diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index b3fbce2fe1c..588b605e9d3 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -65,6 +65,18 @@ replaying the old agent turn or destroying its allocation. Publication requires the offline deployment fence, terminal unowned runs, unchanged descriptors and task bindings, verified local snapshot bytes, and a single transaction publishing the migration receipt and remote authority. +Failed runs with result-less `observed` or `retryable_failure` coordinators can +also be preserved after their leases expire and the offline deployment fence is +in force. Their coordinator, checkpoint, and unresolved copyback remain intact. +Publication journals the original coordinator binding and installs a run-scoped +admission fence: a replacement cannot automatically replay that old turn against +the removed local authority. Explicit file recovery and a new task continuation +use the retained sources and remote workspace. This is not successful recovery, +provider termination evidence, or permission to delete the original allocation. +Accepted results with unfinished copyback remain blockers. Once the original +sync receipt is finalized, assessment and status effects can continue from their +existing database coordinator independently of the file storage migration. + Controller leases, in-flight recovery, active writers, and uncertain workspace operations remain blockers. Tests must prove migration leaves failure state intact, rejects stale copyback even if the old folder exists, and preserves diff --git a/server/src/services/heartbeat/live-recovery-work.ts b/server/src/services/heartbeat/live-recovery-work.ts index 3c008175f91..58dd3f5f8de 100644 --- a/server/src/services/heartbeat/live-recovery-work.ts +++ b/server/src/services/heartbeat/live-recovery-work.ts @@ -1,4 +1,5 @@ import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js"; +import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { settledNativeProcessIdentityCondition } from "../native-local-process-stop.js"; import { and, asc, eq, gt, inArray, isNotNull, isNull, or, sql } from "drizzle-orm"; import { agents, environmentLeases, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db"; @@ -73,6 +74,7 @@ export function registerLiveRecoveryWork(db: Db, coordinator: ReturnType'nativeQuestionCancellation' is null`) : kind === "question" ? and(or(inArray(heartbeatRuns.status, ["queued", "running", "scheduled_retry"]), diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 40351ddfb1a..2912b041065 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,4 +1,5 @@ import { readNativeStartupCleanup, readNativeStartupContext } from "../native-runtime/native-startup-cleanup.js"; +import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { pendingFailedNativeRetryCancellationCondition } from "../native-runtime/native-cancellation-request.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { registerLiveRecoveryWork } from "./live-recovery-work.js"; @@ -962,6 +963,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD inArray(heartbeatRuns.status, ["running", "failed"]), isNull(nativeRunFinalizations.resultId), eq(nativeRunFinalizations.phase, "retryable_failure"), + legacyWorkspaceRecoveryNotMigrated(), or(gt(nativeRunFinalizations.nextAttemptAt, now), eq(nativeRunFinalizations.failureCode, NATIVE_RETAINED_OWNER_WAIT)), ), ); @@ -2051,6 +2053,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD eq(heartbeatRuns.runtimeMode, "native"), eq(nativeRunFinalizations.phase, "retryable_failure"), isNull(nativeRunFinalizations.resultId), + legacyWorkspaceRecoveryNotMigrated(), sql`${heartbeatRuns.contextSnapshot}->'nativeQuestionCancellation' is null`, ), ); diff --git a/server/src/services/legacy-workspace-recovery-migration.ts b/server/src/services/legacy-workspace-recovery-migration.ts index e4b17f07fb9..f9ef179fe41 100644 --- a/server/src/services/legacy-workspace-recovery-migration.ts +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -1,12 +1,22 @@ import { isDeepStrictEqual } from "node:util"; -import { and, eq } from "drizzle-orm"; -import { executionWorkspaces, issues, workspaceOperations, type Db } from "@paperclipai/db"; +import { and, eq, sql } from "drizzle-orm"; +import { executionWorkspaces, heartbeatRuns, issues, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import type { NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; -export type LegacyWorkspaceRecoverySource = { runId: string; issueId: string; reference: NativeWorkspaceSyncReference }; +export type LegacyWorkspaceRecoverySource = { + runId: string; issueId: string; reference: NativeWorkspaceSyncReference | null; + unresolvedFinalization?: { phase: string; leaseOwner: string | null; leaseExpiresAt: string | null }; +}; export const LEGACY_WORKSPACE_RECOVERY_MIGRATION = "paperclip.legacy-workspace-recovery-migration.v1"; +/** Offline migration preserves an unresolved coordinator, but it cannot replay + * an old execution against a local copyback destination that is no longer its + * authority. File recovery and explicit task continuation remain available. */ +export function legacyWorkspaceRecoveryNotMigrated() { + return sql`${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration' is null`; +} + /** Child tasks can explicitly reuse an ancestor's workspace. The run's own * task remains the scope for its recovery files; migration never rebinds it. */ export async function isLegacyWorkspaceTaskRelated(db: Pick, input: { diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index bed067977d1..6a8b1d43ca2 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -63,6 +63,7 @@ import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold import { readRunCancellation } from "../run-cancellation.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; +import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) @@ -115,6 +116,7 @@ export async function claimNativeSessionResumptions(input: { .where( and( eq(heartbeatRuns.runtimeMode, "native"), + legacyWorkspaceRecoveryNotMigrated(), nativeRunnerOwnershipNotHeldCondition(), isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), @@ -156,7 +158,7 @@ export async function claimNativeSessionResumptions(input: { eq(heartbeatRuns.companyId, candidate.companyId), eq(heartbeatRuns.id, candidate.runId), eq(heartbeatRuns.nativeIssueId, issue.id), )).for("update").limit(1); - if (!run || !coordinator) return false; + if (!run || !coordinator || run.runnerProfileJson?.legacyWorkspaceRecoveryMigration != null) return false; if (await retireSupersededNativeRetry(tx as unknown as Db, { run, coordinator, issueExecutionRunId: issue.executionRunId, now, })) return false; diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index a24fccec657..6f0cdaf91d8 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -39,6 +39,7 @@ import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; import { nativeRestartInterruptedTurnId, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { retireSupersededNativeRetry } from "./native-superseded-retry.js"; +import { legacyWorkspaceRecoveryNotMigrated } from "../legacy-workspace-recovery-migration.js"; import { requestNativeStartupCleanup, readNativeStartupContext, promoteNativeStartupCleanup } from "./native-startup-cleanup.js"; export type NativeControllerIdentity = { @@ -524,6 +525,7 @@ export async function claimNativeRestartRecoveries(input: { .where( and( eq(heartbeatRuns.runtimeMode, "native"), + legacyWorkspaceRecoveryNotMigrated(), eq(agents.adapterType, "paperclip_runner"), or(inArray(heartbeatRuns.status, ["running", "failed"]), and( eq(heartbeatRuns.status, "interrupted"), eq(nativeRunFinalizations.phase, "observed"), @@ -603,6 +605,8 @@ export async function claimNativeRestartRecoveries(input: { if (!row) { return busy(); } + if (row.run.runnerProfileJson?.legacyWorkspaceRecoveryMigration != null) + return { kind: "blocked", runId: row.run.id, reason: "legacy_workspace_recovery_migrated" } as const; const stopCleanup = row.coordinator.phase === "observed" && row.coordinator.companyId === row.run.companyId && row.coordinator.issueId === row.run.nativeIssueId && await readPendingNativeRunStop(tx as unknown as Db, row.run); diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index b94a9a4fa57..8dcb754a46c 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -8,7 +8,7 @@ import { createServer, type Server } from "node:http"; import { pathToFileURL } from "node:url"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; -import { activityLog, agents, heartbeatRuns, nativeRunFinalizations, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, +import { activityLog, agents, heartbeatRuns, nativeRunFinalizations, nativeRunResults, completionContracts, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, executionWorkspaceRuntimeLeases, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; @@ -26,6 +26,8 @@ import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { cleanupExecutionWorkspaceArtifacts } from "./workspace-runtime.js"; import { workspaceOperationService } from "./workspace-operations.js"; import { resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; +import { claimNativeSessionResumptions } from "./native-runtime/native-finalization-reconciler.js"; +import { claimNativeRestartRecoveries } from "./native-runtime/native-restart-recovery.js"; import { managedProjectCheckoutName } from "./managed-project-checkout-name.js"; import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js"; @@ -52,7 +54,8 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await Promise.all(servers.splice(0).map(server => new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())))); vi.unstubAllEnvs(); await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); await db.delete(workspaceRuntimeServices); await db.delete(executionWorkspaceRuntimeLeases); - await db.delete(nativeRunFinalizations); await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); + await db.delete(nativeRunFinalizations); await db.delete(nativeRunResults); await db.delete(heartbeatRuns); await db.delete(completionContracts); + await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); await db.delete(companies); await fs.rm(root, { recursive: true, force: true }); }); async function seed() { @@ -143,7 +146,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { }); async function failedCopybackFixture() { const f = await seed(), runId = randomUUID(); - const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent" }).returning(); + const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent", adapterType: "paperclip_runner" }).returning(); const reference = { schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: f.workspaceId, leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/task", descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), finalHostSha256: null, resourceDisposition: "stop_and_retain" }; @@ -167,11 +170,12 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) .rejects.toThrow("native_workspace_sync_source_migrated"); }); - it.each(["controller", "finalizer", "retryable", "task", "descriptor"])("keeps unsafe legacy copyback blocked (%s)", async mode => { + it.each(["controller", "finalizer", "live-finalizer", "task", "descriptor"])("keeps unsafe legacy copyback blocked (%s)", async mode => { const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); if (mode === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId)); if (mode === "finalizer") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId)); - if (mode === "retryable") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "live-finalizer") await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: randomUUID(), + leaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); if (mode === "task") { const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Different task" }).returning(); await db.update(executionWorkspaces).set({ sourceIssueId: other!.id }).where(eq(executionWorkspaces.id, f.workspaceId)); @@ -180,6 +184,63 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_copyback_unresolved"); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); }); + it.each(["retryable_failure", "observed", "no-descriptor"])("preserves an abandoned %s coordinator without replaying its old agent turn", async mode => { + const phase = mode === "no-descriptor" ? "observed" : mode; + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const [oldRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const checkpoint = { providerSessionId: "original-provider-session", sequence: 17 }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...(mode === "no-descriptor" ? {} : oldRun!.runnerProfileJson), + sessionCheckpoint: checkpoint } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(issues).set({ assigneeAgentId: oldRun!.agentId, status: "in_progress" }) + .where(eq(issues.id, f.issueId)); + await db.update(nativeRunFinalizations).set({ phase, leaseOwner: phase === "observed" ? "old-controller" : null, + leaseExpiresAt: phase === "observed" ? new Date(Date.now() - 60_000) : null, + nextAttemptAt: new Date(Date.now() - 30_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "replacement", runIds: [f.runId] })).toEqual([]); + expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).toEqual([]); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]).toMatchObject({ + status: "failed", runnerProfileJson: { ...(mode === "no-descriptor" ? {} : { nativeWorkspaceSync: f.reference }), sessionCheckpoint: checkpoint, + legacyWorkspaceRecoveryMigration: { workspaceId: f.workspaceId } }, + }); + }); + it.each(["prepared", "finalized"])("keeps accepted-result finalization independent of migrated file storage (%s)", async state => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const contractId = randomUUID(), resultId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, revision: 1, + schemaVersion: "v1", policyVersion: "v1", risk: "low", completionAuthority: "agent", incompleteCriteriaPolicy: "block", + contractJson: {}, canonicalSha256: "12".repeat(32), createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ status: "succeeded", completionContractId: contractId, + runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, state, finalHostSha256: state === "finalized" ? "56".repeat(32) : null } } }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: contractId, serverFingerprint: "saved-result", schemaStatus: "accepted", resultJson: { saved: true }, canonicalSha256: "34".repeat(32) }); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId }).where(eq(nativeRunFinalizations.runId, f.runId)); + const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const migrate = () => migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + if (state === "prepared") await expect(migrate()).rejects.toThrow("remote_copyback_unresolved"); + else { + await migrate(); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]!.runnerProfileJson) + .not.toHaveProperty("legacyWorkspaceRecoveryMigration"); + } + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(before); + }); + it("migrates historical records without borrowing another workspace's finalizer", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const otherId = randomUUID(), cwd = path.join(root, "longer-historical-folder"); await fs.mkdir(cwd); + await fs.writeFile(path.join(cwd, "history.txt"), "prior task files\n"); + await db.insert(executionWorkspaces).values({ id: otherId, companyId: f.companyId, sourceIssueId: f.issueId, + name: "Historical workspace", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd }); + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).toBe(2); + for (const id of [f.workspaceId, otherId]) expect(readRemoteWorkspaceAuthority((await workspace(id)).metadata)?.snapshotId).toBeTruthy(); + }); it("rejects a legacy descriptor changed while its local files are uploading", async () => { const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); const put = storage.putFile.bind(storage); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index aed67144fb3..bc892a6c90d 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -109,20 +109,56 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri const legacyRecoverySources: LegacyWorkspaceRecoverySource[] = []; for (const run of unexported.sort((a, b) => a.id.localeCompare(b.id))) { const reference = readNativeWorkspaceSyncReference(run.reference); - const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner }) + const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner, + leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, resultId: nativeRunFinalizations.resultId }) .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.runId, run.id))); + const unsettled = owner && ["observed", "retryable_failure"].includes(owner.phase) && owner.resultId === null; + const owned = owner?.leaseOwner && (!owner.leaseExpiresAt || owner.leaseExpiresAt > new Date()); if (!input.writersStopped || !reference || run.runtimeMode !== "native" || !run.issueId || !await isLegacyWorkspaceTaskRelated(db, { companyId: workspace.companyId, issueId: run.issueId, sourceIssueId: workspace.sourceIssueId }) || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status) || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date() - || !owner || owner.leaseOwner || !["committed", "terminal_failure"].includes(owner.phase)) reasons.push("remote_copyback_unresolved"); - else legacyRecoverySources.push({ runId: run.id, issueId: run.issueId, reference }); + || !owner || owned || !(["committed", "terminal_failure"].includes(owner.phase) || unsettled)) reasons.push("remote_copyback_unresolved"); + else legacyRecoverySources.push({ runId: run.id, issueId: run.issueId, reference, + ...(unsettled ? { unresolvedFinalization: { phase: owner.phase, leaseOwner: owner.leaseOwner, + leaseExpiresAt: owner.leaseExpiresAt?.toISOString() ?? null } } : {}) }); } - const [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + const finalizations = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId, phase: nativeRunFinalizations.phase, + leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + resultId: nativeRunFinalizations.resultId, run: heartbeatRuns }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.id, nativeRunFinalizations.runId))).where(and( eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.issueId, workspace.sourceIssueId), sql`${nativeRunFinalizations.phase} not in ('committed', 'terminal_failure')`, - )).limit(1) : []; - if (finalization) reasons.push("execution_finalization_unresolved"); + )) : []; + for (const row of finalizations) { + if (legacyRecoverySources.some(source => source.runId === row.id && source.unresolvedFinalization)) continue; + const stopped = input.writersStopped && row.run.runtimeMode === "native" + && ["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(row.run.status) + && !(row.run.controllerLeaseExpiresAt && row.run.controllerLeaseExpiresAt > new Date()) + && !(row.leaseOwner && (!row.leaseExpiresAt || row.leaseExpiresAt > new Date())); + const reference = readNativeWorkspaceSyncReference(row.run.runnerProfileJson?.nativeWorkspaceSync); + // Finalized copyback has no outstanding local writer. Assessment/status + // effects stay with their existing durable coordinator after the move. + if (stopped && reference?.state === "finalized") continue; + if (stopped && reference && reference.workspaceId !== workspace.id) { + const [other] = await db.select({ providerType: executionWorkspaces.providerType, cwd: executionWorkspaces.cwd, + sourceIssueId: executionWorkspaces.sourceIssueId, metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, workspace.companyId), eq(executionWorkspaces.id, reference.workspaceId))); + // The offline batch must migrate every local record before the app can + // start. Another workspace's coordinator is checked with its own source. + if (other && ((other.cwd && ["local_fs", "git_worktree"].includes(other.providerType)) || readRemoteWorkspaceAuthority(other.metadata)) + && await isLegacyWorkspaceTaskRelated(db, { companyId: workspace.companyId, issueId: row.run.nativeIssueId, sourceIssueId: other.sourceIssueId })) continue; + } + if (stopped && row.run.runnerProfileJson?.nativeWorkspaceSync == null && row.run.runnerProfileJson?.remoteWorkspace == null + && row.resultId === null && ["observed", "retryable_failure"].includes(row.phase)) { + legacyRecoverySources.push({ runId: row.id, issueId: row.run.nativeIssueId!, reference: null, + unresolvedFinalization: { phase: row.phase, leaseOwner: row.leaseOwner, + leaseExpiresAt: row.leaseExpiresAt?.toISOString() ?? null } }); + continue; + } + reasons.push("execution_finalization_unresolved"); + } + legacyRecoverySources.sort((a, b) => a.runId.localeCompare(b.runId)); const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), allowedOperation ? ne(workspaceOperations.id, allowedOperation) : undefined, @@ -267,8 +303,12 @@ export async function migrateRemoteWorkspace(input: { input.signal?.throwIfAborted(); await lock.execute(sql`select 1`); const publication = await input.db.transaction(async tx => { - for (const source of reserved.legacyRecoverySources) await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) - .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, source.runId))).for("update"); + for (const source of reserved.legacyRecoverySources) { + await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, source.runId))).for("update"); + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, source.runId))).for("update"); + } const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); const existing = readRemoteWorkspaceAuthority(workspace.metadata); @@ -298,6 +338,12 @@ export async function migrateRemoteWorkspace(input: { const metadata: Record = { ...workspace.metadata, remoteWorkspace: authority }; delete metadata.remoteWorkspaceMigration; if (reserved.legacyRecoverySources.length) { metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id }; + for (const source of reserved.legacyRecoverySources) await tx.update(heartbeatRuns).set({ + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ + legacyWorkspaceRecoveryMigration: { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, + operationId: operation.id, workspaceId: workspace.id }, + })}::jsonb`, updatedAt: new Date(), + }).where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, source.runId))); const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : []; metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])]; } diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 231896c716c..13dba6e5c60 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -372,11 +372,19 @@ export async function prepareRemoteWorkspace(input: { } const recoveredFiles: Array<{ operationId: string; sourceRunId: string; path: string }> = []; const recoverySourceRunIds = input.workspace.metadata?.remoteWorkspaceRecoverySourceRunIds; - if (!sameAllocation && Array.isArray(recoverySourceRunIds) && recoverySourceRunIds.length && input.workspace.sourceIssueId) { + if (!sameAllocation && Array.isArray(recoverySourceRunIds) && recoverySourceRunIds.length) { + // Shared and inherited workspaces retain their original source task. The + // current writer's task owns its recovery attachments, including a child's + // preserved files recorded during migration of an ancestor's workspace. + const [run] = await input.db.select({ nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId, + contextIssueId: sql`${heartbeatRuns.contextSnapshot}->'issueId'` }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.workspace.companyId), eq(heartbeatRuns.id, input.runId))); + const issueId = run?.nativeIssueId ?? run?.issueId ?? run?.contextIssueId; + if (typeof issueId !== "string") throw new Error("remote_workspace_recovery_task_missing"); const scratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-recovery-import-")); try { const recovered = await attachPreservedWorkspaceSnapshots({ db: input.db, companyId: input.workspace.companyId, - issueId: input.workspace.sourceIssueId, sourceRunIds: recoverySourceRunIds as string[], cwd: scratch }); + issueId, sourceRunIds: recoverySourceRunIds as string[], cwd: scratch }); for (const item of recovered) { const relative = `.paperclip-recovered/${item.operationId}`; await target.runner!.syncIn!([{ operationId: randomUUID(), files: [{ diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 182be104090..b19405fcfc0 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -195,7 +195,9 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; - it.each([false, true])("preserves an old sandbox after storage migration without writing its old host (removed=%s)", async removeHost => { + it.each(["retained", "removed", "child"])("preserves an old sandbox after storage migration without writing its old host (%s)", async mode => { + vi.stubEnv("DATABASE_URL", connectionString); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); const f = await seed(); // The fixture's command runner supports native copyback; the provider // snapshot boundary separately needs its real-file syncOut capability. @@ -209,14 +211,19 @@ describe("automatic retained workspace recovery", () => { await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; }; - await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId: f.issueId, + const sourceIssueId = mode === "child" ? randomUUID() : f.issueId; + if (mode === "child") { + await db.insert(issues).values({ id: sourceIssueId, companyId: f.companyId, title: "Ancestor workspace owner" }); + await db.update(issues).set({ parentId: sourceIssueId }).where(eq(issues.id, f.issueId)); + } + await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId, name: "Existing files", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.host }); await db.update(issues).set({ executionWorkspaceId: f.issueId }).where(eq(issues.id, f.issueId)); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); try { await migrateRemoteWorkspace({ db, storage: getStorageService(), companyId: f.companyId, workspaceId: f.issueId, apply: true, writersStopped: true }); - if (removeHost) await rm(f.host, { recursive: true }); + if (mode !== "retained") await rm(f.host, { recursive: true }); await requestWorkspaceRecovery({ ...f, idempotencyKey: randomUUID(), actorId: "board" }); await recoverRetainedWorkspace(f.input); expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); @@ -229,6 +236,29 @@ describe("automatic retained workspace recovery", () => { expect(await readFile(path.join(copies[0]!.path, "files", "saved.txt"), "utf8")).toBe("finished work\n"); expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + if (mode === "child") { + // The parent must not receive the child's task-scoped recovery copy. + expect(await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: sourceIssueId, + sourceRunIds: [f.runId], cwd: restored })).toHaveLength(0); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "false"); + const nextRunId = randomUUID(), nextLeaseId = randomUUID(), sourceRun = await run(f.runId); + await db.insert(heartbeatRuns).values({ id: nextRunId, companyId: f.companyId, agentId: sourceRun.agentId, + nativeIssueId: f.issueId, runtimeMode: "native", status: "running" }); + const storage = { ...getStorageService(), provider: "s3" as const }; + const reserved = await reserveRemoteWorkspace({ db, storage, companyId: f.companyId, issueId: f.issueId, + runId: nextRunId, agentId: sourceRun.agentId, environmentId: f.lease.environmentId!, + projectId: null, projectWorkspaceId: null, existingWorkspaceId: f.issueId, source: { kind: "empty" } }); + const [lease] = await db.insert(environmentLeases).values({ id: nextLeaseId, companyId: f.companyId, + issueId: f.issueId, heartbeatRunId: nextRunId, environmentId: f.lease.environmentId, + provider: "daytona", providerLeaseId: randomUUID(), status: "active" }).returning(); + const nextRoot = path.join(root, nextRunId); await mkdir(nextRoot); + const target = { ...targets.get(f.lease.id)!, leaseId: nextLeaseId, remoteCwd: await realpath(nextRoot) }; + const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, + runId: nextRunId, lease: lease as EnvironmentLease, target }); + expect(prepared.workspace.sourceIssueId).toBe(sourceIssueId); + expect(await readFile(path.join(prepared.remoteRoot, ".paperclip-recovered", copies[0]!.operationId, "files", "saved.txt"), "utf8")) + .toBe("finished work\n"); + } } finally { vi.unstubAllEnvs(); } }, 30_000);