修复普通聊天列表未读数仅统计最终回复

src/client/api/sessions.ts

@@ -145,7 +145,7 @@ export async function getSessionUnread(profile?: string): Promise<Record<string,
       const id = typeof item.sessionId === 'string'
         ? item.sessionId
         : typeof item.session_id === 'string' ? item.session_id : ''
-      if (id) result[id] = Math.max(0, number(item.unreadCount ?? item.unread_count ?? item.count))
+      if (id) result[id] = Math.max(0, number(item.final_unread_count ?? item.unreadCount ?? item.unread_count ?? item.count))
     }
     return result
   }
@@ -156,7 +156,7 @@ export async function getSessionUnread(profile?: string): Promise<Record<string,
     else {
       const item = record(value)
       const id = typeof item.sessionId === 'string' ? item.sessionId : typeof item.session_id === 'string' ? item.session_id : sessionId
-      result[id] = Math.max(0, number(item.unreadCount ?? item.unread_count ?? item.count))
+      result[id] = Math.max(0, number(item.final_unread_count ?? item.unreadCount ?? item.unread_count ?? item.count))
     }
   }
   return result

src/server/chatCache.ts

@@ -1,4 +1,5 @@
 import { projectChatEvent } from './chatEventProjection.js'
+import { isFinalChatResult } from './chatUnread.js'
 import { HttpError } from './errors.js'
 import { createHash, randomUUID } from 'node:crypto'
 import { chmodSync, createReadStream, existsSync, mkdirSync, renameSync, statSync, unlinkSync, writeFileSync } from 'node:fs'
@@ -193,7 +194,7 @@ export class ChatCacheStore {
       owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL, event_id TEXT NOT NULL,
       PRIMARY KEY(owner,profile,session_id,event_id));`)
     const localColumns=this.db.prepare('PRAGMA table_info(chat_local_state)').all() as {name:string}[]
-    for(const [name,type] of [['title','TEXT'],['archived','INTEGER'],['read_initialized','INTEGER NOT NULL DEFAULT 0']]){
+    for(const [name,type] of [['title','TEXT'],['archived','INTEGER'],['read_initialized','INTEGER NOT NULL DEFAULT 0'],['final_read_count','INTEGER']]){
       if(!localColumns.some(c=>c.name===name))this.db.exec(`ALTER TABLE chat_local_state ADD COLUMN ${name} ${type}`)
     }
     const sessionColumns = this.db.prepare('PRAGMA table_info(chat_sessions)').all() as Array<{ name: string }>
@@ -345,8 +346,10 @@ export class ChatCacheStore {
       FROM chat_sessions s LEFT JOIN chat_local_state l USING(owner,profile,session_id) WHERE ${where}
       ORDER BY pin DESC, COALESCE(json_extract(s.data,'$.last_active'),0) DESC, s.session_id
       LIMIT ? OFFSET ?`).all(...args, limit, offset) as Array<Record<string,any>>
+    const unread = new Map(this.unread(owner, profile).sessions.map(row => [JSON.stringify([row.profile, row.session_id]), row.final_unread_count]))
     return {status:200,headers:new Headers({'content-type':'application/json'}),body:Buffer.from(JSON.stringify({
       sessions:rows.map(r=>({...JSON.parse(r.data),...(r.local_title!=null?{title:r.local_title}:{}),id:r.session_id,profile:r.profile,owned:true,pinned:!!r.pin,
+        final_unread_count:unread.get(JSON.stringify([r.profile,r.session_id]))??0,
         sync_state:r.sync_state,history_complete:!!r.complete})), total,offset,limit,
       sync_state:this.needsListMetadata(owner,profile)?'syncing':'current'
     }))}
