完善普通聊天 Web 本地存储与实时同步，分离历史后台补齐

docs/verification/2026-09-10-ordinary-chat/README.md

@@ -0,0 +1,34 @@
+# 普通聊天 Web 完整存储与 SSE 验收
+
+日期：2026-09-10。只修改源码与本地测试；没有更新正在使用的 Web 服务或手机。
+
+## 行为
+
+- 普通聊天列表、详情、正文分页、置顶和未读从 Web SQLite 读取。完整数据读取不访问 Hermes；缺失区间返回 `CHAT_HISTORY_SYNC_PENDING`，已存内容不等待补齐。
+- 实时事件与消息更新在同一 SQLite 事务中保存，然后继续 SSE 推送。相同文本的不同事件不会被误去重；SSE 与正文使用同一消息 ID，支持身份提升与运行会话别名迁移。
+- 用户消息带稳定提交身份；发送确认和失败独立保存。Web 的标题、置顶和已读位置不被上游历史覆盖。
+- 旧历史和缺口交给持久化补齐队列，最多两个并发任务，失败退避重试。校验修订号，保留未确认提交与窗口外的旧历史；服务重启后继续未完成任务。
+- 完整历史补齐由 POST `/api/app/sessions/:id/sync` 提交，返回 202。完成后推送 `sessions.changed` / `reason: cache.synced`。原生历史通过显式 `view=history` 或原生入口保留原读取路径。
+- `/healthz` 的 `chatCache.reads` 提供本地读取和补齐统计，不包含凭据或消息内容。
+
+## 验证
+
+- `npm run build` 成功（既有前端 chunk 大小提示仍存在）。
+- `NODE_OPTIONS=--no-experimental-webstorage npx vitest run tests/server tests/client --maxWorkers=2`：122 个文件、871 项通过。设置 Node 参数让 jsdom 提供浏览器存储；不改产品配置。
+- Chromium `tests/e2e/ordinary-local.spec.ts`：1 项通过。通过真实 Web 服务和假 Hermes 测试后端完成发送、SSE 展示、Web 本地正文读取、页面刷新，并检查用户消息和助手回复各一条。截图见 [browser.png](browser.png)。
+- E2E 清除当前进程继承的代理变量，测试只访问回环地址 18801/19119；没有修改系统代理。测试后端为 fixture，不代表真实模型首 token 或设备网络延迟。
+- 全量测试中曾有一次 `pushRoutes` 发生 ECONNRESET；最终限制两个测试 worker 后完整通过。
+
+## 本机性能基准
+
+运行 `npx tsx docs/verification/2026-09-10-ordinary-chat/benchmark.ts`。结果见 [benchmark.json](benchmark.json)。
+
+- 1,000 个已登记会话，200 次首屏读取：P95 约 1.99 ms。
+- 500 次短文本事件落盘：P95 约 0.23 ms。
+- 本轮读取未启动后台补齐。
+
+这是本机 SQLite 微基准，不包含模型生成、网络、真机渲染，不能据此宣称端到端延迟。每个事件当前直接事务保存，没有新增轮询等待。
+
+## 交付边界
+
+没有部署、发布版本、构建分发 IPA 或安装真机。旧库采用增量表结构迁移；普通聊天模式之外的执行模式不迁移成 Web 聊天。附件仍使用现有下载和缓存机制。

docs/verification/2026-09-10-ordinary-chat/benchmark.json

@@ -0,0 +1,22 @@
+{
+  "environment": "local macOS SQLite microbenchmark; not model/network/device latency",
+  "list_sessions": 1000,
+  "event_persistence": {
+    "count": 500,
+    "p50_ms": 0.172333999999978,
+    "p95_ms": 0.22904199999999264,
+    "max_ms": 7.481750000000005
+  },
+  "local_list": {
+    "count": 200,
+    "p50_ms": 1.856583999999998,
+    "p95_ms": 1.9892500000000837,
+    "max_ms": 3.2977500000000077
+  },
+  "diagnostics": {
+    "localReads": 200,
+    "recoveryAttempts": 0,
+    "recoveryFailures": 0,
+    "recoveryCompleted": 0
+  }
+}

docs/verification/2026-09-10-ordinary-chat/benchmark.ts

@@ -0,0 +1,14 @@
+import {mkdtempSync,rmSync} from 'node:fs'
+import {join} from 'node:path'
+import {tmpdir} from 'node:os'
+import {ChatCacheStore,ChatCacheCoordinator} from '../../../src/server/chatCache.js'
+const home=mkdtempSync(join(tmpdir(),'ordinary-bench-'))
+const store=new ChatCacheStore(home),coordinator=new ChatCacheCoordinator(store,{request:async()=>{throw new Error('Unexpected upstream read')}} as any)
+for(let i=0;i<1000;i++)store.recordCommand('bench','p',`s${i}`,'session.create',{})
+const events:number[]=[],reads:number[]=[]
+for(let i=0;i<500;i++){const start=performance.now();coordinator.observe('bench','p','s0',{type:'message.delta',payload:{text:'测试'},delivery_id:`e${i}`});events.push(performance.now()-start)}
+coordinator.observe('bench','p','s0',{type:'message.complete',payload:{},delivery_id:'done'})
+for(let i=0;i<200;i++){const start=performance.now();coordinator.readLocal('bench','list','p',undefined,{limit:20});reads.push(performance.now()-start)}
+const stats=(v:number[])=>{v.sort((a,b)=>a-b);return{count:v.length,p50_ms:v[Math.floor(v.length*.5)],p95_ms:v[Math.floor(v.length*.95)],max_ms:v.at(-1)}}
+console.log(JSON.stringify({environment:'local macOS SQLite microbenchmark; not model/network/device latency',list_sessions:1000,event_persistence:stats(events),local_list:stats(reads),diagnostics:coordinator.diagnostics},null,2))
+coordinator.close();store.close();rmSync(home,{recursive:true,force:true})

src/client/api/sessions.ts

@@ -72,6 +72,7 @@ export async function getMessages(
   limit = 150,
   profile?: string,
   forceRefresh = false,
+  view: 'chat' | 'history' = 'chat',
 ): Promise<MessagePage> {
   const safeOffset = Math.max(0, Math.trunc(offset))
   const safeLimit = Math.max(1, Math.min(500, Math.trunc(limit)))
@@ -81,6 +82,7 @@ export async function getMessages(
     order: 'latest',
     include_compacted: true,
     profile,
+    ...(view === 'history' ? {view} : {}),
   })
   const payload = unwrapData(await apiRequest<unknown>(url, forceRefresh ? { headers: { 'X-Yaoyao-Cache': 'bypass' } } : {}))
   const source = record(payload)
@@ -128,6 +130,10 @@ export async function deleteSession(id: string, profile?: string): Promise<void>
   await apiRequest(apiUrl(`/api/app/sessions/${encodeURIComponent(id)}`, { profile }), { method: 'DELETE' })
 }
 
