统一完整聊天未读计数和快照已读回执

src/server/app.ts

@@ -1,3 +1,4 @@
+import { unreadSnapshot } from './unreadCenter.js'
 import {ExecutionSettings} from './executionSettings.js'
 import {CredentialVaultCoordinator} from './credentialVault/coordinator.js'
 import {CredentialVaultClient} from './credentialVault/transport.js'
@@ -249,6 +250,7 @@ export function createApplication(options: ApplicationOptions = {}): Application
   })
   // The application owns group events. No Dashboard plugin is queried.
   const workspace = new WorkspaceStore(config.home)
+  push.setUnreadCounter(owner => unreadSnapshot(workspace, chatCache!.store, owner).total)
   const openVikingService = new OpenVikingService(workspace, openVikingConfiguration)
   const workspaceNodes = new WorkspaceNodes(workspace, config, { url: config.upstream, client: upstream, session: upstreamSession }, pairings.nodeID)
   workspaceNodes.sourceAllowed = (owner, nodeId, profile) => auth.canUseSource(owner, nodeId, profile)

src/server/chatTranscriptStore.ts

@@ -577,6 +577,20 @@ export class ChatTranscriptStore {
           .prepare('INSERT OR IGNORE INTO chat_transcript_reads VALUES(?,?,?,?)')
           .run(owner, profile, sessionId, m.id)
   }
+  unreadMessages(...scope: Scope): TranscriptMessage[] {
+    this.unread(...scope)
+    return this.db.prepare(`SELECT m.data FROM chat_transcript_messages m LEFT JOIN chat_transcript_reads r USING(owner,profile,session_id,id)
+      WHERE m.owner=? AND m.profile=? AND m.session_id=? AND m.deleted=0 AND r.id IS NULL ORDER BY m.seq,m.id`).all(...scope)
+      .map(row=>JSON.parse(String(row.data)) as TranscriptMessage).filter(m=>m.final_result===true && isFinalChatResult(m))
+  }
+  markMessagesRead(owner:string,profile:string,sessionId:string,ids:string[]):void {
+    const selected=new Set(ids)
+    const head=this.head(owner,profile,sessionId);head.localReadEstablished=true
+    this.db.prepare('UPDATE chat_transcript_heads SET data=? WHERE owner=? AND profile=? AND session_id=?').run(JSON.stringify(head),owner,profile,sessionId)
+    for(const m of this.unreadMessages(owner,profile,sessionId))if(selected.has(m.id))
+      this.db.prepare('INSERT OR IGNORE INTO chat_transcript_reads VALUES(?,?,?,?)').run(owner,profile,sessionId,m.id)
+    this.notify()
+  }
   unread(...scope: Scope): number {
     // Legacy read cursors can finish importing after the initial local snapshot.
     // Map only persisted source rows, never consume a live streaming result.

src/server/pushCoordinator.ts

@@ -896,6 +896,8 @@ export class PushCoordinator {
     })
   }
 
+  private unreadCounter?: (owner: string) => number
+  setUnreadCounter(counter: (owner: string) => number): void { this.unreadCounter = counter }
   resetBadge(userIdValue: string, installationIdValue: string, clientAccountIdValue: string): number {
     const userId = requiredIdentifier(userIdValue, 'userId')
     const installationId = requiredIdentifier(installationIdValue, 'installationId')
@@ -904,8 +906,9 @@ export class PushCoordinator {
       const installation = state.installations.find(item => item.userId === userId
         && item.installationId === installationId
         && item.clientAccountId === clientAccountId)
-      if (installation) installation.badge = 0
-      return 0
+      const count = this.unreadCounter?.(userId) ?? 0
+      if (installation) installation.badge = count
+      return count
     })
   }
 
