From 81612442840976ab9c16ccaadf0bbafb9debfca2 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Wed, 19 Aug 2026 13:39:13 -0700 Subject: [PATCH] feat(sandbox): add opt-in duplex command-stream foundation (capability, protocol, bounded host route, frame codec) (#11738) ## Thinking Path > - Paperclip provides a control plane for companies that run AI agents. > - Sandboxed agents need a safe execution path for persistent command streams. > - The existing callback transport does not provide a bounded, generic duplex route. > - The host must control capability access, route identity, protocol limits, and close behavior. > - This pull request adds an opt-in duplex command-stream foundation across the sandbox layers. > - The feature stays inert because no current provider declares the capability. ## Linked Issues or Issue Description **Subsystem affected** Cross-cutting (multiple of the above) **Problem or motivation** Sandbox command execution needs a persistent host-to-sandbox stream. The current callback bridge uses a file transport and does not provide this generic route. **Proposed solution** Add a fail-closed provider capability, generic worker protocol messages, a host-owned bounded route, cross-layer service mediation, and a versioned newline-delimited frame codec. **Alternatives considered** Keep the file transport and add feature-specific commands. This does not provide one reusable duplex contract or host-owned route bounds. **Roadmap alignment** This work supports the completed Cloud / Sandbox agents roadmap area and the safe autonomy goal in the product definition. **Additional context** The change passed a two-stage security review. The final code review verdict was approve after fixes for active-stream bounds and service-layer capability mediation. ## What Changed - Add the opt-in `duplexCommandStream` provider capability with fail-closed narrowing. - Add duplex open, write, stop, and close requests and data and exit notifications to the plugin worker protocol. - Add a host-owned route with bounds for chunk size, cumulative bytes, lifetime, protocol errors, pending requests, and pre-bind buffering. - Add close acknowledgement handling with worker retirement when the close remains unconfirmed. - Wire `openDuplexChannel` through the execution target, runtime service, and plugin worker. - Add a versioned frame codec with shared wire-compatibility vectors and split UTF-8 handling. ## Verification - `server/src/__tests__/plugin-worker-manager-duplex.test.ts` passes 18 tests. - `server/src/__tests__/environment-execution-target-duplex.test.ts` passes 11 tests. - `packages/adapter-utils/src/duplex-frame-codec.test.ts` passes 38 tests. - `server/src/__tests__/sandbox-capability-contract.test.ts` passes 15 tests. - Setup-token pseudo-terminal regression tests pass 47 tests. - Server TypeScript check passes. - Continuous integration will run the full required test, typecheck, build, and policy checks. ## Risks - Providers that opt into the capability must implement the complete worker protocol. - Route limit defaults can close a stream when a workload exceeds the configured bounds. - The capability remains disabled for current providers, so current production behavior does not change. ## Model Used OpenAI GPT-5 (`gpt-5`), with tool use and code execution. The model reviewed and prepared this pull request from the supplied implementation and verification record. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs: #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../src/command-managed-runtime.ts | 37 ++ .../src/duplex-frame-codec.test.ts | 201 ++++++ .../adapter-utils/src/duplex-frame-codec.ts | 338 ++++++++++ .../src/duplex-frame-vectors.json | 358 +++++++++++ .../adapter-utils/src/execution-target.ts | 2 + packages/plugins/sdk/src/define-plugin.ts | 33 + packages/plugins/sdk/src/index.ts | 2 + .../sdk/src/protocol.duplex-channel.test.ts | 156 +++++ packages/plugins/sdk/src/protocol.ts | 114 ++++ packages/plugins/sdk/src/worker-rpc-host.ts | 48 ++ .../plugins/sdk/tests/worker-rpc-host.test.ts | 188 ++++++ packages/plugins/sdk/vitest.config.ts | 2 +- packages/shared/src/types/plugin.ts | 11 + packages/shared/src/validators/plugin.ts | 1 + ...ment-execution-target-capabilities.test.ts | 2 + ...nvironment-execution-target-duplex.test.ts | 379 ++++++++++++ .../fixtures/plugin-worker-duplex-channel.cjs | 174 ++++++ .../plugin-worker-manager-duplex.test.ts | 505 +++++++++++++++ .../sandbox-capability-contract.test.ts | 42 ++ .../services/environment-execution-target.ts | 15 + server/src/services/environment-runtime.ts | 140 ++++- server/src/services/plugin-worker-manager.ts | 581 +++++++++++++++++- 22 files changed, 3304 insertions(+), 25 deletions(-) create mode 100644 packages/adapter-utils/src/duplex-frame-codec.test.ts create mode 100644 packages/adapter-utils/src/duplex-frame-codec.ts create mode 100644 packages/adapter-utils/src/duplex-frame-vectors.json create mode 100644 packages/plugins/sdk/src/protocol.duplex-channel.test.ts create mode 100644 server/src/__tests__/environment-execution-target-duplex.test.ts create mode 100644 server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs create mode 100644 server/src/__tests__/plugin-worker-manager-duplex.test.ts diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 9c39ed3d06..22a458b135 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -18,6 +18,35 @@ import type { RunProcessResult } from "./server-utils.js"; import type { RuntimeProgressSink, RuntimeStatusSink } from "./runtime-progress.js"; 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 + * `DuplexChannelOpenInput`, which also carries the lease scope fields. + */ +export interface DuplexChannelOpenInput { + command: string; +} + +/** + * A persistent bidirectional channel to one long-lived command in the sandbox. + * The caller writes raw input bytes, reads streamed output, and stops or closes + * the channel. This is the cross-layer channel type: the runner returns it, and + * the sandbox driver adapts the worker manager's host session to it. + */ +export interface CommandManagedDuplexChannel { + /** Writes raw input bytes to the channel. */ + write(data: string): void; + /** Registers the one data listener. The channel streams each raw chunk in order. */ + onData(listener: (chunk: string) => void): void; + /** Registers the one exit listener. The channel calls it one time with the exit. */ + onExit(listener: (exit: { exitCode: number | null }) => void): void; + /** Stops the child process. Safe to call more than one time. */ + stop(): void; + /** Closes the channel and releases the route. Safe to call more than one time. */ + close(): Promise; +} + export interface CommandManagedRuntimeRunner { /** * True when the provider verified the concurrent-sync opt-in. A native runner @@ -76,6 +105,14 @@ export interface CommandManagedRuntimeRunner { syncIn?(operations: SandboxSyncOperation[]): Promise; /** Optional native outbound file transfer. See {@link syncIn}. */ syncOut?(operations: SandboxSyncOperation[]): Promise; + /** + * Optional persistent duplex channel. Present only when the sandbox provider's + * effective capability grants `duplexCommandStream`. The runner opens one + * bidirectional channel to a long-lived command in the sandbox. The SSH runner + * and every provider without the capability omit the member, so a caller gates + * on its presence in the same style as {@link syncIn}/{@link syncOut}. + */ + openDuplexChannel?(input: DuplexChannelOpenInput): Promise; } export interface CommandManagedRuntimeSpec { diff --git a/packages/adapter-utils/src/duplex-frame-codec.test.ts b/packages/adapter-utils/src/duplex-frame-codec.test.ts new file mode 100644 index 0000000000..539d4898ec --- /dev/null +++ b/packages/adapter-utils/src/duplex-frame-codec.test.ts @@ -0,0 +1,201 @@ +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import { describe, expect, it } from "vitest"; +import { + DEFAULT_MAX_DUPLEX_FRAME_BYTES, + DUPLEX_FRAME_VERSION, + DuplexFrameDecoder, + decodeDuplexLine, + encodeDuplexFrame, + type DuplexDecodeResult, + type DuplexFrame, +} from "./duplex-frame-codec.js"; + +// One expected decode result in the fixture: either a decoded frame or the code +// of a protocol error. The codec must never throw on the read path. +type ExpectedResult = { frame: DuplexFrame } | { error: string }; + +interface Vector { + name: string; + category: "valid" | "invalid" | "partial" | "oversized" | "versionMismatch"; + bytes: string; + splitByteOffsets?: number[]; + maxFrameBytes?: number; + roundTrip?: boolean; + expected: ExpectedResult[]; +} + +interface Fixture { + frameVersion: number; + defaultMaxFrameBytes: number; + vectors: Vector[]; +} + +const fixturePath = fileURLToPath( + new URL("./duplex-frame-vectors.json", import.meta.url), +); +const fixture = JSON.parse(readFileSync(fixturePath, "utf8")) as Fixture; + +// Cut one UTF-8 byte stream into chunks at the byte offsets. The decoder must +// keep partial bytes between chunks, so the split can fall inside a multi-byte +// character. +function toChunks(bytes: string, offsets: number[] | undefined): Buffer[] { + const buffer = Buffer.from(bytes, "utf8"); + if (!offsets || offsets.length === 0) return [buffer]; + const bounds = [0, ...offsets, buffer.length]; + const chunks: Buffer[] = []; + for (let i = 0; i < bounds.length - 1; i += 1) { + chunks.push(buffer.subarray(bounds[i], bounds[i + 1])); + } + return chunks; +} + +function pushAll(decoder: DuplexFrameDecoder, chunks: Buffer[]): DuplexDecodeResult[] { + const results: DuplexDecodeResult[] = []; + for (const chunk of chunks) results.push(...decoder.push(chunk)); + return results; +} + +function assertMatches(results: DuplexDecodeResult[], expected: ExpectedResult[]): void { + expect(results).toHaveLength(expected.length); + results.forEach((result, index) => { + const want = expected[index]; + if ("frame" in want) { + expect(result.ok).toBe(true); + if (result.ok) expect(result.frame).toEqual(want.frame); + } else { + expect(result.ok).toBe(false); + if (!result.ok) expect(result.error.code).toBe(want.error); + } + }); +} + +describe("duplex frame codec fixture", () => { + it("the fixture version matches the codec version", () => { + expect(fixture.frameVersion).toBe(DUPLEX_FRAME_VERSION); + expect(fixture.defaultMaxFrameBytes).toBe(DEFAULT_MAX_DUPLEX_FRAME_BYTES); + }); + + it("every category has at least one vector", () => { + const categories = new Set(fixture.vectors.map((vector) => vector.category)); + for (const category of ["valid", "invalid", "partial", "oversized", "versionMismatch"]) { + expect(categories).toContain(category); + } + }); + + for (const vector of fixture.vectors) { + it(`decodes the ${vector.name} vector`, () => { + const decoder = new DuplexFrameDecoder( + vector.maxFrameBytes ? { maxFrameBytes: vector.maxFrameBytes } : undefined, + ); + const chunks = toChunks(vector.bytes, vector.splitByteOffsets); + const results = pushAll(decoder, chunks); + assertMatches(results, vector.expected); + }); + } +}); + +describe("round-trip", () => { + const validVectors = fixture.vectors.filter((vector) => vector.roundTrip); + + it("has round-trip vectors for every frame type", () => { + const types = new Set( + validVectors.flatMap((vector) => + vector.expected.flatMap((result) => + "frame" in result ? [result.frame.type] : [], + ), + ), + ); + for (const type of ["request", "response", "ready", "heartbeat", "close", "error"]) { + expect(types).toContain(type); + } + }); + + for (const vector of validVectors) { + it(`re-encodes the ${vector.name} frame to the same value`, () => { + const want = vector.expected[0]; + expect("frame" in want).toBe(true); + if (!("frame" in want)) return; + const encoded = encodeDuplexFrame(want.frame); + // Encode writes exactly one line: it ends with one newline and holds no + // interior newline, so one frame stays on one line. + expect(encoded.endsWith("\n")).toBe(true); + expect(encoded.slice(0, -1)).not.toContain("\n"); + const decoded = decodeDuplexLine(encoded.slice(0, -1)); + expect(decoded.ok).toBe(true); + if (decoded.ok) expect(decoded.frame).toEqual(want.frame); + }); + } +}); + +describe("streaming decoder behavior", () => { + it("emits nothing until a full line arrives, then the complete frame", () => { + const decoder = new DuplexFrameDecoder(); + const line = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }); + const bytes = Buffer.from(line, "utf8"); + const first = decoder.push(bytes.subarray(0, 3)); + expect(first).toHaveLength(0); + const second = decoder.push(bytes.subarray(3)); + expect(second).toHaveLength(1); + expect(second[0].ok).toBe(true); + }); + + it("keeps a multi-byte UTF-8 sequence valid across a chunk boundary", () => { + const decoder = new DuplexFrameDecoder(); + const frame: DuplexFrame = { + version: DUPLEX_FRAME_VERSION, + type: "request", + id: "req-emoji", + method: "POST", + path: "/x", + query: "", + headers: {}, + body: "😀", + }; + const bytes = Buffer.from(encodeDuplexFrame(frame), "utf8"); + // Split inside the four-byte emoji, right before a continuation byte. + let cut = -1; + for (let i = 1; i < bytes.length; i += 1) { + if ((bytes[i] & 0xc0) === 0x80) { + cut = i; + break; + } + } + expect(cut).toBeGreaterThan(0); + const results = [ + ...decoder.push(bytes.subarray(0, cut)), + ...decoder.push(bytes.subarray(cut)), + ]; + expect(results).toHaveLength(1); + expect(results[0].ok).toBe(true); + if (results[0].ok) expect(results[0].frame).toEqual(frame); + }); + + it("rejects an oversized frame with a protocol error, then resynchronizes", () => { + const decoder = new DuplexFrameDecoder({ maxFrameBytes: 32 }); + const flood = `${"z".repeat(100)}\n`; + const good = encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }); + const results = decoder.push(Buffer.from(flood + good, "utf8")); + expect(results).toHaveLength(2); + expect(results[0].ok).toBe(false); + if (!results[0].ok) expect(results[0].error.code).toBe("frame_too_large"); + expect(results[1].ok).toBe(true); + }); + + it("rejects a version-mismatch frame with a protocol error", () => { + const decoder = new DuplexFrameDecoder(); + const results = decoder.push( + Buffer.from(`${JSON.stringify({ version: 999, type: "heartbeat" })}\n`, "utf8"), + ); + expect(results).toHaveLength(1); + expect(results[0].ok).toBe(false); + if (!results[0].ok) expect(results[0].error.code).toBe("version_mismatch"); + }); + + it("never throws on a malformed read; it returns a protocol error", () => { + const decoder = new DuplexFrameDecoder(); + expect(() => decoder.push(Buffer.from("this is not json\n", "utf8"))).not.toThrow(); + const results = decoder.push(Buffer.from("still not json\n", "utf8")); + expect(results[0].ok).toBe(false); + }); +}); diff --git a/packages/adapter-utils/src/duplex-frame-codec.ts b/packages/adapter-utils/src/duplex-frame-codec.ts new file mode 100644 index 0000000000..2cc4d57c98 --- /dev/null +++ b/packages/adapter-utils/src/duplex-frame-codec.ts @@ -0,0 +1,338 @@ +/** + * Versioned frame codec for the sandbox duplex channel. + * + * The channel carries newline-delimited JSON frames. One frame is one line. The + * host and the generated gateway each hold a copy of this codec. A shared + * fixture file (`duplex-frame-vectors.json`) proves the two copies stay wire + * compatible: both copies decode the same bytes to the same frames. + * + * The codec has two sides: + * - encode: turn a frame object into one line of JSON with a trailing newline. + * - decode: turn a byte stream into frames. The streaming decoder keeps partial + * bytes between chunks, so a frame split across chunks and a multi-byte UTF-8 + * sequence split across chunks both decode correctly. + * + * The decoder never throws on the read path. A malformed, oversized, or + * version-mismatch frame becomes a protocol-error result, not an exception. This + * keeps one bad frame from crashing the read loop. + */ + +/** The wire version this codec reads and writes. */ +export const DUPLEX_FRAME_VERSION = 1; + +/** + * The default maximum size of one frame, in bytes. The decoder rejects a longer + * frame with a `frame_too_large` protocol error. The value matches the per-chunk + * character bound of the host duplex route. + */ +export const DEFAULT_MAX_DUPLEX_FRAME_BYTES = 1_000_000; + +const NEWLINE_BYTE = 0x0a; +const EMPTY = Buffer.alloc(0); + +/** The frame type strings. One value goes in the `type` field of every frame. */ +export const DUPLEX_FRAME_TYPES = { + request: "request", + response: "response", + ready: "ready", + heartbeat: "heartbeat", + close: "close", + error: "error", +} as const; + +/** The outcome of a response frame. A loss response carries a non-completed outcome. */ +export type DuplexResponseOutcome = "completed" | "indeterminate" | "unavailable"; + +/** A request frame. The gateway forwards it to the host API path. */ +export interface DuplexRequestFrame { + version: number; + type: "request"; + id: string; + method: string; + path: string; + query: string; + headers: Record; + body: string; +} + +/** A response frame. The host returns it for one request id. */ +export interface DuplexResponseFrame { + version: number; + type: "response"; + id: string; + status: number; + headers: Record; + body: string; + outcome: DuplexResponseOutcome; +} + +/** + * The READY control frame. The gateway sends it one time after it validates its + * local listener address. The `address` field carries that validated address. + */ +export interface DuplexReadyFrame { + version: number; + type: "ready"; + address: string; +} + +/** The heartbeat control frame. Each side sends it on an interval to prove liveness. */ +export interface DuplexHeartbeatFrame { + version: number; + type: "heartbeat"; +} + +/** The orderly close control frame. A side sends it to end the channel cleanly. */ +export interface DuplexCloseFrame { + version: number; + type: "close"; +} + +/** + * The protocol-error control frame. A peer sends it to report a bad frame. This + * frame is distinct from a decode-time protocol error: the decoder produces a + * {@link DuplexProtocolError} result, while a peer sends this frame on the wire. + */ +export interface DuplexErrorFrame { + version: number; + type: "error"; + code: string; + message?: string; +} + +/** Any frame the codec reads or writes. */ +export type DuplexFrame = + | DuplexRequestFrame + | DuplexResponseFrame + | DuplexReadyFrame + | DuplexHeartbeatFrame + | DuplexCloseFrame + | DuplexErrorFrame; + +/** The reason the decoder rejected one line. */ +export type DuplexProtocolErrorCode = + | "malformed_frame" + | "unknown_type" + | "version_mismatch" + | "frame_too_large"; + +/** A decode-time protocol error. The read path returns it; it never throws. */ +export interface DuplexProtocolError { + code: DuplexProtocolErrorCode; + message: string; +} + +/** One decode result: a valid frame, or a protocol error. */ +export type DuplexDecodeResult = + | { ok: true; frame: DuplexFrame } + | { ok: false; error: DuplexProtocolError }; + +const RESPONSE_OUTCOMES: ReadonlySet = new Set([ + "completed", + "indeterminate", + "unavailable", +]); + +function ok(frame: DuplexFrame): DuplexDecodeResult { + return { ok: true, frame }; +} + +function fail(code: DuplexProtocolErrorCode, message: string): DuplexDecodeResult { + return { ok: false, error: { code, message } }; +} + +function isPlainObject(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function isStringRecord(value: unknown): value is Record { + if (!isPlainObject(value)) return false; + for (const entry of Object.values(value)) { + if (typeof entry !== "string") return false; + } + return true; +} + +/** + * Encode one frame to a single line of JSON with a trailing newline. `JSON.stringify` + * escapes any newline inside a string value, so the returned line holds no + * interior newline. This keeps one frame on one line. + */ +export function encodeDuplexFrame(frame: DuplexFrame): string { + return `${JSON.stringify(frame)}\n`; +} + +/** + * Decode one line (no trailing newline) to a frame or a protocol error. The + * streaming decoder calls this for each complete line. It is exported so a + * caller with its own line splitter can reuse the same validation. + * + * The check order matters. A parseable frame with the wrong version becomes a + * `version_mismatch`, so the version check runs before the type check. + */ +export function decodeDuplexLine(line: string | Buffer): DuplexDecodeResult { + const text = typeof line === "string" ? line : line.toString("utf8"); + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return fail("malformed_frame", "frame is not valid JSON"); + } + if (!isPlainObject(parsed)) { + return fail("malformed_frame", "frame is not a JSON object"); + } + if (parsed.version !== DUPLEX_FRAME_VERSION) { + return fail( + "version_mismatch", + `frame version ${String(parsed.version)} is not ${DUPLEX_FRAME_VERSION}`, + ); + } + return validateFrame(parsed); +} + +function validateFrame(frame: Record): DuplexDecodeResult { + switch (frame.type) { + case "request": + return validateRequest(frame); + case "response": + return validateResponse(frame); + case "ready": + return validateReady(frame); + case "heartbeat": + return validateHeartbeat(frame); + case "close": + return validateClose(frame); + case "error": + return validateError(frame); + default: + return fail("unknown_type", `unknown frame type ${JSON.stringify(frame.type)}`); + } +} + +function validateRequest(frame: Record): DuplexDecodeResult { + if ( + typeof frame.id !== "string" || + typeof frame.method !== "string" || + typeof frame.path !== "string" || + typeof frame.query !== "string" || + typeof frame.body !== "string" || + !isStringRecord(frame.headers) + ) { + return fail("malformed_frame", "request frame has a missing or wrong-typed field"); + } + return ok(frame as unknown as DuplexRequestFrame); +} + +function validateResponse(frame: Record): DuplexDecodeResult { + if ( + typeof frame.id !== "string" || + typeof frame.status !== "number" || + typeof frame.body !== "string" || + !isStringRecord(frame.headers) || + typeof frame.outcome !== "string" || + !RESPONSE_OUTCOMES.has(frame.outcome) + ) { + return fail("malformed_frame", "response frame has a missing or wrong-typed field"); + } + return ok(frame as unknown as DuplexResponseFrame); +} + +function validateReady(frame: Record): DuplexDecodeResult { + if (typeof frame.address !== "string") { + return fail("malformed_frame", "ready frame has a missing or wrong-typed address"); + } + return ok(frame as unknown as DuplexReadyFrame); +} + +function validateHeartbeat(frame: Record): DuplexDecodeResult { + return ok(frame as unknown as DuplexHeartbeatFrame); +} + +function validateClose(frame: Record): DuplexDecodeResult { + return ok(frame as unknown as DuplexCloseFrame); +} + +function validateError(frame: Record): DuplexDecodeResult { + if (typeof frame.code !== "string") { + return fail("malformed_frame", "error frame has a missing or wrong-typed code"); + } + if (frame.message !== undefined && typeof frame.message !== "string") { + return fail("malformed_frame", "error frame has a wrong-typed message"); + } + return ok(frame as unknown as DuplexErrorFrame); +} + +/** Options for a {@link DuplexFrameDecoder}. */ +export interface DuplexFrameDecoderOptions { + /** The maximum size of one frame, in bytes. Defaults to {@link DEFAULT_MAX_DUPLEX_FRAME_BYTES}. */ + maxFrameBytes?: number; +} + +/** + * A streaming decoder for a byte stream of newline-delimited JSON frames. + * + * Call `push` with each chunk. The decoder keeps the bytes of an incomplete + * frame between calls, so a frame that spans two chunks decodes on the chunk + * that completes it. The decoder buffers raw bytes and decodes UTF-8 only on a + * complete line, so a multi-byte sequence split across chunks stays valid. The + * newline byte `0x0A` never appears inside a multi-byte UTF-8 sequence, so a + * split on that byte is always safe. + * + * The decoder enforces the maximum frame size. It rejects an oversized frame + * with a `frame_too_large` protocol error, then discards bytes up to the next + * newline to resynchronize. It never throws on the read path. + */ +export class DuplexFrameDecoder { + private buffer: Buffer = EMPTY; + private discarding = false; + private readonly maxFrameBytes: number; + + constructor(options: DuplexFrameDecoderOptions = {}) { + this.maxFrameBytes = options.maxFrameBytes ?? DEFAULT_MAX_DUPLEX_FRAME_BYTES; + } + + /** Feed one chunk. Return the frames and protocol errors that complete on it. */ + push(chunk: Buffer | Uint8Array | string): DuplexDecodeResult[] { + 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: DuplexDecodeResult[] = []; + for (;;) { + if (this.discarding) { + // Drop the tail of an oversized frame until the next newline. + const newlineIndex = this.buffer.indexOf(NEWLINE_BYTE); + if (newlineIndex === -1) { + this.buffer = EMPTY; + break; + } + this.buffer = this.buffer.subarray(newlineIndex + 1); + this.discarding = false; + continue; + } + + const newlineIndex = this.buffer.indexOf(NEWLINE_BYTE); + if (newlineIndex === -1) { + // No complete line yet. Reject an incomplete frame that already passed + // the size bound, then resynchronize at the next newline. + if (this.buffer.length > this.maxFrameBytes) { + results.push(fail("frame_too_large", "frame exceeds the maximum size")); + this.discarding = true; + this.buffer = EMPTY; + } + break; + } + + const line = this.buffer.subarray(0, newlineIndex); + this.buffer = this.buffer.subarray(newlineIndex + 1); + if (line.length === 0) continue; // Skip a blank line. + if (line.length > this.maxFrameBytes) { + results.push(fail("frame_too_large", "frame exceeds the maximum size")); + continue; + } + results.push(decodeDuplexLine(line)); + } + return results; + } +} diff --git a/packages/adapter-utils/src/duplex-frame-vectors.json b/packages/adapter-utils/src/duplex-frame-vectors.json new file mode 100644 index 0000000000..e1e07fa0c1 --- /dev/null +++ b/packages/adapter-utils/src/duplex-frame-vectors.json @@ -0,0 +1,358 @@ +{ + "frameVersion": 1, + "defaultMaxFrameBytes": 1000000, + "description": "Shared wire-compatibility vectors for the duplex frame codec. Every codec copy decodes the same bytes. bytes is a UTF-8 byte stream; splitByteOffsets cuts it into chunks; expected lists the ordered decode results (frame or protocol-error code).", + "vectors": [ + { + "name": "valid-request", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"body\":\"\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "request", + "id": "req-1", + "method": "GET", + "path": "/api/issues/PAP-1", + "query": "expand=comments", + "headers": { + "accept": "application/json", + "content-type": "application/json" + }, + "body": "" + } + } + ] + }, + { + "name": "valid-response-completed", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"content-type\":\"application/json\"},\"body\":\"{\\\"ok\\\":true}\",\"outcome\":\"completed\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "response", + "id": "req-1", + "status": 200, + "headers": { + "content-type": "application/json" + }, + "body": "{\"ok\":true}", + "outcome": "completed" + } + } + ] + }, + { + "name": "valid-response-indeterminate", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"response\",\"id\":\"req-2\",\"status\":409,\"headers\":{\"content-type\":\"application/json\"},\"body\":\"{\\\"error\\\":\\\"outcome_indeterminate\\\"}\",\"outcome\":\"indeterminate\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "response", + "id": "req-2", + "status": 409, + "headers": { + "content-type": "application/json" + }, + "body": "{\"error\":\"outcome_indeterminate\"}", + "outcome": "indeterminate" + } + } + ] + }, + { + "name": "valid-response-unavailable", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"response\",\"id\":\"req-3\",\"status\":503,\"headers\":{\"content-type\":\"application/json\"},\"body\":\"{\\\"error\\\":\\\"bridge_unavailable\\\"}\",\"outcome\":\"unavailable\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "response", + "id": "req-3", + "status": 503, + "headers": { + "content-type": "application/json" + }, + "body": "{\"error\":\"bridge_unavailable\"}", + "outcome": "unavailable" + } + } + ] + }, + { + "name": "valid-ready", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"ready\",\"address\":\"127.0.0.1:47215\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "ready", + "address": "127.0.0.1:47215" + } + } + ] + }, + { + "name": "valid-heartbeat", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"heartbeat\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "heartbeat" + } + } + ] + }, + { + "name": "valid-close", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"close\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "close" + } + } + ] + }, + { + "name": "valid-error", + "category": "valid", + "bytes": "{\"version\":1,\"type\":\"error\",\"code\":\"malformed_frame\",\"message\":\"peer reported a malformed frame\"}\n", + "roundTrip": true, + "expected": [ + { + "frame": { + "version": 1, + "type": "error", + "code": "malformed_frame", + "message": "peer reported a malformed frame" + } + } + ] + }, + { + "name": "invalid-not-json", + "category": "invalid", + "bytes": "{not valid json\n", + "expected": [ + { + "error": "malformed_frame" + } + ] + }, + { + "name": "invalid-json-array", + "category": "invalid", + "bytes": "[1,2,3]\n", + "expected": [ + { + "error": "malformed_frame" + } + ] + }, + { + "name": "invalid-json-number", + "category": "invalid", + "bytes": "42\n", + "expected": [ + { + "error": "malformed_frame" + } + ] + }, + { + "name": "invalid-unknown-type", + "category": "invalid", + "bytes": "{\"version\":1,\"type\":\"teleport\"}\n", + "expected": [ + { + "error": "unknown_type" + } + ] + }, + { + "name": "invalid-missing-id", + "category": "invalid", + "bytes": "{\"version\":1,\"type\":\"request\",\"method\":\"GET\",\"path\":\"/\",\"query\":\"\",\"headers\":{},\"body\":\"\"}\n", + "expected": [ + { + "error": "malformed_frame" + } + ] + }, + { + "name": "invalid-bad-headers", + "category": "invalid", + "bytes": "{\"version\":1,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"accept\":7},\"body\":\"{\\\"ok\\\":true}\",\"outcome\":\"completed\"}\n", + "expected": [ + { + "error": "malformed_frame" + } + ] + }, + { + "name": "partial-request-split", + "category": "partial", + "bytes": "{\"version\":1,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"body\":\"\"}\n", + "splitByteOffsets": [ + 98 + ], + "expected": [ + { + "frame": { + "version": 1, + "type": "request", + "id": "req-1", + "method": "GET", + "path": "/api/issues/PAP-1", + "query": "expand=comments", + "headers": { + "accept": "application/json", + "content-type": "application/json" + }, + "body": "" + } + } + ] + }, + { + "name": "partial-utf8-split", + "category": "partial", + "bytes": "{\"version\":1,\"type\":\"request\",\"id\":\"req-utf8\",\"method\":\"POST\",\"path\":\"/api/issues/PAP-1/comments\",\"query\":\"\",\"headers\":{\"content-type\":\"application/json\"},\"body\":\"{\\\"text\\\":\\\"héllo 😀 wörld\\\"}\"}\n", + "splitByteOffsets": [ + 177 + ], + "expected": [ + { + "frame": { + "version": 1, + "type": "request", + "id": "req-utf8", + "method": "POST", + "path": "/api/issues/PAP-1/comments", + "query": "", + "headers": { + "content-type": "application/json" + }, + "body": "{\"text\":\"héllo 😀 wörld\"}" + } + } + ] + }, + { + "name": "partial-two-frames", + "category": "partial", + "bytes": "{\"version\":1,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"body\":\"\"}\n{\"version\":1,\"type\":\"response\",\"id\":\"req-1\",\"status\":200,\"headers\":{\"content-type\":\"application/json\"},\"body\":\"{\\\"ok\\\":true}\",\"outcome\":\"completed\"}\n", + "splitByteOffsets": [ + 7, + 199, + 343 + ], + "expected": [ + { + "frame": { + "version": 1, + "type": "request", + "id": "req-1", + "method": "GET", + "path": "/api/issues/PAP-1", + "query": "expand=comments", + "headers": { + "accept": "application/json", + "content-type": "application/json" + }, + "body": "" + } + }, + { + "frame": { + "version": 1, + "type": "response", + "id": "req-1", + "status": 200, + "headers": { + "content-type": "application/json" + }, + "body": "{\"ok\":true}", + "outcome": "completed" + } + } + ] + }, + { + "name": "partial-no-trailing-newline", + "category": "partial", + "bytes": "{\"version\":1,\"type\":\"heartbeat\"}", + "expected": [] + }, + { + "name": "oversized-complete-line", + "category": "oversized", + "maxFrameBytes": 64, + "bytes": "{\"version\":1,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"body\":\"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx\"}\n", + "expected": [ + { + "error": "frame_too_large" + } + ] + }, + { + "name": "oversized-flood-then-resync", + "category": "oversized", + "maxFrameBytes": 64, + "splitByteOffsets": [ + 40 + ], + "bytes": "yyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyyy\n{\"version\":1,\"type\":\"heartbeat\"}\n", + "expected": [ + { + "error": "frame_too_large" + }, + { + "frame": { + "version": 1, + "type": "heartbeat" + } + } + ] + }, + { + "name": "version-mismatch-higher", + "category": "versionMismatch", + "bytes": "{\"version\":2,\"type\":\"request\",\"id\":\"req-1\",\"method\":\"GET\",\"path\":\"/api/issues/PAP-1\",\"query\":\"expand=comments\",\"headers\":{\"accept\":\"application/json\",\"content-type\":\"application/json\"},\"body\":\"\"}\n", + "expected": [ + { + "error": "version_mismatch" + } + ] + }, + { + "name": "version-mismatch-missing", + "category": "versionMismatch", + "bytes": "{\"type\":\"heartbeat\"}\n", + "expected": [ + { + "error": "version_mismatch" + } + ] + } + ] +} diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 9cb50e7dc0..8c1aba3634 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/packages/adapter-utils/src/execution-target.ts @@ -116,6 +116,7 @@ export interface EffectiveSandboxCapabilities { readonly independentControlCommands: boolean; readonly incrementalSessionOutput: boolean; readonly concurrentSyncOperations: boolean; + readonly duplexCommandStream: boolean; } export interface AdapterSandboxExecutionTarget extends AdapterExecutionTargetWorkspaceMetadata { @@ -256,6 +257,7 @@ function parseEffectiveSandboxCapabilities(value: unknown): EffectiveSandboxCapa independentControlCommands: parsed.independentControlCommands === true, incrementalSessionOutput: parsed.incrementalSessionOutput === true, concurrentSyncOperations: parsed.concurrentSyncOperations === true, + duplexCommandStream: parsed.duplexCommandStream === true, }; } diff --git a/packages/plugins/sdk/src/define-plugin.ts b/packages/plugins/sdk/src/define-plugin.ts index c8af779839..69c9a1ca6e 100644 --- a/packages/plugins/sdk/src/define-plugin.ts +++ b/packages/plugins/sdk/src/define-plugin.ts @@ -88,6 +88,12 @@ import type { PluginSetupTokenPtyStopParams, PluginSetupTokenPtyCloseParams, PluginSetupTokenPtyCloseResult, + PluginDuplexChannelOpenParams, + PluginDuplexChannelOpenResult, + PluginDuplexChannelWriteParams, + PluginDuplexChannelStopParams, + PluginDuplexChannelCloseParams, + PluginDuplexChannelCloseResult, } from "./protocol.js"; // --------------------------------------------------------------------------- @@ -464,6 +470,33 @@ export interface PluginDefinition { onSetupTokenPtyClose?( params: PluginSetupTokenPtyCloseParams, ): Promise; + + /** + * Called to open one persistent duplex channel. The worker registers the + * channel under the host route identifier and returns a worker session + * identifier for the data notification binding only. The worker streams data + * and the exit through worker→host notifications, never as a reply. Defining + * the four `onDuplexChannel*` hooks advertises the four methods. The host reads + * the open verb to gate the `duplexCommandStream` capability. + */ + onDuplexChannelOpen?( + params: PluginDuplexChannelOpenParams, + ): Promise; + + /** Called to write raw input to an open duplex channel, keyed by the worker session identifier. */ + onDuplexChannelWrite?(params: PluginDuplexChannelWriteParams): Promise; + + /** Called to stop an open duplex channel child, keyed by the worker session identifier. */ + onDuplexChannelStop?(params: PluginDuplexChannelStopParams): Promise; + + /** + * Called to close an open duplex channel by the host route identifier. The + * worker closes the exact channel registered under that identifier and returns + * a close acknowledgement that carries the same identifier. + */ + onDuplexChannelClose?( + params: PluginDuplexChannelCloseParams, + ): Promise; } // --------------------------------------------------------------------------- diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index c34b1a4d24..d5482a46ad 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -84,6 +84,8 @@ export { JsonRpcCallError, SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION, SETUP_TOKEN_PTY_EXIT_NOTIFICATION, + DUPLEX_CHANNEL_DATA_NOTIFICATION, + DUPLEX_CHANNEL_EXIT_NOTIFICATION, _resetIdCounter, } from "./protocol.js"; diff --git a/packages/plugins/sdk/src/protocol.duplex-channel.test.ts b/packages/plugins/sdk/src/protocol.duplex-channel.test.ts new file mode 100644 index 0000000000..306d7646c5 --- /dev/null +++ b/packages/plugins/sdk/src/protocol.duplex-channel.test.ts @@ -0,0 +1,156 @@ +import { describe, expect, it } from "vitest"; + +import { + DUPLEX_CHANNEL_DATA_NOTIFICATION, + DUPLEX_CHANNEL_EXIT_NOTIFICATION, + type PluginDuplexChannelCloseParams, + type PluginDuplexChannelCloseResult, + type PluginDuplexChannelDataParams, + type PluginDuplexChannelExitParams, + type PluginDuplexChannelOpenParams, + type PluginDuplexChannelOpenResult, + type PluginDuplexChannelStopParams, + type PluginDuplexChannelWriteParams, +} from "./protocol.js"; + +// The generic duplex channel messages model the setup-token pseudo-terminal +// contract: open, write, stop, and close requests, plus data and exit +// notifications. The host owns the route identifier. The worker returns a worker +// session identifier that binds the data and the exit notification only. A close +// keys on the host route identifier, so a lost open reply still permits a +// host-keyed close. +// +// The messages are static types, so a valid payload assigns to its type and an +// invalid payload fails the type check. The `@ts-expect-error` directives make +// the compiler reject each invalid payload. The `tsc --noEmit` check validates +// the directives, so an invalid payload that the type accepts fails the build. + +describe("duplex channel request schemas", () => { + it("accepts a valid open request and its reply", () => { + const open: PluginDuplexChannelOpenParams = { + hostRouteId: "route-1", + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "lease-1", + command: "paperclip-bridge", + }; + const reply: PluginDuplexChannelOpenResult = { workerSessionId: "ws-1" }; + expect(open.hostRouteId).toBe("route-1"); + expect(reply.workerSessionId).toBe("ws-1"); + }); + + it("rejects an open request that omits the host route identifier", () => { + // @ts-expect-error — hostRouteId is required. + const open: PluginDuplexChannelOpenParams = { + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "lease-1", + command: "paperclip-bridge", + }; + expect(open).toBeDefined(); + }); + + it("rejects an open request with a non-string command", () => { + 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, + }; + expect(open).toBeDefined(); + }); + + it("accepts a valid write request", () => { + const write: PluginDuplexChannelWriteParams = { + workerSessionId: "ws-1", + data: "payload", + }; + expect(write.data).toBe("payload"); + }); + + it("rejects a write request that omits the data", () => { + // @ts-expect-error — data is required. + const write: PluginDuplexChannelWriteParams = { workerSessionId: "ws-1" }; + expect(write).toBeDefined(); + }); + + it("accepts a valid stop request", () => { + const stop: PluginDuplexChannelStopParams = { workerSessionId: "ws-1" }; + expect(stop.workerSessionId).toBe("ws-1"); + }); + + it("rejects a stop request that omits the worker session identifier", () => { + // @ts-expect-error — workerSessionId is required. + const stop: PluginDuplexChannelStopParams = {}; + expect(stop).toBeDefined(); + }); + + it("accepts a close request keyed only by the host route identifier", () => { + const close: PluginDuplexChannelCloseParams = { hostRouteId: "route-1" }; + const reply: PluginDuplexChannelCloseResult = { hostRouteId: "route-1" }; + expect(close.hostRouteId).toBe("route-1"); + expect(reply.hostRouteId).toBe("route-1"); + }); + + it("accepts a close request that also carries the worker session identifier", () => { + const close: PluginDuplexChannelCloseParams = { + hostRouteId: "route-1", + workerSessionId: "ws-1", + }; + expect(close.workerSessionId).toBe("ws-1"); + }); + + it("rejects a close request that omits the host route identifier", () => { + // @ts-expect-error — hostRouteId is the authoritative close key and is required. + const close: PluginDuplexChannelCloseParams = { workerSessionId: "ws-1" }; + expect(close).toBeDefined(); + }); +}); + +describe("duplex channel notification schemas", () => { + it("uses the generic notification method names", () => { + expect(DUPLEX_CHANNEL_DATA_NOTIFICATION).toBe("duplexChannel.data"); + expect(DUPLEX_CHANNEL_EXIT_NOTIFICATION).toBe("duplexChannel.exit"); + }); + + it("accepts a valid data notification", () => { + const data: PluginDuplexChannelDataParams = { + workerSessionId: "ws-1", + chunk: "output bytes", + }; + expect(data.chunk).toBe("output bytes"); + }); + + it("rejects a data notification that omits the chunk", () => { + // @ts-expect-error — chunk is required. + const data: PluginDuplexChannelDataParams = { workerSessionId: "ws-1" }; + expect(data).toBeDefined(); + }); + + it("accepts an exit notification with a numeric code and with null", () => { + const exit: PluginDuplexChannelExitParams = { + workerSessionId: "ws-1", + exitCode: 0, + }; + const exitNull: PluginDuplexChannelExitParams = { + workerSessionId: "ws-1", + exitCode: null, + }; + expect(exit.exitCode).toBe(0); + expect(exitNull.exitCode).toBeNull(); + }); + + it("rejects an exit notification with a non-numeric, non-null exit code", () => { + const exit: PluginDuplexChannelExitParams = { + workerSessionId: "ws-1", + // @ts-expect-error — exitCode must be a number or null. + exitCode: "0", + }; + expect(exit).toBeDefined(); + }); +}); diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 519770768d..72585f7324 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -1082,6 +1082,102 @@ export const SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION = "setupTokenPty.output"; /** The worker→host notification method for one pseudo-terminal exit. */ export const SETUP_TOKEN_PTY_EXIT_NOTIFICATION = "setupTokenPty.exit"; +// --------------------------------------------------------------------------- +// Generic duplex channel worker methods. +// --------------------------------------------------------------------------- +// The host drives one persistent duplex channel inside a sandbox provider +// worker. The channel replaces the file transport of the sandbox callback bridge +// with one live bidirectional stream. These messages are generic. They model the +// setup-token pseudo-terminal contract above, but they carry no login command +// allowlist. The host owns the route. It mints an opaque host route identifier, +// carries that identifier in the open request, and keys the close on that +// identifier. The worker registers the channel under the host route identifier +// and returns a worker session identifier for the data and the exit notification +// binding only. The worker never keys a close on the worker session identifier, +// so the host closes a worker-created channel even when the open reply was lost +// and no worker session identifier arrived. The worker sends data and exit as +// notifications, never as a reply, so the host binds them by the worker session +// identifier while the route is open. + +/** The open request for one persistent duplex channel. The worker registers the channel by `hostRouteId`. */ +export interface PluginDuplexChannelOpenParams { + /** The host-owned opaque route identifier. The worker registers the channel by it. */ + hostRouteId: string; + /** The environment driver key, for the worker sandbox scope. */ + driverKey: string; + /** The company that owns the channel. */ + companyId: string; + /** The environment the channel runs in. */ + 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 open reply. It returns the worker session identifier for data binding only. */ +export interface PluginDuplexChannelOpenResult { + /** The worker session identifier. It binds the data and the exit notification only. */ + workerSessionId: string; +} + +/** The write request. It carries the worker session identifier and the raw input bytes. */ +export interface PluginDuplexChannelWriteParams { + /** The worker session identifier that the open reply returned. */ + workerSessionId: string; + /** The raw input bytes to write to the channel. */ + data: string; +} + +/** The stop request. It carries the worker session identifier. */ +export interface PluginDuplexChannelStopParams { + /** The worker session identifier that the open reply returned. */ + workerSessionId: string; +} + +/** The close request. The host route identifier is the authoritative key. */ +export interface PluginDuplexChannelCloseParams { + /** + * The host-owned opaque route identifier. This is the authoritative close key, + * so the host closes the channel even when no worker session identifier arrived + * after a lost open reply. + */ + hostRouteId: string; + /** + * A non-authoritative worker session identifier. The worker never keys the + * close on it. The field is optional, so a close with only the host route + * identifier is a valid request for this lifecycle. + */ + workerSessionId?: string; +} + +/** The close reply. It acknowledges the close and carries the same host route identifier. */ +export interface PluginDuplexChannelCloseResult { + /** The close acknowledgement. It carries the same host route identifier the close sent. */ + hostRouteId: string; +} + +/** The worker→host duplex channel data notification parameters. */ +export interface PluginDuplexChannelDataParams { + /** The worker session identifier that the open reply returned. */ + workerSessionId: string; + /** The raw channel output bytes. */ + chunk: string; +} + +/** The worker→host duplex channel exit notification parameters. */ +export interface PluginDuplexChannelExitParams { + /** The worker session identifier that the open reply returned. */ + workerSessionId: string; + /** The child exit code, or null when the child ended with no code. */ + exitCode: number | null; +} + +/** The worker→host notification method for one duplex channel data chunk. */ +export const DUPLEX_CHANNEL_DATA_NOTIFICATION = "duplexChannel.data"; +/** The worker→host notification method for one duplex channel exit. */ +export const DUPLEX_CHANNEL_EXIT_NOTIFICATION = "duplexChannel.exit"; + /** * Map of host→worker RPC method names to their `[params, result]` types. * @@ -1200,6 +1296,20 @@ export interface HostToWorkerMethods { params: PluginSetupTokenPtyCloseParams, result: PluginSetupTokenPtyCloseResult, ]; + /** Open one persistent duplex channel keyed by a host-owned route identifier. */ + duplexChannelOpen: [ + params: PluginDuplexChannelOpenParams, + result: PluginDuplexChannelOpenResult, + ]; + /** Write raw input to a persistent duplex channel, keyed by the worker session identifier. */ + duplexChannelWrite: [params: PluginDuplexChannelWriteParams, result: void]; + /** Stop a persistent duplex channel child, keyed by the worker session identifier. */ + duplexChannelStop: [params: PluginDuplexChannelStopParams, result: void]; + /** Close a persistent duplex channel by the host route identifier and return a bound acknowledgement. */ + duplexChannelClose: [ + params: PluginDuplexChannelCloseParams, + result: PluginDuplexChannelCloseResult, + ]; } /** Union of all host→worker method names. */ @@ -1245,6 +1355,10 @@ export const HOST_TO_WORKER_OPTIONAL_METHODS: readonly HostToWorkerMethodName[] "setupTokenPtyInput", "setupTokenPtyStop", "setupTokenPtyClose", + "duplexChannelOpen", + "duplexChannelWrite", + "duplexChannelStop", + "duplexChannelClose", ] as const; // --------------------------------------------------------------------------- diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index dc1b095d2c..c7cadbc5e9 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -104,6 +104,10 @@ import type { PluginSetupTokenPtyInputParams, PluginSetupTokenPtyStopParams, PluginSetupTokenPtyCloseParams, + PluginDuplexChannelOpenParams, + PluginDuplexChannelWriteParams, + PluginDuplexChannelStopParams, + PluginDuplexChannelCloseParams, PluginInvocationContext, WorkerToHostMethodName, WorkerToHostMethods, @@ -1624,6 +1628,18 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost case "setupTokenPtyClose": return handleSetupTokenPtyClose(params as PluginSetupTokenPtyCloseParams); + case "duplexChannelOpen": + return handleDuplexChannelOpen(params as PluginDuplexChannelOpenParams); + + case "duplexChannelWrite": + return handleDuplexChannelWrite(params as PluginDuplexChannelWriteParams); + + case "duplexChannelStop": + return handleDuplexChannelStop(params as PluginDuplexChannelStopParams); + + case "duplexChannelClose": + return handleDuplexChannelClose(params as PluginDuplexChannelCloseParams); + default: throw Object.assign( new Error(`Unknown method: ${method}`), @@ -1679,6 +1695,10 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (plugin.definition.onSetupTokenPtyInput) supportedMethods.push("setupTokenPtyInput"); if (plugin.definition.onSetupTokenPtyStop) supportedMethods.push("setupTokenPtyStop"); if (plugin.definition.onSetupTokenPtyClose) supportedMethods.push("setupTokenPtyClose"); + if (plugin.definition.onDuplexChannelOpen) supportedMethods.push("duplexChannelOpen"); + if (plugin.definition.onDuplexChannelWrite) supportedMethods.push("duplexChannelWrite"); + if (plugin.definition.onDuplexChannelStop) supportedMethods.push("duplexChannelStop"); + if (plugin.definition.onDuplexChannelClose) supportedMethods.push("duplexChannelClose"); return { ok: true, supportedMethods }; } @@ -2055,6 +2075,34 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost return plugin.definition.onSetupTokenPtyClose(params); } + async function handleDuplexChannelOpen(params: PluginDuplexChannelOpenParams) { + if (!plugin.definition.onDuplexChannelOpen) { + throw methodNotImplemented("duplexChannelOpen"); + } + return plugin.definition.onDuplexChannelOpen(params); + } + + async function handleDuplexChannelWrite(params: PluginDuplexChannelWriteParams) { + if (!plugin.definition.onDuplexChannelWrite) { + throw methodNotImplemented("duplexChannelWrite"); + } + return plugin.definition.onDuplexChannelWrite(params); + } + + async function handleDuplexChannelStop(params: PluginDuplexChannelStopParams) { + if (!plugin.definition.onDuplexChannelStop) { + throw methodNotImplemented("duplexChannelStop"); + } + return plugin.definition.onDuplexChannelStop(params); + } + + async function handleDuplexChannelClose(params: PluginDuplexChannelCloseParams) { + if (!plugin.definition.onDuplexChannelClose) { + throw methodNotImplemented("duplexChannelClose"); + } + return plugin.definition.onDuplexChannelClose(params); + } + // ----------------------------------------------------------------------- // Event filter helper // ----------------------------------------------------------------------- diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index 9282496281..0b1f336c19 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -905,3 +905,191 @@ describe("worker setup-token pseudo-terminal dispatch", () => { } }); }); + +describe("worker duplex channel dispatch", () => { + it("dispatches open, write, stop, and close, and reports the duplex methods", async () => { + const hostToWorker = new PassThrough(); + const workerToHost = new PassThrough(); + const hostReadline = createInterface({ input: workerToHost }); + const pending = new Map void>(); + let nextRequestId = 1; + + const writes: string[] = []; + let stopped = 0; + let closed = 0; + + // The plugin declares the four duplex channel handlers. The open returns a + // worker session id. The close keys on the host route id and returns a bound + // acknowledgement. + const controllablePlugin = definePlugin({ + async setup() {}, + async onDuplexChannelOpen(params) { + expect(params.hostRouteId).toBe("route-1"); + expect(params.command).toBe("paperclip-bridge"); + expect(params.providerLeaseId).toBe("lease-1"); + return { workerSessionId: "ws-1" }; + }, + async onDuplexChannelWrite(params) { + writes.push(params.data); + }, + async onDuplexChannelStop() { + stopped += 1; + }, + async onDuplexChannelClose(params) { + closed += 1; + return { hostRouteId: params.hostRouteId }; + }, + }); + + const worker = startWorkerRpcHost({ + plugin: controllablePlugin, + stdin: hostToWorker, + stdout: workerToHost, + }); + + function callWorker(method: string, params: unknown) { + const id = `host-${nextRequestId++}`; + const result = new Promise((resolve, reject) => { + pending.set(id, (response) => { + if ("error" in response && response.error) { + reject(new Error(response.error.message)); + return; + } + resolve((response as { result?: unknown }).result); + }); + }); + hostToWorker.write(serializeMessage(createRequest(method, params, id))); + return result; + } + + hostReadline.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { + pending.get(String(message.id))?.(message); + pending.delete(String(message.id)); + } + }); + + try { + await expect( + callWorker("initialize", { + manifest: { + id: "paperclip.duplex-channel", + apiVersion: 1, + version: "1.0.0", + displayName: "Duplex Channel Test", + description: "Test plugin", + author: "Paperclip", + categories: ["automation"], + capabilities: [], + entrypoints: {}, + }, + config: {}, + databaseNamespace: null, + }), + ).resolves.toMatchObject({ + ok: true, + supportedMethods: expect.arrayContaining([ + "duplexChannelOpen", + "duplexChannelWrite", + "duplexChannelStop", + "duplexChannelClose", + ]), + }); + + await expect( + callWorker("duplexChannelOpen", { + hostRouteId: "route-1", + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "lease-1", + command: "paperclip-bridge", + }), + ).resolves.toEqual({ workerSessionId: "ws-1" }); + + await callWorker("duplexChannelWrite", { workerSessionId: "ws-1", data: "payload" }); + await callWorker("duplexChannelStop", { workerSessionId: "ws-1" }); + await expect( + callWorker("duplexChannelClose", { hostRouteId: "route-1" }), + ).resolves.toEqual({ hostRouteId: "route-1" }); + + expect(writes).toEqual(["payload"]); + expect(stopped).toBe(1); + expect(closed).toBe(1); + } finally { + worker.stop(); + hostReadline.close(); + } + }); + + it("reports no duplex methods when the plugin declares no duplex handlers", async () => { + const hostToWorker = new PassThrough(); + const workerToHost = new PassThrough(); + const hostReadline = createInterface({ input: workerToHost }); + const pending = new Map void>(); + let nextRequestId = 1; + + // A plugin with no duplex handlers advertises no duplex method. The Phase 1 + // capability `duplexCommandStream` still resolves false for this provider, + // because the prerequisite verb `duplexChannelOpen` is absent. + const barePlugin = definePlugin({ + async setup() {}, + }); + + const worker = startWorkerRpcHost({ + plugin: barePlugin, + stdin: hostToWorker, + stdout: workerToHost, + }); + + function callWorker(method: string, params: unknown) { + const id = `host-${nextRequestId++}`; + const result = new Promise((resolve, reject) => { + pending.set(id, (response) => { + if ("error" in response && response.error) { + reject(new Error(response.error.message)); + return; + } + resolve((response as { result?: unknown }).result); + }); + }); + hostToWorker.write(serializeMessage(createRequest(method, params, id))); + return result; + } + + hostReadline.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { + pending.get(String(message.id))?.(message); + pending.delete(String(message.id)); + } + }); + + try { + const result = (await callWorker("initialize", { + manifest: { + id: "paperclip.bare", + apiVersion: 1, + version: "1.0.0", + displayName: "Bare Test", + description: "Test plugin", + author: "Paperclip", + categories: ["automation"], + capabilities: [], + entrypoints: {}, + }, + config: {}, + databaseNamespace: null, + })) as { supportedMethods: string[] }; + + expect(result.supportedMethods).not.toContain("duplexChannelOpen"); + expect(result.supportedMethods).not.toContain("duplexChannelWrite"); + expect(result.supportedMethods).not.toContain("duplexChannelStop"); + expect(result.supportedMethods).not.toContain("duplexChannelClose"); + } finally { + worker.stop(); + hostReadline.close(); + } + }); +}); diff --git a/packages/plugins/sdk/vitest.config.ts b/packages/plugins/sdk/vitest.config.ts index faa6d98e10..3834141eee 100644 --- a/packages/plugins/sdk/vitest.config.ts +++ b/packages/plugins/sdk/vitest.config.ts @@ -3,6 +3,6 @@ import { defineConfig } from "vitest/config"; export default defineConfig({ test: { environment: "node", - include: ["tests/**/*.test.ts"], + include: ["tests/**/*.test.ts", "src/**/*.test.ts"], }, }); diff --git a/packages/shared/src/types/plugin.ts b/packages/shared/src/types/plugin.ts index 5a599fa82a..30ff21d289 100644 --- a/packages/shared/src/types/plugin.ts +++ b/packages/shared/src/types/plugin.ts @@ -173,6 +173,17 @@ export interface SandboxProviderCapabilities { * `false`. */ concurrentSyncOperations?: boolean; + /** + * Provider opens one persistent, bidirectional duplex channel that carries the + * command stream, in place of the file transport of the callback bridge. This + * is an opt-in behavioral guarantee, not a worker-method property: a provider + * that keeps persistent sessions and runs independent control commands still + * does not carry a framed duplex stream unless it declares this key. An omitted + * key denies the capability, so the provider keeps the file bridge. Only a + * provider that declares this key `true` and whose worker verifies the duplex + * open method selects the duplex channel path. + */ + duplexCommandStream?: boolean; } export interface PluginEnvironmentDriverDeclaration { diff --git a/packages/shared/src/validators/plugin.ts b/packages/shared/src/validators/plugin.ts index b710c86c02..37b959efef 100644 --- a/packages/shared/src/validators/plugin.ts +++ b/packages/shared/src/validators/plugin.ts @@ -167,6 +167,7 @@ export const sandboxProviderCapabilitiesSchema = z.object({ independentControlCommands: z.boolean().optional(), incrementalSessionOutput: z.boolean().optional(), concurrentSyncOperations: z.boolean().optional(), + duplexCommandStream: z.boolean().optional(), }).strict(); export type SandboxProviderCapabilitiesInput = z.infer; diff --git a/server/src/__tests__/environment-execution-target-capabilities.test.ts b/server/src/__tests__/environment-execution-target-capabilities.test.ts index af1febfcaa..42d8420c7f 100644 --- a/server/src/__tests__/environment-execution-target-capabilities.test.ts +++ b/server/src/__tests__/environment-execution-target-capabilities.test.ts @@ -22,6 +22,7 @@ const SNAPSHOT: EffectiveSandboxCapabilities = { // Concurrent sync operations need BOTH sync verbs; this snapshot verified only // inbound sync, so the opt-in stays off. concurrentSyncOperations: false, + duplexCommandStream: false, }; // A snapshot that grants every capability. A test overrides one flag to prove @@ -34,6 +35,7 @@ const FULL_GRANT: EffectiveSandboxCapabilities = { independentControlCommands: true, incrementalSessionOutput: true, concurrentSyncOperations: true, + duplexCommandStream: true, }; // Build a sandbox execution target with a fixed snapshot and a fixed diff --git a/server/src/__tests__/environment-execution-target-duplex.test.ts b/server/src/__tests__/environment-execution-target-duplex.test.ts new file mode 100644 index 0000000000..7af64eb431 --- /dev/null +++ b/server/src/__tests__/environment-execution-target-duplex.test.ts @@ -0,0 +1,379 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +const { mockResolveEnvironmentDriverConfigForRuntime, mockResolvePluginSandboxProviderDriverById } = + vi.hoisted(() => ({ + mockResolveEnvironmentDriverConfigForRuntime: vi.fn(), + mockResolvePluginSandboxProviderDriverById: vi.fn(), + })); + +vi.mock("../services/environment-config.js", () => ({ + resolveEnvironmentDriverConfigForRuntime: mockResolveEnvironmentDriverConfigForRuntime, +})); + +// The centralized duplex authorization gate resolves the exact lease capability +// snapshot before it calls the driver. The plugin branch of the snapshot reads +// the pinned plugin's declaration from the database. These service tests carry no +// database, so replace only the by-id declaration resolver. The test sets the +// resolved declaration per case; every other export stays real. +vi.mock("../services/plugin-environment-driver.js", async (importActual) => ({ + ...(await importActual()), + resolvePluginSandboxProviderDriverById: mockResolvePluginSandboxProviderDriverById, +})); + +import type { EffectiveSandboxCapabilities } from "@paperclipai/adapter-utils/execution-target"; +import { createSshCommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/ssh"; +import type { Environment, EnvironmentLease } from "@paperclipai/shared"; +import { resolveEnvironmentExecutionTarget } from "../services/environment-execution-target.js"; +import { + buildSandboxCapabilityNarrowing, + environmentRuntimeService, +} from "../services/environment-runtime.js"; +import type { + EnvironmentRuntimeDriver, + EnvironmentRuntimeService, +} from "../services/environment-runtime.js"; +import type { + DuplexChannelHostSession, + PluginWorkerManager, +} from "../services/plugin-worker-manager.js"; + +// A snapshot that grants the opt-in duplex capability, plus the rest true so the +// gate reads only the duplex flag. +const DUPLEX_GRANT: EffectiveSandboxCapabilities = { + reusableLeases: true, + nativeSyncIn: true, + nativeSyncOut: true, + persistentProcessSessions: true, + independentControlCommands: true, + incrementalSessionOutput: true, + duplexCommandStream: true, +}; + +const DUPLEX_ABSENT: EffectiveSandboxCapabilities = { + ...DUPLEX_GRANT, + duplexCommandStream: false, +}; + +function flushMicrotasks(): Promise { + return new Promise((resolve) => setTimeout(resolve, 0)); +} + +// A fake worker manager whose one worker returns a controllable host session. +// The session records every host→worker call and lets the test settle the exit. +function makeFakeWorkerManager() { + let settleWait: (value: { exitCode: number | null }) => void = () => {}; + const waitPromise = new Promise<{ exitCode: number | null }>((resolve) => { + settleWait = resolve; + }); + const hostSession = { + onData: vi.fn(), + write: vi.fn(), + wait: vi.fn(() => waitPromise), + kill: vi.fn(), + close: vi.fn(async () => {}), + } satisfies DuplexChannelHostSession; + const openDuplexChannel = vi.fn(async () => hostSession); + const worker = { openDuplexChannel, supportedMethods: ["duplexChannelOpen"] }; + const manager = { + getWorker: vi.fn(() => worker), + } as unknown as PluginWorkerManager; + return { manager, worker, hostSession, openDuplexChannel, settleWait }; +} + +const PLUGIN_LEASE: EnvironmentLease = { + id: "lease-1", + companyId: "company-1", + providerLeaseId: "provider-lease-1", + metadata: { + sandboxProviderPlugin: true, + pluginId: "test.plugin", + provider: "daytona", + }, +} as unknown as EnvironmentLease; + +const SANDBOX_ENVIRONMENT: Environment = { + id: "env-1", + driver: "sandbox", +} as unknown as Environment; + +// A plugin driver declaration that grants the opt-in duplex capability. The +// centralized gate reads it through the by-id resolver and pairs it with the +// worker's verified `duplexChannelOpen` verb to grant the capability. +const DUPLEX_DECLARED_DRIVER = { + plugin: {}, + driver: { sandboxCapabilities: { duplexCommandStream: true } }, +}; + +// The fixed refusal the centralized gate throws when the lease does not grant the +// opt-in duplex capability. +const DUPLEX_DENIED = /does not grant the duplex command stream capability/; + +beforeEach(() => { + // The default resolves a clean sandbox config and a declaration that grants the + // duplex capability, so the plugin lease reaches the worker. Each refusal test + // overrides one input to deny the capability. + mockResolveEnvironmentDriverConfigForRuntime.mockReset(); + mockResolveEnvironmentDriverConfigForRuntime.mockResolvedValue({ + driver: "sandbox", + config: { provider: "daytona", timeoutMs: 30_000 }, + }); + mockResolvePluginSandboxProviderDriverById.mockReset(); + mockResolvePluginSandboxProviderDriverById.mockResolvedValue(DUPLEX_DECLARED_DRIVER); +}); + +describe("sandbox driver duplex channel wiring", () => { + it("reaches the worker manager route with the lease scope and delegates the channel members", async () => { + const { manager, worker, hostSession, settleWait } = makeFakeWorkerManager(); + const service = environmentRuntimeService({} as never, { pluginWorkerManager: manager }); + + const channel = await service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: PLUGIN_LEASE, + command: "bridge-callback", + }); + + // The driver resolves the worker by the pinned plugin id and passes the same + // lease scope the sandbox execute path uses. + expect(manager.getWorker).toHaveBeenCalledWith("test.plugin"); + expect(worker.openDuplexChannel).toHaveBeenCalledWith({ + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: "provider-lease-1", + command: "bridge-callback", + }); + + // write / stop / close map to write / kill / close on the host session. + channel.write("input-bytes"); + expect(hostSession.write).toHaveBeenCalledWith("input-bytes"); + channel.stop(); + expect(hostSession.kill).toHaveBeenCalledTimes(1); + await channel.close(); + expect(hostSession.close).toHaveBeenCalledTimes(1); + + // onData maps one to one. + const dataListener = vi.fn(); + channel.onData(dataListener); + expect(hostSession.onData).toHaveBeenCalledWith(dataListener); + + // onExit bridges the host session's one-time wait() to the exit listener. + const exitListener = vi.fn(); + channel.onExit(exitListener); + expect(exitListener).not.toHaveBeenCalled(); + settleWait({ exitCode: 7 }); + await flushMicrotasks(); + expect(exitListener).toHaveBeenCalledWith({ exitCode: 7 }); + }); + + it("refuses a lease that is not a plugin-backed sandbox lease before it reaches the worker", async () => { + const { manager, worker } = makeFakeWorkerManager(); + const service = environmentRuntimeService({} as never, { pluginWorkerManager: manager }); + const nonPluginLease = { + id: "lease-2", + companyId: "company-1", + providerLeaseId: "provider-lease-2", + metadata: { pluginId: "test.plugin", provider: "daytona" }, + } as unknown as EnvironmentLease; + + // The lease is not a plugin-backed sandbox lease, so the effective snapshot + // never grants the opt-in duplex capability. The centralized gate refuses the + // open before the driver, so the worker `duplexChannelOpen` RPC never runs. + await expect( + service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: nonPluginLease, + command: "bridge-callback", + }), + ).rejects.toThrow(DUPLEX_DENIED); + expect(worker.openDuplexChannel).not.toHaveBeenCalled(); + }); +}); + +// Direct regressions for the centralized authorization gate on +// EnvironmentRuntimeService.openDuplexChannel. The gate resolves the exact lease +// capability snapshot and refuses unless it grants the opt-in duplex capability. +// Every refusal must happen before the driver, so the worker `duplexChannelOpen` +// RPC never runs. +describe("EnvironmentRuntimeService.openDuplexChannel capability gate", () => { + it("refuses a lease whose provider does not declare the duplex capability", async () => { + const { manager, worker } = makeFakeWorkerManager(); + // The worker verifies the duplex verb, but the declaration omits the opt-in + // capability. An opt-in capability needs an explicit declaration, so the + // effective snapshot denies it. + mockResolvePluginSandboxProviderDriverById.mockResolvedValue({ plugin: {}, driver: {} }); + const service = environmentRuntimeService({} as never, { pluginWorkerManager: manager }); + + await expect( + service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: PLUGIN_LEASE, + command: "bridge-callback", + }), + ).rejects.toThrow(DUPLEX_DENIED); + expect(worker.openDuplexChannel).not.toHaveBeenCalled(); + }); + + it("refuses a lease whose worker does not verify the duplex verb", async () => { + const { manager, worker } = makeFakeWorkerManager(); + // The declaration grants the capability, but the worker does not advertise + // the `duplexChannelOpen` verb, so the runtime never verified it. A + // declaration never grants an unverified capability. + worker.supportedMethods = []; + const service = environmentRuntimeService({} as never, { pluginWorkerManager: manager }); + + await expect( + service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: PLUGIN_LEASE, + command: "bridge-callback", + }), + ).rejects.toThrow(DUPLEX_DENIED); + expect(worker.openDuplexChannel).not.toHaveBeenCalled(); + }); + + it("refuses a lease narrowed away from the duplex capability", async () => { + // A fake driver returns a snapshot that grants every capability except the + // duplex one, which models a lease narrowed away from it. The gate keys on + // the exact duplex flag, so it refuses even though the other capabilities are + // granted, and never calls the driver open. + const openDuplexChannel = vi.fn(); + const narrowedDriver = { + driver: "sandbox", + acquireRunLease: vi.fn(), + releaseRunLease: vi.fn(), + effectiveSandboxCapabilities: vi.fn(async () => ({ ...DUPLEX_ABSENT })), + openDuplexChannel, + } as unknown as EnvironmentRuntimeDriver; + const service = environmentRuntimeService({} as never, { drivers: [narrowedDriver] }); + + await expect( + service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: { ...PLUGIN_LEASE, metadata: { ...PLUGIN_LEASE.metadata, driver: "sandbox" } }, + command: "bridge-callback", + }), + ).rejects.toThrow(DUPLEX_DENIED); + expect(openDuplexChannel).not.toHaveBeenCalled(); + }); + + it("refuses a lease whose provider config fails to resolve", async () => { + const { manager, worker } = makeFakeWorkerManager(); + // The declaration grants the capability and the worker verifies the verb, but + // the provider config cannot be resolved. An untrusted provider fails closed, + // so the config-resolution failure narrows the duplex capability away. + mockResolveEnvironmentDriverConfigForRuntime.mockRejectedValue(new Error("config unresolved")); + const service = environmentRuntimeService({} as never, { pluginWorkerManager: manager }); + + await expect( + service.openDuplexChannel({ + environment: SANDBOX_ENVIRONMENT, + lease: PLUGIN_LEASE, + command: "bridge-callback", + }), + ).rejects.toThrow(DUPLEX_DENIED); + expect(worker.openDuplexChannel).not.toHaveBeenCalled(); + }); + + it("narrows the duplex capability away when the provider config resolution fails", () => { + // Unit cover for the narrowing rule the config-failure case relies on: a + // failed config resolution fails closed on the opt-in duplex capability. + const narrowing = buildSandboxCapabilityNarrowing({ + leasePolicy: "reuse_by_environment", + leaseMetadata: {}, + configResolutionFailed: true, + }); + expect(narrowing.duplexCommandStream).toBe(false); + + const granted = buildSandboxCapabilityNarrowing({ + leasePolicy: "reuse_by_environment", + leaseMetadata: {}, + configResolutionFailed: false, + }); + expect(granted.duplexCommandStream).toBeUndefined(); + }); +}); + +// Build a sandbox execution target with a fixed capability snapshot and a fake +// environment runtime whose openDuplexChannel is a spy. The helper returns the +// runner and the spy so a test reads the capability-gated member. +async function buildSandboxRunner(input: { + snapshot: EffectiveSandboxCapabilities | null; +}) { + mockResolveEnvironmentDriverConfigForRuntime.mockResolvedValue({ + driver: "sandbox", + config: { provider: "daytona", timeoutMs: 30_000 }, + }); + + const openDuplexChannel = vi.fn(async () => ({ + write: vi.fn(), + onData: vi.fn(), + onExit: vi.fn(), + stop: vi.fn(), + close: vi.fn(async () => {}), + })); + const environmentRuntime = { + execute: vi.fn().mockResolvedValue({ + exitCode: 0, + signal: null, + timedOut: false, + stdout: "ok", + stderr: "", + metadata: {}, + }), + supportsSync: () => false, + syncIn: vi.fn(), + syncOut: vi.fn(), + openDuplexChannel, + effectiveSandboxCapabilities: vi.fn(async () => + input.snapshot ? Object.freeze({ ...input.snapshot }) : null, + ), + } as unknown as EnvironmentRuntimeService; + + const target = await resolveEnvironmentExecutionTarget({ + db: {} as never, + companyId: "company-1", + adapterType: "codex_local", + environment: { id: "env-1", driver: "sandbox", config: { provider: "daytona" } }, + leaseId: "lease-1", + leaseMetadata: { remoteCwd: "/work" }, + lease: { id: "lease-1", leasePolicy: "reuse_by_environment" } as never, + environmentRuntime, + }); + if (!target || target.kind !== "remote" || target.transport !== "sandbox") { + throw new Error("expected a sandbox execution target"); + } + return { runner: target.runner, openDuplexChannel }; +} + +describe("inline sandbox runner duplex capability gate", () => { + it("exposes openDuplexChannel that delegates to the runtime when the capability is granted", async () => { + const { runner, openDuplexChannel } = await buildSandboxRunner({ snapshot: DUPLEX_GRANT }); + expect(runner?.openDuplexChannel).toBeDefined(); + + await runner!.openDuplexChannel!({ command: "bridge-callback" }); + expect(openDuplexChannel).toHaveBeenCalledWith({ + environment: expect.objectContaining({ id: "env-1" }), + lease: expect.objectContaining({ id: "lease-1" }), + command: "bridge-callback", + }); + }); + + it("omits openDuplexChannel when the capability is absent", async () => { + const { runner } = await buildSandboxRunner({ snapshot: DUPLEX_ABSENT }); + expect(runner?.openDuplexChannel).toBeUndefined(); + }); + + it("omits openDuplexChannel when the capability snapshot is null", async () => { + const { runner } = await buildSandboxRunner({ snapshot: null }); + expect(runner?.openDuplexChannel).toBeUndefined(); + }); +}); + +describe("ssh runner factory", () => { + it("omits openDuplexChannel", () => { + const runner = createSshCommandManagedRuntimeRunner({ + spec: { remoteCwd: "/work" } as never, + }); + expect(runner.openDuplexChannel).toBeUndefined(); + }); +}); diff --git a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs new file mode 100644 index 0000000000..2cd1967d95 --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs @@ -0,0 +1,174 @@ +// Test worker fixture for the host-owned duplex channel route state machine. The +// fixture drives the manager route state machine through the four typed methods +// (open, write, stop, close) and the data and exit notifications. +// +// The duplex channel is generic. It carries no command allowlist. The test +// encodes a JSON directive in the forwarded `providerLeaseId`, so one fixture +// serves every route case: +// - `mode`: "normal" | "malformed-open" | "no-open-reply" | "duplicate-open-reply" | +// "no-write-reply" +// - `workerSessionId`: the worker session id the open reply returns (default "ws-1") +// - `data`: an array of `{ chunk, sid? }`. The fixture emits each as a data +// notification after the open reply. `sid` defaults to the real worker session +// id; a test sets a wrong `sid` to prove the host drops a mismatched +// notification and counts a protocol error. +// - `exitCode`: when set, the fixture emits an exit notification after the data. +// - `echoInput`: when true, the fixture echoes each `duplexChannelWrite` back as +// one data notification for the bound session. +// - `closeMode`: "ack" | "bad-ack" | "no-ack" (default "ack"). It controls the +// close reply, so a test proves the host retires the worker on an unconfirmed +// close. +const readline = require("node:readline"); + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +// The registered channels, keyed by the host route id. Each entry records the +// bound worker session id and the close directive. +const routes = new Map(); + +function parseDirective(raw) { + try { + const parsed = JSON.parse(raw); + return parsed && typeof parsed === "object" ? parsed : {}; + } catch { + return {}; + } +} + +const rl = readline.createInterface({ input: process.stdin, crlfDelay: Infinity }); + +rl.on("line", (line) => { + if (!line.trim()) return; + const message = JSON.parse(line); + const method = message && typeof message.method === "string" ? message.method : null; + const params = message.params ?? {}; + + if (method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + ok: true, + supportedMethods: [ + "duplexChannelOpen", + "duplexChannelWrite", + "duplexChannelStop", + "duplexChannelClose", + ], + }, + }); + return; + } + + if (method === "duplexChannelOpen") { + const directive = parseDirective(params.providerLeaseId); + const mode = directive.mode ?? "normal"; + const workerSessionId = directive.workerSessionId ?? "ws-1"; + const closeMode = directive.closeMode ?? "ack"; + routes.set(params.hostRouteId, { + workerSessionId, + closeMode, + echoInput: directive.echoInput === true, + noWriteReply: mode === "no-write-reply", + }); + + if (mode === "no-open-reply") { + // Never reply, so the host open call times out. + return; + } + if (mode === "malformed-open") { + // Reply with no worker session id, so the host terminalizes the route. + send({ jsonrpc: "2.0", id: message.id, result: {} }); + return; + } + + const reply = () => + send({ jsonrpc: "2.0", id: message.id, result: { workerSessionId } }); + reply(); + if (mode === "duplicate-open-reply") { + // Send a second open reply for the same request id. The host drops it. + reply(); + } + + // Emit the scripted data and the exit after the open reply, so the host + // binds the route first. + setImmediate(() => { + const data = Array.isArray(directive.data) ? directive.data : []; + for (const entry of data) { + send({ + jsonrpc: "2.0", + method: "duplexChannel.data", + params: { + workerSessionId: entry.sid ?? workerSessionId, + chunk: entry.chunk, + }, + }); + } + if (typeof directive.exitCode === "number") { + send({ + jsonrpc: "2.0", + method: "duplexChannel.exit", + params: { workerSessionId, exitCode: directive.exitCode }, + }); + } + }); + return; + } + + if (method === "duplexChannelWrite") { + const entry = [...routes.values()].find( + (route) => route.workerSessionId === params.workerSessionId, + ); + if (entry && entry.noWriteReply) { + // Never reply, so the host write call stays pending. The test proves the + // host ends the route on the pending-request bound. + return; + } + if (entry && entry.echoInput) { + // Echo the input back as one data notification for the bound session, so a + // test proves the input reaches the worker and the output routes back. + send({ + jsonrpc: "2.0", + method: "duplexChannel.data", + params: { workerSessionId: entry.workerSessionId, chunk: `echo:${params.data}` }, + }); + } + send({ jsonrpc: "2.0", id: message.id, result: null }); + return; + } + + if (method === "duplexChannelStop") { + send({ jsonrpc: "2.0", id: message.id, result: null }); + return; + } + + if (method === "duplexChannelClose") { + const entry = routes.get(params.hostRouteId); + routes.delete(params.hostRouteId); + const closeMode = entry ? entry.closeMode : "ack"; + if (closeMode === "no-ack") { + // Never reply, so the host close call times out and the host retires us. + return; + } + if (closeMode === "bad-ack") { + send({ jsonrpc: "2.0", id: message.id, result: { hostRouteId: "mismatched-route" } }); + return; + } + send({ jsonrpc: "2.0", id: message.id, result: { hostRouteId: params.hostRouteId } }); + return; + } + + if (method === "shutdown") { + send({ jsonrpc: "2.0", id: message.id, result: {} }); + setImmediate(() => process.exit(0)); + return; + } + + send({ + jsonrpc: "2.0", + id: message.id, + error: { code: -32601, message: `Unhandled method: ${method}` }, + }); +}); diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts new file mode 100644 index 0000000000..7a02b92b50 --- /dev/null +++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts @@ -0,0 +1,505 @@ +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { describe, expect, it, vi } from "vitest"; +import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; +import { createPluginWorkerHandle } from "../services/plugin-worker-manager.js"; + +const FIXTURES_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures"); +const DUPLEX_CHANNEL_WORKER_ENTRYPOINT = path.join( + FIXTURES_DIR, + "plugin-worker-duplex-channel.cjs", +); + +const TEST_MANIFEST: PaperclipPluginManifestV1 = { + id: "test.plugin", + apiVersion: 1, + version: "1.0.0", + displayName: "Test plugin", + description: "Test plugin", + author: "Paperclip", + categories: ["automation"], + capabilities: [], + entrypoints: { worker: "dist/worker.js" }, +}; + +function makeDuplexHandle(extra?: Record) { + return createPluginWorkerHandle("test.plugin", { + entrypointPath: DUPLEX_CHANNEL_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { instanceId: "instance-1", hostVersion: "1.0.0" }, + apiVersion: 1, + hostHandlers: {}, + ...extra, + }); +} + +// The test directive rides in `providerLeaseId`, an opaque field the manager +// forwards to the worker unchanged. The duplex channel is generic, so the +// command is a plain fixed string with no allowlist. +function duplexOpenInput(directive: unknown) { + return { + driverKey: "daytona", + companyId: "company-1", + environmentId: "env-1", + providerLeaseId: JSON.stringify(directive), + command: "bridge-callback", + }; +} + +describe("plugin worker manager duplex channel route", () => { + it("delivers data only for the exact bound worker session id and drops a mismatch", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [ + { chunk: "good-1" }, + { chunk: "forged", sid: "ws-EVIL" }, + { chunk: "good-2" }, + ], + exitCode: 0, + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + // The forged notification carries a wrong worker session id, so the host + // drops it. Only the two bound chunks reach the listener, in order. + expect(chunks).toEqual(["good-1", "good-2"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("routes input to the worker and back to the listener", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ workerSessionId: "ws-A", echoInput: true }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + session.write("callback-payload"); + // The worker echoes the input as one data notification for the bound + // session, so the listener receives it. + await vi.waitFor(() => expect(chunks).toContain("echo:callback-payload")); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("buffers early data in order until a listener attaches and drains it in order", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + data: [{ chunk: "one" }, { chunk: "two" }, { chunk: "three" }], + }), + ); + // Wait so the three data notifications arrive and buffer before a listener + // attaches. The drain then delivers them in order. + await new Promise((resolve) => setTimeout(resolve, 60)); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await vi.waitFor(() => expect(chunks.length).toBe(3)); + expect(chunks).toEqual(["one", "two", "three"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("isolates a throwing listener during live delivery so later chunks still route", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [{ chunk: "ok-1" }, { chunk: "boom" }, { chunk: "ok-2" }], + exitCode: 0, + }), + ); + const chunks: string[] = []; + // The listener throws on one chunk. The manager catches the throw, so it + // does not escape the worker stdout notification handler. The later chunk + // still routes and the route still settles. + session.onData((chunk) => { + chunks.push(chunk); + if (chunk === "boom") throw new Error("listener failure"); + }); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual(["ok-1", "boom", "ok-2"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("isolates a throwing listener during the buffered replay so every buffered chunk routes", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + data: [{ chunk: "one" }, { chunk: "boom" }, { chunk: "three" }], + }), + ); + // Wait so the three data notifications arrive and buffer before a listener + // attaches. The drain then delivers them in order. + await new Promise((resolve) => setTimeout(resolve, 60)); + const chunks: string[] = []; + // The listener throws on one buffered chunk. The manager catches the throw + // inside the drain, so it does not escape `onData` and every buffered chunk + // still routes. + expect(() => + session.onData((chunk) => { + chunks.push(chunk); + if (chunk === "boom") throw new Error("listener failure"); + }), + ).not.toThrow(); + expect(chunks).toEqual(["one", "boom", "three"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("binds the worker session id one time and ignores a duplicate open reply", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + mode: "duplicate-open-reply", + workerSessionId: "ws-A", + data: [{ chunk: "hello" }], + exitCode: 0, + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // The duplicate open reply never rebinds or reopens the route, so the + // session runs normally on the one bind. + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual(["hello"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("terminalizes and fails closed on a malformed open reply, then admits a later open", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + await expect( + handle.openDuplexChannel(duplexOpenInput({ mode: "malformed-open" })), + ).rejects.toThrow("DUPLEX_CHANNEL_OPEN_FAILED"); + // The terminalize closed the route by the host route id and the worker + // acknowledged the close, so a later open is admitted. + const session = await handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })); + expect(session).toBeDefined(); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("permits one active duplex channel per worker", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const first = await handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })); + // A second open while the first route is not closed rejects with one fixed + // non-secret error before it reaches the worker. + await expect( + handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })), + ).rejects.toThrow("DUPLEX_CHANNEL_ROUTE_BUSY"); + await first.close(); + // After the first route closes and the worker acknowledges the close, a new + // open is admitted. + const second = await handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })); + await second.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + // ------------------------------------------------------------------------- + // The five explicit bounds. Each bound ends the route when it is exceeded. + // ------------------------------------------------------------------------- + + it("ends the route when the pre-bind buffered bytes pass the bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedChars: 10 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + data: [ + { chunk: "aaaaa" }, // total 5 → buffered + { chunk: "bbbbb" }, // total 10 → buffered + { chunk: "ccccc" }, // total 15 > 10 → end route + ], + }), + ); + // No listener attaches, so the data buffers. The cumulative bytes pass the + // bound and the route ends. The login wait resolves with a null exit code. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route when the pre-bind buffered frame count passes the bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPreBindBufferedFrames: 2 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + data: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + }), + ); + // No listener attaches, so the data buffers. The third frame passes the + // frame-count bound and the route ends. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route when the pending request count passes the bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxPendingRequests: 2 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ mode: "no-write-reply", workerSessionId: "ws-A" }), + ); + const waitResult = session.wait(); + // The worker never replies to a write, so each write stays pending. The + // third write passes the pending-request bound and the route ends. + session.write("one"); + session.write("two"); + session.write("three"); + await expect(waitResult).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route when one host-to-worker write passes the size bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxWriteChars: 8 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ workerSessionId: "ws-A" }), + ); + const waitResult = session.wait(); + // One write is larger than the size bound, so the host rejects it and ends + // the route before it reaches the worker. + session.write("this-write-is-too-large"); + await expect(waitResult).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route when the protocol error count passes the bound", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxProtocolErrors: 2 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [ + { chunk: "e1", sid: "ws-EVIL" }, + { chunk: "e2", sid: "ws-EVIL" }, + { chunk: "e3", sid: "ws-EVIL" }, + ], + }), + ); + // Each mismatched-session data frame is a protocol error. The third frame + // passes the error bound and the route ends. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route when the total data bytes pass the cap for a bound listener", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxTotalDataBytes: 10 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [ + { chunk: "aaaaa" }, // total 5 → deliver + { chunk: "bbbbb" }, // total 10 → deliver + { chunk: "ccccc" }, // total 15 > 10 → end route + ], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // A listener is bound, so the host forwards each chunk until the cumulative + // bytes pass the cap. The third chunk passes the cap, so the host drops it + // and ends the route. The listener never receives data past the cap. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(chunks).toEqual(["aaaaa", "bbbbb"]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("counts inbound bytes, not characters, against the total cap", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxTotalDataBytes: 4 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + // "€" is one character but three bytes in UTF-8. The first chunk is 3 + // bytes (≤ 4), so the host delivers it. The second chunk brings the + // total to 6 bytes (> 4), so the host ends the route. A character count + // would admit both chunks (2 ≤ 4), so one delivered chunk proves the + // host counts bytes. + data: [{ chunk: "€" }, { chunk: "€" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(chunks).toEqual(["€"]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the active route when the lifetime timer expires", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxDurationMs: 100 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ mode: "normal", workerSessionId: "ws-A" }), + ); + const waitResult = session.wait(); + // The route sends no exit. The lifetime timer expires, so the host ends the + // route and resolves the wait with the fixed null exit code. + await expect(waitResult).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route at once when one inbound chunk passes the per-chunk limit before a listener binds", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxChunkChars: 4 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [{ chunk: "this-one-chunk-is-too-large" }], + }), + ); + // No listener attaches. One inbound chunk is larger than the per-chunk + // limit, so the host ends the route at once. The default protocol-error + // budget is far above one, so a single chunk that ends the route proves the + // host does not treat it as a protocol error. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("ends the route at once when one inbound chunk passes the per-chunk limit after a listener binds", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { maxChunkChars: 4 }, + }); + try { + await handle.start(); + const session = await handle.openDuplexChannel( + duplexOpenInput({ + workerSessionId: "ws-A", + data: [{ chunk: "this-one-chunk-is-too-large" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // A listener is bound. One inbound chunk is larger than the per-chunk + // limit, so the host ends the route at once and never forwards the chunk. + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + expect(chunks).toEqual([]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + // ------------------------------------------------------------------------- + // Authoritative closure and worker retirement. + // ------------------------------------------------------------------------- + + it("closes the route with a fixed exit when the worker exits", async () => { + const handle = makeDuplexHandle(); + try { + await handle.start(); + const session = await handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })); + const waitResult = session.wait(); + await handle.stop(); + // A worker exit closes the one route and resolves the wait with the fixed + // non-secret exit. + await expect(waitResult).resolves.toEqual({ exitCode: null }); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("retires the worker on an unconfirmed close acknowledgement", async () => { + const handle = makeDuplexHandle({ + duplexChannelLimits: { closeTimeoutMs: 200 }, + }); + try { + await handle.start(); + const exited = new Promise((resolve) => { + handle.on("exit", () => resolve()); + }); + const session = await handle.openDuplexChannel( + duplexOpenInput({ mode: "normal", closeMode: "bad-ack" }), + ); + await session.close(); + // The close acknowledgement carried a mismatched host route id, so the host + // fails closed and retires the worker before any reuse. + await exited; + await expect( + handle.openDuplexChannel(duplexOpenInput({ mode: "normal" })), + ).rejects.toThrow(); + } finally { + await handle.stop().catch(() => undefined); + } + }); +}); diff --git a/server/src/__tests__/sandbox-capability-contract.test.ts b/server/src/__tests__/sandbox-capability-contract.test.ts index 571a8ddbbc..1c9c933620 100644 --- a/server/src/__tests__/sandbox-capability-contract.test.ts +++ b/server/src/__tests__/sandbox-capability-contract.test.ts @@ -324,6 +324,48 @@ describe("sandbox capability contract normalizer", () => { expect(outOnly.concurrentSyncOperations).toBe(false); }); + it("test_duplex_command_stream_absent_declaration_resolves_false", () => { + // The duplex channel is opt-in and fail-closed. An absent declaration denies + // the capability even when the worker verifies the duplex open verb. This + // matches the incremental-session-output pattern: an opt-in behavioral + // guarantee needs a positive declaration, not just a verified verb. + const undeclared = resolveEffectiveSandboxCapabilities({ + verifiedMethods: ["duplexChannelOpen"], + declared: null, + }); + expect(undeclared.duplexCommandStream).toBe(false); + }); + + it("test_duplex_command_stream_needs_verified_worker_method", () => { + // A declaration never grants the capability without the verified duplex open + // verb. A provider that declares the capability but whose worker does not + // report the duplex open method resolves false. + const declaredButUnverified = resolveEffectiveSandboxCapabilities({ + verifiedMethods: ["environmentExecute"], + declared: { duplexCommandStream: true }, + }); + expect(declaredButUnverified.duplexCommandStream).toBe(false); + }); + + it("test_duplex_command_stream_declared_and_verified_resolves_true_but_narrowing_removes_it", () => { + // A provider that declares the capability and whose worker verifies the + // duplex open verb gets the capability. + const granted = resolveEffectiveSandboxCapabilities({ + verifiedMethods: ["duplexChannelOpen"], + declared: { duplexCommandStream: true }, + }); + expect(granted.duplexCommandStream).toBe(true); + + // Per-target narrowing still removes a verified and declared capability, so a + // lease that cannot use the duplex channel keeps the file bridge. + const narrowed = resolveEffectiveSandboxCapabilities({ + verifiedMethods: ["duplexChannelOpen"], + declared: { duplexCommandStream: true }, + narrowing: { duplexCommandStream: false }, + }); + expect(narrowed.duplexCommandStream).toBe(false); + }); + it("test_unknown_or_unavailable_verification_resolves_false", () => { const declaredAll = { reusableLeases: true, diff --git a/server/src/services/environment-execution-target.ts b/server/src/services/environment-execution-target.ts index 75738c5943..fccf326a4c 100644 --- a/server/src/services/environment-execution-target.ts +++ b/server/src/services/environment-execution-target.ts @@ -505,6 +505,21 @@ export async function resolveEnvironmentExecutionTarget(input: { }), } : {}), + // Expose the duplex channel only when the effective snapshot grants + // the opt-in `duplexCommandStream` capability. A null snapshot + // (resolution failed or the snapshot is not resolvable) leaves the + // member undefined, so the caller keeps the file bridge. This mirrors + // the syncIn/syncOut gate above and fails closed. + ...(effectiveCapabilities?.duplexCommandStream + ? { + openDuplexChannel: (channelInput) => + input.environmentRuntime!.openDuplexChannel({ + environment: input.environment as Environment, + lease: input.lease!, + command: channelInput.command, + }), + } + : {}), } : undefined, }; diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index db554c9dbd..9588740e03 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -14,6 +14,9 @@ import type { } from "@paperclipai/shared"; import { resolveDeclaredSandboxCapabilities } from "@paperclipai/shared"; import type { EffectiveSandboxCapabilities } from "@paperclipai/adapter-utils/execution-target"; +import type { + CommandManagedDuplexChannel, +} from "@paperclipai/adapter-utils/command-managed-runtime"; import type { PluginEnvironmentAcquireLeaseParams, PluginEnvironmentExecuteResult, @@ -52,7 +55,12 @@ import { sandboxConfigFromLeaseMetadataLoose, } from "./sandbox-provider-runtime.js"; import { pluginRegistryService } from "./plugin-registry.js"; -import type { ExecuteLogSink, PluginWorkerManager } from "./plugin-worker-manager.js"; +import type { + ExecuteLogSink, + PluginWorkerManager, + DuplexChannelHostSession, + DuplexChannelOpenInput as WorkerManagerDuplexChannelOpenInput, +} from "./plugin-worker-manager.js"; import { REUSABLE_LEASE_WORKER_METHODS, destroyPluginEnvironmentLease, @@ -78,6 +86,12 @@ import { logger } from "../middleware/logger.js"; // this constant and the plain provider identifiers only. const SANDBOX_ORPHAN_CLEANUP_WRITE_ERROR_KIND = "sandbox_orphan_cleanup_write_failed"; +// The fixed non-secret refusal a duplex channel open returns when the lease does +// not grant the opt-in `duplexCommandStream` capability. The service resolves the +// exact lease capability snapshot and throws this before it reaches any driver. +const DUPLEX_CHANNEL_CAPABILITY_DENIED = + "Sandbox lease does not grant the duplex command stream capability."; + // --------------------------------------------------------------------------- // Sandbox capability contract — one normalizer for both branches // --------------------------------------------------------------------------- @@ -90,6 +104,7 @@ export const SANDBOX_CAPABILITY_KEYS = [ "independentControlCommands", "incrementalSessionOutput", "concurrentSyncOperations", + "duplexCommandStream", ] as const; export type SandboxCapabilityKey = (typeof SANDBOX_CAPABILITY_KEYS)[number]; @@ -106,6 +121,7 @@ export type SandboxCapabilityKey = (typeof SANDBOX_CAPABILITY_KEYS)[number]; const SANDBOX_CAPABILITY_OPT_IN_KEYS: ReadonlySet = new Set([ "incrementalSessionOutput", "concurrentSyncOperations", + "duplexCommandStream", ]); /** @@ -142,6 +158,11 @@ const SANDBOX_CAPABILITY_OPT_IN_KEYS: ReadonlySet = new Se * verifies only one direction cannot get the capability. The verbs are * necessary but not sufficient: this key is opt-in, so the declaration is the * real gate (see {@link SANDBOX_CAPABILITY_OPT_IN_KEYS}). + * - `duplexCommandStream` requires `duplexChannelOpen`, the worker verb that + * opens the persistent duplex channel. The verified verb is necessary but not + * sufficient: this key is opt-in, so the declaration is the real gate. A + * provider that does not implement the duplex open verb resolves `false` and + * keeps the file bridge. */ const SANDBOX_CAPABILITY_PREREQUISITE_METHODS: Record = { // Reusable leases require ALL reuse verbs. Each verb is its own required @@ -156,6 +177,7 @@ const SANDBOX_CAPABILITY_PREREQUISITE_METHODS: Record; @@ -486,6 +518,15 @@ export interface EnvironmentRuntimeDriver { */ syncIn?(input: EnvironmentDriverSyncInput): Promise; syncOut?(input: EnvironmentDriverSyncInput): Promise; + /** + * Optional persistent duplex channel. Present only for a plugin-backed sandbox + * driver whose lease grants the `duplexCommandStream` capability. The driver + * opens the host-owned duplex route on the plugin worker and adapts it to the + * cross-layer {@link CommandManagedDuplexChannel}. Other drivers omit it. + */ + openDuplexChannel?( + input: EnvironmentDriverOpenDuplexChannelInput, + ): Promise; /** True when the lease's plugin worker advertises both sync verbs. */ supportsSync?(input: EnvironmentDriverLeaseInput): boolean; /** @@ -993,6 +1034,39 @@ function createSshEnvironmentDriver(db: Db): EnvironmentRuntimeDriver { }; } +/** + * Adapt the worker manager's duplex host session to the cross-layer channel. The + * two shapes differ in the exit and stop members: the host session resolves the + * exit one time from `wait()`, so the channel bridges it to a one-time `onExit` + * listener; `kill()` maps to `stop()`. The `write`, `onData`, and `close` + * members map one to one. + */ +function adaptDuplexChannelHostSession( + session: DuplexChannelHostSession, +): CommandManagedDuplexChannel { + return { + write(data: string): void { + session.write(data); + }, + onData(listener: (chunk: string) => void): void { + session.onData(listener); + }, + onExit(listener: (exit: { exitCode: number | null }) => void): void { + // `wait()` resolves one time with the exit and never rejects, so a single + // `then` bridges it to the one-time exit listener. + void session.wait().then((exit) => { + listener(exit); + }); + }, + stop(): void { + session.kill(); + }, + close(): Promise { + return session.close(); + }, + }; +} + function createSandboxEnvironmentDriver( db: Db, options: { @@ -2369,6 +2443,37 @@ function createSandboxEnvironmentDriver( return await callPluginEnvironmentSync("environmentSyncOut", input); }, + async openDuplexChannel(input) { + // Plugin-backed sandbox providers only: open the host-owned duplex route on + // the plugin worker. The lease scope mirrors the sandbox execute path — the + // provider driver key, the company, the environment, and the provider lease + // id — so the route binds to the same worker session the runner streams. + if (!input.lease.metadata?.sandboxProviderPlugin || !pluginWorkerManager) { + throw new Error("Sandbox driver does not support duplex channels for this lease."); + } + const pluginId = readString(input.lease.metadata?.pluginId); + const providerKey = readString(input.lease.metadata?.provider); + const providerLeaseId = readString(input.lease.providerLeaseId); + if (!pluginId || !providerKey || !providerLeaseId) { + throw new Error( + "Sandbox duplex channel needs a plugin id, a provider key, and a provider lease id on the lease.", + ); + } + const worker = pluginWorkerManager.getWorker(pluginId); + if (!worker) { + throw new Error(`Plugin worker "${pluginId}" is not running for the duplex channel.`); + } + const managerInput: WorkerManagerDuplexChannelOpenInput = { + driverKey: providerKey, + companyId: input.lease.companyId, + environmentId: input.environment.id, + providerLeaseId, + command: input.command, + }; + const session = await worker.openDuplexChannel(managerInput); + return adaptDuplexChannelHostSession(session); + }, + async effectiveSandboxCapabilities(input) { const metadata = input.lease.metadata ?? {}; const providerKey = @@ -3320,6 +3425,27 @@ export function environmentRuntimeService( } return await driver.syncOut(input); }, + + async openDuplexChannel( + input: EnvironmentDriverOpenDuplexChannelInput, + ): Promise { + const driver = requireDriverKey(getLeaseDriverKey(input.lease, input.environment)); + if (!driver.openDuplexChannel) { + throw new Error(`Environment driver "${driver.driver}" does not support duplex channels.`); + } + // Centralize the duplex channel authorization here. Resolve the exact lease + // capability snapshot and refuse unless the effective snapshot grants the + // opt-in `duplexCommandStream` capability. This gate runs before the driver + // call, so an unauthorized lease never reaches the worker. The + // execution-target member gate stays as defense in depth. A driver that + // cannot resolve the snapshot fails closed with the fixed refusal. + const effective = + (await driver.effectiveSandboxCapabilities?.(input)) ?? null; + if (effective?.duplexCommandStream !== true) { + throw new Error(DUPLEX_CHANNEL_CAPABILITY_DENIED); + } + return await driver.openDuplexChannel(input); + }, }; } diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 5006a9c5c6..8c54cdbeb2 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -39,6 +39,8 @@ import { JsonRpcCallError, SETUP_TOKEN_PTY_OUTPUT_NOTIFICATION, SETUP_TOKEN_PTY_EXIT_NOTIFICATION, + DUPLEX_CHANNEL_DATA_NOTIFICATION, + DUPLEX_CHANNEL_EXIT_NOTIFICATION, } from "@paperclipai/plugin-sdk"; import type { JsonRpcId, @@ -156,6 +158,60 @@ const SETUP_TOKEN_PTY_ROUTE_BUSY = "SETUP_TOKEN_PTY_ROUTE_BUSY"; /** The fixed non-secret error a failed open returns. */ const SETUP_TOKEN_PTY_OPEN_FAILED = "SETUP_TOKEN_PTY_OPEN_FAILED"; +// Bounds and timeouts for the generic duplex channel route. The route mirrors the +// login pseudo-terminal route, but it carries no command allowlist and adds seven +// explicit bounds the pseudo-terminal route lacks. Each bound ends the route when +// it passes the limit, so a faulty or hostile worker cannot flood the host. +/** The default maximum characters for one duplex channel data notification. */ +const MAX_DUPLEX_CHANNEL_CHUNK_CHARS = 1_000_000; +/** + * The default maximum cumulative characters the host buffers for one duplex + * channel route before a data listener attaches. A worker that streams data + * before the consumer binds cannot grow the host buffer without limit. + */ +const MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS = 8 * 1024 * 1024; +/** + * The default maximum number of data frames the host buffers for one duplex + * channel route before a data listener attaches. + */ +const MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES = 10_000; +/** + * The default maximum number of in-flight host→worker requests for one duplex + * channel route. A worker that never replies cannot make the host hold an + * unbounded number of pending requests. + */ +const MAX_DUPLEX_CHANNEL_PENDING_REQUESTS = 256; +/** The default maximum characters for one host→worker duplex channel write. */ +const MAX_DUPLEX_CHANNEL_WRITE_CHARS = 1_000_000; +/** + * The default maximum number of protocol errors for one duplex channel route. + * A protocol error is one malformed or mismatched data frame. The route ends + * when the count passes this budget, so a flood of bad frames bounds the route. + */ +const MAX_DUPLEX_CHANNEL_PROTOCOL_ERRORS = 100; +/** + * The default maximum cumulative bytes the host forwards for one duplex channel + * route over its whole life. The host counts the bytes of every inbound chunk, + * before and after a data listener attaches. The route ends when the count + * passes this cap, so an active route with a bound listener cannot stream an + * unbounded number of bytes. + */ +const MAX_DUPLEX_CHANNEL_TOTAL_DATA_BYTES = 256 * 1024 * 1024; +/** + * The default maximum lifetime for one duplex channel route, in milliseconds. + * The host starts a timer when the route opens and ends the route when the + * timer expires, so a route cannot live without limit. + */ +const MAX_DUPLEX_CHANNEL_DURATION_MS = 60 * 60 * 1000; +/** The default open timeout for one duplex channel route, in milliseconds. */ +const DUPLEX_CHANNEL_OPEN_TIMEOUT_MS = 30_000; +/** The default close timeout for one duplex channel route, in milliseconds. */ +const DUPLEX_CHANNEL_CLOSE_TIMEOUT_MS = 10_000; +/** The fixed non-secret error a rejected second duplex channel open returns. */ +const DUPLEX_CHANNEL_ROUTE_BUSY = "DUPLEX_CHANNEL_ROUTE_BUSY"; +/** The fixed non-secret error a failed duplex channel open returns. */ +const DUPLEX_CHANNEL_OPEN_FAILED = "DUPLEX_CHANNEL_OPEN_FAILED"; + /** Minimum time between two dropped-`execute.log` debug records. The router * rate-limits the record so a flood of dropped chunks writes at most one line * per window with a running count. */ @@ -322,6 +378,36 @@ export interface WorkerStartOptions { /** The close timeout for one login pseudo-terminal route, in milliseconds. */ closeTimeoutMs?: number; }; + + /** + * Bounds and timeouts for the generic duplex channel route. The defaults bound + * one data notification, the pre-bind buffer, the in-flight request count, one + * host→worker write, the protocol-error budget, and the open and close + * timeouts. A test overrides them to exercise each bound without huge inputs or + * long waits. + */ + duplexChannelLimits?: { + /** Max characters for one duplex channel data notification. */ + maxChunkChars?: number; + /** Max cumulative characters the host buffers before a data listener attaches. */ + maxPreBindBufferedChars?: number; + /** Max number of data frames the host buffers before a data listener attaches. */ + maxPreBindBufferedFrames?: number; + /** Max number of in-flight host→worker requests for one route. */ + maxPendingRequests?: number; + /** Max characters for one host→worker duplex channel write. */ + maxWriteChars?: number; + /** Max number of protocol errors for one route before the route ends. */ + maxProtocolErrors?: number; + /** Max cumulative bytes the host forwards for one route over its whole life. */ + maxTotalDataBytes?: number; + /** The maximum lifetime for one route, in milliseconds. */ + maxDurationMs?: number; + /** The open timeout for one duplex channel route, in milliseconds. */ + openTimeoutMs?: number; + /** The close timeout for one duplex channel route, in milliseconds. */ + closeTimeoutMs?: number; + }; } /** @@ -393,6 +479,38 @@ export interface SetupTokenPtyHostSession { close(): Promise; } +/** + * The input the manager needs to open one generic duplex channel. The manager + * mints the host route identifier. The caller supplies the sandbox scope, the + * provider lease id, and the command. The duplex channel carries no command + * allowlist, so the caller owns the command. + */ +export interface DuplexChannelOpenInput { + driverKey: string; + companyId: string; + environmentId: string; + providerLeaseId: string; + command: string; +} + +/** + * One live duplex channel the manager hands to a caller. The shape matches the + * login pseudo-terminal session, so a caller consumes one live bidirectional + * stream with the same methods. + */ +export interface DuplexChannelHostSession { + /** 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 channel. */ + write(data: string): void; + /** Resolves with the child exit code when the command ends or the route ends. */ + wait(): Promise<{ exitCode: number | null }>; + /** Stops the child process. Safe to call more than one time. */ + kill(): void; + /** Closes the route and releases the channel. Safe to call more than one time. */ + close(): Promise; +} + /** * Host-owned route for one active execute call. The host mints the invocation * id and stores the exact company id and log sink here. A worker never selects @@ -484,6 +602,17 @@ export interface PluginWorkerHandle { input: SetupTokenPtyOpenInput, ): Promise; + /** + * Open one generic duplex channel on this worker. The manager mints the host + * route identifier, reserves the route, drives the open, binds the worker + * session identifier one time, and returns a session a caller drives. It + * permits one active duplex channel per worker. It enforces five explicit + * bounds and ends the route when a bound passes its limit. + */ + openDuplexChannel( + input: DuplexChannelOpenInput, + ): Promise; + /** * Authorize the set of companies this worker may act on from proactive * (non-invocation) context. Replaces any previously-authorized set. See the @@ -667,6 +796,34 @@ export function createPluginWorkerHandle( const setupTokenPtyCloseTimeoutMs = options.setupTokenPtyLimits?.closeTimeoutMs ?? SETUP_TOKEN_PTY_CLOSE_TIMEOUT_MS; + // Bounds and timeouts for the generic duplex channel route. A caller (a test) + // can lower them to exercise each bound and the terminalize paths. + const maxDuplexChannelChunkChars = + options.duplexChannelLimits?.maxChunkChars ?? MAX_DUPLEX_CHANNEL_CHUNK_CHARS; + const maxDuplexChannelPreBindChars = + options.duplexChannelLimits?.maxPreBindBufferedChars ?? + MAX_DUPLEX_CHANNEL_PRE_BIND_CHARS; + const maxDuplexChannelPreBindFrames = + options.duplexChannelLimits?.maxPreBindBufferedFrames ?? + MAX_DUPLEX_CHANNEL_PRE_BIND_FRAMES; + const maxDuplexChannelPendingRequests = + options.duplexChannelLimits?.maxPendingRequests ?? + MAX_DUPLEX_CHANNEL_PENDING_REQUESTS; + const maxDuplexChannelWriteChars = + options.duplexChannelLimits?.maxWriteChars ?? MAX_DUPLEX_CHANNEL_WRITE_CHARS; + const maxDuplexChannelProtocolErrors = + options.duplexChannelLimits?.maxProtocolErrors ?? + MAX_DUPLEX_CHANNEL_PROTOCOL_ERRORS; + const maxDuplexChannelTotalDataBytes = + options.duplexChannelLimits?.maxTotalDataBytes ?? + MAX_DUPLEX_CHANNEL_TOTAL_DATA_BYTES; + const maxDuplexChannelDurationMs = + options.duplexChannelLimits?.maxDurationMs ?? MAX_DUPLEX_CHANNEL_DURATION_MS; + const duplexChannelOpenTimeoutMs = + options.duplexChannelLimits?.openTimeoutMs ?? DUPLEX_CHANNEL_OPEN_TIMEOUT_MS; + const duplexChannelCloseTimeoutMs = + options.duplexChannelLimits?.closeTimeoutMs ?? DUPLEX_CHANNEL_CLOSE_TIMEOUT_MS; + // ------------------------------------------------------------------ // Proactive company scopes (LOOA-629) // ------------------------------------------------------------------ @@ -1039,7 +1196,42 @@ export function createPluginWorkerHandle( // verifies a close acknowledgement bound to that identifier; it retires the // worker on an unconfirmed close. - type SetupTokenPtyRouteState = "reserved" | "opening" | "open" | "closed"; + // A single-consumer route state. The login pseudo-terminal route and the + // generic duplex channel route share it. + type RouteState = "reserved" | "opening" | "open" | "closed"; + + // Shared route-binding helpers. The login pseudo-terminal route and the duplex + // channel route both use them, so the two routes bind and settle one way. + + // Settle the route wait exactly once. Replace the settler with a no-op, so a + // later exit or terminalize never settles the wait a second time. + function settleRouteWait( + route: { settleWait: (value: { exitCode: number | null }) => void }, + value: { exitCode: number | null }, + ): void { + const settle = route.settleWait; + route.settleWait = () => {}; + settle(value); + } + + // Read the worker session identifier from an open reply, but only when the + // route can still bind. Return null for a malformed reply, or for a route that + // already left `opening` or terminalized. A late or a duplicate reply never + // binds, revives, or reopens a route. + function readBindableWorkerSessionId( + route: { state: RouteState; terminalized: boolean }, + openResult: unknown, + ): string | null { + const workerSessionId = readNonEmptyString( + isRecord(openResult) ? openResult.workerSessionId : null, + ); + if (!workerSessionId || route.state !== "opening" || route.terminalized) { + return null; + } + return workerSessionId; + } + + type SetupTokenPtyRouteState = RouteState; interface SetupTokenPtyRoute { hostRouteId: string; state: SetupTokenPtyRouteState; @@ -1054,15 +1246,6 @@ export function createPluginWorkerHandle( // blocks a second open until the manager confirms the first route's close. let setupTokenPtyRoute: SetupTokenPtyRoute | null = null; - function settleSetupTokenPtyWait( - route: SetupTokenPtyRoute, - value: { exitCode: number | null }, - ): void { - const settle = route.settleWait; - route.settleWait = () => {}; - settle(value); - } - // Close the worker terminal by the host route identifier and verify the bound // acknowledgement. Return true only when the worker returns an acknowledgement // that carries the exact host route identifier. An absent, malformed, @@ -1092,7 +1275,7 @@ export function createPluginWorkerHandle( route.buffered = []; // A terminalized route reports a null exit code, which the runner treats as a // failure. - settleSetupTokenPtyWait(route, { exitCode: null }); + settleRouteWait(route, { exitCode: null }); const confirmed = await closeSetupTokenPtyTerminal(route.hostRouteId); if (setupTokenPtyRoute === route) setupTokenPtyRoute = null; if (!confirmed) { @@ -1144,7 +1327,7 @@ export function createPluginWorkerHandle( const workerSessionId = readNonEmptyString(params.workerSessionId); if (!workerSessionId || workerSessionId !== route.workerSessionId) return; const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; - settleSetupTokenPtyWait(route, { exitCode }); + settleRouteWait(route, { exitCode }); } // Close the one route on a worker exit. The worker is gone, so the manager @@ -1158,7 +1341,7 @@ export function createPluginWorkerHandle( route.state = "closed"; route.listener = null; route.buffered = []; - settleSetupTokenPtyWait(route, { exitCode: null }); + settleRouteWait(route, { exitCode: null }); } // Open one live login pseudo-terminal route. Reserve the route @@ -1219,10 +1402,8 @@ export function createPluginWorkerHandle( throw err instanceof Error ? err : new Error(SETUP_TOKEN_PTY_OPEN_FAILED); } - const workerSessionId = readNonEmptyString( - isRecord(openResult) ? openResult.workerSessionId : null, - ); - if (!workerSessionId || route.state !== "opening" || route.terminalized) { + const workerSessionId = readBindableWorkerSessionId(route, openResult); + if (!workerSessionId) { // A malformed reply, or a route that already left `opening`. A late or a // duplicate reply never binds, revives, or reopens a route. await terminalizeSetupTokenPtyRoute(route); @@ -1268,6 +1449,345 @@ export function createPluginWorkerHandle( }; } + // ----------------------------------------------------------------------- + // Host-owned generic duplex channel route + // ----------------------------------------------------------------------- + // The duplex channel route mirrors the login pseudo-terminal route model. The + // host owns the route identifier, binds the worker session identifier one time + // on a valid open reply, and keys the close on the host route identifier. It + // rejects a late or a duplicate open reply, and it retires the worker on an + // unconfirmed close. The duplex channel carries no command allowlist, so the + // caller owns the command. + // + // The route adds seven explicit bounds the pseudo-terminal route lacks. Each + // bound ends the route when it passes its limit: + // 1. pre-bind buffered bytes — the cumulative characters the host buffers + // before a data listener attaches; + // 2. pre-bind buffered frame count — the number of data frames the host + // buffers before a data listener attaches; + // 3. pending request count — the number of in-flight host→worker requests; + // 4. host→worker write size — the characters for one write; + // 5. protocol error rate — the count of malformed or mismatched data frames; + // 6. total data bytes — the cumulative inbound bytes over the whole life, + // counted before and after a data listener attaches; + // 7. route lifetime — the milliseconds from the open to the terminal end. + interface DuplexChannelRoute { + hostRouteId: string; + state: RouteState; + workerSessionId: string | null; + listener: ((chunk: string) => void) | null; + buffered: string[]; + bufferedChars: number; + pendingRequests: number; + protocolErrors: number; + totalDataBytes: number; + lifetimeTimer: ReturnType | null; + terminalized: boolean; + settleWait: (value: { exitCode: number | null }) => void; + } + // At most one active duplex channel per worker. A non-null route blocks a + // second open until the manager confirms the first route's close. + let duplexChannelRoute: DuplexChannelRoute | null = null; + + // Close the worker channel by the host route identifier and verify the bound + // acknowledgement. Return true only when the worker returns an acknowledgement + // that carries the exact host route identifier. An absent, malformed, + // mismatched, or timed-out acknowledgement returns false, so the caller fails + // closed. + async function closeDuplexChannelTerminal(hostRouteId: string): Promise { + try { + const ack = await callInternal( + "duplexChannelClose", + { hostRouteId }, + duplexChannelCloseTimeoutMs, + ); + return isRecord(ack) && readNonEmptyString(ack.hostRouteId) === hostRouteId; + } catch { + return false; + } + } + + // Terminalize the route exactly once. Resolve the wait, close the worker + // channel by the host route identifier, and free the per-worker slot only + // after the close resolves. Retire the worker when the close is unconfirmed. + // Clear the route lifetime timer one time. Every terminal path and the + // worker-exit path calls this, so a timer never fires after the route ends. + function clearDuplexChannelLifetimeTimer(route: DuplexChannelRoute): void { + if (route.lifetimeTimer) { + clearTimeout(route.lifetimeTimer); + route.lifetimeTimer = null; + } + } + + async function terminalizeDuplexChannelRoute(route: DuplexChannelRoute): Promise { + if (route.terminalized) return; + route.terminalized = true; + route.state = "closed"; + route.listener = null; + route.buffered = []; + route.bufferedChars = 0; + clearDuplexChannelLifetimeTimer(route); + // A terminalized route reports a null exit code, which the caller treats as a + // failure. + settleRouteWait(route, { exitCode: null }); + const confirmed = await closeDuplexChannelTerminal(route.hostRouteId); + if (duplexChannelRoute === route) duplexChannelRoute = null; + if (!confirmed) { + // The worker did not acknowledge the close, so the host cannot prove the + // channel is gone. Fail closed: retire the worker before any reuse. + log.error( + { pluginId }, + "duplex channel close not acknowledged; retiring worker", + ); + void killProcess(); + } + } + + // Count one protocol error for the route. End the route when the count passes + // the per-route budget, so a flood of malformed or mismatched frames bounds the + // route. + function recordDuplexChannelProtocolError(route: DuplexChannelRoute): void { + route.protocolErrors += 1; + if (route.protocolErrors > maxDuplexChannelProtocolErrors) { + void terminalizeDuplexChannelRoute(route); + } + } + + // Deliver one duplex channel chunk to the bound listener in isolation. A + // listener that throws must not escape the worker stdout notification handler + // or the buffered replay, so a throw here breaks neither the notification + // dispatch loop nor the pre-bind drain. The manager catches the error and logs + // it without the raw bytes. This mirrors the `execute.log` delivery isolation. + function deliverDuplexChannelChunk( + listener: (chunk: string) => void, + chunk: string, + ): void { + try { + listener(chunk); + } catch (err) { + log.error( + { err: err instanceof Error ? err.message : String(err) }, + "duplex channel data delivery threw", + ); + } + } + + // Route one duplex channel data notification to the per-session listener. + // Deliver only while the route is `open` and the notification carries the exact + // bound worker session identifier and a valid chunk. Count a mismatched or + // malformed frame as a protocol error. End the route at once when one chunk is + // larger than the per-chunk limit or when the cumulative bytes pass the total + // cap. Buffer a valid frame under the pre-bind bounds when no listener has + // attached yet. Never log the raw bytes. + function routeDuplexChannelData(notification: JsonRpcNotification): void { + const route = duplexChannelRoute; + if (!route || route.state !== "open") return; + const params = isRecord(notification.params) ? notification.params : {}; + const workerSessionId = readNonEmptyString(params.workerSessionId); + const chunk = params.chunk; + if ( + !workerSessionId || + workerSessionId !== route.workerSessionId || + typeof chunk !== "string" || + chunk.length === 0 + ) { + // A late, unknown, malformed, or mismatched frame. Drop it and count one + // protocol error. + recordDuplexChannelProtocolError(route); + return; + } + if (chunk.length > maxDuplexChannelChunkChars) { + // One inbound chunk is larger than the per-chunk limit. End the route at + // once. Do not count the chunk as a protocol error. + void terminalizeDuplexChannelRoute(route); + return; + } + // Count the bytes of the chunk. Enforce the cumulative total-byte cap before + // and after a listener attaches. End the route when the cap is exceeded, so a + // bound listener cannot receive data past the cap. + const chunkBytes = Buffer.byteLength(chunk); + if (route.totalDataBytes + chunkBytes > maxDuplexChannelTotalDataBytes) { + void terminalizeDuplexChannelRoute(route); + return; + } + route.totalDataBytes += chunkBytes; + if (route.listener) { + deliverDuplexChannelChunk(route.listener, chunk); + return; + } + // No listener attached yet. Buffer the frame under the pre-bind bounds. End + // the route when the cumulative bytes or the frame count passes the bound. + if ( + route.buffered.length + 1 > maxDuplexChannelPreBindFrames || + route.bufferedChars + chunk.length > maxDuplexChannelPreBindChars + ) { + void terminalizeDuplexChannelRoute(route); + return; + } + route.buffered.push(chunk); + route.bufferedChars += chunk.length; + } + + // Route one duplex channel exit notification to the wait. Resolve only while + // the route is `open` and the notification carries the exact bound worker + // session identifier. + function routeDuplexChannelExit(notification: JsonRpcNotification): void { + const route = duplexChannelRoute; + if (!route || route.state !== "open") return; + const params = isRecord(notification.params) ? notification.params : {}; + const workerSessionId = readNonEmptyString(params.workerSessionId); + if (!workerSessionId || workerSessionId !== route.workerSessionId) return; + const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; + settleRouteWait(route, { exitCode }); + } + + // Close the one route on a worker exit. The worker is gone, so the manager + // resolves the wait with the fixed non-secret exit and clears the route one + // time. The pending channel calls reject through `rejectAllPending`. + function closeDuplexChannelRouteOnWorkerExit(): void { + const route = duplexChannelRoute; + if (!route) return; + duplexChannelRoute = null; + route.terminalized = true; + route.state = "closed"; + route.listener = null; + route.buffered = []; + route.bufferedChars = 0; + clearDuplexChannelLifetimeTimer(route); + settleRouteWait(route, { exitCode: null }); + } + + // Open one live generic duplex channel route. Reserve the route before the open + // call, bind the worker session identifier one time on the first successful + // open reply, and return a session a caller drives. Terminalize the route on + // every open failure path. + async function openDuplexChannel( + input: DuplexChannelOpenInput, + ): Promise { + if (duplexChannelRoute) { + // A route for this worker is not yet closed and confirmed. Reject the + // second open with one fixed non-secret error before it reaches the worker. + throw new Error(DUPLEX_CHANNEL_ROUTE_BUSY); + } + const hostRouteId = randomUUID(); + let settleWait: (value: { exitCode: number | null }) => void = () => {}; + const waitPromise = new Promise<{ exitCode: number | null }>((resolve) => { + settleWait = resolve; + }); + const route: DuplexChannelRoute = { + hostRouteId, + state: "reserved", + workerSessionId: null, + listener: null, + buffered: [], + bufferedChars: 0, + pendingRequests: 0, + protocolErrors: 0, + totalDataBytes: 0, + lifetimeTimer: null, + terminalized: false, + settleWait, + }; + duplexChannelRoute = route; + + route.state = "opening"; + let openResult: HostToWorkerMethods["duplexChannelOpen"][1]; + try { + openResult = await callInternal( + "duplexChannelOpen", + { + hostRouteId, + driverKey: input.driverKey, + companyId: input.companyId, + environmentId: input.environmentId, + providerLeaseId: input.providerLeaseId, + command: input.command, + }, + duplexChannelOpenTimeoutMs, + ); + } catch (err) { + // A send failure, an RPC rejection, or an open timeout. Terminalize the + // route exactly once and fail closed. + await terminalizeDuplexChannelRoute(route); + throw err instanceof Error ? err : new Error(DUPLEX_CHANNEL_OPEN_FAILED); + } + + const workerSessionId = readBindableWorkerSessionId(route, openResult); + if (!workerSessionId) { + // A malformed reply, or a route that already left `opening`. A late or a + // duplicate reply never binds, revives, or reopens a route. + await terminalizeDuplexChannelRoute(route); + throw new Error(DUPLEX_CHANNEL_OPEN_FAILED); + } + // Bind the worker session identifier one time and move the route to `open`. + route.workerSessionId = workerSessionId; + route.state = "open"; + + // Start the route lifetime timer now the route is open. The route ends when + // the timer expires. Every terminal path and the worker-exit path clears the + // timer. Unreference the timer so it never blocks the host process shutdown. + route.lifetimeTimer = setTimeout(() => { + void terminalizeDuplexChannelRoute(route); + }, maxDuplexChannelDurationMs); + route.lifetimeTimer.unref?.(); + + // Send one host→worker request under the pending-request bound. End the route + // when too many requests are in-flight, so a worker that never replies cannot + // make the host hold an unbounded number of pending requests. + const sendBoundedRequest = < + M extends "duplexChannelWrite" | "duplexChannelStop", + >( + method: M, + params: HostToWorkerMethods[M][0], + ): void => { + if (route.state !== "open") return; + if (route.pendingRequests >= maxDuplexChannelPendingRequests) { + void terminalizeDuplexChannelRoute(route); + return; + } + route.pendingRequests += 1; + void callInternal(method, params, duplexChannelOpenTimeoutMs) + .catch(() => {}) + .finally(() => { + route.pendingRequests -= 1; + }); + }; + + return { + onData(listener: (chunk: string) => void): void { + route.listener = listener; + if (route.buffered.length > 0) { + const pending = route.buffered; + route.buffered = []; + route.bufferedChars = 0; + for (const chunk of pending) deliverDuplexChannelChunk(listener, chunk); + } + }, + write(data: string): void { + const sid = route.workerSessionId; + if (route.state !== "open" || !sid) return; + if (data.length > maxDuplexChannelWriteChars) { + // The write is larger than the size bound. End the route before the + // write reaches the worker. + void terminalizeDuplexChannelRoute(route); + return; + } + sendBoundedRequest("duplexChannelWrite", { workerSessionId: sid, data }); + }, + wait(): Promise<{ exitCode: number | null }> { + return waitPromise; + }, + kill(): void { + const sid = route.workerSessionId; + if (!sid) return; + sendBoundedRequest("duplexChannelStop", { workerSessionId: sid }); + }, + async close(): Promise { + await terminalizeDuplexChannelRoute(route); + }, + }; + } + /** * Extract the single company a worker→host call references, mirroring the SDK * governed-access gate's own derivation (host-client-factory.ts @@ -1440,6 +1960,18 @@ export function createPluginWorkerHandle( return; } + // Duplex channel notifications: deliver data and the exit to the one + // host-owned duplex route, bound by the worker session identifier while the + // route is open. + if (notification.method === DUPLEX_CHANNEL_DATA_NOTIFICATION) { + routeDuplexChannelData(notification); + return; + } + if (notification.method === DUPLEX_CHANNEL_EXIT_NOTIFICATION) { + routeDuplexChannelExit(notification); + return; + } + // Stream notifications: forward to the stream bus via callback if ( notification.method === "streams.open" || @@ -1590,6 +2122,10 @@ export function createPluginWorkerHandle( // already rejected through `rejectAllPending`. closeSetupTokenPtyRouteOnWorkerExit(); + // Close the one duplex channel route the same way. The pending channel calls + // already rejected through `rejectAllPending`. + closeDuplexChannelRouteOnWorkerExit(); + // Emit synthetic close for any orphaned stream channels so SSE clients // are notified instead of hanging indefinitely. if (openStreamChannels.size > 0 && options.onStreamNotification) { @@ -2063,6 +2599,17 @@ export function createPluginWorkerHandle( return openSetupTokenPtySession(input); }, + openDuplexChannel(input: DuplexChannelOpenInput) { + if (status !== "running" && status !== "starting") { + return Promise.reject( + new Error( + `Cannot open a duplex channel — worker for "${pluginId}" is ${status}`, + ), + ); + } + return openDuplexChannel(input); + }, + notify(method: string, params: unknown) { if (status !== "running") return; const invocationScope = deriveInvocationScope(method, params);