fix(openai-responses): address review findings on the streaming pipeline and converter

Addresses the automated /review findings (doudouOUC, qwen-code-ci-bot) left
on #8169 after the AUTH_PREFLIGHT_ENV_KEYS fix.

Critical fixes in responses-pipeline.ts's streamRequest:
- baseUrl fallback used `??`, which does not catch an empty string --
  USE_OPENAI_RESPONSES has no case in ModelRegistry.getDefaultBaseUrl, so an
  unconfigured baseUrl resolves to '' and every request hit the relative
  path /v1/responses. Switched to `||`.
- streamRequest called the global `fetch` instead of the `fetch`
  buildRuntimeFetchOptions pins alongside its dispatcher. Node's built-in
  undici can be a different major version than the bundled one, and handing
  it a foreign dispatcher throws `invalid onError method` -- this broke
  every streaming request whenever a dispatcher was installed (which is the
  common case, proxy or not).
- Network errors from fetch() and non-ok HTTP responses were thrown
  unredacted, leaking proxy credentials into error messages/logs.
- No Content-Type validation: an HTTP 200 non-SSE body (a gateway/proxy
  block page, or a baseUrl pointed at a non-API endpoint) silently yielded
  zero events instead of a surfaced, retryable failure.
- The SSE parser only flushed a pending event on a blank line and never
  after the read loop ended, so a stream that closes without a trailing
  blank line (a proxy stripping trailing whitespace, or an interrupted
  connection) silently dropped its last event -- including the terminal
  response.completed event carrying usageMetadata/finishReason.

Critical fixes in index.ts's embedContent:
- Missing redactProxyError on the embedding error path (present on the
  sibling openaiContentGenerator and on anthropicContentGenerator).
- The embed client passed baseUrl to the OpenAI SDK unnormalized; this
  generator's own /v1-less baseUrl convention (the streaming pipeline
  strips/re-appends /v1 itself) 404s every embedding call against the SDK,
  which does not auto-append /v1 to a custom baseURL. Normalized the same
  way the streaming pipeline does.

Suggestions fixed in responses-converter.ts:
- `buf?.args ?? fc.arguments`: initFunctionCall seeds buf.args to '', so a
  buffer that received output_item.added but no
  function_call_arguments.delta events never fell through to the done
  item's own complete arguments. Switched to `||`.
- A user turn with text + image parts was split into two separate
  'message' items; the Responses API treats them as two distinct turns, so
  the model could fail to associate a question with its image. Parts now
  accumulate per-turn and flush as one message with a multi-part content
  array, in original relative order around any function_call/
  function_call_output/reasoning items.
- Non-image inlineData and all fileData references were silently dropped
  with no diagnostic. They now become a text placeholder noting the
  unsupported attachment (no input_file part type is wired up for this
  generator yet) instead of vanishing.
- Tool results were sent as the raw { output } / { error } JSON envelope
  instead of the unwrapped string the sibling Chat Completions converter
  sends, double-encoding and quote-escaping any tool output that is itself
  JSON.
- Removed ResponsesStreamState.reset(), an unused method contradicting the
  documented one-instance-per-stream contract.

Other:
- Extracted the embedContent text-extraction logic (byte-identical between
  this generator and the sibling openaiContentGenerator) into a shared
  extract-text-from-contents.ts helper.
- Added test coverage: multi-part message merging, attachment placeholders,
  tool-result envelope unwrapping, the buf.args empty-buffer fallback,
  sanitizePromptCacheKey's truncation/hash path, the data-only SSE framing
  branch (the shape the Responses API actually emits, previously untested),
  and the USE_OPENAI_RESPONSES factory route in createContentGenerator.
- responses-pipeline.test.ts previously stubbed the global fetch, which the
  dispatcher/fetch fix above now bypasses (the pipeline calls the pinned
  fetch instead); switched to mocking buildRuntimeFetchOptions directly so
  the mock intercepts the fetch the pipeline actually uses.

packages/core/src/core/contentGenerator.test.ts

@@ -227,6 +227,35 @@ describe('createContentGenerator', () => {
     expect(openaiMockState.createCount).toBe(1);
   });
 
