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 9bf24ecb59c..0f028d6a037 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -45,3 +45,60 @@ yet make the whole tenant eligible for enrollment. bindings, archived sources, overlap/writer races, and unavailable sources. - Unrelated tenants, production services, and agent task execution remain outside the migration's mutation scope. + +## Legacy recovery storage transition + +The migration owner must distinguish file location from recovery completion. +An offline import can preserve a terminal failed run's local workspace without +claiming that its remote files were recovered. The snapshot and migration +receipt bind the original run and sync descriptor; the failed run, provider +allocation, recovery action, and unresolved state stay recorded. + +After publication, the native sync owner must reject copyback to the old local +folder. Explicit recovery still validates the exact provider allocation and +stop evidence, then preserves remote files as an immutable recovery attachment +instead of merging into retired local authority. New remote allocations attach +those verified recovery copies through the existing attachment owner. Missing +or ambiguous remote sources remain unresolved; migration never authorizes +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 phase, checkpoint, and unresolved copyback remain +intact. Publication journals the original coordinator binding, releases only +that exact expired lease with a recovery-history receipt, 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. +After independently verified sandbox preservation, the task accepts the exact +migrated snapshot receipt for continuation without requiring a fresh workspace. +For older runs without a synchronization descriptor, the migration journals +all declared recovery lease/allocation/directory identities under row locks. +Continuation requires the unchanged complete set and a verified snapshot for +every source. Missing or ambiguous source identity remains held; a partial +archive cannot certify recovery. The recovery journey covers two allocations, +removal of the old filesystem, and the files appearing in the next remote turn. +Accepted results with unfinished copyback remain blockers. Once the original +sync reference is finalized and its successful `workspace_finalize` operation +receipt exists, assessment and status effects can continue from their +existing database coordinator independently of the file storage migration. + +Offline maintenance stops writers; it does not silently change workspace +semantics or retire task ownership. `parallel_shared_writers_unsupported` needs +an operator-approved transition to serialized remote workspace writes. +`task_writer_bound` needs the task lifecycle owner to settle its exact checkout +and execution binding, even if the run row is terminal. `task_binding_required` +needs provenance recovery or a separately approved durable archival path for an +orphaned workspace; the importer never invents a task owner. These records stay +on the original volume and block detach. The batch reports its first failed +workspace through the phase/error receipt; it does not claim to aggregate every +possible blocker in advance or skip blocked rows. + +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 +recovery source validation and company boundaries. diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index 146e551f358..aa703738606 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -17,7 +17,7 @@ import { hasVerifiedWorkspaceRepair, workspaceRecoveryDisplay } from "./workspac const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; -const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace"] as const; +const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration"] as const; const blockerContextColumn = sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null else ( select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb) diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index ba0d2fa8353..b4e3313667a 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -272,7 +272,7 @@ function preserveWorkspaceMetadataForPatch( if (patch[key] !== undefined && patch[key] !== current[key]) throw conflict("Remote workspace coordinates belong to its sandbox storage"); } } - const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles"] as const; + const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles", "legacyWorkspaceRecoveryMigration"] as const; if (patch.metadata !== undefined && remoteAuthorityKeys.some(key => current.metadata?.[key] !== undefined || patch.metadata?.[key] !== undefined)) { nextPatch.metadata = { ...patch.metadata }; for (const key of remoteAuthorityKeys) { diff --git a/server/src/services/explicit-native-continuation.ts b/server/src/services/explicit-native-continuation.ts index b45214afd1c..a153e4202b2 100644 --- a/server/src/services/explicit-native-continuation.ts +++ b/server/src/services/explicit-native-continuation.ts @@ -6,6 +6,8 @@ import { readTaskRetryIntent } from "./task-retry.js"; import { readSystemRecovery } from "./system-recovery-state.js"; import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js"; import { hasVerifiedWorkspaceRepair } from "./workspace-recovery-receipt.js"; +import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { preserveNativeFailedWorkspace, preserveTaskRemoteWorkspaceWriter } from "./native-runtime/native-failed-workspace-recovery.js"; import { createHash } from "node:crypto"; import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; @@ -380,8 +382,19 @@ export async function admitExplicitNativeContinuation(input: { // cleanup failed. That immutable result is history, not an active commit. // Keep it intact and require the same controller/process/lease stop proofs // before admitting new user input; never apply or replay the old result. + const legacyReference = readNativeWorkspaceSyncReference(run.runnerProfileJson?.nativeWorkspaceSync); + const migratedSource = coordinator && !coordinator.resultId && !coordinator.leaseOwner + && ["observed", "retryable_failure"].includes(coordinator.phase) + && await hasVerifiedWorkspaceRepair(db, run) + ? await readMigratedLegacyWorkspaceSource(db, { companyId, runId: run.id, reference: legacyReference }) : null; + // Offline migration fenced this exact unfinished turn and preserved its + // coordinator. A verified attachment permits new user input without + // inventing a terminal result or replaying the old session. All stop and + // ownership checks below still apply. + const preservedFailure = migratedSource?.issueId === issueId + && migratedSource.unresolvedFinalization?.phase === coordinator?.phase; if (!cancelledStartup && coordinator && ( - (coordinator.phase !== "terminal_failure" && !committedFailure) || coordinator.leaseOwner || + (coordinator.phase !== "terminal_failure" && !committedFailure && !preservedFailure) || coordinator.leaseOwner || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", run.status === "cancelled" && !coordinator.leaseOwner ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details." 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 new file mode 100644 index 00000000000..8d014facfe4 --- /dev/null +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -0,0 +1,109 @@ +import { isDeepStrictEqual } from "node:util"; +import { and, eq, sql } from "drizzle-orm"; +import { environmentLeases, 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 | null; + recoveryAllocations?: { leaseId: string; providerLeaseId: string; remoteCwd: string }[] | 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: { + companyId: string; issueId: string | null; sourceIssueId: string | null; +}) { + let current = input.issueId; + const visited = new Set(); + while (current && visited.size < 32 && !visited.has(current)) { + if (current === input.sourceIssueId) return true; + visited.add(current); + const [issue] = await db.select({ parentId: issues.parentId }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, current))); + current = issue?.parentId ?? null; + } + return false; +} + +/** Record every declared source, or none: partial preservation cannot release + * the task's recovery hold. The recovery owner separately verifies provenance, + * provider termination and archive bytes before writing its receipts. */ +export async function readLegacyWorkspaceRecoveryAllocations(db: Pick, run: { + id: string; companyId: string; nativeIssueId: string | null; resultJson: Record | null; +}): Promise { + const recovery = run.resultJson?.workspaceRestoreRecovery as { schema?: unknown; leaseIds?: unknown } | undefined; + const ids = recovery?.leaseIds; + if (recovery?.schema !== "paperclip.workspace-restore-recovery.v1" || !Array.isArray(ids) || !ids.length + || ids.some(id => typeof id !== "string" || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(id)) + || new Set(ids).size !== ids.length || !run.nativeIssueId) return null; + const allocations = []; + for (const id of [...ids].sort()) { + const [lease] = await db.select({ providerLeaseId: environmentLeases.providerLeaseId, metadata: environmentLeases.metadata }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.id, id), + eq(environmentLeases.heartbeatRunId, run.id), eq(environmentLeases.issueId, run.nativeIssueId))); + if (!lease?.providerLeaseId || typeof lease.metadata?.remoteCwd !== "string" || !lease.metadata.remoteCwd.startsWith("/")) return null; + allocations.push({ leaseId: id as string, providerLeaseId: lease.providerLeaseId, remoteCwd: lease.metadata.remoteCwd }); + } + return allocations; +} + +/** This receipt changes where recovery may save files; it never certifies that + * a provider stopped or that its files were recovered. The recovery owner must + * still validate allocation provenance and termination before exporting. */ +export async function readMigratedLegacyWorkspaceSource(db: Pick, input: { + companyId: string; runId: string; reference: NativeWorkspaceSyncReference | null; +}): Promise { + if (input.reference && input.reference.state !== "prepared") return null; + const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, + nativeIssueId: heartbeatRuns.nativeIssueId, + resultJson: sql>`jsonb_build_object('workspaceRestoreRecovery', ${heartbeatRuns.resultJson}->'workspaceRestoreRecovery')`, + runnerProfileJson: sql>`jsonb_build_object( + 'legacyWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration', + 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync', + 'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace')`, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), + )); + const runMarker = run?.runnerProfileJson?.legacyWorkspaceRecoveryMigration as { workspaceId?: unknown } | undefined; + const workspaceId = input.reference?.workspaceId ?? runMarker?.workspaceId; + if (typeof workspaceId !== "string") return null; + const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, workspaceId), + )); + if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return null; + const marker = workspace?.metadata?.legacyWorkspaceRecoveryMigration as { schema?: unknown; operationId?: unknown } | undefined; + if (marker?.schema !== LEGACY_WORKSPACE_RECOVERY_MIGRATION || typeof marker.operationId !== "string") + throw new Error("legacy_workspace_migration_receipt_invalid"); + if (!isDeepStrictEqual(runMarker, { + schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: marker.operationId, workspaceId, + })) throw new Error("legacy_workspace_migration_receipt_invalid"); + const [operation] = await db.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.executionWorkspaceId, workspaceId), + eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"), + eq(workspaceOperations.status, "succeeded"), + )); + const sources = operation?.metadata?.legacyRecoverySources; + const source = Array.isArray(sources) ? sources.find(source => source && typeof source === "object" + && source.runId === input.runId && typeof source.issueId === "string" && isDeepStrictEqual(source.reference, input.reference)) : null; + if (source && source.issueId !== run?.nativeIssueId) throw new Error("legacy_workspace_migration_receipt_invalid"); + if (source && !input.reference && (run?.runnerProfileJson?.nativeWorkspaceSync != null + || run?.runnerProfileJson?.remoteWorkspace != null + || !isDeepStrictEqual(source.recoveryAllocations ?? null, run ? await readLegacyWorkspaceRecoveryAllocations(db, run) : null))) + throw new Error("legacy_workspace_migration_receipt_invalid"); + if (!source) throw new Error("legacy_workspace_migration_receipt_invalid"); + return source as LegacyWorkspaceRecoverySource; +} + +export async function hasMigratedLegacyWorkspaceSource(...args: Parameters) { + return Boolean(await readMigratedLegacyWorkspaceSource(...args)); +} 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/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index aa5a0c76e2a..a684714b53a 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -1,3 +1,4 @@ +import { hasMigratedLegacyWorkspaceSource } from "../legacy-workspace-recovery-migration.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; @@ -921,7 +922,7 @@ export async function prepareNativeWorkspaceSync(input: { throw new Error("native_remote_recovery_lease_mismatch"); } const run = await measureRunPhase("workspace_sync.run_read", () => input.db - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) @@ -1184,7 +1185,7 @@ export async function resumeNativeWorkspaceSync(input: { throw new Error("workspace_sync_out_unrecoverable"); } const run = await input.db - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) @@ -1193,6 +1194,8 @@ export async function resumeNativeWorkspaceSync(input: { parseObject(run?.runnerProfileJson).nativeWorkspaceSync, ); if (!reference) return false; + if (run && await hasMigratedLegacyWorkspaceSource(input.db, { companyId: run.companyId, runId: input.runId, reference })) + throw new Error("native_workspace_sync_source_migrated"); const existing = await readDescriptor({ runId: input.runId, reference }); const providerLeaseId = input.target.sandboxLeaseAcquisition?.providerLeaseId ?? diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index bce64e9e66d..8b75b7be4e1 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, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, +import { activityLog, agents, environmentLeases, 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"; @@ -25,6 +25,10 @@ import type { TaskDirectoryRestorePointRef } from "./isolated-task-directory-ret 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 { attemptNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalizer.js"; import { managedProjectCheckoutName } from "./managed-project-checkout-name.js"; import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js"; @@ -51,7 +55,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(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() { @@ -140,6 +145,182 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: reference!.manifestDigest }); expect(await fs.readFile(path.join(targetDir, ".paperclip/worktrees/child/child.txt"), "utf8")).toBe("saved child edits\n"); }); + async function failedCopybackFixture() { + const f = await seed(), runId = randomUUID(); + 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" }; + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: agent!.id, nativeIssueId: f.issueId, + runtimeMode: "native", status: "failed", finishedAt: new Date(), runnerProfileJson: { nativeWorkspaceSync: reference } }); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId, phase: "terminal_failure" }); + return { ...f, runId, reference }; + } + it("carries unresolved copyback forward during offline migration without declaring recovery complete", async () => { + const f = await failedCopybackFixture(); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_copyback_unresolved"); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + const row = await workspace(f.workspaceId); + expect(readRemoteWorkspaceAuthority(row.metadata)?.snapshotId).toBeTruthy(); + expect(row.metadata?.remoteWorkspaceRecoverySourceRunIds).toEqual([f.runId]); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run).toMatchObject({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: f.reference } }); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + // An old copyback must not recreate or mutate the predecessor's local folder. + 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", "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 === "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)); + } + if (mode === "descriptor") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, descriptorSha256: "invalid" } } }).where(eq(heartbeatRuns.id, f.runId)); + 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([]); + const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + if (before!.leaseOwner) { + expect(after).toMatchObject({ ...before, leaseOwner: null, leaseExpiresAt: null, updatedAt: expect.any(Date), + recoveryHistory: [{ event: "legacy_workspace_migration_owner_released", workspaceId: f.workspaceId, + operationId: expect.any(String), previousOwner: { phase, leaseOwner: before!.leaseOwner, + leaseExpiresAt: before!.leaseExpiresAt!.toISOString() } }] }); + } else expect(after).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", "missing-receipt", "failed-receipt"])("keeps accepted-result finalization independent of migrated file storage (%s)", async mode => { + const state = mode === "prepared" ? "prepared" : "finalized"; + 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 operationId = randomUUID(); + if (mode === "finalized" || mode === "failed-receipt") await db.insert(workspaceOperations).values({ id: operationId, + companyId: f.companyId, executionWorkspaceId: f.workspaceId, heartbeatRunId: f.runId, issueId: f.issueId, + phase: "workspace_finalize", status: mode === "finalized" ? "succeeded" : "failed" }); + const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const migrate = () => migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + if (mode !== "finalized") await expect(migrate()).rejects.toThrow(state === "prepared" + ? "remote_copyback_unresolved" : "execution_finalization_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 attemptNativeWorkspaceFinalization({ db, runId: f.runId })) + .toMatchObject({ kind: "completed", operation: { id: operationId } }); + } + 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); + storage.putFile = async input => { + const result = await put(input); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, providerLeaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId)); + return result; + }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_workspace_migration_source_changed"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("rejects a descriptor-free allocation changed during upload", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await db.insert(environmentLeases).values({ id: f.reference.leaseId, companyId: f.companyId, + issueId: f.issueId, heartbeatRunId: f.runId, provider: "daytona", providerLeaseId: f.reference.providerLeaseId, + status: "released", metadata: { remoteCwd: f.reference.remoteCwd } }); + await db.update(heartbeatRuns).set({ runnerProfileJson: {}, resultJson: { + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.reference.leaseId] }, + } }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "observed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const put = storage.putFile.bind(storage); + storage.putFile = async input => { + const result = await put(input); + await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId)); + return result; + }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })) + .rejects.toThrow("remote_workspace_migration_source_changed"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("does not release a coordinator whose expired owner changed during upload", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: "journaled-owner", + leaseExpiresAt: new Date(Date.now() - 60_000) }).where(eq(nativeRunFinalizations.runId, f.runId)); + const put = storage.putFile.bind(storage); + storage.putFile = async input => { + const result = await put(input); + await db.update(nativeRunFinalizations).set({ leaseOwner: "different-owner" }).where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })) + .rejects.toThrow("remote_workspace_migration_source_changed"); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]) + .toMatchObject({ phase: "observed", leaseOwner: "different-owner", recoveryHistory: [] }); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("preserves a child task's recovery reference to its ancestor's workspace", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Parent task" }).returning(); + await db.update(issues).set({ parentId: parent!.id }).where(eq(issues.id, f.issueId)); + await db.update(executionWorkspaces).set({ sourceIssueId: parent!.id }).where(eq(executionWorkspaces.id, f.workspaceId)); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + expect((await workspace(f.workspaceId)).sourceIssueId).toBe(parent!.id); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.executionWorkspaceId).toBe(f.workspaceId); + await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) + .rejects.toThrow("native_workspace_sync_source_migrated"); + }); + it("fails closed if the durable legacy migration receipt changes", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + const row = await workspace(f.workspaceId); + await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, legacyWorkspaceRecoveryMigration: { operationId: randomUUID() } } }).where(eq(executionWorkspaces.id, f.workspaceId)); + await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) + .rejects.toThrow("legacy_workspace_migration_receipt_invalid"); + }); it("blocks distinct legacy records that alias the same folder", async () => { const f = await seed(), id = randomUUID(); await db.insert(executionWorkspaces).values({ id, companyId: f.companyId, name: "Historical alias", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.cwd }); @@ -536,6 +717,20 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_operation_unresolved"); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); }); + it.each(["parallel", "orphan", "stale-binding"])("keeps offline enrollment blocked on unresolved workspace semantics (%s)", async mode => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await db.update(nativeRunFinalizations).set({ phase: "committed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: null }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "parallel") { + await db.update(executionWorkspaces).set({ mode: "shared_workspace" }).where(eq(executionWorkspaces.id, f.workspaceId)); + await db.update(issues).set({ executionWorkspaceSettings: { sharedWorkspaceConcurrency: "allow" } }).where(eq(issues.id, f.issueId)); + } else if (mode === "orphan") await db.update(executionWorkspaces).set({ sourceIssueId: null }).where(eq(executionWorkspaces.id, f.workspaceId)); + else await db.update(issues).set({ executionRunId: f.runId }).where(eq(issues.id, f.issueId)); + const reason = mode === "parallel" ? "parallel_shared_writers_unsupported" : mode === "orphan" ? "task_binding_required" : "task_writer_bound"; + await expect(migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).rejects.toThrow(reason); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + }); it.each(["running", "starting", "unknown", "failed"])("blocks a %s local service after its control lease expires", async status => { const f = await seed(), serviceId = randomUUID(); await db.insert(executionWorkspaceRuntimeLeases).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId, diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 91cdc407c22..7ba8b29a79c 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -4,7 +4,7 @@ import os from "node:os"; import path from "node:path"; import { and, eq, inArray, ne, or, sql } from "drizzle-orm"; import { - executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaceSnapshots, executionWorkspaces, + environmentLeases, executionWorkspaceRepositories, executionWorkspaceRuntimeLeases, executionWorkspaceSnapshots, executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, projects, workspaceOperations, workspaceRuntimeServices, type Db, } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, remoteWorkspaceSourceSchema, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared"; @@ -17,7 +17,9 @@ import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-rete import { materializeTaskDirectorySnapshot, removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; import { loadTaskDirectoryRestorePoint, mayBeIsolatedTaskDirectoryWorkspace, readTaskDirectoryRetention, validateTaskDirectoryRestorePointRef } from "./isolated-task-directory-retention.js"; import { inspectArchivedGitMigrationSource, materializeArchivedGitMigrationSource, type ArchivedGitMigrationSource } from "./remote-workspace-migration-git.js"; -import { prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { isLegacyWorkspaceTaskRelated, readLegacyWorkspaceRecoveryAllocations, LEGACY_WORKSPACE_RECOVERY_MIGRATION, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js"; +import { offlineWorkspaceMigrationBlockers, prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; const SCHEMA = "paperclip.remote-workspace-migration.v1"; type Workspace = typeof executionWorkspaces.$inferSelect; @@ -46,7 +48,8 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri const marker = object(workspace.metadata?.remoteWorkspaceMigration); const allowedOperation = input.operationId ?? (marker.schema === SCHEMA && typeof marker.operationId === "string" ? marker.operationId : undefined); const reasons: string[] = []; - if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null }; + if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] }; + if (input.writersStopped) reasons.push(...await offlineWorkspaceMigrationBlockers(db)); if (!["local_fs", "git_worktree"].includes(workspace.providerType)) reasons.push("workspace_provider_unsupported"); if (!["active", "idle", "in_review", "archived"].includes(workspace.status)) reasons.push("workspace_not_active"); if (!workspace.sourceIssueId) reasons.push("task_binding_required"); @@ -96,16 +99,76 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri or ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id})`, )).limit(1); if (run) reasons.push("active_execution"); - const [unexported] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId), + const unexported = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + issueId: heartbeatRuns.nativeIssueId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + reference: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId), sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id}`, sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared'`, - )).limit(1); - if (unexported) reasons.push("remote_copyback_unresolved"); - const [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( + )); + 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, + 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 || 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 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); + // The finalizer retries from its operation receipt, not the sync descriptor. + // Without that receipt it would try exporting the newly migrated workspace, + // which intentionally has no live location. Keep the local authority until + // the existing owner records completion; then assessment/status can resume. + if (stopped && reference?.state === "finalized") { + const [finalized] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.heartbeatRunId, row.id), + eq(workspaceOperations.issueId, row.run.nativeIssueId!), eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "succeeded"), + )).limit(1); + if (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, + recoveryAllocations: await readLegacyWorkspaceRecoveryAllocations(db, row.run), + 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, @@ -129,11 +192,14 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ) : undefined), )).limit(1); if (service) reasons.push("workspace_runtime_service_unsettled"); - const [recovery] = workspace.sourceIssueId ? await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, workspace.companyId), eq(issueRecoveryActions.sourceIssueId, workspace.sourceIssueId), + const recoveryIssueIds = [...new Set([workspace.sourceIssueId, ...legacyRecoverySources.map(source => source.issueId)].filter((id): id is string => Boolean(id)))]; + const recoveries = recoveryIssueIds.length ? await db.select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, workspace.companyId), inArray(issueRecoveryActions.sourceIssueId, recoveryIssueIds), inArray(issueRecoveryActions.status, ["active", "escalated"]), - )).limit(1) : []; - if (recovery) reasons.push("workspace_recovery_unresolved"); + )) : []; + if (recoveries.some(recovery => !input.writersStopped + || !legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId) + || (recovery.evidence?.systemRecovery as { state?: unknown } | undefined)?.state === "running")) reasons.push("workspace_recovery_unresolved"); const repositories = await db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, workspace.companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id))); if (repositories.some(repository => repository.state !== "ready")) reasons.push("repository_preparation_unresolved"); @@ -141,7 +207,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ref: archivedGitSource?.commit ?? workspace.baseRef, branchName: workspace.branchName } : { kind: "empty" }; if (repositories.length) source.repositories = repositories.map(repository => ({ id: repository.id, relativePath: repository.relativePath, repoUrl: repository.repoUrl, ref: repository.requestedRef, ...(repository.pinnedCommit ? { pinnedCommit: repository.pinnedCommit } : {}) })); - return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource }; + return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources }; } export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) { @@ -164,7 +230,7 @@ export async function migrateRemoteWorkspace(input: { if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); if (workspace.sourceIssueId) await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, workspace.companyId), eq(issues.id, workspace.sourceIssueId))).for("update"); const current = await inspect(tx as unknown as Db, input); - if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null }; + if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] }; if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source); const sourceFingerprint = fingerprint(workspace); @@ -181,12 +247,14 @@ export async function migrateRemoteWorkspace(input: { if (issueBinding !== current.issueBinding) throw new Error("remote_workspace_migration_source_changed"); if (Boolean(operation.metadata?.retainedDirectory) !== Boolean(current.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.archivedGitSource ?? null) !== canonical(gitSourceProof(current.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); - return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource }; + if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); + return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources }; } const [operation] = await tx.insert(workspaceOperations).values({ companyId: workspace.companyId, executionWorkspaceId: workspace.id, issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding, nestedWorkspaceScope: { version: 1, state: "pending" }, + ...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}), ...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}), ...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id, generation: current.retainedDirectory.generation, digest: current.retainedDirectory.restorePoint.digest } } : {}) } }).returning(); @@ -194,7 +262,7 @@ export async function migrateRemoteWorkspace(input: { await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceMigration: { schema: SCHEMA, operationId: operation!.id, state: "running", sourceFingerprint, source, } }, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); - return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource }; + return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources }; }); if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true }; let operation = reserved.operation!; @@ -245,6 +313,15 @@ 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: 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"); + for (const allocation of source.recoveryAllocations ?? []) await tx.select({ id: environmentLeases.id }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.id, allocation.leaseId))).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); @@ -268,9 +345,36 @@ export async function migrateRemoteWorkspace(input: { if (!saved || !["committed", "superseded"].includes(saved.status) || saved.manifestDigest !== child.manifestDigest || saved.sourceOperationId !== child.operationId) throw new Error("nested_workspace_snapshot_changed"); } + if (canonical(validation.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); const authority: RemoteWorkspaceAuthority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: ref.id, source: reserved.source, location: null, writerRunId: null, dirty: false }; 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) { + // The offline deployment fence has stopped the old controller. The + // locked validation above must still match its journaled expired + // lease exactly. Release only that coordinator lease so explicit + // file recovery can proceed; its phase/checkpoint and the provider's + // independent stop obligation remain unchanged. + if (source.unresolvedFinalization?.leaseOwner) await tx.update(nativeRunFinalizations).set({ + leaseOwner: null, leaseExpiresAt: null, + recoveryHistory: sql`${nativeRunFinalizations.recoveryHistory} || ${JSON.stringify([{ + event: "legacy_workspace_migration_owner_released", operationId: operation.id, + workspaceId: workspace.id, previousOwner: source.unresolvedFinalization, + }])}::jsonb`, updatedAt: new Date(), + }).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), + eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId))); + 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)])]; + } await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); await tx.update(executionWorkspaceSnapshots).set({ status: "committed", deleteAfter: null, metadata: { ...uploaded.metadata, ...(nested.length ? { importedWorkspaceChildren: nested } : {}) }, diff --git a/server/src/services/remote-workspaces.ts b/server/src/services/remote-workspaces.ts index 6c7cdee8dab..18a31f71f5a 100644 --- a/server/src/services/remote-workspaces.ts +++ b/server/src/services/remote-workspaces.ts @@ -394,11 +394,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-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index 0cb51ec397c..be127ec9798 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -1,8 +1,10 @@ -import { and, eq, sql } from "drizzle-orm"; +import { and, desc, eq, sql } from "drizzle-orm"; import { executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type ExecutionBlocker } from "@paperclipai/shared"; import { z } from "zod"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; +import { readSnapshot } from "./workspace-recovery-snapshot.js"; export const workspaceRecoverySchema = z.object({ schema: z.literal("paperclip.workspace-recovery.v1"), @@ -43,6 +45,39 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick>'recoveryActionId'`, + )).where(and(eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId), + eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_recovery_snapshot"), + eq(workspaceOperations.status, "succeeded"), + sql`${workspaceOperations.metadata}->>'leaseId' = ${allocation.leaseId}`, + sql`${workspaceOperations.metadata}->>'providerLeaseId' = ${allocation.providerLeaseId}`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,runId}' = ${run.id}`, + sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`, + )).orderBy(desc(workspaceOperations.createdAt)).limit(1); + if (!saved || readWorkspaceRecovery(saved.evidence)?.status !== "recovered") return null; + try { readSnapshot(saved.metadata, run.companyId, run.id); } catch { return null; } + receiptId = saved.id; + } + return receiptId; + } + } if (typeof issueId !== "string" || (!remote && (!reference || reference.state !== "finalized" || !reference.finalHostSha256))) return null; const [receipt] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and( eq(workspaceOperations.companyId, run.companyId), eq(workspaceOperations.issueId, issueId), diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index ad65d438d2b..87b27c4409d 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -1,3 +1,4 @@ +import { hasMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; import { retiredReusableLeasePredecessors } from "./environment-lease-handoff.js"; import { and, eq, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; @@ -211,7 +212,10 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec } const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId)); if (!environment) throw new SystemRecoveryFailure("source_unavailable", "service_required"); - sources.push({ lease, environment, reference: legacyDescriptor, remoteReference: remoteDescriptor, + const preservationOnly = legacyDescriptor ? await hasMigratedLegacyWorkspaceSource(db, { + companyId: input.companyId, runId: run.id, reference: legacyDescriptor, + }) : false; + sources.push({ lease, environment, reference: legacyDescriptor, preservationOnly, remoteReference: remoteDescriptor, remoteWorkspace: remoteDescriptor ? remoteWorkspace : null, remoteCwd }); } return { run, action, sources }; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 32b57491fa3..7fa4babe6b5 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -33,6 +33,7 @@ import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspac import { readSystemRecovery } from "./system-recovery-state.js"; import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; import { getStorageService } from "../storage/index.js"; +import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; @@ -183,7 +184,7 @@ describe("automatic retained workspace recovery", () => { ownerType: "board", cause: mode === "native" ? "native_continuation_requires_reconciliation" : "legacy_execution_requires_reconciliation", fingerprint: runId, nextAction: "Manual repair required", evidence: { runId, workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery } }); const runtime = { resumeRunLease: vi.fn(async (_input: { lease: EnvironmentLease }) => ({ providerLeaseId, metadata: { remoteCwd: remote } })), - retryPendingSandboxTeardown: vi.fn(async () => ({ providerLeaseId, state: "stopped" })) }; + retryPendingSandboxTeardown: vi.fn(async (_input: { lease: EnvironmentLease }) => ({ providerLeaseId, state: "stopped" })) }; const environmentRuntime = environmentRuntimeService(db, { drivers: [{ driver: "sandbox", acquireRunLease: vi.fn(), releaseRunLease: vi.fn(), ...runtime, resolveCapabilities: async () => classifyEnvironmentCapabilities({ verifiedMethods: [], declared: null, narrowing: null }), @@ -194,6 +195,150 @@ 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(["retained", "removed", "child", "expired-owner", "no-descriptor"])("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("native", mode !== "no-descriptor"); + // The fixture's command runner supports native copyback; the provider + // snapshot boundary separately needs its real-file syncOut capability. + targets.get(f.lease.id)!.runner!.syncIn = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + targets.get(f.lease.id)!.runner!.syncOut = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + if (mode === "no-descriptor") { + const secondId = randomUUID(), secondProviderId = randomUUID(), secondRemote = path.join(root, secondId); + await cp(f.remote, secondRemote, { recursive: true }); + const second: typeof f.lease = { ...f.lease, id: secondId, providerLeaseId: secondProviderId, + metadata: { ...f.lease.metadata, remoteCwd: secondRemote } }; + second.metadata = { ...second.metadata, + remoteExecutionTermination: remoteTerminationReceipt(second, { providerLeaseId: secondProviderId, state: "stopped" }) }; + await db.insert(environmentLeases).values(second); + targets.set(secondId, { ...targets.get(f.lease.id)!, leaseId: secondId, remoteCwd: secondRemote }); + f.runtime.retryPendingSandboxTeardown.mockImplementation(async ({ lease }) => ({ + providerLeaseId: lease.providerLeaseId!, state: "stopped", + })); + f.runtime.resumeRunLease.mockImplementation(async ({ lease }) => ({ providerLeaseId: lease.providerLeaseId!, + metadata: { remoteCwd: lease.metadata!.remoteCwd as string } })); + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id, secondId] }, + } }).where(eq(heartbeatRuns.id, f.runId)); + } + 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)); + if (mode === "expired-owner" || mode === "no-descriptor") await db.update(nativeRunFinalizations).set({ phase: "observed", + leaseOwner: "stopped-old-controller", leaseExpiresAt: new Date(Date.now() - 60_000) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + try { + await migrateRemoteWorkspace({ db, storage: getStorageService(), companyId: f.companyId, + workspaceId: f.issueId, apply: true, writersStopped: 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"); + expect(await readFile(path.join(f.host, "saved.txt")).catch(() => null)).toBeNull(); + expect(await run(f.runId)).toMatchObject({ status: "failed", runnerProfileJson: mode === "no-descriptor" ? {} : { nativeWorkspaceSync: { state: "prepared" } } }); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ workspaceRepairRequired: false }); + const restored = path.join(root, randomUUID()); await mkdir(restored); + const copies = await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId, + sourceRunIds: [f.runId], cwd: restored }); + expect(copies).toHaveLength(mode === "no-descriptor" ? 2 : 1); + 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 === "no-descriptor") { + const source = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...source.resultJson, + workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }, + } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + await db.update(heartbeatRuns).set({ resultJson: source.resultJson }).where(eq(heartbeatRuns.id, f.runId)); + for (const copy of copies) { + await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.id, copy.operationId)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, copy.operationId)); + } + const [allocation] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)); + await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.lease.id)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + await db.update(environmentLeases).set({ providerLeaseId: allocation.providerLeaseId }).where(eq(environmentLeases.id, f.lease.id)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + } + if (mode === "child" || mode === "expired-owner" || mode === "no-descriptor") { + // The parent must not receive the child's task-scoped recovery copy. + if (mode === "child") 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); + const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, + authorType: "user", authorUserId: "operator", body: "/new" }).returning(); + const continuationInput = { db, companyId: f.companyId, issueId: f.issueId, + agentId: sourceRun.agentId, actorType: "user", actorId: "operator", reason: "issue_commented", + commentId: comment.id, successorRunId: nextRunId }; + if (mode === "expired-owner" || mode === "no-descriptor") { + for (const changed of [{ phase: "retryable_failure", leaseOwner: null }, { phase: "observed", leaseOwner: "new-controller" }]) { + await db.update(nativeRunFinalizations).set(changed).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull(); + } + await db.update(nativeRunFinalizations).set({ phase: "observed", leaseOwner: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...sourceRun.runnerProfileJson, + legacyWorkspaceRecoveryMigration: { invalid: true } } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await admitExplicitNativeContinuation({ ...continuationInput, dryRun: true })).toBeNull(); + await db.update(heartbeatRuns).set({ runnerProfileJson: sourceRun.runnerProfileJson }).where(eq(heartbeatRuns.id, f.runId)); + } + expect(await admitExplicitNativeContinuation(continuationInput)).toMatchObject({ previousRunId: f.runId, commentId: comment.id }); + 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"); + } + if (mode === "retained" || mode === "no-descriptor") { + const [receipt] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.id, copies[0]!.operationId)); + for (const metadata of [{ ...receipt.metadata, leaseId: randomUUID() }, { ...receipt.metadata, providerLeaseId: randomUUID() }, + { ...receipt.metadata, recoveryActionId: randomUUID() }, { ...receipt.metadata, archive: { broken: true } }]) { + await db.update(workspaceOperations).set({ metadata }).where(eq(workspaceOperations.id, receipt.id)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + } + await db.update(workspaceOperations).set({ metadata: receipt.metadata, status: "running" }).where(eq(workspaceOperations.id, receipt.id)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(false); + await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, receipt.id)); + expect(await hasVerifiedWorkspaceRepair(db, await run(f.runId))).toBe(true); + } + } finally { + // This journey admits a successor; retire that fixture writer before the + // next case exercises the instance-wide offline migration fence. + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(and( + eq(heartbeatRuns.companyId, f.companyId), inArray(heartbeatRuns.status, ["queued", "running"]))); + vi.unstubAllEnvs(); + } + }, 30_000); + function boardRecoveryApp() { const app = express(); app.use(express.json()); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index cdadb3efba4..ad1d8ab4f6c 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -206,7 +206,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ continue; } } - let reference = source.reference; + let reference = source.preservationOnly ? null : source.reference; if (reference) { try { await readNativeWorkspaceSyncLocalCwd({ runId: input.runId, companyId: input.companyId, workspaceId: reference.workspaceId, reference }); }