protocol_version: enforce on HTTP dispatch path + conformance coverage

HTTP transport doesn't route through RpcServer.serve_one — it has its own
dispatch loop in vgi_rpc/http/server/_app_unary.py and _app_stream.py.
The initial commit's dispatch-boundary check was wired only in serve_one,
so HTTP clients with mismatched protocol_version dispatched straight to
the worker handler with mismatched schemas. Caught by senior review,
fixed here.

Adds the check at the same point in both HTTP unary and stream init —
after _read_request populates _current_request_metadata, before
_deserialize_params / handler dispatch. __describe__ is exempt
(diagnostic path for mismatched clients to discover the server's
version). Mismatch raises ProtocolVersionError → existing
(VersionError, RpcError) catch wraps as _RpcHttpError(BAD_REQUEST) →
client sees IOException with the directional message text intact.
Verified by mutation test against the HTTP integration suite (edit
VGI_PROTOCOL_VERSION → "2.0.0", rebuild, ATTACH fails with full
directional text).

Adds:
- 4 HTTP transport unit tests in tests/test_protocol_version.py
  (matched success, mismatched directional error, describe bypass,
  undeclared opt-out). These are the tests that would have caught the
  blocker had they existed before.
- protocol_version conformance category: ConformanceService now
  declares protocol_version="1.0.0", forcing cross-language ports to
  support the new metadata key. Two describe tests
  (surfaces_declared_version, format) and one round-trip
  (matched_dispatch_succeeds) across every transport. Mismatch
  scenarios stay in unit tests because the conformance runner is
  parameterised on a single proxy — spawning two servers at different
  versions doesn't fit the fixture model.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/test_protocol_version.py

@@ -387,3 +387,89 @@ def serve() -> None:
         finally:
             client_t.close()
             server_t.close()
+
+
+# ---------------------------------------------------------------------------
+# HTTP transport — the path external clients actually use.
+# ---------------------------------------------------------------------------
+
+
+class TestHttpTransportVersionCheck:
+    """HTTP unary and stream init must enforce the dispatch-boundary version check.
+
+    HTTP doesn't route through ``RpcServer.serve_one`` — it has its own
+    dispatch loop in ``vgi_rpc/http/server/_app_unary.py`` and ``_app_stream.py``.
+    Without an HTTP-specific check, a mismatched HTTP client would dispatch
+    straight to the worker handler with the wrong schemas, and the only
+    defense would be the Arrow column-count check (silent corruption in some
+    cases). These tests gate that HTTP-path regression.
+    """
+
+    def test_matched_versions_succeed(self) -> None:
+        """HTTP unary RPC succeeds when client and server speak the same protocol_version."""
+        from vgi_rpc.http import http_connect, make_sync_client
+
+        server = RpcServer(_ProtoV100, _Impl())
+        client = make_sync_client(server, token_key=b"test")
+        try:
+            with http_connect(_ProtoV100, client=client) as proxy:
+                assert proxy.greet(name="x") == "hi x"
+        finally:
+            client.close()
+
+    def test_mismatched_versions_raise_with_direction(self) -> None:
+        """HTTP unary RPC fails with directional message when versions mismatch.
+
+        The C++ extension's user-visible error path runs through this: a
+        DuckDB user issuing ``ATTACH`` against a worker on the wrong version
+        sees the directional text inline, not a generic 'request failed'.
+        """
+        from vgi_rpc.http import http_connect, make_sync_client
+
+        server = RpcServer(_ProtoV200, _Impl())
+        # Client built against the 1.0.0 Protocol talks to a 2.0.0 server.
+        client = make_sync_client(server, token_key=b"test")
+        try:
+            with http_connect(_ProtoV100, client=client) as proxy, pytest.raises(RpcError) as exc_info:
+                proxy.greet(name="x")
+            err_text = str(exc_info.value) + " " + exc_info.value.error_message
+            assert "Client: 1.0.0" in err_text
+            assert "Server: 2.0.0" in err_text
+            assert "upgrade the VGI extension/client" in err_text
+        finally:
+            client.close()
+
+    def test_describe_bypass_over_http(self) -> None:
+        """HTTP __describe__ must dispatch regardless of client/server version mismatch."""
+        from vgi_rpc.http import http_introspect, make_sync_client
+
+        server = RpcServer(_ProtoV200, _Impl(), enable_describe=True)
+        client = make_sync_client(server, token_key=b"test")
+        try:
+            # http_introspect is the framework-internal discovery path; like
+            # the pipe-transport introspect(), it doesn't carry a client
+            # protocol_version. The server must serve describe anyway so
+            # mismatched clients can introspect the expected version.
+            desc = http_introspect(client=client)
+            assert desc.protocol_name == "_ProtoV200"
+            assert desc.protocol_version == "2.0.0"
+        finally:
+            client.close()
+
+    def test_undeclared_server_does_not_check_over_http(self) -> None:
+        """Server without protocol_version -> HTTP path is opt-out too.
+
+        Symmetry with the pipe path. A Protocol that hasn't opted into
+        versioning must not gain the check just because the transport
+        happens to be HTTP.
+        """
+        from vgi_rpc.http import http_connect, make_sync_client
+
+        server = RpcServer(_ProtoNoVersion, _Impl())
+        # Client declares 2.0.0; server opts out → check doesn't fire.
+        client = make_sync_client(server, token_key=b"test")
+        try:
+            with http_connect(_ProtoV200, client=client) as proxy:
+                assert proxy.greet(name="x") == "hi x"
+        finally:
+            client.close()

