fix(coding-agent): repair corrupt REPL protocol frames (#1839)
* refactor(coding-agent): unify kernel shutdown * docs(coding-agent): carry over shutdown rationale comments * fix(coding-agent): repair corrupt REPL protocol * fix(coding-agent): cancel the final kernel snapshot before disposal cleanup * fix(coding-agent): block new executions from splicing ahead of the final dispose snapshot flushSnapshotForDispose captured the execution queue while the kernel state was still 'running', so a concurrent execute() (which has no per-cell timeout) could enqueue after the capture; captureSnapshot then queued behind it and its timeout never started, hanging dispose() and shutdown({snapshot:true}) forever. Claim the flush before capturing the queue: non-internal requests are rejected while the final snapshot is flushing, keeping every wait on the dispose path bounded. * fix(coding-agent): close protocol-repair races and bound the repair - Gate queued requests behind an in-flight repair: a request whose queue slot opens during a repair releases the slot and requeues after the repair, so queued executes/snapshots can no longer run on the empty replacement kernel ahead of restoreState (and overwrite its result or clobber the last good snapshot). - Race the repair wait in execute() with opts.signal so an aborted call returns its aborted result promptly instead of pending through repair. - Bound the repair's restore with REPAIR_RESTORE_TIMEOUT_MS so a stalled replacement cannot wedge protocolRepairPromise (and every later execute) forever; a timed-out restore discards the replacement and returns to clean idle. - Stop respawn-looping when the replacement kernel itself emits a corrupt frame during repair: discard it and settle at clean idle instead of superseding into an unbounded chain of repairs. * fix(coding-agent): re-bootstrap repaired kernels and gate busy-waiting cells - A protocol-repaired kernel restored only the user namespace, so live handles (rlm, bash, skills) were missing and the next cell failed with NameErrors. KernelManagerOptions gains bootstrapCode (the provisioner passes its buildRlmBootstrapCode output); the repair re-runs it after restore, bounded like the restore, and discards the replacement to clean idle when it fails. - The repair requeue gate only ran right after taking the queue slot, so a cell busy-waiting on an interrupted execution when corruption arrived could proceed onto the unrestored replacement while blocking the repair's restore behind its slot. The gate now sits after the busy-wait (still exempting the repair's own requests), covering both windows. - Fold the repeated kill-to-clean-idle teardown into killChildToIdle and rename REPAIR_RESTORE_TIMEOUT_MS to REPAIR_STEP_TIMEOUT_MS since it now bounds restore and bootstrap. * fix(coding-agent): join concurrent teardowns into one final snapshot flush Concurrent dispose() and shutdown({ snapshot: true }) (e.g. a session dispose racing the SIGINT handler) each ran flushSnapshotForDispose: the first finisher's finally cleared flushingSnapshotForDispose while the second was still flushing, dropping the execution guard early and enqueueing a duplicate final snapshot behind the first. Memoize the in-flight flush so every teardown joins the same bounded flush. * fix(coding-agent): close protocol repair gaps for schema-invalid frames, startup chunks, and rebootstrap - Fail the protocol on JSON object frames that violate the frame schema (unknown event kind; done/host_request without a string id) instead of silently dropping them, which left the active request unsettled forever. The protocol version handshake is exact, so unknown kinds are corruption, not forward compatibility. - A corrupt frame sharing one stdout chunk with ready no longer lets the startup mark the corrupt child running: failProtocolFrame records the error while starting and doStart fails the start. - A fresh kernel started lazily after a discarded repair re-runs the configured runtime bootstrap before serving user code, so rlm/bash/skills handles are never silently missing. * test(coding-agent): reconcile chain-collapse suites and cover teardown racing the lazy re-bootstrap - Point the graceful-shutdown flush spies at captureSnapshot: the merged final flush (kernel-snapshot-dispose-timeout) snapshots through the bounded captureSnapshot path, not snapshotState, so the old spies let real snapshot requests hit the mocked wire and leaked fake timers. - The superseded repair's booting replacement is now hard-killed without a protocol shutdown (single-kernel-shutdown only drains running kernels), so it may die before journaling its spawn; assert spawn-count stability after teardown instead of an exact boot count. - New composition test: a teardown racing an in-flight lazy re-bootstrap stays bounded, spawns nothing new, and settles the raced cell. * fix(coding-agent): re-check the final-flush guard after the re-bootstrap wait - An external execute admitted before the final dispose flush could park on the lazy re-bootstrap, then take its queue slot after the flush captured pendingExecutions - splicing between the captured queue and the final snapshot and unbounding the teardown. Re-check the flush guard right after the ensure await; unit test proves the re-check is load-bearing. - Deflake the teardown-vs-re-bootstrap composition test: the fake runtime journals its spawn before ready, so poll isRunning before the teardown to race the re-bootstrap deterministically instead of the boot. * fix(coding-agent): close bot-review gaps in protocol repair and teardown seams - Terminal frames (done/host_request) with an EMPTY string id now fail the protocol like id-less ones: the runtime mints uuid hex ids and echoes the host's uuids, so an empty id is corruption that previously slipped past the validator into the silent-drop hang. - An execution aborted while waiting on the lazy re-bootstrap settles promptly: the wait races the abort signal (same pattern as waitForProtocolRepair) and the aborted request returns before parking on the queue slot behind the still-running bootstrap. - A failed kernel startup awaits its own teardown (final snapshot flush included) before surfacing the error, so a replacement provisioner gated on this dispose can no longer race the still-flushing kernel over the same snapshot files. - restart() during the final dispose flush is rejected instead of taking a queue slot and joining the owning shutdown, which could deadlock the flush's snapshot behind the held slot. * test(coding-agent): cover empty-string host_request ids in the frame validator Kills the surviving mutant that relaxed the host_request half of the non-empty-id requirement back to typeof-only: an empty-id host_request now has a corruption test mirroring the empty-id done one (protocol fails and the kernel is repaired instead of the awaiting cell hanging). * fix(coding-agent): restore the saved namespace on lazily reprovisioned kernels After a discarded repair whose failure never implicated the snapshot (the repair start died, or only its bootstrap failed), the lazily started fresh kernel re-ran the bootstrap but silently abandoned the user's saved namespace forever. Track restore separately: killChildToIdle marks both pending; the two snapshot-culprit discard paths (clean restore failure, replacement corrupted mid-repair) clear the restore flag so a poisoned snapshot is never retried; the lazy path restores one-shot best-effort (ordinary startup semantics) before bootstrapping, and a teardown's final snapshot never triggers reprovisioning. Restore-attempt journaling in the fake runtime pins the no-retry and no-double-restore semantics. * fix(coding-agent): protect the saved snapshot from unrestored kernels; phase-aware corruption attribution - The final dispose flush skips its snapshot while pendingRestore is set: a kernel that never restored the saved namespace must not overwrite it (the on-disk snapshot is strictly fresher). One interleaving-independent condition instead of joining the reprovision promise, which would reintroduce the 30-60s teardown stall the teardown guard prevents. The lazy one-shot clear moves after the corruption check so the flag also covers a flush racing an in-flight lazy restore. - The repair give-up path attributes corruption by phase: performRestore clears pendingRestore on success, so the flag still being set means the corruption struck at or before the restore (snapshot stays the prime suspect, loop-safe); corruption strictly after a successful restore keeps the namespace restorable at the cost of at most one bounded restore per later attempt.
1- Fixed graceful Python kernel disposal so timed-out final snapshots are cancelled before teardown.1- Fixed invalid kernel protocol frames hanging requests by rejecting the affected request and replacing the kernel from its latest state snapshot.1- Fixed kernel teardown so session cleanup and signal handling share one bounded graceful shutdown path.3904 unmodified linesExpand all3905390639073908390939093910391139123913227 unmodified linesExpand all3904 unmodified linesExpand all
expect(finished.status).toBe("ok"); expect(finished.stdout.trim()).toBe("detached-worker"); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });});
227 unmodified linesExpand all91 unmodified lines92939495969697989910033 unmodified lines13413513613713813813914014191 unmodified lines const envResult = await manager.execute('import os\nprint(os.environ["NO_COLOR"])'); expect(envResult.status).toBe("ok"); expect(envResult.stdout.trim()).toBe("1"); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } }, 60_000);
it("emits canonical paths for edits after the kernel changes directories", async () => {33 unmodified lines expect(first.diffs?.[0]?.path).toBe(realpathSync(join(firstDir, "same.txt"))); expect(second.diffs?.[0]?.path).toBe(realpathSync(join(secondDir, "same.txt"))); expect(first.diffs?.[0]?.path).not.toBe(second.diffs?.[0]?.path); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } }, 60_000);});63 unmodified lines6465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091104 unmodified lines115116117118119120121122123124125126127128129130131132133134135136137138139267 unmodified linesExpand all40740840941035541141241341463 unmodified lines } as unknown as ExtensionContext; return { ctx, setWorkingMessage };}
function writeFakeReplRuntime(markerPath: string): string { const python = join(tempDir, "python-repl"); writeFileSync( python, `#!/usr/bin/env nodeconst fs = require("node:fs");const readline = require("node:readline");const emit = (event) => process.stdout.write(JSON.stringify(event) + "\\n");emit({ event: "ready", protocol: 2, python: process.version });const input = readline.createInterface({ input: process.stdin });input.on("line", (line) => { const request = JSON.parse(line); if (request.type === "restore") { emit({ event: "done", id: request.id, status: "ok", restored: [], failed: [] }); return; } if (request.type === "snapshot") { setTimeout(() => { fs.writeFileSync(${JSON.stringify(markerPath)}, "1"); emit({ event: "done", id: request.id, status: "ok", saved: [], skipped: [], bytes: 0 }); }, 150); return; } if (request.type === "execute") { emit({ event: "error", id: request.id, ename: "RuntimeError", evalue: "bootstrap refused", traceback: [] }); emit({ event: "done", id: request.id, status: "error" }); return; } if (request.type === "shutdown") { emit({ event: "done", id: request.id, status: "ok" }); process.exit(0); }});`, ); chmodSync(python, 0o755); return python;}
describe("IpythonKernelProvisioner", () => { beforeEach(() => { tempDir = mkdtempSync(join(tmpdir(), "prime-agent-provisioner-")); });4 unmodified lines tempDir = ""; } });
it("does not surface a failed startup before the kernel's final snapshot flush finished", async () => { const marker = join(tempDir, "snapshot-flushed"); const snapshotDir = join(tempDir, "snapshots"); mkdirSync(snapshotDir, { recursive: true }); const python = writeFakeReplRuntime(marker); const provisioner = new IpythonKernelProvisioner(tempDir, { python, snapshotDir }); try { await expect(provisioner.ensure()).rejects.toThrow(/Failed to initialize rlm runtime/); // The failed kernel's teardown (final snapshot flush included) completed // before the failure surfaced, so a replacement provisioner gated on this // one cannot race the still-flushing kernel over the same snapshot files. expect(existsSync(marker)).toBe(true); } finally { await provisioner.dispose(); } });
it("memoizes concurrent ensure() calls into one startup", async () => { const { python, countRuns } = writeFakePython(); const provisioner = new IpythonKernelProvisioner(tempDir, { python });
267 unmodified linesExpand all cleanupSessionResources(sessionId); await expect(startup).rejects.toThrow(/Kernel exited before ready|disposed during startup/); expect(manager.isRunning).toBe(false); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });});275 unmodified linesExpand all276277278279280280281282275 unmodified linesExpand all
expect(host.lateSentAgentMessageHandlers.size).toBe(256); expect(host.lateSentAgentMessageHandlers.has("request-0")).toBe(false); expect(host.lateSentAgentMessageHandlers.has("request-299")).toBe(true); await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); });});214 unmodified linesExpand all21521621721821921922022122222315 unmodified lines23924024124224324424524624724824925025125025225325425325425525625725825525626025725825926026126226326426526626926726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551 unmodified lines56756856957033157157257357457546 unmodified lines214 unmodified linesExpand all expect(writeLine.mock.calls.some((call) => (call[0] as { type?: string }).type === "interrupt")).toBe(true); manager.disposeSync(); });
it("starts the snapshot timeout after earlier kernel work finishes", async () => { it("cancels a hung final snapshot execution before teardown", async () => { vi.useFakeTimers(); const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" },15 unmodified lines { once: true }, ); }), ); const cleanupResources = vi.fn(); Object.assign( manager as unknown as { state: "running"; executionQueue: Promise<void>; executeInner: typeof executeInner; start: () => Promise<void>; cleanupResources: () => void; }, { state: "running", executionQueue: previousExecution, executeInner, start: async () => {} }, { state: "running", executionQueue: previousExecution, executeInner, start: async () => {}, cleanupResources }, );
const snapshot = ( manager as unknown as { captureSnapshot: (options?: { executionTimeoutMs?: number }) => Promise<unknown>; } ).captureSnapshot({ executionTimeoutMs: 5000 }); await vi.advanceTimersByTimeAsync(5000); const disposal = manager.shutdown({ snapshot: true, drainHostRequests: true }); expect(executeInner).not.toHaveBeenCalled();
releaseQueue(); await waitForCalls(executeInner, 1); const signal = executeInner.mock.calls[0]?.[2].signal; expect(signal?.aborted).toBe(false); await vi.advanceTimersByTimeAsync(4999); expect(signal?.aborted).toBe(false); expect(cleanupResources).not.toHaveBeenCalled(); await vi.advanceTimersByTimeAsync(1);
expect(signal?.aborted).toBe(true); await expect(snapshot).resolves.toBeNull(); await expect(disposal).resolves.toBe(true); expect(cleanupResources).toHaveBeenCalledOnce(); });
it("tears down when the final snapshot is blocked behind a hung execution", async () => { vi.useFakeTimers(); const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" }, }); const executeInner = vi.fn(); const interrupt = vi.fn(async () => {}); const cleanupResources = vi.fn(); Object.assign( manager as unknown as { state: "running"; executionQueue: Promise<void>; activeExecution: object; executeInner: typeof executeInner; interrupt: typeof interrupt; cleanupResources: () => void; }, { state: "running", executionQueue: new Promise<void>(() => {}), activeExecution: {}, executeInner, interrupt, cleanupResources, }, );
const disposal = manager.shutdown({ snapshot: true, drainHostRequests: true }); expect(interrupt).toHaveBeenCalledOnce(); await vi.advanceTimersByTimeAsync(4999); expect(cleanupResources).not.toHaveBeenCalled(); await vi.advanceTimersByTimeAsync(1);
await expect(disposal).resolves.toBe(true); expect(executeInner).not.toHaveBeenCalled(); expect(cleanupResources).toHaveBeenCalledOnce(); });
it("rejects executions enqueued during the final snapshot flush so dispose stays bounded", async () => { vi.useFakeTimers(); const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" }, }); let releaseQueue: () => void = () => {}; const previousExecution = new Promise<void>((resolve) => { releaseQueue = resolve; }); const executeInner = vi.fn( async ( _requestFields: Record<string, unknown>, _code: string, opts: { signal?: AbortSignal }, ): Promise<{ stdout: string; stderr: string; status: "aborted"; durationMs: number }> => await new Promise((resolve) => { opts.signal?.addEventListener( "abort", () => resolve({ stdout: "", stderr: "", status: "aborted", durationMs: 5000 }), { once: true }, ); }), ); const cleanupResources = vi.fn(); Object.assign( manager as unknown as { state: "running"; executionQueue: Promise<void>; executeInner: typeof executeInner; start: () => Promise<void>; cleanupResources: () => void; }, { state: "running", executionQueue: previousExecution, executeInner, start: async () => {}, cleanupResources }, );
const disposal = manager.shutdown({ snapshot: true, drainHostRequests: true }); // A cell arriving mid-flush must not splice ahead of the final snapshot. await expect(manager.execute("1 + 1")).rejects.toThrow("Kernel is shutting down"); releaseQueue(); await waitForCalls(executeInner, 1); expect(executeInner.mock.calls[0]?.[0].type).toBe("snapshot"); await vi.advanceTimersByTimeAsync(5000);
await expect(disposal).resolves.toBe(true); expect(executeInner).toHaveBeenCalledOnce(); expect(cleanupResources).toHaveBeenCalledOnce(); });
it("rejects an execution whose re-bootstrap wait crossed the start of the final flush", async () => { const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" }, bootstrapCode: "boot-code", }); let releaseBootstrap: () => void = () => {}; const bootstrapBlocked = new Promise<void>((resolve) => { releaseBootstrap = resolve; }); let releaseSnapshot: () => void = () => {}; const snapshotBlocked = new Promise<void>((resolve) => { releaseSnapshot = resolve; }); const calls: string[] = []; const ok = { stdout: "", stderr: "", status: "ok" as const, durationMs: 0 }; const executeInner = vi.fn(async (requestFields: Record<string, unknown> & { type: string }, code: string) => { if (requestFields.type === "snapshot") { calls.push("snapshot"); await snapshotBlocked; return { ...ok, doneFields: { saved: [], skipped: [], bytes: 0 } }; } if (code === "boot-code") { calls.push("bootstrap"); await bootstrapBlocked; return ok; } calls.push("cell"); return ok; }); const cleanupResources = vi.fn(); Object.assign( manager as unknown as { state: "running"; pendingRebootstrap: boolean; executeInner: typeof executeInner; start: () => Promise<void>; cleanupResources: () => void; }, { state: "running", pendingRebootstrap: true, executeInner, start: async () => {}, cleanupResources }, );
// Admitted before the flush: passes the first guard, then parks on the // in-flight lazy re-bootstrap. const cell = manager.execute("user-cell"); cell.catch(() => undefined); await waitForCalls(executeInner, 1); expect(calls).toEqual(["bootstrap"]);
// The teardown's final flush starts while the cell is still parked. const teardown = manager.shutdown({ snapshot: true }); await new Promise((resolve) => globalThis.setTimeout(resolve, 0)); releaseBootstrap();
// Un-bootstrapped no more, the cell must NOT splice between the flush's // captured queue and the final snapshot; the guard re-check rejects it. await expect(cell).rejects.toThrow("Kernel is shutting down"); releaseSnapshot(); await expect(teardown).resolves.toBe(true); expect(calls).toEqual(["bootstrap", "snapshot"]); });
it("settles an aborted execution promptly while the lazy re-bootstrap is still running", async () => { const manager = new ReplKernelManager({ cwd: process.cwd(), bootstrapCode: "boot-code" }); let releaseBootstrap: () => void = () => {}; const bootstrapBlocked = new Promise<void>((resolve) => { releaseBootstrap = resolve; }); const ok = { stdout: "", stderr: "", status: "ok" as const, durationMs: 0 }; const executeInner = vi.fn(async (_requestFields: Record<string, unknown> & { type: string }, code: string) => { if (code === "boot-code") { await bootstrapBlocked; } return ok; }); Object.assign( manager as unknown as { state: "running"; pendingRebootstrap: boolean; executeInner: typeof executeInner; start: () => Promise<void>; }, { state: "running", pendingRebootstrap: true, executeInner, start: async () => {} }, );
const controller = new AbortController(); const cell = manager.execute("user-cell", { signal: controller.signal }); await waitForCalls(executeInner, 1);
// Aborted mid-wait: the cell must not ride out the bootstrap bound. controller.abort(); await expect(cell).resolves.toMatchObject({ status: "aborted" }); releaseBootstrap(); });
it("rejects restart() during the final snapshot flush instead of deadlocking the teardown", async () => { const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" }, }); Object.assign( manager as unknown as { state: "running"; flushingSnapshotForDispose: boolean; start: () => Promise<void> }, { state: "running", flushingSnapshotForDispose: true, start: async () => {}, }, );
// Taking a queue slot here and joining the owning shutdown would leave the // flush's snapshot parked behind the slot forever (mutual wait). await expect(manager.restart()).rejects.toThrow("Kernel is shutting down"); });
it("joins concurrent teardowns into a single final snapshot flush", async () => { const manager = new ReplKernelManager({ cwd: process.cwd(), snapshot: { path: "/tmp/test-state.dill", manifestPath: "/tmp/test-state.json" }, }); const executeInner = vi.fn( async ( _requestFields: Record<string, unknown> & { type: string }, ): Promise<{ stdout: string; stderr: string; status: "ok"; durationMs: number }> => ({ stdout: "", stderr: "", status: "ok", durationMs: 0, }), ); const cleanupResources = vi.fn(); Object.assign( manager as unknown as { state: "running"; executionQueue: Promise<void>; executeInner: typeof executeInner; start: () => Promise<void>; cleanupResources: () => void; }, { state: "running", executionQueue: Promise.resolve(), executeInner, start: async () => {}, cleanupResources }, );
// A session dispose racing a signal-handler shutdown must share one final // flush instead of queueing a second snapshot behind the first. await Promise.all([ manager.shutdown({ snapshot: true, drainHostRequests: true }), manager.shutdown({ snapshot: true }), ]);
const snapshotRequests = executeInner.mock.calls.filter((call) => call[0].type === "snapshot"); expect(snapshotRequests).toHaveLength(1); expect(cleanupResources).toHaveBeenCalled(); });
it("routes null-id and stale-id stream events into backgroundOutput, not stdout", async () => { const writeLine = vi.fn(async (_request: Record<string, unknown>) => {}); const { manager, internals } = runningManagerWith(writeLine);
51 unmodified lines pid: undefined, stdin: { destroyed: false, destroy: () => undefined }, };
await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); const types = writeLine.mock.calls.map((call) => (call[0] as { type?: string }).type); expect(types).toContain("shutdown"); expect(killSignals).toContain("SIGTERM"); });46 unmodified lines35 unmodified lines3637383940404142434482 unmodified lines12712812913013113113213313413529 unmodified lines35 unmodified lines dir = mkdtempSync(join(tmpdir(), "prime-agent-repl-execute-")); });
afterEach(async () => { await manager?.dispose(); await manager?.shutdown({ snapshot: true, drainHostRequests: true }); manager = undefined; if (dir) { rmSync(dir, { recursive: true, force: true }); dir = "";82 unmodified lines const r = await manager.execute("from rlm import bash\nh = bash('sleep 600')\nh.pid"); expect(r.status).toBe("ok"); const pid = Number(r.result); expect(Number.isInteger(pid)).toBe(true); await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); let alive = true; for (let i = 0; i < 100; i++) { try { process.kill(pid, 0);29 unmodified lines63 unmodified lines6465666768686970717221 unmodified lines9495969798989910010110211 unmodified lines11411511611711811811912012112236 unmodified lines1591601611621631631641651661674 unmodified lines17217317417517617617717817918085 unmodified lines266267268269270270271272273274134 unmodified linesExpand all63 unmodified lines try { await expect(manager.execute("x")).rejects.toThrow(/Kernel exited before ready/); } finally { errorSpy.mockRestore(); await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); }
expect(readFileSync(envDump, "utf8")).toMatch(new RegExp(`^PRIME_AGENT_KERNEL_OWNER_PID=${process.pid}$`, "m"));
21 unmodified lines
const records = existsSync(journalPath) ? readJournalRecords(journalPath) : []; expect(records.some((r) => r.pid === 999999 && !r.active)).toBe(false); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });
it("writes the inactive journal record when a signaled kill proves ownership", async () => {11 unmodified lines
const records = readJournalRecords(journalPath); expect(records.some((r) => r.pid === 999999 && !r.active)).toBe(true); } finally { await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });
it("a stale doStart resumed after a concurrent teardown never touches the new kernel", async () => {36 unmodified lines expect(records.some((r) => r.pid === childB.pid)).toBe(false); } finally { internals.child = undefined; internals.state = "idle"; await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });
it("a shutdown superseded by a concurrent kill reports not-owner so recovery cannot resurrect to idle", async () => {4 unmodified lines writeLine: (request: Record<string, unknown>) => Promise<void>; shutdown(): Promise<boolean>; kill(): Promise<void>; }; internals.state = "starting"; internals.state = "running"; // A live child handle keeps waitForKernelExit parked so the send actually blocks the shutdown. const child = Object.assign(new EventEmitter(), { exitCode: null, signalCode: null,85 unmodified lines } finally { internals.child = undefined; internals.startPromise = undefined; internals.state = "idle"; await manager.dispose(); await manager.shutdown({ snapshot: true, drainHostRequests: true }); } });});
134 unmodified linesExpand all9 unmodified lines1011121314151617186 unmodified lines2526272829303132333234353637383940414213 unmodified lines56575859546061626364655758666768697071726073747576777838 unmodified lines11711811912010712112212312412512612711412812913013113213313413513613713813914014114214314414514614714812614915015112815215315413015515615715815922 unmodified lines18218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824954 unmodified lines9 unmodified lines handleEvent: (event: Record<string, unknown>) => void; wireChild: (child: ShutdownInternals["child"]) => void; pendingDoneWaiters: Map<string, () => void>; inFlightHostRequests: Set<Promise<void>>; kernelStderr: string; child: EventEmitter & { exitCode: number | null; signalCode: NodeJS.Signals | null; kill: (signal?: NodeJS.Signals | number) => boolean;6 unmodified lines
function configuredManager( onSend: (request: Record<string, unknown>, internals: ShutdownInternals) => void | Promise<void>, hostHandlers?: HostRequestHandlers, withSnapshot = false,): { manager: ReplKernelManager; internals: ShutdownInternals;} { const manager = new ReplKernelManager({ cwd: process.cwd(), hostHandlers }); const manager = new ReplKernelManager({ cwd: process.cwd(), hostHandlers, snapshot: withSnapshot ? { path: "/tmp/shutdown-test.dill", manifestPath: "/tmp/shutdown-test.json" } : undefined, }); const internals = manager as unknown as ShutdownInternals; const child = Object.assign(new EventEmitter(), { exitCode: null, signalCode: null,13 unmodified lines return { manager, internals };}
describe("ReplKernelManager graceful shutdown", () => { it("bounds a stuck stdin write with the aggregate shutdown deadline", async () => { it.each([ ["optional policy", {}], ["snapshot-and-drain policy", { snapshot: true, drainHostRequests: true }], ] as const)("uses the same diagnostic deadline for the %s", async (_label, options) => { vi.useFakeTimers(); try { const { manager, internals } = configuredManager(() => new Promise<void>(() => {})); const shutdown = manager.shutdown(); const { manager, internals } = configuredManager(() => new Promise<void>(() => {}), undefined, true); // The final flush snapshots through the private captureSnapshot (bounded, prune-free). vi.spyOn(manager as unknown as { captureSnapshot: () => Promise<null> }, "captureSnapshot").mockResolvedValue( null, ); const shutdown = manager.shutdown(options); await vi.advanceTimersByTimeAsync(5_000); await shutdown; await expect(shutdown).resolves.toBe(true); expect(internals.kernelStderr).toContain("Kernel did not shut down within 5000ms"); expect(internals.child).toBeUndefined(); } finally { vi.useRealTimers(); }38 unmodified lines vi.useRealTimers(); } });
it("settles an in-flight host request before dispose tears the kernel down", async () => { it("flushes a snapshot and drains host requests when session teardown selects both policies", async () => { let releaseHandler: (() => void) | undefined; const handlerBlocked = new Promise<void>((resolve) => { releaseHandler = resolve; }); const sentReplies: Record<string, unknown>[] = []; const { manager, internals } = configuredManager( async (request) => { async (request, state) => { if (request.type === "host_reply") sentReplies.push(request); if (request.type === "shutdown") { state.handleEvent({ event: "done", id: request.id, status: "ok" }); state.child.exitCode = 0; state.child.emit("exit", 0, null); } }, { "test.slow": async () => { await handlerBlocked; return { answer: 42, status: "weird" }; }, }, true, ); const snapshot = vi .spyOn(manager as unknown as { captureSnapshot: () => Promise<null> }, "captureSnapshot") .mockResolvedValue(null);
internals.handleEvent({ event: "host_request", id: "hr-1", data: { type: "test.slow" } }); expect(internals.inFlightHostRequests.size).toBe(1); const shutdown = manager.shutdown({ snapshot: true, drainHostRequests: true }); await new Promise((resolve) => globalThis.setTimeout(resolve, 0));
const dispose = manager.dispose(); expect(snapshot).toHaveBeenCalledOnce(); expect(internals.writeLine).not.toHaveBeenCalled(); releaseHandler?.(); await dispose; await shutdown;
expect(internals.inFlightHostRequests.size).toBe(0); expect(sentReplies).toEqual([ {22 unmodified lines await handlerBlocked; return { answer: 42 }; }, }, true, ); const snapshot = vi .spyOn(manager as unknown as { captureSnapshot: () => Promise<null> }, "captureSnapshot") .mockResolvedValue(null);
internals.handleEvent({ event: "host_request", id: "hr-1", data: { type: "test.slow" } }); const tracked = [...internals.inFlightHostRequests]; expect(tracked).toHaveLength(1);
await manager.shutdown(); expect(snapshot).not.toHaveBeenCalled(); expect(internals.child).toBeUndefined();
// The reply write fails against the torn-down child; the task must still settle. releaseHandler?.(); await Promise.all(tracked); expect(internals.inFlightHostRequests.size).toBe(0); });
it("shares one wire exchange across concurrent graceful shutdown calls", async () => { let exitKernel: (() => void) | undefined; const { manager, internals } = configuredManager((request, state) => { state.handleEvent({ event: "done", id: request.id, status: "ok" }); exitKernel = () => { state.child.exitCode = 0; state.child.emit("exit", 0, null); }; });
const owner = manager.shutdown(); await new Promise((resolve) => globalThis.setTimeout(resolve, 0)); const concurrent = manager.shutdown(); exitKernel?.();
await expect(owner).resolves.toBe(true); await expect(concurrent).resolves.toBe(false); expect(internals.writeLine).toHaveBeenCalledOnce(); });
it("stands down when a hard teardown supersedes the snapshot flush", async () => { let releaseSnapshot: (() => void) | undefined; const snapshotBlocked = new Promise<null>((resolve) => { releaseSnapshot = () => resolve(null); }); const { manager, internals } = configuredManager(() => undefined, undefined, true); vi.spyOn(manager as unknown as { captureSnapshot: () => Promise<null> }, "captureSnapshot").mockReturnValue( snapshotBlocked, ); const child = internals.child;
const shutdown = manager.shutdown({ snapshot: true }); await new Promise((resolve) => globalThis.setTimeout(resolve, 0)); await manager.kill(); releaseSnapshot?.();
await expect(shutdown).resolves.toBe(false); expect(child.kill).toHaveBeenCalledOnce(); });
it("restart after a graceful shutdown starts the kernel again", async () => { const { manager, internals } = configuredManager((request, state) => { if (request.type !== "shutdown") return; state.handleEvent({ event: "done", id: request.id, status: "ok" });54 unmodified lines