From c5050396c7998fd29dbb6fcf16b2b3e9ae2771ea Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Sat, 22 Aug 2026 16:35:47 -0700 Subject: [PATCH] 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 --- .../src/command-managed-runtime.ts | 9 +- .../src/duplex-bridge-broker.test.ts | 89 +++++++++++ .../adapter-utils/src/duplex-bridge-broker.ts | 23 ++- .../adapter-utils/src/duplex-telemetry.ts | 8 +- .../daytona/src/duplex-command-stream.test.ts | 142 +++++++++++++++++- .../daytona/src/duplex-command-stream.ts | 62 ++++++-- .../sandbox-providers/daytona/src/plugin.ts | 14 +- .../daytona/src/pty-chunked-input.test.ts | 138 +++++++++++++++++ .../daytona/src/pty-chunked-input.ts | 80 ++++++++++ .../daytona/src/setup-token-pty.ts | 17 ++- packages/plugins/sdk/src/protocol.ts | 6 + packages/plugins/sdk/src/types.ts | 8 +- packages/plugins/sdk/src/worker-rpc-host.ts | 11 +- packages/shared/src/telemetry/README.md | 1 + .../fixtures/plugin-worker-duplex-channel.cjs | 14 +- .../plugin-worker-manager-duplex.test.ts | 24 +++ server/src/services/environment-runtime.ts | 5 +- server/src/services/plugin-worker-manager.ts | 34 +++-- 18 files changed, 641 insertions(+), 44 deletions(-) create mode 100644 packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.test.ts create mode 100644 packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.ts diff --git a/packages/adapter-utils/src/command-managed-runtime.ts b/packages/adapter-utils/src/command-managed-runtime.ts index 4c879dbf4d..1121d8365d 100644 --- a/packages/adapter-utils/src/command-managed-runtime.ts +++ b/packages/adapter-utils/src/command-managed-runtime.ts @@ -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. */ diff --git a/packages/adapter-utils/src/duplex-bridge-broker.test.ts b/packages/adapter-utils/src/duplex-bridge-broker.test.ts index fccddb3b82..6c87a1ae87 100644 --- a/packages/adapter-utils/src/duplex-bridge-broker.test.ts +++ b/packages/adapter-utils/src/duplex-bridge-broker.test.ts @@ -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"); + }); +}); diff --git a/packages/adapter-utils/src/duplex-bridge-broker.ts b/packages/adapter-utils/src/duplex-bridge-broker.ts index 43d7b524f3..1a902360a9 100644 --- a/packages/adapter-utils/src/duplex-bridge-broker.ts +++ b/packages/adapter-utils/src/duplex-bridge-broker.ts @@ -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> = { 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."); }; diff --git a/packages/adapter-utils/src/duplex-telemetry.ts b/packages/adapter-utils/src/duplex-telemetry.ts index 4bc08abebf..2bcf8073cb 100644 --- a/packages/adapter-utils/src/duplex-telemetry.ts +++ b/packages/adapter-utils/src/duplex-telemetry.ts @@ -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; diff --git a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts index 1610a3d83e..19339783f5 100644 --- a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts +++ b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.test.ts @@ -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 {}, + async sendInput(data: string | Uint8Array): Promise { + 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 {}, + async disconnect(): Promise {}, + }; + const process: DaytonaPtyProcess = { + async createPty(): Promise { + 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 {}, + async sendInput(data: string | Uint8Array): Promise { + 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 { + killed += 1; + }, + async disconnect(): Promise {}, + }; + const process: DaytonaPtyProcess = { + async createPty(): Promise { + 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", () => { diff --git a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts index 2aba7b76c8..4507ba774e 100644 --- a/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts +++ b/packages/plugins/sandbox-providers/daytona/src/duplex-command-stream.ts @@ -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 = 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); diff --git a/packages/plugins/sandbox-providers/daytona/src/plugin.ts b/packages/plugins/sandbox-providers/daytona/src/plugin.ts index 304947e562..88f9f06032 100644 --- a/packages/plugins/sandbox-providers/daytona/src/plugin.ts +++ b/packages/plugins/sandbox-providers/daytona/src/plugin.ts @@ -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. diff --git a/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.test.ts b/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.test.ts new file mode 100644 index 0000000000..cc1fbbbce5 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.test.ts @@ -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; + chunks: Uint8Array[]; +} { + const chunks: Uint8Array[] = []; + return { + chunks, + async sendInput(chunk: Uint8Array): Promise { + 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 => { + 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 => { + 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); + }); +}); diff --git a/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.ts b/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.ts new file mode 100644 index 0000000000..dd9ee7e627 --- /dev/null +++ b/packages/plugins/sandbox-providers/daytona/src/pty-chunked-input.ts @@ -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, + payload: string | Uint8Array, + onError?: () => void, +): Promise { + 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; + } + } +} diff --git a/packages/plugins/sandbox-providers/daytona/src/setup-token-pty.ts b/packages/plugins/sandbox-providers/daytona/src/setup-token-pty.ts index 89bace5b5c..769e442a19 100644 --- a/packages/plugins/sandbox-providers/daytona/src/setup-token-pty.ts +++ b/packages/plugins/sandbox-providers/daytona/src/setup-token-pty.ts @@ -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 = 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(); diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 4c98b41088..be1c579256 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -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. */ diff --git a/packages/plugins/sdk/src/types.ts b/packages/plugins/sdk/src/types.ts index 8ea10b964e..412054dc77 100644 --- a/packages/plugins/sdk/src/types.ts +++ b/packages/plugins/sdk/src/types.ts @@ -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; } // --------------------------------------------------------------------------- diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index 901410a841..82311ff504 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -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, }); }, }, diff --git a/packages/shared/src/telemetry/README.md b/packages/shared/src/telemetry/README.md index 159501a8dc..bba168e5cb 100644 --- a/packages/shared/src/telemetry/README.md +++ b/packages/shared/src/telemetry/README.md @@ -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. diff --git a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs index 914df984dd..013ecb0362 100644 --- a/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-duplex-channel.cjs @@ -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 : []; diff --git a/server/src/__tests__/plugin-worker-manager-duplex.test.ts b/server/src/__tests__/plugin-worker-manager-duplex.test.ts index 85bc5f3dff..0b746e8bd0 100644 --- a/server/src/__tests__/plugin-worker-manager-duplex.test.ts +++ b/server/src/__tests__/plugin-worker-manager-duplex.test.ts @@ -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 { diff --git a/server/src/services/environment-runtime.ts b/server/src/services/environment-runtime.ts index 6cf5cf33f2..9187a59aa8 100644 --- a/server/src/services/environment-runtime.ts +++ b/server/src/services/environment-runtime.ts @@ -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); }); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 76dea17f0c..77ac1ff63a 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -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 | 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 { 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",