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 <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
d5416fde9a
commit
faab2620ad
|
|
@ -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[];
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -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<string, string>;
|
||||
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<ReturnType<typeof spawn>> = [];
|
||||
|
||||
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<DecodedGatewayFrame>;
|
||||
sendFrame: (frame: Record<string, unknown>) => void;
|
||||
sendRaw: (text: string) => void;
|
||||
endStdin: () => void;
|
||||
exited: Promise<number | null>;
|
||||
stop: () => Promise<void>;
|
||||
}
|
||||
|
||||
// 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<string, string>): Promise<DuplexGatewayHandle> {
|
||||
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<number | null>((resolve) => {
|
||||
child.on("exit", (code) => resolve(code));
|
||||
});
|
||||
|
||||
const waitForFrame = (
|
||||
predicate: (frame: DecodedGatewayFrame) => boolean,
|
||||
timeoutMs = 5000,
|
||||
): Promise<DecodedGatewayFrame> => {
|
||||
const existing = frames.find(predicate);
|
||||
if (existing) return Promise.resolve(existing);
|
||||
return new Promise<DecodedGatewayFrame>((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);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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<string, string> {
|
||||
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();
|
||||
}`;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<Array<{ sessionId?: string }>>;
|
||||
};
|
||||
delete(): Promise<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<void> {
|
||||
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,
|
||||
);
|
||||
});
|
||||
|
|
@ -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<void>): 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<void> {},
|
||||
async sendInput(data: string | Uint8Array): Promise<void> {
|
||||
inputs.push(typeof data === "string" ? data : new TextDecoder().decode(data));
|
||||
},
|
||||
wait(): Promise<{ exitCode?: number; error?: string }> {
|
||||
return waitPromise;
|
||||
},
|
||||
async kill(): Promise<void> {
|
||||
killed += 1;
|
||||
},
|
||||
async disconnect(): Promise<void> {
|
||||
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<typeof createFakePtyHandle> | null;
|
||||
createOptions: DaytonaPtyCreateOptions | null;
|
||||
} {
|
||||
const state: {
|
||||
handle: ReturnType<typeof createFakePtyHandle> | null;
|
||||
createOptions: DaytonaPtyCreateOptions | null;
|
||||
} = { handle: null, createOptions: null };
|
||||
return {
|
||||
get handle() {
|
||||
return state.handle;
|
||||
},
|
||||
get createOptions() {
|
||||
return state.createOptions;
|
||||
},
|
||||
async createPty(options: DaytonaPtyCreateOptions): Promise<DaytonaPtyHandle> {
|
||||
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<void>;
|
||||
executeSessionCommand(sessionId: string, command: unknown): Promise<{ cmdId?: string }>;
|
||||
getSessionCommandLogs(
|
||||
sessionId: string,
|
||||
commandId: string,
|
||||
onStdout: (chunk: string) => void,
|
||||
onStderr: (chunk: string) => void,
|
||||
): Promise<void>;
|
||||
deleteSession(sessionId: string): Promise<void>;
|
||||
}
|
||||
|
||||
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<keyof DaytonaPtyHandle> = [
|
||||
"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<keyof DaytonaSessionCommandSurface> = [
|
||||
"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");
|
||||
});
|
||||
});
|
||||
|
|
@ -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<void>;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<DuplexChannelSession>;
|
||||
|
||||
/** 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>'<diagnosticsPath>'` 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 '<program>' '<arg>'...` 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<DuplexChannelSession> {
|
||||
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<void> {
|
||||
// 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);
|
||||
}
|
||||
|
|
@ -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"],
|
||||
|
|
|
|||
|
|
@ -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<string, unknown>).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";
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, DaytonaDuplexChannelEntry>();
|
||||
const daytonaDuplexChannelBySession = new Map<string, DaytonaDuplexChannelEntry>();
|
||||
|
||||
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<void> {
|
||||
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();
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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");
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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" });
|
||||
|
||||
|
|
|
|||
|
|
@ -253,6 +253,14 @@ export const INSTANCE_FEATURE_CATALOG: Record<InstanceFeatureKey, FeatureCatalog
|
|||
cloudDefault: true,
|
||||
selfHostedDefault: false,
|
||||
},
|
||||
enableSandboxDuplexBridge: {
|
||||
title: "Sandbox Duplex Bridge",
|
||||
description:
|
||||
"Let a run open the sandbox duplex command-stream bridge when the provider grants the capability. The host reads this per run before it selects the transport. Off keeps the file bridge for every run.",
|
||||
tier: "managed",
|
||||
cloudDefault: false,
|
||||
selfHostedDefault: false,
|
||||
},
|
||||
enableWorktreeRunExecution: {
|
||||
title: "Worktree Run Execution",
|
||||
description:
|
||||
|
|
|
|||
|
|
@ -89,6 +89,13 @@ export interface InstanceExperimentalSettings {
|
|||
* instances, which have no trusted cloud tenant path.
|
||||
*/
|
||||
enableOwnerInstanceAdmin: boolean;
|
||||
/**
|
||||
* Kill switch for the sandbox duplex command-stream bridge. Default off. The
|
||||
* host reads this per run before it selects the callback bridge transport.
|
||||
* Off forces the file bridge for every run with no manifest change and no
|
||||
* redeploy.
|
||||
*/
|
||||
enableSandboxDuplexBridge: boolean;
|
||||
/**
|
||||
* Worktree preview instances (`PAPERCLIP_IN_WORKTREE=true`) suppress the
|
||||
* heartbeat run engine by default so previews never self-execute tasks. When
|
||||
|
|
|
|||
|
|
@ -24,6 +24,29 @@ describe("instance experimental settings validators", () => {
|
|||
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({});
|
||||
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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<void>) | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let runtime!: ReturnType<typeof environmentRuntimeService>;
|
||||
let settings!: ReturnType<typeof instanceSettingsService>;
|
||||
|
||||
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 });
|
||||
});
|
||||
});
|
||||
|
|
@ -52,6 +52,7 @@ describe("instance settings service", () => {
|
|||
enableWorkspaceBranchReconcileForward: true,
|
||||
enableWorkspaceDirtyQuarantineRepair: false,
|
||||
enableOwnerInstanceAdmin: false,
|
||||
enableSandboxDuplexBridge: false,
|
||||
enableWorktreeRunExecution: false,
|
||||
worktreeRunExecutionActivatedAt: null,
|
||||
worktreeRunExecutionActivationInstanceId: null,
|
||||
|
|
|
|||
|
|
@ -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<InstanceExperimentalSettings, "enableSandboxDuplexBridge">,
|
||||
): ResolvedSandboxDuplexBridgeInput {
|
||||
return { enableDuplexBridge: experimental.enableSandboxDuplexBridge === true };
|
||||
}
|
||||
|
||||
function stripSecretRefValuesFromPluginLeaseMetadata(input: {
|
||||
metadata: Record<string, unknown> | null | undefined;
|
||||
schema: Record<string, unknown> | 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<ResolvedSandboxDuplexBridgeInput> {
|
||||
const experimental = await instanceSettingsService(db).getExperimental();
|
||||
return resolveSandboxDuplexBridgeInput(experimental);
|
||||
},
|
||||
|
||||
async acquireRunLease(input: {
|
||||
companyId: string;
|
||||
environment: Environment;
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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[];
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -99,6 +99,7 @@ function defaultExperimentalSettings(): InstanceExperimentalSettingsPayload {
|
|||
enableWorkspaceBranchReconcileForward: true,
|
||||
enableWorkspaceDirtyQuarantineRepair: true,
|
||||
enableOwnerInstanceAdmin: false,
|
||||
enableSandboxDuplexBridge: false,
|
||||
enableWorktreeRunExecution: false,
|
||||
worktreeRunExecutionActivatedAt: null,
|
||||
worktreeRunExecutionActivationInstanceId: null,
|
||||
|
|
|
|||
Loading…
Reference in New Issue