diff --git a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts index 4f4373898db..31e563a2dca 100644 --- a/server/src/__tests__/interrupted-preparation-hold-settle.test.ts +++ b/server/src/__tests__/interrupted-preparation-hold-settle.test.ts @@ -11,6 +11,7 @@ import { createDb, environmentLeases, environments, + executionWorkspaces, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, @@ -20,6 +21,8 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro import { currentProcessSpaceHash } from "../services/controller-process-identity.js"; import { heartbeatService } from "../services/heartbeat.js"; import { settleHeldInterruptedPreparation } from "../services/interrupted-preparation-hold-settlement.js"; +import { isAbandonedPreparedRemoteWorkspace } from "../services/remote-workspace-abandoned-preparation.js"; +import { inspectWorkspaceRecoverySource } from "../services/workspace-recovery-source.js"; import { legacyControllerBootId } from "../services/legacy-controller-lease.js"; import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js"; @@ -67,6 +70,7 @@ suite("held interrupted preparations on the shared ownership path", () => { await db.delete(activityLog); await db.delete(issueRecoveryActions); await db.delete(environmentLeases); + await db.delete(executionWorkspaces); await db.delete(heartbeatRunEvents); await db.delete(heartbeatRuns); await db.delete(issues); @@ -171,6 +175,90 @@ suite("held interrupted preparations on the shared ownership path", () => { expect(await hold(f.holdId)).toEqual(before); }); + async function preparedRemoteWorkspace(runId: string, issueId: string, options: { receipt?: Record; + savedSnapshotId?: string | null; authority?: Record; profile?: Record } = {}) { + const workspaceId = randomUUID(), leaseId = randomUUID(), providerLeaseId = `sandbox-${leaseId}`; + const remoteCwd = `/home/daytona/paperclip-workspace/.paperclip-workspaces/${workspaceId}-1-${leaseId}`; + const location = { environmentId, leaseId, providerLeaseId, remoteCwd }; + const source = { kind: "git", repoUrl: "https://github.com/paperclipai/paperclip.git", ref: "a".repeat(40), + branchMode: "create", branchName: "IPH-1-change" }; + const authority = { schema: "paperclip.remote-workspace.v1", epoch: 2, source, location: null, dirty: false, writerRunId: null, + snapshotId: options.savedSnapshotId ?? null, ...options.authority }; + const receipts = [{ schema: "paperclip.remote-workspace-abandoned-epoch.v1", epoch: 1, runId, + snapshotId: options.savedSnapshotId ?? null, location, providerRef: remoteCwd, abandonedAt: new Date().toISOString(), ...options.receipt }]; + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issueId, mode: "isolated_workspace", + strategyType: "git_worktree", name: "Remote workspace", status: "active", providerType: "git_worktree", + metadata: { remoteWorkspace: authority, remoteWorkspaceAbandonedEpochs: receipts } }); + await db.update(heartbeatRuns).set({ runnerProfileJson: { adapterDispatch: { adapterType: "paperclip_runner" }, + remoteWorkspace: { schema: "paperclip.remote-workspace-run.v1", state: "prepared", workspaceId, epoch: 1, leaseId, + providerLeaseId, remoteCwd, snapshotId: null }, ...options.profile } }).where(eq(heartbeatRuns.id, runId)); + return { leaseId }; + } + + const runRow = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]!; + + it("replays a restore-recovery hold saved by an older build once the workspace retired the run's epoch", async () => { + heartbeatService(db); + const f = await interruptedPreparation(); + const { leaseId } = await preparedRemoteWorkspace(f.runId, f.issueId); + const restore = { workspaceRestoreFailure: "restore_failed", + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [leaseId] } }; + const run = await runRow(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...run.resultJson, ...restore } }).where(eq(heartbeatRuns.id, f.runId)); + const saved = await hold(f.holdId); + await db.update(issueRecoveryActions).set({ evidence: { ...saved.evidence, ...restore, + systemRecovery: { schema: "paperclip.system-recovery.v1", policyVersion: 1, state: "exhausted", phase: "verify_execution", + sourceRunId: f.runId, attempts: 1, retriesUsed: 0, revision: 5, nextAttemptAt: null, + failure: { code: "source_binding_invalid", retryClass: "service_required", sourceVerification: { reason: "workspace_reference_mismatch" } } } } }) + .where(eq(issueRecoveryActions.id, f.holdId)); + expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId))).toHaveLength(1); + }); + + it("reports a retired prepared epoch to workspace recovery as an unavailable source, not a binding failure", async () => { + const f = await interruptedPreparation(); + await preparedRemoteWorkspace(f.runId, f.issueId); + await expect(inspectWorkspaceRecoverySource(db, { companyId, issueId: f.issueId, runId: f.runId, actionId: f.holdId })) + .rejects.toMatchObject({ code: "source_unavailable", retryClass: "use_alternative_strategy" }); + }); + + const proofCases: Array<[string, { authority?: Record; profile?: Record; writer?: boolean }, boolean]> = [ + ["the workspace is still at the run's epoch", { authority: { epoch: 1 } }, false], + ["the run is still the workspace writer", { writer: true }, false], + ["the run also has a native workspace sync reference", { profile: { nativeWorkspaceSync: { state: "prepared" } } }, false], + ["the receipt matches the run's exact epoch and allocation", {}, true], + ]; + it.each(proofCases)("treats the prepared workspace as retired only when proven: %s", async (_label, options, expected) => { + const f = await interruptedPreparation(); + const authority = options.writer ? { writerRunId: f.runId } : options.authority; + await preparedRemoteWorkspace(f.runId, f.issueId, { profile: options.profile, authority }); + expect(await isAbandonedPreparedRemoteWorkspace(db, await runRow(f.runId))).toBe(expected); + }); + + it.each([ + ["before its first save finished", null], + ["after its first save committed", randomUUID()], + ] as const)("replays a remote preparation whose unsaved epoch the workspace already retired %s", async (_label, savedSnapshotId) => { + heartbeatService(db); + const f = await interruptedPreparation(); + await preparedRemoteWorkspace(f.runId, f.issueId, { savedSnapshotId }); + expect(await pass()).toMatchObject({ inProgress: 1, settled: 1 }); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toMatchObject({ status: "in_progress" }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, f.runId))).toHaveLength(1); + }); + + it.each([ + ["the retired epoch belongs to another run", { receipt: { runId: randomUUID() } }], + ["the retired epoch was another allocation", { receipt: { location: { leaseId: randomUUID() } } }], + ] as const)("keeps a remote preparation hold when %s", async (_label, options) => { + const f = await interruptedPreparation(); + await preparedRemoteWorkspace(f.runId, f.issueId, options); + const before = await hold(f.holdId); + expect((await pass())?.settled ?? 0).toBe(0); + expect(await hold(f.holdId)).toEqual(before); + }); + it("settles exactly once when two processes judge the same hold", async () => { heartbeatService(db); const f = await interruptedPreparation(); diff --git a/server/src/services/interrupted-preparation-hold-settlement.ts b/server/src/services/interrupted-preparation-hold-settlement.ts index 99de8f7ccc6..c6647d03ed8 100644 --- a/server/src/services/interrupted-preparation-hold-settlement.ts +++ b/server/src/services/interrupted-preparation-hold-settlement.ts @@ -6,6 +6,7 @@ import { logger } from "../middleware/logger.js"; import type { OwnerProof } from "./owner-proof.js"; import type { InProgressRecord, SettleOutcome } from "./recovery/ownership-registry.js"; import { OWNER_GONE_ATTESTATION_KEY } from "./recovery/owner-gone-attestation.js"; +import { abandonedPreparedRemoteWorkspaceExists } from "./remote-workspace-abandoned-preparation.js"; /** * `legacy_interrupted_preparation_hold`: a run a graceful shutdown interrupted @@ -19,9 +20,11 @@ import { OWNER_GONE_ATTESTATION_KEY } from "./recovery/owner-gone-attestation.js * * Only holds the system recorded itself are listed: an operator decision (an * owner-gone attestation, an explicit user continuation, an assigned owner) is - * never reopened, and a workspace-restore hold (restore failure or recovery - * evidence) belongs to restore recovery, not to this kind. A run with a lease still cleaning up is not listed yet; the - * next pass lists it once the lease settles. + * never reopened. A workspace-restore hold (restore failure or recovery + * evidence) belongs to restore recovery, unless the workspace already retired + * the run's exact prepared epoch (`isAbandonedPreparedRemoteWorkspace`): then + * nothing is left to restore and this kind settles it. A run with a lease still + * cleaning up is not listed yet; the next pass lists it once the lease settles. */ export function heldInterruptedPreparationWhere() { return and(inArray(heartbeatRuns.status, ["interrupted", "failed"]), eq(heartbeatRuns.runtimeMode, "legacy"), @@ -32,8 +35,9 @@ export function heldInterruptedPreparationWhere() { and hold.status = 'resolved' and hold.outcome = 'blocked' and hold.owner_user_id is null and hold.evidence->'automaticRecovery'->>'policy' = 'preserve_without_replay_v1' and hold.evidence->'automaticRecovery'->>'replay' = 'blocked' - and not (hold.evidence ?| array['executionReconciliation', ${OWNER_GONE_ATTESTATION_KEY}, 'explicitUserContinuation', - 'workspaceRestoreFailure', 'workspaceRestoreRecovery']))`, + and not (hold.evidence ?| array['executionReconciliation', ${OWNER_GONE_ATTESTATION_KEY}, 'explicitUserContinuation']) + and (not (hold.evidence ?| array['workspaceRestoreFailure', 'workspaceRestoreRecovery']) + or ${abandonedPreparedRemoteWorkspaceExists()}))`, sql`not exists (select 1 from environment_leases lease where lease.company_id = ${heartbeatRuns.companyId} and lease.heartbeat_run_id = ${heartbeatRuns.id} and (lease.released_at is null or lease.status = 'pending_cleanup' or lease.cleanup_status = 'failed'))`); diff --git a/server/src/services/remote-workspace-abandoned-preparation.ts b/server/src/services/remote-workspace-abandoned-preparation.ts new file mode 100644 index 00000000000..dfc94a92cd2 --- /dev/null +++ b/server/src/services/remote-workspace-abandoned-preparation.ts @@ -0,0 +1,37 @@ +import { and, eq, sql } from "drizzle-orm"; +import { executionWorkspaces, heartbeatRuns, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema } from "@paperclipai/shared"; +import { REMOTE_WORKSPACE_ABANDONED_EPOCH_SCHEMA } from "./remote-workspace-epoch-abandonment.js"; + +type Run = Pick; + +export async function isAbandonedPreparedRemoteWorkspace(db: Db, run: Run): Promise { + if (run.runnerProfileJson?.nativeWorkspaceSync != null) return false; + const parsed = remoteWorkspaceRunReferenceSchema.safeParse(run.runnerProfileJson?.remoteWorkspace); + if (!parsed.success || parsed.data.state !== "prepared") return false; + const reference = parsed.data; + const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, run.companyId), eq(executionWorkspaces.id, reference.workspaceId))); + const authority = readRemoteWorkspaceAuthority(workspace?.metadata); + if (!authority || authority.epoch <= reference.epoch || authority.writerRunId === run.id) return false; + const receipts = Array.isArray(workspace?.metadata?.remoteWorkspaceAbandonedEpochs) + ? workspace!.metadata!.remoteWorkspaceAbandonedEpochs as Array> : []; + return receipts.some(receipt => { + const location = receipt?.location as Record | undefined; + return receipt?.schema === REMOTE_WORKSPACE_ABANDONED_EPOCH_SCHEMA && receipt.runId === run.id + && receipt.epoch === reference.epoch + && location?.leaseId === reference.leaseId && location?.providerLeaseId === reference.providerLeaseId + && location?.remoteCwd === reference.remoteCwd; + }); +} + +export function abandonedPreparedRemoteWorkspaceExists() { + const reference = sql`${heartbeatRuns.runnerProfileJson}->'remoteWorkspace'`; + return sql`(${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync' is null and ${reference}->>'state' = 'prepared' + and exists (select 1 from execution_workspaces workspace + cross join lateral jsonb_array_elements(case when jsonb_typeof(workspace.metadata->'remoteWorkspaceAbandonedEpochs') = 'array' + then workspace.metadata->'remoteWorkspaceAbandonedEpochs' else '[]'::jsonb end) receipt + where workspace.company_id = ${heartbeatRuns.companyId} and workspace.id::text = ${reference}->>'workspaceId' + and receipt->>'schema' = ${REMOTE_WORKSPACE_ABANDONED_EPOCH_SCHEMA} and receipt->>'runId' = ${heartbeatRuns.id}::text + and receipt->'epoch' = ${reference}->'epoch'))`; +} diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index a79334c7dc7..12dbd51bc50 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -7,6 +7,7 @@ import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-m import { readSnapshot } from "./workspace-recovery-snapshot.js"; import { exhaustedSystemRecoveryAction, readSystemRecovery } from "./system-recovery-state.js"; import { verifiedWorkspaceSnapshotReuse } from "./remote-workspace-snapshot-reuse-receipt.js"; +import { isAbandonedPreparedRemoteWorkspace } from "./remote-workspace-abandoned-preparation.js"; export const workspaceRecoverySchema = z.object({ schema: z.literal("paperclip.workspace-recovery.v1"), @@ -176,7 +177,7 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick) { - return (await verifiedWorkspaceRepairReceipt(...args)) !== null; + return (await verifiedWorkspaceRepairReceipt(...args)) !== null || await isAbandonedPreparedRemoteWorkspace(...args); } /** Read-only historical bytes have no execution authority. Their immutable diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index 04b4f04cc0e..f35a69fb599 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -7,6 +7,7 @@ import { readNativeWorkspaceSyncReference } from "./native-runtime/native-worksp import { readSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; import type { NativeWorkspaceFinalizationOwnership } from "./native-runtime/native-workspace-finalization-ownership.js"; import { readStopOnlyCleanup } from "./sandbox-stop-and-retain.js"; +import { isAbandonedPreparedRemoteWorkspace } from "./remote-workspace-abandoned-preparation.js"; import { LEGACY_WORKSPACE_RECOVERY_SCHEMA, SETTLED_WORKSPACE_RESTORE_HOLD_KEY } from "./workspace-restore-recovery-state.js"; import { ENVIRONMENT_LEASE_POLICIES, EXECUTION_RECONCILIATION_CAUSES, readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type EnvironmentLeasePolicy, type WorkspaceSourceVerificationFailure } from "@paperclipai/shared"; @@ -109,6 +110,7 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec || remoteAuthority.location.remoteCwd !== remoteReference.remoteCwd || (remoteReference.state === "prepared" ? !remoteAuthority.dirty || remoteAuthority.writerRunId !== run.id : remoteAuthority.dirty || remoteAuthority.writerRunId !== null || remoteAuthority.snapshotId !== remoteReference.snapshotId))) { + if (await isAbandonedPreparedRemoteWorkspace(db, run)) throw new SystemRecoveryFailure("source_unavailable", "use_alternative_strategy"); throw invalid("workspace_reference_mismatch"); } if (remoteReference?.state === "saved") {