Normalize responses SSE events for Codex

gateway.mjs

@@ -480,6 +480,319 @@ function sumUsageSnapshots(current, next) {
   return Object.keys(result).length > 0 ? result : null;
 }
 
+function denormalizeUsageSnapshotToResponsesUsage(usage) {
+  if (!usage || typeof usage !== "object") {
+    return null;
+  }
+  const payload = {};
+  if (Number.isInteger(usage.input_tokens)) {
+    payload.input_tokens = usage.input_tokens;
+  }
+  if (Number.isInteger(usage.output_tokens)) {
+    payload.output_tokens = usage.output_tokens;
+  }
+  if (Number.isInteger(usage.total_tokens)) {
+    payload.total_tokens = usage.total_tokens;
+  } else if (Number.isInteger(usage.input_tokens) && Number.isInteger(usage.output_tokens)) {
+    payload.total_tokens = usage.input_tokens + usage.output_tokens;
+  }
+  if (Number.isInteger(usage.cached_tokens)) {
+    payload.input_tokens_details = {
+      cached_tokens: usage.cached_tokens,
+    };
+  }
+  if (Number.isInteger(usage.reasoning_tokens)) {
+    payload.output_tokens_details = {
+      reasoning_tokens: usage.reasoning_tokens,
+    };
+  }
+  return Object.keys(payload).length > 0 ? payload : null;
+}
+
+function cloneJsonLike(value) {
+  if (value === null || value === undefined) {
+    return value;
+  }
+  return JSON.parse(JSON.stringify(value));
+}
+
+function buildSseBlock(eventName, payloadText) {
+  const lines = [];
+  if (`${eventName || ""}`.trim()) {
+    lines.push(`event: ${`${eventName}`.trim()}`);
+  }
+  lines.push(`data: ${payloadText}`);
+  return Buffer.from(`${lines.join("\n")}\n\n`);
+}
+
+function createEmptySseInspectionResult() {
+  return {
+    reasoning: null,
+    usage: null,
+    response_id: null,
+    thread_id: null,
+    retryable_upstream_error: null,
+  };
+}
+
+function mergeSseInspectionResult(target, next) {
+  if (Number.isInteger(next.reasoning)) {
+    target.reasoning = next.reasoning;
+  }
+  target.usage = mergeUsageSnapshots(target.usage, next.usage);
+  target.response_id = target.response_id || next.response_id || null;
+  target.thread_id = target.thread_id || next.thread_id || null;
+  target.retryable_upstream_error = next.retryable_upstream_error || target.retryable_upstream_error || null;
+  return target;
+}
+
+function inspectSseBlocks(blocks, config) {
+  const result = createEmptySseInspectionResult();
+  for (const block of blocks) {
+    const lines = `${block || ""}`
+      .split(/\r?\n/)
+      .map((line) => line.trimEnd())
+      .filter(Boolean);
+    const eventName = lines
+      .filter((line) => line.startsWith("event:"))
+      .map((line) => line.replace(/^event:\s?/, "").trim())
+      .find(Boolean) || "";
+    const dataLines = lines
+      .filter((line) => line.startsWith("data:"))
+      .map((line) => line.replace(/^data:\s?/, ""));
+
+    if (dataLines.length === 0) {
+      continue;
+    }
+    const payloadText = dataLines.join("\n");
+    if (payloadText === "[DONE]") {
+      continue;
+    }
+    let parsed = null;
+    try {
+      parsed = JSON.parse(payloadText);
+      const reasoning = extractReasoningTokens(parsed);
+      if (reasoning !== null) {
+        result.reasoning = reasoning;
+      }
+      result.usage = mergeUsageSnapshots(result.usage, normalizeUsageSnapshot(parsed));
+      result.response_id = result.response_id || extractStreamingResponseId(parsed) || extractNonStreamingResponseId(parsed);
+      result.thread_id = result.thread_id || extractResponseThreadId(parsed);
+    } catch {
+      // ignore malformed SSE payloads
+    }
+    const retryableUpstreamError = findRetryableStreamErrorMatch(
+      config,
+      parsed,
+      payloadText,
+      eventName,
+    );
+    if (retryableUpstreamError) {
+      result.retryable_upstream_error = retryableUpstreamError;
+    }
+  }
+  return result;
+}
+
+function flushSseInspectionRemainder(state, config) {
+  if (!state?.buffer || !state.buffer.trim()) {
+    state.buffer = "";
+    return createEmptySseInspectionResult();
+  }
+  const remainder = state.buffer;
+  state.buffer = "";
+  return inspectSseBlocks([remainder], config);
+}
+
+function createResponsesCodexSseState() {
+  return {
+    decoder: new TextDecoder("utf8"),
+    buffer: "",
+    saw_response_completed: false,
+    saw_terminal_failure: false,
+    response_id: null,
+    usage: null,
+  };
+}
+
+function normalizeResponsesFailedPayloadForCodex(parsed, eventName = "", fallbackResponseId = null) {
+  const eventType = firstNonEmptyString(parsed?.type, eventName);
+  if (eventType !== "error" && eventType !== "response.failed") {
+    return null;
+  }
+  const normalized = cloneJsonLike(parsed) || {};
+  normalized.type = "response.failed";
+  if (!normalized.response || typeof normalized.response !== "object" || Array.isArray(normalized.response)) {
+    normalized.response = {};
+  }
+  if (!firstNonEmptyString(normalized.response.status)) {
+    normalized.response.status = "failed";
+  }
+  const responseId = firstNonEmptyString(
+    normalized.response.id,
+    normalized.id,
+    fallbackResponseId,
+  );
+  if (responseId && !firstNonEmptyString(normalized.response.id)) {
+    normalized.response.id = responseId;
+  }
+  if (!normalized.response.error || typeof normalized.response.error !== "object" || Array.isArray(normalized.response.error)) {
+    if (normalized.error && typeof normalized.error === "object" && !Array.isArray(normalized.error)) {
+      normalized.response.error = cloneJsonLike(normalized.error);
+    } else {
+      const fallbackError = {};
+      const message = firstNonEmptyString(normalized.message, normalized.error_message);
+      const code = firstNonEmptyString(normalized.code, normalized.error_code);
+      const errorType = firstNonEmptyString(normalized.error_type, normalized.type);
+      if (message) {
+        fallbackError.message = message;
+      }
+      if (code) {
+        fallbackError.code = code;
+      }
+      if (errorType) {
+        fallbackError.type = errorType;
+      }
+      if (Object.keys(fallbackError).length > 0) {
+        normalized.response.error = fallbackError;
+      }
+    }
+  }
+  for (const key of ["usage", "output", "incomplete_details", "metadata"]) {
+    if (normalized.response[key] === undefined && normalized[key] !== undefined) {
+      normalized.response[key] = cloneJsonLike(normalized[key]);
+    }
+  }
+  return normalized;
+}
+
+function normalizeResponsesCompletedPayloadForCodex(parsed, eventName = "", fallbackResponseId = null) {
+  const eventType = firstNonEmptyString(parsed?.type, eventName);
+  if (eventType !== "response.done") {
+    return null;
+  }
+  const normalized = cloneJsonLike(parsed) || {};
+  normalized.type = "response.completed";
+  if (!normalized.response || typeof normalized.response !== "object" || Array.isArray(normalized.response)) {
+    normalized.response = {};
+  }
+  const responseId = firstNonEmptyString(
+    normalized.response.id,
+    normalized.id,
+    fallbackResponseId,
+  );
+  if (responseId && !firstNonEmptyString(normalized.response.id)) {
+    normalized.response.id = responseId;
+  }
+  if (normalized.response.usage === undefined && normalized.usage !== undefined) {
+    normalized.response.usage = cloneJsonLike(normalized.usage);
+  }
+  if (normalized.response.end_turn === undefined && normalized.end_turn !== undefined) {
+    normalized.response.end_turn = normalized.end_turn;
+  }
+  return normalized;
+}
+
+function processResponsesSseBlockForCodex(state, blockText, fallbackResponseId = null) {
+  const lines = `${blockText || ""}`
+    .split(/\r?\n/)
+    .map((line) => line.trimEnd());
+  const eventName = lines
+    .filter((line) => line.startsWith("event:"))
+    .map((line) => line.replace(/^event:\s?/, "").trim())
+    .find(Boolean) || "";
+  const dataLines = lines
+    .filter((line) => line.startsWith("data:"))
+    .map((line) => line.replace(/^data:\s?/, ""));
+
+  if (dataLines.length === 0) {
+    return Buffer.from(`${lines.join("\n")}\n\n`);
+  }
+
+  const payloadText = dataLines.join("\n");
+  if (payloadText === "[DONE]") {
+    return buildSseBlock(eventName, payloadText);
+  }
+
+  let parsed = null;
+  try {
+    parsed = JSON.parse(payloadText);
+  } catch {
+    return Buffer.from(`${lines.join("\n")}\n\n`);
+  }
+
+  state.usage = mergeUsageSnapshots(state.usage, normalizeUsageSnapshot(parsed));
+  state.response_id = state.response_id
+    || extractStreamingResponseId(parsed)
+    || extractNonStreamingResponseId(parsed)
+    || firstNonEmptyString(parsed?.response?.id, fallbackResponseId);
+
+  let rewritten = normalizeResponsesCompletedPayloadForCodex(parsed, eventName, state.response_id || fallbackResponseId);
+  if (!rewritten) {
+    rewritten = normalizeResponsesFailedPayloadForCodex(parsed, eventName, state.response_id || fallbackResponseId);
+  }
+  const outputPayload = rewritten || parsed;
+  const outputEventName = firstNonEmptyString(outputPayload?.type, eventName);
+  state.response_id = state.response_id
+    || extractStreamingResponseId(outputPayload)
+    || extractNonStreamingResponseId(outputPayload)
+    || firstNonEmptyString(outputPayload?.response?.id, fallbackResponseId);
+  state.usage = mergeUsageSnapshots(state.usage, normalizeUsageSnapshot(outputPayload));
+
+  if (outputEventName === "response.completed") {
+    state.saw_response_completed = true;
+  }
+  if (outputEventName === "response.failed" || outputEventName === "response.incomplete") {
+    state.saw_terminal_failure = true;
+  }
+
+  if (!rewritten) {
+    return Buffer.from(`${lines.join("\n")}\n\n`);
+  }
+  return buildSseBlock(outputEventName, JSON.stringify(outputPayload));
+}
+
+function drainResponsesSseForCodex(state, chunk, fallbackResponseId = null) {
+  const decoded = state.decoder.decode(chunk, { stream: true });
+  state.buffer += decoded;
+  const blocks = state.buffer.split(/\r?\n\r?\n/);
+  state.buffer = blocks.pop() ?? "";
+  return blocks
+    .filter((block) => block.length > 0)
+    .map((block) => processResponsesSseBlockForCodex(state, block, fallbackResponseId));
+}
+
+function flushResponsesSseForCodex(state, fallbackResponseId = null, fallbackUsage = null) {
+  const flushed = state.decoder.decode();
+  if (flushed) {
+    state.buffer += flushed;
+  }
+  const outputs = [];
+  if (state.buffer.trim()) {
+    outputs.push(processResponsesSseBlockForCodex(state, state.buffer, fallbackResponseId));
+  }
+  state.buffer = "";
+
+  if (!state.saw_response_completed && !state.saw_terminal_failure) {
+    const responseId = firstNonEmptyString(state.response_id, fallbackResponseId);
+    if (responseId) {
+      const usage = denormalizeUsageSnapshotToResponsesUsage(mergeUsageSnapshots(state.usage, fallbackUsage));
+      const payload = {
+        type: "response.completed",
+        response: {
+          id: responseId,
+        },
+      };
+      if (usage) {
+        payload.response.usage = usage;
+      }
+      outputs.push(buildSseBlock("response.completed", JSON.stringify(payload)));
+      state.saw_response_completed = true;
+    }
+  }
+  return outputs;
+}
+
 function normalizeIntegerList(values, fallback = []) {
   const source = values === undefined || values === null ? fallback : values;
   const normalized = flattenValues(source)
@@ -3425,62 +3738,9 @@ async function fetchUpstreamWithRetry(upstreamUrl, init, config, logger, request
 function inspectSseChunk(state, chunk, config) {
   const decoded = state.decoder.decode(chunk, { stream: true });
   state.buffer += decoded;
-
-  const result = {
-    reasoning: null,
-    usage: null,
-    response_id: null,
-    thread_id: null,
-    retryable_upstream_error: null,
-  };
-
   const blocks = state.buffer.split(/\r?\n\r?\n/);
   state.buffer = blocks.pop() ?? "";
-
+  return inspectSseBlocks(blocks, config);
-  for (const block of blocks) {
-    const lines = block
-      .split(/\r?\n/)
-      .map((line) => line.trimEnd())
-      .filter(Boolean);
-    const eventName = lines
-      .filter((line) => line.startsWith("event:"))
-      .map((line) => line.replace(/^event:\s?/, "").trim())
-      .find(Boolean) || "";
-    const dataLines = lines
-      .filter((line) => line.startsWith("data:"))
-      .map((line) => line.replace(/^data:\s?/, ""));
-
-    if (dataLines.length === 0) {
-      continue;
-    }
-    const payloadText = dataLines.join("\n");
-    if (payloadText === "[DONE]") {
-      continue;
-    }
-    let parsed = null;
-    try {
-      parsed = JSON.parse(payloadText);
-      const reasoning = extractReasoningTokens(parsed);
-      if (reasoning !== null) {
-        result.reasoning = reasoning;
-      }
-      result.usage = mergeUsageSnapshots(result.usage, normalizeUsageSnapshot(parsed));
-      result.response_id = result.response_id || extractStreamingResponseId(parsed);
-      result.thread_id = result.thread_id || extractResponseThreadId(parsed);
-    } catch {
-      // ignore malformed SSE payloads
-    }
-    const retryableUpstreamError = findRetryableStreamErrorMatch(
-      config,
-      parsed,
-      payloadText,
-      eventName,
-    );
-    if (retryableUpstreamError) {
-      result.retryable_upstream_error = retryableUpstreamError;
-    }
-  }
-  return result;
 }
 
 async function handleNonStreaming({
@@ -3631,6 +3891,11 @@ async function handleStreaming({
     decoder: new TextDecoder("utf8"),
     buffer: "",
   };
+  const codexResponsesSseState = isResponsesReasoningRetryPath(pathname) && isSseContentType(
+    upstreamResponse.headers.get("content-type"),
+  )
+    ? createResponsesCodexSseState()
+    : null;
 
   let wroteAnyChunk = false;
   let observedReasoning = null;
@@ -3735,6 +4000,101 @@ async function handleStreaming({
 
     const { done, value } = readResult;
     if (done) {
+      const finalInspection = flushSseInspectionRemainder(sseState, config);
+      const finalCodexChunks = codexResponsesSseState
+        ? flushResponsesSseForCodex(
+          codexResponsesSseState,
+          requestEntry.response_id || null,
+          observedUsage,
+        )
+        : [];
+      if (Number.isInteger(finalInspection.reasoning)) {
+        observedReasoning = finalInspection.reasoning;
+      }
+      observedUsage = mergeUsageSnapshots(observedUsage, finalInspection.usage);
+      if (finalInspection.response_id) {
+        requestEntry.response_id = finalInspection.response_id;
+      }
+      if (finalInspection.thread_id) {
+        requestEntry.thread_id = finalInspection.thread_id;
+      }
+      if (codexResponsesSseState?.response_id && !requestEntry.response_id) {
+        requestEntry.response_id = codexResponsesSseState.response_id;
+      }
+      observedUsage = mergeUsageSnapshots(observedUsage, codexResponsesSseState?.usage || null);
+      if (finalInspection.retryable_upstream_error) {
+        return {
+          inspected: true,
+          matched: true,
+          retry_requested: strict502Mode || !wroteAnyChunk,
+          retryable_upstream_error: finalInspection.retryable_upstream_error,
+          upstream_status_code: upstreamResponse.status,
+          reasoning_tokens: observedReasoning,
+          usage: observedUsage,
+          error: `retryable upstream error: ${finalInspection.retryable_upstream_error.matched_pattern}`,
+          match_reason: "retryable_upstream_error",
+          response_id: requestEntry.response_id,
+          thread_id: requestEntry.thread_id,
+          response_bytes_received: requestEntry.response_bytes_received,
+          stream_chunk_count: requestEntry.stream_chunk_count,
+        };
+      }
+      if (reasoningMatched(config, observedReasoning)) {
+        recordInspectedResponse(monitor, observedReasoning, true);
+        if (config.log_match) {
+          logger(
+            `[match] stream path=${pathname} reasoning_tokens=${observedReasoning} action=${config.stream_action}`,
+          );
+        }
+        if (captureOnly) {
+          return {
+            inspected: true,
+            matched: true,
+            match_reason: "reasoning_guard",
+            status_code: config.non_stream_status_code,
+            upstream_status_code: upstreamResponse.status,
+            reasoning_tokens: observedReasoning,
+            usage: observedUsage,
+            response_id: requestEntry.response_id,
+            thread_id: requestEntry.thread_id,
+            response_bytes_received: requestEntry.response_bytes_received,
+            stream_chunk_count: requestEntry.stream_chunk_count,
+          };
+        }
+        if (strict502Mode || !wroteAnyChunk) {
+          const blockedBody = buildBlockedBody(pathname, observedReasoning, config.non_stream_status_code);
+          res.writeHead(config.non_stream_status_code, {
+            "content-type": "application/json; charset=utf-8",
+            "x-codex-retry-gateway-reason": "reasoning-guard-triggered",
+          });
+          res.end(blockedBody);
+        } else if (!captureOnly && !res.writableEnded) {
+          res.end();
+        }
+        return {
+          inspected: true,
+          matched: true,
+          match_reason: "reasoning_guard",
+          status_code: config.non_stream_status_code,
+          upstream_status_code: upstreamResponse.status,
+          reasoning_tokens: observedReasoning,
+          usage: observedUsage,
+          response_id: requestEntry.response_id,
+          thread_id: requestEntry.thread_id,
+          response_bytes_received: requestEntry.response_bytes_received,
+          stream_chunk_count: requestEntry.stream_chunk_count,
+        };
+      }
+      if (finalCodexChunks.length > 0) {
+        if (strict502Mode) {
+          bufferedChunks.push(...finalCodexChunks);
+        } else if (!captureOnly) {
+          for (const chunk of finalCodexChunks) {
+            wroteAnyChunk = true;
+            res.write(chunk);
+          }
+        }
+      }
       recordInspectedResponse(monitor, observedReasoning, false);
       if (persistEntry) {
         persistStreamingProgress(runtime, requestEntry, { force: true }, new Date());
@@ -3789,6 +4149,13 @@ async function handleStreaming({
     const chunkBuffer = Buffer.from(value);
     const now = new Date();
     const inspection = inspectSseChunk(sseState, value, config);
+    const outputChunks = codexResponsesSseState
+      ? drainResponsesSseForCodex(
+        codexResponsesSseState,
+        value,
+        requestEntry.response_id || null,
+      )
+      : [chunkBuffer];
     const retryableUpstreamError = inspection.retryable_upstream_error;
     if (retryableUpstreamError) {
       abortController.abort();
@@ -3878,11 +4245,16 @@ async function handleStreaming({
     }
 
     if (strict502Mode) {
-      bufferedChunks.push(chunkBuffer);
+      bufferedChunks.push(...outputChunks);
     } else {
-      wroteAnyChunk = true;
       if (!captureOnly) {
-        res.write(chunkBuffer);
+        for (const outputChunk of outputChunks) {
+          if (!outputChunk || outputChunk.length === 0) {
+            continue;
+          }
+          wroteAnyChunk = true;
+          res.write(outputChunk);
+        }
       }
     }
   }

scripts/test-gateway-e2e.mjs

@@ -1043,6 +1043,26 @@ async function run() {
     assert(streamDisconnectedRetryResponse.status === 200, `stream disconnected capacity 抖动后未自动恢复: ${streamDisconnectedRetryResponse.status}`);
     assert(streamDisconnectedRetryText.includes("hello"), "stream disconnected capacity 恢复后未拿到正常 SSE 内容");
 
+    const normalizedFailureStream = await readSseUntilClose(
+      `http://127.0.0.1:${gatewayPort}/responses`,
+      {
+        stream: true,
+        test_capacity_before_success_times: 1,
+        test_capacity_message: "Permanent upstream failure for codex normalization test.",
+        test_capacity_stream_event_name: "error",
+        test_capacity_stream_payload_shape: "default",
+      },
+    );
+    assert(normalizedFailureStream.status === 200, `非重试 fatal stream 首状态异常: ${normalizedFailureStream.status}`);
+    assert(
+      normalizedFailureStream.text.includes('"type":"response.failed"'),
+      "非重试 fatal stream 未归一化为 response.failed",
+    );
+    assert(
+      !normalizedFailureStream.text.includes('"type":"error"'),
+      "非重试 fatal stream 不应继续透传 type=error",
+    );
+
     for (const streamPath of [
       "/responses",
       "/v1/responses",
@@ -1069,6 +1089,12 @@ async function run() {
       assert(okStream.status === 200, `${streamPath} 128 首状态异常: ${okStream.status}`);
       assert(okStream.text.includes("[DONE]"), `${streamPath} 流式 128 未完整结束`);
       assert(!okStream.closedByError, `${streamPath} 流式 128 不应异常断开`);
+      if (streamPath === "/responses" || streamPath === "/v1/responses") {
+        assert(
+          okStream.text.includes('"type":"response.completed"'),
+          `${streamPath} 成功流未补 response.completed`,
+        );
+      }
 
       if (streamPath === "/responses" || streamPath === "/v1/responses") {
         const replayedStream = await readSseUntilClose(