fix(ai): corrected WebSocket fallback to safely replay buffered output over SSE

- Fixed WebSocket stream fallback logic to safely replay buffered output over SSE when WebSocket fails after partial content has been streamed.
- Added tracking flag to prevent unsafe replays of tool calls and terminal events during fallback transitions.
- Enhanced error recovery to reset output state when replaying buffered content over SSE connection.
This commit is contained in:
can1357
2026-03-08 07:50:25 +01:00
parent 85621f37bf
commit 628e68e6ae
4 changed files with 48 additions and 24 deletions
+3
View File
@@ -1,6 +1,9 @@
# Changelog
## [Unreleased]
### Fixed
- Fixed WebSocket stream fallback logic to safely replay buffered output over SSE when WebSocket fails after partial content has been streamed
## [13.9.4] - 2026-03-07
### Changed
@@ -504,6 +504,7 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
let websocketStreamRetries = 0;
let providerRetryAttempt = 0;
let sawTerminalEvent = false;
let canSafelyReplayWebsocketOverSse = true;
while (true) {
try {
for await (const rawEvent of eventStream) {
@@ -674,6 +675,7 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
name: item.name,
arguments: parseStreamingJson(item.arguments || "{}"),
};
canSafelyReplayWebsocketOverSse = false;
stream.push({ type: "toolcall_end", contentIndex: blockIndex(), toolCall, partial: output });
}
} else if (eventType === "response.created") {
@@ -729,48 +731,72 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
break;
} catch (error) {
if (
const canReplayWebsocketFailureOverSse =
usingWebsocket &&
websocketState &&
isCodexWebSocketRetryableStreamError(error) &&
output.content.length === 0 &&
!options?.signal?.aborted
) {
canSafelyReplayWebsocketOverSse &&
!sawTerminalEvent &&
!options?.signal?.aborted;
if (canReplayWebsocketFailureOverSse) {
const state = websocketState!;
const streamError = error instanceof Error ? error : new Error(String(error));
const replayingBufferedOutputOverSse = output.content.length > 0;
const isFatal = isCodexWebSocketFatalError(streamError);
const activateFallback = isFatal || websocketStreamRetries >= getCodexWebSocketRetryBudget();
recordCodexWebSocketFailure(websocketState, activateFallback);
const activateFallback =
replayingBufferedOutputOverSse ||
isFatal ||
websocketStreamRetries >= getCodexWebSocketRetryBudget();
recordCodexWebSocketFailure(state, activateFallback);
logCodexDebug("codex websocket stream fallback", {
error: streamError.message,
retry: websocketStreamRetries,
retryBudget: getCodexWebSocketRetryBudget(),
activated: activateFallback,
fatal: isFatal,
replayedBufferedOutput: replayingBufferedOutputOverSse,
});
if (!activateFallback) {
websocketStreamRetries += 1;
await abortableSleep(getCodexWebSocketRetryDelayMs(websocketStreamRetries), options?.signal);
const websocketRequest = buildCodexWebSocketRequest(transformedBody, websocketState);
const websocketRequest = buildCodexWebSocketRequest(transformedBody, state);
const websocketHeaders = createCodexHeaders(
requestHeaders,
accountId,
apiKey,
options?.sessionId,
"websocket",
websocketState,
state,
);
requestBodyForState = cloneRequestBody(transformedBody);
eventStream = await openCodexWebSocketEventStream(
toWebSocketUrl(url),
websocketHeaders,
websocketRequest,
websocketState,
state,
options?.signal,
);
usingWebsocket = true;
websocketState.lastTransport = "websocket";
state.lastTransport = "websocket";
continue;
}
if (replayingBufferedOutputOverSse) {
canSafelyReplayWebsocketOverSse = true;
currentItem = null;
currentBlock = null;
output.content.length = 0;
nativeOutputItems.length = 0;
output.usage = {
input: 0,
output: 0,
cacheRead: 0,
cacheWrite: 0,
totalTokens: 0,
cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
};
output.stopReason = "stop";
firstTokenTime = undefined;
}
eventStream = await openCodexSseEventStream(
url,
requestHeaders,
@@ -778,11 +804,11 @@ export const streamOpenAICodexResponses: StreamFunction<"openai-codex-responses"
apiKey,
options?.sessionId,
transformedBody,
websocketState,
state,
options?.signal,
);
usingWebsocket = false;
websocketState.lastTransport = "sse";
state.lastTransport = "sse";
requestBodyForState = cloneRequestBody(transformedBody);
continue;
}
@@ -43,4 +43,4 @@ Use exactly this structure:
<instruction>
Additional focus: {{additionalFocus}}
</instruction>
{{/if}}
{{/if}}
@@ -500,7 +500,6 @@ function collectExplicitTraitImplNames(implIds: number[], index: Record<string,
return names;
}
function renderSingleItem(item: RustdocItem, index: Record<string, RustdocItem>, crate_: RustdocCrate): string {
let md = "";
const decl = renderItemDecl(item);
@@ -545,18 +544,14 @@ function renderSingleItem(item: RustdocItem, index: Record<string, RustdocItem>,
}
const methods = collectInherentMethodLines(impls, index);
if (methods.length) md += `## Methods
${methods.join("\n")}
`;
if (methods.length) {
md += `## Methods\n\n${methods.join("\n")}\n\n`;
}
const traitImpls = collectExplicitTraitImplNames(impls, index);
if (traitImpls.length) md += `## Trait Implementations
${traitImpls.map(t => `- ${t}`).join("\n")}
`;
if (traitImpls.length) {
md += `## Trait Implementations\n\n${traitImpls.map(t => `- ${t}`).join("\n")}\n\n`;
}
}
// For enums, show variants