fix: pace captured SSE replay

2026-07-11 13:16:40 +08:00
parent 7d0a40e1ea
commit 29eb91e0c3
3 changed files with 78 additions and 16 deletions
@@ -24,7 +24,7 @@
 - 上游若返回明确的容量错误（默认匹配错误文案 `Selected model is at capacity. Please try a different model.`，以及 `stream disconnected before completion: Concurrency limit exceeded for account, please retry later`），也会自动重试；重试耗尽后转成本地 `502`
 - 除了 `429/503` JSON 错误响应，也会识别 `200` 但返回体本质是错误、以及流式失败事件里携带同样文案的情况
 - 流式命中时默认先缓存并判断；一旦命中 `516`，统一返回 `502`
- 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间让出一次事件循环，不会伪造缺少 response ID 的生命周期事件
+- 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间至少间隔 5ms，并关闭 TCP 小包聚合，避免大量 delta 在同一事件循环突发到达；不会伪造缺少 response ID 的生命周期事件
 - 默认同时拦截 root 路径和 `/v1` 路径：
  - `/responses`
  - `/chat/completions`
@@ -4079,7 +4079,7 @@ function cloneResponseHeaders(sourceHeaders) {
  return headers;
 }

-const CAPTURED_STREAM_REPLAY_CHUNKS_PER_TICK = 1;
+const CAPTURED_STREAM_REPLAY_DELAY_MS = 5;

 function cloneBufferList(chunks) {
  if (!Array.isArray(chunks) || chunks.length === 0) {
@@ -4091,13 +4091,16 @@ function cloneBufferList(chunks) {
 }

 async function writeBufferedStreamChunks(res, chunks) {
-  let writesSinceYield = 0;
-  for (const chunk of chunks || []) {
-    if (!chunk || chunk.length === 0 || res.destroyed || res.writableEnded) {
-      continue;
+  const replayChunks = (chunks || []).filter((chunk) => chunk && chunk.length > 0);
+  res.socket?.setNoDelay(true);
+  res.flushHeaders?.();
+
+  for (let index = 0; index < replayChunks.length; index += 1) {
+    if (res.destroyed || res.writableEnded) {
+      break;
    }
+    const chunk = replayChunks[index];
    const accepted = res.write(chunk);
-    writesSinceYield += 1;
    if (!accepted) {
      await new Promise((resolve) => {
        let settled = false;
@@ -4117,12 +4120,13 @@ async function writeBufferedStreamChunks(res, chunks) {
        res.once("close", onClose);
        res.once("error", onClose);
      });
-      writesSinceYield = 0;
-      continue;
    }
-    if (writesSinceYield >= CAPTURED_STREAM_REPLAY_CHUNKS_PER_TICK) {
-      writesSinceYield = 0;
-      await new Promise((resolve) => setImmediate(resolve));
+    if (
+      index < replayChunks.length - 1 &&
+      !res.destroyed &&
+      !res.writableEnded
+    ) {
+      await sleep(CAPTURED_STREAM_REPLAY_DELAY_MS);
    }
  }
 }
@@ -303,6 +303,7 @@ function startFakeUpstream(port, options = {}) {
            ? parsed.test_stream_delta_chunks
            : 1;
          const deltaTextBase = parsed.test_stream_delta_text || "hello";
+          const outputItemId = "msg_stream";
          const lifecyclePayloadExtra = parsed.test_stream_lifecycle_marker
            ? {
              marker: parsed.test_stream_lifecycle_marker,
@@ -329,17 +330,46 @@ function startFakeUpstream(port, options = {}) {
                  output: lifecyclePayloadExtra,
                },
              })}\n\n`,
+              `data: ${JSON.stringify({
+                type: "response.output_item.added",
+                output_index: 0,
+                item: {
+                  id: outputItemId,
+                  type: "message",
+                  status: "in_progress",
+                  role: "assistant",
+                  content: [],
+                },
+              })}\n\n`,
            ]
            : [];
-          const streamChunks = Array.from({ length: deltaChunkCount }, (_, index) => {
+          const deltaTexts = Array.from({ length: deltaChunkCount }, (_, index) => {
            const deltaText = deltaChunkCount === 1 ? deltaTextBase : `${deltaTextBase}-${index + 1}`;
-            return `data: ${JSON.stringify({ type: "response.output_text.delta", delta: deltaText, response_id: "resp_stream", thread_id: parsed.thread_id || "thread_stream", retry_attempt: reasoningAttempt })}\n\n`;
+            return deltaText;
+          });
+          const streamChunks = deltaTexts.map((deltaText) => {
+            return `data: ${JSON.stringify({ type: "response.output_text.delta", item_id: outputItemId, output_index: 0, content_index: 0, delta: deltaText, response_id: "resp_stream", thread_id: parsed.thread_id || "thread_stream", retry_attempt: reasoningAttempt })}\n\n`;
          });
          if (parsed.test_stream_include_lifecycle) {
            streamChunks.unshift(...lifecycleChunks);
          }
          if (parsed.test_stream_include_lifecycle) {
            streamChunks.push(
+              `data: ${JSON.stringify({
+                type: "response.output_item.done",
+                output_index: 0,
+                item: {
+                  id: outputItemId,
+                  type: "message",
+                  status: "completed",
+                  role: "assistant",
+                  content: [{
+                    type: "output_text",
+                    text: deltaTexts.join(""),
+                    annotations: [],
+                  }],
+                },
+              })}\n\n`,
              `data: ${JSON.stringify({
                type: "response.completed",
                response: {
@@ -553,6 +583,7 @@ async function readSseUntilClose(url, requestBody) {
  }

  text += decoder.decode();
+  const elapsedMs = Date.now() - startedAt;
  return {
    status: response.status,
    headers: response.headers,
@@ -560,6 +591,7 @@ async function readSseUntilClose(url, requestBody) {
    closedByError,
    readCount,
    firstChunkDelayMs,
+    elapsedMs,
  };
 }

@@ -1650,6 +1682,11 @@ async function run() {
    );
    assert(capturedLifecycleStream.status === 200, `/responses 捕获回放状态异常: ${capturedLifecycleStream.status}`);
    assert(capturedLifecycleStream.readCount > 1, "/responses 捕获回放不应退化为单块响应");
+    const observedReplayDurationMs = capturedLifecycleStream.elapsedMs - capturedLifecycleStream.firstChunkDelayMs;
+    assert(
+      observedReplayDurationMs >= 250,
+      `/responses 捕获回放缺少真实传输间隔: ${observedReplayDurationMs}ms`,
+    );
    const capturedEvents = parseSseEvents(capturedLifecycleStream.text);
    const firstReplayEvent = capturedEvents.find((event) => event.payload);
    assert(
@@ -1680,11 +1717,32 @@ async function run() {
      capturedDeltas.every((delta, index) => delta === `captured-chunk-${index + 1}`),
      "/responses 捕获回放的长流 delta 顺序不完整",
    );
+    const outputItemAddedIndex = capturedEvents.findIndex(
+      (event) => event.payload?.type === "response.output_item.added",
+    );
+    const outputItemDoneIndex = capturedEvents.findIndex(
+      (event) => event.payload?.type === "response.output_item.done",
+    );
+    const firstDeltaIndex = capturedEvents.findIndex(
+      (event) => event.payload?.type === "response.output_text.delta",
+    );
    const completedIndex = capturedEvents.findIndex((event) => event.payload?.type === "response.completed");
    const lastDeltaIndex = capturedEvents
      .map((event) => event.payload?.type)
      .lastIndexOf("response.output_text.delta");
-    assert(completedIndex > lastDeltaIndex, "/responses 捕获回放未在所有 delta 后终止");
+    assert(
+      outputItemAddedIndex >= 0 && outputItemAddedIndex < firstDeltaIndex,
+      "/responses 捕获回放未在 delta 前保留 output_item.added",
+    );
+    assert(
+      outputItemDoneIndex > lastDeltaIndex && completedIndex > outputItemDoneIndex,
+      "/responses 捕获回放未按 delta、output_item.done、response.completed 顺序终止",
+    );
+    const completedOutputItem = capturedEvents[outputItemDoneIndex]?.payload?.item;
+    assert(
+      completedOutputItem?.content?.[0]?.text === capturedDeltas.join(""),
+      "/responses 捕获回放的 output_item.done 与全部 delta 内容不一致",
+    );
    assert(!capturedLifecycleStream.text.includes("captured-lifecycle-marker"), "/responses lifecycle 归一化仍透传了巨大的生命周期原始 payload");

    const normalizedLifecycleStream = await readSseUntilClose(