From 2a895ca8dbfb6b8a1f0c992e19368b8a0925acb0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 00:45:59 -0700 Subject: [PATCH] fix(cron): prevent completed runs retrying after claim conflicts (#108682) * fix(cron): avoid replay after lifecycle conflicts * test(cron): type lifecycle race fixture * chore: leave changelog to release flow * fix(protocol): sync Swift session creation model --- .../run.session-lifecycle.test.ts | 69 +++++++++++++++++++ src/cron/isolated-agent/run.ts | 3 + src/cron/retry-hint.test.ts | 53 +++++++++----- src/cron/retry-hint.ts | 20 +++--- ...runs-one-shot-main-job-disables-it.test.ts | 25 +++++++ src/cron/service/timer.ts | 19 +++-- src/cron/types.ts | 2 + 7 files changed, 161 insertions(+), 30 deletions(-) diff --git a/src/cron/isolated-agent/run.session-lifecycle.test.ts b/src/cron/isolated-agent/run.session-lifecycle.test.ts index 73570f04c5a5..38cb66f53e5b 100644 --- a/src/cron/isolated-agent/run.session-lifecycle.test.ts +++ b/src/cron/isolated-agent/run.session-lifecycle.test.ts @@ -1,5 +1,6 @@ // Persistent cron session tests cover lifecycle admission and mutation races. import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { SessionEntry } from "../../config/sessions.js"; import { interruptSessionWorkAdmissions, isSessionWorkAdmissionActive, @@ -14,6 +15,7 @@ import { makeCronSession, makeCronSessionEntry, mockRunCronFallbackPassthrough, + patchSessionEntryMock, preflightCronModelProviderMock, resetRunCronIsolatedAgentTurnHarness, resolveCronSessionMock, @@ -269,6 +271,73 @@ describe("runCronIsolatedAgentTurn session lifecycle", () => { expect(isSessionWorkAdmissionActive(storePath, [sessionKey, sessionId])).toBe(false); }); + it("marks a final lifecycle claim conflict as post-execution (#108428)", async () => { + const sessionKey = "agent:main:main"; + const initialSessionEntry = makeCronSessionEntry({ sessionId: "persistent-session" }); + resolveCronSessionMock.mockReturnValue( + makeCronSession({ + storePath: inMemoryStorePath, + store: { [sessionKey]: { ...initialSessionEntry } }, + initialSessionEntry, + isNewSession: false, + sessionEntry: { ...initialSessionEntry }, + }), + ); + loadSessionEntryMock.mockReturnValue({ ...initialSessionEntry }); + + let agentExecutionStarted = false; + runEmbeddedAgentMock.mockImplementationOnce( + async (runParams: { onExecutionStarted?: () => void }) => { + runParams.onExecutionStarted?.(); + agentExecutionStarted = true; + return { + payloads: [{ text: "completed" }], + meta: { agentMeta: {} }, + }; + }, + ); + + const committedRows = new Map([ + [`${inMemoryStorePath}\0${sessionKey}`, structuredClone(initialSessionEntry) as SessionEntry], + ]); + patchSessionEntryMock.mockImplementation( + async ( + scope: { storePath?: string; sessionKey: string }, + update: ( + entry: SessionEntry, + context: { existingEntry: SessionEntry | undefined }, + ) => SessionEntry | null, + options: { fallbackEntry?: SessionEntry } = {}, + ) => { + const key = `${scope.storePath ?? ""}\0${scope.sessionKey}`; + const current = committedRows.get(key); + const writeBase = current ?? options.fallbackEntry; + if (!writeBase) { + return null; + } + const existingEntry = + agentExecutionStarted && scope.sessionKey === sessionKey + ? { ...writeBase, lifecycleRevision: "replacement-revision" } + : current; + const committed = update(structuredClone(writeBase), { + existingEntry: existingEntry ? structuredClone(existingEntry) : undefined, + }); + if (committed) { + committedRows.set(key, structuredClone(committed)); + } + return committed; + }, + ); + + await expect( + runCronIsolatedAgentTurn(makePersistentCronParams(sessionKey)), + ).resolves.toMatchObject({ + status: "error", + error: `CronSessionLifecycleClaimError: Session "${sessionKey}" changed while starting work. Retry.`, + executionStarted: true, + }); + }); + it("releases a custom cron session lease before delete-after-run cleanup", async () => { const sessionKey = "agent:main:cron:cleanup"; const sessionId = "custom-cron-session"; diff --git a/src/cron/isolated-agent/run.ts b/src/cron/isolated-agent/run.ts index 4ec4ed5a45a1..2e4b9b80a3f0 100644 --- a/src/cron/isolated-agent/run.ts +++ b/src/cron/isolated-agent/run.ts @@ -1654,7 +1654,9 @@ export async function runCronIsolatedAgentTurn(params: { const ownsRunContext = params.job.sessionTarget === "isolated"; let runContextOwnerToken: string | undefined; let runLifecycleGeneration = admittedLifecycleGeneration; + let executionStarted = false; const notifyExecutionStarted = (info?: { lifecycleGeneration?: string }) => { + executionStarted = true; if (info?.lifecycleGeneration) { runLifecycleGeneration = info.lifecycleGeneration; } @@ -1800,6 +1802,7 @@ export async function runCronIsolatedAgentTurn(params: { return prepared.context.withRunSession({ status: "error", error, + executionStarted, // Carry the already-resolved run model into the error/timeout row so // Task-run history keeps provider/model attribution instead of looking like // an un-attributed cron timeout. finalizeCronRun does the same via diff --git a/src/cron/retry-hint.test.ts b/src/cron/retry-hint.test.ts index 0dcea0d09963..8a597941796e 100644 --- a/src/cron/retry-hint.test.ts +++ b/src/cron/retry-hint.test.ts @@ -8,14 +8,16 @@ import { describe("resolveCronExecutionRetryHint", () => { it("matches classified transient errors", () => { - expect(resolveCronExecutionRetryHint("HTTP 529", ["overloaded"])).toEqual({ + expect(resolveCronExecutionRetryHint({ error: "HTTP 529", retryOn: ["overloaded"] })).toEqual({ retryable: true, category: "overloaded", }); - expect(resolveCronExecutionRetryHint("429 rate limit exceeded", ["rate_limit"])).toEqual({ - retryable: true, - category: "rate_limit", - }); + expect( + resolveCronExecutionRetryHint({ + error: "429 rate limit exceeded", + retryOn: ["rate_limit"], + }), + ).toEqual({ retryable: true, category: "rate_limit" }); }); it("treats common network error codes as network when retryOn only includes network", () => { @@ -28,22 +30,26 @@ describe("resolveCronExecutionRetryHint", () => { "ENETUNREACH", "EPIPE", ]) { - expect(resolveCronExecutionRetryHint(`temporary DNS failure: ${code}`, ["network"])).toEqual({ - retryable: true, - category: "network", - }); + expect( + resolveCronExecutionRetryHint({ + error: `temporary DNS failure: ${code}`, + retryOn: ["network"], + }), + ).toEqual({ retryable: true, category: "network" }); } }); it("does not retry permanent errors", () => { - expect(resolveCronExecutionRetryHint("invalid API key", ["network"])).toEqual({ + expect( + resolveCronExecutionRetryHint({ error: "invalid API key", retryOn: ["network"] }), + ).toEqual({ retryable: false, }); }); it("classifies cron pre-execution watchdog failures as timeout retries", () => { for (const message of [setupTimeoutErrorMessage(), preExecutionTimeoutErrorMessage()]) { - expect(resolveCronExecutionRetryHint(message, ["timeout"])).toEqual({ + expect(resolveCronExecutionRetryHint({ error: message, retryOn: ["timeout"] })).toEqual({ retryable: true, category: "timeout", }); @@ -60,7 +66,7 @@ describe("resolveCronExecutionRetryHint", () => { "error 500 got 0", "process exited with code 500", ]) { - expect(resolveCronExecutionRetryHint(message, ["server_error"])).toEqual({ + expect(resolveCronExecutionRetryHint({ error: message, retryOn: ["server_error"] })).toEqual({ retryable: false, }); } @@ -77,7 +83,7 @@ describe("resolveCronExecutionRetryHint", () => { "503", "500", ]) { - expect(resolveCronExecutionRetryHint(message, ["server_error"])).toEqual({ + expect(resolveCronExecutionRetryHint({ error: message, retryOn: ["server_error"] })).toEqual({ retryable: true, category: "server_error", }); @@ -90,15 +96,28 @@ describe("resolveCronExecutionRetryHint", () => { 'Error: Session "agent:main:cron:job-1" changed while starting work. Retry.', 'Error: Session "agent:main:cron:job-1" was deleted while starting work. Retry.', ]) { - expect(resolveCronExecutionRetryHint(message, ["network"])).toEqual({ retryable: true }); + expect(resolveCronExecutionRetryHint({ error: message, retryOn: ["network"] })).toEqual({ + retryable: true, + }); } }); + it("does not retry lifecycle claim conflicts after execution starts (#108428)", () => { + expect( + resolveCronExecutionRetryHint({ + error: + 'CronSessionLifecycleClaimError: Session "agent:main:cron:job-1" changed while starting work. Retry.', + retryOn: ["network"], + executionStarted: true, + }), + ).toEqual({ retryable: false }); + }); + it("does not classify archived-session work-start errors as transient", () => { expect( - resolveCronExecutionRetryHint( - 'Error: Session "agent:main:main" is archived. Restore it before starting new work.', - ), + resolveCronExecutionRetryHint({ + error: 'Error: Session "agent:main:main" is archived. Restore it before starting new work.', + }), ).toEqual({ retryable: false }); }); }); diff --git a/src/cron/retry-hint.ts b/src/cron/retry-hint.ts index ed7614d1e988..e5401cf8db17 100644 --- a/src/cron/retry-hint.ts +++ b/src/cron/retry-hint.ts @@ -7,6 +7,13 @@ type CronRetryHint = { category?: CronRetryOn; }; +type CronRetryHintInput = { + error: string | undefined; + retryOn?: CronRetryOn[]; + classifiedReason?: string | null; + executionStarted?: boolean; +}; + // A bare 5xx-looking number embedded in prose is not an HTTP server error: cron // failure messages routinely contain such numbers ("context limit 512 exceeded", // "exited with 503 lines", "pid 511 killed", a "...-540.sock" path), and @@ -18,8 +25,8 @@ type CronRetryHint = { const SERVER_ERROR_PATTERN = /\b(?:https?|status(?:[ _]code)?|response(?:[ _]code)?|http(?:[ _]status)?)\b[\s:=#"']{0,4}5\d{2}\b|\b5\d{2}\b[\s:)\].,-]*(?:internal server error|server error|bad gateway|service unavailable|gateway time-?out)\b|\binternal server error\b|\bbad gateway\b|\bservice unavailable\b|\bgateway time-?out\b|\b5xx\b|^\s*5\d{2}\s*$/i; -// Lifecycle claims can lose a race before provider execution. Retry the run; -// treating the conflict as permanent disables one-shot jobs without running them. +// Lifecycle claims can lose a race before provider execution. Retry only before +// execution starts; afterward, tools may have produced non-idempotent effects. const SESSION_LIFECYCLE_CLAIM_ERROR_PATTERN = /^(?:(?:CronSessionLifecycleClaimError|Error): )?Session "[^"\n]+" (?:changed|was deleted) while starting work\. Retry\.$/; @@ -35,16 +42,13 @@ const TRANSIENT_PATTERNS: Record = { }; /** Classifies cron execution errors against the configured retryable transient categories. */ -export function resolveCronExecutionRetryHint( - error: string | undefined, - retryOn?: CronRetryOn[], - classifiedReason?: string | null, -): CronRetryHint { +export function resolveCronExecutionRetryHint(input: CronRetryHintInput): CronRetryHint { + const { error, retryOn, classifiedReason, executionStarted } = input; if (!error || typeof error !== "string") { return { retryable: false }; } if (SESSION_LIFECYCLE_CLAIM_ERROR_PATTERN.test(error)) { - return { retryable: true }; + return { retryable: executionStarted !== true }; } const keys = retryOn?.length ? retryOn : (Object.keys(TRANSIENT_PATTERNS) as CronRetryOn[]); const classified = classifiedReason ?? undefined; diff --git a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts index 4dd5321e777e..89535dc549be 100644 --- a/src/cron/service.runs-one-shot-main-job-disables-it.test.ts +++ b/src/cron/service.runs-one-shot-main-job-disables-it.test.ts @@ -668,6 +668,31 @@ describe("CronService", () => { await stopCronAndCleanup(cron, store); }); + it("does not retry a lifecycle claim conflict after agent execution starts (#108428)", async () => { + const runIsolatedAgentJob = vi.fn(async () => ({ + status: "error" as const, + error: + 'CronSessionLifecycleClaimError: Session "agent:main:cron:job-1" changed while starting work. Retry.', + executionStarted: true, + })); + const { store, cron, events } = await createIsolatedAnnounceHarness(runIsolatedAgentJob); + const job = await runIsolatedAnnounceJobAndWait({ + cron, + events, + name: "post-execution lifecycle claim conflict", + status: "error", + }); + + const updated = (await cron.list({ includeDisabled: true })).find( + (entry) => entry.id === job.id, + ); + expect(updated?.enabled).toBe(false); + expect(updated?.state.consecutiveErrors).toBe(1); + expect(updated?.state.nextRunAtMs).toBeUndefined(); + + await stopCronAndCleanup(cron, store); + }); + it("does not post fallback main summary for isolated delivery-target errors", async () => { const runIsolatedAgentJob = vi.fn(async () => ({ status: "error" as const, diff --git a/src/cron/service/timer.ts b/src/cron/service/timer.ts index b9d9cad49156..293bc11688d1 100644 --- a/src/cron/service/timer.ts +++ b/src/cron/service/timer.ts @@ -499,14 +499,16 @@ function resolveTransientCronRetryDecision(params: { cronConfig?: CronConfig; error: string | undefined; lastErrorReason?: string; + executionStarted?: boolean; consecutiveErrors: number | undefined; }): TransientCronRetryDecision { const retryConfig = resolveRetryConfig(params.cronConfig); - const retryHint = resolveCronExecutionRetryHint( - params.error, - retryConfig.retryOn, - params.lastErrorReason, - ); + const retryHint = resolveCronExecutionRetryHint({ + error: params.error, + retryOn: retryConfig.retryOn, + classifiedReason: params.lastErrorReason, + executionStarted: params.executionStarted, + }); const consecutiveErrors = params.consecutiveErrors ?? 0; if (!retryHint.retryable) { return { @@ -710,6 +712,7 @@ export function applyJobResult( result: { status: CronRunStatus; error?: string; + executionStarted?: boolean; deliveryError?: string; diagnostics?: CronRunOutcome["diagnostics"]; delivered?: boolean; @@ -869,6 +872,7 @@ export function applyJobResult( cronConfig: state.deps.cronConfig, error: result.error, lastErrorReason: job.state.lastErrorReason, + executionStarted: result.executionStarted, consecutiveErrors: job.state.consecutiveErrors, }); if (retryDecision.retryable && retryDecision.backoffMs !== undefined) { @@ -910,6 +914,7 @@ export function applyJobResult( cronConfig: state.deps.cronConfig, error: result.error, lastErrorReason: job.state.lastErrorReason, + executionStarted: result.executionStarted, consecutiveErrors: job.state.consecutiveErrors, }); let normalNext: number | undefined; @@ -1139,6 +1144,7 @@ function applyOutcomeToStoredJob( applyJobResult(state, result.job, { status: result.status, error: result.error, + executionStarted: result.executionStarted, deliveryError: result.deliveryError, diagnostics: result.diagnostics, delivered: result.delivered, @@ -1176,6 +1182,7 @@ function applyOutcomeToStoredJob( const shouldDelete = applyJobResult(state, job, { status: result.status, error: result.error, + executionStarted: result.executionStarted, deliveryError: result.deliveryError, diagnostics: result.diagnostics, delivered: result.delivered, @@ -1924,6 +1931,7 @@ async function runStartupCatchupCandidate( activeJobMarker, status: result.status, error: result.error, + executionStarted: result.executionStarted, summary: result.summary, diagnostics: result.diagnostics, delivered: result.delivered, @@ -2394,6 +2402,7 @@ async function executeDetachedCronJob( return { status: res.status, error: res.error, + executionStarted: res.executionStarted, // Forward the post-run delivery failure recorded on an otherwise // successful run so the service can persist it as `lastDeliveryError` and // emit it on the finished event for CLI/UI/API run logs (#95419). diff --git a/src/cron/types.ts b/src/cron/types.ts index 90344789154c..9069b2a29fe6 100644 --- a/src/cron/types.ts +++ b/src/cron/types.ts @@ -187,6 +187,8 @@ export type CronRunDiagnostics = { export type CronRunOutcome = { status: CronRunStatus; error?: string; + /** True once agent execution begins; retries after this point can replay side effects. */ + executionStarted?: boolean; /** Optional classifier for execution errors to guide fallback behavior. */ errorKind?: "delivery-target"; summary?: string;