1
0
forked from boos/BoosAPI

feat: Enhance Gemini and chat proxy routes with improved error handling and fallback mechanisms

This commit is contained in:
cita
2026-03-08 12:54:24 +08:00
parent 23922a5965
commit 8e291b9bf8
5 changed files with 723 additions and 122 deletions
+167
View File
@@ -1909,6 +1909,173 @@ describe('chat proxy stream behavior', () => {
expect(body?.choices?.[0]?.message?.content).toContain('ok via responses fallback after 502');
});
it('continues to /v1/responses when /v1/messages dispatch is denied for /v1/chat/completions', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'generic-site', url: 'https://generic.example.com', platform: 'new-api' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-generic',
actualModel: 'gpt-5.2-codex',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'Unsupported endpoint /v1/chat/completions', type: 'unsupported_endpoint' },
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'This group does not allow /v1/messages dispatch', type: 'forbidden' },
}), {
status: 403,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'resp_dispatch_denied_fallback',
object: 'response',
model: 'gpt-5.2-codex',
status: 'completed',
output_text: 'ok via responses after messages dispatch denied',
usage: { input_tokens: 5, output_tokens: 2, total_tokens: 7 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'gpt-5.2-codex',
stream: false,
messages: [{ role: 'user', content: 'hello' }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(3);
const [firstUrl] = fetchMock.mock.calls[0] as [string, any];
const [secondUrl] = fetchMock.mock.calls[1] as [string, any];
const [thirdUrl] = fetchMock.mock.calls[2] as [string, any];
expect(firstUrl).toContain('/v1/chat/completions');
expect(secondUrl).toContain('/v1/messages');
expect(thirdUrl).toContain('/v1/responses');
expect(response.json()?.choices?.[0]?.message?.content).toContain('ok via responses');
});
it('prefers /v1/responses immediately after explicit legacy protocol rejection on /v1/chat/completions', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'generic-site', url: 'https://generic.example.com', platform: 'sub2api' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-generic',
actualModel: 'gpt-5.2-codex',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: {
message: 'Unsupported legacy protocol: /v1/chat/completions is not supported. Please use /v1/responses.',
type: 'upstream_error',
},
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'resp_legacy_protocol_preferred',
object: 'response',
model: 'gpt-5.2-codex',
status: 'completed',
output_text: 'ok via direct responses preference',
usage: { input_tokens: 5, output_tokens: 2, total_tokens: 7 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'gpt-5.2-codex',
stream: false,
messages: [{ role: 'user', content: 'hello' }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
const [firstUrl] = fetchMock.mock.calls[0] as [string, any];
const [secondUrl] = fetchMock.mock.calls[1] as [string, any];
expect(firstUrl).toContain('/v1/chat/completions');
expect(secondUrl).toContain('/v1/responses');
expect(secondUrl).not.toContain('/v1/messages');
expect(response.json()?.choices?.[0]?.message?.content).toContain('ok via direct responses preference');
});
it('keeps messages-first semantics for claude-family models on generic upstreams', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'generic-site', url: 'https://generic.example.com', platform: 'new-api' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-generic',
actualModel: 'claude-haiku-4-5-20251001',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'This group does not allow /v1/messages dispatch', type: 'forbidden' },
}), {
status: 403,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
error: {
message: 'Unsupported legacy protocol: /v1/chat/completions is not supported. Please use /v1/responses.',
type: 'upstream_error',
},
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'resp_claude_generic_messages_first',
object: 'response',
model: 'claude-haiku-4-5-20251001',
status: 'completed',
output_text: 'ok via responses after preserving messages-first order',
usage: { input_tokens: 5, output_tokens: 2, total_tokens: 7 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'claude-haiku-4-5-20251001',
stream: false,
messages: [{ role: 'user', content: 'hello' }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(3);
const [firstUrl] = fetchMock.mock.calls[0] as [string, any];
const [secondUrl] = fetchMock.mock.calls[1] as [string, any];
const [thirdUrl] = fetchMock.mock.calls[2] as [string, any];
expect(firstUrl).toContain('/v1/messages');
expect(secondUrl).toContain('/v1/chat/completions');
expect(thirdUrl).toContain('/v1/responses');
expect(response.json()?.choices?.[0]?.message?.content).toContain('ok via responses');
});
it('forces openai platform to use /v1/chat/completions for claude downstream requests', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
+36 -5
View File
@@ -13,9 +13,11 @@ import { type DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
buildMinimalJsonHeadersForCompatibility,
buildUpstreamEndpointRequest,
isEndpointDispatchDeniedError,
isEndpointDowngradeError,
isUnsupportedMediaTypeError,
resolveUpstreamEndpointCandidates,
shouldPreferResponsesAfterLegacyChatError,
} from './upstreamEndpoint.js';
import {
ensureModelAllowedForDownstreamKey,
@@ -110,7 +112,8 @@ async function handleChatProxyRequest(
excludeChannelIds.push(selected.channel.id);
const modelName = selected.actualModel || requestedModel;
const endpointCandidates = await resolveUpstreamEndpointCandidates(
const endpointCandidates = [
...await resolveUpstreamEndpointCandidates(
{
site: selected.site,
account: selected.account,
@@ -118,9 +121,31 @@ async function handleChatProxyRequest(
modelName,
downstreamFormat,
requestedModel,
);
),
];
let startTime = Date.now();
const promoteResponsesCandidate = (currentEndpoint: string | null | undefined, upstreamErrorText?: string | null, status = 0) => {
if (!shouldPreferResponsesAfterLegacyChatError({
status,
upstreamErrorText,
downstreamFormat,
sitePlatform: selected.site.platform,
modelName,
requestedModelHint: requestedModel,
currentEndpoint: currentEndpoint as any,
})) {
return;
}
const currentIndex = endpointCandidates.findIndex((endpoint) => endpoint === currentEndpoint);
const responsesIndex = endpointCandidates.indexOf('responses');
if (currentIndex < 0 || responsesIndex < 0 || responsesIndex <= currentIndex + 1) return;
endpointCandidates.splice(responsesIndex, 1);
endpointCandidates.splice(currentIndex + 1, 0, 'responses');
};
try {
const endpointResult = await executeEndpointFlow({
siteUrl: selected.site.url,
@@ -227,9 +252,15 @@ async function handleChatProxyRequest(
return null;
},
shouldDowngrade: (ctx) => (
ctx.response.status >= 500
|| isEndpointDowngradeError(ctx.response.status, ctx.rawErrText)
|| isMessagesRequiredError(ctx.rawErrText)
(() => {
promoteResponsesCandidate(ctx.request.endpoint, ctx.rawErrText, ctx.response.status);
return (
ctx.response.status >= 500
|| isEndpointDowngradeError(ctx.response.status, ctx.rawErrText)
|| isMessagesRequiredError(ctx.rawErrText)
|| isEndpointDispatchDeniedError(ctx.response.status, ctx.rawErrText)
);
})()
),
onDowngrade: (ctx) => {
logProxy(
+272
View File
@@ -4,6 +4,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vites
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordFailureMock = vi.fn();
const invalidateTokenRouterCacheMock = vi.fn();
const authorizeDownstreamTokenMock = vi.fn();
const consumeManagedKeyRequestMock = vi.fn();
@@ -16,6 +17,7 @@ vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
explainSelection: vi.fn(async () => ({ selectedChannelId: 11 })),
},
invalidateTokenRouterCache: (...args: unknown[]) => invalidateTokenRouterCacheMock(...args),
@@ -39,6 +41,7 @@ describe('gemini native proxy routes', () => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordFailureMock.mockReset();
authorizeDownstreamTokenMock.mockReset();
consumeManagedKeyRequestMock.mockReset();
@@ -58,6 +61,7 @@ describe('gemini native proxy routes', () => {
actualModel: 'gemini-2.5-flash',
});
selectNextChannelMock.mockReturnValue(null);
recordFailureMock.mockResolvedValue(undefined);
});
afterAll(async () => {
@@ -93,6 +97,49 @@ describe('gemini native proxy routes', () => {
});
});
it('falls back to the next channel for listModels when first Gemini channel fails', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'first channel failed' },
}), {
status: 500,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
models: [
{ name: 'models/gemini-2.5-flash', displayName: 'Gemini 2.5 Flash' },
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1beta/models',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
const [firstUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
const [secondUrl] = fetchMock.mock.calls[1] as [string, RequestInit];
expect(firstUrl).toContain('key=gemini-key');
expect(secondUrl).toContain('key=gemini-key-2');
});
it('forwards native generateContent requests through the gemini route group', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
candidates: [
@@ -427,4 +474,229 @@ describe('gemini native proxy routes', () => {
expect(response.body).toContain('"parts":[{"text":"first"},{"text":"second","thoughtSignature":"sig-1"}]');
expect(response.body).not.toContain('\r\n\r\n');
});
it('falls back to the next channel when first Gemini channel returns 400 before any bytes are written', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'bad request on first channel' },
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
candidates: [
{
content: { role: 'model', parts: [{ text: 'ok from fallback' }] },
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
const [firstUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
const [secondUrl] = fetchMock.mock.calls[1] as [string, RequestInit];
expect(firstUrl).toContain('key=gemini-key');
expect(secondUrl).toContain('key=gemini-key-2');
expect(response.json().candidates?.[0]?.content?.parts?.[0]?.text).toContain('ok from fallback');
});
it('falls back to the next channel when first Gemini channel returns 403 before any bytes are written', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'forbidden on first channel' },
}), {
status: 403,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
candidates: [
{
content: { role: 'model', parts: [{ text: 'ok from fallback' }] },
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
});
it('falls back to the next channel when first Gemini channel returns 500 before any bytes are written', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
fetchMock
.mockResolvedValueOnce(new Response('upstream crash', {
status: 500,
headers: { 'content-type': 'text/plain' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
candidates: [
{
content: { role: 'model', parts: [{ text: 'ok from fallback' }] },
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
});
it('falls back to the next channel when first Gemini channel throws before any bytes are written', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
fetchMock
.mockRejectedValueOnce(new Error('socket hang up'))
.mockResolvedValueOnce(new Response(JSON.stringify({
candidates: [
{
content: { role: 'model', parts: [{ text: 'ok from fallback' }] },
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
});
it('falls back to the next channel for SSE requests before any bytes are written', async () => {
selectNextChannelMock.mockReturnValue({
channel: { id: 12, routeId: 22 },
site: { id: 45, name: 'gemini-site-2', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 34, username: 'demo-user-2' },
tokenName: 'fallback',
tokenValue: 'gemini-key-2',
actualModel: 'gemini-2.5-flash',
});
const encoder = new TextEncoder();
const upstreamBody = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode('data: {"responseId":"resp-fallback","candidates":[{"content":{"role":"model","parts":[{"text":"hello from fallback sse"}]},"finishReason":"STOP"}]}\r\n\r\n'));
controller.close();
},
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'upstream unavailable' },
}), {
status: 500,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(upstreamBody, {
status: 200,
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:streamGenerateContent?alt=sse',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
});
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(response.body).toContain('hello from fallback sse');
});
});
+203 -117
View File
@@ -7,6 +7,7 @@ import {
geminiGenerateContentTransformer,
} from '../../transformers/gemini/generate-content/index.js';
const MAX_RETRIES = 2;
const GEMINI_MODEL_PROBES = [
'gemini-2.5-flash',
'gemini-2.0-flash',
@@ -23,6 +24,15 @@ async function selectGeminiChannel(request: FastifyRequest) {
return null;
}
async function selectNextGeminiProbeChannel(request: FastifyRequest, excludeChannelIds: number[]) {
const policy = getDownstreamRoutingPolicy(request);
for (const candidate of GEMINI_MODEL_PROBES) {
const selected = await tokenRouter.selectNextChannel(candidate, excludeChannelIds, policy);
if (selected) return selected;
}
return null;
}
function resolveGeminiApiVersion(request: FastifyRequest): string {
const params = request.params as { geminiApiVersion?: string } | undefined;
return (params?.geminiApiVersion || 'v1beta').trim() || 'v1beta';
@@ -49,23 +59,60 @@ function extractGeminiModelActionPath(request: FastifyRequest, apiVersion: strin
export async function geminiProxyRoute(app: FastifyInstance) {
const listModels = async (request: FastifyRequest, reply: FastifyReply) => {
const selected = await selectGeminiChannel(request);
if (!selected) {
return reply.code(503).send({
error: { message: 'No available channels for Gemini models', type: 'server_error' },
});
}
const apiVersion = resolveGeminiApiVersion(request);
const upstream = await fetch(
geminiGenerateContentTransformer.resolveModelsUrl(selected.site.url, apiVersion, selected.tokenValue),
{ method: 'GET' },
);
const text = await upstream.text();
try {
return reply.code(upstream.status).send(JSON.parse(text));
} catch {
return reply.code(upstream.status).type(upstream.headers.get('content-type') || 'application/json').send(text);
const excludeChannelIds: number[] = [];
let retryCount = 0;
let lastStatus = 503;
let lastText = 'No available channels for Gemini models';
let lastContentType = 'application/json';
while (retryCount <= MAX_RETRIES) {
const selected = retryCount === 0
? await selectGeminiChannel(request)
: await selectNextGeminiProbeChannel(request, excludeChannelIds);
if (!selected) {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
excludeChannelIds.push(selected.channel.id);
try {
const upstream = await fetch(
geminiGenerateContentTransformer.resolveModelsUrl(selected.site.url, apiVersion, selected.tokenValue),
{ method: 'GET' },
);
const text = await upstream.text();
if (!upstream.ok) {
lastStatus = upstream.status;
lastText = text;
lastContentType = upstream.headers.get('content-type') || 'application/json';
await tokenRouter.recordFailure?.(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
}
try {
return reply.code(upstream.status).send(JSON.parse(text));
} catch {
return reply.code(upstream.status).type(upstream.headers.get('content-type') || 'application/json').send(text);
}
} catch (error) {
await tokenRouter.recordFailure?.(selected.channel.id);
lastStatus = 502;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: error instanceof Error ? error.message : 'Gemini upstream request failed',
type: 'upstream_error',
},
});
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
}
}
};
@@ -80,112 +127,151 @@ export async function geminiProxyRoute(app: FastifyInstance) {
}
const policy = getDownstreamRoutingPolicy(request);
const selected = await tokenRouter.selectChannel(requestedModel, policy);
if (!selected) {
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
const excludeChannelIds: number[] = [];
let retryCount = 0;
let lastStatus = 503;
let lastText = 'No available channels for this model';
let lastContentType = 'application/json';
const body = geminiGenerateContentTransformer.inbound.normalizeRequest(
request.body || {},
selected.actualModel || requestedModel,
);
const actualModelAction = modelActionPath.replace(
/^models\/[^:]+/,
`models/${selected.actualModel || requestedModel}`,
);
const query = new URLSearchParams(request.query as Record<string, string>).toString();
const upstream = await fetch(
geminiGenerateContentTransformer.resolveActionUrl(
selected.site.url,
apiVersion,
actualModelAction,
selected.tokenValue,
query,
),
{
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(body),
},
);
const contentType = upstream.headers.get('content-type');
if (geminiGenerateContentTransformer.stream.isSseContentType(contentType)) {
reply.hijack();
reply.raw.statusCode = upstream.status;
reply.raw.setHeader('Content-Type', contentType || 'text/event-stream');
const reader = upstream.body?.getReader();
if (!reader) {
reply.raw.end();
return;
}
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
const decoder = new TextDecoder();
let rest = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
const chunkText = decoder.decode(value, { stream: true });
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + chunkText);
rest = parsed.rest;
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
const tail = decoder.decode();
if (tail) {
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + tail);
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
} finally {
reader.releaseLock();
reply.raw.end();
}
return;
}
const text = await upstream.text();
try {
const parsed = JSON.parse(text);
if (!upstream.ok) {
return reply.code(upstream.status).send(parsed);
while (retryCount <= MAX_RETRIES) {
const selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, policy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, policy);
if (!selected) {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
const aggregateState = geminiGenerateContentTransformer.aggregator.createState();
if (Array.isArray(parsed)) {
for (const chunk of geminiGenerateContentTransformer.stream.parseJsonArrayPayload(parsed)) {
geminiGenerateContentTransformer.aggregator.apply(aggregateState, chunk);
}
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
);
}
excludeChannelIds.push(selected.channel.id);
geminiGenerateContentTransformer.aggregator.apply(aggregateState, parsed);
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
const body = geminiGenerateContentTransformer.inbound.normalizeRequest(
request.body || {},
selected.actualModel || requestedModel,
);
} catch {
return reply.code(upstream.status).type(contentType || 'application/json').send(text);
const actualModelAction = modelActionPath.replace(
/^models\/[^:]+/,
`models/${selected.actualModel || requestedModel}`,
);
const query = new URLSearchParams(request.query as Record<string, string>).toString();
try {
const upstream = await fetch(
geminiGenerateContentTransformer.resolveActionUrl(
selected.site.url,
apiVersion,
actualModelAction,
selected.tokenValue,
query,
),
{
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(body),
},
);
const contentType = upstream.headers.get('content-type') || 'application/json';
if (!upstream.ok) {
lastStatus = upstream.status;
lastContentType = contentType;
lastText = await upstream.text();
await tokenRouter.recordFailure?.(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
try {
return reply.code(lastStatus).send(JSON.parse(lastText));
} catch {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
}
if (geminiGenerateContentTransformer.stream.isSseContentType(contentType)) {
reply.hijack();
reply.raw.statusCode = upstream.status;
reply.raw.setHeader('Content-Type', contentType || 'text/event-stream');
const reader = upstream.body?.getReader();
if (!reader) {
reply.raw.end();
return;
}
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
const decoder = new TextDecoder();
let rest = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
const chunkText = decoder.decode(value, { stream: true });
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + chunkText);
rest = parsed.rest;
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
const tail = decoder.decode();
if (tail) {
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + tail);
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
} finally {
reader.releaseLock();
reply.raw.end();
}
return;
}
const text = await upstream.text();
try {
const parsed = JSON.parse(text);
const aggregateState = geminiGenerateContentTransformer.aggregator.createState();
if (Array.isArray(parsed)) {
for (const chunk of geminiGenerateContentTransformer.stream.parseJsonArrayPayload(parsed)) {
geminiGenerateContentTransformer.aggregator.apply(aggregateState, chunk);
}
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
);
}
geminiGenerateContentTransformer.aggregator.apply(aggregateState, parsed);
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
);
} catch {
return reply.code(upstream.status).type(contentType || 'application/json').send(text);
}
} catch (error) {
lastStatus = 502;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: error instanceof Error ? error.message : 'Gemini upstream request failed',
type: 'upstream_error',
},
});
await tokenRouter.recordFailure?.(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
}
};
@@ -568,6 +568,49 @@ export function isUnsupportedMediaTypeError(status: number, upstreamErrorText?:
);
}
export function isEndpointDispatchDeniedError(status: number, upstreamErrorText?: string | null): boolean {
if (status !== 403) return false;
const text = (upstreamErrorText || '').toLowerCase();
if (!text) return false;
return (
/does\s+not\s+allow\s+\/v1\/[a-z0-9/_:-]+\s+dispatch/i.test(upstreamErrorText || '')
|| text.includes('dispatch denied')
);
}
export function shouldPreferResponsesAfterLegacyChatError(input: {
status: number;
upstreamErrorText?: string | null;
downstreamFormat: EndpointPreference;
sitePlatform?: string | null;
modelName?: string | null;
requestedModelHint?: string | null;
currentEndpoint?: UpstreamEndpoint | null;
}): boolean {
if (input.status < 400) return false;
if (input.downstreamFormat !== 'openai') return false;
if (input.currentEndpoint !== 'chat') return false;
const sitePlatform = normalizePlatformName(input.sitePlatform);
if (sitePlatform === 'openai' || sitePlatform === 'claude' || sitePlatform === 'gemini' || sitePlatform === 'anyrouter') {
return false;
}
const modelName = asTrimmedString(input.modelName);
const requestedModelHint = asTrimmedString(input.requestedModelHint);
if (isClaudeFamilyModel(modelName) || isClaudeFamilyModel(requestedModelHint)) {
return false;
}
const text = (input.upstreamErrorText || '').toLowerCase();
return (
text.includes('unsupported legacy protocol')
&& text.includes('/v1/chat/completions')
&& text.includes('/v1/responses')
);
}
export function isEndpointDowngradeError(status: number, upstreamErrorText?: string | null): boolean {
if (status < 400) return false;
const text = (upstreamErrorText || '').toLowerCase();
@@ -592,6 +635,8 @@ export function isEndpointDowngradeError(status: number, upstreamErrorText?: str
}
return (
isEndpointDispatchDeniedError(status, upstreamErrorText)
||
text.includes('convert_request_failed')
|| text.includes('not found')
|| text.includes('unknown endpoint')