From e897b7cee7f827e9778517834f1f89ee9466c791 Mon Sep 17 00:00:00 2001 From: Alexander Kirilin Date: Sat, 11 Jul 2026 16:26:53 -0400 Subject: [PATCH] fix(codex): preserve sequential reasoning summaries --- .../src/providers/openai-codex-responses.ts | 76 +++++++++++++------ .../test/openai-codex-responses-lite.test.ts | 53 +++++++++++++ 2 files changed, 105 insertions(+), 24 deletions(-) diff --git a/packages/ai/src/providers/openai-codex-responses.ts b/packages/ai/src/providers/openai-codex-responses.ts index 64458dfe92d..2623da37d77 100644 --- a/packages/ai/src/providers/openai-codex-responses.ts +++ b/packages/ai/src/providers/openai-codex-responses.ts @@ -691,6 +691,8 @@ class CodexStreamRuntime { nativeOutputItems: Array> = []; /** Sequential-cutoff summary sections/emitted text, global to the response (indices span reasoning items). */ cutoffSummaries: SequentialCutoffSummaryState = createSequentialCutoffSummaryState(); + /** Summary deltas buffered while waiting to see whether atomic `.done` events arrive. */ + pendingSummaryDeltas = new Map(); websocketStreamRetries = 0; providerRetryAttempt = 0; sawTerminalEvent = false; @@ -723,6 +725,7 @@ class CodexStreamRuntime { this.currentItem = null; this.currentBlock = null; this.nativeOutputItems.length = 0; + this.pendingSummaryDeltas.clear(); this.cutoffSummaries = createSequentialCutoffSummaryState(); } @@ -744,6 +747,19 @@ class CodexStreamRuntime { if (outputIndex !== undefined) return this.openItemsByOutputIndex.get(outputIndex) ?? null; return this.currentEntry; } + queueSummaryDelta(entry: CodexOpenItem | null | undefined, delta: string): void { + if (entry?.block?.type !== "thinking" || delta.length === 0) return; + const pending = this.pendingSummaryDeltas.get(entry) ?? []; + pending.push(delta); + this.pendingSummaryDeltas.set(entry, pending); + } + + takeSummaryDeltas(entry: CodexOpenItem | null | undefined): string[] { + if (!entry) return []; + const pending = this.pendingSummaryDeltas.get(entry) ?? []; + this.pendingSummaryDeltas.delete(entry); + return pending; + } closeOpenItem(entry: CodexOpenItem | null | undefined): void { if (!entry) return; @@ -1693,10 +1709,9 @@ class CodexStreamProcessor { /** * Whether the request actually sent (post-`onPayload`) opted into - * sequential-cutoff summary delivery: summaries then arrive as atomic - * `response.reasoning_summary_text.done` events and incremental - * `.delta`/`.part.*` events are ignored (mirrors codex-rs - * `uses_sequential_cutoff_reasoning_summaries`). + * sequential-cutoff summary delivery. Atomic `.done` events are preferred; + * incremental deltas remain buffered as a compatibility fallback when a + * transport emits them instead. */ get #sequentialCutoffSummaries(): boolean { return this.runtime.requestBodyForState.stream_options?.reasoning_summary_delivery === "sequential_cutoff"; @@ -1772,18 +1787,19 @@ class CodexStreamProcessor { } return firstTokenTime; } - if (eventType === "response.reasoning_summary_text.delta") { - if (this.#sequentialCutoffSummaries) return firstTokenTime; - if (this.runtime.currentItem?.type === "reasoning" && this.runtime.currentBlock?.type === "thinking") { - appendReasoningSummaryTextDelta( - this.runtime.currentItem, - this.runtime.currentBlock, - (rawEvent as { delta?: string }).delta || "", - stream, - output, - output.content.length - 1, - ); + const entry = this.runtime.openItemForEvent(rawEvent); + const delta = typeof rawEvent.delta === "string" ? rawEvent.delta : ""; + if (this.#sequentialCutoffSummaries) { + // Some Codex transports still emit incremental summary deltas even + // after opting into sequential-cutoff delivery. Buffer them until + // output_item.done so atomic `.done` events can supersede them + // without duplicate UI output. + this.runtime.queueSummaryDelta(entry, delta); + return firstTokenTime; + } + if (entry?.item.type === "reasoning" && entry.block?.type === "thinking") { + appendReasoningSummaryTextDelta(entry.item, entry.block, delta, stream, output, entry.contentIndex); } return firstTokenTime; } @@ -1793,6 +1809,7 @@ class CodexStreamProcessor { if (!this.#sequentialCutoffSummaries) return firstTokenTime; const entry = this.runtime.openItemForEvent(rawEvent); if (entry?.item.type === "reasoning" && entry.block?.type === "thinking") { + this.runtime.takeSummaryDeltas(entry); if (!firstTokenTime) firstTokenTime = performance.now(); const summaryIndex = typeof rawEvent.summary_index === "number" && Number.isFinite(rawEvent.summary_index) @@ -1827,15 +1844,13 @@ class CodexStreamProcessor { } if (eventType === "response.reasoning_summary_part.done") { - if (this.#sequentialCutoffSummaries) return firstTokenTime; - if (this.runtime.currentItem?.type === "reasoning" && this.runtime.currentBlock?.type === "thinking") { - appendReasoningSummaryPartDone( - this.runtime.currentItem, - this.runtime.currentBlock, - stream, - output, - output.content.length - 1, - ); + const entry = this.runtime.openItemForEvent(rawEvent); + if (this.#sequentialCutoffSummaries) { + this.runtime.queueSummaryDelta(entry, "\n\n"); + return firstTokenTime; + } + if (entry?.item.type === "reasoning" && entry.block?.type === "thinking") { + appendReasoningSummaryPartDone(entry.item, entry.block, stream, output, entry.contentIndex); } return firstTokenTime; } @@ -1930,6 +1945,18 @@ class CodexStreamProcessor { return firstTokenTime; } + #flushSummaryDeltas(entry: CodexOpenItem | null): void { + if (entry?.block?.type !== "thinking") return; + for (const delta of this.runtime.takeSummaryDeltas(entry)) { + entry.block.thinking += delta; + this.stream.push({ + type: "thinking_delta", + contentIndex: entry.contentIndex, + delta, + partial: this.output, + }); + } + } #handleOutputItemDone(rawEvent: Record): void { const { runtime, output, stream } = this; const rawItem = rawEvent.item; @@ -1948,6 +1975,7 @@ class CodexStreamProcessor { const contentIndex = entry?.contentIndex ?? output.content.length - 1; if (item.type === "reasoning" && block?.type === "thinking") { + this.#flushSummaryDeltas(entry); block.thinking = finalizeReasoningThinking( item, block.thinking, diff --git a/packages/ai/test/openai-codex-responses-lite.test.ts b/packages/ai/test/openai-codex-responses-lite.test.ts index 8ef2319bb9d..d3b962a4a9b 100644 --- a/packages/ai/test/openai-codex-responses-lite.test.ts +++ b/packages/ai/test/openai-codex-responses-lite.test.ts @@ -758,6 +758,7 @@ describe("openai-codex concurrent reasoning summaries", () => { const terra = createCodexModel("gpt-5.6-terra"); const withSummary = await transformRequestBody({ model: terra.id }, terra, { reasoningEffort: "medium" }); expect(withSummary.stream_options).toEqual({ reasoning_summary_delivery: "sequential_cutoff" }); + expect(withSummary.reasoning?.summary).toBe("detailed"); const suppressed = await transformRequestBody({ model: terra.id }, terra, { reasoningEffort: "medium", @@ -773,6 +774,58 @@ describe("openai-codex concurrent reasoning summaries", () => { expect(unsupported.stream_options).toBeUndefined(); }); + it("renders summary deltas when sequential-cutoff omits atomic done events", async () => { + const model = createCodexModel("gpt-5.6-terra"); + const events: Array> = [ + { + type: "response.output_item.added", + output_index: 0, + item: { type: "reasoning", id: "reason_delta", summary: [] }, + }, + { + type: "response.reasoning_summary_part.added", + item_id: "reason_delta", + output_index: 0, + summary_index: 0, + part: { type: "summary_text", text: "" }, + }, + { + type: "response.reasoning_summary_text.delta", + item_id: "reason_delta", + output_index: 0, + summary_index: 0, + delta: "Streaming ", + }, + { + type: "response.reasoning_summary_text.delta", + item_id: "reason_delta", + output_index: 0, + summary_index: 0, + delta: "fallback", + }, + { + type: "response.output_item.done", + output_index: 0, + item: { type: "reasoning", id: "reason_delta", summary: [] }, + }, + ...COMPLETED_CODEX_EVENTS, + ]; + const fetchMock = createCodexFetchMock(createCodexSse(events), () => {}); + const stream = streamOpenAICodexResponses(model, createCodexTestContext(), { + apiKey: createCodexTestToken(), + fetch: fetchMock, + reasoning: "medium", + }); + const thinkingDeltas: string[] = []; + for await (const event of stream) { + if (event.type === "thinking_delta") thinkingDeltas.push(event.delta); + } + const result = await stream.result(); + + expect(thinkingDeltas).toEqual(["Streaming ", "fallback"]); + expect(result.content.find(block => block.type === "thinking")?.thinking).toBe("Streaming fallback"); + }); + it("deduplicates cumulative atomic summaries and ignores legacy deltas under sequential cutoff", async () => { const model = createCodexModel("gpt-5.6-terra"); const events: Array> = [