修复聊天模式跨端运行状态订阅与重连恢复

src/client/stores/chat.ts

@@ -112,7 +112,6 @@ export const useChatStore = defineStore('chat', () => {
   let unreadLoadGeneration = 0
   let reconnectResumePromise: Promise<unknown> | undefined
   let persistedRecoveryPromise: Promise<void> | undefined
-  let hasConnectedOnce = false
   const runtimePromises = new Map<string, Promise<ChatRouteState>>()
 
   const activeRouteState = computed(() => {
@@ -248,10 +247,9 @@ export const useChatStore = defineStore('chat', () => {
     connectionState.value = state
     if (state === 'ready') {
       reconnectAttempt = 0
-      const shouldResume = hasConnectedOnce
-      hasConnectedOnce = true
       const active = activeRouteState.value
-      if (shouldResume && active?.isStreaming && !active.runtimeSessionId && !reconnectResumePromise) {
+      if (active?.historySynced && (active.isStreaming || activeSession.value?.owned === true)
+        && !active.route.sessionId.startsWith('draft-') && !active.runtimeSessionId && !reconnectResumePromise) {
         reconnectResumePromise = ensureRuntime(active)
           .catch(cause => { active.error = errorMessage(cause) })
           .finally(() => { reconnectResumePromise = undefined })
@@ -293,7 +291,8 @@ export const useChatStore = defineStore('chat', () => {
     const draftIndex = sessions.value.findIndex(session => session.id === oldState.route.sessionId && session.profile === oldState.route.profile)
     const now = Date.now() / 1000
     const summary: SessionSummary = draftIndex >= 0
-      ? { ...sessions.value[draftIndex], id: storedSessionId, owned: true, updatedAt: now }
+      ? { ...sessions.value[draftIndex], id: storedSessionId, owned: true,
+          updatedAt: newKey === oldKey ? sessions.value[draftIndex].updatedAt : now }
       : { id: storedSessionId, profile: oldState.route.profile, source: 'web', owned: true, title: '新会话', messageCount: 0, toolCallCount: 0, startedAt: now, updatedAt: now }
     if (draftIndex >= 0) sessions.value.splice(draftIndex, 1, summary)
     else sessions.value.unshift(summary)
@@ -370,7 +369,6 @@ export const useChatStore = defineStore('chat', () => {
   function disconnect(): void {
     if (reconnectTimer !== undefined) window.clearTimeout(reconnectTimer)
     reconnectTimer = undefined
-    hasConnectedOnce = false
     socket.close()
     runtimeRoutes.clear()
     runtimePromises.clear()
@@ -525,12 +523,19 @@ export const useChatStore = defineStore('chat', () => {
     activeProfileName.value = selectedProfile
     const state = ensureRoute(selectedProfile, sessionId)
     syncSelectedModel(session?.model, session?.provider)
-    // Reading an old session must not rewrite the server's last_active field.
-    // The runtime is attached lazily by send/interrupt/usage, or when a known
-    // in-flight turn reconnects. REST history remains authoritative here.
+    // REST owns history, but an owned chat also needs a runtime subscription:
+    // another device can start a turn while this viewer has never sent one.
+    // History-only sessions and unsent drafts must remain unattached.
     if (!state.historySynced) await loadHistory(state)
     const refreshed = sessions.value.find(item => item.id === sessionId && item.profile === selectedProfile)
     syncSelectedModel(refreshed?.model, refreshed?.provider)
+    if (refreshed?.owned === true && !sessionId.startsWith('draft-')
+      && activeSessionId.value === sessionId && activeProfileName.value === selectedProfile) {
+      await ensureRuntime(state).catch(cause => {
+        const current = routes[routeKey(selectedProfile, sessionId)]
+        if (current) current.error = errorMessage(cause)
+      })
+    }
   }
 
   async function forceRefreshHistory(): Promise<void> {
@@ -605,6 +610,7 @@ export const useChatStore = defineStore('chat', () => {
     if (!runtimeId) throw new Error('Hermes 未返回运行会话标识')
     const storedId = string(result.stored_session_id ?? result.storedSessionId ?? result.session_key, initialState.route.sessionId)
     const migrated = migrateRoute(initialKey, storedId, runtimeId)
+    migrated.error = undefined
     const info = record(result.info)
     if (!desiredModelRoutes.has(routeKey(migrated.route.profile, migrated.route.sessionId))) {
       syncSelectedModel(string(result.model ?? info.model), string(result.provider ?? info.provider))
@@ -619,13 +625,17 @@ export const useChatStore = defineStore('chat', () => {
     const inflight = record(result.inflight ?? result.active_run)
     const assistant = string(inflight.assistant)
     const inflightError = string(inflight.error)
-    migrated.isStreaming = bool(result.running) || bool(inflight.streaming)
+    migrated.isStreaming = bool(result.running) || bool(info.running) || bool(inflight.streaming)
     const queued = record(result.queued)
     const queuedUser = string(queued.user)
     migrated.isQueued = Boolean(queuedUser)
-    if (assistant || inflightError) {
+    // A turn can be thinking or calling tools before producing any text.
+    // Create its live row now so subsequent tool events have a visible target.
+    const existing = [...migrated.messages].reverse().find(message => message.role === 'assistant' && message.isStreaming)
+    if (migrated.isStreaming || inflightError || (existing && assistant)) {
       migrated.messages = mergeChatMessages(migrated.messages, [{
-        id: `resume:${runtimeId}`,
+        ...existing,
+        id: existing?.id ?? `resume:${runtimeId}`,
         sessionId: storedId,
         profile: migrated.route.profile,
         role: 'assistant',
@@ -636,6 +646,14 @@ export const useChatStore = defineStore('chat', () => {
         error: inflightError || undefined,
       }])
     }
+    if (!migrated.isStreaming) {
+      migrated.liveStatus = undefined
+      migrated.messages = migrated.messages.map(message => message.isStreaming
+        ? { ...message, isStreaming: false, stage: message.stage === 'streaming' ? 'settled' : message.stage }
+        : message)
+    }
+    migrated.pendingApproval = undefined
+    migrated.pendingClarification = undefined
     values(inflight.corrections).forEach((correction, index) => {
       const content = string(correction)
       if (!content) return
@@ -690,7 +708,8 @@ export const useChatStore = defineStore('chat', () => {
     const existing = runtimePromises.get(key)
     if (existing) return existing
     let operation: Promise<ChatRouteState>
-    operation = performEnsureRuntime(initialState).finally(() => {
+    // Register the operation before connect can publish ready and re-enter.
+    operation = Promise.resolve().then(() => performEnsureRuntime(initialState)).finally(() => {
       if (runtimePromises.get(key) === operation) runtimePromises.delete(key)
     })
     runtimePromises.set(key, operation)

tests/client/chatCrossClientSync.test.ts

@@ -0,0 +1,163 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { createPinia, setActivePinia } from 'pinia'
+import type { RealtimeConnectionState, RpcEventFrame, SessionSummary } from '@shared/types'
+import { routeKey } from '@/utils/id'
+
+const realtime = vi.hoisted(() => ({
+  event: undefined as ((event: RpcEventFrame['params']) => void) | undefined,
+  state: undefined as ((state: RealtimeConnectionState) => void) | undefined,
+  request: vi.fn(),
+}))
+const api = vi.hoisted(() => ({ getMessages: vi.fn() }))
+vi.mock('@/api/realtime', () => ({
+  RpcError: class extends Error {},
+  ChatRpcSocket: class {
+    onEvent(handler: typeof realtime.event) { realtime.event = handler }
+    onState(handler: typeof realtime.state) { realtime.state = handler }
+    async connect() { realtime.state?.('ready') }
+    close() {}
+    async request(...args: unknown[]) { return realtime.request(...args) }
+  },
+}))
+vi.mock('@/api/sessions', () => ({
+  getMessages: api.getMessages, getSession: vi.fn(), getSessions: vi.fn(),
+  deleteSession: vi.fn(), getSessionUnread: vi.fn(), markSessionRead: vi.fn(), updateSession: vi.fn(),
+}))
+vi.mock('@/api/profiles', () => ({ getModels: vi.fn() }))
+vi.mock('@/stores/auth', () => ({
+  useAuthStore: () => ({
+    user: { id: 'viewer' }, activeProfile: { name: 'alpha' },
+    status: 'authenticated', isAuthenticated: true,
+  }),
+}))
+
+import { useChatStore } from '@/stores/chat'
+
+function session(id = 'stored-1'): SessionSummary {
+  return { id, profile: 'alpha', owned: true, source: 'ios', title: '跨端聊天',
+    messageCount: 0, toolCallCount: 0, startedAt: 1, updatedAt: 2 }
+}
+function emit(type: string, payload: Record<string, unknown> = {}) {
+  realtime.event?.({ type, session_id: 'runtime-1', profile: 'alpha', payload: payload as never })
+}
+
+describe('cross-client chat subscriptions', () => {
+  let chat: ReturnType<typeof useChatStore>
+  beforeEach(() => {
+    setActivePinia(createPinia())
+    realtime.request.mockReset().mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1', running: false })
+    api.getMessages.mockReset().mockResolvedValue({ messages: [], total: 0, returned: 0, hasMore: false })
+    chat = useChatStore()
+    chat.sessions = [session()]
+  })
+  afterEach(() => { chat.disconnect(); vi.useRealTimers() })
+
+  it('subscribes a viewer before another device starts, then shows thinking, tools, text and completion', async () => {
+    await chat.selectSession('stored-1', 'alpha')
+    expect(realtime.request).toHaveBeenCalledExactlyOnceWith('session.resume', {
+      session_id: 'stored-1', profile: 'alpha', source: 'web', omit_messages: true,
+    })
+    expect(chat.sessions[0].updatedAt).toBe(2)
+    emit('message.start')
+    emit('thinking.delta', { text: '正在检查' })
+    expect(chat.isStreaming).toBe(true)
+    expect(chat.activeRouteState?.liveStatus).toBe('正在检查')
+    emit('tool.start', { tool_id: 'tool-1', name: 'read_file' })
+    expect(chat.messages.at(-1)?.toolCalls?.[0]).toMatchObject({ name: 'read_file', status: 'running' })
+    emit('tool.complete', { tool_id: 'tool-1', name: 'read_file', result: 'ok' })
+    emit('message.delta', { text: '检查完成' })
+    expect(chat.messages.at(-1)?.content).toBe('检查完成')
+    emit('message.complete', { text: '检查完成' })
+    expect(chat.isStreaming).toBe(false)
+    expect(chat.activeRouteState?.liveStatus).toBeUndefined()
+    expect(chat.messages.at(-1)?.toolCalls?.[0].status).toBe('completed')
+  })
+
+  it('restores a run opened before its first text, so tool events are not dropped', async () => {
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1',
+      running: true, inflight: { streaming: true, assistant: '' } })
+    await chat.selectSession('stored-1', 'alpha')
+    expect(chat.isStreaming).toBe(true)
+    expect(chat.messages).toHaveLength(1)
+    emit('tool.start', { tool_id: 'tool-1', name: 'terminal' })
+    expect(chat.messages[0].toolCalls?.[0].name).toBe('terminal')
+  })
+
+  it('restores a partial answer and appends later deltas without an extra row', async () => {
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1',
+      running: true, inflight: { assistant: '前半段' } })
+    await chat.selectSession('stored-1', 'alpha')
+    emit('message.delta', { text: '后半段' })
+    expect(chat.messages.map(message => message.content)).toEqual(['前半段后半段'])
+  })
+
+  it('resubscribes an idle viewer after disconnection and discovers a remote run', async () => {
+    await chat.selectSession('stored-1', 'alpha')
+    expect(chat.isStreaming).toBe(false)
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1',
+      running: true, inflight: { assistant: '断线期间开始的回复' } })
+    vi.useFakeTimers()
+    realtime.state?.('failed')
+    await vi.advanceTimersByTimeAsync(500)
+    expect(realtime.request).toHaveBeenCalledTimes(2)
+    expect(chat.isStreaming).toBe(true)
+    expect(chat.messages.at(-1)?.content).toBe('断线期间开始的回复')
+  })
+
+  it('does not repeatedly resume a subscribed session when revisiting it', async () => {
+    await chat.selectSession('stored-1', 'alpha')
+    await chat.selectSession('stored-1', 'alpha')
+    expect(realtime.request).toHaveBeenCalledTimes(1)
+    expect(api.getMessages).toHaveBeenCalledTimes(1)
+  })
+
+  it('settles a run and clears an interaction completed while the viewer was offline', async () => {
+    await chat.selectSession('stored-1', 'alpha')
+    emit('message.start')
+    emit('message.delta', { text: '部分回复' })
+    emit('thinking.delta', { text: '等待确认' })
+    emit('approval.request', { request_id: 'approval-1' })
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1',
+      running: false, inflight: { assistant: '完整回复', streaming: false } })
+    vi.useFakeTimers()
+    realtime.state?.('failed')
+    await vi.advanceTimersByTimeAsync(500)
+    expect(chat.isStreaming).toBe(false)
+    expect(chat.messages).toHaveLength(1)
+    expect(chat.messages[0]).toMatchObject({ content: '完整回复', stage: 'settled', isStreaming: false })
+    expect(chat.activeRouteState?.liveStatus).toBeUndefined()
+    expect(chat.pendingApproval).toBeUndefined()
+  })
+
+  it('does not duplicate completed history from a retained inactive snapshot', async () => {
+    api.getMessages.mockResolvedValue({ messages: [{ id: 'answer-1', sessionId: 'stored-1',
+      role: 'assistant', content: '已完成回复', timestamp: 1, stage: 'settled' }], total: 1, returned: 1, hasMore: false })
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'stored-1',
+      running: false, inflight: { assistant: '已完成回复', streaming: false } })
+    await chat.selectSession('stored-1', 'alpha')
+    expect(chat.messages.map(message => message.id)).toEqual(['answer-1'])
+  })
+
+  it.each([false, undefined])('keeps history-only sessions unattached with owned=%s', async owned => {
+    chat.sessions[0].owned = owned
+    await chat.selectSession('stored-1', 'alpha')
+    realtime.state?.('ready')
+    expect(realtime.request).not.toHaveBeenCalled()
+  })
+
+  it('never creates a runtime for an unsent draft on ready', () => {
+    const id = chat.createSession('alpha')
+    chat.routes[routeKey('alpha', id)].historySynced = true
+    realtime.state?.('ready')
+    expect(realtime.request).not.toHaveBeenCalled()
+  })
+
+  it('retains readable history when subscribing fails and can retry on selection', async () => {
+    realtime.request.mockRejectedValueOnce(new Error('connection lost'))
+    await chat.selectSession('stored-1', 'alpha')
+    expect(chat.historySynced).toBe(true)
+    expect(chat.activeRouteState?.error).toBe('connection lost')
+    await chat.selectSession('stored-1', 'alpha')
+    expect(chat.activeRouteState?.runtimeSessionId).toBe('runtime-1')
+  })
+})

tests/e2e/workspace-chat.spec.ts

@@ -83,6 +83,45 @@ test('an open Web chat list receives cross-client sessions and real title events
       .__crossClientProjectionSentinel)).toBe(sentinel)
 })
 
+test('an idle Web viewer follows an iOS-origin run and restores thinking after reload', async ({ page }, testInfo) => {
+  await login(page)
+  const capabilities = await (await page.request.get('/api/realtime/capabilities')).json()
+  const headers = { 'X-CSRF-Token': capabilities.csrfToken, Origin: new URL(page.url()).origin }
+  const opened = await page.request.post('/api/realtime/channels', { headers, data: { channel: 'chat' } })
+  expect(opened.ok(), await opened.text()).toBe(true)
+  const channel = (await opened.json()).id
+  let sequence = 0
+  async function command(method: string, params: Record<string, unknown>) {
+    const response = await page.request.post(`/api/realtime/channels/${channel}/commands`, {
+      headers: { ...headers, 'Idempotency-Key': `ios-live-${Date.now()}-${++sequence}` },
+      data: { jsonrpc: '2.0', method, params },
+    })
+    expect(response.ok(), await response.text()).toBe(true)
+    const receipt = await response.json()
+    expect(receipt.state).toBe('confirmed')
+    return receipt.response.result
+  }
+  const session = await command('session.create', { profile: 'default', source: 'ios' })
+  const subscribed = page.waitForResponse(response => response.url().endsWith('/commands')
+    && response.request().postDataJSON()?.method === 'session.resume')
+  await page.goto(`/chat/${session.stored_session_id}?profile=default`)
+  await subscribed
+  await command('prompt.submit', { session_id: session.session_id, text: '[cross-client-live] 检查文件' })
+  await expect(page.getByRole('status', { name: '正在思考', exact: true })).toBeVisible()
+  await page.getByText('思考与工具', { exact: true }).click()
+  await expect(page.getByText('read_file', { exact: true })).toBeVisible()
+  await page.screenshot({ path: testInfo.outputPath('ios-run-web-thinking-tools.png') })
+
+  // This Web page has never submitted a prompt or written an in-flight marker.
+  await page.evaluate(() => sessionStorage.clear())
+  await page.reload()
+  await expect(page.getByRole('status', { name: '正在思考', exact: true })).toBeVisible()
+  await page.request.post('http://127.0.0.1:19120/__release')
+  await expect(page.locator('.message--assistant')).toContainText('会话独立保存')
+  await expect(page.getByRole('status', { name: '正在思考', exact: true })).toHaveCount(0)
+  await page.screenshot({ path: testInfo.outputPath('ios-run-web-completed.png') })
+})
+
 test('created roles and editable teams share a durable chat list without plugin calls', async ({
   page,
   context,

tests/fixtures/workspace-server.ts

@@ -209,7 +209,8 @@ wss.on('connection', (socket) => {
         }
         sessions.set(stored.id, stored)
       }
-      const runtimeId = randomUUID()
+      // A live Hermes resume joins the same runtime, including other viewers.
+      const runtimeId = latestRuntimeBySession.get(stored.id) ?? randomUUID()
       runtimes.set(runtimeId, stored.id)
       latestRuntimeBySession.set(stored.id, runtimeId)
       respond({
@@ -241,12 +242,15 @@ wss.on('connection', (socket) => {
       const requested = /本轮用户指定成员：([^\n]*)/.exec(f.params.text)?.[1] ?? ''
       const delegates = f.params.text.includes('你是管理员。') && !f.params.text.includes('本批次执行结果：') && requested.startsWith('@')
       const markdownStream = String(f.params.text).includes('[streaming-markdown]')
+      const crossClientRun = String(f.params.text).includes('[cross-client-live]')
       const markdownPrefix = '# 流式标题\n\n**即时格式化**\n\n- 第一项\n- 第二项\n\n```ts\nconst answer = 42' + (String(f.params.text).includes('[long]') ? `\n\x60\x60\x60\n\n${'这是持续生成的长回复段落。\n\n'.repeat(80)}流式末尾标记\n\n\x60\x60\x60ts\nconst tail = 1` : '')
       const markdownFinal = `${markdownPrefix}\n\x60\x60\x60\n\n| 名称 | 数量 |\n| --- | --- |\n| 项目 | 2 |\n\n最终完整标记`
       const text = markdownStream ? markdownFinal : `我是${name}。已按角色规则处理这条消息。\n\n- 会话独立保存\n- 可以继续交流\n\n[报告](/tmp/workspace-report.txt)${delegates ? `\n请${requested}处理任务。` : ''}`
       event('message.start', {}, f.params.session_id)
-      setTimeout(() => event('message.delta', { text: markdownStream ? markdownPrefix : text.slice(0, 12) }, f.params.session_id), 80)
+      if (crossClientRun) event('tool.start', { tool_id: 'cross-client-tool', name: 'read_file' }, f.params.session_id)
+      else setTimeout(() => event('message.delta', { text: markdownStream ? markdownPrefix : text.slice(0, 12) }, f.params.session_id), 80)
       const complete = () => {
+        if (crossClientRun) event('tool.complete', { tool_id: 'cross-client-tool', name: 'read_file', result: 'ok' }, f.params.session_id)
         stored!.running = false
         stored!.messages.push({
           id: randomUUID(),
@@ -256,7 +260,7 @@ wss.on('connection', (socket) => {
         })
         event('message.complete', { text, status: 'complete' }, f.params.session_id)
       }
-      if (markdownStream || f.params.text.includes('[hold-workspace]')) heldReplies.push(complete)
+      if (crossClientRun || markdownStream || f.params.text.includes('[hold-workspace]')) heldReplies.push(complete)
       else setTimeout(complete, 250)
       return
     }