diff --git a/.eslintignore b/.eslintignore index 40b211be..3781d7d7 100644 --- a/.eslintignore +++ b/.eslintignore @@ -8,6 +8,7 @@ node_modules/ # Browser binary camoufox/ +camoufox-macos/ # Temporary files tmp/ diff --git a/.gitignore b/.gitignore index 71eac191..20170499 100644 --- a/.gitignore +++ b/.gitignore @@ -35,7 +35,6 @@ tmp/ temp/ ui/dist/ *.css -test/ debug_*.png debug_*.html proxylist.txt diff --git a/.prettierignore b/.prettierignore index 21559cbc..997905f1 100644 --- a/.prettierignore +++ b/.prettierignore @@ -12,6 +12,7 @@ build/ # Camoufox directory camoufox/ +camoufox-macos/ # Logs *.log @@ -32,3 +33,8 @@ Thumbs.db # Temporary files *.tmp *.temp + +# Local tooling and rejected configs +.npm-cache/ +.serena/ +*.REJECTED.yaml diff --git a/README.md b/README.md index 473beea2..4010c67e 100644 --- a/README.md +++ b/README.md @@ -267,7 +267,7 @@ services: | `ENABLE_AUTH_UPDATE` | 是否启用自动保存凭证更新。默认为启用状态，将在每次登录/切换账号成功时以及每 24 小时自动更新 auth 文件。设为 `false` 禁用。 | `true` | | `MAX_RETRIES` | 请求失败后的最大重试次数（仅对假流式和非流式生效）。 | `3` | | `RETRY_DELAY` | 两次重试之间的间隔（毫秒）。 | `2000` | -| `STREAM_TIMEOUT_MS` | 真流式响应相邻数据块之间的超时时间（毫秒），最大 `300000`。 | `60000` | +| `STREAM_TIMEOUT_MS` | 真流式响应相邻数据块之间的超时时间（毫秒），默认值为 `0`（禁用超时），设为正值启用，最大 `300000`。 | `0` | | `FAKE_STREAM_TIMEOUT_MS` | 假流式/非流式缓冲响应的超时时间（毫秒），最大 `300000`。 | `300000` | | `SWITCH_ON_USES` | 自动切换帐户前允许的请求次数（设为 `0` 禁用）。 | `40` | | `FAILURE_THRESHOLD` | 切换帐户前允许的连续失败次数（设为 `0` 禁用）。 | `3` | diff --git a/README_EN.md b/README_EN.md index 9aa299f4..fd9045ad 100644 --- a/README_EN.md +++ b/README_EN.md @@ -265,7 +265,7 @@ Usage: | `ENABLE_AUTH_UPDATE` | Whether to enable automatic auth credential updates. Defaults to enabled. The auth file will be automatically updated upon successful login/account switch and every 24 hours. Set to `false` to disable. | `true` | | `MAX_RETRIES` | Maximum number of retries for failed requests (only effective for fake streaming and non-streaming). | `3` | | `RETRY_DELAY` | Delay between retries in milliseconds. | `2000` | -| `STREAM_TIMEOUT_MS` | Timeout between real streaming chunks, in milliseconds. Maximum: `300000`. | `60000` | +| `STREAM_TIMEOUT_MS` | Timeout between real streaming chunks, in milliseconds. Default is `0` (disabled); set a positive value to enable. Maximum: `300000`. | `0` | | `FAKE_STREAM_TIMEOUT_MS` | Timeout for fake streaming / non-streaming buffered responses, in milliseconds. Maximum: `300000`. | `300000` | | `SWITCH_ON_USES` | Number of requests before automatically switching accounts (`0` to disable). | `40` | | `FAILURE_THRESHOLD` | Number of consecutive failures before switching accounts (`0` to disable). | `3` | diff --git a/package.json b/package.json index d1772eca..10a1066f 100644 --- a/package.json +++ b/package.json @@ -25,7 +25,8 @@ "lint:js:fix": "eslint . --fix", "lint:css:fix": "stylelint \"ui/**/*.{css,less}\" --fix", "format": "prettier --write .", - "format:check": "prettier --check ." + "format:check": "prettier --check .", + "test": "node --test" }, "dependencies": { "archiver": "^7.0.1", diff --git a/scripts/client/index.html b/scripts/client/index.html index bcebc8e3..9356ae65 100644 --- a/scripts/client/index.html +++ b/scripts/client/index.html @@ -1,35 +1,37 @@ - + - - - Your App - - - -
- - + // Immediately request authIndex from parent + if (window.parent && window.parent !== window) { + window.parent.postMessage({ type: "requestAuthIndex" }, "*"); + } + })(); + + + +
+ + diff --git a/src/auth/AuthSwitcher.js b/src/auth/AuthSwitcher.js index 11670468..64b40d15 100644 --- a/src/auth/AuthSwitcher.js +++ b/src/auth/AuthSwitcher.js @@ -10,6 +10,10 @@ * Handles account switching logic including single/multi-account modes and fallback mechanisms */ class AuthSwitcher { + // Dispose a context only after this many consecutive empty-upstream judgments on the SAME context. + // Prevents a hot dispose/recreate loop when every account is judged empty (e.g. detector false-positive). + static EMPTY_DISPOSE_THRESHOLD = 3; + constructor(logger, config, authSource, browserManager) { this.logger = logger; this.config = config; @@ -18,6 +22,8 @@ class AuthSwitcher { this.failureCount = 0; this.usageCount = 0; this.isSystemBusy = false; + // authIndex -> consecutive empty_upstream_response judgment count. + this._emptyJudgmentCounts = new Map(); } get currentAuthIndex() { @@ -27,6 +33,17 @@ class AuthSwitcher { set currentAuthIndex(value) { this.browserManager.currentAuthIndex = value; } + /** + * Reset the consecutive empty-upstream judgment counter for a successful auth index. + * A success on an account means its next empty judgment starts counting from 1 again; + * only consecutive empties without an intervening success may reach the dispose threshold. + * @param {number|null} authIndex - The account index that served a successful request. + */ + resetEmptyJudgmentCountForAuth(authIndex) { + if (Number.isInteger(authIndex) && authIndex >= 0) { + this._emptyJudgmentCounts.delete(authIndex); + } + } // getNextAuthIndex() { // const available = this.authSource.getRotationIndices(); @@ -49,7 +66,7 @@ class AuthSwitcher { // return available[nextIndexInArray]; // } - async switchToNextAuth() { + async switchToNextAuth(failedAuthIndex = this.currentAuthIndex, allowOriginalFallback = true) { const available = this.authSource.getRotationIndices(); if (available.length === 0) { @@ -64,6 +81,28 @@ class AuthSwitcher { this.isSystemBusy = true; try { + const getCurrentCanonicalIndex = () => + failedAuthIndex >= 0 ? this.authSource.getCanonicalIndex(failedAuthIndex) : -1; + + if (failedAuthIndex >= 0) { + const emptyCount = this._emptyJudgmentCounts.get(failedAuthIndex) || 0; + // Churn guard: dispose a context only after K consecutive empty-upstream judgments + // on it. Non-empty failures (429/403/5xx) never dispose — the context stays warm + // and the account recovers after cooldown, keeping switching instant. + if (emptyCount >= AuthSwitcher.EMPTY_DISPOSE_THRESHOLD) { + this.logger.info( + `🗑️ [Auth] Disposing tainted context #${failedAuthIndex} on account switch/retry...` + ); + await this.browserManager.closeContext(failedAuthIndex).catch(err => { + this.logger.warn(`[Auth] Failed to close context #${failedAuthIndex}: ${err.message}`); + }); + if (emptyCount > 0) this._emptyJudgmentCounts.delete(failedAuthIndex); + } else { + this.logger.info( + `🛡️ [Auth] Skipping context #${failedAuthIndex} disposal (${emptyCount}/${AuthSwitcher.EMPTY_DISPOSE_THRESHOLD} consecutive empty judgments) to avoid churn.` + ); + } + } // Single account mode if (available.length === 1) { const singleIndex = available[0]; @@ -76,6 +115,7 @@ class AuthSwitcher { try { await this.browserManager.launchOrSwitchContext(singleIndex); + this._emptyJudgmentCounts.delete(singleIndex); this.resetCounters(); this.browserManager.rebalanceContextPool().catch(err => { this.logger.error(`[Auth] Background rebalance failed: ${err.message}`); @@ -92,18 +132,14 @@ class AuthSwitcher { } // Multi-account mode - const currentCanonicalIndex = - this.currentAuthIndex >= 0 - ? this.authSource.getCanonicalIndex(this.currentAuthIndex) - : this.currentAuthIndex; - const currentIndexInArray = available.indexOf(currentCanonicalIndex); + const currentIndexInArray = available.indexOf(getCurrentCanonicalIndex()); const hasCurrentAccount = currentIndexInArray !== -1; const startIndex = hasCurrentAccount ? currentIndexInArray : 0; const originalStartAccount = hasCurrentAccount ? available[startIndex] : null; this.logger.info("=================================================="); this.logger.info(`🔄 [Auth] Multi-account mode: Starting intelligent account switching`); - this.logger.info(` • Current account: #${this.currentAuthIndex}`); + this.logger.info(` • Failed account: #${failedAuthIndex}`); this.logger.info( ` • Available accounts (dedup by email, keeping latest index): [${available.join(", ")}]` ); @@ -129,6 +165,7 @@ class AuthSwitcher { `🔄 [Auth] Attempting to switch to account #${accountIndex} (${attemptNumber}/${tryCount} accounts)...` ); + const prevIdx = this.currentAuthIndex; try { // Pre-cleanup: remove excess contexts BEFORE creating new one to avoid exceeding maxContexts await this.browserManager.preCleanupForSwitch(accountIndex); @@ -151,13 +188,15 @@ class AuthSwitcher { return { failedAccounts, newIndex: accountIndex, success: true }; } catch (error) { this.logger.error(`❌ [Auth] Account #${accountIndex} failed: ${error.message}`); + if (this.browserManager.currentAuthIndex === accountIndex) { + this.browserManager.currentAuthIndex = prevIdx; + } failedAccounts.push(accountIndex); } } - // If we had a current account, try it as a final fallback - // If we had no current account, we already tried all accounts, so skip fallback - if (hasCurrentAccount && originalStartAccount !== null) { + // Manual rotation may fall back to the original account; failure recovery must not retry it. + if (allowOriginalFallback && hasCurrentAccount && originalStartAccount !== null) { this.logger.warn("=================================================="); this.logger.warn( `⚠️ [Auth] All other accounts failed. Making final attempt with original starting account #${originalStartAccount}...` @@ -255,7 +294,23 @@ class AuthSwitcher { ); } - const isImmediateSwitch = this.config.immediateSwitchStatusCodes.includes(errorDetails.status); + const isImmediateSwitch = + this.config.immediateSwitchStatusCodes.includes(errorDetails.status) || + errorDetails.status === 502 || + errorDetails.reason === "empty_upstream_response"; + + // Track consecutive empty-upstream judgments per context so we don't dispose/recreate + // contexts in a hot loop when every account is judged empty. Reset on any non-empty failure. + const idx = Number.isInteger(errorDetails.authIndex) ? errorDetails.authIndex : this.currentAuthIndex; + if (errorDetails.reason === "empty_upstream_response") { + if (idx >= 0) { + this._emptyJudgmentCounts.set(idx, (this._emptyJudgmentCounts.get(idx) || 0) + 1); + } + } else { + if (idx >= 0) { + this._emptyJudgmentCounts.delete(idx); + } + } const isThresholdReached = this.config.failureThreshold > 0 && this.failureCount >= this.config.failureThreshold; @@ -271,7 +326,7 @@ class AuthSwitcher { } try { - const result = await this.switchToNextAuth(); + const result = await this.switchToNextAuth(idx, false); if (!result.success) { this.logger.warn(`⚠️ [Auth] Account switch skipped: ${result.reason}`); if (sendErrorCallback) { diff --git a/src/core/BrowserManager.js b/src/core/BrowserManager.js index 0ea00ba0..096cf680 100644 --- a/src/core/BrowserManager.js +++ b/src/core/BrowserManager.js @@ -2619,6 +2619,10 @@ class BrowserManager { const contextData = this.contexts.get(authIndex); + // NOTE: no in-flight-deferral here. `activeRequests` is not maintained anywhere, so a + // deferred-disposal path would be dead code that leaks the context (never actually closed). + // Dispose immediately; any in-flight request on this context is already failing over to + // the next account via the auth-switcher before closeContext is called. // Stop health monitor for this context if (contextData.healthMonitorInterval) { clearInterval(contextData.healthMonitorInterval); diff --git a/src/core/ConnectionRegistry.js b/src/core/ConnectionRegistry.js index 96182561..59711eaf 100644 --- a/src/core/ConnectionRegistry.js +++ b/src/core/ConnectionRegistry.js @@ -270,7 +270,7 @@ class ConnectionRegistry extends EventEmitter { ); return; } - this._routeMessage(parsedMessage, entry.queue); + this._routeMessage(parsedMessage, entry.queue, entry.authIndex); } else { this.logger.warn(`[Server] Received message for unknown or outdated request ID: ${requestId}`); } @@ -279,16 +279,16 @@ class ConnectionRegistry extends EventEmitter { } } - _routeMessage(message, queue) { + _routeMessage(message, queue, authIndex = null) { const { event_type } = message; switch (event_type) { case "response_headers": case "chunk": case "error": - queue.enqueue(message); + queue.enqueue(Number.isInteger(authIndex) ? { ...message, authIndex } : message); break; case "stream_close": - queue.enqueue({ type: "STREAM_END" }); + queue.enqueue(Number.isInteger(authIndex) ? { authIndex, type: "STREAM_END" } : { type: "STREAM_END" }); break; default: this.logger.warn(`[Server] Unknown internal event type: ${event_type}`); diff --git a/src/core/FormatConverter.js b/src/core/FormatConverter.js index f13959ae..32196eb9 100644 --- a/src/core/FormatConverter.js +++ b/src/core/FormatConverter.js @@ -592,6 +592,18 @@ class FormatConverter { // Convert conversation messages const conversationMessages = openaiBody.messages.filter(msg => msg.role !== "system"); + // Build tool_call_id to tool name mapping from assistant messages + const toolIdToNameMap = new Map(); + for (const msg of conversationMessages) { + if (msg.role === "assistant" && Array.isArray(msg.tool_calls)) { + for (const tc of msg.tool_calls) { + if (tc.id && tc.function && tc.function.name) { + toolIdToNameMap.set(tc.id, tc.function.name); + } + } + } + } + // Buffer for accumulating consecutive tool message parts // Gemini requires alternating roles, so consecutive tool messages must be merged let pendingToolParts = []; @@ -674,8 +686,11 @@ class FormatConverter { responseContent = { result: message.content }; } - // Use function name from tool message (OpenAI format always includes name) - const functionName = message.name || "unknown_function"; + // Resolve function name from tool message, or from tool_call_id map + const functionName = + message.name || + (message.tool_call_id && toolIdToNameMap.get(message.tool_call_id)) || + "unknown_function"; // Add to buffer instead of pushing directly // This allows merging consecutive tool messages into one user message @@ -816,9 +831,12 @@ class FormatConverter { // Flush any remaining tool parts after the loop flushToolParts(); + // Merge consecutive contents with the same role (Gemini API requires strict role alternation) + const mergedContents = FormatConverter.mergeConsecutiveSameRoleContents(googleContents); + // Build Google request const googleRequest = { - contents: googleContents, + contents: mergedContents, ...(systemInstruction && { systemInstruction: { parts: systemInstruction.parts, role: "user" }, }), @@ -854,19 +872,47 @@ class FormatConverter { thinkingConfig.includeThoughts = rawThinkingConfig.includeThoughts; } + const rawLevel = rawThinkingConfig.thinking_level ?? rawThinkingConfig.thinkingLevel; + if (rawLevel != null) { + const normalizedLevel = String(rawLevel).trim().toLowerCase(); + const mappedLevel = FormatConverter.THINKING_LEVEL_MAP[normalizedLevel]; + if (mappedLevel) { + thinkingConfig.thinkingLevel = mappedLevel; + } + } + + const rawBudget = rawThinkingConfig.thinking_budget ?? rawThinkingConfig.thinkingBudget; + if (rawBudget != null && typeof rawBudget === "number" && !isNaN(rawBudget)) { + thinkingConfig.thinkingBudget = rawBudget; + } + this.logger.info( `[Adapter] Successfully extracted and converted thinking config: ${JSON.stringify(thinkingConfig)}` ); } // Handle OpenAI reasoning_effort parameter - if (!thinkingConfig) { + if (!thinkingConfig || thinkingConfig.thinkingLevel === undefined) { const effort = openaiBody.reasoning_effort || extraBody.reasoning_effort; - if (effort) { - this.logger.debug( - `[Adapter] Detected OpenAI standard reasoning parameter (reasoning_effort: ${effort}), auto-converting to Google format.` - ); - thinkingConfig = { includeThoughts: true }; + if (effort != null) { + const normalizedEffort = String(effort).trim().toLowerCase(); + const mappedLevel = FormatConverter.THINKING_LEVEL_MAP[normalizedEffort]; + if (!thinkingConfig) { + thinkingConfig = { includeThoughts: true }; + } else if (thinkingConfig.includeThoughts === undefined) { + thinkingConfig.includeThoughts = true; + } + + if (mappedLevel) { + thinkingConfig.thinkingLevel = mappedLevel; + this.logger.debug( + `[Adapter] Detected OpenAI reasoning_effort (${normalizedEffort}), mapped thinkingLevel to ${mappedLevel}.` + ); + } else { + this.logger.debug( + "[Adapter] Detected OpenAI standard reasoning parameter (reasoning_effort), auto-converting to Google format." + ); + } } } @@ -1184,20 +1230,7 @@ class FormatConverter { const delta = {}; let hasContent = false; - if (part.thought === true) { - if (part.text) { - delta.reasoning_content = part.text; - hasContent = true; - } - } else if (part.text) { - delta.content = part.text; - hasContent = true; - } else if (part.inlineData) { - const image = part.inlineData; - delta.content = `![Generated Image](data:${image.mimeType};base64,${image.data})`; - this.logger.info("[Adapter] Successfully parsed image from streaming response chunk."); - hasContent = true; - } else if (part.functionCall) { + if (part.functionCall) { // Convert Gemini functionCall to OpenAI tool_calls format const funcCall = part.functionCall; const toolCallId = `call_${this._generateRequestId()}`; @@ -1225,6 +1258,19 @@ class FormatConverter { `[Adapter] Converted Gemini functionCall to OpenAI tool_calls: ${funcCall.name} (index: ${toolCallIndex})` ); hasContent = true; + } else if (part.thought === true) { + if (part.text) { + delta.reasoning_content = part.text; + hasContent = true; + } + } else if (part.text) { + delta.content = part.text; + hasContent = true; + } else if (part.inlineData) { + const image = part.inlineData; + delta.content = `![Generated Image](data:${image.mimeType};base64,${image.data})`; + this.logger.info("[Adapter] Successfully parsed image from streaming response chunk."); + hasContent = true; } if (hasContent) { @@ -1601,9 +1647,55 @@ class FormatConverter { // Parts -> SSE events if (candidate.content && Array.isArray(candidate.content.parts)) { for (const part of candidate.content.parts) { - // The Responses API exposes reasoning summaries via `summary` + `response.reasoning_summary_text.*`. - // Map Gemini "thought" parts to reasoning *summary* to match official expectations. - if (part?.thought === true) { + // Check functionCall FIRST so a part annotated with `thought: true` alongside a + // tool call is not dropped by the reasoning branch below. + if (part?.functionCall) { + const funcCall = part.functionCall; + const itemId = `fc_${this._generateRequestId()}`; + const callId = `call_${this._generateRequestId()}`; + const outputIndex = streamState.nextOutputIndex++; + const args = JSON.stringify(funcCall.args || {}); + + pushEvent("response.output_item.added", { + item: { + arguments: "", + call_id: callId, + id: itemId, + name: funcCall.name, + status: "in_progress", + type: "function_call", + }, + output_index: outputIndex, + }); + + pushEvent("response.function_call_arguments.done", { + arguments: args, + item_id: itemId, + name: funcCall.name, + output_index: outputIndex, + }); + + const completedToolItem = { + arguments: args, + call_id: callId, + id: itemId, + name: funcCall.name, + status: "completed", + type: "function_call", + }; + streamState.outputItemsByIndex[outputIndex] = completedToolItem; + + pushEvent("response.output_item.done", { + item: completedToolItem, + output_index: outputIndex, + }); + + this.logger.info( + `[Adapter] Converted Gemini functionCall to Response API function_call: ${funcCall.name}` + ); + } else if (part?.thought === true) { + // The Responses API exposes reasoning summaries via `summary` + `response.reasoning_summary_text.*`. + // Map Gemini "thought" parts to reasoning *summary* to match official expectations. if (part?.text) { const reasoningItem = ensureReasoningItem(); streamState.reasoningSummaryText += part.text; @@ -1628,10 +1720,7 @@ class FormatConverter { summary_index: reasoningItem.summary_index ?? 0, }); } - continue; - } - - if (part?.text) { + } else if (part?.text) { const messageItem = ensureMessageItem(); streamState.messageText += part.text; @@ -1659,50 +1748,6 @@ class FormatConverter { output_index: messageItem.output_index, }); } - } else if (part?.functionCall) { - const funcCall = part.functionCall; - const itemId = `fc_${this._generateRequestId()}`; - const callId = `call_${this._generateRequestId()}`; - const outputIndex = streamState.nextOutputIndex++; - const args = JSON.stringify(funcCall.args || {}); - - pushEvent("response.output_item.added", { - item: { - arguments: "", - call_id: callId, - id: itemId, - name: funcCall.name, - status: "in_progress", - type: "function_call", - }, - output_index: outputIndex, - }); - - pushEvent("response.function_call_arguments.done", { - arguments: args, - item_id: itemId, - name: funcCall.name, - output_index: outputIndex, - }); - - const completedToolItem = { - arguments: args, - call_id: callId, - id: itemId, - name: funcCall.name, - status: "completed", - type: "function_call", - }; - streamState.outputItemsByIndex[outputIndex] = completedToolItem; - - pushEvent("response.output_item.done", { - item: completedToolItem, - output_index: outputIndex, - }); - - this.logger.info( - `[Adapter] Converted Gemini functionCall to Response API function_call: ${funcCall.name}` - ); } } } @@ -1721,7 +1766,7 @@ class FormatConverter { const responseUsage = { input_tokens: usage.prompt_tokens, input_tokens_details: { - cached_tokens: 0, + cached_tokens: usage.prompt_tokens_details?.cached_tokens || 0, }, output_tokens: usage.completion_tokens, output_tokens_details: { @@ -1819,14 +1864,7 @@ class FormatConverter { if (candidate.content && Array.isArray(candidate.content.parts)) { for (const part of candidate.content.parts) { - if (part.thought === true) { - reasoning_content += part.text || ""; - } else if (part.text) { - content += part.text; - } else if (part.inlineData) { - const image = part.inlineData; - content += `![Generated Image](data:${image.mimeType};base64,${image.data})`; - } else if (part.functionCall) { + if (part.functionCall) { // Convert Gemini functionCall to OpenAI tool_calls format const funcCall = part.functionCall; const toolCallId = `call_${this._generateRequestId()}`; @@ -1842,6 +1880,13 @@ class FormatConverter { }; tool_calls.push(toolCallObj); this.logger.info(`[Adapter] Converted Gemini functionCall to OpenAI tool_calls: ${funcCall.name}`); + } else if (part.thought === true) { + reasoning_content += part.text || ""; + } else if (part.text) { + content += part.text; + } else if (part.inlineData) { + const image = part.inlineData; + content += `![Generated Image](data:${image.mimeType};base64,${image.data})`; } } } @@ -1952,20 +1997,9 @@ class FormatConverter { let reasoningContent = ""; if (candidate.content && Array.isArray(candidate.content.parts)) { for (const part of candidate.content.parts) { - // Responses API supports reasoning output items; map Gemini "thought" parts into a reasoning *summary*. - if (part?.thought === true) { - if (part?.text) reasoningContent += part.text; - continue; - } else if (part.text) { - // Regular text content - messageContent += part.text; - } else if (part.inlineData) { - // Responses API image outputs are intentionally suppressed by this proxy; preserve a text note. - if (!messageContent) { - messageContent = - "[Image output omitted: Responses API image outputs are disabled by this proxy.]"; - } - } else if (part.functionCall) { + // Check functionCall FIRST so a part annotated with `thought: true` alongside a + // tool call is not dropped by the reasoning branch below. + if (part?.functionCall) { // Function call const funcCall = part.functionCall; const callId = `call_${this._generateRequestId()}`; @@ -1980,6 +2014,17 @@ class FormatConverter { this.logger.info( `[Adapter] Converted Gemini functionCall to Response API function_call: ${funcCall.name}` ); + } else if (part?.thought === true) { + if (part?.text) reasoningContent += part.text; + } else if (part.text) { + // Regular text content + messageContent += part.text; + } else if (part.inlineData) { + // Responses API image outputs are intentionally suppressed by this proxy; preserve a text note. + if (!messageContent) { + messageContent = + "[Image output omitted: Responses API image outputs are disabled by this proxy.]"; + } } } } @@ -2051,7 +2096,7 @@ class FormatConverter { usage: { input_tokens: usage.prompt_tokens, input_tokens_details: { - cached_tokens: 0, + cached_tokens: usage.prompt_tokens_details?.cached_tokens || 0, }, output_tokens: usage.completion_tokens, output_tokens_details: { @@ -2086,7 +2131,16 @@ class FormatConverter { } _parseUsage(googleResponse) { - const usage = googleResponse.usageMetadata || {}; + const usage = googleResponse?.usageMetadata || {}; + + let cachedTokens = 0; + const rawCached = usage.cachedContentTokenCount; + if (typeof rawCached === "number" || typeof rawCached === "string") { + const parsed = Number(rawCached); + if (Number.isFinite(parsed) && parsed > 0) { + cachedTokens = Math.floor(parsed); + } + } const inputTokens = usage.promptTokenCount || 0; const toolPromptTokens = usage.toolUsePromptTokenCount || 0; @@ -2105,7 +2159,7 @@ class FormatConverter { const promptTokens = inputTokens + toolPromptTokens; const totalCompletionTokens = completionTextTokens + reasoningTokens; - const totalTokens = googleResponse.usageMetadata?.totalTokenCount || 0; + const totalTokens = googleResponse?.usageMetadata?.totalTokenCount || 0; return { completion_tokens: totalCompletionTokens, @@ -2116,6 +2170,7 @@ class FormatConverter { }, prompt_tokens: promptTokens, prompt_tokens_details: { + cached_tokens: cachedTokens, text_tokens: inputTokens, tool_tokens: toolPromptTokens, }, @@ -2428,9 +2483,12 @@ class FormatConverter { // Flush remaining tool parts flushToolParts(); + // Merge consecutive contents with the same role (Gemini API requires strict role alternation). + const mergedContents = FormatConverter.mergeConsecutiveSameRoleContents(googleContents); + // Build Google request const googleRequest = { - contents: googleContents, + contents: mergedContents, ...(systemInstruction && { systemInstruction: { parts: systemInstruction.parts, role: "user" }, }), @@ -3290,9 +3348,12 @@ class FormatConverter { } } + // Merge consecutive contents with the same role (Gemini API requires strict role alternation). + const mergedContents = FormatConverter.mergeConsecutiveSameRoleContents(googleContents); + // Build Google request const googleRequest = { - contents: googleContents, + contents: mergedContents, ...(systemInstruction && { systemInstruction, }), @@ -3311,8 +3372,39 @@ class FormatConverter { if (reasoning) { thinkingConfig = { includeThoughts: true }; + // Map Responses reasoning.effort (and the compatible top-level reasoning_effort alias) + // through THINKING_LEVEL_MAP, exactly like the chat path maps reasoning_effort. + const effort = reasoning.effort ?? reasoning.reasoning_effort; + if (effort != null) { + const normalizedEffort = String(effort).trim().toLowerCase(); + const mappedLevel = FormatConverter.THINKING_LEVEL_MAP[normalizedEffort]; + if (mappedLevel) { + thinkingConfig.thinkingLevel = mappedLevel; + this.logger.debug( + `[Adapter] Detected OpenAI Response reasoning.effort (${normalizedEffort}), mapped thinkingLevel to ${mappedLevel}.` + ); + } else { + this.logger.debug( + "[Adapter] Detected OpenAI Response reasoning parameter (reasoning.effort), auto-converting to Google format." + ); + } + } + } else if (responseBody.reasoning_effort != null) { + // Compatible top-level alias (chat-style reasoning_effort) without a reasoning object. + const normalizedEffort = String(responseBody.reasoning_effort).trim().toLowerCase(); + const mappedLevel = FormatConverter.THINKING_LEVEL_MAP[normalizedEffort]; + thinkingConfig = { includeThoughts: true }; + if (mappedLevel) { + thinkingConfig.thinkingLevel = mappedLevel; + this.logger.debug( + `[Adapter] Detected OpenAI Response reasoning_effort (${normalizedEffort}), mapped thinkingLevel to ${mappedLevel}.` + ); + } else { + this.logger.debug( + "[Adapter] Detected OpenAI Response reasoning_effort, auto-converting to Google format." + ); + } } - // Force thinking mode (only set includeThoughts=true when missing) if ( this.serverSystem.config.forceThinking && @@ -3565,6 +3657,18 @@ class FormatConverter { this.logger.info("[Adapter] OpenAI Response API to Google translation complete."); return { cleanModelName, googleRequest, modelStreamingMode }; } + static mergeConsecutiveSameRoleContents(contents) { + if (!Array.isArray(contents)) return contents; + const mergedContents = []; + for (const c of contents) { + if (mergedContents.length > 0 && mergedContents[mergedContents.length - 1].role === c.role) { + mergedContents[mergedContents.length - 1].parts.push(...c.parts); + } else { + mergedContents.push(c); + } + } + return mergedContents; + } } module.exports = FormatConverter; diff --git a/src/core/RequestHandler.js b/src/core/RequestHandler.js index ba65309c..d4be8b9f 100644 --- a/src/core/RequestHandler.js +++ b/src/core/RequestHandler.js @@ -20,7 +20,7 @@ const WS_CONNECTION_READY_TIMEOUT_MS = 10000; // Default timeout constants (in milliseconds) const DEFAULT_TIMEOUTS = { FAKE_STREAM: 300000, // 300 seconds (5 minutes) - timeout for fake streaming (buffered response) - STREAM_CHUNK: 60000, // 60 seconds - timeout between stream chunks + STREAM_CHUNK: 0, // 0 = disabled - timeout between stream chunks }; class RequestHandler { @@ -648,14 +648,50 @@ class RequestHandler { return { attemptedAuthIndices }; } + _withFailureAuthIndex(details, requestId, explicitAuthIndex = null) { + if (!details || typeof details !== "object") return details; + if (Number.isInteger(details.authIndex)) return details; + const requestAuthIndex = Number.isInteger(explicitAuthIndex) + ? explicitAuthIndex + : this.connectionRegistry?.getAuthIndexForRequest?.(requestId); + return Number.isInteger(requestAuthIndex) ? { ...details, authIndex: requestAuthIndex } : details; + } + + async _handleAuthFailure(details, requestId, userMessage = null, explicitAuthIndex = null) { + return this.authSwitcher?.handleRequestFailureAndSwitch( + this._withFailureAuthIndex(details, requestId, explicitAuthIndex), + userMessage + ); + } + + /** + * Reset failure bookkeeping after a successful request on a given auth index. + * Shared by every success site so the consecutive empty-upstream judgment counter is + * cleared for the account that actually served the request (in addition to failureCount). + * @param {number|null} authIndex - Auth index that served the successful request. + */ + _resetFailureStateOnSuccess(authIndex = null) { + const index = Number.isInteger(authIndex) && authIndex >= 0 ? authIndex : this.currentAuthIndex; + if (typeof this.authSwitcher?.resetEmptyJudgmentCountForAuth === "function") { + this.authSwitcher.resetEmptyJudgmentCountForAuth(index); + } + if (this.authSwitcher?.failureCount > 0) { + this.logger.debug( + `✅ [Auth] Request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` + ); + this.authSwitcher.failureCount = 0; + } + } _getImmediateStatusRetryCloseReason(status) { return `immediate_status_retry_${status}`; } async _performImmediateSwitchRetry(errorDetails, requestId, tracker) { - await this.authSwitcher.handleRequestFailureAndSwitch( + await this._handleAuthFailure( { message: errorDetails.message, status: Number(errorDetails.status) }, - null + requestId, + null, + errorDetails.authIndex ); const ready = await this._waitForSystemAndConnectionIfBusy(null, { @@ -1226,6 +1262,38 @@ class RequestHandler { this._forwardRequest(proxyRequest, currentQueueAuthIndex); initialMessage = await currentQueue.dequeue(); + if ( + initialMessage && + initialMessage.event_type === "chunk" && + initialMessage.data !== undefined + ) { + // _dumpUpstreamCorrelation records JUDGED-EMPTY responses only (the helper + // itself re-checks _isEmptyUpstreamResponse before any file I/O). It never + // writes for non-empty responses; this call site just feeds it the raw chunk. + this._dumpUpstreamCorrelation( + "processOpenAIRequest:initialMessage", + initialMessage.data, + requestId, + model, + currentQueueAuthIndex + ); + } + if ( + initialMessage && + initialMessage.event_type === "chunk" && + this._isEmptyUpstreamResponse(initialMessage.data) + ) { + this.logger.warn( + `[Request] Detected empty upstream response on account index ${currentQueueAuthIndex}. Preparing retry...` + ); + initialMessage = { + event_type: "error", + message: "Empty upstream completion (zero content, zero tool_calls)", + reason: "empty_upstream_response", + status: 502, + }; + } + const initialStatus = Number(initialMessage?.status); if ( initialMessage.event_type === "error" && @@ -1278,7 +1346,7 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!skipFinalFailureSwitch && !this._isConnectionResetError(initialMessage)) { - await this.authSwitcher.handleRequestFailureAndSwitch(initialMessage, null); + await this._handleAuthFailure(initialMessage, requestId, null, initialMessage.authIndex); } else if (skipFinalFailureSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -1291,12 +1359,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] OpenAI interface request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(currentQueueAuthIndex); res.status(200).set({ "Cache-Control": "no-cache", @@ -1345,7 +1408,7 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!result.error.skipAccountSwitch && !this._isConnectionResetError(result.error)) { - await this.authSwitcher.handleRequestFailureAndSwitch(result.error, null); + await this._handleAuthFailure(result.error, requestId, null, result.queue?.authIndex); } else if (result.error.skipAccountSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -1358,12 +1421,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] OpenAI interface request successful - failure count reset to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(result.queue?.authIndex); // Use the queue that successfully received the initial message const activeQueue = result.queue; @@ -1418,6 +1476,41 @@ class RequestHandler { // Backend errored; don't attempt to translate/send a "normal" stream afterwards. return; } + + // Terminal emptiness judgment for the OpenAI chat fake-stream path, mirroring + // the Response API/Claude fake-stream paths: an empty aggregate body must + // enter the existing single auth-failure + SSE error flow, with no duplicate + // switch and no leaked empty completion. + if (this._isEmptyUpstreamResponse(fullBody)) { + this._dumpUpstreamCorrelation( + "openai-chat-fake-stream", + fullBody, + requestId, + model, + this.currentAuthIndex + ); + this.logger.warn( + `⚠️ [Request] Upstream fake-stream response judged empty (request ${requestId}); switching account and ending stream.` + ); + this._handleRequestError( + { + message: "Empty upstream response (OpenAI chat fake stream)", + reason: "empty_upstream_response", + }, + res, + requestId + ); + this._handleAuthFailure( + { + message: "Empty upstream response (OpenAI chat fake stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId + ); + return; + } + const streamState = {}; const translatedChunk = this.formatConverter.translateGoogleToOpenAIStream( fullBody, @@ -1629,6 +1722,39 @@ class RequestHandler { this._forwardRequest(proxyRequest, currentQueueAuthIndex); initialMessage = await currentQueue.dequeue(); + if ( + initialMessage && + initialMessage.event_type === "chunk" && + initialMessage.data !== undefined + ) { + // _dumpUpstreamCorrelation records JUDGED-EMPTY responses only. + this._dumpUpstreamCorrelation( + "processOpenAIResponseRequest:initialMessage", + initialMessage.data, + requestId, + model, + currentQueueAuthIndex + ); + } + // A complete initial chunk that is already terminal-empty must be converted into + // the existing error/retry flow BEFORE the translator can set responseSent — one + // switch only, exactly like the OpenAI chat real-stream path. + if ( + initialMessage && + initialMessage.event_type === "chunk" && + this._isEmptyUpstreamResponse(initialMessage.data) + ) { + this.logger.warn( + `[Request] Detected empty upstream response on account index ${currentQueueAuthIndex}. Preparing retry...` + ); + initialMessage = { + event_type: "error", + message: "Empty upstream completion (zero content, zero tool_calls)", + reason: "empty_upstream_response", + status: 502, + }; + } + const initialStatus = Number(initialMessage?.status); if ( initialMessage.event_type === "error" && @@ -1681,7 +1807,7 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!skipFinalFailureSwitch && !this._isConnectionResetError(initialMessage)) { - await this.authSwitcher.handleRequestFailureAndSwitch(initialMessage, null); + await this._handleAuthFailure(initialMessage, requestId, null, initialMessage.authIndex); } else if (skipFinalFailureSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -1694,12 +1820,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] OpenAI Response API request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(currentQueueAuthIndex); res.status(200).set({ "Cache-Control": "no-cache", @@ -1755,7 +1876,7 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!result.error.skipAccountSwitch && !this._isConnectionResetError(result.error)) { - await this.authSwitcher.handleRequestFailureAndSwitch(result.error, null); + await this._handleAuthFailure(result.error, requestId, null, result.queue?.authIndex); } else if (result.error.skipAccountSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -1768,12 +1889,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] OpenAI Response API request successful - failure count reset to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(result.queue?.authIndex); // Use the queue that successfully received the initial message const activeQueue = result.queue; @@ -1838,6 +1954,39 @@ class RequestHandler { return; } + // Terminal emptiness judgment for the OpenAI Response API fake stream path. + // _isEmptyUpstreamResponse handles an empty or parseable-but-empty body and treats a + // fragmented unparseable tail as not-conclusively-empty, so translation proceeds unchanged. + if (this._isEmptyUpstreamResponse(fullBody)) { + this._dumpUpstreamCorrelation( + "openai-response-api-fake-stream", + fullBody, + requestId, + model, + this.currentAuthIndex + ); + this.logger.warn( + `⚠️ [Request] Upstream fake-stream response judged empty (request ${requestId}); switching account and ending stream.` + ); + this._handleRequestError( + { + message: "Empty upstream response (Response API fake stream)", + reason: "empty_upstream_response", + }, + res, + requestId + ); + this._handleAuthFailure( + { + message: "Empty upstream response (Response API fake stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId + ); + return; + } + const streamState = {}; streamState.responseDefaults = responseDefaults; const translatedChunk = this.formatConverter.translateGoogleToResponseAPIStream( @@ -2049,7 +2198,7 @@ class RequestHandler { }); this._sendErrorResponse(res, initialMessage.status || 500, initialMessage.message, "api_error"); if (!skipFinalFailureSwitch && !this._isConnectionResetError(initialMessage)) { - await this.authSwitcher.handleRequestFailureAndSwitch(initialMessage, null); + await this._handleAuthFailure(initialMessage, requestId, null, initialMessage.authIndex); } else if (skipFinalFailureSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -2058,10 +2207,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug(`✅ [Auth] Claude request successful - failure count reset to 0`); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(currentQueueAuthIndex); res.status(200).set({ "Cache-Control": "no-cache", @@ -2111,7 +2257,7 @@ class RequestHandler { ); } if (!result.error.skipAccountSwitch && !this._isConnectionResetError(result.error)) { - await this.authSwitcher.handleRequestFailureAndSwitch(result.error, null); + await this._handleAuthFailure(result.error, requestId, null, result.queue?.authIndex); } else if (result.error.skipAccountSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -2120,10 +2266,7 @@ class RequestHandler { return; } - if (this.authSwitcher.failureCount > 0) { - this.logger.debug(`✅ [Auth] Claude request successful - failure count reset to 0`); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(result.queue?.authIndex); // Use the queue that successfully received the initial message const activeQueue = result.queue; @@ -2183,6 +2326,40 @@ class RequestHandler { // Backend errored; don't attempt to translate/send a "normal" stream afterwards. return; } + + // Terminal emptiness judgment for the Claude fake-stream path, mirroring the + // OpenAI Response API fake-stream path: an empty aggregate body must enter the + // existing single auth-failure + SSE error flow, with no duplicate switch. + if (this._isEmptyUpstreamResponse(fullBody)) { + this._dumpUpstreamCorrelation( + "claude-fake-stream", + fullBody, + requestId, + model, + this.currentAuthIndex + ); + this.logger.warn( + `⚠️ [Request] Upstream fake-stream response judged empty (request ${requestId}); switching account and ending stream.` + ); + this._handleRequestError( + { + message: "Empty upstream response (Claude fake stream)", + reason: "empty_upstream_response", + }, + res, + requestId + ); + this._handleAuthFailure( + { + message: "Empty upstream response (Claude fake stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId + ); + return; + } + const streamState = {}; const translatedChunk = this.formatConverter.translateGoogleToClaudeStream( fullBody, @@ -2324,7 +2501,7 @@ class RequestHandler { ); this._sendErrorResponse(res, response.status || 500, response.message, "api_error"); if (!this._isConnectionResetError(response)) { - await this.authSwitcher.handleRequestFailureAndSwitch(response, null); + await this._handleAuthFailure(response, requestId, null, response.authIndex); } return; } @@ -2352,13 +2529,7 @@ class RequestHandler { const geminiResponse = JSON.parse(fullBody || response.body); const totalTokens = geminiResponse.totalTokens || 0; - // Reset failure count on success - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] Count tokens request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(); // Return Claude-compatible response res.status(200).json({ @@ -2468,7 +2639,7 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!this._isConnectionResetError(response)) { - await this.authSwitcher.handleRequestFailureAndSwitch(response, null); + await this._handleAuthFailure(response, requestId, null, response.authIndex); } else { this.logger.info( "[Request] Failure due to connection reset (input_tokens), skipping account switch." @@ -2511,13 +2682,7 @@ class RequestHandler { const totalTokens = geminiResponse.totalTokens || 0; - // Reset failure count on success - if (this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] input_tokens request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; - } + this._resetFailureStateOnSuccess(); res.status(200).json({ input_tokens: totalTokens, @@ -2541,6 +2706,7 @@ class RequestHandler { async _streamClaudeResponse(messageQueue, res, model, requestId) { const streamState = {}; + let sseBuffer = ""; try { // eslint-disable-next-line no-constant-condition @@ -2548,6 +2714,54 @@ class RequestHandler { const message = await messageQueue.dequeue(this.timeouts.STREAM_CHUNK); if (message.type === "STREAM_END") { + // Flush any trailing partial SSE payload before classifying the stream as empty: + // a fragmented final event split across browser network chunks reassembles here and + // may be the only real content the upstream produced. + let flushEmittedOutput = false; + if (sseBuffer.trim() !== "") { + const claudeChunk = this._translateCompleteSseEvent( + sseBuffer, + model, + streamState, + "translateGoogleToClaudeStream" + ); + if (claudeChunk && this._isResponseWritable(res)) { + try { + res.write(claudeChunk); + flushEmittedOutput = true; + } catch (writeError) { + this.logger.debug( + `[Request] Failed to flush Claude stream chunk: ${writeError.message}` + ); + } + } + } + // Terminal empty detection: if the upstream produced no content block (and the + // trailing flush did not emit output either), treat it as empty. Any output — + // including a flush that advanced contentBlockIndex — means the stream is non-empty. + if (!streamState.contentBlockIndex && !flushEmittedOutput) { + this.logger.warn( + `⚠️ [Request] Upstream stream judged empty at STREAM_END (request ${requestId}); switching account and sending SSE error.` + ); + this._handleAuthFailure( + { + message: "Empty upstream response (stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId, + null, + message.authIndex + ); + // Headers are already sent once we reach STREAM_END mid-stream, so the JSON + // _sendErrorResponse would be a silent no-op. Emit a protocol-safe SSE error. + if (res.headersSent) { + this._sendErrorChunkToClient(res, "Empty upstream response", 502); + } else { + this._sendErrorResponse(res, 502, "Empty upstream response"); + } + break; + } this.logger.info(`✅ [Request] Response completed (Claude real stream), request ID: ${requestId}`); break; } @@ -2578,30 +2792,36 @@ class RequestHandler { } if (message.data) { - const claudeChunk = this.formatConverter.translateGoogleToClaudeStream( - message.data, - model, - streamState - ); - if (claudeChunk) { - // Before writing, ensure the response is still writable to avoid - // throwing if the client disconnected mid-stream. - if (!this._isResponseWritable(res)) { - this.logger.debug( - "[Request] Response no longer writable during Claude stream; stopping stream." - ); - break; - } - try { - res.write(claudeChunk); - } catch (writeError) { - this.logger.debug( - `[Request] Failed to write Claude chunk to stream: ${writeError.message}` - ); - // Stop streaming on write failure to avoid misclassifying as a timeout. - break; + sseBuffer += message.data; + const events = this._extractSseEvents(sseBuffer); + for (const eventPayload of events.complete) { + const claudeChunk = this._translateCompleteSseEvent( + eventPayload, + model, + streamState, + "translateGoogleToClaudeStream" + ); + if (claudeChunk) { + // Before writing, ensure the response is still writable to avoid + // throwing if the client disconnected mid-stream. + if (!this._isResponseWritable(res)) { + this.logger.debug( + "[Request] Response no longer writable during Claude stream; stopping stream." + ); + break; + } + try { + res.write(claudeChunk); + } catch (writeError) { + this.logger.debug( + `[Request] Failed to write Claude chunk to stream: ${writeError.message}` + ); + // Stop streaming on write failure to avoid misclassifying as a timeout. + break; + } } } + sseBuffer = events.remainder; } } } catch (error) { @@ -2641,6 +2861,14 @@ class RequestHandler { try { const googleResponse = JSON.parse(fullBody); + // Write a judged-empty-only correlation dump (the helper early-returns on non-empty), + // gated by DUMP_EMPTY_UPSTREAM. Non-empty upstream responses are not recorded here. + this._dumpUpstreamCorrelation("non-stream", fullBody, requestId, model, this.currentAuthIndex); + // Terminal emptiness judgment for the Claude non-stream path. + if (this._isEmptyUpstreamResponse(googleResponse)) { + this._handleEmptyNonStreamResponse(res, requestId); + return; + } const claudeResponse = this.formatConverter.convertGoogleToClaudeNonStream(googleResponse, model); res.type("application/json").send(JSON.stringify(claudeResponse)); this.logger.info(`✅ [Request] Response completed (Claude non-stream), request ID: ${requestId}`); @@ -2695,7 +2923,12 @@ class RequestHandler { // Avoid switching account if the error is just a connection reset if (!result.error.skipAccountSwitch && !this._isConnectionResetError(result.error)) { - await this.authSwitcher.handleRequestFailureAndSwitch(result.error, null); + await this._handleAuthFailure( + result.error, + proxyRequest.request_id, + null, + result.queue?.authIndex + ); } else if (result.error.skipAccountSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -2709,11 +2942,8 @@ class RequestHandler { return; } - if (proxyRequest.is_generative && this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] Generation request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; + if (proxyRequest.is_generative) { + this._resetFailureStateOnSuccess(result.queue?.authIndex); } // Use the queue that successfully received the initial message @@ -2775,6 +3005,37 @@ class RequestHandler { try { const googleResponse = JSON.parse(fullData); this._logGeminiNativeResponseDebug(googleResponse, "pseudo-stream"); + + if (this._isEmptyUpstreamResponse(googleResponse)) { + this._dumpUpstreamCorrelation( + "gemini-native-pseudo-stream", + googleResponse, + proxyRequest.request_id, + proxyRequest.model, + this.currentAuthIndex + ); + this.logger.warn( + `⚠️ [Request] Upstream pseudo-stream response judged empty (request ${proxyRequest.request_id}); switching account and ending stream.` + ); + this._handleRequestError( + { + message: "Empty upstream response (pseudo-stream)", + reason: "empty_upstream_response", + }, + res, + proxyRequest.request_id + ); + this._handleAuthFailure( + { + message: "Empty upstream response (pseudo-stream)", + reason: "empty_upstream_response", + status: 502, + }, + proxyRequest.request_id + ); + return; + } + const candidate = googleResponse.candidates?.[0]; if (candidate && candidate.content && Array.isArray(candidate.content.parts)) { @@ -2948,6 +3209,26 @@ class RequestHandler { ); this._forwardRequest(proxyRequest, currentQueueAuthIndex); headerMessage = await currentQueue.dequeue(); + if (headerMessage?.event_type === "chunk" && headerMessage.data !== undefined) { + this._dumpUpstreamCorrelation( + "gemini-native-real-stream:header", + headerMessage?.data, + proxyRequest.request_id, + proxyRequest.model, + currentQueueAuthIndex + ); + } + if (headerMessage?.event_type === "chunk" && this._isEmptyUpstreamResponse(headerMessage?.data)) { + this.logger.warn( + `[Request] Gemini real stream detected empty upstream response on account index ${currentQueueAuthIndex}. Preparing retry...` + ); + headerMessage = { + event_type: "error", + message: "Empty upstream completion (zero content, zero function calls)", + reason: "empty_upstream_response", + status: 502, + }; + } const headerStatus = Number(headerMessage?.status); if ( @@ -3002,7 +3283,7 @@ class RequestHandler { }); // Avoid switching account if the error is just a connection reset if (!skipFinalFailureSwitch && !this._isConnectionResetError(headerMessage)) { - await this.authSwitcher.handleRequestFailureAndSwitch(headerMessage, null); + await this._handleAuthFailure(headerMessage, proxyRequest.request_id, null, currentQueueAuthIndex); } else if (skipFinalFailureSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -3018,11 +3299,8 @@ class RequestHandler { return; } - if (proxyRequest.is_generative && this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] Generation request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; + if (proxyRequest.is_generative) { + this._resetFailureStateOnSuccess(currentQueueAuthIndex); } this._setResponseHeaders(res, headerMessage, req); @@ -3109,7 +3387,12 @@ class RequestHandler { this._logFinalRequestFailure(result.error, "Gemini non-stream", proxyRequest.request_id); // Avoid switching account if the error is just a connection reset if (!result.error.skipAccountSwitch && !this._isConnectionResetError(result.error)) { - await this.authSwitcher.handleRequestFailureAndSwitch(result.error, null); + await this._handleAuthFailure( + result.error, + proxyRequest.request_id, + null, + result.queue?.authIndex + ); } else if (result.error.skipAccountSwitch) { this.logger.info( "[Request] Immediate-switch retries exhausted, skipping additional account switch." @@ -3124,11 +3407,8 @@ class RequestHandler { } // On success, reset failure count if needed - if (proxyRequest.is_generative && this.authSwitcher.failureCount > 0) { - this.logger.debug( - `✅ [Auth] Non-stream generation request successful - failure count reset from ${this.authSwitcher.failureCount} to 0` - ); - this.authSwitcher.failureCount = 0; + if (proxyRequest.is_generative) { + this._resetFailureStateOnSuccess(result.queue?.authIndex); } // Use the queue that successfully received the initial message @@ -3160,12 +3440,24 @@ class RequestHandler { const fullBodyBuffer = Buffer.concat(chunks); let responseBodyBuffer = fullBodyBuffer; + let fullResponse = null; try { - const fullResponse = JSON.parse(responseBodyBuffer.toString()); + fullResponse = JSON.parse(responseBodyBuffer.toString()); this._logGeminiNativeResponseDebug(fullResponse, "non-stream"); } catch (e) { // Ignore JSON parsing errors for finish reason } + if (fullResponse && this._isEmptyUpstreamResponse(fullResponse)) { + this._dumpUpstreamCorrelation( + "gemini-native-non-stream", + responseBodyBuffer, + proxyRequest.request_id, + proxyRequest.model, + this.currentAuthIndex + ); + this._handleEmptyNonStreamResponse(res, proxyRequest.request_id); + return; + } if (proxyRequest.response_transform === "batchEmbedToEmbedContent") { try { @@ -3230,6 +3522,155 @@ class RequestHandler { return fullBody; } + _isEmptyUpstreamResponse(data) { + if (!data) return true; + let obj = typeof data === "object" ? data : null; + if (typeof data === "string") { + try { + obj = JSON.parse(data); + } catch (e) { + const match = data.match(/data:\s*(\{.*\})/); + if (match) { + try { + obj = JSON.parse(match[1]); + } catch (e2) { + /* empty */ + } + } + } + // The raw chunk may be a fragmented SSE stream: multiple `data:` events in one chunk, + // or a partial event split across browser network chunks. If we can't cleanly parse it + // as a single complete event, do NOT conclude it is empty — more content may be coming. + // Split on `\n\n` and judge based on the parsed events instead. + if (!obj) { + const events = String(data).split("\n\n"); + for (const evt of events) { + const line = evt.trim(); + if (!line) continue; + const dIdx = line.startsWith("data:") ? 5 : line.indexOf("data:"); + if (dIdx < 0) continue; + const payload = line.slice(dIdx > 0 ? dIdx + 5 : 5).trim(); + if (!payload || payload === "[DONE]") continue; + let evtObj = null; + try { + evtObj = JSON.parse(payload); + } catch (e3) { + continue; + } + // If ANY event carries content or is non-terminal, the stream is not empty. + if (!this._isEmptyUpstreamResponse(evtObj)) return false; + } + // Fell through: every parseable event was empty. Still, an unparseable partial + // event means we cannot be certain — treat as not-conclusively-empty. + return false; + } + } + if (!obj) return true; + + if (Array.isArray(obj.content)) { + const hasAnthropicContent = obj.content.some(block => { + if (!block || typeof block !== "object") return false; + if (block.type === "text" || block.type === "thinking" || block.type === "redacted_thinking") { + return typeof block.text === "string" && block.text.trim().length > 0; + } + if (block.type === "tool_use") { + return typeof block.name === "string" && block.name.trim().length > 0; + } + return typeof block.type === "string" && block.type.length > 0; + }); + if (hasAnthropicContent) return false; + if (!obj.stop_reason) return false; + return (obj.usage?.output_tokens ?? 0) === 0; + } + + if (obj.candidates && Array.isArray(obj.candidates)) { + if (obj.promptFeedback && obj.promptFeedback.blockReason) return false; + const cand = obj.candidates[0]; + if (!cand) return false; + const parts = cand.content?.parts || []; + const hasToolCalls = parts.some(p => p.functionCall && p.functionCall.name); + const hasNonWhitespaceText = parts.some(p => typeof p.text === "string" && p.text.trim().length > 0); + const completionTokens = + (obj.usageMetadata?.candidatesTokenCount ?? 0) + (obj.usageMetadata?.thoughtsTokenCount ?? 0); + const isTerminal = !!cand.finishReason; + + // Control finish reasons (safety/blocklist/recitation/prohibited-content/image-safety) are + // VALID terminal results even with zero completion tokens — the upstream answered by + // refusing/blocking the request. Never judge them empty, otherwise the client sees a + // spurious 502 + account switch for a legitimate safety refusal. Do NOT exempt arbitrary + // OTHER/unknown reasons — only the known Gemini control reasons. + const CONTROL_FINISH_REASONS = new Set([ + "SAFETY", + "RECITATION", + "BLOCKLIST", + "PROHIBITED_CONTENT", + "IMAGE_SAFETY", + ]); + if (isTerminal && CONTROL_FINISH_REASONS.has(String(cand.finishReason).toUpperCase())) { + return false; + } + + // Real content (tool call or non-whitespace text) → not empty. + if (hasToolCalls || hasNonWhitespaceText) return false; + // A non-terminal chunk is an in-progress stream (incl. thinking-only chunks) — more content + // is coming. Never abort on it; judgment happens at the terminal/aggregate point. + if (!isTerminal) return false; + // Terminal response: EMPTY iff no real content AND zero completion tokens. Whitespace-only + // text (e.g. parts:[{"text":" "}] or {"text":"\n"}) with stop is empty. Reasoning/thinking + // is NOT final content — it is only a mid-stream signal handled by the !isTerminal guard. + // A whitespace-only-text-with-thoughtSignature terminal response is therefore EMPTY. + // Nonzero completion tokens means a real (oddly-formatted) answer → not empty. + return completionTokens === 0; + } + + if (obj.choices && Array.isArray(obj.choices)) { + const choice = obj.choices[0]; + if (!choice) return false; + const msg = choice.message || choice.delta || {}; + const hasToolCalls = Array.isArray(msg.tool_calls) && msg.tool_calls.length > 0; + const hasNonWhitespaceContent = typeof msg.content === "string" && msg.content.trim().length > 0; + const completionTokens = obj.usage?.completion_tokens ?? 0; + const isTerminal = !!choice.finish_reason; + + // OpenAI safety/content-filter controls are valid terminal results with zero tokens. + if (isTerminal && ["content_filter", "safety"].includes(String(choice.finish_reason).toLowerCase())) { + return false; + } + + if (hasToolCalls || hasNonWhitespaceContent) return false; + if (!isTerminal) return false; + return completionTokens === 0; + } + + return false; + } + + _dumpUpstreamCorrelation(siteTag, rawData, requestId, model = "unknown", authIndex = null) { + const dumpPath = process.env.DUMP_EMPTY_UPSTREAM; + if (!dumpPath) return; // keep out of the hot path when unset + try { + if (!this._isEmptyUpstreamResponse(rawData)) return; // diagnostic is for judged-empty only + const rawText = typeof rawData === "string" ? rawData : JSON.stringify(rawData); + require("fs").appendFileSync( + dumpPath, + JSON.stringify({ + account_index: authIndex, + judged_empty: true, + model, + raw_response: rawText.slice(0, 200000), + raw_response_length: rawText.length, + request_id: requestId, + site: siteTag, + timestamp: new Date().toISOString(), + }) + "\n" + ); + } catch (e) { + this.logger.error( + `❌ [Dump] Failed to write DUMP_EMPTY_UPSTREAM record to "${dumpPath}": ${e?.message || e}. Check the path is writable and exists.` + ); + } + } + async _executeRequestWithRetries(proxyRequest, messageQueue) { let lastError = null; let currentQueue = messageQueue; @@ -3486,12 +3927,63 @@ class RequestHandler { // Keep Response API sequence numbers consistent across helpers that might write to the same SSE response. if (res.__responseApiSeq == null) res.__responseApiSeq = 0; streamState.sequenceNumber = res.__responseApiSeq; + // SSE reassembly buffer: browser network chunks do not align with SSE `\n\n` event boundaries. + let sseBuffer = ""; try { // eslint-disable-next-line no-constant-condition while (true) { const message = await messageQueue.dequeue(this.timeouts.STREAM_CHUNK); if (message.type === "STREAM_END") { + // Flush any trailing partial SSE payload before classifying the stream as empty. + let flushEmittedOutput = false; + if (sseBuffer.trim() !== "") { + const responseAPIChunk = this._translateCompleteSseEvent( + sseBuffer, + model, + streamState, + "translateGoogleToResponseAPIStream" + ); + if (typeof streamState.sequenceNumber === "number") { + res.__responseApiSeq = streamState.sequenceNumber; + } + if (responseAPIChunk && this._isResponseWritable(res)) { + try { + res.write(responseAPIChunk); + flushEmittedOutput = true; + } catch (writeError) { + this.logger.debug( + `[Request] Failed to flush Response API stream chunk: ${writeError.message}` + ); + } + } + } + // Terminal empty detection: if the upstream produced no response object (and the + // trailing flush did not emit output either), treat it as empty. A flush that set + // responseSent means the stream is non-empty. + if (!streamState.responseSent && !flushEmittedOutput) { + this.logger.warn( + `⚠️ [Request] Upstream stream judged empty at STREAM_END (request ${requestId}); switching account and sending SSE error.` + ); + this._handleAuthFailure( + { + message: "Empty upstream response (stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId, + null, + message.authIndex + ); + // Headers are already sent once we reach STREAM_END mid-stream, so the JSON + // _sendErrorResponse would be a silent no-op. Emit a protocol-safe SSE error. + if (res.headersSent) { + this._sendErrorChunkToClient(res, "Empty upstream response", 502); + } else { + this._sendErrorResponse(res, 502, "Empty upstream response"); + } + break; + } this.logger.info( `✅ [Request] Response completed (OpenAI Response API real stream), request ID: ${requestId}` ); @@ -3525,30 +4017,36 @@ class RequestHandler { } if (message.data) { - const responseAPIChunk = this.formatConverter.translateGoogleToResponseAPIStream( - message.data, - model, - streamState - ); - if (typeof streamState.sequenceNumber === "number") { - res.__responseApiSeq = streamState.sequenceNumber; - } - if (responseAPIChunk) { - if (!this._isResponseWritable(res)) { - this.logger.debug( - "[Request] Response no longer writable during Response API stream; stopping stream." - ); - break; + sseBuffer += message.data; + const events = this._extractSseEvents(sseBuffer); + for (const eventPayload of events.complete) { + const responseAPIChunk = this._translateCompleteSseEvent( + eventPayload, + model, + streamState, + "translateGoogleToResponseAPIStream" + ); + if (typeof streamState.sequenceNumber === "number") { + res.__responseApiSeq = streamState.sequenceNumber; } - try { - res.write(responseAPIChunk); - } catch (writeError) { - this.logger.debug( - `[Request] Failed to write Response API chunk (connection likely closed): ${writeError.message}` - ); - break; + if (responseAPIChunk) { + if (!this._isResponseWritable(res)) { + this.logger.debug( + "[Request] Response no longer writable during Response API stream; stopping stream." + ); + break; + } + try { + res.write(responseAPIChunk); + } catch (writeError) { + this.logger.debug( + `[Request] Failed to write Response API chunk (connection likely closed): ${writeError.message}` + ); + break; + } } } + sseBuffer = events.remainder; } } } catch (error) { @@ -3566,11 +4064,62 @@ class RequestHandler { async _streamOpenAIResponse(messageQueue, res, model, requestId) { const streamState = {}; + // SSE reassembly buffer: browser network chunks from the page (build.js stream loop) do not + // align with SSE `\n\n` event boundaries. A raw chunk may carry multiple `data:` events or a + // partial event split across chunks. Accumulate here and split on `\n\n` so each complete + // event is parsed independently — mirroring the non-stream path's `fullBody` accumulation. + let sseBuffer = ""; + try { // eslint-disable-next-line no-constant-condition while (true) { const message = await messageQueue.dequeue(this.timeouts.STREAM_CHUNK); if (message.type === "STREAM_END") { + // Flush any trailing partial SSE payload before classifying the stream as empty: + // a fragmented final event split across browser network chunks reassembles here and + // may be the only real content the upstream produced. + let flushEmittedOutput = false; + if (sseBuffer.trim() !== "") { + const flushed = this._translateCompleteSseEvent(sseBuffer, model, streamState); + if (flushed && this._isResponseWritable(res)) { + try { + res.write(flushed); + flushEmittedOutput = true; + } catch (writeError) { + this.logger.debug( + `[Request] Failed to write flushed SSE event to OpenAI stream: ${writeError.message}` + ); + } + } + } + // Terminal emptiness judgment, consistent with the non-stream path. The first-chunk + // check above only catches a complete empty first payload; a fragmented empty response + // (empty body split across chunks) reassembles here with no content ever emitted. If + // roleSent is still false (and the trailing flush emitted nothing), no text/thought/ + // image/tool_call was produced — treat as an empty upstream response, switch account, + // and send an SSE error. Thinking-only streams keep roleSent=true, so they are NOT + // aborted (mid-stream false-positive protection). + if (!streamState.roleSent && !flushEmittedOutput) { + this.logger.warn( + `⚠️ [Request] Upstream stream judged empty at STREAM_END (request ${requestId}); switching account and sending SSE error.` + ); + this._handleAuthFailure( + { + message: "Empty upstream response (stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId, + null, + message.authIndex + ); + if (res.headersSent) { + this._sendErrorChunkToClient(res, "Empty upstream response", 502); + } else { + this._sendErrorResponse(res, 502, "Empty upstream response"); + } + break; + } if (this._isResponseWritable(res)) { try { res.write("data: [DONE]\n\n"); @@ -3604,25 +4153,27 @@ class RequestHandler { } if (message.data) { - const openAIChunk = this.formatConverter.translateGoogleToOpenAIStream( - message.data, - model, - streamState - ); - if (openAIChunk) { - if (!this._isResponseWritable(res)) { - this.logger.debug( - "[Request] Response no longer writable during OpenAI stream; stopping stream." - ); - break; - } - try { - res.write(openAIChunk); - } catch (writeError) { - this.logger.debug( - `[Request] Failed to write OpenAI chunk to stream: ${writeError.message}` - ); - break; + // Reassemble SSE events from raw network chunks before parsing. + sseBuffer += message.data; + const events = this._extractSseEvents(sseBuffer); + sseBuffer = events.remainder; + for (const eventPayload of events.complete) { + const openAIChunk = this._translateCompleteSseEvent(eventPayload, model, streamState); + if (openAIChunk) { + if (!this._isResponseWritable(res)) { + this.logger.debug( + "[Request] Response no longer writable during OpenAI stream; stopping stream." + ); + return; + } + try { + res.write(openAIChunk); + } catch (writeError) { + this.logger.debug( + `[Request] Failed to write OpenAI chunk to stream: ${writeError.message}` + ); + return; + } } } } @@ -3640,6 +4191,39 @@ class RequestHandler { } } + /** + * Split a raw SSE accumulation into complete events and the trailing partial. + * Accepts both LF (`\n\n`) and CRLF (`\r\n\r\n`) framed events. + * @param {string} buffer accumulated raw SSE text + * @returns {{complete: string[], remainder: string}} complete event payloads (with `data:` lines) and leftover partial + */ + _extractSseEvents(buffer) { + const parts = buffer.split(/\r?\n\r?\n/); + const remainder = parts.pop(); + return { complete: parts, remainder }; + } + + /** + * Parse a single complete SSE event (may contain `data:` lines) and translate it via the given + * Google->target stream translator. + * @param {string} eventPayload raw SSE event text + * @param {string} [translatorName="translateGoogleToOpenAIStream"] FormatConverter stream translator to call + * @returns {string|null} target SSE chunk(s) or null if nothing to emit + */ + _translateCompleteSseEvent(eventPayload, model, streamState, translatorName = "translateGoogleToOpenAIStream") { + const trimmed = (eventPayload || "").trim(); + if (trimmed === "") return null; + // Extract the `data:` payload lines (SSE events may include `event:`/`id:`/`retry:` lines). + const dataLines = trimmed + .split("\n") + .filter(line => line.startsWith("data:")) + .map(line => line.slice(5).trim()); + if (dataLines.length === 0) return null; + // A single event may carry multiple `data:` lines (SSE spec: concatenated with \n). + const payload = dataLines.join("\n"); + return this.formatConverter[translatorName](payload, model, streamState); + } + async _sendOpenAIResponseAPINonStreamResponse(messageQueue, res, model, requestId, responseDefaults = {}) { let fullBody = ""; let receiving = true; @@ -3667,6 +4251,14 @@ class RequestHandler { // Parse and convert to OpenAI Response API format try { const googleResponse = JSON.parse(fullBody); + // Write a judged-empty-only correlation dump (the helper early-returns on non-empty), + // gated by DUMP_EMPTY_UPSTREAM. Non-empty upstream responses are not recorded here. + this._dumpUpstreamCorrelation("non-stream", fullBody, requestId, model, this.currentAuthIndex); + // Terminal emptiness judgment for the OpenAI Response API non-stream path. + if (this._isEmptyUpstreamResponse(googleResponse)) { + this._handleEmptyNonStreamResponse(res, requestId); + return; + } const responseAPIResponse = this.formatConverter.convertGoogleToResponseAPINonStream( googleResponse, model, @@ -3707,6 +4299,16 @@ class RequestHandler { // Parse and convert to OpenAI format try { const googleResponse = JSON.parse(fullBody); + // Write a judged-empty-only correlation dump (the helper early-returns on non-empty), + // gated by DUMP_EMPTY_UPSTREAM. Non-empty upstream responses are not recorded here. + this._dumpUpstreamCorrelation("non-stream", fullBody, requestId, model, this.currentAuthIndex); + // Terminal emptiness judgment for the non-stream path. The full upstream body is a single + // completed response — judge it now (whitespace-only/empty text with stop and ct=0 is + // empty, exactly as the stream path judges). Never leak an empty completion to the client. + if (this._isEmptyUpstreamResponse(googleResponse)) { + this._handleEmptyNonStreamResponse(res, requestId); + return; + } const openAIResponse = this.formatConverter.convertGoogleToOpenAINonStream(googleResponse, model); res.type("application/json").send(JSON.stringify(openAIResponse)); this.logger.info(`✅ [Request] Response completed (OpenAI non-stream), request ID: ${requestId}`); @@ -3716,6 +4318,21 @@ class RequestHandler { } } + _handleEmptyNonStreamResponse(res, requestId) { + this.logger.warn( + `⚠️ [Request] Upstream non-stream response judged empty (request ${requestId}); switching account and returning 502.` + ); + this._handleAuthFailure( + { + message: "Empty upstream response (non-stream)", + reason: "empty_upstream_response", + status: 502, + }, + requestId + ); + this._sendErrorResponse(res, 502, "Empty upstream response"); + } + _setResponseHeaders(res, headerMessage, req) { res.status(headerMessage.status || 200); const headers = headerMessage.headers || {}; diff --git a/src/utils/ConfigLoader.js b/src/utils/ConfigLoader.js index 1400fd82..5cc6ec9e 100644 --- a/src/utils/ConfigLoader.js +++ b/src/utils/ConfigLoader.js @@ -34,13 +34,13 @@ class ConfigLoader { forceWebSearch: false, host: "0.0.0.0", httpPort: 7860, - immediateSwitchStatusCodes: [429, 503], + immediateSwitchStatusCodes: [429, 502, 503, 403], maxContexts: 1, maxRetries: 3, retryDelay: 2000, safetySettingsThreshold: "OFF", streamingMode: "real", - streamTimeoutMs: 60000, + streamTimeoutMs: 0, switchOnUses: 40, wsPort: 9998, }; @@ -68,11 +68,9 @@ class ConfigLoader { const parsed = parseInt(process.env.RETRY_DELAY, 10); config.retryDelay = Number.isFinite(parsed) ? Math.max(50, parsed) : config.retryDelay; } - if (process.env.STREAM_TIMEOUT_MS) { + if (process.env.STREAM_TIMEOUT_MS !== undefined) { const parsed = parseInt(process.env.STREAM_TIMEOUT_MS, 10); - config.streamTimeoutMs = Number.isFinite(parsed) - ? Math.min(300000, Math.max(1, parsed)) - : config.streamTimeoutMs; + config.streamTimeoutMs = Number.isFinite(parsed) ? Math.min(300000, Math.max(0, parsed)) : 0; } if (process.env.FAKE_STREAM_TIMEOUT_MS) { const parsed = parseInt(process.env.FAKE_STREAM_TIMEOUT_MS, 10); diff --git a/src/utils/MessageQueue.js b/src/utils/MessageQueue.js index f813cb26..9436cd48 100644 --- a/src/utils/MessageQueue.js +++ b/src/utils/MessageQueue.js @@ -39,7 +39,8 @@ class MessageQueue extends EventEmitter { super(); this.messages = []; this.waitingResolvers = []; - this.defaultTimeout = timeoutMs; + this.defaultTimeout = + typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 ? Math.floor(timeoutMs) : 0; this.closed = false; this.closeReason = null; } @@ -49,8 +50,11 @@ class MessageQueue extends EventEmitter { if (this.waitingResolvers.length > 0) { const resolver = this.waitingResolvers.shift(); // Check if resolver is still valid (not timed out) - if (resolver && resolver.timeoutId) { - clearTimeout(resolver.timeoutId); + if (resolver && !resolver.timedOut) { + if (resolver.timeoutId) { + clearTimeout(resolver.timeoutId); + resolver.timeoutId = null; + } resolver.resolve(message); } else { // Resolver already timed out, push message to queue instead @@ -66,6 +70,10 @@ class MessageQueue extends EventEmitter { const reason = this.closeReason || "unknown"; throw new QueueClosedError(`Queue is closed (reason: ${reason})`, reason); } + + const effectiveTimeout = + typeof timeoutMs === "number" && Number.isFinite(timeoutMs) && timeoutMs > 0 ? Math.floor(timeoutMs) : 0; + return new Promise((resolve, reject) => { // Check if there are already queued messages if (this.messages.length > 0) { @@ -73,31 +81,29 @@ class MessageQueue extends EventEmitter { return; } - // Create resolver with timeout BEFORE pushing to waitingResolvers - // This prevents race condition where enqueue() sees timeoutId=null - const resolver = { reject, resolve, timeoutId: null }; + const resolver = { reject, resolve, timedOut: false, timeoutId: null }; - // Set timeout first to ensure resolver is fully initialized - resolver.timeoutId = setTimeout(() => { - const index = this.waitingResolvers.indexOf(resolver); - if (index !== -1) { - this.waitingResolvers.splice(index, 1); - } - // Clear timeoutId to mark resolver as invalid - resolver.timeoutId = null; - reject(new QueueTimeoutError()); - }, timeoutMs); + if (effectiveTimeout > 0) { + resolver.timeoutId = setTimeout(() => { + resolver.timedOut = true; + const index = this.waitingResolvers.indexOf(resolver); + if (index !== -1) { + this.waitingResolvers.splice(index, 1); + } + resolver.timeoutId = null; + reject(new QueueTimeoutError()); + }, effectiveTimeout); + } - // Now push to waitingResolvers - resolver is fully initialized this.waitingResolvers.push(resolver); - // CRITICAL: Check again if messages arrived during initialization - // This handles the race where enqueue() was called between the initial - // check (line 70) and push (line 89) + // Check again if messages arrived during initialization if (this.messages.length > 0 && this.waitingResolvers[0] === resolver) { - // We're still the first waiter, consume the message this.waitingResolvers.shift(); - clearTimeout(resolver.timeoutId); + if (resolver.timeoutId) { + clearTimeout(resolver.timeoutId); + resolver.timeoutId = null; + } resolve(this.messages.shift()); } }); @@ -107,7 +113,10 @@ class MessageQueue extends EventEmitter { this.closed = true; this.closeReason = reason; this.waitingResolvers.forEach(resolver => { - clearTimeout(resolver.timeoutId); + if (resolver.timeoutId) { + clearTimeout(resolver.timeoutId); + resolver.timeoutId = null; + } resolver.reject(new QueueClosedError(`Queue is closed (reason: ${reason})`, reason)); }); this.waitingResolvers = []; diff --git a/test/configLoader.test.js b/test/configLoader.test.js new file mode 100644 index 00000000..7fc180ec --- /dev/null +++ b/test/configLoader.test.js @@ -0,0 +1,102 @@ +const { test } = require("node:test"); +const assert = require("node:assert"); +const ConfigLoader = require("../src/utils/ConfigLoader"); + +test("ConfigLoader defaults streamTimeoutMs to 0", () => { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const loader = new ConfigLoader(logger); + + // Save and clear env var + const origEnv = process.env.STREAM_TIMEOUT_MS; + delete process.env.STREAM_TIMEOUT_MS; + + try { + const config = loader.loadConfiguration(); + assert.strictEqual(config.streamTimeoutMs, 0); + } finally { + if (origEnv !== undefined) { + process.env.STREAM_TIMEOUT_MS = origEnv; + } else { + delete process.env.STREAM_TIMEOUT_MS; + } + } +}); + +test("ConfigLoader parses positive STREAM_TIMEOUT_MS environment variable", () => { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const loader = new ConfigLoader(logger); + + const origEnv = process.env.STREAM_TIMEOUT_MS; + process.env.STREAM_TIMEOUT_MS = "60000"; + + try { + const config = loader.loadConfiguration(); + assert.strictEqual(config.streamTimeoutMs, 60000); + } finally { + if (origEnv !== undefined) { + process.env.STREAM_TIMEOUT_MS = origEnv; + } else { + delete process.env.STREAM_TIMEOUT_MS; + } + } +}); + +test("ConfigLoader clamps STREAM_TIMEOUT_MS above hard max to 300000", () => { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const loader = new ConfigLoader(logger); + + const origEnv = process.env.STREAM_TIMEOUT_MS; + process.env.STREAM_TIMEOUT_MS = "600000"; + + try { + const config = loader.loadConfiguration(); + assert.strictEqual(config.streamTimeoutMs, 300000); + } finally { + if (origEnv !== undefined) { + process.env.STREAM_TIMEOUT_MS = origEnv; + } else { + delete process.env.STREAM_TIMEOUT_MS; + } + } +}); + +test("ConfigLoader keeps explicit 0 STREAM_TIMEOUT_MS as 0 (disabled)", () => { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const loader = new ConfigLoader(logger); + + const origEnv = process.env.STREAM_TIMEOUT_MS; + process.env.STREAM_TIMEOUT_MS = "0"; + + try { + const config = loader.loadConfiguration(); + assert.strictEqual(config.streamTimeoutMs, 0); + } finally { + if (origEnv !== undefined) { + process.env.STREAM_TIMEOUT_MS = origEnv; + } else { + delete process.env.STREAM_TIMEOUT_MS; + } + } +}); + +test("ConfigLoader normalizes invalid or negative STREAM_TIMEOUT_MS to 0", () => { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const loader = new ConfigLoader(logger); + + const origEnv = process.env.STREAM_TIMEOUT_MS; + + const invalidInputs = ["-500", "invalid", "-1"]; + for (const input of invalidInputs) { + process.env.STREAM_TIMEOUT_MS = input; + try { + const config = loader.loadConfiguration(); + assert.strictEqual(config.streamTimeoutMs, 0, `Input '${input}' should normalize to 0`); + } finally { + if (origEnv !== undefined) { + process.env.STREAM_TIMEOUT_MS = origEnv; + } else { + delete process.env.STREAM_TIMEOUT_MS; + } + } + } +}); diff --git a/test/formatConverter.test.js b/test/formatConverter.test.js new file mode 100644 index 00000000..49808d86 --- /dev/null +++ b/test/formatConverter.test.js @@ -0,0 +1,623 @@ +"use strict"; +const { test } = require("node:test"); +const assert = require("node:assert"); +const path = require("path"); + +const FormatConverter = require(path.join(__dirname, "..", "src/core/FormatConverter.js")); + +const stubLogger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + +function makeConverter() { + return new FormatConverter(stubLogger, { + get config() { + return { forceThinking: false, thinkingLevel: null, webSearch: false }; + }, + }); +} + +// ---- toolCallId -> name mapping (OpenAI Chat -> Google, translateOpenAIToGoogle) ---- +test("translateOpenAIToGoogle maps tool_call_id via assistant tool_calls; missing -> unknown_function", async () => { + const fc = makeConverter(); + const body = { + messages: [ + { content: "weather?", role: "user" }, + { + content: null, + role: "assistant", + tool_calls: [{ function: { arguments: "{}", name: "get_weather" }, id: "call_1", type: "function" }], + }, + { content: "70", role: "tool", tool_call_id: "call_1" }, + { content: "x", role: "tool", tool_call_id: "missing_id" }, + { content: "y", name: "explicit_now", role: "tool", tool_call_id: "call_2" }, + ], + model: "gpt-4o", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + const fnParts = googleRequest.contents + .filter(c => c.parts && c.parts.some(p => p.functionResponse)) + .flatMap(c => c.parts.filter(p => p.functionResponse)); + const names = fnParts.map(p => p.functionResponse.name); + assert.deepStrictEqual(names, ["get_weather", "unknown_function", "explicit_now"]); +}); + +// ---- toolCallId -> name mapping (OpenAI Responses -> Google, translateOpenAIResponseToGoogle) ---- +test("translateOpenAIResponseToGoogle maps call_id via function_call; missing -> unknown_function; adds thoughtSignature", async () => { + const fc = makeConverter(); + const body = { + input: [ + { content: [{ text: "weather?", type: "input_text" }], role: "user", type: "message" }, + { arguments: "{}", call_id: "fc_1", name: "get_weather", type: "function_call" }, + { call_id: "fc_1", output: "70", type: "function_call_output" }, + { call_id: "missing", output: "x", type: "function_call_output" }, + { call_id: "fc_2", name: "explicit_now", output: "y", type: "function_call_output" }, + ], + model: "gpt-5", + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + const modelParts = googleRequest.contents + .filter(c => c.parts && c.parts.some(p => p.functionCall)) + .flatMap(c => c.parts.filter(p => p.functionCall)); + const fnCall = modelParts[0].functionCall; + assert.strictEqual(fnCall.name, "get_weather"); + assert.strictEqual(modelParts[0].thoughtSignature, FormatConverter.DUMMY_THOUGHT_SIGNATURE); + const fnParts = googleRequest.contents + .filter(c => c.parts && c.parts.some(p => p.functionResponse)) + .flatMap(c => c.parts.filter(p => p.functionResponse)); + const names = fnParts.map(p => p.functionResponse.name); + assert.deepStrictEqual(names, ["get_weather", "unknown_function", "explicit_now"]); +}); + +// ---- consecutive same-role merge (OpenAI Chat -> Google) ---- +test("translateOpenAIToGoogle merges consecutive tool messages into one user message", async () => { + const fc = makeConverter(); + const body = { + messages: [ + { + content: null, + role: "assistant", + tool_calls: [{ function: { arguments: "{}", name: "a" }, id: "c1", type: "function" }], + }, + { content: "1", role: "tool", tool_call_id: "c1" }, + { content: "2", role: "tool", tool_call_id: "c1" }, + ], + model: "gpt-4o", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + const userContents = googleRequest.contents.filter(c => c.role === "user"); + assert.strictEqual(userContents.length, 1, "consecutive tool messages should merge into a single user content"); + const partCount = userContents[0].parts.filter(p => p.functionResponse).length; + assert.strictEqual(partCount, 2); +}); + +test("translateOpenAIResponseToGoogle merges consecutive function_call_output into one user message", async () => { + const fc = makeConverter(); + const body = { + input: [ + { arguments: "{}", call_id: "fc_1", name: "get_weather", type: "function_call" }, + { call_id: "fc_1", output: "1", type: "function_call_output" }, + { call_id: "fc_1", output: "2", type: "function_call_output" }, + ], + model: "gpt-5", + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + const userContents = googleRequest.contents.filter(c => c.role === "user"); + assert.strictEqual( + userContents.length, + 1, + "consecutive function_call_output should merge into a single user content" + ); + const partCount = userContents[0].parts.filter(p => p.functionResponse).length; + assert.strictEqual(partCount, 2); +}); + +// ---- functionCall-with-thoughtSignature preserved (OpenAI Chat stream translator) ---- +test("translateGoogleToOpenAIStream preserves a functionCall part that carries thoughtSignature", () => { + const fc = makeConverter(); + const chunk = JSON.stringify({ + candidates: [ + { + content: { + parts: [{ functionCall: { args: { city: "SF" }, name: "get_weather" }, thoughtSignature: "sig" }], + }, + finishReason: "STOP", + }, + ], + }); + const out = fc.translateGoogleToOpenAIStream(chunk, "gemini-2.5-flash-lite", {}); + assert.ok(typeof out === "string", `expected string, got ${String(out)}`); + assert.ok(out.includes("tool_calls"), out.slice(0, 200)); + assert.ok(out.includes("get_weather"), out.slice(0, 200)); +}); + +// ---- functionCall-with-thoughtSignature preserved (OpenAI Chat non-stream converter) ---- +test("convertGoogleToOpenAINonStream preserves a functionCall part that carries thoughtSignature", () => { + const fc = makeConverter(); + const resp = { + candidates: [ + { + content: { + parts: [{ functionCall: { args: { city: "SF" }, name: "get_weather" }, thoughtSignature: "sig" }], + }, + finishReason: "STOP", + }, + ], + }; + const out = fc.convertGoogleToOpenAINonStream(resp, "gemini-2.5-flash-lite"); + const toolCalls = out.choices[0].message.tool_calls; + assert.ok(Array.isArray(toolCalls) && toolCalls.length === 1, JSON.stringify(out.choices[0].message)); + assert.strictEqual(toolCalls[0].function.name, "get_weather"); +}); + +// ---- OpenAI reasoning_effort fidelity and precedence ---- +test("translateOpenAIToGoogle maps reasoning_effort to thinkingLevel via THINKING_LEVEL_MAP", async () => { + const fc = makeConverter(); + const cases = [ + { effort: "minimal", expected: "MINIMAL" }, + { effort: "low", expected: "LOW" }, + { effort: "medium", expected: "MEDIUM" }, + { effort: "high", expected: "HIGH" }, + { effort: "MEDIUM", expected: "MEDIUM" }, + { effort: " LoW ", expected: "LOW" }, + ]; + + for (const { effort, expected } of cases) { + const body = { + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: effort, + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: expected, + }); + } +}); + +test("translateOpenAIToGoogle preserves model suffix precedence over body reasoning_effort", async () => { + const fc = makeConverter(); + const body = { + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-pro:thinking-high", + reasoning_effort: "low", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "HIGH", + }); +}); + +test("translateOpenAIToGoogle preserves native extra_body.google thinking_config precedence over reasoning_effort", async () => { + const fc = makeConverter(); + const body = { + extra_body: { + google: { + thinking_config: { + include_thoughts: true, + thinking_level: "HIGH", + }, + }, + }, + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: "low", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "HIGH", + }); +}); + +test("translateOpenAIToGoogle supports camelCase native thinkingConfig and thinkingBudget", async () => { + const fc = makeConverter(); + const body = { + extra_body: { + google: { + thinkingConfig: { + includeThoughts: true, + thinkingBudget: 1024, + thinkingLevel: "LOW", + }, + }, + }, + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingBudget: 1024, + thinkingLevel: "LOW", + }); +}); + +test("translateOpenAIToGoogle body reasoning_effort fills missing native level without overwriting explicit level", async () => { + const fc = makeConverter(); + + // Missing native level -> body reasoning_effort fills it + const bodyFill = { + extra_body: { + google: { + thinking_config: { + include_thoughts: true, + }, + }, + }, + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: "high", + }; + const { googleRequest: reqFill } = await fc.translateOpenAIToGoogle(bodyFill); + assert.deepStrictEqual(reqFill.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "HIGH", + }); + + // Explicit native level -> body reasoning_effort ignored + const bodyNoOverwrite = { + extra_body: { + google: { + thinking_config: { + include_thoughts: true, + thinking_level: "HIGH", + }, + }, + }, + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: "low", + }; + const { googleRequest: reqNoOverwrite } = await fc.translateOpenAIToGoogle(bodyNoOverwrite); + assert.deepStrictEqual(reqNoOverwrite.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "HIGH", + }); +}); + +test("translateOpenAIToGoogle preserves explicit include_thoughts:false and native thinking_level over reasoning_effort", async () => { + const fc = makeConverter(); + const body = { + extra_body: { + google: { + thinking_config: { + include_thoughts: false, + thinking_level: "LOW", + }, + }, + }, + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: "high", + }; + const { googleRequest } = await fc.translateOpenAIToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: false, + thinkingLevel: "LOW", + }); +}); + +test("translateOpenAIToGoogle fallback for unknown reasoning_effort or missing reasoning_effort", async () => { + const fc = makeConverter(); + + // Unknown effort -> includeThoughts: true, no invalid thinkingLevel + const unknownBody = { + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + reasoning_effort: "custom_unknown", + }; + const { googleRequest: reqUnknown } = await fc.translateOpenAIToGoogle(unknownBody); + assert.deepStrictEqual(reqUnknown.generationConfig.thinkingConfig, { + includeThoughts: true, + }); + + // No effort -> no thinkingConfig set + const noEffortBody = { + messages: [{ content: "hi", role: "user" }], + model: "gemini-2.5-flash", + }; + const { googleRequest: reqNoEffort } = await fc.translateOpenAIToGoogle(noEffortBody); + assert.strictEqual(reqNoEffort.generationConfig.thinkingConfig, undefined); +}); + +test("FormatConverter._parseUsage handles cachedContentTokenCount and coexistence with reasoning tokens", () => { + const fc = makeConverter(); + const cases = [ + { + expected: { + completion_tokens: 50, + completion_tokens_details: { image_tokens: 0, output_text_tokens: 50, reasoning_tokens: 0 }, + prompt_tokens: 100, + prompt_tokens_details: { cached_tokens: 40, text_tokens: 100, tool_tokens: 0 }, + total_tokens: 150, + }, + input: { + cachedContentTokenCount: 40, + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "1. nonzero cachedContentTokenCount", + }, + { + expected: { + completion_tokens: 50, + completion_tokens_details: { image_tokens: 0, output_text_tokens: 50, reasoning_tokens: 0 }, + prompt_tokens: 100, + prompt_tokens_details: { cached_tokens: 0, text_tokens: 100, tool_tokens: 0 }, + total_tokens: 150, + }, + input: { + cachedContentTokenCount: 0, + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "2. zero cachedContentTokenCount", + }, + { + expected: { + completion_tokens: 50, + completion_tokens_details: { image_tokens: 0, output_text_tokens: 50, reasoning_tokens: 0 }, + prompt_tokens: 100, + prompt_tokens_details: { cached_tokens: 0, text_tokens: 100, tool_tokens: 0 }, + total_tokens: 150, + }, + input: { candidatesTokenCount: 50, promptTokenCount: 100, totalTokenCount: 150 }, + name: "3. absent cachedContentTokenCount", + }, + { + expectedCached: 40, + input: { + cachedContentTokenCount: "40", + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "4a. string numeric cachedContentTokenCount", + }, + { + expectedCached: 0, + input: { + cachedContentTokenCount: "invalid", + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "4b. malformed string cachedContentTokenCount", + }, + { + expectedCached: 0, + input: { + cachedContentTokenCount: -10, + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "4c. negative cachedContentTokenCount", + }, + { + expectedCached: 0, + input: { + cachedContentTokenCount: NaN, + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "4d. NaN / Infinity cachedContentTokenCount", + }, + { + expectedCached: 0, + input: { + cachedContentTokenCount: true, + candidatesTokenCount: 50, + promptTokenCount: 100, + totalTokenCount: 150, + }, + name: "4e. boolean / object cachedContentTokenCount", + }, + { + expected: { + completion_tokens: 110, + completion_tokens_details: { image_tokens: 0, output_text_tokens: 80, reasoning_tokens: 30 }, + prompt_tokens: 200, + prompt_tokens_details: { cached_tokens: 50, text_tokens: 200, tool_tokens: 0 }, + total_tokens: 310, + }, + input: { + cachedContentTokenCount: 50, + candidatesTokenCount: 80, + promptTokenCount: 200, + thoughtsTokenCount: 30, + totalTokenCount: 310, + }, + name: "5 & 6. cached tokens coexisting with reasoning tokens without affecting totals", + }, + ]; + + for (const c of cases) { + const result = fc._parseUsage({ usageMetadata: c.input }); + if (c.expected) { + assert.deepStrictEqual(result, c.expected, `failed on ${c.name}`); + } else if (c.expectedCached !== undefined) { + assert.strictEqual(result.prompt_tokens_details.cached_tokens, c.expectedCached, `failed on ${c.name}`); + assert.strictEqual(result.prompt_tokens, 100, `prompt_tokens changed on ${c.name}`); + assert.strictEqual(result.total_tokens, 150, `total_tokens changed on ${c.name}`); + } + } +}); + +test("convertGoogleToOpenAINonStream includes cached_tokens in usage.prompt_tokens_details", () => { + const fc = makeConverter(); + const googleResponse = { + candidates: [{ content: { parts: [{ text: "Hello" }] }, finishReason: "STOP" }], + usageMetadata: { + cachedContentTokenCount: 8, + candidatesTokenCount: 5, + promptTokenCount: 10, + totalTokenCount: 15, + }, + }; + const res = fc.convertGoogleToOpenAINonStream(googleResponse, "gemini-2.5-flash"); + assert.deepStrictEqual(res.usage, { + completion_tokens: 5, + completion_tokens_details: { image_tokens: 0, output_text_tokens: 5, reasoning_tokens: 0 }, + prompt_tokens: 10, + prompt_tokens_details: { cached_tokens: 8, text_tokens: 10, tool_tokens: 0 }, + total_tokens: 15, + }); +}); + +test("translateGoogleToOpenAIStream includes cached_tokens in usage chunk", () => { + const fc = makeConverter(); + const chunk = JSON.stringify({ + candidates: [{ content: { parts: [{ text: "Hi" }] }, finishReason: "STOP" }], + usageMetadata: { + cachedContentTokenCount: 6, + candidatesTokenCount: 4, + promptTokenCount: 12, + totalTokenCount: 16, + }, + }); + const streamState = {}; + const out = fc.translateGoogleToOpenAIStream(chunk, "gemini-2.5-flash", streamState); + assert.strictEqual(streamState.usage.prompt_tokens_details.cached_tokens, 6); + assert.strictEqual(streamState.usage.prompt_tokens, 12); + assert.ok(out.includes('"cached_tokens":6'), "stream chunk string should include cached_tokens:6"); +}); + +test("convertGoogleToResponseAPINonStream includes cached_tokens under input_tokens_details", () => { + const fc = makeConverter(); + const googleResponse = { + candidates: [{ content: { parts: [{ text: "Resp" }] }, finishReason: "STOP" }], + usageMetadata: { + cachedContentTokenCount: 15, + candidatesTokenCount: 10, + promptTokenCount: 20, + totalTokenCount: 30, + }, + }; + const res = fc.convertGoogleToResponseAPINonStream(googleResponse, "gemini-2.5-flash"); + assert.deepStrictEqual(res.usage, { + input_tokens: 20, + input_tokens_details: { cached_tokens: 15 }, + output_tokens: 10, + output_tokens_details: { reasoning_tokens: 0 }, + total_tokens: 30, + }); +}); + +test("translateGoogleToResponseAPIStream includes cached_tokens under input_tokens_details", () => { + const fc = makeConverter(); + const streamState = {}; + const googleChunk = JSON.stringify({ + candidates: [{ content: { parts: [{ text: "Resp" }] }, finishReason: "STOP" }], + usageMetadata: { + cachedContentTokenCount: 18, + candidatesTokenCount: 12, + promptTokenCount: 25, + totalTokenCount: 37, + }, + }); + const result = fc.translateGoogleToResponseAPIStream(googleChunk, "gemini-2.5-flash", streamState); + assert.ok(result, "should return stream data"); + assert.strictEqual(streamState.usage.prompt_tokens_details.cached_tokens, 18); + assert.ok(result.includes('"cached_tokens":18'), "stream output string should include cached_tokens:18"); +}); + +test("usage outputs do not include invented Claude or prompt-cache resource fields", () => { + const fc = makeConverter(); + const googleResponse = { + candidates: [{ content: { parts: [{ text: "Test" }] }, finishReason: "STOP" }], + usageMetadata: { + cachedContentTokenCount: 7, + candidatesTokenCount: 5, + promptTokenCount: 10, + totalTokenCount: 15, + }, + }; + const chatRes = fc.convertGoogleToOpenAINonStream(googleResponse, "gemini-2.5-flash"); + const respRes = fc.convertGoogleToResponseAPINonStream(googleResponse, "gemini-2.5-flash"); + + const forbiddenFields = [ + "cache_read_input_tokens", + "cache_creation_input_tokens", + "prompt_cache_key", + "cache_key", + "cache_creation_tokens", + ]; + for (const field of forbiddenFields) { + assert.strictEqual(chatRes.usage[field], undefined); + assert.strictEqual(chatRes.usage.prompt_tokens_details[field], undefined); + assert.strictEqual(respRes.usage[field], undefined); + assert.strictEqual(respRes.usage.input_tokens_details[field], undefined); + } +}); + +// ---- Studio PR #228: Responses reasoning.effort mapping through THINKING_LEVEL_MAP ---- +test("translateOpenAIResponseToGoogle maps reasoning.effort to thinkingLevel via THINKING_LEVEL_MAP", async () => { + const fc = makeConverter(); + const cases = [ + { effort: "minimal", expected: "MINIMAL" }, + { effort: "low", expected: "LOW" }, + { effort: "medium", expected: "MEDIUM" }, + { effort: "high", expected: "HIGH" }, + { effort: " HIGH ", expected: "HIGH" }, + ]; + + for (const { effort, expected } of cases) { + const body = { + input: "hi", + model: "gemini-2.5-flash", + reasoning: { effort }, + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + assert.deepStrictEqual( + googleRequest.generationConfig.thinkingConfig, + { includeThoughts: true, thinkingLevel: expected }, + `reasoning.effort=${effort}` + ); + } +}); + +test("translateOpenAIResponseToGoogle supports top-level reasoning_effort alias", async () => { + const fc = makeConverter(); + const body = { + input: "hi", + model: "gemini-2.5-flash", + reasoning_effort: "low", + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "LOW", + }); +}); + +test("translateOpenAIResponseToGoogle preserves model suffix precedence over reasoning.effort", async () => { + const fc = makeConverter(); + const body = { + input: "hi", + model: "gemini-2.5-pro:thinking-high", + reasoning: { effort: "low" }, + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + thinkingLevel: "HIGH", + }); +}); + +test("translateOpenAIResponseToGoogle fallback for unknown reasoning.effort keeps includeThoughts only", async () => { + const fc = makeConverter(); + const body = { + input: "hi", + model: "gemini-2.5-flash", + reasoning: { effort: "custom_unknown" }, + }; + const { googleRequest } = await fc.translateOpenAIResponseToGoogle(body); + assert.deepStrictEqual(googleRequest.generationConfig.thinkingConfig, { + includeThoughts: true, + }); +}); diff --git a/test/messageQueue.test.js b/test/messageQueue.test.js new file mode 100644 index 00000000..a3f4a20c --- /dev/null +++ b/test/messageQueue.test.js @@ -0,0 +1,117 @@ +const { test } = require("node:test"); +const assert = require("node:assert"); +const MessageQueue = require("../src/utils/MessageQueue"); +const { QueueTimeoutError, QueueClosedError } = require("../src/utils/MessageQueue"); + +test("MessageQueue no-arg constructor retains a finite default timeout (300000ms)", async () => { + const queue = new MessageQueue(); + assert.strictEqual(queue.defaultTimeout, 300000, "no-arg must keep the finite 300000ms default"); + + // A no-arg dequeue applies the constructor default: with a small default it times out. + const smallQueue = new MessageQueue(10); + assert.strictEqual(smallQueue.defaultTimeout, 10); + await assert.rejects(async () => { + await smallQueue.dequeue(); + }, QueueTimeoutError); +}); + +test("MessageQueue explicit 0 remains unlimited (no-arg dequeue on a 0-constructed queue)", async () => { + const queue = new MessageQueue(0); + assert.strictEqual(queue.defaultTimeout, 0, "explicit 0 must stay unlimited"); + + let resolved = false; + const promise = queue.dequeue().then(msg => { + resolved = true; + return msg; + }); + await new Promise(r => setTimeout(r, 50)); + assert.strictEqual(resolved, false, "no-arg dequeue on explicit-0 queue must remain pending"); + + queue.enqueue("hello"); + assert.strictEqual(await promise, "hello"); +}); + +test("MessageQueue dequeue(0) remains pending until chunk is enqueued", async () => { + const queue = new MessageQueue(); + let resolved = false; + let result = null; + + const promise = queue.dequeue(0).then(msg => { + resolved = true; + result = msg; + }); + + // Wait short scheduling interval + await new Promise(r => setTimeout(r, 50)); + assert.strictEqual(resolved, false, "dequeue(0) should remain pending"); + + queue.enqueue("hello"); + await promise; + assert.strictEqual(resolved, true); + assert.strictEqual(result, "hello"); +}); + +test("MessageQueue invalid/negative/null/undefined timeouts default to 0 (no timeout)", async () => { + const invalidValues = [-100, null, undefined, NaN, "invalid"]; + + for (const val of invalidValues) { + const queue = new MessageQueue(); + let rejected = false; + + const promise = queue.dequeue(val).catch(err => { + rejected = true; + return err; + }); + + await new Promise(r => setTimeout(r, 20)); + assert.strictEqual(rejected, false, `dequeue(${val}) should not reject automatically`); + + queue.close("cleanup"); + const err = await promise; + assert.ok(err instanceof QueueClosedError); + assert.strictEqual(err.reason, "cleanup"); + } +}); + +test("MessageQueue explicit positive timeout rejects after deadline", async () => { + const queue = new MessageQueue(); + + const start = Date.now(); + await assert.rejects(async () => { + await queue.dequeue(40); + }, QueueTimeoutError); + + const elapsed = Date.now() - start; + assert.ok(elapsed >= 30, `Elapsed time should be near 40ms, was ${elapsed}ms`); +}); + +test("MessageQueue enqueue before positive timeout deadline clears timer and next dequeue gets fresh window", async () => { + const queue = new MessageQueue(); + + const dequeuePromise = queue.dequeue(100); + setTimeout(() => queue.enqueue("chunk1"), 20); + + const res1 = await dequeuePromise; + assert.strictEqual(res1, "chunk1"); + + // Next dequeue gets fresh window and resolves on chunk2 + const dequeuePromise2 = queue.dequeue(100); + setTimeout(() => queue.enqueue("chunk2"), 20); + + const res2 = await dequeuePromise2; + assert.strictEqual(res2, "chunk2"); +}); + +test("MessageQueue close('client_disconnect') rejects pending dequeue(0) promptly", async () => { + const queue = new MessageQueue(); + + const promise = queue.dequeue(0); + queue.close("client_disconnect"); + + await assert.rejects( + async () => { + await promise; + }, + err => err instanceof QueueClosedError && err.reason === "client_disconnect" + ); +}); diff --git a/test/requestHandler.test.js b/test/requestHandler.test.js new file mode 100644 index 00000000..6accfee9 --- /dev/null +++ b/test/requestHandler.test.js @@ -0,0 +1,1165 @@ +"use strict"; +const { test } = require("node:test"); +const assert = require("node:assert"); +const path = require("path"); + +const RequestHandler = require(path.join(__dirname, "..", "src/core/RequestHandler.js")); +const FormatConverter = require(path.join(__dirname, "..", "src/core/FormatConverter.js")); +const ConnectionRegistry = require(path.join(__dirname, "..", "src/core/ConnectionRegistry.js")); + +const stubLogger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + +function makeHandler() { + const rh = Object.create(RequestHandler.prototype); + rh.formatConverter = new FormatConverter(stubLogger, { + get config() { + return { forceThinking: false, thinkingLevel: null, webSearch: false }; + }, + }); + rh.timeouts = { STREAM_CHUNK: 60000 }; + return rh; +} + +test("_isEmptyUpstreamResponse: native Anthropic text and tool_use are not empty", () => { + const rh = makeHandler(); + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + content: [{ text: "hello", type: "text" }], + stop_reason: "end_turn", + usage: { output_tokens: 1 }, + }), + false + ); + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + content: [{ input: {}, name: "ping", type: "tool_use" }], + stop_reason: "tool_use", + usage: { output_tokens: 1 }, + }), + false + ); +}); + +test("_isEmptyUpstreamResponse: unknown Anthropic content blocks fail safe as content", () => { + const rh = makeHandler(); + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + content: [{ payload: { value: "future-output" }, type: "future_block" }], + stop_reason: "end_turn", + usage: { output_tokens: 0 }, + }), + false + ); +}); + +test("_isEmptyUpstreamResponse: native Anthropic empty terminal response is empty", () => { + const rh = makeHandler(); + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + content: [{ text: " ", type: "text" }], + stop_reason: "end_turn", + usage: { output_tokens: 0 }, + }), + true + ); + assert.strictEqual(rh._isEmptyUpstreamResponse({ content: [] }), false, "non-terminal headers stay pass-through"); +}); + +test("_withFailureAuthIndex resolves the request account without reading mutable current account", () => { + const rh = makeHandler(); + rh.connectionRegistry = { + getAuthIndexForRequest: requestId => (requestId === "req-source" ? 4 : null), + }; + const details = rh._withFailureAuthIndex({ status: 429 }, "req-source"); + assert.deepStrictEqual(details, { authIndex: 4, status: 429 }); + const explicit = rh._withFailureAuthIndex({ status: 502 }, "missing", 7); + assert.deepStrictEqual(explicit, { authIndex: 7, status: 502 }); +}); + +test("ConnectionRegistry routes each browser message with its source authIndex", () => { + const registry = Object.create(ConnectionRegistry.prototype); + const messages = []; + const queue = { close: () => {}, enqueue: message => messages.push(message) }; + registry._routeMessage({ data: { text: "x" }, event_type: "chunk" }, queue, 3); + registry._routeMessage({ event_type: "stream_close" }, queue, 5); + assert.strictEqual(messages[0].authIndex, 3); + assert.deepStrictEqual(messages[1], { authIndex: 5, type: "STREAM_END" }); +}); + +test("AuthSwitcher attributes concurrent failure counters to explicit source authIndex", async () => { + const AuthSwitcher = require(path.join(__dirname, "..", "src/auth/AuthSwitcher.js")); + const mockBrowser = { currentAuthIndex: 9 }; + const authSwitcher = new AuthSwitcher( + stubLogger, + { immediateSwitchStatusCodes: [502] }, + { getAuthCount: () => 10, getCanonicalIndex: i => i }, + mockBrowser + ); + let switchStartIndex; + let allowOriginalFallback; + authSwitcher.switchToNextAuth = async (failedAuthIndex, allowFallback) => { + switchStartIndex = failedAuthIndex; + allowOriginalFallback = allowFallback; + return { success: true }; + }; + + await authSwitcher.handleRequestFailureAndSwitch({ authIndex: 2, reason: "empty_upstream_response" }, null); + + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(2), 1); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.has(9), false); + assert.strictEqual(switchStartIndex, 2); + assert.strictEqual(allowOriginalFallback, false); +}); + +// ---- _extractSseEvents ---- +test("_extractSseEvents splits events on LF blank line", () => { + const rh = makeHandler(); + const { complete, remainder } = rh._extractSseEvents("data: a\n\ndata: b\n\n"); + assert.deepStrictEqual(complete, ["data: a", "data: b"]); + assert.strictEqual(remainder, ""); +}); + +test("_extractSseEvents splits events on CRLF blank line", () => { + const rh = makeHandler(); + const { complete, remainder } = rh._extractSseEvents("data: a\r\n\r\ndata: b\r\n\r\n"); + assert.deepStrictEqual(complete, ["data: a", "data: b"]); + assert.strictEqual(remainder, ""); +}); + +test("_extractSseEvents keeps a fragmented trailing event in remainder", () => { + const rh = makeHandler(); + const { complete, remainder } = rh._extractSseEvents("data: a\n\ndata: partial"); + assert.deepStrictEqual(complete, ["data: a"]); + assert.strictEqual(remainder, "data: partial"); +}); + +test("_extractSseEvents returns multiple complete events from one chunk", () => { + const rh = makeHandler(); + const { complete, remainder } = rh._extractSseEvents("data: a\n\ndata: b\n\ndata: c\n\n"); + assert.deepStrictEqual(complete, ["data: a", "data: b", "data: c"]); + assert.strictEqual(remainder, ""); +}); + +test("_extractSseEvents preserves a trailing CR in the remainder", () => { + const rh = makeHandler(); + const { complete, remainder } = rh._extractSseEvents("data: a\n\ndata: b\r"); + assert.deepStrictEqual(complete, ["data: a"]); + assert.strictEqual(remainder, "data: b\r"); +}); + +// ---- _translateCompleteSseEvent ---- +test("_translateCompleteSseEvent returns null for an empty event", () => { + const rh = makeHandler(); + assert.strictEqual(rh._translateCompleteSseEvent("", "gemini-2.5-flash-lite"), null); + assert.strictEqual(rh._translateCompleteSseEvent(" \n ", "gemini-2.5-flash-lite"), null); +}); + +test("_translateCompleteSseEvent skips an event with no data line", () => { + const rh = makeHandler(); + assert.strictEqual(rh._translateCompleteSseEvent("event: ping\nid: 1", "gemini-2.5-flash-lite"), null); +}); + +test("_translateCompleteSseEvent passes through [DONE]", () => { + const rh = makeHandler(); + const out = rh._translateCompleteSseEvent("data: [DONE]", "gemini-2.5-flash-lite"); + assert.ok(typeof out === "string" && out.includes("[DONE]"), `got ${String(out)}`); +}); + +test("_translateCompleteSseEvent skips a non-JSON data payload", () => { + const rh = makeHandler(); + assert.strictEqual(rh._translateCompleteSseEvent("data: not-json", "gemini-2.5-flash-lite"), null); +}); + +test("_translateCompleteSseEvent translates a valid JSON event and trims trailing CR", () => { + const rh = makeHandler(); + const payload = JSON.stringify({ candidates: [{ content: { parts: [{ text: "hello hi" }] } }] }); + const out = rh._translateCompleteSseEvent(`data: ${payload}\r`, "gemini-2.5-flash-lite"); + assert.ok(typeof out === "string" && out.includes("hello hi"), `got ${String(out)}`); +}); + +// ---- _isEmptyUpstreamResponse ---- +test("_isEmptyUpstreamResponse: pure tool call is NOT empty", () => { + const rh = makeHandler(); + const resp = { + candidates: [ + { + content: { parts: [{ functionCall: { args: {}, name: "get_weather" } }] }, + finishReason: "STOP", + }, + ], + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); + +test("_isEmptyUpstreamResponse: text content is NOT empty", () => { + const rh = makeHandler(); + const resp = { candidates: [{ content: { parts: [{ text: "hello" }] }, finishReason: "STOP" }] }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); + +test("_isEmptyUpstreamResponse: reasoning-only non-terminal chunk is NOT empty", () => { + const rh = makeHandler(); + const resp = { candidates: [{ content: { parts: [{ text: "hmm", thought: true }] } }] }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); + +test("_isEmptyUpstreamResponse: whitespace-only + STOP + zero completion tokens is EMPTY", () => { + const rh = makeHandler(); + const resp = { + candidates: [{ content: { parts: [{ text: " " }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), true); +}); + +test("_isEmptyUpstreamResponse: empty string + STOP is EMPTY", () => { + const rh = makeHandler(); + const resp = { candidates: [{ content: { parts: [] }, finishReason: "STOP" }] }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), true); +}); + +test("_isEmptyUpstreamResponse: whitespace text WITH completion tokens is NOT empty", () => { + const rh = makeHandler(); + const resp = { + candidates: [{ content: { parts: [{ text: " " }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 3, thoughtsTokenCount: 0 }, + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); +test("_isEmptyUpstreamResponse: terminal STOP with thought text part is NOT empty (thought parts count as content)", () => { + const rh = makeHandler(); + const resp = { + candidates: [{ content: { parts: [{ text: "hmm", thought: true }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 5 }, + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); + +test("_isEmptyUpstreamResponse: candidates:[] header frame is NOT empty (no terminal evidence)", () => { + const rh = makeHandler(); + assert.strictEqual(rh._isEmptyUpstreamResponse({ candidates: [] }), false); + assert.strictEqual( + rh._isEmptyUpstreamResponse({ candidates: [], usageMetadata: { promptTokenCount: 100 } }), + false + ); +}); + +test("_isEmptyUpstreamResponse: choices:[] usage-only frame is NOT empty", () => { + const rh = makeHandler(); + assert.strictEqual(rh._isEmptyUpstreamResponse({ choices: [] }), false); + assert.strictEqual(rh._isEmptyUpstreamResponse({ choices: [], usage: { prompt_tokens: 50 } }), false); +}); + +test("_isEmptyUpstreamResponse: blocked promptFeedback is NOT empty (pass through, no switch)", () => { + const rh = makeHandler(); + const resp = { + candidates: [], + promptFeedback: { blockReason: "SAFETY" }, + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); +test("_isEmptyUpstreamResponse: terminal STOP thinking-only with thoughtsTokenCount but no content is NOT empty", () => { + const rh = makeHandler(); + const resp = { + candidates: [{ content: { parts: [{ thought: true }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 3 }, + }; + assert.strictEqual(rh._isEmptyUpstreamResponse(resp), false); +}); + +// ---- OpenAI Response API fake stream: terminal empty upstream routes to switch+retry ---- +test("Response API fake stream: empty upstream body is judged and routed to switch+retry, not forwarded", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.config = { forceThinking: false, streamingMode: "fake", switchOnUses: 0, thinkingLevel: null }; + rh.needsSwitchingAfterRequest = false; + rh.timeouts = { FAKE_STREAM: 100 }; + + let switched = false; + let errorSent = false; + let dumped = false; + let translated = false; + + rh.authSwitcher = { + handleRequestFailureAndSwitch: async () => { + switched = true; + }, + incrementUsageCount: () => 0, + }; + + // Empty upstream: the tail queue delivers a STREAM_END with no content data, + // so the accumulated fullBody stays empty and must be judged terminal-empty. + const fakeQueue = { dequeue: async () => ({ type: "STREAM_END" }) }; + + rh.connectionRegistry = { + createMessageQueue: () => fakeQueue, + removeMessageQueue: () => {}, + }; + rh._generateRequestId = () => "test-fake-empty"; + rh._startTrackedRequest = () => {}; + rh._setResponseApiFormat = (res, fmt) => { + res.__responseApiFormat = fmt; + }; + rh._ensureBrowserBackedRequestReady = async () => true; + rh._setupClientDisconnectHandler = () => {}; + rh._initializeProxyRequestAttempt = () => {}; + rh._updateTrackedRequest = () => {}; + rh._getUsageStatsService = () => null; + rh._executeRequestWithRetries = async () => ({ queue: fakeQueue, success: true }); + rh._forwardRequest = async () => {}; + rh._dumpUpstreamCorrelation = () => { + dumped = true; + }; + rh._handleRequestError = () => { + errorSent = true; + }; + rh._finalizeTrackedRequest = () => {}; + rh._isResponseWritable = () => true; + rh._handleQueueTimeout = () => {}; + + // Must NOT be reached: an empty upstream must not translate into a client stream. + rh.formatConverter.translateGoogleToResponseAPIStream = () => { + translated = true; + }; + // Translate the outgoing OpenAI Responses request into Gemini deterministically. + rh.formatConverter.translateOpenAIResponseToGoogle = () => ({ + cleanModelName: "gemini-2.5-flash", + googleRequest: { contents: [{ parts: [{ text: "hi" }], role: "user" }] }, + modelStreamingMode: null, + }); + + const res = { + __responseApiSeq: null, + end: () => { + res.writableEnded = true; + }, + headersSent: false, + status: () => ({ set: () => {} }), + writableEnded: false, + write: () => true, + }; + const req = { + body: { input: "hi", model: "gpt-4o-mini", stream: true }, + headers: {}, + method: "POST", + protocol: "http", + url: "/v1/responses", + }; + + // The fake-stream keep-alive timer (12-18s) is left pending after the request finishes and + // would hold the test runner's event loop open. Replace long timers with an immediate no-op. + const realSetTimeout = global.setTimeout; + global.setTimeout = (fn, ms, ...args) => + ms >= 1000 ? realSetTimeout(() => {}, 0, ...args) : realSetTimeout(fn, ms, ...args); + try { + await rh.processOpenAIResponseRequest(req, res); + } finally { + global.setTimeout = realSetTimeout; + } + + assert.strictEqual(switched, true, "empty upstream fake stream must route to account switch + retry"); + assert.strictEqual(errorSent, true, "empty upstream fake stream must send an error to the client"); + assert.strictEqual(dumped, true, "empty upstream fake stream must write a correlation dump"); + assert.strictEqual( + translated, + false, + "empty upstream fake stream must not translate/send an empty stream to the client" + ); +}); + +// ---- OpenAI chat fake stream: terminal empty upstream routes to switch+retry ---- +test("OpenAI chat fake stream: empty upstream body is judged and routed to switch+retry, not leaked", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.config = { forceThinking: false, streamingMode: "fake", switchOnUses: 0, thinkingLevel: null }; + rh.needsSwitchingAfterRequest = false; + rh.timeouts = { FAKE_STREAM: 100 }; + + let switched = false; + let switchCount = 0; + let errorSent = false; + let dumped = false; + let translated = false; + + rh.authSwitcher = { + handleRequestFailureAndSwitch: async () => { + switchCount++; + switched = true; + }, + incrementUsageCount: () => 0, + }; + + // Empty upstream: whitespace-only STOP body with zero completion tokens accumulated in fullBody. + const emptyBody = JSON.stringify({ + candidates: [{ content: { parts: [{ text: " " }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }); + let dequeues = 0; + const fakeQueue = { + dequeue: async () => { + dequeues++; + if (dequeues === 1) { + return { data: emptyBody, event_type: "chunk" }; + } + return { type: "STREAM_END" }; + }, + }; + + rh.connectionRegistry = { + createMessageQueue: () => fakeQueue, + removeMessageQueue: () => {}, + }; + rh._generateRequestId = () => "test-openai-chat-fake-empty"; + rh._startTrackedRequest = () => {}; + rh._setResponseApiFormat = (res, fmt) => { + res.__responseApiFormat = fmt; + }; + rh._ensureBrowserBackedRequestReady = async () => true; + rh._setupClientDisconnectHandler = () => {}; + rh._initializeProxyRequestAttempt = () => {}; + rh._updateTrackedRequest = () => {}; + rh._getUsageStatsService = () => null; + rh._executeRequestWithRetries = async () => ({ queue: fakeQueue, success: true }); + rh._forwardRequest = async () => {}; + rh._dumpUpstreamCorrelation = () => { + dumped = true; + }; + // _handleRequestError may be the SSE-error path; deflect it to prevent actual writes. + rh._handleRequestError = () => { + errorSent = true; + }; + rh._finalizeTrackedRequest = () => {}; + rh._isResponseWritable = () => true; + rh._handleQueueTimeout = () => {}; + + // Must NOT be reached: an empty upstream must not translate into a client stream. + rh.formatConverter.translateGoogleToOpenAIStream = () => { + translated = true; + }; + // Translate the outgoing OpenAI chat request into Gemini deterministically. + rh.formatConverter.translateOpenAIToGoogle = () => ({ + cleanModelName: "gemini-2.5-flash", + googleRequest: { contents: [{ parts: [{ text: "hi" }], role: "user" }] }, + modelStreamingMode: null, + }); + + const res = { + end: () => { + res.writableEnded = true; + }, + headersSent: false, + status: () => ({ set: () => {} }), + writableEnded: false, + write: () => true, + }; + const req = { + body: { messages: [{ content: "hi", role: "user" }], model: "gpt-4o-mini", stream: true }, + headers: {}, + method: "POST", + protocol: "http", + url: "/v1/chat/completions", + }; + + // The fake-stream keep-alive timer is left pending after the request finishes and would hold + // the test runner's event loop open. Replace long timers with an immediate no-op. + const realSetTimeout = global.setTimeout; + global.setTimeout = (fn, ms, ...args) => + ms >= 1000 ? realSetTimeout(() => {}, 0, ...args) : realSetTimeout(fn, ms, ...args); + try { + await rh.processOpenAIRequest(req, res); + } finally { + global.setTimeout = realSetTimeout; + } + + assert.strictEqual(switched, true, "empty upstream fake stream must route to account switch + retry"); + assert.strictEqual(switchCount, 1, "empty upstream fake stream must trigger exactly one auth switch"); + assert.strictEqual(errorSent, true, "empty upstream fake stream must send an error to the client"); + assert.strictEqual(dumped, true, "empty upstream fake stream must write a correlation dump"); + assert.strictEqual( + translated, + false, + "empty upstream fake stream must not translate/send an empty completion to the client" + ); +}); + +// ---- OpenAI chat fake stream: non-empty upstream still translates (no behavior change) ---- +test("OpenAI chat fake stream: non-empty upstream still translates to the client", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.config = { forceThinking: false, streamingMode: "fake", switchOnUses: 0, thinkingLevel: null }; + rh.needsSwitchingAfterRequest = false; + rh.timeouts = { FAKE_STREAM: 100 }; + + let switchCount = 0; + const written = []; + let translatedChunk = "data: {}\n\n"; + + rh.authSwitcher = { + handleRequestFailureAndSwitch: async () => { + switchCount++; + }, + incrementUsageCount: () => 0, + }; + + const nonEmptyBody = JSON.stringify({ + candidates: [{ content: { parts: [{ text: "hello" }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 5, thoughtsTokenCount: 0 }, + }); + const fakeQueue = { + dequeue: async () => { + if (!fakeQueue._sent) { + fakeQueue._sent = true; + return { data: nonEmptyBody, event_type: "chunk" }; + } + return { type: "STREAM_END" }; + }, + }; + + rh.connectionRegistry = { + createMessageQueue: () => fakeQueue, + removeMessageQueue: () => {}, + }; + rh._generateRequestId = () => "test-openai-chat-fake-nonempty"; + rh._startTrackedRequest = () => {}; + rh._setResponseApiFormat = () => {}; + rh._ensureBrowserBackedRequestReady = async () => true; + rh._setupClientDisconnectHandler = () => {}; + rh._initializeProxyRequestAttempt = () => {}; + rh._updateTrackedRequest = () => {}; + rh._getUsageStatsService = () => null; + rh._executeRequestWithRetries = async () => ({ queue: fakeQueue, success: true }); + rh._forwardRequest = async () => {}; + rh._dumpUpstreamCorrelation = () => {}; + rh._handleRequestError = () => {}; + rh._finalizeTrackedRequest = () => {}; + rh._isResponseWritable = () => true; + rh._handleQueueTimeout = () => {}; + + rh.formatConverter.translateGoogleToOpenAIStream = fullBody => { + translatedChunk = `data: ${JSON.stringify({ content: fullBody })}\n\n`; + return translatedChunk; + }; + rh.formatConverter.translateOpenAIToGoogle = () => ({ + cleanModelName: "gemini-2.5-flash", + googleRequest: { contents: [{ parts: [{ text: "hi" }], role: "user" }] }, + modelStreamingMode: null, + }); + + const res = { + end: () => { + res.writableEnded = true; + }, + headersSent: false, + status: () => ({ set: () => {} }), + writableEnded: false, + write: chunk => { + written.push(chunk); + return true; + }, + }; + const req = { + body: { messages: [{ content: "hi", role: "user" }], model: "gpt-4o-mini", stream: true }, + headers: {}, + method: "POST", + protocol: "http", + url: "/v1/chat/completions", + }; + + const realSetTimeout = global.setTimeout; + global.setTimeout = (fn, ms, ...args) => + ms >= 1000 ? realSetTimeout(() => {}, 0, ...args) : realSetTimeout(fn, ms, ...args); + try { + await rh.processOpenAIRequest(req, res); + } finally { + global.setTimeout = realSetTimeout; + } + + assert.strictEqual(switchCount, 0, "non-empty upstream must not switch accounts"); + assert.ok( + written.some(chunk => chunk.includes("data: ")), + "translated stream must be written to the client" + ); +}); + +// ---- Regression tests for Items 1, 2, 3, 4 ---- +test("Item 1: _dumpUpstreamCorrelation is only called for event_type === 'chunk' with defined data", () => { + const rh = makeHandler(); + let dumpCalled = false; + rh._dumpUpstreamCorrelation = () => { + dumpCalled = true; + }; + + // response_headers frame (no data) must NOT call _dumpUpstreamCorrelation + const headerMsg = { event_type: "response_headers", headers: {} }; + if (headerMsg?.event_type === "chunk" && headerMsg.data !== undefined) { + rh._dumpUpstreamCorrelation("test", headerMsg.data, "req-1", "m", 0); + } + assert.strictEqual(dumpCalled, false); + + // chunk frame with data MUST call _dumpUpstreamCorrelation + const chunkMsg = { data: { foo: "bar" }, event_type: "chunk" }; + if (chunkMsg?.event_type === "chunk" && chunkMsg.data !== undefined) { + rh._dumpUpstreamCorrelation("test", chunkMsg.data, "req-1", "m", 0); + } + assert.strictEqual(dumpCalled, true); +}); + +test("Item 2: _streamOpenAIResponse uses _sendErrorChunkToClient when headers already sent", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.authSwitcher = { handleRequestFailureAndSwitch: () => {} }; + + let sseErrorSent = false; + let jsonErrorSent = false; + + rh._sendErrorChunkToClient = () => { + sseErrorSent = true; + }; + rh._sendErrorResponse = () => { + jsonErrorSent = true; + }; + + const fakeQueue = { + dequeue: async () => ({ type: "STREAM_END" }), + }; + + const res = { + end: () => {}, + headersSent: true, + writableEnded: false, + }; + + await rh._streamOpenAIResponse(fakeQueue, res, "gpt-4o", "req-stream-err"); + + assert.strictEqual(sseErrorSent, true, "must call _sendErrorChunkToClient when res.headersSent is true"); + assert.strictEqual(jsonErrorSent, false, "must NOT call _sendErrorResponse when res.headersSent is true"); +}); + +test("Item 3: AuthSwitcher deletes only failing index on non-empty failure, preserving other accounts", async () => { + const AuthSwitcher = require(path.join(__dirname, "..", "src/auth/AuthSwitcher.js")); + const mockBrowser = { currentAuthIndex: 0 }; + const authSwitcher = new AuthSwitcher( + stubLogger, + { immediateSwitchStatusCodes: [401, 403, 429, 500, 502, 503] }, + { getAuthCount: () => 3, getCanonicalIndex: i => i }, + mockBrowser + ); + authSwitcher.switchToNextAuth = async () => ({ success: true }); + + // Simulate empty failure on account 0 and account 1 + mockBrowser.currentAuthIndex = 0; + await authSwitcher.handleRequestFailureAndSwitch({ reason: "empty_upstream_response" }, null); + + mockBrowser.currentAuthIndex = 1; + await authSwitcher.handleRequestFailureAndSwitch({ reason: "empty_upstream_response" }, null); + + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(0), 1); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(1), 1); + + // Non-empty failure on account 1 should delete account 1 counter ONLY + mockBrowser.currentAuthIndex = 1; + await authSwitcher.handleRequestFailureAndSwitch({ reason: "rate_limit", status: 429 }, null); + + assert.strictEqual(authSwitcher._emptyJudgmentCounts.has(1), false, "account 1 counter deleted"); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(0), 1, "account 0 counter preserved"); +}); + +test("Item 4: FormatConverter.mergeConsecutiveSameRoleContents merges same roles", () => { + const googleContents = [ + { parts: [{ text: "a" }], role: "user" }, + { parts: [{ text: "b" }], role: "user" }, + { parts: [{ text: "c" }], role: "model" }, + ]; + const merged = FormatConverter.mergeConsecutiveSameRoleContents(googleContents); + assert.strictEqual(merged.length, 2); + assert.strictEqual(merged[0].role, "user"); + assert.deepStrictEqual(merged[0].parts, [{ text: "a" }, { text: "b" }]); + assert.strictEqual(merged[1].role, "model"); +}); + +// ---- Studio PR #228 adjudication regressions ---- + +// Fix 2: control finish reasons are valid non-empty results even with zero completion tokens. +test("_isEmptyUpstreamResponse: Gemini control finish reasons are NOT empty with zero tokens", () => { + const rh = makeHandler(); + for (const reason of ["SAFETY", "RECITATION", "BLOCKLIST", "PROHIBITED_CONTENT", "IMAGE_SAFETY"]) { + const resp = { + candidates: [{ content: { parts: [] }, finishReason: reason }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }; + assert.strictEqual( + rh._isEmptyUpstreamResponse(resp), + false, + `finishReason ${reason} must be a valid non-empty control result` + ); + } + // STOP with whitespace + zero tokens stays empty. + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + candidates: [{ content: { parts: [{ text: " " }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }), + true + ); +}); + +test("_isEmptyUpstreamResponse: unknown/OTHER finish reasons are not exempted as controls", () => { + const rh = makeHandler(); + // An OTHER reason with zero content and zero tokens is still empty — do not exempt arbitrary reasons. + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + candidates: [{ content: { parts: [] }, finishReason: "OTHER" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }), + true + ); +}); + +test("_isEmptyUpstreamResponse: OpenAI content_filter finish is NOT empty with zero tokens", () => { + const rh = makeHandler(); + for (const reason of ["content_filter", "safety"]) { + assert.strictEqual( + rh._isEmptyUpstreamResponse({ + choices: [{ finish_reason: reason, message: { content: "" } }], + usage: { completion_tokens: 0 }, + }), + false, + `finish_reason ${reason} must be a valid control result` + ); + } +}); + +// Fix 1: STREAM_END must flush a trailing partial SSE event before classifying empty, and only then +// emit one auth-failure + SSE error (via _sendErrorChunkToClient when headers already sent). +test("_streamClaudeResponse: true-empty STREAM_END emits one switch and one SSE error", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + let switchCount = 0; + let sseErrorCount = 0; + let jsonErrorCount = 0; + rh.authSwitcher = { + failureCount: 0, + handleRequestFailureAndSwitch: async () => { + switchCount++; + }, + }; + rh._handleAuthFailure = async () => { + switchCount++; + }; + rh._sendErrorChunkToClient = () => { + sseErrorCount++; + }; + rh._sendErrorResponse = () => { + jsonErrorCount++; + }; + rh._isResponseWritable = () => true; + rh._translateCompleteSseEvent = () => null; + + const fakeQueue = { + dequeue: async () => ({ type: "STREAM_END" }), + }; + const res = { headersSent: true, writableEnded: false, write: () => true }; + + await rh._streamClaudeResponse(fakeQueue, res, "claude-3-5-sonnet", "req-claude-empty"); + + assert.strictEqual(switchCount, 1, "exactly one auth switch for a true-empty Claude stream"); + assert.strictEqual(sseErrorCount, 1, "one SSE error sent when headers already sent"); + assert.strictEqual(jsonErrorCount, 0, "no silent _sendErrorResponse no-op"); +}); + +test("_streamClaudeResponse: fragmented final event is flushed and NOT judged empty", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + let switchCount = 0; + let sseErrorCount = 0; + const written = []; + rh._handleAuthFailure = async () => { + switchCount++; + }; + rh._sendErrorChunkToClient = () => { + sseErrorCount++; + }; + rh._sendErrorResponse = () => {}; + rh._isResponseWritable = () => true; + // A fragmented final SSE event reassembles in the buffer and translates to real output. + rh._translateCompleteSseEvent = () => "event: content_block_delta\ndata: {}\n\n"; + const res = { + headersSent: true, + writableEnded: false, + write: chunk => { + written.push(chunk); + return true; + }, + }; + // Drive the stream: first a partial chunk, then STREAM_END. + const partialPayload = 'data: {"candidates":[{"content":{"parts":[{"text":"partial"}]}}]}'; + const partialQueue = { + dequeue: async () => { + if (!partialQueue._sent) { + partialQueue._sent = true; + return { data: partialPayload, type: "chunk" }; + } + return { type: "STREAM_END" }; + }, + }; + await rh._streamClaudeResponse(partialQueue, res, "claude-3-5-sonnet", "req-claude-flush"); + + assert.strictEqual(switchCount, 0, "fragmented final event flush must NOT trigger empty judgment"); + assert.strictEqual(sseErrorCount, 0, "no SSE error when the flush produced output"); + assert.ok(written.length > 0, "the fragmented final event must be flushed to the client"); +}); + +test("_streamOpenAIResponseAPIResponse: true-empty STREAM_END emits one switch and one SSE error", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + let switchCount = 0; + let sseErrorCount = 0; + let jsonErrorCount = 0; + rh._handleAuthFailure = async () => { + switchCount++; + }; + rh._sendErrorChunkToClient = () => { + sseErrorCount++; + }; + rh._sendErrorResponse = () => { + jsonErrorCount++; + }; + rh._isResponseWritable = () => true; + rh._translateCompleteSseEvent = () => null; + + const fakeQueue = { + dequeue: async () => ({ type: "STREAM_END" }), + }; + const res = { + __responseApiSeq: null, + headersSent: true, + writableEnded: false, + write: () => true, + }; + + await rh._streamOpenAIResponseAPIResponse(fakeQueue, res, "gpt-5", { + requestId: "req-resp-empty", + responseDefaults: {}, + }); + + assert.strictEqual(switchCount, 1, "exactly one auth switch for a true-empty Responses stream"); + assert.strictEqual(sseErrorCount, 1, "one SSE error sent when headers already sent"); + assert.strictEqual(jsonErrorCount, 0, "no silent _sendErrorResponse no-op"); +}); + +test("_streamOpenAIResponse: true-empty STREAM_END emits one switch and one SSE error", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + let switchCount = 0; + let sseErrorCount = 0; + let jsonErrorCount = 0; + rh._handleAuthFailure = async () => { + switchCount++; + }; + rh._sendErrorChunkToClient = () => { + sseErrorCount++; + }; + rh._sendErrorResponse = () => { + jsonErrorCount++; + }; + rh._isResponseWritable = () => true; + rh._translateCompleteSseEvent = () => null; + + const fakeQueue = { + dequeue: async () => ({ type: "STREAM_END" }), + }; + const res = { headersSent: true, writableEnded: false, write: () => true }; + + await rh._streamOpenAIResponse(fakeQueue, res, "gpt-4o", "req-openai-empty"); + + assert.strictEqual(switchCount, 1, "exactly one auth switch for a true-empty OpenAI stream"); + assert.strictEqual(sseErrorCount, 1, "one SSE error sent when headers already sent"); + assert.strictEqual(jsonErrorCount, 0, "no silent _sendErrorResponse no-op"); +}); + +// Fix 3: Claude fake-stream aggregate terminal-empty must enter the existing single auth-failure + SSE error path. +test("Claude fake stream: empty aggregate body is judged and routed to switch+retry, not translated", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.config = { forceThinking: false, streamingMode: "fake", switchOnUses: 0, thinkingLevel: null }; + rh.needsSwitchingAfterRequest = false; + rh.timeouts = { FAKE_STREAM: 100 }; + + let switched = false; + let errorSent = false; + let dumped = false; + let translated = false; + + rh.authSwitcher = { + handleRequestFailureAndSwitch: async () => { + switched = true; + }, + incrementUsageCount: () => 0, + }; + + const fakeQueue = { dequeue: async () => ({ type: "STREAM_END" }) }; + + rh.connectionRegistry = { + createMessageQueue: () => fakeQueue, + removeMessageQueue: () => {}, + }; + rh._generateRequestId = () => "test-claude-fake-empty"; + rh._startTrackedRequest = () => {}; + rh._setResponseApiFormat = (res, fmt) => { + res.__responseApiFormat = fmt; + }; + rh._ensureBrowserBackedRequestReady = async () => true; + rh._setupClientDisconnectHandler = () => {}; + rh._initializeProxyRequestAttempt = () => {}; + rh._updateTrackedRequest = () => {}; + rh._getUsageStatsService = () => null; + rh._executeRequestWithRetries = async () => ({ queue: fakeQueue, success: true }); + rh._forwardRequest = async () => {}; + rh._dumpUpstreamCorrelation = () => { + dumped = true; + }; + rh._handleRequestError = () => { + errorSent = true; + }; + rh._finalizeTrackedRequest = () => {}; + rh._isResponseWritable = () => true; + rh._handleQueueTimeout = () => {}; + + rh.formatConverter.translateGoogleToClaudeStream = () => { + translated = true; + }; + rh.formatConverter.translateClaudeToGoogle = () => ({ + cleanModelName: "gemini-2.5-flash", + googleRequest: { contents: [{ parts: [{ text: "hi" }], role: "user" }] }, + modelStreamingMode: null, + }); + + const res = { + end: () => { + res.writableEnded = true; + }, + headersSent: false, + status: () => ({ set: () => {} }), + writableEnded: false, + write: () => true, + }; + const req = { + body: { messages: [{ content: "hi", role: "user" }], model: "claude-3-5-sonnet", stream: true }, + headers: {}, + method: "POST", + protocol: "http", + url: "/v1/messages", + }; + + const realSetTimeout = global.setTimeout; + global.setTimeout = (fn, ms, ...args) => + ms >= 1000 ? realSetTimeout(() => {}, 0, ...args) : realSetTimeout(fn, ms, ...args); + try { + await rh.processClaudeRequest(req, res); + } finally { + global.setTimeout = realSetTimeout; + } + + assert.strictEqual(switched, true, "empty Claude fake stream must route to account switch + retry"); + assert.strictEqual(errorSent, true, "empty Claude fake stream must send an error to the client"); + assert.strictEqual(dumped, true, "empty Claude fake stream must write a correlation dump"); + assert.strictEqual( + translated, + false, + "empty Claude fake stream must not translate/send an empty stream to the client" + ); +}); + +// Fix 4: OpenAI Responses real-stream initial complete-empty chunk converts into the existing error/retry flow. +test("Response API real stream: initial complete-empty chunk is converted to error/retry flow", async () => { + const rh = makeHandler(); + rh.logger = stubLogger; + rh.config = { forceThinking: false, immediateSwitchStatusCodes: [502], maxRetries: 0, streamingMode: "real" }; + rh.timeouts = { FAKE_STREAM: 100, STREAM_CHUNK: 100 }; + + let switchCount = 0; + let forwarded = 0; + const rh2 = Object.create(RequestHandler.prototype); + Object.assign(rh2, rh); + rh2.authSwitcher = { + failureCount: 0, + handleRequestFailureAndSwitch: async () => { + switchCount++; + }, + incrementUsageCount: () => 0, + resetEmptyJudgmentCountForAuth: () => {}, + }; + rh2._handleAuthFailure = async () => { + switchCount++; + }; + rh2._withFailureAuthIndex = d => d; + rh2._isResponseWritable = () => true; + rh2._cancelCurrentAttemptBeforeRetry = () => {}; + rh2._logFinalRequestFailure = () => {}; + rh2._sendErrorResponse = () => {}; + rh2._isConnectionResetError = () => false; + // Emulate the real immediate-switch retry: perform the account switch and continue with a new queue. + rh2._prepareImmediateStatusRetry = async () => { + await rh2._handleAuthFailure({ message: "empty", status: 502 }, "req", null, 0); + return true; + }; + rh2._dumpUpstreamCorrelation = () => {}; + rh2._forwardRequest = async () => { + forwarded++; + }; + rh2._advanceProxyRequestAttempt = () => {}; + rh2._initializeProxyRequestAttempt = () => {}; + rh2._setupClientDisconnectHandler = () => {}; + rh2._generateRequestId = () => "req-resp-initial-empty"; + rh2._startTrackedRequest = () => {}; + rh2._setResponseApiFormat = () => {}; + rh2._updateTrackedRequest = () => {}; + rh2._getUsageStatsService = () => null; + rh2._ensureBrowserBackedRequestReady = async () => true; + const emptyPayload = JSON.stringify({ + candidates: [{ content: { parts: [] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 0, thoughtsTokenCount: 0 }, + }); + const nonEmptyPayload = JSON.stringify({ + candidates: [{ content: { parts: [{ text: "hello" }] }, finishReason: "STOP" }], + usageMetadata: { candidatesTokenCount: 5, thoughtsTokenCount: 0 }, + }); + // First queue yields the terminal-empty initial chunk; the post-switch queue yields real content. + let queueCalls = 0; + rh2.connectionRegistry = { + createMessageQueue: () => ({ + close: () => {}, + dequeue: async () => { + queueCalls++; + if (queueCalls === 1) { + return { data: emptyPayload, event_type: "chunk" }; + } + if (queueCalls === 2) { + return { data: nonEmptyPayload, event_type: "chunk" }; + } + if (queueCalls === 3) { + return { data: `data: ${nonEmptyPayload}`, event_type: "chunk" }; + } + return { type: "STREAM_END" }; + }, + }), + getAuthIndexForRequest: () => 0, + removeMessageQueue: () => {}, + }; + rh2.formatConverter = { + translateGoogleToResponseAPIStream: (chunk, model, streamState) => { + // The real translator sets responseSent once it processes a candidate with content. + streamState.responseSent = true; + return "data: {}\n\n"; + }, + translateOpenAIResponseToGoogle: () => ({ + cleanModelName: "gemini-2.5-flash", + googleRequest: {}, + modelStreamingMode: null, + }), + }; + rh2._finalizeTrackedRequest = () => {}; + rh2._handleQueueTimeout = () => {}; + + const res = { + end: () => { + res.writableEnded = true; + }, + headersSent: false, + status: () => ({ set: () => {} }), + writableEnded: false, + write: () => true, + }; + const req = { + body: { input: "hi", model: "gpt-5", stream: true }, + headers: {}, + method: "POST", + protocol: "http", + url: "/v1/responses", + }; + + const realSetTimeout = global.setTimeout; + global.setTimeout = (fn, ms, ...args) => + ms >= 1000 ? realSetTimeout(() => {}, 0, ...args) : realSetTimeout(fn, ms, ...args); + try { + await rh2.processOpenAIResponseRequest(req, res); + } finally { + global.setTimeout = realSetTimeout; + } + + assert.strictEqual(switchCount, 1, "initial complete-empty chunk must trigger exactly one auth switch"); + assert.ok(forwarded >= 2, "the request must be re-forwarded on the retry queue"); + assert.ok(queueCalls >= 4, "retry must use a fresh queue after the switch and stream to completion"); +}); + +// Fix 6: AuthSwitcher success reset clears the consecutive empty judgment counter for the served auth index. +test("AuthSwitcher.resetEmptyJudgmentCountForAuth clears only the successful index", async () => { + const AuthSwitcher = require(path.join(__dirname, "..", "src/auth/AuthSwitcher.js")); + const mockBrowser = { currentAuthIndex: 0 }; + const authSwitcher = new AuthSwitcher( + stubLogger, + { immediateSwitchStatusCodes: [502] }, + { getAuthCount: () => 3, getCanonicalIndex: i => i }, + mockBrowser + ); + authSwitcher._emptyJudgmentCounts.set(0, 2); + authSwitcher._emptyJudgmentCounts.set(1, 1); + + authSwitcher.resetEmptyJudgmentCountForAuth(0); + + assert.strictEqual(authSwitcher._emptyJudgmentCounts.has(0), false, "successful index counter cleared"); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(1), 1, "other index counter preserved"); +}); + +test("AuthSwitcher: empty xN, success, next empty restarts at 1; threshold without success still disposes", async () => { + const AuthSwitcher = require(path.join(__dirname, "..", "src/auth/AuthSwitcher.js")); + const closed = []; + const mockBrowser = { + closeContext: async index => { + closed.push(index); + }, + currentAuthIndex: 0, + preCleanupForSwitch: async () => {}, + rebalanceContextPool: async () => {}, + switchAccount: async index => { + mockBrowser.currentAuthIndex = index; + }, + }; + const authSwitcher = new AuthSwitcher( + stubLogger, + { immediateSwitchStatusCodes: [502] }, + { getAuthCount: () => 3, getCanonicalIndex: i => i, getRotationIndices: () => [0, 1] }, + mockBrowser + ); + + // Three consecutive empties on account 0 -> threshold reached -> dispose on switch. + for (let i = 0; i < 3; i++) { + await authSwitcher.handleRequestFailureAndSwitch({ authIndex: 0, reason: "empty_upstream_response" }, null); + } + assert.strictEqual(closed.includes(0), true, "threshold of 3 empties without success disposes the context"); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.has(0), false, "counter cleared after dispose"); + + // Now empty x2 on account 1, then a success resets the counter, then one more empty -> starts at 1. + authSwitcher._emptyJudgmentCounts.set(1, 2); + authSwitcher.resetEmptyJudgmentCountForAuth(1); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.has(1), false, "success resets the counter"); + await authSwitcher.handleRequestFailureAndSwitch({ authIndex: 1, reason: "empty_upstream_response" }, null); + assert.strictEqual(authSwitcher._emptyJudgmentCounts.get(1), 1, "next empty after success starts counting from 1"); +}); + +// Fix 6: shared helper routes success sites through the AuthSwitcher success reset. +test("_resetFailureStateOnSuccess clears empty counter and failureCount via shared helper", () => { + const rh = makeHandler(); + rh.logger = stubLogger; + let resetIndex = null; + rh.authSwitcher = { + currentAuthIndex: 5, + failureCount: 3, + resetEmptyJudgmentCountForAuth: idx => { + resetIndex = idx; + }, + }; + rh._resetFailureStateOnSuccess(5); + assert.strictEqual(resetIndex, 5, "shared helper must reset empty counter for served auth index"); + assert.strictEqual(rh.authSwitcher.failureCount, 0, "shared helper must reset failureCount"); +}); diff --git a/test/requestHandlerTimeout.test.js b/test/requestHandlerTimeout.test.js new file mode 100644 index 00000000..2a57b122 --- /dev/null +++ b/test/requestHandlerTimeout.test.js @@ -0,0 +1,85 @@ +const { test } = require("node:test"); +const assert = require("node:assert"); +const RequestHandler = require("../src/core/RequestHandler"); +const MessageQueue = require("../src/utils/MessageQueue"); +const { QueueTimeoutError, QueueClosedError } = require("../src/utils/MessageQueue"); + +// Real RequestHandler relies on config for the timeout contract under test. +// AuthSwitcher/FormatConverter constructors only assign fields (no external side effects). +function makeRequestHandler(streamTimeoutMs) { + const logger = { debug: () => {}, error: () => {}, info: () => {}, warn: () => {} }; + const config = { + fakeStreamTimeoutMs: 300000, + streamTimeoutMs, + }; + const authSource = { accountNameMap: new Map() }; + const browserManager = {}; + return new RequestHandler(undefined, undefined, logger, browserManager, config, authSource); +} + +test("RequestHandler STREAM_CHUNK defaults to 0 when config streamTimeoutMs is 0", () => { + const rh = makeRequestHandler(0); + assert.strictEqual(rh.timeouts.STREAM_CHUNK, 0); +}); + +test("RequestHandler STREAM_CHUNK uses explicit positive streamTimeoutMs", () => { + const rh = makeRequestHandler(60000); + assert.strictEqual(rh.timeouts.STREAM_CHUNK, 60000); +}); + +test("Stream loop with streamTimeoutMs=0 does not timeout during long inter-chunk delay", async () => { + const queue = new MessageQueue(0); + const timeoutMs = 0; + + let received = null; + const streamTask = (async () => { + const msg = await queue.dequeue(timeoutMs); + received = msg; + })(); + + // Simulate delay longer than old 60s window (simulated with 60ms delay) + await new Promise(r => setTimeout(r, 60)); + assert.strictEqual(received, null, "Should still be waiting for chunk without timing out"); + + queue.enqueue({ data: "prefill/reasoning payload", type: "chunk" }); + await streamTask; + + assert.deepStrictEqual(received, { data: "prefill/reasoning payload", type: "chunk" }); +}); + +test("Stream loop with positive timeoutMs rejects with QueueTimeoutError when deadline passes", async () => { + const queue = new MessageQueue(0); + const timeoutMs = 30; // 30ms positive timeout + + await assert.rejects(async () => { + await queue.dequeue(timeoutMs); + }, QueueTimeoutError); +}); + +test("Stream loop with streamTimeoutMs=0 terminates cleanly when client disconnects (queue closed)", async () => { + const queue = new MessageQueue(0); + const timeoutMs = 0; + + const streamTask = queue.dequeue(timeoutMs); + + // Client disconnects + queue.close("client_disconnect"); + + await assert.rejects( + async () => { + await streamTask; + }, + err => err instanceof QueueClosedError && err.reason === "client_disconnect" + ); +}); + +test("RequestHandler real stream loop honors this.timeouts.STREAM_CHUNK (empty queue -> QueueTimeoutError)", async () => { + const rh = makeRequestHandler(30); + const queue = new MessageQueue(0); + + // Real production loop (_streamClaudeResponse) reads this.timeouts.STREAM_CHUNK (30ms here). + // Empty queue -> internal dequeue timeout fires -> QueueTimeoutError propagates before any res use. + await assert.rejects(async () => { + await rh._streamClaudeResponse(queue, {}, "gemini-2.5-flash", "req-1"); + }, QueueTimeoutError); +});