perf: reuse authenticated workspace upload readback (#19811)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Workspace import saves local work to an encrypted immutable snapshot
before it changes storage authority.
> - The snapshot uploader reads the whole object back to verify its
ciphertext.
> - The importer then downloads that same object again for authenticated
decryption and restore.
> - Both reads are required by the current call sequence, but they need
not transfer the object twice.
> - This PR lets that caller retain the authenticated upload readback in
private scratch.
> - The importer still restores the full archive, recaptures the source,
checks ownership, and publishes through the existing CAS.

## Linked Issues or Issue Description

Refs #19642, #19805, and #19810 for the existing workspace import and
preservation owners.

**What happened?**

A fresh workspace import performs a PUT and two full GETs of the same
encrypted snapshot. The first GET verifies and discards the object. The
next GET decrypts it for restore. Already-uploaded and lost-reply
recovery paths also repeat the GET.

**Expected behavior**

Use one authenticated readback for the exact final winning reference.
Preserve the default upload behavior and returned reference. Keep every
restore, source, recovery, ownership, and publication check.

**Steps to reproduce**

1. Capture a local source through the real workspace archive codec.
2. Upload it through the real snapshot ledger and filesystem object
backend.
3. Load and restore its archive. Count the object GETs.
4. The three baseline cases each make two GETs. The opt-in path now
makes one and restores the same bytes after the original source is
removed.

**Paperclip version or commit**

Baseline `585c885168`. The first PR head was `5b18121609`; the follow-up
repairs the review findings without changing the storage optimization.

## What Changed

- Add optional `verifiedArchivePath` to the existing upload owner. The
default path and `WorkspaceSnapshotRef` return contract stay unchanged.
- Authenticate each readback with the existing ciphertext, GCM,
key-binding, and plaintext checks.
- Bind retained files to the full winning reference. Recheck private
file ownership and plaintext SHA/size before atomic publication to a new
caller-owned path.
- Clean all invocation scratch on success or failure. Do not overwrite
caller files or reuse state across calls.
- Use this path in POSIX workspace import. Keep full extraction, the
second source capture, closed alias/source/F/writer/task checks, and CAS
publication. Windows keeps the existing download path and does not
bypass private-file checks.
- Cover real cold restore, fresh/replay/lost-reply GET counts, late
winners, private-path rejection, local tampering, and GCM failure.

## Verification

- Genuine baseline RED: three real PostgreSQL/crypto cases each report
two GETs where the new contract requires one.
- GREEN: all 30 snapshot storage cases, one actual online migration
case, and seven focused migration controls pass; 38 total. The other 232
cases in those files were unselected locally.
- Review RED: four obsolete second-GET fault triggers and one
uploaded-journal retry case failed before repair. The triggers now fire
on the retained readback and assert that they fired. The
uploaded-journal case interrupts after its real ledger commit and proves
recovery and cold Git restore through the same journal.
- A focused platform-selection control proves Windows retains the
original two-GET path. This is not native Windows acceptance.
- Strict server TypeScript, module boundaries, and `git diff --check`
pass.
- Local command from `server/`: `node
../cli/node_modules/tsx/dist/cli.mjs
../packages/db/src/test-postgres-command.ts -- node
../node_modules/vitest/vitest.mjs run
src/services/remote-workspace-snapshot-storage.test.ts
src/services/remote-workspace-online-source.test.ts
src/services/remote-workspace-migration.test.ts --testNamePattern
'remote workspace snapshot storage|imports an alias with
unrelated|removes plaintext verification files|rechecks a service that
becomes|rechecks a writer that appears|preserves a local edit|keeps the
existing download path|missing-base-uploaded-retry' --hookTimeout 90000
--maxWorkers 1 --no-file-parallelism --fsModuleCache`.
- The codec timing control used an exact earlier owner and a synthetic
72.9 MB archive. Discarding final-capture output saved only 3%; this PR
keeps that capture unchanged.
- This removes one full object GET. It does not claim a measured hosted
wall-clock improvement.
- Fresh Greptile review is 5/5 on `0933a0a29a`, with no open findings.
Exact-head CI is complete and green: 56 successful checks, four intended
skips, and one neutral result. The second review returned Ship (4/5);
its optional test-maintainability suggestion is deferred. No source
change or CI restart was made for that suggestion.

## Risks

- A late upload can win with a different reference. The retained file
must match all fields of that final reference; otherwise the owner
authenticates the winner.
- A private file can change before relay. Ownership, no-follow reads,
stable file identity, plaintext SHA/size, and no-overwrite publication
protect that boundary.
- Authentication may overlap the encrypted upload scratch. Peak local
disk use can grow by up to one ciphertext file. Data remains streaming;
no whole-object memory buffer is added.
- Opt-in callers must supply a canonical private directory owned by the
process. Existing callers that omit the option retain their behavior.
- No archive format, database schema, provider lifecycle, writer
protocol, admission, or background polling changes.

## Model Used

OpenAI `gpt-6.1-sol`, with reasoning, tool use, and local code
execution. The context window size is not exposed.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>

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

@@ -857,12 +857,18 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
       expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
     }
     if (state === "missing-base-uploaded-retry") {
-      const get = storage.getObject.bind(storage); let reads = 0;
-      storage.getObject = async (...args) => {
-        if (args[1].includes("/workspace-snapshots/") && ++reads === 2) throw new Error("restore reply lost");
-        return get(...args);
-      };
-      await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("restore reply lost");
+      const 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");
+        expect(interrupted).toBe(true);
+      } finally { fault.mockRestore(); }
       expect((await db.select().from(executionWorkspaceSnapshots))[0]?.status).toBe("uploaded");
     }
     const interrupted = state.startsWith("missing-base-") || state.startsWith("nested-")
