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 2c34ce01ae1..5e33e16a758 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 @@ -521,6 +521,22 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).toContain(`Run ${prior.runId} finished without a saved runner checkpoint`); }, 60_000); + it.each(["created", "replacement"] as const)( + "bootstraps again when a restart cut the follow-up off after it retired the finished owner: %s sandbox", async outcome => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + expect(await followUp.start(outcome)).toBe(ADMITTED); + const root = await sessionRoot(); + expect(existsSync(join(root, "control-plane"))).toBe(false); + expect(await archivedOwners(root)).toEqual([prior.runId]); + followUp.logs.length = 0; + expect(await followUp.start(outcome, { kind: "bootstrap_incomplete", runId: followUp.runId, leaseOwner: "successor", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never)).toBe(ADMITTED); + const [copy] = await quarantined(); + expect(copy).toMatch(/\.identity_indeterminate\./); + expect(existsSync(join(stateDirectory, "quarantine", copy!, "continuity-breaks"))).toBe(true); + }, 60_000); + async function keepLocalRootWithoutPointer(prior: Awaited>, followUp: Awaited>) { await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, prior.runId)); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 657c2a96b8a..5f9608db4cd 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1367,6 +1367,11 @@ export async function claimNativeRestartRecoveries(input: { // backend may spawn a runner, and a remote runner exchanges commands only // through a Postgres controller checkpoint. With neither, nothing reached // a runner. The previous controller is gone (takeover was allowed above). + const pendingInheritedRequests = Array.isArray(checkpointRecord.pendingRuntimeRequests) + && checkpointRecord.pendingRuntimeRequests.length > 0; + const continuedSessionOnly = hasCheckpoint && checkpointIdentityMatches && providerEvents.length === 0 + && !pendingInheritedRequests && checkpointRecord.activeTurnId == null && checkpointRecord.semanticResult == null + && checkpointRecord.terminal == null; const remoteRunnerNeverLaunched = !unclaimedBootstrap && remoteWorkspace?.state === "prepared" && !pluginTask && !durableAuthority && stagedAttachment === null && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION @@ -1381,9 +1386,9 @@ export async function claimNativeRestartRecoveries(input: { && row.coordinator.failureDetail?.recoveryMode === "bootstrap_retry" && !row.coordinator.leaseOwner) && !row.coordinator.failureDetail?.successorRunId && !row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt - && !row.run.sessionIdBefore && !row.run.sessionIdAfter + && (!row.run.sessionIdBefore || continuedSessionOnly) && !row.run.sessionIdAfter && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) - && !hasCheckpoint && !hasProviderEvidence + && (continuedSessionOnly || (!hasCheckpoint && !hasProviderEvidence)) && !(await hasNativeExecutionStartEvidence(tx, row.run)); // A settled retry starts new work, so it needs the same current // admission as the resumption reconciler: task pause, agent status, @@ -1473,7 +1478,8 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessAlive: providerProcesses.livePids.length > 0, knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, - hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands && !(unclaimedBootstrap && inheritedCheckpointOnly), + hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands && !(unclaimedBootstrap && inheritedCheckpointOnly) + && !remoteRunnerNeverLaunched, stagedAttachmentVerified: stagedAttachment !== null, checkpointIdentityMatches: checkpointIdentityMatches && diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index a56d5456ae1..335225b7636 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -2960,6 +2960,42 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(disposition.kind).not.toBe("bootstrap_incomplete"); }); + it.each([ + { label: "restarts", startRequested: false, pendingRequest: false, providerEvent: false }, + { label: "keeps a requested runner start held instead of restarting", startRequested: true, pendingRequest: false, + providerEvent: false }, + { label: "keeps a pending runtime request held instead of restarting", startRequested: false, pendingRequest: true, + providerEvent: false }, + { label: "keeps a projected provider event held instead of restarting", startRequested: false, pendingRequest: false, + providerEvent: true }, + ])("$label a claimed remote follow-up that never requested a runner and carries its predecessor's session", + async ({ startRequested, pendingRequest, providerEvent }) => { + const fixture = await seedUnlaunchedRemoteClaim(`PROVEN-CONTINUED-${startRequested}-${pendingRequest}-${providerEvent}`, + startRequested ? { event: { eventType: "native.process_start_requested" } } : {}); + const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(heartbeatRuns).set({ sessionIdBefore: "previous-thread", + runnerProfileJson: { ...seeded.runnerProfileJson, sessionCheckpoint: { + driverKind: "codex_app_server", backendKind: "runner", sessionId: "previous-thread", cursor: "37", + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "previous-thread", activeTurnId: null, terminal: null, + terminalTurns: [{ turnId: "previous-turn", fingerprint: "{}" }], semanticResult: null, + pendingRuntimeRequests: pendingRequest ? [{ requestId: "input-1", kind: "user_input" }] : [], lineage: [], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + if (providerEvent) await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq: 901, + stream: "system", level: "info", eventType: "session.started" }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + if (startRequested || pendingRequest || providerEvent) { + expect(claim).not.toMatchObject({ kind: "bootstrap_incomplete" }); + return; + } + expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId, controllerGeneration: 2 }); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ disposition: "bootstrap_incomplete", + reason: "remote_runner_never_launched" }); + }); + it.each([false, true])("restarts a claimed remote run that never requested a runner (lease expired=%s)", async (leaseExpired) => { const fixture = await seedUnlaunchedRemoteClaim(`PROVEN-${leaseExpired}`, { leaseExpired }); expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index d3573e54916..74ae7795aca 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5809,9 +5809,8 @@ async function migrateRunnerdStateRootForExecution(input: { } const identity = readRunnerdDurableIdentity(scoped); if (!identity && input.restartRecovery?.kind === "bootstrap_incomplete" && - (!input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local") && !existsSync(resolve(scoped, "control-plane", "control-plane-state.json")) && - await noLocalRunnerProcessForRun(input.execution.binding.runId)) { + (input.runnerExecutionTarget?.kind === "remote" || await noLocalRunnerProcessForRun(input.execution.binding.runId))) { // The server was killed between creating the root and the local runner's // first state write. No runner for the run is alive on this host and none // connected, so no provider started and no command was acknowledged.