ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 20:35:39 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(llm): collapse cumulative openai-responses message snapshots instead of concatenating [AI-assisted] (#92399)

Browse Source 
* fix(llm): collapse cumulative openai-responses message snapshots instead of concatenating

Some openai-responses providers (observed: Bedrock Mantle with GPT-5.x
reasoning enabled, confirmed server-side via raw curl) re-emit the
assistant message as many cumulative snapshot items — each a
prefix-superset of the previous one — instead of a single final message
item. Both stream consumers appended one text block per item, so the
final visible reply, transcript, and replay context repeated the answer
once per snapshot (observed 49-80x).

Treat a same-phase message item whose text extends the immediately
preceding text block as a replacement: the prior block takes the longer
text, the duplicate block is dropped, and the first item's signature is
kept so replay and stream-item identity stay stable. Shrinking or
identical adjacent snapshots are dropped. Any non-message output item
(reasoning, tool call) is a real boundary that resets the collapse, so
distinct post-tool messages and reasoning replay pairing are untouched,
as are different-phase (commentary/final_answer) items. Applies to the
agent transport stream, the shared LLM consumer, and completed-response
backfill.

Fixes #91959. Reported by @phoenixyy with server-side evidence from
@DaiMingNJ.

* test(llm): drop redundant stream drains from responses snapshot tests

* fix(llm): collapse only strict snapshot extensions and keep newest item signature

Address ClawSweeper P1 review findings on #92399: text-prefix relation
alone was broader than the observed corruption. Equal or shrinking
adjacent same-phase message items are now always kept as distinct blocks
(the Responses protocol allows multiple message items per response —
verified against the sibling Codex parser, codex-rs/codex-api/src/sse/
responses.rs, which emits every output_item.done message as an
independent item). With extension-only collapse a false positive can
only merge rendering of two messages; it can never remove text.

The merged block now carries the newest item's signature instead of the
first one's, so replay associates the final content with the item that
actually produced it.

* fix(llm): defer snapshot-candidate message blocks to keep the event lifecycle balanced

Address the remaining ClawSweeper P1 on #92399: collapsing a snapshot
used to pop a block whose text_start had already been emitted, leaving
per-index stream subscribers tracking a phantom block.

A message item that follows a finalized text block now defers its public
block: no text_start is emitted and deltas are withheld until the item
either diverges from the prior text (then the block opens and the
withheld prefix replays as one delta) or completes. A collapsed snapshot
therefore never starts a block — it only re-ends the prior index with
grown content, the documented resend shape — and a distinct deferred
item opens and closes its own block normally. No block is ever removed,
so every text_start has exactly one matching text_end at a live index.

Tests now assert the complete ordered event sequence for the collapse,
distinct-item, and divergence cases in both consumers.

* fix(llm): treat any non-message item as a collapse boundary in completed-response backfill

The streaming consumer resets the snapshot-collapse anchor on every
non-message output item ("any other item is a real boundary"), but the
transport's completed-response backfill only dispatched message and
function_call items, so a reasoning item between two strict-prefix
message items did not reset the anchor and the later message could
collapse across it — an asymmetry with the streaming path's documented
invariant. Reset lastTextBlock for every non-message item in the backfill
loop (one canonical place; the per-tool-call reset is now redundant and
removed). Covered by a backfill reasoning-boundary regression test.
This commit is contained in:
Amer Sheeny
2026-06-22 18:49:19 +01:00
committed by GitHub
parent 734f2aa009
commit 9fbc8a74ef
5 changed files with 822 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agents/openai-transport-stream.test.ts
+210
View File
@@ -435,6 +435,216 @@ describe("openai transport stream", () => {
]);
});
it("collapses cumulative message snapshot items into one text block (#91959)", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
const pushSpy = vi.fn();
const snapshot1 = "Scaled dot-product attention";
const snapshot2 = "Scaled dot-product attention divides by sqrt(d_k)";
const snapshot3 = "Scaled dot-product attention divides by sqrt(d_k) before softmax.";
const messageItem = (id: string, text: string) => ({
type: "message",
id,
phase: "final_answer",
content: [{ type: "output_text", text }],
});
await testing.processResponsesStream(
streamChunks([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "final_answer" },
},
{ type: "response.output_text.delta", delta: snapshot1 },
{ type: "response.output_item.done", item: messageItem("msg_1", snapshot1) },
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{ type: "response.output_item.done", item: messageItem("msg_2", snapshot2) },
{
type: "response.output_item.added",
item: { type: "message", id: "msg_3", phase: "final_answer" },
},
{ type: "response.output_item.done", item: messageItem("msg_3", snapshot3) },
{
type: "response.completed",
response: { id: "resp-snapshots", status: "completed" },
},
]),
output,
{ push: pushSpy },
model,
);
expect(output.content).toEqual([
{
type: "text",
text: snapshot3,
textSignature: '{"v":1,"id":"msg_3","phase":"final_answer"}',
},
]);
// Balanced lifecycle: one text_start, all events on index 0, and each
// collapsed snapshot re-ends the same block.
const textEvents = pushSpy.mock.calls
.map(([event]) => event as { type: string; contentIndex?: number })
.filter((event) => event.type.startsWith("text_"));
expect(textEvents.map((event) => [event.type, event.contentIndex])).toEqual([
["text_start", 0],
["text_delta", 0],
["text_end", 0],
["text_end", 0],
["text_end", 0],
]);
});
it("keeps prefix-nested message items separated by a tool call as separate blocks", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
const messageEvents = (id: string, text: string) => [
{ type: "response.output_item.added", item: { type: "message", id } },
{
type: "response.output_item.done",
item: { type: "message", id, content: [{ type: "output_text", text }] },
},
];
await testing.processResponsesStream(
streamChunks([
...messageEvents("msg_1", "Done."),
{
type: "response.output_item.added",
item: {
type: "function_call",
id: "fc_1",
call_id: "call_1",
name: "write",
arguments: "{}",
},
},
{
type: "response.output_item.done",
item: {
type: "function_call",
id: "fc_1",
call_id: "call_1",
name: "write",
arguments: "{}",
},
},
...messageEvents("msg_2", "Done."),
{
type: "response.completed",
response: { id: "resp-tool-boundary", status: "completed" },
},
]),
output,
{ push: vi.fn() },
model,
);
// The post-tool message is a real reply, not a snapshot of the pre-tool one.
expect(output.content.map((block) => block.type)).toEqual(["text", "toolCall", "text"]);
expect(output.content[2]).toMatchObject({ type: "text", text: "Done." });
});
it("collapses cumulative message snapshots in completed-response backfill (#91959)", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
await testing.processResponsesStream(
streamChunks([
{
type: "response.completed",
response: {
id: "resp-backfill-snapshots",
status: "completed",
output: [
{
type: "message",
id: "msg_1",
role: "assistant",
content: [{ type: "output_text", text: "The answer" }],
},
{
type: "message",
id: "msg_2",
role: "assistant",
content: [{ type: "output_text", text: "The answer is 42." }],
},
{
type: "message",
id: "msg_3",
role: "assistant",
content: [{ type: "output_text", text: "The answer" }],
},
],
},
},
]),
output,
{ push: vi.fn() },
model,
);
// msg_2 strictly extends msg_1 and collapses into it; msg_3 shrinks back
// and is an independently identified message, so it stays a real block.
expect(output.content).toEqual([
{
type: "text",
text: "The answer is 42.",
textSignature: '{"v":1,"id":"msg_2"}',
},
{
type: "text",
text: "The answer",
textSignature: '{"v":1,"id":"msg_3"}',
},
]);
});
it("keeps backfill message items separated by a reasoning item as distinct blocks", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
await testing.processResponsesStream(
streamChunks([
{
type: "response.completed",
response: {
id: "resp-backfill-reasoning-boundary",
status: "completed",
output: [
{
type: "message",
id: "msg_1",
role: "assistant",
content: [{ type: "output_text", text: "Step one." }],
},
{ type: "reasoning", id: "rs_1", summary: [] },
{
type: "message",
id: "msg_2",
role: "assistant",
content: [{ type: "output_text", text: "Step one. Step two." }],
},
],
},
},
]),
output,
{ push: vi.fn() },
model,
);
// A reasoning item is a real boundary even in backfill: msg_2 must not
// collapse into msg_1 despite being a strict extension (mirrors streaming).
expect(output.content).toEqual([
{ type: "text", text: "Step one.", textSignature: '{"v":1,"id":"msg_1"}' },
{ type: "text", text: "Step one. Step two.", textSignature: '{"v":1,"id":"msg_2"}' },
]);
});
it("backfills Azure Responses completed function calls when item events are absent", async () => {
const model = createAzureResponsesModel();
const output = createResponsesAssistantOutput(model);
src/agents/openai-transport-stream.ts
+135 -27
View File
@@ -38,6 +38,7 @@ import { isOpenAICompatibleAzureResponsesBaseUrl } from "../shared/azure-openai-
import {
isResponsesTextContentPartType,
isResponsesTextDeltaEventType,
resolveResponsesMessageSnapshotCollapse,
} from "../shared/openai-responses-stream-compat.js";
import { createReasoningTagTextPartitioner } from "../shared/text/reasoning-tag-text-partitioner.js";
import { CHARS_PER_TOKEN_ESTIMATE, estimateStringChars } from "../utils/cjk-chars.js";
@@ -1474,25 +1475,68 @@ async function processResponsesStream(
) {
let currentItem: Record<string, unknown> | null = null;
let currentBlock: Record<string, unknown> | null = null;
let lastTextBlock: {
block: Record<string, unknown>;
index: number;
phase: "commentary" | "final_answer" | undefined;
} | null = null;
// While a message item may still be a cumulative snapshot of lastTextBlock,
// its public block is deferred so a collapsed item never leaves an
// unbalanced text_start behind (#91959). null = no deferral in progress.
let pendingMessageText: string | null = null;
const streamStartedAt = Date.now();
let eventCount = 0;
const eventTypes = new Map<string, number>();
const sseDebugMode = resolveModelSseDebugMode();
const blockIndex = () => output.content.length - 1;
const appendPendingMessageDelta = (delta: string) => {
pendingMessageText = `${pendingMessageText ?? ""}${delta}`;
const priorText = stringifyUnknown(lastTextBlock?.block.text);
if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) {
return;
}
// Diverged from the prior text: this is a distinct message, so open its
// block now and replay the withheld text as one delta.
currentBlock = { type: "text", text: pendingMessageText };
output.content.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
stream.push({ type: "text_delta", contentIndex: blockIndex(), delta: pendingMessageText });
pendingMessageText = null;
};
const appendCompletedResponseTextItem = (item: Record<string, unknown>) => {
const text = readResponsesOutputMessageText(item);
if (!text) {
return;
}
const phase = (item.phase as "commentary" | "final_answer" | undefined) ?? undefined;
const collapse = resolveResponsesMessageSnapshotCollapse({
prior: lastTextBlock && {
text: stringifyUnknown(lastTextBlock.block.text),
phase: lastTextBlock.phase,
},
nextText: text,
nextPhase: phase,
});
if (collapse.kind === "extend" && lastTextBlock) {
// Cumulative snapshot of the prior message item: replace, don't append;
// the newest item's signature carries the content for replay (#91959).
lastTextBlock.block.text = collapse.text;
lastTextBlock.block.textSignature = encodeTextSignatureV1(stringifyUnknown(item.id), phase);
stream.push({
type: "text_end",
contentIndex: lastTextBlock.index,
content: collapse.text,
partial: output,
});
return;
}
const block: Record<string, unknown> = {
type: "text",
text,
textSignature: encodeTextSignatureV1(
stringifyUnknown(item.id),
(item.phase as "commentary" | "final_answer" | undefined) ?? undefined,
),
textSignature: encodeTextSignatureV1(stringifyUnknown(item.id), phase),
};
output.content.push(block);
lastTextBlock = { block, index: blockIndex(), phase };
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
stream.push({
type: "text_end",
@@ -1534,7 +1578,12 @@ async function processResponsesStream(
}
if (rawItem.type === "message") {
appendCompletedResponseTextItem(rawItem);
} else if (rawItem.type === "function_call") {
continue;
}
// Any non-message item (reasoning, tool call) is a real boundary; a later
// message must not collapse across it, mirroring the streaming path.
lastTextBlock = null;
if (rawItem.type === "function_call") {
appendCompletedResponseToolCallItem(rawItem);
}
}
@@ -1569,6 +1618,12 @@ async function processResponsesStream(
output.responseId = stringifyUnknown((event.response as { id?: string } | undefined)?.id);
} else if (type === "response.output_item.added") {
const item = event.item as Record<string, unknown>;
if (item.type !== "message") {
// Snapshot collapse only applies to back-to-back message items; any
// other item is a real boundary (see resolveResponsesMessageSnapshotCollapse).
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning") {
currentItem = item;
currentBlock = { type: "thinking", thinking: "" };
@@ -1576,9 +1631,14 @@ async function processResponsesStream(
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 });
if (lastTextBlock) {
currentBlock = null;
pendingMessageText = "";
} else {
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 = {
@@ -1602,13 +1662,17 @@ async function processResponsesStream(
});
}
} else if (isResponsesTextDeltaEventType(type) || type === "response.refusal.delta") {
if (currentItem?.type === "message" && currentBlock?.type === "text") {
currentBlock.text = `${stringifyUnknown(currentBlock.text)}${stringifyUnknown(event.delta)}`;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: stringifyUnknown(event.delta),
});
if (currentItem?.type === "message") {
if (pendingMessageText !== null) {
appendPendingMessageDelta(stringifyUnknown(event.delta));
} else if (currentBlock?.type === "text") {
currentBlock.text = `${stringifyUnknown(currentBlock.text)}${stringifyUnknown(event.delta)}`;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: stringifyUnknown(event.delta),
});
}
}
} else if (type === "response.function_call_arguments.delta") {
if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") {
@@ -1623,6 +1687,10 @@ async function processResponsesStream(
}
} else if (type === "response.output_item.done") {
const item = event.item as Record<string, unknown>;
if (item.type !== "message") {
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning" && currentBlock?.type === "thinking") {
const summary = Array.isArray(item.summary)
? item.summary
@@ -1648,9 +1716,12 @@ async function processResponsesStream(
partial: output,
});
currentBlock = null;
} else if (item.type === "message" && currentBlock?.type === "text") {
} else if (
item.type === "message" &&
(currentBlock?.type === "text" || pendingMessageText !== null)
) {
const content = Array.isArray(item.content) ? item.content : [];
currentBlock.text = content
const finalText = content
.map((part) => {
const contentPart = part as { type?: string; text?: string; refusal?: string };
return isResponsesTextContentPartType(contentPart.type)
@@ -1658,16 +1729,53 @@ async function processResponsesStream(
: (contentPart.refusal ?? "");
})
.join("");
currentBlock.textSignature = encodeTextSignatureV1(
stringifyUnknown(item.id),
(item.phase as "commentary" | "final_answer" | undefined) ?? undefined,
);
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: stringifyUnknown(currentBlock.text),
partial: output,
});
const phase = (item.phase as "commentary" | "final_answer" | undefined) ?? undefined;
const collapse =
pendingMessageText !== null
? resolveResponsesMessageSnapshotCollapse({
prior: lastTextBlock && {
text: stringifyUnknown(lastTextBlock.block.text),
phase: lastTextBlock.phase,
},
nextText: finalText,
nextPhase: phase,
})
: ({ kind: "keep" } as const);
pendingMessageText = null;
if (collapse.kind === "extend" && lastTextBlock) {
// Cumulative snapshot of the prior message item: replace its text
// instead of appending another copy. The deferred block was never
// started publicly, and the newest item's signature is kept so
// replay carries the item that produced this content (#91959).
lastTextBlock.block.text = collapse.text;
lastTextBlock.block.textSignature = encodeTextSignatureV1(
stringifyUnknown(item.id),
phase,
);
stream.push({
type: "text_end",
contentIndex: lastTextBlock.index,
content: collapse.text,
partial: output,
});
} else {
if (currentBlock?.type !== "text") {
// Deferred distinct message: open its block now, balanced with the
// text_end below.
currentBlock = { type: "text", text: "" };
output.content.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
}
currentBlock.text = finalText;
currentBlock.textSignature = encodeTextSignatureV1(stringifyUnknown(item.id), phase);
lastTextBlock = { block: currentBlock, index: blockIndex(), phase };
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: stringifyUnknown(currentBlock.text),
partial: output,
});
}
currentBlock = null;
} else if (item.type === "function_call") {
const args =
src/llm/providers/openai-responses-shared.test.ts
+317
View File
@@ -587,6 +587,323 @@ describe("processResponsesStream", () => {
"toolcall_end",
]);
});
it("collapses cumulative message snapshot items into one text block (#91959)", async () => {
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
const events: Array<Record<string, unknown>> = [];
const collect = (async () => {
for await (const event of stream) {
events.push(event as unknown as Record<string, unknown>);
}
})();
const snapshot1 = "Self-attention computes";
const snapshot2 = "Self-attention computes Q/K/V projections";
const snapshot3 = "Self-attention computes Q/K/V projections for each token.";
const messageItem = (id: string, text: string) => ({
type: "message",
id,
phase: "final_answer",
content: [{ type: "output_text", text }],
});
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "final_answer" },
},
{ type: "response.content_part.added", part: { type: "output_text", text: "" } },
{ type: "response.output_text.delta", delta: snapshot1 },
{ type: "response.output_item.done", item: messageItem("msg_1", snapshot1) },
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{ type: "response.output_item.done", item: messageItem("msg_2", snapshot2) },
{
type: "response.output_item.added",
item: { type: "message", id: "msg_3", phase: "final_answer" },
},
{ type: "response.output_item.done", item: messageItem("msg_3", snapshot3) },
{ type: "response.completed", response: { id: "resp_1", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
stream.end();
await collect;
expect(output.content).toEqual([
{
type: "text",
text: snapshot3,
textSignature: JSON.stringify({ v: 1, id: "msg_3", phase: "final_answer" }),
},
]);
// Balanced lifecycle: exactly one text_start, every event on index 0, and
// each collapsed snapshot re-ends the same block with its grown content.
expect(events.map((event) => [event.type, event.contentIndex])).toEqual([
["text_start", 0],
["text_delta", 0],
["text_end", 0],
["text_end", 0],
["text_end", 0],
]);
expect(
events.filter((event) => event.type === "text_end").map((event) => event.content),
).toEqual([snapshot1, snapshot2, snapshot3]);
});
it.each([
["identical", "Hello world.", "Hello world."],
["shrinking", "Step one. Step two.", "Step one."],
])("keeps %s adjacent same-phase message items as distinct blocks", async (_label, a, b) => {
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
const events: Array<Record<string, unknown>> = [];
const collect = (async () => {
for await (const event of stream) {
events.push(event as unknown as Record<string, unknown>);
}
})();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_1",
phase: "final_answer",
content: [{ type: "output_text", text: a }],
},
},
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_2",
phase: "final_answer",
content: [{ type: "output_text", text: b }],
},
},
{ type: "response.completed", response: { id: "resp_1", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
stream.end();
await collect;
// Only strict extensions collapse; equal or shrinking items are real,
// independently identified messages and must never be removed.
expect(output.content).toEqual([
{
type: "text",
text: a,
textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "final_answer" }),
},
{
type: "text",
text: b,
textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }),
},
]);
// The deferred second item still opens and closes its own block.
expect(events.map((event) => [event.type, event.contentIndex])).toEqual([
["text_start", 0],
["text_end", 0],
["text_start", 1],
["text_end", 1],
]);
});
it("streams a deferred distinct message live once its text diverges from the prior block", async () => {
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
const events: Array<Record<string, unknown>> = [];
const collect = (async () => {
for await (const event of stream) {
events.push(event as unknown as Record<string, unknown>);
}
})();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_1",
phase: "final_answer",
content: [{ type: "output_text", text: "Hello." }],
},
},
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{ type: "response.content_part.added", part: { type: "output_text", text: "" } },
{ type: "response.output_text.delta", delta: "Good" },
{ type: "response.output_text.delta", delta: "bye" },
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_2",
phase: "final_answer",
content: [{ type: "output_text", text: "Goodbye" }],
},
},
{ type: "response.completed", response: { id: "resp_1", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
stream.end();
await collect;
expect(output.content).toEqual([
{
type: "text",
text: "Hello.",
textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "final_answer" }),
},
{
type: "text",
text: "Goodbye",
textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }),
},
]);
// The withheld prefix is replayed as one delta at divergence ("Good"
// diverges from "Hello."), then later deltas stream live.
expect(events.map((event) => [event.type, event.contentIndex, event.delta ?? null])).toEqual([
["text_start", 0, null],
["text_end", 0, null],
["text_start", 1, null],
["text_delta", 1, "Good"],
["text_delta", 1, "bye"],
["text_end", 1, null],
]);
});
it("keeps prefix-nested message items separated by a reasoning item as separate blocks", async () => {
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_1",
phase: "final_answer",
content: [{ type: "output_text", text: "Step one." }],
},
},
{ type: "response.output_item.added", item: { type: "reasoning" } },
{
type: "response.output_item.done",
item: { type: "reasoning", id: "rs_1", summary: [] },
},
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_2",
phase: "final_answer",
content: [{ type: "output_text", text: "Step one. Step two." }],
},
},
{ type: "response.completed", response: { id: "resp_1", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
stream.end();
// Collapsing across the reasoning block would orphan it for replay.
expect(output.content.map((block) => block.type)).toEqual(["text", "thinking", "text"]);
expect(output.content[2]).toMatchObject({ type: "text", text: "Step one. Step two." });
});
it("keeps prefix-nested message items with different phases as separate blocks", async () => {
const output = createAssistantOutput();
const stream = new AssistantMessageEventStream();
await processResponsesStream(
responseEvents([
{
type: "response.output_item.added",
item: { type: "message", id: "msg_1", phase: "commentary" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_1",
phase: "commentary",
content: [{ type: "output_text", text: "Done" }],
},
},
{
type: "response.output_item.added",
item: { type: "message", id: "msg_2", phase: "final_answer" },
},
{
type: "response.output_item.done",
item: {
type: "message",
id: "msg_2",
phase: "final_answer",
content: [{ type: "output_text", text: "Done." }],
},
},
{ type: "response.completed", response: { id: "resp_1", status: "completed" } },
]),
output,
stream,
nativeOpenAIModel,
);
stream.end();
expect(output.content).toEqual([
{
type: "text",
text: "Done",
textSignature: JSON.stringify({ v: 1, id: "msg_1", phase: "commentary" }),
},
{
type: "text",
text: "Done.",
textSignature: JSON.stringify({ v: 1, id: "msg_2", phase: "final_answer" }),
},
]);
});
});
describe("Azure OpenAI Responses content type support", () => {
src/llm/providers/openai-responses-shared.ts
+133 -37
View File
@@ -20,6 +20,7 @@ import {
type AzureResponsesTextDeltaEvent,
isAzureResponsesTextDeltaEvent,
isResponsesTextContentPartType,
resolveResponsesMessageSnapshotCollapse,
} from "../../shared/openai-responses-stream-compat.js";
import { calculateCost, clampThinkingLevel } from "../model-utils.js";
import type {
@@ -579,14 +580,48 @@ export async function processResponsesStream<TApi extends Api>(
| null = null;
let currentBlock: ThinkingContent | TextContent | (ToolCall & { partialJson: string }) | null =
null;
let lastTextBlock: {
block: TextContent;
index: number;
phase: TextSignatureV1["phase"] | undefined;
} | null = null;
// While a message item may still be a cumulative snapshot of lastTextBlock,
// its public block is deferred so a collapsed item never leaves an
// unbalanced text_start behind (#91959). null = no deferral in progress.
let pendingMessageText: string | null = null;
const blocks = output.content;
const blockIndex = () => blocks.length - 1;
const appendPendingMessageDelta = (delta: string) => {
pendingMessageText = `${pendingMessageText ?? ""}${delta}`;
const priorText = lastTextBlock?.block.text ?? "";
if (priorText.startsWith(pendingMessageText) || pendingMessageText.startsWith(priorText)) {
return;
}
// Diverged from the prior text: this is a distinct message, so open its
// block now and replay the withheld text as one delta.
currentBlock = { type: "text", text: pendingMessageText };
blocks.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: pendingMessageText,
partial: output,
});
pendingMessageText = null;
};
for await (const event of openaiStream) {
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 !== "message") {
// Snapshot collapse only applies to back-to-back message items; any
// other item is a real boundary (see resolveResponsesMessageSnapshotCollapse).
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning") {
currentItem = item;
currentBlock = { type: "thinking", thinking: "" };
@@ -594,9 +629,14 @@ export async function processResponsesStream<TApi extends Api>(
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 });
if (lastTextBlock) {
currentBlock = null;
pendingMessageText = "";
} else {
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 = {
@@ -666,48 +706,39 @@ export async function processResponsesStream<TApi extends Api>(
}
}
} else if (event.type === "response.output_text.delta") {
if (currentItem?.type === "message" && currentBlock?.type === "text") {
if (currentItem?.type === "message") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (isResponsesTextContentPartType(lastPart?.type)) {
currentBlock.text += event.delta;
lastPart.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
}
} else if (isAzureResponsesTextDeltaEvent(event)) {
if (currentItem?.type === "message" && currentBlock?.type === "text") {
if (currentItem?.type === "message") {
currentItem.content = currentItem.content || [];
let lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type !== "text") {
lastPart = { type: "text", text: "" };
currentItem.content.push(lastPart);
}
currentBlock.text += event.delta;
lastPart.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
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") {
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
lastPart.refusal += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
@@ -716,6 +747,27 @@ export async function processResponsesStream<TApi extends Api>(
});
}
}
} else if (event.type === "response.refusal.delta") {
if (currentItem?.type === "message") {
if (!currentItem.content || currentItem.content.length === 0) {
continue;
}
const lastPart = currentItem.content[currentItem.content.length - 1];
if (lastPart?.type === "refusal") {
lastPart.refusal += event.delta;
if (pendingMessageText !== null) {
appendPendingMessageDelta(event.delta);
} else if (currentBlock?.type === "text") {
currentBlock.text += event.delta;
stream.push({
type: "text_delta",
contentIndex: blockIndex(),
delta: event.delta,
partial: output,
});
}
}
}
} else if (event.type === "response.function_call_arguments.delta") {
if (currentItem?.type === "function_call" && currentBlock?.type === "toolCall") {
currentBlock.partialJson += event.delta;
@@ -754,6 +806,10 @@ export async function processResponsesStream<TApi extends Api>(
}
} else if (event.type === "response.output_item.done") {
const item = event.item;
if (item.type !== "message") {
lastTextBlock = null;
pendingMessageText = null;
}
if (item.type === "reasoning" && currentBlock?.type === "thinking") {
const summaryText = item.summary?.map((s) => s.text).join("\n\n") || "";
@@ -767,18 +823,58 @@ export async function processResponsesStream<TApi extends Api>(
partial: output,
});
currentBlock = null;
} else if (item.type === "message" && currentBlock?.type === "text") {
} else if (
item.type === "message" &&
(currentBlock?.type === "text" || pendingMessageText !== null)
) {
// Support both OpenAI "output_text" and Azure "text" content types
currentBlock.text = item.content
const finalText = item.content
.map((c) => (c.type === "output_text" || c.type === "text" ? c.text : c.refusal))
.join("");
currentBlock.textSignature = encodeTextSignatureV1(item.id, item.phase ?? undefined);
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: currentBlock.text,
partial: output,
});
const phase = item.phase ?? undefined;
const collapse =
pendingMessageText !== null
? resolveResponsesMessageSnapshotCollapse({
prior: lastTextBlock && {
text: lastTextBlock.block.text,
phase: lastTextBlock.phase,
},
nextText: finalText,
nextPhase: phase,
})
: ({ kind: "keep" } as const);
pendingMessageText = null;
if (collapse.kind === "extend" && lastTextBlock) {
// Cumulative snapshot of the prior message item: replace its text
// instead of appending another copy. The deferred block was never
// started publicly, and the newest item's signature is kept so
// replay carries the item that produced this content (#91959).
lastTextBlock.block.text = collapse.text;
lastTextBlock.block.textSignature = encodeTextSignatureV1(item.id, phase);
stream.push({
type: "text_end",
contentIndex: lastTextBlock.index,
content: collapse.text,
partial: output,
});
} else {
if (currentBlock?.type !== "text") {
// Deferred distinct message: open its block now, balanced with the
// text_end below.
currentBlock = { type: "text", text: "" };
blocks.push(currentBlock);
stream.push({ type: "text_start", contentIndex: blockIndex(), partial: output });
}
currentBlock.text = finalText;
currentBlock.textSignature = encodeTextSignatureV1(item.id, phase);
lastTextBlock = { block: currentBlock, index: blockIndex(), phase };
stream.push({
type: "text_end",
contentIndex: blockIndex(),
content: currentBlock.text,
partial: output,
});
}
currentBlock = null;
} else if (item.type === "function_call") {
const args =
src/shared/openai-responses-stream-compat.ts
+27
View File
@@ -49,3 +49,30 @@ export function isAzureResponsesTextDeltaEvent(event: {
}): event is AzureResponsesTextDeltaEvent {
return isAzureResponsesTextDeltaEventType(event.type) && typeof event.delta === "string";
}
export type ResponsesMessageSnapshotCollapse = { kind: "extend"; text: string } | { kind: "keep" };
// Some openai-responses providers re-emit the assistant message as cumulative
// snapshot items — each a strict prefix-superset of the previous one — instead
// of one final message item. A same-phase strict extension replaces the prior
// text block, or the visible reply repeats once per snapshot (#91959).
// Extension-only on purpose: equal or shrinking adjacent items stay distinct
// (the Responses protocol allows multiple message items per response), so a
// false positive can only merge rendering — it can never lose text.
// `prior` must be the immediately preceding output item: collapsing across
// reasoning/function_call boundaries would drop real post-tool messages and
// orphan reasoning items, which OpenAI replay rejects.
export function resolveResponsesMessageSnapshotCollapse(params: {
prior: { text: string; phase: string | undefined } | null;
nextText: string;
nextPhase: string | undefined;
}): ResponsesMessageSnapshotCollapse {
const { prior, nextText } = params;
if (!prior?.text || !nextText || prior.phase !== params.nextPhase) {
return { kind: "keep" };
}
if (nextText.length > prior.text.length && nextText.startsWith(prior.text)) {
return { kind: "extend", text: nextText };
}
return { kind: "keep" };
}