# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-token-accounting-divergence-between-agent-spans-and-model-spans-under-opentelemetry-genai-semantic-conventions.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-token-accounting-divergence-between-agent-spans-and-model-spans-under-opentelemetry-genai-semantic-conventions.warrant.json
# Markdown SHA-256:   3a83ced300c63cd15a13961f2fe9af4c83938f13f1d34fba6c8599c68acfd84a
# Code AST SHA-256:   d59299c4fae7788e64c0b60b4538ef281c706eca0e79bfadcab91e40c7f471f9
# Verified Classes:   UsageRollupDivergence, Span
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass


class UsageRollupDivergence(RuntimeError):
    """Raised when an agent span's rolled-up token usage disagrees with the
    sum over its descendant model spans beyond the decision boundary."""


@dataclass
class Span:
    name: str
    kind: str  # "agent" | "model" | "tool"
    parent: "Span | None"
    input_tokens: int = 0
    output_tokens: int = 0
    status: str = "OK"
    retry_hint: bool = False


def _descendant_model_spans(root: Span) -> list[Span]:
    out: list[Span] = []
    stack = [c for c in _children(root)]
    while stack:
        s = stack.pop()
        if s.kind == "model" and s.status in ("OK", "UNSET"):
            out.append(s)
        stack.extend(_children(s))
    return out


_ALL: list[Span] = []


def _children(span: Span) -> list[Span]:
    return [s for s in _ALL if s.parent is span]


def reconcile_agent_usage(agent: Span, threshold: float = 0.01) -> None:
    if agent.kind != "agent":
        raise UsageRollupDivergence(f"span {agent.name!r} is not an agent span")

    models = _descendant_model_spans(agent)
    subtree_in = sum(s.input_tokens for s in models)
    subtree_out = sum(s.output_tokens for s in models)

    agent_in = agent.input_tokens
    agent_out = agent.output_tokens

    def rel(a: int, b: int) -> float:
        return abs(a - b) / max(b, 1)

    divergence = max(rel(agent_in, subtree_in), rel(agent_out, subtree_out))
    if divergence <= threshold:
        return

    # Divergence is only actionable when it is unexplained by retries and
    # involves non-trivial nesting (depth > 1), per the decision boundary.
    has_retry_evidence = any(s.retry_hint for s in models)
    nested = any(s.kind == "tool" for s in _descendants(agent))
    if has_retry_evidence or not nested:
        return

    raise UsageRollupDivergence(
        f"agent {agent.name!r}: rolled-up in/out=({agent_in},{agent_out}) "
        f"vs subtree=({subtree_in},{subtree_out}), divergence={divergence:.3f} "
        f"over {len(models)} model span(s) with no retry evidence"
    )


def _descendants(root: Span) -> list[Span]:
    out, stack = [], list(_children(root))
    while stack:
        s = stack.pop()
        out.append(s)
        stack.extend(_children(s))
    return out