vgi_rpc/conformance/_protocol.py

@@ -13,7 +13,7 @@
 
 import datetime as _dt
 from decimal import Decimal
-from typing import Annotated, Protocol
+from typing import Annotated, ClassVar, Protocol
 
 import pyarrow as pa
 
@@ -35,7 +35,15 @@
 
 
 class ConformanceService(Protocol):
-    """Wire-protocol conformance service exercising all framework capabilities."""
+    """Wire-protocol conformance service exercising all framework capabilities.
+
+    Declares ``protocol_version`` so cross-language conformant implementations
+    must support the per-request ``vgi_rpc.protocol_version`` metadata key
+    that vgi-rpc enforces at the dispatch boundary. Bump major when adding
+    or removing methods; bump minor for additive parameter changes.
+    """
+
+    protocol_version: ClassVar[str] = "1.0.0"
 
     # ------------------------------------------------------------------
     # Unary: Scalar Echo

vgi_rpc/conformance/_runner.py

@@ -1819,6 +1819,67 @@ def _test_desc_protocol_hash_format(desc: ServiceDescription) -> None:
     assert all(c in "0123456789abcdef" for c in desc.protocol_hash), "protocol_hash must be lowercase hex"
 
 
+# ---------------------------------------------------------------------------
+# Application-protocol-version tests
+# ---------------------------------------------------------------------------
+# The mismatch / directional-error scenarios for protocol_version don't fit
+# the single-proxy-per-transport shape of the conformance runner — they require
+# spawning two servers at different versions. Those cases live in
+# ``tests/test_protocol_version.py`` (which now covers pipe AND HTTP). The
+# conformance tests below cover the wire-shape contract every transport must
+# satisfy: describe surfaces the field, request batches carry the metadata key,
+# and matched-version dispatch succeeds across all transports.
+
+
+@_describe_test(category="describe_protocol_version", name="surfaces_declared_version")
+def _test_desc_protocol_version_surfaces(desc: ServiceDescription) -> None:
+    """ConformanceService declares protocol_version, so describe must surface it.
+
+    Round-trip through ``__describe__`` response custom_metadata to
+    ``ServiceDescription.protocol_version``. Cross-language ports that
+    don't yet read the ``vgi_rpc.protocol_version`` key fail this test,
+    which is the intended forcing function.
+    """
+    assert desc.protocol_version, "protocol_version must be present in describe response"
+    # ConformanceService.protocol_version is "1.0.0"; the field is the source
+    # of truth, so don't hardcode the value here — fetch it.
+    from vgi_rpc.conformance._protocol import ConformanceService
+
+    expected = vars(ConformanceService).get("protocol_version")
+    assert desc.protocol_version == expected, (
+        f"describe surfaced protocol_version={desc.protocol_version!r}, "
+        f"expected {expected!r} from ConformanceService.protocol_version"
+    )
+
+
+@_describe_test(category="describe_protocol_version", name="format")
+def _test_desc_protocol_version_format(desc: ServiceDescription) -> None:
+    """The protocol_version field must be canonical semver MAJOR.MINOR.PATCH."""
+    parts = desc.protocol_version.split(".")
+    assert len(parts) == 3, f"expected MAJOR.MINOR.PATCH, got {desc.protocol_version!r}"
+    for component in parts:
+        assert component.isdigit(), f"non-numeric semver component in {desc.protocol_version!r}"
+        # No leading zeros except literal "0".
+        assert component == "0" or not component.startswith("0"), (
+            f"leading zero in semver component of {desc.protocol_version!r}"
+        )
+
+
+@_conformance_test(category="protocol_version", name="matched_dispatch_succeeds")
+def _test_proto_version_matched_dispatch(proxy: ConformanceService, logs: LogCollector) -> None:
+    """A normal dispatched RPC exercises the dispatch-boundary check on the matched-version path.
+
+    The conformance client and server are both built against the same
+    ConformanceService, so their declared protocol_version matches. A
+    successful echo proves the server's dispatch-boundary check accepted
+    the client's ``vgi_rpc.protocol_version`` metadata across whatever
+    transport the runner picked. (Mismatch scenarios live in unit tests —
+    see ``tests/test_protocol_version.py``.)
+    """
+    result = proxy.echo_string(value="versioned")
+    assert result == "versioned"
+
+
 # ---------------------------------------------------------------------------
 # Runner
 # ---------------------------------------------------------------------------

