From bafceb855bdb1ed654dfb837f620c0d2c8b2be40 Mon Sep 17 00:00:00 2001 From: Cita Date: Wed, 1 Apr 2026 12:56:13 +0800 Subject: [PATCH] [codex] Add model tester fixed-channel selection (#352) * feat: add model tester forced channel * fix: keep codex websocket defaults and route refresh checks * fix: scope tester-only forced channel overrides * fix: keep tester hydration and websocket hints internal * fix: reject gemini on legacy tester chat routes * fix: harden forced channel selection edges --- .../proxy-core/channelSelection.test.ts | 56 ++++++ src/server/proxy-core/channelSelection.ts | 158 +++++++++++++++++ .../providers/codexProviderProfile.ts | 2 +- src/server/proxy-core/providers/types.ts | 1 + src/server/proxy-core/surfaces/chatSurface.ts | 137 +++++++++++---- .../proxy-core/surfaces/geminiSurface.ts | 75 +++++++-- .../surfaces/openAiResponsesSurface.ts | 151 +++++++++++------ .../proxy-core/surfaces/sharedSurface.test.ts | 82 ++++++++- .../proxy-core/surfaces/sharedSurface.ts | 41 +---- .../routeRefreshWorkflow.architecture.test.ts | 52 ++++-- src/server/routes/api/test.proxy.test.ts | 10 ++ src/server/routes/api/test.test.ts | 20 +++ src/server/routes/api/test.ts | 50 +++++- src/server/routes/proxy/completions.ts | 73 ++++---- src/server/routes/proxy/embeddings.ts | 75 +++++---- src/server/routes/proxy/gemini.test.ts | 58 +++++++ src/server/routes/proxy/images.ts | 69 +++++--- src/server/routes/proxy/search.ts | 37 ++-- .../routes/proxy/upstreamEndpoint.test.ts | 33 ++++ src/server/routes/proxy/upstreamEndpoint.ts | 17 +- src/server/routes/proxy/videos.ts | 37 ++-- src/web/api.ts | 2 + src/web/pages/ModelTester.tsx | 159 ++++++++++++++++-- .../pages/helpers/modelTesterSession.test.ts | 26 +++ src/web/pages/helpers/modelTesterSession.ts | 30 +++- ...ester.forced-channel.architecture.test.tsx | 14 ++ .../pages/modelTester.forced-channel.test.tsx | 154 +++++++++++++++++ 27 files changed, 1331 insertions(+), 288 deletions(-) create mode 100644 src/server/proxy-core/channelSelection.test.ts create mode 100644 src/server/proxy-core/channelSelection.ts create mode 100644 src/web/pages/modelTester.forced-channel.architecture.test.tsx create mode 100644 src/web/pages/modelTester.forced-channel.test.tsx diff --git a/src/server/proxy-core/channelSelection.test.ts b/src/server/proxy-core/channelSelection.test.ts new file mode 100644 index 0000000..0a93b16 --- /dev/null +++ b/src/server/proxy-core/channelSelection.test.ts @@ -0,0 +1,56 @@ +import { describe, expect, it } from 'vitest'; +import { + getTesterForcedChannelId, + normalizeForcedChannelId, + TESTER_FORCED_CHANNEL_HEADER, + TESTER_REQUEST_HEADER, +} from './channelSelection.js'; + +describe('normalizeForcedChannelId', () => { + it('accepts positive integer ids and rejects fractional or unsafe values', () => { + expect(normalizeForcedChannelId(77)).toBe(77); + expect(normalizeForcedChannelId('78')).toBe(78); + expect(normalizeForcedChannelId(77.9)).toBeNull(); + expect(normalizeForcedChannelId('78.5')).toBeNull(); + expect(normalizeForcedChannelId('9007199254740993')).toBeNull(); + expect(normalizeForcedChannelId(0)).toBeNull(); + expect(normalizeForcedChannelId(-1)).toBeNull(); + }); +}); + +describe('getTesterForcedChannelId', () => { + it('ignores forged forced-channel headers without the trusted tester bridge marker', () => { + expect(getTesterForcedChannelId({ + headers: { + [TESTER_FORCED_CHANNEL_HEADER]: '77', + }, + clientIp: '127.0.0.1', + })).toBeNull(); + + expect(getTesterForcedChannelId({ + headers: { + [TESTER_REQUEST_HEADER]: '1', + [TESTER_FORCED_CHANNEL_HEADER]: '77', + }, + clientIp: '203.0.113.10', + })).toBeNull(); + }); + + it('accepts the forced channel id only for loopback tester bridge traffic', () => { + expect(getTesterForcedChannelId({ + headers: { + [TESTER_REQUEST_HEADER]: '1', + [TESTER_FORCED_CHANNEL_HEADER]: '77', + }, + clientIp: '::1', + })).toBe(77); + + expect(getTesterForcedChannelId({ + headers: { + [TESTER_REQUEST_HEADER]: '1', + [TESTER_FORCED_CHANNEL_HEADER]: '78', + }, + clientIp: '::ffff:127.0.0.1', + })).toBe(78); + }); +}); diff --git a/src/server/proxy-core/channelSelection.ts b/src/server/proxy-core/channelSelection.ts new file mode 100644 index 0000000..c40ac51 --- /dev/null +++ b/src/server/proxy-core/channelSelection.ts @@ -0,0 +1,158 @@ +import * as routeRefreshWorkflow from '../services/routeRefreshWorkflow.js'; +import { proxyChannelCoordinator } from '../services/proxyChannelCoordinator.js'; +import { canRetryProxyChannel } from '../services/proxyChannelRetry.js'; +import type { DownstreamRoutingPolicy } from '../services/downstreamPolicyTypes.js'; +import { tokenRouter } from '../services/tokenRouter.js'; + +type SelectedChannel = Awaited>; + +export const TESTER_FORCED_CHANNEL_HEADER = 'x-metapi-tester-forced-channel-id'; +export const TESTER_REQUEST_HEADER = 'x-metapi-tester-request'; + +function headerValueEquals( + headers: Record | undefined, + expectedKey: string, + expectedValue: string, +): boolean { + if (!headers) return false; + const normalizedExpectedKey = expectedKey.trim().toLowerCase(); + const normalizedExpectedValue = expectedValue.trim().toLowerCase(); + for (const [rawKey, rawValue] of Object.entries(headers)) { + if (rawKey.trim().toLowerCase() !== normalizedExpectedKey) continue; + if (typeof rawValue === 'string' && rawValue.trim().toLowerCase() === normalizedExpectedValue) { + return true; + } + } + return false; +} + +function isLoopbackClientIp(value: string | null | undefined): boolean { + const trimmed = (value || '').trim(); + if (!trimmed) return false; + if (trimmed === '::1' || trimmed === '127.0.0.1') return true; + if (trimmed.startsWith('::ffff:')) { + return trimmed.slice('::ffff:'.length).trim() === '127.0.0.1'; + } + return false; +} + +export function normalizeForcedChannelId(value: unknown): number | null { + const numeric = typeof value === 'number' + ? value + : typeof value === 'string' && value.trim() + ? Number(value.trim()) + : NaN; + if (!Number.isSafeInteger(numeric) || numeric <= 0) return null; + return numeric; +} + +type TesterRequestInput = { + headers?: Record; + clientIp?: string | null; +}; + +export function isTrustedTesterRequest(input?: TesterRequestInput): boolean { + if (!input) return false; + if (!isLoopbackClientIp(input.clientIp)) return false; + return headerValueEquals(input.headers, TESTER_REQUEST_HEADER, '1'); +} + +export function getTesterForcedChannelId(input?: TesterRequestInput): number | null { + if (!isTrustedTesterRequest(input)) return null; + const headers = input?.headers; + if (!headers) return null; + for (const [rawKey, rawValue] of Object.entries(headers)) { + if (rawKey.trim().toLowerCase() !== TESTER_FORCED_CHANNEL_HEADER) continue; + return normalizeForcedChannelId(rawValue); + } + return null; +} + +export function buildForcedChannelUnavailableMessage(forcedChannelId?: number | null): string { + const normalizedForcedChannelId = normalizeForcedChannelId(forcedChannelId); + if (normalizedForcedChannelId === null) { + return 'No available channels for this model'; + } + return `指定通道 #${normalizedForcedChannelId} 当前不可用，固定通道模式不会自动切换其他通道`; +} + +export function canRetryChannelSelection(retryCount: number, forcedChannelId?: number | null): boolean { + if (normalizeForcedChannelId(forcedChannelId) !== null) return false; + return canRetryProxyChannel(retryCount); +} + +export async function selectProxyChannelForAttempt(input: { + requestedModel: string; + downstreamPolicy: DownstreamRoutingPolicy; + excludeChannelIds: number[]; + retryCount: number; + stickySessionKey?: string | null; + forcedChannelId?: number | null; +}): Promise { + const normalizedForcedChannelId = normalizeForcedChannelId(input.forcedChannelId); + if (normalizedForcedChannelId !== null) { + if (input.retryCount > 0) return null; + return await tokenRouter.selectPreferredChannel( + input.requestedModel, + normalizedForcedChannelId, + input.downstreamPolicy, + input.excludeChannelIds, + ); + } + + let selected: SelectedChannel = null; + let refreshedRoutes = false; + + const refreshRoutesForFirstAttempt = async (): Promise => { + if (input.retryCount > 0 || refreshedRoutes) return false; + refreshedRoutes = true; + try { + await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); + return true; + } catch (error) { + console.warn('[proxy/surface] failed to refresh routes after empty selection', error); + return false; + } + }; + + if (input.retryCount === 0 && input.stickySessionKey) { + const preferredChannelId = proxyChannelCoordinator.getStickyChannelId(input.stickySessionKey); + if (preferredChannelId && !input.excludeChannelIds.includes(preferredChannelId)) { + selected = await tokenRouter.selectPreferredChannel( + input.requestedModel, + preferredChannelId, + input.downstreamPolicy, + input.excludeChannelIds, + ); + if (!selected) { + const refreshSucceeded = await refreshRoutesForFirstAttempt(); + selected = await tokenRouter.selectPreferredChannel( + input.requestedModel, + preferredChannelId, + input.downstreamPolicy, + input.excludeChannelIds, + ); + if (!selected && refreshSucceeded) { + proxyChannelCoordinator.clearStickyChannel(input.stickySessionKey, preferredChannelId); + } + } + } + } + + if (!selected) { + selected = input.retryCount === 0 + ? await tokenRouter.selectChannel(input.requestedModel, input.downstreamPolicy) + : await tokenRouter.selectNextChannel( + input.requestedModel, + input.excludeChannelIds, + input.downstreamPolicy, + ); + } + + if (!selected && input.retryCount === 0 && !refreshedRoutes) { + await refreshRoutesForFirstAttempt(); + selected = await tokenRouter.selectChannel(input.requestedModel, input.downstreamPolicy); + } + + return selected; +} diff --git a/src/server/proxy-core/providers/codexProviderProfile.ts b/src/server/proxy-core/providers/codexProviderProfile.ts index 726135b..8db4176 100644 --- a/src/server/proxy-core/providers/codexProviderProfile.ts +++ b/src/server/proxy-core/providers/codexProviderProfile.ts @@ -10,7 +10,7 @@ export const codexProviderProfile: ProviderProfile = { id: 'codex', prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest { const isCodexOauth = asTrimmedString(input.oauthProvider).toLowerCase() === 'codex'; - const websocketTransport = getInputHeader(input.baseHeaders, 'x-metapi-responses-websocket-transport') === '1'; + const websocketTransport = input.responsesWebsocketTransport === true; const configuredUserAgent = isCodexOauth ? asTrimmedString(config.codexHeaderDefaults.userAgent) : ''; const configuredBetaFeatures = ( isCodexOauth && websocketTransport diff --git a/src/server/proxy-core/providers/types.ts b/src/server/proxy-core/providers/types.ts index 080b497..a5d47fa 100644 --- a/src/server/proxy-core/providers/types.ts +++ b/src/server/proxy-core/providers/types.ts @@ -42,6 +42,7 @@ export type PrepareProviderRequestInput = { claudeHeaders?: Record; codexSessionCacheKey?: string | null; codexExplicitSessionId?: string | null; + responsesWebsocketTransport?: boolean; body: Record; action?: ProviderAction; }; diff --git a/src/server/proxy-core/surfaces/chatSurface.ts b/src/server/proxy-core/surfaces/chatSurface.ts index f3a1c1b..c24ee11 100644 --- a/src/server/proxy-core/surfaces/chatSurface.ts +++ b/src/server/proxy-core/surfaces/chatSurface.ts @@ -46,7 +46,7 @@ import { import { summarizeConversationFileInputsInOpenAiBody } from '../capabilities/conversationFileCapabilities.js'; import { getRuntimeResponseReader, readRuntimeResponseText } from '../executors/types.js'; import { detectDownstreamClientContext } from '../../routes/proxy/downstreamClientContext.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; import { acquireSurfaceChannelLease, bindSurfaceStickyChannel, @@ -72,6 +72,11 @@ import { safeUpdateSurfaceProxyDebugSelection, startSurfaceProxyDebugTrace, } from '../../services/proxyDebugTraceRuntime.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, +} from '../channelSelection.js'; function isRecord(value: unknown): value is Record { return !!value && typeof value === 'object' && !Array.isArray(value); @@ -92,6 +97,32 @@ function prioritizeEndpointCandidate( ]; } +function finalizeRetryAsUpstreamFailure(status: number, message: string) { + return { + action: 'respond' as const, + status, + payload: { + error: { + message, + type: 'upstream_error' as const, + }, + }, + }; +} + +function finalizeRetryAsExecutionFailure(message: string) { + return { + action: 'respond' as const, + status: 502, + payload: { + error: { + message: `Upstream error: ${message}`, + type: 'upstream_error' as const, + }, + }, + }; +} + export async function handleChatSurfaceRequest( request: FastifyRequest, reply: FastifyReply, @@ -120,6 +151,10 @@ export async function handleChatSurfaceRequest( } = requestEnvelope.parsed; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const owner = getProxyResourceOwner(request); let resolvedOpenAiBody = upstreamBody; if (owner) { @@ -199,19 +234,21 @@ export async function handleChatSurfaceRequest( excludeChannelIds, retryCount, stickySessionKey, + forcedChannelId, }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); const payload = { - error: { message: 'No available channels for this model', type: 'server_error' as const }, + error: { message: noChannelMessage, type: 'server_error' as const }, }; await finalizeDebugFailure(503, payload, null); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -361,7 +398,7 @@ export async function handleChatSurfaceRequest( errorMessage: busyMessage, retryCount, }); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -472,16 +509,21 @@ export async function handleChatSurfaceRequest( latencyMs: Date.now() - startTime, retryCount, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(endpointResult.status || 502, endpointResult.errText || 'unknown error')) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, + terminalFailureOutcome.status, + terminalFailureOutcome.payload, null, ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } const upstream = endpointResult.upstream; @@ -651,16 +693,21 @@ export async function handleChatSurfaceRequest( totalTokens: parsedUsage.totalTokens, upstreamPath: successfulUpstreamPath, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(failure.status, failure.reason)) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, + terminalFailureOutcome.status, + terminalFailureOutcome.payload, successfulUpstreamPath, ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } const streamResult = streamSession.consumeUpstreamFinalPayload(fallbackData, fallbackText, streamResponse); @@ -848,16 +895,21 @@ export async function handleChatSurfaceRequest( totalTokens: parsedUsage.totalTokens, upstreamPath: successfulUpstreamPath, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(failure.status, failure.reason)) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, + terminalFailureOutcome.status, + terminalFailureOutcome.payload, successfulUpstreamPath, ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } const normalizedFinal = downstreamTransformer.transformFinalResponse(upstreamData, modelName, rawText); const downstreamResponse = downstreamTransformer.serializeFinalResponse(normalizedFinal, parsedUsage); @@ -905,16 +957,21 @@ export async function handleChatSurfaceRequest( latencyMs: Date.now() - startTime, retryCount, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsExecutionFailure(err?.message || 'network failure')) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, + terminalFailureOutcome.status, + terminalFailureOutcome.payload, null, ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } finally { channelLease.release(); } @@ -976,6 +1033,10 @@ export async function handleClaudeCountTokensSurfaceRequest( body: rawBody, }); const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const maxRetries = getProxyMaxChannelRetries(); const failureToolkit = createSurfaceFailureToolkit({ @@ -1034,18 +1095,20 @@ export async function handleClaudeCountTokensSurfaceRequest( excludeChannelIds, retryCount, stickySessionKey, + forcedChannelId, }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); await finalizeDebugFailure(503, { - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -1090,7 +1153,7 @@ export async function handleClaudeCountTokensSurfaceRequest( }, }); if (!endpointCandidates.includes('messages')) { - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -1128,7 +1191,7 @@ export async function handleClaudeCountTokensSurfaceRequest( errorMessage: busyMessage, retryCount, }); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -1240,12 +1303,17 @@ export async function handleClaudeCountTokensSurfaceRequest( latencyMs: latency, retryCount, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(upstream.status, typeof payload === 'string' ? payload : JSON.stringify(payload))) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } - await finalizeDebugFailure(failureOutcome.status, failureOutcome.payload, upstreamRequest.path); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + await finalizeDebugFailure(terminalFailureOutcome.status, terminalFailureOutcome.payload, upstreamRequest.path); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } tokenRouter.recordSuccess(selected.channel.id, latency, 0, modelName); @@ -1284,12 +1352,17 @@ export async function handleClaudeCountTokensSurfaceRequest( latencyMs: Date.now() - startTime, retryCount, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsExecutionFailure(error?.message || 'network failure')) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } - await finalizeDebugFailure(failureOutcome.status, failureOutcome.payload, null); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + await finalizeDebugFailure(terminalFailureOutcome.status, terminalFailureOutcome.payload, null); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } finally { channelLease.release(); } diff --git a/src/server/proxy-core/surfaces/geminiSurface.ts b/src/server/proxy-core/surfaces/geminiSurface.ts index 05f6374..7dc1a54 100644 --- a/src/server/proxy-core/surfaces/geminiSurface.ts +++ b/src/server/proxy-core/surfaces/geminiSurface.ts @@ -40,7 +40,7 @@ import { detectDownstreamClientContext, type DownstreamClientContext } from '../ import { insertProxyLog } from '../../services/proxyLogStore.js'; import { summarizeConversationFileInputsInOpenAiBody } from '../capabilities/conversationFileCapabilities.js'; import { getRuntimeResponseReader, readRuntimeResponseText } from '../executors/types.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; import { buildSurfaceProxyDebugResponseHeaders, captureSurfaceProxyDebugSuccessResponseBody, @@ -53,6 +53,11 @@ import { safeUpdateSurfaceProxyDebugSelection, startSurfaceProxyDebugTrace, } from '../../services/proxyDebugTraceRuntime.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, +} from '../channelSelection.js'; const GEMINI_MODEL_PROBES = [ 'gemini-2.5-flash', 'gemini-2.0-flash', @@ -124,6 +129,24 @@ async function selectNextGeminiProbeChannel(request: FastifyRequest, excludeChan return null; } +async function selectPreferredGeminiProbeChannel( + request: FastifyRequest, + forcedChannelId: number, + excludeChannelIds: number[], +) { + const policy = getDownstreamRoutingPolicy(request); + for (const candidate of GEMINI_MODEL_PROBES) { + const selected = await tokenRouter.selectPreferredChannel( + candidate, + forcedChannelId, + policy, + excludeChannelIds, + ); + if (selected) return selected; + } + return null; +} + function resolveDownstreamPath(request: FastifyRequest): string { const rawUrl = request.raw.url || request.url || ''; const withoutQuery = rawUrl.split('?')[0] || rawUrl; @@ -320,15 +343,23 @@ export async function geminiProxyRoute(app: FastifyInstance) { }); }; const excludeChannelIds: number[] = []; + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); let retryCount = 0; let lastStatus = 503; let lastText = 'No available channels for Gemini models'; let lastContentType = 'application/json'; while (retryCount <= getProxyMaxChannelRetries()) { - const selected = retryCount === 0 - ? await selectGeminiChannel(request) - : await selectNextGeminiProbeChannel(request, excludeChannelIds); + const selected = forcedChannelId !== null + ? (retryCount === 0 + ? await selectPreferredGeminiProbeChannel(request, forcedChannelId, excludeChannelIds) + : null) + : (retryCount === 0 + ? await selectGeminiChannel(request) + : await selectNextGeminiProbeChannel(request, excludeChannelIds)); if (!selected) { await finalizeDebugFailure(lastStatus, lastText, null); return reply.code(lastStatus).type(lastContentType).send(lastText); @@ -409,7 +440,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { status: upstream.status, errorText: text, }); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -438,7 +469,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { type: 'upstream_error', }, }); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -490,6 +521,10 @@ export async function geminiProxyRoute(app: FastifyInstance) { } const policy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamPath = resolveDownstreamPath(request); const clientContext = detectDownstreamClientContext({ downstreamPath, @@ -532,10 +567,24 @@ export async function geminiProxyRoute(app: FastifyInstance) { let lastContentType = 'application/json'; while (retryCount <= getProxyMaxChannelRetries()) { - const selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, policy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, policy); + const selected = forcedChannelId !== null + ? (retryCount === 0 + ? await tokenRouter.selectPreferredChannel(requestedModel, forcedChannelId, policy, excludeChannelIds) + : null) + : (retryCount === 0 + ? await tokenRouter.selectChannel(requestedModel, policy) + : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, policy)); if (!selected) { + if (forcedChannelId !== null) { + lastStatus = 503; + lastContentType = 'application/json'; + lastText = JSON.stringify({ + error: { + message: buildForcedChannelUnavailableMessage(forcedChannelId), + type: 'server_error', + }, + }); + } await finalizeDebugFailure(lastStatus, parseSurfaceProxyDebugTextPayload(lastText), null); return reply.code(lastStatus).type(lastContentType).send(lastText); } @@ -588,7 +637,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { status: 500, errorText: 'Gemini CLI OAuth project is missing', }); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -742,7 +791,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { upstreamPath, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -1258,7 +1307,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { null, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -1344,7 +1393,7 @@ export async function geminiProxyRoute(app: FastifyInstance) { upstreamPath || null, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } diff --git a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts index 9358878..3b041b9 100644 --- a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts +++ b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts @@ -70,6 +70,11 @@ import { safeUpdateSurfaceProxyDebugSelection, startSurfaceProxyDebugTrace, } from '../../services/proxyDebugTraceRuntime.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, +} from '../channelSelection.js'; function isRecord(value: unknown): value is Record { return !!value && typeof value === 'object'; @@ -166,6 +171,32 @@ function carriesResponsesFileUrlInput(value: unknown): boolean { return Object.values(value).some((entry) => carriesResponsesFileUrlInput(entry)); } +function finalizeRetryAsUpstreamFailure(status: number, message: string) { + return { + action: 'respond' as const, + status, + payload: { + error: { + message, + type: 'upstream_error' as const, + }, + }, + }; +} + +function finalizeRetryAsExecutionFailure(message: string) { + return { + action: 'respond' as const, + status: 502, + payload: { + error: { + message: `Upstream error: ${message}`, + type: 'upstream_error' as const, + }, + }, + }; +} + function shouldRefreshOauthResponsesRequest(input: { oauthProvider?: string; status: number; @@ -215,6 +246,10 @@ export async function handleOpenAiResponsesSurfaceRequest( } if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const maxRetries = getProxyMaxChannelRetries(); const failureToolkit = createSurfaceFailureToolkit({ @@ -273,19 +308,21 @@ export async function handleOpenAiResponsesSurfaceRequest( excludeChannelIds, retryCount, stickySessionKey, + forcedChannelId, }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); const payload = { - error: { message: 'No available channels for this model', type: 'server_error' as const }, + error: { message: noChannelMessage, type: 'server_error' as const }, }; await finalizeDebugFailure(503, payload, null); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -479,8 +516,8 @@ export async function handleOpenAiResponsesSurfaceRequest( errorMessage: busyMessage, retryCount, }); - retryCount += 1; - if (retryCount <= maxRetries) { + if (retryCount < maxRetries && canRetryChannelSelection(retryCount, forcedChannelId)) { + retryCount += 1; continue; } await finalizeDebugFailure(503, { @@ -590,16 +627,21 @@ export async function handleOpenAiResponsesSurfaceRequest( latencyMs: Date.now() - startTime, retryCount, }); - if (failureOutcome.action === 'retry') { + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(endpointResult.status || 502, endpointResult.errText || 'unknown error')) + : failureOutcome; + if (!terminalFailureOutcome) { retryCount += 1; continue; } await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, + terminalFailureOutcome.status, + terminalFailureOutcome.payload, null, ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } const upstream = endpointResult.upstream; @@ -748,7 +790,7 @@ export async function handleOpenAiResponsesSurfaceRequest( const failureOutcome = await failureToolkit.handleDetectedFailure({ selected, requestedModel, - modelName, + modelName, failure, latencyMs: latency, retryCount, @@ -756,17 +798,22 @@ export async function handleOpenAiResponsesSurfaceRequest( completionTokens: parsedUsage.completionTokens, totalTokens: parsedUsage.totalTokens, upstreamPath: successfulUpstreamPath, - }); - if (failureOutcome.action === 'retry') { - retryCount += 1; - continue; - } - await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, - successfulUpstreamPath, - ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + }); + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(failure.status, failure.reason)) + : failureOutcome; + if (!terminalFailureOutcome) { + retryCount += 1; + continue; + } + await finalizeDebugFailure( + terminalFailureOutcome.status, + terminalFailureOutcome.payload, + successfulUpstreamPath, + ); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } startSseResponse(); @@ -1005,7 +1052,7 @@ export async function handleOpenAiResponsesSurfaceRequest( const failureOutcome = await failureToolkit.handleDetectedFailure({ selected, requestedModel, - modelName, + modelName, failure, latencyMs: latency, retryCount, @@ -1013,17 +1060,22 @@ export async function handleOpenAiResponsesSurfaceRequest( completionTokens: parsedUsage.completionTokens, totalTokens: parsedUsage.totalTokens, upstreamPath: successfulUpstreamPath, - }); - if (failureOutcome.action === 'retry') { - retryCount += 1; - continue; - } - await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, - successfulUpstreamPath, - ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + }); + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsUpstreamFailure(failure.status, failure.reason)) + : failureOutcome; + if (!terminalFailureOutcome) { + retryCount += 1; + continue; + } + await finalizeDebugFailure( + terminalFailureOutcome.status, + terminalFailureOutcome.payload, + successfulUpstreamPath, + ); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } const normalized = openAiResponsesTransformer.transformFinalResponse( upstreamData, @@ -1077,21 +1129,26 @@ export async function handleOpenAiResponsesSurfaceRequest( const failureOutcome = await failureToolkit.handleExecutionError({ selected, requestedModel, - modelName, - errorMessage: err?.message || 'network failure', - latencyMs: Date.now() - startTime, - retryCount, - }); - if (failureOutcome.action === 'retry') { - retryCount += 1; - continue; + modelName, + errorMessage: err?.message || 'network failure', + latencyMs: Date.now() - startTime, + retryCount, + }); + const terminalFailureOutcome = failureOutcome.action === 'retry' + ? (canRetryChannelSelection(retryCount, forcedChannelId) + ? null + : finalizeRetryAsExecutionFailure(err?.message || 'network failure')) + : failureOutcome; + if (!terminalFailureOutcome) { + retryCount += 1; + continue; } - await finalizeDebugFailure( - failureOutcome.status, - failureOutcome.payload, - null, - ); - return reply.code(failureOutcome.status).send(failureOutcome.payload); + await finalizeDebugFailure( + terminalFailureOutcome.status, + terminalFailureOutcome.payload, + null, + ); + return reply.code(terminalFailureOutcome.status).send(terminalFailureOutcome.payload); } finally { channelLease.release(); } diff --git a/src/server/proxy-core/surfaces/sharedSurface.test.ts b/src/server/proxy-core/surfaces/sharedSurface.test.ts index 7469081..d30fa64 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.test.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.test.ts @@ -206,10 +206,63 @@ describe('selectSurfaceChannelForAttempt', () => { expect(clearStickyChannelMock).not.toHaveBeenCalled(); }); - it('clears stale sticky bindings and falls back to regular selection when the preferred channel is unavailable', async () => { + it('uses the forced tester channel before sticky or automatic selection', async () => { + const selected = { channel: { id: 88 } }; + selectPreferredChannelMock.mockResolvedValueOnce(selected); + + const { selectSurfaceChannelForAttempt } = await import('./sharedSurface.js'); + const result = await selectSurfaceChannelForAttempt({ + requestedModel: 'gpt-5.2', + downstreamPolicy: EMPTY_DOWNSTREAM_ROUTING_POLICY, + excludeChannelIds: [], + retryCount: 0, + stickySessionKey: 'sticky-session', + forcedChannelId: 88, + }); + + expect(result).toBe(selected); + expect(selectPreferredChannelMock).toHaveBeenCalledWith( + 'gpt-5.2', + 88, + EMPTY_DOWNSTREAM_ROUTING_POLICY, + [], + ); + expect(getStickyChannelIdMock).not.toHaveBeenCalled(); + expect(selectChannelMock).not.toHaveBeenCalled(); + expect(selectNextChannelMock).not.toHaveBeenCalled(); + expect(refreshModelsAndRebuildRoutesMock).not.toHaveBeenCalled(); + }); + + it('does not refresh or fall back when the forced tester channel is unavailable', async () => { + selectPreferredChannelMock.mockResolvedValueOnce(null); + + const { selectSurfaceChannelForAttempt } = await import('./sharedSurface.js'); + const result = await selectSurfaceChannelForAttempt({ + requestedModel: 'gpt-5.2', + downstreamPolicy: EMPTY_DOWNSTREAM_ROUTING_POLICY, + excludeChannelIds: [], + retryCount: 0, + forcedChannelId: 91, + }); + + expect(result).toBeNull(); + expect(selectPreferredChannelMock).toHaveBeenCalledWith( + 'gpt-5.2', + 91, + EMPTY_DOWNSTREAM_ROUTING_POLICY, + [], + ); + expect(selectChannelMock).not.toHaveBeenCalled(); + expect(selectNextChannelMock).not.toHaveBeenCalled(); + expect(refreshModelsAndRebuildRoutesMock).not.toHaveBeenCalled(); + }); + + it('refreshes and retries the sticky preferred channel before clearing a stale binding', async () => { const selected = { channel: { id: 22 } }; getStickyChannelIdMock.mockReturnValueOnce(55); - selectPreferredChannelMock.mockResolvedValueOnce(null); + selectPreferredChannelMock + .mockResolvedValueOnce(null) + .mockResolvedValueOnce(null); selectChannelMock.mockResolvedValueOnce(selected); const { selectSurfaceChannelForAttempt } = await import('./sharedSurface.js'); @@ -222,10 +275,35 @@ describe('selectSurfaceChannelForAttempt', () => { }); expect(result).toBe(selected); + expect(refreshModelsAndRebuildRoutesMock).toHaveBeenCalledTimes(1); + expect(selectPreferredChannelMock).toHaveBeenCalledTimes(2); expect(clearStickyChannelMock).toHaveBeenCalledWith('sticky-session', 55); expect(selectChannelMock).toHaveBeenCalledWith('gpt-5.2', EMPTY_DOWNSTREAM_ROUTING_POLICY); }); + it('keeps the sticky binding when route refresh recovers the preferred channel', async () => { + const selected = { channel: { id: 55 } }; + getStickyChannelIdMock.mockReturnValueOnce(55); + selectPreferredChannelMock + .mockResolvedValueOnce(null) + .mockResolvedValueOnce(selected); + + const { selectSurfaceChannelForAttempt } = await import('./sharedSurface.js'); + const result = await selectSurfaceChannelForAttempt({ + requestedModel: 'gpt-5.2', + downstreamPolicy: EMPTY_DOWNSTREAM_ROUTING_POLICY, + excludeChannelIds: [], + retryCount: 0, + stickySessionKey: 'sticky-session', + }); + + expect(result).toBe(selected); + expect(refreshModelsAndRebuildRoutesMock).toHaveBeenCalledTimes(1); + expect(selectPreferredChannelMock).toHaveBeenCalledTimes(2); + expect(clearStickyChannelMock).not.toHaveBeenCalled(); + expect(selectChannelMock).not.toHaveBeenCalled(); + }); + it('logs refresh failures and still retries selection once on the first attempt', async () => { const selected = { channel: { id: 33 } }; selectChannelMock diff --git a/src/server/proxy-core/surfaces/sharedSurface.ts b/src/server/proxy-core/surfaces/sharedSurface.ts index 7ca1dba..b6e6615 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.ts @@ -1,4 +1,3 @@ -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { formatUtcSqlDateTime } from '../../services/localTimeService.js'; import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js'; import type { SiteProxyConfigLike } from '../../services/siteProxy.js'; @@ -19,6 +18,7 @@ import { recordOauthQuotaResetHint } from '../../services/oauth/quota.js'; import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js'; import { proxyChannelCoordinator } from '../../services/proxyChannelCoordinator.js'; import { readRuntimeResponseText } from '../executors/types.js'; +import { selectProxyChannelForAttempt } from '../channelSelection.js'; type SelectedChannel = Awaited>; type SurfaceWarningScope = 'chat' | 'responses'; @@ -107,44 +107,9 @@ export async function selectSurfaceChannelForAttempt(input: { excludeChannelIds: number[]; retryCount: number; stickySessionKey?: string | null; + forcedChannelId?: number | null; }): Promise { - let selected: SelectedChannel = null; - - if (input.retryCount === 0 && input.stickySessionKey) { - const preferredChannelId = proxyChannelCoordinator.getStickyChannelId(input.stickySessionKey); - if (preferredChannelId && !input.excludeChannelIds.includes(preferredChannelId)) { - selected = await tokenRouter.selectPreferredChannel( - input.requestedModel, - preferredChannelId, - input.downstreamPolicy, - input.excludeChannelIds, - ); - if (!selected) { - proxyChannelCoordinator.clearStickyChannel(input.stickySessionKey, preferredChannelId); - } - } - } - - if (!selected) { - selected = input.retryCount === 0 - ? await tokenRouter.selectChannel(input.requestedModel, input.downstreamPolicy) - : await tokenRouter.selectNextChannel( - input.requestedModel, - input.excludeChannelIds, - input.downstreamPolicy, - ); - } - - if (!selected && input.retryCount === 0) { - try { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - } catch (error) { - console.warn('[proxy/surface] failed to refresh routes after empty selection', error); - } - selected = await tokenRouter.selectChannel(input.requestedModel, input.downstreamPolicy); - } - - return selected; + return await selectProxyChannelForAttempt(input); } export function buildSurfaceStickySessionKey(input: { diff --git a/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts b/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts index 3a99ed8..4cd7808 100644 --- a/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts +++ b/src/server/routes/api/routeRefreshWorkflow.architecture.test.ts @@ -10,22 +10,33 @@ function expectNoDirectModelServiceRouteRefresh(source: string): void { expect(source).not.toMatch(/import\s*\{[^}]*\brebuildTokenRoutesFromAvailability\b[^}]*\}\s*from\s*['"][^'"]*modelService\.js['"]/m); } +function expectImportsRouteRefreshWorkflow(source: string): void { + expect(source).toMatch( + /import\s+\*\s+as\s+routeRefreshWorkflow\s+from\s+['"][^'"]*routeRefreshWorkflow\.js['"]/m, + ); +} + +function expectCallsSelectProxyChannelForAttempt(source: string): void { + expect(source).toMatch(/\bselectProxyChannelForAttempt\s*\(/); +} + +function expectCallsRebuildRoutesOnly(source: string): void { + expect(source).toMatch(/\brouteRefreshWorkflow\.rebuildRoutesOnly\s*\(/); +} + describe('route refresh workflow architecture boundaries', () => { it('keeps api controllers on the shared route refresh workflow instead of modelService', () => { const tokensSource = readSource('./tokens.ts'); const settingsSource = readSource('./settings.ts'); const statsSource = readSource('./stats.ts'); - expect(tokensSource).toContain("from '../../services/routeRefreshWorkflow.js'"); - expect(tokensSource).not.toContain("from '../../services/modelService.js'"); + for (const source of [tokensSource, settingsSource, statsSource]) { + expectImportsRouteRefreshWorkflow(source); + expectNoDirectModelServiceRouteRefresh(source); + } - expect(settingsSource).toContain("from '../../services/routeRefreshWorkflow.js'"); - expect(settingsSource).not.toContain("from '../../services/modelService.js'"); - - expect(statsSource).toContain("from '../../services/routeRefreshWorkflow.js'"); - expectNoDirectModelServiceRouteRefresh(statsSource); - expect(tokensSource).toContain('const rebuild = await routeRefreshWorkflow.rebuildRoutesOnly();'); - expect(statsSource).toContain('const rebuild = await routeRefreshWorkflow.rebuildRoutesOnly();'); + expectCallsRebuildRoutesOnly(tokensSource); + expectCallsRebuildRoutesOnly(statsSource); }); it('keeps proxy fallback refreshes and scheduler hooks on the route refresh workflow', () => { @@ -39,6 +50,12 @@ describe('route refresh workflow architecture boundaries', () => { const oauthServiceSource = readSource('../../services/oauth/service.ts'); const sharedSurfaceSource = readSource('../../proxy-core/surfaces/sharedSurface.ts'); const geminiSurfaceSource = readSource('../../proxy-core/surfaces/geminiSurface.ts'); + const channelSelectionSource = readSource('../../proxy-core/channelSelection.ts'); + + for (const source of [schedulerSource, oauthServiceSource, channelSelectionSource]) { + expectImportsRouteRefreshWorkflow(source); + expectNoDirectModelServiceRouteRefresh(source); + } for (const source of [ completionsSource, @@ -47,13 +64,24 @@ describe('route refresh workflow architecture boundaries', () => { modelsRouteSource, searchSource, videosSource, - schedulerSource, - oauthServiceSource, sharedSurfaceSource, geminiSurfaceSource, ]) { - expect(source).toContain('routeRefreshWorkflow'); expectNoDirectModelServiceRouteRefresh(source); } + + for (const source of [ + completionsSource, + embeddingsSource, + imagesSource, + searchSource, + videosSource, + sharedSurfaceSource, + ]) { + expectCallsSelectProxyChannelForAttempt(source); + } + + expect(geminiSurfaceSource).toMatch(/\bselectGeminiChannel\s*\(/); + expect(geminiSurfaceSource).toMatch(/\bselectNextGeminiProbeChannel\s*\(/); }); }); diff --git a/src/server/routes/api/test.proxy.test.ts b/src/server/routes/api/test.proxy.test.ts index 15284d9..dbd0332 100644 --- a/src/server/routes/api/test.proxy.test.ts +++ b/src/server/routes/api/test.proxy.test.ts @@ -73,6 +73,7 @@ describe('testRoutes proxy tester transport', () => { stream: false, jobMode: false, rawMode: true, + forcedChannelId: 77, rawJsonText: JSON.stringify({ model: 'gpt-4o-mini', include: ['reasoning.encrypted_content'], @@ -85,6 +86,10 @@ describe('testRoutes proxy tester transport', () => { expect(response.statusCode).toBe(200); const [url, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit]; expect(url).toMatch(/\/v1\/responses$/); + expect(requestInit.headers).toMatchObject({ + 'x-metapi-tester-request': '1', + 'x-metapi-tester-forced-channel-id': '77', + }); expect(JSON.parse(String(requestInit.body))).toEqual({ model: 'gpt-4o-mini', include: ['reasoning.encrypted_content'], @@ -109,6 +114,7 @@ describe('testRoutes proxy tester transport', () => { payload: { model: 'claude-3-7-sonnet', targetFormat: 'claude', + forcedChannelId: 55, messages: [ { role: 'system', content: 'be concise' }, { role: 'user', content: 'ping' }, @@ -119,6 +125,10 @@ describe('testRoutes proxy tester transport', () => { expect(response.statusCode).toBe(200); const [url, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit]; expect(url).toMatch(/\/v1\/messages$/); + expect(requestInit.headers).toMatchObject({ + 'x-metapi-tester-request': '1', + 'x-metapi-tester-forced-channel-id': '55', + }); expect(JSON.parse(String(requestInit.body))).toEqual({ model: 'claude-3-7-sonnet', stream: false, diff --git a/src/server/routes/api/test.test.ts b/src/server/routes/api/test.test.ts index f133947..60a03d1 100644 --- a/src/server/routes/api/test.test.ts +++ b/src/server/routes/api/test.test.ts @@ -182,4 +182,24 @@ describe('/api/test proxy tester routes', () => { max_output_tokens: 4096, }); }); + + it('rejects gemini on legacy /api/test/chat wrapper until a native mapping exists', async () => { + const response = await app.inject({ + method: 'POST', + url: '/api/test/chat', + payload: { + model: 'gemini-2.5-flash', + targetFormat: 'gemini', + messages: [ + { role: 'user', content: 'hello' }, + ], + }, + }); + + expect(response.statusCode).toBe(400); + expect(response.json()).toEqual({ + error: 'targetFormat=gemini is not supported on legacy /api/test/chat routes; use the proxy tester Gemini path instead', + }); + expect(fetchMock).not.toHaveBeenCalled(); + }); }); diff --git a/src/server/routes/api/test.ts b/src/server/routes/api/test.ts index c4921b7..4264ebe 100644 --- a/src/server/routes/api/test.ts +++ b/src/server/routes/api/test.ts @@ -3,6 +3,11 @@ import { randomUUID } from 'node:crypto'; import { fetch, File as UndiciFile, FormData as UndiciFormData } from 'undici'; import { config } from '../../config.js'; import { readRuntimeResponseText } from '../../proxy-core/executors/types.js'; +import { + normalizeForcedChannelId, + TESTER_FORCED_CHANNEL_HEADER, + TESTER_REQUEST_HEADER, +} from '../../proxy-core/channelSelection.js'; type UndiciRequestInit = Parameters[1]; @@ -14,6 +19,7 @@ type TestChatRequestBody = { messages?: TestChatMessage[]; targetFormat?: TestTargetFormat; stream?: boolean; + forcedChannelId?: number; temperature?: number; top_p?: number; max_tokens?: number; @@ -27,6 +33,7 @@ type ValidatedTestChatPayload = { messages: TestChatMessage[]; targetFormat: TestTargetFormat; stream?: boolean; + forcedChannelId?: number | null; temperature?: number; top_p?: number; max_tokens?: number; @@ -52,6 +59,7 @@ type ProxyTestEnvelope = { stream?: boolean; jobMode?: boolean; rawMode?: boolean; + forcedChannelId?: number | null; jsonBody?: unknown; rawJsonText?: string; multipartFields?: Record; @@ -65,6 +73,7 @@ type ValidatedProxyTestEnvelope = { stream: boolean; jobMode: boolean; rawMode: boolean; + forcedChannelId?: number | null; jsonBody?: unknown; rawJsonText?: string; multipartFields?: Record; @@ -162,19 +171,25 @@ function validateLegacyPayload( return null; } + if (body.targetFormat === 'gemini') { + reply.code(400).send({ + error: 'targetFormat=gemini is not supported on legacy /api/test/chat routes; use the proxy tester Gemini path instead', + }); + return null; + } + const targetFormat: TestTargetFormat = body.targetFormat === 'claude' ? 'claude' : body.targetFormat === 'responses' ? 'responses' - : body.targetFormat === 'gemini' - ? 'gemini' - : 'openai'; + : 'openai'; return { model: body.model, messages: body.messages, targetFormat, stream: body.stream, + forcedChannelId: normalizeForcedChannelId(body.forcedChannelId), temperature: body.temperature, top_p: body.top_p, max_tokens: body.max_tokens, @@ -289,6 +304,7 @@ function convertLegacyPayloadToEnvelope( stream: forceStream, jobMode: false, rawMode: false, + forcedChannelId: payload.forcedChannelId ?? null, jsonBody: convertOpenAiPayloadToClaudeBody(payload, forceStream), }; } @@ -301,6 +317,7 @@ function convertLegacyPayloadToEnvelope( stream: forceStream, jobMode: false, rawMode: false, + forcedChannelId: payload.forcedChannelId ?? null, jsonBody: convertOpenAiPayloadToResponsesBody(payload, forceStream), }; } @@ -312,9 +329,29 @@ function convertLegacyPayloadToEnvelope( stream: forceStream, jobMode: false, rawMode: false, + forcedChannelId: payload.forcedChannelId ?? null, jsonBody: { - ...payload, + model: payload.model, + messages: payload.messages, stream: forceStream, + ...(typeof payload.temperature === 'number' && Number.isFinite(payload.temperature) + ? { temperature: payload.temperature } + : {}), + ...(typeof payload.top_p === 'number' && Number.isFinite(payload.top_p) + ? { top_p: payload.top_p } + : {}), + ...(typeof payload.max_tokens === 'number' && Number.isFinite(payload.max_tokens) + ? { max_tokens: payload.max_tokens } + : {}), + ...(typeof payload.frequency_penalty === 'number' && Number.isFinite(payload.frequency_penalty) + ? { frequency_penalty: payload.frequency_penalty } + : {}), + ...(typeof payload.presence_penalty === 'number' && Number.isFinite(payload.presence_penalty) + ? { presence_penalty: payload.presence_penalty } + : {}), + ...(typeof payload.seed === 'number' && Number.isFinite(payload.seed) + ? { seed: payload.seed } + : {}), }, }; } @@ -356,6 +393,7 @@ function validateProxyEnvelope( stream: body.stream === true, jobMode: body.jobMode === true, rawMode: body.rawMode === true, + forcedChannelId: normalizeForcedChannelId(body.forcedChannelId), }; if (requestKind === 'json') { @@ -472,6 +510,10 @@ async function buildUpstreamRequestInit( forceStream: boolean, ): Promise { const headers: Record = createDefaultHeadersForPath(envelope.path); + headers[TESTER_REQUEST_HEADER] = '1'; + if (typeof envelope.forcedChannelId === 'number' && envelope.forcedChannelId > 0) { + headers[TESTER_FORCED_CHANNEL_HEADER] = String(envelope.forcedChannelId); + } if (envelope.requestKind === 'json') { headers['Content-Type'] = 'application/json'; diff --git a/src/server/routes/proxy/completions.ts b/src/server/routes/proxy/completions.ts index 7c685d2..8e2e703 100644 --- a/src/server/routes/proxy/completions.ts +++ b/src/server/routes/proxy/completions.ts @@ -1,10 +1,9 @@ -import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; -import { fetch } from 'undici'; -import { tokenRouter } from '../../services/tokenRouter.js'; -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; -import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; -import { isTokenExpiredError } from '../../services/alertRules.js'; -import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; +import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; +import { fetch } from 'undici'; +import { tokenRouter } from '../../services/tokenRouter.js'; +import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; +import { isTokenExpiredError } from '../../services/alertRules.js'; +import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; @@ -18,7 +17,13 @@ import { getProxyAuthContext } from '../../middleware/auth.js'; import { buildUpstreamUrl } from './upstreamUrl.js'; import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js'; import { insertProxyLog } from '../../services/proxyLogStore.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, + selectProxyChannelForAttempt, +} from '../../proxy-core/channelSelection.js'; export async function completionsProxyRoute(app: FastifyInstance) { app.post('/v1/completions', async (request: FastifyRequest, reply: FastifyReply) => { @@ -29,6 +34,10 @@ export async function completionsProxyRoute(app: FastifyInstance) { } if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const downstreamPath = '/v1/completions'; const clientContext = detectDownstreamClientContext({ @@ -38,28 +47,28 @@ export async function completionsProxyRoute(app: FastifyInstance) { }); const isStream = body.stream === true; - const excludeChannelIds: number[] = []; - let retryCount = 0; - + const excludeChannelIds: number[] = []; + let retryCount = 0; + while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } - - if (!selected) { - await reportProxyAllFailed({ - model: requestedModel, - reason: 'No available channels after retries', - }); - return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, - }); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); + + if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); + await reportProxyAllFailed({ + model: requestedModel, + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', + }); + return reply.code(503).send({ + error: { message: noChannelMessage, type: 'server_error' }, + }); + } excludeChannelIds.push(selected.channel.id); @@ -112,7 +121,7 @@ export async function completionsProxyRoute(app: FastifyInstance) { }); } - if (shouldRetryProxyRequest(upstream.status, errText) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, errText) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -258,7 +267,7 @@ export async function completionsProxyRoute(app: FastifyInstance) { downstreamPath, ); - if (shouldRetryProxyRequest(failure.status, errText) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(failure.status, errText) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -341,7 +350,7 @@ export async function completionsProxyRoute(app: FastifyInstance) { clientContext, downstreamPath, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } diff --git a/src/server/routes/proxy/embeddings.ts b/src/server/routes/proxy/embeddings.ts index be77670..e488f27 100644 --- a/src/server/routes/proxy/embeddings.ts +++ b/src/server/routes/proxy/embeddings.ts @@ -1,10 +1,9 @@ -import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; -import { fetch } from 'undici'; -import { tokenRouter } from '../../services/tokenRouter.js'; -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; -import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; -import { isTokenExpiredError } from '../../services/alertRules.js'; -import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; +import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; +import { fetch } from 'undici'; +import { tokenRouter } from '../../services/tokenRouter.js'; +import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; +import { isTokenExpiredError } from '../../services/alertRules.js'; +import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js'; import { parseProxyUsage } from '../../services/proxyUsageParser.js'; import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js'; @@ -17,7 +16,13 @@ import { getProxyAuthContext } from '../../middleware/auth.js'; import { buildUpstreamUrl } from './upstreamUrl.js'; import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js'; import { insertProxyLog } from '../../services/proxyLogStore.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, + selectProxyChannelForAttempt, +} from '../../proxy-core/channelSelection.js'; export async function embeddingsProxyRoute(app: FastifyInstance) { app.post('/v1/embeddings', async (request: FastifyRequest, reply: FastifyReply) => { @@ -25,10 +30,14 @@ export async function embeddingsProxyRoute(app: FastifyInstance) { const requestedModel = body?.model; if (!requestedModel) { return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } }); - } - if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; - const downstreamPolicy = getDownstreamRoutingPolicy(request); - const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; + } + if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; + const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); + const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const downstreamPath = '/v1/embeddings'; const clientContext = detectDownstreamClientContext({ downstreamPath, @@ -36,26 +45,26 @@ export async function embeddingsProxyRoute(app: FastifyInstance) { body, }); - const excludeChannelIds: number[] = []; - let retryCount = 0; - + const excludeChannelIds: number[] = []; + let retryCount = 0; + while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } - - if (!selected) { - await reportProxyAllFailed({ - model: requestedModel, - reason: 'No available channels after retries', - }); - return reply.code(503).send({ error: { message: 'No available channels', type: 'server_error' } }); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); + + if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); + await reportProxyAllFailed({ + model: requestedModel, + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', + }); + return reply.code(503).send({ error: { message: noChannelMessage, type: 'server_error' } }); + } excludeChannelIds.push(selected.channel.id); const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/embeddings'); @@ -107,7 +116,7 @@ export async function embeddingsProxyRoute(app: FastifyInstance) { }); } - if (shouldRetryProxyRequest(upstream.status, text) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -178,7 +187,7 @@ export async function embeddingsProxyRoute(app: FastifyInstance) { clientContext, downstreamPath, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } diff --git a/src/server/routes/proxy/gemini.test.ts b/src/server/routes/proxy/gemini.test.ts index 6f58ed1..1c9b1ee 100644 --- a/src/server/routes/proxy/gemini.test.ts +++ b/src/server/routes/proxy/gemini.test.ts @@ -10,6 +10,7 @@ const refreshModelsAndRebuildRoutesMock = vi.fn(); const refreshOauthAccessTokenSingleflightMock = vi.fn(); const selectChannelMock = vi.fn(); const selectNextChannelMock = vi.fn(); +const selectPreferredChannelMock = vi.fn(); const recordSuccessMock = vi.fn(); const recordFailureMock = vi.fn(); const explainSelectionMock = vi.fn(); @@ -68,6 +69,7 @@ vi.mock('../../services/tokenRouter.js', () => ({ tokenRouter: { selectChannel: (...args: unknown[]) => selectChannelMock(...args), selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args), + selectPreferredChannel: (...args: unknown[]) => selectPreferredChannelMock(...args), recordSuccess: (...args: unknown[]) => recordSuccessMock(...args), recordFailure: (...args: unknown[]) => recordFailureMock(...args), explainSelection: (...args: unknown[]) => explainSelectionMock(...args), @@ -181,6 +183,7 @@ describe('gemini native proxy routes', () => { refreshOauthAccessTokenSingleflightMock.mockReset(); selectChannelMock.mockReset(); selectNextChannelMock.mockReset(); + selectPreferredChannelMock.mockReset(); recordSuccessMock.mockReset(); recordFailureMock.mockReset(); explainSelectionMock.mockReset(); @@ -232,6 +235,7 @@ describe('gemini native proxy routes', () => { actualModel: 'gemini-2.5-flash', }); selectNextChannelMock.mockReturnValue(null); + selectPreferredChannelMock.mockReturnValue(null); recordSuccessMock.mockResolvedValue(undefined); recordFailureMock.mockResolvedValue(undefined); resetUpstreamEndpointRuntimeState(); @@ -318,6 +322,60 @@ describe('gemini native proxy routes', () => { expect(secondUrl).toContain('key=gemini-key-2'); }); + it('pins /v1beta/models to the forced tester channel when present', async () => { + selectPreferredChannelMock + .mockReturnValueOnce(null) + .mockReturnValueOnce({ + channel: { id: 77, routeId: 22 }, + site: { id: 88, name: 'forced-gemini-site', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' }, + account: { id: 39, username: 'forced-user' }, + tokenName: 'forced', + tokenValue: 'forced-gemini-key', + actualModel: 'gemini-2.0-flash', + }); + + fetchMock.mockResolvedValue(new Response(JSON.stringify({ + models: [ + { name: 'models/gemini-2.0-flash', displayName: 'Gemini 2.0 Flash' }, + ], + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const response = await app.inject({ + method: 'GET', + url: '/v1beta/models', + remoteAddress: '127.0.0.1', + headers: { + authorization: 'Bearer sk-managed-gemini', + 'x-metapi-tester-request': '1', + 'x-metapi-tester-forced-channel-id': '77', + }, + }); + + expect(response.statusCode).toBe(200); + expect(selectChannelMock).not.toHaveBeenCalled(); + expect(selectNextChannelMock).not.toHaveBeenCalled(); + expect(selectPreferredChannelMock).toHaveBeenCalledTimes(2); + expect(selectPreferredChannelMock).toHaveBeenNthCalledWith( + 1, + 'gemini-2.5-flash', + 77, + expect.anything(), + expect.any(Array), + ); + expect(selectPreferredChannelMock).toHaveBeenNthCalledWith( + 2, + 'gemini-2.0-flash', + 77, + expect.anything(), + expect.any(Array), + ); + const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit]; + expect(targetUrl).toContain('key=forced-gemini-key'); + }); + it('serves gemini-cli model list from local static catalog without upstream fetch', async () => { selectChannelMock.mockReturnValue({ channel: { id: 21, routeId: 22 }, diff --git a/src/server/routes/proxy/images.ts b/src/server/routes/proxy/images.ts index 2b6c2f8..8a5b6c1 100644 --- a/src/server/routes/proxy/images.ts +++ b/src/server/routes/proxy/images.ts @@ -1,7 +1,6 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { estimateProxyCost } from '../../services/modelPricingService.js'; @@ -16,7 +15,13 @@ import { getProxyAuthContext } from '../../middleware/auth.js'; import { buildUpstreamUrl } from './upstreamUrl.js'; import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js'; import { insertProxyLog } from '../../services/proxyLogStore.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, + selectProxyChannelForAttempt, +} from '../../proxy-core/channelSelection.js'; export async function imagesProxyRoute(app: FastifyInstance) { ensureMultipartBufferParser(app); @@ -26,6 +31,10 @@ export async function imagesProxyRoute(app: FastifyInstance) { const requestedModel = body?.model || 'gpt-image-1'; if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const downstreamPath = '/v1/images/generations'; const clientContext = detectDownstreamClientContext({ @@ -37,22 +46,22 @@ export async function imagesProxyRoute(app: FastifyInstance) { let retryCount = 0; while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -101,7 +110,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { detail: `HTTP ${upstream.status}`, }); } - if (shouldRetryProxyRequest(upstream.status, text) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -132,7 +141,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { downstreamPath, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -184,7 +193,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { downstreamPath, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -210,6 +219,10 @@ export async function imagesProxyRoute(app: FastifyInstance) { if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const downstreamPath = '/v1/images/edits'; const clientContext = detectDownstreamClientContext({ @@ -221,22 +234,22 @@ export async function imagesProxyRoute(app: FastifyInstance) { let retryCount = 0; while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -297,7 +310,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { detail: `HTTP ${upstream.status}`, }); } - if (shouldRetryProxyRequest(upstream.status, text) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -328,7 +341,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { downstreamPath, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } @@ -380,7 +393,7 @@ export async function imagesProxyRoute(app: FastifyInstance) { downstreamPath, clientContext, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount++; continue; } diff --git a/src/server/routes/proxy/search.ts b/src/server/routes/proxy/search.ts index 543b9e1..e5dbaa6 100644 --- a/src/server/routes/proxy/search.ts +++ b/src/server/routes/proxy/search.ts @@ -1,7 +1,6 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js'; @@ -14,7 +13,13 @@ import { getProxyAuthContext } from '../../middleware/auth.js'; import { buildUpstreamUrl } from './upstreamUrl.js'; import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js'; import { insertProxyLog } from '../../services/proxyLogStore.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, + selectProxyChannelForAttempt, +} from '../../proxy-core/channelSelection.js'; const DEFAULT_SEARCH_MODEL = '__search'; const DEFAULT_MAX_RESULTS = 10; const MAX_MAX_RESULTS = 20; @@ -54,6 +59,10 @@ export async function searchProxyRoute(app: FastifyInstance) { if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null; const downstreamPath = '/v1/search'; const clientContext = detectDownstreamClientContext({ @@ -65,22 +74,22 @@ export async function searchProxyRoute(app: FastifyInstance) { let retryCount = 0; while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -131,7 +140,7 @@ export async function searchProxyRoute(app: FastifyInstance) { detail: `HTTP ${upstream.status}`, }); } - if (shouldRetryProxyRequest(upstream.status, text) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -170,7 +179,7 @@ export async function searchProxyRoute(app: FastifyInstance) { clientContext, downstreamPath, ); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } diff --git a/src/server/routes/proxy/upstreamEndpoint.test.ts b/src/server/routes/proxy/upstreamEndpoint.test.ts index 4bd285b..b45e170 100644 --- a/src/server/routes/proxy/upstreamEndpoint.test.ts +++ b/src/server/routes/proxy/upstreamEndpoint.test.ts @@ -935,6 +935,39 @@ describe('buildUpstreamEndpointRequest', () => { expect(httpRequest.headers['x-codex-beta-features']).toBeUndefined(); }); + it('uses internal websocket transport hints without forwarding internal metapi headers upstream', () => { + (config as any).codexHeaderDefaults = { + userAgent: 'codex-config-ua/1.0', + betaFeatures: 'multi_agent', + }; + + const request = buildUpstreamEndpointRequest({ + endpoint: 'responses', + modelName: 'gpt-5.4', + stream: true, + tokenValue: 'oauth-access-token', + oauthProvider: 'codex', + sitePlatform: 'codex', + siteUrl: 'https://chatgpt.com/backend-api/codex', + openaiBody: { + model: 'gpt-5.4', + messages: [{ role: 'user', content: 'hello codex' }], + }, + downstreamFormat: 'openai', + downstreamHeaders: { + 'x-metapi-responses-websocket-transport': '1', + 'x-metapi-tester-forced-channel-id': '77', + }, + providerHeaders: { + Originator: 'codex_cli_rs', + }, + } as any); + + expect(request.headers['x-codex-beta-features']).toBe('multi_agent'); + expect(request.headers['x-metapi-responses-websocket-transport']).toBeUndefined(); + expect(request.headers['x-metapi-tester-forced-channel-id']).toBeUndefined(); + }); + it('applies configured payload rules before preparing codex responses requests while forcing store false', () => { (config as any).payloadRules = { default: [ diff --git a/src/server/routes/proxy/upstreamEndpoint.ts b/src/server/routes/proxy/upstreamEndpoint.ts index d3520d1..02d97c2 100644 --- a/src/server/routes/proxy/upstreamEndpoint.ts +++ b/src/server/routes/proxy/upstreamEndpoint.ts @@ -34,6 +34,7 @@ import { import { buildClaudeRuntimeHeaders, buildGeminiCliUserAgent, + getInputHeader, headerValueToString, } from '../../proxy-core/providers/headerUtils.js'; export { @@ -112,11 +113,20 @@ const BLOCKED_PASSTHROUGH_HEADERS = new Set([ 'sec-websocket-version', 'sec-websocket-extensions', ]); +const METAPI_INTERNAL_HEADER_BLOCKLIST = new Set([ + 'x-metapi-tester-request', + 'x-metapi-tester-forced-channel-id', + 'x-metapi-responses-websocket-mode', + 'x-metapi-responses-websocket-transport', +]); const ANTIGRAVITY_RUNTIME_USER_AGENT = 'antigravity/1.19.6 darwin/arm64'; function shouldSkipPassthroughHeader(key: string): boolean { - return HOP_BY_HOP_HEADERS.has(key) || BLOCKED_PASSTHROUGH_HEADERS.has(key); + if (HOP_BY_HOP_HEADERS.has(key) || BLOCKED_PASSTHROUGH_HEADERS.has(key)) return true; + if (METAPI_INTERNAL_HEADER_BLOCKLIST.has(key)) return true; + if (key.startsWith('x-metapi-')) return true; + return false; } function extractSafePassthroughHeaders( @@ -911,6 +921,10 @@ export function buildUpstreamEndpointRequest(input: { } if (input.endpoint === 'responses') { + const responsesWebsocketTransport = getInputHeader( + input.downstreamHeaders, + 'x-metapi-responses-websocket-transport', + ) === '1'; const websocketMode = Object.entries(input.downstreamHeaders || {}).find(([rawKey]) => rawKey.trim().toLowerCase() === 'x-metapi-responses-websocket-mode'); const preserveWebsocketIncrementalMode = asTrimmedString(websocketMode?.[1]).toLowerCase() === 'incremental'; const responsesHeaders = input.downstreamFormat === 'responses' @@ -963,6 +977,7 @@ export function buildUpstreamEndpointRequest(input: { providerHeaders: input.providerHeaders, codexSessionCacheKey: input.codexSessionCacheKey, codexExplicitSessionId: input.codexExplicitSessionId, + responsesWebsocketTransport, body: configuredResponsesBody, }); } diff --git a/src/server/routes/proxy/videos.ts b/src/server/routes/proxy/videos.ts index 9be4c3c..46457bb 100644 --- a/src/server/routes/proxy/videos.ts +++ b/src/server/routes/proxy/videos.ts @@ -1,7 +1,6 @@ import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify'; import { fetch } from 'undici'; import { tokenRouter } from '../../services/tokenRouter.js'; -import * as routeRefreshWorkflow from '../../services/routeRefreshWorkflow.js'; import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js'; import { isTokenExpiredError } from '../../services/alertRules.js'; import { estimateProxyCost } from '../../services/modelPricingService.js'; @@ -17,7 +16,13 @@ import { refreshProxyVideoTaskSnapshot, saveProxyVideoTask, } from '../../services/proxyVideoTaskStore.js'; -import { canRetryProxyChannel, getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { getProxyMaxChannelRetries } from '../../services/proxyChannelRetry.js'; +import { + buildForcedChannelUnavailableMessage, + canRetryChannelSelection, + getTesterForcedChannelId, + selectProxyChannelForAttempt, +} from '../../proxy-core/channelSelection.js'; function rewriteVideoResponsePublicId(payload: unknown, publicId: string): unknown { if (!payload || typeof payload !== 'object') return payload; @@ -47,26 +52,30 @@ export async function videosProxyRoute(app: FastifyInstance) { if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return; const downstreamPolicy = getDownstreamRoutingPolicy(request); + const forcedChannelId = getTesterForcedChannelId({ + headers: request.headers as Record, + clientIp: request.ip, + }); const excludeChannelIds: number[] = []; let retryCount = 0; while (retryCount <= getProxyMaxChannelRetries()) { - let selected = retryCount === 0 - ? await tokenRouter.selectChannel(requestedModel, downstreamPolicy) - : await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy); - - if (!selected && retryCount === 0) { - await routeRefreshWorkflow.refreshModelsAndRebuildRoutes(); - selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy); - } + const selected = await selectProxyChannelForAttempt({ + requestedModel, + downstreamPolicy, + excludeChannelIds, + retryCount, + forcedChannelId, + }); if (!selected) { + const noChannelMessage = buildForcedChannelUnavailableMessage(forcedChannelId); await reportProxyAllFailed({ model: requestedModel, - reason: 'No available channels after retries', + reason: forcedChannelId ? noChannelMessage : 'No available channels after retries', }); return reply.code(503).send({ - error: { message: 'No available channels for this model', type: 'server_error' }, + error: { message: noChannelMessage, type: 'server_error' }, }); } @@ -115,7 +124,7 @@ export async function videosProxyRoute(app: FastifyInstance) { detail: `HTTP ${upstream.status}`, }); } - if (shouldRetryProxyRequest(upstream.status, text) && canRetryProxyChannel(retryCount)) { + if (shouldRetryProxyRequest(upstream.status, text) && canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } @@ -170,7 +179,7 @@ export async function videosProxyRoute(app: FastifyInstance) { errorText: error?.message || 'network failure', modelName: upstreamModel, })); - if (canRetryProxyChannel(retryCount)) { + if (canRetryChannelSelection(retryCount, forcedChannelId)) { retryCount += 1; continue; } diff --git a/src/web/api.ts b/src/web/api.ts index 9878337..3f4d883 100644 --- a/src/web/api.ts +++ b/src/web/api.ts @@ -232,6 +232,7 @@ type TestChatRequestPayload = { messages: Array<{ role: string; content: string }>; targetFormat?: 'openai' | 'claude' | 'responses' | 'gemini'; stream?: boolean; + forcedChannelId?: number | null; temperature?: number; top_p?: number; max_tokens?: number; @@ -257,6 +258,7 @@ export type ProxyTestRequestEnvelope = { stream?: boolean; jobMode?: boolean; rawMode?: boolean; + forcedChannelId?: number | null; jsonBody?: unknown; rawJsonText?: string; multipartFields?: Record; diff --git a/src/web/pages/ModelTester.tsx b/src/web/pages/ModelTester.tsx index 48a6943..cfe940e 100644 --- a/src/web/pages/ModelTester.tsx +++ b/src/web/pages/ModelTester.tsx @@ -18,6 +18,7 @@ import { buildSearchRequestEnvelope, buildVideoCreateRequestEnvelope, buildVideoInspectRequestEnvelope, + attachForcedChannelToEnvelope, countConversationTurns, collectModelTesterModelNames, createLoadingAssistantMessage, @@ -82,6 +83,11 @@ type UploadState = { }; type ConversationFileState = ConversationDraftFile; +type ForcedChannelOption = { + value: string; + label: string; + description?: string; +}; const POLL_INTERVAL_MS = 1200; const wait = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms)); @@ -659,6 +665,11 @@ export default function ModelTester() { const [inputs, setInputs] = useState(DEFAULT_INPUTS); const [modeState, setModeState] = useState(DEFAULT_MODE_STATE); const [parameterEnabled, setParameterEnabled] = useState(DEFAULT_PARAMETER_ENABLED); + const [forcedChannelId, setForcedChannelId] = useState(null); + const [forcedChannelOptions, setForcedChannelOptions] = useState([]); + const [loadingForcedChannels, setLoadingForcedChannels] = useState(false); + const [forcedChannelHint, setForcedChannelHint] = useState(''); + const [forcedChannelHydrationReady, setForcedChannelHydrationReady] = useState(false); const [sending, setSending] = useState(false); const [loadingModels, setLoadingModels] = useState(true); @@ -759,6 +770,7 @@ export default function ModelTester() { setParameterEnabled(restored.parameterEnabled); setPendingPayload(restored.pendingPayload); setPendingJobId(restored.pendingJobId || null); + setForcedChannelId(restored.forcedChannelId ?? null); setCustomRequestMode(restored.customRequestMode); setCustomRequestBody(restored.customRequestBody); setShowDebugPanel(restored.showDebugPanel); @@ -817,6 +829,7 @@ export default function ModelTester() { pushDebug('error', '获取模型列表失败。'); } finally { setLoadingModels(false); + setForcedChannelHydrationReady(true); } }; @@ -824,6 +837,72 @@ export default function ModelTester() { // eslint-disable-next-line react-hooks/exhaustive-deps }, []); + useEffect(() => { + if (!forcedChannelHydrationReady) return; + + if (!inputs.model) { + setForcedChannelOptions([]); + setForcedChannelHint(''); + setForcedChannelId(null); + return; + } + + if (customRequestMode) { + setForcedChannelOptions([]); + setForcedChannelHint('自定义请求模式下固定通道不可用。'); + setForcedChannelId(null); + return; + } + + if (inputs.mode === 'videos.inspect') { + setForcedChannelOptions([]); + setForcedChannelHint('视频查询/删除不会重新选路，不能固定通道。'); + setForcedChannelId(null); + return; + } + + let cancelled = false; + setLoadingForcedChannels(true); + setForcedChannelHint(''); + + void api.getRouteDecision(inputs.model) + .then((result) => { + if (cancelled) return; + const candidates = Array.isArray((result as any)?.decision?.candidates) + ? (result as any).decision.candidates as Array> + : []; + const nextOptions = candidates + .filter((candidate) => candidate?.eligible === true && typeof candidate?.channelId === 'number') + .map((candidate) => ({ + value: String(candidate.channelId), + label: `${candidate.username || `account-${candidate.accountId || 'unknown'}`} @ ${candidate.siteName || 'unknown'} / ${candidate.tokenName || 'default'} (P${candidate.priority ?? 0})`, + description: typeof candidate.reason === 'string' && candidate.reason.trim().length > 0 + ? candidate.reason + : undefined, + })); + setForcedChannelOptions(nextOptions); + if (nextOptions.length === 0) { + setForcedChannelHint('当前模型暂无可固定通道。'); + } + if (typeof forcedChannelId === 'number' && !nextOptions.some((option) => option.value === String(forcedChannelId))) { + setForcedChannelId(null); + } + }) + .catch(() => { + if (cancelled) return; + setForcedChannelOptions([]); + setForcedChannelHint('加载固定通道候选失败。'); + setForcedChannelId(null); + }) + .finally(() => { + if (!cancelled) setLoadingForcedChannels(false); + }); + + return () => { + cancelled = true; + }; + }, [customRequestMode, forcedChannelHydrationReady, inputs.mode, inputs.model]); + useEffect(() => { if (!inputs.model) return; localStorage.setItem(MODEL_TESTER_STORAGE_KEY, serializeModelTesterSession({ @@ -845,6 +924,7 @@ export default function ModelTester() { }, pendingPayload, pendingJobId, + forcedChannelId, customRequestMode, customRequestBody, showDebugPanel, @@ -854,6 +934,7 @@ export default function ModelTester() { activeDebugTab, customRequestBody, customRequestMode, + forcedChannelId, input, inputs, messages, @@ -1158,6 +1239,19 @@ export default function ModelTester() { }; }, [buildApiPayload, buildClaudeBodyFromMessages, buildConversationMessagesWithSystem, buildResponsesBodyFromMessages, customRequestBody, customRequestMode, inputs, parameterEnabled]); + const forcedChannelSelectOptions = useMemo(() => [ + { + value: '__auto__', + label: '自动选路（默认）', + description: '按当前路由正常选择通道', + }, + ...forcedChannelOptions, + ], [forcedChannelOptions]); + + const attachEnvelopeForcedChannel = useCallback((envelope: ProxyTestEnvelope) => ( + attachForcedChannelToEnvelope(envelope, forcedChannelId) + ), [forcedChannelId]); + const buildModeProxyEnvelope = useCallback((): ProxyTestEnvelope | null => { if (inputs.mode === 'embeddings') { const trimmed = embeddingInputText.trim(); @@ -1298,7 +1392,8 @@ export default function ModelTester() { const previewPayload = useMemo(() => { if (inputs.mode !== 'conversation') { - return buildModeProxyEnvelope(); + const envelope = buildModeProxyEnvelope(); + return envelope ? attachEnvelopeForcedChannel(envelope) : null; } if (customRequestMode) { const raw = customRequestBody.trim(); @@ -1310,10 +1405,10 @@ export default function ModelTester() { } } if (inputs.protocol === 'gemini') { - return buildConversationProxyEnvelope(messages); + return attachEnvelopeForcedChannel(buildConversationProxyEnvelope(messages)); } - return buildApiPayload(buildConversationMessagesWithSystem(messages), inputs, parameterEnabled); - }, [buildConversationMessagesWithSystem, buildConversationProxyEnvelope, buildModeProxyEnvelope, customRequestBody, customRequestMode, inputs, messages, parameterEnabled]); + return attachEnvelopeForcedChannel(buildApiPayload(buildConversationMessagesWithSystem(messages), inputs, parameterEnabled)); + }, [attachEnvelopeForcedChannel, buildConversationMessagesWithSystem, buildConversationProxyEnvelope, buildModeProxyEnvelope, customRequestBody, customRequestMode, inputs, messages, parameterEnabled]); useEffect(() => { setDebugPreview(formatJson(previewPayload)); @@ -1737,39 +1832,43 @@ export default function ModelTester() { payload: TestChatPayload, options?: { syncedCustomBody?: string }, ) => { + const effectivePayload = attachEnvelopeForcedChannel(payload); setMessages(nextMessages); if (options?.syncedCustomBody !== undefined) { setCustomRequestBody(options.syncedCustomBody); } setError(''); - setPendingPayload(payload); - setDebugRequest(formatJson(payload)); + setPendingPayload(effectivePayload); + setDebugRequest(formatJson(effectivePayload)); setDebugResponse(''); setActiveDebugTab(DEBUG_TABS.REQUEST); setDebugTimestamp(new Date().toISOString()); - if (payload.stream) { - await startStream(payload); + if (effectivePayload.stream) { + await startStream(effectivePayload); } else { - await startChatJob(payload); + await startChatJob(effectivePayload); } - }, [startChatJob, startStream]); + }, [attachEnvelopeForcedChannel, startChatJob, startStream]); const dispatchProxyEnvelope = useCallback(async (envelope: ProxyTestEnvelope, nextMessages?: ChatMessage[]) => { + const effectiveEnvelope = attachEnvelopeForcedChannel(envelope); setError(''); - setDebugRequest(formatJson(envelope.rawMode ? { path: envelope.path, rawJsonText: envelope.rawJsonText } : envelope)); + setDebugRequest(formatJson(effectiveEnvelope.rawMode + ? { path: effectiveEnvelope.path, rawJsonText: effectiveEnvelope.rawJsonText, forcedChannelId: effectiveEnvelope.forcedChannelId } + : effectiveEnvelope)); setDebugResponse(''); setActiveDebugTab(DEBUG_TABS.REQUEST); setDebugTimestamp(new Date().toISOString()); - if (envelope.stream && nextMessages) { - await startProxyStream(envelope, nextMessages); + if (effectiveEnvelope.stream && nextMessages) { + await startProxyStream(effectiveEnvelope, nextMessages); return; } setSending(true); try { - const result = await api.proxyTest(envelope); + const result = await api.proxyTest(effectiveEnvelope); setDebugResponse(formatJson(result)); setActiveDebugTab(DEBUG_TABS.RESPONSE); setNonConversationResult(result); @@ -1779,7 +1878,7 @@ export default function ModelTester() { } setError(''); - pushDebug('info', `代理请求成功：${envelope.path}`); + pushDebug('info', `代理请求成功：${effectiveEnvelope.path}`); } catch (requestError: any) { const message = requestError?.message || '请求失败'; if (nextMessages) { @@ -1792,7 +1891,7 @@ export default function ModelTester() { } finally { setSending(false); } - }, [pushDebug, startProxyStream]); + }, [attachEnvelopeForcedChannel, pushDebug, startProxyStream]); const buildPayloadWithMessages = useCallback((nextMessages: ChatMessage[]): { payload: TestChatPayload | null; @@ -2380,6 +2479,34 @@ export default function ModelTester() { +
+
+ 固定通道 +
+ { + if (!next || next === '__auto__') { + setForcedChannelId(null); + return; + } + const parsed = Number.parseInt(next, 10); + setForcedChannelId(Number.isFinite(parsed) && parsed > 0 ? parsed : null); + }} + options={forcedChannelSelectOptions} + placeholder={loadingForcedChannels ? '加载通道中...' : '自动选路（默认）'} + disabled={customRequestMode || inputs.mode === 'videos.inspect' || loadingForcedChannels} + emptyLabel="当前模型暂无可固定通道" + menuMaxHeight={300} + /> +
+ {forcedChannelHint + || (typeof forcedChannelId === 'number' + ? `已固定到通道 #${forcedChannelId}，失败不会自动切换。` + : '默认自动选路；如需单独排查，可固定到一个候选通道。')} +
+
+ {inputs.mode === 'conversation' && (
diff --git a/src/web/pages/helpers/modelTesterSession.test.ts b/src/web/pages/helpers/modelTesterSession.test.ts index 2133ec1..1067c7c 100644 --- a/src/web/pages/helpers/modelTesterSession.test.ts +++ b/src/web/pages/helpers/modelTesterSession.test.ts @@ -13,6 +13,7 @@ import { buildGeminiNativeConversationProxyEnvelope, buildRawProxyRequestEnvelope, buildSearchRequestEnvelope, + attachForcedChannelToEnvelope, collectModelTesterModelNames, countConversationTurns, createConversationUserMessage, @@ -74,9 +75,11 @@ describe('modelTesterSession', () => { stream: false, jobMode: false, rawMode: false, + forcedChannelId: 44, jsonBody: { model: '__search', query: 'hello', max_results: 7 }, }, pendingJobId: 'job-1', + forcedChannelId: 44, customRequestMode: true, customRequestBody: '{"model":"gemini-2.5-pro","contents":[]}', showDebugPanel: true, @@ -188,6 +191,29 @@ describe('modelTesterSession', () => { }); }); + it('attaches a forced channel id to tester envelopes without mutating the request body', () => { + const base = buildEmbeddingsRequestEnvelope('hello', { + ...DEFAULT_INPUTS, + model: 'text-embedding-3-small', + }); + + const payload = attachForcedChannelToEnvelope(base, 42); + + expect(payload).toEqual({ + method: 'POST', + path: '/v1/embeddings', + requestKind: 'json', + stream: false, + jobMode: false, + rawMode: false, + forcedChannelId: 42, + jsonBody: { + model: 'text-embedding-3-small', + input: 'hello', + }, + }); + }); + it('builds multipart upload envelopes for /v1/files', () => { expect(buildFileUploadRequestEnvelope({ name: 'paper.pdf', diff --git a/src/web/pages/helpers/modelTesterSession.ts b/src/web/pages/helpers/modelTesterSession.ts index b8ff818..312246f 100644 --- a/src/web/pages/helpers/modelTesterSession.ts +++ b/src/web/pages/helpers/modelTesterSession.ts @@ -93,6 +93,7 @@ export type TesterProxyEnvelope = { stream: boolean; jobMode: boolean; rawMode: boolean; + forcedChannelId?: number | null; jsonBody?: unknown; rawJsonText?: string; multipartFields?: Record; @@ -146,6 +147,7 @@ export type ModelTesterSessionState = { conversationFiles: ConversationDraftFile[]; pendingPayload: TesterProxyEnvelope | null; pendingJobId?: string | null; + forcedChannelId?: number | null; customRequestMode: boolean; customRequestBody: string; showDebugPanel: boolean; @@ -233,6 +235,13 @@ const toNullableFiniteNumber = (value: unknown): number | null => { return null; }; +const toPositiveInteger = (value: unknown): number | null => { + const parsed = toNullableFiniteNumber(value); + if (parsed === null) return null; + const normalized = Math.trunc(parsed); + return normalized > 0 ? normalized : null; +}; + const sanitizeString = (value: unknown, fallback = ''): string => typeof value === 'string' ? value : fallback; @@ -838,6 +847,8 @@ const parsePendingPayload = ( if ('jsonBody' in value) pending.jsonBody = value.jsonBody; if (typeof value.rawJsonText === 'string') pending.rawJsonText = value.rawJsonText; + const forcedChannelId = toPositiveInteger(value.forcedChannelId); + if (forcedChannelId !== null) pending.forcedChannelId = forcedChannelId; if (isRecord(value.multipartFields)) { pending.multipartFields = Object.fromEntries( Object.entries(value.multipartFields) @@ -1152,6 +1163,7 @@ export const parseModelTesterSession = (raw: string | null): ModelTesterSessionS pendingPayload: parsePendingPayload(parsed.pendingPayload, inputs, parameterEnabled), customRequestMode: toBoolean(parsed.customRequestMode, false), customRequestBody: typeof parsed.customRequestBody === 'string' ? parsed.customRequestBody : '', + forcedChannelId: toPositiveInteger(parsed.forcedChannelId), showDebugPanel: toBoolean(parsed.showDebugPanel, false), activeDebugTab: typeof parsed.activeDebugTab === 'string' && VALID_DEBUG_TABS.has(parsed.activeDebugTab) ? parsed.activeDebugTab as DebugTab @@ -1341,6 +1353,23 @@ export const buildApiPayload = ( ): TesterProxyEnvelope => buildConversationRequestEnvelope(messages, inputs, parameterEnabled); +export const attachForcedChannelToEnvelope = ( + envelope: TesterProxyEnvelope, + forcedChannelId?: number | null, +): TesterProxyEnvelope => { + const normalizedForcedChannelId = toPositiveInteger(forcedChannelId); + if (normalizedForcedChannelId === null) { + if (!('forcedChannelId' in envelope)) return envelope; + const { forcedChannelId: _forcedChannelId, ...rest } = envelope; + return rest; + } + + return { + ...envelope, + forcedChannelId: normalizedForcedChannelId, + }; +}; + export const parseCustomRequestBody = (raw: string): Record | null => { if (!raw.trim()) return null; try { @@ -1496,4 +1525,3 @@ export const findLastLoadingAssistantIndex = (messages: ChatMessage[]): number = export const countConversationTurns = (messages: ChatMessage[]): number => messages.reduce((turns, message) => turns + (message.role === 'user' ? 1 : 0), 0); - diff --git a/src/web/pages/modelTester.forced-channel.architecture.test.tsx b/src/web/pages/modelTester.forced-channel.architecture.test.tsx new file mode 100644 index 0000000..8c0e0b0 --- /dev/null +++ b/src/web/pages/modelTester.forced-channel.architecture.test.tsx @@ -0,0 +1,14 @@ +import { describe, expect, it } from 'vitest'; +import { readFileSync } from 'node:fs'; +import { resolve } from 'node:path'; + +describe('ModelTester forced channel architecture', () => { + it('wires a fixed-channel selector through route decisions and tester envelopes', () => { + const source = readFileSync(resolve(process.cwd(), 'src/web/pages/ModelTester.tsx'), 'utf8').replace(/\r\n/g, '\n'); + + expect(source).toContain('固定通道'); + expect(source).toContain('api.getRouteDecision(inputs.model)'); + expect(source).toContain('forcedChannelId'); + expect(source).toContain('attachForcedChannelToEnvelope'); + }); +}); diff --git a/src/web/pages/modelTester.forced-channel.test.tsx b/src/web/pages/modelTester.forced-channel.test.tsx new file mode 100644 index 0000000..483eaed --- /dev/null +++ b/src/web/pages/modelTester.forced-channel.test.tsx @@ -0,0 +1,154 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { act, create, type ReactTestRenderer } from 'react-test-renderer'; +import ModelTester from './ModelTester.js'; +import { + DEBUG_TABS, + DEFAULT_INPUTS, + DEFAULT_MODE_STATE, + DEFAULT_PARAMETER_ENABLED, + MODEL_TESTER_STORAGE_KEY, + serializeModelTesterSession, +} from './helpers/modelTesterSession.js'; + +const { apiMock } = vi.hoisted(() => ({ + apiMock: { + getModelsMarketplace: vi.fn(), + getRoutes: vi.fn(), + getRouteDecision: vi.fn(), + }, +})); + +vi.mock('../api.js', () => ({ + api: apiMock, +})); + +vi.mock('../authSession.js', () => ({ + clearAuthSession: vi.fn(), + getAuthToken: vi.fn(() => null), +})); + +vi.mock('./model-tester/ConversationComposer.js', () => ({ + default: () => null, +})); + +vi.mock('./model-tester/DebugPanel.js', () => ({ + default: () => null, +})); + +vi.mock('../components/useAnimatedVisibility.js', () => ({ + useAnimatedVisibility: () => ({ shouldRender: false, isVisible: false }), +})); + +vi.mock('../components/useIsMobile.js', () => ({ + useIsMobile: () => false, +})); + +vi.mock('../i18n.js', () => ({ + tr: (value: string) => value, +})); + +function collectText(node: any): string { + const children = node?.children || []; + return children.map((child: any) => { + if (typeof child === 'string') return child; + return collectText(child); + }).join(''); +} + +async function flushMicrotasks() { + await act(async () => { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); + }); +} + +describe('ModelTester fixed channel behavior', () => { + const originalLocalStorage = globalThis.localStorage; + + beforeEach(() => { + vi.clearAllMocks(); + apiMock.getModelsMarketplace.mockResolvedValue({ + models: [ + { name: 'gpt-4o-mini' }, + ], + }); + apiMock.getRoutes.mockResolvedValue([]); + apiMock.getRouteDecision.mockResolvedValue({ + decision: { + candidates: [ + { + channelId: 77, + accountId: 12, + username: 'tester', + siteName: 'site-a', + tokenName: 'default', + priority: 0, + eligible: true, + reason: 'ok', + }, + ], + }, + }); + + const session = serializeModelTesterSession({ + input: '', + inputs: { + ...DEFAULT_INPUTS, + model: 'gpt-4o-mini', + }, + parameterEnabled: DEFAULT_PARAMETER_ENABLED, + messages: [], + conversationFiles: [], + pendingPayload: null, + pendingJobId: null, + forcedChannelId: 77, + customRequestMode: false, + customRequestBody: '', + showDebugPanel: false, + activeDebugTab: DEBUG_TABS.PREVIEW, + modeState: DEFAULT_MODE_STATE, + }); + + const storage = new Map([ + [MODEL_TESTER_STORAGE_KEY, session], + ]); + + vi.stubGlobal('localStorage', { + getItem: vi.fn((key: string) => storage.get(key) ?? null), + setItem: vi.fn((key: string, value: string) => { + storage.set(key, value); + }), + removeItem: vi.fn((key: string) => { + storage.delete(key); + }), + }); + }); + + afterEach(() => { + if (originalLocalStorage) { + vi.stubGlobal('localStorage', originalLocalStorage); + } else { + vi.unstubAllGlobals(); + } + vi.clearAllMocks(); + }); + + it('keeps the restored fixed channel selected through initial model hydration', async () => { + let root!: ReactTestRenderer; + + try { + await act(async () => { + root = create(); + }); + await vi.waitFor(async () => { + await flushMicrotasks(); + expect(apiMock.getRouteDecision).toHaveBeenCalledTimes(1); + expect(apiMock.getRouteDecision).toHaveBeenCalledWith('gpt-4o-mini'); + expect(collectText(root.root)).toContain('已固定到通道 #77，失败不会自动切换。'); + }); + } finally { + root?.unmount(); + } + }); +});