ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-27 21:07:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(openai): reconcile rotated terminal message IDs (#122560)

Browse Source 
* fix(ai): reconcile rotated Responses output identities

Retain output position, completion, and block ownership in one shared tracker for terminal text recovery and reasoning backfill. Preserve unindexed aliases and reject changed output types or stable tool-call IDs.

Co-authored-by: snotty <snotty@users.noreply.github.com>

* docs: keep Copilot release notes in PR context

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
Co-authored-by: snotty <snotty@users.noreply.github.com>
This commit is contained in:
Mike
2026-08-27 05:08:42 +08:00
committed by GitHub
parent b3db4758a3
commit 5acfc2dcb8
4 changed files with 241 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/src/transports/openai-responses-stream-internal.ts
+29 -33
View File
@@ -27,7 +27,7 @@ import { encodeTextSignatureV1 } from "./openai-responses-replay-internal.js";
import { adaptResponsesStream } from "./openai-responses-stream-observer-internal.js";
import {
appendResponsesPendingTextDelta,
createResponsesOutputContentIndex,
createResponsesOutputTracker,
createResponsesOutputSlotTracker,
readResponsesOutputIndex,
type ResponsesStreamOutputSlot,
@@ -69,9 +69,7 @@ export async function processResponsesStream<TApi extends Api>(
>;
const streamingToolCalls = createResponsesToolCallTracker<StreamingToolCallState>();
const outputSlots = createResponsesOutputSlotTracker<ResponsesOutputSlot>();
const reasoningBlocksById = new Map<string, ResponsesThinkingBlock>();
const outputItemContentIndexes = createResponsesOutputContentIndex();
const startedTextBlocksByItemId = new Map<string, TextBlockReference>();
const outputs = createResponsesOutputTracker();
let terminalResponse: CompletedResponse | null | undefined;
let lastTextBlock: TextBlockReference | null = null;
const blocks = output.content;
@@ -87,10 +85,10 @@ export async function processResponsesStream<TApi extends Api>(
item,
block,
contentIndex: blocks.length,
outputIndex: readResponsesOutputIndex(event),
} satisfies ResponsesOutputSlot;
blocks.push(block);
reasoningBlocksById.set(item.id, block);
outputItemContentIndexes.set(item, slot.contentIndex);
outputs.set(item, slot.contentIndex, slot.outputIndex);
outputSlots.register(event, slot);
stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output });
return slot;
@@ -112,17 +110,13 @@ export async function processResponsesStream<TApi extends Api>(
item: messageItem,
block,
contentIndex: block ? blocks.length : undefined,
outputIndex: readResponsesOutputIndex(event),
pendingText: collapseCandidate ? "" : null,
collapseCandidate,
} satisfies ResponsesOutputSlot;
if (block) {
blocks.push(block);
outputItemContentIndexes.set(messageItem, slot.contentIndex ?? blocks.length - 1);
startedTextBlocksByItemId.set(messageItem.id, {
block,
index: slot.contentIndex ?? blocks.length - 1,
phase: messageItem.phase ?? undefined,
});
outputs.set(messageItem, slot.contentIndex ?? blocks.length - 1, slot.outputIndex);
}
outputSlots.register(event, slot);
if (slot.contentIndex !== undefined) {
@@ -144,12 +138,6 @@ export async function processResponsesStream<TApi extends Api>(
}
return readResponsesOutputIndex(event) === undefined ? undefined : outputSlots.get(event);
};
const getOrCreateOutputSlot = (
event: object,
item: ResponseOutputItem | ResponsesStreamOutputMessage,
): ResponsesOutputSlot | undefined => {
return resolveOutputItemSlot(event, item) ?? createOutputSlot(event, item);
};
const materializeDeferredTextSlot = (
slot: Extract<ResponsesOutputSlot, { type: "text" }>,
): void => {
@@ -166,12 +154,7 @@ export async function processResponsesStream<TApi extends Api>(
};
blocks.push(slot.block);
slot.contentIndex = blocks.length - 1;
outputItemContentIndexes.set(slot.item, slot.contentIndex);
startedTextBlocksByItemId.set(slot.item.id, {
block: slot.block,
index: slot.contentIndex,
phase: slot.item.phase ?? undefined,
});
outputs.set(slot.item, slot.contentIndex, slot.outputIndex);
stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output });
if (text) {
stream.push({
@@ -199,9 +182,7 @@ export async function processResponsesStream<TApi extends Api>(
stream,
model,
options,
reasoningBlocksById,
startedTextBlocksByItemId,
outputItemContentIndexes,
outputs,
getLastTextBlock: () => lastTextBlock,
setLastTextBlock: (block) => {
lastTextBlock = block;
@@ -262,7 +243,7 @@ export async function processResponsesStream<TApi extends Api>(
outputSlots.register(event, { type: "toolCall", toolCall: toolCallState });
}
output.content.push(toolCallBlock);
outputItemContentIndexes.set(item, contentIndex);
outputs.set(item, contentIndex, readResponsesOutputIndex(event));
stream.push({ type: "toolcall_start", contentIndex, partial: output });
}
} else if (event.type === "response.reasoning_summary_part.added") {
@@ -455,13 +436,19 @@ export async function processResponsesStream<TApi extends Api>(
const existingOutputSlot = resolveOutputItemSlot(event, item);
materializeDeferredTextSlots(existingOutputSlot);
const outputSlot = existingOutputSlot ?? getOrCreateOutputSlot(event, item);
const outputSlot = existingOutputSlot ?? createOutputSlot(event, item);
compactionTracker.completed(item, blocks.length);
if (item.type === "reasoning" && outputSlot?.type === "thinking") {
const summaryText = item.summary?.map((s) => s.text).join("\n\n") || "";
const contentText = item.content?.map((c) => c.text).join("\n\n") || "";
outputSlot.block.thinking = summaryText || contentText || outputSlot.block.thinking;
outputSlot.block.thinkingSignature = JSON.stringify(item);
outputs.set(
item,
outputSlot.contentIndex,
readResponsesOutputIndex(event) ?? outputSlot.outputIndex,
true,
);
if (item.encrypted_content && options?.reasoningReplayMetadata) {
outputSlot.block[OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY] =
options.reasoningReplayMetadata;
@@ -516,7 +503,12 @@ export async function processResponsesStream<TApi extends Api>(
partial: output,
});
lastTextBlock = outputSlot.collapseCandidate;
outputItemContentIndexes.set(item, outputSlot.collapseCandidate.index);
outputs.set(
item,
outputSlot.collapseCandidate.index,
readResponsesOutputIndex(event) ?? outputSlot.outputIndex,
true,
);
} else {
if (!outputSlot.block) {
// Deferred distinct message: open its block now, balanced with the
@@ -541,7 +533,12 @@ export async function processResponsesStream<TApi extends Api>(
throw new Error("Responses stream finalized text without a content index");
}
lastTextBlock = { block: outputSlot.block, index: contentIndex, phase };
outputItemContentIndexes.set(item, contentIndex);
outputs.set(
item,
contentIndex,
readResponsesOutputIndex(event) ?? outputSlot.outputIndex,
true,
);
stream.push({
type: "text_end",
contentIndex,
@@ -550,7 +547,6 @@ export async function processResponsesStream<TApi extends Api>(
});
}
outputSlots.forget(outputSlot);
startedTextBlocksByItemId.delete(item.id);
} else if (item.type === "function_call") {
const streamingToolCall = streamingToolCalls.resolve(
event,
@@ -624,7 +620,7 @@ export async function processResponsesStream<TApi extends Api>(
toolCall,
partial: output,
});
outputItemContentIndexes.set(item, contentIndex);
outputs.set(item, contentIndex, readResponsesOutputIndex(event), true);
}
} else if (event.type === "response.completed" || event.type === "response.incomplete") {
if (streamingToolCalls.hasActive()) {
packages/ai/src/transports/openai-responses-stream-slots-internal.ts
+58 -8
View File
@@ -11,12 +11,14 @@ export type ResponsesStreamOutputSlot<TMessage, TToolCall> =
item: ResponseReasoningItem;
block: ResponsesThinkingBlock;
contentIndex: number;
outputIndex: number | undefined;
}
| {
type: "text";
item: TMessage;
block: TextContent | null;
contentIndex: number | undefined;
outputIndex: number | undefined;
pendingText: string | null;
collapseCandidate: TextBlockReference | null;
}
@@ -33,8 +35,18 @@ type ResponsesOutputIdentityItem = {
call_id?: string | null;
};
export function createResponsesOutputContentIndex() {
const indexes = new Map<string, number>();
type ResponsesOutputState = {
type: string;
callId?: string | null;
outputIndex?: number;
contentIndex: number;
completed: boolean;
};
export type ResponsesOutputTracker = ReturnType<typeof createResponsesOutputTracker>;
export function createResponsesOutputTracker() {
const outputs = new Map<string | number, ResponsesOutputState>();
const identity = (item: ResponsesOutputIdentityItem): string | undefined => {
if ((item.type === "reasoning" || item.type === "message") && item.id) {
return `${item.type}:${item.id}`;
@@ -43,15 +55,53 @@ export function createResponsesOutputContentIndex() {
? `function_call:${item.call_id ?? item.id ?? ""}`
: undefined;
};
const get = (item: ResponsesOutputIdentityItem, outputIndex?: number) => {
const key = identity(item);
const output =
(outputIndex === undefined ? undefined : outputs.get(outputIndex)) ??
(key === undefined ? undefined : outputs.get(key));
if (
!output ||
(outputIndex !== undefined &&
output.outputIndex !== undefined &&
output.outputIndex !== outputIndex)
) {
return undefined;
}
if (
output.type !== item.type ||
(output.callId && item.call_id && output.callId !== item.call_id)
) {
throw new Error("Responses stream changed output item identity");
}
return output;
};
return {
get(item: ResponsesOutputIdentityItem): number | undefined {
const key = identity(item);
return key === undefined ? undefined : indexes.get(key);
},
set(item: ResponsesOutputIdentityItem, contentIndex: number): void {
get,
set(
item: ResponsesOutputIdentityItem,
contentIndex: number,
outputIndex?: number,
completed = false,
): void {
const output: ResponsesOutputState = get(item, outputIndex) ?? {
type: item.type,
contentIndex,
completed,
};
Object.assign(output, { contentIndex, completed });
if (item.call_id) {
output.callId = item.call_id;
}
if (outputIndex !== undefined) {
output.outputIndex = outputIndex;
outputs.set(outputIndex, output);
}
// Output positions survive encrypted ID rotation. Aliases retain the
// supported unindexed stream contract without owning lifecycle state.
const key = identity(item);
if (key !== undefined) {
indexes.set(key, contentIndex);
outputs.set(key, output);
}
},
};
packages/ai/src/transports/openai-responses-stream-terminal-internal.ts
+38 -38
View File
@@ -30,6 +30,7 @@ import {
type OpenAIResponsesReasoningReplayMetadata,
} from "./openai-responses-contracts.js";
import { encodeTextSignatureV1 } from "./openai-responses-replay-internal.js";
import type { ResponsesOutputTracker } from "./openai-responses-stream-slots-internal.js";
import { parseTerminalToolCallArguments } from "./transport-stream-shared.js";
export type ResponsesEventSink = { push(event: AssistantMessageEvent): void };
@@ -109,12 +110,7 @@ export function createResponsesTerminalController(params: {
stream: ResponsesEventSink;
model: Model;
options?: TerminalOptions;
reasoningBlocksById: Map<string, ResponsesThinkingBlock>;
startedTextBlocksByItemId: Map<string, TextBlockReference>;
outputItemContentIndexes: {
get: (item: ResponseOutputItem) => number | undefined;
set: (item: ResponseOutputItem, contentIndex: number) => void;
};
outputs: ResponsesOutputTracker;
getLastTextBlock: () => TextBlockReference | null;
setLastTextBlock: (block: TextBlockReference | null) => void;
markFinalized: () => void;
@@ -122,12 +118,13 @@ export function createResponsesTerminalController(params: {
const { output, stream, model, options } = params;
const blocks = output.content;
const backfillReasoning = (items: ResponseOutputItem[]) => {
for (const item of items) {
for (const [outputIndex, item] of items.entries()) {
if (item.type !== "reasoning" || !item.encrypted_content) {
continue;
}
const block = params.reasoningBlocksById.get(item.id);
if (!block?.thinkingSignature) {
const tracked = params.outputs.get(item, outputIndex);
const block = tracked && blocks[tracked.contentIndex];
if (block?.type !== "thinking" || !block.thinkingSignature) {
continue;
}
const stored = JSON.parse(block.thinkingSignature) as ResponseReasoningItem;
@@ -138,11 +135,13 @@ export function createResponsesTerminalController(params: {
});
}
if (options?.reasoningReplayMetadata) {
block[OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY] = options.reasoningReplayMetadata;
Object.assign(block, {
[OPENAI_RESPONSES_REASONING_REPLAY_BLOCK_META_KEY]: options.reasoningReplayMetadata,
});
}
}
};
const appendText = (item: ResponseOutputMessage): number | undefined => {
const appendText = (item: ResponseOutputMessage, contentIndex?: number): number | undefined => {
const text = (Array.isArray(item.content) ? item.content : [])
.map((part) => {
const content = part as { type: string; text?: string; refusal?: string };
@@ -151,30 +150,30 @@ export function createResponsesTerminalController(params: {
: (content.refusal ?? "");
})
.join("");
const started = params.startedTextBlocksByItemId.get(item.id);
const block = contentIndex === undefined ? undefined : blocks[contentIndex];
const started = block?.type === "text" ? block : undefined;
if (!text && !started) {
return undefined;
}
const phase = item.phase ?? undefined;
if (started) {
const previousText = started.block.text;
started.block.text = text;
started.block.textSignature = encodeTextSignatureV1(item.id, phase);
params.setLastTextBlock({ block: started.block, index: started.index, phase });
params.startedTextBlocksByItemId.delete(item.id);
if (started && contentIndex !== undefined) {
const previousText = started.text;
started.text = text;
started.textSignature = encodeTextSignatureV1(item.id, phase);
params.setLastTextBlock({ block: started, index: contentIndex, phase });
if (text.startsWith(previousText)) {
const delta = text.slice(previousText.length);
if (delta) {
stream.push({ type: "text_delta", contentIndex: started.index, delta });
stream.push({ type: "text_delta", contentIndex, delta });
}
}
stream.push({
type: "text_end",
contentIndex: started.index,
contentIndex,
content: text,
partial: output,
});
return started.index;
return contentIndex;
}
const previous = params.getLastTextBlock();
const collapse = resolveResponsesMessageSnapshotCollapse({
@@ -193,14 +192,14 @@ export function createResponsesTerminalController(params: {
});
return previous.index;
}
const block: TextContent = {
const newBlock: TextContent = {
type: "text",
text,
textSignature: encodeTextSignatureV1(item.id, phase),
};
blocks.push(block);
blocks.push(newBlock);
const index = blocks.length - 1;
params.setLastTextBlock({ block, index, phase });
params.setLastTextBlock({ block: newBlock, index, phase });
stream.push({ type: "text_start", contentIndex: index, partial: output });
stream.push({ type: "text_end", contentIndex: index, content: text, partial: output });
return index;
@@ -221,19 +220,17 @@ export function createResponsesTerminalController(params: {
};
const recoverTerminalOutput = (items: ResponseOutputItem[], includeToolCalls: boolean) => {
let hasCompletedLaterOutput = false;
for (const item of items.toReversed()) {
for (const [outputIndex, item] of [...items.entries()].toReversed()) {
const tracked = params.outputs.get(item, outputIndex);
if (item.type === "reasoning") {
// Terminal snapshots only backfill streamed reasoning; missing reasoning is never emitted.
hasCompletedLaterOutput ||= params.outputItemContentIndexes.get(item) !== undefined;
hasCompletedLaterOutput ||= tracked !== undefined;
continue;
}
if (item.type !== "message" && item.type !== "function_call") {
continue;
}
if (
params.outputItemContentIndexes.get(item) !== undefined ||
(item.type === "message" && params.startedTextBlocksByItemId.has(item.id))
) {
if (tracked) {
hasCompletedLaterOutput = true;
continue;
}
@@ -250,13 +247,13 @@ export function createResponsesTerminalController(params: {
}
for (const [terminalIndex, item] of items.entries()) {
if (item.type === "message") {
const contentIndex = params.outputItemContentIndexes.get(item);
if (contentIndex !== undefined && !params.startedTextBlocksByItemId.has(item.id)) {
const tracked = params.outputs.get(item, terminalIndex);
if (tracked?.completed) {
continue;
}
const appendedIndex = appendText(item);
const appendedIndex = appendText(item, tracked?.contentIndex);
if (appendedIndex !== undefined) {
params.outputItemContentIndexes.set(item, appendedIndex);
params.outputs.set(item, appendedIndex, terminalIndex, true);
}
} else {
params.setLastTextBlock(null);
@@ -267,8 +264,11 @@ export function createResponsesTerminalController(params: {
output.providerReplay.data === item.encrypted_content;
if (item.type === "compaction" && !alreadyCapturedCompaction) {
let replayIndex = blocks.length;
for (const laterItem of items.slice(terminalIndex + 1)) {
const laterContentIndex = params.outputItemContentIndexes.get(laterItem);
for (const [laterIndex, laterItem] of items.entries()) {
if (laterIndex <= terminalIndex) {
continue;
}
const laterContentIndex = params.outputs.get(laterItem, laterIndex)?.contentIndex;
if (laterContentIndex !== undefined) {
replayIndex = laterContentIndex;
break;
@@ -282,10 +282,10 @@ export function createResponsesTerminalController(params: {
options?.reasoningReplayMetadata,
);
} else if (includeToolCalls && item.type === "function_call") {
if (params.outputItemContentIndexes.get(item) !== undefined) {
if (params.outputs.get(item, terminalIndex)) {
continue;
}
params.outputItemContentIndexes.set(item, appendToolCall(item));
params.outputs.set(item, appendToolCall(item), terminalIndex, true);
}
}
}
packages/ai/src/transports/openai-responses-stream-terminal-recovery.test.ts
+116 -2
View File
@@ -692,10 +692,124 @@ const fixtures: ParityFixture[] = [
error: null,
},
},
{
name: "terminal reasoning backfill follows the completed output slot",
events: [
{
type: "response.output_item.added",
output_index: 0,
item: { id: "rs_backfill", type: "reasoning", summary: [] },
},
{
type: "response.output_item.done",
output_index: 0,
item: {
id: "rs_backfill",
type: "reasoning",
summary: [{ type: "summary_text", text: "Checking the answer." }],
},
},
completed("resp_reasoning_backfill", [
{
id: "rs_backfill",
type: "reasoning",
summary: [{ type: "summary_text", text: "Checking the answer." }],
encrypted_content: "fixture-reasoning-ciphertext",
},
]),
],
canonical: {
events: [
{ type: "thinking_start", contentIndex: 0 },
{ type: "thinking_end", contentIndex: 0, content: "Checking the answer." },
],
content: [{ type: "thinking", thinking: "Checking the answer.", encrypted: true }],
responseId: "resp_reasoning_backfill",
stopReason: "stop",
error: null,
},
},
{
name: "equal text in distinct output slots remains distinct",
events: [
...[0, 1].flatMap((outputIndex) => [
{
type: "response.output_item.added",
output_index: outputIndex,
item: { id: `msg_equal_${outputIndex}`, type: "message", content: [] },
},
{
type: "response.output_item.done",
output_index: outputIndex,
item: {
id: `msg_equal_${outputIndex}`,
type: "message",
content: [{ type: "output_text", text: "Again." }],
},
},
]),
completed(
"resp_equal_messages",
[0, 1].map((outputIndex) => ({
id: `msg_equal_${outputIndex}`,
type: "message",
content: [{ type: "output_text", text: "Again." }],
})),
),
],
canonical: {
events: [
{ type: "text_start", contentIndex: 0 },
{ type: "text_end", contentIndex: 0, content: "Again." },
{ type: "text_start", contentIndex: 1 },
{ type: "text_end", contentIndex: 1, content: "Again." },
],
content: [
{ type: "text", text: "Again." },
{ type: "text", text: "Again." },
],
responseId: "resp_equal_messages",
stopReason: "stop",
error: null,
},
},
];
describe("Responses terminal recovery fixtures", () => {
describe.each(["stable", "rotated"])("Responses terminal recovery with %s item IDs", (ids) => {
it.each(fixtures)("$name", async (fixture) => {
expect(await runFixture(fixture.events)).toEqual(fixture.canonical);
let sequence = 0;
const events: ParityFixture["events"] = JSON.parse(
JSON.stringify(fixture.events),
(key, value: unknown) =>
ids === "rotated" &&
(key === "id" || key === "item_id") &&
typeof value === "string" &&
/^(msg|rs)_/.test(value)
? `${value}_event_${sequence++}`
: value,
);
expect(await runFixture(events)).toEqual(fixture.canonical);
});
});
it.each(["type", "call_id"])("rejects terminal output that changes its %s", async (field) => {
const streamed = {
type: "function_call",
id: "fc_original",
call_id: "call_original",
name: "lookup",
arguments: "{}",
status: "completed",
};
const terminal =
field === "type"
? { type: "message", id: "msg_changed", content: [{ type: "output_text", text: "Changed" }] }
: { ...streamed, call_id: "call_changed" };
const result = await runFixture([
{ type: "response.output_item.done", output_index: 0, item: streamed },
completed("resp_changed_identity", [terminal]),
]);
expect(result.error).toBe("Responses stream changed output item identity");
expect(result.content).toHaveLength(1);
expect(result.events.filter((event) => event.type === "toolcall_end")).toHaveLength(1);
});