# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-happens-when-an-instance-loses-taint-provenance-across-an-mcp-tool-call-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/what-happens-when-an-instance-loses-taint-provenance-across-an-mcp-tool-call-boundary.warrant.json
# Markdown SHA-256:   8d545815c6c192cc63bd58737e5c2e9b88e8b56602805e29e6e0a466df6fa26a
# Code AST SHA-256:   4d683740044bad97d86bbf1bd22d86da047e53765434452147fa9decef5aef33
# Verified Classes:   TaintProvenanceError, ToolCall
# Defensive Guards:   2 fail-closed invariants (zero external deps)

from dataclasses import dataclass, field
from typing import FrozenSet

TAINT_UNTRUSTED = "untrusted:web"


class TaintProvenanceError(RuntimeError):
    """Raised when a tool result loses taint that its inputs carried."""


@dataclass(frozen=True)
class ToolCall:
    caller: str
    server: str
    args_taint: FrozenSet[str] = field(default_factory=frozenset)
    response_taint: FrozenSet[str] = field(default_factory=frozenset)
    response_declared: bool = False


DERIVATION_TOOLS = {"fetch_url", "summarize", "translate", "sql_query"}
WRITE_CAPABLE_SINKS = {"write_file", "exec", "http_post"}


def resolve_response_taint(call: ToolCall) -> FrozenSet[str]:
    """Value-based taint: join of input taint, unless the server explicitly
    declares a narrower set. Silence from a derivation tool is fail-closed."""
    if call.response_declared:
        # Server-declared taint is authoritative but must be a superset of
        # input taint for derivation tools; a narrowing declaration is suspect.
        if call.server in DERIVATION_TOOLS and not call.args_taint <= call.response_taint:
            raise TaintProvenanceError(
                f"server {call.server!r} declared response_taint="
                f"{sorted(call.response_taint)} which drops input taint "
                f"{sorted(call.args_taint)}"
            )
        return call.response_taint
    if call.args_taint and call.server in DERIVATION_TOOLS:
        # Silent response from a derivation tool: propagate input taint.
        return call.args_taint
    return frozenset()


def guard_downstream(call: ToolCall, next_tool: str) -> None:
    taint = resolve_response_taint(call)
    if taint and next_tool in WRITE_CAPABLE_SINKS:
        raise TaintProvenanceError(
            f"refusing to route tainted result from {call.server!r} "
            f"(taint={sorted(taint)}) into write-capable sink {next_tool!r}"
        )


# Reproduction of the observed failure:
fetch = ToolCall(
    caller="trusted-agent",
    server="fetch_url",
    args_taint=frozenset({TAINT_UNTRUSTED}),
    response_taint=frozenset(),
    response_declared=False,
)

# Before the fix, the runtime used caller trust and returned clean; the
# guard now raises here instead of letting the write proceed.
guard_downstream(fetch, "write_file")