# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-genai-semantic-convention-conformance-in-agent-observability-pipelines-is-actually-being-tested-rather-than-merely-in.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-genai-semantic-convention-conformance-in-agent-observability-pipelines-is-actually-being-tested-rather-than-merely-in.warrant.json
# Markdown SHA-256:   efd6a6b2bc0b3f7229cbab0e0aaaf30d71859c0c888c0f882bb2619f76a8393f
# Code AST SHA-256:   6d45a8050cfeb12ca62123972f622af569c0f9f2f33f8002f8fb2bc05f3bf575
# Verified Classes:   ConformanceStructuralError, Span
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field


class ConformanceStructuralError(RuntimeError):
    """Raised when a trace passes attribute-domain checks but is structurally
    unreconstructible (orphaned parents, unreconciled tokens, order-dependent
    event pairs, or unbounded span-name cardinality)."""


@dataclass
class Span:
    span_id: str
    parent_span_id: str | None
    name: str
    attributes: dict[str, object] = field(default_factory=dict)
    events: list[dict[str, object]] = field(default_factory=list)


def assert_structurally_conformant(
    spans: list[Span],
    *,
    span_name_cardinality_budget: int = 500,
    token_tolerance: int = 0,
) -> None:
    """Structural conformance gate. Assumes per-span attribute-domain checks
    have already passed; this function only checks the invariants that
    attribute-domain checks cannot see."""

    by_id = {s.span_id: s for s in spans}

    # (a) Parent resolution: every non-root parent must exist in the trace.
    for span in spans:
        if span.parent_span_id is not None and span.parent_span_id not in by_id:
            raise ConformanceStructuralError(
                f"orphaned span {span.span_id!r} ({span.name!r}): "
                f"parent {span.parent_span_id!r} not present in exported trace"
            )

    # (b) Token reconciliation: repeated response ids must not double-count,
    #     and child usage must sum to any parent aggregate within tolerance.
    seen_response_ids: dict[str, str] = {}
    for span in spans:
        rid = span.attributes.get("gen_ai.response.id")
        if rid is not None:
            if rid in seen_response_ids:
                raise ConformanceStructuralError(
                    f"duplicate gen_ai.response.id {rid!r} emitted by spans "
                    f"{seen_response_ids[rid]!r} and {span.span_id!r} with no "
                    f"retry marker; token sums will overcount"
                )
            seen_response_ids[rid] = span.span_id

    # (c) Event-pair order-robustness: argument/result events must be
    #     set-equal, not merely present, so collector reordering is caught.
    for span in spans:
        args = [e for e in span.events if e.get("name") == "gen_ai.tool.call.arguments"]
        results = [e for e in span.events if e.get("name") == "gen_ai.tool.call.result"]
        if len(args) != len(results):
            raise ConformanceStructuralError(
                f"span {span.span_id!r}: {len(args)} argument events vs "
                f"{len(results)} result events; causal pairing unreconstructible"
            )

    # (d) Span-name cardinality against the indexer dimension budget.
    distinct_names = {s.name for s in spans}
    if len(distinct_names) > span_name_cardinality_budget:
        raise ConformanceStructuralError(
            f"span-name cardinality {len(distinct_names)} exceeds budget "
            f"{span_name_cardinality_budget}; indexer will drop spans"
        )