LaPaoPao/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests 1 Activity

[codex] Add model tester fixed-channel selection (#352)

Browse Source 
* 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
This commit is contained in:
Cita
2026-04-01 12:56:13 +08:00
committed by GitHub
parent 8c3be1a613
commit bafceb855b
27 changed files with 1331 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/proxy-core/channelSelection.test.ts
+56
View File
@@ -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);
});
});
src/server/proxy-core/channelSelection.ts
+158
View File
@@ -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<ReturnType<typeof tokenRouter.selectChannel>>;
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<string, unknown> | 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<string, unknown>;
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<SelectedChannel> {
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<boolean> => {
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;
}
src/server/proxy-core/providers/codexProviderProfile.ts
+1 -1
View File
@@ -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
src/server/proxy-core/providers/types.ts
+1
View File
@@ -42,6 +42,7 @@ export type PrepareProviderRequestInput = {
claudeHeaders?: Record<string, string>;
codexSessionCacheKey?: string | null;
codexExplicitSessionId?: string | null;
responsesWebsocketTransport?: boolean;
body: Record<string, unknown>;
action?: ProviderAction;
};
src/server/proxy-core/surfaces/chatSurface.ts
+105 -32
View File
@@ -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<string, unknown> {
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<string, unknown>,
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<string, unknown>,
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();
}
src/server/proxy-core/surfaces/geminiSurface.ts
+62 -13
View File
@@ -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<string, unknown>,
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<string, unknown>,
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;
}
src/server/proxy-core/surfaces/openAiResponsesSurface.ts
+104 -47
View File
@@ -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<string, unknown> {
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<string, unknown>,
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();
}
src/server/proxy-core/surfaces/sharedSurface.test.ts
+80 -2
View File
@@ -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
src/server/proxy-core/surfaces/sharedSurface.ts
+3 -38
View File
@@ -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<ReturnType<typeof tokenRouter.selectChannel>>;
type SurfaceWarningScope = 'chat' | 'responses';
@@ -107,44 +107,9 @@ export async function selectSurfaceChannelForAttempt(input: {
excludeChannelIds: number[];
retryCount: number;
stickySessionKey?: string | null;
forcedChannelId?: number | null;
}): Promise<SelectedChannel> {
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: {