diff --git a/packages/paperclip-runner/src/backends/harness-driver-backend.ts b/packages/paperclip-runner/src/backends/harness-driver-backend.ts index d736206396c..168feb23d27 100644 --- a/packages/paperclip-runner/src/backends/harness-driver-backend.ts +++ b/packages/paperclip-runner/src/backends/harness-driver-backend.ts @@ -168,6 +168,7 @@ export class HarnessDriverBackend implements NativeSessionBackend { reason: recovered.reason, ...(recovered.failureClass ? { failureClass: recovered.failureClass } : {}), ...(recovered.resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}), + ...(recovered.runnerExitConfirmed ? { runnerExitConfirmed: true as const } : {}), }; } try { diff --git a/packages/paperclip-runner/src/backends/native-backend-factory.test.ts b/packages/paperclip-runner/src/backends/native-backend-factory.test.ts index 008d9249cf6..b347533f87b 100644 --- a/packages/paperclip-runner/src/backends/native-backend-factory.test.ts +++ b/packages/paperclip-runner/src/backends/native-backend-factory.test.ts @@ -492,6 +492,20 @@ describe("native backend factory", () => { }); }); + it("carries the transport's proof that a Runner exited before answering session.open", async () => { + const { snapshot } = statelessRecovery(execution()); + const exited = new FakeCodexTransport(); + exited.rejectMethods.set("initialize", new Error("runnerd exited while waiting for session.open")); + Object.assign(exited, { runnerExitConfirmed: () => true }); + const proven = createNativeSessionBackend(execution(), { + codexTransportFactory: () => exited, + environment: { ...process.env, PAPERCLIP_WORKSPACE_CWD: WORKSPACE }, + }); + await expect(proven.recoverSession!(snapshot as never, { signal: new AbortController().signal })) + .resolves.toMatchObject({ recovered: false, reason: expect.stringContaining("runnerd exited while waiting for session.open"), + runnerExitConfirmed: true }); + }); + it.each([ ["an active turn", { activeTurnId: "turn-in-flight" }], ["a pending runtime request", { pendingRuntimeRequests: [{ requestId: "request-1" }] }], diff --git a/packages/paperclip-runner/src/contracts/harness-driver.ts b/packages/paperclip-runner/src/contracts/harness-driver.ts index 922fbd929ff..66a3a161200 100644 --- a/packages/paperclip-runner/src/contracts/harness-driver.ts +++ b/packages/paperclip-runner/src/contracts/harness-driver.ts @@ -561,6 +561,7 @@ export interface HarnessSessionRecoveryResult { * saved state for this session (NativeRunnerResumeStateUnavailableError) and * the checkpoint had no turn or runtime request in flight. */ resumeStateUnavailable?: true; + runnerExitConfirmed?: true; } /** Reads the closed class of an authenticated runner command failure, or of a diff --git a/packages/paperclip-runner/src/contracts/native-session-backend.ts b/packages/paperclip-runner/src/contracts/native-session-backend.ts index b8e9d775fec..bb1812cbabb 100644 --- a/packages/paperclip-runner/src/contracts/native-session-backend.ts +++ b/packages/paperclip-runner/src/contracts/native-session-backend.ts @@ -87,6 +87,7 @@ export interface NativeSessionRecoveryResult { reason?: string; failureClass?: HarnessSessionRecoveryResult["failureClass"]; resumeStateUnavailable?: HarnessSessionRecoveryResult["resumeStateUnavailable"]; + runnerExitConfirmed?: HarnessSessionRecoveryResult["runnerExitConfirmed"]; } /** Runner-owned cause: the process was restored, but its active turn was lost. */ diff --git a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts index 6b3a67ccc6d..c8b15fffb34 100644 --- a/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts +++ b/packages/paperclip-runner/src/drivers/codex/codex-app-server-driver-impl.ts @@ -788,6 +788,7 @@ export class CodexAppServerDriver implements HarnessDriver { reason: redactCodexDiagnostic(String(error)), ...(failureClass ? { failureClass } : {}), ...(resumeStateUnavailable ? { resumeStateUnavailable: true as const } : {}), + ...(transport.runnerExitConfirmed?.() === true ? { runnerExitConfirmed: true as const } : {}), }; } finally { cancellation.detach(); diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index 58329d15648..b55c9229682 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -3982,6 +3982,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { #observedProviderModel: string | undefined; #sessionId: string | null = null; #providerIdentity: Record | null = null; + #openingCommandId: string | null = null; #turnControls: NativeTurnControlCapabilities = { steering: false, queuedFollowUp: false }; supportsTurnReasoning(): boolean { @@ -4294,10 +4295,15 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { * new runner: the old one is gone, and its provider never opened a thread. */ runnerExitConfirmed(): boolean { - return this.#closeFinished && !this.#startupComplete && this.#providerIdentity === null + return this.#closeFinished && !this.#startupComplete + && (this.#providerIdentity === null || this.#openingCommandUnanswered()) && this.#evidence.runnerExited === true && this.#handle?.isExitConfirmed?.() !== false; } + #openingCommandUnanswered(): boolean { + return this.#openingCommandId !== null && this.#core?.getCommand(this.#openingCommandId)?.status === "pending"; + } + async request( method: string, params: Record, @@ -6204,7 +6210,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { core.persistRunAttachTemplate(runAttachTemplate); if (!unclaimedBootstrap) { core.queueCommand("run.prepare", runAttachTemplate); - core.queueCommand("session.open", { reuse: "same_session" }); + const opening: { commandId?: string } | undefined = core.queueCommand("session.open", { reuse: "same_session" }); + this.#openingCommandId = opening?.commandId ?? null; } const registration = this.options.controlPlaneRegistration ? await this.options.controlPlaneRegistration(core) @@ -6880,6 +6887,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport { : null; if (recoveryProbeCommandId !== null) { core.queueCommand(sourceAttachmentTarget ? "session.open" : "runner.drain", {}, recoveryProbeCommandId); + this.#openingCommandId = recoveryProbeCommandId; } // A controller retry can open the exact authority after run.attach has // already reached a durable outcome. Re-observe that command instead of diff --git a/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts b/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts new file mode 100644 index 00000000000..3941f32ea80 --- /dev/null +++ b/packages/paperclip-runner/src/live/runnerd-stale-controller-journal.test.ts @@ -0,0 +1,68 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expect, it } from "vitest"; +import type { PersistedHarnessProviderIdentity } from "../contracts/harness-driver.js"; +import { createCapabilityRunnerdCodexTransport } from "./runnerd-codex-transport.js"; +import { defaultCapabilityRunnerdBinary, fakeCodex, fakeCodexArgs } from "../../test/support/runnerd-transport.js"; + +const identity = { + runnerInstanceId: "stale-journal-runner", + environmentLeaseId: "stale-journal-lease", + runId: "stale-journal-run", + normalizedSessionId: "stale-journal-session", + turnId: "stale-journal-turn", + itemId: "stale-journal-item", +}; + +function transport(root: string, resumeProviderSession?: { driverSessionId: string; providerSessionId: string; providerIdentity?: PersistedHarnessProviderIdentity }) { + return createCapabilityRunnerdCodexTransport({ + runnerBinary: defaultCapabilityRunnerdBinary(), + codexCommand: fakeCodex, + codexArgs: fakeCodexArgs(root), + stateDirectory: root, + prpIdentity: identity, + ...(resumeProviderSession ? { resumeProviderSession } : {}), + }); +} + +function kill(pid: number | null | undefined) { + if (!pid) return; + try { process.kill(pid, "SIGKILL"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error; } +} + +it.each(["stale", "current"] as const)("a follow-up controller resumes the predecessor's suspended runner from a %s journal", async journal => { + const root = await mkdtemp(join(tmpdir(), "runnerd-stale-journal-")); + const controlPath = join(root, "control-plane", "control-plane-state.json"); + const predecessor = transport(root); + let follower: ReturnType | undefined; + try { + const opened = (await predecessor.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] })).thread as { id: string; sessionId: string }; + await predecessor.transport.close(); + const closed = JSON.parse(await readFile(controlPath, "utf8")) as { commands: Array<{ type: string }> } & Record; + expect(closed.commands.map(command => command.type).slice(-2)).toEqual(["runner.drain", "runner.suspend"]); + const runner = JSON.parse(await readFile(join(root, "runner", "runner-state.json"), "utf8")) as { lastControllerCommandSeq: number; lifecycle: string }; + expect(runner).toMatchObject({ lastControllerCommandSeq: closed.commands.length, lifecycle: "suspended" }); + if (journal === "stale") await writeFile(controlPath, JSON.stringify({ ...closed, commands: closed.commands.slice(0, -2) })); + + const providerIdentity = journal === "stale" ? { providerIdentity: { kind: "acpx", normalizedSessionId: identity.normalizedSessionId } as unknown as PersistedHarnessProviderIdentity } : {}; + follower = transport(root, { driverSessionId: opened.id, providerSessionId: opened.sessionId, ...providerIdentity }); + const outcome = await follower.transport.request("thread/read", {}) + .then(() => "resumed", (error: unknown) => String(error)); + if (journal === "stale") { + expect(outcome).not.toBe("resumed"); + expect(outcome).toContain("controller sequence must be contiguous"); + await follower.transport.close().catch(() => undefined); + expect(follower.transport.runnerExitConfirmed?.()).toBe(true); + } else { + expect(outcome).toBe("resumed"); + expect(outcome).not.toContain("controller sequence must be contiguous"); + } + } finally { + await follower?.transport.close().catch(() => undefined); + await predecessor.transport.close().catch(() => undefined); + kill(predecessor.evidence().runnerPid); + kill(follower?.evidence().runnerPid); + await rm(root, { recursive: true, force: true }); + } +}, 60_000); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 1d480fb524f..63044f29808 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2587,9 +2587,13 @@ export async function executeNativeSession( if (recovery.failureClass === PROVIDER_STATE_INCOMPATIBLE) { throw new NativeProviderStateIncompatibleError(recovery.reason ?? "unknown"); } - throw new Error( + const failure = new Error( `native_session_recovery_failed: ${recovery.reason ?? "unknown"}`, ); + if (recovery.runnerExitConfirmed === true) { + Object.defineProperty(failure, "runnerExitConfirmed", { value: true, enumerable: false }); + } + throw failure; } continuityBreak = { reason: governedWaitFollowUpLostState diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index d74c98558c2..f84352cd845 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -102,6 +102,7 @@ export type NativeRestartRecoveryClaim = | { kind: "plugin_task"; leaseId: string }; /** Discovery evidence only; executor revalidates the encrypted header. */ stagedAttachment?: Pick; + exitedBeforeSessionOpen?: true; } | { kind: "resume_dead_runner"; @@ -1985,6 +1986,8 @@ export async function claimNativeRestartRecoveries(input: { providerLeaseId: remoteWorkspace!.providerLeaseId, remoteCwd: remoteWorkspace!.remoteCwd, }, + ...(!pluginTask && row.coordinator.failureDetail?.runnerExitedBeforeSessionOpen === true + ? { exitedBeforeSessionOpen: true as const } : {}), } satisfies NativeRestartRecoveryClaim; } if (claimKind === "reattach_existing_runner") { 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 4d03417458e..21d662aa422 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -446,6 +446,7 @@ import { NATIVE_PROVIDER_SESSION_CLOSED, NATIVE_PROVIDER_STARTUP_REJECTED, nativeRemoteStartupRejectionCode, + nativeRunnerExitedBeforeSessionOpen, nativeRepeatedStartupFailureAction, nativePreviousStartupFailureSignature, nativeStartupFailureSignature, @@ -1021,6 +1022,26 @@ describe("remote controller restart adoption", () => { await expect(verifyRemoteRunnerOrSettledExit({ claim, target: exitedRunner(settledSuspension) as never, identity, runId: "run", normalizedSessionId: "session", providerAttempt })).resolves.toBeUndefined(); }); + const exitedBeforeOpen = { lifecycle: "ready", outbox: [{ id: 127 }], pendingTerminalDelivery: null, pendingProviderCleanup: null }; + it("starts a new runner when the previous attempt's runner exited before answering session.open", async () => { + const error = await verifyRemoteRunnerOrSettledExit({ claim, target: exitedRunner(exitedBeforeOpen) as never, identity, + runId: "run", normalizedSessionId: "session", providerAttempt: 2 }).catch(caught => caught); + expect(readRemoteRunnerRecoveryError(error)).toMatchObject({ stage: "process_verification", reason: "process_not_running", + runnerLifecycle: "ready", runnerSettled: false }); + await expect(verifyRemoteRunnerOrSettledExit({ claim: { ...claim, exitedBeforeSessionOpen: true }, + target: exitedRunner(exitedBeforeOpen) as never, identity, runId: "run", normalizedSessionId: "session", providerAttempt: 2 })) + .resolves.toBeUndefined(); + }); + it.each([ + ["a pending provider cleanup", { ...exitedBeforeOpen, pendingProviderCleanup: { pid: 7 } }, 2], + ["a pending terminal delivery", { ...exitedBeforeOpen, pendingTerminalDelivery: { eventId: "event" } }, 2], + ["a runner that stopped mid-turn", { ...exitedBeforeOpen, lifecycle: "turn_active" }, 2], + ["an exhausted attempt budget", exitedBeforeOpen, 4], + ] as const)("keeps a runner that exited before session.open fail-closed after %s", async (_case, state, providerAttempt) => { + const error = await verifyRemoteRunnerOrSettledExit({ claim: { ...claim, exitedBeforeSessionOpen: true }, + target: exitedRunner(state) as never, identity, runId: "run", normalizedSessionId: "session", providerAttempt }).catch(caught => caught); + expect(readRemoteRunnerRecoveryError(error)).toMatchObject({ stage: "process_verification", reason: "process_not_running" }); + }); it.each([ ["an unsettled suspension", { ...settledSuspension, outbox: [{ id: 1 }] }, 2], ["a pending provider cleanup", { ...settledSuspension, pendingProviderCleanup: { pid: 7 } }, 2], @@ -14241,6 +14262,14 @@ describe("native session bounded recovery", () => { }); }); + it("records a runner proven to have exited before answering session.open as a failed fresh start", () => { + const proven = (message: string) => Object.defineProperty(new Error(message), "runnerExitConfirmed", { value: true }); + expect(nativeRunnerExitedBeforeSessionOpen(new Error("native_session_recovery_failed", { + cause: proven("runnerd exited while waiting for session.open") }))).toBe(true); + expect(nativeRunnerExitedBeforeSessionOpen(new Error("runnerd exited while waiting for session.open"))).toBe(false); + expect(nativeRunnerExitedBeforeSessionOpen(proven("PRP command session.open rejected: {}"))).toBe(false); + }); + it("stops a remote startup rejection at once because a remote same-run bootstrap cannot be verified", () => { const failure = sessionOpenFailure(startupFailure({ message: "Query closed before response received" })); expect(nativeRemoteStartupRejectionCode(nativeSessionFailureSourceCode(failure), failure, true)) @@ -15655,7 +15684,7 @@ describe("runnerd provider runtime wiring", () => { }, ); - it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "suspension-receipt", "suspension-active-owner", "changed", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => { + it.each((["codex", "claude"] as const).flatMap(provider => (["unchanged", "external-checkpoint", "suspension-receipt", "suspension-active-owner", "changed", "changed-external-checkpoint", "runtime-context", "retirement-failed", "stale-owner", "active-owner", "missing-owner", "foreign-environment", "plugin-replaced", "plugin-managed-github"] as const).filter(scenario => provider === "codex" || scenario.startsWith("plugin-")).map(scenario => [provider, scenario] as const)))("rotates an owned warm %s remote runner before checking changed configuration: %s", async (provider, scenario) => { const plugin = scenario === "plugin-replaced" || scenario === "plugin-managed-github"; const suspensionReceipt = scenario.startsWith("suspension-"); const reusable = suspensionReceipt || ["unchanged", "external-checkpoint", "plugin-managed-github"].includes(scenario); @@ -15704,13 +15733,16 @@ describe("runnerd provider runtime wiring", () => { // Use the real control plane and persistence callback, not a handwritten marker. const saved = new Map(); let persistedController: DurablePrpControlPlane | undefined; - const restore = scenario === "external-checkpoint" + const restore = scenario === "external-checkpoint" || scenario === "changed-external-checkpoint" ? vi.spyOn(runnerCheckpoint, "restoreNativeRunnerCheckpoint").mockImplementation(async (_db, binding, directory) => { const checkpoint = saved.get(binding.runId); if (!checkpoint) throw new Error("missing test checkpoint"); checkpoint.restoreSessionDirectory(directory); return checkpoint; }) : undefined; + const loadSaved = scenario === "changed-external-checkpoint" + ? vi.spyOn(runnerCheckpoint, "loadNativeRunnerCheckpoint").mockImplementation(async (_db, binding) => saved.get(binding.runId) ?? null) + : undefined; if (restore) { const completeIdentity = { ...identity, turnId: "turn-first", itemId: "item-first" }; // The authority must begin from a complete real state envelope. @@ -15723,8 +15755,16 @@ describe("runnerd provider runtime wiring", () => { const firstSession = { close: vi.fn(async () => { order.push("retire"); if (failRetirement) throw new Error("retirement unconfirmed"); + if (scenario === "changed-external-checkpoint") { + persistedController!.queueCommand("runner.drain", {}, "command_close_drain_fixture"); + persistedController!.queueCommand("runner.suspend", {}, "command_close_suspend_fixture"); + await persistedController!.flushCheckpoint(); + await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState(identity, "suspended"))); + return; + } await writeRunnerAuthority(root, identity, "suspended"); }) }; + let launchedJournal: string[] | undefined; const nextSession = { close: vi.fn(async () => undefined) }; const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "done", completionClaim: { contractRevision: first.completionContract.contract.revision, objectiveSatisfied: true, @@ -15749,7 +15789,10 @@ describe("runnerd provider runtime wiring", () => { return result; }).mockImplementationOnce(async options => { order.push("execute-next"); - if (persistedController) { + if (scenario === "changed-external-checkpoint") { + launchedJournal = (JSON.parse(await readFile(join(root, "control-plane/control-plane-state.json"), "utf8")).commands as + Array<{ type: string }>).map(command => command.type); + } else if (persistedController) { persistedController.rotateRunIdentity({ ...identity, runId: next.binding.runId, turnId: "turn-next", itemId: "item-next" }); await persistedController.flushCheckpoint(); await writeFile(join(root, "runner/runner-state.json"), JSON.stringify(durableRunnerState( @@ -15817,6 +15860,9 @@ describe("runnerd provider runtime wiring", () => { } if (scenario === "plugin-managed-github") expect(rebind).toHaveBeenCalledOnce(); expect(order).toEqual(reusable ? ["execute-next"] : ["retire", "execute-next"]); + if (scenario === "changed-external-checkpoint") { + expect(launchedJournal?.slice(-2)).toEqual(["runner.drain", "runner.suspend"]); + } if (["unchanged", "external-checkpoint", "suspension-receipt"].includes(scenario)) { if (!persistedController) await writeRunnerAuthority(root, { ...identity, runId: next.binding.runId }, "ready"); if (suspensionReceipt) await rm(join(root, "runner/runner-state.json")); @@ -15850,6 +15896,7 @@ describe("runnerd provider runtime wiring", () => { } } finally { restore?.mockRestore(); + loadSaved?.mockRestore(); await persistedController?.stop(); failRetirement = false; resume?.mockRestore();owners?.mockRestore();feedback.mockRestore(); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6dc376b8069..18fc09d6954 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -104,6 +104,7 @@ import { admitUnclaimedNativeBootstrapRetry } from "./native-bootstrap-retry.js" import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS, runnerdStateIsIncompleteBootstrap, runnerdStateHasNoProviderTurn, runnerdStateIsOpenedSessionBeforeTurn, runnerdStateIsUnconnectedBootstrap } from "./native-bootstrap-proof.js"; import { createNativeDeliverableTextObservation } from "./native-deliverable-text.js"; +import { reconcileControllerJournalWithRunner, refreshRetiredWarmOwnerProjection } from "./warm-controller-journal.js"; import { createNativeGitHubAccess, type NativeGitHubAccess } from "./native-github-access.js"; import { resolveGitHubOperationCredentials } from "../github-operation-credentials.js"; import { syncManagedAiCredentials, type ManagedAiCredentialWritebackBinding } from "../managed-ai-credential-writeback.js"; @@ -7325,6 +7326,7 @@ export async function restoreRemoteSuspendedRunnerCheckpoint(input: { state = await readRemoteRunnerState({ runner: target.runner, stateDirectory: remoteRunnerStateDirectory(retained.remoteCwd, nativeSessionKey(input.execution)) }); } + await reconcileControllerJournalWithRunner({ ...controllerProjectionInput(input, { runId: prior.runId, issueId: input.execution.binding.issueId }), runnerState: state }); // A running warm source keeps the existing live-adoption identity, idle, // controller-fencing and physical-process checks. Only suspension is cached. if (!movedAllocation && input.execution.session.lifecyclePolicy.mode === "warm" && state.lifecycle !== "suspended") return; @@ -7363,6 +7365,18 @@ function isUndeliveredSuspendReceipt(state: Record, root: strin } } +function controllerProjectionInput(input: { db: Db; execution: NativeExecutionInput }, owner: { runId: string; issueId: string }) { + const root = scopedRunnerdStateRoot(input.execution); + const statePath = resolve(root, "control-plane", "control-plane-state.json"); + return { + db: input.db, root, normalizedSessionId: nativeSessionKey(input.execution), + binding: { companyId: input.execution.binding.companyId, issueId: owner.issueId, runId: owner.runId }, + readControlState: () => existsSync(statePath) + ? readBoundedNativeFile(statePath, NATIVE_CONTROL_PLANE_STATE_MAX_BYTES, "native_runner_checkpoint_invalid").toString("utf8") + : null, + }; +} + /** Materialize opaque controller state before ordinary warm admission reads it. * A retained owner or server checkpoint identifies the last completed run; * the original external marker can precede several in-process warm turns. */ @@ -8632,6 +8646,18 @@ export function nativeRunnerExitConfirmed(error: unknown, depth = 0): boolean { || (Array.isArray(value.errors) && value.errors.slice(0, 3).some(item => nativeRunnerExitConfirmed(item, depth + 1))); } +function runnerExitBeforeSessionOpenMessage(error: unknown, depth = 0): boolean { + if (!error || typeof error !== "object" || depth > 5) return false; + const value = error as Record; + if (typeof value.message === "string" && /runnerd exited (?:while waiting for session\.open|unexpectedly)/.test(value.message)) return true; + return runnerExitBeforeSessionOpenMessage(value.cause, depth + 1) + || (Array.isArray(value.errors) && value.errors.slice(0, 3).some(item => runnerExitBeforeSessionOpenMessage(item, depth + 1))); +} + +export function nativeRunnerExitedBeforeSessionOpen(error: unknown): boolean { + return nativeRunnerExitConfirmed(error) && runnerExitBeforeSessionOpenMessage(error); +} + /** Same provider, category and message (numbers ignored): a retry would meet the same fault. */ export function nativeStartupFailureSignature(failure: unknown): string | null { const value = record(failure); @@ -11590,6 +11616,9 @@ async function executePaperclipNativeSessionAttempt( forgetWarmNativeSession(warmSessionId, entry); } finally { entry.busy = false; } if (preparedCopy) throw new Error("native_instruction_preparation_configuration_changed"); + if (input.useRunnerd && input.runnerExecutionTarget?.kind === "remote") { + await refreshRetiredWarmOwnerProjection(controllerProjectionInput(input, { runId: entry.runId, issueId: entry.issueId })); + } persistedWarmSession = loadWarmNativeCheckpoint( input.execution, warmConfigDigest, @@ -12672,6 +12701,8 @@ async function executePaperclipNativeSessionAttempt( // with this proof that the failed runner is gone. ...(recoveryEvidence.recoveryMode === "bootstrap_retry" && nativeRunnerExitConfirmed(error) ? { runnerExitConfirmed: true } : {}), + ...(input.runnerExecutionTarget?.kind === "remote" && nativeRunnerExitedBeforeSessionOpen(error) + ? { runnerExitedBeforeSessionOpen: true } : {}), ...(providerCooldownAt ? { providerCooldownAt: providerCooldownAt.toISOString() } : {}), ...(providerLimit?.plan.resetAt ? { providerLimitResetAt: providerLimit.plan.resetAt.toISOString() } : {}), recoveryOwner: recoveryProjection.recoveryOwner, @@ -14776,7 +14807,8 @@ async function readRemoteRunnerExitEvidence(runner: CommandManagedRuntimeRunner, } export async function verifyRemoteRunnerOrSettledExit( - input: Parameters[0] & { providerAttempt: number }, + input: Parameters[0] & { providerAttempt: number; + claim: { exitedBeforeSessionOpen?: true } }, ) { const recoveryDeadline = Math.min(input.recoveryDeadline ?? Number.POSITIVE_INFINITY, Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS); @@ -14785,18 +14817,25 @@ export async function verifyRemoteRunnerOrSettledExit( } catch (error) { const diagnostic = readRemoteRunnerRecoveryError(error); const target = input.target; + const exitedBeforeSessionOpen = input.claim.exitedBeforeSessionOpen === true; if (input.providerAttempt > NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT || diagnostic?.stage !== "process_verification" || diagnostic.reason !== "process_not_running" || - diagnostic.runnerLifecycle !== "suspended" || diagnostic.runnerSettled !== true || + (!exitedBeforeSessionOpen && (diagnostic.runnerLifecycle !== "suspended" || diagnostic.runnerSettled !== true)) || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) throw error; const state = await readRemoteRunnerState({ runner: target.runner, stateDirectory: remoteRunnerStateDirectory(target.remoteCwd, input.normalizedSessionId), recoveryDeadline }); - if (REMOTE_RUNNER_IDENTITY_FIELDS.some(field => state[field] !== input.identity[field]) || - state.lifecycle !== "suspended" || !describeExitedRemoteRunner(state).runnerSettled) throw error; + if (REMOTE_RUNNER_IDENTITY_FIELDS.some(field => state[field] !== input.identity[field])) throw error; + if (exitedBeforeSessionOpen ? !runnerStateAllowsFreshStart(state) + : state.lifecycle !== "suspended" || !describeExitedRemoteRunner(state).runnerSettled) throw error; return undefined; } } +function runnerStateAllowsFreshStart(state: Record): boolean { + return ["ready", "suspended", "connecting"].includes(String(state.lifecycle)) + && state.pendingTerminalDelivery == null && state.pendingProviderCleanup == null && state.warmTransition == null; +} + /** A crash before source bootstrap leaves a suspended Runner; a crash during * its drain leaves the same live source. Neither permits a replacement until * the physical process and all six persisted identity fields are proven. */ @@ -14830,6 +14869,14 @@ async function verifyStagedRemoteRunner(input: { const claim = { ...input.claim, runId: identity.runId }; const verify = () => verifyRemoteRunnerReattachment({ claim, target, identity: { ...identity }, runId: identity.runId, normalizedSessionId: identity.normalizedSessionId }); + if (state.lifecycle !== "suspended" && input.claim.exitedBeforeSessionOpen === true && runnerStateAllowsFreshStart(state)) { + try { return await verify(); } + catch (error) { + const diagnostic = readRemoteRunnerRecoveryError(error); + if (diagnostic?.stage !== "process_verification" || diagnostic.reason !== "process_not_running") throw error; + return undefined; + } + } if (state.lifecycle !== "suspended") return verify(); const settled = (value: Record) => sameIdentity(value) && value.lifecycle === "suspended" && Array.isArray(value.outbox) && value.outbox.length === 0 && value.pendingTerminalDelivery == null && diff --git a/server/src/services/native-runtime/warm-controller-journal.test.ts b/server/src/services/native-runtime/warm-controller-journal.test.ts new file mode 100644 index 00000000000..48586262ccd --- /dev/null +++ b/server/src/services/native-runtime/warm-controller-journal.test.ts @@ -0,0 +1,119 @@ +import { existsSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const checkpoints = vi.hoisted(() => new Map()); +vi.mock("./native-runner-checkpoint.js", () => ({ + loadNativeRunnerCheckpoint: vi.fn(async (_db: unknown, binding: { runId: string }) => { + const state = checkpoints.get(binding.runId); + if (state === undefined) return null; + const parsed = JSON.parse(state) as { identity: unknown; commands: Array<{ controllerSeq: number }> }; + return { + recoveryEvidence: { normalizedSessionId: "session" }, + records: new Map([["header", JSON.stringify({ schema: "paperclip.runner.controller-checkpoint.v2", state: { identity: parsed.identity } })], + ...parsed.commands.map(entry => [`command:${entry.controllerSeq}`, JSON.stringify(entry)] as [string, string])]), + restoreSessionDirectory(root: string) { + mkdirSync(join(root, "control-plane"), { recursive: true }); + writeFileSync(join(root, "control-plane", "control-plane-state.json"), state); + }, + }; + }), +})); + +const { CONTROLLER_JOURNAL_BEHIND_RUNNER, reconcileControllerJournalWithRunner, refreshRetiredWarmOwnerProjection } = + await import("./warm-controller-journal.js"); + +const identity = { runId: "run-a", runnerInstanceId: "runner", environmentLeaseId: "lease", normalizedSessionId: "session", + turnId: "turn", itemId: "item" }; +const drainId = `command_close_drain_${"d".repeat(32)}`; +const suspendId = `command_close_suspend_${"e".repeat(32)}`; + +function command(controllerSeq: number, type: string, commandId = `command_prp_${String(controllerSeq).padStart(8, "0")}`) { + return { schema: "paperclip.prp.command.v1", commandId, controllerSeq, type, issuedAt: "2026-11-11T00:00:00.000Z", + payload: {}, status: "completed", result: null }; +} + +function control(commands: ReturnType[]) { + return JSON.stringify({ schema: "paperclip.runner.durable.control-plane-state.v1", identity, commands, committedEvents: [] }); +} + +const opened = [command(1, "run.prepare"), command(2, "session.open"), command(3, "turn.start")]; +const closed = [...opened, command(4, "runner.drain", drainId), command(5, "runner.suspend", suspendId)]; + +function runner(processed: Array<{ commandId: string; controllerSeq: number; commandType: string; status?: string }>) { + return { schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "suspended", + lastControllerCommandSeq: Math.max(...processed.map(entry => entry.controllerSeq)), + processedCommands: Object.fromEntries(processed.map(entry => [entry.commandId, { status: "completed", ...entry }])) }; +} + +const processedClose = [...opened.map(entry => ({ commandId: entry.commandId, controllerSeq: entry.controllerSeq, commandType: entry.type })), + { commandId: drainId, controllerSeq: 4, commandType: "runner.drain" }, { commandId: suspendId, controllerSeq: 5, commandType: "runner.suspend" }]; + +describe("warm controller journal", () => { + let root: string; + let statePath: string; + beforeEach(() => { + root = mkdtempSync(join(tmpdir(), "warm-controller-journal-")); + statePath = join(root, "control-plane", "control-plane-state.json"); + mkdirSync(join(root, "control-plane"), { recursive: true }); + checkpoints.clear(); + }); + afterEach(() => rmSync(root, { recursive: true, force: true })); + + const input = () => ({ db: {} as never, root, normalizedSessionId: "session", + binding: { companyId: "company", issueId: "issue", runId: "run-a" }, + readControlState: () => existsSync(statePath) ? readFileSync(statePath, "utf8") : null }); + const projectedTypes = () => (JSON.parse(readFileSync(statePath, "utf8")).commands as Array<{ type: string }>).map(entry => entry.type); + + it("refreshes a retired owner's projection with the close it saved after admission projected it", async () => { + writeFileSync(statePath, control(opened)); + checkpoints.set("run-a", control(closed)); + await expect(refreshRetiredWarmOwnerProjection(input())).resolves.toBe("refreshed"); + expect(projectedTypes().slice(-2)).toEqual(["runner.drain", "runner.suspend"]); + }); + + it("holds a retired owner's projection whose saved journal diverges instead of choosing one", async () => { + writeFileSync(statePath, control(opened)); + checkpoints.set("run-a", control([opened[0], command(2, "session.open", "command_other")])); + await expect(refreshRetiredWarmOwnerProjection(input())).rejects.toThrow("native_runner_checkpoint_projection_held:retired_owner_journal_diverged"); + expect(projectedTypes()).toEqual(["run.prepare", "session.open", "turn.start"]); + }); + + it("leaves another run's projection and a marker untouched", async () => { + writeFileSync(statePath, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity })); + checkpoints.set("run-a", control(closed)); + await expect(refreshRetiredWarmOwnerProjection(input())).resolves.toBe("unchanged"); + writeFileSync(statePath, control(opened)); + await expect(refreshRetiredWarmOwnerProjection({ ...input(), binding: { ...input().binding, runId: "run-b" } })).resolves.toBe("unchanged"); + expect(projectedTypes()).toHaveLength(3); + }); + + it("reconciles a saved journal that is behind the runner only by the predecessor's settled close", async () => { + writeFileSync(statePath, control(opened)); + checkpoints.set("run-a", control(closed)); + await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).resolves.toBe("reconciled"); + expect(projectedTypes()).toEqual(closed.map(entry => entry.type)); + }); + + it("refuses with a precise code when the canonical journal cannot prove the runner's close", async () => { + writeFileSync(statePath, control(opened)); + checkpoints.set("run-a", control(opened)); + await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).rejects.toThrow(CONTROLLER_JOURNAL_BEHIND_RUNNER); + expect(projectedTypes()).toHaveLength(3); + }); + + it("refuses with a precise code when the runner is ahead by anything other than a settled close", async () => { + writeFileSync(statePath, control(opened.slice(0, 2))); + checkpoints.set("run-a", control(closed)); + await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).rejects.toThrow(CONTROLLER_JOURNAL_BEHIND_RUNNER); + }); + + it("leaves a current journal and a live runner without a closing suspend to the existing admission checks", async () => { + writeFileSync(statePath, control(closed)); + await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: runner(processedClose) })).resolves.toBe("unchanged"); + writeFileSync(statePath, control(opened.slice(0, 2))); + const live = runner(processedClose.slice(0, 3)); + await expect(reconcileControllerJournalWithRunner({ ...input(), runnerState: live })).resolves.toBe("unchanged"); + }); +}); diff --git a/server/src/services/native-runtime/warm-controller-journal.ts b/server/src/services/native-runtime/warm-controller-journal.ts new file mode 100644 index 00000000000..29e8583aa7e --- /dev/null +++ b/server/src/services/native-runtime/warm-controller-journal.ts @@ -0,0 +1,119 @@ +import type { Db } from "@paperclipai/db"; +import { RUNNERD_CONTROL_PLANE_STATE_SCHEMAS } from "./native-bootstrap-proof.js"; +import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; + +type Binding = { companyId: string; issueId: string; runId: string }; +type JournalEntry = { commandId: string; controllerSeq: number }; + +const IDENTITY_FIELDS = ["runId", "runnerInstanceId", "environmentLeaseId", "normalizedSessionId", "turnId", "itemId"] as const; +const CLOSE_COMMAND_ID = /^command_close_(drain|suspend)_[0-9a-f]{32}$/; +export const CONTROLLER_JOURNAL_BEHIND_RUNNER = "native_controller_recovery_missing_evidence:controller_journal_behind_runner"; + +function record(value: unknown): Record { + return value !== null && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +} + +function parseFullControlState(text: string | null): Record | null { + if (text === null) return null; + try { + const state = record(JSON.parse(text)); + return RUNNERD_CONTROL_PLANE_STATE_SCHEMAS.has(String(state.schema)) && Array.isArray(state.commands) ? state : null; + } catch { + return null; + } +} + +function journal(state: Record): JournalEntry[] { + return (state.commands as unknown[]).map(record).map(command => ({ + commandId: String(command.commandId), controllerSeq: Number(command.controllerSeq), + })); +} + +function highestSeq(entries: readonly JournalEntry[]): number { + return entries.reduce((highest, entry) => Math.max(highest, entry.controllerSeq), 0); +} + +function sameIdentity(left: unknown, right: unknown): boolean { + const a = record(left), b = record(right); + return IDENTITY_FIELDS.every(field => typeof a[field] === "string" && a[field] === b[field]); +} + +export function controllerJournalExtends(saved: readonly JournalEntry[], projected: readonly JournalEntry[]): boolean { + return saved.length >= projected.length && projected.every((entry, index) => + saved[index]?.commandId === entry.commandId && saved[index]?.controllerSeq === entry.controllerSeq); +} + +async function readCanonicalControlState(db: Db, binding: Binding, normalizedSessionId: string) { + const checkpoint = await loadNativeRunnerCheckpoint(db, binding); + if (!checkpoint || checkpoint.recoveryEvidence.normalizedSessionId !== normalizedSessionId) return null; + try { + const header = record(JSON.parse(checkpoint.records.get("header") ?? "null")); + const commands = [...checkpoint.records].filter(([part]) => /^command:\d+$/.test(part)) + .map(([, contents]) => record(JSON.parse(contents))) + .sort((left, right) => Number(left.controllerSeq) - Number(right.controllerSeq)); + return { checkpoint, state: { identity: record(header.state).identity, commands } as Record }; + } catch { + return null; + } +} + +export type ControllerProjectionInput = { + db: Db; + binding: Binding; + normalizedSessionId: string; + root: string; + readControlState: () => string | null; +}; + +export async function refreshRetiredWarmOwnerProjection(input: ControllerProjectionInput): Promise<"refreshed" | "unchanged"> { + const projectedText = input.readControlState(); + const projected = parseFullControlState(projectedText); + if (!projected || record(projected.identity).runId !== input.binding.runId) return "unchanged"; + const canonical = await readCanonicalControlState(input.db, input.binding, input.normalizedSessionId); + if (!canonical) return "unchanged"; + if (!sameIdentity(canonical.state.identity, projected.identity) + || !controllerJournalExtends(journal(canonical.state), journal(projected))) { + throw new Error("native_runner_checkpoint_projection_held:retired_owner_journal_diverged"); + } + if (journal(canonical.state).length === journal(projected).length) return "unchanged"; + if (input.readControlState() !== projectedText) throw new Error("native_runner_checkpoint_projection_held:journal_changed"); + canonical.checkpoint.restoreSessionDirectory(input.root); + return "refreshed"; +} + +function settledCloseAhead(runnerState: Record, journalSeq: number, runnerSeq: number): JournalEntry[] | null { + const processed = Object.values(record(runnerState.processedCommands)).map(record); + const last = processed.find(command => command.controllerSeq === runnerSeq); + if (last?.commandType !== "runner.suspend" || last.status !== "completed") return null; + const missing = processed.filter(command => Number(command.controllerSeq) > journalSeq) + .sort((left, right) => Number(left.controllerSeq) - Number(right.controllerSeq)); + const provable = missing.length === runnerSeq - journalSeq && missing.every((command, index) => + command.controllerSeq === journalSeq + index + 1 && command.status === "completed" + && (command.commandType === "runner.drain" || command.commandType === "runner.suspend") + && CLOSE_COMMAND_ID.test(String(command.commandId))); + if (!provable) throw new Error(CONTROLLER_JOURNAL_BEHIND_RUNNER); + return missing.map(command => ({ commandId: String(command.commandId), controllerSeq: Number(command.controllerSeq) })); +} + +export async function reconcileControllerJournalWithRunner(input: ControllerProjectionInput & { + runnerState: Record; +}): Promise<"reconciled" | "unchanged"> { + const projectedText = input.readControlState(); + const projected = parseFullControlState(projectedText); + if (!projected || !sameIdentity(projected.identity, input.runnerState)) return "unchanged"; + const projectedJournal = journal(projected); + const journalSeq = highestSeq(projectedJournal); + const runnerSeq = Number(input.runnerState.lastControllerCommandSeq); + if (!Number.isSafeInteger(runnerSeq) || runnerSeq <= journalSeq) return "unchanged"; + const missing = settledCloseAhead(input.runnerState, journalSeq, runnerSeq); + if (!missing) return "unchanged"; + const canonical = await readCanonicalControlState(input.db, input.binding, input.normalizedSessionId); + if (!canonical || !sameIdentity(canonical.state.identity, projected.identity) + || !controllerJournalExtends(journal(canonical.state), [...projectedJournal, ...missing]) + || highestSeq(journal(canonical.state)) !== runnerSeq) { + throw new Error(CONTROLLER_JOURNAL_BEHIND_RUNNER); + } + if (input.readControlState() !== projectedText) throw new Error("native_runner_checkpoint_projection_held:journal_changed"); + canonical.checkpoint.restoreSessionDirectory(input.root); + return "reconciled"; +}