# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-of-a-genai-agent-span-is-closed-before-its-tool-call-child-spans-the-opentelemetry-genai-semantic-convention-invariant-t.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-of-a-genai-agent-span-is-closed-before-its-tool-call-child-spans-the-opentelemetry-genai-semantic-convention-invariant-t.warrant.json
# Markdown SHA-256:   50ec54ad36277eff47070ad071e234561d3bd2218a7dccf3b4ab0d0e108cba7f
# Code AST SHA-256:   e8ab3c92d3cc6d9cbeaebb04cf09ea02a87e1da7cb823e83c4ff358ae107c4a6
# Verified Classes:   TemporalContainmentError, SpanInterval
# Defensive Guards:   5 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass


class TemporalContainmentError(RuntimeError):
    """Raised when a GenAI child span outlives its enclosing agent span.

    Under OpenTelemetry GenAI semantic conventions, an `execute_tool` span
    that is a child of an `invoke_agent` span must have start_time >=
    parent.start_time and end_time <= parent.end_time. Structural parentage
    (parent_span_id) is enforced by the SDK; temporal containment is not.
    """


@dataclass(frozen=True)
class SpanInterval:
    span_id: str
    name: str
    parent_span_id: str | None
    start_ns: int
    end_ns: int


def assert_temporal_containment(
    agent: SpanInterval,
    tool: SpanInterval,
) -> None:
    """Validate that `tool` is temporally contained in `agent`.

    Call this before shipping the trace (or in the eval pipeline) to catch
    async-dispatch bugs where the agent span is ended while tool tasks are
    still in flight.
    """
    if tool.parent_span_id != agent.span_id:
        raise TemporalContainmentError(
            f"tool span {tool.span_id!r} is not a child of agent span "
            f"{agent.span_id!r} (parent={tool.parent_span_id!r}); "
            "check that the context was captured at dispatch time"
        )

    if tool.start_ns < agent.start_ns:
        raise TemporalContainmentError(
            f"tool span {tool.span_id!r} starts at {tool.start_ns} before "
            f"agent span {agent.span_id!r} starts at {agent.start_ns}; "
            "tool must be dispatched after the agent span begins"
        )

    if tool.end_ns > agent.end_ns:
        raise TemporalContainmentError(
            f"tool span {tool.span_id!r} ends at {tool.end_ns} after agent "
            f"span {agent.span_id!r} ends at {agent.end_ns} "
            f"(tail={tool.end_ns - agent.end_ns} ns). "
            "Either await outstanding tool tasks before span.end(), or if "
            "the tool is genuinely fire-and-forget, reparent it to the trace "
            "root and add a link to the agent span instead of a parent edge."
        )


def validate_agent_trace(spans: list[SpanInterval]) -> None:
    """Validate all execute_tool spans against their invoke_agent ancestors."""
    by_id = {s.span_id: s for s in spans}
    for span in spans:
        if span.name != "execute_tool":
            continue
        # Walk up to the nearest invoke_agent ancestor.
        cursor = span.parent_span_id
        while cursor is not None:
            ancestor = by_id.get(cursor)
            if ancestor is None:
                raise TemporalContainmentError(
                    f"tool span {span.span_id!r} references missing ancestor "
                    f"{cursor!r}; trace is incomplete"
                )
            if ancestor.name == "invoke_agent":
                assert_temporal_containment(ancestor, span)
                break
            cursor = ancestor.parent_span_id
        else:
            raise TemporalContainmentError(
                f"tool span {span.span_id!r} has no invoke_agent ancestor; "
                "under GenAI conventions execute_tool must be scoped to an "
                "agent invocation or linked to the trace root"
            )