@@ -367,9 +370,17 @@ export class ChatCacheStore {
 
   unread(owner: string, profile: string) {
     const sessions = (this.db.prepare(`SELECT s.profile,s.session_id,COALESCE(s.message_total,0) message_count,
+      CASE WHEN COALESCE(l.read_initialized,0)=1 THEN COALESCE(l.final_read_count,l.read_count) ELSE COALESCE(s.message_total,0) END final_read_count,
       CASE WHEN COALESCE(l.read_initialized,0)=1 THEN l.read_count ELSE COALESCE(s.message_total,0) END read_message_count FROM chat_sessions s LEFT JOIN chat_local_state l USING(owner,profile,session_id)
       WHERE s.owner=? AND s.owned_at IS NOT NULL AND (?='' OR s.profile=?)`).all(owner,profile,profile) as any[])
-      .map(r=>({...r,updated_at:Date.now(),unread_count:Math.max(0,r.message_count-r.read_message_count)}))
+      .map(({final_read_count,...r})=>{
+        const messages = this.db.prepare(`SELECT data FROM chat_messages
+          WHERE owner=? AND profile=? AND session_id=? AND position>=?
+          AND json_extract(data,'$.role')='assistant'`)
+          .all(owner,r.profile,r.session_id,final_read_count) as {data:string}[]
+        return {...r,updated_at:Date.now(),unread_count:Math.max(0,r.message_count-r.read_message_count),
+          final_unread_count:messages.filter(row=>isFinalChatResult(JSON.parse(row.data))).length}
+      })
     return {profile,supported:true,total_unread:sessions.reduce((n,r)=>n+r.unread_count,0),sessions}
   }
 
@@ -379,8 +390,17 @@ export class ChatCacheStore {
     const total = Number((this.db.prepare('SELECT message_total FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?')
       .get(owner,profile,id) as any).message_total ?? 0)
     const read = Math.max(0,Math.min(total, count ?? total))
-    this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,read_count,read_initialized) VALUES(?,?,?,?,1)
-      ON CONFLICT(owner,profile,session_id) DO UPDATE SET read_count=MAX(read_count,excluded.read_count),read_initialized=1`).run(owner,profile,id,read)
+    // Reading a streaming placeholder must not consume its future final answer.
+    // Keep the original cursor untouched for existing consumers.
+    const pending = this.db.prepare(`SELECT MIN(position) position FROM chat_messages
+      WHERE owner=? AND profile=? AND session_id=? AND position<?
+      AND message_id=(SELECT assistant_id FROM chat_local_state WHERE owner=? AND profile=? AND session_id=?)
+      AND json_extract(data,'$.role')='assistant' AND json_extract(data,'$.status')='streaming'`)
+      .get(owner,profile,id,read,owner,profile,id) as {position:number|null}
+    const finalRead = pending.position ?? read
+    this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,read_count,final_read_count,read_initialized) VALUES(?,?,?,?,?,1)
+      ON CONFLICT(owner,profile,session_id) DO UPDATE SET read_count=MAX(read_count,excluded.read_count),
+      final_read_count=MAX(COALESCE(final_read_count,read_count),excluded.final_read_count),read_initialized=1`).run(owner,profile,id,read,finalRead)
     return this.unread(owner,profile).sessions.find(r=>r.session_id===id)!
   }
 

src/server/chatEventProjection.ts

@@ -29,5 +29,6 @@ export function projectChatEvent(type: string, payload: ChatRecord, previous: Ch
   } else if (type === 'error' || type === 'run.failed') {
     message.status = 'failed'; message.error = String(payload.error ?? payload.message ?? '运行失败')
   } else if (type !== 'message.interim') message.status = 'streaming'
+  message.final_result = (type === 'message.complete' || type === 'run.completed') && message.status === 'complete'
   return message
 }

src/server/chatUnread.ts

@@ -0,0 +1,26 @@
+/** List badges count completed results; transport rows and progress never count. */
+export function isFinalChatResult(message: Record<string, any>): boolean {
+  if (message.role !== 'assistant' || message.error || message.is_streaming || message.isStreaming) return false
+  if (message.final_result === false || message.is_interim || message.display_kind || message.displayKind) return false
+  const status = message.status ?? message.finish_reason ?? message.finishReason
+  if (['streaming', 'pending', 'running', 'interim', 'failed', 'error', 'cancelled', 'interrupted', 'tool_calls', 'function_call'].includes(status)) return false
+  // Live projections retain tool traces on the final answer. Historical tool
+  // invocation rows have no explicit completion marker and must be excluded.
+  if (message.final_result !== true) {
+    let tools = message.tool_calls ?? message.toolCalls ?? message.tools
+    if (typeof tools === 'string') {
+      try { tools = JSON.parse(tools) } catch { return false }
+    }
+    if ((Array.isArray(tools) ? tools.length > 0 : !!tools) || message.function_call) return false
+    if (Array.isArray(message.content) && message.content.some(block =>
+      ['tool_use', 'tool_call', 'function_call'].includes(block?.type))) return false
+  }
+  const content = message.content ?? message.text
+  const visible = (value: unknown) => typeof value === 'string'
+    && value.replace(/<(think|thinking|reasoning)\b[^>]*>[\s\S]*?(?:<\/\1>|$)/gi, '').trim().length > 0
+  return visible(content)
+    || (Array.isArray(content) && content.some(block => block && (
+      ['text', 'output_text'].includes(block.type) && visible(block.text)
+      || ['image', 'image_url', 'file', 'audio', 'video'].includes(block.type))))
+    || (Array.isArray(message.attachments) && message.attachments.length > 0)
+}

