diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 492b8f0cf76..bd85a0b0f72 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4695,7 +4695,18 @@ generation ends at once as `user_action_required`) and goes to the board. It is not retried on the automatic schedule, and it never falls back to the last save, so the run's unsaved files stay in the retained sandbox and the workspace keeps its previous saved version. The -board's Retry saves again; once the save succeeds, the accepted Retry starts one run. +board's Retry saves again. Once the save succeeds, the accepted Retry applies the +run's saved accepted result and finishes the run and task from it, with no new +agent turn and no provider spend. Retry admission waits (`finalizing_saved_result`) +while any saved-work hold of that run is still saving, then preserved-result +finalization coalesces the Retry into the source run (`finalize_result`, activity +`issue.saved_result_finalization_resumed`). This applies to any accepted Retry +(`evaluate`) of a terminal run whose coordinator still holds an accepted, +succeeded result, whose saved-work recovery completed with a verified receipt, +and which has no recorded Stop, newer run, current execution or other owner. A +record that lacks any of these, including one written by an older server, keeps +the previous behaviour: the Retry starts one run. If the save still finds the +disk full, the hold rests again as `sandbox_disk_full` with its Retry. A reclaim whose outcome is unknown keeps the scheduled retry, because its journal resumes on the next attempt and can still free space. diff --git a/server/src/services/__tests__/workspace-recovery-fixture.ts b/server/src/services/__tests__/workspace-recovery-fixture.ts index da034421bfb..55f6dae106a 100644 --- a/server/src/services/__tests__/workspace-recovery-fixture.ts +++ b/server/src/services/__tests__/workspace-recovery-fixture.ts @@ -15,6 +15,7 @@ import { environmentService } from "../environments.js"; import { classifyEnvironmentCapabilities, environmentRuntimeService } from "../environment-runtime.js"; import { readSystemRecovery, createSystemRecovery, failSystemRecovery, reserveSystemRecoveryAttempt, SystemRecoveryFailure } from "../system-recovery-state.js"; import { adoptSystemRecovery } from "../system-recovery.js"; +import { recoverRetainedWorkspace } from "../workspace-recovery.js"; import { requestTaskRetry } from "../task-retry.js"; import { execFile } from "node:child_process"; import { randomUUID } from "node:crypto"; @@ -457,3 +458,12 @@ export async function makeRetryDue(actionId: string) { await db.update(issueRecoveryActions).set({ evidence: { ...current.evidence, systemRecovery: { ...state, nextAttemptAt: 0 } } }).where(eq(issueRecoveryActions.id, actionId)); } + +/** The recovery worker's steps for every saved-work hold of one run (Retry can + * add one beside the board hold), without its post-recovery dispatch. */ +export async function recoverRunHolds(f: Parameters[0]) { + const holds = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, f.companyId), sql`${issueRecoveryActions.evidence}->>'runId' = ${f.runId}`, + sql`${issueRecoveryActions.evidence} ? 'systemRecovery'`)).orderBy(issueRecoveryActions.createdAt); + for (const hold of holds) await recoverRetainedWorkspace({ ...f, actionId: hold.id }); +} diff --git a/server/src/services/explicit-native-continuation.test.ts b/server/src/services/explicit-native-continuation.test.ts index 5c61f923aa8..3e4209de522 100644 --- a/server/src/services/explicit-native-continuation.test.ts +++ b/server/src/services/explicit-native-continuation.test.ts @@ -26,6 +26,7 @@ import { import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js"; import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; +import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; import { adapterExecutionControls, createAdapterExecutionControl } from "./adapter-execution-control.js"; import { CONVERSATION_CONTINUATION_POLICY } from "./conversation-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; @@ -2806,7 +2807,28 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes .toMatchObject({ status: "deferred_issue_execution", runId: null }); }); - it.each(["message", "queued_message", "task_retry"] as const)("continues a terminal retained result after verified copy-back: %s", async origin => { + // A Retry asks for the accepted result, which now has its saved files. It is + // applied by preserved-result finalization, never replaced by a new turn. + it("holds a Retry of a terminal retained result after verified copy-back until its accepted result is applied", async () => { + const f = await recoveredCopybackFixture("task_retry"); + const [before] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); + const waits: string[] = []; + expect(await db.transaction(async tx => { + await tx.select().from(issues).where(eq(issues.id, f.issueId)).for("update"); + return admitExplicitNativeContinuation({ ...f, db: tx as unknown as typeof db, onBlocked: reason => { waits.push(reason); } }); + })).toBeNull(); + expect(waits).toEqual(["finalizing_saved_result"]); + expect(await resumePreservedResultFinalization(db, f.companyId, f.sourceRunId)).toBe(true); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.requestId)))[0]) + .toMatchObject({ status: "coalesced", runId: f.sourceRunId, payload: { taskRetry: { strategy: "finalize_result" } } }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId)))[0]) + .toMatchObject({ phase: "workspace_finalizing", resultId: f.resultId }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0]).toEqual(before); + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + sql`${heartbeatRuns.id} <> ${f.sourceRunId}`))).toEqual([]); + }); + + it.each(["message", "queued_message"] as const)("continues a terminal retained result after verified copy-back: %s", async origin => { const f = await recoveredCopybackFixture(origin); const [before] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)); const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)); diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index c198f1bbfb0..0675377b3ce 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -35,6 +35,7 @@ import { activeIssueInteractionCondition } from "./issue-question-context.js"; import { historicalAdapterType, isConversationAdapter } from "./conversation-continuation.js"; import { queuedCommentIdsFromWakePayload } from "./issue-queued-comment-queue.js"; import { firstTurnDeliveredCommentIds } from "./queued-comment-delivery.js"; +import { canResumePreservedResultFinalization } from "./preserved-result-finalization.js"; type Run = Pick>'runId' = ${runId}`, + )).orderBy(desc(issueRecoveryActions.createdAt)).limit(1); + const recovery = readSystemRecovery(action?.evidence); + const workspace = readWorkspaceRecovery(action?.evidence ?? null); + if (recovery?.state !== "complete" || recovery.sourceRunId !== run.id || !recovery.preservationOperationIds.length + || workspace?.status !== "recovered" || workspace.runId !== run.id + || !await verifiedWorkspaceRepairReceipt(db, run)) return null; + const [intent] = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, run.agentId), + eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.reason, "task_retry"), + sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`, + sql`${agentWakeupRequests.payload}#>>'{taskRetry,sourceRunId}' = ${runId}`, + )).orderBy(desc(agentWakeupRequests.requestedAt)).limit(1); + const generationId = intent?.id ?? recovery.generationId; + if (coordinator.recoveryHistory.some(entry => entry.kind === "preserved_result_finalization" && + entry.generationId === generationId)) return null; + return { coordinator, issue, run, result, recovery, intent, generationId }; +} + +/** The facts that do not depend on the saved-work proof: an accepted, succeeded + * result held by a terminal coordinator, no Stop, no newer run, and the same + * agent owner with no current execution. */ +async function readApplicableResult(db: Db, companyId: string, runId: string, lock: Lock) { + // The finalizer and owner-stop confirmation lock the coordinator before the + // issue; Retry dispatch locks the issue first. This takes the coordinator + // first (like the finalizer). With "nowait" it does not wait for the issue or + // run while it holds the coordinator: a busy row fails the attempt + // (lock_not_available) and the caller tries again. + const coordinatorQuery = db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId))); + const [coordinator] = lock ? await coordinatorQuery.for("update") : await coordinatorQuery; + if (!coordinator?.resultId || coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || + coordinator.failureDetail?.successorRunId) return null; + const issueQuery = db.select().from(issues).where(and(eq(issues.companyId, companyId), eq(issues.id, coordinator.issueId))); + const [issue] = lock === "nowait" ? await issueQuery.for("update", { noWait: true }) + : lock ? await issueQuery.for("update") : await issueQuery; + const runQuery = db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.id, runId))); + const [run] = lock === "nowait" ? await runQuery.for("update", { noWait: true }) + : lock ? await runQuery.for("update") : await runQuery; + if (!issue || !run || run.status === "cancelled" || preservedResultHasStop(run.resultJson) + || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || ["done", "cancelled"].includes(issue.status) || + issue.executionRunId || issue.checkoutRunId) return null; + const [newer] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, companyId), ne(heartbeatRuns.id, runId), + or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), + gte(heartbeatRuns.createdAt, run.createdAt), + )).limit(1); + if (newer) return null; + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.issueId, issue.id), + eq(nativeRunResults.runId, runId), eq(nativeRunResults.id, coordinator.resultId), + eq(nativeRunResults.schemaStatus, "accepted"), + sql`${nativeRunResults.resultJson}#>>'{terminal,runTerminalState}' = 'succeeded'`, + )); + return result ? { coordinator, issue, run, result } : null; +} + +/** Read-only: whether resumePreservedResultFinalization would apply this run's + * saved accepted result now (`ready`), or could once the run's saved-work + * recovery completes (`pending`). Retry admission uses it to wait for that + * result instead of starting a new agent turn. It takes no locks. */ +export async function canResumePreservedResultFinalization(db: Db, companyId: string, runId: string, + when: "ready" | "pending" = "ready") { + return (when === "ready" ? await readPreservedResult(db, companyId, runId, false) + : await readApplicableResult(db, companyId, runId, false)) !== null; +} /** Re-enter the existing finalizer only after preservation. This does not * recreate a result, grant an executor budget, or apply it over newer work. */ export async function resumePreservedResultFinalization(db: Db, companyId: string, runId: string) { - return db.transaction(async tx => { - const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and( - eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, runId), - )).for("update"); - if (!coordinator?.resultId || coordinator.phase !== "terminal_failure" || coordinator.leaseOwner || - coordinator.failureDetail?.successorRunId) return false; - const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, companyId), - eq(issues.id, coordinator.issueId))).for("update"); - const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), - eq(heartbeatRuns.id, runId))).for("update"); - if (!issue || !run || run.status === "cancelled" || preservedResultHasStop(run.resultJson) - || issue.assigneeAgentId !== run.agentId || issue.assigneeUserId || ["done", "cancelled"].includes(issue.status) || - issue.executionRunId || issue.checkoutRunId) return false; - const [newer] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, companyId), ne(heartbeatRuns.id, runId), - or(eq(heartbeatRuns.issueId, issue.id), eq(heartbeatRuns.nativeIssueId, issue.id), - sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issue.id}`), - gte(heartbeatRuns.createdAt, run.createdAt), - )).limit(1); - if (newer) return false; - const [result] = await tx.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( - eq(nativeRunResults.companyId, companyId), eq(nativeRunResults.issueId, issue.id), - eq(nativeRunResults.runId, runId), eq(nativeRunResults.id, coordinator.resultId), - eq(nativeRunResults.schemaStatus, "accepted"), - sql`${nativeRunResults.resultJson}#>>'{terminal,runTerminalState}' = 'succeeded'`, - )); - const [action] = await tx.select().from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, companyId), eq(issueRecoveryActions.sourceIssueId, issue.id), - sql`${issueRecoveryActions.evidence}->>'runId' = ${runId}`, - )).orderBy(desc(issueRecoveryActions.createdAt)).limit(1); - const recovery = readSystemRecovery(action?.evidence); - const workspace = readWorkspaceRecovery(action?.evidence ?? null); - if (!result || recovery?.state !== "complete" || recovery.sourceRunId !== run.id || !recovery.preservationOperationIds.length - || workspace?.status !== "recovered" || workspace.runId !== run.id - || !await verifiedWorkspaceRepairReceipt(tx as unknown as Db, run)) return false; - const [intent] = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( - eq(agentWakeupRequests.companyId, companyId), eq(agentWakeupRequests.agentId, run.agentId), - eq(agentWakeupRequests.status, "deferred_issue_execution"), eq(agentWakeupRequests.reason, "task_retry"), - sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`, - sql`${agentWakeupRequests.payload}#>>'{taskRetry,sourceRunId}' = ${runId}`, - )).orderBy(desc(agentWakeupRequests.requestedAt)).limit(1); - const generationId = intent?.id ?? recovery.generationId; - if (coordinator.recoveryHistory.some(entry => entry.kind === "preserved_result_finalization" && - entry.generationId === generationId)) return false; + // Each attempt rereads every fact. A row held by another writer (usually a + // Retry dispatch for this task) is briefly retried without waiting while the + // coordinator is held. The last attempt waits for its locks, as this + // resumption always did, so a recovered result is never left unapplied + // because a row stayed busy. + for (let attempt = 1; ; attempt += 1) { + const wait = attempt >= PRESERVED_RESULT_LOCK_ATTEMPTS; + const resumed = await attemptPreservedResultFinalization(db, companyId, runId, wait); + if (resumed !== "busy") return resumed; + await new Promise(resolve => setTimeout(resolve, PRESERVED_RESULT_LOCK_DELAY_MS)); + } +} + +const PRESERVED_RESULT_LOCK_ATTEMPTS = 10; +const PRESERVED_RESULT_LOCK_DELAY_MS = 100; + +async function attemptPreservedResultFinalization(db: Db, companyId: string, runId: string, wait: boolean): Promise { + const resumed = await db.transaction(async tx => { + const found = await readPreservedResult(tx as unknown as Db, companyId, runId, wait ? "wait" : "nowait"); + if (!found) return null; + const { coordinator, issue, run, result, recovery, intent, generationId } = found; await notifyNativeFinalizationWork(tx, { companyId, runId }); await tx.update(nativeRunFinalizations).set({ phase: coordinator.assessmentId ? "arbitrating" : "workspace_finalizing", @@ -67,6 +136,16 @@ export async function resumePreservedResultFinalization(db: Db, companyId: strin status: "coalesced", runId, finishedAt: new Date(), updatedAt: new Date(), payload: sql`jsonb_set(${agentWakeupRequests.payload}, '{taskRetry,strategy}', '"finalize_result"'::jsonb)`, }).where(eq(agentWakeupRequests.id, intent.id)); - return true; + return persistActivity(tx as unknown as Db, { companyId, actorType: "system", actorId: "task-recovery", + action: "issue.saved_result_finalization_resumed", entityType: "issue", entityId: issue.id, runId: run.id, + details: { resultId: result.id, preservationGenerationId: recovery.generationId, + ...(intent ? { requestId: intent.id, strategy: "finalize_result" } : {}) } }); + }).catch(error => { + if (isLockNotAvailable(error)) return "busy" as const; + throw error; }); + if (resumed === "busy") return "busy"; + if (!resumed) return false; + try { publishActivity(resumed.publication); } catch { logger.warn({ runId }, "Saved result activity delivery deferred"); } + return true; } diff --git a/server/src/services/workspace-recovery-disk-full.test.ts b/server/src/services/workspace-recovery-disk-full.test.ts index e954a781c75..0e9a7ca539d 100644 --- a/server/src/services/workspace-recovery-disk-full.test.ts +++ b/server/src/services/workspace-recovery-disk-full.test.ts @@ -2,8 +2,8 @@ import { randomUUID } from "node:crypto"; import { readdir, readFile } from "node:fs/promises"; import path from "node:path"; import request from "supertest"; -import { and, eq } from "drizzle-orm"; -import { agentWakeupRequests, executionWorkspaces, heartbeatRuns, issues } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; +import { activityLog, agentWakeupRequests, costEvents, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; import { heartbeatService } from "./heartbeat.js"; import { describe, expect, it, vi } from "vitest"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; @@ -13,8 +13,9 @@ import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js"; import { REMOTE_WORKSPACE_SANDBOX_DISK_FULL } from "./remote-workspace-transport.js"; import { getExecutionBlocker } from "./execution-blocker.js"; import { taskInterruptionPresentation } from "./task-interruption-presentation.js"; +import { finalizeNativeRun } from "./native-runtime/native-run-finalizer.js"; import { createSavedWorkspaceLoader, readRemoteWorkspaceAuthority } from "./remote-workspaces.js"; -import { useWorkspaceRecoveryFixture, acceptedRemoteExport, action, db, receiptOnlyRecoveryApp } from "./__tests__/workspace-recovery-fixture.js"; +import { useWorkspaceRecoveryFixture, acceptedRemoteExport, action, db, receiptOnlyRecoveryApp, recoverRunHolds, run } from "./__tests__/workspace-recovery-fixture.js"; const targets = vi.hoisted(() => new Map()); vi.mock("./environment-execution-target.js", async importOriginal => { const actual = await importOriginal(); @@ -22,31 +23,67 @@ vi.mock("./environment-execution-target.js", async importOriginal => { (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; }); +/** An accepted run whose save failed on a full sandbox disk, handed to the board. */ +async function diskFullHold() { + const f = await acceptedRemoteExport(); + const target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!; + const disk = { full: true, saves: 0 }; + target.runner!.syncOut = async operations => { + disk.saves++; + if (disk.full) throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL); + return syncOut(operations); + }; + await recoverRetainedWorkspace(f.input); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)).toMatchObject({ state: "exhausted", + failure: { retryClass: "user_action_required", cause: { label: "sandbox_disk_full" } } }); + // The route's own dispatch is replaced, so each test decides when it lands. + const retry = async () => (await request(receiptOnlyRecoveryApp()).post(`/api/issues/${f.issueId}/retry`) + .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body as { requestId: string; accepted: boolean; observedRunId: string }; + const runIds = async () => (await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(eq(heartbeatRuns.companyId, f.companyId))).map(row => row.id).sort(); + const recoverAll = () => recoverRunHolds(f.input); + return { ...f, disk, retry, runIds, recoverAll }; +} + +const wake = async (id: string) => (await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, id)))[0]!; +const coordinator = async (runId: string) => (await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0]!; + +/** The Retry applied the saved accepted result: no new run, and the source run and task finish from it. */ +async function expectSavedResultApplied(f: Awaited>, requestId: string, runsBefore: string[]) { + expect(await f.runIds()).toEqual(runsBefore); + expect(await wake(requestId)).toMatchObject({ status: "coalesced", runId: f.runId, + payload: { taskRetry: { strategy: "finalize_result" } } }); + expect(await coordinator(f.runId)).toMatchObject({ phase: "workspace_finalizing", resultId: f.resultId, + recoveryHistory: [expect.objectContaining({ kind: "preserved_result_finalization", generationId: requestId, resultId: f.resultId })] }); + expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), + eq(activityLog.action, "issue.saved_result_finalization_resumed")))).toEqual([expect.objectContaining({ + actorType: "system", entityId: f.issueId, runId: f.runId, + details: expect.objectContaining({ requestId, resultId: f.resultId, strategy: "finalize_result" }) })]); + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]) + .toMatchObject({ status: "done", executionRunId: null, checkoutRunId: null }); + expect((await run(f.runId)).status).toBe("succeeded"); + expect(await f.runIds()).toEqual(runsBefore); + expect(await db.select().from(costEvents).where(eq(costEvents.companyId, f.companyId))).toEqual([]); +} + describe("saved-work recovery on a full sandbox disk", () => { useWorkspaceRecoveryFixture(targets); - it("ends recovery at the first full-disk save with a named cause and a Retry that saves once there is room", async () => { - const f = await acceptedRemoteExport(); - const target = targets.get(f.lease.id)!, syncOut = target.runner!.syncOut!; - let full = true, saves = 0; - target.runner!.syncOut = async operations => { - saves++; - if (full) throw new Error(REMOTE_WORKSPACE_SANDBOX_DISK_FULL); - return syncOut(operations); - }; + it("ends recovery at the first full-disk save with a named cause and a Retry that saves and applies the accepted result without a new turn", async () => { + const f = await diskFullHold(); const workspaceBefore = (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)))[0]; - - await recoverRetainedWorkspace(f.input); const failed = readSystemRecovery((await action(f.actionId)).evidence); expect(failed).toMatchObject({ state: "exhausted", nextAttemptAt: null, attempts: 1, failure: { code: "export_interrupted", retryClass: "user_action_required", cause: { label: "sandbox_disk_full", code: "ENOSPC", status: null } } }); - const savesAtFailure = saves; + const savesAtFailure = f.disk.saves; expect(savesAtFailure).toBeGreaterThan(0); await recoverRetainedWorkspace(f.input); await recoverRetainedWorkspace(f.input); - expect(saves).toBe(savesAtFailure); + expect(f.disk.saves).toBe(savesAtFailure); const handedOff = await action(f.actionId); expect(handedOff).toMatchObject({ status: "active", ownerType: "board" }); expect(handedOff.nextAction).toMatch(/sandbox disk is full/); @@ -64,14 +101,12 @@ describe("saved-work recovery on a full sandbox disk", () => { expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ canRetry: true, observedRunId: f.runId }); expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.runId, workspaceRecovery: { canRetry: true } }); - const runsBefore = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId)); + const runsBefore = await f.runIds(); - full = false; - const app = receiptOnlyRecoveryApp(); - const receipt = (await request(app).post(`/api/issues/${f.issueId}/retry`) - .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; + f.disk.full = false; + const receipt = await f.retry(); expect(receipt).toMatchObject({ accepted: true, observedRunId: f.runId }); - await recoverRetainedWorkspace(f.input); + await f.recoverAll(); const recovered = await action(f.actionId); expect(readWorkspaceRecovery(recovered.evidence)?.status).toBe("recovered"); expect(readSystemRecovery(recovered.evidence)?.preservationOperationIds).toHaveLength(1); @@ -81,20 +116,108 @@ describe("saved-work recovery on a full sandbox disk", () => { snapshotId: authority.snapshotId! }); try { expect(await readFile(path.join(saved.root, "saved.txt"), "utf8")).toBe("finished work\n"); } finally { await saved.cleanup(); } - const savesAfterRetry = saves; - await recoverRetainedWorkspace(f.input); - expect(saves).toBe(savesAfterRetry); - expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))) - .toEqual(runsBefore); + const savesAfterRetry = f.disk.saves; + await f.recoverAll(); + expect(f.disk.saves).toBe(savesAfterRetry); + expect(await f.runIds()).toEqual(runsBefore); + // Retry dispatch lands after the save, before anything applied the result + // (also replayed and concurrent). It applies the accepted result and + // never starts another agent turn to repeat the finished work. const heartbeat = heartbeatService(db); await Promise.all([heartbeat.resumeTaskRetry(f.companyId, receipt.requestId), heartbeat.resumeTaskRetry(f.companyId, receipt.requestId)]); await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); + await expectSavedResultApplied(f, receipt.requestId, runsBefore); + }, 60_000); + + it("holds a Retry dispatched while another saved-work hold of the run is still settling, then applies the accepted result", async () => { + const f = await diskFullHold(); + const runsBefore = await f.runIds(); + f.disk.full = false; + const receipt = await f.retry(); + // The recovery worker saves the files for the board hold first and then + // dispatches, while the hold Retry added for the same run is still + // scheduled. The saved files are already verified at that point. + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + const pending = (await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, f.companyId), + sql`${issueRecoveryActions.evidence}->>'runId' = ${f.runId}`))).filter(hold => hold.id !== f.actionId); + expect(pending.map(hold => readSystemRecovery(hold.evidence)?.state)).toContain("scheduled"); + const heartbeat = heartbeatService(db); + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); + expect(await f.runIds()).toEqual(runsBefore); + expect(await wake(receipt.requestId)).toMatchObject({ status: "deferred_issue_execution", + payload: { executionWait: { reason: "finalizing_saved_result" } } }); + await f.recoverAll(); + await heartbeat.resumeTaskRetry(f.companyId, receipt.requestId); + await expectSavedResultApplied(f, receipt.requestId, runsBefore); + }, 60_000); + + it("does not hold a Retry for a settling hold when the run records an operator Stop", async () => { + const f = await diskFullHold(); + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, cancelledByUserId: "local-board" } }) + .where(eq(heartbeatRuns.id, f.runId)); + const runsBefore = await f.runIds(); + f.disk.full = false; + const receipt = await f.retry(); + await recoverRetainedWorkspace(f.input); + await heartbeatService(db).resumeTaskRetry(f.companyId, receipt.requestId); + expect((await f.runIds()).filter(id => !runsBefore.includes(id))).toHaveLength(1); + expect(await coordinator(f.runId)).toMatchObject({ phase: "terminal_failure", recoveryHistory: [] }); + }, 60_000); + + it("keeps the named full-disk rest and a working Retry when the disk is still full", async () => { + const f = await diskFullHold(); + const runsBefore = await f.runIds(); + const first = await f.retry(); + await f.recoverAll(); + await heartbeatService(db).resumeTaskRetry(f.companyId, first.requestId); + expect(readSystemRecovery((await action(f.actionId)).evidence)).toMatchObject({ state: "exhausted", + failure: { retryClass: "user_action_required", cause: { label: "sandbox_disk_full" } } }); + await f.recoverAll(); + expect((await action(f.actionId)).nextAction).toMatch(/sandbox disk is full/); + expect(await f.runIds()).toEqual(runsBefore); + expect(await coordinator(f.runId)).toMatchObject({ phase: "terminal_failure", recoveryHistory: [] }); + const issue = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; + expect(await taskInterruptionPresentation(db, issue)).toMatchObject({ canRetry: true }); + + // Once there is room, the next Retry saves and applies the result. + f.disk.full = false; + const second = await f.retry(); + await f.recoverAll(); + await heartbeatService(db).resumeTaskRetry(f.companyId, second.requestId); + await expectSavedResultApplied(f, second.requestId, runsBefore); + }, 60_000); + + it.each([ + { record: "no accepted result", prepare: async (f: Awaited>) => { + await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.id, f.resultId)); + } }, + { record: "an accepted result that did not succeed", prepare: async (f: Awaited>) => { + await db.update(nativeRunResults).set({ resultJson: sql`jsonb_set(${nativeRunResults.resultJson}, '{terminal,runTerminalState}', '"failed"'::jsonb)` }) + .where(eq(nativeRunResults.id, f.resultId)); + } }, + // The result and saved files are proven, but the run records an operator + // Stop, so applying the result is not provably wanted. + { record: "an operator Stop recorded on the run", prepare: async (f: Awaited>) => { + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, cancelledByUserId: "local-board" } }) + .where(eq(heartbeatRuns.id, f.runId)); + } }, + ])("keeps today's Retry, one new run, with $record", async ({ prepare }) => { + const f = await diskFullHold(); + await prepare(f); + const runsBefore = await f.runIds(); + f.disk.full = false; + const receipt = await f.retry(); + await f.recoverAll(); + expect(readWorkspaceRecovery((await action(f.actionId)).evidence)?.status).toBe("recovered"); + await heartbeatService(db).resumeTaskRetry(f.companyId, receipt.requestId); const successors = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))) - .filter(row => row.id !== f.runId); + .filter(row => !runsBefore.includes(row.id)); expect(successors).toHaveLength(1); expect(successors[0].contextSnapshot).toMatchObject({ previousRunId: f.runId, taskRetryRequestId: receipt.requestId }); - expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), - eq(agentWakeupRequests.idempotencyKey, `task-retry-dispatch:${receipt.requestId}`)))).toHaveLength(1); + expect(await coordinator(f.runId)).toMatchObject({ phase: "terminal_failure", recoveryHistory: [] }); }, 60_000); }); diff --git a/server/src/services/workspace-recovery-preparation-gates.test.ts b/server/src/services/workspace-recovery-preparation-gates.test.ts index e569bd4f3d4..e23443fb8e1 100644 --- a/server/src/services/workspace-recovery-preparation-gates.test.ts +++ b/server/src/services/workspace-recovery-preparation-gates.test.ts @@ -30,7 +30,7 @@ import { preserveRemoteWorkspacePreparationFailure } from "./native-runtime/nati import { getExecutionBlocker } from "./execution-blocker.js"; import { taskInterruptionPresentation } from "./task-interruption-presentation.js"; import { resumePreservedResultFinalization } from "./preserved-result-finalization.js"; -import { useWorkspaceRecoveryFixture, acceptedRemoteExport, action, boardRecoveryApp, db, interruptedRemotePreparation, makeRetryDue, root, run, seed, withheldPreparationRetry } from "./__tests__/workspace-recovery-fixture.js"; +import { useWorkspaceRecoveryFixture, acceptedRemoteExport, action, boardRecoveryApp, db, interruptedRemotePreparation, makeRetryDue, recoverRunHolds, root, run, seed, withheldPreparationRetry } from "./__tests__/workspace-recovery-fixture.js"; const targets = vi.hoisted(() => new Map()); vi.mock("./environment-execution-target.js", async importOriginal => { const actual = await importOriginal(); @@ -289,10 +289,10 @@ describe("automatic retained workspace recovery", () => { // "exhausted before upgrade": the wave databases, where earlier code already // spent the whole budget on rechecks and recorded no hold. it.each([ - { wave: "s13", spentAttempts: 0, expectedStrategy: "finalize_result", exhaustedBeforeUpgrade: false }, - { wave: "s12", spentAttempts: 1, expectedStrategy: "continue", exhaustedBeforeUpgrade: false }, - { wave: "exhausted before upgrade", spentAttempts: 0, expectedStrategy: "finalize_result", exhaustedBeforeUpgrade: true }, - ])("lets an accepted Retry finish after a killed copyback owner instead of waiting forever ($wave)", async ({ spentAttempts, expectedStrategy, exhaustedBeforeUpgrade }) => { + { wave: "s13", spentAttempts: 0, exhaustedBeforeUpgrade: false }, + { wave: "s12", spentAttempts: 1, exhaustedBeforeUpgrade: false }, + { wave: "exhausted before upgrade", spentAttempts: 0, exhaustedBeforeUpgrade: true }, + ])("lets an accepted Retry finish after a killed copyback owner instead of waiting forever ($wave)", async ({ spentAttempts, exhaustedBeforeUpgrade }) => { const f = await acceptedRemoteExport(); const issue = (await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]; await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.source.agentId, status: "running" }); @@ -354,25 +354,19 @@ describe("automatic retained workspace recovery", () => { stopEvidence: "The killed control-plane process exited and its host was replaced.", }).expect(202); expect((await run(f.runId)).runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeUndefined(); - // The same steps the periodic sweep runs for this action. - await recoverRetainedWorkspace(f.input); + // The same steps the periodic sweep runs for this run's holds. + await recoverRunHolds(f.input); expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); await resumePreservedResultFinalization(db, f.companyId, f.runId); await heartbeatService(db).resumeTaskRetry(f.companyId, retry.requestId!); - // The one accepted click now does work: it finishes the saved result, or, - // after an earlier recovery handoff (s12), continues in exactly one new run. + // The one accepted click now does work: it finishes the saved result, + // also after an earlier recovery failure (s12), with no new agent turn. const [settled] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, retry.requestId!)); - expect(settled).toMatchObject({ status: "coalesced", runId: expect.any(String) }); - if (expectedStrategy === "finalize_result") { - expect(settled).toMatchObject({ runId: f.runId, payload: { taskRetry: { strategy: "finalize_result" } } }); - await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); - expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "done" }); - } else { - expect(settled.payload).toMatchObject({ taskRetry: { strategy: "continue" } }); - expect(await db.select({ id: heartbeatRuns.id, previousRunId: sql`${heartbeatRuns.contextSnapshot}->>'previousRunId'` }) - .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))) - .toEqual([{ id: settled.runId, previousRunId: f.runId }]); - } + expect(settled).toMatchObject({ status: "coalesced", runId: f.runId, payload: { taskRetry: { strategy: "finalize_result" } } }); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued")))).toEqual([]); + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "done" }); }, 60_000); // RC-50b (wave s14/t05): Retry was accepted while saved-work recovery ran. @@ -403,17 +397,17 @@ describe("automatic retained workspace recovery", () => { const second = (await request(app).post(`/api/issues/${f.issueId}/retry`) .send({ observedRunId: f.runId, idempotencyKey: randomUUID() }).expect(202)).body; expect(readSystemRecovery((await action(f.actionId)).evidence)).toMatchObject({ state: "scheduled", attempts: 0 }); - await recoverRetainedWorkspace(f.input); + await recoverRunHolds(f.input); expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); - await resumePreservedResultFinalization(db, f.companyId, f.runId); await heartbeatService(db).resumeTaskRetry(f.companyId, second.requestId!); - // After a board handoff the Retry continues from the recovered files in - // exactly one new run (as in the s12 case), never another wait. + // After a board handoff the Retry applies the accepted result from the + // recovered files, never another wait and never a new agent turn. const [settled] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, second.requestId!)); - expect(settled).toMatchObject({ status: "coalesced", runId: expect.any(String), payload: { taskRetry: { strategy: "continue" } } }); - expect(await db.select({ id: heartbeatRuns.id, previousRunId: sql`${heartbeatRuns.contextSnapshot}->>'previousRunId'` }) - .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))) - .toEqual([{ id: settled.runId, previousRunId: f.runId }]); + expect(settled).toMatchObject({ status: "coalesced", runId: f.runId, payload: { taskRetry: { strategy: "finalize_result" } } }); + expect(await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), + eq(heartbeatRuns.status, "queued")))).toEqual([]); + await finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "done" }); }, 60_000); it("names a deterministic saved-files refusal behind Retry when no fallback can continue, and another Retry shows it again", async () => { diff --git a/ui/src/components/WorkspaceExportRecovery.control.test.tsx b/ui/src/components/WorkspaceExportRecovery.control.test.tsx index b4731f092ac..799f9958366 100644 --- a/ui/src/components/WorkspaceExportRecovery.control.test.tsx +++ b/ui/src/components/WorkspaceExportRecovery.control.test.tsx @@ -373,6 +373,25 @@ describe("governed retained-file recovery", () => { expect(localStorage.getItem("workspace-file-recovery:company-1:issue-1:action-1:run-1:board-1")).toContain(previous); }); + it("hands a full-disk save rest to the task's Retry instead of its own recovery button", async () => { + const diskFull = { ...action, ownerType: "board", evidence: { ...action.evidence, systemRecovery: { + state: "exhausted", preservationOperationIds: [], failure: { code: "export_interrupted", + retryClass: "user_action_required", cause: { label: "sandbox_disk_full", code: "ENOSPC", status: null } } } }, + } as unknown as IssueRecoveryAction; + api.inspect.mockResolvedValue(inspection(diskFull)); + const owns = vi.fn(); + await act(async () => root.render( + + )); + await settle(); + expect(button("Recover saved files")).toBeUndefined(); + expect(button("Try file recovery again")).toBeUndefined(); + expect(owns).toHaveBeenLastCalledWith(false); + expect(owns).not.toHaveBeenCalledWith(true); + expect(container.textContent).toContain("use Retry to save them"); + expect(api.recover).not.toHaveBeenCalled(); + }); + it("only reports recovered files after the durable preservation receipt is present", async () => { const recovered = { ...action, evidence: { ...action.evidence, workspaceRecovery: { ...(action.evidence.workspaceRecovery as object), status: "recovered", diff --git a/ui/src/components/WorkspaceExportRecovery.tsx b/ui/src/components/WorkspaceExportRecovery.tsx index 600a2ccf150..89ad306f0ac 100644 --- a/ui/src/components/WorkspaceExportRecovery.tsx +++ b/ui/src/components/WorkspaceExportRecovery.tsx @@ -125,6 +125,14 @@ function isRetainedExport(action: IssueRecoveryAction, issueId: string, companyI && typeof action.evidence.runId === "string" && action.evidence.runId.length > 0; } +/** Saved-work recovery ended because the sandbox disk was still full (#20758). */ +function isSandboxDiskFullRest(action: IssueRecoveryAction) { + const system = asObject(action.evidence.systemRecovery); + const failure = asObject(system.failure); + return system.state === "exhausted" && failure.retryClass === "user_action_required" + && asObject(failure.cause).label === "sandbox_disk_full"; +} + function isControllerStopRequired(action: IssueRecoveryAction, issueId: string, companyId: string) { return action.companyId === companyId && action.sourceIssueId === issueId && action.ownerType === "board" && ["active", "escalated"].includes(action.status) @@ -343,21 +351,26 @@ function RetainedFilesRecovery({ action, issueId, companyId, className, onRecove const waiting = !failedReceipt && !saved && (restoring || acknowledged || serverAcknowledged); const unconfirmed = Boolean(pending && !acknowledged && !serverAcknowledged); const sourceLoading = sourceView.kind === "loading" || sourceView.kind === "reconnecting"; + // A save that rested on a full sandbox disk is resumed by the task's Retry, + // which saves the files again and then applies the saved result. Its note + // names Retry, so this panel offers no separate control for it. + const diskFullRest = isSandboxDiskFullRest(current ?? action); // A verified source makes this panel the governed recovery path. Otherwise the // task's own Retry (renewed, re-verified recovery) remains the next action. - const ownsRecoveryControl = canMutate && (sourceLoading || Boolean(current && verified && !saved)); + const ownsRecoveryControl = canMutate && !diskFullRest && (sourceLoading || Boolean(current && verified && !saved)); // `waiting` already excludes a failed accepted request, so the reason stays // visible after a failed attempt and hides only while recovery is running. const unverified = Boolean(current && !verified && !saved && !waiting); return

{saved ? "Saved files recovered. Task recovery is still being checked." + : diskFullRest && !waiting ? "The sandbox disk was full when Paperclip saved the files. Free space, then use Retry to save them and finish the task." : failedReceipt ? "File recovery has not finished. The saved source remains protected." : restoring ? "Recovering saved files. Work remains paused until recovery is verified." : waiting ? "File recovery is queued. Waiting for verified saved files." : "The saved files need recovery before work can continue."}

{canMutate ? <> - {current && verified && retryable && !waiting && (!pending || unconfirmed && checkedUnconfirmed || confirmedFailedAttempt) ? : null}