# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-terminate-a-genai-agent-span-whose-gen-ai-usage-input-tokens-was-reported-as-zero-by-the-model-backen.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-refuse-to-terminate-a-genai-agent-span-whose-gen-ai-usage-input-tokens-was-reported-as-zero-by-the-model-backen.warrant.json
# Markdown SHA-256:   61719b5e3f338f4ca8dc1d0adc2ef2c45690c313b79a7178970472afd6c2b85d
# Code AST SHA-256:   551a0d059aff28357f05f6982e8bfe0125d811c97ba440562503294bf4b9e720
# Verified Classes:   GenAIUsageAccountingError, SpanUsageSnapshot
# Defensive Guards:   4 fail-closed invariants (zero external deps)

"""Validator enforcing the GenAI agent-span token-accounting invariant.

Invariant: a chat/agent span whose backend contractually reports usage MUST NOT
close as OK with zero input AND zero output tokens while carrying a non-empty
completion. Absent usage -> omit attributes. Contradictory usage -> quarantine.
"""

from dataclasses import dataclass
from typing import Optional


class GenAIUsageAccountingError(RuntimeError):
    """Raised when a span's token attributes contradict its completion content."""


@dataclass
class SpanUsageSnapshot:
    usage_block_present: bool          # backend contractually reported usage
    input_tokens: Optional[int]        # None means attribute omitted, not zero
    output_tokens: Optional[int]
    completion_nonempty: bool
    finish_reason: str                 # "stop" | "tool_calls" | "length" | "error" | ...


_TERMINAL_OK = {"stop", "tool_calls", "length"}


def validate_agent_span_usage(snap: SpanUsageSnapshot) -> None:
    """Enforce the invariant. Raises on contradictory state; returns otherwise.

    - Absent usage block: attributes MUST be omitted (None), never 0.
    - Present usage block with a completion: at least one token count must be > 0.
    - Error finish reasons are exempt from the nonzero rule.
    """
    if not snap.usage_block_present:
        # Absent usage: the only legal encoding is omission.
        if snap.input_tokens is not None or snap.output_tokens is not None:
            raise GenAIUsageAccountingError(
                "usage block absent but token attributes were set "
                f"(input={snap.input_tokens!r}, output={snap.output_tokens!r}); "
                "omit the attributes instead of fabricating 0"
            )
        return

    if snap.finish_reason not in _TERMINAL_OK:
        # Error/content_filter: zero usage may be legitimate; do not quarantine here.
        return

    inp = snap.input_tokens if snap.input_tokens is not None else 0
    out = snap.output_tokens if snap.output_tokens is not None else 0

    if snap.completion_nonempty and inp == 0 and out == 0:
        raise GenAIUsageAccountingError(
            "contradictory usage: backend reported usage, completion is non-empty, "
            f"yet input_tokens=0 and output_tokens=0 (finish_reason={snap.finish_reason!r}); "
            "quarantine span as ERROR with gen_ai.usage.accounting_error"
        )

    if snap.completion_nonempty and out == 0:
        raise GenAIUsageAccountingError(
            "contradictory usage: non-empty completion implies output_tokens > 0, "
            f"got output_tokens={snap.output_tokens!r}"
        )


def close_agent_span(span, snap: SpanUsageSnapshot) -> None:
    """Close a span, applying the invariant. On violation, quarantine as ERROR."""
    try:
        validate_agent_span_usage(snap)
    except GenAIUsageAccountingError as exc:
        span.set_attribute("gen_ai.usage.accounting_error", str(exc))
        span.set_status(status="ERROR", description=str(exc))
        span.end()
        raise
    span.set_status(status="OK")
    span.end()


# Example call site (span is an OTel Span-like object):
# close_agent_span(span, SpanUsageSnapshot(
#     usage_block_present=True, input_tokens=0, output_tokens=0,
#     completion_nonempty=True, finish_reason="stop",
# ))  # -> raises GenAIUsageAccountingError, span quarantined