shujakuin/codex-retry-gateway
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: preserve responses SSE lifecycle during replay

Browse Source
This commit is contained in:
shujakuin
2026-07-11 02:36:00 +08:00
parent 92d3845eca
commit 7d0a40e1ea
3 changed files with 89 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway.mjs
+12 -161
View File
@@ -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;