[codex] Add Codex WebSocket transport proxy

.gitignore

@@ -17,3 +17,4 @@ skills-lock.json
 videos/
 ccxray-init.png
 docs/src/
+.tmp/

CLAUDE.md

@@ -43,6 +43,8 @@ No build step. No linting. Restart to apply changes.
 | `server/routes/costs.js` | Cost budget endpoints |
 | `server/hub.js` | Multi-project hub: lockfile (`~/.ccxray/hub.json`), discovery (with orphan port probe fallback), client registration, idle shutdown (injectable via setOnShutdown), crash auto-recovery |
 | `server/auth.js` | API key auth middleware (enabled via `AUTH_TOKEN` env) |
+| `server/openai-session.js` | Shared OpenAI/Codex header + session helpers (session id extraction, agent type, turn-metadata sidecar) |
+| `server/ws-proxy.js` | OpenAI WebSocket transport proxy for `/v1/responses` and `/v1/realtime` upgrades. Tunables: `CCXRAY_WS_IDLE_TIMEOUT_MS` (default 60s), `CCXRAY_WS_MAX_QUEUE_BYTES` (default 4 MiB; caps client→upstream buffer while upstream is connecting) |
 | `server/storage/` | Storage adapters (local filesystem, S3/R2). `statShared()` for file mtime. `supportsDelta` flag gates delta-write eligibility |
 
 ### Client (`public/`)

server/config.js

@@ -137,6 +137,7 @@ function getUpstream(provider) {
 function getProviderForRequest(urlPath) {
   const pathname = (urlPath || '').split('?')[0];
   if (pathname === '/v1/responses' || pathname.startsWith('/v1/responses/')) return 'openai';
+  if (pathname === '/v1/realtime' || pathname.startsWith('/v1/realtime/')) return 'openai';
   if (pathname === '/v1/models' || pathname.startsWith('/v1/models/')) return 'openai';
   return 'anthropic';
 }

server/index.js

@@ -18,6 +18,13 @@ const { authMiddleware } = require('./auth');
 const { extractAgentType, extractPromptAgentType, splitB2IntoBlocks } = require('./system-prompt');
 const { findSharedPrefix } = require('./delta-helpers');
 const providers = require('./providers');
+const { handleWebSocketUpgrade } = require('./ws-proxy');
+const {
+  getCodexRawSessionId,
+  isOpenAISubagent,
+  detectOpenAISession,
+  withCodexMetadata,
+} = require('./openai-session');
 
 // ── CLI: parse flags and detect provider launchers ──
 const portIdx = process.argv.indexOf('--port');
@@ -135,53 +142,6 @@ function buildForwardHeaders(clientHeaders, upstream) {
   return fwdHeaders;
 }
 
-function getCodexRawSessionId() {
-  return 'codex-raw';
-}
-
-function firstHeader(headers, name) {
-  const value = headers?.[name.toLowerCase()];
-  return Array.isArray(value) ? value[0] : value;
-}
-
-function getCodexSessionId(headers, parsedBody) {
-  const direct = firstHeader(headers, 'session_id') || firstHeader(headers, 'x-openai-session-id');
-  if (direct) return String(direct);
-  return parsedBody?.metadata?.session_id || null;
-}
-
-function isOpenAISubagent(headers, parsedBody) {
-  const raw = firstHeader(headers, 'x-openai-subagent');
-  if (raw != null) {
-    const text = String(raw).toLowerCase();
-    return text !== '0' && text !== 'false' && text !== 'no';
-  }
-  return Boolean(parsedBody?.metadata?.is_subagent || parsedBody?.metadata?.isSubagent);
-}
-
-function getOpenAIAgentTypeFromHeaders(headers) {
-  const subagent = firstHeader(headers, 'x-openai-subagent');
-  const direct = firstHeader(headers, 'x-openai-agent-type') || firstHeader(headers, 'x-codex-agent-type');
-  const value = direct || subagent;
-  if (!value) return null;
-  const normalized = String(value).toLowerCase();
-  if (normalized === 'explorer' || normalized === 'worker' || normalized === 'default') return normalized;
-  return null;
-}
-
-function detectOpenAISession(headers, parsedBody) {
-  if (!parsedBody) return { sessionId: getCodexRawSessionId(), isNewSession: false, inferred: true };
-  if (!getCodexSessionId(headers, parsedBody)) {
-    return { sessionId: getCodexRawSessionId(), isNewSession: false, inferred: true };
-  }
-  const detected = store.detectSession(parsedBody);
-  return {
-    sessionId: detected.sessionId || getCodexRawSessionId(),
-    isNewSession: detected.isNewSession || false,
-    inferred: detected.inferred || false,
-  };
-}
-
 function getCodexCwdFallback() {
   return hub.lookupClientCwd() || (agentCommand === 'codex' ? process.cwd() : null);
 }
@@ -190,19 +150,6 @@ function getOpenAICwd(parsedBody) {
   return parsedBody?.metadata?.cwd || getCodexCwdFallback();
 }
 
-function withCodexMetadata(parsedBody, headers) {
-  if (!parsedBody || typeof parsedBody !== 'object') return parsedBody;
-  const sessionId = getCodexSessionId(headers, parsedBody);
-  const agentType = getOpenAIAgentTypeFromHeaders(headers);
-  if (!sessionId && !agentType) return parsedBody;
-  const metadata = parsedBody.metadata && typeof parsedBody.metadata === 'object'
-    ? { ...parsedBody.metadata }
-    : {};
-  if (sessionId && !metadata.session_id) metadata.session_id = sessionId;
-  if (agentType && !metadata.agent_type) metadata.agent_type = agentType;
-  return { ...parsedBody, metadata };
-}
-
 function registerPromptVersion({ provider, parsedBody, sharedFile, promptText, firstSeen, notify = true }) {
   if (!promptText) return null;
   const { key: agentKey, label: agentLabel } = extractPromptAgentType(provider, parsedBody);
@@ -317,6 +264,12 @@ const server = http.createServer((clientReq, clientRes) => {
             sharedFile: `openai_instructions_${sysHash}.json`,
             promptText: parsedBody.instructions,
           }) : null;
+          if (promptInfo) {
+            config.storage.writeSharedIfAbsent(`openai_prompt_meta_${sysHash}.json`, JSON.stringify({
+              agentKey: promptInfo.agentKey,
+              agentLabel: promptInfo.agentLabel,
+            })).catch(e => console.error('Write OpenAI prompt metadata failed:', e.message));
+          }
           coreHash = promptInfo?.coreHash || null;
         }
         if (toolsHash) config.storage.writeSharedIfAbsent(`openai_tools_${toolsHash}.json`, JSON.stringify(parsedBody.tools))
@@ -452,6 +405,10 @@ const server = http.createServer((clientReq, clientRes) => {
   });
 });
 
