From d58780e8785f2f1386da04f02a7130d1e2c65f95 Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Fri, 10 Jul 2026 11:55:17 +0300 Subject: [PATCH 1/6] feat: PoC bare-rpc-python as the production wire transport - BareRpcWorker wraps bare_rpc.RPC directly against the spawned worker's Unix socket, replacing poc_heartbeat.QvacWorker's hand-rolled framing for the unary and server-stream call shapes - its call/call_stream already satisfy qvac._transport.Transport's async shape, so it plugs straight into the real generated qvac.methods stubs with no PocTransport-style adapter - bare-rpc-python/compact-encoding-python aren't on PyPI yet -- pinned as an optional "bare-rpc" extra via git dependencies --- packages/sdk-python/pyproject.toml | 10 + .../tests/poc_bare_rpc_transport.py | 177 ++++++++++++++++++ .../sdk-python/tests/test_poc_bare_rpc.py | 76 ++++++++ 3 files changed, 263 insertions(+) create mode 100644 packages/sdk-python/tests/poc_bare_rpc_transport.py create mode 100644 packages/sdk-python/tests/test_poc_bare_rpc.py diff --git a/packages/sdk-python/pyproject.toml b/packages/sdk-python/pyproject.toml index fb54de7c00..311a7cfaed 100644 --- a/packages/sdk-python/pyproject.toml +++ b/packages/sdk-python/pyproject.toml @@ -14,6 +14,16 @@ dependencies = ["pydantic>=2.6,<3"] [project.optional-dependencies] gen = ["datamodel-code-generator==0.68.0", "black==26.5.1", "ruff==0.15.21"] dev = ["pytest==9.1.1", "pytest-asyncio==1.4.0", "ruff==0.15.21", "mypy==2.3.0"] +# QVAC-21806 PoC only -- neither package is on PyPI yet (both v0.0.1, git-only). +# Pinned to their current HEAD commits for reproducibility while unreleased. +bare-rpc = [ + "bare-rpc-python @ git+https://github.com/holepunchto/bare-rpc-python.git@5219c528866678553cb9716f80a199c9accfe8c1", + "compact-encoding-python @ git+https://github.com/holepunchto/compact-encoding-python.git@e7e1f080eb413c6ea06f8c6400a9716857c67266", +] + +[tool.hatch.metadata] +# QVAC-21806 PoC: bare-rpc-python/compact-encoding-python aren't on PyPI yet. +allow-direct-references = true [tool.hatch.build.targets.wheel] packages = ["src/qvac"] diff --git a/packages/sdk-python/tests/poc_bare_rpc_transport.py b/packages/sdk-python/tests/poc_bare_rpc_transport.py new file mode 100644 index 0000000000..e6b1c608af --- /dev/null +++ b/packages/sdk-python/tests/poc_bare_rpc_transport.py @@ -0,0 +1,177 @@ +"""QVAC-21806 PoC: wire bare_rpc.RPC (github.com/holepunchto/bare-rpc-python) +to a real spawned SDK worker, replacing poc_heartbeat.py's hand-rolled frame +encode/decode with the real library. + +Kept deliberately separate from poc_heartbeat.py/poc_transport.py (used by +PR #3100) rather than replacing them -- this is exploratory, async (bare_rpc +is asyncio-native), and not yet wired into qvac._transport.Transport's sync +protocol. See the PR/commit notes for what's proven here vs. still open. + +Wire-level notes: +- Our worker ignores bare-rpc's `command` field entirely and routes purely + on the JSON payload's own `type` field (see poc_heartbeat.py's comment: + "command == id: the server ignores `command` and routes on payload.type"). + So every call below passes command=0 -- bare_rpc.RPC's own internal + monotonic id (not `command`) is what correlates request/response/stream + frames on the wire, and that's handled entirely inside the library. +- Server-stream replies arrive as newline-delimited JSON, but not + necessarily one JSON document per wire STREAM-DATA frame (the worker may + coalesce or split them) -- so chunks from `IncomingStream` still need the + same buffer-and-split-on-newline handling poc_heartbeat.py's call_stream + does, just fed from bare_rpc's discrete async-iterated chunks instead of + raw socket bytes. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +import tempfile +from typing import Any, AsyncIterator + +import bare_rpc + +SDK = os.environ.get( + "QVAC_POC_SDK_DIR", + "/Users/lauri/noxtton/qvac-new/packages/sdk", +) +BARE = f"{SDK}/node_modules/bare-runtime-darwin-arm64/bin/bare" +WORKER = f"{SDK}/dist/server/worker.js" + + +def _json_or_raise(data: bytes) -> Any: + """Parse a JSON payload; the SDK reports failures in-band as {"type":"error"}.""" + obj = json.loads(data.decode("utf-8")) + if isinstance(obj, dict) and obj.get("type") == "error": + raise RuntimeError("worker: " + str(obj.get("message", "unknown error"))) + return obj + + +class BareRpcWorker: + """Spawns the real Bare worker and speaks to it via bare_rpc.RPC instead + of hand-rolled framing. asyncio-native, unlike poc_heartbeat.QvacWorker.""" + + def __init__(self) -> None: + self._sock_path = os.path.join( + tempfile.gettempdir(), f"qvac-bare-rpc-poc-{os.getpid()}.sock" + ) + self._log_path = os.path.join( + tempfile.gettempdir(), f"qvac-bare-rpc-poc-worker-{os.getpid()}.log" + ) + self._server: asyncio.AbstractServer | None = None + self._proc: asyncio.subprocess.Process | None = None + self._log_fh = None + self._writer: asyncio.StreamWriter | None = None + self._read_task: asyncio.Task | None = None + self.rpc: bare_rpc.RPC | None = None + + async def start(self) -> "BareRpcWorker": + if os.path.exists(self._sock_path): + os.unlink(self._sock_path) + + connected: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + async def on_client( + reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + self._writer = writer + self._read_task = asyncio.current_task() + if not connected.done(): + connected.set_result(None) + try: + while True: + chunk = await reader.read(65536) + if not chunk: + break + await self.rpc.receive(chunk) + except asyncio.CancelledError: + pass + + self._server = await asyncio.start_unix_server(on_client, path=self._sock_path) + + config = json.dumps( + { + "QVAC_IPC_SOCKET_PATH": self._sock_path, + "HOME_DIR": os.path.expanduser("~"), + } + ) + self._log_fh = open(self._log_path, "wb") + self._proc = await asyncio.create_subprocess_exec( + BARE, + WORKER, + config, + cwd=SDK, + stdout=self._log_fh, + stderr=subprocess.STDOUT, + ) + + def send(frame: bytes) -> None: + self._writer.write(frame) + + self.rpc = bare_rpc.RPC(send=send) + await asyncio.wait_for(connected, timeout=30) + return self + + async def close(self) -> None: + if self._read_task: + self._read_task.cancel() + if self.rpc: + self.rpc.close() + if self._proc: + self._proc.terminate() + try: + await asyncio.wait_for(self._proc.wait(), timeout=5) + except asyncio.TimeoutError: + self._proc.kill() + if self._writer: + self._writer.close() + if self._server: + self._server.close() + try: + await asyncio.wait_for(self._server.wait_closed(), timeout=5) + except asyncio.TimeoutError: + pass + if self._log_fh: + self._log_fh.close() + if os.path.exists(self._sock_path): + os.unlink(self._sock_path) + + async def __aenter__(self) -> "BareRpcWorker": + return await self.start() + + async def __aexit__(self, *exc: object) -> None: + await self.close() + + def worker_logs(self) -> str: + try: + with open(self._log_path, "r", errors="replace") as f: + return f.read() + except OSError: + return "" + + # ---- the two call shapes exercised so far -------------------------- + + async def call(self, payload: dict) -> dict: + """Unary, via bare_rpc.RPC.request -- no hand-rolled framing at all.""" + data = await self.rpc.request( + command=0, data=json.dumps(payload).encode("utf-8") + ) + return _json_or_raise(data) + + async def call_stream(self, payload: dict) -> AsyncIterator[dict]: + """Server-stream, via bare_rpc.RPC.request_with_response_stream.""" + stream = await self.rpc.request_with_response_stream( + command=0, data=json.dumps(payload).encode("utf-8") + ) + buffer = "" + async for chunk in stream: + buffer += chunk.decode("utf-8") + lines = buffer.split("\n") + buffer = lines.pop() + for line in lines: + if line.strip(): + yield _json_or_raise(line.encode("utf-8")) + if buffer.strip(): + yield _json_or_raise(buffer.encode("utf-8")) diff --git a/packages/sdk-python/tests/test_poc_bare_rpc.py b/packages/sdk-python/tests/test_poc_bare_rpc.py new file mode 100644 index 0000000000..b365ea00a7 --- /dev/null +++ b/packages/sdk-python/tests/test_poc_bare_rpc.py @@ -0,0 +1,76 @@ +"""QVAC-21806 PoC: same real-worker rigor as test_poc_smoke.py/test_poc_progress.py, +but the wire encode/decode is bare_rpc.RPC instead of poc_heartbeat.QvacWorker's +hand-rolled framing. Validates responses against the same generated pydantic +models either way -- proving bare_rpc is a drop-in replacement at the framing +layer without changing anything above it. +""" + +from __future__ import annotations + +import os + +import pytest +import pytest_asyncio + +from qvac.models import QWEN3_600M_INST_Q4 +from qvac.schemas import ( + CompletionStreamRequest, + HeartbeatRequest, + LoadModelRequest, +) +from qvac.methods import completion_stream, heartbeat, load_model + +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif( + "QVAC_POC_SDK_DIR" not in os.environ, + reason="set QVAC_POC_SDK_DIR to a built SDK checkout to run the bare_rpc PoC", + ), +] + + +@pytest_asyncio.fixture +async def worker(): + from poc_bare_rpc_transport import BareRpcWorker + + async with BareRpcWorker() as w: + yield w + + +async def test_heartbeat_unary_via_bare_rpc(worker) -> None: + """BareRpcWorker.call/.call_stream already match qvac._transport.Transport's + async shape exactly, so it plugs straight into the real generated stubs -- + no PocTransport-style adapter needed, unlike poc_heartbeat.QvacWorker.""" + response = await heartbeat(worker, HeartbeatRequest(type="heartbeat")) + assert response.type == "heartbeat" + assert isinstance(response.number, float) + + +async def test_load_model_and_completion_stream_via_bare_rpc(worker) -> None: + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": QWEN3_600M_INST_Q4.src, + "modelType": "llamacpp-completion", + "modelConfig": {}, + } + ) + load_response = await load_model(worker, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + completion_request = CompletionStreamRequest.model_validate( + { + "type": "completionStream", + "modelId": model_id, + "history": [{"role": "user", "content": "Say hello in five words."}], + "stream": True, + } + ) + + text = "" + async for chunk in completion_stream(worker, completion_request): + for event in chunk.events: + if event.type == "contentDelta": + text += event.text + assert text.strip(), "expected real completion text via the bare_rpc server-stream" From f3f8045fa64e0739a97862e121309f8bfcf5c794 Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Fri, 10 Jul 2026 18:52:49 +0300 Subject: [PATCH 2/6] fix: default bare_rpc PoC SDK path to the monorepo-relative dir Same fix as the QVAC-21805-python-models commit, applied to the 21806-only bare_rpc PoC files: poc_bare_rpc_transport.py's SDK default was a hardcoded developer-specific absolute path; default to ../../sdk relative to this file instead, and gate test skips on the worker actually existing rather than on QVAC_POC_SDK_DIR being set. --- packages/sdk-python/tests/poc_bare_rpc_transport.py | 3 ++- packages/sdk-python/tests/test_poc_bare_rpc.py | 5 +++-- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/packages/sdk-python/tests/poc_bare_rpc_transport.py b/packages/sdk-python/tests/poc_bare_rpc_transport.py index e6b1c608af..69a2da4425 100644 --- a/packages/sdk-python/tests/poc_bare_rpc_transport.py +++ b/packages/sdk-python/tests/poc_bare_rpc_transport.py @@ -29,13 +29,14 @@ import os import subprocess import tempfile +from pathlib import Path from typing import Any, AsyncIterator import bare_rpc SDK = os.environ.get( "QVAC_POC_SDK_DIR", - "/Users/lauri/noxtton/qvac-new/packages/sdk", + str(Path(__file__).resolve().parent.parent.parent / "sdk"), ) BARE = f"{SDK}/node_modules/bare-runtime-darwin-arm64/bin/bare" WORKER = f"{SDK}/dist/server/worker.js" diff --git a/packages/sdk-python/tests/test_poc_bare_rpc.py b/packages/sdk-python/tests/test_poc_bare_rpc.py index b365ea00a7..6c6b7f8a84 100644 --- a/packages/sdk-python/tests/test_poc_bare_rpc.py +++ b/packages/sdk-python/tests/test_poc_bare_rpc.py @@ -12,6 +12,7 @@ import pytest import pytest_asyncio +from poc_bare_rpc_transport import WORKER from qvac.models import QWEN3_600M_INST_Q4 from qvac.schemas import ( CompletionStreamRequest, @@ -23,8 +24,8 @@ pytestmark = [ pytest.mark.asyncio, pytest.mark.skipif( - "QVAC_POC_SDK_DIR" not in os.environ, - reason="set QVAC_POC_SDK_DIR to a built SDK checkout to run the bare_rpc PoC", + not os.path.exists(WORKER), + reason=f"no built SDK worker found at {WORKER!r} -- run `bun run build` in packages/sdk, or set QVAC_POC_SDK_DIR", ), ] From 0d898f2b77f239516fd242b40828339e12784607 Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Tue, 14 Jul 2026 01:04:46 +0300 Subject: [PATCH 3/6] QVAC-21806 feat: implement call_duplex against bare-rpc-python's stream runtime - BareRpcWorker.call_duplex maps onto RPC.create_bidirectional_stream: write the JSON payload as the first outgoing chunk followed by up's chunks, end() the outgoing side, and read incoming with the same buffer-and-split-on-newline handling as call_stream. The upstream pump runs as a concurrent background task since the worker can start responding before the client finishes sending. - BareRpcWorker now implements all three Transport call shapes (call/call_stream/ call_duplex), so it plugs directly into the generated duplex stubs (transcribe_stream, text_to_speech_stream, bci_transcribe_stream) with no PocTransport-style adapter. - New test: load a real TTS model and drive text_to_speech_stream end to end against a spawned SDK worker, asserting real synthesized audio and a terminal done=True event. --- .../tests/poc_bare_rpc_transport.py | 54 +++++++++++++++++-- .../sdk-python/tests/test_poc_bare_rpc.py | 42 ++++++++++++++- 2 files changed, 89 insertions(+), 7 deletions(-) diff --git a/packages/sdk-python/tests/poc_bare_rpc_transport.py b/packages/sdk-python/tests/poc_bare_rpc_transport.py index 69a2da4425..b592ba1454 100644 --- a/packages/sdk-python/tests/poc_bare_rpc_transport.py +++ b/packages/sdk-python/tests/poc_bare_rpc_transport.py @@ -3,9 +3,10 @@ encode/decode with the real library. Kept deliberately separate from poc_heartbeat.py/poc_transport.py (used by -PR #3100) rather than replacing them -- this is exploratory, async (bare_rpc -is asyncio-native), and not yet wired into qvac._transport.Transport's sync -protocol. See the PR/commit notes for what's proven here vs. still open. +PR #3100) rather than replacing them -- this is exploratory, and bare_rpc is +asyncio-native so `BareRpcWorker` matches qvac._transport.Transport's async +shape directly (call/call_stream/call_duplex), no PocTransport-style adapter +needed. See the PR/commit notes for what's proven here vs. still open. Wire-level notes: - Our worker ignores bare-rpc's `command` field entirely and routes purely @@ -20,6 +21,15 @@ same buffer-and-split-on-newline handling poc_heartbeat.py's call_stream does, just fed from bare_rpc's discrete async-iterated chunks instead of raw socket bytes. +- Duplex maps onto `RPC.create_bidirectional_stream`: the returned + `(outgoing, incoming)` pair already IS the request/response stream pair + poc_heartbeat.py's hand-rolled `_duplex_call` builds frame-by-frame -- the + library owns the OPEN/RESUME/PAUSE control handshake, we only `write()` + the JSON payload as the first outgoing chunk followed by `up`'s chunks, + `end()` the outgoing side, and read `incoming` with the same + buffer-and-split-on-newline handling as call_stream. The two sides run + concurrently (a background task pumps `up` while `incoming` is iterated) + since the worker can start responding before the client finishes sending. """ from __future__ import annotations @@ -30,7 +40,7 @@ import subprocess import tempfile from pathlib import Path -from typing import Any, AsyncIterator +from typing import Any, AsyncIterable, AsyncIterator import bare_rpc @@ -152,7 +162,7 @@ def worker_logs(self) -> str: except OSError: return "" - # ---- the two call shapes exercised so far -------------------------- + # ---- the three call shapes ----------------------------------------- async def call(self, payload: dict) -> dict: """Unary, via bare_rpc.RPC.request -- no hand-rolled framing at all.""" @@ -176,3 +186,37 @@ async def call_stream(self, payload: dict) -> AsyncIterator[dict]: yield _json_or_raise(line.encode("utf-8")) if buffer.strip(): yield _json_or_raise(buffer.encode("utf-8")) + + async def call_duplex( + self, payload: dict, up: AsyncIterable[bytes] + ) -> AsyncIterator[dict]: + """Duplex, via bare_rpc.RPC.create_bidirectional_stream -- first outgoing + chunk is the JSON payload, then `up`'s chunks; yields parsed response + chunks with the same buffer-and-split-on-newline handling as call_stream.""" + outgoing, incoming = await self.rpc.create_bidirectional_stream(command=0) + await outgoing.write(json.dumps(payload).encode("utf-8")) + + async def _pump_up() -> None: + async for chunk in up: + await outgoing.write(chunk) + await outgoing.end() + + pump_task = asyncio.ensure_future(_pump_up()) + try: + buffer = "" + async for chunk in incoming: + buffer += chunk.decode("utf-8") + lines = buffer.split("\n") + buffer = lines.pop() + for line in lines: + if line.strip(): + yield _json_or_raise(line.encode("utf-8")) + if buffer.strip(): + yield _json_or_raise(buffer.encode("utf-8")) + finally: + if not pump_task.done(): + pump_task.cancel() + try: + await pump_task + except asyncio.CancelledError: + pass diff --git a/packages/sdk-python/tests/test_poc_bare_rpc.py b/packages/sdk-python/tests/test_poc_bare_rpc.py index 6c6b7f8a84..ee20a19c71 100644 --- a/packages/sdk-python/tests/test_poc_bare_rpc.py +++ b/packages/sdk-python/tests/test_poc_bare_rpc.py @@ -13,13 +13,15 @@ import pytest_asyncio from poc_bare_rpc_transport import WORKER -from qvac.models import QWEN3_600M_INST_Q4 +from qvac.models import QWEN3_600M_INST_Q4, TTS_EN_SUPERTONIC_Q4_0 from qvac.schemas import ( CompletionStreamRequest, HeartbeatRequest, LoadModelRequest, + ModelType, + TextToSpeechStreamRequest, ) -from qvac.methods import completion_stream, heartbeat, load_model +from qvac.methods import completion_stream, heartbeat, load_model, text_to_speech_stream pytestmark = [ pytest.mark.asyncio, @@ -75,3 +77,39 @@ async def test_load_model_and_completion_stream_via_bare_rpc(worker) -> None: if event.type == "contentDelta": text += event.text assert text.strip(), "expected real completion text via the bare_rpc server-stream" + + +async def _as_async_iter(items): + for item in items: + yield item + + +async def test_load_model_and_tts_stream_duplex_via_bare_rpc(worker) -> None: + """Exercises call_duplex end to end: text goes up the request stream while + synthesized audio comes down the response stream, concurrently.""" + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": TTS_EN_SUPERTONIC_Q4_0.src, + "modelType": ModelType.tts_ggml, + "modelConfig": {"ttsEngine": "supertonic", "language": "en"}, + } + ) + load_response = await load_model(worker, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + tts_request = TextToSpeechStreamRequest.model_validate( + {"type": "textToSpeechStream", "modelId": model_id} + ) + text = b"Hello from QVAC. This is streaming text to speech." + + samples = [] + saw_done = False + async for chunk in text_to_speech_stream( + worker, tts_request, _as_async_iter([text]) + ): + samples.extend(chunk.buffer) + saw_done = saw_done or chunk.done + assert samples, "expected real synthesized audio via the bare_rpc duplex stream" + assert saw_done, "expected a terminal done=True event on the response stream" From f7e2deb8ba33ea97c671a92e8d249c5e4af53efa Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Tue, 14 Jul 2026 01:12:47 +0300 Subject: [PATCH 4/6] fix: guard bare_rpc import in the PoC transport so CI skips instead of erroring - test_poc_bare_rpc.py's skipif only covered the missing-built-worker case; the unconditional `import bare_rpc` at module level still hard-failed collection in CI, where the pr-checks-sdk-python.yml install step (`.[gen,dev]`) doesn't pull in the unpublished `bare-rpc` extra. - Wrap the import, expose BARE_RPC_AVAILABLE, and add it as a second skip condition alongside the existing worker-not-built check. --- packages/sdk-python/tests/poc_bare_rpc_transport.py | 7 ++++++- packages/sdk-python/tests/test_poc_bare_rpc.py | 7 ++++++- 2 files changed, 12 insertions(+), 2 deletions(-) diff --git a/packages/sdk-python/tests/poc_bare_rpc_transport.py b/packages/sdk-python/tests/poc_bare_rpc_transport.py index b592ba1454..d59533ee52 100644 --- a/packages/sdk-python/tests/poc_bare_rpc_transport.py +++ b/packages/sdk-python/tests/poc_bare_rpc_transport.py @@ -42,7 +42,12 @@ from pathlib import Path from typing import Any, AsyncIterable, AsyncIterator -import bare_rpc +try: + import bare_rpc +except ImportError: + bare_rpc = None + +BARE_RPC_AVAILABLE = bare_rpc is not None SDK = os.environ.get( "QVAC_POC_SDK_DIR", diff --git a/packages/sdk-python/tests/test_poc_bare_rpc.py b/packages/sdk-python/tests/test_poc_bare_rpc.py index ee20a19c71..648ff738f9 100644 --- a/packages/sdk-python/tests/test_poc_bare_rpc.py +++ b/packages/sdk-python/tests/test_poc_bare_rpc.py @@ -12,7 +12,7 @@ import pytest import pytest_asyncio -from poc_bare_rpc_transport import WORKER +from poc_bare_rpc_transport import BARE_RPC_AVAILABLE, WORKER from qvac.models import QWEN3_600M_INST_Q4, TTS_EN_SUPERTONIC_Q4_0 from qvac.schemas import ( CompletionStreamRequest, @@ -25,6 +25,11 @@ pytestmark = [ pytest.mark.asyncio, + pytest.mark.skipif( + not BARE_RPC_AVAILABLE, + reason="bare_rpc not installed -- install the 'bare-rpc' extra " + "(`pip install -e '.[bare-rpc]'`) to run these PoC tests", + ), pytest.mark.skipif( not os.path.exists(WORKER), reason=f"no built SDK worker found at {WORKER!r} -- run `bun run build` in packages/sdk, or set QVAC_POC_SDK_DIR", From cbdf1d53d0813cffae0e1f2148f6ac7818ba7326 Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Tue, 14 Jul 2026 01:37:54 +0300 Subject: [PATCH 5/6] feat: promote bare_rpc PoC to production Transport, extend duplex coverage to transcribe_stream and bci_transcribe_stream - New src/qvac/bare_rpc_transport.py: BareRpcTransport implements qvac._transport.Transport over bare_rpc.RPC, mirroring the JS SDK's client/rpc/node-rpc-client.ts (socket handshake + worker spawn), with the caller supplying the Bare command instead of this package resolving/bundling worker artifacts itself. - Retire tests/poc_bare_rpc_transport.py; test_poc_bare_rpc.py becomes test_bare_rpc_transport.py, testing the production module directly. - Add duplex tests for the two remaining duplex-shaped methods beyond text_to_speech_stream: transcribe_stream (parakeet, real-time-paced 16kHz mono s16le PCM) and bci_transcribe_stream (BCI addon, raw neural feature chunks against a real fixture). - Full sdk-python suite (39 tests) passes against a real spawned worker. --- .../qvac/bare_rpc_transport.py} | 124 ++++----- .../tests/test_bare_rpc_transport.py | 263 ++++++++++++++++++ .../sdk-python/tests/test_poc_bare_rpc.py | 120 -------- 3 files changed, 310 insertions(+), 197 deletions(-) rename packages/sdk-python/{tests/poc_bare_rpc_transport.py => src/qvac/bare_rpc_transport.py} (57%) create mode 100644 packages/sdk-python/tests/test_bare_rpc_transport.py delete mode 100644 packages/sdk-python/tests/test_poc_bare_rpc.py diff --git a/packages/sdk-python/tests/poc_bare_rpc_transport.py b/packages/sdk-python/src/qvac/bare_rpc_transport.py similarity index 57% rename from packages/sdk-python/tests/poc_bare_rpc_transport.py rename to packages/sdk-python/src/qvac/bare_rpc_transport.py index d59533ee52..f60240c3b1 100644 --- a/packages/sdk-python/tests/poc_bare_rpc_transport.py +++ b/packages/sdk-python/src/qvac/bare_rpc_transport.py @@ -1,35 +1,15 @@ -"""QVAC-21806 PoC: wire bare_rpc.RPC (github.com/holepunchto/bare-rpc-python) -to a real spawned SDK worker, replacing poc_heartbeat.py's hand-rolled frame -encode/decode with the real library. - -Kept deliberately separate from poc_heartbeat.py/poc_transport.py (used by -PR #3100) rather than replacing them -- this is exploratory, and bare_rpc is -asyncio-native so `BareRpcWorker` matches qvac._transport.Transport's async -shape directly (call/call_stream/call_duplex), no PocTransport-style adapter -needed. See the PR/commit notes for what's proven here vs. still open. - -Wire-level notes: -- Our worker ignores bare-rpc's `command` field entirely and routes purely - on the JSON payload's own `type` field (see poc_heartbeat.py's comment: - "command == id: the server ignores `command` and routes on payload.type"). - So every call below passes command=0 -- bare_rpc.RPC's own internal - monotonic id (not `command`) is what correlates request/response/stream - frames on the wire, and that's handled entirely inside the library. -- Server-stream replies arrive as newline-delimited JSON, but not - necessarily one JSON document per wire STREAM-DATA frame (the worker may - coalesce or split them) -- so chunks from `IncomingStream` still need the - same buffer-and-split-on-newline handling poc_heartbeat.py's call_stream - does, just fed from bare_rpc's discrete async-iterated chunks instead of - raw socket bytes. -- Duplex maps onto `RPC.create_bidirectional_stream`: the returned - `(outgoing, incoming)` pair already IS the request/response stream pair - poc_heartbeat.py's hand-rolled `_duplex_call` builds frame-by-frame -- the - library owns the OPEN/RESUME/PAUSE control handshake, we only `write()` - the JSON payload as the first outgoing chunk followed by `up`'s chunks, - `end()` the outgoing side, and read `incoming` with the same - buffer-and-split-on-newline handling as call_stream. The two sides run - concurrently (a background task pumps `up` while `incoming` is iterated) - since the worker can start responding before the client finishes sending. +"""Production `qvac._transport.Transport` implementation, backed by +`bare_rpc.RPC` (github.com/holepunchto/bare-rpc-python). + +Mirrors the JS SDK's `client/rpc/node-rpc-client.ts`: create a Unix +socket / Windows named pipe, spawn the worker with that path (plus +`HOME_DIR`) as its JSON config argument, and wire `bare_rpc.RPC` to the +connection the worker dials back in on. + +Locating the Bare binary and the worker's JS entry point is not this +module's job — `command` is supplied by the caller. This is a thin +client that expects the worker to be available separately; bundling +those artifacts into the installed package is a separate concern. """ from __future__ import annotations @@ -37,10 +17,8 @@ import asyncio import json import os -import subprocess import tempfile -from pathlib import Path -from typing import Any, AsyncIterable, AsyncIterator +from typing import Any, AsyncIterable, AsyncIterator, Sequence try: import bare_rpc @@ -49,12 +27,13 @@ BARE_RPC_AVAILABLE = bare_rpc is not None -SDK = os.environ.get( - "QVAC_POC_SDK_DIR", - str(Path(__file__).resolve().parent.parent.parent / "sdk"), -) -BARE = f"{SDK}/node_modules/bare-runtime-darwin-arm64/bin/bare" -WORKER = f"{SDK}/dist/server/worker.js" + +class BareRpcNotInstalledError(ImportError): + def __init__(self) -> None: + super().__init__( + "bare_rpc is not installed -- install the 'bare-rpc' extra " + "(`pip install qvac[bare-rpc]`) to use BareRpcTransport" + ) def _json_or_raise(data: bytes) -> Any: @@ -65,25 +44,32 @@ def _json_or_raise(data: bytes) -> Any: return obj -class BareRpcWorker: - """Spawns the real Bare worker and speaks to it via bare_rpc.RPC instead - of hand-rolled framing. asyncio-native, unlike poc_heartbeat.QvacWorker.""" +class BareRpcTransport: + """Spawns a QVAC SDK worker and speaks to it via `bare_rpc.RPC`, + satisfying `qvac._transport.Transport`'s async call/call_stream/call_duplex + shape directly. - def __init__(self) -> None: + `command` is the Bare invocation up to (not including) the worker's + JSON config argument, e.g. `["bare", "/path/to/worker.js"]` — this + class appends `{"QVAC_IPC_SOCKET_PATH": ..., "HOME_DIR": ...}` itself, + since that handshake is protocol, not caller, concern. + """ + + def __init__(self, command: Sequence[str], *, home_dir: str | None = None) -> None: + if bare_rpc is None: + raise BareRpcNotInstalledError() + self._command = list(command) + self._home_dir = home_dir or os.path.expanduser("~") self._sock_path = os.path.join( - tempfile.gettempdir(), f"qvac-bare-rpc-poc-{os.getpid()}.sock" - ) - self._log_path = os.path.join( - tempfile.gettempdir(), f"qvac-bare-rpc-poc-worker-{os.getpid()}.log" + tempfile.gettempdir(), f"qvac-worker-{os.getpid()}-{id(self)}.sock" ) self._server: asyncio.AbstractServer | None = None self._proc: asyncio.subprocess.Process | None = None - self._log_fh = None self._writer: asyncio.StreamWriter | None = None self._read_task: asyncio.Task | None = None self.rpc: bare_rpc.RPC | None = None - async def start(self) -> "BareRpcWorker": + async def connect(self, *, timeout: float = 30) -> "BareRpcTransport": if os.path.exists(self._sock_path): os.unlink(self._sock_path) @@ -108,26 +94,19 @@ async def on_client( self._server = await asyncio.start_unix_server(on_client, path=self._sock_path) config = json.dumps( - { - "QVAC_IPC_SOCKET_PATH": self._sock_path, - "HOME_DIR": os.path.expanduser("~"), - } - ) - self._log_fh = open(self._log_path, "wb") - self._proc = await asyncio.create_subprocess_exec( - BARE, - WORKER, - config, - cwd=SDK, - stdout=self._log_fh, - stderr=subprocess.STDOUT, + {"QVAC_IPC_SOCKET_PATH": self._sock_path, "HOME_DIR": self._home_dir} ) + self._proc = await asyncio.create_subprocess_exec(*self._command, config) def send(frame: bytes) -> None: self._writer.write(frame) self.rpc = bare_rpc.RPC(send=send) - await asyncio.wait_for(connected, timeout=30) + try: + await asyncio.wait_for(connected, timeout=timeout) + except asyncio.TimeoutError: + await self.close() + raise return self async def close(self) -> None: @@ -149,25 +128,16 @@ async def close(self) -> None: await asyncio.wait_for(self._server.wait_closed(), timeout=5) except asyncio.TimeoutError: pass - if self._log_fh: - self._log_fh.close() if os.path.exists(self._sock_path): os.unlink(self._sock_path) - async def __aenter__(self) -> "BareRpcWorker": - return await self.start() + async def __aenter__(self) -> "BareRpcTransport": + return await self.connect() async def __aexit__(self, *exc: object) -> None: await self.close() - def worker_logs(self) -> str: - try: - with open(self._log_path, "r", errors="replace") as f: - return f.read() - except OSError: - return "" - - # ---- the three call shapes ----------------------------------------- + # ---- Transport protocol ---------------------------------------------- async def call(self, payload: dict) -> dict: """Unary, via bare_rpc.RPC.request -- no hand-rolled framing at all.""" diff --git a/packages/sdk-python/tests/test_bare_rpc_transport.py b/packages/sdk-python/tests/test_bare_rpc_transport.py new file mode 100644 index 0000000000..c8a3bd5200 --- /dev/null +++ b/packages/sdk-python/tests/test_bare_rpc_transport.py @@ -0,0 +1,263 @@ +"""Same real-worker rigor as test_poc_smoke.py/test_poc_progress.py, but +exercising the production `qvac.bare_rpc_transport.BareRpcTransport` +against a real spawned SDK worker for all three wire call shapes, +including both duplex-shaped methods beyond text-to-speech: parakeet's +`transcribeStream` and the BCI addon's `bciTranscribeStream`. +""" + +from __future__ import annotations + +import array +import asyncio +import os +import wave +from pathlib import Path + +import pytest +import pytest_asyncio + +from qvac.bare_rpc_transport import BARE_RPC_AVAILABLE, BareRpcTransport +from qvac.models import ( + BCI_WINDOWED, + PARAKEET_CTC_0_6B_Q4_0, + QWEN3_600M_INST_Q4, + TTS_EN_SUPERTONIC_Q4_0, +) +from qvac.schemas import ( + BciTranscribeStreamRequest, + CompletionStreamRequest, + HeartbeatRequest, + LoadModelRequest, + ModelType, + TextToSpeechStreamRequest, + TranscribeStreamRequest, +) +from qvac.methods import ( + bci_transcribe_stream, + completion_stream, + heartbeat, + load_model, + text_to_speech_stream, + transcribe_stream, +) + +SDK_DIR = os.environ.get( + "QVAC_POC_SDK_DIR", + str(Path(__file__).resolve().parent.parent.parent / "sdk"), +) +BARE_BIN = f"{SDK_DIR}/node_modules/bare-runtime-darwin-arm64/bin/bare" +WORKER_PATH = f"{SDK_DIR}/dist/server/worker.js" +AUDIO_FIXTURE = f"{SDK_DIR}/e2e/assets/audio/transcription-short-wav.wav" +NEURAL_FIXTURE = f"{SDK_DIR}/e2e/assets/neural/neural-not-too-controversial.bin" + +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif( + not BARE_RPC_AVAILABLE, + reason="bare_rpc not installed -- install the 'bare-rpc' extra " + "(`pip install -e '.[bare-rpc]'`) to run these tests", + ), + pytest.mark.skipif( + not os.path.exists(WORKER_PATH), + reason=f"no built SDK worker found at {WORKER_PATH!r} -- run `bun run build` in packages/sdk, or set QVAC_POC_SDK_DIR", + ), +] + + +@pytest_asyncio.fixture +async def transport(): + async with BareRpcTransport([BARE_BIN, WORKER_PATH]) as t: + yield t + + +async def test_heartbeat_unary(transport) -> None: + response = await heartbeat(transport, HeartbeatRequest(type="heartbeat")) + assert response.type == "heartbeat" + assert isinstance(response.number, float) + + +async def test_load_model_and_completion_stream(transport) -> None: + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": QWEN3_600M_INST_Q4.src, + "modelType": "llamacpp-completion", + "modelConfig": {}, + } + ) + load_response = await load_model(transport, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + completion_request = CompletionStreamRequest.model_validate( + { + "type": "completionStream", + "modelId": model_id, + "history": [{"role": "user", "content": "Say hello in five words."}], + "stream": True, + } + ) + + text = "" + async for chunk in completion_stream(transport, completion_request): + for event in chunk.events: + if event.type == "contentDelta": + text += event.text + assert text.strip(), "expected real completion text via the bare_rpc server-stream" + + +async def _as_async_iter(items): + for item in items: + yield item + + +async def _paced_chunks(chunks, delay_s): + for i, chunk in enumerate(chunks): + yield chunk + if delay_s > 0 and i < len(chunks) - 1: + await asyncio.sleep(delay_s) + + +async def test_load_model_and_tts_stream_duplex(transport) -> None: + """Exercises call_duplex end to end: text goes up the request stream while + synthesized audio comes down the response stream, concurrently.""" + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": TTS_EN_SUPERTONIC_Q4_0.src, + "modelType": ModelType.tts_ggml, + "modelConfig": {"ttsEngine": "supertonic", "language": "en"}, + } + ) + load_response = await load_model(transport, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + tts_request = TextToSpeechStreamRequest.model_validate( + {"type": "textToSpeechStream", "modelId": model_id} + ) + text = b"Hello from QVAC. This is streaming text to speech." + + samples = [] + saw_done = False + async for chunk in text_to_speech_stream( + transport, tts_request, _as_async_iter([text]) + ): + samples.extend(chunk.buffer) + saw_done = saw_done or chunk.done + assert samples, "expected real synthesized audio via the bare_rpc duplex stream" + assert saw_done, "expected a terminal done=True event on the response stream" + + +def _wav_to_s16le_mono_16k(path: str) -> bytes: + """Decode a 16-bit PCM wav to 16 kHz mono s16le -- the wire format parakeet's + duplex `transcribeStream` expects (see the SDK e2e `parakeet-stream-runner.ts`, + which converts its f32 fixture samples to s16le bytes before writing them).""" + with wave.open(path, "rb") as wf: + channels, width, rate = wf.getnchannels(), wf.getsampwidth(), wf.getframerate() + raw = wf.readframes(wf.getnframes()) + if width != 2: + raise RuntimeError(f"expected 16-bit PCM wav, got sampwidth={width}") + samples = array.array("h") + samples.frombytes(raw) + mono = array.array("h", samples[0::channels]) if channels > 1 else samples + target = 16000 + if rate != target: + if rate % target != 0: + raise RuntimeError(f"can't cleanly decimate {rate}Hz to {target}Hz") + mono = array.array("h", mono[0 :: rate // target]) + return mono.tobytes() + + +async def test_transcribe_stream_duplex(transport) -> None: + """Parakeet's streaming session only decodes when fed at roughly real-time + cadence (see transcription-parakeet's live-stream-simulation.test.js / + duplex-streaming tests) -- chunks are paced with a real `asyncio.sleep` + between writes, matching the SDK e2e runner's `writeInChunks(delayMs)`.""" + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": PARAKEET_CTC_0_6B_Q4_0.src, + "modelType": ModelType.parakeet_transcription, + "modelConfig": {}, + } + ) + load_response = await load_model(transport, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + chunk_ms = 1000 + pcm = _wav_to_s16le_mono_16k(AUDIO_FIXTURE) + bytes_per_chunk = int(16000 * chunk_ms / 1000) * 2 + trailing_silence = bytes( + int(16000 * 1.5) * 2 + ) # settle time so the stream finalizes + chunks = [pcm[i : i + bytes_per_chunk] for i in range(0, len(pcm), bytes_per_chunk)] + chunks += [ + trailing_silence[i : i + bytes_per_chunk] + for i in range(0, len(trailing_silence), bytes_per_chunk) + ] + + transcribe_request = TranscribeStreamRequest.model_validate( + { + "type": "transcribeStream", + "modelId": model_id, + "parakeetStreamingConfig": {"chunkMs": chunk_ms, "emitPartials": True}, + } + ) + + text = "" + async for response in transcribe_stream( + transport, transcribe_request, _paced_chunks(chunks, chunk_ms / 1000) + ): + piece = response.text or (response.segment.text if response.segment else None) + if piece: + text += piece + assert text.strip(), "expected real transcript text via the bare_rpc duplex stream" + + +async def test_bci_transcribe_stream_duplex(transport) -> None: + """The BCI addon's sliding-window driver is fed arbitrary-size chunks with + no real-time pacing requirement (see the SDK e2e `bci-executor.ts`, which + writes 64 KiB slices back-to-back) -- unlike parakeet's transcribeStream.""" + load_request = LoadModelRequest.model_validate( + { + "type": "loadModel", + "modelSrc": BCI_WINDOWED.src, + "modelType": ModelType.bci_whispercpp_transcription, + "modelConfig": { + "whisperConfig": {"language": "en", "temperature": 0.0}, + "miscConfig": {"caption_enabled": False}, + # neural-not-too-controversial.bin was recorded on session day 1. + "bciConfig": {"day_idx": 1}, + }, + } + ) + load_response = await load_model(transport, load_request) + assert load_response.success, load_response.error + model_id = load_response.model_id + + bci_request = BciTranscribeStreamRequest.model_validate( + { + "type": "bciTranscribeStream", + "modelId": model_id, + "streamOpts": {"emit": "delta"}, + } + ) + + with open(NEURAL_FIXTURE, "rb") as f: + neural_bytes = f.read() + chunk_size = 64 * 1024 + chunks = [ + neural_bytes[i : i + chunk_size] + for i in range(0, len(neural_bytes), chunk_size) + ] + + text = "" + async for response in bci_transcribe_stream( + transport, bci_request, _as_async_iter(chunks) + ): + piece = response.text or (response.segment.text if response.segment else None) + if piece: + text += piece + assert "controversial" in text.lower(), f"unexpected BCI transcript: {text!r}" diff --git a/packages/sdk-python/tests/test_poc_bare_rpc.py b/packages/sdk-python/tests/test_poc_bare_rpc.py deleted file mode 100644 index 648ff738f9..0000000000 --- a/packages/sdk-python/tests/test_poc_bare_rpc.py +++ /dev/null @@ -1,120 +0,0 @@ -"""QVAC-21806 PoC: same real-worker rigor as test_poc_smoke.py/test_poc_progress.py, -but the wire encode/decode is bare_rpc.RPC instead of poc_heartbeat.QvacWorker's -hand-rolled framing. Validates responses against the same generated pydantic -models either way -- proving bare_rpc is a drop-in replacement at the framing -layer without changing anything above it. -""" - -from __future__ import annotations - -import os - -import pytest -import pytest_asyncio - -from poc_bare_rpc_transport import BARE_RPC_AVAILABLE, WORKER -from qvac.models import QWEN3_600M_INST_Q4, TTS_EN_SUPERTONIC_Q4_0 -from qvac.schemas import ( - CompletionStreamRequest, - HeartbeatRequest, - LoadModelRequest, - ModelType, - TextToSpeechStreamRequest, -) -from qvac.methods import completion_stream, heartbeat, load_model, text_to_speech_stream - -pytestmark = [ - pytest.mark.asyncio, - pytest.mark.skipif( - not BARE_RPC_AVAILABLE, - reason="bare_rpc not installed -- install the 'bare-rpc' extra " - "(`pip install -e '.[bare-rpc]'`) to run these PoC tests", - ), - pytest.mark.skipif( - not os.path.exists(WORKER), - reason=f"no built SDK worker found at {WORKER!r} -- run `bun run build` in packages/sdk, or set QVAC_POC_SDK_DIR", - ), -] - - -@pytest_asyncio.fixture -async def worker(): - from poc_bare_rpc_transport import BareRpcWorker - - async with BareRpcWorker() as w: - yield w - - -async def test_heartbeat_unary_via_bare_rpc(worker) -> None: - """BareRpcWorker.call/.call_stream already match qvac._transport.Transport's - async shape exactly, so it plugs straight into the real generated stubs -- - no PocTransport-style adapter needed, unlike poc_heartbeat.QvacWorker.""" - response = await heartbeat(worker, HeartbeatRequest(type="heartbeat")) - assert response.type == "heartbeat" - assert isinstance(response.number, float) - - -async def test_load_model_and_completion_stream_via_bare_rpc(worker) -> None: - load_request = LoadModelRequest.model_validate( - { - "type": "loadModel", - "modelSrc": QWEN3_600M_INST_Q4.src, - "modelType": "llamacpp-completion", - "modelConfig": {}, - } - ) - load_response = await load_model(worker, load_request) - assert load_response.success, load_response.error - model_id = load_response.model_id - - completion_request = CompletionStreamRequest.model_validate( - { - "type": "completionStream", - "modelId": model_id, - "history": [{"role": "user", "content": "Say hello in five words."}], - "stream": True, - } - ) - - text = "" - async for chunk in completion_stream(worker, completion_request): - for event in chunk.events: - if event.type == "contentDelta": - text += event.text - assert text.strip(), "expected real completion text via the bare_rpc server-stream" - - -async def _as_async_iter(items): - for item in items: - yield item - - -async def test_load_model_and_tts_stream_duplex_via_bare_rpc(worker) -> None: - """Exercises call_duplex end to end: text goes up the request stream while - synthesized audio comes down the response stream, concurrently.""" - load_request = LoadModelRequest.model_validate( - { - "type": "loadModel", - "modelSrc": TTS_EN_SUPERTONIC_Q4_0.src, - "modelType": ModelType.tts_ggml, - "modelConfig": {"ttsEngine": "supertonic", "language": "en"}, - } - ) - load_response = await load_model(worker, load_request) - assert load_response.success, load_response.error - model_id = load_response.model_id - - tts_request = TextToSpeechStreamRequest.model_validate( - {"type": "textToSpeechStream", "modelId": model_id} - ) - text = b"Hello from QVAC. This is streaming text to speech." - - samples = [] - saw_done = False - async for chunk in text_to_speech_stream( - worker, tts_request, _as_async_iter([text]) - ): - samples.extend(chunk.buffer) - saw_done = saw_done or chunk.done - assert samples, "expected real synthesized audio via the bare_rpc duplex stream" - assert saw_done, "expected a terminal done=True event on the response stream" From c9d5d60b8e53b796b38e4d6e5034e1b4307f22f6 Mon Sep 17 00:00:00 2001 From: Lauri Piisang Date: Fri, 17 Jul 2026 01:27:39 +0300 Subject: [PATCH 6/6] QVAC-21806 fix: bring bare-rpc transport up to main's lint/typecheck gates Rebasing onto merged #3100 pulls in ruff+mypy CI for sdk-python; the bare-rpc transport predates it. Bring it into line: - ruff: typing.AsyncIterable/AsyncIterator/Sequence -> collections.abc, drop redundant forward-ref quotes, sort the test import block - mypy: assert post-connect invariants (rpc/writer set) and add a _require_rpc guard so the Transport methods don't deref Optional; add a scoped ignore_missing_imports override for the optional, git-only bare_rpc/compact_encoding extras - test: load the Qwen3 completion model with an explicit n_ctx -- it's a thinking model, so the worker reserves context for the reasoning trace and the metadata-default budget overflowed even a tiny prompt --- packages/sdk-python/pyproject.toml | 7 +++++ .../sdk-python/src/qvac/bare_rpc_transport.py | 26 ++++++++++++++----- .../tests/test_bare_rpc_transport.py | 21 ++++++++------- 3 files changed, 39 insertions(+), 15 deletions(-) diff --git a/packages/sdk-python/pyproject.toml b/packages/sdk-python/pyproject.toml index 311a7cfaed..d7519a1a7d 100644 --- a/packages/sdk-python/pyproject.toml +++ b/packages/sdk-python/pyproject.toml @@ -59,3 +59,10 @@ module = "qvac._generated.models._internal" # enum member on construction (verified: `Model().field == SomeEnum.x` # is True). Scoped to this one generated module, not project-wide. disable_error_code = ["assignment"] + +[[tool.mypy.overrides]] +module = ["bare_rpc", "compact_encoding"] +# The bare-rpc extra is optional and git-only (no PyPI release, no py.typed +# marker), so it's absent from the default install CI type-checks against. +# bare_rpc_transport.py already guards the import at runtime. +ignore_missing_imports = true diff --git a/packages/sdk-python/src/qvac/bare_rpc_transport.py b/packages/sdk-python/src/qvac/bare_rpc_transport.py index f60240c3b1..b2cee750fe 100644 --- a/packages/sdk-python/src/qvac/bare_rpc_transport.py +++ b/packages/sdk-python/src/qvac/bare_rpc_transport.py @@ -18,7 +18,8 @@ import json import os import tempfile -from typing import Any, AsyncIterable, AsyncIterator, Sequence +from collections.abc import AsyncIterable, AsyncIterator, Sequence +from typing import Any try: import bare_rpc @@ -69,7 +70,7 @@ def __init__(self, command: Sequence[str], *, home_dir: str | None = None) -> No self._read_task: asyncio.Task | None = None self.rpc: bare_rpc.RPC | None = None - async def connect(self, *, timeout: float = 30) -> "BareRpcTransport": + async def connect(self, *, timeout: float = 30) -> BareRpcTransport: if os.path.exists(self._sock_path): os.unlink(self._sock_path) @@ -82,6 +83,9 @@ async def on_client( self._read_task = asyncio.current_task() if not connected.done(): connected.set_result(None) + # rpc is created right after the worker is spawned below, long + # before the worker can start and dial back into this callback. + assert self.rpc is not None try: while True: chunk = await reader.read(65536) @@ -99,6 +103,9 @@ async def on_client( self._proc = await asyncio.create_subprocess_exec(*self._command, config) def send(frame: bytes) -> None: + # bare_rpc only calls send once it has a connection, at which + # point on_client above has already set the writer. + assert self._writer is not None self._writer.write(frame) self.rpc = bare_rpc.RPC(send=send) @@ -131,7 +138,7 @@ async def close(self) -> None: if os.path.exists(self._sock_path): os.unlink(self._sock_path) - async def __aenter__(self) -> "BareRpcTransport": + async def __aenter__(self) -> BareRpcTransport: return await self.connect() async def __aexit__(self, *exc: object) -> None: @@ -139,16 +146,21 @@ async def __aexit__(self, *exc: object) -> None: # ---- Transport protocol ---------------------------------------------- + def _require_rpc(self) -> Any: + if self.rpc is None: + raise RuntimeError("BareRpcTransport used before connect()") + return self.rpc + async def call(self, payload: dict) -> dict: """Unary, via bare_rpc.RPC.request -- no hand-rolled framing at all.""" - data = await self.rpc.request( + data = await self._require_rpc().request( command=0, data=json.dumps(payload).encode("utf-8") ) return _json_or_raise(data) async def call_stream(self, payload: dict) -> AsyncIterator[dict]: """Server-stream, via bare_rpc.RPC.request_with_response_stream.""" - stream = await self.rpc.request_with_response_stream( + stream = await self._require_rpc().request_with_response_stream( command=0, data=json.dumps(payload).encode("utf-8") ) buffer = "" @@ -168,7 +180,9 @@ async def call_duplex( """Duplex, via bare_rpc.RPC.create_bidirectional_stream -- first outgoing chunk is the JSON payload, then `up`'s chunks; yields parsed response chunks with the same buffer-and-split-on-newline handling as call_stream.""" - outgoing, incoming = await self.rpc.create_bidirectional_stream(command=0) + outgoing, incoming = await self._require_rpc().create_bidirectional_stream( + command=0 + ) await outgoing.write(json.dumps(payload).encode("utf-8")) async def _pump_up() -> None: diff --git a/packages/sdk-python/tests/test_bare_rpc_transport.py b/packages/sdk-python/tests/test_bare_rpc_transport.py index c8a3bd5200..0ebac275e7 100644 --- a/packages/sdk-python/tests/test_bare_rpc_transport.py +++ b/packages/sdk-python/tests/test_bare_rpc_transport.py @@ -17,6 +17,14 @@ import pytest_asyncio from qvac.bare_rpc_transport import BARE_RPC_AVAILABLE, BareRpcTransport +from qvac.methods import ( + bci_transcribe_stream, + completion_stream, + heartbeat, + load_model, + text_to_speech_stream, + transcribe_stream, +) from qvac.models import ( BCI_WINDOWED, PARAKEET_CTC_0_6B_Q4_0, @@ -32,14 +40,6 @@ TextToSpeechStreamRequest, TranscribeStreamRequest, ) -from qvac.methods import ( - bci_transcribe_stream, - completion_stream, - heartbeat, - load_model, - text_to_speech_stream, - transcribe_stream, -) SDK_DIR = os.environ.get( "QVAC_POC_SDK_DIR", @@ -82,7 +82,10 @@ async def test_load_model_and_completion_stream(transport) -> None: "type": "loadModel", "modelSrc": QWEN3_600M_INST_Q4.src, "modelType": "llamacpp-completion", - "modelConfig": {}, + # Qwen3 is a thinking model: the worker reserves context for the + # reasoning trace, so the metadata-default budget overflows even a + # tiny prompt. Give it an explicit window (matches the SDK e2e). + "modelConfig": {"n_ctx": 2048}, } ) load_response = await load_model(transport, load_request)