# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-token-accounting-drift-when-genai-agent-spans-report-tool-call-latency-but-omit-gen-ai-usage-on-nested-tool-invocations.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-token-accounting-drift-when-genai-agent-spans-report-tool-call-latency-but-omit-gen-ai-usage-on-nested-tool-invocations.warrant.json
# Markdown SHA-256:   149cad4ab907e66cda93d4cda74e7c8fe88afaf177b7adf87da8b62f396c5787
# Code AST SHA-256:   f2506e6143d987ee2416e9cad3a63a057b2976c433e94ddb92d4c98bc3c83c5a
# Verified Classes:   GenAIAccountingDriftError, Span, Trace
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Iterable


class GenAIAccountingDriftError(RuntimeError):
    """Raised when a trace's GenAI span structure violates the usage-accounting invariant."""


@dataclass
class Span:
    span_id: str
    parent_id: str | None
    operation: str          # e.g. "invoke_agent", "chat", "execute_tool"
    input_tokens: int = 0
    output_tokens: int = 0
    tool_call_id: str | None = None
    children: list["Span"] = field(default_factory=list)


@dataclass
class Trace:
    trace_id: str
    spans: list[Span]


def _flatten(spans: Iterable[Span]) -> list[Span]:
    out: list[Span] = []
    for s in spans:
        out.append(s)
        out.extend(_flatten(s.children))
    return out


def check_trace(trace: Trace, tolerance: float = 0.05) -> dict:
    """
    Enforce the GenAI usage-accounting invariant for one trace.

    Invariants:
      I1. Every tool-call span must coexist with at least one generation span
          (a model must have run to emit a tool call).
      I2. The agent span's declared usage must be >= the sum of its generation
          children's usage, within `tolerance`.
      I3. No single generation child may exceed the agent span's declared usage.
    """
    all_spans = _flatten(trace.spans)
    gens = [s for s in all_spans if s.operation == "chat"]
    tools = [s for s in all_spans if s.operation == "execute_tool"]
    agents = [s for s in all_spans if s.operation == "invoke_agent"]

    # I1: silent-drop detection.
    if tools and not gens:
        raise GenAIAccountingDriftError(
            f"trace {trace.trace_id}: {len(tools)} tool-call span(s) but zero "
            f"generation spans; generation span was dropped, not absent"
        )

    if not agents:
        return {"trace_id": trace.trace_id, "status": "no_agent_span"}

    agent = agents[0]
    declared_in = agent.input_tokens
    declared_out = agent.output_tokens
    child_in = sum(g.input_tokens for g in gens)
    child_out = sum(g.output_tokens for g in gens)

    # I3: child cannot exceed parent.
    for g in gens:
        if g.input_tokens > declared_in or g.output_tokens > declared_out:
            raise GenAIAccountingDriftError(
                f"trace {trace.trace_id}: generation span {g.span_id} usage "
                f"({g.input_tokens}/{g.output_tokens}) exceeds agent span "
                f"declared usage ({declared_in}/{declared_out}); parent "
                f"attribute written from stale counter"
            )

    # I2: aggregate drift.
    if child_in > 0:
        drift_in = (child_in - declared_in) / child_in
        drift_out = (child_out - declared_out) / child_out if child_out else 0.0
        if drift_in > tolerance or drift_out > tolerance:
            raise GenAIAccountingDriftError(
                f"trace {trace.trace_id}: agent span undercounts children by "
                f"in={drift_in:.1%} out={drift_out:.1%} (tolerance {tolerance:.0%})"
            )

    return {
        "trace_id": trace.trace_id,
        "status": "ok",
        "generations": len(gens),
        "tool_calls": len(tools),
        "drift_in": round((child_in - declared_in) / child_in, 4) if child_in else 0.0,
    }