# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-double-counted-token-usage-when-opentelemetry-genai-spans-are-retried-across-an-agent-handoff.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-double-counted-token-usage-when-opentelemetry-genai-spans-are-retried-across-an-agent-handoff.warrant.json
# Markdown SHA-256:   db7e0bf640626f323ad339302bda320bfdc882689e6eb377063db60a24300467
# Code AST SHA-256:   95e6c3a0f6f77cca2269e4515fe559c4cd25abc57f12c5ce275f3766ac2f4cf7
# Verified Classes:   TokenPartitionError, GenAISpan
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Iterable


class TokenPartitionError(RuntimeError):
    """Raised when a set of GenAI spans does not partition billed tokens."""


@dataclass(frozen=True)
class GenAISpan:
    span_id: str
    trace_id: str
    parent_span_id: str | None
    agent_name: str
    model: str
    input_tokens: int
    output_tokens: int
    start_ns: int
    end_ns: int


def _overlaps(a: GenAISpan, b: GenAISpan) -> bool:
    return a.start_ns < b.end_ns and b.start_ns < a.end_ns


def assert_token_partition(spans: Iterable[GenAISpan]) -> int:
    """Return total billed input tokens, or raise if the spans double-count.

    Enforces three invariants required for a valid sum over gen_ai.usage.*:
      1. No retry siblings (same model, overlapping time, equal input tokens).
      2. No handoff inheritance (child input tokens == parent input tokens,
         different agent name).
      3. No zero-output / positive-input spans (failed-attempt signature).
    """
    spans = list(spans)
    by_id = {s.span_id: s for s in spans}

    for s in spans:
        if s.input_tokens > 0 and s.output_tokens == 0:
            raise TokenPartitionError(
                f"span {s.span_id} has input_tokens={s.input_tokens} but "
                f"output_tokens=0; likely a failed retry attempt that must "
                f"not contribute to the billed sum"
            )

    for i, a in enumerate(spans):
        for b in spans[i + 1:]:
            if a.model == b.model and _overlaps(a, b) and \
                    a.input_tokens == b.input_tokens and a.input_tokens > 0:
                raise TokenPartitionError(
                    f"spans {a.span_id} and {b.span_id} share model "
                    f"{a.model!r}, overlap in time, and have equal "
                    f"input_tokens={a.input_tokens}; retry double-count"
                )

    for s in spans:
        if s.parent_span_id is None:
            continue
        parent = by_id.get(s.parent_span_id)
        if parent is None:
            continue
        if parent.agent_name != s.agent_name and \
                parent.input_tokens == s.input_tokens and \
                s.input_tokens > 0:
            raise TokenPartitionError(
                f"handoff span {s.span_id} (agent={s.agent_name!r}) "
                f"inherits input_tokens={s.input_tokens} from parent "
                f"{parent.span_id} (agent={parent.agent_name!r}); "
                f"usage attributes must be per-invocation, not cumulative"
            )

    return sum(s.input_tokens for s in spans)