mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 16:02:36 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
71313207eb
commit
f513f0efd9
3 files changed
+136
-3
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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<string, unknown> | undefined;
|
||||
const preparationFailure = run.errorCode === "setup_failed" || historicalRetryAncestor && run.errorCode === "adapter_failed";
|
||||
if (run.status !== "failed" || !run.finishedAt || run.runtimeMode !== "legacy" || run.runtimeModeResolvedAt
|
||||
|
||||
@@ -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<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -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));
|
||||
|
||||
Reference in new issue
Block a user