From 48419f96f0fc7cfed2602eeb9d5df3760477ffe9 Mon Sep 17 00:00:00 2001 From: Phillip Cunliffe Date: Fri, 22 May 2026 12:26:35 -0700 Subject: [PATCH] Add Codex metadata projection --- .../ai-gateway/src/message_projector.js | 266 +++++++++++++++--- .../smoke/flows/gateway_codex_capture.js | 141 +++++++--- .../ai-gateway-message-projector.test.js | 133 ++++++++- 3 files changed, 473 insertions(+), 67 deletions(-) diff --git a/hypaware-core/plugins-workspace/ai-gateway/src/message_projector.js b/hypaware-core/plugins-workspace/ai-gateway/src/message_projector.js index a89b4f27..b369fd81 100644 --- a/hypaware-core/plugins-workspace/ai-gateway/src/message_projector.js +++ b/hypaware-core/plugins-workspace/ai-gateway/src/message_projector.js @@ -146,12 +146,19 @@ export function createAiGatewayMessageProjector(opts) { conversation_source: projection.conversation_source, cwd: projection.cwd, git_branch: projection.git_branch, - claude_version: projection.claude_version, + client_version: projection.client_version, + client_name: projection.client_name, + entrypoint: projection.entrypoint, + user_type: projection.user_type, + permission_mode: projection.permission_mode, + is_sidechain: projection.is_sidechain, user_id: projection.user_id, provider: projection.provider, model: projection.model, system_text: projection.system_text, tools: projection.tools, + request_id: projection.request_id, + prompt_id: projection.prompt_id, message_index: i, previous_message_id: [...conversationMessageIds], message_created_at: projection.ts_start, @@ -175,7 +182,10 @@ export function createAiGatewayMessageProjector(opts) { row.date = utcDate(row.message_created_at) row.session_id = projection.session_id row.content = content - row.attributes = mergeJsonObjects(row.attributes, gatewayAttributes) + row.attributes = mergeJsonObjects( + mergeJsonObjects(row.attributes, projection.attributes), + gatewayAttributes + ) /** @type {Record} */ let enriched = row @@ -212,14 +222,15 @@ function buildProjection(exchange) { if (!isPlainObject(reqBody)) return undefined const requestPath = stringValue(exchange.path) ?? stringValue(readPath(exchange, ['request', 'path'])) ?? '' const provider = resolveProvider(exchange, reqBody, requestPath) + const codexContext = resolveCodexContext(exchange, provider, requestPath, resolveRecordedCwd(reqBody, exchange)) const responseBody = parseMaybeJson(readRawResponseBody(exchange)) const ts_start = stringValue(exchange.ts_start) ?? new Date().toISOString() const messages = messagesForProvider(provider, requestPath, reqBody, responseBody, exchange) if (messages.length === 0) return undefined - const conversation_id = resolveConversationId(reqBody, exchange) - const session_id = resolveSessionId(reqBody, exchange) - const recordedContext = resolveRecordedContext(reqBody, exchange) + const conversation_id = resolveConversationId(reqBody, exchange, provider, requestPath, codexContext) + const session_id = resolveSessionId(reqBody, exchange, codexContext) + const recordedContext = resolveRecordedContext(reqBody, exchange, codexContext) return { provider, conversation_id, @@ -228,7 +239,15 @@ function buildProjection(exchange) { conversation_source: resolveConversationSource(exchange, provider), cwd: recordedContext.cwd, git_branch: recordedContext.git_branch, - claude_version: recordedContext.claude_version, + client_version: recordedContext.client_version, + client_name: recordedContext.client_name, + entrypoint: recordedContext.entrypoint, + user_type: recordedContext.user_type, + permission_mode: recordedContext.permission_mode, + is_sidechain: recordedContext.is_sidechain, + request_id: resolveRequestId(exchange, codexContext), + prompt_id: codexContext?.turn_id, + attributes: codexContext?.attributes ? { codex: codexContext.attributes } : undefined, model: resolveModel(reqBody, responseBody), system_text: extractSystemText(reqBody.system), tools: reqBody.tools, @@ -496,10 +515,21 @@ function isOpenAiResponsesPath(path) { /** * @param {Record} reqBody * @param {Record} exchange + * @param {string} provider + * @param {string} path + * @param {ReturnType} codexContext * @returns {string} */ -function resolveConversationId(reqBody, exchange) { - const sessionId = resolveSessionId(reqBody, exchange) +function resolveConversationId(reqBody, exchange, provider, path, codexContext) { + if (isCodexExchange(provider, path, exchange)) { + const codexConversationId = firstString( + codexContext?.thread_id, + readHeader(exchange, 'thread-id'), + readHeader(exchange, 'session-id') + ) + if (codexConversationId) return codexConversationId + } + const sessionId = resolveSessionId(reqBody, exchange, codexContext) if (sessionId) return sessionId const messages = Array.isArray(reqBody.messages) ? reqBody.messages : responsesInputMessages(reqBody.input) if (messages.length > 0 && isPlainObject(messages[0])) { @@ -512,8 +542,13 @@ function resolveConversationId(reqBody, exchange) { /** * @param {Record} reqBody * @param {Record} exchange + * @param {ReturnType} [codexContext] */ -function resolveSessionId(reqBody, exchange) { +function resolveSessionId(reqBody, exchange, codexContext) { + if (codexContext) { + const sessionId = firstString(codexContext.session_id, readHeader(exchange, 'session-id')) + if (sessionId) return sessionId + } return readMetadataSessionId(reqBody) ?? readHeader(exchange, 'x-claude-code-session-id') } @@ -559,17 +594,25 @@ function resolveConversationSource(exchange, provider) { /** * @param {Record} reqBody * @param {Record} exchange + * @param {ReturnType} [codexContext] */ -function resolveRecordedContext(reqBody, exchange) { +function resolveRecordedContext(reqBody, exchange, codexContext) { const meta = readKey(reqBody, 'metadata') const userId = isPlainObject(meta) ? parseMaybeJson(meta.user_id) : undefined + const claudeVersion = firstString( + readStringKey(exchange, 'claude_version'), + readStringKey(exchange, 'claudeVersion'), + readStringKey(reqBody, 'claude_version'), + readStringKey(reqBody, 'claudeVersion'), + readStringKey(meta, 'claude_version'), + readStringKey(meta, 'claudeVersion'), + readStringKey(userId, 'claude_version'), + readStringKey(userId, 'claudeVersion'), + claudeVersionFromUserAgent(readPath(exchange, ['client', 'user_agent'])) + ) + const cwd = firstString(codexContext?.cwd, resolveRecordedCwd(reqBody, exchange)) return { - cwd: firstString( - readStringKey(exchange, 'cwd'), - readStringKey(reqBody, 'cwd'), - readStringKey(meta, 'cwd'), - readStringKey(userId, 'cwd') - ), + cwd, git_branch: firstString( readStringKey(exchange, 'git_branch'), readStringKey(exchange, 'gitBranch'), @@ -580,17 +623,142 @@ function resolveRecordedContext(reqBody, exchange) { readStringKey(userId, 'git_branch'), readStringKey(userId, 'gitBranch') ), - claude_version: firstString( - readStringKey(exchange, 'claude_version'), - readStringKey(exchange, 'claudeVersion'), - readStringKey(reqBody, 'claude_version'), - readStringKey(reqBody, 'claudeVersion'), - readStringKey(meta, 'claude_version'), - readStringKey(meta, 'claudeVersion'), - readStringKey(userId, 'claude_version'), - readStringKey(userId, 'claudeVersion'), - claudeVersionFromUserAgent(readPath(exchange, ['client', 'user_agent'])) - ), + client_version: firstString(codexContext?.client_version, claudeVersion), + client_name: codexContext ? 'codex' : claudeVersion ? 'claude' : undefined, + entrypoint: codexContext?.entrypoint, + user_type: codexContext?.thread_source, + permission_mode: codexContext?.sandbox, + is_sidechain: codexContext?.thread_source ? codexContext.thread_source === 'subagent' : undefined, + } +} + +/** + * @param {Record} reqBody + * @param {Record} exchange + */ +function resolveRecordedCwd(reqBody, exchange) { + const meta = readKey(reqBody, 'metadata') + const userId = isPlainObject(meta) ? parseMaybeJson(meta.user_id) : undefined + return firstString( + readStringKey(exchange, 'cwd'), + readStringKey(reqBody, 'cwd'), + readStringKey(meta, 'cwd'), + readStringKey(userId, 'cwd') + ) +} + +/** + * @param {Record} exchange + * @param {ReturnType} codexContext + */ +function resolveRequestId(exchange, codexContext) { + if (!codexContext) return undefined + return readResponseHeader(exchange, 'x-oai-request-id') ?? readHeader(exchange, 'x-client-request-id') +} + +/** + * @param {Record} exchange + * @param {string} provider + * @param {string} path + * @param {string | undefined} preferredCwd + */ +function resolveCodexContext(exchange, provider, path, preferredCwd) { + if (!isCodexExchange(provider, path, exchange)) return undefined + const metadata = readCodexTurnMetadata(exchange) + const userAgent = readHeader(exchange, 'user-agent') ?? stringValue(readPath(exchange, ['client', 'user_agent'])) + const client = codexClientFromUserAgent(userAgent) + const workspace = selectCodexWorkspace(metadata, preferredCwd) + const workspaceInfo = workspace?.info + const remoteUrls = isPlainObject(workspaceInfo?.associated_remote_urls) + ? workspaceInfo.associated_remote_urls + : undefined + const thread_id = firstString(readStringKey(metadata, 'thread_id'), readHeader(exchange, 'thread-id')) + const session_id = firstString(readStringKey(metadata, 'session_id'), readHeader(exchange, 'session-id')) + const turn_id = readStringKey(metadata, 'turn_id') + const thread_source = readStringKey(metadata, 'thread_source') + const originator = firstString(readHeader(exchange, 'originator'), client.entrypoint) + const sandbox = readStringKey(metadata, 'sandbox') + const turn_started_at_unix_ms = numberValue(readKey(metadata, 'turn_started_at_unix_ms')) + const window_id = readHeader(exchange, 'x-codex-window-id') + const git_origin_url = readStringKey(remoteUrls, 'origin') + const git_commit = readStringKey(workspaceInfo, 'latest_git_commit_hash') + const has_changes = typeof workspaceInfo?.has_changes === 'boolean' ? workspaceInfo.has_changes : undefined + + /** @type {Record} */ + const attributes = {} + setIfString(attributes, 'thread_id', thread_id) + setIfString(attributes, 'session_id', session_id) + setIfString(attributes, 'turn_id', turn_id) + setIfString(attributes, 'thread_source', thread_source) + setIfString(attributes, 'originator', originator) + setIfString(attributes, 'window_id', window_id) + setIfString(attributes, 'sandbox', sandbox) + if (turn_started_at_unix_ms !== undefined) attributes.turn_started_at_unix_ms = turn_started_at_unix_ms + setIfString(attributes, 'workspace', workspace?.path) + setIfString(attributes, 'git_origin_url', git_origin_url) + setIfString(attributes, 'git_commit', git_commit) + if (has_changes !== undefined) attributes.has_changes = has_changes + + return { + thread_id, + session_id, + turn_id, + thread_source, + cwd: workspace?.path, + client_version: client.version, + entrypoint: originator, + sandbox, + attributes: Object.keys(attributes).length > 0 ? attributes : undefined, + } +} + +/** + * @param {string} provider + * @param {string} path + * @param {Record} exchange + */ +function isCodexExchange(provider, path, exchange) { + return provider === 'chatgpt' || + path === '/backend-api/codex/responses' || + path.startsWith('/backend-api/codex/responses/') || + readHeader(exchange, 'x-codex-turn-metadata') !== undefined +} + +/** @param {Record} exchange */ +function readCodexTurnMetadata(exchange) { + const parsed = parseMaybeJson(readHeader(exchange, 'x-codex-turn-metadata')) + return isPlainObject(parsed) ? parsed : undefined +} + +/** @param {string | undefined} userAgent */ +function codexClientFromUserAgent(userAgent) { + if (typeof userAgent !== 'string') return {} + const match = /^([^/]+)\/([^/\s]+)/.exec(userAgent) + if (!match) return {} + const product = match[1].trim() + if (!/^codex(?:\b|[-_\s])/i.test(product)) return {} + return { + entrypoint: product, + version: match[2], + } +} + +/** + * @param {Record | undefined} metadata + * @param {string | undefined} preferredCwd + * @returns {{ path: string, info?: Record } | undefined} + */ +function selectCodexWorkspace(metadata, preferredCwd) { + const workspaces = readKey(metadata, 'workspaces') + if (!isPlainObject(workspaces)) return undefined + const workspacePath = preferredCwd && Object.hasOwn(workspaces, preferredCwd) + ? preferredCwd + : Object.keys(workspaces).find((key) => key.length > 0) + if (!workspacePath) return undefined + const info = readKey(workspaces, workspacePath) + return { + path: workspacePath, + info: isPlainObject(info) ? info : undefined, } } @@ -625,7 +793,11 @@ export function extractMessageParts(exchange, message, ctx) { if (content.length === 0) return [] const message_id = computeMessageId(String(ctx.conversation_id), role, content) - const attributes = withClientAttributes(extractAttributes(exchange, message), stringValue(ctx.claude_version)) + const attributes = withClientAttributes( + extractAttributes(exchange, message), + stringValue(ctx.client_version), + stringValue(ctx.client_name) + ) const finishReason = mapFinishReason(stringValue(message.stop_reason)) const base = { @@ -640,9 +812,15 @@ export function extractMessageParts(exchange, message, ctx) { conversation_source: ctx.conversation_source, cwd: ctx.cwd, git_branch: ctx.git_branch, - client_version: ctx.claude_version, + client_version: ctx.client_version, + entrypoint: ctx.entrypoint, + user_type: ctx.user_type, + permission_mode: ctx.permission_mode, + is_sidechain: ctx.is_sidechain, message_id, previous_message_id: ctx.previous_message_id, + request_id: ctx.request_id, + prompt_id: ctx.prompt_id, message_index: ctx.message_index, message_created_at: ctx.message_created_at, role, @@ -862,13 +1040,19 @@ function extractAttributes(exchange, message) { /** * @param {Record | undefined} attributes - * @param {string | undefined} claudeVersion + * @param {string | undefined} clientVersion + * @param {string | undefined} clientName */ -function withClientAttributes(attributes, claudeVersion) { - if (!claudeVersion) return attributes +function withClientAttributes(attributes, clientVersion, clientName) { + if (!clientVersion) return attributes const out = attributes ? { ...attributes } : {} const client = isPlainObject(out.client) ? { ...out.client } : {} - client.claude_version = claudeVersion + if (clientName === 'claude') { + client.claude_version = clientVersion + } else { + client.version = clientVersion + if (clientName) client.name = clientName + } out.client = client return out } @@ -962,6 +1146,17 @@ function readStreamEvents(exchange) { /** @param {unknown} exchange @param {string} name */ function readHeader(exchange, name) { const headers = parseMaybeJson(readKey(exchange, 'request_headers')) ?? readPath(exchange, ['request', 'headers']) + return readHeaderValue(headers, name) +} + +/** @param {unknown} exchange @param {string} name */ +function readResponseHeader(exchange, name) { + const headers = parseMaybeJson(readKey(exchange, 'response_headers')) ?? readPath(exchange, ['response', 'headers']) + return readHeaderValue(headers, name) +} + +/** @param {unknown} headers @param {string} name */ +function readHeaderValue(headers, name) { if (!isPlainObject(headers)) return undefined const wanted = name.toLowerCase() for (const [key, value] of Object.entries(headers)) { @@ -993,6 +1188,11 @@ function readStringKey(obj, key) { return typeof value === 'string' && value.length > 0 ? value : undefined } +/** @param {Record} target @param {string} key @param {string | undefined} value */ +function setIfString(target, key, value) { + if (value !== undefined) target[key] = value +} + /** * @param {Record[]} streamEvents * @returns {Record | null} diff --git a/hypaware-core/smoke/flows/gateway_codex_capture.js b/hypaware-core/smoke/flows/gateway_codex_capture.js index 3383d74c..02d9c73b 100644 --- a/hypaware-core/smoke/flows/gateway_codex_capture.js +++ b/hypaware-core/smoke/flows/gateway_codex_capture.js @@ -12,25 +12,26 @@ import { dispatch } from '../../../src/core/cli/dispatch.js' /** * Phase 7 smoke — OpenAI `/v1` passthrough plus the Codex-specific - * `/v1/responses` capture under the daemon boot path. + * ChatGPT `/backend-api/codex/responses` capture under the daemon boot path. * - * Boots `runDaemon` with `@hypaware/ai-gateway` activated and an - * `openai`-named upstream rooted at `/v1`. Two requests run through - * it: + * Boots `runDaemon` with `@hypaware/ai-gateway` activated and + * OpenAI plus ChatGPT Codex upstreams. Two requests run through it: * * - `POST /v1/chat/completions` — the legacy OpenAI Chat path, a * proxy of every non-streaming inference call. - * - `POST /v1/responses` — the OpenAI Responses API endpoint Codex - * uses; the response is an SSE stream so the recorder also - * exercises the `is_sse=true` / `stream_event_count>0` columns. + * - `POST /backend-api/codex/responses` — the ChatGPT endpoint Codex + * Desktop uses; the response is an SSE stream so the recorder also + * exercises the `is_sse=true` / `stream_event_count>0` columns and + * Codex metadata projection. * * Bead `hy-bbyi` assertions: * * - Four normalized rows land in `ai_gateway_messages` filterable by * `dev_run_id`: user+assistant for chat completions and user+assistant - * for Responses. - * - The `/v1/responses` rows carry `is_sse=true` and a positive - * `stream_event_count` under `attributes.gateway`. + * for Codex Responses. + * - The `/backend-api/codex/responses` rows carry `is_sse=true`, a + * positive `stream_event_count` under `attributes.gateway`, and + * Codex turn metadata projected into first-class columns. * - Daemon self-telemetry (`source.start`, `sink.tick`, * `cache.append`, `daemon.shutdown`) is present in JSONL. * @@ -57,10 +58,11 @@ export async function run({ harness, expect }) { listen: '127.0.0.1:0', upstreams: [ // `path_prefix: '/v1'` is the same value `@hypaware/codex` - // registers via `registerUpstreamPreset()` in production — - // matches `/v1/chat/completions`, `/v1/responses`, and any - // other Responses-API path Codex emits. + // registers for API-key mode in production. { name: 'openai', base_url: openai.url, path_prefix: '/v1', provider: 'openai' }, + // `path_prefix: '/backend-api/codex'` is the same ChatGPT-auth + // route `@hypaware/codex` registers in production. + { name: 'chatgpt', base_url: openai.url, path_prefix: '/backend-api/codex', provider: 'chatgpt' }, ], }, }, @@ -95,16 +97,47 @@ export async function run({ harness, expect }) { const chatResp = await postJson(`${gatewayUrl}/v1/chat/completions`, harness.devRunId, chatBody) expect.that('gateway: /v1/chat/completions returned 200', chatResp.statusCode, (v) => v === 200) - // ----- 2. Streaming /v1/responses (the Codex contract path) ----- + // ----- 2. Streaming /backend-api/codex/responses (the Codex contract path) ----- + const codexWorkspace = '/Users/phil/workspace/hypaware' + const codexThreadId = `thread-${harness.devRunId}` + const codexSessionId = `session-${harness.devRunId}` + const codexTurnId = `turn-${harness.devRunId}` const responsesBody = JSON.stringify({ model: 'gpt-5-codex', input: [{ role: 'user', content: [{ type: 'input_text', text: 'help refactor' }] }], stream: true, }) - const responsesResp = await postJson(`${gatewayUrl}/v1/responses`, harness.devRunId, responsesBody) - expect.that('gateway: /v1/responses returned 200', responsesResp.statusCode, (v) => v === 200) + const responsesResp = await postJson( + `${gatewayUrl}/backend-api/codex/responses`, + harness.devRunId, + responsesBody, + { + 'thread-id': codexThreadId, + 'session-id': codexSessionId, + 'x-client-request-id': `client-request-${harness.devRunId}`, + originator: 'Codex Desktop', + 'user-agent': 'Codex Desktop/0.133.0-alpha.1', + 'x-codex-window-id': `window-${harness.devRunId}`, + 'x-codex-turn-metadata': JSON.stringify({ + session_id: codexSessionId, + thread_id: codexThreadId, + thread_source: 'user', + turn_id: codexTurnId, + workspaces: { + [codexWorkspace]: { + associated_remote_urls: { origin: 'https://github.com/hyparam/hypaware.git' }, + latest_git_commit_hash: '072b240f2c82e15de26022a8b9bb29e13be826a9', + has_changes: true, + }, + }, + sandbox: 'seatbelt', + turn_started_at_unix_ms: 1779476507669, + }), + } + ) + expect.that('gateway: /backend-api/codex/responses returned 200', responsesResp.statusCode, (v) => v === 200) expect.that( - 'gateway: /v1/responses body looks like an SSE stream', + 'gateway: /backend-api/codex/responses body looks like an SSE stream', responsesResp.body, (v) => typeof v === 'string' && v.includes('data: ') && v.includes('response.completed'), ) @@ -122,10 +155,22 @@ export async function run({ harness, expect }) { model, role, content_text, + conversation_id, + cwd, + client_version, + entrypoint, + user_type, + permission_mode, + is_sidechain, + request_id, + prompt_id, JSON_VALUE(attributes, '$.gateway.path') as path, JSON_VALUE(attributes, '$.gateway.status_code') as status_code, JSON_VALUE(attributes, '$.gateway.is_sse') as is_sse, - JSON_VALUE(attributes, '$.gateway.stream_event_count') as stream_event_count + JSON_VALUE(attributes, '$.gateway.stream_event_count') as stream_event_count, + JSON_VALUE(attributes, '$.codex.thread_id') as codex_thread_id, + JSON_VALUE(attributes, '$.codex.workspace') as codex_workspace, + JSON_VALUE(attributes, '$.codex.git_origin_url') as codex_git_origin_url from ai_gateway_messages where JSON_VALUE(attributes, '$.dev_run_id') = '${harness.devRunId}' order by message_created_at, message_index, part_index @@ -148,11 +193,16 @@ export async function run({ harness, expect }) { ) const chatRows = rows.filter((r) => r.path === '/v1/chat/completions') - const responseRows = rows.filter((r) => r.path === '/v1/responses') + const responseRows = rows.filter((r) => r.path === '/backend-api/codex/responses') + expect.that( + 'query: /v1/chat/completions rows carry provider=openai', + chatRows.map((r) => r.provider), + (v) => Array.isArray(v) && v.length === 2 && v.every((provider) => provider === 'openai'), + ) expect.that( - 'query: all rows carry provider=openai', - rows.map((r) => r.provider), - (v) => Array.isArray(v) && v.length === 4 && v.every((provider) => provider === 'openai'), + 'query: /backend-api/codex/responses rows carry provider=chatgpt', + responseRows.map((r) => r.provider), + (v) => Array.isArray(v) && v.length === 2 && v.every((provider) => provider === 'chatgpt'), ) expect.that( 'query: /v1/chat/completions has user and assistant rows', @@ -165,20 +215,38 @@ export async function run({ harness, expect }) { (v) => v.length === 2 && v.every((r) => Number(r.status_code) === 200 && (r.is_sse === false || r.is_sse === 0 || r.is_sse === 'false')), ) expect.that( - 'query: /v1/responses has user and assistant rows', + 'query: /backend-api/codex/responses has user and assistant rows', responseRows.map((r) => r.role).sort(), (v) => Array.isArray(v) && v.join(',') === 'assistant,user', ) expect.that( - 'query: /v1/responses rows have is_sse=true', + 'query: /backend-api/codex/responses rows have is_sse=true', responseRows, (v) => v.length === 2 && v.every((r) => r.is_sse === true || r.is_sse === 1 || r.is_sse === 'true'), ) expect.that( - 'query: /v1/responses rows have stream_event_count > 0', + 'query: /backend-api/codex/responses rows have stream_event_count > 0', responseRows, (v) => v.length === 2 && v.every((r) => Number(r.stream_event_count) > 0), ) + expect.that( + 'query: /backend-api/codex/responses rows have projected Codex columns', + responseRows, + (v) => v.length === 2 && v.every((r) => + r.conversation_id === codexThreadId && + r.cwd === codexWorkspace && + r.client_version === '0.133.0-alpha.1' && + r.entrypoint === 'Codex Desktop' && + r.user_type === 'user' && + r.permission_mode === 'seatbelt' && + (r.is_sidechain === false || r.is_sidechain === 0 || r.is_sidechain === 'false') && + r.request_id === 'oai-request-codex-smoke' && + r.prompt_id === codexTurnId && + r.codex_thread_id === codexThreadId && + r.codex_workspace === codexWorkspace && + r.codex_git_origin_url === 'https://github.com/hyparam/hypaware.git' + ), + ) // ----- Daemon-self-telemetry assertions ----- const traces = await expect.traces() @@ -221,8 +289,8 @@ export async function run({ harness, expect }) { (rows) => rows.length === 2, ) expect.that( - 'logs: aigw.exchange for /v1/responses carries is_sse=true', - exchangeLogs.find((/** @type {any} */ l) => l.attributes?.path === '/v1/responses')?.attributes?.is_sse, + 'logs: aigw.exchange for /backend-api/codex/responses carries is_sse=true', + exchangeLogs.find((/** @type {any} */ l) => l.attributes?.path === '/backend-api/codex/responses')?.attributes?.is_sse, (v) => v === true, ) } @@ -236,11 +304,13 @@ export async function run({ harness, expect }) { * exercises: * * - `POST /chat/completions` — non-streaming JSON response. - * - `POST /responses` — SSE stream with three events ending in + * - `POST /responses` and `/backend-api/codex/responses` + * — SSE stream with three events ending in * `response.completed`. * - * The proxy prefix `/v1` is consumed by the gateway's path matcher; - * the upstream sees `/chat/completions` and `/responses`. + * The fake upstream accepts both the OpenAI `/v1` paths and the + * ChatGPT Codex backend path because the gateway preserves the + * request path when forwarding. * * @returns {Promise<{ url: string, close: () => Promise }>} */ @@ -253,10 +323,15 @@ async function startOpenAiUpstream() { } req.resume() req.on('end', () => { - if (req.url === '/v1/responses' || req.url === '/responses') { + if ( + req.url === '/v1/responses' || + req.url === '/responses' || + req.url === '/backend-api/codex/responses' + ) { res.writeHead(200, { 'content-type': 'text/event-stream', 'cache-control': 'no-cache', + 'x-oai-request-id': 'oai-request-codex-smoke', }) res.write('event: response.created\ndata: {"id":"resp_1","status":"in_progress"}\n\n') res.write('event: response.output_text.delta\ndata: {"delta":"ok"}\n\n') @@ -300,9 +375,10 @@ async function startOpenAiUpstream() { * @param {string} url * @param {string} runId * @param {string} body + * @param {Record} [extraHeaders] * @returns {Promise<{ statusCode: number, body: string }>} */ -function postJson(url, runId, body) { +function postJson(url, runId, body, extraHeaders = {}) { return new Promise((resolve, reject) => { const parsed = new URL(url) const req = http.request( @@ -315,6 +391,7 @@ function postJson(url, runId, body) { 'content-type': 'application/json', 'content-length': String(Buffer.byteLength(body)), 'x-hyp-dev-run-id': runId, + ...extraHeaders, }, }, (res) => { diff --git a/test/plugins/ai-gateway-message-projector.test.js b/test/plugins/ai-gateway-message-projector.test.js index 9f75ee55..8bd48a6d 100644 --- a/test/plugins/ai-gateway-message-projector.test.js +++ b/test/plugins/ai-gateway-message-projector.test.js @@ -219,6 +219,135 @@ test('reconstructs OpenAI Responses SSE text into an assistant part row', async assert.equal(rows[1].attributes.gateway.is_sse, true) }) +test('projects Codex turn metadata from ChatGPT gateway headers', async () => { + const projector = createAiGatewayMessageProjector({ gatewayId: 'gw-test' }) + const turnMetadata = { + session_id: 'codex-session-1', + thread_id: 'codex-thread-1', + thread_source: 'user', + turn_id: 'codex-turn-1', + workspaces: { + '/Users/phil/workspace/hypaware': { + associated_remote_urls: { origin: 'https://github.com/hyparam/hypaware.git' }, + latest_git_commit_hash: '072b240f2c82e15de26022a8b9bb29e13be826a9', + has_changes: true, + }, + }, + sandbox: 'seatbelt', + turn_started_at_unix_ms: 1779476507669, + } + const rows = await projector.projectExchange(exchange({ + provider: 'chatgpt', + path: '/backend-api/codex/responses', + request_headers: { + 'thread-id': 'codex-thread-header', + 'session-id': 'codex-session-header', + 'x-client-request-id': 'client-request-1', + originator: 'Codex Desktop', + 'user-agent': 'Codex Desktop/0.133.0-alpha.1', + 'x-codex-window-id': 'window-1', + 'x-codex-turn-metadata': JSON.stringify(turnMetadata), + }, + request_body: { + model: 'gpt-5-codex', + input: [{ role: 'user', content: [{ type: 'input_text', text: 'help refactor' }] }], + stream: false, + }, + response_headers: { + 'x-oai-request-id': 'oai-request-1', + }, + response_body: { + output_text: 'ok', + usage: { input_tokens: 8, output_tokens: 1 }, + }, + })) + + assert.equal(rows.length, 2) + assert.ok(rows.every((row) => row.provider === 'chatgpt')) + assert.ok(rows.every((row) => row.conversation_id === 'codex-thread-1')) + assert.ok(rows.every((row) => row.cwd === '/Users/phil/workspace/hypaware')) + assert.ok(rows.every((row) => row.git_branch === undefined)) + assert.ok(rows.every((row) => row.client_version === '0.133.0-alpha.1')) + assert.ok(rows.every((row) => row.entrypoint === 'Codex Desktop')) + assert.ok(rows.every((row) => row.user_type === 'user')) + assert.ok(rows.every((row) => row.permission_mode === 'seatbelt')) + assert.ok(rows.every((row) => row.is_sidechain === false)) + assert.ok(rows.every((row) => row.request_id === 'oai-request-1')) + assert.ok(rows.every((row) => row.prompt_id === 'codex-turn-1')) + assert.equal(rows[0].attributes.codex.thread_id, 'codex-thread-1') + assert.equal(rows[0].attributes.codex.session_id, 'codex-session-1') + assert.equal(rows[0].attributes.codex.turn_id, 'codex-turn-1') + assert.equal(rows[0].attributes.codex.thread_source, 'user') + assert.equal(rows[0].attributes.codex.originator, 'Codex Desktop') + assert.equal(rows[0].attributes.codex.window_id, 'window-1') + assert.equal(rows[0].attributes.codex.sandbox, 'seatbelt') + assert.equal(rows[0].attributes.codex.turn_started_at_unix_ms, 1779476507669) + assert.equal(rows[0].attributes.codex.workspace, '/Users/phil/workspace/hypaware') + assert.equal(rows[0].attributes.codex.git_origin_url, 'https://github.com/hyparam/hypaware.git') + assert.equal(rows[0].attributes.codex.git_commit, '072b240f2c82e15de26022a8b9bb29e13be826a9') + assert.equal(rows[0].attributes.codex.has_changes, true) +}) + +test('marks Codex subagent turns as sidechain rows without workspace metadata', async () => { + const projector = createAiGatewayMessageProjector({ gatewayId: 'gw-test' }) + const rows = await projector.projectExchange(exchange({ + provider: 'chatgpt', + path: '/backend-api/codex/responses', + request_headers: { + 'thread-id': 'subagent-thread-header', + 'session-id': 'subagent-session-header', + originator: 'Codex Desktop', + 'user-agent': 'Codex Desktop/0.133.0-alpha.1', + 'x-codex-turn-metadata': JSON.stringify({ + session_id: 'subagent-session-1', + thread_id: 'subagent-thread-1', + thread_source: 'subagent', + turn_id: 'subagent-turn-1', + sandbox: 'seatbelt', + }), + }, + request_body: { + model: 'gpt-5-codex', + input: [{ role: 'user', content: 'check status' }], + }, + response_body: { output_text: 'ok' }, + })) + + assert.equal(rows.length, 2) + assert.ok(rows.every((row) => row.conversation_id === 'subagent-thread-1')) + assert.ok(rows.every((row) => row.cwd === undefined)) + assert.ok(rows.every((row) => row.is_sidechain === true)) + assert.ok(rows.every((row) => row.prompt_id === 'subagent-turn-1')) +}) + +test('falls back to Codex header identifiers when turn metadata is invalid', async () => { + const projector = createAiGatewayMessageProjector({ gatewayId: 'gw-test' }) + const rows = await projector.projectExchange(exchange({ + provider: 'chatgpt', + path: '/backend-api/codex/responses', + request_headers: { + 'thread-id': 'fallback-thread', + 'session-id': 'fallback-session', + 'x-client-request-id': 'client-request-fallback', + originator: 'Codex Desktop', + 'user-agent': 'Codex Desktop/0.133.0-alpha.1', + 'x-codex-turn-metadata': '{', + }, + request_body: { + model: 'gpt-5-codex', + input: [{ role: 'user', content: 'hello' }], + }, + response_body: { output_text: 'ok' }, + })) + + assert.equal(rows.length, 2) + assert.ok(rows.every((row) => row.conversation_id === 'fallback-thread')) + assert.ok(rows.every((row) => row.request_id === 'client-request-fallback')) + assert.ok(rows.every((row) => row.prompt_id === undefined)) + assert.equal(rows[0].attributes.codex.thread_id, 'fallback-thread') + assert.equal(rows[0].attributes.codex.session_id, 'fallback-session') +}) + test('applies enrichers before stripping non-schema draft fields', async () => { const projector = createAiGatewayMessageProjector({ gatewayId: 'gw-test', @@ -255,9 +384,9 @@ function exchange(overrides = {}) { response_bytes: 20, is_sse: overrides.is_sse ?? false, stream_event_count: overrides.stream_events?.length ?? 0, - request_headers: JSON.stringify({ 'x-hyp-dev-run-id': 'run-1' }), + request_headers: JSON.stringify({ 'x-hyp-dev-run-id': 'run-1', ...(overrides.request_headers ?? {}) }), request_body: JSON.stringify(overrides.request_body), - response_headers: JSON.stringify({ 'content-type': 'application/json' }), + response_headers: JSON.stringify({ 'content-type': 'application/json', ...(overrides.response_headers ?? {}) }), response_body: overrides.response_body === null ? null : JSON.stringify(overrides.response_body), error: null, metadata: JSON.stringify({ dev_run_id: 'run-1' }),