boos/BoosAPI
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[codex] improve proxy stability, routing, and site subscription visibility (#194)

Browse Source 
* wip: preserve local changes before upstream merge

* Improve proxy routing stability and add stable-first strategy
This commit is contained in:
Bainily
2026-03-20 16:18:28 +08:00
committed by GitHub
parent e2595bcab3
commit e9c1bd898e
37 changed files with 3424 additions and 236 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/api/sites.subscription-summary.test.ts
+128
View File
@@ -0,0 +1,128 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { mkdtempSync } from 'node:fs';
import {
buildStoredSub2ApiSubscriptionSummary,
mergeAccountExtraConfig,
} from '../../services/accountExtraConfig.js';
type DbModule = typeof import('../../db/index.js');
describe('sites route subscription summary aggregation', () => {
let app: FastifyInstance;
let db: DbModule['db'];
let schema: DbModule['schema'];
let dataDir = '';
beforeAll(async () => {
dataDir = mkdtempSync(join(tmpdir(), 'metapi-sites-subscription-'));
process.env.DATA_DIR = dataDir;
await import('../../db/migrate.js');
const dbModule = await import('../../db/index.js');
const routesModule = await import('./sites.js');
db = dbModule.db;
schema = dbModule.schema;
app = Fastify();
await app.register(routesModule.sitesRoutes);
});
beforeEach(async () => {
await db.delete(schema.accounts).run();
await db.delete(schema.sites).run();
});
afterAll(async () => {
await app.close();
delete process.env.DATA_DIR;
});
it('aggregates stored sub2api subscription summaries into /api/sites', async () => {
const site = await db.insert(schema.sites).values({
name: 'sub2api-site',
url: 'https://sub2api.example.com',
platform: 'sub2api',
}).returning().get();
await db.insert(schema.accounts).values([
{
siteId: site.id,
username: 'user-a',
accessToken: 'token-a',
balance: 10,
status: 'active',
extraConfig: mergeAccountExtraConfig(null, {
sub2apiSubscription: buildStoredSub2ApiSubscriptionSummary({
activeCount: 1,
totalUsedUsd: 3,
subscriptions: [
{
id: 11,
groupName: 'Pro',
expiresAt: '2026-04-10T00:00:00.000Z',
monthlyUsedUsd: 3,
monthlyLimitUsd: 20,
},
],
}, 1760000000000),
}),
},
{
siteId: site.id,
username: 'user-b',
accessToken: 'token-b',
balance: 5,
status: 'active',
extraConfig: mergeAccountExtraConfig(null, {
sub2apiSubscription: buildStoredSub2ApiSubscriptionSummary({
activeCount: 1,
totalUsedUsd: 2,
subscriptions: [
{
id: 12,
groupName: 'Lite',
expiresAt: '2026-03-25T00:00:00.000Z',
monthlyUsedUsd: 2,
monthlyLimitUsd: 10,
},
],
}, 1760000002000),
}),
},
]).run();
const response = await app.inject({
method: 'GET',
url: '/api/sites',
});
expect(response.statusCode).toBe(200);
const body = response.json() as Array<{
totalBalance: number;
subscriptionSummary: {
activeCount: number;
totalUsedUsd: number;
totalMonthlyLimitUsd: number | null;
totalRemainingUsd: number | null;
nextExpiresAt: string | null;
planNames: string[];
updatedAt: number | null;
} | null;
}>;
expect(body).toHaveLength(1);
expect(body[0]?.totalBalance).toBe(15);
expect(body[0]?.subscriptionSummary).toEqual({
activeCount: 2,
totalUsedUsd: 5,
totalMonthlyLimitUsd: 30,
totalRemainingUsd: 25,
nextExpiresAt: '2026-03-25T00:00:00.000Z',
planNames: ['Pro', 'Lite'],
updatedAt: 1760000002000,
});
});
});
src/server/routes/api/sites.ts
+70 -8
View File
@@ -1,11 +1,12 @@
import { FastifyInstance } from 'fastify';
import { db, schema } from '../../db/index.js';
import { and, eq, sql } from 'drizzle-orm';
import { and, eq } from 'drizzle-orm';
import { detectSite } from '../../services/siteDetector.js';
import { invalidateSiteProxyCache } from '../../services/siteProxy.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { invalidateTokenRouterCache } from '../../services/tokenRouter.js';
import { parseSiteCustomHeadersInput } from '../../services/siteCustomHeaders.js';
import { getSub2ApiSubscriptionFromExtraConfig } from '../../services/accountExtraConfig.js';
function normalizeSiteStatus(input: unknown): 'active' | 'disabled' | null {
if (input === undefined || input === null) return null;
@@ -75,6 +76,65 @@ function normalizeOptionalExternalCheckinUrl(input: unknown): {
return { valid: true, present: true, url: parsed.toString().replace(/\/+$/, '') };
}
type SiteSubscriptionAggregate = {
activeCount: number;
totalUsedUsd: number;
totalMonthlyLimitUsd: number | null;
totalRemainingUsd: number | null;
nextExpiresAt: string | null;
planNames: string[];
updatedAt: number | null;
};
function roundMetric(value: number): number {
return Math.round(value * 1_000_000) / 1_000_000;
}
function pickEarlierIsoDate(current?: string | null, next?: string | null): string | null {
if (!current) return next || null;
if (!next) return current;
const currentMs = Date.parse(current);
const nextMs = Date.parse(next);
if (!Number.isFinite(currentMs)) return next;
if (!Number.isFinite(nextMs)) return current;
return nextMs < currentMs ? next : current;
}
function aggregateSiteSubscription(
current: SiteSubscriptionAggregate | undefined,
extraConfig?: string | null,
): SiteSubscriptionAggregate | undefined {
const stored = getSub2ApiSubscriptionFromExtraConfig(extraConfig);
if (!stored) return current;
const planNames = new Set(current?.planNames || []);
let totalMonthlyLimitUsd = current?.totalMonthlyLimitUsd ?? null;
let nextExpiresAt = current?.nextExpiresAt ?? null;
for (const item of stored.subscriptions) {
if (item.groupName) planNames.add(item.groupName);
if (typeof item.monthlyLimitUsd === 'number' && Number.isFinite(item.monthlyLimitUsd)) {
totalMonthlyLimitUsd = roundMetric((totalMonthlyLimitUsd ?? 0) + item.monthlyLimitUsd);
}
nextExpiresAt = pickEarlierIsoDate(nextExpiresAt, item.expiresAt);
}
const totalUsedUsd = roundMetric((current?.totalUsedUsd || 0) + stored.totalUsedUsd);
const totalRemainingUsd = totalMonthlyLimitUsd == null
? null
: roundMetric(Math.max(0, totalMonthlyLimitUsd - totalUsedUsd));
return {
activeCount: (current?.activeCount || 0) + stored.activeCount,
totalUsedUsd,
totalMonthlyLimitUsd,
totalRemainingUsd,
nextExpiresAt,
planNames: Array.from(planNames),
updatedAt: Math.max(current?.updatedAt || 0, stored.updatedAt || 0) || null,
};
}
export async function sitesRoutes(app: FastifyInstance) {
function invalidateSiteCaches() {
invalidateSiteProxyCache();
@@ -137,21 +197,23 @@ export async function sitesRoutes(app: FastifyInstance) {
// List all sites
app.get('/api/sites', async () => {
const siteRows = await db.select().from(schema.sites).all();
const accountBalanceRows = await db.select({
const accountRows = await db.select({
siteId: schema.accounts.siteId,
totalBalance: sql<number>`coalesce(sum(${schema.accounts.balance}), 0)`,
}).from(schema.accounts)
.groupBy(schema.accounts.siteId)
.all();
balance: schema.accounts.balance,
extraConfig: schema.accounts.extraConfig,
}).from(schema.accounts).all();
const totalBalanceBySiteId: Record<number, number> = {};
for (const row of accountBalanceRows) {
totalBalanceBySiteId[row.siteId] = Number(row.totalBalance || 0);
const subscriptionBySiteId: Record<number, SiteSubscriptionAggregate | undefined> = {};
for (const row of accountRows) {
totalBalanceBySiteId[row.siteId] = roundMetric((totalBalanceBySiteId[row.siteId] || 0) + Number(row.balance || 0));
subscriptionBySiteId[row.siteId] = aggregateSiteSubscription(subscriptionBySiteId[row.siteId], row.extraConfig);
}
return siteRows.map((site) => ({
...site,
totalBalance: Math.round((totalBalanceBySiteId[site.id] || 0) * 1_000_000) / 1_000_000,
subscriptionSummary: subscriptionBySiteId[site.id] || null,
}));
});
src/server/routes/proxy/chat.stream.test.ts
+13
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();
@@ -102,6 +103,7 @@ describe('chat proxy stream behavior', () => {
fetchModelPricingCatalogMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
dbInsertMock.mockClear();
resetUpstreamEndpointRuntimeState();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
@@ -113,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 = [];
});
src/server/routes/proxy/completions.ts
+81 -11
View File
@@ -79,8 +79,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, downstreamApiKeyId, 0, 0, 0, 0, null, clientContext, downstreamPath);
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({
@@ -178,11 +198,24 @@ export async function completionsProxyRoute(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', 200, latency, null, retryCount, downstreamApiKeyId,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails, clientContext, downstreamPath,
selected,
requestedModel,
'success',
200,
latency,
null,
retryCount,
downstreamApiKeyId,
resolvedUsage.promptTokens,
resolvedUsage.completionTokens,
resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
clientContext,
downstreamPath,
);
return;
}
@@ -199,7 +232,11 @@ 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,
@@ -256,16 +293,49 @@ export async function completionsProxyRoute(app: FastifyInstance) {
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
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,
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);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, downstreamApiKeyId, 0, 0, 0, 0, null, clientContext, downstreamPath);
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;
src/server/routes/proxy/embeddings.ts
+45 -5
View File
@@ -75,8 +75,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, downstreamApiKeyId, 0, 0, 0, 0, null, clientContext, downstreamPath);
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({
@@ -126,7 +146,7 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
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, downstreamApiKeyId,
@@ -134,8 +154,28 @@ export async function embeddingsProxyRoute(app: FastifyInstance) {
);
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, downstreamApiKeyId, 0, 0, 0, 0, null, clientContext, downstreamPath);
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;
src/server/routes/proxy/endpointFlow.test.ts
+34
View File
@@ -155,6 +155,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' },
src/server/routes/proxy/endpointFlow.ts
+28
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;
@@ -61,6 +69,8 @@ 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 {
@@ -100,6 +110,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 +137,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 +159,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);
src/server/routes/proxy/gemini.test.ts
+25 -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();
@@ -188,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);
});
@@ -261,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');
@@ -1417,7 +1422,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');
@@ -1467,7 +1475,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 () => {
@@ -1510,7 +1521,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 () => {
@@ -1550,7 +1564,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 () => {
@@ -1596,7 +1612,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');
});
src/server/routes/proxy/images.ts
+74 -10
View File
@@ -74,8 +74,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, downstreamApiKeyId, 0, downstreamPath, clientContext);
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,
@@ -107,13 +123,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, 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, downstreamApiKeyId, 0, downstreamPath, clientContext);
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;
@@ -200,8 +232,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, downstreamApiKeyId, 0, downstreamPath, clientContext);
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,
@@ -233,13 +281,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, 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, downstreamApiKeyId, 0, downstreamPath, clientContext);
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;
src/server/routes/proxy/search.ts
+35 -5
View File
@@ -105,8 +105,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, downstreamApiKeyId, clientContext, downstreamPath);
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,
@@ -130,13 +145,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, 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, downstreamApiKeyId, clientContext, downstreamPath);
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;
src/server/routes/proxy/upstreamEndpoint.test.ts
+46
View File
@@ -12,6 +12,9 @@ import {
buildUpstreamEndpointRequest,
isUnsupportedMediaTypeError,
isEndpointDowngradeError,
recordUpstreamEndpointFailure,
recordUpstreamEndpointSuccess,
resetUpstreamEndpointRuntimeState,
resolveUpstreamEndpointCandidates,
} from './upstreamEndpoint.js';
@@ -33,6 +36,7 @@ describe('resolveUpstreamEndpointCandidates', () => {
beforeEach(() => {
fetchModelPricingCatalogMock.mockReset();
fetchModelPricingCatalogMock.mockResolvedValue(null);
resetUpstreamEndpointRuntimeState();
(config as any).codexHeaderDefaults = {
userAgent: '',
betaFeatures: '',
@@ -192,6 +196,48 @@ 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('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: [
src/server/routes/proxy/upstreamEndpoint.ts
+245 -22
View File
@@ -19,7 +19,7 @@ import {
import {
buildGeminiGenerateContentRequestFromOpenAi,
} from './geminiCliCompat.js';
export {
import {
buildMinimalJsonHeadersForCompatibility,
isEndpointDispatchDeniedError,
isEndpointDowngradeError,
@@ -27,10 +27,30 @@ 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;
wantsNativeResponsesReasoning: boolean;
};
type EndpointRuntimeState = {
preferredEndpoint: UpstreamEndpoint | null;
preferredUpdatedAtMs: number;
blockedUntilMsByEndpoint: Partial<Record<UpstreamEndpoint, number>>;
};
type ChannelContext = {
site: {
id: number;
@@ -45,6 +65,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';
}
@@ -577,6 +601,197 @@ 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
),
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.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 (isEndpointDowngradeError(status, errorText)) return true;
const text = (errorText || '').toLowerCase();
return (
text.includes('unsupported legacy protocol')
|| text.includes('please use /v1/')
|| text.includes('does not allow /v1/')
);
}
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 {
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,
@@ -651,29 +866,39 @@ export async function resolveUpstreamEndpointCandidates(
},
): 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 conversationFileSummary = requestCapabilities?.conversationFileSummary ?? {
hasImage: false,
hasAudio: false,
hasDocument: requestCapabilities?.hasNonImageFileInput === true,
hasDocument: hasNonImageFileInput,
hasRemoteDocumentUrl: false,
};
const hasNonImageFileInput = conversationFileSummary.hasDocument;
const wantsNativeResponsesReasoning = requestCapabilities?.wantsNativeResponsesReasoning === true;
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(
@@ -732,20 +957,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 : [];
@@ -765,7 +990,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>();
@@ -776,19 +1001,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);
}
}
@@ -1195,5 +1420,3 @@ export function buildClaudeCountTokensUpstreamRequest(input: {
},
};
}
src/server/routes/proxy/videos.ts
+3 -3
View File
@@ -100,7 +100,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,
@@ -153,11 +153,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;