# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-treat-tool-call-spans-whose-attributes-straddle-the-genai-semantic-conventions-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-treat-tool-call-spans-whose-attributes-straddle-the-genai-semantic-conventions-boundary.warrant.json
# Markdown SHA-256:   18f389c4b611fcfebceabcf94767c76234b75f775b374818e943ecf7d5a10fe9
# Code AST SHA-256:   725cbdcd56e58e8c113ac0d71391de1764dff04b1137868db0c17067ddd7ee09
# Verified Classes:   GenAISpanContractError, SpanVerdict
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from typing import Mapping


class GenAISpanContractError(RuntimeError):
    """Raised when a span's gen_ai.* attributes violate the operation-closure invariant."""


# Attributes admissible per operation family. Anything outside the declared
# family's set is a closure violation and must not reach the cost pipeline.
_TOOL_ONLY = frozenset({
    "gen_ai.tool.name",
    "gen_ai.tool.call.id",
    "gen_ai.tool.description",
})
_LLM_ONLY = frozenset({
    "gen_ai.request.model",
    "gen_ai.response.model",
    "gen_ai.usage.input_tokens",
    "gen_ai.usage.output_tokens",
    "gen_ai.response.finish_reasons",
})
_COMMON = frozenset({
    "gen_ai.operation.name",
    "gen_ai.system",
    "gen_ai.agent.id",
    "gen_ai.agent.name",
})

_ALLOWED: Mapping[str, frozenset[str]] = {
    "execute_tool": _COMMON | _TOOL_ONLY,
    "invoke_agent": _COMMON,
    "create_agent": _COMMON,
    "chat": _COMMON | _LLM_ONLY,
    "text_completion": _COMMON | _LLM_ONLY,
    "embeddings": _COMMON | _LLM_ONLY,
}


@dataclass(frozen=True)
class SpanVerdict:
    accept: bool
    flag_duplicate_call_id: bool
    reason: str


def classify_genai_span(
    attributes: Mapping[str, str | int | float],
    *,
    seen_tool_call_ids: set[str],
) -> SpanVerdict:
    genai_keys = {k for k in attributes if k.startswith("gen_ai.")}
    if not genai_keys:
        return SpanVerdict(True, False, "no gen_ai attributes; not in scope")

    op = attributes.get("gen_ai.operation.name")
    if op is None:
        # Ambiguous: gen_ai.* present but no discriminant. Never guess.
        raise GenAISpanContractError(
            f"gen_ai attributes present without gen_ai.operation.name: "
            f"{sorted(genai_keys)}"
        )

    allowed = _ALLOWED.get(str(op))
    if allowed is None:
        raise GenAISpanContractError(f"unknown gen_ai.operation.name={op!r}")

    extra = genai_keys - allowed
    if extra:
        # Closure violation, e.g. usage.* bled onto an execute_tool span.
        raise GenAISpanContractError(
            f"operation={op!r} carries out-of-family attributes {sorted(extra)}; "
            f"admissible set is {sorted(allowed)}"
        )

    dup = False
    if op == "execute_tool":
        call_id = attributes.get("gen_ai.tool.call.id")
        if call_id is not None:
            dup = str(call_id) in seen_tool_call_ids
            seen_tool_call_ids.add(str(call_id))
    elif "gen_ai.tool.call.id" in genai_keys:
        raise GenAISpanContractError(
            f"gen_ai.tool.call.id present on operation={op!r}; only execute_tool may carry it"
        )

    return SpanVerdict(True, dup, "closure satisfied" + (" (retry)" if dup else ""))