From e4ca2acfa69bd6bde809c6b27813f9250fb91f19 Mon Sep 17 00:00:00 2001 From: weichen Date: Sat, 3 Oct 2026 15:21:28 +0000 Subject: [PATCH] Show a run as active as soon as the send returns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: Stamping the user message with the run id made the live bubble look settled, and the composer stayed on 发送 until the next reconcile. Evidence: npx tsc -b passed. vitest 75, including hasSettledReply and mergeStartedRun. On a mock server the button became 停止/排队 within 250ms and the reply grew while the run was open. Impact: Composer, live bubble, and agent tracker only. A stored user message no longer hides the stream. Reopening replays from resumeSeq when the server sends it. Refs: Dev fix/live-run-visibility Co-authored-by: Cursor --- src/api.ts | 2 +- src/components/AgentTracker.tsx | 6 ++- src/components/Conversation.tsx | 4 +- src/live.ts | 6 +++ src/store.ts | 75 +++++++++++++++++++++++---------- test/live.test.ts | 16 ++++++- test/send.test.ts | 37 ++++++++++++++++ 7 files changed, 117 insertions(+), 29 deletions(-) create mode 100644 test/send.test.ts diff --git a/src/api.ts b/src/api.ts index 29faedd..dd39e5c 100644 --- a/src/api.ts +++ b/src/api.ts @@ -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(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 }) }) diff --git a/src/components/AgentTracker.tsx b/src/components/AgentTracker.tsx index 55e2b3c..232a388 100644 --- a/src/components/AgentTracker.tsx +++ b/src/components/AgentTracker.tsx @@ -19,10 +19,12 @@ export function AgentTracker() { // 中间过程：该会话消息树中归属子代理 run 的消息（run_id 过滤）。 const subMessages: Record = {} 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 双层窗框）。 diff --git a/src/components/Conversation.tsx b/src/components/Conversation.tsx index 83972de..3662403 100644 --- a/src/components/Conversation.tsx +++ b/src/components/Conversation.tsx @@ -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 return })} - {live && !messages.some(m => m.runId === live.runId) && + {live && !hasSettledReply(messages, live.runId) && } {activeHitl.map(c => )} {messages.length === 0 &&
diff --git a/src/live.ts b/src/live.ts index c14015a..49103f5 100644 --- a/src/live.ts +++ b/src/live.ts @@ -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) +} diff --git a/src/store.ts b/src/store.ts index e146ff3..5ab7768 100644 --- a/src/store.ts +++ b/src/store.ts @@ -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 { 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 +} diff --git a/test/live.test.ts b/test/live.test.ts index 6cf7d40..45d61a7 100644 --- a/test/live.test.ts +++ b/test/live.test.ts @@ -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) + }) +}) diff --git a/test/send.test.ts b/test/send.test.ts new file mode 100644 index 0000000..af661c6 --- /dev/null +++ b/test/send.test.ts @@ -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 { + 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) + }) +}) -- 2.54.0