perf(storage): upload workspace snapshots in parallel verified parts (#19917)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - With remote (sandbox) execution, every agent turn ends by saving the
workspace: capture an archive, encrypt it, upload it to object storage,
and verify it.
> - On a busy S3-backed instance, the upload step is the slowest part of
a save. A 1.5 GB workspace can hold a run open for minutes after the
agent finished.
> - The upload step reads the archive several times, uploads through one
stream with 4 parts in flight, and then downloads the whole object again
to verify it.
> - Parts that use the stream path share one process-wide 32 MiB buffer
window, so saves that run at the same time slow each other down.
> - This pull request uploads parts in parallel straight from the
encrypted file, lets the backend verify every part with Content-MD5, and
replaces the full read-back with a size check for those verified
uploads.
> - The benefit is faster saves, a smaller run-finalization window, and
no interference between concurrent saves.

## Linked Issues or Issue Description

**What existing behavior does this improve?**
Remote workspace snapshot upload (`uploadWorkspaceSnapshot`) to
S3-compatible storage.

**Subsystem affected**
Server storage (`server/src/storage/*`) and remote workspace snapshot
storage.

**Current behavior**
For one save, the upload step:
1. hashes the archive;
2. reads the archive again to encrypt it into a temporary file;
3. hashes the encrypted file;
4. uploads it as one stream with 4 parts in flight;
5. downloads the whole object to verify it.

On one hosted instance over 7 days (724 saves, median 731 MB), the
upload step ran at a median of 6.5 MB/s. With no other save running it
reached 11–14 MB/s. With six or more saves at the same time it fell to
4.4 MB/s. From the same region and with the same part settings, a plain
client reached 31 MB/s to the same kind of bucket.

**Proposed behavior**
Upload parts in parallel from byte ranges of the encrypted file, with
Content-MD5 on each part. Confirm only the stored size after a verified
upload. Digest the archive in the same pass that encrypts it.

**Reason and benefit**
Saves finish faster and hold their run open for less time. Concurrent
saves no longer serialize on the shared buffer window. A transient part
failure is retried from disk instead of failing the whole save.

**Breaking changes**
None. `StorageProvider.putObject` may now return a receipt. Providers
that return nothing keep the current behavior, including the full
read-back.

Related: none found.

## What Changed

- `storage/types.ts`, `storage/service.ts`: an upload can name its
`sourceFile`, a private file whose bytes are exactly the body.
`putObject` may return `{ integrity: "server_verified_md5" }`, and
`putFile` passes it through as `PutFileResult.integrity`.
- `storage/s3-multipart.ts`: new `putS3MultipartFile`.
- Each part reads its own byte range of the file. Up to 8 parts upload
at once, with no shared buffer window.
  - Each part carries the Content-MD5 of its range.
  - A failed part is resent from disk, up to 3 attempts.
- The first part that runs out of attempts stops its siblings. The
multipart upload is aborted only after every part settles.
- A source file whose size differs from the declared length is refused
before any request.
- `storage/s3-provider.ts`:
- A `sourceFile` larger than 16 MiB uses `putS3MultipartFile`. A smaller
one uses one PUT with Content-MD5.
- These requests use a second client with `requestChecksumCalculation:
"WHEN_REQUIRED"`, so bodies are plain and the backend verifies
Content-MD5 directly. The default client wraps bodies in aws-chunked
framing with a CRC32 trailer.
  - Other uploads are unchanged.
- `services/remote-workspace-snapshot-storage.ts`:
- Encryption digests the plaintext in the same read. An unchanged file
identity (dev, inode, size, mtime, ctime) across that read replaces the
separate digest pass. A change still fails with `archive_changed` before
any upload.
- The encrypted file is still digested from disk, so the recorded hash
covers the exact bytes that the parts read.
- The upload passes `sourceFile`. When the provider reports
`server_verified_md5`, a HEAD confirms the stored size instead of
downloading the object.
  - The full read-back stays for:
    - other providers (local disk);
    - recovery of earlier attempts;
    - the caller that keeps a verified archive file.

## Verification

- `pnpm --filter @paperclipai/server typecheck`: exit 0.
- `vitest run src/__tests__/storage-s3-multipart.test.ts
src/services/remote-workspace-snapshot-storage.test.ts`: 59 passed. New
tests:
- every part carries the Content-MD5 of its range, and 8 parts are in
flight;
  - a failed part is resent and the upload completes;
- a part that keeps failing aborts the upload only after in-flight
siblings settle;
  - a size mismatch is refused before any request;
- a small file goes as one PUT with Content-MD5, through a real HTTP
server;
- the read-back is skipped only for a server-verified upload, and the
snapshot still restores;
  - the read-back stays when the caller keeps the read-back file;
  - a stored size mismatch is rejected;
  - an archive that changes during encryption is rejected before upload.
- With the source change reverted, the three new snapshot tests fail.
- Real S3-compatible bucket (a Railway bucket, run from a workstation):
- A 3 MiB single PUT stored exactly: HEAD size, ETag and read-back
SHA-256 all matched.
  - A 96 MiB, 13-part upload stored exactly, with the same checks.
  - A part with a wrong Content-MD5 was rejected with `BadDigest`.
- The same backend stored a wrong `x-amz-checksum-sha256` without
checking it. That is why this change uses Content-MD5.
- Measured from a container in the app's region, 1 GiB with the app's
settings (8 MiB parts, 4 in flight), to the same kind of bucket: 31
MB/s. With 16 in flight: 69 MB/s.
- 77 storage-related suites: the only failures are the same 19 that fail
on unmodified `master` in this checkout. They are in 4 files: missing
bundled connector skills and an unbuilt local runner daemon.

## Risks

- Integrity now depends on the backend enforcing Content-MD5 for file
uploads. AWS S3 and the tested S3-compatible backend enforce it. A
backend that ignores Content-MD5 would accept corrupted parts without a
read-back. A later restore still fails closed: AES-GCM authentication
and the plaintext SHA-256 are checked on load.
- Up to 8 parallel part requests per save, instead of 4 shared ones.
Memory per part is a 64 KiB stream buffer.
- Archive-change detection now uses file identity instead of a second
digest. The archive lives in a private temporary directory that the
capture step creates.
- Medium risk overall: the change is limited to file-sourced S3 uploads
and the snapshot verify step.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Claude Opus 5.5 (`claude-opus-5-5`) in Claude Code, with tool use
(shell, file edits, test runs, S3 API probes). AI-assisted
investigation, change and tests, with human direction.

## 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
- [ ] I have updated relevant documentation to reflect my changes (no
docs describe this path)
- [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 <noreply@paperclip.ing>

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

@@ -19,6 +19,17 @@ 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";
 
+// Pass-through, so one test can report a changed archive between the two
+// identity reads that bracket encryption.
+const lstatOverride = vi.hoisted(() => ({ current: null as null | ((file: string, real: (file: string) => Promise<import("node:fs").Stats>) => Promise<import("node:fs").Stats>) }));
+vi.mock("node:fs/promises", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("node:fs/promises")>();
+  const lstat = ((file: Parameters<typeof actual.lstat>[0], ...rest: unknown[]) => lstatOverride.current
+    ? lstatOverride.current(String(file), target => actual.lstat(target))
+    : (actual.lstat as (...args: unknown[]) => unknown)(file, ...rest)) as typeof actual.lstat;
+  return { ...actual, lstat, default: { ...actual, lstat } };
+});
+
 const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
 if (externalDatabaseUrl && !/^remote_workspace_[a-z_]*test$/.test(new URL(externalDatabaseUrl).pathname.slice(1))) {
   throw new Error("Workspace storage tests require a disposable remote_workspace_*test database");
@@ -136,6 +147,51 @@ suite("remote workspace snapshot storage", () => {
     expect(await fs.lstat(verifiedArchivePath).catch(() => null)).toBeNull();
     expect((await fs.readdir(root)).some(name => name.startsWith(".workspace-upload-readback-"))).toBe(false);
   });
+  it("confirms only the stored size when the backend verified every uploaded part", async () => {
+    const f = await fixture(randomBytes(2 * 1024 * 1024 + 91)); const put = storage.putFile.bind(storage);
+    let sourceFile: string | undefined;
+    storage.putFile = async input => {
+      sourceFile = "sourceFile" in input ? input.sourceFile : undefined;
+      return { ...await put(input), integrity: "server_verified_md5" };
+    };
+    const get = vi.spyOn(storage, "getObject"), head = vi.spyOn(storage, "headObject");
+    const ref = await uploadWorkspaceSnapshot(f.upload);
+    expect(sourceFile).toMatch(/archive\.encrypted$/);
+    expect(get).not.toHaveBeenCalled();
+    expect(head).toHaveBeenCalledWith(f.companyId, ref.objectKey, expect.anything());
+    const restoredArchive = path.join(root, "restored-verified.pcws");
+    await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: restoredArchive });
+    expect((await fs.readFile(restoredArchive)).equals(await fs.readFile(f.archivePath))).toBe(true);
+  });
+  it("still reads back a server-verified upload when the caller keeps the read-back file", async () => {
+    const f = await fixture(); const put = storage.putFile.bind(storage);
+    storage.putFile = async input => ({ ...await put(input), integrity: "server_verified_md5" });
+    const get = vi.spyOn(storage, "getObject"), verifiedArchivePath = path.join(root, "kept.pcws");
+    await uploadWorkspaceSnapshot({ ...f.upload, verifiedArchivePath });
+    expect(get).toHaveBeenCalled();
+    expect((await fs.readFile(verifiedArchivePath)).equals(await fs.readFile(f.archivePath))).toBe(true);
+  });
+  it("rejects a server-verified upload whose stored size differs", async () => {
+    const f = await fixture(); const put = storage.putFile.bind(storage), head = storage.headObject.bind(storage);
+    storage.putFile = async input => ({ ...await put(input), integrity: "server_verified_md5" });
+    storage.headObject = async (...args) => { const result = await head(...args); return { ...result, contentLength: (result.contentLength ?? 0) + 1 }; };
+    await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("object_size_mismatch");
+    const [row] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, f.row.id));
+    expect(row?.status).toBe("pending");
+  });
+  it("rejects an archive that changes while it is encrypted, before any upload", async () => {
+    const f = await fixture(); const put = vi.spyOn(storage, "putFile");
+    let reads = 0;
+    lstatOverride.current = async (file, real) => {
+      const stats = await real(file);
+      if (file !== f.archivePath || ++reads < 2) return stats;
+      return Object.assign(Object.create(Object.getPrototypeOf(stats)), stats, { mtimeMs: stats.mtimeMs + 1 });
+    };
+    try { await expect(uploadWorkspaceSnapshot(f.upload)).rejects.toThrow("archive_changed"); }
+    finally { lstatOverride.current = null; }
+    expect(reads).toBe(2);
+    expect(put).not.toHaveBeenCalled();
+  });
   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; };

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

