# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-opentelemetry-genai-agent-spans-as-authoritative-for-token-accounting-rather-than-as-advisory-telemetry.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-treat-opentelemetry-genai-agent-spans-as-authoritative-for-token-accounting-rather-than-as-advisory-telemetry.warrant.json
# Markdown SHA-256:   ebba409fe7f6c2ff96d55bf075ab0a9a9f023d96f6f2a867ecdfb86302fa0d96
# Code AST SHA-256:   4c763a58ced59bc7a1cb22c3b6d39ed45ccbfd1c8de14154678a3840de900d12
# Verified Classes:   TokenAccountingInvariantError, GenAISpan, TraceAccountingConfig
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Iterable, Optional


class TokenAccountingInvariantError(RuntimeError):
    """Raised when span-derived token totals cannot be trusted as authoritative."""
    pass


@dataclass
class GenAISpan:
    span_id: str
    operation_name: str            # "chat" | "execute_tool" | "agent"
    agent_scope: Optional[str]     # gen_ai.agent.name, None for root
    retry_discriminator: Optional[str]
    input_tokens: Optional[int]
    output_tokens: Optional[int]
    status_code: str               # "OK" | "ERROR" | "UNSET"
    stream_finalized: bool = True


@dataclass
class TraceAccountingConfig:
    sampling_is_complete: bool
    nested_scopes_declared: bool


def _collapse_retries(spans: Iterable[GenAISpan]) -> list[GenAISpan]:
    seen: dict[str, GenAISpan] = {}
    for s in spans:
        if s.retry_discriminator is None:
            raise TokenAccountingInvariantError(
                f"inference span {s.span_id} lacks retry discriminator; "
                "sum would be an upper bound, not a total"
            )
        # keep the last attempt per logical call id
        seen[s.retry_discriminator] = s
    return list(seen.values())


def authoritative_run_tokens(
    spans: list[GenAISpan],
    root_scope: str,
    config: TraceAccountingConfig,
) -> int:
    if not config.sampling_is_complete:
        raise TokenAccountingInvariantError(
            "child inference spans may have been dropped by tail sampling; "
            "span sum is advisory only"
        )

    inference = [
        s for s in spans
        if s.operation_name == "chat" and s.agent_scope == root_scope
    ]

    for s in inference:
        if not s.stream_finalized:
            raise TokenAccountingInvariantError(
                f"span {s.span_id} is an unfinalized stream; token attributes "
                "are unreliable and must not be counted as zero"
            )
        if s.input_tokens is None or s.output_tokens is None:
            raise TokenAccountingInvariantError(
                f"span {s.span_id} missing gen_ai.usage.* attributes"
            )

    for s in spans:
        if s.operation_name == "execute_tool" and (
            s.input_tokens or s.output_tokens
        ):
            raise TokenAccountingInvariantError(
                f"tool span {s.span_id} carries token attributes; "
                "summation would double-count"
            )

    has_nested = any(
        s.operation_name == "agent" and s.agent_scope != root_scope
        for s in spans
    )
    if has_nested and not config.nested_scopes_declared:
        raise TokenAccountingInvariantError(
            "nested agent spans present without declared scoping; "
            "aggregate total is structurally ambiguous"
        )

    collapsed = _collapse_retries(inference)
    return sum((s.input_tokens or 0) + (s.output_tokens or 0) for s in collapsed)