# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-prompt-injection-taint-appear-to-vanish-when-an-agent-delegates-to-a-sub-agent-whose-trace-spans-are-emitted-on-a-different-cl.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-prompt-injection-taint-appear-to-vanish-when-an-agent-delegates-to-a-sub-agent-whose-trace-spans-are-emitted-on-a-different-cl.warrant.json
# Markdown SHA-256:   961b5b9ec7aa4efd9590080f0209cca32d292ff7048d470692724348ec7840c1
# Code AST SHA-256:   3f3989aae410546871e0321adc795a4b1e6b7aac3477977b9c11a8bc61efe26c
# Verified Classes:   TaintLineageError, Span
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from __future__ import annotations
from dataclasses import dataclass


class TaintLineageError(RuntimeError):
    """Raised when taint lineage cannot be established from trace data."""


@dataclass(frozen=True)
class Span:
    span_id: str
    parent_span_id: str | None
    trace_id: str
    start_time_ns: int
    taint: str | None  # e.g. "external_untrusted" or None


def reconstruct_taint_lineage(spans: list[Span], leaf_span_id: str) -> list[Span]:
    """Walk parent_span_id edges (NOT start_time) to find tainted ancestors.

    Fails closed: if the causal chain is broken or a clock-skew anomaly is
    detected, raises TaintLineageError rather than returning a clean result.
    """
    by_id = {s.span_id: s for s in spans}
    if leaf_span_id not in by_id:
        raise TaintLineageError(
            f"leaf span {leaf_span_id!r} absent from trace; cannot establish taint lineage"
        )

    chain: list[Span] = []
    cursor: Span | None = by_id[leaf_span_id]
    seen: set[str] = set()

    while cursor is not None:
        if cursor.span_id in seen:
            raise TaintLineageError(f"cycle detected at span {cursor.span_id!r}")
        seen.add(cursor.span_id)
        chain.append(cursor)

        if cursor.parent_span_id is None:
            break

        parent = by_id.get(cursor.parent_span_id)
        if parent is None:
            # Parent span was dropped at ingestion (often due to clock-skew
            # validation). We cannot prove the ancestor is clean.
            raise TaintLineageError(
                f"parent span {cursor.parent_span_id!r} of {cursor.span_id!r} "
                f"missing from trace; taint state is UNKNOWN, not clean"
            )

        # Detect the clock-domain anomaly explicitly: a child that starts
        # before its parent indicates cross-domain skew, so timestamp-based
        # ordering downstream would be unsound.
        if cursor.start_time_ns < parent.start_time_ns:
            raise TaintLineageError(
                f"clock-skew anomaly: span {cursor.span_id!r} starts before parent "
                f"{parent.span_id!r}; timestamp lineage is unsound, fail closed"
            )

        cursor = parent

    return chain


def enforce_taint_policy(spans: list[Span], leaf_span_id: str) -> None:
    """Block side-effecting calls when any ancestor is tainted."""
    try:
        lineage = reconstruct_taint_lineage(spans, leaf_span_id)
    except TaintLineageError as exc:
        # Fail closed: unknown taint state blocks the action.
        raise TaintLineageError(
            f"refusing side-effecting call: {exc}"
        ) from exc

    tainted = [s for s in lineage if s.taint]
    if tainted:
        raise TaintLineageError(
            f"side-effecting call blocked: tainted ancestor "
            f"{tainted[0].span_id!r} (taint={tainted[0].taint!r})"
        )