@@ -1099,6 +1105,15 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
     expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(1);
     expect(await db.select().from(activityLog)).toHaveLength(1);
   });
+  it("keeps the existing download path for Windows imports", async () => {
+    const f = await seed(), get = vi.spyOn(storage, "getObject"), platform = vi.spyOn(process, "platform", "get").mockReturnValue("win32");
+    try {
+      await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true });
+      expect(get).toHaveBeenCalledTimes(2);
+      expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)?.snapshotId).toBeTruthy();
+      expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n");
+    } finally { platform.mockRestore(); }
+  });
   it.each([false, true])("removes plaintext verification files with read-only folders after import (source changed: %s)", async sourceChanged => {
     const f = await seed(), protectedDir = path.join(f.cwd, "protected"), scratchRoot = path.join(root, "scratch");
     await fs.mkdir(protectedDir); await fs.mkdir(scratchRoot);
@@ -1107,13 +1122,14 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
     vi.stubEnv("TMPDIR", scratchRoot);
     const get = storage.getObject.bind(storage); let reads = 0;
     if (sourceChanged) storage.getObject = async (...args) => {
-      if (++reads === 2) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n");
+      if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n");
       return get(...args);
     };
     try {
       const migration = migrateRemoteWorkspace({ db, storage, ...f, apply: true });
       if (sourceChanged) {
         await expect(migration).rejects.toThrow("remote_workspace_migration_source_changed");
+        expect(reads).toBe(1);
         expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
       } else {
         await expect(migration).resolves.toMatchObject({ applied: true, alreadyMigrated: true });
@@ -1452,29 +1468,32 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
   it("rechecks a service that becomes unresolved during snapshot verification", async () => {
     const f = await seed(), get = storage.getObject.bind(storage); let reads = 0;
     storage.getObject = async (...args) => {
-      if (++reads === 2) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId,
+      if (++reads === 1) await db.insert(workspaceRuntimeServices).values({ id: randomUUID(), companyId: f.companyId,
         executionWorkspaceId: f.workspaceId, scopeType: "execution_workspace", scopeId: f.workspaceId,
         serviceName: "Late synthetic preview", status: "running", lifecycle: "shared", provider: "local_process", cwd: f.cwd });
       return get(...args);
     };
     await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_runtime_service_unsettled");
+    expect(reads).toBe(1);
     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("rechecks a writer that appears while the restored copy is being verified", async () => {
     const f = await seed(), get = storage.getObject.bind(storage); let reads = 0;
     storage.getObject = async (...args) => {
-      if (++reads === 2) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId,
+      if (++reads === 1) await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.workspaceId,
         phase: "workspace_command", status: "running" });
       return get(...args);
     };
     await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("workspace_operation_unresolved");
+    expect(reads).toBe(1);
     expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
   });
   it("preserves a local edit made during verification and refuses activation", async () => {
     const f = await seed(), get = storage.getObject.bind(storage); let reads = 0;
-    storage.getObject = async (...args) => { if (++reads === 2) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); };
+    storage.getObject = async (...args) => { if (++reads === 1) await fs.writeFile(path.join(f.cwd, "saved.txt"), "new local work\n"); return get(...args); };
     await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("source_changed");
+    expect(reads).toBe(1);
     expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull();
     expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("new local work\n");
   });

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

@@ -402,7 +402,7 @@ export async function migrateRemoteWorkspace(input: {
     }
     const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId,
       operationId: operation.id, generation: 1, epoch: 1 });
-    const scratch = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"));
+    const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-")));
     try {
       input.signal?.throwIfAborted();
       const retained = reserved.retainedDirectory?.restorePoint;
@@ -435,15 +435,18 @@ export async function migrateRemoteWorkspace(input: {
       }
       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");
+      const verifiedArchivePath = process.platform === "win32" ? undefined : restoredArchive;
       const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id,
         companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal,
+        verifiedArchivePath,
         // Private Git reconstruction can change administrative bytes between
         // attempts. Its journaled logical recipe is checked at reservation and
         // again under source locks before authority publication below.
         ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}),
       });