tests/client/sessionUnread.test.ts

@@ -0,0 +1,13 @@
+import { afterEach, expect, it, vi } from 'vitest'
+import { getSessionUnread } from '@/api/sessions'
+
+afterEach(() => vi.unstubAllGlobals())
+
+it('uses final result counts for list badges, including an explicit zero', async () => {
+  vi.stubGlobal('fetch', vi.fn(async () => Response.json({ sessions: [
+    { session_id: 'result', unread_count: 12, final_unread_count: 1 },
+    { session_id: 'tools', unread_count: 8, final_unread_count: 0 },
+    { session_id: 'legacy', unread_count: 2 },
+  ] })))
+  expect(await getSessionUnread('default')).toEqual({ result: 1, tools: 0, legacy: 2 })
+})

tests/server/chatCacheRoutes.test.ts

@@ -150,6 +150,11 @@ describe('source=web chat cache routes', () => {
       source: 'ios',
       owned: true,
       title: 'Hermes 仍标记为 iOS',
+      final_unread_count: 1,
+    })])
+    const unread = await agent.get('/api/app/sessions/unread?profile=default').set('Host', host).expect(200)
+    expect(unread.body.sessions).toEqual([expect.objectContaining({
+      session_id: 'owned-session', final_unread_count: 1,
     })])
 
     const history = await agent

tests/server/chatUnread.test.ts

