# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-terminate-an-agent-span-when-a-tool-call-is-retried-and-token-usage-is-double-counted-under-opentelemetry-genai-semantic-conven.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-terminate-an-agent-span-when-a-tool-call-is-retried-and-token-usage-is-double-counted-under-opentelemetry-genai-semantic-conven.warrant.json
# Markdown SHA-256:   51cf1ec5d7f9ff1ec805e4a8a7735cf3e9bd364b400c694a4fa5ba9c925117a0
# Code AST SHA-256:   78e6e2494c21b2c0c13da2bc92522b3fbaa0c1f7b64dfa1f091d81dc40d2fdf8
# Verified Classes:   GenAiUsageInvariantError, ToolAttempt, LogicalToolCall
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class GenAiUsageInvariantError(RuntimeError):
    """Raised when agent-span token accounting violates the logical-call boundary."""


@dataclass
class ToolAttempt:
    call_id: str
    attempt: int
    input_tokens: int
    output_tokens: int
    succeeded: bool
    error_type: Optional[str] = None


@dataclass
class LogicalToolCall:
    call_id: str
    attempts: list[ToolAttempt] = field(default_factory=list)

    def record(self, attempt: ToolAttempt) -> None:
        if attempt.call_id != self.call_id:
            raise GenAiUsageInvariantError(
                f"attempt call_id {attempt.call_id!r} does not match logical call "
                f"{self.call_id!r}; cannot merge into one span"
            )
        self.attempts.append(attempt)

    def final_usage(self) -> tuple[int, int]:
        """Usage must be attributed to exactly one attempt: the terminal one."""
        if not self.attempts:
            raise GenAiUsageInvariantError(
                f"logical call {self.call_id!r} closed with zero attempts"
            )
        terminal = self.attempts[-1]
        if len(self.attempts) > 1 and any(a.succeeded for a in self.attempts[:-1]):
            raise GenAiUsageInvariantError(
                f"call {self.call_id!r} has a successful non-terminal attempt; "
                "retries must not follow success"
            )
        return terminal.input_tokens, terminal.output_tokens


def emit_agent_span_usage(
    logical_calls: list[LogicalToolCall],
    model_turn_input: int,
    model_turn_output: int,
    per_attempt_usage_emitted: bool,
) -> dict[str, int]:
    if per_attempt_usage_emitted:
        # Parent must not also accrue tool usage; doing so double-counts retries.
        return {
            "gen_ai.usage.input_tokens": model_turn_input,
            "gen_ai.usage.output_tokens": model_turn_output,
        }

    tool_in = 0
    tool_out = 0
    seen: set[str] = set()
    for call in logical_calls:
        if call.call_id in seen:
            raise GenAiUsageInvariantError(
                f"duplicate logical call id {call.call_id!r} in one agent span"
            )
        seen.add(call.call_id)
        i, o = call.final_usage()
        tool_in += i
        tool_out += o

    return {
        "gen_ai.usage.input_tokens": model_turn_input + tool_in,
        "gen_ai.usage.output_tokens": model_turn_output + tool_out,
    }