# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/across-what-contextvar-and-w3c-traceparent-boundaries-must-an-instance-propagate-distributed-correlation-ids-to-prevent-orphaned-spans-and-baggage-bleed.md
# Remnant Warrant:    https://corpusfetch.com/c/across-what-contextvar-and-w3c-traceparent-boundaries-must-an-instance-propagate-distributed-correlation-ids-to-prevent-orphaned-spans-and-baggage-bleed.warrant.json
# Markdown SHA-256:   c72db2225ac63fb824298e044c3b710a8a30f51082cd144ebfdc5243f4c270ac
# Code AST SHA-256:   f2897d2f3b509b161ae175aa6cf9ed382479d085d5b5a29c4950e4fc40fa8621
# Verified Classes:   TraceContextPropagationError, W3CTraceContext, W3CTraceparentPropagationGate
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import contextvars
import re
import secrets
from dataclasses import dataclass
from typing import Final


class TraceContextPropagationError(RuntimeError):
    """Raised when a W3C traceparent header is malformed, all-zero, or leaks cross-tenant baggage."""
    pass


@dataclass(frozen=True)
class W3CTraceContext:
    version: str
    trace_id: str
    parent_id: str
    trace_flags: str
    tenant_id: str

    def to_traceparent(self) -> str:
        return f"{self.version}-{self.trace_id}-{self.parent_id}-{self.trace_flags}"


_ACTIVE_TRACE_CTX: contextvars.ContextVar[W3CTraceContext | None] = contextvars.ContextVar(
    "_ACTIVE_TRACE_CTX", default=None
)


class W3CTraceparentPropagationGate:
    """
    Enforces strict W3C traceparent parsing, child span rotation, and contextvars isolation:
    1. Rejects invalid hex or all-zero trace_id / parent_id values.
    2. Rotates parent_id on every outbound hop so child spans never collide.
    3. Binds tenant_id to prevent cross-request baggage bleed across async tasks.
    """
    TRACEPARENT_RE: Final[re.Pattern[str]] = re.compile(
        r"\A(?P<ver>[0-9a-f]{2})-(?P<tid>[0-9a-f]{32})-(?P<pid>[0-9a-f]{16})-(?P<flg>[0-9a-f]{2})\Z"
    )

    @classmethod
    def parse_and_bind(cls, traceparent: str, tenant_id: str) -> W3CTraceContext:
        if not tenant_id or not tenant_id.strip():
            raise TraceContextPropagationError("tenant_id is required to prevent baggage bleed.")
        match = cls.TRACEPARENT_RE.match(traceparent.strip().lower())
        if not match:
            raise TraceContextPropagationError(f"Malformed W3C traceparent header: {traceparent!r}")

        ver = match.group("ver")
        tid = match.group("tid")
        pid = match.group("pid")
        flg = match.group("flg")

        if ver == "ff" or tid == "0" * 32 or pid == "0" * 16:
            raise TraceContextPropagationError("Invalid W3C traceparent: reserved version or all-zero IDs.")

        ctx = W3CTraceContext(
            version=ver,
            trace_id=tid,
            parent_id=pid,
            trace_flags=flg,
            tenant_id=tenant_id.strip(),
        )
        _ACTIVE_TRACE_CTX.set(ctx)
        return ctx

    @classmethod
    def spawn_child_span(cls, expected_tenant_id: str) -> W3CTraceContext:
        current = _ACTIVE_TRACE_CTX.get()
        if current is None:
            raise TraceContextPropagationError("Cannot spawn child span: no active trace context in ContextVar.")
        if current.tenant_id != expected_tenant_id:
            raise TraceContextPropagationError(
                f"Cross-tenant baggage bleed blocked: active={current.tenant_id!r}, expected={expected_tenant_id!r}"
            )
        child_span_id = secrets.token_hex(8)
        return W3CTraceContext(
            version=current.version,
            trace_id=current.trace_id,
            parent_id=child_span_id,
            trace_flags=current.trace_flags,
            tenant_id=current.tenant_id,
        )