# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-of-gen-ai-tool-call-end-the-span-closure-invariant-that-opentelemetry-genai-semantic-conventions-leave-underspecified-for-agent.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-of-gen-ai-tool-call-end-the-span-closure-invariant-that-opentelemetry-genai-semantic-conventions-leave-underspecified-for-agent.warrant.json
# Markdown SHA-256:   e08bea072552b8f1d47f50b48b8770f0927e3a9b2503893f5360826bfdb1412f
# Code AST SHA-256:   61f0d6327cdf9f511dcaa76df336b2862d27085fe721da10af3801c1f6acb969
# Verified Classes:   GenAiSpanClosureError, SpanRecord, AgentSpanGuard
# Defensive Guards:   6 fail-closed invariants (zero external deps)

from __future__ import annotations

import time
from dataclasses import dataclass, field


class GenAiSpanClosureError(RuntimeError):
    """Raised when a GenAI agent span violates lifetime containment or usage reconciliation."""


@dataclass
class SpanRecord:
    span_id: str
    parent_span_id: str | None
    operation: str
    start_time: float
    end_time: float | None = None
    tool_call_id: str | None = None
    input_tokens: int = 0
    output_tokens: int = 0


@dataclass
class AgentSpanGuard:
    agent: SpanRecord
    children: list[SpanRecord] = field(default_factory=list)

    def add_child(self, child: SpanRecord) -> None:
        if child.parent_span_id != self.agent.span_id:
            raise GenAiSpanClosureError(
                f"child {child.span_id} is not parented to agent {self.agent.span_id}"
            )
        self.children.append(child)

    def close_agent(self, now: float | None = None) -> None:
        now = time.time() if now is None else now
        self.agent.end_time = now

        # Invariant 1: lifetime containment — every child must end before the agent.
        for child in self.children:
            if child.end_time is None:
                raise GenAiSpanClosureError(
                    f"agent {self.agent.span_id} closed with unfinished child "
                    f"{child.span_id} ({child.operation}); tool span outlived parent"
                )
            if child.end_time > self.agent.end_time:
                raise GenAiSpanClosureError(
                    f"child {child.span_id} end_time {child.end_time:.3f} exceeds "
                    f"agent end_time {self.agent.end_time:.3f}; "
                    f"containment violated by {child.end_time - self.agent.end_time:.3f}s"
                )

        # Invariant 2: usage reconciliation — parent usage must cover child usage.
        child_input = sum(c.input_tokens for c in self.children)
        child_output = sum(c.output_tokens for c in self.children)
        if child_input > self.agent.input_tokens:
            raise GenAiSpanClosureError(
                f"agent input_tokens {self.agent.input_tokens} < child sum {child_input}; "
                f"retry likely reused gen_ai.tool.call.id"
            )
        if child_output > self.agent.output_tokens:
            raise GenAiSpanClosureError(
                f"agent output_tokens {self.agent.output_tokens} < child sum {child_output}"
            )

        # Invariant 3: tool call id uniqueness per attempt.
        seen: set[str] = set()
        for child in self.children:
            if child.tool_call_id is None:
                continue
            if child.tool_call_id in seen:
                raise GenAiSpanClosureError(
                    f"duplicate gen_ai.tool.call.id {child.tool_call_id!r} in trace; "
                    f"retries must mint a new call id"
                )
            seen.add(child.tool_call_id)