weichen
e10e447dab
Why: Status words shared with a final reply could suppress a child completion record. Evidence: 272 frontend unit tests, build, and 15 isolated browser scenarios. Impact: Tag client-side collaboration replay so partial-delta deduplication preserves agent notes. Refs: Agent-System-Dev feature/work-companion-visibility UI1
316 lines
13 KiB
TypeScript
316 lines
13 KiB
TypeScript
// 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 }
|
||
| {
|
||
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<LiveItem, { kind: 'tool' }>
|
||
|
||
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 }]
|
||
}
|
||
|
||
// 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<number, boolean>()
|
||
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 })
|
||
}
|
||
}
|
||
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): 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<string, unknown> | 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}」已开始` })
|
||
break
|
||
case 'agent.progress':
|
||
if (typeof p?.note === 'string' && p.note.trim()) items.push({ kind: 'agent', text: `协作进展：${p.note}` })
|
||
break
|
||
case 'agent.result':
|
||
items.push({ kind: 'agent', text: `协作结束：${typeof p?.summary === 'string' && p.summary.trim() ? p.summary : String(p?.status ?? '已结束')}` })
|
||
break
|
||
case 'thinking.delta':
|
||
if (typeof p?.text === 'string') items = applyThinkingDelta(items, p.text, now)
|
||
break
|
||
case 'message.delta':
|
||
if (!finalizing && 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 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)
|
||
}
|