Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: finish review-driven single-source cleanup

Browse Source
This commit is contained in:
Cita
2026-03-23 11:26:01 +08:00
parent 29c9fcbd29
commit 328b492ab3
21 changed files with 926 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/proxy-core/providers/headerUtils.test.ts
+9 -2
View File
@@ -25,7 +25,7 @@ describe('provider header utils', () => {
).toBe('beta-a,beta-b,beta-c');
expect(
mergeClaudeBetaHeader('custom-a,custom-b', 'beta-a,beta-b', ['beta-c']),
).toBe('custom-a,custom-b');
).toBe('beta-a,beta-b,custom-a,custom-b,beta-c');
});
it('preserves gemini cli runtime metadata when rebuilding user agents', async () => {
@@ -79,9 +79,12 @@ describe('provider header utils', () => {
const headers = buildClaudeRuntimeHeaders({
baseHeaders: {
'Content-Type': 'application/json',
authorization: 'Bearer stale-token',
},
claudeHeaders: {
'anthropic-beta': 'custom-beta',
'x-api-key': 'stale-api-key',
'user-agent': 'custom-agent',
},
anthropicVersion: '2023-06-01',
stream: true,
@@ -90,9 +93,13 @@ describe('provider header utils', () => {
});
expect(headers['anthropic-version']).toBe('2023-06-01');
expect(headers['anthropic-beta']).toBe('custom-beta');
expect(headers['anthropic-beta']).toContain('claude-code-20250219');
expect(headers['anthropic-beta']).toContain('custom-beta');
expect(headers.Authorization).toBe('Bearer oauth-token');
expect(headers.authorization).toBeUndefined();
expect(headers['x-api-key']).toBeUndefined();
expect(headers['user-agent']).toBeUndefined();
expect(headers['User-Agent']).toBe('custom-agent');
expect(headers.Accept).toBe('text/event-stream');
});
src/server/proxy-core/providers/headerUtils.ts
+37 -11
View File
@@ -32,6 +32,24 @@ export function getInputHeader(
return null;
}
function normalizeLowerCaseHeaderMap(
sources: Array<Record<string, unknown> | Record<string, string> | undefined>,
shouldSkip: (key: string) => boolean,
): Record<string, string> {
const normalized: Record<string, string> = {};
for (const source of sources) {
if (!source) continue;
for (const [key, value] of Object.entries(source)) {
const normalizedValue = headerValueToString(value);
if (!normalizedValue) continue;
const normalizedKey = key.toLowerCase();
if (shouldSkip(normalizedKey)) continue;
normalized[normalizedKey] = normalizedValue;
}
}
return normalized;
}
export function uuidFromSeed(seed: string): string {
const derived = pbkdf2Sync(seed, 'metapi-runtime-header-seed', 10_000, 16, 'sha256');
const bytes = new Uint8Array(derived);
@@ -52,17 +70,10 @@ export function mergeClaudeBetaHeader(
defaultValue: string,
extraBetas: string[] = [],
): string {
const source = explicitValue || defaultValue;
const seen = new Set<string>();
const merged: string[] = [];
for (const entry of source.split(',')) {
const normalized = entry.trim();
if (!normalized || seen.has(normalized)) continue;
seen.add(normalized);
merged.push(normalized);
}
if (!explicitValue) {
for (const entry of extraBetas) {
for (const source of [defaultValue, explicitValue ?? '', ...extraBetas]) {
for (const entry of source.split(',')) {
const normalized = entry.trim();
if (!normalized || seen.has(normalized)) continue;
seen.add(normalized);
@@ -187,9 +198,24 @@ export function buildClaudeRuntimeHeaders(input: {
input.defaultBetaHeader || CLAUDE_DEFAULT_BETA_HEADER,
input.extraBetas,
);
const passthroughHeaders = normalizeLowerCaseHeaderMap(
[input.baseHeaders, input.claudeHeaders],
(key) => (
key === 'accept'
|| key === 'accept-encoding'
|| key === 'anthropic-beta'
|| key === 'anthropic-dangerous-direct-browser-access'
|| key === 'anthropic-version'
|| key === 'authorization'
|| key === 'connection'
|| key === 'user-agent'
|| key === 'x-api-key'
|| key === 'x-app'
|| key.startsWith('x-stainless-')
),
);
const headers: Record<string, string> = {
...input.baseHeaders,
...input.claudeHeaders,
...passthroughHeaders,
'anthropic-version': input.anthropicVersion,
...(anthropicBeta ? { 'anthropic-beta': anthropicBeta } : {}),
'Anthropic-Dangerous-Direct-Browser-Access': 'true',
src/server/proxy-core/surfaces/chatSurface.ts
+45 -107
View File
@@ -2,7 +2,6 @@ import { TextDecoder } from 'node:util';
import type { FastifyRequest, FastifyReply } from 'fastify';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed } from '../../services/alertService.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { type DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
@@ -19,8 +18,6 @@ import {
} from '../../routes/proxy/downstreamPolicy.js';
import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js';
import { detectProxyFailure } from '../../routes/proxy/proxyFailureJudge.js';
import { buildUpstreamUrl } from '../../routes/proxy/upstreamUrl.js';
import { resolveProxyLogBilling } from '../../routes/proxy/proxyBilling.js';
import { openAiChatTransformer } from '../../transformers/openai/chat/index.js';
import { anthropicMessagesTransformer } from '../../transformers/anthropic/messages/index.js';
import { getProxyAuthContext, getProxyResourceOwner } from '../../middleware/auth.js';
@@ -32,7 +29,6 @@ import {
buildOauthProviderHeaders,
} from '../../services/oauth/service.js';
import { getOauthInfoFromAccount } from '../../services/oauth/oauthAccount.js';
import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js';
import {
collectResponsesFinalPayloadFromSse,
collectResponsesFinalPayloadFromSseText,
@@ -48,7 +44,9 @@ import { detectDownstreamClientContext } from '../../routes/proxy/downstreamClie
import {
createSurfaceFailureToolkit,
createSurfaceDispatchRequest,
recordSurfaceSuccess,
selectSurfaceChannelForAttempt,
trySurfaceOauthRefreshRecovery,
} from './sharedSurface.js';
const MAX_RETRIES = 2;
@@ -224,28 +222,15 @@ export async function handleChatSurfaceRequest(
});
const tryRecover = async (ctx: Parameters<NonNullable<typeof endpointStrategy.tryRecover>>[0]) => {
if ((ctx.response.status === 401 || ctx.response.status === 403) && oauth) {
try {
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
selected.tokenValue = refreshed.accessToken;
selected.account = {
...selected.account,
accessToken: refreshed.accessToken,
extraConfig: refreshed.extraConfig ?? selected.account.extraConfig,
};
const refreshedRequest = buildEndpointRequest(ctx.request.endpoint);
const refreshedTargetUrl = buildUpstreamUrl(selected.site.url, refreshedRequest.path);
const refreshedResponse = await dispatchRequest(refreshedRequest, refreshedTargetUrl);
if (refreshedResponse.ok) {
return {
upstream: refreshedResponse,
upstreamPath: refreshedRequest.path,
};
}
ctx.request = refreshedRequest;
ctx.response = refreshedResponse;
ctx.rawErrText = await refreshedResponse.text().catch(() => 'unknown error');
} catch {
return endpointStrategy.tryRecover(ctx);
const recovered = await trySurfaceOauthRefreshRecovery({
ctx,
selected,
siteUrl: selected.site.url,
buildRequest: (endpoint) => buildEndpointRequest(endpoint),
dispatchRequest,
});
if (recovered?.upstream?.ok) {
return recovered;
}
}
return endpointStrategy.tryRecover(ctx);
@@ -477,46 +462,19 @@ export async function handleChatSurfaceRequest(
}
const latency = Date.now() - startTime;
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
await recordSurfaceSuccess({
selected,
requestedModel,
modelName,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, modelName);
recordDownstreamCostUsage(request, estimatedCost);
await failureToolkit.log({
selected,
modelRequested: requestedModel,
status: 'success',
httpStatus: 200,
requestStartedAtMs: startTime,
latencyMs: latency,
errorMessage: null,
retryCount,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
upstreamPath: successfulUpstreamPath,
logSuccess: failureToolkit.log,
recordDownstreamCost: (estimatedCost) => {
recordDownstreamCostUsage(request, estimatedCost);
},
});
return;
}
@@ -570,46 +528,19 @@ export async function handleChatSurfaceRequest(
const normalizedFinal = downstreamTransformer.transformFinalResponse(upstreamData, modelName, rawText);
const downstreamResponse = downstreamTransformer.serializeFinalResponse(normalizedFinal, parsedUsage);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
await recordSurfaceSuccess({
selected,
requestedModel,
modelName,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, modelName);
recordDownstreamCostUsage(request, estimatedCost);
await failureToolkit.log({
selected,
modelRequested: requestedModel,
status: 'success',
httpStatus: 200,
requestStartedAtMs: startTime,
latencyMs: latency,
errorMessage: null,
retryCount,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
upstreamPath: successfulUpstreamPath,
logSuccess: failureToolkit.log,
recordDownstreamCost: (estimatedCost) => {
recordDownstreamCostUsage(request, estimatedCost);
},
});
return reply.send(downstreamResponse);
@@ -768,18 +699,25 @@ export async function handleClaudeCountTokensSurfaceRequest(
let upstream = await dispatchRequest(upstreamRequest);
if ((upstream.status === 401 || upstream.status === 403) && oauth) {
try {
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
selected.tokenValue = refreshed.accessToken;
selected.account = {
...selected.account,
accessToken: refreshed.accessToken,
extraConfig: refreshed.extraConfig ?? selected.account.extraConfig,
};
const recoverContext = {
request: upstreamRequest,
response: upstream,
rawErrText: '',
};
const recovered = await trySurfaceOauthRefreshRecovery({
ctx: recoverContext,
selected,
siteUrl: selected.site.url,
buildRequest: () => buildRequest(),
dispatchRequest,
captureFailureBody: false,
});
if (recovered?.upstream?.ok) {
upstreamRequest = buildRequest();
upstream = await dispatchRequest(upstreamRequest);
} catch {
// Fall through to the regular upstream error handling below.
upstream = recovered.upstream;
} else {
upstreamRequest = recoverContext.request;
upstream = recoverContext.response;
}
}
src/server/proxy-core/surfaces/openAiResponsesSurface.ts
+38 -119
View File
@@ -1,8 +1,6 @@
import { TextDecoder } from 'node:util';
import type { FastifyReply, FastifyRequest } from 'fastify';
import { tokenRouter } from '../../services/tokenRouter.js';
import { reportProxyAllFailed } from '../../services/alertService.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import {
@@ -14,8 +12,6 @@ import {
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from '../../routes/proxy/downstreamPolicy.js';
import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js';
import { detectProxyFailure } from '../../routes/proxy/proxyFailureJudge.js';
import { buildUpstreamUrl } from '../../routes/proxy/upstreamUrl.js';
import { resolveProxyLogBilling } from '../../routes/proxy/proxyBilling.js';
import { getProxyAuthContext, getProxyResourceOwner } from '../../middleware/auth.js';
import { normalizeInputFileBlock } from '../../transformers/shared/inputFile.js';
import {
@@ -26,7 +22,6 @@ import {
buildOauthProviderHeaders,
} from '../../services/oauth/service.js';
import { getOauthInfoFromAccount } from '../../services/oauth/oauthAccount.js';
import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js';
import {
collectResponsesFinalPayloadFromSse,
collectResponsesFinalPayloadFromSseText,
@@ -46,7 +41,9 @@ import { detectDownstreamClientContext } from '../../routes/proxy/downstreamClie
import {
createSurfaceFailureToolkit,
createSurfaceDispatchRequest,
recordSurfaceSuccess,
selectSurfaceChannelForAttempt,
trySurfaceOauthRefreshRecovery,
} from './sharedSurface.js';
const MAX_RETRIES = 2;
@@ -325,28 +322,15 @@ export async function handleOpenAiResponsesSurfaceRequest(
response: ctx.response,
rawErrText: ctx.rawErrText || '',
})) {
try {
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
selected.tokenValue = refreshed.accessToken;
selected.account = {
...selected.account,
accessToken: refreshed.accessToken,
extraConfig: refreshed.extraConfig ?? selected.account.extraConfig,
};
const refreshedRequest = buildEndpointRequest(ctx.request.endpoint);
const refreshedTargetUrl = buildUpstreamUrl(selected.site.url, refreshedRequest.path);
const refreshedResponse = await dispatchRequest(refreshedRequest, refreshedTargetUrl);
if (refreshedResponse.ok) {
return {
upstream: refreshedResponse,
upstreamPath: refreshedRequest.path,
};
}
ctx.request = refreshedRequest;
ctx.response = refreshedResponse;
ctx.rawErrText = await refreshedResponse.text().catch(() => 'unknown error');
} catch {
return endpointStrategy.tryRecover(ctx);
const recovered = await trySurfaceOauthRefreshRecovery({
ctx,
selected,
siteUrl: selected.site.url,
buildRequest: (endpoint) => buildEndpointRequest(endpoint),
dispatchRequest,
});
if (recovered?.upstream?.ok) {
return recovered;
}
}
return endpointStrategy.tryRecover(ctx);
@@ -410,65 +394,23 @@ export async function handleOpenAiResponsesSurfaceRequest(
const upstream = endpointResult.upstream;
const successfulUpstreamPath = endpointResult.upstreamPath;
const finalizeStreamSuccess = async (parsedUsage: UsageSummary, latency: number) => {
let usageForLog = {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
};
let estimatedCost = 0;
let billingDetails: unknown = null;
try {
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
usageForLog = {
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
};
const billing = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
estimatedCost = billing.estimatedCost;
billingDetails = billing.billingDetails;
} catch (error) {
console.error('[responses] post-stream bookkeeping failed:', error);
}
try {
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, modelName);
recordDownstreamCostUsage(request, estimatedCost);
await failureToolkit.log({
await recordSurfaceSuccess({
selected,
modelRequested: requestedModel,
status: 'success',
httpStatus: 200,
requestedModel,
modelName,
parsedUsage,
requestStartedAtMs: startTime,
latencyMs: latency,
errorMessage: null,
retryCount,
promptTokens: usageForLog.promptTokens,
completionTokens: usageForLog.completionTokens,
totalTokens: usageForLog.totalTokens,
estimatedCost,
billingDetails,
upstreamPath: successfulUpstreamPath,
logSuccess: failureToolkit.log,
recordDownstreamCost: (estimatedCost) => {
recordDownstreamCostUsage(request, estimatedCost);
},
bestEffortMetrics: {
errorLabel: '[responses] post-stream bookkeeping failed:',
},
});
} catch (error) {
console.error('[responses] post-stream success logging failed:', error);
@@ -716,48 +658,25 @@ export async function handleOpenAiResponsesSurfaceRequest(
serializationMode: isCompactRequest ? 'compact' : 'response',
});
try {
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
await recordSurfaceSuccess({
selected,
requestedModel,
modelName,
parsedUsage,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
latencyMs: latency,
retryCount,
upstreamPath: successfulUpstreamPath,
logSuccess: failureToolkit.log,
recordDownstreamCost: (estimatedCost) => {
recordDownstreamCostUsage(request, estimatedCost);
},
bestEffortMetrics: {
errorLabel: '[responses] post-response bookkeeping failed:',
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost, modelName);
recordDownstreamCostUsage(request, estimatedCost);
await failureToolkit.log({
selected,
modelRequested: requestedModel,
status: 'success',
httpStatus: 200,
latencyMs: latency,
errorMessage: null,
retryCount,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
upstreamPath: successfulUpstreamPath,
});
} catch (error) {
console.error('[responses] post-response bookkeeping failed:', error);
console.error('[responses] post-response success logging failed:', error);
}
return reply.send(downstreamData);
} catch (err: any) {
src/server/proxy-core/surfaces/sharedSurface.test.ts
+357 -1
View File
@@ -1,4 +1,4 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest';
import { EMPTY_DOWNSTREAM_ROUTING_POLICY } from '../../services/downstreamPolicyTypes.js';
const selectChannelMock = vi.fn();
@@ -16,12 +16,18 @@ const reportTokenExpiredMock = vi.fn();
const isTokenExpiredErrorMock = vi.fn();
const shouldRetryProxyRequestMock = vi.fn();
const recordOauthQuotaResetHintMock = vi.fn();
const recordSuccessMock = vi.fn();
const resolveProxyUsageWithSelfLogFallbackMock = vi.fn();
const resolveProxyLogBillingMock = vi.fn();
const refreshOauthAccessTokenSingleflightMock = vi.fn();
const consoleWarnMock = vi.spyOn(console, 'warn').mockImplementation(() => {});
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
},
}));
@@ -67,7 +73,23 @@ vi.mock('../../services/oauth/quota.js', () => ({
recordOauthQuotaResetHint: (...args: unknown[]) => recordOauthQuotaResetHintMock(...args),
}));
vi.mock('../../services/proxyUsageFallbackService.js', () => ({
resolveProxyUsageWithSelfLogFallback: (...args: unknown[]) => resolveProxyUsageWithSelfLogFallbackMock(...args),
}));
vi.mock('../../routes/proxy/proxyBilling.js', () => ({
resolveProxyLogBilling: (...args: unknown[]) => resolveProxyLogBillingMock(...args),
}));
vi.mock('../../services/oauth/refreshSingleflight.js', () => ({
refreshOauthAccessTokenSingleflight: (...args: unknown[]) => refreshOauthAccessTokenSingleflightMock(...args),
}));
describe('selectSurfaceChannelForAttempt', () => {
afterAll(() => {
consoleWarnMock.mockRestore();
});
beforeEach(() => {
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
@@ -84,6 +106,11 @@ describe('selectSurfaceChannelForAttempt', () => {
isTokenExpiredErrorMock.mockReset();
shouldRetryProxyRequestMock.mockReset();
recordOauthQuotaResetHintMock.mockReset();
recordSuccessMock.mockReset();
resolveProxyUsageWithSelfLogFallbackMock.mockReset();
resolveProxyLogBillingMock.mockReset();
refreshOauthAccessTokenSingleflightMock.mockReset();
consoleWarnMock.mockClear();
});
it('refreshes models and retries selectChannel on the first attempt when no channel is available', async () => {
@@ -308,6 +335,40 @@ describe('selectSurfaceChannelForAttempt', () => {
}));
});
it('keeps retryable failures on the retry path even when quota hint recording fails', async () => {
composeProxyLogMessageMock.mockReturnValue('normalized error');
formatUtcSqlDateTimeMock.mockReturnValue('2026-03-21 22:00:00');
insertProxyLogMock.mockResolvedValue(undefined);
shouldRetryProxyRequestMock.mockReturnValue(true);
isTokenExpiredErrorMock.mockReturnValue(false);
recordOauthQuotaResetHintMock.mockRejectedValue(new Error('hint failed'));
const { createSurfaceFailureToolkit } = await import('./sharedSurface.js');
const toolkit = createSurfaceFailureToolkit({
warningScope: 'chat',
downstreamPath: '/v1/chat/completions',
maxRetries: 2,
clientContext: null,
downstreamApiKeyId: null,
});
await expect(toolkit.handleUpstreamFailure({
selected: {
channel: { id: 11, routeId: 22 },
account: { id: 33, username: 'oauth-user' },
site: { name: 'Codex OAuth' },
actualModel: 'upstream-model',
},
requestedModel: 'gpt-5.2',
modelName: 'upstream-model',
status: 429,
errText: 'quota exceeded',
rawErrText: '{"error":"quota exceeded"}',
latencyMs: 1200,
retryCount: 0,
})).resolves.toEqual({ action: 'retry' });
});
it('returns a terminal upstream error response and reports token expiration when retries stop', async () => {
composeProxyLogMessageMock.mockReturnValue('normalized error');
formatUtcSqlDateTimeMock.mockReturnValue('2026-03-21 22:00:00');
@@ -363,6 +424,50 @@ describe('selectSurfaceChannelForAttempt', () => {
});
});
it('returns terminal failures even when final alerting throws', async () => {
composeProxyLogMessageMock.mockReturnValue('normalized error');
formatUtcSqlDateTimeMock.mockReturnValue('2026-03-21 22:00:00');
insertProxyLogMock.mockResolvedValue(undefined);
shouldRetryProxyRequestMock.mockReturnValue(false);
isTokenExpiredErrorMock.mockReturnValue(true);
recordOauthQuotaResetHintMock.mockResolvedValue(null);
reportTokenExpiredMock.mockRejectedValue(new Error('token alert failed'));
const { createSurfaceFailureToolkit } = await import('./sharedSurface.js');
const toolkit = createSurfaceFailureToolkit({
warningScope: 'responses',
downstreamPath: '/v1/responses',
maxRetries: 2,
clientContext: null,
downstreamApiKeyId: null,
});
await expect(toolkit.handleUpstreamFailure({
selected: {
channel: { id: 11, routeId: 22 },
account: { id: 33, username: 'oauth-user' },
site: { name: 'Codex OAuth' },
actualModel: 'upstream-model',
},
requestedModel: 'gpt-5.2',
modelName: 'upstream-model',
status: 401,
errText: 'expired token',
rawErrText: 'expired token',
latencyMs: 900,
retryCount: 2,
})).resolves.toEqual({
action: 'respond',
status: 401,
payload: {
error: {
message: 'expired token',
type: 'upstream_error',
},
},
});
});
it('handles detected proxy failures through the shared failure toolkit', async () => {
composeProxyLogMessageMock.mockReturnValue('normalized error');
formatUtcSqlDateTimeMock.mockReturnValue('2026-03-21 22:00:00');
@@ -468,4 +573,255 @@ describe('selectSurfaceChannelForAttempt', () => {
reason: 'socket hang up',
});
});
it('records stream failures with error text even without a runtime status code', async () => {
composeProxyLogMessageMock.mockReturnValue('normalized error');
formatUtcSqlDateTimeMock.mockReturnValue('2026-03-21 22:00:00');
insertProxyLogMock.mockResolvedValue(undefined);
const { createSurfaceFailureToolkit } = await import('./sharedSurface.js');
const toolkit = createSurfaceFailureToolkit({
warningScope: 'responses',
downstreamPath: '/v1/responses',
maxRetries: 2,
clientContext: null,
downstreamApiKeyId: null,
});
await toolkit.recordStreamFailure({
selected: {
channel: { id: 11, routeId: 22 },
account: { id: 33, username: 'oauth-user' },
site: { name: 'Codex OAuth' },
actualModel: 'upstream-model',
},
requestedModel: 'gpt-5.2',
modelName: 'upstream-model',
errorMessage: 'stream exploded',
latencyMs: 450,
retryCount: 1,
});
expect(recordFailureMock).toHaveBeenCalledWith(11, {
errorText: 'stream exploded',
modelName: 'upstream-model',
});
});
it('refreshes oauth tokens through the shared recover helper and retries the rebuilt request', async () => {
const refreshedResponse = {
ok: true,
status: 200,
text: vi.fn(),
};
const selected = {
account: {
id: 33,
accessToken: 'old-access-token',
extraConfig: '{"oauth":{"refreshToken":"refresh"}}',
},
tokenValue: 'old-access-token',
};
const ctx = {
request: {
endpoint: 'responses' as const,
path: '/v1/responses',
headers: { authorization: 'Bearer old-access-token' },
body: { model: 'gpt-5.2' },
},
response: {
ok: false,
status: 401,
text: vi.fn(),
},
rawErrText: 'expired token',
};
refreshOauthAccessTokenSingleflightMock.mockResolvedValue({
accessToken: 'new-access-token',
extraConfig: '{"oauth":{"refreshToken":"refresh-next"}}',
});
const dispatchRequest = vi.fn().mockResolvedValue(refreshedResponse);
const { trySurfaceOauthRefreshRecovery } = await import('./sharedSurface.js');
const result = await trySurfaceOauthRefreshRecovery({
ctx,
selected,
siteUrl: 'https://upstream.example.com',
buildRequest: () => ({
endpoint: 'responses',
path: '/v1/responses',
headers: { authorization: `Bearer ${selected.tokenValue}` },
body: { model: 'gpt-5.2' },
}),
dispatchRequest,
});
expect(refreshOauthAccessTokenSingleflightMock).toHaveBeenCalledWith(33);
expect(selected.tokenValue).toBe('new-access-token');
expect(selected.account.accessToken).toBe('new-access-token');
expect(selected.account.extraConfig).toBe('{"oauth":{"refreshToken":"refresh-next"}}');
expect(dispatchRequest).toHaveBeenCalledWith(expect.objectContaining({
headers: { authorization: 'Bearer new-access-token' },
}), 'https://upstream.example.com/v1/responses');
expect(result).toEqual({
upstream: refreshedResponse,
upstreamPath: '/v1/responses',
});
});
it('updates the recover context with the refreshed failure response when oauth refresh retry still fails', async () => {
const refreshedResponse = {
ok: false,
status: 403,
text: vi.fn().mockResolvedValue('account mismatch'),
};
const ctx = {
request: {
endpoint: 'responses' as const,
path: '/v1/responses',
headers: { authorization: 'Bearer old-access-token' },
body: { model: 'gpt-5.2' },
},
response: {
ok: false,
status: 401,
text: vi.fn(),
},
rawErrText: 'expired token',
};
const selected = {
account: {
id: 33,
accessToken: 'old-access-token',
extraConfig: '{"oauth":{"refreshToken":"refresh"}}',
},
tokenValue: 'old-access-token',
};
refreshOauthAccessTokenSingleflightMock.mockResolvedValue({
accessToken: 'new-access-token',
extraConfig: '{"oauth":{"refreshToken":"refresh-next"}}',
});
const { trySurfaceOauthRefreshRecovery } = await import('./sharedSurface.js');
const result = await trySurfaceOauthRefreshRecovery({
ctx,
selected,
siteUrl: 'https://upstream.example.com',
buildRequest: () => ({
endpoint: 'responses',
path: '/v1/responses',
headers: { authorization: `Bearer ${selected.tokenValue}` },
body: { model: 'gpt-5.2' },
}),
dispatchRequest: vi.fn().mockResolvedValue(refreshedResponse),
});
expect(result).toBeNull();
expect(ctx.request.headers).toEqual({ authorization: 'Bearer new-access-token' });
expect(ctx.response).toBe(refreshedResponse);
expect(ctx.rawErrText).toBe('account mismatch');
});
it('records shared success bookkeeping with usage fallback, billing, and success logging', async () => {
resolveProxyUsageWithSelfLogFallbackMock.mockResolvedValue({
promptTokens: 20,
completionTokens: 8,
totalTokens: 28,
});
resolveProxyLogBillingMock.mockResolvedValue({
estimatedCost: 0.42,
billingDetails: { source: 'pricing-test' },
});
const logSuccess = vi.fn().mockResolvedValue(undefined);
const recordDownstreamCost = vi.fn();
const { recordSurfaceSuccess } = await import('./sharedSurface.js');
const result = await recordSurfaceSuccess({
selected: {
channel: { id: 11, routeId: 22 },
account: { id: 33, username: 'oauth-user' },
site: { id: 44, url: 'https://upstream.example.com', name: 'Codex OAuth' },
tokenValue: 'live-token',
tokenName: 'default',
actualModel: 'upstream-model',
},
requestedModel: 'gpt-5.2',
modelName: 'upstream-model',
parsedUsage: {
promptTokens: 10,
completionTokens: 5,
totalTokens: 15,
},
requestStartedAtMs: 1000,
latencyMs: 250,
retryCount: 1,
upstreamPath: '/v1/responses',
logSuccess,
recordDownstreamCost,
});
expect(resolveProxyUsageWithSelfLogFallbackMock).toHaveBeenCalledWith({
site: { id: 44, url: 'https://upstream.example.com', name: 'Codex OAuth' },
account: { id: 33, username: 'oauth-user' },
tokenValue: 'live-token',
tokenName: 'default',
modelName: 'upstream-model',
requestStartedAtMs: 1000,
requestEndedAtMs: 1250,
localLatencyMs: 250,
usage: {
promptTokens: 10,
completionTokens: 5,
totalTokens: 15,
},
});
expect(resolveProxyLogBillingMock).toHaveBeenCalledWith({
site: { id: 44, url: 'https://upstream.example.com', name: 'Codex OAuth' },
account: { id: 33, username: 'oauth-user' },
modelName: 'upstream-model',
parsedUsage: {
promptTokens: 10,
completionTokens: 5,
totalTokens: 15,
},
resolvedUsage: {
promptTokens: 20,
completionTokens: 8,
totalTokens: 28,
},
});
expect(recordSuccessMock).toHaveBeenCalledWith(11, 250, 0.42, 'upstream-model');
expect(recordDownstreamCost).toHaveBeenCalledWith(0.42);
expect(logSuccess).toHaveBeenCalledWith({
selected: {
channel: { id: 11, routeId: 22 },
account: { id: 33, username: 'oauth-user' },
site: { id: 44, url: 'https://upstream.example.com', name: 'Codex OAuth' },
tokenValue: 'live-token',
tokenName: 'default',
actualModel: 'upstream-model',
},
modelRequested: 'gpt-5.2',
status: 'success',
httpStatus: 200,
latencyMs: 250,
errorMessage: null,
retryCount: 1,
promptTokens: 20,
completionTokens: 8,
totalTokens: 28,
estimatedCost: 0.42,
billingDetails: { source: 'pricing-test' },
upstreamPath: '/v1/responses',
});
expect(result).toEqual({
resolvedUsage: {
promptTokens: 20,
completionTokens: 8,
totalTokens: 28,
},
estimatedCost: 0.42,
billingDetails: { source: 'pricing-test' },
});
});
});
src/server/proxy-core/surfaces/sharedSurface.ts
+198 -11
View File
@@ -3,16 +3,20 @@ import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import type { SiteProxyConfigLike } from '../../services/siteProxy.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import type { DownstreamRoutingPolicy } from '../../services/downstreamPolicyTypes.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { composeProxyLogMessage } from '../../routes/proxy/logPathMeta.js';
import { resolveProxyLogBilling } from '../../routes/proxy/proxyBilling.js';
import type { DownstreamClientContext } from '../../routes/proxy/downstreamClientContext.js';
import { insertProxyLog } from '../../services/proxyLogStore.js';
import { dispatchRuntimeRequest } from '../../routes/proxy/runtimeExecutor.js';
import type { BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js';
import { buildUpstreamUrl } from '../../routes/proxy/upstreamUrl.js';
import { recordOauthQuotaResetHint } from '../../services/oauth/quota.js';
import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js';
type SelectedChannel = Awaited<ReturnType<typeof tokenRouter.selectChannel>>;
type SurfaceWarningScope = 'chat' | 'responses';
@@ -39,6 +43,34 @@ type SurfaceFailureOutcome =
| { action: 'retry' }
| SurfaceFailureResponse;
type SurfaceOauthRefreshSelectedChannel = {
account: {
id: number;
accessToken?: string | null;
extraConfig?: string | null;
};
tokenValue: string;
};
type SurfaceOauthRefreshContext<TRequest extends BuiltEndpointRequest> = {
request: TRequest;
response: Awaited<ReturnType<typeof dispatchRuntimeRequest>>;
rawErrText: string;
};
type SurfaceSuccessSelectedChannel = SurfaceSelectedChannel & {
account: Record<string, unknown> & { id: number };
site: Record<string, unknown>;
tokenValue: string;
tokenName?: string | null;
};
type SurfaceUsageSummary = {
promptTokens: number;
completionTokens: number;
totalTokens: number;
};
export async function selectSurfaceChannelForAttempt(input: {
requestedModel: string;
downstreamPolicy: DownstreamRoutingPolicy;
@@ -146,6 +178,150 @@ export function createSurfaceDispatchRequest(input: {
);
}
export async function trySurfaceOauthRefreshRecovery<TRequest extends BuiltEndpointRequest>(input: {
ctx: SurfaceOauthRefreshContext<TRequest>;
selected: SurfaceOauthRefreshSelectedChannel;
siteUrl: string;
buildRequest: (endpoint: TRequest['endpoint']) => TRequest;
dispatchRequest: (
request: TRequest,
targetUrl: string,
) => Promise<Awaited<ReturnType<typeof dispatchRuntimeRequest>>>;
captureFailureBody?: boolean;
}): Promise<{
upstream: Awaited<ReturnType<typeof dispatchRuntimeRequest>>;
upstreamPath: string;
} | null> {
try {
const refreshed = await refreshOauthAccessTokenSingleflight(input.selected.account.id);
input.selected.tokenValue = refreshed.accessToken;
input.selected.account = {
...input.selected.account,
accessToken: refreshed.accessToken,
extraConfig: refreshed.extraConfig ?? input.selected.account.extraConfig,
};
const refreshedRequest = input.buildRequest(input.ctx.request.endpoint);
const refreshedTargetUrl = buildUpstreamUrl(input.siteUrl, refreshedRequest.path);
const refreshedResponse = await input.dispatchRequest(refreshedRequest, refreshedTargetUrl);
if (refreshedResponse.ok) {
return {
upstream: refreshedResponse,
upstreamPath: refreshedRequest.path,
};
}
input.ctx.request = refreshedRequest;
input.ctx.response = refreshedResponse;
if (input.captureFailureBody !== false) {
input.ctx.rawErrText = await refreshedResponse.text().catch(() => 'unknown error');
}
} catch {
return null;
}
return null;
}
export async function recordSurfaceSuccess(input: {
selected: SurfaceSuccessSelectedChannel;
requestedModel: string;
modelName: string;
parsedUsage: SurfaceUsageSummary;
requestStartedAtMs: number;
latencyMs: number;
retryCount: number;
upstreamPath?: string | null;
logSuccess: (args: {
selected: SurfaceSelectedChannel;
modelRequested: string;
status: string;
httpStatus: number;
latencyMs: number;
errorMessage: string | null;
retryCount: number;
promptTokens?: number;
completionTokens?: number;
totalTokens?: number;
estimatedCost?: number;
billingDetails?: unknown;
upstreamPath?: string | null;
}) => Promise<void>;
recordDownstreamCost?: (estimatedCost: number) => void;
bestEffortMetrics?: {
errorLabel: string;
};
}) {
let resolvedUsage = {
promptTokens: input.parsedUsage.promptTokens,
completionTokens: input.parsedUsage.completionTokens,
totalTokens: input.parsedUsage.totalTokens,
};
let estimatedCost = 0;
let billingDetails: unknown = null;
try {
resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: input.selected.site,
account: input.selected.account,
tokenValue: input.selected.tokenValue,
tokenName: input.selected.tokenName,
modelName: input.modelName,
requestStartedAtMs: input.requestStartedAtMs,
requestEndedAtMs: input.requestStartedAtMs + input.latencyMs,
localLatencyMs: input.latencyMs,
usage: {
promptTokens: input.parsedUsage.promptTokens,
completionTokens: input.parsedUsage.completionTokens,
totalTokens: input.parsedUsage.totalTokens,
},
});
const billing = await resolveProxyLogBilling({
site: input.selected.site,
account: input.selected.account,
modelName: input.modelName,
parsedUsage: input.parsedUsage,
resolvedUsage,
});
estimatedCost = billing.estimatedCost;
billingDetails = billing.billingDetails;
} catch (error) {
if (!input.bestEffortMetrics) {
throw error;
}
console.error(input.bestEffortMetrics.errorLabel, error);
}
tokenRouter.recordSuccess(
input.selected.channel.id,
input.latencyMs,
estimatedCost,
input.modelName,
);
input.recordDownstreamCost?.(estimatedCost);
await input.logSuccess({
selected: input.selected,
modelRequested: input.requestedModel,
status: 'success',
httpStatus: 200,
latencyMs: input.latencyMs,
errorMessage: null,
retryCount: input.retryCount,
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
estimatedCost,
billingDetails,
upstreamPath: input.upstreamPath,
});
return {
resolvedUsage,
estimatedCost,
billingDetails,
};
}
export function createSurfaceFailureToolkit(input: {
warningScope: SurfaceWarningScope;
downstreamPath: string;
@@ -193,6 +369,14 @@ export function createSurfaceFailureToolkit(input: {
? { action: 'retry' as const }
: null;
const runBestEffort = (label: string, fn: () => Promise<unknown>) => {
void Promise.resolve()
.then(fn)
.catch((error) => {
console.warn(`[proxy/${input.warningScope}] failed to ${label}`, error);
});
};
return {
log,
async handleUpstreamFailure(args: {
@@ -220,19 +404,19 @@ export function createSurfaceFailureToolkit(input: {
errorMessage: args.errText,
retryCount: args.retryCount,
});
await recordOauthQuotaResetHint({
runBestEffort('record oauth quota reset hint', () => recordOauthQuotaResetHint({
accountId: args.selected.account.id,
statusCode: args.status,
errorText: rawErrText,
});
}));
if (isTokenExpiredError({ status: args.status, message: args.errText })) {
await reportTokenExpired({
runBestEffort('report token expired', () => reportTokenExpired({
accountId: args.selected.account.id,
username: args.selected.account.username,
siteName: args.selected.site.name,
detail: `HTTP ${args.status}`,
});
}));
}
if (shouldRetryProxyRequest(args.status, args.errText)) {
@@ -240,10 +424,10 @@ export function createSurfaceFailureToolkit(input: {
if (retry) return retry;
}
await reportProxyAllFailed({
runBestEffort('report proxy all failed', () => reportProxyAllFailed({
model: args.requestedModel,
reason: `upstream returned HTTP ${args.status}`,
});
}));
return {
action: 'respond',
@@ -293,10 +477,10 @@ export function createSurfaceFailureToolkit(input: {
if (retry) return retry;
}
await reportProxyAllFailed({
runBestEffort('report proxy all failed', () => reportProxyAllFailed({
model: args.requestedModel,
reason: args.failure.reason,
});
}));
return {
action: 'respond',
@@ -335,10 +519,10 @@ export function createSurfaceFailureToolkit(input: {
const retry = maybeRetry(args.retryCount);
if (retry) return retry;
await reportProxyAllFailed({
runBestEffort('report proxy all failed', () => reportProxyAllFailed({
model: args.requestedModel,
reason: args.errorMessage || 'network failure',
});
}));
return {
action: 'respond',
@@ -374,7 +558,10 @@ export function createSurfaceFailureToolkit(input: {
modelName: args.modelName,
});
} else {
await tokenRouter.recordFailure(args.selected.channel.id, args.modelName);
await tokenRouter.recordFailure(args.selected.channel.id, {
errorText: errorMessage,
modelName: args.modelName,
});
}
await log({
selected: args.selected,