feat: add TCP transport support (vgi-rpc 0.21.0)

Wire up the raw Arrow-IPC TCP transport added in vgi-rpc 0.21.0:

- Worker.main() gains --tcp [HOST:]PORT, mutually exclusive with
  --http/--unix, emitting the TCP:<host>:<port> discovery line and
  serving via serve_tcp(). The meta/fixture worker (run_server) picks
  up --tcp directly from vgi-rpc.
- Client gains transport="tcp" + Client.from_tcp(host, port), spawning
  connections via tcp_connect(); catalog_mixin routes catalog calls
  over TCP too.
- New round-trip tests spawn vgi-fixture-worker --tcp, parse the
  discovery line, and drive table-function + catalog calls through
  Client.from_tcp.

TCP framing carries no auth/encryption (loopback/trusted networks
only); HTTP remains the transport for untrusted networks.

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

Author: rustyconover

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
     "pyarrow",
     "typer>=0.9",
     "platformdirs",
-    "vgi-rpc>=0.20.5",
+    "vgi-rpc>=0.21.0",
     "httpx>=0.24",
 ]
 

tests/test_tcp_transport.py

@@ -0,0 +1,122 @@
+# Copyright 2025, 2026 Query Farm LLC - https://query.farm
+
+"""Round-trip tests for the TCP transport.
+
+Spawns ``vgi-fixture-worker --tcp 127.0.0.1:0`` (raw Arrow-IPC framing over a
+TCP socket, served by ``vgi_rpc.rpc.run_server``), parses the ``TCP:host:port``
+discovery line it prints on stdout, then drives it through
+``Client.from_tcp(...)`` — the TCP analog of the HTTP round-trip in
+``tests/_http_fixtures.py``.
+"""
+
+from __future__ import annotations
+
+import queue
+import subprocess
+import sys
+import threading
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+import pyarrow as pa
+import pytest
+
+from vgi.arguments import Arguments
+from vgi.client import Client
+
+
+@contextmanager
+def run_tcp_worker(*, bind: str = "127.0.0.1:0") -> Iterator[tuple[str, int]]:
+    """Run ``vgi-fixture-worker --tcp`` and yield the bound ``(host, port)``.
+
+    The worker prints ``TCP:<host>:<port>`` once bound and then must not write
+    further to stdout (the cross-language launcher discovery contract), so we
+    read exactly one line to learn the port. stderr is drained in the
+    background to keep the worker from blocking on a full pipe buffer.
+    """
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "vgi._test_fixtures.worker", "--tcp", bind],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        bufsize=1,
+    )
+
+    def _drain(pipe: object) -> None:
+        for _ in pipe:  # type: ignore[attr-defined]
+            pass
+
+    stderr_thread = threading.Thread(target=_drain, args=(proc.stderr,), daemon=True)
+    stderr_thread.start()
+
+    # Read the discovery line off stdout with a timeout so a worker that never
+    # binds fails the test instead of hanging it.
+    line_q: queue.Queue[str] = queue.Queue(maxsize=1)
+
+    def _read_line() -> None:
+        assert proc.stdout is not None
+        for line in proc.stdout:
+            if line.startswith("TCP:"):
+                line_q.put(line.strip())
+                return
+
+    reader = threading.Thread(target=_read_line, daemon=True)
+    reader.start()
+    try:
+        discovery = line_q.get(timeout=30)
+    except queue.Empty:
+        proc.terminate()
+        raise TimeoutError("worker did not emit a TCP: discovery line within 30s") from None
+
+    _, host, port_str = discovery.split(":", 2)
+    try:
+        yield host, int(port_str)
+    finally:
+        proc.terminate()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            proc.kill()
+            proc.wait(timeout=5)
+        stderr_thread.join(timeout=5)
+
+
+def test_tcp_round_trip_table_function() -> None:
+    """A table function streams rows correctly over the TCP transport."""
+    with run_tcp_worker() as (host, port), Client.from_tcp(host, port) as client:
+        batches = list(
+            client.table_function(
+                function_name="sequence",
+                arguments=Arguments(positional=(pa.scalar(5),)),
+            )
+        )
+
+    table = pa.Table.from_batches(batches)
+    assert table.column("n").to_pylist() == [0, 1, 2, 3, 4]
+
+
+def test_tcp_round_trip_catalog_listing() -> None:
+    """Catalog discovery works over the TCP transport (catalog_mixin path)."""
+    with run_tcp_worker() as (host, port), Client.from_tcp(host, port) as client:
+        catalogs = client.catalogs()
+
+    assert any(c.name == "example" for c in catalogs)
+
+
+class TestTcpConstructorValidation:
+    """``transport='tcp'`` argument validation."""
+
+    def test_requires_host_and_port(self) -> None:
+        """Tcp transport without host/port is rejected."""
+        with pytest.raises(ValueError, match="requires tcp_host and tcp_port"):
+            Client(transport="tcp", pool=None)
+
+    def test_rejects_server_path(self) -> None:
+        """server_path is subprocess-only."""
+        with pytest.raises(ValueError, match="server_path is only meaningful"):
+            Client("some-worker", transport="tcp", tcp_host="127.0.0.1", tcp_port=1, pool=None)
+
+    def test_rejects_base_url(self) -> None:
+        """base_url is http-only."""
+        with pytest.raises(ValueError, match="base_url is only meaningful"):
+            Client(transport="tcp", tcp_host="127.0.0.1", tcp_port=1, base_url="http://x", pool=None)

