From 4bda22583eab6fcddc09e0b611838bfe8693acfe Mon Sep 17 00:00:00 2001 From: Markus Hartung Date: Thu, 13 Aug 2026 09:02:30 +0200 Subject: [PATCH] fix(sse): provider-response summary format bugs (dashboard Provider Response panel) (#10037) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 * 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 --- open-sse/utils/stream.ts | 49 +- open-sse/utils/streamPayloadCollector.ts | 830 +++++++++++--------- tests/unit/stream-payload-collector.test.ts | 100 ++- tests/unit/stream-utils.test.ts | 57 ++ 4 files changed, 661 insertions(+), 375 deletions(-) diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 121b3950e6..0eeccec17d 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -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, { diff --git a/open-sse/utils/streamPayloadCollector.ts b/open-sse/utils/streamPayloadCollector.ts index 63cb901179..9ca089dfe4 100644 --- a/open-sse/utils/streamPayloadCollector.ts +++ b/open-sse/utils/streamPayloadCollector.ts @@ -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; +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((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((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 { diff --git a/tests/unit/stream-payload-collector.test.ts b/tests/unit/stream-payload-collector.test.ts index 669edcdd1e..610d0861a0 100644 --- a/tests/unit/stream-payload-collector.test.ts +++ b/tests/unit/stream-payload-collector.test.ts @@ -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); +}); diff --git a/tests/unit/stream-utils.test.ts b/tests/unit/stream-utils.test.ts index e4a0bcdb5e..380099b1b6 100644 --- a/tests/unit/stream-utils.test.ts +++ b/tests/unit/stream-utils.test.ts @@ -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( [