fix(workspaces): cancel stalled snapshot reads and preserve restore refusals (#19894)

Cancel encrypted snapshot reads after five minutes without incoming bytes.
Renew the deadline on progress, preserve caller cancellation, settle the
physical request and private scratch, and retain saved snapshots for normal
retry admission. Keep Node warnings out of server-owned helper refusal labels.

Verified both complete owner suites (145 tests), the prior unchanged storage
and transport coverage, server typechecks, final-head CI and Greptile 5/5.
The earlier CI cancellation-signal assertion failure was reproduced locally
and corrected with checks for abort state, reason, stream closure and cleanup.

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

doc/DEVELOPING.md

@@ -2193,6 +2193,18 @@ or `cleanup`) and its timeout in milliseconds. Command output is excluded from
 that diagnostic. A timeout does not prove the remote process stopped or that a
 save failed before publication. Reconcile the existing workspace and operation
 receipts before recovery; the transport does not automatically repeat the command.
+The internal helper commands suppress Node runtime warnings so their fixed
+failure labels stay separate from warning text. Actual command errors still
+fail the operation. This does not change an agent process's environment.
+
+Loading an encrypted saved snapshot cancels its storage request after five
+minutes without incoming bytes, including a wait for response headers. Incoming
+bytes renew that idle deadline, so a large download can take longer while it
+makes progress. The failure is `remote_workspace_snapshot_download_stalled`;
+private download scratch is removed before returning and the saved snapshot is
+unchanged. Preparation offers its existing bounded retry only after rechecking
+the same clean saved version and workspace owner. Caller cancellation still
+cancels the request and does not become a download timeout.
 Export cleanup keeps this typed timeout diagnostic in the operation's
 `metadata.exportCleanup.lastError`. Other cleanup errors retain a generic message.
 The saved snapshot stays acknowledged and the cleanup obligation stays pending

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

@@ -2,6 +2,9 @@ import { randomUUID } from "node:crypto";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
+import { createServer, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+import { setTimeout as wait } from "node:timers/promises";
 import { Readable } from "node:stream";
 import { appendFileSync, readdirSync } from "node:fs";
 import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
@@ -9,6 +12,7 @@ import { eq } from "drizzle-orm";
 import { applyPendingMigrations, closeRegisteredClients, createDb, executionWorkspaceSnapshots } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
 import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js";
+import { createS3StorageProvider } from "../storage/s3-provider.js";
 import { createStorageService } from "../storage/service.js";
 import type { StorageService } from "../storage/types.js";
 import { localEncryptedProvider } from "../secrets/local-encrypted-provider.js";
@@ -341,7 +345,8 @@ suite("remote workspace snapshot storage", () => {
       destroy(error, callback) { closed = true; callback(error); },
     });
     storage.getObject = vi.fn(async (_companyId, _objectKey, options) => {
-      expect(options?.signal).toBe(controller.signal);
+      expect(options?.signal).toBeInstanceOf(AbortSignal);
+      expect(options?.signal?.aborted).toBe(false);
       return { stream: body, contentLength: ciphertext.length };
     });
     const scratchRoot = path.join(root, "private-scratch"); await fs.mkdir(scratchRoot);
@@ -357,4 +362,98 @@ suite("remote workspace snapshot storage", () => {
       await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
     } finally { temp.mockRestore(); }
   });
