From 6cd94f6049b9ce90a663ae3442d0cfbcdee386fe Mon Sep 17 00:00:00 2001 From: qingminlong Date: Thu, 9 Jul 2026 18:58:56 +0800 Subject: [PATCH] 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 --- extensions/searxng/src/searxng-client.test.ts | 24 +++++++++++++ extensions/searxng/src/searxng-client.ts | 2 +- src/agents/tools/web-fetch.ts | 2 +- src/agents/tools/web-shared.test.ts | 19 ++++++++++ src/agents/tools/web-shared.ts | 3 +- src/agents/tools/web-tools.fetch.test.ts | 36 ++++++++++++++++++- 6 files changed, 82 insertions(+), 4 deletions(-) diff --git a/extensions/searxng/src/searxng-client.test.ts b/extensions/searxng/src/searxng-client.test.ts index 4ed889c403cd..e608e375ffc3 100644 --- a/extensions/searxng/src/searxng-client.test.ts +++ b/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({ + 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); diff --git a/extensions/searxng/src/searxng-client.ts b/extensions/searxng/src/searxng-client.ts index e07e040676ba..52848f831d4b 100644 --- a/extensions/searxng/src/searxng-client.ts +++ b/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); }, diff --git a/src/agents/tools/web-fetch.ts b/src/agents/tools/web-fetch.ts index 8780f7378e55..a22476f4e4e9 100644 --- a/src/agents/tools/web-fetch.ts +++ b/src/agents/tools/web-fetch.ts @@ -645,7 +645,7 @@ async function runWebFetch(params: WebFetchRuntimeParams): Promise { 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(); diff --git a/src/agents/tools/web-shared.ts b/src/agents/tools/web-shared.ts index a40ee1f9a057..582cf0a0d1d8 100644 --- a/src/agents/tools/web-shared.ts +++ b/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 diff --git a/src/agents/tools/web-tools.fetch.test.ts b/src/agents/tools/web-tools.fetch.test.ts index 7f0045690688..475e50246b88 100644 --- a/src/agents/tools/web-tools.fetch.test.ts +++ b/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({ + 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 () => {