# 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-dynamic-ift-for-tool-and-rag-pipelines.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-treat-retrieved-context-as-tainted-the-label-laundering-boundary-in-dynamic-ift-for-tool-and-rag-pipelines.warrant.json
# Markdown SHA-256:   52be3d0504696402d57ff7c1f6d78897d3f7fcca422d90f79f631680893cad18
# Code AST SHA-256:   f7835584e2384a66a0c3677cc094ca6cfe1d512edcebdb0c9a43a1acc391fe96
# Verified Classes:   TaintLaunderingError, ProvenanceNode, SinkCall
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import Optional


class TaintLaunderingError(RuntimeError):
    """Raised when a sink is invoked with an argument whose provenance
    reaches a tainted node but the runtime failed to propagate the label."""
    pass


@dataclass
class ProvenanceNode:
    node_id: str
    tainted: bool
    parents: list["ProvenanceNode"] = field(default_factory=list)

    def is_tainted(self, _seen: Optional[set] = None) -> bool:
        seen = _seen or set()
        if self.node_id in seen:
            return False
        seen.add(self.node_id)
        if self.tainted:
            return True
        return any(p.is_tainted(seen) for p in self.parents)


@dataclass
class SinkCall:
    sink: str
    args: dict[str, ProvenanceNode]
    declassified: bool = False

    def assert_safe(self) -> None:
        if self.declassified:
            return
        for arg_name, node in self.args.items():
            if node.is_tainted():
                raise TaintLaunderingError(
                    f"Sink '{self.sink}' argument '{arg_name}' has tainted "
                    f"provenance (node {node.node_id}); label was laundered "
                    f"across a model-mediated transform. Declassify explicitly "
                    f"or block the call."
                )


# --- Demonstration of the boundary failure ---
retrieval = ProvenanceNode("rag_chunk_42", tainted=True)
summary = ProvenanceNode("planner_note", tainted=False, parents=[retrieval])
send = SinkCall("send_email", {"body": summary})
send.assert_safe()  # raises: summary's provenance reaches a tainted node