mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(recovery): finish the passive recovery pass before restarting it (#20680)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The workspace recovery worker saves the files of failed runs. One passive pass of the worker visits holds with obligations: saved input that waits, cleanup, and due re-arms and fallbacks of exhausted holds > - #17741 restarts that pass after each committed workspace-recovery change, so new work is found quickly > - On staging, changes commit continuously (stop receipts, new holds, recovery progress). Every pass restarted from the first candidates, which are settled holds whose saved input waits behind another hold, and it never reached later candidates > - After #20295, #20569 and #20661 were deployed, no due re-arm ran: five stop-proof holds and about twenty capture-policy holds kept their exhausted state, with no marker and no activity > - This pull request restarts the pass only after it reaches its tail, and it continues an unfinished pass on the next run > - The benefit is that every candidate is visited on a busy instance, so the exhausted holds re-arm, save their files and free their tasks without an administrator ## Linked Issues or Issue Description Refs #17741 Refs #20295 Refs #20569 Refs #20661 **What happened?** Exhausted workspace holds were not re-armed on staging, even after the fixes that add their re-arm were deployed. Their recovery actions stayed unchanged for days (`updatedAt` on Nov 5 or Nov 10). No `stopProofRecovery` or `capturePolicyRecovery` marker and no `issue.workspace_recovery_rearmed` activity was recorded. The worker did process new restore holds, which are selected by its unfinished pass. **What did you expect to happen?** The worker's passive pass reaches every hold with a due re-arm or fallback, re-arms it, and saves the files. **Steps to reproduce** 1. Create several settled holds with saved user input that waits, with action ids that sort before an exhausted stop-proof hold. 2. Start the real workspace recovery worker. 3. Commit a workspace-recovery notification every 50 ms, as stop receipts and new holds do on a busy instance. 4. The worker never re-arms the exhausted hold. Without the notifications, it re-arms the hold. ## What Changed - `workspace-recovery-work.ts`: a committed change marks the passive pass for restart (`restart`). It no longer resets the cursor. While a restart is pending, an unfinished pass that made progress continues on the next run. - `workspace-recovery.ts`: when the passive cursor passes the last candidate and a restart is pending, the same run starts a fresh pass from the first candidate, so a new change is not delayed by another run interval. Without a pending restart the pass records `complete` and does not wrap, so paused input does not keep the worker active. - `doc/execution-semantics.md`: document that a pass is never restarted midway. - `workspace-recovery-stop-proof.test.ts`: add a real-worker regression in the staging shape (six earlier settled holds with waiting user input, one exhausted stop-proof hold, continuous notifications). ## Verification - Red: on master, the new test "re-arms the hold on a busy instance whose passive sweep keeps selecting earlier settled holds" fails (`stopProofRecovery` stays undefined). Control: the same test without the notifications passes on master. This shows that the restarts cause the starvation. - Green: with the fix, the test passes. These suites pass: `workspace-recovery-stop-proof` (12), `execution-control-work` (27), which includes "visits a newly committed earlier Retry during a busy recovered-source pass, then becomes quiet", plus `workspace-recovery-capture-policy`, `workspace-recovery-preparation-dispatch`, `workspace-recovery-warm-handoff`, `workspace-recovery-preparation` and `server-startup-feedback-export` (279). The timing-sensitive `heartbeat-observed-workspace-preservation` suite passed 6 of 6 runs (an earlier revision that restarted only on the next run made it flaky). - Staging evidence (read-only GET requests): the 28 exhausted holds of the company have no re-arm marker and no fallback check. Many blocked tasks with settled holds have deferred wakes. ## Risks - A change committed during a pass is now picked up at the next pass, not at once. Unfinished physical work still uses its own pass first, so new recoveries are not delayed. Saved input in a passive obligation can wait up to one full pass. - Previous-release story: no persisted state changes. Holds saved by earlier releases are selected on the first full pass after deploy, with no migration. - Quiet behavior stays: a completed pass with no new change keeps the worker idle. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use (shell, code editing, browser GET reads) in Claude Code. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
a8f21a171c
commit
69c69e4c20
4 files changed
+91
-10
No files matched your search
@@ -3776,7 +3776,10 @@ infrastructure certification, and grants no agent or provider execution authorit
|
||||
Each completed export frees its own slot and wakes the next scan. A slow export
|
||||
does not hold a free slot. Completed cleanup and saved-input obligations receive
|
||||
one bounded pass through unvisited sources. The pass stops at its tail so paused
|
||||
input does not keep the worker active; a committed change starts a fresh pass.
|
||||
input does not keep the worker active; a committed change starts a fresh pass
|
||||
once the current pass reaches its tail. A pass is never restarted midway, so on
|
||||
a busy instance, where changes commit continuously, every candidate is still
|
||||
visited: the due re-arms and fallbacks of older exhausted holds included.
|
||||
Shutdown joins this worker's exports before closing the database. Repeated Retry
|
||||
during physical preservation binds the canonical intent
|
||||
without changing the attempt's generation or revision. Progress writes preserve
|
||||
|
||||
@@ -3,21 +3,25 @@ import { readFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { activityLog, agentWakeupRequests, environmentLeases, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target";
|
||||
import { readRemoteWorkspaceAuthority } from "@paperclipai/shared";
|
||||
import { readRemoteWorkspaceWriterHold, remoteWorkspaceHoldRefusesStart } from "./remote-workspace-writer-hold.js";
|
||||
import { readSystemRecovery } from "./system-recovery-state.js";
|
||||
import { adoptSystemRecovery } from "./system-recovery.js";
|
||||
import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { DELIVERY_QUEUES, notifyDeliveryWork, subscribeDeliveryWork } from "./delivery-work-notifications.js";
|
||||
import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { registerWorkspaceRecoveryWork } from "./workspace-recovery-work.js";
|
||||
import { prepareSandboxStopAndRetain, settleStopOnlyCleanup } from "./sandbox-stop-and-retain.js";
|
||||
import { nextWorkspaceRecoveryAt } from "./system-recovery.js";
|
||||
import { RETAINED_SOURCE_FALLBACK_OUTCOME } from "./workspace-recovery-fallback.js";
|
||||
import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js";
|
||||
import { reconcileRetainedWorkspaces, recoverRetainedWorkspace, retainedWorkspaceCandidates } from "./workspace-recovery.js";
|
||||
import { reconcileRetainedWorkspaces, recoverRetainedWorkspace, retainedWorkspaceCandidates, type PassiveSweep } from "./workspace-recovery.js";
|
||||
import { acceptedRemoteExport, action, connectionString, db, run, seed, useWorkspaceRecoveryFixture } from "./__tests__/workspace-recovery-fixture.js";
|
||||
|
||||
const targets = vi.hoisted(() => new Map<string, AdapterSandboxExecutionTarget>());
|
||||
vi.mock("./execution-control-deadline.js", async importOriginal => ({
|
||||
...await importOriginal<typeof import("./execution-control-deadline.js")>(), EXECUTION_RECONCILIATION_INTERVAL_MS: 200 }));
|
||||
vi.mock("./environment-execution-target.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("./environment-execution-target.js")>();
|
||||
return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters<typeof actual.resolveEnvironmentExecutionTarget>[0]) =>
|
||||
@@ -292,6 +296,69 @@ describe("exhausted stop-proof holds", () => {
|
||||
expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).toContain(f.actionId);
|
||||
}, 60_000);
|
||||
|
||||
async function waitingSettledHold(id: string) {
|
||||
const f = await seed("native", true);
|
||||
settledRuns.set(id, f.runId);
|
||||
await adoptSystemRecovery(db, f);
|
||||
const prior = await action(f.actionId);
|
||||
const system = readSystemRecovery(prior.evidence)!;
|
||||
await db.update(issueRecoveryActions).set({ id, status: "resolved", outcome: "blocked", ownerType: "system", evidence: { ...prior.evidence,
|
||||
systemRecovery: { ...system, state: "complete", phase: "reconcile_effects", attempts: 1 },
|
||||
workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "recovered", stage: "saving_receipts",
|
||||
attempt: 1, attemptLimit: 6, retryable: false, requestIds: [] },
|
||||
automaticRecovery: { runId: f.runId, policy: "preserve_without_replay_v1", replay: "blocked" } } })
|
||||
.where(eq(issueRecoveryActions.id, f.actionId));
|
||||
const source = await run(f.runId);
|
||||
await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: source.agentId, source: "on_demand",
|
||||
reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "board",
|
||||
payload: { issueId: f.issueId } });
|
||||
return f;
|
||||
}
|
||||
|
||||
const settledRuns = new Map<string, string>();
|
||||
|
||||
it("restarts a finished passive pass when a change adds work both before and after its cursor", async () => {
|
||||
const first = await waitingSettledHold("00000002-0000-4000-8000-000000000000");
|
||||
await waitingSettledHold("00000003-0000-4000-8000-000000000000");
|
||||
const visited: string[] = [];
|
||||
const onRecovered = async (source: { id: string }) => { visited.push(source.id); };
|
||||
const sweep: PassiveSweep = {};
|
||||
for (let pass = 0; pass < 4 && !sweep.complete; pass += 1)
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, first.environmentRuntime, onRecovered, true, undefined, sweep) ?? []);
|
||||
expect(sweep.complete).toBe(true);
|
||||
await waitingSettledHold("00000001-0000-4000-8000-000000000000");
|
||||
const f = await exhaustedStopProofHold();
|
||||
await db.update(issueRecoveryActions).set({ id: "ffffffff-0000-4000-8000-000000000000" }).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
sweep.restart = true;
|
||||
visited.length = 0;
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true, undefined, sweep) ?? []);
|
||||
expect(sweep).toMatchObject({ complete: false, restart: true });
|
||||
for (let pass = 0; pass < 6 && !(sweep.complete && !sweep.restart && visited.includes(settledRuns.get("00000001-0000-4000-8000-000000000000")!)); pass += 1)
|
||||
await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true, undefined, sweep) ?? []);
|
||||
expect(visited).toContain(settledRuns.get("00000001-0000-4000-8000-000000000000"));
|
||||
expect((await action("ffffffff-0000-4000-8000-000000000000")).evidence.stopProofRecovery).toMatchObject({ rearms: 1 });
|
||||
}, 60_000);
|
||||
|
||||
it("re-arms the hold on a busy instance whose passive sweep keeps selecting earlier settled holds", async () => {
|
||||
for (const prefix of ["00000001", "00000002", "00000003", "00000004", "00000005", "00000006"])
|
||||
await waitingSettledHold(`${prefix}-0000-4000-8000-000000000000`);
|
||||
const f = await exhaustedStopProofHold();
|
||||
const holdId = "ffffffff-0000-4000-8000-000000000000";
|
||||
await db.update(issueRecoveryActions).set({ id: holdId }).where(eq(issueRecoveryActions.id, f.actionId));
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, canReconcile: () => true, onError: vi.fn() });
|
||||
const worker = registerWorkspaceRecoveryWork(db, coordinator, f.environmentRuntime, async () => undefined);
|
||||
const busy = setInterval(() => { void db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery)).catch(() => undefined); }, 50);
|
||||
try {
|
||||
await worker.ready;
|
||||
await vi.waitFor(async () => expect((await action(holdId)).evidence.stopProofRecovery).toMatchObject({ rearms: 1 }),
|
||||
{ timeout: 20_000, interval: 250 });
|
||||
} finally {
|
||||
clearInterval(busy);
|
||||
await worker.stop();
|
||||
await coordinator.stop();
|
||||
}
|
||||
}, 60_000);
|
||||
|
||||
it("wakes the workspace recovery worker when a stop receipt is recorded", async () => {
|
||||
const f = await acceptedRemoteExport("retained");
|
||||
const woken: string[] = [];
|
||||
|
||||
@@ -3,7 +3,7 @@ import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { executionControlSchedule, registerExecutionControlWork } from "./execution-control-work.js";
|
||||
import type { EnvironmentRuntimeService } from "./environment-runtime.js";
|
||||
import { reconcileRetainedWorkspaces } from "./workspace-recovery.js";
|
||||
import { reconcileRetainedWorkspaces, type PassiveSweep } from "./workspace-recovery.js";
|
||||
|
||||
export function registerWorkspaceRecoveryWork(db: Db, coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
runtime: EnvironmentRuntimeService, onRecovered: Parameters<typeof reconcileRetainedWorkspaces>[2],
|
||||
@@ -12,15 +12,17 @@ export function registerWorkspaceRecoveryWork(db: Db, coordinator: ReturnType<ty
|
||||
let stopped = false;
|
||||
let scan: Promise<void> | undefined;
|
||||
let wakeWhileBusy = false;
|
||||
let passiveSweep: { cursor?: string } = {};
|
||||
const passiveSweep: PassiveSweep = {};
|
||||
const schedule = executionControlSchedule(db);
|
||||
const worker = registerExecutionControlWork(coordinator, [{
|
||||
queue: DELIVERY_QUEUES.workspaceRecovery,
|
||||
onSettled: () => { passiveSweep = {}; if (pending.size >= 2) wakeWhileBusy = true; },
|
||||
onSettled: () => { passiveSweep.restart = true; if (pending.size >= 2) wakeWhileBusy = true; },
|
||||
run: () => scan = (async () => {
|
||||
const sweep = passiveSweep, cursor = sweep.cursor;
|
||||
const work = await reconcileRetainedWorkspaces(db, runtime, onRecovered, true, continueSweep => {
|
||||
if (!stopped && (continueSweep || wakeWhileBusy)) { wakeWhileBusy = false; worker.wake(); }
|
||||
}, passiveSweep);
|
||||
}, sweep);
|
||||
if (sweep.restart && !sweep.complete && sweep.cursor !== cursor && !stopped) worker.wake();
|
||||
for (const job of work ?? []) {
|
||||
pending.add(job);
|
||||
const settled = () => { pending.delete(job); };
|
||||
|
||||
@@ -644,11 +644,13 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO
|
||||
)).orderBy(asc(issueRecoveryActions.id)).limit(5);
|
||||
}
|
||||
|
||||
export type PassiveSweep = { cursor?: string; complete?: boolean; restart?: boolean };
|
||||
|
||||
/** Bounded discovery shares the periodic recovery lifecycle. A repaired source
|
||||
* still owes cleanup even without saved input or after task reconciliation. */
|
||||
export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRuntimeService,
|
||||
onRecovered: (run: typeof heartbeatRuns.$inferSelect) => Promise<void>, obligationsOnly = false,
|
||||
onFinished?: (continueSweep: boolean) => void, passiveSweep?: { cursor?: string }) {
|
||||
onFinished?: (continueSweep: boolean) => void, passiveSweep?: PassiveSweep) {
|
||||
await adoptHistoricalWorkspaceRecoveries(db);
|
||||
let pending = workInFlight.get(db);
|
||||
if (!pending) { pending = new Set(); workInFlight.set(db, pending); }
|
||||
@@ -659,7 +661,14 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu
|
||||
let rows = await retainedWorkspaceCandidates(db, unfinishedCursors.get(db), true, false, inFlightRunIds);
|
||||
if (!rows.length) rows = await retainedWorkspaceCandidates(db, undefined, true, false, inFlightRunIds);
|
||||
const unfinished = rows.length > 0;
|
||||
if (!rows.length) rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds);
|
||||
if (!rows.length) {
|
||||
rows = await retainedWorkspaceCandidates(db, passiveSweep ? passiveSweep.cursor : cursors.get(db), false, obligationsOnly, inFlightRunIds);
|
||||
if (passiveSweep && !rows.length && passiveSweep.restart) {
|
||||
Object.assign(passiveSweep, { cursor: undefined, restart: false });
|
||||
rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds);
|
||||
}
|
||||
if (passiveSweep) passiveSweep.complete = !rows.length;
|
||||
}
|
||||
if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds);
|
||||
const work: Promise<void>[] = [];
|
||||
let hasUnvisited = false;
|
||||
|
||||
Reference in new issue
Block a user