fix(daytona-duplex): chunk host-to-sandbox writes and make a transport close legible (#11986)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Paperclip runs agent work through adapters and sandbox providers > - The Daytona duplex path sends host input through a provider pseudo-terminal WebSocket > - Large messages exceed the provider limit, and a transport close can look like a process exit > - This pull request chunks UTF-8 input and carries transport-close state through the duplex path > - The benefit is reliable large input and accurate loss reporting ## Linked Issues or Issue Description **What happened?** The Daytona duplex path sent a full input payload as one WebSocket message. A payload above the provider limit closed the channel. The wait path also mapped a non-numeric exit result to a process exit without exit data. **Expected behavior** The provider must receive large input as ordered UTF-8 chunks. A transport close without exit data must record `transport_closed`, while a numeric exit must record `provider_exit`. **Steps to reproduce** 1. Start a Daytona duplex session. 2. Send an input payload larger than 65536 bytes. 3. Observe that one message closes the provider channel. 4. End a session without a numeric exit code. 5. Observe that the loss reason reports a process exit. **Paperclip version or commit** Commit `1761e79ec9097c65d94f90a8ba20416f8ab718a6`. **Deployment mode** Built from source with the Daytona sandbox provider. ## What Changed - Add a shared UTF-8 byte chunker with a 32768-byte cap. - Route both Daytona pseudo-terminal write paths through the chunker. - Preserve multi-byte UTF-8 sequences across read-side chunks. - Carry an explicit `transportClosed` state through the worker and host wait paths. - Record `transport_closed` for a reason-less transport close and `provider_exit` for a numeric exit. - Keep orderly completion suppression for both exit paths. ## Verification - The Daytona plugin suite passes 194 tests. - The adapter-utils broker, codec, and telemetry suites pass 73 tests. - The plugin SDK duplex and worker RPC host suites pass 37 tests. - The server plugin worker manager duplex suite passes 78 tests. - The execution target sandbox and ACPX execute suites pass 257 tests. - TypeScript checks pass for adapter-utils, plugin SDK, server, and the standalone Daytona plugin. ## Risks The chunk size adds a loop for large input payloads. The 32768-byte cap stays below the provider limit. The optional loss field preserves compatibility for other providers. ## Model Used OpenAI Codex, GPT-5, extended reasoning, tool use, and code execution. The runtime does not expose a separate context-window value. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes: #` / `Refs: #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
141b815294
commit
c5050396c7
|
|
@ -40,8 +40,13 @@ export interface CommandManagedDuplexChannel {
|
|||
write(data: string): void;
|
||||
/** Registers the one data listener. The channel streams each raw chunk in order. */
|
||||
onData(listener: (chunk: string) => void): void;
|
||||
/** Registers the one exit listener. The channel calls it one time with the exit. */
|
||||
onExit(listener: (exit: { exitCode: number | null }) => void): void;
|
||||
/**
|
||||
* 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
|
||||
* provider transport closed with no exit data, so a reader can tell a real
|
||||
* process exit from a reason-less transport close.
|
||||
*/
|
||||
onExit(listener: (exit: { exitCode: number | null; transportClosed?: boolean }) => void): void;
|
||||
/** Stops the child process. Safe to call more than one time. */
|
||||
stop(): void;
|
||||
/** Closes the channel and releases the route. Safe to call more than one time. */
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import {
|
|||
} from "./duplex-frame-codec.js";
|
||||
import {
|
||||
createDuplexTelemetry,
|
||||
DUPLEX_COUNTER_LOSS_TOTAL,
|
||||
DUPLEX_SPAN_REQUEST,
|
||||
type DuplexTelemetryCounterRecord,
|
||||
type DuplexTelemetryEventRecord,
|
||||
|
|
@ -855,3 +856,91 @@ describe("duplex bridge broker request limits", () => {
|
|||
});
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
/**
|
||||
* A channel harness that captures the broker exit listener. `emitExit` invokes it,
|
||||
* so a test drives one channel exit into the broker.
|
||||
*/
|
||||
function createExitChannelHarness(): {
|
||||
channel: CommandManagedDuplexChannel;
|
||||
emitExit: (exit: { exitCode: number | null; transportClosed?: boolean }) => void;
|
||||
} {
|
||||
let exitListener:
|
||||
| ((exit: { exitCode: number | null; transportClosed?: boolean }) => void)
|
||||
| null = null;
|
||||
const channel: CommandManagedDuplexChannel = {
|
||||
write: () => undefined,
|
||||
onData: () => undefined,
|
||||
onExit: (listener) => {
|
||||
exitListener = listener;
|
||||
},
|
||||
stop: () => undefined,
|
||||
close: () => Promise.resolve(),
|
||||
};
|
||||
return {
|
||||
channel,
|
||||
emitExit: (exit) => {
|
||||
if (!exitListener) throw new Error("The broker did not bind the exit listener.");
|
||||
exitListener(exit);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("duplex bridge broker exit taxonomy", () => {
|
||||
const brokers: DuplexBridgeBroker[] = [];
|
||||
|
||||
afterEach(async () => {
|
||||
while (brokers.length > 0) {
|
||||
const broker = brokers.pop();
|
||||
if (broker) await broker.close();
|
||||
}
|
||||
});
|
||||
|
||||
it("records a transport close as a distinct loss, not a process exit", () => {
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
const harness = createExitChannelHarness();
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: () =>
|
||||
Promise.resolve({ status: 200, headers: {}, body: "" }),
|
||||
telemetry,
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
// A reason-less transport close carries the discriminator and no exit code.
|
||||
harness.emitExit({ exitCode: null, transportClosed: true });
|
||||
|
||||
expect(broker.lossRecord?.reason).toBe("transport_closed");
|
||||
expect(broker.runDisposition).toEqual({ failed: true, lossReason: "transport_closed" });
|
||||
const lossCounter = capture.counters.find(
|
||||
(record) => record.metric === DUPLEX_COUNTER_LOSS_TOTAL,
|
||||
);
|
||||
expect(lossCounter?.dimensions.loss_reason).toBe("transport_closed");
|
||||
});
|
||||
|
||||
it("records a numeric exit as a process exit", () => {
|
||||
const { telemetry, capture } = createTelemetryCapture();
|
||||
const harness = createExitChannelHarness();
|
||||
const broker = createDuplexBridgeBroker({
|
||||
channel: harness.channel,
|
||||
forwardRequest: () =>
|
||||
Promise.resolve({ status: 200, headers: {}, body: "" }),
|
||||
telemetry,
|
||||
});
|
||||
brokers.push(broker);
|
||||
broker.start();
|
||||
|
||||
// A numeric exit code is a real process exit, so the broker keeps the existing
|
||||
// channel_exit -> provider_exit mapping.
|
||||
harness.emitExit({ exitCode: 0 });
|
||||
|
||||
expect(broker.lossRecord?.reason).toBe("channel_exit");
|
||||
expect(broker.runDisposition).toEqual({ failed: true, lossReason: "provider_exit" });
|
||||
const lossCounter = capture.counters.find(
|
||||
(record) => record.metric === DUPLEX_COUNTER_LOSS_TOTAL,
|
||||
);
|
||||
expect(lossCounter?.dimensions.loss_reason).toBe("provider_exit");
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -66,6 +66,7 @@ export type DuplexBrokerState = "opening" | "open" | "lost" | "closing" | "close
|
|||
/** The reason the broker classified a loss. The broker records it for metrics only. */
|
||||
export type DuplexBrokerLossReason =
|
||||
| "channel_exit"
|
||||
| "transport_closed"
|
||||
| "stream_failure"
|
||||
| "protocol_failure"
|
||||
| "heartbeat_write_failure"
|
||||
|
|
@ -77,6 +78,8 @@ export type DuplexBrokerLossReason =
|
|||
* carries only the closed enum value. The map is total over the internal reasons,
|
||||
* so no raw text ever reaches the typed reason.
|
||||
* - `channel_exit` -> `provider_exit`: the provider channel process exited.
|
||||
* - `transport_closed` -> `transport_closed`: the provider transport closed with
|
||||
* no exit data, so the loss is a transport close, not a process exit.
|
||||
* - `stream_failure` -> `write_error`: a write to the channel failed.
|
||||
* - `protocol_failure` -> `rpc_failure`: a malformed or mismatched frame.
|
||||
* - `heartbeat_write_failure` -> `heartbeat_timeout`: the liveness write failed.
|
||||
|
|
@ -84,6 +87,7 @@ export type DuplexBrokerLossReason =
|
|||
*/
|
||||
const BROKER_LOSS_REASON_TO_TYPED: Readonly<Record<DuplexBrokerLossReason, DuplexLossReason>> = {
|
||||
channel_exit: "provider_exit",
|
||||
transport_closed: "transport_closed",
|
||||
stream_failure: "write_error",
|
||||
protocol_failure: "rpc_failure",
|
||||
heartbeat_write_failure: "heartbeat_timeout",
|
||||
|
|
@ -457,10 +461,12 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
if (stopped) return;
|
||||
const afterOrderlyCompletion = orderlyCompletionSeq !== null;
|
||||
// A clean channel end that orders after a host-observed orderly completion is
|
||||
// a normal teardown, not a loss. Stop cleanly, emit no loss event, and leave
|
||||
// the run a success. This keeps the closed telemetry contract: an orderly
|
||||
// close is not a loss and emits no loss event.
|
||||
if (afterOrderlyCompletion && reason === "channel_exit") {
|
||||
// a normal teardown, not a loss. A process exit and a reason-less transport
|
||||
// close both end the channel, so both count as the normal teardown here. Stop
|
||||
// cleanly, emit no loss event, and leave the run a success. This keeps the
|
||||
// closed telemetry contract: an orderly close is not a loss and emits no loss
|
||||
// event.
|
||||
if (afterOrderlyCompletion && (reason === "channel_exit" || reason === "transport_closed")) {
|
||||
stopped = true;
|
||||
clearHeartbeat();
|
||||
clearPending();
|
||||
|
|
@ -909,7 +915,14 @@ export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBr
|
|||
}
|
||||
};
|
||||
|
||||
const onExit = (): void => {
|
||||
const onExit = (exit: { exitCode: number | null; transportClosed?: boolean }): void => {
|
||||
// A reason-less transport close is not a process exit. Record it as a distinct
|
||||
// loss, so a transport close stays legible in the loss taxonomy. A real process
|
||||
// exit stays `channel_exit` -> `provider_exit`.
|
||||
if (exit.transportClosed === true) {
|
||||
recordLoss("transport_closed", "The sandbox channel transport closed with no exit.");
|
||||
return;
|
||||
}
|
||||
recordLoss("channel_exit", "The sandbox channel process exited.");
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -85,9 +85,10 @@ export type DuplexLossClass = "pre_dispatch" | "post_dispatch";
|
|||
* cause to one of these values before any sink reads it. The set covers the
|
||||
* loss-detection modes the transport names: a gateway stdin end of file, a
|
||||
* provider process exit, a heartbeat timeout, and an RPC failure. It adds
|
||||
* `write_error` for a rejected host-to-sandbox write and `other` for an unknown
|
||||
* cause. The host maps an unknown cause or any caught provider text to `other`,
|
||||
* so no raw provider text reaches a sink.
|
||||
* `write_error` for a rejected host-to-sandbox write, `transport_closed` for a
|
||||
* reason-less provider transport close with no exit data, and `other` for an
|
||||
* unknown cause. The host maps an unknown cause or any caught provider text to
|
||||
* `other`, so no raw provider text reaches a sink.
|
||||
*/
|
||||
export const DUPLEX_LOSS_REASONS = [
|
||||
"stdin_eof",
|
||||
|
|
@ -95,6 +96,7 @@ export const DUPLEX_LOSS_REASONS = [
|
|||
"heartbeat_timeout",
|
||||
"rpc_failure",
|
||||
"write_error",
|
||||
"transport_closed",
|
||||
"other",
|
||||
] as const;
|
||||
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ import {
|
|||
type DaytonaPtyHandle,
|
||||
type DaytonaPtyProcess,
|
||||
} from "./duplex-command-stream.js";
|
||||
import { PTY_INPUT_CHUNK_BYTES, PTY_MESSAGE_CAP_BYTES } from "./pty-chunked-input.js";
|
||||
|
||||
// The carriage return that submits the launch wrapper line to the terminal.
|
||||
const ENTER = "\r";
|
||||
|
|
@ -273,6 +274,8 @@ describe("openDaytonaDuplexChannelSession", () => {
|
|||
|
||||
// The launch wrapper is the first input; the host frame write is the second.
|
||||
session.write('{"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');
|
||||
// The fake terminal echoes nothing, so the write produces no data callback.
|
||||
// A real terminal in raw mode with echo off behaves the same way.
|
||||
|
|
@ -357,22 +360,104 @@ describe("openDaytonaDuplexChannelSession", () => {
|
|||
expect(received).toEqual(['{"version":1,"type":"heartbeat"}\n']);
|
||||
});
|
||||
|
||||
it("resolves wait with the gateway exit code", async () => {
|
||||
it("maps a numeric SDK exit code to a process exit", async () => {
|
||||
const process = createFakeProcess();
|
||||
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
|
||||
process.handle?.finish(3);
|
||||
|
||||
await expect(session.wait()).resolves.toEqual({ exitCode: 3 });
|
||||
// A numeric exit code is a real process exit, not a transport close.
|
||||
await expect(session.wait()).resolves.toEqual({ exitCode: 3, transportClosed: false });
|
||||
});
|
||||
|
||||
it("maps an absent exit code to null", async () => {
|
||||
it("maps an absent SDK exit code to a transport close", async () => {
|
||||
const process = createFakeProcess();
|
||||
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
|
||||
process.handle?.finish(undefined);
|
||||
|
||||
await expect(session.wait()).resolves.toEqual({ exitCode: null });
|
||||
// A non-numeric SDK result marks a reason-less transport close with no exit
|
||||
// data, so the seam reports a transport close, not a process exit.
|
||||
await expect(session.wait()).resolves.toEqual({ exitCode: null, transportClosed: true });
|
||||
});
|
||||
|
||||
it("splits a write above the provider cap into more than one send under the cap", async () => {
|
||||
const process = createFakeProcess();
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
|
||||
const handle = process.handle;
|
||||
if (!handle) throw new Error("The fake process opened no handle.");
|
||||
// The launch wrapper is the first recorded input. Count only the write's sends.
|
||||
const before = handle.inputs.length;
|
||||
|
||||
// A fixed payload well above the provider message cap. It stays fixed, so a
|
||||
// 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);
|
||||
// The chunker awaits each send, so let the write settle before the assertion.
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
const sends = handle.inputs.slice(before);
|
||||
expect(sends.length).toBeGreaterThan(1);
|
||||
for (const chunk of sends) {
|
||||
expect(Buffer.byteLength(chunk, "utf8")).toBeLessThanOrEqual(PTY_INPUT_CHUNK_BYTES);
|
||||
expect(Buffer.byteLength(chunk, "utf8")).toBeLessThanOrEqual(PTY_MESSAGE_CAP_BYTES);
|
||||
}
|
||||
// The sends rejoin to the exact payload, so the chunker loses no byte.
|
||||
expect(sends.join("")).toBe(payload);
|
||||
});
|
||||
|
||||
it("sends a write below the provider cap as exactly one send", async () => {
|
||||
const process = createFakeProcess();
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
|
||||
const handle = process.handle;
|
||||
if (!handle) throw new Error("The fake process opened no handle.");
|
||||
const before = handle.inputs.length;
|
||||
|
||||
session.write('{"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(handle.inputs.slice(before)).toHaveLength(1);
|
||||
});
|
||||
|
||||
it("keeps two writes in order and never interleaves their chunks", async () => {
|
||||
// A handle whose `sendInput` yields a microtask before it records the chunk.
|
||||
// The yield opens a window where a second write could jump ahead. The session
|
||||
// chains the writes, so the second write starts only after the first ends.
|
||||
const inputs: string[] = [];
|
||||
const handle: DaytonaPtyHandle = {
|
||||
async waitForConnection(): Promise<void> {},
|
||||
async sendInput(data: string | Uint8Array): Promise<void> {
|
||||
await Promise.resolve();
|
||||
inputs.push(typeof data === "string" ? data : new TextDecoder().decode(data));
|
||||
},
|
||||
wait(): Promise<{ exitCode?: number; error?: string }> {
|
||||
return new Promise(() => {});
|
||||
},
|
||||
async kill(): Promise<void> {},
|
||||
async disconnect(): Promise<void> {},
|
||||
};
|
||||
const process: DaytonaPtyProcess = {
|
||||
async createPty(): Promise<DaytonaPtyHandle> {
|
||||
return handle;
|
||||
},
|
||||
};
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY);
|
||||
// Two payloads above the cap, so each write is more than one chunk. "a" marks
|
||||
// the first write and "b" marks the second write.
|
||||
const first = "a".repeat(150_416);
|
||||
const second = "b".repeat(150_416);
|
||||
|
||||
session.write(first);
|
||||
session.write(second);
|
||||
// Let both writes settle.
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
// Drop the launch wrapper (the first input) and rejoin the chunks. The rejoin
|
||||
// equals the first payload then the second payload, so no chunk of the second
|
||||
// write lands between the chunks of the first write.
|
||||
expect(inputs.slice(1).join("")).toBe(`${first}${second}`);
|
||||
});
|
||||
|
||||
it("kills the child on stop and releases the socket on close", async () => {
|
||||
|
|
@ -475,6 +560,55 @@ describe("openDaytonaDuplexChannelSession", () => {
|
|||
expect(writeErrors).toEqual(["write_error"]);
|
||||
expect(killed).toBe(1);
|
||||
});
|
||||
|
||||
it("does not send a queued write after a first-send rejection terminalizes the channel", async () => {
|
||||
// Two writes queue on the chain. The launch wrapper write succeeds, then the
|
||||
// first frame write rejects and terminalizes the channel. The second queued
|
||||
// write must not reach `sendInput`, so no write runs on the closed transport.
|
||||
const sends: string[] = [];
|
||||
let killed = 0;
|
||||
const handle: DaytonaPtyHandle = {
|
||||
async waitForConnection(): Promise<void> {},
|
||||
async sendInput(data: string | Uint8Array): Promise<void> {
|
||||
const text = typeof data === "string" ? data : new TextDecoder().decode(data);
|
||||
sends.push(text);
|
||||
// The launch wrapper write (the first send) succeeds. The first frame
|
||||
// write (the second send) rejects and terminalizes the channel.
|
||||
if (sends.length === 1) return;
|
||||
throw new Error("broken");
|
||||
},
|
||||
wait(): Promise<{ exitCode?: number; error?: string }> {
|
||||
return new Promise(() => {});
|
||||
},
|
||||
async kill(): Promise<void> {
|
||||
killed += 1;
|
||||
},
|
||||
async disconnect(): Promise<void> {},
|
||||
};
|
||||
const process: DaytonaPtyProcess = {
|
||||
async createPty(): Promise<DaytonaPtyHandle> {
|
||||
return handle;
|
||||
},
|
||||
};
|
||||
const writeErrors: string[] = [];
|
||||
const session = await openDaytonaDuplexChannelSession(process, GATEWAY, {
|
||||
onWriteError: (reason) => writeErrors.push(reason),
|
||||
});
|
||||
|
||||
session.write("first\n");
|
||||
session.write("second\n");
|
||||
// Let the write chain settle both queued writes.
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
// The channel terminalized one time on the first frame write. The second
|
||||
// queued write sent no chunk, so exactly two sends ran: the launch wrapper and
|
||||
// the first frame write. The second frame never reached the transport.
|
||||
expect(writeErrors).toEqual(["write_error"]);
|
||||
expect(killed).toBe(1);
|
||||
expect(sends).toHaveLength(2);
|
||||
expect(sends.some((text) => text.includes("second"))).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createDaytonaDuplexChannelSessionOpener", () => {
|
||||
|
|
|
|||
|
|
@ -45,6 +45,8 @@ export type {
|
|||
DaytonaPtyProcess,
|
||||
} from "./setup-token-pty.js";
|
||||
|
||||
import { sendPtyInputInChunks } from "./pty-chunked-input.js";
|
||||
|
||||
/**
|
||||
* A live duplex channel session for one command. The session allocates a real
|
||||
* pseudo-terminal in raw mode, streams the raw output, accepts host input, and
|
||||
|
|
@ -56,8 +58,13 @@ export interface DuplexChannelSession {
|
|||
onData(listener: (chunk: string) => void): void;
|
||||
/** Writes raw input bytes to the pseudo-terminal. */
|
||||
write(data: string): void;
|
||||
/** Resolves with the child exit code when the command ends. */
|
||||
wait(): Promise<{ exitCode: number | null }>;
|
||||
/**
|
||||
* Resolves when the command ends or the transport closes. A numeric `exitCode`
|
||||
* is a real process exit. `transportClosed` is true when the pseudo-terminal
|
||||
* socket closed with no exit data, so the caller can tell a real process exit
|
||||
* from a reason-less transport close.
|
||||
*/
|
||||
wait(): Promise<{ exitCode: number | null; transportClosed: boolean }>;
|
||||
/** Stops the child process. Safe to call more than one time. */
|
||||
kill(): void;
|
||||
/** Releases the session resources. Safe to call more than one time. */
|
||||
|
|
@ -203,14 +210,28 @@ export async function openDaytonaDuplexChannelSession(
|
|||
// Report a host-to-sandbox write failure one time and end the channel at once.
|
||||
// The seam carries only the typed reason; the raw provider error never leaves
|
||||
// this scope. The channel end propagates the loss up through the exit.
|
||||
//
|
||||
// `channelTerminated` turns true on the first rejected chunk, before the early
|
||||
// return, so a later queued write reads it and sends no chunk. Without the flag
|
||||
// a queued write runs after the terminal kill and calls `sendInput` on the
|
||||
// closed transport.
|
||||
let channelTerminated = false;
|
||||
let writeErrorReported = false;
|
||||
const endOnWriteError = (): void => {
|
||||
channelTerminated = true;
|
||||
if (writeErrorReported) return;
|
||||
writeErrorReported = true;
|
||||
options?.onWriteError?.(DAYTONA_DUPLEX_WRITE_ERROR_REASON);
|
||||
void handle.kill().catch(() => undefined);
|
||||
};
|
||||
|
||||
// The tail of the write chain. The chunker awaits each chunk, so one write can
|
||||
// suspend between its chunks. The chain runs each write after the previous
|
||||
// write ends, so the chunks of two writes never interleave on the wire. The
|
||||
// chain never rejects, because the chunker maps a rejected send to
|
||||
// `endOnWriteError` and returns.
|
||||
let writeChain: Promise<void> = Promise.resolve();
|
||||
|
||||
return {
|
||||
onData(next: (chunk: string) => void): void {
|
||||
listener = next;
|
||||
|
|
@ -221,17 +242,38 @@ export async function openDaytonaDuplexChannelSession(
|
|||
}
|
||||
},
|
||||
write(data: string): void {
|
||||
// Fire the input write. A write error must not throw into the transport, so
|
||||
// the transport's stream stays the single result path. On a rejected write,
|
||||
// end the channel at once and report the typed reason. The raw provider
|
||||
// error never reaches a sink.
|
||||
void handle.sendInput(data).catch(() => {
|
||||
endOnWriteError();
|
||||
// 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
|
||||
// runs this write after the previous write ends, so two writes keep their
|
||||
// order and never interleave their chunks. A write error must not throw into
|
||||
// the transport, so the transport's stream stays the single result path. On
|
||||
// a rejected chunk, the chunker ends the channel one time through
|
||||
// `endOnWriteError` and sends no later chunk. The raw provider error never
|
||||
// reaches a sink.
|
||||
//
|
||||
// A rejected chunk terminalizes the channel and kills the transport. So this
|
||||
// write skips the send when `channelTerminated` is true, and a queued write
|
||||
// never calls `sendInput` on the closed transport.
|
||||
writeChain = writeChain.then(() => {
|
||||
if (channelTerminated) return;
|
||||
return sendPtyInputInChunks(
|
||||
(chunk) => handle.sendInput(chunk),
|
||||
data,
|
||||
endOnWriteError,
|
||||
);
|
||||
});
|
||||
},
|
||||
async wait(): Promise<{ exitCode: number | null }> {
|
||||
async wait(): Promise<{ exitCode: number | null; transportClosed: boolean }> {
|
||||
const result = await handle.wait();
|
||||
return { exitCode: typeof result.exitCode === "number" ? result.exitCode : null };
|
||||
if (typeof result.exitCode === "number") {
|
||||
// A numeric exit code is a real process exit. The SDK parses it from the
|
||||
// pseudo-terminal WebSocket close reason (for example `{"exitCode":0}`).
|
||||
return { exitCode: result.exitCode, transportClosed: false };
|
||||
}
|
||||
// A non-numeric exit code marks a reason-less transport close: the socket
|
||||
// closed with no exit data. It is a transport close, not a process exit.
|
||||
return { exitCode: null, transportClosed: true };
|
||||
},
|
||||
kill(): void {
|
||||
void handle.kill().catch(() => undefined);
|
||||
|
|
|
|||
|
|
@ -2828,9 +2828,19 @@ const plugin = definePlugin({
|
|||
session.onData((chunk) => {
|
||||
pluginContext?.duplexChannel.data(entry.hostRouteId, workerSessionId, chunk);
|
||||
});
|
||||
// Forward the child exit one time. The host resolves the open route on it.
|
||||
// Forward the child exit one time. The host resolves the open route on it. A
|
||||
// numeric exit code is a real process exit; a resolved transport close carries
|
||||
// `transportClosed`, so the host keeps the two apart in the loss taxonomy. A
|
||||
// rejected wait is not a resolved transport close, so it reports no code and no
|
||||
// transport-close mark.
|
||||
void session.wait().then(
|
||||
(result) => pluginContext?.duplexChannel.exit(entry.hostRouteId, workerSessionId, result.exitCode),
|
||||
(result) =>
|
||||
pluginContext?.duplexChannel.exit(
|
||||
entry.hostRouteId,
|
||||
workerSessionId,
|
||||
result.exitCode,
|
||||
result.transportClosed,
|
||||
),
|
||||
() => pluginContext?.duplexChannel.exit(entry.hostRouteId, workerSessionId, null),
|
||||
);
|
||||
// Echo the host route id on the reply, so the host binds the exact pair.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,138 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
PTY_INPUT_CHUNK_BYTES,
|
||||
PTY_MESSAGE_CAP_BYTES,
|
||||
sendPtyInputInChunks,
|
||||
} from "./pty-chunked-input.js";
|
||||
|
||||
/**
|
||||
* A fake `sendInput`. It records each raw byte chunk, so a test asserts the chunk
|
||||
* count, the chunk sizes, and the rejoined bytes. It copies each chunk, because
|
||||
* the chunker passes a `subarray` view of one shared byte array.
|
||||
*/
|
||||
function createByteRecorder(): {
|
||||
sendInput: (chunk: Uint8Array) => Promise<void>;
|
||||
chunks: Uint8Array[];
|
||||
} {
|
||||
const chunks: Uint8Array[] = [];
|
||||
return {
|
||||
chunks,
|
||||
async sendInput(chunk: Uint8Array): Promise<void> {
|
||||
chunks.push(Uint8Array.from(chunk));
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Concatenate the recorded chunks into one byte array. */
|
||||
function concatChunks(chunks: readonly Uint8Array[]): Uint8Array {
|
||||
const total = chunks.reduce((sum, chunk) => sum + chunk.length, 0);
|
||||
const joined = new Uint8Array(total);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
joined.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
return joined;
|
||||
}
|
||||
|
||||
describe("sendPtyInputInChunks", () => {
|
||||
it("sends a payload below the cap as exactly one send", async () => {
|
||||
const recorder = createByteRecorder();
|
||||
|
||||
await sendPtyInputInChunks(recorder.sendInput, "a small frame\n");
|
||||
|
||||
expect(recorder.chunks).toHaveLength(1);
|
||||
expect(recorder.chunks[0]?.length).toBeLessThanOrEqual(PTY_MESSAGE_CAP_BYTES);
|
||||
});
|
||||
|
||||
it("splits a payload above the cap into more than one send, each at or below the cap", async () => {
|
||||
const recorder = createByteRecorder();
|
||||
// A fixed payload well above the provider message cap. It stays fixed, so a
|
||||
// raised chunk size above this size makes the payload one send and fails this
|
||||
// test.
|
||||
const payload = new Uint8Array(150_416).fill(65);
|
||||
|
||||
await sendPtyInputInChunks(recorder.sendInput, payload);
|
||||
|
||||
expect(recorder.chunks.length).toBeGreaterThan(1);
|
||||
for (const chunk of recorder.chunks) {
|
||||
expect(chunk.length).toBeLessThanOrEqual(PTY_INPUT_CHUNK_BYTES);
|
||||
expect(chunk.length).toBeLessThanOrEqual(PTY_MESSAGE_CAP_BYTES);
|
||||
}
|
||||
expect(concatChunks(recorder.chunks)).toEqual(payload);
|
||||
});
|
||||
|
||||
it("slices a multi-byte character across a boundary by bytes and rejoins it", async () => {
|
||||
const recorder = createByteRecorder();
|
||||
// A fixed payload: one 1-byte "x", then the 3-byte character "€" repeated. It
|
||||
// stays fixed, so a raised chunk size above this size makes the payload one
|
||||
// send and fails this test. The chunk size is a multiple of 3, so a pure "€"
|
||||
// run would put every boundary on a character edge. The 1-byte prefix shifts
|
||||
// the "€" run by one byte, so the first chunk boundary lands inside one "€"
|
||||
// sequence.
|
||||
const text = `x${"€".repeat(30_000)}`;
|
||||
const inputBytes = new TextEncoder().encode(text);
|
||||
expect(inputBytes.length).toBe(90_001);
|
||||
|
||||
await sendPtyInputInChunks(recorder.sendInput, text);
|
||||
|
||||
expect(recorder.chunks.length).toBeGreaterThan(1);
|
||||
// The first chunk holds exactly the byte cap and ends inside a "€" sequence,
|
||||
// because the 1-byte prefix shifts the 3-byte run off the cap boundary. A
|
||||
// character-index slice could never end mid-sequence, so the slice is by bytes.
|
||||
expect(recorder.chunks[0]?.length).toBe(PTY_INPUT_CHUNK_BYTES);
|
||||
// The rejoined bytes equal the input bytes, so a split multi-byte sequence is
|
||||
// whole again. A streaming decoder on the read side then decodes it.
|
||||
const joined = concatChunks(recorder.chunks);
|
||||
expect(joined).toEqual(inputBytes);
|
||||
expect(new TextDecoder("utf-8").decode(joined)).toBe(text);
|
||||
});
|
||||
|
||||
it("resolves each send before it starts the next send", async () => {
|
||||
let active = 0;
|
||||
let maxActive = 0;
|
||||
const sendCount = { value: 0 };
|
||||
// A `sendInput` that stays active across a microtask. The awaited loop starts
|
||||
// the next send only after the previous send resolves, so at most one send is
|
||||
// active at a time. A synchronous burst would start every send at once.
|
||||
const sendInput = async (): Promise<void> => {
|
||||
active += 1;
|
||||
maxActive = Math.max(maxActive, active);
|
||||
await Promise.resolve();
|
||||
sendCount.value += 1;
|
||||
active -= 1;
|
||||
};
|
||||
// Three chunks of the byte cap plus a short tail.
|
||||
const payload = new Uint8Array(PTY_INPUT_CHUNK_BYTES * 3 + 10).fill(65);
|
||||
|
||||
await sendPtyInputInChunks(sendInput, payload);
|
||||
|
||||
expect(sendCount.value).toBe(4);
|
||||
expect(maxActive).toBe(1);
|
||||
});
|
||||
|
||||
it("stops sending later chunks after a chunk rejects", async () => {
|
||||
const chunks: Uint8Array[] = [];
|
||||
let errorCount = 0;
|
||||
// A `sendInput` that records the chunk, then rejects on a later microtask. The
|
||||
// rejection is asynchronous, so a synchronous burst could not stop the later
|
||||
// sends. The awaited loop learns the rejection before the next send, so it
|
||||
// sends one chunk only.
|
||||
const sendInput = async (chunk: Uint8Array): Promise<void> => {
|
||||
chunks.push(chunk);
|
||||
await Promise.resolve();
|
||||
throw new Error("RAW-PROVIDER-BROKEN-PIPE");
|
||||
};
|
||||
|
||||
// A payload above the cap, so the chunker would fire more than one chunk with
|
||||
// no fail-fast. The loop stops after the first rejected chunk.
|
||||
await sendPtyInputInChunks(sendInput, new Uint8Array(150_416).fill(65), () => {
|
||||
errorCount += 1;
|
||||
});
|
||||
|
||||
// The loop sent one chunk, learned the rejection, and started no later send.
|
||||
expect(chunks).toHaveLength(1);
|
||||
// The error handler ran one time for the whole payload.
|
||||
expect(errorCount).toBe(1);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,80 @@
|
|||
// The shared host-to-sandbox write chunker for the Daytona pseudo-terminal (PTY)
|
||||
// sessions. The provider PTY WebSocket rejects any single message over 65536
|
||||
// bytes with WebSocket close code 1009 and closes the connection. `sendInput`
|
||||
// sends a whole payload in one message, so a payload above the cap takes the
|
||||
// channel down. Real duplex frames cross the cap (single frames of 127582 and
|
||||
// 150416 bytes are observed). This module slices each payload into byte-bounded
|
||||
// chunks under the cap and sends each chunk in order.
|
||||
|
||||
/**
|
||||
* The provider PTY WebSocket message cap, in bytes. The provider rejects a single
|
||||
* PTY message over this size with WebSocket close code 1009 and closes the
|
||||
* connection.
|
||||
*/
|
||||
export const PTY_MESSAGE_CAP_BYTES = 65536;
|
||||
|
||||
/**
|
||||
* The maximum byte length of one host-to-sandbox chunk. It holds about 5 KiB of
|
||||
* headroom under {@link PTY_MESSAGE_CAP_BYTES}, so a chunk never sits on the cap
|
||||
* boundary. A later change in how the provider counts message overhead does not
|
||||
* push one chunk over the cap and back into the close-code-1009 failure. The
|
||||
* headroom costs about nine percent more messages for a large payload.
|
||||
*
|
||||
* The chunker slices bytes, not string characters, so a chunk boundary can split
|
||||
* a multi-byte UTF-8 sequence. The read-side streaming `TextDecoder` rejoins a
|
||||
* split sequence, so a byte slice is safe. A string-index slice could split a
|
||||
* surrogate pair, so the chunker never slices by character.
|
||||
*/
|
||||
export const PTY_INPUT_CHUNK_BYTES = 60 * 1024;
|
||||
|
||||
/** The single UTF-8 encoder the chunker reuses for every string payload. */
|
||||
const PTY_INPUT_TEXT_ENCODER = new TextEncoder();
|
||||
|
||||
/**
|
||||
* Send one host-to-sandbox payload as byte-bounded chunks under the provider
|
||||
* message cap. The function encodes a string payload to UTF-8 bytes, slices the
|
||||
* byte array into chunks of at most {@link PTY_INPUT_CHUNK_BYTES}, and calls
|
||||
* `sendInput` for each chunk in order.
|
||||
*
|
||||
* The loop awaits each chunk before it sends the next chunk. The `await` keeps
|
||||
* the wire order equal to the slice order, and it makes the chunk send fail-fast:
|
||||
* after a chunk rejects, the function sends no further chunk of this payload. A
|
||||
* synchronous loop could not fail-fast, because `sendInput` reaches `ws.send`
|
||||
* before its first `await`, so a synchronous loop puts every chunk on the wire
|
||||
* before the first rejection settles. The awaited loop learns the rejection
|
||||
* before it sends the next chunk, so it stops.
|
||||
*
|
||||
* The function does not overlap two calls for the same channel. The caller
|
||||
* serializes its writes (for example, through a promise chain), so the awaited
|
||||
* loop of one payload never interleaves its chunks with the chunks of the next
|
||||
* payload.
|
||||
*
|
||||
* `onError` runs at most one time. The function reports the first rejected chunk
|
||||
* and returns. A rejected send never throws out of this function, so the caller's
|
||||
* stream stays the single result path.
|
||||
*
|
||||
* @param sendInput - Sends one raw byte chunk to the pseudo-terminal.
|
||||
* @param payload - The host-to-sandbox payload, as a string or raw bytes.
|
||||
* @param onError - Optional. The function calls it one time on the first rejected chunk.
|
||||
*/
|
||||
export async function sendPtyInputInChunks(
|
||||
sendInput: (chunk: Uint8Array) => Promise<void>,
|
||||
payload: string | Uint8Array,
|
||||
onError?: () => void,
|
||||
): Promise<void> {
|
||||
const bytes = typeof payload === "string" ? PTY_INPUT_TEXT_ENCODER.encode(payload) : payload;
|
||||
for (let offset = 0; offset < bytes.length; offset += PTY_INPUT_CHUNK_BYTES) {
|
||||
const chunk = bytes.subarray(offset, offset + PTY_INPUT_CHUNK_BYTES);
|
||||
try {
|
||||
// Await the send before the next chunk. The await keeps the wire order and
|
||||
// lets the loop learn a rejection before it starts a later send.
|
||||
await sendInput(chunk);
|
||||
} catch {
|
||||
// Report the first rejected chunk one time and stop. The function starts no
|
||||
// later send for this payload, so the wire keeps the ordered, fail-fast
|
||||
// behavior. The raw provider error never leaves this function.
|
||||
onError?.();
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -27,6 +27,8 @@
|
|||
|
||||
import { randomUUID } from "node:crypto";
|
||||
|
||||
import { sendPtyInputInChunks } from "./pty-chunked-input.js";
|
||||
|
||||
/**
|
||||
* A live pseudo-terminal session for one setup-token login command. The session
|
||||
* allocates a real pseudo-terminal, streams the raw terminal output, accepts
|
||||
|
|
@ -151,6 +153,11 @@ export async function openDaytonaSetupTokenPtySession(
|
|||
// to the command, not to a shell.
|
||||
await handle.sendInput(`exec ${command}${PTY_COMMAND_TERMINATOR}`);
|
||||
|
||||
// The tail of the write chain. The chunker awaits each chunk, so one write can
|
||||
// suspend between its chunks. The chain runs each write after the previous
|
||||
// write ends, so the chunks of two writes never interleave on the wire.
|
||||
let writeChain: Promise<void> = Promise.resolve();
|
||||
|
||||
return {
|
||||
onData(next: (chunk: string) => void): void {
|
||||
listener = next;
|
||||
|
|
@ -161,9 +168,15 @@ export async function openDaytonaSetupTokenPtySession(
|
|||
}
|
||||
},
|
||||
write(data: string): void {
|
||||
// Fire the input write. A write error must not throw into the runner, so
|
||||
// Send the input as byte-bounded chunks under the provider message cap. The
|
||||
// login input is short today, so the latent defect never fires, but the same
|
||||
// unbounded write goes through the one chunker. The chain runs this write
|
||||
// after the previous write ends, so two writes keep their order and never
|
||||
// interleave their chunks. A write error must not throw into the runner, so
|
||||
// the runner's fixed status stays the single result path.
|
||||
void handle.sendInput(data).catch(() => undefined);
|
||||
writeChain = writeChain.then(() =>
|
||||
sendPtyInputInChunks((chunk) => handle.sendInput(chunk), data),
|
||||
);
|
||||
},
|
||||
async wait(): Promise<{ exitCode: number | null }> {
|
||||
const result = await handle.wait();
|
||||
|
|
|
|||
|
|
@ -1191,6 +1191,12 @@ export interface PluginDuplexChannelExitParams {
|
|||
workerSessionId: string;
|
||||
/** The child exit code, or null when the child ended with no code. */
|
||||
exitCode: number | null;
|
||||
/**
|
||||
* True when the provider transport closed with no exit data, so the exit is a
|
||||
* reason-less transport close, not a process exit. Absent or false marks a real
|
||||
* process exit. The host maps a transport close to a distinct loss reason.
|
||||
*/
|
||||
transportClosed?: boolean;
|
||||
}
|
||||
|
||||
/** The worker→host notification method for one duplex channel data chunk. */
|
||||
|
|
|
|||
|
|
@ -2065,8 +2065,14 @@ 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 exitCode - The child exit code, or null when the child ended with no code.
|
||||
* @param transportClosed - True when the transport closed with no exit data, so the exit is a reason-less transport close, not a process exit. Absent marks a real process exit.
|
||||
*/
|
||||
exit(hostRouteId: string, workerSessionId: string, exitCode: number | null): void;
|
||||
exit(
|
||||
hostRouteId: string,
|
||||
workerSessionId: string,
|
||||
exitCode: number | null,
|
||||
transportClosed?: boolean,
|
||||
): void;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -1415,16 +1415,23 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost
|
|||
if (typeof chunk !== "string" || chunk.length === 0) return;
|
||||
notifyHost(DUPLEX_CHANNEL_DATA_NOTIFICATION, { hostRouteId, workerSessionId, chunk });
|
||||
},
|
||||
exit(hostRouteId: string, workerSessionId: string, exitCode: number | null): void {
|
||||
exit(
|
||||
hostRouteId: string,
|
||||
workerSessionId: string,
|
||||
exitCode: number | null,
|
||||
transportClosed?: boolean,
|
||||
): void {
|
||||
// Forward the child exit of a persistent duplex channel. The host
|
||||
// resolves the open route's wait promise by the exact live pair while
|
||||
// the route is open.
|
||||
// the route is open. `transportClosed` carries the exit discriminator, so
|
||||
// the host tells a real process exit from a reason-less transport close.
|
||||
if (typeof hostRouteId !== "string" || hostRouteId.length === 0) return;
|
||||
if (typeof workerSessionId !== "string" || workerSessionId.length === 0) return;
|
||||
notifyHost(DUPLEX_CHANNEL_EXIT_NOTIFICATION, {
|
||||
hostRouteId,
|
||||
workerSessionId,
|
||||
exitCode: typeof exitCode === "number" ? exitCode : null,
|
||||
transportClosed: transportClosed === true,
|
||||
});
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -366,6 +366,7 @@ key never reaches a sink by accident.
|
|||
| `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. |
|
||||
| `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. |
|
||||
|
||||
To add a name or an enum value, extend the literal constant in
|
||||
`duplex-telemetry.ts` first, then update the test that asserts the closed set.
|
||||
|
|
|
|||
|
|
@ -14,6 +14,8 @@
|
|||
// `sid` or `rid` to prove the host drops a mismatched-pair notification and
|
||||
// counts a protocol error.
|
||||
// - `exitCode`: when set, the fixture emits an exit notification after the data.
|
||||
// - `transportClosed`: when true, the exit notification marks a reason-less
|
||||
// transport close and carries no exit code.
|
||||
// - `dataAfterExit`: an array of `{ chunk, sid? }`, same shape as `data`. The
|
||||
// fixture emits these as data notifications after the exit notification, so a
|
||||
// test can script a batch where a data frame arrives after the exit in the
|
||||
|
|
@ -52,11 +54,19 @@ function scriptedFrameLines(directive, hostRouteId, workerSessionId) {
|
|||
},
|
||||
})}\n`;
|
||||
}
|
||||
if (typeof directive.exitCode === "number") {
|
||||
if (typeof directive.exitCode === "number" || directive.transportClosed === true) {
|
||||
const exitParams = {
|
||||
hostRouteId,
|
||||
workerSessionId,
|
||||
exitCode: typeof directive.exitCode === "number" ? directive.exitCode : null,
|
||||
};
|
||||
// A transport-close directive marks a reason-less transport close, so the
|
||||
// fixture carries the discriminator the same way a real worker does.
|
||||
if (directive.transportClosed === true) exitParams.transportClosed = true;
|
||||
lines += `${JSON.stringify({
|
||||
jsonrpc: "2.0",
|
||||
method: "duplexChannel.exit",
|
||||
params: { hostRouteId, workerSessionId, exitCode: directive.exitCode },
|
||||
params: exitParams,
|
||||
})}\n`;
|
||||
}
|
||||
const dataAfterExit = Array.isArray(directive.dataAfterExit) ? directive.dataAfterExit : [];
|
||||
|
|
|
|||
|
|
@ -163,6 +163,30 @@ describe("plugin worker manager duplex channel route", () => {
|
|||
}
|
||||
});
|
||||
|
||||
it("carries a transport-close exit through to the wait result", async () => {
|
||||
const handle = makeDuplexHandle();
|
||||
try {
|
||||
await handle.start();
|
||||
const session = await handle.openDuplexChannel(
|
||||
duplexOpenInput({
|
||||
workerSessionId: "ws-A",
|
||||
data: [{ chunk: "one" }],
|
||||
// The worker reports a reason-less transport close with no exit code.
|
||||
transportClosed: true,
|
||||
}),
|
||||
);
|
||||
const chunks: string[] = [];
|
||||
session.onData((chunk) => chunks.push(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 });
|
||||
expect(chunks).toEqual(["one"]);
|
||||
await session.close();
|
||||
} finally {
|
||||
await handle.stop().catch(() => undefined);
|
||||
}
|
||||
});
|
||||
|
||||
it("isolates a throwing listener during the buffered replay so every buffered chunk routes", async () => {
|
||||
const handle = makeDuplexHandle();
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -1080,9 +1080,10 @@ function adaptDuplexChannelHostSession(
|
|||
onData(listener: (chunk: string) => void): void {
|
||||
session.onData(listener);
|
||||
},
|
||||
onExit(listener: (exit: { exitCode: number | null }) => void): void {
|
||||
onExit(listener: (exit: { exitCode: number | null; transportClosed?: boolean }) => void): void {
|
||||
// `wait()` resolves one time with the exit and never rejects, so a single
|
||||
// `then` bridges it to the one-time exit listener.
|
||||
// `then` bridges it to the one-time exit listener. The exit carries
|
||||
// `transportClosed`, so the broker tells a real exit from a transport close.
|
||||
void session.wait().then((exit) => {
|
||||
listener(exit);
|
||||
});
|
||||
|
|
|
|||
|
|
@ -572,8 +572,13 @@ export interface DuplexChannelHostSession {
|
|||
onData(listener: (chunk: string) => void): void;
|
||||
/** Writes raw input bytes to the channel. */
|
||||
write(data: string): void;
|
||||
/** Resolves with the child exit code when the command ends or the route ends. */
|
||||
wait(): Promise<{ exitCode: number | null }>;
|
||||
/**
|
||||
* Resolves when the command ends or the route ends. A numeric `exitCode` is a
|
||||
* real process exit. `transportClosed` is true when the provider transport
|
||||
* closed with no exit data, so the broker can tell a real process exit from a
|
||||
* reason-less transport close.
|
||||
*/
|
||||
wait(): Promise<{ exitCode: number | null; transportClosed?: boolean }>;
|
||||
/** Stops the child process. Safe to call more than one time. */
|
||||
kill(): void;
|
||||
/** Closes the route and releases the channel. Safe to call more than one time. */
|
||||
|
|
@ -1280,8 +1285,8 @@ export function createPluginWorkerHandle(
|
|||
// Settle the route wait exactly once. Replace the settler with a no-op, so a
|
||||
// later exit or terminalize never settles the wait a second time.
|
||||
function settleRouteWait(
|
||||
route: { settleWait: (value: { exitCode: number | null }) => void },
|
||||
value: { exitCode: number | null },
|
||||
route: { settleWait: (value: { exitCode: number | null; transportClosed?: boolean }) => void },
|
||||
value: { exitCode: number | null; transportClosed?: boolean },
|
||||
): void {
|
||||
const settle = route.settleWait;
|
||||
route.settleWait = () => {};
|
||||
|
|
@ -1557,7 +1562,7 @@ export function createPluginWorkerHandle(
|
|||
totalDataBytes: number;
|
||||
lifetimeTimer: ReturnType<typeof setTimeout> | null;
|
||||
terminalized: boolean;
|
||||
settleWait: (value: { exitCode: number | null }) => void;
|
||||
settleWait: (value: { exitCode: number | null; transportClosed?: boolean }) => void;
|
||||
// The data frames that arrived before the bind, held in order. The bind
|
||||
// replays them through the exact-pair routing, so an early frame is never
|
||||
// lost and a frame whose pair does not match the bound pair still fails
|
||||
|
|
@ -1869,6 +1874,14 @@ export function createPluginWorkerHandle(
|
|||
return;
|
||||
}
|
||||
const exitCode = typeof params.exitCode === "number" ? params.exitCode : null;
|
||||
// A reason-less transport close carries `transportClosed`; a real process exit
|
||||
// does not. Carry the discriminator to the wait only when it is a transport
|
||||
// close, so a real exit keeps the plain `{ exitCode }` result and the broker
|
||||
// still keeps the two apart in the loss taxonomy.
|
||||
if (params.transportClosed === true) {
|
||||
settleRouteWait(resolved, { exitCode, transportClosed: true });
|
||||
return;
|
||||
}
|
||||
settleRouteWait(resolved, { exitCode });
|
||||
}
|
||||
|
||||
|
|
@ -1953,10 +1966,13 @@ export function createPluginWorkerHandle(
|
|||
input: DuplexChannelOpenInput,
|
||||
): Promise<DuplexChannelHostSession> {
|
||||
const hostRouteId = nextDuplexHostRouteId();
|
||||
let settleWait: (value: { exitCode: number | null }) => void = () => {};
|
||||
const waitPromise = new Promise<{ exitCode: number | null }>((resolve) => {
|
||||
settleWait = resolve;
|
||||
});
|
||||
let settleWait: (value: { exitCode: number | null; transportClosed?: boolean }) => void =
|
||||
() => {};
|
||||
const waitPromise = new Promise<{ exitCode: number | null; transportClosed?: boolean }>(
|
||||
(resolve) => {
|
||||
settleWait = resolve;
|
||||
},
|
||||
);
|
||||
const route: DuplexChannelRoute = {
|
||||
hostRouteId,
|
||||
state: "reserved",
|
||||
|
|
|
|||
Loading…
Reference in New Issue