diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 5b4c5639891..5ecc0a6309b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2285,7 +2285,7 @@ unknown operations or partial transcripts do not. Replacement uses a fresh session and retires only the exact predecessor's obsolete recovery hold while recording the proof and successor lineage. Retained provider files are not edited. -Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A resume after a crash, which leaves no drain record, still consumes one. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A resume after a crash, which leaves no drain record, still consumes one, unless the previous restart resume made progress before the crash. Each restart claim records the run's event-log position. Progress means that, after that position, the provider accepted a new turn and then completed work in that turn: a tool execution or a completed item other than a usage report. Restore activity, failed turns, and accepted turns without completed work are not progress, so a crash loop without progress and a resume that fails each time still use up the budget. This credit stops after 12 controller generations, so a run that makes progress and then crashes the controller each time is still bounded. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. A due failed remote retry can retain its provider PID after heartbeat completion releases the task execution lock. Both remote workspace reference formats and diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index e2e15b1b010..180e858a82d 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -159,6 +159,82 @@ export function suspendedByCleanDrain(coordinator: { return false; } +/** + * A controller restart (a crash, so no drain record) that interrupts a resume + * which had already made progress is also not the run's failure. Each claim + * records the run's event-log position; the next claim does not charge an + * attempt when, after that position, the provider accepted a new turn and then + * completed work in it (a tool execution or a non-usage item). A resume that + * dies before it completes any work in a new turn still charges, so a crash + * loop without progress, a resume that fails each time, and provider failures + * stay bounded. Credit stops after NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT + * controller generations, so a run that makes progress and then crashes the + * controller every time is still bounded. + */ +export const NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT = 12; +const RESUME_PROGRESS_EVENT_TYPES = ["tool.execution.completed", "item.completed"] as const; + +type RecoveryClaimProgressEntry = { disposition?: unknown; progressWatermarkSeq?: unknown; progressAttempt?: unknown }; + +/** + * The watermark of the most recent claim, only while the run is still on the + * provider attempt that claim resumed. A normal retry in between starts a new + * attempt, so the earlier resume's work no longer counts for it. + */ +export function lastRecoveryClaimProgressWatermark( + recoveryHistory: ReadonlyArray, + currentAttempt: number, +): number | null { + for (let index = recoveryHistory.length - 1; index >= 0; index -= 1) { + const entry = recoveryHistory[index]; + if (!RECOVERY_CLAIM_DISPOSITIONS.has(entry?.disposition)) continue; + if (entry?.progressAttempt !== currentAttempt) return null; + const watermark = entry?.progressWatermarkSeq; + return typeof watermark === "number" && Number.isSafeInteger(watermark) && watermark >= 0 ? watermark : null; + } + return null; +} + +async function nativeRunEventWatermark(tx: Pick, runId: string): Promise { + const [row] = await tx + .select({ seq: sql`max(${heartbeatRunEvents.seq})` }) + .from(heartbeatRunEvents) + .where(eq(heartbeatRunEvents.runId, runId)); + return Number(row?.seq ?? 0); +} + +export async function previousResumeMadeProgress(tx: Pick, input: { + runId: string; + coordinator: { + attempt: number; + controllerGeneration: number; + recoveryHistory: ReadonlyArray; + }; +}): Promise { + if (input.coordinator.attempt <= 0) return false; + if (input.coordinator.controllerGeneration >= NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT) return false; + const watermark = lastRecoveryClaimProgressWatermark(input.coordinator.recoveryHistory, input.coordinator.attempt); + if (watermark === null) return false; + const workTurnId = sql`${heartbeatRunEvents.payload} -> 'prpEvent' ->> 'turnId'`; + const [work] = await tx + .select({ seq: heartbeatRunEvents.seq }) + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.runId, input.runId), + gt(heartbeatRunEvents.seq, watermark), + inArray(heartbeatRunEvents.eventType, [...RESUME_PROGRESS_EVENT_TYPES]), + sql`coalesce(${heartbeatRunEvents.payload} -> 'prpEvent' -> 'payload' ->> 'kind', '') <> 'usage'`, + // The inner alias hides the table name, so unqualified outer references + // below resolve to the outer completed-work row. + sql`exists (select 1 from ${heartbeatRunEvents} as accepted_turn where accepted_turn.run_id = ${input.runId} + and accepted_turn.seq > ${watermark} and accepted_turn.seq < ${heartbeatRunEvents.seq} + and accepted_turn.event_type = 'turn.accepted' + and accepted_turn.payload -> 'prpEvent' ->> 'turnId' = ${workTurnId})`, + )) + .limit(1); + return work !== undefined; +} + const controllerBootId = randomUUID(); function processIsAlive(pid: number | null | undefined): boolean { @@ -1184,6 +1260,9 @@ export async function claimNativeRestartRecoveries(input: { // Read before this claim appends its own history entry. const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator); + const resumeKeepsAttempt = cleanDrainSuspension || await previousResumeMadeProgress(tx, { + runId: row.run.id, coordinator: row.coordinator, + }); const classification = classifyNativeRunnerRecoveryEvidence({ remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, @@ -1207,7 +1286,7 @@ export async function claimNativeRestartRecoveries(input: { ?.runTerminalState === "failed" && nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, providerAttempt: row.coordinator.attempt, - resumeChargesAttempt: !cleanDrainSuspension, + resumeChargesAttempt: !resumeKeepsAttempt, }); if (stopCleanup && classification.claimKind !== "reattach_remote_runner") { return { kind: "blocked", runId: row.run.id, reason: "native_stop_requires_existing_remote_session" } as const; @@ -1381,6 +1460,11 @@ export async function claimNativeRestartRecoveries(input: { recycledProviderPids: providerProcesses.recycledPids, stderrTail: row.run.stderrExcerpt, }); + // The next claim credits this resume only for work recorded after here. + const progressWatermarkSeq = await nativeRunEventWatermark(tx, row.run.id); + const attemptKeptForProgress = resumeKeepsAttempt && !cleanDrainSuspension; + // The attempt the executor runs this resume as (see its execution claim). + const progressAttempt = nextNativeProviderAttempt(providerAttempt, claimKind, !resumeKeepsAttempt); await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); const claimed = await tx .update(nativeRunFinalizations) @@ -1403,7 +1487,9 @@ export async function claimNativeRestartRecoveries(input: { ? "resuming_session" : "bootstrap_incomplete", recoveryRequestId: input.recoveryRequestId ?? null, - recoveryHistory: appendBoundedRecoveryHistory(event), + recoveryHistory: appendBoundedRecoveryHistory({ + ...event, progressWatermarkSeq, progressAttempt, ...(attemptKeptForProgress ? { attemptKeptForProgress } : {}), + }), failureCode: null, nextAttemptAt: null, updatedAt: now, @@ -1471,7 +1557,7 @@ export async function claimNativeRestartRecoveries(input: { providerAttempt, restartKind: input.restartKind, recoveryRequestId: input.recoveryRequestId ?? null, - chargesProviderAttempt: !cleanDrainSuspension, + chargesProviderAttempt: !resumeKeepsAttempt, }; if (claimKind === "reattach_remote_runner") { return { 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 73044a9a9f8..e1f9a1c7b08 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 @@ -86,7 +86,7 @@ vi.mock("../../sentry.js", async () => { captureRunFailure: mockCaptureRunFailure, }; }); -import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_RESTART_RECOVERY_FAILED, type NativeControllerIdentity } from "./native-restart-recovery.js"; +import { claimNativeRestartRecoveries, currentNativeControllerIdentity, NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT, NATIVE_RESTART_RECOVERY_FAILED, nextNativeProviderAttempt, type NativeControllerIdentity } from "./native-restart-recovery.js"; // Exact v5 wire prompt from before the task-monitor upgrade (#15446). const historicalPrompt = { @@ -2606,6 +2606,138 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( : { kind: "blocked", runId: fixture.runId, reason: "execution_recovery_budget_exhausted" })]); }); + // RC-22b: controller crashes (no drain record) that each interrupted a resume + // which had already done work exhausted the three-attempt budget. + async function seedInterruptedCheckpoint(label: string) { + const fixture = await seedRun(label); + await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { + driverKind: "opencode_server", + sessionId: fixture.native.normalizedSessionId, + identity: { companyId, issueId: fixture.issueId, runId: fixture.runId, agentId, sessionId: fixture.native.normalizedSessionId }, + providerSessionId: "interrupted-provider", + activeTurnId: null, + semanticResult: null, + terminal: { runTerminalState: "failed", turnTerminalState: "failed" }, + terminalTurns: [{ turnId: "interrupted-turn", fingerprint: JSON.stringify({ + terminalState: "failed", result: null, + error: { code: "provider_turn_lost_on_restore", recoverable: true }, + }) }], + } } }).where(eq(heartbeatRuns.id, fixture.runId)); + // The original execution claimed attempt 1. + await fixture.db.update(nativeRunFinalizations).set({ attempt: 1, controllerGeneration: 1 }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + return fixture; + } + + async function appendRunEvents(fixture: Awaited>, events: Array<{ + eventType: string; turnId?: string; kind?: string; + }>) { + const [last] = await fixture.db.select({ seq: sql`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` }) + .from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, fixture.runId)); + let seq = last?.seq ?? 0; + for (const event of events) { + seq += 1; + await fixture.db.insert(heartbeatRunEvents).values({ companyId, runId: fixture.runId, agentId, seq, + eventType: event.eventType, stream: "system", level: "info", + payload: { prpEvent: { runId: fixture.runId, eventType: event.eventType, sourceKind: "runner", + ...(event.turnId ? { turnId: event.turnId } : {}), + payload: event.kind ? { kind: event.kind } : {} } } }); + } + } + + // The new controller resumes, the executor applies the claim's charge to the + // attempt, and then the controller dies (a crash leaves no drain record). + async function resumeThenCrash(fixture: Awaited>) { + const [disposition] = await claimNativeRestartRecoveries({ + db: fixture.db, controller: successor, restartKind: "hard", runIds: [fixture.runId], + }); + if (disposition?.kind !== "resume_dead_runner") return { disposition }; + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + const attempt = nextNativeProviderAttempt(coordinator.attempt, disposition.kind, disposition.chargesProviderAttempt !== false); + await fixture.db.update(nativeRunFinalizations).set({ + attempt, controllerBootId: randomUUID(), controllerPid: 2_000_000_001, + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + return { disposition, attempt }; + } + + // What a resumed turn records on restore before it does any new work. + const restoreNoise = (round: number) => [ + { eventType: "item.completed", kind: "thread_lineage" }, + { eventType: "turn.failed", turnId: "interrupted-turn" }, + { eventType: "tool.execution.completed", turnId: "interrupted-turn" }, + { eventType: "turn.submitted" }, + { eventType: "turn.accepted", turnId: `continuation-${round}` }, + { eventType: "item.completed", turnId: `continuation-${round}`, kind: "usage" }, + ]; + + it("keeps recovering a run whose restart resumes each completed work before the next crash", async () => { + const fixture = await seedInterruptedCheckpoint("PROGRESS-BETWEEN-CRASHES"); + const attempts: number[] = []; + const charges: boolean[] = []; + for (let round = 1; round <= 5; round += 1) { + const { disposition, attempt } = await resumeThenCrash(fixture); + expect(disposition).toMatchObject({ kind: "resume_dead_runner", runId: fixture.runId }); + charges.push((disposition as { chargesProviderAttempt?: boolean }).chargesProviderAttempt !== false); + attempts.push(attempt!); + await appendRunEvents(fixture, [...restoreNoise(round), + { eventType: "tool.execution.completed", turnId: `continuation-${round}` }]); + } + // The first restart interrupted the original execution, not a resume. + expect(charges).toEqual([true, false, false, false, false]); + expect(attempts).toEqual([2, 2, 2, 2, 2]); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect((coordinator.recoveryHistory as Array>).filter(entry => entry.attemptKeptForProgress)) + .toHaveLength(4); + }); + + it("does not credit a resume's work to a later provider attempt started by a normal retry", async () => { + const fixture = await seedInterruptedCheckpoint("PROGRESS-THEN-RETRY"); + await resumeThenCrash(fixture); + await appendRunEvents(fixture, [{ eventType: "turn.accepted", turnId: "resumed" }, + { eventType: "tool.execution.completed", turnId: "resumed" }]); + // The resume (attempt 2) did work. A transport failure then started a + // normal local retry at attempt 3, which crashed before doing any work. + await fixture.db.update(nativeRunFinalizations).set({ attempt: 3 }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + const { disposition } = await resumeThenCrash(fixture); + expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" }); + }); + + it("still exhausts the budget when restart resumes complete no work between crashes", async () => { + const fixture = await seedInterruptedCheckpoint("NO-PROGRESS-BETWEEN-CRASHES"); + const outcomes: Array<{ kind?: string; attempt?: number; reason?: string }> = []; + for (let round = 1; round <= 3; round += 1) { + const { disposition, attempt } = await resumeThenCrash(fixture); + outcomes.push({ kind: disposition?.kind, attempt, reason: (disposition as { reason?: string }).reason }); + // Restore activity and an accepted turn that fails before any work. + await appendRunEvents(fixture, [...restoreNoise(round), { eventType: "turn.failed", turnId: `continuation-${round}` }]); + } + expect(outcomes).toEqual([ + expect.objectContaining({ kind: "resume_dead_runner", attempt: 2 }), + expect.objectContaining({ kind: "resume_dead_runner", attempt: 3 }), + expect.objectContaining({ kind: "blocked", reason: "execution_recovery_budget_exhausted" }), + ]); + }); + + it("bounds a crash loop that makes progress before every crash", async () => { + const fixture = await seedInterruptedCheckpoint("PROGRESS-CRASH-LOOP"); + const outcomes: Array<{ kind?: string; attempt?: number }> = []; + for (let round = 1; round <= NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT + 3; round += 1) { + const { disposition, attempt } = await resumeThenCrash(fixture); + outcomes.push({ kind: disposition?.kind, attempt }); + if (disposition?.kind !== "resume_dead_runner") break; + await appendRunEvents(fixture, [{ eventType: "turn.accepted", turnId: `loop-${round}` }, + { eventType: "item.completed", turnId: `loop-${round}`, kind: "assistant_message" }]); + } + // Generations 2..12 are claimed; credit stops at the generation limit and + // the remaining attempt is spent before recovery blocks. + expect(outcomes.at(-1)).toEqual({ kind: "blocked", attempt: undefined }); + expect(outcomes.slice(0, -1).map(outcome => outcome.attempt)) + .toEqual([2, ...Array(NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT - 2).fill(2), 3]); + }); + it("terminalizes the recorded failed-checkpoint incident atomically instead of resuming it on upgrade", async () => { const fixture = await seedRun("FAILED-CHECKPOINT"); await fixture.db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: {