From 3e28d64a72849b4ba6e716144a051f32a5fde994 Mon Sep 17 00:00:00 2001 From: Nicky Leach Date: Tue, 25 Aug 2026 13:42:13 -0700 Subject: [PATCH] fix(plugin-worker-manager): queue and replay pre-bind login pseudo-terminal frames (#12173) ## Thinking Path > - Paperclip routes plugin worker messages to agent sessions. > - The login pseudo-terminal route opens after the host receives the open reply. > - `readline` can deliver later frames from the same pipe read before that reply continuation runs. > - The host dropped early output and exit frames. > - The fix queues valid early frames, preserves arrival order, and replays them after the route opens. > - The route uses bounded memory and closes fail-closed when a bound breaks. > - The final tests also pin child issue ordering so the serialized suite remains deterministic. ## Linked Issues or Issue Description Fixes #12122 ## What Changed - Add a bounded queue for login pseudo-terminal output and exit frames during route opening. - Validate session ids, chunk types, and per-chunk limits before queue insertion. - Bound the queue by 10,000 frames and 8 MiB of characters. - Charge retained worker session identifiers against the character bound. - Preserve arrival order and stop replay after the first valid exit. - Drop repeated exits without changing the first exit position or code. - Bound the repeat-exit lookup and clear queued state on all terminal paths. - Add regression tests and fixture support for coalesced frames, ordering, limits, cleanup, and log safety. - Pin issue numbers in the child-wake test so its expected child order remains deterministic. ## Verification - Build the plugin SDK with `pnpm --filter @paperclipai/plugin-sdk build`. - Run `npx vitest run server/src/__tests__/plugin-worker-manager.test.ts` from the repository root. - Run `npx vitest run server/src/__tests__/issues-service.test.ts` from the repository root. - The focused plugin worker suite passes 66 of 66 tests at the prior reviewed head. - The issue service file passes 120 of 120 tests in two isolated runs at the current head. - Confirm that GitHub Actions passes all required checks. - Confirm that Greptile reports 5/5 with no unresolved review threads. - Storybook visual regression remains skipped because the PR has no `storybook-visual` label. ## Risks - The queue adds bounded memory use while the login pseudo-terminal route opens. - A queue limit breach closes the route and prevents unbounded buffering. - A hostile worker can fail only its own login route when it breaches a bound. - The first valid exit closes the route, so later records do not reach the session. - The child-wake test now uses distinct issue numbers to match the service sort contract. ## Model Used OpenAI Codex, GPT-5, extended reasoning, tool use, and code review support. 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 that 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 linked the existing public issue with `Fixes: #12122` - [x] I have not referenced internal Paperclip issues or links - [x] My branch name describes the change and contains no internal ticket id - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation where needed - [x] I have considered and documented risks above - [x] All required Paperclip CI gates are green - [x] Greptile is 5/5 with no unresolved review threads - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip --- .../fixtures/plugin-worker-login-pty.cjs | 140 ++++- server/src/__tests__/issues-service.test.ts | 8 + .../__tests__/plugin-worker-manager.test.ts | 543 ++++++++++++++++++ server/src/services/plugin-worker-manager.ts | 159 ++++- 4 files changed, 819 insertions(+), 31 deletions(-) diff --git a/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs b/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs index 24ecd95e7d..27d9f1c1e5 100644 --- a/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs +++ b/server/src/__tests__/fixtures/plugin-worker-login-pty.cjs @@ -6,22 +6,103 @@ // The manager allowlists `command` to the fixed `CLAUDE_SETUP_TOKEN_COMMAND`. The // test encodes a JSON directive in the forwarded `providerLeaseId`, so one fixture // serves every route-gate case: -// - `mode`: "normal" | "malformed-open" | "no-open-reply" | "duplicate-open-reply" +// - `mode`: "normal" | "malformed-open" | "no-open-reply" | "duplicate-open-reply" | +// "exit-before-open-reply" // - `workerSessionId`: the worker session id the open reply returns (default "ws-1") // - `outputs`: an array of `{ chunk, sid? }`. The fixture emits each as an output // notification after the open reply. `sid` defaults to the real worker session // id; a test sets a wrong `sid` to prove the host drops a mismatched // notification. // - `exitCode`: when set, the fixture emits an exit notification after the outputs. +// - `extraExits`: an array of `{ exitCode, sid? }`. The fixture emits each as a +// further exit notification, after the main `exitCode` exit. `sid` defaults +// to the real worker session id; a test sets a wrong `sid` to script a worker +// that sends a valid exit, then a mismatched exit, before the bind. +// - `outputsAfterExit`: an array of `{ chunk, sid? }`, same shape as `outputs`. +// The fixture emits these after the exit notification, so a test proves the +// host drops output that arrives behind a queued exit. +// - `sequence`: an array of `{ type: "output", chunk, sid? }` or +// `{ type: "exit", exitCode, sid? }` entries. When set, the fixture emits +// exactly this sequence, in order, instead of the fixed +// outputs/exitCode/extraExits/outputsAfterExit composition. A test uses +// this to script an arrival order the fixed composition cannot express, +// for example a mismatched exit that arrives before a valid one. // - `closeMode`: "ack" | "bad-ack" | "no-ack" (default "ack"). It controls the // close reply, so a test proves the host retires the worker on an unconfirmed // close. +// - `batchWithOpenReply`: when true, the fixture writes the open reply and the +// scripted outputs and exit in one stdout write. The host then reads the open +// reply and the notifications in one batch, so a test proves the host queues +// and replays a record that arrives before the route binds. +// - `mode: "exit-before-open-reply"`: the fixture emits the scripted outputs, +// then exits with no open reply, so a test proves the host clears the +// pre-bind queue on a worker exit during the open window. const readline = require("node:readline"); function send(message) { process.stdout.write(`${JSON.stringify(message)}\n`); } +// Serialize one array of `{ chunk, sid? }` entries as newline-delimited output +// notification lines. A test sets a wrong `sid` on one entry to force a +// mismatch. +function outputLines(entries, workerSessionId) { + let lines = ""; + for (const entry of entries) { + lines += `${JSON.stringify({ + jsonrpc: "2.0", + method: "loginPty.output", + params: { + workerSessionId: entry.sid ?? workerSessionId, + chunk: entry.chunk, + }, + })}\n`; + } + return lines; +} + +// Serialize one exit notification line for the given worker session id. +function exitLine(workerSessionId, exitCode) { + return `${JSON.stringify({ + jsonrpc: "2.0", + method: "loginPty.exit", + params: { workerSessionId, exitCode }, + })}\n`; +} + +// Serialize the scripted output and exit notifications as newline-delimited +// lines. The `sequence` directive, when set, emits exactly that order. The +// fixed composition otherwise emits, in order: the pre-exit outputs, then +// the main exit, then every extra exit, then the post-exit outputs. The +// batch mode writes these together with the open reply in one stdout write. +function scriptedOutputLines(directive, workerSessionId) { + if (Array.isArray(directive.sequence)) { + let lines = ""; + for (const entry of directive.sequence) { + if (entry.type === "output") { + lines += outputLines([{ chunk: entry.chunk, sid: entry.sid }], workerSessionId); + } else if (entry.type === "exit") { + lines += exitLine(entry.sid ?? workerSessionId, entry.exitCode); + } + } + return lines; + } + const outputs = Array.isArray(directive.outputs) ? directive.outputs : []; + const outputsAfterExit = Array.isArray(directive.outputsAfterExit) + ? directive.outputsAfterExit + : []; + const extraExits = Array.isArray(directive.extraExits) ? directive.extraExits : []; + let lines = outputLines(outputs, workerSessionId); + if (typeof directive.exitCode === "number") { + lines += exitLine(workerSessionId, directive.exitCode); + } + for (const exit of extraExits) { + lines += exitLine(exit.sid ?? workerSessionId, exit.exitCode); + } + lines += outputLines(outputsAfterExit, workerSessionId); + return lines; +} + // The registered terminals, keyed by the host route id. Each entry records the // bound worker session id and the close directive. const routes = new Map(); @@ -71,41 +152,48 @@ rl.on("line", (line) => { // Never reply, so the host open call times out. return; } - if (mode === "malformed-open") { - // Reply with no worker session id, so the host terminalizes the route. - send({ jsonrpc: "2.0", id: message.id, result: {} }); + if (mode === "exit-before-open-reply") { + // Emit the scripted pre-bind outputs, then exit with no open reply. The + // route never binds, so a test proves the worker-exit path clears the + // pre-bind queue. + process.stdout.write(scriptedOutputLines(directive, workerSessionId)); + process.exit(1); return; } - const reply = () => - send({ jsonrpc: "2.0", id: message.id, result: { workerSessionId } }); - reply(); + // A malformed reply carries no worker session id, so the host cannot bind + // and terminalizes the route. + const isMalformedOpen = mode === "malformed-open"; + const openReplyLine = `${JSON.stringify({ + jsonrpc: "2.0", + id: message.id, + result: isMalformedOpen ? {} : { workerSessionId }, + })}\n`; + + if (directive.batchWithOpenReply === true) { + // Write the open reply and the scripted outputs and exit in one stdout + // write. The host reads them in one batch, so an output or an exit + // notification arrives before the route binds — even a malformed reply + // that never binds. The host must queue and, on a bind, replay it; on a + // malformed reply, it must clear the queue instead. + process.stdout.write(openReplyLine + scriptedOutputLines(directive, workerSessionId)); + return; + } + + process.stdout.write(openReplyLine); if (mode === "duplicate-open-reply") { // Send a second open reply for the same request id. The host drops it. - reply(); + process.stdout.write(openReplyLine); + } + if (isMalformedOpen) { + // No scripted output follows a non-batched malformed reply. + return; } // Emit the scripted output and the exit after the open reply, so the host // binds the route first. setImmediate(() => { - const outputs = Array.isArray(directive.outputs) ? directive.outputs : []; - for (const entry of outputs) { - send({ - jsonrpc: "2.0", - method: "loginPty.output", - params: { - workerSessionId: entry.sid ?? workerSessionId, - chunk: entry.chunk, - }, - }); - } - if (typeof directive.exitCode === "number") { - send({ - jsonrpc: "2.0", - method: "loginPty.exit", - params: { workerSessionId, exitCode: directive.exitCode }, - }); - } + process.stdout.write(scriptedOutputLines(directive, workerSessionId)); }); return; } diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index ee0632e673..f2b3bbb347 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -4590,6 +4590,13 @@ describeEmbeddedPostgres("issueService blockers and dependency wake readiness", title: "Child A", status: "done", priority: "medium", + // Give the children distinct, ordered issue numbers. The service + // sorts direct children by issueNumber, then createdAt. A batched + // insert gives every row in the statement the same defaultNow() + // createdAt, so without a distinct issueNumber the two children + // tie on both sort keys and the database is free to return them + // in either order. + issueNumber: 1, }, { id: childB, @@ -4598,6 +4605,7 @@ describeEmbeddedPostgres("issueService blockers and dependency wake readiness", title: "Child B", status: "blocked", priority: "medium", + issueNumber: 2, }, ]); diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index e34ad8d86d..6dba39b371 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -9,6 +9,24 @@ import { type HostServices, type HostToWorkerMethods, } from "@paperclipai/plugin-sdk"; + +// Mock the shared logger, so a test reads the exact calls the manager makes +// when it logs a route event. The child logger returns the same mock object, +// so `log.warn`/`log.error` inside the manager are this mock's `warn`/`error`. +vi.mock("../middleware/logger.js", () => { + const mockLogger: Record = { + trace: vi.fn(), + debug: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + error: vi.fn(), + fatal: vi.fn(), + child: vi.fn(() => mockLogger), + }; + return { logger: mockLogger, httpLogger: vi.fn() }; +}); + +import { logger } from "../middleware/logger.js"; import { appendStderrExcerpt, createPluginWorkerHandle, @@ -1345,3 +1363,528 @@ describe("plugin worker manager setup-token pty route gate", () => { } }); }); + +// --------------------------------------------------------------------------- +// Login pseudo-terminal pre-bind queue (GH #12122 / PAP-5132) +// --------------------------------------------------------------------------- +// The host reads the worker pipe and `readline` dispatches every line of one +// chunk synchronously. The route only becomes `open` inside the `await` +// continuation of the `loginPtyOpen` reply, which runs as a microtask after +// the whole synchronous line loop. So a worker that batches an output or an +// exit notification with the open reply floods the host before the bind. The +// tests below prove the host queues these pre-bind records and replays them +// through the live router right after the bind, instead of dropping them. +// The host holds every pre-bind record — output and exit alike — in one +// arrival-ordered queue, and it replays each record through the live router +// in that exact order. A replayed exit that carries the bound worker session +// identifier settles the route, so a record that arrived behind it — a real +// worker process cannot emit output after it exits, so this only matters for +// a forged or a queued record — replays into a route that already settled +// and is dropped. + +describe("plugin worker manager login pseudo-terminal pre-bind queue", () => { + it("queues and replays a coalesced output notification that arrives before the bind", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The fixture writes the open reply and the output notification in one + // stdout write, so the host reads both before the route binds. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + outputs: [{ chunk: "batched-output" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // The bind already replayed the queued record into `buffered`, so + // `onData` drains it synchronously with no wait. + expect(chunks).toEqual(["batched-output"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("delivers a coalesced output before the coalesced exit settles the wait", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + outputs: [{ chunk: "batched-output" }], + exitCode: 0, + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // Both the output and the exit arrived before the bind and queued in + // order. The replay preserves that order, so the output reaches the + // listener before the wait settles. + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual(["batched-output"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("settles the wait with a valid pre-bind exit that a later mismatched exit cannot displace", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The worker batches a valid exit for the real worker session id, then a + // second exit for a forged worker session id, both before the open + // reply. The bind still verifies the real session id, so the held valid + // exit settles the wait, and the mismatched exit that arrived after it + // never displaces it. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + exitCode: 0, + extraExits: [{ exitCode: 1, sid: "ws-EVIL" }], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("delivers a pre-bind output, then the exit, and drops output that arrives behind the exit", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The worker batches an output, then a valid exit, then a further + // output, all before the open reply. The host replays the three held + // records in this exact arrival order. The exit settles the route + // right after the first output, so the record behind the exit finds a + // route that already settled and never reaches the listener. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + outputs: [{ chunk: "before-exit" }], + exitCode: 0, + outputsAfterExit: [{ chunk: "after-exit" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual(["before-exit"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops output that arrives behind a valid pre-bind exit", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + exitCode: 0, + // The worker batched this output behind the exit. The replay sends + // the exit first, in arrival order, which settles the route. The + // output record behind it then finds a route that is no longer + // `open`, the same drop the live path applies to output a real + // worker process could never emit after its own exit. + outputsAfterExit: [{ chunk: "late-output" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual([]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("settles the wait with the valid exit code and drops output behind it, even past the total-chars bound", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxTotalChars: 10, maxPreBindFrames: 1000, maxPreBindChars: 1000 }, + }); + try { + await handle.start(); + // The worker batches a valid exit with the open reply, then an output + // record that would push the cumulative delivered total past the + // 10-character bound. The replay sends the exit first, in arrival + // order. The exit settles the route, so the replay drops the output + // record behind it before the total-chars check ever runs — the + // bound violation the check exists to catch never reaches it, because + // a real worker process could never emit that output in the first + // place. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + exitCode: 0, + outputsAfterExit: [{ chunk: "aaaaaaaaaaaa" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual([]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("delivers output that arrived behind a mismatched pre-bind exit and settles with the later valid exit", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The worker batches a mismatched exit, then a genuine output, then + // the valid exit, all before the open reply. The mismatched exit + // arrives first, but the exact-match gate fails it, so it changes no + // state and the replay continues. The output that follows it still + // reaches the listener, and the valid exit that follows the output + // settles the wait. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + sequence: [ + { type: "exit", exitCode: 1, sid: "ws-EVIL" }, + { type: "output", chunk: "genuine-output" }, + { type: "exit", exitCode: 0 }, + ], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: 0 }); + expect(chunks).toEqual(["genuine-output"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("drops a forged pre-bind worker session id sent before the valid bind", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + outputs: [ + { chunk: "forged", sid: "ws-EVIL" }, + { chunk: "good" }, + ], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // Both records queued before the bind, since the host cannot yet check + // a worker session id against a route that has not bound one. The + // replay applies the exact-match gate against the real bind ("ws-A"), + // so the forged record never reaches the listener. + expect(chunks).toEqual(["good"]); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("terminalizes the route when batched pre-bind records pass the frame-count bound", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindFrames: 2 }, + }); + try { + await handle.start(); + // The worker batches three output notifications with the open reply, so + // all three arrive, and queue, before the bind. The third record passes + // the frame-count bound, so the host terminalizes the route before the + // bind can complete, and the open call itself fails. + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + outputs: [{ chunk: "a" }, { chunk: "b" }, { chunk: "c" }], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("terminalizes the route when batched pre-bind output passes the character bound", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindChars: 10 }, + }); + try { + await handle.start(); + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + // 5 + 5 = 10 admits; the third record brings the queued total to + // 15, past the 10-character bound, so the host terminalizes the + // route before the bind can complete. + outputs: [{ chunk: "aaaaa" }, { chunk: "bbbbb" }, { chunk: "ccccc" }], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("charges the retained worker session id characters on the pre-bind output path", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindChars: 10 }, + }); + try { + await handle.start(); + // A 6-character worker session id and one 5-character chunk charge 11 + // characters against the 10-character bound on the very first record, + // even though the chunk alone is under the bound. Without the identifier + // charge, this single record would pass. + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ABCDEF", + outputs: [{ chunk: "aaaaa" }], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("terminalizes the route when batched pre-bind exit notifications with a large worker session id pass the character bound", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindChars: 10 }, + }); + try { + await handle.start(); + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + // An exit record carries no chunk, but it still retains the worker + // session id. This 5-character id makes 5 + 5 = 10 admit the first + // two exits; the third brings the queued total to 15, past the + // 10-character bound, so the host terminalizes the route before + // the bind can complete. + workerSessionId: "AAAAA", + sequence: [ + { type: "exit", exitCode: 1 }, + { type: "exit", exitCode: 2 }, + { type: "exit", exitCode: 3 }, + ], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("settles with the first exit code against a repeated pre-bind exit for the same session, behind a filled output queue", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindFrames: 50 }, + }); + try { + await handle.start(); + // Fill the pre-bind queue with 40 output records, then repeat the exit + // for the same worker session id after more output arrives. The first + // exit settles the route during the replay, so the repeat exit and the + // output around it never reach the listener. + const fillerOutputs = Array.from({ length: 40 }, (_, index) => ({ + type: "output" as const, + chunk: `filler-${index}`, + })); + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + sequence: [ + ...fillerOutputs, + { type: "exit", exitCode: 1 }, + { type: "output", chunk: "between-exits" }, + { type: "exit", exitCode: 2 }, + { type: "output", chunk: "after-repeat" }, + ], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + // The wait settles with the FIRST exit code, and the exit drops every + // output that arrived behind it, so neither "between-exits" nor + // "after-repeat" reaches the listener. + await expect(session.wait()).resolves.toEqual({ exitCode: 1 }); + expect(chunks).toEqual(fillerOutputs.map((entry) => entry.chunk)); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("settles with the first exit code against N repeated pre-bind exits for the same session", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The worker sends five repeat exits for the same session before the + // bind. The first exit settles the route during the replay, so every + // repeat drops there and the wait still settles with the first code. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + workerSessionId: "ws-A", + exitCode: 1, + extraExits: [ + { exitCode: 2 }, + { exitCode: 3 }, + { exitCode: 4 }, + { exitCode: 5 }, + ], + }), + ); + await expect(session.wait()).resolves.toEqual({ exitCode: 1 }); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("clears the pre-bind queue on a malformed open reply", async () => { + const handle = makeLoginPtyHandle(); + try { + await handle.start(); + // The malformed reply carries no worker session id, so the host cannot + // bind. It terminalizes the route and clears the queued output before + // it ever reaches a listener. + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + mode: "malformed-open", + batchWithOpenReply: true, + outputs: [{ chunk: "leaked" }], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + // A later open on the same worker starts a fresh route and receives + // only its own scripted output, never the cleared queue. + const session = await handle.openLoginPtySession( + ptyOpenInput({ mode: "normal", outputs: [{ chunk: "fresh" }] }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await vi.waitFor(() => expect(chunks).toContain("fresh")); + expect(chunks).not.toContain("leaked"); + await session.close(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("clears the pre-bind queue when the worker exits during the open window", async () => { + const handle = makeLoginPtyHandle({ autoRestart: false }); + try { + await handle.start(); + // The fixture emits one output notification, then exits before it ever + // sends the open reply. The route never binds. The worker-exit path + // must clear the queued output along with the route. + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + mode: "exit-before-open-reply", + outputs: [{ chunk: "queued-before-exit" }], + }), + ), + ).rejects.toThrow(); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("terminalizes at replay when the queued output would pass the cumulative total-chars bound", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxTotalChars: 10, maxPreBindFrames: 1000, maxPreBindChars: 1000 }, + }); + try { + await handle.start(); + // The generous pre-bind bounds admit all three records at intake, so + // the bind completes and the replay runs. The replay sends each record + // through the same live router an open route uses, so the cumulative + // `maxTotalChars` gate still applies: the third record would bring the + // delivered total to 15, past the 10-character bound, so the replay + // terminalizes the route partway through. The directive carries no exit + // notification, so only a mid-replay terminalize can settle the wait; + // without the cumulative gate applying during replay, this call would + // hang. + const session = await handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + outputs: [{ chunk: "aaaaa" }, { chunk: "bbbbb" }, { chunk: "ccccc" }], + }), + ); + const chunks: string[] = []; + session.onData((chunk) => chunks.push(chunk)); + await expect(session.wait()).resolves.toEqual({ exitCode: null }); + // The terminalize clears `route.buffered` by design (a terminalized + // login route settles with a null exit code, so buffered data has no + // consumer), so even the two records the replay delivered before the + // bound tripped never reach the listener. + expect(chunks).toEqual([]); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("never logs the raw chunk content, including on the pre-bind overflow path", async () => { + const handle = makeLoginPtyHandle({ + loginPtyLimits: { maxPreBindFrames: 1 }, + }); + const secretMarker = "super-secret-login-code-must-never-reach-a-log-line"; + vi.mocked(logger.warn).mockClear(); + vi.mocked(logger.error).mockClear(); + vi.mocked(logger.info).mockClear(); + vi.mocked(logger.debug).mockClear(); + try { + await handle.start(); + // The first record admits, then the second passes the frame-count + // bound and terminalizes the route. The overflow path logs a fixed + // warning that must never carry the chunk text. + await expect( + handle.openLoginPtySession( + ptyOpenInput({ + batchWithOpenReply: true, + outputs: [{ chunk: secretMarker }, { chunk: "overflow" }], + }), + ), + ).rejects.toThrow("LOGIN_PTY_OPEN_FAILED"); + const loggedText = [ + ...vi.mocked(logger.warn).mock.calls, + ...vi.mocked(logger.error).mock.calls, + ...vi.mocked(logger.info).mock.calls, + ...vi.mocked(logger.debug).mock.calls, + ] + .flat() + .map((arg) => JSON.stringify(arg)) + .join("\n"); + expect(loggedText).not.toContain(secretMarker); + } finally { + await handle.stop().catch(() => undefined); + } + }); +}); diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 43b7c820b4..0b533a3385 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -169,6 +169,10 @@ const MAX_EXECUTE_LOG_TOTAL_CHARS = 128 * 1024 * 1024; const MAX_LOGIN_PTY_CHUNK_CHARS = 1_000_000; /** Maximum cumulative output characters for one login pseudo-terminal route. */ const MAX_LOGIN_PTY_TOTAL_CHARS = 8 * 1024 * 1024; +/** Default maximum output and exit records the host queues before the bind. */ +const MAX_LOGIN_PTY_PRE_BIND_FRAMES = 10_000; +/** Default maximum cumulative output characters the host queues before the bind. */ +const MAX_LOGIN_PTY_PRE_BIND_CHARS = 8 * 1024 * 1024; /** The default open timeout for one login pseudo-terminal route, in milliseconds. */ const LOGIN_PTY_OPEN_TIMEOUT_MS = 30_000; /** The default close timeout for one login pseudo-terminal route, in milliseconds. */ @@ -472,6 +476,10 @@ export interface WorkerStartOptions { maxChunkChars?: number; /** Max cumulative output characters for one login pseudo-terminal route. */ maxTotalChars?: number; + /** Max number of pre-bind output and exit records the host queues before the bind. */ + maxPreBindFrames?: number; + /** Max cumulative output characters the host queues before the bind. */ + maxPreBindChars?: number; /** The open timeout for one login pseudo-terminal route, in milliseconds. */ openTimeoutMs?: number; /** The close timeout for one login pseudo-terminal route, in milliseconds. */ @@ -909,6 +917,10 @@ export function createPluginWorkerHandle( options.loginPtyLimits?.maxChunkChars ?? MAX_LOGIN_PTY_CHUNK_CHARS; const maxLoginPtyTotalChars = options.loginPtyLimits?.maxTotalChars ?? MAX_LOGIN_PTY_TOTAL_CHARS; + const maxLoginPtyPreBindFrames = + options.loginPtyLimits?.maxPreBindFrames ?? MAX_LOGIN_PTY_PRE_BIND_FRAMES; + const maxLoginPtyPreBindChars = + options.loginPtyLimits?.maxPreBindChars ?? MAX_LOGIN_PTY_PRE_BIND_CHARS; const loginPtyOpenTimeoutMs = options.loginPtyLimits?.openTimeoutMs ?? LOGIN_PTY_OPEN_TIMEOUT_MS; const loginPtyCloseTimeoutMs = @@ -1389,6 +1401,23 @@ export function createPluginWorkerHandle( } type LoginPtyRouteState = RouteState; + + // One pre-bind login pseudo-terminal record: an output chunk or an exit, + // normalized to a narrow scalar shape (never the raw notification object). + interface LoginPtyPreBindOutputRecord { + kind: "output"; + workerSessionId: string; + chunk: string; + } + + interface LoginPtyPreBindExitRecord { + kind: "exit"; + workerSessionId: string; + exitCode: number | null; + } + + type LoginPtyPreBindRecord = LoginPtyPreBindOutputRecord | LoginPtyPreBindExitRecord; + interface LoginPtyRoute { hostRouteId: string; state: LoginPtyRouteState; @@ -1398,6 +1427,12 @@ export function createPluginWorkerHandle( deliveredChars: number; terminalized: boolean; settleWait: (value: { exitCode: number | null }) => void; + // Every output and exit record that arrived before the bind, in arrival order. + preBind: LoginPtyPreBindRecord[]; + // The cumulative characters `preBind` holds. Each record charges its + // `workerSessionId` characters, plus the `chunk` characters for an output + // record. + preBindChars: number; } // At most one active credential pseudo-terminal per worker. A non-null route // blocks a second open until the manager confirms the first route's close. @@ -1430,6 +1465,9 @@ export function createPluginWorkerHandle( route.state = "closed"; route.listener = null; route.buffered = []; + // A terminalized route never replays a queued pre-bind record. + route.preBind = []; + route.preBindChars = 0; // A terminalized route reports a null exit code, which the runner treats as a // failure. settleRouteWait(route, { exitCode: null }); @@ -1448,11 +1486,17 @@ export function createPluginWorkerHandle( // Route one login pseudo-terminal output notification to the per-session // listener. Deliver only while the route is `open` and the notification carries - // the exact bound worker session identifier and valid bounded bytes. Drop an - // unknown, late, malformed, or mismatched notification. Never log the raw bytes. + // the exact bound worker session identifier and valid bounded bytes. Queue the + // notification while the route is still `opening`. Drop an unknown, late, + // malformed, or mismatched notification. Never log the raw bytes. function routeLoginPtyOutput(notification: JsonRpcNotification): void { const route = loginPtyRoute; - if (!route || route.state !== "open") return; + if (!route) return; + if (route.state === "opening") { + queuePreBindLoginPtyOutput(route, notification); + return; + } + if (route.state !== "open") return; const params = isRecord(notification.params) ? notification.params : {}; const workerSessionId = readNonEmptyString(params.workerSessionId); if (!workerSessionId || workerSessionId !== route.workerSessionId) return; @@ -1476,17 +1520,116 @@ export function createPluginWorkerHandle( // Route one login pseudo-terminal exit notification to the login wait. Resolve // only while the route is `open` and the notification carries the exact bound - // worker session identifier. + // worker session identifier. Queue the notification while the route is still + // `opening`. A resolved exit moves the state off `open`, so a later record — + // live or replayed — finds a closed route and drops there. function routeLoginPtyExit(notification: JsonRpcNotification): void { const route = loginPtyRoute; - if (!route || route.state !== "open") return; + if (!route) return; + if (route.state === "opening") { + queuePreBindLoginPtyExit(route, notification); + return; + } + if (route.state !== "open") return; const params = isRecord(notification.params) ? notification.params : {}; const workerSessionId = readNonEmptyString(params.workerSessionId); if (!workerSessionId || workerSessionId !== route.workerSessionId) return; const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; + route.state = "closed"; settleRouteWait(route, { exitCode }); } + // Queue one login pseudo-terminal output notification that arrived before the + // bind, in arrival order. Terminalize the route fail-closed on a frame-count + // or character-bound breach. Never log the raw chunk. + function queuePreBindLoginPtyOutput( + route: LoginPtyRoute, + notification: JsonRpcNotification, + ): void { + const params = isRecord(notification.params) ? notification.params : {}; + const workerSessionId = readNonEmptyString(params.workerSessionId); + const chunk = params.chunk; + if ( + !workerSessionId || + typeof chunk !== "string" || + chunk.length === 0 || + chunk.length > maxLoginPtyChunkChars + ) { + return; + } + // Charge the record's full retained size: the worker session identifier + // plus the chunk. A record that retains only the identifier still counts. + const recordChars = workerSessionId.length + chunk.length; + if ( + route.preBind.length + 1 > maxLoginPtyPreBindFrames || + route.preBindChars + recordChars > maxLoginPtyPreBindChars + ) { + log.warn( + { pluginId }, + "login pseudo-terminal pre-bind queue exceeded a bound; terminalizing route", + ); + void terminalizeLoginPtyRoute(route); + return; + } + route.preBind.push({ kind: "output", workerSessionId, chunk }); + route.preBindChars += recordChars; + } + + // Queue one login pseudo-terminal exit notification that arrived before the + // bind, in the same queue the output records use. An exit record carries no + // chunk, but it still retains a worker session identifier, so it charges + // against the character bound too. + function queuePreBindLoginPtyExit( + route: LoginPtyRoute, + notification: JsonRpcNotification, + ): void { + const params = isRecord(notification.params) ? notification.params : {}; + const workerSessionId = readNonEmptyString(params.workerSessionId); + if (!workerSessionId) return; + const exitCode = typeof params.exitCode === "number" ? params.exitCode : null; + const recordChars = workerSessionId.length; + if ( + route.preBind.length + 1 > maxLoginPtyPreBindFrames || + route.preBindChars + recordChars > maxLoginPtyPreBindChars + ) { + log.warn( + { pluginId }, + "login pseudo-terminal pre-bind queue exceeded a bound; terminalizing route", + ); + void terminalizeLoginPtyRoute(route); + return; + } + route.preBind.push({ kind: "exit", workerSessionId, exitCode }); + route.preBindChars += recordChars; + } + + // Replay the records a route held before it bound, in arrival order, + // through the same live router the open route uses (`routeLoginPtyOutput`, + // `routeLoginPtyExit`), so a forged worker session identifier still fails + // the exact-match gate and a replayed exit still closes the route and + // drops every record behind it. + function replayPreBindLoginPtyRecords(route: LoginPtyRoute): void { + const held = route.preBind; + route.preBind = []; + route.preBindChars = 0; + for (const record of held) { + if (route.terminalized) return; + if (record.kind === "output") { + routeLoginPtyOutput({ + jsonrpc: "2.0", + method: LOGIN_PTY_OUTPUT_NOTIFICATION, + params: { workerSessionId: record.workerSessionId, chunk: record.chunk }, + }); + } else { + routeLoginPtyExit({ + jsonrpc: "2.0", + method: LOGIN_PTY_EXIT_NOTIFICATION, + params: { workerSessionId: record.workerSessionId, exitCode: record.exitCode }, + }); + } + } + } + // Close the one route on a worker exit. The worker is gone, so the manager // resolves the login wait with the fixed non-secret exit and clears the route // one time. The pending pseudo-terminal calls reject through `rejectAllPending`. @@ -1498,6 +1641,8 @@ export function createPluginWorkerHandle( route.state = "closed"; route.listener = null; route.buffered = []; + route.preBind = []; + route.preBindChars = 0; settleRouteWait(route, { exitCode: null }); } @@ -1538,6 +1683,8 @@ export function createPluginWorkerHandle( deliveredChars: 0, terminalized: false, settleWait, + preBind: [], + preBindChars: 0, }; loginPtyRoute = route; @@ -1574,6 +1721,8 @@ export function createPluginWorkerHandle( // Bind the worker session identifier one time and move the route to `open`. route.workerSessionId = workerSessionId; route.state = "open"; + // Replay every record the route queued before the bind, in arrival order. + replayPreBindLoginPtyRecords(route); return { onData(listener: (chunk: string) => void): void {