diff --git a/src/services/api/openaiShim.architecture.test.ts b/src/services/api/openaiShim.architecture.test.ts new file mode 100644 index 000000000..fd4aa9e9d --- /dev/null +++ b/src/services/api/openaiShim.architecture.test.ts @@ -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')) + } + }) +}) diff --git a/src/services/api/openaiShim.test.ts b/src/services/api/openaiShim.test.ts index 27718f87b..1e1db35c3 100644 --- a/src/services/api/openaiShim.test.ts +++ b/src/services/api/openaiShim.test.ts @@ -376,7 +376,6 @@ function importFreshOpenAIShim( type StreamIdleTestApi = { StreamIdleTimeoutError: new (timeoutMs: number) => Error - getApiTimeoutMs: () => number getStreamIdleTimeoutMs: () => number readWithIdleTimeout: ( reader: ReadableStreamDefaultReader, @@ -389,7 +388,6 @@ async function getStreamIdleTestApi(cacheKey: string): Promise 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 } } }) diff --git a/src/services/api/openaiShim.ts b/src/services/api/openaiShim.ts index d1c165b30..d64b84b58 100644 --- a/src/services/api/openaiShim.ts +++ b/src/services/api/openaiShim.ts @@ -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 = { '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({ - 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 { - 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, -) { - 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 { return new Promise(resolve => setTimeout(resolve, ms)) } diff --git a/src/services/api/openaiShim/architecture.test.ts b/src/services/api/openaiShim/architecture.test.ts deleted file mode 100644 index e4decc59d..000000000 --- a/src/services/api/openaiShim/architecture.test.ts +++ /dev/null @@ -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) - }) -}) diff --git a/src/services/api/openaiShim/geminiStreamConversion.test.ts b/src/services/api/openaiShim/geminiStreamConversion.test.ts new file mode 100644 index 000000000..78b56d94d --- /dev/null +++ b/src/services/api/openaiShim/geminiStreamConversion.test.ts @@ -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>): 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>> { + const events: Array> = [] + for await (const event of geminiSseToAnthropic( + response, + 'gemini-test', + signal, + dependencies, + )) { + events.push(event as unknown as Record) + } + 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({ + 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) +}) diff --git a/src/services/api/openaiShim/transport.test.ts b/src/services/api/openaiShim/transport.test.ts new file mode 100644 index 000000000..7c5c79ffe --- /dev/null +++ b/src/services/api/openaiShim/transport.test.ts @@ -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((_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((_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 | undefined + let capturedSignal: AbortSignal | undefined + globalThis.fetch = (async (_input, init) => { + capturedSignal = init?.signal + return new Response(new ReadableStream({ + 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({ + 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 + } + } +}) diff --git a/src/services/api/openaiShim/transport.ts b/src/services/api/openaiShim/transport.ts new file mode 100644 index 000000000..44ab70025 --- /dev/null +++ b/src/services/api/openaiShim/transport.ts @@ -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({ + 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 { + 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, +) { + 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) +}