# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-decide-to-drop-a-span-the-tail-sampling-invariant-that-cost-aware-retention-silently-violates.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-decide-to-drop-a-span-the-tail-sampling-invariant-that-cost-aware-retention-silently-violates.warrant.json
# Markdown SHA-256:   e3693c3ae2b1d6cc0ba4ac1c82e6729b76f1ac863b7b0dd68adc7cf7276ee190
# Code AST SHA-256:   7d9300d1553ea387d99167c60b2a52adbde6155676de98a9958bb4a511b824e8
# Verified Classes:   RetentionInvariantError, RetentionConfig
# Defensive Guards:   4 fail-closed invariants (zero external deps)

"""Guard: reject span-retention configurations that violate the
trace-admission / span-retention independence invariant."""

from dataclasses import dataclass


class RetentionInvariantError(RuntimeError):
    """Raised when a retention config makes span retention depend on payload."""


@dataclass(frozen=True)
class RetentionConfig:
    trace_admission_rate: float          # e.g. 1.0 for "all errors"
    span_drop_rule: str                  # "uniform" | "cost_threshold"
    per_span_budget_tokens: int | None
    cost_estimator_source: str           # must not be the span itself
    raw_fidelity_sink: bool


def validate_retention(cfg: RetentionConfig) -> None:
    if not (0.0 <= cfg.trace_admission_rate <= 1.0):
        raise RetentionInvariantError(
            f"trace_admission_rate out of range: {cfg.trace_admission_rate}"
        )

    # The core invariant: a trace-level completeness guarantee is only
    # sound if span retention is content-blind.
    if cfg.trace_admission_rate >= 1.0 and cfg.span_drop_rule != "uniform":
        if not cfg.raw_fidelity_sink:
            raise RetentionInvariantError(
                "trace_admission_rate=1.0 with content-aware span dropping "
                f"({cfg.span_drop_rule!r}) and no raw-fidelity sink: the "
                "completeness guarantee is false for high-payload spans. "
                "Either switch span_drop_rule to 'uniform' or provision a "
                "raw_fidelity_sink for incident replay."
            )

    # Cost estimator must not read the span being retained.
    if cfg.cost_estimator_source in {"span_payload", "self", "serialized_span"}:
        raise RetentionInvariantError(
            "cost_estimator_source reads the span under decision; the drop "
            "boundary shifts with serialization changes. Pin to an "
            "independent accounting source."
        )

    if cfg.span_drop_rule == "cost_threshold" and cfg.per_span_budget_tokens is None:
        raise RetentionInvariantError(
            "cost_threshold rule requires per_span_budget_tokens"
        )


if __name__ == "__main__":
    # Reproduces the observed failure: 100% error admission + cost-threshold
    # span dropping + no raw sink -> invariant violation.
    bad = RetentionConfig(
        trace_admission_rate=1.0,
        span_drop_rule="cost_threshold",
        per_span_budget_tokens=2048,
        cost_estimator_source="span_payload",
        raw_fidelity_sink=False,
    )
    validate_retention(bad)  # raises RetentionInvariantError