fix(azure-openai-responses): bound SSE response reads via buildGuardedModelFetch (#97349)

wangmiao0668000666 · claude · web-flow · commit 66ffad117610 · 2026-06-28T12:14:53.000-07:00
Inject buildGuardedModelFetch with resolved Azure base URL into both
SDK constructors (OpenAI and AzureOpenAI path) so Azure Responses
requests route through OpenClaw's guarded transport (SSRF policy,
timeout, retry limiting, SSE/JSON synthesis caps).

The non-OK response body cap is applied lazily inside the shared
sanitizeOpenAISdkSseResponse guard via TransformStream — closes the
unbounded response.text() OOM path for hostile 4xx/5xx responses
while preserving the OpenAI SDK's ability to cancel and retry.

- src/llm/providers/azure-openai-responses.ts: pass guardedFetch into
  both OpenAI and AzureOpenAI SDK constructors (no per-provider wrapper)
- src/agents/provider-transport-fetch.ts: add capNonOkResponseBodyLazily
  helper and wire into sanitizeOpenAISdkSseResponse for !response.ok
- src/agents/provider-transport-fetch.test.ts: 2 inline tests
  (cap fires + SDK cancel preserves source)

Co-authored-by: Claude &lt;noreply@anthropic.com&gt;
diff --git a/src/agents/provider-transport-fetch.test.ts b/src/agents/provider-transport-fetch.test.ts
@@ -1465,6 +1465,90 @@ describe("buildGuardedModelFetch", () => {
     expect(String(caught)).toMatch(/exceeded.*bytes while synthesizing SSE/i);
   });
 