vgi_rpc/http/server/_app_stream.py

@@ -20,7 +20,7 @@
 
 from vgi_rpc.external import predict_externalize_bytes_for_collector, resolve_external_location
 from vgi_rpc.log import Message
-from vgi_rpc.metadata import CANCEL_KEY, STATE_KEY, strip_keys
+from vgi_rpc.metadata import CANCEL_KEY, PROTOCOL_VERSION_KEY, STATE_KEY, strip_keys
 from vgi_rpc.rpc import (
     _EMPTY_SCHEMA,
     _TICK_BATCH,
@@ -50,6 +50,7 @@
     CallStatistics,
     HookToken,
     _current_call_stats,
+    _current_request_metadata,
     _current_stream_id,
     _DispatchHook,
     _record_input,
@@ -205,6 +206,14 @@ def _run_stream_init_sync(
                     f"Method name mismatch: URL path has '{method_name}' but Arrow IPC "
                     f"custom_metadata 'vgi_rpc.method' has '{ipc_method}'. These must match."
                 )
+            # Application-protocol-version gate (mirror of RpcServer.serve_one). HTTP
+            # stream init dispatches directly here; the check has to be wired in
+            # independently. Stream methods include no synthetic ``__describe__``,
+            # but the same exemption is kept for consistency with the unary path.
+            # Server opts out by not declaring ``protocol_version`` on its Protocol.
+            if app._server._protocol_version_parts is not None and method_name != "__describe__":
+                md = _current_request_metadata.get()
+                app._server._check_protocol_version(md.get(PROTOCOL_VERSION_KEY) if md is not None else None)
             _deserialize_params(kwargs, info.param_types, app._server.ipc_validation)
             _validate_params(info.name, kwargs, info.param_types)
         except (pa.ArrowInvalid, TypeError, StopIteration, RpcError, VersionError) as exc:

vgi_rpc/http/server/_app_unary.py

@@ -15,6 +15,7 @@
 from pyarrow import ipc
 
 from vgi_rpc.external import predict_externalize_bytes_for_batch
+from vgi_rpc.metadata import PROTOCOL_VERSION_KEY
 from vgi_rpc.rpc import (
     CallContext,
     RpcError,
@@ -37,6 +38,7 @@
     CallStatistics,
     HookToken,
     _current_call_stats,
+    _current_request_metadata,
     _DispatchHook,
     _record_output,
 )
@@ -77,6 +79,15 @@ def _run_unary_sync(
                     f"Method name mismatch: URL path has '{method_name}' but Arrow IPC "
                     f"custom_metadata 'vgi_rpc.method' has '{ipc_method}'. These must match."
                 )
+            # Application-protocol-version gate (mirror of RpcServer.serve_one). HTTP
+            # transport dispatches directly here instead of going through serve_one,
+            # so the check has to be wired in independently. ``__describe__`` is
+            # exempt — it's the diagnostic path a mismatched client uses to learn
+            # the server's expected version. Server opts out by not declaring
+            # ``protocol_version`` on its Protocol class.
+            if app._server._protocol_version_parts is not None and method_name != "__describe__":
+                md = _current_request_metadata.get()
+                app._server._check_protocol_version(md.get(PROTOCOL_VERSION_KEY) if md is not None else None)
             _deserialize_params(kwargs, info.param_types, app._server.ipc_validation)
             _validate_params(info.name, kwargs, info.param_types)
         except (pa.ArrowInvalid, TypeError, StopIteration, RpcError, VersionError) as exc: