# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-reconcile-opentelemetry-genai-tool-call-span-attributes-when-a-single-tool-invocation-produces-multiple-token-level-model-spans.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-reconcile-opentelemetry-genai-tool-call-span-attributes-when-a-single-tool-invocation-produces-multiple-token-level-model-spans.warrant.json
# Markdown SHA-256:   cb7839796c2316c016af9198dc1ec326888f8503ec1037cc3d1bc3267f515e76
# Code AST SHA-256:   6adbc589a76c3563216900fa529fd53aff02d35e1f2914d3fb8735a35d0eb1ba
# Verified Classes:   ToolCallAttributionError, Span
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from typing import Iterable, Mapping


class ToolCallAttributionError(RuntimeError):
    """Raised when token usage cannot be attributed unambiguously to a tool call."""


@dataclass(frozen=True)
class Span:
    span_id: str
    operation: str          # e.g. "chat", "execute_tool"
    tool_call_id: str | None
    input_tokens: int
    output_tokens: int
    retry_index: int = 0


INFERENCE_OPS = frozenset({"chat", "text_completion", "embeddings"})


def attribute_tokens(spans: Iterable[Span]) -> dict[str, tuple[int, int]]:
    """
    Return {tool_call_id: (input_tokens, output_tokens)} counting each
    inference span exactly once. Mirrored usage on execute_tool spans is
    excluded. Ambiguous multi-span attribution raises rather than guessing.
    """
    per_call: dict[str, list[Span]] = {}
    for s in spans:
        # Mirrored usage on tool-execution spans is never authoritative.
        if s.operation == "execute_tool":
            continue
        if s.operation not in INFERENCE_OPS:
            continue
        if s.tool_call_id is None:
            continue
        per_call.setdefault(s.tool_call_id, []).append(s)

    result: dict[str, tuple[int, int]] = {}
    for call_id, group in per_call.items():
        # Multiple inference spans for one call id are only legal if each
        # carries a distinct retry_index; otherwise attribution is ambiguous.
        retry_indices = {s.retry_index for s in group}
        if len(group) > 1 and len(retry_indices) != len(group):
            raise ToolCallAttributionError(
                f"tool_call_id={call_id!r} maps to {len(group)} inference spans "
                f"with non-distinct retry_index values {sorted(retry_indices)}; "
                "cannot attribute tokens without a retry discriminator"
            )
        inp = sum(s.input_tokens for s in group)
        out = sum(s.output_tokens for s in group)
        result[call_id] = (inp, out)
    return result


def reconcile_against_invoice(
    attributed: Mapping[str, tuple[int, int]],
    invoice_input: int,
    invoice_output: int,
    tolerance: float = 0.02,
) -> None:
    total_in = sum(v[0] for v in attributed.values())
    total_out = sum(v[1] for v in attributed.values())
    if invoice_input and abs(total_in - invoice_input) / invoice_input > tolerance:
        raise ToolCallAttributionError(
            f"input token mismatch: attributed={total_in} invoice={invoice_input} "
            f"(delta {total_in - invoice_input:+d}, tolerance {tolerance:.0%})"
        )
    if invoice_output and abs(total_out - invoice_output) / invoice_output > tolerance:
        raise ToolCallAttributionError(
            f"output token mismatch: attributed={total_out} invoice={invoice_output} "
            f"(delta {total_out - invoice_output:+d}, tolerance {tolerance:.0%})"
        )