fix(core): address review round on the Responses wire (7 criticals + coverage)

Addresses the review round on #8169. Every finding was re-verified against
the code before acting; all were confirmed. Each behavior fix has a
regression test that was checked red-before / green-after in place.

Retry, timeout and error classification:
- generateContentStream returned a lazy async generator, so its promise
  resolved before any network I/O and connection-time 5xx/429 could never
  reach retryWithBackoff. Split the pipeline into an eager connect phase
  (fetch + non-2xx handling + content-type guard) and a lazy body-iteration
  phase behind connectStream(), and wired index.ts to it. executeStream is
  retained and now delegates, so existing callers are unaffected.
- The stream read loop had no timeout layer at all and the undici dispatcher
  uses headersTimeout:0/bodyTimeout:0, so a 200-then-silent upstream hung
  the turn forever with no user present to cancel. Added an idle watchdog
  honoring streamIdleTimeoutMs / QWEN_STREAM_IDLE_TIMEOUT_MS, reusing the
  Chat pipeline's shared constants and its retryable ETIMEDOUT semantics.
  (Note: the reviewer's cited helper `withStreamInactivityTimeout` does not
  exist; the real sibling is `withStreamGuards`, whose structure is mirrored
  here rather than imported, since importing it pulls in the whole OpenAI
  SDK graph.)
- Mid-stream `error` / `response.failed` frames arrive after a 200 OK and
  were thrown as plain Errors with no `.status` and no `.code`, so they
  classified as `unknown` and missed the rate-limit retry, transport replay,
  model fallback and persistent-mode wait. Map the known codes onto a status
  before throwing.
- getOpenAIClient passed raw timeout/maxRetries to the SDK, bypassing
  resolveRequestTimeout (so `timeout: 0`, a documented "disable" knob, meant
  immediate abort) and DEFAULT_MAX_RETRIES.

Tool-call argument handling:
- A corrupt non-empty delta buffer shadowed the done item's authoritative
  `arguments`, dispatching tool calls with `{}`. Fall back to the done
  item's string on parse failure.
- Malformed-but-repairable arguments bypassed the safeJsonParse/jsonrepair
  fallback every sibling wire applies to the same input class.
- Arguments parsing to an array/null/primitive flowed through the
  `as Record` cast into functionCall.args; collapse non-objects to `{}` as
  streamingToolCallParser does.

