# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-broken-causal-linkage-between-prompt-spans-tool-spans-and-memory-operations-in-agent-execution-graphs.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-broken-causal-linkage-between-prompt-spans-tool-spans-and-memory-operations-in-agent-execution-graphs.warrant.json
# Markdown SHA-256:   7ab502382586c9fe95ce5d371dd4e626323565a2c13ca3f45c166934d7df23bb
# Code AST SHA-256:   ba71fae5eae753286ad8a941155683d4cc14e60f93dc0ab011d62f2995518485
# Verified Classes:   CausalCorrelationError, Span
# Defensive Guards:   7 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class CausalCorrelationError(RuntimeError):
    """Raised when a span graph is structurally complete but causally invalid."""
    pass


@dataclass
class Span:
    span_id: str
    parent_id: str | None
    start_ns: int
    end_ns: int
    kind: str  # "prompt" | "tool" | "memory_read" | "memory_write"
    tool_call_id: str | None = None


CLOCK_SKEW_TOLERANCE_NS = 5_000_000  # 5 ms


def validate_causal_graph(spans: dict[str, Span]) -> None:
    by_parent: dict[str | None, list[Span]] = {}
    for s in spans.values():
        by_parent.setdefault(s.parent_id, []).append(s)

    for s in spans.values():
        if s.parent_id is None:
            continue
        parent = spans.get(s.parent_id)
        if parent is None:
            raise CausalCorrelationError(
                f"span {s.span_id} references missing parent {s.parent_id}"
            )
        if s.start_ns < parent.start_ns - CLOCK_SKEW_TOLERANCE_NS:
            raise CausalCorrelationError(
                f"span {s.span_id} starts before its parent {parent.span_id}: "
                f"causal inversion (child {s.start_ns} < parent {parent.start_ns})"
            )
        if s.end_ns > parent.end_ns + CLOCK_SKEW_TOLERANCE_NS:
            raise CausalCorrelationError(
                f"span {s.span_id} ends after its parent {parent.span_id}: "
                f"interval not contained (child {s.end_ns} > parent {parent.end_ns})"
            )

    # Every tool span must be a child of the prompt span whose completion
    # requested it, keyed by tool_call_id.
    prompt_by_tool_call: dict[str, Span] = {}
    for s in spans.values():
        if s.kind == "prompt" and s.tool_call_id is not None:
            prompt_by_tool_call[s.tool_call_id] = s

    for s in spans.values():
        if s.kind != "tool":
            continue
        if s.tool_call_id is None:
            raise CausalCorrelationError(
                f"tool span {s.span_id} has no tool_call_id; cannot bind to prompt"
            )
        expected_parent = prompt_by_tool_call.get(s.tool_call_id)
        if expected_parent is None:
            raise CausalCorrelationError(
                f"tool span {s.span_id} has tool_call_id {s.tool_call_id} "
                f"with no originating prompt span"
            )
        if s.parent_id != expected_parent.span_id:
            raise CausalCorrelationError(
                f"tool span {s.span_id} parented to {s.parent_id}, "
                f"expected {expected_parent.span_id} (the prompting span)"
            )

    # Memory operations must be parented to the step that issued them,
    # never to the root, and must fall inside their parent's interval.
    for s in spans.values():
        if s.kind not in ("memory_read", "memory_write"):
            continue
        if s.parent_id is None:
            raise CausalCorrelationError(
                f"memory span {s.span_id} parented to root; "
                f"issuing agent step is unrecoverable"
            )