Drive the composer from one shell run phase

Why: The send button, stop button, and queued sends each looked at a different slice of run state, so a reconnect or a late assistant message could disagree.
Evidence: A depth-4 walk of send, HTTP results, SSE, reconnect, conversation switch, and flush keeps busy aligned with an in-flight run, hides the live bubble once an assistant message exists, and keeps one user message per request id.
Impact: The event listener list is exhaustive over the stream contract, including run.check. A disconnected stream shows 重连中.
Refs: feature/runstate-shell
Co-authored-by: Cursor <cursoragent@cursor.com>

src/api.ts

@@ -3,7 +3,7 @@
 import { auth } from './auth'
 import type {
   Project, Conversation, Message, Run, Artifact, HITLCard, HITLResponse,
-  AgentProfile, SettingDesc, StreamEvent, PlanSnapshot, TodoList, UploadInfo, PromptModule,
+  AgentProfile, SettingDesc, StreamEvent, StreamEventType, PlanSnapshot, TodoList, UploadInfo, PromptModule,
   ActivityCard,
 } from './contracts'
 
@@ -118,6 +118,34 @@ export interface SSEHandlers {
   onError?: (err: Error) => void
 }
 
+// 契约里每一种事件都要有监听。漏掉一种时 tsc 会报缺字段。
+const streamEventListeners: Record<StreamEventType, true> = {
+  'message.delta': true,
+  'thinking.delta': true,
+  'tool.call.delta': true,
+  'tool.call': true,
+  'tool.output': true,
+  'tool.result': true,
+  'run.phase': true,
+  'run.check': true,
+  'hitl.request': true,
+  'hitl.resolved': true,
+  'plan.update': true,
+  'todo.update': true,
+  'agent.spawn': true,
+  'agent.progress': true,
+  'agent.result': true,
+  'artifact.created': true,
+  'artifact.updated': true,
+  'run.completed': true,
+  'run.failed': true,
+  'error': true,
+  'card.created': true,
+  'card.updated': true,
+  'card.status': true,
+  'card.resolved': true,
+}
+
 export function connectEvents(conversationId: string, since: number, handlers: SSEHandlers, projectId?: string): { close: () => void } {
   let closed = false
   let es: EventSource | null = null
@@ -137,10 +165,7 @@ export function connectEvents(conversationId: string, since: number, handlers: S
     }
     es.onmessage = ev => handle('message', ev.data)
     // 同时监听具名事件与默认（服务端 event: 字段）。
-    const types = ['message.delta', 'thinking.delta', 'tool.call.delta', 'tool.call', 'tool.output', 'tool.result', 'run.phase',
-      'hitl.request', 'hitl.resolved', 'plan.update', 'todo.update', 'agent.spawn', 'agent.progress',
-      'agent.result', 'artifact.created', 'artifact.updated', 'run.completed', 'run.failed', 'error',
-      'card.created', 'card.updated', 'card.status', 'card.resolved']
+    const types = Object.keys(streamEventListeners) as StreamEventType[]
     for (const t of types) es.addEventListener(t, ev => handle(t, (ev as MessageEvent).data))
   }
 

src/components/Conversation.tsx

@@ -7,7 +7,8 @@ import { marked } from 'marked'
 import { api } from '../api'
 import { store, useStore, activeRun, type LiveState, type ContextObs } from '../store'
 import { scrollRatio, scrollTopForRatio } from '../scrollsync'
-import { hasSettledReply, liveToMessage, livePhaseLabel, showProcessInline } from '../live'
+import { liveToMessage, livePhaseLabel, showProcessInline } from '../live'
+import { deriveShellPhase, isShellBusy, showLiveBubble } from '../runstate'
 import type { Message, Part, HITLCard, ActivityCard, Artifact } from '../contracts'
 import { UIRenderer, InlineConfirmCard, hasOpenConfirm, CardAnchor } from '@ui'
 import { docBridge } from './Workbench'
