Merge pull request 'Show a run as active as soon as the send returns' (#38) from fix/live-run-visibility into main

This commit was merged in pull request #38.
This commit is contained in:
2026-10-03 15:25:37 +00:00
7 changed files with 117 additions and 29 deletions
+1 -1
View File
@@ -46,7 +46,7 @@ export const api = {
setConversationProfile: (id: string, profileId: string) => auth.fetch(`${BASE}/api/conversations/${id}`, { method: 'PATCH', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ profileId }) }).then(r => json<Conversation>(r)),
conversationDetail: (id: string) => auth.fetch(`${BASE}/api/conversations/${id}`).then(r =>
json<{ conversation: Conversation; messages: Message[]; runs: Run[]; activeHitl: HITLCard[]; lastSeq: number }>(r)),
json<{ conversation: Conversation; messages: Message[]; runs: Run[]; activeHitl: HITLCard[]; lastSeq: number; resumeSeq?: number }>(r)),
sendMessage: (convId: string, text: string, attachmentIds: string[], requestId?: string) =>
auth.fetch(`${BASE}/api/conversations/${convId}/messages`, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ text, attachmentIds, requestId }) })
+4 -2
View File
@@ -19,10 +19,12 @@ export function AgentTracker() {
// 中间过程:该会话消息树中归属子代理 run 的消息(run_id 过滤)。
const subMessages: Record<string, Message[]> = {}
for (const m of messagesAll) {
if (m.runId && m.runId !== runId) (subMessages[m.runId] ??= []).push(m)
if (m.role === 'assistant' && m.runId && m.runId !== runId) (subMessages[m.runId] ??= []).push(m)
}
const focus = selected ?? children[0]?.id ?? runId
const focusMsgs = focus === runId ? messagesAll.filter(m => m.runId === runId) : subMessages[focus] ?? []
const focusMsgs = focus === runId
? messagesAll.filter(m => m.role === 'assistant' && m.runId === runId)
: subMessages[focus] ?? []
return (
// 阶段③ 收敛:只保留 OverlayHost 一层壳(此前在其内又嵌 .modal 双层窗框)。
+2 -2
View File
@@ -7,7 +7,7 @@ import { marked } from 'marked'
import { api } from '../api'
import { store, useStore, activeRun, type LiveState, type ContextObs } from '../store'
import { scrollRatio, scrollTopForRatio } from '../scrollsync'
import { liveToMessage, livePhaseLabel, showProcessInline } from '../live'
import { hasSettledReply, liveToMessage, livePhaseLabel, showProcessInline } from '../live'
import type { Message, Part, HITLCard, ActivityCard, Artifact } from '../contracts'
import { UIRenderer, InlineConfirmCard, hasOpenConfirm, CardAnchor } from '@ui'
import { docBridge } from './Workbench'
@@ -541,7 +541,7 @@ export function Conversation() {
if (m.role === 'system') return <NoticeMessage key={m.id} msg={m} />
return <AssistantMessage key={m.id} msg={m} live={live?.runId === m.runId ? live : null} />
})}
{live && !messages.some(m => m.runId === live.runId) &&
{live && !hasSettledReply(messages, live.runId) &&
<AssistantMessage key={`live-${live.runId}`} msg={liveToMessage(live)} live={live} />}
{activeHitl.map(c => <HITLCardView key={c.id} card={c} />)}
{messages.length === 0 && <div className="empty-state">
+6
View File
@@ -250,3 +250,9 @@ export function livePhaseLabel(live: LiveState, now = Date.now()): string {
export function showProcessInline(live: LiveState | null, autoCollapse: boolean): boolean {
return live != null || !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)
}
+52 -23
View File
@@ -305,16 +305,18 @@ export const store = {
if (sse) { sse.close(); sse = null }
set({ currentConvId: convId, messages: [], runs: [], activeHitl: [], lastSeq: 0, live: null, plan: null, todos: null, cards: [], inboxScope: 'conv', outbox: [], outboxDraining: false })
syncPreviewRail((state.previewTabs[convId] ?? []).length > 0)
await store.reconcile()
await store.connectLive()
const resumeSeq = await store.reconcile()
store.connectLive(resumeSeq ?? state.lastSeq)
},
// 终态对账:以 Query 的权威消息树/运行树/游标为准收敛。
async reconcile() {
// 返回值是实时流应从这个 seq 之后重放;有进行中的 run 时它早于 lastSeq。
async reconcile(): Promise<number | undefined> {
if (!state.currentConvId) return
try {
const d = await api.conversationDetail(state.currentConvId)
const liveRunActive = state.live && !isTerminal(d.runs.find(r => r.id === state.live!.runId)?.status)
const resumeSeq = d.resumeSeq ?? d.lastSeq
set({
messages: d.messages,
runs: d.runs,
@@ -327,15 +329,16 @@ export const store = {
set({ plan, todos })
store.refreshCards(d.conversation.id)
store.refreshTasks()
return resumeSeq
} catch (e) {
set({ error: String(e) })
}
},
connectLive() {
connectLive(since?: number) {
if (!state.currentConvId) return
const convId = state.currentConvId
sse = connectEvents(convId, state.lastSeq, {
sse = connectEvents(convId, since ?? state.lastSeq, {
onOpen: () => set({ connected: true }),
onError: () => set({ connected: false }),
onEvent: e => store.applyEvent(e),
@@ -343,40 +346,42 @@ export const store = {
},
applyEvent(e: StreamEvent) {
const now = eventTime(e)
const project = (mutate: (l: LiveState) => void) => touchLive(e.runId ?? '', mutate, now)
switch (e.type) {
case 'message.delta': {
const p = e.payload as { index: number; text: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyTextDelta(l.items, p.index, p.text) }) })
set({ live: project(l => { l.items = applyTextDelta(l.items, p.index, p.text) }) })
break
}
case 'thinking.delta': {
const p = e.payload as { index: number; text: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyThinkingDelta(l.items, p.text) }) })
set({ live: project(l => { l.items = applyThinkingDelta(l.items, p.text, now) }) })
break
}
case 'tool.call.delta': {
const p = e.payload as { index: number; callId?: string; name?: string; delta: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyToolCallDelta(l.items, p.index, p.delta, p.callId, p.name) }) })
set({ live: project(l => { l.items = applyToolCallDelta(l.items, p.index, p.delta, p.callId, p.name, now) }) })
break
}
case 'tool.call': {
const p = e.payload as { callId: string; name: string; arguments?: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyToolCall(l.items, p.name, p.callId, p.arguments, Date.now()) }) })
set({ live: project(l => { l.items = applyToolCall(l.items, p.name, p.callId, p.arguments, now) }) })
break
}
case 'tool.output': {
const p = e.payload as { callId?: string; text: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyToolOutput(l.items, p.callId, p.text) }) })
set({ live: project(l => { l.items = applyToolOutput(l.items, p.callId, p.text) }) })
break
}
case 'tool.result': {
const p = e.payload as { callId: string; status: string; summary?: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyToolResult(l.items, p.callId, p.status, p.summary, Date.now()) }) })
set({ live: project(l => { l.items = applyToolResult(l.items, p.callId, p.status, p.summary, now) }) })
break
}
case 'run.phase': {
const p = e.payload as { phase: string }
set({ live: touchLive(e.runId ?? '', l => {
set({ live: project(l => {
l.phase = p.phase
if (p.phase === 'finalizing') l.items = applyFinalizing(l.items)
}) })
@@ -384,7 +389,7 @@ export const store = {
}
case 'run.check': {
const p = e.payload as { passed?: boolean; reason?: string }
set({ live: touchLive(e.runId ?? '', l => { l.items = applyCheck(l.items, !!p.passed, p.reason ?? '') }) })
set({ live: project(l => { l.items = applyCheck(l.items, !!p.passed, p.reason ?? '') }) })
break
}
case 'hitl.request': {
@@ -397,17 +402,17 @@ export const store = {
}
case 'agent.spawn': {
const p = e.payload as { agentRun: string; name: string; task: string }
set({ live: touchLive(e.runId ?? '', l => { l.agentNotes = { ...l.agentNotes, [p.agentRun]: `已派生 ${p.name}:${p.task}` } }) })
set({ live: project(l => { l.agentNotes = { ...l.agentNotes, [p.agentRun]: `已派生 ${p.name}:${p.task}` } }) })
break
}
case 'agent.progress': {
const p = e.payload as { agentRun: string; note?: string }
set({ live: touchLive(e.runId ?? '', l => { if (p.note) l.agentNotes = { ...l.agentNotes, [p.agentRun]: p.note! } }) })
set({ live: project(l => { if (p.note) l.agentNotes = { ...l.agentNotes, [p.agentRun]: p.note! } }) })
break
}
case 'agent.result': {
const p = e.payload as { agentRun: string; summary?: string }
set({ live: touchLive(e.runId ?? '', l => { l.agentNotes = { ...l.agentNotes, [p.agentRun]: p.summary ?? '完成' } }) })
set({ live: project(l => { l.agentNotes = { ...l.agentNotes, [p.agentRun]: p.summary ?? '完成' } }) })
store.reconcile()
break
}
@@ -444,7 +449,7 @@ export const store = {
const p = e.payload as { artifact?: { id: string; name: string; kind?: string } }
if (p?.artifact) {
const ref = { artifactId: p.artifact.id, name: p.artifact.name, kind: p.artifact.kind }
set({ live: touchLive(e.runId ?? '', l => {
set({ live: project(l => {
l.items = e.type === 'artifact.updated' ? applyArtifactUpdated(l.items, ref) : applyArtifactCreated(l.items, ref)
}) })
}
@@ -481,12 +486,23 @@ export const store = {
return 'ok'
}
if (out.run) {
const started = out.run
// 立刻把 run 合进列表,停止/排队不必等终态对账。已有同 id 记录保持不动。
const runs = mergeStartedRun(state.runs, started)
const messages = state.messages.some(m => m.id === out.message.id)
? state.messages
: [...state.messages, out.message]
if (isTerminal(runs.find(r => r.id === started.id)?.status)) {
set({ runs, messages })
await store.reconcile()
return 'ok'
}
// 竞态保护:POST 返回前 SSE 事件可能已先行到达并累积(同 run),
// 此时保留已累积的 live,只补挂用户消息;否则初始化空 live。
if (!state.live || state.live.runId !== out.run.id) {
set({ live: { runId: out.run.id, phase: 'thinking', items: [], agentNotes: {}, startedAt: Date.now() } })
}
set({ messages: [...state.messages, out.message] })
const live = !state.live || state.live.runId !== started.id
? { runId: started.id, phase: 'thinking', items: [], agentNotes: {}, startedAt: Date.now() }
: state.live
set({ live, runs, messages })
store.connectIfNeeded()
}
return 'ok'
@@ -780,8 +796,8 @@ export const store = {
},
}
function touchLive(_runId: string, mutate: (l: LiveState) => void): LiveState | null {
const base = state.live ?? { runId: _runId, phase: 'thinking', items: [], agentNotes: {}, startedAt: Date.now() }
function touchLive(_runId: string, mutate: (l: LiveState) => void, now = Date.now()): LiveState | null {
const base = state.live ?? { runId: _runId, phase: 'thinking', items: [], agentNotes: {}, startedAt: now }
const next: LiveState = { ...base, agentNotes: { ...base.agentNotes } }
mutate(next)
return next
@@ -798,3 +814,16 @@ function isConversationBusy(err: unknown): boolean {
export function activeRun(runs: Run[]): Run | undefined {
return runs.find(r => r.kind !== 'subagent' && !isTerminal(r.status))
}
// mergeStartedRun 把刚启动的 run 补进列表。同 id 已在(通常是对账拿到的更新记录)时不覆盖。
export function mergeStartedRun(runs: Run[], run: Run): Run[] {
if (runs.some(r => r.id === run.id)) return runs
return [...runs, run]
}
// eventTime 取事件上的服务端时间。重放时墙上时钟会把「已工作」和工具耗时算成 0。
export function eventTime(e: { ts?: string }, now = Date.now()): number {
if (!e.ts) return now
const t = Date.parse(e.ts)
return Number.isNaN(t) ? now : t
}
+15 -1
View File
@@ -4,7 +4,7 @@ import { describe, it, expect } from 'vitest'
import {
applyTextDelta, applyThinkingDelta, applyToolCall, applyToolCallDelta, applyToolResult, applyToolOutput, applyArtifactCreated,
applyCheck, applyFinalizing,
liveToMessage, livePhaseLabel, showProcessInline, type LiveItem,
hasSettledReply, liveToMessage, livePhaseLabel, showProcessInline, type LiveItem,
} from '../src/live'
import type { LiveState } from '../src/store'
@@ -191,3 +191,17 @@ describe('showProcessInline', () => {
expect(showProcessInline(null, false)).toBe(true)
})
})
describe('hasSettledReply', () => {
const user = { role: 'user', runId: 'r1' }
const assistant = { role: 'assistant', runId: 'r1' }
it('stays false while only the user message carries the run id', () => {
expect(hasSettledReply([user], 'r1')).toBe(false)
expect(hasSettledReply([user, { role: 'assistant', runId: 'other' }], 'r1')).toBe(false)
})
it('turns true once an assistant message for that run is stored', () => {
expect(hasSettledReply([user, assistant], 'r1')).toBe(true)
})
})
+37
View File
@@ -0,0 +1,37 @@
// 发送后的运行态投影:把刚启动的 run 合进列表,并用事件上的服务端时间。
import { describe, expect, it } from 'vitest'
import type { Run } from '../src/contracts'
import { activeRun, eventTime, mergeStartedRun } from '../src/store'
function run(id: string, status: Run['status'], over: Partial<Run> = {}): Run {
return {
id, conversationId: 'c', projectId: 'p', kind: 'conversation', status,
createdAt: '2026-10-03T00:00:00Z', ...over,
}
}
describe('mergeStartedRun', () => {
it('adds a pending run so the composer treats the conversation as active', () => {
const merged = mergeStartedRun([], run('r1', 'pending'))
expect(activeRun(merged)?.id).toBe('r1')
})
it('keeps an existing record when the same id arrives again', () => {
const settled = run('r1', 'completed', { phase: 'done' })
const merged = mergeStartedRun([settled], run('r1', 'pending'))
expect(merged).toHaveLength(1)
expect(merged[0].status).toBe('completed')
expect(activeRun(merged)).toBeUndefined()
})
})
describe('eventTime', () => {
it('parses the server timestamp', () => {
expect(eventTime({ ts: '2026-10-03T14:50:25.073Z' }, 0)).toBe(Date.parse('2026-10-03T14:50:25.073Z'))
})
it('falls back when the timestamp is missing or unparseable', () => {
expect(eventTime({}, 42)).toBe(42)
expect(eventTime({ ts: 'not-a-time' }, 42)).toBe(42)
})
})