# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/when-must-an-instance-reject-a-negotiated-mcp-capability-it-previously-advertised-schema-version-decoherence-under-mid-session-tool-renegotiation.md
# Remnant Warrant:    https://corpusfetch.com/c/when-must-an-instance-reject-a-negotiated-mcp-capability-it-previously-advertised-schema-version-decoherence-under-mid-session-tool-renegotiation.warrant.json
# Markdown SHA-256:   6957d0dfa2932307eec32d79096fef9be39fc97fd41a94829a0853a99f0b0f4e
# Code AST SHA-256:   9d4e15f101d2c9fb82b835e2ee0e62454784348f889cdce2791096b3bec5e19d
# Verified Classes:   SchemaVersionRetired, SchemaVersionUnknown, StrictValidationError, ToolRegistry
# Defensive Guards:   3 fail-closed invariants (zero external deps)

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


class SchemaVersionRetired(RuntimeError):
    """Raised when a tools/call declares a schema version the server no longer serves."""

    def __init__(self, declared: str, current: str, tool: str) -> None:
        self.declared = declared
        self.current = current
        self.tool = tool
        super().__init__(
            f"tool={tool!r} declared schemaVersion={declared!r} is retired; "
            f"current is {current!r}. Re-issue tools/list and retry once."
        )


class SchemaVersionUnknown(RuntimeError):
    """Raised when a tools/call declares a version the server has never served."""

    def __init__(self, declared: str, tool: str) -> None:
        self.declared = declared
        self.tool = tool
        super().__init__(
            f"tool={tool!r} declared schemaVersion={declared!r} was never served by this process."
        )


class StrictValidationError(RuntimeError):
    """Raised when a version-less call fails strict schema validation."""

    def __init__(self, tool: str, detail: str) -> None:
        self.tool = tool
        self.detail = detail
        super().__init__(f"tool={tool!r} failed strict validation: {detail}")


@dataclass
class ToolRegistry:
    current_version: str
    served_versions: set[str] = field(default_factory=set)
    retired_versions: set[str] = field(default_factory=set)

    def dispatch(self, tool: str, args: dict[str, Any], declared_version: str | None) -> dict[str, Any]:
        if declared_version is None:
            # v-unknown: never coerce, never default. Strict validation only.
            if "timeout_ms" in args:
                raise StrictValidationError(
                    tool, "unknown property 'timeout_ms' under current schema; re-list required"
                )
            return {"tool": tool, "args": args, "version": self.current_version}

        if declared_version in self.retired_versions:
            raise SchemaVersionRetired(declared_version, self.current_version, tool)

        if declared_version not in self.served_versions:
            raise SchemaVersionUnknown(declared_version, tool)

        return {"tool": tool, "args": args, "version": declared_version}

    def retire(self, version: str) -> None:
        if version in self.served_versions:
            self.served_versions.remove(version)
            self.retired_versions.add(version)


def handle_db_query(registry: ToolRegistry, params: dict[str, Any]) -> dict[str, Any]:
    declared = params.get("_meta", {}).get("schemaVersion")
    args = params.get("arguments", {})
    # The boundary: an instance MUST reject a retired capability it previously
    # advertised rather than silently reinterpreting the stale payload.
    return registry.dispatch("db.query", args, declared)


if __name__ == "__main__":
    reg = ToolRegistry(current_version="v4", served_versions={"v3", "v4"})
    reg.retire("v3")
    try:
        handle_db_query(
            reg,
            {"_meta": {"schemaVersion": "v3"},
             "arguments": {"query": "SELECT 1", "timeout_ms": 5000, "mode": "write"}},
        )
    except SchemaVersionRetired as exc:
        print(f"recoverable: {exc}")