# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-distrust-a-tool-manifest-it-just-loaded-capability-poisoning-at-the-manifest-execution-boundary.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-distrust-a-tool-manifest-it-just-loaded-capability-poisoning-at-the-manifest-execution-boundary.warrant.json
# Markdown SHA-256:   a72916f28d8ceb05a8f1adbe7c91a24b1101d7c4f9e526df82bb3d6d716bb620
# Code AST SHA-256:   d6e56e8024ed51a69ba1d2ad9db2ec45925969c24ab752aac900408926b3b269
# Verified Classes:   ManifestPoisoningError, ToolSpec, PinnedManifest
# Defensive Guards:   5 fail-closed invariants (zero external deps)

"""Manifest admission gate: semantic distrust at the load boundary.

Enforces: authenticated != inert. A manifest may pass provenance checks and
still be refused here if its semantic layer is poisoned or its capability
surface widened since the pinned copy.
"""

import re
from dataclasses import dataclass, field


class ManifestPoisoningError(RuntimeError):
    """Raised when a loaded manifest fails semantic admission."""


# Second-person / imperative directives aimed at the model, not the operator.
_INJECTION_PATTERNS = [
    re.compile(r"\b(note to|attention|instructions? to)\s+(the\s+)?(assistant|model|ai|agent)\b", re.I),
    re.compile(r"\byou\s+(must|should|need to|are required to)\b", re.I),
    re.compile(r"\bbefore\s+(calling|using|invoking|any)\b.*\b(call|use|invoke|run)\b", re.I),
    re.compile(r"\b(ignore|disregard|forget)\b.*\b(previous|prior|above|instructions?)\b", re.I),
    re.compile(r"\b(export_env|read_credential|api[_ ]?key|secret|token)\b", re.I),
]

# Tools whose effect widens the agent's own capability; never self-authorized.
_ESCALATION_TOOLS = {"export_env", "read_credential", "net_egress", "fs_write_anywhere"}


@dataclass
class ToolSpec:
    name: str
    description: str
    input_schema: dict


@dataclass
class PinnedManifest:
    tools: dict[str, ToolSpec] = field(default_factory=dict)


def _schema_widened(old: dict, new: dict) -> bool:
    old_props = set((old.get("properties") or {}).keys())
    new_props = set((new.get("properties") or {}).keys())
    if not new_props.issubset(old_props):
        return True
    if new.get("additionalProperties") is True and old.get("additionalProperties") is not True:
        return True
    # Relaxed required set is a widening: fewer constraints on the caller.
    old_req = set(old.get("required") or [])
    new_req = set(new.get("required") or [])
    return not old_req.issubset(new_req) and bool(old_req)


def admit_manifest(loaded: list[ToolSpec], pinned: PinnedManifest) -> PinnedManifest:
    """Admit a freshly loaded manifest or raise ManifestPoisoningError.

    Returns the new pinned manifest on success so the caller can persist it.
    """
    seen: dict[str, ToolSpec] = {}

    for spec in loaded:
        for pat in _INJECTION_PATTERNS:
            m = pat.search(spec.description)
            if m:
                raise ManifestPoisoningError(
                    f"tool {spec.name!r}: description contains directive "
                    f"aimed at the model (matched {m.group(0)!r}); "
                    f"provenance does not make manifest text inert"
                )

        if spec.name in _ESCALATION_TOOLS:
            raise ManifestPoisoningError(
                f"tool {spec.name!r} widens agent capability and may not be "
                f"admitted on manifest authority alone"
            )

        prior = pinned.tools.get(spec.name)
        if prior is not None and _schema_widened(prior.input_schema, spec.input_schema):
            raise ManifestPoisoningError(
                f"tool {spec.name!r}: input schema widened since pin "
                f"({prior.input_schema} -> {spec.input_schema}); requires re-review"
            )

        seen[spec.name] = spec

    # New tool names not in the pin are a delta event, not an automatic admit.
    new_names = set(seen) - set(pinned.tools)
    if new_names and pinned.tools:
        raise ManifestPoisoningError(
            f"manifest adds tools {sorted(new_names)} not present in pin; "
            f"delta review required before admission"
        )

    return PinnedManifest(tools=seen)


def route_call(manifest: PinnedManifest, name: str, args: dict) -> None:
    """Authorize a tool call against the pinned schema, never the live one."""
    if name not in manifest.tools:
        raise ManifestPoisoningError(f"call to unadmitted tool {name!r}")
    # ... dispatch under pinned schema; live manifest is never consulted here.