From 7d0a40e1ea636af0926ea7a2151a6846edd80efb Mon Sep 17 00:00:00 2001 From: yunyaozhou Date: Sat, 11 Jul 2026 02:36:00 +0800 Subject: [PATCH] fix: preserve responses SSE lifecycle during replay --- README.md | 7 +- gateway.mjs | 173 +++-------------------------------- scripts/test-gateway-e2e.mjs | 106 ++++++++++++++------- 3 files changed, 89 insertions(+), 197 deletions(-) diff --git a/README.md b/README.md index d226b1a..6154f7e 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` -- 流式成功响应在严格检查模式下仍会先缓存完成，但会按分块 SSE 回放给 Codex，而不是单次大包 `res.end(...)` +- 流式成功响应在严格检查模式下仍会先缓存完成；成功后会按真实且规范化的 Responses 生命周期与输出顺序逐块回放给 Codex，每个 SSE 块之间让出一次事件循环，不会伪造缺少 response ID 的生命周期事件 - 默认同时拦截 root 路径和 `/v1` 路径： - `/responses` - `/chat/completions` @@ -241,7 +241,7 @@ codex --dangerously-bypass-approvals-and-sandbox -c model="gpt-5.4" hello 建议同时检查： - `curl http://:4610/__codex_retry_gateway/health` -- `/responses` 成功流能完整结束，不会卡 pending，必要时应能看到 `response.completed` +- `/responses` 成功流能完整结束，不会卡 pending；`response.created` 必须带真实 response ID，随后依次可见 `response.in_progress`、输出 delta 与 `response.completed` - 容量错误 `Selected model is at capacity. Please try a different model.` 仍能按网关策略自动重试 如果当前对话依赖本机 `4610`，不要把本机作为首个发布目标；先在另一台已接管相同 profile 的机器验证，再回到本机切换。 @@ -395,6 +395,9 @@ macOS / Linux: ~/.codex-retry-gateway - `test-gateway-e2e.ps1` - 已通过 - 验证 `/responses`、`/chat/completions`、`/v1/responses`、`/v1/chat/completions` +- `node scripts/test-gateway-e2e.mjs` + - 已通过 + - 验证 strict capture 下 `/responses` 生命周期不注入空事件，长流 delta 按序完整回放并以 `response.completed` 结束 - `test-install-restore.ps1` - 已通过 - 验证安装、透传、UI 页面、热更新配置、实时日志、516 统计、恢复闭环 diff --git a/gateway.mjs b/gateway.mjs index 478bf9a..65fb4fa 100644 --- a/gateway.mjs +++ b/gateway.mjs @@ -538,13 +538,7 @@ function buildSseBlock(eventName, payloadText) { return Buffer.from(`${lines.join("\n")}\n\n`); } -const RESPONSES_PREVIEW_SUPPRESSED_EVENT_NAMES = new Set([ - "response.created", - "response.in_progress", -]); -const RESPONSES_CAPTURE_PREVIEW_START_AFTER_MS = 1000; const RESPONSES_MODEL_HEADER_NAMES = new Set(["openai-model", "x-openai-model"]); -const RESPONSES_CAPTURE_PREVIEW_HEARTBEAT_MS = 2000; function normalizeResponsesHeaderSubset(...sources) { const headers = {}; @@ -562,95 +556,6 @@ function normalizeResponsesHeaderSubset(...sources) { return Object.keys(headers).length > 0 ? headers : null; } -function buildResponsesPreviewPayload(type = "response.in_progress") { - return { - type, - response: {}, - }; -} - -function buildResponsesFailedPayload(message, options = {}) { - const payload = { - type: "response.failed", - response: { - status: "failed", - error: { - message: firstNonEmptyString(message, "response failed"), - type: firstNonEmptyString(options.errorType, "server_error"), - }, - }, - }; - const responseId = firstNonEmptyString(options.responseId); - if (responseId) { - payload.response.id = responseId; - } - const code = firstNonEmptyString(options.code); - if (code) { - payload.response.error.code = code; - } - return payload; -} - -function buildResponsesFailedSseDelivery(message, options = {}) { - const payload = buildResponsesFailedPayload(message, options); - const chunk = buildSseBlock("response.failed", JSON.stringify(payload)); - return buildCapturedDelivery( - 200, - new Headers({ - "content-type": "text/event-stream; charset=utf-8", - "cache-control": "no-cache", - connection: "keep-alive", - }), - chunk, - { stream_chunks: [chunk] }, - ); -} - -function createResponsesCapturedPreview(res) { - return { - res, - created_at_ms: Date.now(), - started: false, - last_sent_at_ms: 0, - }; -} - -function maybeWriteResponsesCapturedPreview(preview, sourceHeaders = null, nowMs = Date.now()) { - if (!preview?.res || preview.res.writableEnded || preview.res.destroyed) { - return false; - } - if (!preview.started) { - if (nowMs - preview.created_at_ms < RESPONSES_CAPTURE_PREVIEW_START_AFTER_MS) { - return false; - } - preview.started = true; - preview.last_sent_at_ms = nowMs; - if (sourceHeaders) { - copyHeadersToClient(sourceHeaders, preview.res); - } - preview.res.writeHead(200, { - "content-type": "text/event-stream; charset=utf-8", - "cache-control": "no-cache", - connection: "keep-alive", - }); - preview.res.write( - buildSseBlock("response.created", JSON.stringify(buildResponsesPreviewPayload("response.created"))), - ); - return true; - } - if (nowMs - preview.last_sent_at_ms < RESPONSES_CAPTURE_PREVIEW_HEARTBEAT_MS) { - return false; - } - preview.last_sent_at_ms = nowMs; - preview.res.write( - buildSseBlock( - "response.in_progress", - JSON.stringify(buildResponsesPreviewPayload("response.in_progress")), - ), - ); - return true; -} - function createEmptySseInspectionResult() { return { reasoning: null, @@ -845,7 +750,7 @@ function normalizeResponsesCompletedPayloadForCodex(parsed, eventName = "", fall return normalizeResponsesLifecyclePayloadForCodex(parsed, eventName, fallbackResponseId); } -function processResponsesSseBlockForCodex(state, blockText, fallbackResponseId = null, options = {}) { +function processResponsesSseBlockForCodex(state, blockText, fallbackResponseId = null) { const lines = `${blockText || ""}` .split(/\r?\n/) .map((line) => line.trimEnd()); @@ -905,37 +810,31 @@ function processResponsesSseBlockForCodex(state, blockText, fallbackResponseId = state.saw_terminal_failure = true; } - if ( - options?.suppressLifecycleEvents && - RESPONSES_PREVIEW_SUPPRESSED_EVENT_NAMES.has(outputEventName) - ) { - return Buffer.alloc(0); - } if (!rewritten) { return Buffer.from(`${lines.join("\n")}\n\n`); } return buildSseBlock(outputEventName, JSON.stringify(outputPayload)); } -function drainResponsesSseForCodex(state, chunk, fallbackResponseId = null, options = {}) { +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, options)) + .map((block) => processResponsesSseBlockForCodex(state, block, fallbackResponseId)) .filter((chunkBuffer) => chunkBuffer && chunkBuffer.length > 0); } -function flushResponsesSseForCodex(state, fallbackResponseId = null, fallbackUsage = null, options = {}) { +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, options)); + outputs.push(processResponsesSseBlockForCodex(state, state.buffer, fallbackResponseId)); } state.buffer = ""; @@ -4180,7 +4079,7 @@ function cloneResponseHeaders(sourceHeaders) { return headers; } -const CAPTURED_STREAM_REPLAY_CHUNKS_PER_TICK = 16; +const CAPTURED_STREAM_REPLAY_CHUNKS_PER_TICK = 1; function cloneBufferList(chunks) { if (!Array.isArray(chunks) || chunks.length === 0) { @@ -4244,20 +4143,6 @@ async function writeCapturedResponse(res, delivery) { res.end(delivery.body); } -async function appendCapturedResponse(res, delivery) { - if (!delivery) { - throw new Error("missing captured response delivery"); - } - if (Array.isArray(delivery.stream_chunks) && delivery.stream_chunks.length > 0) { - await writeBufferedStreamChunks(res, delivery.stream_chunks); - } else if (delivery.body?.length) { - res.write(delivery.body); - } - if (!res.writableEnded) { - res.end(); - } -} - function buildCapturedDelivery(statusCode, headers, body, options = {}) { const delivery = { status_code: statusCode, @@ -4933,7 +4818,6 @@ async function handleStreaming({ requestAbortSignal = null, captureOnly = false, persistEntry = true, - capturedResponsesPreview = null, }) { const strict502Mode = captureOnly || config.stream_action !== "disconnect"; const reader = upstreamResponse.body.getReader(); @@ -5056,7 +4940,6 @@ async function handleStreaming({ codexResponsesSseState, requestEntry.response_id || null, observedUsage, - { suppressLifecycleEvents: Boolean(capturedResponsesPreview?.started) }, ) : []; if (Number.isInteger(finalInspection.reasoning)) { @@ -5290,15 +5173,11 @@ async function handleStreaming({ }; } - if (captureOnly && codexResponsesSseState && capturedResponsesPreview) { - maybeWriteResponsesCapturedPreview(capturedResponsesPreview, upstreamResponse.headers, now.getTime()); - } const outputChunks = codexResponsesSseState ? drainResponsesSseForCodex( codexResponsesSseState, value, requestEntry.response_id || null, - { suppressLifecycleEvents: Boolean(capturedResponsesPreview?.started) }, ) : [chunkBuffer]; @@ -5377,7 +5256,6 @@ async function executeGatewayQuery({ captureOnly = false, persistEntry = true, externalAbortSignals = [], - capturedResponsesPreview = null, }) { const maxUpstreamAttempts = normalizePositiveInteger( config.upstream_fetch_retry_attempts, @@ -5481,7 +5359,6 @@ async function executeGatewayQuery({ requestAbortSignal: queryAbortLink.controller.signal, captureOnly, persistEntry, - capturedResponsesPreview, }) : await handleNonStreaming({ runtime, @@ -5710,7 +5587,6 @@ async function runResponsesReasoningRetry({ monitor, pathname, req, - res, requestBody, requestIsStream, upstreamUrl, @@ -5721,9 +5597,6 @@ async function runResponsesReasoningRetry({ let round = 1; let totalUpstreamAttempts = 0; const retryExtraState = createReasoningRetryExtraState(); - const capturedResponsesPreview = requestIsStream && res && isResponsesReasoningRetryPath(pathname) - ? createResponsesCapturedPreview(res) - : null; while (true) { if (clientAbortSignal?.aborted) { @@ -5766,7 +5639,6 @@ async function runResponsesReasoningRetry({ upstreamAuth, requestEntry: attemptEntry, externalAbortSignals: [clientAbortSignal, slotController.signal], - capturedResponsesPreview, }); return new Promise((resolve, reject) => { let settled = false; @@ -5852,7 +5724,6 @@ async function runResponsesReasoningRetry({ ...winner.result, response_stream: winner.result.response_stream, total_upstream_attempts: totalUpstreamAttempts, - responses_capture_preview_started: Boolean(capturedResponsesPreview?.started), }; } @@ -5875,18 +5746,8 @@ async function runResponsesReasoningRetry({ requestEntry.reasoning_retry_stop_reason = fatalOutcome.result?.match_reason || "fatal"; return { ...fatalOutcome.result, - delivery: capturedResponsesPreview?.started - ? buildResponsesFailedSseDelivery( - fatalOutcome.result?.error || "reasoning retry fatal stream failure", - { - responseId: requestEntry.response_id || fatalOutcome.result?.response_id || null, - code: "reasoning_retry_failed", - }, - ) - : fatalOutcome.result?.delivery, response_stream: fatalOutcome.result.response_stream, total_upstream_attempts: totalUpstreamAttempts, - responses_capture_preview_started: Boolean(capturedResponsesPreview?.started), }; } @@ -5899,22 +5760,16 @@ async function runResponsesReasoningRetry({ return { inspected: false, matched: false, - status_code: capturedResponsesPreview?.started ? 200 : 502, + status_code: 502, upstream_status_code: null, error: "reasoning retry ended without a successful response", response_stream: requestIsStream, total_upstream_attempts: totalUpstreamAttempts, - delivery: capturedResponsesPreview?.started - ? buildResponsesFailedSseDelivery("reasoning retry ended without a successful response", { - responseId: requestEntry.response_id || null, - code: "reasoning_retry_exhausted", - }) - : buildCapturedDelivery( - 502, - new Headers({ "content-type": "application/json; charset=utf-8" }), - buildGatewayErrorBody("reasoning retry ended without a successful response"), - ), - responses_capture_preview_started: Boolean(capturedResponsesPreview?.started), + delivery: buildCapturedDelivery( + 502, + new Headers({ "content-type": "application/json; charset=utf-8" }), + buildGatewayErrorBody("reasoning retry ended without a successful response"), + ), }; } } @@ -6014,7 +5869,6 @@ async function proxyRequest(runtime, req, res) { monitor: runtime.monitor, pathname, req, - res, requestBody, requestIsStream, upstreamUrl, @@ -6041,13 +5895,10 @@ async function proxyRequest(runtime, req, res) { const { delivery, total_upstream_attempts: observedUpstreamAttempts, - responses_capture_preview_started: responsesCapturePreviewStarted, ...resultFields } = result; if (delivery && !res.headersSent) { await writeCapturedResponse(res, delivery); - } else if (delivery && responsesCapturePreviewStarted && !res.writableEnded) { - await appendCapturedResponse(res, delivery); } if (Number.isInteger(observedUpstreamAttempts)) { requestEntry.upstream_attempt_count = observedUpstreamAttempts; diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index 611127e..86f2703 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/scripts/test-gateway-e2e.mjs @@ -562,6 +562,30 @@ async function readSseUntilClose(url, requestBody) { firstChunkDelayMs, }; } + +function parseSseEvents(text) { + return `${text || ""}` + .split(/\r?\n\r?\n/) + .filter((block) => block.trim()) + .map((block) => { + const lines = block.split(/\r?\n/); + const eventName = lines + .filter((line) => line.startsWith("event:")) + .map((line) => line.replace(/^event:\s?/, "").trim()) + .find(Boolean) || ""; + const payloadText = lines + .filter((line) => line.startsWith("data:")) + .map((line) => line.replace(/^data:\s?/, "")) + .join("\n"); + let payload = null; + try { + payload = JSON.parse(payloadText); + } catch { + // [DONE] and malformed test payloads intentionally remain raw. + } + return { eventName, payloadText, payload }; + }); +} async function run() { assert( @@ -1611,43 +1635,57 @@ async function run() { } } - let previewResolved = false; - const previewFetchPromise = fetch(`http://127.0.0.1:${gatewayPort}/responses`, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ + const capturedLifecycleStream = await readSseUntilClose( + `http://127.0.0.1:${gatewayPort}/responses`, + { stream: true, - thread_id: "thread-preview", + thread_id: "thread-captured-lifecycle", test_reasoning_tokens: 128, test_stream_include_lifecycle: true, - test_stream_lifecycle_marker: "preview-marker", - test_stream_delta_chunks: 4, - test_stream_chunk_delay_ms: 320, - }), - }).then((response) => { - previewResolved = true; - return response; - }); - await new Promise((resolve) => setTimeout(resolve, 1350)); - assert(previewResolved, "responses thread 重打预览未在首个上游 chunk 后尽早返回头部"); - const previewResponse = await previewFetchPromise; - assert(previewResponse.status === 200, `responses thread 重打预览状态异常: ${previewResponse.status}`); - const previewReader = previewResponse.body.getReader(); - const previewDecoder = new TextDecoder(); - const firstPreviewRead = await previewReader.read(); - const firstPreviewText = previewDecoder.decode(firstPreviewRead.value || new Uint8Array(), { stream: true }); - assert(firstPreviewText.includes('"type":"response.created"'), "responses thread 重打预览未先发 response.created"); - let previewText = firstPreviewText; - while (true) { - const { done, value } = await previewReader.read(); - if (done) { - break; - } - previewText += previewDecoder.decode(value, { stream: true }); - } - previewText += previewDecoder.decode(); - assert(previewText.includes('"delta":"hello-1"'), "responses thread 重打预览未回放最终 delta"); - assert(!previewText.includes("preview-marker"), "responses thread 重打预览不应透传巨大的 lifecycle 原始 payload"); + test_stream_lifecycle_marker: "captured-lifecycle-marker", + test_stream_delta_chunks: 64, + test_stream_delta_text: "captured-chunk", + test_stream_chunk_delay_ms: 2, + }, + ); + assert(capturedLifecycleStream.status === 200, `/responses 捕获回放状态异常: ${capturedLifecycleStream.status}`); + assert(capturedLifecycleStream.readCount > 1, "/responses 捕获回放不应退化为单块响应"); + const capturedEvents = parseSseEvents(capturedLifecycleStream.text); + const firstReplayEvent = capturedEvents.find((event) => event.payload); + assert( + firstReplayEvent?.payload?.type === "response.created" && + firstReplayEvent.payload.response?.id === "resp_stream", + "/responses 捕获回放的首个生命周期事件必须带真实 response ID", + ); + const lifecycleEvents = capturedEvents.filter((event) => [ + "response.created", + "response.in_progress", + "response.completed", + ].includes(event.payload?.type)); + assert(lifecycleEvents.length === 3, "/responses 捕获回放不应注入额外生命周期事件"); + assert( + lifecycleEvents.every((event) => event.payload?.response?.id === "resp_stream"), + "/responses 捕获回放不应发送缺少真实 response ID 的生命周期事件", + ); + assert( + lifecycleEvents.map((event) => event.payload.type).join(",") === + "response.created,response.in_progress,response.completed", + "/responses 捕获回放未保留生命周期顺序", + ); + const capturedDeltas = capturedEvents + .filter((event) => event.payload?.type === "response.output_text.delta") + .map((event) => event.payload.delta); + assert(capturedDeltas.length === 64, "/responses 捕获回放未保留全部长流 delta"); + assert( + capturedDeltas.every((delta, index) => delta === `captured-chunk-${index + 1}`), + "/responses 捕获回放的长流 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(!capturedLifecycleStream.text.includes("captured-lifecycle-marker"), "/responses lifecycle 归一化仍透传了巨大的生命周期原始 payload"); const normalizedLifecycleStream = await readSseUntilClose( `http://127.0.0.1:${gatewayPort}/responses`,