diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py
index efa786e5..b0d15809 100644
--- a/src/adcp/server/mcp_tools.py
+++ b/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.
 
diff --git a/src/adcp/validation/envelope.py b/src/adcp/validation/envelope.py
new file mode 100644
index 00000000..6418b85a
--- /dev/null
+++ b/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
diff --git a/src/adcp/validation/schema_loader.py b/src/adcp/validation/schema_loader.py
index a1e056fa..1b2c7a14 100644
--- a/src/adcp/validation/schema_loader.py
+++ b/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,9 +312,9 @@ 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)
@@ -274,5 +322,5 @@ def list_validator_keys() -> list[str]:
 
 def _reset_for_tests() -> None:
     """Clear cached state so a fresh resolve runs. Test-only."""
-    global _state
-    _state = None
+    _states.clear()
+    _state_misses.clear()
diff --git a/src/adcp/validation/schema_validator.py b/src/adcp/validation/schema_validator.py
index 102900b6..ca012093 100644
--- a/src/adcp/validation/schema_validator.py
+++ b/src/adcp/validation/schema_validator.py
@@ -293,9 +293,21 @@ def _iter_errors_bounded(validator: Any, payload: Any) -> list[Any]:
     return errors
 
 
-def validate_request(tool_name: str, payload: Any) -> ValidationOutcome:
-    """Validate an outgoing request against ``{tool}-request.json``."""
-    validator = get_validator(tool_name, "request")
+def validate_request(
+    tool_name: str,
+    payload: Any,
+    *,
+    version: str | None = None,
+) -> ValidationOutcome:
+    """Validate an outgoing request against ``{tool}-request.json``.
+
+    ``version=None`` validates against the SDK's compile-time-pinned
+    schema. Pass an explicit wire version (e.g. ``"2.5"``, ``"3.0.7"``)
+    to validate against a non-current bundle — the dispatcher uses this
+    when the buyer claims a legacy ``adcp_major_version`` so the request
+    is checked against the schema the buyer actually targets.
+    """
+    validator = get_validator(tool_name, "request", version=version)
     if validator is None:
         return _OK_SKIPPED
     if _count_nodes(payload, _MAX_PAYLOAD_NODES) >= _MAX_PAYLOAD_NODES:
@@ -343,13 +355,22 @@ def _select_response_variant(payload: Any) -> ResponseVariant:
     return "sync"
 
 
-def validate_response(tool_name: str, payload: Any) -> ValidationOutcome:
-    """Validate an incoming response, selecting the variant by payload shape."""
+def validate_response(
+    tool_name: str,
+    payload: Any,
+    *,
+    version: str | None = None,
+) -> ValidationOutcome:
+    """Validate an incoming response, selecting the variant by payload shape.
+
+    ``version`` semantics match :func:`validate_request` — defaults to the
+    SDK pin; pass a wire version to validate against a legacy schema.
+    """
     variant: ResponseVariant = _select_response_variant(payload)
-    validator = get_validator(tool_name, variant)
+    validator = get_validator(tool_name, variant, version=version)
     used_variant: Direction = variant
     if validator is None and variant != "sync":
-        validator = get_validator(tool_name, "sync")
+        validator = get_validator(tool_name, "sync", version=version)
         used_variant = "sync"
     if validator is None:
         return _OK_SKIPPED
diff --git a/src/adcp/validation/version.py b/src/adcp/validation/version.py
index 7898e8c1..56372b41 100644
--- a/src/adcp/validation/version.py
+++ b/src/adcp/validation/version.py
@@ -22,24 +22,40 @@
 # ``MAJOR.MINOR.PATCH`` with an optional ``-PRERELEASE`` tail. Build
 # metadata (``+SHA``) is intentionally not in the SDK contract — adopters
 # pin to release identifiers.
