// Renders "separated"-mode reasoning (streamed on its own `reasoning_chunk` SSE // channel) to stderr, so --think matches inline (see think-filter.ts). const THINK_OPEN = '' const THINK_CLOSE = '' export type ReasoningChunk = { reasoning: string nodeId: string isFinal: boolean } // reasoning_chunk nests its payload under `data` (not top-level like `message`). export function parseReasoningChunk(parsed: Record): ReasoningChunk | undefined { const data = parsed.data if (data === null || typeof data !== 'object' || Array.isArray(data)) return undefined const rec = data as Record return { reasoning: typeof rec.reasoning === 'string' ? rec.reasoning : '', nodeId: typeof rec.node_id === 'string' ? rec.node_id : '', isFinal: rec.is_final === true, } } // Bucket key for a chunk; falls back to a single bucket so live rendering and // buffered collection key reasoning the same way. export function reasoningKey(chunk: ReasoningChunk): string { return chunk.nodeId !== '' ? chunk.nodeId : '_' } // Appends a reasoning delta to a per-node accumulator. export function accumulateReasoning(acc: Record, chunk: ReasoningChunk): void { if (chunk.reasoning === '') return const key = reasoningKey(chunk) acc[key] = (acc[key] ?? '') + chunk.reasoning } // Frames a live reasoning stream into stderr: on the first delta, // raw deltas thereafter, on is_final. Parallel branches can interleave // chunks from different nodes on one stream, so it keeps at most one block open, // switches blocks on node change, and tags each block with its node id so the // interleaved fragments stay distinguishable. export class ReasoningChunkRenderer { private openNode: string | undefined push(chunk: ReasoningChunk, errOut: NodeJS.WritableStream): void { const key = reasoningKey(chunk) if (chunk.reasoning !== '') { if (this.openNode !== key) { this.closeActive(errOut) errOut.write(chunk.nodeId !== '' ? `${THINK_OPEN} [${chunk.nodeId}]\n` : `${THINK_OPEN}\n`) this.openNode = key } errOut.write(chunk.reasoning) } if (chunk.isFinal && this.openNode === key) this.closeActive(errOut) } // Close a block left open by a truncated stream. flush(errOut: NodeJS.WritableStream): void { this.closeActive(errOut) } private closeActive(errOut: NodeJS.WritableStream): void { if (this.openNode === undefined) return errOut.write(`${THINK_CLOSE}\n`) this.openNode = undefined } } // Frames fully-buffered reasoning (one entry per LLM node id) into blocks. export function formatReasoningBlocks(reasoning: Record): string { const blocks: string[] = [] for (const text of Object.values(reasoning)) { const trimmed = text.trim() if (trimmed !== '') blocks.push(`${THINK_OPEN}\n${trimmed}\n${THINK_CLOSE}`) } return blocks.join('\n---\n') } // Frames per-node reasoning from a message_end `metadata` object; '' when absent. export function reasoningBlocksFromMetadata(metadata: unknown): string { if (metadata === null || typeof metadata !== 'object' || Array.isArray(metadata)) return '' const reasoning = (metadata as Record).reasoning if (reasoning === null || typeof reasoning !== 'object' || Array.isArray(reasoning)) return '' const map: Record = {} for (const [key, value] of Object.entries(reasoning as Record)) { if (typeof value === 'string') map[key] = value } return formatReasoningBlocks(map) }