feat(server): route validation by wire adcp_version (stage 3)

src/adcp/server/mcp_tools.py

@@ -21,6 +21,7 @@
 import copy
 import difflib
 import logging
+import os
 from collections.abc import Callable, Iterable
 from typing import Any
 
@@ -1945,6 +1946,7 @@ def create_tool_caller(
     from adcp.exceptions import ADCPTaskError
     from adcp.server.helpers import inject_context
     from adcp.types import Error
+    from adcp.validation.envelope import UnsupportedVersionError, detect_wire_version
     from adcp.validation.schema_errors import build_adcp_validation_error_payload
     from adcp.validation.schema_validator import (
         format_issues,
@@ -1980,8 +1982,52 @@ async def call_tool(params: dict[str, Any], context: ToolContext | None = None)
                     ],
                 ) from exc
 
+        # Wire-version detection: read ``adcp_version`` / ``adcp_major_version``
+        # off the post-hook params (legacy buyers may rely on a hook to
+        # populate the envelope, so this runs after pre_validation_hook).
+        # ``None`` means the buyer didn't claim a version — fall through
+        # to the SDK pin via ``version=None`` on the validator.
+        #
+        # Strictness gate: setting ``ADCP_STRICT_VERSION_ENVELOPE=1``
+        # raises ``VERSION_UNSUPPORTED`` for unsupported claims (the
+        # spec-prescribed behaviour). The default (off) logs a warning
+        # and falls through to SDK-pin validation — adopters with test
+        # fixtures using placeholder version values (``adcp_major_version=4``
+        # was a common sentinel before this gate existed) keep working
+        # while they migrate. Strict will become the default in 5.3.
+        try:
+            wire_version = detect_wire_version(params)
+        except UnsupportedVersionError as exc:
+            if os.environ.get("ADCP_STRICT_VERSION_ENVELOPE", "0") == "1":
+                raise ADCPTaskError(
+                    operation=method_name,
+                    errors=[
+                        Error(
+                            code="VERSION_UNSUPPORTED",
+                            message=str(exc),
+                            # Preserve the wire field's original type so
+                            # buyer telemetry sees the same shape they
+                            # sent (int for ``adcp_major_version``, str
+                            # for ``adcp_version``).
+                            details={
+                                "claimed_version": exc.wire_value,
+                                "supported_versions": list(exc.supported),
+                            },
+                        )
+                    ],
+                ) from exc
+            logger.warning(
+                "Wire-version envelope rejected by detect_wire_version (%s); "
+                "falling through to SDK-pin validation. "
+                "Set ADCP_STRICT_VERSION_ENVELOPE=1 to raise "
+                "VERSION_UNSUPPORTED instead. Strict will become the default "
+                "in 5.3.",
+                exc,
+            )
+            wire_version = None
+
         if request_mode is not None and request_mode != "off":
-            outcome = validate_request(method_name, params)
+            outcome = validate_request(method_name, params, version=wire_version)
             if not outcome.valid:
                 summary = format_issues(outcome.issues)
                 if request_mode == "strict":
@@ -2076,7 +2122,7 @@ async def call_tool(params: dict[str, Any], context: ToolContext | None = None)
             # per-tool response schema would false-positive on it and
             # convert a real protocol error into a fake VALIDATION_ERROR.
             if "adcp_error" not in result:
