diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 622b5652c99..4e92bd2d158 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2306,7 +2306,7 @@ replacement behavior. When a replacement does archive remote continuity state and the remote move fails, the local state is not moved, and the run error keeps the bounded, redacted exit status and output of that move. -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. +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. A resume that launches a new runner and that the controller died before starting is not charged either (a reattachment adopts a provider that already started, so only progress counts for it): after the previous claim's position, no runner was requested and no runner event was recorded. That resume used no provider work, so the next claim replaces it and the attempt ends at the attempt that resume was to run as. 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 b62e1c68d7e..480ad9248a1 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -169,7 +169,8 @@ export function suspendedByCleanDrain(coordinator: { * 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 + * stay bounded. A resume the controller died before starting is not charged + * either (restartResumeKeepsAttempt). 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. */ @@ -179,24 +180,49 @@ const RESUME_PROGRESS_EVENT_TYPES = ["tool.execution.completed", "item.completed 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. + * The most recent recovery claim's event-log position and the provider attempt + * its resume runs as. Entries from releases that did not record them give none. */ -export function lastRecoveryClaimProgressWatermark( +export function lastRecoveryClaimResume( recoveryHistory: ReadonlyArray, - currentAttempt: number, -): number | null { +): { watermark: number; attempt: number; launchesRunner: boolean } | 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; + const attempt = entry?.progressAttempt; + // A reattachment adopts a runner that had already started its provider. + const launchesRunner = entry?.disposition === "resume_dead_runner" || entry?.disposition === "bootstrap_incomplete"; + return typeof watermark === "number" && Number.isSafeInteger(watermark) && watermark >= 0 + && typeof attempt === "number" && Number.isSafeInteger(attempt) + ? { watermark, attempt, launchesRunner } : null; } return null; } +/** + * Whether the resume claimed at `watermark` reached execution. The executor + * records native.process_start_requested before any backend may spawn a + * runner, and every runner event carries a source event ID. + */ +async function nativeExecutionStartedSince(tx: Pick, runId: string, watermark: number): Promise { + const [started] = await tx + .select({ seq: heartbeatRunEvents.seq }) + .from(heartbeatRunEvents) + .where(and( + eq(heartbeatRunEvents.runId, runId), + gt(heartbeatRunEvents.seq, watermark), + or( + isNotNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [ + "adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, "tool.execution.started", + ]), + ), + )) + .limit(1); + return started !== undefined; +} + async function nativeRunEventWatermark(tx: Pick, runId: string): Promise { const [row] = await tx .select({ seq: sql`max(${heartbeatRunEvents.seq})` }) @@ -205,7 +231,19 @@ async function nativeRunEventWatermark(tx: Pick, runId: string): P return Number(row?.seq ?? 0); } -export async function previousResumeMadeProgress(tx: Pick, input: { +/** + * Whether this restart claim keeps the provider attempt instead of charging + * one. Both cases are judged from the previous claim's recorded position: + * - That resume launched a new runner (not a reattachment) and never started: + * no runner was requested and no runner event was + * recorded before the controller died (for example in a K6 crash loop). It + * used no provider work, so this claim replaces it and carries its charge: + * the attempt ends at the attempt that resume was to run as, never above it. + * - That resume made progress on the current attempt (see above). A normal + * retry in between starts a new attempt, and earlier work does not count. + * Neither applies at or after NATIVE_PROGRESS_CREDIT_GENERATION_LIMIT. + */ +export async function restartResumeKeepsAttempt(tx: Pick, input: { runId: string; coordinator: { attempt: number; @@ -215,8 +253,14 @@ export async function previousResumeMadeProgress(tx: Pick, input: }): 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 previous = lastRecoveryClaimResume(input.coordinator.recoveryHistory); + if (previous === null) return false; + // Only a claim that launches a new runner can be unstarted. A reattachment + // adopted provider work that had already started, so only progress counts. + if (previous.launchesRunner && !await nativeExecutionStartedSince(tx, input.runId, previous.watermark)) + return input.coordinator.attempt >= previous.attempt; + if (previous.attempt !== input.coordinator.attempt) return false; + const watermark = previous.watermark; const workTurnId = sql`${heartbeatRunEvents.payload} -> 'prpEvent' ->> 'turnId'`; const [work] = await tx .select({ seq: heartbeatRunEvents.seq }) @@ -1262,7 +1306,7 @@ 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, { + const resumeKeepsAttempt = cleanDrainSuspension || await restartResumeKeepsAttempt(tx, { runId: row.run.id, coordinator: row.coordinator, }); const classification = classifyNativeRunnerRecoveryEvidence({ 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 2d01395321f..887f5b1c5f9 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 @@ -2808,6 +2808,7 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); // What a resumed turn records on restore before it does any new work. const restoreNoise = (round: number) => [ + { eventType: "native.process_start_requested" }, { eventType: "item.completed", kind: "thread_lineage" }, { eventType: "turn.failed", turnId: "interrupted-turn" }, { eventType: "tool.execution.completed", turnId: "interrupted-turn" }, @@ -2840,12 +2841,13 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); 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" }]); + await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }, + { 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)); + await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }]); const { disposition } = await resumeThenCrash(fixture); expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" }); }); @@ -2866,6 +2868,82 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); ]); }); + // RC-22c: in a K6 crash loop the controller died again before the resumed + // execution requested a runner, and the next claim charged that resume. + it.each([ + { label: "after the executor took its lease", leaseTaken: true }, + { label: "before the executor took its lease", leaseTaken: false }, + ])("does not charge for a resume killed before it started the provider ($label)", async ({ leaseTaken }) => { + const fixture = await seedInterruptedCheckpoint(`UNSTARTED-RESUME-${leaseTaken ? "LEASE" : "NO-LEASE"}`); + // The first restart resume runs and makes progress. + expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 2 }); + await appendRunEvents(fixture, [...restoreNoise(1), { eventType: "tool.execution.completed", turnId: "continuation-1" }]); + const outcomes: Array<{ kind?: string; attempt?: number }> = []; + // Four crash restarts in a row, each before the resumed execution started. + for (let round = 0; round < 4; round += 1) { + if (leaseTaken) { + const { disposition, attempt } = await resumeThenCrash(fixture); + outcomes.push({ kind: disposition?.kind, attempt }); + continue; + } + const [disposition] = await claimNativeRestartRecoveries({ + db: fixture.db, controller: successor, restartKind: "hard", runIds: [fixture.runId], + }); + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), controllerPid: 2_000_000_001 }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + outcomes.push({ kind: disposition?.kind, attempt: coordinator.attempt }); + } + expect(outcomes).toEqual(Array(4).fill({ kind: "resume_dead_runner", attempt: 2 })); + // A resume that starts and then fails before any work is charged again. + await appendRunEvents(fixture, [...restoreNoise(2), { eventType: "turn.failed", turnId: "continuation-2" }]); + expect(await resumeThenCrash(fixture)).toMatchObject({ disposition: { kind: "resume_dead_runner" }, attempt: 3 }); + }); + + it("does not treat a reattachment with no later events as an unstarted resume", async () => { + const fixture = await seedInterruptedCheckpoint("REATTACH-NO-EVENTS"); + await resumeThenCrash(fixture); + await appendRunEvents(fixture, [...restoreNoise(1), { eventType: "turn.failed", turnId: "continuation-1" }]); + expect(await resumeThenCrash(fixture)).toMatchObject({ attempt: 3 }); + // That claim was a reattachment to a runner whose provider had already + // started. The controller then died before recording anything else, and the + // adopted runner died too. + await fixture.db.update(nativeRunFinalizations).set({ + recoveryHistory: sql`jsonb_set(${nativeRunFinalizations.recoveryHistory}, + array[(jsonb_array_length(${nativeRunFinalizations.recoveryHistory}) - 1)::text, 'disposition'], + '"reattach_existing_runner"'::jsonb)`, + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const { disposition } = await resumeThenCrash(fixture); + expect(disposition).toMatchObject({ kind: "blocked", reason: "execution_recovery_budget_exhausted" }); + }); + + it("charges the owed attempt once when an unstarted first resume is replaced", async () => { + const fixture = await seedInterruptedCheckpoint("UNSTARTED-FIRST-RESUME"); + // The first restart charges, but the controller dies before the executor + // takes its lease, so the charge was never applied. + await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, restartKind: "hard", runIds: [fixture.runId] }); + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: randomUUID(), controllerPid: 2_000_000_001 }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + // The replacement carries that charge, once. + const { disposition, attempt } = await resumeThenCrash(fixture); + expect(disposition).toMatchObject({ kind: "resume_dead_runner", chargesProviderAttempt: true }); + expect(attempt).toBe(2); + }); + + it("bounds a crash loop that never starts the provider", async () => { + const fixture = await seedInterruptedCheckpoint("UNSTARTED-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; + } + 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("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 }> = []; @@ -2873,7 +2951,8 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); 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}` }, + await appendRunEvents(fixture, [{ eventType: "native.process_start_requested" }, + { 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