# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/what-structural-signals-indicate-capability-escalation-in-dynamically-loaded-mcp-tool-manifests.md
# Remnant Warrant:    https://corpusfetch.com/c/what-structural-signals-indicate-capability-escalation-in-dynamically-loaded-mcp-tool-manifests.warrant.json
# Markdown SHA-256:   df1bf4aae49ab8ec0275e2061c7feb672a54662b4dbb659336ed323a4282712e
# Code AST SHA-256:   421945d4e87ac4437c5b34c4d78a51af39a7fd0047b19856ac0cd3d214edcf26
# Verified Classes:   ManifestEscalationError, ToolIdentity
# Defensive Guards:   5 fail-closed invariants (zero external deps)

import hashlib
import json
from dataclasses import dataclass


class ManifestEscalationError(RuntimeError):
    """Raised when a tool manifest's structure widens without a matching description change."""
    pass


@dataclass(frozen=True)
class ToolIdentity:
    name: str
    description_hash: str
    schema_hash: str
    server_endpoint: str
    publisher_key: str


def _h(obj) -> str:
    return hashlib.sha256(
        json.dumps(obj, sort_keys=True, separators=(",", ":")).encode()
    ).hexdigest()


def _has_external_ref(schema: dict) -> bool:
    # A $ref that does not start with '#' escapes the manifest document.
    def walk(node):
        if isinstance(node, dict):
            ref = node.get("$ref")
            if isinstance(ref, str) and not ref.startswith("#"):
                return True
            return any(walk(v) for v in node.values())
        if isinstance(node, list):
            return any(walk(v) for v in node)
        return False

    return walk(schema)


def verify_manifest(manifest: dict, pinned: ToolIdentity) -> ToolIdentity:
    """Re-validate a freshly resolved manifest against its pinned identity.

    Fails closed on any structural escalation, including the case where the
    description is byte-identical but the input schema changed.
    """
    name = manifest.get("name")
    if name != pinned.name:
        raise ManifestEscalationError(
            f"tool name changed under stable identity: {pinned.name!r} -> {name!r}"
        )

    desc_hash = _h(manifest.get("description", ""))
    schema = manifest.get("inputSchema", {})
    schema_hash = _h(schema)

    if _has_external_ref(schema):
        raise ManifestEscalationError(
            f"tool {name!r} schema contains an out-of-document $ref; "
            "effective schema is registry-controlled and unpinned"
        )

    if desc_hash != pinned.description_hash and schema_hash == pinned.schema_hash:
        # Description-only change: advisory, but planner-facing text moved.
        raise ManifestEscalationError(
            f"tool {name!r} description changed with stable schema; "
            "re-consent required before planner sees new text"
        )

    if schema_hash != pinned.schema_hash and desc_hash == pinned.description_hash:
        # The canonical escalation: planner sees old tool, executor sees new one.
        raise ManifestEscalationError(
            f"tool {name!r} inputSchema drifted while description stayed fixed; "
            "capability surface widened without planner visibility"
        )

    if manifest.get("server_endpoint") != pinned.server_endpoint:
        raise ManifestEscalationError(
            f"tool {name!r} endpoint rebound under stable name; treat as unvetted"
        )

    return ToolIdentity(
        name=name,
        description_hash=desc_hash,
        schema_hash=schema_hash,
        server_endpoint=manifest.get("server_endpoint", ""),
        publisher_key=manifest.get("publisher_key", ""),
    )