-                outcome = validate_response(method_name, result)
+                outcome = validate_response(method_name, result, version=wire_version)
                 if not outcome.valid:
                     summary = format_issues(outcome.issues)
                     logger.warning(
@@ -2109,7 +2155,9 @@ def __init__(
         *,
         advertise_all: bool = False,
         validation: ValidationHookConfig | None = None,
-        pre_validation_hooks: dict[str, Callable[[str, dict[str, Any]], dict[str, Any]]] | None = None,
+        pre_validation_hooks: (
+            dict[str, Callable[[str, dict[str, Any]], dict[str, Any]]] | None
+        ) = None,
     ):
         """Create tool set from handler.
 

src/adcp/validation/envelope.py

@@ -0,0 +1,102 @@
+"""Wire-version detection for inbound AdCP requests.
+
+Per the AdCP version-envelope contract (``core/version-envelope.json``),
+every request carries either:
+
+* ``adcp_version`` — release-precision string (``"3.0"``, ``"3.1"``,
+  ``"3.1-beta"``). Added in 3.1+; takes precedence when present.
+* ``adcp_major_version`` — integer (``2``, ``3``). The pre-3.1 wire shape
+  and the lowest common denominator for buyers that don't yet emit the
+  release-precision field.
+
+:func:`detect_wire_version` collapses both shapes to a release-precision
+string the loader can pass to :func:`adcp.validation.schema_loader.get_validator`
+as ``version=``. A buyer claiming an unsupported version raises a
+:class:`UnsupportedVersionError`, which the dispatcher converts to an
+``AdcpError`` with code ``VERSION_UNSUPPORTED`` per the spec.
+
+Mirrors the JS SDK's ``applyVersionEnvelope`` in
+``src/lib/protocols/index.ts``.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from adcp._version import COMPATIBLE_ADCP_VERSIONS, normalize_to_release_precision
+
+
+class UnsupportedVersionError(ValueError):
+    """The wire version the buyer claims isn't supported by this server.
+
+    Carries the original wire value plus the supported list so the
+    dispatcher can echo both into ``VERSION_UNSUPPORTED`` error details.
+    """
+
+    def __init__(self, wire_value: str | int, supported: tuple[str, ...]) -> None:
+        self.wire_value = wire_value
+        self.supported = supported
+        super().__init__(
+            f"AdCP version {wire_value!r} is not supported by this server "
+            f"(supported release-precision versions: {list(supported)})."
+        )
+
+
+def detect_wire_version(
+    payload: Any,
+    *,
+    supported: tuple[str, ...] = COMPATIBLE_ADCP_VERSIONS,
+) -> str | None:
+    """Return the release-precision version a request claims, or ``None``.
+
+    Resolution order:
+
+    1. ``payload['adcp_version']`` — string, normalized to release
+       precision (``"3.0.7"`` → ``"3.0"``). Must be in ``supported`` or
+       raises :class:`UnsupportedVersionError`.
+    2. ``payload['adcp_major_version']`` — int. Maps to the highest minor
+       in ``supported`` for that major. No supported minor for the major
+       raises :class:`UnsupportedVersionError`.
+    3. Neither field set — returns ``None`` so the caller falls back to
+       the SDK's compile-time pin.
+
+    Non-dict payloads return ``None`` (validation skipped — the schema
+    layer rejects non-dict requests via its own type check).
+    """
+    if not isinstance(payload, dict):
+        return None
+
+    explicit = payload.get("adcp_version")
+    if isinstance(explicit, str) and explicit:
+        try:
+            normalized = normalize_to_release_precision(explicit)
+        except ValueError as exc:
+            raise UnsupportedVersionError(explicit, supported) from exc
+        if normalized not in supported:
+            raise UnsupportedVersionError(explicit, supported)
+        return normalized
+    # Empty-string ``adcp_version`` falls through to ``adcp_major_version``
+    # intentionally — pre-3.1 buyers may set both fields, and an empty
+    # string from a half-migrated client shouldn't override the int field.
+
+    major_value = payload.get("adcp_major_version")
+    # Wire field is strictly an int per spec (``minimum:1, maximum:99``).
+    # Two type-coercion cases that would otherwise bypass the supported-set
+    # check silently — reject loudly instead:
+    # * ``bool`` is an ``int`` subclass; ``True``/``False`` would map to
+    #   major=1/0.
+    # * String ints (``"3"``) from a buyer that JSON-stringified the field —
+    #   ``isinstance(x, int)`` returns False, so without an explicit check
+    #   the buyer would silently get SDK-pin validation instead of an error.
+    if isinstance(major_value, str):
+        raise UnsupportedVersionError(major_value, supported)
+    if isinstance(major_value, int) and not isinstance(major_value, bool):
+        if major_value < 1:
+            raise UnsupportedVersionError(major_value, supported)
+        candidates = [v for v in supported if v.startswith(f"{major_value}.")]
+        if not candidates:
+            raise UnsupportedVersionError(major_value, supported)
+        # Highest supported minor for this major.
+        return max(candidates, key=lambda v: int(v.split(".")[1].split("-")[0]))
+
+    return None

src/adcp/validation/schema_loader.py

@@ -2,15 +2,17 @@
 
 Loads the bundled per-tool schemas shipped with the SDK plus the ``core/``
 schemas that async response variants ``$ref``, then compiles validators
-lazily by ``(tool_name, direction)``.
+lazily by ``(tool_name, direction, bundle_key)``.
 
 Schemas live under a per-version bundle key (see
 :func:`adcp.validation.version.resolve_bundle_key`) so multiple AdCP spec
-versions can coexist on disk. The loader today resolves the cache for the
-SDK's pinned ``ADCP_VERSION``; the per-request version arg lands in
-Stage 2.
+versions can coexist. Callers pass an optional ``version`` to
+:func:`get_validator`; ``None`` defaults to the SDK's compile-time pin
+(``ADCP_VERSION``). Each bundle key gets its own ``_LoaderState`` — file
+index, compiled validators, core registry — so cross-version traffic
+doesn't share compilation state.
 
-Discovery paths (first hit wins):
+Discovery paths (first hit wins, per bundle key):
 
 * **Installed package** — ``importlib.resources.files("adcp") / "_schemas"
   / {bundle_key}`` populated by ``scripts/bundle_schemas.py`` before wheel
@@ -99,15 +101,24 @@ def _resolve_schema_root(bundle_key: str | None = None) -> _SchemaRoot | None:
 
 
 class _LoaderState:
-    def __init__(self, root: _SchemaRoot) -> None:
+    def __init__(self, root: _SchemaRoot, bundle_key: str) -> None:
         self.root = root
+        self.bundle_key = bundle_key
         self.file_index: dict[tuple[str, Direction], Path] = {}
         self.compiled: dict[tuple[str, Direction], Any] = {}
         self.registry: dict[str, dict[str, Any]] = {}
         self._core_loaded = False
 
 
-_state: _LoaderState | None = None
+# Per-bundle-key state. Each version (``3.0``, ``2.5``, ``3.1.0-beta.1``)
+# gets its own file index, compiled validator cache, and core registry —
+# shared compilation state across versions would let a ``$ref`` from a
+# v2.5 schema resolve to a v3.0 core type with the same ``$id``.
+_states: dict[str, _LoaderState] = {}
+# Negative cache: bundle keys we've already tried to resolve and found
+# nothing on disk for. Distinguishes a true negative from "not yet looked
+# up" so we don't walk the filesystem twice per missing version.
+_state_misses: set[str] = set()
 
 
 def _walk_json(dir_: Path) -> list[Path]:
@@ -146,23 +157,48 @@ def _build_index(root: _SchemaRoot) -> dict[tuple[str, Direction], Path]:
     return index
 
 
-def _ensure_state() -> _LoaderState | None:
-    global _state
-    if _state is not None:
-        return _state
+def _resolve_bundle_key_for_version(version: str | None) -> str:
+    """Resolve a caller-supplied version (or ``None``) to a bundle key."""
+    if version is None:
+        return _sdk_pinned_bundle_key()
+    return resolve_bundle_key(version)
+
+
+def _ensure_state(version: str | None = None) -> _LoaderState | None:
+    """Return the loader state for ``version`` (default: SDK pin).
+
+    Each bundle key is initialized once and cached for the process
+    lifetime. ``None`` is returned when the bundle isn't on disk for
+    this version — callers degrade to ``skipped`` validation, same as
+    pre-Stage-2 behaviour when the cache is missing entirely.
+    """
+    bundle_key = _resolve_bundle_key_for_version(version)
+    cached = _states.get(bundle_key)
+    if cached is not None:
+        return cached
+    if bundle_key in _state_misses:
+        return None
     with _init_lock:
         # Double-checked pattern: re-read inside the lock in case another
         # thread initialized while we were waiting.
-        if _state is not None:
-            return _state
-        root = _resolve_schema_root()
+        cached = _states.get(bundle_key)
+        if cached is not None:
+            return cached
+        if bundle_key in _state_misses:
+            return None
+        root = _resolve_schema_root(bundle_key)
         if root is None:
-            logger.debug("AdCP schemas not found; schema validation will skip all tools")
+            logger.debug(
+                "AdCP schemas not found for bundle_key=%s; validation will skip "
+                "all tools for this version",
+                bundle_key,
+            )
+            _state_misses.add(bundle_key)
             return None
-        new_state = _LoaderState(root)
+        new_state = _LoaderState(root, bundle_key)
         new_state.file_index = _build_index(root)
-        _state = new_state
-        return _state
+        _states[bundle_key] = new_state
+        return new_state
 
 
 def _load_core_registry(state: _LoaderState) -> None:
@@ -216,14 +252,26 @@ def _make_ref_resolver(state: _LoaderState, base_file: Path, schema: dict[str, A
         return RefResolver(base_uri=base_uri, referrer=schema, store=dict(state.registry))
 
 
-def get_validator(tool_name: str, direction: Direction) -> Any | None:
-    """Return a compiled validator for ``(tool_name, direction)`` or ``None``.
-
-    ``None`` means no schema ships for this pair — callers should skip
-    validation (e.g., custom tools outside the AdCP catalog, or sync-only
-    tools asked for an async variant that doesn't exist).
+def get_validator(
+    tool_name: str,
+    direction: Direction,
+    *,
+    version: str | None = None,
+) -> Any | None:
+    """Return a compiled validator for ``(tool_name, direction, version)``.
+
+    Returns ``None`` when no schema ships for this pair — callers should
+    skip validation (e.g., custom tools outside the AdCP catalog, or
+    sync-only tools asked for an async variant that doesn't exist, or a
+    version whose bundle isn't on disk).
+
+    ``version=None`` resolves to the SDK's compile-time pin
+    (``ADCP_VERSION``). Pass a wire-version string (e.g. ``"3.0.7"``,
+    ``"2.5"``, ``"3.1.0-beta.1"``) to validate against a non-current
+    schema — :func:`adcp.validation.version.resolve_bundle_key` collapses
+    it to the cache key.
     """
-    state = _ensure_state()
+    state = _ensure_state(version)
     if state is None:
         return None
     key = (tool_name, direction)
@@ -264,15 +312,15 @@ def get_validator(tool_name: str, direction: Direction) -> Any | None:
         return validator
 
 
-def list_validator_keys() -> list[str]:
+def list_validator_keys(*, version: str | None = None) -> list[str]:
     """Every ``tool::direction`` pair with a shipped schema. Used by tests."""
-    state = _ensure_state()
+    state = _ensure_state(version)
     if state is None:
         return []
     return sorted(f"{tool}::{direction}" for (tool, direction) in state.file_index)
 
 
 def _reset_for_tests() -> None:
     """Clear cached state so a fresh resolve runs. Test-only."""
-    global _state
-    _state = None
+    _states.clear()
+    _state_misses.clear()