-      const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored");
-      await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive, signal: input.signal });
+      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 });
       if (!recovered) {
         const currentArchive = path.join(scratch, "current.pcws");

server/src/services/remote-workspace-online-source.test.ts

@@ -267,7 +267,9 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", ()
     const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Unrelated active agent" }).returning();
     const run = await remoteWriter(f.companyId, agent.id);
     expect(await inventoryRemoteWorkspaceMigration(db, f)).toMatchObject({ eligible: true });
+    const get = vi.spyOn(storage, "getObject");
     await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).resolves.toMatchObject({ applied: true });
+    expect(get).toHaveBeenCalledTimes(1);
     expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)))[0]).toEqual(run);
     expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("saved legacy edits\n");
     expect(await stored(f.aliases[1].workspace.id)).toEqual(f.aliases[1].workspace);

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

@@ -3,6 +3,7 @@ import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { Readable } from "node:stream";
+import { appendFileSync, readdirSync } from "node:fs";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
 import { eq } from "drizzle-orm";
 import { applyPendingMigrations, closeRegisteredClients, createDb, executionWorkspaceSnapshots } from "@paperclipai/db";
@@ -10,6 +11,7 @@ import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } fro
 import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
 import { createStorageService } from "../storage/service.js";
 import type { StorageService } from "../storage/types.js";
+import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
 import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js";
 import { beginWorkspaceSnapshot, getWorkspaceSnapshot, loadWorkspaceSnapshotArchive, recoverWorkspaceSnapshotArchive, uploadWorkspaceSnapshot } from "./remote-workspace-snapshot-storage.js";
 
