# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-agent-span-close-its-tool-call-child-before-emitting-token-usage.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-agent-span-close-its-tool-call-child-before-emitting-token-usage.warrant.json
# Markdown SHA-256:   2fa66b9bb8005533ddb88d7e9ea22fa6781f9ffe1188d119a2b591d17e821ad8
# Code AST SHA-256:   51ce199fed5b3424abefab700633dcc6af6bb3992aead12d52b96bea43b01169
# Verified Classes:   SpanOrderingViolation, SpanRecord
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations
from dataclasses import dataclass, field
from typing import Optional


class SpanOrderingViolation(RuntimeError):
    """Raised when GenAI usage emission violates tool-call containment ordering."""


@dataclass
class SpanRecord:
    name: str
    operation: str  # "invoke_agent" | "chat" | "execute_tool"
    start_ns: int
    end_ns: Optional[int] = None
    usage_emitted_ns: Optional[int] = None
    parent_id: Optional[str] = None
    span_id: str = ""


def validate_agent_span_ordering(
    agent: SpanRecord, children: list[SpanRecord]
) -> None:
    """
    Enforce: on an agent span that carries usage, no execute_tool child may
    start before usage emission, and usage emission must not fall inside any
    open tool-span window.
    """
    if agent.usage_emitted_ns is None:
        return  # usage carried elsewhere; invariant does not apply

    tool_children = [c for c in children if c.operation == "execute_tool"]

    for tool in tool_children:
        if tool.start_ns < agent.usage_emitted_ns:
            raise SpanOrderingViolation(
                f"tool span {tool.span_id or tool.name!r} started at "
                f"{tool.start_ns} before agent usage emission at "
                f"{agent.usage_emitted_ns}; output tokens will be "
                f"misattributed to this tool by containment-based processors"
            )
        if tool.end_ns is not None and tool.start_ns <= agent.usage_emitted_ns <= tool.end_ns:
            raise SpanOrderingViolation(
                f"agent usage emission at {agent.usage_emitted_ns} falls inside "
                f"open tool span {tool.span_id or tool.name!r} "
                f"[{tool.start_ns}, {tool.end_ns}]"
            )


def attribute_usage(agent: SpanRecord, children: list[SpanRecord]) -> dict[str, int]:
    """Attribute agent usage to tool calls under the valid ordering only."""
    validate_agent_span_ordering(agent, children)
    # After validation, usage belongs to the model turn that produced the
    # *next* tool call, i.e. the earliest tool child starting after emission.
    eligible = sorted(
        (c for c in children if c.operation == "execute_tool"),
        key=lambda c: c.start_ns,
    )
    if not eligible:
        return {}
    return {eligible[0].span_id or eligible[0].name: 1}