@@ -59,6 +59,10 @@ async function digestFile(file: string, signal?: AbortSignal) {
   for await (const part of createReadStream(file, { signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES })) { digest.update(part); byteSize += part.length; }
   return { sha256: digest.digest("hex"), byteSize };
 }
+async function verifyStoredSize(storage: StorageService, ref: WorkspaceSnapshotRef, signal?: AbortSignal) {
+  const head = await storage.headObject(ref.companyId, ref.objectKey, { signal });
+  if (!head.exists || head.contentLength !== ref.byteSize) failure("object_size_mismatch");
+}
 async function verifyObject(storage: StorageService, ref: WorkspaceSnapshotRef, signal?: AbortSignal) {
   const object = await storage.getObject(ref.companyId, ref.objectKey, { signal });
   const digest = createHash("sha256"); let size = 0;
@@ -135,7 +139,7 @@ type WorkspaceSnapshotUpload = {
 /** 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));
+    return uploadVerifiedWorkspaceSnapshot(input, ref => verifyObject(input.storage, ref, input.signal), { acceptServerVerifiedUpload: true });
   input.signal?.throwIfAborted();
   const target = input.verifiedArchivePath, parent = path.dirname(target);
   if (!path.isAbsolute(target) || path.normalize(target) !== target) failure("readback_parent_invalid");
@@ -160,7 +164,8 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P
     verified.push({ ref: structuredClone(ref), file });
   };
   try {
-    const ref = await uploadVerifiedWorkspaceSnapshot(input, verify);
+    // This caller keeps the read-back file, so every upload is read back.
+    const ref = await uploadVerifiedWorkspaceSnapshot(input, verify, { acceptServerVerifiedUpload: false });
     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);
@@ -183,7 +188,8 @@ export async function uploadWorkspaceSnapshot(input: WorkspaceSnapshotUpload): P
   } finally { await rm(scratch, { recursive: true, force: true }); }
 }
 
-async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise<void>): Promise<WorkspaceSnapshotRef> {
+async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, verify: (ref: WorkspaceSnapshotRef) => Promise<void>,
+  options: { acceptServerVerifiedUpload: boolean }): 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");
@@ -235,17 +241,23 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v
   try {
     const encrypted = path.join(scratch, "archive.encrypted"), dataKey = randomBytes(32), iv = randomBytes(12);
     const cipher = createCipheriv("aes-256-gcm", dataKey, iv); cipher.setAAD(aad(identity));
-    const plaintext = await digestFile(input.archivePath, input.signal);
-    const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ schema: KEY_SCHEMA,
-      ...identity, dataKey: dataKey.toString("base64"), plaintextSha256: plaintext.sha256, plaintextBytes: plaintext.byteSize }) });
     const header = await open(encrypted, "wx", 0o600);
     try { await header.writeFile(Buffer.concat([MAGIC, iv])); } finally { await header.close(); }
+    // One read of the archive both digests and encrypts it. An unchanged file
+    // identity across that read replaces the separate digest pass.
+    const before = await lstat(input.archivePath);
     const incomingDigest = createHash("sha256"); let incomingBytes = 0;
     const observed = new Transform({ transform(bytes, _encoding, callback) { incomingDigest.update(bytes); incomingBytes += bytes.length; callback(null, bytes); } });
     await pipeline(createReadStream(input.archivePath, { signal: input.signal, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), observed, cipher, createWriteStream(encrypted, { flags: "a", mode: 0o600, highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }), { signal: input.signal });
-    if (incomingBytes !== plaintext.byteSize || incomingDigest.digest("hex") !== plaintext.sha256) failure("archive_changed");
+    const after = await lstat(input.archivePath);
+    if (!before.isFile() || before.isSymbolicLink() || incomingBytes !== before.size || after.dev !== before.dev || after.ino !== before.ino
+      || after.size !== before.size || after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs) failure("archive_changed");
+    const plaintext = { sha256: incomingDigest.digest("hex"), byteSize: incomingBytes };
     const footer = await open(encrypted, "a"); try { await footer.writeFile(cipher.getAuthTag()); } finally { await footer.close(); }
+    const { material } = await localEncryptedProvider.createVersion({ value: JSON.stringify({ schema: KEY_SCHEMA,
+      ...identity, dataKey: dataKey.toString("base64"), plaintextSha256: plaintext.sha256, plaintextBytes: plaintext.byteSize }) });
     dataKey.fill(0);
+    // Digest the encrypted file as stored: upload parts are read from these bytes.
     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 };
@@ -275,10 +287,15 @@ async function uploadVerifiedWorkspaceSnapshot(input: WorkspaceSnapshotUpload, v
     }
     // 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,
+    const stored = await input.storage.putFile({ companyId: identity.companyId, namespace: "workspace-snapshots", objectKey: attempt.objectKey,
       originalFilename: null, contentType: "application/octet-stream", body: createReadStream(encrypted, { highWaterMark: WORKSPACE_ARCHIVE_IO_BUFFER_BYTES }),
-      byteSize: attempt.byteSize, sha256: attempt.sha256, signal: input.signal });
-    await verify({ schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef);
+      byteSize: attempt.byteSize, sha256: attempt.sha256, sourceFile: encrypted, signal: input.signal });
+    const uploaded = { schema: SCHEMA, ...identity, ...attempt } as WorkspaceSnapshotRef;
+    // The backend checked every part of this upload against digests of these
+    // exact bytes, so only its size needs confirming. Any other provider, and
+    // a caller that needs the bytes back, still reads the object back.
+    if (options.acceptServerVerifiedUpload && stored.integrity === "server_verified_md5") await verifyStoredSize(input.storage, uploaded, input.signal);
+    else await verify(uploaded);
     if (recapture) {
       const recovered = await availableAttempt(attempts);
       if (recovered) return publish(recovered);