# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-trace-context-corruption-at-the-federation-boundary-when-parent-id-lineage-is-silently-re-rooted.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-trace-context-corruption-at-the-federation-boundary-when-parent-id-lineage-is-silently-re-rooted.warrant.json
# Markdown SHA-256:   eb95a951e0aca3e8e4db73b1974c02ce6b551d4c30f141f538f18a23eb1e2fc1
# Code AST SHA-256:   31ff7e05ebc118b33ce8c940246a67a39520b61740bf191cac7073f287998979
# Verified Classes:   TraceLineageCorruptionError, Span
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Iterable


class TraceLineageCorruptionError(RuntimeError):
    """Raised when a child span claims a parent-id that the attributed upstream agent never emitted."""


@dataclass(frozen=True)
class Span:
    trace_id: str
    span_id: str
    parent_id: str | None
    agent: str  # from tracestate attribution


def verify_lineage(
    spans: Iterable[Span],
    expected_orphan_rate: float = 0.0,
    tolerance: float = 0.005,
) -> float:
    """
    Reconcile claimed parent-ids against emitted span-ids per trace.
    Returns the observed orphan rate. Raises if it exceeds expectation + tolerance.
    """
    by_trace: dict[str, dict[str, set[str]]] = {}
    for s in spans:
        by_trace.setdefault(s.trace_id, {}).setdefault(s.agent, set()).add(s.span_id)

    total_children = 0
    orphans = 0
    for trace_id, agents in by_trace.items():
        emitted = set().union(*agents.values())
        for s in spans:
            if s.trace_id != trace_id or s.parent_id is None:
                continue
            total_children += 1
            if s.parent_id not in emitted:
                orphans += 1

    if total_children == 0:
        return 0.0

    observed = orphans / total_children
    if observed > expected_orphan_rate + tolerance:
        raise TraceLineageCorruptionError(
            f"trace lineage corruption: orphan rate {observed:.4f} exceeds "
            f"expected {expected_orphan_rate:.4f} + tolerance {tolerance:.4f}; "
            f"{orphans}/{total_children} child spans reference unemitted parent-ids. "
            f"Suspect a federation egress gateway rewriting traceparent parent-id."
        )
    return observed