fix(openai): bound embedding-batch and realtime session JSON response reads

hugenshen · hugenshen · commit db255d677fcc · 2026-06-29T00:58:46.000+08:00
diff --git a/extensions/openai/embedding-batch.test.ts b/extensions/openai/embedding-batch.test.ts
@@ -266,6 +266,70 @@ describe("OpenAI embedding batch output", () => {
     ]);
   });
 
+  it("bounds batch status success body via readProviderJsonResponse", async () => {
+    const chunkSize = 1024 * 1024;
+    const chunkCount = 20; // 20 MiB, well over 16 MiB cap
+    let readCount = 0;
+    let canceled = false;
+    const oversizedStatus = new Response(
+      new ReadableStream<Uint8Array>({
+        pull(controller) {
+          if (readCount >= chunkCount) {
+            controller.close();
+            return;
+          }
+          readCount += 1;
+          controller.enqueue(new Uint8Array(chunkSize));
+        },
+        cancel() {
+          canceled = true;
+        },
+      }),
+      { status: 200, headers: { "Content-Type": "application/json" } },
+    );
+    let batchStatusCalled = false;
+    const fetchImpl = vi.fn(async (input: RequestInfo | URL, init?: RequestInit) => {
+      const url = fetchInputUrl(input);
+      if (url.endsWith("/files") && init?.method === "POST") {
+        return jsonResponse({ id: "file-0" });
+      }
+      if (url.endsWith("/batches") && init?.method === "POST") {
+        return jsonResponse({ id: "batch-0", status: "in_progress" });
+      }
+      if (url.endsWith("/batches/batch-0") && !batchStatusCalled) {
+        batchStatusCalled = true;
+        return oversizedStatus;
+      }
+      return new Response("unexpected request", { status: 500 });
+    });
+
+    await expect(
+      runOpenAiEmbeddingBatches({
+        openAi: {
+          baseUrl: "https://openai-compatible.example/v1",
+          headers: { Authorization: "Bearer test" },
+          model: "text-embedding-3-small",
+          fetchImpl,
+        },
+        agentId: "main",
+        requests: [
+          {
+            custom_id: "0",
+            method: "POST",
+            url: "/v1/embeddings",
+            body: { model: "text-embedding-3-small", input: "payload" },
+          },
+        ],
+        wait: true,
+        concurrency: 1,
+        pollIntervalMs: 1000,
+        timeoutMs: 60_000,
+      }),
+    ).rejects.toThrow(/openai\.batch-status/);
+    expect(canceled).toBe(true);
+    expect(readCount).toBeLessThan(chunkCount);
+  });
+
   it("bounds batch resource error bodies without using response.text()", async () => {
     const tracked = cancelTrackedResponse(`${"batch status unavailable ".repeat(1024)}tail`, {
       status: 400,
diff --git a/extensions/openai/embedding-batch.ts b/extensions/openai/embedding-batch.ts
@@ -18,7 +18,10 @@ import {
   uploadBatchJsonlFile,
   withRemoteHttpResponse,
 } from "openclaw/plugin-sdk/memory-core-host-engine-embeddings";
-import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http";
+import {
+  readProviderJsonResponse,
+  readResponseTextLimited,
+} from "openclaw/plugin-sdk/provider-http";
 import { normalizeStringEntries } from "openclaw/plugin-sdk/string-coerce-runtime";
 import type { OpenAiEmbeddingClient } from "./embedding-provider.js";
 
@@ -96,7 +99,7 @@ async function fetchOpenAiBatchStatus(params: {
     openAi: params.openAi,
     path: `/batches/${params.batchId}`,
     errorPrefix: "openai batch status",
-    parse: async (res) => (await res.json()) as OpenAiBatchStatus,
+    parse: async (res) => readProviderJsonResponse<OpenAiBatchStatus>(res, "openai.batch-status"),
   });
 }
 
diff --git a/extensions/openai/realtime-provider-shared.test.ts b/extensions/openai/realtime-provider-shared.test.ts
@@ -0,0 +1,100 @@
+// Openai tests cover realtime session secret creation behavior.
+import { describe, expect, it, vi } from "vitest";
+import {
+  createOpenAIRealtimeClientSecret,
+  createOpenAIRealtimeTranscriptionClientSecret,
+} from "./realtime-provider-shared.js";
+
+const { fetchWithSsrFGuardMock } = vi.hoisted(() => ({
+  fetchWithSsrFGuardMock: vi.fn(),
+}));
+
+vi.mock("openclaw/plugin-sdk/ssrf-runtime", () => ({
+  fetchWithSsrFGuard: fetchWithSsrFGuardMock,
+}));
+
+function makeStreamingResponse(params: { chunkCount: number; chunkSize: number }): {
+  response: Response;
+  getReadCount: () => number;
+  wasCanceled: () => boolean;
+} {
+  let readCount = 0;
+  let canceled = false;
+  const chunk = new Uint8Array(params.chunkSize);
+  const response = new Response(
+    new ReadableStream<Uint8Array>({
+      pull(controller) {
+        if (readCount >= params.chunkCount) {
+          controller.close();
+          return;
+        }
+        readCount += 1;
+        controller.enqueue(chunk);
+      },
+      cancel() {
+        canceled = true;
+      },
+    }),
+    { status: 200, headers: { "Content-Type": "application/json" } },
+  );
+  return { response, getReadCount: () => readCount, wasCanceled: () => canceled };
+}
+
+function guardedFetch(response: Response): void {
+  fetchWithSsrFGuardMock.mockResolvedValue({ response, release: vi.fn() });
+}
+
+describe("createOpenAIRealtimeClientSecret", () => {
+  it("returns client secret from a well-formed response", async () => {
+    guardedFetch(
+      new Response(
+        JSON.stringify({
+          client_secret: { value: "eph-secret-abc" },
+          expires_at: Math.floor(Date.now() / 1000) + 60,
+        }),
+        { status: 200, headers: { "Content-Type": "application/json" } },
+      ),
+    );
+
+    const result = await createOpenAIRealtimeClientSecret({
+      authToken: "sk-test",
+      auditContext: "test",
+      session: { model: "gpt-4o-realtime-preview" },
+    });
+
+    expect(result.value).toBe("eph-secret-abc");
+    expect(typeof result.expiresAt).toBe("number");
+  });
+
+  it("bounds oversized success response and cancels the stream", async () => {
+    // 20 MiB in 1 MiB chunks — well over the 16 MiB cap
+    const streamed = makeStreamingResponse({ chunkCount: 20, chunkSize: 1024 * 1024 });
+    guardedFetch(streamed.response);
+
+    await expect(
+      createOpenAIRealtimeClientSecret({
+        authToken: "sk-test",
+        auditContext: "test",
+        session: { model: "gpt-4o-realtime-preview" },
+      }),
+    ).rejects.toThrow(/openai\.realtime-session/);
+
+    expect(streamed.wasCanceled()).toBe(true);
+    expect(streamed.getReadCount()).toBeLessThan(20);
+  });
+
+  it("throws the provider error label on oversized body", async () => {
+    const streamed = makeStreamingResponse({ chunkCount: 20, chunkSize: 1024 * 1024 });
+    guardedFetch(streamed.response);
+
+    await expect(
+      createOpenAIRealtimeTranscriptionClientSecret({
+        authToken: "sk-test",
+        auditContext: "test",
+        session: { model: "gpt-4o-transcribe" },
+      }),
+    ).rejects.toThrow(/openai\.realtime-session/);
+
+    expect(streamed.wasCanceled()).toBe(true);
+  });
+});
diff --git a/extensions/openai/realtime-provider-shared.ts b/extensions/openai/realtime-provider-shared.ts
@@ -2,6 +2,7 @@
 import { resolveExpiresAtMsFromEpochSeconds } from "openclaw/plugin-sdk/number-runtime";
 import {
   createProviderHttpError,
+  readProviderJsonResponse,
   resolveProviderRequestHeaders,
 } from "openclaw/plugin-sdk/provider-http";
 import { captureWsEvent } from "openclaw/plugin-sdk/proxy-capture";
@@ -110,7 +111,7 @@ async function createOpenAIRealtimeSecret(
       if (!response.ok) {
         throw await createProviderHttpError(response, params.errorMessage);
       }
-      return (await response.json()) as unknown;
+      return await readProviderJsonResponse<unknown>(response, "openai.realtime-session");
     } finally {
       await release();
     }
