diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 8f0c30290c2..e4473596965 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4198,6 +4198,14 @@ its normalized current run ID does not prove that the current controller attache Fenced remote Runner recovery requires a matching database controller checkpoint or a valid staged attachment for the target run before selecting exact-checkpoint resume. Local Runner recovery keeps its existing local controller checks. +A follow-up run that a restart interrupted before its remote Runner launched +carries only that copied checkpoint. Restart recovery does not treat the copied +provider session as this run's provider work. When the run saved no controller +checkpoint, or its controller checkpoint records no provider turn, and no Runner +process identity or Runner-sent event exists for the run, recovery bootstraps +the run again instead of attempting to reattach a Runner that never started. A +copied session counts only when a settled earlier run of the same session saved +that exact provider session. If the first attachment finds the retained runner gone, or replacement cannot archive its remote state before the new turn starts, ordinary failure settlement requests one normal Retry under the diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 28a6505dea8..a24545c1ad6 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -25,7 +25,7 @@ import { recordNativeLocalProcessStopInTransaction, } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; -import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; +import { and, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, ne, or, sql } from "drizzle-orm"; import { alias } from "drizzle-orm/pg-core"; import type { Db } from "@paperclipai/db"; import { @@ -821,6 +821,24 @@ async function hasNativeRunnerOriginatedEvidence( return event !== undefined; } +/** A settled earlier run of the same native session saved this exact provider + * session. Admission rebinds that checkpoint to a follow-up run, so the + * follow-up's copy of the provider session is carried, not its own work. */ +async function carriesSettledPredecessorSession( + tx: Pick, + run: { companyId: string; id: string; nativeSessionId: string | null; createdAt: Date }, + providerSessionId: unknown, +): Promise { + if (typeof providerSessionId !== "string" || !providerSessionId || !run.nativeSessionId) return false; + const [predecessor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.nativeSessionId, run.nativeSessionId), + ne(heartbeatRuns.id, run.id), lt(heartbeatRuns.createdAt, run.createdAt), isNotNull(heartbeatRuns.finishedAt), + inArray(heartbeatRuns.status, ["succeeded", "failed", "cancelled", "timed_out", "interrupted"]), + sql`${heartbeatRuns.runnerProfileJson}#>>'{sessionCheckpoint,providerSessionId}' = ${providerSessionId}`, + )).limit(1); + return predecessor !== undefined; +} + async function hasNativeRunEvent( tx: Pick, run: { companyId: string; id: string }, eventType: string, ): Promise { @@ -1516,18 +1534,29 @@ export async function claimNativeRestartRecoveries(input: { // provider session: session-open events may already be projected here, // but no session snapshot or turn exists. Those events alone do not make // the checkpoint proof ambiguous; any turn or provider event does. - const providerEvidenceIsSessionOpenOnly = !hasCheckpointProviderIdentity + // A follow-up on a continued session carries only its rebound admission + // checkpoint: the previous turn's provider session, with no cursor, + // turn, result or request of this run. When a settled earlier run of the + // same session saved that exact provider session, the carried session is + // not this run's provider work, so it does not make the controller + // checkpoint proof ambiguous (Daytona waves s9202, s9205). + const reboundSessionOnly = continuedSessionOnly && Boolean(durableAuthority) && remoteWorkspace !== null + && hasCheckpointProviderIdentity && !unclaimedBootstrap && !pluginTask && stagedAttachment === null + && await carriesSettledPredecessorSession(tx, row.run, checkpointRecord.providerSessionId); + const providerEvidenceIsSessionOpenOnly = (!hasCheckpointProviderIdentity || reboundSessionOnly) && providerEvents.every(event => SESSION_OPEN_PROVIDER_EVENT_TYPES.has(event.eventType)); const remoteBootstrapProof = !unclaimedBootstrap && remoteWorkspace !== null && !pluginTask - && Boolean(durableAuthority) && stagedAttachment === null && !hasCheckpoint && providerEvidenceIsSessionOpenOnly + && Boolean(durableAuthority) && stagedAttachment === null && (!hasCheckpoint || reboundSessionOnly) + && providerEvidenceIsSessionOpenOnly && Boolean(row.run.runnerInstanceId && row.run.nativeSessionId) ? await nativeRunnerCheckpointRemoteBootstrapProof(tx, { companyId: row.run.companyId, issueId: row.coordinator.issueId, runId: row.run.id, }, { runnerInstanceId: row.run.runnerInstanceId!, normalizedSessionId: row.run.nativeSessionId! }) : null; // Projected session-open events need the stronger session-opened proof. + const ownProviderEvidence = reboundSessionOnly ? providerEvents.length > 0 : hasProviderEvidence; const remoteBootstrapProvenIncomplete = remoteBootstrapProof !== null - && (!hasProviderEvidence || remoteBootstrapProof === "session_opened_without_turn"); + && (!ownProviderEvidence || remoteBootstrapProof === "session_opened_without_turn"); const remoteRunnerNeverReceivedCommands = !unclaimedBootstrap && !remoteRunnerNeverLaunched && !remoteRunnerExitedBeforeProviderWork && !remoteBootstrapProvenIncomplete @@ -1538,8 +1567,10 @@ export async function claimNativeRestartRecoveries(input: { && 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 + // A follow-up may carry its predecessor's settled runner identity and + // session; neither is this run's runner or provider work. + && (inheritedRunnerIdentity || (!row.run.processPid && !row.run.processGroupId && !row.run.processStartedAt)) + && (!row.run.sessionIdBefore || continuedSessionOnly) && !row.run.sessionIdAfter && Boolean(row.run.nativeSessionId && row.run.runnerInstanceId) && !(await hasNativeRunnerOriginatedEvidence(tx, row.run)) && !(await hasNativeRunEvent(tx, row.run, PROCESS_IDENTITY_RECORDED)); @@ -1565,7 +1596,7 @@ export async function claimNativeRestartRecoveries(input: { knownProviderProcessIdentityAmbiguous: providerProcesses.ambiguousLivePids.length > 0, hasCheckpoint: hasCheckpoint && !remoteRunnerNeverReceivedCommands && !(unclaimedBootstrap && inheritedCheckpointOnly) - && !remoteRunnerNeverLaunched, + && !remoteRunnerNeverLaunched && !(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 ba518e9e8b5..dc06ac192ac 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 @@ -2588,18 +2588,20 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); attempt?: number; mutate?: (core: DurablePrpControlPlane) => void; checkpoint?: boolean; + /** False: the controller saved its checkpoint, but no remote runner was launched yet. */ + launched?: boolean; } = {}) { const fixture = await seedRun(`REMOTE-CLAIMED-${label}`); const { runId, issueId, native } = fixture; const remote = remotePreparedWorkspace(); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.preparedProfile, nativeWorkspaceSync: remote }, - processPid: 443, processStartedAt: new Date(Date.now() - 30_000), + ...(options.launched === false ? {} : { processPid: 443, processStartedAt: new Date(Date.now() - 30_000) }), }).where(eq(heartbeatRuns.id, runId)); await fixture.db.insert(heartbeatRunEvents).values([ { companyId, runId, agentId, seq: 900, stream: "system", level: "info", eventType: "native.process_start_requested" }, - { companyId, runId, agentId, seq: 901, stream: "system", level: "info", eventType: "native.process_identity_recorded", - payload: { targetKind: "remote", processPid: 443 } }, + ...(options.launched === false ? [] : [{ companyId, runId, agentId, seq: 901, stream: "system", level: "info", + eventType: "native.process_identity_recorded", payload: { targetKind: "remote", processPid: 443 } }]), ]); const bootId = randomUUID(); const leaseOwner = `${bootId}:1:${randomUUID()}`; @@ -2926,6 +2928,86 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_runner_never_received_commands" }); }); + // Daytona waves s9202 t02 (B1) and s9205 t04 (B2): a follow-up run on a + // continued session was killed after native.process_start_requested but + // before its remote runner launched. Admission rebinds the previous turn's + // checkpoint to the new run: its identity names this run and it keeps the + // provider session ID, but it records no cursor, turn, result or request. + // That carried session is not this run's provider work. + async function reboundFollowUpCheckpoint(fixture: Awaited>, extra: Record = {}, + predecessor = true) { + const [seeded] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + // The settled previous turn of the same session saved the provider session. + if (predecessor) await fixture.db.insert(heartbeatRuns).values({ companyId, agentId, status: "succeeded", runtimeMode: "native", + nativeIssueId: fixture.issueId, nativeSessionId: seeded.nativeSessionId, runnerInstanceId: seeded.runnerInstanceId, + createdAt: new Date(seeded.createdAt.getTime() - 120_000), finishedAt: new Date(seeded.createdAt.getTime() - 10_000), + contextSnapshot: { issueId: fixture.issueId }, runnerProfileJson: { sessionCheckpoint: { providerSessionId: "inherited-thread" } } }); + await fixture.db.update(heartbeatRuns).set({ sessionIdBefore: "inherited-thread", 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: [], ...extra, + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + } + + it.each([ + { label: "restarts", mode: "unlaunched" }, + { label: "restarts with its predecessor's remote runner identity", mode: "inherited_process" }, + { label: "keeps its own recorded runner held", mode: "identity_recorded" }, + { label: "keeps a carried active turn held", mode: "active_turn" }, + ] as const)("$label a follow-up whose controller never saved a checkpoint (s9205 t04)", async ({ mode }) => { + const fixture = await seedUnlaunchedRemoteClaim(`FOLLOW-UP-UNCOMMANDED-${mode}`, + { event: { eventType: "native.process_start_requested" } }); + await reboundFollowUpCheckpoint(fixture, mode === "active_turn" ? { activeTurnId: "turn-in-flight" } : {}); + if (mode === "inherited_process") { + const processStartedAt = new Date(Date.now() - 120_000); + const [current] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(heartbeatRuns).set({ processPid: 596, processGroupId: null, processStartedAt, processHostHash: "remote" }) + .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 } }); + } + if (mode === "identity_recorded") 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 (mode === "identity_recorded" || mode === "active_turn") { + 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.each([ + { label: "restarts", mode: "bootstrap" }, + { label: "keeps a provider session no settled earlier run saved held in", mode: "no_predecessor" }, + { label: "keeps a turn command held", mode: "turn_command" }, + { label: "keeps a projected provider turn event held", mode: "turn_event" }, + ] as const)("$label a follow-up whose saved controller checkpoint proves no provider turn (s9202 t02)", async ({ mode }) => { + const fixture = await seedClaimedRemoteBootstrap(`FOLLOW-UP-CLAIMED-${mode}`, { + launched: false, + mutate: mode === "turn_command" ? core => { openSession(core); + core.queueCommand("turn.start", { prompt: "Prior work must stay held." }); } : undefined, + }); + await reboundFollowUpCheckpoint(fixture, {}, mode !== "no_predecessor"); + if (mode === "turn_event") await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, + seq: 902, stream: "system", level: "info", eventType: "turn.started" }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + if (mode !== "bootstrap") { + expect(claim).not.toMatchObject({ kind: "bootstrap_incomplete" }); + return; + } + expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId, providerAttempt: 1, controllerGeneration: 2 }); + }); + it.each([ { label: "restarts", providerEvent: false, process: "none" }, { label: "restarts with its predecessor's runner identity", providerEvent: false, process: "inherited" },