diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 62622d23111..a7f91e7ba13 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1715,6 +1715,16 @@ starts by itself once the gate clears. A Retry accepted during the pre-deploy drain stays saved and is dispatched by the deferred-receipt worker after the restart. +When a remote native run fails and its close cannot prove that the Runner +suspended (for example, the Runner is unreachable after `provider_transport_failed` +or a `turn.start` timeout), the controller no longer force-stops that Runner. It +leaves the Runner and its saved state running, keeps the session quarantined, and +the same-run reattachment adopts the live Runner with its real turn state. Before +this, the close stopped the Runner mid-turn, so the reattachment always found it +gone with an unsettled journal (`runner_remote_process_identity_unavailable`) and +the task waited for Retry. A Stop, a cancelled turn and a `stop_cleanup` recovery +still stop the Runner, and a Runner that settled or exited is handled as before. + A remote native run can lose its original runner or sandbox before reattachment. This happens after a server restart, and for a same-run retry after a turn ends without a semantic result. The reattachment then fails with diff --git a/doc/run-log-events.md b/doc/run-log-events.md index ecca5d5a430..cbf3b62079b 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -358,7 +358,10 @@ runner diagnostic. Its evidence includes runner suspension, provider drain, final provider state, pending call/operation/source-event IDs and input digests, and incomplete result-delivery command IDs and statuses. If execution and cleanup both fail, execution retains its original error identity and cleanup is -attached as `cleanupError`. +attached as `cleanupError`. `runnerRetained: true` means a remote Runner that +could not prove its suspension was left running for the same-run +reattachment instead of being force-stopped; the diagnostic +`remote runner left alive unsettled for same-run reattachment` records it. Semantic settlement also includes up to 20 content-free failure records, with the call ID, operation ID, stage (`dispatch` or `persist_result`), and cause. diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts index f9af4a42915..cd2d6975156 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.test.ts @@ -124,6 +124,22 @@ describe("HarnessDriverBackend", () => { expect(startTurn).toHaveBeenCalledExactlyOnceWith({ message }); }); + it.each([ + { cancelled: false, retainUnsettledRemoteRunner: true }, + { cancelled: true, retainUnsettledRemoteRunner: false }, + ])("asks to retain an unsettled remote Runner on close only when the run was not cancelled (cancelled=$cancelled)", + async ({ cancelled, retainUnsettledRemoteRunner }) => { + const close = vi.fn(async () => {}); + const session = Object.assign(new FakeHarnessSession(), { close }); + const backend = new HarnessDriverBackend({ ...driver, openSession: async () => session }); + const opened = await backend.openSession({ identity: { + runId: "run-1", sessionId: "session-1", companyId: "company-1", issueId: "issue-1", agentId: "agent-1", + } }); + if (cancelled) await cancelSession(opened, { reason: "Stop requested", signal: new AbortController().signal }).cleanup; + await opened.close({ reason: "native session execution complete" }); + expect(close).toHaveBeenCalledExactlyOnceWith({ reason: "native session execution complete", retainUnsettledRemoteRunner }); + }); + it("forwards reasoning when the provider explicitly supports it", async () => { const startTurn = vi.fn(async () => ({ turnId: "turn-1" })); const session = Object.assign(new FakeHarnessSession(), { startTurn, supportsTurnReasoning: () => true }); diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index e6623853776..d736206396c 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -917,7 +917,7 @@ class HarnessNativeSession implements NativeSession { } close(input: { reason: string }) { - return this.#session.close(input); + return this.#session.close({ ...input, retainUnsettledRemoteRunner: !this.#explicitlyCancelled }); } } diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 7a9396eac64..922fbd929ff 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -660,7 +660,7 @@ export interface HarnessSession { release?(): Promise; /** Relinquish controller authority without semantically closing the session. */ detachControllerForRestart?(): Promise; - close(input: { reason: string; force?: boolean }): Promise; + close(input: { reason: string; force?: boolean; retainUnsettledRemoteRunner?: boolean }): Promise; } /** A local harness implementation hidden behind the runner daemon boundary. */ diff --git a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts index 2502927f526..c24c3825526 100644 --- a/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts +++ b/packages/paperclip-runner/src/drivers/codex/app-server-transport.ts @@ -75,7 +75,7 @@ export interface CodexAppServerTransport { * Close the provider transport. The optional reason is controller-owned * diagnostic context; transports must not forward it to the provider. */ - close(reason?: string): Promise; + close(reason?: string, options?: { retainUnsettledRemoteRunner?: boolean }): Promise; /** After close: the runner settled and its process exit was observed. */ runnerExitConfirmed?(): boolean; /** Return the confirmed released source, or false when active use was not released. */ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts index 6de9c319e24..e3bb904288b 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-harness-session.ts @@ -933,11 +933,11 @@ export class CodexHarnessSession } // `force` is part of the contract. Codex closes the same way with or without it. - async close(input: { reason: string; force?: boolean }): Promise { + async close(input: { reason: string; force?: boolean; retainUnsettledRemoteRunner?: boolean }): Promise { this.cancelPendingRequests("session_closed"); this.eventQueue.close(); // Untyped callers (for example generated scripts) may still omit the input. - await this.transport.close(input?.reason); + await this.transport.close(input?.reason, { retainUnsettledRemoteRunner: input?.retainUnsettledRemoteRunner === true }); } async release(): Promise { diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport-remote-retention.test.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport-remote-retention.test.ts new file mode 100644 index 00000000000..c68c0f131e5 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport-remote-retention.test.ts @@ -0,0 +1,188 @@ +import { execFileSync, spawn } from "node:child_process"; +import { mkdtemp, readFile, rm } from "node:fs/promises"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { expect, it, vi } from "vitest"; +import { NativeSessionCloseUnrecoverableError } from "../contracts/native-session-backend.js"; +import type * as durableControlPlane from "../control-plane/durable-prp-control-plane.js"; +import type { DurablePrpControlPlane } from "../control-plane/durable-prp-control-plane.js"; +import { codexSemanticToolSpecs } from "../drivers/codex/codex-app-server-driver.js"; +import { createCapabilityRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, expectTurnStarted, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +function processState(pid: number): "alive" | "gone" { + try { process.kill(pid, 0); } catch { return "gone"; } + try { + return execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim().startsWith("Z") ? "gone" : "alive"; + } catch { return "gone"; } +} + +const signal = (pid: number | null | undefined, name: NodeJS.Signals) => { + if (!pid) return; + try { process.kill(pid, name); } catch (error) { if (!["ESRCH", "EPERM"].includes(String((error as NodeJS.ErrnoException).code))) throw error; } +}; + +async function unreachableRemoteRunnerClose(input: { + retainUnsettledRemoteRunner?: boolean; + recoveryPurpose?: "execution" | "stop_cleanup"; +}) { + const directory = await mkdtemp(join(tmpdir(), "runnerd-remote-retention-")); + const children: ReturnType[] = []; + const launch = vi.fn((spec: durableControlPlane.RunnerProcessLaunchSpec) => { + const child = spawn(spec.command, [...spec.args], { cwd: spec.cwd, env: spec.environment, stdio: "ignore" }); + children.push(child); + return { + child, + completion: new Promise((resolveExit, rejectExit) => { + child.once("error", rejectExit); + child.once("exit", (code, exitSignal) => resolveExit({ code, signal: exitSignal, stdout: "", stderr: "" })); + }), + isExitConfirmed: () => child.exitCode !== null || child.signalCode !== null, + }; + }); + const diagnostics: string[] = []; + const bundle = createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(directory, "--linger-after-turn-start"), stateDirectory: directory, + runnerReconnectGraceMs: 60_000, runnerProcessLauncher: launch, closeGraceMs: 600, + onDiagnostic: message => diagnostics.push(message), + ...(input.recoveryPurpose ? { recoveryPurpose: input.recoveryPurpose } : {}), + }); + bundle.transport.setServerRequestHandler(async () => ({ success: true, contentItems: [] })); + let providerPid: number | null = null; + try { + await bundle.transport.request("initialize", {}); + await bundle.transport.request("thread/start", { cwd: directory, dynamicTools: [] }); + await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Keep working." }] }); + await expectTurnStarted(bundle.transport.notifications()[Symbol.asyncIterator]()); + providerPid = bundle.evidence().codexPid; + const runner = children[0]!; + signal(runner.pid, "SIGSTOP"); + const closed = await bundle.transport.close("native session execution complete", + input.retainUnsettledRemoteRunner === undefined ? undefined + : { retainUnsettledRemoteRunner: input.retainUnsettledRemoteRunner }) + .then(() => null, (error: unknown) => error); + const runnerState = processState(runner.pid!); + signal(runner.pid, "SIGCONT"); + const runnerJournal = JSON.parse(await readFile(join(directory, "runner", "runner-state.json"), "utf8")) as Record; + return { closed, runnerState, runnerJournal, launch, diagnostics }; + } finally { + for (const child of children) { signal(child.pid, "SIGCONT"); signal(child.pid, "SIGKILL"); } + signal(providerPid, "SIGKILL"); + await rm(directory, { recursive: true, force: true }); + } +} + +it("leaves an unreachable remote Runner alive and unsettled when a failed run closes, so the same-run reattach can adopt it", async () => { + const result = await unreachableRemoteRunnerClose({ retainUnsettledRemoteRunner: true }); + expect(result.closed).toBeInstanceOf(NativeSessionCloseUnrecoverableError); + expect(result.runnerState).toBe("alive"); + expect(result.runnerJournal.lifecycle).toBe("ready"); + expect(result.launch).toHaveBeenCalledTimes(1); + const settlement = result.diagnostics.find(message => message.startsWith("native_session_settlement_incomplete ")); + expect(JSON.parse(settlement!.slice("native_session_settlement_incomplete ".length))) + .toMatchObject({ runnerSuspended: false, runnerRetained: true }); +}, 30_000); + +it.each([ + { title: "an explicit cancellation", options: { retainUnsettledRemoteRunner: false } }, + { title: "a caller that does not ask to retain it", options: {} }, + { title: "a Stop cleanup", options: { retainUnsettledRemoteRunner: true, recoveryPurpose: "stop_cleanup" as const } }, +])("still stops an unreachable remote Runner on $title", async ({ options }) => { + const result = await unreachableRemoteRunnerClose(options); + expect(result.closed).toBeInstanceOf(Error); + expect(result.runnerState).toBe("gone"); +}, 30_000); + +it("keeps a retained remote Runner for the reattaching controller, which settles it into a journal the next attempt can relaunch", async () => { + const root = await mkdtemp(join(tmpdir(), "runnerd-remote-retention-adopt-")); + const server = createServer(); + let authority: DurablePrpControlPlane | null = null; + server.on("upgrade", (request, socket, head) => { + if (!authority) { socket.destroy(); return; } + authority.handleUpgrade(request, socket, "/runner", head); + }); + await new Promise(resolveListen => server.listen(0, "127.0.0.1", resolveListen)); + const address = server.address(); + if (!address || typeof address === "string") throw new Error("Missing retention test listener"); + const registration = async (next: DurablePrpControlPlane) => { + authority = next; + return { connectUrl: `ws://127.0.0.1:${address.port}/runner`, release: () => { if (authority === next) authority = null; } }; + }; + const children: ReturnType[] = []; + const launch = vi.fn((spec: durableControlPlane.RunnerProcessLaunchSpec) => { + const child = spawn(spec.command, [...spec.args], { cwd: spec.cwd, env: spec.environment, stdio: "ignore", detached: true }); + children.push(child); + return { + child, + completion: new Promise((resolveExit, rejectExit) => { + child.once("error", rejectExit); + child.once("exit", (code, exitSignal) => resolveExit({ code, signal: exitSignal, stdout: "", stderr: "" })); + }), + isExitConfirmed: () => child.exitCode !== null || child.signalCode !== null, + }; + }); + const shared = { + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: resolve(import.meta.dirname, "../../runner/target/debug/fake-codex-app-server"), + codexArgs: ["--state-file", join(root, "provider-state.json"), "--hold-first-durable-turn"], + stateDirectory: root, + prpIdentity: { + runnerInstanceId: "runner-retained", environmentLeaseId: "lease-retained", runId: "run-retained", + normalizedSessionId: "session-retained", turnId: "turn-retained", itemId: "item-retained", + }, + runnerReconnectGraceMs: 5_000, + runnerProcessReconnectGraceMs: null, + controlPlaneRegistration: registration, + }; + const first = createCapabilityRunnerdCodexTransport({ ...shared, runnerProcessLauncher: launch, closeGraceMs: 600 }); + let adopted: ReturnType | undefined; + let pid: number | null = null; + let providerPid: number | null = null; + try { + const opened = await first.transport.request("thread/start", { cwd: root, dynamicTools: codexSemanticToolSpecs() }); + await first.transport.request("turn/start", { input: [{ type: "text", text: "Keep this task active." }] }); + for await (const notification of first.transport.notifications()) { + if (notification.method === "turn/started") break; + } + pid = children[0]!.pid!; + providerPid = first.evidence().codexPid; + signal(pid, "SIGSTOP"); + await expect(first.transport.close("native session execution complete", { retainUnsettledRemoteRunner: true })) + .rejects.toBeInstanceOf(NativeSessionCloseUnrecoverableError); + expect(processState(pid)).toBe("alive"); + signal(pid, "SIGCONT"); + const journal = async () => JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")) as Record; + await new Promise(resolveWait => setTimeout(resolveWait, 500)); + expect(processState(pid)).toBe("alive"); + expect((await journal()).lifecycle).toBe("ready"); + const duplicateLaunch = vi.fn(() => { throw new Error("Must adopt the existing runner"); }); + const thread = opened.thread as Record; + const adoptedPid = pid; + adopted = createCapabilityRunnerdCodexTransport({ + ...shared, + resumeDynamicTools: codexSemanticToolSpecs(), + resumeProviderSession: { driverSessionId: String(thread.id), providerSessionId: String(thread.sessionId) }, + runnerProcessLauncher: duplicateLaunch, + adoptExistingRunner: { + pid: adoptedPid, processGroupId: adoptedPid, startedAt: new Date().toISOString(), + isAlive: () => processState(adoptedPid) === "alive", + signal: name => { try { process.kill(-adoptedPid, name); return true; } catch { return false; } }, + }, + }); + adopted.transport.setServerRequestHandler(async () => ({ success: true, contentItems: [] })); + await adopted.transport.request("thread/read", {}).catch(() => undefined); + await vi.waitFor(() => expect(processState(adoptedPid)).toBe("gone"), { timeout: 10_000, interval: 50 }); + expect(await journal()).toMatchObject({ lifecycle: "suspended", outbox: [], pendingTerminalDelivery: null }); + expect(duplicateLaunch).not.toHaveBeenCalled(); + expect(launch).toHaveBeenCalledTimes(1); + } finally { + await adopted?.transport.close().catch(() => undefined); + for (const child of children) { signal(child.pid, "SIGCONT"); signal(child.pid ? -child.pid : null, "SIGKILL"); } + signal(providerPid, "SIGKILL"); + server.closeAllConnections(); + if (server.listening) await new Promise(resolveClose => server.close(() => resolveClose())); + await rm(root, { recursive: true, force: true }); + } +}, 30_000); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 791f0afda6c..58329d15648 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -4049,6 +4049,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { readonly #closedSignal: Promise; #resolveClosedSignal!: () => void; #closePromise: Promise | null = null; + #retainUnsettledRemoteRunner = false; #pendingCloseCheckpointRetry: CloseCheckpointRetry | null = null; #failedCloseCheckpoint: CloseCheckpointRetry | null = null; #shutdownSettled = false; @@ -5304,11 +5305,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { return false; } - close(reason?: string): Promise { + close(reason?: string, options?: { retainUnsettledRemoteRunner?: boolean }): Promise { // Detachment relinquishes process ownership. A late execution finalizer // must not suspend or signal the runner now owned by the next controller. if (this.#controllerDetachedForRestart) return this.#detachPromise ?? Promise.resolve(); + if (this.#closePromise === null) this.#retainUnsettledRemoteRunner = options?.retainUnsettledRemoteRunner === true; if (reason) { this.#diagnostic( `runner transport close requested: ${reason.replaceAll(/[\r\n]/g, " ").slice(0, 1_000)}`, @@ -5516,6 +5518,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { // runnerd exited on its own, unsuspended, while the session was open: it // was lost (killed or crashed), not stopped or broken by this close. let runnerLost = false; + let runnerRetained = false; + const remoteRunnerRetainable = this.#retainUnsettledRemoteRunner + && this.options.recoveryPurpose !== "stop_cleanup" + && (this.options.runnerProcessLauncher !== undefined || adoptedRunner !== undefined); let lastSuspensionState: Record | null = null; let suspensionState: { commandStatus: string | null; @@ -5604,9 +5610,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { this.#diagnostic( "runner did not prove durable suspension before checkpoint", ); + runnerRetained = remoteRunnerRetainable && !runnerLost && !(await this.#runnerHasExited()); + if (runnerRetained) this.#diagnostic("remote runner left alive unsettled for same-run reattachment"); } try { - if (this.#handle) { + if (runnerRetained) { + this.#evidence.runnerExited = false; + } else if (this.#handle) { const result = await waitForProcess( this.#handle, Math.max(0, closeDeadline - Date.now()), @@ -5690,6 +5700,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { if (suspensionRequired && !runnerSettled && !runnerLossReaped) { const settlement = { runnerSuspended, + runnerRetained, providerDrained, suspensionState, semanticTools: this.#core?.semanticToolSettlementDiagnostics(), @@ -5738,7 +5749,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { }), forceKill: async () => { const handle = this.#handle; - if (!handle || this.#evidence.runnerExited) return; + if (!handle || this.#evidence.runnerExited || runnerRetained) return; handle.child.kill("SIGKILL"); // A remote kill dispatches an asynchronous, ownership-fenced RPC. // Do not release the session for reuse until its process monitor has