pjandro/OmniRoute
1
0
Fork 0
mirror of https://github.com/diegosouzapw/OmniRoute.git synced 2026-08-18 13:14:56 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(sse): provider-response summary format bugs (dashboard Provider Response panel) (#10037)

Browse Source 
* fix(sse): provider-response summary reconstructed from truncated events

The dashboard's "Provider Response" panel showed a stale, incomplete
snapshot for long streamed responses. Root cause: open-sse/utils/stream.ts
reconstructed the summary from
buildStreamSummaryFromEvents(providerPayloadCollector.getEvents(), ...)
-- but getEvents() only returns whatever survived the collector's
maxEvents/maxBytes cap, so once a stream exceeded it (easy with a
reasoning + tool-calling model), everything after the cutoff (final
finish_reason, tool_calls, rest of reasoning_content, usage) was
silently dropped from the reconstruction, even though the client
actually received the correct, complete response.

Fix: streamPayloadCollector.ts's per-format summary builders
(buildOpenAISummary/buildResponsesSummary/buildClaudeSummary/
buildGeminiSummary) are now also available as incremental reducers
(createXReducer: ingest one chunk at a time, finalize at the end).
createStructuredSSECollector accepts a format + fallbackModel and feeds
the reducer on every push() -- including chunks that get dropped from
the retained event array once the cap is hit -- via a new getSummary()
method. stream.ts's error-path call site now uses
collector.getSummary() instead of reconstructing from the (possibly
truncated) getEvents().

Extracted from a squashed commit (originally authored alongside a
conversation-tracking continuation fix in the same commit) -- only the
files relevant to this SSE-summary bug are included here
(stream.ts/streamPayloadCollector.ts + their test); the unrelated
conversationTracker.ts continuation fix stays with the conversation-
tracking PR it belongs to.

Test plan:
- New TDD regression tests in tests/unit/stream-payload-collector.test.ts,
  confirmed failing before the fix and passing after.

* fix(sse): provider-response summary used the client's format, not the provider's

providerPayloadCollector (dashboard "Provider Response" panel) was keyed on
sourceFormat (the CLIENT's wire format) instead of targetFormat (the
PROVIDER's — see createSSEStream's own @param doc: "targetFormat - Provider
format", "sourceFormat - Client format"). Whenever a request translates
between two different formats — e.g. a Responses-API client routed to a
plain-OpenAI-chat-completions upstream, the common OpenClaw/opencode-zen
shape — the reducer picked for sourceFormat could never recognize the
provider's actual raw event shape, so it stayed stuck at its empty initial
state. The dashboard's "Provider Response" panel showed a permanently empty
`output: []` while "Client Response" (built from separately-accumulated
state, unaffected by this bug) correctly showed full content — reading as
if the two panels simply disagreed about the same request.

Confirmed live via a wire-level pcap capture (scripts/sre/tcp-close-
analyzer.py) cross-referenced against the dashboard log
(1786032832181-1c6275): the actual response was complete and correct: this
was purely a logging/summary bug, never a wire-format bug.

Fix is mode-aware: TRANSLATE mode uses targetFormat (the provider's true
format); PASSTHROUGH mode keeps sourceFormat, since passthrough has no
separate provider/client format split — nothing gets translated there, and
real passthrough callers (createPassthroughStreamWithLogger) don't even
pass targetFormat.

New regression test reproduces the exact live scenario (Responses-API
source, OpenAI target, real chat.completion.chunk deltas) and asserts the
provider summary reflects them — confirmed it fails with the old
`sourceFormat`-keyed code (reproducing the live `output: []`-style
symptom) and passes with the fix.

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>

* fix(sse): stamp object: chat.completion on the provider-summary fallback

createSSEStream's providerPayloadCollector.build() falls back to the
synthesized responseBody as the "Provider Response" dashboard summary
whenever sourceFormat/targetFormat isn't OPENAI_RESPONSES (in both the
passthrough and translate branches) -- but responseBody is built purely
for the client and never carries an `object` field at all, so the
summary ended up with `object: undefined` instead of the expected
"chat.completion", even though everything else (choices, usage) was
correct.

Caught by this PR's own new regression test ("createSSEStream translate
mode: providerPayload summary reflects the PROVIDER's format, not the
client's") -- the code itself was unchanged by the rebase (applied
cleanly from the original commit), so this was a latent gap in the
original fix, not a rebase regression.

Fix: stamp `object: "chat.completion"` on a shallow copy used only for
the provider summary in both branches; responseBody itself (sent to the
client elsewhere) stays untouched.

