From 328b492ab3eb4e22a21bf87a5e1cfa5d7589b556 Mon Sep 17 00:00:00 2001 From: Cita Date: Mon, 23 Mar 2026 11:26:01 +0800 Subject: [PATCH] fix: finish review-driven single-source cleanup --- .../proxy-core/providers/headerUtils.test.ts | 11 +- .../proxy-core/providers/headerUtils.ts | 48 ++- src/server/proxy-core/surfaces/chatSurface.ts | 152 +++----- .../surfaces/openAiResponsesSurface.ts | 157 ++------ .../proxy-core/surfaces/sharedSurface.test.ts | 358 +++++++++++++++++- .../proxy-core/surfaces/sharedSurface.ts | 209 +++++++++- .../api/accountTokens.architecture.test.ts | 13 + src/server/routes/api/accountTokens.ts | 1 - src/server/routes/api/accounts.ts | 19 +- .../proxy/architecture-boundaries.test.ts | 23 ++ src/server/routes/proxy/upstreamEndpoint.ts | 43 +-- .../services/accountMutationWorkflow.test.ts | 35 ++ .../services/accountMutationWorkflow.ts | 20 +- .../platforms/standardApiProvider.test.ts | 38 ++ .../services/platforms/standardApiProvider.ts | 23 +- .../openai/responses/conversion.test.ts | 41 ++ .../openai/responses/normalization.ts | 9 +- src/shared/conversationFileTypes.js | 14 +- src/shared/conversationFileTypes.test.ts | 10 + src/shared/tokenRoutePatterns.js | 10 +- src/shared/tokenRoutePatterns.test.ts | 12 + 21 files changed, 926 insertions(+), 320 deletions(-) create mode 100644 src/server/routes/api/accountTokens.architecture.test.ts diff --git a/src/server/proxy-core/providers/headerUtils.test.ts b/src/server/proxy-core/providers/headerUtils.test.ts index a762d3a..b1c4398 100644 --- a/src/server/proxy-core/providers/headerUtils.test.ts +++ b/src/server/proxy-core/providers/headerUtils.test.ts @@ -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'); }); diff --git a/src/server/proxy-core/providers/headerUtils.ts b/src/server/proxy-core/providers/headerUtils.ts index f31600e..5846694 100644 --- a/src/server/proxy-core/providers/headerUtils.ts +++ b/src/server/proxy-core/providers/headerUtils.ts @@ -32,6 +32,24 @@ export function getInputHeader( return null; } +function normalizeLowerCaseHeaderMap( + sources: Array | Record | undefined>, + shouldSkip: (key: string) => boolean, +): Record { + const normalized: Record = {}; + 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(); 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 = { - ...input.baseHeaders, - ...input.claudeHeaders, + ...passthroughHeaders, 'anthropic-version': input.anthropicVersion, ...(anthropicBeta ? { 'anthropic-beta': anthropicBeta } : {}), 'Anthropic-Dangerous-Direct-Browser-Access': 'true', diff --git a/src/server/proxy-core/surfaces/chatSurface.ts b/src/server/proxy-core/surfaces/chatSurface.ts index 2e0d0e3..37e93c5 100644 --- a/src/server/proxy-core/surfaces/chatSurface.ts +++ b/src/server/proxy-core/surfaces/chatSurface.ts @@ -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>[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; } } diff --git a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts index 49e7eb2..eff8055 100644 --- a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts +++ b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts @@ -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) { diff --git a/src/server/proxy-core/surfaces/sharedSurface.test.ts b/src/server/proxy-core/surfaces/sharedSurface.test.ts index 8a0ce8c..78a120c 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.test.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.test.ts @@ -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' }, + }); + }); }); diff --git a/src/server/proxy-core/surfaces/sharedSurface.ts b/src/server/proxy-core/surfaces/sharedSurface.ts index 8e4c559..e2a41f7 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.ts @@ -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>; 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 = { + request: TRequest; + response: Awaited>; + rawErrText: string; +}; + +type SurfaceSuccessSelectedChannel = SurfaceSelectedChannel & { + account: Record & { id: number }; + site: Record; + 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(input: { + ctx: SurfaceOauthRefreshContext; + selected: SurfaceOauthRefreshSelectedChannel; + siteUrl: string; + buildRequest: (endpoint: TRequest['endpoint']) => TRequest; + dispatchRequest: ( + request: TRequest, + targetUrl: string, + ) => Promise>>; + captureFailureBody?: boolean; +}): Promise<{ + upstream: Awaited>; + 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; + 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) => { + 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, diff --git a/src/server/routes/api/accountTokens.architecture.test.ts b/src/server/routes/api/accountTokens.architecture.test.ts new file mode 100644 index 0000000..9cdd600 --- /dev/null +++ b/src/server/routes/api/accountTokens.architecture.test.ts @@ -0,0 +1,13 @@ +import { readFileSync } from 'node:fs'; +import { describe, expect, it } from 'vitest'; + +function readSource(relativePath: string): string { + return readFileSync(new URL(relativePath, import.meta.url), 'utf8'); +} + +describe('account token route architecture boundaries', () => { + it('keeps legacy token repair from forcing route rebuilds inline', () => { + const source = readSource('./accountTokens.ts'); + expect(source).not.toContain('rebuildRoutes: true'); + }); +}); diff --git a/src/server/routes/api/accountTokens.ts b/src/server/routes/api/accountTokens.ts index 1f981fd..ebffde4 100644 --- a/src/server/routes/api/accountTokens.ts +++ b/src/server/routes/api/accountTokens.ts @@ -233,7 +233,6 @@ async function executeAccountTokenSync(row: AccountWithSiteRow): Promise('/api/accounts/:id', async (request) => { const id = parseInt(request.params.id); await db.delete(schema.accounts).where(eq(schema.accounts.id, id)).run(); - try { - await rebuildTokenRoutesFromAvailability(); - } catch { } + await rebuildRoutesBestEffort(); return { success: true }; }); @@ -1386,9 +1386,7 @@ export async function accountsRoutes(app: FastifyInstance) { } if (shouldRebuildRoutes) { - try { - await rebuildTokenRoutesFromAvailability(); - } catch { } + await rebuildRoutesBestEffort(); } return { @@ -1596,10 +1594,7 @@ export async function accountsRoutes(app: FastifyInstance) { } } }); - - try { - await rebuildTokenRoutesFromAvailability(); - } catch { } + await rebuildRoutesBestEffort(); return { success: true }; } catch (err: any) { diff --git a/src/server/routes/proxy/architecture-boundaries.test.ts b/src/server/routes/proxy/architecture-boundaries.test.ts index c60c140..28f5e20 100644 --- a/src/server/routes/proxy/architecture-boundaries.test.ts +++ b/src/server/routes/proxy/architecture-boundaries.test.ts @@ -92,6 +92,12 @@ describe('proxy route architecture boundaries', () => { expect(source).not.toContain('normalizeContentText('); }); + it('keeps codex runtime header and prompt-cache derivation inside provider profiles', () => { + const source = readSource('./upstreamEndpoint.ts'); + expect(source).not.toContain('buildCodexRuntimeHeaders('); + expect(source).not.toContain('shouldInjectDerivedPromptCacheKey'); + }); + it('keeps gemini runtime closure in transformer-owned helpers', () => { const source = readSource('./gemini.ts'); const surfaceSource = readSource('../../proxy-core/surfaces/geminiSurface.ts'); @@ -139,6 +145,23 @@ describe('proxy route architecture boundaries', () => { expect(surfaceSource).toContain('openAiResponsesTransformer.proxyStream.createSession('); }); + it('keeps oauth refresh recovery and success bookkeeping behind shared surface helpers', () => { + const chatSurfaceSource = readSource('../../proxy-core/surfaces/chatSurface.ts'); + const responsesSurfaceSource = readSource('../../proxy-core/surfaces/openAiResponsesSurface.ts'); + + expect(chatSurfaceSource).toContain('trySurfaceOauthRefreshRecovery('); + expect(chatSurfaceSource).toContain('recordSurfaceSuccess('); + expect(chatSurfaceSource).not.toContain('refreshOauthAccessTokenSingleflight('); + expect(chatSurfaceSource).not.toContain('resolveProxyUsageWithSelfLogFallback('); + expect(chatSurfaceSource).not.toContain('resolveProxyLogBilling('); + + expect(responsesSurfaceSource).toContain('trySurfaceOauthRefreshRecovery('); + expect(responsesSurfaceSource).toContain('recordSurfaceSuccess('); + expect(responsesSurfaceSource).not.toContain('refreshOauthAccessTokenSingleflight('); + expect(responsesSurfaceSource).not.toContain('resolveProxyUsageWithSelfLogFallback('); + expect(responsesSurfaceSource).not.toContain('resolveProxyLogBilling('); + }); + it('keeps canonical transformer contracts imported from the transformer boundary only', () => { const chatSource = readSource('./chat.ts'); const responsesSource = readSource('./responses.ts'); diff --git a/src/server/routes/proxy/upstreamEndpoint.ts b/src/server/routes/proxy/upstreamEndpoint.ts index 2441004..9a7ffaf 100644 --- a/src/server/routes/proxy/upstreamEndpoint.ts +++ b/src/server/routes/proxy/upstreamEndpoint.ts @@ -29,10 +29,7 @@ import { } from '../../transformers/shared/endpointCompatibility.js'; import { buildClaudeRuntimeHeaders, - buildCodexRuntimeHeaders, - buildGeminiCliRuntimeHeaders, buildGeminiCliUserAgent, - getInputHeader, headerValueToString, } from '../../proxy-core/providers/headerUtils.js'; export { @@ -1234,7 +1231,10 @@ export function buildUpstreamEndpointRequest(input: { ); const configuredResponsesBody = applyConfiguredPayloadRules(body); - if (providerProfile?.id === 'codex') { + if (sitePlatform === 'codex') { + if (providerProfile?.id !== 'codex') { + throw new Error(`missing codex provider profile for platform: ${sitePlatform}`); + } return providerProfile.prepareRequest({ endpoint: 'responses', modelName: input.modelName, @@ -1254,39 +1254,14 @@ export function buildUpstreamEndpointRequest(input: { }); } - const headers = sitePlatform === 'codex' - ? buildCodexRuntimeHeaders({ - baseHeaders: { - ...commonHeaders, - ...responsesHeaders, - }, - providerHeaders: input.providerHeaders, - explicitSessionId: asTrimmedString(input.codexExplicitSessionId) || null, - continuityKey: asTrimmedString(input.codexSessionCacheKey) || null, - }) - : ensureStreamAcceptHeader({ - ...commonHeaders, - ...responsesHeaders, - }, input.stream); - const codexSessionId = sitePlatform === 'codex' - ? (getInputHeader(headers, 'session_id') || getInputHeader(headers, 'session-id')) - : null; - const shouldInjectDerivedPromptCacheKey = sitePlatform === 'codex' - && !!codexSessionId - && !asTrimmedString((configuredResponsesBody as Record).prompt_cache_key) - && !asTrimmedString(input.codexExplicitSessionId) - && !!asTrimmedString(input.codexSessionCacheKey); - const runtimeBody = shouldInjectDerivedPromptCacheKey - ? { - ...configuredResponsesBody, - prompt_cache_key: codexSessionId, - } - : configuredResponsesBody; - + const headers = ensureStreamAcceptHeader({ + ...commonHeaders, + ...responsesHeaders, + }, input.stream); return { path: resolveEndpointPath('responses'), headers, - body: runtimeBody, + body: configuredResponsesBody, runtime, }; } diff --git a/src/server/services/accountMutationWorkflow.test.ts b/src/server/services/accountMutationWorkflow.test.ts index 221fbdc..9be12b5 100644 --- a/src/server/services/accountMutationWorkflow.test.ts +++ b/src/server/services/accountMutationWorkflow.test.ts @@ -107,6 +107,19 @@ describe('accountMutationWorkflow', () => { expect(rebuildTokenRoutesFromAvailabilityMock).toHaveBeenCalledTimes(1); }); + it('only marks refreshedModels when the refresh result explicitly says it refreshed', async () => { + refreshModelsForAccountMock.mockResolvedValue({ accountId: 4, refreshed: false, status: 'skipped' }); + + const { convergeAccountMutation } = await import('./accountMutationWorkflow.js'); + const result = await convergeAccountMutation({ + accountId: 4, + refreshModels: true, + }); + + expect(result.modelRefreshResult).toEqual({ accountId: 4, refreshed: false, status: 'skipped' }); + expect(result.refreshedModels).toBe(false); + }); + it('refreshes model coverage in batches and maps failures', async () => { refreshModelsForAccountMock .mockResolvedValueOnce({ accountId: 1, refreshed: true, status: 'success' }) @@ -129,4 +142,26 @@ describe('accountMutationWorkflow', () => { result: { createdRoutes: 2 }, }); }); + + it('rejects invalid batch sizes before starting the batch loop', async () => { + const { refreshAccountCoverageBatch } = await import('./accountMutationWorkflow.js'); + + await expect(refreshAccountCoverageBatch({ + accountIds: [1, 2], + batchSize: 1.5, + mapFailure: (accountId, errorMessage) => ({ accountId, errorMessage }), + })).rejects.toThrow('batchSize must be a positive integer'); + + expect(refreshModelsForAccountMock).not.toHaveBeenCalled(); + }); + + it('exposes a best-effort route rebuild helper for controller callers', async () => { + rebuildTokenRoutesFromAvailabilityMock.mockResolvedValueOnce({ createdRoutes: 1 }); + + const { rebuildRoutesBestEffort } = await import('./accountMutationWorkflow.js'); + await expect(rebuildRoutesBestEffort()).resolves.toBe(true); + + rebuildTokenRoutesFromAvailabilityMock.mockRejectedValueOnce(new Error('rebuild failed')); + await expect(rebuildRoutesBestEffort()).resolves.toBe(false); + }); }); diff --git a/src/server/services/accountMutationWorkflow.ts b/src/server/services/accountMutationWorkflow.ts index f873261..b4c1575 100644 --- a/src/server/services/accountMutationWorkflow.ts +++ b/src/server/services/accountMutationWorkflow.ts @@ -20,6 +20,15 @@ export type CoverageBatchRebuildResult = | { success: true; result: Awaited> } | { success: false; error: string }; +export async function rebuildRoutesBestEffort(): Promise { + try { + await rebuildTokenRoutesFromAvailability(); + return true; + } catch { + return false; + } +} + export async function convergeAccountMutation(input: { accountId: number; preferredApiToken?: string | null; @@ -110,7 +119,7 @@ export async function convergeAccountMutation(input: { const modelRefreshResult = await runStep(() => refreshModelsForAccount(input.accountId)); if (modelRefreshResult) { result.modelRefreshResult = modelRefreshResult; - result.refreshedModels = true; + result.refreshedModels = modelRefreshResult.refreshed === true; } } @@ -133,6 +142,11 @@ export async function refreshAccountCoverageBatch(input: { refresh: Array; rebuild: CoverageBatchRebuildResult | null; }> { + if (!Number.isInteger(input.batchSize) || input.batchSize <= 0) { + throw new Error('batchSize must be a positive integer'); + } + + const batchSize = input.batchSize; const uniqueAccountIds = Array.from(new Set( input.accountIds.filter((id) => Number.isFinite(id) && id > 0), )); @@ -142,8 +156,8 @@ export async function refreshAccountCoverageBatch(input: { } const refresh: Array = []; - for (let offset = 0; offset < uniqueAccountIds.length; offset += input.batchSize) { - const batch = uniqueAccountIds.slice(offset, offset + input.batchSize); + for (let offset = 0; offset < uniqueAccountIds.length; offset += batchSize) { + const batch = uniqueAccountIds.slice(offset, offset + batchSize); const settled = await Promise.allSettled( batch.map(async (accountId) => refreshModelsForAccount(accountId)), ); diff --git a/src/server/services/platforms/standardApiProvider.test.ts b/src/server/services/platforms/standardApiProvider.test.ts index fe2e04b..f3cfaba 100644 --- a/src/server/services/platforms/standardApiProvider.test.ts +++ b/src/server/services/platforms/standardApiProvider.test.ts @@ -7,6 +7,7 @@ import { class TestStandardApiProviderAdapter extends StandardApiProviderAdapterBase { readonly platformName = 'test-standard'; + fetchJsonImpl = async () => ({ data: [] as Array<{ id: string }> }); async detect(_url: string): Promise { return false; @@ -15,6 +16,14 @@ class TestStandardApiProviderAdapter extends StandardApiProviderAdapterBase { async getModels(_baseUrl: string, _token: string): Promise { return []; } + + protected override async fetchJson(url: string, options?: Parameters[1]): Promise { + return this.fetchJsonImpl(url, options) as Promise; + } + + async fetchModelsForTest(options: Parameters[0]) { + return this.fetchModelsFromStandardEndpoint(options); + } } describe('standardApiProvider helpers', () => { @@ -43,4 +52,33 @@ describe('standardApiProvider helpers', () => { quota: 0, }); }); + + it('does not swallow mapper bugs while still returning empty lists for network failures', async () => { + const adapter = new TestStandardApiProviderAdapter(); + adapter.fetchJsonImpl = async () => ({ data: [{ id: 'gpt-5' }] }); + + await expect(adapter.fetchModelsForTest({ + baseUrl: 'https://api.example.com', + mapResponse: () => { + throw new Error('mapper exploded'); + }, + })).rejects.toThrow('mapper exploded'); + + adapter.fetchJsonImpl = async () => { + throw new Error('network failed'); + }; + + await expect(adapter.fetchModelsForTest({ + baseUrl: 'https://api.example.com', + })).resolves.toEqual([]); + }); + + it('rejects invalid payload shapes instead of silently treating them as no models', async () => { + const adapter = new TestStandardApiProviderAdapter(); + adapter.fetchJsonImpl = async () => ({ data: 'not-an-array' }); + + await expect(adapter.fetchModelsForTest({ + baseUrl: 'https://api.example.com', + })).rejects.toThrow('invalid standard models payload'); + }); }); diff --git a/src/server/services/platforms/standardApiProvider.ts b/src/server/services/platforms/standardApiProvider.ts index 1448168..76c516b 100644 --- a/src/server/services/platforms/standardApiProvider.ts +++ b/src/server/services/platforms/standardApiProvider.ts @@ -60,18 +60,27 @@ export abstract class StandardApiProviderAdapterBase extends BasePlatformAdapter ? options.resolveUrl(normalizedBaseUrl) : resolveVersionedModelsUrl(normalizedBaseUrl); + let payload: any; try { - const payload = await this.fetchJson(url, { + payload = await this.fetchJson(url, { headers: options.headers, }); - const rows = options.mapResponse - ? options.mapResponse(payload) - : (payload?.data || []).map((item: any) => item?.id); - return rows - .map((item) => (typeof item === 'string' ? item.trim() : '')) - .filter((item) => item.length > 0); } catch { return []; } + + const rows = options.mapResponse + ? options.mapResponse(payload) + : Array.isArray(payload?.data) + ? payload.data.map((item: any) => item?.id) + : null; + + if (!Array.isArray(rows)) { + throw new Error('invalid standard models payload'); + } + + return rows + .map((item) => (typeof item === 'string' ? item.trim() : '')) + .filter((item) => item.length > 0); } } diff --git a/src/server/transformers/openai/responses/conversion.test.ts b/src/server/transformers/openai/responses/conversion.test.ts index e06bc31..e767c47 100644 --- a/src/server/transformers/openai/responses/conversion.test.ts +++ b/src/server/transformers/openai/responses/conversion.test.ts @@ -34,6 +34,47 @@ describe('responses conversion single source of truth', () => { expect(convertOpenAiBodyToResponsesBodyViaCompatibility).toBe(convertOpenAiBodyToResponsesBody); expect(sanitizeResponsesBodyForProxyViaCompatibility).toBe(sanitizeResponsesBodyForProxy); }); + + it('preserves extra properties when normalizing role-only message items', () => { + expect(normalizeResponsesInputForCompatibility([ + { + role: 'assistant', + content: 'done', + id: 'msg_1', + status: 'completed', + }, + ])).toEqual([ + { + type: 'message', + role: 'assistant', + content: [{ type: 'output_text', text: 'done' }], + id: 'msg_1', + status: 'completed', + }, + ]); + }); + + it('filters whitespace-only string entries from normalized responses input arrays', () => { + expect(normalizeResponsesInputForCompatibility([ + 'hello', + ' ', + { + role: 'user', + content: 'world', + }, + ])).toEqual([ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'world' }], + }, + ]); + }); }); describe('sanitizeResponsesBodyForProxy', () => { diff --git a/src/server/transformers/openai/responses/normalization.ts b/src/server/transformers/openai/responses/normalization.ts index d27cfb2..9dc820f 100644 --- a/src/server/transformers/openai/responses/normalization.ts +++ b/src/server/transformers/openai/responses/normalization.ts @@ -136,6 +136,7 @@ export function normalizeResponsesMessageItem(item: Record): Re if (asTrimmedString(item.role)) { return { + ...item, type: 'message', role, content: normalizedContent, @@ -158,13 +159,13 @@ export function normalizeResponsesInputForCompatibility(input: unknown): unknown } if (Array.isArray(input)) { - return input.map((item) => { + return input.flatMap((item) => { if (typeof item === 'string') { const normalized = item.trim(); - return normalized ? toResponsesInputMessageFromText(normalized) : item; + return normalized ? [toResponsesInputMessageFromText(normalized)] : []; } - if (!isRecord(item)) return item; - return normalizeResponsesMessageItem(item); + if (!isRecord(item)) return [item]; + return [normalizeResponsesMessageItem(item)]; }); } diff --git a/src/shared/conversationFileTypes.js b/src/shared/conversationFileTypes.js index 095cd6d..d0af49c 100644 --- a/src/shared/conversationFileTypes.js +++ b/src/shared/conversationFileTypes.js @@ -17,6 +17,12 @@ function normalizeValue(value) { return (value || '').trim().toLowerCase(); } +function normalizeMimeType(value) { + const normalized = normalizeValue(value); + const [essence] = normalized.split(';'); + return (essence || '').trim(); +} + export function inferConversationFileMimeType(filename) { const normalized = normalizeValue(filename); if (normalized.endsWith('.pdf')) return 'application/pdf'; @@ -41,7 +47,7 @@ export function inferConversationFileMimeType(filename) { } export function resolveConversationFileMimeType(mimeType, filename) { - const normalizedMimeType = normalizeValue(mimeType); + const normalizedMimeType = normalizeMimeType(mimeType); if (normalizedMimeType && !GENERIC_MIME_TYPES.has(normalizedMimeType)) { return normalizedMimeType; } @@ -49,14 +55,14 @@ export function resolveConversationFileMimeType(mimeType, filename) { } export function classifyConversationFileMimeType(mimeType) { - const normalized = normalizeValue(mimeType); + const normalized = normalizeMimeType(mimeType); if (normalized.startsWith('image/')) return 'image'; if (normalized.startsWith('audio/')) return 'audio'; return 'document'; } export function detectConversationFileKind(file) { - const mimeType = normalizeValue(file?.mimeType); + const mimeType = normalizeMimeType(file?.mimeType); if (mimeType) { if (mimeType.startsWith('image/')) return 'image'; if (mimeType.startsWith('audio/')) return 'audio'; @@ -73,7 +79,7 @@ export function detectConversationFileKind(file) { } export function isSupportedConversationFileMimeType(mimeType) { - const normalized = normalizeValue(mimeType); + const normalized = normalizeMimeType(mimeType); return DOCUMENT_MIME_TYPES.has(normalized) || normalized.startsWith('image/') || normalized.startsWith('audio/'); diff --git a/src/shared/conversationFileTypes.test.ts b/src/shared/conversationFileTypes.test.ts index 740fc70..fef7cb3 100644 --- a/src/shared/conversationFileTypes.test.ts +++ b/src/shared/conversationFileTypes.test.ts @@ -42,4 +42,14 @@ describe('conversationFileTypes', () => { mimeType: 'application/octet-stream', })).toBe('document'); }); + + it('strips MIME parameters before matching support and kind rules', () => { + expect(resolveConversationFileMimeType('text/plain; charset=utf-8', 'notes.txt')).toBe('text/plain'); + expect(classifyConversationFileMimeType('text/plain; charset=utf-8')).toBe('document'); + expect(detectConversationFileKind({ + filename: 'notes.txt', + mimeType: 'text/plain; charset=utf-8', + })).toBe('document'); + expect(isSupportedConversationFileMimeType('text/plain; charset=utf-8')).toBe(true); + }); }); diff --git a/src/shared/tokenRoutePatterns.js b/src/shared/tokenRoutePatterns.js index b35b1b1..99c0ed5 100644 --- a/src/shared/tokenRoutePatterns.js +++ b/src/shared/tokenRoutePatterns.js @@ -86,6 +86,11 @@ function isSafeRegexPatternBody(body) { continue; } if (ch === '\\') { + const next = body[index + 1]; + if (!next) return false; + if (/[a-z]/i.test(next) && next !== 'd') { + return false; + } escaped = true; continue; } @@ -99,10 +104,7 @@ function isSafeRegexPatternBody(body) { } if (ch === '(') { if (body[index + 1] === '?') { - if (body[index + 2] !== ':') return false; - groupStack.push({ hasInnerQuantifier: false, hasAlternation: false }); - index += 2; - continue; + return false; } groupStack.push({ hasInnerQuantifier: false, hasAlternation: false }); continue; diff --git a/src/shared/tokenRoutePatterns.test.ts b/src/shared/tokenRoutePatterns.test.ts index f5da960..d1adf22 100644 --- a/src/shared/tokenRoutePatterns.test.ts +++ b/src/shared/tokenRoutePatterns.test.ts @@ -21,6 +21,18 @@ describe('token route pattern helpers', () => { expect(matchesTokenRouteModelPattern('claude-sonnet-4-6', 're:(?=claude)')).toBe(false); }); + it('rejects regex syntax that the lightweight parser does not implement', async () => { + const { + matchesTokenRouteModelPattern, + parseTokenRouteRegexPattern, + } = await import('./tokenRoutePatterns.js'); + + expect(parseTokenRouteRegexPattern('re:^(?:gpt|claude)-5$').regex).toBeNull(); + expect(matchesTokenRouteModelPattern('gpt-5', 're:^(?:gpt|claude)-5$')).toBe(false); + expect(parseTokenRouteRegexPattern('re:^gpt-\\s+$').regex).toBeNull(); + expect(matchesTokenRouteModelPattern('gpt- ', 're:^gpt-\\s+$')).toBe(false); + }); + it('supports exact, glob, and safe regex route matches', async () => { const { matchesTokenRouteModelPattern } = await import('./tokenRoutePatterns.js');