fix(workspaces): retain migration scope and restart bindings

Co-Authored-By: Paperclip <noreply@paperclip.ing>

server/src/services/remote-workspace-migration.test.ts

@@ -847,6 +847,60 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
     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("resumes a later archived upload after an earlier missing attempt with a different manifest", async () => {
+    const f = await archivedPreparedFixture(), upload = storage.putFile.bind(storage);
+    const interrupted = vi.spyOn(storage, "putFile").mockImplementationOnce(async input => {
+      await upload(input); throw new Error("later archived upload reply lost");
+    });
+    await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("later archived upload reply lost");
+    interrupted.mockRestore();
+    const [pending] = await db.select().from(executionWorkspaceSnapshots);
+    expect(pending.status).toBe("pending");
+    const [existing] = pending.metadata!.uploads as Array<{ objectKey: string; manifestDigest: string; [key: string]: unknown }>;
+    const missing = { ...existing, objectKey: `${existing.objectKey}.missing`, manifestDigest: "00".repeat(32) };
+    expect(missing.manifestDigest).not.toBe(existing.manifestDigest);
+    expect(await storage.headObject(f.companyId, missing.objectKey)).toMatchObject({ exists: false });
+    expect(await storage.headObject(f.companyId, existing.objectKey)).toMatchObject({ exists: true });
+    const attempts = [missing, existing];
+    await db.update(executionWorkspaceSnapshots).set({ metadata: { ...pending.metadata, uploads: attempts } })
+      .where(eq(executionWorkspaceSnapshots.id, pending.id));
+    const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory");
+    const get = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("resume must not download"));
+    const put = vi.spyOn(storage, "putFile");
+    await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true });
+    expect(capture).not.toHaveBeenCalled(); expect(get).not.toHaveBeenCalled(); expect(put).not.toHaveBeenCalled();
+    const [saved] = await db.select().from(executionWorkspaceSnapshots);
+    expect(saved).toMatchObject({ id: pending.id, sourceOperationId: pending.sourceOperationId, status: "committed",
+      objectKey: existing.objectKey, manifestDigest: existing.manifestDigest, metadata: { uploads: attempts } });
+    expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBe(pending.id);
+  });
+  it("refuses an old archived journal with unknown nonempty repository intent bindings before resume work", async () => {
+    const f = await archivedRepositoryFixture(), upload = storage.putFile.bind(storage);
+    const interrupted = vi.spyOn(storage, "putFile").mockImplementationOnce(async input => {
+      await upload(input); throw new Error("old archived upload reply lost");
+    });
+    await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("old archived upload reply lost");
+    interrupted.mockRestore();
+    const [operation] = await db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_migration"));
+    const metadata = { ...operation.metadata }; delete metadata.archivedRepositoryIntents;
+    await db.update(workspaceOperations).set({ metadata }).where(eq(workspaceOperations.id, operation.id));
+    await db.update(executionWorkspaceRepositories).set({ relativePath: "untracked.txt" })
+      .where(eq(executionWorkspaceRepositories.id, f.repository.id));
+    const journals = await db.select().from(workspaceOperations), snapshots = await db.select().from(executionWorkspaceSnapshots);
+    const before = await workspace(f.workspaceId), repositories = await db.select().from(executionWorkspaceRepositories);
+    const baseBytes = await fs.readFile(path.join(f.base, "base.txt"));
+    const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory");
+    const get = vi.spyOn(storage, "getObject"), put = vi.spyOn(storage, "putFile"), head = vi.spyOn(storage, "headObject");
+    await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_workspace_migration_source_changed");
+    expect(capture).not.toHaveBeenCalled(); expect(get).not.toHaveBeenCalled(); expect(put).not.toHaveBeenCalled(); expect(head).not.toHaveBeenCalled();
+    expect(await db.select().from(workspaceOperations)).toEqual(journals);
+    expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(snapshots);
+    expect(await workspace(f.workspaceId)).toEqual(before);
+    expect(await db.select().from(executionWorkspaceRepositories)).toEqual(repositories);
+    expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]).toEqual(f.task);
+    expect(await fs.readFile(path.join(f.base, "base.txt"))).toEqual(baseBytes);
+    await expect(fs.access(f.cwd)).rejects.toMatchObject({ code: "ENOENT" });
+  });
   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 });

server/src/services/remote-workspace-migration.ts

@@ -413,8 +413,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");
-      if (Object.hasOwn(operation.metadata ?? {}, "archivedRepositoryIntents")
+      if (canonical(operation.metadata?.archivedRepositoryIntents ?? []) !== canonical(repositoryIntentBindings(current.archivedRepositoryIntents)))
-        && canonical(operation.metadata!.archivedRepositoryIntents) !== canonical(repositoryIntentBindings(current.archivedRepositoryIntents)))
         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, archivedRepositoryIntents: current.archivedRepositoryIntents };
     }
