fix(workspaces): recover unpublished archived Git uploads (#19459)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Remote workspace imports preserve existing files before the server changes workspace authority. > - Archived Git imports reconstruct saved files and Git state in a private repository. > - A failed upload can leave a pending snapshot journal with no object in storage. > - A later reconstruction can have different Git administrative bytes for the same saved logical source. > - This pull request lets that import record a new unpublished capture after it proves that prior objects are absent. > - The benefit is safe retry through the original import journal while source checks and atomic publication remain in place. ## Linked Issues or Issue Description Refs #19396. Refs #19409. A snapshot upload can fail before object persistence. The caller then removes its private archive. A retry reconstructs the same saved Git workspace. If its administrative bytes differ, the snapshot owner rejects the retry with `remote_workspace_snapshot_operation_content_changed`. The pending journal remains blocked even though no snapshot object or remote authority was published. ## What Changed - Add an optional unpublished recapture capability to snapshot storage. Bind it to the original source operation. Existing callers keep the exact manifest check. - Enable this capability only for archived Git reconstruction. The migration owner keeps the source recipe, task binding, ownership, and publication checks. - Verify every prior object before new capture admission. Unknown or corrupt objects stop the retry. Reuse an available prior object, including an object that appears while the replacement upload runs. - Compare the observed upload journal under the snapshot row lock before appending an attempt. Keep network I/O outside that lock. Preserve all prior attempt metadata. - Keep the first atomic publication as the winner. A losing uploader verifies and returns that saved reference. - Add real PostgreSQL, filesystem, and Git regression tests. Document the retry behavior. ## Verification - Baseline `70d23612dd9176cae8ee6b2b6dcf2f6eade9a42a`: the new real Git fixture fails with `remote_workspace_snapshot_operation_content_changed`. It fails the first upload before persistence, removes the old source cache, and advances unrelated remote refs. The saved task commit and restore point stay fixed. - Final frozen source: 20 tests pass in a fresh process. This includes the real Git regression and all snapshot storage cases. Tests cover prior object recovery, corrupt and unknown objects, normal caller mismatch rejection, operation binding, stale journal admission, and late upload publication in both orders. - Reproduce the final focused check from `server/`: ```sh node ../node_modules/vitest/vitest.mjs run --config vitest.config.ts src/services/remote-workspace-snapshot-storage.test.ts src/services/remote-workspace-migration.test.ts -t 'failed-upload-retry|snapshot storage' --reporter=dot ``` - An earlier six-suite run had 201 passing cases and one failing late-object case. That run started before the final prior-object recheck was added. The failing case passes in the fresh final run above. This earlier run is not a clean full-family result for the final source. - That run skipped 16 lifecycle cases because `PAPERCLIP_REMOTE_WORKSPACE_LIFECYCLE_TEST_DATABASE_URL` was unset. Migration, storage, reservation, and concurrency tests used real disposable embedded PostgreSQL. - `git diff --check` passes. Full repository typecheck passed before rebasing. Full test, build, and current-head CI results are pending. ## Risks - The optional capability permits different physical captures. The migration owner must prove that the logical source and current ownership are unchanged. Ordinary mutable workspace uploads keep the previous mismatch guard. - An old upload can complete after the last object check. The first authenticated atomic publication wins. A late upload cannot replace its object or snapshot reference. Every candidate remains in the retention journal. - Unknown storage results and journal changes fail closed. They can require another canonical retry. This change adds no retry loop and changes no database schema. ## Model Used OpenAI Codex, GPT-6 family. The worker used reasoning, tool use, code editing, and local test execution. The worker runtime did not expose an exact model ID or context window. ## 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 - [ ] All Paperclip CI gates are green - [ ] 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 <[email protected]>
2026-11-07 06:16:06 -10:00 · 2026-11-06 08:04:35 -06:00
1 parent 2a6aae3963
commit 6cafcaf89b
@@ -1997,6 +1997,13 @@ project with the same repository URL; publication locks and rechecks those sourc
 rows. Imports from mutable local folders still recapture and compare the
 current files; a saved upload does not permit later local edits to be discarded.

+If every recorded snapshot upload is definitively absent, an unpublished
+archived Git import can recapture the same validated logical source. Private Git
+administrative bytes can differ between reconstructions. The original journal
+and every upload attempt remain recorded; available prior uploads and concurrent
+publication winners are reused. Unknown or corrupt objects block recapture.
+The source recipe and ownership checks still gate authority publication.
+
 When the verified Git source still exists, reconstruction also copies its local
 `info/exclude` rules into the private repository before applying a saved restore
 point. This keeps saved ignored files, including recovery receipts and bundles,
@@ -727,7 +727,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
    const address = server.address(); if (!address || typeof address === "string") throw new Error("test server not bound");
    return { bare, repoUrl: `http://127.0.0.1:${address.port}/repository.git` };
  }
