# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-treat-retrieved-context-as-tainted-the-label-laundering-boundary-in-prompt-injection-taint-tracking.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-treat-retrieved-context-as-tainted-the-label-laundering-boundary-in-prompt-injection-taint-tracking.warrant.json
# Markdown SHA-256:   d12316d3b5d715f0a9a3298955935317c025c907053388434957caab628f6d06
# Code AST SHA-256:   fa4ce2856cd301ecd204d9701f1a2563b11b4ce0b854b09055eeef0b4c2ddfcf
# Verified Classes:   TaintLaunderingError, Label, Declassifier
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Iterable


class TaintLaunderingError(RuntimeError):
    """Raised when a derived value would be treated as trusted despite
    consuming untrusted spans without an audited declassifier."""


@dataclass(frozen=True)
class Label:
    untrusted_sources: frozenset[str] = field(default_factory=frozenset)

    @property
    def is_tainted(self) -> bool:
        return bool(self.untrusted_sources)

    def join(self, other: "Label") -> "Label":
        return Label(self.untrusted_sources | other.untrusted_sources)


TRUSTED = Label(frozenset())


@dataclass(frozen=True)
class Declassifier:
    name: str
    closed_schema: bool  # must be True to legally lower a label
    covered_sinks: frozenset[str]

    def may_declassify(self, sink: str) -> bool:
        return self.closed_schema and sink in self.covered_sinks


def derive_label(
    consumed: Iterable[Label],
    *,
    produced_by_model: bool,
    declassifier: Declassifier | None = None,
    sink: str | None = None,
) -> Label:
    """Compute the label of a derived value.

    Invariant: a derived value inherits the join of all inputs actually
    consumed. Model-mediated rewriting never lowers the label on its own.
    """
    joined = TRUSTED
    for lbl in consumed:
        joined = joined.join(lbl)

    if not joined.is_tainted:
        return joined

    # Model output over tainted input stays tainted unless a closed-schema
    # declassifier explicitly covers the sink.
    if produced_by_model:
        if declassifier is None:
            return joined
        if sink is None:
            raise TaintLaunderingError(
                f"declassifier {declassifier.name!r} invoked without a sink class; "
                "cannot certify the output is safe for any sink"
            )
        if not declassifier.may_declassify(sink):
            raise TaintLaunderingError(
                f"declassifier {declassifier.name!r} cannot lower taint for sink "
                f"{sink!r}: closed_schema={declassifier.closed_schema}, "
                f"covered={sorted(declassifier.covered_sinks)}"
            )
        return TRUSTED

    return joined


def guard_sink(value_label: Label, sink: str) -> None:
    if value_label.is_tainted:
        raise TaintLaunderingError(
            f"refusing to pass tainted value to sink {sink!r}; "
            f"untrusted_sources={sorted(value_label.untrusted_sources)}"
        )


# --- Example: the observed failure, now blocked ---
if __name__ == "__main__":
    fetched = Label(frozenset({"retrieval:web_fetch:doc#7"}))
    # Model summarizes the fetched doc -> still tainted.
    summary = derive_label([fetched], produced_by_model=True)
    # Planner consumes the summary -> inherits taint.
    plan = derive_label([summary], produced_by_model=True)
    try:
        guard_sink(plan, "send_email")
    except TaintLaunderingError as exc:
        print("BLOCKED:", exc)