diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 986c65cbbf5..e5451eb7ace 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2379,6 +2379,22 @@ All of these must be true: - Each quarantined root for the scope passes the same journal and local Runner evidence checks for the same owner. +A local per-turn Runner can end the same way: the result was accepted, but its +`runner.suspend` command did not finish, so its own receipt is still +unsuspended. The next local run retires that owner with the same checks and +these differences: + +- The Runner's own receipt may be unsuspended. It may not be unreadable or + name another owner. +- Only the Runner's own lifecycle commands (`runner.suspend`, `runner.drain`, + `runner.shutdown`) may be unfinished. All other commands must be complete. +- The prior run's recorded Runner process is proven stopped by exact PID and + start time, as in restart recovery. A PID that now belongs to another + process counts as stopped. A live recorded process, a live process group, a + live recorded provider process, a missing identity, or a start time that + cannot be compared keeps the rejection, for example + `retirement_refused=runner_process_alive`. A later Retry checks again. + The provider session then follows the normal resume-or-replace policy. A resumed sandbox, a warm session, a plugin task, a restart claim, or any missing proof keeps the existing `prior_owner_terminal_state_indeterminate` @@ -3317,6 +3333,13 @@ A graceful shutdown can also suspend a restart-recovery claim whose execution has not yet reached provider dispatch. That execution has lost its authority: it leaves the run running, its budget hold held and its workspace unmarked, and the next controller adopts the same run. It does not settle the run as failed. +Shutdown closes the runtime API listener before it drains runs. A local Runner +execution that then needs the listener (for example, to give a new runner its +callback origin) detaches the same way instead of failing. It keeps its claim, +the drain suspends it, and the next controller adopts the run. A remote Runner +execution has no process evidence on the controller host, so it keeps the +same-run bootstrap retry. A listener that never started is still a startup +failure. Remote Runner processes use their authenticated connection lease as the bound on disconnected lifetime. The controller's shorter connection-recovery budget limits its own attempt; it must not terminate the remote owner while another diff --git a/doc/recoverable-disk-cleanup.md b/doc/recoverable-disk-cleanup.md index 95adcd1ac05..0a26bfc799a 100644 --- a/doc/recoverable-disk-cleanup.md +++ b/doc/recoverable-disk-cleanup.md @@ -71,6 +71,9 @@ copy is behind, the run fails with `native_event_replay_conflict` and automatic recovery stops. No files are copied to the sandbox. Resuming that copy would make Runner reject the controller's acknowledgement, and it could repeat work the controller already accepted. +When a controller is registered, the comparison uses its current live cursor. +The local marker written for external checkpoint storage does not replace that +authority. An hourly background sweep expires the copy once the session can only resume by reopening closed work. It checks at most 200 sessions and expires at most 50 per diff --git a/packages/plugins/sandbox-providers/daytona/pnpm-lock.yaml b/packages/plugins/sandbox-providers/daytona/pnpm-lock.yaml index c4483ecd951..38f45f94b93 100644 --- a/packages/plugins/sandbox-providers/daytona/pnpm-lock.yaml +++ b/packages/plugins/sandbox-providers/daytona/pnpm-lock.yaml @@ -483,7 +483,9 @@ packages: '@smithy/core@3.33.3': resolution: {integrity: sha512-CsOeKq/9kA3y6VJHt+/+VTCtBaxJ4OTFpgrjIUhPpDIKxBci1k2bJaQASF2h/ELWrulGp+t97DZ0mevfAD8idg==} engines: {node: '>=18.0.0'} - deprecated: Deprecated due to a bug in custom logger data redaction. See https://github.com/aws/aws-sdk-js-v3/issues/8393. + deprecated: |- + Deprecated due to a bug in custom logger data redaction. Upgrade to the latest version. If you used a custom logger with this + version, review your logs and rotate any credentials or other secrets. See https://github.com/aws/aws-sdk-js-v3/issues/8393. '@smithy/credential-provider-imds@4.5.2': resolution: {integrity: sha512-A9uSdn72ozbRUSit0eib0TW7nXuNPlaeM0zcGkJ+nE6tFcSDbnmtwoxbTCFBukVQcszDAyvsd7+rTduPTXpygg==} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a87f8e749f5..97cbf7b33af 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -5048,7 +5048,9 @@ packages: '@smithy/core@3.35.1': resolution: {integrity: sha512-i4YPS4B6ts7bjn7UwLnGjiZdprOvHvgGobFZsYK3GIY3E5hIqtj0rReU69BcTpGp+fvtraSNXeG1l+jtJvF55w==} engines: {node: '>=18.0.0'} - deprecated: Deprecated due to a bug in custom logger data redaction. See https://github.com/aws/aws-sdk-js-v3/issues/8393. + deprecated: |- + Deprecated due to a bug in custom logger data redaction. Upgrade to the latest version. If you used a custom logger with this + version, review your logs and rotate any credentials or other secrets. See https://github.com/aws/aws-sdk-js-v3/issues/8393. '@smithy/credential-provider-imds@4.5.2': resolution: {integrity: sha512-A9uSdn72ozbRUSit0eib0TW7nXuNPlaeM0zcGkJ+nE6tFcSDbnmtwoxbTCFBukVQcszDAyvsd7+rTduPTXpygg==} diff --git a/releases/beta/v2026.1023.0-beta.0.md b/releases/beta/v2026.1023.0-beta.0.md index ef11a77a4b2..16f05f22d4d 100644 --- a/releases/beta/v2026.1023.0-beta.0.md +++ b/releases/beta/v2026.1023.0-beta.0.md @@ -1,8 +1,8 @@ -# Paperclip v2026.1103.0 +# Paperclip v2026.1106.0 -> Released: 2026-11-03 +> Released: 2026-11-06 -Paperclip v2026.1103.0 carries 515 commits, promoted from `2026.1023.0-beta.0`. The headline is that the Paperclip Runner is now the default for new agents: new Claude, Codex, OpenCode, and Pi agents start on the native Runner on qualified macOS and Linux hosts, existing agents get a one-click "Upgrade runner" action, and GitHub Copilot joins as a native harness. Around that: project repositories that automatically become skill sources, model-backed prompt injection prevention, agent observability to Braintrust or any OTLP endpoint, a unified Google Workspace connection, bulk edits on the Agents roster, workspaces that compact and expire themselves with encrypted restore points, and Grok Bot and Muse as experimental Runner providers. Underneath, server restarts report ready sooner, the Docker image boots from a precompiled bundle, task pages load faster, and more background services stop polling the database. +Paperclip v2026.1106.0 carries 515 commits, promoted from `2026.1023.0-beta.0`. The headline is that the Paperclip Runner is now the default for new agents: new Claude, Codex, OpenCode, and Pi agents start on the native Runner on qualified macOS and Linux hosts, existing agents get a one-click "Upgrade runner" action, and GitHub Copilot joins as a native harness. Around that: project repositories that automatically become skill sources, model-backed prompt injection prevention, agent observability to Braintrust or any OTLP endpoint, a unified Google Workspace connection, bulk edits on the Agents roster, workspaces that compact and expire themselves with encrypted restore points, and Grok Bot and Muse as experimental Runner providers. Underneath, server restarts report ready sooner, the Docker image boots from a precompiled bundle, task pages load faster, and more background services stop polling the database. ## Breaking Changes diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index 892cb063735..654f152bcf3 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { authUsers, budgetReservations, companyMemberships, completionContracts, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; import { expect, it, vi } from "vitest"; +import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; import { currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.js"; @@ -77,6 +78,56 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut 90_000, ); + it("hands a recovery to the next controller when shutdown closed the listener before backend setup", async () => { + const f = await seedRecoverableRun("LISTENER-CLOSED"); + const listener = fixture.current.server!; + const address = listener.address(); + if (!address || typeof address === "string") throw new Error("Expected a TCP runtime API listener"); + let listenerClosed: Promise | undefined; + // Graceful shutdown closes HTTP admission before it drains heartbeat runs. + // Backend setup then asks for the callback origin, as runnerd setup does. + const dying = vi.fn(() => ({ + descriptor: async () => { resolveLocalRuntimeApiUrl(); throw new Error("unreachable"); }, + openSession: async () => { resolveLocalRuntimeApiUrl(); throw new Error("unreachable"); }, + })); + const heartbeat = heartbeatService(f.db, { + nativeSessionBackendFactory: dying as never, + beforeNativeRuntimeSelection: async () => { + listener.closeIdleConnections(); + listenerClosed = new Promise(resolve => listener.close(() => resolve())); + }, + }); + let drained: unknown; + try { + try { + const { claims } = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId })]); + } finally { await heartbeat.drainActiveRunExecutions(); } + expect(dying).toHaveBeenCalled(); + drained = await heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + await heartbeat.drainActiveRunExecutions(); + } finally { + await listenerClosed; + await new Promise(resolve => listener.listen(address.port, "127.0.0.1", resolve)); + } + expect(drained).toMatchObject({ restartSuspendedRunIds: [f.runId] }); + const [run] = await f.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect({ status: run.status, errorCode: run.errorCode, finishedAt: run.finishedAt }).toEqual({ + status: "running", errorCode: null, finishedAt: null }); + expect(owner).toMatchObject({ phase: "observed", resultId: null, recoveryState: "awaiting_runner_reattach" }); + + await f.db.update(nativeRunFinalizations).set({ controllerPid: 2_000_000_001 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const resumed = vi.fn(() => { throw new Error("next boot reached provider dispatch"); }); + const next = heartbeatService(f.db, { nativeSessionBackendFactory: resumed }); + try { + const { claims } = await next.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId })]); + } finally { await next.drainActiveRunExecutions(); } + expect(resumed).toHaveBeenCalledOnce(); + }, 90_000); + it("still settles a recovery whose claim merely expired without a shutdown suspension", async () => { const f = await seedRecoverableRun("EXPIRED-NOT-SUSPENDED"); const dying = vi.fn(() => { throw new Error("An expired claim must not open a provider session"); }); diff --git a/server/src/__tests__/helpers/native-restart.ts b/server/src/__tests__/helpers/native-restart.ts index deccd592af1..8d642f4ca12 100644 --- a/server/src/__tests__/helpers/native-restart.ts +++ b/server/src/__tests__/helpers/native-restart.ts @@ -244,7 +244,7 @@ export function nativeRestartFixture() { }; } - return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor }; } }; + return { seedRun, transportOptions, get current() { return { temporary, runtimeRoot, paperclipHome, apiUrl, companyId, agentId, successor, server }; } }; } export function remotePreparedWorkspace() { diff --git a/server/src/http/server.test.ts b/server/src/http/server.test.ts index b548858d23f..c4616ec6751 100644 --- a/server/src/http/server.test.ts +++ b/server/src/http/server.test.ts @@ -8,7 +8,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import express from "express"; import { DurablePrpControlPlane } from "../vendor/paperclip-runner/index.js"; import { registerRunnerPrpAuthority, runnerPrpWebSocketInternals } from "../realtime/runner-prp-ws.js"; -import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; +import { LocalRuntimeApiListenerClosedError, resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { createPaperclipHttpServer } from "./server.js"; import { privateHostnameGuard } from "../middleware/private-hostname-guard.js"; @@ -206,6 +206,30 @@ describe("Paperclip HTTP server runner transport", () => { } }); + it("distinguishes a listener closed for shutdown from one that never served", async () => { + expect(() => resolveLocalRuntimeApiUrl()).toThrow("local_runtime_api_listener_unavailable"); + expect(() => resolveLocalRuntimeApiUrl()).not.toThrow(LocalRuntimeApiListenerClosedError); + const server = createPaperclipHttpServer((_req, res) => res.end("closing"), { apiUrl: "http://ignored.example.test" }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const client = await fetch(resolveLocalRuntimeApiUrl(), { keepalive: true }); + // `close()` stops admission at once; its close event waits for open requests. + const closed = new Promise(resolve => server.close(() => resolve())); + expect(() => resolveLocalRuntimeApiUrl()).toThrow(LocalRuntimeApiListenerClosedError); + await client.text(); + server.closeAllConnections(); + await closed; + expect(() => resolveLocalRuntimeApiUrl()).toThrow(LocalRuntimeApiListenerClosedError); + expect(() => resolveLocalRuntimeApiUrl()).toThrow("local_runtime_api_listener_unavailable"); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + try { + expect(resolveLocalRuntimeApiUrl()).toMatch(/^http:\/\/127\.0\.0\.1:\d+$/); + } finally { + await new Promise(resolve => server.close(() => resolve())); + } + }); + it.each(["127.0.0.1", "::1", "::ffff:127.0.0.1"])("routes a guarded native read through the real %s listener", async host => { await guardedNativeRead(host); }); diff --git a/server/src/runtime-api.ts b/server/src/runtime-api.ts index 4c4f87868a6..b7d1a3d64f4 100644 --- a/server/src/runtime-api.ts +++ b/server/src/runtime-api.ts @@ -4,15 +4,37 @@ import { isIP } from "node:net"; const runtimeApiListeners = new Set(); const registeredRuntimeApiListeners = new WeakSet(); +let runtimeApiListenerClosed = false; + +/** + * The process had a runtime API listener and has since closed it, which only + * happens while it shuts down. Work that needs a callback origin must hand off + * to the next controller instead of failing as a provider or startup error. + */ +export class LocalRuntimeApiListenerClosedError extends Error { + constructor() { + // Keep the established failure code for classifiers and persisted rows. + super("local_runtime_api_listener_unavailable"); + this.name = "LocalRuntimeApiListenerClosedError"; + } +} /** Track listener ownership without treating configured or advertised URLs as authority. */ export function registerLocalRuntimeApiServer(server: Server): void { if (!registeredRuntimeApiListeners.has(server)) { registeredRuntimeApiListeners.add(server); - server.on("listening", () => runtimeApiListeners.add(server)); - server.on("close", () => runtimeApiListeners.delete(server)); + server.on("listening", () => { + runtimeApiListeners.add(server); + runtimeApiListenerClosed = false; + }); + server.on("close", () => { + if (runtimeApiListeners.delete(server)) runtimeApiListenerClosed = true; + }); + } + if (server.listening) { + runtimeApiListeners.add(server); + runtimeApiListenerClosed = false; } - if (server.listening) runtimeApiListeners.add(server); } function normalizeHost(value: string | null | undefined): string { @@ -94,6 +116,11 @@ export function chooseLocalRuntimeApiUrl(input: { export function resolveLocalRuntimeApiUrl(): string { const listeners = [...runtimeApiListeners].filter(server => server.listening); + // `server.close()` clears `listening` at once; its close event waits for + // admitted requests. Either state means this process stopped serving. + if (listeners.length === 0 && (runtimeApiListenerClosed || runtimeApiListeners.size > 0)) { + throw new LocalRuntimeApiListenerClosedError(); + } if (listeners.length !== 1) { throw new Error(listeners.length ? "local_runtime_api_listener_ambiguous" : "local_runtime_api_listener_unavailable"); } @@ -103,7 +130,10 @@ export function resolveLocalRuntimeApiUrl(): string { } export const runtimeApiInternals = { - resetForTests(): void { runtimeApiListeners.clear(); }, + resetForTests(): void { + runtimeApiListeners.clear(); + runtimeApiListenerClosed = false; + }, }; export function choosePrimaryRuntimeApiUrl(input: { diff --git a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts index f12aad1871f..e969b20d0c2 100644 --- a/server/src/services/native-runtime/native-recovered-owner-followup.test.ts +++ b/server/src/services/native-runtime/native-recovered-owner-followup.test.ts @@ -25,7 +25,7 @@ import { import { nativeRuntimeContextFixture } from "./runtime-context.test-fixture.js"; const toolDefinitions = vi.hoisted(() => vi.fn()); -vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" })); +vi.mock("../../runtime-api.js", async importOriginal => ({ ...await importOriginal(), resolveLocalRuntimeApiUrl: () => "http://127.0.0.1:43219" })); vi.mock("./paperclip-runner-tool-authority.js", () => ({ PaperclipRunnerToolAuthority: class { readonly binding = {}; @@ -185,7 +185,11 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered runner: { execute: vi.fn(async () => { throw new Error("admission must not reach the sandbox"); }) }, sandboxLeaseAcquisition: { outcome, providerLeaseId: `sandbox-${runId}` } } as never, }).then(() => "constructed", (error: Error) => error.message); - return { runId, logs, start, selectedRunId: selected?.id ?? null }; + // A local Runner keeps its controller root on this host between turns. + const startLocal = () => createRunnerdBackend({ + db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, + }).then(() => "constructed", (error: Error) => error.message); + return { runId, logs, start, startLocal, selectedRunId: selected?.id ?? null }; } async function quarantined() { @@ -465,6 +469,104 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); }, 60_000); + /** A spawned process that stands in for the prior local Runner. */ + async function runnerProcess() { + const child = spawn(process.execPath, ["-e", "setTimeout(() => {}, 60_000)"], { stdio: "ignore" }); + await new Promise(resolve => child.once("spawn", resolve)); + const startedAt = await readProcessStartedAt(child.pid!); + expect(startedAt).toBeTruthy(); + return { child, pid: child.pid!, processStartedAt: new Date(startedAt!) }; + } + + async function stoppedRunnerProcess() { + const runner = await runnerProcess(); + const exited = new Promise(resolve => runner.child.once("exit", resolve)); + runner.child.kill("SIGKILL"); + await exited; + return runner; + } + + /** The state a local per-turn Runner leaves when its result was accepted but + * its suspend command never finished: the controller journal holds a pending + * runner.suspend and the Runner's own receipt is still `ready`. */ + async function seedUnsuspendedLocalOwner(runner: { pid: number; processStartedAt: Date }) { + const prior = await seedRecoveredPriorRun("pending-command"); + await db.update(heartbeatRuns).set({ processPid: runner.pid, processGroupId: runner.pid, processStartedAt: runner.processStartedAt }) + .where(eq(heartbeatRuns.id, prior.runId)); + const followUp = await admitFollowUp(prior); + // Learn the scope key from a held attempt, then keep its root in place. + expect(await followUp.start()).toMatch(/^runner_state_identity_mismatch: prior_owner_terminal_state_indeterminate/); + const [copy] = await quarantined(); + const root = join(stateDirectory, copy!.split(".")[0]!); + await rename(join(stateDirectory, "quarantine", copy!), root); + await rm(join(stateDirectory, "quarantine"), { recursive: true, force: true }); + await mkdir(join(root, "runner"), { recursive: true }); + await writeFile(join(root, "runner", "runner-state.json"), JSON.stringify({ + schema: "paperclip.runner.durable.state.v1", lifecycle: "ready", ...prior.identity })); + return { prior, followUp, root }; + } + + it("retires a finished local owner whose Runner stopped before it proved suspension", async () => { + const { prior, followUp, root } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + expect(await followUp.startLocal()).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(existsSync(join(root, "runner"))).toBe(false); + expect(await archivedOwners(root)).toEqual([prior.runId]); + expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); + }, 60_000); + + it("keeps a finished local owner held while its recorded Runner process is alive", async () => { + const runner = await runnerProcess(); + try { + const { followUp } = await seedUnsuspendedLocalOwner(runner); + expect(await followUp.startLocal()).toBe(stranded("runner_process_alive")); + expect(followUp.logs.join("")).not.toContain("finished without a saved runner checkpoint"); + } finally { + runner.child.kill("SIGKILL"); + } + }, 60_000); + + it.each(["no identity", "no process group"] as const)("keeps a finished local owner held with %s", async scenario => { + const { prior, followUp } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + await db.update(heartbeatRuns).set(scenario === "no identity" + ? { processPid: null, processGroupId: null, processStartedAt: null } : { processGroupId: null }) + .where(eq(heartbeatRuns.id, prior.runId)); + expect(await followUp.startLocal()).toBe(stranded("runner_process_identity_missing")); + }, 60_000); + + it("retires a finished local owner from its saved stop receipt after settlement cleared the process fields", async () => { + const runner = await stoppedRunnerProcess(); + const { prior, followUp, root } = await seedUnsuspendedLocalOwner(runner); + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: prior.runId, seq: 900, eventType: "native.local_process_stopped", + stream: "system", payload: { processPid: runner.pid, processGroupId: runner.pid, processStartedAt: runner.processStartedAt.toISOString() } }); + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, prior.runId)); + expect(await followUp.startLocal()).toBe(ADMITTED); + expect(await archivedOwners(root)).toEqual([prior.runId]); + }, 60_000); + + it("keeps a stopped local owner held while a provider process from its run events is alive", async () => { + const provider = await runnerProcess(); + try { + const { prior, followUp } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + await db.insert(heartbeatRunEvents).values({ companyId, agentId, runId: prior.runId, seq: 901, eventType: "provider.event", + stream: "system", payload: { prpEvent: { payload: { processId: provider.pid, + processStartedAt: provider.processStartedAt.toISOString() } } } }); + expect(await followUp.startLocal()).toBe(stranded("provider_process_alive")); + } finally { + provider.child.kill("SIGKILL"); + } + }, 60_000); + + it("keeps a stopped local owner held by an unfinished non-lifecycle command", async () => { + const { followUp, root } = await seedUnsuspendedLocalOwner(await stoppedRunnerProcess()); + const path = join(root, "control-plane", "control-plane-state.json"); + const state = JSON.parse(await readFile(path, "utf8")); + state.commands.find((command: { type: string }) => command.type === "turn.start").status = "pending"; + await writeFile(path, JSON.stringify(state)); + expect(await followUp.startLocal()).toBe(stranded("journal_commands_unsettled(turn.start.pending)")); + }, 60_000); + it.each(["uncommitted result", "committed without result", "live controller lease", "unexpired controller lease", "recovery pending", "pending command", "unanswered tool input", "foreign quarantine"] as const)("keeps the prior owner held with %s", async scenario => { const prior = await seedRecoveredPriorRun(scenario === "pending command" ? "pending-command" diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index fd79245b6e5..52d8d15c9b7 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -5,8 +5,48 @@ import { evaluateNativeControllerTakeover, evaluateNativeProviderProcesses, nextNativeProviderAttempt, + proveNativeLocalRunnerStopped, } from "./native-restart-recovery.js"; +describe("local runner stop proof", () => { + const startedAt = new Date("2026-11-06T17:58:54.000Z"); + const base = { processPid: 72454, processGroupId: 72454, processStartedAt: startedAt }; + const probe = (alive: number[], starts: Record, groups: number[] = []) => ({ + isProcessAlive: (pid: number) => alive.includes(pid), + isProcessGroupAlive: (group: number) => groups.includes(group), + readProcessStartedAt: async (pid: number) => starts[pid] ?? null, + }); + + it.each([ + ["the runner and its group exited", base, probe([], {}), { stopped: true }], + ["the runner PID now belongs to another process", base, + probe([72454], { 72454: new Date(startedAt.getTime() + 60_000) }), { stopped: true }], + ["the recorded runner is alive", base, probe([72454], { 72454: startedAt }), + { stopped: false, reason: "runner_process_alive" }], + ["the live PID start time is unknown", base, probe([72454], {}), + { stopped: false, reason: "runner_process_identity_unverifiable" }], + ["the runner group is alive", base, probe([], {}, [72454]), + { stopped: false, reason: "runner_process_group_alive" }], + ["no runner PID was recorded", { ...base, processPid: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ["no runner start time was recorded", { ...base, processStartedAt: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ["no runner process group was recorded", { ...base, processGroupId: null }, probe([], {}), + { stopped: false, reason: "runner_process_identity_missing" }], + ["a stop receipt proved the runner absent", { ...base, processStartedAt: null, runnerStopRecorded: true }, + probe([], {}), { stopped: true }], + ] as const)("decides when %s", async (_case, identity, checks, expected) => { + await expect(proveNativeLocalRunnerStopped({ ...identity, ...checks })).resolves.toEqual(expected); + }); + + it("refuses while a recorded provider process is alive", async () => { + const providerStartedAt = new Date("2026-11-06T17:58:56.000Z"); + await expect(proveNativeLocalRunnerStopped({ ...base, + providerProcesses: [{ pid: 72460, processStartedAt: providerStartedAt }], + ...probe([72460], { 72460: providerStartedAt }) })).resolves.toEqual({ stopped: false, reason: "provider_process_alive" }); + }); +}); + describe("native restart recovery classification", () => { it.each([false, true])("recognizes verified staged remote evidence with target snapshot=%s", hasCheckpoint => { const evidence = { remoteSandbox: true, runnerPidAlive: false, runnerGroupAlive: false, diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index d750a83a141..089d421450e 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -336,6 +336,118 @@ export async function evaluateNativeProviderProcesses(input: { }; } +/** Run event types that can carry a provider process identity. */ +export const NATIVE_PROVIDER_PROCESS_EVENT_TYPES = [ + "harness.ready", + "session.started", + "session.resumed", + "session.updated", + "turn.started", + "provider.event", + "provider.rpc_result", +] as const; + +function parsedProcessStart(value: unknown): Date | null { + const parsed = typeof value === "string" ? new Date(value) : null; + return parsed && !Number.isNaN(parsed.getTime()) ? parsed : null; +} + +/** Provider process identities from a saved checkpoint and recent run events. */ +export function nativeProviderProcessIdentities( + checkpoint: Record, + providerEvents: ReadonlyArray<{ payload: Record | null }>, +): NativeProviderProcessIdentity[] { + const identities: NativeProviderProcessIdentity[] = []; + const checkpointProcess = + checkpoint.process && typeof checkpoint.process === "object" && !Array.isArray(checkpoint.process) + ? (checkpoint.process as Record) + : {}; + for (const [field, startedAtField] of [ + ["providerPid", "providerProcessStartedAt"], + ["codexPid", "codexProcessStartedAt"], + ["sidecarPid", "sidecarProcessStartedAt"], + ["agentPid", "agentProcessStartedAt"], + ] as const) { + const value = checkpointProcess[field]; + if (typeof value === "number" && Number.isInteger(value) && value > 0) { + identities.push({ pid: value, processStartedAt: parsedProcessStart(checkpointProcess[startedAtField]) }); + } + } + for (const providerEvent of providerEvents) { + const prpEvent = providerEvent.payload?.prpEvent; + const event = prpEvent && typeof prpEvent === "object" && !Array.isArray(prpEvent) + ? (prpEvent as Record) : {}; + const eventPayload = event.payload && typeof event.payload === "object" && !Array.isArray(event.payload) + ? (event.payload as Record) : {}; + const processId = eventPayload.processId; + if (typeof processId === "number" && Number.isInteger(processId) && processId > 0) { + identities.push({ pid: processId, + processStartedAt: parsedProcessStart(eventPayload.processStartedAt ?? eventPayload.providerProcessStartedAt) }); + } + } + return identities; +} + +export type NativeLocalRunnerStopProof = + | { stopped: true } + | { + stopped: false; + reason: + | "runner_process_identity_missing" + | "runner_process_alive" + | "runner_process_identity_unverifiable" + | "runner_process_group_alive" + | "provider_process_alive" + | "provider_process_identity_unverifiable"; + }; + +/** + * Proves that a finished local runner and its recorded provider processes + * stopped, with the same exact PID and start-time checks as restart recovery. + * A PID that now belongs to a different process counts as stopped. A live + * process with the recorded identity, a live process group, or a live PID + * whose start time cannot be compared is not proof. + */ +export async function proveNativeLocalRunnerStopped(input: { + processPid: number | null; + processGroupId: number | null; + processStartedAt: Date | null; + providerProcesses?: readonly NativeProviderProcessIdentity[]; + /** A current, server-authored stop receipt already proved this exact Runner and group absent. */ + runnerStopRecorded?: boolean; + isProcessAlive?: (pid: number) => boolean; + isProcessGroupAlive?: (processGroupId: number) => boolean; + readProcessStartedAt?: (pid: number) => Promise; +}): Promise { + const refuse = (reason: Extract["reason"]) => + ({ stopped: false as const, reason }); + // A local Runner leads its own process group. Without both identities, a + // live child in that group could outlive the Runner unseen. + if (!input.processPid || !Number.isInteger(input.processPid) || input.processPid <= 1 || !input.processGroupId + || (!input.processStartedAt && !input.runnerStopRecorded)) { + return refuse("runner_process_identity_missing"); + } + if (!input.runnerStopRecorded) { + const runner = await evaluateNativeProviderProcesses({ + identities: [{ pid: input.processPid, processStartedAt: input.processStartedAt }], + isProcessAlive: input.isProcessAlive, + readProcessStartedAt: input.readProcessStartedAt, + }); + if (runner.livePids.length > 0) return refuse("runner_process_alive"); + if (runner.ambiguousLivePids.length > 0) return refuse("runner_process_identity_unverifiable"); + } + const groupAlive = input.isProcessGroupAlive ?? processGroupIsAlive; + if (groupAlive(input.processGroupId)) return refuse("runner_process_group_alive"); + const providers = await evaluateNativeProviderProcesses({ + identities: (input.providerProcesses ?? []).filter(identity => identity.pid !== input.processPid), + isProcessAlive: input.isProcessAlive, + readProcessStartedAt: input.readProcessStartedAt, + }); + if (providers.livePids.length > 0) return refuse("provider_process_alive"); + if (providers.ambiguousLivePids.length > 0) return refuse("provider_process_identity_unverifiable"); + return { stopped: true }; +} + export function classifyNativeRunnerRecoveryEvidence(input: { runnerPidAlive: boolean; runnerGroupAlive: boolean; @@ -898,79 +1010,12 @@ export async function claimNativeRestartRecoveries(input: { .where( and( eq(heartbeatRunEvents.runId, row.run.id), - inArray(heartbeatRunEvents.eventType, [ - "harness.ready", - "session.started", - "session.resumed", - "session.updated", - "turn.started", - "provider.event", - "provider.rpc_result", - ]), + inArray(heartbeatRunEvents.eventType, [...NATIVE_PROVIDER_PROCESS_EVENT_TYPES]), ), ) .orderBy(desc(heartbeatRunEvents.id)) .limit(100); - const checkpointProcess = - checkpointRecord.process && - typeof checkpointRecord.process === "object" && - !Array.isArray(checkpointRecord.process) - ? (checkpointRecord.process as Record) - : {}; - const providerProcessIdentities: NativeProviderProcessIdentity[] = []; - const checkpointProcessFields = [ - ["providerPid", "providerProcessStartedAt"], - ["codexPid", "codexProcessStartedAt"], - ["sidecarPid", "sidecarProcessStartedAt"], - ["agentPid", "agentProcessStartedAt"], - ] as const; - for (const [field, startedAtField] of checkpointProcessFields) { - const value = checkpointProcess[field]; - if (typeof value === "number" && Number.isInteger(value) && value > 0) { - const rawStartedAt = checkpointProcess[startedAtField]; - const parsedStartedAt = - typeof rawStartedAt === "string" ? new Date(rawStartedAt) : null; - providerProcessIdentities.push({ - pid: value, - processStartedAt: - parsedStartedAt && !Number.isNaN(parsedStartedAt.getTime()) - ? parsedStartedAt - : null, - }); - } - } - for (const providerEvent of providerEvents) { - const prpEvent = providerEvent.payload?.prpEvent; - const event = - prpEvent && typeof prpEvent === "object" && !Array.isArray(prpEvent) - ? (prpEvent as Record) - : {}; - const eventPayload = - event.payload && - typeof event.payload === "object" && - !Array.isArray(event.payload) - ? (event.payload as Record) - : {}; - const processId = eventPayload.processId; - if ( - typeof processId === "number" && - Number.isInteger(processId) && - processId > 0 - ) { - const rawStartedAt = - eventPayload.processStartedAt ?? - eventPayload.providerProcessStartedAt; - const parsedStartedAt = - typeof rawStartedAt === "string" ? new Date(rawStartedAt) : null; - providerProcessIdentities.push({ - pid: processId, - processStartedAt: - parsedStartedAt && !Number.isNaN(parsedStartedAt.getTime()) - ? parsedStartedAt - : null, - }); - } - } + const providerProcessIdentities = nativeProviderProcessIdentities(checkpointRecord, providerEvents); const providerProcesses = await evaluateNativeProviderProcesses({ identities: (remoteProcess ? [] : providerProcessIdentities).filter( (identity) => identity.pid !== row.run.processPid, diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index d3e76df91ba..62d420874d0 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -275,7 +275,7 @@ const state = vi.hoisted(() => ({ release: null as null | (() => void), })); -vi.mock("../../runtime-api.js", () => ({ resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl })); +vi.mock("../../runtime-api.js", async importOriginal => ({ ...await importOriginal(), resolveLocalRuntimeApiUrl: state.localRuntimeApiUrl })); const accountingEvents = vi.hoisted(() => ({ committed: undefined as undefined | ((event: PrpEvent) => Promise), duplicate: undefined as undefined | ((event: PrpEvent) => Promise) })); vi.mock("./paperclip-control-plane-port.js", async importOriginal => { @@ -18090,10 +18090,12 @@ describe("runnerd provider runtime wiring", () => { }); it.each([ - { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v1" }, - { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v2" }, - { backup: "current", nextSourceSeq: 1518, controlSchema: "paperclip.runner.durable.control-plane-state.v2" }, - ])("checks the controller cursor before restoring a $backup failover backup ($controlSchema)", async ({ backup, nextSourceSeq, controlSchema }) => { + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v1", registered: false }, + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: false }, + { backup: "current", nextSourceSeq: 1518, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: false }, + { backup: "stale", nextSourceSeq: 120, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: true }, + { backup: "current", nextSourceSeq: 1518, controlSchema: "paperclip.runner.durable.control-plane-state.v2", registered: true }, + ])("checks the controller cursor before restoring a $backup failover backup ($controlSchema, registered=$registered)", async ({ backup, nextSourceSeq, controlSchema, registered }) => { const remoteCwd = join(isolatedStateDirectory, "remote"); await mkdir(join(remoteCwd, ".paperclip-runtime", "paperclip-runner"), { recursive: true }); const syncIn = vi.fn(async () => undefined); @@ -18114,14 +18116,29 @@ describe("runnerd provider runtime wiring", () => { }; return { exitCode: 0, timedOut: false, stdout: "", stderr: "" }; }); + const checkpointSink = vi.fn(async () => undefined); + const checkpointRegistration = registered ? vi.spyOn(runnerCheckpoint, "nativeRunnerCheckpointRegistration") + .mockReturnValue(async authority => { + authority.setCheckpointPersistence(checkpointSink); + await authority.flushCheckpoint(); + }) : undefined; + let controller: DurablePrpControlPlane | undefined; + let registration: Awaited>> | undefined; + onTestFinished(async () => { await registration?.release(); await controller?.stop(); checkpointRegistration?.mockRestore(); }); + const controllerArtifact = join(isolatedStateDirectory, "failover-cursor-controller"); + if (registered) await writeFile(controllerArtifact, "fixture bootstrap artifact"); const runnerInstanceId = "runner-failover-cursor"; await createRunnerdBackend({ db: leaseDb(execution), execution, runnerInstanceId, + ...(registered ? { controllerPersistence: { bootId: "failover-cursor-boot" }, + runnerRemoteBinaryPath: controllerArtifact, runnerRemoteCodexPath: controllerArtifact } : {}), runnerIngressAuthorized: true, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd, environmentId: "environment", leaseId: "lease-replacement", providerKey: "daytona", effectiveCapabilities: { runnerWebSocketIngress: true }, + getRunnerIngressEndpoint: async () => ({ kind: "authenticated_websocket", websocketUrl: "wss://sandbox.invalid/runner", + secretHeaders: [{ name: "Authorization", value: "fixture-ingress" }], generation: "1", refresh: vi.fn(), close: vi.fn(async () => undefined) }), sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "sandbox-replacement", reason: "not_found" }, runner: { execute: remoteExecute, syncIn }, } as never, @@ -18129,10 +18146,11 @@ describe("runnerd provider runtime wiring", () => { state.createBackend.mock.calls.at(-1)![1].codexTransportFactory!(); const options = state.createTransport.mock.calls.at(-1)![0] as RunnerTransportOptions & { prepareExternalRunnerState: () => Promise; + controlPlaneRegistration: NonNullable; }; const root = options.stateDirectory!; await mkdir(join(root, "control-plane"), { recursive: true }); - await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ + if (!registered) await writeFile(join(root, "control-plane", "control-plane-state.json"), JSON.stringify({ schema: controlSchema, identity: options.prpIdentity, ackedSourceSeq: 1517, connectionCount: 1, committedEvents: [], commands: [], })); @@ -18150,6 +18168,20 @@ describe("runnerd provider runtime wiring", () => { sourceProviderLeaseId: "sandbox-previous", }))); + if (registered) { + controller = new DurablePrpControlPlane({ stateDirectory: join(root, "control-plane"), + identity: options.prpIdentity! as ConstructorParameters[0]["identity"], + expectedRunnerVersion: "failover-cursor-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + controller.store.state.ackedSourceSeq = 1517; + await controller.flushCheckpoint(); + registration = await options.controlPlaneRegistration(controller); + expect(checkpointRegistration).toHaveBeenCalled(); + expect(checkpointSink).toHaveBeenCalled(); + const marker = JSON.parse(await readFile(join(root, "control-plane", "control-plane-state.json"), "utf8")); + expect(marker.schema).toBe("paperclip.runner.external-control-plane-state.v1"); + expect(marker.ackedSourceSeq).toBeUndefined(); + syncIn.mockClear(); + } const prepared = options.prepareExternalRunnerState(); if (backup === "stale") { await expect(prepared).rejects.toThrow( diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0e8b4a60a79..4cbadac50c5 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5,7 +5,7 @@ import { logger } from "../../middleware/logger.js"; import { canRestoreMigratedNativeSession } from "./native-workspace-migration.js"; import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES, nativeHarnessBackupManifestSha256, retiredSourceAuthorizesNativeBackupRestore } from "./native-retired-backup-source.js"; import { prepareCursorCloudRuntime } from "./cursor-cloud-runtime.js"; -import { resolveLocalRuntimeApiUrl } from "../../runtime-api.js"; +import { LocalRuntimeApiListenerClosedError, resolveLocalRuntimeApiUrl } from "../../runtime-api.js"; import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js"; import { readProviderFailure } from "@paperclipai/shared"; import { nativeCancellationHasUnsettledOwner, notifyNativeCancellation, waitForNativeCancellation } from "./native-cancellation-delivery.js"; @@ -292,6 +292,10 @@ import { import { currentNativeControllerIdentity, evaluateNativeControllerTakeover, + proveNativeLocalRunnerStopped, + nativeProviderProcessIdentities, + NATIVE_PROVIDER_PROCESS_EVENT_TYPES, + type NativeLocalRunnerStopProof, isCurrentNativeControllerProcess, nextNativeProviderAttempt, type NativeControllerIdentity, @@ -5286,15 +5290,25 @@ function moveNativeContinuityState(root: string, archiveRoot: string): void { * Runner outbox can hold only post-terminal telemetry. Local lifecycle * evidence, even an unsuspended receipt, and unknown entries are owner * claims that this proof never overrides. + * + * `localRunner` names a local Runner that wrote its own lifecycle receipt in + * this root. Its unsuspended receipt and its own unfinished lifecycle commands + * (suspend, drain, shutdown) are claims of that one process. The caller must + * separately prove that process stopped before it relies on this projection. */ type SettledRunnerdProjection = { bytes: string; refused?: never } | { bytes?: never; refused: string }; -function readSettledRunnerdProjection(root: string, identity: Record): SettledRunnerdProjection { +const LOCAL_RUNNER_LIFECYCLE_COMMANDS = new Set(["runner.suspend", "runner.drain", "runner.shutdown"]); + +function readSettledRunnerdProjection(root: string, identity: Record, + options: { localRunner?: boolean } = {}): SettledRunnerdProjection { const refuse = (refused: string): SettledRunnerdProjection => ({ refused }); try { if (!isSafeNativeStateDirectory(root)) return refuse("root_unsafe"); const lifecycle = runnerdAuthorityLifecycle(root, identity as RunnerdDurableIdentity); - if (lifecycle !== "absent") return refuse(`runner_lifecycle_${lifecycle}`); + if (lifecycle !== "absent" && !(options.localRunner && lifecycle === "not_suspended")) { + return refuse(`runner_lifecycle_${lifecycle}`); + } // Every legitimate writer's entry is classified in native-session-root-entries.ts. // Held entries are known evidence that needs its own settlement; anything // else is writer drift. Name each kind separately. @@ -5306,6 +5320,11 @@ function readSettledRunnerdProjection(root: string, identity: Record !LOCAL_RUNNER_LIFECYCLE_COMMANDS.has(String(command.type)) || command.status === "completed") + : commands; + const settledControl = settledCommands === commands ? control : { ...control, commands: settledCommands }; const answered = new Set(commands .filter(command => command.type === "semantic_tool.result" && command.status === "completed") .map(command => record(command.payload).sourceEventId)); @@ -5315,11 +5334,11 @@ function readSettledRunnerdProjection(root: string, identity: Record record(control.identity)[key] !== value)) { return refuse("journal_identity_mismatch"); } - if (!isSettledRemoteWarmControlState(control, identity)) { + if (!isSettledRemoteWarmControlState(settledControl, identity)) { // Closed vocabulary only: known command types and statuses, never payloads. - const pending = commands.filter(command => command.status !== "completed") + const unsettled = settledCommands.filter(command => command.status !== "completed") .map(command => `${retirementReasonName("commandType", command.type)}.${retirementReasonName("commandStatus", command.status)}`); - return refuse(pending.length > 0 ? `journal_commands_unsettled(${[...new Set(pending)].slice(0, 4).join(",")})` + return refuse(unsettled.length > 0 ? `journal_commands_unsettled(${[...new Set(unsettled)].slice(0, 4).join(",")})` : "journal_terminal_unsettled"); } if ((control.committedEvents as unknown[]).map(record).some(event => @@ -5363,6 +5382,12 @@ function retirementReasonName(kind: keyof typeof RETIREMENT_REASON_NAMES, value: * that authority as a continuity break instead. This run then bootstraps its * own authority, and the provider session follows the ordinary * resume-or-replace policy. Any missing proof keeps the original rejection. + * + * A local per-turn Runner can end the same way: the result is accepted but its + * suspend command does not finish, so its own receipt stays unsuspended. That + * receipt is a claim of one recorded Runner process. Once that process and its + * recorded provider processes are proven stopped by exact PID and start time, + * the claim has no owner left and the same retirement applies. */ async function retireFinishedPerTurnRunnerdOwner(input: { db: Db; @@ -5377,11 +5402,14 @@ async function retireFinishedPerTurnRunnerdOwner(input: { // existing recovery path reads. Plugin tasks own a separate saved checkpoint. const refuse = (refused: string) => ({ retired: false as const, refused }); if (input.execution.session.lifecyclePolicy.mode !== "per_turn") return refuse("lifecycle_not_per_turn"); - if (target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { - return refuse("target_not_sandbox_runner"); - } - if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { - return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`); + const localRunner = !target || target.kind === "local"; + if (!localRunner) { + if (target.kind !== "remote" || target.transport !== "sandbox" || !target.runner || target.runnerTask) { + return refuse("target_not_sandbox_runner"); + } + if (!["created", "replacement"].includes(target.sandboxLeaseAcquisition?.outcome ?? "")) { + return refuse(`sandbox_${retirementReasonName("sandboxOutcome", target.sandboxLeaseAcquisition?.outcome)}`); + } } const identity: Record = {}; for (const key of ["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId", "turnId", "itemId"]) { @@ -5389,7 +5417,7 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (typeof value !== "string" || !value) return refuse("durable_identity_incomplete"); identity[key] = value; } - const projection = readSettledRunnerdProjection(input.root, identity); + const projection = readSettledRunnerdProjection(input.root, identity, { localRunner }); if (projection.refused !== undefined) return refuse(projection.refused); const controlBytes = projection.bytes; const { binding } = input.execution; @@ -5421,6 +5449,29 @@ async function retireFinishedPerTurnRunnerdOwner(input: { remoteLeaseFenced: readNativeControllerRecovery(profile.nativeControllerRecovery, profile.sessionCheckpoint) !== null }); if (!stopped.allowed) return refuse(`controller_${retirementReasonName("controllerReason", stopped.reason)}`); const provenFinishedHere = finishedHere(); + const proveLocalRunnerStopped = async (): Promise => { + const run = prior!.run; + // Settlement may already have cleared the process fields after it saved a + // server-authored stop receipt; that receipt is the identity to check. + const released = run.processPid === null && run.processGroupId === null && run.processStartedAt === null; + const receipt = released ? await readNativeLocalProcessStop(input.db, run.companyId, run.id) : null; + if (released && !receipt) return { stopped: false, reason: "runner_process_identity_missing" }; + const providerEvents = await input.db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + inArray(heartbeatRunEvents.eventType, [...NATIVE_PROVIDER_PROCESS_EVENT_TYPES]))) + .orderBy(desc(heartbeatRunEvents.id)).limit(100); + return proveNativeLocalRunnerStopped({ + processPid: receipt?.processPid ?? run.processPid, + processGroupId: receipt?.processGroupId ?? run.processGroupId, + processStartedAt: receipt ? null : run.processStartedAt, + runnerStopRecorded: receipt !== null, + providerProcesses: nativeProviderProcessIdentities(record(profile.sessionCheckpoint), providerEvents), + }); + }; + if (localRunner) { + const runnerStopped = await proveLocalRunnerStopped(); + if (!runnerStopped.stopped) return refuse(runnerStopped.reason); + } // Earlier rejected admissions can quarantine the same projection. Each copy // must pass the same proof: a rebuilt root never carries the local Runner // evidence that held an earlier attempt, so that evidence stays decisive. @@ -5429,13 +5480,14 @@ async function retireFinishedPerTurnRunnerdOwner(input: { if (!isSafeNativeStateDirectory(quarantine)) return refuse("quarantine_unsafe"); for (const name of readdirSync(quarantine)) { if (!name.startsWith(`${basename(input.root)}.`)) continue; - const copy = readSettledRunnerdProjection(resolve(quarantine, name), identity); + const copy = readSettledRunnerdProjection(resolve(quarantine, name), identity, { localRunner }); if (copy.refused !== undefined) return refuse(`quarantine_${copy.refused}`); } } // Database reads can await. Retire only the exact evidence proven above. - if (readSettledRunnerdProjection(input.root, identity).bytes !== controlBytes) return refuse("journal_changed"); + if (readSettledRunnerdProjection(input.root, identity, { localRunner }).bytes !== controlBytes) return refuse("journal_changed"); if (provenFinishedHere && !finishedHere()) return refuse("controller_restarted_here"); + if (localRunner && !(await proveLocalRunnerStopped()).stopped) return refuse("runner_process_restarted"); const archivedAt = Date.now(); const archiveRoot = resolve(input.root, "continuity-breaks", `${archivedAt}-${randomUUID()}`); mkdirSync(archiveRoot, { recursive: true, mode: 0o700 }); @@ -7008,6 +7060,7 @@ export function verifyNativeHarnessBackup(input: { export function nativeHarnessBackupCursorConflict(input: { backupRoot: string; stateRoot: string; + controllerState?: { schema: string; identity: { runnerInstanceId: string }; ackedSourceSeq: number }; }): string | null { const read = (path: string, maxBytes: number) => { try { @@ -7018,7 +7071,7 @@ export function nativeHarnessBackupCursorConflict(input: { return null; } }; - const control = read(resolve(input.stateRoot, "control-plane", "control-plane-state.json"), + const control = input.controllerState ?? read(resolve(input.stateRoot, "control-plane", "control-plane-state.json"), NATIVE_CONTROL_PLANE_STATE_MAX_BYTES); if (!control || !RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(control.schema))) return null; const runner = read(resolve(input.backupRoot, "runner", "runner-state.json"), NATIVE_RUNNER_STATE_MAX_BYTES); @@ -10704,7 +10757,14 @@ async function executePaperclipNativeSessionWithinScope( } else { error = identityRedactor.redact(error); } - if (nativeRunsDetachingForRestart.has(input.execution.binding.runId)) { + // A closed runtime API listener means this process is shutting down; it + // closes HTTP admission before it drains runs. For a local Runner, keep the + // claim for the drain to suspend: the next controller proves the Runner + // process identity and adopts the run without spending a failure or a + // same-run retry. A remote Runner has no process evidence on this host, so + // it keeps the bootstrap retry below. + if (nativeRunsDetachingForRestart.has(input.execution.binding.runId) + || (error instanceof LocalRuntimeApiListenerClosedError && input.runnerExecutionTarget?.kind !== "remote")) { await leaseRenewal.stop().catch(() => undefined); liveQuestions.close(); activeNativeSessions.delete(input.execution.binding.runId); @@ -14374,7 +14434,8 @@ async function createRunnerdBackendWithinSessionClaim( runnerInstanceId: input.runnerInstanceId, }); if (!backup) throw new Error("runner_harness_backup_unavailable"); - const cursorConflict = nativeHarnessBackupCursorConflict({ backupRoot: backup.root, stateRoot: root }); + const cursorConflict = nativeHarnessBackupCursorConflict({ backupRoot: backup.root, stateRoot: root, + controllerState: runnerControlPlane?.store.state }); if (cursorConflict) throw new Error(cursorConflict); await measureNativeRunnerSpan( input.trace, diff --git a/tests/runner-e2e/report-lock.json b/tests/runner-e2e/report-lock.json index bd6c65c89b8..d6740c79df5 100644 --- a/tests/runner-e2e/report-lock.json +++ b/tests/runner-e2e/report-lock.json @@ -1,6 +1,6 @@ { - "inputSha256": "14049cb20d48f448b6f24e5445a75bf8d22ed3052fe9ff885f4c127775d958d3", - "outputSha256": "53f6c7c7f09a5cf0442ee2992a656eb2336ab6b13e665219a0cd1247475f13d9", - "patchSha256": "eb4adb2f3c8a2ebefe65d3100b581c5765318023d96aa8a80394c848d7299b9d", + "inputSha256": "53f6c7c7f09a5cf0442ee2992a656eb2336ab6b13e665219a0cd1247475f13d9", + "outputSha256": "5e2b012b50bb8b198384da4f5427c27acd948af405f89cde15bfa48bb10c0ad4", + "patchSha256": "2abdd6ec8e374bfb20b7e168ef010858493a50a044360175e1af5a2ca393f909", "basis": "Generated with the trusted master lockfile refresh. Review this offline correction and its hashes with the dependency changes." } diff --git a/tests/runner-e2e/report-lock.patch b/tests/runner-e2e/report-lock.patch index 6e5bb651368..090ee867256 100644 --- a/tests/runner-e2e/report-lock.patch +++ b/tests/runner-e2e/report-lock.patch @@ -1,14 +1,15 @@ diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml -index 45b3aaf..a87f8e7 100644 +index a87f8e7..97cbf7b 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml -@@ -1306,6 +1306,9 @@ importers: - lucide-react: - specifier: ^1.48.0 - version: 1.48.0(react@19.2.8) -+ mdast-util-from-markdown: -+ specifier: ^2.0.3 -+ version: 2.0.3 - mermaid: - specifier: ^12.1.0 - version: 12.1.0 +@@ -5048,7 +5048,9 @@ packages: + '@smithy/core@3.35.1': + resolution: {integrity: sha512-i4YPS4B6ts7bjn7UwLnGjiZdprOvHvgGobFZsYK3GIY3E5hIqtj0rReU69BcTpGp+fvtraSNXeG1l+jtJvF55w==} + engines: {node: '>=18.0.0'} +- deprecated: Deprecated due to a bug in custom logger data redaction. See https://github.com/aws/aws-sdk-js-v3/issues/8393. ++ deprecated: |- ++ Deprecated due to a bug in custom logger data redaction. Upgrade to the latest version. If you used a custom logger with this ++ version, review your logs and rotate any credentials or other secrets. See https://github.com/aws/aws-sdk-js-v3/issues/8393. + + '@smithy/credential-provider-imds@4.5.2': + resolution: {integrity: sha512-A9uSdn72ozbRUSit0eib0TW7nXuNPlaeM0zcGkJ+nE6tFcSDbnmtwoxbTCFBukVQcszDAyvsd7+rTduPTXpygg==} diff --git a/tests/runner-e2e/workflow-target.test.ts b/tests/runner-e2e/workflow-target.test.ts index f98b85c128b..d02c01d647b 100644 --- a/tests/runner-e2e/workflow-target.test.ts +++ b/tests/runner-e2e/workflow-target.test.ts @@ -16,8 +16,9 @@ async function resolveTarget(event: string, cadence: string, branch = "master") const output = path.join(directory, "output"); const result = spawnSync("bash", ["-c", `gh() { printf '%s\\n' "$MOCK_BRANCH_SHA"; }\n${script}`], { encoding: "utf8", env: { ...process.env, EVENT_NAME: event, EVAL_CADENCE: cadence, + IMAGE_ONLY: "false", TESTOPS_REQUESTED: "false", TARGET_BRANCH: branch, DEFAULT_BRANCH: "master", GITHUB_SHA: queuedSha, - MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip", IMAGE_ONLY: "false" }, + MOCK_BRANCH_SHA: advancedSha, GITHUB_OUTPUT: output, REPOSITORY: "paperclipai/paperclip" }, }); return { status: result.status, stderr: result.stderr, output: await readFile(output, "utf8").catch(() => "") };