refactor(openai-shim): extract transport lifecycle (#2071)

* refactor(openai-shim): extract transport lifecycle * refactor(openai-shim): rebase transport extraction and address review Rebase onto current main and clean up the transport extraction follow-ups: drop stale facade imports left after the move and restore the full API timeout parser negative-case coverage in transport.test.ts. * test(openai-shim): address CodeRabbit review findings Use path.join in the architecture guard, require transport.ts in the mandatory extraction slice, strengthen Gemini stream conversion coverage, and add transport deadline/cancellation regression tests with fake timers. * test(openai-shim): assert manual signal cleanup after body cancel Exercise the combineRequestSignals fallback without AbortSignal.any so early body cancellation removes caller listeners and a later caller.abort does not abort the combined fetch signal. * test(openai-shim): restore AbortSignal.any when initially absent Delete the temporary AbortSignal.any override when the runtime did not define an own property, so transport and facade signal-cleanup tests leave global AbortSignal state unchanged for later cases.
2026-08-24 10:14:19 -05:00 · 2026-08-07 09:51:48 +08:00
parent 1bf8076d48
commit bac012f70b
7 changed files with 745 additions and 443 deletions
@@ -0,0 +1,52 @@
+import { existsSync, readFileSync, readdirSync } from 'node:fs'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { describe, expect, test } from 'bun:test'
+
+const facadePath = fileURLToPath(new URL('./openaiShim.ts', import.meta.url))
+const moduleDirectory = fileURLToPath(new URL('./openaiShim/', import.meta.url))
+
+const extractionDeltas = [
+  ['streamControl.ts', 169],
+  ['providerCompatibility.ts', 115],
+  ['ollamaAdapter.ts', 387],
+  ['messageConversion.ts', 474],
+  ['rawToolCallParsing.ts', 291],
+  ['xmlToolCallParsing.ts', 356],
+  ['streamConversion.ts', 1_072],
+  ['clientDispatch.ts', 182],
+  ['requestPlanner.ts', 304],
+  ['requestExecutor.ts', 704],
+  ['transport.ts', 361],
+  ['responseAdapters.ts', 189],
+  ['requestPreparation.ts', 247],
+  ['codexDispatch.ts', 109],
+] as const
+
+const upstreamExtractionCount = 11
+
+describe('openaiShim facade architecture', () => {
+  test('does not regain logic removed by the independent extractions', () => {
+    for (const [moduleName] of extractionDeltas.slice(0, upstreamExtractionCount)) {
+      expect(existsSync(join(moduleDirectory, moduleName))).toBe(true)
+    }
+    const activeReduction = extractionDeltas
+      .filter(([moduleName]) => existsSync(join(moduleDirectory, moduleName)))
+      .reduce(
+        (total, [, reduction]) => total + reduction,
+        0,
+      )
+    const facadeLines = readFileSync(facadePath, 'utf8').trimEnd().split('\n').length
+    expect(facadeLines).toBeLessThanOrEqual(5_636 - activeReduction)
+  })
+
+  test('keeps every extracted production module paired with its own test', () => {
+    const files = readdirSync(moduleDirectory)
+    const productionModules = files.filter(file =>
+      file.endsWith('.ts') && !file.endsWith('.test.ts'),
+    )
+    for (const moduleName of productionModules) {
+      expect(files).toContain(moduleName.replace(/\.ts$/, '.test.ts'))
+    }
+  })
+})
@@ -376,7 +376,6 @@ function importFreshOpenAIShim(

 type StreamIdleTestApi = {
  StreamIdleTimeoutError: new (timeoutMs: number) => Error
-  getApiTimeoutMs: () => number
  getStreamIdleTimeoutMs: () => number
  readWithIdleTimeout: (
    reader: ReadableStreamDefaultReader<Uint8Array>,
@@ -389,7 +388,6 @@ async function getStreamIdleTestApi(cacheKey: string): Promise<StreamIdleTestApi
  const mod = await importFreshOpenAIShim(cacheKey)
  const testApi = mod.__test as unknown as Partial<StreamIdleTestApi>
  expect(typeof testApi.StreamIdleTimeoutError).toBe('function')
-  expect(typeof testApi.getApiTimeoutMs).toBe('function')
  expect(typeof testApi.getStreamIdleTimeoutMs).toBe('function')
  expect(typeof testApi.readWithIdleTimeout).toBe('function')
  return testApi as StreamIdleTestApi
@@ -1374,27 +1372,6 @@ test('stream idle timeout env parser parses and bounds overrides', async () => {
 })
 // openaiShim test extraction seam 024 end

-test('API timeout env parser accepts safe positive integers and falls back otherwise', async () => {
-  const testApi = await getStreamIdleTestApi('api-timeout-env-parser')
-
-  delete process.env.API_TIMEOUT_MS
-  expect(testApi.getApiTimeoutMs()).toBe(600_000)
-
-  process.env.API_TIMEOUT_MS = '50'
-  expect(testApi.getApiTimeoutMs()).toBe(50)
-
-  process.env.API_TIMEOUT_MS = ' 50 '
-  expect(testApi.getApiTimeoutMs()).toBe(50)
-
-  process.env.API_TIMEOUT_MS = '3000000000'
-  expect(testApi.getApiTimeoutMs()).toBe(2_147_483_647)
-
-  for (const invalid of ['abc', '-5', '', '0', '1.5', '9007199254740993']) {
-    process.env.API_TIMEOUT_MS = invalid
-    expect(testApi.getApiTimeoutMs()).toBe(600_000)
-  }
-})
-
 // openaiShim test extraction seam 025 start: Anthropic-compatible passthrough stream rejects with idle timeout when it stalls
 test('Anthropic-compatible passthrough stream rejects with idle timeout when it stalls', async () => {
  process.env.CLAUDE_STREAM_IDLE_TIMEOUT_MS = '25'
@@ -4855,6 +4832,8 @@ test('manual signal fallback removes caller forwarding after the body settles',
  } finally {
    if (originalAbortSignalAny) {
      Object.defineProperty(AbortSignal, 'any', originalAbortSignalAny)
+    } else {
+      delete (AbortSignal as { any?: unknown }).any
    }
  }
 })
@@ -95,9 +95,15 @@ import {
 import { geminiSseToAnthropic as convertGeminiStream } from './openaiShim/geminiStreamConversion.js'
 import { compressToolHistory } from './compressToolHistory.js'
 import {
-  fetchWithProxyRetry,
-  type ProxyRetryFetcher,
-} from './fetchWithProxyRetry.js'
+  createClassifiedTransportError,
+  fetchWithHeadersDeadline,
+  getApiTimeoutMs,
+  preserveCallerAbortError,
+  redactUrlForDiagnostics,
+  redactUrlsInMessage,
+  ResponseHeadersTimeoutError,
+} from './openaiShim/transport.js'
+export { getApiTimeoutMs } from './openaiShim/transport.js'
 import { executeOpenAIRequest } from './openaiShim/requestExecutor.js'
 import {
  getLocalFastPathConfig,
@@ -121,15 +127,6 @@ import {
  markOpenAIRequestNonReplayable,
 } from './openaiErrorClassification.js'
 import { redactSecretValueForDisplay, type SecretValueSource } from '../../utils/providerProfile.js'
-import {
-  redactEncodedSecretSubstringsForDisplay,
-  redactSecretSubstringsForDisplay,
-} from '../../utils/providerSecrets.js'
-import {
-  redactUrlForDisplay,
-  shouldRedactUrlQueryParam,
-} from '../../utils/redaction.js'
-import { createCombinedAbortSignal } from '../../utils/combinedAbortSignal.js'
 import {
  normalizeToolArguments,
  hasToolFieldMapping,
@@ -202,9 +199,7 @@ const GITHUB_429_MAX_RETRIES = 3
 const GITHUB_429_BASE_DELAY_SEC = 1
 const GITHUB_429_MAX_DELAY_SEC = 32
 const CREDENTIAL_POOL_COOLDOWN_MS = 30_000
-const DEFAULT_API_TIMEOUT_MS = 600_000
 const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 90_000
-const MAX_STREAM_IDLE_TIMEOUT_MS = 2_147_483_647
 const GEMINI_API_HOST = 'generativelanguage.googleapis.com'
 const COPILOT_HEADERS: Record<string, string> = {
  'User-Agent': 'GitHubCopilotChat/0.26.7',
@@ -218,231 +213,6 @@ function isCopilotTokenExpiredError(text: string): boolean {
  return lower.includes('token expired') || lower.includes('token has expired')
 }

-class ResponseHeadersTimeoutError extends Error {
-  constructor(timeoutMs: number, url: string) {
-    super(
-      `OpenAI-compatible request received no response headers within ${timeoutMs}ms (API_TIMEOUT_MS) from ${url}`,
-    )
-    this.name = 'ResponseHeadersTimeoutError'
-  }
-}
-
-function preserveCallerAbortError(
-  error: unknown,
-  callerSignal: AbortSignal,
-): unknown {
-  return error instanceof ResponseHeadersTimeoutError || isAbortError(error)
-    ? callerSignal.reason ?? error
-    : error
-}
-
-function isAbortError(error: unknown): boolean {
-  return (
-    (typeof DOMException !== 'undefined' &&
-      error instanceof DOMException &&
-      error.name === 'AbortError') ||
-    (typeof error === 'object' &&
-      error !== null &&
-      'name' in error &&
-      error.name === 'AbortError')
-  )
-}
-
-export function getApiTimeoutMs(): number {
-  const raw = process.env.API_TIMEOUT_MS?.trim()
-  if (!raw || !/^\d+$/.test(raw)) return DEFAULT_API_TIMEOUT_MS
-  const parsed = Number(raw)
-  return Number.isSafeInteger(parsed) && parsed > 0
-    ? Math.min(parsed, MAX_STREAM_IDLE_TIMEOUT_MS)
-    : DEFAULT_API_TIMEOUT_MS
-}
-
-function combineRequestSignals(
-  callerSignal: AbortSignal | undefined,
-  deadlineSignal: AbortSignal,
-): {
-  signal: AbortSignal
-  cleanupAfterHeaders: () => void
-  cleanup: () => void
-  cleanupAfterBody?: () => void
-} {
-  if (!callerSignal) {
-    return {
-      signal: deadlineSignal,
-      cleanupAfterHeaders: () => {},
-      cleanup: () => {},
-    }
-  }
-
-  if (typeof AbortSignal.any === 'function') {
-    return {
-      // The deadline controller is request-local and its timer is the only
-      // abort source, so clearing that timer after headers permanently disarms it.
-      signal: AbortSignal.any([callerSignal, deadlineSignal]),
-      cleanupAfterHeaders: () => {},
-      cleanup: () => {},
-    }
-  }
-
-  const combined = new AbortController()
-  const abortFromCaller = () => {
-    deadlineSignal.removeEventListener('abort', abortFromDeadline)
-    combined.abort(callerSignal.reason)
-  }
-  const abortFromDeadline = () => {
-    callerSignal.removeEventListener('abort', abortFromCaller)
-    combined.abort(deadlineSignal.reason)
-  }
-  const cleanupAfterHeaders = () => {
-    deadlineSignal.removeEventListener('abort', abortFromDeadline)
-  }
-  const cleanup = () => {
-    callerSignal.removeEventListener('abort', abortFromCaller)
-    cleanupAfterHeaders()
-  }
-
-  callerSignal.addEventListener('abort', abortFromCaller, { once: true })
-  deadlineSignal.addEventListener('abort', abortFromDeadline, { once: true })
-  if (callerSignal.aborted) {
-    abortFromCaller()
-  } else if (deadlineSignal.aborted) {
-    abortFromDeadline()
-  }
-
-  return {
-    signal: combined.signal,
-    cleanupAfterHeaders,
-    cleanup,
-    cleanupAfterBody: cleanup,
-  }
-}
-
-function wrapResponseBodyWithCleanup(
-  response: Response,
-  cleanup: () => void,
-): Response {
-  if (!response.body) {
-    cleanup()
-    return response
-  }
-
-  const reader = response.body.getReader()
-  let cleanedUp = false
-  const cleanupOnce = () => {
-    if (cleanedUp) return
-    cleanedUp = true
-    cleanup()
-  }
-  const body = new ReadableStream<Uint8Array>({
-    async pull(controller) {
-      try {
-        const result = await reader.read()
-        if (result.done) {
-          cleanupOnce()
-          controller.close()
-        } else {
-          controller.enqueue(result.value)
-        }
-      } catch (error) {
-        cleanupOnce()
-        controller.error(error)
-      }
-    },
-    async cancel(reason) {
-      try {
-        await reader.cancel(reason)
-      } finally {
-        cleanupOnce()
-      }
-    },
-  })
-  const wrapped = new Response(body, {
-    status: response.status,
-    statusText: response.statusText,
-    headers: response.headers,
-  })
-  for (const property of ['url', 'type', 'redirected'] as const) {
-    try {
-      Object.defineProperty(wrapped, property, {
-        value: response[property],
-        configurable: true,
-      })
-    } catch {
-      /* non-fatal: standard response metadata remains available */
-    }
-  }
-  return wrapped
-}
-
-async function fetchWithHeadersDeadline(
-  url: string,
-  init: RequestInit,
-  options: {
-    callerSignal?: AbortSignal
-    timeoutMs: number
-  },
-): Promise<Response> {
-  const redactedUrl = redactUrlForDiagnostics(url)
-  const fetchWithAttemptDeadline: ProxyRetryFetcher = async (input, attemptInit) => {
-    const deadlineController = new AbortController()
-    const timeoutReason = new ResponseHeadersTimeoutError(
-      options.timeoutMs,
-      redactedUrl,
-    )
-    const {
-      signal,
-      cleanupAfterHeaders,
-      cleanup,
-      cleanupAfterBody,
-    } = combineRequestSignals(options.callerSignal, deadlineController.signal)
-    const timer = setTimeout(
-      () => deadlineController.abort(timeoutReason),
-      options.timeoutMs,
-    )
-    timer.unref?.()
-
-    let headersReceived = false
-    try {
-      const response = await fetch(input, { ...attemptInit, signal })
-      if (signal.aborted) {
-        void response.body?.cancel().catch(() => {})
-        throw (
-          signal.reason ??
-          new DOMException('The operation was aborted.', 'AbortError')
-        )
-      }
-      headersReceived = true
-      return cleanupAfterBody
-        ? wrapResponseBodyWithCleanup(response, cleanupAfterBody)
-        : response
-    } catch (error) {
-      if (options.callerSignal?.aborted) {
-        throw preserveCallerAbortError(error, options.callerSignal)
-      }
-      if (
-        deadlineController.signal.aborted &&
-        deadlineController.signal.reason === timeoutReason
-      ) {
-        throw timeoutReason
-      }
-      throw error
-    } finally {
-      clearTimeout(timer)
-      if (headersReceived) {
-        cleanupAfterHeaders()
-      } else {
-        cleanup()
-      }
-    }
-  }
-
-  return fetchWithProxyRetry(
-    url,
-    { ...init, signal: options.callerSignal },
-    { fetcher: fetchWithAttemptDeadline },
-  )
-}
-
 function hasGeminiApiHost(baseUrl: string | undefined): boolean {
  return matchesGeminiApiHost(baseUrl, GEMINI_API_HOST)
 }
@@ -464,150 +234,6 @@ function formatRetryAfterHint(response: Response): string {
  return ra ? ` (Retry-After: ${ra})` : ''
 }

-function decodeValidPercentRun(encoded: string): string {
-  const escapes = encoded.match(/%[0-9A-Fa-f]{2}/g)
-  if (!escapes) return encoded
-
-  let decoded = ''
-  let offset = 0
-  while (offset < escapes.length) {
-    const firstByte = Number.parseInt(escapes[offset].slice(1), 16)
-    const sequenceLength =
-      firstByte <= 0x7f
-        ? 1
-        : firstByte >= 0xc2 && firstByte <= 0xdf
-          ? 2
-          : firstByte >= 0xe0 && firstByte <= 0xef
-            ? 3
-            : firstByte >= 0xf0 && firstByte <= 0xf4
-              ? 4
-              : 1
-    try {
-      decoded += decodeURIComponent(
-        escapes.slice(offset, offset + sequenceLength).join(''),
-      )
-      offset += sequenceLength
-    } catch {
-      decoded += escapes[offset]
-      offset++
-    }
-  }
-  return decoded
-}
-
-function decodeValidUrlEscapesOnce(value: string): string {
-  return value.replace(/(?:%[0-9A-Fa-f]{2})+/g, decodeValidPercentRun)
-}
-
-const MAX_URL_SECRET_DECODING_LAYERS = 4
-
-function redactDecodedUrlComponentSecrets(value: string): string {
-  let decoded = value
-  let foundSecret = false
-  for (let layer = 0; layer <= MAX_URL_SECRET_DECODING_LAYERS; layer++) {
-    const redacted =
-      redactSecretSubstringsForDisplay(
-        decoded,
-        process.env as SecretValueSource,
-      ) ?? decoded
-    if (redacted !== decoded) foundSecret = true
-    if (layer === MAX_URL_SECRET_DECODING_LAYERS) {
-      decoded = redacted
-      break
-    }
-    const next = decodeValidUrlEscapesOnce(redacted)
-    if (next === redacted) {
-      decoded = redacted
-      break
-    }
-    decoded = next
-  }
-  return foundSecret ? decoded : value
-}
-
-function redactUrlForDiagnostics(url: string): string {
-  let redacted = redactUrlForDisplay(url)
-  try {
-    const parsed = new URL(redacted)
-    const redactedPathname = redactDecodedUrlComponentSecrets(parsed.pathname)
-    const redactedSearch = redactDecodedUrlComponentSecrets(parsed.search)
-    let componentRedacted = false
-    if (redactedPathname !== parsed.pathname) {
-      parsed.pathname = redactedPathname
-      componentRedacted = true
-    }
-    if (redactedSearch !== parsed.search) {
-      parsed.search = redactedSearch
-      componentRedacted = true
-    }
-    if (componentRedacted) redacted = parsed.toString()
-  } catch {
-    // Keep the URL-level redaction when the URL cannot be parsed.
-  }
-  const redactedSubstrings =
-    redactSecretSubstringsForDisplay(
-      redacted,
-      process.env as SecretValueSource,
-    ) ?? redacted
-  return (
-    redactSecretValueForDisplay(
-      redactedSubstrings,
-      process.env as SecretValueSource,
-    ) ?? redactedSubstrings
-  )
-}
-
-function redactUrlsInMessage(message: string): string {
-  return message.replace(/https?:\/\/\S+/g, match => redactUrlForDiagnostics(match))
-}
-
-function createClassifiedTransportError(
-  error: unknown,
-  requestUrl: string,
-  model: string,
-  preclassifiedFailure?: ReturnType<typeof classifyOpenAINetworkFailure>,
-) {
-  const failure =
-    preclassifiedFailure ??
-    classifyOpenAINetworkFailure(error, {
-      url: requestUrl,
-    })
-  const redactedUrl = redactUrlForDiagnostics(requestUrl)
-  const encodedSecretRedactedMessage =
-    redactEncodedSecretSubstringsForDisplay(
-      redactUrlsInMessage(failure.message),
-      process.env as SecretValueSource,
-    ) ?? 'Request failed'
-  const redactedMessage =
-    redactSecretSubstringsForDisplay(
-      encodedSecretRedactedMessage,
-      process.env as SecretValueSource,
-    ) ?? 'Request failed'
-  const safeMessage =
-    redactSecretValueForDisplay(
-      redactedMessage,
-      process.env as SecretValueSource,
-    ) || 'Request failed'
-
-  logForDebugging(
-    `[OpenAIShim] transport failure category=${failure.category} retryable=${failure.retryable} code=${failure.code ?? 'unknown'} method=POST url=${redactedUrl} model=${model} message=${safeMessage}`,
-    { level: 'warn' },
-  )
-
-  const apiError = APIError.generate(
-    0,
-    undefined,
-    buildOpenAICompatibilityErrorMessage(
-      `OpenAI API transport error: ${safeMessage}${failure.code ? ` (code=${failure.code})` : ''}`,
-      failure,
-    ),
-    new Headers(),
-  )
-  return failure.retryable
-    ? apiError
-    : markOpenAIRequestNonReplayable(apiError)
-}
-
 function sleepMs(ms: number): Promise<void> {
  return new Promise(resolve => setTimeout(resolve, ms))
 }
@@ -1,37 +0,0 @@
-import { existsSync, readFileSync } from 'node:fs'
-import { fileURLToPath } from 'node:url'
-import { describe, expect, test } from 'bun:test'
-
-const facadePath = fileURLToPath(new URL('../openaiShim.ts', import.meta.url))
-const moduleDirectory = fileURLToPath(new URL('.', import.meta.url))
-
-// The rebased shared extraction seam is 5,636 lines. The ten representative
-// extractions below remove 4,054 net lines, yielding the verified 1,582-line
-// fully extracted façade.
-const extractionDeltas = [
-  ['streamControl.ts', 169],
-  ['providerCompatibility.ts', 115],
-  ['ollamaAdapter.ts', 387],
-  ['messageConversion.ts', 474],
-  ['rawToolCallParsing.ts', 291],
-  ['xmlToolCallParsing.ts', 356],
-  ['streamConversion.ts', 1_072],
-  ['clientDispatch.ts', 182],
-  ['requestPlanner.ts', 304],
-  ['requestExecutor.ts', 704],
-] as const
-
-describe('openaiShim façade architecture', () => {
-  test('does not regain logic removed by the independent extractions', () => {
-    for (const [moduleName] of extractionDeltas) {
-      expect(existsSync(`${moduleDirectory}/${moduleName}`)).toBe(true)
-    }
-    const activeReduction = extractionDeltas.reduce(
-      (total, [, reduction]) => total + reduction,
-      0,
-    )
-    const facadeLines = readFileSync(facadePath, 'utf8').trimEnd().split('\n').length
-
-    expect(facadeLines).toBeLessThanOrEqual(5_636 - activeReduction)
-  })
-})
@@ -0,0 +1,144 @@
+import { expect, test } from 'bun:test'
+import {
+  createReaderCanceller,
+  createStreamAbortError,
+  readWithIdleTimeout,
+  throwIfStreamAborted,
+} from './streamControl.js'
+import { geminiSseToAnthropic } from './geminiStreamConversion.js'
+
+const dependencies = {
+  createReaderCanceller,
+  createStreamAbortError,
+  getStreamIdleTimeoutMs: () => 1_000,
+  makeMessageId: () => 'msg_gemini_test',
+  readWithIdleTimeout,
+  throwIfStreamAborted,
+}
+
+function responseFor(...payloads: Array<Record<string, unknown>>): Response {
+  const frames = [
+    ...payloads.map(payload => `data: ${JSON.stringify(payload)}\n\n`),
+    'data: [DONE]\n\n',
+  ].join('')
+  return new Response(new TextEncoder().encode(frames), {
+    headers: { 'Content-Type': 'text/event-stream' },
+  })
+}
+
+async function collectEvents(
+  response: Response,
+  signal?: AbortSignal,
+): Promise<Array<Record<string, unknown>>> {
+  const events: Array<Record<string, unknown>> = []
+  for await (const event of geminiSseToAnthropic(
+    response,
+    'gemini-test',
+    signal,
+    dependencies,
+  )) {
+    events.push(event as unknown as Record<string, unknown>)
+  }
+  return events
+}
+
+test('converts Gemini text, tool calls, usage, and finish state', async () => {
+  const events = await collectEvents(responseFor({
+    candidates: [{
+      content: {
+        parts: [
+          { text: 'Inspecting.' },
+          { functionCall: { name: 'Read', args: { file_path: 'a.ts' } } },
+        ],
+      },
+      finishReason: 'STOP',
+    }],
+    usageMetadata: {
+      promptTokenCount: 4,
+      candidatesTokenCount: 2,
+      thoughtsTokenCount: 1,
+    },
+  }))
+
+  expect(events[0]).toMatchObject({
+    type: 'message_start',
+    message: { id: 'msg_gemini_test', model: 'gemini-test' },
+  })
+  expect(events.some(event =>
+    event.type === 'content_block_delta' &&
+    (event.delta as { text?: string })?.text === 'Inspecting.',
+  )).toBe(true)
+
+  const toolStartIndex = events.findIndex(event =>
+    event.type === 'content_block_start' &&
+    (event.content_block as { name?: string })?.name === 'Read',
+  )
+  expect(toolStartIndex).toBeGreaterThan(-1)
+  expect(events[toolStartIndex + 1]).toMatchObject({
+    type: 'content_block_delta',
+    delta: {
+      type: 'input_json_delta',
+      partial_json: '{"file_path":"a.ts"}',
+    },
+  })
+  expect(events[toolStartIndex + 2]).toEqual({
+    type: 'content_block_stop',
+    index: (events[toolStartIndex] as { index: number }).index,
+  })
+  expect(events.at(-2)).toMatchObject({
+    type: 'message_delta',
+    delta: { stop_reason: 'tool_use' },
+    usage: { input_tokens: 4, output_tokens: 3 },
+  })
+  expect(events.at(-1)).toEqual({ type: 'message_stop' })
+})
+
+test('maps STOP to end_turn when no tool call is present', async () => {
+  const events = await collectEvents(responseFor({
+    candidates: [{
+      content: { parts: [{ text: 'Done.' }] },
+      finishReason: 'STOP',
+    }],
+  }))
+
+  expect(events.at(-2)).toMatchObject({
+    type: 'message_delta',
+    delta: { stop_reason: 'end_turn' },
+  })
+  expect(events.at(-1)).toEqual({ type: 'message_stop' })
+})
+
+test('maps MAX_TOKENS to max_tokens when no tool call is present', async () => {
+  const events = await collectEvents(responseFor({
+    candidates: [{
+      content: { parts: [{ text: 'Truncated.' }] },
+      finishReason: 'MAX_TOKENS',
+    }],
+  }))
+
+  expect(events.at(-2)).toMatchObject({
+    type: 'message_delta',
+    delta: { stop_reason: 'max_tokens' },
+  })
+  expect(events.at(-1)).toEqual({ type: 'message_stop' })
+})
+
+test('rejects an already-aborted Gemini stream without yielding events', async () => {
+  const cancelReasons: unknown[] = []
+  const response = new Response(new ReadableStream<Uint8Array>({
+    cancel(reason) {
+      cancelReasons.push(reason)
+    },
+  }), { headers: { 'Content-Type': 'text/event-stream' } })
+  const controller = new AbortController()
+  controller.abort()
+  const stream = geminiSseToAnthropic(
+    response,
+    'gemini-test',
+    controller.signal,
+    dependencies,
+  )
+
+  await expect(stream.next()).rejects.toMatchObject({ name: 'AbortError' })
+  expect(cancelReasons).toHaveLength(1)
+})
@@ -0,0 +1,155 @@
+import { afterEach, expect, jest, test } from 'bun:test'
+import {
+  fetchWithHeadersDeadline,
+  getApiTimeoutMs,
+  redactUrlForDiagnostics,
+  ResponseHeadersTimeoutError,
+} from './transport.js'
+
+const originalFetch = globalThis.fetch
+const originalApiTimeoutMs = process.env.API_TIMEOUT_MS
+const originalApiKey = process.env.OPENAI_API_KEY
+
+afterEach(() => {
+  globalThis.fetch = originalFetch
+  if (originalApiTimeoutMs === undefined) delete process.env.API_TIMEOUT_MS
+  else process.env.API_TIMEOUT_MS = originalApiTimeoutMs
+  if (originalApiKey === undefined) delete process.env.OPENAI_API_KEY
+  else process.env.OPENAI_API_KEY = originalApiKey
+  jest.useRealTimers()
+})
+
+test('API timeout parser accepts bounded positive integers', () => {
+  delete process.env.API_TIMEOUT_MS
+  expect(getApiTimeoutMs()).toBe(600_000)
+  process.env.API_TIMEOUT_MS = '25'
+  expect(getApiTimeoutMs()).toBe(25)
+  process.env.API_TIMEOUT_MS = ' 25 '
+  expect(getApiTimeoutMs()).toBe(25)
+  process.env.API_TIMEOUT_MS = '3000000000'
+  expect(getApiTimeoutMs()).toBe(2_147_483_647)
+
+  for (const invalid of ['abc', '', '1.5', '9007199254740993', '25ms', '0', '-5']) {
+    process.env.API_TIMEOUT_MS = invalid
+    expect(getApiTimeoutMs()).toBe(600_000)
+  }
+})
+
+test('API timeout parser can inspect a request-local environment', () => {
+  process.env.API_TIMEOUT_MS = '10'
+  expect(getApiTimeoutMs({ API_TIMEOUT_MS: '42' })).toBe(42)
+  expect(getApiTimeoutMs({})).toBe(600_000)
+})
+
+test('redacts configured and encoded secrets from diagnostic URLs', () => {
+  process.env.OPENAI_API_KEY = 'secret/value'
+  const diagnostic = redactUrlForDiagnostics(
+    'https://example.test/secret%252Fvalue?token=secret%2Fvalue',
+  )
+
+  expect(diagnostic).not.toContain('secret')
+  expect(diagnostic).not.toContain('value')
+  expect(diagnostic.toLowerCase()).toContain('redact')
+})
+
+test('rejects a request that receives no headers before its deadline', async () => {
+  globalThis.fetch = ((_input, init) => new Promise<Response>((_resolve, reject) => {
+    const signal = init?.signal
+    signal?.addEventListener('abort', () => reject(signal.reason), { once: true })
+  })) as typeof globalThis.fetch
+
+  await expect(fetchWithHeadersDeadline(
+    'https://example.test/v1/chat/completions',
+    {},
+    { timeoutMs: 20 },
+  )).rejects.toBeInstanceOf(ResponseHeadersTimeoutError)
+})
+
+test('preserves caller cancellation instead of reporting a deadline', async () => {
+  const caller = new AbortController()
+  globalThis.fetch = ((_input, init) => new Promise<Response>((_resolve, reject) => {
+    const signal = init?.signal
+    signal?.addEventListener('abort', () => {
+      reject(new DOMException('Aborted', 'AbortError'))
+    }, { once: true })
+  })) as typeof globalThis.fetch
+
+  const pending = fetchWithHeadersDeadline(
+    'https://example.test/v1/chat/completions',
+    {},
+    { callerSignal: caller.signal, timeoutMs: 60_000 },
+  )
+  caller.abort(new DOMException('Caller stopped', 'AbortError'))
+
+  await expect(pending).rejects.toBe(caller.signal.reason)
+})
+
+test('disarms the deadline once response headers arrive', async () => {
+  jest.useFakeTimers()
+  let controller: ReadableStreamDefaultController<Uint8Array> | undefined
+  let capturedSignal: AbortSignal | undefined
+  globalThis.fetch = (async (_input, init) => {
+    capturedSignal = init?.signal
+    return new Response(new ReadableStream<Uint8Array>({
+      start(value) {
+        controller = value
+      },
+    }))
+  }) as typeof globalThis.fetch
+
+  const responsePromise = fetchWithHeadersDeadline(
+    'https://example.test/v1/chat/completions',
+    {},
+    { timeoutMs: 20 },
+  )
+  await Promise.resolve()
+  jest.advanceTimersByTime(30)
+  expect(capturedSignal?.aborted).toBe(false)
+
+  controller?.enqueue(new TextEncoder().encode('ok'))
+  controller?.close()
+  const response = await responsePromise
+  expect(await response.text()).toBe('ok')
+})
+
+test('cleans up the caller signal after early body cancellation', async () => {
+  jest.useFakeTimers()
+  const originalAbortSignalAny = Object.getOwnPropertyDescriptor(
+    AbortSignal,
+    'any',
+  )
+  Object.defineProperty(AbortSignal, 'any', {
+    value: undefined,
+    configurable: true,
+  })
+  try {
+    let capturedSignal: AbortSignal | undefined
+    const caller = new AbortController()
+    globalThis.fetch = (async (_input, init) => {
+      capturedSignal = init?.signal
+      return new Response(new ReadableStream<Uint8Array>({
+        start(value) {
+          value.enqueue(new TextEncoder().encode('partial'))
+        },
+      }))
+    }) as typeof globalThis.fetch
+
+    const response = await fetchWithHeadersDeadline(
+      'https://example.test/v1/chat/completions',
+      {},
+      { callerSignal: caller.signal, timeoutMs: 20 },
+    )
+    await response.body?.cancel()
+    jest.advanceTimersByTime(30)
+    expect(capturedSignal?.aborted).toBe(false)
+
+    caller.abort(new DOMException('Caller stopped', 'AbortError'))
+    expect(capturedSignal?.aborted).toBe(false)
+  } finally {
+    if (originalAbortSignalAny) {
+      Object.defineProperty(AbortSignal, 'any', originalAbortSignalAny)
+    } else {
+      delete (AbortSignal as { any?: unknown }).any
+    }
+  }
+})
@@ -0,0 +1,383 @@
+import { APIError } from '@anthropic-ai/sdk'
+import { logForDebugging } from '../../../utils/debug.js'
+import {
+  redactSecretValueForDisplay,
+  type SecretValueSource,
+} from '../../../utils/providerProfile.js'
+import {
+  redactEncodedSecretSubstringsForDisplay,
+  redactSecretSubstringsForDisplay,
+} from '../../../utils/providerSecrets.js'
+import { redactUrlForDisplay } from '../../../utils/redaction.js'
+import {
+  buildOpenAICompatibilityErrorMessage,
+  classifyOpenAINetworkFailure,
+  markOpenAIRequestNonReplayable,
+} from '../openaiErrorClassification.js'
+import {
+  fetchWithProxyRetry,
+  type ProxyRetryFetcher,
+} from '../fetchWithProxyRetry.js'
+
+const DEFAULT_API_TIMEOUT_MS = 600_000
+const MAX_API_TIMEOUT_MS = 2_147_483_647
+const MAX_URL_SECRET_DECODING_LAYERS = 4
+
+export class ResponseHeadersTimeoutError extends Error {
+  constructor(timeoutMs: number, url: string) {
+    super(
+      `OpenAI-compatible request received no response headers within ${timeoutMs}ms (API_TIMEOUT_MS) from ${url}`,
+    )
+    this.name = 'ResponseHeadersTimeoutError'
+  }
+}
+
+function isAbortError(error: unknown): boolean {
+  return (
+    (typeof DOMException !== 'undefined' &&
+      error instanceof DOMException &&
+      error.name === 'AbortError') ||
+    (typeof error === 'object' &&
+      error !== null &&
+      'name' in error &&
+      error.name === 'AbortError')
+  )
+}
+
+export function preserveCallerAbortError(
+  error: unknown,
+  callerSignal: AbortSignal,
+): unknown {
+  return error instanceof ResponseHeadersTimeoutError || isAbortError(error)
+    ? callerSignal.reason ?? error
+    : error
+}
+
+export function getApiTimeoutMs(
+  processEnv: NodeJS.ProcessEnv = process.env,
+): number {
+  const raw = processEnv.API_TIMEOUT_MS?.trim()
+  if (!raw || !/^\d+$/.test(raw)) return DEFAULT_API_TIMEOUT_MS
+  const parsed = Number(raw)
+  return Number.isSafeInteger(parsed) && parsed > 0
+    ? Math.min(parsed, MAX_API_TIMEOUT_MS)
+    : DEFAULT_API_TIMEOUT_MS
+}
+
+function combineRequestSignals(
+  callerSignal: AbortSignal | undefined,
+  deadlineSignal: AbortSignal,
+): {
+  signal: AbortSignal
+  cleanupAfterHeaders: () => void
+  cleanup: () => void
+  cleanupAfterBody?: () => void
+} {
+  if (!callerSignal) {
+    return {
+      signal: deadlineSignal,
+      cleanupAfterHeaders: () => {},
+      cleanup: () => {},
+    }
+  }
+
+  if (typeof AbortSignal.any === 'function') {
+    return {
+      signal: AbortSignal.any([callerSignal, deadlineSignal]),
+      cleanupAfterHeaders: () => {},
+      cleanup: () => {},
+    }
+  }
+
+  const combined = new AbortController()
+  const abortFromCaller = () => {
+    deadlineSignal.removeEventListener('abort', abortFromDeadline)
+    combined.abort(callerSignal.reason)
+  }
+  const abortFromDeadline = () => {
+    callerSignal.removeEventListener('abort', abortFromCaller)
+    combined.abort(deadlineSignal.reason)
+  }
+  const cleanupAfterHeaders = () => {
+    deadlineSignal.removeEventListener('abort', abortFromDeadline)
+  }
+  const cleanup = () => {
+    callerSignal.removeEventListener('abort', abortFromCaller)
+    cleanupAfterHeaders()
+  }
+
+  callerSignal.addEventListener('abort', abortFromCaller, { once: true })
+  deadlineSignal.addEventListener('abort', abortFromDeadline, { once: true })
+  if (callerSignal.aborted) abortFromCaller()
+  else if (deadlineSignal.aborted) abortFromDeadline()
+
+  return {
+    signal: combined.signal,
+    cleanupAfterHeaders,
+    cleanup,
+    cleanupAfterBody: cleanup,
+  }
+}
+
+function wrapResponseBodyWithCleanup(
+  response: Response,
+  cleanup: () => void,
+): Response {
+  if (!response.body) {
+    cleanup()
+    return response
+  }
+
+  const reader = response.body.getReader()
+  let cleanedUp = false
+  const cleanupOnce = () => {
+    if (cleanedUp) return
+    cleanedUp = true
+    cleanup()
+  }
+  const body = new ReadableStream<Uint8Array>({
+    async pull(controller) {
+      try {
+        const result = await reader.read()
+        if (result.done) {
+          cleanupOnce()
+          controller.close()
+        } else {
+          controller.enqueue(result.value)
+        }
+      } catch (error) {
+        cleanupOnce()
+        controller.error(error)
+      }
+    },
+    async cancel(reason) {
+      try {
+        await reader.cancel(reason)
+      } finally {
+        cleanupOnce()
+      }
+    },
+  })
+  const wrapped = new Response(body, {
+    status: response.status,
+    statusText: response.statusText,
+    headers: response.headers,
+  })
+  for (const property of ['url', 'type', 'redirected'] as const) {
+    try {
+      Object.defineProperty(wrapped, property, {
+        value: response[property],
+        configurable: true,
+      })
+    } catch {
+      // Standard response metadata remains available when this is unsupported.
+    }
+  }
+  return wrapped
+}
+
+export async function fetchWithHeadersDeadline(
+  url: string,
+  init: RequestInit,
+  options: {
+    callerSignal?: AbortSignal
+    timeoutMs: number
+  },
+): Promise<Response> {
+  const redactedUrl = redactUrlForDiagnostics(url)
+  const fetchWithAttemptDeadline: ProxyRetryFetcher = async (input, attemptInit) => {
+    const deadlineController = new AbortController()
+    const timeoutReason = new ResponseHeadersTimeoutError(
+      options.timeoutMs,
+      redactedUrl,
+    )
+    const {
+      signal,
+      cleanupAfterHeaders,
+      cleanup,
+      cleanupAfterBody,
+    } = combineRequestSignals(options.callerSignal, deadlineController.signal)
+    const timer = setTimeout(
+      () => deadlineController.abort(timeoutReason),
+      options.timeoutMs,
+    )
+    timer.unref?.()
+
+    let headersReceived = false
+    try {
+      const response = await fetch(input, { ...attemptInit, signal })
+      if (signal.aborted) {
+        void response.body?.cancel().catch(() => {})
+        throw (
+          signal.reason ??
+          new DOMException('The operation was aborted.', 'AbortError')
+        )
+      }
+      headersReceived = true
+      return cleanupAfterBody
+        ? wrapResponseBodyWithCleanup(response, cleanupAfterBody)
+        : response
+    } catch (error) {
+      if (options.callerSignal?.aborted) {
+        throw preserveCallerAbortError(error, options.callerSignal)
+      }
+      if (
+        deadlineController.signal.aborted &&
+        deadlineController.signal.reason === timeoutReason
+      ) {
+        throw timeoutReason
+      }
+      throw error
+    } finally {
+      clearTimeout(timer)
+      if (headersReceived) cleanupAfterHeaders()
+      else cleanup()
+    }
+  }
+
+  return fetchWithProxyRetry(
+    url,
+    { ...init, signal: options.callerSignal },
+    { fetcher: fetchWithAttemptDeadline },
+  )
+}
+
+function decodeValidPercentRun(encoded: string): string {
+  const escapes = encoded.match(/%[0-9A-Fa-f]{2}/g)
+  if (!escapes) return encoded
+
+  let decoded = ''
+  let offset = 0
+  while (offset < escapes.length) {
+    const firstByte = Number.parseInt(escapes[offset].slice(1), 16)
+    const sequenceLength =
+      firstByte <= 0x7f
+        ? 1
+        : firstByte >= 0xc2 && firstByte <= 0xdf
+          ? 2
+          : firstByte >= 0xe0 && firstByte <= 0xef
+            ? 3
+            : firstByte >= 0xf0 && firstByte <= 0xf4
+              ? 4
+              : 1
+    try {
+      decoded += decodeURIComponent(
+        escapes.slice(offset, offset + sequenceLength).join(''),
+      )
+      offset += sequenceLength
+    } catch {
+      decoded += escapes[offset]
+      offset++
+    }
+  }
+  return decoded
+}
+
+function decodeValidUrlEscapesOnce(value: string): string {
+  return value.replace(/(?:%[0-9A-Fa-f]{2})+/g, decodeValidPercentRun)
+}
+
+function redactDecodedUrlComponentSecrets(value: string): string {
+  let decoded = value
+  let foundSecret = false
+  for (let layer = 0; layer <= MAX_URL_SECRET_DECODING_LAYERS; layer++) {
+    const redacted =
+      redactSecretSubstringsForDisplay(
+        decoded,
+        process.env as SecretValueSource,
+      ) ?? decoded
+    if (redacted !== decoded) foundSecret = true
+    if (layer === MAX_URL_SECRET_DECODING_LAYERS) {
+      decoded = redacted
+      break
+    }
+    const next = decodeValidUrlEscapesOnce(redacted)
+    if (next === redacted) {
+      decoded = redacted
+      break
+    }
+    decoded = next
+  }
+  return foundSecret ? decoded : value
+}
+
+export function redactUrlForDiagnostics(url: string): string {
+  let redacted = redactUrlForDisplay(url)
+  try {
+    const parsed = new URL(redacted)
+    const redactedPathname = redactDecodedUrlComponentSecrets(parsed.pathname)
+    const redactedSearch = redactDecodedUrlComponentSecrets(parsed.search)
+    let componentRedacted = false
+    if (redactedPathname !== parsed.pathname) {
+      parsed.pathname = redactedPathname
+      componentRedacted = true
+    }
+    if (redactedSearch !== parsed.search) {
+      parsed.search = redactedSearch
+      componentRedacted = true
+    }
+    if (componentRedacted) redacted = parsed.toString()
+  } catch {
+    // Keep the URL-level redaction when the URL cannot be parsed.
+  }
+  const redactedSubstrings =
+    redactSecretSubstringsForDisplay(
+      redacted,
+      process.env as SecretValueSource,
+    ) ?? redacted
+  return (
+    redactSecretValueForDisplay(
+      redactedSubstrings,
+      process.env as SecretValueSource,
+    ) ?? redactedSubstrings
+  )
+}
+
+export function redactUrlsInMessage(message: string): string {
+  return message.replace(/https?:\/\/\S+/g, match => redactUrlForDiagnostics(match))
+}
+
+export function createClassifiedTransportError(
+  error: unknown,
+  requestUrl: string,
+  model: string,
+  preclassifiedFailure?: ReturnType<typeof classifyOpenAINetworkFailure>,
+) {
+  const failure =
+    preclassifiedFailure ??
+    classifyOpenAINetworkFailure(error, { url: requestUrl })
+  const redactedUrl = redactUrlForDiagnostics(requestUrl)
+  const encodedSecretRedactedMessage =
+    redactEncodedSecretSubstringsForDisplay(
+      redactUrlsInMessage(failure.message),
+      process.env as SecretValueSource,
+    ) ?? 'Request failed'
+  const redactedMessage =
+    redactSecretSubstringsForDisplay(
+      encodedSecretRedactedMessage,
+      process.env as SecretValueSource,
+    ) ?? 'Request failed'
+  const safeMessage =
+    redactSecretValueForDisplay(
+      redactedMessage,
+      process.env as SecretValueSource,
+    ) || 'Request failed'
+
+  logForDebugging(
+    `[OpenAIShim] transport failure category=${failure.category} retryable=${failure.retryable} code=${failure.code ?? 'unknown'} method=POST url=${redactedUrl} model=${model} message=${safeMessage}`,
+    { level: 'warn' },
+  )
+
+  const apiError = APIError.generate(
+    0,
+    undefined,
+    buildOpenAICompatibilityErrorMessage(
+      `OpenAI API transport error: ${safeMessage}${failure.code ? ` (code=${failure.code})` : ''}`,
+      failure,
+    ),
+    new Headers(),
+  )
+  return failure.retryable
+    ? apiError
+    : markOpenAIRequestNonReplayable(apiError)
+}