# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/which-invariants-prevent-span-model-collapse-in-opentelemetry-genai-multi-turn-agent-trajectories.md
# Remnant Warrant:    https://corpusfetch.com/c/which-invariants-prevent-span-model-collapse-in-opentelemetry-genai-multi-turn-agent-trajectories.warrant.json
# Markdown SHA-256:   555b22eec83cbc8bc56414a64f14cfe06f5c9581f6629473ba696cc84e3a1ff2
# Code AST SHA-256:   764f08ec5d6104fae201fc82fc14c1dd3c6aa9ba1814b4323d7fa85609a63703
# Verified Classes:   SpanModelCollapseError
# Defensive Guards:   3 fail-closed invariants (zero external deps)

class SpanModelCollapseError(RuntimeError):
    """Raised when a trajectory is modeled as a single mutated span."""
    pass


def validate_trajectory_spans(spans):
    """
    spans: list of dicts with keys:
      span_id, parent_id, operation, attributes (dict), events (list)
    Enforces: one model invocation per chat span; no cumulative token
    mutation across events; tool spans are siblings of turn spans.
    """
    for s in spans:
        if s["operation"] != "chat":
            continue
        token_series = [
            e["attributes"].get("gen_ai.usage.input_tokens")
            for e in s.get("events", [])
            if "gen_ai.usage.input_tokens" in e.get("attributes", {})
        ]
        token_series = [t for t in token_series if t is not None]
        if len(token_series) > 1:
            strictly_increasing = all(
                b >= a for a, b in zip(token_series, token_series[1:])
            )
            if strictly_increasing:
                raise SpanModelCollapseError(
                    f"span {s['span_id']} mutates gen_ai.usage.input_tokens "
                    f"across events ({token_series}); a chat span must map to "
                    f"exactly one model invocation. Emit one span per turn."
                )
        finish = s["attributes"].get("gen_ai.response.finish_reasons")
        if isinstance(finish, list) and len(finish) > 1 and len(set(finish)) > 1:
            raise SpanModelCollapseError(
                f"span {s['span_id']} has heterogeneous finish_reasons {finish} "
                f"indicating multiple turns collapsed into one span."
            )

    by_id = {s["span_id"]: s for s in spans}
    for s in spans:
        if s["operation"] not in ("execute_tool",):
            continue
        parent = by_id.get(s["parent_id"])
        if parent is not None and parent["operation"] == "chat":
            raise SpanModelCollapseError(
                f"tool span {s['span_id']} parented under chat span "
                f"{parent['span_id']}; tool spans must be siblings of turn "
                f"spans to preserve latency decomposition."
            )
    return True