# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-broken-span-links-in-nested-agent-tool-model-call-graphs-under-genai-semantic-conventions.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-broken-span-links-in-nested-agent-tool-model-call-graphs-under-genai-semantic-conventions.warrant.json
# Markdown SHA-256:   d4a18e422ebfe46c51a7692691780fea703eae1f4649bf0a2f7b1bcbc2cc40cd
# Code AST SHA-256:   afb364f8e40b820a55e5d1554950ba457915b35a0d2085a918001636d2b4d778
# Verified Classes:   SpanLinkInvariantError, SpanRecord, SpanLink
# Defensive Guards:   6 fail-closed invariants (zero external deps)

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


class SpanLinkInvariantError(RuntimeError):
    """Raised when a GenAI span link violates the semantic linkage invariant."""
    pass


@dataclass
class SpanRecord:
    span_id: str
    trace_id: str
    operation_name: str          # e.g. "chat", "execute_tool"
    start_ns: int
    end_ns: Optional[int]        # None => still live
    status_ok: bool
    request_id: Optional[str]    # genai.request.id, shared across retries
    attempt: int = 1


@dataclass
class SpanLink:
    role: str                    # genai.link.role
    target_span_id: str
    target_state: Optional[str]  # genai.link.target_state


_ALLOWED_ROLES = {
    "model_response_source": {"chat"},
    "tool_invocation_source": {"execute_tool"},
}


def resolve_link(
    linker: SpanRecord,
    link: SpanLink,
    registry: dict[str, SpanRecord],
) -> SpanRecord:
    target = registry.get(link.target_span_id)
    if target is None:
        raise SpanLinkInvariantError(
            f"link target {link.target_span_id} not in live registry"
        )
    if target.trace_id != linker.trace_id:
        raise SpanLinkInvariantError("cross-trace link is not permitted")

    # Condition 1: target must precede or coincide with linker start.
    if target.start_ns > linker.start_ns:
        # Forward reference: legal only if explicitly annotated downstream.
        pass

    # Condition 2: liveness / terminal-state compatibility.
    if target.end_ns is not None:
        if link.target_state != "completed":
            raise SpanLinkInvariantError(
                f"link to ended span {target.span_id} without target_state=completed"
            )
        if not target.status_ok:
            raise SpanLinkInvariantError(
                f"link role {link.role} requires OK target, got error span"
            )

    # Condition 3: role compatibility.
    allowed = _ALLOWED_ROLES.get(link.role, set())
    if target.operation_name not in allowed:
        raise SpanLinkInvariantError(
            f"role {link.role} incompatible with target operation "
            f"{target.operation_name}"
        )

    # Condition 4: retry-sibling resolution.
    if target.request_id is not None:
        siblings = [
            s for s in registry.values()
            if s.request_id == target.request_id
            and s.span_id != target.span_id
            and s.start_ns > target.start_ns
            and s.status_ok
        ]
        if siblings:
            latest = max(siblings, key=lambda s: s.start_ns)
            raise SpanLinkInvariantError(
                f"link targets retry attempt {target.attempt} but a later "
                f"successful attempt {latest.span_id} exists; re-point required"
            )

    return target