From 3e8d9d6f6b75007195ad03efea581feb03505e83 Mon Sep 17 00:00:00 2001 From: Priya Raman Date: Thu, 20 Aug 2026 03:21:45 +0000 Subject: [PATCH] test(sandbox): add the local real-process duplex end-to-end harness Spawn the generated gateway with plain node from a temporary file, attach the real host broker to the child stdin and stdout, and forward each request to a local fake API server on a real HTTP call. The harness needs no provider credentials and adds no dependency. The harness proves the composed duplex path over real kernel pipes: - The child sends a valid liveness-only READY frame to the broker. - One HTTP request travels through the child and the broker to the API and the response body returns unchanged. - A large response spans many pipe chunks and returns complete. - A multi-byte UTF-8 sequence split across chunk borders decodes correctly. - Concurrent requests through one child each receive the matching response. - A real stdin EOF makes the gateway answer 409 outcome_indeterminate to every outstanding request and 503 bridge_unavailable to every new one. - A real process kill moves the broker to lost, stops all dispatch, and leaves no open pipe or socket handle after teardown. Co-authored-by: Paperclip --- .../src/duplex-bridge-local-e2e.test.ts | 646 ++++++++++++++++++ 1 file changed, 646 insertions(+) create mode 100644 packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts diff --git a/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts new file mode 100644 index 00000000000..a4e2942c974 --- /dev/null +++ b/packages/adapter-utils/src/duplex-bridge-local-e2e.test.ts @@ -0,0 +1,646 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { createServer, type IncomingMessage, type ServerResponse } from "node:http"; +import net from "node:net"; +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { StringDecoder } from "node:string_decoder"; +import { afterEach, describe, expect, it } from "vitest"; + +import { + authorizeSandboxCallbackBridgeRequestWithRoutes, + getSandboxCallbackBridgeServerSource, + SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE, +} from "./sandbox-callback-bridge.js"; +import { + DuplexFrameDecoder, + type DuplexFrame, + type DuplexReadyFrame, + type DuplexRequestFrame, +} from "./duplex-frame-codec.js"; +import { + createDuplexBridgeBroker, + type DuplexBridgeBroker, + type DuplexBrokerForwardResult, +} from "./duplex-bridge-broker.js"; +import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js"; + +/** + * Local real-process end-to-end harness for the composed duplex path. + * + * The harness spawns the real generated gateway with plain `node` from a + * temporary file, attaches the real host broker to the child stdin and stdout, + * and forwards each request to a local fake API server on a real HTTP call. It + * uses no provider credentials. + * + * The child stdio pipes are kernel pipes. They give partial reads, split a + * multi-byte UTF-8 sequence across two chunks, apply backpressure, and give a + * true end of file when the child dies. The harness proves the composed path + * handles each of these real conditions. + */ + +/** One recorded call the fake API server received. */ +interface FakeApiRequest { + method: string; + url: string; + auth: string | null; + runId: string | null; + body: string; +} + +/** The responder the fake API server calls for each request. */ +type FakeApiResponder = (req: IncomingMessage, res: ServerResponse, body: string) => void; + +/** The handle for one fake API server. */ +interface FakeApiServer { + origin: string; + requests: FakeApiRequest[]; + setResponder: (responder: FakeApiResponder) => void; + /** The count of sockets the server holds open right now. */ + openSocketCount: () => number; + /** True while the server still accepts connections. */ + listening: () => boolean; + close: () => Promise; +} + +/** + * Start a local fake API server. The forward handler targets it, so one real + * HTTP call proves the composed path end to end. The server tracks each open + * socket, so a test asserts teardown leaves no open socket handle. + */ +async function startFakeApiServer(): Promise { + const requests: FakeApiRequest[] = []; + const sockets = new Set(); + let responder: FakeApiResponder = (req, res) => { + const url = new URL(req.url ?? "/", "http://127.0.0.1"); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, path: url.pathname })); + }; + + const server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (chunk: Buffer) => chunks.push(chunk)); + req.on("end", () => { + const body = Buffer.concat(chunks).toString("utf8"); + requests.push({ + method: req.method ?? "GET", + url: req.url ?? "/", + auth: req.headers.authorization ?? null, + runId: + typeof req.headers["x-paperclip-run-id"] === "string" + ? req.headers["x-paperclip-run-id"] + : null, + body, + }); + responder(req, res, body); + }); + }); + server.on("connection", (socket) => { + sockets.add(socket); + socket.on("close", () => sockets.delete(socket)); + }); + + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", () => resolve()); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("The fake API server did not expose a TCP port."); + } + + return { + origin: `http://127.0.0.1:${address.port}`, + requests, + setResponder: (next) => { + responder = next; + }, + openSocketCount: () => sockets.size, + listening: () => server.listening, + close: () => + new Promise((resolve) => { + // Destroy each open socket first. A keep-alive client socket keeps the + // server open, so `server.close` alone could stall. The destroy makes the + // close callback fire and drives the open socket count to zero. + for (const socket of sockets) socket.destroy(); + server.close(() => resolve()); + }), + }; +} + +/** The channel view over the spawned child, plus the frames the child sent host-ward. */ +interface ChildDuplexChannel { + channel: CommandManagedDuplexChannel; + observedFrames: DuplexFrame[]; + stderr: () => string; +} + +/** + * Wrap the spawned child as a {@link CommandManagedDuplexChannel}. The broker + * writes to the child stdin, reads the child stdout, and learns of the child + * exit through this channel. + * + * The host end reads a byte stream from the stdout pipe. A pipe read can split + * one multi-byte UTF-8 character across two chunks. The `StringDecoder` holds + * the bytes of an incomplete character until the next chunk, so the broker only + * ever reads whole characters. The channel keeps a second decoder for + * observation only; it lets the harness assert the READY frame and the request + * frames the child produced, and it never feeds the broker. + */ +function attachChildDuplexChannel(child: ChildProcessWithoutNullStreams): ChildDuplexChannel { + const observed = new DuplexFrameDecoder(); + const observedFrames: DuplexFrame[] = []; + const stdoutDecoder = new StringDecoder("utf8"); + let dataListener: ((chunk: string) => void) | null = null; + let exitListener: ((exit: { exitCode: number | null }) => void) | null = null; + let pendingText = ""; + let pendingExit: { exitCode: number | null } | null = null; + let stderrText = ""; + + child.stdout.on("data", (buffer: Buffer) => { + for (const result of observed.push(buffer)) { + if (result.ok) observedFrames.push(result.frame); + } + const text = stdoutDecoder.write(buffer); + if (text.length === 0) return; + if (dataListener) dataListener(text); + else pendingText += text; + }); + child.stderr.on("data", (buffer: Buffer) => { + stderrText += buffer.toString("utf8"); + }); + child.on("exit", (code) => { + const exit = { exitCode: code }; + if (exitListener) exitListener(exit); + else pendingExit = exit; + }); + // Swallow a stdin EPIPE. The broker can write one more frame while the child + // exits; the write fails and the broker records the loss on its own path. + child.stdin.on("error", () => undefined); + + const channel: CommandManagedDuplexChannel = { + write: (data) => { + child.stdin.write(data); + }, + onData: (listener) => { + dataListener = listener; + if (pendingText.length > 0) { + const replay = pendingText; + pendingText = ""; + listener(replay); + } + }, + onExit: (listener) => { + exitListener = listener; + if (pendingExit) { + const exit = pendingExit; + pendingExit = null; + listener(exit); + } + }, + stop: () => { + child.kill("SIGKILL"); + }, + close: () => + new Promise((resolve) => { + // Close the write side. The child stdin reaches end of file, so the + // gateway sees a real EOF. + child.stdin.end(() => resolve()); + }), + }; + + return { channel, observedFrames, stderr: () => stderrText }; +} + +/** The forward-handler mode. `proxy` calls the fake API; `hang` blocks until abort. */ +type ForwardMode = "proxy" | "hang"; + +/** The options for one harness. */ +interface HarnessOptions { + hostApiToken?: string; + runId?: string; + maxBodyBytes?: number; + lossExitGraceMs?: number; +} + +/** The full harness handle for one composed duplex path. */ +interface DuplexE2EHarness { + baseUrl: string; + bridgeToken: string; + hostApiToken: string; + runId: string; + nonce: string; + broker: DuplexBridgeBroker; + api: FakeApiServer; + child: ChildProcessWithoutNullStreams; + observedFrames: DuplexFrame[]; + forwardedRequests: DuplexRequestFrame[]; + setForwardMode: (mode: ForwardMode) => void; + stderr: () => string; + killChild: () => Promise; + waitFor: (predicate: () => boolean, message: string, timeoutMs?: number) => Promise; + teardown: () => Promise; +} + +/** Reserve one free loopback port. The host assigns it to the gateway. */ +async function reserveLoopbackPort(): Promise { + return new Promise((resolve, reject) => { + const probe = net.createServer(); + probe.once("error", reject); + probe.listen(0, "127.0.0.1", () => { + const address = probe.address(); + if (!address || typeof address === "string") { + probe.close(() => reject(new Error("Could not reserve a loopback port."))); + return; + } + const reserved = address.port; + probe.close(() => resolve(reserved)); + }); + }); +} + +/** + * Build the composed duplex path: a spawned gateway child, the real broker on + * the child stdio, and a local fake API server as the forward target. + */ +async function createHarness(options: HarnessOptions = {}): Promise { + const hostApiToken = options.hostApiToken ?? "real-run-jwt"; + const runId = options.runId ?? "run-e2e"; + const maxBodyBytes = options.maxBodyBytes ?? 1_000_000; + const bridgeToken = "duplex-e2e-bridge-token"; + const nonce = "e2e112233445566778899aabbccddeeff"; + let forwardMode: ForwardMode = "proxy"; + const forwardedRequests: DuplexRequestFrame[] = []; + + const api = await startFakeApiServer(); + const assignedPort = await reserveLoopbackPort(); + + const tmpDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-e2e-")); + const entrypoint = path.join(tmpDir, "gateway.mjs"); + await writeFile(entrypoint, getSandboxCallbackBridgeServerSource(), "utf8"); + + const child = spawn(process.execPath, [entrypoint], { + stdio: ["pipe", "pipe", "pipe"], + env: { + ...process.env, + PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE, + PAPERCLIP_BRIDGE_HOST: "127.0.0.1", + PAPERCLIP_BRIDGE_PORT: String(assignedPort), + PAPERCLIP_BRIDGE_NONCE: nonce, + PAPERCLIP_BRIDGE_TOKEN: bridgeToken, + PAPERCLIP_BRIDGE_MAX_BODY_BYTES: String(maxBodyBytes), + ...(options.lossExitGraceMs != null + ? { PAPERCLIP_BRIDGE_LOSS_EXIT_GRACE_MS: String(options.lossExitGraceMs) } + : {}), + }, + }) as ChildProcessWithoutNullStreams; + + const { channel, observedFrames, stderr } = attachChildDuplexChannel(child); + + const forwardRequest = async ( + request: DuplexRequestFrame, + opts: { signal: AbortSignal }, + ): Promise => { + forwardedRequests.push(request); + // Keep the real route allowlist on the forward seam. The broker forwards + // only an allowed route; it denies any other route with a 403. + const denial = authorizeSandboxCallbackBridgeRequestWithRoutes(request); + if (denial) { + return { + status: 403, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: denial }), + }; + } + if (forwardMode === "hang") { + // Block until the broker aborts the forward. This lets a test hold an + // outstanding request open while it forces a loss. + await new Promise((_resolve, reject) => { + const onAbort = () => reject(new Error("The forward call was aborted.")); + if (opts.signal.aborted) { + onAbort(); + return; + } + opts.signal.addEventListener("abort", onAbort, { once: true }); + }); + } + const method = request.method.trim().toUpperCase() || "GET"; + const headers = new Headers(); + for (const [key, value] of Object.entries(request.headers)) { + if (value.trim().length === 0) continue; + headers.set(key, value); + } + // Apply the real host token and the run id, the same as the file bridge path. + headers.set("authorization", `Bearer ${hostApiToken}`); + headers.set("x-paperclip-run-id", runId); + const target = new URL(`${request.path}${request.query ?? ""}`, api.origin); + const response = await fetch(target, { + method, + headers, + ...(method === "GET" || method === "HEAD" ? {} : { body: request.body }), + signal: opts.signal, + }); + const body = await response.text(); + const outHeaders: Record = {}; + response.headers.forEach((value, key) => { + if (key.toLowerCase() === "content-length") return; + outHeaders[key] = value; + }); + return { status: response.status, headers: outHeaders, body }; + }; + + const broker = createDuplexBridgeBroker({ + channel, + forwardRequest, + logger: () => undefined, + }); + broker.start(); + + const waitFor = async ( + predicate: () => boolean, + message: string, + timeoutMs = 5000, + ): Promise => { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (predicate()) return; + if (Date.now() > deadline) { + throw new Error(`Timed out waiting for ${message}. stderr: ${stderr()}`); + } + await new Promise((resolve) => { + const timer = setTimeout(resolve, 20); + timer.unref?.(); + }); + } + }; + + const waitForExit = (): Promise => + new Promise((resolve) => { + if (child.exitCode !== null || child.signalCode !== null) { + resolve(); + return; + } + child.once("exit", () => resolve()); + }); + + const killChild = async (): Promise => { + child.kill("SIGKILL"); + await waitForExit(); + }; + + let torndown = false; + const teardown = async (): Promise => { + if (torndown) return; + torndown = true; + child.kill("SIGKILL"); + await waitForExit(); + child.stdin.destroy(); + child.stdout.destroy(); + child.stderr.destroy(); + await api.close(); + await rm(tmpDir, { recursive: true, force: true }); + }; + + return { + baseUrl: `http://127.0.0.1:${assignedPort}`, + bridgeToken, + hostApiToken, + runId, + nonce, + broker, + api, + child, + observedFrames, + forwardedRequests, + setForwardMode: (mode) => { + forwardMode = mode; + }, + stderr, + killChild, + waitFor, + teardown, + }; +} + +/** + * Build a large body of multi-byte UTF-8 characters. "€" uses three UTF-8 bytes + * and "😀" uses four. A pipe read boundary at a 65536-byte multiple lands inside + * a "€" sequence, because 65536 is not a multiple of three. This guarantees a + * multi-byte character split across two pipe chunks. + */ +function buildMultiByteBody(targetBytes: number): string { + const marker = "😀-start-😀"; + const filler = "€".repeat(Math.ceil(targetBytes / 3)); + return `${marker}${filler}${marker}`; +} + +describe("duplex bridge local end-to-end harness", () => { + const harnesses: DuplexE2EHarness[] = []; + + afterEach(async () => { + while (harnesses.length > 0) { + const harness = harnesses.pop(); + if (harness) await harness.teardown(); + } + }); + + const readyPredicate = (harness: DuplexE2EHarness) => () => + harness.observedFrames.some((frame) => frame.type === "ready"); + + it("delivers a valid READY frame from the spawned gateway child to the attached broker", async () => { + const harness = await createHarness(); + harnesses.push(harness); + + await harness.waitFor(readyPredicate(harness), "a READY frame from the gateway child"); + + const ready = harness.observedFrames.find( + (frame) => frame.type === "ready", + ) as DuplexReadyFrame; + expect(ready.version).toBe(1); + expect(ready.nonce).toBe(harness.nonce); + // READY is liveness only. It carries no address data. + expect((ready as unknown as Record).address).toBeUndefined(); + expect((ready as unknown as Record).port).toBeUndefined(); + // The broker read the READY frame and stayed open with no loss. + expect(harness.broker.state).toBe("open"); + expect(harness.broker.lossRecord).toBeNull(); + }, 20000); + + it("carries one real HTTP request through the child and the broker to the fake API unchanged", async () => { + const harness = await createHarness(); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + harness.api.setResponder((req, res) => { + const url = new URL(req.url ?? "/", "http://127.0.0.1"); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ ok: true, echoedPath: url.pathname })); + }); + + const response = await fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + expect(response.status).toBe(200); + await expect(response.json()).resolves.toEqual({ ok: true, echoedPath: "/api/agents/me" }); + + // The fake API saw one request with the real host token and the run id. The + // broker replaced the bridge token, so the fake API never saw it. + expect(harness.api.requests).toHaveLength(1); + expect(harness.api.requests[0]).toMatchObject({ + method: "GET", + url: "/api/agents/me", + auth: `Bearer ${harness.hostApiToken}`, + runId: harness.runId, + }); + }, 20000); + + it("reassembles a large response that spans many pipe chunks", async () => { + const harness = await createHarness(); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // The body is larger than the pipe buffer, so the response frame crosses the + // child stdin pipe in many chunks. An ASCII body isolates the chunk-span + // behavior from the multi-byte behavior the next test covers. + const largeBody = "x".repeat(256 * 1024); + expect(Buffer.byteLength(largeBody, "utf8")).toBeGreaterThan(200 * 1024); + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "text/plain; charset=utf-8" }); + res.end(largeBody); + }); + + const response = await fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + expect(response.status).toBe(200); + const received = await response.text(); + // The whole body returns complete after it crossed the pipe in many chunks. + expect(received.length).toBe(largeBody.length); + expect(received).toBe(largeBody); + }, 20000); + + it("decodes a multi-byte UTF-8 sequence split across chunk borders", async () => { + const harness = await createHarness(); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + const multiByteBody = buildMultiByteBody(256 * 1024); + expect(Buffer.byteLength(multiByteBody, "utf8")).toBeGreaterThan(200 * 1024); + harness.api.setResponder((_req, res) => { + res.writeHead(200, { "content-type": "text/plain; charset=utf-8" }); + res.end(multiByteBody); + }); + + const response = await fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + expect(response.status).toBe(200); + const received = await response.text(); + // A multi-byte character split across a pipe chunk border decodes to the + // same character, so the whole body returns unchanged. + expect(received.length).toBe(multiByteBody.length); + expect(received).toBe(multiByteBody); + }, 20000); + + it("matches each concurrent request through one child to its own response", async () => { + const harness = await createHarness(); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + harness.api.setResponder((req, res) => { + const url = new URL(req.url ?? "/", "http://127.0.0.1"); + res.writeHead(200, { "content-type": "application/json" }); + res.end(JSON.stringify({ path: url.pathname })); + }); + + const ids = Array.from({ length: 8 }, (_unused, index) => `issue-${index}`); + const responses = await Promise.all( + ids.map((id) => + fetch(`${harness.baseUrl}/api/issues/${id}`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }).then(async (response) => ({ status: response.status, body: await response.json() })), + ), + ); + + responses.forEach((result, index) => { + expect(result.status).toBe(200); + expect(result.body).toEqual({ path: `/api/issues/${ids[index]}` }); + }); + expect(harness.api.requests).toHaveLength(8); + }, 20000); + + it("answers 409 to outstanding and 503 to new requests when the broker closes its write side", async () => { + const harness = await createHarness({ lossExitGraceMs: 3000 }); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // Hold the forward open, so the broker sends no response frame. The gateway + // keeps the request outstanding until the loss. + harness.setForwardMode("hang"); + const outstanding = fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + void outstanding.catch(() => undefined); + await harness.waitFor( + () => harness.forwardedRequests.length >= 1, + "the broker to receive the forwarded request", + ); + + // The broker closes its write side, so the child stdin reaches a real EOF. + await harness.broker.close(); + + const lossResponse = await outstanding; + expect(lossResponse.status).toBe(409); + expect(lossResponse.headers.get("x-paperclip-bridge-outcome")).toBe("indeterminate"); + await expect(lossResponse.json()).resolves.toEqual({ error: "outcome_indeterminate" }); + + const afterLoss = await fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + expect(afterLoss.status).toBe(503); + await expect(afterLoss.json()).resolves.toEqual({ error: "bridge_unavailable" }); + }, 20000); + + it("moves the broker to lost on a child kill, dispatches nothing more, and leaks no handle after teardown", async () => { + const harness = await createHarness(); + harnesses.push(harness); + await harness.waitFor(readyPredicate(harness), "the gateway to become ready"); + + // Hold one forward open, so a request is in flight when the child dies. + harness.setForwardMode("hang"); + const outstanding = fetch(`${harness.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${harness.bridgeToken}` }, + }); + void outstanding.catch(() => undefined); + await harness.waitFor( + () => harness.forwardedRequests.length >= 1, + "the broker to receive the forwarded request", + ); + const forwardedBeforeKill = harness.forwardedRequests.length; + + // A real process kill closes the stdout pipe and ends the child. + await harness.killChild(); + await harness.waitFor( + () => harness.broker.state === "lost", + "the broker to record the channel loss", + ); + expect(harness.broker.lossRecord?.reason).toBe("channel_exit"); + + // The broker dispatches nothing more after the loss. + await new Promise((resolve) => { + const timer = setTimeout(resolve, 100); + timer.unref?.(); + }); + expect(harness.forwardedRequests.length).toBe(forwardedBeforeKill); + // The outstanding fetch fails because the child died before it answered. + await expect(outstanding).rejects.toThrow(); + + await harness.teardown(); + // Teardown left no open pipe or socket handle. + expect(harness.child.stdin.destroyed).toBe(true); + expect(harness.child.stdout.destroyed).toBe(true); + expect(harness.child.stderr.destroyed).toBe(true); + expect(harness.api.listening()).toBe(false); + expect(harness.api.openSocketCount()).toBe(0); + }, 20000); +});