# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-gen-ai-usage-input-tokens-be-attributed-when-an-agent-span-both-invokes-a-tool-and-re-enters-the-model.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-gen-ai-usage-input-tokens-be-attributed-when-an-agent-span-both-invokes-a-tool-and-re-enters-the-model.warrant.json
# Markdown SHA-256:   a1a08ceb1852b5f3a95f85b3c818d903eea1cc9aff7d18a607097dc485941f22
# Code AST SHA-256:   0e4c68d8f1aa6440e7b82f6b154d6e9bd0e8c4a5a8590ef88afc11564744e2e6
# Verified Classes:   GenAIUsageAttributionError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from typing import Iterable, Mapping, Any

INFERENCE_OPS = frozenset({"chat", "text_completion", "generate_content"})
USAGE_KEYS = ("gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens")


class GenAIUsageAttributionError(RuntimeError):
    """Raised when gen_ai.usage.* is attached to a span that is not an inference span."""


def validate_usage_attribution(spans: Iterable[Mapping[str, Any]]) -> None:
    """
    Enforce the OpenTelemetry GenAI invariant: usage attributes belong ONLY to
    inference spans (operation.name in INFERENCE_OPS). Agent, workflow, and
    execute_tool spans must not carry usage; doing so double-counts when
    downstream consumers sum span-level tokens against provider billing.

    `spans` is a list of dicts with keys: 'name', 'attributes'.
    """
    for span in spans:
        attrs = span.get("attributes", {})
        op = attrs.get("gen_ai.operation.name")
        present = [k for k in USAGE_KEYS if k in attrs]
        if not present:
            continue
        if op not in INFERENCE_OPS:
            raise GenAIUsageAttributionError(
                f"span {span.get('name')!r} (operation.name={op!r}) carries "
                f"{present}; usage attributes are reserved for inference spans "
                f"{sorted(INFERENCE_OPS)}. Hoist usage to the child chat span "
                f"or emit a gen_ai.usage.reported span event instead."
            )
        # Sanity: input/output must be non-negative integers.
        for k in present:
            v = attrs[k]
            if not isinstance(v, int) or v < 0:
                raise GenAIUsageAttributionError(
                    f"span {span.get('name')!r} has malformed {k}={v!r}; "
                    f"expected non-negative int."
                )


def reconcile_turn_tokens(spans: Iterable[Mapping[str, Any]], billed_input: int) -> None:
    """Compare summed inference-span input tokens against provider billing."""
    total = 0
    for span in spans:
        attrs = span.get("attributes", {})
        if attrs.get("gen_ai.operation.name") in INFERENCE_OPS:
            total += int(attrs.get("gen_ai.usage.input_tokens", 0))
    if total != billed_input:
        raise GenAIUsageAttributionError(
            f"summed inference input tokens ({total}) != billed ({billed_input}); "
            f"check for usage hoisted onto agent/tool spans."
        )