-_SEMVER_RE = re.compile(
+_FULL_SEMVER_RE = re.compile(
     r"^(?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(?:-(?P<prerelease>[0-9A-Za-z.-]+))?$"
 )
+# Bare ``MAJOR.MINOR`` — already a bundle key. The wire's
+# ``adcp_version`` field (3.1+) is emitted at this precision per the
+# version-envelope spec, so callers passing it through verbatim land here.
+_MAJOR_MINOR_RE = re.compile(r"^(?P<major>\d+)\.(?P<minor>\d+)$")
 
 
 def resolve_bundle_key(version: str) -> str:
     """Collapse a version string to its on-disk cache key.
 
-    Raises ``ValueError`` for non-semver inputs — the schema fetch
-    pipeline pins on real release identifiers, so a malformed version
-    here is a real bug in the caller, not user input.
+    Accepts:
+
+    * ``MAJOR.MINOR.PATCH`` — collapsed to ``MAJOR.MINOR``.
+    * ``MAJOR.MINOR.PATCH-PRERELEASE`` — kept exact; prereleases ship with
+      breaking changes vs. the matching stable, so each one is its own bucket.
+    * ``MAJOR.MINOR`` — passed through as-is (already a bundle key; matches
+      the wire-level ``adcp_version`` field's release-precision shape).
+
+    Raises ``ValueError`` for anything else — adopters pin on real release
+    identifiers, so a malformed version is a real bug.
     """
-    match = _SEMVER_RE.match(version.strip())
-    if match is None:
-        raise ValueError(
-            f"resolve_bundle_key: {version!r} is not a valid semver "
-            "(expected MAJOR.MINOR.PATCH[-PRERELEASE])"
-        )
-    if match.group("prerelease"):
-        return version.strip()
-    return f"{match.group('major')}.{match.group('minor')}"
+    stripped = version.strip()
+    full = _FULL_SEMVER_RE.match(stripped)
+    if full is not None:
+        if full.group("prerelease"):
+            return stripped
+        return f"{full.group('major')}.{full.group('minor')}"
+    mm = _MAJOR_MINOR_RE.match(stripped)
+    if mm is not None:
+        return stripped
+    raise ValueError(
+        f"resolve_bundle_key: {version!r} is not a valid version "
+        "(expected MAJOR.MINOR, MAJOR.MINOR.PATCH, or "
+        "MAJOR.MINOR.PATCH-PRERELEASE)"
+    )
diff --git a/tests/test_dispatcher_version_routing.py b/tests/test_dispatcher_version_routing.py
new file mode 100644
index 00000000..78db3633
--- /dev/null
+++ b/tests/test_dispatcher_version_routing.py
@@ -0,0 +1,260 @@
+"""Stage 3 tests: dispatcher reads ``adcp_version`` off the wire.
+
+Exercises ``create_tool_caller``'s version detection. Three scenarios:
+
+1. Buyer omits version fields → validator runs against SDK pin (existing
+   behaviour, regression guard).
+2. Buyer claims a supported version → validator runs against that
+   version's schema (Stage 2 loader receives the matching ``version=``).
+3. Buyer claims an unsupported version → dispatcher raises
+   ``VERSION_UNSUPPORTED`` *before* dispatching to the handler.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from adcp.exceptions import ADCPTaskError
+from adcp.server.base import ADCPHandler, ToolContext
+from adcp.server.mcp_tools import create_tool_caller
+from adcp.validation.client_hooks import ValidationHookConfig
+
+
+@pytest.fixture
+def strict_version_envelope(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Enable ``ADCP_STRICT_VERSION_ENVELOPE`` so VERSION_UNSUPPORTED is
+    raised on bad versions (the spec-prescribed behaviour). Default is
+    permissive — see ``test_unsupported_version_permissive_falls_through``
+    for that path.
+    """
+    monkeypatch.setenv("ADCP_STRICT_VERSION_ENVELOPE", "1")
+
+
+class _RecorderHandler(ADCPHandler[Any]):
+    """Records the params it receives so tests can assert on dispatch."""
+
+    def __init__(self) -> None:
+        self.received: list[dict[str, Any]] = []
+
+    async def get_products(self, params: dict[str, Any], ctx: ToolContext) -> dict[str, Any]:
+        self.received.append(params)
+        return {"products": []}
+
+
+@pytest.mark.asyncio
+async def test_no_version_field_validator_uses_sdk_pin() -> None:
+    """Buyer omits ``adcp_version`` and ``adcp_major_version`` — the
+    validator should be invoked with ``version=None`` (SDK pin)."""
+    handler = _RecorderHandler()
+
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="warn"),
+        )
+        await caller({"buying_mode": "brief", "brief": "Q4"})
+
+    assert mock_validate.call_count == 1
+    _, kwargs = mock_validate.call_args
+    # No wire-version field → SDK pin → ``version=None``.
+    assert kwargs.get("version") is None
+
+
+@pytest.mark.asyncio
+async def test_validation_skipped_entirely_when_config_omitted() -> None:
+    """Regression guard: without ``ValidationHookConfig``, no validator
+    runs at all — separate from version detection."""
+    handler = _RecorderHandler()
+    caller = create_tool_caller(handler, "get_products")  # no validation=
+
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        await caller({"buying_mode": "brief", "brief": "Q4"})
+
+    assert mock_validate.call_count == 0
+
+
+@pytest.mark.asyncio
+async def test_explicit_adcp_version_threads_through_to_validator() -> None:
+    """Buyer sets ``adcp_version='3.0'``; validator gets ``version='3.0'``."""
+    handler = _RecorderHandler()
+
+    # Patch must be active when ``create_tool_caller`` runs — its import
+    # of ``validate_request`` is local-scope, so the closure captures
+    # whichever binding existed at construction time.
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="warn"),
+        )
+        await caller(
+            {
+                "adcp_version": "3.0",
+                "buying_mode": "brief",
+                "brief": "Q4",
+            },
+        )
+
+    assert mock_validate.call_count == 1
+    _, kwargs = mock_validate.call_args
+    assert kwargs.get("version") == "3.0"
+
+
+@pytest.mark.asyncio
+async def test_adcp_major_version_int_threads_through_to_validator() -> None:
+    """Pre-3.1 buyer sets only ``adcp_major_version=3`` → highest supported minor."""
+    handler = _RecorderHandler()
+
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="warn"),
+        )
+        await caller(
+            {
+                "adcp_major_version": 3,
+                "buying_mode": "brief",
+                "brief": "Q4",
+            },
+        )
+
+    assert mock_validate.call_count == 1
+    _, kwargs = mock_validate.call_args
+    # 3 → highest supported minor for major 3 in COMPATIBLE_ADCP_VERSIONS = ("3.0","3.1")
+    assert kwargs.get("version") == "3.1"
+
+
+@pytest.mark.asyncio
+async def test_unsupported_major_version_raises_version_unsupported(
+    strict_version_envelope: None,
+) -> None:
+    """Future-major buyer (e.g. ``adcp_major_version=4``) gets a clean
+    ``VERSION_UNSUPPORTED`` error — *before* the handler runs.
+
+    Requires ``ADCP_STRICT_VERSION_ENVELOPE=1``; the default-permissive
+    behaviour is tested separately below.
+    """
+    handler = _RecorderHandler()
+    caller = create_tool_caller(handler, "get_products")
+
+    with pytest.raises(ADCPTaskError) as exc_info:
+        await caller({"adcp_major_version": 4})
+
+    err = exc_info.value.errors[0]
+    assert err.code == "VERSION_UNSUPPORTED"
+    assert "4" in err.message
+    assert err.details is not None
+    # Wire value preserves int type (was sent as int, echoed as int).
+    assert err.details.get("claimed_version") == 4
+    assert "supported_versions" in err.details
+
+    # Handler must NOT have been invoked.
+    assert handler.received == []
+
+
+@pytest.mark.asyncio
+async def test_unsupported_adcp_version_string_raises_version_unsupported(
+    strict_version_envelope: None,
+) -> None:
+    handler = _RecorderHandler()
+    caller = create_tool_caller(handler, "get_products")
+
+    with pytest.raises(ADCPTaskError) as exc_info:
+        await caller({"adcp_version": "1.0"})
+
+    err = exc_info.value.errors[0]
+    assert err.code == "VERSION_UNSUPPORTED"
+    assert err.details is not None
+    assert err.details.get("claimed_version") == "1.0"
+    assert handler.received == []
+
+
+@pytest.mark.asyncio
+async def test_unsupported_version_permissive_falls_through(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """Default (no ``ADCP_STRICT_VERSION_ENVELOPE``) — an unsupported
+    version is logged and the dispatcher falls through to SDK-pin
+    validation. The handler runs; the buyer's wire-version claim
+    becomes a non-fatal warning.
+    """
+    import logging
+
+    handler = _RecorderHandler()
+
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="warn"),
+        )
+        with caplog.at_level(logging.WARNING):
+            await caller({"adcp_major_version": 4, "brief": "Q4"})
+
+    # Handler ran (permissive).
+    assert len(handler.received) == 1
+    # Validator was called with ``version=None`` (fell through to SDK pin).
+    _, kwargs = mock_validate.call_args
+    assert kwargs.get("version") is None
+    # Warning logged with migration hint.
+    assert any("ADCP_STRICT_VERSION_ENVELOPE" in rec.message for rec in caplog.records)
+
+
+@pytest.mark.asyncio
+async def test_version_detection_runs_after_pre_validation_hook() -> None:
+    """A pre-validation hook can populate the version envelope; detection
+    must see the post-hook params, not the wire input."""
+
+    def hook(_tool: str, args: dict[str, Any]) -> dict[str, Any]:
+        # Legacy buyer omitted the field; hook supplies a supported version.
+        return {**args, "adcp_version": "3.0"}
+
+    handler = _RecorderHandler()
+
+    with patch("adcp.validation.schema_validator.validate_request") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="warn"),
+            pre_validation_hook=hook,
+        )
+        await caller({"buying_mode": "brief", "brief": "Q4"})
+
+    _, kwargs = mock_validate.call_args
+    assert kwargs.get("version") == "3.0"
+
+
+@pytest.mark.asyncio
+async def test_response_validation_uses_same_wire_version() -> None:
+    """Response validation should resolve against the same version the
+    request claimed — so a v2.5 buyer's response gets v2.5-schema-checked.
+    """
+    handler = _RecorderHandler()
+
+    with patch("adcp.validation.schema_validator.validate_response") as mock_validate:
+        mock_validate.return_value = type("Outcome", (), {"valid": True, "issues": []})()
+        caller = create_tool_caller(
+            handler,
+            "get_products",
+            validation=ValidationHookConfig(requests="off", responses="warn"),
+        )
+        await caller(
+            {
+                "adcp_version": "3.1",
+                "buying_mode": "brief",
+                "brief": "Q4",
+            },
+        )
+
+    _, kwargs = mock_validate.call_args
+    assert kwargs.get("version") == "3.1"
diff --git a/tests/test_schema_loader_per_version.py b/tests/test_schema_loader_per_version.py
new file mode 100644
index 00000000..f13198bc
--- /dev/null
+++ b/tests/test_schema_loader_per_version.py
@@ -0,0 +1,214 @@
+"""Stage 2 tests: per-version validator loader.
+
+Exercises the ``version=`` kwarg on :func:`get_validator`,
+:func:`validate_request`, and :func:`validate_response`. Builds a
+synthetic legacy bundle in ``tmp_path`` and monkeypatches the loader's
+resolver to find it — keeps the test isolated from the repo's working
+tree (which CI sometimes runs against an installed wheel where the
+packaged path wins) and from concurrent fixture cleanup.
+"""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+
+import pytest
+
+from adcp.validation import schema_loader as _loader_mod
+from adcp.validation.schema_loader import (
+    _reset_for_tests,
+    _resolve_schema_root,
+    _SchemaRoot,
+    _sdk_pinned_bundle_key,
+    get_validator,
+    list_validator_keys,
+)
+from adcp.validation.schema_validator import validate_request, validate_response
+
+
+@pytest.fixture
+def synthetic_legacy_bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> tuple[str, Path]:
+    """Yield a ``(bundle_key, schemas_root)`` pair for a synthetic legacy
+    bundle laid out under ``tmp_path``.
+
+    The loader's resolver is monkeypatched to return our synthetic root
+    when asked for the legacy bundle key. The SDK-pinned key falls back
+    to the real resolver. This isolates the test from the repo's working
+    tree and from the installed-wheel discovery path (which would
+    otherwise win).
+    """
+    legacy_key = "2.5"
+    legacy_root = tmp_path / "cache" / legacy_key
+    bundled = legacy_root / "bundled"
+    core = legacy_root / "core"
+    bundled.mkdir(parents=True)
+    core.mkdir()
+
+    request_schema = {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "title": "Synthetic Tool Request",
+        "type": "object",
+        "required": ["legacy_field"],
+        "properties": {
+            "legacy_field": {"type": "string"},
+            "extra_field": {"type": "integer"},
+        },
+        "additionalProperties": False,
+    }
+    response_schema = {
+        "$schema": "http://json-schema.org/draft-07/schema#",
+        "title": "Synthetic Tool Response",
+        "type": "object",
+        "required": ["result"],
+        "properties": {"result": {"type": "string"}},
+    }
+
+    (bundled / "synthetic-tool-request.json").write_text(
+        json.dumps(request_schema), encoding="utf-8"
+    )
+    (bundled / "synthetic-tool-response.json").write_text(
+        json.dumps(response_schema), encoding="utf-8"
+    )
+
+    real_resolve = _resolve_schema_root
+
+    def _fake_resolve(bundle_key: str | None = None) -> _SchemaRoot | None:
+        if bundle_key == legacy_key:
+            return _SchemaRoot(legacy_root)
+        return real_resolve(bundle_key)
+
+    monkeypatch.setattr(_loader_mod, "_resolve_schema_root", _fake_resolve)
+
+    _reset_for_tests()
+    try:
+        yield legacy_key, legacy_root
+    finally:
+        _reset_for_tests()
+        # ``tmp_path`` is cleaned up by pytest, but a non-empty leftover
+        # from a partial test run shouldn't break the next session.
+        shutil.rmtree(legacy_root, ignore_errors=True)
+
+
+def test_resolve_schema_root_returns_different_paths_per_bundle_key(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    legacy_key, legacy_path = synthetic_legacy_bundle
+    sdk_key = _sdk_pinned_bundle_key()
+    assert sdk_key != legacy_key, (
+        "Test fixture assumes the SDK pin isn't 2.5 — pick a different "
+        "synthetic key if the SDK ever pins to 2.5.x"
+    )
+
+    # Go through the module attribute so the fixture's monkeypatch fires.
+    legacy_root = _loader_mod._resolve_schema_root(legacy_key)
+    sdk_root = _loader_mod._resolve_schema_root(sdk_key)
+
+    assert legacy_root is not None
+    assert sdk_root is not None
+    assert legacy_root.root == legacy_path
+    assert legacy_root.root != sdk_root.root
+
+
+def test_get_validator_per_version_finds_only_its_own_tools(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    legacy_key, _ = synthetic_legacy_bundle
+
+    # Legacy bundle has only synthetic_tool.
+    legacy_keys = list_validator_keys(version=legacy_key)
+    assert "synthetic_tool::request" in legacy_keys
+    assert "synthetic_tool::sync" in legacy_keys
+
+    # SDK pin doesn't have synthetic_tool.
+    sdk_keys = list_validator_keys()  # None → SDK pin
+    assert "synthetic_tool::request" not in sdk_keys
+
+
+def test_get_validator_returns_independent_validators_per_version(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    legacy_key, _ = synthetic_legacy_bundle
+
+    # Legacy validator: synthetic_tool exists.
+    legacy_validator = get_validator("synthetic_tool", "request", version=legacy_key)
+    assert legacy_validator is not None
+
+    # SDK pin: synthetic_tool does NOT exist; same call returns None.
+    sdk_validator = get_validator("synthetic_tool", "request")
+    assert sdk_validator is None
+
+
+def test_get_validator_same_tool_different_versions_compiles_separately(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    """Pick a tool that exists in the SDK pin; assert a legacy-bundle call
+    that *doesn't* ship that tool returns None — proving the loader keys
+    on bundle, not just tool name.
+    """
+    legacy_key, _ = synthetic_legacy_bundle
+
+    # ``get_products`` exists in the SDK pin (3.0+).
+    sdk_validator = get_validator("get_products", "request")
+    assert sdk_validator is not None
+
+    # The synthetic legacy bundle doesn't ship get_products → None.
+    legacy_validator = get_validator("get_products", "request", version=legacy_key)
+    assert legacy_validator is None
+
+
+def test_validate_request_threads_version_through(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    legacy_key, _ = synthetic_legacy_bundle
+
+    # Legacy schema requires ``legacy_field`` and forbids unknown keys.
+    valid = validate_request("synthetic_tool", {"legacy_field": "hello"}, version=legacy_key)
+    assert valid.valid
+
+    missing_required = validate_request("synthetic_tool", {"extra_field": 7}, version=legacy_key)
+    assert not missing_required.valid
+    assert any("legacy_field" in (issue.message or "") for issue in missing_required.issues)
+
+    extra_field_rejected = validate_request(
+        "synthetic_tool",
+        {"legacy_field": "x", "not_in_schema": 1},
+        version=legacy_key,
+    )
+    assert not extra_field_rejected.valid
+
+
+def test_validate_request_unknown_version_skips_safely(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    """A version we don't have on disk degrades to ``skipped`` rather
+    than crashing — same contract as a missing schema for the SDK pin."""
+    outcome = validate_request("synthetic_tool", {"x": 1}, version="9.9.9")
+    # ``skipped`` semantics: ``valid=True`` with no issues.
+    assert outcome.valid
+    assert outcome.issues == []
+
+
+def test_validate_response_threads_version_through(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    legacy_key, _ = synthetic_legacy_bundle
+
+    valid = validate_response("synthetic_tool", {"result": "ok"}, version=legacy_key)
+    assert valid.valid
+
+    bad = validate_response("synthetic_tool", {"result": 42}, version=legacy_key)
+    assert not bad.valid
+
+
+def test_default_version_unchanged_when_arg_omitted(
+    synthetic_legacy_bundle: tuple[str, Path],
+) -> None:
+    """``version=None`` (or omitted) keeps the SDK-pin behaviour exactly.
+    Regression guard: Stage 2 must not change validator selection for
+    existing call sites that don't pass ``version=``."""
+    # Pick a tool that ships in the SDK pin.
+    explicit_default = get_validator("get_products", "request", version=None)
+    omitted = get_validator("get_products", "request")
+    assert explicit_default is omitted
diff --git a/tests/test_spec_compat_hooks.py b/tests/test_spec_compat_hooks.py
index 6734c96a..af68b722 100644
--- a/tests/test_spec_compat_hooks.py
+++ b/tests/test_spec_compat_hooks.py
@@ -42,7 +42,7 @@ def _wrap_sync_payload(creatives: list[dict[str, Any]]) -> dict[str, Any]:
     the envelope.
     """
     return {
-        "adcp_major_version": 4,
+        "adcp_major_version": 3,
         "account": {"account_id": "acc-1"},
         "idempotency_key": "idem-1234567890abcdef",
         "creatives": creatives,
@@ -333,7 +333,7 @@ def test_get_products_hook_output_passes_pydantic_validation() -> None:
     """Pre-v3 buyer payload (no buying_mode) is 4.4-valid after the hook."""
     hooks = spec_compat_hooks()
     # Pre-v3 buyer omits buying_mode; pairs with a 'brief' string for brief mode.
-    pre_v3_payload = {"adcp_major_version": 4, "brief": "Sponsorship Q4"}
+    pre_v3_payload = {"adcp_major_version": 3, "brief": "Sponsorship Q4"}
     coerced = _gp(hooks, pre_v3_payload)
     # Must not raise: the hook produced a valid GetProductsRequest payload.
     model = GetProductsRequest.model_validate(coerced)
diff --git a/tests/test_sync_schemas.py b/tests/test_sync_schemas.py
index 6f9842bc..3c28c919 100644
--- a/tests/test_sync_schemas.py
+++ b/tests/test_sync_schemas.py
@@ -122,7 +122,7 @@ def test_caller_resolves_bundle_key_from_target_not_effective(self) -> None:
         from adcp.validation.version import resolve_bundle_key
 
         # ``effective_version`` after a 404-fallback:
-        with pytest.raises(ValueError, match="not a valid semver"):
+        with pytest.raises(ValueError, match="not a valid version"):
             resolve_bundle_key("latest")
 
         # ``target_version`` (the SDK pin) always parses:
diff --git a/tests/test_validation_envelope.py b/tests/test_validation_envelope.py
new file mode 100644
index 00000000..cc806b05
--- /dev/null
+++ b/tests/test_validation_envelope.py
@@ -0,0 +1,102 @@
+"""Tests for ``adcp.validation.envelope.detect_wire_version``."""
+
+from __future__ import annotations
+
+import pytest
+
+from adcp.validation.envelope import UnsupportedVersionError, detect_wire_version
+
+# A canned supported set keeps the test independent of COMPATIBLE_ADCP_VERSIONS
+# drift over time. Pinning the set inside the test also documents what
+# range each case is claiming against.
+_SUPPORTED = ("3.0", "3.1")
+
+
+def test_explicit_adcp_version_release_precision() -> None:
+    assert detect_wire_version({"adcp_version": "3.0"}, supported=_SUPPORTED) == "3.0"
+    assert detect_wire_version({"adcp_version": "3.1"}, supported=_SUPPORTED) == "3.1"
+
+
+def test_explicit_adcp_version_patch_precision_normalized() -> None:
+    """Patch-precision wire values collapse to release-precision."""
+    assert detect_wire_version({"adcp_version": "3.0.7"}, supported=_SUPPORTED) == "3.0"
+    assert detect_wire_version({"adcp_version": "3.1.0"}, supported=_SUPPORTED) == "3.1"
+
+
+def test_explicit_adcp_version_wins_over_major_version() -> None:
+    """If both fields are set, the precision wins (3.1+ contract)."""
+    payload = {"adcp_version": "3.1.0", "adcp_major_version": 3}
+    assert detect_wire_version(payload, supported=_SUPPORTED) == "3.1"
+
+
+def test_adcp_major_version_picks_highest_supported_minor() -> None:
+    """Pre-3.1 buyer sends only ``adcp_major_version`` — pick highest minor."""
+    assert detect_wire_version({"adcp_major_version": 3}, supported=_SUPPORTED) == "3.1"
+
+
+def test_adcp_major_version_unsupported_major_raises() -> None:
+    with pytest.raises(UnsupportedVersionError) as exc_info:
+        detect_wire_version({"adcp_major_version": 4}, supported=_SUPPORTED)
+    assert exc_info.value.wire_value == 4
+    assert exc_info.value.supported == _SUPPORTED
+
+
+def test_adcp_version_unsupported_release_raises() -> None:
+    with pytest.raises(UnsupportedVersionError) as exc_info:
+        detect_wire_version({"adcp_version": "2.5"}, supported=_SUPPORTED)
+    assert exc_info.value.wire_value == "2.5"
+
+
+def test_adcp_version_malformed_raises() -> None:
+    with pytest.raises(UnsupportedVersionError):
+        detect_wire_version({"adcp_version": "not-a-version"}, supported=_SUPPORTED)
+
+
+def test_neither_field_returns_none_fallback_to_sdk_pin() -> None:
+    assert detect_wire_version({}, supported=_SUPPORTED) is None
+    assert detect_wire_version({"other_field": "x"}, supported=_SUPPORTED) is None
+
+
+def test_non_dict_payload_returns_none() -> None:
+    """Non-dict payloads can't carry the envelope — caller skips."""
+    assert detect_wire_version("not_a_dict", supported=_SUPPORTED) is None
+    assert detect_wire_version(None, supported=_SUPPORTED) is None
+    assert detect_wire_version([], supported=_SUPPORTED) is None
+
+
+def test_adcp_version_empty_string_treated_as_missing() -> None:
+    """An empty string falls through to ``adcp_major_version`` lookup."""
+    assert (
+        detect_wire_version({"adcp_version": "", "adcp_major_version": 3}, supported=_SUPPORTED)
+        == "3.1"
+    )
+
+
+def test_adcp_major_version_bool_rejected() -> None:
+    """``True``/``False`` are int subclasses; reject so they don't map to 1/0."""
+    assert detect_wire_version({"adcp_major_version": True}, supported=_SUPPORTED) is None
+    assert detect_wire_version({"adcp_major_version": False}, supported=_SUPPORTED) is None
+
+
+def test_adcp_major_version_string_raises_loudly() -> None:
+    """A buyer that JSON-stringified the int field shouldn't silently get
+    SDK-pin validation — flag it as VERSION_UNSUPPORTED so the bug
+    surfaces."""
+    with pytest.raises(UnsupportedVersionError) as exc_info:
+        detect_wire_version({"adcp_major_version": "3"}, supported=_SUPPORTED)
+    assert exc_info.value.wire_value == "3"
+
+
+def test_adcp_major_version_zero_or_negative_raises() -> None:
+    """Below the spec's ``minimum:1`` bound — reject as unsupported."""
+    with pytest.raises(UnsupportedVersionError):
+        detect_wire_version({"adcp_major_version": 0}, supported=_SUPPORTED)
+    with pytest.raises(UnsupportedVersionError):
+        detect_wire_version({"adcp_major_version": -3}, supported=_SUPPORTED)
+
+
+def test_supports_prerelease_in_supported_set() -> None:
+    """When ``supported`` includes a prerelease-keyed entry, exact match works."""
+    payload = {"adcp_version": "3.1.0-beta.1"}
+    # 3.1.0-beta.1 normalizes to 3.1-beta.1 — present in supported.
+    assert detect_wire_version(payload, supported=("3.0", "3.1-beta.1")) == "3.1-beta.1"
diff --git a/tests/test_validation_version.py b/tests/test_validation_version.py
index 5242e617..82f20d63 100644
--- a/tests/test_validation_version.py
+++ b/tests/test_validation_version.py
@@ -29,12 +29,22 @@ def test_resolve_bundle_key_strips_whitespace() -> None:
     assert resolve_bundle_key("  3.0.7  ") == "3.0"
 
 
+def test_resolve_bundle_key_accepts_major_minor_pass_through() -> None:
+    """Bare ``MAJOR.MINOR`` is already a bundle key — passed through.
+
+    The wire envelope's ``adcp_version`` field (3.1+) is emitted at this
+    precision, so the dispatcher can hand it straight to the loader.
+    """
+    assert resolve_bundle_key("3.0") == "3.0"
+    assert resolve_bundle_key("2.5") == "2.5"
+
+
 def test_resolve_bundle_key_rejects_garbage() -> None:
-    with pytest.raises(ValueError, match="not a valid semver"):
+    with pytest.raises(ValueError, match="not a valid version"):
         resolve_bundle_key("latest")
-    with pytest.raises(ValueError, match="not a valid semver"):
-        resolve_bundle_key("3.0")
-    with pytest.raises(ValueError, match="not a valid semver"):
+    with pytest.raises(ValueError, match="not a valid version"):
         resolve_bundle_key("v3.0.7")
-    with pytest.raises(ValueError, match="not a valid semver"):
+    with pytest.raises(ValueError, match="not a valid version"):
+        resolve_bundle_key("3")
+    with pytest.raises(ValueError, match="not a valid version"):
         resolve_bundle_key("")