Nvidia/BossAI
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

complete codex cliproxyapi parity

Browse Source
This commit is contained in:
cita
2026-03-20 11:23:29 +08:00
parent 5403a946c6
commit adc4fb1be7
28 changed files with 2969 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/server/routes/proxy/responses.websocket.test.ts
+375 -53
View File
@@ -1,6 +1,7 @@
import Fastify, { type FastifyInstance } from 'fastify';
import { createServer, type Server } from 'node:http';
import { AddressInfo } from 'node:net';
import WebSocket from 'ws';
import WebSocket, { WebSocketServer } from 'ws';
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest';
const fetchMock = vi.fn();
@@ -175,6 +176,14 @@ function waitForSocketMessages(socket: WebSocket, count: number, timeoutMs = 100
describe('responses websocket transport', () => {
let app: FastifyInstance;
let baseUrl: string;
let upstreamServer: WebSocketServer;
let upstreamSiteUrl: string;
let upstreamConnectionCount: number;
let upstreamUpgradeHeaders: Record<string, string>;
let upstreamRequests: Record<string, unknown>[];
let upstreamMessageHandler: (socket: WebSocket, parsed: Record<string, unknown>, requestIndex: number) => void;
let rejectedUpgradeServer: Server;
let rejectedUpgradeSiteUrl: string;
beforeAll(async () => {
const { responsesProxyRoute } = await import('./responses.js');
@@ -183,6 +192,39 @@ describe('responses websocket transport', () => {
await app.listen({ port: 0, host: '127.0.0.1' });
const address = app.server.address() as AddressInfo;
baseUrl = `ws://127.0.0.1:${address.port}`;
upstreamServer = new WebSocketServer({ port: 0 });
upstreamServer.on('connection', (socket, request) => {
upstreamConnectionCount += 1;
upstreamUpgradeHeaders = Object.fromEntries(
Object.entries(request.headers)
.map(([key, value]) => [key, Array.isArray(value) ? value[0] || '' : value || '']),
);
socket.on('message', (payload) => {
const parsed = JSON.parse(String(payload)) as Record<string, unknown>;
upstreamRequests.push(parsed);
upstreamMessageHandler(socket, parsed, upstreamRequests.length);
});
});
await new Promise<void>((resolve) => upstreamServer.once('listening', () => resolve()));
const upstreamAddress = upstreamServer.address() as AddressInfo;
upstreamSiteUrl = `http://127.0.0.1:${upstreamAddress.port}/backend-api/codex`;
rejectedUpgradeServer = createServer();
rejectedUpgradeServer.on('upgrade', (_request, socket) => {
socket.write(
'HTTP/1.1 426 Upgrade Required\r\n'
+ 'Content-Type: text/plain\r\n'
+ 'Content-Length: 16\r\n'
+ 'Connection: close\r\n'
+ '\r\n'
+ 'Upgrade Required',
);
socket.destroy();
});
await new Promise<void>((resolve) => rejectedUpgradeServer.listen(0, '127.0.0.1', () => resolve()));
const rejectedAddress = rejectedUpgradeServer.address() as AddressInfo;
rejectedUpgradeSiteUrl = `http://127.0.0.1:${rejectedAddress.port}/backend-api/codex`;
});
beforeEach(() => {
@@ -202,24 +244,90 @@ describe('responses websocket transport', () => {
selectChannelMock.mockReturnValue(selectedChannel);
selectNextChannelMock.mockReturnValue(null);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamConnectionCount = 0;
upstreamUpgradeHeaders = {};
upstreamRequests = [];
upstreamMessageHandler = (socket, parsed, requestIndex) => {
const responseId = `resp_upstream_${requestIndex}`;
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: responseId,
object: 'response',
model: parsed.model || 'gpt-5.4',
status: 'completed',
output: [],
usage: {
input_tokens: 1,
output_tokens: 1,
total_tokens: 2,
},
},
}));
};
});
afterAll(async () => {
await new Promise<void>((resolve) => rejectedUpgradeServer.close(() => resolve()));
await new Promise<void>((resolve) => upstreamServer.close(() => resolve()));
await app.close();
});
it('accepts response.create over GET /v1/responses websocket and forwards streamed responses events', async () => {
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_ws","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_item.added\n',
'data: {"type":"response.output_item.added","output_index":0,"item":{"id":"msg_ws","type":"message","role":"assistant","status":"in_progress","content":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_ws","delta":"pong"}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws","model":"gpt-5.4","status":"completed","output":[{"id":"msg_ws","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"pong"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket) => {
socket.send(JSON.stringify({
type: 'response.created',
response: {
id: 'resp_ws',
model: 'gpt-5.4',
created_at: 1706000000,
status: 'in_progress',
output: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_item.added',
output_index: 0,
item: {
id: 'msg_ws',
type: 'message',
role: 'assistant',
status: 'in_progress',
content: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_text.delta',
output_index: 0,
item_id: 'msg_ws',
delta: 'pong',
}));
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_ws',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'pong' }],
}],
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
}));
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -247,7 +355,8 @@ describe('responses websocket transport', () => {
'response.completed',
]);
expect(messages[3]?.response?.output?.[0]?.content?.[0]?.text).toBe('pong');
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamConnectionCount).toBe(1);
});
it('echoes x-codex-turn-state on websocket upgrade responses', async () => {
@@ -266,6 +375,89 @@ describe('responses websocket transport', () => {
expect(upgrade.headers['x-codex-turn-state']).toBe('turn-state-123');
});
it('reuses one upstream codex websocket session across sequential websocket turns', async () => {
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
const socket = new WebSocket(`${baseUrl}/v1/responses`, {
headers: {
'x-codex-turn-state': 'turn-state-123',
'x-codex-beta-features': 'feature-a,feature-b',
},
});
await waitForSocketOpen(socket);
const firstMessagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
input: [],
}));
const firstMessages = await firstMessagesPromise;
const secondMessagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
previous_response_id: firstMessages[0]?.response?.id,
input: [],
}));
const secondMessages = await secondMessagesPromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(firstMessages[0]?.type).toBe('response.completed');
expect(secondMessages[0]?.type).toBe('response.completed');
expect(upstreamConnectionCount).toBe(1);
expect(upstreamRequests).toHaveLength(2);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
model: 'gpt-5.4',
});
expect(upstreamRequests[1]).toMatchObject({
type: 'response.create',
previous_response_id: firstMessages[0]?.response?.id,
});
expect(upstreamUpgradeHeaders['x-codex-turn-state']).toBe('turn-state-123');
expect(upstreamUpgradeHeaders['x-codex-beta-features']).toBe('feature-a,feature-b');
expect(upstreamUpgradeHeaders['openai-beta']).toContain('responses_websockets=');
});
it('falls back to the HTTP responses executor when the upstream codex websocket upgrade returns 426', async () => {
const selectedChannel = createSelectedChannel({
siteUrl: rejectedUpgradeSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
fetchMock.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_fallback","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n',
'data: [DONE]\n\n',
]));
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const messagesPromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
input: [],
}));
const messages = await messagesPromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(1);
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
expect(messages[0]?.response?.id).toBe('resp_http_fallback');
});
it('merges follow-up response.create payloads when the selected upstream does not support incremental mode', async () => {
const selectedChannel = createSelectedChannel({
sitePlatform: 'openai',
@@ -364,21 +556,56 @@ describe('responses websocket transport', () => {
});
it('preserves incremental response.create payloads with previous_response_id for websocket-capable upstreams', async () => {
fetchMock
.mockResolvedValueOnce(createSseResponse([
'event: response.output_item.done\n',
'data: {"type":"response.output_item.done","output_index":0,"item":{"id":"fc_1","type":"function_call","call_id":"call_1"}}\n\n',
'event: response.output_item.done\n',
'data: {"type":"response.output_item.done","output_index":1,"item":{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}}\n\n',
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws_1","model":"gpt-5.4","status":"completed","output":[{"id":"fc_1","type":"function_call","call_id":"call_1"},{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"call tool"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]))
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_ws_2","model":"gpt-5.4","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"done"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket, _parsed, requestIndex) => {
if (requestIndex === 1) {
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws_1',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_1',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'call tool' }],
}],
usage: {
input_tokens: 3,
output_tokens: 1,
total_tokens: 4,
},
},
}));
return;
}
socket.send(JSON.stringify({
type: 'response.completed',
response: {
id: 'resp_ws_2',
model: 'gpt-5.4',
status: 'completed',
output: [{
id: 'msg_2',
type: 'message',
role: 'assistant',
status: 'completed',
content: [{ type: 'output_text', text: 'done' }],
}],
usage: {
input_tokens: 5,
output_tokens: 1,
total_tokens: 6,
},
},
}));
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -415,12 +642,92 @@ describe('responses websocket transport', () => {
await secondResponsePromise;
socket.close();
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamConnectionCount).toBe(1);
expect(upstreamRequests).toHaveLength(2);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
model: 'gpt-5.4',
instructions: 'be helpful',
});
expect(upstreamRequests[1]).toMatchObject({
type: 'response.create',
previous_response_id: 'resp_ws_1',
model: 'gpt-5.4',
instructions: 'be helpful',
input: [
{
id: 'tool_out_1',
type: 'function_call_output',
call_id: 'call_1',
output: 'tool result',
},
],
});
});
it('disables codex websocket incremental transport when the selected account marks websockets as disabled', async () => {
const selectedChannel = createSelectedChannel({
extraConfig: JSON.stringify({
credentialMode: 'session',
websockets: false,
oauth: {
provider: 'codex',
accountId: 'chatgpt-account-123',
email: 'codex-user@example.com',
},
}),
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
fetchMock
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_1","model":"gpt-5.4","status":"completed","output":[{"id":"msg_1","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"first"}]}],"usage":{"input_tokens":3,"output_tokens":1,"total_tokens":4}}}\n\n',
'data: [DONE]\n\n',
]))
.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_http_2","model":"gpt-5.4","status":"completed","output":[{"id":"msg_2","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"second"}]}],"usage":{"input_tokens":5,"output_tokens":1,"total_tokens":6}}}\n\n',
'data: [DONE]\n\n',
]));
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const firstResponsePromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
instructions: 'be helpful',
input: [],
}));
const firstMessages = await firstResponsePromise;
const secondResponsePromise = waitForSocketMessages(socket, 1);
socket.send(JSON.stringify({
type: 'response.create',
model: 'gpt-5.4',
previous_response_id: 'resp_http_1',
input: [
{
id: 'tool_out_1',
type: 'function_call_output',
call_id: 'call_1',
output: 'tool result',
},
],
}));
await secondResponsePromise;
socket.close();
expect(firstMessages[0]?.type).toBe('response.completed');
expect(upstreamConnectionCount).toBe(0);
expect(fetchMock).toHaveBeenCalledTimes(2);
const [, secondOptions] = fetchMock.mock.calls[1] as [string, RequestInit];
const secondBody = JSON.parse(String(secondOptions.body));
expect(secondBody.previous_response_id).toBe('resp_ws_1');
expect(secondBody.model).toBe('gpt-5.4');
expect(secondBody.instructions).toBe('be helpful');
expect(secondBody.previous_response_id).toBeUndefined();
expect(secondBody.input).toEqual([
{
id: 'tool_out_1',
@@ -490,11 +797,11 @@ describe('responses websocket transport', () => {
});
it('forwards generate=false upstream for websocket-capable upstreams instead of synthesizing prewarm events', async () => {
fetchMock.mockResolvedValueOnce(createSseResponse([
'event: response.completed\n',
'data: {"type":"response.completed","response":{"id":"resp_capable_generate_false","model":"gpt-5.4","status":"completed","output":[],"usage":{"input_tokens":0,"output_tokens":0,"total_tokens":0}}}\n\n',
'data: [DONE]\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
@@ -510,23 +817,42 @@ describe('responses websocket transport', () => {
socket.close();
expect(messages.map((message) => message.type)).toEqual(['response.completed']);
expect(fetchMock).toHaveBeenCalledTimes(1);
const [, options] = fetchMock.mock.calls[0] as [string, RequestInit];
const forwardedBody = JSON.parse(String(options.body));
expect(forwardedBody.generate).toBe(false);
expect(fetchMock).toHaveBeenCalledTimes(0);
expect(upstreamRequests[0]).toMatchObject({
type: 'response.create',
generate: false,
});
});
it('emits websocket error when the upstream stream closes before a terminal responses event', async () => {
fetchMock.mockResolvedValue(createSseResponse([
'event: response.created\n',
'data: {"type":"response.created","response":{"id":"resp_incomplete","model":"gpt-5.4","created_at":1706000000,"status":"in_progress","output":[]}}\n\n',
'event: response.output_text.delta\n',
'data: {"type":"response.output_text.delta","output_index":0,"item_id":"msg_ws","delta":"partial"}\n\n',
]));
const selectedChannel = createSelectedChannel({
siteUrl: upstreamSiteUrl,
});
selectChannelMock.mockReturnValue(selectedChannel);
previewSelectedChannelMock.mockResolvedValue(selectedChannel);
upstreamMessageHandler = (socket) => {
socket.send(JSON.stringify({
type: 'response.created',
response: {
id: 'resp_incomplete',
model: 'gpt-5.4',
created_at: 1706000000,
status: 'in_progress',
output: [],
},
}));
socket.send(JSON.stringify({
type: 'response.output_text.delta',
output_index: 0,
item_id: 'msg_ws',
delta: 'partial',
}));
socket.close();
};
const socket = new WebSocket(`${baseUrl}/v1/responses`);
await waitForSocketOpen(socket);
const messagesPromise = waitForSocketMessages(socket, 6, 400);
const messagesPromise = waitForSocketMessages(socket, 3, 400);
socket.send(JSON.stringify({
type: 'response.create',
@@ -546,12 +872,8 @@ describe('responses websocket transport', () => {
expect(messages.map((message) => message.type)).toEqual([
'response.created',
'response.output_text.delta',
'response.output_text.done',
'response.content_part.done',
'response.output_item.done',
'response.failed',
'error',
]);
expect(messages[5]?.response?.status).toBe('failed');
expect(messages[5]?.error?.message).toContain('stream closed before response.completed');
expect(messages[2]?.error?.message).toContain('stream closed before response.completed');
});
});
src/server/routes/proxy/responsesWebsocket.ts
+259 -56
View File
@@ -2,13 +2,19 @@ import { randomUUID } from 'node:crypto';
import type { FastifyInstance } from 'fastify';
import type { IncomingMessage } from 'node:http';
import { WebSocketServer, type RawData, type WebSocket } from 'ws';
import { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } from '../../proxy-core/runtime/codexWebsocketRuntime.js';
import { tokenRouter } from '../../services/tokenRouter.js';
import { buildOauthProviderHeaders } from '../../services/oauth/service.js';
import { openAiResponsesTransformer } from '../../transformers/openai/responses/index.js';
import { buildUpstreamEndpointRequest } from './upstreamEndpoint.js';
const installedApps = new WeakSet<FastifyInstance>();
const WS_TURN_STATE_HEADER = 'x-codex-turn-state';
const RESPONSES_WEBSOCKET_MODE_HEADER = 'x-metapi-responses-websocket-mode';
const RESPONSES_WEBSOCKET_TRANSPORT_HEADER = 'x-metapi-responses-websocket-transport';
const codexWebsocketRuntime = createCodexWebsocketRuntime();
type SelectedChannel = NonNullable<Awaited<ReturnType<typeof tokenRouter.selectChannel>>>;
type NormalizedResponsesWebsocketRequest =
| {
@@ -42,6 +48,93 @@ function headerValueToTrimmedString(value: unknown): string {
return '';
}
function toBooleanLike(value: unknown): boolean | null {
if (typeof value === 'boolean') return value;
if (typeof value === 'string') {
const normalized = value.trim().toLowerCase();
if (normalized === 'true' || normalized === '1' || normalized === 'yes' || normalized === 'on') return true;
if (normalized === 'false' || normalized === '0' || normalized === 'no' || normalized === 'off') return false;
}
return null;
}
function parseExtraConfigRecord(extraConfig: unknown): Record<string, unknown> | null {
if (typeof extraConfig !== 'string') return null;
try {
const parsed = JSON.parse(extraConfig);
return isRecord(parsed) ? parsed : null;
} catch {
return null;
}
}
function readNestedRecord(value: unknown, key: string): Record<string, unknown> | null {
if (!isRecord(value)) return null;
const nested = value[key];
return isRecord(nested) ? nested : null;
}
function selectedChannelModelMatches(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const selectedModel = asTrimmedString(selectedChannel.actualModel).toLowerCase();
const normalizedRequestModel = asTrimmedString(requestModel).toLowerCase();
if (!selectedModel || !normalizedRequestModel) return true;
return selectedModel === normalizedRequestModel;
}
function selectedChannelSupportsCodexWebsocketTransport(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const platform = asTrimmedString(selectedChannel.site?.platform).toLowerCase();
if (platform !== 'codex') return false;
if (!selectedChannelModelMatches(selectedChannel, requestModel)) return false;
const extraConfig = parseExtraConfigRecord(selectedChannel.account.extraConfig);
const oauth = readNestedRecord(extraConfig, 'oauth');
const providerData = readNestedRecord(oauth, 'providerData');
const candidateFlags = [
extraConfig?.websockets,
readNestedRecord(extraConfig, 'attributes')?.websockets,
readNestedRecord(extraConfig, 'metadata')?.websockets,
providerData?.websockets,
readNestedRecord(providerData, 'attributes')?.websockets,
readNestedRecord(providerData, 'metadata')?.websockets,
];
for (const candidate of candidateFlags) {
const parsed = toBooleanLike(candidate);
if (parsed !== null) return parsed;
}
return true;
}
function selectedChannelSupportsIncrementalInput(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
return selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel);
}
function shouldReuseSelectedChannel(
selectedChannel: SelectedChannel | null,
requestModel: string,
): boolean {
if (!selectedChannel) return false;
const selectedModel = asTrimmedString(selectedChannel.actualModel).toLowerCase();
const normalizedRequestModel = asTrimmedString(requestModel).toLowerCase();
if (!selectedModel || !normalizedRequestModel) return true;
return selectedModel === normalizedRequestModel;
}
function deriveCodexExplicitSessionId(body: Record<string, unknown>, sessionId: string): string {
const promptCacheKey = asTrimmedString(body.prompt_cache_key);
return promptCacheKey || sessionId;
}
function parseJsonObject(raw: RawData): Record<string, unknown> | null {
try {
const parsed = JSON.parse(String(raw));
@@ -232,6 +325,77 @@ function collectResponsesOutput(payloads: unknown[]): unknown[] {
.map(([, value]) => value);
}
async function forwardResponsesRequestViaHttp(input: {
app: FastifyInstance;
socket: WebSocket;
request: IncomingMessage;
payload: Record<string, unknown>;
preserveIncrementalMode: boolean;
}): Promise<unknown[] | null> {
const response = await input.app.inject({
method: 'POST',
url: '/v1/responses',
headers: {
...buildInjectHeaders(input.request),
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
...(input.preserveIncrementalMode ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
},
payload: input.payload,
});
if (response.statusCode < 200 || response.statusCode >= 300) {
let payload: unknown = null;
try {
payload = JSON.parse(response.body);
} catch {
payload = null;
}
writeResponsesWebsocketError(
input.socket,
response.statusCode,
response.statusMessage || 'Upstream error',
payload,
);
return null;
}
const contentType = String(response.headers['content-type'] || '').toLowerCase();
if (!contentType.includes('text/event-stream')) {
try {
const payload = JSON.parse(response.body);
input.socket.send(JSON.stringify(payload));
return isRecord(payload?.response) && Array.isArray(payload.response.output)
? cloneJsonObject(payload.response.output)
: [];
} catch {
writeResponsesWebsocketError(input.socket, 502, 'Unexpected non-JSON websocket proxy response');
return null;
}
}
const pulled = openAiResponsesTransformer.pullSseEvents(response.body);
const forwardedPayloads: unknown[] = [];
let sawTerminalPayload = false;
for (const event of pulled.events) {
if (event.data === '[DONE]') continue;
try {
const payload = JSON.parse(event.data);
forwardedPayloads.push(payload);
const type = isRecord(payload) ? asTrimmedString(payload.type) : '';
if (type === 'response.completed' || type === 'response.failed') {
sawTerminalPayload = true;
}
input.socket.send(JSON.stringify(payload));
} catch {
// Ignore malformed SSE frames; the HTTP route already normalizes them.
}
}
if (!sawTerminalPayload) {
writeResponsesWebsocketError(input.socket, 408, 'stream closed before response.completed');
}
return collectResponsesOutput(forwardedPayloads);
}
function buildInjectHeaders(request: IncomingMessage): Record<string, string | string[]> {
const headers: Record<string, string | string[]> = {};
for (const [rawKey, rawValue] of Object.entries(request.headers)) {
@@ -262,7 +426,7 @@ async function supportsResponsesWebsocketIncrementalInput(
try {
const selected = await tokenRouter.previewSelectedChannel(requestModel);
return asTrimmedString(selected?.site?.platform).toLowerCase() === 'codex';
return selectedChannelSupportsIncrementalInput(selected, requestModel);
} catch {
return false;
}
@@ -273,8 +437,16 @@ async function handleResponsesWebsocketConnection(
socket: WebSocket,
request: IncomingMessage,
) {
const websocketSessionId = headerValueToTrimmedString(request.headers['session_id'])
|| headerValueToTrimmedString(request.headers['session-id'])
|| randomUUID();
let lastRequest: Record<string, unknown> | null = null;
let lastResponseOutput: unknown[] = [];
let selectedChannel: SelectedChannel | null = null;
socket.once('close', () => {
void codexWebsocketRuntime.closeSession(websocketSessionId);
});
socket.on('message', async (raw) => {
const parsed = parseJsonObject(raw);
@@ -283,7 +455,9 @@ async function handleResponsesWebsocketConnection(
return;
}
const supportsIncrementalInput = await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
const requestModel = asTrimmedString(parsed.model) || asTrimmedString(lastRequest?.model);
const supportsIncrementalInput = selectedChannelSupportsIncrementalInput(selectedChannel, requestModel)
|| await supportsResponsesWebsocketIncrementalInput(parsed, lastRequest);
const shouldHandleLocalPrewarm = shouldHandleResponsesWebsocketPrewarmLocally(
parsed,
lastRequest,
@@ -309,67 +483,95 @@ async function handleResponsesWebsocketConnection(
return;
}
const response = await app.inject({
method: 'POST',
url: '/v1/responses',
headers: {
...buildInjectHeaders(request),
if (!shouldReuseSelectedChannel(selectedChannel, requestModel)) {
selectedChannel = requestModel
? await tokenRouter.selectChannel(requestModel)
: null;
}
const codexWebsocketChannel = selectedChannelSupportsCodexWebsocketTransport(selectedChannel, requestModel)
? selectedChannel
: null;
if (codexWebsocketChannel) {
const downstreamHeaders: Record<string, unknown> = {
...(request.headers as Record<string, unknown>),
[RESPONSES_WEBSOCKET_TRANSPORT_HEADER]: '1',
...(supportsIncrementalInput ? { [RESPONSES_WEBSOCKET_MODE_HEADER]: 'incremental' } : {}),
},
payload: normalized.request,
});
};
const providerHeaders = buildOauthProviderHeaders({
extraConfig: typeof codexWebsocketChannel.account.extraConfig === 'string'
? codexWebsocketChannel.account.extraConfig
: null,
downstreamHeaders,
});
const prepared = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: asTrimmedString(codexWebsocketChannel.actualModel) || requestModel,
stream: true,
tokenValue: codexWebsocketChannel.tokenValue,
sitePlatform: codexWebsocketChannel.site.platform,
siteUrl: codexWebsocketChannel.site.url,
openaiBody: normalized.request,
downstreamFormat: 'responses',
responsesOriginalBody: normalized.request,
downstreamHeaders,
providerHeaders,
codexExplicitSessionId: deriveCodexExplicitSessionId(normalized.request, websocketSessionId),
});
const requestUrl = `${codexWebsocketChannel.site.url.replace(/\/+$/, '')}${prepared.path}`;
if (response.statusCode < 200 || response.statusCode >= 300) {
let payload: unknown = null;
try {
payload = JSON.parse(response.body);
} catch {
payload = null;
}
writeResponsesWebsocketError(
socket,
response.statusCode,
response.statusMessage || 'Upstream error',
payload,
);
return;
}
const contentType = String(response.headers['content-type'] || '').toLowerCase();
if (!contentType.includes('text/event-stream')) {
try {
const payload = JSON.parse(response.body);
socket.send(JSON.stringify(payload));
lastResponseOutput = isRecord(payload?.response) && Array.isArray(payload.response.output)
? cloneJsonObject(payload.response.output)
: [];
} catch {
writeResponsesWebsocketError(socket, 502, 'Unexpected non-JSON websocket proxy response');
}
return;
}
const pulled = openAiResponsesTransformer.pullSseEvents(response.body);
const forwardedPayloads: unknown[] = [];
let sawTerminalPayload = false;
for (const event of pulled.events) {
if (event.data === '[DONE]') continue;
try {
const payload = JSON.parse(event.data);
forwardedPayloads.push(payload);
const type = isRecord(payload) ? asTrimmedString(payload.type) : '';
if (type === 'response.completed' || type === 'response.failed') {
sawTerminalPayload = true;
const runtimeResult = await codexWebsocketRuntime.sendRequest({
sessionId: websocketSessionId,
requestUrl,
headers: prepared.headers,
body: prepared.body,
});
lastResponseOutput = collectResponsesOutput(runtimeResult.events);
for (const payload of runtimeResult.events) {
socket.send(JSON.stringify(payload));
}
socket.send(JSON.stringify(payload));
} catch {
// Ignore malformed SSE frames; the HTTP route already normalizes them.
} catch (error) {
const runtimeError = error instanceof CodexWebsocketRuntimeError
? error
: new CodexWebsocketRuntimeError('upstream websocket request failed');
if (runtimeError.status === 426) {
const fallbackOutput = await forwardResponsesRequestViaHttp({
app,
socket,
request,
payload: normalized.request,
preserveIncrementalMode: false,
});
if (fallbackOutput) {
lastResponseOutput = fallbackOutput;
}
return;
}
lastResponseOutput = collectResponsesOutput(runtimeError.events);
for (const payload of runtimeError.events) {
socket.send(JSON.stringify(payload));
}
writeResponsesWebsocketError(
socket,
runtimeError.status || 408,
runtimeError.message,
runtimeError.payload,
);
}
return;
}
lastResponseOutput = collectResponsesOutput(forwardedPayloads);
if (!sawTerminalPayload) {
writeResponsesWebsocketError(socket, 408, 'stream closed before response.completed');
const forwardedOutput = await forwardResponsesRequestViaHttp({
app,
socket,
request,
payload: normalized.request,
preserveIncrementalMode: supportsIncrementalInput,
});
if (forwardedOutput) {
lastResponseOutput = forwardedOutput;
}
});
}
@@ -394,6 +596,7 @@ export function ensureResponsesWebsocketTransport(app: FastifyInstance) {
});
app.addHook('onClose', async () => {
await codexWebsocketRuntime.closeAllSessions();
await new Promise<void>((resolve) => {
websocketServer.close(() => resolve());
});
src/server/routes/proxy/upstreamEndpoint.test.ts
+145 -3
View File
@@ -1,4 +1,5 @@
import { beforeEach, describe, expect, it, vi } from 'vitest';
import { config } from '../../config.js';
const fetchModelPricingCatalogMock = vi.fn(async (_arg?: unknown): Promise<any> => null);
@@ -32,6 +33,17 @@ describe('resolveUpstreamEndpointCandidates', () => {
beforeEach(() => {
fetchModelPricingCatalogMock.mockReset();
fetchModelPricingCatalogMock.mockResolvedValue(null);
(config as any).codexHeaderDefaults = {
userAgent: '',
betaFeatures: '',
};
(config as any).payloadRules = {
default: [],
defaultRaw: [],
override: [],
overrideRaw: [],
filter: [],
};
});
it('uses downstream-aligned endpoint priority for unknown platforms', async () => {
@@ -468,6 +480,136 @@ describe('buildUpstreamEndpointRequest', () => {
expect(request.body.prompt_cache_key).toBe('codex-cache-123');
});
it('applies configured codex header defaults with CLIProxyAPI-compatible precedence', () => {
(config as any).codexHeaderDefaults = {
userAgent: 'codex-config-ua/1.0',
betaFeatures: 'multi_agent',
};
const websocketRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: true,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello codex' }],
},
downstreamFormat: 'openai',
downstreamHeaders: {
'x-metapi-responses-websocket-transport': '1',
},
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(websocketRequest.headers['User-Agent']).toBe('codex-config-ua/1.0');
expect(websocketRequest.headers['x-codex-beta-features']).toBe('multi_agent');
const clientHeaderRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: true,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello again codex' }],
},
downstreamFormat: 'openai',
downstreamHeaders: {
'x-metapi-responses-websocket-transport': '1',
'user-agent': 'client-ua/2.0',
'x-codex-beta-features': 'client-beta',
},
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(clientHeaderRequest.headers['User-Agent']).toBe('codex-config-ua/1.0');
expect(clientHeaderRequest.headers['x-codex-beta-features']).toBe('client-beta');
const httpRequest = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'plain http codex' }],
},
downstreamFormat: 'openai',
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(httpRequest.headers['x-codex-beta-features']).toBeUndefined();
});
it('applies configured payload rules before preparing codex responses requests', () => {
(config as any).payloadRules = {
default: [
{
models: [{ name: 'gpt-*', protocol: 'codex' }],
params: {
'reasoning.effort': 'high',
},
},
],
defaultRaw: [],
override: [
{
models: [{ name: 'gpt-5.4', protocol: 'codex' }],
params: {
'text.verbosity': 'low',
},
},
],
overrideRaw: [],
filter: [
{
models: [{ name: 'gpt-5.4', protocol: 'codex' }],
params: ['safety_identifier'],
},
],
};
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'gpt-5.4',
stream: false,
tokenValue: 'oauth-access-token',
oauthProvider: 'codex',
sitePlatform: 'codex',
siteUrl: 'https://chatgpt.com/backend-api/codex',
openaiBody: {
model: 'gpt-5.4',
messages: [{ role: 'user', content: 'hello codex' }],
verbosity: 'high',
safety_identifier: 'drop-me',
},
downstreamFormat: 'openai',
providerHeaders: {
Originator: 'codex_cli_rs',
},
} as any);
expect(request.body.reasoning).toEqual({ effort: 'high' });
expect(request.body.text).toEqual({ verbosity: 'low' });
expect(request.body.safety_identifier).toBeUndefined();
});
it('builds gemini-cli native requests with project envelope and bearer headers', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
@@ -850,7 +992,7 @@ describe('buildUpstreamEndpointRequest', () => {
]);
});
it('applies global responses standardization and drops non-standard fields', () => {
it('preserves unknown native responses fields while still normalizing known compatibility fields', () => {
const request = buildUpstreamEndpointRequest({
endpoint: 'responses',
modelName: 'upstream-gpt',
@@ -865,13 +1007,13 @@ describe('buildUpstreamEndpointRequest', () => {
input: 'hello',
metadata: { trace: 'abc123' },
max_completion_tokens: 512,
custom_vendor_flag: 'drop-me',
custom_vendor_flag: 'keep-me',
},
});
expect(request.path).toBe('/v1/responses');
expect(request.body.metadata).toEqual({ trace: 'abc123' });
expect(request.body.custom_vendor_flag).toBeUndefined();
expect(request.body.custom_vendor_flag).toBe('keep-me');
expect(request.body.max_completion_tokens).toBeUndefined();
expect(request.body.max_output_tokens).toBe(512);
expect(request.body.input).toEqual([
src/server/routes/proxy/upstreamEndpoint.ts
+58 -12
View File
@@ -1,6 +1,12 @@
import { createHash, randomUUID } from 'node:crypto';
import {
rankConversationFileEndpoints,
type ConversationFileInputSummary,
} from '../../proxy-core/capabilities/conversationFileCapabilities.js';
import { resolveProviderProfile } from '../../proxy-core/providers/registry.js';
import { config } from '../../config.js';
import { fetchModelPricingCatalog } from '../../services/modelPricingService.js';
import { applyPayloadRules } from '../../services/payloadRules.js';
import type { DownstreamFormat } from '../../transformers/shared/normalized.js';
import {
convertOpenAiBodyToResponsesBody as convertOpenAiBodyToResponsesBodyViaTransformer,
@@ -47,6 +53,20 @@ function asTrimmedString(value: unknown): string {
return typeof value === 'string' ? value.trim() : '';
}
function resolveRequestedModelForPayloadRules(input: {
modelName: string;
openaiBody: Record<string, unknown>;
claudeOriginalBody?: Record<string, unknown>;
responsesOriginalBody?: Record<string, unknown>;
}): string {
return (
asTrimmedString(input.responsesOriginalBody?.model)
|| asTrimmedString(input.claudeOriginalBody?.model)
|| asTrimmedString(input.openaiBody.model)
|| asTrimmedString(input.modelName)
);
}
function normalizePlatformName(platform: unknown): string {
return asTrimmedString(platform).toLowerCase();
}
@@ -626,6 +646,7 @@ export async function resolveUpstreamEndpointCandidates(
requestedModelHint?: string,
requestCapabilities?: {
hasNonImageFileInput?: boolean;
conversationFileSummary?: ConversationFileInputSummary;
wantsNativeResponsesReasoning?: boolean;
},
): Promise<UpstreamEndpoint[]> {
@@ -634,7 +655,13 @@ export async function resolveUpstreamEndpointCandidates(
isClaudeFamilyModel(modelName)
|| isClaudeFamilyModel(asTrimmedString(requestedModelHint))
);
const hasNonImageFileInput = requestCapabilities?.hasNonImageFileInput === true;
const conversationFileSummary = requestCapabilities?.conversationFileSummary ?? {
hasImage: false,
hasAudio: false,
hasDocument: requestCapabilities?.hasNonImageFileInput === true,
hasRemoteDocumentUrl: false,
};
const hasNonImageFileInput = conversationFileSummary.hasDocument;
const wantsNativeResponsesReasoning = requestCapabilities?.wantsNativeResponsesReasoning === true;
if (sitePlatform === 'anyrouter') {
// anyrouter deployments are effectively anthropic-protocol first.
@@ -659,8 +686,14 @@ export async function resolveUpstreamEndpointCandidates(
if (sitePlatform === 'claude') return ['messages'] as UpstreamEndpoint[];
if (sitePlatform === 'gemini') return ['responses', 'chat'] as UpstreamEndpoint[];
if (sitePlatform === 'gemini-cli' || sitePlatform === 'antigravity') return ['chat'] as UpstreamEndpoint[];
if (preferMessagesForClaudeModel) return ['messages', 'responses', 'chat'] as UpstreamEndpoint[];
return ['responses', 'messages', 'chat'] as UpstreamEndpoint[];
return rankConversationFileEndpoints({
sitePlatform,
requestedOrder: preferMessagesForClaudeModel
? ['messages', 'responses', 'chat']
: ['responses', 'messages', 'chat'],
summary: conversationFileSummary,
preferMessagesForClaudeModel,
});
})()
: preferred;
const prioritizedPreferredEndpoints: UpstreamEndpoint[] = (
@@ -885,6 +918,16 @@ export function buildUpstreamEndpointRequest(input: {
stream: input.stream,
oauthProjectId: asTrimmedString(input.oauthProjectId) || null,
};
const requestedModelForPayloadRules = resolveRequestedModelForPayloadRules(input);
const applyConfiguredPayloadRules = <T extends Record<string, unknown>>(body: T): T => (
applyPayloadRules({
rules: config.payloadRules,
payload: body,
modelName: input.modelName,
requestedModel: requestedModelForPayloadRules,
protocol: sitePlatform,
}) as T
);
if (isInternalGeminiUpstream) {
const instructions = (
@@ -898,6 +941,7 @@ export function buildUpstreamEndpointRequest(input: {
modelName: input.modelName,
instructions,
});
const configuredGeminiRequest = applyConfiguredPayloadRules(geminiRequest);
if (!providerProfile) {
throw new Error(`missing provider profile for platform: ${sitePlatform}`);
}
@@ -911,7 +955,7 @@ export function buildUpstreamEndpointRequest(input: {
sitePlatform,
baseHeaders: commonHeaders,
providerHeaders: input.providerHeaders,
body: geminiRequest,
body: configuredGeminiRequest,
action: input.stream ? 'streamGenerateContent' : 'generateContent',
});
}
@@ -952,6 +996,7 @@ export function buildUpstreamEndpointRequest(input: {
?? sanitizeAnthropicMessagesBody(
convertOpenAiBodyToAnthropicMessagesBody(openaiBody, input.modelName, input.stream),
);
const configuredClaudeBody = applyConfiguredPayloadRules(sanitizedBody);
if (providerProfile?.id === 'claude') {
return providerProfile.prepareRequest({
@@ -964,7 +1009,7 @@ export function buildUpstreamEndpointRequest(input: {
sitePlatform,
baseHeaders: commonHeaders,
claudeHeaders,
body: sanitizedBody,
body: configuredClaudeBody,
});
}
@@ -980,7 +1025,7 @@ export function buildUpstreamEndpointRequest(input: {
return {
path: resolveEndpointPath('messages'),
headers,
body: sanitizedBody,
body: configuredClaudeBody,
runtime,
};
}
@@ -1015,6 +1060,7 @@ export function buildUpstreamEndpointRequest(input: {
),
sitePlatform,
);
const configuredResponsesBody = applyConfiguredPayloadRules(body);
if (providerProfile?.id === 'codex') {
return providerProfile.prepareRequest({
@@ -1032,7 +1078,7 @@ export function buildUpstreamEndpointRequest(input: {
providerHeaders: input.providerHeaders,
codexSessionCacheKey: input.codexSessionCacheKey,
codexExplicitSessionId: input.codexExplicitSessionId,
body,
body: configuredResponsesBody,
});
}
@@ -1055,15 +1101,15 @@ export function buildUpstreamEndpointRequest(input: {
: null;
const shouldInjectDerivedPromptCacheKey = sitePlatform === 'codex'
&& !!codexSessionId
&& !asTrimmedString((body as Record<string, unknown>).prompt_cache_key)
&& !asTrimmedString((configuredResponsesBody as Record<string, unknown>).prompt_cache_key)
&& !asTrimmedString(input.codexExplicitSessionId)
&& !!asTrimmedString(input.codexSessionCacheKey);
const runtimeBody = shouldInjectDerivedPromptCacheKey
? {
...body,
...configuredResponsesBody,
prompt_cache_key: codexSessionId,
}
: body;
: configuredResponsesBody;
return {
path: resolveEndpointPath('responses'),
@@ -1077,11 +1123,11 @@ export function buildUpstreamEndpointRequest(input: {
return {
path: resolveEndpointPath('chat'),
headers,
body: {
body: applyConfiguredPayloadRules({
...openaiBody,
model: input.modelName,
stream: input.stream,
},
}),
runtime,
};
}