fix route grouping and codex websocket regressions
This commit is contained in:
@@ -149,4 +149,180 @@ describe('PUT /api/routes/:id route rebuild', () => {
|
||||
expect(rebuiltAuto?.priority).toBe(0);
|
||||
expect(rebuiltAuto?.weight).toBe(10);
|
||||
});
|
||||
|
||||
it('creates explicit-group routes with sourceRouteIds and aggregates source channels', async () => {
|
||||
const sourceA = await seedAccountWithToken('claude-opus-4-5');
|
||||
const sourceB = await seedAccountWithToken('claude-sonnet-4-5');
|
||||
|
||||
const exactRouteA = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-opus-4-5',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
const exactRouteB = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-sonnet-4-5',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.routeChannels).values([
|
||||
{
|
||||
routeId: exactRouteA.id,
|
||||
accountId: sourceA.account.id,
|
||||
tokenId: sourceA.token.id,
|
||||
sourceModel: 'claude-opus-4-5',
|
||||
priority: 0,
|
||||
weight: 10,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
},
|
||||
{
|
||||
routeId: exactRouteB.id,
|
||||
accountId: sourceB.account.id,
|
||||
tokenId: sourceB.token.id,
|
||||
sourceModel: 'claude-sonnet-4-5',
|
||||
priority: 1,
|
||||
weight: 8,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
},
|
||||
]).run();
|
||||
|
||||
const createResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/routes',
|
||||
payload: {
|
||||
routeMode: 'explicit_group',
|
||||
displayName: 'claude-opus-4-6',
|
||||
sourceRouteIds: [exactRouteA.id, exactRouteB.id],
|
||||
routingStrategy: 'weighted',
|
||||
},
|
||||
});
|
||||
|
||||
expect(createResponse.statusCode).toBe(200);
|
||||
expect(createResponse.json()).toMatchObject({
|
||||
displayName: 'claude-opus-4-6',
|
||||
routeMode: 'explicit_group',
|
||||
sourceRouteIds: [exactRouteA.id, exactRouteB.id],
|
||||
});
|
||||
|
||||
const createdRouteId = (createResponse.json() as { id: number }).id;
|
||||
|
||||
const storedChannels = await db.select().from(schema.routeChannels)
|
||||
.where(eq(schema.routeChannels.routeId, createdRouteId))
|
||||
.all();
|
||||
expect(storedChannels).toHaveLength(0);
|
||||
|
||||
const summaryResponse = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/routes/summary',
|
||||
});
|
||||
expect(summaryResponse.statusCode).toBe(200);
|
||||
expect(summaryResponse.json()).toContainEqual(expect.objectContaining({
|
||||
id: createdRouteId,
|
||||
routeMode: 'explicit_group',
|
||||
sourceRouteIds: [exactRouteA.id, exactRouteB.id],
|
||||
channelCount: 2,
|
||||
enabledChannelCount: 2,
|
||||
siteNames: expect.arrayContaining([sourceA.site.name, sourceB.site.name]),
|
||||
}));
|
||||
|
||||
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: 'claude-opus-4-5',
|
||||
}),
|
||||
expect.objectContaining({
|
||||
routeId: exactRouteB.id,
|
||||
accountId: sourceB.account.id,
|
||||
sourceModel: 'claude-sonnet-4-5',
|
||||
}),
|
||||
]));
|
||||
});
|
||||
|
||||
it('rejects invalid explicit-group payloads', async () => {
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/routes',
|
||||
payload: {
|
||||
routeMode: 'explicit_group',
|
||||
displayName: '',
|
||||
sourceRouteIds: [],
|
||||
},
|
||||
});
|
||||
|
||||
expect(response.statusCode).toBe(400);
|
||||
expect(response.json()).toMatchObject({
|
||||
success: false,
|
||||
});
|
||||
});
|
||||
|
||||
it('prefers an exact route over a colliding explicit-group display name', async () => {
|
||||
const exactCandidate = await seedAccountWithToken('claude-opus-4-6');
|
||||
const groupedCandidate = await seedAccountWithToken('claude-opus-4-5');
|
||||
|
||||
const exactRoute = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-opus-4-6',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
const sourceRoute = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-opus-4-5',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.routeChannels).values([
|
||||
{
|
||||
routeId: exactRoute.id,
|
||||
accountId: exactCandidate.account.id,
|
||||
tokenId: exactCandidate.token.id,
|
||||
sourceModel: 'claude-opus-4-6',
|
||||
priority: 0,
|
||||
weight: 10,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
},
|
||||
{
|
||||
routeId: sourceRoute.id,
|
||||
accountId: groupedCandidate.account.id,
|
||||
tokenId: groupedCandidate.token.id,
|
||||
sourceModel: 'claude-opus-4-5',
|
||||
priority: 0,
|
||||
weight: 10,
|
||||
enabled: true,
|
||||
manualOverride: false,
|
||||
},
|
||||
]).run();
|
||||
|
||||
const groupResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/routes',
|
||||
payload: {
|
||||
routeMode: 'explicit_group',
|
||||
displayName: 'claude-opus-4-6',
|
||||
sourceRouteIds: [sourceRoute.id],
|
||||
},
|
||||
});
|
||||
|
||||
expect(groupResponse.statusCode).toBe(200);
|
||||
|
||||
const decisionResponse = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/api/routes/decision?model=claude-opus-4-6',
|
||||
});
|
||||
|
||||
expect(decisionResponse.statusCode).toBe(200);
|
||||
expect(decisionResponse.json()).toMatchObject({
|
||||
success: true,
|
||||
decision: {
|
||||
matched: true,
|
||||
routeId: exactRoute.id,
|
||||
modelPattern: 'claude-opus-4-6',
|
||||
actualModel: 'claude-opus-4-6',
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
+430
-186
@@ -22,6 +22,138 @@ function isExactModelPattern(modelPattern: string): boolean {
|
||||
if (normalized.toLowerCase().startsWith('re:')) return false;
|
||||
return !/[\*\?]/.test(normalized);
|
||||
}
|
||||
|
||||
type RouteMode = 'pattern' | 'explicit_group';
|
||||
type RouteRow = typeof schema.tokenRoutes.$inferSelect & {
|
||||
routeMode: RouteMode;
|
||||
sourceRouteIds: number[];
|
||||
};
|
||||
|
||||
function normalizeRouteMode(routeMode: unknown): RouteMode {
|
||||
return routeMode === 'explicit_group' ? 'explicit_group' : 'pattern';
|
||||
}
|
||||
|
||||
function isExplicitGroupRoute(route: Pick<RouteRow, 'routeMode'> | Pick<typeof schema.tokenRoutes.$inferSelect, 'routeMode'>): boolean {
|
||||
return normalizeRouteMode(route.routeMode) === 'explicit_group';
|
||||
}
|
||||
|
||||
function normalizeSourceRouteIdsInput(input: unknown): number[] {
|
||||
const rawValues = Array.isArray(input) ? input : [];
|
||||
const normalized: number[] = [];
|
||||
for (const raw of rawValues) {
|
||||
const value = Number(raw);
|
||||
if (!Number.isFinite(value)) continue;
|
||||
const routeId = Math.trunc(value);
|
||||
if (routeId <= 0 || normalized.includes(routeId)) continue;
|
||||
normalized.push(routeId);
|
||||
if (normalized.length >= 500) break;
|
||||
}
|
||||
return normalized;
|
||||
}
|
||||
|
||||
async function loadRouteSourceIdsMap(routeIds: number[]): Promise<Map<number, number[]>> {
|
||||
const normalizedRouteIds = Array.from(new Set(routeIds.filter((routeId) => Number.isFinite(routeId) && routeId > 0)));
|
||||
if (normalizedRouteIds.length === 0) return new Map();
|
||||
|
||||
const rows = await db.select().from(schema.routeGroupSources)
|
||||
.where(inArray(schema.routeGroupSources.groupRouteId, normalizedRouteIds))
|
||||
.all();
|
||||
const sourceRouteIdsByRouteId = new Map<number, number[]>();
|
||||
for (const row of rows) {
|
||||
if (!sourceRouteIdsByRouteId.has(row.groupRouteId)) {
|
||||
sourceRouteIdsByRouteId.set(row.groupRouteId, []);
|
||||
}
|
||||
sourceRouteIdsByRouteId.get(row.groupRouteId)!.push(row.sourceRouteId);
|
||||
}
|
||||
for (const [routeId, sourceRouteIds] of sourceRouteIdsByRouteId.entries()) {
|
||||
sourceRouteIdsByRouteId.set(routeId, Array.from(new Set(sourceRouteIds)));
|
||||
}
|
||||
return sourceRouteIdsByRouteId;
|
||||
}
|
||||
|
||||
function decorateRoutesWithSources(
|
||||
routes: Array<typeof schema.tokenRoutes.$inferSelect>,
|
||||
sourceRouteIdsByRouteId: Map<number, number[]>,
|
||||
): RouteRow[] {
|
||||
return routes.map((route) => ({
|
||||
...route,
|
||||
routeMode: normalizeRouteMode(route.routeMode),
|
||||
sourceRouteIds: sourceRouteIdsByRouteId.get(route.id) ?? [],
|
||||
}));
|
||||
}
|
||||
|
||||
async function listRoutesWithSources(): Promise<RouteRow[]> {
|
||||
const routes = await db.select().from(schema.tokenRoutes).all();
|
||||
const sourceRouteIdsByRouteId = await loadRouteSourceIdsMap(routes.map((route) => route.id));
|
||||
return decorateRoutesWithSources(routes, sourceRouteIdsByRouteId);
|
||||
}
|
||||
|
||||
async function getRouteWithSources(routeId: number): Promise<RouteRow | null> {
|
||||
const route = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, routeId)).get();
|
||||
if (!route) return null;
|
||||
const sourceRouteIdsByRouteId = await loadRouteSourceIdsMap([routeId]);
|
||||
return decorateRoutesWithSources([route], sourceRouteIdsByRouteId)[0] ?? null;
|
||||
}
|
||||
|
||||
async function validateExplicitGroupSourceRoutes(sourceRouteIds: number[], currentRouteId?: number): Promise<{ ok: true } | { ok: false; message: string }> {
|
||||
if (sourceRouteIds.length === 0) {
|
||||
return { ok: false, message: '显式群组至少需要选择一个来源模型' };
|
||||
}
|
||||
|
||||
const routes = await db.select().from(schema.tokenRoutes)
|
||||
.where(inArray(schema.tokenRoutes.id, sourceRouteIds))
|
||||
.all();
|
||||
if (routes.length !== sourceRouteIds.length) {
|
||||
return { ok: false, message: '来源模型中存在不存在的路由' };
|
||||
}
|
||||
|
||||
for (const route of routes) {
|
||||
if (currentRouteId && route.id === currentRouteId) {
|
||||
return { ok: false, message: '显式群组不能引用自身作为来源模型' };
|
||||
}
|
||||
if (normalizeRouteMode(route.routeMode) === 'explicit_group') {
|
||||
return { ok: false, message: '显式群组只能选择精确模型路由作为来源模型' };
|
||||
}
|
||||
if (!isExactModelPattern(route.modelPattern)) {
|
||||
return { ok: false, message: '显式群组只能选择精确模型路由作为来源模型' };
|
||||
}
|
||||
}
|
||||
|
||||
return { ok: true };
|
||||
}
|
||||
|
||||
async function replaceRouteSourceRouteIds(routeId: number, sourceRouteIds: number[]): Promise<void> {
|
||||
await db.delete(schema.routeGroupSources).where(eq(schema.routeGroupSources.groupRouteId, routeId)).run();
|
||||
if (sourceRouteIds.length === 0) return;
|
||||
await db.insert(schema.routeGroupSources).values(
|
||||
sourceRouteIds.map((sourceRouteId) => ({
|
||||
groupRouteId: routeId,
|
||||
sourceRouteId,
|
||||
})),
|
||||
).run();
|
||||
}
|
||||
|
||||
async function clearDependentExplicitGroupSnapshotsBySourceRouteIds(sourceRouteIds: number[]): Promise<void> {
|
||||
const normalizedSourceRouteIds = Array.from(new Set(
|
||||
sourceRouteIds.filter((routeId): routeId is number => Number.isFinite(routeId) && routeId > 0),
|
||||
));
|
||||
if (normalizedSourceRouteIds.length === 0) return;
|
||||
|
||||
const rows = await db.select({ groupRouteId: schema.routeGroupSources.groupRouteId })
|
||||
.from(schema.routeGroupSources)
|
||||
.where(inArray(schema.routeGroupSources.sourceRouteId, normalizedSourceRouteIds))
|
||||
.all();
|
||||
const dependentRouteIdSet = new Set<number>();
|
||||
for (const row of rows) {
|
||||
const routeId = Number(row.groupRouteId);
|
||||
if (Number.isFinite(routeId) && routeId > 0) {
|
||||
dependentRouteIdSet.add(routeId);
|
||||
}
|
||||
}
|
||||
const dependentRouteIds = Array.from(dependentRouteIdSet);
|
||||
if (dependentRouteIds.length === 0) return;
|
||||
await clearRouteDecisionSnapshots(dependentRouteIds);
|
||||
}
|
||||
|
||||
async function getDefaultTokenId(accountId: number): Promise<number | null> {
|
||||
const token = await db.select().from(schema.accountTokens)
|
||||
@@ -390,25 +522,54 @@ function parseBatchRouteWideDecisionRouteIds(
|
||||
};
|
||||
}
|
||||
|
||||
async function fetchChannelsForRoutes(routeIds: number[]): Promise<Map<number, any[]>> {
|
||||
if (routeIds.length === 0) return new Map();
|
||||
|
||||
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))
|
||||
.leftJoin(schema.accountTokens, eq(schema.routeChannels.tokenId, schema.accountTokens.id))
|
||||
.where(inArray(schema.routeChannels.routeId, routeIds))
|
||||
.all();
|
||||
|
||||
const channelsByRoute = new Map<number, any[]>();
|
||||
|
||||
for (const row of channelRows) {
|
||||
const routeId = row.route_channels.routeId;
|
||||
if (!channelsByRoute.has(routeId)) channelsByRoute.set(routeId, []);
|
||||
channelsByRoute.get(routeId)!.push({
|
||||
...row.route_channels,
|
||||
account: row.accounts,
|
||||
site: row.sites,
|
||||
type RouteChannelSummary = {
|
||||
channelCount: number;
|
||||
enabledChannelCount: number;
|
||||
siteNames: Set<string>;
|
||||
};
|
||||
|
||||
async function fetchChannelsForRouteRows(routes: RouteRow[]): Promise<Map<number, any[]>> {
|
||||
if (routes.length === 0) return new Map();
|
||||
|
||||
const explicitSourceRouteIds = Array.from(new Set(routes
|
||||
.filter((route) => isExplicitGroupRoute(route))
|
||||
.flatMap((route) => route.sourceRouteIds)));
|
||||
const enabledExplicitSourceRouteIds = explicitSourceRouteIds.length > 0
|
||||
? (await db.select({
|
||||
id: schema.tokenRoutes.id,
|
||||
modelPattern: schema.tokenRoutes.modelPattern,
|
||||
routeMode: schema.tokenRoutes.routeMode,
|
||||
enabled: schema.tokenRoutes.enabled,
|
||||
}).from(schema.tokenRoutes)
|
||||
.where(inArray(schema.tokenRoutes.id, explicitSourceRouteIds))
|
||||
.all())
|
||||
.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,
|
||||
]));
|
||||
if (actualRouteIds.length === 0) {
|
||||
return new Map(routes.map((route) => [route.id, []]));
|
||||
}
|
||||
|
||||
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))
|
||||
.leftJoin(schema.accountTokens, eq(schema.routeChannels.tokenId, schema.accountTokens.id))
|
||||
.where(inArray(schema.routeChannels.routeId, actualRouteIds))
|
||||
.all();
|
||||
|
||||
const channelsByActualRouteId = new Map<number, any[]>();
|
||||
|
||||
for (const row of channelRows) {
|
||||
const routeId = row.route_channels.routeId;
|
||||
if (!channelsByActualRouteId.has(routeId)) channelsByActualRouteId.set(routeId, []);
|
||||
channelsByActualRouteId.get(routeId)!.push({
|
||||
...row.route_channels,
|
||||
account: row.accounts,
|
||||
site: row.sites,
|
||||
token: row.account_tokens
|
||||
? {
|
||||
id: row.account_tokens.id,
|
||||
@@ -417,63 +578,83 @@ async function fetchChannelsForRoutes(routeIds: number[]): Promise<Map<number, a
|
||||
enabled: row.account_tokens.enabled,
|
||||
isDefault: row.account_tokens.isDefault,
|
||||
}
|
||||
: null,
|
||||
});
|
||||
}
|
||||
|
||||
return channelsByRoute;
|
||||
}
|
||||
|
||||
export async function tokensRoutes(app: FastifyInstance) {
|
||||
// List routes with basic info only (lightweight for selectors)
|
||||
app.get('/api/routes/lite', async () => {
|
||||
return await db.select({
|
||||
id: schema.tokenRoutes.id,
|
||||
modelPattern: schema.tokenRoutes.modelPattern,
|
||||
displayName: schema.tokenRoutes.displayName,
|
||||
displayIcon: schema.tokenRoutes.displayIcon,
|
||||
routingStrategy: schema.tokenRoutes.routingStrategy,
|
||||
enabled: schema.tokenRoutes.enabled,
|
||||
}).from(schema.tokenRoutes).all();
|
||||
: null,
|
||||
});
|
||||
}
|
||||
|
||||
const channelsByRoute = new Map<number, any[]>();
|
||||
for (const route of routes) {
|
||||
if (isExplicitGroupRoute(route)) {
|
||||
channelsByRoute.set(route.id, route.sourceRouteIds.flatMap((sourceRouteId) => channelsByActualRouteId.get(sourceRouteId) || []));
|
||||
continue;
|
||||
}
|
||||
channelsByRoute.set(route.id, channelsByActualRouteId.get(route.id) || []);
|
||||
}
|
||||
|
||||
return channelsByRoute;
|
||||
}
|
||||
|
||||
async function fetchChannelsForRoutes(routeIds: number[]): Promise<Map<number, any[]>> {
|
||||
if (routeIds.length === 0) return new Map();
|
||||
return await fetchChannelsForRouteRows(await listRoutesWithSources()).then((channelsByRoute) => {
|
||||
const filtered = new Map<number, any[]>();
|
||||
for (const routeId of routeIds) {
|
||||
filtered.set(routeId, channelsByRoute.get(routeId) || []);
|
||||
}
|
||||
return filtered;
|
||||
});
|
||||
}
|
||||
|
||||
async function buildRouteChannelSummaryMap(routes: RouteRow[]): Promise<Map<number, RouteChannelSummary>> {
|
||||
const channelsByRoute = await fetchChannelsForRouteRows(routes);
|
||||
const summaryByRoute = new Map<number, RouteChannelSummary>();
|
||||
for (const route of routes) {
|
||||
const channels = channelsByRoute.get(route.id) || [];
|
||||
const siteNames = new Set<string>();
|
||||
let enabledChannelCount = 0;
|
||||
for (const channel of channels) {
|
||||
if (channel.enabled) enabledChannelCount += 1;
|
||||
if (channel.site?.name) siteNames.add(channel.site.name);
|
||||
}
|
||||
summaryByRoute.set(route.id, {
|
||||
channelCount: channels.length,
|
||||
enabledChannelCount,
|
||||
siteNames,
|
||||
});
|
||||
}
|
||||
return summaryByRoute;
|
||||
}
|
||||
|
||||
// Route summary (no channel details) for first-screen rendering
|
||||
app.get('/api/routes/summary', async () => {
|
||||
const routes = await db.select().from(schema.tokenRoutes).all();
|
||||
if (routes.length === 0) return [];
|
||||
|
||||
const routeIds = routes.map((route) => route.id);
|
||||
|
||||
// Aggregate channel counts and site names per route
|
||||
const channelRows = await db.select({
|
||||
routeId: schema.routeChannels.routeId,
|
||||
enabled: schema.routeChannels.enabled,
|
||||
siteName: schema.sites.name,
|
||||
}).from(schema.routeChannels)
|
||||
.innerJoin(schema.accounts, eq(schema.routeChannels.accountId, schema.accounts.id))
|
||||
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
|
||||
.where(inArray(schema.routeChannels.routeId, routeIds))
|
||||
.all();
|
||||
|
||||
const aggByRoute = new Map<number, { channelCount: number; enabledChannelCount: number; siteNames: Set<string> }>();
|
||||
for (const row of channelRows) {
|
||||
let agg = aggByRoute.get(row.routeId);
|
||||
if (!agg) {
|
||||
agg = { channelCount: 0, enabledChannelCount: 0, siteNames: new Set() };
|
||||
aggByRoute.set(row.routeId, agg);
|
||||
}
|
||||
agg.channelCount += 1;
|
||||
if (row.enabled) agg.enabledChannelCount += 1;
|
||||
if (row.siteName) agg.siteNames.add(row.siteName);
|
||||
}
|
||||
|
||||
return routes.map((route) => {
|
||||
const agg = aggByRoute.get(route.id);
|
||||
return {
|
||||
id: route.id,
|
||||
modelPattern: route.modelPattern,
|
||||
displayName: route.displayName ?? null,
|
||||
export async function tokensRoutes(app: FastifyInstance) {
|
||||
// List routes with basic info only (lightweight for selectors)
|
||||
app.get('/api/routes/lite', async () => {
|
||||
return (await listRoutesWithSources()).map((route) => ({
|
||||
id: route.id,
|
||||
modelPattern: route.modelPattern,
|
||||
displayName: route.displayName,
|
||||
displayIcon: route.displayIcon,
|
||||
routeMode: route.routeMode,
|
||||
sourceRouteIds: route.sourceRouteIds,
|
||||
routingStrategy: route.routingStrategy,
|
||||
enabled: route.enabled,
|
||||
}));
|
||||
});
|
||||
|
||||
// Route summary (no channel details) for first-screen rendering
|
||||
app.get('/api/routes/summary', async () => {
|
||||
const routes = await listRoutesWithSources();
|
||||
if (routes.length === 0) return [];
|
||||
const aggByRoute = await buildRouteChannelSummaryMap(routes);
|
||||
|
||||
return routes.map((route) => {
|
||||
const agg = aggByRoute.get(route.id);
|
||||
return {
|
||||
id: route.id,
|
||||
modelPattern: route.modelPattern,
|
||||
displayName: route.displayName ?? null,
|
||||
displayIcon: route.displayIcon ?? null,
|
||||
routeMode: route.routeMode,
|
||||
sourceRouteIds: route.sourceRouteIds,
|
||||
modelMapping: route.modelMapping ?? null,
|
||||
routingStrategy: route.routingStrategy ?? 'weighted',
|
||||
enabled: route.enabled,
|
||||
@@ -486,26 +667,29 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
});
|
||||
});
|
||||
|
||||
// Get channels for a single route (on-demand loading)
|
||||
app.get<{ Params: { id: string } }>('/api/routes/:id/channels', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const route = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, routeId)).get();
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
const channelsByRoute = await fetchChannelsForRoutes([routeId]);
|
||||
return channelsByRoute.get(routeId) || [];
|
||||
});
|
||||
// Get channels for a single route (on-demand loading)
|
||||
app.get<{ Params: { id: string } }>('/api/routes/:id/channels', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const route = await getRouteWithSources(routeId);
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
const channelsByRoute = await fetchChannelsForRouteRows([route]);
|
||||
return channelsByRoute.get(routeId) || [];
|
||||
});
|
||||
|
||||
// Batch add channels to a route
|
||||
app.post<{ Params: { id: string }; Body: { channels: Array<{ accountId: number; tokenId?: number; sourceModel?: string }> } }>('/api/routes/:id/channels/batch', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const body = request.body;
|
||||
|
||||
const route = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, routeId)).get();
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
app.post<{ Params: { id: string }; Body: { channels: Array<{ accountId: number; tokenId?: number; sourceModel?: string }> } }>('/api/routes/:id/channels/batch', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const body = request.body;
|
||||
|
||||
const route = await getRouteWithSources(routeId);
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
if (isExplicitGroupRoute(route)) {
|
||||
return reply.code(400).send({ success: false, message: '显式群组不支持直接维护通道' });
|
||||
}
|
||||
|
||||
if (!body?.channels || !Array.isArray(body.channels) || body.channels.length === 0) {
|
||||
return reply.code(400).send({ success: false, message: 'channels 必须是非空数组' });
|
||||
@@ -566,24 +750,24 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
}
|
||||
}
|
||||
|
||||
if (created > 0) {
|
||||
await clearRouteDecisionSnapshot(routeId);
|
||||
invalidateTokenRouterCache();
|
||||
}
|
||||
if (created > 0) {
|
||||
await clearRouteDecisionSnapshot(routeId);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([routeId]);
|
||||
invalidateTokenRouterCache();
|
||||
}
|
||||
|
||||
return { success: true, created, skipped, errors };
|
||||
});
|
||||
|
||||
// List all routes
|
||||
app.get('/api/routes', async () => {
|
||||
const routes = await db.select().from(schema.tokenRoutes).all();
|
||||
if (routes.length === 0) return [];
|
||||
|
||||
const routeIds = routes.map((route) => route.id);
|
||||
const channelsByRoute = await fetchChannelsForRoutes(routeIds);
|
||||
|
||||
return routes.map((route) => ({
|
||||
...route,
|
||||
// List all routes
|
||||
app.get('/api/routes', async () => {
|
||||
const routes = await listRoutesWithSources();
|
||||
if (routes.length === 0) return [];
|
||||
|
||||
const channelsByRoute = await fetchChannelsForRouteRows(routes);
|
||||
|
||||
return routes.map((route) => ({
|
||||
...route,
|
||||
decisionSnapshot: parseRouteDecisionSnapshot(route.decisionSnapshot),
|
||||
decisionRefreshedAt: route.decisionRefreshedAt ?? null,
|
||||
channels: channelsByRoute.get(route.id) || [],
|
||||
@@ -689,87 +873,143 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
return { success: true, decisions };
|
||||
});
|
||||
|
||||
// Create a route
|
||||
app.post<{ Body: { modelPattern: string; displayName?: string; displayIcon?: string; modelMapping?: string; routingStrategy?: string; enabled?: boolean } }>('/api/routes', async (request) => {
|
||||
// Create a route
|
||||
app.post<{ Body: { routeMode?: string; modelPattern?: string; displayName?: string; displayIcon?: string; modelMapping?: string; routingStrategy?: string; enabled?: boolean; sourceRouteIds?: number[] } }>('/api/routes', async (request, reply) => {
|
||||
const body = request.body;
|
||||
const routeMode = normalizeRouteMode(body.routeMode);
|
||||
const displayName = typeof body.displayName === 'string' ? body.displayName.trim() : '';
|
||||
const sourceRouteIds = normalizeSourceRouteIdsInput(body.sourceRouteIds);
|
||||
const modelPattern = routeMode === 'explicit_group'
|
||||
? displayName
|
||||
: (typeof body.modelPattern === 'string' ? body.modelPattern.trim() : '');
|
||||
|
||||
if (routeMode === 'explicit_group') {
|
||||
if (!displayName) {
|
||||
return reply.code(400).send({ success: false, message: '显式群组必须填写对外模型名' });
|
||||
}
|
||||
const validation = await validateExplicitGroupSourceRoutes(sourceRouteIds);
|
||||
if (!validation.ok) {
|
||||
return reply.code(400).send({ success: false, message: validation.message });
|
||||
}
|
||||
} else if (!modelPattern) {
|
||||
return reply.code(400).send({ success: false, message: '模型匹配不能为空' });
|
||||
}
|
||||
|
||||
const insertedRoute = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: body.modelPattern,
|
||||
displayName: body.displayName,
|
||||
modelPattern,
|
||||
displayName: displayName || body.displayName,
|
||||
displayIcon: body.displayIcon,
|
||||
routeMode,
|
||||
modelMapping: body.modelMapping,
|
||||
routingStrategy: normalizeRouteRoutingStrategy(body.routingStrategy),
|
||||
enabled: body.enabled ?? true,
|
||||
}).run();
|
||||
const routeId = Number(insertedRoute.lastInsertRowid || 0);
|
||||
if (routeId <= 0) {
|
||||
return { success: false, message: '创建路由失败' };
|
||||
}
|
||||
const route = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, routeId)).get();
|
||||
if (!route) {
|
||||
return { success: false, message: '创建路由失败' };
|
||||
}
|
||||
|
||||
await populateRouteChannelsByModelPattern(route.id, body.modelPattern);
|
||||
invalidateTokenRouterCache();
|
||||
return route;
|
||||
});
|
||||
|
||||
// Update a route
|
||||
app.put<{ Params: { id: string }; Body: any }>('/api/routes/:id', async (request, reply) => {
|
||||
const id = parseInt(request.params.id, 10);
|
||||
const body = request.body as Record<string, unknown>;
|
||||
const existingRoute = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, id)).get();
|
||||
if (!existingRoute) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
|
||||
const updates: Record<string, unknown> = {};
|
||||
let nextModelPattern = existingRoute.modelPattern;
|
||||
|
||||
if (body.displayName !== undefined) updates.displayName = body.displayName;
|
||||
if (body.displayIcon !== undefined) updates.displayIcon = body.displayIcon;
|
||||
if (body.modelPattern !== undefined) {
|
||||
const routeId = Number(insertedRoute.lastInsertRowid || 0);
|
||||
if (routeId <= 0) {
|
||||
return { success: false, message: '创建路由失败' };
|
||||
}
|
||||
const route = await getRouteWithSources(routeId);
|
||||
if (!route) {
|
||||
return { success: false, message: '创建路由失败' };
|
||||
}
|
||||
|
||||
if (routeMode === 'explicit_group') {
|
||||
await replaceRouteSourceRouteIds(route.id, sourceRouteIds);
|
||||
} else {
|
||||
await populateRouteChannelsByModelPattern(route.id, modelPattern);
|
||||
}
|
||||
invalidateTokenRouterCache();
|
||||
return await getRouteWithSources(routeId);
|
||||
});
|
||||
|
||||
// Update a route
|
||||
app.put<{ Params: { id: string }; Body: any }>('/api/routes/:id', async (request, reply) => {
|
||||
const id = parseInt(request.params.id, 10);
|
||||
const body = request.body as Record<string, unknown>;
|
||||
const existingRoute = await getRouteWithSources(id);
|
||||
if (!existingRoute) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
const routeMode = normalizeRouteMode(body.routeMode ?? existingRoute.routeMode);
|
||||
if (routeMode !== existingRoute.routeMode) {
|
||||
return reply.code(400).send({ success: false, message: '暂不支持在不同群组模式之间直接切换' });
|
||||
}
|
||||
|
||||
const updates: Record<string, unknown> = {};
|
||||
let nextModelPattern = existingRoute.modelPattern;
|
||||
let nextDisplayName = existingRoute.displayName ?? '';
|
||||
let nextSourceRouteIds = existingRoute.sourceRouteIds;
|
||||
|
||||
if (body.displayName !== undefined) {
|
||||
nextDisplayName = String(body.displayName || '').trim();
|
||||
updates.displayName = nextDisplayName || null;
|
||||
}
|
||||
if (body.displayIcon !== undefined) updates.displayIcon = body.displayIcon;
|
||||
if (routeMode === 'explicit_group') {
|
||||
nextModelPattern = nextDisplayName;
|
||||
updates.modelPattern = nextModelPattern;
|
||||
if (body.sourceRouteIds !== undefined) {
|
||||
nextSourceRouteIds = normalizeSourceRouteIdsInput(body.sourceRouteIds);
|
||||
}
|
||||
if (!nextDisplayName) {
|
||||
return reply.code(400).send({ success: false, message: '显式群组必须填写对外模型名' });
|
||||
}
|
||||
const validation = await validateExplicitGroupSourceRoutes(nextSourceRouteIds, id);
|
||||
if (!validation.ok) {
|
||||
return reply.code(400).send({ success: false, message: validation.message });
|
||||
}
|
||||
} else if (body.modelPattern !== undefined) {
|
||||
nextModelPattern = String(body.modelPattern);
|
||||
updates.modelPattern = nextModelPattern;
|
||||
}
|
||||
if (body.modelMapping !== undefined) updates.modelMapping = body.modelMapping;
|
||||
if (body.routingStrategy !== undefined) updates.routingStrategy = normalizeRouteRoutingStrategy(body.routingStrategy);
|
||||
if (body.enabled !== undefined) updates.enabled = body.enabled;
|
||||
if (body.routeMode !== undefined) updates.routeMode = routeMode;
|
||||
updates.updatedAt = new Date().toISOString();
|
||||
|
||||
await db.update(schema.tokenRoutes).set(updates).where(eq(schema.tokenRoutes.id, id)).run();
|
||||
const modelPatternChanged = body.modelPattern !== undefined && nextModelPattern !== existingRoute.modelPattern;
|
||||
if (routeMode === 'explicit_group' && body.sourceRouteIds !== undefined) {
|
||||
await replaceRouteSourceRouteIds(id, nextSourceRouteIds);
|
||||
}
|
||||
const modelPatternChanged = nextModelPattern !== existingRoute.modelPattern;
|
||||
const routeBehaviorChanged = modelPatternChanged
|
||||
|| (routeMode === 'explicit_group' && body.sourceRouteIds !== undefined)
|
||||
|| body.modelMapping !== undefined
|
||||
|| body.routingStrategy !== undefined
|
||||
|| body.enabled !== undefined;
|
||||
if (modelPatternChanged) {
|
||||
await rebuildAutomaticRouteChannelsByModelPattern(id, nextModelPattern);
|
||||
}
|
||||
if (routeBehaviorChanged) {
|
||||
await clearRouteDecisionSnapshot(id);
|
||||
}
|
||||
invalidateTokenRouterCache();
|
||||
return await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, id)).get();
|
||||
});
|
||||
if (routeMode === 'pattern' && modelPatternChanged) {
|
||||
await rebuildAutomaticRouteChannelsByModelPattern(id, nextModelPattern);
|
||||
}
|
||||
if (routeBehaviorChanged) {
|
||||
await clearRouteDecisionSnapshot(id);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([id]);
|
||||
}
|
||||
invalidateTokenRouterCache();
|
||||
return await getRouteWithSources(id);
|
||||
});
|
||||
|
||||
// Delete a route
|
||||
app.delete<{ Params: { id: string } }>('/api/routes/:id', async (request) => {
|
||||
const id = parseInt(request.params.id, 10);
|
||||
await db.delete(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, id)).run();
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true };
|
||||
});
|
||||
app.delete<{ Params: { id: string } }>('/api/routes/:id', async (request) => {
|
||||
const id = parseInt(request.params.id, 10);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([id]);
|
||||
await db.delete(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, id)).run();
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true };
|
||||
});
|
||||
|
||||
// Add a channel to a route
|
||||
app.post<{ Params: { id: string }; Body: { accountId: number; tokenId?: number; sourceModel?: string; priority?: number; weight?: number } }>('/api/routes/:id/channels', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const body = request.body;
|
||||
|
||||
const route = await db.select().from(schema.tokenRoutes).where(eq(schema.tokenRoutes.id, routeId)).get();
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
app.post<{ Params: { id: string }; Body: { accountId: number; tokenId?: number; sourceModel?: string; priority?: number; weight?: number } }>('/api/routes/:id/channels', async (request, reply) => {
|
||||
const routeId = parseInt(request.params.id, 10);
|
||||
const body = request.body;
|
||||
|
||||
const route = await getRouteWithSources(routeId);
|
||||
if (!route) {
|
||||
return reply.code(404).send({ success: false, message: '路由不存在' });
|
||||
}
|
||||
if (isExplicitGroupRoute(route)) {
|
||||
return reply.code(400).send({ success: false, message: '显式群组不支持直接维护通道' });
|
||||
}
|
||||
|
||||
const sourceModel = typeof body.sourceModel === 'string'
|
||||
? body.sourceModel.trim()
|
||||
@@ -811,11 +1051,12 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
const created = await db.select().from(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).get();
|
||||
if (!created) {
|
||||
return reply.code(500).send({ success: false, message: '创建通道失败' });
|
||||
}
|
||||
await clearRouteDecisionSnapshot(routeId);
|
||||
invalidateTokenRouterCache();
|
||||
return created;
|
||||
});
|
||||
}
|
||||
await clearRouteDecisionSnapshot(routeId);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([routeId]);
|
||||
invalidateTokenRouterCache();
|
||||
return created;
|
||||
});
|
||||
|
||||
// Batch update channel priorities
|
||||
app.put<{ Body: { updates: Array<{ id: number; priority: number }> } }>('/api/channels/batch', async (request, reply) => {
|
||||
@@ -841,13 +1082,14 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
}).where(eq(schema.routeChannels.id, update.id)).run();
|
||||
}
|
||||
|
||||
const updatedChannels = await db.select().from(schema.routeChannels)
|
||||
.where(inArray(schema.routeChannels.id, channelIds))
|
||||
.all();
|
||||
await clearRouteDecisionSnapshots(existingChannels.map((channel) => channel.routeId));
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true, channels: updatedChannels };
|
||||
});
|
||||
const updatedChannels = await db.select().from(schema.routeChannels)
|
||||
.where(inArray(schema.routeChannels.id, channelIds))
|
||||
.all();
|
||||
await clearRouteDecisionSnapshots(existingChannels.map((channel) => channel.routeId));
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds(existingChannels.map((channel) => channel.routeId));
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true, channels: updatedChannels };
|
||||
});
|
||||
|
||||
// Update a channel
|
||||
app.put<{ Params: { channelId: string }; Body: any }>('/api/channels/:channelId', async (request, reply) => {
|
||||
@@ -889,23 +1131,25 @@ export async function tokensRoutes(app: FastifyInstance) {
|
||||
if (body[key] !== undefined) updates[key] = body[key];
|
||||
}
|
||||
|
||||
await db.update(schema.routeChannels).set(updates).where(eq(schema.routeChannels.id, channelId)).run();
|
||||
await clearRouteDecisionSnapshot(channel.routeId);
|
||||
invalidateTokenRouterCache();
|
||||
return await db.select().from(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).get();
|
||||
});
|
||||
await db.update(schema.routeChannels).set(updates).where(eq(schema.routeChannels.id, channelId)).run();
|
||||
await clearRouteDecisionSnapshot(channel.routeId);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([channel.routeId]);
|
||||
invalidateTokenRouterCache();
|
||||
return await db.select().from(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).get();
|
||||
});
|
||||
|
||||
// Delete a channel
|
||||
app.delete<{ Params: { channelId: string } }>('/api/channels/:channelId', async (request) => {
|
||||
const channelId = parseInt(request.params.channelId, 10);
|
||||
const channel = await db.select().from(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).get();
|
||||
await db.delete(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).run();
|
||||
if (channel) {
|
||||
await clearRouteDecisionSnapshot(channel.routeId);
|
||||
}
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true };
|
||||
});
|
||||
await db.delete(schema.routeChannels).where(eq(schema.routeChannels.id, channelId)).run();
|
||||
if (channel) {
|
||||
await clearRouteDecisionSnapshot(channel.routeId);
|
||||
await clearDependentExplicitGroupSnapshotsBySourceRouteIds([channel.routeId]);
|
||||
}
|
||||
invalidateTokenRouterCache();
|
||||
return { success: true };
|
||||
});
|
||||
|
||||
// Rebuild routes/channels from model availability.
|
||||
app.post<{ Body?: { refreshModels?: boolean; wait?: boolean } }>('/api/routes/rebuild', async (request, reply) => {
|
||||
|
||||
@@ -1384,6 +1384,15 @@ describe('chat proxy stream behavior', () => {
|
||||
status: 400,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
error: {
|
||||
message: 'request validation failed',
|
||||
type: 'invalid_request_error',
|
||||
},
|
||||
}), {
|
||||
status: 400,
|
||||
headers: { 'content-type': 'application/json' },
|
||||
}))
|
||||
.mockResolvedValueOnce(new Response(JSON.stringify({
|
||||
id: 'resp_retry_minimal_headers',
|
||||
object: 'response',
|
||||
@@ -1413,21 +1422,27 @@ describe('chat proxy stream behavior', () => {
|
||||
const body = response.json();
|
||||
expect(body.output_text).toContain('ok after minimal headers retry');
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(3);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(4);
|
||||
const [, firstOptions] = fetchMock.mock.calls[0] as [string, any];
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, any];
|
||||
const [, thirdOptions] = fetchMock.mock.calls[2] as [string, any];
|
||||
const [, fourthOptions] = fetchMock.mock.calls[3] as [string, any];
|
||||
|
||||
expect(firstOptions.headers['openai-beta']).toBe('responses-2025-03-11');
|
||||
expect(secondOptions.headers['openai-beta']).toBe('responses-2025-03-11');
|
||||
expect(thirdOptions.headers['openai-beta']).toBeUndefined();
|
||||
expect(thirdOptions.headers['openai-beta']).toBe('responses-2025-03-11');
|
||||
expect(fourthOptions.headers['openai-beta']).toBeUndefined();
|
||||
|
||||
const firstBody = JSON.parse(firstOptions.body);
|
||||
const secondBody = JSON.parse(secondOptions.body);
|
||||
const thirdBody = JSON.parse(thirdOptions.body);
|
||||
const fourthBody = JSON.parse(fourthOptions.body);
|
||||
expect(firstBody.user).toBe('user-123');
|
||||
expect(secondBody.user).toBeUndefined();
|
||||
expect(thirdBody.user).toBeUndefined();
|
||||
expect(secondBody.include).toEqual(['reasoning.encrypted_content']);
|
||||
expect(thirdBody.include).toBeUndefined();
|
||||
expect(fourthBody.user).toBeUndefined();
|
||||
});
|
||||
|
||||
it('returns concise Cloudflare host error on /v1/responses 502 html failures', async () => {
|
||||
|
||||
@@ -7,11 +7,13 @@ import { join } from 'node:path';
|
||||
type DbModule = typeof import('../../db/index.js');
|
||||
type ProxyRouterModule = typeof import('./router.js');
|
||||
type TokenRouterModule = typeof import('../../services/tokenRouter.js');
|
||||
type TokensRoutesModule = typeof import('../api/tokens.js');
|
||||
|
||||
describe('/v1/models route', () => {
|
||||
let db: DbModule['db'];
|
||||
let schema: DbModule['schema'];
|
||||
let proxyRoutes: ProxyRouterModule['proxyRoutes'];
|
||||
let tokensRoutes: TokensRoutesModule['tokensRoutes'];
|
||||
let invalidateTokenRouterCache: TokenRouterModule['invalidateTokenRouterCache'];
|
||||
let app: FastifyInstance;
|
||||
let dataDir = '';
|
||||
@@ -24,13 +26,16 @@ describe('/v1/models route', () => {
|
||||
const dbModule = await import('../../db/index.js');
|
||||
const proxyRouterModule = await import('./router.js');
|
||||
const tokenRouterModule = await import('../../services/tokenRouter.js');
|
||||
const tokensRoutesModule = await import('../api/tokens.js');
|
||||
|
||||
db = dbModule.db;
|
||||
schema = dbModule.schema;
|
||||
proxyRoutes = proxyRouterModule.proxyRoutes;
|
||||
tokensRoutes = tokensRoutesModule.tokensRoutes;
|
||||
invalidateTokenRouterCache = tokenRouterModule.invalidateTokenRouterCache;
|
||||
|
||||
app = Fastify();
|
||||
await app.register(tokensRoutes);
|
||||
await app.register(proxyRoutes);
|
||||
});
|
||||
|
||||
@@ -285,6 +290,106 @@ describe('/v1/models route', () => {
|
||||
expect(ids).not.toContain('claude-sonnet-4-5');
|
||||
});
|
||||
|
||||
it('returns only explicit-group public name while hiding source exact routes', async () => {
|
||||
const site = await db.insert(schema.sites).values({
|
||||
name: 'explicit-group-site',
|
||||
url: 'https://explicit-group.example.com',
|
||||
platform: 'openai',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
|
||||
const account = await db.insert(schema.accounts).values({
|
||||
siteId: site.id,
|
||||
accessToken: 'explicit-group-access-token',
|
||||
status: 'active',
|
||||
}).returning().get();
|
||||
|
||||
const token = await db.insert(schema.accountTokens).values({
|
||||
accountId: account.id,
|
||||
name: 'default',
|
||||
token: 'explicit-group-api-token',
|
||||
enabled: true,
|
||||
isDefault: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.modelAvailability).values([
|
||||
{
|
||||
accountId: account.id,
|
||||
modelName: 'claude-opus-4-5',
|
||||
available: true,
|
||||
},
|
||||
{
|
||||
accountId: account.id,
|
||||
modelName: 'claude-sonnet-4-5',
|
||||
available: true,
|
||||
},
|
||||
]).run();
|
||||
|
||||
const sourceRouteA = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-opus-4-5',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
const sourceRouteB = await db.insert(schema.tokenRoutes).values({
|
||||
modelPattern: 'claude-sonnet-4-5',
|
||||
enabled: true,
|
||||
}).returning().get();
|
||||
|
||||
await db.insert(schema.routeChannels).values([
|
||||
{
|
||||
routeId: sourceRouteA.id,
|
||||
accountId: account.id,
|
||||
tokenId: token.id,
|
||||
sourceModel: 'claude-opus-4-5',
|
||||
enabled: true,
|
||||
},
|
||||
{
|
||||
routeId: sourceRouteB.id,
|
||||
accountId: account.id,
|
||||
tokenId: token.id,
|
||||
sourceModel: 'claude-sonnet-4-5',
|
||||
enabled: true,
|
||||
},
|
||||
]).run();
|
||||
|
||||
const groupResponse = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/api/routes',
|
||||
payload: {
|
||||
routeMode: 'explicit_group',
|
||||
displayName: 'claude-opus-4-6',
|
||||
sourceRouteIds: [sourceRouteA.id, sourceRouteB.id],
|
||||
},
|
||||
});
|
||||
expect(groupResponse.statusCode).toBe(200);
|
||||
const groupId = (groupResponse.json() as { id: number }).id;
|
||||
|
||||
await db.insert(schema.downstreamApiKeys).values({
|
||||
name: 'managed-explicit-group-key',
|
||||
key: 'sk-managed-explicit-group',
|
||||
enabled: true,
|
||||
allowedRouteIds: JSON.stringify([groupId]),
|
||||
}).run();
|
||||
|
||||
const response = await app.inject({
|
||||
method: 'GET',
|
||||
url: '/v1/models',
|
||||
headers: {
|
||||
authorization: 'Bearer sk-managed-explicit-group',
|
||||
},
|
||||
});
|
||||
|
||||
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('claude-opus-4-6');
|
||||
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',
|
||||
|
||||
@@ -1,6 +1,4 @@
|
||||
import { FastifyInstance } from 'fastify';
|
||||
import { db, schema } from '../../db/index.js';
|
||||
import { and, eq } from 'drizzle-orm';
|
||||
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
|
||||
import { getDownstreamRoutingPolicy } from './downstreamPolicy.js';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
@@ -17,28 +15,7 @@ export async function modelsProxyRoute(app: FastifyInstance) {
|
||||
const downstreamPolicy = getDownstreamRoutingPolicy(request);
|
||||
|
||||
const readModels = async () => {
|
||||
const rows = await db.select({ modelName: schema.modelAvailability.modelName })
|
||||
.from(schema.modelAvailability)
|
||||
.innerJoin(schema.accounts, eq(schema.modelAvailability.accountId, schema.accounts.id))
|
||||
.innerJoin(schema.sites, eq(schema.accounts.siteId, schema.sites.id))
|
||||
.where(
|
||||
and(
|
||||
eq(schema.modelAvailability.available, true),
|
||||
eq(schema.accounts.status, 'active'),
|
||||
eq(schema.sites.status, 'active'),
|
||||
),
|
||||
)
|
||||
.all();
|
||||
const routeAliases = (await db.select({ displayName: schema.tokenRoutes.displayName })
|
||||
.from(schema.tokenRoutes)
|
||||
.where(eq(schema.tokenRoutes.enabled, true))
|
||||
.all())
|
||||
.map((row) => (row.displayName || '').trim())
|
||||
.filter((name) => name.length > 0);
|
||||
const deduped = Array.from(new Set([
|
||||
...rows.map((r) => r.modelName),
|
||||
...routeAliases,
|
||||
]))
|
||||
const deduped = Array.from(new Set(await tokenRouter.getAvailableModels()))
|
||||
.filter((modelName) => !isSearchPseudoModel(modelName))
|
||||
.sort();
|
||||
const allowed: string[] = [];
|
||||
|
||||
@@ -204,6 +204,9 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
model: modelName,
|
||||
stream: isStream,
|
||||
};
|
||||
if (body.generate === false) {
|
||||
normalizedResponsesBody.generate = false;
|
||||
}
|
||||
if (owner) {
|
||||
try {
|
||||
normalizedResponsesBody = await resolveResponsesBodyInputFiles(normalizedResponsesBody, owner);
|
||||
@@ -446,6 +449,9 @@ export async function responsesProxyRoute(app: FastifyInstance) {
|
||||
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') {
|
||||
|
||||
@@ -6,6 +6,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vites
|
||||
const fetchMock = vi.fn();
|
||||
const selectChannelMock = vi.fn();
|
||||
const selectNextChannelMock = vi.fn();
|
||||
const previewSelectedChannelMock = vi.fn();
|
||||
const recordSuccessMock = vi.fn();
|
||||
const recordFailureMock = vi.fn();
|
||||
const refreshModelsAndRebuildRoutesMock = vi.fn();
|
||||
@@ -30,6 +31,7 @@ vi.mock('../../services/tokenRouter.js', () => ({
|
||||
tokenRouter: {
|
||||
selectChannel: (...args: unknown[]) => selectChannelMock(...args),
|
||||
selectNextChannel: (...args: unknown[]) => selectNextChannelMock(...args),
|
||||
previewSelectedChannel: (...args: unknown[]) => previewSelectedChannelMock(...args),
|
||||
recordSuccess: (...args: unknown[]) => recordSuccessMock(...args),
|
||||
recordFailure: (...args: unknown[]) => recordFailureMock(...args),
|
||||
},
|
||||
@@ -91,6 +93,44 @@ function createSseResponse(chunks: string[], status = 200) {
|
||||
});
|
||||
}
|
||||
|
||||
function createSelectedChannel(options?: {
|
||||
siteName?: string;
|
||||
siteUrl?: string;
|
||||
sitePlatform?: string;
|
||||
username?: string;
|
||||
extraConfig?: string;
|
||||
tokenValue?: string;
|
||||
actualModel?: string;
|
||||
}) {
|
||||
const sitePlatform = options?.sitePlatform ?? 'codex';
|
||||
const isCodex = sitePlatform === 'codex';
|
||||
return {
|
||||
channel: { id: 11, routeId: 22 },
|
||||
site: {
|
||||
name: options?.siteName ?? (isCodex ? 'codex-site' : 'openai-site'),
|
||||
url: options?.siteUrl ?? (isCodex ? 'https://chatgpt.com/backend-api/codex' : 'https://api.openai.com'),
|
||||
platform: sitePlatform,
|
||||
},
|
||||
account: {
|
||||
id: 33,
|
||||
username: options?.username ?? (isCodex ? 'codex-user@example.com' : 'openai-user@example.com'),
|
||||
extraConfig: options?.extraConfig ?? (isCodex
|
||||
? JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-123',
|
||||
email: 'codex-user@example.com',
|
||||
},
|
||||
})
|
||||
: '{}'),
|
||||
},
|
||||
tokenName: 'default',
|
||||
tokenValue: options?.tokenValue ?? (isCodex ? 'oauth-access-token' : 'sk-openai-token'),
|
||||
actualModel: options?.actualModel ?? (isCodex ? 'gpt-5.4' : 'gpt-4.1'),
|
||||
};
|
||||
}
|
||||
|
||||
function waitForSocketOpen(socket: WebSocket) {
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
socket.once('open', () => resolve());
|
||||
@@ -98,19 +138,37 @@ function waitForSocketOpen(socket: WebSocket) {
|
||||
});
|
||||
}
|
||||
|
||||
function waitForSocketMessages(socket: WebSocket, count: number) {
|
||||
function waitForSocketUpgrade(socket: WebSocket) {
|
||||
return new Promise<{ headers: Record<string, string | string[] | undefined> }>((resolve, reject) => {
|
||||
socket.once('upgrade', (response) => resolve({ headers: response.headers as Record<string, string | string[] | undefined> }));
|
||||
socket.once('error', reject);
|
||||
});
|
||||
}
|
||||
|
||||
function waitForSocketMessages(socket: WebSocket, count: number, timeoutMs = 1000) {
|
||||
return new Promise<any[]>((resolve, reject) => {
|
||||
const messages: any[] = [];
|
||||
const timeout = setTimeout(() => {
|
||||
socket.off('message', onMessage);
|
||||
socket.off('error', onError);
|
||||
reject(new Error(`Timed out waiting for ${count} websocket messages`));
|
||||
}, timeoutMs);
|
||||
const onMessage = (payload: WebSocket.RawData) => {
|
||||
messages.push(JSON.parse(String(payload)));
|
||||
if (messages.length >= count) {
|
||||
clearTimeout(timeout);
|
||||
socket.off('message', onMessage);
|
||||
socket.off('error', reject);
|
||||
socket.off('error', onError);
|
||||
resolve(messages);
|
||||
}
|
||||
};
|
||||
const onError = (error: Error) => {
|
||||
clearTimeout(timeout);
|
||||
socket.off('message', onMessage);
|
||||
reject(error);
|
||||
};
|
||||
socket.on('message', onMessage);
|
||||
socket.once('error', reject);
|
||||
socket.once('error', onError);
|
||||
});
|
||||
}
|
||||
|
||||
@@ -131,6 +189,7 @@ describe('responses websocket transport', () => {
|
||||
fetchMock.mockReset();
|
||||
selectChannelMock.mockReset();
|
||||
selectNextChannelMock.mockReset();
|
||||
previewSelectedChannelMock.mockReset();
|
||||
recordSuccessMock.mockReset();
|
||||
recordFailureMock.mockReset();
|
||||
refreshModelsAndRebuildRoutesMock.mockReset();
|
||||
@@ -139,26 +198,10 @@ describe('responses websocket transport', () => {
|
||||
resolveProxyUsageWithSelfLogFallbackMock.mockClear();
|
||||
dbInsertMock.mockClear();
|
||||
|
||||
selectChannelMock.mockReturnValue({
|
||||
channel: { id: 11, routeId: 22 },
|
||||
site: { name: 'codex-site', url: 'https://chatgpt.com/backend-api/codex', platform: 'codex' },
|
||||
account: {
|
||||
id: 33,
|
||||
username: 'codex-user@example.com',
|
||||
extraConfig: JSON.stringify({
|
||||
credentialMode: 'session',
|
||||
oauth: {
|
||||
provider: 'codex',
|
||||
accountId: 'chatgpt-account-123',
|
||||
email: 'codex-user@example.com',
|
||||
},
|
||||
}),
|
||||
},
|
||||
tokenName: 'default',
|
||||
tokenValue: 'oauth-access-token',
|
||||
actualModel: 'gpt-5.4',
|
||||
});
|
||||
const selectedChannel = createSelectedChannel();
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
selectNextChannelMock.mockReturnValue(null);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
@@ -206,4 +249,306 @@ describe('responses websocket transport', () => {
|
||||
expect(messages[3]?.response?.output?.[0]?.content?.[0]?.text).toBe('pong');
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('echoes x-codex-turn-state on websocket upgrade responses', async () => {
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
|
||||
headers: {
|
||||
'x-codex-turn-state': 'turn-state-123',
|
||||
},
|
||||
});
|
||||
|
||||
const [upgrade] = await Promise.all([
|
||||
waitForSocketUpgrade(socket),
|
||||
waitForSocketOpen(socket),
|
||||
]);
|
||||
socket.close();
|
||||
|
||||
expect(upgrade.headers['x-codex-turn-state']).toBe('turn-state-123');
|
||||
});
|
||||
|
||||
it('merges follow-up response.create payloads when the selected upstream does not support incremental mode', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
sitePlatform: 'openai',
|
||||
actualModel: 'gpt-4.1',
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1"}}\n\n',
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}}\n\n',
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_1","model":"gpt-4.1","status":"completed","output":[{"id":"fc_1","type":"function_call","call_id":"call_1"},{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]))
|
||||
.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_2","model":"gpt-4.1","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-4.1',
|
||||
instructions: 'be helpful',
|
||||
input: [
|
||||
{
|
||||
id: 'msg_user_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'call the tool' }],
|
||||
},
|
||||
],
|
||||
}));
|
||||
await firstResponsePromise;
|
||||
|
||||
const secondResponsePromise = waitForSocketMessages(socket, 1);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
previous_response_id: 'resp_ws_1',
|
||||
input: [
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
],
|
||||
}));
|
||||
await secondResponsePromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, firstOptions] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, RequestInit];
|
||||
const firstBody = JSON.parse(String(firstOptions.body));
|
||||
const secondBody = JSON.parse(String(secondOptions.body));
|
||||
|
||||
expect(firstBody.input).toHaveLength(1);
|
||||
expect(secondBody.previous_response_id).toBeUndefined();
|
||||
expect(secondBody.model).toBe('gpt-4.1');
|
||||
expect(secondBody.instructions).toBe('be helpful');
|
||||
expect(secondBody.input).toEqual([
|
||||
{
|
||||
id: 'msg_user_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'call the tool' }],
|
||||
},
|
||||
{
|
||||
id: 'fc_1',
|
||||
type: 'function_call',
|
||||
call_id: 'call_1',
|
||||
status: 'completed',
|
||||
},
|
||||
{
|
||||
id: 'msg_1',
|
||||
type: 'message',
|
||||
role: 'assistant',
|
||||
status: 'completed',
|
||||
content: [{ type: 'output_text', text: 'call tool' }],
|
||||
},
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('preserves incremental response.create payloads with previous_response_id for websocket-capable upstreams', async () => {
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1"}}\n\n',
|
||||
'event: response.output_item.done\n',
|
||||
'data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}}\n\n',
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_1","model":"gpt-5.4","status":"completed","output":[{"id":"fc_1","type":"function_call","call_id":"call_1"},{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]))
|
||||
.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_2","model":"gpt-5.4","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const firstResponsePromise = waitForSocketMessages(socket, 1);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
instructions: 'be helpful',
|
||||
input: [
|
||||
{
|
||||
id: 'msg_user_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'call the tool' }],
|
||||
},
|
||||
],
|
||||
}));
|
||||
await firstResponsePromise;
|
||||
|
||||
const secondResponsePromise = waitForSocketMessages(socket, 1);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
previous_response_id: 'resp_ws_1',
|
||||
input: [
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
],
|
||||
}));
|
||||
await secondResponsePromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
const [, secondOptions] = fetchMock.mock.calls[1] as [string, RequestInit];
|
||||
const secondBody = JSON.parse(String(secondOptions.body));
|
||||
expect(secondBody.previous_response_id).toBe('resp_ws_1');
|
||||
expect(secondBody.model).toBe('gpt-5.4');
|
||||
expect(secondBody.instructions).toBe('be helpful');
|
||||
expect(secondBody.input).toEqual([
|
||||
{
|
||||
id: 'tool_out_1',
|
||||
type: 'function_call_output',
|
||||
call_id: 'call_1',
|
||||
output: 'tool result',
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('handles generate=false locally only for non-websocket-capable upstreams', async () => {
|
||||
const selectedChannel = createSelectedChannel({
|
||||
sitePlatform: 'openai',
|
||||
actualModel: 'gpt-4.1',
|
||||
});
|
||||
selectChannelMock.mockReturnValue(selectedChannel);
|
||||
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
|
||||
fetchMock.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_ws_after_prewarm","model":"gpt-4.1","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
await waitForSocketOpen(socket);
|
||||
|
||||
const prewarmMessagesPromise = waitForSocketMessages(socket, 2);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-4.1',
|
||||
generate: false,
|
||||
}));
|
||||
const prewarmMessages = await prewarmMessagesPromise;
|
||||
expect(prewarmMessages.map((message) => message.type)).toEqual(['response.created', 'response.completed']);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(0);
|
||||
|
||||
const secondResponsePromise = waitForSocketMessages(socket, 1);
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
previous_response_id: prewarmMessages[0]?.response?.id,
|
||||
input: [
|
||||
{
|
||||
id: 'msg_followup_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'continue' }],
|
||||
},
|
||||
],
|
||||
}));
|
||||
await secondResponsePromise;
|
||||
socket.close();
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
const [, options] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||
const forwardedBody = JSON.parse(String(options.body));
|
||||
expect(forwardedBody.generate).toBeUndefined();
|
||||
expect(forwardedBody.previous_response_id).toBeUndefined();
|
||||
expect(forwardedBody.model).toBe('gpt-4.1');
|
||||
expect(forwardedBody.input).toEqual([
|
||||
{
|
||||
id: 'msg_followup_1',
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'continue' }],
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('forwards generate=false upstream for websocket-capable upstreams instead of synthesizing prewarm events', async () => {
|
||||
fetchMock.mockResolvedValueOnce(createSseResponse([
|
||||
'event: response.completed\n',
|
||||
'data: {"type":"response.completed","response":{"id":"resp_capable_generate_false","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}\n\n',
|
||||
'data: [DONE]\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 1);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
generate: false,
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
const [, options] = fetchMock.mock.calls[0] as [string, RequestInit];
|
||||
const forwardedBody = JSON.parse(String(options.body));
|
||||
expect(forwardedBody.generate).toBe(false);
|
||||
});
|
||||
|
||||
it('emits websocket error when the upstream stream closes before a terminal responses event', async () => {
|
||||
fetchMock.mockResolvedValue(createSseResponse([
|
||||
'event: response.created\n',
|
||||
'data: {"type":"response.created","response":{"id":"resp_incomplete","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
|
||||
'event: response.output_text.delta\n',
|
||||
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_ws","delta":"partial"}\n\n',
|
||||
]));
|
||||
|
||||
const socket = new WebSocket(`${baseUrl}/v1/responses`);
|
||||
await waitForSocketOpen(socket);
|
||||
const messagesPromise = waitForSocketMessages(socket, 3, 400);
|
||||
|
||||
socket.send(JSON.stringify({
|
||||
type: 'response.create',
|
||||
model: 'gpt-5.4',
|
||||
input: [
|
||||
{
|
||||
type: 'message',
|
||||
role: 'user',
|
||||
content: [{ type: 'input_text', text: 'hello websocket' }],
|
||||
},
|
||||
],
|
||||
}));
|
||||
|
||||
const messages = await messagesPromise;
|
||||
socket.close();
|
||||
|
||||
expect(messages.map((message) => message.type)).toEqual([
|
||||
'response.created',
|
||||
'response.output_text.delta',
|
||||
'error',
|
||||
]);
|
||||
expect(messages[2]?.status).toBe(408);
|
||||
expect(messages[2]?.error?.message).toContain('stream closed before response.completed');
|
||||
});
|
||||
});
|
||||
|
||||
@@ -2,9 +2,13 @@ import { randomUUID } from 'node:crypto';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { IncomingMessage } from 'node:http';
|
||||
import { WebSocketServer, type RawData, type WebSocket } from 'ws';
|
||||
import { tokenRouter } from '../../services/tokenRouter.js';
|
||||
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
|
||||
|
||||
const installedApps = new WeakSet<FastifyInstance>();
|
||||
const WS_TURN_STATE_HEADER = 'x-codex-turn-state';
|
||||
const RESPONSES_WEBSOCKET_MODE_HEADER = 'x-metapi-responses-websocket-mode';
|
||||
const RESPONSES_WEBSOCKET_TRANSPORT_HEADER = 'x-metapi-responses-websocket-transport';
|
||||
|
||||
type NormalizedResponsesWebsocketRequest =
|
||||
| {
|
||||
@@ -26,6 +30,18 @@ function asTrimmedString(value: unknown): string {
|
||||
return typeof value === 'string' ? value.trim() : '';
|
||||
}
|
||||
|
||||
function headerValueToTrimmedString(value: unknown): string {
|
||||
if (typeof value === 'string') return value.trim();
|
||||
if (Array.isArray(value)) {
|
||||
for (const item of value) {
|
||||
if (typeof item !== 'string') continue;
|
||||
const trimmed = item.trim();
|
||||
if (trimmed) return trimmed;
|
||||
}
|
||||
}
|
||||
return '';
|
||||
}
|
||||
|
||||
function parseJsonObject(raw: RawData): Record<string, unknown> | null {
|
||||
try {
|
||||
const parsed = JSON.parse(String(raw));
|
||||
@@ -44,19 +60,11 @@ function toResponseInputArray(value: unknown): unknown[] {
|
||||
}
|
||||
|
||||
function normalizeResponsesWebsocketRequest(
|
||||
raw: RawData,
|
||||
parsed: Record<string, unknown>,
|
||||
lastRequest: Record<string, unknown> | null,
|
||||
lastResponseOutput: unknown[],
|
||||
supportsIncrementalInput: boolean,
|
||||
): NormalizedResponsesWebsocketRequest {
|
||||
const parsed = parseJsonObject(raw);
|
||||
if (!parsed) {
|
||||
return {
|
||||
ok: false,
|
||||
status: 400,
|
||||
message: 'Invalid websocket JSON payload',
|
||||
};
|
||||
}
|
||||
|
||||
const requestType = asTrimmedString(parsed.type);
|
||||
if (requestType !== 'response.create' && requestType !== 'response.append') {
|
||||
return {
|
||||
@@ -66,27 +74,6 @@ function normalizeResponsesWebsocketRequest(
|
||||
};
|
||||
}
|
||||
|
||||
if (requestType === 'response.create' && parsed.generate === false) {
|
||||
const next = cloneJsonObject(parsed);
|
||||
delete next.type;
|
||||
delete next.generate;
|
||||
next.stream = true;
|
||||
if (!Array.isArray(next.input)) next.input = [];
|
||||
const modelName = asTrimmedString(next.model);
|
||||
if (!modelName) {
|
||||
return {
|
||||
ok: false,
|
||||
status: 400,
|
||||
message: 'missing model in response.create request',
|
||||
};
|
||||
}
|
||||
return {
|
||||
ok: true,
|
||||
request: next,
|
||||
nextRequestSnapshot: cloneJsonObject(next),
|
||||
};
|
||||
}
|
||||
|
||||
if (!lastRequest) {
|
||||
if (requestType !== 'response.create') {
|
||||
return {
|
||||
@@ -97,6 +84,9 @@ function normalizeResponsesWebsocketRequest(
|
||||
}
|
||||
const next = cloneJsonObject(parsed);
|
||||
delete next.type;
|
||||
if (!supportsIncrementalInput && parsed.generate === false) {
|
||||
delete next.generate;
|
||||
}
|
||||
next.stream = true;
|
||||
if (!Array.isArray(next.input)) next.input = [];
|
||||
const modelName = asTrimmedString(next.model);
|
||||
@@ -122,15 +112,8 @@ function normalizeResponsesWebsocketRequest(
|
||||
};
|
||||
}
|
||||
|
||||
const mergedInput = [
|
||||
...toResponseInputArray(lastRequest.input),
|
||||
...cloneJsonObject(lastResponseOutput),
|
||||
...cloneJsonObject(parsed.input),
|
||||
];
|
||||
const next = cloneJsonObject(parsed);
|
||||
delete next.type;
|
||||
delete next.previous_response_id;
|
||||
next.input = mergedInput;
|
||||
next.stream = true;
|
||||
if (!('model' in next) && typeof lastRequest.model === 'string') {
|
||||
next.model = lastRequest.model;
|
||||
@@ -139,6 +122,22 @@ function normalizeResponsesWebsocketRequest(
|
||||
next.instructions = cloneJsonObject(lastRequest.instructions);
|
||||
}
|
||||
|
||||
if (supportsIncrementalInput && requestType === 'response.create' && asTrimmedString(parsed.previous_response_id)) {
|
||||
return {
|
||||
ok: true,
|
||||
request: next,
|
||||
nextRequestSnapshot: cloneJsonObject(next),
|
||||
};
|
||||
}
|
||||
|
||||
const mergedInput = [
|
||||
...toResponseInputArray(lastRequest.input),
|
||||
...cloneJsonObject(lastResponseOutput),
|
||||
...cloneJsonObject(parsed.input),
|
||||
];
|
||||
delete next.previous_response_id;
|
||||
next.input = mergedInput;
|
||||
|
||||
return {
|
||||
ok: true,
|
||||
request: next,
|
||||
@@ -146,6 +145,16 @@ function normalizeResponsesWebsocketRequest(
|
||||
};
|
||||
}
|
||||
|
||||
function shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed: Record<string, unknown>,
|
||||
lastRequest: Record<string, unknown> | null,
|
||||
supportsIncrementalInput: boolean,
|
||||
): boolean {
|
||||
if (supportsIncrementalInput || lastRequest) return false;
|
||||
if (asTrimmedString(parsed.type) !== 'response.create') return false;
|
||||
return parsed.generate === false;
|
||||
}
|
||||
|
||||
function writeResponsesWebsocketError(
|
||||
socket: WebSocket,
|
||||
status: number,
|
||||
@@ -244,6 +253,21 @@ function buildInjectHeaders(request: IncomingMessage): Record<string, string | s
|
||||
return headers;
|
||||
}
|
||||
|
||||
async function supportsResponsesWebsocketIncrementalInput(
|
||||
parsed: Record<string, unknown>,
|
||||
lastRequest: Record<string, unknown> | null,
|
||||
): Promise<boolean> {
|
||||
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
|
||||
if (!requestModel) return false;
|
||||
|
||||
try {
|
||||
const selected = await tokenRouter.previewSelectedChannel(requestModel);
|
||||
return asTrimmedString(selected?.site?.platform).toLowerCase() === 'codex';
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function handleResponsesWebsocketConnection(
|
||||
app: FastifyInstance,
|
||||
socket: WebSocket,
|
||||
@@ -259,14 +283,25 @@ async function handleResponsesWebsocketConnection(
|
||||
return;
|
||||
}
|
||||
|
||||
const normalized = normalizeResponsesWebsocketRequest(raw, lastRequest, lastResponseOutput);
|
||||
const supportsIncrementalInput = await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
|
||||
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
|
||||
parsed,
|
||||
lastRequest,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
const normalized = normalizeResponsesWebsocketRequest(
|
||||
parsed,
|
||||
lastRequest,
|
||||
lastResponseOutput,
|
||||
supportsIncrementalInput,
|
||||
);
|
||||
if (!normalized.ok) {
|
||||
writeResponsesWebsocketError(socket, normalized.status, normalized.message);
|
||||
return;
|
||||
}
|
||||
|
||||
lastRequest = normalized.nextRequestSnapshot;
|
||||
if (parsed.generate === false && asTrimmedString(parsed.type) === 'response.create') {
|
||||
if (shouldHandleLocalPrewarm) {
|
||||
lastResponseOutput = [];
|
||||
for (const payload of synthesizePrewarmResponsePayloads(normalized.request)) {
|
||||
socket.send(JSON.stringify(payload));
|
||||
@@ -277,7 +312,11 @@ async function handleResponsesWebsocketConnection(
|
||||
const response = await app.inject({
|
||||
method: 'POST',
|
||||
url: '/v1/responses',
|
||||
headers: buildInjectHeaders(request),
|
||||
headers: {
|
||||
...buildInjectHeaders(request),
|
||||
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
|
||||
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
|
||||
},
|
||||
payload: normalized.request,
|
||||
});
|
||||
|
||||
@@ -313,17 +352,25 @@ async function handleResponsesWebsocketConnection(
|
||||
|
||||
const pulled = openAiResponsesTransformer.pullSseEvents(response.body);
|
||||
const forwardedPayloads: unknown[] = [];
|
||||
let sawTerminalPayload = false;
|
||||
for (const event of pulled.events) {
|
||||
if (event.data === '[DONE]') continue;
|
||||
try {
|
||||
const payload = JSON.parse(event.data);
|
||||
forwardedPayloads.push(payload);
|
||||
const type = isRecord(payload) ? asTrimmedString(payload.type) : '';
|
||||
if (type === 'response.completed' || type === 'response.failed') {
|
||||
sawTerminalPayload = true;
|
||||
}
|
||||
socket.send(JSON.stringify(payload));
|
||||
} catch {
|
||||
// Ignore malformed SSE frames; the HTTP route already normalizes them.
|
||||
}
|
||||
}
|
||||
lastResponseOutput = collectResponsesOutput(forwardedPayloads);
|
||||
if (!sawTerminalPayload) {
|
||||
writeResponsesWebsocketError(socket, 408, 'stream closed before response.completed');
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -332,6 +379,11 @@ export function ensureResponsesWebsocketTransport(app: FastifyInstance) {
|
||||
installedApps.add(app);
|
||||
|
||||
const websocketServer = new WebSocketServer({ noServer: true });
|
||||
websocketServer.on('headers', (headers, request) => {
|
||||
const turnState = headerValueToTrimmedString(request.headers[WS_TURN_STATE_HEADER]);
|
||||
if (!turnState) return;
|
||||
headers.push(`${WS_TURN_STATE_HEADER}: ${turnState}`);
|
||||
});
|
||||
|
||||
app.server.on('upgrade', (request, socket, head) => {
|
||||
const url = new URL(request.url || '/', 'http://localhost');
|
||||
|
||||
@@ -470,6 +470,9 @@ function convertCodexSystemRoleToDeveloper(input: unknown): unknown {
|
||||
function applyCodexResponsesCompatibility(
|
||||
body: Record<string, unknown>,
|
||||
sitePlatform: string,
|
||||
options?: {
|
||||
preservePreviousResponseId?: boolean;
|
||||
},
|
||||
): Record<string, unknown> {
|
||||
if (sitePlatform !== 'codex') return body;
|
||||
|
||||
@@ -494,12 +497,14 @@ function applyCodexResponsesCompatibility(
|
||||
'truncation',
|
||||
'user',
|
||||
'context_management',
|
||||
'previous_response_id',
|
||||
'prompt_cache_retention',
|
||||
'safety_identifier',
|
||||
]) {
|
||||
delete next[key];
|
||||
}
|
||||
if (!options?.preservePreviousResponseId) {
|
||||
delete next.previous_response_id;
|
||||
}
|
||||
|
||||
if (asTrimmedString(next.service_tier).toLowerCase() !== 'priority') {
|
||||
delete next.service_tier;
|
||||
@@ -974,6 +979,8 @@ export function buildUpstreamEndpointRequest(input: {
|
||||
}
|
||||
|
||||
if (input.endpoint === 'responses') {
|
||||
const websocketMode = Object.entries(input.downstreamHeaders || {}).find(([rawKey]) => rawKey.trim().toLowerCase() === 'x-metapi-responses-websocket-mode');
|
||||
const preserveWebsocketIncrementalMode = asTrimmedString(websocketMode?.[1]).toLowerCase() === 'incremental';
|
||||
const responsesHeaders = input.downstreamFormat === 'responses'
|
||||
? extractResponsesPassthroughHeaders(input.downstreamHeaders)
|
||||
: {};
|
||||
@@ -986,11 +993,16 @@ export function buildUpstreamEndpointRequest(input: {
|
||||
}
|
||||
: convertOpenAiBodyToResponsesBodyViaTransformer(openaiBody, input.modelName, input.stream)
|
||||
);
|
||||
const sanitizedResponsesBody = sanitizeResponsesBodyForProxyViaTransformer(rawBody, input.modelName, input.stream);
|
||||
if (preserveWebsocketIncrementalMode && rawBody.generate === false) {
|
||||
sanitizedResponsesBody.generate = false;
|
||||
}
|
||||
const body = ensureCodexResponsesStoreFalse(
|
||||
ensureCodexResponsesInstructions(
|
||||
applyCodexResponsesCompatibility(
|
||||
sanitizeResponsesBodyForProxyViaTransformer(rawBody, input.modelName, input.stream),
|
||||
sanitizedResponsesBody,
|
||||
sitePlatform,
|
||||
{ preservePreviousResponseId: preserveWebsocketIncrementalMode },
|
||||
),
|
||||
sitePlatform,
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user