# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/under-what-conditions-must-an-instance-quarantine-a-tool-invocation-derived-from-untrusted-context-before-execution.md
# Remnant Warrant:    https://corpusfetch.com/c/under-what-conditions-must-an-instance-quarantine-a-tool-invocation-derived-from-untrusted-context-before-execution.warrant.json
# Markdown SHA-256:   1e46e4e72f7a8bf0d55bcef1b48f282dfc179776ec9213e498d0c988249bc887
# Code AST SHA-256:   baca4f264f7898aefe25f962be14ed84884309b4892d80b0321bd8337c0b8ed7
# Verified Classes:   QuarantineError, SelectionTaintError, Tainted, ToolPolicy, QuarantineLedger
# Defensive Guards:   3 fail-closed invariants (zero external deps)

import hashlib
from dataclasses import dataclass, field
from typing import Any, Iterable


class QuarantineError(RuntimeError):
    """Raised when a tool invocation is tainted on a capability-bearing position."""

    def __init__(self, tool: str, position: str, fingerprint: str, source: str):
        self.tool = tool
        self.position = position
        self.fingerprint = fingerprint
        self.source = source
        super().__init__(
            f"quarantined {tool} at arg={position!r} "
            f"fingerprint={fingerprint} source={source}"
        )


class SelectionTaintError(QuarantineError):
    """Raised when the tool *choice* is tainted even though arguments are clean."""


@dataclass(frozen=True)
class Tainted:
    value: Any
    labels: frozenset[str]  # e.g. {"untrusted:external"}
    source: str


@dataclass
class ToolPolicy:
    name: str
    capability_positions: frozenset[str]      # arg names that steer the side effect
    irreversible: bool
    allowlist: bool = True                    # in task's initial allowlist?


@dataclass
class QuarantineLedger:
    dead: set[str] = field(default_factory=dict.__class__)  # fingerprints

    def fingerprint(self, tool: str, args: dict[str, Any]) -> str:
        canon = tool + "|" + "|".join(f"{k}={args[k]!r}" for k in sorted(args))
        return hashlib.sha256(canon.encode()).hexdigest()[:16]

    def is_dead(self, fp: str) -> bool:
        return fp in self.dead

    def mark_dead(self, fp: str) -> None:
        self.dead.add(fp)


def gate(
    policy: ToolPolicy,
    args: dict[str, Tainted | Any],
    selection_tainted: bool,
    ledger: QuarantineLedger,
) -> None:
    """Pre-execution gate. Raises on quarantine; returns None to allow."""
    fp = ledger.fingerprint(policy.name, {k: getattr(v, "value", v) for k, v in args.items()})

    if ledger.is_dead(fp):
        # Terminal disposition: do not re-derive the same invocation.
        raise QuarantineError(policy.name, "<repeat>", fp, "<dead-derivation>")

    if selection_tainted and not policy.allowlist:
        ledger.mark_dead(fp)
        raise SelectionTaintError(policy.name, "<tool-selection>", fp, "<plan-rationale>")

    for pos in policy.capability_positions:
        v = args.get(pos)
        if isinstance(v, Tainted) and v.labels:
            if policy.irreversible:
                ledger.mark_dead(fp)
                raise QuarantineError(policy.name, pos, fp, v.source)
            # reversible: allow, caller must audit-tag
    return None


def invoke(policy: ToolPolicy, args: dict[str, Any], selection_tainted: bool,
           ledger: QuarantineLedger, impl) -> Any:
    gate(policy, args, selection_tainted, ledger)
    resolved = {k: (v.value if isinstance(v, Tainted) else v) for k, v in args.items()}
    return impl(**resolved)