fix(storage): name the S3 read when a response body is cut off (#19284)

Record read_object_body with HTTP status, S3 request id and attempts when a GetObject body stream errors after send() resolved, and add the storage fields to run failure execution diagnostics. Diagnostics only; no retry behavior changes.

server/src/__tests__/storage-s3-failure-diagnostics.test.ts

@@ -3,6 +3,8 @@ import type { AddressInfo } from "node:net";
 import { Readable } from "node:stream";
 import { CreateMultipartUploadCommand, S3Client, UploadPartCommand } from "@aws-sdk/client-s3";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import type { heartbeatRuns } from "@paperclipai/db";
+import { collectRunFailureDiagnostics } from "../services/run-failure-diagnostics.js";
 import { storageFailureDiagnosticsFor } from "../storage/failure-diagnostics.js";
 import { createS3StorageProvider } from "../storage/s3-provider.js";
 import { putS3Multipart } from "../storage/s3-multipart.js";
@@ -82,6 +84,43 @@ describe("S3 storage failure diagnostics", () => {
     await expect(s3.headObject({ objectKey: "missing" })).resolves.toEqual({ exists: false });
   });
 
+  it("names a response body that was cut off after its headers arrived", async () => {
+    handler = (req, res) => {
+      res.writeHead(200, { "content-type": "application/octet-stream", "content-length": "1024", "x-amz-request-id": "req-body-1" });
+      res.write(Buffer.alloc(16), () => req.socket.destroy());
+    };
+    const object = await provider().getObject({ objectKey: "company/snapshot.bin" });
+    const error = await (async () => { for await (const _chunk of object.stream) { /* drain */ } })().catch((err: unknown) => err);
+
+    expect(error).toMatchObject({ message: "aborted", code: "ECONNRESET" });
+    expect(storageFailureDiagnosticsFor(error)).toEqual({
+      provider: "s3",
+      operation: "read_object_body",
+      errorCode: "ECONNRESET",
+      httpStatus: 200,
+      requestId: "req-body-1",
+      attempts: 1,
+    });
+    const report = collectRunFailureDiagnostics(
+      { id: "run", errorCode: "setup_failed", resultJson: null } as typeof heartbeatRuns.$inferSelect,
+      { error: new Error("Workspace restore failed", { cause: error }), phase: "setup" },
+    );
+    expect(report.execution).toMatchObject({
+      storageOperation: "read_object_body",
+      storageErrorCode: "ECONNRESET",
+      storageHttpStatus: 200,
+      storageRequestId: "req-body-1",
+    });
+  });
+
+  it("does not handle a body error that its consumer left unhandled", async () => {
+    handler = (_req, res) => res.end("ok");
+    const { stream } = await provider().getObject({ objectKey: "k" });
+    const failure = new Error("unhandled body failure");
+    expect(() => stream.emit("error", failure)).toThrow(failure);
+    expect(storageFailureDiagnosticsFor(failure)).toMatchObject({ operation: "read_object_body" });
+  });
+
   it("names the multipart step that failed and ignores unsafe provider fields", async () => {
     const client = new S3Client({ region: "us-east-1" });
     const failure = Object.assign(new Error("private provider message"), {

server/src/services/run-failure-diagnostics.ts

@@ -13,6 +13,7 @@ import {
   ACP_TOOL_KINDS,
 } from "@paperclipai/adapter-utils/acpx-engine/constants";
 import { redactCurrentUserText } from "../log-redaction.js";
+import { storageFailureDiagnosticsFor } from "../storage/failure-diagnostics.js";
 import { redactSensitiveText, REDACTED_EVENT_VALUE } from "../redaction.js";
 import { readNativeModelRejectionDiagnostic, readNativeTurnFailureDiagnostic } from "./native-runtime/native-provider-failure.js";
 import { readRemoteRunnerRecoveryDiagnostic, readRemoteRunnerRecoveryError } from "./native-runtime/remote-runner-recovery-diagnostics.js";
@@ -435,6 +436,17 @@ export function collectRunFailureDiagnostics(run: Run, options: RunFailureReport
         }
       }
     }
+    // A bare socket error ("aborted", ECONNRESET) cannot say which storage
+    // read it interrupted; the provider's fixed-shape record can.
+    const storage = storageFailureDiagnosticsFor(error);
+    if (storage && execution.storageOperation === undefined) {
+      execution.storageProvider = storage.provider;
+      execution.storageOperation = storage.operation;
+      execution.storageErrorCode = storage.errorCode;
+      if (storage.httpStatus !== null) execution.storageHttpStatus = storage.httpStatus;
+      if (storage.requestId !== null) execution.storageRequestId = storage.requestId;
+      if (storage.attempts !== null) execution.storageAttempts = storage.attempts;
+    }
     const entry: RunFailureException = {};
     for (const field of ["name", "message", "stack", "code"] as const) {
       const value = read(error, field);

server/src/storage/failure-diagnostics.ts

@@ -5,6 +5,7 @@
 export type StorageFailureOperation =
   | "put_object"
   | "get_object"
+  | "read_object_body"
   | "head_object"
   | "delete_object"
   | "create_multipart_upload"

server/src/storage/s3-failure.ts

@@ -36,6 +36,23 @@ export function recordS3Failure(error: unknown, operation: StorageFailureOperati
   });
 }
 
+/**
+ * A GetObject body can fail after send() resolved, for example when the
+ * socket resets mid-download. The response names the request; the stream
+ * error's own code (ECONNRESET) names the failure better than its name.
+ */
+export function recordS3BodyFailure(error: unknown, response: unknown): void {
+  const metadata = ownValue(response, "$metadata");
+  recordStorageFailure(error, {
+    provider: "s3",
+    operation: "read_object_body",
+    errorCode: matching(ownValue(error, "code"), ERROR_CODE) ?? matching(ownValue(error, "name"), ERROR_CODE) ?? "unknown",
+    httpStatus: integerIn(ownValue(metadata, "httpStatusCode"), 100, 599),
+    requestId: matching(ownValue(metadata, "requestId"), REQUEST_ID),
+    attempts: integerIn(ownValue(metadata, "attempts"), 1, 100),
+  });
+}
+
 export async function withS3FailureDiagnostics<T>(operation: StorageFailureOperation, send: () => Promise<T>): Promise<T> {
   try {
     return await send();

server/src/storage/s3-provider.ts

@@ -8,7 +8,7 @@ import {
   ListObjectVersionsCommand,
 } from "@aws-sdk/client-s3";
 import { putS3Multipart } from "./s3-multipart.js";
-import { withS3FailureDiagnostics } from "./s3-failure.js";
+import { recordS3BodyFailure, withS3FailureDiagnostics } from "./s3-failure.js";
 import { addAbortSignal, Readable } from "node:stream";
 import type { StorageProvider, GetObjectResult, HeadObjectResult } from "./types.js";
 import { notFound, unprocessable } from "../errors.js";
@@ -167,6 +167,12 @@ export function createS3StorageProvider(config: S3ProviderConfig): StorageProvid
         ));
 
         const stream = await toReadableStream(output.Body);
+        // Observe, do not handle: a consumer without an error listener still
+        // sees the failure exactly as it would without this one.
+        stream.on("error", function recordBodyFailure(error) {
+          recordS3BodyFailure(error, output);
+          if (stream.listenerCount("error") === 1) throw error;
+        });
         if (input.signal) addAbortSignal(input.signal, stream);
         return {
           stream,