diff --git a/server/src/services/native-runtime/native-inherited-runner-retry.ts b/server/src/services/native-runtime/native-inherited-runner-retry.ts index 63c686304b8..0568842db20 100644 --- a/server/src/services/native-runtime/native-inherited-runner-retry.ts +++ b/server/src/services/native-runtime/native-inherited-runner-retry.ts @@ -89,6 +89,28 @@ export function readInheritedRunnerRetainedSandbox(profile: Record; + +/** The settled predecessor whose exact runner process identity this run carries. */ +async function settledPredecessorWithRunnerIdentity(tx: Pick, run: RunnerIdentityRun) { + if (!run.nativeSessionId || (run.processPid === null && run.processGroupId === null)) return null; + const [predecessor] = await tx.select({ id: heartbeatRuns.id, processStartedAt: heartbeatRuns.processStartedAt }) + .from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), + eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), ne(heartbeatRuns.id, run.id), + lt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES), + run.processPid === null ? isNull(heartbeatRuns.processPid) : eq(heartbeatRuns.processPid, run.processPid), + run.processGroupId === null ? isNull(heartbeatRuns.processGroupId) : eq(heartbeatRuns.processGroupId, run.processGroupId), + )).orderBy(desc(heartbeatRuns.createdAt)).limit(1); + return predecessor && sameTime(predecessor.processStartedAt, run.processStartedAt) ? predecessor : null; +} + +/** The run's recorded runner process identity is exactly a settled predecessor's, not its own. */ +export async function carriesInheritedRunnerIdentity(tx: Pick, run: RunnerIdentityRun): Promise { + return (await settledPredecessorWithRunnerIdentity(tx, run)) !== null; +} + export async function retireInheritedRemoteRunnerIdentity(db: Db, runId: string): Promise { return db.transaction(async transaction => { const tx = transaction as unknown as Db; @@ -127,16 +149,8 @@ export async function retireInheritedRemoteRunnerIdentity(db: Db, runId: string) if ((await nativeSessionProviderEvidence(tx, [run.id])).has(run.id)) return "not_applicable"; if (await hasNativeRunnerCheckpoint(tx, { companyId: run.companyId, issueId: issue.id, runId: run.id })) return "not_applicable"; - // The identity must be exactly the one a settled predecessor recorded. - const [predecessor] = await tx.select({ id: heartbeatRuns.id, processStartedAt: heartbeatRuns.processStartedAt }) - .from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.agentId, run.agentId), - eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), ne(heartbeatRuns.id, run.id), - lt(heartbeatRuns.createdAt, run.createdAt), inArray(heartbeatRuns.status, TERMINAL_RUN_STATUSES), - run.processPid === null ? isNull(heartbeatRuns.processPid) : eq(heartbeatRuns.processPid, run.processPid), - run.processGroupId === null ? isNull(heartbeatRuns.processGroupId) : eq(heartbeatRuns.processGroupId, run.processGroupId), - )).orderBy(desc(heartbeatRuns.createdAt)).limit(1); - if (!predecessor || !sameTime(predecessor.processStartedAt, run.processStartedAt)) return "not_applicable"; + const predecessor = await settledPredecessorWithRunnerIdentity(tx, run); + if (!predecessor) return "not_applicable"; // Keep the run's own sandbox when its lease is still live. const [lease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and( 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 3e6af4c6106..af4aeaf8ddd 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 @@ -337,6 +337,16 @@ describe.skipIf(!recoveryPostgresSupport.supported)("follow-up after a recovered expect(followUp.logs.join("")).not.toContain("starting a new runner for this conversation"); }, 60_000); + it("starts an unclaimed remote follow-up from the previous owner's external marker under a bootstrap claim", async () => { + const fixture = await staleProjection("codex", true); + expect(await fixture.followUp.start("created", { kind: "bootstrap_incomplete", runId: fixture.followUp.runId, + leaseOwner: "successor", controllerGeneration: 1, providerAttempt: 0, restartKind: "hard", recoveryRequestId: null } as never)) + .toBe(ADMITTED); + expect(JSON.parse(await readFile(fixture.path, "utf8")).identity).toEqual(fixture.latest.identity); + await fixture.assertBackup(); + expect(await quarantined()).toEqual([]); + }, 60_000); + it("rehydrates a same-source external marker before reading the suspended resumed sandbox", async () => { const fixture = await staleProjection("codex", true), { latest, followUp, root, path } = fixture; await writeFile(path, JSON.stringify({ schema: "paperclip.runner.external-control-plane-state.v1", identity: latest.identity })); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 14d3954a745..b9d19c33f7d 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { carriesInheritedRunnerIdentity } from "./native-inherited-runner-retry.js"; import { readPendingNativeRunStop } from "./native-cancellation-request.js"; import { CONTROLLER_LEASE_RENEW_MS } from "../controller-instance-lease.js"; import { currentControllerProcessStamp, UNKNOWN_PROCESS_SPACE } from "../controller-process-identity.js"; @@ -1324,6 +1325,13 @@ export async function claimNativeRestartRecoveries(input: { const hasProviderEvidence = hasCheckpointProviderIdentity || providerEvents.length > 0; + 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 inheritedCheckpointOnly = hasCheckpoint && admissionCheckpointOnly && remoteSandbox && !pluginTask + && !durableAuthority && stagedAttachment === null && providerEvents.length === 0; + const inheritedRunnerIdentity = inheritedCheckpointOnly && Boolean(row.run.processPid || row.run.processGroupId) + && await carriesInheritedRunnerIdentity(tx, row.run); let unclaimedBootstrap = remoteWorkspace?.state === "prepared" && row.run.runnerProfileJson?.nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION && @@ -1338,14 +1346,12 @@ export async function claimNativeRestartRecoveries(input: { // suspends. Without an owner, an elapsed expiry grants no authority. (!row.coordinator.leaseExpiresAt || row.coordinator.leaseExpiresAt <= now) && !row.coordinator.failureDetail?.successorRunId && - !row.run.processPid && - !row.run.processGroupId && - !row.run.processStartedAt && + (inheritedRunnerIdentity || (!row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt)) && !row.run.sessionIdBefore && !row.run.sessionIdAfter && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) && - !hasCheckpoint && - !hasProviderEvidence; + (!hasCheckpoint || inheritedCheckpointOnly) && + (!hasProviderEvidence || inheritedCheckpointOnly); if (unclaimedBootstrap) { if (activeNativeExecutionRunIds.has(row.run.id) && controller.bootId === (await currentNativeControllerIdentity()).bootId) { @@ -1430,9 +1436,6 @@ 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 @@ -1468,12 +1471,12 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessAlive: providerProcesses.livePids.length > 0, knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, - hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands, + hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands && !(unclaimedBootstrap && inheritedCheckpointOnly), stagedAttachmentVerified: stagedAttachment !== null, checkpointIdentityMatches: checkpointIdentityMatches && (!remoteSandbox || remoteWorkspace !== null), - hasProviderEvidence, + hasProviderEvidence: hasProviderEvidence && !(unclaimedBootstrap && inheritedCheckpointOnly), checkpointFailed: (checkpointRecord.terminal as Record | undefined) ?.runTerminalState === "failed" && 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 8b8ebc17c71..866f51003e3 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 @@ -2757,6 +2757,50 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_runner_never_received_commands" }); }); + it.each([ + { label: "restarts", providerEvent: false, process: "none" }, + { label: "restarts with its predecessor's runner identity", providerEvent: false, process: "inherited" }, + { label: "keeps its own runner identity held instead of restarting", providerEvent: false, process: "own" }, + { label: "keeps a projected provider event held instead of restarting", providerEvent: true, process: "none" }, + ])("$label an unclaimed remote follow-up that carries only its inherited checkpoint", async ({ providerEvent, process }) => { + const fixture = await seedUnlaunchedRemoteClaim(`UNCLAIMED-INHERITED-${providerEvent}-${process}`); + if (process !== "none") { + 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 (process === "inherited") { + const [current] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + 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 } }); + } + } + await fixture.db.update(nativeRunFinalizations).set({ attempt: 0, controllerGeneration: 0, controllerBootId: null, + controllerPid: null, controllerProcessStartedAt: null, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + 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 (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 (providerEvent || process === "own") { + expect(claim).not.toMatchObject({ kind: "bootstrap_incomplete" }); + return; + } + expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId }); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_execution_never_claimed" }); + }); + 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, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 7a4bc5ed81f..15aacf4dbd1 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -6744,9 +6744,14 @@ async function restorePerTurnRunnerdController(input: { runnerExecutionTarget?: AdapterExecutionTarget | null; restartRecovery?: NativeRestartRecoveryClaim; }): Promise { - if (input.restartRecovery || input.runnerExecutionTarget?.kind !== "remote" || + if ((input.restartRecovery && input.restartRecovery.kind !== "bootstrap_incomplete") || + input.runnerExecutionTarget?.kind !== "remote" || input.execution.session.lifecyclePolicy.mode !== "per_turn") return; const root = scopedRunnerdStateRoot(input.execution); + if (input.restartRecovery) { + const prior = readRunnerdDurableIdentity(root) ?? readRunnerdDurableIdentity(root, true); + if (!prior || prior.runId === input.execution.binding.runId) return; + } await restoreDurableHarnessForExecution(input.db, input.execution, root); const identity = readRunnerdDurableIdentity(root); const reference = identity ?? readRunnerdDurableIdentity(root, true);