diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 12197298fd0..c479e918c3f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2536,6 +2536,30 @@ recovery worker discovers these holds itself, so holds saved by earlier releases take the same path after deploy. An exhausted hold whose run's files a sibling recovery action already saved settles as recovered. +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 +could be proven. This applies to every workspace recovery the worker owns, +export holds and restore holds alike. Once the source run is provably over (its +run is terminal, no local execution remains, and neither its controller lease +nor its finalizer lease is live), the recovery worker re-arms it with a new +bounded system generation. A live lease defers the re-arm to that lease's +expiry; the worker is not woken earlier. The new generation repeats the strict +stop proof (a provider stop receipt for every allocation of the run) and the +save, then settles the hold exactly like any completed recovery, so other tasks +on the workspace can start. Re-arms are bounded: at most four, the first at once +and the next ones 30 minutes, 2 hours and 8 hours after the previous re-arm, +each recorded in the action's `stopProofRecovery` marker and as +`issue.workspace_recovery_rearmed` system activity. When every re-arm of an +export hold is spent and the stop still cannot be proven, the hold takes the +retained-source fallback above: the unsaved files are attached to the task, +unapplied, and the task continues from the last save. A fallback whose bounded +checks are spent is reconsidered once whenever an allocation of the source run +changes afterwards, for example when a late provider stop receipt arrives. +While a re-arm remains, the hold names the wait and says that Paperclip tries +again automatically; after that it names Retry. Holds exhausted by earlier +releases carry no marker and are re-armed on the worker's first pass after +deploy, with no migration. + An exhausted workspace export records the exact Blocked status version it creates, bound to its source run, result, contract and identity. Recovery may restore the accepted passive wait only while that receipt still owns the task @@ -3397,7 +3421,8 @@ Recorded system recovery failures retain their original attempt and intent. When system workspace recovery exhausts its bounded attempts or deadline, the recovery action stays active and its owner changes from system to board. Its next action names the attempt count and last failure code and tells the board to -use **Retry recovery**. The task blocker shows that same text. Retry starts one +use **Retry recovery**. An unproven agent stop (`agent_stop_unconfirmed`) instead +names the automatic re-arm described above; Retry remains available. The task blocker shows that same text. Retry starts one new bounded system generation and returns ownership to the system. A sweep that finds an action already exhausted and still system-owned applies the same handoff as bookkeeping only; it grants no attempt. Exhaustion never submits a provider diff --git a/server/src/services/issue-recovery-actions.ts b/server/src/services/issue-recovery-actions.ts index bddd3c5ec96..df53a07525e 100644 --- a/server/src/services/issue-recovery-actions.ts +++ b/server/src/services/issue-recovery-actions.ts @@ -93,7 +93,7 @@ export type ResolveIssueRecoveryActionInput = { function readModelNextAction(row: IssueRecoveryActionRow): string { const system = readSystemRecovery(row.evidence); return system?.state === "exhausted" && row.evidence?.workspaceRecovery - ? exhaustedSystemRecoveryAction(system) : row.nextAction; + ? exhaustedSystemRecoveryAction(system, row.evidence) : row.nextAction; } function toReadModel(row: IssueRecoveryActionRow): IssueRecoveryAction { diff --git a/server/src/services/sandbox-stop-and-retain.ts b/server/src/services/sandbox-stop-and-retain.ts index 27e4122b3cd..fe0c2b1ee99 100644 --- a/server/src/services/sandbox-stop-and-retain.ts +++ b/server/src/services/sandbox-stop-and-retain.ts @@ -1,5 +1,6 @@ import { notifyRetainedNativeSessionCleanup } from "./retained-native-session-cleanup-notifications.js"; import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { randomUUID } from "node:crypto"; import { and, eq, sql, type SQL } from "drizzle-orm"; import { environmentLeases, heartbeatRuns, type Db } from "@paperclipai/db"; @@ -198,7 +199,11 @@ export async function settleStopOnlyCleanup(db: Db, lease: Lease, options: { att ? { [ORPHANED_SANDBOX_RECOVERY_KEY]: { recoveredOrphanedRunLease: true, ownerProof: recordedIntent.ownerProof ?? null, runId: lease.heartbeatRunId, recoveredAt: now.toISOString(), fromStopIntent: true } } : {}; const [updated] = await writeEnvironmentCleanup(db, async tx => { - if (stopped) await notifyRetainedNativeSessionCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId! }); + if (stopped) { + await notifyRetainedNativeSessionCleanup(tx, { companyId: lease.companyId, runId: lease.heartbeatRunId! }); + // A stop receipt is new evidence for a save hold waiting on it. + await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); + } return tx.update(environmentLeases).set({ status: stopped ? "released" : "pending_cleanup", cleanupStatus: stopped ? "success" : "failed", failureReason: stopped ? null : "sandbox_stop_pending", diff --git a/server/src/services/system-recovery-state.test.ts b/server/src/services/system-recovery-state.test.ts index 0a99d42720b..f57f28cac11 100644 --- a/server/src/services/system-recovery-state.test.ts +++ b/server/src/services/system-recovery-state.test.ts @@ -82,6 +82,17 @@ describe("durable system recovery budget", () => { expect(exhaustedSystemRecoveryAction({ ...once, attempts: 6 })).toContain("stopped after 6 attempts"); }); + it("promises an automatic retry for an unproven stop only while a re-arm remains", () => { + const stop = { ...failSystemRecovery(reserveSystemRecoveryAttempt(createSystemRecovery(randomUUID()), 0)!, + new SystemRecoveryFailure("agent_stop_unconfirmed", "recheck_evidence"), { now: 0 }), state: "exhausted" as const }; + const marker = (rearms: number) => ({ stopProofRecovery: { schema: "paperclip.stop-proof-recovery.v1", rearms, nextRearmAt: null } }); + expect(exhaustedSystemRecoveryAction(stop, {})).toMatch(/tries again automatically/); + expect(exhaustedSystemRecoveryAction(stop, marker(3))).toMatch(/tries again automatically/); + // Spent re-arms, or no evidence to judge them, keep the Retry text. + expect(exhaustedSystemRecoveryAction(stop, marker(4))).toContain("Retry to try again"); + expect(exhaustedSystemRecoveryAction(stop)).toContain("Retry to try again"); + }); + it("names the last failure and its cause in the board handoff", () => { const state = reserveSystemRecoveryAttempt(createSystemRecovery(randomUUID()), 0)!; const withCause = failSystemRecovery(state, new SystemRecoveryFailure("provider_unavailable", "retry_transient"), diff --git a/server/src/services/system-recovery-state.ts b/server/src/services/system-recovery-state.ts index a70b6ed6c0c..abf58eca9a9 100644 --- a/server/src/services/system-recovery-state.ts +++ b/server/src/services/system-recovery-state.ts @@ -127,13 +127,46 @@ export function failSystemRecovery( }; } +/** + * An exhausted save whose last failure was an unproven agent stop is not a + * permanent refusal: once the source run has ended, a later attempt can prove + * the stop. Recovery re-arms it a bounded number of times on a backoff + * (workspace-recovery-stop-proof), then an export hold continues from the last + * save with the unsaved files attached, like a refused source. + */ +export const STOP_PROOF_RECOVERY_KEY = "stopProofRecovery"; +export const STOP_PROOF_RECOVERY_SCHEMA = "paperclip.stop-proof-recovery.v1"; +export const STOP_PROOF_MAX_REARMS = 4; +export const STOP_PROOF_WAIT_CODE = "agent_stop_unconfirmed"; + +export function readStopProofRecovery(evidence: Record | null | undefined): { rearms: number; nextRearmAt: number | null } | null { + const marker = evidence?.[STOP_PROOF_RECOVERY_KEY] as Record | undefined; + if (!marker || marker.schema !== STOP_PROOF_RECOVERY_SCHEMA || typeof marker.rearms !== "number") return null; + return { rearms: marker.rearms, nextRearmAt: typeof marker.nextRearmAt === "number" ? marker.nextRearmAt : null }; +} + +export function isExhaustedStopProofWait(state: Pick | null | undefined) { + return state?.state === "exhausted" && state.failure?.code === STOP_PROOF_WAIT_CODE + && state.failure.retryClass === "recheck_evidence"; +} + +/** Every bounded re-arm of an unproven stop is spent. */ +export function stopProofRearmsSpent(evidence: Record | null | undefined) { + return (readStopProofRecovery(evidence)?.rearms ?? 0) >= STOP_PROOF_MAX_REARMS; +} + /** User-facing text for a spent recovery budget. It names what happens next and * never asks a person to inspect a sandbox or environment. The code stays for support. */ -export function exhaustedSystemRecoveryAction(state: SystemRecoveryState): string { +export function exhaustedSystemRecoveryAction(state: SystemRecoveryState, evidence?: Record | null): string { // Another Retry cannot release a copyback owner that stopped without reporting. if (state.failure?.code === "copyback_stop_unconfirmed") return "Recovery of saved work is waiting for the previous server's file copy to stop. " + "Recovery resumes once Paperclip can confirm that it stopped. No provider turn is submitted."; + // An unproven agent stop is re-proven automatically once the run is over + // (workspace-recovery-stop-proof), while a bounded re-arm remains. + if (evidence !== undefined && isExhaustedStopProofWait(state) && !stopProofRearmsSpent(evidence)) + return "Recovery of saved work is waiting to confirm that the previous run stopped. " + + "Paperclip tries again automatically, and the agent will not repeat its work."; const cause = state.failure?.cause ? `, cause: ${state.failure.cause.label}` : ""; if (isDeterministicRecoveryRefusal(state)) { const code = state.failure!.code; diff --git a/server/src/services/system-recovery.ts b/server/src/services/system-recovery.ts index adade4b9da6..020f2354d9e 100644 --- a/server/src/services/system-recovery.ts +++ b/server/src/services/system-recovery.ts @@ -5,7 +5,8 @@ import { writeExecutionControl } from "./execution-control-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { createSystemRecovery, exhaustedSystemRecoveryAction, readSystemRecovery, type SystemRecoveryState } from "./system-recovery-state.js"; import { workspaceRecoverySourceEligible, type WorkspaceRecoverySourceIdentity } from "./workspace-recovery-source.js"; -import { retainedSourceFallbackDueAt, retainedSourceFallbackPendingCondition } from "./workspace-recovery-fallback-discovery.js"; +import { retainedSourceFallbackDueAt, retainedSourceFallbackPendingCondition, stopProofRearmDueAt, + stopProofRearmPendingCondition } from "./workspace-recovery-fallback-discovery.js"; export async function adoptSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) { return db.transaction(async tx => { @@ -59,7 +60,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour // A spent budget is not system work any more. Hand it to the board with the // one action that can progress it, instead of leaving an inert system owner. const exhausted = next.state === "exhausted"; - const handoff = exhausted ? exhaustedSystemRecoveryAction(next) : null; + const handoff = exhausted ? exhaustedSystemRecoveryAction(next, await loadActionEvidence(db, source)) : null; const compatibility = { schema: "paperclip.workspace-recovery.v1", runId: source.runId, status, stage: next.phase === "verify_execution" ? "verifying_stop" : next.phase === "preserve_files" ? "restoring_files" : "saving_receipts", @@ -90,7 +91,7 @@ export async function updateSystemRecovery(db: Db, source: WorkspaceRecoverySour export async function handOffExhaustedSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity, state: SystemRecoveryState) { if (state.state !== "exhausted") return false; - const handoff = exhaustedSystemRecoveryAction(state); + const handoff = exhaustedSystemRecoveryAction(state, await loadActionEvidence(db, source)); const [updated] = await db.update(issueRecoveryActions).set({ ownerType: "board", ownerAgentId: null, ownerUserId: null, nextAction: handoff, evidence: sql`jsonb_set(${issueRecoveryActions.evidence}, '{workspaceRecovery,message}', ${JSON.stringify(handoff)}::jsonb)`, @@ -105,6 +106,12 @@ export async function handOffExhaustedSystemRecovery(db: Db, source: WorkspaceRe return Boolean(updated); } +async function loadActionEvidence(db: Db, source: WorkspaceRecoverySourceIdentity) { + const [row] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.id, source.actionId))); + return row?.evidence ?? null; +} + export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySourceIdentity) { const [row] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, source.companyId), eq(issueRecoveryActions.sourceIssueId, source.issueId), @@ -115,6 +122,7 @@ export async function loadSystemRecovery(db: Db, source: WorkspaceRecoverySource export async function nextWorkspaceRecoveryAt(db: Db): Promise { const rows = await db.select({ at: sql`case when ${retainedSourceFallbackPendingCondition()} then ${retainedSourceFallbackDueAt()} + when ${stopProofRearmPendingCondition()} then ${stopProofRearmDueAt()} when jsonb_typeof(${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}') = 'number' then (${issueRecoveryActions.evidence}#>>'{systemRecovery,nextAttemptAt}')::bigint else 0 end` }) .from(issueRecoveryActions).innerJoin(heartbeatRuns, and( @@ -126,6 +134,7 @@ export async function nextWorkspaceRecoveryAt(db: Db): Promise { workspaceRecoverySourceEligible(true), sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence') or ${retainedSourceFallbackPendingCondition()} + or ${stopProofRearmPendingCondition()} or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and ${issueRecoveryActions.evidence}->>'bootstrapPreparationSettledAt' is not null and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') = 'pending' diff --git a/server/src/services/workspace-recovery-fallback-discovery.ts b/server/src/services/workspace-recovery-fallback-discovery.ts index 35d54b28451..23cc1e26f51 100644 --- a/server/src/services/workspace-recovery-fallback-discovery.ts +++ b/server/src/services/workspace-recovery-fallback-discovery.ts @@ -1,13 +1,72 @@ import { sql } from "drizzle-orm"; import { issueRecoveryActions } from "@paperclipai/db"; +import { isExhaustedStopProofWait, STOP_PROOF_MAX_REARMS, STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, STOP_PROOF_WAIT_CODE, + stopProofRearmsSpent, type SystemRecoveryState } from "./system-recovery-state.js"; /** Source refusals that repeat on every retry (see workspace-recovery-fallback). */ export const UNVERIFIABLE_SOURCE_CODES = ["allocation_ownership_ambiguous", "source_binding_invalid", "source_unavailable", "historical_source_missing"] as const; export const PERMANENT_SOURCE_RETRY_CLASSES = ["service_required", "use_alternative_strategy"] as const; export const RETAINED_SOURCE_FALLBACK_CHECK_KEY = "retainedSourceFallbackCheck"; +/** Whether an exhausted hold may continue from the last save: its source is + * refused permanently, or its stop stayed unproven through every re-arm. */ +export function retainedHoldFallbackEligible(evidence: Record | null | undefined, state: SystemRecoveryState | null) { + if (!state || state.state !== "exhausted") return false; + const failure = state.failure; + if (failure && (UNVERIFIABLE_SOURCE_CODES as readonly string[]).includes(failure.code) + && (PERMANENT_SOURCE_RETRY_CLASSES as readonly string[]).includes(failure.retryClass)) return true; + return isExhaustedStopProofWait(state) && stopProofRearmsSpent(evidence); +} + +const stopProofMarker = sql`${issueRecoveryActions.evidence}->${STOP_PROOF_RECOVERY_KEY}`; +const stopProofRearms = sql`case when ${stopProofMarker}->>'schema' = ${STOP_PROOF_RECOVERY_SCHEMA} + and jsonb_typeof(${stopProofMarker}->'rearms') = 'number' then (${stopProofMarker}->>'rearms')::int else 0 end`; +const exhaustedStopProofWait = sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'exhausted' + and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,code}' = ${STOP_PROOF_WAIT_CODE} + and ${issueRecoveryActions.evidence}#>>'{systemRecovery,failure,retryClass}' = 'recheck_evidence')`; + +/** An exhausted unproven-stop hold whose next bounded re-arm is pending. Holds + * saved by older releases carry no marker and are due immediately. A finalizer + * owner whose lease is not live waits for the abandoned-owner settlement (the + * attempt would refuse it, and the passive sweep revisits the hold after it); + * a live one makes the re-arm due at its expiry (stopProofRearmDueAt). */ +export function stopProofRearmPendingCondition() { + return sql`(${issueRecoveryActions.status} in ('active', 'escalated') + and ${exhaustedStopProofWait} + and not exists (select 1 from native_run_finalizations finalizer + where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id::text = ${issueRecoveryActions.evidence}->>'runId' + and finalizer.lease_owner is not null and coalesce(finalizer.lease_expires_at <= now(), true)) + and ${issueRecoveryActions.evidence}->'retainedSourceFallback' is null + and ${issueRecoveryActions.evidence}->'executionReconciliation' is null + and ${stopProofRearms} < ${STOP_PROOF_MAX_REARMS})`; +} + +/** When the next re-arm is due, in epoch milliseconds (0 = now): its backoff, + * and no earlier than the expiry of a controller or finalizer lease that the + * source run still holds, so a held run is checked at that deadline only. */ +export function stopProofRearmDueAt() { + return sql`greatest(case when ${stopProofMarker}->>'schema' = ${STOP_PROOF_RECOVERY_SCHEMA} + and jsonb_typeof(${stopProofMarker}->'nextRearmAt') = 'number' then (${stopProofMarker}->>'nextRearmAt')::bigint else 0 end, + coalesce((select (extract(epoch from source.controller_lease_expires_at) * 1000)::bigint from heartbeat_runs source + where source.company_id = ${issueRecoveryActions.companyId} and source.id::text = ${issueRecoveryActions.evidence}->>'runId'), 0), + coalesce((select max((extract(epoch from finalizer.lease_expires_at) * 1000)::bigint) from native_run_finalizations finalizer + where finalizer.company_id = ${issueRecoveryActions.companyId} and finalizer.run_id::text = ${issueRecoveryActions.evidence}->>'runId' + and finalizer.lease_owner is not null), 0))`; +} + +export function stopProofRearmDueCondition(now: number) { + return sql`(${stopProofRearmPendingCondition()} and ${stopProofRearmDueAt()} <= ${now}::bigint)`; +} + const list = (values: readonly string[]) => sql.join(values.map(value => sql`${value}`), sql`, `); const check = sql`${issueRecoveryActions.evidence}->${RETAINED_SOURCE_FALLBACK_CHECK_KEY}`; +// New stop evidence (any change to an allocation of the source run, such as a +// late provider stop receipt) after the last deferred check reconsiders the +// hold once, even after its bounded checks are spent. +const leaseChangedSinceCheck = sql`(jsonb_typeof(${check}->'checkedAt') = 'number' and exists ( + select 1 from environment_leases lease where lease.company_id = ${issueRecoveryActions.companyId} + and lease.heartbeat_run_id::text = ${issueRecoveryActions.evidence}->>'runId' + and lease.updated_at > to_timestamp((${check}->>'checkedAt')::bigint / 1000.0)))`; /** * An exhausted hold that the fallback may still settle: its source is refused @@ -27,6 +86,11 @@ export function retainedSourceFallbackPendingCondition() { and exists (select 1 from heartbeat_runs source where source.company_id = ${issueRecoveryActions.companyId} and source.id::text = ${issueRecoveryActions.evidence}->>'runId' and source.runner_profile_json#>>'{remoteWorkspace,state}' = 'prepared')) + or (${issueRecoveryActions.cause} in ('native_workspace_sync_out_unrecoverable', 'native_workspace_sync_out_retry_exhausted') + and ${exhaustedStopProofWait} and ${stopProofRearms} >= ${STOP_PROOF_MAX_REARMS} + and exists (select 1 from heartbeat_runs source where source.company_id = ${issueRecoveryActions.companyId} + and source.id::text = ${issueRecoveryActions.evidence}->>'runId' + and source.runner_profile_json#>>'{remoteWorkspace,state}' = 'prepared')) or exists (select 1 from issue_recovery_actions sibling where sibling.company_id = ${issueRecoveryActions.companyId} and sibling.source_issue_id = ${issueRecoveryActions.sourceIssueId} and sibling.id <> ${issueRecoveryActions.id} and sibling.evidence->>'runId' = ${issueRecoveryActions.evidence}->>'runId' @@ -34,6 +98,7 @@ export function retainedSourceFallbackPendingCondition() { and sibling.evidence#>>'{workspaceRecovery,status}' = 'recovered')) and (${check} is null or ${check}->>'generationId' is distinct from ${issueRecoveryActions.evidence}#>>'{systemRecovery,generationId}' + or ${leaseChangedSinceCheck} or jsonb_typeof(${check}->'nextCheckAt') = 'number'))`; } diff --git a/server/src/services/workspace-recovery-fallback.ts b/server/src/services/workspace-recovery-fallback.ts index 53fb7216130..00278498886 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 } from "./workspace-recovery-fallback-discovery.js"; +import { PERMANENT_SOURCE_RETRY_CLASSES, RETAINED_SOURCE_FALLBACK_CHECK_KEY, UNVERIFIABLE_SOURCE_CODES, retainedHoldFallbackEligible } from "./workspace-recovery-fallback-discovery.js"; import { captureReadOnlyWorkspaceArchive, sandboxProviderFailureLabel } from "./remote-workspace-transport.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; import { readSystemRecovery, type SystemRecoveryState } from "./system-recovery-state.js"; @@ -115,7 +115,7 @@ async function inspectFallback(db: Db, input: Source, lock: boolean, requireStop const system = readSystemRecovery(action.evidence); if (!exportCauses.includes(action.cause) || !["active", "escalated"].includes(action.status) || action.evidence.runId !== run.id || action.evidence.executionReconciliation || system?.state !== "exhausted" || system.sourceRunId !== run.id - || !isUnverifiableSourceFailure(system.failure)) return refuse("not_unverifiable_source_hold"); + || !retainedHoldFallbackEligible(action.evidence, system)) return refuse("not_unverifiable_source_hold"); if (run.runtimeMode !== "native" || !terminal.includes(run.status) || !run.finishedAt || (run.nativeIssueId ?? run.issueId ?? run.contextSnapshot?.issueId) !== issue.id) return refuse("source_run_not_terminal"); // No second executor: the source run, its controller and its finalizer have stopped. @@ -417,7 +417,7 @@ export async function recordRetainedSourceFallbackDeferral(db: Db, input: Source /** * The recovery worker's entry for an exhausted hold: settle it when a sibling * saved its files, or continue from the last save when its source is refused - * permanently. Runs under the workspace ownership lock, like any physical + * permanently or its stop stayed unproven through every bounded re-arm. Runs under the workspace ownership lock, like any physical * recovery attempt. Returns whether the hold is settled. */ export async function settleExhaustedRetainedHold(input: Source & { db: Db; environmentRuntime: EnvironmentRuntimeService }, @@ -425,7 +425,10 @@ export async function settleExhaustedRetainedHold(input: Source & { db: Db; envi const { db } = input; const sibling = await settleSiblingRecoveredHold(db, input); if (sibling === "settled") return true; - if (!options.fallback || !isUnverifiableSourceFailure(state.failure)) { + const [row] = options.fallback ? await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), + eq(issueRecoveryActions.id, input.actionId))) : []; + if (!options.fallback || !retainedHoldFallbackEligible(row?.evidence, state)) { // A sibling marked recovered without a receipt that still verifies cannot // settle this hold. Record it so discovery does not select it again. if (sibling === "unverified") await recordRetainedSourceFallbackDeferral(db, input, "sibling_receipt_unverified"); diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index 1126d87aa97..c0365f4eaf5 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -67,7 +67,7 @@ export function workspaceRecoveryDisplay(evidence: Record, runI * described from its current state instead of the stored text. */ export function blockedWorkspaceRecoveryMessage(evidence: Record | null | undefined): string | null { const system = readSystemRecovery(evidence ?? undefined); - return system?.state === "exhausted" ? exhaustedSystemRecoveryAction(system) : null; + return system?.state === "exhausted" ? exhaustedSystemRecoveryAction(system, evidence ?? null) : null; } async function verifiedNativeCopybackReceipt(db: Db, input: { diff --git a/server/src/services/workspace-recovery-stop-proof.test.ts b/server/src/services/workspace-recovery-stop-proof.test.ts new file mode 100644 index 00000000000..520c4c63b16 --- /dev/null +++ b/server/src/services/workspace-recovery-stop-proof.test.ts @@ -0,0 +1,250 @@ +import { randomUUID } from "node:crypto"; +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 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 { 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 { acceptedRemoteExport, action, connectionString, db, run, seed, useWorkspaceRecoveryFixture } from "./__tests__/workspace-recovery-fixture.js"; + +const targets = vi.hoisted(() => new Map()); +vi.mock("./environment-execution-target.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => + (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; +}); + +// Recovery text never asks a person to inspect or fix a sandbox. +const adminText = /sandbox|environment|administrator|board member|owner must|inspect the run/i; +const workspace = async (id: string) => (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!; + +describe("exhausted stop-proof holds", () => { + useWorkspaceRecoveryFixture(targets); + beforeEach(() => { + vi.stubEnv("DATABASE_URL", connectionString); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); + }); + afterEach(() => { vi.unstubAllEnvs(); }); + + /** + * The staging shape (Nov 5): a parent task and three children that an older + * release bound to the parent's remote workspace. One child's run wrote the + * workspace, failed to save it, and its save recovery was exhausted after one + * attempt with `agent_stop_unconfirmed`: the recovery window ran out while + * the stop could not be proven yet. The run has since ended and its sandbox + * has a provider stop receipt, so the agent is provably gone. + */ + async function exhaustedStopProofHold() { + const f = await acceptedRemoteExport("retained"); + const source = await run(f.runId); + const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Landing page", status: "blocked", + assigneeAgentId: source.agentId, executionWorkspaceId: f.workspaceId, executionWorkspacePreference: "reuse_existing" }).returning(); + await db.update(executionWorkspaces).set({ sourceIssueId: parent.id }).where(eq(executionWorkspaces.id, f.workspaceId)); + await db.update(issues).set({ parentId: parent.id, executionWorkspacePreference: "reuse_existing" }).where(eq(issues.id, f.issueId)); + const siblings = await db.insert(issues).values(["Hero copy", "Pricing table"].map(title => ({ companyId: f.companyId, title, + parentId: parent.id, status: "blocked" as const, assigneeAgentId: source.agentId, executionWorkspaceId: f.workspaceId, + executionWorkspacePreference: "reuse_existing" }))).returning(); + const firstFailureAt = Date.now() - 5 * 86_400_000; + // The writer went silent days ago, past the workspace holder cutoff. + await db.update(heartbeatRuns).set({ finishedAt: new Date(firstFailureAt - 60_000) }).where(eq(heartbeatRuns.id, f.runId)); + const nextAction = "Automatic recovery of saved work stopped after 1 attempts (last failure: agent_stop_unconfirmed). " + + "Retry to try again; the agent will not repeat its work."; + const prior = await action(f.actionId); + const current = readSystemRecovery(prior.evidence)!; + // The exact evidence the Nov 5 release saved; no field this change adds. + await db.update(issueRecoveryActions).set({ ownerType: "system", nextAction, evidence: { ...prior.evidence, + systemRecovery: { ...current, revision: 3, phase: "verify_execution", state: "exhausted", attempts: 1, retriesUsed: 0, + attemptId: randomUUID(), firstFailureAt, deadlineAt: firstFailureAt + 30 * 60_000, nextAttemptAt: null, + failure: { code: "agent_stop_unconfirmed", retryClass: "recheck_evidence", evidenceFingerprint: "d".repeat(64) } }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: f.runId, status: "blocked", stage: "verifying_stop", + attempt: 1, attemptLimit: 6, retryable: true, requestIds: [] }, + coordinatorAttempt: 1, originalFailureCode: "native_workspace_sync_out_failed", workspaceFinalizeAttempt: 3 } }) + .where(eq(issueRecoveryActions.id, f.actionId)); + return { ...f, parentIssueId: parent.id, siblingIssueIds: siblings.map(row => row.id) }; + } + + async function parentHold(f: { companyId: string; workspaceId: string }) { + const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!; + return readRemoteWorkspaceWriterHold(db as never, { companyId: f.companyId, authority, runId: randomUUID() }); + } + + it("re-proves the stop, saves the files and frees the shared workspace without a person", async () => { + const f = await exhaustedStopProofHold(); + // The hold refuses every other task on the shared workspace, the parent included. + expect(remoteWorkspaceHoldRefusesStart(await parentHold(f))).toBe(true); + + // The recovery worker discovers it and wakes for it. + expect(await nextWorkspaceRecoveryAt(db)).toBeLessThanOrEqual(Date.now()); + const ids = (await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id); + expect(ids).toContain(f.actionId); + const onRecovered = vi.fn(async () => undefined); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, onRecovered, true) ?? []); + + const settled = await action(f.actionId); + expect(readSystemRecovery(settled.evidence)).toMatchObject({ state: "complete" }); + expect(readSystemRecovery(settled.evidence)!.preservationOperationIds.length).toBeGreaterThan(0); + expect(readWorkspaceRecovery(settled.evidence)?.status).toBe("recovered"); + expect(settled.evidence.stopProofRecovery).toMatchObject({ schema: "paperclip.stop-proof-recovery.v1", rearms: 1 }); + expect(settled.ownerType).toBe("system"); + expect(String(settled.nextAction)).not.toMatch(adminText); + expect(onRecovered).toHaveBeenCalledTimes(1); + + // The child's files are saved into the workspace it wrote; the hold is gone. + const authority = readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)!; + expect(authority).toMatchObject({ dirty: false }); + expect(authority.snapshotId).not.toBeNull(); + expect(await parentHold(f)).toEqual({ kind: "free" }); + expect(await readFile(path.join(f.remote, "saved.txt"), "utf8")).toBe("finished work\n"); + + const [rearmed] = await db.select().from(activityLog).where(eq(activityLog.action, "issue.workspace_recovery_rearmed")); + expect(rearmed).toMatchObject({ actorType: "system", entityId: f.issueId, runId: f.runId }); + }, 60_000); + + it("keeps the hold while the source run's controller still holds its lease", async () => { + const f = await exhaustedStopProofHold(); + const expiresAt = Date.now() + 60 * 60_000; + await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(expiresAt) }).where(eq(heartbeatRuns.id, f.runId)); + + // The worker is not woken until the lease can expire, and does not select the hold before that. + expect(await nextWorkspaceRecoveryAt(db)).toBeGreaterThanOrEqual(expiresAt - 1_000); + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.actionId); + await recoverRetainedWorkspace(f.input); + + const held = await action(f.actionId); + expect(readSystemRecovery(held.evidence)).toMatchObject({ state: "exhausted", failure: { code: "agent_stop_unconfirmed" } }); + expect(held.evidence.stopProofRecovery).toBeUndefined(); + expect(f.runtime.resumeRunLease).not.toHaveBeenCalled(); + }, 60_000); + + it("waits for the next re-arm on its backoff", async () => { + const f = await exhaustedStopProofHold(); + const prior = await action(f.actionId); + await db.update(issueRecoveryActions).set({ evidence: { ...prior.evidence, stopProofRecovery: { + schema: "paperclip.stop-proof-recovery.v1", rearms: 1, generationId: randomUUID(), previousGenerationId: randomUUID(), + rearmedAt: Date.now() - 60_000, nextRearmAt: Date.now() + 60 * 60_000 } } }).where(eq(issueRecoveryActions.id, f.actionId)); + + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.actionId); + expect(await nextWorkspaceRecoveryAt(db)).toBeGreaterThan(Date.now()); + await recoverRetainedWorkspace(f.input); + const waiting = await action(f.actionId); + expect(readSystemRecovery(waiting.evidence)?.state).toBe("exhausted"); + // The wait names the automatic retry; it asks no person to act. + expect(waiting.nextAction).toMatch(/tries again automatically/); + expect(String(waiting.nextAction)).not.toMatch(adminText); + }, 60_000); + + it("after its bounded re-arms, attaches the unsaved files and continues from the last save", async () => { + const f = await exhaustedStopProofHold(); + const prior = await action(f.actionId); + await db.update(issueRecoveryActions).set({ evidence: { ...prior.evidence, stopProofRecovery: { + schema: "paperclip.stop-proof-recovery.v1", rearms: 4, generationId: randomUUID(), previousGenerationId: randomUUID(), + rearmedAt: Date.now() - 86_400_000, nextRearmAt: null } } }).where(eq(issueRecoveryActions.id, f.actionId)); + + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).toContain(f.actionId); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => undefined, true) ?? []); + + const settled = await action(f.actionId); + expect(settled).toMatchObject({ status: "resolved", resolutionNote: RETAINED_SOURCE_FALLBACK_OUTCOME }); + const decision = settled.evidence.executionReconciliation as { retainedSourceFallback: { quarantine: { state: string } } }; + // No unsaved work is discarded silently: it is attached to the child task. + expect(decision.retainedSourceFallback.quarantine.state).toBe("captured"); + expect(await parentHold(f)).toEqual({ kind: "free" }); + }, 60_000); + + it("reconsiders a spent fallback when a late stop receipt arrives", async () => { + const f = await exhaustedStopProofHold(); + const prior = await action(f.actionId); + const generationId = readSystemRecovery(prior.evidence)!.generationId; + const checkedAt = Date.now() - 60_000; + await db.update(issueRecoveryActions).set({ evidence: { ...prior.evidence, + stopProofRecovery: { schema: "paperclip.stop-proof-recovery.v1", rearms: 4, generationId, previousGenerationId: randomUUID(), + rearmedAt: Date.now() - 86_400_000, nextRearmAt: null }, + retainedSourceFallbackCheck: { generationId, reason: "execution_not_stopped", checks: 12, checkedAt, nextCheckAt: null } } }) + .where(eq(issueRecoveryActions.id, f.actionId)); + await db.update(environmentLeases).set({ updatedAt: new Date(checkedAt - 60_000) }).where(eq(environmentLeases.heartbeatRunId, f.runId)); + const ids = async () => (await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id); + // Every bounded check is spent: the worker stops selecting the hold. + expect(await ids()).not.toContain(f.actionId); + + // A provider stop receipt lands on the source allocation afterwards. + await db.update(environmentLeases).set({ updatedAt: new Date() }).where(eq(environmentLeases.heartbeatRunId, f.runId)); + expect(await ids()).toContain(f.actionId); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => undefined, true) ?? []); + expect(await action(f.actionId)).toMatchObject({ status: "resolved", resolutionNote: RETAINED_SOURCE_FALLBACK_OUTCOME }); + expect(await parentHold(f)).toEqual({ kind: "free" }); + }, 60_000); + + /** Exhausts a saved-work recovery with the Nov 5 evidence shape: one attempt, `agent_stop_unconfirmed`, window spent. */ + async function exhaustStopWait(actionId: string, runId: string) { + const prior = await action(actionId); + const current = readSystemRecovery(prior.evidence)!; + const firstFailureAt = Date.now() - 5 * 86_400_000; + await db.update(issueRecoveryActions).set({ evidence: { ...prior.evidence, + systemRecovery: { ...current, revision: current.revision + 3, state: "exhausted", attempts: 1, retriesUsed: 0, attemptId: randomUUID(), + firstFailureAt, deadlineAt: firstFailureAt + 30 * 60_000, nextAttemptAt: null, + failure: { code: "agent_stop_unconfirmed", retryClass: "recheck_evidence", evidenceFingerprint: "e".repeat(64) } }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId, status: "blocked", stage: "verifying_stop", + attempt: 1, attemptLimit: 6, retryable: true, requestIds: [] } } }).where(eq(issueRecoveryActions.id, actionId)); + } + + it("re-arms an exhausted restore hold too, through the worker's own discovery", async () => { + // A restore hold (native continuation reconciliation), not a workspace export. + const f = await seed("native", true); + await adoptSystemRecovery(db, f); + expect((await action(f.actionId)).cause).toBe("native_continuation_requires_reconciliation"); + await exhaustStopWait(f.actionId, f.runId); + + expect(await nextWorkspaceRecoveryAt(db)).toBeLessThanOrEqual(Date.now()); + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).toContain(f.actionId); + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, async () => undefined, true) ?? []); + + const rearmed = await action(f.actionId); + expect(rearmed.evidence.stopProofRecovery).toMatchObject({ rearms: 1 }); + expect(readSystemRecovery(rearmed.evidence)).toMatchObject({ state: "complete" }); + expect(rearmed.ownerType).toBe("system"); + }, 60_000); + + it("leaves a stale finalizer owner to the abandoned-owner settlement instead of spending re-arms", async () => { + const f = await exhaustedStopProofHold(); + await db.update(nativeRunFinalizations).set({ leaseOwner: `native-finalizer:${randomUUID()}`, leaseExpiresAt: new Date(Date.now() - 60_000) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).not.toContain(f.actionId); + const next = await nextWorkspaceRecoveryAt(db); + expect(next === null || next > Date.now()).toBe(true); + await recoverRetainedWorkspace(f.input); + const held = await action(f.actionId); + expect(held.evidence.stopProofRecovery).toBeUndefined(); + expect(readSystemRecovery(held.evidence)?.state).toBe("exhausted"); + + // Once that settlement releases the owner, the hold is due again. + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect((await retainedWorkspaceCandidates(db, undefined, false, true)).map(row => row.action.id)).toContain(f.actionId); + }, 60_000); + + it("wakes the workspace recovery worker when a stop receipt is recorded", async () => { + const f = await acceptedRemoteExport("retained"); + const woken: string[] = []; + const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.workspaceRecovery, () => woken.push("workspace-recovery")); + try { + const [active] = await db.update(environmentLeases).set({ status: "active", releasedAt: null, cleanupStatus: null, + metadata: { ...f.lease.metadata, remoteExecutionTermination: undefined } }).where(eq(environmentLeases.id, f.lease.id)).returning(); + const stopping = await prepareSandboxStopAndRetain(db, active); + expect(stopping).not.toBeNull(); + woken.length = 0; + expect(await settleStopOnlyCleanup(db, stopping!, { attemptId: String(stopping!.metadata?.pendingCleanupAttemptId ?? randomUUID()), + receipt: { providerLeaseId: f.providerLeaseId, state: "stopped" } })).not.toBeNull(); + await vi.waitFor(() => expect(woken.length).toBeGreaterThan(0), { timeout: 5_000 }); + } finally { stop(); } + }, 60_000); +}); diff --git a/server/src/services/workspace-recovery-stop-proof.ts b/server/src/services/workspace-recovery-stop-proof.ts new file mode 100644 index 00000000000..704bb144eb0 --- /dev/null +++ b/server/src/services/workspace-recovery-stop-proof.ts @@ -0,0 +1,89 @@ +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { logger } from "../middleware/logger.js"; +import { adapterExecutionControls } from "./adapter-execution-control.js"; +import { persistActivity, publishActivity } from "./activity-log.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; +import { notifyIssueRecovery } from "./issue-recovery-notifications.js"; +import { createSystemRecovery, isExhaustedStopProofWait, readStopProofRecovery, readSystemRecovery, STOP_PROOF_MAX_REARMS, + STOP_PROOF_RECOVERY_KEY, STOP_PROOF_RECOVERY_SCHEMA, type SystemRecoveryState } from "./system-recovery-state.js"; +import { terminalWorkspaceRunStates, type WorkspaceRecoverySourceIdentity } from "./workspace-recovery-source.js"; + +/** Delay before re-arm n+1, measured from re-arm n. The first re-arm is due at once. */ +const STOP_PROOF_REARM_DELAYS_MS = [30 * 60_000, 2 * 60 * 60_000, 8 * 60 * 60_000]; + +/** User-facing text while a re-armed save runs. It names no person who must act. */ +export const STOP_PROOF_REARM_MESSAGE = "Paperclip is saving the previous run's files again now that the run has ended. " + + "No provider turn is submitted."; + +/** + * Re-arms an exhausted save whose last failure was an unproven agent stop. + * Exhaustion there only means the recovery window ran out before the stop + * could be proven; it is not a refusal of the saved files. Once the source run + * is provably over (terminal, no local execution, no live controller or + * finalizer lease), a new recovery generation repeats the strict stop proof + * (a provider stop receipt) and the save. Bounded: at most + * STOP_PROOF_MAX_REARMS re-arms on a growing backoff, after which the hold + * continues from the last save with the unsaved files attached + * (workspace-recovery-fallback). Holds exhausted by older releases carry no + * marker and are re-armed on the first pass, with no migration. + * Returns whether a new generation was written. + */ +export async function rearmExhaustedStopProof(db: Db, input: WorkspaceRecoverySourceIdentity, observed: SystemRecoveryState, + now = Date.now()): Promise { + if (!isExhaustedStopProofWait(observed)) return false; + const outcome = await db.transaction(async tx => { + // Lock order shared with the fallback: task, coordinator, run, recovery action. + const [issue] = await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, input.companyId), + eq(issues.id, input.issueId))).for("update"); + const [coordinator] = await tx.select({ leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.runId, input.runId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.id, input.runId))).for("update"); + const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, input.companyId), + eq(issueRecoveryActions.sourceIssueId, input.issueId), eq(issueRecoveryActions.id, input.actionId))).for("update"); + if (!issue || !run || !action) return null; + const system = readSystemRecovery(action.evidence); + if (!system || system.generationId !== observed.generationId || system.revision !== observed.revision + || !isExhaustedStopProofWait(system) || system.sourceRunId !== run.id) return null; + if (!["active", "escalated"].includes(action.status) + || action.evidence.runId !== run.id || action.evidence.executionReconciliation || action.evidence.retainedSourceFallback) return null; + // The agent is provably gone: its run ended and nothing still executes or + // finalizes it. The new attempt still proves the sandbox stop itself. + const at = new Date(now); + if (!terminalWorkspaceRunStates.includes(run.status) || !run.finishedAt || adapterExecutionControls.has(run.id) + || (run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > at) + // The attempt refuses any recorded finalizer owner (inspectWorkspaceRecoverySource); + // a stale one is released by the abandoned-owner settlement first. + || coordinator?.leaseOwner) return null; + const marker = readStopProofRecovery(action.evidence); + const rearms = marker?.rearms ?? 0; + if (rearms >= STOP_PROOF_MAX_REARMS || (marker?.nextRearmAt ?? 0) > now) return null; + // Any workspace recovery the worker owns: export holds and restore holds alike. + const next: SystemRecoveryState = { ...createSystemRecovery(run.id), revision: system.revision + 1, + preservationOperationIds: system.preservationOperationIds, providerCooldownAt: system.providerCooldownAt, + nextAttemptAt: system.providerCooldownAt }; + const count = rearms + 1; + const delay = STOP_PROOF_REARM_DELAYS_MS[Math.min(count - 1, STOP_PROOF_REARM_DELAYS_MS.length - 1)]; + const stopProofRecovery = { schema: STOP_PROOF_RECOVERY_SCHEMA, rearms: count, generationId: next.generationId, + previousGenerationId: system.generationId, rearmedAt: now, nextRearmAt: count < STOP_PROOF_MAX_REARMS ? now + delay : null }; + const prior = action.evidence.workspaceRecovery as { requestIds?: unknown } | undefined; + await notifyIssueRecovery(tx, { companyId: input.companyId }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.workspaceRecovery); + await tx.update(issueRecoveryActions).set({ ownerType: "system", ownerAgentId: null, ownerUserId: null, + nextAction: STOP_PROOF_REARM_MESSAGE, updatedAt: at, + evidence: { ...action.evidence, systemRecovery: next, [STOP_PROOF_RECOVERY_KEY]: stopProofRecovery, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId: run.id, status: "queued", stage: "verifying_stop", + attempt: 0, attemptLimit: 6, retryable: true, message: STOP_PROOF_REARM_MESSAGE, + requestIds: Array.isArray(prior?.requestIds) ? prior.requestIds : [] } }, + }).where(and(eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, action.id))); + return persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "workspace-recovery", + action: "issue.workspace_recovery_rearmed", entityType: "issue", entityId: input.issueId, runId: run.id, + details: { recoveryActionId: action.id, failureCode: system.failure?.code ?? null, rearm: count, + maxRearms: STOP_PROOF_MAX_REARMS, previousGenerationId: system.generationId, generationId: next.generationId } }); + }); + if (!outcome) return false; + try { publishActivity(outcome.publication); } catch { logger.warn({ runId: input.runId }, "Recovery activity delivery deferred"); } + return true; +} diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 656d8ba83d9..5be4f8bf03c 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -36,7 +36,8 @@ import { adoptHistoricalWorkspaceRecoveries } from "./workspace-recovery-backfil import { canAdoptNativeSourceRetention, retainNativeRecoverySource, settleRecoveredNativeSourceRetention } from "./native-runtime/native-source-retention.js"; import { saveRemoteWorkspace } from "./remote-workspaces.js"; import { continueFromLastSavedWorkspace, isUnverifiableSourceFailure, recordRetainedSourceFallbackDeferral, settleExhaustedRetainedHold } from "./workspace-recovery-fallback.js"; -import { retainedSourceFallbackDueCondition } from "./workspace-recovery-fallback-discovery.js"; +import { retainedSourceFallbackDueCondition, stopProofRearmDueCondition } from "./workspace-recovery-fallback-discovery.js"; +import { rearmExhaustedStopProof } from "./workspace-recovery-stop-proof.js"; import { settleRecoveredWorkspaceRestoreHolds } from "./workspace-restore-hold-settlement.js"; const exportCauses = ["native_workspace_sync_out_unrecoverable", "native_workspace_sync_out_retry_exhausted"]; @@ -198,13 +199,16 @@ export async function requestWorkspaceRecovery(input: Source & { db: Db; idempot export async function recoverRetainedWorkspace(input: Source & { db: Db; environmentRuntime: EnvironmentRuntimeService }) { const { db } = input; await adoptSystemRecovery(db, input); - const initial = await loadSystemRecovery(db, input); + let initial = await loadSystemRecovery(db, input); // A hold whose files a sibling saved, or whose source is refused permanently // (including holds saved by older releases), settles without a person. // An adoptable native source keeps its retention path; only the sibling // settlement applies to it. if (initial?.state === "exhausted" && await settleExhaustedRetainedHold(input, initial, { fallback: !canAdoptNativeSourceRetention(initial) })) return; + // An exhausted wait for stop proof is re-armed, bounded, once the run is over. + if (initial?.state === "exhausted" && await rearmExhaustedStopProof(db, input, initial)) + initial = await loadSystemRecovery(db, input); if (initial?.state === "exhausted" && !canAdoptNativeSourceRetention(initial)) { // A budget spent before this hold existed still names its receipt, so the // board confirmation in its handoff can release it (RC-50 wave databases). @@ -624,7 +628,7 @@ export function retainedWorkspaceCandidates(db: Db, cursor?: string, unfinishedO and coalesce(${issueRecoveryActions.evidence}->>'bootstrapPreparationContinuation', 'pending') = 'pending'))` : undefined, obligationsOnly ? sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence') or (${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' = 'complete' and (${cleanup} or ${interruptedPreparation})) or ${waitingInput} - or ${retainedSourceFallbackDueCondition(Date.now())})` : undefined, + or ${retainedSourceFallbackDueCondition(Date.now())} or ${stopProofRearmDueCondition(Date.now())})` : undefined, sql`(${issueRecoveryActions.evidence}#>>'{systemRecovery,state}' in ('scheduled', 'running', 'awaiting_evidence', 'complete', 'exhausted') and (${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' = 'null'::jsonb or ${issueRecoveryActions.evidence}#>'{systemRecovery,nextAttemptAt}' <= to_jsonb(${Date.now()}::bigint)))`,