# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-an-audit-chain-fork-in-federated-agent-traces-and-why-hash-link-continuity-is-not-sufficient-evidence-of-linearity.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-an-audit-chain-fork-in-federated-agent-traces-and-why-hash-link-continuity-is-not-sufficient-evidence-of-linearity.warrant.json
# Markdown SHA-256:   ea130131f91af7a9ae1f38a6fdfd7df6591d6e48696d1eae929398385a450dd8
# Code AST SHA-256:   0893315b64dbc2dbd5314dfdea029f149a108d2c48514ddaccb76ce4e63260f9
# Verified Classes:   TraceForkError, Span
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from collections import defaultdict
from dataclasses import dataclass


class TraceForkError(RuntimeError):
    """Raised when a federated trace has >1 successor at the same causal epoch."""


@dataclass(frozen=True)
class Span:
    trace_id: str
    span_id: str
    parent_span: str | None
    causal_epoch: int
    domain: str
    payload_hash: str


def assert_trace_linearity(spans: list[Span]) -> None:
    """Reject forks even when every per-domain hash chain verified cleanly.

    A fork is two or more spans sharing (trace_id, parent_span, causal_epoch).
    Per-domain chain integrity cannot observe this; only this cross-domain
    join can.
    """
    by_parent: dict[tuple[str, str | None, int], list[Span]] = defaultdict(list)
    for s in spans:
        by_parent[(s.trace_id, s.parent_span, s.causal_epoch)].append(s)

    for key, group in by_parent.items():
        if len(group) < 2:
            continue
        trace_id, parent_span, epoch = key
        domains = sorted({s.domain for s in group})
        hashes = sorted({s.payload_hash for s in group})
        # Divergent payloads across the fork mean side-effect ordering is
        # unrecoverable from the audit trail -> hard failure, not a warning.
        if len(hashes) > 1:
            raise TraceForkError(
                f"trace fork with divergent side effects: trace={trace_id} "
                f"parent={parent_span} epoch={epoch} domains={domains} "
                f"payload_hashes={hashes}; audit trail is non-authoritative "
                f"for this trace"
            )
        raise TraceForkError(
            f"trace fork (identical payloads, still non-linear): trace={trace_id} "
            f"parent={parent_span} epoch={epoch} domains={domains}"
        )