@@ -0,0 +1,106 @@
+// @vitest-environment node
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it } from 'vitest'
+import { ChatCacheStore } from '../../src/server/chatCache.js'
+import { isFinalChatResult } from '../../src/server/chatUnread.js'
+
+const stores: ChatCacheStore[] = [], homes: string[] = []
+function setup() {
+  const home = mkdtempSync(join(tmpdir(), 'chat-unread-')); homes.push(home)
+  const store = new ChatCacheStore(home); stores.push(store)
+  store.recordCommand('owner', 'default', 'chat', 'session.create', {})
+  return { store, home }
+}
+function snapshot(store: ChatCacheStore, messages: object[], total = messages.length) {
+  store.putSnapshot('owner', 'messages', 'messages', 'default', 'chat', {
+    status: 200, headers: new Headers(), body: Buffer.from(JSON.stringify({
+      messages, pagination: { total, offset: 0 },
+    })),
+  })
+}
+afterEach(() => {
+  for (const store of stores.splice(0)) if (!store.isClosed) store.close()
+  for (const home of homes.splice(0)) rmSync(home, { recursive: true, force: true })
+})
+
+describe('ordinary conversation list unread results', () => {
+  it('counts final history replies while preserving the existing counters and raw read positions', () => {
+    const { store } = setup()
+    const messages = [
+      { id: 'u', role: 'user', content: '问题' },
+      { id: 'call', role: 'assistant', content: '先查一下', tool_calls: JSON.stringify([{ id: 'tool' }]) },
+      { id: 'tool', role: 'tool', content: '查询结果' },
+      { id: 'reasoning', role: 'assistant', content: '<think>思考</think>' },
+      { id: 'interim', role: 'assistant', content: '继续处理', final_result: false, status: 'complete' },
+      { id: 'final', role: 'assistant', content: '最终结果' },
+      { id: 'user2', role: 'user', content: '再问' },
+      { id: 'final2', role: 'assistant', content: '第二个结果', tool_calls: '[]' },
+    ]
+    snapshot(store, messages)
+    expect(store.unread('owner', 'default')).toMatchObject({ total_unread: 8, sessions: [
+      { unread_count: 8, message_count: 8, read_message_count: 0, final_unread_count: 2 },
+    ] })
+    expect(store.markRead('owner', 'default', 'chat', 4)).toMatchObject({ read_message_count: 4, final_unread_count: 2 })
+    expect(store.markRead('owner', 'default', 'chat', 6)).toMatchObject({ read_message_count: 6, final_unread_count: 1 })
+    const list = JSON.parse(store.localList('owner', 'default', {}).body.toString())
+    expect(list.sessions[0]).toMatchObject({ message_count: 8, final_unread_count: 1 })
+    snapshot(store, messages)
+    expect(store.markRead('owner', 'default', 'chat', 0).final_unread_count).toBe(1)
+    expect(store.unread('other', 'default').sessions).toEqual([])
+    expect(store.unread('owner', 'other').sessions).toEqual([])
+    expect(store.markRead('owner', 'default', 'chat', 999).final_unread_count).toBe(0)
+  })
+
+  it('does not count tools, progress, or an unfinished answer, then counts completion once across restart', () => {
+    const { store, home } = setup()
+    let seq = 0
+    const event = (type: string, payload: object = {}) => store.recordEvent('owner', 'default', 'chat', {type, payload, delivery_id: String(++seq)})
+    event('message.start', { message_id: 'a' })
+    event('message.delta', { text: '处理中' })
+    event('message.interim')
+    event('tool.start', { tool_id: 't', name: 'search' })
+    event('tool.complete', { tool_id: 't', result: 'found' })
+    expect(store.unread('owner', 'default').sessions[0].final_unread_count).toBe(0)
+    store.markRead('owner', 'default', 'chat')
+    event('message.complete', { text: '最终结果' })
+    event('run.completed')
+    expect(store.unread('owner', 'default').sessions[0]).toMatchObject({ unread_count: 0, final_unread_count: 1 })
+    store.close()
+    const restored = new ChatCacheStore(home); stores.push(restored)
+    expect(restored.unread('owner', 'default').sessions[0].final_unread_count).toBe(1)
+    expect(restored.markRead('owner', 'default', 'chat').final_unread_count).toBe(0)
+  })
+
+  it('uses existing read cursors for old caches and does not guess from missing history rows', () => {
+    const { store } = setup()
+    snapshot(store, [{ id: 'a', role: 'assistant', content: '结果' }], 20)
+    store.db.prepare('UPDATE chat_local_state SET read_count=19, final_read_count=NULL').run()
+    expect(store.unread('owner', 'default').sessions[0].final_unread_count).toBe(1)
+    store.db.prepare('UPDATE chat_local_state SET read_count=20').run()
+    expect(store.unread('owner', 'default').sessions[0].final_unread_count).toBe(0)
+  })
+
+  it('does not let an abandoned streaming row prevent later results from being marked read', () => {
+    const { store } = setup()
+    store.recordEvent('owner', 'default', 'chat', {type:'message.delta',payload:{text:'旧回复'},delivery_id:'old'})
+    store.recordCommand('owner', 'default', 'chat', 'prompt.submit', {text:'新问题',_delivery_id:'new'})
+    store.recordEvent('owner', 'default', 'chat', {type:'message.complete',payload:{text:'新结果'},delivery_id:'done'})
+    expect(store.unread('owner', 'default').sessions[0].final_unread_count).toBe(1)
+    expect(store.markRead('owner', 'default', 'chat').final_unread_count).toBe(0)
+  })
+
+  it('recognizes attachment-only results and ignores failed or reasoning-only outputs', () => {
+    expect(isFinalChatResult({ role: 'assistant', attachments: [{ path: 'result.pdf' }] })).toBe(true)
+    expect(isFinalChatResult({ role: 'assistant', content: [{ type: 'text', text: '结果' }] })).toBe(true)
+    for (const message of [
+      { role: 'system', content: 'notice' },
+      { role: 'assistant', reasoning: '思考', content: '' },
+      { role: 'assistant', content: '部分结果', status: 'failed' },
+      { role: 'assistant', content: '处理中', status: 'streaming' },
+      { role: 'assistant', content: [{ type: 'tool_use', id: 't' }] },
+      { role: 'assistant', content: [{ type: 'text', text: '查询中' }, { type: 'tool_use', id: 't' }] },
+    ]) expect(isFinalChatResult(message)).toBe(false)
+  })
+})