+server.on('upgrade', (req, socket, head) => {
+  handleWebSocketUpgrade(req, socket, head);
+});
+
 
 // ── Spawn agent CLI with proxy routing ──
 function spawnAgent(command, port, args, onExit) {

server/openai-session.js

@@ -0,0 +1,94 @@
+'use strict';
+
+const store = require('./store');
+
+function getCodexRawSessionId() {
+  return 'codex-raw';
+}
+
+function firstHeader(headers, name) {
+  const value = headers?.[name.toLowerCase()];
+  return Array.isArray(value) ? value[0] : value;
+}
+
+function parseCodexTurnMetadata(headers) {
+  const raw = firstHeader(headers, 'x-codex-turn-metadata');
+  if (!raw) return null;
+  try {
+    const parsed = JSON.parse(String(raw));
+    return parsed && typeof parsed === 'object' ? parsed : null;
+  } catch {
+    return null;
+  }
+}
+
+function getCodexSessionId(headers, parsedBody) {
+  const direct = firstHeader(headers, 'session_id') || firstHeader(headers, 'x-openai-session-id');
+  if (direct) return String(direct);
+  const turnMetadata = parseCodexTurnMetadata(headers);
+  if (typeof turnMetadata?.session_id === 'string') return turnMetadata.session_id;
+  return parsedBody?.metadata?.session_id || null;
+}
+
+function getOpenAIAgentTypeFromHeaders(headers) {
+  const subagent = firstHeader(headers, 'x-openai-subagent');
+  const direct = firstHeader(headers, 'x-openai-agent-type') || firstHeader(headers, 'x-codex-agent-type');
+  const turnMetadata = parseCodexTurnMetadata(headers);
+  const value = direct || turnMetadata?.agent_type || subagent;
+  if (!value) return null;
+  const normalized = String(value).toLowerCase();
+  if (normalized === 'explorer' || normalized === 'worker' || normalized === 'default') return normalized;
+  return null;
+}
+
+function isOpenAISubagent(headers, parsedBody) {
+  const raw = firstHeader(headers, 'x-openai-subagent');
+  if (raw != null) {
+    const text = String(raw).toLowerCase();
+    return text !== '0' && text !== 'false' && text !== 'no';
+  }
+  return Boolean(parsedBody?.metadata?.is_subagent || parsedBody?.metadata?.isSubagent);
+}
+
+// Used by both HTTP (parsedBody present) and WebSocket upgrade (no body) paths.
+// When parsedBody is null but headers carry session_id, we synthesize a minimal
+// body so store.detectSession honors header-derived sessions consistently —
+// otherwise WS upgrades and body-less HTTP retries would collapse into the
+// `codex-raw` bucket.
+function detectOpenAISession(headers, parsedBody) {
+  const sessionId = getCodexSessionId(headers, parsedBody);
+  if (!sessionId) {
+    return { sessionId: getCodexRawSessionId(), isNewSession: false, inferred: true };
+  }
+  const bodyForDetection = parsedBody || { metadata: { session_id: sessionId } };
+  const detected = store.detectSession(bodyForDetection);
+  return {
+    sessionId: detected.sessionId || sessionId || getCodexRawSessionId(),
+    isNewSession: detected.isNewSession || false,
+    inferred: detected.inferred || false,
+  };
+}
+
+function withCodexMetadata(parsedBody, headers) {
+  if (!parsedBody || typeof parsedBody !== 'object') return parsedBody;
+  const sessionId = getCodexSessionId(headers, parsedBody);
+  const agentType = getOpenAIAgentTypeFromHeaders(headers);
+  if (!sessionId && !agentType) return parsedBody;
+  const metadata = parsedBody.metadata && typeof parsedBody.metadata === 'object'
+    ? { ...parsedBody.metadata }
+    : {};
+  if (sessionId && !metadata.session_id) metadata.session_id = sessionId;
+  if (agentType && !metadata.agent_type) metadata.agent_type = agentType;
+  return { ...parsedBody, metadata };
+}
+
+module.exports = {
+  getCodexRawSessionId,
+  firstHeader,
+  parseCodexTurnMetadata,
+  getCodexSessionId,
+  getOpenAIAgentTypeFromHeaders,
+  isOpenAISubagent,
+  detectOpenAISession,
+  withCodexMetadata,
+};

server/restore.js

@@ -206,10 +206,19 @@ async function buildVersionIndex() {
       const b0 = isOpenAI ? '' : (sys[0]?.text || '');
       const b2 = isOpenAI ? (typeof sys === 'string' ? sys : JSON.stringify(sys, null, 2)) : (sys[2]?.text || '');
       const m = b0.match(/cc_version=(\S+?)[; ]/);
-      const { key: agentKey, label: agentLabel } = isOpenAI
-        ? extractPromptAgentType('openai', { instructions: b2 })
-        : extractAgentType(sys);
       const sysHash = filename.replace(/^sys_/, '').replace(/^openai_instructions_/, '').replace(/\.json$/, '');
+      let agentInfo = null;
+      if (isOpenAI) {
+        const meta = await config.storage.readShared(`openai_prompt_meta_${sysHash}.json`)
+          .then(JSON.parse)
+          .catch(() => null);
+        if (meta?.agentKey) {
+          agentInfo = { key: meta.agentKey, label: meta.agentLabel || meta.agentKey };
+        }
+      }
+      const { key: agentKey, label: agentLabel } = agentInfo || (isOpenAI
+        ? extractPromptAgentType('openai', { instructions: b2 })
+        : extractAgentType(sys));
       if (sysHash && agentKey) sysHashToAgentKey.set(sysHash, agentKey);
       if (b2.length >= (isOpenAI ? 1 : 500)) {
         const coreText = isOpenAI ? b2 : (splitB2IntoBlocks(b2).coreInstructions || '');