feat: Add normalized types and utility functions for handling chat formats

- Introduced `NormalizedContentBlock`, `NormalizedUsage`, and related types for structured data handling.
- Implemented utility functions for creating and merging normalized usage statistics.
- Added a new module for shared transformers to manage chat formats.

feat: Implement app local state management

- Created functions to clear application installation state and manage local storage keys.
- Added constants for theme and user profile storage keys.

test: Add tests for BrandIcon component rendering

- Implemented tests to ensure BrandIcon does not force a colored background for recognized brands.

test: Enhance ModelAnalysisPanel tests

- Added tests for rendering token summaries with compact units in ModelAnalysisPanel.

test: Add tests for ModernSelect component

- Implemented tests to verify icon rendering for selected options in ModernSelect.

test: Add number formatting tests

- Created tests for `formatCompactTokenMetric` to validate token formatting logic.

feat: Implement account creation intent handling

- Added logic to handle session and API key creation intents in the Accounts page.

test: Add tests for dashboard performance card

- Implemented tests to verify rendering of performance metrics in the dashboard.

feat: Implement default connection segment resolution

- Added logic to determine initial connection segments based on platform types.

test: Add tests for settings factory reset functionality

- Implemented tests to verify the factory reset process and local state clearing.

test: Add tests for site creation redirect logic

- Implemented tests to verify redirection based on site capabilities during creation.

test: Add tests for token routes refresh decision action

- Implemented tests to verify refresh decision actions in TokenRoutes.

test: Add tests for token routes snapshot load

