diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 32a3074e2b0..c584ec07510 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2801,7 +2801,12 @@ take the same path after deploy. An exhausted hold whose run's files a sibling recovery action already saved settles as recovered. This includes a no-replay hold recorded resolved-but-blocked: settling it removes its replay block, so it stops blocking the task, and starts no turn itself; the run's own result and -any waiting input decide what continues. +any waiting input decide what continues. When that no-replay hold still owns +the task's Blocked status (its recorded failure block names the current status +version), the task returns to `todo`; a newer decision's Blocked status is kept. +A settled hold is never selected for recovery again. The worker handles several +holds of one run one after another in the same pass, so a hold that stays +selectable (for example while saved input waits) cannot starve a later one. An exhausted save whose last failure was `agent_stop_unconfirmed` is not a refusal of the saved files: its recovery window ran out before the agent's stop diff --git a/server/src/services/workspace-recovery-capture-policy.test.ts b/server/src/services/workspace-recovery-capture-policy.test.ts index 146a3896a3d..0cf7c81b8ad 100644 --- a/server/src/services/workspace-recovery-capture-policy.test.ts +++ b/server/src/services/workspace-recovery-capture-policy.test.ts @@ -3,7 +3,7 @@ import { mkdir, readFile, symlink, writeFile } 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, executionWorkspaces, heartbeatRuns, issueRecoveryActions } from "@paperclipai/db"; +import { activityLog, agentWakeupRequests, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues } from "@paperclipai/db"; import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { getExecutionBlocker } from "./execution-blocker.js"; @@ -110,7 +110,7 @@ describe("saves exhausted under an older workspace capture policy", () => { * Its six attempts failed the same way and it was recorded * resolved-but-blocked. That second hold is the task's execution blocker. */ - async function stagingIncident() { + async function stagingIncident(failureBlock: boolean) { const f = await exhaustedCaptureHold(); const exportHold = await action(f.actionId); const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }; @@ -129,6 +129,7 @@ describe("saves exhausted under an older workspace capture policy", () => { historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId, leaseIds: [f.lease.id] } } }) .where(eq(issueRecoveryActions.id, f.actionId)); const exhausted = readSystemRecovery((await action(f.actionId)).evidence)!; + const [task] = await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)).returning(); const noReplayHoldId = randomUUID(), resolvedAt = new Date(Date.now() - 30 * 60_000); await db.insert(issueRecoveryActions).values({ id: noReplayHoldId, companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", status: "resolved", outcome: "blocked", ownerType: "board", resolvedAt, @@ -140,12 +141,21 @@ describe("saves exhausted under an older workspace capture policy", () => { workspaceRecovery: exportHold.evidence.workspaceRecovery, originalFailureCode: "native_workspace_sync_out_retry_exhausted", continuationDelivery: "invalidated", systemRecoveryActiveAt: Date.now() - 40 * 60_000, workspaceRestoreFailure: "restore_failed", - workspaceRestoreRecovery: recovery, previousWorkspaceRecovery: null } }); - return { ...f, noReplayHoldId }; + workspaceRestoreRecovery: recovery, previousWorkspaceRecovery: null, + // Rows of the same release that recorded the Blocked status their run's + // failure set, and the historical source an older adoption named. + ...(failureBlock ? { nativeFailureBlock: { runId: f.runId, statusVersion: Number(task.statusVersion) }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId, leaseIds: [f.lease.id] } } : {}) } }); + // The user's "try again" comment waits for the task to be free. + const [waiting] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: source.agentId, source: "on_demand", + reason: "issue_commented", status: "deferred_issue_execution", requestedByActorType: "user", requestedByActorId: "operator", + payload: { issueId: f.issueId } }).returning(); + return { ...f, noReplayHoldId, waitingInputId: waiting.id }; } - it("settles the staging incident's export hold and its resolved no-replay sibling without a person", async () => { - const f = await stagingIncident(); + it.each([["as recorded on staging", false], ["that owns the task's Blocked status", true]] as const)( + "settles the export hold and a resolved no-replay sibling %s without a person", async (_shape, failureBlock) => { + const f = await stagingIncident(failureBlock); // As on staging: the no-replay sibling is the execution blocker, and it // offers no task Retry. expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: f.noReplayHoldId, @@ -154,13 +164,17 @@ describe("saves exhausted under an older workspace capture policy", () => { // The recovery worker's own discovery, as after a deploy: it runs while // work is due and goes idle once nothing is. const blockersSeenByContinuation: unknown[] = []; - const onRecovered = vi.fn(async () => { blockersSeenByContinuation.push(await getExecutionBlocker(db, f.companyId, f.issueId)); }); + // Normal dispatch consumes the saved input once nothing blocks the task. + const onRecovered = vi.fn(async () => { + const blocker = await getExecutionBlocker(db, f.companyId, f.issueId); + blockersSeenByContinuation.push(blocker); + if (!blocker) await db.update(agentWakeupRequests).set({ status: "queued" }).where(eq(agentWakeupRequests.id, f.waitingInputId)); + }); let passes = 0; for (; passes < 6; passes += 1) { const due = await nextWorkspaceRecoveryAt(db); if (due === null || due > Date.now()) break; - await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []); - } + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []); } expect(passes).toBeGreaterThan(0); expect(passes).toBeLessThan(6); @@ -182,11 +196,23 @@ describe("saves exhausted under an older workspace capture policy", () => { // The continuation hook ran once the task was free, so saved input is dispatched. expect(blockersSeenByContinuation.at(-1)).toBeNull(); expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).map(row => row.id)).toEqual([f.runId]); + expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, f.waitingInputId)))[0]!.status).toBe("queued"); + // A hold that owned the Blocked status releases it; otherwise the status is + // left to the run's own result finalization. + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + if (failureBlock) expect(task!.status).toBe("todo"); // Bounded: one re-arm, and the worker has nothing left to do. const next = await nextWorkspaceRecoveryAt(db); expect(next === null || next > Date.now()).toBe(true); expect((await db.select().from(activityLog).where(eq(activityLog.action, "issue.workspace_recovery_rearmed"))) .filter(row => row.entityId === f.issueId)).toHaveLength(1); + // A later pass, even with new saved input, never selects or hands off the settled hold again. + await db.update(agentWakeupRequests).set({ status: "deferred_issue_execution" }).where(eq(agentWakeupRequests.id, f.waitingInputId)); + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.noReplayHoldId); + expect((await retainedWorkspaceCandidates(db)).map(row => row.action.id)).not.toContain(f.noReplayHoldId); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []); + expect(await action(f.noReplayHoldId)).toEqual(sibling); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); }, 90_000); it("does not re-arm a save that the current policy already exhausted", async () => { diff --git a/server/src/services/workspace-recovery-fallback.test.ts b/server/src/services/workspace-recovery-fallback.test.ts index 13a95cff050..762e0b441d4 100644 --- a/server/src/services/workspace-recovery-fallback.test.ts +++ b/server/src/services/workspace-recovery-fallback.test.ts @@ -465,10 +465,13 @@ describe("retained source fallback", () => { expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); await db.update(issueRecoveryActions).set({ status: "resolved", resolvedAt: new Date() }).where(eq(issueRecoveryActions.id, f.actionId)); const staleId = randomUUID(), now = Date.now(); + // An export hold that names the task's Blocked status leaves the status to + // its saved result's finalization; the sibling settlement must not change it. + const [task] = await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, f.issueId)).returning(); await db.insert(issueRecoveryActions).values({ id: staleId, companyId: f.companyId, sourceIssueId: f.issueId, kind: "active_run_watchdog", status: "active", ownerType: "board", cause: "native_workspace_sync_out_retry_exhausted", fingerprint: randomUUID(), nextAction: "Automatic recovery of saved work stopped after 6 attempts.", - evidence: { runId: f.runId, + evidence: { runId: f.runId, nativeFailureBlock: { runId: f.runId, statusVersion: Number(task.statusVersion) }, systemRecovery: { schema: "paperclip.system-recovery.v1", policyVersion: 1, sourceRunId: f.runId, revision: 54, generationId: randomUUID(), intentId: null, phase: "preserve_files", state: "exhausted", attemptId: randomUUID(), attempts: 6, retriesUsed: 5, firstFailureAt: now - 60_000, nextAttemptAt: null, deadlineAt: now - 1, providerCooldownAt: null, @@ -488,6 +491,7 @@ describe("retained source fallback", () => { const live = await db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), inArray(issueRecoveryActions.status, ["active", "escalated"]))); expect(live).toHaveLength(0); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.status).toBe("blocked"); }, 60_000); it("stops discovering a stale hold whose sibling receipt no longer verifies", async () => { diff --git a/server/src/services/workspace-recovery-fallback.ts b/server/src/services/workspace-recovery-fallback.ts index ffec097f372..9c1844dcc5f 100644 --- a/server/src/services/workspace-recovery-fallback.ts +++ b/server/src/services/workspace-recovery-fallback.ts @@ -26,7 +26,7 @@ import { abandonUnverifiedRemoteWorkspaceEpoch } from "./remote-workspace-epoch- import { hasStopOnlyCleanup, prepareHistoricalSandboxStopVerification, prepareSandboxStopAndRetain, restoreSandboxStopAfterFailedRestart, settleStopOnlyCleanup } from "./sandbox-stop-and-retain.js"; import { nativeWorkspaceFinalizationBlockSchema } from "./native-runtime/preserved-native-result-finalization.js"; import { withNativeWorkspaceFinalizationOwnership } from "./native-runtime/native-workspace-finalization-ownership.js"; -import { PERMANENT_SOURCE_RETRY_CLASSES, RETAINED_SOURCE_FALLBACK_CHECK_KEY, UNVERIFIABLE_SOURCE_CODES, isRetainedFileHold, +import { NATIVE_NO_REPLAY_HOLD_CAUSE, PERMANENT_SOURCE_RETRY_CLASSES, RETAINED_SOURCE_FALLBACK_CHECK_KEY, UNVERIFIABLE_SOURCE_CODES, isRetainedFileHold, retainedHoldFallbackEligible } from "./workspace-recovery-fallback-discovery.js"; import { captureReadOnlyWorkspaceArchive, sandboxProviderFailureLabel } from "./remote-workspace-transport.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; @@ -303,6 +303,27 @@ async function confirmSourceAllocationStopped(db: Db, input: Source & { environm * the recovery delivery starts one continuation. Bounded and idempotent: a * settled hold is never revisited, and a lost race leaves the hold unchanged. */ +/** + * Whether the task's Blocked status still belongs to this hold: its own + * receipt names the current status version or, for holds saved before + * receipts existed, this run's terminal failure set the block and nothing + * changed it since. Otherwise a newer decision owns the block. + */ +function holdOwnsBlockedStatus(action: typeof issueRecoveryActions.$inferSelect, issue: typeof issues.$inferSelect, + run: typeof heartbeatRuns.$inferSelect) { + const block = nativeWorkspaceFinalizationBlockSchema.safeParse(action.evidence.workspaceFinalizationBlock); + // A native no-replay hold records the Blocked status version its run's failure set. + const failureBlock = object(action.evidence.nativeFailureBlock); + const blockedAt = issue.blockedTransitionAt?.getTime(); + return issue.status === "blocked" && (block.success + ? block.data.issueId === issue.id && block.data.runId === run.id + && block.data.statusVersion === Number(issue.statusVersion) && block.data.priorDecisionId === issue.lastStatusDecisionId + : action.evidence.nativeFailureBlock !== undefined + ? failureBlock.runId === run.id && failureBlock.statusVersion === Number(issue.statusVersion) + : action.evidence.workspaceFinalizationBlock === undefined && issue.lastStatusDecisionId === null && blockedAt !== undefined + && run.finishedAt !== null && blockedAt >= run.finishedAt.getTime() - 1_000 && blockedAt <= action.createdAt.getTime() + 60_000); +} + export async function continueFromLastSavedWorkspace(input: Source & { db: Db; environmentRuntime: EnvironmentRuntimeService; ownership: Ownership; }): Promise { @@ -346,17 +367,7 @@ export async function continueFromLastSavedWorkspace(input: Source & { // holds saved before receipts existed, a block set by this run's terminal // failure that nothing changed since) still names it. Otherwise a newer // decision owns the block: it is kept and no continuation is started. - const block = nativeWorkspaceFinalizationBlockSchema.safeParse(action.evidence.workspaceFinalizationBlock); - // A native no-replay hold records the Blocked status version its run's failure set. - const failureBlock = object(action.evidence.nativeFailureBlock); - const blockedAt = issue.blockedTransitionAt?.getTime(); - const ownsBlock = issue.status === "blocked" && (block.success - ? block.data.issueId === issue.id && block.data.runId === run.id - && block.data.statusVersion === Number(issue.statusVersion) && block.data.priorDecisionId === issue.lastStatusDecisionId - : action.evidence.nativeFailureBlock !== undefined - ? failureBlock.runId === run.id && failureBlock.statusVersion === Number(issue.statusVersion) - : action.evidence.workspaceFinalizationBlock === undefined && issue.lastStatusDecisionId === null && blockedAt !== undefined - && blockedAt >= run.finishedAt!.getTime() - 1_000 && blockedAt <= action.createdAt.getTime() + 60_000); + const ownsBlock = holdOwnsBlockedStatus(action, issue, run); const dispatchable = issue.status !== "blocked" || ownsBlock; const userText = quarantine.state === "captured" ? "Paperclip could not verify the previous run's unsaved files, so the task continues from the last saved version. Those files are attached to the task, unapplied." @@ -460,7 +471,7 @@ export async function settleExhaustedRetainedHold(input: Source & { db: Db; envi */ export async function settleSiblingRecoveredHold(db: Db, input: Source) { const settled = await db.transaction(async tx => { - const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId), + const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId))).for("update"); @@ -468,10 +479,13 @@ export async function settleSiblingRecoveredHold(db: Db, input: Source) { eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).for("update"); if (!issue || !run || !action || action.evidence.runId !== run.id) return null; if (object(action.evidence.siblingWorkspaceRecovery).schema === SIBLING_WORKSPACE_RECOVERY_SCHEMA) return "already" as const; + // isRetainedFileHold also accepts export holds; only the no-replay cause + // may release the status here. An export hold leaves it to its result. + const noReplayHold = action.cause === NATIVE_NO_REPLAY_HOLD_CAUSE && isRetainedFileHold(action); const system = readSystemRecovery(action.evidence); // A no-replay hold is recorded resolved-but-blocked. It blocks the task // until its run's files are saved, so a sibling's save settles it too. - if ((!["active", "escalated"].includes(action.status) && !isRetainedFileHold(action)) + if ((!["active", "escalated"].includes(action.status) && !noReplayHold) || system?.state !== "exhausted" || system.sourceRunId !== run.id) return null; const receiptId = await verifiedWorkspaceRepairReceipt(tx as unknown as Db, run); const [receipt] = receiptId ? await tx.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and( @@ -498,6 +512,18 @@ export async function settleSiblingRecoveredHold(db: Db, input: Source) { siblingWorkspaceRecovery: { schema: SIBLING_WORKSPACE_RECOVERY_SCHEMA, receiptId, recoveryActionId: siblingActionId, recordedAt: now.toISOString() } }, }).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, action.id))); + // A no-replay hold that still owns the task's Blocked status releases it, + // so saved input and normal wakes continue the task. An export hold's + // status stays with its own result finalization. A newer decision's + // Blocked status is kept. + if (noReplayHold && holdOwnsBlockedStatus(action, issue, run)) { + await tx.update(issues).set({ status: "todo", unblockDescriptor: null, blockedTransitionAt: null, + blockedOwnerNotifiedAt: null, updatedAt: now }).where(eq(issues.id, issue.id)); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.issueMonitors); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.agentTimers); + await notifyStatusCardIssueChange(tx, input.companyId); + await notifyRoutineExecutionStatus(tx, issue.id, input.companyId); + } await notifyDeliveryWork(tx, DELIVERY_QUEUES.executionContinuation); return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery", action: "issue.workspace_recovered", entityType: "issue", entityId: issue.id, runId: run.id, diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index f35a69fb599..19dc219c2a1 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -26,13 +26,15 @@ export function workspaceSourceBindingFailure(reason: WorkspaceSourceVerificatio export class WorkspaceSourceInspectionLimit extends Error {} +// A resolved hold that a sibling's save settled has nothing left to recover, +// so it is never selected again, even when it names a historical source. export function workspaceRecoverySourceEligible(includeCompleted = false) { const actionEligible = sql`(${issueRecoveryActions.status} in ('active', 'escalated') or (${issueRecoveryActions.status} = 'cancelled' and ${inArray(issueRecoveryActions.cause, workspaceExportCauses)} and jsonb_typeof(${issueRecoveryActions.evidence}#>'{workspaceRecovery,requestIds}') = 'array' and ${issueRecoveryActions.evidence}#>'{workspaceRecovery,requestIds}' <> '[]'::jsonb) or - (${issueRecoveryActions.status} = 'resolved' and ( + (${issueRecoveryActions.status} = 'resolved' and ${issueRecoveryActions.evidence}->'siblingWorkspaceRecovery' is null and ( (${issueRecoveryActions.evidence}#>>'{automaticRecovery,policy}' = 'preserve_without_replay_v1' and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,replay}' = 'blocked' and ${issueRecoveryActions.evidence}#>>'{automaticRecovery,runId}' = ${heartbeatRuns.id}::text) or diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index fc0078036d0..101af1e339d 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -663,16 +663,36 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu if (!rows.length && !passiveSweep) rows = await retainedWorkspaceCandidates(db, undefined, false, obligationsOnly, inFlightRunIds); const work: Promise[] = []; let hasUnvisited = false; + const advance = (actionId: string) => { + if (unfinished) unfinishedCursors.set(db, actionId); + else if (passiveSweep) passiveSweep.cursor = actionId; + else cursors.set(db, actionId); + }; + // Several actions of one run (an export hold and a later no-replay hold) + // run one after another in that run's job. Skipping the later one would + // starve it while the earlier one stays selectable (for example while saved + // input waits for the later one to settle). + const groups = new Map(); for (const { action, run } of rows) { + const group = groups.get(run.id); + if (group) { group.actions.push(action); advance(action.id); continue; } if (pending.size >= 2) { hasUnvisited = true; break; } - if (unfinished) unfinishedCursors.set(db, action.id); - else if (passiveSweep) passiveSweep.cursor = action.id; - else cursors.set(db, action.id); + advance(action.id); if (pending.has(run.id)) continue; pending.add(run.id); + groups.set(run.id, { run, actions: [action] }); + } + for (const { run, actions } of groups.values()) { const finish = beginIdleTrackedWork(); - const physical = ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? ""); + const physical = actions.some(action => ["scheduled", "running", "awaiting_evidence"].includes(readSystemRecovery(action.evidence)?.state ?? "")); work.push((async () => { try { + for (const action of actions) await recoverRunAction(action, run); + } finally { pending.delete(run.id); finish(); onFinished?.(physical || hasUnvisited); } })()); + } + return work; + + async function recoverRunAction(action: typeof issueRecoveryActions.$inferSelect, run: typeof heartbeatRuns.$inferSelect) { + try { await recoverRetainedWorkspace({ db, environmentRuntime: runtime, companyId: action.companyId, issueId: action.sourceIssueId, runId: run.id, actionId: action.id }); const [updated] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id)); const recovered = readWorkspaceRecovery(updated?.evidence ?? null)?.status === "recovered"; @@ -691,7 +711,5 @@ export async function reconcileRetainedWorkspaces(db: Db, runtime: EnvironmentRu } } } catch { logger.warn({ runId: run.id }, "Retained workspace remains pending"); } - finally { pending.delete(run.id); finish(); onFinished?.(physical || hasUnvisited); } })()); } - return work; }