+  it.each(["headers", "body"])("closes a stalled S3 %s request before returning a retryable download failure", async phase => {
+    const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload);
+    const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey));
+    let ready!: () => void, closed!: () => void;
+    const started = new Promise<void>(resolve => { ready = resolve; });
+    const disconnected = new Promise<void>(resolve => { closed = resolve; });
+    const server = createServer((req, res) => {
+      req.socket.once("close", closed);
+      if (phase === "headers") { ready(); return; }
+      res.writeHead(200, { "content-length": String(ciphertext.length) });
+      res.write(ciphertext.subarray(0, 1));
+    });
+    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+    vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture-access-key");
+    vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture-secret-key");
+    vi.stubEnv("AWS_SESSION_TOKEN", "");
+    const remote = createStorageService(createS3StorageProvider({ bucket: "bucket", region: "us-east-1", forcePathStyle: true,
+      endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}` }));
+    const get = remote.getObject.bind(remote);
+    remote.getObject = async (...args) => {
+      const object = await get(...args);
+      object.stream.once("data", ready);
+      return object;
+    };
+    const controller = new AbortController(), targetPath = path.join(root, "stalled.pcws");
+    const scratch = path.join(root, "download-scratch"); await fs.mkdir(scratch);
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+    const result = loadWorkspaceSnapshotArchive({ storage: remote, reference: ref, targetPath,
+      signal: controller.signal, temporaryParent: scratch }).then(() => ({ succeeded: true }), error => error);
+    try {
+      await started;
+      await vi.advanceTimersByTimeAsync(300_001);
+      const outcome = await Promise.race([result, wait(100).then(() => ({ pending: true }))]);
+      expect(outcome).toMatchObject({ name: "TimeoutError", message: "remote_workspace_snapshot_download_stalled" });
+      await Promise.race([disconnected, wait(1_000).then(() => { throw new Error("S3 request still connected after timeout"); })]);
+      expect(controller.signal.aborted).toBe(false);
+      expect(await fs.readdir(scratch)).toEqual([]);
+      await expect(fs.access(targetPath)).rejects.toMatchObject({ code: "ENOENT" });
+      expect(await getWorkspaceSnapshot(db, { ...f, snapshotId: ref.id })).toEqual(ref);
+      await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath });
+      expect(await fs.readFile(targetPath)).toEqual(await fs.readFile(f.archivePath));
+    } finally {
+      controller.abort(); await result;
+      vi.useRealTimers(); server.closeAllConnections();
+      await new Promise<void>(resolve => server.close(() => resolve()));
+    }
+  });
+  it("lets an S3 download exceed five minutes while bytes keep arriving", async () => {
+    const f = await fixture(), ref = await uploadWorkspaceSnapshot(f.upload);
+    const ciphertext = await fs.readFile(path.join(root, "objects", ref.objectKey));
+    let response!: ServerResponse, consumed = 0;
+    const observed = Array.from({ length: 4 }, () => {
+      let resolve!: () => void;
+      const promise = new Promise<void>(ready => { resolve = ready; });
+      return { promise, resolve };
+    });
+    const width = Math.ceil(ciphertext.length / 4);
+    const server = createServer((_req, res) => {
+      response = res; res.writeHead(200, { "content-length": String(ciphertext.length) });
+      res.write(ciphertext.subarray(0, width));
+    });
+    await new Promise<void>(resolve => server.listen(0, "127.0.0.1", resolve));
+    vi.stubEnv("AWS_ACCESS_KEY_ID", "fixture-access-key");
+    vi.stubEnv("AWS_SECRET_ACCESS_KEY", "fixture-secret-key");
+    vi.stubEnv("AWS_SESSION_TOKEN", "");
+    const remote = createStorageService(createS3StorageProvider({ bucket: "bucket", region: "us-east-1", forcePathStyle: true,
+      endpoint: `http://127.0.0.1:${(server.address() as AddressInfo).port}` }));
+    const get = remote.getObject.bind(remote);
+    remote.getObject = async (...args) => {
+      const object = await get(...args);
+      object.stream.on("data", () => observed[consumed++]?.resolve());
+      return object;
+    };
+    const controller = new AbortController(), targetPath = path.join(root, "progressing.pcws");
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] });
+    const result = loadWorkspaceSnapshotArchive({ storage: remote, reference: ref, targetPath, signal: controller.signal });
+    const settled = result.catch(error => error);
+    try {
+      await observed[0]!.promise;
+      for (let index = 1; index < 4; index += 1) {
+        await vi.advanceTimersByTimeAsync(240_000);
+        response.write(ciphertext.subarray(index * width, (index + 1) * width));
+        await observed[index]!.promise;
+      }
+      response.end(); await result;
+      expect(await fs.readFile(targetPath)).toEqual(await fs.readFile(f.archivePath));
+      await vi.advanceTimersByTimeAsync(300_001);
+      expect(controller.signal.aborted).toBe(false);
+    } finally {
+      controller.abort(); await settled;
+      vi.useRealTimers(); server.closeAllConnections();
+      await new Promise<void>(resolve => server.close(() => resolve()));
+    }
+  });
 });

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

