# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-gen-ai-usage-tokens-across-nested-agent-spans-in-opentelemetry-genai-traces.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-double-counting-of-gen-ai-usage-tokens-across-nested-agent-spans-in-opentelemetry-genai-traces.warrant.json
# Markdown SHA-256:   527e821285baabf62302b3be04832ff8907d0f14c1e0efbf5b8e18d60f50d942
# Code AST SHA-256:   a3406da57e6639fbffe793f7a03d43613a0a47499a645309d32696d16686ee50
# Verified Classes:   UsageAggregationInvariantError, SpanModel
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class UsageAggregationInvariantError(RuntimeError):
    """Raised when a span that models a logical turn or agent orchestration
    carries gen_ai.usage.* attributes, which would double-count tokens under
    sum-over-all-spans backends."""


# Attributes that are span-local per the GenAI semantic conventions and must
# never appear on a span that aggregates descendants.
LEAF_ONLY_USAGE_ATTRS = frozenset(
    {"gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens"}
)

# Explicit, non-authoritative aggregate namespace. Distinct from gen_ai.usage.*
# so generic summation never picks it up.
AGGREGATE_NAMESPACE = "agent.turn."


@dataclass
class SpanModel:
    name: str
    attributes: dict[str, object] = field(default_factory=dict)
    has_or_will_have_children: bool = False
    is_physical_model_call: bool = False


def validate_usage_placement(span: SpanModel) -> None:
    """Enforce that gen_ai.usage.* lives only on physical-call (leaf) spans.

    A span that is an orchestrator or a logical turn (i.e., has or will have
    children) must not carry gen_ai.usage.*; it may carry aggregated totals
    under the AGGREGATE_NAMESPACE instead.
    """
    usage_attrs_present = LEAF_ONLY_USAGE_ATTRS & span.attributes.keys()

    if not usage_attrs_present:
        return

    if span.has_or_will_have_children and not span.is_physical_model_call:
        raise UsageAggregationInvariantError(
            f"span {span.name!r} carries {sorted(usage_attrs_present)} but "
            f"also aggregates child spans; this double-counts tokens under "
            f"sum-over-all-spans backends. Move totals to the "
            f"{AGGREGATE_NAMESPACE!r} namespace or emit usage only on leaves."
        )

    # A physical call may legitimately have children (e.g., a streaming span
    # with chunk events), but those children must not themselves emit
    # gen_ai.usage.*; that is enforced at the child's own validation call.


def emit_aggregate(span: SpanModel, input_tokens: int, output_tokens: int) -> None:
    """Record a derived, non-authoritative aggregate on an ancestor span.

    Never writes into gen_ai.usage.*; consumers that sum leaves must be able
    to ignore this namespace entirely.
    """
    span.attributes[f"{AGGREGATE_NAMESPACE}input_tokens_aggregate"] = input_tokens
    span.attributes[f"{AGGREGATE_NAMESPACE}output_tokens_aggregate"] = output_tokens


# Example: the failure this guard is designed to catch.
if __name__ == "__main__":
    orchestrator = SpanModel(
        name="invoke_agent",
        attributes={
            "gen_ai.usage.input_tokens": 8400,   # WRONG: aggregated from children
            "gen_ai.usage.output_tokens": 1200,
        },
        has_or_will_have_children=True,
        is_physical_model_call=False,
    )

    try:
        validate_usage_placement(orchestrator)
    except UsageAggregationInvariantError as exc:
        # Correct remediation: strip leaf-only attrs, emit explicit aggregate.
        for attr in LEAF_ONLY_USAGE_ATTRS:
            orchestrator.attributes.pop(attr, None)
        emit_aggregate(orchestrator, input_tokens=8400, output_tokens=1200)
        validate_usage_placement(orchestrator)  # now passes
        print(f"remediated: {exc}")