-  it.each(["present", "root-history", "pruned", "retry", "source-excludes", "source-excludes-symlink", "missing-base", "missing-base-retry", "missing-base-uploaded-retry", "missing-base-changed-retry", "nested", "nested-linked", "nested-retry", "nested-source-changed-retry", "unapproved-nested", "missing", "expired", "corrupt"])("preserves archived Git edits or refuses an unavailable restore point (%s)", async state => {
+  it.each(["present", "root-history", "pruned", "retry", "missing-base-failed-upload-retry", "source-excludes", "source-excludes-symlink", "missing-base", "missing-base-retry", "missing-base-uploaded-retry", "missing-base-changed-retry", "nested", "nested-linked", "nested-retry", "nested-source-changed-retry", "unapproved-nested", "missing", "expired", "corrupt"])("preserves archived Git edits or refuses an unavailable restore point (%s)", async state => {
    const remote = state.startsWith("missing-base") ? await serveRepository() : null;
    const f = await gitArchiveFixture(remote?.repoUrl);
    if (remote) { await git(remote.bare, "fetch", f.base, "main:refs/heads/main"); await git(remote.bare, "update-server-info"); }
@@ -838,6 +838,20 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
      storage.putFile = async input => { const result = await put(input); if (fail) { fail = false; throw new Error("upload reply lost"); } return result; };
      await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("upload reply lost");
    }
+    let failedAttempt: Record<string, unknown> | undefined;
+    if (state === "missing-base-failed-upload-retry") {
+      const put = storage.putFile.bind(storage); let fail = true;
+      storage.putFile = async input => {
+        if (fail && input.namespace === "workspace-snapshots") { fail = false; throw new Error("upload_part InternalError"); }
+        return put(input);
+      };
+      await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("upload_part InternalError");
+      const [pending] = await db.select().from(executionWorkspaceSnapshots);
+      expect(pending).toMatchObject({ status: "pending", objectKey: null });
+      failedAttempt = (pending!.metadata!.uploads as Record<string, unknown>[])[0]!;
+      expect((await storage.headObject(f.companyId, failedAttempt.objectKey as string)).exists).toBe(false);
+      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) => {
@@ -886,6 +900,16 @@ describe.skipIf(!support.supported)("remote workspace migration", () => {
      return;
    }
    await migrateRemoteWorkspace({ db, storage, ...f, apply: true });
+    if (failedAttempt) {
+      const [saved] = await db.select().from(executionWorkspaceSnapshots);
+      const attempts = saved!.metadata!.uploads as Record<string, unknown>[];
+      expect(saved!.status).toBe("committed");
+      expect(attempts).toHaveLength(2);
+      expect(attempts[0]).toEqual(failedAttempt);
+      expect(attempts[1]!.manifestDigest).not.toBe(failedAttempt.manifestDigest);
+      expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_migration")))
+        .toEqual([expect.objectContaining({ id: saved!.sourceOperationId, status: "succeeded" })]);
+    }
    if (interrupted) {
      const operations = await db.select().from(workspaceOperations).where(eq(workspaceOperations.phase, "remote_workspace_migration"));
      expect(operations).toHaveLength(1);
@@ -376,7 +376,12 @@ export async function migrateRemoteWorkspace(input: {
      const exclude = nested.map(child => child.relativePath);
      const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, signal: input.signal });
      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 });
+        companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal,
+        // 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 });
      await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal });
@@ -75,6 +75,128 @@ suite("remote workspace snapshot storage", () => {
    expect(row?.status).toBe("pending"); expect((row?.metadata?.uploads as unknown[]).length).toBe(1);
    await expect(uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64) })).rejects.toThrow("operation_content_changed");
  });
