feat(gateway): a gateway module can hold a two-way byte channel to a worker module

src/contract/AGENTS.md

@@ -57,7 +57,7 @@ Three identities:
 
 ## Protocol version
 
-- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 63, max: 63 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`.
+- `PROTOCOL_VERSION_RANGE` in `protocol.ts` holds it, now `{ min: 64, max: 64 }`. `DAEMON_PROTOCOL_VERSION` (`src/daemon-protocol/`) is its `max`.
 - Every change to a contract shape bumps it by one in the same PR. Move `min` too, unless
   a compatibility path stays side by side. `.agents/scripts/protocol-bump-check.mjs` checks
   it on push and in CI; it counts every file here as a shape but tests, `index.ts` and the

src/contract/operations.test.ts

@@ -1724,6 +1724,16 @@ describe("relay.request on a byte channel", () => {
     expect(parse(request)).toBe(false);
   });
 
+  it("accepts twoWay with GET and no body, and refuses it with another method, with a body, or as false, where the same request without twoWay is accepted", () => {
+    const channel = { ...request, channel: "a" };
+    expect(parse({ ...channel, twoWay: true })).toBe(true);
+    expect(parse({ ...channel, method: "POST" })).toBe(true);
+    expect(parse({ ...channel, method: "POST", twoWay: true })).toBe(false);
+    expect(parse({ ...channel, hasBody: true, method: "PUT" })).toBe(true);
+    expect(parse({ ...channel, hasBody: true, method: "PUT", twoWay: true })).toBe(false);
+    expect(parse({ ...channel, twoWay: false })).toBe(false);
+  });
+
   it("OPERATIONS has no relay.body and PUSH_SCHEMAS has no relay-chunk", () => {
     expect(Object.keys(OPERATIONS)).not.toContain("relay.body");
     expect(Object.keys(PUSH_SCHEMAS)).not.toContain("relay-chunk");

src/contract/operations.ts

@@ -972,7 +972,8 @@ const relayUrlSchema = z
  * request body arrives, and the response body leaves, on the byte channel `channel` names (ADR 0028);
  * the head leaves as a `relay-head` push keyed by this call's frame id. The
  * answer comes once the response body has ended, or `{ cancelled: true }` once a `relay.cancel` took
- * effect. Admin only, and on a worker only the gateway's own uplink session may call it.
+ * effect. With `twoWay` (GET, no body) both directions of the channel stay open: the answer comes once
+ * both have ended. Admin only, and on a worker only the gateway's own uplink session may call it.
  */
 // fallow-ignore-next-line unused-export -- consumed only through the OPERATIONS registry, not by name; still public contract surface.
 export const relayRequest = defineOperation({
@@ -989,11 +990,15 @@ export const relayRequest = defineOperation({
       tenant: z.string().min(1).max(300),
       leaseId: leaseIdSchema.optional(),
       hasBody: z.boolean(),
+      twoWay: z.literal(true).optional(),
       channel: relayChannelSchema,
     })
     .strict()
     .refine((input) => !input.hasBody || (input.method !== "GET" && input.method !== "HEAD"), {
       message: "hasBody is not allowed with GET or HEAD",
+    })
+    .refine((input) => input.twoWay !== true || (input.method === "GET" && !input.hasBody), {
+      message: "twoWay needs GET and no body",
     }),
   output: z.object({ cancelled: z.literal(true).optional() }),
 });

src/contract/protocol.ts

@@ -10,7 +10,7 @@ export interface ProtocolRange {
 const protocolRangeSchema = z.object({ min: z.number().int(), max: z.number().int() });
 
 /** Widen only once a second version is kept alive side by side with the first; a bump that leaves no shim behind moves both ends. */
-export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 63, max: 63 };
+export const PROTOCOL_VERSION_RANGE: ProtocolRange = { min: 64, max: 64 };
 
 const LEGACY_DAEMON_PROTOCOL_VERSION = 2;
 

src/daemon-protocol/index.test.ts

@@ -9,7 +9,7 @@ import {
 
 describe("daemon protocol", () => {
   it("keeps the current protocol version and newline framing", () => {
-    expect(DAEMON_PROTOCOL_VERSION).toBe(63);
+    expect(DAEMON_PROTOCOL_VERSION).toBe(64);
     expect(serializeFrame({ id: 1, type: "hello" })).toBe('{"id":1,"type":"hello"}\n');
   });
 

src/daemon/AGENTS.md

@@ -59,7 +59,7 @@ Its stdout lines move the state; a failed line, a non-zero exit or a failed laun
 2. **Hello** — first frame must be `hello`, else `HANDSHAKE_REQUIRED` and close.
    Role: the uplink connection is `admin` by acceptance; otherwise the resolver. Principal: `hello.principal`, else `SIMLOCK_AGENT_ID`, else the daemon's pid. Both are fixed for the connection's life.
 3. **Version** — ranges overlap → highest common version. `PROTOCOL_VERSION_RANGE`
-   is `{ min: 63, max: 63 }` (`src/contract/protocol.ts`). No overlap → the
+   is `{ min: 64, max: 64 }` (`src/contract/protocol.ts`). No overlap → the
    connection stays open, and every request but `daemon.stop` gets
    `PROTOCOL_VERSION_UNSUPPORTED`.
 4. **Request** — one frame → one `dispatch()` call. While stopping: `DAEMON_STOPPING`.
@@ -122,6 +122,7 @@ Security model: cooperative identity, not a hostile-process boundary.
 - `GatewayUplink` dials `gateway.url` with `gateway.token`, after the first host load reading. Backoff from 1 s, ×2, cap 30 s, jittered; retries forever.
 - `DaemonServer.acceptUplink` serves it as one more connection, role `admin`.
 - `relay.request` runs only on that connection; targets come from `relayTargets` (today `agent-device` and `stream`). See [gateway](../gateway/AGENTS.md).
+- A `relay.request` with `twoWay` takes its target from `relayTwoWayTargets` (a module's `relayTwoWay`, by module name; none → `RELAY_TARGET_MISSING`) and runs no body cut-off: the target gets `pipe = {readable, writable, signal}` on the channel's two directions. It answers a refusal before touching the pipe, or a 2xx with no body and then uses the pipe; the call answers once both directions have ended. `relay.cancel`, `abortAll` and a closed uplink abort `pipe.signal`; a target that fails after its head aborts `pipe.writable`. The 15 s silence bound is the gateway's.
 
 ## Logging and event records
 

src/daemon/main.ts

@@ -150,7 +150,7 @@ import { describeLeaseRequestFailure } from "./error-code.js";
 import { wakingQueueAfter } from "./enrollment-wake.js";
 import { execBaseEnvironment } from "./exec-env.js";
 import { GatewayUplink } from "./gateway-uplink.js";
-import type { RelayTargets } from "./relay.js";
+import type { RelayTargets, TwoWayRelayTargets } from "./relay.js";
 
 export interface StartDaemonOptions {
   readonly clock?: Clock;
@@ -168,6 +168,7 @@ export interface StartDaemonOptions {
   /** why: test-only, replaces the registry of optional modules. */
   readonly optionalModules?: readonly OptionalModuleDefinition<unknown>[];
   readonly relayTargets?: RelayTargets;
+  readonly relayTwoWayTargets?: TwoWayRelayTargets;
   readonly role?: DaemonMode;
   readonly prerequisiteChecks?: readonly PrerequisiteCheck[];
   readonly serviceCheck?: ServiceCheck;
@@ -665,6 +666,10 @@ async function composeDaemon(
       ...moduleHost.relayTargets,
       ...options.relayTargets,
     },
+    relayTwoWayTargets: {
+      ...moduleHost.relayTwoWayTargets,
+      ...options.relayTwoWayTargets,
+    },
     enrollment: wakingQueueAfter(core.enrollment, () => leasing.corePorts.notifyAvailability()),
     nuke,
     registry,
@@ -938,6 +943,7 @@ async function startGatewayDaemon(options: GatewayDaemonOptions): Promise<Daemon
       instanceId,
       leases: leaseIndex,
       relay: (workerId, call) => gatewayRelay.relay(workerId, call),
+      relayTwoWay: (workerId, call) => gatewayRelay.relayTwoWay(workerId, call),
       requesterPrefix: gatewayRequesterPrefix,
       startedAt: clock.now(),
       worker: (id) => gatewayService.workers.view(id),

src/daemon/relay-gateway.e2e.test.ts

@@ -1,10 +1,15 @@
 import { afterEach, describe, expect, it } from "vitest";
 
-import { GatewayRelay, type GatewayWorkerClient, type WorkerDirectory } from "../gateway/index.js";
+import {
+  GatewayRelay,
+  type GatewayWorkerClient,
+  type TwoWayRelay,
+  type WorkerDirectory,
+} from "../gateway/index.js";
 import { linkPair, TYPE } from "../link/testing.js";
 import { connectSimlockClient } from "../simlock-client/client.js";
 import type { DispatchSession } from "./dispatch.js";
-import type { RelayTarget } from "./relay.js";
+import type { RelayTarget, TwoWayRelayTarget } from "./relay.js";
 import type { DaemonServer } from "./server.js";
 import {
   aborted,
@@ -111,6 +116,11 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => {
     leaseId: string | undefined;
   }[] = [];
 
+  let twoWayBehavior: TwoWayRelayTarget = () =>
+    Promise.resolve(new Response(null, { status: 200 }));
+  const twoWayTarget: TwoWayRelayTarget = (request, tenant, leaseId, pipe) =>
+    twoWayBehavior(request, tenant, leaseId, pipe);
+
   const target: RelayTarget = (request, tenant, leaseId) => {
     targetCalls.push({
       headers: request.headers,
@@ -124,17 +134,20 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => {
 
   afterEach(async () => {
     behavior = () => Promise.resolve(new Response("ok"));
+    twoWayBehavior = () => Promise.resolve(new Response(null, { status: 200 }));
     targetCalls.length = 0;
     await cleanUp(daemons, directories);
   });
 
   async function startPair(
     options: { readonly deadlineMs?: number; readonly idleMs?: number } = {},
   ) {
-    const fixture: WorkerFixture = await startWorker(directories, daemons, {
-      "agent-device": target,
-      echo: target,
-    });
+    const fixture: WorkerFixture = await startWorker(
+      directories,
+      daemons,
+      { "agent-device": target, echo: target },
+      { relayTwoWayTargets: { echo: twoWayTarget } },
+    );
     const pair = linkPair();
     const gatewayEnd = pair.gatewayEnd;
     fixture.worker.acceptUplink(pair.worker);
@@ -163,7 +176,10 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => {
   }
 
   const startFleet = (execCommand?: FleetExec) =>
-    startFleet_({ daemons, directories, target }, execCommand === undefined ? {} : { execCommand });
+    startFleet_(
+      { daemons, directories, target, twoWayTarget },
+      execCommand === undefined ? {} : { execCommand },
+    );
 
   const agent = (overrides: Partial<DispatchSession> = {}): DispatchSession => ({
     manageEventSubscription: () => undefined,
@@ -213,6 +229,132 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => {
     expect(taps.flatMap((tap) => tap.binaries(TYPE.OPEN)).length).toBeGreaterThanOrEqual(1);
   }, 30_000);
 
+  it("a gateway module's two-way relay carries bytes both ways at once until either side ends", async () => {
+    const encoder = new TextEncoder();
+    const decoder = new TextDecoder();
+    const { relay, workerId } = await startFleet();
+    const open = async () => {
+      const opened = await relay.relayTwoWay(workerId, buildCall({}));
+      expect(opened).not.toBeInstanceOf(Response);
+      const two = opened as TwoWayRelay;
+      const reader = two.readable.getReader();
+      const readText = async (length: number): Promise<string> => {
+        let seen = "";
+        while (seen.length < length) {
+          const step = await reader.read();
+          if (step.done) break;
+          seen += decoder.decode(step.value);
+        }
+        return seen;
+      };
+      return { reader, readText, two, writer: two.writable.getWriter() };
+    };
+
+    twoWayBehavior = (_request, _tenant, _leaseId, pipe) => {
+      void (async () => {
+        const writer = pipe.writable.getWriter();
+        const reader = pipe.readable.getReader();
+        await writer.write(encoder.encode("hello from worker"));
+        for (;;) {
+          const step = await reader.read();
+          if (step.done) return;
+          await writer.write(step.value);
+        }
+      })().catch(() => undefined);
+      return Promise.resolve(new Response(null, { status: 200 }));
+    };
+    const both = await open();
+    const first = await both.readText("hello from worker".length);
+    await both.writer.write(encoder.encode("ping"));
+    const echoed = await both.readText("ping".length);
+
+    const heard = gate();
+    let heardText = "";
+    let endedSignal: AbortSignal | undefined;
+    twoWayBehavior = (_request, _tenant, _leaseId, pipe) => {
+      endedSignal = pipe.signal;
+      void (async () => {
+        const writer = pipe.writable.getWriter();
+        await writer.write(encoder.encode("bye"));
+        await writer.close();
+        const reader = pipe.readable.getReader();
+        for (;;) {
+          const step = await reader.read();
+          if (step.done) break;
+          heardText += decoder.decode(step.value);
+        }
+        heard.open();
+      })().catch(() => undefined);
+      return Promise.resolve(new Response(null, { status: 200 }));
+    };
+    const workerEnds = await open();
+    const bye = await workerEnds.readText(3);
+    const afterBye = await workerEnds.reader.read();
+    await workerEnds.writer.write(encoder.encode("late"));
+    await workerEnds.writer.close();
+    await heard.promise;
+
+    const more = gate();
+    twoWayBehavior = (_request, _tenant, _leaseId, pipe) => {
+      void (async () => {
+        const writer = pipe.writable.getWriter();
+        await writer.write(encoder.encode("one"));
+        await more.promise;
+        await writer.write(encoder.encode("two"));
+        await writer.close();
+      })().catch(() => undefined);
+      return Promise.resolve(new Response(null, { status: 200 }));
+    };
+    const gatewayEnds = await open();
+    const one = await gatewayEnds.readText(3);
+    await gatewayEnds.writer.close();
+    more.open();
+    const two = await settleWithin(gatewayEnds.readText(3));
+    const afterTwo = await settleWithin(gatewayEnds.reader.read());
+
+    const signals: AbortSignal[] = [];
+    twoWayBehavior = (_request, _tenant, _leaseId, pipe) => {
+      signals.push(pipe.signal);
+      return Promise.resolve(new Response(null, { status: 200 }));
+    };
+    const cancelled = await open();
+    const abortedWrites = await open();
+    await cancelled.reader.cancel();
+    await abortedWrites.writer.abort(new Error("gone"));
+    await poll(() => signals.map((signal) => signal.aborted)).toEqual([true, true]);
+    await both.reader.cancel();
+
+    expect(both.two.status).toBe(200);
+    expect(first).toBe("hello from worker");
+    expect(echoed).toBe("ping");
+    expect(bye).toBe("bye");
+    expect(afterBye.done).toBe(true);
+    expect(heardText).toBe("late");
+    expect(endedSignal?.aborted).toBe(false);
+    expect(one).toBe("one");
+    expect(two).toEqual({ state: "fulfilled", value: "two" });
+    expect(afterTwo).toMatchObject({ state: "fulfilled", value: { done: true } });
+    expect(signals).toHaveLength(2);
+  }, 30_000);
+
+  it("a gateway module opening a two-way relay gets the worker's refusal or an error", async () => {
+    twoWayBehavior = () =>
+      Promise.resolve(new Response("not for you", { headers: { "x-why": "no" }, status: 403 }));
+    const { relay, workerId } = await startFleet();
+
+    const refused = await relay.relayTwoWay(workerId, buildCall({}));
+    const missing = await settleWithin(
+      relay.relayTwoWay(workerId, buildCall({ target: "agent-device" })),
+    );
+
+    expect(refused).toBeInstanceOf(Response);
+    const refusal = refused as Response;
+    expect(refusal.status).toBe(403);
+    expect(refusal.headers.get("x-why")).toBe("no");
+    expect(await refusal.text()).toBe("not for you");
+    expect(missing).toMatchObject({ code: "RELAY_TARGET_MISSING", state: "rejected" });
+  }, 30_000);
+
   it("a 20 MiB PUT body arrives at the target byte for byte, and a 20 MiB response body comes back byte for byte", async () => {
     let received = "";
     behavior = async (request) => {
@@ -629,6 +771,35 @@ describe("a gateway relaying to a worker", { timeout: 15_000 }, () => {
     await poll(() => signal?.aborted).toBe(true);
   }, 30_000);
 
+  it("closing the uplink ends every two-way relay: the gateway side gets WORKER_UNREACHABLE and the target's signal aborts", async () => {
+    const signals: AbortSignal[] = [];
+    twoWayBehavior = (_request, _tenant, _leaseId, pipe) => {
+      signals.push(pipe.signal);
+      return Promise.resolve(new Response(null, { status: 200 }));
+    };
+    const { gatewayEnd, relay } = await startPair();
+    const open = async () => {
+      const opened = (await relay.relayTwoWay(WORKER_ID, buildCall({}))) as TwoWayRelay;
+      return { reader: opened.readable.getReader(), writer: opened.writable.getWriter() };
+    };
+    const first = await open();
+    const second = await open();
+    await poll(() => signals.map((signal) => signal.aborted)).toEqual([false, false]);
+
+    await gatewayEnd.close();
+    const outcomes = await Promise.all([
+      settleWithin(first.reader.read()),
+      settleWithin(first.writer.write(new Uint8Array(1))),
+      settleWithin(second.reader.read()),
+      settleWithin(second.writer.write(new Uint8Array(1))),
+    ]);
+
+    for (const settled of outcomes) {
+      expect(settled).toMatchObject({ code: "WORKER_UNREACHABLE", state: "rejected" });
+    }
+    await poll(() => signals.map((signal) => signal.aborted)).toEqual([true, true]);
+  }, 30_000);
+
   it("a relay to a worker with no uplink fails with WORKER_UNREACHABLE and the target sees nothing", async () => {
     const { relay } = await startPair();