# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-double-counting-and-span-nesting-corruption-in-opentelemetry-genai-agent-spans-under-retry-and-fan-out.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-double-counting-and-span-nesting-corruption-in-opentelemetry-genai-agent-spans-under-retry-and-fan-out.warrant.json
# Markdown SHA-256:   1594247685fb7e28bea6b84f033d0fafafe801681caed9d05ed8fdae8c8f8e53
# Code AST SHA-256:   c6d01d18ebe279753ff2ed09f3e5df2008a30da3c84ac1b38e6e0365ee6ddfcc
# Verified Classes:   GenAiSpanInvariantError, GenAiSpan
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class GenAiSpanInvariantError(RuntimeError):
    """Raised when a GenAI span violates leaf-local accounting or call-id invariants."""


@dataclass
class GenAiSpan:
    span_id: str
    operation_name: str  # "chat" | "execute_tool" | "invoke_agent"
    parent_id: str | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None
    tool_call_id: str | None = None
    attempt: int = 1
    status_ok: bool = True
    children: list["GenAiSpan"] = field(default_factory=list)


def validate_trace(spans: list[GenAiSpan]) -> None:
    by_id = {s.span_id: s for s in spans}

    # Rule 1: usage attributes are leaf-local; a span with children must not
    # carry gen_ai.usage.* (prevents double-counting on rollup).
    for span in spans:
        if span.children and (span.input_tokens is not None or span.output_tokens is not None):
            raise GenAiSpanInvariantError(
                f"span {span.span_id} ({span.operation_name}) carries usage attributes "
                f"but has {len(span.children)} children; usage must be leaf-local"
            )

    # Rule 2: tool-call identity is (call_id, attempt); duplicates corrupt dedup.
    seen: set[tuple[str, int]] = set()
    for span in spans:
        if span.operation_name == "execute_tool":
            if span.tool_call_id is None:
                raise GenAiSpanInvariantError(
                    f"execute_tool span {span.span_id} missing gen_ai.tool.call.id"
                )
            key = (span.tool_call_id, span.attempt)
            if key in seen:
                raise GenAiSpanInvariantError(
                    f"duplicate tool attempt {key}: retries must increment 'attempt' "
                    f"while preserving tool.call.id"
                )
            seen.add(key)

    # Rule 3: parent linkage must resolve.
    for span in spans:
        if span.parent_id is not None and span.parent_id not in by_id:
            raise GenAiSpanInvariantError(
                f"span {span.span_id} references unknown parent {span.parent_id}"
            )


def rollup_input_tokens(spans: list[GenAiSpan]) -> int:
    """Hierarchy-aware rollup: sum only leaf, successful attempts."""
    validate_trace(spans)
    total = 0
    for span in spans:
        if span.children:
            continue
        if not span.status_ok:
            continue
        total += span.input_tokens or 0
    return total