# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-tokens-when-an-agent-span-and-its-child-tool-call-spans-both-report-usage.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-tokens-when-an-agent-span-and-its-child-tool-call-spans-both-report-usage.warrant.json
# Markdown SHA-256:   0b30091e0d0a1e879059259c95df45ca523582c383803b88328726f1e830e413
# Code AST SHA-256:   618fc805e368b40cb8468eccef1e9074f5156da9330288e86f2217a3c3b7b7ab
# Verified Classes:   UsageAttributionError
# Defensive Guards:   2 fail-closed invariants (zero external deps)

class UsageAttributionError(RuntimeError):
    """Raised when a span tree's usage attributes cannot be safely aggregated."""


def aggregate_input_tokens(spans: list[dict]) -> int:
    """Sum input tokens across a trace using leaf-only attribution.

    Each span is a dict with keys: span_id, parent_id, genai.usage.input_tokens.
    Invariant enforced: every non-leaf span's usage must be >= the sum of its
    direct children's usage. Violation means the parent is emitting a delta or
    look-ahead value, and leaf-summation would undercount.
    """
    by_parent: dict[str | None, list[dict]] = {}
    for s in spans:
        by_parent.setdefault(s["parent_id"], []).append(s)

    for parent_id, children in by_parent.items():
        if parent_id is None:
            continue
        parent = next((s for s in spans if s["span_id"] == parent_id), None)
        if parent is None:
            raise UsageAttributionError(
                f"orphan span {children[0]['span_id']} references missing parent {parent_id}"
            )
        child_sum = sum(c["genai.usage.input_tokens"] for c in children)
        if parent["genai.usage.input_tokens"] < child_sum:
            raise UsageAttributionError(
                f"parent {parent_id} reports {parent['genai.usage.input_tokens']} "
                f"input tokens but children sum to {child_sum}; "
                "parent is emitting a delta, not a rollup — leaf summation unsafe"
            )

    leaves = [s for s in spans if s["span_id"] not in by_parent]
    return sum(s["genai.usage.input_tokens"] for s in leaves)