@@ -1258,7 +1261,7 @@ export class PushCoordinator {
       if (state.outbox.length + targets.length > MAX_OUTBOX_ITEMS) throw new Error('Push outbox limit reached')
       state.processedEvents[key] = now
       for (const target of targets) {
-        target.badge += 1
+        target.badge = this.unreadCounter ? this.unreadCounter(userId) : target.badge + 1
         state.outbox.push({
           id: randomUUID(),
           eventId,

src/server/routes.ts

@@ -1,3 +1,5 @@
+import { acknowledgeUnread, unreadSnapshot } from './unreadCenter.js'
+import { z } from 'zod'
 import { RegistrationLimiter } from './subaccountAccess.js'
 import { readServerIdentity, updateServerIdentity } from './serverIdentity.js'
 import { authorizeFileRead, fileAccessWorkingDirectory, readFileAccess, saveFileAccess } from './fileAccess.js'
@@ -2153,6 +2155,19 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
     const owner=dependencies.auth.require(ctx).id
     ctx.body={session_ids:dependencies.chatCache!.store.pins(owner,String(ctx.query.profile??''))}
   })
+  const unreadInput=z.object({items:z.array(z.object({mode:z.enum(['bot','chat']),id:z.string().min(1).max(256),profile:z.string().max(256).optional(),messages:z.array(z.object({id:z.string().min(1).max(512),seq:z.number().int().nonnegative(),version:z.number().int().nonnegative().optional(),taskId:z.string().max(256).optional()}).strict()).max(20000)}).strict()).max(1000)}).strict()
+  router.get('/api/app/unread',ctx=>{
+    const owner=dependencies.auth.require(ctx).id
+    ctx.set('Cache-Control','no-store')
+    ctx.body=unreadSnapshot(dependencies.workspace,dependencies.chatCache!.store,owner,(node,profile)=>dependencies.auth.canUseSource(owner,node,profile))
+  })
+  router.post('/api/app/unread/read',ctx=>{
+    const owner=dependencies.auth.require(ctx).id,input=unreadInput.safeParse(body(ctx))
+    if(!input.success)throw new HttpError(400,'未读快照格式无效','invalid_unread_snapshot')
+    acknowledgeUnread(dependencies.workspace,dependencies.chatCache!.store,owner,input.data.items,(node,profile)=>dependencies.auth.canUseSource(owner,node,profile))
+    for(const item of input.data.items)if(item.mode==='chat')dependencies.onChatListChanged?.(owner,item.profile ?? 'default',item.id)
+    ctx.body=unreadSnapshot(dependencies.workspace,dependencies.chatCache!.store,owner,(node,profile)=>dependencies.auth.canUseSource(owner,node,profile))
+  })
   router.get('/api/app/sessions/unread', ctx => {
     const owner=dependencies.auth.require(ctx).id
     ctx.body=dependencies.chatCache!.store.unread(owner,String(ctx.query.profile??''))

src/server/unreadCenter.ts

@@ -0,0 +1,51 @@
+import type { WorkspaceStore } from './workspaceStore.js'
+import type { ChatCacheStore } from './chatCache.js'
+import type { WorkspaceAgent, WorkspaceConversation } from '../shared/workspace.js'
+import { emptyUnread, type UnreadSnapshot, type UnreadConversation } from '../shared/unread.js'
+import { notificationPlainText } from './notificationText.js'
+
+export function unreadSnapshot(workspace:WorkspaceStore, cache:ChatCacheStore, owner:string,
+  allowed:(node:string,profile:string)=>boolean = ()=>true):UnreadSnapshot {
+  const result=emptyUnread()
+  for(const c of workspace.list<WorkspaceConversation>(owner,'conversation')) {
+    if(c.archived || !c.memberIds.every(id=>{const a=workspace.get<WorkspaceAgent>(owner,'agent',id);return !a || allowed(a.nodeId,a.profile)}))continue
+    const messages=workspace.chatUnreadMessages(owner,c.id)
+    if(!messages.length)continue
+    result.conversations.push({mode:'bot',id:c.id,name:c.name,preview:notificationPlainText(messages.at(-1)!.content,{maximum:160,fallback:messages.at(-1)!.attachments[0]?.name ?? '附件'}),count:messages.length,
+      updatedAt:messages.at(-1)!.createdAt,messages:messages.map(m=>({id:m.id,seq:m.seq,version:m.chatUnreadVersion,taskId:m.conversationTaskId}))})
+  }
+  for(const row of cache.unread(owner,'').sessions) {
+    if(!allowed('local',row.profile))continue
+    const stored=cache.db.prepare('SELECT s.data,l.archived FROM chat_sessions s LEFT JOIN chat_local_state l USING(owner,profile,session_id) WHERE s.owner=? AND s.profile=? AND s.session_id=?').get(owner,row.profile,row.session_id)
+    if(!stored || stored.archived===1)continue
+    const summary=JSON.parse(String(stored.data))
+    if(summary.archived || summary.source==='yaoyao_workspace')continue
+    const messages=cache.transcripts.unreadMessages(owner,row.profile,row.session_id)
+    if(!messages.length)continue
+    const last=messages.at(-1)!
+    result.conversations.push({mode:'chat',id:row.session_id,profile:row.profile,name:summary.title || summary.name || '未命名聊天',preview:notificationPlainText(typeof last.content==='string'?last.content:'',{maximum:160,fallback:'附件'}),count:messages.length,
+      updatedAt:Number(last.timestamp ?? last.created_at ?? last.createdAt ?? 0),messages:messages.map(m=>({id:m.id,seq:m.seq}))})
+  }
+  result.conversations.sort((a,b)=>b.updatedAt-a.updatedAt || a.id.localeCompare(b.id))
+  for(const row of result.conversations)result[row.mode]+=row.count
+  result.total=result.bot+result.chat
+  return result
+}
+
+/** Only IDs in the user's click/render snapshot are acknowledged; later completions survive. */
+export function acknowledgeUnread(workspace:WorkspaceStore,cache:ChatCacheStore,owner:string,
+  items:Array<Pick<UnreadConversation,'mode'|'id'|'profile'|'messages'>>,allowed:(node:string,profile:string)=>boolean=()=>true):void {
+  for(const item of items) {
+    if(item.mode==='bot') {
+      const c=workspace.require<WorkspaceConversation>(owner,'conversation',item.id)
+      if(!c.memberIds.every(id=>{const a=workspace.get<WorkspaceAgent>(owner,'agent',id);return !a || allowed(a.nodeId,a.profile)}))continue
+      workspace.markChatMessagesRead(owner,c.id,item.messages)
+    } else {
+      const profile=item.profile ?? 'default'
+      if(!allowed('local',profile))continue
+      cache.requireOwned(owner,profile,item.id)
+      cache.transcripts.seed(owner,profile,item.id)
+      cache.transcripts.markMessagesRead(owner,profile,item.id,item.messages.map(m=>m.id))
+    }
+  }
+}

src/server/workspaceStore.ts

@@ -10,6 +10,7 @@ import { isVisibleMessageFile } from '../shared/messageFiles.js'
 import { normalizeUploadName } from '../shared/uploadNames.js'
 import type { StoredWorkspaceFile } from './workspaceAssets.js'
 import type { SharedComputer } from './sharedComputers.js'
+import { isWorkspaceChatReply } from '../shared/unread.js'
 import { notificationPlainText } from './notificationText.js'
 import {readHostTools} from './hostToolSettings.js'
 import {supportsHostEnvironment,deriveComputer} from '../shared/workspace.js'
@@ -270,8 +271,7 @@ export class WorkspaceStore {
         for (const task of this.list<Task>(owner, 'conversation-task')) {
           task.unread ??= task.unreadCount > 0
           task.unreadVersion ??= task.unread ? 1 : 0
-          task.unreadCount = task.unread ? 1 : 0
-          this.put(owner, 'conversation-task', task.id, task)
+            this.put(owner, 'conversation-task', task.id, task)
         }
         for (const conversation of this.list<Conversation>(owner, 'conversation')) {
           conversation.unread ??= conversation.kind === 'group'
@@ -284,6 +284,25 @@ export class WorkspaceStore {
       }
       this.db.prepare('INSERT INTO workspace_migrations VALUES(?)').run(migration)
     })
+    this.atomic(() => {
+      this.db.exec('CREATE TABLE IF NOT EXISTS workspace_chat_reads(owner TEXT NOT NULL,id TEXT NOT NULL,PRIMARY KEY(owner,id))')
+      const migration = 'workspace-chat-unread-v2'
+      if (this.db.prepare('SELECT id FROM workspace_migrations WHERE id=?').get(migration)) return
+      for (const owner of this.owners()) for (const c of this.list<Conversation>(owner, 'conversation')) {
+        let version = c.unreadVersion ?? 0
+        for (const m of this.list<Message>(owner, 'message').filter(m => m.conversationId === c.id).sort((a,b) => a.seq-b.seq)) {
+          if (!isWorkspaceChatReply(m)) continue
+          m.chatUnreadVersion = ++version
+          const scope = m.conversationTaskId ? this.get<Task>(owner,'conversation-task',m.conversationTaskId) : c
+          if (!scope?.unread || m.seq < scope.readSeq || m.seq === scope.readSeq && m.seq !== scope.lastSeq)
+            this.db.prepare('INSERT OR IGNORE INTO workspace_chat_reads VALUES(?,?)').run(owner,m.id)
+          this.put(owner,'message',m.id,m)
+        }
+        c.unreadVersion = version
+        this.refreshChatUnread(owner,c)
+      }
+      this.db.prepare('INSERT INTO workspace_migrations VALUES(?)').run(migration)
+    })
     this.atomic(() => {
       const migration='host-environment-option-v1'
       if(this.db.prepare('SELECT id FROM workspace_migrations WHERE id=?').get(migration))return
@@ -294,6 +313,42 @@ export class WorkspaceStore {
       this.db.prepare('INSERT INTO workspace_migrations VALUES(?)').run(migration)
     })
   }
+  chatUnreadMessages(owner: string, conversationId: string, taskId?: string): Message[] {
+    return this.db.prepare(`SELECT e.data FROM workspace_entities e LEFT JOIN workspace_chat_reads r ON r.owner=e.owner AND r.id=e.id
+      WHERE e.owner=? AND e.kind='message' AND json_extract(e.data,'$.conversationId')=? AND r.id IS NULL`).all(owner,conversationId)
+      .map(row => JSON.parse(String(row.data)) as Message).filter(m => (!taskId || m.conversationTaskId === taskId) && isWorkspaceChatReply(m))
+      .sort((a,b) => a.seq-b.seq || a.id.localeCompare(b.id))
+  }
+  private refreshChatUnread(owner: string, c: Conversation): void {
+    const messages = this.chatUnreadMessages(owner,c.id)
+    c.unreadCount = messages.length; c.unread = messages.length > 0
+    this.put(owner,'conversation',c.id,c)
+    if (c.kind === 'group') for (const task of this.tasks(owner,c.id)) {
+      const pending = messages.filter(m => m.conversationTaskId === task.id)
+      task.unreadCount = pending.length;task.unread = pending.length > 0
+      task.unreadVersion = Math.max(task.unreadVersion ?? 0,...pending.map(m => m.chatUnreadVersion ?? 0))
+      this.put(owner,'conversation-task',task.id,task)
+    }
+  }
+  markChatMessagesRead(owner: string, conversationId: string, ids: Array<{id:string;version?:number}>): Conversation {
+    return this.atomic(() => {
+      const c = this.require<Conversation>(owner,'conversation',conversationId)
+      const selected = new Map(ids.map(m => [m.id,m.version ?? Number.MAX_SAFE_INTEGER]))
+      for (const m of this.chatUnreadMessages(owner,c.id)) {
+        if (!selected.has(m.id) || (m.chatUnreadVersion ?? 0) > selected.get(m.id)!) continue
+        this.db.prepare('INSERT OR IGNORE INTO workspace_chat_reads VALUES(?,?)').run(owner,m.id)
+        c.readSeq = Math.max(c.readSeq,m.seq)
+        if (m.conversationTaskId) {
+          const task=this.requireTask(owner,c.id,m.conversationTaskId)
+          task.readSeq=Math.max(task.readSeq,m.seq);this.put(owner,'conversation-task',task.id,task)
+        }
+      }
+      this.refreshChatUnread(owner,c)
+      if(c.kind==='group')for(const task of this.tasks(owner,c.id))this.event(owner,'task.changed',task,c.id)
+      this.event(owner,'conversation.changed',c,c.id)
+      return this.conversationSummary(owner,c)
+    })
+  }
   private hasUnreadTask(owner: string, conversationId: string): boolean {
     return !!this.db.prepare("SELECT 1 FROM workspace_entities WHERE owner=? AND kind='conversation-task' AND json_extract(data,'$.conversationId')=? AND json_extract(data,'$.unread')=1 LIMIT 1").get(owner, conversationId)
   }
@@ -907,31 +962,25 @@ export class WorkspaceStore {
     })
   }
   markTaskRead(owner: string, conversationId: string, taskId: string, seq: number, unreadVersion?: number): Task {
-    return this.atomic(() => {
-      const task = this.requireTask(owner, conversationId, taskId)
-      task.readSeq = Math.max(task.readSeq, Math.min(seq, task.lastSeq))
-      if (unreadVersion === undefined ? seq >= task.lastSeq : unreadVersion === (task.unreadVersion ?? 0)) task.unread = false
-      task.unreadCount = task.unread ? 1 : 0
-      this.put(owner, 'conversation-task', task.id, task)
-      const conversation = this.require<Conversation>(owner, 'conversation', conversationId)
-      conversation.unread = this.hasUnreadTask(owner, conversationId)
-      conversation.unreadCount = conversation.unread ? 1 : 0
-      this.put(owner, 'conversation', conversationId, conversation)
-      this.event(owner, 'task.changed', task, conversationId)
-      this.event(owner, 'conversation.changed', conversation, conversationId)
-      return task
-    })
+    const task=this.requireTask(owner,conversationId,taskId)
+    this.markChatMessagesRead(owner,conversationId,this.chatUnreadMessages(owner,conversationId,taskId)
+      .filter(m => m.seq<=seq && (unreadVersion===undefined || (m.chatUnreadVersion ?? 0)<=unreadVersion))
+      .map(m=>({id:m.id,version:m.chatUnreadVersion})))
+    task.readSeq=Math.max(task.readSeq,Math.min(seq,task.lastSeq))
+    const updated=this.requireTask(owner,conversationId,taskId)
+    updated.readSeq=Math.max(updated.readSeq,task.readSeq);this.put(owner,'conversation-task',taskId,updated)
+    this.event(owner,'task.changed',updated,conversationId)
+    return updated
   }
   markConversationRead(owner: string, conversationId: string, seq: number, unreadVersion?: number): Conversation {
-    return this.atomic(() => {
-      const conversation = this.require<Conversation>(owner, 'conversation', conversationId)
-      conversation.readSeq = Math.max(conversation.readSeq, Math.min(seq, conversation.lastSeq))
-      if (conversation.kind === 'direct' && (unreadVersion === undefined ? seq >= conversation.lastSeq : unreadVersion === (conversation.unreadVersion ?? 0))) conversation.unread = false
-      conversation.unreadCount = conversation.unread ? 1 : 0
-      this.put(owner, 'conversation', conversationId, conversation)
-      this.event(owner, 'conversation.changed', conversation, conversationId)
-      return this.conversationSummary(owner, conversation)
-    })
+    const c=this.require<Conversation>(owner,'conversation',conversationId)
+    if(c.kind==='group')return this.conversationSummary(owner,c)
+    const updated=this.markChatMessagesRead(owner,conversationId,this.chatUnreadMessages(owner,conversationId)
+      .filter(m=>m.seq<=seq && (unreadVersion===undefined || (m.chatUnreadVersion ?? 0)<=unreadVersion))
+      .map(m=>({id:m.id,version:m.chatUnreadVersion})))
+    updated.readSeq=Math.max(updated.readSeq,Math.min(seq,c.lastSeq));this.put(owner,'conversation',c.id,updated)
+    this.event(owner,'conversation.changed',updated,c.id)
+    return updated
   }
   deleteTask(owner: string, conversationId: string, taskId: string): void {
     this.atomic(() => {
@@ -958,8 +1007,7 @@ export class WorkspaceStore {
       conversation.previewAgentId = last?.role === 'assistant' ? last.agentId : undefined
       conversation.lastMessageAt = last?.createdAt
       conversation.updatedAt = Date.now()
-      conversation.unread = this.hasUnreadTask(owner, conversationId)
-      conversation.unreadCount = conversation.unread ? 1 : 0
+      this.refreshChatUnread(owner, conversation)
       this.projectConversationRuns(owner, conversation)
       this.put(owner, 'conversation', conversationId, conversation)
       this.event(owner, 'conversation.changed', conversation, conversationId)
@@ -1079,15 +1127,11 @@ export class WorkspaceStore {
       if (previous?.visible === false && message.visible !== false) message.seq = 0
       if (!message.seq) message.seq = c.lastSeq + 1
       c.lastSeq = Math.max(c.lastSeq, message.seq)
-      const needsAttention = (value: Message | undefined) => !!value && value.role !== 'user' && value.visible !== false
-        && ['complete', 'failed', 'interrupted'].includes(value.status)
-        && !!(value.content.trim() || value.error || value.browserCard || value.serviceWarnings?.length || this.messageForDisplay(owner, value).attachments.length)
-      if (needsAttention(message) && !needsAttention(previous)) {
-        c.unread = true
-        c.unreadVersion = (c.unreadVersion ?? 0) + 1
-        if (task) { task.unread = true; task.unreadVersion = (task.unreadVersion ?? 0) + 1 }
-      }
-      c.unreadCount = c.unread ? 1 : 0
+      if (isWorkspaceChatReply(message) && !previous?.chatUnreadVersion) {
+        c.unreadVersion=(c.unreadVersion ?? 0)+1
+        message.chatUnreadVersion=c.unreadVersion
+        if(task)task.unreadVersion=c.unreadVersion
+      } else if (previous?.chatUnreadVersion) message.chatUnreadVersion=previous.chatUnreadVersion
       if (updateSummary && message.visible !== false && (message.content.trim() || message.attachments.length || message.browserCard && !previous?.browserCard || message.serviceWarnings?.length && !previous?.serviceWarnings?.length)) {
         c.updatedAt = Date.now()
         c.lastMessageAt = Math.max(c.lastMessageAt ?? 0, message.createdAt)
@@ -1102,7 +1146,6 @@ export class WorkspaceStore {
         if (becameVisible) task.messageCount += 1
         if (becameHidden) task.messageCount = Math.max(0, task.messageCount - 1)
         task.lastSeq = Math.max(task.lastSeq, message.seq)
-        task.unreadCount = task.unread ? 1 : 0
         if (message.visible !== false && (message.content.trim() || message.attachments.length || message.browserCard && !previous?.browserCard || message.serviceWarnings?.length && !previous?.serviceWarnings?.length)) {
           if (message.role === 'user' && task.titleSource === 'automatic' && task.messageCount <= 1) {
             const source = notificationPlainText(message.content, { maximum: 48, fallback: message.attachments[0]?.name ?? '' }).trim()
@@ -1112,8 +1155,9 @@ export class WorkspaceStore {
           task.updatedAt = Date.now()
         }
         this.put(owner, 'conversation-task', task.id, task)
-        this.event(owner, 'task.changed', task, c.id)
       }
+      this.refreshChatUnread(owner,c)
+      if(task && updateSummary)this.event(owner,'task.changed',this.requireTask(owner,c.id,task.id),c.id)
       if (patch) this.event(owner, 'message.patch', patch, c.id, { type: 'message.changed', data: message })
       else this.event(owner, 'message.changed', message, c.id)
       if (updateSummary) {
@@ -1133,7 +1177,7 @@ export class WorkspaceStore {
       ...conversation,
       unread: conversation.unread ?? false,
       unreadVersion: conversation.unreadVersion ?? 0,
-      unreadCount: conversation.unread ? 1 : 0,
+      unreadCount: conversation.unreadCount ?? 0,
       ...(member ? { avatar: this.agentSummary(member).avatar } : {}),
       lastMessageAt: conversation.lastMessageAt
         ?? (conversation.lastSeq > 0 ? this.messages(owner, conversation.id, Number.MAX_SAFE_INTEGER, 1).at(-1)?.createdAt : undefined)

src/shared/unread.ts

@@ -0,0 +1,13 @@
+export type UnreadMode = 'bot' | 'chat'
+export interface UnreadMessage { id: string; seq: number; version?: number; taskId?: string }
+export interface UnreadConversation {
+  mode: UnreadMode; id: string; profile?: string; name: string; preview: string;
+  count: number; updatedAt: number; messages: UnreadMessage[];
+}
+export interface UnreadSnapshot { total: number; bot: number; chat: number; conversations: UnreadConversation[] }
+export const emptyUnread = (): UnreadSnapshot => ({ total: 0, bot: 0, chat: 0, conversations: [] })
+/** Tool traces may accompany a final answer; standalone process/system records never qualify. */
+export function isWorkspaceChatReply(m: { role: string; status: string; visible?: boolean; error?: string; content: string; attachments?: unknown[] }): boolean {
+  return m.role === 'assistant' && m.status === 'complete' && m.visible !== false && !m.error
+    && (!!m.content.replace(/<(think|thinking|reasoning)\b[^>]*>[\s\S]*?(?:<\/\1>|$)/gi, '').trim() || !!m.attachments?.length)
+}

src/shared/workspace.ts

@@ -179,6 +179,8 @@ export interface WorkspaceFile {
   sender: 'user' | 'agent'
 }
 export interface WorkspaceMessage {
+  /** Completion version assigned once to the stable user-facing reply ID. */
+  chatUnreadVersion?: number
   browserCard?: import('./managedBrowser.js').ManagedBrowserCard
   /** Server-owned, nonfatal availability notices; never model-authored markup. */
   serviceWarnings?: Array<{code:string;service:string;message:string}>

tests/server/unreadCenter.test.ts

@@ -0,0 +1,32 @@
+// @vitest-environment node
+import { afterEach, expect, it, vi } from 'vitest'
+import { mkdtempSync,rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { WorkspaceStore } from '../../src/server/workspaceStore'
+import { ChatCacheStore,ChatCacheCoordinator } from '../../src/server/chatCache'
+import { unreadSnapshot,acknowledgeUnread } from '../../src/server/unreadCenter'
+const homes:string[]=[];const stores:Array<{close():void}>=[]
+afterEach(()=>{stores.reverse().forEach(s=>s.close());stores.length=0;homes.splice(0).forEach(h=>rmSync(h,{recursive:true,force:true}))})
+it('shares exact snapshots between devices, isolates owners/profiles, and excludes task duplication and process rows',()=>{
+ const home=mkdtempSync(join(tmpdir(),'unread-integration-'));homes.push(home)
+ const workspace=new WorkspaceStore(home),cache=new ChatCacheStore(home),coordinator=new ChatCacheCoordinator(cache,{request:vi.fn()} as any);stores.push(workspace,cache,coordinator)
+ const a=workspace.createAgent('a',{name:'bot',profile:'p'}),b=workspace.createAgent('a',{name:'two',profile:'p'}),c=workspace.createGroup('a',{name:'group',memberIds:[a.id,b.id],administratorId:a.id}),task=workspace.tasks('a',c.id)[0]!
+ workspace.saveMessage('a',{id:'bot-final',conversationId:c.id,conversationTaskId:task.id,seq:0,role:'assistant',status:'complete',content:'附件答复',reasoning:'',attachments:[],tools:[],createdAt:1})
+ for(const profile of ['p','other']){
+  cache.recordCommand('a',profile,'s','session.create',{})
+  coordinator.observe('a',profile,'s',{type:'message.interim',payload:{text:'检查中'},delivery_id:'progress'})
+  coordinator.observe('a',profile,'s',{type:'message.complete',payload:{text:'答复'},delivery_id:'final'})
+  coordinator.observe('a',profile,'s',{type:'message.complete',payload:{text:'答复'},delivery_id:'final'})
+ }
+ const device1=unreadSnapshot(workspace,cache,'a'),device2=unreadSnapshot(workspace,cache,'a')
+ expect(device1).toMatchObject({total:3,bot:1,chat:2});expect(device2).toEqual(device1)
+ expect(unreadSnapshot(workspace,cache,'b').total).toBe(0)
+ expect(unreadSnapshot(workspace,cache,'a',(_,p)=>p==='p').total).toBe(2)
+ coordinator.observe('a','p','s',{type:'message.start',payload:{},delivery_id:'new-start'})
+ coordinator.observe('a','p','s',{type:'message.complete',payload:{text:'随后新消息'},delivery_id:'new-final'})
+ acknowledgeUnread(workspace,cache,'a',device1.conversations)
+ acknowledgeUnread(workspace,cache,'a',device2.conversations)
+ expect(unreadSnapshot(workspace,cache,'a')).toMatchObject({total:1,chat:1,bot:0})
+ expect(workspace.require<any>('a','conversation-task',task.id).unreadCount).toBe(0)
+})

tests/server/workspaceUnread.test.ts

@@ -24,9 +24,9 @@ beforeEach(() => {
 })
 afterEach(() => { vi.restoreAllMocks(); store.close(); rmSync(home, { recursive: true, force: true }) })
 
-it('stores one unread flag for any number of completed replies and persists read state', () => {
+it('counts each completed reply and persists read state', () => {
   for (let i = 0; i < 12; i++) store.saveMessage(owner, message())
-  expect(current(direct.id)).toMatchObject({ unread: true, unreadCount: 1 })
+  expect(current(direct.id)).toMatchObject({ unread: true, unreadCount: 12 })
   const display = vi.spyOn(store, 'messageForDisplay')
   expect(store.conversationSummary(owner, current(direct.id)).unread).toBe(true)
   store.markConversationRead(owner, direct.id, current(direct.id).lastSeq, current(direct.id).unreadVersion)
@@ -71,7 +71,7 @@ it('does not let a stale task read clear another completion and recomputes flags
   const first = store.tasks(owner, group.id)[0]!
   store.saveMessage(owner, message(group.id, { conversationTaskId: first.id }))
   const snapshot = task(first.id)
-  store.saveMessage(owner, message(group.id, { conversationTaskId: first.id, status: 'failed', error: '执行失败' }))
+  store.saveMessage(owner, message(group.id, { conversationTaskId: first.id, status: 'complete', content: '新消息' }))
   store.markTaskRead(owner, group.id, first.id, snapshot.lastSeq, snapshot.unreadVersion)
   expect(task(first.id).unread).toBe(true)
   expect(() => store.markTaskRead('another-owner', group.id, first.id, 100, 1)).toThrow()
@@ -86,11 +86,12 @@ it('migrates legacy counts once without loading message bodies and does not resu
   store.saveMessage(owner, message(group.id, { conversationTaskId: first.id }))
   store.db.exec("UPDATE workspace_entities SET data=json_remove(data,'$.unread','$.unreadVersion') WHERE kind IN ('conversation','conversation-task')")
   store.db.prepare('DELETE FROM workspace_migrations WHERE id=?').run('workspace-unread-flag-v1')
+  store.db.prepare('DELETE FROM workspace_migrations WHERE id=?').run('workspace-chat-unread-v2')
   store.close(); store = new WorkspaceStore(home)
-  expect(current(direct.id)).toMatchObject({ unread: true, unreadVersion: 1, unreadCount: 1 })
+  expect(current(direct.id)).toMatchObject({ unread: true, unreadCount: 1 })
   expect(current(group.id).unread).toBe(true)
   expect(task(first.id).unread).toBe(true)
-  store.markConversationRead(owner, direct.id, current(direct.id).lastSeq, 1)
+  store.markConversationRead(owner, direct.id, current(direct.id).lastSeq, current(direct.id).unreadVersion)
   store.close(); store = new WorkspaceStore(home)
   expect(current(direct.id).unread).toBe(false)
 })
@@ -108,3 +109,25 @@ it('reads only hidden IDs while preserving owner and task boundaries', () => {
   const plan = store.db.prepare("EXPLAIN QUERY PLAN SELECT id FROM workspace_entities WHERE owner=? AND kind='message' AND json_extract(data,'$.conversationId')=? AND json_extract(data,'$.visible')=0 AND json_extract(data,'$.conversationTaskId')=?").all(owner, group.id, first.id)
   expect(plan.some(row => String(row.detail).includes('workspace_message_hidden'))).toBe(true)
 })
+
+it('snapshot IDs exclude future completions and keep gaps, retries and restart idempotent', () => {
+  const a=message(), b=message(), streaming=message(direct.id,{status:'streaming'})
+  for(const m of [a,b,streaming])store.saveMessage(owner,m)
+  const snapshot=store.chatUnreadMessages(owner,direct.id).map(m=>({id:m.id,version:m.chatUnreadVersion}))
+  streaming.status='complete';store.saveMessage(owner,streaming)
+  store.markChatMessagesRead(owner,direct.id,[snapshot[1]!])
+  expect(current(direct.id).unreadCount).toBe(2)
+  store.markChatMessagesRead(owner,direct.id,snapshot)
+  store.markChatMessagesRead(owner,direct.id,snapshot)
+  expect(store.chatUnreadMessages(owner,direct.id).map(m=>m.id)).toEqual([streaming.id])
+  store.close();store=new WorkspaceStore(home)
+  expect(current(direct.id).unreadCount).toBe(1)
+  store.saveMessage(owner,a)
+  expect(current(direct.id).unreadCount).toBe(1)
+})
+it('excludes failure, interruption, reasoning and system; counts attachments once',()=>{
+  for(const patch of [{status:'failed',error:'失败'},{status:'interrupted'},{role:'system'},{content:'<think>过程</think>'}])store.saveMessage(owner,message(direct.id,patch as Partial<WorkspaceMessage>))
+  const attachment=message(direct.id,{content:'',attachments:[{id:'f',name:'结果.pdf',size:1,mime:'application/pdf'} as any]})
+  store.saveMessage(owner,attachment);store.saveMessage(owner,attachment)
+  expect(current(direct.id).unreadCount).toBe(1)
+})