# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-delegated-a2a-task-lose-its-parent-trace-span-when-the-delegating-agent-reuses-a-pooled-http-client-across-tenants.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-delegated-a2a-task-lose-its-parent-trace-span-when-the-delegating-agent-reuses-a-pooled-http-client-across-tenants.warrant.json
# Markdown SHA-256:   7cc09f9d3a5f02fa70f3741a828aaa7b5ee64423a12f3b68b042e381e8be17db
# Code AST SHA-256:   6277695550ec51cf05e70948138cb53697c2a2abf52442eb3f48652ca61ce462
# Verified Classes:   TraceBindingError, TraceContext
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Optional


class TraceBindingError(RuntimeError):
    """Raised when a delegated task's trace context cannot be bound to the caller's active span."""
    pass


@dataclass(frozen=True)
class TraceContext:
    trace_id: str
    span_id: str


def bind_delegation_context(
    caller_span: Optional[TraceContext],
    captured_at_call_time: bool,
    pooled_client_shared: bool,
) -> TraceContext:
    # The invariant requires capture at call time; ambient re-reads at send
    # time are the failure mode under pooled, shared clients.
    if caller_span is None:
        raise TraceBindingError(
            "no active caller span at delegation time; refusing to delegate "
            "with an unbound trace context"
        )
    if not captured_at_call_time:
        raise TraceBindingError(
            f"trace context for trace_id={caller_span.trace_id} was not captured "
            "at delegation call time; ambient injection under a pooled client "
            "will orphan or mis-parent the delegated span"
        )
    if pooled_client_shared and captured_at_call_time is False:
        # Defensive: explicit conjunction check, never reachable given above,
        # but documents the exact condition of validity.
        raise TraceBindingError("pooled shared client with deferred context read")
    return TraceContext(trace_id=caller_span.trace_id, span_id=caller_span.span_id)


def assert_parent_continuity(
    delegated_trace_id: str,
    delegated_parent_span_id: str,
    known_spans_in_trace: set[str],
) -> None:
    if delegated_trace_id not in known_spans_in_trace and delegated_parent_span_id not in known_spans_in_trace:
        raise TraceBindingError(
            f"delegated span parent={delegated_parent_span_id} not present in "
            f"trace_id={delegated_trace_id}; trace graph edge is broken"
        )