@@ -17,6 +17,7 @@ const SCHEMA = "paperclip.workspace-snapshot.v1";
 const KEY_SCHEMA = "paperclip.workspace-snapshot-key.v1";
 const MAGIC = Buffer.from("PCWSE1\n");
 const FORMAT = "paperclip.remote-workspace-archive.v1";
+const DOWNLOAD_IDLE_TIMEOUT_MS = 5 * 60_000;
 type SnapshotRow = typeof executionWorkspaceSnapshots.$inferSelect;
 export interface WorkspaceSnapshotRef {
   schema: typeof SCHEMA;
@@ -304,14 +305,25 @@ export async function loadWorkspaceSnapshotArchive(input: {
   try {
     input.signal?.throwIfAborted();
     const ciphertext = path.join(scratch, "ciphertext"), plaintext = path.join(scratch, "plaintext");
-    const object = await input.storage.getObject(ref.companyId, ref.objectKey, { signal: input.signal });
     let received = 0; const digest = createHash("sha256");
-    const bounded = new Transform({ transform(bytes, _encoding, callback) {
-      received += bytes.length; digest.update(bytes);
-      callback(received > ref.byteSize ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes);
-    } });
-    try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600 }), { signal: input.signal }); }
-    finally { object.stream.destroy(); }
+    const stalled = new DOMException("remote_workspace_snapshot_download_stalled", "TimeoutError");
+    const idle = new AbortController();
+    const signal = input.signal ? AbortSignal.any([input.signal, idle.signal]) : idle.signal;
+    const timer = setTimeout(() => idle.abort(stalled), DOWNLOAD_IDLE_TIMEOUT_MS);
+    timer.unref();
+    try {
+      const object = await input.storage.getObject(ref.companyId, ref.objectKey, { signal });
+      const bounded = new Transform({ transform(bytes, _encoding, callback) {
+        if (bytes.length > 0) timer.refresh();
+        received += bytes.length; digest.update(bytes);
+        callback(received > ref.byteSize ? new Error("remote_workspace_snapshot_object_size_mismatch") : null, bytes);
+      } });
+      try { await pipeline(object.stream, bounded, createWriteStream(ciphertext, { flags: "wx", mode: 0o600 }), { signal }); }
+      finally { object.stream.destroy(); }
+    } catch (error) {
+      if (signal.aborted && signal.reason === stalled) throw stalled;
+      throw error;
+    } finally { clearTimeout(timer); }
     if (received !== ref.byteSize || digest.digest("hex") !== ref.sha256) failure("object_hash_mismatch");
     const file = await open(ciphertext, "r"), header = Buffer.alloc(MAGIC.length + 12), tag = Buffer.alloc(16);
     try { await file.read(header, 0, header.length, 0); await file.read(tag, 0, tag.length, ref.byteSize - tag.length); } finally { await file.close(); }

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

@@ -208,7 +208,7 @@ const WORKSPACE_COMMAND_TIMEOUT_MS = 300_000;
 async function command(target: RemoteWorkspaceTarget, operation: RemoteWorkspaceOperation,
   args: string[], env?: Record<string, string>, timeoutMs = WORKSPACE_COMMAND_TIMEOUT_MS) {
   const execute = () => target.runner!.execute({ command: "node", args, cwd: target.remoteCwd,
-    env: { NODE_OPTIONS: "", NODE_PATH: "", ...env }, timeoutMs, bypassSession: true });
+    env: { NODE_OPTIONS: "", NODE_PATH: "", ...env, NODE_NO_WARNINGS: "1" }, timeoutMs, bypassSession: true });
   // Copy-back and restore record why the provider refused the command.
   const result = operation === "capture" || operation === "archive_helper" || operation === "restore"
     ? await execute().catch((error: unknown) => { noteProviderRefusal(operation, error); throw error; })

server/src/services/remote-workspaces.integration.test.ts

@@ -5,6 +5,7 @@ import { createHash, randomUUID } from "node:crypto";
 import { execFile, spawn } from "node:child_process";
 import net from "node:net";
 import { createServer } from "node:http";
+import { Readable } from "node:stream";
 import { promisify } from "node:util";
 import fs from "node:fs/promises";
 import os from "node:os";
@@ -406,7 +407,7 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
     }
   }, 60_000);
 
