diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 75ee5f55fe8..fd0e4f61ab8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4932,6 +4932,29 @@ only while its recorded retry is still due or its matching retry run is verified live. An expired or missing retry, exhausted budget, or board-owned failure shows recovery needed rather than “Observing active run.” +Finalization of a saved result has its own budget of three failed attempts, +counted in `heartbeat_runs.result_json.finalizationFailureCount`. When the +budget runs out on an accepted result of a provider-succeeded run and the cause +was contention (`status_cas_exhausted` or `side_effect_planning_failed`), the +finalizer tries the same saved result again after 2 minutes, then after 10 +minutes, each with a fresh budget, before it hands the task to the board. Each +delay is recorded once in the coordinator's `recoveryHistory` +(`automatic_finalization_backoff`). No provider work runs. + +When the coordinator still rests `terminal_failure` with +`native_finalization_retry_exhausted` for an accepted result of a succeeded +run, the task offers Retry. That Retry re-finalizes the saved result: it reopens +the same coordinator with a fresh finalization budget, records +`exhausted_finalization_retry` in its `recoveryHistory`, resolves the board +card, writes a coalesced `task_retry` receipt with strategy `finalize_result`, +and runs the finalizer for the same run. It starts no provider run and keeps +the provider attempt count. Later clicks join that Retry. Until the result +commits or the budget runs out again, the task shows work in progress +("Finishing recovery of saved work"). If that finalization fails, the normal bounded +finalization retries and recovery card apply, and Retry is offered again once +they run out. Coordinators that older builds left in this state qualify with or +without `finalizationFailureCount`; no migration is needed. + Native recovery reads include a read-only `nativeRunActivity` projection, bound to the company, source issue, and exact `resume_native_run.runId`. It reports a queued/running native heartbeat, or a running `workspace_finalize` operation diff --git a/server/src/__tests__/native-exhausted-finalization-retry.test.ts b/server/src/__tests__/native-exhausted-finalization-retry.test.ts new file mode 100644 index 00000000000..a1838746a9e --- /dev/null +++ b/server/src/__tests__/native-exhausted-finalization-retry.test.ts @@ -0,0 +1,246 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { and, eq, sql } from "drizzle-orm"; +import { + activityLog, + agents, + agentWakeupRequests, + companies, + completionContracts, + createDb, + heartbeatRuns, + issueRecoveryActions, + issues, + nativeRunFinalizations, + nativeRunResults, + workspaceOperations, +} from "@paperclipai/db"; +import { + CONTROL_PLANE_CONFORMANCE_OPEN, + CONTROL_PLANE_CONFORMANCE_RESULT, + CONTROL_PLANE_CONFORMANCE_TERMINAL, +} from "../vendor/paperclip-runner/testing.js"; +import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { reconcileNativeFinalizations } from "../services/native-runtime/native-finalization-reconciler.js"; +import { finalizeNativeRun, NATIVE_FINALIZATION_AUTOMATIC_BACKOFF_MS } from "../services/native-runtime/native-run-finalizer.js"; +import { refinalizeSavedNativeResult } from "../services/native-runtime/exhausted-finalization-retry.js"; +import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; +import { taskInterruptionPresentation } from "../services/task-interruption-presentation.js"; +import { requestTaskRetry } from "../services/task-retry.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_FINALIZATION_TEST_DATABASE_URL"); + +// A provider-succeeded native run whose accepted result could not be applied +// to its task used to rest in review with a board recovery card and no Retry. +// Retry must apply the same saved result again, never start a provider run. +describe("Retry after native finalization runs out of retries", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = "72200000-0000-4000-8000-000000000001"; + const agentId = "72200000-0000-4000-8000-000000000002"; + const userId = "board-user"; + let seq = 0; + + beforeAll(async () => { + if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-exhausted-retry-"); + db = createDb(external ?? temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Exhausted finalization", issuePrefix: "NEF" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Finalizing agent", adapterType: "codex_local", status: "running" }); + }, 30_000); + + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await temporary?.cleanup(); + }); + + async function seedAcceptedRun(options: { brokenResult?: boolean } = {}) { + seq += 1; + const id = (n: number) => `72200000-0000-4000-8${String(seq).padStart(3, "0")}-${String(n).padStart(12, "0")}`; + const issueId = id(1), contractId = id(2), runId = id(3); + await db.insert(issues).values({ id: issueId, companyId, title: `Finalize saved result ${seq}`, status: "in_progress", + assigneeAgentId: agentId, workMode: "standard" }); + await db.insert(completionContracts).values({ + id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v1", risk: "standard", + completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: "phase6-v1", objective: "Finalize the saved result", + criteria: [{ id: "objective", requirement: "The saved result is applied" }] }, + canonicalSha256: `exhausted-contract-${seq}`, createdByActorType: "system", createdByActorId: "test", + }); + await db.insert(heartbeatRuns).values({ + id: runId, companyId, agentId, status: "running", runtimeMode: "native", + runtimeModeReason: "persisted_before_kill_switch", nativeIssueId: issueId, nativeSessionId: runId, + runnerInstanceId: contractId, completionContractId: contractId, + completionContractSha256: `exhausted-contract-${seq}`, contextSnapshot: { issueId }, + }); + const port = new PaperclipControlPlanePort(db, { + companyId, issueId, runId, agentId, sessionId: runId, completionContractId: contractId, + completionContractSha256: `exhausted-contract-${seq}`, sourceInstanceId: contractId, + controlPlaneSourceInstanceId: `exhausted-control-${seq}`, + }); + await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, + identity: { companyId, issueId, runId, agentId, sessionId: runId }, sourceInstanceId: contractId }); + await port.completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, + callerResultId: `exhausted-result-${seq}` }); + if (options.brokenResult) await setResultTerminal(runId, "unknown"); + await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: runId, issueId, + phase: "workspace_finalize", status: "succeeded" }); + return { issueId, runId }; + } + + async function setResultTerminal(runId: string, runTerminalState: string) { + const stored = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, runId)).then(rows => rows[0]!); + await db.update(nativeRunResults).set({ resultJson: { ...(stored.resultJson as Record), + terminal: { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, runTerminalState } } }).where(eq(nativeRunResults.id, stored.id)); + } + + /** Three failed finalizations of a deterministic cause: no automatic delay applies. */ + async function exhaustFinalization(runId: string) { + for (let attempt = 0; attempt < 3; attempt += 1) { + await db.update(nativeRunFinalizations).set({ nextAttemptAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, runId)); + await reconcileNativeFinalizations(db, [runId]); + } + } + + const coordinator = (runId: string) => db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, runId)).then(rows => rows[0]!); + const issueRow = (issueId: string) => db.select().from(issues).where(eq(issues.id, issueId)).then(rows => rows[0]!); + const runCount = (issueId: string) => db.select({ count: sql`count(*)::int` }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.nativeIssueId, issueId))).then(rows => rows[0]!.count); + const presentation = async (issueId: string) => taskInterruptionPresentation(db, await issueRow(issueId)); + const retry = (issueId: string, runId: string, key: string) => requestTaskRetry({ db, companyId, issueId, + actorId: userId, actorSource: "session", idempotencyKey: key, observedRunId: runId }); + + async function expectStuckButRetryable(issueId: string, runId: string) { + await expect(coordinator(runId)).resolves.toMatchObject({ phase: "terminal_failure", + failureCode: "native_finalization_retry_exhausted" }); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0])) + .resolves.toMatchObject({ status: "succeeded" }); + await expect(issueRow(issueId)).resolves.toMatchObject({ status: "in_review" }); + await expect(presentation(issueId)).resolves.toMatchObject({ state: "failed", canRetry: true, observedRunId: runId }); + } + + it("re-finalizes the saved result on Retry with no new provider run or attempt", async () => { + const { issueId, runId } = await seedAcceptedRun({ brokenResult: true }); + await exhaustFinalization(runId); + await expectStuckButRetryable(issueId, runId); + const before = await coordinator(runId); + // The cause clears (as row contention does); the saved result is intact. + await setResultTerminal(runId, CONTROL_PLANE_CONFORMANCE_TERMINAL.runTerminalState); + + const receipt = await retry(issueId, runId, "refinalize-1"); + expect(receipt).toMatchObject({ accepted: true, stale: false, observedRunId: runId, finalizeRunId: runId }); + await expect(retry(issueId, runId, "refinalize-1")).resolves.toMatchObject({ accepted: true, requestId: receipt.requestId }); + await expect(coordinator(runId)).resolves.toMatchObject({ phase: "workspace_finalizing", failureCode: null, + recoveryHistory: [expect.objectContaining({ kind: "exhausted_finalization_retry", generationId: receipt.requestId, + previousFailureCode: "native_finalization_retry_exhausted" })] }); + await expect(db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!))) + .resolves.toEqual([expect.objectContaining({ status: "coalesced", runId, + payload: expect.objectContaining({ taskRetry: expect.objectContaining({ strategy: "finalize_result" }) }) })]); + await expect(db.select().from(activityLog).where(and(eq(activityLog.entityId, issueId), + eq(activityLog.action, "issue.retry_requested")))).resolves.toEqual([expect.objectContaining({ + actorType: "user", actorId: userId, runId, details: expect.objectContaining({ strategy: "finalize_result" }) })]); + + // A second click (another tab, another user, a new key) joins the pending + // saved-result Retry instead of queueing a provider Retry. + await expect(retry(issueId, runId, "refinalize-2")).resolves.toMatchObject({ accepted: true, + requestId: receipt.requestId, finalizeRunId: runId }); + const providerRetries = () => db.select().from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.reason, "task_retry"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`, + sql`${agentWakeupRequests.status} <> 'coalesced' or ${agentWakeupRequests.payload}#>>'{taskRetry,strategy}' <> 'finalize_result'`)); + await expect(providerRetries()).resolves.toHaveLength(0); + // A finalizer holds the coordinator (and would take the task next). A + // click must join without waiting on it, or the two could deadlock. + await db.transaction(async finalizer => { + await finalizer.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)).for("update"); + const joined = await Promise.race([retry(issueId, runId, "refinalize-locked"), + new Promise(resolve => setTimeout(() => resolve("blocked on the coordinator lock"), 5_000))]); + expect(joined).toMatchObject({ accepted: true, requestId: receipt.requestId }); + }); + // The Retry visibly took effect while the saved result is applied. + await expect(presentation(issueId)).resolves.toMatchObject({ state: "working", canRetry: false, + recoveryStep: "finishing_recovery" }); + + await refinalizeSavedNativeResult(db, { companyId, runId }); + // A stale click after the result applied also joins; nothing new runs. + const late = await retry(issueId, runId, "refinalize-3"); + expect(late).toMatchObject({ accepted: true, requestId: receipt.requestId }); + expect(late).not.toHaveProperty("finalizeRunId"); + await expect(providerRetries()).resolves.toHaveLength(0); + const after = await coordinator(runId); + expect(after).toMatchObject({ phase: "committed" }); + expect(after.decisionId).toBeTruthy(); + expect(after.attempt).toBe(before.attempt); + expect(await runCount(issueId)).toBe(1); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0])) + .resolves.toMatchObject({ status: "succeeded", nativePhase: "committed" }); + const settled = await presentation(issueId); + expect(settled?.canRetry ?? false).toBe(false); + expect(settled?.state ?? "resolved").toBe("resolved"); + await expect(db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, issueId), + eq(issueRecoveryActions.status, "active"), eq(issueRecoveryActions.ownerType, "board")))).resolves.toHaveLength(0); + }); + + it("offers the same Retry for a coordinator an older build left terminal", async () => { + const { issueId, runId } = await seedAcceptedRun({ brokenResult: true }); + await exhaustFinalization(runId); + // The previous release's shape: no finalization count, budget judged on provider attempts. + await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'finalizationFailureCount'` }) + .where(eq(heartbeatRuns.id, runId)); + await db.update(nativeRunFinalizations).set({ attempt: 3 }).where(eq(nativeRunFinalizations.runId, runId)); + await expectStuckButRetryable(issueId, runId); + await setResultTerminal(runId, CONTROL_PLANE_CONFORMANCE_TERMINAL.runTerminalState); + + await expect(retry(issueId, runId, "refinalize-old")).resolves.toMatchObject({ finalizeRunId: runId }); + await refinalizeSavedNativeResult(db, { companyId, runId }); + await expect(coordinator(runId)).resolves.toMatchObject({ phase: "committed", attempt: 3 }); + expect(await runCount(issueId)).toBe(1); + }); + + it("keeps a Retry whose re-finalization fails again bounded and visible", async () => { + const { issueId, runId } = await seedAcceptedRun({ brokenResult: true }); + await exhaustFinalization(runId); + await expectStuckButRetryable(issueId, runId); + + await expect(retry(issueId, runId, "refinalize-fails")).resolves.toMatchObject({ finalizeRunId: runId }); + await refinalizeSavedNativeResult(db, { companyId, runId }); + // The click granted a fresh budget: the first failure stays automatic. + await expect(coordinator(runId)).resolves.toMatchObject({ phase: "retryable_failure", + failureCode: "native_finalization_invalid" }); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]?.resultJson)) + .resolves.toMatchObject({ finalizationFailureCount: 1 }); + await expect(presentation(issueId)).resolves.toMatchObject({ state: "working", canRetry: false }); + await exhaustFinalization(runId); + await expectStuckButRetryable(issueId, runId); + expect(await runCount(issueId)).toBe(1); + }); + + it("re-applies a result that lost status races after longer delays before asking for Retry", async () => { + const { issueId, runId } = await seedAcceptedRun(); + const failOnce = async () => { + // The quick retries are already spent; the next failure exhausts them. + await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) + || '{"finalizationFailureCount": 2}'::jsonb` }).where(eq(heartbeatRuns.id, runId)); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true, + failpoint: "status_projection" }); + }; + for (const delay of NATIVE_FINALIZATION_AUTOMATIC_BACKOFF_MS) { + const started = Date.now(); + await failOnce(); + const row = await coordinator(runId); + expect(row).toMatchObject({ phase: "retryable_failure", failureCode: "side_effect_planning_failed" }); + expect(row.nextAttemptAt!.getTime()).toBeGreaterThanOrEqual(started + delay - 1_000); + await expect(issueRow(issueId)).resolves.toMatchObject({ status: "in_progress" }); + await expect(db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).then(rows => rows[0]?.resultJson)) + .resolves.toMatchObject({ finalizationFailureCount: 0 }); + } + // Each delay is used once; then the board gets a working Retry. + await failOnce(); + await expectStuckButRetryable(issueId, runId); + await expect(retry(issueId, runId, "refinalize-after-backoff")).resolves.toMatchObject({ finalizeRunId: runId }); + await refinalizeSavedNativeResult(db, { companyId, runId }); + await expect(coordinator(runId)).resolves.toMatchObject({ phase: "committed" }); + expect(await runCount(issueId)).toBe(1); + }); +}); diff --git a/server/src/routes/issues.ts b/server/src/routes/issues.ts index 389157ed47b..c68ce450ee8 100644 --- a/server/src/routes/issues.ts +++ b/server/src/routes/issues.ts @@ -10242,7 +10242,14 @@ export function issueRoutes( // The receipt commits before dispatch. Maintenance redelivers it after a // crash or a lost response; this nudge only reduces perceived latency. const chatActionId = "chatActionId" in receipt ? receipt.chatActionId : undefined; - if (chatActionId) void opts.chatRunRetries?.processFailedChatRunRetry(chatActionId).catch(err => + const finalizeRunId = "finalizeRunId" in receipt ? receipt.finalizeRunId : undefined; + if (finalizeRunId) { + // The saved result is applied again; no provider run starts. The + // result-commit worker repeats this if the process stops first. + const { refinalizeSavedNativeResult } = await import("../services/native-runtime/exhausted-finalization-retry.js"); + void refinalizeSavedNativeResult(db, { companyId: issue.companyId, runId: finalizeRunId }).catch(err => + logger.warn({ err, issueId: issue.id, runId: finalizeRunId }, "Saved result finalization remains queued")); + } else if (chatActionId) void opts.chatRunRetries?.processFailedChatRunRetry(chatActionId).catch(err => logger.warn({ err, issueId: issue.id }, "Connected task Retry remains queued")); else if (receipt.accepted && receipt.requestId) void heartbeat.resumeTaskRetry(issue.companyId, receipt.requestId).catch(err => logger.warn({ err, issueId: issue.id }, "Task Retry remains queued")); diff --git a/server/src/services/native-runtime/exhausted-finalization-retry.ts b/server/src/services/native-runtime/exhausted-finalization-retry.ts new file mode 100644 index 00000000000..c1430d0c75f --- /dev/null +++ b/server/src/services/native-runtime/exhausted-finalization-retry.ts @@ -0,0 +1,155 @@ +import { and, eq, sql, type SQL } from "drizzle-orm"; +import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, workspaceOperations, type Db } from "@paperclipai/db"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { nativeSha256 } from "./canonical.js"; + +/** + * A provider-succeeded native run whose accepted result could not be applied + * to its task: finalization used its whole retry budget and the coordinator + * rests `terminal_failure` with `native_finalization_retry_exhausted`. Retry + * re-applies that saved result. It never starts a provider run. + * + * Rows saved by older builds carry the same phase and code, with or without + * `finalizationFailureCount`, so they qualify without a migration. + */ +export const EXHAUSTED_FINALIZATION_FAILURE_CODE = "native_finalization_retry_exhausted"; +export const EXHAUSTED_FINALIZATION_RETRY_KIND = "exhausted_finalization_retry"; +/** If the immediate re-finalization is lost (for example, a restart), the result-commit worker runs it then. */ +const REFINALIZATION_FALLBACK_DELAY_MS = 30_000; + +/** True for a run row whose saved result only needs finalizing again. Correlated on `heartbeat_runs`. */ +export function exhaustedNativeFinalizationCondition(): SQL { + return sql`(${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'succeeded' + and exists (select 1 from ${nativeRunFinalizations} + inner join ${nativeRunResults} on ${nativeRunResults.id} = ${nativeRunFinalizations.resultId} + and ${nativeRunResults.companyId} = ${nativeRunFinalizations.companyId} + and ${nativeRunResults.runId} = ${nativeRunFinalizations.runId} + and ${nativeRunResults.schemaStatus} = 'accepted' + where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.issueId} = ${heartbeatRuns.nativeIssueId} + and ${nativeRunFinalizations.phase} = 'terminal_failure' + and ${nativeRunFinalizations.failureCode} = ${EXHAUSTED_FINALIZATION_FAILURE_CODE} + and ${nativeRunFinalizations.leaseOwner} is null + and ${nativeRunFinalizations.failureDetail}->'successorRunId' is null))`; +} + +/** + * True while a saved-result Retry of this succeeded run is still being applied + * (including its bounded automatic retries). The task shows work in progress. + * The finalizer ends it by committing or by going terminal again. + */ +export function savedResultRetryPendingCondition(): SQL { + return sql`(${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} = 'succeeded' + and exists (select 1 from ${nativeRunFinalizations} + where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} + and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} + and ${nativeRunFinalizations.resultId} is not null + and ${nativeRunFinalizations.phase} not in ('committed', 'terminal_failure') + and ${nativeRunFinalizations.recoveryHistory} @> ${JSON.stringify([{ kind: EXHAUSTED_FINALIZATION_RETRY_KIND }])}::jsonb))`; +} + +/** + * Caller holds the task lock and has proven `runId` is the task's latest run + * by its current assignee. Reopens the coordinator for the same saved result + * with a fresh finalization budget. + * + * - `rearmed`: this call reopened the coordinator. + * - `joined`: an earlier saved-result Retry of this run already reopened it + * (still finalizing, or applied). A later click joins it and never starts + * provider work; `generationId` names that earlier Retry. + * - `not_eligible`: the caller keeps its ordinary Retry. + */ +export async function rearmExhaustedNativeFinalization(tx: Db, input: { + companyId: string; issueId: string; runId: string; generationId: string; actorId: string; +}): Promise<{ kind: "rearmed" } | { kind: "joined"; generationId: string | null; finalizing: boolean } + | { kind: "not_eligible" }> { + const [eligible] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.nativeIssueId, input.issueId), exhaustedNativeFinalizationCondition(), + )).limit(1); + // The caller holds the task row. The finalizer locks coordinator then task, + // so never wait for a coordinator that may be finalizing: read it unlocked, + // and lock it only when it rests terminal, where no finalizer holds it. + const readCoordinator = (lock: boolean) => { + const query = tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.issueId, input.issueId))); + return (lock ? query.for("update") : query).then(rows => rows[0]); + }; + const classify = (row: typeof nativeRunFinalizations.$inferSelect) => { + // Joining is read-only. If the earlier Retry goes terminal meanwhile, this + // click still starts no provider work and the task offers Retry again. + const earlier = row.phase === "terminal_failure" ? undefined + : row.recoveryHistory.findLast(entry => entry.kind === EXHAUSTED_FINALIZATION_RETRY_KIND); + return earlier ? { kind: "joined" as const, finalizing: row.phase !== "committed", + generationId: typeof earlier.generationId === "string" ? earlier.generationId : null } : { kind: "not_eligible" as const }; + }; + const exhausted = (row: typeof nativeRunFinalizations.$inferSelect) => row.phase === "terminal_failure" + && row.failureCode === EXHAUSTED_FINALIZATION_FAILURE_CODE && !row.leaseOwner; + const observed = await readCoordinator(false); + if (!observed?.resultId) return { kind: "not_eligible" }; + if (!eligible || !exhausted(observed)) return classify(observed); + const coordinator = await readCoordinator(true); + if (!coordinator?.resultId) return { kind: "not_eligible" }; + if (!exhausted(coordinator)) { + const changed = classify(coordinator); + // Changed under us: fail closed rather than fall back to a provider Retry. + return changed.kind === "joined" ? changed : { kind: "joined", finalizing: true, generationId: null }; + } + const now = new Date(); + const nextAttemptAt = new Date(now.getTime() + REFINALIZATION_FALLBACK_DELAY_MS); + const phase = coordinator.assessmentId ? "arbitrating" : "workspace_finalizing"; + await notifyNativeFinalizationWork(tx, { companyId: input.companyId, runId: input.runId, issueId: input.issueId }); + await tx.update(nativeRunFinalizations).set({ + phase, nextAttemptAt, failureCode: null, failureDetail: null, updatedAt: now, + recoveryHistory: [...coordinator.recoveryHistory, { + kind: EXHAUSTED_FINALIZATION_RETRY_KIND, generationId: input.generationId, requestedByUserId: input.actorId, + resultId: coordinator.resultId, previousFailureCode: coordinator.failureCode, + previousFailureDetail: coordinator.failureDetail, at: now.toISOString(), + }], + }).where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.phase, "terminal_failure"))); + // The board card asked for this Retry. A new failure raises its own card. + // Loaded on use: task presentation reads this module's condition. + const { issueRecoveryActionService } = await import("../issue-recovery-actions.js"); + await issueRecoveryActionService(tx).resolveActiveForIssue({ + companyId: input.companyId, sourceIssueId: input.issueId, kind: "active_run_watchdog", + cause: EXHAUSTED_FINALIZATION_FAILURE_CODE, + fingerprint: nativeSha256({ runId: input.runId, failureCode: EXHAUSTED_FINALIZATION_FAILURE_CODE }), + status: "resolved", outcome: "handed_back", + resolutionNote: "Retry applies the saved result again, without another provider turn.", + }, tx); + // A deliberate Retry is a new durable intent, so it gets a fresh budget. + await tx.update(heartbeatRuns).set({ + nativePhase: phase, nativePhaseUpdatedAt: now, updatedAt: now, + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + finalizationPhase: phase, finalizationFailureCount: 0, failureCode: null, nextAttemptAt: nextAttemptAt.toISOString(), + })}::jsonb`, + }).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))); + return { kind: "rearmed" }; +} + +/** + * Applies the saved result now, after `rearmExhaustedNativeFinalization` + * committed. Same-run finalization only: it keeps the provider attempt count + * and records a failure through the finalizer's own bounded budget. + */ +export async function refinalizeSavedNativeResult(db: Db, input: { companyId: string; runId: string }) { + const [barrier] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.phase, "workspace_finalize"), eq(workspaceOperations.status, "succeeded"), + )).limit(1); + // Without a saved workspace the reconciler resumes the workspace step first. + if (!barrier) return null; + // Loaded on use: task presentation reads this module's condition. + const { finalizeNativeRun, NativeFinalizationDeferredError, recordNativeFinalizationFailure } = + await import("./native-run-finalizer.js"); + try { + return await finalizeNativeRun({ db, runId: input.runId, workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, preserveProviderAttempt: true }); + } catch (error) { + if (error instanceof NativeFinalizationDeferredError) return null; + return recordNativeFinalizationFailure({ db, runId: input.runId, error, projectRunStatus: true }); + } +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 3b62586d7e7..a29e848e59c 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -327,6 +327,15 @@ function terminalFinalizationFailureMessage(failureCode: string, failureScope: " } const NATIVE_FINALIZATION_RETRY_LIMIT = 3; +/** + * Contention while applying a saved, accepted, provider-succeeded result is + * transient. When its quick retries run out, the finalizer waits these longer + * delays and tries the same saved result again, with no provider work, before + * it hands the task to the board with a Retry. Bounded: one entry per delay. + */ +export const NATIVE_FINALIZATION_AUTOMATIC_BACKOFF_MS = [2 * 60_000, 10 * 60_000] as const; +const TRANSIENT_FINALIZATION_FAILURE_CODES = new Set(["status_cas_exhausted", "side_effect_planning_failed"]); +export const AUTOMATIC_FINALIZATION_BACKOFF_KIND = "automatic_finalization_backoff"; /** Failed finalizations of this run's saved result. Rows saved before the count existed start at zero. */ function readFinalizationFailureCount(resultJson: unknown) { @@ -462,14 +471,31 @@ async function recordRetryableFailure(input: { // turn its first transient finalization failure into a terminal one and // strand an accepted result. The count lives on the run's result because // every claim clears the coordinator's failure detail. - const finalizationFailureCount = input.failureScope === "workspace" + let finalizationFailureCount = input.failureScope === "workspace" ? null : readFinalizationFailureCount(input.run.resultJson) + 1; - const exhausted = + const budgetExhausted = input.permanent === true || (workspaceFinalizeAttempt !== null ? workspaceFinalizeAttempt >= 3 : finalizationFailureCount! >= NATIVE_FINALIZATION_RETRY_LIMIT); + // A saved, accepted, provider-succeeded result that only lost status races + // is re-applied after a longer delay instead of stranding the task. Each + // delay is used once; the result is never recreated and no provider runs. + const automaticBackoffsUsed = input.coordinator.recoveryHistory + .filter((entry) => entry.kind === AUTOMATIC_FINALIZATION_BACKOFF_KIND).length; + const automaticBackoffMs = budgetExhausted && !supersededByNewerRun && input.permanent !== true + && input.failureScope !== "workspace" && Boolean(input.coordinator.resultId) + && acceptedRunTerminalState === "succeeded" && input.run.status !== "cancelled" + && !preservedResultHasStop(input.run.resultJson) + && TRANSIENT_FINALIZATION_FAILURE_CODES.has(input.failureCode) + ? NATIVE_FINALIZATION_AUTOMATIC_BACKOFF_MS[automaticBackoffsUsed] ?? null + : null; + const exhausted = budgetExhausted && automaticBackoffMs === null; + if (automaticBackoffMs !== null) { + nextAttemptAt.setTime(now.getTime() + automaticBackoffMs); + finalizationFailureCount = 0; + } const phase = supersededByNewerRun || exhausted ? ("terminal_failure" as const) @@ -513,6 +539,13 @@ async function recordRetryableFailure(input: { nextAction: input.nextAction, }, nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt, + ...(automaticBackoffMs === null ? {} : { + recoveryHistory: [...input.coordinator.recoveryHistory, { + kind: AUTOMATIC_FINALIZATION_BACKOFF_KIND, resultId: input.coordinator.resultId, + previousFailureCode: input.failureCode, delayMs: automaticBackoffMs, + nextAttemptAt: nextAttemptAt.toISOString(), at: now.toISOString(), + }], + }), updatedAt: now, }) .where(eq(nativeRunFinalizations.runId, input.run.id)); diff --git a/server/src/services/task-interruption-presentation.test.ts b/server/src/services/task-interruption-presentation.test.ts index 93154e7e4d3..d6f87fca344 100644 --- a/server/src/services/task-interruption-presentation.test.ts +++ b/server/src/services/task-interruption-presentation.test.ts @@ -455,6 +455,30 @@ describe("setup command failure evidence", () => { }); }); +describe("saved result that ran out of finalization retries", () => { + const review = { ...issue, status: "in_review" }; + const succeeded = { ...run, status: "succeeded", runtimeMode: "native" as const, errorCode: null, resultJson: null }; + + it("offers Retry for a succeeded run whose accepted result could not be finalized", () => { + expect(projectTaskInterruption(review, { run: { ...succeeded, nativeFinalizationRetry: true }, agent, + paused: false, approval: false })).toMatchObject({ state: "failed", canRetry: true, observedRunId: "run" }); + }); + + it("shows the saved result being applied after Retry", () => { + expect(projectTaskInterruption(review, { run: { ...succeeded, nativeFinalizationRetryPending: true }, agent, + paused: false, approval: false })).toMatchObject({ state: "working", canRetry: false, recoveryStep: "finishing_recovery" }); + }); + + it("keeps ordinary succeeded and failed runs unchanged", () => { + expect(projectTaskInterruption(review, { run: succeeded, agent, paused: false, approval: false })) + .toMatchObject({ state: "resolved", canRetry: false }); + expect(projectTaskInterruption(review, { run: { ...succeeded, status: "failed", errorCode: "adapter_failed" }, agent, + paused: false, approval: false })).toMatchObject({ state: "failed", canRetry: true }); + expect(projectTaskInterruption({ ...review, status: "done" }, { run: { ...succeeded, nativeFinalizationRetry: true }, + agent, paused: false, approval: false })).toMatchObject({ state: "resolved", canRetry: false }); + }); +}); + // Remote native run shape: the provider turn completed and its work was // saved, then a same-run recovery was refused by the cleanup gate. The task // showed "Couldn't finish this attempt" with a Retry the server refuses (409). diff --git a/server/src/services/task-interruption-presentation.ts b/server/src/services/task-interruption-presentation.ts index 2163074b6b1..3118c05667f 100644 --- a/server/src/services/task-interruption-presentation.ts +++ b/server/src/services/task-interruption-presentation.ts @@ -21,6 +21,7 @@ import { canonicalExecutionUuid } from "./execution-identity.js"; import { chatActionDeadline } from "./chat-action-work.js"; import { NATIVE_CLEANUP_SETTLING_MESSAGE, NATIVE_SESSION_CLEANUP_QUARANTINED } from "./native-runtime/completed-turn-cleanup.js"; import { isQueuedDependencyWait } from "./run-cancellation.js"; +import { exhaustedNativeFinalizationCondition, savedResultRetryPendingCondition } from "./native-runtime/exhausted-finalization-retry.js"; import { AI_CONNECTION_BUSY_RETRY_REASON, AI_CONNECTION_POOL_WAIT_RETRY_REASON, RUNNER_REVISION_WAIT_RETRY_REASON, WORKSPACE_BUSY_RETRY_REASON } from "../modules/run-dispatch/index.js"; @@ -45,6 +46,10 @@ type ProjectionRun = Pick`(select ${nativeRunFinalizations.updatedAt} from ${nativeRunFinalizations} where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id} and ${nativeRunFinalizations.phase} = 'terminal_failure')`.mapWith(value => value == null ? null : new Date(value)), + nativeFinalizationRetry: exhaustedNativeFinalizationCondition(), + nativeFinalizationRetryPending: savedResultRetryPendingCondition(), // The run whose resultJson supplied providerFailure. A scheduled retry // shows its predecessor's reason, so redaction must use that run's registry. providerFailureRunId: sql`case when ${heartbeatRuns.resultJson}->'providerFailure' is null diff --git a/server/src/services/task-retry.ts b/server/src/services/task-retry.ts index 1f75c4d8292..72050465f52 100644 --- a/server/src/services/task-retry.ts +++ b/server/src/services/task-retry.ts @@ -22,6 +22,7 @@ import { deferredChatInputPredicate, resumeDeferredChatWake } from "./durable-ch import { notifyDeferredExecutionWork } from "./deferred-execution-notifications.js"; import { hasPendingRunCleanup } from "./execution-blocker.js"; import { PersistedExecutionWaitRefusal } from "./deferred-execution-wait.js"; +import { rearmExhaustedNativeFinalization } from "./native-runtime/exhausted-finalization-retry.js"; import { NativeChatReviewPresentationContentionError } from "./native-runtime/native-chat-review-presentation.js"; import { isExternalChatWaitAuthorizationContention } from "./native-runtime/chat-attachment-reuse.js"; @@ -300,6 +301,40 @@ function admitTaskRetry(input: TaskRetryRequest) { .where(eq(agentWakeupRequests.id, active.id)); active = undefined; } + // An accepted result that only failed to finalize is applied again, never + // replaced by a new provider run. The receipt coalesces into that run. + const finalizeId = randomUUID(); + const savedResult = latest.runtimeMode === "native" && latest.status === "succeeded" + ? await rearmExhaustedNativeFinalization(db, { companyId: issue.companyId, issueId: issue.id, runId: latest.id, + generationId: active?.id ?? finalizeId, actorId: input.actorId }) + : { kind: "not_eligible" as const }; + if (savedResult.kind !== "not_eligible") { + // A later click joins the saved-result Retry already under way (or applied). + const canonicalId = (savedResult.kind === "joined" ? savedResult.generationId : null) ?? active?.id ?? null; + const finalizeIntent: TaskRetryIntent = { schema: "paperclip.task-retry.v1", generationId: canonicalId ?? finalizeId, + sourceRunId: latest.id, actorId: input.actorId, + actorSource: taskRetryIntentSchema.shape.actorSource.parse(input.actorSource ?? "session"), strategy: "finalize_result", + ...(canonicalId ? { canonicalIntentId: canonicalId } : {}) }; + const settled = { status: "coalesced", runId: latest.id, finishedAt: new Date(), updatedAt: new Date() } as const; + // An unadmitted Retry for this run would start a provider turn; it joins the re-finalization instead. + if (active && !active.runId) await tx.update(agentWakeupRequests).set({ ...settled, + payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}', '"finalize_result"'::jsonb)` }) + .where(eq(agentWakeupRequests.id, active.id)); + await tx.insert(agentWakeupRequests).values({ id: finalizeId, companyId: issue.companyId, agentId: issue.assigneeAgentId, + source: "on_demand", triggerDetail: "manual", reason: "task_retry", idempotencyKey: key, + requestedByActorType: "user", requestedByActorId: input.actorId, ...settled, + payload: { issueId: issue.id, taskRetry: finalizeIntent } }); + const restartRecovery = input.initiatedBy === "restart_recovery"; + await persistActivity(db, { companyId: issue.companyId, actorType: restartRecovery ? "system" : "user", + actorId: restartRecovery ? "restart-recovery" : input.actorId, + action: "issue.retry_requested", entityType: "issue", entityId: issue.id, runId: latest.id, + details: { requestId: canonicalId ?? finalizeId, sourceRunId: latest.id, attachedToExistingWork: Boolean(canonicalId), + strategy: "finalize_result", + ...(restartRecovery ? { initiatedBy: "restart_recovery", onBehalfOfUserId: input.actorId } : {}) } }); + const finalizing = savedResult.kind === "rearmed" || savedResult.finalizing; + return { requestId: canonicalId ?? finalizeId, accepted: true, stale: false, observedRunId: latest.id, + chatActionId: undefined, ...(finalizing ? { finalizeRunId: latest.id } : {}) }; + } const busy = ["queued", "running", "scheduled_retry"].includes(latest.status); if (!busy) { await preserveTaskRemoteWorkspaceWriter(db, issue.companyId, issue.id);