From 59bd71dff28c41ba02c6254bbfdf4573b407d1ea Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 2 Oct 2026 09:41:24 -0500 Subject: [PATCH] Preserve the owned Pi runner across controller-only E2E restart Separate durable runner preservation from final process retirement. Bind restart preservation to the public run identity and observed ancestry, and drain every retained owner after the replacement controller shuts down. Co-authored-by: Paperclip --- tests/runner-e2e/FIXTURES.md | 6 + tests/runner-e2e/pi-native-flow.ts | 5 +- .../runner-e2e/pi-native-restart-flow.test.ts | 14 +- tests/runner-e2e/pi-native-restart-flow.ts | 8 +- tests/runner-e2e/process-tree-owner.ts | 52 ++++++ tests/runner-e2e/process-tree.ts | 11 +- tests/runner-e2e/runner.spec.ts | 6 +- ...erver-restart-preservation-process.test.ts | 75 ++++++++ .../server-restart-preservation.test.ts | 176 ++++++++++++++++++ tests/runner-e2e/server-stop.ts | 73 +++++++- tests/runner-e2e/server.ts | 28 ++- 11 files changed, 419 insertions(+), 35 deletions(-) create mode 100644 tests/runner-e2e/server-restart-preservation-process.test.ts create mode 100644 tests/runner-e2e/server-restart-preservation.test.ts diff --git a/tests/runner-e2e/FIXTURES.md b/tests/runner-e2e/FIXTURES.md index 4b5e075e311..38b63440426 100644 --- a/tests/runner-e2e/FIXTURES.md +++ b/tests/runner-e2e/FIXTURES.md @@ -309,6 +309,12 @@ no database writes, private hooks, or fabricated provider results are allowed. `native-pending-controller-restart` restarts the public controller while one Pi input callback remains unanswered. It requires the same durable interaction, request, live run, native session, turn and producer before and after restart. +For local execution, the public run's exact PID, process group and start identity +must identify an already-observed durable runner under this controller. The test +preserves only that runner tree during restart, checks the same live identity +afterward, and keeps the old process owner for complete final cleanup alongside +the replacement controller. Other controller children are retired normally. +Remote execution does not infer local process authority from remote PIDs. Only then does the browser submit previously undisclosed text. One durable resolution, one original successful turn and independently read exact workspace JSON prove delivery. A replacement run, replay, cancellation, expiry, rewritten diff --git a/tests/runner-e2e/pi-native-flow.ts b/tests/runner-e2e/pi-native-flow.ts index 2b771161cdb..fea7a89d27a 100644 --- a/tests/runner-e2e/pi-native-flow.ts +++ b/tests/runner-e2e/pi-native-flow.ts @@ -1,3 +1,4 @@ +import type { RestartRunnerIdentity } from "./process-tree-owner.js"; import { observeRunProcesses, createDeniedTargetFixture, bindDeniedTargetPrompt } from "./copilot-local-fixtures.js"; import { hasDeliveredPiDenial, piPermissionRequests, hasPiRemoteRetirement, hasUnchangedPiRemoteTarget } from "./pi-native-evidence.js"; import { randomBytes } from "node:crypto"; @@ -25,7 +26,7 @@ type Check = { id: string; passed: boolean; detail: string }; export async function runPiNativeFlow(input: { page: Page; api: RunnerApi; fixtures: LiveFixtureValues; execution: MatrixExecution; nonce: string; - workspacePath: string; deadlineAt: number; restart(): Promise; + workspacePath: string; deadlineAt: number; restart(preserveRunner?: RestartRunnerIdentity): Promise; observe(issue: Row, runs: Row[]): void; capture(id: string, label: string, file: string): Promise; evidence(name: string, data: unknown): Promise; @@ -112,7 +113,7 @@ export async function runPiNativeFlow(input: { await create(execution.task.buildTitle(nonce), execution.task.buildPrompt(nonce), { targets: ["pi-native-restart-answer.json"] }); checks.push(...await runPiPendingControllerRestart({ page, companyId: fixtures.company.id, deadlineAt: input.deadlineAt, load, events, - restart: input.restart, settle: () => settle(1), capture: input.capture, evidence: input.evidence, + preserveLocalRunner: !remote, restart: input.restart, settle: () => settle(1), capture: input.capture, evidence: input.evidence, readProof: async () => { if (remote) await finishRemote("native-restart-final"); return JSON.parse(await readWorkspace("pi-native-restart-answer.json")); diff --git a/tests/runner-e2e/pi-native-restart-flow.test.ts b/tests/runner-e2e/pi-native-restart-flow.test.ts index aca08bbcad6..475b5b72c11 100644 --- a/tests/runner-e2e/pi-native-restart-flow.test.ts +++ b/tests/runner-e2e/pi-native-restart-flow.test.ts @@ -18,7 +18,7 @@ function fixture(adapter = "acpx-runtime-sidecar") { normalizedSessionId: "session", sourceInstanceId: "runner", sourceSeq, sourceEventId: `runner:run:${sourceSeq}`, payload } }, }); const state = { issue: { id: "issue", companyId: "company", status: "in_progress" }, - runs: [{ id: "run", companyId: "company", nativeIssueId: "issue", runtimeMode: "native", status: "running", nativeSessionId: "session", runnerInstanceId: "runner" } as Row], + runs: [{ id: "run", companyId: "company", nativeIssueId: "issue", runtimeMode: "native", status: "running", nativeSessionId: "session", runnerInstanceId: "runner", processPid: 200, processGroupId: 200, processStartedAt: "2026-10-02T13:55:27.000Z" } as Row], interactions: [{ id: "interaction", companyId: "company", issueId: "issue", kind: "ask_user_questions", status: "pending", result: null, sourceRunId: "run", continuationPolicy: "none", resolverPolicy: "human_only", idempotencyKey: "paperclip-runner-question:run:request", payload: { runtimeRequestId: "request", questionSet } } as Row] }; @@ -164,7 +164,7 @@ it.each([null, {}, { status: "answered", value: "guess" }, { status: "cancelled" }, ); -async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution") { +async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution" | "replaced-runner") { const f = fixture(), evidence = new Map(), checkpoints: string[] = []; let answer = "", restarts = 0, submissions = 0, proof: unknown, resolvePost: ((value: unknown) => void) | undefined; let postPredicate: ((value: any) => boolean) | undefined; @@ -181,23 +181,25 @@ async function flow(failure?: "replaced" | "wrong-file" | "missing-resolution") waitForRequest: (predicate: (value: any) => boolean) => { postPredicate = predicate; return new Promise(resolve => { resolvePost = resolve; }); } }; const result = runPiPendingControllerRestart({ page, companyId: "company", deadlineAt: Date.now() + 1000, load: async () => structuredClone(f.state), events: async () => structuredClone(f.events), - restart: async () => { + restart: async identity => { + expect(identity).toEqual({ processPid: 200, processGroupId: 200, processStartedAt: "2026-10-02T13:55:27.000Z" }); expect(submissions).toBe(0); expect(f.state.runs[0]!.status).toBe("running"); expect(f.state.interactions[0]!.status).toBe("pending"); expect(answer).toBe(""); restarts++; checkpoints.push("restart"); if (failure === "replaced") f.state.interactions[0]!.id = "replacement"; + if (failure === "replaced-runner") f.state.runs[0]!.processPid = f.state.runs[0]!.processGroupId = 201; }, settle: async () => structuredClone(f.state), readProof: async () => failure === "wrong-file" ? { status: "answered", value: "guess" } : proof, capture: async id => { checkpoints.push(id); }, evidence: async (name, data) => { evidence.set(name, structuredClone(data)); }, }); if (failure) await expect(result).rejects.toThrow("Pi native restart"); else expect((await result).every(check => check.passed)).toBe(true); - expect(restarts).toBe(1); expect(submissions).toBe(failure === "replaced" ? 0 : 1); + expect(restarts).toBe(1); expect(submissions).toBe(["replaced", "replaced-runner"].includes(failure ?? "") ? 0 : 1); expect(checkpoints.slice(0, 3)).toEqual(["pi-restart-pending", "restart", "reload"]); expect(evidence.has("pi-native-restart-before.json")).toBe(true); expect(evidence.has("pi-native-restart-after.json")).toBe(true); expect(evidence.has("pi-native-restart-checks.json")).toBe(true); expect(JSON.stringify(evidence.get("pi-native-restart-before.json"))).not.toContain("PI-RESTART-"); - if (failure !== "replaced") expect(evidence.has("pi-native-restart-final.json")).toBe(true); + if (!["replaced", "replaced-runner"].includes(failure ?? "")) expect(evidence.has("pi-native-restart-final.json")).toBe(true); } it("restarts while unanswered, reloads, then submits hidden text through the exact browser route", () => flow()); -it.each(["replaced", "wrong-file", "missing-resolution"] as const)("retains evidence and fails whole flow on %s", failure => flow(failure)); +it.each(["replaced", "wrong-file", "missing-resolution", "replaced-runner"] as const)("retains evidence and fails whole flow on %s", failure => flow(failure)); diff --git a/tests/runner-e2e/pi-native-restart-flow.ts b/tests/runner-e2e/pi-native-restart-flow.ts index 1605c749347..b2488a0214e 100644 --- a/tests/runner-e2e/pi-native-restart-flow.ts +++ b/tests/runner-e2e/pi-native-restart-flow.ts @@ -1,3 +1,4 @@ +import { parseRestartRunnerIdentity, type RestartRunnerIdentity } from "./process-tree-owner.js"; import { createHash, randomBytes } from "node:crypto"; import { expect, type Page } from "@playwright/test"; import { canonicalJson } from "../../packages/shared/src/portability-hash.js"; @@ -132,7 +133,7 @@ export function gradePiRestartCompletion(state: State, events: readonly Row[], p export async function runPiPendingControllerRestart(input: { page: Page; companyId: string; deadlineAt: number; load(): Promise; events(runId: string): Promise; - restart(): Promise; settle(): Promise; readProof(): Promise; + preserveLocalRunner?: boolean; restart(preserveRunner?: RestartRunnerIdentity): Promise; settle(): Promise; readProof(): Promise; capture(id: string, label: string, file: string): Promise; evidence(name: string, data: unknown): Promise; }): Promise { const checks: Check[] = [], answer = `PI-RESTART-${randomBytes(16).toString("hex")}`; @@ -150,10 +151,13 @@ export async function runPiPendingControllerRestart(input: { const composer = () => input.page.getByTestId("question-text-answer-composer").filter({ visible: true }); await expect(composer()).toHaveCount(1); await input.capture("pi-restart-pending", "Pi native question before controller restart", "pi-restart-pending.png"); - await input.restart(); + const runnerIdentity = input.preserveLocalRunner === false ? undefined : parseRestartRunnerIdentity(before.state.runs[0]); + await input.restart(runnerIdentity); await input.page.reload(); const after = await snapshot("pi-native-restart-after.json"), resumed = observePiRestartPending(after.state, after.events, input.companyId); check("restart-same-pending-native-request", digest(resumed) === digest(pending), "Controller restart retained the same unanswered request, run, turn, native session and producer"); + if (runnerIdentity) check("restart-same-live-runner-identity", digest(parseRestartRunnerIdentity(after.state.runs[0])) === digest(runnerIdentity), + "The original durable runner PID, process group and start identity survived the controller restart"); await expect(composer()).toHaveCount(1); await input.capture("pi-restart-reconnected", "Same Pi question after controller restart", "pi-restart-reconnected.png"); await composer().locator('[contenteditable="true"],textarea').first().fill(answer); diff --git a/tests/runner-e2e/process-tree-owner.ts b/tests/runner-e2e/process-tree-owner.ts index a5da0dbbd7f..531c0df646e 100644 --- a/tests/runner-e2e/process-tree-owner.ts +++ b/tests/runner-e2e/process-tree-owner.ts @@ -5,6 +5,20 @@ import { type ObservedProcessGroup, type ProcessObservation, } from "./process-tree.js"; +/** Supplied by the company-scoped run API, never by process-name discovery. */ +export interface RestartRunnerIdentity { + processPid: number; + processGroupId: number; + processStartedAt: string; +} +export function parseRestartRunnerIdentity(value: unknown): RestartRunnerIdentity { + const v = value as RestartRunnerIdentity | null; + if (!v || !Number.isSafeInteger(v.processPid) || v.processPid <= 1 + || v.processGroupId !== v.processPid || typeof v.processStartedAt !== "string" + || !Number.isFinite(Date.parse(v.processStartedAt))) throw new Error("Invalid restart runner identity"); + return { processPid: v.processPid, processGroupId: v.processGroupId, processStartedAt: v.processStartedAt }; +} + export function createProcessTreeOwner(root: ChildProcess, options: { readTable?: () => Promise; signalGroup?: (pid: number, signal: NodeJS.Signals) => void; @@ -12,6 +26,8 @@ export function createProcessTreeOwner(root: ChildProcess, options: { let groups: ObservedProcessGroup[] = []; let table: ProcessObservation[] = []; let rootStarted: string | undefined; + let preserved: ObservedProcessGroup[] = []; + let restartRunner: ProcessObservation | undefined; let observing: Promise | undefined; const covered = new Set(); const readTable = options.readTable ?? readProcessTable; @@ -47,6 +63,14 @@ export function createProcessTreeOwner(root: ChildProcess, options: { } } groups = [...byGroup.values()]; + // Follow only continuously owned members of the admitted daemon tree. + // Orphaned descendants stay owned, but unrelated controller children do not. + const retainedPreserved = refreshContinuouslyLiveProcessGroups(revalidateObservedProcessGroups(preserved, next), next); + const preservedIds = new Set(retainedPreserved.map(group => group.processGroupId)); + for (const member of retainedPreserved.flatMap(group => group.members)) { + for (const group of observeDescendantProcessTree(next, member.pid).groups) preservedIds.add(group.processGroupId); + } + preserved = groups.filter(group => preservedIds.has(group.processGroupId)); table = next; })().finally(() => { observing = undefined; }); return observing; @@ -60,6 +84,33 @@ export function createProcessTreeOwner(root: ChildProcess, options: { function liveGroups() { return groups.filter(group => table.some(row => row.processGroupId === group.processGroupId && running(row))); } + async function preserveRunnerForRestart(identity: RestartRunnerIdentity) { + const expected = parseRestartRunnerIdentity(identity); + await observe(); + if (restartRunner || exited()) throw new Error("Restart runner admission must precede controller exit"); + const tree = observeDescendantProcessTree(table, root.pid!); + const candidate = tree.members.map(member => member.process).find(row => row.pid === expected.processPid); + if (!candidate || !running(candidate) || candidate.kind !== "paperclip-runnerd" + || candidate.processGroupId !== expected.processGroupId + || Date.parse(candidate.started) !== Date.parse(expected.processStartedAt) + || candidate.processGroupId === root.pid) throw new Error("Restart runner is not the exact owned durable daemon"); + const subtree = observeDescendantProcessTree(table, candidate.pid); + const members = new Set(subtree.members.map(member => member.process.pid)); + const ids = new Set(subtree.groups.map(group => group.processGroupId)); + if (table.some(row => ids.has(row.processGroupId) && !members.has(row.pid) && running(row))) { + throw new Error("Restart runner shares a process group with an unrelated process"); + } + restartRunner = { ...candidate }; + preserved = groups.filter(group => ids.has(group.processGroupId)); + } + function assertRestartRunnerAlive() { + if (!restartRunner || !table.some(row => row.pid === restartRunner!.pid && row.started === restartRunner!.started + && row.processGroupId === restartRunner!.processGroupId && running(row))) throw new Error("Admitted restart runner did not survive controller restart"); + } + function restartCleanupGroups() { + const ids = new Set(preserved.map(group => group.processGroupId)); + return new Set(liveGroups().filter(group => !ids.has(group.processGroupId)).map(group => group.processGroupId)); + } function signalSnapshot(signal: NodeJS.Signals, selected?: ReadonlySet) { const currentProcessGroupId = table.find(row => row.pid === process.pid)?.processGroupId ?? null; if (currentProcessGroupId === null) throw new Error("Cleanup caller group identity is unavailable"); @@ -118,6 +169,7 @@ export function createProcessTreeOwner(root: ChildProcess, options: { graceComplete = selected.size > 0 && delivered.length === selected.size; } return { observe, signal, liveGroups, gracefulRoots, signalGracefully, + preserveRunnerForRestart, assertRestartRunnerAlive, restartCleanupGroups, directGraceDelivered: () => directGraceDelivered, stopObserving: () => { if (timer) clearInterval(timer); } }; } diff --git a/tests/runner-e2e/process-tree.ts b/tests/runner-e2e/process-tree.ts index 1ab4ebcddc9..7a21642d9cd 100644 --- a/tests/runner-e2e/process-tree.ts +++ b/tests/runner-e2e/process-tree.ts @@ -159,6 +159,14 @@ const diagnosticProcessKinds = new Set([ "tsx", ]); +export function diagnosticProcessKind(command: string, platform: NodeJS.Platform = process.platform) { + const name = path.basename(command); + // Linux comm is limited to 15 bytes. This is a diagnostic role only; + // preservation still requires the trusted run PID/start/group and ancestry. + if (platform === "linux" && name === "paperclip-runne") return "paperclip-runnerd"; + return diagnosticProcessKinds.has(name) ? name : "other"; +} + export async function readProcessTable(startInspector?: () => ChildProcess): Promise { if (process.platform === "win32") { return null; @@ -212,8 +220,7 @@ export async function readProcessTable(startInspector?: () => ChildProcess): Pro // A target process can choose its own argv and process name. Emit a // fixed category instead of target-controlled text so diagnostics // can never turn that metadata into a secret-exfiltration channel. - const command = path.basename(match[6]!); - const kind = diagnosticProcessKinds.has(command) ? command : "other"; + const kind = diagnosticProcessKind(match[6]!); return { pid: Number(match[1]), parentPid: Number(match[2]), diff --git a/tests/runner-e2e/runner.spec.ts b/tests/runner-e2e/runner.spec.ts index cf284fa310b..eb5135d436c 100644 --- a/tests/runner-e2e/runner.spec.ts +++ b/tests/runner-e2e/runner.spec.ts @@ -1,3 +1,4 @@ +import type { RestartRunnerIdentity } from "./process-tree-owner.js"; import { runNativeActiveStopFlow } from "./native-active-stop-flow.js"; import { runPiControlsFlow } from "./pi-controls-flow.js"; import { runCursorNativeFlow } from "./cursor-native-flow.js"; @@ -227,6 +228,7 @@ async function restartIsolatedPaperclipServer(input: { api: RunnerApi; requestId: string; deadlineAt: number; + preserveRunner?: RestartRunnerIdentity; }): Promise { const { controlDirectory, @@ -237,7 +239,7 @@ async function restartIsolatedPaperclipServer(input: { const temporaryRequestPath = `${requestPath}.${process.pid}.${input.requestId}.tmp`; await writeFile( temporaryRequestPath, - JSON.stringify({ requestId: input.requestId }), + JSON.stringify({ requestId: input.requestId, ...(input.preserveRunner ? { preserveRunner: input.preserveRunner } : {}) }), { encoding: "utf8", mode: 0o600 }, ); await rename(temporaryRequestPath, requestPath); @@ -976,7 +978,7 @@ for (const execution of executions) { } else if (execution.task.flow === "pi_native") { const story = await runPiNativeFlow({ page, api, fixtures, execution, nonce, workspacePath, deadlineAt: startedAtMs + deadlineMs, - restart: () => restartIsolatedPaperclipServer({ api, requestId: `pi-native-${nonce}`, deadlineAt: startedAtMs + deadlineMs }), + restart: preserveRunner => restartIsolatedPaperclipServer({ api, requestId: `pi-native-${nonce}`, deadlineAt: startedAtMs + deadlineMs, preserveRunner }), observe: (currentIssue, currentRuns) => { issue = currentIssue as IssueRecord; selectedRuns = currentRuns as RunRecord[]; }, capture: captureScreenshot, evidence: (name, data) => writeSanitizedJson(snapshotsDir, name, data, secrets), diff --git a/tests/runner-e2e/server-restart-preservation-process.test.ts b/tests/runner-e2e/server-restart-preservation-process.test.ts new file mode 100644 index 00000000000..10b1e720635 --- /dev/null +++ b/tests/runner-e2e/server-restart-preservation-process.test.ts @@ -0,0 +1,75 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { once } from "node:events"; +import { mkdtemp, realpath, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { expect, it } from "vitest"; +import { createRunnerE2EServerStopper } from "./server-stop.js"; +import { createProcessTreeOwner } from "./process-tree-owner.js"; +import { readProcessTable } from "./process-tree.js"; + +it.skipIf(process.platform === "win32")("keeps the real admitted daemon and child alive across controller exit, then drains both controller generations", async () => { + const directory = await mkdtemp(path.join(await realpath(os.tmpdir()), "pc-restart-owner-")); + const daemon = "/bin/sh"; + const fixtureDaemons = new Set(); + const entry = path.join(directory, "controller.cjs"); + const stop = createRunnerE2EServerStopper({ gracefulTimeoutMs: 3000, forcedTimeoutMs: 2000, + hasSpawnError: () => false, markExpectedStop: () => {}, log: () => {}, + // Only the fixture's declared role is synthetic. PID, start, ancestry, + // groups, observation and signal delivery all use the real kernel table. + createOwner: child => createProcessTreeOwner(child, { readTable: async () => { + const table = await readProcessTable(); + return table?.map(row => fixtureDaemons.has(row.pid) && row.kind === "sh" ? { ...row, kind: "paperclip-runnerd" } : row) ?? null; + } }), + }); + const closed: Promise[] = []; + try { + // This is a tiny inert process-role fixture, not a provider or real daemon. + await writeFile(entry, ` + const { spawn } = require('node:child_process'); + const fs = require('node:fs'); + const marker = ${JSON.stringify(path.join(directory, "child-ready"))} + process.pid; + const daemon = spawn(${JSON.stringify(daemon)}, ['-c', ${JSON.stringify("trap 'exit 0' TERM; /bin/sleep 600 & printf '%s' \"$!\" > \"$1\"; wait")}, 'fixture', marker], { detached: true, stdio: ['ignore', 'ignore', 'pipe'] }); + let daemonError = ''; daemon.stderr.on('data', chunk => { daemonError += chunk; }); + daemon.once('exit', (code, signal) => { if (process.connected) process.send({ failure: { code, signal, stderr: daemonError } }); }); + const background = spawn('/bin/sleep', ['600'], { detached: true, stdio: 'ignore' }); + process.on('SIGTERM', () => process.exit(0)); + const ready = setInterval(() => { + if (fs.existsSync(marker)) { clearInterval(ready); process.send({ daemon: daemon.pid, background: background.pid }); } + }, 10); + setInterval(() => {}, 1000); + `); + const start = async () => { + const child = spawn(process.execPath, [entry], { detached: true, stdio: ["ignore", "ignore", "pipe", "ipc"] }); + closed.push(new Promise(resolve => child.once("close", () => resolve()))); + const message = once(child, "message"); + await stop.watch(child); + const result = (await message)[0]; + expect(result.failure, JSON.stringify(result)).toBeUndefined(); + fixtureDaemons.add(result.daemon); + return { child, ids: result as { daemon: number; background: number } }; + }; + const first = await start(); + const before = await readProcessTable(); + expect(before).not.toBeNull(); + const runner = before!.find(row => row.pid === first.ids.daemon)!; + expect(runner.kind).toBe("sh"); + const descendants = before!.filter(row => row.parentPid === runner.pid); + expect(descendants.length).toBeGreaterThan(0); + await stop.forRestart(first.child, { processPid: runner.pid, processGroupId: runner.processGroupId, + processStartedAt: new Date(runner.started).toISOString() }); + const after = await readProcessTable(); + expect(after!.find(row => row.pid === runner.pid)?.started).toBe(runner.started); + for (const descendant of descendants) expect(after!.find(row => row.pid === descendant.pid)?.started).toBe(descendant.started); + expect(after!.some(row => row.pid === first.ids.background && !row.state?.startsWith("Z"))).toBe(false); + const second = await start(); + await stop.stopAll(); await Promise.all(closed); + const final = await readProcessTable(); + const expectedGone = [first.child.pid, first.ids.daemon, first.ids.background, ...descendants.map(row => row.pid), + second.child.pid, second.ids.daemon, second.ids.background]; + expect(final!.filter(row => expectedGone.includes(row.pid) && !row.state?.startsWith("Z"))).toEqual([]); + } finally { + try { await stop.stopAll(); await Promise.all(closed); } + finally { await rm(directory, { recursive: true, force: true }); } + } +}, 15000); diff --git a/tests/runner-e2e/server-restart-preservation.test.ts b/tests/runner-e2e/server-restart-preservation.test.ts new file mode 100644 index 00000000000..110fe76dfdb --- /dev/null +++ b/tests/runner-e2e/server-restart-preservation.test.ts @@ -0,0 +1,176 @@ +import type { ChildProcess } from "node:child_process"; +import { expect, it, vi } from "vitest"; +import { createProcessTreeOwner, parseRestartRunnerIdentity } from "./process-tree-owner.js"; +import { createRunnerE2EServerStopper } from "./server-stop.js"; +import { diagnosticProcessKind, type ProcessObservation } from "./process-tree.js"; + +const started = "2026-10-02T13:55:27.000Z"; +const identity = { processPid: 200, processGroupId: 200, processStartedAt: started }; +const row = (pid: number, parentPid: number, group = pid, kind = "node"): ProcessObservation => + ({ pid, parentPid, processGroupId: group, started, kind, state: "S" }); +function fixture() { + let table = [row(process.pid, 1, 10), row(100, process.pid), row(200, 100, 200, "paperclip-runnerd"), + row(201, 200), row(300, 100), row(900, process.pid)]; + const signals: Array<[number, string]> = []; + const roots: Array = []; + const owners: Array> = []; + const root = (pid: number) => { + const child = { pid, exitCode: null as number | null, signalCode: null as NodeJS.Signals | null, + kill: (signal: string) => { + signals.push([pid, signal]); child.exitCode = 0; + table = table.filter(row => row.pid !== pid).map(row => row.parentPid === pid ? { ...row, parentPid: 1 } : row); + return true; + } }; + const handle = child as unknown as ChildProcess; + roots.push(handle); return handle; + }; + const stop = createRunnerE2EServerStopper({ gracefulTimeoutMs: 100, forcedTimeoutMs: 100, + hasSpawnError: () => false, markExpectedStop: () => {}, log: () => {}, + createOwner: child => { + const owner = createProcessTreeOwner(child, { readTable: async () => table.map(row => ({ ...row })), + signalGroup: (group, signal) => { signals.push([group, signal]); table = table.filter(row => row.processGroupId !== group); } }); + owners.push(owner); return owner; + } }); + return { stop, child: root(100), root, signals, owners, table: () => table, + replace: (next: ProcessObservation[]) => { table = next; }, close: () => owners.forEach(owner => owner.stopObserving()) }; +} + +it.skipIf(process.platform === "win32")("preserves only the admitted daemon tree across restart and finally retires old and new owners", async () => { + const f = fixture(); + try { + await f.stop.watch(f.child); + await f.stop.forRestart(f.child, identity); + expect(f.signals).toEqual([[100, "SIGTERM"], [300, "SIGTERM"]]); + expect(f.table().map(row => row.pid)).toContain(200); + // The daemon forks a new private group after controller loss; it stays owned. + f.replace([...f.table(), row(202, 201), row(400, process.pid), row(401, 400)]); + const replacement = f.root(400); await f.stop.watch(replacement); + await f.stop.stopAll(); + expect(f.table().map(row => row.pid).sort()).toEqual([process.pid, 900].sort()); + expect(f.signals.filter(([pid]) => pid === 200)).toEqual([[200, "SIGTERM"]]); + expect(f.signals.findIndex(([pid]) => pid === 400)).toBeLessThan(f.signals.findIndex(([pid]) => pid === 200)); + expect(f.signals.some(([pid]) => pid === 202)).toBe(true); + expect(f.signals.some(([pid]) => pid === 900)).toBe(false); + } finally { f.close(); } +}); + +it.skipIf(process.platform === "win32").each([ + ["foreign controller", (rows: ProcessObservation[]) => { rows.find(row => row.pid === 200)!.parentPid = 900; }], + ["reused PID", (rows: ProcessObservation[]) => { rows.find(row => row.pid === 200)!.started = "2026-10-02T13:55:28.000Z"; }], + ["wrong role", (rows: ProcessObservation[]) => { rows.find(row => row.pid === 200)!.kind = "node"; }], + ["mixed process group", (rows: ProcessObservation[]) => { rows.push(row(901, 900, 200)); }], + ["same controller group", (rows: ProcessObservation[]) => { rows.find(row => row.pid === 200)!.processGroupId = 100; }], + ["missing runner", (rows: ProcessObservation[]) => { rows.splice(rows.findIndex(row => row.pid === 200), 1); }], +] as const)("rejects %s without sending a restart signal", async (_name, mutate) => { + const f = fixture(); + try { + const rows = f.table(); mutate(rows); f.replace(rows); + await expect(f.stop.forRestart(f.child, identity)).rejects.toThrow(/runner|daemon|process group/); + expect(f.signals).toEqual([]); + } finally { f.close(); } +}); + +it.skipIf(process.platform === "win32")("fails if the admitted daemon dies instead of pretending recovery preserved it", async () => { + const f = fixture(); + try { + const kill = f.child.kill.bind(f.child); + f.child.kill = signal => { const result = kill(signal); f.replace(f.table().filter(row => row.pid !== 200)); return result; }; + await expect(f.stop.forRestart(f.child, identity)).rejects.toThrow("did not survive"); + await f.stop.stopAll(); + expect(f.table().map(row => row.pid).sort()).toEqual([process.pid, 900].sort()); + } finally { f.close(); } +}); + +it.skipIf(process.platform === "win32")("does not transfer ownership to a reused retained daemon group during final cleanup", async () => { + const f = fixture(); + try { + await f.stop.forRestart(f.child, identity); + f.replace(f.table().map(row => row.pid === 200 ? { ...row, started: "2026-10-02T14:00:00.000Z" } : row)); + await expect(f.stop.stopAll()).rejects.toThrow("cleanup incomplete"); + expect(f.signals.some(([pid]) => pid === 200)).toBe(false); + } finally { f.close(); } +}); + +it.each([null, {}, { ...identity, processPid: 0 }, { ...identity, processGroupId: 100 }, { ...identity, processStartedAt: "bad" }])( + "rejects malformed restart identity %j", value => { expect(() => parseRestartRunnerIdentity(value)).toThrow(); }); + +it.skipIf(process.platform === "win32")("keeps the first final-cleanup deadline after an inspection failure and repeated stop", async () => { + const f = fixture(); + const clock = vi.spyOn(Date, "now"); + try { + await f.stop.forRestart(f.child, identity); + clock.mockReturnValue(10_000); + const owner = f.owners[0]!; + const observe = owner.observe; + let fail = true; + owner.observe = async () => { if (fail) { fail = false; throw new Error("one inspection failure"); } await observe(); }; + await expect(f.stop(f.child)).rejects.toThrow("inspection failure"); + clock.mockReturnValue(20_000); + await f.stop(f.child); + expect(f.signals.filter(([pid]) => pid === 200)).toEqual([[200, "SIGKILL"]]); + } finally { clock.mockRestore(); f.close(); } +}); + +it.skipIf(process.platform === "win32")("final cancellation interrupts restart and retires its preserved descendants", async () => { + const f = fixture(); + let cleanup: Promise | undefined; + try { + const kill = f.child.kill.bind(f.child); + f.child.kill = signal => { const result = kill(signal); cleanup = f.stop.stopAll(); return result; }; + await expect(f.stop.forRestart(f.child, identity)).rejects.toThrow("Final shutdown interrupted"); + await cleanup; + expect(f.table().map(row => row.pid).sort()).toEqual([process.pid, 900].sort()); + } finally { f.close(); } +}); + +it.skipIf(process.platform === "win32")("final cleanup still retires the old runner when replacement startup failed", async () => { + const f = fixture(); + try { + await f.stop.forRestart(f.child, identity); + const replacement = f.root(400); + Object.assign(replacement, { exitCode: 1 }); + await f.stop.watch(replacement); + await f.stop.stopAll(); + expect(f.table().map(row => row.pid).sort()).toEqual([process.pid, 900].sort()); + expect(f.signals.filter(([pid]) => pid === 200)).toEqual([[200, "SIGTERM"]]); + } finally { f.close(); } +}); + + +it("normalizes only the exact Linux comm truncation, never a nearby role name", () => { + expect(diagnosticProcessKind("/tmp/paperclip-runne", "linux")).toBe("paperclip-runnerd"); + expect(diagnosticProcessKind("/tmp/paperclip-runnerd", "darwin")).toBe("paperclip-runnerd"); + expect(diagnosticProcessKind("paperclip-runne", "darwin")).toBe("other"); + for (const name of ["paperclip-runn", "paperclip-runner", "paperclip-runneX", "paperclip-runnerd-other"]) { + expect(diagnosticProcessKind(name, "linux")).toBe("other"); + } +}); + + +it.skipIf(process.platform === "win32")("still drains the retained daemon after replacement cleanup reports failure", async () => { + const f = fixture(); + try { + await f.stop.forRestart(f.child, identity); + f.replace([...f.table(), row(400, process.pid)]); + const replacement = f.root(400); await f.stop.watch(replacement); + const latestOwner = f.owners[1]!; + latestOwner.observe = async () => { throw new Error("replacement inspection failed"); }; + await expect(f.stop.stopAll()).rejects.toThrow("cleanup incomplete"); + expect(f.signals.filter(([pid]) => pid === 200)).toEqual([[200, "SIGTERM"]]); + expect(f.signals.findIndex(([pid]) => pid === 400)).toBeLessThan(f.signals.findIndex(([pid]) => pid === 200)); + } finally { f.close(); } +}); + + +it.skipIf(process.platform === "win32")("does not reuse an already-retired ordinary restart controller's old cleanup deadline", async () => { + const f = fixture(); + const clock = vi.spyOn(Date, "now").mockReturnValue(10_000); + try { + await f.stop(f.child); + clock.mockReturnValue(20_000); + f.replace([...f.table(), row(400, process.pid), row(401, 400)]); + const replacement = f.root(400); await f.stop.watch(replacement); + await f.stop.stopAll(); + expect(f.signals.filter(([pid]) => pid === 401)).toEqual([[401, "SIGTERM"]]); + } finally { clock.mockRestore(); f.close(); } +}); diff --git a/tests/runner-e2e/server-stop.ts b/tests/runner-e2e/server-stop.ts index bbeddc734c1..bcd6ff59c74 100644 --- a/tests/runner-e2e/server-stop.ts +++ b/tests/runner-e2e/server-stop.ts @@ -1,5 +1,5 @@ import type { ChildProcess } from "node:child_process"; -import { createProcessTreeOwner, incompleteTreeFallback } from "./process-tree-owner.js"; +import { createProcessTreeOwner, incompleteTreeFallback, type RestartRunnerIdentity } from "./process-tree-owner.js"; // The wrapper receives Playwright's group signal; only it signals Paperclip. export const runnerE2EServerDetached = process.platform !== "win32"; @@ -16,6 +16,10 @@ export function createRunnerE2EServerStopper(options: { type State = { owner: ReturnType; promise?: Promise; + restartPromise?: Promise; + finalRequested?: boolean; + finalPhaseStarted?: boolean; + finalComplete?: boolean; gracefulDeadline?: number; forcedDeadline?: number; gracefulSent: boolean; @@ -23,16 +27,21 @@ export function createRunnerE2EServerStopper(options: { escalationLogged: boolean; }; const states = new WeakMap(); + const watched = new Set(); + let finalGraceDeadline: number | undefined; + let finalForcedDeadline: number | undefined; + let finalAllPromise: Promise | undefined; const exited = (child: ChildProcess) => child.exitCode !== null || child.signalCode !== null || options.hasSpawnError(child); function watch(child: ChildProcess) { let state = states.get(child); if (!state) { state = { owner: (options.createOwner ?? createProcessTreeOwner)(child), gracefulSent: false, groupSignals: new Set(), escalationLogged: false }; states.set(child, state); + watched.add(child); } return state; } - async function stopOnce(child: ChildProcess, state: State, signal: NodeJS.Signals) { + async function stopOnce(child: ChildProcess, state: State, signal: NodeJS.Signals, restarting = false) { state.gracefulDeadline ??= Date.now() + options.gracefulTimeoutMs; await state.owner.observe(); if (!exited(child) && !state.gracefulSent) { @@ -40,22 +49,28 @@ export function createRunnerE2EServerStopper(options: { } while (Date.now() < state.gracefulDeadline) { await state.owner.observe(); + if (restarting && state.finalRequested) throw new Error("Final shutdown interrupted controller restart"); if (exited(child)) { + if (restarting) { + state.owner.assertRestartRunnerAlive(); + if (!state.owner.restartCleanupGroups().size) return; + } if (!state.owner.liveGroups().length) { state.owner.stopObserving(); return; } // The leader is gone. Retire only descendants whose start identities // we observed while they belonged to this server, including private groups. - const unsignaled = new Set(state.owner.liveGroups().map(group => group.processGroupId).filter(pid => !state.groupSignals.has(pid))); + const unsignaled = new Set((restarting ? [...state.owner.restartCleanupGroups()] : state.owner.liveGroups().map(group => group.processGroupId)).filter(pid => !state.groupSignals.has(pid))); for (const pid of await state.owner.signal("SIGTERM", unsignaled)) state.groupSignals.add(pid); } await wait(25); } + if (restarting) throw new Error("Controller restart did not retire non-durable children within its graceful deadline"); if (!state.escalationLogged) { options.log(`\nPaperclip did not stop within ${options.gracefulTimeoutMs}ms; sending SIGKILL\n`); state.escalationLogged = true; } if (runnerE2EServerDetached) await state.owner.signal("SIGKILL"); else if (!exited(child)) child.kill("SIGKILL"); - state.forcedDeadline ??= Date.now() + options.forcedTimeoutMs; + state.forcedDeadline ??= finalForcedDeadline ?? Date.now() + options.forcedTimeoutMs; do { await state.owner.observe(); if (exited(child) && !state.owner.liveGroups().length) { state.owner.stopObserving(); return; } @@ -66,7 +81,19 @@ export function createRunnerE2EServerStopper(options: { const stop = (child: ChildProcess, signal: NodeJS.Signals = "SIGTERM"): Promise => { options.markExpectedStop(child); const state = watch(child); - state.promise ??= Promise.resolve().then(() => stopOnce(child, state, signal)).catch(async error => { + state.finalRequested = true; + state.promise ??= Promise.resolve().then(async () => { + if (state.restartPromise && !state.finalPhaseStarted) { + await state.restartPromise.catch(() => {}); + state.finalPhaseStarted = true; + // A completed restart is a separate phase, not elapsed cleanup time. + state.gracefulDeadline = finalGraceDeadline ?? Date.now() + options.gracefulTimeoutMs; + state.forcedDeadline = undefined; + state.groupSignals.clear(); + } + await stopOnce(child, state, signal); + state.finalComplete = true; + }).catch(async error => { try { await incompleteTreeFallback(error, child, state, state.gracefulDeadline!, options.forcedTimeoutMs); } finally { @@ -77,5 +104,39 @@ export function createRunnerE2EServerStopper(options: { repeating the graceful signal. */ return state.promise; }; - return Object.assign(stop, { watch: (child: ChildProcess) => watch(child).owner.observe() }); + return Object.assign(stop, { + watch: (child: ChildProcess) => watch(child).owner.observe(), + forRestart: (child: ChildProcess, identity: RestartRunnerIdentity): Promise => { + options.markExpectedStop(child); + const state = watch(child); + if (state.finalRequested || state.restartPromise) return Promise.reject(new Error("Restart already requested or final cleanup active")); + state.restartPromise = (async () => { + await state.owner.preserveRunnerForRestart(identity); + options.log(`Controller restart preserves admitted runner pid=${identity.processPid} pgid=${identity.processGroupId} started=${identity.processStartedAt}\n`); + await stopOnce(child, state, "SIGTERM", true); + })(); + return state.restartPromise; + }, + stopAll: (signal: NodeJS.Signals = "SIGTERM"): Promise => { + // Replacement controller drain precedes old daemon retirement. All + // generations share the original final-cleanup window, including retries. + finalGraceDeadline ??= Math.min(Date.now() + options.gracefulTimeoutMs, + ...[...watched].map(child => states.get(child)!).filter(state => state.finalRequested && !state.finalComplete + && (!state.restartPromise || state.finalPhaseStarted)).map(state => state.gracefulDeadline ?? Infinity)); + finalForcedDeadline ??= finalGraceDeadline + options.forcedTimeoutMs; + for (const child of watched) { + const state = states.get(child)!; + state.finalRequested = true; + state.gracefulDeadline ??= finalGraceDeadline; + } + finalAllPromise ??= Promise.resolve().then(async () => { + const failures: unknown[] = []; + for (const child of [...watched].reverse()) { + try { await stop(child, signal); } catch (error) { failures.push(error); } + } + if (failures.length) throw new AggregateError(failures, "Owned server cleanup incomplete"); + }).catch(error => { finalAllPromise = undefined; throw error; }); + return finalAllPromise; + }, + }); } diff --git a/tests/runner-e2e/server.ts b/tests/runner-e2e/server.ts index 846235401be..997c517b57d 100644 --- a/tests/runner-e2e/server.ts +++ b/tests/runner-e2e/server.ts @@ -1,3 +1,4 @@ +import { parseRestartRunnerIdentity, type RestartRunnerIdentity } from "./process-tree-owner.js"; import { usesInstalledCli, verifyInstalledCli } from "./installed-cli.js"; import { runnerMatrix } from "./catalog.js"; import { createRunnerE2EServerStopper, runnerE2EServerDetached } from "./server-stop.js"; @@ -229,6 +230,7 @@ async function waitForHealthToStop() { interface RestartRequest { requestId: string; + preserveRunner?: RestartRunnerIdentity; } async function readRestartRequest(): Promise { @@ -254,7 +256,8 @@ async function readRestartRequest(): Promise { ) { return null; } - return { requestId }; + const preserveRunner = (value as { preserveRunner?: unknown }).preserveRunner; + return { requestId, ...(preserveRunner === undefined ? {} : { preserveRunner: parseRestartRunnerIdentity(preserveRunner) }) }; } async function writeRestartAck( @@ -276,12 +279,13 @@ async function writeRestartAck( await rename(temporaryAckPath, restartAckPath); } -async function restartServer(requestId: string) { +async function restartServer({ requestId, preserveRunner }: RestartRequest) { activeRestartRequestId = requestId; appendLog(`\nRestart request ${requestId}: stopping Paperclip\n`); const previous = child; if (!previous) throw new Error("No Paperclip server is available to restart"); - await stopServer(previous); + if (preserveRunner) await stopServer.forRestart(previous, preserveRunner); + else await stopServer(previous); if (child === previous) child = null; // Do not mistake an orphaned old server for a healthy replacement. The port // must stop answering before the next launcher is allowed to start. @@ -331,13 +335,12 @@ async function supervise() { const request = await readRestartRequest(); if (request && request.requestId !== lastRestartRequestId) { lastRestartRequestId = request.requestId; - await restartServer(request.requestId); + await restartServer(request); } await delay(200); } - const running = child; - if (running) await stopServer(running, shutdownSignal ?? "SIGTERM"); + await stopServer.stopAll(shutdownSignal ?? "SIGTERM"); } let exitCode = 0; @@ -356,15 +359,10 @@ try { ); } } - const running = child; - if (running) { - try { - await stopServer(running); - } catch (stopError) { - appendLog( - `Failed to stop Paperclip after supervisor failure: ${stopError instanceof Error ? stopError.message : String(stopError)}\n`, - ); - } + try { + await stopServer.stopAll(); + } catch (stopError) { + appendLog(`Failed to stop Paperclip after supervisor failure: ${stopError instanceof Error ? stopError.message : String(stopError)}\n`); } }