Skip to content

fix(codex): preserve sequential reasoning summaries #5233

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
can1357 merged 1 commit into from
Jul 14, 2026
+105 −24
Merged

fix(codex): preserve sequential reasoning summaries #5233

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 52 additions & 24 deletions packages/ai/src/providers/openai-codex-responses.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -691,6 +691,8 @@ class CodexStreamRuntime {
nativeOutputItems: Array<Record<string, unknown>> = [];
/** 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<CodexOpenItem, string[]>();
Comment on lines +694 to +695

@roboomp roboomp Jul 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should-fix: This changes observable Codex stream assembly in packages/ai, but the two-file PR omits the required packages/ai/CHANGELOG.md entry under ## [Unreleased] → ### Fixed. Please record the user-visible fix.

websocketStreamRetries = 0;
providerRetryAttempt = 0;
sawTerminalEvent = false;
Expand Down Expand Up @@ -723,6 +725,7 @@ class CodexStreamRuntime {
this.currentItem = null;
this.currentBlock = null;
this.nativeOutputItems.length = 0;
this.pendingSummaryDeltas.clear();
this.cutoffSummaries = createSequentialCutoffSummaryState();
}

Expand All @@ -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;
Expand Down Expand Up @@ -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";
Expand Down Expand Up @@ -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;
}
Expand All @@ -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)
Expand Down Expand Up @@ -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");
Comment on lines 1846 to +1849

@roboomp roboomp Jul 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

blocking: response.reasoning_summary_part.done normally follows response.reasoning_summary_text.done (the repo's Responses server emits that exact order at openai-responses-server.ts:1114-1125). The text-done branch has already discarded the legacy deltas and emitted the atomic text, but this line then buffers a separator that #flushSummaryDeltas() emits at item completion. A standard delta("IGNORED") → text.done("Plan") → part.done → output_item.done stream therefore produces thinking_deltas ["Plan", "\n\n"] and final thinking "Plan\n\n", so atomic delivery is not actually authoritative. Please only queue the separator for an entry that has not received an atomic done, and cover the real event order in the regression test.

return firstTokenTime;
}
if (entry?.item.type === "reasoning" && entry.block?.type === "thinking") {
appendReasoningSummaryPartDone(entry.item, entry.block, stream, output, entry.contentIndex);
}
return firstTokenTime;
}
Expand Down Expand Up @@ -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)) {
Comment on lines +1948 to +1950

@roboomp roboomp Jul 14, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

blocking: This flush updates only the current block; it does not advance the response-global cutoffSummaries.emitted/summary state. If item 1 falls back to buffered delta "Plan" and item 2 later supplies atomic cumulative summaries "Plan" and "Plan\n\nInspect", item 2 re-emits the first section: the resulting thinking blocks are ["Plan", "Plan\n\nInspect"]. That violates the existing cross-item replay-dedup contract. Please fold fallback text into the global cutoff state (or otherwise reconcile it before later .done events) and add a mixed fallback/atomic regression case.

entry.block.thinking += delta;
this.stream.push({
type: "thinking_delta",
contentIndex: entry.contentIndex,
delta,
partial: this.output,
});
}
}
#handleOutputItemDone(rawEvent: Record<string, unknown>): void {
const { runtime, output, stream } = this;
const rawItem = rawEvent.item;
Expand All @@ -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,
Expand Down
53 changes: 53 additions & 0 deletions packages/ai/test/openai-codex-responses-lite.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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<Record<string, unknown>> = [
{
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<Record<string, unknown>> = [
Expand Down
Loading