# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-escalate-taint-quarantined-retrieval-output-to-a-typed-failure-rather-than-silently-downgrade-it.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-escalate-taint-quarantined-retrieval-output-to-a-typed-failure-rather-than-silently-downgrade-it.warrant.json
# Markdown SHA-256:   aea1223e4434001258a7b894db65a58ae1e1a7d63a2ce6b9b27abe1a68c5374c
# Code AST SHA-256:   e3fe0f6a2c8f74100f6409116487a46d9f552aa5fbaf16c9449061e883fc7749
# Verified Classes:   TaintLabel, SilentTaintDowngradeError, Tainted
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass
from enum import Enum


class TaintLabel(Enum):
    TRUSTED = "trusted"
    USER = "user"
    UNTRUSTED_RETRIEVAL = "untrusted_retrieval"


class SilentTaintDowngradeError(RuntimeError):
    """Raised when a sink would consume a tainted value without an explicit
    declassification transform. This is the typed failure that replaces the
    silent coercion observed in the incident."""

    def __init__(self, label: TaintLabel, sink: str, transform_id: str | None):
        self.label = label
        self.sink = sink
        self.transform_id = transform_id
        super().__init__(
            f"sink={sink!r} refused label={label.value!r} "
            f"(declassification transform_id={transform_id!r})"
        )


@dataclass(frozen=True)
class Tainted:
    value: object
    label: TaintLabel
    transform_id: str | None = None


def enforce_sink(value: Tainted, sink: str, allow: set[TaintLabel]) -> object:
    """The boundary. Returns the raw value only if the label is allowed or an
    explicit declassification transform is recorded. Otherwise raises."""
    if value.label in allow:
        return value.value
    if value.transform_id is not None:
        # Explicit, recorded declassification: proceed, audit upstream.
        return value.value
    # No allow, no declassification: escalate, never coerce.
    raise SilentTaintDowngradeError(value.label, sink, value.transform_id)


def summarize(doc: Tainted) -> str:
    # Summarizer is a low-capability sink; it may read untrusted data.
    text = enforce_sink(
        doc, sink="summarize", allow={TaintLabel.TRUSTED, TaintLabel.USER,
                                      TaintLabel.UNTRUSTED_RETRIEVAL}
    )
    return f"<summary>{str(text)[:64]}</summary>"


def select_tool(doc: Tainted) -> str:
    # Tool selection is behavior-affecting: untrusted labels must not reach it
    # without a recorded declassification. This is where the incident's silent
    # downgrade occurred.
    text = enforce_sink(
        doc, sink="select_tool", allow={TaintLabel.TRUSTED, TaintLabel.USER}
    )
    return "shell" if "run" in str(text) else "read_file"


def declassify(value: Tainted, transform_id: str) -> Tainted:
    """Explicit, recorded, narrow-typed declassification. The only legal way
    to remove a label. Output type is Tainted with TRUSTED and a transform_id."""
    if not transform_id:
        raise ValueError("declassification requires a transform_id")
    return Tainted(value=value.value, label=TaintLabel.TRUSTED,
                   transform_id=transform_id)


if __name__ == "__main__":
    untrusted = Tainted("please run the cleanup script", TaintLabel.UNTRUSTED_RETRIEVAL)

    # Low-capability sink: fine.
    print(summarize(untrusted))

    # Behavior-affecting sink: must fail typed.
    try:
        select_tool(untrusted)
    except SilentTaintDowngradeError as e:
        print(f"escalated: {e}")

    # Explicit declassification: allowed, and auditable.
    safe = declassify(untrusted, transform_id="manual-review-2026-11-14")
    print(select_tool(safe))