aibo/BoosAPI
1
0
Fork 0
forked from boos/BoosAPI
Code Pull Requests Activity

fix: align explicit group route picker and details (#158)

Browse Source 
* fix: align explicit group route picker and details

* fix: move responses stream closeout into transformer facade
This commit is contained in:
Cita
2026-03-19 00:47:22 +08:00
committed by GitHub
parent 45fe687c0d
commit c0c993f448
12 changed files with 1065 additions and 199 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/api/tokens.route-update-rebuild.test.ts
+69
View File
@@ -244,6 +244,75 @@ describe('PUT /api/routes/:id route rebuild', () => {
]));
});
it('fills missing sourceModel from source exact routes when loading explicit-group channels', async () => {
const sourceA = await seedAccountWithToken('deepseek-chat');
const sourceB = await seedAccountWithToken('deepseek-reasoner');
const exactRouteA = await db.insert(schema.tokenRoutes).values({
modelPattern: 'deepseek-chat',
enabled: true,
}).returning().get();
const exactRouteB = await db.insert(schema.tokenRoutes).values({
modelPattern: 'deepseek-reasoner',
enabled: true,
}).returning().get();
await db.insert(schema.routeChannels).values([
{
routeId: exactRouteA.id,
accountId: sourceA.account.id,
tokenId: sourceA.token.id,
sourceModel: null,
priority: 0,
weight: 10,
enabled: true,
manualOverride: false,
},
{
routeId: exactRouteB.id,
accountId: sourceB.account.id,
tokenId: sourceB.token.id,
sourceModel: null,
priority: 1,
weight: 8,
enabled: true,
manualOverride: false,
},
]).run();
const createResponse = await app.inject({
method: 'POST',
url: '/api/routes',
payload: {
routeMode: 'explicit_group',
displayName: 'deepseekv1',
sourceRouteIds: [exactRouteA.id, exactRouteB.id],
},
});
expect(createResponse.statusCode).toBe(200);
const createdRouteId = (createResponse.json() as { id: number }).id;
const channelsResponse = await app.inject({
method: 'GET',
url: `/api/routes/${createdRouteId}/channels`,
});
expect(channelsResponse.statusCode).toBe(200);
expect(channelsResponse.json()).toEqual(expect.arrayContaining([
expect.objectContaining({
routeId: exactRouteA.id,
accountId: sourceA.account.id,
sourceModel: 'deepseek-chat',
}),
expect.objectContaining({
routeId: exactRouteB.id,
accountId: sourceB.account.id,
sourceModel: 'deepseek-reasoner',
}),
]));
});
it('rejects invalid explicit-group payloads', async () => {
const response = await app.inject({
method: 'POST',
src/server/routes/api/tokens.ts
+20 -5
View File
@@ -534,7 +534,7 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
const explicitSourceRouteIds = Array.from(new Set(routes
.filter((route) => isExplicitGroupRoute(route))
.flatMap((route) => route.sourceRouteIds)));
const enabledExplicitSourceRouteIds = explicitSourceRouteIds.length > 0
const explicitSourceRoutes = explicitSourceRouteIds.length > 0
? (await db.select({
id: schema.tokenRoutes.id,
modelPattern: schema.tokenRoutes.modelPattern,
@@ -543,9 +543,10 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
}).from(schema.tokenRoutes)
.where(inArray(schema.tokenRoutes.id, explicitSourceRouteIds))
.all())
.filter((route) => route.enabled && !isExplicitGroupRoute(route) && isExactModelPattern(route.modelPattern))
.map((route) => route.id)
: [];
const enabledExplicitSourceRouteIds = explicitSourceRoutes
.filter((route) => route.enabled && !isExplicitGroupRoute(route) && isExactModelPattern(route.modelPattern))
.map((route) => route.id);
const actualRouteIds = Array.from(new Set([
...routes.filter((route) => !isExplicitGroupRoute(route)).map((route) => route.id),
...enabledExplicitSourceRouteIds,
@@ -554,6 +555,14 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
return new Map(routes.map((route) => [route.id, []]));
}
const actualRouteById = new Map<number, { modelPattern: string; routeMode: string | null }>();
for (const route of routes.filter((item) => !isExplicitGroupRoute(item))) {
actualRouteById.set(route.id, { modelPattern: route.modelPattern, routeMode: route.routeMode ?? null });
}
for (const route of explicitSourceRoutes) {
actualRouteById.set(route.id, { modelPattern: route.modelPattern, routeMode: route.routeMode ?? null });
}
const channelRows = await db.select().from(schema.routeChannels)
.innerJoin(schema.accounts, eq(schema.routeChannels.accountId, schema.accounts.id))
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
@@ -565,13 +574,19 @@ async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number
for (const row of channelRows) {
const routeId = row.route_channels.routeId;
const actualRoute = actualRouteById.get(routeId);
const fallbackSourceModel = actualRoute && !isExplicitGroupRoute(actualRoute) && isExactModelPattern(actualRoute.modelPattern)
? actualRoute.modelPattern
: null;
const resolvedSourceModel = (row.route_channels.sourceModel || fallbackSourceModel || '').trim();
if (!channelsByActualRouteId.has(routeId)) channelsByActualRouteId.set(routeId, []);
channelsByActualRouteId.get(routeId)!.push({
...row.route_channels,
sourceModel: resolvedSourceModel || null,
account: row.accounts,
site: row.sites,
token: row.account_tokens
? {
token: row.account_tokens
? {
id: row.account_tokens.id,
name: row.account_tokens.name,
accountId: row.account_tokens.accountId,
src/server/routes/proxy/responses.ts
+54 -66
View File
@@ -434,6 +434,34 @@ export async function responsesProxyRoute(app: FastifyInstance) {
reply.raw.setHeader('Connection', 'keep-alive');
reply.raw.setHeader('X-Accel-Buffering', 'no');
};
let parsedUsage: UsageSummary = {
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
cacheReadTokens: 0,
cacheCreationTokens: 0,
promptTokensIncludeCache: null,
};
const writeLines = (lines: string[]) => {
for (const line of lines) reply.raw.write(line);
};
const streamSession = openAiResponsesTransformer.proxyStream.createSession({
modelName,
successfulUpstreamPath,
strictTerminalEvents: Object.entries(request.headers as Record<string, unknown>)
.some(([rawKey, rawValue]) => rawKey.trim().toLowerCase() === 'x-metapi-responses-websocket-transport'
&& String(rawValue).trim() === '1'),
getUsage: () => parsedUsage,
onParsedPayload: (payload) => {
if (payload && typeof payload === 'object') {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(payload));
}
},
writeLines,
writeRaw: (chunk) => {
reply.raw.write(chunk);
},
});
if (!upstreamContentType.includes('text/event-stream')) {
const rawText = await upstream.text();
@@ -447,7 +475,7 @@ export async function responsesProxyRoute(app: FastifyInstance) {
upstreamData = unwrapGeminiCliPayload(upstreamData);
}
const parsedUsage = parseProxyUsage(upstreamData);
parsedUsage = parseProxyUsage(upstreamData);
const latency = Date.now() - startTime;
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
if (failure) {
@@ -483,17 +511,31 @@ export async function responsesProxyRoute(app: FastifyInstance) {
return reply.code(failure.status).send({ error: { message: failure.reason, type: 'upstream_error' } });
}
const normalized = openAiResponsesTransformer.transformFinalResponse(
upstreamData,
modelName,
rawText,
);
const downstreamData = openAiResponsesTransformer.outbound.serializeFinal({
upstreamPayload: upstreamData,
normalized,
usage: parsedUsage,
serializationMode: 'response',
});
startSseResponse();
const streamResult = streamSession.consumeUpstreamFinalPayload(upstreamData, rawText, reply.raw);
if (streamResult.status === 'failed') {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
200,
latency,
streamResult.errorMessage,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return;
}
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
@@ -526,32 +568,6 @@ export async function responsesProxyRoute(app: FastifyInstance) {
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
const streamPayload = isRecord(downstreamData) ? downstreamData : {
id: `resp_${Date.now()}`,
object: 'response',
status: 'completed',
model: modelName,
output: [],
output_text: '',
usage: {
input_tokens: parsedUsage.promptTokens,
output_tokens: parsedUsage.completionTokens,
total_tokens: parsedUsage.totalTokens,
},
};
const createdPayload = {
...streamPayload,
status: 'in_progress',
output: [],
output_text: '',
};
startSseResponse();
reply.raw.write(`event: response.created\ndata: ${JSON.stringify({ type: 'response.created', response: createdPayload })}\n\n`);
reply.raw.write(`event: response.completed\ndata: ${JSON.stringify({ type: 'response.completed', response: streamPayload })}\n\n`);
reply.raw.write('data: [DONE]\n\n');
reply.raw.end();
return;
}
@@ -561,14 +577,6 @@ export async function responsesProxyRoute(app: FastifyInstance) {
const baseReader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
? createGeminiCliStreamReader(upstreamReader)
: upstreamReader;
let parsedUsage: UsageSummary = {
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
cacheReadTokens: 0,
cacheCreationTokens: 0,
promptTokensIncludeCache: null,
};
let rawText = '';
const decoder = new TextDecoder();
const reader = baseReader
@@ -588,26 +596,6 @@ export async function responsesProxyRoute(app: FastifyInstance) {
},
}
: baseReader;
const writeLines = (lines: string[]) => {
for (const line of lines) reply.raw.write(line);
};
const streamSession = openAiResponsesTransformer.proxyStream.createSession({
modelName,
successfulUpstreamPath,
strictTerminalEvents: Object.entries(request.headers as Record<string, unknown>)
.some(([rawKey, rawValue]) => rawKey.trim().toLowerCase() === 'x-metapi-responses-websocket-transport'
&& String(rawValue).trim() === '1'),
getUsage: () => parsedUsage,
onParsedPayload: (payload) => {
if (payload && typeof payload === 'object') {
parsedUsage = mergeProxyUsage(parsedUsage, parseProxyUsage(payload));
}
},
writeLines,
writeRaw: (chunk) => {
reply.raw.write(chunk);
},
});
const streamResult = await streamSession.run(reader, reply.raw);
rawText += decoder.decode();
src/server/transformers/openai/responses/proxyStream.test.ts
+72
View File
@@ -0,0 +1,72 @@
import { describe, expect, it } from 'vitest';
import { createResponsesProxyStreamSession } from './proxyStream.js';
describe('createResponsesProxyStreamSession', () => {
it('serializes non-SSE fallback payloads into canonical responses SSE closeout events', () => {
const lines: string[] = [];
let ended = false;
const usage = {
promptTokens: 5,
completionTokens: 3,
totalTokens: 8,
cacheReadTokens: 0,
cacheCreationTokens: 0,
promptTokensIncludeCache: null,
};
const payload = {
id: 'resp_fallback_1',
object: 'response',
status: 'completed',
model: 'gpt-5.2',
output_text: 'hello from responses upstream',
output: [
{
id: 'msg_fallback_1',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'hello from responses upstream' }],
},
],
usage: {
input_tokens: usage.promptTokens,
output_tokens: usage.completionTokens,
total_tokens: usage.totalTokens,
},
};
const session = createResponsesProxyStreamSession({
modelName: 'gpt-5.2',
successfulUpstreamPath: '/v1/responses',
getUsage: () => usage,
writeLines: (nextLines) => {
lines.push(...nextLines);
},
writeRaw: () => {},
});
const result = session.consumeUpstreamFinalPayload(
payload,
JSON.stringify(payload),
{
end() {
ended = true;
},
},
);
expect(result).toEqual({
status: 'completed',
errorMessage: null,
});
expect(ended).toBe(true);
const output = lines.join('');
expect(output).toContain('event: response.created');
expect(output).toContain('event: response.completed');
expect(output).toContain('"type":"response.completed"');
expect(output).toContain('"output_text":"hello from responses upstream"');
expect(output).toContain('data: [DONE]');
});
});
src/server/transformers/openai/responses/proxyStream.ts
+46
View File
@@ -1,6 +1,7 @@
import { createProxyStreamLifecycle } from '../../shared/protocolLifecycle.js';
import { type ParsedSseEvent } from '../../shared/normalized.js';
import { completeResponsesStream, createOpenAiResponsesAggregateState, failResponsesStream, serializeConvertedResponsesEvents } from './aggregator.js';
import { openAiResponsesOutbound } from './outbound.js';
import { openAiResponsesStream } from './stream.js';
type StreamReader = {
@@ -148,6 +149,51 @@ export function createResponsesProxyStreamSession(input: ResponsesProxyStreamSes
};
return {
consumeUpstreamFinalPayload(payload: unknown, fallbackText: string, response?: ResponseSink): ResponsesProxyStreamResult {
if (payload && typeof payload === 'object') {
input.onParsedPayload?.(payload);
}
const payloadType = (isRecord(payload) && typeof payload.type === 'string')
? payload.type
: '';
if (payloadType === 'error' || payloadType === 'response.failed') {
fail(payload);
response?.end();
return terminalResult;
}
const normalizedFinal = openAiResponsesOutbound.normalizeFinal(payload, input.modelName, fallbackText);
streamContext.id = normalizedFinal.id;
streamContext.model = normalizedFinal.model;
streamContext.created = normalizedFinal.created;
const streamPayload = openAiResponsesOutbound.serializeFinal({
upstreamPayload: payload,
normalized: normalizedFinal,
usage: input.getUsage(),
serializationMode: 'response',
});
const createdPayload = {
...streamPayload,
status: 'in_progress',
output: [],
output_text: '',
};
finalized = true;
terminalResult = {
status: 'completed',
errorMessage: null,
};
input.writeLines([
`event: response.created\ndata: ${JSON.stringify({ type: 'response.created', response: createdPayload })}\n\n`,
`event: response.completed\ndata: ${JSON.stringify({ type: 'response.completed', response: streamPayload })}\n\n`,
'data: [DONE]\n\n',
]);
response?.end();
return terminalResult;
},
async run(reader: StreamReader | null | undefined, response: ResponseSink): Promise<ResponsesProxyStreamResult> {
const lifecycle = createProxyStreamLifecycle<ParsedSseEvent>({
reader,