diff --git a/README.md b/README.md index 6154f7e..e9190fc 100644 --- a/README.md +++ b/README.md @@ -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` diff --git a/gateway.mjs b/gateway.mjs index 65fb4fa..9d004a7 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -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); } } } diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index 86f2703..0865ce2 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -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: { @@ -552,7 +582,8 @@ async function readSseUntilClose(url, requestBody) { } } - text += decoder.decode(); + 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(