fix(openai): stop double-prefixing SSE bodies mislabeled as JSON (#96503)

* fix(openai): stop double-prefixing SSE bodies mislabeled as JSON

OpenAI-compatible gateways that stream real SSE (data: {...}) but label the
response application/json hit the streaming JSON-wrap fallback, which re-prefixed
each frame as 'data: data: {...}' and broke JSON.parse in the OpenAI SDK. Sniff
JSON-labeled streaming bodies and relabel genuine SSE as text/event-stream so the
SSE sanitizer parses them verbatim.

* test(openai): prove mislabeled SSE stays streaming

---------

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

src/agents/provider-transport-fetch.test.ts

@@ -1193,6 +1193,58 @@ describe("buildGuardedModelFetch", () => {
     expect(items).toEqual([{ ok: true }]);
   });
 
+  it("does not re-prefix SSE bodies mislabeled as JSON by streaming gateways", async () => {
+    const source = openResponseStreamText(
+      'data: {"id":"a","choices":[{"index":0,"delta":{"content":"Hi","role":"assistant"}}]}\n\n' +
+        'data: {"id":"a","choices":[{"index":0,"delta":{"content":" there"}}]}\n\n' +
+        "data: [DONE]\n\n",
+    );
+    fetchWithSsrFGuardMock.mockResolvedValue({
+      response: new Response(
+        source.stream,
+        // Mislabeled: SSE body served with a JSON content-type.
+        { headers: { "content-type": "application/json; charset=utf-8" } },
+      ),
+      finalUrl: "https://gateway.example/v1/chat/completions",
+      release: vi.fn(async () => undefined),
+    });
+    const model = {
+      id: "MiniMax-M3",
+      provider: "hetu",
+      api: "openai-completions",
+      baseUrl: "https://gateway.example/v1",
+    } as unknown as Model<"openai-completions">;
+
+    const responsePromise = buildGuardedModelFetch(model)(
+      "https://gateway.example/v1/chat/completions",
+      {
+        method: "POST",
+        headers: { "content-type": "application/json" },
+        body: JSON.stringify({ model: "MiniMax-M3", stream: true }),
+      },
+    );
+    const timeout = Symbol("timeout");
+    const result = await Promise.race<Response | typeof timeout>([
+      responsePromise,
+      new Promise<typeof timeout>((resolve) => {
+        setTimeout(() => resolve(timeout), 100);
+      }),
+    ]);
+    source.close();
+
+    expect(result).not.toBe(timeout);
+    const response = result as Response;
+    expect(response.headers.get("content-type")).toContain("text/event-stream");
+    const items = [];
+    for await (const item of Stream.fromSSEResponse(response, new AbortController())) {
+      items.push(item);
+    }
+    expect(items).toEqual([
+      { id: "a", choices: [{ index: 0, delta: { content: "Hi", role: "assistant" } }] },
+      { id: "a", choices: [{ index: 0, delta: { content: " there" } }] },
+    ]);
+  });
+
   it("does not clone Request bodies while checking for streaming JSON fallbacks", async () => {
     const cloneSpy = vi.spyOn(Request.prototype, "clone");
     fetchWithSsrFGuardMock.mockResolvedValue({

src/agents/provider-transport-fetch.ts

@@ -293,10 +293,6 @@ function isJsonContentType(contentType: string): boolean {
   return /\bapplication\/json\b/i.test(contentType) || /\+json\b/i.test(contentType);
 }
 
-function isOpenAISdkStreamContentType(contentType: string): boolean {
-  return /\btext\/event-stream\b/i.test(contentType) || isJsonContentType(contentType);
-}
-
 type OpenAISdkStreamBodyKind = "html" | "json" | "sse" | "unknown";
 
 function classifyOpenAISdkStreamBodyPrefix(text: string): OpenAISdkStreamBodyKind {
@@ -371,7 +367,20 @@ async function normalizeOpenAISdkStreamContentType(params: {
   localServiceLease?: ProviderLocalServiceLease;
 }): Promise<Response> {
   const contentType = params.response.headers.get("content-type") ?? "";
-  if (!params.response.ok || !params.response.body || isOpenAISdkStreamContentType(contentType)) {
+  if (!params.response.ok || !params.response.body) {
+    return params.response;
+  }
+  if (/\btext\/event-stream\b/i.test(contentType)) {
+    return params.response;
+  }
+  if (isJsonContentType(contentType)) {
+    // Some OpenAI-compatible gateways stream real SSE (`data: {...}`) but mislabel
+    // the response as JSON. Without relabeling, the JSON-wrap fallback below would
+    // re-prefix each frame as `data: data: {...}`, breaking JSON.parse in the SDK.
+    const kind = await classifyOpenAISdkStreamBody(params.response).catch(() => "unknown" as const);
+    if (kind === "sse") {
+      return withOpenAISdkStreamContentType(params.response, "text/event-stream; charset=utf-8");
+    }
     return params.response;
   }
   if (!contentType.trim()) {