Add comprehensive tests for Anthropics and Gemini transformers
- Implement tests for anthropicMessagesTransformer including serialization and event handling in stream.test.ts. - Add usage extraction tests for anthropicMessagesUsage in usage.test.ts. - Create extensive tests for Gemini generate content aggregation and outbound usage in aggregator-outbound-usage.test.ts. - Introduce tests for Gemini stream handling and payload parsing in stream.test.ts. - Enhance OpenAI response serialization tests to ensure proper mapping of usage details and preservation of annotations in outbound.test.ts.
This commit is contained in:
cita
@@ -14,16 +14,44 @@ describe('proxy route architecture boundaries', () => {
|
||||
});
|
||||
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(');
|
||||
expect(source).not.toContain('serializeAnthropicRawSseEvent');
|
||||
expect(source).not.toContain('syncAnthropicRawStreamStateFromEvent');
|
||||
expect(source).not.toContain('isAnthropicRawSseEventName');
|
||||
expect(source).not.toContain('serializeAnthropicFinalAsStream');
|
||||
expect(source).not.toContain('function shouldRetryClaudeMessagesWithNormalizedBody(');
|
||||
expect(source).not.toContain('const emitNormalizedFinalAsStream =');
|
||||
expect(source).not.toContain("from './protocolCompat.js'");
|
||||
expect(source).not.toContain("from './chatStreamCompat.js'");
|
||||
expect(source).not.toContain('const promoteResponsesCandidate =');
|
||||
expect(source).not.toContain('shouldRetryClaudeMessagesWithNormalizedBody(');
|
||||
expect(source).not.toContain('buildOpenAiSyntheticFinalStream(');
|
||||
expect(source).toContain('anthropicMessagesTransformer.consumeSseEventBlock(');
|
||||
expect(source).toContain('anthropicMessagesTransformer.serializeUpstreamFinalAsStream(');
|
||||
expect(source).toContain('openAiChatTransformer.serializeUpstreamFinalAsStream(');
|
||||
});
|
||||
|
||||
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(');
|
||||
expect(source).not.toContain("from '../../transformers/openai/responses/conversion.js'");
|
||||
expect(source).not.toContain("from '../../transformers/openai/responses/outbound.js'");
|
||||
expect(source).not.toContain("from '../../transformers/openai/responses/aggregator.js'");
|
||||
expect(source).not.toContain('function buildResponsesCompatibilityBodies(');
|
||||
expect(source).not.toContain('function buildResponsesCompatibilityHeaderCandidates(');
|
||||
expect(source).not.toContain('function shouldRetryResponsesCompatibility(');
|
||||
expect(source).not.toContain("from './protocolCompat.js'");
|
||||
expect(source).not.toContain('function shouldDowngradeFromChatToMessagesForResponses(');
|
||||
expect(source).not.toContain('function normalizeText(');
|
||||
expect(source).toContain('openAiResponsesTransformer.inbound.toOpenAiBody(');
|
||||
expect(source).toContain('openAiResponsesTransformer.compatibility.buildRetryBodies(');
|
||||
expect(source).toContain('openAiResponsesTransformer.compatibility.buildRetryHeaders(');
|
||||
expect(source).toContain('openAiResponsesTransformer.compatibility.shouldRetry(');
|
||||
expect(source).toContain('openAiResponsesTransformer.compatibility.shouldDowngradeChatToMessages(');
|
||||
expect(source).toContain('openAiResponsesTransformer.aggregator.createState(');
|
||||
expect(source).toContain('openAiResponsesTransformer.aggregator.serialize(');
|
||||
expect(source).toContain('openAiResponsesTransformer.aggregator.complete(');
|
||||
expect(source).toContain('openAiResponsesTransformer.outbound.serializeFinal(');
|
||||
});
|
||||
|
||||
it('removes normalizeContentText from upstream endpoint routing', () => {
|
||||
@@ -34,8 +62,14 @@ describe('proxy route architecture boundaries', () => {
|
||||
|
||||
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)');
|
||||
expect(source).not.toContain('outbound.serializeAggregateResponse(');
|
||||
expect(source).not.toContain('aggregator.apply(');
|
||||
expect(source).not.toContain('stream.serializeAggregateSsePayload(');
|
||||
expect(source).not.toContain('stream.serializeAggregateJsonPayload(');
|
||||
expect(source).not.toContain('stream.applyJsonPayloadToAggregate(');
|
||||
expect(source).not.toContain('stream.parseSsePayloads(');
|
||||
expect(source).toContain('stream.consumeUpstreamSseBuffer(');
|
||||
expect(source).toContain('stream.serializeUpstreamJsonPayload(');
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -565,8 +565,15 @@ describe('chat proxy stream behavior', () => {
|
||||
|
||||
expect(Array.isArray(firstBody.messages)).toBe(true);
|
||||
expect(Array.isArray(secondBody.messages)).toBe(true);
|
||||
expect(typeof secondBody.messages[0]?.content).toBe('string');
|
||||
expect(secondBody.messages[0]?.content).toContain('hello');
|
||||
expect(Array.isArray(secondBody.messages[0]?.content)).toBe(true);
|
||||
expect(secondBody.messages[0]?.content).toEqual(
|
||||
expect.arrayContaining([
|
||||
expect.objectContaining({
|
||||
type: 'text',
|
||||
text: expect.stringContaining('hello'),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
it('downgrades to next endpoint when normalized Claude fallback still returns messages is required', async () => {
|
||||
|
||||
@@ -16,8 +16,8 @@ import {
|
||||
isEndpointDispatchDeniedError,
|
||||
isEndpointDowngradeError,
|
||||
isUnsupportedMediaTypeError,
|
||||
promoteResponsesCandidateAfterLegacyChatError,
|
||||
resolveUpstreamEndpointCandidates,
|
||||
shouldPreferResponsesAfterLegacyChatError,
|
||||
} from './upstreamEndpoint.js';
|
||||
import {
|
||||
ensureModelAllowedForDownstreamKey,
|
||||
@@ -29,30 +29,9 @@ 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';
|
||||
import { anthropicMessagesTransformer } from '../../transformers/anthropic/messages/index.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
function shouldRetryClaudeMessagesWithNormalizedBody(
|
||||
downstreamFormat: DownstreamFormat,
|
||||
endpointPath: string,
|
||||
status: number,
|
||||
upstreamErrorText: string,
|
||||
): boolean {
|
||||
if (downstreamFormat !== 'claude') return false;
|
||||
if (!endpointPath.includes('/v1/messages')) return false;
|
||||
if (status < 400 || status >= 500) return false;
|
||||
return /messages\s+is\s+required/i.test(upstreamErrorText);
|
||||
}
|
||||
|
||||
function isMessagesRequiredError(upstreamErrorText: string): boolean {
|
||||
return /messages\s+is\s+required/i.test(upstreamErrorText);
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object';
|
||||
@@ -125,27 +104,6 @@ async function handleChatProxyRequest(
|
||||
];
|
||||
let startTime = Date.now();
|
||||
|
||||
const promoteResponsesCandidate = (currentEndpoint: string | null | undefined, upstreamErrorText?: string | null, status = 0) => {
|
||||
if (!shouldPreferResponsesAfterLegacyChatError({
|
||||
status,
|
||||
upstreamErrorText,
|
||||
downstreamFormat,
|
||||
sitePlatform: selected.site.platform,
|
||||
modelName,
|
||||
requestedModelHint: requestedModel,
|
||||
currentEndpoint: currentEndpoint as any,
|
||||
})) {
|
||||
return;
|
||||
}
|
||||
|
||||
const currentIndex = endpointCandidates.findIndex((endpoint) => endpoint === currentEndpoint);
|
||||
const responsesIndex = endpointCandidates.indexOf('responses');
|
||||
if (currentIndex < 0 || responsesIndex < 0 || responsesIndex <= currentIndex + 1) return;
|
||||
|
||||
endpointCandidates.splice(responsesIndex, 1);
|
||||
endpointCandidates.splice(currentIndex + 1, 0, 'responses');
|
||||
};
|
||||
|
||||
try {
|
||||
const endpointResult = await executeEndpointFlow({
|
||||
siteUrl: selected.site.url,
|
||||
@@ -172,12 +130,12 @@ async function handleChatProxyRequest(
|
||||
};
|
||||
},
|
||||
tryRecover: async (ctx) => {
|
||||
if (shouldRetryClaudeMessagesWithNormalizedBody(
|
||||
if (anthropicMessagesTransformer.compatibility.shouldRetryNormalizedBody({
|
||||
downstreamFormat,
|
||||
ctx.request.path,
|
||||
ctx.response.status,
|
||||
ctx.rawErrText,
|
||||
)) {
|
||||
endpointPath: ctx.request.path,
|
||||
status: ctx.response.status,
|
||||
upstreamErrorText: ctx.rawErrText,
|
||||
})) {
|
||||
const normalizedClaudeRequest = buildUpstreamEndpointRequest({
|
||||
endpoint: ctx.request.endpoint,
|
||||
modelName,
|
||||
@@ -253,11 +211,19 @@ async function handleChatProxyRequest(
|
||||
},
|
||||
shouldDowngrade: (ctx) => (
|
||||
(() => {
|
||||
promoteResponsesCandidate(ctx.request.endpoint, ctx.rawErrText, ctx.response.status);
|
||||
promoteResponsesCandidateAfterLegacyChatError(endpointCandidates, {
|
||||
status: ctx.response.status,
|
||||
upstreamErrorText: ctx.rawErrText,
|
||||
downstreamFormat,
|
||||
sitePlatform: selected.site.platform,
|
||||
modelName,
|
||||
requestedModelHint: requestedModel,
|
||||
currentEndpoint: ctx.request.endpoint,
|
||||
});
|
||||
return (
|
||||
ctx.response.status >= 500
|
||||
|| isEndpointDowngradeError(ctx.response.status, ctx.rawErrText)
|
||||
|| isMessagesRequiredError(ctx.rawErrText)
|
||||
|| anthropicMessagesTransformer.compatibility.isMessagesRequiredError(ctx.rawErrText)
|
||||
|| isEndpointDispatchDeniedError(ctx.response.status, ctx.rawErrText)
|
||||
);
|
||||
})()
|
||||
@@ -338,23 +304,6 @@ async function handleChatProxyRequest(
|
||||
writeLines(downstreamTransformer.serializeDone(streamContext, claudeContext));
|
||||
};
|
||||
|
||||
const emitNormalizedFinalAsStream = (upstreamData: unknown, 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 = openAiChatTransformer.buildSyntheticChunks(normalizedFinal);
|
||||
for (const chunk of syntheticChunks) {
|
||||
reply.raw.write(`data: ${JSON.stringify(chunk)}\n\n`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
writeLines(serializeAnthropicFinalAsStream(normalizedFinal, streamContext, claudeContext));
|
||||
};
|
||||
|
||||
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
|
||||
if (!upstreamContentType.includes('text/event-stream')) {
|
||||
const fallbackText = await upstream.text();
|
||||
@@ -366,7 +315,25 @@ async function handleChatProxyRequest(
|
||||
}
|
||||
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(fallbackData));
|
||||
emitNormalizedFinalAsStream(fallbackData, fallbackText);
|
||||
if (downstreamFormat === 'openai') {
|
||||
const syntheticLines = openAiChatTransformer.serializeUpstreamFinalAsStream(
|
||||
fallbackData,
|
||||
modelName,
|
||||
fallbackText,
|
||||
streamContext,
|
||||
);
|
||||
writeLines(syntheticLines);
|
||||
} else {
|
||||
writeLines(
|
||||
anthropicMessagesTransformer.serializeUpstreamFinalAsStream(
|
||||
fallbackData,
|
||||
modelName,
|
||||
fallbackText,
|
||||
streamContext,
|
||||
claudeContext,
|
||||
),
|
||||
);
|
||||
}
|
||||
writeDone();
|
||||
reply.raw.end();
|
||||
|
||||
@@ -437,37 +404,37 @@ async function handleChatProxyRequest(
|
||||
}
|
||||
|
||||
let parsedPayload: unknown = null;
|
||||
try {
|
||||
parsedPayload = JSON.parse(eventBlock.data);
|
||||
} catch {
|
||||
parsedPayload = null;
|
||||
if (downstreamFormat === 'claude') {
|
||||
const consumed = anthropicMessagesTransformer.consumeSseEventBlock(
|
||||
eventBlock,
|
||||
streamContext,
|
||||
claudeContext,
|
||||
modelName,
|
||||
);
|
||||
parsedPayload = consumed.parsedPayload;
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
|
||||
}
|
||||
if (consumed.handled) {
|
||||
writeLines(consumed.lines);
|
||||
if (consumed.done) {
|
||||
shouldTerminateEarly = true;
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
parsedPayload = JSON.parse(eventBlock.data);
|
||||
} catch {
|
||||
parsedPayload = null;
|
||||
}
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(parsedPayload));
|
||||
}
|
||||
}
|
||||
|
||||
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 = isAnthropicRawSseEventName(eventBlock.event)
|
||||
? eventBlock.event
|
||||
: (isAnthropicRawSseEventName(payloadType) ? payloadType : '');
|
||||
|
||||
if (claudeEventName) {
|
||||
syncAnthropicRawStreamStateFromEvent(
|
||||
claudeEventName,
|
||||
parsedPayload,
|
||||
streamContext,
|
||||
claudeContext,
|
||||
);
|
||||
reply.raw.write(serializeAnthropicRawSseEvent(claudeEventName, eventBlock.data));
|
||||
if (claudeContext.doneSent) {
|
||||
shouldTerminateEarly = true;
|
||||
break;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
}
|
||||
const normalizedEvent = downstreamTransformer.transformStreamEvent(parsedPayload, streamContext, modelName);
|
||||
writeLines(downstreamTransformer.serializeStreamEvent(normalizedEvent, streamContext, claudeContext));
|
||||
if (downstreamFormat === 'claude' && claudeContext.doneSent) {
|
||||
|
||||
@@ -28,6 +28,21 @@ vi.mock('../../services/downstreamApiKeyService.js', () => ({
|
||||
consumeManagedKeyRequest: (...args: unknown[]) => consumeManagedKeyRequestMock(...args),
|
||||
}));
|
||||
|
||||
function parseSsePayloads(body: string): Array<Record<string, unknown>> {
|
||||
return body
|
||||
.split('\n\n')
|
||||
.map((block) => block.trim())
|
||||
.filter(Boolean)
|
||||
.map((block) => block
|
||||
.split('\n')
|
||||
.filter((line) => line.startsWith('data:'))
|
||||
.map((line) => line.slice(5).trimStart())
|
||||
.join('\n'),
|
||||
)
|
||||
.filter((data) => data && data !== '[DONE]')
|
||||
.map((data) => JSON.parse(data) as Record<string, unknown>);
|
||||
}
|
||||
|
||||
describe('gemini native proxy routes', () => {
|
||||
let app: FastifyInstance;
|
||||
|
||||
@@ -301,7 +316,7 @@ describe('gemini native proxy routes', () => {
|
||||
});
|
||||
});
|
||||
|
||||
it('aggregates non-sse json-array streaming payloads into a final Gemini response', async () => {
|
||||
it('keeps non-sse json-array streaming payloads on the wire as chunk responses', async () => {
|
||||
fetchMock.mockResolvedValue(new Response(JSON.stringify([
|
||||
{
|
||||
candidates: [
|
||||
@@ -348,32 +363,45 @@ describe('gemini native proxy routes', () => {
|
||||
});
|
||||
|
||||
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' },
|
||||
],
|
||||
expect(response.json()).toEqual([
|
||||
{
|
||||
responseId: '',
|
||||
modelVersion: '',
|
||||
candidates: [
|
||||
{
|
||||
index: 0,
|
||||
finishReason: 'STOP',
|
||||
content: {
|
||||
role: 'model',
|
||||
parts: [{ text: 'first' }],
|
||||
},
|
||||
groundingMetadata: { source: 'web' },
|
||||
},
|
||||
groundingMetadata: { source: 'web' },
|
||||
citationMetadata: { citations: [{ startIndex: 0, endIndex: 5 }] },
|
||||
},
|
||||
],
|
||||
usageMetadata: {
|
||||
promptTokenCount: 11,
|
||||
candidatesTokenCount: 6,
|
||||
totalTokenCount: 17,
|
||||
cachedContentTokenCount: 2,
|
||||
thoughtsTokenCount: 3,
|
||||
],
|
||||
},
|
||||
});
|
||||
{
|
||||
responseId: '',
|
||||
modelVersion: '',
|
||||
candidates: [
|
||||
{
|
||||
index: 0,
|
||||
finishReason: 'STOP',
|
||||
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,
|
||||
},
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('derives gemini-3 thinkingLevel from OpenAI-style reasoning inputs in the runtime request path', async () => {
|
||||
@@ -428,12 +456,12 @@ describe('gemini native proxy routes', () => {
|
||||
},
|
||||
],
|
||||
generationConfig: {
|
||||
thinkingConfig: { thinkingLevel: 'high' },
|
||||
thinkingConfig: { thinkingLevel: 'high', includeThoughts: true },
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('aggregates SSE payloads through the transformer instead of raw passthrough', async () => {
|
||||
it('streams SSE payloads as normalized chunk events instead of cumulative aggregate snapshots', async () => {
|
||||
const encoder = new TextEncoder();
|
||||
const upstreamBody = new ReadableStream<Uint8Array>({
|
||||
start(controller) {
|
||||
@@ -466,12 +494,53 @@ describe('gemini native proxy routes', () => {
|
||||
|
||||
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"}]');
|
||||
const events = parseSsePayloads(response.body);
|
||||
|
||||
expect(events).toHaveLength(2);
|
||||
expect(events[0]).toMatchObject({
|
||||
responseId: 'resp-sse',
|
||||
modelVersion: 'gemini-2.5-flash',
|
||||
candidates: [
|
||||
{
|
||||
content: {
|
||||
parts: [{ text: 'first' }],
|
||||
},
|
||||
groundingMetadata: { source: 'web' },
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(events[0]).not.toMatchObject({
|
||||
candidates: [
|
||||
{
|
||||
citationMetadata: expect.anything(),
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(events[1]).toMatchObject({
|
||||
usageMetadata: {
|
||||
promptTokenCount: 11,
|
||||
candidatesTokenCount: 6,
|
||||
totalTokenCount: 17,
|
||||
cachedContentTokenCount: 2,
|
||||
thoughtsTokenCount: 3,
|
||||
},
|
||||
candidates: [
|
||||
{
|
||||
finishReason: 'STOP',
|
||||
content: {
|
||||
parts: [{ text: 'second', thoughtSignature: 'sig-1' }],
|
||||
},
|
||||
citationMetadata: { citations: [{ startIndex: 0, endIndex: 5 }] },
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(events[1]).not.toMatchObject({
|
||||
candidates: [
|
||||
{
|
||||
groundingMetadata: expect.anything(),
|
||||
},
|
||||
],
|
||||
});
|
||||
expect(response.body).not.toContain('\r\n\r\n');
|
||||
});
|
||||
|
||||
|
||||
@@ -119,6 +119,7 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
const generateContent = async (request: FastifyRequest, reply: FastifyReply) => {
|
||||
const apiVersion = resolveGeminiApiVersion(request);
|
||||
const modelActionPath = extractGeminiModelActionPath(request, apiVersion);
|
||||
const isStreamAction = modelActionPath.endsWith(':streamGenerateContent');
|
||||
const requestedModel = modelActionPath.replace(/^models\//, '').split(':')[0].trim();
|
||||
if (!requestedModel) {
|
||||
return reply.code(400).send({
|
||||
@@ -206,27 +207,23 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
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 consumed = geminiGenerateContentTransformer.stream.consumeUpstreamSseBuffer(
|
||||
aggregateState,
|
||||
rest + chunkText,
|
||||
);
|
||||
rest = consumed.rest;
|
||||
for (const line of consumed.lines) {
|
||||
reply.raw.write(line);
|
||||
}
|
||||
}
|
||||
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),
|
||||
),
|
||||
);
|
||||
const consumed = geminiGenerateContentTransformer.stream.consumeUpstreamSseBuffer(
|
||||
aggregateState,
|
||||
rest + tail,
|
||||
);
|
||||
for (const line of consumed.lines) {
|
||||
reply.raw.write(line);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
@@ -239,19 +236,13 @@ export async function geminiProxyRoute(app: FastifyInstance) {
|
||||
const text = await upstream.text();
|
||||
try {
|
||||
const parsed = JSON.parse(text);
|
||||
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);
|
||||
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
|
||||
return reply.code(upstream.status).send(
|
||||
geminiGenerateContentTransformer.outbound.serializeAggregateResponse(aggregateState),
|
||||
geminiGenerateContentTransformer.stream.serializeUpstreamJsonPayload(
|
||||
aggregateState,
|
||||
parsed,
|
||||
isStreamAction,
|
||||
),
|
||||
);
|
||||
} catch {
|
||||
return reply.code(upstream.status).type(contentType || 'application/json').send(text);
|
||||
|
||||
@@ -9,23 +9,13 @@ 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 {
|
||||
completeResponsesStream,
|
||||
createOpenAiResponsesAggregateState,
|
||||
failResponsesStream,
|
||||
serializeConvertedResponsesEvents,
|
||||
} from '../../transformers/openai/responses/aggregator.js';
|
||||
import {
|
||||
buildMinimalJsonHeadersForCompatibility,
|
||||
buildUpstreamEndpointRequest,
|
||||
isEndpointDowngradeError,
|
||||
isUnsupportedMediaTypeError,
|
||||
resolveUpstreamEndpointCandidates,
|
||||
type UpstreamEndpoint,
|
||||
} from './upstreamEndpoint.js';
|
||||
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from './downstreamPolicy.js';
|
||||
import { composeProxyLogMessage } from './logPathMeta.js';
|
||||
@@ -35,222 +25,10 @@ import { resolveProxyLogBilling } from './proxyBilling.js';
|
||||
|
||||
const MAX_RETRIES = 2;
|
||||
|
||||
function shouldDowngradeFromChatToMessagesForResponses(
|
||||
endpointPath: string,
|
||||
status: number,
|
||||
upstreamErrorText: string,
|
||||
): boolean {
|
||||
if (!endpointPath.includes('/chat/completions')) return false;
|
||||
if (status < 400 || status >= 500) return false;
|
||||
return /messages\s+is\s+required/i.test(upstreamErrorText);
|
||||
}
|
||||
|
||||
function parseUpstreamErrorShape(rawText: string): {
|
||||
type: string;
|
||||
code: string;
|
||||
message: string;
|
||||
} {
|
||||
try {
|
||||
const parsed = JSON.parse(rawText) as Record<string, unknown>;
|
||||
const error = (parsed.error && typeof parsed.error === 'object')
|
||||
? parsed.error as Record<string, unknown>
|
||||
: parsed;
|
||||
return {
|
||||
type: typeof error.type === 'string' ? error.type.trim().toLowerCase() : '',
|
||||
code: typeof error.code === 'string' ? error.code.trim().toLowerCase() : '',
|
||||
message: typeof error.message === 'string' ? error.message.trim() : '',
|
||||
};
|
||||
} catch {
|
||||
return { type: '', code: '', message: '' };
|
||||
}
|
||||
}
|
||||
|
||||
function stripResponsesMetadata(
|
||||
body: Record<string, unknown>,
|
||||
): Record<string, unknown> | null {
|
||||
if (!Object.prototype.hasOwnProperty.call(body, 'metadata')) return null;
|
||||
const next = { ...body };
|
||||
delete next.metadata;
|
||||
return next;
|
||||
}
|
||||
|
||||
function toFiniteNumber(value: unknown): number | null {
|
||||
return typeof value === 'number' && Number.isFinite(value) ? value : null;
|
||||
}
|
||||
|
||||
function buildCoreResponsesBody(
|
||||
body: Record<string, unknown>,
|
||||
): Record<string, unknown> | null {
|
||||
const model = typeof body.model === 'string' ? body.model.trim() : '';
|
||||
if (!model) return null;
|
||||
if (body.input === undefined) return null;
|
||||
|
||||
const core: Record<string, unknown> = {
|
||||
model,
|
||||
input: body.input,
|
||||
stream: body.stream === true,
|
||||
};
|
||||
|
||||
const maxOutputTokens = toFiniteNumber(body.max_output_tokens);
|
||||
if (maxOutputTokens !== null && maxOutputTokens > 0) {
|
||||
core.max_output_tokens = Math.trunc(maxOutputTokens);
|
||||
}
|
||||
|
||||
const temperature = toFiniteNumber(body.temperature);
|
||||
if (temperature !== null) core.temperature = temperature;
|
||||
|
||||
const topP = toFiniteNumber(body.top_p);
|
||||
if (topP !== null) core.top_p = topP;
|
||||
|
||||
const instructions = typeof body.instructions === 'string' ? body.instructions.trim() : '';
|
||||
if (instructions) core.instructions = instructions;
|
||||
|
||||
return core;
|
||||
}
|
||||
|
||||
function buildStrictResponsesBody(
|
||||
body: Record<string, unknown>,
|
||||
): Record<string, unknown> | null {
|
||||
const model = typeof body.model === 'string' ? body.model.trim() : '';
|
||||
if (!model) return null;
|
||||
if (body.input === undefined) return null;
|
||||
|
||||
return {
|
||||
model,
|
||||
input: body.input,
|
||||
stream: body.stream === true,
|
||||
};
|
||||
}
|
||||
|
||||
function buildResponsesCompatibilityBodies(
|
||||
body: Record<string, unknown>,
|
||||
): Record<string, unknown>[] {
|
||||
const candidates: Record<string, unknown>[] = [];
|
||||
const seen = new Set<string>();
|
||||
try {
|
||||
const originalKey = JSON.stringify(body);
|
||||
if (originalKey) seen.add(originalKey);
|
||||
} catch {
|
||||
// ignore non-serializable bodies
|
||||
}
|
||||
const push = (next: Record<string, unknown> | null) => {
|
||||
if (!next) return;
|
||||
let key = '';
|
||||
try {
|
||||
key = JSON.stringify(next);
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
if (!key || seen.has(key)) return;
|
||||
seen.add(key);
|
||||
candidates.push(next);
|
||||
};
|
||||
|
||||
push(stripResponsesMetadata(body));
|
||||
push(buildCoreResponsesBody(body));
|
||||
push(buildStrictResponsesBody(body));
|
||||
return candidates;
|
||||
}
|
||||
|
||||
function buildResponsesCompatibilityHeaderCandidates(
|
||||
headers: Record<string, string>,
|
||||
stream: boolean,
|
||||
): Record<string, string>[] {
|
||||
const candidates: Record<string, string>[] = [];
|
||||
const seen = new Set<string>();
|
||||
const push = (next: Record<string, string>) => {
|
||||
const normalizedEntries = Object.entries(next)
|
||||
.filter(([, value]) => typeof value === 'string' && value.trim().length > 0)
|
||||
.map(([key, value]) => [key.toLowerCase(), value] as const)
|
||||
.sort(([a], [b]) => a.localeCompare(b));
|
||||
const key = JSON.stringify(normalizedEntries);
|
||||
if (!key || seen.has(key)) return;
|
||||
seen.add(key);
|
||||
candidates.push(Object.fromEntries(normalizedEntries));
|
||||
};
|
||||
|
||||
push(headers);
|
||||
|
||||
const minimal: Record<string, string> = {};
|
||||
for (const [rawKey, rawValue] of Object.entries(headers)) {
|
||||
const key = rawKey.toLowerCase();
|
||||
if (
|
||||
key === 'authorization'
|
||||
|| key === 'x-api-key'
|
||||
|| key === 'content-type'
|
||||
|| key === 'accept'
|
||||
) {
|
||||
minimal[key] = rawValue;
|
||||
}
|
||||
}
|
||||
if (!minimal['content-type']) minimal['content-type'] = 'application/json';
|
||||
if (stream && !minimal.accept) minimal.accept = 'text/event-stream';
|
||||
push(minimal);
|
||||
|
||||
return candidates;
|
||||
}
|
||||
|
||||
function shouldRetryResponsesCompatibility(input: {
|
||||
endpoint: UpstreamEndpoint;
|
||||
status: number;
|
||||
rawErrText: string;
|
||||
}): boolean {
|
||||
if (input.endpoint !== 'responses') return false;
|
||||
if (input.status !== 400) return false;
|
||||
const parsedError = parseUpstreamErrorShape(input.rawErrText);
|
||||
const type = parsedError.type.trim().toLowerCase();
|
||||
const code = parsedError.code.trim().toLowerCase();
|
||||
const message = parsedError.message.trim().toLowerCase();
|
||||
const compact = `${type} ${code} ${message}`.trim();
|
||||
const rawCompact = (input.rawErrText || '').toLowerCase();
|
||||
|
||||
// Authentication/authorization failures should not enter compatibility retries.
|
||||
if (
|
||||
compact.includes('invalid_api_key')
|
||||
|| compact.includes('authentication')
|
||||
|| compact.includes('unauthorized')
|
||||
|| compact.includes('forbidden')
|
||||
|| compact.includes('insufficient_quota')
|
||||
|| compact.includes('rate_limit')
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (type === 'upstream_error' || code === 'upstream_error') return true;
|
||||
if (message === 'upstream_error' || message === 'upstream request failed') return true;
|
||||
if (rawCompact.includes('upstream_error')) return true;
|
||||
|
||||
// Many sub2api-compatible gateways return generic 400 for field incompatibilities.
|
||||
// Retry with progressively stricter payload/header candidates to maximize compatibility.
|
||||
return true;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return !!value && typeof value === 'object';
|
||||
}
|
||||
|
||||
function asTrimmedString(value: unknown): string {
|
||||
return typeof value === 'string' ? value.trim() : '';
|
||||
}
|
||||
|
||||
function normalizeText(value: unknown): string {
|
||||
if (typeof value === 'string') return value;
|
||||
if (Array.isArray(value)) {
|
||||
return value
|
||||
.map((item) => normalizeText(item))
|
||||
.filter((item) => item.length > 0)
|
||||
.join('\n');
|
||||
}
|
||||
if (isRecord(value)) {
|
||||
if (typeof value.text === 'string') return value.text;
|
||||
if (typeof value.content === 'string') return value.content;
|
||||
if (typeof value.input_text === 'string') return value.input_text;
|
||||
if (typeof value.output_text === 'string') return value.output_text;
|
||||
if (Array.isArray(value.content)) return normalizeText(value.content);
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
type UsageSummary = ReturnType<typeof parseProxyUsage>;
|
||||
|
||||
export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
@@ -295,7 +73,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
excludeChannelIds.push(selected.channel.id);
|
||||
|
||||
const modelName = selected.actualModel || requestedModel;
|
||||
const openAiBody = convertResponsesBodyToOpenAiBody(body, modelName, isStream);
|
||||
const openAiBody = openAiResponsesTransformer.inbound.toOpenAiBody(body, modelName, isStream);
|
||||
const endpointCandidates = await resolveUpstreamEndpointCandidates(
|
||||
{
|
||||
site: selected.site,
|
||||
@@ -340,15 +118,15 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
headers: endpointRequest.headers,
|
||||
body: endpointRequest.body as Record<string, unknown>,
|
||||
};
|
||||
},
|
||||
tryRecover: async (ctx) => {
|
||||
if (shouldRetryResponsesCompatibility({
|
||||
},
|
||||
tryRecover: async (ctx) => {
|
||||
if (openAiResponsesTransformer.compatibility.shouldRetry({
|
||||
endpoint: ctx.request.endpoint,
|
||||
status: ctx.response.status,
|
||||
rawErrText: ctx.rawErrText,
|
||||
})) {
|
||||
const compatibilityBodies = buildResponsesCompatibilityBodies(ctx.request.body);
|
||||
const compatibilityHeaders = buildResponsesCompatibilityHeaderCandidates(
|
||||
const compatibilityBodies = openAiResponsesTransformer.compatibility.buildRetryBodies(ctx.request.body);
|
||||
const compatibilityHeaders = openAiResponsesTransformer.compatibility.buildRetryHeaders(
|
||||
ctx.request.headers,
|
||||
isStream,
|
||||
);
|
||||
@@ -416,7 +194,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
shouldDowngrade: (ctx) => (
|
||||
ctx.response.status >= 500
|
||||
|| isEndpointDowngradeError(ctx.response.status, ctx.rawErrText)
|
||||
|| shouldDowngradeFromChatToMessagesForResponses(
|
||||
|| openAiResponsesTransformer.compatibility.shouldDowngradeChatToMessages(
|
||||
ctx.request.path,
|
||||
ctx.response.status,
|
||||
ctx.rawErrText,
|
||||
@@ -492,7 +270,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
|
||||
const passthroughResponsesStream = successfulUpstreamPath === '/v1/responses';
|
||||
const streamContext = openAiResponsesTransformer.createStreamContext(modelName);
|
||||
const responsesState = createOpenAiResponsesAggregateState(modelName);
|
||||
const responsesState = openAiResponsesTransformer.aggregator.createState(modelName);
|
||||
|
||||
const writeLines = (lines: string[]) => {
|
||||
for (const line of lines) reply.raw.write(line);
|
||||
@@ -505,7 +283,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
if (passthroughResponsesStream) {
|
||||
reply.raw.write('data: [DONE]\n\n');
|
||||
} else if (!responsesState.completed) {
|
||||
writeLines(completeResponsesStream(responsesState, streamContext, parsedUsage));
|
||||
writeLines(openAiResponsesTransformer.aggregator.complete(responsesState, streamContext, parsedUsage));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
@@ -537,13 +315,13 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
|| payloadType === 'response.failed'
|
||||
);
|
||||
if (isFailureEvent) {
|
||||
writeLines(failResponsesStream(responsesState, streamContext, parsedUsage, parsedPayload));
|
||||
writeLines(openAiResponsesTransformer.aggregator.fail(responsesState, streamContext, parsedUsage, parsedPayload));
|
||||
continue;
|
||||
}
|
||||
|
||||
if (parsedPayload && typeof parsedPayload === 'object') {
|
||||
const normalizedEvent = openAiResponsesTransformer.transformStreamEvent(parsedPayload, streamContext, modelName);
|
||||
writeLines(serializeConvertedResponsesEvents({
|
||||
writeLines(openAiResponsesTransformer.aggregator.serialize({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: normalizedEvent,
|
||||
@@ -552,7 +330,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
continue;
|
||||
}
|
||||
|
||||
writeLines(serializeConvertedResponsesEvents({
|
||||
writeLines(openAiResponsesTransformer.aggregator.serialize({
|
||||
state: responsesState,
|
||||
streamContext,
|
||||
event: { contentDelta: eventBlock.data },
|
||||
@@ -580,7 +358,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
} finally {
|
||||
reader.releaseLock();
|
||||
if (!passthroughResponsesStream && !responsesState.completed) {
|
||||
writeLines(completeResponsesStream(responsesState, streamContext, parsedUsage));
|
||||
writeLines(openAiResponsesTransformer.aggregator.complete(responsesState, streamContext, parsedUsage));
|
||||
}
|
||||
reply.raw.end();
|
||||
}
|
||||
@@ -632,7 +410,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
modelName,
|
||||
rawText,
|
||||
);
|
||||
const downstreamData = serializeResponsesFinalPayload({
|
||||
const downstreamData = openAiResponsesTransformer.outbound.serializeFinal({
|
||||
upstreamPayload: upstreamData,
|
||||
normalized,
|
||||
usage: parsedUsage,
|
||||
|
||||
@@ -578,7 +578,13 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
});
|
||||
|
||||
expect(request.path).toBe('/v1/messages');
|
||||
expect(request.body.system).toBe('be careful');
|
||||
expect(request.body.system).toEqual([
|
||||
{
|
||||
type: 'text',
|
||||
text: 'be careful',
|
||||
cache_control: { type: 'ephemeral' },
|
||||
},
|
||||
]);
|
||||
expect(request.body.messages).toEqual([
|
||||
{
|
||||
role: 'user',
|
||||
@@ -733,7 +739,13 @@ describe('buildUpstreamEndpointRequest', () => {
|
||||
});
|
||||
|
||||
expect(request.path).toBe('/v1/messages');
|
||||
expect(request.body.system).toBe('system prompt');
|
||||
expect(request.body.system).toEqual([
|
||||
{
|
||||
type: 'text',
|
||||
text: 'system prompt',
|
||||
cache_control: { type: 'ephemeral' },
|
||||
},
|
||||
]);
|
||||
expect(request.body.messages).toEqual([
|
||||
{
|
||||
role: 'user',
|
||||
|
||||
@@ -611,6 +611,20 @@ export function shouldPreferResponsesAfterLegacyChatError(input: {
|
||||
);
|
||||
}
|
||||
|
||||
export function promoteResponsesCandidateAfterLegacyChatError(
|
||||
endpointCandidates: UpstreamEndpoint[],
|
||||
input: Parameters<typeof shouldPreferResponsesAfterLegacyChatError>[0],
|
||||
): void {
|
||||
if (!shouldPreferResponsesAfterLegacyChatError(input)) return;
|
||||
|
||||
const currentIndex = endpointCandidates.findIndex((endpoint) => endpoint === input.currentEndpoint);
|
||||
const responsesIndex = endpointCandidates.indexOf('responses');
|
||||
if (currentIndex < 0 || responsesIndex < 0 || responsesIndex <= currentIndex + 1) return;
|
||||
|
||||
endpointCandidates.splice(responsesIndex, 1);
|
||||
endpointCandidates.splice(currentIndex + 1, 0, 'responses');
|
||||
}
|
||||
|
||||
export function isEndpointDowngradeError(status: number, upstreamErrorText?: string | null): boolean {
|
||||
if (status < 400) return false;
|
||||
const text = (upstreamErrorText || '').toLowerCase();
|
||||
|
||||
Reference in New Issue
Block a user