diff --git a/gateway.mjs b/gateway.mjs index 71cbe0b..664b64d 100644 --- a/gateway.mjs +++ b/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() ?? ""; - - 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; + return inspectSseBlocks(blocks, config); } 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); + } } } } diff --git a/scripts/test-gateway-e2e.mjs b/scripts/test-gateway-e2e.mjs index a36780f..4f07115 100644 --- a/scripts/test-gateway-e2e.mjs +++ b/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", @@ -1065,10 +1085,16 @@ async function run() { const okStream = await readSseUntilClose( `http://127.0.0.1:${gatewayPort}${streamPath}`, { stream: true, test_reasoning_tokens: 128 }, - ); + ); 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(