uv.lock

@@ -90,21 +90,25 @@ class CatalogClientMixin:
     Catalog methods spawn ephemeral connections under the hood — for
     subprocess transport a pooled subprocess worker; for HTTP transport a
     short-lived ``http_connect`` session reusing the ``Client``'s shared
-    ``httpx.Client`` (bearer token, headers). Browsing catalogs over HTTP
-    is the canonical non-DuckDB use case this mixin supports.
+    ``httpx.Client`` (bearer token, headers); for TCP transport a short-lived
+    ``tcp_connect`` session. Browsing catalogs over HTTP is the canonical
+    non-DuckDB use case this mixin supports.
 
-    Other attributes expected from ``Client``: ``_transport`` (subprocess vs
-    http), ``_base_url`` (HTTP base URL), and ``_get_or_create_httpx_client()``
-    (shared HTTP client factory).
+    Other attributes expected from ``Client``: ``_transport`` (subprocess,
+    http, or tcp), ``_base_url`` (HTTP base URL), ``_tcp_host`` / ``_tcp_port``
+    (TCP endpoint), and ``_get_or_create_httpx_client()`` (shared HTTP client
+    factory).
 
     Attributes:
         server_path: Worker shell command used for subprocess transport.
     """
 
     # Type hints for attributes expected from Client
     server_path: str
-    _transport: Literal["subprocess", "http"]
+    _transport: Literal["subprocess", "http", "tcp"]
     _base_url: str | None
+    _tcp_host: str | None
+    _tcp_port: int | None
     _external_location: Any | None
 
     def _get_or_create_httpx_client(self) -> Any:  # implemented by Client
@@ -128,7 +132,8 @@ def _catalog_connect(self) -> Iterator[VgiProtocol]:
             A typed `[`VgiProtocol`][]` proxy bound to the active transport.
         """
         try:
-            if getattr(self, "_transport", "subprocess") == "http":
+            transport = getattr(self, "_transport", "subprocess")
+            if transport == "http":
                 from vgi_rpc.http import http_connect
 
                 httpx_client = self._get_or_create_httpx_client()
@@ -139,6 +144,17 @@ def _catalog_connect(self) -> Iterator[VgiProtocol]:
                     external_location=getattr(self, "_external_location", None),
                 ) as proxy:
                     yield proxy
+            elif transport == "tcp":
+                from vgi_rpc.rpc import tcp_connect
+
+                assert self._tcp_host is not None and self._tcp_port is not None
+                with tcp_connect(
+                    VgiProtocol,  # type: ignore[type-abstract]
+                    self._tcp_host,
+                    self._tcp_port,
+                    external_location=getattr(self, "_external_location", None),
+                ) as proxy:
+                    yield proxy
             else:
                 cmd = shlex.split(self.server_path, posix=sys.platform != "win32")
                 with _catalog_pool.connect(VgiProtocol, cmd) as proxy:  # type: ignore[type-abstract]