+  it("recaptures only an unavailable unpublished operation while retaining every attempt", async () => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    storage.putFile = async () => { throw new Error("upload_part InternalError"); };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload_part InternalError");
+    const [pending] = await db.select().from(executionWorkspaceSnapshots);
+    const original = (pending!.metadata!.uploads as Record<string, unknown>[])[0]!;
+    await fs.rm(f.archivePath);
+    await fs.writeFile(path.join(f.cwd, "saved.txt"), "equivalent logical source, different physical capture\n");
+    const captured = await captureRemoteWorkspaceDirectory({ cwd: f.cwd, archivePath: f.archivePath });
+    const retry = { ...f.upload, manifestDigest: captured.manifestDigest, unpublishedRecapture: { sourceOperationId: f.operationId } };
+    await expect(uploadWorkspaceSnapshot({ ...retry, unpublishedRecapture: { sourceOperationId: randomUUID() } })).rejects.toThrow("operation_binding_changed");
+    storage.putFile = put;
+    const ref = await uploadWorkspaceSnapshot(retry);
+    expect(ref.manifestDigest).toBe(captured.manifestDigest);
+    const [saved] = await db.select().from(executionWorkspaceSnapshots);
+    expect(saved!.metadata!.uploads).toEqual([original, expect.objectContaining({ objectKey: ref.objectKey, manifestDigest: captured.manifestDigest })]);
+    // The capability does not reopen a published snapshot.
+    await expect(uploadWorkspaceSnapshot({ ...retry, manifestDigest: f.captured.manifestDigest })).rejects.toThrow("operation_content_changed");
+  });
+  it("prefers an available prior capture over a different recapture", async () => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply");
+    await fs.rm(f.archivePath);
+    const ref = await uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } });
+    expect(ref.manifestDigest).toBe(f.captured.manifestDigest);
+    expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(1);
+  });
+  it("rechecks prior objects after encryption before recording a recapture", async () => {
+    const f = await fixture(), put = storage.putFile.bind(storage), head = storage.headObject.bind(storage);
+    let savedInput!: Parameters<StorageService["putFile"]>[0];
+    storage.putFile = async input => {
+      const parts: Buffer[] = [];
+      for await (const bytes of input.body) parts.push(Buffer.from(bytes));
+      savedInput = { ...input, body: Buffer.concat(parts) };
+      throw new Error("upload interrupted before persistence");
+    };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload interrupted before persistence");
+    let heads = 0;
+    storage.headObject = async (...args) => { if (++heads === 2) await put(savedInput); return head(...args); };
+    storage.putFile = put;
+    const ref = await uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } });
+    expect(ref.manifestDigest).toBe(f.captured.manifestDigest);
+    expect(heads).toBe(2);
+    expect((await db.select().from(executionWorkspaceSnapshots))[0]!.metadata!.uploads).toHaveLength(1);
+  });
+  it("refuses stale recapture admission when another attempt enters the ledger", async () => {
+    const f = await fixture(), head = storage.headObject.bind(storage);
+    storage.putFile = async () => { throw new Error("upload unavailable"); };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload unavailable");
+    let ready!: () => void, release!: () => void;
+    const paused = new Promise<void>(resolve => { ready = resolve; }), resume = new Promise<void>(resolve => { release = resolve; });
+    let heads = 0;
+    storage.headObject = async (...args) => { if (++heads === 1) { ready(); await resume; } return head(...args); };
+    const stale = uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } });
+    try {
+      await paused;
+      await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("upload unavailable");
+      release();
+      await expect(stale).rejects.toThrow("upload_attempts_changed");
+      const [saved] = await db.select().from(executionWorkspaceSnapshots);
+      expect(saved).toMatchObject({ status: "pending", objectKey: null });
+      expect(saved!.metadata!.uploads).toHaveLength(2);
+    } finally { release(); await Promise.allSettled([stale]); }
+  });
+  it.each(["unknown", "corrupt"])("refuses recapture when a prior object is %s", async state => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("lost upload reply");
+    if (state === "unknown") storage.headObject = async () => { throw new Error("HEAD unavailable"); };
+    else {
+      const [pending] = await db.select().from(executionWorkspaceSnapshots);
+      const [attempt] = pending!.metadata!.uploads as { objectKey: string }[];
+      await fs.writeFile(path.join(root, "objects", attempt!.objectKey), "corrupt object");
+    }
+    await expect(uploadWorkspaceSnapshot({ ...f.upload, manifestDigest: "b".repeat(64), unpublishedRecapture: { sourceOperationId: f.operationId } }))
+      .rejects.toThrow(state === "unknown" ? "HEAD unavailable" : "object_hash_mismatch");
+    const [saved] = await db.select().from(executionWorkspaceSnapshots);
+    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 => {
+    const f = await fixture(), put = storage.putFile.bind(storage);
+    const deferred = () => {
+      let resolve!: () => void;
+      const promise = new Promise<void>(ready => { resolve = ready; });
+      return { resolve, promise };
+    };
+    const originalReady = deferred(), originalRelease = deferred(), recaptureReady = deferred(), recaptureRelease = deferred();
+    let calls = 0;
+    storage.putFile = async input => {
+      const first = ++calls === 1;
+      if (first) { originalReady.resolve(); await originalRelease.promise; }
+      else { recaptureReady.resolve(); await recaptureRelease.promise; }
+      const saved = await put(input);
+      if (first && winner === "original-with-lost-reply") throw new Error("late original upload reply lost");
+      return saved;
+    };
+    const unpublishedRecapture = { sourceOperationId: f.operationId };
+    const original = uploadWorkspaceSnapshot({ ...f.upload, unpublishedRecapture });
+    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 });
+      await recaptureReady.promise;
+      (winner === "recapture" ? recaptureRelease : originalRelease).resolve();
+      if (winner === "original-with-lost-reply") {
+        await expect(original).rejects.toThrow("late original upload reply lost");
+        recaptureRelease.resolve();
+      }
+      const saved = await (winner === "original" ? original : recapture);
+      expect(saved.manifestDigest).toBe(winner === "recapture" ? captured.manifestDigest : f.captured.manifestDigest);
+      originalRelease.resolve(); recaptureRelease.resolve();
+      if (winner !== "original-with-lost-reply") expect(await original).toEqual(saved);
+      expect(await recapture).toEqual(saved);
+      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]); }
+  });
  it("authenticates a lost-reply archive without publishing its pending upload", async () => {
    const f = await fixture(), put = storage.putFile.bind(storage);
    storage.putFile = async input => { await put(input); throw new Error("lost upload reply"); };
@@ -5,6 +5,7 @@ import os from "node:os";
 import path from "node:path";
 import { pipeline } from "node:stream/promises";
 import { Transform } from "node:stream";
+import { isDeepStrictEqual } from "node:util";
 import { and, eq } from "drizzle-orm";
 import { executionWorkspaceSnapshots, type Db } from "@paperclipai/db";
 import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
@@ -124,15 +125,27 @@ export async function recoverWorkspaceSnapshotArchive(input: {
 export async function uploadWorkspaceSnapshot(input: {
  db: Db; storage: StorageService; snapshotId: string; companyId: string; workspaceId: string;
  archivePath: string; manifestDigest: string; signal?: AbortSignal;
+  // 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> {
  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;
  }
  if (row!.status !== "pending" || !/^[a-f0-9]{64}$/.test(input.manifestDigest)) failure("not_writable");
+  const recapture = Boolean(input.unpublishedRecapture);
+  const requireUnpublished = (current: SnapshotRow) => {
+    if (current.sourceOperationId !== input.unpublishedRecapture!.sourceOperationId
+      || current.objectKey !== null || current.sha256 !== null || current.byteSize !== null
+      || current.key !== null || current.manifestDigest !== null) failure("not_unpublished");
+  };
+  if (recapture) requireUnpublished(row!);
  const identity = { id: row!.id, companyId: row!.companyId, workspaceId: row!.executionWorkspaceId, generation: row!.generation };
  const publish = async (attempt: UploadAttempt) => {
    const updated = await input.db.transaction(async tx => {
@@ -143,20 +156,27 @@ export async function uploadWorkspaceSnapshot(input: {
    const winning = updated ?? (await input.db.select().from(executionWorkspaceSnapshots).where(scope(input)))[0];
    if (!winning || !["uploaded", "committed", "superseded"].includes(winning.status)) failure("publication_changed");
    const saved = reference(winning!);
-    if (saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
+    if (!recapture && saved.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
+    if (saved.objectKey !== attempt.objectKey) await verifyObject(input.storage, saved, input.signal);
    return saved;
  };
  // Recover a completed upload even when the previous process lost its reply
  // and its scratch directory. Failed attempts remain referenced for cleanup.
-  const attempts = Array.isArray(row!.metadata?.uploads) ? row!.metadata!.uploads as UploadAttempt[] : [];
-  for (const attempt of attempts) {
-    if (attempt.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
-    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 publish(attempt);
+  const availableAttempt = async (attempts: UploadAttempt[]) => {
+    for (const attempt of attempts) {
+      input.signal?.throwIfAborted();
+      if (!recapture && attempt.manifestDigest !== input.manifestDigest) failure("operation_content_changed");
+      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;
+      }
    }
-  }
+    return null;
+  };
+  const attempts = Array.isArray(row!.metadata?.uploads) ? row!.metadata!.uploads as UploadAttempt[] : [];
+  const available = await availableAttempt(attempts);
+  if (available) return publish(available);
  const scratch = await mkdtemp(path.join(os.tmpdir(), "pc-workspace-encrypt-"));
  try {
    const encrypted = path.join(scratch, "archive.encrypted"), dataKey = randomBytes(32), iv = randomBytes(12);
@@ -175,17 +195,28 @@ export async function uploadWorkspaceSnapshot(input: {
    const content = await digestFile(encrypted, input.signal);
    const attempt: UploadAttempt = { objectKey: `${identity.companyId}/workspace-snapshots/${identity.workspaceId}/${identity.id}/${randomUUID()}.pcws`,
      ...content, key: material, manifestDigest: input.manifestDigest };
+    // Recheck after encryption, then bind admission to the unchanged ledger.
+    // Remote I/O stays outside the row lock. An older upload completing after
+    // this check can still win the pending-to-uploaded publication below.
+    if (recapture) {
+      const recovered = await availableAttempt(attempts);
+      if (recovered) return publish(recovered);
+    }
    const recorded = await input.db.transaction(async tx => {
      const [current] = await tx.select().from(executionWorkspaceSnapshots).where(scope(input)).for("update");
-      if (!current || current.status !== "pending") return false;
-      const previous = Array.isArray(current.metadata?.uploads) ? current.metadata.uploads : [];
+      if (!current || current.status !== "pending") return null;
+      const previous = Array.isArray(current.metadata?.uploads) ? current.metadata.uploads as UploadAttempt[] : [];
+      if (recapture) {
+        requireUnpublished(current);
+        if (!isDeepStrictEqual(previous, attempts)) failure("upload_attempts_changed");
+      } else if (previous.some(saved => saved.manifestDigest !== input.manifestDigest)) failure("operation_content_changed");
      await notifyRemoteWorkspaceSnapshots(tx, `workspace:${input.workspaceId}`);
      await tx.update(executionWorkspaceSnapshots).set({ metadata: { ...current.metadata, uploads: [...previous, attempt] }, updatedAt: new Date() }).where(scope(input));
      return true;
    });
    if (!recorded) {
      const saved = await getWorkspaceSnapshot(input.db, input);
-      if (!saved || saved.manifestDigest !== input.manifestDigest) failure("publication_changed");
+      if (!saved || (!recapture && saved.manifestDigest !== input.manifestDigest)) failure("publication_changed");
      await verifyObject(input.storage, saved, input.signal); return saved;
    }
    // The object key is unique per attempt, so a late losing uploader cannot
@@ -194,6 +225,10 @@ export async function uploadWorkspaceSnapshot(input: {
      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);
+    if (recapture) {
+      const recovered = await availableAttempt(attempts);
+      if (recovered) return publish(recovered);
+    }
    return await publish(attempt);
  } finally { await rm(scratch, { recursive: true, force: true }); }
 }