diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e683dead6d6..4e604ccd82b 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -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 diff --git a/server/src/services/workspace-recovery-stop-proof.test.ts b/server/src/services/workspace-recovery-stop-proof.test.ts index e3ece5d881c..3546d1c845d 100644 --- a/server/src/services/workspace-recovery-stop-proof.test.ts +++ b/server/src/services/workspace-recovery-stop-proof.test.ts @@ -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()); +vi.mock("./execution-control-deadline.js", async importOriginal => ({ + ...await importOriginal(), EXECUTION_RECONCILIATION_INTERVAL_MS: 200 })); vi.mock("./environment-execution-target.js", async importOriginal => { const actual = await importOriginal(); return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[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(); + + 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[] = []; diff --git a/server/src/services/workspace-recovery-work.ts b/server/src/services/workspace-recovery-work.ts index dea2bf28b46..279fa77cc00 100644 --- a/server/src/services/workspace-recovery-work.ts +++ b/server/src/services/workspace-recovery-work.ts @@ -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, runtime: EnvironmentRuntimeService, onRecovered: Parameters[2], @@ -12,15 +12,17 @@ export function registerWorkspaceRecoveryWork(db: Db, coordinator: ReturnType | 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); }; diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 101af1e339d..6c8712a031a 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -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, 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[] = []; let hasUnvisited = false;