feat: Add support for OpenAI Responses endpoint and enhance proxy retry logic

- Updated proxyRetryPolicy to retry on all 4xx and 5xx status codes.
- Introduced 'responses' target format in TestChatRequestPayload.
- Implemented extractResponsesContent function to handle responses format in ModelTester.
- Enhanced ModelTester to support responses target format and display appropriate labels.
- Updated ProxyLogs to extract and display upstream paths from error messages.
- Added tests to ensure responses target format is preserved during payload sync.
- Created responses proxy route to handle requests to the new OpenAI Responses endpoint.
- Updated upstreamEndpoint logic to include responses endpoint handling.
- Added modelService tests to validate token route rebuilding logic.
This commit is contained in:
cita
2026-03-01 15:21:16 +08:00
parent 44af9a91cf
commit 4e22b379a8
21 changed files with 1794 additions and 56 deletions
+1 -1
View File
@@ -545,7 +545,7 @@ export async function settingsRoutes(app: FastifyInstance) {
successMessage: (currentTask) => {
const rebuild = (currentTask.result as any)?.rebuild;
if (!rebuild) return '缓存清理后重建路由已完成';
return `缓存清理后重建完成:新增路由 ${rebuild.createdRoutes},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
return `缓存清理后重建完成:新增路由 ${rebuild.createdRoutes},移除旧路由 ${rebuild.removedRoutes ?? 0},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
},
failureMessage: (currentTask) => `缓存清理后重建失败:${currentTask.error || 'unknown error'}`,
},
+1 -1
View File
@@ -192,7 +192,7 @@ export async function statsRoutes(app: FastifyInstance) {
successMessage: (currentTask) => {
const rebuild = (currentTask.result as any)?.rebuild;
if (!rebuild) return '模型广场刷新已完成';
return `模型广场刷新完成:新增路由 ${rebuild.createdRoutes},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
return `模型广场刷新完成:新增路由 ${rebuild.createdRoutes},移除旧路由 ${rebuild.removedRoutes ?? 0},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
},
failureMessage: (currentTask) => `模型广场刷新失败:${currentTask.error || 'unknown error'}`,
},
+67 -2
View File
@@ -4,7 +4,7 @@ import { fetch } from 'undici';
import { config } from '../../config.js';
type TestChatMessage = { role: string; content: string };
type TestTargetFormat = 'openai' | 'claude';
type TestTargetFormat = 'openai' | 'claude' | 'responses';
type TestChatRequestBody = {
model?: string;
@@ -84,7 +84,11 @@ const validatePayload = (
return null;
}
const targetFormat: TestTargetFormat = body.targetFormat === 'claude' ? 'claude' : 'openai';
const targetFormat: TestTargetFormat = body.targetFormat === 'claude'
? 'claude'
: body.targetFormat === 'responses'
? 'responses'
: 'openai';
return {
model: body.model,
@@ -146,6 +150,56 @@ const convertOpenAiPayloadToClaudeBody = (
return body;
};
const convertOpenAiPayloadToResponsesBody = (
payload: ValidatedTestChatPayload,
forceStream: boolean,
): Record<string, unknown> => {
const systemContents: string[] = [];
const messages: Array<{ role: 'user' | 'assistant'; content: string }> = [];
for (const message of payload.messages) {
const role = typeof message.role === 'string' ? message.role : 'user';
const content = typeof message.content === 'string' ? message.content : '';
if (!content.trim()) continue;
if (role === 'system') {
systemContents.push(content);
continue;
}
messages.push({
role: role === 'assistant' ? 'assistant' : 'user',
content,
});
}
const body: Record<string, unknown> = {
model: payload.model,
stream: forceStream,
};
if (messages.length === 1 && messages[0].role === 'user' && systemContents.length === 0) {
body.input = messages[0].content;
} else {
body.input = messages;
if (systemContents.length > 0) {
body.instructions = systemContents.join('\n\n');
}
}
if (typeof payload.temperature === 'number' && Number.isFinite(payload.temperature)) {
body.temperature = payload.temperature;
}
if (typeof payload.top_p === 'number' && Number.isFinite(payload.top_p)) {
body.top_p = payload.top_p;
}
body.max_output_tokens = typeof payload.max_tokens === 'number' && Number.isFinite(payload.max_tokens)
? payload.max_tokens
: 4096;
return body;
};
const buildUpstreamRequest = (
payload: ValidatedTestChatPayload,
forceStream: boolean,
@@ -162,6 +216,17 @@ const buildUpstreamRequest = (
};
}
if (payload.targetFormat === 'responses') {
return {
url: `http://127.0.0.1:${config.port}/v1/responses`,
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${config.proxyToken}`,
},
body: convertOpenAiPayloadToResponsesBody(payload, forceStream),
};
}
const { targetFormat: _targetFormat, ...openAiPayload } = payload;
return {
url: `http://127.0.0.1:${config.port}/v1/chat/completions`,
+1 -1
View File
@@ -336,7 +336,7 @@ export async function tokensRoutes(app: FastifyInstance) {
successMessage: (currentTask) => {
const rebuild = (currentTask.result as any)?.rebuild;
if (!rebuild) return '刷新模型并重建路由已完成';
return `刷新模型并重建路由完成:新增路由 ${rebuild.createdRoutes},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
return `刷新模型并重建路由完成:新增路由 ${rebuild.createdRoutes},移除旧路由 ${rebuild.removedRoutes ?? 0},新增通道 ${rebuild.createdChannels},移除通道 ${rebuild.removedChannels}`;
},
failureMessage: (currentTask) => `刷新模型并重建路由失败:${currentTask.error || 'unknown error'}`,
},
+431
View File
@@ -10,6 +10,7 @@ const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async (_arg?: any) => 0);
const fetchModelPricingCatalogMock = vi.fn(async (_arg?: any): Promise<any> => null);
const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) => ({
...usage,
estimatedCostFromQuota: 0,
@@ -49,6 +50,7 @@ vi.mock('../../services/alertRules.js', () => ({
vi.mock('../../services/modelPricingService.js', () => ({
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
fetchModelPricingCatalog: (arg: any) => fetchModelPricingCatalogMock(arg),
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
@@ -73,9 +75,11 @@ describe('chat proxy stream behavior', () => {
beforeAll(async () => {
const { chatProxyRoute, claudeMessagesProxyRoute } = await import('./chat.js');
const { responsesProxyRoute } = await import('./responses.js');
app = Fastify();
await app.register(chatProxyRoute);
await app.register(claudeMessagesProxyRoute);
await app.register(responsesProxyRoute);
});
beforeEach(() => {
@@ -88,6 +92,7 @@ describe('chat proxy stream behavior', () => {
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
fetchModelPricingCatalogMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
dbInsertMock.mockClear();
@@ -100,6 +105,7 @@ describe('chat proxy stream behavior', () => {
actualModel: 'upstream-gpt',
});
selectNextChannelMock.mockReturnValue(null);
fetchModelPricingCatalogMock.mockResolvedValue(null);
});
afterAll(async () => {
@@ -141,6 +147,23 @@ describe('chat proxy stream behavior', () => {
expect(response.body).toContain('data: [DONE]');
});
it('returns clear 400 when /v1/chat/completions receives responses-style input without messages', async () => {
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'claude-haiku-4-5-20251001',
input: 'hello',
},
});
expect(response.statusCode).toBe(400);
const body = response.json();
expect(body?.error?.type).toBe('invalid_request_error');
expect(body?.error?.message).toContain('/v1/responses');
expect(fetchMock).not.toHaveBeenCalled();
});
it('sets anti-buffering SSE headers for streamed chat responses', async () => {
const encoder = new TextEncoder();
const upstreamBody = new ReadableStream<Uint8Array>({
@@ -314,4 +337,412 @@ describe('chat proxy stream behavior', () => {
expect(response.body).toContain('\"text\":\"hello\"');
expect(response.body).toContain('event: message_stop');
});
it('serves /v1/responses via protocol translation when upstream is OpenAI-compatible', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'resp_123',
object: 'response',
output_text: 'hello from responses',
usage: { input_tokens: 10, output_tokens: 5, total_tokens: 15 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.2',
input: 'hello',
},
});
expect(response.statusCode).toBe(200);
const body = response.json();
expect(body.object).toBe('response');
expect(body.output_text).toContain('hello from responses');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/chat/completions');
const forwarded = JSON.parse(options.body);
expect(forwarded.model).toBe('upstream-gpt');
});
it('passes through /v1/responses SSE payloads', async () => {
const encoder = new TextEncoder();
const upstreamBody = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode('event: response.output_text.delta\ndata: {"type":"response.output_text.delta","delta":"hello"}\n\n'));
controller.enqueue(encoder.encode('data: [DONE]\n\n'));
controller.close();
},
});
fetchMock.mockResolvedValue(new Response(upstreamBody, {
status: 200,
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.2',
input: 'hello',
stream: true,
},
});
expect(response.statusCode).toBe(200);
expect(response.headers['content-type']).toContain('text/event-stream');
expect(response.body).toContain('response.output_text.delta');
expect(response.body).toContain('[DONE]');
});
it('routes /v1/responses to /v1/messages when upstream catalog is anthropic-only', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['anthropic'],
},
],
groupRatio: {},
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'msg_900',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'hello from anthropic messages upstream' }],
stop_reason: 'end_turn',
usage: { input_tokens: 7, output_tokens: 3, total_tokens: 10 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'claude-haiku-4-5-20251001',
input: 'hello',
},
});
expect(response.statusCode).toBe(200);
const body = response.json();
expect(body.object).toBe('response');
expect(body.output_text).toContain('hello from anthropic messages upstream');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/messages');
});
it('forces anyrouter platform to prefer /v1/messages even when catalog says openai', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'anyrouter-site', url: 'https://anyrouter.example.com', platform: 'anyrouter' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'upstream-gpt',
});
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['openai'],
},
],
groupRatio: {},
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'msg_anyrouter',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'anyrouter prefers messages' }],
stop_reason: 'end_turn',
usage: { input_tokens: 6, output_tokens: 2, total_tokens: 8 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('anyrouter prefers messages');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/messages');
});
it('chooses /v1/messages upstream when catalog indicates messages-only endpoint support', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['/v1/messages'],
},
],
groupRatio: {},
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'msg_100',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'hello from messages only' }],
stop_reason: 'end_turn',
usage: { input_tokens: 8, output_tokens: 4, total_tokens: 12 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('hello from messages only');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/messages');
});
it('prefers OpenAI-compatible endpoint when catalog uses generic openai/anthropic labels', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['anthropic', 'openai'],
},
],
groupRatio: {},
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'chatcmpl-openai-first',
object: 'chat.completion',
created: 1_706_000_002,
model: 'upstream-gpt',
choices: [{
index: 0,
message: { role: 'assistant', content: 'hello from openai endpoint' },
finish_reason: 'stop',
}],
usage: { prompt_tokens: 5, completion_tokens: 3, total_tokens: 8 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('hello from openai endpoint');
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/chat/completions');
});
it('falls back to /v1/messages when catalog only declares openai and chat endpoint fails', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['openai'],
},
],
groupRatio: {},
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: {
message: 'openai_error',
type: 'bad_response_status_code',
code: 'bad_response_status_code',
},
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'msg_fallback_500',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'fallback to messages from openai-only catalog' }],
stop_reason: 'end_turn',
usage: { input_tokens: 12, output_tokens: 7, total_tokens: 19 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('fallback to messages from openai-only catalog');
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/messages');
});
it('downgrades endpoint when upstream returns convert_request_failed/not implemented', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['/v1/chat/completions', '/v1/messages'],
},
],
groupRatio: {},
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: {
message: 'not implemented (request id: abc123)',
type: 'new_api_error',
code: 'convert_request_failed',
},
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'msg_200',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'fallback from messages' }],
stop_reason: 'end_turn',
usage: { input_tokens: 11, output_tokens: 6, total_tokens: 17 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('fallback from messages');
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/messages');
});
it('downgrades endpoint when upstream returns openai_error bad_response_status_code', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
{
modelName: 'upstream-gpt',
supportedEndpointTypes: ['/v1/chat/completions', '/v1/messages'],
},
],
groupRatio: {},
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: {
message: 'openai_error',
type: 'bad_response_status_code',
code: 'bad_response_status_code',
},
}), {
status: 400,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'msg_300',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'fallback from bad_response_status_code' }],
stop_reason: 'end_turn',
usage: { input_tokens: 9, output_tokens: 5, total_tokens: 14 },
}), {
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);
const body = response.json();
expect(body?.choices?.[0]?.message?.content).toContain('fallback from bad_response_status_code');
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/messages');
});
});
+80 -26
View File
@@ -22,9 +22,18 @@ import {
serializeFinalResponse,
buildSyntheticOpenAiChunks,
} from './chatFormats.js';
import {
buildUpstreamEndpointRequest,
isEndpointDowngradeError,
resolveUpstreamEndpointCandidates,
} from './upstreamEndpoint.js';
const MAX_RETRIES = 2;
function withUpstreamPath(path: string, message: string): string {
return `[upstream:${path}] ${message}`;
}
export async function chatProxyRoute(app: FastifyInstance) {
app.post('/v1/chat/completions', async (request: FastifyRequest, reply: FastifyReply) =>
handleChatProxyRequest(request, reply, 'openai'));
@@ -72,55 +81,94 @@ async function handleChatProxyRequest(
excludeChannelIds.push(selected.channel.id);
const targetUrl = `${selected.site.url}/v1/chat/completions`;
const forwardBody = {
...upstreamBody,
model: selected.actualModel,
stream: isStream,
};
const startTime = Date.now();
const modelName = selected.actualModel || requestedModel;
const endpointCandidates = await resolveUpstreamEndpointCandidates(
{
site: selected.site,
account: selected.account,
},
modelName,
downstreamFormat,
);
let startTime = Date.now();
try {
const upstream = await fetch(targetUrl, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify(forwardBody),
});
let upstream: Awaited<ReturnType<typeof fetch>> | null = null;
let successfulUpstreamPath: string | null = null;
let finalStatus = 0;
let finalErrText = 'unknown error';
if (!upstream.ok) {
const errText = await upstream.text().catch(() => 'unknown error');
for (let endpointIndex = 0; endpointIndex < endpointCandidates.length; endpointIndex += 1) {
const endpointRequest = buildUpstreamEndpointRequest({
endpoint: endpointCandidates[endpointIndex],
modelName,
stream: isStream,
tokenValue: selected.tokenValue,
openaiBody: upstreamBody,
});
const targetUrl = `${selected.site.url}${endpointRequest.path}`;
startTime = Date.now();
const response = await fetch(targetUrl, {
method: 'POST',
headers: endpointRequest.headers,
body: JSON.stringify(endpointRequest.body),
});
if (response.ok) {
upstream = response;
successfulUpstreamPath = endpointRequest.path;
break;
}
const rawErrText = await response.text().catch(() => 'unknown error');
const errText = withUpstreamPath(endpointRequest.path, rawErrText);
const shouldDowngradeEndpoint = (
endpointIndex < endpointCandidates.length - 1
&& isEndpointDowngradeError(response.status, rawErrText)
);
if (shouldDowngradeEndpoint) {
logProxy(selected, requestedModel, 'failed', response.status, Date.now() - startTime, errText, retryCount);
continue;
}
finalStatus = response.status;
finalErrText = errText;
break;
}
if (!upstream) {
const status = finalStatus || 502;
const errText = finalErrText || 'unknown error';
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount);
logProxy(selected, requestedModel, 'failed', status, Date.now() - startTime, errText, retryCount);
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
if (isTokenExpiredError({ status, message: errText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
detail: `HTTP ${status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
if (shouldRetryProxyRequest(status, errText) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
reason: `upstream returned HTTP ${status}`,
});
return reply.code(upstream.status).send({
return reply.code(status).send({
error: { message: errText, type: 'upstream_error' },
});
}
const modelName = selected.actualModel || requestedModel;
if (isStream) {
reply.hijack();
reply.raw.statusCode = 200;
@@ -232,6 +280,7 @@ async function handleChatProxyRequest(
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
successfulUpstreamPath,
);
return;
}
@@ -348,6 +397,7 @@ async function handleChatProxyRequest(
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
successfulUpstreamPath,
);
return;
}
@@ -407,6 +457,7 @@ async function handleChatProxyRequest(
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
successfulUpstreamPath,
);
return reply.send(downstreamResponse);
@@ -446,8 +497,11 @@ function logProxy(
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
upstreamPath: string | null = null,
) {
try {
const normalizedErrorMessage = errorMessage
|| (upstreamPath ? `[upstream:${upstreamPath}]` : null);
db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
@@ -461,7 +515,7 @@ function logProxy(
completionTokens,
totalTokens,
estimatedCost,
errorMessage,
errorMessage: normalizedErrorMessage,
retryCount,
}).run();
} catch {}
+58
View File
@@ -247,6 +247,21 @@ function parseClaudeMessageContent(content: unknown): string {
return extractTextAndReasoning(content).content;
}
function parseResponsesOutputText(payload: Record<string, unknown>): string {
const direct = typeof payload.output_text === 'string' ? payload.output_text : '';
if (direct) return direct;
const output = Array.isArray(payload.output) ? payload.output : [];
const parts: string[] = [];
for (const item of output) {
if (!isRecord(item)) continue;
const parsed = extractTextAndReasoning(item.content ?? item);
if (parsed.content) parts.push(parsed.content);
}
return parts.join('\n\n');
}
function convertClaudeRequestToOpenAiBody(body: Record<string, unknown>): {
model: string;
stream: boolean;
@@ -354,6 +369,19 @@ export function parseDownstreamChatRequest(
};
}
const hasMessages = Array.isArray(raw.messages) && raw.messages.length > 0;
if (!hasMessages) {
const hint = raw.input !== undefined
? 'messages is required for /v1/chat/completions. For Responses payload, use /v1/responses.'
: 'messages is required';
return {
error: {
statusCode: 400,
payload: { error: { message: hint, type: 'invalid_request_error' } },
},
};
}
return {
value: {
requestedModel: model,
@@ -396,6 +424,17 @@ export function normalizeUpstreamFinalResponse(
};
}
if (isRecord(payload) && ((payload as any).object === 'response' || Array.isArray((payload as any).output))) {
return {
id: isNonEmptyString(payload.id) ? payload.id : fallbackId,
model: isNonEmptyString(payload.model) ? payload.model : fallbackModel,
created: ensureIntegerTimestamp(payload.created, now),
content: parseResponsesOutputText(payload) || fallbackText,
reasoningContent: '',
finishReason: normalizeStopReason(payload.finish_reason ?? payload.status) || 'stop',
};
}
if (isRecord(payload) && Array.isArray(payload.candidates)) {
const candidate = payload.candidates[0] || {};
const parsedCandidate = extractTextAndReasoning(candidate?.content?.parts || candidate?.content);
@@ -468,6 +507,25 @@ export function normalizeUpstreamStreamEvent(
}
const type = typeof payload.type === 'string' ? payload.type : '';
if (type.startsWith('response.output_text')) {
const deltaText = typeof payload.delta === 'string'
? payload.delta
: extractTextAndReasoning(payload.delta).content;
return {
contentDelta: deltaText || undefined,
};
}
if (type === 'response.completed' && isRecord((payload as any).response)) {
const responsePayload = (payload as any).response as Record<string, unknown>;
if (isNonEmptyString(responsePayload.id)) context.id = responsePayload.id;
if (isNonEmptyString(responsePayload.model)) context.model = responsePayload.model;
return {
finishReason: normalizeStopReason(responsePayload.status) || 'stop',
done: true,
};
}
const message = isRecord(payload.message) ? payload.message : null;
if (message) {
+441
View File
@@ -0,0 +1,441 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
import { estimateProxyCost } from '../../services/modelPricingService.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
import { normalizeUpstreamFinalResponse } from './chatFormats.js';
import {
buildUpstreamEndpointRequest,
isEndpointDowngradeError,
resolveUpstreamEndpointCandidates,
type UpstreamEndpoint,
} from './upstreamEndpoint.js';
const MAX_RETRIES = 2;
function withUpstreamPath(path: string, message: string): string {
return `[upstream:${path}] ${message}`;
}
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function normalizeText(value: unknown): string {
if (typeof value === 'string') return value;
if (Array.isArray(value)) {
return value
.map((item) => normalizeText(item))
.filter((item) => item.length > 0)
.join('\n');
}
if (isRecord(value)) {
if (typeof value.text === 'string') return value.text;
if (typeof value.content === 'string') return value.content;
if (typeof value.input_text === 'string') return value.input_text;
if (typeof value.output_text === 'string') return value.output_text;
if (Array.isArray(value.content)) return normalizeText(value.content);
}
return '';
}
function convertResponsesBodyToOpenAiBody(
body: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const messages: Array<{ role: string; content: string }> = [];
const input = body.input;
if (typeof input === 'string') {
const text = input.trim();
if (text) messages.push({ role: 'user', content: text });
} else if (Array.isArray(input)) {
for (const item of input) {
if (!isRecord(item)) continue;
const role = typeof item.role === 'string' ? item.role : 'user';
const text = normalizeText(item.content ?? item).trim();
if (!text) continue;
messages.push({ role: role === 'assistant' ? 'assistant' : (role === 'system' ? 'system' : 'user'), content: text });
}
} else if (isRecord(input)) {
const text = normalizeText(input).trim();
if (text) messages.push({ role: 'user', content: text });
}
const instructions = typeof body.instructions === 'string' ? body.instructions.trim() : '';
if (instructions) {
messages.unshift({ role: 'system', content: instructions });
}
const payload: Record<string, unknown> = {
model: modelName,
stream,
messages,
};
if (typeof body.temperature === 'number' && Number.isFinite(body.temperature)) {
payload.temperature = body.temperature;
}
if (typeof body.top_p === 'number' && Number.isFinite(body.top_p)) {
payload.top_p = body.top_p;
}
if (typeof body.max_output_tokens === 'number' && Number.isFinite(body.max_output_tokens)) {
payload.max_tokens = body.max_output_tokens;
}
if (body.tools !== undefined) payload.tools = body.tools;
if (body.tool_choice !== undefined) payload.tool_choice = body.tool_choice;
return payload;
}
function toResponsesPayload(
normalized: ReturnType<typeof normalizeUpstreamFinalResponse>,
usage: { promptTokens: number; completionTokens: number; totalTokens: number },
): Record<string, unknown> {
const normalizedId = typeof normalized.id === 'string' && normalized.id.trim()
? normalized.id.trim()
: `resp_${Date.now()}`;
const responseId = normalizedId.startsWith('resp_') ? normalizedId : `resp_${normalizedId}`;
const messageId = normalizedId.startsWith('msg_') ? normalizedId : `msg_${normalizedId}`;
return {
id: responseId,
object: 'response',
created: normalized.created,
status: 'completed',
model: normalized.model,
output: [{
id: messageId,
type: 'message',
role: 'assistant',
status: 'completed',
content: [{
type: 'output_text',
text: normalized.content || '',
}],
}],
output_text: normalized.content || '',
usage: {
input_tokens: usage.promptTokens,
output_tokens: usage.completionTokens,
total_tokens: usage.totalTokens,
},
};
}
export async function responsesProxyRoute(app: FastifyInstance) {
app.post('/v1/responses', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any;
const requestedModel = typeof body?.model === 'string' ? body.model.trim() : '';
if (!requestedModel) {
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
}
const isStream = body.stream === true;
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? tokenRouter.selectChannel(requestedModel)
: tokenRouter.selectNextChannel(requestedModel, excludeChannelIds);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = tokenRouter.selectChannel(requestedModel);
}
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' },
});
}
excludeChannelIds.push(selected.channel.id);
const modelName = selected.actualModel || requestedModel;
const openAiBody = convertResponsesBodyToOpenAiBody(body, modelName, isStream);
const resolvedCandidates = await resolveUpstreamEndpointCandidates(
{
site: selected.site,
account: selected.account,
},
modelName,
'openai',
);
const endpointCandidates = resolvedCandidates
.filter((endpoint) => endpoint !== 'responses');
if (endpointCandidates.length === 0) {
endpointCandidates.push('chat', 'messages');
}
const startTime = Date.now();
try {
let upstream: Awaited<ReturnType<typeof fetch>> | null = null;
let successfulUpstreamPath: string | null = null;
let finalStatus = 0;
let finalErrText = 'unknown error';
for (let endpointIndex = 0; endpointIndex < endpointCandidates.length; endpointIndex += 1) {
const endpoint = endpointCandidates[endpointIndex] as UpstreamEndpoint;
const endpointRequest = buildUpstreamEndpointRequest({
endpoint,
modelName,
stream: isStream,
tokenValue: selected.tokenValue,
openaiBody: openAiBody,
});
const targetUrl = `${selected.site.url}${endpointRequest.path}`;
const response = await fetch(targetUrl, {
method: 'POST',
headers: endpointRequest.headers,
body: JSON.stringify(endpointRequest.body),
});
if (response.ok) {
upstream = response;
successfulUpstreamPath = endpointRequest.path;
break;
}
const rawErrText = await response.text().catch(() => 'unknown error');
const errText = withUpstreamPath(endpointRequest.path, rawErrText);
const shouldDowngradeEndpoint = (
endpointIndex < endpointCandidates.length - 1
&& isEndpointDowngradeError(response.status, rawErrText)
);
if (shouldDowngradeEndpoint) {
logProxy(selected, requestedModel, 'failed', response.status, Date.now() - startTime, errText, retryCount);
continue;
}
finalStatus = response.status;
finalErrText = errText;
break;
}
if (!upstream) {
const status = finalStatus || 502;
const errText = finalErrText || 'unknown error';
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', status, Date.now() - startTime, errText, retryCount);
if (isTokenExpiredError({ status, message: errText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if (shouldRetryProxyRequest(status, errText) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${status}`,
});
return reply.code(status).send({ error: { message: errText, type: 'upstream_error' } });
}
if (isStream) {
reply.raw.writeHead(200, {
'Content-Type': 'text/event-stream',
'Cache-Control': 'no-cache',
Connection: 'keep-alive',
});
const reader = upstream.body?.getReader();
if (!reader) {
reply.raw.end();
return;
}
const decoder = new TextDecoder();
let parsedUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 };
let sseBuffer = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = decoder.decode(value, { stream: true });
reply.raw.write(chunk);
sseBuffer += chunk;
const pulled = pullSseDataEvents(sseBuffer);
sseBuffer = pulled.rest;
for (const eventPayload of pulled.events) {
try {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
} catch {}
}
}
if (sseBuffer.trim().length > 0) {
const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
for (const eventPayload of pulled.events) {
try {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
} catch {}
}
}
} finally {
reader.releaseLock();
reply.raw.end();
}
const latency = Date.now() - startTime;
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
let estimatedCost = await estimateProxyCost({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
});
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
estimatedCost = resolvedUsage.estimatedCostFromQuota;
}
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost,
successfulUpstreamPath,
);
return;
}
const rawText = await upstream.text();
let upstreamData: unknown = rawText;
try {
upstreamData = JSON.parse(rawText);
} catch {
upstreamData = rawText;
}
const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(upstreamData);
const normalized = normalizeUpstreamFinalResponse(
upstreamData,
modelName,
rawText,
);
const downstreamData = toResponsesPayload(normalized, parsedUsage);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
let estimatedCost = await estimateProxyCost({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
});
if (resolvedUsage.estimatedCostFromQuota > 0 && (resolvedUsage.recoveredFromSelfLog || estimatedCost <= 0)) {
estimatedCost = resolvedUsage.estimatedCostFromQuota;
}
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost,
successfulUpstreamPath,
);
return reply.send(downstreamData);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
});
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
});
}
}
});
}
function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
upstreamPath: string | null = null,
) {
try {
const normalizedErrorMessage = errorMessage
|| (upstreamPath ? `[upstream:${upstreamPath}]` : null);
db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
errorMessage: normalizedErrorMessage,
retryCount,
}).run();
} catch {}
}
+2
View File
@@ -4,6 +4,7 @@ import { chatProxyRoute, claudeMessagesProxyRoute } from './chat.js';
import { modelsProxyRoute } from './models.js';
import { embeddingsProxyRoute } from './embeddings.js';
import { completionsProxyRoute } from './completions.js';
import { responsesProxyRoute } from './responses.js';
import { imagesProxyRoute } from './images.js';
export async function proxyRoutes(app: FastifyInstance) {
@@ -15,6 +16,7 @@ export async function proxyRoutes(app: FastifyInstance) {
await app.register(chatProxyRoute);
await app.register(claudeMessagesProxyRoute);
await app.register(completionsProxyRoute);
await app.register(responsesProxyRoute);
await app.register(modelsProxyRoute);
await app.register(embeddingsProxyRoute);
await app.register(imagesProxyRoute);
+345
View File
@@ -0,0 +1,345 @@
import { fetchModelPricingCatalog } from '../../services/modelPricingService.js';
import type { DownstreamFormat } from './chatFormats.js';
export type UpstreamEndpoint = 'chat' | 'messages' | 'responses';
type ChannelContext = {
site: {
id: number;
url: string;
platform: string;
apiKey?: string | null;
};
account: {
id: number;
accessToken?: string | null;
apiToken?: string | null;
};
};
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function toFiniteNumber(value: unknown): number | null {
return typeof value === 'number' && Number.isFinite(value) ? value : null;
}
function normalizeContentText(content: unknown): string {
if (typeof content === 'string') return content;
if (Array.isArray(content)) {
return content
.map((item) => {
if (typeof item === 'string') return item;
if (!isRecord(item)) return '';
const text = asTrimmedString(item.text ?? item.content ?? item.output_text);
return text;
})
.filter((item) => item.length > 0)
.join('\n');
}
if (isRecord(content)) {
const text = asTrimmedString(content.text ?? content.content ?? content.output_text);
return text;
}
return '';
}
function convertOpenAiBodyToMessagesBody(
openaiBody: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const rawMessages = Array.isArray(openaiBody.messages) ? openaiBody.messages : [];
const systemContents: string[] = [];
const messages: Array<{ role: 'user' | 'assistant'; content: string }> = [];
for (const item of rawMessages) {
if (!isRecord(item)) continue;
const role = asTrimmedString(item.role) || 'user';
const content = normalizeContentText(item.content);
if (!content) continue;
if (role === 'system') {
systemContents.push(content);
continue;
}
messages.push({
role: role === 'assistant' ? 'assistant' : 'user',
content,
});
}
const body: Record<string, unknown> = {
model: modelName,
stream,
messages,
max_tokens: toFiniteNumber(openaiBody.max_tokens) ?? 4096,
};
if (systemContents.length > 0) {
body.system = systemContents.join('\n\n');
}
const temperature = toFiniteNumber(openaiBody.temperature);
if (temperature !== null) body.temperature = temperature;
const topP = toFiniteNumber(openaiBody.top_p);
if (topP !== null) body.top_p = topP;
if (Array.isArray(openaiBody.stop) && openaiBody.stop.length > 0) {
body.stop_sequences = openaiBody.stop;
}
if (openaiBody.tools !== undefined) body.tools = openaiBody.tools;
if (openaiBody.tool_choice !== undefined) body.tool_choice = openaiBody.tool_choice;
return body;
}
function convertOpenAiBodyToResponsesBody(
openaiBody: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const rawMessages = Array.isArray(openaiBody.messages) ? openaiBody.messages : [];
const systemContents: string[] = [];
const messages: Array<{ role: 'user' | 'assistant'; content: string }> = [];
for (const item of rawMessages) {
if (!isRecord(item)) continue;
const role = asTrimmedString(item.role) || 'user';
const content = normalizeContentText(item.content);
if (!content) continue;
if (role === 'system') {
systemContents.push(content);
continue;
}
messages.push({
role: role === 'assistant' ? 'assistant' : 'user',
content,
});
}
const body: Record<string, unknown> = {
model: modelName,
stream,
max_output_tokens: toFiniteNumber(openaiBody.max_tokens) ?? 4096,
};
if (messages.length === 1 && messages[0].role === 'user' && systemContents.length === 0) {
body.input = messages[0].content;
} else {
body.input = messages;
if (systemContents.length > 0) {
body.instructions = systemContents.join('\n\n');
}
}
const temperature = toFiniteNumber(openaiBody.temperature);
if (temperature !== null) body.temperature = temperature;
const topP = toFiniteNumber(openaiBody.top_p);
if (topP !== null) body.top_p = topP;
if (openaiBody.tools !== undefined) body.tools = openaiBody.tools;
if (openaiBody.tool_choice !== undefined) body.tool_choice = openaiBody.tool_choice;
return body;
}
function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] {
const raw = asTrimmedString(value).toLowerCase();
if (!raw) return [];
const normalized = new Set<UpstreamEndpoint>();
if (
raw.includes('/v1/messages')
|| raw === 'messages'
|| raw.includes('anthropic')
|| raw.includes('claude')
) {
normalized.add('messages');
}
if (
raw.includes('/v1/responses')
|| raw === 'responses'
|| raw.includes('response')
) {
// Treat responses family as OpenAI-compatible for upstream selection.
normalized.add('chat');
}
if (
raw.includes('/v1/chat/completions')
|| raw.includes('chat/completions')
|| raw === 'chat'
|| raw === 'chat_completions'
|| raw === 'completions'
|| raw.includes('chat')
) {
normalized.add('chat');
}
// Some upstreams return protocol families instead of concrete endpoint paths.
if (raw === 'openai' || raw.includes('openai')) {
normalized.add('chat');
}
return Array.from(normalized);
}
function preferredEndpointOrder(downstreamFormat: DownstreamFormat): UpstreamEndpoint[] {
return downstreamFormat === 'claude'
? ['messages', 'chat']
: ['chat', 'messages'];
}
export async function resolveUpstreamEndpointCandidates(
context: ChannelContext,
modelName: string,
downstreamFormat: DownstreamFormat,
): Promise<UpstreamEndpoint[]> {
if ((context.site.platform || '').toLowerCase() === 'anyrouter') {
// anyrouter deployments are effectively anthropic-protocol first.
return ['messages', 'chat'];
}
const preferred = preferredEndpointOrder(downstreamFormat);
try {
const catalog = await fetchModelPricingCatalog({
site: {
id: context.site.id,
url: context.site.url,
platform: context.site.platform,
apiKey: context.site.apiKey ?? null,
},
account: {
id: context.account.id,
accessToken: context.account.accessToken ?? null,
apiToken: context.account.apiToken ?? null,
},
modelName,
totalTokens: 0,
});
if (!catalog || !Array.isArray(catalog.models) || catalog.models.length === 0) {
return preferred;
}
const matched = catalog.models.find((item) =>
asTrimmedString(item?.modelName).toLowerCase() === modelName.toLowerCase(),
);
if (!matched) return preferred;
const supportedRaw = Array.isArray(matched.supportedEndpointTypes) ? matched.supportedEndpointTypes : [];
const supported = new Set<UpstreamEndpoint>();
for (const endpoint of supportedRaw) {
const normalizedList = normalizeEndpointTypes(endpoint);
for (const normalized of normalizedList) {
supported.add(normalized);
}
}
if (supported.size === 0) return preferred;
const filtered = preferred.filter((endpoint) => supported.has(endpoint));
if (filtered.length === 0) return preferred;
// Keep non-catalog endpoints as best-effort fallbacks because some
// upstreams expose incomplete/incorrect endpoint metadata.
const fallback = preferred.filter((endpoint) => !filtered.includes(endpoint));
return [...filtered, ...fallback];
} catch {
return preferred;
}
}
export function buildUpstreamEndpointRequest(input: {
endpoint: UpstreamEndpoint;
modelName: string;
stream: boolean;
tokenValue: string;
openaiBody: Record<string, unknown>;
}): {
path: string;
headers: Record<string, string>;
body: Record<string, unknown>;
} {
const commonHeaders = {
'Content-Type': 'application/json',
Authorization: `Bearer ${input.tokenValue}`,
};
if (input.endpoint === 'messages') {
return {
path: '/v1/messages',
headers: {
...commonHeaders,
'x-api-key': input.tokenValue,
'anthropic-version': '2023-06-01',
},
body: convertOpenAiBodyToMessagesBody(input.openaiBody, input.modelName, input.stream),
};
}
if (input.endpoint === 'responses') {
return {
path: '/v1/responses',
headers: commonHeaders,
body: convertOpenAiBodyToResponsesBody(input.openaiBody, input.modelName, input.stream),
};
}
return {
path: '/v1/chat/completions',
headers: commonHeaders,
body: {
...input.openaiBody,
model: input.modelName,
stream: input.stream,
},
};
}
export function isEndpointDowngradeError(status: number, upstreamErrorText?: string | null): boolean {
if (status < 400) return false;
const text = (upstreamErrorText || '').toLowerCase();
if (!text) return false;
let parsedCode = '';
let parsedType = '';
let parsedMessage = '';
try {
const parsed = JSON.parse(upstreamErrorText || '{}') as Record<string, unknown>;
const error = (parsed.error && typeof parsed.error === 'object')
? parsed.error as Record<string, unknown>
: parsed;
parsedCode = asTrimmedString(error.code).toLowerCase();
parsedType = asTrimmedString(error.type).toLowerCase();
parsedMessage = asTrimmedString(error.message).toLowerCase();
} catch {
parsedCode = '';
parsedType = '';
parsedMessage = '';
}
return (
text.includes('convert_request_failed')
|| text.includes('not implemented')
|| text.includes('api not implemented')
|| text.includes('unsupported legacy protocol')
|| parsedCode === 'convert_request_failed'
|| parsedCode === 'bad_response_status_code'
|| parsedType === 'bad_response_status_code'
|| parsedMessage.includes('bad_response_status_code')
);
}
+123
View File
@@ -0,0 +1,123 @@
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import { mkdtempSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { and, eq } from 'drizzle-orm';
type DbModule = typeof import('../db/index.js');
type ModelServiceModule = typeof import('./modelService.js');
describe('rebuildTokenRoutesFromAvailability', () => {
let db: DbModule['db'];
let schema: DbModule['schema'];
let rebuildTokenRoutesFromAvailability: ModelServiceModule['rebuildTokenRoutesFromAvailability'];
let dataDir = '';
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-model-service-'));
process.env.DATA_DIR = dataDir;
await import('../db/migrate.js');
const dbModule = await import('../db/index.js');
const modelService = await import('./modelService.js');
db = dbModule.db;
schema = dbModule.schema;
rebuildTokenRoutesFromAvailability = modelService.rebuildTokenRoutesFromAvailability;
});
beforeEach(() => {
db.delete(schema.routeChannels).run();
db.delete(schema.tokenRoutes).run();
db.delete(schema.tokenModelAvailability).run();
db.delete(schema.modelAvailability).run();
db.delete(schema.accountTokens).run();
db.delete(schema.accounts).run();
db.delete(schema.sites).run();
});
afterAll(() => {
delete process.env.DATA_DIR;
});
it('removes stale exact routes and keeps wildcard routes on rebuild', () => {
const site = db.insert(schema.sites).values({
name: 'site-1',
url: 'https://site-1.example.com',
platform: 'new-api',
}).returning().get();
const account = db.insert(schema.accounts).values({
siteId: site.id,
username: 'user-1',
accessToken: 'access-1',
status: 'active',
}).returning().get();
const token = db.insert(schema.accountTokens).values({
accountId: account.id,
name: 'default',
token: 'sk-test',
source: 'manual',
enabled: true,
isDefault: true,
}).returning().get();
db.insert(schema.tokenModelAvailability).values({
tokenId: token.id,
modelName: 'latest-model',
available: true,
}).run();
const staleRoute = db.insert(schema.tokenRoutes).values({
modelPattern: 'old-model',
enabled: true,
}).returning().get();
db.insert(schema.routeChannels).values({
routeId: staleRoute.id,
accountId: account.id,
tokenId: token.id,
priority: 0,
weight: 10,
enabled: true,
manualOverride: false,
}).run();
const wildcardRoute = db.insert(schema.tokenRoutes).values({
modelPattern: 'gpt-*',
enabled: true,
}).returning().get();
db.insert(schema.routeChannels).values({
routeId: wildcardRoute.id,
accountId: account.id,
tokenId: token.id,
priority: 0,
weight: 10,
enabled: true,
manualOverride: false,
}).run();
const rebuild = rebuildTokenRoutesFromAvailability();
expect(rebuild.models).toBe(1);
expect(rebuild.removedRoutes).toBe(1);
const oldRoute = db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, staleRoute.id)).get();
expect(oldRoute).toBeUndefined();
const oldChannels = db.select().from(schema.routeChannels).where(eq(schema.routeChannels.routeId, staleRoute.id)).all();
expect(oldChannels).toHaveLength(0);
const latestRoute = db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.modelPattern, 'latest-model')).get();
expect(latestRoute).toBeDefined();
const latestChannels = db.select().from(schema.routeChannels)
.where(and(eq(schema.routeChannels.routeId, latestRoute!.id), eq(schema.routeChannels.tokenId, token.id)))
.all();
expect(latestChannels.length).toBeGreaterThan(0);
const wildcardRouteAfter = db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, wildcardRoute.id)).get();
expect(wildcardRouteAfter).toBeDefined();
});
});
+24
View File
@@ -16,6 +16,10 @@ function normalizeModels(models: string[]): string[] {
return Array.from(new Set(models.filter((model) => typeof model === 'string' && model.trim().length > 0)));
}
function isExactModelPattern(modelPattern: string): boolean {
return !/[\*\?\[]/.test(modelPattern);
}
async function withTimeout<T>(fn: () => Promise<T>, timeoutMs: number, timeoutMessage: string): Promise<T> {
let timer: ReturnType<typeof setTimeout> | null = null;
try {
@@ -214,6 +218,7 @@ export function rebuildTokenRoutesFromAvailability() {
let createdRoutes = 0;
let createdChannels = 0;
let removedChannels = 0;
let removedRoutes = 0;
for (const [modelName, tokenAccountMap] of modelTokens.entries()) {
let route = routes.find((r) => r.modelPattern === modelName);
@@ -269,11 +274,30 @@ export function rebuildTokenRoutesFromAvailability() {
}
}
const latestModelNames = new Set<string>(Array.from(modelTokens.keys()));
for (const route of routes) {
const modelPattern = (route.modelPattern || '').trim();
if (!modelPattern || !isExactModelPattern(modelPattern) || latestModelNames.has(modelPattern)) {
continue;
}
const routeChannelCount = channels.filter((channel) => channel.routeId === route.id).length;
if (routeChannelCount > 0) {
removedChannels += routeChannelCount;
}
const deleted = db.delete(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, route.id)).run().changes;
if (deleted > 0) {
removedRoutes += deleted;
}
}
return {
models: modelTokens.size,
createdRoutes,
createdChannels,
removedChannels,
removedRoutes,
};
}
+10 -4
View File
@@ -20,15 +20,21 @@ describe('proxyRetryPolicy', () => {
).toBe(true);
});
it('does not retry on generic client request errors', () => {
it('retries on generic client request errors for cross-channel fallback', () => {
expect(
shouldRetryProxyRequest(400, '{"error":{"message":"invalid request body"}}'),
).toBe(false);
).toBe(true);
expect(
shouldRetryProxyRequest(401, '{"error":{"message":"invalid access token"}}'),
).toBe(false);
).toBe(true);
expect(
shouldRetryProxyRequest(403, '{"error":{"message":"forbidden"}}'),
).toBe(false);
).toBe(true);
expect(
shouldRetryProxyRequest(404, '{"error":{"message":"not found"}}'),
).toBe(true);
expect(
shouldRetryProxyRequest(422, '{"error":{"message":"unprocessable"}}'),
).toBe(true);
});
});
+2 -2
View File
@@ -21,7 +21,7 @@ function isModelUnsupportedErrorMessage(rawMessage?: string | null): boolean {
}
export function shouldRetryProxyRequest(status: number, upstreamErrorText?: string | null): boolean {
if (status === 429 || status >= 500) return true;
if (status !== 400 && status !== 404 && status !== 422) return false;
if (status >= 400) return true;
if (!upstreamErrorText) return false;
return isModelUnsupportedErrorMessage(upstreamErrorText);
}