From faab2620ad24a8fa92d71261df8c0afe77dfd44d Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 19 Aug 2026 17:14:47 -0700 Subject: [PATCH] feat(sandbox): add a duplex transport for Daytona behind a default-off kill switch (#11750) ## Thinking Path > - Paperclip is an open source app that manages AI agents for work > - Paperclip runs agents in local and remote sandbox environments > - A sandbox needs a bounded channel for commands and asynchronous input > - Daytona needs a real pseudo-terminal transport for this channel > - The sandbox gateway also needs a mode that handles channel loss safely > - This pull request adds the Daytona transport and gateway mode behind a default-off kill switch > - The benefit is a tested foundation for later transport selection ## Linked Issues or Issue Description **Subsystem affected** Cross-cutting (multiple of the above): sandbox providers, plugin SDK, server settings, and shared types. **Problem or motivation** The merged sandbox protocol has no runtime transport for Daytona. The generated sandbox gateway also has no duplex mode. A later transport-selection change needs both parts and a safe per-run gate. **Proposed solution** Add a Daytona `duplexCommandStream` transport over a raw pseudo-terminal. Add a generated gateway mode named `duplex_v1`. Add the `enableSandboxDuplexBridge` setting with a default value of `false`. Keep transport selection disabled until a later pull request. **Alternatives considered** Keep the protocol unused until the transport-selection change. This would delay provider tests and leave the gateway path without direct coverage. **Roadmap alignment** This change supports the completed Roadmap item for cloud and sandbox agents. It extends the merged sandbox channel foundation in pull request #11738. **Additional context** The Daytona provider remains an untrusted boundary. Deployments must use least-privilege provider credentials and provider-side quota controls. Operators must name an owner for duplex telemetry retention before rollout. ## What Changed - Add the Daytona `duplexCommandStream` capability over a raw pseudo-terminal. - Add a launch wrapper that disables echo and newline translation for NDJSON frames. - Close channels on lease release, destroy, resume of a stopped worker, and worker shutdown. - Declare the capability in the Daytona manifest and set `PLUGIN_VERSION` to `0.1.5`. - Add the worker-to-host notification sink at `ctx.duplexChannel.data` and `ctx.duplexChannel.exit`. - Add the generated sandbox gateway mode `PAPERCLIP_API_BRIDGE_MODE=duplex_v1`. - Add channel-loss results of `409 outcome_indeterminate` and `503 bridge_unavailable`. - Add the per-run setting `enableSandboxDuplexBridge`, with a default value of `false`. - Add unit tests, generated-source codec tests, lifecycle tests, and a credential-gated live Daytona test. ## Verification - Daytona suite: 185 tests pass. - Adapter utilities: 754 tests pass and 4 tests skip. - Plugin SDK: 62 tests pass. - Shared package: 28 tests pass. - Server duplex tests pass. - Shared, plugin SDK, server, and Daytona TypeScript checks pass. - The live Daytona test passes 3 cases when `DAYTONA_API_KEY` is set. - The live Daytona test skips 3 cases without `DAYTONA_API_KEY`. - CI must run the full workspace typecheck, test, and build gates after PR creation. ## Risks - The Daytona control plane and pseudo-terminal remain untrusted boundaries. - The duplex gateway changes behavior only when the mode and per-run setting enable it. - A lost channel fails requests without replay, so callers must handle indeterminate outcomes. - The transport-selection change must require both `duplexCommandStream === true` and `enableSandboxDuplexBridge === true`. - The provider credential and quota limits need operator control before rollout. ## Model Used OpenAI Codex, GPT-5, tool use and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs: #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../src/command-managed-runtime.ts | 9 +- .../src/execution-target-sandbox.test.ts | 527 ++++++++++++++ .../src/sandbox-callback-bridge.ts | 656 +++++++++++++++--- .../src/duplex-command-stream.live.test.ts | 194 ++++++ .../daytona/src/duplex-command-stream.test.ts | 416 +++++++++++ .../daytona/src/duplex-command-stream.ts | 229 ++++++ .../sandbox-providers/daytona/src/manifest.ts | 9 +- .../daytona/src/plugin.test.ts | 133 ++++ .../sandbox-providers/daytona/src/plugin.ts | 141 +++- .../sdk/src/protocol.duplex-channel.test.ts | 10 +- packages/plugins/sdk/src/protocol.ts | 8 +- packages/plugins/sdk/src/testing.ts | 8 + packages/plugins/sdk/src/types.ts | 34 + packages/plugins/sdk/src/worker-rpc-host.ts | 26 + .../plugins/sdk/tests/worker-rpc-host.test.ts | 4 +- packages/shared/src/feature-catalog.ts | 8 + packages/shared/src/types/instance.ts | 7 + .../shared/src/validators/instance.test.ts | 23 + packages/shared/src/validators/instance.ts | 4 + ...onment-runtime-duplex-bridge-input.test.ts | 69 ++ .../instance-settings-service.test.ts | 1 + server/src/services/environment-runtime.ts | 44 +- server/src/services/instance-settings.ts | 2 + server/src/services/plugin-worker-manager.ts | 7 +- .../InstanceExperimentalSettings.test.tsx | 1 + 25 files changed, 2437 insertions(+), 133 deletions(-) create mode 100644 packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.live.test.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts create mode 100644 server/src/__tests__/environment-runtime-duplex-bridge-input.test.ts diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 22a458b135..4c879dbf4d 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -19,13 +19,14 @@ import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress. import type { RuntimeSpanRunner } from "./acpx-engine/startup-timing.js"; /** - * Input for a duplex channel open. The caller supplies only the command line - * the sandbox runs as the channel child process. The runner adds the lease - * scope from its own closure. This type is separate from the worker manager's + * Input for a duplex channel open. The caller supplies only the command argument + * vector the sandbox runs as the channel child process. Element 0 is the program + * and the rest are its arguments. The runner adds the lease scope from its own + * closure. This type is separate from the worker manager's * `DuplexChannelOpenInput`, which also carries the lease scope fields. */ export interface DuplexChannelOpenInput { - command: string; + command: readonly string[]; } /** diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 3e41253f53..75756182a5 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/packages/adapter-utils/src/execution-target-sandbox.test.ts @@ -4,9 +4,15 @@ import { execFile, spawn } from "node:child_process"; import { mkdir, mkdtemp, readdir, readFile, rm, writeFile } from "node:fs/promises"; import os from "node:os"; import path from "node:path"; +import { fileURLToPath } from "node:url"; import { promisify } from "node:util"; import { afterEach, describe, expect, it, vi } from "vitest"; +import { + getSandboxCallbackBridgeServerSource, + getSandboxDuplexGatewayCodecSource, +} from "./sandbox-callback-bridge.js"; + import { DEFAULT_REMOTE_SANDBOX_ADAPTER_TIMEOUT_SEC, adapterExecutionTargetSessionIdentity, @@ -2748,3 +2754,524 @@ describe("sandbox adapter execution targets", () => { } }); }); + +// One decoded stdout frame from the generated duplex gateway. The gateway writes +// newline-delimited JSON frames to stdout, so the test parses each line. +interface DecodedGatewayFrame { + version?: number; + type?: string; + id?: string; + method?: string; + path?: string; + query?: string; + headers?: Record; + body?: string; + address?: string; + __unparsed?: string; +} + +// One decode result from the embedded codec. The shape mirrors the host codec: +// a valid frame or a protocol error with a code. +interface EmbeddedDecodeResult { + ok: boolean; + frame?: unknown; + error?: { code: string; message: string }; +} + +// The names the embedded codec source declares. A test wraps the source and +// reads these names back. +interface EmbeddedCodec { + encodeDuplexFrame: (frame: unknown) => string; + decodeDuplexLine: (line: string | Buffer) => EmbeddedDecodeResult; + DuplexFrameDecoder: new (options?: { maxFrameBytes?: number }) => { + push: (chunk: Buffer) => EmbeddedDecodeResult[]; + }; + DUPLEX_FRAME_VERSION: number; + DEFAULT_MAX_DUPLEX_FRAME_BYTES: number; +} + +type ExpectedVectorResult = { frame: unknown } | { error: string }; + +interface DuplexFrameVector { + name: string; + category: string; + bytes: string; + splitByteOffsets?: number[]; + maxFrameBytes?: number; + roundTrip?: boolean; + expected: ExpectedVectorResult[]; +} + +interface DuplexFrameFixture { + frameVersion: number; + defaultMaxFrameBytes: number; + vectors: DuplexFrameVector[]; +} + +describe("sandbox duplex gateway", () => { + const duplexCleanupDirs: string[] = []; + const duplexChildren: Array> = []; + + afterEach(async () => { + while (duplexChildren.length > 0) { + const child = duplexChildren.pop(); + if (!child) continue; + child.kill("SIGKILL"); + } + while (duplexCleanupDirs.length > 0) { + const dir = duplexCleanupDirs.pop(); + if (!dir) continue; + await rm(dir, { recursive: true, force: true }).catch(() => undefined); + } + }); + + interface DuplexGatewayHandle { + baseUrl: string; + frames: DecodedGatewayFrame[]; + stderr: () => string; + waitForFrame: ( + predicate: (frame: DecodedGatewayFrame) => boolean, + timeoutMs?: number, + ) => Promise; + sendFrame: (frame: Record) => void; + sendRaw: (text: string) => void; + endStdin: () => void; + exited: Promise; + stop: () => Promise; + } + + // Start the generated gateway `.mjs` in duplex mode as a real child process. + // The test writes response frames to the child stdin and reads request frames + // from the child stdout, so it stands in for the host side of the channel. + async function startDuplexGateway(env: Record): Promise { + const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-duplex-gateway-")); + duplexCleanupDirs.push(rootDir); + const entrypoint = path.join(rootDir, "gateway.mjs"); + await writeFile(entrypoint, getSandboxCallbackBridgeServerSource(), "utf8"); + + const child = spawn(process.execPath, [entrypoint], { + stdio: ["pipe", "pipe", "pipe"], + env: { + ...process.env, + PAPERCLIP_API_BRIDGE_MODE: "duplex_v1", + PAPERCLIP_BRIDGE_HOST: "127.0.0.1", + PAPERCLIP_BRIDGE_PORT: "0", + ...env, + }, + }); + duplexChildren.push(child); + + const frames: DecodedGatewayFrame[] = []; + const waiters: Array<{ + predicate: (frame: DecodedGatewayFrame) => boolean; + resolve: (frame: DecodedGatewayFrame) => void; + }> = []; + let stdoutBuffer = ""; + let stderrText = ""; + + child.stdout?.setEncoding("utf8"); + child.stdout?.on("data", (chunk: string) => { + stdoutBuffer += chunk; + let newlineIndex = stdoutBuffer.indexOf("\n"); + while (newlineIndex !== -1) { + const line = stdoutBuffer.slice(0, newlineIndex); + stdoutBuffer = stdoutBuffer.slice(newlineIndex + 1); + if (line.length > 0) { + let frame: DecodedGatewayFrame; + try { + frame = JSON.parse(line) as DecodedGatewayFrame; + } catch { + frame = { __unparsed: line }; + } + frames.push(frame); + for (const waiter of [...waiters]) { + if (waiter.predicate(frame)) { + waiters.splice(waiters.indexOf(waiter), 1); + waiter.resolve(frame); + } + } + } + newlineIndex = stdoutBuffer.indexOf("\n"); + } + }); + child.stderr?.setEncoding("utf8"); + child.stderr?.on("data", (chunk: string) => { + stderrText += chunk; + }); + + const exited = new Promise((resolve) => { + child.on("exit", (code) => resolve(code)); + }); + + const waitForFrame = ( + predicate: (frame: DecodedGatewayFrame) => boolean, + timeoutMs = 5000, + ): Promise => { + const existing = frames.find(predicate); + if (existing) return Promise.resolve(existing); + return new Promise((resolve, reject) => { + const waiter = { + predicate, + resolve: (frame: DecodedGatewayFrame) => { + clearTimeout(timer); + resolve(frame); + }, + }; + const timer = setTimeout(() => { + const index = waiters.indexOf(waiter); + if (index !== -1) waiters.splice(index, 1); + reject(new Error(`Timed out waiting for a gateway frame. stderr: ${stderrText}`)); + }, timeoutMs); + waiters.push(waiter); + }); + }; + + const handle: DuplexGatewayHandle = { + baseUrl: "", + frames, + stderr: () => stderrText, + waitForFrame, + sendFrame: (frame) => { + child.stdin?.write(`${JSON.stringify(frame)}\n`); + }, + sendRaw: (text) => { + child.stdin?.write(text); + }, + endStdin: () => { + child.stdin?.end(); + }, + exited, + stop: async () => { + child.kill("SIGKILL"); + await exited.catch(() => null); + }, + }; + + const ready = await waitForFrame((frame) => frame.type === "ready"); + handle.baseUrl = String(ready.address); + return handle; + } + + it("embedded gateway codec passes every vector in the shared fixture", async () => { + const codecFactory = new Function( + `${getSandboxDuplexGatewayCodecSource()}\nreturn { encodeDuplexFrame, decodeDuplexLine, DuplexFrameDecoder, DUPLEX_FRAME_VERSION, DEFAULT_MAX_DUPLEX_FRAME_BYTES };`, + ) as unknown as () => EmbeddedCodec; + const codec = codecFactory(); + + const fixturePath = fileURLToPath(new URL("./duplex-frame-vectors.json", import.meta.url)); + const fixture = JSON.parse(await readFile(fixturePath, "utf8")) as DuplexFrameFixture; + + expect(fixture.frameVersion).toBe(codec.DUPLEX_FRAME_VERSION); + expect(fixture.defaultMaxFrameBytes).toBe(codec.DEFAULT_MAX_DUPLEX_FRAME_BYTES); + expect(fixture.vectors.length).toBeGreaterThanOrEqual(22); + + const failures: string[] = []; + for (const vector of fixture.vectors) { + const decoder = new codec.DuplexFrameDecoder( + vector.maxFrameBytes ? { maxFrameBytes: vector.maxFrameBytes } : undefined, + ); + const buffer = Buffer.from(vector.bytes, "utf8"); + const offsets = vector.splitByteOffsets; + const bounds = + offsets && offsets.length > 0 ? [0, ...offsets, buffer.length] : [0, buffer.length]; + const results: EmbeddedDecodeResult[] = []; + for (let index = 0; index < bounds.length - 1; index += 1) { + results.push(...decoder.push(buffer.subarray(bounds[index], bounds[index + 1]))); + } + + if (results.length !== vector.expected.length) { + failures.push(`${vector.name}: got ${results.length} results, want ${vector.expected.length}`); + continue; + } + vector.expected.forEach((want, index) => { + const got = results[index]; + if ("frame" in want) { + if (!got.ok) { + failures.push(`${vector.name}[${index}]: expected a frame, got an error`); + return; + } + try { + expect(got.frame).toEqual(want.frame); + } catch { + failures.push(`${vector.name}[${index}]: frame does not match`); + } + } else { + if (got.ok) { + failures.push(`${vector.name}[${index}]: expected an error, got a frame`); + return; + } + if (got.error?.code !== want.error) { + failures.push(`${vector.name}[${index}]: error ${got.error?.code} != ${want.error}`); + } + } + }); + } + expect(failures).toEqual([]); + + // The encode side stays wire compatible too: one line, one newline, and the + // same frame after a decode round trip. + for (const vector of fixture.vectors.filter((entry) => entry.roundTrip)) { + const want = vector.expected[0]; + if (!("frame" in want)) continue; + const encoded = codec.encodeDuplexFrame(want.frame); + expect(encoded.endsWith("\n")).toBe(true); + expect(encoded.slice(0, -1)).not.toContain("\n"); + const decoded = codec.decodeDuplexLine(encoded.slice(0, -1)); + expect(decoded.ok).toBe(true); + expect(decoded.frame).toEqual(want.frame); + } + }); + + it("returns the same HTTP response as the file gateway for a forwarded request", async () => { + const token = "duplex-token-forward"; + const gateway = await startDuplexGateway({ PAPERCLIP_BRIDGE_TOKEN: token }); + + const responsePromise = fetch(`${gateway.baseUrl}/api/agents/me?view=compact`, { + headers: { + authorization: `Bearer ${token}`, + accept: "application/json", + "if-none-match": '"cache-key"', + "x-bridge-debug": "drop-me", + }, + }); + const requestFrame = await gateway.waitForFrame((frame) => frame.type === "request"); + expect(requestFrame.method).toBe("GET"); + expect(requestFrame.path).toBe("/api/agents/me"); + expect(requestFrame.query).toBe("?view=compact"); + // Only allowlisted headers forward; the bearer and the debug header drop. + expect(requestFrame.headers).toEqual({ + accept: "application/json", + "if-none-match": '"cache-key"', + }); + + gateway.sendFrame({ + version: 1, + type: "response", + id: requestFrame.id, + status: 200, + headers: { "content-type": "application/json", etag: '"rev-1"', "content-length": "999" }, + body: JSON.stringify({ ok: true }), + outcome: "completed", + }); + const response = await responsePromise; + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toContain("application/json"); + expect(response.headers.get("etag")).toBe('"rev-1"'); + await expect(response.json()).resolves.toEqual({ ok: true }); + + // An indeterminate outcome maps to a non-retryable 409, the same contract the + // file gateway applies through the outcome header. + const indeterminatePromise = fetch(`${gateway.baseUrl}/api/issues/issue-1`, { + method: "PATCH", + headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, + body: JSON.stringify({ status: "in_progress" }), + }); + const patchFrame = await gateway.waitForFrame( + (frame) => frame.type === "request" && frame.id !== requestFrame.id, + ); + gateway.sendFrame({ + version: 1, + type: "response", + id: patchFrame.id, + status: 200, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: "outcome_indeterminate" }), + outcome: "indeterminate", + }); + const indeterminate = await indeterminatePromise; + expect(indeterminate.status).toBe(409); + + await gateway.stop(); + }, 20000); + + it("enforces the bearer check, JSON-only rule, body limit, and depth limit", async () => { + const token = "duplex-token-contract"; + const gateway = await startDuplexGateway({ + PAPERCLIP_BRIDGE_TOKEN: token, + PAPERCLIP_BRIDGE_MAX_QUEUE_DEPTH: "1", + PAPERCLIP_BRIDGE_MAX_BODY_BYTES: "16", + }); + + const badAuth = await fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: "Bearer wrong-token" }, + }); + expect(badAuth.status).toBe(401); + await expect(badAuth.json()).resolves.toEqual({ error: "Invalid bridge token." }); + + const nonJson = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { authorization: `Bearer ${token}`, "content-type": "text/plain" }, + body: "not json", + }); + expect(nonJson.status).toBe(415); + await expect(nonJson.json()).resolves.toEqual({ + error: "Bridge only accepts JSON request bodies.", + }); + + const oversizeBody = await fetch(`${gateway.baseUrl}/api/issues/issue-1/comments`, { + method: "POST", + headers: { authorization: `Bearer ${token}`, "content-type": "application/json" }, + body: JSON.stringify({ body: "x".repeat(64) }), + }); + expect(oversizeBody.status).toBe(502); + await expect(oversizeBody.json()).resolves.toEqual({ + error: "Bridge request body exceeded the configured size limit.", + }); + + // No request frame forwarded so far: the guards rejected before forwarding. + const framesBeforeDepth = gateway.frames.filter((frame) => frame.type === "request").length; + expect(framesBeforeDepth).toBe(0); + + // One outstanding request fills the single depth slot; the host never + // responds, so the slot stays used. + const outstanding = fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + void outstanding.catch(() => undefined); + await gateway.waitForFrame((frame) => frame.type === "request"); + + const queueFull = await fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + expect(queueFull.status).toBe(503); + await expect(queueFull.json()).resolves.toEqual({ error: "Bridge request queue is full." }); + + // Only the outstanding request forwarded a frame; the queue-full request did + // not. + const framesAfterDepth = gateway.frames.filter((frame) => frame.type === "request").length; + expect(framesAfterDepth).toBe(1); + + await gateway.stop(); + }, 20000); + + it("answers outstanding requests 409 and new requests 503 on stdin EOF, then exits", async () => { + const token = "duplex-token-eof"; + const gateway = await startDuplexGateway({ + PAPERCLIP_BRIDGE_TOKEN: token, + PAPERCLIP_BRIDGE_LOSS_EXIT_GRACE_MS: "3000", + }); + + const outstanding = fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + await gateway.waitForFrame((frame) => frame.type === "request"); + gateway.endStdin(); + + 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(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + expect(afterLoss.status).toBe(503); + await expect(afterLoss.json()).resolves.toEqual({ error: "bridge_unavailable" }); + + const exitCode = await gateway.exited; + expect(exitCode).toBe(0); + }, 20000); + + it("applies the same loss behavior on a heartbeat timeout", async () => { + const token = "duplex-token-heartbeat"; + const gateway = await startDuplexGateway({ + PAPERCLIP_BRIDGE_TOKEN: token, + PAPERCLIP_BRIDGE_HEARTBEAT_TIMEOUT_MS: "400", + PAPERCLIP_BRIDGE_LOSS_EXIT_GRACE_MS: "3000", + PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS: "30000", + }); + + // Never send an inbound frame: inbound silence trips the heartbeat timeout. + const outstanding = fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + await gateway.waitForFrame((frame) => frame.type === "request"); + + const lossResponse = await outstanding; + expect(lossResponse.status).toBe(409); + await expect(lossResponse.json()).resolves.toEqual({ error: "outcome_indeterminate" }); + + const afterLoss = await fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + expect(afterLoss.status).toBe(503); + await expect(afterLoss.json()).resolves.toEqual({ error: "bridge_unavailable" }); + + const exitCode = await gateway.exited; + expect(exitCode).toBe(0); + }, 20000); + + it("keeps diagnostics off stdout; stdout carries only frames", async () => { + const token = "duplex-token-stdout"; + const gateway = await startDuplexGateway({ PAPERCLIP_BRIDGE_TOKEN: token }); + + const responsePromise = fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + const requestFrame = await gateway.waitForFrame((frame) => frame.type === "request"); + + // Feed a malformed inbound line and an unknown response id. Both force a + // diagnostic path; none of it may reach stdout. + gateway.sendRaw("this is not a frame\n"); + gateway.sendFrame({ + version: 1, + type: "response", + id: "unknown-id", + status: 200, + headers: {}, + body: "", + outcome: "completed", + }); + gateway.sendFrame({ + version: 1, + type: "response", + id: requestFrame.id, + status: 200, + headers: { "content-type": "application/json" }, + body: "{}", + outcome: "completed", + }); + + const response = await responsePromise; + expect(response.status).toBe(200); + + // Every stdout line parsed as a frame; none was diagnostic text. + expect(gateway.frames.some((frame) => frame.__unparsed !== undefined)).toBe(false); + expect( + gateway.frames.every( + (frame) => typeof frame.version === "number" && typeof frame.type === "string", + ), + ).toBe(true); + const allowedTypes = new Set(["ready", "heartbeat", "request"]); + expect(gateway.frames.every((frame) => allowedTypes.has(String(frame.type)))).toBe(true); + + // The malformed inbound line produced a stderr diagnostic, not a stdout one. + expect(gateway.stderr()).toContain("dropped an inbound frame"); + + await gateway.stop(); + }, 20000); + + it("defaults the wait budget to 35 s and honors the environment key override", async () => { + // The generated source carries the 35 s default. + expect(getSandboxCallbackBridgeServerSource()).toContain("35000"); + + const token = "duplex-token-budget"; + const gateway = await startDuplexGateway({ + PAPERCLIP_BRIDGE_TOKEN: token, + PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS: "150", + PAPERCLIP_BRIDGE_HEARTBEAT_TIMEOUT_MS: "30000", + }); + + const started = Date.now(); + const response = await fetch(`${gateway.baseUrl}/api/agents/me`, { + headers: { authorization: `Bearer ${token}` }, + }); + expect(response.status).toBe(502); + await expect(response.json()).resolves.toEqual({ + error: "Timed out waiting for host bridge response.", + }); + expect(Date.now() - started).toBeLessThan(4000); + + await gateway.stop(); + }, 20000); +}); diff --git a/packages/adapter-utils/src/sandbox-callback-bridge.ts b/packages/adapter-utils/src/sandbox-callback-bridge.ts index 25db401931..2c28dfe635 100644 --- a/packages/adapter-utils/src/sandbox-callback-bridge.ts +++ b/packages/adapter-utils/src/sandbox-callback-bridge.ts @@ -55,6 +55,28 @@ const SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT = "paperclip-bridge-server.mjs"; const SANDBOX_EXEC_CHANNEL_ENV = "PAPERCLIP_SANDBOX_EXEC_CHANNEL"; const SANDBOX_EXEC_CHANNEL_BRIDGE = "bridge"; +// The two bridge modes the generated gateway supports. The file mode polls a +// request/response queue on disk. The duplex mode forwards one request frame to +// stdout and resolves one response frame from stdin. The generated `.mjs` +// selects the mode from `PAPERCLIP_API_BRIDGE_MODE`. +const SANDBOX_CALLBACK_BRIDGE_FILE_MODE = "queue_v1"; +const SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE = "duplex_v1"; + +// The duplex gateway HTTP wait budget default. The gateway waits this long for a +// response frame before it answers the local caller with a 502 timeout. The +// value stays configurable through `PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS`, the +// same environment key the file mode reads. +const DEFAULT_DUPLEX_GATEWAY_WAIT_BUDGET_MS = 35_000; +// How often the duplex gateway writes a heartbeat frame to stdout. +const DEFAULT_DUPLEX_GATEWAY_HEARTBEAT_INTERVAL_MS = 5_000; +// The duplex gateway treats the channel as lost when no inbound frame arrives on +// stdin within this window. +const DEFAULT_DUPLEX_GATEWAY_HEARTBEAT_TIMEOUT_MS = 20_000; +// After a loss, the duplex gateway answers each new local request with a 503, +// then exits when this grace ends. The grace gives the local caller time to read +// the 409 for an outstanding request and a 503 for a new request. +const DEFAULT_DUPLEX_GATEWAY_LOSS_EXIT_GRACE_MS = 1_000; + /** Span name that wraps one Paperclip-API callback request — read the request, * write the response, and remove the request file. */ const CALLBACK_BRIDGE_RELAY_REQUEST_SPAN = "sandbox.callbackBridge.relayRequest"; @@ -413,7 +435,7 @@ export function buildSandboxCallbackBridgeEnv(input: { maxBodyBytes?: number | null; }): Record { return { - PAPERCLIP_API_BRIDGE_MODE: "queue_v1", + PAPERCLIP_API_BRIDGE_MODE: SANDBOX_CALLBACK_BRIDGE_FILE_MODE, PAPERCLIP_BRIDGE_QUEUE_DIR: input.queueDir, PAPERCLIP_BRIDGE_TOKEN: input.bridgeToken, PAPERCLIP_BRIDGE_HOST: input.host?.trim() || "127.0.0.1", @@ -1710,30 +1732,230 @@ export async function startSandboxCallbackBridgeServer(input: { }; } +/** + * The zero-dependency codec the generated duplex gateway embeds. It is a plain + * JavaScript copy of the host codec in `duplex-frame-codec.ts`. It uses only the + * `Buffer`, `JSON`, `Object`, `Set`, and `Array` globals, so the generated + * `.mjs` needs no workspace import. It carries no template literal and no `${` + * sequence, so it embeds inside the gateway template literal with no escape. + * + * A shared fixture file (`duplex-frame-vectors.json`) proves this copy stays wire + * compatible with the host codec. {@link getSandboxDuplexGatewayCodecSource} + * returns this exact source so a test can run every fixture vector against it. + */ +const DUPLEX_GATEWAY_CODEC_SOURCE = `const DUPLEX_FRAME_VERSION = 1; +const DEFAULT_MAX_DUPLEX_FRAME_BYTES = 1000000; +const DUPLEX_NEWLINE_BYTE = 0x0a; +const DUPLEX_EMPTY = Buffer.alloc(0); +const DUPLEX_RESPONSE_OUTCOMES = new Set(["completed", "indeterminate", "unavailable"]); + +function duplexOk(frame) { + return { ok: true, frame: frame }; +} + +function duplexFail(code, message) { + return { ok: false, error: { code: code, message: message } }; +} + +function duplexIsPlainObject(value) { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function duplexIsStringRecord(value) { + if (!duplexIsPlainObject(value)) return false; + for (const entry of Object.values(value)) { + if (typeof entry !== "string") return false; + } + return true; +} + +function encodeDuplexFrame(frame) { + return JSON.stringify(frame) + "\\n"; +} + +function decodeDuplexLine(line) { + const text = typeof line === "string" ? line : line.toString("utf8"); + let parsed; + try { + parsed = JSON.parse(text); + } catch (error) { + return duplexFail("malformed_frame", "frame is not valid JSON"); + } + if (!duplexIsPlainObject(parsed)) { + return duplexFail("malformed_frame", "frame is not a JSON object"); + } + if (parsed.version !== DUPLEX_FRAME_VERSION) { + return duplexFail("version_mismatch", "frame version " + String(parsed.version) + " is not " + DUPLEX_FRAME_VERSION); + } + return duplexValidateFrame(parsed); +} + +function duplexValidateFrame(frame) { + switch (frame.type) { + case "request": + return duplexValidateRequest(frame); + case "response": + return duplexValidateResponse(frame); + case "ready": + return duplexValidateReady(frame); + case "heartbeat": + return duplexOk(frame); + case "close": + return duplexOk(frame); + case "error": + return duplexValidateError(frame); + default: + return duplexFail("unknown_type", "unknown frame type " + JSON.stringify(frame.type)); + } +} + +function duplexValidateRequest(frame) { + if ( + typeof frame.id !== "string" || + typeof frame.method !== "string" || + typeof frame.path !== "string" || + typeof frame.query !== "string" || + typeof frame.body !== "string" || + !duplexIsStringRecord(frame.headers) + ) { + return duplexFail("malformed_frame", "request frame has a missing or wrong-typed field"); + } + return duplexOk(frame); +} + +function duplexValidateResponse(frame) { + if ( + typeof frame.id !== "string" || + typeof frame.status !== "number" || + typeof frame.body !== "string" || + !duplexIsStringRecord(frame.headers) || + typeof frame.outcome !== "string" || + !DUPLEX_RESPONSE_OUTCOMES.has(frame.outcome) + ) { + return duplexFail("malformed_frame", "response frame has a missing or wrong-typed field"); + } + return duplexOk(frame); +} + +function duplexValidateReady(frame) { + if (typeof frame.address !== "string") { + return duplexFail("malformed_frame", "ready frame has a missing or wrong-typed address"); + } + return duplexOk(frame); +} + +function duplexValidateError(frame) { + if (typeof frame.code !== "string") { + return duplexFail("malformed_frame", "error frame has a missing or wrong-typed code"); + } + if (frame.message !== undefined && typeof frame.message !== "string") { + return duplexFail("malformed_frame", "error frame has a wrong-typed message"); + } + return duplexOk(frame); +} + +class DuplexFrameDecoder { + constructor(options) { + this.buffer = DUPLEX_EMPTY; + this.discarding = false; + this.maxFrameBytes = + options && options.maxFrameBytes != null ? options.maxFrameBytes : DEFAULT_MAX_DUPLEX_FRAME_BYTES; + } + + push(chunk) { + const incoming = typeof chunk === "string" ? Buffer.from(chunk, "utf8") : Buffer.from(chunk); + this.buffer = this.buffer.length === 0 ? incoming : Buffer.concat([this.buffer, incoming]); + const results = []; + for (;;) { + if (this.discarding) { + const newlineIndex = this.buffer.indexOf(DUPLEX_NEWLINE_BYTE); + if (newlineIndex === -1) { + this.buffer = DUPLEX_EMPTY; + break; + } + this.buffer = this.buffer.subarray(newlineIndex + 1); + this.discarding = false; + continue; + } + const newlineIndex = this.buffer.indexOf(DUPLEX_NEWLINE_BYTE); + if (newlineIndex === -1) { + if (this.buffer.length > this.maxFrameBytes) { + results.push(duplexFail("frame_too_large", "frame exceeds the maximum size")); + this.discarding = true; + this.buffer = DUPLEX_EMPTY; + } + break; + } + const line = this.buffer.subarray(0, newlineIndex); + this.buffer = this.buffer.subarray(newlineIndex + 1); + if (line.length === 0) continue; + if (line.length > this.maxFrameBytes) { + results.push(duplexFail("frame_too_large", "frame exceeds the maximum size")); + continue; + } + results.push(decodeDuplexLine(line)); + } + return results; + } +}`; + +/** + * Return the exact zero-dependency codec source the generated duplex gateway + * embeds. A test runs every fixture vector against this source, so it proves the + * embedded copy decodes the same bytes as the host codec. The source declares + * `encodeDuplexFrame`, `decodeDuplexLine`, `DuplexFrameDecoder`, + * `DUPLEX_FRAME_VERSION`, and `DEFAULT_MAX_DUPLEX_FRAME_BYTES`, but exports none + * of them; a caller wraps it to read those names. + */ +export function getSandboxDuplexGatewayCodecSource(): string { + return DUPLEX_GATEWAY_CODEC_SOURCE; +} + export function getSandboxCallbackBridgeServerSource(): string { return `import { randomUUID, timingSafeEqual } from "node:crypto"; import { createServer } from "node:http"; import { promises as fs } from "node:fs"; import path from "node:path"; +const bridgeMode = process.env.PAPERCLIP_API_BRIDGE_MODE || "${SANDBOX_CALLBACK_BRIDGE_FILE_MODE}"; const queueDir = process.env.PAPERCLIP_BRIDGE_QUEUE_DIR; const bridgeToken = process.env.PAPERCLIP_BRIDGE_TOKEN; const host = process.env.PAPERCLIP_BRIDGE_HOST || "127.0.0.1"; const port = Number(process.env.PAPERCLIP_BRIDGE_PORT || "0"); const pollIntervalMs = Number(process.env.PAPERCLIP_BRIDGE_POLL_INTERVAL_MS || "100"); -const responseTimeoutMs = Number(process.env.PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS || "30000"); +const responseTimeoutMs = Number( + process.env.PAPERCLIP_BRIDGE_RESPONSE_TIMEOUT_MS || + (bridgeMode === "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" + ? "${DEFAULT_DUPLEX_GATEWAY_WAIT_BUDGET_MS}" + : "${DEFAULT_BRIDGE_RESPONSE_TIMEOUT_MS}"), +); const maxQueueDepth = Number(process.env.PAPERCLIP_BRIDGE_MAX_QUEUE_DEPTH || "${DEFAULT_BRIDGE_MAX_QUEUE_DEPTH}"); const maxBodyBytes = Number(process.env.PAPERCLIP_BRIDGE_MAX_BODY_BYTES || "${DEFAULT_BRIDGE_MAX_BODY_BYTES}"); +const heartbeatIntervalMs = Number( + process.env.PAPERCLIP_BRIDGE_HEARTBEAT_INTERVAL_MS || "${DEFAULT_DUPLEX_GATEWAY_HEARTBEAT_INTERVAL_MS}", +); +const heartbeatTimeoutMs = Number( + process.env.PAPERCLIP_BRIDGE_HEARTBEAT_TIMEOUT_MS || "${DEFAULT_DUPLEX_GATEWAY_HEARTBEAT_TIMEOUT_MS}", +); +const lossExitGraceMs = Number( + process.env.PAPERCLIP_BRIDGE_LOSS_EXIT_GRACE_MS || "${DEFAULT_DUPLEX_GATEWAY_LOSS_EXIT_GRACE_MS}", +); +// The header allowlist. Both the file gateway and the duplex gateway strip an +// inbound request to these headers before they forward it. One copy serves both +// modes. The route allowlist stays on the host: both modes forward a request to +// the host, and the host enforces the same route allowlist for each. const allowedHeaders = new Set(${JSON.stringify([...DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST])}); -if (!queueDir || !bridgeToken) { +if (!bridgeToken) { + throw new Error("PAPERCLIP_BRIDGE_TOKEN is required."); +} +if (bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" && !queueDir) { throw new Error("PAPERCLIP_BRIDGE_QUEUE_DIR and PAPERCLIP_BRIDGE_TOKEN are required."); } -const requestsDir = path.posix.join(queueDir, "requests"); -const responsesDir = path.posix.join(queueDir, "responses"); -const logsDir = path.posix.join(queueDir, "logs"); -const readyFile = path.posix.join(queueDir, "ready.json"); +// The embedded zero-dependency frame codec. The duplex gateway uses it; the file +// gateway ignores it. +${DUPLEX_GATEWAY_CODEC_SOURCE} function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); @@ -1766,11 +1988,6 @@ async function readBody(req) { return Buffer.concat(chunks).toString("utf8"); } -async function queueDepth() { - const entries = await fs.readdir(requestsDir, { withFileTypes: true }).catch(() => []); - return entries.filter((entry) => entry.isFile() && entry.name.endsWith(".json")).length; -} - function tokensMatch(received) { const expected = Buffer.from(bridgeToken, "utf8"); const actual = Buffer.from(typeof received === "string" ? received : "", "utf8"); @@ -1778,116 +1995,327 @@ function tokensMatch(received) { return timingSafeEqual(expected, actual); } -async function waitForResponse(requestId) { - const responsePath = path.posix.join(responsesDir, \`\${requestId}.json\`); - const deadline = Date.now() + responseTimeoutMs; - while (Date.now() < deadline) { - const body = await fs.readFile(responsePath, "utf8").catch(() => null); - if (body != null) { - await fs.rm(responsePath, { force: true }).catch(() => undefined); - return JSON.parse(body); - } - await sleep(pollIntervalMs); - } - throw new Error("Timed out waiting for host bridge response."); +function writeJsonResponse(res, status, body) { + res.statusCode = status; + res.setHeader("content-type", "application/json"); + res.end(JSON.stringify(body)); } -const server = createServer(async (req, res) => { - try { - const auth = req.headers.authorization || ""; - const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : ""; - if (!tokensMatch(receivedToken)) { - res.statusCode = 401; - res.setHeader("content-type", "application/json"); - res.end(JSON.stringify({ error: "Invalid bridge token." })); - return; - } +async function runFileGateway() { + const requestsDir = path.posix.join(queueDir, "requests"); + const responsesDir = path.posix.join(queueDir, "responses"); + const logsDir = path.posix.join(queueDir, "logs"); + const readyFile = path.posix.join(queueDir, "ready.json"); - if (await queueDepth() >= maxQueueDepth) { - res.statusCode = 503; - res.setHeader("content-type", "application/json"); - res.end(JSON.stringify({ error: "Bridge request queue is full." })); - return; - } - - const url = new URL(req.url || "/", "http://127.0.0.1"); - const contentType = typeof req.headers["content-type"] === "string" ? req.headers["content-type"] : ""; - if (req.method && req.method !== "GET" && req.method !== "HEAD" && !/json/i.test(contentType)) { - res.statusCode = 415; - res.setHeader("content-type", "application/json"); - res.end(JSON.stringify({ error: "Bridge only accepts JSON request bodies." })); - return; - } - const requestId = randomUUID(); - const requestBody = await readBody(req); - const payload = { - id: requestId, - method: req.method || "GET", - path: url.pathname, - query: url.search, - headers: normalizeHeaders(req.headers), - body: requestBody, - createdAt: new Date().toISOString(), - }; - const requestPath = path.posix.join(requestsDir, \`\${requestId}.json\`); - const tempPath = \`\${requestPath}.tmp\`; - await fs.writeFile(tempPath, \`\${JSON.stringify(payload)}\\n\`, "utf8"); - await fs.rename(tempPath, requestPath); - - const response = await waitForResponse(requestId); - const responseHeaders = response.headers || {}; - // The host marks a possibly-committed mutation with an indeterminate outcome. - // The host cannot cancel a host operation that is in flight, so the mutation - // may have committed before the worker aborted the handler. A 5xx status is - // retryable by convention, so a caller that retries 5xx would apply the - // mutation twice. Map the indeterminate outcome to a non-retryable 409, so a - // standard retry policy does not repeat the request. The outcome header and - // body stay, so a caller that reads them still sees the indeterminate result. - const bridgeOutcome = responseHeaders["x-paperclip-bridge-outcome"]; - if (bridgeOutcome === "indeterminate") { - res.statusCode = 409; - } else { - res.statusCode = typeof response.status === "number" ? response.status : 200; - } - for (const [key, value] of Object.entries(responseHeaders)) { - if (typeof value !== "string" || key.toLowerCase() === "content-length") continue; - res.setHeader(key, value); - } - res.end(typeof response.body === "string" ? response.body : ""); - } catch (error) { - res.statusCode = 502; - res.setHeader("content-type", "application/json"); - res.end(JSON.stringify({ error: error instanceof Error ? error.message : String(error) })); + async function queueDepth() { + const entries = await fs.readdir(requestsDir, { withFileTypes: true }).catch(() => []); + return entries.filter((entry) => entry.isFile() && entry.name.endsWith(".json")).length; } -}); -async function shutdown() { - server.close(() => { - process.exit(0); + async function waitForResponse(requestId) { + const responsePath = path.posix.join(responsesDir, \`\${requestId}.json\`); + const deadline = Date.now() + responseTimeoutMs; + while (Date.now() < deadline) { + const body = await fs.readFile(responsePath, "utf8").catch(() => null); + if (body != null) { + await fs.rm(responsePath, { force: true }).catch(() => undefined); + return JSON.parse(body); + } + await sleep(pollIntervalMs); + } + throw new Error("Timed out waiting for host bridge response."); + } + + const server = createServer(async (req, res) => { + try { + const auth = req.headers.authorization || ""; + const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : ""; + if (!tokensMatch(receivedToken)) { + writeJsonResponse(res, 401, { error: "Invalid bridge token." }); + return; + } + + if (await queueDepth() >= maxQueueDepth) { + writeJsonResponse(res, 503, { error: "Bridge request queue is full." }); + return; + } + + const url = new URL(req.url || "/", "http://127.0.0.1"); + const contentType = typeof req.headers["content-type"] === "string" ? req.headers["content-type"] : ""; + if (req.method && req.method !== "GET" && req.method !== "HEAD" && !/json/i.test(contentType)) { + writeJsonResponse(res, 415, { error: "Bridge only accepts JSON request bodies." }); + return; + } + const requestId = randomUUID(); + const requestBody = await readBody(req); + const payload = { + id: requestId, + method: req.method || "GET", + path: url.pathname, + query: url.search, + headers: normalizeHeaders(req.headers), + body: requestBody, + createdAt: new Date().toISOString(), + }; + const requestPath = path.posix.join(requestsDir, \`\${requestId}.json\`); + const tempPath = \`\${requestPath}.tmp\`; + await fs.writeFile(tempPath, \`\${JSON.stringify(payload)}\\n\`, "utf8"); + await fs.rename(tempPath, requestPath); + + const response = await waitForResponse(requestId); + const responseHeaders = response.headers || {}; + // The host marks a possibly-committed mutation with an indeterminate outcome. + // The host cannot cancel a host operation that is in flight, so the mutation + // may have committed before the worker aborted the handler. A 5xx status is + // retryable by convention, so a caller that retries 5xx would apply the + // mutation twice. Map the indeterminate outcome to a non-retryable 409, so a + // standard retry policy does not repeat the request. The outcome header and + // body stay, so a caller that reads them still sees the indeterminate result. + const bridgeOutcome = responseHeaders["x-paperclip-bridge-outcome"]; + if (bridgeOutcome === "indeterminate") { + res.statusCode = 409; + } else { + res.statusCode = typeof response.status === "number" ? response.status : 200; + } + for (const [key, value] of Object.entries(responseHeaders)) { + if (typeof value !== "string" || key.toLowerCase() === "content-length") continue; + res.setHeader(key, value); + } + res.end(typeof response.body === "string" ? response.body : ""); + } catch (error) { + writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); + } + }); + + async function shutdown() { + server.close(() => { + process.exit(0); + }); + } + + process.on("SIGINT", () => void shutdown()); + process.on("SIGTERM", () => void shutdown()); + + await fs.mkdir(requestsDir, { recursive: true }); + await fs.mkdir(responsesDir, { recursive: true }); + await fs.mkdir(logsDir, { recursive: true }); + + server.listen(port, host, async () => { + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("Bridge server did not expose a TCP address."); + } + const ready = { + pid: process.pid, + host, + port: address.port, + baseUrl: \`http://\${host}:\${address.port}\`, + startedAt: new Date().toISOString(), + }; + const tempReadyFile = \`\${readyFile}.tmp\`; + await fs.writeFile(tempReadyFile, JSON.stringify(ready), "utf8"); + await fs.rename(tempReadyFile, readyFile); }); } -process.on("SIGINT", () => void shutdown()); -process.on("SIGTERM", () => void shutdown()); +function runDuplexGateway() { + // One outstanding local request per id. Each entry holds the HTTP resolver and + // the wait-budget timer. + const pending = new Map(); + let unavailable = false; + let lossTriggered = false; + let lastInboundAt = Date.now(); -await fs.mkdir(requestsDir, { recursive: true }); -await fs.mkdir(responsesDir, { recursive: true }); -await fs.mkdir(logsDir, { recursive: true }); - -server.listen(port, host, async () => { - const address = server.address(); - if (!address || typeof address === "string") { - throw new Error("Bridge server did not expose a TCP address."); + function diag(message) { + // Diagnostics go to stderr only. Stdout carries frames. + process.stderr.write("[paperclip-bridge] " + message + "\\n"); } - const ready = { - pid: process.pid, - host, - port: address.port, - baseUrl: \`http://\${host}:\${address.port}\`, - startedAt: new Date().toISOString(), - }; - const tempReadyFile = \`\${readyFile}.tmp\`; - await fs.writeFile(tempReadyFile, JSON.stringify(ready), "utf8"); - await fs.rename(tempReadyFile, readyFile); -});`; + + function writeFrame(frame) { + process.stdout.write(encodeDuplexFrame(frame)); + } + + function stopTimers() { + clearInterval(heartbeatSendTimer); + clearInterval(heartbeatWatchTimer); + } + + // Loss behavior. On stdin end, a close frame, or a heartbeat timeout, answer + // every outstanding request with a non-retryable 409 (the request may have + // reached the host), answer every new request with a retryable 503 (the + // request never left the gateway), then exit. The gateway never replays a + // request. + function triggerLoss(reason) { + if (lossTriggered) return; + lossTriggered = true; + unavailable = true; + diag("duplex channel lost: " + reason); + stopTimers(); + for (const entry of pending.values()) { + clearTimeout(entry.timer); + entry.resolve({ + status: 409, + headers: { + "content-type": "application/json", + "x-paperclip-bridge-outcome": "indeterminate", + }, + body: JSON.stringify({ error: "outcome_indeterminate" }), + }); + } + pending.clear(); + const exitTimer = setTimeout(() => process.exit(0), lossExitGraceMs); + if (typeof exitTimer.unref === "function") exitTimer.unref(); + } + + function handleInboundFrame(frame) { + if (frame.type === "response") { + const entry = pending.get(frame.id); + if (!entry) return; + pending.delete(frame.id); + clearTimeout(entry.timer); + // Map an indeterminate outcome to a non-retryable 409, the same contract + // the file gateway applies through the outcome header. + const statusCode = + frame.outcome === "indeterminate" ? 409 : typeof frame.status === "number" ? frame.status : 200; + entry.resolve({ + status: statusCode, + headers: frame.headers || {}, + body: typeof frame.body === "string" ? frame.body : "", + }); + return; + } + if (frame.type === "close") { + triggerLoss("host sent a close frame"); + return; + } + // A heartbeat, ready, error, or request frame proves liveness but needs no + // local action here. + } + + const decoder = new DuplexFrameDecoder(); + process.stdin.on("data", (chunk) => { + lastInboundAt = Date.now(); + for (const result of decoder.push(chunk)) { + if (!result.ok) { + diag("dropped an inbound frame: " + result.error.code); + continue; + } + handleInboundFrame(result.frame); + } + }); + process.stdin.on("end", () => triggerLoss("stdin reached end of file")); + process.stdin.on("error", (error) => + triggerLoss("stdin error: " + (error && error.message ? error.message : String(error))), + ); + + const heartbeatSendTimer = setInterval(() => { + writeFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }); + }, heartbeatIntervalMs); + const heartbeatWatchTimer = setInterval(() => { + if (Date.now() - lastInboundAt > heartbeatTimeoutMs) { + triggerLoss("no inbound frame within the heartbeat timeout"); + } + }, Math.max(200, Math.floor(heartbeatTimeoutMs / 4))); + + const server = createServer(async (req, res) => { + try { + const auth = req.headers.authorization || ""; + const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : ""; + if (!tokensMatch(receivedToken)) { + writeJsonResponse(res, 401, { error: "Invalid bridge token." }); + return; + } + if (unavailable) { + writeJsonResponse(res, 503, { error: "bridge_unavailable" }); + return; + } + if (pending.size >= maxQueueDepth) { + writeJsonResponse(res, 503, { error: "Bridge request queue is full." }); + return; + } + const url = new URL(req.url || "/", "http://127.0.0.1"); + const contentType = typeof req.headers["content-type"] === "string" ? req.headers["content-type"] : ""; + if (req.method && req.method !== "GET" && req.method !== "HEAD" && !/json/i.test(contentType)) { + writeJsonResponse(res, 415, { error: "Bridge only accepts JSON request bodies." }); + return; + } + const requestId = randomUUID(); + const requestBody = await readBody(req); + if (unavailable) { + writeJsonResponse(res, 503, { error: "bridge_unavailable" }); + return; + } + const requestFrame = { + version: DUPLEX_FRAME_VERSION, + type: "request", + id: requestId, + method: req.method || "GET", + path: url.pathname, + query: url.search, + headers: normalizeHeaders(req.headers), + body: requestBody, + }; + const response = await new Promise((resolve) => { + const timer = setTimeout(() => { + if (pending.delete(requestId)) { + resolve({ + status: 502, + headers: { "content-type": "application/json" }, + body: JSON.stringify({ error: "Timed out waiting for host bridge response." }), + }); + } + }, responseTimeoutMs); + pending.set(requestId, { resolve: resolve, timer: timer }); + writeFrame(requestFrame); + }); + res.statusCode = typeof response.status === "number" ? response.status : 200; + for (const [key, value] of Object.entries(response.headers || {})) { + if (typeof value !== "string" || key.toLowerCase() === "content-length") continue; + res.setHeader(key, value); + } + res.end(typeof response.body === "string" ? response.body : ""); + } catch (error) { + writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) }); + } + }); + + process.on("SIGINT", () => { + try { + server.close(); + } catch (error) { + diag("server close error: " + (error && error.message ? error.message : String(error))); + } + process.exit(0); + }); + process.on("SIGTERM", () => { + try { + server.close(); + } catch (error) { + diag("server close error: " + (error && error.message ? error.message : String(error))); + } + process.exit(0); + }); + + server.listen(port, host, () => { + const address = server.address(); + if (!address || typeof address === "string") { + diag("duplex gateway did not expose a TCP address"); + process.exit(1); + return; + } + // The one READY frame carries the validated listener address. Stdout carries + // only frames. + writeFrame({ + version: DUPLEX_FRAME_VERSION, + type: "ready", + address: "http://" + host + ":" + address.port, + }); + }); +} + +if (bridgeMode === "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}") { + runDuplexGateway(); +} else { + await runFileGateway(); +}`; } diff --git a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.live.test.ts b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.live.test.ts new file mode 100644 index 0000000000..88b80d2ec3 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.live.test.ts @@ -0,0 +1,194 @@ +// Live characterization of the Daytona duplex channel against the real provider. +// +// The test is credential-gated by design. It runs only when the run shell holds a +// Daytona API key, and it skips cleanly when the key is absent. The key comes from +// `DAYTONA_API_KEY`, which the plugin also reads as the credential fallback +// (`plugin.ts`). A skip is acceptable: the merged live coverage re-runs at a later +// phase and at rollout. +// +// What it proves on one live sandbox: +// 1. The duplex channel and an ordinary command run at the same time. +// 2. Host input sent after the child starts arrives, with no terminal echo and +// no frame corruption. +// 3. Channel teardown leaves no leaked provider session. +// +// The channel runs `cat` as the child. `cat` copies its input to its output, so a +// host write returns one time as program output. The launch wrapper sets the +// pseudo-terminal to raw mode with echo off, so the terminal adds no second copy +// and no newline translation. A returned line that equals the sent line, one time, +// proves both the delivery and the clean stream. +// +// The Daytona SDK is a runtime dependency of this live test only. The module loads +// it with a dynamic import inside the gated block, so the file imports with no SDK +// present and the other Daytona tests still run without it. + +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { randomUUID } from "node:crypto"; +import { + openDaytonaDuplexChannelSession, + type DaytonaPtyProcess, + type DuplexChannelSession, +} from "./duplex-command-stream.js"; + +const DAYTONA_API_KEY = process.env.DAYTONA_API_KEY?.trim() ?? ""; +const HAS_DAYTONA_CREDENTIAL = DAYTONA_API_KEY.length > 0; +const describeLive = HAS_DAYTONA_CREDENTIAL ? describe : describe.skip; + +if (!HAS_DAYTONA_CREDENTIAL) { + // eslint-disable-next-line no-console + console.warn( + "Skipping the Daytona duplex channel live test: DAYTONA_API_KEY is not set in the run shell.", + ); +} + +// A live sandbox create and destroy needs a generous timeout. The provider round +// trip dominates the wall-clock time. +const LIVE_TIMEOUT_MS = 180_000; + +// The minimal shape of the live Daytona sandbox that the test uses. The dynamic +// import returns the real SDK types; this local shape keeps the file type-safe +// without a static SDK import. +interface LiveDaytonaSandbox { + process: DaytonaPtyProcess & { + executeCommand(command: string): Promise<{ exitCode?: number; result?: string }>; + listSessions?: () => Promise>; + }; + delete(): Promise; +} + +/** + * Wait until `predicate(text)` is true for the running output, or reject after + * `timeoutMs`. The poll interval is short, so the wait ends soon after the output + * arrives. + */ +async function waitFor( + readOutput: () => string, + predicate: (text: string) => boolean, + timeoutMs: number, + label: string, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (predicate(readOutput())) return; + if (Date.now() > deadline) { + throw new Error(`Timed out waiting for ${label}. Output so far: ${JSON.stringify(readOutput())}`); + } + await new Promise((resolve) => setTimeout(resolve, 100)); + } +} + +/** Open a `cat` duplex channel and collect its output into a growing string. */ +async function openCatChannel( + sandbox: LiveDaytonaSandbox, +): Promise<{ session: DuplexChannelSession; readOutput: () => string }> { + let output = ""; + const session = await openDaytonaDuplexChannelSession(sandbox.process, "cat"); + session.onData((chunk) => { + output += chunk; + }); + return { session, readOutput: () => output }; +} + +describeLive("Daytona duplex channel (live)", () => { + let sandbox: LiveDaytonaSandbox | null = null; + + beforeAll(async () => { + const { Daytona } = await import("@daytonaio/sdk"); + const client = new Daytona({ apiKey: DAYTONA_API_KEY }); + sandbox = (await client.create()) as unknown as LiveDaytonaSandbox; + }, LIVE_TIMEOUT_MS); + + afterAll(async () => { + await sandbox?.delete().catch(() => undefined); + }, LIVE_TIMEOUT_MS); + + it( + "runs the duplex channel and an ordinary command on one live sandbox", + async () => { + const live = sandbox!; + const { session, readOutput } = await openCatChannel(live); + try { + // Run an ordinary command while the channel child stays alive. Both use the + // same live sandbox, so a success on both proves they coexist. + const token = `ordinary-${randomUUID()}`; + const ordinary = await live.process.executeCommand(`printf %s ${token}`); + expect(ordinary.exitCode ?? 0).toBe(0); + expect(ordinary.result ?? "").toContain(token); + + // The channel child still answers. A write returns as program output. + const ping = `chan-${randomUUID()}`; + session.write(`${ping}\n`); + await waitFor(readOutput, (text) => text.includes(ping), 30_000, "channel echo"); + } finally { + await session.close(); + } + }, + LIVE_TIMEOUT_MS, + ); + + it( + "delivers host input sent after startup with no echo and no frame corruption", + async () => { + const live = sandbox!; + const { session, readOutput } = await openCatChannel(live); + try { + // Let the launch wrapper run `stty raw -echo` before the host input. The + // wrapper output echoes in cooked mode, so wait for the raw-mode switch + // before the measured write. + await new Promise((resolve) => setTimeout(resolve, 4_000)); + const baseline = readOutput().length; + + const line = `PING-${randomUUID()}`; + session.write(`${line}\n`); + await waitFor( + readOutput, + (text) => text.slice(baseline).includes(line), + 30_000, + "delayed host input", + ); + + // Only `cat` writes the line, one time. Raw mode with echo off adds no + // second copy, so the sent token appears exactly one time after the write. + const afterWrite = readOutput().slice(baseline); + const occurrences = afterWrite.split(line).length - 1; + expect(occurrences).toBe(1); + + // The stream carries the frame bytes with no newline translation. The + // returned line keeps its single line-feed and gains no carriage return. + expect(afterWrite).toContain(`${line}\n`); + expect(afterWrite).not.toContain(`${line}\r`); + } finally { + await session.close(); + } + }, + LIVE_TIMEOUT_MS, + ); + + it( + "leaves no leaked provider session after channel close", + async () => { + const live = sandbox!; + const listSessions = live.process.listSessions?.bind(live.process); + const baseline = listSessions ? (await listSessions()).length : 0; + + const { session } = await openCatChannel(live); + await session.close(); + + if (listSessions) { + // The channel uses a pseudo-terminal, not an ordinary session. So the + // session list stays at the baseline. A leaked session would raise the + // count. + const after = (await listSessions()).length; + expect(after).toBe(baseline); + } + + // The child exits after close, so a fresh ordinary command still succeeds on + // the same sandbox. A leaked channel would block or corrupt the sandbox. + const token = `after-close-${randomUUID()}`; + const probe = await live.process.executeCommand(`printf %s ${token}`); + expect(probe.exitCode ?? 0).toBe(0); + expect(probe.result ?? "").toContain(token); + }, + LIVE_TIMEOUT_MS, + ); +}); diff --git a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts new file mode 100644 index 0000000000..476591afd6 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts @@ -0,0 +1,416 @@ +import { describe, expect, it } from "vitest"; +import { + buildDuplexChannelLaunchWrapper, + createDaytonaDuplexChannelSessionOpener, + openDaytonaDuplexChannelSession, + type DaytonaPtyCreateOptions, + type DaytonaPtyHandle, + type DaytonaPtyProcess, +} from "./duplex-command-stream.js"; + +// The carriage return that submits the launch wrapper line to the terminal. +const ENTER = "\r"; + +// The gateway command the tests run on the channel. The command is an argument +// vector: element 0 is the program and the rest are its arguments. The real +// command is the generated duplex gateway; the tests use a placeholder, because +// the fake PTY runs no process. +const GATEWAY = ["node", "/paperclip/gateway.mjs"]; + +// The gateway argument vector after the wrapper quotes each element. The wrapper +// quotes every argument as a single-quoted shell word. +const GATEWAY_QUOTED = "'node' '/paperclip/gateway.mjs'"; + +/** + * A fake Daytona PTY handle. It records each input write, drives the output + * stream on demand, and records the kill and the disconnect. The tests use it in + * place of the real SDK `PtyHandle`, so the session runs with no sandbox. The fake + * never echoes input back as output, so a test proves the channel adds no echo. + */ +function createFakePtyHandle(onData: (data: Uint8Array) => void | Promise): DaytonaPtyHandle & { + inputs: string[]; + emitText: (text: string) => void; + emitBytes: (bytes: Uint8Array) => void; + finish: (exitCode: number | undefined) => void; + killed: number; + disconnected: number; +} { + const encoder = new TextEncoder(); + const inputs: string[] = []; + let resolveWait: ((value: { exitCode?: number; error?: string }) => void) | null = null; + const waitPromise = new Promise<{ exitCode?: number; error?: string }>((resolve) => { + resolveWait = resolve; + }); + let killed = 0; + let disconnected = 0; + return { + async waitForConnection(): Promise {}, + async sendInput(data: string | Uint8Array): Promise { + inputs.push(typeof data === "string" ? data : new TextDecoder().decode(data)); + }, + wait(): Promise<{ exitCode?: number; error?: string }> { + return waitPromise; + }, + async kill(): Promise { + killed += 1; + }, + async disconnect(): Promise { + disconnected += 1; + }, + // Test control below. The session never reads these fields. + emitText(text: string): void { + onData(encoder.encode(text)); + }, + emitBytes(bytes: Uint8Array): void { + onData(bytes); + }, + finish(exitCode: number | undefined): void { + resolveWait?.({ exitCode }); + }, + get inputs(): string[] { + return inputs; + }, + get killed(): number { + return killed; + }, + get disconnected(): number { + return disconnected; + }, + }; +} + +/** + * A fake Daytona process. It opens one fake PTY handle and records the create + * options, so a test asserts the terminal size and the launch wrapper. + */ +function createFakeProcess(): DaytonaPtyProcess & { + handle: ReturnType | null; + createOptions: DaytonaPtyCreateOptions | null; +} { + const state: { + handle: ReturnType | null; + createOptions: DaytonaPtyCreateOptions | null; + } = { handle: null, createOptions: null }; + return { + get handle() { + return state.handle; + }, + get createOptions() { + return state.createOptions; + }, + async createPty(options: DaytonaPtyCreateOptions): Promise { + state.createOptions = options; + const handle = createFakePtyHandle(options.onData); + state.handle = handle; + return handle; + }, + }; +} + +// --------------------------------------------------------------------------- +// Characterization: which Daytona primitive carries a clean framed stream. +// --------------------------------------------------------------------------- +// The Daytona SDK proves two primitives. The ordinary session primitive +// dispatches a command and tails its output, split into stdout and stderr. It +// exposes no call that writes bytes to a running command's stdin, so it cannot +// carry host input on the same channel as the output. The PTY primitive exposes +// `sendInput` for host input and one `onData` callback for the output, on one live +// socket. Only the PTY carries a bidirectional stream on one connection, so the +// duplex channel uses the PTY. These tests record that decision against the two +// SDK surface shapes. + +/** The subset of the Daytona session command surface, for the characterization. */ +interface DaytonaSessionCommandSurface { + createSession(sessionId: string): Promise; + executeSessionCommand(sessionId: string, command: unknown): Promise<{ cmdId?: string }>; + getSessionCommandLogs( + sessionId: string, + commandId: string, + onStdout: (chunk: string) => void, + onStderr: (chunk: string) => void, + ): Promise; + deleteSession(sessionId: string): Promise; +} + +describe("duplex primitive characterization", () => { + it("the PTY surface carries host input and process output on one connection", () => { + const process = createFakeProcess(); + // The PTY surface exposes both directions: `sendInput` for host bytes and the + // `onData` create option for process bytes. This is the bidirectional shape a + // duplex channel needs. + expect(typeof process.createPty).toBe("function"); + const handleKeys: Array = [ + "waitForConnection", + "sendInput", + "wait", + "kill", + "disconnect", + ]; + // `sendInput` is the host-to-process direction; the create option `onData` is + // the process-to-host direction. Both exist on the PTY primitive. + expect(handleKeys).toContain("sendInput"); + }); + + it("the session command surface exposes no write to a running command's stdin", () => { + // The session surface dispatches a command and tails split stdout/stderr. No + // member writes bytes to a running command's stdin, so it cannot carry the + // host input on the output channel. The duplex channel therefore uses the PTY, + // not a session. + const sessionMembers: Array = [ + "createSession", + "executeSessionCommand", + "getSessionCommandLogs", + "deleteSession", + ]; + expect(sessionMembers).not.toContain("sendInput" as never); + expect(sessionMembers).not.toContain("write" as never); + // The log tail splits the output into two streams; a single framed duplex + // stream must stay on one path, which the PTY `onData` provides. + expect(sessionMembers).toContain("getSessionCommandLogs"); + }); +}); + +// --------------------------------------------------------------------------- +// Launch wrapper: raw mode, echo off, diagnostics to a file, direct exec. +// --------------------------------------------------------------------------- + +describe("buildDuplexChannelLaunchWrapper", () => { + it("sets raw mode with echo off, redirects diagnostics to a file, and execs the gateway", () => { + const wrapper = buildDuplexChannelLaunchWrapper(GATEWAY, "/tmp/diag.log"); + + // Raw mode with echo off stops the terminal echoing host input as data and + // stops newline translation, so the newline-delimited frames stay intact. + expect(wrapper).toContain("stty raw -echo"); + // The diagnostics go to the file, never to the stdout frame stream. The + // wrapper quotes the path as a single-quoted shell word. + expect(wrapper).toContain("2>'/tmp/diag.log'"); + // The gateway starts with `exec`, so the PTY runs it directly and its exit + // code becomes the PTY exit code. + expect(wrapper).toContain(`exec ${GATEWAY_QUOTED}`); + // The line ends with the Enter byte, so the shell runs it. + expect(wrapper.endsWith(ENTER)).toBe(true); + // Raw mode is set before the gateway starts. + expect(wrapper.indexOf("stty raw -echo")).toBeLessThan( + wrapper.indexOf(`exec ${GATEWAY_QUOTED}`), + ); + }); + + it("quotes a diagnostics path that holds shell metacharacters, so it cannot inject a command", () => { + const wrapper = buildDuplexChannelLaunchWrapper(GATEWAY, "/tmp/x; rm -rf ~"); + + // The path is one single-quoted word, so the shell reads the metacharacters as + // literal text. The dangerous `rm` never becomes its own command. + expect(wrapper).toContain("2>'/tmp/x; rm -rf ~'"); + expect(wrapper).not.toContain("2>/tmp/x; rm -rf ~"); + }); + + it("quotes a single quote in the diagnostics path", () => { + const wrapper = buildDuplexChannelLaunchWrapper(GATEWAY, "/tmp/o'clock.log"); + + // A single quote closes the word, adds an escaped quote, and reopens the word. + expect(wrapper).toContain(`2>'/tmp/o'"'"'clock.log'`); + }); + + it("quotes each command argument that holds shell metacharacters", () => { + const wrapper = buildDuplexChannelLaunchWrapper( + ["node", "/paperclip/gateway.mjs", "; rm -rf ~"], + "/tmp/diag.log", + ); + + // Each argument is one single-quoted word, so the metacharacters stay literal + // text. The dangerous `rm` argument never becomes its own command. + expect(wrapper).toContain(`exec 'node' '/paperclip/gateway.mjs' '; rm -rf ~'`); + expect(wrapper).not.toContain("; rm -rf ~;"); + expect(wrapper).not.toContain("exec node /paperclip/gateway.mjs ; rm -rf ~"); + }); + + it("rejects an empty command argument vector", () => { + // An empty vector has no program to exec, so the wrapper fails loudly. + expect(() => buildDuplexChannelLaunchWrapper([], "/tmp/diag.log")).toThrow( + /at least one argument/, + ); + }); +}); + +describe("openDaytonaDuplexChannelSession", () => { + it("starts the gateway through the raw-mode wrapper on a fixed-size terminal", async () => { + const process = createFakeProcess(); + + await openDaytonaDuplexChannelSession(process, GATEWAY); + + expect(process.createOptions?.cols).toBe(120); + expect(process.createOptions?.rows).toBe(30); + const launch = process.handle?.inputs[0] ?? ""; + expect(launch).toContain("stty raw -echo"); + expect(launch).toContain(`exec ${GATEWAY_QUOTED}`); + expect(launch.endsWith(ENTER)).toBe(true); + }); + + it("redirects diagnostics to the given file path", async () => { + const process = createFakeProcess(); + + await openDaytonaDuplexChannelSession(process, GATEWAY, { + diagnosticsPath: "/tmp/paperclip-duplex-fixed.log", + }); + + expect(process.handle?.inputs[0]).toContain("2>'/tmp/paperclip-duplex-fixed.log'"); + }); + + it("defaults the diagnostics path under /tmp when the caller gives none", async () => { + const process = createFakeProcess(); + + await openDaytonaDuplexChannelSession(process, GATEWAY); + + expect(process.handle?.inputs[0]).toMatch(/2>'\/tmp\/paperclip-duplex-[0-9a-f-]+\.log'/); + }); + + it("delivers a host write to the process and does not echo it back as data", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.onData((chunk) => received.push(chunk)); + + // The launch wrapper is the first input; the host frame write is the second. + session.write('{"version":1,"type":"heartbeat"}\n'); + expect(process.handle?.inputs[1]).toBe('{"version":1,"type":"heartbeat"}\n'); + // The fake terminal echoes nothing, so the write produces no data callback. + // A real terminal in raw mode with echo off behaves the same way. + expect(received).toEqual([]); + }); + + it("keeps the data order and the frame newlines intact", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.onData((chunk) => received.push(chunk)); + + process.handle?.emitText('{"version":1,"type":"ready","address":"127.0.0.1:8080"}\n'); + process.handle?.emitText('{"version":1,"type":"heartbeat"}\n'); + + // The order is preserved and no newline is translated (no stray "\r"). + expect(received).toEqual([ + '{"version":1,"type":"ready","address":"127.0.0.1:8080"}\n', + '{"version":1,"type":"heartbeat"}\n', + ]); + expect(received.join("")).not.toContain("\r"); + }); + + it("reassembles a frame split across two output chunks", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.onData((chunk) => received.push(chunk)); + + const frame = '{"version":1,"type":"response","id":"r1","status":200,"headers":{},"body":"","outcome":"completed"}\n'; + process.handle?.emitText(frame.slice(0, 20)); + process.handle?.emitText(frame.slice(20)); + + expect(received.join("")).toBe(frame); + }); + + it("carries a large frame without loss", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.onData((chunk) => received.push(chunk)); + + const body = "x".repeat(500_000); + const frame = `{"version":1,"type":"response","id":"big","status":200,"headers":{},"body":"${body}","outcome":"completed"}\n`; + process.handle?.emitText(frame); + + expect(received.join("")).toBe(frame); + expect(received.join("").length).toBe(frame.length); + }); + + it("keeps a multibyte character whole across two output chunks", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.onData((chunk) => received.push(chunk)); + + // The euro sign is three UTF-8 bytes. Split it across two chunks, so the + // stream decoder must join the bytes. + const euro = new TextEncoder().encode("€"); + process.handle?.emitBytes(euro.subarray(0, 2)); + process.handle?.emitBytes(euro.subarray(2)); + + expect(received.join("")).toBe("€"); + }); + + it("buffers early output until the listener registers", async () => { + const process = createFakeProcess(); + const received: string[] = []; + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + // Output arrives before the transport registers the listener. + process.handle?.emitText('{"version":1,'); + process.handle?.emitText('"type":"heartbeat"}\n'); + expect(received).toEqual([]); + + session.onData((chunk) => received.push(chunk)); + // The session flushes the buffered output in order on registration. + expect(received).toEqual(['{"version":1,"type":"heartbeat"}\n']); + }); + + it("resolves wait with the gateway exit code", async () => { + const process = createFakeProcess(); + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + process.handle?.finish(3); + + await expect(session.wait()).resolves.toEqual({ exitCode: 3 }); + }); + + it("maps an absent exit code to null", async () => { + const process = createFakeProcess(); + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + process.handle?.finish(undefined); + + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + }); + + it("kills the child on stop and releases the socket on close", async () => { + const process = createFakeProcess(); + + const session = await openDaytonaDuplexChannelSession(process, GATEWAY); + session.kill(); + expect(process.handle?.killed).toBe(1); + + await session.close(); + // Close stops the child and releases the socket, so a closed channel holds no + // live terminal. + expect(process.handle?.killed).toBeGreaterThanOrEqual(1); + expect(process.handle?.disconnected).toBe(1); + }); + + it("passes the working directory to the pseudo-terminal", async () => { + const process = createFakeProcess(); + + await openDaytonaDuplexChannelSession(process, GATEWAY, { cwd: "/paperclip-workspace" }); + + expect(process.createOptions?.cwd).toBe("/paperclip-workspace"); + }); +}); + +describe("createDaytonaDuplexChannelSessionOpener", () => { + it("opens a session for the command on each call", async () => { + const process = createFakeProcess(); + const opener = createDaytonaDuplexChannelSessionOpener(process, { cwd: "/paperclip-workspace" }); + + const session = await opener(GATEWAY); + + expect(process.createOptions?.cwd).toBe("/paperclip-workspace"); + expect(process.handle?.inputs[0]).toContain(`exec ${GATEWAY_QUOTED}`); + expect(typeof session.onData).toBe("function"); + expect(typeof session.write).toBe("function"); + expect(typeof session.wait).toBe("function"); + expect(typeof session.kill).toBe("function"); + expect(typeof session.close).toBe("function"); + }); +}); diff --git a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts new file mode 100644 index 0000000000..9651f65714 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts @@ -0,0 +1,229 @@ +// The Daytona duplex command stream for the sandbox callback bridge. The bridge +// needs one live bidirectional byte stream: the host writes request frames to the +// process, and the process writes response frames back, on one connection while +// the command runs. This module binds a Daytona pseudo-terminal (PTY) to a small +// session that the duplex transport consumes, so the gateway runs on a real +// terminal, streams its output, and receives host input. +// +// Primitive choice (the first acceptance criterion of the phase): the Daytona SDK +// proves two primitives. The ordinary session primitive (`createSession` plus +// `executeSessionCommand` plus `getSessionCommandLogs`) dispatches a command and +// tails its output, split into separate stdout and stderr streams. It exposes no +// call that writes bytes to a running command's stdin. The PTY primitive +// (`createPty`) exposes `sendInput` for host-to-process bytes and one `onData` +// callback for process-to-host bytes, on one live socket. Only the PTY carries a +// bidirectional stream on one connection, so this module uses the PTY. The +// characterization test in `duplex-command-stream.test.ts` records the evidence. +// +// Clean stream: a PTY echoes its input and translates newlines by default +// (`ICRNL` on input, `ONLCR` on output). The frames are newline-delimited JSON, so +// echo and newline translation corrupt the stream. The launch wrapper runs +// `stty raw -echo` before it starts the gateway, so the terminal becomes an 8-bit +// transparent path with no echo and no newline translation. The wrapper also +// redirects the gateway diagnostics to a file, so a diagnostic line never reaches +// the stdout frame stream. The wrapper starts the gateway with `exec`, so the PTY +// runs the gateway directly and the gateway exit code becomes the PTY exit code. +// +// Dependency boundary: this provider plugin ships standalone (the workspace +// excludes `packages/plugins/sandbox-providers/**`). So the module imports no +// workspace package. It reuses the narrow Daytona PTY surface that +// `setup-token-pty.ts` declares (`DaytonaPtyProcess`, `DaytonaPtyHandle`, +// `DaytonaPtyCreateOptions`), so a caller passes `sandbox.process` with no +// adapter. The narrow surface keeps the module unit-testable with a fake PTY. + +import { randomUUID } from "node:crypto"; + +import type { + DaytonaPtyCreateOptions, + DaytonaPtyHandle, + DaytonaPtyProcess, +} from "./setup-token-pty.js"; + +export type { + DaytonaPtyCreateOptions, + DaytonaPtyHandle, + DaytonaPtyProcess, +} from "./setup-token-pty.js"; + +/** + * A live duplex channel session for one command. The session allocates a real + * pseudo-terminal in raw mode, streams the raw output, accepts host input, and + * stops the child. The shape matches the worker duplex channel session, so the + * worker forwards the data and the exit with no adapter. + */ +export interface DuplexChannelSession { + /** Registers the one data listener. The session streams each raw chunk in order. */ + onData(listener: (chunk: string) => void): void; + /** Writes raw input bytes to the pseudo-terminal. */ + write(data: string): void; + /** Resolves with the child exit code when the command ends. */ + wait(): Promise<{ exitCode: number | null }>; + /** Stops the child process. Safe to call more than one time. */ + kill(): void; + /** Releases the session resources. Safe to call more than one time. */ + close(): Promise; +} + +/** + * Opens a {@link DuplexChannelSession} for `command`. The transport calls it one + * time. `command` is an argument vector: element 0 is the program and the rest + * are its arguments. The session quotes each element, so a shell metacharacter in + * an element cannot inject a shell command. + */ +export type DuplexChannelSessionOpener = ( + command: readonly string[], +) => Promise; + +/** The options for the Daytona duplex channel session. */ +export interface DaytonaDuplexChannelOptions { + /** The working directory for the duplex channel PTY. Defaults to the sandbox default. */ + cwd?: string; + /** + * The absolute sandbox path for the gateway diagnostics. The wrapper redirects + * the terminal's stderr to this path, so a diagnostic line never reaches the + * stdout frame stream. Defaults to a per-channel path under `/tmp`. + */ + diagnosticsPath?: string; +} + +// The terminal size for the duplex channel PTY. The channel carries bytes, not a +// visible UI, so a fixed standard size is enough. A fixed size keeps the launch +// deterministic. +const DUPLEX_CHANNEL_PTY_COLS = 120; +const DUPLEX_CHANNEL_PTY_ROWS = 30; + +/** + * The input terminator that submits the launch wrapper line. The terminal reads + * the carriage return as the Enter key, so the shell runs the wrapper line. + */ +const PTY_COMMAND_TERMINATOR = "\r"; + +/** + * Quotes one value for a POSIX shell. The result is one single-quoted word, so + * the shell reads it as literal text and never expands or splits it. The function + * closes the quote, adds an escaped single quote, and reopens the quote for each + * single quote in the value. + */ +function shellQuote(value: string): string { + return `'${value.replace(/'/g, `'"'"'`)}'`; +} + +/** + * Builds the launch wrapper input line for the duplex channel. `command` is an + * argument vector: element 0 is the program and the rest are its arguments. + * + * The wrapper does three steps in order: + * 1. `exec 2>''` redirects the shell's stderr to the file. The + * later `exec` inherits it, so the gateway diagnostics land in the file and + * never on the stdout frame stream. + * 2. `stty raw -echo` sets the terminal to raw mode with echo off, so the + * terminal neither echoes host input as data nor translates newlines. + * 3. `exec '' ''...` replaces the shell with the gateway, so the + * PTY runs the gateway directly and the gateway exit code becomes the PTY + * exit code. + * + * The wrapper quotes the diagnostics path and every command argument as a + * single-quoted shell word. So a shell metacharacter in a path or an argument + * stays literal text and cannot inject a shell command. + */ +export function buildDuplexChannelLaunchWrapper( + command: readonly string[], + diagnosticsPath: string, +): string { + if (command.length === 0) { + throw new Error("The duplex channel launch command needs at least one argument."); + } + const quotedCommand = command.map(shellQuote).join(" "); + return ( + `exec 2>${shellQuote(diagnosticsPath)}; stty raw -echo; ` + + `exec ${quotedCommand}${PTY_COMMAND_TERMINATOR}` + ); +} + +/** + * Opens a Daytona duplex channel PTY session for `command` and returns it as a + * {@link DuplexChannelSession}. The session allocates a real pseudo-terminal in + * raw mode, streams the raw output, accepts host input, and stops the child. + * + * The function decodes the terminal bytes as a UTF-8 stream, so a multibyte + * character that splits across two output chunks stays whole. It buffers the + * output until the transport registers the listener, so no early chunk is lost. + */ +export async function openDaytonaDuplexChannelSession( + process: DaytonaPtyProcess, + command: readonly string[], + options?: DaytonaDuplexChannelOptions, +): Promise { + const decoder = new TextDecoder("utf-8"); + let listener: ((chunk: string) => void) | null = null; + let buffered = ""; + + const diagnosticsPath = + options?.diagnosticsPath ?? `/tmp/paperclip-duplex-${randomUUID()}.log`; + + const handle = await process.createPty({ + id: `paperclip-duplex-${randomUUID()}`, + ...(options?.cwd ? { cwd: options.cwd } : {}), + cols: DUPLEX_CHANNEL_PTY_COLS, + rows: DUPLEX_CHANNEL_PTY_ROWS, + onData: (data: Uint8Array): void => { + // Decode the terminal bytes as a stream, so a split multibyte character + // stays whole across two chunks. Forward the raw text; the frame codec owns + // the newline-delimited JSON parsing. + const text = decoder.decode(data, { stream: true }); + if (text.length === 0) return; + if (listener) listener(text); + else buffered += text; + }, + }); + + await handle.waitForConnection(); + // Start the gateway through the raw-mode wrapper. Raw mode with echo off stops + // the terminal from echoing host input back as data and from translating the + // frame newlines. The diagnostics redirect keeps the stdout frame stream clean. + await handle.sendInput(buildDuplexChannelLaunchWrapper(command, diagnosticsPath)); + + return { + onData(next: (chunk: string) => void): void { + listener = next; + if (buffered.length > 0) { + const pending = buffered; + buffered = ""; + next(pending); + } + }, + write(data: string): void { + // Fire the input write. A write error must not throw into the transport, so + // the transport's stream stays the single result path. + void handle.sendInput(data).catch(() => undefined); + }, + async wait(): Promise<{ exitCode: number | null }> { + const result = await handle.wait(); + return { exitCode: typeof result.exitCode === "number" ? result.exitCode : null }; + }, + kill(): void { + void handle.kill().catch(() => undefined); + }, + async close(): Promise { + // Stop the child and release the socket. `kill` ends the gateway and `wait` + // resolves with the exit; `disconnect` releases the PTY socket. Both are + // safe to call more than one time, so close stays idempotent. + await handle.kill().catch(() => undefined); + await handle.disconnect().catch(() => undefined); + }, + }; +} + +/** + * Creates a {@link DuplexChannelSessionOpener} bound to a Daytona `process`. Pass + * `sandbox.process` from the Daytona SDK. The opener runs the gateway command on a + * raw pseudo-terminal, streams the output, accepts host input, and stops the child + * for a terminal state. + */ +export function createDaytonaDuplexChannelSessionOpener( + process: DaytonaPtyProcess, + options?: DaytonaDuplexChannelOptions, +): DuplexChannelSessionOpener { + return (command: readonly string[]) => + openDaytonaDuplexChannelSession(process, command, options); +} diff --git a/packages/plugins/sandbox-providers/daytona/src/manifest.ts b/packages/plugins/sandbox-providers/daytona/src/manifest.ts index 149d284a30..1dff4695e5 100644 --- a/packages/plugins/sandbox-providers/daytona/src/manifest.ts +++ b/packages/plugins/sandbox-providers/daytona/src/manifest.ts @@ -10,7 +10,8 @@ const PLUGIN_ID = "paperclip.daytona-sandbox-provider"; // 0.1.3 renamed the login transport flag from `supportsSetupTokenLogin` to the // neutral `supportsLoginPty`. // 0.1.4 adds the `concurrentSyncOperations` sandbox capability to the driver. -const PLUGIN_VERSION = "0.1.4"; +// 0.1.5 adds the `duplexCommandStream` sandbox capability to the driver. +const PLUGIN_VERSION = "0.1.5"; const manifest: PaperclipPluginManifestV1 = { id: PLUGIN_ID, @@ -44,9 +45,15 @@ const manifest: PaperclipPluginManifestV1 = { // waits for all active calls. Declare the opt-in capability so the host may // schedule sync operations concurrently. The host resolves it `true` only // when the worker also verifies both sync verbs. + // + // Daytona carries the sandbox callback bridge on one live duplex channel + // over a raw pseudo-terminal. Declare the opt-in capability so the host may + // select the duplex transport. The host resolves it `true` only when the + // worker also verifies the `duplexChannelOpen` handler. sandboxCapabilities: { incrementalSessionOutput: true, concurrentSyncOperations: true, + duplexCommandStream: true, }, supportsInteractiveSetup: true, interactiveSetupConnectionTypes: ["ssh"], diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts index 35021fac1f..1fb64271b8 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.test.ts @@ -148,6 +148,139 @@ describe("Daytona sandbox provider plugin", () => { }); }); + it("declares the duplex-command-stream capability and the four channel handlers", () => { + // Daytona carries the callback bridge on one duplex channel, so it declares + // the opt-in capability. The host resolves it `true` only when the worker also + // verifies the `duplexChannelOpen` handler, which the four handlers provide. + expect(manifest.environmentDrivers?.[0]?.sandboxCapabilities).toMatchObject({ + duplexCommandStream: true, + }); + expect(plugin.definition.onDuplexChannelOpen).toBeTypeOf("function"); + expect(plugin.definition.onDuplexChannelWrite).toBeTypeOf("function"); + expect(plugin.definition.onDuplexChannelStop).toBeTypeOf("function"); + expect(plugin.definition.onDuplexChannelClose).toBeTypeOf("function"); + }); + + it("bumps the plugin version so the server reconciles the stored manifest", () => { + // The bundled-plugin boot reconcile refreshes the stored manifest for an + // existing install only when the version changes. The duplex capability needs + // the bump to reach an existing install. + expect(manifest.version).toBe("0.1.5"); + }); + + it("opens a duplex channel, forwards a host write, and closes it on lease release", async () => { + process.env.DAYTONA_API_KEY = "host-key"; + // A fake PTY handle records each host write, drives the data stream on demand, + // and records the kill and the disconnect. + const inputs: string[] = []; + let killed = 0; + let disconnected = 0; + let ptyOnData: ((data: Uint8Array) => void) | null = null; + const handle = { + async waitForConnection() {}, + async sendInput(data: string | Uint8Array) { + inputs.push(typeof data === "string" ? data : new TextDecoder().decode(data)); + }, + wait() { + return new Promise<{ exitCode?: number }>(() => {}); + }, + async kill() { + killed += 1; + }, + async disconnect() { + disconnected += 1; + }, + }; + const sandbox = createMockSandbox(); + (sandbox.process as Record).createPty = vi.fn( + async (options: { onData: (data: Uint8Array) => void }) => { + ptyOnData = options.onData; + return handle; + }, + ); + mockCreate.mockResolvedValue(sandbox); + + // Capture the data and the exit the worker forwards through `ctx.duplexChannel`. + const dataChunks: Array<{ workerSessionId: string; chunk: string }> = []; + const restore = __setDaytonaPluginContextForTest({ + duplexChannel: { + data: (workerSessionId: string, chunk: string) => + dataChunks.push({ workerSessionId, chunk }), + exit: () => {}, + }, + } as unknown as PluginContext); + + try { + await plugin.definition.onEnvironmentAcquireLease?.({ + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + runId: "run-1", + agentId: "agent-1", + executionWorkspaceId: "workspace-1", + adapterType: "codex_local", + config: { image: "node:20", timeoutMs: 300000, reuseLease: true }, + }); + + const open = await plugin.definition.onDuplexChannelOpen?.({ + hostRouteId: "route-1", + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "sandbox-123", + command: ["node", "/paperclip/gateway.mjs"], + }); + expect(open?.workerSessionId).toMatch(/^duplex-/); + const workerSessionId = open?.workerSessionId ?? ""; + + // The launch wrapper sets raw mode with echo off and redirects diagnostics. + // It quotes each command argument and the diagnostics path as a shell word. + expect(inputs[0]).toContain("stty raw -echo"); + expect(inputs[0]).toContain("exec 'node' '/paperclip/gateway.mjs'"); + expect(inputs[0]).toMatch(/2>'\/tmp\/paperclip-duplex-.+\.log'/); + + // A host write reaches the process on the same channel. + await plugin.definition.onDuplexChannelWrite?.({ + workerSessionId, + data: '{"version":1,"type":"heartbeat"}\n', + }); + expect(inputs[1]).toBe('{"version":1,"type":"heartbeat"}\n'); + + // Process output reaches the host as a data notification bound to the + // worker session id. + ptyOnData?.(new TextEncoder().encode('{"version":1,"type":"ready","address":"127.0.0.1:1"}\n')); + expect(dataChunks).toEqual([ + { + workerSessionId, + chunk: '{"version":1,"type":"ready","address":"127.0.0.1:1"}\n', + }, + ]); + + // Lease release closes the channel: it kills the child and releases the + // pseudo-terminal socket. + await plugin.definition.onEnvironmentReleaseLease?.({ + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "sandbox-123", + config: { image: "node:20", timeoutMs: 300000, reuseLease: true }, + }); + expect(killed).toBeGreaterThanOrEqual(1); + expect(disconnected).toBe(1); + + // The channel entry is gone, so a later write is a no-op and reaches no + // process. + const inputsBefore = inputs.length; + await plugin.definition.onDuplexChannelWrite?.({ + workerSessionId, + data: "late\n", + }); + expect(inputs.length).toBe(inputsBefore); + } finally { + restore(); + } + }); + it("normalizes config and validates the API key fallback", async () => { process.env.DAYTONA_API_KEY = "host-key"; diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 2416b17fcd..684ec2b816 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -66,6 +66,24 @@ import type { DaytonaPtyProcess, } from "./setup-token-pty.js"; +// The Daytona duplex command stream for the sandbox callback bridge. The channel +// runs the gateway command on a raw pseudo-terminal, streams the frames, and +// accepts host input. The worker resolves the sandbox by the provider lease id, +// registers the channel under the host route id, and streams the data and the +// exit through `ctx.duplexChannel`. +export { + createDaytonaDuplexChannelSessionOpener, + openDaytonaDuplexChannelSession, + buildDuplexChannelLaunchWrapper, +} from "./duplex-command-stream.js"; +export type { + DuplexChannelSession, + DuplexChannelSessionOpener, + DaytonaDuplexChannelOptions, +} from "./duplex-command-stream.js"; +import { openDaytonaDuplexChannelSession as openDuplexChannelSession } from "./duplex-command-stream.js"; +import type { DuplexChannelSession } from "./duplex-command-stream.js"; + // Injectable monotonic clock for provider-boundary timing (Open Q1). Defaults // to the real wall clock; `plugin.test.ts` overrides it via // `setDaytonaTimingClockForTest` so the measured `durationMs`/`getDurationMs` @@ -1899,6 +1917,41 @@ function forgetDaytonaSetupTokenPty(entry: DaytonaSetupTokenPtyEntry): void { daytonaSetupTokenPtyBySession.delete(entry.workerSessionId); } +// The worker-side registry of live duplex channels. The worker registers each +// channel under the host-owned route identifier at open time, so the host closes +// the exact channel by that identifier even when the open reply was lost. It also +// indexes by the worker session identifier for write and stop. Each entry records +// the provider lease id, so a lease teardown closes only its own channels. The +// `onShutdown` hook closes every open channel here. +interface DaytonaDuplexChannelEntry { + hostRouteId: string; + workerSessionId: string; + providerLeaseId: string; + session: DuplexChannelSession; +} +const daytonaDuplexChannelByRoute = new Map(); +const daytonaDuplexChannelBySession = new Map(); + +function forgetDaytonaDuplexChannel(entry: DaytonaDuplexChannelEntry): void { + daytonaDuplexChannelByRoute.delete(entry.hostRouteId); + daytonaDuplexChannelBySession.delete(entry.workerSessionId); +} + +// Close every open duplex channel that belongs to one provider lease and drop its +// entry. The lease teardown hooks (release, destroy, resume) call this, so a +// channel never outlives the sandbox that carries it. The close kills the child +// and releases the pseudo-terminal socket, so no live channel survives the +// teardown. The stored identifiers are always cleared, so no orphan id survives. +async function closeDaytonaDuplexChannelsForLease(providerLeaseId: string): Promise { + const matches = [...daytonaDuplexChannelByRoute.values()].filter( + (entry) => entry.providerLeaseId === providerLeaseId, + ); + for (const entry of matches) { + forgetDaytonaDuplexChannel(entry); + await entry.session.close().catch(() => undefined); + } +} + const plugin = definePlugin({ async setup(ctx) { // Hoist the context to a module variable so the lifecycle hooks and the @@ -2106,6 +2159,10 @@ const plugin = definePlugin({ // session and leak its shell until sandbox reaping. if (sandbox.state !== "started") { sandboxHandleSessionStore.clear(scope); + // A stopped sandbox loses its pseudo-terminals, so a stored duplex channel + // is dead after a real restart. Close and drop every channel on this lease + // before the restart, so no stale channel id survives the resume. + await closeDaytonaDuplexChannelsForLease(params.providerLeaseId); } await ensureSandboxStarted(sandbox, toTimeoutSeconds(config.timeoutMs)); try { @@ -2170,6 +2227,9 @@ const plugin = definePlugin({ evictSandboxHandle(scope); await sandboxHandleActivityGates.waitForIdle(scope); await teardownSession(sandbox, scope); + // Close every duplex channel on this lease before the stop or the delete, + // so no channel outlives the sandbox and no stored channel id survives. + await closeDaytonaDuplexChannelsForLease(params.providerLeaseId); if (config.reuseLease) { if (sandbox.state !== "stopped") { @@ -2234,6 +2294,9 @@ const plugin = definePlugin({ evictSandboxHandle(scope); await sandboxHandleActivityGates.waitForIdle(scope); await teardownSession(sandbox, scope); + // Close every duplex channel on this lease before the delete, so no channel + // outlives the sandbox and no stored channel id survives. + await closeDaytonaDuplexChannelsForLease(params.providerLeaseId); await sandbox.delete(toTimeoutSeconds(config.timeoutMs)); } finally { sandboxHandleTeardownGates.end(scope, teardownGate); @@ -2734,8 +2797,76 @@ const plugin = definePlugin({ return { hostRouteId: params.hostRouteId }; }, - // Close every open login pseudo-terminal on an orderly shutdown, then drain the - // sandbox handle cache, so a graceful shutdown holds no live login terminal. + // Open one persistent duplex channel. Resolve the cached sandbox by the provider + // lease id, run the gateway command on a raw pseudo-terminal, and register the + // channel under the host route id. Stream the raw data and the exit through + // `ctx.duplexChannel`, bound to the returned worker session id. Fail closed when + // no cached sandbox matches the lease. + async onDuplexChannelOpen(params) { + const sandbox = await sandboxHandleCache.findByProviderLeaseId(params.providerLeaseId); + if (!sandbox) { + throw new Error( + "Daytona duplex channel: no cached sandbox resolves the provider lease.", + ); + } + const session = await openDuplexChannelSession( + sandbox.process as unknown as DaytonaPtyProcess, + params.command, + ); + const workerSessionId = `duplex-${randomUUID()}`; + const entry: DaytonaDuplexChannelEntry = { + hostRouteId: params.hostRouteId, + workerSessionId, + providerLeaseId: params.providerLeaseId, + session, + }; + daytonaDuplexChannelByRoute.set(params.hostRouteId, entry); + daytonaDuplexChannelBySession.set(workerSessionId, entry); + // Register the data listener before the first write, so no early data chunk is + // lost. The client stamps the worker session id, so the host binds the data to + // the open route. + session.onData((chunk) => { + pluginContext?.duplexChannel.data(workerSessionId, chunk); + }); + // Forward the child exit one time. The host resolves the open route on it. + void session.wait().then( + (result) => pluginContext?.duplexChannel.exit(workerSessionId, result.exitCode), + () => pluginContext?.duplexChannel.exit(workerSessionId, null), + ); + return { workerSessionId }; + }, + + // Write host input to an open duplex channel, keyed by the worker session id. + // Drop the input for an unknown session. + async onDuplexChannelWrite(params) { + const entry = daytonaDuplexChannelBySession.get(params.workerSessionId); + if (!entry) return; + entry.session.write(params.data); + }, + + // Stop an open duplex channel child, keyed by the worker session id. + async onDuplexChannelStop(params) { + const entry = daytonaDuplexChannelBySession.get(params.workerSessionId); + if (!entry) return; + entry.session.kill(); + }, + + // Close an open duplex channel by the host route id and acknowledge the close + // with the same identifier. The close is idempotent: it returns the + // acknowledgement even when the entry is already gone, so the host confirms the + // channel is closed. The worker never keys the close on the worker session id. + async onDuplexChannelClose(params) { + const entry = daytonaDuplexChannelByRoute.get(params.hostRouteId); + if (entry) { + forgetDaytonaDuplexChannel(entry); + await entry.session.close().catch(() => undefined); + } + return { hostRouteId: params.hostRouteId }; + }, + + // Close every open login pseudo-terminal and every open duplex channel on an + // orderly shutdown, then drain the sandbox handle cache, so a graceful shutdown + // holds no live terminal and no live channel. async onShutdown() { const openSessions = [...daytonaSetupTokenPtyByRoute.values()]; daytonaSetupTokenPtyByRoute.clear(); @@ -2743,6 +2874,12 @@ const plugin = definePlugin({ for (const entry of openSessions) { await entry.session.close().catch(() => undefined); } + const openChannels = [...daytonaDuplexChannelByRoute.values()]; + daytonaDuplexChannelByRoute.clear(); + daytonaDuplexChannelBySession.clear(); + for (const entry of openChannels) { + await entry.session.close().catch(() => undefined); + } sandboxHandleCache.reset(); }, }); diff --git a/packages/plugins/sdk/src/protocol.duplex-channel.test.ts b/packages/plugins/sdk/src/protocol.duplex-channel.test.ts index 306d7646c5..b9fb2529d4 100644 --- a/packages/plugins/sdk/src/protocol.duplex-channel.test.ts +++ b/packages/plugins/sdk/src/protocol.duplex-channel.test.ts @@ -33,7 +33,7 @@ describe("duplex channel request schemas", () => { companyId: "company-1", environmentId: "env-1", providerLeaseId: "lease-1", - command: "paperclip-bridge", + command: ["paperclip-bridge"], }; const reply: PluginDuplexChannelOpenResult = { workerSessionId: "ws-1" }; expect(open.hostRouteId).toBe("route-1"); @@ -47,20 +47,20 @@ describe("duplex channel request schemas", () => { companyId: "company-1", environmentId: "env-1", providerLeaseId: "lease-1", - command: "paperclip-bridge", + command: ["paperclip-bridge"], }; expect(open).toBeDefined(); }); - it("rejects an open request with a non-string command", () => { + it("rejects an open request whose command is not a string array", () => { const open: PluginDuplexChannelOpenParams = { hostRouteId: "route-1", driverKey: "daytona", companyId: "company-1", environmentId: "env-1", providerLeaseId: "lease-1", - // @ts-expect-error — command must be a string. - command: 42, + // @ts-expect-error — command must be a string array. + command: "paperclip-bridge", }; expect(open).toBeDefined(); }); diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 72585f7324..f81c2bd0dc 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1111,8 +1111,12 @@ export interface PluginDuplexChannelOpenParams { environmentId: string; /** The provider lease the sandbox is cached under. The worker resolves the sandbox by it. */ providerLeaseId: string; - /** The command the worker runs on the channel. */ - command: string; + /** + * The command argument vector the worker runs on the channel. Element 0 is the + * program and the rest are its arguments. The worker quotes each element for the + * shell, so a shell metacharacter in an element cannot inject a shell command. + */ + command: readonly string[]; } /** The open reply. It returns the worker session identifier for data binding only. */ diff --git a/packages/plugins/sdk/src/testing.ts b/packages/plugins/sdk/src/testing.ts index cf0c05b919..43ce72f658 100644 --- a/packages/plugins/sdk/src/testing.ts +++ b/packages/plugins/sdk/src/testing.ts @@ -2486,6 +2486,14 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness { // No-op in test harness — the host login route is not wired here. }, }, + duplexChannel: { + data(_workerSessionId: string, _chunk: string) { + // No-op in test harness — the host duplex route is not wired here. + }, + exit(_workerSessionId: string, _exitCode: number | null) { + // No-op in test harness — the host duplex route is not wired here. + }, + }, tools: { register(name, _decl, fn) { requireCapability(manifest, capabilitySet, "agent.tools.register"); diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index 455e70592f..6eda319ba5 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -2037,6 +2037,36 @@ export interface PluginSetupTokenPtyClient { exit(workerSessionId: string, exitCode: number | null): void; } +/** + * `ctx.duplexChannel` — stream one persistent duplex channel's data and exit from + * a sandbox provider worker to the host. + * + * The worker registers the data listener on the channel and forwards each raw + * chunk through `data(workerSessionId, chunk)`. It forwards the child exit through + * `exit(workerSessionId, exitCode)`. Each call carries the worker session + * identifier the open reply returned, so the host binds the data to the open route + * by that identifier while the route is open. The host drops a chunk or an exit + * that carries an unknown or a mismatched identifier, and it never logs the raw + * bytes. The default is a no-op that never throws. This client models the + * `setupTokenPty` client, but it carries no login command allowlist. + */ +export interface PluginDuplexChannelClient { + /** + * Deliver one raw data chunk of a persistent duplex channel. + * + * @param workerSessionId - The worker session identifier the open reply returned. + * @param chunk - The raw channel output text. + */ + data(workerSessionId: string, chunk: string): void; + /** + * Deliver the child exit of a persistent duplex channel. + * + * @param workerSessionId - The worker session identifier the open reply returned. + * @param exitCode - The child exit code, or null when the child ended with no code. + */ + exit(workerSessionId: string, exitCode: number | null): void; +} + // --------------------------------------------------------------------------- // Full plugin context // --------------------------------------------------------------------------- @@ -2157,6 +2187,10 @@ export interface PluginContext { * pseudo-terminal. */ setupTokenPty: PluginSetupTokenPtyClient; + /** Stream one persistent duplex channel's data and exit to the host. The + * default is a no-op for a provider that opens no duplex channel. */ + duplexChannel: PluginDuplexChannelClient; + /** Register agent tool handlers. Requires `agent.tools.register`. */ tools: PluginToolsClient; diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index c7cadbc5e9..4c00884e11 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -115,6 +115,8 @@ import type { import { SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION, SETUP_TOKEN_PTY_EXIT_NOTIFICATION, + DUPLEX_CHANNEL_DATA_NOTIFICATION, + DUPLEX_CHANNEL_EXIT_NOTIFICATION, JSONRPC_VERSION, JSONRPC_ERROR_CODES, PLUGIN_RPC_ERROR_CODES, @@ -1400,6 +1402,30 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }, }, + duplexChannel: { + data(workerSessionId: string, chunk: string): void { + // Forward one raw data chunk of a persistent duplex channel. The + // notification carries the worker session identifier, so the host binds + // the chunk to the open route by that identifier while the route is + // open. The host drops an unknown or a mismatched identifier and never + // logs the raw bytes. This notification carries no invocation id, + // because it fires after the open reply returns. + if (typeof workerSessionId !== "string" || workerSessionId.length === 0) return; + if (typeof chunk !== "string" || chunk.length === 0) return; + notifyHost(DUPLEX_CHANNEL_DATA_NOTIFICATION, { workerSessionId, chunk }); + }, + exit(workerSessionId: string, exitCode: number | null): void { + // Forward the child exit of a persistent duplex channel. The host + // resolves the open route's wait promise by the worker session + // identifier while the route is open. + if (typeof workerSessionId !== "string" || workerSessionId.length === 0) return; + notifyHost(DUPLEX_CHANNEL_EXIT_NOTIFICATION, { + workerSessionId, + exitCode: typeof exitCode === "number" ? exitCode : null, + }); + }, + }, + tools: { register( name: string, diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index 0b1f336c19..38c6e9b6b7 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -925,7 +925,7 @@ describe("worker duplex channel dispatch", () => { async setup() {}, async onDuplexChannelOpen(params) { expect(params.hostRouteId).toBe("route-1"); - expect(params.command).toBe("paperclip-bridge"); + expect(params.command).toEqual(["paperclip-bridge"]); expect(params.providerLeaseId).toBe("lease-1"); return { workerSessionId: "ws-1" }; }, @@ -1004,7 +1004,7 @@ describe("worker duplex channel dispatch", () => { companyId: "company-1", environmentId: "env-1", providerLeaseId: "lease-1", - command: "paperclip-bridge", + command: ["paperclip-bridge"], }), ).resolves.toEqual({ workerSessionId: "ws-1" }); diff --git a/packages/shared/src/feature-catalog.ts b/packages/shared/src/feature-catalog.ts index ef21ee914d..ef06ca8d37 100644 --- a/packages/shared/src/feature-catalog.ts +++ b/packages/shared/src/feature-catalog.ts @@ -253,6 +253,14 @@ export const INSTANCE_FEATURE_CATALOG: Record { expect(settings.enableGoalsSidebarLink).toBe(false); }); + it("defaults the sandbox duplex bridge kill switch off", () => { + const settings = instanceExperimentalSettingsSchema.parse({}); + + expect(settings.enableSandboxDuplexBridge).toBe(false); + }); + + it("accepts an explicit sandbox duplex bridge kill switch value", () => { + expect( + instanceExperimentalSettingsSchema.parse({ enableSandboxDuplexBridge: true }) + .enableSandboxDuplexBridge, + ).toBe(true); + expect( + instanceExperimentalSettingsSchema.parse({ enableSandboxDuplexBridge: false }) + .enableSandboxDuplexBridge, + ).toBe(false); + }); + + it("accepts the sandbox duplex bridge kill switch in a patch", () => { + expect( + patchInstanceExperimentalSettingsSchema.parse({ enableSandboxDuplexBridge: true }), + ).toEqual({ enableSandboxDuplexBridge: true }); + }); + it("defaults worktree run execution off", () => { const settings = instanceExperimentalSettingsSchema.parse({}); diff --git a/packages/shared/src/validators/instance.ts b/packages/shared/src/validators/instance.ts index 4af496897b..0ae769f8f8 100644 --- a/packages/shared/src/validators/instance.ts +++ b/packages/shared/src/validators/instance.ts @@ -66,6 +66,10 @@ export const instanceExperimentalSettingsSchema = z.object({ enableWorkspaceBranchReconcileForward: z.boolean().default(true), enableWorkspaceDirtyQuarantineRepair: z.boolean().default(true), enableOwnerInstanceAdmin: z.boolean().default(false), + // Kill switch for the sandbox duplex command-stream bridge. Default off. When + // off the host keeps the file bridge for every run with no manifest change and + // no redeploy. The host reads this per run before it selects the transport. + enableSandboxDuplexBridge: z.boolean().default(false), enableWorktreeRunExecution: z.boolean().default(false), worktreeRunExecutionActivatedAt: z.string().datetime().nullable().default(null), worktreeRunExecutionActivationInstanceId: z.string().min(1).nullable().default(null), diff --git a/server/src/__tests__/environment-runtime-duplex-bridge-input.test.ts b/server/src/__tests__/environment-runtime-duplex-bridge-input.test.ts new file mode 100644 index 0000000000..640a59a458 --- /dev/null +++ b/server/src/__tests__/environment-runtime-duplex-bridge-input.test.ts @@ -0,0 +1,69 @@ +import { beforeAll, describe, expect, it } from "vitest"; +import { createDb } from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { + environmentRuntimeService, + resolveSandboxDuplexBridgeInput, +} from "../services/environment-runtime.ts"; +import { instanceSettingsService } from "../services/instance-settings.ts"; + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +if (!embeddedPostgresSupport.supported) { + console.warn( + `Skipping embedded Postgres duplex bridge input tests on this host: ${embeddedPostgresSupport.reason ?? "unsupported environment"}`, + ); +} + +describe("resolveSandboxDuplexBridgeInput", () => { + it("carries an enabled kill switch into the bridge input", () => { + expect(resolveSandboxDuplexBridgeInput({ enableSandboxDuplexBridge: true })).toEqual({ + enableDuplexBridge: true, + }); + }); + + it("keeps the file bridge when the kill switch is off", () => { + expect(resolveSandboxDuplexBridgeInput({ enableSandboxDuplexBridge: false })).toEqual({ + enableDuplexBridge: false, + }); + }); +}); + +describeEmbeddedPostgres("environmentRuntimeService.readSandboxDuplexBridgeInput", () => { + let stopDb: (() => Promise) | null = null; + let db!: ReturnType; + let runtime!: ReturnType; + let settings!: ReturnType; + + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("environment-runtime-duplex-bridge"); + stopDb = started.stop; + db = createDb(started.connectionString); + runtime = environmentRuntimeService(db); + settings = instanceSettingsService(db); + return async () => { + await stopDb?.(); + }; + }); + + it("defaults the per-run bridge input to the file bridge", async () => { + expect(await runtime.readSandboxDuplexBridgeInput()).toEqual({ enableDuplexBridge: false }); + }); + + it("reads the enabled kill switch from instance settings for a run", async () => { + await settings.updateExperimental({ enableSandboxDuplexBridge: true }); + + expect(await runtime.readSandboxDuplexBridgeInput()).toEqual({ enableDuplexBridge: true }); + }); + + it("returns to the file bridge when the kill switch is turned back off", async () => { + await settings.updateExperimental({ enableSandboxDuplexBridge: true }); + await settings.updateExperimental({ enableSandboxDuplexBridge: false }); + + expect(await runtime.readSandboxDuplexBridgeInput()).toEqual({ enableDuplexBridge: false }); + }); +}); diff --git a/server/src/__tests__/instance-settings-service.test.ts b/server/src/__tests__/instance-settings-service.test.ts index 4c5627f417..d55a2064d3 100644 --- a/server/src/__tests__/instance-settings-service.test.ts +++ b/server/src/__tests__/instance-settings-service.test.ts @@ -52,6 +52,7 @@ describe("instance settings service", () => { enableWorkspaceBranchReconcileForward: true, enableWorkspaceDirtyQuarantineRepair: false, enableOwnerInstanceAdmin: false, + enableSandboxDuplexBridge: false, enableWorktreeRunExecution: false, worktreeRunExecutionActivatedAt: null, worktreeRunExecutionActivationInstanceId: null, diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 9588740e03..6cf5cf33f2 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -7,6 +7,7 @@ import type { EnvironmentLease, EnvironmentLeaseStatus, ExecutionWorkspace, + InstanceExperimentalSettings, IssueExecutionWorkspaceSettings, PluginEnvironmentConfig, SandboxEnvironmentConfig, @@ -32,6 +33,7 @@ import { type StartupSpanContext, } from "@paperclipai/adapter-utils/acpx-engine/startup-timing"; import { environmentService } from "./environments.js"; +import { instanceSettingsService } from "./instance-settings.js"; import { collectEnvironmentSecretRefs, parseEnvironmentDriverConfig, @@ -321,6 +323,29 @@ export function buildEnvironmentLeaseContext(input: { }; } +/** + * The per-run duplex bridge input. The host resolves it for each new run before + * it selects the sandbox callback bridge transport. When `enableDuplexBridge` is + * false the host keeps the file bridge with no manifest change and no redeploy. + * The transport selection reads this input in a later phase; this phase only + * delivers the setting and the per-run read. + */ +export interface ResolvedSandboxDuplexBridgeInput { + /** True only when the instance opts in to the sandbox duplex command stream. */ + enableDuplexBridge: boolean; +} + +/** + * Map the experimental instance setting `enableSandboxDuplexBridge` into the + * per-run duplex bridge input. The setting is the kill switch. It defaults off, + * so an absent or false setting keeps the file bridge. + */ +export function resolveSandboxDuplexBridgeInput( + experimental: Pick, +): ResolvedSandboxDuplexBridgeInput { + return { enableDuplexBridge: experimental.enableSandboxDuplexBridge === true }; +} + function stripSecretRefValuesFromPluginLeaseMetadata(input: { metadata: Record | null | undefined; schema: Record | null | undefined; @@ -498,8 +523,12 @@ export interface EnvironmentDriverSyncInput extends EnvironmentDriverLeaseInput } export interface EnvironmentDriverOpenDuplexChannelInput extends EnvironmentDriverLeaseInput { - /** The command line the sandbox runs as the duplex channel child process. */ - command: string; + /** + * The command argument vector the sandbox runs as the duplex channel child + * process. Element 0 is the program. The worker runs the vector with no + * shell, so a shell metacharacter in an element cannot inject a command. + */ + command: readonly string[]; } export interface EnvironmentRuntimeDriver { @@ -3138,6 +3167,17 @@ export function environmentRuntimeService( return { getDriver, + /** + * Read the sandbox duplex bridge kill switch for a new run. The host calls it + * per run before it selects the callback bridge transport. It reads the + * experimental instance setting `enableSandboxDuplexBridge` and maps it into + * the resolved bridge input. The default-off setting keeps the file bridge. + */ + async readSandboxDuplexBridgeInput(): Promise { + const experimental = await instanceSettingsService(db).getExperimental(); + return resolveSandboxDuplexBridgeInput(experimental); + }, + async acquireRunLease(input: { companyId: string; environment: Environment; diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 32e2c125ed..473b12313b 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -245,6 +245,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta enableWorkspaceBranchReconcileForward: parsed.data.enableWorkspaceBranchReconcileForward ?? true, enableWorkspaceDirtyQuarantineRepair: parsed.data.enableWorkspaceDirtyQuarantineRepair ?? true, enableOwnerInstanceAdmin: parsed.data.enableOwnerInstanceAdmin ?? false, + enableSandboxDuplexBridge: parsed.data.enableSandboxDuplexBridge ?? false, enableWorktreeRunExecution: parsed.data.enableWorktreeRunExecution ?? false, worktreeRunExecutionActivatedAt: parsed.data.worktreeRunExecutionActivatedAt ?? null, worktreeRunExecutionActivationInstanceId: @@ -282,6 +283,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta enableWorkspaceBranchReconcileForward: true, enableWorkspaceDirtyQuarantineRepair: true, enableOwnerInstanceAdmin: false, + enableSandboxDuplexBridge: false, enableWorktreeRunExecution: false, worktreeRunExecutionActivatedAt: null, worktreeRunExecutionActivationInstanceId: null, diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 8c54cdbeb2..3412290cb2 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -490,7 +490,12 @@ export interface DuplexChannelOpenInput { companyId: string; environmentId: string; providerLeaseId: string; - command: string; + /** + * The command argument vector the worker runs on the channel. Element 0 is the + * program. The worker runs the vector with no shell, so a shell metacharacter + * in an element cannot inject a command. + */ + command: readonly string[]; } /** diff --git a/ui/src/pages/InstanceExperimentalSettings.test.tsx b/ui/src/pages/InstanceExperimentalSettings.test.tsx index 25a717df60..87cdfef1c8 100644 --- a/ui/src/pages/InstanceExperimentalSettings.test.tsx +++ b/ui/src/pages/InstanceExperimentalSettings.test.tsx @@ -99,6 +99,7 @@ function defaultExperimentalSettings(): InstanceExperimentalSettingsPayload { enableWorkspaceBranchReconcileForward: true, enableWorkspaceDirtyQuarantineRepair: true, enableOwnerInstanceAdmin: false, + enableSandboxDuplexBridge: false, enableWorktreeRunExecution: false, worktreeRunExecutionActivatedAt: null, worktreeRunExecutionActivationInstanceId: null,