From 7e1fb8eb1be1fbd28d7cd42dfa6de750e8adb82d Mon Sep 17 00:00:00 2001 From: Cita Date: Wed, 25 Mar 2026 21:01:01 +0800 Subject: [PATCH] [codex] HTTP-first Codex upstream websocket settings (#280) * http-first codex upstream websocket settings * address websocket runtime review comments --------- Co-authored-by: apple --- src/server/config.ts | 1 + src/server/index.ts | 23 ++ .../runtime/codexWebsocketRuntime.test.ts | 237 ++++++++++++++++-- .../runtime/codexWebsocketRuntime.ts | 6 +- src/server/proxy-core/surfaces/chatSurface.ts | 2 + .../surfaces/openAiResponsesSurface.ts | 1 + .../proxy-core/surfaces/sharedSurface.ts | 6 +- src/server/routes/api/settings.events.test.ts | 35 +++ src/server/routes/api/settings.ts | 67 +++++ .../proxy/responses.codex-oauth.test.ts | 71 ++++++ .../routes/proxy/responses.websocket.test.ts | 19 +- src/server/routes/proxy/responsesWebsocket.ts | 45 +--- src/web/pages/Settings.tsx | 108 ++++++++ .../pages/settings.proxy-transport.test.tsx | 142 +++++++++++ 14 files changed, 677 insertions(+), 86 deletions(-) create mode 100644 src/web/pages/settings.proxy-transport.test.tsx diff --git a/src/server/config.ts b/src/server/config.ts index 5e5cb7c..d754fcc 100644 --- a/src/server/config.ts +++ b/src/server/config.ts @@ -115,6 +115,7 @@ export function buildConfig(env: NodeJS.ProcessEnv) { proxySessionChannelQueueWaitMs: Math.max(0, Math.trunc(parseNumber(env.PROXY_SESSION_CHANNEL_QUEUE_WAIT_MS, 1_500))), proxySessionChannelLeaseTtlMs: Math.max(5_000, Math.trunc(parseNumber(env.PROXY_SESSION_CHANNEL_LEASE_TTL_MS, 90_000))), proxySessionChannelLeaseKeepaliveMs: Math.max(1_000, Math.trunc(parseNumber(env.PROXY_SESSION_CHANNEL_LEASE_KEEPALIVE_MS, 15_000))), + codexUpstreamWebsocketEnabled: parseBoolean(env.CODEX_UPSTREAM_WEBSOCKET_ENABLED, false), proxyLogRetentionDays: Math.max(0, Math.trunc(parseNumber(env.PROXY_LOG_RETENTION_DAYS, 30))), proxyLogRetentionPruneIntervalMinutes: Math.max(1, Math.trunc(parseNumber(env.PROXY_LOG_RETENTION_PRUNE_INTERVAL_MINUTES, 30))), proxyFileRetentionDays: Math.max(0, Math.trunc(parseNumber(env.PROXY_FILE_RETENTION_DAYS, 30))), diff --git a/src/server/index.ts b/src/server/index.ts index 1a6f764..70c4a77 100644 --- a/src/server/index.ts +++ b/src/server/index.ts @@ -137,6 +137,11 @@ function applyRuntimeSettings(settingsMap: Map) { const systemProxyUrl = parseSettingFromMap(settingsMap, 'system_proxy_url'); if (typeof systemProxyUrl === 'string') config.systemProxyUrl = systemProxyUrl; + const codexUpstreamWebsocketEnabled = parseSettingFromMap(settingsMap, 'codex_upstream_websocket_enabled'); + if (typeof codexUpstreamWebsocketEnabled === 'boolean') { + config.codexUpstreamWebsocketEnabled = codexUpstreamWebsocketEnabled; + } + const proxyErrorKeywords = parseSettingFromMap(settingsMap, 'proxy_error_keywords'); if (proxyErrorKeywords !== undefined) { config.proxyErrorKeywords = toStringList(proxyErrorKeywords); @@ -203,6 +208,24 @@ function applyRuntimeSettings(settingsMap: Map) { config.logCleanupRetentionDays = normalizeLogCleanupRetentionDays(logCleanupRetentionDays); } + const proxySessionChannelConcurrencyLimit = parseSettingFromMap(settingsMap, 'proxy_session_channel_concurrency_limit'); + if ( + typeof proxySessionChannelConcurrencyLimit === 'number' + && Number.isFinite(proxySessionChannelConcurrencyLimit) + && proxySessionChannelConcurrencyLimit >= 0 + ) { + config.proxySessionChannelConcurrencyLimit = Math.trunc(proxySessionChannelConcurrencyLimit); + } + + const proxySessionChannelQueueWaitMs = parseSettingFromMap(settingsMap, 'proxy_session_channel_queue_wait_ms'); + if ( + typeof proxySessionChannelQueueWaitMs === 'number' + && Number.isFinite(proxySessionChannelQueueWaitMs) + && proxySessionChannelQueueWaitMs >= 0 + ) { + config.proxySessionChannelQueueWaitMs = Math.trunc(proxySessionChannelQueueWaitMs); + } + const routingWeights = parseSettingFromMap>(settingsMap, 'routing_weights'); if (routingWeights && typeof routingWeights === 'object') { config.routingWeights = { diff --git a/src/server/proxy-core/runtime/codexWebsocketRuntime.test.ts b/src/server/proxy-core/runtime/codexWebsocketRuntime.test.ts index 228c9fb..5630e97 100644 --- a/src/server/proxy-core/runtime/codexWebsocketRuntime.test.ts +++ b/src/server/proxy-core/runtime/codexWebsocketRuntime.test.ts @@ -139,7 +139,7 @@ describe('codexWebsocketRuntime', () => { await runtime.closeSession('exec-session-close'); }); - it('treats response.incomplete as a terminal websocket failure instead of waiting for socket close', async () => { + it('returns response.incomplete as a terminal websocket event without rejecting the session turn', async () => { upstreamMessageHandler = (socket, parsed) => { socket.send(JSON.stringify({ type: 'response.incomplete', @@ -154,37 +154,224 @@ describe('codexWebsocketRuntime', () => { })); }; - const { createCodexWebsocketRuntime, CodexWebsocketRuntimeError } = await import('./codexWebsocketRuntime.js'); + const { createCodexWebsocketRuntime } = await import('./codexWebsocketRuntime.js'); const runtime = createCodexWebsocketRuntime(); - let error: unknown; - try { - await runtime.sendRequest({ - sessionId: 'exec-session-incomplete', - requestUrl: upstreamWsUrl, - headers: { - Authorization: 'Bearer oauth-access-token', - 'OpenAI-Beta': 'responses_websockets=2026-02-06', - }, - body: { - model: 'gpt-5.4', - input: [], - }, - }); - } catch (caught) { - error = caught; - } - - expect(error).toBeInstanceOf(CodexWebsocketRuntimeError); - expect(error).toMatchObject({ - message: 'max_output_tokens', - status: 502, + const result = await runtime.sendRequest({ + sessionId: 'exec-session-incomplete', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + input: [], + }, }); - expect((error as CodexWebsocketRuntimeError).events).toEqual([ + + expect(result.events).toEqual([ expect.objectContaining({ type: 'response.incomplete', }), ]); + expect(result.reusedSession).toBe(false); + + await runtime.closeSession('exec-session-incomplete'); + }); + + it('keeps the upstream websocket session alive across response.failed terminal turns', async () => { + upstreamMessageHandler = (socket, parsed, requestIndex) => { + if (requestIndex === 1) { + socket.send(JSON.stringify({ + type: 'response.failed', + response: { + id: 'resp-failed', + model: parsed.model || 'gpt-5.4', + status: 'failed', + error: { + message: 'tool execution failed', + type: 'server_error', + }, + }, + })); + return; + } + + socket.send(JSON.stringify({ + type: 'response.completed', + response: { + id: `resp-${requestIndex}`, + object: 'response', + model: parsed.model || 'gpt-5.4', + status: 'completed', + output: [], + usage: { + input_tokens: 1, + output_tokens: 1, + total_tokens: 2, + }, + }, + })); + }; + + const { createCodexWebsocketRuntime } = await import('./codexWebsocketRuntime.js'); + const runtime = createCodexWebsocketRuntime(); + + const first = await runtime.sendRequest({ + sessionId: 'exec-session-failed-turn', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + input: [], + }, + }); + + const second = await runtime.sendRequest({ + sessionId: 'exec-session-failed-turn', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + previous_response_id: 'resp-failed', + input: [], + }, + }); + + expect(first.events).toEqual([ + expect.objectContaining({ + type: 'response.failed', + response: expect.objectContaining({ + id: 'resp-failed', + error: expect.objectContaining({ + message: 'tool execution failed', + }), + }), + }), + ]); + expect(first.reusedSession).toBe(false); + expect(second.events[0]).toMatchObject({ + type: 'response.completed', + response: { id: 'resp-2' }, + }); + expect(second.reusedSession).toBe(true); + expect(upstreamConnectionCount).toBe(1); + expect(upstreamRequests).toHaveLength(2); + + await runtime.closeSession('exec-session-failed-turn'); + }); + + it('fails the current turn and opens a fresh websocket on the next turn when a reused session closes before yielding any events', async () => { + upstreamMessageHandler = (socket, parsed, requestIndex) => { + if (requestIndex === 1) { + socket.send(JSON.stringify({ + type: 'response.completed', + response: { + id: 'resp-1', + object: 'response', + model: parsed.model || 'gpt-5.4', + status: 'completed', + output: [], + usage: { + input_tokens: 1, + output_tokens: 1, + total_tokens: 2, + }, + }, + })); + return; + } + + if (requestIndex === 2) { + socket.close(); + return; + } + + socket.send(JSON.stringify({ + type: 'response.completed', + response: { + id: 'resp-3', + object: 'response', + model: parsed.model || 'gpt-5.4', + status: 'completed', + output: [], + usage: { + input_tokens: 1, + output_tokens: 1, + total_tokens: 2, + }, + }, + })); + }; + + const { createCodexWebsocketRuntime } = await import('./codexWebsocketRuntime.js'); + const runtime = createCodexWebsocketRuntime(); + + await runtime.sendRequest({ + sessionId: 'exec-session-retry-stale', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + input: [], + }, + }); + + await expect(runtime.sendRequest({ + sessionId: 'exec-session-retry-stale', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + previous_response_id: 'resp-1', + input: [], + }, + })).rejects.toThrow('stream closed before response.completed'); + + const recovered = await runtime.sendRequest({ + sessionId: 'exec-session-retry-stale', + requestUrl: upstreamWsUrl, + headers: { + Authorization: 'Bearer oauth-access-token', + 'OpenAI-Beta': 'responses_websockets=2026-02-06', + }, + body: { + model: 'gpt-5.4', + previous_response_id: 'resp-1', + input: [], + }, + }); + + expect(recovered.events[0]).toMatchObject({ + type: 'response.completed', + response: { id: 'resp-3' }, + }); + expect(recovered.reusedSession).toBe(false); + expect(upstreamConnectionCount).toBe(2); + expect(upstreamRequests).toHaveLength(3); + expect(upstreamRequests[1]).toMatchObject({ + type: 'response.create', + previous_response_id: 'resp-1', + }); + expect(upstreamRequests[2]).toMatchObject({ + type: 'response.create', + previous_response_id: 'resp-1', + }); + + await runtime.closeSession('exec-session-retry-stale'); }); it('treats top-level error frames as terminal websocket failures', async () => { diff --git a/src/server/proxy-core/runtime/codexWebsocketRuntime.ts b/src/server/proxy-core/runtime/codexWebsocketRuntime.ts index c5186e0..4e1eba0 100644 --- a/src/server/proxy-core/runtime/codexWebsocketRuntime.ts +++ b/src/server/proxy-core/runtime/codexWebsocketRuntime.ts @@ -29,9 +29,9 @@ function isTerminalEvent(payload: Record): boolean { || type === 'error'; } -function isFailureTerminalEvent(payload: Record): boolean { +function isRuntimeErrorEvent(payload: Record): boolean { const type = asTrimmedString(payload.type); - return type === 'response.failed' || type === 'response.incomplete' || type === 'error'; + return type === 'error'; } function asFiniteNumber(value: unknown): number | undefined { @@ -262,7 +262,7 @@ async function sendSessionRequest( events.push(parsed); if (!isTerminalEvent(parsed)) return; if (settled) return; - if (isFailureTerminalEvent(parsed)) { + if (isRuntimeErrorEvent(parsed)) { settled = true; cleanup(); clearSessionSocket(session, socket); diff --git a/src/server/proxy-core/surfaces/chatSurface.ts b/src/server/proxy-core/surfaces/chatSurface.ts index ade6cb9..255c36c 100644 --- a/src/server/proxy-core/surfaces/chatSurface.ts +++ b/src/server/proxy-core/surfaces/chatSurface.ts @@ -264,6 +264,7 @@ export async function handleChatSurfaceRequest( }; let startTime = Date.now(); const leaseResult = await acquireSurfaceChannelLease({ + stickySessionKey, selected, }); if (leaseResult.status === 'timeout') { @@ -789,6 +790,7 @@ export async function handleClaudeCountTokensSurfaceRequest( const oauth = getOauthInfoFromAccount(selected.account); const startTime = Date.now(); const leaseResult = await acquireSurfaceChannelLease({ + stickySessionKey, selected, }); if (leaseResult.status === 'timeout') { diff --git a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts index 17d05f8..a7ef658 100644 --- a/src/server/proxy-core/surfaces/openAiResponsesSurface.ts +++ b/src/server/proxy-core/surfaces/openAiResponsesSurface.ts @@ -380,6 +380,7 @@ export async function handleOpenAiResponsesSurfaceRequest( const startTime = Date.now(); const leaseResult = await acquireSurfaceChannelLease({ + stickySessionKey, selected, }); if (leaseResult.status === 'timeout') { diff --git a/src/server/proxy-core/surfaces/sharedSurface.ts b/src/server/proxy-core/surfaces/sharedSurface.ts index 177955e..5d4b9d7 100644 --- a/src/server/proxy-core/surfaces/sharedSurface.ts +++ b/src/server/proxy-core/surfaces/sharedSurface.ts @@ -187,13 +187,17 @@ export function clearSurfaceStickyChannel(input: { } export async function acquireSurfaceChannelLease(input: { + stickySessionKey?: string | null; selected: { channel: { id: number }; account?: { extraConfig?: string | null } | null; }; }) { return await proxyChannelCoordinator.acquireChannelLease({ - channelId: input.selected.channel.id, + // Only session-addressable requests should consume the guarded per-channel + // lease pool. Requests without a stable downstream session key should keep + // the pre-sticky-session parallel behavior instead of contending globally. + channelId: input.stickySessionKey ? input.selected.channel.id : 0, accountExtraConfig: input.selected.account?.extraConfig, }); } diff --git a/src/server/routes/api/settings.events.test.ts b/src/server/routes/api/settings.events.test.ts index 7d12aa2..a65c2da 100644 --- a/src/server/routes/api/settings.events.test.ts +++ b/src/server/routes/api/settings.events.test.ts @@ -52,6 +52,9 @@ describe('settings and auth events', () => { config.logCleanupUsageLogsEnabled = false; config.logCleanupProgramLogsEnabled = false; config.logCleanupRetentionDays = 30; + config.codexUpstreamWebsocketEnabled = false; + config.proxySessionChannelConcurrencyLimit = 2; + config.proxySessionChannelQueueWaitMs = 1500; config.routingFallbackUnitCost = 1; (config as any).telegramEnabled = false; (config as any).telegramApiBaseUrl = 'https://api.telegram.org'; @@ -111,6 +114,38 @@ describe('settings and auth events', () => { expect(savedInterval?.value).toBe(JSON.stringify(8)); }); + it('persists codex upstream websocket and session lease settings from runtime settings', async () => { + const updateResponse = await app.inject({ + method: 'PUT', + url: '/api/settings/runtime', + payload: { + codexUpstreamWebsocketEnabled: true, + proxySessionChannelConcurrencyLimit: 6, + proxySessionChannelQueueWaitMs: 4200, + }, + }); + + expect(updateResponse.statusCode).toBe(200); + const updated = updateResponse.json() as { + codexUpstreamWebsocketEnabled?: boolean; + proxySessionChannelConcurrencyLimit?: number; + proxySessionChannelQueueWaitMs?: number; + }; + expect(updated.codexUpstreamWebsocketEnabled).toBe(true); + expect(updated.proxySessionChannelConcurrencyLimit).toBe(6); + expect(updated.proxySessionChannelQueueWaitMs).toBe(4200); + expect(config.codexUpstreamWebsocketEnabled).toBe(true); + expect(config.proxySessionChannelConcurrencyLimit).toBe(6); + expect(config.proxySessionChannelQueueWaitMs).toBe(4200); + + const savedWebsocket = await db.select().from(schema.settings).where(eq(schema.settings.key, 'codex_upstream_websocket_enabled')).get(); + const savedConcurrency = await db.select().from(schema.settings).where(eq(schema.settings.key, 'proxy_session_channel_concurrency_limit')).get(); + const savedQueueWait = await db.select().from(schema.settings).where(eq(schema.settings.key, 'proxy_session_channel_queue_wait_ms')).get(); + expect(savedWebsocket?.value).toBe(JSON.stringify(true)); + expect(savedConcurrency?.value).toBe(JSON.stringify(6)); + expect(savedQueueWait?.value).toBe(JSON.stringify(4200)); + }); + it('returns current recognized admin IP in runtime settings response', async () => { const response = await app.inject({ method: 'GET', diff --git a/src/server/routes/api/settings.ts b/src/server/routes/api/settings.ts index 7c12cdf..6ae6646 100644 --- a/src/server/routes/api/settings.ts +++ b/src/server/routes/api/settings.ts @@ -38,6 +38,9 @@ type RoutingWeights = typeof config.routingWeights; interface RuntimeSettingsBody { proxyToken?: string; systemProxyUrl?: string; + codexUpstreamWebsocketEnabled?: boolean; + proxySessionChannelConcurrencyLimit?: number; + proxySessionChannelQueueWaitMs?: number; checkinCron?: string; checkinScheduleMode?: 'cron' | 'interval'; checkinIntervalHours?: number; @@ -381,6 +384,11 @@ function applyImportedSettingToRuntime(key: string, value: unknown) { config.systemProxyUrl = normalizeSiteProxyUrl(value) || ''; return; } + case 'codex_upstream_websocket_enabled': { + if (typeof value !== 'boolean') return; + config.codexUpstreamWebsocketEnabled = value; + return; + } case 'proxy_error_keywords': { try { config.proxyErrorKeywords = parseProxyErrorKeywords(value); @@ -389,6 +397,18 @@ function applyImportedSettingToRuntime(key: string, value: unknown) { } return; } + case 'proxy_session_channel_concurrency_limit': { + const limit = Number(value); + if (!Number.isFinite(limit) || limit < 0) return; + config.proxySessionChannelConcurrencyLimit = Math.trunc(limit); + return; + } + case 'proxy_session_channel_queue_wait_ms': { + const queueWaitMs = Number(value); + if (!Number.isFinite(queueWaitMs) || queueWaitMs < 0) return; + config.proxySessionChannelQueueWaitMs = Math.trunc(queueWaitMs); + return; + } case 'proxy_empty_content_fail_enabled': { try { config.proxyEmptyContentFailEnabled = parseBooleanFlag(value, '空内容判定失败开关'); @@ -557,6 +577,9 @@ function getRuntimeSettingsResponse(currentAdminIp = '') { logCleanupUsageLogsEnabled: config.logCleanupUsageLogsEnabled, logCleanupProgramLogsEnabled: config.logCleanupProgramLogsEnabled, logCleanupRetentionDays: config.logCleanupRetentionDays, + codexUpstreamWebsocketEnabled: config.codexUpstreamWebsocketEnabled, + proxySessionChannelConcurrencyLimit: config.proxySessionChannelConcurrencyLimit, + proxySessionChannelQueueWaitMs: config.proxySessionChannelQueueWaitMs, routingFallbackUnitCost: config.routingFallbackUnitCost, routingWeights: config.routingWeights, webhookUrl: config.webhookUrl, @@ -940,6 +963,50 @@ export async function settingsRoutes(app: FastifyInstance) { invalidateSiteProxyCache(); } + if (body.codexUpstreamWebsocketEnabled !== undefined) { + let nextValue = false; + try { + nextValue = parseBooleanFlag(body.codexUpstreamWebsocketEnabled, 'Codex 上游 WebSocket 开关'); + } catch (err: any) { + return reply.code(400).send({ + success: false, + message: err?.message || 'Codex 上游 WebSocket 开关格式无效', + }); + } + + if (nextValue !== config.codexUpstreamWebsocketEnabled) { + changedLabels.push('Codex 上游 WebSocket 默认策略'); + } + config.codexUpstreamWebsocketEnabled = nextValue; + upsertSetting('codex_upstream_websocket_enabled', config.codexUpstreamWebsocketEnabled); + } + + if (body.proxySessionChannelConcurrencyLimit !== undefined) { + const limit = Number(body.proxySessionChannelConcurrencyLimit); + if (!Number.isFinite(limit) || limit < 0) { + return reply.code(400).send({ success: false, message: '会话通道并发上限必须是大于等于 0 的整数' }); + } + const nextLimit = Math.trunc(limit); + if (nextLimit !== config.proxySessionChannelConcurrencyLimit) { + changedLabels.push(`会话通道并发上限（${config.proxySessionChannelConcurrencyLimit} -> ${nextLimit}）`); + } + config.proxySessionChannelConcurrencyLimit = nextLimit; + upsertSetting('proxy_session_channel_concurrency_limit', config.proxySessionChannelConcurrencyLimit); + } + + if (body.proxySessionChannelQueueWaitMs !== undefined) { + const rawQueueWaitMs = Number(body.proxySessionChannelQueueWaitMs); + if (!Number.isFinite(rawQueueWaitMs) || rawQueueWaitMs < 0) { + return reply.code(400).send({ success: false, message: '会话通道排队等待时间必须是大于等于 0 的整数毫秒' }); + } + const nextQueueWaitMs = Math.trunc(rawQueueWaitMs); + if (nextQueueWaitMs !== config.proxySessionChannelQueueWaitMs) { + changedLabels.push(`会话通道排队等待（${config.proxySessionChannelQueueWaitMs}ms -> ${nextQueueWaitMs}ms）`); + } + config.proxySessionChannelQueueWaitMs = nextQueueWaitMs; + upsertSetting('proxy_session_channel_queue_wait_ms', config.proxySessionChannelQueueWaitMs); + } + if (body.proxyErrorKeywords !== undefined) { let nextKeywords: string[] = []; try { diff --git a/src/server/routes/proxy/responses.codex-oauth.test.ts b/src/server/routes/proxy/responses.codex-oauth.test.ts index 3a51d92..7948f25 100644 --- a/src/server/routes/proxy/responses.codex-oauth.test.ts +++ b/src/server/routes/proxy/responses.codex-oauth.test.ts @@ -20,6 +20,9 @@ const refreshOauthAccessTokenSingleflightMock = vi.fn(); const recordOauthQuotaResetHintMock = vi.fn(); const insertedProxyLogs: Record[] = []; const originalProxyEmptyContentFailEnabled = config.proxyEmptyContentFailEnabled; +const originalProxyStickySessionEnabled = config.proxyStickySessionEnabled; +const originalProxySessionChannelConcurrencyLimit = config.proxySessionChannelConcurrencyLimit; +const originalProxySessionChannelQueueWaitMs = config.proxySessionChannelQueueWaitMs; const dbInsertMock = vi.fn((_arg?: any) => ({ values: (values: Record) => { insertedProxyLogs.push(values); @@ -136,6 +139,9 @@ describe('responses proxy codex oauth refresh', () => { beforeEach(() => { resetCodexHttpSessionQueue(); config.proxyEmptyContentFailEnabled = false; + config.proxyStickySessionEnabled = originalProxyStickySessionEnabled; + config.proxySessionChannelConcurrencyLimit = originalProxySessionChannelConcurrencyLimit; + config.proxySessionChannelQueueWaitMs = originalProxySessionChannelQueueWaitMs; fetchMock.mockReset(); selectChannelMock.mockReset(); selectNextChannelMock.mockReset(); @@ -180,6 +186,9 @@ describe('responses proxy codex oauth refresh', () => { afterAll(async () => { config.proxyEmptyContentFailEnabled = originalProxyEmptyContentFailEnabled; + config.proxyStickySessionEnabled = originalProxyStickySessionEnabled; + config.proxySessionChannelConcurrencyLimit = originalProxySessionChannelConcurrencyLimit; + config.proxySessionChannelQueueWaitMs = originalProxySessionChannelQueueWaitMs; await app.close(); }); @@ -561,6 +570,68 @@ describe('responses proxy codex oauth refresh', () => { expect(secondOptions.headers.Session_id || secondOptions.headers.session_id).toBe('session-http-serial-1'); }); + it('does not gate codex responses requests without a downstream session id behind the session lease queue', async () => { + config.proxyStickySessionEnabled = true; + config.proxySessionChannelConcurrencyLimit = 1; + config.proxySessionChannelQueueWaitMs = 20; + + const firstUpstream = createDeferred(); + fetchMock + .mockImplementationOnce(() => firstUpstream.promise) + .mockResolvedValueOnce(new Response(JSON.stringify({ + id: 'resp_codex_parallel_2', + object: 'response', + model: 'gpt-5.2-codex', + status: 'completed', + output_text: 'second request finished', + usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 }, + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const firstResponsePromise = app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2-codex', + input: 'first request', + }, + }); + + await waitFor(() => fetchMock.mock.calls.length === 1); + + const secondResponsePromise = app.inject({ + method: 'POST', + url: '/v1/responses', + payload: { + model: 'gpt-5.2-codex', + input: 'second request', + }, + }); + + await waitFor(() => fetchMock.mock.calls.length === 2); + + firstUpstream.resolve(new Response(JSON.stringify({ + id: 'resp_codex_parallel_1', + object: 'response', + model: 'gpt-5.2-codex', + status: 'completed', + output_text: 'first request finished', + usage: { input_tokens: 4, output_tokens: 2, total_tokens: 6 }, + }), { + status: 200, + headers: { 'content-type': 'application/json' }, + })); + + const [firstResponse, secondResponse] = await Promise.all([ + firstResponsePromise, + secondResponsePromise, + ]); + expect(firstResponse.statusCode).toBe(200); + expect(secondResponse.statusCode).toBe(200); + }); + it('records codex usage_limit_reached reset hints on upstream 429 failures', async () => { fetchMock.mockResolvedValue(new Response(JSON.stringify({ error: { diff --git a/src/server/routes/proxy/responses.websocket.test.ts b/src/server/routes/proxy/responses.websocket.test.ts index 615cfa8..72fe18c 100644 --- a/src/server/routes/proxy/responses.websocket.test.ts +++ b/src/server/routes/proxy/responses.websocket.test.ts @@ -248,6 +248,7 @@ function createClientSocketForPath(path: string, headers: Record describe('responses websocket transport', () => { const originalCodexResponsesWebsocketBeta = config.codexResponsesWebsocketBeta; + const originalCodexUpstreamWebsocketEnabled = config.codexUpstreamWebsocketEnabled; let app: FastifyInstance; let baseUrl: string; let upstreamServer: WebSocketServer; @@ -328,6 +329,7 @@ describe('responses websocket transport', () => { upstreamUpgradeHeaders = {}; upstreamRequests = []; (config as any).codexResponsesWebsocketBeta = originalCodexResponsesWebsocketBeta; + (config as any).codexUpstreamWebsocketEnabled = true; rejectedUpgradeStatus = 426; rejectedUpgradeStatusText = 'Upgrade Required'; rejectedUpgradeBody = 'Upgrade Required'; @@ -363,6 +365,7 @@ describe('responses websocket transport', () => { }); afterAll(async () => { + (config as any).codexUpstreamWebsocketEnabled = originalCodexUpstreamWebsocketEnabled; await new Promise((resolve) => rejectedUpgradeServer.close(() => resolve())); await new Promise((resolve) => upstreamServer.close(() => resolve())); await app.close(); @@ -1430,20 +1433,8 @@ describe('responses websocket transport', () => { }); }); - 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); + it('falls back to the HTTP responses route when codex upstream websocket is globally disabled', async () => { + (config as any).codexUpstreamWebsocketEnabled = false; fetchMock .mockResolvedValueOnce(createSseResponse([ 'event: response.completed\n', diff --git a/src/server/routes/proxy/responsesWebsocket.ts b/src/server/routes/proxy/responsesWebsocket.ts index 019645d..ff76dd7 100644 --- a/src/server/routes/proxy/responsesWebsocket.ts +++ b/src/server/routes/proxy/responsesWebsocket.ts @@ -14,6 +14,7 @@ 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'; +import { config } from '../../config.js'; const installedApps = new WeakSet(); const WS_TURN_STATE_HEADER = 'x-codex-turn-state'; @@ -56,32 +57,6 @@ 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 | 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 | null { - if (!isRecord(value)) return null; - const nested = value[key]; - return isRecord(nested) ? nested : null; -} - function selectedChannelModelMatches( selectedChannel: SelectedChannel | null, requestModel: string, @@ -101,23 +76,7 @@ function selectedChannelSupportsCodexWebsocketTransport( 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; + return !!config.codexUpstreamWebsocketEnabled; } function selectedChannelSupportsIncrementalInput( diff --git a/src/web/pages/Settings.tsx b/src/web/pages/Settings.tsx index 6d960c0..2f701de 100644 --- a/src/web/pages/Settings.tsx +++ b/src/web/pages/Settings.tsx @@ -49,6 +49,9 @@ type RuntimeSettings = { logCleanupUsageLogsEnabled: boolean; logCleanupProgramLogsEnabled: boolean; logCleanupRetentionDays: number; + codexUpstreamWebsocketEnabled: boolean; + proxySessionChannelConcurrencyLimit: number; + proxySessionChannelQueueWaitMs: number; routingFallbackUnitCost: number; routingWeights: RoutingWeights; systemProxyUrl: string; @@ -190,6 +193,9 @@ export default function Settings() { logCleanupUsageLogsEnabled: false, logCleanupProgramLogsEnabled: false, logCleanupRetentionDays: 30, + codexUpstreamWebsocketEnabled: false, + proxySessionChannelConcurrencyLimit: 2, + proxySessionChannelQueueWaitMs: 1500, routingFallbackUnitCost: 1, routingWeights: defaultWeights, systemProxyUrl: '', @@ -204,6 +210,7 @@ export default function Settings() { const [testingCheckin, setTestingCheckin] = useState(false); const [savingToken, setSavingToken] = useState(false); const [savingSystemProxy, setSavingSystemProxy] = useState(false); + const [savingProxyTransport, setSavingProxyTransport] = useState(false); const [testingSystemProxy, setTestingSystemProxy] = useState(false); const [systemProxyTestState, setSystemProxyTestState] = useState(null); const [savingProxyFailureRules, setSavingProxyFailureRules] = useState(false); @@ -412,6 +419,13 @@ export default function Settings() { logCleanupRetentionDays: Number(runtimeInfo.logCleanupRetentionDays) >= 1 ? Math.trunc(Number(runtimeInfo.logCleanupRetentionDays)) : 30, + codexUpstreamWebsocketEnabled: !!runtimeInfo.codexUpstreamWebsocketEnabled, + proxySessionChannelConcurrencyLimit: Number(runtimeInfo.proxySessionChannelConcurrencyLimit) >= 0 + ? Math.trunc(Number(runtimeInfo.proxySessionChannelConcurrencyLimit)) + : 2, + proxySessionChannelQueueWaitMs: Number(runtimeInfo.proxySessionChannelQueueWaitMs) >= 0 + ? Math.trunc(Number(runtimeInfo.proxySessionChannelQueueWaitMs)) + : 1500, routingFallbackUnitCost: Number(runtimeInfo.routingFallbackUnitCost) > 0 ? Number(runtimeInfo.routingFallbackUnitCost) : 1, @@ -569,6 +583,34 @@ export default function Settings() { } }; + const saveProxyTransportSettings = async () => { + setSavingProxyTransport(true); + try { + const res = await api.updateRuntimeSettings({ + codexUpstreamWebsocketEnabled: runtime.codexUpstreamWebsocketEnabled, + proxySessionChannelConcurrencyLimit: runtime.proxySessionChannelConcurrencyLimit, + proxySessionChannelQueueWaitMs: runtime.proxySessionChannelQueueWaitMs, + }); + setRuntime((prev) => ({ + ...prev, + codexUpstreamWebsocketEnabled: typeof res?.codexUpstreamWebsocketEnabled === 'boolean' + ? res.codexUpstreamWebsocketEnabled + : prev.codexUpstreamWebsocketEnabled, + proxySessionChannelConcurrencyLimit: Number(res?.proxySessionChannelConcurrencyLimit) >= 0 + ? Math.trunc(Number(res.proxySessionChannelConcurrencyLimit)) + : prev.proxySessionChannelConcurrencyLimit, + proxySessionChannelQueueWaitMs: Number(res?.proxySessionChannelQueueWaitMs) >= 0 + ? Math.trunc(Number(res.proxySessionChannelQueueWaitMs)) + : prev.proxySessionChannelQueueWaitMs, + })); + toast.success('传输与会话并发设置已保存'); + } catch (err: any) { + toast.error(err?.message || '保存失败'); + } finally { + setSavingProxyTransport(false); + } + }; + const testSystemProxy = async () => { const proxyUrl = runtime.systemProxyUrl.trim(); if (!proxyUrl) { @@ -1220,6 +1262,72 @@ export default function Settings() { +
+
Codex 上游传输与会话并发
+
+ 默认采用 HTTP 优先。只有这里开启后，metapi 才会在 Codex 请求上尝试把上游升级为 WebSocket。 + 下游 Codex 客户端也必须同时启用 `/v1/responses` websocket，单开这里不会生效。 +
+
+ 从旧版本升级时，原先账号 `extraConfig.websockets` 的行为不再单独生效；现在统一以这里的全局设置和下游客户端是否开启为准。 +
+ +
+
+
会话通道并发上限
+ { + const nextValue = Number(e.target.value); + setRuntime((prev) => ({ + ...prev, + proxySessionChannelConcurrencyLimit: Number.isFinite(nextValue) && nextValue >= 0 + ? Math.trunc(nextValue) + : prev.proxySessionChannelConcurrencyLimit, + })); + }} + style={inputStyle} + /> +
+
+
排队等待时间（毫秒）
+ { + const nextValue = Number(e.target.value); + setRuntime((prev) => ({ + ...prev, + proxySessionChannelQueueWaitMs: Number.isFinite(nextValue) && nextValue >= 0 + ? Math.trunc(nextValue) + : prev.proxySessionChannelQueueWaitMs, + })); + }} + style={inputStyle} + /> +
+
+
+ 这组 lease 只作用于能识别稳定 `session_id` 的会话型请求；没有稳定会话标识的普通请求不会进入这个池。 +
+
+ +
+
+
下游访问令牌（PROXY_TOKEN）
diff --git a/src/web/pages/settings.proxy-transport.test.tsx b/src/web/pages/settings.proxy-transport.test.tsx new file mode 100644 index 0000000..071fe6c --- /dev/null +++ b/src/web/pages/settings.proxy-transport.test.tsx @@ -0,0 +1,142 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { act, create, type ReactTestInstance, type ReactTestRenderer } from 'react-test-renderer'; +import { MemoryRouter } from 'react-router-dom'; +import { ToastProvider } from '../components/Toast.js'; +import Settings from './Settings.js'; + +const { apiMock } = vi.hoisted(() => ({ + apiMock: { + getAuthInfo: vi.fn(), + getRuntimeSettings: vi.fn(), + getDownstreamApiKeys: vi.fn(), + getRoutesLite: vi.fn(), + getRuntimeDatabaseConfig: vi.fn(), + getBrandList: vi.fn(), + updateRuntimeSettings: vi.fn(), + }, +})); + +vi.mock('../api.js', () => ({ + api: apiMock, +})); + +vi.mock('../components/BrandIcon.js', () => ({ + BrandGlyph: () => null, + InlineBrandIcon: () => null, + getBrand: () => null, + normalizeBrandIconKey: (icon: string) => icon, +})); + +function collectText(node: ReactTestInstance): string { + return (node.children || []).map((child) => { + if (typeof child === 'string') return child; + return collectText(child); + }).join(''); +} + +async function flushMicrotasks() { + await act(async () => { + await Promise.resolve(); + await Promise.resolve(); + }); +} + +describe('Settings proxy transport', () => { + beforeEach(() => { + vi.clearAllMocks(); + apiMock.getAuthInfo.mockResolvedValue({ masked: 'sk-****' }); + apiMock.getRuntimeSettings.mockResolvedValue({ + checkinCron: '0 8 * * *', + checkinScheduleMode: 'interval', + checkinIntervalHours: 6, + balanceRefreshCron: '0 * * * *', + logCleanupCron: '15 4 * * *', + logCleanupUsageLogsEnabled: true, + logCleanupProgramLogsEnabled: true, + logCleanupRetentionDays: 14, + codexUpstreamWebsocketEnabled: false, + proxySessionChannelConcurrencyLimit: 4, + proxySessionChannelQueueWaitMs: 3200, + routingFallbackUnitCost: 1, + routingWeights: {}, + adminIpAllowlist: [], + systemProxyUrl: '', + }); + apiMock.getDownstreamApiKeys.mockResolvedValue({ items: [] }); + apiMock.getRoutesLite.mockResolvedValue([]); + apiMock.getBrandList.mockResolvedValue({ brands: [] }); + apiMock.getRuntimeDatabaseConfig.mockResolvedValue({ + active: { dialect: 'sqlite', connection: '(default sqlite path)', ssl: false }, + saved: null, + restartRequired: false, + }); + apiMock.updateRuntimeSettings.mockResolvedValue({ + success: true, + codexUpstreamWebsocketEnabled: true, + proxySessionChannelConcurrencyLimit: 6, + proxySessionChannelQueueWaitMs: 4200, + }); + }); + + afterEach(() => { + vi.clearAllMocks(); + }); + + it('saves codex upstream websocket and session lease settings from the settings page', async () => { + let root!: ReactTestRenderer; + try { + await act(async () => { + root = create( + + + + + , + ); + }); + await flushMicrotasks(); + + const websocketToggleLabel = root.root.find((node) => ( + node.type === 'label' + && collectText(node).includes('允许 metapi 到 Codex 上游使用 WebSocket') + )); + const websocketToggle = websocketToggleLabel.findByType('input'); + expect(websocketToggle.props.checked).toBe(false); + + const concurrencyInput = root.root.find((node) => ( + node.type === 'input' + && node.props.type === 'number' + && node.props.value === 4 + )); + const queueWaitInput = root.root.find((node) => ( + node.type === 'input' + && node.props.type === 'number' + && node.props.value === 3200 + )); + + await act(async () => { + websocketToggle.props.onChange({ target: { checked: true } }); + concurrencyInput.props.onChange({ target: { value: '6' } }); + queueWaitInput.props.onChange({ target: { value: '4200' } }); + }); + + const saveButton = root.root.find((node) => ( + node.type === 'button' + && typeof node.props.onClick === 'function' + && collectText(node).trim() === '保存传输与并发' + )); + await act(async () => { + saveButton.props.onClick(); + }); + await flushMicrotasks(); + + expect(apiMock.updateRuntimeSettings).toHaveBeenCalledWith({ + codexUpstreamWebsocketEnabled: true, + proxySessionChannelConcurrencyLimit: 6, + proxySessionChannelQueueWaitMs: 4200, + }); + } finally { + root?.unmount(); + } + }); +});