# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-agent-runtime-refuse-to-attest-a-tool-call-dag-whose-edges-were-reconstructed-from-post-hoc-logs.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-agent-runtime-refuse-to-attest-a-tool-call-dag-whose-edges-were-reconstructed-from-post-hoc-logs.warrant.json
# Markdown SHA-256:   5a778afaf24bb13ef4c673d57169b428f574d29d433052745c3ec50e1d21074d
# Code AST SHA-256:   6bcbb5dc6f4aba0fe776d0cf834c4d3db0cdcfe5821ad107578e3f2bac00003e
# Verified Classes:   AmbiguousAttestationError, Span, Edge
# Defensive Guards:   2 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field


class AmbiguousAttestationError(RuntimeError):
    """Raised when a tool-call DAG cannot be causally attested."""


@dataclass
class Span:
    span_id: str
    tool: str
    output_hash: str
    parent_pointer: str | None = None  # inline, minted by caller
    timestamp: int = 0


@dataclass
class Edge:
    src: str
    dst: str
    kind: str  # "asserted" | "inferred"


def reconstruct_edges(spans: list[Span]) -> list[Edge]:
    edges: list[Edge] = []
    by_output: dict[str, list[str]] = {}
    by_time: dict[int, list[str]] = {}

    for s in spans:
        if s.parent_pointer is not None:
            edges.append(Edge(s.parent_pointer, s.span_id, "asserted"))
        else:
            by_output.setdefault(s.output_hash, []).append(s.span_id)
            by_time.setdefault(s.timestamp, []).append(s.span_id)

    # Only promote inferred edges when the join key is unique.
    for s in spans:
        if s.parent_pointer is not None:
            continue
        candidates = [sid for sid in by_output.get(s.output_hash, []) if sid != s.span_id]
        if len(candidates) == 1:
            edges.append(Edge(candidates[0], s.span_id, "inferred"))
        elif len(candidates) > 1:
            raise AmbiguousAttestationError(
                f"output hash {s.output_hash!r} for span {s.span_id} "
                f"matches {len(candidates)} candidates {candidates}; "
                f"cannot attest causal edge"
            )
        concurrent = [sid for sid in by_time.get(s.timestamp, []) if sid != s.span_id]
        if len(concurrent) >= 1 and s.parent_pointer is None:
            raise AmbiguousAttestationError(
                f"span {s.span_id} shares tick {s.timestamp} with {concurrent}; "
                f"ordering is not happens-before"
            )
    return edges


def attest(spans: list[Span], signing_key: bytes) -> dict:
    edges = reconstruct_edges(spans)  # may raise AmbiguousAttestationError
    payload = ";".join(f"{e.src}->{e.dst}:{e.kind}" for e in sorted(edges, key=lambda e: (e.src, e.dst)))
    digest = hashlib.sha256(signing_key + payload.encode()).hexdigest()
    return {"edges": edges, "digest": digest, "asserted_only": all(e.kind == "asserted" for e in edges)}