# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-continue-a-trace-context-it-did-not-originate-otel-genai-span-linking-and-a2a-handoff-propagation-boundar.md
# Remnant Warrant:    https://corpusfetch.com/c/in-what-scenarios-must-an-instance-refuse-to-continue-a-trace-context-it-did-not-originate-otel-genai-span-linking-and-a2a-handoff-propagation-boundar.warrant.json
# Markdown SHA-256:   494a4685a19cd5230c0bdfebbd22be27a26d4be1135c3dd20278f1515b947df3
# Code AST SHA-256:   f68405f68e2a5e2457b8d45e68779166932cd3df09403c3d9ae3722e7de56d83
# Verified Classes:   UntrustedTraceContextError, InboundContext
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import re
from dataclasses import dataclass

_TRACEPARENT = re.compile(
    r"^00-([0-9a-f]{32})-([0-9a-f]{16})-([0-9a-f]{2})$"
)


class UntrustedTraceContextError(RuntimeError):
    """Raised when an inbound trace context may not be continued."""


@dataclass(frozen=True)
class InboundContext:
    trace_id: str
    span_id: str
    sampled: bool
    sender_authenticated: bool
    sender_in_trust_domain: bool


def parse_traceparent(header: str) -> InboundContext:
    m = _TRACEPARENT.match(header.strip())
    if not m:
        raise UntrustedTraceContextError(
            f"malformed traceparent, refusing to continue: {header!r}"
        )
    trace_id, span_id, flags = m.groups()
    if trace_id == "0" * 32 or span_id == "0" * 16:
        raise UntrustedTraceContextError("all-zero trace/span id is invalid")
    # Authentication/domain fields are populated by the A2A envelope layer,
    # not by the header itself; defaults are the untrusted case.
    return InboundContext(
        trace_id=trace_id,
        span_id=span_id,
        sampled=bool(int(flags, 16) & 0x01),
        sender_authenticated=False,
        sender_in_trust_domain=False,
    )


def decide_propagation(ctx: InboundContext) -> str:
    """Return 'continue' or 'link'. Fails closed on any doubt."""
    if not ctx.sender_authenticated:
        return "link"
    if not ctx.sender_in_trust_domain:
        return "link"
    return "continue"


def continue_or_link(header: str, *, authenticated: bool, in_domain: bool) -> str:
    ctx = parse_traceparent(header)
    ctx = InboundContext(
        trace_id=ctx.trace_id,
        span_id=ctx.span_id,
        sampled=ctx.sampled,
        sender_authenticated=authenticated,
        sender_in_trust_domain=in_domain,
    )
    decision = decide_propagation(ctx)
    if decision == "continue" and not ctx.sampled:
        # Even in-domain, a not-sampled inbound bit must not suppress a
        # security-relevant span; the local sampler is authoritative.
        raise UntrustedTraceContextError(
            "in-domain continuation with sampled=0 would suppress local export; "
            "force local sampling decision instead of inheriting"
        )
    return decision