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 <noreply@paperclip.ing>
This commit is contained in:
Nicky Leach
GitHub
parent
1fc4591327
commit
3e28d64a72
|
|
@ -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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
},
|
||||
]);
|
||||
|
||||
|
|
|
|||
|
|
@ -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<string, unknown> = {
|
||||
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);
|
||||
}
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
Loading…
Reference in New Issue