# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-reports-genai-token-usage-on-agent-spans-without-matching-tool-call-attributes-a-silent-double-counting-boundary-in-open.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-reports-genai-token-usage-on-agent-spans-without-matching-tool-call-attributes-a-silent-double-counting-boundary-in-open.warrant.json
# Markdown SHA-256:   6fe13d856472ab26cbd50edbdb41e437b002ec57a13d0f144ab35e0d86087e6c
# Code AST SHA-256:   9eb4a5fb858ecfab40e7cb5af5cd6fb85366d007a0ba1d8fb74c0c9a50e3f787
# Verified Classes:   GenAISemconvViolation, SpanRecord
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


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


@dataclass
class SpanRecord:
    span_id: str
    parent_id: str | None
    operation_name: str  # gen_ai.operation.name
    attributes: dict[str, object] = field(default_factory=dict)

    def has_usage(self) -> bool:
        return any(
            k in self.attributes
            for k in ("gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens")
        )


def _descendants(span: SpanRecord, all_spans: list[SpanRecord]) -> list[SpanRecord]:
    children = [s for s in all_spans if s.parent_id == span.span_id]
    out: list[SpanRecord] = []
    for c in children:
        out.append(c)
        out.extend(_descendants(c, all_spans))
    return out


def validate_usage_attribution(spans: list[SpanRecord]) -> None:
    """Enforce: a span carrying gen_ai.usage.* must have no descendant that also carries it.

    A violation means a trace-wide sum of usage attributes will double-count.
    """
    for span in spans:
        if not span.has_usage():
            continue
        for desc in _descendants(span, spans):
            if desc.has_usage():
                raise GenAISemconvViolation(
                    f"span {span.span_id} ({span.operation_name}) carries gen_ai.usage.* "
                    f"but descendant {desc.span_id} ({desc.operation_name}) also carries it; "
                    f"trace-wide usage sum will double-count. Set usage on leaves only."
                )


def validate_tool_call_attribution(spans: list[SpanRecord]) -> None:
    """Enforce: each execute_tool span names exactly one tool and one call id."""
    for span in spans:
        if span.operation_name != "execute_tool":
            continue
        name = span.attributes.get("gen_ai.tool.name")
        call_id = span.attributes.get("gen_ai.tool.call.id")
        if not isinstance(name, str) or not name:
            raise GenAISemconvViolation(
                f"execute_tool span {span.span_id} missing gen_ai.tool.name"
            )
        if any(sep in name for sep in (",", ";", " ")):
            raise GenAISemconvViolation(
                f"execute_tool span {span.span_id} has batched tool name {name!r}; "
                f"emit one span per gen_ai.tool.call.id"
            )
        if not isinstance(call_id, str) or not call_id:
            raise GenAISemconvViolation(
                f"execute_tool span {span.span_id} missing gen_ai.tool.call.id"
            )


def validate_trace(spans: list[SpanRecord]) -> None:
    validate_usage_attribution(spans)
    validate_tool_call_attribution(spans)