From f513f0efd9616a5f8e8e76ca6980e7624e8fc05a Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 6 Nov 2026 13:03:21 -0600 Subject: [PATCH] fix(recovery): admit certified process-loss preparation retries (#19552) ## Thinking Path > - Paperclip keeps task work durable across server restarts. > - Restart recovery records a certificate when a Runner preparation ended before provider work started. > - Automatic recovery already verifies that certificate together with saved files and stopped leases. > - Explicit Retry called a preparation helper that only reached this proof for status `interrupted`. > - A certified `failed/process_lost` source therefore stayed blocked after its files were recovered. > - This pull request makes the helper reuse the same existing proof. > - A queued user Retry can then admit one new turn while all current safety and policy gates remain effective. ## Linked Issues or Issue Description **What happened?** A task had a certified process-lost preparation, verified saved-workspace receipts, a destroyed source sandbox, and one or two later attempts refused before dispatch. Its accepted Retry request stayed deferred with `execution_recovery` and created no successor. The explicit admission helper excluded this source before it could use the preparation proof added in #19464. This also prevented the refused-attempt continuation path from #19466 from advancing. **Expected behavior** Explicit Retry must use the same positive pre-provider proof as restart recovery. It must preserve source history, refuse contradictory execution evidence, honor current policy gates, and admit at most one successor for the existing request. **Steps to reproduce** Persist a legacy preparing run as failed with `process_lost` and the host bootstrap certificate that provider work did not start. Record its canonical committed snapshot, recovery/finalization operations, clean workspace authority, and exact destroyed-lease receipt. Add one or two unstarted cancelled descendants with the dispatcher's execution-wait receipt. Accept one Retry request and resume it concurrently twice. Before this fix, no successor exists and the same request remains deferred. **Paperclip version or commit** Reproduced on master `a6c22806308e8a3c7b21145f80fb471756bb443d`. Related repairs: #19464 and #19466. ## What Changed - Reuse `isInterruptedNativePreparation` before the existing failed-preparation checks. The existing no-coordinator and historical-ancestor restrictions remain. - Add real-database regressions for one and two refused descendants, duplicate clicks, concurrent resume, continuation dispatch, preserved history, and unchanged recovery attempt counts. - Add late-gate checks for provider work, provider events, a newer native checkpoint, missing bootstrap proof, missing stop proof, a changed snapshot, formal approval, user Stop, agent/company pause, and a spent budget. - Clarify that explicit admission uses the existing certified preparation proof. ## Verification - Red: both real-database cases passed the saved-file, interrupted-preparation, and refused-dispatch proof checks, but the original successor-count assertion found zero queued runs instead of one. The accepted requests retained `execution_recovery`. - Green: the same cases admit one successor after two concurrent resume calls. Continuation dispatch resolves to the latest refused attempt. Source and cancelled-attempt history stay intact. - Complete relevant local suites: **456/456 passed** before rebase. Command from `server`: `../node_modules/.bin/vitest run --config vitest.config.ts --fsModuleCache src/services/task-retry.test.ts src/services/explicit-native-continuation.test.ts`. - Final head `9788382c019df57899af06e81211e7c5165e7f0d`: the thirteen new cases passed **13/13** again after rebase. Required CI and review remain pending. - Environment: macOS ARM64, Node 26, Vitest 5.0.3, disposable real PostgreSQL. Tests exercise production request, admission, and continuation services. No provider executes and no hosted task is changed. - Hosted recovery still requires rollout and verification through the original queued requests. Final reviewed head: `9788382c019df57899af06e81211e7c5165e7f0d`. [Required CI](https://github.com/paperclipai/paperclip/actions/runs/42273663624) passed (56 successful checks, four expected skips, none pending or failed). Greptile and Paperclip Review both rated this head 5/5 with no findings; all review threads are resolved. ## Risks The change accepts only the existing verified preparation contract. An error label alone is insufficient. Provider-started or contradictory state remains blocked. Saved-file, lease, approval, Stop, pause, and budget checks remain effective. No schema, retry limit, provider setting, account selection, or production scheduler changes. ## Model Used OpenAI GPT-6 family via Codex, with coding and tool use. The exact model ID and context window are unavailable in this session's metadata. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/execution-semantics.md | 5 +- .../src/services/cancelled-native-startup.ts | 2 +- server/src/services/task-retry.test.ts | 132 +++++++++++++++++- 3 files changed, 136 insertions(+), 3 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index eb7f8cff4d3..84141d4f0d5 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2657,7 +2657,10 @@ leases before the task can continue. Once these checks pass, the server retires its own unchanged bootstrap-failure block and schedules one bounded retry. Startup also discovers this obligation when file recovery completed before the restart. A pending Retry or saved message owns its continuation instead of -starting a competing automatic attempt. If a replacement was already withheld +starting a competing automatic attempt. Explicit admission uses the same verified +preparation proof, including the certified failed `process_lost` form. A prior +unstarted retry refused by the execution gate remains historical evidence; its +host dispatch receipt must also verify before the new turn can start. If a replacement was already withheld by the workspace gate before execution, continuation retains that cancelled row and uses its existing predecessor claim and remaining failure budget. Recovery adopts that withheld row only when it is the source's sole retry child. diff --git a/server/src/services/cancelled-native-startup.ts b/server/src/services/cancelled-native-startup.ts index d151d2020b5..bd6eaac678f 100644 --- a/server/src/services/cancelled-native-startup.ts +++ b/server/src/services/cancelled-native-startup.ts @@ -31,7 +31,7 @@ export async function isFailedNativePreparationRetryAncestor(db: Db, run: Run, c } async function failedNativePreparation(db: Db, run: Run, coordinator: Coordinator | undefined, historicalRetryAncestor: boolean) { - if (run.status === "interrupted") return !historicalRetryAncestor && !coordinator && isInterruptedNativePreparation(db, run); + if (!historicalRetryAncestor && !coordinator && await isInterruptedNativePreparation(db, run)) return true; const recovery = run.resultJson?.executionRecovery as Record | undefined; const preparationFailure = run.errorCode === "setup_failed" || historicalRetryAncestor && run.errorCode === "adapter_failed"; if (run.status !== "failed" || !run.finishedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt diff --git a/server/src/services/task-retry.test.ts b/server/src/services/task-retry.test.ts index 995923a4e77..b2b2c42a25e 100644 --- a/server/src/services/task-retry.test.ts +++ b/server/src/services/task-retry.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { agents, agentWakeupRequests, chatConversations, chatEndpoints, companies, createDb, type Db, environmentLeases, environments, executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueComments, issues, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; +import { agents, agentWakeupRequests, chatConversations, chatEndpoints, companies, budgetPolicies, costEvents, createDb, type Db, environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, workspaceOperations, heartbeatRunEvents, heartbeatRuns, issueComments, issues, approvals, issueApprovals, issueRecoveryActions, nativeRunFinalizations, toolApplications, toolConnections, toolInvocations, applyPendingMigrations } from "@paperclipai/db"; import { startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; import { priorRecoveryApiReceipt, unresolvedRecoveryToolInvocation } from "./recovery-effect-receipts.js"; import { requestTaskRetry } from "./task-retry.js"; @@ -16,6 +16,9 @@ import { taskChatRetryCanAwaitRecovery, taskChatRetryExecutionStopped } from "./ import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js"; import { buildExecutionContinuation } from "./execution-continuation.js"; import { taskRetryRunAncestry } from "./recovery-run-ancestry.js"; +import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { isInterruptedNativePreparation } from "./cancelled-native-startup.js"; +import { isPreDispatchExecutionWaitVerified } from "./pre-dispatch-execution-wait.js"; describe("durable task Retry admission", () => { let postgres: Awaited>; @@ -257,6 +260,133 @@ describe("durable task Retry admission", () => { } }); + async function seedRecoveredLostPreparation(refusedCount: number) { + const { input, source, refused, hold } = await seedRefusedRetry(); + const workspaceId = randomUUID(), leaseId = randomUUID(), snapshotId = randomUUID(); + const snapshotOperationId = randomUUID(), recoveryOperationId = randomUUID(); + const providerLeaseId = `fixture-sandbox-${randomUUID()}`, sha256 = "a".repeat(64); + const restore = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] }; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: input.companyId, + sourceIssueId: input.issueId, mode: "isolated_workspace", strategyType: "isolated_workspace", name: "Recovered work", + metadata: { remoteWorkspace: { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId, + source: { kind: "empty" }, location: null, writerRunId: null, dirty: false } } }); + const [recoveredSource] = await db.update(heartbeatRuns).set({ errorCode: "process_lost", + startedAt: source.createdAt, controllerBootId: randomUUID(), controllerLeaseExpiresAt: source.finishedAt, + runnerProfileJson: { ...source.runnerProfileJson, remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", + workspaceId, leaseId, providerLeaseId, remoteCwd: "/fixture/saved-work", epoch: 1, state: "saved", snapshotId } }, + resultJson: { stopReason: "process_lost", executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: restore }, + }).where(eq(heartbeatRuns.id, source.id)).returning(); + await db.insert(environmentLeases).values({ id: leaseId, companyId: input.companyId, issueId: input.issueId, + heartbeatRunId: source.id, executionWorkspaceId: workspaceId, provider: "daytona", providerLeaseId, + status: "expired", leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "success", + metadata: { remoteExecutionTermination: { schema: "paperclip.remote-termination.v1", companyId: input.companyId, + runId: source.id, leaseId, provider: "daytona", providerLeaseId, state: "destroyed", confirmedAt: new Date().toISOString() } } }); + await db.insert(executionWorkspaceSnapshots).values({ id: snapshotId, companyId: input.companyId, + executionWorkspaceId: workspaceId, sourceOperationId: snapshotOperationId, sourceRunId: source.id, + generation: 1, epoch: 1, status: "committed", sha256 }); + const scope = { companyId: input.companyId, issueId: input.issueId, heartbeatRunId: source.id, + executionWorkspaceId: workspaceId, status: "succeeded", finishedAt: new Date() }; + await db.insert(workspaceOperations).values([ + { ...scope, id: snapshotOperationId, phase: "remote_workspace_snapshot", metadata: { snapshotId, sha256, epoch: 1, generation: 1 } }, + { ...scope, id: recoveryOperationId, phase: "workspace_recovery", metadata: { recoveryActionId: hold.id, + storage: "remote_workspace", leaseId, providerLeaseId, snapshotId, sha256, epoch: 1, generation: 1 } }, + { ...scope, phase: "workspace_finalize", metadata: { owningService: "workspace_recovery", preservationOperationIds: [recoveryOperationId] } }, + ]); + await db.update(issueRecoveryActions).set({ evidence: { ...hold.evidence, workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: restore, continuationDelivery: "invalidated", + systemRecovery: { ...createSystemRecovery(source.id), state: "complete", phase: "reconcile_effects", attempts: 1, + preservationOperationIds: [recoveryOperationId] }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: source.id, status: "recovered", + stage: "saving_receipts", attempt: 1, attemptLimit: 6, requestIds: [] } } }).where(eq(issueRecoveryActions.id, hold.id)); + let latest = refused; + if (refusedCount === 2) [latest] = await db.insert(heartbeatRuns).values({ companyId: input.companyId, + agentId: source.agentId, issueId: input.issueId, status: "cancelled", errorCode: refused.errorCode, + retryOfRunId: refused.id, createdAt: new Date(refused.finishedAt!.getTime() + 1), finishedAt: new Date(), + contextSnapshot: { issueId: input.issueId, retryOfRunId: refused.id }, resultJson: refused.resultJson }).returning(); + expect(await hasVerifiedWorkspaceRepair(db, recoveredSource)).toBe(true); + expect(await isInterruptedNativePreparation(db, recoveredSource)).toBe(true); + expect(await isPreDispatchExecutionWaitVerified(db, latest)).toBe(true); + return { input: { ...input, observedRunId: latest.id }, source: recoveredSource, refused: latest, + hold, leaseId, snapshotId }; + } + + it.each([1, 2])("dispatches a recovered process-lost preparation after %i refused automatic attempts", async refusedCount => { + const { input, source, refused, hold } = await seedRecoveredLostPreparation(refusedCount); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId)); + await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, status: "running" }); + const receipt = await requestTaskRetry(input); + expect((await requestTaskRetry({ ...input, idempotencyKey: randomUUID() })).requestId).toBe(receipt.requestId); + const heartbeat = heartbeatService(db); + await Promise.all([heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!), + heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!)]); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, receipt.requestId!)); + const queued = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.status, "queued"))); + expect(queued, JSON.stringify(wake.payload)).toHaveLength(1); + expect(wake).toMatchObject({ status: "coalesced", runId: queued[0].id }); + await expect(dispatchRun(input, queued[0])).resolves.toMatchObject({ interruptedRunId: refused.id }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, refused.id)))[0]) + .toMatchObject({ status: "cancelled", resultJson: refused.resultJson }); + const [settled] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)); + expect(settled.evidence.systemRecovery).toMatchObject({ attempts: 1, retriesUsed: 0 }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, source.id)))[0]) + .toMatchObject({ status: "failed", resultJson: source.resultJson }); + }); + + it.each(["provider_work_started", "provider_event", "native_checkpoint", "missing_bootstrap_proof", + "missing_stop_receipt", "changed_snapshot", "pending_approval", "late_operator_stop", "agent_paused", "company_paused", "budget_hard_stop"] as const)( + "holds a recovered process-lost preparation Retry when a current gate changes: %s", async gate => { + const { input, source, hold, leaseId, snapshotId } = await seedRecoveredLostPreparation(2); + await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId)); + await db.insert(heartbeatRuns).values({ companyId: input.companyId, agentId: source.agentId, status: "running" }); + const receipt = await requestTaskRetry(input); + if (gate === "provider_work_started" || gate === "missing_bootstrap_proof") await db.update(heartbeatRuns) + .set({ resultJson: { ...source.resultJson, executionRecovery: { kind: "bootstrap", + ...(gate === "provider_work_started" ? { providerWorkStarted: true } : {}) } } }).where(eq(heartbeatRuns.id, source.id)); + if (gate === "provider_event") await db.insert(heartbeatRunEvents).values({ companyId: input.companyId, + agentId: source.agentId, runId: source.id, seq: 1, eventType: "tool.execution.started" }); + if (gate === "native_checkpoint") { + await db.update(heartbeatRuns).set({ nativeIssueId: input.issueId }).where(eq(heartbeatRuns.id, source.id)); + await db.insert(nativeRunFinalizations).values({ companyId: input.companyId, issueId: input.issueId, + runId: source.id, phase: "running", attempt: 1, controllerGeneration: 1 }); + } + if (gate === "missing_stop_receipt") await db.update(environmentLeases).set({ metadata: {} }).where(eq(environmentLeases.id, leaseId)); + if (gate === "changed_snapshot") await db.update(executionWorkspaceSnapshots).set({ sha256: "b".repeat(64) }) + .where(eq(executionWorkspaceSnapshots.id, snapshotId)); + if (gate === "pending_approval") { + const [approval] = await db.insert(approvals).values({ companyId: input.companyId, type: "hire_agent", status: "pending", payload: {} }).returning(); + await db.insert(issueApprovals).values({ companyId: input.companyId, issueId: input.issueId, approvalId: approval.id }); + } + if (gate === "late_operator_stop") await db.update(issues).set({ status: "cancelled", cancelledAt: new Date() }).where(eq(issues.id, input.issueId)); + if (gate === "agent_paused") await db.update(agents).set({ status: "paused", pausedAt: new Date() }).where(eq(agents.id, source.agentId)); + if (gate === "company_paused") await db.update(companies).set({ status: "paused", pausedAt: new Date() }).where(eq(companies.id, input.companyId)); + if (gate === "budget_hard_stop") { + await db.insert(budgetPolicies).values({ companyId: input.companyId, scopeType: "company", scopeId: input.companyId, + metric: "billed_cents", windowKind: "lifetime", amount: 1, notifyEnabled: false }); + await db.insert(costEvents).values({ companyId: input.companyId, agentId: source.agentId, heartbeatRunId: source.id, + provider: "fixture", model: "fixture", costCents: 1, occurredAt: new Date() }); + } + const heartbeat = heartbeatService(db); + const attempts = await Promise.allSettled([heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!), + heartbeat.resumeTaskRetry(input.companyId, receipt.requestId!)]); + for (const attempt of attempts) { + if (["agent_paused", "company_paused", "budget_hard_stop"].includes(gate)) { + expect(attempt.status).toBe("rejected"); + if (attempt.status === "rejected") { + expect(attempt.reason).toBeInstanceOf(HttpError); + expect(attempt.reason.status).toBe(409); + expect(attempt.reason.message).toMatch(/not invokable|not active|budget hard-stop/); + } + } else expect(attempt.status).toBe("fulfilled"); + } + expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.status, "queued")))).toHaveLength(0); + if (!["late_operator_stop", "agent_paused", "company_paused", "budget_hard_stop"].includes(gate)) { + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)))[0].evidence.explicitUserContinuation) + .toBeUndefined(); + } + }); + it("dispatches one Retry continuation after a refused automatic attempt", async () => { const { input, source, refused } = await seedRefusedRetry(); await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, source.agentId));