- Implemented tests to verify loading of cached snapshot data in TokenRoutes.
This commit is contained in:
cita
2026-03-08 12:08:57 +08:00
parent ed05b37626
commit 23922a5965
115 changed files with 18597 additions and 4411 deletions
@@ -0,0 +1,41 @@
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('proxy route architecture boundaries', () => {
it('keeps shared protocol helpers out of chat route', () => {
const source = readSource('./chat.ts');
expect(source).not.toContain("from './chatFormats.js'");
expect(source).toContain("from '../../transformers/openai/chat/index.js'");
expect(source).toContain("from '../../transformers/anthropic/messages/index.js'");
});
it('keeps anthropic-specific stream orchestration out of chat route', () => {
const source = readSource('./chat.ts');
expect(source).not.toContain('function syncClaudeStreamStateFromRawEvent(');
expect(source).not.toContain('function serializeRawSseEvent(');
expect(source).not.toContain('function isClaudeSseEventName(');
});
it('keeps responses protocol assembly out of responses route', () => {
const source = readSource('./responses.ts');
expect(source).not.toContain('function toResponsesPayload(');
expect(source).not.toContain('function createResponsesStreamState(');
expect(source).not.toContain('function convertResponsesBodyToOpenAiBody(');
});
it('removes normalizeContentText from upstream endpoint routing', () => {
const source = readSource('./upstreamEndpoint.ts');
expect(source).not.toContain('function normalizeContentText(');
expect(source).not.toContain('normalizeContentText(');
});
it('keeps gemini runtime closure in transformer-owned helpers', () => {
const source = readSource('./gemini.ts');
expect(source).toContain('serializeAggregateResponse');
expect(source).toContain('parseSsePayloads');
expect(source).toContain('aggregator.apply(aggregateState, parsed)');
});
});
+30 -110
View File
@@ -9,19 +9,7 @@ import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { resolveProxyUrlForSite, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import {
type DownstreamFormat,
createStreamTransformContext,
createClaudeDownstreamContext,
parseDownstreamChatRequest,
pullSseEventsWithDone,
normalizeUpstreamStreamEvent,
serializeNormalizedStreamEvent,
serializeStreamDone,
normalizeUpstreamFinalResponse,
serializeFinalResponse,
buildSyntheticOpenAiChunks,
} from './chatFormats.js';
import { type DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
buildMinimalJsonHeadersForCompatibility,
buildUpstreamEndpointRequest,
@@ -38,19 +26,16 @@ import { composeProxyLogMessage } from './logPathMeta.js';
import { executeEndpointFlow, withUpstreamPath } from './endpointFlow.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveProxyLogBilling } from './proxyBilling.js';
import { openAiChatTransformer } from '../../transformers/openai/chat/index.js';
import {
anthropicMessagesTransformer,
isAnthropicRawSseEventName,
serializeAnthropicFinalAsStream,
serializeAnthropicRawSseEvent,
syncAnthropicRawStreamStateFromEvent,
} from '../../transformers/anthropic/messages/index.js';
const MAX_RETRIES = 2;
const CLAUDE_SSE_EVENT_NAMES = new Set([
'message_start',
'content_block_start',
'content_block_delta',
'content_block_stop',
'message_delta',
'message_stop',
'ping',
'error',
]);
function shouldRetryClaudeMessagesWithNormalizedBody(
downstreamFormat: DownstreamFormat,
endpointPath: string,
@@ -71,53 +56,6 @@ function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function isClaudeSseEventName(value: unknown): value is string {
return typeof value === 'string' && CLAUDE_SSE_EVENT_NAMES.has(value);
}
function serializeRawSseEvent(event: string, data: string): string {
const dataLines = data.split('\n').map((line) => `data: ${line}`).join('\n');
if (event) {
return `event: ${event}\n${dataLines}\n\n`;
}
return `${dataLines}\n\n`;
}
function syncClaudeStreamStateFromRawEvent(
eventName: string,
parsedPayload: unknown,
streamContext: { id: string; model: string },
claudeContext: { messageStarted: boolean; contentBlockStarted: boolean; doneSent: boolean },
) {
if (eventName === 'message_start') {
claudeContext.messageStarted = true;
if (isRecord(parsedPayload) && isRecord(parsedPayload.message)) {
const message = parsedPayload.message;
if (typeof message.id === 'string' && message.id.trim().length > 0) {
streamContext.id = message.id;
}
if (typeof message.model === 'string' && message.model.trim().length > 0) {
streamContext.model = message.model;
}
}
return;
}
if (eventName === 'content_block_start') {
claudeContext.contentBlockStarted = true;
return;
}
if (eventName === 'content_block_stop') {
claudeContext.contentBlockStarted = false;
return;
}
if (eventName === 'message_stop') {
claudeContext.doneSent = true;
}
}
export async function chatProxyRoute(app: FastifyInstance) {
app.post('/v1/chat/completions', async (request: FastifyRequest, reply: FastifyReply) =>
handleChatProxyRequest(request, reply, 'openai'));
@@ -133,7 +71,10 @@ async function handleChatProxyRequest(
reply: FastifyReply,
downstreamFormat: DownstreamFormat,
) {
const parsedRequest = parseDownstreamChatRequest(request.body, downstreamFormat);
const downstreamTransformer = downstreamFormat === 'claude'
? anthropicMessagesTransformer
: openAiChatTransformer;
const parsedRequest = downstreamTransformer.transformRequest(request.body);
if (parsedRequest.error) {
return reply.code(parsedRequest.error.statusCode).send(parsedRequest.error.payload);
}
@@ -345,8 +286,8 @@ async function handleChatProxyRequest(
reply.raw.setHeader('Connection', 'keep-alive');
reply.raw.setHeader('X-Accel-Buffering', 'no');
const streamContext = createStreamTransformContext(modelName);
const claudeContext = createClaudeDownstreamContext();
const streamContext = downstreamTransformer.createStreamContext(modelName);
const claudeContext = anthropicMessagesTransformer.createDownstreamContext();
let parsedUsage: ReturnType<typeof parseProxyUsage> = {
promptTokens: 0,
completionTokens: 0,
@@ -363,38 +304,24 @@ async function handleChatProxyRequest(
};
const writeDone = () => {
writeLines(serializeStreamDone(downstreamFormat, streamContext, claudeContext));
writeLines(downstreamTransformer.serializeDone(streamContext, claudeContext));
};
const emitNormalizedFinalAsStream = (upstreamData: unknown, fallbackText = '') => {
const normalizedFinal = normalizeUpstreamFinalResponse(upstreamData, modelName, fallbackText);
const normalizedFinal = downstreamTransformer.transformFinalResponse(upstreamData, modelName, fallbackText);
streamContext.id = normalizedFinal.id;
streamContext.model = normalizedFinal.model;
streamContext.created = normalizedFinal.created;
if (downstreamFormat === 'openai') {
const syntheticChunks = buildSyntheticOpenAiChunks(normalizedFinal);
const syntheticChunks = openAiChatTransformer.buildSyntheticChunks(normalizedFinal);
for (const chunk of syntheticChunks) {
reply.raw.write(`data: ${JSON.stringify(chunk)}\n\n`);
}
return;
}
writeLines(serializeNormalizedStreamEvent('claude', { role: 'assistant' }, streamContext, claudeContext));
const combinedText = [normalizedFinal.reasoningContent, normalizedFinal.content]
.filter((item) => typeof item === 'string' && item.trim().length > 0)
.join('\n\n');
if (combinedText) {
writeLines(serializeNormalizedStreamEvent('claude', {
contentDelta: combinedText,
}, streamContext, claudeContext));
}
writeLines(serializeNormalizedStreamEvent('claude', {
finishReason: normalizedFinal.finishReason,
}, streamContext, claudeContext));
writeLines(serializeAnthropicFinalAsStream(normalizedFinal, streamContext, claudeContext));
};
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
@@ -470,7 +397,7 @@ async function handleChatProxyRequest(
let shouldTerminateEarly = false;
const consumeSseBuffer = (incoming: string): string => {
const pulled = pullSseEventsWithDone(incoming);
const pulled = downstreamTransformer.pullSseEvents(incoming);
for (const eventBlock of pulled.events) {
if (eventBlock.data === '[DONE]') {
writeDone();
@@ -487,23 +414,22 @@ async function handleChatProxyRequest(
if (parsedPayload && typeof parsedPayload === 'object') {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
if (downstreamFormat === 'claude') {
const payloadType = (isRecord(parsedPayload) && typeof parsedPayload.type === 'string')
? parsedPayload.type
: '';
const claudeEventName = isClaudeSseEventName(eventBlock.event)
const claudeEventName = isAnthropicRawSseEventName(eventBlock.event)
? eventBlock.event
: (isClaudeSseEventName(payloadType) ? payloadType : '');
: (isAnthropicRawSseEventName(payloadType) ? payloadType : '');
if (claudeEventName) {
syncClaudeStreamStateFromRawEvent(
syncAnthropicRawStreamStateFromEvent(
claudeEventName,
parsedPayload,
streamContext,
claudeContext,
);
reply.raw.write(serializeRawSseEvent(claudeEventName, eventBlock.data));
reply.raw.write(serializeAnthropicRawSseEvent(claudeEventName, eventBlock.data));
if (claudeContext.doneSent) {
shouldTerminateEarly = true;
break;
@@ -511,19 +437,13 @@ async function handleChatProxyRequest(
continue;
}
}
const normalizedEvent = normalizeUpstreamStreamEvent(parsedPayload, streamContext, modelName);
writeLines(serializeNormalizedStreamEvent(
downstreamFormat,
normalizedEvent,
streamContext,
claudeContext,
));
const normalizedEvent = downstreamTransformer.transformStreamEvent(parsedPayload, streamContext, modelName);
writeLines(downstreamTransformer.serializeStreamEvent(normalizedEvent, streamContext, claudeContext));
if (downstreamFormat === 'claude' && claudeContext.doneSent) {
shouldTerminateEarly = true;
break;
}
if (downstreamFormat === 'openai' && streamContext.doneSent) {
if (streamContext.doneSent) {
shouldTerminateEarly = true;
break;
}
@@ -533,7 +453,7 @@ async function handleChatProxyRequest(
if (downstreamFormat === 'openai') {
reply.raw.write(`data: ${eventBlock.data}\n\n`);
} else {
writeLines(serializeNormalizedStreamEvent('claude', {
writeLines(anthropicMessagesTransformer.serializeStreamEvent({
contentDelta: eventBlock.data,
}, streamContext, claudeContext));
if (claudeContext.doneSent) {
@@ -628,8 +548,8 @@ async function handleChatProxyRequest(
const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(upstreamData);
const normalizedFinal = normalizeUpstreamFinalResponse(upstreamData, modelName, rawText);
const downstreamResponse = serializeFinalResponse(downstreamFormat, normalizedFinal, parsedUsage);
const normalizedFinal = downstreamTransformer.transformFinalResponse(upstreamData, modelName, rawText);
const downstreamResponse = downstreamTransformer.serializeFinalResponse(normalizedFinal, parsedUsage);
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
File diff suppressed because it is too large Load Diff
+430
View File
@@ -0,0 +1,430 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const invalidateTokenRouterCacheMock = vi.fn();
const authorizeDownstreamTokenMock = vi.fn();
const consumeManagedKeyRequestMock = vi.fn();
vi.mock('undici', () => ({
fetch: (...args: unknown[]) => fetchMock(...args),
}));
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
explainSelection: vi.fn(async () => ({ selectedChannelId: 11 })),
},
invalidateTokenRouterCache: (...args: unknown[]) => invalidateTokenRouterCacheMock(...args),
}));
vi.mock('../../services/downstreamApiKeyService.js', () => ({
authorizeDownstreamToken: (...args: unknown[]) => authorizeDownstreamTokenMock(...args),
consumeManagedKeyRequest: (...args: unknown[]) => consumeManagedKeyRequestMock(...args),
}));
describe('gemini native proxy routes', () => {
let app: FastifyInstance;
beforeAll(async () => {
const { proxyRoutes } = await import('./router.js');
app = Fastify();
await app.register(proxyRoutes);
});
beforeEach(() => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
authorizeDownstreamTokenMock.mockReset();
consumeManagedKeyRequestMock.mockReset();
authorizeDownstreamTokenMock.mockResolvedValue({
ok: true,
source: 'global',
token: 'sk-managed-gemini',
policy: {},
});
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'gemini-site', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'gemini-key',
actualModel: 'gemini-2.5-flash',
});
selectNextChannelMock.mockReturnValue(null);
});
afterAll(async () => {
await app.close();
});
it('accepts x-goog-api-key on /v1beta/models and returns gemini model list shape', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
models: [
{ name: 'models/gemini-2.5-flash', displayName: 'Gemini 2.5 Flash' },
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1beta/models',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
});
expect(response.statusCode).toBe(200);
expect(response.json()).toEqual({
models: [
{
name: 'models/gemini-2.5-flash',
displayName: 'Gemini 2.5 Flash',
},
],
});
});
it('forwards native generateContent requests through the gemini route group', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
candidates: [
{
content: {
parts: [{ text: 'hello from gemini' }],
role: 'model',
},
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
},
});
expect(response.statusCode).toBe(200);
const [targetUrl, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toContain('/v1beta/models/gemini-2.5-flash:generateContent');
expect(targetUrl).toContain('key=gemini-key');
expect(JSON.parse(String(requestInit.body))).toEqual({
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
});
expect(response.json()).toEqual({
responseId: '',
modelVersion: '',
candidates: [
{
index: 0,
content: {
parts: [{ text: 'hello from gemini' }],
role: 'model',
},
finishReason: 'STOP',
},
],
});
});
it('preserves structured Gemini-native fields instead of narrowing them to a bare passthrough shell', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
candidates: [
{
content: {
parts: [{ text: 'ok' }],
role: 'model',
},
finishReason: 'STOP',
},
],
usageMetadata: {
promptTokenCount: 10,
candidatesTokenCount: 5,
totalTokenCount: 15,
cachedContentTokenCount: 2,
thoughtsTokenCount: 1,
},
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
systemInstruction: {
parts: [{ text: 'be concise' }],
},
cachedContent: 'cached/abc123',
safetySettings: [{ category: 'HARM_CATEGORY_HARASSMENT', threshold: 'BLOCK_NONE' }],
generationConfig: {
responseModalities: ['TEXT'],
responseMimeType: 'application/json',
temperature: 0.2,
topP: 0.8,
topK: 20,
maxOutputTokens: 256,
thinkingConfig: { thinkingBudget: 512 },
imageConfig: { aspectRatio: '1:1' },
},
tools: [
{
functionDeclarations: [
{
name: 'lookup_weather',
description: 'look up weather',
},
],
},
],
contents: [
{
role: 'user',
parts: [{ text: 'weather in shanghai' }],
},
],
},
});
expect(response.statusCode).toBe(200);
const [, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(JSON.parse(String(requestInit.body))).toEqual({
systemInstruction: {
parts: [{ text: 'be concise' }],
},
cachedContent: 'cached/abc123',
safetySettings: [{ category: 'HARM_CATEGORY_HARASSMENT', threshold: 'BLOCK_NONE' }],
generationConfig: {
responseModalities: ['TEXT'],
responseMimeType: 'application/json',
temperature: 0.2,
topP: 0.8,
topK: 20,
maxOutputTokens: 256,
thinkingConfig: { thinkingBudget: 512 },
imageConfig: { aspectRatio: '1:1' },
},
tools: [
{
functionDeclarations: [
{
name: 'lookup_weather',
description: 'look up weather',
},
],
},
],
contents: [
{
role: 'user',
parts: [{ text: 'weather in shanghai' }],
},
],
});
});
it('aggregates non-sse json-array streaming payloads into a final Gemini response', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify([
{
candidates: [
{
content: { role: 'model', parts: [{ text: 'first' }] },
groundingMetadata: { source: 'web' },
},
],
},
{
candidates: [
{
content: { role: 'model', parts: [{ text: 'second', thoughtSignature: 'sig-1' }] },
citationMetadata: { citations: [{ startIndex: 0, endIndex: 5 }] },
},
],
usageMetadata: {
promptTokenCount: 11,
candidatesTokenCount: 6,
totalTokenCount: 17,
cachedContentTokenCount: 2,
thoughtsTokenCount: 3,
},
},
]), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:streamGenerateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
},
});
expect(response.statusCode).toBe(200);
expect(response.json()).toEqual({
responseId: '',
modelVersion: '',
candidates: [
{
index: 0,
finishReason: 'STOP',
content: {
role: 'model',
parts: [
{ text: 'first' },
{ text: 'second', thoughtSignature: 'sig-1' },
],
},
groundingMetadata: { source: 'web' },
citationMetadata: { citations: [{ startIndex: 0, endIndex: 5 }] },
},
],
usageMetadata: {
promptTokenCount: 11,
candidatesTokenCount: 6,
totalTokenCount: 17,
cachedContentTokenCount: 2,
thoughtsTokenCount: 3,
},
});
});
it('derives gemini-3 thinkingLevel from OpenAI-style reasoning inputs in the runtime request path', async () => {
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'gemini-site', url: 'https://generativelanguage.googleapis.com', platform: 'gemini' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'gemini-key',
actualModel: 'gemini-3-pro',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
candidates: [
{
content: {
parts: [{ text: 'ok' }],
role: 'model',
},
finishReason: 'STOP',
},
],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-3-pro:generateContent',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
reasoning_effort: 'high',
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
},
});
expect(response.statusCode).toBe(200);
const [, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(JSON.parse(String(requestInit.body))).toEqual({
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
generationConfig: {
thinkingConfig: { thinkingLevel: 'high' },
},
});
});
it('aggregates SSE payloads through the transformer instead of raw passthrough', async () => {
const encoder = new TextEncoder();
const upstreamBody = new ReadableStream<Uint8Array>({
start(controller) {
controller.enqueue(encoder.encode('data: {"responseId":"resp-sse","modelVersion":"gemini-2.5-flash","candidates":[{"content":{"role":"model","parts":[{"text":"first"}]},"groundingMetadata":{"source":"web"}}]}\r\n\r\n'));
controller.enqueue(encoder.encode('data: {"candidates":[{"content":{"role":"model","parts":[{"text":"second","thoughtSignature":"sig-1"}]},"citationMetadata":{"citations":[{"startIndex":0,"endIndex":5}]},"finishReason":"STOP"}],"usageMetadata":{"promptTokenCount":11,"candidatesTokenCount":6,"totalTokenCount":17,"cachedContentTokenCount":2,"thoughtsTokenCount":3}}\r\n\r\n'));
controller.close();
},
});
fetchMock.mockResolvedValue(new Response(upstreamBody, {
status: 200,
headers: { 'content-type': 'text/event-stream; charset=utf-8' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1beta/models/gemini-2.5-flash:streamGenerateContent?alt=sse',
headers: {
'x-goog-api-key': 'sk-managed-gemini',
},
payload: {
contents: [
{
role: 'user',
parts: [{ text: 'hello' }],
},
],
},
});
expect(response.statusCode).toBe(200);
expect(response.headers['content-type']).toContain('text/event-stream');
expect(response.body).toContain('data: {"responseId":"resp-sse"');
expect(response.body).toContain('"groundingMetadata":{"source":"web"}');
expect(response.body).toContain('"citationMetadata":{"citations":[{"startIndex":0,"endIndex":5}]}');
expect(response.body).toContain('"cachedContentTokenCount":2');
expect(response.body).toContain('"thoughtsTokenCount":3');
expect(response.body).toContain('"parts":[{"text":"first"},{"text":"second","thoughtSignature":"sig-1"}]');
expect(response.body).not.toContain('\r\n\r\n');
});
});
+196
View File
@@ -0,0 +1,196 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { TextDecoder } from 'node:util';
import { fetch } from 'undici';
import { tokenRouter } from '../../services/tokenRouter.js';
import { getDownstreamRoutingPolicy } from './downstreamPolicy.js';
import {
geminiGenerateContentTransformer,
} from '../../transformers/gemini/generate-content/index.js';
const GEMINI_MODEL_PROBES = [
'gemini-2.5-flash',
'gemini-2.0-flash',
'gemini-1.5-flash',
'gemini-pro',
];
async function selectGeminiChannel(request: FastifyRequest) {
const policy = getDownstreamRoutingPolicy(request);
for (const candidate of GEMINI_MODEL_PROBES) {
const selected = await tokenRouter.selectChannel(candidate, policy);
if (selected) return selected;
}
return null;
}
function resolveGeminiApiVersion(request: FastifyRequest): string {
const params = request.params as { geminiApiVersion?: string } | undefined;
return (params?.geminiApiVersion || 'v1beta').trim() || 'v1beta';
}
function getRawRequestUrl(request: FastifyRequest): string {
return request.raw.url || request.url || '';
}
function extractGeminiModelActionPath(request: FastifyRequest, apiVersion: string): string {
const rawUrl = getRawRequestUrl(request);
const withoutQuery = rawUrl.split('?')[0] || rawUrl;
const normalizedVersion = apiVersion.replace(/^\/+/, '');
const geminiPrefix = `/gemini/${normalizedVersion}/`;
const aliasPrefix = `/${normalizedVersion}/`;
if (withoutQuery.startsWith(geminiPrefix)) {
return withoutQuery.slice(geminiPrefix.length);
}
if (withoutQuery.startsWith(aliasPrefix)) {
return withoutQuery.slice(aliasPrefix.length);
}
return withoutQuery.replace(/^\/+/, '');
}
export async function geminiProxyRoute(app: FastifyInstance) {
const listModels = async (request: FastifyRequest, reply: FastifyReply) => {
const selected = await selectGeminiChannel(request);
if (!selected) {
return reply.code(503).send({
error: { message: 'No available channels for Gemini models', type: 'server_error' },
});
}
const apiVersion = resolveGeminiApiVersion(request);
const upstream = await fetch(
geminiGenerateContentTransformer.resolveModelsUrl(selected.site.url, apiVersion, selected.tokenValue),
{ method: 'GET' },
);
const text = await upstream.text();
try {
return reply.code(upstream.status).send(JSON.parse(text));
} catch {
return reply.code(upstream.status).type(upstream.headers.get('content-type') || 'application/json').send(text);
}
};
const generateContent = async (request: FastifyRequest, reply: FastifyReply) => {
const apiVersion = resolveGeminiApiVersion(request);
const modelActionPath = extractGeminiModelActionPath(request, apiVersion);
const requestedModel = modelActionPath.replace(/^models\//, '').split(':')[0].trim();
if (!requestedModel) {
return reply.code(400).send({
error: { message: 'Gemini model path is required', type: 'invalid_request_error' },
});
}
const policy = getDownstreamRoutingPolicy(request);
const selected = await tokenRouter.selectChannel(requestedModel, policy);
if (!selected) {
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
const body = geminiGenerateContentTransformer.inbound.normalizeRequest(
request.body || {},
selected.actualModel || requestedModel,
);
const actualModelAction = modelActionPath.replace(
/^models\/[^:]+/,
`models/${selected.actualModel || requestedModel}`,
);
const query = new URLSearchParams(request.query as Record<string, string>).toString();
const upstream = await fetch(
geminiGenerateContentTransformer.resolveActionUrl(
selected.site.url,
apiVersion,
actualModelAction,
selected.tokenValue,
query,
),
{
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(body),
},
);
const contentType = upstream.headers.get('content-type');
if (geminiGenerateContentTransformer.stream.isSseContentType(contentType)) {
reply.hijack();
reply.raw.statusCode = upstream.status;
reply.raw.setHeader('Content-Type', contentType || 'text/event-stream');
const reader = upstream.body?.getReader();
if (!reader) {
reply.raw.end();
return;
}
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
const decoder = new TextDecoder();
let rest = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
const chunkText = decoder.decode(value, { stream: true });
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + chunkText);
rest = parsed.rest;
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
const tail = decoder.decode();
if (tail) {
const parsed = geminiGenerateContentTransformer.stream.parseSsePayloads(rest + tail);
for (const event of parsed.events) {
geminiGenerateContentTransformer.stream.applyAggregate(aggregateState, event);
reply.raw.write(
geminiGenerateContentTransformer.stream.serializeSsePayload(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
),
);
}
}
} finally {
reader.releaseLock();
reply.raw.end();
}
return;
}
const text = await upstream.text();
try {
const parsed = JSON.parse(text);
if (!upstream.ok) {
return reply.code(upstream.status).send(parsed);
}
const aggregateState = geminiGenerateContentTransformer.aggregator.createState();
if (Array.isArray(parsed)) {
for (const chunk of geminiGenerateContentTransformer.stream.parseJsonArrayPayload(parsed)) {
geminiGenerateContentTransformer.aggregator.apply(aggregateState, chunk);
}
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
);
}
geminiGenerateContentTransformer.aggregator.apply(aggregateState, parsed);
return reply.code(upstream.status).send(
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
);
} catch {
return reply.code(upstream.status).type(contentType || 'application/json').send(text);
}
};
app.get('/v1beta/models', listModels);
app.get('/gemini/:geminiApiVersion/models', listModels);
app.post('/v1beta/models/*', generateContent);
app.post('/gemini/:geminiApiVersion/models/*', generateContent);
}
@@ -0,0 +1,160 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordSuccessMock = vi.fn();
const recordFailureMock = vi.fn();
const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async () => 0);
const dbInsertMock = vi.fn((_arg?: any) => ({
values: () => ({
run: () => undefined,
}),
}));
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
return {
...actual,
fetch: (...args: unknown[]) => fetchMock(...args),
};
});
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
},
}));
vi.mock('../../services/modelService.js', () => ({
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
}));
vi.mock('../../services/alertService.js', () => ({
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
}));
vi.mock('../../services/alertRules.js', () => ({
isTokenExpiredError: () => false,
}));
vi.mock('../../services/modelPricingService.js', () => ({
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
}));
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
schema: {
proxyLogs: {},
},
}));
function buildMultipartBody(boundary: string) {
return Buffer.from(
`--${boundary}\r\n`
+ `Content-Disposition: form-data; name="model"\r\n\r\n`
+ `gpt-image-1\r\n`
+ `--${boundary}\r\n`
+ `Content-Disposition: form-data; name="prompt"\r\n\r\n`
+ `edit this\r\n`
+ `--${boundary}\r\n`
+ `Content-Disposition: form-data; name="image"; filename="cat.png"\r\n`
+ `Content-Type: image/png\r\n\r\n`
+ `pngdata\r\n`
+ `--${boundary}--\r\n`,
);
}
describe('/v1/images/edits route', () => {
let app: FastifyInstance;
beforeAll(async () => {
const { imagesProxyRoute } = await import('./images.js');
app = Fastify();
await app.register(imagesProxyRoute);
});
beforeEach(() => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordSuccessMock.mockReset();
recordFailureMock.mockReset();
refreshModelsAndRebuildRoutesMock.mockReset();
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
dbInsertMock.mockClear();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'demo-site', url: 'https://upstream.example.com', platform: 'openai' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'upstream-gpt-image',
});
selectNextChannelMock.mockReturnValue(null);
});
afterAll(async () => {
await app.close();
});
it('accepts multipart image edit requests and forwards them to /v1/images/edits', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
created: 1,
data: [{ b64_json: 'iVBORw0KGgo=' }],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const boundary = 'metapi-boundary';
const response = await app.inject({
method: 'POST',
url: '/v1/images/edits',
headers: {
authorization: 'Bearer sk-demo',
'content-type': `multipart/form-data; boundary=${boundary}`,
},
payload: buildMultipartBody(boundary),
});
expect(response.statusCode).toBe(200);
const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://upstream.example.com/v1/images/edits');
});
it('returns explicit not-supported error for /v1/images/variations', async () => {
const response = await app.inject({
method: 'POST',
url: '/v1/images/variations',
payload: {
model: 'gpt-image-1',
},
});
expect(response.statusCode).toBe(400);
expect(response.json()).toEqual({
error: {
message: 'Image variations are not supported',
type: 'invalid_request_error',
},
});
});
});
+143 -11
View File
@@ -11,11 +11,14 @@ import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordD
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
const MAX_RETRIES = 2;
export async function imagesProxyRoute(app: FastifyInstance) {
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
export async function imagesProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app);
app.post('/v1/images/generations', async (request: FastifyRequest, reply: FastifyReply) => {
const body = request.body as any;
const requestedModel = body?.model || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
@@ -112,25 +115,154 @@ export async function imagesProxyRoute(app: FastifyInstance) {
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
});
}
}
});
}
}
}
});
app.post('/v1/images/edits', async (request: FastifyRequest, reply: FastifyReply) => {
const multipartForm = await parseMultipartFormData(request);
const jsonBody = (!multipartForm && request.body && typeof request.body === 'object')
? request.body as Record<string, unknown>
: null;
const requestedModel = typeof multipartForm?.get('model') === 'string'
? String(multipartForm.get('model')).trim()
: (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '') || 'gpt-image-1';
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
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({
model: requestedModel,
reason: 'No available channels after retries',
});
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = `${selected.site.url}/v1/images/edits`;
const startTime = Date.now();
try {
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel,
}) as any,
})
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: selected.actualModel || requestedModel,
}),
});
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount, 0, '/v1/images/edits');
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
let data: any = {};
try { data = JSON.parse(text); } catch { data = { data: [] }; }
const latency = Date.now() - startTime;
const estimatedCost = await estimateProxyCost({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount, estimatedCost, '/v1/images/edits');
return reply.code(upstream.status).send(data);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, err.message, retryCount, 0, '/v1/images/edits');
if (retryCount < MAX_RETRIES) {
retryCount++;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
});
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
});
}
}
});
app.post('/v1/images/variations', async (_request: FastifyRequest, reply: FastifyReply) => {
return reply.code(400).send({
error: {
message: 'Image variations are not supported',
type: 'invalid_request_error',
},
});
});
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
estimatedCost = 0,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
estimatedCost = 0,
downstreamPath = '/v1/images/generations',
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath: '/v1/images/generations',
downstreamPath,
errorMessage,
});
await db.insert(schema.proxyLogs).values({
+86
View File
@@ -284,4 +284,90 @@ describe('/v1/models route', () => {
expect(ids).not.toContain('claude-opus-4-5');
expect(ids).not.toContain('claude-sonnet-4-5');
});
it('filters search pseudo models out of /v1/models', async () => {
const site = await db.insert(schema.sites).values({
name: 'search-site',
url: 'https://search.example.com',
platform: 'openai',
status: 'active',
}).returning().get();
const account = await db.insert(schema.accounts).values({
siteId: site.id,
accessToken: 'search-access-token',
status: 'active',
}).returning().get();
const token = await db.insert(schema.accountTokens).values({
accountId: account.id,
name: 'default',
token: 'search-api-token',
enabled: true,
isDefault: true,
}).returning().get();
await db.insert(schema.modelAvailability).values([
{
accountId: account.id,
modelName: '__search',
available: true,
},
{
accountId: account.id,
modelName: '__tavily_search',
available: true,
},
{
accountId: account.id,
modelName: 'gpt-4.1',
available: true,
},
]).run();
const searchRoute = await db.insert(schema.tokenRoutes).values({
modelPattern: '__search',
enabled: true,
}).returning().get();
const llmRoute = await db.insert(schema.tokenRoutes).values({
modelPattern: 'gpt-4.1',
enabled: true,
}).returning().get();
await db.insert(schema.routeChannels).values([
{
routeId: searchRoute.id,
accountId: account.id,
tokenId: token.id,
sourceModel: '__search',
enabled: true,
},
{
routeId: llmRoute.id,
accountId: account.id,
tokenId: token.id,
sourceModel: 'gpt-4.1',
enabled: true,
},
]).run();
const response = await app.inject({
method: 'GET',
url: '/v1/models',
headers: {
authorization: 'Bearer change-me-proxy-sk-token',
},
});
expect(response.statusCode).toBe(200);
const body = response.json() as {
object: 'list';
data: Array<{ id: string }>;
};
const ids = body.data.map((item) => item.id);
expect(ids).toContain('gpt-4.1');
expect(ids).not.toContain('__search');
expect(ids).not.toContain('__tavily_search');
});
});
+9 -1
View File
@@ -6,6 +6,12 @@ import { getDownstreamRoutingPolicy } from './downstreamPolicy.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { isModelAllowedByPolicyOrAllowedRoutes } from '../../services/downstreamApiKeyService.js';
function isSearchPseudoModel(modelName: string): boolean {
const normalized = (modelName || '').trim().toLowerCase();
if (!normalized) return false;
return normalized === '__search' || /^__.+_search$/.test(normalized);
}
export async function modelsProxyRoute(app: FastifyInstance) {
app.get('/v1/models', async (request) => {
const downstreamPolicy = getDownstreamRoutingPolicy(request);
@@ -32,7 +38,9 @@ export async function modelsProxyRoute(app: FastifyInstance) {
const deduped = Array.from(new Set([
...rows.map((r) => r.modelName),
...routeAliases,
])).sort();
]))
.filter((modelName) => !isSearchPseudoModel(modelName))
.sort();
const allowed: string[] = [];
for (const modelName of deduped) {
if (!await isModelAllowedByPolicyOrAllowedRoutes(modelName, downstreamPolicy)) {
+71
View File
@@ -0,0 +1,71 @@
import type { FastifyInstance, FastifyRequest } from 'fastify';
type MultipartAwareFastify = FastifyInstance & {
__metapiMultipartParserRegistered?: boolean;
};
function getContentType(request: FastifyRequest): string {
return typeof request.headers['content-type'] === 'string'
? request.headers['content-type']
: '';
}
export function ensureMultipartBufferParser(app: FastifyInstance): void {
const target = app as MultipartAwareFastify;
if (target.__metapiMultipartParserRegistered) return;
app.addContentTypeParser(/^multipart\/form-data(?:;.*)?$/i, { parseAs: 'buffer' }, (_request, body, done) => {
done(null, body);
});
target.__metapiMultipartParserRegistered = true;
}
export function isMultipartRequest(request: FastifyRequest): boolean {
return /^multipart\/form-data(?:;.*)?$/i.test(getContentType(request));
}
export async function parseMultipartFormData(request: FastifyRequest): Promise<FormData | null> {
if (!isMultipartRequest(request)) return null;
const contentType = getContentType(request);
const body = request.body;
if (!Buffer.isBuffer(body) && !(body instanceof Uint8Array)) return null;
const response = new Response(new Blob([Buffer.from(body)]), {
headers: {
'content-type': contentType,
},
});
return response.formData();
}
export function cloneFormDataWithOverrides(
formData: FormData,
overrides: Record<string, string>,
): FormData {
const next = new FormData();
const applied = new Set<string>();
for (const [key, value] of formData.entries()) {
if (Object.prototype.hasOwnProperty.call(overrides, key)) {
next.append(key, overrides[key] ?? '');
applied.add(key);
continue;
}
if (typeof value === 'string') {
next.append(key, value);
continue;
}
const fileLike = value as unknown as File;
next.append(key, value, fileLike.name || 'upload.bin');
}
for (const [key, value] of Object.entries(overrides)) {
if (applied.has(key)) continue;
next.append(key, value);
}
return next;
}
+19 -898
View File
@@ -9,13 +9,16 @@ import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { resolveProxyUrlForSite, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { type NormalizedStreamEvent } from '../../transformers/shared/normalized.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import { serializeResponsesFinalPayload } from '../../transformers/openai/responses/outbound.js';
import { convertResponsesBodyToOpenAiBody } from '../../transformers/openai/responses/conversion.js';
import {
createStreamTransformContext,
normalizeUpstreamFinalResponse,
normalizeUpstreamStreamEvent,
pullSseEventsWithDone,
type NormalizedStreamEvent,
} from './chatFormats.js';
completeResponsesStream,
createOpenAiResponsesAggregateState,
failResponsesStream,
serializeConvertedResponsesEvents,
} from '../../transformers/openai/responses/aggregator.js';
import {
buildMinimalJsonHeadersForCompatibility,
buildUpstreamEndpointRequest,
@@ -248,894 +251,8 @@ function normalizeText(value: unknown): string {
return '';
}
function stringifyJson(value: unknown): string {
try {
return JSON.stringify(value);
} catch {
return '';
}
}
function normalizeToolArguments(value: unknown): string {
if (typeof value === 'string') return value;
if (value === undefined || value === null) return '';
if (typeof value === 'number' || typeof value === 'boolean') return String(value);
if (Array.isArray(value) || isRecord(value)) return stringifyJson(value);
return '';
}
function normalizeToolOutput(value: unknown): string {
const text = normalizeText(value).trim();
if (text) return text;
if (typeof value === 'string') return value;
if (value === undefined || value === null) return '';
if (typeof value === 'number' || typeof value === 'boolean') return String(value);
if (Array.isArray(value) || isRecord(value)) return stringifyJson(value);
return '';
}
type OpenAiToolCall = {
id: string;
type: 'function';
function: {
name: string;
arguments: string;
};
};
function toOpenAiToolCall(item: Record<string, unknown>, fallbackIndex: number): OpenAiToolCall | null {
const callId = (
asTrimmedString(item.call_id)
|| asTrimmedString(item.id)
|| `call_${Date.now()}_${fallbackIndex}`
);
const name = asTrimmedString(item.name);
if (!name) return null;
return {
id: callId,
type: 'function',
function: {
name,
arguments: normalizeToolArguments(item.arguments),
},
};
}
function extractTextAndToolsFromResponsesMessage(
rawContent: unknown,
fallbackPrefix: string,
): {
text: string;
toolCalls: OpenAiToolCall[];
toolResults: Array<{ id: string; content: string }>;
} {
const contentItems = Array.isArray(rawContent) ? rawContent : [rawContent];
const textParts: string[] = [];
const toolCalls: OpenAiToolCall[] = [];
const toolResults: Array<{ id: string; content: string }> = [];
for (let index = 0; index < contentItems.length; index += 1) {
const part = contentItems[index];
if (!isRecord(part)) {
const text = normalizeText(part).trim();
if (text) textParts.push(text);
continue;
}
const partType = asTrimmedString(part.type).toLowerCase();
if (partType === 'tool_use') {
const callId = asTrimmedString(part.id) || `call_${fallbackPrefix}_${index}`;
const name = asTrimmedString(part.name);
if (!name) continue;
toolCalls.push({
id: callId,
type: 'function',
function: {
name,
arguments: normalizeToolArguments(part.input),
},
});
continue;
}
if (partType === 'tool_result') {
const id = asTrimmedString(part.tool_use_id) || asTrimmedString(part.id);
if (!id) continue;
toolResults.push({
id,
content: normalizeToolOutput(part.content),
});
continue;
}
const text = normalizeText(part).trim();
if (text) textParts.push(text);
}
return {
text: textParts.join('\n').trim(),
toolCalls,
toolResults,
};
}
function convertResponsesToolsToOpenAi(rawTools: unknown): unknown {
if (!Array.isArray(rawTools)) return rawTools;
const converted = rawTools
.map((item) => {
if (!isRecord(item)) return item;
const type = asTrimmedString(item.type).toLowerCase();
if (type !== 'function') return item;
if (isRecord(item.function) && asTrimmedString(item.function.name)) {
return item;
}
const name = asTrimmedString(item.name);
if (!name) return item;
const fn: Record<string, unknown> = { name };
const description = asTrimmedString(item.description);
if (description) fn.description = description;
if (item.parameters !== undefined) fn.parameters = item.parameters;
if (item.strict !== undefined) fn.strict = item.strict;
return {
type: 'function',
function: fn,
};
});
return converted;
}
function convertResponsesToolChoiceToOpenAi(rawToolChoice: unknown): unknown {
if (rawToolChoice === undefined) return undefined;
if (typeof rawToolChoice === 'string') return rawToolChoice;
if (!isRecord(rawToolChoice)) return rawToolChoice;
const type = asTrimmedString(rawToolChoice.type).toLowerCase();
if (type === 'function') {
if (isRecord(rawToolChoice.function) && asTrimmedString(rawToolChoice.function.name)) {
return rawToolChoice;
}
const name = asTrimmedString(rawToolChoice.name);
if (!name) return 'required';
return {
type: 'function',
function: { name },
};
}
if (type === 'auto' || type === 'none' || type === 'required') {
return type;
}
return rawToolChoice;
}
function convertResponsesBodyToOpenAiBody(
body: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const messages: Array<Record<string, unknown>> = [];
const input = body.input;
let functionCallIndex = 0;
let pendingToolCalls: OpenAiToolCall[] = [];
const flushPendingToolCalls = () => {
if (pendingToolCalls.length <= 0) return;
messages.push({
role: 'assistant',
content: '',
tool_calls: pendingToolCalls,
});
pendingToolCalls = [];
};
const pushToolOutputMessage = (callIdRaw: unknown, outputRaw: unknown) => {
const toolCallId = asTrimmedString(callIdRaw);
if (!toolCallId) return;
const content = normalizeToolOutput(outputRaw);
messages.push({
role: 'tool',
tool_call_id: toolCallId,
content,
});
};
const pushUserMessage = (content: string) => {
if (!content) return;
messages.push({ role: 'user', content });
};
const processInputItem = (item: unknown) => {
if (typeof item === 'string') {
flushPendingToolCalls();
const text = item.trim();
if (text) pushUserMessage(text);
return;
}
if (!isRecord(item)) return;
const itemType = asTrimmedString(item.type).toLowerCase();
if (itemType === 'function_call') {
const toolCall = toOpenAiToolCall(item, functionCallIndex);
functionCallIndex += 1;
if (!toolCall) return;
pendingToolCalls.push(toolCall);
return;
}
if (itemType === 'function_call_output') {
flushPendingToolCalls();
pushToolOutputMessage(item.call_id ?? item.id, item.output ?? item.content);
return;
}
flushPendingToolCalls();
const role = asTrimmedString(item.role).toLowerCase() || 'user';
const messageContent = item.content ?? item.input ?? item;
const parsed = extractTextAndToolsFromResponsesMessage(
messageContent,
`${Date.now()}_${messages.length}`,
);
if (role === 'assistant') {
if (!parsed.text && parsed.toolCalls.length <= 0) return;
const assistantMessage: Record<string, unknown> = {
role: 'assistant',
content: parsed.text,
};
if (parsed.toolCalls.length > 0) {
assistantMessage.tool_calls = parsed.toolCalls;
}
messages.push(assistantMessage);
return;
}
const normalizedRole = role === 'system' || role === 'developer'
? 'system'
: role === 'tool'
? 'tool'
: 'user';
if (normalizedRole === 'tool') {
pushToolOutputMessage(item.tool_call_id ?? item.call_id ?? item.id, messageContent);
return;
}
if (parsed.text) {
messages.push({
role: normalizedRole,
content: parsed.text,
});
}
for (const toolResult of parsed.toolResults) {
messages.push({
role: 'tool',
tool_call_id: toolResult.id,
content: toolResult.content,
});
}
};
if (typeof input === 'string') {
const text = input.trim();
if (text) {
messages.push({ role: 'user', content: text });
}
} else if (Array.isArray(input)) {
for (const item of input) {
processInputItem(item);
}
} else if (isRecord(input)) {
processInputItem(input);
}
flushPendingToolCalls();
const instructions = asTrimmedString(body.instructions);
if (instructions) {
messages.unshift({ role: 'system', content: instructions });
}
const payload: Record<string, unknown> = {
model: modelName,
stream,
messages,
};
if (typeof body.temperature === 'number' && Number.isFinite(body.temperature)) {
payload.temperature = body.temperature;
}
if (typeof body.top_p === 'number' && Number.isFinite(body.top_p)) {
payload.top_p = body.top_p;
}
if (typeof body.max_output_tokens === 'number' && Number.isFinite(body.max_output_tokens)) {
payload.max_tokens = body.max_output_tokens;
}
if (body.parallel_tool_calls !== undefined) payload.parallel_tool_calls = body.parallel_tool_calls;
if (body.tools !== undefined) payload.tools = convertResponsesToolsToOpenAi(body.tools);
if (body.tool_choice !== undefined) payload.tool_choice = convertResponsesToolChoiceToOpenAi(body.tool_choice);
return payload;
}
type UsageSummary = ReturnType<typeof parseProxyUsage>;
type ResponsesToolCall = {
id: string;
name: string;
arguments: string;
};
type ResponsesMessageItemState = {
itemId: string;
outputIndex: number;
contentIndex: number;
text: string;
};
type ResponsesToolItemState = {
toolIndex: number;
itemId: string;
callId: string;
outputIndex: number;
name: string;
arguments: string;
};
type ResponsesStreamState = {
started: boolean;
completed: boolean;
responseId: string;
model: string;
createdAt: number;
sequenceNumber: number;
outputCursor: number;
messageItem: ResponsesMessageItemState | null;
toolItems: Map<number, ResponsesToolItemState>;
};
function ensureResponseId(rawId: string): string {
const trimmed = rawId.trim() || `resp_${Date.now()}`;
return trimmed.startsWith('resp_') ? trimmed : `resp_${trimmed}`;
}
function ensureMessageId(rawId: string): string {
const trimmed = rawId.trim() || `msg_${Date.now()}`;
return trimmed.startsWith('msg_') ? trimmed : `msg_${trimmed}`;
}
function ensureFunctionCallId(rawId: string): string {
const trimmed = rawId.trim();
if (!trimmed) return `call_${Date.now()}`;
return trimmed.startsWith('call_') ? trimmed : `call_${trimmed}`;
}
function stringifyToolInput(input: unknown): string {
if (typeof input === 'string') return input;
if (!input || typeof input !== 'object') return '';
try {
return JSON.stringify(input);
} catch {
return '';
}
}
function extractToolCallsFromUpstream(payload: unknown): ResponsesToolCall[] {
if (!isRecord(payload)) return [];
if (Array.isArray(payload.choices)) {
const choice = payload.choices[0];
const message = isRecord((choice as any)?.message) ? (choice as any).message : {};
const toolCalls = Array.isArray((message as any).tool_calls) ? (message as any).tool_calls : [];
return toolCalls
.map((item: unknown, index: number) => {
if (!isRecord(item)) return null;
const fn = isRecord(item.function) ? item.function : {};
const id = typeof item.id === 'string' && item.id.trim().length > 0
? item.id
: `call_${index}`;
const name = typeof fn.name === 'string' ? fn.name : '';
const args = typeof fn.arguments === 'string' ? fn.arguments : '';
return {
id: ensureFunctionCallId(id),
name,
arguments: args,
};
})
.filter((item: ResponsesToolCall | null): item is ResponsesToolCall => !!item);
}
if (payload.type === 'message' && Array.isArray(payload.content)) {
return payload.content
.map((item: unknown, index: number) => {
if (!isRecord(item) || item.type !== 'tool_use') return null;
const id = typeof item.id === 'string' && item.id.trim().length > 0
? item.id
: `call_${index}`;
const name = typeof item.name === 'string' ? item.name : '';
const args = stringifyToolInput(item.input);
return {
id: ensureFunctionCallId(id),
name,
arguments: args,
};
})
.filter((item: ResponsesToolCall | null): item is ResponsesToolCall => !!item);
}
return [];
}
function toResponsesPayload(
upstreamPayload: unknown,
normalized: ReturnType<typeof normalizeUpstreamFinalResponse>,
usage: UsageSummary,
): Record<string, unknown> {
if (isRecord(upstreamPayload) && upstreamPayload.object === 'response') {
return upstreamPayload;
}
const normalizedId = typeof normalized.id === 'string' && normalized.id.trim()
? normalized.id.trim()
: `resp_${Date.now()}`;
const responseId = ensureResponseId(normalizedId);
const messageId = ensureMessageId(normalizedId);
const toolCalls = extractToolCallsFromUpstream(upstreamPayload);
const output: Array<Record<string, unknown>> = [];
if (normalized.content || toolCalls.length === 0) {
output.push({
id: messageId,
type: 'message',
role: 'assistant',
status: 'completed',
content: [{
type: 'output_text',
text: normalized.content || '',
}],
});
}
for (const toolCall of toolCalls) {
output.push({
id: toolCall.id,
type: 'function_call',
status: 'completed',
call_id: toolCall.id,
name: toolCall.name,
arguments: toolCall.arguments,
});
}
return {
id: responseId,
object: 'response',
created: normalized.created,
status: 'completed',
model: normalized.model,
output,
output_text: normalized.content || '',
usage: {
input_tokens: usage.promptTokens,
output_tokens: usage.completionTokens,
total_tokens: usage.totalTokens,
},
};
}
function createResponsesStreamState(modelName: string): ResponsesStreamState {
return {
started: false,
completed: false,
responseId: `resp_meta_${Date.now()}`,
model: modelName,
createdAt: Math.floor(Date.now() / 1000),
sequenceNumber: 0,
outputCursor: 0,
messageItem: null,
toolItems: new Map(),
};
}
function toResponsesUsage(usage: UsageSummary): Record<string, number> {
return {
input_tokens: usage.promptTokens,
output_tokens: usage.completionTokens,
total_tokens: usage.totalTokens,
};
}
function nextSequence(state: ResponsesStreamState): number {
const value = state.sequenceNumber;
state.sequenceNumber += 1;
return value;
}
function serializeResponsesSse(eventType: string, payload: Record<string, unknown>): string {
return `event: ${eventType}\ndata: ${JSON.stringify(payload)}\n\n`;
}
function emitResponsesEvent(
state: ResponsesStreamState,
eventType: string,
payload: Record<string, unknown>,
): string {
return serializeResponsesSse(eventType, {
type: eventType,
sequence_number: nextSequence(state),
...payload,
});
}
function buildResponseObject(
state: ResponsesStreamState,
status: 'in_progress' | 'completed' | 'failed',
usage?: UsageSummary,
): Record<string, unknown> {
const response: Record<string, unknown> = {
id: state.responseId,
object: 'response',
created_at: state.createdAt,
status,
model: state.model,
output: [],
};
if (usage) {
response.usage = toResponsesUsage(usage);
}
return response;
}
function ensureResponsesStarted(
state: ResponsesStreamState,
streamContext: { id: string; model: string; created: number },
): string[] {
if (state.started) return [];
state.started = true;
state.responseId = ensureResponseId(streamContext.id || state.responseId);
state.model = streamContext.model || state.model;
state.createdAt = streamContext.created || state.createdAt;
const inProgress = buildResponseObject(state, 'in_progress');
return [
emitResponsesEvent(state, 'response.created', { response: inProgress }),
emitResponsesEvent(state, 'response.in_progress', { response: inProgress }),
];
}
function ensureMessageItem(state: ResponsesStreamState): string[] {
if (state.messageItem) return [];
const outputIndex = state.outputCursor;
state.outputCursor += 1;
const itemId = ensureMessageId(`${state.responseId}_${outputIndex}`);
state.messageItem = {
itemId,
outputIndex,
contentIndex: 0,
text: '',
};
return [
emitResponsesEvent(state, 'response.output_item.added', {
output_index: outputIndex,
item: {
id: itemId,
type: 'message',
status: 'in_progress',
role: 'assistant',
content: [],
},
}),
emitResponsesEvent(state, 'response.content_part.added', {
item_id: itemId,
output_index: outputIndex,
content_index: 0,
part: {
type: 'output_text',
text: '',
},
}),
];
}
function toIncrementalText(existingText: string, incomingText: string): string {
if (!incomingText) return '';
if (!existingText) return incomingText;
if (incomingText === existingText) return '';
if (incomingText.startsWith(existingText)) {
return incomingText.slice(existingText.length);
}
if (existingText.endsWith(incomingText)) return '';
// Some upstreams emit overlapping windows instead of strict deltas.
// Keep only the non-overlapping suffix to avoid duplicated text.
const maxOverlap = Math.min(existingText.length, incomingText.length);
const MIN_OVERLAP = 8;
for (let overlap = maxOverlap; overlap >= MIN_OVERLAP; overlap -= 1) {
if (existingText.slice(-overlap) === incomingText.slice(0, overlap)) {
return incomingText.slice(overlap);
}
}
return incomingText;
}
function appendMessageDelta(state: ResponsesStreamState, delta: string): string[] {
if (!delta) return [];
const events: string[] = [];
events.push(...ensureMessageItem(state));
if (!state.messageItem) return events;
const normalizedDelta = toIncrementalText(state.messageItem.text, delta);
if (!normalizedDelta) return events;
state.messageItem.text += normalizedDelta;
events.push(emitResponsesEvent(state, 'response.output_text.delta', {
item_id: state.messageItem.itemId,
output_index: state.messageItem.outputIndex,
content_index: state.messageItem.contentIndex,
delta: normalizedDelta,
}));
return events;
}
function closeMessageItem(state: ResponsesStreamState): string[] {
if (!state.messageItem) return [];
const item = state.messageItem;
const events = [
emitResponsesEvent(state, 'response.output_text.done', {
item_id: item.itemId,
output_index: item.outputIndex,
content_index: item.contentIndex,
text: item.text,
}),
emitResponsesEvent(state, 'response.content_part.done', {
item_id: item.itemId,
output_index: item.outputIndex,
content_index: item.contentIndex,
part: {
type: 'output_text',
text: item.text,
},
}),
emitResponsesEvent(state, 'response.output_item.done', {
output_index: item.outputIndex,
item: {
id: item.itemId,
type: 'message',
status: 'completed',
role: 'assistant',
content: [{
type: 'output_text',
text: item.text,
}],
},
}),
];
state.messageItem = null;
return events;
}
function ensureToolItem(
state: ResponsesStreamState,
toolIndex: number,
id?: string,
name?: string,
): string[] {
const existing = state.toolItems.get(toolIndex);
if (existing) {
if (id && !existing.callId) existing.callId = ensureFunctionCallId(id);
if (id && !existing.itemId) existing.itemId = ensureFunctionCallId(id);
if (name && !existing.name) existing.name = name;
return [];
}
const outputIndex = state.outputCursor;
state.outputCursor += 1;
const callId = ensureFunctionCallId(id || `${state.responseId}_${toolIndex}`);
const itemId = callId;
const toolState: ResponsesToolItemState = {
toolIndex,
itemId,
callId,
outputIndex,
name: name || '',
arguments: '',
};
state.toolItems.set(toolIndex, toolState);
return [
emitResponsesEvent(state, 'response.output_item.added', {
output_index: outputIndex,
item: {
id: itemId,
type: 'function_call',
status: 'in_progress',
call_id: callId,
name: toolState.name,
arguments: '',
},
}),
];
}
function appendToolCallDelta(
state: ResponsesStreamState,
toolDelta: NonNullable<NormalizedStreamEvent['toolCallDeltas']>[number],
): string[] {
const toolIndex = Number.isFinite(toolDelta.index) ? Math.max(0, Math.trunc(toolDelta.index)) : 0;
const events: string[] = [];
events.push(...ensureToolItem(state, toolIndex, toolDelta.id, toolDelta.name));
const toolState = state.toolItems.get(toolIndex);
if (!toolState) return events;
if (toolDelta.name && !toolState.name) toolState.name = toolDelta.name;
if (toolDelta.argumentsDelta !== undefined) {
toolState.arguments += toolDelta.argumentsDelta;
events.push(emitResponsesEvent(state, 'response.function_call_arguments.delta', {
item_id: toolState.itemId,
output_index: toolState.outputIndex,
delta: toolDelta.argumentsDelta,
}));
}
return events;
}
function closeToolItems(state: ResponsesStreamState): string[] {
if (state.toolItems.size <= 0) return [];
const ordered = Array.from(state.toolItems.values())
.sort((a, b) => a.outputIndex - b.outputIndex);
const events: string[] = [];
for (const toolItem of ordered) {
events.push(emitResponsesEvent(state, 'response.function_call_arguments.done', {
item_id: toolItem.itemId,
output_index: toolItem.outputIndex,
arguments: toolItem.arguments,
}));
events.push(emitResponsesEvent(state, 'response.output_item.done', {
output_index: toolItem.outputIndex,
item: {
id: toolItem.itemId,
type: 'function_call',
status: 'completed',
call_id: toolItem.callId,
name: toolItem.name,
arguments: toolItem.arguments,
},
}));
}
state.toolItems.clear();
return events;
}
function completeResponsesStream(
state: ResponsesStreamState,
streamContext: { id: string; model: string; created: number },
usage: UsageSummary,
): string[] {
if (state.completed) return [];
const events: string[] = [];
events.push(...ensureResponsesStarted(state, streamContext));
events.push(...closeMessageItem(state));
events.push(...closeToolItems(state));
events.push(emitResponsesEvent(state, 'response.completed', {
response: buildResponseObject(state, 'completed', usage),
}));
events.push('data: [DONE]\n\n');
state.completed = true;
return events;
}
function failResponsesStream(
state: ResponsesStreamState,
streamContext: { id: string; model: string; created: number },
usage: UsageSummary,
payload?: unknown,
): string[] {
if (state.completed) return [];
const events: string[] = [];
events.push(...ensureResponsesStarted(state, streamContext));
const responsePayload = isRecord(payload) && isRecord((payload as any).response)
? ((payload as any).response as Record<string, unknown>)
: null;
const errorPayload = isRecord(payload) && isRecord((payload as any).error)
? ((payload as any).error as Record<string, unknown>)
: null;
if (responsePayload && asTrimmedString(responsePayload.id)) {
state.responseId = ensureResponseId(asTrimmedString(responsePayload.id));
}
if (responsePayload && asTrimmedString(responsePayload.model)) {
state.model = asTrimmedString(responsePayload.model);
}
const failedResponse = buildResponseObject(state, 'failed', usage);
if (errorPayload) {
failedResponse.error = errorPayload;
}
events.push(emitResponsesEvent(state, 'response.failed', {
response: failedResponse,
}));
events.push('data: [DONE]\n\n');
state.completed = true;
return events;
}
function serializeConvertedResponsesEvents(input: {
state: ResponsesStreamState;
streamContext: { id: string; model: string; created: number };
event: NormalizedStreamEvent;
usage: UsageSummary;
}): string[] {
const { state, streamContext, event, usage } = input;
if (state.completed) return [];
const shouldStart = (
event.role === 'assistant'
|| !!event.contentDelta
|| !!event.reasoningDelta
|| (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0)
|| !!event.done
|| !!event.finishReason
);
const events: string[] = [];
if (shouldStart) {
events.push(...ensureResponsesStarted(state, streamContext));
}
if (event.contentDelta) {
events.push(...appendMessageDelta(state, event.contentDelta));
}
if (event.reasoningDelta) {
events.push(...appendMessageDelta(state, event.reasoningDelta));
}
if (Array.isArray(event.toolCallDeltas) && event.toolCallDeltas.length > 0) {
events.push(...closeMessageItem(state));
for (const toolDelta of event.toolCallDeltas) {
events.push(...appendToolCallDelta(state, toolDelta));
}
}
if (event.finishReason) {
// Keep compatibility with clients that expect finish before completed.
}
if (event.done) {
events.push(...completeResponsesStream(state, streamContext, usage));
}
return events;
}
export async function responsesProxyRoute(app: FastifyInstance) {
const handleResponsesRequest = async (
request: FastifyRequest,
@@ -1374,15 +491,15 @@ export async function responsesProxyRoute(app: FastifyInstance) {
let sseBuffer = '';
const passthroughResponsesStream = successfulUpstreamPath === '/v1/responses';
const streamContext = createStreamTransformContext(modelName);
const responsesState = createResponsesStreamState(modelName);
const streamContext = openAiResponsesTransformer.createStreamContext(modelName);
const responsesState = createOpenAiResponsesAggregateState(modelName);
const writeLines = (lines: string[]) => {
for (const line of lines) reply.raw.write(line);
};
const consumeSseBuffer = (incoming: string): string => {
const pulled = pullSseEventsWithDone(incoming);
const pulled = openAiResponsesTransformer.pullSseEvents(incoming);
for (const eventBlock of pulled.events) {
if (eventBlock.data === '[DONE]') {
if (passthroughResponsesStream) {
@@ -1425,7 +542,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
}
if (parsedPayload && typeof parsedPayload === 'object') {
const normalizedEvent = normalizeUpstreamStreamEvent(parsedPayload, streamContext, modelName);
const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(parsedPayload, streamContext, modelName);
writeLines(serializeConvertedResponsesEvents({
state: responsesState,
streamContext,
@@ -1510,12 +627,16 @@ export async function responsesProxyRoute(app: FastifyInstance) {
}
const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(upstreamData);
const normalized = normalizeUpstreamFinalResponse(
const normalized = openAiResponsesTransformer.transformFinalResponse(
upstreamData,
modelName,
rawText,
);
const downstreamData = toResponsesPayload(upstreamData, normalized, parsedUsage);
const downstreamData = serializeResponsesFinalPayload({
upstreamPayload: upstreamData,
normalized,
usage: parsedUsage,
});
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
+6
View File
@@ -6,6 +6,9 @@ import { embeddingsProxyRoute } from './embeddings.js';
import { completionsProxyRoute } from './completions.js';
import { responsesProxyRoute } from './responses.js';
import { imagesProxyRoute } from './images.js';
import { searchProxyRoute } from './search.js';
import { geminiProxyRoute } from './gemini.js';
import { videosProxyRoute } from './videos.js';
export async function proxyRoutes(app: FastifyInstance) {
// Auth middleware for all /v1 routes
@@ -19,5 +22,8 @@ export async function proxyRoutes(app: FastifyInstance) {
await app.register(responsesProxyRoute);
await app.register(modelsProxyRoute);
await app.register(embeddingsProxyRoute);
await app.register(searchProxyRoute);
await app.register(imagesProxyRoute);
await app.register(videosProxyRoute);
await app.register(geminiProxyRoute);
}
+197
View File
@@ -0,0 +1,197 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordSuccessMock = vi.fn();
const recordFailureMock = vi.fn();
const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async () => 0);
const dbInsertMock = vi.fn((_arg?: any) => ({
values: () => ({
run: () => undefined,
}),
}));
vi.mock('undici', () => ({
fetch: (...args: unknown[]) => fetchMock(...args),
}));
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
},
}));
vi.mock('../../services/modelService.js', () => ({
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
}));
vi.mock('../../services/alertService.js', () => ({
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
}));
vi.mock('../../services/alertRules.js', () => ({
isTokenExpiredError: () => false,
}));
vi.mock('../../services/modelPricingService.js', () => ({
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
}));
vi.mock('../../db/index.js', () => ({
db: {
insert: (arg: any) => dbInsertMock(arg),
},
schema: {
proxyLogs: {},
},
}));
describe('/v1/search route', () => {
let app: FastifyInstance;
beforeAll(async () => {
const { searchProxyRoute } = await import('./search.js');
app = Fastify();
await app.register(searchProxyRoute);
});
beforeEach(() => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordSuccessMock.mockReset();
recordFailureMock.mockReset();
refreshModelsAndRebuildRoutesMock.mockReset();
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
dbInsertMock.mockClear();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'demo-site', url: 'https://upstream.example.com', platform: 'openai' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: '__search',
});
selectNextChannelMock.mockReturnValue(null);
});
afterAll(async () => {
await app.close();
});
it('defaults model to __search and forwards to the upstream /v1/search endpoint', async () => {
fetchMock.mockResolvedValue(new Response(JSON.stringify({
object: 'search.result',
data: [{ title: 'AxonHub' }],
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/search',
headers: {
authorization: 'Bearer sk-demo',
},
payload: {
query: 'axonhub',
},
});
expect(response.statusCode).toBe(200);
expect(selectChannelMock).toHaveBeenCalledWith('__search', expect.anything());
const [targetUrl, requestInit] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://upstream.example.com/v1/search');
expect(JSON.parse(String(requestInit.body))).toEqual({
query: 'axonhub',
max_results: 10,
model: '__search',
});
});
it('rejects max_results outside the allowed range', async () => {
const response = await app.inject({
method: 'POST',
url: '/v1/search',
headers: {
authorization: 'Bearer sk-demo',
},
payload: {
query: 'axonhub',
max_results: 21,
},
});
expect(response.statusCode).toBe(400);
expect(response.json()).toMatchObject({
error: {
message: 'max_results must be an integer between 1 and 20',
type: 'invalid_request_error',
},
});
expect(fetchMock).not.toHaveBeenCalled();
});
it('rejects max_results below the allowed range', async () => {
const response = await app.inject({
method: 'POST',
url: '/v1/search',
headers: {
authorization: 'Bearer sk-demo',
},
payload: {
query: 'axonhub',
max_results: 0,
},
});
expect(response.statusCode).toBe(400);
expect(response.json()).toMatchObject({
error: {
message: 'max_results must be an integer between 1 and 20',
type: 'invalid_request_error',
},
});
expect(fetchMock).not.toHaveBeenCalled();
});
it('rejects streaming search requests', async () => {
const response = await app.inject({
method: 'POST',
url: '/v1/search',
headers: {
authorization: 'Bearer sk-demo',
},
payload: {
query: 'axonhub',
stream: true,
},
});
expect(response.statusCode).toBe(400);
expect(response.json()).toMatchObject({
error: {
message: 'search does not support streaming',
type: 'invalid_request_error',
},
});
expect(fetchMock).not.toHaveBeenCalled();
});
});
+180
View File
@@ -0,0 +1,180 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
import { db, 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 { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { composeProxyLogMessage } from './logPathMeta.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
const MAX_RETRIES = 2;
const DEFAULT_SEARCH_MODEL = '__search';
const DEFAULT_MAX_RESULTS = 10;
const MAX_MAX_RESULTS = 20;
export async function searchProxyRoute(app: FastifyInstance) {
app.post('/v1/search', async (request: FastifyRequest, reply: FastifyReply) => {
const body = (request.body || {}) as Record<string, unknown>;
const query = typeof body.query === 'string' ? body.query.trim() : '';
if (!query) {
return reply.code(400).send({
error: { message: 'query is required', type: 'invalid_request_error' },
});
}
if (body.stream === true) {
return reply.code(400).send({
error: { message: 'search does not support streaming', type: 'invalid_request_error' },
});
}
const rawMaxResults = body.max_results;
const maxResults = rawMaxResults == null
? DEFAULT_MAX_RESULTS
: (typeof rawMaxResults === 'number' && Number.isInteger(rawMaxResults)
? rawMaxResults
: NaN);
if (!Number.isInteger(maxResults) || maxResults < 1 || maxResults > MAX_MAX_RESULTS) {
return reply.code(400).send({
error: {
message: `max_results must be an integer between 1 and ${MAX_MAX_RESULTS}`,
type: 'invalid_request_error',
},
});
}
const requestedModel = typeof body.model === 'string' && body.model.trim().length > 0
? body.model.trim()
: DEFAULT_SEARCH_MODEL;
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
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({
model: requestedModel,
reason: 'No available channels after retries',
});
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = `${selected.site.url}/v1/search`;
const forwardBody = {
...body,
max_results: maxResults,
model: selected.actualModel || requestedModel,
};
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 text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', upstream.status, Date.now() - startTime, text, retryCount);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
let data: any = {};
try { data = JSON.parse(text); } catch { data = { data: [] }; }
const latency = Date.now() - startTime;
tokenRouter.recordSuccess(selected.channel.id, latency, 0);
recordDownstreamCostUsage(request, 0);
logProxy(selected, requestedModel, 'success', upstream.status, latency, null, retryCount);
return reply.code(upstream.status).send(data);
} catch (error: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(selected, requestedModel, 'failed', 0, Date.now() - startTime, error?.message || 'network error', retryCount);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: error?.message || 'network failure',
});
return reply.code(502).send({
error: { message: error?.message || 'network failure', type: 'upstream_error' },
});
}
}
});
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
estimatedCost: 0,
errorMessage: composeProxyLogMessage({
downstreamPath: '/v1/search',
errorMessage,
}),
retryCount,
createdAt,
}).run();
} catch (error) {
console.warn('[proxy/search] failed to write proxy log', error);
}
}
@@ -313,6 +313,84 @@ describe('buildUpstreamEndpointRequest', () => {
]);
});
it('preserves multimodal user content when converting chat to messages', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'messages',
modelName: 'claude-3-7-sonnet',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'anthropic',
siteUrl: 'https://example.com',
openaiBody: {
model: 'gpt-4o-mini',
messages: [
{
role: 'user',
content: [
{ type: 'text', text: 'describe this' },
{ type: 'image_url', image_url: { url: 'https://example.com/cat.png' } },
],
},
],
},
downstreamFormat: 'openai',
});
expect(request.path).toBe('/v1/messages');
expect(request.body.messages).toEqual([
{
role: 'user',
content: [
{ type: 'text', text: 'describe this' },
{
type: 'image',
cache_control: { type: 'ephemeral' },
source: {
type: 'url',
url: 'https://example.com/cat.png',
},
},
],
},
]);
});
it('preserves multimodal user content when converting chat to responses', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-4.1',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'openai',
siteUrl: 'https://example.com',
openaiBody: {
model: 'gpt-4o-mini',
messages: [
{
role: 'user',
content: [
{ type: 'text', text: 'describe this' },
{ type: 'image_url', image_url: { url: 'https://example.com/cat.png' } },
],
},
],
},
downstreamFormat: 'openai',
});
expect(request.path).toBe('/v1/responses');
expect(request.body.input).toEqual([
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'describe this' },
{ type: 'input_image', image_url: { url: 'https://example.com/cat.png' } },
],
},
]);
});
it('applies global responses standardization and drops non-standard fields', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
@@ -419,4 +497,283 @@ describe('buildUpstreamEndpointRequest', () => {
expect(request.headers['content-type']).toBeUndefined();
expect(request.headers['Content-Type']).toBe('application/json');
});
it('preserves structured responses content blocks instead of flattening them', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'upstream-gpt',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'openai',
siteUrl: 'https://example.com',
openaiBody: {},
downstreamFormat: 'responses',
responsesOriginalBody: {
model: 'gpt-5.2',
input: [
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'describe this image' },
{ type: 'input_image', image_url: 'https://example.com/cat.png' },
{
type: 'input_audio',
input_audio: {
data: 'UklGRg==',
format: 'wav',
},
},
],
},
],
},
});
expect(request.path).toBe('/v1/responses');
expect(request.body.input).toEqual([
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'describe this image' },
{ type: 'input_image', image_url: 'https://example.com/cat.png' },
{
type: 'input_audio',
input_audio: {
data: 'UklGRg==',
format: 'wav',
},
},
],
},
]);
});
it('preserves multimodal OpenAI user content when converting to Anthropic messages', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'messages',
modelName: 'claude-sonnet-4-5',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'claude',
siteUrl: 'https://example.com',
downstreamFormat: 'openai',
openaiBody: {
model: 'claude-sonnet-4-5',
messages: [
{
role: 'system',
content: 'be careful',
},
{
role: 'user',
content: [
{ type: 'text', text: 'describe this image' },
{ type: 'image_url', image_url: { url: 'https://example.com/cat.png' } },
],
},
],
},
});
expect(request.path).toBe('/v1/messages');
expect(request.body.system).toBe('be careful');
expect(request.body.messages).toEqual([
{
role: 'user',
content: [
{ type: 'text', text: 'describe this image' },
{
type: 'image',
cache_control: { type: 'ephemeral' },
source: {
type: 'url',
url: 'https://example.com/cat.png',
},
},
],
},
]);
});
it('preserves assistant reasoning blocks for Anthropic and strips cache_control from thinking blocks', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'messages',
modelName: 'claude-sonnet-4-5',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'claude',
siteUrl: 'https://example.com',
downstreamFormat: 'openai',
openaiBody: {
model: 'claude-sonnet-4-5',
messages: [
{
role: 'assistant',
content: [
{
type: 'reasoning',
text: 'internal thinking',
cache_control: { type: 'ephemeral' },
},
{
type: 'text',
text: 'final answer',
cache_control: { type: 'ephemeral' },
},
],
},
],
},
});
expect(request.path).toBe('/v1/messages');
expect(request.body.messages).toEqual([
{
role: 'assistant',
content: [
{
type: 'thinking',
thinking: 'internal thinking',
},
{
type: 'text',
text: 'final answer',
cache_control: { type: 'ephemeral' },
},
],
},
]);
});
it('preserves multimodal OpenAI user content when converting to Responses input', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.2',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'openai',
siteUrl: 'https://example.com',
downstreamFormat: 'openai',
openaiBody: {
model: 'gpt-5.2',
messages: [
{
role: 'user',
content: [
{ type: 'text', text: 'describe the upload' },
{ type: 'image_url', image_url: { url: 'https://example.com/cat.png' } },
],
},
],
},
});
expect(request.path).toBe('/v1/responses');
expect(request.body.input).toEqual([
{
type: 'message',
role: 'user',
content: [
{ type: 'input_text', text: 'describe the upload' },
{ type: 'input_image', image_url: { url: 'https://example.com/cat.png' } },
],
},
]);
});
it('preserves Anthropic image and tool_result blocks instead of flattening to plain text', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'messages',
modelName: 'claude-opus-4-6',
stream: false,
tokenValue: 'sk-test',
sitePlatform: 'claude',
siteUrl: 'https://example.com',
openaiBody: {
model: 'gpt-5.2',
messages: [
{
role: 'system',
content: 'system prompt',
},
{
role: 'user',
content: [
{ type: 'text', text: 'look at this' },
{ type: 'image_url', image_url: 'https://example.com/cat.png' },
],
},
{
role: 'assistant',
tool_calls: [
{
id: 'call_1',
type: 'function',
function: {
name: 'lookup',
arguments: '{"topic":"cat"}',
},
},
],
},
{
role: 'tool',
tool_call_id: 'call_1',
content: '{"ok":true}',
},
{
role: 'user',
content: 'thanks',
},
],
},
downstreamFormat: 'openai',
});
expect(request.path).toBe('/v1/messages');
expect(request.body.system).toBe('system prompt');
expect(request.body.messages).toEqual([
{
role: 'user',
content: [
{ type: 'text', text: 'look at this' },
{
type: 'image',
source: {
type: 'url',
url: 'https://example.com/cat.png',
},
},
],
},
{
role: 'assistant',
content: [
{
type: 'tool_use',
id: 'call_1',
name: 'lookup',
input: { topic: 'cat' },
},
],
},
{
role: 'user',
content: [
{
type: 'tool_result',
tool_use_id: 'call_1',
content: '{"ok":true}',
},
{
type: 'text',
text: 'thanks',
cache_control: { type: 'ephemeral' },
},
],
},
]);
});
});
+14 -659
View File
@@ -1,5 +1,13 @@
import { fetchModelPricingCatalog } from '../../services/modelPricingService.js';
import type { DownstreamFormat } from './chatFormats.js';
import type { DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
convertOpenAiBodyToResponsesBody as convertOpenAiBodyToResponsesBodyViaTransformer,
sanitizeResponsesBodyForProxy as sanitizeResponsesBodyForProxyViaTransformer,
} from '../../transformers/openai/responses/conversion.js';
import {
convertOpenAiBodyToAnthropicMessagesBody,
sanitizeAnthropicMessagesBody,
} from '../../transformers/anthropic/messages/conversion.js';
export type UpstreamEndpoint = 'chat' | 'messages' | 'responses';
export type EndpointPreference = DownstreamFormat | 'responses';
@@ -170,660 +178,6 @@ function toFiniteNumber(value: unknown): number | null {
return typeof value === 'number' && Number.isFinite(value) ? value : null;
}
function sanitizeMessagesBodyForAnthropic(body: Record<string, unknown>): Record<string, unknown> {
const sanitized: Record<string, unknown> = { ...body };
const hasTemperature = toFiniteNumber(sanitized.temperature) !== null;
const hasTopP = toFiniteNumber(sanitized.top_p) !== null;
// Some Anthropic-compatible upstreams reject requests carrying both fields.
if (hasTemperature && hasTopP) {
delete sanitized.top_p;
}
// Claude Code may send thinking.type = "adaptive". Many Anthropic-compatible
// upstreams only accept enabled/disabled.
const thinking = sanitized.thinking;
if (isRecord(thinking)) {
const rawType = asTrimmedString(thinking.type).toLowerCase();
if (rawType === 'adaptive') {
const budgetTokens = toFiniteNumber(thinking.budget_tokens);
if (budgetTokens !== null && budgetTokens > 0) {
sanitized.thinking = {
...thinking,
type: 'enabled',
budget_tokens: Math.trunc(budgetTokens),
};
} else {
sanitized.thinking = { type: 'disabled' };
}
} else if (rawType && rawType !== 'enabled' && rawType !== 'disabled') {
sanitized.thinking = { type: 'disabled' };
}
}
return sanitized;
}
function normalizeContentText(content: unknown): string {
if (typeof content === 'string') return content;
if (Array.isArray(content)) {
return content
.map((item) => {
if (typeof item === 'string') return item;
if (!isRecord(item)) return '';
const text = asTrimmedString(item.text ?? item.content ?? item.output_text);
return text;
})
.filter((item) => item.length > 0)
.join('\n');
}
if (isRecord(content)) {
const text = asTrimmedString(content.text ?? content.content ?? content.output_text);
return text;
}
return '';
}
function safeJsonStringify(value: unknown): string {
try {
return JSON.stringify(value);
} catch {
return '';
}
}
function parseJsonString(raw: string): unknown {
const trimmed = raw.trim();
if (!trimmed) return {};
try {
return JSON.parse(trimmed);
} catch {
return { value: raw };
}
}
function normalizeOpenAiToolArguments(raw: unknown): string {
if (typeof raw === 'string') return raw;
if (raw === undefined || raw === null) return '';
if (typeof raw === 'number' || typeof raw === 'boolean') return String(raw);
if (Array.isArray(raw) || isRecord(raw)) {
return safeJsonStringify(raw);
}
return '';
}
function normalizeAnthropicToolInput(raw: unknown): unknown {
if (raw === undefined || raw === null) return {};
if (isRecord(raw) || Array.isArray(raw)) return raw;
if (typeof raw === 'string') return parseJsonString(raw);
if (typeof raw === 'number' || typeof raw === 'boolean') return raw;
return {};
}
function normalizeToolMessageContent(raw: unknown): string {
const contentText = normalizeContentText(raw).trim();
if (contentText) return contentText;
if (raw === undefined || raw === null) return '';
if (typeof raw === 'string') return raw;
if (typeof raw === 'number' || typeof raw === 'boolean') return String(raw);
if (Array.isArray(raw) || isRecord(raw)) return safeJsonStringify(raw);
return '';
}
function convertOpenAiToolsToAnthropic(rawTools: unknown): unknown {
if (!Array.isArray(rawTools)) return rawTools;
const converted = rawTools
.map((item) => {
if (!isRecord(item)) return null;
const type = asTrimmedString(item.type).toLowerCase();
if (type === 'function' && isRecord(item.function)) {
const fn = item.function;
const name = asTrimmedString(fn.name);
if (!name) return null;
const mapped: Record<string, unknown> = { name };
const description = asTrimmedString(fn.description);
if (description) mapped.description = description;
if (fn.parameters !== undefined) mapped.input_schema = fn.parameters;
if (item.cache_control !== undefined) mapped.cache_control = item.cache_control;
return mapped;
}
if (asTrimmedString(item.name) && item.input_schema !== undefined) {
return item;
}
return null;
})
.filter((item): item is Record<string, unknown> => !!item);
return converted.length > 0 ? converted : rawTools;
}
function convertOpenAiToolChoiceToAnthropic(rawToolChoice: unknown): unknown {
if (rawToolChoice === undefined) return undefined;
if (typeof rawToolChoice === 'string') {
const normalized = rawToolChoice.trim().toLowerCase();
if (normalized === 'required') return { type: 'any' };
if (normalized === 'none') return { type: 'none' };
if (normalized === 'auto') return { type: 'auto' };
if (normalized === 'any') return { type: 'any' };
return rawToolChoice;
}
if (!isRecord(rawToolChoice)) return rawToolChoice;
const type = asTrimmedString(rawToolChoice.type).toLowerCase();
if (type === 'function' && isRecord(rawToolChoice.function)) {
const name = asTrimmedString(rawToolChoice.function.name);
return name ? { type: 'tool', name } : { type: 'any' };
}
if (type === 'tool' || type === 'auto' || type === 'any' || type === 'none') {
return rawToolChoice;
}
return rawToolChoice;
}
function convertOpenAiToolsToResponses(rawTools: unknown): unknown {
if (!Array.isArray(rawTools)) return rawTools;
const converted = rawTools
.map((item) => {
if (!isRecord(item)) return null;
const type = asTrimmedString(item.type).toLowerCase();
if (type === 'function' && isRecord(item.function)) {
const fn = item.function;
const name = asTrimmedString(fn.name);
if (!name) return null;
const mapped: Record<string, unknown> = {
type: 'function',
name,
};
const description = asTrimmedString(fn.description);
if (description) mapped.description = description;
if (fn.parameters !== undefined) mapped.parameters = fn.parameters;
if (fn.strict !== undefined) mapped.strict = fn.strict;
return mapped;
}
if (type === 'function' && asTrimmedString(item.name)) {
return item;
}
if (type === 'image_generation') {
return item;
}
return null;
})
.filter((item): item is Record<string, unknown> => !!item);
return converted.length > 0 ? converted : rawTools;
}
function convertOpenAiToolChoiceToResponses(rawToolChoice: unknown): unknown {
if (rawToolChoice === undefined) return undefined;
if (typeof rawToolChoice === 'string') return rawToolChoice;
if (!isRecord(rawToolChoice)) return rawToolChoice;
const type = asTrimmedString(rawToolChoice.type).toLowerCase();
if (type === 'function' && isRecord(rawToolChoice.function)) {
const name = asTrimmedString(rawToolChoice.function.name);
if (!name) return 'required';
return { type: 'function', name };
}
return rawToolChoice;
}
function toResponsesInputMessageFromText(text: string): Record<string, unknown> {
return {
type: 'message',
role: 'user',
content: [{ type: 'input_text', text }],
};
}
function normalizeResponsesMessageItem(item: Record<string, unknown>): Record<string, unknown> {
const type = asTrimmedString(item.type).toLowerCase();
if (type === 'function_call' || type === 'function_call_output') {
return item;
}
const role = asTrimmedString(item.role).toLowerCase();
const textContent = normalizeContentText(item.content ?? item.text).trim();
if (type === 'message') {
if (!textContent) return item;
const normalizedRole = role || 'user';
const textType = normalizedRole === 'assistant' ? 'output_text' : 'input_text';
return {
...item,
role: normalizedRole,
content: [{ type: textType, text: textContent }],
};
}
if (role) {
if (!textContent) return item;
const textType = role === 'assistant' ? 'output_text' : 'input_text';
return {
type: 'message',
role,
content: [{ type: textType, text: textContent }],
};
}
if (textContent) {
return toResponsesInputMessageFromText(textContent);
}
return item;
}
function normalizeResponsesInputForCompatibility(input: unknown): unknown {
if (typeof input === 'string') {
const normalized = input.trim();
if (!normalized) return input;
return [toResponsesInputMessageFromText(normalized)];
}
if (Array.isArray(input)) {
return input.map((item) => {
if (typeof item === 'string') {
const normalized = item.trim();
return normalized ? toResponsesInputMessageFromText(normalized) : item;
}
if (!isRecord(item)) return item;
return normalizeResponsesMessageItem(item);
});
}
if (isRecord(input)) {
return [normalizeResponsesMessageItem(input)];
}
return input;
}
function normalizeResponsesBodyForCompatibility(
body: Record<string, unknown>,
): Record<string, unknown> {
const nextInput = normalizeResponsesInputForCompatibility(body.input);
if (nextInput === body.input) return body;
return {
...body,
input: nextInput,
};
}
const ALLOWED_RESPONSES_FIELDS = new Set([
'model',
'input',
'instructions',
'max_output_tokens',
'max_completion_tokens',
'temperature',
'top_p',
'truncation',
'tools',
'tool_choice',
'parallel_tool_calls',
'metadata',
'reasoning',
'store',
'stream',
'user',
'previous_response_id',
'text',
'audio',
'include',
'response_format',
'service_tier',
'stop',
'n',
]);
function sanitizeResponsesBodyForProxy(
body: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
let normalized = normalizeResponsesBodyForCompatibility({
...body,
model: modelName,
stream,
});
if (normalized.input === undefined) {
if (Array.isArray((normalized as Record<string, unknown>).messages)) {
const converted = convertOpenAiBodyToResponsesBody(normalized, modelName, stream);
normalized = normalizeResponsesBodyForCompatibility(converted);
} else {
const prompt = asTrimmedString((normalized as Record<string, unknown>).prompt);
if (prompt) {
normalized = {
...normalized,
input: [toResponsesInputMessageFromText(prompt)],
};
}
}
}
const sanitized: Record<string, unknown> = {};
for (const [key, value] of Object.entries(normalized)) {
if (!ALLOWED_RESPONSES_FIELDS.has(key)) continue;
if (key === 'max_completion_tokens') continue;
sanitized[key] = value;
}
const maxOutputTokens = toFiniteNumber(normalized.max_output_tokens);
if (maxOutputTokens !== null && maxOutputTokens > 0) {
sanitized.max_output_tokens = Math.trunc(maxOutputTokens);
} else {
const maxCompletionTokens = toFiniteNumber(normalized.max_completion_tokens);
if (maxCompletionTokens !== null && maxCompletionTokens > 0) {
sanitized.max_output_tokens = Math.trunc(maxCompletionTokens);
}
}
sanitized.model = modelName;
sanitized.stream = stream;
return sanitized;
}
function convertOpenAiBodyToMessagesBody(
openaiBody: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const rawMessages = Array.isArray(openaiBody.messages) ? openaiBody.messages : [];
const systemContents: string[] = [];
const messages: Array<{
role: 'user' | 'assistant';
content: string | Array<Record<string, unknown>>;
}> = [];
const appendAssistantMessage = (item: Record<string, unknown>) => {
const role = asTrimmedString(item.role).toLowerCase() || 'assistant';
if (role !== 'assistant') return;
const contentBlocks: Array<Record<string, unknown>> = [];
const textContent = normalizeContentText(item.content).trim();
if (textContent) {
contentBlocks.push({
type: 'text',
text: textContent,
});
}
const rawToolCalls = Array.isArray(item.tool_calls) ? item.tool_calls : [];
for (let index = 0; index < rawToolCalls.length; index += 1) {
const toolCall = rawToolCalls[index];
if (!isRecord(toolCall)) continue;
const functionPart = isRecord(toolCall.function) ? toolCall.function : {};
const id = asTrimmedString(toolCall.id) || `toolu_${Date.now()}_${index}`;
const name = (
asTrimmedString(functionPart.name)
|| asTrimmedString(toolCall.name)
|| `tool_${index}`
);
const input = normalizeAnthropicToolInput(functionPart.arguments ?? toolCall.arguments);
contentBlocks.push({
type: 'tool_use',
id,
name,
input,
});
}
if (contentBlocks.length <= 0) return;
if (contentBlocks.length === 1 && contentBlocks[0].type === 'text') {
const singleText = asTrimmedString(contentBlocks[0].text);
if (singleText) {
messages.push({
role: 'assistant',
content: singleText,
});
}
return;
}
messages.push({
role: 'assistant',
content: contentBlocks,
});
};
for (let messageIndex = 0; messageIndex < rawMessages.length; messageIndex += 1) {
const item = rawMessages[messageIndex];
if (!isRecord(item)) continue;
const role = asTrimmedString(item.role).toLowerCase() || 'user';
if (role === 'system' || role === 'developer') {
const content = normalizeContentText(item.content);
if (!content) continue;
systemContents.push(content);
continue;
}
if (role === 'tool') {
const toolResultBlocks: Array<Record<string, unknown>> = [];
let cursor = messageIndex;
while (cursor < rawMessages.length) {
const toolCandidate = rawMessages[cursor];
if (!isRecord(toolCandidate)) break;
const toolRole = asTrimmedString(toolCandidate.role).toLowerCase();
if (toolRole !== 'tool') break;
const toolUseId = (
asTrimmedString(toolCandidate.tool_call_id)
|| asTrimmedString(toolCandidate.id)
);
const toolResultContent = normalizeToolMessageContent(toolCandidate.content).trim();
if (toolUseId && toolResultContent) {
toolResultBlocks.push({
type: 'tool_result',
tool_use_id: toolUseId,
content: toolResultContent,
});
}
cursor += 1;
}
if (toolResultBlocks.length > 0 && cursor < rawMessages.length) {
const nextItem = rawMessages[cursor];
if (isRecord(nextItem)) {
const nextRole = asTrimmedString(nextItem.role).toLowerCase();
if (nextRole === 'user') {
const userText = normalizeContentText(nextItem.content).trim();
if (userText) {
toolResultBlocks.push({
type: 'text',
text: userText,
});
}
cursor += 1;
}
}
}
if (toolResultBlocks.length > 0) {
messages.push({
role: 'user',
content: toolResultBlocks,
});
}
messageIndex = cursor - 1;
continue;
}
if (role === 'assistant') {
appendAssistantMessage(item);
continue;
}
const content = normalizeContentText(item.content).trim();
if (!content) continue;
messages.push({
role: 'user',
content,
});
}
const body: Record<string, unknown> = {
model: modelName,
stream,
messages,
max_tokens: toFiniteNumber(openaiBody.max_tokens) ?? 4096,
};
if (systemContents.length > 0) {
body.system = systemContents.join('\n\n');
}
const temperature = toFiniteNumber(openaiBody.temperature);
if (temperature !== null) body.temperature = temperature;
const topP = toFiniteNumber(openaiBody.top_p);
if (topP !== null) body.top_p = topP;
if (Array.isArray(openaiBody.stop) && openaiBody.stop.length > 0) {
body.stop_sequences = openaiBody.stop;
}
if (openaiBody.tools !== undefined) body.tools = convertOpenAiToolsToAnthropic(openaiBody.tools);
const anthropicToolChoice = convertOpenAiToolChoiceToAnthropic(openaiBody.tool_choice);
if (anthropicToolChoice !== undefined) body.tool_choice = anthropicToolChoice;
return body;
}
function convertOpenAiBodyToResponsesBody(
openaiBody: Record<string, unknown>,
modelName: string,
stream: boolean,
): Record<string, unknown> {
const rawMessages = Array.isArray(openaiBody.messages) ? openaiBody.messages : [];
const systemContents: string[] = [];
const inputItems: Array<Record<string, unknown>> = [];
for (const item of rawMessages) {
if (!isRecord(item)) continue;
const role = asTrimmedString(item.role).toLowerCase() || 'user';
if (role === 'system' || role === 'developer') {
const content = normalizeContentText(item.content).trim();
if (!content) continue;
systemContents.push(content);
continue;
}
if (role === 'assistant') {
const textContent = normalizeContentText(item.content).trim();
if (textContent) {
inputItems.push({
type: 'message',
role: 'assistant',
content: [{ type: 'output_text', text: textContent }],
});
}
const rawToolCalls = Array.isArray(item.tool_calls) ? item.tool_calls : [];
for (let index = 0; index < rawToolCalls.length; index += 1) {
const toolCall = rawToolCalls[index];
if (!isRecord(toolCall)) continue;
const functionPart = isRecord(toolCall.function) ? toolCall.function : {};
const callId = asTrimmedString(toolCall.id) || `call_${Date.now()}_${index}`;
const name = (
asTrimmedString(functionPart.name)
|| asTrimmedString(toolCall.name)
|| `tool_${index}`
);
const argumentsValue = normalizeOpenAiToolArguments(
functionPart.arguments ?? toolCall.arguments,
);
inputItems.push({
type: 'function_call',
call_id: callId,
name,
arguments: argumentsValue,
});
}
continue;
}
if (role === 'tool') {
const callId = asTrimmedString(item.tool_call_id) || asTrimmedString(item.id);
if (!callId) continue;
inputItems.push({
type: 'function_call_output',
call_id: callId,
output: normalizeToolMessageContent(item.content),
});
continue;
}
const content = normalizeContentText(item.content).trim();
if (!content) continue;
inputItems.push({
type: 'message',
role: 'user',
content: [{ type: 'input_text', text: content }],
});
}
const maxOutputTokens = (
toFiniteNumber(openaiBody.max_output_tokens)
?? toFiniteNumber(openaiBody.max_completion_tokens)
?? toFiniteNumber(openaiBody.max_tokens)
?? 4096
);
const body: Record<string, unknown> = {
model: modelName,
stream,
max_output_tokens: maxOutputTokens,
input: inputItems,
};
if (systemContents.length > 0) {
body.instructions = systemContents.join('\n\n');
}
const temperature = toFiniteNumber(openaiBody.temperature);
if (temperature !== null) body.temperature = temperature;
const topP = toFiniteNumber(openaiBody.top_p);
if (topP !== null) body.top_p = topP;
if (openaiBody.metadata !== undefined) body.metadata = openaiBody.metadata;
if (openaiBody.reasoning !== undefined) body.reasoning = openaiBody.reasoning;
if (openaiBody.parallel_tool_calls !== undefined) body.parallel_tool_calls = openaiBody.parallel_tool_calls;
if (openaiBody.tools !== undefined) body.tools = convertOpenAiToolsToResponses(openaiBody.tools);
const responsesToolChoice = convertOpenAiToolChoiceToResponses(openaiBody.tool_choice);
if (responsesToolChoice !== undefined) body.tool_choice = responsesToolChoice;
return normalizeResponsesBodyForCompatibility(body);
}
function normalizeEndpointTypes(value: unknown): UpstreamEndpoint[] {
const raw = asTrimmedString(value).toLowerCase();
@@ -1104,9 +458,9 @@ export function buildUpstreamEndpointRequest(input: {
model: input.modelName,
stream: input.stream,
}
: convertOpenAiBodyToMessagesBody(input.openaiBody, input.modelName, input.stream)
: convertOpenAiBodyToAnthropicMessagesBody(input.openaiBody, input.modelName, input.stream)
);
const sanitizedBody = sanitizeMessagesBodyForAnthropic(body);
const sanitizedBody = sanitizeAnthropicMessagesBody(body);
const headers = ensureStreamAcceptHeader({
...commonHeaders,
@@ -1133,9 +487,9 @@ export function buildUpstreamEndpointRequest(input: {
model: input.modelName,
stream: input.stream,
}
: convertOpenAiBodyToResponsesBody(input.openaiBody, input.modelName, input.stream)
: convertOpenAiBodyToResponsesBodyViaTransformer(input.openaiBody, input.modelName, input.stream)
);
const body = sanitizeResponsesBodyForProxy(rawBody, input.modelName, input.stream);
const body = sanitizeResponsesBodyForProxyViaTransformer(rawBody, input.modelName, input.stream);
const headers = ensureStreamAcceptHeader({
...commonHeaders,
@@ -1291,3 +645,4 @@ export function isEndpointDowngradeError(status: number, upstreamErrorText?: str
)
);
}
+248
View File
@@ -0,0 +1,248 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
const selectChannelMock = vi.fn();
const selectNextChannelMock = vi.fn();
const recordSuccessMock = vi.fn();
const recordFailureMock = vi.fn();
const refreshModelsAndRebuildRoutesMock = vi.fn();
const reportProxyAllFailedMock = vi.fn();
const reportTokenExpiredMock = vi.fn();
const estimateProxyCostMock = vi.fn(async () => 0);
const saveProxyVideoTaskMock = vi.fn();
const getProxyVideoTaskByPublicIdMock = vi.fn();
const deleteProxyVideoTaskByPublicIdMock = vi.fn();
const refreshProxyVideoTaskSnapshotMock = vi.fn();
vi.mock('undici', async () => {
const actual = await vi.importActual<typeof import('undici')>('undici');
return {
...actual,
fetch: (...args: unknown[]) => fetchMock(...args),
};
});
vi.mock('../../services/tokenRouter.js', () => ({
tokenRouter: {
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
},
}));
vi.mock('../../services/modelService.js', () => ({
refreshModelsAndRebuildRoutes: (...args: unknown[]) => refreshModelsAndRebuildRoutesMock(...args),
}));
vi.mock('../../services/alertService.js', () => ({
reportProxyAllFailed: (...args: unknown[]) => reportProxyAllFailedMock(...args),
reportTokenExpired: (...args: unknown[]) => reportTokenExpiredMock(...args),
}));
vi.mock('../../services/alertRules.js', () => ({
isTokenExpiredError: () => false,
}));
vi.mock('../../services/modelPricingService.js', () => ({
estimateProxyCost: (arg: any) => estimateProxyCostMock(arg),
}));
vi.mock('../../services/proxyRetryPolicy.js', () => ({
shouldRetryProxyRequest: () => false,
}));
vi.mock('../../services/proxyVideoTaskStore.js', () => ({
saveProxyVideoTask: (...args: unknown[]) => saveProxyVideoTaskMock(...args),
getProxyVideoTaskByPublicId: (...args: unknown[]) => getProxyVideoTaskByPublicIdMock(...args),
deleteProxyVideoTaskByPublicId: (...args: unknown[]) => deleteProxyVideoTaskByPublicIdMock(...args),
refreshProxyVideoTaskSnapshot: (...args: unknown[]) => refreshProxyVideoTaskSnapshotMock(...args),
}));
describe('/v1/videos routes', () => {
let app: FastifyInstance;
const buildMultipartBody = (boundary: string) => Buffer.from(
`--${boundary}\r\n`
+ `Content-Disposition: form-data; name="model"\r\n\r\n`
+ `sora-2\r\n`
+ `--${boundary}\r\n`
+ `Content-Disposition: form-data; name="prompt"\r\n\r\n`
+ `a cat walking\r\n`
+ `--${boundary}\r\n`
+ `Content-Disposition: form-data; name="input_reference"; filename="cat.png"\r\n`
+ `Content-Type: image/png\r\n\r\n`
+ `pngdata\r\n`
+ `--${boundary}--\r\n`,
);
beforeAll(async () => {
const { videosProxyRoute } = await import('./videos.js');
app = Fastify();
await app.register(videosProxyRoute);
});
beforeEach(() => {
fetchMock.mockReset();
selectChannelMock.mockReset();
selectNextChannelMock.mockReset();
recordSuccessMock.mockReset();
recordFailureMock.mockReset();
refreshModelsAndRebuildRoutesMock.mockReset();
reportProxyAllFailedMock.mockReset();
reportTokenExpiredMock.mockReset();
estimateProxyCostMock.mockClear();
saveProxyVideoTaskMock.mockReset();
getProxyVideoTaskByPublicIdMock.mockReset();
deleteProxyVideoTaskByPublicIdMock.mockReset();
refreshProxyVideoTaskSnapshotMock.mockReset();
selectChannelMock.mockReturnValue({
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'demo-site', url: 'https://upstream.example.com', platform: 'openai' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'sora-2',
});
selectNextChannelMock.mockReturnValue(null);
});
afterAll(async () => {
await app.close();
});
it('creates an upstream video task and stores a local public id mapping', async () => {
saveProxyVideoTaskMock.mockResolvedValue({
publicId: 'vid_local_123',
upstreamVideoId: 'vid_upstream_123',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_123',
object: 'video',
status: 'queued',
model: 'sora-2',
prompt: 'a cat walking',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'POST',
url: '/v1/videos',
payload: {
model: 'sora-2',
prompt: 'a cat walking',
},
});
expect(response.statusCode).toBe(200);
expect(saveProxyVideoTaskMock).toHaveBeenCalledWith(expect.objectContaining({
upstreamVideoId: 'vid_upstream_123',
requestedModel: 'sora-2',
actualModel: 'sora-2',
lastUpstreamStatus: 200,
statusSnapshot: expect.objectContaining({
id: 'vid_upstream_123',
status: 'queued',
}),
}));
expect(response.json()).toMatchObject({
id: 'vid_local_123',
object: 'video',
status: 'queued',
});
});
it('accepts multipart video create requests', async () => {
saveProxyVideoTaskMock.mockResolvedValue({
publicId: 'vid_local_456',
upstreamVideoId: 'vid_upstream_456',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_456',
object: 'video',
status: 'queued',
model: 'sora-2',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const boundary = 'metapi-video-boundary';
const response = await app.inject({
method: 'POST',
url: '/v1/videos',
headers: {
'content-type': `multipart/form-data; boundary=${boundary}`,
},
payload: buildMultipartBody(boundary),
});
expect(response.statusCode).toBe(200);
expect(response.json()).toMatchObject({
id: 'vid_local_456',
object: 'video',
status: 'queued',
});
});
it('resolves local video ids back to the upstream task on GET', async () => {
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_123',
upstreamVideoId: 'vid_upstream_123',
siteUrl: 'https://upstream.example.com',
tokenValue: 'sk-demo',
});
fetchMock.mockResolvedValue(new Response(JSON.stringify({
id: 'vid_upstream_123',
object: 'video',
status: 'running',
model: 'sora-2',
}), {
status: 200,
headers: { 'content-type': 'application/json' },
}));
const response = await app.inject({
method: 'GET',
url: '/v1/videos/vid_local_123',
});
expect(response.statusCode).toBe(200);
const [targetUrl] = fetchMock.mock.calls[0] as [string, RequestInit];
expect(targetUrl).toBe('https://upstream.example.com/v1/videos/vid_upstream_123');
expect(refreshProxyVideoTaskSnapshotMock).toHaveBeenCalledWith('vid_local_123', expect.objectContaining({
lastUpstreamStatus: 200,
statusSnapshot: expect.objectContaining({
id: 'vid_upstream_123',
status: 'running',
}),
}));
expect(response.json()).toMatchObject({
id: 'vid_local_123',
object: 'video',
status: 'running',
});
});
it('deletes the upstream task and local mapping on DELETE', async () => {
getProxyVideoTaskByPublicIdMock.mockResolvedValue({
publicId: 'vid_local_123',
upstreamVideoId: 'vid_upstream_123',
siteUrl: 'https://upstream.example.com',
tokenValue: 'sk-demo',
});
fetchMock.mockResolvedValue(new Response(null, { status: 204 }));
const response = await app.inject({
method: 'DELETE',
url: '/v1/videos/vid_local_123',
});
expect(response.statusCode).toBe(204);
expect(deleteProxyVideoTaskByPublicIdMock).toHaveBeenCalledWith('vid_local_123');
});
});
+229
View File
@@ -0,0 +1,229 @@
import { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { fetch } from 'undici';
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 { estimateProxyCost } from '../../services/modelPricingService.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
import { withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { cloneFormDataWithOverrides, ensureMultipartBufferParser, parseMultipartFormData } from './multipart.js';
import {
deleteProxyVideoTaskByPublicId,
getProxyVideoTaskByPublicId,
refreshProxyVideoTaskSnapshot,
saveProxyVideoTask,
} from '../../services/proxyVideoTaskStore.js';
const MAX_RETRIES = 2;
function rewriteVideoResponsePublicId(payload: unknown, publicId: string): unknown {
if (!payload || typeof payload !== 'object') return payload;
return {
...(payload as Record<string, unknown>),
id: publicId,
};
}
export async function videosProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app);
app.post('/v1/videos', async (request: FastifyRequest, reply: FastifyReply) => {
const multipartForm = await parseMultipartFormData(request);
const jsonBody = (!multipartForm && request.body && typeof request.body === 'object')
? request.body as Record<string, unknown>
: null;
const requestedModel = typeof multipartForm?.get('model') === 'string'
? String(multipartForm.get('model')).trim()
: (typeof jsonBody?.model === 'string' ? jsonBody.model.trim() : '');
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 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({
model: requestedModel,
reason: 'No available channels after retries',
});
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
excludeChannelIds.push(selected.channel.id);
const targetUrl = `${selected.site.url}/v1/videos`;
const startTime = Date.now();
try {
const requestInit = multipartForm
? withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
Authorization: `Bearer ${selected.tokenValue}`,
},
body: cloneFormDataWithOverrides(multipartForm, {
model: selected.actualModel || requestedModel,
}) as any,
})
: withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: {
'Content-Type': 'application/json',
Authorization: `Bearer ${selected.tokenValue}`,
},
body: JSON.stringify({
...(jsonBody || {}),
model: selected.actualModel || requestedModel,
}),
});
const upstream = await fetch(targetUrl, requestInit);
const text = await upstream.text();
if (!upstream.ok) {
tokenRouter.recordFailure(selected.channel.id);
if (isTokenExpiredError({ status: upstream.status, message: text })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${upstream.status}`,
});
}
if (shouldRetryProxyRequest(upstream.status, text) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${upstream.status}`,
});
return reply.code(upstream.status).send({ error: { message: text, type: 'upstream_error' } });
}
let data: any = {};
try { data = JSON.parse(text); } catch { data = {}; }
const upstreamVideoId = typeof data?.id === 'string' ? data.id.trim() : '';
if (!upstreamVideoId) {
return reply.code(502).send({
error: { message: 'Upstream video response did not include id', type: 'upstream_error' },
});
}
const mapping = await saveProxyVideoTask({
upstreamVideoId,
siteUrl: selected.site.url,
tokenValue: selected.tokenValue,
requestedModel,
actualModel: selected.actualModel || requestedModel,
channelId: typeof selected.channel.id === 'number' ? selected.channel.id : null,
accountId: typeof selected.account.id === 'number' ? selected.account.id : null,
statusSnapshot: data,
upstreamResponseMeta: {
contentType: upstream.headers.get('content-type') || 'application/json',
},
lastUpstreamStatus: upstream.status,
});
const latency = Date.now() - startTime;
const estimatedCost = await estimateProxyCost({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
return reply.code(upstream.status).send(rewriteVideoResponsePublicId(data, mapping.publicId));
} catch (error: any) {
tokenRouter.recordFailure(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: error?.message || 'network failure',
});
return reply.code(502).send({
error: { message: error?.message || 'network failure', type: 'upstream_error' },
});
}
}
});
app.get('/v1/videos/:id', async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => {
const mapping = await getProxyVideoTaskByPublicId(request.params.id);
if (!mapping) {
return reply.code(404).send({
error: { message: 'Video task not found', type: 'not_found_error' },
});
}
const upstream = await fetch(`${mapping.siteUrl}/v1/videos/${encodeURIComponent(mapping.upstreamVideoId)}`, {
method: 'GET',
headers: {
Authorization: `Bearer ${mapping.tokenValue}`,
},
});
const text = await upstream.text();
try {
const data = JSON.parse(text);
await refreshProxyVideoTaskSnapshot(mapping.publicId, {
statusSnapshot: data,
upstreamResponseMeta: {
contentType: upstream.headers.get('content-type') || 'application/json',
},
lastUpstreamStatus: upstream.status,
});
return reply.code(upstream.status).send(rewriteVideoResponsePublicId(data, mapping.publicId));
} catch {
return reply.code(upstream.status).type(upstream.headers.get('content-type') || 'application/json').send(text);
}
});
app.delete('/v1/videos/:id', async (request: FastifyRequest<{ Params: { id: string } }>, reply: FastifyReply) => {
const mapping = await getProxyVideoTaskByPublicId(request.params.id);
if (!mapping) {
return reply.code(404).send({
error: { message: 'Video task not found', type: 'not_found_error' },
});
}
const upstream = await fetch(`${mapping.siteUrl}/v1/videos/${encodeURIComponent(mapping.upstreamVideoId)}`, {
method: 'DELETE',
headers: {
Authorization: `Bearer ${mapping.tokenValue}`,
},
});
if (upstream.ok) {
await deleteProxyVideoTaskByPublicId(mapping.publicId);
return reply.code(upstream.status).send();
}
const text = await upstream.text();
return reply.code(upstream.status).send({
error: { message: text || 'Upstream delete failed', type: 'upstream_error' },
});
});
}