# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-quarantine-a-tainted-tool-result-rather-than-pass-it-to-a-downstream-tool.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-quarantine-a-tainted-tool-result-rather-than-pass-it-to-a-downstream-tool.warrant.json
# Markdown SHA-256:   91eb4306f2e941a41272337df34cb1a3fd7ca93b23fd3c25305e87057ff1bbd3
# Code AST SHA-256:   e952ac63681f8d308505b6afdbee2f7d7023562f10e58e123260f405c6b4b52f
# Verified Classes:   SinkEffectClass, TaintLaunderingError, Labeled
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Iterable


class SinkEffectClass(Enum):
    REVERSIBLE = "reversible"   # in-memory, sandboxed
    PERSISTENT = "persistent"   # filesystem, DB, durable store
    EGRESS = "egress"           # network, mail, external API


class TaintLaunderingError(RuntimeError):
    """Raised when a tainted value would cross a sink without quarantine."""


@dataclass(frozen=True)
class Labeled:
    value: str
    tainted: bool
    provenance: tuple[str, ...]  # ordered source ids; empty == trusted


def _declassified_by_trusted_principal(prov: Iterable[str]) -> bool:
    # A declassification is only valid if the *last* provenance entry is an
    # operator-granted capability id, not a model-issued one.
    prov = tuple(prov)
    return bool(prov) and prov[-1].startswith("cap:operator:")


def dispatch_to_tool(
    arg: Labeled,
    sink: SinkEffectClass,
    *,
    destination_namespace: str | None = None,
    tainted_namespaces: set[str] | None = None,
) -> None:
    """Gate a labeled value before it reaches a tool with side effects.

    Quarantine is required when a tainted value reaches a PERSISTENT or
    EGRESS sink and has not been declassified by a trusted principal.
    """
    if tainted_namespaces is None:
        tainted_namespaces = set()

    # Reads from a namespace previously written by a tainted call inherit
    # the taint, even if the value's own label claims otherwise.
    namespace_tainted = (
        destination_namespace is not None
        and destination_namespace in tainted_namespaces
    )

    effective_taint = arg.tainted or namespace_tainted

    if not effective_taint:
        return

    if _declassified_by_trusted_principal(arg.provenance):
        return

    if sink is SinkEffectClass.REVERSIBLE:
        # Propagation alone is correct; label is preserved by the caller.
        return

    # PERSISTENT or EGRESS with taint and no trusted declassification:
    # this is the laundering boundary. Refuse and quarantine.
    raise TaintLaunderingError(
        f"tainted value (provenance={arg.provenance!r}) reached "
        f"{sink.value} sink without operator declassification; "
        f"namespace={destination_namespace!r} "
        f"namespace_tainted={namespace_tainted}. Quarantine required."
    )


def record_persistent_write(
    destination_namespace: str,
    tainted_namespaces: set[str],
    *,
    was_tainted: bool,
) -> None:
    """After a successful persistent write, mark the namespace if tainted.

    Omitting this is the exact step that produced the laundering bug:
    the write succeeds, but the subsequent read is labeled trusted.
    """
    if was_tainted:
        tainted_namespaces.add(destination_namespace)