@@ -30,7 +32,7 @@ suite("remote workspace snapshot storage", () => {
   afterAll(async () => { if (externalDatabaseUrl) await closeRegisteredClients(externalDatabaseUrl); else await database?.cleanup(); });
   beforeEach(async () => {
     vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "41".repeat(32));
-    root = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-store-test-"));
+    root = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-store-test-")));
     storage = createStorageService(createLocalDiskStorageProvider(path.join(root, "objects")));
   });
   afterEach(async () => { vi.unstubAllEnvs(); await db.delete(executionWorkspaceSnapshots); await fs.rm(root, { recursive: true, force: true }); });
@@ -57,6 +59,67 @@ 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(["fresh", "uploaded", "lost-reply"])("reuses one authenticated %s readback for a complete cold restore", async state => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    if (state === "uploaded") await uploadWorkspaceSnapshot(f.upload);
+    if (state === "lost-reply") {
+      storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); };
+      await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply");
+      storage.putFile = put;
+    }
+    const get = vi.spyOn(storage, "getObject"), verifiedArchivePath = path.join(root, "verified.pcws");
+    const request = { ...f.upload, verifiedArchivePath }, ref = await uploadWorkspaceSnapshot(request);
+    if (!await fs.lstat(verifiedArchivePath).catch(() => null))
+      await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: verifiedArchivePath });
+    expect(get).toHaveBeenCalledTimes(1);
+    expect(await fs.readFile(verifiedArchivePath)).toEqual(await fs.readFile(f.archivePath));
+    expect((await fs.stat(verifiedArchivePath)).mode & 0o777).toBe(0o600);
+    expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: ref.id })).toEqual(ref);
+    await fs.rm(f.cwd, { recursive: true }); await fs.rm(f.archivePath);
+    await restoreRemoteWorkspaceArchive({ archivePath: verifiedArchivePath, targetDir: path.join(root, "cold"), expectedManifestDigest: ref.manifestDigest });
+    expect(await fs.readFile(path.join(root, "cold", "saved.txt"), "utf8")).toBe("acknowledged workspace content\n");
+    expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false);
+  });
+  it.each(["existing", "public-parent", "linked-parent"])("refuses a caller-owned readback path with %s ownership", async mode => {
+    const f = await fixture(), parent = path.join(root, "destination"); await fs.mkdir(parent, { mode: 0o700 });
+    let verifiedArchivePath = path.join(parent, "archive.pcws");
+    if (mode === "existing") await fs.writeFile(verifiedArchivePath, "keep caller bytes");
+    if (mode === "public-parent") await fs.chmod(parent, 0o755);
+    if (mode === "linked-parent") {
+      const alias = path.join(root, "alias"); await fs.symlink(parent, alias); verifiedArchivePath = path.join(alias, "archive.pcws");
+    }
+    const get = vi.spyOn(storage, "getObject"), put = vi.spyOn(storage, "putFile");
+    await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow("readback_");
+    expect(get).not.toHaveBeenCalled(); expect(put).not.toHaveBeenCalled();
+    if (mode === "existing") expect(await fs.readFile(verifiedArchivePath, "utf8")).toBe("keep caller bytes");
+    expect((await db.select().from(executionWorkspaceSnapshots))[0].status).toBe("pending");
+  });
+  it("rejects a tampered verified plaintext before publishing the caller path", async () => {
+    const f = await fixture(), verifiedArchivePath = path.join(root, "tampered.pcws"), debug = db.$client.options.debug;
+    let changed = false;
+    db.$client.options.debug = (_connection, query) => {
+      if (changed || !query.startsWith('update "execution_workspace_snapshots"') || !query.includes('"status" =')) return;
+      const directory = readdirSync(root).find(name => name.startsWith(".workspace-upload-readback-"));
+      if (!directory) return;
+      const file = readdirSync(path.join(root, directory)).find(name => name.endsWith(".pcws"));
+      if (file) { appendFileSync(path.join(root, directory, file), "changed plaintext"); changed = true; }
+    };
+    try {
+      await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow("readback_file_changed");
+      expect(changed).toBe(true);
+      expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull();
+      expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false);
+    } finally { db.$client.options.debug = debug; }
+  });
+  it("keeps GCM authentication mandatory for an uploaded readback and removes all failed plaintext", async () => {
+    const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload), verifiedArchivePath = path.join(root, "bad-key.pcws");
+    const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null }));
+    const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ ...wrapped, dataKey: Buffer.alloc(32, 8).toString("base64") }) });
+    await db.update(executionWorkspaceSnapshots).set({ key: material }).where(eq(executionWorkspaceSnapshots.id, ref.id));
+    await expect(uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath })).rejects.toThrow();
+    expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull();
+    expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false);
+  });
   it("reconciles an upload whose reply was lost after object persistence, without the old scratch file", async () => {
     const f = await fixture(); const put = storage.putFile.bind(storage); let loseReply = true;
     storage.putFile = async input => { const result = await put(input); if (loseReply) { loseReply = false; throw new Error("lost upload reply"); } return result; };
@@ -156,7 +219,8 @@ suite("remote workspace snapshot storage", () => {
     expect(saved).toMatchObject({ status: "pending", objectKey: null });
     expect(saved!.metadata!.uploads).toHaveLength(1);
   });
-  it.each(["original", "recapture", "original-with-lost-reply"])("keeps a %s winner when another physical capture finishes late", async winner => {
+  it.each(["original", "recapture", "original-with-lost-reply"].flatMap(winner => [false, true].map(readback => ({ winner, readback }))))(
+    "keeps a $winner winner when another physical capture finishes late (readback: $readback)", async ({ winner, readback }) => {
     const f = await fixture(), put = storage.putFile.bind(storage);
     const deferred = () => {
       let resolve!: () => void;
@@ -174,14 +238,16 @@ suite("remote workspace snapshot storage", () => {
       return saved;
     };
     const unpublishedRecapture = { sourceOperationId: f.operationId };
-    const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture });
+    const originalPath = path.join(root, "original-verified.pcws"), recapturedPath = path.join(root, "recaptured-verified.pcws");
+    const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture, ...(readback ? { verifiedArchivePath: originalPath } : {}) });
     let recapture: ReturnType<typeof uploadWorkspaceSnapshot> | undefined;
     try {
       await originalReady.promise;
       const archivePath = path.join(root, "recaptured.pcws");
       await fs.writeFile(path.join(f.cwd, "saved.txt"), "different physical capture\n");
       const captured = await captureRemoteWorkspaceDirectory({ cwd: f.cwd, archivePath });
-      recapture = uploadWorkspaceSnapshot({ ...f.upload, archivePath, manifestDigest: captured.manifestDigest, unpublishedRecapture });
+      recapture = uploadWorkspaceSnapshot({ ...f.upload, archivePath, manifestDigest: captured.manifestDigest, unpublishedRecapture,
+        ...(readback ? { verifiedArchivePath: recapturedPath } : {}) });
       await recaptureReady.promise;
       (winner === "recapture" ? recaptureRelease : originalRelease).resolve();
       if (winner === "original-with-lost-reply") {
@@ -193,6 +259,11 @@ suite("remote workspace snapshot storage", () => {
       originalRelease.resolve(); recaptureRelease.resolve();
       if (winner !== "original-with-lost-reply") expect(await original).toEqual(saved);
       expect(await recapture).toEqual(saved);
+      if (readback) {
+        const winningBytes = await fs.readFile(winner === "recapture" ? archivePath : f.archivePath);
+        expect(await fs.readFile(recapturedPath)).toEqual(winningBytes);
+        if (winner !== "original-with-lost-reply") expect(await fs.readFile(originalPath)).toEqual(winningBytes);
+      }
       expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: f.row.id })).toEqual(saved);
       expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(2);
     } finally { originalRelease.resolve(); recaptureRelease.resolve(); await Promise.allSettled([original, recapture]); }

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

@@ -1,6 +1,6 @@
 import { createCipheriv, createDecipheriv, createHash, randomBytes, randomUUID } from "node:crypto";
-import { createReadStream, createWriteStream } from "node:fs";
-import { link, mkdtemp, open, rm } from "node:fs/promises";
+import { constants, createReadStream, createWriteStream } from "node:fs";
+import { link, lstat, mkdtemp, open, realpath, rm } from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
 import { pipeline } from "node:stream/promises";
@@ -121,22 +121,74 @@ export async function recoverWorkspaceSnapshotArchive(input: {
   return null;
 }
 
-/** Upload attempts are journaled before remote I/O. The caller owns publication. */
-export async function uploadWorkspaceSnapshot(input: {
+type WorkspaceSnapshotUpload = {
   db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string;
-  archivePath: string; manifestDigest: string; signal?: AbortSignal;
+  archivePath: string; manifestDigest: string; signal?: AbortSignal; verifiedArchivePath?: string;
   // Only an owner with an immutable logical source may replace an unavailable,
   // unpublished capture. A recoverable or concurrently published attempt wins,
   // and its returned manifest may differ from the new physical capture.
   unpublishedRecapture?: { sourceOperationId: string };
-}): Promise<WorkspaceSnapshotRef> {
+};
+
+/** Upload attempts are journaled before remote I/O. The caller owns publication. */
+export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): Promise<WorkspaceSnapshotRef> {
+  if (input.verifiedArchivePath === undefined)
+    return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal));
+  input.signal?.throwIfAborted();
+  const target = input.verifiedArchivePath, parent = path.dirname(target);
+  if (!path.isAbsolute(target) || path.normalize(target) !== target) failure("readback_parent_invalid");
+  const owner = await lstat(parent);
+  const assertParent = async () => {
+    const current = await lstat(parent);
+    if (!current.isDirectory() || current.isSymbolicLink() || (current.mode & 0o777) !== 0o700
+      || typeof process.getuid === "function" && current.uid !== process.getuid()
+      || current.dev !== owner.dev || current.ino !== owner.ino || await realpath(parent) !== parent) failure("readback_parent_invalid");
+  };
+  await assertParent();
+  await lstat(target).then(() => failure("readback_target_exists"), (error: NodeJS.ErrnoException) => {
+    if (error.code !== "ENOENT") throw error;
+  });
+  const scratch = await mkdtemp(path.join(parent, ".workspace-upload-readback-"));
+  const verified: Array<{ ref: WorkspaceSnapshotRef; file: string }> = [];
+  const verify = async (ref: WorkspaceSnapshotRef) => {
+    if (verified.some(saved => isDeepStrictEqual(saved.ref, ref))) return;
+    const file = path.join(scratch, `${randomUUID()}.pcws`);
+    await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: file,
+      signal: input.signal, temporaryParent: scratch });
+    verified.push({ ref: structuredClone(ref), file });
+  };
+  try {
+    const ref = await uploadVerifiedWorkspaceSnapshot(input, verify);
+    await verify(ref);
+    const file = verified.find(saved => isDeepStrictEqual(saved.ref, ref))!.file;
+    const wrapped = await snapshotKey(ref), handle = await open(file, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK);
+    try {
+      const before = await handle.stat(), digest = createHash("sha256"); let bytes = 0;
+      if (!before.isFile() || (before.mode & 0o777) !== 0o600 || before.nlink !== 1
+        || typeof process.getuid === "function" && before.uid !== process.getuid()) failure("readback_file_changed");
+      for await (const chunk of createReadStream(file, { fd: handle.fd, autoClose: false, signal: input.signal })) {
+        digest.update(chunk); bytes += chunk.length;
+      }
+      const after = await handle.stat(), named = await lstat(file);
+      if (!named.isFile() || named.isSymbolicLink() || named.dev !== before.dev || named.ino !== before.ino
+        || named.nlink !== 1 || named.mode !== before.mode || after.size !== before.size
+        || after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs
+        || bytes !== wrapped.plaintextBytes || digest.digest("hex") !== wrapped.plaintextSha256) failure("readback_file_changed");
+      await assertParent(); input.signal?.throwIfAborted();
+      await link(file, target);
+    } finally { await handle.close(); }
+    return ref;
+  } finally { await rm(scratch, { recursive: true, force: true }); }
+}
+
+async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise<void>): Promise<WorkspaceSnapshotRef> {
   const [row] = await input.db.select().from(executionWorkspaceSnapshots).where(scope(input));
   if (!row) failure("missing");
   if (input.unpublishedRecapture && row!.sourceOperationId !== input.unpublishedRecapture.sourceOperationId) failure("operation_binding_changed");
   if (["uploaded", "committed", "superseded"].includes(row!.status)) {
     const saved = reference(row!);
     if (saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
-    await verifyObject(input.storage, saved, input.signal); return saved;
+    await verify(saved); return saved;
   }
   if (row!.status !== "pending" || !/^[a-f0-9]{64}$/.test(input.manifestDigest)) failure("not_writable");
   const recapture = Boolean(input.unpublishedRecapture);
@@ -157,7 +209,7 @@ export async function uploadWorkspaceSnapshot(input: {
     if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed");
     const saved = reference(winning!);
     if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
-    if (saved.objectKey !== attempt.objectKey) await verifyObject(input.storage, saved, input.signal);
+    if (saved.objectKey !== attempt.objectKey) await verify(saved);
     return saved;
   };
   // Recover a completed upload even when the previous process lost its reply
@@ -169,7 +221,7 @@ export async function uploadWorkspaceSnapshot(input: {
       const ref = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef;
       validateReference(ref);
       if ((await input.storage.headObject(ref.companyId, ref.objectKey, { signal: input.signal })).exists) {
-        await verifyObject(input.storage, ref, input.signal); return attempt;
+        await verify(ref); return attempt;
       }
     }
     return null;
@@ -217,14 +269,14 @@ export async function uploadWorkspaceSnapshot(input: {
     if (!recorded) {
       const saved = await getWorkspaceSnapshot(input.db, input);
       if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) failure("publication_changed");
-      await verifyObject(input.storage, saved, input.signal); return saved;
+      await verify(saved); return saved;
     }
     // The object key is unique per attempt, so a late losing uploader cannot
     // overwrite an object already used by a committed generation.
     await input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey,
       originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted),
       byteSize: attempt.byteSize, sha256: attempt.sha256, signal: input.signal });
-    await verifyObject(input.storage, { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef, input.signal);
+    await verify({ schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef);
     if (recapture) {
       const recovered = await availableAttempt(attempts);
       if (recovered) return publish(recovered);
@@ -233,16 +285,21 @@ export async function uploadWorkspaceSnapshot(input: {
   } finally { await rm(scratch, { recursive: true, force: true }); }
 }
 
+async function snapshotKey(ref: WorkspaceSnapshotRef) {
+  validateReference(ref);
+  const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null }));
+  if (wrapped.schema !== KEY_SCHEMA || wrapped.companyId !== ref.companyId || wrapped.workspaceId !== ref.workspaceId
+    || wrapped.id !== ref.id || wrapped.generation !== ref.generation || !/^[a-f0-9]{64}$/.test(wrapped.plaintextSha256 ?? "")
+    || !Number.isSafeInteger(wrapped.plaintextBytes) || wrapped.plaintextBytes < 0) failure("key_binding_invalid");
+  return wrapped;
+}
+
 /** Authentication finishes in private scratch before any plaintext is published. */
 export async function loadWorkspaceSnapshotArchive(input: {
   storage: StorageService; reference: WorkspaceSnapshotRef; targetPath: string; signal?: AbortSignal; temporaryParent?: string;
 }): Promise<void> {
   input.signal?.throwIfAborted();
-  const ref = input.reference; validateReference(ref);
-  const wrapped = JSON.parse(await localEncryptedProvider.resolveVersion({ material: ref.key, externalRef: null }));
-  if (wrapped.schema !== KEY_SCHEMA || wrapped.companyId !== ref.companyId || wrapped.workspaceId !== ref.workspaceId
-    || wrapped.id !== ref.id || wrapped.generation !== ref.generation || !/^[a-f0-9]{64}$/.test(wrapped.plaintextSha256 ?? "")
-    || !Number.isSafeInteger(wrapped.plaintextBytes) || wrapped.plaintextBytes < 0) failure("key_binding_invalid");
+  const ref = input.reference, wrapped = await snapshotKey(ref);
   const scratch = await mkdtemp(path.join(input.temporaryParent ?? os.tmpdir(), "pc-workspace-decrypt-"));
   try {
     input.signal?.throwIfAborted();