Fidelity and observability:
- max_output_tokens: buildRequest never read request.config.maxOutputTokens,
  dropping the per-send window clamp geminiChat computes (#5950).
- functionResponse.parts media was silently discarded; tool-returned images
  now reach the wire as a follow-up input_image message.
- The converter never called setGenAiUsageProvenance while always
  materializing cachedContentTokenCount, so every turn reported cache
  statistics as provider-reported.

Reachability and diagnostics:
- AuthType.USE_OPENAI_RESPONSES was missing from the --auth-type yargs
  choices, so `qwen --auth-type=openai-responses` was rejected outright.
- The new PROTOCOL_ITEMS entry was unreachable UI. Added the auth type to
  customProvider.protocolOptions; verified the type is wired end to end
  (contentGenerator creation, AUTH_PREFLIGHT_ENV_KEYS, validateAuthMethod,
  DEFAULT_BASE_URLS) so the wizard option resolves to a working generator.
- systemInfo.ts and the RUM base_url gate were not extended for the new auth
  type, degrading /bug and /about output and losing endpoint attribution.
- De-duplicated the part->text lambda in extract-text-from-contents.

Coverage: filled the 29 flagged untested-behavior sites (multi-line data:
frames, UTF-8 split across chunk boundaries, extra_body precedence with an
explicit 0, custom header forwarding, abort-signal identity and mid-stream
abort, dispatcher forwarding, proxy-credential redaction, the non-SSE
content-type guard, malformed frames, empty-merge, and the auth/telemetry
entries).

Verification: 111/111 in openaiResponsesContentGenerator, 54/54 in the other
touched core suites, 402/404 in the touched CLI suites (the 2 failures are
pre-existing and reproduce on a pristine checkout -- config.test.ts reads
the ambient OTEL_EXPORTER_OTLP_ENDPOINT). tsc clean, eslint clean, prettier
clean.

packages/cli/src/acp-integration/authPreflight.test.ts

@@ -49,6 +49,12 @@ describe('AUTH_PREFLIGHT_ENV_KEYS drift detection', () => {
     expect(uncovered).toEqual([]);
   });
 
+  it('checks OPENAI_API_KEY for OpenAI Responses auth', () => {
+    expect(AUTH_PREFLIGHT_ENV_KEYS[AuthType.USE_OPENAI_RESPONSES]).toContain(
+      'OPENAI_API_KEY',
+    );
+  });
+
   it('does not list waived auth types as keyed', () => {
     // A misconfiguration where a type is in BOTH maps would mean ambiguity.
     const keyed = Object.keys(AUTH_PREFLIGHT_ENV_KEYS);

packages/cli/src/config/auth.test.ts

@@ -43,6 +43,18 @@ describe('validateAuthMethod', () => {
     expect(validateAuthMethod(AuthType.USE_OPENAI)).toBeNull();
   });
 
+  it('validates USE_OPENAI_RESPONSES with the default OpenAI API key', () => {
+    process.env['OPENAI_API_KEY'] = 'fake-key';
+
+    expect(validateAuthMethod(AuthType.USE_OPENAI_RESPONSES)).toBeNull();
+  });
+
+  it('reports the OpenAI API key requirement for USE_OPENAI_RESPONSES', () => {
+    expect(validateAuthMethod(AuthType.USE_OPENAI_RESPONSES)).toBe(
+      "Missing API key for OpenAI-compatible auth. Set settings.security.auth.apiKey, or set the 'OPENAI_API_KEY' environment variable.",
+    );
+  });
+
   it('should return an error message for USE_OPENAI if no API key is available', () => {
     expect(validateAuthMethod(AuthType.USE_OPENAI)).toBe(
       "Missing API key for OpenAI-compatible auth. Set settings.security.auth.apiKey, or set the 'OPENAI_API_KEY' environment variable.",

packages/cli/src/config/config.test.ts

@@ -12,6 +12,7 @@ import {
   DEFAULT_QWEN_MODEL,
   OutputFormat,
   NativeLspService,
+  AuthType,
   Storage,
 } from '@qwen-code/qwen-code-core';
 import {
@@ -319,6 +320,19 @@ describe('parseArguments', () => {
     expect(argv.promptInteractive).toBeUndefined();
   });
 
+  it('accepts OpenAI Responses as an auth type', async () => {
+    process.argv = [
+      'node',
+      'script.js',
+      '--auth-type',
+      AuthType.USE_OPENAI_RESPONSES,
+    ];
+
+    const argv = await parseArguments();
+
+    expect(argv.authType).toBe(AuthType.USE_OPENAI_RESPONSES);
+  });
+
   it('registers update as an exiting subcommand', async () => {
     process.argv = ['node', 'script.js', 'update'];
     mockUpdateHandler.mockResolvedValue(undefined);

packages/cli/src/config/config.ts

@@ -959,6 +959,7 @@ export async function parseArguments(): Promise<CliArgs> {
           type: 'string',
           choices: [
             AuthType.USE_OPENAI,
+            AuthType.USE_OPENAI_RESPONSES,
             AuthType.USE_ANTHROPIC,
             AuthType.QWEN_OAUTH,
             AuthType.USE_GEMINI,

packages/cli/src/ui/auth/ProviderSetupSteps.test.tsx

@@ -7,7 +7,7 @@
 import { renderWithProviders } from '../../test-utils/render.js';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import { act } from 'react';
-import { AuthType } from '@qwen-code/qwen-code-core';
+import { AuthType, customProvider } from '@qwen-code/qwen-code-core';
 import type { KeypressHandler, Key } from '../contexts/KeypressContext.js';
 import { useKeypress } from '../hooks/useKeypress.js';
 import { ProviderSetupSteps } from './ProviderSetupSteps.js';
@@ -73,6 +73,54 @@ describe('ProviderSetupSteps', () => {
     });
   };
 
+  const createProtocolFlow = (): ProviderSetupFlow => {
+    const noop = vi.fn();
+    return {
+      state: {
+        provider: customProvider,
+        step: 'protocol',
+        stepIndex: 0,
+        totalSteps: 1,
+        protocol: AuthType.USE_OPENAI,
+        baseUrl: '',
+        baseUrlPlaceholder: '',
+        baseUrlOptionIndex: 0,
+        baseUrlError: null,
+        apiKey: '',
+        apiKeyError: null,
+        modelIds: '',
+        modelIdsError: null,
+        thinkingEnabled: false,
+        modalityEnabled: false,
+        modalityImage: true,
+        modalityVideo: true,
+        modalityAudio: true,
+        modalityPdf: false,
+        contextWindowSize: '',
+        focusedConfigIndex: 0,
+        previewJson: '',
+      },
+      start: noop,
+      reset: noop,
+      goBack: noop,
+      selectProtocol: noop,
+      selectBaseUrl: noop,
+      highlightBaseUrl: noop,
+      submitBaseUrl: noop,
+      changeBaseUrl: noop,
+      changeApiKey: noop,
+      submitApiKey: noop,
+      changeModelIds: noop,
+      submitModelIds: noop,
+      moveAdvancedFocusUp: noop,
+      moveAdvancedFocusDown: noop,
+      toggleFocusedAdvancedOption: noop,
+      changeContextWindowSize: noop,
+      submitAdvancedConfig: noop,
+      submit: noop,
+    } as unknown as ProviderSetupFlow;
+  };
+
   const createAdvancedConfigFlow = (): ProviderSetupFlow => {
     const noop = vi.fn();
     return {
@@ -259,6 +307,17 @@ describe('ProviderSetupSteps', () => {
     } as unknown as ProviderSetupFlow;
   };
 
+  it('renders OpenAI Responses for the custom provider protocol step', () => {
+    const flow = createProtocolFlow();
+
+    const { lastFrame, unmount } = renderWithProviders(
+      <ProviderSetupSteps flow={flow} />,
+    );
+
+    expect(lastFrame()).toContain('OpenAI Responses');
+    unmount();
+  });
+
   it('maps Ctrl+P/N to advanced-config focus navigation', () => {
     const flow = createAdvancedConfigFlow();
 

packages/cli/src/ui/auth/useAuth.test.ts

@@ -74,6 +74,19 @@ const createConfig = () => {
 describe('useAuthCommand', () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    vi.unstubAllEnvs();
+  });
+
+  it('accepts OpenAI Responses as QWEN_DEFAULT_AUTH_TYPE', () => {
+    vi.stubEnv('QWEN_DEFAULT_AUTH_TYPE', AuthType.USE_OPENAI_RESPONSES);
+    const settings = createSettings();
+    const config = createConfig();
+
+    const { result } = renderHook(() =>
+      useAuthCommand(settings as never, config as never, vi.fn()),
+    );
+
+    expect(result.current.authError).toBeNull();
   });
 
   it('exposes closeAuthDialog that flips isAuthDialogOpen to false', () => {

packages/cli/src/utils/systemInfo.test.ts

@@ -282,6 +282,30 @@ describe('systemInfo', () => {
       expect(extendedInfo.baseUrl).toBe('https://api.openai.com');
     });
 
+    it('should include provider diagnostics for OpenAI Responses auth', async () => {
+      vi.mocked(IdeClient.getInstance).mockResolvedValue({
+        getDetectedIdeDisplayName: vi.fn().mockReturnValue('test-ide'),
+      } as unknown as IdeClient);
+      setExecFileStdout('10.0.0');
+
+      const { AuthType } = await import('@qwen-code/qwen-code-core');
+      vi.mocked(mockContext.services.config!.getAuthType).mockReturnValue(
+        AuthType.USE_OPENAI_RESPONSES,
+      );
+      vi.mocked(
+        mockContext.services.config!.getContentGeneratorConfig,
+      ).mockReturnValue({
+        model: 'gpt-5',
+        baseUrl: 'https://api.example.com',
+        apiKeyEnvKey: 'OPENAI_API_KEY',
+      });
+
+      const extendedInfo = await getExtendedSystemInfo(mockContext);
+
+      expect(extendedInfo.baseUrl).toBe('https://api.example.com');
+      expect(extendedInfo.apiKeyEnvKey).toBe('OPENAI_API_KEY');
+    });
+
     it('should use sandbox env without prefix for bug reports', async () => {
       process.env['SANDBOX'] = 'qwen-code-test-sandbox';
       vi.mocked(IdeClient.getInstance).mockResolvedValue({

packages/cli/src/utils/systemInfo.ts

@@ -215,6 +215,7 @@ export async function getExtendedSystemInfo(
   // Get base URL and apiKeyEnvKey if using OpenAI or Anthropic auth
   const contentGeneratorConfig =
     baseInfo.selectedAuthType === AuthType.USE_OPENAI ||
+    baseInfo.selectedAuthType === AuthType.USE_OPENAI_RESPONSES ||
     baseInfo.selectedAuthType === AuthType.USE_ANTHROPIC
       ? context.services.config?.getContentGeneratorConfig()
       : undefined;

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

@@ -15,10 +15,12 @@ vi.mock('../../utils/request-tokenizer/index.js', () => ({
 
 const mockExecute = vi.fn();
 const mockExecuteStream = vi.fn();
+const mockConnectStream = vi.fn();
 vi.mock('./responses-pipeline.js', () => ({
   ResponsesPipeline: vi.fn().mockImplementation(() => ({
     execute: mockExecute,
     executeStream: mockExecuteStream,
+    connectStream: mockConnectStream,
   })),
 }));
 
@@ -38,6 +40,11 @@ import type { Config } from '../../config/config.js';
 import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import { GenerateContentResponse } from '@google/genai';
 import { preloadRuntimeFetchModule } from '../../utils/runtimeFetchOptions.js';
+import {
+  DEFAULT_MAX_RETRIES,
+  DEFAULT_TIMEOUT,
+  DISABLED_REQUEST_TIMEOUT_MS,
+} from '../openaiContentGenerator/constants.js';
 
 function makeCliConfig(): Config {
   return { getProxy: () => undefined } as unknown as Config;
@@ -87,18 +94,33 @@ describe('OpenAIResponsesContentGenerator', () => {
     async function* fakeStream() {
       yield new GenerateContentResponse();
     }
-    mockExecuteStream.mockResolvedValue(fakeStream());
-    await generator.generateContentStream(
+    const stream = fakeStream();
+    mockConnectStream.mockResolvedValue(stream);
+    const result = await generator.generateContentStream(
       { model: 'gpt-5', contents: [] },
       'prompt-1',
     );
-    expect(mockExecuteStream).toHaveBeenCalledWith(
+    expect(result).toBe(stream);
+    expect(mockConnectStream).toHaveBeenCalledWith(
       { model: 'gpt-5', contents: [] },
       'prompt-1',
       undefined,
     );
   });
 
+  it('rejects generateContentStream when the initial connection fails', async () => {
+    mockConnectStream.mockRejectedValue(
+      Object.assign(new Error('Responses API error 500'), { status: 500 }),
+    );
+
+    await expect(
+      generator.generateContentStream(
+        { model: 'gpt-5', contents: [] },
+        'prompt-1',
+      ),
+    ).rejects.toThrow('Responses API error 500');
+  });
+
   it('forwards a real abortSignal from request.config to the pipeline', async () => {
     const expected = new GenerateContentResponse();
     mockExecute.mockResolvedValue(expected);
@@ -118,13 +140,13 @@ describe('OpenAIResponsesContentGenerator', () => {
     async function* fakeStream() {
       yield new GenerateContentResponse();
     }
-    mockExecuteStream.mockResolvedValue(fakeStream());
+    mockConnectStream.mockResolvedValue(fakeStream());
     const { signal } = new AbortController();
     await generator.generateContentStream(
       { model: 'gpt-5', contents: [], config: { abortSignal: signal } },
       'prompt-1',
     );
-    expect(mockExecuteStream).toHaveBeenCalledWith(
+    expect(mockConnectStream).toHaveBeenCalledWith(
       expect.anything(),
       'prompt-1',
       signal,
@@ -185,7 +207,7 @@ describe('OpenAIResponsesContentGenerator', () => {
         makeCliConfig(),
       );
       await embedModelGenerator.embedContent({
-        model: 'text-embedding-3-small',
+        model: 'request-model-that-does-not-embed',
         contents: [{ role: 'user', parts: [{ text: 'hello world' }] }],
       });
       expect(mockEmbeddingsCreate).toHaveBeenCalledWith(
@@ -206,6 +228,19 @@ describe('OpenAIResponsesContentGenerator', () => {
       );
     });
 
+    it('joins an array of string contents for embedding', async () => {
+      mockEmbeddingsCreate.mockResolvedValue({
+        data: [{ embedding: [0.3] }],
+      });
+      await generator.embedContent({
+        model: 'text-embedding-ada-002',
+        contents: ['first text', 'second text'],
+      });
+      expect(mockEmbeddingsCreate).toHaveBeenCalledWith(
+        expect.objectContaining({ input: 'first text second text' }),
+      );
+    });
+
     it('extracts a plain string content directly', async () => {
       mockEmbeddingsCreate.mockResolvedValue({
         data: [{ embedding: [0.4] }],
@@ -254,24 +289,44 @@ describe('OpenAIResponsesContentGenerator', () => {
       ).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.
+    it('applies SDK client defaults and configured credentials', async () => {
       mockEmbeddingsCreate.mockResolvedValue({
         data: [{ embedding: [0.1] }],
       });
-      const bareOriginGenerator = new OpenAIResponsesContentGenerator(
-        { ...makeGeneratorConfig(), baseUrl: 'https://api.openai.com' },
+      const configuredGenerator = new OpenAIResponsesContentGenerator(
+        {
+          ...makeGeneratorConfig(),
+          baseUrl: 'https://api.openai.com/',
+          timeout: 0,
+        },
         makeCliConfig(),
       );
-      await bareOriginGenerator.embedContent({
+      await configuredGenerator.embedContent({
         model: 'text-embedding-ada-002',
         contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
       });
       expect(mockOpenAIConstructor).toHaveBeenCalledWith(
-        expect.objectContaining({ baseURL: 'https://api.openai.com/v1' }),
+        expect.objectContaining({
+          apiKey: 'test-key',
+          baseURL: 'https://api.openai.com/v1',
+          timeout: DISABLED_REQUEST_TIMEOUT_MS,
+          maxRetries: DEFAULT_MAX_RETRIES,
+        }),
+      );
+    });
+
+    it('applies the repository default timeout when none is configured', async () => {
+      mockEmbeddingsCreate.mockResolvedValue({
+        data: [{ embedding: [0.1] }],
+      });
+
+      await generator.embedContent({
+        model: 'text-embedding-ada-002',
+        contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+      });
+
+      expect(mockOpenAIConstructor).toHaveBeenCalledWith(
+        expect.objectContaining({ timeout: DEFAULT_TIMEOUT }),
       );
     });
 

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

@@ -26,6 +26,10 @@ import {
 } from '../../utils/runtimeFetchOptions.js';
 import { extractTextFromContents } from '../../utils/extract-text-from-contents.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
+import {
+  DEFAULT_MAX_RETRIES,
+  resolveRequestTimeout,
+} from '../openaiContentGenerator/constants.js';
 
 const debugLogger = createDebugLogger('RESPONSES');
 
@@ -68,8 +72,8 @@ export class OpenAIResponsesContentGenerator implements ContentGenerator {
     this.openaiClient = new OpenAISDK({
       apiKey,
       baseURL,
-      timeout: this.contentGeneratorConfig.timeout,
-      maxRetries: this.contentGeneratorConfig.maxRetries,
+      timeout: resolveRequestTimeout(this.contentGeneratorConfig.timeout),
+      maxRetries: this.contentGeneratorConfig.maxRetries ?? DEFAULT_MAX_RETRIES,
       // Mirror the streaming pipeline (responses-pipeline.ts), which
       // applies customHeaders to every request -- without this, headers
       // configured for the streaming path (e.g. a proxy auth header)
@@ -98,7 +102,7 @@ export class OpenAIResponsesContentGenerator implements ContentGenerator {
     userPromptId: string,
   ): Promise<AsyncGenerator<GenerateContentResponse>> {
     const signal = request.config?.abortSignal ?? undefined;
-    return this.pipeline.executeStream(request, userPromptId, signal);
+    return this.pipeline.connectStream(request, userPromptId, signal);
   }
 
   async countTokens(

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

@@ -21,6 +21,7 @@ import type {
   ResponsesApiReasoningItem,
   ResponsesSSEEvent,
 } from './types.js';
+import { getGenAiUsageProvenance } from '../../telemetry/gen-ai-usage.js';
 
 describe('convertResponsesEventToGemini', () => {
   it('emits a plain text chunk for response.output_text.delta', () => {
@@ -47,7 +48,10 @@ describe('convertResponsesEventToGemini', () => {
 
   it('buffers function_call args across deltas and emits on output_item.done', () => {
     const state = new ResponsesStreamState();
-    convertResponsesEventToGemini(
+    // The buffering events (output_item.added and each arguments.delta) must
+    // return null: emitting a functionCall part early would land a spurious
+    // (empty/partial) tool call in history that replays as a duplicate call.
+    const added = convertResponsesEventToGemini(
       {
         event: 'response.output_item.added',
         data: {
@@ -63,7 +67,8 @@ describe('convertResponsesEventToGemini', () => {
       'gpt-5',
       state,
     );
-    convertResponsesEventToGemini(
+    expect(added).toBeNull();
+    const delta1 = convertResponsesEventToGemini(
       {
         event: 'response.function_call_arguments.delta',
         data: { output_index: 0, delta: '{"path":' },
@@ -71,7 +76,8 @@ describe('convertResponsesEventToGemini', () => {
       'gpt-5',
       state,
     );
-    convertResponsesEventToGemini(
+    expect(delta1).toBeNull();
+    const delta2 = convertResponsesEventToGemini(
       {
         event: 'response.function_call_arguments.delta',
         data: { output_index: 0, delta: '"a.ts"}' },
@@ -79,6 +85,7 @@ describe('convertResponsesEventToGemini', () => {
       'gpt-5',
       state,
     );
+    expect(delta2).toBeNull();
     const resp = convertResponsesEventToGemini(
       {
         event: 'response.output_item.done',
@@ -226,6 +233,174 @@ describe('convertResponsesEventToGemini', () => {
     ]);
   });
 
+  it("prefers the done item's authoritative arguments when the delta buffer is corrupt", () => {
+    // A non-compliant proxy delivers output_item.added, a corrupt/partial
+    // arguments delta (non-empty, so `||` keeps it), then output_item.done
+    // with the complete valid arguments. Without the fallback the corrupt
+    // buffer short-circuits and the tool dispatches with empty args.
+    const state = new ResponsesStreamState();
+    convertResponsesEventToGemini(
+      {
+        event: 'response.output_item.added',
+        data: {
+          output_index: 0,
+          item: {
+            type: 'function_call',
+            id: 'fc_1',
+            call_id: 'call_1',
+            name: 'read_file',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    convertResponsesEventToGemini(
+      {
+        event: 'response.function_call_arguments.delta',
+        data: { output_index: 0, delta: '{"path":' },
+      },
+      'gpt-5',
+      state,
+    );
+    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' },
+        },
+      },
+    ]);
+  });
+
+  it('repairs malformed-but-repairable tool-call arguments via the jsonrepair fallback (trailing comma)', () => {
+    // The authoritative arguments string itself is malformed but repairable
+    // (trailing comma). Bare JSON.parse throws; safeJsonParse/jsonrepair
+    // recovers it the same way every sibling wire does.
+    const state = new ResponsesStreamState();
+    convertResponsesEventToGemini(
+      {
+        event: 'response.output_item.added',
+        data: {
+          output_index: 0,
+          item: {
+            type: 'function_call',
+            id: 'fc_1',
+            call_id: 'call_1',
+            name: 'read_file',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    convertResponsesEventToGemini(
+      {
+        event: 'response.function_call_arguments.delta',
+        data: { output_index: 0, delta: '{"path": "a.ts",}' },
+      },
+      'gpt-5',
+      state,
+    );
+    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',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    expect(resp?.candidates?.[0]?.content?.parts).toEqual([
+      {
+        functionCall: {
+          id: 'call_1',
+          name: 'read_file',
+          args: { path: 'a.ts' },
+        },
+      },
+    ]);
+  });
+
+  it('collapses non-object parsed tool-call arguments to {} (array payload)', () => {
+    // A polluted buffer can parse to a valid non-object JSON value (array /
+    // null / primitive). Downstream consumers spread args as a Record, so
+    // anything non-object must collapse to {}.
+    const state = new ResponsesStreamState();
+    convertResponsesEventToGemini(
+      {
+        event: 'response.output_item.added',
+        data: {
+          output_index: 0,
+          item: {
+            type: 'function_call',
+            id: 'fc_1',
+            call_id: 'call_1',
+            name: 'x',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    convertResponsesEventToGemini(
+      {
+        event: 'response.function_call_arguments.delta',
+        data: { output_index: 0, delta: '["a","b"]' },
+      },
+      'gpt-5',
+      state,
+    );
+    const resp = convertResponsesEventToGemini(
+      {
+        event: 'response.output_item.done',
+        data: {
+          output_index: 0,
+          item: {
+            type: 'function_call',
+            id: 'fc_1',
+            call_id: 'call_1',
+            name: 'x',
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    expect(
+      (
+        resp?.candidates?.[0]?.content?.parts?.[0] as {
+          functionCall?: { args?: unknown };
+        }
+      ).functionCall?.args,
+    ).toEqual({});
+  });
+
   describe('reasoning item completion (thoughtSignature round-trip)', () => {
     it('emits a signature-only thought chunk when encrypted_content is present', () => {
       const state = new ResponsesStreamState();
@@ -309,6 +484,59 @@ describe('convertResponsesEventToGemini', () => {
       thoughtsTokenCount: 2,
       cachedContentTokenCount: 1,
     });
+    // A successful completed turn must map to STOP; a wrong reason here would
+    // trigger MAX_TOKENS truncation recovery downstream on every turn.
+    expect(resp?.candidates?.[0]?.finishReason).toBe('STOP');
+  });
+
+  it('records cache provenance as reported when cached_tokens is present', () => {
+    const state = new ResponsesStreamState();
+    const resp = convertResponsesEventToGemini(
+      {
+        event: 'response.completed',
+        data: {
+          response: {
+            id: 'resp_1',
+            usage: {
+              input_tokens: 10,
+              output_tokens: 5,
+              total_tokens: 15,
+              input_tokens_details: { cached_tokens: 3 },
+            },
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    expect(
+      getGenAiUsageProvenance(resp?.usageMetadata ?? undefined)
+        ?.cachedInputTokensReported,
+    ).toBe(true);
+  });
+
+  it('records cache provenance as NOT reported when cached_tokens is absent (absent vs zero)', () => {
+    // cachedContentTokenCount is always materialized to 0, so provenance is the
+    // only signal that distinguishes "provider reported 0" from "not reported".
+    const state = new ResponsesStreamState();
+    const resp = convertResponsesEventToGemini(
+      {
+        event: 'response.completed',
+        data: {
+          response: {
+            id: 'resp_1',
+            usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 },
+          },
+        },
+      },
+      'gpt-5',
+      state,
+    );
+    expect(resp?.usageMetadata?.cachedContentTokenCount).toBe(0);
+    expect(
+      getGenAiUsageProvenance(resp?.usageMetadata ?? undefined)
+        ?.cachedInputTokensReported,
+    ).toBe(false);
   });
 
   it('throws on response.failed', () => {
@@ -336,6 +564,49 @@ describe('convertResponsesEventToGemini', () => {
     ).toThrow(/Responses API error: boom/);
   });
 
+  it('stamps .status/.code on a response.failed with a known error code so retry/fallback gates classify it', () => {
+    // A mid-stream failure arrives after 200 OK; without .status it classifies
+    // as `unknown` and misses every retry / rate-limit / fallback gate.
+    const state = new ResponsesStreamState();
+    let thrown: unknown;
+    try {
+      convertResponsesEventToGemini(
+        {
+          event: 'response.failed',
+          data: {
+            response: {
+              error: { code: 'rate_limit_exceeded', message: 'slow down' },
+            },
+          },
+        },
+        'gpt-5',
+        state,
+      );
+    } catch (err) {
+      thrown = err;
+    }
+    expect(thrown).toBeInstanceOf(Error);
+    expect((thrown as { status?: number }).status).toBe(429);
+    expect((thrown as { code?: string }).code).toBe('rate_limit_exceeded');
+    expect((thrown as Error).message).toContain('rate_limit_exceeded');
+  });
+
+  it('maps a mid-stream error event server_error code to HTTP 500 status', () => {
+    const state = new ResponsesStreamState();
+    let thrown: unknown;
+    try {
+      convertResponsesEventToGemini(
+        { event: 'error', data: { message: 'boom', code: 'server_error' } },
+        'gpt-5',
+        state,
+      );
+    } catch (err) {
+      thrown = err;
+    }
+    expect((thrown as { status?: number }).status).toBe(500);
+    expect((thrown as { code?: string }).code).toBe('server_error');
+  });
+
   describe('response.incomplete', () => {
     it('extracts usage and maps incomplete_details.reason to MAX_TOKENS', () => {
       const state = new ResponsesStreamState();
@@ -731,6 +1002,48 @@ describe('convertGeminiContentsToResponsesInput', () => {
     ]);
   });
 
+  it('surfaces tool-result image media as a follow-up user input_image message instead of dropping it', () => {
+    // A tool that returns an image stores it in functionResponse.parts; the
+    // string-only function_call_output.output can't carry it, so it must be
+    // emitted as a follow-up user message rather than silently dropped.
+    const { input } = convertGeminiContentsToResponsesInput(
+      request([
+        {
+          role: 'user',
+          parts: [
+            {
+              functionResponse: {
+                id: 'call_1',
+                response: { output: 'see image' },
+                parts: [
+                  { inlineData: { mimeType: 'image/png', data: 'YWJj' } },
+                ],
+              },
+            },
+          ],
+        },
+      ]),
+    );
+    expect(input).toEqual([
+      {
+        type: 'function_call_output',
+        call_id: 'call_1',
+        output: 'see image',
+      } satisfies ResponsesApiFunctionCallOutputItem,
+      {
+        type: 'message',
+        role: 'user',
+        content: [
+          {
+            type: 'input_text',
+            text: '(attached media from previous tool call)',
+          },
+          { type: 'input_image', image_url: 'data:image/png;base64,YWJj' },
+        ],
+      } satisfies ResponsesApiMessageItem,
+    ]);
+  });
+
   it('unwraps a { error } tool response envelope to the bare string', () => {
     const { input } = convertGeminiContentsToResponsesInput(
       request([
@@ -837,6 +1150,43 @@ describe('convertGeminiToolsToResponsesTools', () => {
     ).toBeUndefined();
   });
 
+  it('converts every declaration across multiple Tool entries (not just the first)', () => {
+    const tools = convertGeminiToolsToResponsesTools({
+      model: 'gpt-5',
+      contents: [],
+      config: {
+        tools: [
+          {
+            functionDeclarations: [
+              {
+                name: 'read_file',
+                parametersJsonSchema: { type: 'object', properties: {} },
+              },
+              {
+                name: 'write_file',
+                parametersJsonSchema: { type: 'object', properties: {} },
+              },
+            ],
+          },
+        ],
+      },
+    });
+    expect(tools?.map((t) => t.name)).toEqual(['read_file', 'write_file']);
+  });
+
+  it('returns undefined (not tools: []) when declarations are present but empty', () => {
+    // Exercises the `result.length > 0 ? result : undefined` ternary: an empty
+    // array would otherwise serialize `tools: []` alongside the unconditional
+    // tool_choice.
+    expect(
+      convertGeminiToolsToResponsesTools({
+        model: 'gpt-5',
+        contents: [],
+        config: { tools: [{ functionDeclarations: [] }] },
+      }),
+    ).toBeUndefined();
+  });
+
   it('normalizes a zero-arg tool schema missing properties (Azure/litellm compatibility)', () => {
     // Every other case here already has `properties`, so
     // normalizeResponsesParameters is a no-op for them -- this is the only
@@ -897,11 +1247,37 @@ describe('normalizeResponsesParameters', () => {
     });
   });
 
+  it('recurses into bare object schemas nested under oneOf and allOf', () => {
+    const schema = {
+      type: 'object',
+      properties: {},
+      oneOf: [{ type: 'object' }],
+      allOf: [{ type: 'object' }],
+    };
+    expect(normalizeResponsesParameters(schema)).toEqual({
+      type: 'object',
+      properties: {},
+      oneOf: [{ type: 'object', properties: {} }],
+      allOf: [{ type: 'object', properties: {} }],
+    });
+  });
+
   it('passes through a well-formed schema unchanged', () => {
     const schema = { type: 'object', properties: { path: { type: 'string' } } };
     expect(normalizeResponsesParameters(schema)).toEqual(schema);
   });
 
+  it('does not mutate the caller-provided schema object in place', () => {
+    // Tool declarations are shared across requests; an in-place mutation would
+    // corrupt the caller's object. The normalizer must return a fresh tree.
+    const schema = { type: 'object' };
+    const snapshot = structuredClone(schema);
+    const result = normalizeResponsesParameters(schema);
+    expect(schema).toEqual(snapshot);
+    expect(result).not.toBe(schema);
+    expect(result).toEqual({ type: 'object', properties: {} });
+  });
+
   it('passes through undefined', () => {
     expect(normalizeResponsesParameters(undefined)).toBeUndefined();
   });

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

@@ -26,9 +26,42 @@ import type {
   ResponsesApiContentPart,
 } from './types.js';
 import { createDebugLogger } from '../../utils/debugLogger.js';
+import { safeJsonParse } from '../../utils/safeJsonParse.js';
+import { setGenAiUsageProvenance } from '../../telemetry/gen-ai-usage.js';
 
 const debugLogger = createDebugLogger('RESPONSES_CONVERTER');
 
+/**
+ * Known OpenAI Responses API error codes that carry a well-defined HTTP
+ * status. Mid-stream `error` / `response.failed` frames arrive after a 200 OK,
+ * so unless the thrown error is stamped with a `.status`/`.code` it classifies
+ * as `unknown` and misses every retry / rate-limit / model-fallback gate (see
+ * utils/retryErrorClassification.ts and utils/rateLimit.ts) — the same failure
+ * at connection time carries `.status` (responses-pipeline.ts) and is retried.
+ */
+const RESPONSES_ERROR_CODE_TO_STATUS: Record<string, number> = {
+  server_error: 500,
+  rate_limit_exceeded: 429,
+};
+
+interface ResponsesStreamError extends Error {
+  status?: number;
+  code?: string;
+}
+
+function makeResponsesStreamError(
+  message: string,
+  code?: string,
+): ResponsesStreamError {
+  const err: ResponsesStreamError = new Error(message);
+  if (code) {
+    err.code = code;
+    const status = RESPONSES_ERROR_CODE_TO_STATUS[code];
+    if (status !== undefined) err.status = status;
+  }
+  return err;
+}
+
 /**
  * Opaque payload stashed in `part.thoughtSignature` for a Responses API
  * reasoning item, so it round-trips through this app's own Content[] history
@@ -173,6 +206,23 @@ export function convertResponsesEventToGemini(
         try {
           args = JSON.parse(rawArgs) as Record<string, unknown>;
         } catch {
+          // The accumulated delta buffer can be corrupt or truncated (a
+          // non-compliant proxy) while the done item still carries the
+          // complete authoritative `arguments` string — prefer that when it
+          // differs, so a valid final payload isn't shadowed by a broken
+          // buffer. Otherwise repair the string we have with the same
+          // jsonrepair fallback every sibling wire applies (safeJsonParse,
+          // streamingToolCallParser.getCompletedToolCalls, the Chat converter).
+          const authoritative =
+            fc.arguments && fc.arguments !== rawArgs ? fc.arguments : rawArgs;
+          args = safeJsonParse(authoritative, {}) as Record<string, unknown>;
+        }
+        // Tool arguments are always JSON objects; a polluted buffer or proxy
+        // payload can parse/repair to an array/null/primitive, which would
+        // otherwise flow through the cast into functionCall.args and break the
+        // Record-spreading consumers downstream (geminiChat, sessionService).
+        // Collapse anything else to {}, matching the sibling's guard.
+        if (typeof args !== 'object' || args === null || Array.isArray(args)) {
           args = {};
         }
         return makeChunkResponse(model, state, [
@@ -226,12 +276,15 @@ export function convertResponsesEventToGemini(
     case 'response.failed': {
       const raw = event.data as Record<string, unknown>;
       const envelope = (raw['response'] ?? raw) as {
-        error?: { code: string; message: string };
+        error?: { code?: string; message?: string };
       };
       const errMsg = envelope.error
         ? `${envelope.error.code}: ${envelope.error.message}`
         : 'Response failed';
-      throw new Error(`Responses API failed: ${errMsg}`);
+      throw makeResponsesStreamError(
+        `Responses API failed: ${errMsg}`,
+        envelope.error?.code,
+      );
     }
 
     case 'response.incomplete': {
@@ -254,9 +307,10 @@ export function convertResponsesEventToGemini(
     }
 
     case 'error': {
-      const data = event.data as { message?: string };
-      throw new Error(
+      const data = event.data as { message?: string; code?: string };
+      throw makeResponsesStreamError(
         `Responses API error: ${data.message ?? 'Unknown error'}`,
+        data.code,
       );
     }
 
@@ -305,6 +359,15 @@ function makeFinalResponse(
       thoughtsTokenCount: usage.output_tokens_details?.reasoning_tokens ?? 0,
       cachedContentTokenCount: usage.input_tokens_details?.cached_tokens ?? 0,
     };
+    // cachedContentTokenCount is always materialized (`?? 0`), so without
+    // provenance the logging layer's absent-vs-zero fallback
+    // (loggingContentGenerator.usageSpanMetadata) would report a provider
+    // cache measurement on every turn even when none was sent. Mirror the
+    // Chat converter: record whether cached_tokens was actually present.
+    setGenAiUsageProvenance(resp.usageMetadata, {
+      cachedInputTokensReported:
+        typeof usage.input_tokens_details?.cached_tokens === 'number',
+    });
   }
 
   return resp;
@@ -375,6 +438,11 @@ export function convertGeminiContentsToResponsesInput(
     // Flushed before any function_call/function_call_output/reasoning item
     // so relative ordering within the turn is preserved.
     let pendingContentParts: ResponsesApiContentPart[] = [];
+    // Media attached to tool results (functionResponse.parts) is staged here
+    // and flushed as one follow-up user message after all tool outputs in this
+    // turn, so tool-returned images aren't silently dropped. Mirrors the Chat
+    // converter's split-tool-media follow-up.
+    const pendingToolMediaParts: ResponsesApiContentPart[] = [];
     const flushPendingMessage = () => {
       if (pendingContentParts.length === 0) return;
       const content: string | ResponsesApiContentPart[] =
@@ -465,6 +533,47 @@ export function convertGeminiContentsToResponsesInput(
         } else {
           output = JSON.stringify(fr.response ?? {});
         }
+        // A tool can return media (e.g. read_file on an image) in
+        // functionResponse.parts alongside its textual output. The
+        // function_call_output.output field is string-only, so surface the
+        // media the way the sibling Chat converter does
+        // (createToolMessage/splitToolMedia) instead of dropping it: append
+        // any text entries (e.g. compaction-slimmer placeholders) to the
+        // output, and stage image entries for a follow-up user message.
+        for (const frPart of fr.parts ?? []) {
+          if (
+            'text' in frPart &&
+            typeof (frPart as { text?: unknown }).text === 'string' &&
+            (frPart as { text: string }).text.length > 0
+          ) {
+            const text = (frPart as { text: string }).text;
+            output += (output ? '\n' : '') + text;
+          } else if (frPart.inlineData) {
+            const mimeType = frPart.inlineData.mimeType ?? 'image/png';
+            if (mimeType.startsWith('image/')) {
+              pendingToolMediaParts.push({
+                type: 'input_image',
+                image_url: `data:${mimeType};base64,${frPart.inlineData.data}`,
+              });
+            } else {
+              debugLogger.warn(
+                `Dropping unsupported tool-result inline media type: ${mimeType}`,
+              );
+              pendingToolMediaParts.push({
+                type: 'input_text',
+                text: `[Unsupported tool-result media type: ${mimeType}]`,
+              });
+            }
+          } else if (frPart.fileData) {
+            debugLogger.warn(
+              `Dropping unsupported tool-result file reference: ${frPart.fileData.mimeType ?? 'unknown mime type'}`,
+            );
+            pendingToolMediaParts.push({
+              type: 'input_text',
+              text: `[Unsupported tool-result file reference: ${frPart.fileData.mimeType ?? 'unknown mime type'}]`,
+            });
+          }
+        }
         items.push({
           type: 'function_call_output',
           call_id: fr.id || `call_${Date.now()}_${callIdCounter++}`,
@@ -501,6 +610,21 @@ export function convertGeminiContentsToResponsesInput(
       }
     }
     flushPendingMessage();
+    if (pendingToolMediaParts.length > 0) {
+      // Emitted after all function_call_output items in this turn so tool
+      // outputs stay grouped; the model still sees tool-returned images.
+      items.push({
+        type: 'message',
+        role: 'user',
+        content: [
+          {
+            type: 'input_text',
+            text: '(attached media from previous tool call)',
+          },
+          ...pendingToolMediaParts,
+        ],
+      } as ResponsesApiMessageItem);
+    }
   }
 
   return { instructions, input: items };

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

@@ -14,8 +14,12 @@ import {
   afterEach,
 } from 'vitest';
 import type { GenerateContentParameters } from '@google/genai';
-import { FunctionCallingConfigMode } from '@google/genai';
-import { ResponsesPipeline } from './responses-pipeline.js';
+import { FunctionCallingConfigMode, FinishReason } from '@google/genai';
+import {
+  ResponsesPipeline,
+  mergeStreamResponses,
+  StreamInactivityTimeoutError,
+} from './responses-pipeline.js';
 import type { Config } from '../../config/config.js';
 import type { ContentGeneratorConfig } from '../contentGenerator.js';
 import type { ResponsesApiRequest } from './types.js';
@@ -651,4 +655,414 @@ describe('ResponsesPipeline', () => {
     ]);
     expect(result.usageMetadata?.totalTokenCount).toBe(2);
   });
+
+  // --- Critical (a): per-send window-clamped output budget (#5950) ---
+
+  it('sends request.config.maxOutputTokens as max_output_tokens when no samplingParams cap is set', async () => {
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const request: GenerateContentParameters = {
+      model: 'gpt-5',
+      contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+      config: { maxOutputTokens: 512 },
+    };
+    for await (const _ of pipeline.executeStream(request, 'p1')) {
+      // drain
+    }
+    const body = JSON.parse(
+      fetchMock.mock.calls[0]![1].body,
+    ) as ResponsesApiRequest;
+    expect(body.max_output_tokens).toBe(512);
+  });
+
+  it('lets request.config.maxOutputTokens override the static samplingParams.max_tokens', async () => {
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig({ samplingParams: { max_tokens: 2048 } }),
+      makeCliConfig(),
+    );
+    const request: GenerateContentParameters = {
+      model: 'gpt-5',
+      contents: [{ role: 'user', parts: [{ text: 'hi' }] }],
+      config: { maxOutputTokens: 512 },
+    };
+    for await (const _ of pipeline.executeStream(request, 'p1')) {
+      // drain
+    }
+    const body = JSON.parse(
+      fetchMock.mock.calls[0]![1].body,
+    ) as ResponsesApiRequest;
+    expect(body.max_output_tokens).toBe(512);
+  });
+
+  // --- Critical (b): mid-stream idle-timeout watchdog ---
+
+  it('fails a silent mid-stream stall with a retryable ETIMEDOUT after the idle timeout', async () => {
+    vi.useFakeTimers();
+    try {
+      const encoder = new TextEncoder();
+      // Emit one delta frame, then go silent forever: the never-resolving
+      // pull() keeps the second reader.read() pending so only the idle
+      // watchdog can end the stream.
+      const body = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(
+            encoder.encode(
+              `data: ${JSON.stringify({
+                type: 'response.output_text.delta',
+                delta: 'hi',
+              })}\n`,
+            ),
+          );
+        },
+        pull() {
+          return new Promise<void>(() => {});
+        },
+      });
+      fetchMock.mockResolvedValue({
+        ok: true,
+        status: 200,
+        headers: { get: () => 'text/event-stream' },
+        body,
+        text: async () => '',
+      });
+      const pipeline = new ResponsesPipeline(
+        makeGeneratorConfig({ streamIdleTimeoutMs: 1000 }),
+        makeCliConfig(),
+      );
+      const gen = pipeline.executeStream(textRequest('hi'), 'p1');
+      const first = await gen.next();
+      expect(first.value?.candidates?.[0]?.content?.parts).toEqual([
+        { text: 'hi' },
+      ]);
+      const pending = gen.next();
+      // eslint-disable-next-line vitest/valid-expect -- awaited via `assertion` below, after the fake timers advance (handler attached early so the timeout rejection is not unhandled)
+      const assertion = expect(pending).rejects.toBeInstanceOf(
+        StreamInactivityTimeoutError,
+      );
+      await vi.advanceTimersByTimeAsync(1000);
+      await assertion;
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it('does not fire the idle watchdog while chunks keep arriving', async () => {
+    // A fully delivered, promptly-closing stream must complete cleanly even
+    // with a tiny idle timeout configured -- the timer resets per chunk and
+    // the terminal `done` resolves before it can fire.
+    mockResponse([
+      ...sseEvent('response.output_text.delta', { delta: 'foo' }),
+      ...sseEvent('response.completed', { response: { status: 'completed' } }),
+    ]);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig({ streamIdleTimeoutMs: 1000 }),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      chunks.push(chunk);
+    }
+    expect(chunks.map((c) => c.candidates?.[0]?.content?.parts)).toEqual([
+      [{ text: 'foo' }],
+      [],
+    ]);
+  });
+
+  // --- Critical (c): eager connect so retryWithBackoff sees connect errors ---
+
+  it('connectStream performs the fetch eagerly, before the body is iterated', async () => {
+    mockResponse([
+      ...sseEvent('response.output_text.delta', { delta: 'hi' }),
+      ...sseEvent('response.completed', { response: { status: 'completed' } }),
+    ]);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const gen = await pipeline.connectStream(textRequest('hi'), 'p1');
+    // Network I/O already happened while awaiting the returned promise -- a
+    // lazy `async *` generator would leave this at 0 until the first next().
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+    for await (const _ of gen) {
+      // drain
+    }
+  });
+
+  it('connectStream rejects on a connection-time HTTP error so retry sees it', async () => {
+    // A 5xx at request time must reject the awaited promise (which
+    // generateContentStream returns into retryWithBackoff) rather than
+    // escaping later during lazy iteration outside the retry wrapper.
+    fetchMock.mockResolvedValue({
+      ok: false,
+      status: 500,
+      text: async () => '{"error":"server"}',
+    });
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    await expect(
+      pipeline.connectStream(textRequest('hi'), 'p1'),
+    ).rejects.toThrow(/Responses API error 500/);
+    expect(fetchMock).toHaveBeenCalledTimes(1);
+  });
+
+  // --- Suggestions: previously untested added behavior ---
+
+  it('accumulates a multi-line data: block (joined with \\n) into a single frame', async () => {
+    // Every other test emits exactly one data: line per event, so the
+    // `event: ` + multi-`data:` accumulation path never runs; a last-line-wins
+    // mutant would silently drop a split frame.
+    const lines = [
+      'event: response.output_text.delta',
+      'data: {"type":"response.output_text.delta",',
+      'data: "delta":"multi"}',
+      '',
+    ];
+    mockResponse(lines);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      chunks.push(chunk);
+    }
+    expect(chunks.map((c) => c.candidates?.[0]?.content?.parts)).toEqual([
+      [{ text: 'multi' }],
+    ]);
+  });
+
+  it('decodes multi-byte UTF-8 split across reader.read() chunks', async () => {
+    // JSON.stringify emits raw (unescaped) non-ASCII, so byte-chunking splits
+    // real multi-byte characters across reads -- exercising the decoder's
+    // { stream: true } flag. Dropping it corrupts split code points to U+FFFD.
+    const text = '你好😀世界';
+    const lines = [
+      `data: ${JSON.stringify({ type: 'response.output_text.delta', delta: text })}`,
+      `data: ${JSON.stringify({
+        type: 'response.completed',
+        response: { id: 'r1', status: 'completed' },
+      })}`,
+    ];
+    fetchMock.mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: { get: () => 'text/event-stream' },
+      body: sseStreamChunked(lines, 3),
+      text: async () => '',
+    });
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      chunks.push(chunk);
+    }
+    expect(chunks[0]?.candidates?.[0]?.content?.parts).toEqual([{ text }]);
+  });
+
+  it('does not let extra_body overwrite an explicit samplingParams.temperature of 0', async () => {
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig({
+        samplingParams: { temperature: 0 },
+        extra_body: { temperature: 1 },
+      }),
+      makeCliConfig(),
+    );
+    for await (const _ of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      // drain
+    }
+    const body = JSON.parse(
+      fetchMock.mock.calls[0]![1].body,
+    ) as ResponsesApiRequest;
+    // The production guard checks `requestRecord[key] === undefined`, so an
+    // explicit 0 is preserved; a truthiness guard would let extra_body win.
+    expect(body.temperature).toBe(0);
+  });
+
+  it('propagates responseId and finishReason through execute()/mergeStreamResponses', async () => {
+    mockResponse([
+      ...sseEvent('response.output_text.delta', { delta: 'foo' }),
+      ...sseEvent('response.completed', {
+        response: {
+          id: 'r1',
+          status: 'completed',
+          usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 },
+        },
+      }),
+    ]);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const result = await pipeline.execute(textRequest('hi'), 'p1');
+    expect(result.responseId).toBe('r1');
+    expect(result.candidates?.[0]?.finishReason).toBe(FinishReason.STOP);
+  });
+
+  it('applies customHeaders to the fetch request', async () => {
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig({ customHeaders: { 'X-Proxy-Auth': 'token' } }),
+      makeCliConfig(),
+    );
+    for await (const _ of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      // drain
+    }
+    expect(fetchMock.mock.calls[0]![1].headers['X-Proxy-Auth']).toBe('token');
+  });
+
+  it('forwards the exact AbortSignal to fetch', async () => {
+    const controller = new AbortController();
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    for await (const _ of pipeline.executeStream(
+      textRequest('hi'),
+      'p1',
+      controller.signal,
+    )) {
+      // drain
+    }
+    expect(fetchMock.mock.calls[0]![1].signal).toBe(controller.signal);
+  });
+
+  it('aborts the read loop when the AbortSignal fires mid-stream', async () => {
+    const controller = new AbortController();
+    const body = new ReadableStream<Uint8Array>({
+      start(c) {
+        controller.signal.addEventListener('abort', () =>
+          c.error(new DOMException('Aborted', 'AbortError')),
+        );
+      },
+    });
+    fetchMock.mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: { get: () => 'text/event-stream' },
+      body,
+      text: async () => '',
+    });
+    // Idle watchdog off so the abort, not the timer, is what ends the stream.
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig({ streamIdleTimeoutMs: 0 }),
+      makeCliConfig(),
+    );
+    const gen = pipeline.executeStream(
+      textRequest('hi'),
+      'p1',
+      controller.signal,
+    );
+    const pending = gen.next();
+    await Promise.resolve();
+    controller.abort();
+    await expect(pending).rejects.toThrow(/Abort/);
+  });
+
+  it('forwards the runtime dispatcher to fetch', async () => {
+    const dispatcher = { marker: 'dispatcher' };
+    buildRuntimeFetchOptionsMock.mockReturnValue({
+      fetch: fetchMock,
+      fetchOptions: { dispatcher },
+    });
+    mockResponse(
+      sseEvent('response.completed', { response: { status: 'completed' } }),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    for await (const _ of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      // drain
+    }
+    expect(fetchMock.mock.calls[0]![1].dispatcher).toBe(dispatcher);
+  });
+
+  it('redacts proxy credentials from a fetch rejection', async () => {
+    fetchMock.mockRejectedValue(
+      new Error('fetch failed http://user:[email protected]:8080'),
+    );
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    let caught: unknown;
+    try {
+      for await (const _ of pipeline.executeStream(textRequest('hi'), 'p1')) {
+        // drain
+      }
+    } catch (err) {
+      caught = err;
+    }
+    const message = (caught as Error).message;
+    expect(message).toMatch(/<redacted>@proxy\.example/);
+    expect(message).not.toContain('secret');
+  });
+
+  it('rejects a 200 response whose content-type is not SSE', async () => {
+    fetchMock.mockResolvedValue({
+      ok: true,
+      status: 200,
+      headers: { get: () => 'application/json' },
+      body: sseStream(
+        sseEvent('response.completed', { response: { status: 'completed' } }),
+      ),
+      text: async () => '',
+    });
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    await expect(async () => {
+      for await (const _ of pipeline.executeStream(textRequest('hi'), 'p1')) {
+        // drain
+      }
+    }).rejects.toThrow(/non-SSE content-type/);
+  });
+
+  it('skips an unparseable data: frame and still yields the surrounding valid events', async () => {
+    const lines = [
+      `data: ${JSON.stringify({ type: 'response.output_text.delta', delta: 'a' })}`,
+      'data: {not valid json',
+      `data: ${JSON.stringify({ type: 'response.output_text.delta', delta: 'b' })}`,
+      'data: [DONE]',
+    ];
+    mockResponse(lines);
+    const pipeline = new ResponsesPipeline(
+      makeGeneratorConfig(),
+      makeCliConfig(),
+    );
+    const chunks = [];
+    for await (const chunk of pipeline.executeStream(textRequest('hi'), 'p1')) {
+      chunks.push(chunk);
+    }
+    expect(chunks.map((c) => c.candidates?.[0]?.content?.parts)).toEqual([
+      [{ text: 'a' }],
+      [{ text: 'b' }],
+    ]);
+  });
+
+  it('mergeStreamResponses([]) returns an empty candidates array, not undefined', () => {
+    const merged = mergeStreamResponses([]);
+    expect(merged.candidates).toEqual([]);
+  });
 });

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

@@ -25,18 +25,76 @@ import {
   buildRuntimeFetchOptions,
   redactProxyError,
 } from '../../utils/runtimeFetchOptions.js';
+import {
+  DEFAULT_STREAM_IDLE_TIMEOUT_MS,
+  MAX_STREAM_GUARD_TIMEOUT_MS,
+  QWEN_STREAM_IDLE_TIMEOUT_MS_ENV,
+} from '../openaiContentGenerator/constants.js';
 import { createHash } from 'node:crypto';
 import { createDebugLogger } from '../../utils/debugLogger.js';
 
 const debugLogger = createDebugLogger('RESPONSES_PIPELINE');
 
+/**
+ * Thrown when the SSE read loop goes silent past the inactivity timeout.
+ * `code: 'ETIMEDOUT'` makes `classifyRetryError` treat it as a retryable
+ * transport error, identical to a real socket read timeout -- mirroring the
+ * sibling Chat pipeline's StreamInactivityTimeoutError (which cannot be reused
+ * directly here because it is typed to an OpenAI chunk async-iterable, whereas
+ * this wire reads a raw byte ReadableStream).
+ */
+export class StreamInactivityTimeoutError extends Error {
+  readonly code = 'ETIMEDOUT' as const;
+
+  constructor(
+    readonly idleMs: number,
+    readonly chunksReceived: number,
+  ) {
+    super(
+      `No stream activity for ${idleMs}ms after ${chunksReceived} chunks. ` +
+        `Set ${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV} to increase this window ` +
+        `(or 0 to disable it).`,
+    );
+    this.name = 'StreamInactivityTimeoutError';
+  }
+}
+
+// Resolve the streaming inactivity timeout (ms): explicit config field wins,
+// then the env deployment knob, then the default -- mirroring the sibling Chat
+// pipeline's resolveStreamIdleTimeoutMs. `<= 0` disables the watchdog; a value
+// above the JS timer ceiling or a non-integer is rejected (setTimeout silently
+// compresses oversized delays to ~1ms, which would fire near-immediately).
+function resolveStreamIdleTimeoutMs(config: ContentGeneratorConfig): number {
+  const fromConfig = config.streamIdleTimeoutMs;
+  if (typeof fromConfig === 'number') {
+    if (
+      Number.isInteger(fromConfig) &&
+      fromConfig <= MAX_STREAM_GUARD_TIMEOUT_MS
+    ) {
+      return fromConfig;
+    }
+  }
+  const raw = process.env[QWEN_STREAM_IDLE_TIMEOUT_MS_ENV]?.trim();
+  if (raw && /^\d+$/.test(raw)) {
+    const parsed = Number(raw);
+    if (parsed <= MAX_STREAM_GUARD_TIMEOUT_MS) {
+      return parsed;
+    }
+  }
+  return DEFAULT_STREAM_IDLE_TIMEOUT_MS;
+}
+
 export class ResponsesPipeline {
   private readonly config: ContentGeneratorConfig;
   private readonly cliConfig: Config;
+  // Resolved once (config field > env > default) so the env read happens per
+  // pipeline, not per streaming request.
+  private readonly streamIdleTimeoutMs: number;
 
   constructor(config: ContentGeneratorConfig, cliConfig: Config) {
     this.config = config;
     this.cliConfig = cliConfig;
+    this.streamIdleTimeoutMs = resolveStreamIdleTimeoutMs(config);
   }
 
   async *executeStream(
@@ -44,9 +102,28 @@ export class ResponsesPipeline {
     userPromptId: string,
     signal?: AbortSignal,
   ): AsyncGenerator<GenerateContentResponse> {
+    yield* await this.connectStream(request, userPromptId, signal);
+  }
+
+  /**
+   * Eagerly performs the connect phase (fetch + non-2xx handling + content-type
+   * guard) and returns a lazy generator over the response body. Callers that
+   * wrap the returned promise in retryWithBackoff (index.ts's
+   * generateContentStream) then see connection-time HTTP errors -- a 500/502/504
+   * or a 429 storm -- surface from the awaited promise and get retried, instead
+   * of the errors escaping later during lazy iteration outside the retry
+   * wrapper (a lazy `async *` generator resolves its promise before any network
+   * I/O runs).
+   */
+  async connectStream(
+    request: GenerateContentParameters,
+    userPromptId: string,
+    signal?: AbortSignal,
+  ): Promise<AsyncGenerator<GenerateContentResponse>> {
     const activeRequest = this.buildRequest(request, userPromptId);
     const streamState = new ResponsesStreamState();
-    yield* this.streamRequest(activeRequest, streamState, signal);
+    const reader = await this.connect(activeRequest, signal);
+    return this.iterateBody(reader, streamState, signal);
   }
 
   async execute(
@@ -91,7 +168,7 @@ export class ResponsesPipeline {
       prompt_cache_key: sanitizePromptCacheKey(userPromptId),
       // We never reference previous_response_id, so server-side storage of
       // the response buys nothing. The spec pairs reasoning.encrypted_content
-      // with store:false as the intended stateless-replay recipe — set it
+      // with store:false as the intended stateless-replay recipe -- set it
       // unconditionally for consistency, not just when reasoning is on.
       store: false,
     };
@@ -126,9 +203,16 @@ export class ResponsesPipeline {
       if (this.config.samplingParams.top_p != null) {
         apiRequest.top_p = this.config.samplingParams.top_p;
       }
-      if (this.config.samplingParams.max_tokens != null) {
-        apiRequest.max_output_tokens = this.config.samplingParams.max_tokens;
-      }
+    }
+
+    // The per-send output budget geminiChat window-clamps lives on
+    // request.config.maxOutputTokens (issue #5950 invariant: prompt +
+    // max_tokens <= window); it must win over the static samplingParams cap so
+    // subagent budgets and the MAX_TOKENS escalation re-clamp reach this wire.
+    const maxOut =
+      request.config?.maxOutputTokens ?? this.config.samplingParams?.max_tokens;
+    if (maxOut != null) {
+      apiRequest.max_output_tokens = maxOut;
     }
 
     if (this.config.extra_body) {
@@ -136,7 +220,7 @@ export class ResponsesPipeline {
       for (const [key, value] of Object.entries(this.config.extra_body)) {
         // apiRequest is built from an object literal, so optional fields
         // like `tools`/`instructions` are already present as keys even when
-        // their value is undefined — check the value, not just key presence,
+        // their value is undefined -- check the value, not just key presence,
         // so extra_body can still fill in a field nothing else set.
         if (!(key in requestRecord) || requestRecord[key] === undefined) {
           requestRecord[key] = value;
@@ -165,11 +249,16 @@ export class ResponsesPipeline {
     return Object.keys(reasoning).length > 0 ? reasoning : undefined;
   }
 
-  private async *streamRequest(
+  /**
+   * Connect phase: build and send the request, validate the HTTP response, and
+   * return a reader over the SSE body. Eagerly awaited so connection-time
+   * failures (non-2xx, missing body, non-SSE content-type, transport errors)
+   * reject the promise the caller can wrap in retryWithBackoff.
+   */
+  private async connect(
     apiRequest: ResponsesApiRequest,
-    streamState: ResponsesStreamState,
     signal?: AbortSignal,
-  ): AsyncGenerator<GenerateContentResponse> {
+  ): Promise<ReadableStreamDefaultReader<Uint8Array>> {
     const baseUrl = (this.config.baseUrl || 'https://api.openai.com')
       .replace(/\/v1\/?$/, '')
       .replace(/\/$/, '');
@@ -251,6 +340,58 @@ export class ResponsesPipeline {
       );
     }
 
+    return response.body.getReader();
+  }
+
+  /**
+   * Body-iteration phase: lazily drain the SSE reader, parse frames, and yield
+   * Gemini responses. An idle watchdog around each `reader.read()` fails a
+   * silent mid-stream stall with a retryable ETIMEDOUT (mirroring the Chat
+   * pipeline's inactivity guard) instead of hanging forever; the timer resets
+   * on every chunk, so an actively streaming model is never interrupted, and
+   * `streamIdleTimeoutMs <= 0` disables it.
+   */
+  private async *iterateBody(
+    reader: ReadableStreamDefaultReader<Uint8Array>,
+    streamState: ResponsesStreamState,
+    signal?: AbortSignal,
+  ): AsyncGenerator<GenerateContentResponse> {
+    const idleMs = this.streamIdleTimeoutMs;
+    let chunksReceived = 0;
+
+    const readChunk = (): Promise<ReadableStreamReadResult<Uint8Array>> => {
+      if (idleMs <= 0) return reader.read();
+      const readPromise = reader.read();
+      let timer: ReturnType<typeof setTimeout> | undefined;
+      const timeout = new Promise<never>((_resolve, reject) => {
+        timer = setTimeout(() => {
+          // A user cancellation wins over the watchdog's retryable ETIMEDOUT.
+          // Reject only -- do NOT cancel the reader here: cancel() resolves the
+          // pending read() as { done: true } synchronously, which would win the
+          // race over this rejection. The generator's finally cancels the
+          // reader (freeing the socket) once this error propagates out.
+          if (signal?.aborted) {
+            const abortErr = new Error('Aborted');
+            abortErr.name = 'AbortError';
+            reject(abortErr);
+          } else {
+            reject(new StreamInactivityTimeoutError(idleMs, chunksReceived));
+          }
+        }, idleMs);
+        timer.unref?.();
+      });
+      return Promise.race([readPromise, timeout])
+        .catch((err) => {
+          // The orphaned read() rejects once the body is cancelled; swallow it
+          // so it is not an unhandled rejection.
+          void Promise.resolve(readPromise).catch(() => {});
+          throw err;
+        })
+        .finally(() => {
+          if (timer !== undefined) clearTimeout(timer);
+        });
+    };
+
     const convertParsedFrame = (
       eventType: ResponsesSSEEventType,
       data: Record<string, unknown>,
@@ -286,7 +427,6 @@ export class ResponsesPipeline {
       }
     };
 
-    const reader = response.body.getReader();
     const decoder = new TextDecoder();
     let buffer = '';
     let currentEventType: ResponsesSSEEventType | null = null;
@@ -294,8 +434,9 @@ export class ResponsesPipeline {
 
     try {
       while (true) {
-        const { done, value } = await reader.read();
+        const { done, value } = await readChunk();
         if (done) break;
+        chunksReceived += 1;
 
         buffer += decoder.decode(value, { stream: true });
         const lines = buffer.split('\n');

packages/core/src/providers/__tests__/presets/custom-provider.test.ts

@@ -7,14 +7,14 @@
 import { describe, expect, it } from 'vitest';
 import {
   AuthType,
-  customProvider,
   CUSTOM_API_KEY_ENV_PREFIX,
   buildInstallPlan,
   shouldShowStep,
 } from '@qwen-code/qwen-code-core';
-// Re-import generateCustomEnvKey from the relative source path so the new
-// hash-suffix format is exercised even before dist/ is rebuilt.
-import { generateCustomEnvKey } from '../../presets/custom-provider.js';
+import {
+  customProvider,
+  generateCustomEnvKey,
+} from '../../presets/custom-provider.js';
 
 describe('generateCustomEnvKey', () => {
   it('produces a deterministic URL-based key with a stable hash suffix', () => {
@@ -105,11 +105,26 @@ describe('customProvider', () => {
   it('offers multiple protocol options', () => {
     expect(customProvider.protocolOptions).toEqual([
       AuthType.USE_OPENAI,
+      AuthType.USE_OPENAI_RESPONSES,
       AuthType.USE_ANTHROPIC,
       AuthType.USE_GEMINI,
     ]);
   });
 
+  it('builds an OpenAI Responses install plan', () => {
+    const plan = buildInstallPlan(customProvider, {
+      protocol: AuthType.USE_OPENAI_RESPONSES,
+      baseUrl: 'https://api.example.com',
+      apiKey: 'sk-responses',
+      modelIds: ['gpt-5'],
+    });
+
+    expect(plan.authType).toBe(AuthType.USE_OPENAI_RESPONSES);
+    expect(plan.modelProviders?.[0]?.authType).toBe(
+      AuthType.USE_OPENAI_RESPONSES,
+    );
+  });
+
   it('keeps custom ownership detection but merges installs by model identity', () => {
     expect(customProvider.ownsModel).toBeTypeOf('function');
     expect(

packages/core/src/providers/presets/custom-provider.ts

@@ -102,6 +102,7 @@ export const customProvider: ProviderConfig = {
   protocol: AuthType.USE_OPENAI,
   protocolOptions: [
     AuthType.USE_OPENAI,
+    AuthType.USE_OPENAI_RESPONSES,
     AuthType.USE_ANTHROPIC,
     AuthType.USE_GEMINI,
   ],

packages/core/src/telemetry/qwen-logger/qwen-logger.test.ts

@@ -169,6 +169,23 @@ describe('QwenLogger', () => {
       );
     });
 
+    it('includes the base URL for OpenAI Responses auth', async () => {
+      const config = makeFakeConfig({
+        getAuthType: () => AuthType.USE_OPENAI_RESPONSES,
+        getContentGeneratorConfig: () => ({
+          model: 'gpt-5',
+          baseUrl: 'https://api.example.com',
+        }),
+      });
+      const logger = QwenLogger.getInstance(config)!;
+
+      const payload = await (
+        logger as unknown as { createRumPayload(): Promise<RumPayload> }
+      ).createRumPayload();
+
+      expect(payload.properties?.['base_url']).toBe('https://api.example.com');
+    });
+
     it('includes source when source.json exists with valid source', async () => {
       // Note: Testing source information requires actual file system operations
       // This test verifies that the payload structure is correct

packages/core/src/telemetry/qwen-logger/qwen-logger.ts

@@ -285,7 +285,8 @@ export class QwenLogger {
         auth_type: authType,
         model: this.config?.getModel(),
         base_url:
-          authType === AuthType.USE_OPENAI
+          authType === AuthType.USE_OPENAI ||
+          authType === AuthType.USE_OPENAI_RESPONSES
             ? this.config?.getContentGeneratorConfig().baseUrl || ''
             : '',
         ...(this.config?.getChannel?.()