feat(duplex): run the Daytona sandbox callback bridge over Node HTTP/2 (#12120)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - Sandbox providers carry agent work through controlled execution
channels
> - The Daytona callback bridge uses a bespoke line-framed protocol over
its duplex channel
> - The bespoke protocol adds framing work and does not use the Node
transport that already supports multiplexed streams
> - This pull request carries raw bytes across the channel, adds a Node
HTTP/2 bridge, and selects it for Daytona
> - The benefit is one authenticated, multiplexed callback session with
queue_v1 as the bounded fallback

## Linked Issues or Issue Description

**Subsystem affected**

The packages/plugins Daytona provider and the shared duplex execution
path.

**Problem or motivation**

The Daytona callback bridge uses a bespoke line-framed protocol over the
provider duplex channel. This adds protocol work and limits stream
handling.

**Proposed solution**

Carry raw bytes through the cross-layer channel. Add an authenticated
Node HTTP/2 host server and sandbox client gateway. Select http2_v1 for
Daytona and retain queue_v1 as the fallback.

**Alternatives considered**

Keep the current duplex_v1 protocol. This keeps the bespoke framing path
and does not provide one HTTP/2 session for callback streams.

**Roadmap alignment**

ROADMAP.md lists Daytona under cloud and sandbox agents. This change
improves the shipped Daytona provider path.

**Additional context**

The branch adds no dependency. Node 24 provides the http2 module. The
host token check and canonical path parser remain the single dispatch
path.

## What Changed

- Carry raw Uint8Array chunks through the adapter, plugin, worker,
runtime, and Daytona layers.
- Encode bytes as base64 only across the JSON-RPC hop, because JSON has
no binary type.
- Add the bounded host HTTP/2 server and the in-sandbox HTTP/2 client
gateway.
- Authenticate every stream with the per-run bridge token before route
work.
- Parse the path once and reuse the canonical result for route and
forwarding work.
- Select http2_v1 for Daytona and fall back once to queue_v1 when the
client preface is absent.
- Add transport, session, stream, and fallback telemetry.
- Mark HTTP/2 as the preferred transport and queue_v1 as the
soft-deprecated fallback.

## Verification

- `npx vitest run packages/adapter-utils/src` — 990 passed and 4
skipped.
- `npx vitest run
server/src/__tests__/plugin-worker-manager-duplex.test.ts` — 32 passed.
- `npx vitest run --config
packages/plugins/sandbox-providers/daytona/vitest.config.ts` — 220
passed and 6 skipped.
- `npx tsc --noEmit` in `packages/adapter-utils`, `packages/shared`,
`packages/plugins/sdk`, and `server` — clean.
- No `package.json` or `pnpm-lock.yaml` file changed.
- The live Daytona test skips when `DAYTONA_API_KEY` is absent.
- The root `npx tsc --noEmit` command has a pre-existing missing
`packages/adapters/droid-local` reference on this branch and on
`master`.

## Risks

- The transport change affects several duplex layers and could expose
byte-boundary errors.
- A missing HTTP/2 client preface falls back once to queue_v1 and
records `preface_missing`.
- The host token check and canonical path parser must remain on the
shared dispatch path.
- The live Daytona test needs `DAYTONA_API_KEY` and does not run in this
agent sandbox.

## Model Used

OpenAI GPT-5, tool-enabled coding agent with repository inspection,
GitHub CLI, and shell execution.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach 2026-08-25 07:35:39 -07:00 committed by GitHub
parent 0f0e544317
commit 445547c989
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
43 changed files with 5631 additions and 740 deletions

View File

@ -1610,7 +1610,13 @@ describe("worktree helpers", () => {
fs.rmSync(tempRoot, { recursive: true, force: true });
}
},
30_000,
// This test starts three separate embedded Postgres lifecycles (the
// source database, the worktree init's internal target database, and a
// third instance opened here to verify the seeded rows), so it needs
// more headroom than the other embedded-Postgres tests in this file.
// It normally finishes in well under 10s; the 60s budget absorbs CI
// runner contention without masking a real hang.
60_000,
);
itEmbeddedPostgres(

View File

@ -419,9 +419,9 @@ key never reaches a sink by accident.
| Key | Type | Optional | Value set |
| --- | --- | --- | --- |
| `provider` | string | no | `daytona`, or `other` for any other plugin key. |
| `transport` | string | no | `duplex` or `file`. A fallback record uses `file`; every other record uses `duplex`. |
| `transport` | string | no | `duplex`, `http2`, or `file`. `duplex` names the retired bespoke frame protocol; `http2` names the Node HTTP/2 session over the sandbox channel; a fallback record uses `file`. |
| `outcome` | string | yes | `ok` or `error`. |
| `fallback_reason` | string | yes | `gate_off`, `capability_absent`, `open_failed`, `ready_invalid`, `ready_nonce_mismatch`, `ready_timeout`, or `contaminated`. It rides only a fallback record. |
| `fallback_reason` | string | yes | `gate_off`, `capability_absent`, `route_busy`, `entrypoint_sync_failed`, `broker_construction_failed`, `channel_open_failed`, `ready_invalid`, `ready_nonce_mismatch`, `ready_timeout`, `contaminated`, `aggregate_bytes_exceeded`, or `preface_missing`. It rides only a fallback record. `route_busy` marks the process-scoped route ceiling full. `entrypoint_sync_failed` and `broker_construction_failed` mark the named build step. `channel_open_failed` marks a failed channel open. `aggregate_bytes_exceeded` marks a readiness handshake, or an `http2` post-preface pre-bind buffer, where the host fell back because the process aggregate byte ceiling had no room. `preface_missing` marks a missing or an invalid HTTP/2 client connection preface inside the bounded readiness buffer: the host found no valid preface after the accepted READY line, aborted the `http2` open, and moved the run to the file bridge (`queue_v1`) one time. |
| `loss_class` | string | yes | `pre_dispatch` or `post_dispatch`, relative to the first request dispatch. It rides only a loss record. |
| `loss_reason` | string | yes | `stdin_eof`, `provider_exit`, `heartbeat_timeout`, `rpc_failure`, `write_error`, `transport_closed`, or `other`. The host maps every loss cause to one of these values, so no raw provider text reaches a sink. `write_error` marks a rejected host-to-sandbox write. `transport_closed` marks a reason-less provider transport close with no exit data. It rides only a loss record. |

View File

@ -8,11 +8,34 @@ import { afterEach, describe, expect, it } from "vitest";
import {
createCommandManagedRuntimeClient,
prepareCommandManagedRuntime,
type CommandManagedDuplexChannel,
type CommandManagedRuntimeRunner,
} from "./command-managed-runtime.js";
import type { SandboxSyncOperation } from "./sandbox-managed-runtime.js";
import type { RunProcessResult } from "./server-utils.js";
/**
* An in-memory fake `CommandManagedDuplexChannel`. It holds no real process; a
* write echoes straight to the one registered data listener, so a test proves
* the channel contract carries a `Uint8Array` chunk with no string coercion in
* between. The fake never exits on its own; a test calls the listener it
* registers with `onExit` only when it needs one.
*/
function createFakeEchoDuplexChannel(): CommandManagedDuplexChannel {
let dataListener: ((chunk: Uint8Array) => void) | null = null;
return {
write(data: Uint8Array): void {
dataListener?.(data);
},
onData(listener: (chunk: Uint8Array) => void): void {
dataListener = listener;
},
onExit(): void {},
stop(): void {},
close: async (): Promise<void> => {},
};
}
const execFile = promisify(execFileCallback);
interface SpawnRunnerHandle {
@ -1057,4 +1080,22 @@ describe("command managed runtime", () => {
/stdout: tar: workspace-download\.tar: Cannot open: Permission denied/,
);
});
it("test_channel_round_trips_all_byte_values", () => {
const channel = createFakeEchoDuplexChannel();
const allByteValues = Uint8Array.from({ length: 256 }, (_, value) => value);
const received: Uint8Array[] = [];
channel.onData((chunk) => {
received.push(chunk);
});
channel.write(allByteValues);
expect(received).toHaveLength(1);
// A byte value of zero must survive. A UTF-8 string channel loses it: a
// JavaScript string can hold the code point U+0000, but a C-style consumer
// downstream of a string channel often treats it as a terminator.
expect(received[0]).toEqual(allByteValues);
expect(Array.from(received[0] ?? [])).toEqual(Array.from({ length: 256 }, (_, value) => value));
});
});

View File

@ -37,9 +37,9 @@ export interface DuplexChannelOpenInput {
*/
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;
write(data: Uint8Array): void;
/** Registers the one data listener. The channel streams each raw byte chunk in order. */
onData(listener: (chunk: Uint8Array) => void): void;
/**
* Registers the one exit listener. The channel calls it one time with the exit.
* A numeric `exitCode` is a real process exit. `transportClosed` is true when the
@ -117,6 +117,8 @@ export interface CommandManagedRuntimeRunner {
* 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}.
*
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
*/
openDuplexChannel?(input: DuplexChannelOpenInput): Promise<CommandManagedDuplexChannel>;
}

View File

@ -163,6 +163,8 @@ describe("DuplexAggregateByteLedger", () => {
"decoder_buffer",
"readiness_buffer",
"readiness_replay",
"http2_preface_scan",
"http2_preface_replay",
"pending_write",
"stdin_write",
]);

View File

@ -80,6 +80,17 @@ export const MIN_HOST_PROCESS_MEMORY_BYTES_FOR_DUPLEX = 2 * 1024 * 1024 * 1024;
* frame decoder charges its own retention. This is a separate retention from
* `readiness_buffer`: the gate drops the whole pre-READY buffer on READY, then
* charges only the retained suffix and each later pre-bind chunk under this owner.
* - `http2_preface_scan`: the raw untrusted bytes the `http2_v1` preface scan
* retains while it searches for the client connection preface, from the
* readiness-replay handoff until the preface match or the scan cap. This
* is a separate retention from `readiness_replay`: it starts only after
* the readiness gate hands its own retained suffix to the preface scan.
* - `http2_preface_replay`: the bytes the `http2_v1` preface scan holds after the
* client connection preface, from the preface match until the HTTP/2 server
* binds its downstream listener. This is a separate retention from
* `http2_preface_scan`: the scan drops its own buffer on the preface match,
* then charges only the retained suffix and each later pre-bind chunk under
* this owner.
* - `pending_write`: the raw host-to-worker write payload a pending duplex write
* RPC retains, from the enqueue seam until the RPC settles.
* - `stdin_write`: the serialized host-to-worker frame the child-stdin transport
@ -99,6 +110,8 @@ export const DUPLEX_AGGREGATE_TOKEN_OWNERS = [
"decoder_buffer",
"readiness_buffer",
"readiness_replay",
"http2_preface_scan",
"http2_preface_replay",
"pending_write",
"stdin_write",
] as const;

View File

