ton/pi_harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ai): preserve responses reasoning for out-of-order items

Browse Source 
closes #6009
This commit is contained in:
Mario Zechner
2026-06-25 11:57:02 +02:00
parent 9cd2c81ada
commit 8c9dbffa36
2 changed files with 156 additions and 171 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/ai/CHANGELOG.md
+1
View File
@@ -8,6 +8,7 @@
### Fixed
- Fixed OpenAI Responses streams to preserve reasoning replay state when output items finish out of order ([#6009](https://github.com/earendil-works/pi/issues/6009)).
- Fixed retry classification for provider errors that explicitly tell callers to retry the request ([#6019](https://github.com/earendil-works/pi/issues/6019)).
## [0.80.2] - 2026-06-23
packages/ai/src/api/openai-responses-shared.ts
+155 -171
View File
@@ -3,11 +3,11 @@ import type {
Tool as OpenAITool,
ResponseCreateParamsStreaming,
ResponseFunctionCallOutputItemList,
ResponseFunctionToolCall,
ResponseInput,
ResponseInputContent,
ResponseInputImage,
ResponseInputText,
ResponseOutputItem,
ResponseOutputMessage,
ResponseReasoningItem,
ResponseStreamEvent,
@@ -285,6 +285,13 @@ export function convertResponsesTools(tools: Tool[], options?: ConvertResponsesT
// Stream processing
// =============================================================================
type StreamingToolCall = ToolCall & { partialJson: string };
type ResponsesOutputSlot =
| { type: "thinking"; block: ThinkingContent; contentIndex: number }
| { type: "text"; block: TextContent; contentIndex: number }
| { type: "toolCall"; block: StreamingToolCall; contentIndex: number };
export async function processResponsesStream<TApi extends Api>(
openaiStream: AsyncIterable<ResponseStreamEvent>,
output: AssistantMessage,
@@ -292,11 +299,59 @@ export async function processResponsesStream<TApi extends Api>(
model: Model<TApi>,
options?: OpenAIResponsesStreamOptions,
): Promise<void> {
let currentItem: ResponseReasoningItem | ResponseOutputMessage | ResponseFunctionToolCall | null = null;
let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null = null;
let sawTerminalResponseEvent = false;
const blocks = output.content;
const blockIndex = () => blocks.length - 1;
const outputSlots = new Map<number, ResponsesOutputSlot>();
const getSlot = <TType extends ResponsesOutputSlot["type"]>(
outputIndex: number,
type: TType,
): Extract<ResponsesOutputSlot, { type: TType }> | undefined => {
const slot = outputSlots.get(outputIndex);
return slot?.type === type ? (slot as Extract<ResponsesOutputSlot, { type: TType }>) : undefined;
};
const createSlot = (outputIndex: number, item: ResponseOutputItem): ResponsesOutputSlot | undefined => {
if (item.type === "reasoning") {
const block: ThinkingContent = { type: "thinking", thinking: "" };
output.content.push(block);
const slot = {
type: "thinking",
block,
contentIndex: output.content.length - 1,
} satisfies ResponsesOutputSlot;
outputSlots.set(outputIndex, slot);
stream.push({ type: "thinking_start", contentIndex: slot.contentIndex, partial: output });
return slot;
}
if (item.type === "message") {
const block: TextContent = { type: "text", text: "" };
output.content.push(block);
const slot = { type: "text", block, contentIndex: output.content.length - 1 } satisfies ResponsesOutputSlot;
outputSlots.set(outputIndex, slot);
stream.push({ type: "text_start", contentIndex: slot.contentIndex, partial: output });
return slot;
}
if (item.type === "function_call") {
const block: StreamingToolCall = {
type: "toolCall",
id: `${item.call_id}|${item.id}`,
name: item.name,
arguments: {},
partialJson: item.arguments || "",
};
output.content.push(block);
const slot = {
type: "toolCall",
block,
contentIndex: output.content.length - 1,
} satisfies ResponsesOutputSlot;
outputSlots.set(outputIndex, slot);
stream.push({ type: "toolcall_start", contentIndex: slot.contentIndex, partial: output });
return slot;
}
return undefined;
};
const getOrCreateSlot = (outputIndex: number, item: ResponseOutputItem): ResponsesOutputSlot | undefined => {
return outputSlots.get(outputIndex) ?? createSlot(outputIndex, item);
};
const finalizeResponse = (
response: Extract<ResponseStreamEvent, { type: "response.completed" | "response.incomplete" }>["response"],
): void => {
@@ -335,195 +390,124 @@ export async function processResponsesStream<TApi extends Api>(
if (event.type === "response.created") {
output.responseId = event.response.id;
} else if (event.type === "response.output_item.added") {
const item = event.item;
if (item.type === "reasoning") {
currentItem = item;
currentBlock = { type: "thinking", thinking: "" };
output.content.push(currentBlock);
stream.push({ type: "thinking_start", contentIndex: blockIndex(), partial: output });
} else if (item.type === "message") {
currentItem = item;
currentBlock = { type: "text", text: "" };
output.content.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
} else if (item.type === "function_call") {
currentItem = item;
currentBlock = {
type: "toolCall",
id: `${item.call_id}|${item.id}`,
name: item.name,
arguments: {},
partialJson: item.arguments || "",
};
output.content.push(currentBlock);
stream.push({ type: "toolcall_start", contentIndex: blockIndex(), partial: output });
}
} else if (event.type === "response.reasoning_summary_part.added") {
if (currentItem && currentItem.type === "reasoning") {
currentItem.summary = currentItem.summary || [];
currentItem.summary.push(event.part);
}
createSlot(event.output_index, event.item);
} else if (event.type === "response.reasoning_summary_text.delta") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentItem.summary = currentItem.summary || [];
const lastPart = currentItem.summary[currentItem.summary.length - 1];
if (lastPart) {
currentBlock.thinking += event.delta;
lastPart.text += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
const slot = getSlot(event.output_index, "thinking");
if (!slot) continue;
slot.block.thinking += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.reasoning_summary_part.done") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentItem.summary = currentItem.summary || [];
const lastPart = currentItem.summary[currentItem.summary.length - 1];
if (lastPart) {
currentBlock.thinking += "\n\n";
lastPart.text += "\n\n";
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
delta: "\n\n",
partial: output,
});
}
}
const slot = getSlot(event.output_index, "thinking");
if (!slot) continue;
slot.block.thinking += "\n\n";
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: "\n\n",
partial: output,
});
} else if (event.type === "response.reasoning_text.delta") {
if (currentItem?.type === "reasoning" && currentBlock?.type === "thinking") {
currentBlock.thinking += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
} else if (event.type === "response.content_part.added") {
if (currentItem?.type === "message") {
currentItem.content = currentItem.content || [];
// Filter out ReasoningText, only accept output_text and refusal
if (event.part.type === "output_text" || event.part.type === "refusal") {
currentItem.content.push(event.part);
}
}
const slot = getSlot(event.output_index, "thinking");
if (!slot) continue;
slot.block.thinking += event.delta;
stream.push({
type: "thinking_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.output_text.delta") {
if (currentItem?.type === "message" && currentBlock?.type === "text") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type === "output_text") {
currentBlock.text += event.delta;
lastPart.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
const slot = getSlot(event.output_index, "text");
if (!slot) continue;
slot.block.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.refusal.delta") {
if (currentItem?.type === "message" && currentBlock?.type === "text") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type === "refusal") {
currentBlock.text += event.delta;
lastPart.refusal += event.delta;
const slot = getSlot(event.output_index, "text");
if (!slot) continue;
slot.block.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.function_call_arguments.delta") {
const slot = getSlot(event.output_index, "toolCall");
if (!slot) continue;
slot.block.partialJson += event.delta;
slot.block.arguments = parseStreamingJson(slot.block.partialJson);
stream.push({
type: "toolcall_delta",
contentIndex: slot.contentIndex,
delta: event.delta,
partial: output,
});
} else if (event.type === "response.function_call_arguments.done") {
const slot = getSlot(event.output_index, "toolCall");
if (!slot) continue;
const previousPartialJson = slot.block.partialJson;
slot.block.partialJson = event.arguments;
slot.block.arguments = parseStreamingJson(slot.block.partialJson);
if (event.arguments.startsWith(previousPartialJson)) {
const delta = event.arguments.slice(previousPartialJson.length);
if (delta.length > 0) {
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
type: "toolcall_delta",
contentIndex: slot.contentIndex,
delta,
partial: output,
});
}
}
} else if (event.type === "response.function_call_arguments.delta") {
if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") {
currentBlock.partialJson += event.delta;
currentBlock.arguments = parseStreamingJson(currentBlock.partialJson);
stream.push({
type: "toolcall_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
} else if (event.type === "response.function_call_arguments.done") {
if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") {
const previousPartialJson = currentBlock.partialJson;
currentBlock.partialJson = event.arguments;
currentBlock.arguments = parseStreamingJson(currentBlock.partialJson);
if (event.arguments.startsWith(previousPartialJson)) {
const delta = event.arguments.slice(previousPartialJson.length);
if (delta.length > 0) {
stream.push({
type: "toolcall_delta",
contentIndex: blockIndex(),
delta,
partial: output,
});
}
}
}
} else if (event.type === "response.output_item.done") {
const item = event.item;
const slot = getOrCreateSlot(event.output_index, item);
if (item.type === "reasoning" && currentBlock?.type === "thinking") {
if (item.type === "reasoning" && slot?.type === "thinking") {
const summaryText = item.summary?.map((s) => s.text).join("\n\n") || "";
const contentText = item.content?.map((c) => c.text).join("\n\n") || "";
currentBlock.thinking = summaryText || contentText || currentBlock.thinking;
currentBlock.thinkingSignature = JSON.stringify(item);
slot.block.thinking = summaryText || contentText || slot.block.thinking;
slot.block.thinkingSignature = JSON.stringify(item);
stream.push({
type: "thinking_end",
contentIndex: blockIndex(),
content: currentBlock.thinking,
contentIndex: slot.contentIndex,
content: slot.block.thinking,
partial: output,
});
currentBlock = null;
} else if (item.type === "message" && currentBlock?.type === "text") {
currentBlock.text =
item.content?.map((c) => (c.type === "output_text" ? c.text : c.refusal)).join("") || "";
currentBlock.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined);
outputSlots.delete(event.output_index);
} else if (item.type === "message" && slot?.type === "text") {
slot.block.text = item.content?.map((c) => (c.type === "output_text" ? c.text : c.refusal)).join("") || "";
slot.block.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined);
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: currentBlock.text,
contentIndex: slot.contentIndex,
content: slot.block.text,
partial: output,
});
currentBlock = null;
} else if (item.type === "function_call") {
const args =
currentBlock?.type === "toolCall" && currentBlock.partialJson
? parseStreamingJson(currentBlock.partialJson)
: parseStreamingJson(item.arguments || "{}");
let toolCall: ToolCall;
if (currentBlock?.type === "toolCall") {
// Finalize in-place and strip the scratch buffer so replay only
// carries parsed arguments.
currentBlock.arguments = args;
delete (currentBlock as { partialJson?: string }).partialJson;
toolCall = currentBlock;
} else {
toolCall = {
type: "toolCall",
id: `${item.call_id}|${item.id}`,
name: item.name,
arguments: args,
};
}
currentBlock = null;
stream.push({ type: "toolcall_end", contentIndex: blockIndex(), toolCall, partial: output });
outputSlots.delete(event.output_index);
} else if (item.type === "function_call" && slot?.type === "toolCall") {
slot.block.arguments = parseStreamingJson(item.arguments || slot.block.partialJson || "{}");
// Finalize in-place and strip the scratch buffer so replay only
// carries parsed arguments.
delete (slot.block as { partialJson?: string }).partialJson;
stream.push({
type: "toolcall_end",
contentIndex: slot.contentIndex,
toolCall: slot.block,
partial: output,
});
outputSlots.delete(event.output_index);
}
} else if (event.type === "response.completed" || event.type === "response.incomplete") {
finalizeResponse(event.response);