diff --git a/packages/ai/CHANGELOG.md b/packages/ai/CHANGELOG.md index 5788005ed..3a7177a5b 100644 --- a/packages/ai/CHANGELOG.md +++ b/packages/ai/CHANGELOG.md @@ -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 diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 2624b1a87..c503dfb96 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -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; } diff --git a/packages/coding-agent/src/prompts/system/handoff-document.md b/packages/coding-agent/src/prompts/system/handoff-document.md index 9d32e8534..afb032404 100644 --- a/packages/coding-agent/src/prompts/system/handoff-document.md +++ b/packages/coding-agent/src/prompts/system/handoff-document.md @@ -43,4 +43,4 @@ Use exactly this structure: Additional focus: {{additionalFocus}} -{{/if}} +{{/if}} \ No newline at end of file diff --git a/packages/coding-agent/src/web/scrapers/docs-rs.ts b/packages/coding-agent/src/web/scrapers/docs-rs.ts index 7bcec57ba..d5cbadcb4 100644 --- a/packages/coding-agent/src/web/scrapers/docs-rs.ts +++ b/packages/coding-agent/src/web/scrapers/docs-rs.ts @@ -500,7 +500,6 @@ function collectExplicitTraitImplNames(implIds: number[], index: Record, crate_: RustdocCrate): string { let md = ""; const decl = renderItemDecl(item); @@ -545,18 +544,14 @@ function renderSingleItem(item: RustdocItem, index: Record, } 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