@@ -528,7 +527,8 @@ async function captureMigration(input: MigrationInput, reserved: ReservedMigrati
     if (recovered) {
       if (scope.state === "pending") throw new Error("nested_workspace_scope_invalid");
       const ref = await uploadWorkspaceSnapshot({ ...input, snapshotId: snapshot.id, archivePath,
-        manifestDigest: recovered.manifestDigest, verification: "upload" });
+        manifestDigest: recovered.manifestDigest, verification: "upload",
+        ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}) });
       return { operation, ref, nested: scope.state === "fixed" ? validateImportedWorkspaceChildren(scope.children) : [], scratch };
     }
     const sourceRoot = retained ? await materializeTaskDirectorySnapshot({

server/src/services/remote-workspace-snapshot-archive.test.ts

@@ -131,8 +131,8 @@ describe("portable remote workspace archives", () => {
     expect(await sourceBytes(cwd)).toBe(before);
     await expect(capture(cwd)).resolves.toMatchObject({ manifestDigest: expect.any(String) });
   });
-  it.each(["linked", "borrowed"].flatMap(kind => ["local", "serialized"].map(mode => ({ kind, mode }))))(
+  it.each(["linked", "borrowed"].flatMap(kind => ["local", "serialized"].flatMap(mode => [false, true].map(singlePass => ({ kind, mode, singlePass })))))(
-    "enforces owned Git source roots for $kind $mode capture without changing archive bytes", async ({ kind, mode }) => {
+    "enforces owned Git source roots for $kind $mode capture (single pass: $singlePass)", async ({ kind, mode, singlePass }) => {
       const base = await source(); await git(base, "init", "-q", "-b", "main");
       await fs.writeFile(path.join(base, "work.txt"), "base work\n"); await git(base, "add", "."); await git(base, "commit", "-qm", "base");
       const cwd = path.join(root, "selected");
@@ -143,7 +143,7 @@ describe("portable remote workspace archives", () => {
       const baseline = await capture(cwd), archivePath = path.join(root, "owned.pcws");
       const prefix = remoteWorkspaceArchiveScript().split("const [kind, first, second, digest]")[0];
       const fenced = async (roots: string[]) => {
-        const input = { kind: "capture", cwd, archivePath, gitSourceRoots: roots };
+        const input = { kind: "capture", cwd, archivePath, gitSourceRoots: roots, singlePass };
         return mode === "local" ? captureRemoteWorkspaceDirectory(input)
           : JSON.parse((await execute(process.execPath, ["--eval", `${prefix}archive(${JSON.stringify(input)}).then(value=>process.stdout.write(JSON.stringify(value))).catch(error=>{process.stderr.write(error.message);process.exitCode=1;});`], { env })).stdout);
       };

server/src/services/remote-workspace-snapshot-archive.ts

@@ -516,6 +516,7 @@ export async function workspaceArchive(input: {
       };
       for (let index = 0; index < repositories.length; index++) {
         const relative = repositories[index]!, repo = path.join(root, relative), normalized = path.join(scratch, `repo-${index}.git`);
+        if (singlePass) await assertGitSourceFence(repo);
         const settings = await portableGitSettings(repo);
         if (!singlePass) gitSignatures.set(relative, await signature(repo, settings));
         const offline = await gitStateHash(repo, "shallow") !== null

server/src/services/remote-workspace-snapshot-storage.test.ts

@@ -77,6 +77,23 @@ suite("remote workspace snapshot storage", () => {
     const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref.id));
     expect(row?.status).toBe("uploaded"); // The owning service still controls atomic publication.
   });
+  it.each(["uploaded", "lost-reply"])("refuses an unreadable %s snapshot key without downloading", async state => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    if (state === "lost-reply") {
+      storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); };
+      await expect(uploadWorkspaceSnapshot({ ...f.upload, verification: "upload" })).rejects.toThrow("lost upload reply");
+      storage.putFile = put;
+    } else await uploadWorkspaceSnapshot({ ...f.upload, verification: "upload" });
+    await fs.rm(f.archivePath);
+    const get = vi.spyOn(storage, "getObject").mockRejectedValue(new Error("must not download"));
+    const upload = vi.spyOn(storage, "putFile");
+    const [before] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id));
+    vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "42".repeat(32));
+    await expect(uploadWorkspaceSnapshot({ ...f.upload, verification: "upload" })).rejects.toThrow();
+    const [after] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id));
+    expect(after).toEqual(before);
+    expect(get).not.toHaveBeenCalled(); expect(upload).not.toHaveBeenCalled();
+  });
   it("records where a save and a restore spend their time", async () => {
     const f = await fixture();
     const saveTiming = newWorkspaceSnapshotStorageTiming();

server/src/services/remote-workspace-snapshot-storage.ts

@@ -212,7 +212,7 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P
 
 async function uploadWorkspaceSnapshotTimed(input: WorkspaceSnapshotUpload): Promise<WorkspaceSnapshotRef> {
   if (input.verification === "upload")
-    return uploadVerifiedWorkspaceSnapshot(input, async () => {}, { acceptServerVerifiedUpload: false });
+    return uploadVerifiedWorkspaceSnapshot(input, assertWorkspaceSnapshotKeyReadable, { acceptServerVerifiedUpload: false });
   if (input.verifiedArchivePath === undefined)
     return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal), { acceptServerVerifiedUpload: true });
   input.signal?.throwIfAborted();