# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-reject-an-opentelemetry-genai-agent-span-whose-tool-call-token-accounting-has-drifted-from-its-par.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-reject-an-opentelemetry-genai-agent-span-whose-tool-call-token-accounting-has-drifted-from-its-par.warrant.json
# Markdown SHA-256:   bb2cb0c885bce1339228fd49c3d8a167cd0f7bf9f09bd68c8767c713ab90361b
# Code AST SHA-256:   7fe52ad65480453988868c3d06735a57d7b5e286884b439f0d7b9016d814158d
# Verified Classes:   GenAISpanAccountingError, Span
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field


class GenAISpanAccountingError(RuntimeError):
    """Raised when a GenAI agent span violates the additive-leaf token invariant."""


@dataclass
class Span:
    name: str
    operation: str  # "invoke_agent" | "execute_tool" | "chat"
    input_tokens: int | None = None
    output_tokens: int | None = None
    children: list["Span"] = field(default_factory=list)

    def descendants(self):
        for c in self.children:
            yield c
            yield from c.descendants()


def _usage(span: Span) -> tuple[int, int]:
    # Absence and explicit zero are equivalent per the semantic conventions.
    return (span.input_tokens or 0, span.output_tokens or 0)


def validate_agent_trace(root: Span) -> None:
    if root.operation != "invoke_agent":
        raise GenAISpanAccountingError(
            f"trace root must be invoke_agent, got {root.operation!r}"
        )

    roots = [
        s for s in [root, *root.descendants()]
        if s.operation == "invoke_agent" and s is not root
    ]
    # Nested invoke_agent spans are allowed only as sub-agents; flag duplicate
    # same-depth roots only if the caller passed a forest root by mistake.
    if any(s.name == root.name for s in roots):
        raise GenAISpanAccountingError(
            f"ambiguous rollup root: multiple invoke_agent spans named {root.name!r}"
        )

    for span in [root, *root.descendants()]:
        if span.operation != "execute_tool":
            continue
        tin, tout = _usage(span)
        if tin == 0 and tout == 0:
            continue
        for chat in span.descendants():
            if chat.operation != "chat":
                continue
            cin, cout = _usage(chat)
            if (tin, tout) == (cin, cout) and (cin or cout):
                raise GenAISpanAccountingError(
                    f"copy-up signature on execute_tool span {span.name!r}: "
                    f"usage ({tin},{tout}) duplicates nested chat span "
                    f"{chat.name!r}; this guarantees double-counting in tree sums"
                )

    root_in, root_out = _usage(root)
    direct_chat_out = sum(
        _usage(c)[1] for c in root.children if c.operation == "chat"
    )
    if root_out < direct_chat_out:
        raise GenAISpanAccountingError(
            f"parent {root.name!r} output_tokens={root_out} is less than "
            f"sum of direct chat children={direct_chat_out}"
        )