+  it('routes USE_OPENAI_RESPONSES to the Responses API generator', async () => {
+    // Regression guard for the import path / auth-type check in the
+    // USE_OPENAI_RESPONSES branch: a typo there would silently break
+    // generator creation only once a real method is invoked (the returned
+    // generator is a LazyContentGenerator), so this must exercise a call,
+    // not just check the returned object's shape.
+    const mockConfig = {
+      getUsageStatisticsEnabled: () => false,
+      getContentGeneratorConfig: () => ({}),
+      getCliVersion: () => '1.0.0',
+      getTelemetryEnabled: () => false,
+      getSessionId: () => 'test-session',
+    } as unknown as Config;
+    const generator = await createContentGenerator(
+      {
+        model: 'gpt-5',
+        apiKey: 'test-key',
+        authType: AuthType.USE_OPENAI_RESPONSES,
+      },
+      mockConfig,
+    );
+
+    const result = await generator.countTokens({
+      model: 'gpt-5',
+      contents: 'hello world',
+    });
+    expect(result.totalTokens).toBeGreaterThan(0);
+  });
+
   it('does not preload non-lazy content generators', async () => {
     const generator: ContentGenerator = {
       generateContent: vi.fn(),

packages/core/src/core/openaiContentGenerator/openaiContentGenerator.ts

@@ -17,6 +17,7 @@ import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import { isAbortError } from '../../utils/errors.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 import { redactProxyError } from '../../utils/runtimeFetchOptions.js';
+import { extractTextFromContents } from '../../utils/extract-text-from-contents.js';
 
 const debugLogger = createDebugLogger('OPENAI');
 
@@ -110,37 +111,7 @@ export class OpenAIContentGenerator implements ContentGenerator {
   async embedContent(
     request: EmbedContentParameters,
   ): Promise<EmbedContentResponse> {
-    // Extract text from contents
-    let text = '';
-    if (Array.isArray(request.contents)) {
-      text = request.contents
-        .map((content) => {
-          if (typeof content === 'string') return content;
-          if ('parts' in content && content.parts) {
-            return content.parts
-              .map((part) =>
-                typeof part === 'string'
-                  ? part
-                  : 'text' in part
-                    ? (part as { text?: string }).text || ''
-                    : '',
-              )
-              .join(' ');
-          }
-          return '';
-        })
-        .join(' ');
-    } else if (request.contents) {
-      if (typeof request.contents === 'string') {
-        text = request.contents;
-      } else if ('parts' in request.contents && request.contents.parts) {
-        text = request.contents.parts
-          .map((part) =>
-            typeof part === 'string' ? part : 'text' in part ? part.text : '',
-          )
-          .join(' ');
-      }
-    }
+    const text = extractTextFromContents(request.contents);
 
     try {
       const embedding = await this.pipeline.client.embeddings.create({

packages/core/src/core/openaiResponsesContentGenerator/index.test.ts

@@ -23,8 +23,9 @@ vi.mock('./responses-pipeline.js', () => ({
 }));
 
 const mockEmbeddingsCreate = vi.fn();
+const mockOpenAIConstructor = vi.fn();
 vi.mock('openai', () => ({
-  default: vi.fn().mockImplementation(() => ({
+  default: mockOpenAIConstructor.mockImplementation(() => ({
     embeddings: { create: mockEmbeddingsCreate },
   })),
 }));
@@ -180,6 +181,59 @@ describe('OpenAIResponsesContentGenerator', () => {
         }),
       ).rejects.toThrow(/Embedding error: network down/);
     });
+
+    it('redacts proxy credentials from a thrown error message', async () => {
+      // The sibling openaiContentGenerator already redacts here; this
+      // generator's embedContent skipped it, leaking a configured proxy's
+      // credentials into the error surfaced to the caller/logs on failure.
+      mockEmbeddingsCreate.mockRejectedValue(
+        new Error('connect ECONNREFUSED http://user:[email protected]:8080'),
+      );
+      await expect(
+        generator.embedContent({
+          model: 'text-embedding-ada-002',
+          contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+        }),
+      ).rejects.toThrow(/<redacted>@proxy\.local:8080/);
+    });
+
+    it('appends /v1 to a bare-origin baseUrl before constructing the OpenAI SDK client', async () => {
+      // This generator's own baseUrl convention (used by the streaming
+      // pipeline) is /v1-less, but the OpenAI SDK does not append /v1 to a
+      // custom baseURL on its own -- only to its own built-in default.
+      // Passing the bare origin straight through 404s every embedding call.
+      mockEmbeddingsCreate.mockResolvedValue({
+        data: [{ embedding: [0.1] }],
+      });
+      const bareOriginGenerator = new OpenAIResponsesContentGenerator(
+        { ...makeGeneratorConfig(), baseUrl: 'https://api.openai.com' },
+        makeCliConfig(),
+      );
+      await bareOriginGenerator.embedContent({
+        model: 'text-embedding-ada-002',
+        contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+      });
+      expect(mockOpenAIConstructor).toHaveBeenCalledWith(
+        expect.objectContaining({ baseURL: 'https://api.openai.com/v1' }),
+      );
+    });
+
+    it('does not append a second /v1 when baseUrl already has one', async () => {
+      mockEmbeddingsCreate.mockResolvedValue({
+        data: [{ embedding: [0.1] }],
+      });
+      const generatorWithV1 = new OpenAIResponsesContentGenerator(
+        { ...makeGeneratorConfig(), baseUrl: 'https://api.openai.com/v1' },
+        makeCliConfig(),
+      );
+      await generatorWithV1.embedContent({
+        model: 'text-embedding-ada-002',
+        contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+      });
+      expect(mockOpenAIConstructor).toHaveBeenCalledWith(
+        expect.objectContaining({ baseURL: 'https://api.openai.com/v1' }),
+      );
+    });
   });
 
   it('createOpenAIResponsesContentGenerator returns an OpenAIResponsesContentGenerator', () => {

packages/core/src/core/openaiResponsesContentGenerator/index.ts

@@ -20,7 +20,11 @@ import type {
 import type OpenAI from 'openai';
 import { ResponsesPipeline } from './responses-pipeline.js';
 import { RequestTokenEstimator } from '../../utils/request-tokenizer/index.js';
-import { buildRuntimeFetchOptions } from '../../utils/runtimeFetchOptions.js';
+import {
+  buildRuntimeFetchOptions,
+  redactProxyError,
+} from '../../utils/runtimeFetchOptions.js';
+import { extractTextFromContents } from '../../utils/extract-text-from-contents.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 
 const debugLogger = createDebugLogger('RESPONSES');
@@ -52,9 +56,18 @@ export class OpenAIResponsesContentGenerator implements ContentGenerator {
       'openai',
       this.cliConfig.getProxy(),
     );
+    // This generator's own baseUrl convention is /v1-less (the streaming
+    // pipeline strips any trailing /v1 and appends /v1/responses itself),
+    // but the OpenAI SDK does not append /v1 to a custom baseURL on its
+    // own -- only to its own built-in default. Normalize the same way the
+    // streaming pipeline does so a bare-origin baseUrl still resolves to
+    // .../v1/embeddings instead of .../embeddings (404).
+    const baseURL = this.contentGeneratorConfig.baseUrl
+      ? `${this.contentGeneratorConfig.baseUrl.replace(/\/v1\/?$/, '').replace(/\/$/, '')}/v1`
+      : undefined;
     this.openaiClient = new OpenAISDK({
       apiKey,
-      baseURL: this.contentGeneratorConfig.baseUrl,
+      baseURL,
       timeout: this.contentGeneratorConfig.timeout,
       maxRetries: this.contentGeneratorConfig.maxRetries,
       ...(runtimeOptions || {}),
@@ -98,36 +111,7 @@ export class OpenAIResponsesContentGenerator implements ContentGenerator {
   async embedContent(
     request: EmbedContentParameters,
   ): Promise<EmbedContentResponse> {
-    let text = '';
-    if (Array.isArray(request.contents)) {
-      text = request.contents
-        .map((content) => {
-          if (typeof content === 'string') return content;
-          if ('parts' in content && content.parts) {
-            return content.parts
-              .map((part) =>
-                typeof part === 'string'
-                  ? part
-                  : 'text' in part
-                    ? (part as { text?: string }).text || ''
-                    : '',
-              )
-              .join(' ');
-          }
-          return '';
-        })
-        .join(' ');
-    } else if (request.contents) {
-      if (typeof request.contents === 'string') {
-        text = request.contents;
-      } else if ('parts' in request.contents && request.contents.parts) {
-        text = request.contents.parts
-          .map((part) =>
-            typeof part === 'string' ? part : 'text' in part ? part.text : '',
-          )
-          .join(' ');
-      }
-    }
+    const text = extractTextFromContents(request.contents);
 
     const client = await this.getOpenAIClient();
     try {
@@ -145,9 +129,10 @@ export class OpenAIResponsesContentGenerator implements ContentGenerator {
         embeddings: [{ values: first.embedding }],
       };
     } catch (error) {
-      debugLogger.error('Embedding error:', error);
+      const redactedError = redactProxyError(error);
+      debugLogger.error('Embedding error:', redactedError);
       throw new Error(
-        `Embedding error: ${error instanceof Error ? error.message : String(error)}`,
+        `Embedding error: ${redactedError instanceof Error ? redactedError.message : String(redactedError)}`,
       );
     }
   }

packages/core/src/core/openaiResponsesContentGenerator/responses-converter.test.ts

@@ -189,6 +189,43 @@ describe('convertResponsesEventToGemini', () => {
     ]);
   });
 
+  it("falls back to the done item's arguments when output_item.added created the buffer but no delta events ever arrived", () => {
+    // initFunctionCall seeds buf.args to ''. A `??` fallback here would
+    // never trigger for an empty string, so a proxy that sends
+    // output_item.added followed directly by output_item.done (no
+    // function_call_arguments.delta in between) must still fall through to
+    // the done item's own complete `arguments` rather than losing every
+    // argument to `JSON.parse('')` throwing.
+    const state = new ResponsesStreamState();
+    state.initFunctionCall(0, 'fc_1', 'call_1', 'read_file');
+    const resp = convertResponsesEventToGemini(
+      {
+        event: 'response.output_item.done',
+        data: {
+          output_index: 0,
+          item: {
+            type: 'function_call',
+            id: 'fc_1',
+            call_id: 'call_1',
+            name: 'read_file',
+            arguments: '{"path":"a.ts"}',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    expect(resp?.candidates?.[0]?.content?.parts).toEqual([
+      {
+        functionCall: {
+          id: 'call_1',
+          name: 'read_file',
+          args: { path: 'a.ts' },
+        },
+      },
+    ]);
+  });
+
   describe('reasoning item completion (thoughtSignature round-trip)', () => {
     it('emits a signature-only thought chunk when encrypted_content is present', () => {
       const state = new ResponsesStreamState();
@@ -558,6 +595,167 @@ describe('convertGeminiContentsToResponsesInput', () => {
     ]);
   });
 
+  it('merges a text part and an image part in the same turn into one message with a multi-part content array', () => {
+    // Regression guard: pushing one 'message' item per part would make the
+    // Responses API treat the question and the image as two separate turns.
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            { text: 'What is in this image?' },
+            { inlineData: { mimeType: 'image/png', data: 'YWJj' } },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'message',
+        role: 'user',
+        content: [
+          { type: 'input_text', text: 'What is in this image?' },
+          { type: 'input_image', image_url: 'data:image/png;base64,YWJj' },
+        ],
+      } satisfies ResponsesApiMessageItem,
+    ]);
+  });
+
+  it('flushes accumulated text as its own message before a function_call so relative order is preserved', () => {
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'model',
+          parts: [
+            { text: "I'll check that file." },
+            {
+              functionCall: {
+                id: 'call_1',
+                name: 'read_file',
+                args: { path: 'a.ts' },
+              },
+            },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'message',
+        role: 'assistant',
+        content: "I'll check that file.",
+      } satisfies ResponsesApiMessageItem,
+      {
+        type: 'function_call',
+        call_id: 'call_1',
+        name: 'read_file',
+        arguments: JSON.stringify({ path: 'a.ts' }),
+      } satisfies ResponsesApiFunctionCallItem,
+    ]);
+  });
+
+  it('replaces non-image inlineData with a text placeholder instead of silently dropping it', () => {
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            { text: 'Summarize my PDF' },
+            { inlineData: { mimeType: 'application/pdf', data: 'JVBERi0' } },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'message',
+        role: 'user',
+        content: [
+          { type: 'input_text', text: 'Summarize my PDF' },
+          {
+            type: 'input_text',
+            text: '[Unsupported inline media type: application/pdf]',
+          },
+        ],
+      } satisfies ResponsesApiMessageItem,
+    ]);
+  });
+
+  it('replaces a fileData reference with a text placeholder instead of silently dropping it', () => {
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            {
+              fileData: {
+                mimeType: 'application/pdf',
+                fileUri: 'gs://bucket/doc.pdf',
+              },
+            },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'message',
+        role: 'user',
+        content: '[Unsupported file reference: application/pdf]',
+      } satisfies ResponsesApiMessageItem,
+    ]);
+  });
+
+  it('unwraps a { output } tool response envelope to the bare string instead of double-encoding it', () => {
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            {
+              functionResponse: {
+                id: 'call_1',
+                response: { output: '{"key":"value"}' },
+              },
+            },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'function_call_output',
+        call_id: 'call_1',
+        output: '{"key":"value"}',
+      } satisfies ResponsesApiFunctionCallOutputItem,
+    ]);
+  });
+
+  it('unwraps a { error } tool response envelope to the bare string', () => {
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            {
+              functionResponse: {
+                id: 'call_1',
+                response: { error: 'file not found' },
+              },
+            },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'function_call_output',
+        call_id: 'call_1',
+        output: 'file not found',
+      } satisfies ResponsesApiFunctionCallOutputItem,
+    ]);
+  });
+
   it('extracts systemInstruction into instructions', () => {
     const { instructions } = convertGeminiContentsToResponsesInput({
       model: 'gpt-5',

packages/core/src/core/openaiResponsesContentGenerator/responses-converter.ts

@@ -92,11 +92,6 @@ export class ResponsesStreamState {
     const buf = this.funcCallArgs.get(outputIndex);
     if (buf) buf.args += delta;
   }
-
-  reset(): void {
-    this.responseId = null;
-    this.funcCallArgs.clear();
-  }
 }
 
 /**
@@ -167,7 +162,13 @@ export function convertResponsesEventToGemini(
         // non-compliant proxy doesn't silently drop the whole tool call.
         const id = buf?.id ?? fc.call_id;
         const name = buf?.name ?? fc.name;
-        const rawArgs = buf?.args ?? fc.arguments;
+        // `||` not `??`: initFunctionCall seeds buf.args to '', so a buffer
+        // that received output_item.added but no function_call_arguments.delta
+        // events (e.g. a non-compliant proxy that only sends output_item.done
+        // with the complete arguments) must still fall through to fc.arguments
+        // -- an empty string is not nullish and would otherwise short-circuit
+        // this fallback, sending an empty-args tool call to the model.
+        const rawArgs = buf?.args || fc.arguments;
         let args: Record<string, unknown> = {};
         try {
           args = JSON.parse(rawArgs) as Record<string, unknown>;
@@ -365,17 +366,34 @@ export function convertGeminiContentsToResponsesInput(
     const role = content.role === 'model' ? 'assistant' : 'user';
     const parts = content.parts ?? [];
 
+    // Text and image parts within one turn accumulate here and are flushed
+    // as a single 'message' item with a multi-part content array, rather
+    // than one 'message' item per part -- the Responses API treats separate
+    // message items as separate turns, so a {text, inlineData} pair (e.g.
+    // "What is in this image?" + the image) would otherwise become two
+    // consecutive user turns the model may not associate with each other.
+    // Flushed before any function_call/function_call_output/reasoning item
+    // so relative ordering within the turn is preserved.
+    let pendingContentParts: ResponsesApiContentPart[] = [];
+    const flushPendingMessage = () => {
+      if (pendingContentParts.length === 0) return;
+      const content: string | ResponsesApiContentPart[] =
+        pendingContentParts.length === 1 &&
+        pendingContentParts[0]!.type === 'input_text'
+          ? pendingContentParts[0]!.text
+          : pendingContentParts;
+      items.push({ type: 'message', role, content } as ResponsesApiMessageItem);
+      pendingContentParts = [];
+    };
+
     for (const part of parts) {
       if (typeof part === 'string') {
-        items.push({
-          type: 'message',
-          role,
-          content: part,
-        } as ResponsesApiMessageItem);
+        pendingContentParts.push({ type: 'input_text', text: part });
         continue;
       }
 
       if ('thought' in part && part.thought) {
+        flushPendingMessage();
         // Prior-turn reasoning can only be replayed as a real 'reasoning'
         // item if we captured its id + encrypted_content via thoughtSignature
         // when it first streamed in. If it's missing (older history, a model
@@ -410,14 +428,11 @@ export function convertGeminiContentsToResponsesInput(
       }
 
       if ('text' in part && part.text) {
-        items.push({
-          type: 'message',
-          role,
-          content: part.text,
-        } as ResponsesApiMessageItem);
+        pendingContentParts.push({ type: 'input_text', text: part.text });
       }
 
       if ('functionCall' in part && part.functionCall) {
+        flushPendingMessage();
         const callId =
           part.functionCall.id || `call_${Date.now()}_${callIdCounter++}`;
         items.push({
@@ -429,10 +444,24 @@ export function convertGeminiContentsToResponsesInput(
       }
 
       if ('functionResponse' in part && part.functionResponse) {
+        flushPendingMessage();
         const fr = part.functionResponse;
+        const responseRecord = fr.response as
+          | Record<string, unknown>
+          | null
+          | undefined;
         let output: string;
         if (typeof fr.response === 'string') {
           output = fr.response;
+        } else if (typeof responseRecord?.['output'] === 'string') {
+          // Unwrap the { output } envelope coreToolScheduler wraps tool
+          // results in (and { error } for failures), matching the sibling
+          // Chat Completions converter -- sending the raw JSON envelope
+          // instead double-encodes and quote-escapes tool output that is
+          // itself JSON, and costs extra tokens on every turn.
+          output = responseRecord['output'] as string;
+        } else if (typeof responseRecord?.['error'] === 'string') {
+          output = responseRecord['error'] as string;
         } else {
           output = JSON.stringify(fr.response ?? {});
         }
@@ -446,20 +475,32 @@ export function convertGeminiContentsToResponsesInput(
       if ('inlineData' in part && part.inlineData && role === 'user') {
         const mimeType = part.inlineData.mimeType ?? 'image/png';
         if (mimeType.startsWith('image/')) {
-          const contentParts: ResponsesApiContentPart[] = [
-            {
-              type: 'input_image',
-              image_url: `data:${mimeType};base64,${part.inlineData.data}`,
-            },
-          ];
-          items.push({
-            type: 'message',
-            role: 'user',
-            content: contentParts,
-          } as ResponsesApiMessageItem);
+          pendingContentParts.push({
+            type: 'input_image',
+            image_url: `data:${mimeType};base64,${part.inlineData.data}`,
+          });
+        } else {
+          // No input_file part type is wired up for this generator yet
+          // (unlike the sibling Chat Completions converter, which sends
+          // PDFs as a file part) -- rather than silently dropping the
+          // attachment with zero diagnostic, tell the model it was there
+          // but isn't supported so a "summarize my PDF" prompt gets an
+          // explanation instead of "I don't see a document".
+          pendingContentParts.push({
+            type: 'input_text',
+            text: `[Unsupported inline media type: ${mimeType}]`,
+          });
         }
       }
+
+      if ('fileData' in part && part.fileData && role === 'user') {
+        pendingContentParts.push({
+          type: 'input_text',
+          text: `[Unsupported file reference: ${part.fileData.mimeType ?? 'unknown mime type'}]`,
+        });
+      }
     }
+    flushPendingMessage();
   }
 
   return { instructions, input: items };

packages/core/src/core/openaiResponsesContentGenerator/responses-pipeline.test.ts

@@ -21,6 +21,22 @@ import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import type { ResponsesApiRequest } from './types.js';
 import { preloadRuntimeFetchModule } from '../../utils/runtimeFetchOptions.js';
 
+// The pipeline calls the `fetch` buildRuntimeFetchOptions returns (pinned
+// alongside its dispatcher) rather than the global `fetch`, so the mock must
+// intercept it there -- stubbing global `fetch` alone is bypassed and the
+// real undici fetch attempts a live network call (see #8169 review).
+const { buildRuntimeFetchOptionsMock } = vi.hoisted(() => ({
+  buildRuntimeFetchOptionsMock: vi.fn(),
+}));
+vi.mock('../../utils/runtimeFetchOptions.js', async (importOriginal) => {
+  const actual =
+    await importOriginal<typeof import('../../utils/runtimeFetchOptions.js')>();
+  return {
+    ...actual,
+    buildRuntimeFetchOptions: buildRuntimeFetchOptionsMock,
+  };
+});
+
 function sseStream(lines: string[]): ReadableStream<Uint8Array> {
   const body = lines.join('\n') + '\n';
   const encoder = new TextEncoder();
@@ -86,17 +102,18 @@ describe('ResponsesPipeline', () => {
 
   beforeEach(() => {
     fetchMock = vi.fn();
-    vi.stubGlobal('fetch', fetchMock);
+    buildRuntimeFetchOptionsMock.mockReturnValue({ fetch: fetchMock });
   });
 
   afterEach(() => {
-    vi.unstubAllGlobals();
+    vi.clearAllMocks();
   });
 
   function mockResponse(lines: string[], status = 200) {
     fetchMock.mockResolvedValue({
       ok: status < 400,
       status,
+      headers: { get: () => 'text/event-stream' },
       body: sseStream(lines),
       text: async () => '',
     });
@@ -143,6 +160,49 @@ describe('ResponsesPipeline', () => {
     ]);
   });
 
+  it('passes userPromptId through as prompt_cache_key when within the length limit', async () => {
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    for await (const _ of pipeline.executeStream(
+      textRequest('hi'),
+      'short-id',
+    )) {
+      // drain
+    }
+    const body = JSON.parse(
+      fetchMock.mock.calls[0]![1].body,
+    ) as ResponsesApiRequest;
+    expect(body.prompt_cache_key).toBe('short-id');
+  });
+
+  it('hashes userPromptId into a fixed-length prompt_cache_key when it exceeds 64 characters', async () => {
+    // A mutation removing the truncation branch would silently disable
+    // prompt caching (the Responses API rejects/ignores an over-length
+    // prompt_cache_key) with no visible error -- pin the hashed shape.
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const longId = 'x'.repeat(100);
+    for await (const _ of pipeline.executeStream(textRequest('hi'), longId)) {
+      // drain
+    }
+    const body = JSON.parse(
+      fetchMock.mock.calls[0]![1].body,
+    ) as ResponsesApiRequest;
+    expect(body.prompt_cache_key).not.toBe(longId);
+    expect(body.prompt_cache_key).toHaveLength(64);
+    expect(body.prompt_cache_key).toMatch(/^[0-9a-f]{64}$/);
+  });
+
   it('strips a trailing /v1 from baseUrl before appending /v1/responses', async () => {
     mockResponse(
       sseEvent('response.completed', { response: { status: 'completed' } }),
@@ -440,6 +500,70 @@ describe('ResponsesPipeline', () => {
     expect(result.usageMetadata?.totalTokenCount).toBe(5);
   });
 
+  it('parses data-only SSE frames (no event: line) -- the shape the Responses API actually emits', async () => {
+    // Every other test builds frames with sseEvent(), which always prepends
+    // an `event: ` line, so it never exercises the data-only branch that
+    // parses `data['type']` directly and handles `[DONE]`. A regression
+    // there would leave the whole suite green while breaking every real
+    // OpenAI call.
+    const dataOnlyLines = [
+      `data: ${JSON.stringify({ type: 'response.output_text.delta', delta: 'hi' })}`,
+      `data: ${JSON.stringify({
+        type: 'response.completed',
+        response: { id: 'r1', status: 'completed' },
+      })}`,
+      'data: [DONE]',
+    ];
+    mockResponse(dataOnlyLines);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(
+      textRequest('hello'),
+      'prompt-1',
+    )) {
+      chunks.push(chunk);
+    }
+    expect(chunks.map((c) => c.candidates?.[0]?.content?.parts)).toEqual([
+      [{ text: 'hi' }],
+      [],
+    ]);
+  });
+
+  it('parses data-only SSE frames split across multiple reader.read() chunks', async () => {
+    const dataOnlyLines = [
+      `data: ${JSON.stringify({ type: 'response.output_text.delta', delta: 'hi' })}`,
+      `data: ${JSON.stringify({
+        type: 'response.completed',
+        response: { id: 'r1', status: 'completed' },
+      })}`,
+    ];
+    fetchMock.mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: { get: () => 'text/event-stream' },
+      body: sseStreamChunked(dataOnlyLines, 5),
+      text: async () => '',
+    });
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(
+      textRequest('hello'),
+      'prompt-1',
+    )) {
+      chunks.push(chunk);
+    }
+    expect(chunks.map((c) => c.candidates?.[0]?.content?.parts)).toEqual([
+      [{ text: 'hi' }],
+      [],
+    ]);
+  });
+
   it('parses correctly when frames are split across multiple reader.read() chunks', async () => {
     const lines = [
       ...sseEvent('response.output_text.delta', { delta: 'foo' }),
@@ -466,6 +590,7 @@ describe('ResponsesPipeline', () => {
     fetchMock.mockResolvedValue({
       ok: true,
       status: 200,
+      headers: { get: () => 'text/event-stream' },
       body: sseStreamChunked(lines, 5),
       text: async () => '',
     });

packages/core/src/core/openaiResponsesContentGenerator/responses-pipeline.ts

@@ -21,7 +21,10 @@ import {
   convertGeminiToolsToResponsesTools,
   cleanOrphanedFunctionCalls,
 } from './responses-converter.js';
-import { buildRuntimeFetchOptions } from '../../utils/runtimeFetchOptions.js';
+import {
+  buildRuntimeFetchOptions,
+  redactProxyError,
+} from '../../utils/runtimeFetchOptions.js';
 import { createHash } from 'node:crypto';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 
@@ -167,7 +170,7 @@ export class ResponsesPipeline {
     streamState: ResponsesStreamState,
     signal?: AbortSignal,
   ): AsyncGenerator<GenerateContentResponse> {
-    const baseUrl = (this.config.baseUrl ?? 'https://api.openai.com')
+    const baseUrl = (this.config.baseUrl || 'https://api.openai.com')
       .replace(/\/v1\/?$/, '')
       .replace(/\/$/, '');
     const url = `${baseUrl}/v1/responses`;
@@ -207,8 +210,20 @@ export class ResponsesPipeline {
     if (runtimeOptions?.fetchOptions?.dispatcher) {
       fetchOpts.dispatcher = runtimeOptions.fetchOptions.dispatcher;
     }
+    // Use the fetch buildRuntimeFetchOptions pins alongside the dispatcher
+    // (same undici version as the dispatcher) rather than Node's global
+    // fetch -- Node's built-in undici can be a different major version than
+    // the bundled one, and handing it a foreign dispatcher throws `invalid
+    // onError method`.
+    const fetchFn =
+      (runtimeOptions as { fetch?: typeof fetch } | undefined)?.fetch ?? fetch;
 
-    const response = await fetch(url, fetchOpts as RequestInit);
+    let response: Response;
+    try {
+      response = await fetchFn(url, fetchOpts as RequestInit);
+    } catch (error) {
+      throw redactProxyError(error);
+    }
 
     if (!response.ok) {
       const errBody = await response.text().catch(() => '');
@@ -217,13 +232,60 @@ export class ResponsesPipeline {
       );
       (err as ResponsesApiError).status = response.status;
       (err as ResponsesApiError).responseBody = errBody;
-      throw err;
+      throw redactProxyError(err);
     }
 
     if (!response.body) {
       throw new Error('Responses API returned no body');
     }
 
+    const contentType = response.headers.get('content-type') ?? '';
+    if (
+      contentType &&
+      !/text\/event-stream|application\/x-ndjson|application\/stream\+json/.test(
+        contentType,
+      )
+    ) {
+      throw new Error(
+        `Responses API returned non-SSE content-type: ${contentType}`,
+      );
+    }
+
+    const convertParsedFrame = (
+      eventType: ResponsesSSEEventType,
+      data: Record<string, unknown>,
+    ): GenerateContentResponse | null => {
+      const sseEvent: ResponsesSSEEvent = { event: eventType, data };
+      return convertResponsesEventToGemini(
+        sseEvent,
+        this.config.model,
+        streamState,
+      );
+    };
+
+    // Shared by both SSE framing styles this method has to accept: an
+    // `event: ` line followed by a blank-line-terminated `data: ` block
+    // (the standard-compliant shape), and a data-only `data: {"type":...}`
+    // line with no `event: ` line (the shape the OpenAI Responses API
+    // actually emits on the wire).
+    const parseSSEFrame = (
+      eventType: ResponsesSSEEventType,
+      dataStr: string,
+    ): GenerateContentResponse | null => {
+      try {
+        const data = JSON.parse(dataStr) as Record<string, unknown>;
+        return convertParsedFrame(eventType, data);
+      } catch (err) {
+        if (err instanceof SyntaxError) {
+          debugLogger.debug(
+            `Failed to parse SSE data: ${dataStr.substring(0, 200)}`,
+          );
+          return null;
+        }
+        throw redactProxyError(err);
+      }
+    };
+
     const reader = response.body.getReader();
     const decoder = new TextDecoder();
     let buffer = '';
@@ -259,15 +321,7 @@ export class ResponsesPipeline {
                   | ResponsesSSEEventType
                   | undefined;
                 if (eventType) {
-                  const sseEvent: ResponsesSSEEvent = {
-                    event: eventType,
-                    data,
-                  };
-                  const geminiResp = convertResponsesEventToGemini(
-                    sseEvent,
-                    this.config.model,
-                    streamState,
-                  );
+                  const geminiResp = convertParsedFrame(eventType, data);
                   if (geminiResp) {
                     yield geminiResp;
                   }
@@ -278,7 +332,7 @@ export class ResponsesPipeline {
                     `Failed to parse SSE data: ${dataContent.substring(0, 200)}`,
                   );
                 } else {
-                  throw err;
+                  throw redactProxyError(err);
                 }
               }
             }
@@ -286,31 +340,9 @@ export class ResponsesPipeline {
           }
 
           if (line.trim() === '' && currentEventType && dataAccumulator) {
-            try {
-              const data = JSON.parse(dataAccumulator) as Record<
-                string,
-                unknown
-              >;
-              const sseEvent: ResponsesSSEEvent = {
-                event: currentEventType,
-                data,
-              };
-              const geminiResp = convertResponsesEventToGemini(
-                sseEvent,
-                this.config.model,
-                streamState,
-              );
-              if (geminiResp) {
-                yield geminiResp;
-              }
-            } catch (err) {
-              if (err instanceof SyntaxError) {
-                debugLogger.debug(
-                  `Failed to parse SSE data: ${dataAccumulator.substring(0, 200)}`,
-                );
-              } else {
-                throw err;
-              }
+            const geminiResp = parseSSEFrame(currentEventType, dataAccumulator);
+            if (geminiResp) {
+              yield geminiResp;
             }
             currentEventType = null;
             dataAccumulator = '';
@@ -320,6 +352,17 @@ export class ResponsesPipeline {
           }
         }
       }
+
+      // The stream can close without a trailing blank line (a proxy that
+      // strips trailing whitespace, or an interrupted connection) -- flush
+      // any event still buffered so the final `response.completed` event
+      // (carrying usageMetadata/finishReason) isn't silently dropped.
+      if (currentEventType && dataAccumulator) {
+        const geminiResp = parseSSEFrame(currentEventType, dataAccumulator);
+        if (geminiResp) {
+          yield geminiResp;
+        }
+      }
     } finally {
       await reader.cancel().catch(() => {});
       reader.releaseLock();

packages/core/src/utils/extract-text-from-contents.ts

@@ -0,0 +1,53 @@
+/**
+ * @license
+ * Copyright 2025 Qwen Team
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+import type { ContentListUnion } from '@google/genai';
+
+/**
+ * Extract the concatenated text of every text part across a `@google/genai`
+ * `ContentListUnion`, used by the OpenAI-compatible generators' `embedContent`
+ * to build the embedding input string. Shared so a change to how `Part` text
+ * is shaped only needs to be made once -- this was previously duplicated
+ * verbatim between `openaiContentGenerator.ts` and
+ * `openaiResponsesContentGenerator/index.ts`.
+ */
+export function extractTextFromContents(contents: ContentListUnion): string {
+  let text = '';
+  if (Array.isArray(contents)) {
+    text = contents
+      .map((content) => {
+        if (typeof content === 'string') return content;
+        if ('parts' in content && content.parts) {
+          return content.parts
+            .map((part) =>
+              typeof part === 'string'
+                ? part
+                : 'text' in part
+                  ? (part as { text?: string }).text || ''
+                  : '',
+            )
+            .join(' ');
+        }
+        return '';
+      })
+      .join(' ');
+  } else if (contents) {
+    if (typeof contents === 'string') {
+      text = contents;
+    } else if ('parts' in contents && contents.parts) {
+      text = contents.parts
+        .map((part) =>
+          typeof part === 'string'
+            ? part
+            : 'text' in part
+              ? (part as { text?: string }).text || ''
+              : '',
+        )
+        .join(' ');
+    }
+  }
+  return text;
+}