Verified: tests/unit/stream-utils.test.ts 51/52 passing (the one
remaining failure is an unrelated, pre-existing v3.6.6-era test,
confirmed present and failing identically on a pristine
upstream/release/v3.8.50 checkout -- base-red inherited: #9985).
typecheck/lint clean (pre-existing unrelated errors elsewhere in the
file, confirmed identical to upstream).

---------

Co-authored-by: Markus Hartung <markus.hartung@gmail.com>
This commit is contained in:
Markus Hartung
2026-08-13 09:02:30 +02:00
committed by GitHub
parent ae54c6221c
commit 4bda22583e
4 changed files with 661 additions and 375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
open-sse/utils/stream.ts
View File

@@ -787,6 +787,29 @@ export function createSSEStream(options: StreamOptions = {}) {
let upstreamErrorForwarded = false;
const providerPayloadCollector = createStructuredSSECollector({
stage: "provider_response",
// #9315: compute the summary live from every pushed chunk (not just the
// ones that survive the storage cap below) so a long stream never shows a
// stale/incomplete "provider response" in the dashboard.
//
// Real bug: this was unconditionally `sourceFormat` (the CLIENT's wire
// format — see this function's own @param doc above). In TRANSLATE mode
// the chunks pushed here are the RAW PROVIDER response, whose format is
// `targetFormat` (@param "Provider format (for translate mode)"), not
// sourceFormat. Whenever a client's format differs from the provider's
// (e.g. a Responses-API client routed to a plain-OpenAI-chat-completions
// upstream — the OpenClaw/opencode-zen case that surfaced this live), the
// reducer picked for `sourceFormat` could never recognize the provider's
// actual event shape, so it never left its empty initial state — the
// dashboard's "Provider Response" panel permanently showed
// `output: []`/empty while "Client Response" (built from
// separately-accumulated state, unaffected by this) correctly showed full
// content, reading as if the two panels simply disagreed. PASSTHROUGH
// mode has no separate provider/client format split — nothing gets
// translated, so the provider's raw chunks genuinely ARE in sourceFormat
// (and real passthrough callers, e.g. createPassthroughStreamWithLogger,
// don't even pass targetFormat) — keep using sourceFormat there.
format: mode === STREAM_MODE.TRANSLATE ? targetFormat : sourceFormat,
fallbackModel: model,
});
const clientPayloadCollector = createStructuredSSECollector({
stage: "client_response",
@@ -1641,7 +1664,9 @@ export function createSSEStream(options: StreamOptions = {}) {
// retry." with finish_reason: "stop" — clients (Goose/opencode) feed that
// text back as a turn and spin in a retry loop. This restores the #3400
// behavior that #3422 inadvertently reverted (regression #3388/#3502).
if (Array.isArray(parsed.choices) && (parsed.choices.length === 0 ||
if (
Array.isArray(parsed.choices) &&
(parsed.choices.length === 0 ||
(parsed.choices.length === 1 &&
parsed.choices[0]?.delta &&
typeof parsed.choices[0].delta === "object" &&
@@ -2483,7 +2508,11 @@ export function createSSEStream(options: StreamOptions = {}) {
// #9315 switched the summary to the accumulated responseBody to avoid
// stale/truncated event data — but responseBody here is synthesized in
// chat-completion shape, which loses the Responses API `response` object.
// Keep the events-derived summary for OPENAI_RESPONSES only.
// Keep the events-derived summary for OPENAI_RESPONSES only. responseBody
// itself never carries an `object` marker (it's built purely for the
// client, which doesn't need one) — the dashboard's Provider Response
// panel does, so stamp `object: "chat.completion"` on a shallow copy
// used only for this summary, leaving responseBody itself untouched.
providerPayload: providerPayloadCollector.build(
sourceFormat === FORMATS.OPENAI_RESPONSES
? buildStreamSummaryFromEvents(
@@ -2491,7 +2520,7 @@ export function createSSEStream(options: StreamOptions = {}) {
sourceFormat,
model
)
: responseBody,
: { object: "chat.completion", ...responseBody },
{ includeEvents: false }
),
clientPayload: clientPayloadCollector.build(responseBody, {
@@ -2600,11 +2629,7 @@ export function createSSEStream(options: StreamOptions = {}) {
error: err.message,
errorCode: err.code,
providerPayload: providerPayloadCollector.build(
buildStreamSummaryFromEvents(
providerPayloadCollector.getEvents(),
targetFormat,
model
),
providerPayloadCollector.getSummary(),
{ includeEvents: false }
),
clientPayload: clientPayloadCollector.build(errorBody, {
@@ -2783,7 +2808,11 @@ export function createSSEStream(options: StreamOptions = {}) {
usage: state?.usage,
responseBody,
// Same OPENAI_RESPONSES carve-out as the passthrough branch above —
// the synthesized chat-shaped responseBody drops the `response` object.
// the synthesized chat-shaped responseBody drops the `response` object,
// and (like the passthrough branch) never carries an `object` marker at
// all — stamp `object: "chat.completion"` on a shallow copy used only
// for this summary; responseBody itself (sent to the client / below)
// stays untouched.
providerPayload: providerPayloadCollector.build(
targetFormat === FORMATS.OPENAI_RESPONSES
? buildStreamSummaryFromEvents(
@@ -2791,7 +2820,7 @@ export function createSSEStream(options: StreamOptions = {}) {
targetFormat,
model
)
: responseBody,
: { object: "chat.completion", ...responseBody },
{ includeEvents: false }
),
clientPayload: clientPayloadCollector.build(responseBody, {

830
open-sse/utils/streamPayloadCollector.ts
View File

@@ -12,6 +12,16 @@ type CollectorOptions = {
maxEvents?: number;
maxBytes?: number;
stage?: string;
// When set, every pushed payload — even ones dropped from the retained
// `events` array once maxEvents/maxBytes is hit — is also fed to a live
// per-format summary reducer, so build()'s summary reflects the FULL
// stream, not just the surviving (possibly truncated) event slice.
// See #9315: reconstructing the summary from getEvents() after the fact
// means a long stream that exceeds the cap gets a stale/incomplete
// "provider response" (missing tool_calls, wrong finish_reason, cut-off
// content) even though the actual served response was correct.
format?: string | null;
fallbackModel?: string | null;
};
type BuildOptions = {
@@ -20,6 +30,11 @@ type BuildOptions = {
type JsonRecord = Record<string, unknown>;
interface SummaryReducer {
ingest(payload: JsonRecord): void;
finalize(): unknown;
}
function getEventName(payload: unknown): string | undefined {
if (!payload || typeof payload !== "object" || Array.isArray(payload)) return undefined;
@@ -113,13 +128,15 @@ function tryParseJson(raw: string): unknown {
}
}
function buildOpenAISummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const payloads = events
.map((evt) => asRecord(evt.data))
.filter((payload) => Object.keys(payload).length);
if (payloads.length === 0) return null;
// ─── Per-format live reducers ────────────────────────────────────────────────
// Each reducer mirrors the corresponding build*Summary()'s original for-loop
// body exactly (ingest = one loop iteration, finalize = the post-loop return),
// just restructured so it can be fed one payload at a time as chunks arrive —
// including chunks that will later be dropped from the retained event array
// once the collector's storage cap is hit.
const first = payloads[0];
function createOpenAIReducer(fallbackModel?: string | null): SummaryReducer {
let first: JsonRecord | null = null;
const contentParts: string[] = [];
const reasoningParts: string[] = [];
type ToolCall = {
@@ -156,124 +173,126 @@ function buildOpenAISummary(events: StructuredSSEEvent[], fallbackModel?: string
return `seq:${unknownToolCallSeq}`;
};
for (const chunk of payloads) {
const choice = asRecord(Array.isArray(chunk.choices) ? chunk.choices[0] : null);
const delta = asRecord(choice.delta);
return {
ingest(chunk: JsonRecord) {
if (Object.keys(chunk).length === 0) return;
if (!first) first = chunk;
if (typeof delta.content === "string" && delta.content.length > 0) {
contentParts.push(delta.content);
}
if (Array.isArray(delta.content)) {
for (const part of delta.content) {
const partObj = asRecord(part);
if (typeof partObj.text === "string" && partObj.text.length > 0) {
contentParts.push(partObj.text);
const choice = asRecord(Array.isArray(chunk.choices) ? chunk.choices[0] : null);
const delta = asRecord(choice.delta);
if (typeof delta.content === "string" && delta.content.length > 0) {
contentParts.push(delta.content);
}
if (Array.isArray(delta.content)) {
for (const part of delta.content) {
const partObj = asRecord(part);
if (typeof partObj.text === "string" && partObj.text.length > 0) {
contentParts.push(partObj.text);
}
}
}
}
if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) {
reasoningParts.push(delta.reasoning_content);
}
// Normalize `reasoning` alias (NVIDIA kimi-k2.5 etc.)
if (
typeof delta.reasoning === "string" &&
delta.reasoning.length > 0 &&
!delta.reasoning_content
) {
reasoningParts.push(delta.reasoning);
}
if (Array.isArray(delta.tool_calls)) {
for (const item of delta.tool_calls) {
const toolCall = asRecord(item);
const key = getToolCallKey(toolCall);
const existing = toolCalls.get(key);
const deltaArgs =
typeof asRecord(toolCall.function).arguments === "string"
? String(asRecord(toolCall.function).arguments)
: "";
if (!existing) {
toolCalls.set(key, {
id: typeof toolCall.id === "string" ? toolCall.id : null,
index: Number.isInteger(toolCall.index) ? Number(toolCall.index) : toolCalls.size,
type: toString(toolCall.type, "function"),
function: {
name: toString(asRecord(toolCall.function).name, "unknown"),
arguments: deltaArgs,
},
});
continue;
}
existing.id = existing.id || (typeof toolCall.id === "string" ? toolCall.id : null);
if (
(!Number.isInteger(existing.index) || existing.index < 0) &&
Number.isInteger(toolCall.index)
) {
existing.index = Number(toolCall.index);
}
if (typeof asRecord(toolCall.function).name === "string" && !existing.function.name) {
existing.function.name = String(asRecord(toolCall.function).name);
}
existing.function.arguments += deltaArgs;
if (typeof delta.reasoning_content === "string" && delta.reasoning_content.length > 0) {
reasoningParts.push(delta.reasoning_content);
}
// Normalize `reasoning` alias (NVIDIA kimi-k2.5 etc.)
if (
typeof delta.reasoning === "string" &&
delta.reasoning.length > 0 &&
!delta.reasoning_content
) {
reasoningParts.push(delta.reasoning);
}
}
if (typeof choice.finish_reason === "string" && choice.finish_reason.length > 0) {
finishReason = choice.finish_reason;
}
if (chunk.usage && typeof chunk.usage === "object") {
usage = { ...asRecord(chunk.usage) };
}
}
if (Array.isArray(delta.tool_calls)) {
for (const item of delta.tool_calls) {
const toolCall = asRecord(item);
const key = getToolCallKey(toolCall);
const existing = toolCalls.get(key);
const deltaArgs =
typeof asRecord(toolCall.function).arguments === "string"
? String(asRecord(toolCall.function).arguments)
: "";
const joinedContent = contentParts.length > 0 ? contentParts.join("").trim() : null;
const joinedReasoning = reasoningParts.length > 0 ? reasoningParts.join("").trim() : null;
const message: JsonRecord = {
role: "assistant",
content: joinedContent || null,
if (!existing) {
toolCalls.set(key, {
id: typeof toolCall.id === "string" ? toolCall.id : null,
index: Number.isInteger(toolCall.index) ? Number(toolCall.index) : toolCalls.size,
type: toString(toolCall.type, "function"),
function: {
name: toString(asRecord(toolCall.function).name, "unknown"),
arguments: deltaArgs,
},
});
continue;
}
existing.id = existing.id || (typeof toolCall.id === "string" ? toolCall.id : null);
if (
(!Number.isInteger(existing.index) || existing.index < 0) &&
Number.isInteger(toolCall.index)
) {
existing.index = Number(toolCall.index);
}
if (typeof asRecord(toolCall.function).name === "string" && !existing.function.name) {
existing.function.name = String(asRecord(toolCall.function).name);
}
existing.function.arguments += deltaArgs;
}
}
if (typeof choice.finish_reason === "string" && choice.finish_reason.length > 0) {
finishReason = choice.finish_reason;
}
if (chunk.usage && typeof chunk.usage === "object") {
usage = { ...asRecord(chunk.usage) };
}
},
finalize(): unknown {
if (!first) return null;
const joinedContent = contentParts.length > 0 ? contentParts.join("").trim() : null;
const joinedReasoning = reasoningParts.length > 0 ? reasoningParts.join("").trim() : null;
const message: JsonRecord = {
role: "assistant",
content: joinedContent || null,
};
if (joinedReasoning) {
message.reasoning_content = joinedReasoning;
}
const finalToolCalls = [...toolCalls.values()].sort((a, b) => a.index - b.index);
if (finalToolCalls.length > 0) {
finishReason = "tool_calls";
message.tool_calls = finalToolCalls;
}
const result: JsonRecord = {
id: toString(first.id, `chatcmpl-${Date.now()}`),
object: "chat.completion",
created: toNumber(first.created, Math.floor(Date.now() / 1000)),
model: toString(first.model, fallbackModel || "unknown"),
choices: [
{
index: 0,
message,
finish_reason: finishReason,
},
],
};
if (usage && Object.keys(usage).length > 0) {
result.usage = usage;
}
return result;
},
};
if (joinedReasoning) {
message.reasoning_content = joinedReasoning;
}
const finalToolCalls = [...toolCalls.values()].sort((a, b) => a.index - b.index);
if (finalToolCalls.length > 0) {
finishReason = "tool_calls";
message.tool_calls = finalToolCalls;
}
const result: JsonRecord = {
id: toString(first.id, `chatcmpl-${Date.now()}`),
object: "chat.completion",
created: toNumber(first.created, Math.floor(Date.now() / 1000)),
model: toString(first.model, fallbackModel || "unknown"),
choices: [
{
index: 0,
message,
finish_reason: finishReason,
},
],
};
if (usage && Object.keys(usage).length > 0) {
result.usage = usage;
}
return result;
}
function buildResponsesSummary(
events: StructuredSSEEvent[],
fallbackModel?: string | null
): unknown {
const payloads = events
.map((evt) => asRecord(evt.data))
.filter((payload) => Object.keys(payload).length);
if (payloads.length === 0) return null;
function createResponsesReducer(fallbackModel?: string | null): SummaryReducer {
let sawAny = false;
let completed: JsonRecord | null = null;
let latestResponse: JsonRecord | null = null;
let usage: JsonRecord | null = null;
@@ -289,67 +308,72 @@ function buildResponsesSummary(
]
: [];
for (const payload of payloads) {
const eventType = toString(payload.type);
if (
eventType === "response.completed" &&
payload.response &&
typeof payload.response === "object"
) {
completed = asRecord(payload.response);
}
if (payload.response && typeof payload.response === "object") {
latestResponse = asRecord(payload.response);
} else if (payload.object === "response") {
latestResponse = payload;
}
if (
eventType === "response.output_text.delta" &&
typeof payload.delta === "string" &&
payload.delta.length > 0
) {
textParts.push(payload.delta);
}
if (payload.usage && typeof payload.usage === "object") {
usage = { ...asRecord(payload.usage) };
} else if (payload.response && typeof asRecord(payload.response).usage === "object") {
usage = { ...asRecord(asRecord(payload.response).usage) };
}
}
const picked = completed || latestResponse;
if (picked && Object.keys(picked).length > 0) {
const pickedOutput = Array.isArray(picked.output) ? picked.output : [];
return {
id: toString(picked.id, `resp_${Date.now()}`),
object: "response",
model: toString(picked.model, fallbackModel || "unknown"),
output: pickedOutput.length > 0 ? pickedOutput : buildOutputFromText(),
usage: picked.usage ?? usage ?? null,
status: toString(picked.status, completed ? "completed" : "in_progress"),
created_at: toNumber(picked.created_at, Math.floor(Date.now() / 1000)),
metadata: asRecord(picked.metadata),
};
}
return {
id: `resp_${Date.now()}`,
object: "response",
model: fallbackModel || "unknown",
output: buildOutputFromText(),
usage: usage ?? null,
status: "completed",
created_at: Math.floor(Date.now() / 1000),
metadata: {},
ingest(payload: JsonRecord) {
if (Object.keys(payload).length === 0) return;
sawAny = true;
const eventType = toString(payload.type);
if (
eventType === "response.completed" &&
payload.response &&
typeof payload.response === "object"
) {
completed = asRecord(payload.response);
}
if (payload.response && typeof payload.response === "object") {
latestResponse = asRecord(payload.response);
} else if (payload.object === "response") {
latestResponse = payload;
}
if (
eventType === "response.output_text.delta" &&
typeof payload.delta === "string" &&
payload.delta.length > 0
) {
textParts.push(payload.delta);
}
if (payload.usage && typeof payload.usage === "object") {
usage = { ...asRecord(payload.usage) };
} else if (payload.response && typeof asRecord(payload.response).usage === "object") {
usage = { ...asRecord(asRecord(payload.response).usage) };
}
},
finalize(): unknown {
if (!sawAny) return null;
const picked = completed || latestResponse;
if (picked && Object.keys(picked).length > 0) {
const pickedOutput = Array.isArray(picked.output) ? picked.output : [];
return {
id: toString(picked.id, `resp_${Date.now()}`),
object: "response",
model: toString(picked.model, fallbackModel || "unknown"),
output: pickedOutput.length > 0 ? pickedOutput : buildOutputFromText(),
usage: picked.usage ?? usage ?? null,
status: toString(picked.status, completed ? "completed" : "in_progress"),
created_at: toNumber(picked.created_at, Math.floor(Date.now() / 1000)),
metadata: asRecord(picked.metadata),
};
}
return {
id: `resp_${Date.now()}`,
object: "response",
model: fallbackModel || "unknown",
output: buildOutputFromText(),
usage: usage ?? null,
status: "completed",
created_at: Math.floor(Date.now() / 1000),
metadata: {},
};
},
};
}
function buildClaudeSummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const payloads = events
.map((evt) => asRecord(evt.data))
.filter((payload) => Object.keys(payload).length);
if (payloads.length === 0) return null;
function createClaudeReducer(fallbackModel?: string | null): SummaryReducer {
let sawAny = false;
type ClaudeBlock =
| { type: "text"; index: number; text: string }
| { type: "thinking"; index: number; thinking: string; signature?: string }
@@ -379,172 +403,177 @@ function buildClaudeSummary(events: StructuredSSEEvent[], fallbackModel?: string
// non-streaming JSON path. Last-writer-wins: the final snapshot is authoritative.
let contextManagement: JsonRecord | null = null;
for (const payload of payloads) {
const eventType = toString(payload.type);
if (
payload.context_management &&
typeof payload.context_management === "object" &&
!Array.isArray(payload.context_management)
) {
contextManagement = asRecord(payload.context_management);
}
if (eventType === "message_start") {
const message = asRecord(payload.message);
messageId = toString(message.id, messageId || `msg_${Date.now()}`);
model = toString(message.model, model);
role = toString(message.role, role);
mergeUsage(usage, message.usage);
continue;
}
return {
ingest(payload: JsonRecord) {
if (Object.keys(payload).length === 0) return;
sawAny = true;
if (eventType === "content_block_start") {
const index = toNumber(payload.index, blocks.size);
const contentBlock = asRecord(payload.content_block);
const blockType = toString(contentBlock.type);
if (blockType === "thinking") {
blocks.set(index, {
type: "thinking",
index,
thinking: toString(contentBlock.thinking),
signature:
typeof contentBlock.signature === "string" ? contentBlock.signature : undefined,
});
} else if (blockType === "tool_use") {
blocks.set(index, {
type: "tool_use",
index,
id: toString(contentBlock.id, `toolu_${Date.now()}_${index}`),
name: toString(contentBlock.name),
input: cloneLogPayload(contentBlock.input ?? {}),
inputJson: "",
});
} else {
blocks.set(index, {
type: "text",
index,
text: toString(contentBlock.text),
});
const eventType = toString(payload.type);
if (
payload.context_management &&
typeof payload.context_management === "object" &&
!Array.isArray(payload.context_management)
) {
contextManagement = asRecord(payload.context_management);
}
if (eventType === "message_start") {
const message = asRecord(payload.message);
messageId = toString(message.id, messageId || `msg_${Date.now()}`);
model = toString(message.model, model);
role = toString(message.role, role);
mergeUsage(usage, message.usage);
return;
}
continue;
}
if (eventType === "content_block_delta") {
const index = toNumber(payload.index, 0);
const delta = asRecord(payload.delta);
const deltaType = toString(delta.type);
const existing = blocks.get(index);
if (eventType === "content_block_start") {
const index = toNumber(payload.index, blocks.size);
const contentBlock = asRecord(payload.content_block);
const blockType = toString(contentBlock.type);
if (deltaType === "input_json_delta") {
const toolUse =
existing && existing.type === "tool_use"
if (blockType === "thinking") {
blocks.set(index, {
type: "thinking",
index,
thinking: toString(contentBlock.thinking),
signature:
typeof contentBlock.signature === "string" ? contentBlock.signature : undefined,
});
} else if (blockType === "tool_use") {
blocks.set(index, {
type: "tool_use",
index,
id: toString(contentBlock.id, `toolu_${Date.now()}_${index}`),
name: toString(contentBlock.name),
input: cloneLogPayload(contentBlock.input ?? {}),
inputJson: "",
});
} else {
blocks.set(index, {
type: "text",
index,
text: toString(contentBlock.text),
});
}
return;
}
if (eventType === "content_block_delta") {
const index = toNumber(payload.index, 0);
const delta = asRecord(payload.delta);
const deltaType = toString(delta.type);
const existing = blocks.get(index);
if (deltaType === "input_json_delta") {
const toolUse =
existing && existing.type === "tool_use"
? existing
: {
type: "tool_use" as const,
index,
id: `toolu_${Date.now()}_${index}`,
name: "",
input: {},
inputJson: "",
};
toolUse.inputJson += toString(delta.partial_json);
blocks.set(index, toolUse);
return;
}
if (deltaType === "thinking_delta" || typeof delta.thinking === "string") {
const thinking =
existing && existing.type === "thinking"
? existing
: { type: "thinking" as const, index, thinking: "", signature: undefined };
thinking.thinking += toString(delta.thinking);
blocks.set(index, thinking);
return;
}
const textBlock =
existing && existing.type === "text"
? existing
: {
type: "tool_use" as const,
type: "text" as const,
index,
id: `toolu_${Date.now()}_${index}`,
name: "",
input: {},
inputJson: "",
text: "",
};
toolUse.inputJson += toString(delta.partial_json);
blocks.set(index, toolUse);
continue;
textBlock.text += toString(delta.text);
blocks.set(index, textBlock);
return;
}
if (deltaType === "thinking_delta" || typeof delta.thinking === "string") {
const thinking =
existing && existing.type === "thinking"
? existing
: { type: "thinking" as const, index, thinking: "", signature: undefined };
thinking.thinking += toString(delta.thinking);
blocks.set(index, thinking);
continue;
if (eventType === "message_delta") {
const delta = asRecord(payload.delta);
stopReason = toString(delta.stop_reason, stopReason);
stopSequence =
typeof delta.stop_sequence === "string" ? String(delta.stop_sequence) : stopSequence;
mergeUsage(usage, payload.usage);
return;
}
const textBlock =
existing && existing.type === "text"
? existing
: {
type: "text" as const,
index,
text: "",
};
textBlock.text += toString(delta.text);
blocks.set(index, textBlock);
continue;
}
if (eventType === "message_delta") {
const delta = asRecord(payload.delta);
stopReason = toString(delta.stop_reason, stopReason);
stopSequence =
typeof delta.stop_sequence === "string" ? String(delta.stop_sequence) : stopSequence;
mergeUsage(usage, payload.usage);
continue;
}
},
mergeUsage(usage, payload.usage);
}
finalize(): unknown {
if (!sawAny) return null;
const content = [...blocks.values()]
.sort((a, b) => a.index - b.index)
.flatMap<ClaudeContentBlock>((block) => {
if (block.type === "text") {
return block.text
? [
{
type: "text",
text: block.text,
},
]
: [];
}
if (block.type === "thinking") {
return block.thinking
? [
{
type: "thinking",
thinking: block.thinking,
...(block.signature ? { signature: block.signature } : {}),
},
]
: [];
}
const content = [...blocks.values()]
.sort((a, b) => a.index - b.index)
.flatMap<ClaudeContentBlock>((block) => {
if (block.type === "text") {
return block.text
? [
{
type: "text",
text: block.text,
},
]
: [];
}
if (block.type === "thinking") {
return block.thinking
? [
{
type: "thinking",
thinking: block.thinking,
...(block.signature ? { signature: block.signature } : {}),
},
]
: [];
}
const parsedInput =
block.inputJson.trim().length > 0
? tryParseJson(block.inputJson)
: cloneLogPayload(block.input);
return [
{
type: "tool_use",
id: block.id,
name: block.name,
input: parsedInput,
},
];
});
const parsedInput =
block.inputJson.trim().length > 0
? tryParseJson(block.inputJson)
: cloneLogPayload(block.input);
return [
{
type: "tool_use",
id: block.id,
name: block.name,
input: parsedInput,
},
];
});
return {
id: messageId || `msg_${Date.now()}`,
type: "message",
role,
model,
content,
stop_reason: stopReason,
...(stopSequence ? { stop_sequence: stopSequence } : {}),
...(Object.keys(usage).length > 0 ? { usage } : {}),
...(contextManagement ? { context_management: contextManagement } : {}),
return {
id: messageId || `msg_${Date.now()}`,
type: "message",
role,
model,
content,
stop_reason: stopReason,
...(stopSequence ? { stop_sequence: stopSequence } : {}),
...(Object.keys(usage).length > 0 ? { usage } : {}),
...(contextManagement ? { context_management: contextManagement } : {}),
};
},
};
}
function buildGeminiSummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const payloads = events
.map((evt) => asRecord(evt.data))
.filter((payload) => Object.keys(payload).length);
if (payloads.length === 0) return null;
function createGeminiReducer(fallbackModel?: string | null): SummaryReducer {
let sawAny = false;
const parts: JsonRecord[] = [];
const usageMetadata: JsonRecord = {};
let modelVersion = fallbackModel || "gemini";
@@ -565,54 +594,110 @@ function buildGeminiSummary(events: StructuredSSEEvent[], fallbackModel?: string
parts.push(part);
};
for (const payload of payloads) {
if (typeof payload.modelVersion === "string" && payload.modelVersion.length > 0) {
modelVersion = payload.modelVersion;
}
mergeUsage(usageMetadata, payload.usageMetadata);
const candidate = asRecord(Array.isArray(payload.candidates) ? payload.candidates[0] : null);
if (typeof candidate.finishReason === "string" && candidate.finishReason.length > 0) {
finishReason = candidate.finishReason;
}
const content = asRecord(candidate.content);
if (typeof content.role === "string" && content.role.length > 0) {
role = content.role;
}
if (!Array.isArray(content.parts)) continue;
for (const item of content.parts) {
const part = asRecord(item);
if (part.functionCall && typeof part.functionCall === "object") {
parts.push({
functionCall: cloneLogPayload(part.functionCall),
});
} else if (typeof part.text === "string" && part.text.length > 0) {
appendPart({
text: part.text,
...(part.thought === true ? { thought: true } : {}),
});
}
}
}
return {
candidates: [
{
index: 0,
content: {
role,
parts,
},
finishReason,
},
],
...(Object.keys(usageMetadata).length > 0 ? { usageMetadata } : {}),
modelVersion,
ingest(payload: JsonRecord) {
if (Object.keys(payload).length === 0) return;
sawAny = true;
if (typeof payload.modelVersion === "string" && payload.modelVersion.length > 0) {
modelVersion = payload.modelVersion;
}
mergeUsage(usageMetadata, payload.usageMetadata);
const candidate = asRecord(Array.isArray(payload.candidates) ? payload.candidates[0] : null);
if (typeof candidate.finishReason === "string" && candidate.finishReason.length > 0) {
finishReason = candidate.finishReason;
}
const content = asRecord(candidate.content);
if (typeof content.role === "string" && content.role.length > 0) {
role = content.role;
}
if (!Array.isArray(content.parts)) return;
for (const item of content.parts) {
const part = asRecord(item);
if (part.functionCall && typeof part.functionCall === "object") {
parts.push({
functionCall: cloneLogPayload(part.functionCall),
});
} else if (typeof part.text === "string" && part.text.length > 0) {
appendPart({
text: part.text,
...(part.thought === true ? { thought: true } : {}),
});
}
}
},
finalize(): unknown {
if (!sawAny) return null;
return {
candidates: [
{
index: 0,
content: {
role,
parts,
},
finishReason,
},
],
...(Object.keys(usageMetadata).length > 0 ? { usageMetadata } : {}),
modelVersion,
};
},
};
}
function createSummaryReducer(
format: string | null | undefined,
fallbackModel?: string | null
): SummaryReducer | undefined {
const normalized = normalizeFormat(format);
if (!normalized) return undefined;
switch (normalized) {
case FORMATS.OPENAI_RESPONSES:
return createResponsesReducer(fallbackModel);
case FORMATS.CLAUDE:
return createClaudeReducer(fallbackModel);
case FORMATS.GEMINI:
case FORMATS.ANTIGRAVITY:
return createGeminiReducer(fallbackModel);
default:
return createOpenAIReducer(fallbackModel);
}
}
function buildOpenAISummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const reducer = createOpenAIReducer(fallbackModel);
for (const evt of events) reducer.ingest(asRecord(evt.data));
return reducer.finalize();
}
function buildResponsesSummary(
events: StructuredSSEEvent[],
fallbackModel?: string | null
): unknown {
const reducer = createResponsesReducer(fallbackModel);
for (const evt of events) reducer.ingest(asRecord(evt.data));
return reducer.finalize();
}
function buildClaudeSummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const reducer = createClaudeReducer(fallbackModel);
for (const evt of events) reducer.ingest(asRecord(evt.data));
return reducer.finalize();
}
function buildGeminiSummary(events: StructuredSSEEvent[], fallbackModel?: string | null): unknown {
const reducer = createGeminiReducer(fallbackModel);
for (const evt of events) reducer.ingest(asRecord(evt.data));
return reducer.finalize();
}
export function buildStreamSummaryFromEvents(
events: StructuredSSEEvent[],
fallbackFormat?: string | null,
@@ -666,19 +751,25 @@ export function compactStructuredStreamPayload(payload: unknown): unknown {
}
export function createStructuredSSECollector(options: CollectorOptions = {}) {
const { maxEvents = 200, maxBytes = 49152, stage } = options;
const { maxEvents = 200, maxBytes = 49152, stage, format, fallbackModel } = options;
const events: StructuredSSEEvent[] = [];
let usedBytes = 0;
let droppedEvents = 0;
// Live-updated on every push() regardless of the storage cap above — see
// the CollectorOptions.format doc comment for why (#9315).
const reducer = createSummaryReducer(format, fallbackModel);
return {
push(payload: unknown, explicitEvent?: string) {
if (payload === null || payload === undefined) return;
const clonedData = cloneLogPayload(payload);
reducer?.ingest(asRecord(clonedData));
const event: StructuredSSEEvent = {
index: events.length + droppedEvents,
timestamp: new Date().toISOString(),
data: cloneLogPayload(payload),
data: clonedData,
};
const eventName = explicitEvent || getEventName(payload);
@@ -700,6 +791,17 @@ export function createStructuredSSECollector(options: CollectorOptions = {}) {
return events.map((event) => cloneLogPayload(event));
},
// The reducer-computed summary, built incrementally from EVERY pushed
// payload (see CollectorOptions.format) — unlike
// buildStreamSummaryFromEvents(getEvents(), ...), this is correct even
// once the collector has truncated its retained event array. Returns
// undefined if no format was configured (e.g. the client-response
// collector, which builds its summary from independently-accumulated
// response state instead).
getSummary(): unknown {
return reducer?.finalize();
},
build(summary?: unknown, buildOptions: BuildOptions = {}) {
const { includeEvents = true } = buildOptions;
return {

100
tests/unit/stream-payload-collector.test.ts
View File

@@ -110,7 +110,9 @@ test("buildStreamSummaryFromEvents merges tool_call deltas when every chunk carr
],
}),
toolCallEvent({
tool_calls: [{ index: 0, id: "call_a", type: "function", function: { arguments: '{"x":1}' } }],
tool_calls: [
{ index: 0, id: "call_a", type: "function", function: { arguments: '{"x":1}' } },
],
}),
toolCallEvent({}, "tool_calls"),
];
@@ -215,3 +217,99 @@ test("buildStreamSummaryFromEvents keeps two genuinely different interleaved too
assert.equal(toolCalls[1].function.name, "Read");
assert.equal(toolCalls[1].function.arguments, '{"path":"b"}');
});
type OpenAIStreamSummary = {
choices: Array<{
finish_reason: string;
message: {
tool_calls?: Array<{ function: { name: string; arguments: string } }>;
reasoning_content?: string;
};
}>;
usage?: { total_tokens: number };
};
// #9315 — the dashboard's "Provider Response" panel went stale/incomplete for
// long streamed responses because it was reconstructed from
// buildStreamSummaryFromEvents(collector.getEvents(), ...) — and getEvents()
// only returns whatever survived the collector's maxEvents/maxBytes cap. Once
// a stream exceeded that cap, every chunk after the cutoff (final
// finish_reason, tool_calls, rest of reasoning_content, usage) was silently
// dropped from the reconstruction, even though the client actually received
// the complete, correct response.
test("#9315: collector.getSummary() reflects the full stream even after maxEvents truncation", () => {
const c = collector.createStructuredSSECollector({
maxEvents: 3,
format: "openai",
fallbackModel: "test-model",
});
// First 3 chunks fill the cap.
c.push({
id: "chatcmpl-1",
object: "chat.completion.chunk",
created: 1,
model: "test-model",
choices: [{ index: 0, delta: { role: "assistant", content: "Thinking" } }],
});
c.push({ choices: [{ index: 0, delta: { content: " about it" } }] });
c.push({ choices: [{ index: 0, delta: { reasoning_content: "step one. " } }] });
// These all arrive AFTER the cap is full — the OLD reconstruction-from-
// getEvents() approach silently loses every one of them.
c.push({ choices: [{ index: 0, delta: { reasoning_content: "step two." } }] });
c.push({
choices: [
{
index: 0,
delta: {
tool_calls: [
{
index: 0,
id: "call_1",
type: "function",
function: { name: "Bash", arguments: '{"cmd":"date"}' },
},
],
},
},
],
});
c.push({ choices: [{ index: 0, delta: {}, finish_reason: "tool_calls" }] });
c.push({
choices: [{ index: 0, delta: {} }],
usage: { prompt_tokens: 10, completion_tokens: 20, total_tokens: 30 },
});
// Sanity check: this test is only meaningful if truncation genuinely happened.
const retained = c.getEvents();
assert.equal(retained.length, 3, "expected the raw event array to be capped at maxEvents");
// Characterize the pre-fix bug: reconstructing from the truncated retained
// events (the old approach every call site in stream.ts used) misses
// everything that arrived after the cap.
const staleSummary = collector.buildStreamSummaryFromEvents(
retained,
"openai",
"test-model"
) as OpenAIStreamSummary;
assert.equal(staleSummary.choices[0].finish_reason, "stop");
assert.equal(staleSummary.choices[0].message.tool_calls, undefined);
assert.equal(staleSummary.choices[0].message.reasoning_content, "step one.");
// The fix: getSummary() was fed every pushed chunk, truncated from storage
// or not, so it reflects the true final state.
const liveSummary = c.getSummary() as OpenAIStreamSummary;
assert.equal(liveSummary.choices[0].finish_reason, "tool_calls");
assert.equal(liveSummary.choices[0].message.tool_calls.length, 1);
assert.equal(liveSummary.choices[0].message.tool_calls[0].function.name, "Bash");
assert.equal(liveSummary.choices[0].message.tool_calls[0].function.arguments, '{"cmd":"date"}');
assert.equal(liveSummary.choices[0].message.reasoning_content, "step one. step two.");
assert.equal(liveSummary.usage.total_tokens, 30);
});
test("#9315: getSummary() returns undefined when no format was configured (unaffected client-response collector)", () => {
const c = collector.createStructuredSSECollector({ maxEvents: 200 });
c.push({ choices: [{ index: 0, delta: { content: "hi" } }] });
assert.equal(c.getSummary(), undefined);
});

57
tests/unit/stream-utils.test.ts
View File

@@ -1098,6 +1098,63 @@ test("createSSEStream passthrough preserves Responses API events and completion
assert.equal(onCompletePayload.providerPayload.summary.object, "response");
});
// Real bug found live (dashboard log id 1786032832181-1c6275, #9315 follow-up):
// providerPayloadCollector was keyed on `sourceFormat` (the CLIENT's format)
// instead of `targetFormat` (the PROVIDER's format — see createSSEStream's own
// @param doc). A Responses-API client routed to a plain-OpenAI-chat-completions
// upstream (exactly this OpenClaw/opencode-zen combo) fed the provider's real
// chat.completion.chunk deltas into the Responses-API reducer, which never
// recognizes them — so the dashboard's "Provider Response" panel stayed stuck
// empty (`output: []`) forever while "Client Response" correctly showed full
// content, reading as if the two panels disagreed about the same request.
test("createSSEStream translate mode: providerPayload summary reflects the PROVIDER's format, not the client's", async () => {
let onCompletePayload = null;
await readTransformed(
[
`data: ${JSON.stringify({
id: "chatcmpl-1",
object: "chat.completion.chunk",
created: 1,
model: "big-pickle",
choices: [
{ index: 0, delta: { role: "assistant", content: "Hello " }, finish_reason: null },
],
})}\n\n`,
`data: ${JSON.stringify({
id: "chatcmpl-1",
object: "chat.completion.chunk",
created: 1,
model: "big-pickle",
choices: [{ index: 0, delta: { content: "world" }, finish_reason: "stop" }],
usage: { prompt_tokens: 5, completion_tokens: 2, total_tokens: 7 },
})}\n\n`,
`data: [DONE]\n\n`,
],
{
mode: "translate",
// Client speaks Responses API; the upstream provider (opencode-zen-style)
// speaks plain OpenAI chat-completions — exactly the OpenClaw combo that
// surfaced this live.
sourceFormat: FORMATS.OPENAI_RESPONSES,
targetFormat: FORMATS.OPENAI,
provider: "opencode-zen",
model: "big-pickle",
body: { input: "hi" },
onComplete(payload) {
onCompletePayload = payload;
},
}
);
const summary = onCompletePayload.providerPayload.summary;
assert.ok(summary, "providerPayload.summary must not be null/undefined");
// The bug's exact symptom: a Responses-API reducer fed chat-completion chunks
// never recognizes them, so it stays at "no output" — assert the OPPOSITE.
assert.equal(summary.object, "chat.completion");
assert.equal(summary.choices?.[0]?.message?.content, "Hello world");
assert.equal(summary.choices?.[0]?.finish_reason, "stop");
});
test("createSSEStream passthrough drops leaked empty chat bootstrap chunks for Responses clients", async () => {
const text = await readTransformed(
[