-  it.each(["same root", "different root", "destroyed allocation", "occupied root", "missing object", "transient restore", "restore lock busy"] as const)(
+  it.each(["same root", "different root", "destroyed allocation", "occupied root", "missing object", "stalled download", "transient restore", "restore lock busy"] as const)(
     "preserves the published workspace path and files on a cold restore (%s)", async mode => {
     const reserved = await reserve({ kind: "empty" }), first = await target("sandbox-A");
     const prepared = await prepareRemoteWorkspace({ db, storage, workspace: reserved, runId: ids.run, lease: await lease(), target: first });
@@ -512,6 +513,29 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
         message: "remote_workspace_restore_failed: step=snapshot_download; remote_workspace_snapshot_object_missing" });
       expect(await fs.lstat(expectedRemoteRoot).catch(() => null)).toBeNull();
     }
+    if (mode === "stalled download") {
+      let ready!: () => void;
+      const started = new Promise<void>(resolve => { ready = resolve; });
+      const body = new Readable({ read() {} });
+      const stalledStorage: StorageService = { ...storage, getObject: async () => { ready(); return { stream: body }; } };
+      const realTimeout = globalThis.setTimeout;
+      const deadline = vi.spyOn(globalThis, "setTimeout").mockImplementation((fn, ms, ...args) =>
+        realTimeout(fn, ms === 300_000 ? 25 : ms, ...args));
+      const restoring = prepareRemoteWorkspace({ db, storage: stalledStorage, workspace: reservedAgain, runId: nextRun, lease: newLease, target: second })
+        .then(() => null, (error: unknown) => error as RemoteWorkspaceRestoreError);
+      try {
+        await started;
+        expect(await restoring).toMatchObject({ step: "snapshot_download", retryable: true,
+          message: "remote_workspace_restore_failed: step=snapshot_download; remote_workspace_snapshot_download_stalled" });
+        expect(body.destroyed).toBe(true);
+        expect(readRemoteWorkspaceAuthority((await workspace(reserved.id)).metadata))
+          .toEqual(readRemoteWorkspaceAuthority(reservedAgain.metadata));
+        expect(await fs.lstat(expectedRemoteRoot).catch(() => null)).toBeNull();
+      } finally {
+        body.destroy(new Error("test cleanup")); await restoring;
+        deadline.mockRestore();
+      }
+    }
     const restored = await prepareRemoteWorkspace({ db, storage, workspace: reservedAgain, runId: nextRun, lease: newLease, target: second });
     expect(restored.remoteRoot).toBe(expectedRemoteRoot);
     if (mode !== "different root") expect(restored.remoteRoot).toBe(prepared.remoteRoot);
@@ -974,10 +998,12 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
     for (const mode of ["abort", "storage_failure"] as const) {
       const controller = new AbortController(), failure = new Error("Saved object unavailable");
       let observedSignal: AbortSignal | undefined;
+      let observedBody: Readable | undefined;
       const interruptedStorage: StorageService = { ...storage, getObject: async (companyId, objectKey, options) => {
         observedSignal = options?.signal;
         if (mode === "storage_failure") throw failure;
         const object = await storage.getObject(companyId, objectKey, options);
+        observedBody = object.stream;
         controller.abort(new DOMException("Saved read cancelled", "AbortError"));
         return object;
       } };
@@ -986,9 +1012,15 @@ suite("remote workspace lifecycle with real Postgres and independent filesystem
       const access = await remoteFiles.open({ companyId: ids.company, workspaceId: reserved.id, signal: controller.signal })
         .catch(error => { readError = error; return null; });
       await access?.close();
-      expect(observedSignal).toBe(controller.signal);
+      expect(observedSignal).toBeInstanceOf(AbortSignal);
+      expect(observedSignal?.aborted).toBe(controller.signal.aborted);
       if (mode === "storage_failure") expect(readError).toBe(failure);
-      else expect(readError).toMatchObject({ name: "AbortError" });
+      else {
+        expect(observedSignal?.reason).toBe(controller.signal.reason);
+        expect(readError).toMatchObject({ name: "AbortError" });
+        expect(observedBody?.destroyed).toBe(true);
+        expect(observedBody?.closed).toBe(true);
+      }
       expect(await fs.readdir(savedTemp)).toEqual([]);
     }
     const cancelled = new DOMException("Already cancelled", "AbortError");