diff --git a/server/src/services/heartbeat/queue.test.ts b/server/src/services/heartbeat/queue.test.ts index b3f0d0e97b7..6dee76c29f3 100644 --- a/server/src/services/heartbeat/queue.test.ts +++ b/server/src/services/heartbeat/queue.test.ts @@ -1026,6 +1026,19 @@ describe.skipIf(!support.supported)("heartbeat queue database wiring", () => { expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(run.id); }); + it("leaves a queued run for the next server when shutdown closes admission during the queue scan", async () => { + const { run, agent, wake } = await fixture("queued"); + await db.update(heartbeatRuns).set({ responsibleUserId: "queue-test-user" }).where(eq(heartbeatRuns.id, run.id)); + let admissionClosed = false; + const deps = callbacks(db); + deps.isRunAdmissionClosed = () => admissionClosed; + deps.getAgent.mockImplementation(async () => { admissionClosed = true; return agent; }); + expect(await createHeartbeatQueue(db, deps).startNextQueuedRunForAgent(agent.id)).toEqual([]); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toMatchObject({ status: "queued", startedAt: null }); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)))[0].status).toBe(wake.status); + expect(deps.executeRun).not.toHaveBeenCalled(); + }); + it("rechecks the connection hold when Stop races the queue scan", async () => { const { run, agent } = await fixture("queued"); const deps = { ...callbacks(db), canStartNewExternalBotWork: vi.fn(async () => false) }; diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index 509b8c6bed7..030a3abbf18 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -2284,7 +2284,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende const claimedRuns: Array = []; for (const queuedRun of prioritizedRuns) { - if (claimedRuns.length >= availableSlots) break; + if (claimedRuns.length >= availableSlots || isRunAdmissionClosed()) break; let claimed: typeof heartbeatRuns.$inferSelect | null; try { claimed = await claimQueuedRun(queuedRun, companyAgents, deferredPostCommitEffects, deferredCancellations); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index f84352cd845..d71af628d3e 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -745,7 +745,8 @@ 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_never_received_commands"; + | "remote_runner_exited_before_provider_work" | "remote_runner_never_received_commands" + | "remote_follow_up_never_attached"; processStartMatches: boolean; knownProviderProcessAlive?: boolean; knownProviderProcessIdentityAmbiguous?: boolean; @@ -1592,7 +1593,22 @@ export async function claimNativeRestartRecoveries(input: { // Fresh evidence wins over the saved failure: any event the runner or // provider sent contradicts "no provider work". && !(await hasNativeRunnerOriginatedEvidence(tx, row.run)); - if ((remoteRunnerNeverLaunched || remoteRunnerExitedBeforeProviderWork) && row.coordinator.phase === "retryable_failure" && !admission.exhausted + const remoteFollowUpNeverAttached = !unclaimedBootstrap && !remoteRunnerNeverLaunched && !remoteRunnerExitedBeforeProviderWork + && remoteWorkspace?.state === "prepared" && !pluginTask && !durableAuthority && stagedAttachment === null + && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION + && row.coordinator.phase === "retryable_failure" && row.run.status === "failed" && !row.coordinator.leaseOwner + && settledFailure?.recoveryMode === "bootstrap_retry" + && settledFailure?.originalFailureCode === "local_runtime_api_listener_unavailable" + && settledFailure?.providerSessionEstablished === false && settledFailure?.providerEventsExist === false + && !settledFailure?.successorRunId + && continuedSessionOnly && inheritedCheckpointOnly && !row.run.sessionIdAfter + && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) + && (inheritedRunnerIdentity || (!row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt)) + && await carriesSettledPredecessorSession(tx, row.run, checkpointRecord.providerSessionId) + && !(await hasNativeRunnerOriginatedEvidence(tx, row.run)) + && !(await hasNativeRunEvent(tx, row.run, PROCESS_IDENTITY_RECORDED)); + if ((remoteRunnerNeverLaunched || remoteRunnerExitedBeforeProviderWork || remoteFollowUpNeverAttached) + && row.coordinator.phase === "retryable_failure" && !admission.exhausted && !await readNativeSessionResumptionAgent(tx as unknown as Db, { issue, run: row.run, coordinator: row.coordinator })) { const deferred = await deferRemoteRetry("native_retry_admission_ineligible", ["done", "cancelled"].includes(issue.status)); if (deferred) return deferred; @@ -1633,7 +1649,7 @@ export async function claimNativeRestartRecoveries(input: { && (!ownProviderEvidence || remoteBootstrapProof === "session_opened_without_turn"); const remoteRunnerNeverReceivedCommands = !unclaimedBootstrap && !remoteRunnerNeverLaunched - && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete + && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete && !remoteFollowUpNeverAttached && remoteSandbox && remoteWorkspace !== null && !pluginTask && !durableAuthority && stagedAttachment === null && admissionCheckpointOnly && providerEvents.length === 0 && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION @@ -1659,10 +1675,11 @@ export async function claimNativeRestartRecoveries(input: { remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, remoteBootstrapProvenIncomplete: remoteBootstrapProvenIncomplete || remoteRunnerNeverLaunched - || remoteRunnerExitedBeforeProviderWork || remoteRunnerNeverReceivedCommands, + || remoteRunnerExitedBeforeProviderWork || remoteRunnerNeverReceivedCommands || remoteFollowUpNeverAttached, remoteBootstrapIncompleteReason: remoteRunnerNeverLaunched ? "remote_runner_never_launched" : remoteRunnerExitedBeforeProviderWork ? "remote_runner_exited_before_provider_work" : remoteRunnerNeverReceivedCommands ? "remote_runner_never_received_commands" + : remoteFollowUpNeverAttached ? "remote_follow_up_never_attached" : remoteBootstrapProof === "session_opened_without_turn" ? "remote_session_opened_without_turn" : undefined, runnerPidAlive, runnerGroupAlive, @@ -1671,7 +1688,7 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands && !(unclaimedBootstrap && inheritedCheckpointOnly) - && !remoteRunnerNeverLaunched && !(remoteBootstrapProvenIncomplete && reboundSessionOnly), + && !remoteRunnerNeverLaunched && !remoteFollowUpNeverAttached && !(remoteBootstrapProvenIncomplete && reboundSessionOnly), 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 5b1251bb68d..cf50f4d54b1 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 @@ -3105,6 +3105,55 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_execution_never_claimed" }); }); + it.each([ + { label: "bootstraps a follow-up carrying its predecessor's Runner identity", mode: "inherited", restarts: true }, + { label: "bootstraps a follow-up that requested its warm Runner on its predecessor's identity", mode: "inherited_requested", restarts: true }, + { label: "bootstraps a follow-up with no Runner identity that requested a Runner", mode: "requested", restarts: true }, + { label: "keeps a follow-up with its own recorded Runner identity held", mode: "own_identity", restarts: false }, + { label: "keeps a follow-up with a Runner-originated event held", mode: "runner_event", restarts: false }, + { label: "keeps a follow-up with a carried active turn held", mode: "active_turn", restarts: false }, + { label: "keeps a follow-up whose provider session no settled run saved held", mode: "no_predecessor", restarts: false }, + { label: "keeps a follow-up whose saved failure established a provider session held", mode: "session_established", restarts: false }, + ] as const)("$label after its own server closed the runtime listener before it attached (s9511, s9513)", async ({ mode, restarts }) => { + const fixture = await seedUnlaunchedRemoteClaim(`FOLLOW-UP-LISTENER-${mode}`, { bootstrapRetry: true, + ...(["inherited_requested", "requested", "own_identity", "runner_event"].includes(mode) + ? { event: { eventType: "native.process_start_requested" } } : {}) }); + await reboundFollowUpCheckpoint(fixture, mode === "active_turn" ? { activeTurnId: "turn-in-flight" } : {}, mode !== "no_predecessor"); + await fixture.db.update(nativeRunFinalizations).set({ failureDetail: { + message: "local_runtime_api_listener_unavailable", recoverable: true, recoveryMode: "bootstrap_retry", + checkpointExists: true, providerEventsExist: false, providerSessionEstablished: mode === "session_established", + originalFailureCode: "local_runtime_api_listener_unavailable", + } }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const [current] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + if (mode !== "requested") { + const processStartedAt = new Date(Date.now() - 120_000); + await fixture.db.update(heartbeatRuns).set({ processPid: 596, processGroupId: null, processStartedAt, processHostHash: "remote" }) + .where(eq(heartbeatRuns.id, fixture.runId)); + if (mode !== "own_identity") await fixture.db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", + runtimeMode: "native", nativeIssueId: fixture.issueId, nativeSessionId: current.nativeSessionId, + runnerInstanceId: current.runnerInstanceId, createdAt: new Date(current.createdAt.getTime() - 60_000), + finishedAt: new Date(current.createdAt.getTime() - 5_000), processPid: 596, processGroupId: null, processStartedAt, + processHostHash: "remote", contextSnapshot: { issueId: fixture.issueId } }); + } + if (mode === "own_identity") await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, + seq: 901, stream: "system", level: "info", eventType: "native.process_identity_recorded" }); + if (mode === "runner_event") await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, + seq: 902, stream: "system", level: "info", eventType: "item.updated", sourceEventId: "runner-event-after-failure" }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + if (!restarts) { + 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).toMatchObject({ phase: "observed", recoveryState: "bootstrap_incomplete", failureCode: null }); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_follow_up_never_attached" }); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run).toMatchObject({ status: "running", errorCode: null, finishedAt: null }); + }); + 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,