fix(web): mark partial response reads when streams fail (#102550)

* fix(web): mark partial response reads when streams fail

* fix(web): report partial stream bytes accurately

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/searxng/src/searxng-client.test.ts

@@ -150,6 +150,30 @@ describe("searxng client", () => {
     });
   });
 
+  it("rejects partial response bodies without blaming the size limit", async () => {
+    const chunk = new TextEncoder().encode("partial");
+    let sentChunk = false;
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (!sentChunk) {
+          sentChunk = true;
+          controller.enqueue(chunk);
+          return;
+        }
+        controller.error(new Error("stream reset"));
+      },
+    });
+    endpointMockState.responses.push(new Response(stream, { status: 200 }));
+
+    await expect(
+      runSearxngSearch({
+        baseUrl: "http://127.0.0.1:8888",
+        query: "openclaw",
+        categories: "general",
+      }),
+    ).rejects.toThrow("SearXNG response incomplete after 7 bytes.");
+  });
+
   it("detects category searches that should retry with general", () => {
     expect(testing.shouldRetryEmptyCategorySearchWithGeneral("weather")).toBe(true);
     expect(testing.shouldRetryEmptyCategorySearchWithGeneral("weather,news")).toBe(true);

extensions/searxng/src/searxng-client.ts

@@ -222,7 +222,7 @@ async function fetchSearxngResults(params: {
 
       const body = await readResponseText(response, { maxBytes: MAX_RESPONSE_BYTES });
       if (body.truncated) {
-        throw new Error("SearXNG response too large.");
+        throw new Error(`SearXNG response incomplete after ${body.bytesRead} bytes.`);
       }
       return parseSearxngResponseText(body.text, params.count);
     },

src/agents/tools/web-fetch.ts

@@ -645,7 +645,7 @@ async function runWebFetch(params: WebFetchRuntimeParams): Promise<Record<string
     throwIfFetchAborted(params.signal);
     const body = bodyResult.text;
     const responseTruncatedWarning = bodyResult.truncated
-      ? `Response body truncated after ${params.maxResponseBytes} bytes.`
+      ? `Response body incomplete after ${bodyResult.bytesRead} bytes.`
       : undefined;
 
     let title: string | undefined;

src/agents/tools/web-shared.test.ts

@@ -159,6 +159,25 @@ describe("readResponseText", () => {
     expect(releaseLock).toHaveBeenCalledTimes(1);
   });
 
+  it("marks bounded response readers truncated after stream errors", async () => {
+    const cancel = vi.fn(async () => undefined);
+    const releaseLock = vi.fn();
+    const response = responseFromReader({
+      chunks: ["partial"],
+      cancel,
+      releaseLock,
+      readError: new Error("stream reset"),
+    });
+
+    await expect(readResponseText(response, { maxBytes: 64 })).resolves.toEqual({
+      text: "partial",
+      truncated: true,
+      bytesRead: 7,
+    });
+    expect(cancel).toHaveBeenCalledTimes(1);
+    expect(releaseLock).toHaveBeenCalledTimes(1);
+  });
+
   it("does not mark exact-limit streamed responses as truncated", async () => {
     const cancel = vi.fn(async () => undefined);
     const releaseLock = vi.fn();

src/agents/tools/web-shared.ts

@@ -295,7 +295,8 @@ export async function readResponseText(
         }
       }
     } catch {
-      // Best-effort: return whatever we read so far.
+      // Stream errors mean the accumulated bytes are only a partial body.
+      truncated = true;
     } finally {
       if (truncated) {
         // Some mocked or non-compliant streams never settle cancel(); do not

src/agents/tools/web-tools.fetch.test.ts

@@ -669,7 +669,41 @@ describe("web_fetch extraction fallbacks", () => {
     });
     const result = await tool?.execute?.("call", { url: "https://example.com/stream" });
     const details = result?.details as { warning?: string } | undefined;
-    expect(details?.warning).toContain("Response body truncated");
+    expect(details?.warning).toContain("Response body incomplete after 32000 bytes");
+  });
+
+  it("reports the retained byte count when a response stream fails", async () => {
+    const chunk = new TextEncoder().encode("partial");
+    let sentChunk = false;
+    const stream = new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (!sentChunk) {
+          sentChunk = true;
+          controller.enqueue(chunk);
+          return;
+        }
+        controller.error(new Error("stream reset"));
+      },
+    });
+    installMockFetch((input: RequestInfo | URL) =>
+      Promise.resolve(
+        responseWithUrl(
+          stream,
+          { status: 200, headers: { "content-type": "text/plain; charset=utf-8" } },
+          resolveRequestUrl(input),
+        ),
+      ),
+    );
+
+    const tool = createFetchTool({
+      maxResponseBytes: 64,
+      firecrawl: { enabled: false },
+    });
+    const result = await tool?.execute?.("call", { url: "https://example.com/reset" });
+    const details = result?.details as { text?: string; warning?: string } | undefined;
+
+    expect(details?.text).toContain("partial");
+    expect(details?.warning).toContain("Response body incomplete after 7 bytes");
   });
 
   it("keeps DNS pinning for web_fetch by default even when HTTP_PROXY is configured", async () => {