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 ae9be61febe..e1887754d71 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 @@ -202,8 +202,10 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered const remoteExecute = vi.fn(async (_request: unknown): Promise<{ exitCode: number; timedOut: boolean; stdout: string; stderr: string }> => { throw new Error("admission must not reach the sandbox"); }); - const start = (outcome: "created" | "replacement" | "resumed" = "created") => createRunnerdBackend({ + const start = (outcome: "created" | "replacement" | "resumed" = "created", + restartRecovery?: Parameters[0]["restartRecovery"]) => createRunnerdBackend({ db, execution, runnerInstanceId: next.runnerInstanceId, onLog: async (_stream, chunk) => { logs.push(chunk); }, + ...(restartRecovery ? { restartRecovery } : {}), runnerExecutionTarget: { kind: "remote", transport: "sandbox", providerKey: "daytona", environmentId, leaseId, remoteCwd: "/workspace", reusableLeaseConfigured: true, runner: { execute: remoteExecute }, @@ -658,6 +660,16 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(await followUp.start()).toBe(outcome); }, 60_000); + it("starts a follow-up whose own bootstrap never reached its root from the previous owner's settled root", async () => { + const prior = await seedRecoveredPriorRun(); + const followUp = await admitFollowUp(prior); + const root = await restoreHeldRoot(prior, followUp); + expect(await followUp.start("created", { kind: "bootstrap_incomplete", runId: followUp.runId, leaseOwner: "successor", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null } as never)).toBe(ADMITTED); + expect(await quarantined()).toEqual([]); + expect(await archivedOwners(root)).toEqual([prior.runId]); + }, 60_000); + it("admits again when a retry rebuilds the same projection from the database", async () => { const prior = await seedRecoveredPriorRun(); const followUp = await admitFollowUp(prior); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index bf8033add40..23f9ba7c2e6 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -678,7 +678,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: { /** Durable evidence proves a claimed remote bootstrap reached no provider work. */ remoteBootstrapProvenIncomplete?: boolean; remoteBootstrapIncompleteReason?: "remote_runner_never_launched" | "remote_session_opened_without_turn" - | "remote_runner_exited_before_provider_work"; + | "remote_runner_exited_before_provider_work" | "remote_runner_never_received_commands"; processStartMatches: boolean; knownProviderProcessAlive?: boolean; knownProviderProcessIdentityAmbiguous?: boolean; @@ -793,6 +793,16 @@ async function hasNativeRunnerOriginatedEvidence( return event !== undefined; } +async function hasNativeRunEvent( + tx: Pick, run: { companyId: string; id: string }, eventType: string, +): Promise { + const [event] = await tx.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, eventType))) + .limit(1); + return event !== undefined; +} + export async function hasNativeExecutionStartEvidence( tx: Pick, run: { companyId: string; id: string }, ): Promise { @@ -1408,6 +1418,24 @@ export async function claimNativeRestartRecoveries(input: { const remoteBootstrapProvenIncomplete = remoteBootstrapProof !== null && (!hasProviderEvidence || remoteBootstrapProof === "session_opened_without_turn"); + const admissionCheckpointOnly = !hasCheckpoint || (checkpointIdentityMatches && checkpointRecord.cursor == null + && (!Array.isArray(checkpointRecord.terminalTurns) || checkpointRecord.terminalTurns.length === 0) + && checkpointRecord.activeTurnId == null && checkpointRecord.semanticResult == null && checkpointRecord.terminal == null); + const remoteRunnerNeverReceivedCommands = !unclaimedBootstrap && !remoteRunnerNeverLaunched + && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete + && remoteSandbox && remoteWorkspace !== null && !pluginTask && !durableAuthority && stagedAttachment === null + && admissionCheckpointOnly && providerEvents.length === 0 + && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION + && row.coordinator.phase === "observed" && Boolean(row.coordinator.leaseOwner) + && row.coordinator.attempt >= 1 && row.coordinator.controllerGeneration >= 1 + && Boolean(row.coordinator.controllerBootId && row.coordinator.controllerPid && row.coordinator.controllerProcessStartedAt) + && !row.coordinator.failureDetail?.successorRunId + && !row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt + && !row.run.sessionIdBefore && !row.run.sessionIdAfter + && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) + && !(await hasNativeRunnerOriginatedEvidence(tx, row.run)) + && !(await hasNativeRunEvent(tx, row.run, PROCESS_IDENTITY_RECORDED)); + // Read before this claim appends its own history entry. const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator); const resumeKeepsAttempt = cleanDrainSuspension || await restartResumeKeepsAttempt(tx, { @@ -1417,9 +1445,10 @@ export async function claimNativeRestartRecoveries(input: { remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, remoteBootstrapProvenIncomplete: remoteBootstrapProvenIncomplete || remoteRunnerNeverLaunched - || remoteRunnerExitedBeforeProviderWork, + || remoteRunnerExitedBeforeProviderWork || remoteRunnerNeverReceivedCommands, remoteBootstrapIncompleteReason: remoteRunnerNeverLaunched ? "remote_runner_never_launched" : remoteRunnerExitedBeforeProviderWork ? "remote_runner_exited_before_provider_work" + : remoteRunnerNeverReceivedCommands ? "remote_runner_never_received_commands" : remoteBootstrapProof === "session_opened_without_turn" ? "remote_session_opened_without_turn" : undefined, runnerPidAlive, runnerGroupAlive, @@ -1427,7 +1456,7 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessAlive: providerProcesses.livePids.length > 0, knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, - hasCheckpoint, + hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands, 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 5dad4687c4e..26b9a82da58 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 @@ -2690,6 +2690,34 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); return fixture; } + it.each([ + { label: "restarts a remote follow-up", identityRecorded: false, inherited: true }, + { label: "restarts a first remote run", identityRecorded: false, inherited: false }, + { label: "keeps a recorded runner process held instead of restarting", identityRecorded: true, inherited: true }, + ])("$label whose controller never sent its runner a command", async ({ identityRecorded, inherited }) => { + const fixture = await seedUnlaunchedRemoteClaim(`NEVER-COMMANDED-${identityRecorded}-${inherited}`, + { event: { eventType: "native.process_start_requested" } }); + const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + if (inherited) await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...seeded.runnerProfileJson, sessionCheckpoint: { + driverKind: "codex_app_server", backendKind: "runner", sessionId: "inherited-thread", cursor: null, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "inherited-thread", activeTurnId: null, terminal: null, terminalTurns: [], semanticResult: null, + pendingRuntimeRequests: [], lineage: [], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + if (identityRecorded) await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq: 901, + stream: "system", level: "info", eventType: "native.process_identity_recorded" }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + if (identityRecorded) { + 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({ reason: "remote_runner_never_received_commands" }); + }); + it("restarts a remote bootstrap retry that shutdown cut before any runner was requested", async () => { const fixture = await seedUnlaunchedRemoteClaim("BOOTSTRAP-RETRY", { bootstrapRetry: true }); expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, @@ -2865,8 +2893,6 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); }); it.each<[string, Parameters[1], { kind: string; reason: string }]>([ - ["a runner process start was requested", { event: { eventType: "native.process_start_requested" } }, - { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }], ["a runner process identity was recorded", { event: { eventType: "native.process_identity_recorded" } }, { kind: "blocked", reason: "ambiguous_remote_recovery_evidence" }], ["the adapter was invoked", { event: { eventType: "adapter.invoke" } }, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 4782abec0a4..8e006540a3f 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -5808,7 +5808,7 @@ async function migrateRunnerdStateRootForExecution(input: { if (result.kind !== "state") throw new Error("Plugin Runner checkpoint unavailable"); savePluginRunnerCheckpoint(scoped, result.state, identity); } - if (input.restartRecovery?.kind === "bootstrap_incomplete") { + if (input.restartRecovery?.kind === "bootstrap_incomplete" && identity.runId === input.execution.binding.runId) { // A local runner's process ids belong to this host, so a fresh provider // session it opened before any turn can be proven gone here. const localProcesses = !input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local"; @@ -5883,7 +5883,8 @@ async function migrateRunnerdStateRootForExecution(input: { // Name the failed retirement proof: the shared error otherwise hides // which of several independent fences held the follow-up. let retirementRefusal = ""; - if (verification === "terminal_state_indeterminate" && !input.restartRecovery && !input.retainedWarmOwnerRunId) { + if (verification === "terminal_state_indeterminate" && !input.retainedWarmOwnerRunId + && (!input.restartRecovery || input.restartRecovery.kind === "bootstrap_incomplete")) { const retirement = await retireFinishedPerTurnRunnerdOwner({ ...input, root: scoped, identity }); if (retirement.retired) return; if (retirement.refused !== "lifecycle_not_per_turn") retirementRefusal = `; retirement_refused=${retirement.refused}`;