Skip to content

Commit 1d50004

Browse files
rustyconoverclaude
andcommitted
feat(client): Client.list_protocols / describe_protocol via reflection
Step 0 of the attach-tickets plan: every worker hosts vgi_rpc.Reflection.v1 on every transport, but the Python Client had no way to ask what a worker hosts. Optional protocols (attach tickets, reporting) are discovered this way, so the client needs it first. - Client.list_protocols() -> list[HostedProtocol(name, version, hash)], in server order (vgi.v2 first, Worker.hosted_protocols() extras, then the framework's own). HostedProtocol is a frozen dataclass exported from vgi.client. - Client.describe_protocol(name) -> vgi_rpc.introspect.ServiceDescription. - Both reuse the primary connection instead of opening a new one: HTTP binds a second http_connect proxy to the shared httpx2.Client; every other transport (subprocess direct/pooled, launch, TCP, Iroh) binds an RpcConnection(Reflection, ...) to the primary proxy's RpcTransport. - A worker that predates reflection (UNIMPLEMENTED / protocol_not_supported / method_not_implemented) yields [HostedProtocol("vgi.v2", "", "")] instead of raising; any other error is a ClientError. vgi-rpc 0.49 exports no public Reflection class, no public listing entry point (introspect()/http_introspect() only describe one protocol), and no way to rebind a proxy to another protocol. So this imports Reflection from vgi_rpc.rpc._reflection (as tests/test_hosted_protocols.py already does), reads the proxy's _transport, and adapts describe() with vgi_rpc.introspect._adapt_description. A public vgi-rpc entry point would remove all three. Tests (tests/test_client_reflection.py) cover launch, pooled and direct stdio, and HTTP, and check the listing against the fixture worker's own build_rpc_server bindings, hash format (64 lowercase hex), and the fallback. No tests/conformance/test_reflection.py: the cross-language coverage is the C++ repo's test/sql/integration/reflection/, recorded as a _EXEMPTIONS entry in test_directory_parity.py. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
1 parent 6be6eec commit 1d50004

8 files changed

Lines changed: 276 additions & 6 deletions

File tree

‎CLAUDE.md‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -523,6 +523,15 @@ hook. `MetaWorker` hosts its children's lists concatenated. Identity is not
523523
supplied through this hook. It stays opt-in via `resolve_token`/`mint_grant`
524524
and is hosted on HTTP only. See `docs/api/worker.md`.
525525

526+
Client side, `Client.list_protocols()` returns the hosted set as
527+
`vgi_rpc.HostedProtocol` (re-exported from `vgi.client`) in server order, over
528+
any transport, and `Client.describe_protocol(name)` returns one protocol's
529+
methods. Both are thin wrappers over vgi-rpc's public `list_protocols` /
530+
`describe_protocol` on the primary proxy (no private vgi-rpc imports). A worker
531+
without reflection (`ReflectionNotSupportedError`) yields
532+
`[HostedProtocol("vgi.v2", "", "")]` rather than an error. Tests: `tests/test_client_reflection.py`; cross-language coverage is the
533+
C++ repo's `test/sql/integration/reflection/`. See `docs/api/client.md`.
534+
526535
### Access Log Sampling and Async Emission
527536

528537
`--access-log-sample 0.05` keeps 5% of *successful* calls. Errors are always

‎docs/api/client.md‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,4 +3,36 @@
33
The Python `Client` spawns or connects to a worker, streams Arrow data to and from its functions,
44
and surfaces errors as `ClientError`. It is the pure-Python counterpart to the DuckDB extension.
55

6+
## Discovering hosted protocols
7+
8+
Every worker hosts the vgi-rpc reflection protocol, `vgi_rpc.Reflection.v1`, on every transport.
9+
`Client.list_protocols()` asks it what the worker hosts, in the worker's order: `vgi.v2` first,
10+
then any protocols the worker adds through `Worker.hosted_protocols()`, then the framework's own.
11+
Each entry is a `HostedProtocol` (vgi-rpc's, re-exported from `vgi.client`) with `name`,
12+
`version`, `hash`, `deprecated`, `deprecation_message` and `features`. The hash is a SHA-256 hex digest of the
13+
protocol's canonical description, so two equal hashes mean the same method surface.
14+
`Client.describe_protocol(name)` returns one protocol's methods and schemas as a
15+
`vgi_rpc.ServiceDescription`. Both are thin wrappers over vgi-rpc's `list_protocols(proxy)` /
16+
`describe_protocol(proxy, name)` on the client's primary connection.
17+
18+
```python test="skip"
19+
from vgi.client import Client
20+
21+
with Client("vgi-fixture-worker") as client:
22+
hosted = {p.name for p in client.list_protocols()}
23+
if "conformance.Secondary.v1" in hosted:
24+
print(client.describe_protocol("conformance.Secondary.v1"))
25+
```
26+
27+
A worker that does not host reflection (one built before it existed answers `UNIMPLEMENTED` /
28+
`protocol_not_supported`, which vgi-rpc raises as `ReflectionNotSupportedError`) is not an
29+
error for `list_protocols()`. It returns the single entry
30+
`HostedProtocol("vgi.v2", "", "")`, because `vgi.v2` is the one protocol every worker hosts. The
31+
empty hash marks the entry as inferred rather than reported.
32+
33+
Both calls use the client's primary connection. Do not call them while a stream is open on
34+
the same client.
35+
36+
## API reference
37+
638
::: vgi.client

‎pyproject.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ dependencies = [
4040
"pyarrow",
4141
"typer>=0.9",
4242
"platformdirs",
43-
"vgi-rpc>=0.49.0",
43+
"vgi-rpc>=0.50.0",
4444
"httpx2>=2.12.0",
4545
]
4646

‎tests/conformance/test_directory_parity.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,12 @@
8585
"it is a DuckDB setting with no counterpart in the Python client, which is "
8686
"handed a transport directly."
8787
),
88+
"reflection": (
89+
"Covered by sqllogictests: vgi_rpc.Reflection.v1 discovery is asserted "
90+
"cross-language by vgi/test/sql/integration/reflection/, which every SDK's "
91+
"fixture worker runs. Python Client coverage (Client.list_protocols / "
92+
"describe_protocol on stdio, launch and HTTP) is in tests/test_client_reflection.py."
93+
),
8894
"simple_writable": (
8995
"INSERT/UPDATE/DELETE/RETURNING write paths are mirrored by "
9096
"test_writable.py, which drives the same operations against the writable "

‎tests/test_client_reflection.py‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
# Copyright 2025, 2026 Query Farm LLC - https://query.farm
2+
3+
"""``Client.list_protocols`` / ``Client.describe_protocol`` over reflection.
4+
5+
Every worker hosts ``vgi_rpc.Reflection.v1`` on every transport. These tests
6+
ask the fixture worker what it hosts through the Python ``Client`` on each
7+
transport it speaks, and check the answer against the worker's own
8+
``RpcServer`` bindings. The cross-language coverage of the same surface is the
9+
C++ repo's ``test/sql/integration/reflection/``.
10+
"""
11+
12+
from __future__ import annotations
13+
14+
import re
15+
from collections.abc import Callable
16+
from typing import Any
17+
18+
import pytest
19+
from vgi_rpc import ReflectionNotSupportedError
20+
from vgi_rpc.rpc import RpcError
21+
22+
from tests.conftest import SUBPROCESS_FIXTURE_WORKER
23+
from vgi._test_fixtures.worker import ExampleWorker
24+
from vgi.client import Client, ClientError, HostedProtocol
25+
from vgi.protocol import VgiProtocol
26+
from vgi.rpc_server import build_rpc_server
27+
28+
_SECONDARY = "conformance.Secondary.v1"
29+
_REFLECTION = "vgi_rpc.Reflection.v1"
30+
_EXPECTED_NAMES = ["vgi.v2", _SECONDARY, _REFLECTION]
31+
_HASH = re.compile(r"^[0-9a-f]{64}$")
32+
33+
34+
def _server_view() -> list[HostedProtocol]:
35+
"""Return what the fixture worker's own server binds, as the client would report it."""
36+
server = build_rpc_server(ExampleWorker(quiet=True), transport="pipe")
37+
return [
38+
HostedProtocol(name=b.name, version=b.version or "", hash=b.protocol_hash) for b in server.bindings.values()
39+
]
40+
41+
42+
def _assert_fixture_listing(listing: list[HostedProtocol]) -> None:
43+
"""The fixture's hosted set: ``vgi.v2`` first, the secondary, then reflection."""
44+
assert all(isinstance(p, HostedProtocol) for p in listing)
45+
assert [p.name for p in listing] == _EXPECTED_NAMES
46+
assert listing[0].version == VgiProtocol.protocol_version
47+
for p in listing:
48+
assert _HASH.match(p.hash), f"{p.name} hash {p.hash!r} is not 64 lowercase hex chars"
49+
assert listing == _server_view()
50+
51+
52+
class TestListProtocols:
53+
"""Each transport the fixture worker speaks reports the same hosted set."""
54+
55+
def test_client_transport(self, client_transport: Callable[[], Client]) -> None:
56+
"""Launch / pooled subprocess / HTTP, per the shared transport matrix."""
57+
with client_transport() as client:
58+
_assert_fixture_listing(client.list_protocols())
59+
60+
def test_stdio_direct(self) -> None:
61+
"""A directly spawned stdin/stdout worker, independent of the default matrix."""
62+
with Client(SUBPROCESS_FIXTURE_WORKER, pool=None) as client:
63+
first = client.list_protocols()
64+
_assert_fixture_listing(first)
65+
# The pipe is still usable for vgi.v2 after a reflection call.
66+
assert client.list_protocols() == first
67+
68+
def test_http(self, _shared_http_base_url: Callable[[], str]) -> None:
69+
"""``Client.from_http`` against the fixture HTTP server."""
70+
pytest.importorskip("vgi_rpc.http")
71+
with Client.from_http(_shared_http_base_url()) as client:
72+
_assert_fixture_listing(client.list_protocols())
73+
74+
def test_requires_started_client(self) -> None:
75+
"""An unstarted client raises ``ClientError`` rather than spawning a worker."""
76+
with pytest.raises(ClientError, match="not started"):
77+
Client(SUBPROCESS_FIXTURE_WORKER, pool=None).list_protocols()
78+
79+
80+
class TestDescribeProtocol:
81+
"""``describe_protocol`` returns one protocol's methods."""
82+
83+
def test_describes_secondary(self, client_transport: Callable[[], Client]) -> None:
84+
"""The secondary's description matches its listed hash."""
85+
with client_transport() as client:
86+
listed = {p.name: p for p in client.list_protocols()}
87+
described = client.describe_protocol(_SECONDARY)
88+
assert described.protocol_name == _SECONDARY
89+
assert described.protocol_hash == listed[_SECONDARY].hash
90+
assert "echo_string" in described.methods
91+
92+
def test_describes_vgi_v2(self) -> None:
93+
"""``vgi.v2`` describes with its version and the client's own method set."""
94+
with Client(SUBPROCESS_FIXTURE_WORKER, pool=None) as client:
95+
described = client.describe_protocol("vgi.v2")
96+
assert described.protocol_version == VgiProtocol.protocol_version
97+
assert {"bind", "init"} <= set(described.methods)
98+
99+
def test_unhosted_protocol_raises(self) -> None:
100+
"""A protocol the worker does not host is a ``ClientError``."""
101+
with (
102+
Client(SUBPROCESS_FIXTURE_WORKER, pool=None) as client,
103+
pytest.raises(ClientError, match="nope.v1"),
104+
):
105+
client.describe_protocol("nope.v1")
106+
107+
108+
def _patch_listing(monkeypatch: pytest.MonkeyPatch, error: RpcError) -> None:
109+
"""Make vgi-rpc's ``list_protocols``, as the client calls it, raise *error*."""
110+
111+
def _fail(target: object) -> Any:
112+
raise error
113+
114+
monkeypatch.setattr("vgi.client.client.rpc_list_protocols", _fail)
115+
116+
117+
class TestPreReflectionWorker:
118+
"""A worker without reflection is reported as hosting only ``vgi.v2``.
119+
120+
Which server answers mean "no reflection" is vgi-rpc's call (it raises
121+
``ReflectionNotSupportedError``, tested there); the client's job is only
122+
to map that one exception to the fallback and everything else to
123+
``ClientError``.
124+
"""
125+
126+
def test_falls_back_to_vgi_v2(self, monkeypatch: pytest.MonkeyPatch) -> None:
127+
"""``ReflectionNotSupportedError`` yields one ``vgi.v2`` entry with an empty hash."""
128+
error = ReflectionNotSupportedError(
129+
"ProtocolNotSupportedError",
130+
"not hosted",
131+
"",
132+
error_code="UNIMPLEMENTED",
133+
error_kind="protocol_not_supported",
134+
)
135+
_patch_listing(monkeypatch, error)
136+
with Client(SUBPROCESS_FIXTURE_WORKER, pool=None) as client:
137+
assert client.list_protocols() == [HostedProtocol(name="vgi.v2", version="", hash="")]
138+
139+
def test_other_errors_raise(self, monkeypatch: pytest.MonkeyPatch) -> None:
140+
"""Any other reflection failure is a ``ClientError``, not a silent fallback."""
141+
_patch_listing(monkeypatch, RpcError("RuntimeError", "boom", "", error_code="INTERNAL"))
142+
with Client(SUBPROCESS_FIXTURE_WORKER, pool=None) as client, pytest.raises(ClientError, match="boom"):
143+
client.list_protocols()

‎uv.lock‎

Lines changed: 4 additions & 4 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎vgi/client/__init__.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@
4040

4141
from typing import TYPE_CHECKING, Any
4242

43+
from vgi_rpc import HostedProtocol
44+
4345
from vgi.client.aggregate import (
4446
AggregateClientMixin,
4547
AggregateSession,
@@ -60,6 +62,7 @@
6062
"CatalogSnapshot",
6163
"Client",
6264
"ClientError",
65+
"HostedProtocol",
6366
"OutputWriter",
6467
"ResumableTableScan",
6568
"ResumeUnsupported",

‎vgi/client/client.py‎

Lines changed: 78 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,8 @@
6666
client.copy_from(...) — read a custom ``COPY ... FROM`` format
6767
client.copy_to(...) — write a custom ``COPY ... TO`` format
6868
client.server_capabilities() — HTTP only; upload-URL caps
69+
client.list_protocols() — protocols the worker hosts (reflection)
70+
client.describe_protocol(...) — one hosted protocol's methods (reflection)
6971
7072
See Also:
7173
--------
@@ -94,7 +96,14 @@
9496
from typing import IO, Any, Literal, cast
9597

9698
import pyarrow as pa
97-
from vgi_rpc import WorkerPool
99+
from vgi_rpc import (
100+
HostedProtocol,
101+
ReflectionNotSupportedError,
102+
ServiceDescription,
103+
WorkerPool,
104+
)
105+
from vgi_rpc import describe_protocol as rpc_describe_protocol
106+
from vgi_rpc import list_protocols as rpc_list_protocols
98107
from vgi_rpc.log import Message
99108
from vgi_rpc.rpc import (
100109
AnnotatedBatch,
@@ -226,6 +235,10 @@ def _validate_accepted_max_response_bytes(value: int | None) -> int | None:
226235
return value
227236

228237

238+
#: Wire name of the primary VGI protocol, the first protocol every worker hosts.
239+
VGI_PROTOCOL_NAME = "vgi.v2"
240+
241+
229242
@dataclass
230243
class WorkerConnection:
231244
"""Holds state for a single worker connection (subprocess, HTTP, TCP, or launch).
@@ -1599,6 +1612,70 @@ def server_capabilities(self) -> Any:
15991612
return http_capabilities(base_url=self._base_url, prefix=self._http_prefix, client=httpx_client)
16001613
return http_capabilities(base_url=self._base_url, client=httpx_client)
16011614

1615+
def _require_primary(self) -> WorkerConnection:
1616+
"""Return the primary connection, or raise if the client is not started."""
1617+
if self._primary is None:
1618+
raise ClientError("Client not started. Call start() or use context manager.")
1619+
return self._primary
1620+
1621+
def list_protocols(self) -> list[HostedProtocol]:
1622+
"""Return the protocols the worker hosts, in the worker's order.
1623+
1624+
Calls ``vgi_rpc.Reflection.v1.list_protocols`` through
1625+
``vgi_rpc.list_protocols`` on the primary worker's connection, whatever
1626+
the transport, without opening a new one. ``vgi.v2`` is always first; a
1627+
worker's extra protocols (``Worker.hosted_protocols()``) follow, then
1628+
the framework's own. Use it to discover an optional protocol before
1629+
calling it, rather than calling it and reading an error.
1630+
1631+
Do not call it while a stream is open on this client: over pipe-like
1632+
transports it shares the stream's connection.
1633+
1634+
A worker that does not host reflection (vgi-rpc raises
1635+
``ReflectionNotSupportedError``) is not an error: the result is a single
1636+
``HostedProtocol("vgi.v2", "", "")``, since ``vgi.v2`` is the one
1637+
protocol every worker hosts. The empty hash marks the entry as inferred.
1638+
1639+
Returns:
1640+
One [`HostedProtocol`][] per hosted protocol.
1641+
1642+
Raises:
1643+
[`ClientError`][]: If the client is not started, or the worker
1644+
answers reflection with any other error.
1645+
1646+
"""
1647+
primary = self._require_primary()
1648+
try:
1649+
return rpc_list_protocols(primary.proxy)
1650+
except ReflectionNotSupportedError:
1651+
return [HostedProtocol(name=VGI_PROTOCOL_NAME, version="", hash="")]
1652+
except RpcError as e:
1653+
raise ClientError(f"list_protocols failed: {e}") from e
1654+
1655+
def describe_protocol(self, name: str) -> ServiceDescription:
1656+
"""Return one hosted protocol's full description, via reflection.
1657+
1658+
Calls ``vgi_rpc.describe_protocol`` on the primary connection (two
1659+
round trips: ``list_protocols``, then ``describe(name)``).
1660+
1661+
Args:
1662+
name: The protocol's wire name, as ``list_protocols`` reports it.
1663+
1664+
Returns:
1665+
A ``vgi_rpc.ServiceDescription`` with each method's type and
1666+
parameter/result schemas.
1667+
1668+
Raises:
1669+
[`ClientError`][]: If the client is not started, the worker does
1670+
not host *name*, or the worker does not answer reflection.
1671+
1672+
"""
1673+
primary = self._require_primary()
1674+
try:
1675+
return rpc_describe_protocol(primary.proxy, name)
1676+
except RpcError as e:
1677+
raise ClientError(f"describe_protocol({name!r}) failed: {e}") from e
1678+
16021679
def __enter__(self) -> Client:
16031680
"""Enter the context manager by starting the worker subprocess."""
16041681
self.start()

0 commit comments

Comments
 (0)