// live 流式投影纯逻辑(中栏过程/回复分离):事件按到达序并入 items——thinking/text // 增量各并入其末段,工具按当前位置落位。SSE 有序且后端每轮按 思考→正文→工具 追加落库 // 部件,事件到达序即落库序,故合成消息与落库部件同构,终态对账不重排。 // 文本段是否为过程(process)由「本段之后、下一个思考/文本段之前是否出现工具」判定。 // 参数增量(编写中)一到即算工具,解说不用等工具真正开始执行才离开回复区。 import type { Message, Part, StreamEvent } from './contracts' import type { LiveState } from './store' import { splitTimeline, type ReplayPart } from './messageparts' export type ToolLiveState = 'composing' | 'running' | 'done' // 到达序部件流的元素。 export type LiveItem = | { kind: 'thinking'; text: string; startedAt: number; endedAt: number } | { kind: 'text'; index: number; text: string } | { kind: 'agent'; text: string; agentRun?: string; agentPhase?: 'spawn' | 'progress' | 'result' } | { kind: 'tool' callId?: string index?: number name: string args: string state: ToolLiveState startedAt: number endedAt: number output: string status?: string summary?: string } | { kind: 'artifact'; artifact: { artifactId: string; name: string; kind?: string } } | { kind: 'check'; passed: boolean; reason: string } | { kind: 'final' } type LiveTool = Extract const OUTPUT_TAIL = 4096 function isTool(it: LiveItem): it is LiveTool { return it.kind === 'tool' } // applyThinkingDelta 并入思考增量:末段是思考则追加,否则开启新思考段(新轮开始)。 // now 供测试固定首尾时间;缺省用墙上时钟,durationMs = endedAt - startedAt。 export function applyThinkingDelta(items: LiveItem[], delta: string, now = Date.now()): LiveItem[] { const last = items[items.length - 1] if (last && last.kind === 'thinking') { const out = items.slice() out[out.length - 1] = { kind: 'thinking', text: last.text + delta, startedAt: last.startedAt, endedAt: now } return out } return [...items, { kind: 'thinking', text: delta, startedAt: now, endedAt: now }] } // applyTextDelta 并入正文增量:末段是同 index 的文本段则追加,否则开启新文本段。 export function applyTextDelta(items: LiveItem[], index: number, delta: string): LiveItem[] { const last = items[items.length - 1] if (last && last.kind === 'text' && last.index === index) { const out = items.slice() out[out.length - 1] = { kind: 'text', index, text: last.text + delta } return out } return [...items, { kind: 'text', index, text: delta }] } // applyToolCallDelta 并入正在编写的参数。同 index(或同 callId)的编写中条目追加,否则新开一条。 export function applyToolCallDelta(items: LiveItem[], index: number, delta: string, callId?: string, name?: string, now = Date.now()): LiveItem[] { for (let i = items.length - 1; i >= 0; i--) { const it = items[i] if (!isTool(it) || it.state !== 'composing') continue if (it.index !== index && !(callId && it.callId === callId)) continue const out = items.slice() out[i] = { ...it, args: it.args + delta, callId: callId || it.callId, name: name || it.name, endedAt: now, } return out } return [...items, { kind: 'tool', index, callId, name: name || '工具', args: delta, state: 'composing', startedAt: now, endedAt: now, output: '', }] } // applyToolCall 把一条调用标成运行中。优先按 callId 接上编写中的条目,否则接第一条编写中的条目。 export function applyToolCall(items: LiveItem[], name: string, callId?: string, args?: string, now = 0): LiveItem[] { let idx = -1 if (callId) { for (let i = 0; i < items.length; i++) { const it = items[i] if (isTool(it) && it.state === 'composing' && it.callId === callId) { idx = i; break } } } if (idx < 0) { for (let i = 0; i < items.length; i++) { const it = items[i] if (isTool(it) && it.state === 'composing') { idx = i; break } } } if (idx >= 0) { const it = items[idx] as LiveTool const out = items.slice() out[idx] = { ...it, name: name || it.name, callId: callId || it.callId, args: args || it.args, state: 'running', startedAt: now || it.startedAt, endedAt: now, } return out } return [...items, { kind: 'tool', callId, name, args: args ?? '', state: 'running', startedAt: now, endedAt: now, output: '', }] } // applyToolOutput 把执行中的输出追加到对应工具,只保留尾部,避免长日志撑住内存。 export function applyToolOutput(items: LiveItem[], callId: string | undefined, text: string, maxChars = OUTPUT_TAIL): LiveItem[] { for (let i = items.length - 1; i >= 0; i--) { const it = items[i] if (!isTool(it) || it.state === 'done') continue if (callId && it.callId && it.callId !== callId) continue let output = it.output + text if (output.length > maxChars) output = output.slice(output.length - maxChars) const out = items.slice() out[i] = { ...it, output } return out } return items } // applyToolResult 回填对应工具(优先按 callId 配对,回落到最近一个未决工具)。 export function applyToolResult(items: LiveItem[], callId: string | undefined, status: string, summary?: string, now = 0): LiveItem[] { for (let i = items.length - 1; i >= 0; i--) { const it = items[i] if (!isTool(it) || it.state === 'done') continue if (callId && it.callId && it.callId !== callId) continue const out = items.slice() out[i] = { ...it, callId: it.callId, name: it.name, status, summary, state: 'done', endedAt: now || it.endedAt } return out } return items } export function applyCheck(items: LiveItem[], passed: boolean, reason: string): LiveItem[] { return [...items, { kind: 'check', passed, reason }] } /** A child streams cumulative previews. Replace that child's latest preview in * place, while preserving real phase changes and the order of other children. */ function applyAgentProgress(items: LiveItem[], agentRun: string, note: string): LiveItem[] { const text = note.trim() if (!text) return items for (let i = items.length - 1; i >= 0; i--) { const previous = items[i] if (previous.kind !== 'agent' || previous.agentRun !== agentRun) continue if (previous.agentPhase !== 'progress') break const value = previous.text.slice('协作进展:'.length) if (value === text || value.startsWith(text)) return items if (text.startsWith(value) && !/^调用\s/.test(text)) { const next = items.slice() next[i] = { ...previous, text: `协作进展:${text}` } return next } break } return [...items, { kind: 'agent', text: `协作进展:${text}`, agentRun, agentPhase: 'progress' }] } // applyFinalizing 标记收尾开始:此前的正文都是过程,此后的正文才是给用户的回复。 export function applyFinalizing(items: LiveItem[]): LiveItem[] { if (items.some(it => it.kind === 'final')) return items return [...items, { kind: 'final' }] } export function applyArtifactCreated(items: LiveItem[], artifact: { artifactId: string; name: string; kind?: string }): LiveItem[] { return [...items, { kind: 'artifact', artifact }] } export function applyArtifactUpdated(items: LiveItem[], artifact: { artifactId: string; name: string; kind?: string }): LiveItem[] { const out = items.slice() for (let i = out.length - 1; i >= 0; i--) { const it = out[i] if (it.kind === 'artifact' && it.artifact.artifactId === artifact.artifactId) { out[i] = { kind: 'artifact', artifact: { ...it.artifact, ...artifact } } return out } } return [...out, { kind: 'artifact', artifact }] } // liveToMessage:run 进行中没有已落库的助手消息,把 LiveState 合成同构临时消息。 export function liveToMessage(live: LiveState): Message { const finalAt = live.items.findIndex(it => it.kind === 'final') const textProcess = new Map() for (let i = 0; i < live.items.length; i++) { if (live.items[i].kind !== 'text') continue if (finalAt >= 0) { textProcess.set(i, i < finalAt) continue } let hasCalls = false for (let j = i + 1; j < live.items.length && live.items[j].kind === 'tool'; j++) hasCalls = true textProcess.set(i, hasCalls) } const parts: Part[] = [] for (let i = 0; i < live.items.length; i++) { const it = live.items[i] if (it.kind === 'final' || it.kind === 'artifact') continue if (it.kind === 'check') { parts.push({ type: 'check', process: true, check: { passed: it.passed, reason: it.reason || undefined } }) continue } if (it.kind === 'agent') { parts.push({ type: 'text', text: it.text, process: true, replaySource: 'agent' } as ReplayPart) } else if (it.kind === 'thinking') { const durationMs = it.endedAt - it.startedAt if (it.text) parts.push({ type: 'thinking', text: it.text, process: true, ...(durationMs > 0 ? { durationMs } : {}), }) } else if (it.kind === 'tool') { const durationMs = it.state === 'done' ? it.endedAt - it.startedAt : 0 const call = { id: it.callId ?? '', name: it.name, ...(it.args ? { arguments: it.args } : {}), ...(durationMs > 0 ? { durationMs } : {}), } parts.push({ type: 'tool_call', process: true, toolCall: call, toolPhase: it.state, ...(it.output ? { toolOutput: it.output } : {}), } as Part) if (it.state === 'done') { parts.push({ type: 'tool_result', process: true, toolCall: { id: call.id, name: call.name }, toolResult: { callId: call.id, status: it.status === 'error' ? 'error' : 'ok', summary: it.summary }, }) } } else { parts.push({ type: 'text', text: it.text, process: textProcess.get(i) || undefined }) } } const createdAt = live.startedAt > 0 && Number.isFinite(live.startedAt) ? new Date(live.startedAt).toISOString() : '' return { id: `live:${live.runId}`, conversationId: '', role: 'assistant', runId: live.runId, parts, createdAt } } /** Rebuild one run's process from its durable ordered events, without the live preview's output-tail limit. */ export function runEventsToTimeline(events: StreamEvent[], parentRunId = events[0]?.runId, completeRecord = false): ReplayPart[] { let items: LiveItem[] = [] let finalizing = false for (const event of [...events].sort((a, b) => a.seq - b.seq)) { const p = event.payload as Record | undefined const now = Date.parse(event.ts) || 0 if (event.runId !== parentRunId && event.type !== 'agent.progress' && event.type !== 'agent.result') continue switch (event.type) { case 'agent.spawn': if (typeof p?.name === 'string') items.push({ kind: 'agent', text: `协作代理「${p.name}」已开始`, agentRun: String(p.agentRun ?? event.runId ?? ''), agentPhase: 'spawn' }) break case 'agent.progress': if (typeof p?.note === 'string') items = applyAgentProgress(items, String(p.agentRun ?? event.runId ?? ''), p.note) break case 'agent.result': items.push({ kind: 'agent', text: `协作结束:${typeof p?.summary === 'string' && p.summary.trim() ? p.summary : String(p?.status ?? '已结束')}`, agentRun: String(p?.agentRun ?? event.runId ?? ''), agentPhase: 'result' }) break case 'thinking.delta': if (typeof p?.text === 'string') items = applyThinkingDelta(items, p.text, now) break case 'message.delta': if ((!finalizing || completeRecord) && typeof p?.text === 'string') items = applyTextDelta(items, typeof p.index === 'number' ? p.index : 0, p.text) break case 'tool.call.delta': if (typeof p?.delta === 'string') items = applyToolCallDelta(items, typeof p.index === 'number' ? p.index : 0, p.delta, typeof p.callId === 'string' ? p.callId : undefined, typeof p.name === 'string' ? p.name : undefined, now) break case 'tool.call': items = applyToolCall(items, typeof p?.name === 'string' ? p.name : '工具', typeof p?.callId === 'string' ? p.callId : undefined, typeof p?.arguments === 'string' ? p.arguments : undefined, now) break case 'tool.output': if (typeof p?.text === 'string') items = applyToolOutput(items, typeof p.callId === 'string' ? p.callId : undefined, p.text, Number.POSITIVE_INFINITY) break case 'tool.result': items = applyToolResult(items, typeof p?.callId === 'string' ? p.callId : undefined, typeof p?.status === 'string' ? p.status : 'ok', typeof p?.summary === 'string' ? p.summary : undefined, now) break case 'run.phase': if (p?.phase === 'finalizing') { finalizing = true; items = applyFinalizing(items) } break case 'run.check': items = applyCheck(items, !!p?.passed, typeof p?.reason === 'string' ? p.reason : '') break } } const live: LiveState = { runId: events[0]?.runId ?? '', phase: 'done', items, agentNotes: {}, startedAt: 0 } return completeRecord ? liveToMessage(live).parts : splitTimeline(liveToMessage(live).parts).timeline } // livePhaseLabel 用最后一个实时条目给出中文阶段,而不是后端的英文 phase。 export function livePhaseLabel(live: LiveState, now = Date.now()): string { if (live.phase === 'finalizing') return '整理回复中' let last: LiveItem | undefined for (let i = live.items.length - 1; i >= 0; i--) { const kind = live.items[i].kind if (kind !== 'artifact' && kind !== 'check' && kind !== 'final') { last = live.items[i]; break } } if (!last || last.kind === 'thinking') return '思考中' if (last.kind === 'tool') { if (last.state === 'composing') return `正在编写 ${last.name}` if (last.state === 'running') { const s = Math.max(0, Math.floor((now - last.startedAt) / 1000)) const m = Math.floor(s / 60) const clock = m > 0 ? `${m} 分 ${s % 60} 秒` : `${s} 秒` return `正在运行 ${last.name} · ${clock}` } return '回复中' } return '回复中' } // 默认阅读层只显示阶段;运行中也把技术记录留给用户主动展开。 export function showProcessInline(_live: LiveState | null, autoCollapse: boolean): boolean { return !autoCollapse } // hasSettledReply:助手回复落库后,这个 run 的实时气泡才该收起。 // 用户消息也会带同一个 runId,不能拿它判断回复已经落库。 export function hasSettledReply(messages: { role: string; runId?: string }[], runId: string): boolean { return messages.some(m => m.role === 'assistant' && m.runId === runId) }