+export async function requestHistorySync(sessionId:string,profile?:string):Promise<void>{
+  await apiRequest(apiUrl(`/api/app/sessions/${encodeURIComponent(sessionId)}/sync`,{profile}),{method:'POST',body:{}})
+}
+
 export async function getSessionUnread(profile?: string): Promise<Record<string, number>> {
   const payload = unwrapData(await apiRequest<unknown>(apiUrl('/api/app/sessions/unread', { profile })))
   const source = record(payload)

src/client/stores/chat.ts

@@ -4,7 +4,7 @@ import type {
   ChatAttachment, ChatMessage, ChatRouteState, JsonValue, ModelOption, RealtimeConnectionState, SessionSummary,
 } from '@shared/types'
 import {
-  deleteSession as deleteSessionApi, getMessages, getSession, getSessions, getSessionUnread, markSessionRead, updateSession,
+  deleteSession as deleteSessionApi, getMessages, getSession, getSessions, getSessionUnread, markSessionRead, updateSession, requestHistorySync,
 } from '@/api/sessions'
 import { getModels } from '@/api/profiles'
 import { ChatRpcSocket, RpcError } from '@/api/realtime'
@@ -303,6 +303,11 @@ export const useChatStore = defineStore('chat', () => {
     if (event.type === 'gateway.ready') return
     const payload = record(event.payload)
     if (event.type === 'sessions.changed') {
+      if(payload.reason==='cache.synced'){
+        const state=activeRouteState.value
+        if(state && state.route.profile===payload.profile && state.route.sessionId===payload.session_id && !state.isLoadingHistory)
+          void loadHistory(state).catch(()=>{})
+      }
       scheduleSessionListRefresh()
       return
     }
@@ -492,7 +497,7 @@ export const useChatStore = defineStore('chat', () => {
       state.loadedMessageCount = cached.messages.length
     }
     try {
-      const page = await getMessages(state.route.sessionId, 0, 150, state.route.profile, forceRefresh)
+      const page = await getMessages(state.route.sessionId, 0, 150, state.route.profile, false, sessionView)
       if (generation !== state.generation) return
       if (forceRefresh && routes[routeKey(state.route.profile, state.route.sessionId)]?.isStreaming) {
         throw new Error('正在回复，请结束后再刷新历史')
@@ -510,6 +515,7 @@ export const useChatStore = defineStore('chat', () => {
       await historyCache.set(scope, state.route.sessionId, { messages: state.messages, total: page.total, savedAt: Date.now() })
     } catch (cause) {
       if (generation === state.generation) state.error = errorMessage(cause)
+      if(cause instanceof ApiError && cause.code==='CHAT_HISTORY_SYNC_PENDING')return
       throw cause
     } finally {
       if (generation === state.generation) state.isLoadingHistory = false
@@ -541,7 +547,9 @@ export const useChatStore = defineStore('chat', () => {
   async function forceRefreshHistory(): Promise<void> {
     const state = activeRouteState.value
     if (!state || state.isLoadingHistory || state.isStreaming || state.isQueued || state.pendingApproval || state.pendingClarification || state.route.sessionId.startsWith('draft-')) return
-    await loadHistory(state, true)
+    if(sessionView==='history'){await loadHistory(state);return}
+    await requestHistorySync(state.route.sessionId,state.route.profile)
+    state.error='历史正在后台补齐，完成后会自动更新'
   }
 
   async function loadOlder(): Promise<void> {
@@ -550,7 +558,7 @@ export const useChatStore = defineStore('chat', () => {
     const generation = state.generation
     state.isLoadingHistory = true
     try {
-      const page = await getMessages(state.route.sessionId, state.loadedMessageCount, 150, state.route.profile)
+      const page = await getMessages(state.route.sessionId, state.loadedMessageCount, 150, state.route.profile, false, sessionView)
       if (generation !== state.generation) return
       state.messages = mergeChatMessages(state.messages, page.messages, 'prepend')
       state.loadedMessageCount += page.returned

src/client/utils/messageReducer.ts

@@ -236,6 +236,15 @@ export function applyChatEvent(state: ChatRouteState, event: RpcEventFrame['para
   if (eventSessionId(event, state.route.sessionId) !== state.route.sessionId) return state
   const next: ChatRouteState = { ...state, error: undefined }
   switch (event.type) {
+    case 'run.peer_user_message':
+    case 'peer.user.message': {
+      const id=string(payload.message_id),clientMessageId=string(payload.client_message_id ?? payload.queue_id)
+      const existing=state.messages.find(m=>m.id===id||m.clientMessageId===clientMessageId)
+      const stage=payload.status==='failed'?'failed':payload.status==='accepted'?'accepted':existing?.stage??'pending'
+      next.messages=mergeChatMessages(state.messages,[{...existing,id,clientMessageId,sessionId:state.route.sessionId,profile:state.route.profile,
+        role:'user',content:string(payload.text ?? payload.content),timestamp:existing?.timestamp??number(payload.timestamp,Date.now()/1000),stage}])
+      break
+    }
     case 'message.start':
     case 'run.started':
       next.messages = updateStreamingMessage(state, payload, 'start')

src/server/app.ts

@@ -258,6 +258,10 @@ export function createApplication(options: ApplicationOptions = {}): Application
   realtime.broker.protectedSession = id => workspace.ownsUpstream(id)
   realtime.broker.onNativeEvent = (owner, profile, storedId, frame) => {
     chatCache?.observe(owner, profile, storedId, frame)
+    if(/^(message\.(start|delta|interim|complete)|reasoning\.|tool\.|run\.(completed|failed))/.test(frame.type)){
+      const id=chatCache?.store.projectedMessageID(owner,profile,storedId)
+      if(id)frame.payload={...frame.payload,message_id:id}
+    }
     if (owner.startsWith('device:')) return
     if (!['message.complete', 'attachment.staged'].includes(frame.type)) return
     const data = frame.payload ?? {}, text = frame.type === 'attachment.staged' ? JSON.stringify(data) : nativeMessageFileText(data)
@@ -439,6 +443,7 @@ export function createApplication(options: ApplicationOptions = {}): Application
     }
   })
   const router = createApiRouter({
+    onChatListChanged: (owner,profile,id) => realtime.broker.publishOwnerSessionsChanged(owner,'local-state',profile,id),
     onServerIdentityChanged: identity => realtime.broker.publishServerIdentity(identity, config.upstream.href),
     onUserAccessChanged: async owner => {
       for (const c of workspace.list<{ id: string }>(owner, 'conversation')) await workspaceRuntime.stopConversation(owner, c.id)
@@ -513,6 +518,7 @@ export function createApplication(options: ApplicationOptions = {}): Application
       push.close()
       uploads.close()
       workspace.close()
+      chatCache?.close()
       chatCache?.store.close()
     },
   }

src/server/chatCache.ts

@@ -1,4 +1,6 @@
-import { createHash } from 'node:crypto'
+import { projectChatEvent } from './chatEventProjection.js'
+import { HttpError } from './errors.js'
+import { createHash, randomUUID } from 'node:crypto'
 import { chmodSync, createReadStream, existsSync, mkdirSync, renameSync, statSync, unlinkSync, writeFileSync } from 'node:fs'
 import { join } from 'node:path'
 import { DatabaseSync } from 'node:sqlite'
@@ -136,6 +138,8 @@ function referencedPaths(value: unknown, output = new Set<string>()): Set<string
 export class ChatCacheStore {
   readonly db: DatabaseSync
   readonly assetsRoot: string
+  private closed = false
+  get isClosed(): boolean { return this.closed }
 
   constructor(home: string, legacySingleUserID?: string) {
     mkdirSync(home, { recursive: true, mode: 0o700 })
@@ -179,6 +183,19 @@ export class ChatCacheStore {
         saved_at INTEGER NOT NULL, PRIMARY KEY(owner,cache_key));
       CREATE TABLE IF NOT EXISTS chat_meta(
         key TEXT PRIMARY KEY, value TEXT NOT NULL);`)
+    this.db.exec(`CREATE TABLE IF NOT EXISTS chat_local_state(
+      owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
+      pinned INTEGER, read_count INTEGER NOT NULL DEFAULT 0,
+      assistant_id TEXT, PRIMARY KEY(owner,profile,session_id));
+      CREATE TABLE IF NOT EXISTS chat_session_aliases(owner TEXT NOT NULL,profile TEXT NOT NULL,alias_id TEXT NOT NULL,session_id TEXT NOT NULL,PRIMARY KEY(owner,profile,alias_id));
+      CREATE TABLE IF NOT EXISTS chat_recovery_jobs(owner TEXT NOT NULL,profile TEXT NOT NULL,session_id TEXT NOT NULL,force INTEGER NOT NULL DEFAULT 0,PRIMARY KEY(owner,profile,session_id));
+      CREATE TABLE IF NOT EXISTS chat_event_receipts(
+      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']]){
+      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 }>
     this.db.exec('BEGIN IMMEDIATE')
     try {
@@ -286,6 +303,8 @@ export class ChatCacheStore {
       this.db.prepare("UPDATE chat_snapshots SET sync_state='stale' WHERE kind='list'").run()
       this.db.prepare(`INSERT INTO chat_meta(key,value) VALUES('ownership-registry-v1','complete')
         ON CONFLICT(key) DO UPDATE SET value=excluded.value`).run()
+    this.db.prepare(`INSERT OR IGNORE INTO chat_local_state(owner,profile,session_id)
+      SELECT owner,profile,session_id FROM chat_sessions WHERE owned_at IS NOT NULL`).run()
       this.db.exec('COMMIT')
     } catch (error) {
       this.db.exec('ROLLBACK')
@@ -293,7 +312,7 @@ export class ChatCacheStore {
     }
   }
 
-  close(): void { this.db.close() }
+  close(): void { this.closed = true; this.db.close() }
 
   stats(): Record<string, number> {
     const count = (table: string) => Number((this.db.prepare(`SELECT COUNT(*) count FROM ${table}`).get() as { count: number }).count)
@@ -315,10 +334,134 @@ export class ChatCacheStore {
   }
 
   localList(owner: string, profile: string, page: { offset?: number; limit?: number; archived?: string }): UpstreamResponse {
-    return this.mergeOwnedSessionsIntoList(owner, profile, {
-      status: 200, headers: new Headers({ 'content-type': 'application/json' }),
-      body: Buffer.from('{"sessions":[]}'),
-    }, page)
+    const offset = Math.max(0, Math.trunc(page.offset ?? 0)), limit = Math.max(1, Math.min(200, Math.trunc(page.limit ?? 100)))
+    const where = `s.owner=? AND s.owned_at IS NOT NULL AND (?='' OR s.profile=?)
+      AND s.source NOT IN ('cron','ios_group','yaoyao_workspace')
+      AND (?='include' OR COALESCE(l.archived,json_extract(s.data,'$.archived'),json_extract(s.data,'$.is_archived'),0)=CASE WHEN ?='only' THEN 1 ELSE 0 END)`
+    const args = [owner, profile, profile, page.archived ?? 'exclude', page.archived ?? 'exclude']
+    const total = Number((this.db.prepare(`SELECT count(*) n FROM chat_sessions s LEFT JOIN chat_local_state l USING(owner,profile,session_id) WHERE ${where}`).get(...args) as {n:number}).n)
+    const rows = this.db.prepare(`SELECT s.profile,s.session_id,s.data,s.sync_state,s.complete,l.title local_title,
+      COALESCE(l.pinned,json_extract(s.data,'$.pinned'),json_extract(s.data,'$.is_pinned'),0) pin
+      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>>
+    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,
+        sync_state:r.sync_state,history_complete:!!r.complete})), total,offset,limit,
+      sync_state:this.needsListMetadata(owner,profile)?'syncing':'current'
+    }))}
+  }
+
+  missingSessions(owner: string, profile: string): Array<{profile:string;session_id:string}> {
+    return (this.db.prepare(`SELECT profile,session_id FROM chat_sessions WHERE owner=? AND owned_at IS NOT NULL
+      AND (?='' OR profile=?) AND (metadata_complete=0 OR complete=0) LIMIT 100`)
+      .all(owner,profile,profile) as {profile:string;session_id:string}[])
+  }
+
+  pins(owner: string, profile: string): string[] {
+    return (this.db.prepare(`SELECT s.session_id 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=?)
+      AND COALESCE(l.pinned,json_extract(s.data,'$.pinned'),json_extract(s.data,'$.is_pinned'),0)=1 ORDER BY s.last_synced_at DESC`)
+      .all(owner,profile,profile) as {session_id:string}[]).map(r=>r.session_id)
+  }
+
+  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 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)}))
+    return {profile,supported:true,total_unread:sessions.reduce((n,r)=>n+r.unread_count,0),sessions}
+  }
+
+  markRead(owner: string, profile: string, id: string, count?: number) {
+    id=this.canonicalSessionID(owner,profile,id)
+    this.requireOwned(owner,profile,id)
+    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)
+    return this.unread(owner,profile).sessions.find(r=>r.session_id===id)!
+  }
+
+  requireOwned(owner: string, profile: string, id: string): void {
+    if (!this.ownsSession(owner,profile,id)) throw new HttpError(404,'会话不存在','session_not_found')
+  }
+
+  patchLocal(owner: string, profile: string, id: string, patch: Record<string,unknown>): void {
+    id=this.canonicalSessionID(owner,profile,id)
+    this.requireOwned(owner,profile,id)
+    this.db.exec('SAVEPOINT local_chat_patch')
+    try {
+      if (typeof patch.pinned==='boolean') this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,pinned) VALUES(?,?,?,?)
+        ON CONFLICT(owner,profile,session_id) DO UPDATE SET pinned=excluded.pinned`).run(owner,profile,id,patch.pinned?1:0)
+      const row = this.localDetail(owner,profile,id)!
+      const data = {...JSON.parse(row.response.body.toString()),...patch}
+      this.db.prepare('UPDATE chat_sessions SET data=? WHERE owner=? AND profile=? AND session_id=?').run(JSON.stringify(data),owner,profile,id)
+      for(const key of ['title','archived'] as const)if(patch[key]!==undefined){
+        const value=key==='archived'?(patch[key]?1:0):String(patch[key])
+        this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,${key}) VALUES(?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET ${key}=excluded.${key}`).run(owner,profile,id,value)
+      }
+      if(typeof patch.title==='string') this.recordAuthoritativeTitle(owner,profile,id,patch.title)
+      this.db.prepare("UPDATE chat_sessions SET sync_state=CASE WHEN complete=1 THEN 'current' ELSE sync_state END WHERE owner=? AND profile=? AND session_id=?").run(owner,profile,id)
+      this.db.exec('RELEASE local_chat_patch')
+    } catch(e) {this.db.exec('ROLLBACK TO local_chat_patch; RELEASE local_chat_patch');throw e}
+  }
+
+  private writeLiveMessage(owner:string,profile:string,id:string,message:Record<string,any>): void {
+    const prior = this.db.prepare('SELECT position FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?').get(owner,profile,id,message.id) as {position:number}|undefined
+    const row = this.db.prepare('SELECT message_total,complete,data FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,id) as any
+    const extent = Number((this.db.prepare('SELECT COALESCE(MAX(position)+1,0) n FROM chat_messages WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,id) as any).n)
+    const position = prior?.position ?? Math.max(extent,Number(row.message_total??0))
+    const data=JSON.stringify(message)
+    this.db.prepare(`INSERT INTO chat_messages(owner,profile,session_id,message_id,position,data,content_hash) VALUES(?,?,?,?,?,?,?)
+      ON CONFLICT(owner,profile,session_id,message_id) DO UPDATE SET data=excluded.data,content_hash=excluded.content_hash`).run(owner,profile,id,message.id,position,data,createHash('sha256').update(data).digest('hex'))
+    const total=Math.max(Number(row.message_total??0),position+1), summary=JSON.parse(row.data)
+    const now=Date.now()
+    summary.message_count=total; summary.last_active=now/1000
+    if (typeof message.content==='string' && message.content.trim()) summary.preview=message.content.slice(0,160)
+    this.db.prepare('UPDATE chat_sessions SET data=?,message_total=?,last_synced_at=? WHERE owner=? AND profile=? AND session_id=?').run(JSON.stringify(summary),total,now,owner,profile,id)
+    for(const path of referencedPaths(message)) this.db.prepare(`INSERT OR IGNORE INTO chat_attachments(owner,profile,session_id,source_path,state,updated_at) VALUES(?,?,?,?,'pending',?)`).run(owner,profile,id,path,now)
+  }
+
+  projectedMessageID(owner:string,profile:string,id:string):string|undefined {
+    return (this.db.prepare(`SELECT message_id FROM chat_messages WHERE owner=? AND profile=? AND session_id=?
+      AND json_extract(data,'$.role')='assistant' ORDER BY position DESC LIMIT 1`).get(owner,profile,id) as {message_id:string}|undefined)?.message_id
+  }
+
+  private projectEvent(owner:string,profile:string,id:string,type:string,p:Record<string,any>,eventID:string): void {
+    if(!this.ownsSession(owner,profile,id))return
+    if(type==='command.confirmed'||type==='command.rejected'){
+      const userID=`user:${p.delivery_id}`
+      const row=this.db.prepare('SELECT data FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?').get(owner,profile,id,userID) as any
+      if(row)this.writeLiveMessage(owner,profile,id,{...JSON.parse(row.data),status:type==='command.confirmed'?'complete':'failed',...(p.error?{error:p.error}:{})})
+      this.db.prepare("UPDATE chat_sessions SET sync_state=CASE WHEN sync_state='gap' THEN 'gap' WHEN complete=1 THEN 'current' ELSE 'stale' END WHERE owner=? AND profile=? AND session_id=?").run(owner,profile,id)
+      return
+    }
+    if (type==='message.user' || type==='peer_user_message') {
+      this.writeLiveMessage(owner,profile,id,{...p,id:String(p.message_id??p.id??eventID),role:'user',content:p.text??p.content??'',timestamp:Date.now()/1000});return
+    }
+    const state=this.db.prepare('SELECT assistant_id FROM chat_local_state WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,id) as any
+    const last=state?.assistant_id?this.db.prepare('SELECT data FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?').get(owner,profile,id,state.assistant_id) as any:undefined
+    const previous=last?JSON.parse(last.data):undefined
+    // A new run must never append to the assistant row from the preceding turn.
+    if(['message.complete','run.completed'].includes(type) && !previous && !p.text && !p.content && !p.message){
+      this.db.prepare("UPDATE chat_sessions SET sync_state='gap' WHERE owner=? AND profile=? AND session_id=?").run(owner,profile,id);return
+    }
+    const startsNew=type==='message.start' && previous?.status!=='streaming'
+    const message=projectChatEvent(type,p,startsNew?undefined:previous,`event:${eventID}`,Date.now())
+    if(message){
+      if(previous && previous.id !== message.id && !startsNew){
+        const exists=this.db.prepare('SELECT 1 FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?').get(owner,profile,id,message.id)
+        if(!exists)this.db.prepare('UPDATE chat_messages SET message_id=? WHERE owner=? AND profile=? AND session_id=? AND message_id=?').run(message.id,owner,profile,id,previous.id)
+        else {this.db.prepare("UPDATE chat_sessions SET sync_state='gap' WHERE owner=? AND profile=? AND session_id=?").run(owner,profile,id);return}
+      }
+      this.writeLiveMessage(owner,profile,id,message)
+      this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,assistant_id) VALUES(?,?,?,?)
+        ON CONFLICT(owner,profile,session_id) DO UPDATE SET assistant_id=excluded.assistant_id`).run(owner,profile,id,type==='message.interim'?null:message.id)
+      this.db.prepare(`UPDATE chat_sessions SET sync_state=CASE WHEN sync_state='gap' THEN 'gap' WHEN complete=1 THEN 'current' ELSE 'stale' END
+        WHERE owner=? AND profile=? AND session_id=?`).run(owner,profile,id)
+    }
   }
 
   needsListMetadata(owner: string, profile: string): boolean {
@@ -331,8 +474,9 @@ export class ChatCacheStore {
       WHERE owner=? AND profile=? AND session_id=? AND owned_at IS NOT NULL`)
       .get(owner, profile, sessionID) as { data: string; sync_state: ChatCacheState } | undefined
     if (!row) return undefined
+    const local=this.db.prepare('SELECT title,archived,pinned FROM chat_local_state WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,sessionID) as any
     return { response: { status: 200, headers: new Headers({ 'content-type': 'application/json' }),
-      body: Buffer.from(row.data) }, state: row.sync_state }
+      body: Buffer.from(JSON.stringify({...JSON.parse(row.data),...(local?.title!=null?{title:local.title}:{}),...(local?.archived!=null?{archived:!!local.archived}:{}),owned:true,pinned:local?.pinned!=null?!!local.pinned:bool(JSON.parse(row.data).pinned),sync_state:row.sync_state})) }, state: row.sync_state }
   }
 
   revision(owner: string, profile: string, sessionID: string): number {
@@ -381,8 +525,29 @@ export class ChatCacheStore {
     this.db.exec('SAVEPOINT history_sync')
     try {
       if (force) this.db.prepare('DELETE FROM chat_messages WHERE owner=? AND profile=? AND session_id=?').run(owner, profile, sessionID)
+      // Recovery promotes locally generated stream identities to upstream row IDs.
+      // Remove only replaced rows inside the verified window; keep older coverage.
+      if(!force){
+        const windows=pages.filter(p=>p.kind==='messages').map(p=>parseResponse(p.response)).filter(Boolean) as Record<string,any>[]
+        for(const window of windows){
+          const pagination=window.pagination??{},messages=window.messages as Record<string,any>[]
+          if(!Array.isArray(messages)||!Number.isSafeInteger(pagination.total))continue
+          const end=pagination.total-Number(pagination.offset??0),start=end-messages.length
+          const ids=new Set(messages.map((m,index)=>messageID(m,String(start+index))))
+          const existing=this.db.prepare('SELECT message_id,data FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND position>=? AND position<?').all(owner,profile,sessionID,start,end) as {message_id:string;data:string}[]
+          for(const row of existing){
+            if(ids.has(row.message_id))continue
+            const value=JSON.parse(row.data)
+            if(value.role==='user'&&value.status==='pending'&&!messages.some(m=>m.role==='user'&&m.content===value.content))
+              throw new Error('Unconfirmed submission must be reconciled before history replacement')
+            this.db.prepare('DELETE FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND message_id=?').run(owner,profile,sessionID,row.message_id)
+          }
+        }
+      }
       this.db.prepare("DELETE FROM chat_snapshots WHERE owner=? AND profile=? AND session_id=? AND kind='messages'").run(owner, profile, sessionID)
       for (const page of pages) this.putSnapshot(owner, page.key, page.kind, profile, sessionID, page.response, false, page.startedAt)
+      const latest=this.db.prepare("SELECT message_id FROM chat_messages WHERE owner=? AND profile=? AND session_id=? AND json_extract(data,'$.role')='assistant' ORDER BY position DESC LIMIT 1").get(owner,profile,sessionID) as {message_id:string}|undefined
+      if(latest)this.db.prepare('UPDATE chat_local_state SET assistant_id=? WHERE owner=? AND profile=? AND session_id=?').run(latest.message_id,owner,profile,sessionID)
       this.db.exec('RELEASE history_sync')
       return true
     } catch (error) {
@@ -407,7 +572,7 @@ export class ChatCacheStore {
     if (messages.length !== end - start) return undefined // Fetch only an actually missing page.
     const body = Buffer.from(JSON.stringify({
       session_id: sessionID,
-      session: JSON.parse(String(session.data)),
+      session: JSON.parse(this.localDetail(owner,profile,sessionID)!.response.body.toString()),
       messages,
       pagination: {
         total,
@@ -427,20 +592,23 @@ export class ChatCacheStore {
     const rawSeq = Number(frame.seq ?? frame.sequence ?? 0)
     const type = String(frame.type ?? 'unknown')
     const payload = JSON.stringify(frame.payload ?? {})
-    const seq = Number.isSafeInteger(rawSeq) && rawSeq > 0
-      ? rawSeq
-      : Number.parseInt(createHash('sha256').update(`${type}:${payload}`).digest('hex').slice(0, 12), 16)
+    const eventID = String(frame.delivery_id ?? (rawSeq > 0 ? `${frame.epoch ?? ''}:${rawSeq}:${type}` : randomUUID()))
+    const seq = Number.parseInt(createHash('sha256').update(eventID).digest('hex').slice(0,12),16)
     const now = Date.now()
+    this.db.exec('SAVEPOINT live_event')
+    try {
+    const receipt=this.db.prepare('INSERT OR IGNORE INTO chat_event_receipts VALUES(?,?,?,?)').run(owner,profile,sessionID,eventID)
+    if(!receipt.changes){this.db.exec('RELEASE live_event');return false}
     const previous = this.db.prepare(`SELECT last_event_seq FROM chat_sessions
       WHERE owner=? AND profile=? AND session_id=?`).get(owner, profile, sessionID) as { last_event_seq?: number } | undefined
     const lastSeen = Number(previous?.last_event_seq ?? 0)
     const inserted = this.db.prepare(`INSERT OR IGNORE INTO chat_events(owner,profile,session_id,event_seq,event_type,payload,received_at)
       VALUES(?,?,?,?,?,?,?)`).run(owner, profile, sessionID, Number.isSafeInteger(seq) ? seq : 0, type, payload, now)
-    if (!inserted.changes) return false
+    if (!inserted.changes) {this.db.exec('RELEASE live_event');return false}
     const state: ChatCacheState = type === 'gateway.reset' || type.includes('reset')
-      || (rawSeq > 0 && lastSeen > 0 && rawSeq > lastSeen + 1) ? 'gap' : 'stale'
+      || (!frame.epoch && rawSeq > 0 && lastSeen > 0 && rawSeq > lastSeen + 1) ? 'gap' : 'stale'
     const metadataOnly = ['session.info', 'session.title', 'session.title.updated', 'session.usage', 'gateway.ready'].includes(type) && state !== 'gap'
-    this.db.prepare(`UPDATE chat_sessions SET sync_state=CASE WHEN ? THEN sync_state ELSE ? END,
+    this.db.prepare(`UPDATE chat_sessions SET sync_state=CASE WHEN ? THEN sync_state WHEN sync_state='gap' THEN 'gap' ELSE ? END,
       sync_revision=sync_revision+?,last_event_seq=MAX(last_event_seq,?),last_synced_at=?
       WHERE owner=? AND profile=? AND session_id=?`).run(metadataOnly ? 1 : 0, state, metadataOnly ? 0 : 1, rawSeq > 0 ? rawSeq : lastSeen, now, owner, profile, sessionID)
     if (type === 'session.info') {
@@ -465,10 +633,39 @@ export class ChatCacheStore {
       VALUES(?,?,?,?,?) ON CONFLICT(owner,profile) DO UPDATE SET event_cursor=MAX(event_cursor,excluded.event_cursor),
       sync_state=excluded.sync_state,last_synced_at=excluded.last_synced_at`)
       .run(owner, profile, rawSeq > 0 ? rawSeq : lastSeen, state, now)
+    this.projectEvent(owner,profile,sessionID,type,object(frame.payload)??{},eventID)
+    this.db.exec('RELEASE live_event')
     return true
+    }catch(e){this.db.exec('ROLLBACK TO live_event; RELEASE live_event');throw e}
+  }
+
+  canonicalSessionID(owner:string,profile:string,id:string):string {
+    let current=id
+    const seen=new Set<string>()
+    for(let n=0;n<32&&!seen.has(current);n++){
+      seen.add(current)
+      const row=this.db.prepare('SELECT session_id FROM chat_session_aliases WHERE owner=? AND profile=? AND alias_id=?').get(owner,profile,current) as {session_id:string}|undefined
+      if(!row)break
+      current=row.session_id
+    }
+    return current
+  }
+  private migrateRuntimeSession(owner:string,profile:string,id:string,runtime:string):void {
+    const old=this.db.prepare('SELECT session_id FROM chat_sessions WHERE owner=? AND profile=? AND runtime_id=? AND session_id<>? AND owned_at IS NOT NULL').get(owner,profile,runtime,id) as {session_id:string}|undefined
+    if(!old||this.ownsSession(owner,profile,id))return
+    this.db.exec('SAVEPOINT migrate_live_route')
+    try{
+      for(const table of ['chat_sessions','chat_messages','chat_events','chat_snapshots','chat_attachments','chat_local_state','chat_event_receipts','chat_recovery_jobs'])
+        this.db.prepare(`UPDATE ${table} SET session_id=? WHERE owner=? AND profile=? AND session_id=?`).run(id,owner,profile,old.session_id)
+      this.db.prepare('INSERT OR REPLACE INTO chat_session_aliases VALUES(?,?,?,?)').run(owner,profile,old.session_id,id)
+      this.db.prepare('UPDATE chat_session_aliases SET session_id=? WHERE owner=? AND profile=? AND session_id=?').run(id,owner,profile,old.session_id)
+      this.db.prepare("UPDATE chat_sessions SET data=json_set(data,'$.id',?) WHERE owner=? AND profile=? AND session_id=?").run(id,owner,profile,id)
+      this.db.exec('RELEASE migrate_live_route')
+    }catch(e){this.db.exec('ROLLBACK TO migrate_live_route; RELEASE migrate_live_route');throw e}
   }
 
   recordRoute(owner: string, profile: string, sessionID: string, runtimeID: string): void {
+    this.migrateRuntimeSession(owner,profile,sessionID,runtimeID)
     const now = Date.now()
     const summary = { id: sessionID, profile, source: 'unknown', title: '新对话', started_at: now, last_active: now }
     this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,runtime_id,source,data,sync_state,complete,last_synced_at,owned_at,ownership_evidence)
@@ -488,11 +685,18 @@ export class ChatCacheStore {
     this.db.prepare(`INSERT INTO chat_sessions(owner,profile,session_id,source,data,sync_state,complete,last_synced_at,owned_at,ownership_evidence)
       VALUES(?,?,?,?,?,?,?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET
       source=CASE WHEN excluded.source='unknown' THEN chat_sessions.source ELSE excluded.source END,
-      sync_state='stale',sync_revision=chat_sessions.sync_revision+1,last_synced_at=excluded.last_synced_at,
+      sync_state=CASE WHEN chat_sessions.sync_state='gap' THEN 'gap' ELSE 'stale' END,sync_revision=chat_sessions.sync_revision+1,last_synced_at=excluded.last_synced_at,
       owned_at=COALESCE(chat_sessions.owned_at,excluded.owned_at),
       ownership_evidence=COALESCE(chat_sessions.ownership_evidence,excluded.ownership_evidence)`)
       .run(owner, profile, sessionID, source, JSON.stringify(summary), 'stale', 0, now,
         establishesOwnership ? now : null, establishesOwnership ? `command:${method}` : null)
+    if(method==='session.create') this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,read_initialized) VALUES(?,?,?,1) ON CONFLICT(owner,profile,session_id) DO UPDATE SET read_initialized=1`).run(owner,profile,sessionID)
+    if(method==='session.create') this.db.prepare(`UPDATE chat_sessions SET complete=1,metadata_complete=1,message_total=0,sync_state='current' WHERE owner=? AND profile=? AND session_id=? AND message_total IS NULL`).run(owner,profile,sessionID)
+    if((method==='prompt.submit'||method==='session.steer') && params._delivery_id){
+      const userID=`user:${params._delivery_id}`
+      this.writeLiveMessage(owner,profile,sessionID,{id:userID,role:'user',content:params.text??params.input??'',timestamp:now/1000,status:'pending',client_message_id:String(params._delivery_id).replace(/^(web|ios):prompt:/,'')})
+      this.db.prepare(`INSERT INTO chat_local_state(owner,profile,session_id,assistant_id) VALUES(?,?,?,NULL) ON CONFLICT(owner,profile,session_id) DO UPDATE SET assistant_id=NULL`).run(owner,profile,sessionID)
+    }
     const payload = JSON.stringify(params)
     const seq = Number.parseInt(createHash('sha256').update(`${method}:${payload}`).digest('hex').slice(0, 12), 16)
     this.db.prepare(`INSERT OR IGNORE INTO chat_events(owner,profile,session_id,event_seq,event_type,payload,received_at)
@@ -525,7 +729,7 @@ export class ChatCacheStore {
       .all(owner, profile, sessionID) as Array<{ local_path: string }>
     this.db.exec('BEGIN IMMEDIATE')
     try {
-      for (const table of ['chat_messages', 'chat_events', 'chat_attachments', 'chat_sessions']) {
+      for (const table of ['chat_messages', 'chat_events', 'chat_attachments', 'chat_local_state', 'chat_event_receipts', 'chat_recovery_jobs', 'chat_session_aliases', 'chat_sessions']) {
         this.db.prepare(`DELETE FROM ${table} WHERE owner=? AND profile=? AND session_id=?`).run(owner, profile, sessionID)
       }
       this.db.prepare("DELETE FROM chat_snapshots WHERE owner=? AND ((profile=? AND session_id=?) OR kind='list')")
@@ -550,6 +754,7 @@ export class ChatCacheStore {
   }
 
   ownsSession(owner: string, profile: string, sessionID: string): boolean {
+    sessionID=this.canonicalSessionID(owner,profile,sessionID)
     return Boolean(this.db.prepare(
       'SELECT 1 ok FROM chat_sessions WHERE owner=? AND profile=? AND session_id=? AND owned_at IS NOT NULL',
     ).get(owner, profile, sessionID))
@@ -562,7 +767,8 @@ export class ChatCacheStore {
     const row = this.db.prepare(`SELECT authoritative_title FROM chat_sessions
       WHERE owner=? AND profile=? AND session_id=? AND owned_at IS NOT NULL`)
       .get(owner, profile, sessionID) as { authoritative_title?: string | null } | undefined
-    const authoritativeTitle = String(row?.authoritative_title ?? '').trim()
+    const local=this.db.prepare('SELECT title FROM chat_local_state WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,sessionID) as {title?:string}|undefined
+    const authoritativeTitle = String(local?.title ?? row?.authoritative_title ?? '').trim()
     return {
       owned: Boolean(row),
       ...(authoritativeTitle ? { authoritativeTitle } : {}),
@@ -717,7 +923,8 @@ export class ChatCacheStore {
     const row = this.db.prepare(`SELECT authoritative_title FROM chat_sessions
       WHERE owner=? AND profile=? AND session_id=? AND owned_at IS NOT NULL`).get(owner, profile, id) as { authoritative_title?: string | null } | undefined
     const owned = Boolean(row)
-    const title = String(row?.authoritative_title ?? '').trim()
+    const local=this.db.prepare('SELECT title FROM chat_local_state WHERE owner=? AND profile=? AND session_id=?').get(owner,profile,id) as {title?:string}|undefined
+    const title = String(local?.title ?? row?.authoritative_title ?? '').trim()
     const body = session
       ? Buffer.from(JSON.stringify({
           ...payload,
@@ -928,7 +1135,92 @@ export class ChatCacheCoordinator {
   private reconciling = new Map<string, Promise<void>>()
   onSynchronized: (owner: string, profile: string, sessionID: string) => void = () => {}
   constructor(readonly store: ChatCacheStore, readonly upstream: UpstreamServiceSession,
-    readonly mode: ChatCacheMode = 'prefer-local') {}
+    readonly mode: ChatCacheMode = 'prefer-local') {
+      const timer=setTimeout(()=>{
+        this.retryTimers.delete(timer)
+        if(this.store.isClosed||this.stopped)return
+        void this.restoreReadPositions()
+        for(const row of this.store.db.prepare('SELECT * FROM chat_recovery_jobs').all() as any[])this.schedule(row.owner,row.profile,row.session_id,!!row.force)
+      },0);timer.unref();this.retryTimers.add(timer)
+    }
+
+  readonly diagnostics = {localReads:0, recoveryAttempts:0, recoveryFailures:0, recoveryCompleted:0}
+  private async restoreReadPositions():Promise<void>{
+    if(this.stopped||this.store.isClosed)return
+    const scopes=this.store.db.prepare('SELECT DISTINCT owner,profile FROM chat_local_state WHERE read_initialized=0').all() as {owner:string;profile:string}[]
+    let failed=false
+    for(let i=0;i<scopes.length;i+=2)await Promise.all(scopes.slice(i,i+2).map(async({owner,profile})=>{
+      if(this.stopped||this.store.isClosed)return
+      try{
+        const response=await this.upstream.request('/api/session-unread',{search:new URLSearchParams({profile})})
+        if(this.stopped||this.store.isClosed)return
+        if(response.status>=500)throw new Error('Read-position import unavailable')
+        const payload=parseResponse(response),rows=Array.isArray(payload?.sessions)?payload.sessions:[]
+        this.store.db.exec('SAVEPOINT read_position_import')
+        try{
+          for(const raw of rows){const r=object(raw);if(!r)continue;const id=String(r.session_id??r.sessionId??'')
+            const count=Math.max(0,Number(r.read_message_count??r.readMessageCount??0)||0)
+            this.store.db.prepare('UPDATE chat_local_state SET read_count=?,read_initialized=1 WHERE owner=? AND profile=? AND session_id=? AND read_initialized=0').run(count,owner,profile,id)
+          }
+          this.store.db.prepare(`UPDATE chat_local_state SET read_count=COALESCE((SELECT message_total FROM chat_sessions s WHERE s.owner=chat_local_state.owner AND s.profile=chat_local_state.profile AND s.session_id=chat_local_state.session_id),0),read_initialized=1 WHERE owner=? AND profile=? AND read_initialized=0`).run(owner,profile)
+          this.store.db.exec('RELEASE read_position_import')
+        }catch(e){this.store.db.exec('ROLLBACK TO read_position_import; RELEASE read_position_import');throw e}
+      }catch{failed=true}
+    }))
+    if(failed&&!this.stopped&&!this.store.isClosed){const timer=setTimeout(()=>{this.retryTimers.delete(timer);void this.restoreReadPositions()},30_000);timer.unref();this.retryTimers.add(timer)}
+  }
+  private jobs = new Map<string,{owner:string;profile:string;id:string;force:boolean;attempt:number}>()
+  private activeJobs = new Set<string>()
+  private retryTimers = new Set<ReturnType<typeof setTimeout>>()
+  private stopped = false
+  close(): void {this.stopped=true;for(const timer of this.retryTimers)clearTimeout(timer);this.retryTimers.clear();this.jobs.clear()}
+
+  schedule(owner:string,profile:string,id:string,force=false): void {
+    if(this.stopped||this.store.isClosed||!this.store.ownsSession(owner,profile,id))return
+    const key=JSON.stringify([owner,profile,id]),previous=this.jobs.get(key)
+    if(previous){previous.force ||= force;if(force)this.store.db.prepare('UPDATE chat_recovery_jobs SET force=1 WHERE owner=? AND profile=? AND session_id=?').run(owner,profile,id);return}
+    this.store.db.prepare('INSERT INTO chat_recovery_jobs VALUES(?,?,?,?) ON CONFLICT(owner,profile,session_id) DO UPDATE SET force=MAX(force,excluded.force)').run(owner,profile,id,force?1:0)
+    this.jobs.set(key,{owner,profile,id,force,attempt:0})
+    const timer=setTimeout(()=>{this.retryTimers.delete(timer);this.drain()},0);timer.unref();this.retryTimers.add(timer)
+  }
+  private drain(): void {
+    if(this.stopped||this.store.isClosed)return
+    for(const [key,job] of this.jobs){
+      if(this.activeJobs.size>=2)break
+      if(this.activeJobs.has(key)||job.attempt<0)continue
+      this.activeJobs.add(key)
+      this.diagnostics.recoveryAttempts++
+      const force=job.force;job.force=false
+      void this.reconcile(job.owner,job.profile,job.id,force).then(()=>{
+        if(this.stopped||this.store.isClosed)return
+        if(job.force){job.attempt=0;return}
+        this.diagnostics.recoveryCompleted++
+        this.jobs.delete(key)
+        this.store.db.prepare('DELETE FROM chat_recovery_jobs WHERE owner=? AND profile=? AND session_id=?').run(job.owner,job.profile,job.id)
+      }).catch(()=>{
+        if(this.stopped||this.store.isClosed)return
+        this.diagnostics.recoveryFailures++
+        job.force ||= force
+        const attempt=Math.abs(job.attempt)+1;job.attempt=-attempt
+        const timer=setTimeout(()=>{this.retryTimers.delete(timer);job.attempt=attempt;this.drain()},Math.min(30_000,1000*2**Math.min(attempt-1,5)))
+        timer.unref();this.retryTimers.add(timer)
+      }).finally(()=>{this.activeJobs.delete(key);this.drain()})
+    }
+  }
+  readLocal(owner:string,kind:'list'|'detail'|'messages',profile:string,id:string|undefined,
+    page:{offset?:number;limit?:number;archived?:string},force=false): {response:UpstreamResponse;source:ChatCacheSource;state:ChatCacheState} {
+    this.diagnostics.localReads++
+    if(id)id=this.store.canonicalSessionID(owner,profile,id)
+    if(kind==='list'){
+      for(const row of this.store.missingSessions(owner,profile)) this.schedule(owner,row.profile,row.session_id,force)
+      return {response:this.store.localList(owner,profile,page),source:'local',state:this.store.needsListMetadata(owner,profile)?'syncing':'current'}
+    }
+    this.store.requireOwned(owner,profile,id!)
+    const local=kind==='detail'?this.store.localDetail(owner,profile,id!):this.store.messagePage(owner,profile,id!,page.offset??0,page.limit??100)
+    if(force||!local||local.state==='gap')this.schedule(owner,profile,id!,force)
+    if(!local)throw new HttpError(409,'历史正在后台补齐，请稍后重试','CHAT_HISTORY_SYNC_PENDING')
+    return {response:local.response,source:'local',state:this.jobs.has(JSON.stringify([owner,profile,id]))?'syncing':local.state}
+  }
 
   async read(owner: string, key: string, kind: string, profile: string, sessionID: string | undefined,
     load: () => Promise<UpstreamResponse>, ownedList = false,
@@ -994,8 +1286,9 @@ export class ChatCacheCoordinator {
 
   observe(owner: string, profile: string, sessionID: string, frame: Record<string, unknown>): void {
     if (!this.store.recordEvent(owner, profile, sessionID, frame)) return
-    if (['message.complete', 'gateway.reset'].includes(String(frame.type))) {
-      void this.reconcile(owner, profile, sessionID).catch(() => {})
+    if (String(frame.type).includes('reset') || this.store.localDetail(owner,profile,sessionID)?.state==='gap'
+      || (['message.complete','run.completed'].includes(String(frame.type)) && this.store.localDetail(owner,profile,sessionID)?.state!=='current')) {
+      this.schedule(owner, profile, sessionID)
     }
   }
 
@@ -1013,11 +1306,13 @@ export class ChatCacheCoordinator {
   }
 
   async reconcile(owner: string, profile: string, sessionID: string, force = false): Promise<void> {
-    if (this.mode === 'upstream-only' || !this.store.ownsSession(owner, profile, sessionID)) return
+    if(this.store.isClosed)return
+    sessionID=this.store.canonicalSessionID(owner,profile,sessionID)
+    if (this.stopped || this.store.isClosed || !this.store.ownsSession(owner, profile, sessionID)) return
     const key = JSON.stringify([owner, profile, sessionID])
     const pending = this.reconciling.get(key)
     if (pending) {
-      await pending
+      await pending.catch(() => {})
       if (force || this.store.localDetail(owner, profile, sessionID)?.state !== 'current') {
         return this.reconcile(owner, profile, sessionID, force)
       }
@@ -1030,6 +1325,7 @@ export class ChatCacheCoordinator {
       const detailPath = `/api/sessions/${encodeURIComponent(sessionID)}`
       const detailStartedAt = Date.now()
       const detail = await this.upstream.request(detailPath, { search: new URLSearchParams({ profile }), cache: 'reload' })
+      if (this.stopped || this.store.isClosed) return
       if (detail.status !== 200) throw new Error('Unable to synchronize conversation metadata')
       const detailPayload = parseResponse(detail) ?? {}
       const detailSummary = object(detailPayload.session) ?? detailPayload
@@ -1045,6 +1341,7 @@ export class ChatCacheCoordinator {
         const search = new URLSearchParams({ profile, offset: String(offset), limit: '100', order: 'latest', include_compacted: 'true' })
         const startedAt = Date.now()
         const response = await this.upstream.request(`${detailPath}/messages`, { search, cache: 'reload' })
+        if (this.stopped || this.store.isClosed) return
         const payload = parseResponse(response)
         if (response.status !== 200 || !Array.isArray(payload?.messages)) throw new Error('Unable to synchronize conversation messages')
         const messages = payload.messages.flatMap(value => object(value) ? [object(value)!] : [])
@@ -1093,7 +1390,7 @@ export class ChatCacheCoordinator {
       if (!this.store.applySync(owner, profile, sessionID, revision, pages, force)) {
         // Preserve the invalidation when a newer event arrived during the read.
         // Its completion joins this flight and schedules the next tail read.
-        if (force) throw new Error('Conversation changed during refresh; retry after the reply completes')
+        throw new Error('Conversation changed during refresh; retry after the reply completes')
       } else {
         this.onSynchronized(owner, profile, sessionID)
       }

src/server/chatEventProjection.ts

@@ -0,0 +1,33 @@
+/** Shared durable representation of an ordinary Hermes turn. No client timers or UI state. */
+export type ChatRecord = Record<string, any>
+const text = (p: ChatRecord) => String(p.delta ?? p.text_delta ?? p.content_delta ?? p.text ?? '')
+export function projectChatEvent(type: string, payload: ChatRecord, previous: ChatRecord | undefined,
+  fallbackID: string, now: number): ChatRecord | undefined {
+  const supported = /^(message\.(start|delta|interim|complete)|reasoning\.(delta|complete)|tool\.|run\.(completed|failed)|error$)/.test(type)
+  if (!supported) return undefined
+  const supplied = payload.message && typeof payload.message === 'object' ? payload.message : {}
+  const id = String(supplied.id ?? payload.message_id ?? previous?.id ?? fallbackID)
+  const message: ChatRecord = { role: 'assistant', content: '', reasoning: '', timestamp: now / 1000,
+    ...previous, ...supplied, id }
+  if (type === 'message.delta') message.content = String(message.content ?? '') + text(payload)
+  if (type === 'message.interim') message.status = 'complete'
+  if (type === 'reasoning.delta') message.reasoning = String(message.reasoning ?? '') + text(payload)
+  if (type.startsWith('tool.')) {
+    const tools: ChatRecord[] = Array.isArray(message.tool_calls) ? [...message.tool_calls] : []
+    const toolID = String(payload.tool_id ?? payload.tool_call_id ?? payload.id ?? '')
+    const index = tools.findIndex(t => t.id === toolID)
+    const value = { ...(index >= 0 ? tools[index] : {}), ...payload, id: toolID, status: type }
+    if (index >= 0) tools[index] = value; else tools.push(value)
+    message.tool_calls = tools
+  }
+  if (type === 'message.complete' || type === 'run.completed') {
+    if (typeof payload.output === 'string') message.content = payload.output
+    else if (typeof payload.content === 'string') message.content = payload.content
+    else if (typeof payload.text === 'string') message.content = payload.text
+    message.status = payload.error || ['error','failed'].includes(payload.status) ? 'failed' : 'complete'
+    if(payload.error)message.error=payload.error
+  } 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'
+  return message
+}

src/server/realtimeBroker.ts

@@ -156,7 +156,7 @@ export class RealtimeBroker {
     const existing = this.receipts.reserve(c.principal.key, requestId, fingerprint)
     const key = `${c.principal.key}:${requestId}`
     if (existing) return this.commands.get(key) ?? Promise.resolve(this.completed.get(key)?.receipt ?? existing)
-    const work = this.perform(c, frame).then(response => {
+    const work = this.perform(c, frame, requestId).then(response => {
       const state = response.error ? 'rejected' : 'confirmed'
       this.receipts.finish(c.principal.key, requestId, state, response)
       return { requestId, state, response } as CommandReceipt
@@ -178,7 +178,7 @@ export class RealtimeBroker {
     return work
   }
 
-  private async perform(c: RealtimeChannel, frame: Frame): Promise<Frame> {
+  private async perform(c: RealtimeChannel, frame: Frame, deliveryID: string): Promise<Frame> {
     const u = this.upstream(c.principal)
     const p = frame.params as Frame
     if (p.session_id && this.protectedSession(String(p.session_id))) throw new HttpError(404, '会话不存在', 'session_not_found')
@@ -241,16 +241,28 @@ export class RealtimeBroker {
         }
         if (!c.principal.valid()) throw new HttpError(401, 'Authentication expired', 'authentication_required')
       }
+      const submittedAt = this.now()/1000
       const previouslyActive = route?.active ?? false
       if (route && (method === 'prompt.submit' || method === 'session.steer')) route.active = true
       const nativeOwner = this.nativeOwner(c.principal)
       if (nativeOwner && route && ['prompt.submit', 'session.steer'].includes(method)) {
-        this.onNativeCommand(nativeOwner, route.profile, route.stored, method, p)
+        this.onNativeCommand(nativeOwner, route.profile, route.stored, method, {...p,_delivery_id:deliveryID})
+        const peer={jsonrpc:'2.0',method:'event',params:{type:'run.peer_user_message',session_id:route.runtime,profile:route.profile,
+          payload:{message_id:`user:${deliveryID}`,queue_id:deliveryID.replace(/^(web|ios):prompt:/,''),client_message_id:deliveryID.replace(/^(web|ios):prompt:/,''),text:p.text??'',timestamp:submittedAt}}}
+        for(const channel of this.channels.values())if(channel.routes.has(this.routeKey(route.profile,route.stored))&&channel.principal.valid()&&this.nativeOwner(channel.principal)===nativeOwner)this.emit(channel,'frame',peer)
       }
       const activity: RealtimeActivity = { kind: 'command', name: method, sessionId: route?.stored }
       this.onActivity(activity)
       const response = await this.rpc(u, method, p, f => c.principal.observeEvent?.(JSON.stringify(f)), f => c.principal.observeCommand?.(JSON.stringify(f)))
         .finally(() => this.onActivity(activity))
+      if (nativeOwner && route && ['prompt.submit','session.steer'].includes(method)) {
+        this.onNativeEvent(nativeOwner,route.profile,route.stored,{type:response.error?'command.rejected':'command.confirmed',
+          delivery_id:`command-result:${deliveryID}`,payload:{delivery_id:deliveryID,error:response.error?.message}})
+        const peer={jsonrpc:'2.0',method:'event',params:{type:'run.peer_user_message',session_id:route.runtime,profile:route.profile,
+          payload:{message_id:`user:${deliveryID}`,queue_id:deliveryID.replace(/^(web|ios):prompt:/,''),client_message_id:deliveryID.replace(/^(web|ios):prompt:/,''),text:p.text??'',status:response.error?'failed':'accepted',timestamp:submittedAt}}}
+        for(const channel of this.channels.values())if(channel.routes.has(this.routeKey(route.profile,route.stored))&&channel.principal.valid()&&this.nativeOwner(channel.principal)===nativeOwner)this.emit(channel,'frame',peer)
+
+      }
       if (response.error && route && (method === 'prompt.submit' || method === 'session.steer')) route.active = previouslyActive
       if (!response.error && route && (method === 'prompt.submit' || method === 'session.steer')) route.active = true
       if ((opening || method === 'session.branch') && response.result) {
@@ -273,7 +285,7 @@ export class RealtimeBroker {
         r.observers.set(c.principal.key, c.principal)
         if (nativeOwner) this.onNativeRoute(nativeOwner, profile, stored, runtime)
         if (nativeOwner && (method === 'session.create' || method === 'session.branch')) {
-          this.onNativeCommand(nativeOwner, profile, stored, method, p)
+          this.onNativeCommand(nativeOwner, profile, stored, method, {...p,_delivery_id:deliveryID})
         }
         for (const pending of [result.pending_approval, result.pending_clarify, result.inflight?.pending_approval, result.inflight?.pending_clarify]) {
           if (pending?.request_id) this.interactions.set(`${u.principal.upstreamKey}:${pending.request_id}`, runtime)
@@ -396,6 +408,22 @@ export class RealtimeBroker {
         if (r.seq > 0 && p.seq > r.seq + 1) this.reset(u, 'upstream_sequence_gap')
         r.seq = p.seq
       }
+      if(p.type==='session.info'){
+        const stored=p.payload?.stored_session_id??p.payload?.session_key??p.payload?.info?.stored_session_id
+        if(typeof stored==='string'&&stored&&stored!==r.stored){
+          const oldKey=this.routeKey(r.profile,r.stored),newKey=this.routeKey(r.profile,stored)
+          if(!u.routes.has(newKey)){
+            this.routeOwners.delete(JSON.stringify([u.principal.instanceKey??u.principal.upstreamKey,r.profile,r.stored]));
+            this.routeOwners.set(JSON.stringify([u.principal.instanceKey??u.principal.upstreamKey,r.profile,stored]),u.principal.upstreamKey);
+            u.routes.delete(oldKey);r.stored=stored;u.routes.set(newKey,r)
+            for(const channel of this.channels.values())if(channel.principal.upstreamKey===u.principal.upstreamKey&&channel.routes.delete(oldKey))channel.routes.add(newKey)
+            for(const principal of r.observers.values()){
+              const owner=this.nativeOwner(principal)
+              if(owner&&principal.valid())this.onNativeRoute(owner,r.profile,r.stored,r.runtime)
+            }
+          }
+        }
+      }
       if (p.type === 'session.info' && typeof p.payload?.cwd === 'string') {
         r.cwd = p.payload.cwd
         if (r.cwd !== r.workingDirectory?.resolved) r.workingDirectory = undefined
@@ -405,11 +433,13 @@ export class RealtimeBroker {
       if (p.payload?.request_id) this.interactions.set(`${u.principal.upstreamKey}:${p.payload.request_id}`, sid)
       while (this.interactions.size > 10_000) this.interactions.delete(this.interactions.keys().next().value!)
     }
+    const deliveryID = Number.isSafeInteger(p.seq) ? `${this.epoch}:${sid}:${p.seq}` : randomUUID()
+    const persistedOwners = new Set<string>()
     if (r) for (const principal of r.observers.values()) {
       if (principal.valid()) {
         principal.observeEvent?.(JSON.stringify(f))
         const owner = this.nativeOwner(principal)
-        if (owner) this.onNativeEvent(owner, r.profile, r.stored, p)
+        if (owner && !persistedOwners.has(owner)) { persistedOwners.add(owner); const projected={...p,delivery_id:deliveryID,epoch:this.epoch}; this.onNativeEvent(owner, r.profile, r.stored, projected); p.payload=projected.payload }
       }
     }
     if (!r && p.type === 'sessions.changed') {

src/server/routes.ts

@@ -53,6 +53,7 @@ import { chatCacheKey, type ChatCacheCoordinator } from './chatCache.js'
 type JsonObject = Record<string, unknown>
 
 export interface RouteDependencies {
+  onChatListChanged?: (owner:string,profile:string,id:string)=>void
   onServerIdentityChanged?: (identity: ServerIdentity) => void
   onUserAccessChanged?: (owner: string) => Promise<void>
   workspace: WorkspaceStore
@@ -762,58 +763,14 @@ async function cachedChatRead(
   if (!dependencies.chatCache) {
     throw new HttpError(503, 'Chat ownership registry is unavailable', 'chat_registry_unavailable')
   }
-  if (ctx.get('x-yaoyao-cache').toLowerCase() === 'bypass') {
-    if (options.kind === 'messages' && options.sessionID
-      && dependencies.chatCache.mode !== 'upstream-only'
-      && dependencies.chatCache.store.ownsSession(user.id, options.profile, options.sessionID)) {
-      await dependencies.chatCache.reconcile(user.id, options.profile, options.sessionID, true)
-      const page = dependencies.chatCache.store.messagePage(user.id, options.profile, options.sessionID, options.offset ?? 0, options.limit ?? 100)
-      if (!page) throw new HttpError(503, 'History refresh is incomplete', 'chat_refresh_incomplete')
-      ctx.set('X-Yaoyao-Data-Source', 'upstream')
-      ctx.set('X-Yaoyao-Sync-State', page.state)
-      sendUpstreamResponse(ctx, page.response, dependencies.upstreamSession.jar)
-      return
-    }
-    const requestStartedAt = Date.now()
-    const upstreamResponse = await load()
-    const response = options.kind === 'list' && options.ownedList
-      ? dependencies.chatCache.store.mergeOwnedSessionsIntoList(
-          user.id,
-          options.profile,
-          upstreamResponse,
-          { offset: options.offset, limit: options.limit, archived: options.archived },
-        )
-      : upstreamResponse
-    dependencies.chatCache.store.putSnapshot(
-      user.id, options.key, options.kind, options.profile,
-      options.sessionID, response, options.ownedList, requestStartedAt, upstreamResponse,
-    )
-    ctx.set('X-Yaoyao-Data-Source', 'upstream')
-    ctx.set('X-Yaoyao-Sync-State', 'current')
-    sendUpstreamResponse(
-      ctx,
-      options.sessionID
-        ? dependencies.chatCache.store.markSessionOwnership(
-            user.id,
-            options.profile,
-            options.sessionID,
-            response,
-          )
-        : response,
-      dependencies.upstreamSession.jar,
-    )
+  // Explicit native history browsing retains its independent upstream path.
+  if(ctx.query.view==='history' || ctx.state.hermesBotNative || !ctx.path.startsWith('/api/app/')){
+    const response=await load()
+    sendUpstreamResponse(ctx,options.sessionID?dependencies.chatCache.store.markSessionOwnership(user.id,options.profile,options.sessionID,response):response,dependencies.upstreamSession.jar)
     return
   }
-  const result = options.kind === 'messages' && options.sessionID
-    ? await dependencies.chatCache.messages(
-      user.id, options.key, options.profile, options.sessionID,
-      options.offset ?? 0, options.limit ?? 100, load,
-    )
-    : await dependencies.chatCache.read(
-      user.id, options.key, options.kind, options.profile,
-      options.sessionID, load, options.ownedList,
-      { offset: options.offset, limit: options.limit, archived: options.archived },
-    )
+  const result = dependencies.chatCache.readLocal(user.id, options.kind, options.profile, options.sessionID,
+    {offset:options.offset,limit:options.limit,archived:options.archived},ctx.get('x-yaoyao-cache').toLowerCase()==='bypass')
   ctx.set('X-Yaoyao-Data-Source', result.source)
   ctx.set('X-Yaoyao-Sync-State', result.state)
   sendUpstreamResponse(
@@ -1194,7 +1151,7 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
     json(ctx, 200, {
       ok: true,
       chatCache: dependencies.chatCache
-        ? { available: true, mode: dependencies.chatCache.mode, ...dependencies.chatCache.store.stats() }
+        ? { available: true, mode: dependencies.chatCache.mode, ...dependencies.chatCache.store.stats(), reads: dependencies.chatCache.diagnostics }
         : { available: false, mode: dependencies.config.chatCacheMode ?? 'upstream-only' },
     })
   })
@@ -2051,29 +2008,25 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
   router.get('/api/app/sessions/search', async (ctx) => {
     await searchSessions(ctx, dependencies)
   })
-  router.get('/api/app/sessions/unread', async (ctx) => {
-    const search = searchFrom(ctx, ['profile'])
-    await proxyOptionalUnread(
-      ctx,
-      dependencies.upstream,
-      '/api/session-unread',
-      { search },
-      { profile: search.get('profile') || '', total_unread: 0, sessions: [], supported: false },
-    )
+  router.get('/api/app/sessions/pins', ctx => {
+    const owner=dependencies.auth.require(ctx).id
+    ctx.body={session_ids:dependencies.chatCache!.store.pins(owner,String(ctx.query.profile??''))}
   })
-  router.patch('/api/app/sessions/unread/:sessionID', async (ctx) => {
-    const id = safeIdentifier(ctx.params.sessionID, 'session ID')
-    await proxyOptionalUnread(
-      ctx,
-      dependencies.upstream,
-      `/api/session-unread/${encodeURIComponent(id)}`,
-      {
-        method: 'PATCH',
-        search: searchFrom(ctx, ['profile']),
-        requestBody: body(ctx),
-      },
-      { ok: true, supported: false },
-    )
+  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??''))
+  })
+  router.patch('/api/app/sessions/unread/:sessionID', ctx => {
+    const owner=dependencies.auth.require(ctx).id,profile=String(ctx.query.profile??'default'),id=safeIdentifier(ctx.params.sessionID,'session ID')
+    const input=body(ctx) as Record<string,unknown>,value=input.readMessageCount??input.read_message_count
+    ctx.body=dependencies.chatCache!.store.markRead(owner,profile,id,typeof value==='number'&&Number.isFinite(value)?value:undefined)
+    dependencies.onChatListChanged?.(owner,profile,id)
+  })
+  router.post('/api/app/sessions/:sessionID/sync',ctx=>{
+    const owner=dependencies.auth.require(ctx).id,profile=String(ctx.query.profile??'default'),id=safeIdentifier(ctx.params.sessionID,'session ID')
+    dependencies.chatCache!.store.requireOwned(owner,profile,id)
+    dependencies.chatCache!.schedule(owner,profile,id,true)
+    ctx.status=202;ctx.body={sync_state:'syncing'}
   })
   router.get('/api/app/sessions', async (ctx) => {
     const view = ctx.query.view === 'history' || ctx.query.view === 'chat' ? ctx.query.view : undefined
@@ -2152,6 +2105,14 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
   router.patch('/api/app/sessions/:sessionID', async (ctx) => {
     const id = safeIdentifier(ctx.params.sessionID, 'session ID')
     const search = searchFrom(ctx, ['profile'])
+    const localPatch=writableSessionPatch(body(ctx))
+    if(Object.keys(localPatch).every(key=>['title','pinned','archived'].includes(key))){
+      const owner=dependencies.auth.require(ctx).id,profile=search.get('profile')||'default'
+      if(!dependencies.chatCache!.store.ownsSession(owner,profile,id))throw new HttpError(410,'原生历史只读','native_sessions_read_only')
+      dependencies.chatCache!.store.patchLocal(owner,profile,id,localPatch)
+      dependencies.onChatListChanged?.(owner,profile,id)
+      ctx.body={ok:true};return
+    }
     await withJar(ctx, async jar => {
       const profile = search.get('profile') || 'default'
       const owner = dependencies.auth.require(ctx).id

tests/client/chatModelSync.test.ts

@@ -10,6 +10,7 @@ const sessionsApi = vi.hoisted(() => ({
   getSession: vi.fn(),
   getMessages: vi.fn(),
   getSessions: vi.fn(),
+  requestHistorySync: vi.fn(),
 }))
 
 vi.mock('@/api/realtime', () => {
@@ -29,6 +30,7 @@ vi.mock('@/api/sessions', () => ({
   getMessages: sessionsApi.getMessages,
   getSession: sessionsApi.getSession,
   getSessions: sessionsApi.getSessions,
+  requestHistorySync: sessionsApi.requestHistorySync,
   getSessionUnread: vi.fn(),
   markSessionRead: vi.fn(),
   updateSession: vi.fn(),
@@ -49,6 +51,7 @@ describe('chat model realtime synchronization', () => {
     realtime.eventHandler = undefined
     realtime.request.mockReset().mockResolvedValue({})
     sessionsApi.getMessages.mockReset()
+    sessionsApi.requestHistorySync.mockReset().mockResolvedValue(undefined)
     sessionsApi.getSession.mockReset()
     sessionsApi.getSessions.mockReset().mockResolvedValue({ items: [], nextCursor: null })
     setActivePinia(createPinia())
@@ -242,8 +245,10 @@ describe('chat model realtime synchronization', () => {
       { id: 'new', sessionId: 'session-1', role: 'assistant', content: 'authoritative', timestamp: 1, stage: 'settled' },
     ], returned: 1, total: 1, hasMore: false })
     await chat.forceRefreshHistory()
-    expect(sessionsApi.getMessages).toHaveBeenCalledWith('session-1', 0, 150, 'alpha', true)
-    expect(chat.messages.map(message => message.id)).toEqual(['new', 'unsent'])
+    expect(sessionsApi.requestHistorySync).toHaveBeenCalledWith('session-1','alpha')
+    expect(sessionsApi.getMessages).not.toHaveBeenCalled()
+    expect(chat.messages.map(message => message.id)).toEqual(['old', 'unsent'])
+    expect(chat.routes[key].error).toContain('后台补齐')
   })
 
 })

tests/e2e/ordinary-local.spec.ts

@@ -0,0 +1,25 @@
+import {test,expect} from '@playwright/test'
+
+test('ordinary SSE replies remain readable from Web without a second history source',async({page})=>{
+  await page.goto('/chat')
+  if(await page.getByRole('heading',{name:'登录夭夭'}).isVisible()){
+    await page.getByLabel('账号').fill('admin');await page.getByLabel('密码').fill('e2e-password')
+    await page.getByRole('button',{name:'登录',exact:true}).click()
+  }
+  await expect(page.getByRole('navigation')).toBeVisible()
+  await page.getByRole('button',{name:'新建聊天',exact:true}).click()
+  const prompt=`本地落盘验收 ${Date.now()}`
+  await page.locator('.composer-textarea').fill(prompt)
+  await page.getByRole('button',{name:'发送消息'}).click()
+  await expect(page.locator('.message__content').filter({hasText:'这是来自假 Gateway 的流式回复。'})).toHaveCount(1,{timeout:15000})
+  const result=await page.evaluate(async()=>{
+    const response=await fetch('/api/app/sessions/session-new/messages?profile=yaoyao&limit=100')
+    return{status:response.status,source:response.headers.get('X-Yaoyao-Data-Source'),body:await response.json()}
+  })
+  expect(result.status).toBe(200);expect(result.source).toBe('local')
+  expect(result.body.messages.filter((m:any)=>m.role==='user').map((m:any)=>m.content)).toEqual([prompt])
+  expect(result.body.messages.filter((m:any)=>m.role==='assistant').map((m:any)=>m.content)).toEqual(['这是来自假 Gateway 的流式回复。'])
+  await page.reload()
+  await expect(page.locator('.message__content').filter({hasText:'这是来自假 Gateway 的流式回复。'})).toHaveCount(1)
+  await page.screenshot({path:'/tmp/yaoyao-ordinary-local-browser.png',fullPage:true})
+})

tests/fixtures/e2e-web-server.mjs

@@ -50,6 +50,10 @@ for (const [sessionID, profile] of ownedSessions) {
     source: 'web',
   })
 }
+// These are existing Hermes histories, not empty sessions created by this build.
+// Exercise the durable background import, never a blocking list fallback.
+registry.db.exec('UPDATE chat_sessions SET metadata_complete=0,complete=0,message_total=NULL')
+registry.db.exec('INSERT OR IGNORE INTO chat_recovery_jobs(owner,profile,session_id) SELECT owner,profile,session_id FROM chat_sessions WHERE owned_at IS NOT NULL')
 registry.close()
 
 Object.assign(process.env, {

tests/server/app.test.ts

@@ -252,12 +252,13 @@ describe('15300 BFF', () => {
       .get('/api/app/sessions/unread?profile=yaoer')
       .set('Host', '127.0.0.1:15300')
       .expect(200)
-    expect(unread.body).toEqual({ profile: 'yaoer', total_unread: 0, sessions: [], supported: false })
+    expect(unread.body).toEqual({ profile: 'yaoer', total_unread: 0, sessions: [], supported: true })
 
     const bootstrap = await request(runtime.app.callback())
       .get('/api/app/bootstrap')
       .set('Host', '127.0.0.1:15300')
       .expect(200)
+    runtime.chatCache!.store.recordCommand(String(bootstrap.body.user.id),'yaoer','session-1','session.create',{})
     const marked = await request(runtime.app.callback())
       .patch('/api/app/sessions/unread/session-1?profile=yaoer')
       .set('Host', '127.0.0.1:15300')
@@ -266,7 +267,7 @@ describe('15300 BFF', () => {
       .set('X-CSRF-Token', bootstrap.body.csrfToken)
       .send({ readMessageCount: 12 })
       .expect(200)
-    expect(marked.body).toEqual({ ok: true, supported: false })
+    expect(marked.body).toMatchObject({session_id:'session-1',read_message_count:0,unread_count:0})
   })
 
   it('requires exact Origin and CSRF on local login without forwarding credentials upstream', async () => {
@@ -316,7 +317,7 @@ describe('15300 BFF', () => {
     expect(records.at(-1)?.search.get('exclude_sources')).toBe('cron,ios_group,yaoyao_workspace')
 
     await request(runtime.app.callback())
-      .get('/api/app/sessions/session-1/messages?offset=3&limit=50&order=oldest&include_compacted=false&profile=default')
+      .get('/api/app/sessions/session-1/messages?view=history&offset=3&limit=50&order=oldest&include_compacted=false&profile=default')
       .set('Host', '127.0.0.1:15300')
       .expect(200)
     const history = records.at(-1)!

tests/server/chatCache.test.ts

@@ -116,6 +116,8 @@ describe('durable source=web chat cache', () => {
     const load = vi.fn(async () => response({ sessions: [
       { id: 'unowned-history', profile, source: 'web', title: 'Private native history' }, ...sessions,
     ] }))
+    // Existing installations have registered sessions whose metadata has not been imported.
+    f.store.db.prepare('UPDATE chat_sessions SET metadata_complete=0').run()
     const first = await f.coordinator.read(owner, 'first', 'list', profile, undefined, load, true, { limit: 100 })
     const firstPage = JSON.parse(first.response.body.toString())
     expect(firstPage.sessions[0]).toMatchObject({ id: 'owned-102', last_active: sessions[102]!.last_active_at })
@@ -391,10 +393,10 @@ describe('durable source=web chat cache', () => {
     expect(route.runtime_id).toBe('runtime-1')
     const fallback = await f.coordinator.read(owner, key, 'detail', profile, sessionID,
       async () => { throw new Error('9119 offline') })
-    expect(fallback).toMatchObject({ source: 'local', state: 'stale' })
+    expect(fallback).toMatchObject({ source: 'local', state: 'current' })
     const serverFailure = await f.coordinator.read(owner, key, 'detail', profile, sessionID,
       async () => response({ error: 'unavailable' }, 503))
-    expect(serverFailure).toMatchObject({ source: 'local', state: 'stale' })
+    expect(serverFailure).toMatchObject({ source: 'local', state: 'current' })
     f.store.close()
   })
 
@@ -644,6 +646,7 @@ describe('local history and bounded tail synchronization', () => {
     const rows = [...f.rows, { id: 'new-user', role: 'user', content: 'next' }, { id: 'new-answer', role: 'assistant', content: 'answer' }]
     const { coordinator, request } = upstreamFor(f, rows)
     coordinator.observe(owner, profile, sessionID, { type: 'message.complete', seq: 1 })
+    await coordinator.reconcile(owner, profile, sessionID)
     const page = await coordinator.messages(owner, 'last', profile, sessionID, 0, 10, async () => { throw new Error('unexpected read') })
     expect(JSON.parse(page.response.body.toString()).messages.at(-1).id).toBe('new-answer')
     expect(request.mock.calls.filter(([path]) => path.endsWith('/messages'))).toHaveLength(1)
@@ -711,8 +714,8 @@ describe('local history and bounded tail synchronization', () => {
       if (path.endsWith('/messages')) f.store.recordEvent(owner, profile, sessionID, { type: 'message.delta', seq: 2, payload: { delta: 'newer' } })
       return result
     })
-    await coordinator.reconcile(owner, profile, sessionID)
-    expect(f.store.localDetail(owner, profile, sessionID)?.state).toBe('stale')
+    await expect(coordinator.reconcile(owner, profile, sessionID)).rejects.toThrow('Conversation changed')
+    expect(JSON.parse(f.store.messagePage(owner,profile,sessionID,0,1)!.response.body.toString()).messages[0].content).toBe('newer')
     f.store.close()
   })
 
@@ -738,7 +741,9 @@ describe('local history and bounded tail synchronization', () => {
     f.store.recordEvent(owner, profile, sessionID, { type: 'message.complete', seq: 2 })
     const latestSync = coordinator.reconcile(owner, profile, sessionID)
     release()
-    await Promise.all([firstSync, latestSync])
+    const results = await Promise.allSettled([firstSync, latestSync])
+    expect(results[0].status).toBe('rejected')
+    expect(results[1].status).toBe('fulfilled')
     expect(published).toHaveBeenCalledTimes(1)
     expect(request).toHaveBeenCalledTimes(4)
     expect(f.store.localDetail(owner, profile, sessionID)?.state).toBe('current')

tests/server/chatCacheRoutes.test.ts

@@ -167,9 +167,10 @@ describe('source=web chat cache routes', () => {
     const ownedDetail = await agent.get('/api/app/sessions/owned-session?profile=default')
       .set('Host', host).expect(200)
     expect(ownedDetail.body).toMatchObject({ id: 'owned-session', source: 'ios', owned: true })
-    const historyDetail = await agent.get('/api/app/sessions/history-only?profile=default')
+    const historyDetail = await agent.get('/api/app/sessions/history-only?profile=default&view=history')
       .set('Host', host).expect(200)
     expect(historyDetail.body).toMatchObject({ id: 'history-only', source: 'ios', owned: false })
+    await runtime.chatCache!.reconcile(String(setup.body.user.id), 'default', 'owned-session')
     const ownedMessages = await agent
       .get('/api/app/sessions/owned-session/messages?profile=default&offset=0&limit=100')
       .set('Host', host).expect(200)
@@ -231,11 +232,7 @@ describe('source=web chat cache routes', () => {
       .send({ title: 'Web 可继续管理' })
       .expect(200)
     const patchCalls = fetchImpl.mock.calls.filter(([, init]) => init?.method === 'PATCH')
-    expect(patchCalls).toHaveLength(1)
-    expect(JSON.parse(String(patchCalls[0]?.[1]?.body))).toEqual({
-      title: 'Web 可继续管理',
-      profile: 'default',
-    })
+    expect(patchCalls).toHaveLength(0) // Ordinary presentation state is owned by Web.
     const immediateRenamedDetail = await agent
       .get('/api/app/sessions/owned-session?profile=default')
       .set('Host', host)
@@ -386,13 +383,14 @@ describe('source=web chat cache routes', () => {
     first.chatCache!.store.recordRoute(owner, 'default', 'session-web', 'runtime-web')
     first.chatCache!.store.recordCommand(owner, 'default', 'session-web', 'session.create', { source: 'web' })
 
+    await first.chatCache!.reconcile(owner,'default','session-web')
     const listPath = '/api/app/sessions?view=chat&profile=default&limit=100'
-    await agent.get(listPath).set('Host', host).expect('X-Yaoyao-Data-Source', 'upstream').expect(200)
+    await agent.get(listPath).set('Host', host).expect('X-Yaoyao-Data-Source', 'local').expect(200)
     await agent.get(listPath).set('Host', host).expect('X-Yaoyao-Data-Source', 'local').expect(200)
     await agent.get(listPath).set('Host', host).set('X-Yaoyao-Cache', 'bypass')
-      .expect('X-Yaoyao-Data-Source', 'upstream').expect(200)
+      .expect('X-Yaoyao-Data-Source', 'local').expect(200)
     const messagesPath = '/api/app/sessions/session-web/messages?profile=default&offset=0&limit=100'
-    await agent.get(messagesPath).set('Host', host).expect('X-Yaoyao-Data-Source', 'upstream').expect(200)
+    await agent.get(messagesPath).set('Host', host).expect('X-Yaoyao-Data-Source', 'local').expect(200)
     const cachedMessages = await agent.get(messagesPath).set('Host', host)
       .expect('X-Yaoyao-Data-Source', 'local').expect(200)
     expect(cachedMessages.body.messages[0].content).toContain('持久消息')
@@ -405,7 +403,7 @@ describe('source=web chat cache routes', () => {
     expect(Buffer.from(rangedMedia.body).toString()).toBe('png')
     await agent.get('/api/app/sessions?view=history').set('Host', host).expect(200)
     await agent.get('/api/app/sessions?view=history').set('Host', host).expect(200)
-    expect(counts).toMatchObject({ list: 2, messages: 1, history: 1, media: 1 })
+    expect(counts).toMatchObject({ list: 0, messages: 1, history: 1, media: 1 })
     const historySnapshots = first.chatCache!.store.db.prepare("SELECT COUNT(*) count FROM chat_snapshots WHERE kind='history'").get() as { count: number }
     expect(historySnapshots.count).toBe(0)
 

tests/server/ordinaryChatLocal.test.ts

@@ -0,0 +1,76 @@
+// @vitest-environment node
+import {mkdtempSync,rmSync} from 'node:fs'
+import {tmpdir} from 'node:os'
+import {join} from 'node:path'
+import {afterEach,describe,it,expect,vi} from 'vitest'
+import {ChatCacheStore,ChatCacheCoordinator} from '../../src/server/chatCache.js'
+import type {UpstreamServiceSession} from '../../src/server/localAuth.js'
+const fixtures:Array<{home:string;store:ChatCacheStore;coordinator:ChatCacheCoordinator}>=[]
+function setup(){const home=mkdtempSync(join(tmpdir(),'ordinary-local-')),store=new ChatCacheStore(home),request=vi.fn();const coordinator=new ChatCacheCoordinator(store,{request} as unknown as UpstreamServiceSession);fixtures.push({home,store,coordinator});return{store,coordinator,request,home}}
+afterEach(()=>{for(const f of fixtures.splice(0)){f.coordinator.close();if(!f.store.isClosed)f.store.close();rmSync(f.home,{recursive:true,force:true})}})
+function json(r:{response:{body:Buffer}}){return JSON.parse(r.response.body.toString())}
+describe('ordinary chat authoritative local reads',()=>{
+ it('persists a complete streaming turn without upstream reads, including repeated chunks',()=>{
+  const {store,coordinator,request}=setup();store.recordCommand('o','p','s','session.create',{source:'ios'})
+  store.recordCommand('o','p','s','prompt.submit',{text:'问题',_delivery_id:'submit-1'})
+  const send=(type:string,payload:object,delivery_id:string)=>coordinator.observe('o','p','s',{type,payload,delivery_id})
+  send('message.start',{message_id:'a'},'1');send('reasoning.delta',{delta:'思考'},'2')
+  send('message.delta',{delta:'哈'},'3');send('message.delta',{delta:'哈'},'4');send('message.delta',{delta:'哈'},'4')
+  send('tool.start',{tool_id:'t',name:'read',arguments:{path:'a'}},'5');send('tool.complete',{tool_id:'t',result:'ok'},'6')
+  send('message.complete',{},'7')
+  const page=json(coordinator.readLocal('o','messages','p','s',{limit:20}))
+  expect(page.messages).toHaveLength(2);expect(page.messages[1]).toMatchObject({content:'哈哈',reasoning:'思考',status:'complete',tool_calls:[{id:'t',result:'ok'}]})
+  expect(json(coordinator.readLocal('o','list','p',undefined,{})).sessions[0].message_count).toBe(2)
+  expect(request).not.toHaveBeenCalled()
+  store.close();const reopened=new ChatCacheStore(fixtures.at(-1)!.home);expect(JSON.parse(reopened.messagePage('o','p','s',0,20)!.response.body.toString()).messages).toEqual(page.messages);reopened.close()
+ })
+ it('keeps sequence-less identical deltas and interim segments independently',()=>{
+  const {store,coordinator}=setup();store.recordCommand('o','p','s','session.create',{})
+  for(const [type,payload] of [['message.delta',{text:'ha'}],['message.delta',{text:'ha'}],['message.interim',{}],['message.delta',{text:'done'}],['message.complete',{}]] as const)coordinator.observe('o','p','s',{type,payload})
+  const page=json(coordinator.readLocal('o','messages','p','s',{}));expect(page.messages.map((m:any)=>m.content)).toEqual(['haha','done'])
+ })
+ it('isolates preferences and clamps monotonic read cursors locally',()=>{
+  const {store,coordinator,request}=setup();store.recordCommand('o','p','s','session.create',{});coordinator.observe('o','p','s',{type:'message.delta',payload:{text:'a'}})
+  store.patchLocal('o','p','s',{pinned:true,title:'标题'});expect(store.pins('o','p')).toEqual(['s']);expect(store.pins('other','p')).toEqual([])
+  expect(store.unread('o','p').total_unread).toBe(1);store.markRead('o','p','s',900);store.markRead('o','p','s',0);expect(store.unread('o','p').total_unread).toBe(0)
+  expect(()=>coordinator.readLocal('other','messages','p','s',{})).toThrow('会话不存在');expect(request).not.toHaveBeenCalled()
+ })
+ it('returns cached data immediately and reports a missing page without awaiting repair',()=>{
+  const {store,coordinator,request}=setup();store.recordRoute('o','p','legacy','r')
+  expect(json(coordinator.readLocal('o','list','p',undefined,{})).sessions).toHaveLength(1)
+  expect(()=>coordinator.readLocal('o','messages','p','legacy',{})).toThrow('历史正在后台补齐')
+  expect(request).not.toHaveBeenCalled()
+ })
+ it('keeps newer local pin settings when remote metadata is ingested',()=>{
+  const {store}=setup();store.recordCommand('o','p','s','session.create',{});store.patchLocal('o','p','s',{pinned:true})
+  store.putSnapshot('o','d','detail','p','s',{status:200,headers:new Headers(),body:Buffer.from(JSON.stringify({id:'s',pinned:false,message_count:0}))})
+  expect(store.pins('o','p')).toEqual(['s']);expect(JSON.parse(store.localList('o','p',{}).body.toString()).sessions[0].pinned).toBe(true)
+ })
+})
+
+describe('ordinary chat migration and bounded recovery',()=>{
+ it('preserves local messages and pins when a live session becomes a compressed tip',()=>{
+   const {store,coordinator}=setup();store.recordRoute('o','p','old','runtime');store.recordCommand('o','p','old','session.create',{})
+   coordinator.observe('o','p','old',{type:'message.delta',payload:{text:'kept'},delivery_id:'one'})
+   store.patchLocal('o','p','old',{pinned:true,title:'local'})
+   store.recordRoute('o','p','new','runtime')
+   expect(store.canonicalSessionID('o','p','old')).toBe('new')
+   expect(json(coordinator.readLocal('o','messages','p','old',{})).messages[0].content).toBe('kept')
+   expect(store.pins('o','p')).toEqual(['new'])
+   expect(store.canonicalSessionID('other','p','old')).toBe('old')
+ })
+ it('limits background repair to two concurrent conversations',async()=>{
+   const {store,coordinator,request}=setup();let active=0,maximum=0
+   request.mockImplementation(async(path:string)=>{active++;maximum=Math.max(maximum,active);await new Promise(r=>setTimeout(r,10));active--;
+     return {status:200,headers:new Headers(),body:Buffer.from(JSON.stringify(path.endsWith('/messages')?{messages:[],pagination:{total:0,has_more:false}}:{id:path.split('/').at(-1),message_count:0}))}})
+   for(let i=0;i<5;i++){store.recordRoute('o','p',`legacy-${i}`,`r-${i}`);coordinator.schedule('o','p',`legacy-${i}`)}
+   await vi.waitFor(()=>expect(store.db.prepare('SELECT * FROM chat_recovery_jobs').all()).toHaveLength(0))
+   expect(maximum).toBe(2)
+ })
+ it('promotes an assistant identity without creating a second history row',()=>{
+   const {store,coordinator}=setup();store.recordCommand('o','p','s','session.create',{})
+   coordinator.observe('o','p','s',{type:'message.delta',payload:{text:'hello'},delivery_id:'delta'})
+   coordinator.observe('o','p','s',{type:'message.complete',payload:{message_id:'canonical',text:'hello'},delivery_id:'done'})
+   const page=json(coordinator.readLocal('o','messages','p','s',{}));expect(page.messages).toHaveLength(1);expect(page.messages[0].id).toBe('canonical')
+ })
+})

tests/server/realtimeApi.test.ts

@@ -74,7 +74,7 @@ describe('native Hermes chat and history are separated on 15300', () => {
       .send({ title: '不能修改' })
       .expect(410)
     expect(pairedWrite.body.code).toBe('native_sessions_read_only')
-    expect(upstreamCalls).toBeGreaterThan(beforeWrites)
+    expect(upstreamCalls).toBe(beforeWrites) // Ownership checks do not read Hermes.
     expect(upstreamWrites).toBe(beforeUpstreamWrites)
 
     const unownedWeb = await agent.patch('/api/app/sessions/session-web?profile=default')
@@ -115,6 +115,6 @@ describe('native Hermes chat and history are separated on 15300', () => {
     expect(await ownershipGate.canResume('default', 'session-web', undefined, owner)).toBe(true)
     await agent.patch('/api/app/sessions/session-web?profile=default').set('Host', '127.0.0.1:15300')
       .set('Origin', origin).set('X-CSRF-Token', csrf).send({ title: '可修改的聊天' }).expect(200)
-    expect(upstreamWrites).toBe(beforeUpstreamWrites + 1)
+    expect(upstreamWrites).toBe(beforeUpstreamWrites)
   })
 })