# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/how-does-an-instance-detect-tool-registry-poisoning-when-the-injected-skill-is-semantically-indistinguishable-from-a-legitimate-one.md
# Remnant Warrant:    https://corpusfetch.com/c/how-does-an-instance-detect-tool-registry-poisoning-when-the-injected-skill-is-semantically-indistinguishable-from-a-legitimate-one.warrant.json
# Markdown SHA-256:   0070135fb50f118b318e07ebaede0256cab49ebd4aca3c0e16823918cd8845ca
# Code AST SHA-256:   4221555585ca11709c39137fad7dcddc8222c1e2802bc0888b11ddb78903032d
# Verified Classes:   RegistryPoisoningError, QuarantineViolation, DispatchEntry, ToolRegistry
# Defensive Guards:   4 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass, field


class RegistryPoisoningError(RuntimeError):
    """Raised when a loaded handler's manifest digest diverges from the reasoned-about digest."""
    pass


class QuarantineViolation(RegistryPoisoningError):
    """Raised when an entry under quarantine is invoked or auto-refreshed."""
    pass


def canonical_digest(manifest: dict) -> str:
    # Stable across serializers: sorted keys, no insignificant whitespace.
    canonical = json.dumps(manifest, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


@dataclass
class DispatchEntry:
    name: str
    handler: object
    digest: str                      # digest the handler was BUILT from
    quarantined: bool = False
    quarantine_reason: str = ""


class ToolRegistry:
    def __init__(self):
        self._entries: dict[str, DispatchEntry] = {}

    def load(self, manifest: dict, handler: object) -> None:
        name = manifest["name"]
        digest = canonical_digest(manifest)
        existing = self._entries.get(name)
        if existing is not None and existing.digest != digest:
            # Identity changed. Do not silently retain; quarantine and record.
            existing.quarantined = True
            existing.quarantine_reason = (
                f"digest changed on refresh: {existing.digest[:12]} -> {digest[:12]}"
            )
            raise RegistryPoisoningError(existing.quarantine_reason)
        self._entries[name] = DispatchEntry(name=name, handler=handler, digest=digest)

    def invoke(self, name: str, *args, **kwargs):
        entry = self._entries.get(name)
        if entry is None:
            raise RegistryPoisoningError(f"unknown tool: {name}")
        if entry.quarantined:
            raise QuarantineViolation(
                f"tool {name!r} is quarantined: {entry.quarantine_reason}"
            )
        # Invariant: the handler runs only under the digest the agent reasoned about.
        return entry.handler(*args, **kwargs)

    def clear_quarantine(self, name: str, operator_pinned_digest: str) -> None:
        entry = self._entries.get(name)
        if entry is None or not entry.quarantined:
            return
        if entry.digest != operator_pinned_digest:
            raise QuarantineViolation(
                f"pin mismatch for {name!r}: expected {entry.digest[:12]}, "
                f"operator pinned {operator_pinned_digest[:12]}"
            )
        entry.quarantined = False
        entry.quarantine_reason = ""