@ -85,7 +85,7 @@ function createFakeChannelHarness(): FakeChannelHarness {
string,
{ frame: DuplexResponseFrame; received: number; chunks: Buffer[] }
>();
let dataListener: ((chunk: string) => void) | null = null;
let dataListener: ((chunk: Uint8Array) => void) | null = null;
let exitListener: ((exit: { exitCode: number | null }) => void) | null = null;
const channel: CommandManagedDuplexChannel = {
@ -129,14 +129,14 @@ function createFakeChannelHarness(): FakeChannelHarness {
channel,
feed: ({ frame, bodyText }) => {
if (!dataListener) throw new Error("The broker did not bind the data listener.");
dataListener(encodeDuplexFrame(frame));
dataListener(Buffer.from(encodeDuplexFrame(frame), "utf8"));
if (bodyText.length > 0) {
for (const chunk of splitBodyIntoChunkFrames(
frame.id,
Buffer.from(bodyText, "utf8"),
DUPLEX_FRAME_VERSION,
)) {
dataListener(encodeDuplexFrame(chunk));
dataListener(Buffer.from(encodeDuplexFrame(chunk), "utf8"));
}
}
},

View File

@ -111,7 +111,7 @@ function createFakeChannelHarness(): FakeChannelHarness {
string,
{ frame: DuplexResponseFrame; received: number; chunks: Buffer[] }
>();
let dataListener: ((chunk: string) => void) | null = null;
let dataListener: ((chunk: Uint8Array) => void) | null = null;
const channel: CommandManagedDuplexChannel = {
write: (data) => {
@ -152,14 +152,14 @@ function createFakeChannelHarness(): FakeChannelHarness {
channel,
feed: ({ frame, bodyText }) => {
if (!dataListener) throw new Error("The broker did not bind the data listener.");
dataListener(encodeDuplexFrame(frame));
dataListener(Buffer.from(encodeDuplexFrame(frame), "utf8"));
if (bodyText.length > 0) {
const chunks = splitBodyIntoChunkFrames(
frame.id,
Buffer.from(bodyText, "utf8"),
DUPLEX_FRAME_VERSION,
);
for (const chunk of chunks) dataListener(encodeDuplexFrame(chunk));
for (const chunk of chunks) dataListener(Buffer.from(encodeDuplexFrame(chunk), "utf8"));
}
},
responses,

View File

@ -699,7 +699,10 @@ export async function createDuplexBridgeBroker(
const writeLine = (line: string): boolean => {
try {
channel.write(line);
// The channel carries raw bytes (see `ChannelBytesWireValue` in the plugin
// SDK's protocol.ts). Encode the frame's UTF-8 text to bytes at this one
// write seam.
channel.write(Buffer.from(line, "utf8"));
return true;
} catch (error) {
recordLoss("stream_failure", errorMessage(error));
@ -1337,7 +1340,7 @@ export async function createDuplexBridgeBroker(
}
};
const onData = (chunk: string): void => {
const onData = (chunk: Uint8Array): void => {
if (stopped) return;
const results = decoder.push(chunk);
for (const result of results) {
@ -1364,7 +1367,7 @@ export async function createDuplexBridgeBroker(
const sendHeartbeat = (): void => {
if (state !== "open") return;
try {
channel.write(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }));
channel.write(Buffer.from(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }), "utf8"));
} catch (error) {
recordLoss("heartbeat_write_failure", errorMessage(error));
}
@ -1387,7 +1390,7 @@ export async function createDuplexBridgeBroker(
// Send an orderly close frame. Ignore a write failure here; the broker is
// already closing, so a dead channel needs no loss record.
try {
channel.write(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "close" }));
channel.write(Buffer.from(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "close" }), "utf8"));
} catch (error) {
options.logger?.(`Duplex broker could not send the close frame: ${errorMessage(error)}`);
}

View File

@ -5,7 +5,6 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";
import { Readable } from "node:stream";
import { StringDecoder } from "node:string_decoder";
import { afterEach, describe, expect, it } from "vitest";
import {
@ -145,20 +144,15 @@ interface ChildDuplexChannel {
* writes to the child stdin, reads the child stdout, and learns of the child
* exit through this channel.
*
* The host end reads a byte stream from the stdout pipe. A pipe read can split
* one multi-byte UTF-8 character across two chunks. The `StringDecoder` holds
* the bytes of an incomplete character until the next chunk, so the broker only
* ever reads whole characters. The channel keeps a second decoder for
* observation only; it lets the harness assert the READY frame and the request
* frames the child produced, and it never feeds the broker.
* The host end forwards each raw stdout chunk unchanged; the channel carries
* bytes, so no decode step sits between the pipe and the broker.
*/
function attachChildDuplexChannel(child: ChildProcessWithoutNullStreams): ChildDuplexChannel {
const observed = new DuplexFrameDecoder();
const observedFrames: DuplexFrame[] = [];
const stdoutDecoder = new StringDecoder("utf8");
let dataListener: ((chunk: string) => void) | null = null;
let dataListener: ((chunk: Uint8Array) => void) | null = null;
let exitListener: ((exit: { exitCode: number | null }) => void) | null = null;
let pendingText = "";
let pendingBytes: Buffer = Buffer.alloc(0);
let pendingExit: { exitCode: number | null } | null = null;
let stderrText = "";
@ -166,10 +160,9 @@ function attachChildDuplexChannel(child: ChildProcessWithoutNullStreams): ChildD
for (const result of observed.push(buffer)) {
if (result.ok) observedFrames.push(result.frame);
}
const text = stdoutDecoder.write(buffer);
if (text.length === 0) return;
if (dataListener) dataListener(text);
else pendingText += text;
if (buffer.length === 0) return;
if (dataListener) dataListener(buffer);
else pendingBytes = pendingBytes.length === 0 ? buffer : Buffer.concat([pendingBytes, buffer]);
});
child.stderr.on("data", (buffer: Buffer) => {
stderrText += buffer.toString("utf8");
@ -189,9 +182,9 @@ function attachChildDuplexChannel(child: ChildProcessWithoutNullStreams): ChildD
},
onData: (listener) => {
dataListener = listener;
if (pendingText.length > 0) {
const replay = pendingText;
pendingText = "";
if (pendingBytes.length > 0) {
const replay = pendingBytes;
pendingBytes = Buffer.alloc(0);
listener(replay);
}
},

View File

@ -15,6 +15,13 @@
* 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.
*
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
* The `http2_v1` host readiness gate imports {@link decodeDuplexLine} from this
* file to read the one READY line every gateway sends, so this file's READY
* frame path stays live for both transports. `duplex-bridge-broker.ts` and
* `duplex-body-spool.ts` still import the request, response, and body-chunk
* frame types from this file, so this phase keeps every frame type here.
*/
import {

View File

@ -0,0 +1,70 @@
import { describe, expect, it } from "vitest";
import {
DUPLEX_LOSS_REASONS,
HTTP2_TELEMETRY_EVENT_NAMES,
mapHttp2EventToDuplexLossReason,
normalizeDuplexLossReason,
normalizeDuplexProvider,
type DuplexFallbackReason,
type DuplexTransportValue,
} from "./duplex-observability.js";
describe("duplex observability: HTTP/2 event mapping (accepted security fix 7)", () => {
it("test_http2_events_map_to_the_loss_taxonomy", () => {
// Every closed HTTP/2 event name maps to a value from the existing,
// closed DuplexLossReason set. The map reuses one taxonomy across every
// transport, so no separate HTTP/2-only reason list exists.
for (const event of HTTP2_TELEMETRY_EVENT_NAMES) {
const reason = mapHttp2EventToDuplexLossReason(event);
expect(DUPLEX_LOSS_REASONS).toContain(reason);
}
expect(mapHttp2EventToDuplexLossReason("session_error")).toBe("rpc_failure");
expect(mapHttp2EventToDuplexLossReason("session_goaway")).toBe("transport_closed");
expect(mapHttp2EventToDuplexLossReason("session_stall")).toBe("heartbeat_timeout");
expect(mapHttp2EventToDuplexLossReason("write_error")).toBe("write_error");
expect(mapHttp2EventToDuplexLossReason("transport_closed")).toBe("transport_closed");
expect(mapHttp2EventToDuplexLossReason("channel_exit")).toBe("provider_exit");
});
it("test_an_unknown_telemetry_input_maps_to_other", () => {
// An unknown event name, a raw provider error string, and a missing value
// all map to `other`. No raw text ever reaches a sink through this map.
expect(mapHttp2EventToDuplexLossReason("ECONNRESET: read failed at socket.js:42")).toBe("other");
expect(mapHttp2EventToDuplexLossReason("some_future_event")).toBe("other");
expect(mapHttp2EventToDuplexLossReason(null)).toBe("other");
expect(mapHttp2EventToDuplexLossReason(undefined)).toBe("other");
expect(mapHttp2EventToDuplexLossReason("")).toBe("other");
});
it("keeps the closed HTTP/2 event-name set to the seven named signals", () => {
expect([...HTTP2_TELEMETRY_EVENT_NAMES].sort()).toEqual(
[
"session_error",
"session_goaway",
"session_stall",
"write_error",
"transport_closed",
"channel_exit",
].sort(),
);
});
});
describe("duplex observability: closed transport and fallback-reason values", () => {
it("accepts the http2 transport value and the preface_missing fallback reason", () => {
// A type-level check: these string literals must widen to the exported
// union types with no cast, so a drift in either union breaks the build.
const transport: DuplexTransportValue = "http2";
const fallbackReason: DuplexFallbackReason = "preface_missing";
expect(transport).toBe("http2");
expect(fallbackReason).toBe("preface_missing");
});
});
describe("duplex observability: existing normalization stays intact", () => {
it("still maps an unknown loss cause and an unknown provider key to their closed defaults", () => {
expect(normalizeDuplexLossReason("not_a_real_reason")).toBe("other");
expect(normalizeDuplexProvider("some-unlisted-plugin")).toBe("other");
});
});

View File

@ -86,8 +86,12 @@ export const DUPLEX_DIMENSION_KEYS = [
/** One dimension key from the closed set. */
export type DuplexDimensionKey = (typeof DUPLEX_DIMENSION_KEYS)[number];
/** The transport a record is about. */
export type DuplexTransportValue = "duplex" | "file";
/**
* The transport a record is about. `duplex` names the retired bespoke frame
* protocol; `http2` names the Node HTTP/2 session over the sandbox channel;
* `file` names the queue-file bridge.
*/
export type DuplexTransportValue = "duplex" | "http2" | "file";
/** The outcome of a record. */
export type DuplexOutcomeValue = "ok" | "error";
@ -98,9 +102,13 @@ export type DuplexOutcomeValue = "ok" | "error";
* stage: the process-scoped route ceiling was full (`route_busy`), the entrypoint
* sync failed (`entrypoint_sync_failed`), the broker construction failed
* (`broker_construction_failed`), or the channel open failed (`channel_open_failed`).
* The `aggregate_bytes_exceeded` reason names a readiness handshake the host fell
* back because the process aggregate byte ceiling had no room for the readiness
* buffer.
* The `aggregate_bytes_exceeded` reason names a readiness handshake, or an
* `http2` post-preface pre-bind buffer, the host fell back because the
* process aggregate byte ceiling had no room. The `preface_missing` reason
* names a missing or an invalid HTTP/2 client connection preface inside the
* bounded readiness buffer: the host found no valid preface after the
* accepted READY line, aborted the HTTP/2 open, and moved the run to
* `queue_v1` one time.
*/
export type DuplexFallbackReason =
| "gate_off"
@ -113,7 +121,8 @@ export type DuplexFallbackReason =
| "ready_nonce_mismatch"
| "ready_timeout"
| "contaminated"
| "aggregate_bytes_exceeded";
| "aggregate_bytes_exceeded"
| "preface_missing";
/** The class of a terminal loss, relative to the first request dispatch. */
export type DuplexLossClass = "pre_dispatch" | "post_dispatch";
@ -155,6 +164,58 @@ export function normalizeDuplexLossReason(value: string | null | undefined): Dup
: "other";
}
/**
* The closed set of HTTP/2 session and stream event names the host maps to a
* loss reason. Each name spells one distinct signal the `http2_v1` transport
* can observe: a session-level protocol fault, a peer GOAWAY, a stalled PING
* watchdog, a rejected host-to-sandbox write, a reason-less transport close,
* or the pseudo-terminal channel process exit.
*/
export const HTTP2_TELEMETRY_EVENT_NAMES = [
"session_error",
"session_goaway",
"session_stall",
"write_error",
"transport_closed",
"channel_exit",
] as const;
/** One event name from the closed HTTP/2 event set. */
export type Http2TelemetryEventName = (typeof HTTP2_TELEMETRY_EVENT_NAMES)[number];
/** The host-owned closed HTTP/2 event-name set. It backs {@link mapHttp2EventToDuplexLossReason}. */
const HTTP2_EVENT_NAMES: ReadonlySet<string> = new Set<string>(HTTP2_TELEMETRY_EVENT_NAMES);
/**
* The map from one closed HTTP/2 event name to the existing, closed
* {@link DuplexLossReason} taxonomy. The map reuses that one taxonomy instead
* of a second, HTTP/2-only reason list, so the closed-set pattern stays one
* set of values across every transport.
*/
const HTTP2_EVENT_TO_LOSS_REASON: Readonly<Record<Http2TelemetryEventName, DuplexLossReason>> = {
session_error: "rpc_failure",
session_goaway: "transport_closed",
session_stall: "heartbeat_timeout",
write_error: "write_error",
transport_closed: "transport_closed",
channel_exit: "provider_exit",
};
/**
* Map one HTTP/2 session or stream event name to the closed
* {@link DuplexLossReason} taxonomy (accepted security fix 7). Return the
* mapped reason when the closed event-name set holds the value. Return
* `other` for any other value or a missing value, so an unknown event name or
* a raw provider string never reaches a sink.
*/
export function mapHttp2EventToDuplexLossReason(
event: string | null | undefined,
): DuplexLossReason {
return typeof event === "string" && HTTP2_EVENT_NAMES.has(event)
? HTTP2_EVENT_TO_LOSS_REASON[event as Http2TelemetryEventName]
: "other";
}
/** The one approved public provider value. */
export const DUPLEX_APPROVED_PROVIDER = "daytona";
/** The constant for any provider key outside the allowlist. */
@ -278,6 +339,13 @@ export interface DuplexObservabilityOptions {
recorder?: DuplexObservabilityRecorder | null;
/** The raw provider key. The facade maps it through the allowlist one time. */
providerKey?: string | null;
/**
* The transport value the facade stamps on every non-file record (a channel
* open, a request, a loss, a session leak). The default is `duplex`, so an
* existing caller that names no transport sees no change. The host passes
* `http2` for the `http2_v1` path.
*/
transport?: DuplexTransportValue;
}
/**
@ -289,6 +357,7 @@ export interface DuplexObservabilityOptions {
export function createDuplexObservability(options: DuplexObservabilityOptions = {}): DuplexObservability {
const recorder = options.recorder ?? NOOP_DUPLEX_OBSERVABILITY_RECORDER;
const provider = normalizeDuplexProvider(options.providerKey);
const transport: DuplexTransportValue = options.transport ?? "duplex";
const safeSpan = (record: DuplexObservabilitySpanRecord): void => {
try {
@ -332,7 +401,7 @@ export function createDuplexObservability(options: DuplexObservabilityOptions =
settled = true;
const dimensions: DuplexObservabilityDimensions = {
provider,
transport: "duplex",
transport,
outcome: "ok",
};
safeSpan({ name: DUPLEX_SPAN_CHANNEL_OPEN, dimensions });
@ -349,7 +418,7 @@ export function createDuplexObservability(options: DuplexObservabilityOptions =
// closed dimension key, so no new key reaches a sink.
safeSpan({
name: DUPLEX_SPAN_CHANNEL_OPEN,
dimensions: { provider, transport: "duplex", outcome: "error", fallback_reason: reason },
dimensions: { provider, transport, outcome: "error", fallback_reason: reason },
});
recordFallback(reason);
},
@ -359,14 +428,14 @@ export function createDuplexObservability(options: DuplexObservabilityOptions =
recordRequest(record: { latencyMs: number; outcome: DuplexOutcomeValue }): void {
safeSpan({
name: DUPLEX_SPAN_REQUEST,
dimensions: { provider, transport: "duplex", outcome: record.outcome },
dimensions: { provider, transport, outcome: record.outcome },
latencyMs: record.latencyMs,
});
},
recordLoss(lossClass: DuplexLossClass, lossReason: DuplexLossReason): void {
const dimensions: DuplexObservabilityDimensions = {
provider,
transport: "duplex",
transport,
outcome: "error",
loss_class: lossClass,
loss_reason: lossReason,
@ -377,7 +446,7 @@ export function createDuplexObservability(options: DuplexObservabilityOptions =
recordSessionLeak(): void {
safeCounter({
metric: DUPLEX_COUNTER_SESSION_LEAK_TOTAL,
dimensions: { provider, transport: "duplex", outcome: "error" },
dimensions: { provider, transport, outcome: "error" },
});
},
};

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,969 @@
/**
* Host HTTP/2 server for the sandbox callback bridge transport.
*
* The server wraps one {@link CommandManagedDuplexChannel} as a Node `Duplex`
* and runs one plaintext HTTP/2 session on it. It maps every stream on that
* session to one call of the caller-supplied `forwardRequest` handler, then
* writes the result back as the stream response. The handler applies the real
* host token and the run attribution, so those rules stay in one place, next
* to the existing file-bridge and duplex-bridge forward path.
*
* This file does not select the transport for a run. It builds and tests the
* host half of the pair in isolation; a later phase wires the pair into the
* transport-selection path.
*
* Requests flow from the sandbox to the host only: the host never opens a
* stream to the sandbox. The server enforces three checks, in this order, for
* every stream:
* 1. a constant-time compare of the bridge token against the per-run token
* (accepted security fix 4), before any other processing;
* 2. one canonical parse of the `:path` pseudo-header (accepted security fix
* 3), whose result feeds both the route allowlist and the forward URL;
* 3. the route allowlist and the header allowlist, reused unchanged from
* `sandbox-callback-bridge.ts`.
* The server also bounds ten `http2.createServer` options (accepted security
* fix 1), so Node enforces the session, header, and stream-reset limits on
* every connection with no new component.
*/
import { Duplex } from "node:stream";
import http2 from "node:http2";
import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
import {
authorizeSandboxCallbackBridgeRequestWithRoutes,
compareBridgeTokensConstantTime,
sanitizeSandboxCallbackBridgeHeaders,
DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST,
DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES,
DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST,
type SandboxCallbackBridgeRouteRule,
} from "./sandbox-callback-bridge.js";
// ---------------------------------------------------------------------------
// Bounded server options (accepted security fix 1). Node enforces each value,
// so naming them adds configuration and no new component. Every value and
// every name below matches the board-approved table exactly.
// ---------------------------------------------------------------------------
/** Server push. The transport never needs it. */
export const HTTP2_BRIDGE_ENABLE_PUSH = false;
/** Open streams. This matches the current broker limit ({@link DEFAULT_DUPLEX_BROKER_MAX_IN_FLIGHT_REQUESTS} in `duplex-bridge-broker.ts`). */
export const HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS = 64;
/** One decompressed header list. The Node default is 65535. */
export const HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE = 16384;
/** The header-compression table. This keeps the Node default. */
export const HTTP2_BRIDGE_HEADER_TABLE_SIZE = 4096;
/** Session memory in mebibytes. The Node default is 10. */
export const HTTP2_BRIDGE_MAX_SESSION_MEMORY = 16;
/** Header pairs per request. This names the Node default. */
export const HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS = 128;
/** The outbound compression table. */
export const HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE = 4096;
/** Invalid frames before Node closes the session. */
export const HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES = 100;
/** Rejected streams before Node closes the session. */
export const HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS = 100;
/** The stream-reset budget (frames per interval). Node sends GOAWAY past the budget. */
export const HTTP2_BRIDGE_STREAM_RESET_RATE = 10;
/** The stream-reset budget (burst allowance). Node sends GOAWAY past the budget. */
export const HTTP2_BRIDGE_STREAM_RESET_BURST = 100;
/**
* The full bounded options object. The server passes this object, unchanged,
* to `http2.createServer`. A test asserts every value on this object, so it
* proves the running server actually carries the bound, not only that the
* named constant exists.
*/
export const HTTP2_BRIDGE_SERVER_OPTIONS: http2.ServerOptions = {
settings: {
enablePush: HTTP2_BRIDGE_ENABLE_PUSH,
maxConcurrentStreams: HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS,
maxHeaderListSize: HTTP2_BRIDGE_MAX_HEADER_LIST_SIZE,
headerTableSize: HTTP2_BRIDGE_HEADER_TABLE_SIZE,
},
maxSessionMemory: HTTP2_BRIDGE_MAX_SESSION_MEMORY,
maxHeaderListPairs: HTTP2_BRIDGE_MAX_HEADER_LIST_PAIRS,
maxDeflateDynamicTableSize: HTTP2_BRIDGE_MAX_DEFLATE_DYNAMIC_TABLE_SIZE,
maxSessionInvalidFrames: HTTP2_BRIDGE_MAX_SESSION_INVALID_FRAMES,
maxSessionRejectedStreams: HTTP2_BRIDGE_MAX_SESSION_REJECTED_STREAMS,
streamResetRate: HTTP2_BRIDGE_STREAM_RESET_RATE,
streamResetBurst: HTTP2_BRIDGE_STREAM_RESET_BURST,
};
// ---------------------------------------------------------------------------
// Duplex channel adapter
// ---------------------------------------------------------------------------
/** The default cap, in bytes, on the read-side queue {@link wrapDuplexChannelAsNodeDuplex}
* holds once `Duplex.push()` reports the readable side is full (a `false`
* return). A sandbox-controlled channel has no upstream pause: `onData` below
* keeps delivering bytes whether or not the HTTP/2 session keeps up with
* them. Past this cap the wrapper treats the channel as stuck, not merely
* slow, and fails closed: it stops the channel and destroys the `Duplex`, so
* a producer that keeps outpacing its reader cannot grow host memory without
* bound. This cap also bounds one single chunk: the wrapper checks a chunk's
* own size against it before `push()` ever runs, so one oversized chunk
* cannot cross the cap on its first delivery, before the queue holds
* anything to compare it against. */
export const DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES = HTTP2_BRIDGE_MAX_SESSION_MEMORY * 1024 * 1024;
/** The default bound, in milliseconds, on how long the read-side queue
* {@link wrapDuplexChannelAsNodeDuplex} holds can stay non-empty with no
* chunk draining from it. The byte cap above bounds how much memory a stuck
* reader can hold; it does not bound how long the reader can stay stuck. A
* consumer that never resumes reading would otherwise hold the channel open,
* backpressured, for as long as the queue stays under the byte cap. Each
* drained chunk renews this bound, so a consumer that keeps making real
* progress never trips it; only a consumer that stops resuming entirely
* does. */
export const DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS = 30_000;
/**
* Wrap a {@link CommandManagedDuplexChannel} as a Node `Duplex`, so an
* `Http2Server` can run one session directly on it (`server.emit("connection",
* duplex)`). The wrapper never buffers more than one write in flight: it calls
* the stream write callback only after the channel's own write call settles
* (the backpressure constraint), never as a delivery signal. The provider
* accepts many megabytes in milliseconds and holds them in its own buffer, so
* this direction stays governed by the channel's own write-settle timing.
*
* The read direction needs its own bound. The channel exposes no pause: once
* `onData` below is registered, the channel keeps calling it for every byte
* the sandbox sends, with no way for this wrapper to slow it down. Node's
* `Duplex.push()` reports back-pressure through its boolean return, not by
* refusing the call, so a caller that ignores a `false` return and keeps
* pushing grows the readable side's internal buffer with no limit. This
* wrapper honors that signal instead: while `push()` reports room, it pushes
* directly; once `push()` reports the readable side is full, it queues each
* later chunk instead of pushing past that signal, and drains the queue from
* `read()`, which Node calls again only once the consumer wants more. Every
* chunk, on either path, first checks against
* {@link DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES} (or the caller's
* `maxBufferedReadBytes`) on its own size, and the queue checks against the
* same cap on its cumulative size: past either check the wrapper fails
* closed instead of buffering further, because the channel has no pause to
* fall back on. A second, independent bound
* {@link DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS} (or the caller's
* `readBackpressureStallMs`) covers the case the byte cap does not: a
* consumer that stops reading entirely, so the queue never grows past the
* byte cap but also never drains. This bound renews on every chunk the
* queue drains, so a consumer that keeps making real progress never trips
* it.
*/
export function wrapDuplexChannelAsNodeDuplex(
channel: CommandManagedDuplexChannel,
options: { maxBufferedReadBytes?: number; readBackpressureStallMs?: number } = {},
): Duplex {
const maxBufferedReadBytes = options.maxBufferedReadBytes ?? DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES;
const readBackpressureStallMs =
options.readBackpressureStallMs ?? DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS;
// Chunks `onData` already delivered that `push()` has not yet accepted,
// in arrival order. `read()` drains this queue before it lets Node pull
// any new bytes, so the delivery order the channel used stays intact.
const pendingReads: Buffer[] = [];
let pendingReadBytes = 0;
// True once `push()` last reported room for more, or before the first
// push call. `onData` pushes directly while this holds; once a `push()`
// call reports no room, later chunks queue in `pendingReads` instead.
let canPushMore = true;
// True once the channel exited. `endReadableIfDrained` pushes `null` only
// after the queue this wrapper still holds fully drains, so a chunk that
// arrived before the exit is never dropped.
let channelExited = false;
// Arms while the queue holds at least one chunk; clears once it fully
// drains. Fires `readBackpressureStallMs` after the queue's last drain (or
// its first chunk, if it never drained at all) with no further drain, so a
// consumer that stops resuming does not hold the channel open forever
// under the byte cap.
let backpressureStallTimer: ReturnType<typeof setTimeout> | undefined;
function clearBackpressureStallTimer(): void {
if (backpressureStallTimer === undefined) return;
clearTimeout(backpressureStallTimer);
backpressureStallTimer = undefined;
}
function armBackpressureStallTimer(): void {
clearBackpressureStallTimer();
backpressureStallTimer = setTimeout(() => {
failClosed(
"Sandbox HTTP/2 channel's read backpressure queue did not drain within the stall bound; the reader appears stuck.",
);
}, readBackpressureStallMs);
backpressureStallTimer.unref?.();
}
function failClosed(message: string): void {
clearBackpressureStallTimer();
channel.stop();
duplex.destroy(new Error(message));
}
function endReadableIfDrained(): void {
if (!channelExited || pendingReads.length > 0 || duplex.destroyed) return;
duplex.push(null);
}
const duplex: Duplex = new Duplex({
read() {
canPushMore = true;
let drainedAChunk = false;
while (canPushMore && pendingReads.length > 0) {
const next = pendingReads.shift();
if (next === undefined) break;
pendingReadBytes -= next.byteLength;
drainedAChunk = true;
canPushMore = duplex.push(next);
}
if (pendingReads.length === 0) {
clearBackpressureStallTimer();
} else if (drainedAChunk) {
// The queue still holds chunks, but at least one drained just now:
// real progress, so the stall bound renews instead of expiring under
// a consumer that is still reading, only slowly.
armBackpressureStallTimer();
}
endReadableIfDrained();
},
write(chunk: unknown, _encoding, callback) {
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as ArrayBufferLike);
let settleResult: unknown;
try {
settleResult = channel.write(bytes);
} catch (error) {
callback(error instanceof Error ? error : new Error(String(error)));
return;
}
if (
settleResult != null &&
typeof (settleResult as Promise<void>).then === "function"
) {
(settleResult as Promise<void>).then(
() => callback(),
(error) => callback(error instanceof Error ? error : new Error(String(error))),
);
} else {
callback();
}
},
final(callback) {
channel
.close()
.then(() => callback(), (error) => callback(error instanceof Error ? error : new Error(String(error))));
},
});
channel.onData((chunk) => {
if (duplex.destroyed) return;
const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk);
// Check one chunk's own size against the cap before either path below
// runs. `push()` never refuses a call on its size, so a single chunk
// larger than the whole cap would otherwise reach Node's internal
// buffer unbounded on the direct-push path, before the queue this
// wrapper owns ever holds anything to compare a later chunk against.
if (bytes.byteLength > maxBufferedReadBytes) {
failClosed(
"Sandbox HTTP/2 channel delivered one chunk larger than the bounded read backpressure buffer.",
);
return;
}
if (canPushMore && pendingReads.length === 0) {
canPushMore = duplex.push(bytes);
return;
}
// The readable side already reported it is full, and the channel has no
// pause to slow the sandbox side down: queue this chunk instead of
// pushing past that signal. Bound the queue, so a producer that keeps
// outpacing its reader cannot grow it without limit.
if (pendingReads.length === 0) {
// The queue was empty until this chunk: arm the stall bound, so a
// consumer that never resumes reading still ends the channel, even
// though this chunk alone stays under the byte cap.
armBackpressureStallTimer();
}
pendingReadBytes += bytes.byteLength;
if (pendingReadBytes > maxBufferedReadBytes) {
failClosed(
"Sandbox HTTP/2 channel exceeded the bounded read backpressure buffer; the reader could not keep up.",
);
return;
}
pendingReads.push(bytes);
});
channel.onExit(() => {
channelExited = true;
endReadableIfDrained();
});
// The channel exited, or `read()`/`onData` above failed the `Duplex`
// closed: either way, no further chunk will ever drain, so the stall
// timer serves no purpose and only holds a stray handle open.
duplex.once("close", clearBackpressureStallTimer);
return duplex;
}
// ---------------------------------------------------------------------------
// Canonical `:path` parsing (accepted security fix 3)
// ---------------------------------------------------------------------------
/** The reason {@link parseCanonicalBridgeRequestPath} rejected one request. */
export type CanonicalBridgeRequestPathRejection =
| "missing_path"
| "duplicate_pseudo_header"
| "non_origin_form"
| "encoded_slash"
| "backslash"
| "nul_byte"
| "dot_segment";
/** The one parsed pathname and query. Both the route allowlist and the forward
* URL builder read this same value; the host never parses `:path` twice. */
export interface CanonicalBridgeRequestPath {
pathname: string;
/** The query string, in `URL.search` form: empty, or a leading `?`. */
query: string;
}
export type CanonicalBridgeRequestPathResult =
| { ok: true; value: CanonicalBridgeRequestPath }
| { ok: false; reason: CanonicalBridgeRequestPathRejection };
/** The request pseudo-headers HTTP/2 allows exactly one of, per request. */
const REQUEST_PSEUDO_HEADER_NAMES = [":method", ":scheme", ":authority", ":path"] as const;
/**
* Parse the `:path` pseudo-header exactly one time. The caller passes the
* returned pathname and query to both the route allowlist and the forward URL
* builder never a second, independent parse of the raw header.
*
* The parser rejects a request that carries any of: a duplicate pseudo-header,
* a missing or empty `:path`, a non-origin-form path, an encoded slash, a
* backslash, a NUL byte, or a dot segment (checked before URL normalization
* would silently remove it, and after percent-decoding each segment, so an
* encoded dot segment cannot slip through).
*/
export function parseCanonicalBridgeRequestPath(
headers: http2.IncomingHttpHeaders,
): CanonicalBridgeRequestPathResult {
for (const name of REQUEST_PSEUDO_HEADER_NAMES) {
if (Array.isArray((headers as Record<string, unknown>)[name])) {
return { ok: false, reason: "duplicate_pseudo_header" };
}
}
const rawPath = headers[":path"];
if (typeof rawPath !== "string" || rawPath.length === 0) {
return { ok: false, reason: "missing_path" };
}
// Origin-form only: a single leading "/", never "//" (network-path form) and
// never an absolute-form URI ("scheme://...").
if (!rawPath.startsWith("/") || rawPath.startsWith("//") || rawPath.includes("://")) {
return { ok: false, reason: "non_origin_form" };
}
if (/%2f/i.test(rawPath)) {
return { ok: false, reason: "encoded_slash" };
}
if (rawPath.includes("\\")) {
return { ok: false, reason: "backslash" };
}
if (rawPath.includes("\0") || /%00/i.test(rawPath)) {
return { ok: false, reason: "nul_byte" };
}
const queryIndex = rawPath.indexOf("?");
const rawPathname = queryIndex === -1 ? rawPath : rawPath.slice(0, queryIndex);
for (const segment of rawPathname.split("/")) {
let decoded: string;
try {
decoded = decodeURIComponent(segment);
} catch {
return { ok: false, reason: "non_origin_form" };
}
if (decoded === "." || decoded === "..") {
return { ok: false, reason: "dot_segment" };
}
}
// Every raw-string check above passed, and the path carries no dot segment,
// so `URL` normalization here changes nothing but percent-encoding; it stays
// safe to build the canonical pathname and query from it.
let url: URL;
try {
url = new URL(rawPath, "http://bridge.internal");
} catch {
return { ok: false, reason: "non_origin_form" };
}
return { ok: true, value: { pathname: url.pathname, query: url.search } };
}
/**
* Build the forward URL from the one canonical parse. This mirrors
* `buildBridgeForwardUrl` in `execution-target.ts`, which the file bridge and
* the duplex bridge use today; a later phase wires the HTTP/2 host handler to
* that same forward path and can consolidate the two into one export.
*/
export function buildHttp2BridgeForwardUrl(
baseUrl: string,
request: CanonicalBridgeRequestPath,
): URL {
const url = new URL(request.pathname, baseUrl);
const query = request.query.trim();
url.search = query.startsWith("?") ? query.slice(1) : query;
return url;
}
// ---------------------------------------------------------------------------
// PING stall detection
// ---------------------------------------------------------------------------
/** The default interval between two liveness PING frames, in milliseconds. */
export const DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS = 5_000;
/** The default bound a sent PING waits for its ack before the session counts as stalled. */
export const DEFAULT_HTTP2_BRIDGE_PING_STALL_MS = 20_000;
/** The default idle bound on a request body read: the maximum gap between
* two received chunks (or between the token check and the first chunk)
* before the server treats the stream as stalled. Each received chunk resets
* this bound, so a slow peer that keeps making real progress completes; only
* a peer that stops sending trips it. */
export const DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS = 30_000;
/** The default lifetime bound on a request body read, measured from the
* start of the read or from the most recent chunk that carried real
* progress, whichever is later. This bound is independent of
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS}: the idle bound
* resets on every chunk to catch a peer that stops sending; this bound
* renews on every chunk too, but only up to the hard ceiling below, so it
* catches a peer that never stops sending but also never finishes. See
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS} for the
* bound that keeps this renewal from running forever. */
export const DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS = 120_000;
/** The default hard ceiling on a request body read's total lifetime: an
* absolute bound armed once, at the start of the read, and never pushed out
* past this point no matter how much progress a later chunk reports. A peer
* cannot use a steady trickle of small chunks to keep renewing
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS} and hold a
* {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} stream slot open forever,
* because this ceiling still ends the read once the read's total age passes
* it. Set well above {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS}
* so a legitimate upload that makes real but slow progress a chunk every
* few seconds, well inside the renewable bound above still has room to
* finish, while a peer that never finishes is still bounded to this ceiling
* instead of running forever. This is an intentional design limit, not a
* defect and not open for removal without a replacement bound. */
export const DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS =
DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS * 4;
/** The default bound {@link Http2BridgeServerHandle.close} waits for an
* active session to close on its own before it force-destroys the session. A
* session that carries a stalled stream would otherwise hold `close()` open
* forever, because `session.close()` waits for every open stream to end. */
export const DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS = 5_000;
function startHttp2BridgePingWatchdog(
session: http2.ServerHttp2Session,
input: { intervalMs: number; stallMs: number; onStall: (error: Error) => void },
): () => void {
let stopped = false;
let pingTimer: ReturnType<typeof setTimeout> | undefined;
let stallTimer: ReturnType<typeof setTimeout> | undefined;
function sendOnePing(): void {
if (stopped) return;
stallTimer = setTimeout(() => {
if (stopped) return;
stopped = true;
input.onStall(new Error("HTTP/2 bridge session stalled: no PING ack within the stall bound."));
}, input.stallMs);
stallTimer.unref?.();
try {
session.ping((error) => {
if (stopped) return;
if (stallTimer) clearTimeout(stallTimer);
if (error) {
stopped = true;
input.onStall(error instanceof Error ? error : new Error(String(error)));
return;
}
pingTimer = setTimeout(sendOnePing, input.intervalMs);
pingTimer.unref?.();
});
} catch (error) {
if (stallTimer) clearTimeout(stallTimer);
stopped = true;
input.onStall(error instanceof Error ? error : new Error(String(error)));
}
}
pingTimer = setTimeout(sendOnePing, input.intervalMs);
pingTimer.unref?.();
return () => {
stopped = true;
if (pingTimer) clearTimeout(pingTimer);
if (stallTimer) clearTimeout(stallTimer);
};
}
// ---------------------------------------------------------------------------
// The server
// ---------------------------------------------------------------------------
/** The result of one forward call. The server turns it into one stream response. */
export interface Http2BridgeForwardResult {
status: number;
headers?: Record<string, string>;
body?: Buffer | string;
}
/**
* The one canonically-parsed, route-authorized, header-sanitized request the
* server hands to the forward handler.
*/
export interface Http2BridgeForwardRequest {
method: string;
pathname: string;
query: string;
headers: Record<string, string>;
body: Buffer;
}
export type Http2BridgeForwardHandler = (
request: Http2BridgeForwardRequest,
) => Promise<Http2BridgeForwardResult>;
/** The GOAWAY the server observed, naming the last stream ID the peer processed. */
export interface Http2BridgeGoawayRecord {
lastStreamId: number;
errorCode: number;
}
/** Classify one stream ID against an observed GOAWAY's last processed stream ID. */
export function classifyStreamAgainstGoaway(
streamId: number,
lastStreamId: number,
): "accepted" | "not_accepted" {
return streamId <= lastStreamId ? "accepted" : "not_accepted";
}
export interface CreateHttp2BridgeServerOptions {
/** The per-run bridge token. The server compares it, constant-time, against
* the token on every stream before route or header processing. */
bridgeToken: string;
/** The forward handler the server calls for each authorized request. */
forwardRequest: Http2BridgeForwardHandler;
/** The route allowlist. The default is {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST}. */
routes?: readonly SandboxCallbackBridgeRouteRule[];
/** The header allowlist. The default is {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST}. */
headerAllowlist?: readonly string[];
/** The maximum request body size, in bytes. The default is {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES}. */
maxBodyBytes?: number;
/** The interval between two liveness PING frames, in milliseconds. */
pingIntervalMs?: number;
/** The bound a sent PING waits for its ack before the server closes the session. */
pingStallMs?: number;
/** The idle bound on a request body read: the maximum gap between two
* received chunks. The default is
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS}. */
requestBodyTimeoutMs?: number;
/** The lifetime bound a request body read gets, renewed on every chunk
* that carries real progress, up to {@link requestBodyLifetimeCeilingMs}.
* See {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS} for why the
* server enforces it. The default is
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS}. */
requestBodyMaxLifetimeMs?: number;
/** The hard ceiling on a request body read's total lifetime, armed once
* and never pushed out by progress. This bound caps worst-case
* stream-slot occupancy even for a peer that keeps renewing
* {@link requestBodyMaxLifetimeMs} with a steady trickle of chunks. See
* {@link DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS} for the
* default and the reasoning behind it. */
requestBodyLifetimeCeilingMs?: number;
/** The bound {@link Http2BridgeServerHandle.close} waits for an active
* session to close on its own before it force-destroys the session. The
* default is {@link DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS}. */
closeGraceMs?: number;
/** The cap, in bytes, on data this server holds once a bound `Duplex`
* reports its readable side is full (`push()` returns `false`). Past this
* cap the server treats the channel as stuck, not merely slow: see
* {@link wrapDuplexChannelAsNodeDuplex}. The default is
* {@link DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES}. */
maxBufferedReadBytes?: number;
/** The bound, in milliseconds, on how long the read backpressure queue
* {@link wrapDuplexChannelAsNodeDuplex} holds can stay non-empty with no
* chunk draining from it. The default is
* {@link DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS}. */
readBackpressureStallMs?: number;
/** The sink for a GOAWAY the server observed on a session (one the sandbox
* side sent to the host). */
onGoaway?: (record: Http2BridgeGoawayRecord) => void;
/** The sink for a session-level fault (a stall, a protocol fault). */
onSessionError?: (error: Error) => void;
/**
* Fires once for each new session. A caller uses the live
* `ServerHttp2Session` to send its own GOAWAY (accepted security fix's
* GOAWAY-classification behavior is meaningful only from the side that
* names the last stream it processed the host, since every stream
* originates from the sandbox). A test uses this hook to drive the
* GOAWAY test deterministically.
*/
onSession?: (session: http2.ServerHttp2Session) => void;
}
/** The handle {@link createHttp2BridgeServer} returns. */
export interface Http2BridgeServerHandle {
/** The underlying `Http2Server`. It is never `listen()`-ed; every session
* binds through {@link Http2BridgeServerHandle.bindChannel}. */
readonly server: http2.Http2Server;
/** Wrap the channel as a `Duplex` and run one HTTP/2 session on it. Returns
* the wrapped `Duplex`, so a caller can also drive it directly (a test uses
* this to bind one side of a paired in-memory `Duplex`). */
bindChannel(channel: CommandManagedDuplexChannel): Duplex;
/** Close every active session. Safe to call more than one time. */
close(): Promise<void>;
}
function normalizeStreamMethod(value: string | string[] | undefined): string {
return typeof value === "string" && value.trim().length > 0 ? value.trim().toUpperCase() : "GET";
}
function readBridgeTokenHeader(headers: http2.IncomingHttpHeaders): string | undefined {
const raw = headers.authorization;
if (typeof raw !== "string" || !raw.startsWith("Bearer ")) return undefined;
return raw.slice("Bearer ".length);
}
function toOutboundHeaderRecord(headers: http2.IncomingHttpHeaders): Record<string, string> {
const out: Record<string, string> = {};
for (const [key, value] of Object.entries(headers)) {
if (key.startsWith(":") || value == null) continue;
out[key.toLowerCase()] = Array.isArray(value) ? value.join(", ") : String(value);
}
return out;
}
/**
* Read one request body, bounded on size and on two independent time
* bounds.
*
* The idle bound resets on each received chunk, so a slow peer that keeps
* making real progress completes, while a peer that stops sending
* mid-stream the same failure a stalled network path or a hung sandbox
* process produces still trips it.
*
* The lifetime bound also renews on each received chunk, by the same
* amount, so a slow peer that keeps sending real progress gets more time
* instead of losing the stream mid-upload. That renewal never pushes the
* deadline past `startedAt + lifetimeCeilingMs`: the hard ceiling arms once,
* when the read starts, and holds no matter how much progress a later chunk
* reports. This is an intentional design limit, not a defect, and it is not
* open for removal or relaxation without a replacement bound. It exists to
* cap the worst-case time one read can occupy a
* {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} stream slot a cap the idle
* bound cannot provide alone, because a peer that keeps every chunk gap
* under the idle bound never trips it, no matter how long the read runs. A
* peer cannot use a steady trickle of small chunks to hold the slot forever,
* because the ceiling still ends the read once the read's total age passes
* it, regardless of how the peer paces its chunks.
*
* The `close` listener is the settle-of-last-resort: it fires whenever the
* stream ends for any reason at all a normal end, an error, a timeout- or
* shutdown-triggered `destroy()`, or a peer reset so the promise always
* settles and the caller never awaits a stream that already went away.
*/
function readHttp2StreamBody(
stream: http2.ServerHttp2Stream,
maxBodyBytes: number,
idleTimeoutMs: number,
maxLifetimeMs: number,
lifetimeCeilingMs: number,
): Promise<Buffer> {
return new Promise((resolve, reject) => {
const chunks: Buffer[] = [];
let totalBytes = 0;
let settled = false;
let idleTimer: ReturnType<typeof setTimeout>;
let lifetimeTimer: ReturnType<typeof setTimeout>;
const hardDeadline = Date.now() + lifetimeCeilingMs;
const settle = (run: () => void) => {
if (settled) return;
settled = true;
clearTimeout(idleTimer);
clearTimeout(lifetimeTimer);
run();
};
const armIdleTimer = () => {
clearTimeout(idleTimer);
idleTimer = setTimeout(() => {
settle(() => reject(new Error("Bridge request body stalled before it completed.")));
stream.destroy();
}, idleTimeoutMs);
idleTimer.unref?.();
};
// Renews on every chunk that carries real progress, but the delay never
// exceeds the time left before the hard ceiling armed below: see the
// function's own doc comment for why the ceiling must stay independent
// of progress.
const armLifetimeTimer = () => {
clearTimeout(lifetimeTimer);
const delayMs = Math.max(0, Math.min(maxLifetimeMs, hardDeadline - Date.now()));
lifetimeTimer = setTimeout(() => {
settle(() => reject(new Error("Bridge request body exceeded the maximum lifetime bound.")));
stream.destroy();
}, delayMs);
lifetimeTimer.unref?.();
};
armLifetimeTimer();
armIdleTimer();
stream.on("data", (chunk: Buffer) => {
totalBytes += chunk.byteLength;
if (totalBytes > maxBodyBytes) {
settle(() => reject(new Error("Bridge request body exceeded the configured size limit.")));
stream.destroy();
return;
}
chunks.push(chunk);
// The chunk is real progress, so the peer is not stalled: reset the
// idle bound and renew the lifetime bound, instead of letting either
// expire under a slow but active upload. The lifetime renewal above is
// still capped by the hard ceiling.
armIdleTimer();
armLifetimeTimer();
});
stream.once("end", () => settle(() => resolve(Buffer.concat(chunks))));
stream.once("error", (error) => settle(() => reject(error instanceof Error ? error : new Error(String(error)))));
stream.once("aborted", () => settle(() => reject(new Error("Bridge request stream aborted."))));
stream.once("close", () => settle(() => reject(new Error("Bridge request stream closed before it completed."))));
});
}
function respondJson(stream: http2.ServerHttp2Stream, status: number, body: unknown): void {
if (stream.destroyed || stream.closed) return;
try {
stream.respond({ ":status": status, "content-type": "application/json" });
stream.end(JSON.stringify(body));
} catch {
// The peer reset the stream (RST_STREAM) before the server could answer.
// That fault stays local to this one stream; every other stream and the
// session itself stay unaffected.
}
}
/**
* Answer a denied stream, then consume and discard its request body under
* the same idle and total lifetime bounds an authenticated request gets.
* A denied request's body content never reaches the forward handler, but
* the inbound half of the stream still needs a bound: without one, a peer
* that leaves the body unfinished keeps the stream open, holding one of the
* {@link HTTP2_BRIDGE_MAX_CONCURRENT_STREAMS} slots for as long as it
* chooses. Nothing awaits the discard; the caller has already answered the
* request and moves on to the next stream.
*/
function denyRequest(
stream: http2.ServerHttp2Stream,
status: number,
body: unknown,
maxBodyBytes: number,
idleTimeoutMs: number,
maxLifetimeMs: number,
lifetimeCeilingMs: number,
): void {
respondJson(stream, status, body);
if (stream.destroyed || stream.closed) return;
readHttp2StreamBody(stream, maxBodyBytes, idleTimeoutMs, maxLifetimeMs, lifetimeCeilingMs).catch(() => {
// The idle or lifetime bound above already destroyed the stream, or the
// peer reset it first. Either way the slot is free; the discarded body
// content is irrelevant to a denial.
});
}
/**
* Create the host HTTP/2 bridge server. The server runs no listener of its
* own: a caller wraps one duplex channel through {@link Http2BridgeServerHandle.bindChannel}
* per sandbox session.
*/
export function createHttp2BridgeServer(options: CreateHttp2BridgeServerOptions): Http2BridgeServerHandle {
const routes = options.routes ?? DEFAULT_SANDBOX_CALLBACK_BRIDGE_ROUTE_ALLOWLIST;
const headerAllowlist = options.headerAllowlist ?? DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST;
const maxBodyBytes = options.maxBodyBytes ?? DEFAULT_SANDBOX_CALLBACK_BRIDGE_MAX_BODY_BYTES;
const pingIntervalMs = options.pingIntervalMs ?? DEFAULT_HTTP2_BRIDGE_PING_INTERVAL_MS;
const pingStallMs = options.pingStallMs ?? DEFAULT_HTTP2_BRIDGE_PING_STALL_MS;
const requestBodyTimeoutMs = options.requestBodyTimeoutMs ?? DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_TIMEOUT_MS;
const requestBodyMaxLifetimeMs =
options.requestBodyMaxLifetimeMs ?? DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_MAX_LIFETIME_MS;
const requestBodyLifetimeCeilingMs =
options.requestBodyLifetimeCeilingMs ?? DEFAULT_HTTP2_BRIDGE_REQUEST_BODY_LIFETIME_CEILING_MS;
const closeGraceMs = options.closeGraceMs ?? DEFAULT_HTTP2_BRIDGE_CLOSE_GRACE_MS;
const maxBufferedReadBytes = options.maxBufferedReadBytes ?? DEFAULT_HTTP2_BRIDGE_MAX_BUFFERED_READ_BYTES;
const readBackpressureStallMs =
options.readBackpressureStallMs ?? DEFAULT_HTTP2_BRIDGE_READ_BACKPRESSURE_STALL_MS;
const server = http2.createServer(HTTP2_BRIDGE_SERVER_OPTIONS);
const activeSessions = new Set<http2.ServerHttp2Session>();
async function handleStream(
stream: http2.ServerHttp2Stream,
headers: http2.IncomingHttpHeaders,
): Promise<void> {
// Accepted security fix 4: the constant-time bridge-token compare runs
// before route processing and before header processing. This host check
// is independent of the gateway's own token check on the sandbox side.
if (!compareBridgeTokensConstantTime(options.bridgeToken, readBridgeTokenHeader(headers))) {
denyRequest(
stream,
401,
{ error: "Invalid bridge token." },
maxBodyBytes,
requestBodyTimeoutMs,
requestBodyMaxLifetimeMs,
requestBodyLifetimeCeilingMs,
);
return;
}
// Accepted security fix 3: parse `:path` exactly one time; both the route
// allowlist and the forward request below read this one result.
const parsedPath = parseCanonicalBridgeRequestPath(headers);
if (!parsedPath.ok) {
denyRequest(
stream,
400,
{ error: `Invalid request path: ${parsedPath.reason}` },
maxBodyBytes,
requestBodyTimeoutMs,
requestBodyMaxLifetimeMs,
requestBodyLifetimeCeilingMs,
);
return;
}
const method = normalizeStreamMethod(headers[":method"]);
const denialReason = authorizeSandboxCallbackBridgeRequestWithRoutes(
{ method, path: parsedPath.value.pathname },
routes,
);
if (denialReason) {
denyRequest(
stream,
403,
{ error: denialReason },
maxBodyBytes,
requestBodyTimeoutMs,
requestBodyMaxLifetimeMs,
requestBodyLifetimeCeilingMs,
);
return;
}
const sanitizedHeaders = sanitizeSandboxCallbackBridgeHeaders(
toOutboundHeaderRecord(headers),
headerAllowlist,
);
let body: Buffer;
try {
body = await readHttp2StreamBody(
stream,
maxBodyBytes,
requestBodyTimeoutMs,
requestBodyMaxLifetimeMs,
requestBodyLifetimeCeilingMs,
);
} catch (error) {
respondJson(stream, 413, { error: error instanceof Error ? error.message : String(error) });
return;
}
let result: Http2BridgeForwardResult;
try {
result = await options.forwardRequest({
method,
pathname: parsedPath.value.pathname,
query: parsedPath.value.query,
headers: sanitizedHeaders,
body,
});
} catch (error) {
respondJson(stream, 502, { error: error instanceof Error ? error.message : String(error) });
return;
}
if (stream.destroyed || stream.closed) return;
const responseHeaders: http2.OutgoingHttpHeaders = { ":status": result.status };
for (const [key, value] of Object.entries(result.headers ?? {})) {
if (key.toLowerCase() === "content-length") continue;
responseHeaders[key] = value;
}
try {
stream.respond(responseHeaders);
stream.end(result.body);
} catch {
// The peer reset the stream (RST_STREAM) between dispatch and response.
// One stream's write fault stays local to that stream.
}
}
server.on("session", (session) => {
activeSessions.add(session);
options.onSession?.(session);
const stopWatchdog = startHttp2BridgePingWatchdog(session, {
intervalMs: pingIntervalMs,
stallMs: pingStallMs,
// Destroying the session with an error routes back through this same
// session's own `error` listener below, which reports it exactly once.
onStall: (error) => {
if (!session.destroyed) session.destroy(error);
},
});
session.on("goaway", (errorCode: number, lastStreamId: number) => {
options.onGoaway?.({ lastStreamId, errorCode });
});
session.on("close", () => {
stopWatchdog();
activeSessions.delete(session);
});
session.on("error", (error) => {
stopWatchdog();
options.onSessionError?.(error instanceof Error ? error : new Error(String(error)));
});
});
server.on("stream", (stream, headers) => {
void handleStream(stream, headers).catch((error) => {
// A fault inside `handleStream` itself (not a forward-handler or
// stream-body rejection, both already caught above) is a defensive
// last resort. Destroy only this stream; the session stays open.
if (!stream.destroyed) {
stream.destroy(error instanceof Error ? error : new Error(String(error)));
}
});
});
return {
server,
bindChannel(channel: CommandManagedDuplexChannel): Duplex {
const duplex = wrapDuplexChannelAsNodeDuplex(channel, { maxBufferedReadBytes, readBackpressureStallMs });
server.emit("connection", duplex);
return duplex;
},
async close(): Promise<void> {
// `session.close()` sends GOAWAY and waits for every open stream to end
// on its own; a stalled stream (its body never completes, and its own
// timeout has not yet fired) would hold this wait open forever. The
// grace timer bounds it: past `closeGraceMs`, the server force-destroys
// the session, which ends its streams at once and settles their body
// reads through the `close` backstop in `readHttp2StreamBody`.
await Promise.all(
[...activeSessions].map(
(session) =>
new Promise<void>((resolve) => {
if (session.closed || session.destroyed) {
resolve();
return;
}
const forceDestroyTimer = setTimeout(() => {
if (!session.destroyed) session.destroy();
}, closeGraceMs);
forceDestroyTimer.unref?.();
session.close(() => {
clearTimeout(forceDestroyTimer);
resolve();
});
}),
),
);
},
};
}

View File

@ -2940,4 +2940,79 @@ describe("sandbox callback bridge", () => {
expect(stderr).toContain("[paperclip-bridge] server error");
expect(stderr).toContain("EADDRINUSE");
}, 15_000);
it("test_http2_gateway_writes_no_frame_between_ready_and_the_preface", async () => {
// Spawn the real generated gateway in http2_v1 mode and read its raw
// stdout bytes. The only frame-codec write on this path is the READY
// line; the very next bytes must be the HTTP/2 client connection preface
// with nothing in between, because the gateway hands stdout to the
// HTTP/2 client immediately after it writes READY and starts no
// heartbeat timer and writes no envelope frame on this path.
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-bridge-http2-gateway-"));
cleanupDirs.push(rootDir);
const entrypoint = path.join(rootDir, "paperclip-bridge-server.mjs");
await writeFile(entrypoint, getSandboxCallbackBridgeServerSource(), "utf8");
const probe = createServer();
const assignedPort = await new Promise<number>((resolve, reject) => {
probe.once("error", reject);
probe.listen(0, "127.0.0.1", () => {
const address = probe.address();
if (!address || typeof address === "string") {
reject(new Error("Could not reserve a loopback port for the test."));
return;
}
probe.close(() => resolve(address.port));
});
});
const nonce = "test-nonce-http2";
const child = spawn(process.execPath, [entrypoint], {
env: {
...process.env,
PAPERCLIP_API_BRIDGE_MODE: "http2_v1",
PAPERCLIP_BRIDGE_TOKEN: "test-token",
PAPERCLIP_BRIDGE_PORT: String(assignedPort),
PAPERCLIP_BRIDGE_NONCE: nonce,
},
stdio: ["pipe", "pipe", "pipe"],
});
cleanupFns.push(async () => {
child.kill();
});
let stderr = "";
child.stderr.on("data", (chunk: Buffer) => {
stderr += chunk.toString("utf8");
});
const chunks: Buffer[] = [];
const preface = Buffer.from("505249202a20485454502f322e300d0a0d0a534d0d0a0d0a", "hex");
const firstBytes = await new Promise<Buffer>((resolve, reject) => {
const timer = setTimeout(
() => reject(new Error("Timed out waiting for the http2 gateway stdout. stderr: " + stderr)),
5000,
);
child.stdout.on("data", (chunk: Buffer) => {
chunks.push(chunk);
const total = Buffer.concat(chunks);
const newlineIndex = total.indexOf(0x0a);
if (newlineIndex !== -1 && total.length >= newlineIndex + 1 + preface.length) {
clearTimeout(timer);
resolve(total);
}
});
child.once("error", reject);
child.once("exit", (code) => {
clearTimeout(timer);
reject(new Error("The http2 gateway exited early with code " + String(code) + ". stderr: " + stderr));
});
});
const newlineIndex = firstBytes.indexOf(0x0a);
expect(newlineIndex).toBeGreaterThan(0);
const readyLine = firstBytes.subarray(0, newlineIndex).toString("utf8");
expect(JSON.parse(readyLine)).toMatchObject({ type: "ready", nonce });
const afterReady = firstBytes.subarray(newlineIndex + 1, newlineIndex + 1 + preface.length);
expect(afterReady).toEqual(preface);
}, 15_000);
});

View File

@ -1,7 +1,9 @@
import { createHash, randomBytes, randomUUID } from "node:crypto";
import { createHash, randomBytes, randomUUID, timingSafeEqual } from "node:crypto";
import { promises as fs } from "node:fs";
import http2 from "node:http2";
import os from "node:os";
import path from "node:path";
import type { Duplex } from "node:stream";
import {
runWithoutActiveStep,
@ -70,12 +72,20 @@ export const SANDBOX_CALLBACK_BRIDGE_ENTRYPOINT = "paperclip-bridge-server.mjs";
const SANDBOX_EXEC_CHANNEL_ENV = "PAPERCLIP_SANDBOX_EXEC_CHANNEL";
const SANDBOX_EXEC_CHANNEL_BRIDGE = "bridge";
// The two bridge modes the generated gateway supports. The file mode polls a
// request/response queue on disk. The duplex mode forwards one request frame to
// stdout and resolves one response frame from stdin. The generated `.mjs`
// selects the mode from `PAPERCLIP_API_BRIDGE_MODE`.
// The bridge modes the generated gateway supports. The file mode polls a
// request/response queue on disk. The retired duplex mode forwarded one
// request frame to stdout and resolved one response frame from stdin; the
// generated gateway still defines it, but no mode dispatch selects it anymore
// — the http2 mode replaced it as the active non-file transport. The http2
// mode runs one Node HTTP/2 client session directly on stdin/stdout, after it
// sends the one READY line the host readiness gate expects. The generated
// `.mjs` selects the mode from `PAPERCLIP_API_BRIDGE_MODE`.
// HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
const SANDBOX_CALLBACK_BRIDGE_FILE_MODE = "queue_v1";
export const SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE = "duplex_v1";
/** The active non-file transport mode. It replaced {@link SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}
* in the mode-selection path. */
export const SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE = "http2_v1";
// The duplex gateway HTTP wait budget default. The gateway waits this long for a
// response frame before it answers the local caller with a 502 timeout. The
@ -1832,6 +1842,209 @@ export async function startSandboxCallbackBridgeServer(input: {
};
}
// ---------------------------------------------------------------------------
// Sandbox HTTP/2 client gateway
//
// This gateway turns each local loopback request into one HTTP/2 stream to
// the host server in `http2-bridge-server.ts`. It sits beside the file-mode
// and duplex-mode gateways above; it changes neither of them, and no mode
// dispatch selects it yet — a later phase wires it into the generated
// in-sandbox entrypoint and the transport-selection path.
// ---------------------------------------------------------------------------
/**
* Constant-time bridge-token compare. Both this sandbox gateway and the host
* server in `http2-bridge-server.ts` import this one helper, so the gateway
* check and the independent host check (accepted security fix 4) apply the
* exact same comparison rule. A length mismatch returns `false` without a
* `timingSafeEqual` call, because `timingSafeEqual` throws on unequal buffer
* lengths; both operands are bridge tokens of near-fixed length, so this one
* length branch leaks no useful timing signal.
*/
export function compareBridgeTokensConstantTime(
expected: string,
received: string | null | undefined,
): boolean {
const expectedBytes = Buffer.from(expected, "utf8");
const receivedBytes = Buffer.from(typeof received === "string" ? received : "", "utf8");
if (expectedBytes.length !== receivedBytes.length) return false;
return timingSafeEqual(expectedBytes, receivedBytes);
}
const SANDBOX_HTTP2_GATEWAY_DEFAULT_AUTHORITY = "bridge.internal";
/** One local request the gateway forwards as one HTTP/2 stream. */
export interface SandboxHttp2BridgeGatewayRequest {
method: string;
path: string;
query: string;
headers: Record<string, string>;
body: Buffer;
/**
* The token the local caller presented. The gateway check (accepted
* security fix 4 keeps this alongside the independent host check) compares
* it against the per-run bridge token before it opens a stream.
*/
receivedToken: string | null | undefined;
}
/** The response one forwarded HTTP/2 stream carried back. */
export interface SandboxHttp2BridgeGatewayResponse {
status: number;
headers: Record<string, string>;
body: Buffer;
}
export interface SandboxHttp2BridgeGateway {
/** Forward one local request as one HTTP/2 stream over the client session. */
forwardRequest(
request: SandboxHttp2BridgeGatewayRequest,
): Promise<SandboxHttp2BridgeGatewayResponse>;
/** Close the HTTP/2 client session. Safe to call more than one time. */
close(): Promise<void>;
}
export interface CreateSandboxHttp2BridgeGatewayOptions {
/**
* The per-run bridge token. The gateway checks every request against it,
* then attaches it to the outbound HTTP/2 stream as the `authorization`
* header, so the host can run its own independent check.
*/
bridgeToken: string;
/**
* Open the transport the HTTP/2 client session runs on. Returns a `Duplex`
* already connected to the host the sandbox process's own channel in
* production, or one side of a paired in-memory `Duplex` in a test.
*/
createConnection: () => Duplex;
/** The `:authority` pseudo-header value. The channel carries no real network
* address, so this is a fixed label. The default is `bridge.internal`. */
authority?: string;
/** The header allowlist applied to every outbound request. The default is
* {@link DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST}. */
headerAllowlist?: readonly string[];
/**
* The sink for a GOAWAY the host sends. The host names the last client
* stream ID it processed; a caller classifies each of its own dispatched
* stream IDs against it with `classifyStreamAgainstGoaway` in
* `http2-bridge-server.ts` to know which requests need a retry elsewhere.
*/
onGoaway?: (record: { lastStreamId: number; errorCode: number }) => void;
}
function forwardOneHttp2Request(
session: http2.ClientHttp2Session,
request: {
bridgeToken: string;
method: string;
path: string;
query: string;
headers: Record<string, string>;
body: Buffer;
},
): Promise<SandboxHttp2BridgeGatewayResponse> {
return new Promise((resolve, reject) => {
const query = request.query.trim();
const pathWithQuery =
query.length === 0 ? request.path : `${request.path}${query.startsWith("?") ? query : `?${query}`}`;
const requestHeaders: http2.OutgoingHttpHeaders = {
":method": request.method,
":path": pathWithQuery,
authorization: `Bearer ${request.bridgeToken}`,
...request.headers,
};
let stream: http2.ClientHttp2Stream;
try {
stream = session.request(requestHeaders, { endStream: request.body.length === 0 });
} catch (error) {
reject(error instanceof Error ? error : new Error(String(error)));
return;
}
const chunks: Buffer[] = [];
let responseHeaders: Record<string, string> = {};
let status = 502;
let settled = false;
const settle = (run: () => void) => {
if (settled) return;
settled = true;
run();
};
stream.on("response", (headers) => {
const rawStatus = headers[":status"];
status = typeof rawStatus === "number" ? rawStatus : Number(rawStatus) || 502;
responseHeaders = {};
for (const [key, value] of Object.entries(headers)) {
if (key.startsWith(":") || value == null) continue;
responseHeaders[key] = Array.isArray(value) ? value.join(", ") : String(value);
}
});
stream.on("data", (chunk: Buffer) => chunks.push(chunk));
stream.once("end", () => settle(() => resolve({ status, headers: responseHeaders, body: Buffer.concat(chunks) })));
stream.once("error", (error) =>
settle(() => reject(error instanceof Error ? error : new Error(String(error)))),
);
stream.once("aborted", () => settle(() => reject(new Error("Bridge HTTP/2 stream aborted."))));
if (request.body.length > 0) {
stream.end(request.body);
} else if (!stream.writableEnded) {
stream.end();
}
});
}
/**
* Create the sandbox HTTP/2 client gateway. It opens one HTTP/2 client
* session on the transport `createConnection` returns, and forwards each
* local request the caller hands it (already checked against the bridge
* token see {@link SandboxHttp2BridgeGatewayRequest.receivedToken}) as one
* HTTP/2 stream. It keeps the header allowlist on the sandbox side, exactly
* as the file-mode and duplex-mode gateways do.
*/
export function createSandboxHttp2BridgeGateway(
options: CreateSandboxHttp2BridgeGatewayOptions,
): SandboxHttp2BridgeGateway {
const authority = options.authority?.trim() || SANDBOX_HTTP2_GATEWAY_DEFAULT_AUTHORITY;
const headerAllowlist = options.headerAllowlist ?? DEFAULT_SANDBOX_CALLBACK_BRIDGE_HEADER_ALLOWLIST;
const session = http2.connect(`http://${authority}`, {
createConnection: options.createConnection,
});
// A session-level fault fails every in-flight `forwardRequest` call through
// that stream's own `error`/`aborted` handler. This listener only stops
// Node from raising an unhandled `error` event for the session itself.
session.on("error", () => undefined);
session.on("goaway", (errorCode: number, lastStreamId: number) => {
options.onGoaway?.({ lastStreamId, errorCode });
});
return {
forwardRequest(request: SandboxHttp2BridgeGatewayRequest): Promise<SandboxHttp2BridgeGatewayResponse> {
// The gateway check (accepted security fix 4 keeps this as well as the
// independent host-side check): a request whose token does not match
// the per-run bridge token never opens a stream.
if (!compareBridgeTokensConstantTime(options.bridgeToken, request.receivedToken)) {
return Promise.reject(new Error("Invalid bridge token."));
}
return forwardOneHttp2Request(session, {
bridgeToken: options.bridgeToken,
method: request.method,
path: request.path,
query: request.query,
headers: sanitizeSandboxCallbackBridgeHeaders(request.headers, headerAllowlist),
body: request.body,
});
},
close(): Promise<void> {
return new Promise((resolve) => {
if (session.closed || session.destroyed) {
resolve();
return;
}
session.close(() => resolve());
});
},
};
}
/**
* The zero-dependency codec the generated duplex gateway embeds. It is a plain
* JavaScript copy of the host codec in `duplex-frame-codec.ts`. It uses only the
@ -2084,6 +2297,8 @@ export function getSandboxCallbackBridgeServerSource(): string {
import { createServer } from "node:http";
import { promises as fs } from "node:fs";
import path from "node:path";
import http2 from "node:http2";
import { Duplex } from "node:stream";
const bridgeMode = process.env.PAPERCLIP_API_BRIDGE_MODE || "${SANDBOX_CALLBACK_BRIDGE_FILE_MODE}";
const queueDir = process.env.PAPERCLIP_BRIDGE_QUEUE_DIR;
@ -2129,7 +2344,11 @@ const allowedHeaders = new Set(${JSON.stringify([...DEFAULT_SANDBOX_CALLBACK_BRI
if (!bridgeToken) {
throw new Error("PAPERCLIP_BRIDGE_TOKEN is required.");
}
if (bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" && !queueDir) {
if (
bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}" &&
bridgeMode !== "${SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE}" &&
!queueDir
) {
throw new Error("PAPERCLIP_BRIDGE_QUEUE_DIR and PAPERCLIP_BRIDGE_TOKEN are required.");
}
@ -2723,7 +2942,231 @@ function runDuplexGateway() {
});
}
if (bridgeMode === "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}") {
// ---------------------------------------------------------------------------
// http2_v1: run one Node HTTP/2 client session directly on stdin/stdout.
//
// This gateway writes exactly one frame-codec line before it hands stdin and
// stdout to the HTTP/2 client: the READY line, so the host readiness gate
// accepts the same handshake it already accepts for every mode. After that
// one write, only the HTTP/2 client touches stdout: this function calls
// writeFrame no more, and it starts no heartbeat timer, so no non-HTTP/2
// writer can put a byte on stdout between the READY line and the client
// connection preface.
// ---------------------------------------------------------------------------
function createStdioDuplex() {
const duplex = new Duplex({
read() {
// process.stdin pushes bytes through the "data" listener below; there
// is nothing to pull on demand here.
},
write(chunk, _encoding, callback) {
const flushed = process.stdout.write(chunk);
if (flushed) {
callback();
} else {
process.stdout.once("drain", () => callback());
}
},
});
process.stdin.on("data", (chunk) => {
duplex.push(chunk);
});
process.stdin.on("end", () => {
duplex.push(null);
});
process.stdin.on("error", (error) => {
duplex.destroy(error instanceof Error ? error : new Error(String(error)));
});
return duplex;
}
function runHttp2Gateway() {
function diag(message) {
// Diagnostics go to stderr only, the same as every other mode.
process.stderr.write("[paperclip-bridge] " + message + "\\n");
}
function writeFrame(frame) {
process.stdout.write(encodeDuplexFrame(frame));
}
const authority = "bridge.internal";
let session = null;
let unavailable = false;
function openSession() {
if (session) return session;
const stdio = createStdioDuplex();
session = http2.connect("http://" + authority, {
createConnection: () => stdio,
});
session.on("error", () => {
unavailable = true;
});
session.on("close", () => {
unavailable = true;
});
session.on("goaway", (errorCode, lastStreamId) => {
diag("host sent GOAWAY (errorCode=" + errorCode + ", lastStreamId=" + lastStreamId + ")");
});
return session;
}
function forwardOverHttp2(request) {
return new Promise((resolve, reject) => {
const activeSession = openSession();
const query = request.query || "";
const pathWithQuery =
query.length === 0 ? request.path : request.path + (query.charAt(0) === "?" ? query : "?" + query);
const outboundHeaders = Object.assign(
{
":method": request.method,
":path": pathWithQuery,
authorization: "Bearer " + bridgeToken,
},
request.headers,
);
let stream;
try {
stream = activeSession.request(outboundHeaders, { endStream: request.body.length === 0 });
} catch (error) {
reject(error instanceof Error ? error : new Error(String(error)));
return;
}
const chunks = [];
let status = 502;
let responseHeaders = {};
let settled = false;
const settle = (run) => {
if (settled) return;
settled = true;
run();
};
stream.on("response", (headers) => {
const raw = headers[":status"];
status = typeof raw === "number" ? raw : Number(raw) || 502;
responseHeaders = {};
for (const [key, value] of Object.entries(headers)) {
if (key.charAt(0) === ":" || value == null) continue;
responseHeaders[key] = Array.isArray(value) ? value.join(", ") : String(value);
}
});
stream.on("data", (chunk) => chunks.push(chunk));
stream.once("end", () =>
settle(() => resolve({ status: status, headers: responseHeaders, body: Buffer.concat(chunks) })),
);
stream.once("error", (error) =>
settle(() => reject(error instanceof Error ? error : new Error(String(error)))),
);
stream.once("aborted", () => settle(() => reject(new Error("Bridge HTTP/2 stream aborted."))));
if (request.body.length > 0) {
stream.end(request.body);
} else if (!stream.writableEnded) {
stream.end();
}
});
}
const server = createServer(async (req, res) => {
try {
const auth = req.headers.authorization || "";
const receivedToken = auth.startsWith("Bearer ") ? auth.slice("Bearer ".length) : "";
if (!tokensMatch(receivedToken)) {
writeJsonResponse(res, 401, { error: "Invalid bridge token." });
return;
}
if (unavailable) {
writeJsonResponse(res, 503, { error: "bridge_unavailable" });
return;
}
const url = new URL(req.url || "/", "http://127.0.0.1");
const contentType = typeof req.headers["content-type"] === "string" ? req.headers["content-type"] : "";
if (req.method && req.method !== "GET" && req.method !== "HEAD" && !/json/i.test(contentType)) {
writeJsonResponse(res, 415, { error: "Bridge only accepts JSON request bodies." });
return;
}
const requestBodyBuffer = Buffer.from(await readBody(req), "utf8");
let response;
try {
response = await forwardOverHttp2({
method: req.method || "GET",
path: url.pathname,
query: url.search,
headers: normalizeHeaders(req.headers),
body: requestBodyBuffer,
});
} catch (error) {
writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) });
return;
}
res.statusCode = typeof response.status === "number" ? response.status : 200;
for (const [key, value] of Object.entries(response.headers || {})) {
if (typeof value !== "string" || key.toLowerCase() === "content-length") continue;
res.setHeader(key, value);
}
res.end(response.body);
} catch (error) {
writeJsonResponse(res, 502, { error: error instanceof Error ? error.message : String(error) });
}
});
process.on("SIGINT", () => {
try {
server.close();
} catch (error) {
diag("server close error: " + (error && error.message ? error.message : String(error)));
}
process.exit(0);
});
process.on("SIGTERM", () => {
try {
server.close();
} catch (error) {
diag("server close error: " + (error && error.message ? error.message : String(error)));
}
process.exit(0);
});
// Bind-or-exit, the same rule every mode applies: the host assigns a
// positive loopback port, and the gateway binds exactly that port or exits
// nonzero. It never selects a different port.
if (!Number.isInteger(port) || port <= 0) {
diag("http2 gateway requires a positive assigned PAPERCLIP_BRIDGE_PORT; got " + String(port));
process.exit(1);
}
server.on("error", (error) => {
diag(
"http2 gateway could not bind port " +
String(port) +
": " +
(error && error.message ? error.message : String(error)),
);
process.exit(1);
});
server.listen(port, host, () => {
const address = server.address();
if (!address || typeof address === "string") {
diag("http2 gateway did not expose a TCP address");
process.exit(1);
return;
}
// Send the one frame-codec line on this path: the READY line the host
// readiness gate expects. Open the HTTP/2 client session on the very next
// statement, so the client connection preface is the next byte the host
// sees after READY, with no other writer in between.
writeFrame({ version: DUPLEX_FRAME_VERSION, type: "ready", nonce: bridgeNonce });
gatewayReady = true;
openSession();
});
}
if (bridgeMode === "${SANDBOX_CALLBACK_BRIDGE_HTTP2_MODE}") {
runHttp2Gateway();
} else if (bridgeMode === "${SANDBOX_CALLBACK_BRIDGE_DUPLEX_MODE}") {
// No host selection path sets this mode anymore (http2_v1 replaced it), but
// the generated gateway keeps the mode reachable: it stays defined here,
// unchanged, so nothing that still spawns the gateway directly with this
// mode name breaks.
runDuplexGateway();
} else {
await runFileGateway();

View File

@ -106,29 +106,38 @@ function entriesMatch(left: SnapshotEntry | null | undefined, right: SnapshotEnt
return false;
}
async function isHolderAlive(lockDir: string): Promise<boolean> {
const LOCK_STALE_MS = 30_000;
async function isLockStale(lockDir: string): Promise<boolean> {
try {
const raw = await fs.readFile(path.join(lockDir, "owner.json"), "utf8");
const owner = JSON.parse(raw) as { pid?: unknown };
const pid = typeof owner.pid === "number" && Number.isFinite(owner.pid) && owner.pid > 0 ? owner.pid : null;
if (pid === null) {
// Owner record is unparseable / missing pid — treat as stale.
return false;
return true;
}
try {
process.kill(pid, 0);
return true;
} catch {
return false;
} catch {
return true;
}
} catch {
// owner.json missing or unreadable — treat as stale.
return false;
// owner.json is missing or unreadable. A live holder also passes through
// this exact state, briefly, between its own `fs.mkdir(lockDir)` and its
// `fs.writeFile(owner.json)` below. Reading "missing" as "stale" here would
// let a concurrent acquirer delete a live holder's lock directory during
// that window. Mirror the materializePaperclipSkillCopy lock pattern: fall
// back to the lock directory's own mtime, and only call it stale once the
// directory itself has outlived the stale threshold.
const stat = await fs.stat(lockDir).catch(() => null);
return !stat || Date.now() - stat.mtimeMs > LOCK_STALE_MS;
}
}
async function acquireDirectoryMergeLock(lockDir: string): Promise<() => Promise<void>> {
const deadline = Date.now() + 30_000;
const deadline = Date.now() + LOCK_STALE_MS;
while (true) {
try {
await fs.mkdir(lockDir);
@ -146,7 +155,7 @@ async function acquireDirectoryMergeLock(lockDir: string): Promise<() => Promise
// Stale-lock detection: if the owner PID is dead (SIGKILL / OOM / crash),
// the lockDir would otherwise persist forever and stall restores. Mirror
// the materializePaperclipSkillCopy lock pattern — remove and retry.
if (!(await isHolderAlive(lockDir))) {
if (await isLockStale(lockDir)) {
await fs.rm(lockDir, { recursive: true, force: true }).catch(() => undefined);
continue;
}

View File

@ -24,6 +24,8 @@
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import { randomUUID } from "node:crypto";
import http2 from "node:http2";
import { Duplex } from "node:stream";
import {
openDaytonaDuplexChannelSession,
type DaytonaPtyProcess,
@ -117,7 +119,7 @@ describeLive("Daytona duplex channel (live)", () => {
// The channel child still answers. A write returns as program output.
const ping = `chan-${randomUUID()}`;
session.write(`${ping}\n`);
session.write(Buffer.from(`${ping}\n`));
await waitFor(readOutput, (text) => text.includes(ping), 30_000, "channel echo");
} finally {
await session.close();
@ -139,7 +141,7 @@ describeLive("Daytona duplex channel (live)", () => {
const baseline = readOutput().length;
const line = `PING-${randomUUID()}`;
session.write(`${line}\n`);
session.write(Buffer.from(`${line}\n`));
await waitFor(
readOutput,
(text) => text.slice(baseline).includes(line),
@ -214,7 +216,7 @@ describeLive("Daytona duplex channel (live)", () => {
const line = `RTT-${index}-${randomUUID()}`;
const baseline = readOutput().length;
const start = Date.now();
session.write(`${line}\n`);
session.write(Buffer.from(`${line}\n`));
await waitFor(
readOutput,
(text) => text.slice(baseline).includes(line),
@ -262,7 +264,7 @@ describeLive("Daytona duplex channel (live)", () => {
// before its echo returns. The abrupt close models a lost provider channel.
await new Promise((resolve) => setTimeout(resolve, 4_000));
const inFlight = `INFLIGHT-${randomUUID()}`;
session.write(`${inFlight}\n`);
session.write(Buffer.from(`${inFlight}\n`));
// Close at once, without waiting for the echo. The pending round trip never
// settles through the stream; the channel tears down instead.
await session.close();
@ -288,4 +290,124 @@ describeLive("Daytona duplex channel (live)", () => {
},
LIVE_TIMEOUT_MS,
);
it(
"test_live_daytona_run_uses_http2_v1_end_to_end",
async () => {
// Phase 4 selects http2_v1 by running one Node HTTP/2 session directly
// on this same pseudo-terminal channel, right after the READY line.
// This package ships standalone (see the file header), so it cannot
// import the host readiness gate or the preface scan from
// `@paperclipai/adapter-utils`; this test reimplements the minimal,
// self-contained version of both, using only `node:http2`, so the
// proof runs against a real Daytona PTY end to end.
const live = sandbox!;
const nonce = randomUUID();
// The 24-octet HTTP/2 client connection preface (RFC 9113, Section 3.4).
const CLIENT_PREFACE = Buffer.from("505249202a20485454502f322e300d0a0d0a534d0d0a0d0a", "hex");
// The sandbox-side child: send the READY line, then hand stdin/stdout to
// a real HTTP/2 client session and dispatch one request — the same
// shape `runHttp2Gateway()` in `sandbox-callback-bridge.ts` runs.
const nodeScript = [
`process.stdout.write(JSON.stringify({version:2,type:"ready",nonce:${JSON.stringify(nonce)}})+"\\n");`,
`const http2=require("node:http2");`,
`const {Duplex}=require("node:stream");`,
`const stdio=new Duplex({read(){},write(chunk,enc,cb){const ok=process.stdout.write(chunk);if(ok)cb();else process.stdout.once("drain",cb);}});`,
`process.stdin.on("data",c=>stdio.push(c));`,
`process.stdin.on("end",()=>stdio.push(null));`,
`const session=http2.connect("http://bridge.internal",{createConnection:()=>stdio});`,
`session.on("error",()=>process.exit(1));`,
`const stream=session.request({":method":"GET",":path":"/ping"});`,
`let body="";`,
`stream.on("data",c=>{body+=c;});`,
`stream.on("error",()=>process.exit(3));`,
`stream.on("end",()=>{session.close(()=>process.exit(body==="pong"?0:2));});`,
`stream.end();`,
].join("");
const session = await openDaytonaDuplexChannelSession(live.process, ["node", "-e", nodeScript]);
try {
// Reads bytes from the channel until it finds one complete newline-
// terminated READY line, then hands every later byte — including any
// already-buffered suffix of the same chunk — to `onAfterReady`.
const readyAndAfter = await new Promise<{ nonce: string; afterReady: Buffer }>((resolve, reject) => {
let buffer = Buffer.alloc(0);
const timer = setTimeout(
() => reject(new Error(`Timed out waiting for the READY line. Bytes so far: ${buffer.length}`)),
30_000,
);
session.onData((chunk) => {
buffer = Buffer.concat([buffer, Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)]);
const newlineIndex = buffer.indexOf(0x0a);
if (newlineIndex === -1) return;
clearTimeout(timer);
const line = buffer.subarray(0, newlineIndex).toString("utf8");
const decoded = JSON.parse(line) as { type?: string; nonce?: string };
resolve({ nonce: decoded.nonce ?? "", afterReady: Buffer.from(buffer.subarray(newlineIndex + 1)) });
});
});
expect(readyAndAfter.nonce).toBe(nonce);
// Scan the retained suffix for the client preface, the same rule
// `createHttp2PrefaceScanningChannel` in `execution-target.ts` applies:
// the scan window opens only on bytes after the accepted READY line.
let sawPreface = false;
let downstream: ((chunk: Buffer) => void) | null = null;
let pendingAfterPreface = Buffer.alloc(0);
let scanBuffer = readyAndAfter.afterReady;
function deliver(chunk: Buffer): void {
if (downstream) downstream(chunk);
else pendingAfterPreface = Buffer.concat([pendingAfterPreface, chunk]);
}
function handleChunk(chunk: Buffer): void {
if (sawPreface) {
deliver(chunk);
return;
}
scanBuffer = Buffer.concat([scanBuffer, chunk]);
const offset = scanBuffer.indexOf(CLIENT_PREFACE);
if (offset === -1) return;
sawPreface = true;
const fromPreface = Buffer.from(scanBuffer.subarray(offset));
scanBuffer = Buffer.alloc(0);
deliver(fromPreface);
}
// The already-retained suffix might already hold the preface.
handleChunk(Buffer.alloc(0));
session.onData((chunk) => handleChunk(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
// Wrap the channel as a Node `Duplex` starting at the preface offset,
// and bind one plaintext HTTP/2 server session on it.
const boundDuplex: Duplex = new Duplex({
read() {},
write(chunk: unknown, _encoding, callback) {
session.write(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as ArrayBufferLike));
callback();
},
});
downstream = (chunk) => boundDuplex.push(chunk);
if (pendingAfterPreface.length > 0) {
boundDuplex.push(pendingAfterPreface);
pendingAfterPreface = Buffer.alloc(0);
}
const server = http2.createServer();
server.on("stream", (stream) => {
stream.respond({ ":status": 200 });
stream.end("pong");
});
server.emit("connection", boundDuplex);
const exit = await session.wait();
expect(exit.exitCode).toBe(0);
// eslint-disable-next-line no-console
console.log("[duplex-live] http2_v1: READY, then the real client preface, then one full HTTP/2 round trip, all over one live Daytona PTY.");
} finally {
await session.close();
}
},
LIVE_TIMEOUT_MS,
);
});

View File

@ -270,10 +270,10 @@ describe("openDaytonaDuplexChannelSession", () => {
const received: string[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
session.onData((chunk) => received.push(new TextDecoder().decode(chunk)));
// The launch wrapper is the first input; the host frame write is the second.
session.write('{"version":1,"type":"heartbeat"}\n');
session.write(new TextEncoder().encode('{"version":1,"type":"heartbeat"}\n'));
// The chunker awaits each send, so let the write settle before the assertion.
await new Promise((resolve) => setImmediate(resolve));
expect(process.handle?.inputs[1]).toBe('{"version":1,"type":"heartbeat"}\n');
@ -287,7 +287,7 @@ describe("openDaytonaDuplexChannelSession", () => {
const received: string[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
session.onData((chunk) => received.push(new TextDecoder().decode(chunk)));
process.handle?.emitText('{"version":1,"type":"ready","address":"127.0.0.1:8080"}\n');
process.handle?.emitText('{"version":1,"type":"heartbeat"}\n');
@ -305,7 +305,7 @@ describe("openDaytonaDuplexChannelSession", () => {
const received: string[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
session.onData((chunk) => received.push(new TextDecoder().decode(chunk)));
const frame = '{"version":1,"type":"response","id":"r1","status":200,"headers":{},"body":"","outcome":"completed"}\n';
process.handle?.emitText(frame.slice(0, 20));
@ -319,7 +319,7 @@ describe("openDaytonaDuplexChannelSession", () => {
const received: string[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
session.onData((chunk) => received.push(new TextDecoder().decode(chunk)));
const body = "x".repeat(500_000);
const frame = `{"version":1,"type":"response","id":"big","status":200,"headers":{},"body":"${body}","outcome":"completed"}\n`;
@ -329,20 +329,41 @@ describe("openDaytonaDuplexChannelSession", () => {
expect(received.join("").length).toBe(frame.length);
});
it("keeps a multibyte character whole across two output chunks", async () => {
it("test_stream_forwards_bytes_without_utf8_conversion", async () => {
const process = createFakeProcess();
const received: string[] = [];
const received: Uint8Array[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
// The euro sign is three UTF-8 bytes. Split it across two chunks, so the
// stream decoder must join the bytes.
// The euro sign is three UTF-8 bytes. Split it across two chunks. The
// session no longer runs a UTF-8 stream decoder (see `duplex-command-stream.ts`),
// so it forwards each raw byte chunk unchanged; it does not wait for a whole
// character. The two chunks concatenate back to the exact original bytes —
// proof that no layer here converts the data to a string.
const euro = new TextEncoder().encode("€");
process.handle?.emitBytes(euro.subarray(0, 2));
process.handle?.emitBytes(euro.subarray(2));
expect(received.join("")).toBe("€");
expect(received).toHaveLength(2);
expect(received[0]).toEqual(euro.subarray(0, 2));
expect(received[1]).toEqual(euro.subarray(2));
const joined = Buffer.concat(received);
expect(joined).toEqual(Buffer.from(euro));
expect(new TextDecoder().decode(joined)).toBe("€");
});
it("sends the full 256-value byte corpus through the channel unchanged", async () => {
const process = createFakeProcess();
const received: Uint8Array[] = [];
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
session.onData((chunk) => received.push(chunk));
const allByteValues = Uint8Array.from({ length: 256 }, (_, value) => value);
process.handle?.emitBytes(allByteValues);
expect(Buffer.concat(received)).toEqual(Buffer.from(allByteValues));
});
it("buffers early output until the listener registers", async () => {
@ -355,7 +376,7 @@ describe("openDaytonaDuplexChannelSession", () => {
process.handle?.emitText('"type":"heartbeat"}\n');
expect(received).toEqual([]);
session.onData((chunk) => received.push(chunk));
session.onData((chunk) => received.push(new TextDecoder().decode(chunk)));
// The session flushes the buffered output in order on registration.
expect(received).toEqual(['{"version":1,"type":"heartbeat"}\n']);
});
@ -393,7 +414,7 @@ describe("openDaytonaDuplexChannelSession", () => {
// raised chunk size above this size makes the write one send and fails this
// test. The payload is ASCII, so the fake's per-send decode keeps each byte.
const payload = "x".repeat(150_416);
session.write(payload);
session.write(new TextEncoder().encode(payload));
// The chunker awaits each send, so let the write settle before the assertion.
await new Promise((resolve) => setImmediate(resolve));
@ -414,7 +435,7 @@ describe("openDaytonaDuplexChannelSession", () => {
if (!handle) throw new Error("The fake process opened no handle.");
const before = handle.inputs.length;
session.write('{"version":1,"type":"heartbeat"}\n');
session.write(new TextEncoder().encode('{"version":1,"type":"heartbeat"}\n'));
// The chunker awaits each send, so let the write settle before the assertion.
await new Promise((resolve) => setImmediate(resolve));
@ -449,8 +470,8 @@ describe("openDaytonaDuplexChannelSession", () => {
const first = "a".repeat(150_416);
const second = "b".repeat(150_416);
session.write(first);
session.write(second);
session.write(new TextEncoder().encode(first));
session.write(new TextEncoder().encode(second));
// Let both writes settle.
await new Promise((resolve) => setImmediate(resolve));
@ -514,7 +535,7 @@ describe("openDaytonaDuplexChannelSession", () => {
onWriteError: (reason) => writeErrors.push(reason),
});
session.write("frame\n");
session.write(new TextEncoder().encode("frame\n"));
// Let the rejected write settle.
await new Promise((resolve) => setImmediate(resolve));
@ -553,8 +574,8 @@ describe("openDaytonaDuplexChannelSession", () => {
onWriteError: (reason) => writeErrors.push(reason),
});
session.write("a");
session.write("b");
session.write(new TextEncoder().encode("a"));
session.write(new TextEncoder().encode("b"));
await new Promise((resolve) => setImmediate(resolve));
expect(writeErrors).toEqual(["write_error"]);
@ -595,8 +616,8 @@ describe("openDaytonaDuplexChannelSession", () => {
onWriteError: (reason) => writeErrors.push(reason),
});
session.write("first\n");
session.write("second\n");
session.write(new TextEncoder().encode("first\n"));
session.write(new TextEncoder().encode("second\n"));
// Let the write chain settle both queued writes.
await new Promise((resolve) => setImmediate(resolve));
await new Promise((resolve) => setImmediate(resolve));

View File

@ -54,10 +54,10 @@ import { sendPtyInputInChunks } from "./pty-chunked-input.js";
* worker forwards the data and the exit with no adapter.
*/
export interface DuplexChannelSession {
/** Registers the one data listener. The session streams each raw chunk in order. */
onData(listener: (chunk: string) => void): void;
/** Registers the one data listener. The session streams each raw byte chunk in order. */
onData(listener: (chunk: Uint8Array) => void): void;
/** Writes raw input bytes to the pseudo-terminal. */
write(data: string): void;
write(data: Uint8Array): void;
/**
* Resolves when the command ends or the transport closes. A numeric `exitCode`
* is a real process exit. `transportClosed` is true when the pseudo-terminal
@ -169,18 +169,16 @@ export function buildDuplexChannelLaunchWrapper(
* {@link DuplexChannelSession}. The session allocates a real pseudo-terminal in
* raw mode, streams the raw output, accepts host input, and stops the child.
*
* The function decodes the terminal bytes as a UTF-8 stream, so a multibyte
* character that splits across two output chunks stays whole. It buffers the
* output until the transport registers the listener, so no early chunk is lost.
* The function forwards the terminal bytes unchanged. It buffers the output
* until the transport registers the listener, so no early chunk is lost.
*/
export async function openDaytonaDuplexChannelSession(
process: DaytonaPtyProcess,
command: readonly string[],
options?: DaytonaDuplexChannelOptions,
): Promise<DuplexChannelSession> {
const decoder = new TextDecoder("utf-8");
let listener: ((chunk: string) => void) | null = null;
let buffered = "";
let listener: ((chunk: Uint8Array) => void) | null = null;
let buffered: Buffer = Buffer.alloc(0);
const diagnosticsPath =
options?.diagnosticsPath ?? `/tmp/paperclip-duplex-${randomUUID()}.log`;
@ -191,13 +189,12 @@ export async function openDaytonaDuplexChannelSession(
cols: DUPLEX_CHANNEL_PTY_COLS,
rows: DUPLEX_CHANNEL_PTY_ROWS,
onData: (data: Uint8Array): void => {
// Decode the terminal bytes as a stream, so a split multibyte character
// stays whole across two chunks. Forward the raw text; the frame codec owns
// the newline-delimited JSON parsing.
const text = decoder.decode(data, { stream: true });
if (text.length === 0) return;
if (listener) listener(text);
else buffered += text;
// Forward the raw bytes unchanged; the frame codec owns the
// newline-delimited JSON parsing and any multi-byte character
// reassembly on the read side.
if (data.byteLength === 0) return;
if (listener) listener(data);
else buffered = buffered.length === 0 ? Buffer.from(data) : Buffer.concat([buffered, data]);
},
});
@ -233,15 +230,15 @@ export async function openDaytonaDuplexChannelSession(
let writeChain: Promise<void> = Promise.resolve();
return {
onData(next: (chunk: string) => void): void {
onData(next: (chunk: Uint8Array) => void): void {
listener = next;
if (buffered.length > 0) {
const pending = buffered;
buffered = "";
buffered = Buffer.alloc(0);
next(pending);
}
},
write(data: string): void {
write(data: Uint8Array): void {
// Send the input as byte-bounded chunks under the provider message cap. A
// whole payload in one message can cross the cap and take the channel down,
// so the chunker slices the payload and sends each chunk in order. The chain

View File

@ -50,6 +50,10 @@ const manifest: PaperclipPluginManifestV1 = {
// over a raw pseudo-terminal. Declare the opt-in capability so the host may
// select the duplex transport. The host resolves it `true` only when the
// worker also verifies the `duplexChannelOpen` handler.
//
// Daytona is the first provider that runs the HTTP/2 transport over this
// channel. HTTP/2 is the preferred transport. `queue_v1` is the
// soft-deprecated fallback.
sandboxCapabilities: {
incrementalSessionOutput: true,
concurrentSyncOperations: true,

View File

@ -201,11 +201,13 @@ describe("Daytona sandbox provider plugin", () => {
mockCreate.mockResolvedValue(sandbox);
// Capture the data and the exit the worker forwards through `ctx.duplexChannel`.
// `ctx.duplexChannel.data` carries raw bytes; decode each chunk to text so the
// assertion below reads the plain-text payload.
const dataChunks: Array<{ hostRouteId: string; workerSessionId: string; chunk: string }> = [];
const restore = __setDaytonaPluginContextForTest({
duplexChannel: {
data: (hostRouteId: string, workerSessionId: string, chunk: string) =>
dataChunks.push({ hostRouteId, workerSessionId, chunk }),
data: (hostRouteId: string, workerSessionId: string, chunk: Uint8Array) =>
dataChunks.push({ hostRouteId, workerSessionId, chunk: Buffer.from(chunk).toString("utf8") }),
exit: () => {},
},
} as unknown as PluginContext);
@ -242,10 +244,12 @@ describe("Daytona sandbox provider plugin", () => {
expect(inputs[0]).toMatch(/2>'\/tmp\/paperclip-duplex-.+\.log'/);
// A host write on the exact pair reaches the process on the same channel.
// `data` arrives in the wire-safe base64 form (see `ChannelBytesWireValue`
// in the plugin SDK's protocol.ts).
await plugin.definition.onDuplexChannelWrite?.({
hostRouteId: "route-1",
workerSessionId,
data: '{"version":1,"type":"heartbeat"}\n',
data: Buffer.from('{"version":1,"type":"heartbeat"}\n', "utf8").toString("base64"),
});
expect(inputs[1]).toBe('{"version":1,"type":"heartbeat"}\n');
@ -255,7 +259,7 @@ describe("Daytona sandbox provider plugin", () => {
await plugin.definition.onDuplexChannelWrite?.({
hostRouteId: "route-foreign",
workerSessionId,
data: "foreign\n",
data: Buffer.from("foreign\n", "utf8").toString("base64"),
});
expect(inputs.length).toBe(inputsBeforeForeign);
// A stop whose pair does not match the bound entry stops nothing.
@ -297,7 +301,7 @@ describe("Daytona sandbox provider plugin", () => {
await plugin.definition.onDuplexChannelWrite?.({
hostRouteId: "route-1",
workerSessionId,
data: "late\n",
data: Buffer.from("late\n", "utf8").toString("base64"),
});
expect(inputs.length).toBe(inputsBefore);
} finally {

View File

@ -9,7 +9,7 @@ import type {
Resources,
Sandbox,
} from "@daytonaio/sdk";
import { definePlugin, NOOP_PLUGIN_TRACER } from "@paperclipai/plugin-sdk";
import { decodeChannelBytes, definePlugin, NOOP_PLUGIN_TRACER } from "@paperclipai/plugin-sdk";
import type {
PluginContext,
PluginTracer,
@ -2861,10 +2861,18 @@ const plugin = definePlugin({
// Write host input to an open duplex channel. Act only on the exact live pair.
// A write whose pair does not match the bound entry applies no bytes.
//
// `params.data` arrives in the wire-safe base64 form (JSON carries no binary
// type; see `ChannelBytesWireValue` in the plugin SDK's protocol.ts). Decode it
// back to raw bytes before it reaches the pseudo-terminal. A malformed value
// decodes to `null`; the worker applies no bytes rather than sending an empty
// write to the sandbox.
async onDuplexChannelWrite(params) {
const entry = daytonaDuplexChannelBySession.get(params.workerSessionId);
if (!entry || entry.hostRouteId !== params.hostRouteId) return;
entry.session.write(params.data);
const data = decodeChannelBytes(params.data);
if (data === null) return;
entry.session.write(data);
},
// Stop an open duplex channel child. Act only on the exact live pair. A stop

View File

@ -478,6 +478,8 @@ export interface PluginDefinition {
* 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.
*
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
*/
onDuplexChannelOpen?(
params: PluginDuplexChannelOpenParams,

View File

@ -86,6 +86,8 @@ export {
LOGIN_PTY_EXIT_NOTIFICATION,
DUPLEX_CHANNEL_DATA_NOTIFICATION,
DUPLEX_CHANNEL_EXIT_NOTIFICATION,
encodeChannelBytes,
decodeChannelBytes,
_resetIdCounter,
} from "./protocol.js";

View File

@ -1,8 +1,13 @@
import { describe, expect, it } from "vitest";
import {
createNotification,
decodeChannelBytes,
DUPLEX_CHANNEL_DATA_NOTIFICATION,
DUPLEX_CHANNEL_EXIT_NOTIFICATION,
encodeChannelBytes,
parseMessage,
serializeMessage,
type PluginDuplexChannelCloseParams,
type PluginDuplexChannelCloseResult,
type PluginDuplexChannelDataParams,
@ -219,3 +224,52 @@ describe("duplex channel notification schemas", () => {
expect(exit).toBeDefined();
});
});
// JSON-RPC travels as one line of JSON text (see `serializeMessage`), and JSON
// carries no binary type. `chunk` and `data` cross this hop as a base64 string
// (`ChannelBytesWireValue`); `encodeChannelBytes`/`decodeChannelBytes` are the
// one shared codec both ends use.
describe("duplex channel byte-safe wire representation", () => {
it("test_json_rpc_hop_preserves_all_byte_values", () => {
const allByteValues = Uint8Array.from({ length: 256 }, (_, value) => value);
// Encode the full byte corpus into a data notification, exactly as
// worker-rpc-host.ts does at the worker→host boundary, then send it through
// the same newline-delimited JSON serialization the host and the worker use
// over stdio.
const params: PluginDuplexChannelDataParams = {
hostRouteId: "route-1",
workerSessionId: "ws-1",
chunk: encodeChannelBytes(allByteValues),
};
const line = serializeMessage(
createNotification(DUPLEX_CHANNEL_DATA_NOTIFICATION, params),
);
// Parse the line back, exactly as the reading side does, and decode the
// chunk back to raw bytes.
const parsed = parseMessage(line);
expect("params" in parsed).toBe(true);
const receivedParams = (parsed as { params: PluginDuplexChannelDataParams }).params;
const decoded = decodeChannelBytes(receivedParams.chunk);
expect(decoded).not.toBeNull();
expect(decoded).toEqual(allByteValues);
// The byte value zero is the case a UTF-8 string hop loses or mistreats as a
// terminator. Assert it by name, not only as part of the full-corpus check.
expect(decoded?.[0]).toBe(0);
});
it("round-trips an empty byte chunk through encode and decode", () => {
const empty = new Uint8Array(0);
const encoded = encodeChannelBytes(empty);
expect(encoded).toBe("");
expect(decodeChannelBytes(encoded)).toBeNull();
});
it("decodeChannelBytes rejects a malformed wire value", () => {
expect(decodeChannelBytes("not valid base64!!")).toBeNull();
expect(decodeChannelBytes(undefined)).toBeNull();
expect(decodeChannelBytes(42)).toBeNull();
});
});

View File

@ -1099,6 +1099,50 @@ export const LOGIN_PTY_OUTPUT_NOTIFICATION = "loginPty.output";
/** The worker→host notification method for one pseudo-terminal exit. */
export const LOGIN_PTY_EXIT_NOTIFICATION = "loginPty.exit";
// ---------------------------------------------------------------------------
// Byte-safe duplex channel wire representation.
// ---------------------------------------------------------------------------
// A JSON-RPC message travels as one line of JSON text (see `serializeMessage`
// below). JSON has no binary type, so a raw byte chunk cannot cross this hop
// unchanged. `ChannelBytesWireValue` is the one JSON-safe encoding this
// protocol uses for a duplex channel chunk: a base64 string.
//
// Every layer above this hop carries the chunk as `Uint8Array`. This includes
// the plugin context, the worker RPC host's public duplex methods, and the
// host-side plugin worker manager. Only the JSON-RPC message itself holds the
// base64 form, and only for the one hop between the host process and the
// worker process.
//
// This base64 form is not the sandbox provider channel's wire format. That
// channel carries raw bytes with no base64 armor: a live measurement of the
// provider transport proved that every byte value survives it unchanged.
//
// HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
/** The wire-safe JSON-RPC form of one duplex channel byte chunk: a base64 string. */
export type ChannelBytesWireValue = string;
/** Encodes raw channel bytes into the wire-safe JSON-RPC representation. */
export function encodeChannelBytes(bytes: Uint8Array): ChannelBytesWireValue {
return Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength).toString("base64");
}
/**
* Decodes the wire-safe JSON-RPC representation back to raw channel bytes.
* Returns `null` for a value that is not a well-formed base64 string, so a
* caller on the trust boundary treats a malformed frame as a protocol error
* instead of silently substituting the empty byte array.
*/
export function decodeChannelBytes(value: unknown): Uint8Array | null {
if (typeof value !== "string" || value.length === 0) return null;
// `Buffer.from(str, "base64")` silently drops an invalid character instead
// of throwing, so re-encode the decoded bytes and compare. A well-formed
// base64 string round-trips to itself; a malformed one does not.
const decoded = Buffer.from(value, "base64");
if (decoded.toString("base64") !== value) return null;
return new Uint8Array(decoded.buffer, decoded.byteOffset, decoded.byteLength);
}
// ---------------------------------------------------------------------------
// Generic duplex channel worker methods.
// ---------------------------------------------------------------------------
@ -1150,8 +1194,8 @@ export interface PluginDuplexChannelWriteParams {
hostRouteId: string;
/** The worker session identifier that the open reply returned. */
workerSessionId: string;
/** The raw input bytes to write to the channel. */
data: string;
/** The raw input bytes to write to the channel, in the {@link ChannelBytesWireValue} wire form. */
data: ChannelBytesWireValue;
}
/** The stop request. It carries the exact route pair. */
@ -1196,8 +1240,8 @@ export interface PluginDuplexChannelDataParams {
hostRouteId: string;
/** The worker session identifier that the open reply returned. */
workerSessionId: string;
/** The raw channel output bytes. */
chunk: string;
/** The raw channel output bytes, in the {@link ChannelBytesWireValue} wire form. */
chunk: ChannelBytesWireValue;
}
/** The worker→host duplex channel exit notification parameters. */

View File

@ -2487,7 +2487,7 @@ export function createTestHarness(options: TestHarnessOptions): TestHarness {
},
},
duplexChannel: {
data(_hostRouteId: string, _workerSessionId: string, _chunk: string) {
data(_hostRouteId: string, _workerSessionId: string, _chunk: Uint8Array) {
// No-op in test harness — the host duplex route is not wired here.
},
exit(_hostRouteId: string, _workerSessionId: string, _exitCode: number | null) {

View File

@ -2056,9 +2056,9 @@ export interface PluginDuplexChannelClient {
*
* @param hostRouteId - The host route identifier the open request carried. The worker echoes it, so the host routes the exact pair.
* @param workerSessionId - The worker session identifier the open reply returned.
* @param chunk - The raw channel output text.
* @param chunk - The raw channel output bytes.
*/
data(hostRouteId: string, workerSessionId: string, chunk: string): void;
data(hostRouteId: string, workerSessionId: string, chunk: Uint8Array): void;
/**
* Deliver the child exit of a persistent duplex channel.
*

View File

@ -133,6 +133,7 @@ import {
isJsonRpcErrorResponse,
JsonRpcParseError,
JsonRpcCallError,
encodeChannelBytes,
} from "./protocol.js";
// ---------------------------------------------------------------------------
@ -1403,17 +1404,25 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost
},
duplexChannel: {
data(hostRouteId: string, workerSessionId: string, chunk: string): void {
data(hostRouteId: string, workerSessionId: string, chunk: Uint8Array): void {
// Forward one raw data chunk of a persistent duplex channel. The
// notification echoes the host route identifier and the worker session
// identifier, so the host routes the chunk to the exact live pair while
// the route is open. The host drops an unknown or a mismatched pair and
// never logs the raw bytes. This notification carries no invocation id,
// because it fires after the open reply returns.
//
// JSON-RPC travels as JSON text, which carries no binary type, so this
// is the one point where the chunk crosses from `Uint8Array` to the
// wire-safe base64 form. See `ChannelBytesWireValue` in protocol.ts.
if (typeof hostRouteId !== "string" || hostRouteId.length === 0) return;
if (typeof workerSessionId !== "string" || workerSessionId.length === 0) return;
if (typeof chunk !== "string" || chunk.length === 0) return;
notifyHost(DUPLEX_CHANNEL_DATA_NOTIFICATION, { hostRouteId, workerSessionId, chunk });
if (!(chunk instanceof Uint8Array) || chunk.byteLength === 0) return;
notifyHost(DUPLEX_CHANNEL_DATA_NOTIFICATION, {
hostRouteId,
workerSessionId,
chunk: encodeChannelBytes(chunk),
});
},
exit(
hostRouteId: string,

View File

@ -182,6 +182,8 @@ export interface SandboxProviderCapabilities {
* 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.
*
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
*/
duplexCommandStream?: boolean;
}

View File

@ -149,8 +149,9 @@ describe("sandbox driver duplex channel wiring", () => {
});
// write / stop / close map to write / kill / close on the host session.
channel.write("input-bytes");
expect(hostSession.write).toHaveBeenCalledWith("input-bytes");
const inputBytes = new TextEncoder().encode("input-bytes");
channel.write(inputBytes);
expect(hostSession.write).toHaveBeenCalledWith(inputBytes);
channel.stop();
expect(hostSession.kill).toHaveBeenCalledTimes(1);
await channel.close();

View File

@ -22,7 +22,15 @@
// read order — proving the host still bounds these frames on their own terms
// and never lets the exit crowd a data frame out of the pre-open hold.
// - `echoInput`: when true, the fixture echoes each `duplexChannelWrite` back as
// one data notification for the bound session.
// one data notification for the bound session, prefixed with `echo:` at the
// byte level (so the echo round-trips a non-UTF-8 payload unchanged).
//
// The JSON-RPC hop carries a duplex chunk as a base64 string (see
// `ChannelBytesWireValue` in packages/plugins/sdk/src/protocol.ts). This fixture
// is a hand-rolled worker, not the real SDK, so it encodes and decodes base64
// itself with the two helpers below. A `chunk` directive value in a test is
// always the plain-text payload; the fixture is the one place that puts it on
// the wire as base64.
// - `writeReplyDelayMs`: when a positive number, the fixture delays each
// `duplexChannelWrite` reply by that many milliseconds, so a test proves the
// host holds the pending-write reservation until the RPC settles.
@ -48,6 +56,19 @@ function send(message) {
process.stdout.write(`${JSON.stringify(message)}\n`);
}
// Encode one plain-text directive chunk to the wire-safe base64 form. The real
// worker (worker-rpc-host.ts, `encodeChannelBytes`) does the same encoding.
function toWireChunk(text) {
return Buffer.from(text, "utf8").toString("base64");
}
// Decode the wire-safe base64 form of one host→worker write back to a Buffer,
// so the echo path works on exact bytes rather than a UTF-8 string. This keeps
// the echo byte-exact for a payload that is not valid UTF-8.
function fromWireChunk(wireValue) {
return Buffer.from(wireValue, "base64");
}
// Serialize the scripted data and exit frames as newline-delimited lines. The
// batch mode writes these together with the open reply in one stdout write. Each
// frame carries the exact pair, so the host binds and routes by the pair. A test
@ -62,7 +83,7 @@ function scriptedFrameLines(directive, hostRouteId, workerSessionId) {
params: {
hostRouteId: entry.rid ?? hostRouteId,
workerSessionId: entry.sid ?? workerSessionId,
chunk: entry.chunk,
chunk: toWireChunk(entry.chunk),
},
})}\n`;
}
@ -89,7 +110,7 @@ function scriptedFrameLines(directive, hostRouteId, workerSessionId) {
params: {
hostRouteId: entry.rid ?? hostRouteId,
workerSessionId: entry.sid ?? workerSessionId,
chunk: entry.chunk,
chunk: toWireChunk(entry.chunk),
},
})}\n`;
}
@ -223,13 +244,16 @@ rl.on("line", (line) => {
if (entry.echoInput) {
// Echo the input back as one data notification for the bound pair, so a
// test proves the input reaches the worker and the output routes back.
// Work on the decoded Buffer, not a string, so a non-UTF-8 payload (the
// full 256-value byte corpus) echoes byte-exact under the `echo:` prefix.
const echoBytes = Buffer.concat([Buffer.from("echo:", "utf8"), fromWireChunk(params.data)]);
send({
jsonrpc: "2.0",
method: "duplexChannel.data",
params: {
hostRouteId: entry.hostRouteId,
workerSessionId: entry.workerSessionId,
chunk: `echo:${params.data}`,
chunk: echoBytes.toString("base64"),
},
});
}
@ -278,7 +302,7 @@ rl.on("line", (line) => {
params: {
hostRouteId: entry.hostRouteId,
workerSessionId: entry.workerSessionId,
chunk: entry.emitAfterCloseChunk,
chunk: toWireChunk(entry.emitAfterCloseChunk),
},
});
});

View File

@ -104,7 +104,10 @@ describe("plugin worker manager duplex aggregate byte ledger", () => {
});
// A late listener drains the terminal buffered record and releases its token.
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session now streams raw `Uint8Array` chunks. Decode each one back
// to text, so the assertion below still compares the plain-text payload
// the fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await vi.waitFor(() => {
expect(chunks).toEqual(["ok"]);
expect(ledger.bytesInUse).toBe(0);
@ -142,7 +145,10 @@ describe("plugin worker manager duplex aggregate byte ledger", () => {
expect(ledger.liveTokenCount).toBe(2);
});
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session now streams raw `Uint8Array` chunks. Decode each one back
// to text, so the assertion below still compares the plain-text payload
// the fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await vi.waitFor(() => {
expect(chunks).toEqual(["aa", "bb"]);
expect(ledger.bytesInUse).toBe(0);

View File

@ -133,9 +133,10 @@ describe("plugin worker manager duplex pending-write byte ledger", () => {
const route = await handle.openDuplexChannel(
duplexOpenInput({ workerSessionId: "ws-a", mode: "no-write-reply" }),
);
route.write(writeData);
route.write(writeData);
route.write(writeData);
const writeBytesEncoded = new TextEncoder().encode(writeData);
route.write(writeBytesEncoded);
route.write(writeBytesEncoded);
route.write(writeBytesEncoded);
// The worker reads its stdin, so each transport token flushes and releases.
// Only the three held raw payloads remain, so the gauge settles at three
// times the payload byte count with three live tokens.
@ -168,7 +169,7 @@ describe("plugin worker manager duplex pending-write byte ledger", () => {
duplexOpenInput({ workerSessionId: "ws-d", writeReplyDelayMs: 150 }),
);
const writeBytes = 1000;
route.write("x".repeat(writeBytes));
route.write(new TextEncoder().encode("x".repeat(writeBytes)));
// The transport token flushes at once, so one raw-payload token stays held for
// the full write byte count while the RPC is in flight.
await vi.waitFor(() => {
@ -201,7 +202,7 @@ describe("plugin worker manager duplex pending-write byte ledger", () => {
const data = "a€";
expect(data.length).toBe(2);
expect(Buffer.byteLength(data, "utf8")).toBe(4);
route.write(data);
route.write(new TextEncoder().encode(data));
// The transport token flushes, so one raw-payload token of four bytes remains.
await vi.waitFor(() => {
expect(ledger.liveTokenCount).toBe(1);
@ -232,7 +233,7 @@ describe("plugin worker manager duplex pending-write byte ledger", () => {
// The single write is larger than the ceiling. The raw-payload reservation
// fails, so the manager ends the route with the aggregate marker, never the
// route-busy marker, and never writes the frame.
route.write("x".repeat(200));
route.write(new TextEncoder().encode("x".repeat(200)));
expect(telemetry.rejections).toBeGreaterThanOrEqual(1);
expect(ledger.bytesInUse).toBe(0);
expect(telemetry.peak).toBeLessThanOrEqual(ledger.ceilingBytes);

View File

@ -125,7 +125,7 @@ function loggedWarnReasons(): string[] {
const HELD_WRITE_CHARS = 200_000;
describe("plugin worker manager duplex stdin transport byte ledger", () => {
it("reserves the encoded serialized-frame size, larger than the raw payload, including JSON escaping", async () => {
it("reserves the encoded serialized-frame size, larger than the raw payload, including the base64 wire inflation", async () => {
const telemetry = peakTrackingTelemetry();
const ledger = new DuplexAggregateByteLedger({ ceilingBytes: 8 << 20, telemetry });
const handle = makeDuplexHandle({ duplexAggregateByteLedger: ledger });
@ -138,19 +138,24 @@ describe("plugin worker manager duplex stdin transport byte ledger", () => {
exitAfterStopMs: 500,
}),
);
// Every character is a quote. The JSON serialization escapes each quote to two
// bytes, so the serialized frame is at least twice the raw payload byte count.
const data = '"'.repeat(50_000);
const rawBytes = Buffer.byteLength(data, "utf8");
// The JSON-RPC hop carries the write payload as a base64 string, because JSON
// has no binary type (see `ChannelBytesWireValue` in protocol.ts). Base64
// encodes three raw bytes as four characters, so the serialized frame is
// larger than the raw payload by about that ratio, plus the small fixed cost
// of the surrounding JSON envelope (the method name and the route identifiers).
const rawBytes = 50_000;
const data = new Uint8Array(rawBytes).fill(0x22); // an arbitrary byte value
const encodedLength = Buffer.from(data).toString("base64").length;
route.write(data);
// The raw-payload token and the transport token both hold at once. The worker
// does not read its stdin, so the transport token never flushes.
expect(ledger.liveTokenCount).toBe(2);
const transportBytes = ledger.bytesInUse - rawBytes;
// The transport reservation covers the serialized frame, so it is larger than
// the raw payload and it includes the doubled escaped quotes.
// The transport reservation covers the serialized frame: at least the base64
// form of the raw payload, plus a bounded JSON envelope around it.
expect(transportBytes).toBeGreaterThan(rawBytes);
expect(transportBytes).toBeGreaterThanOrEqual(2 * rawBytes);
expect(transportBytes).toBeGreaterThanOrEqual(encodedLength);
expect(transportBytes).toBeLessThan(encodedLength + 5_000);
} finally {
await handle.stop().catch(() => undefined);
}
@ -182,7 +187,7 @@ describe("plugin worker manager duplex stdin transport byte ledger", () => {
);
const data = "x".repeat(HELD_WRITE_CHARS);
const rawBytes = Buffer.byteLength(data, "utf8");
route.write(data);
route.write(new TextEncoder().encode(data));
// Both tokens hold at first: the raw payload and the serialized frame.
expect(ledger.liveTokenCount).toBe(2);
const transportBytes = ledger.bytesInUse - rawBytes;
@ -219,7 +224,7 @@ describe("plugin worker manager duplex stdin transport byte ledger", () => {
);
const data = "x".repeat(HELD_WRITE_CHARS);
const rawBytes = Buffer.byteLength(data, "utf8");
route.write(data);
route.write(new TextEncoder().encode(data));
expect(ledger.liveTokenCount).toBe(2);
const transportBytes = ledger.bytesInUse - rawBytes;
// End the route. Route terminalization releases the route-owned tokens, but it
@ -263,12 +268,10 @@ describe("plugin worker manager duplex stdin transport byte ledger", () => {
exitAfterStopMs: 900,
}),
);
// Each write mixes escaped quotes and multi-byte characters near the per-write
// size. The serialized frame escapes each quote and keeps each euro sign as
// three UTF-8 bytes, so the reservation must count the encoded size.
const escaped = '"'.repeat(40_000);
const multiByte = "€".repeat(20_000);
const nearLimit = escaped + multiByte;
// Each write is near the per-write size. The JSON-RPC hop carries it as
// base64, so the reservation must count the encoded (larger) wire size, not
// the raw payload size.
const nearLimit = new Uint8Array(100_000).fill(0x22);
const routes = [routeA, routeB, routeA, routeB];
for (const route of routes) {
route.write(nearLimit);

View File

@ -64,7 +64,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
// The forged pair is an ownership violation. The host reaches no listener and
// retires the worker, so the wait settles with the fixed non-secret null exit.
await expect(session.wait()).resolves.toEqual({ exitCode: null });
@ -87,7 +90,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await expect(session.wait()).resolves.toEqual({ exitCode: null });
expect(chunks).toEqual([]);
} finally {
@ -103,8 +109,11 @@ describe("plugin worker manager duplex channel route", () => {
duplexOpenInput({ workerSessionId: "ws-A", echoInput: true }),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
session.write("callback-payload");
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
session.write(new TextEncoder().encode("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"));
@ -127,7 +136,10 @@ describe("plugin worker manager duplex channel route", () => {
// attaches. The drain then delivers them in order.
await new Promise((resolve) => setTimeout(resolve, 60));
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await vi.waitFor(() => expect(chunks.length).toBe(3));
expect(chunks).toEqual(["one", "two", "three"]);
await session.close();
@ -152,8 +164,9 @@ describe("plugin worker manager duplex channel route", () => {
// 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");
const text = new TextDecoder().decode(chunk);
chunks.push(text);
if (text === "boom") throw new Error("listener failure");
});
await expect(session.wait()).resolves.toEqual({ exitCode: 0 });
expect(chunks).toEqual(["ok-1", "boom", "ok-2"]);
@ -176,7 +189,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
// The discriminator survives the worker exit notification, so the host wait
// resolves with the transport-close mark and no exit code.
await expect(session.wait()).resolves.toEqual({ exitCode: null, transportClosed: true });
@ -205,8 +221,9 @@ describe("plugin worker manager duplex channel route", () => {
// still routes.
expect(() =>
session.onData((chunk) => {
chunks.push(chunk);
if (chunk === "boom") throw new Error("listener failure");
const text = new TextDecoder().decode(chunk);
chunks.push(text);
if (text === "boom") throw new Error("listener failure");
}),
).not.toThrow();
expect(chunks).toEqual(["one", "boom", "three"]);
@ -229,7 +246,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(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 });
@ -269,8 +289,8 @@ describe("plugin worker manager duplex channel route", () => {
);
const aChunks: string[] = [];
const bChunks: string[] = [];
routeA.onData((chunk) => aChunks.push(chunk));
routeB.onData((chunk) => bChunks.push(chunk));
routeA.onData((chunk) => aChunks.push(new TextDecoder().decode(chunk)));
routeB.onData((chunk) => bChunks.push(new TextDecoder().decode(chunk)));
await expect(routeA.wait()).resolves.toEqual({ exitCode: 0 });
await expect(routeB.wait()).resolves.toEqual({ exitCode: 0 });
// The host routes each frame by the exact pair, so each route receives only
@ -313,7 +333,7 @@ describe("plugin worker manager duplex channel route", () => {
duplexOpenInput({ workerSessionId: "ws-A", emitAfterCloseChunk: "after-close" }),
);
const chunks: string[] = [];
first.onData((chunk) => chunks.push(chunk));
first.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
// Close the route. The host installs the tombstone atomically before the slot
// frees. The worker then emits one late frame for the closed pair.
await first.close();
@ -327,7 +347,7 @@ describe("plugin worker manager duplex channel route", () => {
duplexOpenInput({ workerSessionId: "ws-B", data: [{ chunk: "b-1" }], exitCode: 0 }),
);
const secondChunks: string[] = [];
second.onData((chunk) => secondChunks.push(chunk));
second.onData((chunk) => secondChunks.push(new TextDecoder().decode(chunk)));
await expect(second.wait()).resolves.toEqual({ exitCode: 0 });
expect(secondChunks).toEqual(["b-1"]);
await second.close();
@ -394,9 +414,9 @@ describe("plugin worker manager duplex channel route", () => {
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");
session.write(new TextEncoder().encode("one"));
session.write(new TextEncoder().encode("two"));
session.write(new TextEncoder().encode("three"));
await expect(waitResult).resolves.toEqual({ exitCode: null });
} finally {
await handle.stop().catch(() => undefined);
@ -415,7 +435,7 @@ describe("plugin worker manager duplex channel route", () => {
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");
session.write(new TextEncoder().encode("this-write-is-too-large"));
await expect(waitResult).resolves.toEqual({ exitCode: null });
} finally {
await handle.stop().catch(() => undefined);
@ -463,7 +483,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(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.
@ -492,7 +515,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await expect(session.wait()).resolves.toEqual({ exitCode: null });
expect(chunks).toEqual(["€"]);
} finally {
@ -521,7 +547,10 @@ describe("plugin worker manager duplex channel route", () => {
// drop a buffered valid chunk when the route ends before a listener binds.
await expect(session.wait()).resolves.toEqual({ exitCode: null });
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
expect(chunks).toEqual(["€"]);
} finally {
await handle.stop().catch(() => undefined);
@ -581,7 +610,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(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 });
@ -612,7 +644,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await expect(session.wait()).resolves.toEqual({ exitCode: 0 });
expect(chunks).toEqual(["batched-one", "batched-two"]);
await session.close();
@ -736,7 +771,10 @@ describe("plugin worker manager duplex channel route", () => {
}),
);
const chunks: string[] = [];
session.onData((chunk) => chunks.push(chunk));
// The session streams raw `Uint8Array` chunks. Decode each one back to
// text, so the assertion below compares the plain-text payload the
// fixture directive scripted.
session.onData((chunk) => chunks.push(new TextDecoder().decode(chunk)));
await expect(session.wait()).resolves.toEqual({ exitCode: null });
expect(chunks).toEqual(["€"]);
} finally {
@ -744,6 +782,41 @@ describe("plugin worker manager duplex channel route", () => {
}
});
// -------------------------------------------------------------------------
// Byte fidelity across the worker remote-procedure-call hop.
// -------------------------------------------------------------------------
it("test_worker_channel_preserves_all_byte_values", async () => {
const handle = makeDuplexHandle();
try {
await handle.start();
const session = await handle.openDuplexChannel(
duplexOpenInput({ workerSessionId: "ws-A", echoInput: true }),
);
const allByteValues = Uint8Array.from({ length: 256 }, (_, value) => value);
const received: Uint8Array[] = [];
session.onData((chunk) => received.push(chunk));
session.write(allByteValues);
// The fixture echoes the write as one data notification, prefixed with the
// five ASCII bytes "echo:". It builds the echo on the decoded byte buffer,
// not a string, so the round trip through the base64 JSON-RPC wire form
// (`ChannelBytesWireValue`) carries every one of the 256 byte values
// unchanged, including the byte value zero, which a UTF-8 string hop would
// not preserve reliably end to end.
await vi.waitFor(() => expect(received.length).toBe(1));
const echoPrefix = new TextEncoder().encode("echo:");
const echoed = received[0]!;
expect(echoed.byteLength).toBe(echoPrefix.byteLength + allByteValues.byteLength);
expect(echoed.subarray(0, echoPrefix.byteLength)).toEqual(echoPrefix);
expect(echoed.subarray(echoPrefix.byteLength)).toEqual(allByteValues);
await session.close();
} finally {
await handle.stop().catch(() => undefined);
}
});
// -------------------------------------------------------------------------
// Authoritative closure and worker retirement.
// -------------------------------------------------------------------------

View File

@ -557,6 +557,7 @@ export async function resolveEnvironmentExecutionTarget(input: {
// (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.
// HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
...(effectiveCapabilities?.duplexCommandStream
? {
openDuplexChannel: (channelInput) =>

View File

@ -577,6 +577,8 @@ export interface EnvironmentRuntimeDriver {
* 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.
*
* HTTP/2 is the preferred transport. `queue_v1` is the soft-deprecated fallback.
*/
openDuplexChannel?(
input: EnvironmentDriverOpenDuplexChannelInput,
@ -1137,10 +1139,10 @@ function adaptDuplexChannelHostSession(
session: DuplexChannelHostSession,
): CommandManagedDuplexChannel {
return {
write(data: string): void {
write(data: Uint8Array): void {
session.write(data);
},
onData(listener: (chunk: string) => void): void {
onData(listener: (chunk: Uint8Array) => void): void {
session.onData(listener);
},
onExit(listener: (exit: { exitCode: number | null; transportClosed?: boolean }) => void): void {

View File

@ -41,6 +41,8 @@ import {
LOGIN_PTY_EXIT_NOTIFICATION,
DUPLEX_CHANNEL_DATA_NOTIFICATION,
DUPLEX_CHANNEL_EXIT_NOTIFICATION,
encodeChannelBytes,
decodeChannelBytes,
} from "@paperclipai/plugin-sdk";
import type {
JsonRpcId,
@ -610,10 +612,10 @@ export interface DuplexChannelOpenInput {
* 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;
/** Registers the one data listener. The session streams each raw byte chunk in order. */
onData(listener: (chunk: Uint8Array) => void): void;
/** Writes raw input bytes to the channel. */
write(data: string): void;
write(data: Uint8Array): void;
/**
* Resolves when the command ends or the route ends. A numeric `exitCode` is a
* real process exit. `transportClosed` is true when the provider transport
@ -913,7 +915,10 @@ export function createPluginWorkerHandle(
options.loginPtyLimits?.closeTimeoutMs ?? LOGIN_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.
// can lower them to exercise each bound and the terminalize paths. Each
// "Chars" name is a historical holdover: the channel now carries `Uint8Array`
// chunks, so every one of these bounds counts raw bytes (`.length` on a
// `Uint8Array` is its byte count), not UTF-16 code units.
const maxDuplexChannelChunkChars =
options.duplexChannelLimits?.maxChunkChars ?? MAX_DUPLEX_CHANNEL_CHUNK_CHARS;
const maxDuplexChannelPreBindChars =
@ -1630,10 +1635,10 @@ export function createPluginWorkerHandle(
// 7. route lifetime — the milliseconds from the open to the terminal end.
// One buffered data chunk retained for a late listener drain. It carries the raw
// chunk string and the aggregate byte token that reserved its raw bytes. The
// chunk bytes and the aggregate byte token that reserved its raw bytes. The
// token is `null` when no ledger is injected.
interface BufferedDuplexChunk {
chunk: string;
chunk: Uint8Array;
token: ReservationToken | null;
}
// One pre-bind data event, normalized to the narrow duplex-event schema. The host
@ -1643,7 +1648,7 @@ export function createPluginWorkerHandle(
// fails closed.
interface HeldDuplexEvent {
workerSessionId: string;
chunk: string;
chunk: Uint8Array;
token: ReservationToken | null;
}
// One pre-bind exit event, normalized to the narrow duplex-event schema.
@ -1657,7 +1662,7 @@ export function createPluginWorkerHandle(
hostRouteId: string;
state: RouteState;
workerSessionId: string | null;
listener: ((chunk: string) => void) | null;
listener: ((chunk: Uint8Array) => void) | null;
buffered: BufferedDuplexChunk[];
bufferedChars: number;
pendingRequests: number;
@ -1959,8 +1964,8 @@ export function createPluginWorkerHandle(
// 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,
listener: (chunk: Uint8Array) => void,
chunk: Uint8Array,
): void {
try {
listener(chunk);
@ -2036,25 +2041,32 @@ export function createPluginWorkerHandle(
return;
}
const route = resolved;
const chunk = params.chunk;
if (typeof chunk !== "string" || chunk.length === 0) {
// The exact pair matches, but the chunk is malformed. Count one per-route
// protocol error. Release a carried replay token first.
// The wire carries the chunk as a base64 string (JSON has no binary type; see
// `ChannelBytesWireValue` in protocol.ts). Decode it back to raw bytes before
// any bound check, so every bound below counts real bytes, not base64 text.
// A bind replay (`replayPreBindDuplexFrames`) reuses this function with a
// held event whose chunk the host already decoded once; it carries the
// already-decoded `Uint8Array` straight through with no second decode.
const rawChunk = params.chunk;
const chunk = rawChunk instanceof Uint8Array ? rawChunk : decodeChannelBytes(rawChunk);
if (chunk === null || chunk.byteLength === 0) {
// The exact pair matches, but the chunk is malformed or empty. Count one
// per-route protocol error. Release a carried replay token first.
releaseRouteToken(route, carried?.token ?? null);
recordDuplexChannelProtocolError(route);
return;
}
if (chunk.length > maxDuplexChannelChunkChars) {
if (chunk.byteLength > 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.
releaseRouteToken(route, carried?.token ?? null);
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);
// 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 = chunk.byteLength;
if (route.totalDataBytes + chunkBytes > maxDuplexChannelTotalDataBytes) {
releaseRouteToken(route, carried?.token ?? null);
void terminalizeDuplexChannelRoute(route);
@ -2178,9 +2190,10 @@ export function createPluginWorkerHandle(
return;
}
// A data event. Validate and normalize it to the narrow duplex-event schema
// before any retention.
const chunk = params.chunk;
if (typeof chunk !== "string" || chunk.length === 0) {
// before any retention. The wire carries the chunk as base64; decode it back
// to raw bytes before any bound check or retention.
const chunk = decodeChannelBytes(params.chunk);
if (chunk === null || chunk.byteLength === 0) {
recordDuplexChannelProtocolError(route);
return;
}
@ -2189,7 +2202,7 @@ export function createPluginWorkerHandle(
return;
}
// Reserve the exact retained raw byte count before the host holds the event.
const reserved = reserveRouteBytes(route, "pre_bind_event", Buffer.byteLength(chunk));
const reserved = reserveRouteBytes(route, "pre_bind_event", chunk.byteLength);
if (reserved === null) {
// The aggregate ceiling rejected the reservation. The caller retains nothing
// and the route fails closed with the fixed marker.
@ -2406,25 +2419,27 @@ export function createPluginWorkerHandle(
>(
method: M,
params: HostToWorkerMethods[M][0],
// The exact raw byte count of a `duplexChannelWrite` payload, before base64
// encoding. The caller supplies it: `params.data` on the wire is the base64
// form (`ChannelBytesWireValue`), so its string length no longer equals the
// real payload bytes. A stop request carries no payload and omits this.
writeByteCount?: number,
): void => {
if (route.state !== "open") return;
if (route.pendingRequests >= maxDuplexChannelPendingRequests) {
void terminalizeDuplexChannelRoute(route);
return;
}
// Reserve the exact UTF-8 byte count of a host→worker write against the
// Reserve the exact raw byte count of a host→worker write against the
// aggregate ledger before `callInternal` retains the payload. A pending write
// RPC holds `params.data` until it settles, so this reservation bounds the
// aggregate host→worker pending-write bytes across every route. Compute the
// byte count with `Buffer.byteLength`, not `data.length`, because one
// character can encode as several UTF-8 bytes. A stop request carries no
// payload, so it reserves nothing. When no ledger is present, admit the write
// with no token (a unit test constructs the handle this way).
// aggregate host→worker pending-write bytes across every route. A stop
// request carries no payload, so it reserves nothing. When no ledger is
// present, admit the write with no token (a unit test constructs the handle
// this way).
let pendingWriteToken: ReservationToken | null = null;
if (method === "duplexChannelWrite" && duplexAggregateByteLedger) {
const data = (params as HostToWorkerMethods["duplexChannelWrite"][0]).data;
const bytes = Buffer.byteLength(data, "utf8");
pendingWriteToken = duplexAggregateByteLedger.reserve("pending_write", bytes);
pendingWriteToken = duplexAggregateByteLedger.reserve("pending_write", writeByteCount ?? 0);
if (!pendingWriteToken) {
// The reservation would pass the aggregate ceiling. Retain nothing, do
// not enqueue the RPC, and end the route fail-closed with the aggregate
@ -2471,7 +2486,7 @@ export function createPluginWorkerHandle(
};
return {
onData(listener: (chunk: string) => void): void {
onData(listener: (chunk: Uint8Array) => void): void {
route.listener = listener;
if (route.buffered.length > 0) {
const pending = route.buffered;
@ -2489,20 +2504,28 @@ export function createPluginWorkerHandle(
terminalDuplexRoutes.delete(route);
}
},
write(data: string): void {
write(data: Uint8Array): void {
const sid = route.workerSessionId;
if (route.state !== "open" || !sid) return;
if (data.length > maxDuplexChannelWriteChars) {
if (data.byteLength > maxDuplexChannelWriteChars) {
// The write is larger than the size bound. End the route before the
// write reaches the worker.
void terminalizeDuplexChannelRoute(route);
return;
}
sendBoundedRequest("duplexChannelWrite", {
hostRouteId: route.hostRouteId,
workerSessionId: sid,
data,
});
// Encode the raw bytes to the wire-safe base64 form (JSON carries no
// binary type) and pass the exact raw byte count separately, so the
// pending-write ledger reservation charges the real payload bytes, not
// the inflated base64 string length.
sendBoundedRequest(
"duplexChannelWrite",
{
hostRouteId: route.hostRouteId,
workerSessionId: sid,
data: encodeChannelBytes(data),
},
data.byteLength,
);
},
wait(): Promise<{ exitCode: number | null }> {
return waitPromise;