@@ -414,6 +415,7 @@ export function Conversation() {
   const activeHitl = useStore(s => s.activeHitl)
   const live = useStore(s => s.live)
   const runs = useStore(s => s.runs)
+  const posting = useStore(s => s.posting)
   const connected = useStore(s => s.connected)
   const artifacts = useStore(s => s.artifacts)
   const profiles = useStore(s => s.profiles)
@@ -432,7 +434,14 @@ export function Conversation() {
   artifactCache = Object.fromEntries(artifacts.map(a => [a.id, a]))
   cardCache = Object.fromEntries(cards.map(c => [c.id, c]))
 
-  const active = activeRun(runs)
+  const phase = deriveShellPhase({
+    posting,
+    openRun: !!activeRun(runs),
+    liveRunId: live?.runId ?? null,
+    assistantRunIds: messages.filter(m => m.role === 'assistant' && m.runId).map(m => m.runId as string),
+  })
+  const busy = isShellBusy(phase)
+  const showStop = phase === 'running'
 
   useEffect(() => {
     if (!stickRef.current) return
@@ -445,7 +454,7 @@ export function Conversation() {
     const ids = attachmentIds
     const names = attachmentNames
     setText(''); setAttachmentIds([]); setAttachmentNames([])
-    if (active) {
+    if (busy) {
       store.enqueueOutbox(t, ids, names)
       return
     }
@@ -499,8 +508,9 @@ export function Conversation() {
           onClick={() => currentConvId && store.setCrossFlags(currentConvId, { allowCrossRead: conversation?.allowCrossRead === false })}>
           <Icon name="users" size={13} />跨读{conversation?.allowCrossRead === false ? '·关' : '·开'}
         </button>
-        <span title={connected ? '事件流已连接' : '事件流重连中'} className={`conn-dot ${connected ? 'on' : 'off'}`} />
-        {active && <button className="btn ghost" onClick={() => store.cancelActive()}>停止</button>}
+        <span title={connected ? '事件流已连接' : '重连中'} className={`conn-dot ${connected ? 'on' : 'off'}`} />
+        {!connected && <span className="conn-label">重连中</span>}
+        {showStop && <button className="btn ghost" onClick={() => store.cancelActive()}>停止</button>}
       </div>
 
       <div className="chat-body" ref={bodyRef}
@@ -541,7 +551,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 && !hasSettledReply(messages, live.runId) &&
+        {showLiveBubble(live?.runId ?? null, messages.filter(m => m.role === 'assistant' && m.runId).map(m => m.runId as string)) && live &&
           <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">
@@ -579,14 +589,14 @@ export function Conversation() {
             }} />
           </label>
           <textarea rows={1} value={text} placeholder={hasOpenConfirm(cards) ? '有待确认的命令——请先处理上方确认卡'
-            : active ? '继续输入以排队'
+            : busy ? '继续输入以排队'
             : artifacts.length > 0 ? '输入消息…（/ 命令，@knowledge 检索注入）' : '输入消息…（/ 命令）'}
             onChange={e => setText(e.target.value)}
             onKeyDown={e => { if (e.key === 'Enter' && !e.shiftKey) { e.preventDefault(); send() } }} />
-          {active && (
+          {showStop && (
             <button className="btn danger" onClick={() => store.cancelActive()}>停止</button>
           )}
-          <button className="btn" onClick={send} disabled={!text.trim() && attachmentIds.length === 0}>{active ? '排队' : '发送'}</button>
+          <button className="btn" onClick={send} disabled={!text.trim() && attachmentIds.length === 0}>{busy ? '排队' : '发送'}</button>
         </div>
       </div>
     </div>

src/runstate.test.ts

@@ -0,0 +1,136 @@
+import { describe, expect, it } from 'vitest'
+import { canFlushOutbox, deriveShellPhase, hasInFlight, isShellBusy, showLiveBubble, type ShellFacts } from './runstate'
+
+interface Model {
+  facts: ShellFacts
+  outbox: string[]
+  userRequestIds: string[]
+  flushed: string[]
+}
+
+const idle: ShellFacts = { posting: false, openRun: false, liveRunId: null, assistantRunIds: [] }
+
+function model(): Model {
+  return { facts: { ...idle, assistantRunIds: [] }, outbox: [], userRequestIds: [], flushed: [] }
+}
+
+type Step =
+  | 'send'
+  | 'post-ok'
+  | 'post-409'
+  | 'post-fail'
+  | 'delta'
+  | 'assistant'
+  | 'done'
+  | 'reconnect'
+  | 'switch'
+  | 'queue'
+  | 'flush'
+
+function apply(m: Model, step: Step): Model {
+  const facts = {
+    ...m.facts,
+    assistantRunIds: m.facts.assistantRunIds.slice(),
+  }
+  const next: Model = { facts, outbox: m.outbox.slice(), userRequestIds: m.userRequestIds.slice(), flushed: m.flushed.slice() }
+  const phase = deriveShellPhase(next.facts)
+  const requestId = 'req-1'
+  switch (step) {
+    case 'send':
+      if (isShellBusy(phase)) {
+        next.outbox.push(requestId)
+        return next
+      }
+      next.facts.posting = true
+      return next
+    case 'post-ok':
+      next.facts.posting = false
+      next.facts.openRun = true
+      next.facts.liveRunId = 'run-1'
+      if (!next.userRequestIds.includes(requestId)) next.userRequestIds.push(requestId)
+      return next
+    case 'post-409':
+      next.facts.posting = false
+      if (!next.outbox.includes(requestId)) next.outbox.unshift(requestId)
+      return next
+    case 'post-fail':
+      next.facts.posting = false
+      return next
+    case 'delta':
+      if (next.facts.openRun) next.facts.liveRunId = 'run-1'
+      return next
+    case 'assistant':
+      if (next.facts.liveRunId && !next.facts.assistantRunIds.includes(next.facts.liveRunId)) {
+        next.facts.assistantRunIds.push(next.facts.liveRunId)
+      }
+      return next
+    case 'done':
+      next.facts.openRun = false
+      return next
+    case 'reconnect':
+      return next
+    case 'switch':
+      return model()
+    case 'queue':
+      next.outbox.push('queued')
+      return next
+    case 'flush': {
+      if (!canFlushOutbox(deriveShellPhase(next.facts)) || next.outbox.length === 0) return next
+      const id = next.outbox.shift() ?? ''
+      next.flushed.push(id)
+      next.facts.posting = true
+      if (id === requestId && !next.userRequestIds.includes(id)) next.userRequestIds.push(id)
+      return next
+    }
+    default: {
+      const unexpected: never = step
+      throw new Error(unexpected)
+    }
+  }
+}
+
+function assertInvariants(m: Model) {
+  const phase = deriveShellPhase(m.facts)
+  expect(isShellBusy(phase)).toBe(hasInFlight(m.facts))
+  expect(showLiveBubble(m.facts.liveRunId, m.facts.assistantRunIds)).toBe(
+    !!m.facts.liveRunId && !m.facts.assistantRunIds.includes(m.facts.liveRunId),
+  )
+  if (isShellBusy(phase)) {
+    expect(canFlushOutbox(phase)).toBe(false)
+  }
+  expect(new Set(m.userRequestIds).size).toBe(m.userRequestIds.length)
+}
+
+const steps: Step[] = ['send', 'post-ok', 'post-409', 'post-fail', 'delta', 'assistant', 'done', 'reconnect', 'switch', 'queue', 'flush']
+
+describe('shell run state', () => {
+  it('walks sequences up to depth 4 without breaking the four invariants', () => {
+    let frontier: Model[] = [model()]
+    for (let depth = 0; depth < 4; depth++) {
+      const grown: Model[] = []
+      for (const current of frontier) {
+        for (const step of steps) {
+          const next = apply(current, step)
+          assertInvariants(next)
+          grown.push(next)
+        }
+      }
+      frontier = grown
+    }
+    expect(frontier.length).toBe(steps.length ** 4)
+  })
+
+  it('follows sending, running, settling, then idle', () => {
+    let m = model()
+    m = apply(m, 'send')
+    expect(deriveShellPhase(m.facts)).toBe('sending')
+    m = apply(m, 'post-ok')
+    expect(deriveShellPhase(m.facts)).toBe('running')
+    expect(showLiveBubble(m.facts.liveRunId, m.facts.assistantRunIds)).toBe(true)
+    m = apply(m, 'done')
+    expect(deriveShellPhase(m.facts)).toBe('settling')
+    m = apply(m, 'assistant')
+    expect(deriveShellPhase(m.facts)).toBe('idle')
+    expect(showLiveBubble(m.facts.liveRunId, m.facts.assistantRunIds)).toBe(false)
+  })
+})

src/runstate.ts

@@ -0,0 +1,39 @@
+// 产品壳运行态的唯一来源。按钮、停止、取消和排队发送都读这里，不再各自看 runs 或 live。
+// sending：请求已发出、run 还没出现。running：有未结束的 run。
+// settling：run 已结束，但这条 run 的助手消息还没落库。idle：可以发送下一条。
+
+export type ShellPhase = 'sending' | 'running' | 'settling' | 'idle'
+
+export interface ShellFacts {
+  posting: boolean
+  openRun: boolean
+  liveRunId: string | null
+  assistantRunIds: string[]
+}
+
+export function deriveShellPhase(facts: ShellFacts): ShellPhase {
+  const livePending = showLiveBubble(facts.liveRunId, facts.assistantRunIds)
+  if (facts.posting && !facts.openRun) return 'sending'
+  if (facts.openRun) return 'running'
+  if (livePending) return 'settling'
+  return 'idle'
+}
+
+export function isShellBusy(phase: ShellPhase): boolean {
+  return phase !== 'idle'
+}
+
+// 流式气泡只在这条 run 还没有助手消息时显示。用户消息上的同一个 runId 不算回复已落库。
+export function showLiveBubble(liveRunId: string | null, assistantRunIds: string[]): boolean {
+  if (!liveRunId) return false
+  return !assistantRunIds.includes(liveRunId)
+}
+
+export function canFlushOutbox(phase: ShellPhase): boolean {
+  return phase === 'idle'
+}
+
+// 在途：正在发送、有未结束的 run，或实时气泡还在等助手消息。
+export function hasInFlight(facts: ShellFacts): boolean {
+  return facts.posting || facts.openRun || showLiveBubble(facts.liveRunId, facts.assistantRunIds)
+}

src/store.ts

@@ -5,6 +5,7 @@ import { api, connectEvents } from './api'
 import { applyTextDelta, applyThinkingDelta, applyToolCall, applyToolCallDelta, applyToolResult, applyToolOutput, applyArtifactCreated, applyArtifactUpdated, applyCheck, applyFinalizing, type LiveItem } from './live'
 import { closePreviewTab, openPreviewTab, type PreviewTab } from './previewtabs'
 import { deliveryRequestId, enqueueMessage, enqueueMessageFront, newQueuedMessage, promoteQueuedMessage, removeQueuedMessage, replaceQueuedText, type QueuedMessage } from './outbox'
+import { canFlushOutbox, deriveShellPhase } from './runstate'
 import { flashElement } from './scrollsync'
 import { isThemeChoice, resolveTheme, type ThemeChoice } from './theme'
 import type {
@@ -80,6 +81,8 @@ interface AppState {
   serverWorkspaces: { id: string; title: string; icon?: string; shortcut?: string; body?: any[] }[]
   ui: UIState
   connected: boolean
+  /** 发送请求已经出去，run 记录还没回来。 */
+  posting: boolean
   error: string | null
   toast: string | null
   outbox: QueuedMessage[]
@@ -129,7 +132,7 @@ let state: AppState = {
   messages: [], runs: [], activeHitl: [], lastSeq: 0, live: null,
   plan: null, todos: null, artifacts: [], cards: [], projectCards: [], inboxScope: 'conv', runtimeMock: false, profiles: [], pluginPanels: [],
   serverWorkspaces: [],
-  ui: loadUI(), connected: false, error: null, toast: null,
+  ui: loadUI(), connected: false, posting: false, error: null, toast: null,
   outbox: [], outboxDraining: false,
   previewTabs: {}, previewActive: {},
 }
@@ -303,7 +306,7 @@ export const store = {
 
   async openConversation(convId: string) {
     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 })
+    set({ currentConvId: convId, messages: [], runs: [], activeHitl: [], lastSeq: 0, live: null, plan: null, todos: null, cards: [], inboxScope: 'conv', outbox: [], outboxDraining: false, posting: false })
     syncPreviewRail((state.previewTabs[convId] ?? []).length > 0)
     const resumeSeq = await store.reconcile()
     store.connectLive(resumeSeq ?? state.lastSeq)
@@ -470,12 +473,17 @@ export const store = {
         setTimeout(() => { void store.reconcile().then(() => store.flushOutbox()) }, 250)
         break
       }
+      default: {
+        const unexpected: never = e.type
+        void unexpected
+      }
     }
   },
 
   async sendMessage(text: string, attachmentIds: string[], requestId?: string): Promise<'ok' | 'busy'> {
     if (!state.currentConvId) return 'ok'
     const id = requestId ?? newQueuedMessage(text, attachmentIds, []).id
+    set({ posting: true })
     try {
       const out = await api.sendMessage(state.currentConvId, text, attachmentIds, id)
       if (out.command) {
@@ -514,6 +522,8 @@ export const store = {
       store.toast(`发送失败：${e}`)
       await store.reconcile()
       return 'ok'
+    } finally {
+      set({ posting: false })
     }
   },
 
@@ -533,7 +543,7 @@ export const store = {
     const item = state.outbox.find(q => q.id === id)
     if (!item) return
     const rest = removeQueuedMessage(state.outbox, id)
-    if (activeRun(state.runs) || state.live) {
+    if (!canFlushOutbox(shellPhase())) {
       set({ outbox: promoteQueuedMessage(state.outbox, id) })
       store.toast('回合进行中，已放到队首，结束后自动发送')
       return
@@ -544,7 +554,7 @@ export const store = {
 
   async flushOutbox() {
     if (state.outboxDraining) return
-    if (activeRun(state.runs) || state.live) return
+    if (!canFlushOutbox(shellPhase())) return
     const next = state.outbox[0]
     if (!next) return
     set({ outbox: state.outbox.slice(1), outboxDraining: true })
@@ -555,7 +565,7 @@ export const store = {
       set({ outboxDraining: false })
     }
     if (result === 'busy') return
-    if (!activeRun(state.runs) && !state.live && state.outbox.length > 0) {
+    if (canFlushOutbox(shellPhase()) && state.outbox.length > 0) {
       await store.flushOutbox()
     }
   },
@@ -811,6 +821,16 @@ function isConversationBusy(err: unknown): boolean {
   return String(err).includes('409:')
 }
 
+function shellPhase() {
+  const liveRunId = state.live?.runId ?? null
+  return deriveShellPhase({
+    posting: state.posting,
+    openRun: !!activeRun(state.runs),
+    liveRunId,
+    assistantRunIds: state.messages.filter(m => m.role === 'assistant' && m.runId).map(m => m.runId as string),
+  })
+}
+
 export function activeRun(runs: Run[]): Run | undefined {
   return runs.find(r => r.kind !== 'subagent' && !isTerminal(r.status))
 }

src/styles.css

@@ -330,6 +330,7 @@ button { font-family: inherit; font-size: inherit; color: inherit; }
 .conn-dot { width: 8px; height: 8px; border-radius: 50%; }
 .conn-dot.on { background: var(--ok); }
 .conn-dot.off { background: var(--err); }
+.conn-label { font-size: var(--fs-ui-sm); color: var(--text-faint); }
 .doc-toolbar { display: flex; gap: 8px; margin-bottom: 10px; align-items: center; }
 ::-webkit-scrollbar { width: 10px; height: 10px; }
 ::-webkit-scrollbar-thumb { background: var(--bg-active); border-radius: 6px; border: 3px solid transparent; background-clip: content-box; }