Merge remote-tracking branch 'origin/main' into feat/account-level-proxy

# Conflicts:
#	src/server/routes/proxy/chat.ts
#	src/server/routes/proxy/responses.ts
This commit is contained in:
Hureru
2026-03-19 23:18:41 +08:00
93 changed files with 10221 additions and 4191 deletions
@@ -0,0 +1,53 @@
import type { CliProfileDefinition } from './types.js';
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object' && !Array.isArray(value);
}
const claudeCodeUserIdPattern = /^user_[0-9a-f]{64}_account__session_[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
function isClaudeSurface(path: string): boolean {
const normalizedPath = path.trim().toLowerCase();
return normalizedPath === '/v1/messages'
|| normalizedPath === '/anthropic/v1/messages'
|| normalizedPath === '/v1/messages/count_tokens';
}
export function extractClaudeCodeSessionId(userId: string): string | null {
const trimmed = userId.trim();
if (!claudeCodeUserIdPattern.test(trimmed)) return null;
const sessionPrefix = '__session_';
const sessionIndex = trimmed.lastIndexOf(sessionPrefix);
if (sessionIndex === -1) return null;
const sessionId = trimmed.slice(sessionIndex + sessionPrefix.length).trim();
return sessionId || null;
}
export const claudeCodeCliProfile: CliProfileDefinition = {
id: 'claude_code',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: true,
supportsCountTokens: true,
echoesTurnState: false,
},
detect(input) {
if (!isClaudeSurface(input.downstreamPath)) return null;
if (!isRecord(input.body) || !isRecord(input.body.metadata)) return null;
const userId = typeof input.body.metadata.user_id === 'string'
? input.body.metadata.user_id.trim()
: '';
const sessionId = userId ? extractClaudeCodeSessionId(userId) : null;
if (!sessionId) return null;
return {
id: 'claude_code',
sessionId,
traceHint: sessionId,
};
},
};
@@ -0,0 +1,91 @@
import type { CliProfileDefinition, DetectCliProfileInput } from './types.js';
function headerValueToString(value: unknown): string | null {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed || null;
}
if (Array.isArray(value)) {
for (const item of value) {
if (typeof item !== 'string') continue;
const trimmed = item.trim();
if (trimmed) return trimmed;
}
}
return null;
}
function getHeaderValue(headers: Record<string, unknown> | undefined, targetKey: string): string | null {
if (!headers) return null;
const normalizedTarget = targetKey.trim().toLowerCase();
for (const [rawKey, rawValue] of Object.entries(headers)) {
if (rawKey.trim().toLowerCase() !== normalizedTarget) continue;
return headerValueToString(rawValue);
}
return null;
}
function hasHeaderPrefix(headers: Record<string, unknown> | undefined, prefix: string): boolean {
if (!headers) return false;
const normalizedPrefix = prefix.trim().toLowerCase();
return Object.entries(headers).some(([rawKey, rawValue]) => {
const key = rawKey.trim().toLowerCase();
return key.startsWith(normalizedPrefix) && !!headerValueToString(rawValue);
});
}
function isCodexPath(path: string): boolean {
const normalizedPath = path.trim().toLowerCase();
return normalizedPath.startsWith('/v1/responses')
|| normalizedPath === '/v1/chat/completions'
|| normalizedPath.startsWith('/v1/messages');
}
export function isCodexResponsesSurface(headers?: Record<string, unknown>): boolean {
return isCodexRequest({
downstreamPath: '/v1/responses',
headers,
});
}
export function getCodexSessionId(headers?: Record<string, unknown>): string | null {
return getHeaderValue(headers, 'session_id') || getHeaderValue(headers, 'session-id');
}
export function isCodexRequest(input: DetectCliProfileInput): boolean {
if (!isCodexPath(input.downstreamPath)) return false;
const headers = input.headers;
if (!headers) return false;
const originator = getHeaderValue(headers, 'originator');
if (originator?.toLowerCase() === 'codex_cli_rs') return true;
if (getHeaderValue(headers, 'openai-beta')) return true;
if (hasHeaderPrefix(headers, 'x-stainless-')) return true;
if (getCodexSessionId(headers)) return true;
if (getHeaderValue(headers, 'x-codex-turn-state')) return true;
return false;
}
export const codexCliProfile: CliProfileDefinition = {
id: 'codex',
capabilities: {
supportsResponsesCompact: true,
supportsResponsesWebsocketIncremental: true,
preservesContinuation: true,
supportsCountTokens: false,
echoesTurnState: true,
},
detect(input) {
if (!isCodexRequest(input)) return null;
const sessionId = getCodexSessionId(input.headers) || undefined;
return {
id: 'codex',
...(sessionId ? { sessionId, traceHint: sessionId } : {}),
};
},
};
@@ -0,0 +1,37 @@
import type { CliProfileDefinition } from './types.js';
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object' && !Array.isArray(value);
}
function isGeminiCliPath(path: string): boolean {
const normalizedPath = path.trim().toLowerCase();
return normalizedPath === '/v1internal:generatecontent'
|| normalizedPath === '/v1internal:streamgeneratecontent'
|| normalizedPath === '/v1internal:counttokens';
}
function hasGeminiCliBodyShape(body: unknown): boolean {
if (!isRecord(body)) return false;
return typeof body.model === 'string'
|| Array.isArray(body.contents)
|| (isRecord(body.request) && (Array.isArray(body.request.contents) || typeof body.request.model === 'string'));
}
export const geminiCliProfile: CliProfileDefinition = {
id: 'gemini_cli',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: false,
supportsCountTokens: true,
echoesTurnState: false,
},
detect(input) {
if (!isGeminiCliPath(input.downstreamPath)) return null;
if (input.body !== undefined && !hasGeminiCliBodyShape(input.body)) return null;
return {
id: 'gemini_cli',
};
},
};
@@ -0,0 +1,17 @@
import type { CliProfileDefinition } from './types.js';
export const genericCliProfile: CliProfileDefinition = {
id: 'generic',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: false,
supportsCountTokens: false,
echoesTurnState: false,
},
detect() {
return {
id: 'generic',
};
},
};
@@ -0,0 +1,100 @@
import { describe, expect, it } from 'vitest';
import { detectCliProfile } from './registry.js';
describe('detectCliProfile', () => {
it('detects Codex responses requests and exposes Codex capability flags', () => {
expect(detectCliProfile({
downstreamPath: '/v1/responses',
headers: {
originator: 'codex_cli_rs',
Session_id: 'codex-session-123',
},
})).toEqual({
id: 'codex',
sessionId: 'codex-session-123',
traceHint: 'codex-session-123',
capabilities: {
supportsResponsesCompact: true,
supportsResponsesWebsocketIncremental: true,
preservesContinuation: true,
supportsCountTokens: false,
echoesTurnState: true,
},
});
});
it('treats x-codex-turn-state as a Codex marker even when session_id is absent', () => {
expect(detectCliProfile({
downstreamPath: '/v1/responses',
headers: {
'x-codex-turn-state': 'turn-state-123',
},
})).toMatchObject({
id: 'codex',
capabilities: {
supportsResponsesWebsocketIncremental: true,
echoesTurnState: true,
},
});
});
it('detects Claude Code requests on the count_tokens surface and exposes token counting support', () => {
expect(detectCliProfile({
downstreamPath: '/v1/messages/count_tokens',
body: {
metadata: {
user_id: 'user_20836b5653ed68aa981604f502c0a491397f6053826a93c953423632578d38ad_account__session_f25958b8-e75c-455d-8b40-f006d87cc2a4',
},
},
})).toEqual({
id: 'claude_code',
sessionId: 'f25958b8-e75c-455d-8b40-f006d87cc2a4',
traceHint: 'f25958b8-e75c-455d-8b40-f006d87cc2a4',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: true,
supportsCountTokens: true,
echoesTurnState: false,
},
});
});
it('detects Gemini CLI internal routes and exposes Gemini CLI capability flags', () => {
expect(detectCliProfile({
downstreamPath: '/v1internal:countTokens',
body: {
model: 'gpt-4.1',
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
})).toEqual({
id: 'gemini_cli',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: false,
supportsCountTokens: true,
echoesTurnState: false,
},
});
});
it('falls back to generic for native Gemini routes', () => {
expect(detectCliProfile({
downstreamPath: '/v1beta/models/gemini-2.5-flash:generateContent',
body: {
contents: [{ role: 'user', parts: [{ text: 'hello' }] }],
},
})).toEqual({
id: 'generic',
capabilities: {
supportsResponsesCompact: false,
supportsResponsesWebsocketIncremental: false,
preservesContinuation: false,
supportsCountTokens: false,
echoesTurnState: false,
},
});
});
});
@@ -0,0 +1,38 @@
import { claudeCodeCliProfile } from './claudeCodeProfile.js';
import { codexCliProfile } from './codexProfile.js';
import { geminiCliProfile } from './geminiCliProfile.js';
import { genericCliProfile } from './genericProfile.js';
import type { CliProfileDefinition, CliProfileId, DetectCliProfileInput, DetectedCliProfile } from './types.js';
const orderedProfiles: CliProfileDefinition[] = [
claudeCodeCliProfile,
codexCliProfile,
geminiCliProfile,
];
export const cliProfileRegistry: Record<CliProfileId, CliProfileDefinition> = {
generic: genericCliProfile,
codex: codexCliProfile,
claude_code: claudeCodeCliProfile,
gemini_cli: geminiCliProfile,
};
export function getCliProfileDefinition(id: CliProfileId): CliProfileDefinition {
return cliProfileRegistry[id];
}
export function detectCliProfile(input: DetectCliProfileInput): DetectedCliProfile {
for (const profile of orderedProfiles) {
const detected = profile.detect(input);
if (!detected) continue;
return {
...detected,
capabilities: profile.capabilities,
};
}
return {
id: genericCliProfile.id,
capabilities: genericCliProfile.capabilities,
};
}
@@ -0,0 +1,32 @@
export type CliProfileId =
| 'generic'
| 'codex'
| 'claude_code'
| 'gemini_cli';
export type CliProfileCapabilities = {
supportsResponsesCompact: boolean;
supportsResponsesWebsocketIncremental: boolean;
preservesContinuation: boolean;
supportsCountTokens: boolean;
echoesTurnState: boolean;
};
export type DetectCliProfileInput = {
downstreamPath: string;
headers?: Record<string, unknown>;
body?: unknown;
};
export type DetectedCliProfile = {
id: CliProfileId;
sessionId?: string;
traceHint?: string;
capabilities: CliProfileCapabilities;
};
export type CliProfileDefinition = {
id: CliProfileId;
capabilities: CliProfileCapabilities;
detect(input: DetectCliProfileInput): Omit<DetectedCliProfile, 'capabilities'> | null;
};
@@ -0,0 +1,241 @@
import { describe, expect, it, vi } from 'vitest';
import { DefaultProxyConductor } from './DefaultProxyConductor.js';
import { terminalStreamFailure } from './streamTermination.js';
const baseSelectedChannel = {
channel: { id: 11, routeId: 22 },
site: { id: 44, name: 'demo-site', url: 'https://upstream.example.com', platform: 'openai' },
account: { id: 33, username: 'demo-user' },
tokenName: 'default',
tokenValue: 'sk-demo',
actualModel: 'upstream-gpt',
};
describe('DefaultProxyConductor', () => {
it('returns the first selected channel when the first attempt succeeds', async () => {
const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
const selectNextChannel = vi.fn();
const recordSuccess = vi.fn().mockResolvedValue(undefined);
const recordFailure = vi.fn().mockResolvedValue(undefined);
const conductor = new DefaultProxyConductor({
selectChannel,
selectNextChannel,
recordSuccess,
recordFailure,
});
const attempt = vi.fn().mockResolvedValue({
ok: true,
response: new Response('ok', { status: 200 }),
latencyMs: 12,
cost: 0.25,
});
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt,
});
expect(result).toMatchObject({
ok: true,
selected: baseSelectedChannel,
attempts: 1,
});
expect(selectChannel).toHaveBeenCalledWith('gpt-5.4', undefined);
expect(selectNextChannel).not.toHaveBeenCalled();
expect(recordFailure).not.toHaveBeenCalled();
expect(recordSuccess).toHaveBeenCalledWith(11, {
latencyMs: 12,
cost: 0.25,
});
});
it('retries on the same channel when the attempt asks for a same-channel retry', async () => {
const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
const selectNextChannel = vi.fn();
const recordSuccess = vi.fn().mockResolvedValue(undefined);
const recordFailure = vi.fn().mockResolvedValue(undefined);
const conductor = new DefaultProxyConductor({
selectChannel,
selectNextChannel,
recordSuccess,
recordFailure,
});
const attempt = vi.fn()
.mockResolvedValueOnce({
ok: false,
action: 'retry_same_channel',
status: 429,
rawErrorText: 'rate limited',
})
.mockResolvedValueOnce({
ok: true,
response: new Response('ok', { status: 200 }),
});
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt,
});
expect(result).toMatchObject({
ok: true,
attempts: 2,
});
expect(selectNextChannel).not.toHaveBeenCalled();
expect(attempt).toHaveBeenCalledTimes(2);
expect(recordFailure).toHaveBeenCalledWith(11, {
status: 429,
rawErrorText: 'rate limited',
});
});
it('fails over to the next channel when the attempt asks for failover', async () => {
const nextSelectedChannel = {
...baseSelectedChannel,
channel: { id: 12, routeId: 22 },
tokenValue: 'sk-next',
};
const selectChannel = vi.fn().mockResolvedValue(baseSelectedChannel);
const selectNextChannel = vi.fn().mockResolvedValue(nextSelectedChannel);
const recordSuccess = vi.fn().mockResolvedValue(undefined);
const recordFailure = vi.fn().mockResolvedValue(undefined);
const conductor = new DefaultProxyConductor({
selectChannel,
selectNextChannel,
recordSuccess,
recordFailure,
});
const attempt = vi.fn()
.mockResolvedValueOnce({
ok: false,
action: 'failover',
status: 503,
rawErrorText: 'upstream unavailable',
})
.mockResolvedValueOnce({
ok: true,
response: new Response('ok', { status: 200 }),
});
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt,
});
expect(result).toMatchObject({
ok: true,
selected: nextSelectedChannel,
attempts: 2,
});
expect(selectNextChannel).toHaveBeenCalledWith('gpt-5.4', [11], undefined);
expect(recordFailure).toHaveBeenCalledWith(11, {
status: 503,
rawErrorText: 'upstream unavailable',
});
expect(recordSuccess).toHaveBeenCalledWith(12, {
latencyMs: null,
cost: null,
});
});
it('refreshes auth on 401 and retries the same channel with the refreshed selection', async () => {
const refreshedChannel = {
...baseSelectedChannel,
tokenValue: 'sk-refreshed',
};
const refreshAuth = vi.fn().mockResolvedValue(refreshedChannel);
const conductor = new DefaultProxyConductor({
selectChannel: vi.fn().mockResolvedValue(baseSelectedChannel),
selectNextChannel: vi.fn(),
recordSuccess: vi.fn().mockResolvedValue(undefined),
recordFailure: vi.fn().mockResolvedValue(undefined),
refreshAuth,
});
const attempt = vi.fn()
.mockResolvedValueOnce({
ok: false,
action: 'refresh_auth',
status: 401,
rawErrorText: 'expired token',
})
.mockResolvedValueOnce({
ok: true,
response: new Response('ok', { status: 200 }),
});
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt,
});
expect(result).toMatchObject({
ok: true,
selected: refreshedChannel,
attempts: 2,
});
expect(refreshAuth).toHaveBeenCalledWith(baseSelectedChannel, {
status: 401,
rawErrorText: 'expired token',
});
});
it('returns a no_channel result when no channel is available', async () => {
const conductor = new DefaultProxyConductor({
selectChannel: vi.fn().mockResolvedValue(null),
selectNextChannel: vi.fn(),
recordSuccess: vi.fn(),
recordFailure: vi.fn(),
previewSelectedChannel: vi.fn().mockResolvedValue(null),
});
expect(await conductor.previewSelectedChannel('gpt-5.4')).toBe(null);
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt: vi.fn(),
});
expect(result).toEqual({
ok: false,
reason: 'no_channel',
attempts: 0,
});
});
it('propagates terminal stream failures and calls the terminal failure hook', async () => {
const onTerminalFailure = vi.fn().mockResolvedValue(undefined);
const conductor = new DefaultProxyConductor({
selectChannel: vi.fn().mockResolvedValue(baseSelectedChannel),
selectNextChannel: vi.fn(),
recordSuccess: vi.fn(),
recordFailure: vi.fn().mockResolvedValue(undefined),
});
const attempt = vi.fn().mockResolvedValue({
ok: false,
...terminalStreamFailure({
status: 502,
rawErrorText: 'stream disconnected before completion',
}),
});
const result = await conductor.execute({
requestedModel: 'gpt-5.4',
attempt,
onTerminalFailure,
});
expect(result).toEqual({
ok: false,
reason: 'terminal',
selected: baseSelectedChannel,
status: 502,
rawErrorText: 'stream disconnected before completion',
attempts: 1,
});
expect(onTerminalFailure).toHaveBeenCalledWith(baseSelectedChannel, {
status: 502,
rawErrorText: 'stream disconnected before completion',
});
});
});
@@ -0,0 +1,127 @@
import {
failureActionOf,
isTerminalFailure,
shouldFailover,
shouldRefreshAuth,
shouldRetrySameChannel,
} from './retryPolicy.js';
import type { ExecuteInput, ExecuteResult, ProxyConductorDependencies, SelectedChannelLike } from './types.js';
import { recordFailedAttempt, recordSuccessfulAttempt } from './usageHooks.js';
export class DefaultProxyConductor {
constructor(private readonly deps: ProxyConductorDependencies) {}
async previewSelectedChannel(requestedModel: string, downstreamPolicy?: unknown): Promise<SelectedChannelLike | null> {
if (this.deps.previewSelectedChannel) {
return this.deps.previewSelectedChannel(requestedModel, downstreamPolicy);
}
return this.deps.selectChannel(requestedModel, downstreamPolicy);
}
async execute(input: ExecuteInput): Promise<ExecuteResult> {
const excludeChannelIds: number[] = [];
let attempts = 0;
let selected = await this.deps.selectChannel(input.requestedModel, input.downstreamPolicy);
if (!selected) {
return {
ok: false,
reason: 'no_channel',
attempts: 0,
};
}
while (selected) {
const result = await input.attempt({
selected,
attemptIndex: attempts,
excludeChannelIds: [...excludeChannelIds],
});
attempts += 1;
if (result.ok) {
await recordSuccessfulAttempt(this.deps, selected.channel.id, {
latencyMs: result.latencyMs ?? null,
cost: result.cost ?? null,
});
return {
ok: true,
selected,
response: result.response,
attempts,
};
}
const action = failureActionOf(result);
await recordFailedAttempt(this.deps, selected.channel.id, {
status: result.status,
rawErrorText: result.rawErrorText,
});
if (isTerminalFailure(action)) {
await input.onTerminalFailure?.(selected, {
status: result.status,
rawErrorText: result.rawErrorText,
});
return {
ok: false,
reason: 'terminal',
selected,
status: result.status,
rawErrorText: result.rawErrorText,
attempts,
};
}
if (shouldRetrySameChannel(action)) {
continue;
}
if (shouldRefreshAuth(action) && this.deps.refreshAuth) {
const refreshed = await this.deps.refreshAuth(selected, {
status: result.status,
rawErrorText: result.rawErrorText,
});
if (refreshed) {
selected = refreshed;
continue;
}
}
if (shouldFailover(action)) {
excludeChannelIds.push(selected.channel.id);
const next = await this.deps.selectNextChannel(
input.requestedModel,
excludeChannelIds,
input.downstreamPolicy,
);
if (!next) {
return {
ok: false,
reason: 'failed',
selected,
status: result.status,
rawErrorText: result.rawErrorText,
attempts,
};
}
selected = next;
continue;
}
return {
ok: false,
reason: 'failed',
selected,
status: result.status,
rawErrorText: result.rawErrorText,
attempts,
};
}
return {
ok: false,
reason: 'failed',
attempts,
};
}
}
@@ -0,0 +1,21 @@
import type { AttemptFailure, AttemptFailureAction } from './types.js';
export function failureActionOf(result: AttemptFailure): AttemptFailureAction {
return result.action;
}
export function shouldRetrySameChannel(action: AttemptFailureAction): boolean {
return action === 'retry_same_channel';
}
export function shouldRefreshAuth(action: AttemptFailureAction): boolean {
return action === 'refresh_auth';
}
export function shouldFailover(action: AttemptFailureAction): boolean {
return action === 'failover' || action === 'refresh_auth';
}
export function isTerminalFailure(action: AttemptFailureAction): boolean {
return action === 'terminal';
}
@@ -0,0 +1,10 @@
export function terminalStreamFailure(input: {
status?: number;
rawErrorText?: string;
}) {
return {
action: 'terminal' as const,
...(input.status !== undefined ? { status: input.status } : {}),
...(input.rawErrorText ? { rawErrorText: input.rawErrorText } : {}),
};
}
+80
View File
@@ -0,0 +1,80 @@
export type SelectedChannelLike = {
channel: { id: number; routeId?: number };
site: Record<string, unknown>;
account: Record<string, unknown>;
tokenName?: string;
tokenValue?: string;
actualModel?: string;
};
export type AttemptSuccess = {
ok: true;
response: Response;
latencyMs?: number | null;
cost?: number | null;
};
export type AttemptFailureAction =
| 'retry_same_channel'
| 'refresh_auth'
| 'failover'
| 'terminal'
| 'stop';
export type AttemptFailure = {
ok: false;
action: AttemptFailureAction;
status?: number;
rawErrorText?: string;
error?: unknown;
};
export type AttemptResult = AttemptSuccess | AttemptFailure;
export type ExecuteAttemptContext = {
selected: SelectedChannelLike;
attemptIndex: number;
excludeChannelIds: number[];
};
export type ProxyConductorDependencies = {
selectChannel: (requestedModel: string, downstreamPolicy?: unknown) => Promise<SelectedChannelLike | null>;
previewSelectedChannel?: (requestedModel: string, downstreamPolicy?: unknown) => Promise<SelectedChannelLike | null>;
selectNextChannel: (
requestedModel: string,
excludeChannelIds: number[],
downstreamPolicy?: unknown,
) => Promise<SelectedChannelLike | null>;
recordSuccess?: (channelId: number, metrics: { latencyMs: number | null; cost: number | null }) => Promise<void> | void;
recordFailure?: (channelId: number, failure: { status?: number; rawErrorText?: string }) => Promise<void> | void;
refreshAuth?: (
selected: SelectedChannelLike,
failure: { status?: number; rawErrorText?: string },
) => Promise<SelectedChannelLike | null>;
};
export type ExecuteInput = {
requestedModel: string;
downstreamPolicy?: unknown;
attempt: (context: ExecuteAttemptContext) => Promise<AttemptResult>;
onTerminalFailure?: (
selected: SelectedChannelLike,
failure: { status?: number; rawErrorText?: string },
) => Promise<void> | void;
};
export type ExecuteResult =
| {
ok: true;
selected: SelectedChannelLike;
response: Response;
attempts: number;
}
| {
ok: false;
reason: 'no_channel' | 'failed' | 'terminal';
selected?: SelectedChannelLike;
status?: number;
rawErrorText?: string;
attempts: number;
};
@@ -0,0 +1,20 @@
import type { ProxyConductorDependencies } from './types.js';
export async function recordSuccessfulAttempt(
deps: ProxyConductorDependencies,
channelId: number,
metrics: { latencyMs?: number | null; cost?: number | null },
): Promise<void> {
await deps.recordSuccess?.(channelId, {
latencyMs: metrics.latencyMs ?? null,
cost: metrics.cost ?? null,
});
}
export async function recordFailedAttempt(
deps: ProxyConductorDependencies,
channelId: number,
failure: { status?: number; rawErrorText?: string },
): Promise<void> {
await deps.recordFailure?.(channelId, failure);
}
@@ -0,0 +1,212 @@
import { createHash, randomUUID } from 'node:crypto';
import { Response } from 'undici';
import type { RuntimeDispatchInput, RuntimeExecutor, RuntimeResponse } from './types.js';
import {
asTrimmedString,
materializeErrorResponse,
performFetch,
withRequestBody,
} from './types.js';
const ANTIGRAVITY_RUNTIME_BASE_URLS = [
'https://daily-cloudcode-pa.googleapis.com',
'https://daily-cloudcode-pa.sandbox.googleapis.com',
] as const;
function antigravityRequestType(modelName: string): 'image_gen' | 'agent' {
return modelName.includes('image') ? 'image_gen' : 'agent';
}
function generateAntigravityProjectId(): string {
const adjectives = ['useful', 'bright', 'swift', 'calm', 'bold'];
const nouns = ['signal', 'river', 'rocket', 'forest', 'bridge'];
const adjective = adjectives[Math.floor(Math.random() * adjectives.length)] || 'useful';
const noun = nouns[Math.floor(Math.random() * nouns.length)] || 'signal';
const suffix = Math.floor(100000 + Math.random() * 900000);
return `${adjective}-${noun}-${suffix}`;
}
function extractFirstUserText(value: unknown): string {
if (!Array.isArray(value)) return '';
for (const content of value) {
if (!content || typeof content !== 'object' || Array.isArray(content)) continue;
const record = content as Record<string, unknown>;
if (asTrimmedString(record.role) !== 'user') continue;
const parts = Array.isArray(record.parts) ? record.parts : [];
for (const part of parts) {
if (!part || typeof part !== 'object' || Array.isArray(part)) continue;
const text = asTrimmedString((part as Record<string, unknown>).text);
if (text) return text;
}
}
return '';
}
function generateStableAntigravitySessionId(payload: Record<string, unknown>): string {
const firstUserText = extractFirstUserText(
payload.request && typeof payload.request === 'object' && !Array.isArray(payload.request)
? (payload.request as Record<string, unknown>).contents
: undefined,
);
if (!firstUserText) {
return `-${BigInt(`0x${randomUUID().replace(/-/g, '').slice(0, 16)}`).toString()}`;
}
const digest = createHash('sha256').update(firstUserText).digest('hex').slice(0, 16);
const bigint = BigInt(`0x${digest}`) & BigInt('0x7fffffffffffffff');
return `-${bigint.toString()}`;
}
function renameParametersJsonSchema(value: unknown): unknown {
if (Array.isArray(value)) {
return value.map((item) => renameParametersJsonSchema(item));
}
if (!value || typeof value !== 'object') return value;
const input = value as Record<string, unknown>;
const output: Record<string, unknown> = {};
for (const [key, entry] of Object.entries(input)) {
const nextKey = key === 'parametersJsonSchema' ? 'parameters' : key;
output[nextKey] = renameParametersJsonSchema(entry);
}
return output;
}
function deleteNestedMaxOutputTokens(payload: Record<string, unknown>): void {
const request = payload.request;
if (!request || typeof request !== 'object' || Array.isArray(request)) return;
const generationConfig = (request as Record<string, unknown>).generationConfig;
if (!generationConfig || typeof generationConfig !== 'object' || Array.isArray(generationConfig)) return;
delete (generationConfig as Record<string, unknown>).maxOutputTokens;
}
function buildAntigravityRuntimeBody(
originalBody: Record<string, unknown>,
modelName: string,
action?: NonNullable<RuntimeDispatchInput['request']['runtime']>['action'],
): Record<string, unknown> {
const payload = renameParametersJsonSchema(structuredClone(originalBody)) as Record<string, unknown>;
if (action === 'countTokens') {
return payload;
}
const requestType = antigravityRequestType(modelName);
const projectId = asTrimmedString(payload.project) || generateAntigravityProjectId();
payload.model = modelName;
payload.project = projectId;
payload.userAgent = 'antigravity';
payload.requestType = requestType;
payload.requestId = requestType === 'image_gen'
? `image_gen/${Date.now()}/${randomUUID()}/12`
: `agent-${randomUUID()}`;
const request = payload.request;
if (request && typeof request === 'object' && !Array.isArray(request)) {
delete (request as Record<string, unknown>).safetySettings;
if (requestType !== 'image_gen') {
(request as Record<string, unknown>).sessionId = generateStableAntigravitySessionId(payload);
}
if (modelName.includes('claude')) {
const toolConfig = (
(request as Record<string, unknown>).toolConfig
&& typeof (request as Record<string, unknown>).toolConfig === 'object'
&& !Array.isArray((request as Record<string, unknown>).toolConfig)
)
? (request as Record<string, unknown>).toolConfig as Record<string, unknown>
: (((request as Record<string, unknown>).toolConfig = {}) as Record<string, unknown>);
toolConfig.functionCallingConfig = { mode: 'VALIDATED' };
} else {
deleteNestedMaxOutputTokens(payload);
}
}
return payload;
}
function antigravityShouldRetryNoCapacity(
status: number,
responseText: string,
): boolean {
return status === 503 && responseText.toLowerCase().includes('no capacity available');
}
function antigravityNoCapacityRetryDelay(attempt: number): number {
return Math.min((attempt + 1) * 250, 2000);
}
async function waitMs(ms: number): Promise<void> {
if (ms <= 0) return;
await new Promise((resolve) => setTimeout(resolve, ms));
}
export const antigravityExecutor: RuntimeExecutor = {
async dispatch(input: RuntimeDispatchInput) {
const modelName = asTrimmedString(input.request.runtime?.modelName) || asTrimmedString(input.request.body.model);
const runtimeBody = buildAntigravityRuntimeBody(
input.request.body,
modelName,
input.request.runtime?.action,
);
const baseAttempts = 3;
let lastResponse: RuntimeResponse | null = null;
attemptLoop:
for (let attempt = 0; attempt < baseAttempts; attempt += 1) {
for (const baseUrl of ANTIGRAVITY_RUNTIME_BASE_URLS) {
const requestUrl = `${baseUrl}${input.request.path}`;
const minimalHeaders: Record<string, string> = {
Authorization: input.request.headers.Authorization || input.request.headers.authorization || '',
'Content-Type': 'application/json',
Accept: input.request.runtime?.stream ? 'text/event-stream' : 'application/json',
'User-Agent': 'antigravity/1.19.6 darwin/arm64',
};
let response: RuntimeResponse;
try {
response = await performFetch(
input,
withRequestBody(input.request, runtimeBody, minimalHeaders),
requestUrl,
);
} catch (error) {
if (baseUrl !== ANTIGRAVITY_RUNTIME_BASE_URLS[ANTIGRAVITY_RUNTIME_BASE_URLS.length - 1]) {
continue;
}
if (attempt + 1 < baseAttempts) {
continue attemptLoop;
}
throw error;
}
if (response.ok) return response;
const errorResponse = await materializeErrorResponse(response);
const errorText = await errorResponse.text().catch(() => '');
lastResponse = new Response(errorText, {
status: errorResponse.status,
headers: errorResponse.headers,
});
if (errorResponse.status === 429) {
continue;
}
if (antigravityShouldRetryNoCapacity(errorResponse.status, errorText)) {
if (baseUrl !== ANTIGRAVITY_RUNTIME_BASE_URLS[ANTIGRAVITY_RUNTIME_BASE_URLS.length - 1]) {
continue;
}
if (attempt + 1 < baseAttempts) {
await waitMs(antigravityNoCapacityRetryDelay(attempt));
continue attemptLoop;
}
}
return lastResponse;
}
}
return lastResponse || performFetch(input, withRequestBody(input.request, runtimeBody, {
Authorization: input.request.headers.Authorization || input.request.headers.authorization || '',
'Content-Type': 'application/json',
Accept: input.request.runtime?.stream ? 'text/event-stream' : 'application/json',
'User-Agent': 'antigravity/1.19.6 darwin/arm64',
}));
},
};
@@ -0,0 +1,8 @@
import type { RuntimeDispatchInput, RuntimeExecutor } from './types.js';
import { performFetch } from './types.js';
export const claudeExecutor: RuntimeExecutor = {
async dispatch(input: RuntimeDispatchInput) {
return performFetch(input, input.request);
},
};
@@ -0,0 +1,8 @@
import type { RuntimeDispatchInput, RuntimeExecutor } from './types.js';
import { performFetch } from './types.js';
export const codexExecutor: RuntimeExecutor = {
async dispatch(input: RuntimeDispatchInput) {
return performFetch(input, input.request);
},
};
@@ -0,0 +1,60 @@
import type { RuntimeDispatchInput, RuntimeExecutor, RuntimeResponse } from './types.js';
import {
asTrimmedString,
materializeErrorResponse,
performFetch,
withRequestBody,
} from './types.js';
function replaceGeminiCliModelInUserAgent(userAgent: string | undefined, modelName: string): string | undefined {
const raw = asTrimmedString(userAgent);
if (!raw) return undefined;
return raw.replace(/^GeminiCLI\/([^/]+)\/[^ ]+ /i, `GeminiCLI/$1/${modelName} `);
}
function buildGeminiCliAttemptRequest(
request: RuntimeDispatchInput['request'],
modelName: string,
) {
const body = structuredClone(request.body);
const action = request.runtime?.action;
if (action === 'countTokens') {
delete body.model;
delete body.project;
} else {
body.model = modelName;
}
const headers = { ...request.headers };
const nextUserAgent = replaceGeminiCliModelInUserAgent(
headers['User-Agent'] || headers['user-agent'],
modelName,
);
if (nextUserAgent) {
headers['User-Agent'] = nextUserAgent;
delete headers['user-agent'];
}
return withRequestBody(request, body, headers);
}
export const geminiCliExecutor: RuntimeExecutor = {
async dispatch(input: RuntimeDispatchInput) {
const baseModel = asTrimmedString(input.request.runtime?.modelName) || asTrimmedString(input.request.body.model);
const models = [baseModel].filter(Boolean);
let lastResponse: RuntimeResponse | null = null;
for (const modelName of models.length > 0 ? models : [baseModel || 'unknown']) {
const attemptRequest = buildGeminiCliAttemptRequest(input.request, modelName);
const response = await performFetch(input, attemptRequest);
if (response.ok) return response;
if (response.status === 429) {
lastResponse = await materializeErrorResponse(response);
continue;
}
return materializeErrorResponse(response);
}
return lastResponse || performFetch(input, input.request);
},
};
+75
View File
@@ -0,0 +1,75 @@
import {
Response,
fetch,
type RequestInit as UndiciRequestInit,
type Response as UndiciResponse,
} from 'undici';
export type ProxyRuntimeRequest = {
endpoint: 'chat' | 'messages' | 'responses';
path: string;
headers: Record<string, string>;
body: Record<string, unknown>;
runtime?: {
executor: 'default' | 'codex' | 'gemini-cli' | 'antigravity' | 'claude';
modelName?: string;
stream?: boolean;
oauthProjectId?: string | null;
action?: 'generateContent' | 'streamGenerateContent' | 'countTokens';
};
};
export type RuntimeDispatchInput = {
siteUrl: string;
request: ProxyRuntimeRequest;
targetUrl?: string;
buildInit: (requestUrl: string, request: ProxyRuntimeRequest) => Promise<UndiciRequestInit> | UndiciRequestInit;
};
export type RuntimeResponse = UndiciResponse;
export type RuntimeExecutor = {
dispatch(input: RuntimeDispatchInput): Promise<RuntimeResponse>;
};
export function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
export function withRequestBody(
request: ProxyRuntimeRequest,
body: Record<string, unknown>,
headers?: Record<string, string>,
): ProxyRuntimeRequest {
return {
...request,
headers: headers ? { ...headers } : { ...request.headers },
body,
};
}
function buildUpstreamUrl(siteUrl: string, path: string): string {
const normalizedBase = siteUrl.replace(/\/+$/, '');
const normalizedPath = path.startsWith('/') ? path : `/${path}`;
return `${normalizedBase}${normalizedPath}`;
}
export async function performFetch(
input: RuntimeDispatchInput,
request: ProxyRuntimeRequest,
requestUrl = input.targetUrl || buildUpstreamUrl(input.siteUrl, request.path),
): Promise<RuntimeResponse> {
const init = await input.buildInit(requestUrl, request);
return fetch(requestUrl, init);
}
export async function materializeErrorResponse(
response: RuntimeResponse,
): Promise<RuntimeResponse> {
if (response.ok) return response;
const text = await response.text().catch(() => '');
return new Response(text, {
status: response.status,
headers: response.headers,
});
}
@@ -0,0 +1,47 @@
import type { PreparedProviderRequest, PrepareProviderRequestInput, ProviderAction, ProviderProfile } from './types.js';
const ANTIGRAVITY_RUNTIME_USER_AGENT = 'antigravity/1.19.6 darwin/arm64';
function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function resolveAction(action: ProviderAction | undefined, stream: boolean): ProviderAction {
if (action) return action;
return stream ? 'streamGenerateContent' : 'generateContent';
}
function resolvePath(action: ProviderAction): string {
if (action === 'countTokens') return '/v1internal:countTokens';
if (action === 'streamGenerateContent') return '/v1internal:streamGenerateContent?alt=sse';
return '/v1internal:generateContent';
}
export const antigravityProviderProfile: ProviderProfile = {
id: 'antigravity',
prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest {
const action = resolveAction(input.action, input.stream);
const projectId = asTrimmedString(input.oauthProjectId);
return {
path: resolvePath(action),
headers: {
Authorization: input.baseHeaders.Authorization,
'Content-Type': 'application/json',
Accept: action === 'streamGenerateContent' ? 'text/event-stream' : 'application/json',
'User-Agent': ANTIGRAVITY_RUNTIME_USER_AGENT,
},
body: {
project: projectId,
model: input.modelName,
request: input.body,
},
runtime: {
executor: 'antigravity',
modelName: input.modelName,
stream: action === 'streamGenerateContent',
oauthProjectId: projectId,
action,
},
};
},
};
@@ -0,0 +1,128 @@
import type { PreparedProviderRequest, PrepareProviderRequestInput, ProviderProfile } from './types.js';
const CLAUDE_DEFAULT_USER_AGENT = 'claude-cli/2.1.63 (external, cli)';
const CLAUDE_DEFAULT_BETA_HEADER = 'claude-code-20250219,oauth-2025-04-20,interleaved-thinking-2025-05-14,context-management-2025-06-27,prompt-caching-scope-2026-01-05';
function headerValueToString(value: unknown): string | null {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed || null;
}
if (Array.isArray(value)) {
for (const item of value) {
if (typeof item !== 'string') continue;
const trimmed = item.trim();
if (trimmed) return trimmed;
}
}
return null;
}
function getInputHeader(
headers: Record<string, unknown> | Record<string, string> | undefined,
key: string,
): string | null {
if (!headers) return null;
for (const [candidateKey, candidateValue] of Object.entries(headers)) {
if (candidateKey.toLowerCase() !== key.toLowerCase()) continue;
return headerValueToString(candidateValue);
}
return null;
}
function mergeClaudeBetaHeader(
explicitValue: string | null,
extraBetas: string[] = [],
): string {
const source = explicitValue || CLAUDE_DEFAULT_BETA_HEADER;
const seen = new Set<string>();
const merged: string[] = [];
for (const entry of source.split(',')) {
const normalized = entry.trim();
if (!normalized || seen.has(normalized)) continue;
seen.add(normalized);
merged.push(normalized);
}
if (!explicitValue) {
for (const entry of extraBetas) {
const normalized = entry.trim();
if (!normalized || seen.has(normalized)) continue;
seen.add(normalized);
merged.push(normalized);
}
}
return merged.join(',');
}
function buildClaudeRuntimeHeaders(input: {
baseHeaders: Record<string, string>;
claudeHeaders: Record<string, string>;
anthropicVersion: string;
stream: boolean;
isClaudeOauthUpstream: boolean;
tokenValue: string;
}): Record<string, string> {
const anthropicBeta = mergeClaudeBetaHeader(
getInputHeader(input.claudeHeaders, 'anthropic-beta'),
);
const headers: Record<string, string> = {
...input.baseHeaders,
...input.claudeHeaders,
'anthropic-version': input.anthropicVersion,
...(anthropicBeta ? { 'anthropic-beta': anthropicBeta } : {}),
'Anthropic-Dangerous-Direct-Browser-Access': 'true',
'X-App': 'cli',
'X-Stainless-Retry-Count': getInputHeader(input.claudeHeaders, 'x-stainless-retry-count') || '0',
'X-Stainless-Runtime-Version': getInputHeader(input.claudeHeaders, 'x-stainless-runtime-version') || 'v24.3.0',
'X-Stainless-Package-Version': getInputHeader(input.claudeHeaders, 'x-stainless-package-version') || '0.74.0',
'X-Stainless-Runtime': getInputHeader(input.claudeHeaders, 'x-stainless-runtime') || 'node',
'X-Stainless-Lang': getInputHeader(input.claudeHeaders, 'x-stainless-lang') || 'js',
'X-Stainless-Arch': getInputHeader(input.claudeHeaders, 'x-stainless-arch') || 'x64',
'X-Stainless-Os': getInputHeader(input.claudeHeaders, 'x-stainless-os') || 'Windows',
'X-Stainless-Timeout': getInputHeader(input.claudeHeaders, 'x-stainless-timeout') || '600',
'User-Agent': getInputHeader(input.claudeHeaders, 'user-agent') || CLAUDE_DEFAULT_USER_AGENT,
Connection: 'keep-alive',
Accept: input.stream ? 'text/event-stream' : 'application/json',
'Accept-Encoding': input.stream ? 'identity' : 'gzip, deflate, br, zstd',
};
if (input.isClaudeOauthUpstream) {
headers.Authorization = `Bearer ${input.tokenValue}`;
} else {
headers['x-api-key'] = input.tokenValue;
}
return headers;
}
export const claudeProviderProfile: ProviderProfile = {
id: 'claude',
prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest {
const anthropicVersion = (
input.claudeHeaders?.['anthropic-version']
|| input.baseHeaders['anthropic-version']
|| '2023-06-01'
);
const isClaudeOauthUpstream = input.sitePlatform?.trim().toLowerCase() === 'claude'
&& input.oauthProvider === 'claude';
const isCountTokens = input.action === 'countTokens';
return {
path: isCountTokens ? '/v1/messages/count_tokens?beta=true' : '/v1/messages',
headers: buildClaudeRuntimeHeaders({
baseHeaders: input.baseHeaders,
claudeHeaders: input.claudeHeaders ?? {},
anthropicVersion,
stream: isCountTokens ? false : input.stream,
isClaudeOauthUpstream,
tokenValue: input.tokenValue,
}),
body: input.body,
runtime: {
executor: 'claude',
modelName: input.modelName,
stream: isCountTokens ? false : input.stream,
oauthProjectId: null,
...(isCountTokens ? { action: 'countTokens' } : {}),
},
};
},
};
@@ -0,0 +1,131 @@
import { createHash, randomUUID } from 'node:crypto';
import type { PreparedProviderRequest, PrepareProviderRequestInput, ProviderProfile } from './types.js';
const CODEX_CLIENT_VERSION = '0.101.0';
const CODEX_DEFAULT_USER_AGENT = 'codex_cli_rs/0.101.0 (Mac OS 26.0.1; arm64) Apple_Terminal/464';
function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function headerValueToString(value: unknown): string | null {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed || null;
}
if (Array.isArray(value)) {
for (const item of value) {
if (typeof item !== 'string') continue;
const trimmed = item.trim();
if (trimmed) return trimmed;
}
}
return null;
}
function getInputHeader(
headers: Record<string, unknown> | Record<string, string> | undefined,
key: string,
): string | null {
if (!headers) return null;
for (const [candidateKey, candidateValue] of Object.entries(headers)) {
if (candidateKey.toLowerCase() !== key.toLowerCase()) continue;
return headerValueToString(candidateValue);
}
return null;
}
function uuidFromSeed(seed: string): string {
const hash = createHash('sha1').update(seed).digest();
const bytes = new Uint8Array(hash.subarray(0, 16));
bytes[6] = (bytes[6]! & 0x0f) | 0x50;
bytes[8] = (bytes[8]! & 0x3f) | 0x80;
const hex = Array.from(bytes, (value) => value.toString(16).padStart(2, '0')).join('');
return [
hex.slice(0, 8),
hex.slice(8, 12),
hex.slice(12, 16),
hex.slice(16, 20),
hex.slice(20, 32),
].join('-');
}
function buildCodexRuntimeHeaders(input: {
baseHeaders: Record<string, string>;
providerHeaders?: Record<string, string>;
explicitSessionId?: string | null;
continuityKey?: string | null;
}): Record<string, string> {
const authorization = (
getInputHeader(input.baseHeaders, 'authorization')
|| getInputHeader(input.baseHeaders, 'Authorization')
|| ''
);
const originator = getInputHeader(input.providerHeaders, 'originator') || 'codex_cli_rs';
const accountId = getInputHeader(input.providerHeaders, 'chatgpt-account-id');
const version = getInputHeader(input.baseHeaders, 'version') || CODEX_CLIENT_VERSION;
const userAgent = getInputHeader(input.baseHeaders, 'user-agent') || CODEX_DEFAULT_USER_AGENT;
const explicitSessionId = asTrimmedString(input.explicitSessionId);
const continuityKey = asTrimmedString(input.continuityKey);
const sessionId = (
getInputHeader(input.baseHeaders, 'session_id')
|| getInputHeader(input.baseHeaders, 'session-id')
|| explicitSessionId
|| (continuityKey ? uuidFromSeed(`metapi:codex:${continuityKey}`) : null)
|| randomUUID()
);
const conversationId = (
getInputHeader(input.baseHeaders, 'conversation_id')
|| getInputHeader(input.baseHeaders, 'conversation-id')
|| explicitSessionId
|| (continuityKey ? sessionId : null)
);
return {
Authorization: authorization,
'Content-Type': 'application/json',
...(accountId ? { 'Chatgpt-Account-Id': accountId } : {}),
Originator: originator,
Version: version,
Session_id: sessionId,
...(conversationId ? { Conversation_id: conversationId } : {}),
'User-Agent': userAgent,
Accept: 'text/event-stream',
Connection: 'Keep-Alive',
};
}
export const codexProviderProfile: ProviderProfile = {
id: 'codex',
prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest {
const headers = buildCodexRuntimeHeaders({
baseHeaders: input.baseHeaders,
providerHeaders: input.providerHeaders,
explicitSessionId: asTrimmedString(input.codexExplicitSessionId) || null,
continuityKey: asTrimmedString(input.codexSessionCacheKey) || null,
});
const codexSessionId = getInputHeader(headers, 'session_id') || getInputHeader(headers, 'session-id');
const shouldInjectDerivedPromptCacheKey = !!codexSessionId
&& !asTrimmedString(input.body.prompt_cache_key)
&& !asTrimmedString(input.codexExplicitSessionId)
&& !!asTrimmedString(input.codexSessionCacheKey);
const body = shouldInjectDerivedPromptCacheKey
? {
...input.body,
prompt_cache_key: codexSessionId,
}
: input.body;
return {
path: '/responses',
headers,
body,
runtime: {
executor: 'codex',
modelName: input.modelName,
stream: input.stream,
oauthProjectId: asTrimmedString(input.oauthProjectId) || null,
},
};
},
};
@@ -0,0 +1,113 @@
import type { PreparedProviderRequest, PrepareProviderRequestInput, ProviderAction, ProviderProfile } from './types.js';
function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function headerValueToString(value: unknown): string | null {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed || null;
}
if (Array.isArray(value)) {
for (const item of value) {
if (typeof item !== 'string') continue;
const trimmed = item.trim();
if (trimmed) return trimmed;
}
}
return null;
}
function getInputHeader(
headers: Record<string, unknown> | Record<string, string> | undefined,
key: string,
): string | null {
if (!headers) return null;
for (const [candidateKey, candidateValue] of Object.entries(headers)) {
if (candidateKey.toLowerCase() !== key.toLowerCase()) continue;
return headerValueToString(candidateValue);
}
return null;
}
function parseGeminiCliUserAgentRuntime(userAgent: string | null): {
version: string;
platform: string;
arch: string;
} | null {
if (!userAgent) return null;
const match = /^GeminiCLI\/([^/]+)\/[^ ]+ \(([^;]+); ([^)]+)\)$/i.exec(userAgent.trim());
if (!match) return null;
return {
version: match[1] || '0.31.0',
platform: match[2] || 'win32',
arch: match[3] || 'x64',
};
}
function buildGeminiCLIUserAgent(modelName: string, existingUserAgent?: string | null): string {
const parsed = parseGeminiCliUserAgentRuntime(existingUserAgent ?? null);
const version = parsed?.version || '0.31.0';
const platform = parsed?.platform || 'win32';
const arch = parsed?.arch || 'x64';
const effectiveModel = asTrimmedString(modelName) || 'unknown';
return `GeminiCLI/${version}/${effectiveModel} (${platform}; ${arch})`;
}
function resolveAction(action: ProviderAction | undefined, stream: boolean): ProviderAction {
if (action) return action;
return stream ? 'streamGenerateContent' : 'generateContent';
}
function resolvePath(action: ProviderAction): string {
if (action === 'countTokens') return '/v1internal:countTokens';
if (action === 'streamGenerateContent') return '/v1internal:streamGenerateContent?alt=sse';
return '/v1internal:generateContent';
}
export const geminiCliProviderProfile: ProviderProfile = {
id: 'gemini-cli',
prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest {
const projectId = asTrimmedString(input.oauthProjectId);
if (!projectId) {
throw new Error('gemini-cli oauth project id missing');
}
const action = resolveAction(input.action, input.stream);
const apiClient = (
getInputHeader(input.providerHeaders, 'x-goog-api-client')
|| getInputHeader(input.baseHeaders, 'x-goog-api-client')
);
const userAgent = buildGeminiCLIUserAgent(
input.modelName,
getInputHeader(input.providerHeaders, 'user-agent') || getInputHeader(input.baseHeaders, 'user-agent'),
);
const headers: Record<string, string> = {
Authorization: input.baseHeaders.Authorization,
'Content-Type': 'application/json',
'User-Agent': userAgent,
};
if (apiClient) {
headers['X-Goog-Api-Client'] = apiClient;
}
if (action === 'streamGenerateContent') {
headers.Accept = 'text/event-stream';
}
return {
path: resolvePath(action),
headers,
body: {
project: projectId,
model: input.modelName,
request: input.body,
},
runtime: {
executor: 'gemini-cli',
modelName: input.modelName,
stream: action === 'streamGenerateContent',
oauthProjectId: projectId,
action,
},
};
},
};
@@ -0,0 +1,181 @@
import { describe, expect, it } from 'vitest';
import { resolveProviderProfile } from './registry.js';
describe('resolveProviderProfile', () => {
it('builds codex provider requests with codex-specific path, headers, and runtime metadata', () => {
const profile = resolveProviderProfile('codex');
expect(profile?.id).toBe('codex');
const result = profile!.prepareRequest({
endpoint: 'responses',
modelName: 'gpt-5.2-codex',
stream: true,
tokenValue: 'oauth-access-token',
sitePlatform: 'codex',
baseHeaders: {
Authorization: 'Bearer oauth-access-token',
},
providerHeaders: {
Originator: 'codex_cli_rs',
'Chatgpt-Account-Id': 'chatgpt-account-123',
},
codexSessionCacheKey: 'gpt-5.2-codex:user-123',
body: {
model: 'gpt-5.2-codex',
stream: true,
store: false,
},
});
expect(result.path).toBe('/responses');
expect(result.headers.Authorization).toBe('Bearer oauth-access-token');
expect(result.headers.Originator).toBe('codex_cli_rs');
expect(result.headers['Chatgpt-Account-Id']).toBe('chatgpt-account-123');
expect(result.headers.Session_id).toMatch(/^[0-9a-f-]{36}$/i);
expect(result.headers.Conversation_id).toBe(result.headers.Session_id);
expect(result.runtime).toMatchObject({
executor: 'codex',
modelName: 'gpt-5.2-codex',
stream: true,
});
expect(result.body.prompt_cache_key).toBe(result.headers.Session_id);
});
it('builds claude provider requests without rebuilding protocol bodies', () => {
const profile = resolveProviderProfile('claude');
expect(profile?.id).toBe('claude');
const protocolBody = {
model: 'claude-opus-4-6',
max_tokens: 256,
messages: [{ role: 'user', content: 'hello' }],
};
const result = profile!.prepareRequest({
endpoint: 'messages',
modelName: 'claude-opus-4-6',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'claude',
sitePlatform: 'claude',
baseHeaders: {
'Content-Type': 'application/json',
},
claudeHeaders: {},
body: protocolBody,
});
expect(result.path).toBe('/v1/messages');
expect(result.headers.Authorization).toBe('Bearer oauth-access-token');
expect(result.headers['x-api-key']).toBeUndefined();
expect(result.headers['anthropic-version']).toBe('2023-06-01');
expect(result.runtime).toMatchObject({
executor: 'claude',
modelName: 'claude-opus-4-6',
stream: false,
});
expect(result.body).toBe(protocolBody);
});
it('builds gemini-cli provider requests with wrapped runtime envelope and project validation', () => {
const profile = resolveProviderProfile('gemini-cli');
expect(profile?.id).toBe('gemini-cli');
const protocolBody = {
contents: [{ role: 'user', parts: [{ text: 'hello gemini cli' }] }],
};
const result = profile!.prepareRequest({
endpoint: 'chat',
modelName: 'gemini-2.5-pro',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'gemini-cli',
oauthProjectId: 'project-demo',
sitePlatform: 'gemini-cli',
baseHeaders: {
Authorization: 'Bearer oauth-access-token',
},
providerHeaders: {
'User-Agent': 'GeminiCLI/0.31.0/unknown (win32; x64)',
'X-Goog-Api-Client': 'google-genai-sdk/1.41.0 gl-node/v22.19.0',
},
body: protocolBody,
});
expect(result.path).toBe('/v1internal:generateContent');
expect(result.headers.Authorization).toBe('Bearer oauth-access-token');
expect(result.headers['User-Agent']).toBe('GeminiCLI/0.31.0/gemini-2.5-pro (win32; x64)');
expect(result.headers['X-Goog-Api-Client']).toContain('google-genai-sdk/');
expect(result.runtime).toMatchObject({
executor: 'gemini-cli',
modelName: 'gemini-2.5-pro',
oauthProjectId: 'project-demo',
action: 'generateContent',
});
expect(result.body).toEqual({
project: 'project-demo',
model: 'gemini-2.5-pro',
request: protocolBody,
});
expect((result.body as Record<string, unknown>).request).toBe(protocolBody);
expect(() => profile!.prepareRequest({
endpoint: 'chat',
modelName: 'gemini-2.5-pro',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'gemini-cli',
sitePlatform: 'gemini-cli',
baseHeaders: {
Authorization: 'Bearer oauth-access-token',
},
body: protocolBody,
})).toThrow(/project id missing/i);
});
it('builds antigravity provider requests without leaking gemini-cli client headers', () => {
const profile = resolveProviderProfile('antigravity');
expect(profile?.id).toBe('antigravity');
const protocolBody = {
contents: [{ role: 'user', parts: [{ text: 'hello antigravity' }] }],
};
const result = profile!.prepareRequest({
endpoint: 'chat',
modelName: 'gemini-3-pro-preview',
stream: true,
tokenValue: 'oauth-access-token',
oauthProjectId: 'project-demo',
sitePlatform: 'antigravity',
baseHeaders: {
Authorization: 'Bearer oauth-access-token',
},
providerHeaders: {
'User-Agent': 'GeminiCLI/0.31.0/unknown (win32; x64)',
'X-Goog-Api-Client': 'google-genai-sdk/1.41.0 gl-node/v22.19.0',
},
body: protocolBody,
});
expect(result.path).toBe('/v1internal:streamGenerateContent?alt=sse');
expect(result.headers.Authorization).toBe('Bearer oauth-access-token');
expect(result.headers['User-Agent']).toBe('antigravity/1.19.6 darwin/arm64');
expect(result.headers['X-Goog-Api-Client']).toBeUndefined();
expect(result.headers.Accept).toBe('text/event-stream');
expect(result.runtime).toMatchObject({
executor: 'antigravity',
modelName: 'gemini-3-pro-preview',
oauthProjectId: 'project-demo',
action: 'streamGenerateContent',
});
expect(result.body).toEqual({
project: 'project-demo',
model: 'gemini-3-pro-preview',
request: protocolBody,
});
expect((result.body as Record<string, unknown>).request).toBe(protocolBody);
});
});
@@ -0,0 +1,17 @@
import { antigravityProviderProfile } from './antigravityProviderProfile.js';
import { claudeProviderProfile } from './claudeProviderProfile.js';
import { codexProviderProfile } from './codexProviderProfile.js';
import { geminiCliProviderProfile } from './geminiCliProviderProfile.js';
import type { ProviderProfile } from './types.js';
const providerProfilesByPlatform: Record<string, ProviderProfile> = {
codex: codexProviderProfile,
claude: claudeProviderProfile,
'gemini-cli': geminiCliProviderProfile,
antigravity: antigravityProviderProfile,
};
export function resolveProviderProfile(sitePlatform?: string | null): ProviderProfile | null {
const normalized = typeof sitePlatform === 'string' ? sitePlatform.trim().toLowerCase() : '';
return providerProfilesByPlatform[normalized] ?? null;
}
+52
View File
@@ -0,0 +1,52 @@
export type ProviderProfileId =
| 'codex'
| 'claude'
| 'gemini-cli'
| 'antigravity';
export type ProviderEndpoint =
| 'chat'
| 'messages'
| 'responses';
export type ProviderAction =
| 'generateContent'
| 'streamGenerateContent'
| 'countTokens';
export type ProviderRuntimeDescriptor = {
executor: 'default' | 'codex' | 'gemini-cli' | 'antigravity' | 'claude';
modelName?: string;
stream?: boolean;
oauthProjectId?: string | null;
action?: ProviderAction;
};
export type PreparedProviderRequest = {
path: string;
headers: Record<string, string>;
body: Record<string, unknown>;
runtime: ProviderRuntimeDescriptor;
};
export type PrepareProviderRequestInput = {
endpoint: ProviderEndpoint;
modelName: string;
stream: boolean;
tokenValue: string;
oauthProvider?: string;
oauthProjectId?: string;
sitePlatform?: string;
baseHeaders: Record<string, string>;
providerHeaders?: Record<string, string>;
claudeHeaders?: Record<string, string>;
codexSessionCacheKey?: string | null;
codexExplicitSessionId?: string | null;
body: Record<string, unknown>;
action?: ProviderAction;
};
export type ProviderProfile = {
id: ProviderProfileId;
prepareRequest(input: PrepareProviderRequestInput): PreparedProviderRequest;
};
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,159 @@
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { getProxyResourceOwner } from '../../middleware/auth.js';
import {
getProxyFileByPublicIdForOwner,
getProxyFileContentByPublicIdForOwner,
listProxyFilesByOwner,
saveProxyFile,
softDeleteProxyFileByPublicIdForOwner,
} from '../../services/proxyFileStore.js';
import { ensureMultipartBufferParser, parseMultipartFormData } from '../../routes/proxy/multipart.js';
function invalidRequest(reply: FastifyReply, message: string) {
return reply.code(400).send({ error: { message, type: 'invalid_request_error' } });
}
function notFound(reply: FastifyReply, message = 'file not found') {
return reply.code(404).send({ error: { message, type: 'not_found_error' } });
}
function toUnixSeconds(sqlDateTime: string | null | undefined): number {
if (!sqlDateTime) return Math.floor(Date.now() / 1000);
const parsed = Date.parse(sqlDateTime.replace(' ', 'T') + 'Z');
if (!Number.isFinite(parsed)) return Math.floor(Date.now() / 1000);
return Math.floor(parsed / 1000);
}
function toFileObject(record: {
publicId: string;
filename: string;
mimeType: string;
purpose: string | null;
byteSize: number;
createdAt: string | null;
}) {
return {
id: record.publicId,
object: 'file',
bytes: record.byteSize,
created_at: toUnixSeconds(record.createdAt),
filename: record.filename,
purpose: record.purpose || 'assistants',
mime_type: record.mimeType,
};
}
function inferMimeTypeFromFilename(filename: string): string {
const normalized = filename.trim().toLowerCase();
if (normalized.endsWith('.pdf')) return 'application/pdf';
if (normalized.endsWith('.txt')) return 'text/plain';
if (normalized.endsWith('.md') || normalized.endsWith('.markdown')) return 'text/markdown';
if (normalized.endsWith('.json')) return 'application/json';
if (normalized.endsWith('.png')) return 'image/png';
if (normalized.endsWith('.jpg') || normalized.endsWith('.jpeg')) return 'image/jpeg';
if (normalized.endsWith('.gif')) return 'image/gif';
if (normalized.endsWith('.webp')) return 'image/webp';
if (normalized.endsWith('.wav')) return 'audio/wav';
if (normalized.endsWith('.mp3')) return 'audio/mpeg';
return 'application/octet-stream';
}
function isSupportedProxyFileMimeType(mimeType: string): boolean {
return mimeType === 'application/pdf'
|| mimeType === 'text/plain'
|| mimeType === 'text/markdown'
|| mimeType === 'application/json'
|| mimeType.startsWith('image/')
|| mimeType.startsWith('audio/');
}
export async function filesProxyRoute(app: FastifyInstance) {
ensureMultipartBufferParser(app);
app.post('/v1/files', async (request: FastifyRequest, reply: FastifyReply) => {
const owner = getProxyResourceOwner(request);
if (!owner) {
return reply.code(401).send({ error: { message: 'Missing proxy auth context', type: 'authentication_error' } });
}
const formData = await parseMultipartFormData(request);
if (!formData) {
return invalidRequest(reply, 'multipart/form-data with a file field is required');
}
const fileEntry = formData.get('file');
if (!fileEntry || typeof fileEntry !== 'object' || typeof (fileEntry as File).arrayBuffer !== 'function') {
return invalidRequest(reply, 'file field is required');
}
const filename = fileEntry.name || 'upload.bin';
const mimeType = (fileEntry.type || inferMimeTypeFromFilename(filename)).trim().toLowerCase();
if (!isSupportedProxyFileMimeType(mimeType)) {
return invalidRequest(reply, `unsupported file mime type: ${mimeType || 'application/octet-stream'}`);
}
const purposeValue = formData.get('purpose');
const purpose = typeof purposeValue === 'string' && purposeValue.trim().length > 0
? purposeValue.trim()
: 'assistants';
const buffer = Buffer.from(await fileEntry.arrayBuffer());
const saved = await saveProxyFile({
...owner,
filename,
mimeType,
purpose,
contentBase64: buffer.toString('base64'),
});
return reply.send(toFileObject(saved));
});
app.get('/v1/files', async (request: FastifyRequest, reply: FastifyReply) => {
const owner = getProxyResourceOwner(request);
if (!owner) {
return reply.code(401).send({ error: { message: 'Missing proxy auth context', type: 'authentication_error' } });
}
const files = await listProxyFilesByOwner(owner);
return reply.send({
object: 'list',
data: files.map((item) => toFileObject(item)),
has_more: false,
});
});
app.get<{ Params: { fileId: string } }>('/v1/files/:fileId', async (request, reply) => {
const owner = getProxyResourceOwner(request);
if (!owner) {
return reply.code(401).send({ error: { message: 'Missing proxy auth context', type: 'authentication_error' } });
}
const file = await getProxyFileByPublicIdForOwner(request.params.fileId, owner);
if (!file) return notFound(reply);
return reply.send(toFileObject(file));
});
app.get<{ Params: { fileId: string } }>('/v1/files/:fileId/content', async (request, reply) => {
const owner = getProxyResourceOwner(request);
if (!owner) {
return reply.code(401).send({ error: { message: 'Missing proxy auth context', type: 'authentication_error' } });
}
const file = await getProxyFileContentByPublicIdForOwner(request.params.fileId, owner);
if (!file) return notFound(reply);
reply.type(file.mimeType);
reply.header('Content-Disposition', `inline; filename="${encodeURIComponent(file.filename)}"`);
return reply.send(file.buffer);
});
app.delete<{ Params: { fileId: string } }>('/v1/files/:fileId', async (request, reply) => {
const owner = getProxyResourceOwner(request);
if (!owner) {
return reply.code(401).send({ error: { message: 'Missing proxy auth context', type: 'authentication_error' } });
}
const deleted = await softDeleteProxyFileByPublicIdForOwner(request.params.fileId, owner);
if (!deleted) return notFound(reply);
return reply.send({
id: request.params.fileId,
object: 'file',
deleted: true,
});
});
}
@@ -0,0 +1,843 @@
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
import { TextDecoder } from 'node:util';
import { fetch } from 'undici';
import { and, eq } from 'drizzle-orm';
import { db, schema } from '../../db/index.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { parseProxyUsage } from '../../services/proxyUsageParser.js';
import { isModelAllowedByPolicyOrAllowedRoutes } from '../../services/downstreamApiKeyService.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { buildOauthProviderHeaders } from '../../services/oauth/service.js';
import { getOauthInfoFromExtraConfig } from '../../services/oauth/oauthAccount.js';
import { withSiteProxyRequestInit } from '../../services/siteProxy.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { getDownstreamRoutingPolicy } from '../../routes/proxy/downstreamPolicy.js';
import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js';
import { composeProxyLogMessage } from '../../routes/proxy/logPathMeta.js';
import {
buildUpstreamEndpointRequest,
resolveUpstreamEndpointCandidates,
} from '../../routes/proxy/upstreamEndpoint.js';
import {
geminiGenerateContentTransformer,
} from '../../transformers/gemini/generate-content/index.js';
import { createChatEndpointStrategy } from '../../transformers/shared/chatEndpointStrategy.js';
import { normalizeUpstreamFinalResponse } from '../../transformers/shared/normalized.js';
import {
createGeminiCliStreamReader,
unwrapGeminiCliPayload,
wrapGeminiCliRequest,
} from '../../routes/proxy/geminiCliCompat.js';
import { dispatchRuntimeRequest } from '../../routes/proxy/runtimeExecutor.js';
const MAX_RETRIES = 2;
const GEMINI_MODEL_PROBES = [
'gemini-2.5-flash',
'gemini-2.0-flash',
'gemini-1.5-flash',
'gemini-pro',
];
const GEMINI_CLI_STATIC_MODELS = [
{ name: 'models/gemini-2.5-pro', displayName: 'Gemini 2.5 Pro' },
{ name: 'models/gemini-2.5-flash', displayName: 'Gemini 2.5 Flash' },
{ name: 'models/gemini-2.5-flash-lite', displayName: 'Gemini 2.5 Flash Lite' },
{ name: 'models/gemini-3-pro-preview', displayName: 'Gemini 3 Pro Preview' },
{ name: 'models/gemini-3.1-pro-preview', displayName: 'Gemini 3.1 Pro Preview' },
{ name: 'models/gemini-3-flash-preview', displayName: 'Gemini 3 Flash Preview' },
{ name: 'models/gemini-3.1-flash-lite-preview', displayName: 'Gemini 3.1 Flash Lite Preview' },
];
const EMPTY_PROXY_USAGE = {
promptTokens: 0,
completionTokens: 0,
totalTokens: 0,
};
function isGeminiCliPlatform(platform: unknown): boolean {
return String(platform || '').trim().toLowerCase() === 'gemini-cli';
}
function isAntigravityPlatform(platform: unknown): boolean {
return String(platform || '').trim().toLowerCase() === 'antigravity';
}
function isInternalGeminiPlatform(platform: unknown): boolean {
return isGeminiCliPlatform(platform) || isAntigravityPlatform(platform);
}
function buildGeminiCliActionPath(input: {
isStreamAction: boolean;
isCountTokensAction: boolean;
}) {
if (input.isCountTokensAction) return '/v1internal:countTokens';
if (input.isStreamAction) return '/v1internal:streamGenerateContent?alt=sse';
return '/v1internal:generateContent';
}
function isDirectGeminiFamilyPlatform(platform: unknown): boolean {
const normalized = String(platform || '').trim().toLowerCase();
return normalized === 'gemini' || normalized === 'gemini-cli' || normalized === 'antigravity';
}
function omitGeminiCliModelField(body: unknown): Record<string, unknown> {
if (!body || typeof body !== 'object' || Array.isArray(body)) return {};
const { model: _model, ...rest } = body as Record<string, unknown>;
return rest;
}
async function selectGeminiChannel(request: FastifyRequest) {
const policy = getDownstreamRoutingPolicy(request);
for (const candidate of GEMINI_MODEL_PROBES) {
const selected = await tokenRouter.selectChannel(candidate, policy);
if (selected) return selected;
}
return null;
}
async function selectNextGeminiProbeChannel(request: FastifyRequest, excludeChannelIds: number[]) {
const policy = getDownstreamRoutingPolicy(request);
for (const candidate of GEMINI_MODEL_PROBES) {
const selected = await tokenRouter.selectNextChannel(candidate, excludeChannelIds, policy);
if (selected) return selected;
}
return null;
}
function resolveDownstreamPath(request: FastifyRequest): string {
const rawUrl = request.raw.url || request.url || '';
const withoutQuery = rawUrl.split('?')[0] || rawUrl;
return withoutQuery || '/v1beta/models';
}
function resolveUpstreamPath(apiVersion: string, modelActionPath: string): string {
const normalizedVersion = apiVersion.replace(/^\/+/, '');
const normalizedAction = modelActionPath.replace(/^\/+/, '');
return `/${normalizedVersion}/${normalizedAction}`;
}
function hasDownstreamModelRestrictions(policy: { supportedModels?: unknown; allowedRouteIds?: unknown }): boolean {
const supportedModels = Array.isArray(policy.supportedModels) ? policy.supportedModels : [];
const allowedRouteIds = Array.isArray(policy.allowedRouteIds) ? policy.allowedRouteIds : [];
return supportedModels.length > 0 || allowedRouteIds.length > 0;
}
function extractGeminiListedModelName(item: unknown): string {
if (!item || typeof item !== 'object') return '';
const rawName = typeof (item as { name?: unknown }).name === 'string'
? (item as { name: string }).name.trim()
: '';
if (!rawName) return '';
return rawName.startsWith('models/') ? rawName.slice('models/'.length) : rawName;
}
async function filterGeminiListedModelsForPolicy(
payload: unknown,
request: FastifyRequest,
): Promise<unknown> {
if (!payload || typeof payload !== 'object' || !Array.isArray((payload as { models?: unknown[] }).models)) {
return payload;
}
const policy = getDownstreamRoutingPolicy(request);
if (!hasDownstreamModelRestrictions(policy)) {
return payload;
}
const filteredModels: unknown[] = [];
for (const item of (payload as { models: unknown[] }).models) {
const modelName = extractGeminiListedModelName(item);
if (!modelName) continue;
if (!await isModelAllowedByPolicyOrAllowedRoutes(modelName, policy)) continue;
const decision = await tokenRouter.explainSelection?.(modelName, [], policy);
if (decision && typeof decision.selectedChannelId !== 'number') continue;
filteredModels.push(item);
}
return {
...(payload as Record<string, unknown>),
models: filteredModels,
};
}
async function readRouteAwareGeminiModels(request: FastifyRequest): Promise<Array<{ name: string; displayName: string }>> {
const policy = getDownstreamRoutingPolicy(request);
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();
const deduped = Array.from(new Set([
...rows.map((row) => String(row.modelName || '').trim()).filter(Boolean),
...routeAliases.map((row) => String(row.displayName || '').trim()).filter(Boolean),
])).sort();
const allowed: Array<{ name: string; displayName: string }> = [];
for (const modelName of deduped) {
if (!await isModelAllowedByPolicyOrAllowedRoutes(modelName, policy)) continue;
const decision = await tokenRouter.explainSelection?.(modelName, [], policy);
if (decision && typeof decision.selectedChannelId !== 'number') continue;
allowed.push({
name: `models/${modelName}`,
displayName: modelName,
});
}
return allowed;
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamPath: string,
upstreamPath: string | null,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
downstreamPath,
upstreamPath,
errorMessage,
});
await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
modelRequested,
modelActual: selected.actualModel || modelRequested,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost: 0,
billingDetails: null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
}).run();
} catch (error) {
console.warn('[proxy/gemini] failed to write proxy log', error);
}
}
export async function geminiProxyRoute(app: FastifyInstance) {
const listModels = async (request: FastifyRequest, reply: FastifyReply) => {
const apiVersion = geminiGenerateContentTransformer.resolveProxyApiVersion(
request.params as { geminiApiVersion?: string } | undefined,
);
const excludeChannelIds: number[] = [];
let retryCount = 0;
let lastStatus = 503;
let lastText = 'No available channels for Gemini models';
let lastContentType = 'application/json';
while (retryCount <= MAX_RETRIES) {
const selected = retryCount === 0
? await selectGeminiChannel(request)
: await selectNextGeminiProbeChannel(request, excludeChannelIds);
if (!selected) {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
excludeChannelIds.push(selected.channel.id);
try {
if (!isDirectGeminiFamilyPlatform(selected.site.platform)) {
let models = await readRouteAwareGeminiModels(request);
if (models.length <= 0) {
await refreshModelsAndRebuildRoutes();
models = await readRouteAwareGeminiModels(request);
}
return reply.code(200).send({ models });
}
if (isGeminiCliPlatform(selected.site.platform)) {
const filtered = await filterGeminiListedModelsForPolicy(
{ models: GEMINI_CLI_STATIC_MODELS },
request,
);
return reply.code(200).send(filtered);
}
const upstream = await fetch(
geminiGenerateContentTransformer.resolveModelsUrl(selected.site.url, apiVersion, selected.tokenValue),
{ method: 'GET' },
);
const text = await upstream.text();
if (!upstream.ok) {
lastStatus = upstream.status;
lastText = text;
lastContentType = upstream.headers.get('content-type') || 'application/json';
await tokenRouter.recordFailure?.(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
}
try {
const parsed = JSON.parse(text);
const filtered = await filterGeminiListedModelsForPolicy(parsed, request);
return reply.code(upstream.status).send(filtered);
} catch {
return reply.code(upstream.status).type(upstream.headers.get('content-type') || 'application/json').send(text);
}
} catch (error) {
await tokenRouter.recordFailure?.(selected.channel.id);
lastStatus = 502;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: error instanceof Error ? error.message : 'Gemini upstream request failed',
type: 'upstream_error',
},
});
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
}
}
};
const handleGenerateContent = async (
request: FastifyRequest,
reply: FastifyReply,
options?: {
downstreamProtocol?: 'gemini' | 'gemini-cli';
action?: 'generateContent' | 'streamGenerateContent' | 'countTokens';
},
) => {
const downstreamProtocol = options?.downstreamProtocol || 'gemini';
const isGeminiCliDownstream = downstreamProtocol === 'gemini-cli';
const cliRequestedModel = isGeminiCliDownstream
? (typeof (request.body as Record<string, unknown> | null | undefined)?.model === 'string'
? String((request.body as Record<string, unknown>).model).trim()
: '')
: '';
const parsedPath = isGeminiCliDownstream
? {
apiVersion: 'v1beta',
modelActionPath: `models/${cliRequestedModel}:${options?.action || 'generateContent'}`,
isStreamAction: options?.action === 'streamGenerateContent',
requestedModel: cliRequestedModel,
}
: geminiGenerateContentTransformer.parseProxyRequestPath({
rawUrl: request.raw.url || request.url || '',
params: request.params as { geminiApiVersion?: string } | undefined,
});
const { apiVersion, modelActionPath, isStreamAction, requestedModel } = parsedPath;
const isCountTokensAction = isGeminiCliDownstream
? options?.action === 'countTokens'
: modelActionPath.endsWith(':countTokens');
if (!requestedModel) {
return reply.code(400).send({
error: { message: 'Gemini model path is required', type: 'invalid_request_error' },
});
}
const policy = getDownstreamRoutingPolicy(request);
const downstreamPath = resolveDownstreamPath(request);
const excludeChannelIds: number[] = [];
let retryCount = 0;
let lastStatus = 503;
let lastText = 'No available channels for this model';
let lastContentType = 'application/json';
while (retryCount <= MAX_RETRIES) {
const selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, policy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, policy);
if (!selected) {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
excludeChannelIds.push(selected.channel.id);
const actualModel = selected.actualModel || requestedModel;
const normalizedBody = geminiGenerateContentTransformer.inbound.normalizeRequest(
isGeminiCliDownstream ? omitGeminiCliModelField(request.body) : (request.body || {}),
actualModel,
);
const oauth = getOauthInfoFromExtraConfig(selected.account.extraConfig);
const isGeminiCli = isGeminiCliPlatform(selected.site.platform);
const isInternalGemini = isInternalGeminiPlatform(selected.site.platform);
const isDirectGeminiFamily = isDirectGeminiFamilyPlatform(selected.site.platform);
const startTime = Date.now();
let upstreamPath = '';
try {
if (isDirectGeminiFamily) {
if (isGeminiCli && !oauth?.projectId) {
lastStatus = 500;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: 'Gemini CLI OAuth project is missing',
type: 'server_error',
},
});
await tokenRouter.recordFailure?.(selected.channel.id);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
const actualModelAction = modelActionPath.replace(
/^models\/[^:]+/,
`models/${actualModel}`,
);
upstreamPath = isInternalGemini
? buildGeminiCliActionPath({ isStreamAction, isCountTokensAction })
: resolveUpstreamPath(apiVersion, actualModelAction);
const query = new URLSearchParams(request.query as Record<string, string>).toString();
const requestBody = isInternalGemini
? (
isCountTokensAction
? { request: normalizedBody }
: wrapGeminiCliRequest({
modelName: actualModel,
projectId: oauth?.projectId || '',
request: normalizedBody as Record<string, unknown>,
})
)
: normalizedBody;
const requestHeaders = isInternalGemini
? {
'Content-Type': 'application/json',
...(isStreamAction ? { Accept: 'text/event-stream' } : {}),
Authorization: `Bearer ${selected.tokenValue}`,
...buildOauthProviderHeaders({
extraConfig: typeof selected.account.extraConfig === 'string' ? selected.account.extraConfig : null,
downstreamHeaders: request.headers as Record<string, unknown>,
}),
}
: {
'Content-Type': 'application/json',
};
const targetUrl = isInternalGemini
? `${selected.site.url}${upstreamPath}`
: geminiGenerateContentTransformer.resolveActionUrl(
selected.site.url,
apiVersion,
actualModelAction,
selected.tokenValue,
query,
);
const upstream = isInternalGemini
? await dispatchRuntimeRequest({
siteUrl: selected.site.url,
targetUrl,
request: {
endpoint: 'chat',
path: upstreamPath,
headers: requestHeaders,
body: requestBody as Record<string, unknown>,
runtime: {
executor: isGeminiCli ? 'gemini-cli' : 'antigravity',
modelName: actualModel,
stream: isStreamAction,
oauthProjectId: oauth?.projectId || null,
action: isCountTokensAction
? 'countTokens'
: (isStreamAction ? 'streamGenerateContent' : 'generateContent'),
},
},
buildInit: async (requestUrl, requestForFetch) => withSiteProxyRequestInit(requestUrl, {
method: 'POST',
headers: requestForFetch.headers,
body: JSON.stringify(requestForFetch.body),
}),
})
: await fetch(targetUrl, {
method: 'POST',
headers: requestHeaders,
body: JSON.stringify(requestBody),
});
const contentType = upstream.headers.get('content-type') || 'application/json';
if (!upstream.ok) {
lastStatus = upstream.status;
lastContentType = contentType;
lastText = await upstream.text();
await tokenRouter.recordFailure?.(selected.channel.id);
await logProxy(
selected,
requestedModel,
'failed',
lastStatus,
Date.now() - startTime,
lastText,
retryCount,
downstreamPath,
upstreamPath,
);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
try {
return reply.code(lastStatus).send(JSON.parse(lastText));
} catch {
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
}
if (geminiGenerateContentTransformer.stream.isSseContentType(contentType)) {
reply.hijack();
reply.raw.statusCode = upstream.status;
reply.raw.setHeader('Content-Type', contentType || 'text/event-stream');
const upstreamReader = upstream.body?.getReader();
const reader = isInternalGemini && !isGeminiCliDownstream && upstreamReader
? createGeminiCliStreamReader(upstreamReader)
: upstreamReader;
if (!reader) {
const latency = Date.now() - startTime;
await tokenRouter.recordSuccess?.(selected.channel.id, latency, 0);
await logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamPath,
upstreamPath,
);
reply.raw.end();
return;
}
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
const decoder = new TextDecoder();
let rest = '';
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
if (!value) continue;
const chunkText = decoder.decode(value, { stream: true });
const consumed = geminiGenerateContentTransformer.stream.consumeUpstreamSseBuffer(
aggregateState,
rest + chunkText,
);
rest = consumed.rest;
for (const line of consumed.lines) {
reply.raw.write(line);
}
}
const tail = decoder.decode();
if (tail) {
const consumed = geminiGenerateContentTransformer.stream.consumeUpstreamSseBuffer(
aggregateState,
rest + tail,
);
for (const line of consumed.lines) {
reply.raw.write(line);
}
}
} finally {
reader.releaseLock();
reply.raw.end();
}
const parsedUsage = parseProxyUsage(aggregateState);
const latency = Date.now() - startTime;
await tokenRouter.recordSuccess?.(selected.channel.id, latency, 0);
await logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamPath,
upstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
);
return;
}
const text = await upstream.text();
const aggregateState = geminiGenerateContentTransformer.stream.createAggregateState();
let parsedUsage = EMPTY_PROXY_USAGE;
try {
const parsed = JSON.parse(text);
const unwrappedPayload = isInternalGemini && !isGeminiCliDownstream
? unwrapGeminiCliPayload(parsed)
: parsed;
const responsePayload = isCountTokensAction
? unwrappedPayload
: geminiGenerateContentTransformer.stream.serializeUpstreamJsonPayload(
aggregateState,
unwrappedPayload,
isStreamAction,
);
parsedUsage = parseProxyUsage(aggregateState);
const latency = Date.now() - startTime;
await tokenRouter.recordSuccess?.(selected.channel.id, latency, 0);
await logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamPath,
upstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
);
return reply.code(upstream.status).send(
isGeminiCliDownstream && !isCountTokensAction
? { response: responsePayload }
: responsePayload,
);
} catch {
const latency = Date.now() - startTime;
await tokenRouter.recordSuccess?.(selected.channel.id, latency, 0);
await logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamPath,
upstreamPath,
);
return reply.code(upstream.status).type(contentType || 'application/json').send(text);
}
}
if (isCountTokensAction) {
lastStatus = 501;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: 'Gemini countTokens compatibility is not implemented for this upstream',
type: 'invalid_request_error',
},
});
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
const openAiBody = geminiGenerateContentTransformer.compatibility.buildOpenAiBodyFromGeminiRequest({
body: normalizedBody as Record<string, unknown>,
modelName: actualModel,
stream: isStreamAction,
});
const endpointCandidates = await resolveUpstreamEndpointCandidates(
{
site: selected.site,
account: selected.account,
},
actualModel,
'openai',
requestedModel,
);
const buildEndpointRequest = (
endpoint: 'chat' | 'messages' | 'responses',
requestOptions: { forceNormalizeClaudeBody?: boolean } = {},
) => {
const endpointRequest = buildUpstreamEndpointRequest({
endpoint,
modelName: actualModel,
stream: isStreamAction,
tokenValue: selected.tokenValue,
oauthProvider: oauth?.provider,
oauthProjectId: oauth?.projectId,
sitePlatform: selected.site.platform,
siteUrl: selected.site.url,
openaiBody: openAiBody,
downstreamFormat: 'openai',
forceNormalizeClaudeBody: requestOptions.forceNormalizeClaudeBody,
downstreamHeaders: request.headers as Record<string, unknown>,
providerHeaders: buildOauthProviderHeaders({
extraConfig: typeof selected.account.extraConfig === 'string' ? selected.account.extraConfig : null,
downstreamHeaders: request.headers as Record<string, unknown>,
}),
});
return {
endpoint,
path: endpointRequest.path,
headers: endpointRequest.headers,
body: endpointRequest.body as Record<string, unknown>,
runtime: endpointRequest.runtime,
};
};
const dispatchRequest = (compatibilityRequest: BuiltEndpointRequest, targetUrl?: string) => (
dispatchRuntimeRequest({
siteUrl: selected.site.url,
targetUrl,
request: compatibilityRequest,
buildInit: async (requestUrl, requestForFetch) => withSiteProxyRequestInit(requestUrl, {
method: 'POST',
headers: requestForFetch.headers,
body: JSON.stringify(requestForFetch.body),
}),
})
);
const endpointStrategy = createChatEndpointStrategy({
downstreamFormat: 'openai',
endpointCandidates,
modelName: actualModel,
requestedModelHint: requestedModel,
sitePlatform: selected.site.platform,
isStream: isStreamAction,
buildRequest: ({ endpoint, forceNormalizeClaudeBody }) => buildEndpointRequest(
endpoint,
{ forceNormalizeClaudeBody },
),
dispatchRequest,
});
const endpointResult = await executeEndpointFlow({
siteUrl: selected.site.url,
endpointCandidates,
buildRequest: (endpoint) => buildEndpointRequest(endpoint),
dispatchRequest,
tryRecover: endpointStrategy.tryRecover,
shouldDowngrade: endpointStrategy.shouldDowngrade,
});
if (!endpointResult.ok) {
lastStatus = endpointResult.status;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: endpointResult.errText,
type: 'upstream_error',
},
});
await tokenRouter.recordFailure?.(selected.channel.id);
await logProxy(
selected,
requestedModel,
'failed',
lastStatus,
Date.now() - startTime,
endpointResult.errText,
retryCount,
downstreamPath,
null,
);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
upstreamPath = endpointResult.upstreamPath;
const upstream = endpointResult.upstream;
const rawText = await upstream.text();
let upstreamData: unknown = rawText;
try {
upstreamData = JSON.parse(rawText);
} catch {}
const parsedUsage = parseProxyUsage(upstreamData);
const normalizedFinal = normalizeUpstreamFinalResponse(upstreamData, actualModel, rawText);
const geminiResponse = geminiGenerateContentTransformer.compatibility.serializeNormalizedFinalToGemini({
normalized: normalizedFinal,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
const latency = Date.now() - startTime;
await tokenRouter.recordSuccess?.(selected.channel.id, latency, 0);
await logProxy(
selected,
requestedModel,
'success',
upstream.status,
latency,
null,
retryCount,
downstreamPath,
upstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
);
return reply.code(upstream.status).send(
isGeminiCliDownstream
? { response: geminiResponse }
: geminiResponse,
);
} catch (error) {
lastStatus = 502;
lastContentType = 'application/json';
lastText = JSON.stringify({
error: {
message: error instanceof Error ? error.message : 'Gemini upstream request failed',
type: 'upstream_error',
},
});
await tokenRouter.recordFailure?.(selected.channel.id);
await logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
error instanceof Error ? error.message : 'Gemini upstream request failed',
retryCount,
downstreamPath,
upstreamPath || null,
);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
return reply.code(lastStatus).type(lastContentType).send(lastText);
}
}
};
const generateContent = async (request: FastifyRequest, reply: FastifyReply) => handleGenerateContent(request, reply);
const geminiCliGenerateContent = async (request: FastifyRequest, reply: FastifyReply) => handleGenerateContent(request, reply, {
downstreamProtocol: 'gemini-cli',
action: 'generateContent',
});
const geminiCliStreamGenerateContent = async (request: FastifyRequest, reply: FastifyReply) => handleGenerateContent(request, reply, {
downstreamProtocol: 'gemini-cli',
action: 'streamGenerateContent',
});
const geminiCliCountTokens = async (request: FastifyRequest, reply: FastifyReply) => handleGenerateContent(request, reply, {
downstreamProtocol: 'gemini-cli',
action: 'countTokens',
});
app.get('/v1beta/models', listModels);
app.get('/gemini/:geminiApiVersion/models', listModels);
app.post('/v1beta/models/*', generateContent);
app.post('/gemini/:geminiApiVersion/models/*', generateContent);
app.post('/v1internal::generateContent', geminiCliGenerateContent);
app.post('/v1internal::streamGenerateContent', geminiCliStreamGenerateContent);
app.post('/v1internal::countTokens', geminiCliCountTokens);
}
@@ -0,0 +1,114 @@
import type { ProxyResourceOwner } from '../../middleware/auth.js';
import { getProxyFileByPublicIdForOwner } from '../../services/proxyFileStore.js';
import {
inferInputFileMimeType,
normalizeInputFileBlock,
toOpenAiChatFileBlock,
toResponsesInputFileBlock,
type NormalizedInputFile,
} from '../../transformers/shared/inputFile.js';
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function isSupportedFileMimeType(mimeType: string): boolean {
return mimeType === 'application/pdf'
|| mimeType === 'text/plain'
|| mimeType === 'text/markdown'
|| mimeType === 'application/json'
|| mimeType.startsWith('image/')
|| mimeType.startsWith('audio/');
}
export class ProxyInputFileResolutionError extends Error {
statusCode: number;
payload: unknown;
constructor(statusCode: number, payload: unknown) {
super(typeof payload === 'object' && payload && 'error' in (payload as Record<string, unknown>)
? String(((payload as { error?: { message?: unknown } }).error?.message) || 'input file resolution failed')
: 'input file resolution failed');
this.name = 'ProxyInputFileResolutionError';
this.statusCode = statusCode;
this.payload = payload;
}
}
function buildInvalidRequest(message: string): ProxyInputFileResolutionError {
return new ProxyInputFileResolutionError(400, {
error: {
message,
type: 'invalid_request_error',
},
});
}
function buildNotFound(message: string): ProxyInputFileResolutionError {
return new ProxyInputFileResolutionError(404, {
error: {
message,
type: 'not_found_error',
},
});
}
async function resolveInputFileBlock(
item: Record<string, unknown>,
owner: ProxyResourceOwner,
): Promise<Record<string, unknown>> {
const normalized = normalizeInputFileBlock(item);
if (!normalized) return item;
let resolved: NormalizedInputFile = { ...normalized };
if (!resolved.fileData && resolved.fileId?.startsWith('file-metapi-')) {
const stored = await getProxyFileByPublicIdForOwner(resolved.fileId, owner);
if (!stored) {
throw buildNotFound(`file not found: ${resolved.fileId}`);
}
resolved = {
filename: stored.filename,
mimeType: stored.mimeType,
fileData: stored.contentBase64,
};
}
const inferredMimeType = inferInputFileMimeType(resolved);
if (resolved.fileData && (!inferredMimeType || !isSupportedFileMimeType(inferredMimeType))) {
throw buildInvalidRequest(`unsupported file mime type: ${inferredMimeType || 'application/octet-stream'}`);
}
if (inferredMimeType) {
resolved.mimeType = inferredMimeType;
}
const type = typeof item.type === 'string' ? item.type : '';
if (type === 'input_file') {
return toResponsesInputFileBlock(resolved);
}
if (type === 'file') {
return toOpenAiChatFileBlock(resolved);
}
return item;
}
export async function inlineLocalInputFileReferences(
value: unknown,
owner: ProxyResourceOwner,
): Promise<unknown> {
if (Array.isArray(value)) {
return Promise.all(value.map((item) => inlineLocalInputFileReferences(item, owner)));
}
if (!isRecord(value)) return value;
const type = typeof value.type === 'string' ? value.type : '';
if (type === 'input_file' || type === 'file') {
return resolveInputFileBlock(value, owner);
}
const entries = await Promise.all(
Object.entries(value).map(async ([key, entryValue]) => [key, await inlineLocalInputFileReferences(entryValue, owner)] as const),
);
return Object.fromEntries(entries);
}
@@ -0,0 +1,97 @@
import { describe, expect, it, vi } from 'vitest';
import { listModelsSurface } from './modelsSurface.js';
describe('listModelsSurface', () => {
it('returns OpenAI list shape and hides models without a resolvable channel', async () => {
const result = await listModelsSurface({
downstreamPolicy: { type: 'all' },
responseFormat: 'openai',
tokenRouter: {
getAvailableModels: vi.fn().mockResolvedValue(['routable-model', 'orphan-model']),
explainSelection: vi.fn()
.mockResolvedValueOnce({ selectedChannelId: null })
.mockResolvedValueOnce({ selectedChannelId: 11 }),
},
refreshModelsAndRebuildRoutes: vi.fn(),
isModelAllowed: vi.fn().mockResolvedValue(true),
now: () => new Date('2026-03-19T00:00:00.000Z'),
});
expect(result).toEqual({
object: 'list',
data: [
{
id: 'routable-model',
object: 'model',
created: 1773878400,
owned_by: 'metapi',
},
],
});
});
it('returns Claude list shape when requested', async () => {
const result = await listModelsSurface({
downstreamPolicy: { type: 'all' },
responseFormat: 'claude',
tokenRouter: {
getAvailableModels: vi.fn().mockResolvedValue(['claude-opus-4-6']),
explainSelection: vi.fn().mockResolvedValue({ selectedChannelId: 22 }),
},
refreshModelsAndRebuildRoutes: vi.fn(),
isModelAllowed: vi.fn().mockResolvedValue(true),
now: () => new Date('2026-03-19T00:00:00.000Z'),
});
expect(result).toEqual({
data: [
{
id: 'claude-opus-4-6',
type: 'model',
display_name: 'claude-opus-4-6',
created_at: '2026-03-19T00:00:00.000Z',
},
],
first_id: 'claude-opus-4-6',
last_id: 'claude-opus-4-6',
has_more: false,
});
});
it('applies downstream policy filtering before selection checks and refreshes once when the first read is empty', async () => {
const getAvailableModels = vi.fn()
.mockResolvedValueOnce(['blocked-model'])
.mockResolvedValueOnce(['allowed-model']);
const refreshModelsAndRebuildRoutes = vi.fn().mockResolvedValue(undefined);
const isModelAllowed = vi.fn()
.mockResolvedValueOnce(false)
.mockResolvedValueOnce(true);
const explainSelection = vi.fn().mockResolvedValue({ selectedChannelId: 33 });
const result = await listModelsSurface({
downstreamPolicy: { type: 'whitelist' },
responseFormat: 'openai',
tokenRouter: {
getAvailableModels,
explainSelection,
},
refreshModelsAndRebuildRoutes,
isModelAllowed,
now: () => new Date('2026-03-19T00:00:00.000Z'),
});
expect(refreshModelsAndRebuildRoutes).toHaveBeenCalledTimes(1);
expect(result).toEqual({
object: 'list',
data: [
{
id: 'allowed-model',
object: 'model',
created: 1773878400,
owned_by: 'metapi',
},
],
});
});
});
@@ -0,0 +1,70 @@
function isSearchPseudoModel(modelName: string): boolean {
const normalized = (modelName || '').trim().toLowerCase();
if (!normalized) return false;
return normalized === '__search' || /^__.+_search$/.test(normalized);
}
type ModelsSurfaceInput = {
downstreamPolicy: unknown;
responseFormat: 'openai' | 'claude';
tokenRouter: {
getAvailableModels(): Promise<string[]>;
explainSelection(modelName: string, excludeChannelIds: number[], downstreamPolicy: unknown): Promise<{
selectedChannelId?: number | null;
}>;
};
refreshModelsAndRebuildRoutes(): Promise<unknown>;
isModelAllowed(modelName: string, downstreamPolicy: unknown): Promise<boolean>;
now?: () => Date;
};
async function readVisibleModels(input: ModelsSurfaceInput): Promise<string[]> {
const deduped = Array.from(new Set(await input.tokenRouter.getAvailableModels()))
.filter((modelName) => !isSearchPseudoModel(modelName))
.sort();
const allowed: string[] = [];
for (const modelName of deduped) {
if (!await input.isModelAllowed(modelName, input.downstreamPolicy)) {
continue;
}
const decision = await input.tokenRouter.explainSelection(modelName, [], input.downstreamPolicy);
if (typeof decision.selectedChannelId === 'number') {
allowed.push(modelName);
}
}
return allowed;
}
export async function listModelsSurface(input: ModelsSurfaceInput) {
let models = await readVisibleModels(input);
if (models.length === 0) {
await input.refreshModelsAndRebuildRoutes();
models = await readVisibleModels(input);
}
const now = input.now?.() ?? new Date();
if (input.responseFormat === 'claude') {
const data = models.map((id) => ({
id,
type: 'model' as const,
display_name: id,
created_at: now.toISOString(),
}));
return {
data,
first_id: data[0]?.id || null,
last_id: data[data.length - 1]?.id || null,
has_more: false,
};
}
return {
object: 'list' as const,
data: models.map((id) => ({
id,
object: 'model' as const,
created: Math.floor(now.getTime() / 1000),
owned_by: 'metapi',
})),
};
}
@@ -0,0 +1,870 @@
import { TextDecoder } from 'node:util';
import type { FastifyReply, FastifyRequest } from 'fastify';
import { db, hasProxyLogDownstreamApiKeyIdColumn, schema } from '../../db/index.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { refreshModelsAndRebuildRoutes } from '../../services/modelService.js';
import { reportProxyAllFailed, reportTokenExpired } from '../../services/alertService.js';
import { isTokenExpiredError } from '../../services/alertRules.js';
import { shouldRetryProxyRequest } from '../../services/proxyRetryPolicy.js';
import { resolveProxyUsageWithSelfLogFallback } from '../../services/proxyUsageFallbackService.js';
import { mergeProxyUsage, parseProxyUsage } from '../../services/proxyUsageParser.js';
import { resolveChannelProxyUrl, withSiteRecordProxyRequestInit } from '../../services/siteProxy.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import {
buildUpstreamEndpointRequest,
resolveUpstreamEndpointCandidates,
} from '../../routes/proxy/upstreamEndpoint.js';
import { ensureModelAllowedForDownstreamKey, getDownstreamRoutingPolicy, recordDownstreamCostUsage } from '../../routes/proxy/downstreamPolicy.js';
import { composeProxyLogMessage } from '../../routes/proxy/logPathMeta.js';
import { executeEndpointFlow, type BuiltEndpointRequest } from '../../routes/proxy/endpointFlow.js';
import { detectProxyFailure } from '../../routes/proxy/proxyFailureJudge.js';
import { formatUtcSqlDateTime } from '../../services/localTimeService.js';
import { resolveProxyLogBilling } from '../../routes/proxy/proxyBilling.js';
import { getProxyAuthContext, getProxyResourceOwner } from '../../middleware/auth.js';
import { dispatchRuntimeRequest } from '../../routes/proxy/runtimeExecutor.js';
import { normalizeInputFileBlock } from '../../transformers/shared/inputFile.js';
import {
ProxyInputFileResolutionError,
hasNonImageFileInputInOpenAiBody,
resolveResponsesBodyInputFiles,
} from '../../services/proxyInputFileResolver.js';
import {
buildOauthProviderHeaders,
} from '../../services/oauth/service.js';
import { getOauthInfoFromExtraConfig } from '../../services/oauth/oauthAccount.js';
import { recordOauthQuotaResetHint } from '../../services/oauth/quota.js';
import { refreshOauthAccessTokenSingleflight } from '../../services/oauth/refreshSingleflight.js';
import { collectResponsesFinalPayloadFromSse } from '../../routes/proxy/responsesSseFinal.js';
import {
createGeminiCliStreamReader,
unwrapGeminiCliPayload,
} from '../../routes/proxy/geminiCliCompat.js';
import { isCodexResponsesSurface } from '../cliProfiles/codexProfile.js';
import { detectCliProfile } from '../cliProfiles/registry.js';
import type { CliProfileId } from '../cliProfiles/types.js';
const MAX_RETRIES = 2;
function isRecord(value: unknown): value is Record<string, unknown> {
return !!value && typeof value === 'object';
}
function normalizeIncludeList(value: unknown): string[] {
if (typeof value === 'string') {
const trimmed = value.trim();
return trimmed ? [trimmed] : [];
}
if (!Array.isArray(value)) return [];
return value
.map((item) => (typeof item === 'string' ? item.trim() : ''))
.filter((item) => item.length > 0);
}
function hasExplicitInclude(body: Record<string, unknown>): boolean {
return Object.prototype.hasOwnProperty.call(body, 'include');
}
function hasResponsesReasoningRequest(value: unknown): boolean {
if (!isRecord(value)) return false;
const relevantKeys = ['effort', 'budget_tokens', 'budgetTokens', 'max_tokens', 'maxTokens', 'summary'];
return relevantKeys.some((key) => {
const entry = value[key];
if (typeof entry === 'string') return entry.trim().length > 0;
return entry !== undefined && entry !== null;
});
}
function carriesResponsesReasoningContinuity(value: unknown): boolean {
if (Array.isArray(value)) {
return value.some((item) => carriesResponsesReasoningContinuity(item));
}
if (!isRecord(value)) return false;
const type = typeof value.type === 'string' ? value.type.trim().toLowerCase() : '';
if (type === 'reasoning') {
if (typeof value.encrypted_content === 'string' && value.encrypted_content.trim()) {
return true;
}
if (Array.isArray(value.summary) && value.summary.length > 0) {
return true;
}
}
if (typeof value.reasoning_signature === 'string' && value.reasoning_signature.trim()) {
return true;
}
return carriesResponsesReasoningContinuity(value.input)
|| carriesResponsesReasoningContinuity(value.content);
}
function wantsNativeResponsesReasoning(body: unknown): boolean {
if (!isRecord(body)) return false;
const include = normalizeIncludeList(body.include);
if (include.some((item) => item.toLowerCase() === 'reasoning.encrypted_content')) {
return true;
}
if (carriesResponsesReasoningContinuity(body.input)) {
return true;
}
if (hasExplicitInclude(body)) {
return false;
}
return hasResponsesReasoningRequest(body.reasoning);
}
function carriesResponsesFileUrlInput(value: unknown): boolean {
if (Array.isArray(value)) {
return value.some((item) => carriesResponsesFileUrlInput(item));
}
if (!isRecord(value)) return false;
const normalizedFile = normalizeInputFileBlock(value);
if (normalizedFile?.fileUrl) return true;
return Object.values(value).some((entry) => carriesResponsesFileUrlInput(entry));
}
type UsageSummary = ReturnType<typeof parseProxyUsage>;
type DownstreamClientContext = {
clientKind: CliProfileId;
sessionId?: string;
traceHint?: string;
};
function detectDownstreamClientContext(input: {
downstreamPath: string;
headers?: Record<string, unknown>;
body?: unknown;
}): DownstreamClientContext {
const detected = detectCliProfile(input);
return {
clientKind: detected.id,
...(detected.sessionId ? { sessionId: detected.sessionId } : {}),
...(detected.traceHint ? { traceHint: detected.traceHint } : {}),
};
}
function deriveCodexExplicitSessionId(body: Record<string, unknown>): string | null {
const promptCacheKey = typeof body.prompt_cache_key === 'string'
? body.prompt_cache_key.trim()
: '';
return promptCacheKey || null;
}
export async function handleOpenAiResponsesSurfaceRequest(
request: FastifyRequest,
reply: FastifyReply,
downstreamPath: '/v1/responses' | '/v1/responses/compact',
) {
const body = request.body as Record<string, unknown>;
const clientContext = detectDownstreamClientContext({
downstreamPath,
headers: request.headers as Record<string, unknown>,
body,
});
const defaultEncryptedReasoningInclude = isCodexResponsesSurface(
request.headers as Record<string, unknown>,
);
const parsedRequestEnvelope = openAiResponsesTransformer.transformRequest(body, {
defaultEncryptedReasoningInclude,
});
if (parsedRequestEnvelope.error) {
return reply.code(parsedRequestEnvelope.error.statusCode).send(parsedRequestEnvelope.error.payload);
}
const requestEnvelope = parsedRequestEnvelope.value!;
const requestedModel = requestEnvelope.model;
const isStream = requestEnvelope.stream;
const isCompactRequest = downstreamPath === '/v1/responses/compact';
if (isCompactRequest && isStream) {
return reply.code(400).send({
error: {
message: 'stream is not supported on /v1/responses/compact',
type: 'invalid_request_error',
},
});
}
if (!await ensureModelAllowedForDownstreamKey(request, reply, requestedModel)) return;
const downstreamPolicy = getDownstreamRoutingPolicy(request);
const downstreamApiKeyId = getProxyAuthContext(request)?.keyId ?? null;
const logDownstreamApiKeyId = downstreamApiKeyId !== null
&& await hasProxyLogDownstreamApiKeyIdColumn();
const excludeChannelIds: number[] = [];
let retryCount = 0;
while (retryCount <= MAX_RETRIES) {
let selected = retryCount === 0
? await tokenRouter.selectChannel(requestedModel, downstreamPolicy)
: await tokenRouter.selectNextChannel(requestedModel, excludeChannelIds, downstreamPolicy);
if (!selected && retryCount === 0) {
await refreshModelsAndRebuildRoutes();
selected = await tokenRouter.selectChannel(requestedModel, downstreamPolicy);
}
if (!selected) {
await reportProxyAllFailed({
model: requestedModel,
reason: 'No available channels after retries',
});
return reply.code(503).send({
error: { message: 'No available channels for this model', type: 'server_error' },
});
}
excludeChannelIds.push(selected.channel.id);
const modelName = selected.actualModel || requestedModel;
const oauth = getOauthInfoFromExtraConfig(selected.account.extraConfig);
const isCodexSite = String(selected.site.platform || '').trim().toLowerCase() === 'codex';
const owner = getProxyResourceOwner(request);
let normalizedResponsesBody: Record<string, unknown> = {
...requestEnvelope.parsed.normalizedBody,
model: modelName,
stream: isStream,
};
if (body.generate === false) {
normalizedResponsesBody.generate = false;
}
if (owner) {
try {
normalizedResponsesBody = await resolveResponsesBodyInputFiles(normalizedResponsesBody, owner);
} catch (error) {
if (error instanceof ProxyInputFileResolutionError) {
return reply.code(error.statusCode).send(error.payload);
}
throw error;
}
}
const openAiBody = openAiResponsesTransformer.inbound.toOpenAiBody(
normalizedResponsesBody,
modelName,
isStream,
{ defaultEncryptedReasoningInclude },
);
const hasNonImageFileInput = hasNonImageFileInputInOpenAiBody(openAiBody);
const prefersNativeResponsesReasoning = wantsNativeResponsesReasoning(normalizedResponsesBody);
const requiresNativeResponsesFileUrl = carriesResponsesFileUrlInput(normalizedResponsesBody.input);
if (requiresNativeResponsesFileUrl && String(selected.site.platform || '').trim().toLowerCase() === 'claude') {
return reply.code(400).send({
error: {
message: 'Responses input_file.file_url requires an upstream /v1/responses endpoint; current site only supports /v1/messages.',
type: 'invalid_request_error',
},
});
}
const endpointCandidates = await resolveUpstreamEndpointCandidates(
{
site: selected.site,
account: selected.account,
},
modelName,
'responses',
requestedModel,
{
hasNonImageFileInput,
wantsNativeResponsesReasoning: prefersNativeResponsesReasoning,
},
);
if (requiresNativeResponsesFileUrl) {
endpointCandidates.splice(0, endpointCandidates.length, 'responses');
}
if (endpointCandidates.length === 0) {
endpointCandidates.push('responses', 'chat', 'messages');
}
const buildProviderHeaders = () => (
buildOauthProviderHeaders({
extraConfig: typeof selected.account.extraConfig === 'string' ? selected.account.extraConfig : null,
downstreamHeaders: request.headers as Record<string, unknown>,
})
);
const buildEndpointRequest = (endpoint: 'chat' | 'messages' | 'responses') => {
const upstreamStream = isStream || (isCodexSite && endpoint === 'responses');
const codexExplicitSessionId = deriveCodexExplicitSessionId(normalizedResponsesBody);
const endpointRequest = buildUpstreamEndpointRequest({
endpoint,
modelName,
stream: upstreamStream,
tokenValue: selected.tokenValue,
oauthProvider: oauth?.provider,
oauthProjectId: oauth?.projectId,
sitePlatform: selected.site.platform,
siteUrl: selected.site.url,
openaiBody: openAiBody,
downstreamFormat: 'responses',
responsesOriginalBody: normalizedResponsesBody,
downstreamHeaders: request.headers as Record<string, unknown>,
providerHeaders: buildProviderHeaders(),
codexExplicitSessionId,
});
const upstreamPath = (
isCompactRequest && endpoint === 'responses'
? `${endpointRequest.path}/compact`
: endpointRequest.path
);
return {
endpoint,
path: upstreamPath,
headers: endpointRequest.headers,
body: endpointRequest.body as Record<string, unknown>,
runtime: endpointRequest.runtime,
};
};
const dispatchRequest = (compatibilityRequest: BuiltEndpointRequest, targetUrl?: string) => (
dispatchRuntimeRequest({
siteUrl: selected.site.url,
targetUrl,
request: compatibilityRequest,
buildInit: (_requestUrl, requestForFetch) => withSiteRecordProxyRequestInit(selected.site, {
method: 'POST',
headers: requestForFetch.headers,
body: JSON.stringify(requestForFetch.body),
}),
})
);
const endpointStrategy = openAiResponsesTransformer.compatibility.createEndpointStrategy({
isStream: isStream || isCodexSite,
requiresNativeResponsesFileUrl,
dispatchRequest,
});
const tryRecover = async (ctx: Parameters<NonNullable<typeof endpointStrategy.tryRecover>>[0]) => {
if (ctx.response.status === 401 && oauth) {
try {
const refreshed = await refreshOauthAccessTokenSingleflight(selected.account.id);
selected.tokenValue = refreshed.accessToken;
selected.account = {
...selected.account,
accessToken: refreshed.accessToken,
extraConfig: refreshed.extraConfig ?? selected.account.extraConfig,
};
const refreshedRequest = buildEndpointRequest(ctx.request.endpoint);
const refreshedTargetUrl = `${selected.site.url}${refreshedRequest.path}`;
const refreshedResponse = await dispatchRequest(refreshedRequest, refreshedTargetUrl);
if (refreshedResponse.ok) {
return {
upstream: refreshedResponse,
upstreamPath: refreshedRequest.path,
};
}
ctx.request = refreshedRequest;
ctx.response = refreshedResponse;
ctx.rawErrText = await refreshedResponse.text().catch(() => 'unknown error');
} catch {
return endpointStrategy.tryRecover(ctx);
}
}
return endpointStrategy.tryRecover(ctx);
};
const startTime = Date.now();
try {
const endpointResult = await executeEndpointFlow({
siteUrl: selected.site.url,
proxyUrl: resolveChannelProxyUrl(selected.site, selected.account.extraConfig),
endpointCandidates,
buildRequest: (endpoint) => buildEndpointRequest(endpoint),
dispatchRequest,
tryRecover,
shouldDowngrade: endpointStrategy.shouldDowngrade,
onDowngrade: (ctx) => {
logProxy(
selected,
requestedModel,
'failed',
ctx.response.status,
Date.now() - startTime,
ctx.errText,
retryCount,
downstreamPath,
0,
0,
0,
0,
null,
null,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
},
});
if (!endpointResult.ok) {
const status = endpointResult.status || 502;
const errText = endpointResult.errText || 'unknown error';
const rawErrText = endpointResult.rawErrText || errText;
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
status,
Date.now() - startTime,
errText,
retryCount,
downstreamPath,
0,
0,
0,
0,
null,
null,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
await recordOauthQuotaResetHint({
accountId: selected.account.id,
statusCode: status,
errorText: rawErrText,
});
if (isTokenExpiredError({ status, message: errText })) {
await reportTokenExpired({
accountId: selected.account.id,
username: selected.account.username,
siteName: selected.site.name,
detail: `HTTP ${status}`,
});
}
if (shouldRetryProxyRequest(status, errText) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: `upstream returned HTTP ${status}`,
});
return reply.code(status).send({ error: { message: errText, type: 'upstream_error' } });
}
const upstream = endpointResult.upstream;
const successfulUpstreamPath = endpointResult.upstreamPath;
const finalizeStreamSuccess = async (parsedUsage: UsageSummary, latency: number) => {
let usageForLog = {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
};
let estimatedCost = 0;
let billingDetails: unknown = null;
try {
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
usageForLog = {
promptTokens: resolvedUsage.promptTokens,
completionTokens: resolvedUsage.completionTokens,
totalTokens: resolvedUsage.totalTokens,
};
const billing = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
estimatedCost = billing.estimatedCost;
billingDetails = billing.billingDetails;
} catch (error) {
console.error('[responses] post-stream bookkeeping failed:', error);
}
try {
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount, downstreamPath,
usageForLog.promptTokens, usageForLog.completionTokens, usageForLog.totalTokens, estimatedCost, billingDetails,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
} catch (error) {
console.error('[responses] post-stream success logging failed:', error);
}
};
if (isStream) {
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
const startSseResponse = () => {
reply.hijack();
reply.raw.statusCode = 200;
reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
reply.raw.setHeader('Cache-Control', 'no-cache, no-transform');
reply.raw.setHeader('Connection', 'keep-alive');
reply.raw.setHeader('X-Accel-Buffering', 'no');
};
let parsedUsage: 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();
let upstreamData: unknown = rawText;
try {
upstreamData = JSON.parse(rawText);
} catch {
upstreamData = rawText;
}
if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
upstreamData = unwrapGeminiCliPayload(upstreamData);
}
parsedUsage = parseProxyUsage(upstreamData);
const latency = Date.now() - startTime;
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
if (failure) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
failure.status,
latency,
failure.reason,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: failure.reason,
});
return reply.code(failure.status).send({ error: { message: failure.reason, type: 'upstream_error' } });
}
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;
}
await finalizeStreamSuccess(parsedUsage, latency);
return;
}
startSseResponse();
const upstreamReader = upstream.body?.getReader();
const baseReader = String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli' && upstreamReader
? createGeminiCliStreamReader(upstreamReader)
: upstreamReader;
let rawText = '';
const decoder = new TextDecoder();
const reader = baseReader
? {
async read() {
const result = await baseReader.read();
if (result.value) {
rawText += decoder.decode(result.value, { stream: true });
}
return result;
},
async cancel(reason?: unknown) {
return baseReader.cancel(reason);
},
releaseLock() {
return baseReader.releaseLock();
},
}
: baseReader;
const streamResult = await streamSession.run(reader, reply.raw);
rawText += decoder.decode();
const latency = Date.now() - startTime;
if (streamResult.status === 'failed') {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
200,
latency,
streamResult.errorMessage,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return;
}
// Once SSE has been hijacked and bytes may already be on the wire, we
// must not attempt to convert stream failures into a fresh HTTP error
// response or retry on another channel. Responses stream failures are
// handled in-band by the proxy stream session.
await finalizeStreamSuccess(parsedUsage, latency);
return;
}
const upstreamContentType = (upstream.headers.get('content-type') || '').toLowerCase();
let rawText = '';
let upstreamData: unknown;
if (
upstreamContentType.includes('text/event-stream')
&& (
successfulUpstreamPath.endsWith('/responses')
|| successfulUpstreamPath.endsWith('/responses/compact')
)
) {
const collected = await collectResponsesFinalPayloadFromSse(upstream, modelName);
rawText = collected.rawText;
upstreamData = collected.payload;
} else {
rawText = await upstream.text();
upstreamData = rawText;
try {
upstreamData = JSON.parse(rawText);
} catch {
upstreamData = rawText;
}
}
if (String(selected.site.platform || '').trim().toLowerCase() === 'gemini-cli') {
upstreamData = unwrapGeminiCliPayload(upstreamData);
}
const latency = Date.now() - startTime;
const parsedUsage = parseProxyUsage(upstreamData);
const failure = detectProxyFailure({ rawText, usage: parsedUsage });
if (failure) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
failure.status,
latency,
failure.reason,
retryCount,
downstreamPath,
parsedUsage.promptTokens,
parsedUsage.completionTokens,
parsedUsage.totalTokens,
0,
null,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
if (shouldRetryProxyRequest(failure.status, failure.reason) && retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: failure.reason,
});
return reply.code(failure.status).send({ error: { message: failure.reason, type: 'upstream_error' } });
}
const normalized = openAiResponsesTransformer.transformFinalResponse(
upstreamData,
modelName,
rawText,
);
const downstreamData = openAiResponsesTransformer.outbound.serializeFinal({
upstreamPayload: upstreamData,
normalized,
usage: parsedUsage,
serializationMode: isCompactRequest ? 'compact' : 'response',
});
const resolvedUsage = await resolveProxyUsageWithSelfLogFallback({
site: selected.site,
account: selected.account,
tokenValue: selected.tokenValue,
tokenName: selected.tokenName,
modelName: selected.actualModel || requestedModel,
requestStartedAtMs: startTime,
requestEndedAtMs: startTime + latency,
localLatencyMs: latency,
usage: {
promptTokens: parsedUsage.promptTokens,
completionTokens: parsedUsage.completionTokens,
totalTokens: parsedUsage.totalTokens,
},
});
const { estimatedCost, billingDetails } = await resolveProxyLogBilling({
site: selected.site,
account: selected.account,
modelName: selected.actualModel || requestedModel,
parsedUsage,
resolvedUsage,
});
tokenRouter.recordSuccess(selected.channel.id, latency, estimatedCost);
recordDownstreamCostUsage(request, estimatedCost);
logProxy(
selected, requestedModel, 'success', 200, latency, null, retryCount, downstreamPath,
resolvedUsage.promptTokens, resolvedUsage.completionTokens, resolvedUsage.totalTokens, estimatedCost, billingDetails,
successfulUpstreamPath,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
return reply.send(downstreamData);
} catch (err: any) {
tokenRouter.recordFailure(selected.channel.id);
logProxy(
selected,
requestedModel,
'failed',
0,
Date.now() - startTime,
err.message,
retryCount,
downstreamPath,
0,
0,
0,
0,
null,
null,
clientContext,
logDownstreamApiKeyId ? downstreamApiKeyId : null,
);
if (retryCount < MAX_RETRIES) {
retryCount += 1;
continue;
}
await reportProxyAllFailed({
model: requestedModel,
reason: err.message || 'network failure',
});
return reply.code(502).send({
error: { message: `Upstream error: ${err.message}`, type: 'upstream_error' },
});
}
}
}
async function logProxy(
selected: any,
modelRequested: string,
status: string,
httpStatus: number,
latencyMs: number,
errorMessage: string | null,
retryCount: number,
downstreamPath: string,
promptTokens = 0,
completionTokens = 0,
totalTokens = 0,
estimatedCost = 0,
billingDetails: unknown = null,
upstreamPath: string | null = null,
clientContext: DownstreamClientContext | null = null,
downstreamApiKeyId: number | null = null,
) {
try {
const createdAt = formatUtcSqlDateTime(new Date());
const normalizedErrorMessage = composeProxyLogMessage({
clientKind: clientContext?.clientKind && clientContext.clientKind !== 'generic'
? clientContext.clientKind
: null,
sessionId: clientContext?.sessionId || null,
traceHint: clientContext?.traceHint || null,
downstreamPath,
upstreamPath,
errorMessage,
});
await db.insert(schema.proxyLogs).values({
routeId: selected.channel.routeId,
channelId: selected.channel.id,
accountId: selected.account.id,
...(downstreamApiKeyId !== null ? { downstreamApiKeyId } : {}),
modelRequested,
modelActual: selected.actualModel,
status,
httpStatus,
latencyMs,
promptTokens,
completionTokens,
totalTokens,
estimatedCost,
billingDetails: billingDetails ? JSON.stringify(billingDetails) : null,
errorMessage: normalizedErrorMessage,
retryCount,
createdAt,
}).run();
} catch (error) {
console.warn('[proxy/responses] failed to write proxy log', error);
}
}