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 <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach 2026-08-19 13:39:13 -07:00 committed by GitHub
parent bd059a073d
commit 8161244284
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
22 changed files with 3304 additions and 25 deletions

View File

@ -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<void>;
}
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<SandboxSyncResult>;
/** Optional native outbound file transfer. See {@link syncIn}. */
syncOut?(operations: SandboxSyncOperation[]): Promise<SandboxSyncResult>;
/**
* 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<CommandManagedDuplexChannel>;
}
export interface CommandManagedRuntimeSpec {

View File

@ -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);
});
});

View File

@ -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<string, string>;
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<string, string>;
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<string> = new Set<DuplexResponseOutcome>([
"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<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value);
}
function isStringRecord(value: unknown): value is Record<string, string> {
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<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<string, unknown>): 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<string, unknown>): DuplexDecodeResult {
return ok(frame as unknown as DuplexHeartbeatFrame);
}
function validateClose(frame: Record<string, unknown>): DuplexDecodeResult {
return ok(frame as unknown as DuplexCloseFrame);
}
function validateError(frame: Record<string, unknown>): 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;
}
}

View File

@ -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"
}
]
}
]
}

View File

@ -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,
};
}

View File

@ -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<PluginSetupTokenPtyCloseResult>;
/**
* 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 workerhost 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<PluginDuplexChannelOpenResult>;
/** Called to write raw input to an open duplex channel, keyed by the worker session identifier. */
onDuplexChannelWrite?(params: PluginDuplexChannelWriteParams): Promise<void>;
/** Called to stop an open duplex channel child, keyed by the worker session identifier. */
onDuplexChannelStop?(params: PluginDuplexChannelStopParams): Promise<void>;
/**
* 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<PluginDuplexChannelCloseResult>;
}
// ---------------------------------------------------------------------------

View File

@ -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";

View File

@ -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();
});
});

View File

@ -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 hostworker 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;
// ---------------------------------------------------------------------------

View File

@ -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
// -----------------------------------------------------------------------

View File

@ -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<string, (response: JsonRpcResponse) => 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<unknown>((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<string, (response: JsonRpcResponse) => 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<unknown>((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();
}
});
});

View File

@ -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"],
},
});

View File

@ -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 {

View File

@ -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<typeof sandboxProviderCapabilitiesSchema>;

View File

@ -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

View File

@ -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<typeof import("../services/plugin-environment-driver.js")>()),
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<void> {
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();
});
});

View File

@ -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}` },
});
});

View File

@ -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<string, unknown>) {
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<void>((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);
}
});
});

View File

@ -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,

View File

@ -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,
};

View File

@ -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<SandboxCapabilityKey> = new Set([
"incrementalSessionOutput",
"concurrentSyncOperations",
"duplexCommandStream",
]);
/**
@ -142,6 +158,11 @@ const SANDBOX_CAPABILITY_OPT_IN_KEYS: ReadonlySet<SandboxCapabilityKey> = 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<SandboxCapabilityKey, readonly (readonly string[])[]> = {
// Reusable leases require ALL reuse verbs. Each verb is its own required
@ -156,6 +177,7 @@ const SANDBOX_CAPABILITY_PREREQUISITE_METHODS: Record<SandboxCapabilityKey, read
independentControlCommands: [["environmentExecute"]],
incrementalSessionOutput: [["environmentExecute"]],
concurrentSyncOperations: [["environmentSyncIn"], ["environmentSyncOut"]],
duplexCommandStream: [["duplexChannelOpen"]],
};
function capabilityIsVerified(
@ -240,6 +262,7 @@ export function resolveEffectiveSandboxCapabilities(input: {
independentControlCommands: resolve("independentControlCommands"),
incrementalSessionOutput: resolve("incrementalSessionOutput"),
concurrentSyncOperations: resolve("concurrentSyncOperations"),
duplexCommandStream: resolve("duplexCommandStream"),
};
}
@ -254,9 +277,9 @@ export function resolveEffectiveSandboxCapabilities(input: {
* which falls back for a `job` backend or a `nativeFileSyncUnsupported` lease).
* - `configResolutionFailed` marks that the runtime could not resolve the
* provider config. A provider whose config cannot be resolved is untrusted, so
* the runtime fails closed and narrows `persistentProcessSessions` and
* `incrementalSessionOutput` to false. An empty config alone does not fail
* closed; only a resolution error does.
* the runtime fails closed and narrows `persistentProcessSessions`,
* `incrementalSessionOutput`, and `duplexCommandStream` to false. An empty
* config alone does not fail closed; only a resolution error does.
*/
export function buildSandboxCapabilityNarrowing(input: {
leasePolicy?: EnvironmentLease["leasePolicy"] | null;
@ -275,11 +298,15 @@ export function buildSandboxCapabilityNarrowing(input: {
if (input.configResolutionFailed === true) {
// The runtime could not resolve the provider config, so it fails closed and
// denies persistent process sessions and incremental session output. The
// session-output streaming gate reads `incrementalSessionOutput`, so it must
// narrow with the persistent-session gate to keep the fail-closed behavior.
// denies persistent process sessions, incremental session output, and the
// duplex command stream. The session-output streaming gate reads
// `incrementalSessionOutput`, so it must narrow with the persistent-session
// gate to keep the fail-closed behavior. The duplex command stream opens a
// host-owned bidirectional channel, so an untrusted provider must not keep
// it either.
narrowing.persistentProcessSessions = false;
narrowing.incrementalSessionOutput = false;
narrowing.duplexCommandStream = false;
}
return narrowing;
@ -470,6 +497,11 @@ export interface EnvironmentDriverSyncInput extends EnvironmentDriverLeaseInput
operations: PluginSyncOperation[];
}
export interface EnvironmentDriverOpenDuplexChannelInput extends EnvironmentDriverLeaseInput {
/** The command line the sandbox runs as the duplex channel child process. */
command: string;
}
export interface EnvironmentRuntimeDriver {
readonly driver: string;
acquireRunLease(input: EnvironmentDriverAcquireInput): Promise<EnvironmentLease>;
@ -486,6 +518,15 @@ export interface EnvironmentRuntimeDriver {
*/
syncIn?(input: EnvironmentDriverSyncInput): Promise<PluginEnvironmentSyncResult>;
syncOut?(input: EnvironmentDriverSyncInput): Promise<PluginEnvironmentSyncResult>;
/**
* 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<CommandManagedDuplexChannel>;
/** 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<void> {
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<CommandManagedDuplexChannel> {
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);
},
};
}

View File

@ -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 hostworker 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
* hostworker 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<void>;
}
/**
* 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<void>;
}
/**
* 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<SetupTokenPtyHostSession>;
/**
* 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<DuplexChannelHostSession>;
/**
* 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<typeof setTimeout> | 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<boolean> {
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<void> {
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<DuplexChannelHostSession> {
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<void> {
await terminalizeDuplexChannelRoute(route);
},
};
}
/**
* Extract the single company a workerhost 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);