feat: serveStream / VgiRpcServer.serveConnection — serve over any byte stream

Add a public server-side serve-over-stream seam (the sibling of the client's
pipeConnect), so a worker can be served over a duplex channel that isn't a
process or socket — e.g. a Web Worker MessagePort bridge, or an in-memory pipe.

- VgiRpcServer.serveConnection(readable, writable, transportKind?): the
  transport-agnostic core; run() (stdin/stdout) now delegates to it, so the
  stdio path is unchanged.
- serveStream(protocol, { readable, writable }): convenience wrapper, exported
  next to the server.

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

Author: rustyconover

src/index.core.ts

@@ -92,6 +92,7 @@ export {
   uint64,
 } from "./schema.js";
 export { VgiRpcServer } from "./server.js";
+export { serveStream, type ServeStreamOptions } from "./serve-stream.js";
 export {
   type CallContext,
   type CallStatistics,

src/serve-stream.ts

@@ -0,0 +1,38 @@
+// © Copyright 2025-2026, Query.Farm LLC - https://query.farm
+// SPDX-License-Identifier: Apache-2.0
+
+//! Serve a protocol over a caller-provided byte-stream pair — the stream
+//! sibling of `serveTcp` / `serveUnix`, with no socket/listener of its own.
+//!
+//! Useful for transports the launcher helpers don't cover: a Web Worker /
+//! `MessagePort` bridge (postMessage), an in-memory pipe, or a pre-connected
+//! socket. The host side already has this symmetry via `pipeConnect`.
+
+import type { Socket } from "node:net";
+
+import type { Protocol } from "./protocol.js";
+import { VgiRpcServer } from "./server.js";
+import { TransportKind } from "./types.js";
+
+export interface ServeStreamOptions {
+  /** Incoming request bytes — a web `ReadableStream<Uint8Array>` or a Node
+   *  `Readable` (e.g. a `Duplex` bridging a MessagePort). */
+  readable: ReadableStream<Uint8Array> | NodeJS.ReadableStream;
+  /** Outgoing response sink — a stdout-like fd number, or a `net.Socket` /
+   *  structurally-compatible `Duplex`. Omit for the stdout fd. */
+  writable?: number | Socket;
+  /** Passed through to the `VgiRpcServer` constructor (describe, hooks, …). */
+  serverOptions?: ConstructorParameters<typeof VgiRpcServer>[1];
+  /** Reported to the `on_serve_start` hook. Defaults to `PIPE`. */
+  transportKind?: TransportKind;
+}
+
+/**
+ * Serve `protocol` over the provided `readable`/`writable` until the readable
+ * ends. Thin wrapper over {@link VgiRpcServer.serveConnection}. Resolves on
+ * clean EOF; rejects on a real protocol/transport error.
+ */
+export async function serveStream(protocol: Protocol, options: ServeStreamOptions): Promise<void> {
+  const server = new VgiRpcServer(protocol, options.serverOptions);
+  await server.serveConnection(options.readable, options.writable, options.transportKind);
+}

src/server.ts

@@ -161,10 +161,8 @@ export class VgiRpcServer {
     );
   }
 
-  /** Start the server loop. Reads requests until stdin closes. */
+  /** Start the server loop over stdin/stdout. Reads requests until stdin closes. */
   async run(): Promise<void> {
-    const stdin = process.stdin as unknown as ReadableStream<Uint8Array>;
-
     // Warn if running interactively
     if (process.stdin.isTTY || process.stdout.isTTY) {
       process.stderr.write(
@@ -174,21 +172,44 @@ export class VgiRpcServer {
           "(e.g. vgi_rpc.connect()).\n",
       );
     }
+    const stdin = process.stdin as unknown as ReadableStream<Uint8Array>;
+    // writable omitted → IpcStreamWriter defaults to the stdout fd.
+    await this.serveConnection(stdin);
+  }
 
-    const reader = await IpcStreamReader.create(stdin);
-    const writer = new IpcStreamWriter();
+  /**
+   * Serve requests over an explicit byte-stream pair until the readable ends —
+   * the transport-agnostic core that {@link run} (stdin/stdout) is built on.
+   *
+   * Use this to serve over any duplex channel that the stdio/unix/tcp helpers
+   * don't cover: a Web Worker / `MessagePort` bridge, an in-memory pipe, or a
+   * pre-connected socket. The loop, on_serve_start firing, and EOF/broken-pipe
+   * handling are identical to {@link run}.
+   *
+   * @param readable incoming request bytes — a web `ReadableStream<Uint8Array>`
+   *   or a Node `Readable` (e.g. a `Duplex` bridging a MessagePort).
+   * @param writable outgoing response sink — a stdout-like fd number, or a
+   *   `net.Socket` / structurally-compatible `Duplex`. Omit for the stdout fd.
+   * @param transportKind reported to the `on_serve_start` hook (default `PIPE`).
+   */
+  async serveConnection(
+    readable: ReadableStream<Uint8Array> | NodeJS.ReadableStream,
+    writable?: number | import("node:net").Socket,
+    transportKind: TransportKind = TransportKind.PIPE,
+  ): Promise<void> {
+    const reader = await IpcStreamReader.create(readable);
+    const writer = new IpcStreamWriter(writable);
 
     try {
       while (true) {
-        // Fire on_serve_start lazily so the hook can do work that
-        // depends on the transport binding (matches Python which fires
-        // it inside serve()). Inside the loop so a failure on the very
+        // Fire on_serve_start lazily so the hook can do work that depends on
+        // the transport binding. Inside the loop so a failure on the very
         // first request can be retried.
-        await this.notifyTransport(TransportKind.PIPE);
+        await this.notifyTransport(transportKind);
         await this.serveOne(reader, writer);
       }
     } catch (e: any) {
-      // EOF or broken pipe → clean exit
+      // EOF or broken pipe / closed channel → clean exit
       if (
         e.message?.includes("closed") ||
         e.message?.includes("Expected Schema Message") ||