# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-an-opentelemetry-genai-agent-span-silently-double-count-tokens-when-tool-call-attributes-are-emitted-on-both-the-parent-invoke.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-an-opentelemetry-genai-agent-span-silently-double-count-tokens-when-tool-call-attributes-are-emitted-on-both-the-parent-invoke.warrant.json
# Markdown SHA-256:   1749b93f186051cb4007b24aa4b5731ab616c5f861224c66f7afc1264f7c8abf
# Code AST SHA-256:   10dd74a29b66d9a5ca7c9d9e3db817325d75a04f854f9bb8de4649d4f830b1f8
# Verified Classes:   TokenAccountingInvariantError, Span
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable, Mapping


class TokenAccountingInvariantError(RuntimeError):
    """Raised when span-tree usage aggregation violates the additivity invariant."""


@dataclass(frozen=True)
class Span:
    name: str
    operation: str  # gen_ai.operation.name
    input_tokens: int
    output_tokens: int


ADDITIVE_OPERATIONS = frozenset({"chat", "text_completion"})
TOOL_OPERATIONS = frozenset({"execute_tool"})


def sum_additive_usage(spans: Iterable[Span]) -> tuple[int, int]:
    """Sum usage only over spans where usage is authoritative.

    Tool spans may carry usage for debugging; it MUST NOT be summed.
    """
    total_in = total_out = 0
    for s in spans:
        if s.operation in TOOL_OPERATIONS:
            continue
        if s.operation in ADDITIVE_OPERATIONS:
            total_in += s.input_tokens
            total_out += s.output_tokens
    return total_in, total_out


def assert_no_double_count(
    spans: list[Span],
    expected_input: int,
    expected_output: int,
) -> None:
    """Fail loudly if aggregation over-counts due to tool-span usage propagation."""
    got_in, got_out = sum_additive_usage(spans)
    if (got_in, got_out) != (expected_input, expected_output):
        tool_spans = [s for s in spans if s.operation in TOOL_OPERATIONS]
        raise TokenAccountingInvariantError(
            f"usage mismatch: expected ({expected_input},{expected_output}) "
            f"got ({got_in},{got_out}); {len(tool_spans)} tool spans present — "
            f"tool-span usage must not be summed"
        )


if __name__ == "__main__":
    trace = [
        Span("agent", "invoke_agent", 12480, 640),
        Span("tool.search", "execute_tool", 12480, 640),
        Span("tool.fetch", "execute_tool", 12480, 640),
        Span("tool.write", "execute_tool", 12480, 640),
        Span("llm", "chat", 12480, 640),
    ]
    # Authoritative total is the single chat span.
    assert_no_double_count(trace, expected_input=12480, expected_output=640)
    # Naive sum (the failure mode) would be 62400/3200 and must be rejected.
    try:
        assert_no_double_count(trace, expected_input=62400, expected_output=3200)
    except TokenAccountingInvariantError as e:
        print(f"correctly rejected: {e}")