diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index 6ea398b49ae..5f5f643027d 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -637,13 +637,15 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { branchName, strategyType: "git_worktree", providerType: "git_worktree", metadata: { createdByRuntime: true, gitBranchOwnershipVersion: 1 } }).where(eq(executionWorkspaces.id, f.workspaceId)); return { ...f, cwd, base, upstream, commit, projectId, projectWorkspaceId, repoUrl, branchName }; } - async function archivedPreparedFixture(sourceKind: "recipe" | "cleanup" | "branch" = "recipe") { + async function archivedPreparedFixture(sourceKind: "recipe" | "cleanup" | "branch" = "recipe", + beforeArchive?: (f: Awaited>) => Promise) { const base = await gitArchiveFixture(), f = { ...base, ...await failedCopybackFixture(base) }; let cleanup: typeof workspaceOperations.$inferSelect | null = null; if (sourceKind === "recipe") { await fs.writeFile(path.join(f.cwd, "saved.txt"), "archived staged edit\n"); await git(f.cwd, "add", "saved.txt"); await fs.writeFile(path.join(f.cwd, "saved.txt"), "archived unstaged edit\n"); await fs.writeFile(path.join(f.cwd, "untracked.txt"), "archived private file\n"); + await beforeArchive?.(f); const captured = (await captureRestorePoint({ cwd: f.cwd, upstreamCommits: [f.upstream] }))!; const id = randomUUID(), generation = 1, sealed = await sealRestorePoint(captured.payload, { companyId: f.companyId, workspaceId: f.workspaceId, restorePointId: id, generation }); @@ -696,6 +698,86 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { const history = await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId)); return { ...f, run, lease, hold, task, owner, history, cleanup, before: await workspace(f.workspaceId) }; } + async function archivedRepositoryFixture(layout: "absent" | "partial" | "symlink" = "absent", state: "pending" | "needs_repair" = "needs_repair") { + const relativePath = layout === "partial" ? "pending-repository" : layout === "symlink" ? "pending-parent/missing" : ".paperclip-repositories/pending"; + const f = await archivedPreparedFixture("recipe", async f => { + if (layout === "partial") { await fs.mkdir(path.join(f.cwd, relativePath)); await fs.writeFile(path.join(f.cwd, relativePath, "partial.txt"), "retained partial files\n"); } + if (layout === "symlink") { await fs.mkdir(path.join(f.cwd, "saved-target")); await fs.writeFile(path.join(f.cwd, "saved-target", "keep.txt"), "saved target\n"); + await fs.symlink("saved-target", path.join(f.cwd, "pending-parent")); } + }); + const pending = await pendingRepository(f); + const [repository] = await db.update(executionWorkspaceRepositories).set({ state, relativePath, + failureCode: state === "needs_repair" ? "repository_preparation_failed" : null }).where(eq(executionWorkspaceRepositories.id, pending.id)).returning(); + const [task] = await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)).returning(); + return { ...f, repository, task }; + } + it.each(["pending", "needs_repair"] as const)("preserves an archived repository intent after authenticated absence proof (%s)", async state => { + const f = await archivedRepositoryFixture("absent", state); + expect(await migrateRemoteWorkspace({ db, storage, ...f, apply: true })).toMatchObject({ applied: true }); + const row = await workspace(f.workspaceId), authority = readRemoteWorkspaceAuthority(row.metadata)!; + expect(row.status).toBe("archived"); expect(row.closedAt).toEqual(f.before.closedAt); + expect(row.metadata?.gitCacheRestoration).toEqual(f.before.metadata?.gitCacheRestoration); + expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, f.repository.id)))[0]).toEqual(f.repository); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); + expect(await db.select().from(activityLog).where(eq(activityLog.entityId, f.issueId))).toEqual(f.history); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0]).toEqual(f.lease); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0]).toEqual(f.owner); + await expect(readMigratedLegacyWorkspaceSource(db, { companyId: f.companyId, runId: f.runId, reference: f.reference })).resolves.toMatchObject({ reference: f.reference }); + const ref = (await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }))!, archivePath = path.join(root, "intent.pcws"), targetDir = path.join(root, "restored-intent"); + await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "untracked.txt"), "utf8")).toBe("archived private file\n"); + await expect(fs.lstat(path.join(targetDir, f.repository.relativePath))).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.access(f.cwd)).rejects.toMatchObject({ code: "ENOENT" }); + }); + it.each(["partial", "symlink"] as const)("refuses an archived repository intent with saved %s state before upload", async layout => { + const f = await archivedRepositoryFixture(layout), put = storage.putFile.bind(storage); let uploads = 0; + storage.putFile = async input => { uploads++; return put(input); }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); + expect(uploads).toBe(0); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + expect((await workspace(f.workspaceId)).metadata?.gitCacheRestoration).toEqual(f.before.metadata?.gitCacheRestoration); + expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, f.repository.id)))[0]).toEqual(f.repository); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); + }); + it("refuses an archived repository intent changed to ready during upload", async () => { + const f = await archivedRepositoryFixture(), put = storage.putFile.bind(storage); let uploaded = false; + storage.putFile = async input => { const saved = await put(input); uploaded = true; + await db.update(executionWorkspaceRepositories).set({ state: "ready", pinnedCommit: "b".repeat(40) }).where(eq(executionWorkspaceRepositories.id, f.repository.id)); return saved; }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed"); + expect(uploaded).toBe(true); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); + }); + it("rechecks the current archived repository intent path at publication", async () => { + const f = await archivedRepositoryFixture(), put = storage.putFile.bind(storage); let uploaded = false; + storage.putFile = async input => { const saved = await put(input); uploaded = true; + await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" }).where(eq(executionWorkspaceRepositories.id, f.repository.id)); return saved; }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); + expect(uploaded).toBe(true); expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task); + expect((await workspace(f.workspaceId)).metadata?.gitCacheRestoration).toEqual(f.before.metadata?.gitCacheRestoration); + }); + it("rechecks an archived repository intent against a recovered uploaded archive", async () => { + const f = await archivedPreparedFixture(), transaction = db.transaction.bind(db); let interrupted = false; + const fault = vi.spyOn(db, "transaction").mockImplementation(async (action, config) => { + const result = await transaction(action, config); + if (!interrupted && result && typeof result === "object" && "status" in result && result.status === "uploaded") { + interrupted = true; throw new Error("uploaded checkpoint reply lost"); + } + return result; + }); + try { await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("uploaded checkpoint reply lost"); } + finally { fault.mockRestore(); } + expect(interrupted).toBe(true); + const [snapshot] = await db.select().from(executionWorkspaceSnapshots); expect(snapshot.status).toBe("uploaded"); + const pending = await pendingRepository(f); + const [repository] = await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" }).where(eq(executionWorkspaceRepositories.id, pending.id)).returning(); + await fs.rm(f.base, { recursive: true }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("repository_preparation_unresolved"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + expect((await db.select().from(executionWorkspaceSnapshots))[0]).toMatchObject({ id: snapshot.id, status: "uploaded" }); + expect((await db.select().from(executionWorkspaceRepositories).where(eq(executionWorkspaceRepositories.id, repository.id)))[0]).toEqual(repository); + }); it.each(["cleanup", "branch"] as const)("preserves a no-recipe archived %s source with prepared recovery", async kind => { const f = await archivedPreparedFixture(kind); expect(await migrateRemoteWorkspace({ db, storage, ...f, apply: true })).toMatchObject({ applied: true }); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 5b9965abd6e..b2b8c4df0cf 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -59,14 +59,21 @@ const gitSourceProof = (source: ArchivedGitMigrationSource | null) => source ? { ...(source.retainedBranch ? { retainedBranch: source.retainedBranch } : {}), } : null; -async function isUnmaterializedRepositoryRequest(workspace: Workspace, - repository: typeof executionWorkspaceRepositories.$inferSelect) { +type RepositoryRequest = typeof executionWorkspaceRepositories.$inferSelect; + +function isUnmaterializedRepositoryCandidate(repository: RepositoryRequest) { const unprepared = repository.state === "pending" && !repository.failureCode || repository.state === "needs_repair" && repository.failureCode === "repository_preparation_failed"; - if (!workspace.cwd || !unprepared || repository.pinnedCommit || repository.branchName) return false; + if (!unprepared || repository.pinnedCommit || repository.branchName) return false; const relative = repository.relativePath; if (!relative || relative.includes("\\") || path.posix.isAbsolute(relative) || path.posix.normalize(relative) !== relative || relative.split("/").some(part => !part || part === "." || part === "..")) return false; + return true; +} + +async function isUnmaterializedRepositoryRequest(workspace: Pick, repository: RepositoryRequest) { + if (!workspace.cwd || !isUnmaterializedRepositoryCandidate(repository)) return false; + const relative = repository.relativePath; const root = await fs.lstat(workspace.cwd).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return null; throw error; @@ -88,6 +95,13 @@ async function isUnmaterializedRepositoryRequest(workspace: Workspace, }); } +async function assertArchivedRepositoryIntents(root: string, repositories: readonly RepositoryRequest[]) { + for (const repository of repositories) { + if (!await isUnmaterializedRepositoryRequest({ cwd: root }, repository)) + throw new Error("remote_workspace_migration_blocked:repository_preparation_unresolved"); + } +} + async function inspect(db: ReadDb, input: { companyId: string; workspaceId: string; operationId?: string; writersStopped?: boolean; onlineSource?: OnlineWorkspaceSource; onlineSourceUnproven?: boolean }, cohort?: WorkspaceMigrationCohort) { const [workspace] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))); if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); @@ -95,7 +109,7 @@ 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, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[] }; + if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[], archivedRepositoryIntents: [] as RepositoryRequest[] }; if (input.writersStopped) reasons.push(...await offlineWorkspaceMigrationBlockers(db)); if (input.onlineSourceUnproven) reasons.push("online_workspace_source_scope_unproven"); if (input.onlineSource) { @@ -301,7 +315,9 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri legacyRecoverySources.sort((a, b) => a.runId.localeCompare(b.runId)); const repositories = await db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, workspace.companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id))); + const archivedRepositoryIntents: RepositoryRequest[] = []; for (const repository of repositories.filter(repository => repository.state !== "ready")) { + if (archivedGitSource?.recipe && isUnmaterializedRepositoryCandidate(repository)) { archivedRepositoryIntents.push(repository); continue; } if ((!input.writersStopped && !input.onlineSource) || !await isUnmaterializedRepositoryRequest(workspace, repository)) reasons.push("repository_preparation_unresolved"); } const source: RemoteWorkspaceSource = workspace.repoUrl ? { kind: "git", repoUrl: workspace.repoUrl, @@ -309,7 +325,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri const materialized = repositories.filter(repository => repository.state === "ready"); if (materialized.length) source.repositories = materialized.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, legacyRecoverySources, finalizedRecoveryHolds, onlineRecoveryHolds }; + return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources, finalizedRecoveryHolds, onlineRecoveryHolds, archivedRepositoryIntents }; } export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) { @@ -365,7 +381,7 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); if (workspace.sourceIssueId !== selected.sourceIssueId) throw new Error("remote_workspace_migration_source_changed"); const current = await inspect(tx as unknown as Db, { ...input, onlineSource }, cohort); - if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[] }; + if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[], archivedRepositoryIntents: [] as RepositoryRequest[] }; if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); const nativeSessionContinuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace); const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source); @@ -390,7 +406,7 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork if (canonical(recoverySourcesResumeIdentity(operation.metadata?.legacyRecoverySources ?? [])) !== canonical(recoverySourcesResumeIdentity(current.legacyRecoverySources))) throw new Error("remote_workspace_migration_source_changed"); if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) 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, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; + return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds, archivedRepositoryIntents: current.archivedRepositoryIntents }; } const [operation] = await tx.insert(workspaceOperations).values({ ...(cohort ? { id: cohort.members.find(member => member.workspaceId === input.workspaceId)!.operationId } : {}), companyId: workspace.companyId, executionWorkspaceId: workspace.id, issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", @@ -408,7 +424,7 @@ async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWork 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, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; + return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds, archivedRepositoryIntents: current.archivedRepositoryIntents }; }); } type ReservedMigration = Awaited>; @@ -505,6 +521,7 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati if (!saved) throw new Error("remote_workspace_migration_operation_changed"); operation = saved; } + if (sourceRoot) await assertArchivedRepositoryIntents(sourceRoot, reserved.archivedRepositoryIntents); const exclude = nested.map(child => child.relativePath); const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); @@ -520,6 +537,7 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati if (!verifiedArchivePath) await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive, signal: input.signal }); await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); + await assertArchivedRepositoryIntents(restored, reserved.archivedRepositoryIntents); if (!recovered) { const currentArchive = path.join(scratch, "current.pcws"); const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath: currentArchive, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); @@ -532,7 +550,7 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati if (canonical(verified) !== canonical(nested)) throw new Error("nested_workspace_source_changed"); } input.signal?.throwIfAborted(); - return { operation, ref, nested, scratch }; + return { operation, ref, nested, scratch, restored }; } catch (error) { await removeTaskDirectoryTree(scratch); throw error; } } type CapturedMigration = Awaited>; @@ -574,12 +592,16 @@ async function validateMigrationPublication(tx: MigrationTransaction, input: Mig .from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), eq(environmentLeases.id, allocation.leaseId))).for("update", { noWait: true }); } + if (reserved.archivedGitSource?.recipe) await tx.select({ id: executionWorkspaceRepositories.id }).from(executionWorkspaceRepositories) + .where(and(eq(executionWorkspaceRepositories.companyId, input.companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, input.workspaceId))) + .orderBy(executionWorkspaceRepositories.id).for("update", { noWait: true }); const validation = await inspect(tx as unknown as Db, { ...input, operationId: operation.id, onlineSource }, cohort); if (!input.source && canonical(validation.source) !== canonical(reserved.source)) throw new Error("remote_workspace_migration_source_changed"); if (validation.issueBinding !== reserved.issueBinding) throw new Error("remote_workspace_migration_source_changed"); if (Boolean(validation.retainedDirectory) !== Boolean(reserved.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); if (canonical(gitSourceProof(validation.archivedGitSource)) !== canonical(gitSourceProof(reserved.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); if (validation.reasons.length) throw new Error(`remote_workspace_migration_blocked:${validation.reasons.join(",")}`); + await assertArchivedRepositoryIntents(captured.restored, validation.archivedRepositoryIntents); if (reserved.archivedGitSource) await lockArchivedGitMigrationSources(tx as unknown as Db, workspace, reserved.archivedGitSource); const [uploaded] = await tx.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.id, ref.id))).for("update");