# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/why-does-an-instance-of-a-genai-agent-span-report-zero-genai-usage-output-tokens-while-the-tool-call-child-span-carries-a-full-genai-tool-call-argumen.md
# Remnant Warrant:    https://corpusfetch.com/c/why-does-an-instance-of-a-genai-agent-span-report-zero-genai-usage-output-tokens-while-the-tool-call-child-span-carries-a-full-genai-tool-call-argumen.warrant.json
# Markdown SHA-256:   4293c373e8b6b6edfe895c1e86482383876cb0b5f44241b28297e101eab714f2
# Code AST SHA-256:   cf73edba90addc18e86b8e90984903756050e3ea2f189593daddc4e9cecb55b1
# Verified Classes:   GenAIUsageInvariantError
# Defensive Guards:   4 fail-closed invariants (zero external deps)

class GenAIUsageInvariantError(RuntimeError):
    """Raised when a GenAI span violates the usage-attribution invariant."""


_MODEL_OPS = frozenset({"chat", "text_completion", "embeddings"})
_AGGREGATE_OPS = frozenset({"invoke_agent", "execute_tool"})


def validate_span_usage(span_attrs: dict) -> None:
    """Enforce that usage attributes live only on model-invocation spans.

    span_attrs is the flattened attribute dict of a single emitted span,
    e.g. {"genai.operation.name": "invoke_agent",
          "genai.usage.input_tokens": 1841, ...}
    """
    op = span_attrs.get("genai.operation.name")
    has_usage = any(k.startswith("genai.usage.") for k in span_attrs)
    has_tool_args = "genai.tool.call.arguments" in span_attrs

    if op in _AGGREGATE_OPS and has_usage:
        raise GenAIUsageInvariantError(
            f"span op={op!r} carries genai.usage.* attributes; usage must be "
            f"attached only to model-invocation spans {sorted(_MODEL_OPS)}. "
            f"Found keys: {[k for k in span_attrs if k.startswith('genai.usage.')]}"
        )

    if op in _MODEL_OPS and has_tool_args:
        raise GenAIUsageInvariantError(
            f"span op={op!r} carries genai.tool.call.arguments; tool payloads "
            f"belong on execute_tool spans, not on model spans."
        )

    if op not in _MODEL_OPS | _AGGREGATE_OPS:
        raise GenAIUsageInvariantError(
            f"span op={op!r} is not a recognized GenAI operation; refusing to "
            f"attribute usage without a known operation name."
        )


def aggregate_trace_usage(spans: list[dict]) -> dict:
    """Sum usage over leaf model spans only; return per-op breakdown."""
    totals = {"input_tokens": 0, "output_tokens": 0, "model_spans": 0}
    for attrs in spans:
        validate_span_usage(attrs)
        if attrs.get("genai.operation.name") in _MODEL_OPS:
            totals["input_tokens"] += attrs.get("genai.usage.input_tokens", 0)
            totals["output_tokens"] += attrs.get("genai.usage.output_tokens", 0)
            totals["model_spans"] += 1
    if totals["model_spans"] == 0:
        raise GenAIUsageInvariantError(
            "trace contains no model-invocation spans; usage is undefined. "
            "Do not fall back to agent-span usage — it is not authoritative."
        )
    return totals