# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-refuse-to-emit-a-genai-agent-span-double-counted-tokens-and-orphaned-tool-call-attributes-at-the-span-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-refuse-to-emit-a-genai-agent-span-double-counted-tokens-and-orphaned-tool-call-attributes-at-the-span-boundary.warrant.json
# Markdown SHA-256:   7bd501489cbbd9c780b7a313a7fcb330edb051862367364e73a6c071be8bb921
# Code AST SHA-256:   da32a7a1104970743aea1ba3daa98f985d7c4fdec8652cc45dfe252aba1282c4
# Verified Classes:   GenAISpanInvariantError, GenAISpan
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class GenAISpanInvariantError(RuntimeError):
    """Raised when a GenAI span violates usage or tool-call join invariants."""


@dataclass
class GenAISpan:
    operation_name: str
    has_descendant_usage: bool = False
    input_tokens: Optional[int] = None
    output_tokens: Optional[int] = None
    tool_call_id: Optional[str] = None
    tool_call_id_origin: Optional[str] = None  # "model" | "local"
    originating_span_present: bool = True


AGENT_OPERATIONS = {"invoke_agent", "create_agent", "execute_agent"}


def validate_before_export(span: GenAISpan, consumer_aggregates_across_spans: bool) -> None:
    """Enforce the GenAI span boundary invariants at finalization time."""

    # Invariant 1: agent spans must not re-emit descendant usage when the
    # consumer sums usage across all spans in the trace.
    emits_usage = span.input_tokens is not None or span.output_tokens is not None
    if (
        span.operation_name in AGENT_OPERATIONS
        and span.has_descendant_usage
        and emits_usage
        and consumer_aggregates_across_spans
    ):
        raise GenAISpanInvariantError(
            f"agent span '{span.operation_name}' emits usage "
            f"(in={span.input_tokens}, out={span.output_tokens}) while descendants "
            f"also emit usage and consumer aggregates across spans; "
            f"this double-counts tokens by nesting depth"
        )

    # Invariant 2: a tool call ID must be joinable to its originating model span.
    if span.tool_call_id is not None:
        if span.tool_call_id_origin == "local":
            raise GenAISpanInvariantError(
                f"tool span carries locally-minted gen_ai.tool.call.id="
                f"'{span.tool_call_id}'; this looks joinable but is not. "
                f"Omit the attribute or propagate the model-assigned ID."
            )
        if span.tool_call_id_origin == "model" and not span.originating_span_present:
            raise GenAISpanInvariantError(
                f"tool span references model-assigned gen_ai.tool.call.id="
                f"'{span.tool_call_id}' but the originating model span is absent "
                f"from the trace; the tool-call attribute will be orphaned"
            )
