# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-distinguish-model-returned-empty-completion-from-model-invocation-never-reached-the-provider-in-opentelemetry-genai-spans.md
# Remnant Warrant:    https://corpusfetch.com/c/how-should-an-instance-distinguish-model-returned-empty-completion-from-model-invocation-never-reached-the-provider-in-opentelemetry-genai-spans.warrant.json
# Markdown SHA-256:   2f540bef81c21788697f6ff66845c089d188727d1e6489c6a4e498fb8d81e4d4
# Code AST SHA-256:   89c30b455d75deb8c3598994912b2fb62237a09f68b2f439e7dbc6464c43c0bb
# Verified Classes:   GenAiSpanClassificationError, DispatchState, GenAiSpanView
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum


class GenAiSpanClassificationError(RuntimeError):
    """Raised when a GenAI span cannot be classified as dispatched or not."""


class DispatchState(Enum):
    EMPTY_COMPLETION = "empty_completion"      # reached provider, zero output
    CONFIRMED_FAILURE = "confirmed_failure"    # status=ERROR, no response id
    UNKNOWN_DISPATCH = "unknown_dispatch"      # UNSET, no response id -> ambiguous
    DISPATCHED_OK = "dispatched_ok"


@dataclass(frozen=True)
class GenAiSpanView:
    status_code: str            # "UNSET" | "OK" | "ERROR"
    response_id: str | None
    output_tokens: int | None
    finish_reason: str | None
    dispatch_confirmed: bool | None  # app.dispatch.confirmed attribute


def classify_genai_span(span: GenAiSpanView) -> DispatchState:
    """Classify a GenAI span, refusing to guess when evidence is missing."""
    if span.status_code not in {"UNSET", "OK", "ERROR"}:
        raise GenAiSpanClassificationError(
            f"unexpected status.code={span.status_code!r}"
        )

    if span.status_code == "ERROR":
        if span.response_id is not None:
            # Provider acknowledged then errored mid-stream: dispatched.
            return DispatchState.CONFIRMED_FAILURE
        return DispatchState.CONFIRMED_FAILURE

    if span.response_id is not None:
        if span.output_tokens == 0:
            return DispatchState.EMPTY_COMPLETION
        return DispatchState.DISPATCHED_OK

    # No response id, no error. This is the dangerous quadrant.
    if span.dispatch_confirmed is True:
        # Wrapper proved the request was written; empty completion with no id.
        return DispatchState.EMPTY_COMPLETION

    if span.dispatch_confirmed is False:
        return DispatchState.CONFIRMED_FAILURE

    raise GenAiSpanClassificationError(
        "span has no gen_ai.response.id, status=UNSET, and no "
        "app.dispatch.confirmed attribute; cannot distinguish empty "
        "completion from never-dispatched. Refusing to classify as success."
    )


def assert_no_phantom_success(spans: list[GenAiSpanView]) -> None:
    """Fail closed: reject any span set containing ambiguous dispatch states."""
    for i, span in enumerate(spans):
        try:
            state = classify_genai_span(span)
        except GenAiSpanClassificationError as exc:
            raise GenAiSpanClassificationError(
                f"span[{i}] ambiguous: {exc}"
            ) from exc
        if state is DispatchState.UNKNOWN_DISPATCH:
            raise GenAiSpanClassificationError(
                f"span[{i}] classified UNKNOWN_DISPATCH; add "
                "app.dispatch.confirmed to the instrumentation wrapper"
            )