Merge remote-tracking branch 'origin/main' into feat/account-level-proxy

# Conflicts:
#	src/server/proxy-core/surfaces/chatSurface.ts
#	src/server/proxy-core/surfaces/openAiResponsesSurface.ts
#	src/server/routes/proxy/embeddings.ts
#	src/server/routes/proxy/images.ts
#	src/server/services/modelService.ts
This commit is contained in:
Hureru
2026-03-21 10:09:19 +08:00
198 changed files with 20039 additions and 1105 deletions
@@ -14,6 +14,7 @@ const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) =>
estimatedCostFromQuota: 0,
recoveredFromSelfLog: false,
}));
const refreshOauthAccessTokenSingleflightMock = vi.fn();
const dbInsertMock = vi.fn((_arg?: any) => ({
values: () => ({
run: () => undefined,
@@ -60,10 +61,16 @@ vi.mock('../../services/proxyUsageFallbackService.js', () => ({
resolveProxyUsageWithSelfLogFallback: (arg: any) => resolveProxyUsageWithSelfLogFallbackMock(arg),
}));
vi.mock('../../services/oauth/refreshSingleflight.js', () => ({
refreshOauthAccessTokenSingleflight: (...args: unknown[]) => refreshOauthAccessTokenSingleflightMock(...args),
}));
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
@@ -105,6 +112,7 @@ describe('chat proxy codex oauth compatibility', () => {
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
refreshOauthAccessTokenSingleflightMock.mockReset();
dbInsertMock.mockClear();
selectChannelMock.mockReturnValue({
@@ -128,6 +136,11 @@ describe('chat proxy codex oauth compatibility', () => {
actualModel: 'gpt-5.4',
});
selectNextChannelMock.mockReturnValue(null);
refreshOauthAccessTokenSingleflightMock.mockResolvedValue({
accessToken: 'fresh-access-token',
accountId: 33,
accountKey: 'chatgpt-account-123',
});
});
afterAll(async () => {
@@ -245,4 +258,107 @@ describe('chat proxy codex oauth compatibility', () => {
expect(response.body).not.toContain('event: response.created');
expect(response.body).not.toContain('"type":"response.output_text.delta"');
});
it('still translates raw codex SSE into OpenAI chat completion chunks when upstream mislabels the content-type', async () => {
fetchMock.mockResolvedValue(new Response([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_codex_openai_stream_header_miss","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_item.added\n',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_codex_openai_stream_header_miss","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_codex_openai_stream_header_miss","delta":"pong from codex"}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_codex_openai_stream_header_miss","model":"gpt-5.4","status":"completed","usage":{"input_tokens":9,"output_tokens":3,"total_tokens":12}}}\n\n',
'data: [DONE]\n\n',
].join(''), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'gpt-5.4',
stream: true,
messages: [{ role: 'user', content: 'hello codex' }],
},
});
expect(response.statusCode).toBe(200);
expect(response.headers['content-type']).toContain('text/event-stream');
expect(response.body).toContain('"object":"chat.completion.chunk"');
expect(response.body).toContain('pong from codex');
expect(response.body).not.toContain('event: response.created');
expect(response.body).not.toContain('"type":"response.output_text.delta"');
});
it('retries oauth chat requests with a normalized upstream URL after refresh', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'openai-site', url: 'https://gateway.example.com/v1/', platform: 'openai' },
account: {
id: 33,
username: 'oauth-user@example.com',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
accountId: 'chatgpt-account-123',
email: 'oauth-user@example.com',
planType: 'plus',
},
}),
},
tokenName: 'default',
tokenValue: 'expired-access-token',
actualModel: 'gpt-4o-mini',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'expired token', type: 'invalid_request_error' },
}), {
status: 401,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'chatcmpl_refreshed',
object: 'chat.completion',
created: 1706000000,
model: 'gpt-4o-mini',
choices: [
{
index: 0,
message: { role: 'assistant', content: 'ok after refresh' },
finish_reason: 'stop',
},
],
usage: { prompt_tokens: 4, completion_tokens: 2, total_tokens: 6 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/chat/completions',
payload: {
model: 'gpt-4o-mini',
messages: [{ role: 'user', content: 'hello oauth' }],
},
});
expect(response.statusCode).toBe(200);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
expect(fetchMock).toHaveBeenCalledTimes(2);
const [firstUrl, firstOptions] = fetchMock.mock.calls[0] as [string, any];
const [secondUrl, secondOptions] = fetchMock.mock.calls[1] as [string, any];
expect(firstUrl).toBe('https://gateway.example.com/v1/responses');
expect(secondUrl).toBe('https://gateway.example.com/v1/responses');
expect(firstOptions.headers.Authorization).toBe('Bearer expired-access-token');
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
expect(response.json()?.choices?.[0]?.message?.content).toBe('ok after refresh');
});
});
@@ -68,6 +68,8 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
+223 -7
View File
@@ -1,6 +1,7 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { config } from '../../config.js';
import { resetUpstreamEndpointRuntimeState } from './upstreamEndpoint.js';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
@@ -68,6 +69,9 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
},
@@ -99,6 +103,7 @@ describe('chat proxy stream behavior', () => {
fetchModelPricingCatalogMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
dbInsertMock.mockClear();
resetUpstreamEndpointRuntimeState();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
@@ -110,6 +115,17 @@ describe('chat proxy stream behavior', () => {
});
selectNextChannelMock.mockReturnValue(null);
fetchModelPricingCatalogMock.mockResolvedValue(null);
(config as any).codexHeaderDefaults = {
userAgent: '',
betaFeatures: '',
};
(config as any).payloadRules = {
default: [],
defaultRaw: [],
override: [],
overrideRaw: [],
filter: [],
};
config.proxyEmptyContentFailEnabled = false;
config.proxyErrorKeywords = [];
});
@@ -2016,6 +2032,87 @@ describe('chat proxy stream behavior', () => {
expect(targetUrl).toContain('/v1/messages');
});
it('does not stick generic /v1/responses traffic to /v1/messages after a fallback success', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'generic-site', url: 'https://upstream.example.com', platform: 'new-api' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'upstream-gpt',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'Gateway time-out', type: 'upstream_error' },
}), {
status: 504,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'Bad gateway', type: 'upstream_error' },
}), {
status: 502,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'msg_fallback_1',
type: 'message',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'ok via messages fallback' }],
stop_reason: 'end_turn',
usage: { input_tokens: 7, output_tokens: 3, total_tokens: 10 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'resp_recovered_1',
object: 'response',
model: 'upstream-gpt',
status: 'completed',
output_text: 'ok via recovered responses',
usage: { input_tokens: 6, output_tokens: 2, total_tokens: 8 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const firstResponse = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.4',
input: 'hello',
},
});
expect(firstResponse.statusCode).toBe(200);
expect(firstResponse.json().output_text).toContain('ok via messages fallback');
const secondResponse = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.4',
input: 'hello again',
},
});
expect(secondResponse.statusCode).toBe(200);
expect(secondResponse.json().output_text).toContain('ok via recovered responses');
expect(fetchMock).toHaveBeenCalledTimes(4);
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];
const [fourthUrl] = fetchMock.mock.calls[3] as [string, any];
expect(firstUrl).toContain('/v1/responses');
expect(secondUrl).toContain('/v1/chat/completions');
expect(thirdUrl).toContain('/v1/messages');
expect(fourthUrl).toContain('/v1/responses');
});
it('prefers native /v1/responses for claude-family /v1/responses requests that explicitly ask for encrypted reasoning', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
@@ -2193,7 +2290,7 @@ describe('chat proxy stream behavior', () => {
});
});
it('returns clear 400 when input_file file_url is sent to a claude-only upstream without native responses support', async () => {
it('converts input_file file_url into Claude document url blocks for claude-only upstreams', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'claude-site', url: 'https://upstream.example.com', platform: 'claude' },
@@ -2202,6 +2299,18 @@ describe('chat proxy stream behavior', () => {
tokenValue: 'sk-demo',
actualModel: 'upstream-claude',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'msg_file_url_1',
type: 'message',
role: 'assistant',
model: 'upstream-claude',
content: [{ type: 'text', text: 'hello from claude messages' }],
stop_reason: 'end_turn',
usage: { input_tokens: 7, output_tokens: 3 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
@@ -2225,12 +2334,119 @@ describe('chat proxy stream behavior', () => {
},
});
expect(response.statusCode).toBe(400);
const body = response.json();
expect(body?.error?.type).toBe('invalid_request_error');
expect(body?.error?.message).toContain('input_file.file_url');
expect(body?.error?.message).toContain('/v1/responses');
expect(fetchMock).not.toHaveBeenCalled();
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(1);
const [targetUrl, options] = fetchMock.mock.calls[0] as [string, any];
expect(targetUrl).toContain('/v1/messages');
const forwardedBody = JSON.parse(options.body);
expect(forwardedBody.messages[0].content[1]).toMatchObject({
type: 'document',
title: 'remote.pdf',
source: {
type: 'url',
url: 'https://example.com/remote.pdf',
},
});
});
it('does not let remote document url success poison later inline document endpoint preference', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'generic-site', url: 'https://upstream.example.com', platform: 'new-api' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'upstream-gpt',
});
fetchMock.mockImplementation(async (target: unknown) => {
const url = String(target);
if (url.includes('/v1/responses')) {
return new Response(JSON.stringify({
id: 'resp_file_url_runtime_1',
object: 'response',
model: 'upstream-gpt',
output_text: 'hello from responses upstream',
output: [
{
id: 'msg_file_url_runtime_1',
type: 'message',
role: 'assistant',
content: [{ type: 'output_text', text: 'hello from responses upstream' }],
},
],
status: 'completed',
usage: { input_tokens: 7, output_tokens: 3, total_tokens: 10 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}
if (url.includes('/v1/messages')) {
return new Response(JSON.stringify({
id: 'msg_inline_runtime_1',
type: 'message',
role: 'assistant',
model: 'upstream-gpt',
content: [{ type: 'text', text: 'hello from messages upstream' }],
stop_reason: 'end_turn',
usage: { input_tokens: 7, output_tokens: 3 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}
throw new Error(`unexpected target url: ${url}`);
});
const remoteResponse = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'claude-haiku-4-5-20251001',
input: [
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'read this remote file' },
{
type: 'input_file',
filename: 'remote.pdf',
file_url: 'https://example.com/remote.pdf',
},
],
},
],
},
});
expect(remoteResponse.statusCode).toBe(200);
expect(String(fetchMock.mock.calls[0]?.[0])).toContain('/v1/responses');
const inlineResponse = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'claude-haiku-4-5-20251001',
input: [
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'read this inline file' },
{
type: 'input_file',
filename: 'brief.pdf',
file_data: 'data:application/pdf;base64,JVBERi0xLjQK',
},
],
},
],
},
});
expect(inlineResponse.statusCode).toBe(200);
expect(String(fetchMock.mock.calls[1]?.[0])).toContain('/v1/messages');
});
it('prefers native /v1/responses for claude-family /v1/responses requests that opt into reasoning without injecting a generic default include', async () => {
+159 -66
View File
@@ -1,6 +1,5 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
@@ -15,8 +14,10 @@ import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { detectProxyFailure } from './proxyFailureJudge.js';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
const MAX_RETRIES = 2;
@@ -26,12 +27,16 @@ export async function completionsProxyRoute(app: FastifyInstance) {
const requestedModel = body?.model;
if (!requestedModel) {
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
}
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
}
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const downstreamPath = '/v1/completions';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const isStream = body.stream === true;
const excludeChannelIds: number[] = [];
@@ -75,8 +80,28 @@ export async function completionsProxyRoute(app: FastifyInstance) {
if (!upstream.ok) {
const errText = await upstream.text().catch(() => 'unknown error');
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: errText,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
errText,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
await reportTokenExpired({
@@ -174,12 +199,25 @@ export async function completionsProxyRoute(app: FastifyInstance) {
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected,
requestedModel,
'success',
200,
latency,
null,
retryCount,
downstreamApiKeyId,
resolvedUsage.promptTokens,
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
clientContext,
downstreamPath,
);
return;
}
@@ -195,17 +233,28 @@ export async function completionsProxyRoute(app: FastifyInstance) {
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
if (failure) {
const errText = failure.reason;
tokenRouter.recordFailure(selected.channel.id);
tokenRouter.recordFailure(selected.channel.id, {
status: failure.status,
errorText: errText,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
failure.status,
latency,
errText,
retryCount,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
failure.status,
latency,
errText,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (shouldRetryProxyRequest(failure.status, errText) && retryCount < MAX_RETRIES) {
retryCount += 1;
@@ -245,16 +294,49 @@ export async function completionsProxyRoute(app: FastifyInstance) {
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
);
return reply.send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected,
requestedModel,
'success',
200,
latency,
null,
retryCount,
downstreamApiKeyId,
resolvedUsage.promptTokens,
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
clientContext,
downstreamPath,
);
return reply.send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
@@ -271,47 +353,58 @@ export async function completionsProxyRoute(app: FastifyInstance) {
});
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath: '/v1/completions',
errorMessage,
});
await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
modelRequested,
modelActual: selected.actualModel,
status,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/completions',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
}).run();
} catch (error) {
completionTokens,
totalTokens,
estimatedCost,
billingDetails,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
});
} catch (error) {
console.warn('[proxy/completions] failed to write proxy log', error);
}
}
@@ -71,6 +71,9 @@ vi.mock('../../db/index.js', () => ({
schema: {
proxyLogs: {},
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
}));
describe('downstream client context route logging', () => {
@@ -33,6 +33,16 @@ describe('isCodexResponsesSurface', () => {
})).toBe(true);
});
it('detects broader Codex official-client family headers from user-agent and originator prefixes', () => {
expect(isCodexResponsesSurface({
'user-agent': 'Mozilla/5.0 codex_chatgpt_desktop/1.2.3',
})).toBe(true);
expect(isCodexResponsesSurface({
originator: 'codex_vscode',
})).toBe(true);
});
it('returns false for generic responses clients', () => {
expect(isCodexResponsesSurface({
'content-type': 'application/json',
@@ -50,6 +60,9 @@ describe('detectDownstreamClientContext', () => {
},
})).toEqual({
clientKind: 'codex',
clientAppId: 'codex_cli_rs',
clientAppName: 'Codex CLI',
clientConfidence: 'exact',
sessionId: 'codex-session-123',
traceHint: 'codex-session-123',
});
@@ -63,6 +76,37 @@ describe('detectDownstreamClientContext', () => {
},
})).toEqual({
clientKind: 'codex',
clientAppId: 'codex',
clientAppName: 'Codex',
clientConfidence: 'heuristic',
});
});
it('recognizes broader Codex official-client user-agent families without requiring stainless headers', () => {
expect(detectDownstreamClientContext({
downstreamPath: '/v1/responses',
headers: {
'user-agent': 'Mozilla/5.0 codex_chatgpt_desktop/1.2.3',
},
})).toEqual({
clientKind: 'codex',
clientAppId: 'codex_chatgpt_desktop',
clientAppName: 'Codex Desktop',
clientConfidence: 'exact',
});
});
it('recognizes broader Codex official-client originator prefixes', () => {
expect(detectDownstreamClientContext({
downstreamPath: '/v1/responses',
headers: {
originator: 'codex_exec',
},
})).toEqual({
clientKind: 'codex',
clientAppId: 'codex_exec',
clientAppName: 'Codex Exec',
clientConfidence: 'exact',
});
});
@@ -79,6 +123,9 @@ describe('detectDownstreamClientContext', () => {
body,
})).toEqual({
clientKind: 'claude_code',
clientAppId: 'claude_code',
clientAppName: 'Claude Code',
clientConfidence: 'exact',
sessionId: 'f25958b8-e75c-455d-8b40-f006d87cc2a4',
traceHint: 'f25958b8-e75c-455d-8b40-f006d87cc2a4',
});
@@ -128,6 +175,54 @@ describe('detectDownstreamClientContext', () => {
},
})).toEqual({
clientKind: 'gemini_cli',
clientAppId: 'gemini_cli',
clientAppName: 'Gemini CLI',
clientConfidence: 'exact',
});
});
it('recognizes app fingerprints alongside a generic protocol family', () => {
expect(detectDownstreamClientContext({
downstreamPath: '/v1/chat/completions',
headers: {
'x-title': 'Cherry Studio',
'http-referer': 'https://cherry-ai.com',
},
})).toEqual({
clientKind: 'generic',
clientAppId: 'cherry_studio',
clientAppName: 'Cherry Studio',
clientConfidence: 'exact',
});
});
it('keeps protocol family detection when an app fingerprint also matches', () => {
expect(detectDownstreamClientContext({
downstreamPath: '/v1/responses',
headers: {
originator: 'codex_cli_rs',
'x-title': 'Cherry Studio',
'http-referer': 'https://cherry-ai.com',
},
})).toEqual({
clientKind: 'codex',
clientAppId: 'cherry_studio',
clientAppName: 'Cherry Studio',
clientConfidence: 'exact',
});
});
it('marks weak app-only matches as heuristic instead of upgrading protocol behavior', () => {
expect(detectDownstreamClientContext({
downstreamPath: '/v1/chat/completions',
headers: {
'user-agent': 'CherryStudio/1.2.3',
},
})).toEqual({
clientKind: 'generic',
clientAppId: 'cherry_studio',
clientAppName: 'Cherry Studio',
clientConfidence: 'heuristic',
});
});
});
@@ -6,13 +6,269 @@ import { isCodexResponsesSurface as isCodexResponsesSurfaceViaProfile } from '..
import type { CliProfileId } from '../../proxy-core/cliProfiles/types.js';
export type DownstreamClientKind = CliProfileId;
export type DownstreamClientConfidence = 'exact' | 'heuristic';
export type DownstreamClientContext = {
clientKind: DownstreamClientKind;
sessionId?: string;
traceHint?: string;
clientAppId?: string;
clientAppName?: string;
clientConfidence?: DownstreamClientConfidence;
};
type NormalizedClientHeaders = Record<string, string[]>;
type DownstreamClientBodySummary = {
topLevelKeys: string[];
metadataUserId: string | null;
};
type DownstreamClientFingerprintInput = {
downstreamPath: string;
headers: NormalizedClientHeaders;
bodySummary: DownstreamClientBodySummary;
};
type DownstreamClientFingerprintRule = {
id: string;
name: string;
priority: number;
match(input: DownstreamClientFingerprintInput): DownstreamClientConfidence | null;
};
type DownstreamProtocolClientApp = {
clientAppId: string;
clientAppName: string;
clientConfidence: DownstreamClientConfidence;
};
type HeaderPrefixMatcherRule = {
id: string;
name: string;
userAgentPrefixes?: string[];
originatorPrefixes?: string[];
};
const codexOfficialClientAppRules: HeaderPrefixMatcherRule[] = [
{
id: 'codex_cli_rs',
name: 'Codex CLI',
userAgentPrefixes: ['codex_cli_rs/'],
originatorPrefixes: ['codex_cli_rs'],
},
{
id: 'codex_vscode',
name: 'Codex VSCode',
userAgentPrefixes: ['codex_vscode/'],
originatorPrefixes: ['codex_vscode'],
},
{
id: 'codex_app',
name: 'Codex App',
userAgentPrefixes: ['codex_app/'],
originatorPrefixes: ['codex_app'],
},
{
id: 'codex_chatgpt_desktop',
name: 'Codex Desktop',
userAgentPrefixes: ['codex_chatgpt_desktop/', 'codex desktop/'],
originatorPrefixes: ['codex_chatgpt_desktop', 'codex desktop'],
},
{
id: 'codex_atlas',
name: 'Codex Atlas',
userAgentPrefixes: ['codex_atlas/'],
originatorPrefixes: ['codex_atlas'],
},
{
id: 'codex_exec',
name: 'Codex Exec',
userAgentPrefixes: ['codex_exec/'],
originatorPrefixes: ['codex_exec'],
},
{
id: 'codex_sdk_ts',
name: 'Codex SDK TS',
userAgentPrefixes: ['codex_sdk_ts/'],
originatorPrefixes: ['codex_sdk_ts'],
},
];
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object' && !Array.isArray(value);
}
function normalizeHeaderValues(value: unknown): string[] {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed ? [trimmed] : [];
}
if (!Array.isArray(value)) return [];
return value
.filter((item): item is string => typeof item === 'string')
.map((item) => item.trim())
.filter((item) => item.length > 0);
}
function normalizeHeaders(headers?: Record<string, unknown>): NormalizedClientHeaders {
if (!headers) return {};
const normalized: NormalizedClientHeaders = {};
for (const [rawKey, rawValue] of Object.entries(headers)) {
const key = rawKey.trim().toLowerCase();
if (!key) continue;
const values = normalizeHeaderValues(rawValue);
if (values.length === 0) continue;
normalized[key] = normalized[key]
? [...normalized[key], ...values]
: values;
}
return normalized;
}
function headerEquals(headers: NormalizedClientHeaders, key: string, expected: string): boolean {
const normalizedExpected = expected.trim().toLowerCase();
return (headers[key.trim().toLowerCase()] || []).some((value) => value.trim().toLowerCase() === normalizedExpected);
}
function headerIncludes(headers: NormalizedClientHeaders, key: string, expectedFragment: string): boolean {
const normalizedExpected = expectedFragment.trim().toLowerCase();
return (headers[key.trim().toLowerCase()] || []).some((value) => value.trim().toLowerCase().includes(normalizedExpected));
}
function headerMatchesPrefixes(headers: NormalizedClientHeaders, key: string, prefixes: string[]): boolean {
return prefixes.some((prefix) => headerIncludes(headers, key, prefix));
}
function buildBodySummary(body: unknown): DownstreamClientBodySummary {
if (!isRecord(body)) {
return {
topLevelKeys: [],
metadataUserId: null,
};
}
const metadataUserId = isRecord(body.metadata) && typeof body.metadata.user_id === 'string'
? body.metadata.user_id.trim() || null
: null;
return {
topLevelKeys: Object.keys(body).sort((left, right) => left.localeCompare(right)),
metadataUserId,
};
}
const appFingerprintRegistry: DownstreamClientFingerprintRule[] = [
{
id: 'cherry_studio',
name: 'Cherry Studio',
priority: 100,
match(input) {
const hasTitle = headerEquals(input.headers, 'x-title', 'Cherry Studio');
const hasReferer = headerEquals(input.headers, 'http-referer', 'https://cherry-ai.com')
|| headerEquals(input.headers, 'referer', 'https://cherry-ai.com');
if (hasTitle && hasReferer) {
return 'exact';
}
const weakSignals = [
headerIncludes(input.headers, 'user-agent', 'cherrystudio'),
headerIncludes(input.headers, 'x-title', 'cherry studio'),
headerIncludes(input.headers, 'http-referer', 'cherry-ai.com'),
headerIncludes(input.headers, 'referer', 'cherry-ai.com'),
];
return weakSignals.some(Boolean) ? 'heuristic' : null;
},
},
];
function detectDownstreamClientFingerprint(input: {
downstreamPath: string;
headers?: Record<string, unknown>;
body?: unknown;
}) {
const fingerprintInput: DownstreamClientFingerprintInput = {
downstreamPath: input.downstreamPath,
headers: normalizeHeaders(input.headers),
bodySummary: buildBodySummary(input.body),
};
let matchedRule: DownstreamClientFingerprintRule | null = null;
let matchedConfidence: DownstreamClientConfidence | null = null;
for (const rule of appFingerprintRegistry) {
const confidence = rule.match(fingerprintInput);
if (!confidence) continue;
if (!matchedRule || rule.priority > matchedRule.priority) {
matchedRule = rule;
matchedConfidence = confidence;
}
}
if (!matchedRule || !matchedConfidence) {
return null;
}
return {
clientAppId: matchedRule.id,
clientAppName: matchedRule.name,
clientConfidence: matchedConfidence,
};
}
function detectCodexOfficialClientApp(headers: NormalizedClientHeaders): DownstreamProtocolClientApp | null {
for (const rule of codexOfficialClientAppRules) {
const matchesOriginator = Array.isArray(rule.originatorPrefixes)
&& headerMatchesPrefixes(headers, 'originator', rule.originatorPrefixes);
const matchesUserAgent = Array.isArray(rule.userAgentPrefixes)
&& headerMatchesPrefixes(headers, 'user-agent', rule.userAgentPrefixes);
if (!matchesOriginator && !matchesUserAgent) continue;
return {
clientAppId: rule.id,
clientAppName: rule.name,
clientConfidence: 'exact',
};
}
return null;
}
function detectProtocolClientApp(input: {
clientKind: DownstreamClientKind;
headers?: Record<string, unknown>;
}): DownstreamProtocolClientApp | null {
switch (input.clientKind) {
case 'claude_code':
return {
clientAppId: 'claude_code',
clientAppName: 'Claude Code',
clientConfidence: 'exact',
};
case 'gemini_cli':
return {
clientAppId: 'gemini_cli',
clientAppName: 'Gemini CLI',
clientConfidence: 'exact',
};
case 'codex': {
const headers = normalizeHeaders(input.headers);
return detectCodexOfficialClientApp(headers) || {
clientAppId: 'codex',
clientAppName: 'Codex',
clientConfidence: 'heuristic',
};
}
default:
return null;
}
}
export function isCodexResponsesSurface(headers?: Record<string, unknown>): boolean {
return isCodexResponsesSurfaceViaProfile(headers);
}
@@ -27,9 +283,15 @@ export function detectDownstreamClientContext(input: {
body?: unknown;
}): DownstreamClientContext {
const detected = detectCliProfile(input);
const fingerprint = detectDownstreamClientFingerprint(input);
const protocolClientApp = fingerprint ? null : detectProtocolClientApp({
clientKind: detected.id,
headers: input.headers,
});
return {
clientKind: detected.id,
...(detected.sessionId ? { sessionId: detected.sessionId } : {}),
...(detected.traceHint ? { traceHint: detected.traceHint } : {}),
...(fingerprint || protocolClientApp || {}),
};
}
+3 -1
View File
@@ -4,7 +4,9 @@ import { isModelAllowedByPolicyOrAllowedRoutes, recordManagedKeyCostUsage } from
import { EMPTY_DOWNSTREAM_ROUTING_POLICY, type DownstreamRoutingPolicy } from '../../services/downstreamPolicyTypes.js';
export function getDownstreamRoutingPolicy(request: FastifyRequest): DownstreamRoutingPolicy {
return getProxyAuthContext(request)?.policy || EMPTY_DOWNSTREAM_ROUTING_POLICY;
const authContext = getProxyAuthContext(request);
if (!authContext) return EMPTY_DOWNSTREAM_ROUTING_POLICY;
return authContext.policy;
}
export async function ensureModelAllowedForDownstreamKey(
+73 -16
View File
@@ -1,6 +1,5 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
@@ -16,6 +15,8 @@ import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
const MAX_RETRIES = 2;
@@ -29,8 +30,12 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
const downstreamPath = '/v1/embeddings';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -71,8 +76,28 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
@@ -97,8 +122,8 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
let data: any = {};
try { data = JSON.parse(text); } catch { data = {}; }
const parsedUsage = parseProxyUsage(data);
const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(data);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
@@ -121,16 +146,37 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, clientContext, downstreamPath,
);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
0,
0,
0,
null,
clientContext,
downstreamPath,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
@@ -159,18 +205,25 @@ async function logProxy(
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/embeddings',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath: '/v1/embeddings',
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await db.insert(schema.proxyLogs).values({
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
@@ -180,11 +233,15 @@ async function logProxy(
completionTokens,
totalTokens,
estimatedCost,
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
billingDetails,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
}).run();
});
} catch (error) {
console.warn('[proxy/embeddings] failed to write proxy log', error);
}
+35 -16
View File
@@ -73,6 +73,7 @@ describe('executeEndpointFlow', () => {
expect(result.ok).toBe(true);
expect(dispatchRequest).toHaveBeenCalledTimes(1);
expect(dispatchRequest.mock.calls[0]?.[1]).toBe('https://example.com/v1/responses');
expect(fetchMock).not.toHaveBeenCalled();
});
@@ -155,6 +156,40 @@ describe('executeEndpointFlow', () => {
expect(fetchMock).toHaveBeenCalledTimes(2);
});
it('emits attempt callbacks for failed and successful endpoint probes', async () => {
fetchMock
.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({
error: { message: 'unsupported endpoint', type: 'invalid_request_error' },
}), {
status: 404,
headers: { 'content-type': 'application/json' },
})))
.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { 'content-type': 'application/json' },
})));
const onAttemptFailure = vi.fn();
const onAttemptSuccess = vi.fn();
const result = await executeEndpointFlow({
siteUrl: 'https://example.com',
endpointCandidates: ['responses', 'chat'],
buildRequest: (endpoint) => endpoint === 'responses'
? requestFor('/v1/responses')
: { ...requestFor('/v1/chat/completions'), endpoint },
shouldDowngrade: () => true,
onAttemptFailure,
onAttemptSuccess,
});
expect(result.ok).toBe(true);
expect(onAttemptFailure).toHaveBeenCalledTimes(1);
expect(onAttemptFailure.mock.calls[0]?.[0]?.request?.path).toBe('/v1/responses');
expect(onAttemptSuccess).toHaveBeenCalledTimes(1);
expect(onAttemptSuccess.mock.calls[0]?.[0]?.request?.path).toBe('/v1/chat/completions');
});
it('accepts recovered response from tryRecover hook', async () => {
fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({
error: { message: 'upstream_error', type: 'upstream_error' },
@@ -185,22 +220,6 @@ describe('executeEndpointFlow', () => {
expect(fetchMock).toHaveBeenCalledTimes(1);
});
it('uses proxyUrl for the default fetch path when no dispatch hook is provided', async () => {
fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({ ok: true }), {
status: 200,
headers: { 'content-type': 'application/json' },
})));
await executeEndpointFlow({
siteUrl: 'https://example.com',
proxyUrl: 'https://proxy.internal/base',
endpointCandidates: ['responses'],
buildRequest: () => requestFor('/v1/responses'),
});
expect(fetchMock.mock.calls[0]?.[0]).toBe('https://proxy.internal/base/v1/responses');
});
it('returns normalized final error when all endpoints fail', async () => {
fetchMock.mockResolvedValueOnce(toUndiciResponse(new Response(JSON.stringify({
error: { message: 'upstream_error', type: 'upstream_error' },
+29 -11
View File
@@ -27,6 +27,14 @@ export type EndpointAttemptContext = {
rawErrText: string;
};
export type EndpointAttemptSuccessContext = {
endpointIndex: number;
endpointCount: number;
request: BuiltEndpointRequest;
targetUrl: string;
response: Awaited<ReturnType<typeof fetch>>;
};
export type EndpointRecoverResult = {
upstream: Awaited<ReturnType<typeof fetch>>;
upstreamPath: string;
@@ -51,7 +59,6 @@ export function withUpstreamPath(path: string, message: string): string {
type ExecuteEndpointFlowInput = {
siteUrl: string;
proxyUrl?: string | null;
endpointCandidates: UpstreamEndpoint[];
buildRequest: (endpoint: UpstreamEndpoint, endpointIndex: number) => BuiltEndpointRequest;
dispatchRequest?: (
@@ -61,14 +68,10 @@ type ExecuteEndpointFlowInput = {
tryRecover?: (ctx: EndpointAttemptContext) => Promise<EndpointRecoverResult>;
shouldDowngrade?: (ctx: EndpointAttemptContext) => boolean;
onDowngrade?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise<void>;
onAttemptFailure?: (ctx: EndpointAttemptContext & { errText: string }) => void | Promise<void>;
onAttemptSuccess?: (ctx: EndpointAttemptSuccessContext) => void | Promise<void>;
};
function buildAbsoluteUrl(base: string, path: string): string {
const normalizedBase = base.replace(/\/+$/, '');
const normalizedPath = path.replace(/^\/+/, '');
return normalizedPath ? `${normalizedBase}/${normalizedPath}` : normalizedBase;
}
export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Promise<EndpointFlowResult> {
const endpointCount = input.endpointCandidates.length;
if (endpointCount <= 0) {
@@ -86,10 +89,7 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
for (let endpointIndex = 0; endpointIndex < endpointCount; endpointIndex += 1) {
const endpoint = input.endpointCandidates[endpointIndex] as UpstreamEndpoint;
const request = input.buildRequest(endpoint, endpointIndex);
const defaultTarget = buildUpstreamUrl(input.siteUrl, request.path);
const targetUrl = input.proxyUrl
? buildAbsoluteUrl(input.proxyUrl, request.path)
: defaultTarget;
const targetUrl = buildUpstreamUrl(input.siteUrl, request.path);
let response = input.dispatchRequest
? await input.dispatchRequest(request, targetUrl)
@@ -100,6 +100,13 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
}));
if (response.ok) {
await input.onAttemptSuccess?.({
endpointIndex,
endpointCount,
request,
targetUrl,
response,
});
return {
ok: true,
upstream: response,
@@ -120,6 +127,13 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
if (input.tryRecover) {
const recovered = await input.tryRecover(baseContext);
if (recovered?.upstream?.ok) {
await input.onAttemptSuccess?.({
endpointIndex,
endpointCount,
request: baseContext.request,
targetUrl: baseContext.targetUrl,
response: recovered.upstream,
});
return {
ok: true,
upstream: recovered.upstream,
@@ -135,6 +149,10 @@ export async function executeEndpointFlow(input: ExecuteEndpointFlowInput): Prom
baseContext.request.path,
summarizeUpstreamError(response.status, rawErrText),
);
await input.onAttemptFailure?.({
...baseContext,
errText,
});
const isLastEndpoint = endpointIndex >= endpointCount - 1;
const shouldDowngrade = !isLastEndpoint && !!input.shouldDowngrade?.(baseContext);
+114 -6
View File
@@ -2,6 +2,7 @@ import Fastify, { type FastifyInstance } from 'fastify';
import { readFileSync } from 'node:fs';
import path from 'node:path';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { resetUpstreamEndpointRuntimeState } from './upstreamEndpoint.js';
const fetchMock = vi.fn();
const fetchModelPricingCatalogMock = vi.fn();
@@ -74,6 +75,9 @@ vi.mock('../../db/index.js', () => ({
select: (..._args: unknown[]) => createDbSelectChain(),
insert: (arg: unknown) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
modelAvailability: {
@@ -185,6 +189,7 @@ describe('gemini native proxy routes', () => {
selectNextChannelMock.mockReturnValue(null);
recordSuccessMock.mockResolvedValue(undefined);
recordFailureMock.mockResolvedValue(undefined);
resetUpstreamEndpointRuntimeState();
explainSelectionMock.mockResolvedValue({ selectedChannelId: 11 });
isModelAllowedByPolicyOrAllowedRoutesMock.mockResolvedValue(true);
});
@@ -258,7 +263,10 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
status: 500,
errorText: JSON.stringify({ error: { message: 'first channel failed' } }),
}));
const [firstUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
const [secondUrl] = fetchMock.mock.calls[1] as [string, RequestInit];
expect(firstUrl).toContain('key=gemini-key');
@@ -681,6 +689,92 @@ describe('gemini native proxy routes', () => {
});
});
it('routes Gemini native document requests to responses endpoints on openai-compatible upstreams', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 42, routeId: 22 },
site: { id: 78, name: 'openai-site', url: 'https://api.openai.com', platform: 'openai' },
account: { id: 38, username: 'openai-user@example.com' },
tokenName: 'default',
tokenValue: 'openai-access-token',
actualModel: 'gpt-4.1',
});
fetchMock.mockImplementation(async (target: unknown) => {
const url = String(target);
if (url === 'https://api.openai.com/v1/responses') {
return new Response(JSON.stringify({
id: 'resp-openai-file-1',
object: 'response',
model: 'gpt-4.1',
output: [
{
type: 'message',
role: 'assistant',
content: [{ type: 'output_text', text: 'document summary from responses' }],
},
],
usage: {
input_tokens: 9,
output_tokens: 4,
total_tokens: 13,
},
}), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}
if (url === 'https://api.openai.com/v1/chat/completions') {
return new Response(JSON.stringify({
id: 'chatcmpl-openai-file-1',
object: 'chat.completion',
created: 1_742_160_002,
model: 'gpt-4.1',
choices: [
{
index: 0,
message: {
role: 'assistant',
content: 'document summary from chat',
},
finish_reason: 'stop',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
});
}
throw new Error(`unexpected target url: ${url}`);
});
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
authorization: 'Bearer sk-managed-gemini',
},
payload: {
contents: [
{
role: 'user',
parts: [
{ text: 'summarize this pdf' },
{
fileData: {
fileUri: 'https://example.com/brief.pdf',
mimeType: 'application/pdf',
},
},
],
},
],
},
});
expect(response.statusCode).toBe(200);
const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://api.openai.com/v1/responses');
});
it('routes Gemini native generateContent requests to antigravity upstreams through the internal content endpoint', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 43, routeId: 22 },
@@ -1414,7 +1508,10 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
status: 400,
errorText: JSON.stringify({ error: { message: 'bad request on first channel' } }),
}));
const [firstUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
const [secondUrl] = fetchMock.mock.calls[1] as [string, RequestInit];
expect(firstUrl).toContain('key=gemini-key');
@@ -1464,7 +1561,10 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
status: 403,
errorText: JSON.stringify({ error: { message: 'forbidden on first channel' } }),
}));
});
it('falls back to the next channel when first Gemini channel returns 500 before any bytes are written', async () => {
@@ -1507,7 +1607,10 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
status: 500,
errorText: 'upstream crash',
}));
});
it('falls back to the next channel when first Gemini channel throws before any bytes are written', async () => {
@@ -1547,7 +1650,9 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
errorText: 'socket hang up',
}));
});
it('falls back to the next channel for SSE requests before any bytes are written', async () => {
@@ -1593,7 +1698,10 @@ describe('gemini native proxy routes', () => {
expect(response.statusCode).toBe(200);
expect(fetchMock).toHaveBeenCalledTimes(2);
expect(recordFailureMock).toHaveBeenCalledWith(11);
expect(recordFailureMock).toHaveBeenCalledWith(11, expect.objectContaining({
status: 500,
errorText: JSON.stringify({ error: { message: 'upstream unavailable' } }),
}));
expect(response.body).toContain('hello from fallback sse');
});
@@ -58,6 +58,9 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
},
+103 -20
View File
@@ -1,6 +1,5 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
@@ -15,6 +14,8 @@ import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
const MAX_RETRIES = 2;
@@ -27,8 +28,12 @@ export async function imagesProxyRoute(app: FastifyInstance) {
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
const downstreamPath = '/v1/images/generations';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -70,8 +75,24 @@ export async function imagesProxyRoute(app: FastifyInstance) {
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
@@ -103,13 +124,29 @@ export async function imagesProxyRoute(app: FastifyInstance) {
completionTokens: 0,
totalTokens: 0,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
@@ -137,8 +174,12 @@ export async function imagesProxyRoute(app: FastifyInstance) {
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
const downstreamPath = '/v1/images/edits';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body: jsonBody || Object.fromEntries(multipartForm?.entries?.() || []),
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -192,8 +233,24 @@ export async function imagesProxyRoute(app: FastifyInstance) {
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null, 0, '/v1/images/edits');
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
@@ -225,13 +282,29 @@ export async function imagesProxyRoute(app: FastifyInstance) {
completionTokens: 0,
totalTokens: 0,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null, estimatedCost, '/v1/images/edits');
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, estimatedCost, downstreamPath, clientContext);
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null, 0, '/v1/images/edits');
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: err.message,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamApiKeyId,
0,
downstreamPath,
clientContext,
);
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
@@ -268,18 +341,24 @@ async function logProxy(
downstreamApiKeyId: number | null = null,
estimatedCost = 0,
downstreamPath = '/v1/images/generations',
clientContext: DownstreamClientContext | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
});
await db.insert(schema.proxyLogs).values({
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
@@ -289,10 +368,14 @@ async function logProxy(
completionTokens: 0,
totalTokens: 0,
estimatedCost,
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
}).run();
});
} catch (error) {
console.warn('[proxy/images] failed to write proxy log', error);
}
+3 -3
View File
@@ -7,7 +7,7 @@ type ComposeProxyLogMessageArgs = {
errorMessage?: string | null;
};
type ParsedPathMeta = {
export type ParsedProxyLogMessageMeta = {
clientKind: string | null;
sessionId: string | null;
downstreamPath: string | null;
@@ -15,7 +15,7 @@ type ParsedPathMeta = {
messageText: string;
};
function parseExistingPathMeta(rawMessage: string): ParsedPathMeta {
export function parseProxyLogMessageMeta(rawMessage: string): ParsedProxyLogMessageMeta {
const clientMatch = rawMessage.match(/\[client:([^\]]+)\]/i);
const sessionMatch = rawMessage.match(/\[session:([^\]]+)\]/i);
const downstreamMatch = rawMessage.match(/\[downstream:([^\]]+)\]/i);
@@ -42,7 +42,7 @@ export function composeProxyLogMessage({
errorMessage,
}: ComposeProxyLogMessageArgs): string | null {
const rawMessage = typeof errorMessage === 'string' ? errorMessage.trim() : '';
const parsed = parseExistingPathMeta(rawMessage);
const parsed = parseProxyLogMessageMeta(rawMessage);
const finalClientKind = (clientKind || parsed.clientKind || '').trim();
const finalSessionId = (sessionId || traceHint || parsed.sessionId || '').trim();
const finalDownstreamPath = (downstreamPath || parsed.downstreamPath || '').trim();
+117
View File
@@ -8,6 +8,7 @@ type DbModule = typeof import('../../db/index.js');
type ProxyRouterModule = typeof import('./router.js');
type TokenRouterModule = typeof import('../../services/tokenRouter.js');
type TokensRoutesModule = typeof import('../api/tokens.js');
type ConfigModule = typeof import('../../config.js');
describe('/v1/models route', () => {
let db: DbModule['db'];
@@ -15,6 +16,7 @@ describe('/v1/models route', () => {
let proxyRoutes: ProxyRouterModule['proxyRoutes'];
let tokensRoutes: TokensRoutesModule['tokensRoutes'];
let invalidateTokenRouterCache: TokenRouterModule['invalidateTokenRouterCache'];
let config: ConfigModule['config'];
let app: FastifyInstance;
let dataDir = '';
@@ -27,12 +29,15 @@ describe('/v1/models route', () => {
const proxyRouterModule = await import('./router.js');
const tokenRouterModule = await import('../../services/tokenRouter.js');
const tokensRoutesModule = await import('../api/tokens.js');
const configModule = await import('../../config.js');
db = dbModule.db;
schema = dbModule.schema;
proxyRoutes = proxyRouterModule.proxyRoutes;
tokensRoutes = tokensRoutesModule.tokensRoutes;
invalidateTokenRouterCache = tokenRouterModule.invalidateTokenRouterCache;
config = configModule.config;
config.proxyToken = 'sk-global-proxy-token';
app = Fastify();
await app.register(tokensRoutes);
@@ -108,6 +113,7 @@ describe('/v1/models route', () => {
name: 'managed-key',
key: 'sk-managed-models',
enabled: true,
supportedModels: JSON.stringify(['routable-model']),
}).run();
const response = await app.inject({
@@ -129,6 +135,64 @@ describe('/v1/models route', () => {
expect(ids).not.toContain('orphan-model');
});
it('keeps global proxy token unrestricted when no managed key matches', async () => {
const site = await db.insert(schema.sites).values({
name: 'global-site',
url: 'https://global.example.com',
platform: 'openai',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
accessToken: 'global-access-token',
status: 'active',
}).returning().get();
const token = await db.insert(schema.accountTokens).values({
accountId: account.id,
name: 'default',
token: 'global-api-token',
enabled: true,
isDefault: true,
}).returning().get();
await db.insert(schema.modelAvailability).values({
accountId: account.id,
modelName: 'global-routable-model',
available: true,
}).run();
const route = await db.insert(schema.tokenRoutes).values({
modelPattern: 'global-routable-model',
enabled: true,
}).returning().get();
await db.insert(schema.routeChannels).values({
routeId: route.id,
accountId: account.id,
tokenId: token.id,
sourceModel: 'global-routable-model',
enabled: true,
}).run();
const response = await app.inject({
method: 'GET',
url: '/v1/models',
headers: {
authorization: 'Bearer sk-global-proxy-token',
},
});
expect(response.statusCode).toBe(200);
const body = response.json() as {
object: 'list';
data: Array<{ id: string }>;
};
expect(body.data.map((item) => item.id)).toContain('global-routable-model');
});
it('returns only whitelist models for managed key with supportedModels policy', async () => {
const site = await db.insert(schema.sites).values({
name: 'test-site',
@@ -290,6 +354,58 @@ describe('/v1/models route', () => {
expect(ids).not.toContain('claude-sonnet-4-5');
});
it('returns no models for managed key with empty model and group selections', async () => {
const site = await db.insert(schema.sites).values({
name: 'deny-all-site',
url: 'https://deny-all.example.com',
platform: 'openai',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
accessToken: 'deny-all-access-token',
status: 'active',
}).returning().get();
await db.insert(schema.modelAvailability).values([
{
accountId: account.id,
modelName: 'gpt-4o-mini',
available: true,
},
{
accountId: account.id,
modelName: 'claude-opus-4-6',
available: true,
},
]).run();
await db.insert(schema.downstreamApiKeys).values({
name: 'managed-key-deny-all',
key: 'sk-managed-deny-all',
enabled: true,
supportedModels: JSON.stringify([]),
allowedRouteIds: JSON.stringify([]),
}).run();
const response = await app.inject({
method: 'GET',
url: '/v1/models',
headers: {
authorization: 'Bearer sk-managed-deny-all',
},
});
expect(response.statusCode).toBe(200);
const body = response.json() as {
object: 'list';
data: Array<{ id: string }>;
};
expect(body.data).toEqual([]);
});
it('returns only explicit-group public name while hiding source exact routes', async () => {
const site = await db.insert(schema.sites).values({
name: 'explicit-group-site',
@@ -461,6 +577,7 @@ describe('/v1/models route', () => {
name: 'search-key',
key: 'sk-search-key',
enabled: true,
supportedModels: JSON.stringify(['gpt-4.1']),
}).run();
const response = await app.inject({
@@ -1,5 +1,6 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { config } from '../../config.js';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
@@ -17,6 +18,7 @@ const resolveProxyUsageWithSelfLogFallbackMock = vi.fn(async ({ usage }: any) =>
const refreshOauthAccessTokenSingleflightMock = vi.fn();
const recordOauthQuotaResetHintMock = vi.fn();
const insertedProxyLogs: Record<string, unknown>[] = [];
const originalProxyEmptyContentFailEnabled = config.proxyEmptyContentFailEnabled;
const dbInsertMock = vi.fn((_arg?: any) => ({
values: (values: Record<string, unknown>) => {
insertedProxyLogs.push(values);
@@ -78,6 +80,8 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
@@ -109,6 +113,7 @@ describe('responses proxy codex oauth refresh', () => {
});
beforeEach(() => {
config.proxyEmptyContentFailEnabled = false;
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
@@ -152,6 +157,7 @@ describe('responses proxy codex oauth refresh', () => {
});
afterAll(async () => {
config.proxyEmptyContentFailEnabled = originalProxyEmptyContentFailEnabled;
await app.close();
});
@@ -208,6 +214,69 @@ describe('responses proxy codex oauth refresh', () => {
expect(response.json()?.output_text).toContain('ok after codex token refresh');
});
it('retries oauth responses requests with a normalized upstream URL after refresh', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { name: 'openai-site', url: 'https://gateway.example.com/v1/', platform: 'openai' },
account: {
id: 33,
username: 'oauth-user@example.com',
extraConfig: JSON.stringify({
credentialMode: 'session',
oauth: {
provider: 'codex',
accountId: 'chatgpt-account-123',
email: 'oauth-user@example.com',
planType: 'plus',
},
}),
},
tokenName: 'default',
tokenValue: 'expired-access-token',
actualModel: 'gpt-4.1-mini',
});
fetchMock
.mockResolvedValueOnce(new Response(JSON.stringify({
error: { message: 'expired token', type: 'invalid_request_error' },
}), {
status: 401,
headers: { 'content-type': 'application/json' },
}))
.mockResolvedValueOnce(new Response(JSON.stringify({
id: 'resp_openai_refreshed',
object: 'response',
model: 'gpt-4.1-mini',
status: 'completed',
output_text: 'ok after refresh',
usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 },
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-4.1-mini',
input: 'hello oauth',
},
});
expect(response.statusCode).toBe(200);
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
expect(fetchMock).toHaveBeenCalledTimes(2);
const [firstUrl, firstOptions] = fetchMock.mock.calls[0] as [string, any];
const [secondUrl, secondOptions] = fetchMock.mock.calls[1] as [string, any];
expect(firstUrl).toBe('https://gateway.example.com/v1/responses');
expect(secondUrl).toBe('https://gateway.example.com/v1/responses');
expect(firstOptions.headers.Authorization).toBe('Bearer expired-access-token');
expect(secondOptions.headers.Authorization).toBe('Bearer fresh-access-token');
expect(response.json()?.output_text).toBe('ok after refresh');
});
it('sends an explicit empty instructions field to codex responses when downstream body has no system prompt', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'resp_codex_no_system',
@@ -501,6 +570,39 @@ describe('responses proxy codex oauth refresh', () => {
expect(String(insertedProxyLogs.at(-1)?.errorMessage || '')).toContain('stream closed before response.completed');
});
it('does not record success when a native responses stream completes with empty content and empty usage while empty-content failure is enabled', async () => {
config.proxyEmptyContentFailEnabled = true;
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_codex_empty","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_codex_empty","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}\n\n',
'data: [DONE]\n\n',
]));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.4',
input: 'hello codex',
stream: true,
},
});
expect(response.statusCode).toBe(200);
expect(response.body).toContain('event: response.failed');
expect(response.body).not.toContain('event: response.completed');
expect(recordSuccessMock).not.toHaveBeenCalled();
expect(recordFailureMock).toHaveBeenCalledTimes(1);
expect(insertedProxyLogs.at(-1)).toMatchObject({
status: 'failed',
httpStatus: 200,
});
expect(String(insertedProxyLogs.at(-1)?.errorMessage || '')).toContain('empty content');
});
it('does not retry or mark failure after converting a non-stream upstream payload into SSE when post-stream usage accounting fails', async () => {
resolveProxyUsageWithSelfLogFallbackMock.mockRejectedValueOnce(new Error('usage accounting failed'));
fetchMock.mockResolvedValue(new Response(JSON.stringify({
@@ -578,4 +680,39 @@ describe('responses proxy codex oauth refresh', () => {
expect(recordFailureMock).not.toHaveBeenCalled();
expect(recordSuccessMock).toHaveBeenCalledTimes(1);
});
it('replays raw codex SSE when upstream mislabels a streaming responses body as application/json', async () => {
fetchMock.mockResolvedValue(new Response([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_codex_stream_header_miss","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_item.added\n',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_codex_stream_header_miss","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_codex_stream_header_miss","delta":"pong"}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_codex_stream_header_miss","model":"gpt-5.4","status":"completed","usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
].join(''), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
payload: {
model: 'gpt-5.4',
input: 'hello codex',
stream: true,
},
});
expect(response.statusCode).toBe(200);
expect(response.body).toContain('event: response.output_text.delta');
expect(response.body).toContain('"delta":"pong"');
expect(response.body).not.toContain('"output_text":"event: response.created');
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(recordFailureMock).not.toHaveBeenCalled();
expect(recordSuccessMock).toHaveBeenCalledTimes(1);
});
});
@@ -67,6 +67,9 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
},
@@ -1,6 +1,7 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { createServer, type Server } from 'node:http';
import { AddressInfo } from 'node:net';
import WebSocket from 'ws';
import WebSocket, { WebSocketServer } from 'ws';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
@@ -72,6 +73,8 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
@@ -175,6 +178,14 @@ function waitForSocketMessages(socket: WebSocket, count: number, timeoutMs = 100
describe('responses websocket transport', () => {
let app: FastifyInstance;
let baseUrl: string;
let upstreamServer: WebSocketServer;
let upstreamSiteUrl: string;
let upstreamConnectionCount: number;
let upstreamUpgradeHeaders: Record<string, string>;
let upstreamRequests: Record<string, unknown>[];
let upstreamMessageHandler: (socket: WebSocket, parsed: Record<string, unknown>, requestIndex: number) => void;
let rejectedUpgradeServer: Server;
let rejectedUpgradeSiteUrl: string;
beforeAll(async () => {
const { responsesProxyRoute } = await import('./responses.js');
@@ -183,6 +194,39 @@ describe('responses websocket transport', () => {
await app.listen({ port: 0, host: '127.0.0.1' });
const address = app.server.address() as AddressInfo;
baseUrl = `ws://127.0.0.1:${address.port}`;
upstreamServer = new WebSocketServer({ port: 0 });
upstreamServer.on('connection', (socket, request) => {
upstreamConnectionCount += 1;
upstreamUpgradeHeaders = Object.fromEntries(
Object.entries(request.headers)
.map(([key, value]) => [key, Array.isArray(value) ? value[0] || '' : value || '']),
);
socket.on('message', (payload) => {
const parsed = JSON.parse(String(payload)) as Record<string, unknown>;
upstreamRequests.push(parsed);
upstreamMessageHandler(socket, parsed, upstreamRequests.length);
});
});
await new Promise<void>((resolve) => upstreamServer.once('listening', () => resolve()));
const upstreamAddress = upstreamServer.address() as AddressInfo;
upstreamSiteUrl = `http://127.0.0.1:${upstreamAddress.port}/backend-api/codex`;
rejectedUpgradeServer = createServer();
rejectedUpgradeServer.on('upgrade', (_request, socket) => {
socket.write(
'HTTP/1.1 426 Upgrade Required\r\n'
+ 'Content-Type: text/plain\r\n'
+ 'Content-Length: 16\r\n'
+ 'Connection: close\r\n'
+ '\r\n'
+ 'Upgrade Required',
);
socket.destroy();
});
await new Promise<void>((resolve) => rejectedUpgradeServer.listen(0, '127.0.0.1', () => resolve()));
const rejectedAddress = rejectedUpgradeServer.address() as AddressInfo;
rejectedUpgradeSiteUrl = `http://127.0.0.1:${rejectedAddress.port}/backend-api/codex`;
});
beforeEach(() => {
@@ -202,24 +246,90 @@ describe('responses websocket transport', () => {
selectChannelMock.mockReturnValue(selectedChannel);
selectNextChannelMock.mockReturnValue(null);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamConnectionCount = 0;
upstreamUpgradeHeaders = {};
upstreamRequests = [];
upstreamMessageHandler = (socket, parsed, requestIndex) => {
const responseId = `resp_upstream_${requestIndex}`;
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: responseId,
object: 'response',
model: parsed.model || 'gpt-5.4',
status: 'completed',
output: [],
usage: {
input_tokens: 1,
output_tokens: 1,
total_tokens: 2,
},
},
}));
};
});
afterAll(async () => {
await new Promise<void>((resolve) => rejectedUpgradeServer.close(() => resolve()));
await new Promise<void>((resolve) => upstreamServer.close(() => resolve()));
await app.close();
});
it('accepts response.create over GET /v1/responses websocket and forwards streamed responses events', async () => {
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_ws","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_item.added\n',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_ws","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_ws","delta":"pong"}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws","model":"gpt-5.4","status":"completed","output":[{"id":"msg_ws","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"pong"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket) => {
socket.send(JSON.stringify({
type: 'response.created',
response: {
id: 'resp_ws',
model: 'gpt-5.4',
created_at: 1706000000,
status: 'in_progress',
output: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_item.added',
output_index: 0,
item: {
id: 'msg_ws',
type: 'message',
role: 'assistant',
status: 'in_progress',
content: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_text.delta',
output_index: 0,
item_id: 'msg_ws',
delta: 'pong',
}));
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_ws',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'pong' }],
}],
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
}));
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -247,7 +357,8 @@ describe('responses websocket transport', () => {
'response.completed',
]);
expect(messages[3]?.response?.output?.[0]?.content?.[0]?.text).toBe('pong');
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamConnectionCount).toBe(1);
});
it('echoes x-codex-turn-state on websocket upgrade responses', async () => {
@@ -266,6 +377,89 @@ describe('responses websocket transport', () => {
expect(upgrade.headers['x-codex-turn-state']).toBe('turn-state-123');
});
it('reuses one upstream codex websocket session across sequential websocket turns', async () => {
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
headers: {
'x-codex-turn-state': 'turn-state-123',
'x-codex-beta-features': 'feature-a,feature-b',
},
});
await waitForSocketOpen(socket);
const firstMessagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
input: [],
}));
const firstMessages = await firstMessagesPromise;
const secondMessagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
previous_response_id: firstMessages[0]?.response?.id,
input: [],
}));
const secondMessages = await secondMessagesPromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(firstMessages[0]?.type).toBe('response.completed');
expect(secondMessages[0]?.type).toBe('response.completed');
expect(upstreamConnectionCount).toBe(1);
expect(upstreamRequests).toHaveLength(2);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
model: 'gpt-5.4',
});
expect(upstreamRequests[1]).toMatchObject({
type: 'response.create',
previous_response_id: firstMessages[0]?.response?.id,
});
expect(upstreamUpgradeHeaders['x-codex-turn-state']).toBe('turn-state-123');
expect(upstreamUpgradeHeaders['x-codex-beta-features']).toBe('feature-a,feature-b');
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=');
});
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 426', async () => {
const selectedChannel = createSelectedChannel({
siteUrl: rejectedUpgradeSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
fetchMock.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_fallback","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
'data: [DONE]\n\n',
]));
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const messagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
input: [],
}));
const messages = await messagesPromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
expect(messages[0]?.response?.id).toBe('resp_http_fallback');
});
it('merges follow-up response.create payloads when the selected upstream does not support incremental mode', async () => {
const selectedChannel = createSelectedChannel({
sitePlatform: 'openai',
@@ -364,21 +558,56 @@ describe('responses websocket transport', () => {
});
it('preserves incremental response.create payloads with previous_response_id for websocket-capable upstreams', async () => {
fetchMock
.mockResolvedValueOnce(createSseResponse([
'event: response.output_item.done\n',
'data: {"type":"response.output_item.done","output_index":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1"}}\n\n',
'event: response.output_item.done\n',
'data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws_1","model":"gpt-5.4","status":"completed","output":[{"id":"fc_1","type":"function_call","call_id":"call_1"},{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]))
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws_2","model":"gpt-5.4","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket, _parsed, requestIndex) => {
if (requestIndex === 1) {
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws_1',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_1',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'call tool' }],
}],
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
}));
return;
}
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws_2',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_2',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'done' }],
}],
usage: {
input_tokens: 5,
output_tokens: 1,
total_tokens: 6,
},
},
}));
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -415,12 +644,92 @@ describe('responses websocket transport', () => {
await secondResponsePromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamConnectionCount).toBe(1);
expect(upstreamRequests).toHaveLength(2);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
model: 'gpt-5.4',
instructions: 'be helpful',
});
expect(upstreamRequests[1]).toMatchObject({
type: 'response.create',
previous_response_id: 'resp_ws_1',
model: 'gpt-5.4',
instructions: 'be helpful',
input: [
{
id: 'tool_out_1',
type: 'function_call_output',
call_id: 'call_1',
output: 'tool result',
},
],
});
});
it('disables codex websocket incremental transport when the selected account marks websockets as disabled', async () => {
const selectedChannel = createSelectedChannel({
extraConfig: JSON.stringify({
credentialMode: 'session',
websockets: false,
oauth: {
provider: 'codex',
accountId: 'chatgpt-account-123',
email: 'codex-user@example.com',
},
}),
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
fetchMock
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_1","model":"gpt-5.4","status":"completed","output":[{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"first"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]))
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_2","model":"gpt-5.4","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"second"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
'data: [DONE]\n\n',
]));
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const firstResponsePromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
instructions: 'be helpful',
input: [],
}));
const firstMessages = await firstResponsePromise;
const secondResponsePromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
previous_response_id: 'resp_http_1',
input: [
{
id: 'tool_out_1',
type: 'function_call_output',
call_id: 'call_1',
output: 'tool result',
},
],
}));
await secondResponsePromise;
socket.close();
expect(firstMessages[0]?.type).toBe('response.completed');
expect(upstreamConnectionCount).toBe(0);
expect(fetchMock).toHaveBeenCalledTimes(2);
const [, secondOptions] = fetchMock.mock.calls[1] as [string, RequestInit];
const secondBody = JSON.parse(String(secondOptions.body));
expect(secondBody.previous_response_id).toBe('resp_ws_1');
expect(secondBody.model).toBe('gpt-5.4');
expect(secondBody.instructions).toBe('be helpful');
expect(secondBody.previous_response_id).toBeUndefined();
expect(secondBody.input).toEqual([
{
id: 'tool_out_1',
@@ -490,11 +799,11 @@ describe('responses websocket transport', () => {
});
it('forwards generate=false upstream for websocket-capable upstreams instead of synthesizing prewarm events', async () => {
fetchMock.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_capable_generate_false","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -510,23 +819,42 @@ describe('responses websocket transport', () => {
socket.close();
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
expect(fetchMock).toHaveBeenCalledTimes(1);
const [, options] = fetchMock.mock.calls[0] as [string, RequestInit];
const forwardedBody = JSON.parse(String(options.body));
expect(forwardedBody.generate).toBe(false);
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
generate: false,
});
});
it('emits websocket error when the upstream stream closes before a terminal responses event', async () => {
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_incomplete","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_ws","delta":"partial"}\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket) => {
socket.send(JSON.stringify({
type: 'response.created',
response: {
id: 'resp_incomplete',
model: 'gpt-5.4',
created_at: 1706000000,
status: 'in_progress',
output: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_text.delta',
output_index: 0,
item_id: 'msg_ws',
delta: 'partial',
}));
socket.close();
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const messagesPromise = waitForSocketMessages(socket, 6, 400);
const messagesPromise = waitForSocketMessages(socket, 3, 400);
socket.send(JSON.stringify({
type: 'response.create',
@@ -546,12 +874,8 @@ describe('responses websocket transport', () => {
expect(messages.map((message) => message.type)).toEqual([
'response.created',
'response.output_text.delta',
'response.output_text.done',
'response.content_part.done',
'response.output_item.done',
'response.failed',
'error',
]);
expect(messages[5]?.response?.status).toBe('failed');
expect(messages[5]?.error?.message).toContain('stream closed before response.completed');
expect(messages[2]?.error?.message).toContain('stream closed before response.completed');
});
});
@@ -20,4 +20,56 @@ describe('collectResponsesFinalPayloadFromSse', () => {
.rejects
.toThrow('quota exceeded');
});
it('prefers aggregated stream content when response.completed only carries an empty output array', async () => {
const upstream = {
async text() {
return [
'event: response.created',
'data: {"type":"response.created","response":{"id":"resp_empty_completed","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}',
'',
'event: response.output_item.added',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_empty_completed","type":"message","role":"assistant","status":"in_progress","content":[]}}',
'',
'event: response.output_text.delta',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_empty_completed","delta":"pong"}',
'',
'event: response.completed',
'data: {"type":"response.completed","response":{"id":"resp_empty_completed","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}',
'',
'data: [DONE]',
'',
].join('\n');
},
};
await expect(collectResponsesFinalPayloadFromSse(upstream, 'gpt-5.4'))
.resolves
.toMatchObject({
payload: {
id: 'resp_empty_completed',
status: 'completed',
output: [
{
id: 'msg_empty_completed',
type: 'message',
role: 'assistant',
status: 'completed',
content: [
{
type: 'output_text',
text: 'pong',
},
],
},
],
output_text: 'pong',
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
});
});
});
+149 -10
View File
@@ -24,6 +24,36 @@ function getResponsesFailureMessage(payload: Record<string, unknown>): string {
return 'upstream stream failed';
}
function hasMeaningfulMessageContent(content: unknown): boolean {
if (!Array.isArray(content)) return false;
return content.some((part) => {
if (!isRecord(part)) return false;
const partType = typeof part.type === 'string' ? part.type.trim().toLowerCase() : '';
if (partType === 'output_text' || partType === 'text') {
return typeof part.text === 'string' && part.text.length > 0;
}
return true;
});
}
function hasMeaningfulResponsesOutput(output: unknown): boolean {
if (!Array.isArray(output)) return false;
return output.some((item) => {
if (!isRecord(item)) return false;
const itemType = typeof item.type === 'string' ? item.type.trim().toLowerCase() : '';
if (itemType === 'message') {
return hasMeaningfulMessageContent(item.content);
}
if (itemType === 'reasoning') {
return (
(Array.isArray(item.summary) && item.summary.length > 0)
|| (typeof item.encrypted_content === 'string' && item.encrypted_content.trim().length > 0)
);
}
return itemType.length > 0;
});
}
function hasCompleteFinalResponsesPayload(payload: Record<string, unknown>): boolean {
return (
payload.object === 'response.compaction'
@@ -32,11 +62,101 @@ function hasCompleteFinalResponsesPayload(payload: Record<string, unknown>): boo
);
}
export async function collectResponsesFinalPayloadFromSse(
upstream: { text(): Promise<string> },
function hasMeaningfulFinalResponsesPayload(payload: Record<string, unknown>): boolean {
if (payload.object === 'response.compaction') {
return Array.isArray(payload.output) && payload.output.length > 0;
}
if (typeof payload.output_text === 'string' && payload.output_text.length > 0) {
return true;
}
return hasMeaningfulResponsesOutput(payload.output);
}
function rememberStreamResponseEnvelope(
streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
payload: Record<string, unknown>,
): void {
const responsePayload = isRecord(payload.response) ? payload.response : payload;
if (typeof responsePayload.id === 'string' && responsePayload.id.trim().length > 0) {
streamContext.id = responsePayload.id;
}
if (typeof responsePayload.model === 'string' && responsePayload.model.trim().length > 0) {
streamContext.model = responsePayload.model;
}
const createdAt = (
typeof responsePayload.created_at === 'number' && Number.isFinite(responsePayload.created_at)
? responsePayload.created_at
: (typeof responsePayload.created === 'number' && Number.isFinite(responsePayload.created)
? responsePayload.created
: null)
);
if (createdAt !== null) {
streamContext.created = createdAt;
}
}
function materializeCompletedPayloadFromAggregate(
aggregateState: ReturnType<typeof openAiResponsesTransformer.aggregator.createState>,
streamContext: ReturnType<typeof openAiResponsesTransformer.createStreamContext>,
usage: ReturnType<typeof parseProxyUsage>,
): Record<string, unknown> | null {
const lines = openAiResponsesTransformer.aggregator.complete(
aggregateState,
streamContext,
usage,
);
const { events } = openAiResponsesTransformer.pullSseEvents(lines.join(''));
for (const event of events) {
if (event.data === '[DONE]') continue;
const payload = parseResponsesSsePayload(event.data);
if (!payload) continue;
if (event.event === 'response.completed' && isRecord(payload.response)) {
return payload.response;
}
if (payload.type === 'response.completed' && hasCompleteFinalResponsesPayload(payload)) {
return payload;
}
}
return null;
}
export function looksLikeResponsesSseText(rawText: string): boolean {
const { events, rest } = openAiResponsesTransformer.pullSseEvents(rawText);
if (events.length === 0 || rest.trim().length > 0) return false;
return events.some((event) => {
if (event.data === '[DONE]') return true;
if (event.event === 'error' || event.event.startsWith('response.')) return true;
const payload = parseResponsesSsePayload(event.data);
const payloadType = typeof payload?.type === 'string' ? payload.type : '';
return payloadType === 'error' || payloadType.startsWith('response.');
});
}
export function createSingleChunkStreamReader(rawText: string): {
read(): Promise<{ done: boolean; value?: Uint8Array }>;
cancel(reason?: unknown): Promise<unknown>;
releaseLock(): void;
} {
const chunk = Buffer.from(rawText, 'utf8');
let done = false;
return {
async read() {
if (done) return { done: true };
done = true;
return { done: false, value: chunk };
},
async cancel() {
done = true;
return undefined;
},
releaseLock() {},
};
}
export function collectResponsesFinalPayloadFromSseText(
rawText: string,
modelName: string,
): Promise<{ payload: Record<string, unknown>; rawText: string }> {
const rawText = await upstream.text();
): { payload: Record<string, unknown>; rawText: string } {
const { events } = openAiResponsesTransformer.pullSseEvents(rawText);
const streamContext = openAiResponsesTransformer.createStreamContext(modelName);
const aggregateState = openAiResponsesTransformer.aggregator.createState(modelName);
@@ -92,12 +212,17 @@ export async function collectResponsesFinalPayloadFromSse(
const payloadType = typeof payload.type === 'string' ? payload.type : '';
const eventType = payloadType || event.event;
rememberStreamResponseEnvelope(streamContext, payload);
usage = mergeProxyUsage(usage, parseProxyUsage(payload));
captureCompletedPayloadFromEvent(eventType, payload);
if (completedPayload) {
continue;
}
const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(payload, streamContext, modelName);
const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(
payload,
streamContext,
modelName,
);
captureCompletedPayloadFromLines(openAiResponsesTransformer.aggregator.serialize({
state: aggregateState,
streamContext,
@@ -106,12 +231,19 @@ export async function collectResponsesFinalPayloadFromSse(
}));
}
if (
completedPayload
&& !hasMeaningfulFinalResponsesPayload(completedPayload)
&& hasMeaningfulResponsesOutput(aggregateState.outputItems)
) {
completedPayload = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage);
}
if (!completedPayload) {
captureCompletedPayloadFromLines(openAiResponsesTransformer.aggregator.complete(
aggregateState,
streamContext,
usage,
));
const materialized = materializeCompletedPayloadFromAggregate(aggregateState, streamContext, usage);
if (materialized) {
completedPayload = materialized;
}
}
if (completedPayload) {
@@ -123,3 +255,10 @@ export async function collectResponsesFinalPayloadFromSse(
throw new Error('stream disconnected before response.completed');
}
export async function collectResponsesFinalPayloadFromSse(
upstream: { text(): Promise<string> },
modelName: string,
): Promise<{ payload: Record<string, unknown>; rawText: string }> {
return collectResponsesFinalPayloadFromSseText(await upstream.text(), modelName);
}
+259 -56
View File
@@ -2,13 +2,19 @@ import { randomUUID } from 'node:crypto';
import type { FastifyInstance } from 'fastify';
import type { IncomingMessage } from 'node:http';
import { WebSocketServer, type RawData, type WebSocket } from 'ws';
import { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } from '../../proxy-core/runtime/codexWebsocketRuntime.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { buildOauthProviderHeaders } from '../../services/oauth/service.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import { buildUpstreamEndpointRequest } from './upstreamEndpoint.js';
const installedApps = new WeakSet<FastifyInstance>();
const WS_TURN_STATE_HEADER = 'x-codex-turn-state';
const RESPONSES_WEBSOCKET_MODE_HEADER = 'x-metapi-responses-websocket-mode';
const RESPONSES_WEBSOCKET_TRANSPORT_HEADER = 'x-metapi-responses-websocket-transport';
const codexWebsocketRuntime = createCodexWebsocketRuntime();
type SelectedChannel = NonNullable<Awaited<ReturnType<typeof tokenRouter.selectChannel>>>;
type NormalizedResponsesWebsocketRequest =
| {
@@ -42,6 +48,93 @@ function headerValueToTrimmedString(value: unknown): string {
return '';
}
function toBooleanLike(value: unknown): boolean | null {
if (typeof value === 'boolean') return value;
if (typeof value === 'string') {
const normalized = value.trim().toLowerCase();
if (normalized === 'true' || normalized === '1' || normalized === 'yes' || normalized === 'on') return true;
if (normalized === 'false' || normalized === '0' || normalized === 'no' || normalized === 'off') return false;
}
return null;
}
function parseExtraConfigRecord(extraConfig: unknown): Record<string, unknown> | null {
if (typeof extraConfig !== 'string') return null;
try {
const parsed = JSON.parse(extraConfig);
return isRecord(parsed) ? parsed : null;
} catch {
return null;
}
}
function readNestedRecord(value: unknown, key: string): Record<string, unknown> | null {
if (!isRecord(value)) return null;
const nested = value[key];
return isRecord(nested) ? nested : null;
}
function selectedChannelModelMatches(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const selectedModel = asTrimmedString(selectedChannel.actualModel).toLowerCase();
const normalizedRequestModel = asTrimmedString(requestModel).toLowerCase();
if (!selectedModel || !normalizedRequestModel) return true;
return selectedModel === normalizedRequestModel;
}
function selectedChannelSupportsCodexWebsocketTransport(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const platform = asTrimmedString(selectedChannel.site?.platform).toLowerCase();
if (platform !== 'codex') return false;
if (!selectedChannelModelMatches(selectedChannel, requestModel)) return false;
const extraConfig = parseExtraConfigRecord(selectedChannel.account.extraConfig);
const oauth = readNestedRecord(extraConfig, 'oauth');
const providerData = readNestedRecord(oauth, 'providerData');
const candidateFlags = [
extraConfig?.websockets,
readNestedRecord(extraConfig, 'attributes')?.websockets,
readNestedRecord(extraConfig, 'metadata')?.websockets,
providerData?.websockets,
readNestedRecord(providerData, 'attributes')?.websockets,
readNestedRecord(providerData, 'metadata')?.websockets,
];
for (const candidate of candidateFlags) {
const parsed = toBooleanLike(candidate);
if (parsed !== null) return parsed;
}
return true;
}
function selectedChannelSupportsIncrementalInput(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
return selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel);
}
function shouldReuseSelectedChannel(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const selectedModel = asTrimmedString(selectedChannel.actualModel).toLowerCase();
const normalizedRequestModel = asTrimmedString(requestModel).toLowerCase();
if (!selectedModel || !normalizedRequestModel) return true;
return selectedModel === normalizedRequestModel;
}
function deriveCodexExplicitSessionId(body: Record<string, unknown>, sessionId: string): string {
const promptCacheKey = asTrimmedString(body.prompt_cache_key);
return promptCacheKey || sessionId;
}
function parseJsonObject(raw: RawData): Record<string, unknown> | null {
try {
const parsed = JSON.parse(String(raw));
@@ -232,6 +325,77 @@ function collectResponsesOutput(payloads: unknown[]): unknown[] {
.map(([, value]) => value);
}
async function forwardResponsesRequestViaHttp(input: {
app: FastifyInstance;
socket: WebSocket;
request: IncomingMessage;
payload: Record<string, unknown>;
preserveIncrementalMode: boolean;
}): Promise<unknown[] | null> {
const response = await input.app.inject({
method: 'POST',
url: '/v1/responses',
headers: {
...buildInjectHeaders(input.request),
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
...(input.preserveIncrementalMode ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
},
payload: input.payload,
});
if (response.statusCode < 200 || response.statusCode >= 300) {
let payload: unknown = null;
try {
payload = JSON.parse(response.body);
} catch {
payload = null;
}
writeResponsesWebsocketError(
input.socket,
response.statusCode,
response.statusMessage || 'Upstream error',
payload,
);
return null;
}
const contentType = String(response.headers['content-type'] || '').toLowerCase();
if (!contentType.includes('text/event-stream')) {
try {
const payload = JSON.parse(response.body);
input.socket.send(JSON.stringify(payload));
return isRecord(payload?.response) && Array.isArray(payload.response.output)
? cloneJsonObject(payload.response.output)
: [];
} catch {
writeResponsesWebsocketError(input.socket, 502, 'Unexpected non-JSON websocket proxy response');
return null;
}
}
const pulled = openAiResponsesTransformer.pullSseEvents(response.body);
const forwardedPayloads: unknown[] = [];
let sawTerminalPayload = false;
for (const event of pulled.events) {
if (event.data === '[DONE]') continue;
try {
const payload = JSON.parse(event.data);
forwardedPayloads.push(payload);
const type = isRecord(payload) ? asTrimmedString(payload.type) : '';
if (type === 'response.completed' || type === 'response.failed') {
sawTerminalPayload = true;
}
input.socket.send(JSON.stringify(payload));
} catch {
// Ignore malformed SSE frames; the HTTP route already normalizes them.
}
}
if (!sawTerminalPayload) {
writeResponsesWebsocketError(input.socket, 408, 'stream closed before response.completed');
}
return collectResponsesOutput(forwardedPayloads);
}
function buildInjectHeaders(request: IncomingMessage): Record<string, string | string[]> {
const headers: Record<string, string | string[]> = {};
for (const [rawKey, rawValue] of Object.entries(request.headers)) {
@@ -262,7 +426,7 @@ async function supportsResponsesWebsocketIncrementalInput(
try {
const selected = await tokenRouter.previewSelectedChannel(requestModel);
return asTrimmedString(selected?.site?.platform).toLowerCase() === 'codex';
return selectedChannelSupportsIncrementalInput(selected, requestModel);
} catch {
return false;
}
@@ -273,8 +437,16 @@ async function handleResponsesWebsocketConnection(
socket: WebSocket,
request: IncomingMessage,
) {
const websocketSessionId = headerValueToTrimmedString(request.headers['session_id'])
|| headerValueToTrimmedString(request.headers['session-id'])
|| randomUUID();
let lastRequest: Record<string, unknown> | null = null;
let lastResponseOutput: unknown[] = [];
let selectedChannel: SelectedChannel | null = null;
socket.once('close', () => {
void codexWebsocketRuntime.closeSession(websocketSessionId);
});
socket.on('message', async (raw) => {
const parsed = parseJsonObject(raw);
@@ -283,7 +455,9 @@ async function handleResponsesWebsocketConnection(
return;
}
const supportsIncrementalInput = await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
parsed,
lastRequest,
@@ -309,67 +483,95 @@ async function handleResponsesWebsocketConnection(
return;
}
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
headers: {
...buildInjectHeaders(request),
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
selectedChannel = requestModel
? await tokenRouter.selectChannel(requestModel)
: null;
}
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
? selectedChannel
: null;
if (codexWebsocketChannel) {
const downstreamHeaders: Record<string, unknown> = {
...(request.headers as Record<string, unknown>),
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
},
payload: normalized.request,
});
};
const providerHeaders = buildOauthProviderHeaders({
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
? codexWebsocketChannel.account.extraConfig
: null,
downstreamHeaders,
});
const prepared = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
stream: true,
tokenValue: codexWebsocketChannel.tokenValue,
sitePlatform: codexWebsocketChannel.site.platform,
siteUrl: codexWebsocketChannel.site.url,
openaiBody: normalized.request,
downstreamFormat: 'responses',
responsesOriginalBody: normalized.request,
downstreamHeaders,
providerHeaders,
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
});
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
if (response.statusCode < 200 || response.statusCode >= 300) {
let payload: unknown = null;
try {
payload = JSON.parse(response.body);
} catch {
payload = null;
}
writeResponsesWebsocketError(
socket,
response.statusCode,
response.statusMessage || 'Upstream error',
payload,
);
return;
}
const contentType = String(response.headers['content-type'] || '').toLowerCase();
if (!contentType.includes('text/event-stream')) {
try {
const payload = JSON.parse(response.body);
socket.send(JSON.stringify(payload));
lastResponseOutput = isRecord(payload?.response) && Array.isArray(payload.response.output)
? cloneJsonObject(payload.response.output)
: [];
} catch {
writeResponsesWebsocketError(socket, 502, 'Unexpected non-JSON websocket proxy response');
}
return;
}
const pulled = openAiResponsesTransformer.pullSseEvents(response.body);
const forwardedPayloads: unknown[] = [];
let sawTerminalPayload = false;
for (const event of pulled.events) {
if (event.data === '[DONE]') continue;
try {
const payload = JSON.parse(event.data);
forwardedPayloads.push(payload);
const type = isRecord(payload) ? asTrimmedString(payload.type) : '';
if (type === 'response.completed' || type === 'response.failed') {
sawTerminalPayload = true;
const runtimeResult = await codexWebsocketRuntime.sendRequest({
sessionId: websocketSessionId,
requestUrl,
headers: prepared.headers,
body: prepared.body,
});
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
for (const payload of runtimeResult.events) {
socket.send(JSON.stringify(payload));
}
socket.send(JSON.stringify(payload));
} catch {
// Ignore malformed SSE frames; the HTTP route already normalizes them.
} catch (error) {
const runtimeError = error instanceof CodexWebsocketRuntimeError
? error
: new CodexWebsocketRuntimeError('upstream websocket request failed');
if (runtimeError.status === 426) {
const fallbackOutput = await forwardResponsesRequestViaHttp({
app,
socket,
request,
payload: normalized.request,
preserveIncrementalMode: false,
});
if (fallbackOutput) {
lastResponseOutput = fallbackOutput;
}
return;
}
lastResponseOutput = collectResponsesOutput(runtimeError.events);
for (const payload of runtimeError.events) {
socket.send(JSON.stringify(payload));
}
writeResponsesWebsocketError(
socket,
runtimeError.status || 408,
runtimeError.message,
runtimeError.payload,
);
}
return;
}
lastResponseOutput = collectResponsesOutput(forwardedPayloads);
if (!sawTerminalPayload) {
writeResponsesWebsocketError(socket, 408, 'stream closed before response.completed');
const forwardedOutput = await forwardResponsesRequestViaHttp({
app,
socket,
request,
payload: normalized.request,
preserveIncrementalMode: supportsIncrementalInput,
});
if (forwardedOutput) {
lastResponseOutput = forwardedOutput;
}
});
}
@@ -394,6 +596,7 @@ export function ensureResponsesWebsocketTransport(app: FastifyInstance) {
});
app.addHook('onClose', async () => {
await codexWebsocketRuntime.closeAllSessions();
await new Promise<void>((resolve) => {
websocketServer.close(() => resolve());
});
+3
View File
@@ -54,6 +54,9 @@ vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
hasProxyLogBillingDetailsColumn: async () => false,
hasProxyLogClientColumns: async () => false,
hasProxyLogDownstreamApiKeyIdColumn: async () => false,
schema: {
proxyLogs: {},
},
+59 -13
View File
@@ -1,6 +1,5 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
@@ -13,6 +12,8 @@ import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { getProxyAuthContext } from '../../middleware/auth.js';
import { buildUpstreamUrl } from './upstreamUrl.js';
import { detectDownstreamClientContext, type DownstreamClientContext } from './downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
const MAX_RETRIES = 2;
const DEFAULT_SEARCH_MODEL = '__search';
@@ -55,8 +56,12 @@ export async function searchProxyRoute(app: FastifyInstance) {
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
const downstreamPath = '/v1/search';
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const excludeChannelIds: number[] = [];
let retryCount = 0;
@@ -101,8 +106,23 @@ export async function searchProxyRoute(app: FastifyInstance) {
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: upstream.status,
errorText: text,
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
upstream.status,
Date.now() - startTime,
text,
retryCount,
downstreamApiKeyId,
clientContext,
downstreamPath,
);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
@@ -126,13 +146,28 @@ export async function searchProxyRoute(app: FastifyInstance) {
try { data = JSON.parse(text); } catch { data = { data: [] }; }
const latency = Date.now() - startTime;
tokenRouter.recordSuccess(selected.channel.id, latency, 0);
tokenRouter.recordSuccess(selected.channel.id, latency, 0, selected.actualModel);
recordDownstreamCostUsage(request, 0);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, downstreamApiKeyId, clientContext, downstreamPath);
return reply.code(upstream.status).send(data);
} catch (error: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, error?.message || 'network error', retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
tokenRouter.recordFailure(selected.channel.id, {
status: 0,
errorText: error?.message || 'network error',
modelName: selected.actualModel,
});
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
error?.message || 'network error',
retryCount,
downstreamApiKeyId,
clientContext,
downstreamPath,
);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
@@ -158,14 +193,16 @@ async function logProxy(
errorMessage: string | null,
retryCount: number,
downstreamApiKeyId: number | null = null,
clientContext: DownstreamClientContext | null = null,
downstreamPath = '/v1/search',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
await db.insert(schema.proxyLogs).values({
await insertProxyLog({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
downstreamApiKeyId,
modelRequested,
modelActual: selected.actualModel,
status,
@@ -176,12 +213,21 @@ async function logProxy(
totalTokens: 0,
estimatedCost: 0,
errorMessage: composeProxyLogMessage({
downstreamPath: '/v1/search',
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
errorMessage,
}),
clientFamily: clientContext?.clientKind || null,
clientAppId: clientContext?.clientAppId || null,
clientAppName: clientContext?.clientAppName || null,
clientConfidence: clientContext?.clientConfidence || null,
retryCount,
createdAt,
}).run();
});
} catch (error) {
console.warn('[proxy/search] failed to write proxy log', error);
}
@@ -1,4 +1,5 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { config } from '../../config.js';
const fetchModelPricingCatalogMock = vi.fn(async (_arg?: unknown): Promise<any> => null);
@@ -11,6 +12,9 @@ import {
buildUpstreamEndpointRequest,
isUnsupportedMediaTypeError,
isEndpointDowngradeError,
recordUpstreamEndpointFailure,
recordUpstreamEndpointSuccess,
resetUpstreamEndpointRuntimeState,
resolveUpstreamEndpointCandidates,
} from './upstreamEndpoint.js';
@@ -32,6 +36,18 @@ describe('resolveUpstreamEndpointCandidates', () => {
beforeEach(() => {
fetchModelPricingCatalogMock.mockReset();
fetchModelPricingCatalogMock.mockResolvedValue(null);
resetUpstreamEndpointRuntimeState();
(config as any).codexHeaderDefaults = {
userAgent: '',
betaFeatures: '',
};
(config as any).payloadRules = {
default: [],
defaultRaw: [],
override: [],
overrideRaw: [],
filter: [],
};
});
it('uses downstream-aligned endpoint priority for unknown platforms', async () => {
@@ -180,6 +196,129 @@ describe('resolveUpstreamEndpointCandidates', () => {
expect(order).toEqual(['responses', 'messages', 'chat']);
});
it('remembers the last successful endpoint per site capability profile', async () => {
recordUpstreamEndpointSuccess({
siteId: baseContext.site.id,
endpoint: 'responses',
downstreamFormat: 'openai',
modelName: 'gpt-5.3',
});
const order = await resolveUpstreamEndpointCandidates(
{
...baseContext,
site: { ...baseContext.site, platform: 'new-api' },
},
'gpt-5.3',
'openai',
);
expect(order).toEqual(['responses', 'chat', 'messages']);
});
it('keeps remote-document-url requests on a separate runtime preference bucket from inline document requests', async () => {
recordUpstreamEndpointSuccess({
siteId: baseContext.site.id,
endpoint: 'chat',
downstreamFormat: 'openai',
modelName: 'gpt-5.3',
requestCapabilities: {
hasNonImageFileInput: true,
conversationFileSummary: {
hasImage: false,
hasAudio: false,
hasDocument: true,
hasRemoteDocumentUrl: false,
},
},
});
const order = await resolveUpstreamEndpointCandidates(
{
...baseContext,
site: { ...baseContext.site, platform: 'new-api' },
},
'gpt-5.3',
'openai',
undefined,
{
hasNonImageFileInput: true,
conversationFileSummary: {
hasImage: false,
hasAudio: false,
hasDocument: true,
hasRemoteDocumentUrl: true,
},
},
);
expect(order).toEqual(['responses', 'messages', 'chat']);
});
it('does not remember messages fallback success for generic /v1/responses requests', async () => {
recordUpstreamEndpointSuccess({
siteId: baseContext.site.id,
endpoint: 'messages',
downstreamFormat: 'responses',
modelName: 'gpt-5.3',
});
const order = await resolveUpstreamEndpointCandidates(
{
...baseContext,
site: { ...baseContext.site, platform: 'new-api' },
},
'gpt-5.3',
'responses',
);
expect(order).toEqual(['responses', 'chat', 'messages']);
});
it('does not block generic /v1/responses endpoints on transient upstream errors', async () => {
recordUpstreamEndpointFailure({
siteId: baseContext.site.id,
endpoint: 'responses',
downstreamFormat: 'responses',
modelName: 'gpt-5.3',
status: 504,
errorText: '{"error":{"message":"Gateway time-out","type":"upstream_error"}}',
});
const order = await resolveUpstreamEndpointCandidates(
{
...baseContext,
site: { ...baseContext.site, platform: 'new-api' },
},
'gpt-5.3',
'responses',
);
expect(order).toEqual(['responses', 'chat', 'messages']);
});
it('learns a better endpoint from explicit upstream protocol errors', async () => {
recordUpstreamEndpointFailure({
siteId: baseContext.site.id,
endpoint: 'chat',
downstreamFormat: 'openai',
modelName: 'gpt-5.3',
status: 400,
errorText: 'Unsupported legacy protocol: /v1/chat/completions is not supported. Please use /v1/responses.',
});
const order = await resolveUpstreamEndpointCandidates(
{
...baseContext,
site: { ...baseContext.site, platform: 'new-api' },
},
'gpt-5.3',
'openai',
);
expect(order).toEqual(['responses', 'messages']);
});
it('keeps claude models messages-first even when openai platform catalog prefers chat', async () => {
fetchModelPricingCatalogMock.mockResolvedValue({
models: [
@@ -468,6 +607,136 @@ describe('buildUpstreamEndpointRequest', () => {
expect(request.body.prompt_cache_key).toBe('codex-cache-123');
});
it('applies configured codex header defaults with CLIProxyAPI-compatible precedence', () => {
(config as any).codexHeaderDefaults = {
userAgent: 'codex-config-ua/1.0',
betaFeatures: 'multi_agent',
};
const websocketRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: true,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello codex' }],
},
downstreamFormat: 'openai',
downstreamHeaders: {
'x-metapi-responses-websocket-transport': '1',
},
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(websocketRequest.headers['User-Agent']).toBe('codex-config-ua/1.0');
expect(websocketRequest.headers['x-codex-beta-features']).toBe('multi_agent');
const clientHeaderRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: true,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello again codex' }],
},
downstreamFormat: 'openai',
downstreamHeaders: {
'x-metapi-responses-websocket-transport': '1',
'user-agent': 'client-ua/2.0',
'x-codex-beta-features': 'client-beta',
},
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(clientHeaderRequest.headers['User-Agent']).toBe('codex-config-ua/1.0');
expect(clientHeaderRequest.headers['x-codex-beta-features']).toBe('client-beta');
const httpRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'plain http codex' }],
},
downstreamFormat: 'openai',
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(httpRequest.headers['x-codex-beta-features']).toBeUndefined();
});
it('applies configured payload rules before preparing codex responses requests', () => {
(config as any).payloadRules = {
default: [
{
models: [{ name: 'gpt-*', protocol: 'codex' }],
params: {
'reasoning.effort': 'high',
},
},
],
defaultRaw: [],
override: [
{
models: [{ name: 'gpt-5.4', protocol: 'codex' }],
params: {
'text.verbosity': 'low',
},
},
],
overrideRaw: [],
filter: [
{
models: [{ name: 'gpt-5.4', protocol: 'codex' }],
params: ['safety_identifier'],
},
],
};
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello codex' }],
verbosity: 'high',
safety_identifier: 'drop-me',
},
downstreamFormat: 'openai',
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(request.body.reasoning).toEqual({ effort: 'high' });
expect(request.body.text).toEqual({ verbosity: 'low' });
expect(request.body.safety_identifier).toBeUndefined();
});
it('builds gemini-cli native requests with project envelope and bearer headers', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
@@ -850,7 +1119,7 @@ describe('buildUpstreamEndpointRequest', () => {
]);
});
it('applies global responses standardization and drops non-standard fields', () => {
it('preserves unknown native responses fields while still normalizing known compatibility fields', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'upstream-gpt',
@@ -865,13 +1134,13 @@ describe('buildUpstreamEndpointRequest', () => {
input: 'hello',
metadata: { trace: 'abc123' },
max_completion_tokens: 512,
custom_vendor_flag: 'drop-me',
custom_vendor_flag: 'keep-me',
},
});
expect(request.path).toBe('/v1/responses');
expect(request.body.metadata).toEqual({ trace: 'abc123' });
expect(request.body.custom_vendor_flag).toBeUndefined();
expect(request.body.custom_vendor_flag).toBe('keep-me');
expect(request.body.max_completion_tokens).toBeUndefined();
expect(request.body.max_output_tokens).toBe(512);
expect(request.body.input).toEqual([
+330 -32
View File
@@ -1,6 +1,12 @@
import { createHash, randomUUID } from 'node:crypto';
import {
rankConversationFileEndpoints,
type ConversationFileInputSummary,
} from '../../proxy-core/capabilities/conversationFileCapabilities.js';
import { resolveProviderProfile } from '../../proxy-core/providers/registry.js';
import { config } from '../../config.js';
import { fetchModelPricingCatalog } from '../../services/modelPricingService.js';
import { applyPayloadRules } from '../../services/payloadRules.js';
import type { DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
convertOpenAiBodyToResponsesBody as convertOpenAiBodyToResponsesBodyViaTransformer,
@@ -13,7 +19,7 @@ import {
import {
buildGeminiGenerateContentRequestFromOpenAi,
} from './geminiCliCompat.js';
export {
import {
buildMinimalJsonHeadersForCompatibility,
isEndpointDispatchDeniedError,
isEndpointDowngradeError,
@@ -21,10 +27,31 @@ export {
promoteResponsesCandidateAfterLegacyChatError,
shouldPreferResponsesAfterLegacyChatError,
} from '../../transformers/shared/endpointCompatibility.js';
export {
buildMinimalJsonHeadersForCompatibility,
isEndpointDispatchDeniedError,
isEndpointDowngradeError,
isUnsupportedMediaTypeError,
promoteResponsesCandidateAfterLegacyChatError,
shouldPreferResponsesAfterLegacyChatError,
};
export type UpstreamEndpoint = 'chat' | 'messages' | 'responses';
export type EndpointPreference = DownstreamFormat | 'responses';
type EndpointCapabilityProfile = {
preferMessagesForClaudeModel: boolean;
hasNonImageFileInput: boolean;
hasRemoteDocumentUrl: boolean;
wantsNativeResponsesReasoning: boolean;
};
type EndpointRuntimeState = {
preferredEndpoint: UpstreamEndpoint | null;
preferredUpdatedAtMs: number;
blockedUntilMsByEndpoint: Partial<Record<UpstreamEndpoint, number>>;
};
type ChannelContext = {
site: {
id: number;
@@ -39,6 +66,10 @@ type ChannelContext = {
};
};
const ENDPOINT_RUNTIME_PREFERRED_TTL_MS = 24 * 60 * 60 * 1000;
const ENDPOINT_RUNTIME_BLOCK_TTL_MS = 6 * 60 * 60 * 1000;
const endpointRuntimeStates = new Map<string, EndpointRuntimeState>();
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
@@ -47,6 +78,20 @@ function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function resolveRequestedModelForPayloadRules(input: {
modelName: string;
openaiBody: Record<string, unknown>;
claudeOriginalBody?: Record<string, unknown>;
responsesOriginalBody?: Record<string, unknown>;
}): string {
return (
asTrimmedString(input.responsesOriginalBody?.model)
|| asTrimmedString(input.claudeOriginalBody?.model)
|| asTrimmedString(input.openaiBody.model)
|| asTrimmedString(input.modelName)
);
}
function normalizePlatformName(platform: unknown): string {
return asTrimmedString(platform).toLowerCase();
}
@@ -557,6 +602,225 @@ function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] {
return Array.from(normalized);
}
function buildEndpointCapabilityProfile(input?: {
modelName?: string;
requestedModelHint?: string;
requestCapabilities?: {
hasNonImageFileInput?: boolean;
conversationFileSummary?: ConversationFileInputSummary;
wantsNativeResponsesReasoning?: boolean;
};
}): EndpointCapabilityProfile {
return {
preferMessagesForClaudeModel: (
isClaudeFamilyModel(asTrimmedString(input?.modelName))
|| isClaudeFamilyModel(asTrimmedString(input?.requestedModelHint))
),
hasNonImageFileInput: (
input?.requestCapabilities?.conversationFileSummary?.hasDocument === true
|| input?.requestCapabilities?.hasNonImageFileInput === true
),
hasRemoteDocumentUrl: (
input?.requestCapabilities?.conversationFileSummary?.hasRemoteDocumentUrl === true
),
wantsNativeResponsesReasoning: input?.requestCapabilities?.wantsNativeResponsesReasoning === true,
};
}
function buildEndpointRuntimeStateKey(input: {
siteId: number;
downstreamFormat: EndpointPreference;
capabilityProfile: EndpointCapabilityProfile;
}): string {
const capabilityProfile = input.capabilityProfile;
return [
String(input.siteId),
input.downstreamFormat,
capabilityProfile.preferMessagesForClaudeModel ? 'claude' : 'generic',
capabilityProfile.hasNonImageFileInput ? 'files' : 'nofiles',
capabilityProfile.hasRemoteDocumentUrl ? 'remoteurl' : 'noremoteurl',
capabilityProfile.wantsNativeResponsesReasoning ? 'reasoning' : 'noreasoning',
].join(':');
}
function getOrCreateEndpointRuntimeState(key: string, nowMs = Date.now()): EndpointRuntimeState {
const existing = endpointRuntimeStates.get(key);
if (existing) return existing;
const initial: EndpointRuntimeState = {
preferredEndpoint: null,
preferredUpdatedAtMs: nowMs,
blockedUntilMsByEndpoint: {},
};
endpointRuntimeStates.set(key, initial);
return initial;
}
function maybeDeleteEndpointRuntimeState(key: string, nowMs = Date.now()): void {
const state = endpointRuntimeStates.get(key);
if (!state) return;
const hasActiveBlock = Object.values(state.blockedUntilMsByEndpoint).some((untilMs) => (
typeof untilMs === 'number' && untilMs > nowMs
));
const preferredFresh = (
!!state.preferredEndpoint
&& (state.preferredUpdatedAtMs + ENDPOINT_RUNTIME_PREFERRED_TTL_MS) > nowMs
);
if (!hasActiveBlock && !preferredFresh) {
endpointRuntimeStates.delete(key);
}
}
function applyEndpointRuntimePreference(
candidates: UpstreamEndpoint[],
key: string,
nowMs = Date.now(),
): UpstreamEndpoint[] {
const state = endpointRuntimeStates.get(key);
if (!state || candidates.length <= 1) return candidates;
const blocked = new Set<UpstreamEndpoint>();
for (const endpoint of candidates) {
const untilMs = state.blockedUntilMsByEndpoint[endpoint];
if (typeof untilMs === 'number' && untilMs > nowMs) {
blocked.add(endpoint);
}
}
let next = candidates.filter((endpoint) => !blocked.has(endpoint));
if (next.length === 0) {
next = [...candidates];
}
const preferredFresh = (
!!state.preferredEndpoint
&& (state.preferredUpdatedAtMs + ENDPOINT_RUNTIME_PREFERRED_TTL_MS) > nowMs
);
if (preferredFresh && state.preferredEndpoint && next.includes(state.preferredEndpoint)) {
next = [
state.preferredEndpoint,
...next.filter((endpoint) => endpoint !== state.preferredEndpoint),
];
}
maybeDeleteEndpointRuntimeState(key, nowMs);
return next;
}
function inferSuggestedEndpointFromError(errorText?: string | null): UpstreamEndpoint | null {
const text = (errorText || '').toLowerCase();
if (!text) return null;
if (text.includes('/v1/responses')) return 'responses';
if (text.includes('/v1/messages')) return 'messages';
if (text.includes('/v1/chat/completions')) return 'chat';
return null;
}
function shouldBlockEndpointByError(status: number, errorText?: string | null): boolean {
if (isEndpointDispatchDeniedError(status, errorText)) return true;
if (status === 404 || status === 405 || status === 415 || status === 501) return true;
if (isUnsupportedMediaTypeError(status, errorText)) return true;
const text = (errorText || '').toLowerCase();
return (
text.includes('convert_request_failed')
|| text.includes('endpoint_not_found')
|| text.includes('unknown_endpoint')
|| text.includes('unsupported_endpoint')
|| text.includes('unsupported_path')
|| text.includes('not_found_error')
|| text.includes('unsupported legacy protocol')
|| text.includes('please use /v1/')
|| text.includes('does not allow /v1/')
|| text.includes('unknown endpoint')
|| text.includes('unsupported endpoint')
|| text.includes('unsupported path')
|| text.includes('unrecognized request url')
|| text.includes('no route matched')
|| text.includes('does not exist')
);
}
function shouldRememberSuccessfulEndpoint(input: {
endpoint: UpstreamEndpoint;
downstreamFormat: EndpointPreference;
}): boolean {
if (input.downstreamFormat !== 'responses') return true;
return input.endpoint === 'responses';
}
export function resetUpstreamEndpointRuntimeState(): void {
endpointRuntimeStates.clear();
}
export function recordUpstreamEndpointSuccess(input: {
siteId: number;
endpoint: UpstreamEndpoint;
downstreamFormat: EndpointPreference;
modelName?: string;
requestedModelHint?: string;
requestCapabilities?: {
hasNonImageFileInput?: boolean;
conversationFileSummary?: ConversationFileInputSummary;
wantsNativeResponsesReasoning?: boolean;
};
}): void {
if (!shouldRememberSuccessfulEndpoint(input)) return;
const nowMs = Date.now();
const key = buildEndpointRuntimeStateKey({
siteId: input.siteId,
downstreamFormat: input.downstreamFormat,
capabilityProfile: buildEndpointCapabilityProfile({
modelName: input.modelName,
requestedModelHint: input.requestedModelHint,
requestCapabilities: input.requestCapabilities,
}),
});
const state = getOrCreateEndpointRuntimeState(key, nowMs);
state.preferredEndpoint = input.endpoint;
state.preferredUpdatedAtMs = nowMs;
delete state.blockedUntilMsByEndpoint[input.endpoint];
}
export function recordUpstreamEndpointFailure(input: {
siteId: number;
endpoint: UpstreamEndpoint;
downstreamFormat: EndpointPreference;
status: number;
errorText?: string | null;
modelName?: string;
requestedModelHint?: string;
requestCapabilities?: {
hasNonImageFileInput?: boolean;
conversationFileSummary?: ConversationFileInputSummary;
wantsNativeResponsesReasoning?: boolean;
};
}): void {
if (!shouldBlockEndpointByError(input.status, input.errorText)) return;
const nowMs = Date.now();
const key = buildEndpointRuntimeStateKey({
siteId: input.siteId,
downstreamFormat: input.downstreamFormat,
capabilityProfile: buildEndpointCapabilityProfile({
modelName: input.modelName,
requestedModelHint: input.requestedModelHint,
requestCapabilities: input.requestCapabilities,
}),
});
const state = getOrCreateEndpointRuntimeState(key, nowMs);
state.blockedUntilMsByEndpoint[input.endpoint] = nowMs + ENDPOINT_RUNTIME_BLOCK_TTL_MS;
const suggestedEndpoint = inferSuggestedEndpointFromError(input.errorText);
if (suggestedEndpoint && suggestedEndpoint !== input.endpoint) {
state.preferredEndpoint = suggestedEndpoint;
state.preferredUpdatedAtMs = nowMs;
delete state.blockedUntilMsByEndpoint[suggestedEndpoint];
}
}
function preferredEndpointOrder(
downstreamFormat: EndpointPreference,
sitePlatform?: string,
@@ -626,27 +890,44 @@ export async function resolveUpstreamEndpointCandidates(
requestedModelHint?: string,
requestCapabilities?: {
hasNonImageFileInput?: boolean;
conversationFileSummary?: ConversationFileInputSummary;
wantsNativeResponsesReasoning?: boolean;
},
): Promise<UpstreamEndpoint[]> {
const sitePlatform = normalizePlatformName(context.site.platform);
const preferMessagesForClaudeModel = (
isClaudeFamilyModel(modelName)
|| isClaudeFamilyModel(asTrimmedString(requestedModelHint))
const capabilityProfile = buildEndpointCapabilityProfile({
modelName,
requestedModelHint,
requestCapabilities,
});
const preferMessagesForClaudeModel = capabilityProfile.preferMessagesForClaudeModel;
const hasNonImageFileInput = capabilityProfile.hasNonImageFileInput;
const wantsNativeResponsesReasoning = capabilityProfile.wantsNativeResponsesReasoning;
const runtimeStateKey = buildEndpointRuntimeStateKey({
siteId: context.site.id,
downstreamFormat,
capabilityProfile,
});
const applyRuntimePreference = (candidates: UpstreamEndpoint[]) => (
applyEndpointRuntimePreference(candidates, runtimeStateKey)
);
const hasNonImageFileInput = requestCapabilities?.hasNonImageFileInput === true;
const wantsNativeResponsesReasoning = requestCapabilities?.wantsNativeResponsesReasoning === true;
const conversationFileSummary = requestCapabilities?.conversationFileSummary ?? {
hasImage: false,
hasAudio: false,
hasDocument: hasNonImageFileInput,
hasRemoteDocumentUrl: false,
};
if (sitePlatform === 'anyrouter') {
// anyrouter deployments are effectively anthropic-protocol first.
if (hasNonImageFileInput) {
return downstreamFormat === 'responses'
return applyRuntimePreference(downstreamFormat === 'responses'
? ['responses', 'messages', 'chat']
: ['messages', 'responses', 'chat'];
: ['messages', 'responses', 'chat']);
}
if (downstreamFormat === 'responses') {
return ['responses', 'messages', 'chat'];
return applyRuntimePreference(['responses', 'messages', 'chat']);
}
return ['messages', 'chat', 'responses'];
return applyRuntimePreference(['messages', 'chat', 'responses']);
}
const preferred = preferredEndpointOrder(
@@ -659,8 +940,14 @@ export async function resolveUpstreamEndpointCandidates(
if (sitePlatform === 'claude') return ['messages'] as UpstreamEndpoint[];
if (sitePlatform === 'gemini') return ['responses', 'chat'] as UpstreamEndpoint[];
if (sitePlatform === 'gemini-cli' || sitePlatform === 'antigravity') return ['chat'] as UpstreamEndpoint[];
if (preferMessagesForClaudeModel) return ['messages', 'responses', 'chat'] as UpstreamEndpoint[];
return ['responses', 'messages', 'chat'] as UpstreamEndpoint[];
return rankConversationFileEndpoints({
sitePlatform,
requestedOrder: preferMessagesForClaudeModel
? ['messages', 'responses', 'chat']
: ['responses', 'messages', 'chat'],
summary: conversationFileSummary,
preferMessagesForClaudeModel,
});
})()
: preferred;
const prioritizedPreferredEndpoints: UpstreamEndpoint[] = (
@@ -699,20 +986,20 @@ export async function resolveUpstreamEndpointCandidates(
});
if (!catalog || !Array.isArray(catalog.models) || catalog.models.length === 0) {
return prioritizedPreferredEndpoints;
return applyRuntimePreference(prioritizedPreferredEndpoints);
}
const matched = catalog.models.find((item) =>
asTrimmedString(item?.modelName).toLowerCase() === modelName.toLowerCase(),
);
if (!matched) return prioritizedPreferredEndpoints;
if (!matched) return applyRuntimePreference(prioritizedPreferredEndpoints);
const shouldIgnoreCatalogOrderingForClaudeMessages = (
preferMessagesForClaudeModel
&& (downstreamFormat !== 'responses' || sitePlatform !== 'openai')
);
if (shouldIgnoreCatalogOrderingForClaudeMessages) {
return prioritizedPreferredEndpoints;
return applyRuntimePreference(prioritizedPreferredEndpoints);
}
const supportedRaw = Array.isArray(matched.supportedEndpointTypes) ? matched.supportedEndpointTypes : [];
@@ -732,7 +1019,7 @@ export async function resolveUpstreamEndpointCandidates(
if (forceMessagesFirstForClaudeModel && !hasConcreteEndpointHint) {
// Generic labels like openai/anthropic are too coarse for Claude models;
// keep messages-first order in this case.
return prioritizedPreferredEndpoints;
return applyRuntimePreference(prioritizedPreferredEndpoints);
}
const supported = new Set<UpstreamEndpoint>();
@@ -743,19 +1030,19 @@ export async function resolveUpstreamEndpointCandidates(
}
}
if (supported.size === 0) return prioritizedPreferredEndpoints;
if (supported.size === 0) return applyRuntimePreference(prioritizedPreferredEndpoints);
const firstSupported = prioritizedPreferredEndpoints.find((endpoint) => supported.has(endpoint));
if (!firstSupported) return prioritizedPreferredEndpoints;
if (!firstSupported) return applyRuntimePreference(prioritizedPreferredEndpoints);
// Catalog metadata can be incomplete/inaccurate, so only use it to pick
// the first attempt. Keep downstream-driven fallback order unchanged.
return [
return applyRuntimePreference([
firstSupported,
...prioritizedPreferredEndpoints.filter((endpoint) => endpoint !== firstSupported),
];
]);
} catch {
return prioritizedPreferredEndpoints;
return applyRuntimePreference(prioritizedPreferredEndpoints);
}
}
@@ -885,6 +1172,16 @@ export function buildUpstreamEndpointRequest(input: {
stream: input.stream,
oauthProjectId: asTrimmedString(input.oauthProjectId) || null,
};
const requestedModelForPayloadRules = resolveRequestedModelForPayloadRules(input);
const applyConfiguredPayloadRules = <T extends Record<string, unknown>>(body: T): T => (
applyPayloadRules({
rules: config.payloadRules,
payload: body,
modelName: input.modelName,
requestedModel: requestedModelForPayloadRules,
protocol: sitePlatform,
}) as T
);
if (isInternalGeminiUpstream) {
const instructions = (
@@ -898,6 +1195,7 @@ export function buildUpstreamEndpointRequest(input: {
modelName: input.modelName,
instructions,
});
const configuredGeminiRequest = applyConfiguredPayloadRules(geminiRequest);
if (!providerProfile) {
throw new Error(`missing provider profile for platform: ${sitePlatform}`);
}
@@ -911,7 +1209,7 @@ export function buildUpstreamEndpointRequest(input: {
sitePlatform,
baseHeaders: commonHeaders,
providerHeaders: input.providerHeaders,
body: geminiRequest,
body: configuredGeminiRequest,
action: input.stream ? 'streamGenerateContent' : 'generateContent',
});
}
@@ -952,6 +1250,7 @@ export function buildUpstreamEndpointRequest(input: {
?? sanitizeAnthropicMessagesBody(
convertOpenAiBodyToAnthropicMessagesBody(openaiBody, input.modelName, input.stream),
);
const configuredClaudeBody = applyConfiguredPayloadRules(sanitizedBody);
if (providerProfile?.id === 'claude') {
return providerProfile.prepareRequest({
@@ -964,7 +1263,7 @@ export function buildUpstreamEndpointRequest(input: {
sitePlatform,
baseHeaders: commonHeaders,
claudeHeaders,
body: sanitizedBody,
body: configuredClaudeBody,
});
}
@@ -980,7 +1279,7 @@ export function buildUpstreamEndpointRequest(input: {
return {
path: resolveEndpointPath('messages'),
headers,
body: sanitizedBody,
body: configuredClaudeBody,
runtime,
};
}
@@ -1015,6 +1314,7 @@ export function buildUpstreamEndpointRequest(input: {
),
sitePlatform,
);
const configuredResponsesBody = applyConfiguredPayloadRules(body);
if (providerProfile?.id === 'codex') {
return providerProfile.prepareRequest({
@@ -1032,7 +1332,7 @@ export function buildUpstreamEndpointRequest(input: {
providerHeaders: input.providerHeaders,
codexSessionCacheKey: input.codexSessionCacheKey,
codexExplicitSessionId: input.codexExplicitSessionId,
body,
body: configuredResponsesBody,
});
}
@@ -1055,15 +1355,15 @@ export function buildUpstreamEndpointRequest(input: {
: null;
const shouldInjectDerivedPromptCacheKey = sitePlatform === 'codex'
&& !!codexSessionId
&& !asTrimmedString((body as Record<string, unknown>).prompt_cache_key)
&& !asTrimmedString((configuredResponsesBody as Record<string, unknown>).prompt_cache_key)
&& !asTrimmedString(input.codexExplicitSessionId)
&& !!asTrimmedString(input.codexSessionCacheKey);
const runtimeBody = shouldInjectDerivedPromptCacheKey
? {
...body,
...configuredResponsesBody,
prompt_cache_key: codexSessionId,
}
: body;
: configuredResponsesBody;
return {
path: resolveEndpointPath('responses'),
@@ -1077,11 +1377,11 @@ export function buildUpstreamEndpointRequest(input: {
return {
path: resolveEndpointPath('chat'),
headers,
body: {
body: applyConfiguredPayloadRules({
...openaiBody,
model: input.modelName,
stream: input.stream,
},
}),
runtime,
};
}
@@ -1149,5 +1449,3 @@ export function buildClaudeCountTokensUpstreamRequest(input: {
},
};
}
+3 -3
View File
@@ -102,7 +102,7 @@ export async function videosProxyRoute(app: FastifyInstance) {
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
tokenRouter.recordFailure(selected.channel.id, selected.actualModel);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
@@ -155,11 +155,11 @@ export async function videosProxyRoute(app: FastifyInstance) {
completionTokens: 0,
totalTokens: 0,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, selected.actualModel);
recordDownstreamCostUsage(request, estimatedCost);
return reply.code(upstream.status).send(rewriteVideoResponsePublicId(data, mapping.publicId));
} catch (error: any) {
tokenRouter.recordFailure(selected.channel.id);
tokenRouter.recordFailure(selected.channel.id, selected.actualModel);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;