[codex] HTTP-first Codex upstream websocket settings (#280)

* http-first codex upstream websocket settings

* address websocket runtime review comments

---------

Co-authored-by: apple <apple@appledeMacBook-Pro.local>

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',

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 {

src/server/routes/proxy/responses.codex-oauth.test.ts

@@ -20,6 +20,9 @@ const refreshOauthAccessTokenSingleflightMock = vi.fn();
 const recordOauthQuotaResetHintMock = vi.fn();
 const insertedProxyLogs: Record<string, unknown>[] = [];
 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<string, unknown>) => {
     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<Response>();
+    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: {

src/server/routes/proxy/responses.websocket.test.ts

@@ -248,6 +248,7 @@ function createClientSocketForPath(path: string, headers: Record<string, string>
 
 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<void>((resolve) => rejectedUpgradeServer.close(() => resolve()));
     await new Promise<void>((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',

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<FastifyInstance>();
 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<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,
@@ -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(