增加了空回为失败和自定义失败关键词的功能，修复了机器人提出的问题 (#153)
* Add proxy failure rules for keywords and empty responses * fix(proxy): detect failure keywords for streamed chat responses * Fix(tester): treat empty stream output as error * feat(server): default listen host to IPv6 any-address * fix(proxy): treat empty upstream content as failure * Revert "feat(server): default listen host to IPv6 any-address" This reverts commit 2ca49a3891cc3a2e37138156b17b660b6a3cb7ea. * fix(settings): validate proxy failure settings * fix: address CodeRabbit feedback in proxy failure detection * test: force NODE_ENV=test for vitest to avoid React act() prod build * fix(proxy/chat): preserve true SSE streaming (no upstream.text buffering) * fix: avoid http fallback after hijacked SSE stream * fix: wire proxy failure detection into chat and responses --------- Co-authored-by: metapi-bot <metapi-bot@local> Co-authored-by: openclaw <bot@local> Co-authored-by: openclaw <openclaw@local> Co-authored-by: openclaw-executor <executor@openclaw.local> Co-authored-by: xuyufengfei <xuyufengfei@users.noreply.github.com> Co-authored-by: cita <juricek.chen@gmail.com>
This commit is contained in:
xuyufengfei
@@ -94,6 +94,8 @@ export function buildConfig(env: NodeJS.ProcessEnv) {
|
||||
tokenRouterCacheTtlMs: Math.max(100, Math.trunc(parseNumber(env.TOKEN_ROUTER_CACHE_TTL_MS, 1_500))),
|
||||
proxyLogRetentionDays: Math.max(0, Math.trunc(parseNumber(env.PROXY_LOG_RETENTION_DAYS, 30))),
|
||||
proxyLogRetentionPruneIntervalMinutes: Math.max(1, Math.trunc(parseNumber(env.PROXY_LOG_RETENTION_PRUNE_INTERVAL_MINUTES, 30))),
|
||||
proxyErrorKeywords: parseCsvList(env.PROXY_ERROR_KEYWORDS),
|
||||
proxyEmptyContentFailEnabled: parseBoolean(env.PROXY_EMPTY_CONTENT_FAIL, false),
|
||||
routingWeights: {
|
||||
baseWeightFactor: parseNumber(env.BASE_WEIGHT_FACTOR, 0.5),
|
||||
valueScoreFactor: parseNumber(env.VALUE_SCORE_FACTOR, 0.5),
|
||||
|
||||
@@ -129,6 +129,16 @@ function applyRuntimeSettings(settingsMap: Map<string, string>) {
|
||||
const systemProxyUrl = parseSettingFromMap<string>(settingsMap, 'system_proxy_url');
|
||||
if (typeof systemProxyUrl === 'string') config.systemProxyUrl = systemProxyUrl;
|
||||
|
||||
const proxyErrorKeywords = parseSettingFromMap<string[] | string>(settingsMap, 'proxy_error_keywords');
|
||||
if (proxyErrorKeywords !== undefined) {
|
||||
config.proxyErrorKeywords = toStringList(proxyErrorKeywords);
|
||||
}
|
||||
|
||||
const proxyEmptyContentFailEnabled = parseSettingFromMap<boolean>(settingsMap, 'proxy_empty_content_fail_enabled');
|
||||
if (typeof proxyEmptyContentFailEnabled === 'boolean') {
|
||||
config.proxyEmptyContentFailEnabled = proxyEmptyContentFailEnabled;
|
||||
}
|
||||
|
||||
const checkinCron = parseSettingFromMap<string>(settingsMap, 'checkin_cron');
|
||||
if (typeof checkinCron === 'string' && checkinCron) config.checkinCron = checkinCron;
|
||||
|
||||
|
||||
@@ -273,6 +273,60 @@ describe('settings and auth events', () => {
|
||||
expect(runtime.systemProxyUrl).toBe('http://127.0.0.1:7890');
|
||||
});
|
||||
|
||||
it('splits proxy error keywords on newlines and commas when saving runtime settings', async () => {
|
||||
const updateResponse = await app.inject({
|
||||
method: 'PUT',
|
||||
url: '/api/settings/runtime',
|
||||
payload: {
|
||||
proxyErrorKeywords: 'quota exceeded\nbad gateway,too many requests',
|
||||
proxyEmptyContentFailEnabled: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect(updateResponse.statusCode).toBe(200);
|
||||
const updated = updateResponse.json() as {
|
||||
proxyErrorKeywords?: string[];
|
||||
proxyEmptyContentFailEnabled?: boolean;
|
||||
};
|
||||
expect(updated.proxyErrorKeywords).toEqual([
|
||||
'quota exceeded',
|
||||
'bad gateway',
|
||||
'too many requests',
|
||||
]);
|
||||
expect(updated.proxyEmptyContentFailEnabled).toBe(true);
|
||||
expect(config.proxyErrorKeywords).toEqual([
|
||||
'quota exceeded',
|
||||
'bad gateway',
|
||||
'too many requests',
|
||||
]);
|
||||
expect(config.proxyEmptyContentFailEnabled).toBe(true);
|
||||
|
||||
const rows = await db.select().from(schema.settings).all();
|
||||
const settingsMap = new Map(rows.map((row) => [row.key, row.value]));
|
||||
expect(settingsMap.get('proxy_error_keywords')).toBe(JSON.stringify([
|
||||
'quota exceeded',
|
||||
'bad gateway',
|
||||
'too many requests',
|
||||
]));
|
||||
expect(settingsMap.get('proxy_empty_content_fail_enabled')).toBe(JSON.stringify(true));
|
||||
|
||||
const getResponse = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/settings/runtime',
|
||||
});
|
||||
expect(getResponse.statusCode).toBe(200);
|
||||
const runtime = getResponse.json() as {
|
||||
proxyErrorKeywords?: string[];
|
||||
proxyEmptyContentFailEnabled?: boolean;
|
||||
};
|
||||
expect(runtime.proxyErrorKeywords).toEqual([
|
||||
'quota exceeded',
|
||||
'bad gateway',
|
||||
'too many requests',
|
||||
]);
|
||||
expect(runtime.proxyEmptyContentFailEnabled).toBe(true);
|
||||
});
|
||||
|
||||
it('persists and returns log cleanup settings from runtime settings', async () => {
|
||||
const updateResponse = await app.inject({
|
||||
method: 'PUT',
|
||||
|
||||
@@ -55,6 +55,8 @@ interface RuntimeSettingsBody {
|
||||
adminIpAllowlist?: string[] | string;
|
||||
routingFallbackUnitCost?: number;
|
||||
routingWeights?: Partial<RoutingWeights>;
|
||||
proxyErrorKeywords?: string[] | string;
|
||||
proxyEmptyContentFailEnabled?: boolean;
|
||||
}
|
||||
|
||||
interface DatabaseMigrationBody {
|
||||
@@ -127,6 +129,33 @@ function toStringList(value: unknown): string[] {
|
||||
return [];
|
||||
}
|
||||
|
||||
function parseProxyErrorKeywords(value: unknown): string[] {
|
||||
const splitKeywords = (input: string): string[] => input
|
||||
.split(/\r?\n|,/)
|
||||
.map((item) => item.trim())
|
||||
.filter((item) => item.length > 0);
|
||||
|
||||
if (Array.isArray(value)) {
|
||||
const keywords = value.flatMap((item) => {
|
||||
if (typeof item !== 'string') return [];
|
||||
return splitKeywords(item);
|
||||
});
|
||||
return keywords;
|
||||
}
|
||||
|
||||
if (typeof value === 'string') {
|
||||
const keywords = splitKeywords(value);
|
||||
return keywords;
|
||||
}
|
||||
|
||||
throw new Error('上游错误关键词格式无效：需要 string 或 string[]');
|
||||
}
|
||||
|
||||
function parseBooleanFlag(value: unknown, label: string): boolean {
|
||||
if (typeof value === 'boolean') return value;
|
||||
throw new Error(`${label}格式无效：需要 boolean`);
|
||||
}
|
||||
|
||||
function isValidHttpUrl(raw: string): boolean {
|
||||
const value = String(raw || '').trim();
|
||||
if (!value) return false;
|
||||
@@ -197,6 +226,22 @@ function applyImportedSettingToRuntime(key: string, value: unknown) {
|
||||
config.systemProxyUrl = normalizeSiteProxyUrl(value) || '';
|
||||
return;
|
||||
}
|
||||
case 'proxy_error_keywords': {
|
||||
try {
|
||||
config.proxyErrorKeywords = parseProxyErrorKeywords(value);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
return;
|
||||
}
|
||||
case 'proxy_empty_content_fail_enabled': {
|
||||
try {
|
||||
config.proxyEmptyContentFailEnabled = parseBooleanFlag(value, '空内容判定失败开关');
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
return;
|
||||
}
|
||||
case 'webhook_url': {
|
||||
if (typeof value !== 'string') return;
|
||||
config.webhookUrl = value.trim();
|
||||
@@ -347,6 +392,8 @@ function getRuntimeSettingsResponse(currentAdminIp = '') {
|
||||
adminIpAllowlist: config.adminIpAllowlist,
|
||||
currentAdminIp,
|
||||
systemProxyUrl: config.systemProxyUrl,
|
||||
proxyErrorKeywords: config.proxyErrorKeywords,
|
||||
proxyEmptyContentFailEnabled: config.proxyEmptyContentFailEnabled,
|
||||
proxyTokenMasked: maskSecret(config.proxyToken),
|
||||
};
|
||||
}
|
||||
@@ -603,6 +650,42 @@ export async function settingsRoutes(app: FastifyInstance) {
|
||||
invalidateSiteProxyCache();
|
||||
}
|
||||
|
||||
if (body.proxyErrorKeywords !== undefined) {
|
||||
let nextKeywords: string[] = [];
|
||||
try {
|
||||
nextKeywords = parseProxyErrorKeywords(body.proxyErrorKeywords);
|
||||
} catch (err: any) {
|
||||
return reply.code(400).send({
|
||||
success: false,
|
||||
message: err?.message || '上游错误关键词格式无效',
|
||||
});
|
||||
}
|
||||
|
||||
if (JSON.stringify(nextKeywords) !== JSON.stringify(config.proxyErrorKeywords || [])) {
|
||||
changedLabels.push('上游错误关键词');
|
||||
}
|
||||
config.proxyErrorKeywords = nextKeywords;
|
||||
upsertSetting('proxy_error_keywords', config.proxyErrorKeywords);
|
||||
}
|
||||
|
||||
if (body.proxyEmptyContentFailEnabled !== undefined) {
|
||||
let nextValue = false;
|
||||
try {
|
||||
nextValue = parseBooleanFlag(body.proxyEmptyContentFailEnabled, '空内容判定失败开关');
|
||||
} catch (err: any) {
|
||||
return reply.code(400).send({
|
||||
success: false,
|
||||
message: err?.message || '空内容判定失败开关格式无效',
|
||||
});
|
||||
}
|
||||
|
||||
if (nextValue !== config.proxyEmptyContentFailEnabled) {
|
||||
changedLabels.push('空内容判定失败');
|
||||
}
|
||||
config.proxyEmptyContentFailEnabled = nextValue;
|
||||
upsertSetting('proxy_empty_content_fail_enabled', config.proxyEmptyContentFailEnabled);
|
||||
}
|
||||
|
||||
if (body.webhookUrl !== undefined) {
|
||||
if (String(body.webhookUrl || '').trim() !== config.webhookUrl) {
|
||||
changedLabels.push('Webhook 地址');
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import Fastify, { type FastifyInstance } from 'fastify';
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import { config } from '../../config.js';
|
||||
|
||||
const fetchMock = vi.fn();
|
||||
const selectChannelMock = vi.fn();
|
||||
@@ -109,6 +110,8 @@ describe('chat proxy stream behavior', () => {
|
||||
});
|
||||
selectNextChannelMock.mockReturnValue(null);
|
||||
fetchModelPricingCatalogMock.mockResolvedValue(null);
|
||||
config.proxyEmptyContentFailEnabled = false;
|
||||
config.proxyErrorKeywords = [];
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -150,6 +153,77 @@ describe('chat proxy stream behavior', () => {
|
||||
expect(response.body).toContain('data: [DONE]');
|
||||
});
|
||||
|
||||
it('returns upstream_error for empty non-stream chat responses when empty-content failure is enabled', async () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'chatcmpl-empty',
|
||||
object: 'chat.completion',
|
||||
created: 1_706_000_000,
|
||||
model: 'upstream-gpt',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: '' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 6, completion_tokens: 0, total_tokens: 6 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
messages: [{ role: 'user', content: 'hi' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(502);
|
||||
expect(response.json()?.error?.type).toBe('upstream_error');
|
||||
expect(response.json()?.error?.message).toContain('empty content');
|
||||
expect(recordSuccessMock).not.toHaveBeenCalled();
|
||||
expect(recordFailureMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('returns HTTP upstream_error instead of hijacking when streamed chat requests receive empty non-SSE payloads', async () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'chatcmpl-empty-stream',
|
||||
object: 'chat.completion',
|
||||
created: 1_706_000_000,
|
||||
model: 'upstream-gpt',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: '' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 4, completion_tokens: 0, total_tokens: 4 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/chat/completions',
|
||||
payload: {
|
||||
model: 'gpt-4o-mini',
|
||||
stream: true,
|
||||
messages: [{ role: 'user', content: 'hi' }],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(502);
|
||||
expect(response.headers['content-type']).not.toContain('text/event-stream');
|
||||
expect(response.json()?.error?.type).toBe('upstream_error');
|
||||
expect(recordSuccessMock).not.toHaveBeenCalled();
|
||||
expect(recordFailureMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('returns clear 400 when /v1/chat/completions receives responses-style input without messages', async () => {
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
@@ -1989,6 +2063,71 @@ describe('chat proxy stream behavior', () => {
|
||||
expect(targetUrl).toContain('/v1/responses');
|
||||
});
|
||||
|
||||
it('returns upstream_error for empty non-stream /v1/responses payloads when empty-content failure is enabled', async () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp-empty',
|
||||
object: 'response',
|
||||
model: 'gpt-5.4',
|
||||
status: 'completed',
|
||||
output: [],
|
||||
output_text: '',
|
||||
usage: { input_tokens: 3, output_tokens: 0, total_tokens: 3 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.4',
|
||||
input: 'hello',
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(502);
|
||||
expect(response.json()?.error?.type).toBe('upstream_error');
|
||||
expect(response.json()?.error?.message).toContain('empty content');
|
||||
expect(recordSuccessMock).not.toHaveBeenCalled();
|
||||
expect(recordFailureMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('returns HTTP upstream_error instead of hijacking when streamed /v1/responses receives empty non-SSE payloads', async () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify({
|
||||
id: 'resp-empty-stream',
|
||||
object: 'response',
|
||||
model: 'gpt-5.4',
|
||||
status: 'completed',
|
||||
output: [],
|
||||
output_text: '',
|
||||
usage: { input_tokens: 2, output_tokens: 0, total_tokens: 2 },
|
||||
}), {
|
||||
status: 200,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}));
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
payload: {
|
||||
model: 'gpt-5.4',
|
||||
input: 'hello',
|
||||
stream: true,
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(502);
|
||||
expect(response.headers['content-type']).not.toContain('text/event-stream');
|
||||
expect(response.json()?.error?.type).toBe('upstream_error');
|
||||
expect(recordSuccessMock).not.toHaveBeenCalled();
|
||||
expect(recordFailureMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('prefers native /v1/responses for claude-family /v1/responses requests that include input_file file_url', async () => {
|
||||
selectChannelMock.mockReturnValue({
|
||||
channel: { id: 11, routeId: 22 },
|
||||
|
||||
+142
-82
@@ -20,7 +20,8 @@ import {
|
||||
recordDownstreamCostUsage,
|
||||
} from './downstreamPolicy.js';
|
||||
import { composeProxyLogMessage } from './logPathMeta.js';
|
||||
import { executeEndpointFlow, type BuiltEndpointRequest } from './endpointFlow.js';
|
||||
import { executeEndpointFlow, withUpstreamPath, type BuiltEndpointRequest } from './endpointFlow.js';
|
||||
import { detectProxyFailure } from './proxyFailureJudge.js';
|
||||
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
|
||||
import { resolveProxyLogBilling } from './proxyBilling.js';
|
||||
import { openAiChatTransformer } from '../../transformers/openai/chat/index.js';
|
||||
@@ -342,12 +343,15 @@ async function handleChatProxyRequest(
|
||||
const successfulUpstreamPath = endpointResult.upstreamPath;
|
||||
|
||||
if (isStream) {
|
||||
reply.hijack();
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
|
||||
const startSseResponse = () => {
|
||||
reply.hijack();
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
};
|
||||
|
||||
let parsedUsage: ReturnType<typeof parseProxyUsage> = {
|
||||
promptTokens: 0,
|
||||
@@ -377,10 +381,11 @@ async function handleChatProxyRequest(
|
||||
reply.raw.write(chunk);
|
||||
},
|
||||
});
|
||||
let rawText = '';
|
||||
|
||||
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
|
||||
if (!upstreamContentType.includes('text/event-stream')) {
|
||||
const fallbackText = await upstream.text();
|
||||
rawText = fallbackText;
|
||||
let fallbackData: unknown = null;
|
||||
try {
|
||||
fallbackData = JSON.parse(fallbackText);
|
||||
@@ -390,7 +395,96 @@ async function handleChatProxyRequest(
|
||||
if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
|
||||
fallbackData = unwrapGeminiCliPayload(fallbackData);
|
||||
}
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(fallbackData));
|
||||
const latency = Date.now() - startTime;
|
||||
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
|
||||
if (failure) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
failure.status,
|
||||
latency,
|
||||
failure.reason,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
|
||||
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: failure.reason,
|
||||
});
|
||||
|
||||
return reply.code(failure.status).send({
|
||||
error: { message: failure.reason, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
|
||||
startSseResponse();
|
||||
const streamResult = streamSession.consumeUpstreamFinalPayload(fallbackData, fallbackText, reply.raw);
|
||||
if (streamResult.status === 'failed') {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
200,
|
||||
latency,
|
||||
streamResult.errorMessage,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
return;
|
||||
}
|
||||
} else {
|
||||
startSseResponse();
|
||||
// Preserve real SSE streaming instead of buffering the whole upstream body.
|
||||
const upstreamReader = upstream.body?.getReader();
|
||||
const baseReader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
|
||||
? createGeminiCliStreamReader(upstreamReader)
|
||||
: upstreamReader;
|
||||
const decoder = new TextDecoder();
|
||||
const reader = baseReader
|
||||
? {
|
||||
async read() {
|
||||
const result = await baseReader.read();
|
||||
if (result.value) {
|
||||
rawText += decoder.decode(result.value, { stream: true });
|
||||
}
|
||||
return result;
|
||||
},
|
||||
async cancel(reason?: unknown) {
|
||||
return baseReader.cancel(reason);
|
||||
},
|
||||
releaseLock() {
|
||||
return baseReader.releaseLock();
|
||||
},
|
||||
}
|
||||
: baseReader;
|
||||
const streamResult = await streamSession.run(reader, reply.raw);
|
||||
rawText += decoder.decode();
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
if (streamResult.status === 'failed') {
|
||||
@@ -416,83 +510,13 @@ async function handleChatProxyRequest(
|
||||
return;
|
||||
}
|
||||
|
||||
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,
|
||||
modelName,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'success',
|
||||
200,
|
||||
latency,
|
||||
null,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
resolvedUsage.promptTokens,
|
||||
resolvedUsage.completionTokens,
|
||||
resolvedUsage.totalTokens,
|
||||
estimatedCost,
|
||||
billingDetails,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
return;
|
||||
// Once SSE has been hijacked and streamed downstream, we can no longer
|
||||
// safely fall back to an HTTP error response or retry by switching the
|
||||
// channel mid-flight. Stream-level failures must be handled in-band by
|
||||
// the proxy stream session itself.
|
||||
}
|
||||
|
||||
const upstreamReader = upstream.body?.getReader();
|
||||
const reader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
|
||||
? createGeminiCliStreamReader(upstreamReader)
|
||||
: upstreamReader;
|
||||
const streamResult = await streamSession.run(reader, reply.raw);
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
if (streamResult.status === 'failed') {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
200,
|
||||
latency,
|
||||
streamResult.errorMessage,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
@@ -562,6 +586,42 @@ async function handleChatProxyRequest(
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(upstreamData);
|
||||
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
|
||||
if (failure) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
failure.status,
|
||||
latency,
|
||||
failure.reason,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
|
||||
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: failure.reason,
|
||||
});
|
||||
|
||||
return reply.code(failure.status).send({
|
||||
error: { message: failure.reason, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
const normalizedFinal = downstreamTransformer.transformFinalResponse(upstreamData, modelName, rawText);
|
||||
const downstreamResponse = downstreamTransformer.serializeFinalResponse(normalizedFinal, parsedUsage);
|
||||
|
||||
|
||||
@@ -1,49 +1,50 @@
|
||||
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import { fetch } from 'undici';
|
||||
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
|
||||
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
|
||||
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
|
||||
import { composeProxyLogMessage } from './logPathMeta.js';
|
||||
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
|
||||
import { resolveProxyLogBilling } from './proxyBilling.js';
|
||||
import { getProxyAuthContext } from '../../middleware/auth.js';
|
||||
import { buildUpstreamUrl } from './upstreamUrl.js';
|
||||
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
|
||||
import { isTokenExpiredError } from '../../services/alertRules.js';
|
||||
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
|
||||
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
|
||||
import { mergeProxyUsage, parseProxyUsage, pullSseDataEvents } from '../../services/proxyUsageParser.js';
|
||||
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
|
||||
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
|
||||
import { composeProxyLogMessage } from './logPathMeta.js';
|
||||
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
|
||||
import { detectProxyFailure } from './proxyFailureJudge.js';
|
||||
import { resolveProxyLogBilling } from './proxyBilling.js';
|
||||
import { getProxyAuthContext } from '../../middleware/auth.js';
|
||||
import { buildUpstreamUrl } from './upstreamUrl.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
app.post('/v1/completions', async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const body = request.body as any;
|
||||
const requestedModel = body?.model;
|
||||
if (!requestedModel) {
|
||||
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
|
||||
}
|
||||
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
|
||||
const downstreamPolicy = getDownstreamRoutingPolicy(request);
|
||||
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
|
||||
const logDownstreamApiKeyId = downstreamApiKeyId !== null
|
||||
&& await hasProxyLogDownstreamApiKeyIdColumn();
|
||||
|
||||
const isStream = body.stream === true;
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
const requestedModel = body?.model;
|
||||
if (!requestedModel) {
|
||||
return reply.code(400).send({ error: { message: 'model is required', type: 'invalid_request_error' } });
|
||||
}
|
||||
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
|
||||
const downstreamPolicy = getDownstreamRoutingPolicy(request);
|
||||
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
|
||||
const logDownstreamApiKeyId = downstreamApiKeyId !== null
|
||||
&& await hasProxyLogDownstreamApiKeyIdColumn();
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
|
||||
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
|
||||
}
|
||||
const isStream = body.stream === true;
|
||||
const excludeChannelIds: number[] = [];
|
||||
let retryCount = 0;
|
||||
|
||||
while (retryCount <= MAX_RETRIES) {
|
||||
let selected = retryCount === 0
|
||||
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
|
||||
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
|
||||
|
||||
if (!selected && retryCount === 0) {
|
||||
await refreshModelsAndRebuildRoutes();
|
||||
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
|
||||
}
|
||||
|
||||
if (!selected) {
|
||||
await reportProxyAllFailed({
|
||||
@@ -55,26 +56,26 @@ export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
});
|
||||
}
|
||||
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/completions');
|
||||
const forwardBody = { ...body, model: selected.actualModel };
|
||||
const startTime = Date.now();
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const targetUrl = buildUpstreamUrl(selected.site.url, '/v1/completions');
|
||||
const forwardBody = { ...body, model: selected.actualModel };
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
}));
|
||||
const upstream = await fetch(targetUrl, withSiteRecordProxyRequestInit(selected.site, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Authorization': `Bearer ${selected.tokenValue}`,
|
||||
},
|
||||
body: JSON.stringify(forwardBody),
|
||||
}));
|
||||
|
||||
if (!upstream.ok) {
|
||||
const errText = await upstream.text().catch(() => 'unknown error');
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
|
||||
if (!upstream.ok) {
|
||||
const errText = await upstream.text().catch(() => 'unknown error');
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, errText, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
|
||||
|
||||
if (isTokenExpiredError({ status: upstream.status, message: errText })) {
|
||||
await reportTokenExpired({
|
||||
@@ -85,10 +86,10 @@ export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
});
|
||||
}
|
||||
|
||||
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
if (shouldRetryProxyRequest(upstream.status, errText) && retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
@@ -110,117 +111,153 @@ export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
return;
|
||||
}
|
||||
|
||||
const decoder = new TextDecoder();
|
||||
let parsedUsage: ReturnType<typeof parseProxyUsage> = {
|
||||
promptTokens: 0,
|
||||
completionTokens: 0,
|
||||
totalTokens: 0,
|
||||
cacheReadTokens: 0,
|
||||
cacheCreationTokens: 0,
|
||||
promptTokensIncludeCache: null,
|
||||
};
|
||||
let sseBuffer = '';
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
reply.raw.write(chunk);
|
||||
|
||||
sseBuffer += chunk;
|
||||
const pulled = pullSseDataEvents(sseBuffer);
|
||||
sseBuffer = pulled.rest;
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
if (sseBuffer.trim().length > 0) {
|
||||
const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
reply.raw.end();
|
||||
}
|
||||
const decoder = new TextDecoder();
|
||||
let parsedUsage: ReturnType<typeof parseProxyUsage> = {
|
||||
promptTokens: 0,
|
||||
completionTokens: 0,
|
||||
totalTokens: 0,
|
||||
cacheReadTokens: 0,
|
||||
cacheCreationTokens: 0,
|
||||
promptTokensIncludeCache: null,
|
||||
};
|
||||
let sseBuffer = '';
|
||||
try {
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
const chunk = decoder.decode(value, { stream: true });
|
||||
reply.raw.write(chunk);
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const data = await upstream.json() as any;
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(data);
|
||||
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,
|
||||
},
|
||||
});
|
||||
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
|
||||
);
|
||||
return reply.send(data);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
sseBuffer += chunk;
|
||||
const pulled = pullSseDataEvents(sseBuffer);
|
||||
sseBuffer = pulled.rest;
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
if (sseBuffer.trim().length > 0) {
|
||||
const pulled = pullSseDataEvents(`${sseBuffer}\n\n`);
|
||||
for (const eventPayload of pulled.events) {
|
||||
try {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(JSON.parse(eventPayload)));
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
reply.raw.end();
|
||||
}
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
tokenValue: selected.tokenValue,
|
||||
tokenName: selected.tokenName,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
requestStartedAtMs: startTime,
|
||||
requestEndedAtMs: startTime + latency,
|
||||
localLatencyMs: latency,
|
||||
usage: {
|
||||
promptTokens: parsedUsage.promptTokens,
|
||||
completionTokens: parsedUsage.completionTokens,
|
||||
totalTokens: parsedUsage.totalTokens,
|
||||
},
|
||||
});
|
||||
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
const rawText = await upstream.text();
|
||||
let data: any = rawText;
|
||||
try {
|
||||
data = JSON.parse(rawText);
|
||||
} catch {
|
||||
data = rawText;
|
||||
}
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(data);
|
||||
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
|
||||
if (failure) {
|
||||
const errText = failure.reason;
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
failure.status,
|
||||
latency,
|
||||
errText,
|
||||
retryCount,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
|
||||
if (shouldRetryProxyRequest(failure.status, errText) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: failure.reason,
|
||||
});
|
||||
|
||||
return reply.code(failure.status).send({
|
||||
error: { message: errText, type: 'upstream_error' },
|
||||
});
|
||||
}
|
||||
|
||||
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,
|
||||
},
|
||||
});
|
||||
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
|
||||
);
|
||||
return reply.send(data);
|
||||
} catch (err: any) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, logDownstreamApiKeyId ? downstreamApiKeyId : null);
|
||||
if (retryCount < MAX_RETRIES) {
|
||||
retryCount++;
|
||||
continue;
|
||||
}
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: err.message || 'network failure',
|
||||
@@ -233,48 +270,48 @@ export async function completionsProxyRoute(app: FastifyInstance) {
|
||||
});
|
||||
}
|
||||
|
||||
async function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
downstreamApiKeyId: number | null = null,
|
||||
promptTokens = 0,
|
||||
completionTokens = 0,
|
||||
totalTokens = 0,
|
||||
estimatedCost = 0,
|
||||
billingDetails: unknown = null,
|
||||
) {
|
||||
try {
|
||||
const createdAt = formatUtcSqlDateTime(new Date());
|
||||
const normalizedErrorMessage = composeProxyLogMessage({
|
||||
downstreamPath: '/v1/completions',
|
||||
errorMessage,
|
||||
});
|
||||
await db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
totalTokens,
|
||||
estimatedCost,
|
||||
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
|
||||
errorMessage: normalizedErrorMessage,
|
||||
retryCount,
|
||||
createdAt,
|
||||
}).run();
|
||||
} catch (error) {
|
||||
console.warn('[proxy/completions] failed to write proxy log', error);
|
||||
}
|
||||
}
|
||||
async function logProxy(
|
||||
selected: any,
|
||||
modelRequested: string,
|
||||
status: string,
|
||||
httpStatus: number,
|
||||
latencyMs: number,
|
||||
errorMessage: string | null,
|
||||
retryCount: number,
|
||||
downstreamApiKeyId: number | null = null,
|
||||
promptTokens = 0,
|
||||
completionTokens = 0,
|
||||
totalTokens = 0,
|
||||
estimatedCost = 0,
|
||||
billingDetails: unknown = null,
|
||||
) {
|
||||
try {
|
||||
const createdAt = formatUtcSqlDateTime(new Date());
|
||||
const normalizedErrorMessage = composeProxyLogMessage({
|
||||
downstreamPath: '/v1/completions',
|
||||
errorMessage,
|
||||
});
|
||||
await db.insert(schema.proxyLogs).values({
|
||||
routeId: selected.channel.routeId,
|
||||
channelId: selected.channel.id,
|
||||
accountId: selected.account.id,
|
||||
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
|
||||
modelRequested,
|
||||
modelActual: selected.actualModel,
|
||||
status,
|
||||
httpStatus,
|
||||
latencyMs,
|
||||
promptTokens,
|
||||
completionTokens,
|
||||
totalTokens,
|
||||
estimatedCost,
|
||||
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
|
||||
errorMessage: normalizedErrorMessage,
|
||||
retryCount,
|
||||
createdAt,
|
||||
}).run();
|
||||
} catch (error) {
|
||||
console.warn('[proxy/completions] failed to write proxy log', error);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,139 @@
|
||||
import { afterEach, describe, expect, it } from 'vitest';
|
||||
import { config } from '../../config.js';
|
||||
import { detectProxyFailure } from './proxyFailureJudge.js';
|
||||
|
||||
describe('detectProxyFailure (empty content)', () => {
|
||||
const originalEmptyFail = config.proxyEmptyContentFailEnabled;
|
||||
const originalKeywords = Array.isArray(config.proxyErrorKeywords) ? [...config.proxyErrorKeywords] : config.proxyErrorKeywords;
|
||||
|
||||
afterEach(() => {
|
||||
config.proxyEmptyContentFailEnabled = originalEmptyFail;
|
||||
config.proxyErrorKeywords = originalKeywords as any;
|
||||
});
|
||||
|
||||
it('flags empty assistant content even when total tokens > 0', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = JSON.stringify({
|
||||
id: 'chatcmpl_empty',
|
||||
object: 'chat.completion',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: '' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 12, completion_tokens: 0, total_tokens: 12 },
|
||||
});
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 12, completionTokens: 0, totalTokens: 12 },
|
||||
});
|
||||
|
||||
expect(failure).toMatchObject({ status: 502 });
|
||||
});
|
||||
|
||||
it('does not flag when output exists even if usage is missing', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = JSON.stringify({
|
||||
id: 'chatcmpl_has_output',
|
||||
object: 'chat.completion',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: 'hi' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
});
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
|
||||
});
|
||||
|
||||
expect(failure).toBeNull();
|
||||
});
|
||||
|
||||
it('does not treat tool call payloads as empty content', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = JSON.stringify({
|
||||
id: 'resp_tool',
|
||||
object: 'response',
|
||||
status: 'completed',
|
||||
output: [{
|
||||
type: 'function_call',
|
||||
id: 'fc_1',
|
||||
call_id: 'call_abc',
|
||||
name: 'Glob',
|
||||
arguments: '{"pattern":"README*"}',
|
||||
}],
|
||||
output_text: '',
|
||||
});
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
|
||||
});
|
||||
|
||||
expect(failure).toBeNull();
|
||||
});
|
||||
|
||||
it('flags empty SSE streams that contain no content deltas', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = [
|
||||
'data: {"id":"evt_1","choices":[{"delta":{}}]}',
|
||||
'',
|
||||
'data: [DONE]',
|
||||
'',
|
||||
'',
|
||||
].join('\n');
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
|
||||
});
|
||||
|
||||
expect(failure).toMatchObject({ status: 502 });
|
||||
});
|
||||
|
||||
it('treats DONE-only SSE as no output and flags failure', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = [
|
||||
'data: [DONE]',
|
||||
'',
|
||||
'',
|
||||
].join('\n');
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 },
|
||||
});
|
||||
|
||||
expect(failure).toMatchObject({ status: 502 });
|
||||
});
|
||||
|
||||
it('truncates fractional token counts before empty-content detection', () => {
|
||||
config.proxyEmptyContentFailEnabled = true;
|
||||
|
||||
const rawText = JSON.stringify({
|
||||
id: 'chatcmpl_fractional_empty',
|
||||
object: 'chat.completion',
|
||||
choices: [{
|
||||
index: 0,
|
||||
message: { role: 'assistant', content: '' },
|
||||
finish_reason: 'stop',
|
||||
}],
|
||||
usage: { prompt_tokens: 3.9, completion_tokens: 0.6, total_tokens: 4.5 },
|
||||
});
|
||||
|
||||
const failure = detectProxyFailure({
|
||||
rawText,
|
||||
usage: { promptTokens: 3.9, completionTokens: 0.6, totalTokens: 4.5 } as any,
|
||||
});
|
||||
|
||||
expect(failure).toMatchObject({ status: 502 });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,203 @@
|
||||
import { config } from '../../config.js';
|
||||
import { pullSseDataEvents } from '../../services/proxyUsageParser.js';
|
||||
|
||||
type FailureResult = {
|
||||
status: number;
|
||||
reason: string;
|
||||
};
|
||||
|
||||
type UsageSummary = {
|
||||
promptTokens?: number;
|
||||
completionTokens?: number;
|
||||
totalTokens?: number;
|
||||
};
|
||||
|
||||
function normalizeKeywords(values: string[]): string[] {
|
||||
return values
|
||||
.map((item) => (typeof item === 'string' ? item.trim() : ''))
|
||||
.filter((item) => item.length > 0)
|
||||
.map((item) => item.toLowerCase());
|
||||
}
|
||||
|
||||
function toNonNegativeInt(value: unknown): number {
|
||||
const n = typeof value === 'number' ? value : Number(value);
|
||||
if (!Number.isFinite(n)) return 0;
|
||||
return Math.max(0, Math.trunc(n));
|
||||
}
|
||||
|
||||
function hasNonEmptyString(value: unknown): boolean {
|
||||
return typeof value === 'string' && value.trim().length > 0;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object' && !Array.isArray(value);
|
||||
}
|
||||
|
||||
function hasToolCallLike(value: unknown): boolean {
|
||||
if (!value) return false;
|
||||
if (Array.isArray(value)) return value.length > 0;
|
||||
if (isRecord(value)) return Object.keys(value).length > 0;
|
||||
return false;
|
||||
}
|
||||
|
||||
function hasCompletionContentFromChoice(choice: any): boolean {
|
||||
if (hasNonEmptyString(choice?.text)) return true;
|
||||
if (hasNonEmptyString(choice?.completion)) return true;
|
||||
if (hasNonEmptyString(choice?.output_text)) return true;
|
||||
|
||||
const message = choice?.message;
|
||||
if (hasNonEmptyString(message?.content)) return true;
|
||||
if (Array.isArray(message?.content)) {
|
||||
for (const part of message.content) {
|
||||
if (hasNonEmptyString(part?.text) || hasNonEmptyString(part?.output_text) || hasNonEmptyString(part?.content)) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (hasNonEmptyString(message?.refusal)) return true;
|
||||
if (hasToolCallLike(message?.tool_calls)) return true;
|
||||
if (hasToolCallLike(message?.toolCalls)) return true;
|
||||
if (hasToolCallLike(message?.function_call)) return true;
|
||||
if (hasToolCallLike(message?.functionCall)) return true;
|
||||
|
||||
if (hasToolCallLike(choice?.tool_calls)) return true;
|
||||
if (hasToolCallLike(choice?.toolCalls)) return true;
|
||||
if (hasToolCallLike(choice?.function_call)) return true;
|
||||
if (hasToolCallLike(choice?.functionCall)) return true;
|
||||
|
||||
const delta = choice?.delta;
|
||||
if (hasNonEmptyString(delta?.content)) return true;
|
||||
if (hasNonEmptyString(delta?.refusal)) return true;
|
||||
if (hasToolCallLike(delta?.tool_calls)) return true;
|
||||
if (hasToolCallLike(delta?.toolCalls)) return true;
|
||||
if (hasToolCallLike(delta?.function_call)) return true;
|
||||
if (hasToolCallLike(delta?.functionCall)) return true;
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
function hasCompletionContentFromPayload(payload: unknown): boolean {
|
||||
if (!payload || typeof payload !== 'object') return false;
|
||||
const obj: any = payload;
|
||||
|
||||
if (Array.isArray(obj?.choices)) {
|
||||
for (const choice of obj.choices) {
|
||||
if (hasCompletionContentFromChoice(choice)) return true;
|
||||
}
|
||||
if (hasCompletionContentFromChoice(obj)) return true;
|
||||
}
|
||||
|
||||
if (hasNonEmptyString(obj?.output_text)) return true;
|
||||
if (hasNonEmptyString(obj?.outputText)) return true;
|
||||
|
||||
if (Array.isArray(obj?.output)) {
|
||||
for (const item of obj.output) {
|
||||
if (!isRecord(item)) continue;
|
||||
const type = String((item as any).type || '').toLowerCase();
|
||||
if (type.includes('function_call') || type.includes('tool_call')) return true;
|
||||
|
||||
if (hasNonEmptyString((item as any).text) || hasNonEmptyString((item as any).output_text)) return true;
|
||||
|
||||
if (Array.isArray((item as any).content)) {
|
||||
for (const part of (item as any).content) {
|
||||
if (hasNonEmptyString((part as any)?.text) || hasNonEmptyString((part as any)?.output_text) || hasNonEmptyString((part as any)?.content)) {
|
||||
return true;
|
||||
}
|
||||
const partType = String((part as any)?.type || '').toLowerCase();
|
||||
if (partType.includes('function_call') || partType.includes('tool_call')) return true;
|
||||
}
|
||||
}
|
||||
|
||||
if (hasToolCallLike((item as any).tool_calls) || hasToolCallLike((item as any).toolCalls)) return true;
|
||||
if (hasToolCallLike((item as any).function_call) || hasToolCallLike((item as any).functionCall)) return true;
|
||||
}
|
||||
}
|
||||
|
||||
if (Array.isArray(obj?.content)) {
|
||||
for (const part of obj.content) {
|
||||
if (hasNonEmptyString((part as any)?.text) || hasNonEmptyString((part as any)?.output_text) || hasNonEmptyString((part as any)?.content)) {
|
||||
return true;
|
||||
}
|
||||
const partType = String((part as any)?.type || '').toLowerCase();
|
||||
if (partType.includes('function_call') || partType.includes('tool_call')) return true;
|
||||
}
|
||||
}
|
||||
|
||||
if (hasNonEmptyString(obj?.delta)) return true;
|
||||
if (hasNonEmptyString(obj?.text)) return true;
|
||||
if (hasToolCallLike(obj?.tool_calls) || hasToolCallLike(obj?.toolCalls)) return true;
|
||||
if (hasToolCallLike(obj?.function_call) || hasToolCallLike(obj?.functionCall)) return true;
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
function detectHasUpstreamOutput(rawText: string): boolean {
|
||||
const text = typeof rawText === 'string' ? rawText : '';
|
||||
const trimmed = text.trim();
|
||||
if (!trimmed) return false;
|
||||
|
||||
try {
|
||||
const parsed = JSON.parse(trimmed);
|
||||
return hasCompletionContentFromPayload(parsed);
|
||||
} catch {
|
||||
// Important: don't trim before SSE parsing, otherwise `data: [DONE]\n\n` can be lost.
|
||||
const pulled = pullSseDataEvents(text);
|
||||
if (pulled.events.length > 0) {
|
||||
for (const event of pulled.events) {
|
||||
const payload = event.trim();
|
||||
if (!payload || payload === '[DONE]') continue;
|
||||
try {
|
||||
const parsedEvent = JSON.parse(payload);
|
||||
if (hasCompletionContentFromPayload(parsedEvent)) return true;
|
||||
} catch {
|
||||
// Non-JSON payload still counts as upstream output.
|
||||
return true;
|
||||
}
|
||||
}
|
||||
// SSE payloads exist but none contain output.
|
||||
return false;
|
||||
}
|
||||
|
||||
// Looks like SSE but contains no non-DONE payloads.
|
||||
if (text.includes('data:')) return false;
|
||||
|
||||
// Not JSON and not SSE: assume it's plain text output.
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
export function detectProxyFailure(input: {
|
||||
rawText: string;
|
||||
usage?: UsageSummary | null;
|
||||
// Backward-compatible fields (older call sites)
|
||||
completionTokens?: number;
|
||||
totalTokens?: number;
|
||||
}): FailureResult | null {
|
||||
const rawText = typeof input.rawText === 'string' ? input.rawText : '';
|
||||
const keywords = normalizeKeywords(config.proxyErrorKeywords || []);
|
||||
if (keywords.length > 0) {
|
||||
const normalizedText = rawText.toLowerCase();
|
||||
const matched = keywords.find((keyword) => normalizedText.includes(keyword));
|
||||
if (matched) {
|
||||
return {
|
||||
status: 502,
|
||||
reason: `Upstream response matched failure keyword: ${matched}`,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (config.proxyEmptyContentFailEnabled) {
|
||||
const completionTokens = toNonNegativeInt(input.usage?.completionTokens ?? input.completionTokens);
|
||||
const hasOutput = detectHasUpstreamOutput(rawText);
|
||||
|
||||
if (!hasOutput && completionTokens <= 0) {
|
||||
return {
|
||||
status: 502,
|
||||
reason: 'Upstream returned empty content',
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
@@ -16,6 +16,7 @@ import {
|
||||
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
|
||||
import { composeProxyLogMessage } from './logPathMeta.js';
|
||||
import { executeEndpointFlow, withUpstreamPath, type BuiltEndpointRequest } from './endpointFlow.js';
|
||||
import { detectProxyFailure } from './proxyFailureJudge.js';
|
||||
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
|
||||
import { resolveProxyLogBilling } from './proxyBilling.js';
|
||||
import { getProxyAuthContext, getProxyResourceOwner } from '../../middleware/auth.js';
|
||||
@@ -424,15 +425,140 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
const successfulUpstreamPath = endpointResult.upstreamPath;
|
||||
|
||||
if (isStream) {
|
||||
reply.hijack();
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
|
||||
const startSseResponse = () => {
|
||||
reply.hijack();
|
||||
reply.raw.statusCode = 200;
|
||||
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
|
||||
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
|
||||
reply.raw.setHeader('Connection', 'keep-alive');
|
||||
reply.raw.setHeader('X-Accel-Buffering', 'no');
|
||||
};
|
||||
|
||||
if (!upstreamContentType.includes('text/event-stream')) {
|
||||
const rawText = await upstream.text();
|
||||
let upstreamData: unknown = rawText;
|
||||
try {
|
||||
upstreamData = JSON.parse(rawText);
|
||||
} catch {
|
||||
upstreamData = rawText;
|
||||
}
|
||||
if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
|
||||
upstreamData = unwrapGeminiCliPayload(upstreamData);
|
||||
}
|
||||
|
||||
const parsedUsage = parseProxyUsage(upstreamData);
|
||||
const latency = Date.now() - startTime;
|
||||
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
|
||||
if (failure) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
failure.status,
|
||||
latency,
|
||||
failure.reason,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
|
||||
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: failure.reason,
|
||||
});
|
||||
return reply.code(failure.status).send({ error: { message: failure.reason, type: 'upstream_error' } });
|
||||
}
|
||||
|
||||
const normalized = openAiResponsesTransformer.transformFinalResponse(
|
||||
upstreamData,
|
||||
modelName,
|
||||
rawText,
|
||||
);
|
||||
const downstreamData = openAiResponsesTransformer.outbound.serializeFinal({
|
||||
upstreamPayload: upstreamData,
|
||||
normalized,
|
||||
usage: parsedUsage,
|
||||
serializationMode: 'response',
|
||||
});
|
||||
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,
|
||||
},
|
||||
});
|
||||
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
modelName: selected.actualModel || requestedModel,
|
||||
parsedUsage,
|
||||
resolvedUsage,
|
||||
});
|
||||
|
||||
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
|
||||
recordDownstreamCostUsage(request, estimatedCost);
|
||||
logProxy(
|
||||
selected, requestedModel, 'success', 200, latency, null, retryCount, downstreamPath,
|
||||
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
|
||||
const streamPayload = isRecord(downstreamData) ? downstreamData : {
|
||||
id: `resp_${Date.now()}`,
|
||||
object: 'response',
|
||||
status: 'completed',
|
||||
model: modelName,
|
||||
output: [],
|
||||
output_text: '',
|
||||
usage: {
|
||||
input_tokens: parsedUsage.promptTokens,
|
||||
output_tokens: parsedUsage.completionTokens,
|
||||
total_tokens: parsedUsage.totalTokens,
|
||||
},
|
||||
};
|
||||
const createdPayload = {
|
||||
...streamPayload,
|
||||
status: 'in_progress',
|
||||
output: [],
|
||||
output_text: '',
|
||||
};
|
||||
|
||||
startSseResponse();
|
||||
reply.raw.write(`event: response.created\ndata: ${JSON.stringify({ type: 'response.created', response: createdPayload })}\n\n`);
|
||||
reply.raw.write(`event: response.completed\ndata: ${JSON.stringify({ type: 'response.completed', response: streamPayload })}\n\n`);
|
||||
reply.raw.write('data: [DONE]\n\n');
|
||||
reply.raw.end();
|
||||
return;
|
||||
}
|
||||
|
||||
startSseResponse();
|
||||
|
||||
const upstreamReader = upstream.body?.getReader();
|
||||
const reader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
|
||||
const baseReader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
|
||||
? createGeminiCliStreamReader(upstreamReader)
|
||||
: upstreamReader;
|
||||
let parsedUsage: UsageSummary = {
|
||||
@@ -443,6 +569,25 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
cacheCreationTokens: 0,
|
||||
promptTokensIncludeCache: null,
|
||||
};
|
||||
let rawText = '';
|
||||
const decoder = new TextDecoder();
|
||||
const reader = baseReader
|
||||
? {
|
||||
async read() {
|
||||
const result = await baseReader.read();
|
||||
if (result.value) {
|
||||
rawText += decoder.decode(result.value, { stream: true });
|
||||
}
|
||||
return result;
|
||||
},
|
||||
async cancel(reason?: unknown) {
|
||||
return baseReader.cancel(reason);
|
||||
},
|
||||
releaseLock() {
|
||||
return baseReader.releaseLock();
|
||||
},
|
||||
}
|
||||
: baseReader;
|
||||
const writeLines = (lines: string[]) => {
|
||||
for (const line of lines) reply.raw.write(line);
|
||||
};
|
||||
@@ -464,6 +609,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
},
|
||||
});
|
||||
const streamResult = await streamSession.run(reader, reply.raw);
|
||||
rawText += decoder.decode();
|
||||
|
||||
const latency = Date.now() - startTime;
|
||||
if (streamResult.status === 'failed') {
|
||||
@@ -489,6 +635,11 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Once SSE has been hijacked and bytes may already be on the wire, we
|
||||
// must not attempt to convert stream failures into a fresh HTTP error
|
||||
// response or retry on another channel. Responses stream failures are
|
||||
// handled in-band by the proxy stream session.
|
||||
|
||||
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
|
||||
site: selected.site,
|
||||
account: selected.account,
|
||||
@@ -544,6 +695,38 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
}
|
||||
const latency = Date.now() - startTime;
|
||||
const parsedUsage = parseProxyUsage(upstreamData);
|
||||
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
|
||||
if (failure) {
|
||||
tokenRouter.recordFailure(selected.channel.id);
|
||||
logProxy(
|
||||
selected,
|
||||
requestedModel,
|
||||
'failed',
|
||||
failure.status,
|
||||
latency,
|
||||
failure.reason,
|
||||
retryCount,
|
||||
downstreamPath,
|
||||
parsedUsage.promptTokens,
|
||||
parsedUsage.completionTokens,
|
||||
parsedUsage.totalTokens,
|
||||
0,
|
||||
null,
|
||||
successfulUpstreamPath,
|
||||
clientContext,
|
||||
logDownstreamApiKeyId ? downstreamApiKeyId : null,
|
||||
);
|
||||
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
|
||||
retryCount += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
await reportProxyAllFailed({
|
||||
model: requestedModel,
|
||||
reason: failure.reason,
|
||||
});
|
||||
return reply.code(failure.status).send({ error: { message: failure.reason, type: 'upstream_error' } });
|
||||
}
|
||||
const normalized = openAiResponsesTransformer.transformFinalResponse(
|
||||
upstreamData,
|
||||
modelName,
|
||||
|
||||
@@ -1413,6 +1413,8 @@ export default function ModelTester() {
|
||||
setActiveDebugTab(DEBUG_TABS.RESPONSE);
|
||||
const reader = response.body.getReader();
|
||||
let doneReceived = false;
|
||||
let hasAnyContent = false;
|
||||
let hasAnyReasoning = false;
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
@@ -1447,6 +1449,12 @@ export default function ModelTester() {
|
||||
}
|
||||
|
||||
const delta = parseAnyStreamDelta(eventPayload);
|
||||
if (typeof delta.reasoningDelta === 'string' && delta.reasoningDelta.trim().length > 0) {
|
||||
hasAnyReasoning = true;
|
||||
}
|
||||
if (typeof delta.contentDelta === 'string' && delta.contentDelta.trim().length > 0) {
|
||||
hasAnyContent = true;
|
||||
}
|
||||
if (delta.reasoningDelta || delta.contentDelta) {
|
||||
setMessages((prev) => applyAssistantDelta(prev, {
|
||||
reasoningDelta: delta.reasoningDelta,
|
||||
@@ -1457,21 +1465,38 @@ export default function ModelTester() {
|
||||
}
|
||||
}
|
||||
|
||||
const emptyOutput = !hasAnyContent && !hasAnyReasoning;
|
||||
|
||||
setMessages((prev) => {
|
||||
const idx = findLastLoadingAssistantIndex(prev);
|
||||
if (idx === -1) return prev;
|
||||
const finalized = finalizeIncompleteMessage(prev[idx]);
|
||||
if (emptyOutput && !(finalized.content || '').trim() && !(finalized.reasoningContent || '').trim()) {
|
||||
return replaceMessageAt(prev, idx, {
|
||||
...finalized,
|
||||
content: '空回复（上游未返回任何内容）',
|
||||
status: MESSAGE_STATUS.ERROR,
|
||||
isThinkingComplete: true,
|
||||
});
|
||||
}
|
||||
return replaceMessageAt(prev, idx, {
|
||||
...finalizeIncompleteMessage(prev[idx]),
|
||||
...finalized,
|
||||
status: MESSAGE_STATUS.COMPLETE,
|
||||
isThinkingComplete: true,
|
||||
});
|
||||
});
|
||||
|
||||
setPendingPayload(null);
|
||||
setError('');
|
||||
pushDebug(doneReceived ? 'info' : 'warn', doneReceived
|
||||
? '流式传输已成功完成。'
|
||||
: '流式传输未收到 [DONE] 信号，已在本地完成。');
|
||||
if (emptyOutput) {
|
||||
const message = '上游返回空内容';
|
||||
setError(message);
|
||||
pushDebug('error', '流式传输完成但内容为空。');
|
||||
} else {
|
||||
setError('');
|
||||
pushDebug(doneReceived ? 'info' : 'warn', doneReceived
|
||||
? '流式传输已成功完成。'
|
||||
: '流式传输未收到 [DONE] 信号，已在本地完成。');
|
||||
}
|
||||
} catch (streamError: any) {
|
||||
const abortedByUser = controller.signal.aborted && streamStopRequestedRef.current;
|
||||
const abortedUnexpectedly = controller.signal.aborted
|
||||
@@ -1539,6 +1564,8 @@ export default function ModelTester() {
|
||||
|
||||
const reader = response.body.getReader();
|
||||
let doneReceived = false;
|
||||
let hasAnyContent = false;
|
||||
let hasAnyReasoning = false;
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
@@ -1571,6 +1598,12 @@ export default function ModelTester() {
|
||||
}
|
||||
|
||||
const delta = parseAnyStreamDelta(eventPayload);
|
||||
if (typeof delta.reasoningDelta === 'string' && delta.reasoningDelta.trim().length > 0) {
|
||||
hasAnyReasoning = true;
|
||||
}
|
||||
if (typeof delta.contentDelta === 'string' && delta.contentDelta.trim().length > 0) {
|
||||
hasAnyContent = true;
|
||||
}
|
||||
if (delta.reasoningDelta || delta.contentDelta) {
|
||||
setMessages((prev) => applyAssistantDelta(prev, {
|
||||
reasoningDelta: delta.reasoningDelta,
|
||||
@@ -1581,20 +1614,37 @@ export default function ModelTester() {
|
||||
}
|
||||
}
|
||||
|
||||
const emptyOutput = !hasAnyContent && !hasAnyReasoning;
|
||||
|
||||
setMessages((prev) => {
|
||||
const idx = findLastLoadingAssistantIndex(prev);
|
||||
if (idx === -1) return prev;
|
||||
const finalized = finalizeIncompleteMessage(prev[idx]);
|
||||
if (emptyOutput && !(finalized.content || '').trim() && !(finalized.reasoningContent || '').trim()) {
|
||||
return replaceMessageAt(prev, idx, {
|
||||
...finalized,
|
||||
content: '空回复（上游未返回任何内容）',
|
||||
status: MESSAGE_STATUS.ERROR,
|
||||
isThinkingComplete: true,
|
||||
});
|
||||
}
|
||||
return replaceMessageAt(prev, idx, {
|
||||
...finalizeIncompleteMessage(prev[idx]),
|
||||
...finalized,
|
||||
status: MESSAGE_STATUS.COMPLETE,
|
||||
isThinkingComplete: true,
|
||||
});
|
||||
});
|
||||
|
||||
setError('');
|
||||
pushDebug(doneReceived ? 'info' : 'warn', doneReceived
|
||||
? '代理流式传输已成功完成。'
|
||||
: '代理流式传输未收到 [DONE] 信号，已在本地完成。');
|
||||
if (emptyOutput) {
|
||||
const message = '上游返回空内容';
|
||||
setError(message);
|
||||
pushDebug('error', '代理流式传输完成但内容为空。');
|
||||
} else {
|
||||
setError('');
|
||||
pushDebug(doneReceived ? 'info' : 'warn', doneReceived
|
||||
? '代理流式传输已成功完成。'
|
||||
: '代理流式传输未收到 [DONE] 信号，已在本地完成。');
|
||||
}
|
||||
} catch (streamError: any) {
|
||||
const abortedByUser = controller.signal.aborted && streamStopRequestedRef.current;
|
||||
const abortedUnexpectedly = controller.signal.aborted
|
||||
|
||||
@@ -34,6 +34,8 @@ type RuntimeSettings = {
|
||||
routingFallbackUnitCost: number;
|
||||
routingWeights: RoutingWeights;
|
||||
systemProxyUrl: string;
|
||||
proxyErrorKeywords: string[];
|
||||
proxyEmptyContentFailEnabled: boolean;
|
||||
proxyTokenMasked?: string;
|
||||
adminIpAllowlist?: string[];
|
||||
currentAdminIp?: string;
|
||||
@@ -193,13 +195,17 @@ export default function Settings() {
|
||||
routingFallbackUnitCost: 1,
|
||||
routingWeights: defaultWeights,
|
||||
systemProxyUrl: '',
|
||||
proxyErrorKeywords: [],
|
||||
proxyEmptyContentFailEnabled: false,
|
||||
});
|
||||
const [proxyTokenSuffix, setProxyTokenSuffix] = useState('');
|
||||
const [proxyErrorKeywordsText, setProxyErrorKeywordsText] = useState('');
|
||||
const [maskedToken, setMaskedToken] = useState('');
|
||||
const [loading, setLoading] = useState(true);
|
||||
const [savingSchedule, setSavingSchedule] = useState(false);
|
||||
const [savingToken, setSavingToken] = useState(false);
|
||||
const [savingSystemProxy, setSavingSystemProxy] = useState(false);
|
||||
const [savingProxyFailureRules, setSavingProxyFailureRules] = useState(false);
|
||||
const [savingRouting, setSavingRouting] = useState(false);
|
||||
const [showAdvancedRouting, setShowAdvancedRouting] = useState(false);
|
||||
const [savingSecurity, setSavingSecurity] = useState(false);
|
||||
@@ -406,12 +412,21 @@ export default function Settings() {
|
||||
...(runtimeInfo.routingWeights || {}),
|
||||
},
|
||||
systemProxyUrl: typeof runtimeInfo.systemProxyUrl === 'string' ? runtimeInfo.systemProxyUrl : '',
|
||||
proxyErrorKeywords: Array.isArray(runtimeInfo.proxyErrorKeywords)
|
||||
? runtimeInfo.proxyErrorKeywords.filter((item: unknown) => typeof item === 'string')
|
||||
: [],
|
||||
proxyEmptyContentFailEnabled: !!runtimeInfo.proxyEmptyContentFailEnabled,
|
||||
proxyTokenMasked: runtimeInfo.proxyTokenMasked || '',
|
||||
adminIpAllowlist: Array.isArray(runtimeInfo.adminIpAllowlist)
|
||||
? runtimeInfo.adminIpAllowlist.filter((item: unknown) => typeof item === 'string')
|
||||
: [],
|
||||
currentAdminIp: typeof runtimeInfo.currentAdminIp === 'string' ? runtimeInfo.currentAdminIp : '',
|
||||
});
|
||||
setProxyErrorKeywordsText(
|
||||
Array.isArray(runtimeInfo.proxyErrorKeywords)
|
||||
? runtimeInfo.proxyErrorKeywords.filter((item: unknown) => typeof item === 'string').join('\n')
|
||||
: '',
|
||||
);
|
||||
setAdminIpAllowlistText(
|
||||
Array.isArray(runtimeInfo.adminIpAllowlist)
|
||||
? runtimeInfo.adminIpAllowlist.join('\n')
|
||||
@@ -463,6 +478,11 @@ export default function Settings() {
|
||||
return compact;
|
||||
};
|
||||
|
||||
const parseProxyErrorKeywords = (raw: string) => raw
|
||||
.split(/\r?\n|,/g)
|
||||
.map((item) => item.trim())
|
||||
.filter((item) => item.length > 0);
|
||||
|
||||
const saveSchedule = async () => {
|
||||
setSavingSchedule(true);
|
||||
try {
|
||||
@@ -521,6 +541,33 @@ export default function Settings() {
|
||||
}
|
||||
};
|
||||
|
||||
const saveProxyFailureRules = async () => {
|
||||
setSavingProxyFailureRules(true);
|
||||
try {
|
||||
const keywords = parseProxyErrorKeywords(proxyErrorKeywordsText);
|
||||
const res = await api.updateRuntimeSettings({
|
||||
proxyErrorKeywords: keywords,
|
||||
proxyEmptyContentFailEnabled: runtime.proxyEmptyContentFailEnabled,
|
||||
});
|
||||
const nextKeywords = Array.isArray(res?.proxyErrorKeywords)
|
||||
? res.proxyErrorKeywords
|
||||
: keywords;
|
||||
setRuntime((prev) => ({
|
||||
...prev,
|
||||
proxyErrorKeywords: nextKeywords,
|
||||
proxyEmptyContentFailEnabled: typeof res?.proxyEmptyContentFailEnabled === 'boolean'
|
||||
? res.proxyEmptyContentFailEnabled
|
||||
: prev.proxyEmptyContentFailEnabled,
|
||||
}));
|
||||
setProxyErrorKeywordsText(nextKeywords.join('\n'));
|
||||
toast.success('代理失败规则已保存');
|
||||
} catch (err: any) {
|
||||
toast.error(err?.message || '保存失败');
|
||||
} finally {
|
||||
setSavingProxyFailureRules(false);
|
||||
}
|
||||
};
|
||||
|
||||
const resetDownstreamForm = () => {
|
||||
setEditingDownstreamId(null);
|
||||
setDownstreamCreate({
|
||||
@@ -1007,6 +1054,38 @@ export default function Settings() {
|
||||
</button>
|
||||
</div>
|
||||
|
||||
<div className="card animate-slide-up stagger-4" style={{ padding: 20 }}>
|
||||
<div style={{ fontWeight: 600, fontSize: 14, marginBottom: 8 }}>代理失败判定</div>
|
||||
<div style={{ fontSize: 12, color: 'var(--color-text-muted)', marginBottom: 12 }}>
|
||||
命中任一关键词或空内容时判定失败，可触发重试。
|
||||
</div>
|
||||
<textarea
|
||||
value={proxyErrorKeywordsText}
|
||||
onChange={(e) => setProxyErrorKeywordsText(e.target.value)}
|
||||
placeholder="一行一个关键词，或逗号分隔"
|
||||
style={{
|
||||
...inputStyle,
|
||||
fontFamily: 'var(--font-mono)',
|
||||
minHeight: 96,
|
||||
resize: 'vertical',
|
||||
marginBottom: 12,
|
||||
}}
|
||||
/>
|
||||
<label style={{ display: 'inline-flex', alignItems: 'center', gap: 8, fontSize: 13, color: 'var(--color-text-secondary)', marginBottom: 12 }}>
|
||||
<input
|
||||
type="checkbox"
|
||||
checked={runtime.proxyEmptyContentFailEnabled}
|
||||
onChange={(e) => setRuntime((prev) => ({ ...prev, proxyEmptyContentFailEnabled: e.target.checked }))}
|
||||
/>
|
||||
空内容（completion=0，即使 prompt 有 token 也算）判定失败
|
||||
</label>
|
||||
<div>
|
||||
<button onClick={saveProxyFailureRules} disabled={savingProxyFailureRules} className="btn btn-primary">
|
||||
{savingProxyFailureRules ? <><span className="spinner spinner-sm" style={{ borderTopColor: 'white', borderColor: 'rgba(255,255,255,0.3)' }} /> 保存中...</> : '保存失败规则'}
|
||||
</button>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="card animate-slide-up stagger-4" style={{ padding: 20 }}>
|
||||
<div style={{ fontWeight: 600, fontSize: 14, marginBottom: 8 }}>下游访问令牌（PROXY_TOKEN）</div>
|
||||
<div style={{ fontSize: 12, color: 'var(--color-text-muted)', marginBottom: 12 }}>
|
||||
|
||||
Reference in New Issue
Block a user