feat: enhance Codex response handling with standardized event stream and logging

Author: caozhiyuan

src/routes/provider/responses/handler.ts

@@ -38,6 +38,10 @@ export async function handleProviderResponsesForProvider(
   },
 ): Promise<Response> {
   const { payload, provider } = options
+  debugJson(logger, "Responses request payload:", {
+    payload,
+    provider,
+  })
   const providerConfig = await resolveProviderConfig(provider)
   if (providerConfig?.type !== "openai-responses") {
     return c.json(
@@ -51,15 +55,19 @@ export async function handleProviderResponsesForProvider(
     )
   }
 
-  const selectedModel =
+  const model =
     providerConfig.name === "codex" ?
       getCodexModels().data.find((model) => model.id === payload.model)
     : undefined
 
-  applyResponsesApiContextManagement(
-    payload,
-    selectedModel?.capabilities.limits.max_prompt_tokens,
-  )
+  const maxPromptTokens = model?.capabilities.limits.max_prompt_tokens ?? 0
+  applyResponsesApiContextManagement(payload, maxPromptTokens)
+
+  const contextManagement = payload.context_management
+  debugJson(logger, "Translated Responses request payload:", {
+    contextManagement,
+    provider,
+  })
 
   const upstreamResponse =
     providerConfig.name === "codex" ?
@@ -79,6 +87,31 @@ export async function handleProviderResponsesForProvider(
 
   const recordUsage = createProviderResponsesUsageRecorder(payload, provider)
 
+  if (providerConfig.name === "codex" && payload.stream) {
+    let usage: UsageTokens = {}
+
+    return createProviderProxyResponse(
+      upstreamResponse,
+      createStandardizedCodexResponsesEventStream(
+        getResponsesEvents(upstreamResponse),
+        {
+          onClose: () => {
+            recordUsage(usage)
+          },
+          onChunk: (chunk) => {
+            debugJson(logger, "Responses stream chunk:", chunk)
+          },
+          onEvent: (event) => {
+            const nextUsage = getResponsesStreamEventUsage(event)
+            if (nextUsage) {
+              usage = nextUsage
+            }
+          },
+        },
+      ),
+    )
+  }
+
   if (payload.stream) {
     void recordProviderResponsesStreamUsage(upstreamResponse.clone(), {
       normalizeCodex: providerConfig.name === "codex",
@@ -97,15 +130,6 @@ export async function handleProviderResponsesForProvider(
     recordUsage(normalizeResponsesUsage(responseBody.usage))
   }
 
-  if (providerConfig.name === "codex" && payload.stream) {
-    return createProviderProxyResponse(
-      upstreamResponse,
-      createStandardizedCodexResponsesEventStream(
-        getResponsesEvents(upstreamResponse),
-      ),
-    )
-  }
-
   return createProviderProxyResponse(upstreamResponse)
 }
 
@@ -153,12 +177,11 @@ const recordProviderResponsesStreamUsage = async (
         normalizeCodex: options.normalizeCodex,
         provider: options.provider,
       })
-      if (
-        parsed?.type === "response.completed"
-        || parsed?.type === "response.failed"
-        || parsed?.type === "response.incomplete"
-      ) {
-        usage = normalizeResponsesUsage(parsed.response.usage)
+      if (parsed) {
+        const nextUsage = getResponsesStreamEventUsage(parsed)
+        if (nextUsage) {
+          usage = nextUsage
+        }
       }
     }
   } finally {
@@ -188,6 +211,20 @@ const parseProviderResponsesStreamEvent = (
   }
 }
 
+const getResponsesStreamEventUsage = (
+  event: ResponseStreamEvent,
+): UsageTokens | null => {
+  if (
+    event.type === "response.completed"
+    || event.type === "response.failed"
+    || event.type === "response.incomplete"
+  ) {
+    return normalizeResponsesUsage(event.response.usage)
+  }
+
+  return null
+}
+
 const getResponsesEvents = (response: unknown) => {
   return events(response as Parameters<typeof events>[0])
 }

src/routes/responses/handler.ts

@@ -40,7 +40,6 @@ export const responsesHandlerDependencies = {
 
 export const handleResponses = async (c: Context) => {
   const payload = await c.req.json<ResponsesPayload>()
-  debugJson(logger, "Responses request payload:", payload)
 
   const providerModelAlias = parseProviderModelAlias(payload.model)
   if (providerModelAlias) {
@@ -51,6 +50,7 @@ export const handleResponses = async (c: Context) => {
     })
   }
 
+  debugJson(logger, "Responses request payload:", payload)
   await responsesHandlerDependencies.checkRateLimit(state)
 
   // not support subagent marker for now , set sessionId = getUUID(requestId)

src/services/codex/create-responses.ts

@@ -32,6 +32,17 @@ type ServerSentEventChunk = {
 
 type CodexResponsesWebSocketChunk = ServerSentEventChunk
 
+type StandardizedCodexResponsesChunk = {
+  chunk: ServerSentEventChunk
+  event: ResponseStreamEvent | null
+}
+
+interface CodexResponsesStandardStreamOptions {
+  onClose?: () => void | Promise<void>
+  onChunk?: (chunk: ServerSentEventChunk) => void | Promise<void>
+  onEvent?: (event: ResponseStreamEvent) => void | Promise<void>
+}
+
 type CodexResponsesWebSocketRequest =
   PooledWebSocketRequest<CodexResponsesWebSocketPayload>
 
@@ -415,9 +426,10 @@ const createCodexResponsesWebSocketProxyResponse = (
 
 export const createStandardizedCodexResponsesEventStream = (
   source: AsyncIterable<ServerSentEventChunk>,
+  options: CodexResponsesStandardStreamOptions = {},
 ): ReadableStream<Uint8Array> => {
   return createServerSentEventStream(
-    normalizeCodexResponsesStandardStream(source),
+    normalizeCodexResponsesStandardStream(source, options),
   )
 }
 
@@ -431,9 +443,20 @@ const normalizeCodexResponsesWebSocketStream = async function* (
 
 const normalizeCodexResponsesStandardStream = async function* (
   stream: AsyncIterable<ServerSentEventChunk>,
+  options: CodexResponsesStandardStreamOptions,
 ): AsyncIterable<ServerSentEventChunk> {
-  for await (const chunk of stream) {
-    yield normalizeCodexResponsesStandardChunk(chunk)
+  try {
+    for await (const chunk of stream) {
+      const normalized = normalizeCodexResponsesStandardChunk(chunk)
+      await options.onChunk?.(normalized.chunk)
+      if (normalized.event) {
+        await options.onEvent?.(normalized.event)
+      }
+
+      yield normalized.chunk
+    }
+  } finally {
+    await options.onClose?.()
   }
 }
 
@@ -465,26 +488,38 @@ const normalizeCodexResponsesProxyChunk = (
 
 const normalizeCodexResponsesStandardChunk = (
   chunk: ServerSentEventChunk,
-): ServerSentEventChunk => {
+): StandardizedCodexResponsesChunk => {
   if (!chunk.data || chunk.data === "[DONE]") {
-    return chunk
+    return {
+      chunk,
+      event: null,
+    }
   }
 
   try {
     const parsed = JSON.parse(chunk.data) as Record<string, unknown>
     logCodexRateLimitsEvent(parsed)
     const normalized = normalizeCodexResponsesEvent(parsed)
     if (!normalized) {
-      return chunk
+      return {
+        chunk,
+        event: null,
+      }
     }
 
     return {
-      ...chunk,
-      data: JSON.stringify(normalized),
-      event: normalized.type,
+      chunk: {
+        ...chunk,
+        data: JSON.stringify(normalized),
+        event: normalized.type,
+      },
+      event: normalized,
     }
   } catch {
-    return chunk
+    return {
+      chunk,
+      event: null,
+    }
   }
 }
 

tests/codex-create-responses.test.ts

@@ -128,6 +128,65 @@ describe("codex api helpers", () => {
     expect(body).toContain("data: [DONE]")
   })
 
+  test("reports normalized terminal events while standardizing Codex SSE", async () => {
+    const seenChunkEvents: Array<string | undefined> = []
+    const seenEvents: Array<string> = []
+    let streamClosed = false
+    const stream = createStandardizedCodexResponsesEventStream(
+      streamChunks([
+        {
+          data: JSON.stringify({
+            response: {
+              created_at: 0,
+              error: null,
+              id: "resp_123",
+              incomplete_details: null,
+              instructions: null,
+              metadata: null,
+              model: "gpt-5.4",
+              object: "response",
+              output: [],
+              output_text: "hello",
+              parallel_tool_calls: true,
+              status: "completed",
+              temperature: null,
+              tool_choice: "auto",
+              tools: [],
+              top_p: null,
+              usage: {
+                input_tokens: 5,
+                output_tokens: 2,
+                total_tokens: 7,
+              },
+            },
+            sequence_number: 1,
+            type: "response.done",
+          }),
+        },
+        {
+          data: "[DONE]",
+        },
+      ]),
+      {
+        onClose: () => {
+          streamClosed = true
+        },
+        onChunk: (chunk) => {
+          seenChunkEvents.push(chunk.event)
+        },
+        onEvent: (event) => {
+          seenEvents.push(event.type)
+        },
+      },
+    )
+
+    await new Response(stream).text()
+
+    expect(seenChunkEvents).toEqual(["response.completed", undefined])
+    expect(seenEvents).toEqual(["response.completed"])
+    expect(streamClosed).toBe(true)
+  })
+
   test("builds the ChatGPT Codex websocket responses path", () => {
     expect(buildCodexResponsesWebSocketUrl()).toBe(
       "wss://chatgpt.com/backend-api/codex/responses",