+  it("caps non-OK response body lazily so SDK can still cancel retryable responses", async () => {
+    // Regression: a 429/5xx non-OK body used to be returned unchanged by the
+    // shared sanitizer, so a hostile endpoint could OOM the SDK when it called
+    // response.text(). The cap is applied lazily via TransformStream so the SDK
+    // can still cancel the response before reading any body.
+    const OVER_LIMIT = 100 * 1024;
+    fetchWithSsrFGuardMock.mockResolvedValue({
+      response: new Response(
+        new ReadableStream({
+          start(controller) {
+            controller.enqueue(new Uint8Array(OVER_LIMIT));
+            controller.close();
+          },
+        }),
+        { status: 429, statusText: "Too Many Requests" },
+      ),
+      finalUrl: "https://custom-azure.openai.azure.com/openai/v1/responses",
+      release: vi.fn(async () => undefined),
+    });
+    const model = {
+      id: "gpt-5.5",
+      provider: "azure",
+      api: "azure-openai-responses",
+      baseUrl: "https://custom-azure.openai.azure.com/openai/v1",
+    } as unknown as Model<"azure-openai-responses">;
+
+    const response = await buildGuardedModelFetch(model)(
+      "https://custom-azure.openai.azure.com/openai/v1/responses",
+      { method: "POST" },
+    );
+
+    expect(response.status).toBe(429);
+    expect(response.ok).toBe(false);
+
+    const text = await response.text();
+    expect(text.length).toBeLessThanOrEqual(64 * 1024);
+    expect(text.length).toBeLessThan(OVER_LIMIT);
+  });
+
+  it("preserves SDK ability to cancel retryable non-OK responses before reading body", async () => {
+    // Regression: a non-OK body wrapper must be lazy. The OpenAI SDK may decide
+    // to cancel a 429/5xx response and retry before reading the body. If the
+    // wrapper eagerly reads the body, the SDK loses that capability.
+    const TOTAL_BYTES = 80 * 1024;
+    let bytesPulled = 0;
+    fetchWithSsrFGuardMock.mockResolvedValue({
+      response: new Response(
+        new ReadableStream({
+          pull(controller) {
+            if (bytesPulled < TOTAL_BYTES) {
+              const chunk = new Uint8Array(8 * 1024);
+              bytesPulled += chunk.byteLength;
+              controller.enqueue(chunk);
+            } else {
+              controller.close();
+            }
+          },
+        }),
+        { status: 503, statusText: "Service Unavailable" },
+      ),
+      finalUrl: "https://custom-azure.openai.azure.com/openai/v1/responses",
+      release: vi.fn(async () => undefined),
+    });
+    const model = {
+      id: "gpt-5.5",
+      provider: "azure",
+      api: "azure-openai-responses",
+      baseUrl: "https://custom-azure.openai.azure.com/openai/v1",
+    } as unknown as Model<"azure-openai-responses">;
+
+    const response = await buildGuardedModelFetch(model)(
+      "https://custom-azure.openai.azure.com/openai/v1/responses",
+      { method: "POST" },
+    );
+
+    expect(response.status).toBe(503);
+    // SDK cancels the response body before reading anything. The lazy cap must
+    // not pull the full body from the source — a small pre-buffered chunk is
+    // allowed (ReadableStream default high-water-mark), but the full payload
+    // must remain unconsumed so the SDK can still retry.
+    await response.body?.cancel();
+    expect(bytesPulled).toBeLessThan(TOTAL_BYTES);
+  });
+
   describe("long retry-after handling", () => {
     const anthropicModel = {
       id: "sonnet-4.6",
diff --git a/src/agents/provider-transport-fetch.ts b/src/agents/provider-transport-fetch.ts
@@ -96,14 +96,44 @@ function findSseEventBoundary(buffer: string): { index: number; length: number }
   return best;
 }
 
+function capNonOkResponseBodyLazily(response: Response, maxBytes: number): Response {
+  const source = response.body;
+  if (!source) {
+    return response;
+  }
+  let total = 0;
+  const capped = source.pipeThrough(
+    new TransformStream<Uint8Array, Uint8Array>({
+      transform(chunk, controller) {
+        const nextTotal = total + chunk.byteLength;
+        if (nextTotal > maxBytes) {
+          const remaining = maxBytes - total;
+          if (remaining > 0) {
+            controller.enqueue(chunk.subarray(0, remaining));
+          }
+          total = maxBytes;
+          controller.terminate();
+          return;
+        }
+        total = nextTotal;
+        controller.enqueue(chunk);
+      },
+    }),
+  );
+  return new Response(capped, response);
+}
+
 function sanitizeOpenAISdkSseResponse(
   response: Response,
   options?: { synthesizeJsonAsSse?: boolean },
 ): Response {
   const contentType = response.headers.get("content-type") ?? "";
-  if (!response.ok || !response.body) {
+  if (!response.body) {
     return response;
   }
+  if (!response.ok) {
+    return capNonOkResponseBodyLazily(response, SSE_SANITIZE_BUFFER_MAX_BYTES);
+  }
   if (
     options?.synthesizeJsonAsSse === true &&
     (/\bapplication\/json\b/i.test(contentType) || /\+json\b/i.test(contentType))
diff --git a/src/llm/providers/azure-openai-responses.ts b/src/llm/providers/azure-openai-responses.ts
@@ -1,6 +1,7 @@
 // Azure OpenAI Responses provider adapts Azure deployments to Responses API streams.
 import OpenAI, { AzureOpenAI } from "openai";
 import type { ResponseCreateParamsStreaming } from "openai/resources/responses/responses.js";
+import { buildGuardedModelFetch } from "../../agents/provider-transport-fetch.js";
 import { isOpenAICompatibleAzureResponsesBaseUrl } from "../../shared/azure-openai-responses-client-compat.js";
 import { getEnvApiKey } from "../env-api-keys.js";
 import type {
@@ -198,13 +199,15 @@ function createClient(
   }
 
   const { baseUrl, apiVersion } = resolveAzureConfig(model, options);
+  const guardedFetch = buildGuardedModelFetch({ ...model, baseUrl });
 
   if (isOpenAICompatibleAzureResponsesBaseUrl(baseUrl)) {
     return new OpenAI({
       apiKey,
       dangerouslyAllowBrowser: true,
       defaultHeaders: headers,
       baseURL: baseUrl,
+      fetch: guardedFetch,
     });
   }
 
@@ -214,6 +217,7 @@ function createClient(
     dangerouslyAllowBrowser: true,
     defaultHeaders: headers,
     baseURL: baseUrl,
+    fetch: guardedFetch,
   });
 }
 
