fix: isolate request-scoped HTTP POST failures

src/transports/http-stream.test.ts

@@ -1187,6 +1187,242 @@ describe("createHttpStream", () => {
     }
   });
 
+  it("fails only the request when its connected POST returns an HTTP error", async () => {
+    const controlledFetch = createControlledFetch({
+      respond: (request) =>
+        request.method === "POST" && JSON.parse(request.body).id === 4
+          ? new Response(JSON.stringify({ error: "boom" }), { status: 500 })
+          : undefined,
+    });
+    const stream = createHttpStream("https://agent.example/acp", {
+      fetch: controlledFetch.fetch,
+    });
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    try {
+      await writer.write(initializeRequest);
+      await readMessage(reader);
+      await controlledFetch.sendSse(0, sessionNewResponse);
+      await readMessage(reader);
+      await writer.write(promptRequest);
+
+      await writer.write({ ...promptRequest, id: 4 });
+      expect(await readMessage(reader)).toEqual({
+        jsonrpc: "2.0",
+        id: 4,
+        error: {
+          code: -32603,
+          message: expect.stringContaining("ACP POST failed: 500"),
+          data: { status: 500, body: { error: "boom" } },
+        },
+      });
+
+      // The shared connection and the other in-flight request keep working.
+      expect(
+        controlledFetch.requests.some((request) => request.method === "DELETE"),
+      ).toBe(false);
+      expect(sseAt(controlledFetch.sseRequests, 0).signal.aborted).toBe(false);
+      expect(sseAt(controlledFetch.sseRequests, 1).signal.aborted).toBe(false);
+      const promptResponse = {
+        jsonrpc: "2.0",
+        id: 2,
+        result: { stopReason: "end_turn" },
+      } satisfies AnyMessage;
+      await controlledFetch.sendSse(1, promptResponse);
+      expect(await readMessage(reader)).toEqual(promptResponse);
+
+      await writer.write({ ...promptRequest, id: 5 });
+      const nextPost = requestAt(
+        controlledFetch.requests,
+        controlledFetch.requests.length - 1,
+      );
+      expect(JSON.parse(nextPost.body)).toMatchObject({ id: 5 });
+    } finally {
+      reader.releaseLock();
+      writer.releaseLock();
+      await closeStream(stream);
+    }
+  });
+
+  it.each([403, 404, 410])(
+    "tears down the connection when a connected POST returns %i",
+    async (status) => {
+      const controlledFetch = createControlledFetch({
+        postStatus: status,
+        postBody: "Unknown Acp-Connection-Id",
+      });
+      const stream = createHttpStream("https://agent.example/acp", {
+        fetch: controlledFetch.fetch,
+      });
+      const writer = stream.writable.getWriter();
+      const reader = stream.readable.getReader();
+
+      try {
+        await writer.write(initializeRequest);
+        await readMessage(reader);
+
+        await expect(writer.write(sessionNewRequest)).rejects.toThrow(
+          `ACP POST failed: ${status}`,
+        );
+        expect(requestAt(controlledFetch.requests, 3).method).toBe("DELETE");
+        await expect(reader.read()).rejects.toThrow("ACP POST failed");
+      } finally {
+        reader.releaseLock();
+        writer.releaseLock();
+        await closeStream(stream);
+      }
+    },
+  );
+
+  it("tears down the connection when a notification POST fails", async () => {
+    const controlledFetch = createControlledFetch({
+      respond: (request) =>
+        request.method === "POST" &&
+        JSON.parse(request.body).method === "session/cancel"
+          ? new Response("boom", { status: 500 })
+          : undefined,
+    });
+    const stream = createHttpStream("https://agent.example/acp", {
+      fetch: controlledFetch.fetch,
+    });
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    try {
+      await writer.write(initializeRequest);
+      await readMessage(reader);
+
+      await expect(
+        writer.write({
+          jsonrpc: "2.0",
+          method: "session/cancel",
+          params: { sessionId: "session-1" },
+        }),
+      ).rejects.toThrow("ACP POST failed: 500");
+      expect(
+        controlledFetch.requests.some((request) => request.method === "DELETE"),
+      ).toBe(true);
+    } finally {
+      reader.releaseLock();
+      writer.releaseLock();
+      await closeStream(stream);
+    }
+  });
+
+  it.each([
+    { maxMessageBytes: undefined, bodyBytes: 8 * 1024 },
+    { maxMessageBytes: 1024, bodyBytes: 1024 },
+  ])(
+    "truncates connected POST error bodies to $bodyBytes bytes",
+    async ({ maxMessageBytes, bodyBytes }) => {
+      const controlledFetch = createControlledFetch({
+        postStatus: 500,
+        postBody: "x".repeat(20_000),
+      });
+      const stream = createHttpStream("https://agent.example/acp", {
+        fetch: controlledFetch.fetch,
+        maxMessageBytes,
+      });
+      const writer = stream.writable.getWriter();
+      const reader = stream.readable.getReader();
+
+      try {
+        await writer.write(initializeRequest);
+        await readMessage(reader);
+
+        await writer.write(sessionNewRequest);
+        expect(await readMessage(reader)).toMatchObject({
+          id: 1,
+          error: { data: { status: 500, body: "x".repeat(bodyBytes) } },
+        });
+      } finally {
+        reader.releaseLock();
+        writer.releaseLock();
+        await closeStream(stream);
+      }
+    },
+  );
+
+  it("drops a multi-byte character split by error body truncation", async () => {
+    const controlledFetch = createControlledFetch({
+      postStatus: 500,
+      postBody: `${"x".repeat(8 * 1024 - 1)}é`,
+    });
+    const stream = createHttpStream("https://agent.example/acp", {
+      fetch: controlledFetch.fetch,
+    });
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    try {
+      await writer.write(initializeRequest);
+      await readMessage(reader);
+
+      await writer.write(sessionNewRequest);
+      expect(await readMessage(reader)).toMatchObject({
+        error: { data: { body: "x".repeat(8 * 1024 - 1) } },
+      });
+    } finally {
+      reader.releaseLock();
+      writer.releaseLock();
+      await closeStream(stream);
+    }
+  });
+
+  it("does not answer a request twice when its POST fails after it settled", async () => {
+    const releasePost = createDeferred<void>();
+    const controlledFetch = createControlledFetch({
+      respond: (request) =>
+        request.method === "POST" && JSON.parse(request.body).id === 6
+          ? new Response("boom", { status: 500 })
+          : undefined,
+    });
+    const fetch: typeof globalThis.fetch = async (input, init) => {
+      const response = await controlledFetch.fetch(input, init);
+      if (
+        init?.method === "POST" &&
+        bodyToString(init.body).includes('"id":2')
+      ) {
+        await releasePost.promise;
+        return new Response("boom", { status: 500 });
+      }
+
+      return response;
+    };
+    const stream = createHttpStream("https://agent.example/acp", { fetch });
+    const writer = stream.writable.getWriter();
+    const reader = stream.readable.getReader();
+
+    try {
+      await writer.write(initializeRequest);
+      await readMessage(reader);
+      await controlledFetch.sendSse(0, sessionNewResponse);
+      await readMessage(reader);
+
+      const write = writer.write(promptRequest);
+      await flushMicrotasks();
+      await controlledFetch.closeSse(1);
+      expect(await readMessage(reader)).toMatchObject({
+        id: 2,
+        error: { message: "ACP session SSE stream closed" },
+      });
+
+      releasePost.resolve();
+      await write;
+
+      await writer.write({ ...promptRequest, id: 6 });
+      expect(await readMessage(reader)).toMatchObject({
+        id: 6,
+        error: { data: { status: 500 } },
+      });
+    } finally {
+      reader.releaseLock();
+      writer.releaseLock();
+      await closeStream(stream);
+    }
+  });
+
   it("errors and tears down local resources when connected POST fails", async () => {
     const clearSpy = vi.spyOn(MemoryAcpCookieStore.prototype, "clear");
     const controlledFetch = createControlledFetch({

src/transports/http-stream.ts

@@ -44,6 +44,17 @@ const SESSION_REATTACH_METHODS: ReadonlySet<string> = new Set([
   AGENT_METHODS.session_resume,
 ]);
 
+/**
+ * POST statuses that mean the whole connection is unusable: credentials were
+ * rejected, or the server no longer knows the `Acp-Connection-Id`.
+ */
+const CONNECTION_FAILURE_STATUSES: ReadonlySet<number> = new Set([
+  401, 403, 404, 410,
+]);
+
+/** Error bodies are diagnostic text, so they are capped well below ACP messages. */
+const MAX_HTTP_ERROR_BODY_BYTES = 8 * 1024;
+
 export interface HttpStreamOptions {
   /** Fetch implementation to use. Defaults to `globalThis.fetch`. */
   readonly fetch?: typeof globalThis.fetch;
@@ -87,6 +98,14 @@ export {
  * `session/load` or `session/resume`. If a session's SSE stream closes while
  * requests are in flight, those requests fail and the next message for the
  * session reopens its stream.
+ *
+ * If a request's POST is rejected with any other HTTP error, only that request
+ * fails: it receives a JSON-RPC error whose `data` is `{ status, body }`, with
+ * the body truncated to 8 KiB. HTTP 401, 403, 404, and 410 instead mean the
+ * connection is unusable or gone, and close the transport, as do network
+ * errors and failed POSTs of notifications or responses. A request failed
+ * this way may still have been processed by the agent (e.g. a gateway 502
+ * after the agent accepted it).
  */
 export function createHttpStream(
   serverUrl: string,
@@ -294,7 +313,8 @@ class HttpStreamTransport {
         throw await httpError(
           "ACP POST failed",
           response,
-          this.maxMessageBytes,
+          Math.min(this.maxMessageBytes, MAX_HTTP_ERROR_BODY_BYTES),
+          { truncate: true },
         );
       }
       void response.body?.cancel().catch(() => {});
@@ -306,16 +326,31 @@ class HttpStreamTransport {
         }
       }
     } catch (error) {
-      if (pendingSessionRequestKey) {
-        this.pendingSessionRequests.delete(pendingSessionRequestKey);
-      }
+      // A session request may already have been settled while its POST was in
+      // flight (its response arrived or its session stream closed).
+      const isStillPending = pendingSessionRequestKey
+        ? this.pendingSessionRequests.delete(pendingSessionRequestKey)
+        : true;
 
       if (sessionId && isSessionConflict(error)) {
         this.failSession(sessionId, error);
         this.dropSessionMessage(message, error);
         return;
       }
 
+      if (
+        error instanceof HttpResponseError &&
+        !CONNECTION_FAILURE_STATUSES.has(error.status) &&
+        "method" in message &&
+        "id" in message
+      ) {
+        // Reject only this request without poisoning the shared writable stream.
+        if (isStillPending) {
+          this.enqueueRequestError(message.id, error);
+        }
+        return;
+      }
+
       this.errorReadable(error);
       throw error;
     }
@@ -575,7 +610,7 @@ class HttpStreamTransport {
     for (const [key, pending] of this.pendingSessionRequests) {
       if (pending.sessionId === sessionId) {
         this.pendingSessionRequests.delete(key);
-        this.enqueueSessionError(pending.id, error);
+        this.enqueueRequestError(pending.id, error);
       }
     }
   }
@@ -595,7 +630,7 @@ class HttpStreamTransport {
     }
 
     if ("method" in message) {
-      this.enqueueSessionError(message.id, error);
+      this.enqueueRequestError(message.id, error);
       return;
     }
 
@@ -605,7 +640,7 @@ class HttpStreamTransport {
     }
   }
 
-  private enqueueSessionError(id: string | number | null, error: Error): void {
+  private enqueueRequestError(id: string | number | null, error: Error): void {
     this.enqueue({
       jsonrpc: "2.0",
       id,
@@ -753,6 +788,7 @@ function resolveFetch(
 async function readResponseText(
   response: Response,
   maxMessageBytes: number,
+  { truncate = false }: { truncate?: boolean } = {},
 ): Promise<string> {
   if (!response.body) {
     return "";
@@ -766,6 +802,12 @@ async function readResponseText(
       if (done) {
         return new TextDecoder().decode(buffer.take());
       }
+      if (truncate && value.byteLength >= maxMessageBytes - buffer.byteLength) {
+        buffer.append(value.subarray(0, maxMessageBytes - buffer.byteLength));
+        void reader.cancel().catch(() => {});
+        // Streaming decode drops a multi-byte character split by the cut.
+        return new TextDecoder().decode(buffer.take(), { stream: true });
+      }
       buffer.append(value);
     }
   } catch (error) {
@@ -808,10 +850,11 @@ async function httpError(
   prefix: string,
   response: Response,
   maxMessageBytes: number,
+  options: { truncate?: boolean } = {},
 ): Promise<Error> {
   let text = "";
   try {
-    text = await readResponseText(response, maxMessageBytes);
+    text = await readResponseText(response, maxMessageBytes, options);
   } catch (error) {
     if (error instanceof MessageTooLargeError) {
       throw error;