修复普通聊天消息重复回显并升级至 v0.4.8

docs/ordinary-chat-incremental-sync.md

@@ -11,11 +11,18 @@
 - 旧消息删除、重排等改变历史位置的情况保留本地内容，等待明确强制刷新，不把它当作普通尾部追加。
 - 模型变化继续由实时事件和进入页面时的必要校验处理，取消每 3 秒无条件读取会话详情。
 
+## 用户消息身份
+
+- `chat_message_identities` 按账号、Profile 和会话保存正式消息 ID 与 `client_message_id` 的一对一关联；强制刷新、重启后继续保留，会话压缩迁移和删除时同步处理。
+- 上游携带客户端 ID 时直接保留。旧版上游缺少 ID 时，仅在历史覆盖完整后，结合已确认的提交回执、运行开始时间、下一次提交边界和完整文本寻找唯一对应记录；未确认、被拒绝、排队或对应关系不唯一时不猜测。该兼容路径要求上游时间落在本地记录的提交窗口内。
+- 已有缓存缺失 ID 时，在本地消息读取中恢复关联，无需额外请求上游。前端通过正式 ID 与客户端 ID 合并已有气泡；迟到的发送回执不能覆盖正式消息状态。相同文字的多次独立提交始终保留。
+
 ## 验证
 
 - 1,000 条已缓存消息，20 次历史分页读取不请求上游。
 - 新增回复只读取一页尾部；漏掉 130 条消息只补两页；原有消息不重新写入。
 - 明确强制刷新才覆盖旧编辑和删除；失败时保留旧记录；并发完成事件提交后再通知客户端。
 - 浏览器验证普通重载命中本地、强制刷新回源，以及只读历史和普通聊天的边界。
+- 用户消息回归覆盖历史与回执乱序、已有重复缓存、连续相同文本、跨页歧义、重启及强制刷新后的身份保留。
 
 这些请求数来自隔离测试，并非对真实网络耗时的估计。

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "yaoyao",
-  "version": "0.4.7",
+  "version": "0.4.8",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "yaoyao",
-      "version": "0.4.7",
+      "version": "0.4.8",
       "license": "Apache-2.0",
       "dependencies": {
         "@koa/bodyparser": "^5.0.0",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "yaoyao",
-  "version": "0.4.7",
+  "version": "0.4.8",
   "private": true,
   "license": "Apache-2.0",
   "description": "夭夭 AI：聊天、多 Agent 团队、文件库与电脑工作空间",

release.json

@@ -1,6 +1,6 @@
 {
   "schemaVersion": 1,
-  "releaseVersion": "0.4.7",
-  "webVersion": "0.4.7",
-  "gitTag": "v0.4.7"
+  "releaseVersion": "0.4.8",
+  "webVersion": "0.4.8",
+  "gitTag": "v0.4.8"
 }

src/client/stores/chat.ts

@@ -726,7 +726,7 @@ export const useChatStore = defineStore('chat', () => {
 
   function updateDelivery(state: ChatRouteState, clientMessageId: string, patch: Partial<ChatMessage>): void {
     state.messages = state.messages.map(message => message.clientMessageId === clientMessageId || message.id === clientMessageId
-      ? { ...message, ...patch }
+      ? { ...message, ...patch, stage: message.stage === 'settled' && patch.stage === 'accepted' ? 'settled' : patch.stage ?? message.stage }
       : message)
   }
 

src/client/utils/messageReducer.ts

@@ -36,6 +36,10 @@ function sameMessage(left: ChatMessage, right: ChatMessage): boolean {
 }
 
 function mergeOne(previous: ChatMessage, incoming: ChatMessage): ChatMessage {
+  const persistedUser = (message: ChatMessage) => message.role === 'user' && message.stage === 'settled'
+    && message.serverMessageId === message.id && message.id !== message.clientMessageId && !message.id.startsWith('user:')
+  // A late receipt or a cached optimistic row must not demote persisted history.
+  if (persistedUser(previous) && !persistedUser(incoming)) [previous, incoming] = [incoming, previous]
   const replacesContent = Boolean(incoming.content) || Boolean(incoming.attachments?.length)
   return {
     ...previous,
@@ -55,9 +59,16 @@ export function mergeChatMessages(existing: ChatMessage[], incoming: ChatMessage
   const base = position === 'prepend' || position === 'snapshot' ? [...incoming, ...existing] : [...existing, ...incoming]
   const result: ChatMessage[] = []
   for (const message of base) {
-    const index = result.findIndex(candidate => sameMessage(candidate, message))
-    if (index < 0) result.push(message)
-    else result[index] = mergeOne(result[index], message)
+    let merged = message
+    let index: number
+    let insertionIndex = result.length
+    // An enriched history row can bridge both an old server row and its local
+    // echo already in the cache. Consume every identity match, not just the first.
+    while ((index = result.findIndex(candidate => sameMessage(candidate, merged))) >= 0) {
+      insertionIndex = Math.min(insertionIndex, index)
+      merged = mergeOne(result.splice(index, 1)[0]!, merged)
+    }
+    result.splice(insertionIndex, 0, merged)
   }
   return orderChatMessages(result)
 }

src/server/chatCache.ts

@@ -192,7 +192,12 @@ export class ChatCacheStore {
       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));`)
+      PRIMARY KEY(owner,profile,session_id,event_id));
+      CREATE TABLE IF NOT EXISTS chat_message_identities(
+        owner TEXT NOT NULL, profile TEXT NOT NULL, session_id TEXT NOT NULL,
+        message_id TEXT NOT NULL, client_message_id TEXT NOT NULL,
+        PRIMARY KEY(owner,profile,session_id,message_id),
+        UNIQUE(owner,profile,session_id,client_message_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'],['final_read_count','INTEGER']]){
       if(!localColumns.some(c=>c.name===name))this.db.exec(`ALTER TABLE chat_local_state ADD COLUMN ${name} ${type}`)
@@ -552,13 +557,15 @@ export class ChatCacheStore {
         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 identified = this.userMessageIdentities(owner, profile, sessionID, messages, false)
           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))
+            if(value.role==='user'&&value.status==='pending'&&!identified.some(m=>m.role==='user'
+              && value.client_message_id && m.client_message_id===value.client_message_id))
               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)
           }
@@ -566,6 +573,14 @@ export class ChatCacheStore {
       }
       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)
+      // Resolve legacy identities only after all pages are stored. Resolving
+      // pages individually could consume a send before seeing an ambiguous row.
+      const coverage = this.db.prepare('SELECT complete FROM chat_sessions WHERE owner=? AND profile=? AND session_id=?')
+        .get(owner, profile, sessionID) as {complete: number} | undefined
+      if (coverage?.complete) this.userMessageIdentities(owner, profile, sessionID, pages.flatMap(page => {
+        const messages = page.kind === 'messages' ? parseResponse(page.response)?.messages : undefined
+        return Array.isArray(messages) ? messages.flatMap(value => object(value) ? [object(value)!] : []) : []
+      }))
       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')
@@ -576,6 +591,82 @@ export class ChatCacheStore {
     }
   }
 
+  private userMessageIdentities(owner: string, profile: string, sessionID: string,
+    messages: Record<string, unknown>[], recoverLegacy = true): Record<string, unknown>[] {
+    const scope = [owner, profile, sessionID]
+    const bindings = this.db.prepare(`SELECT message_id,client_message_id FROM chat_message_identities
+      WHERE owner=? AND profile=? AND session_id=?`).all(...scope) as {message_id: string; client_message_id: string}[]
+    const byServer = new Map(bindings.map(row => [row.message_id, row.client_message_id]))
+    const byClient = new Map(bindings.map(row => [row.client_message_id, row.message_id]))
+    const serverID = (message: Record<string, unknown>) => {
+      const id = String(message.id ?? message.message_id ?? '').trim()
+      return message.role === 'user' && id && !id.startsWith('user:') ? id : ''
+    }
+    const bind = (id: string, client: string) => {
+      if (!id || !client || byServer.has(id) || byClient.has(client)) return
+      this.db.prepare('INSERT INTO chat_message_identities VALUES(?,?,?,?,?)').run(...scope, id, client)
+      byServer.set(id, client)
+      byClient.set(client, id)
+    }
+    for (const message of messages) {
+      bind(serverID(message), String(message.client_message_id ?? message.clientMessageId ?? ''))
+    }
+    if (recoverLegacy && messages.some(message => serverID(message) && !byServer.has(serverID(message)))) {
+      // Older gateways omit client IDs. Recover only a mutually unique match
+      // inside a confirmed, non-queued submission's time window. Never dedupe
+      // by text or position alone, and include other pages when testing uniqueness.
+      const events = this.db.prepare(`SELECT event_type,payload,received_at FROM chat_events
+        WHERE owner=? AND profile=? AND session_id=?
+        AND event_type IN ('command:prompt.submit','command:session.steer','command.confirmed','command.rejected','message.start')
+        ORDER BY received_at,event_seq`)
+        .all(...scope) as {event_type: string; payload: string; received_at: number}[]
+      const confirmed = new Map<string, number>(), rejected = new Set<string>()
+      for (const event of events) {
+        const payload = JSON.parse(event.payload)
+        if (event.event_type === 'command.confirmed' && payload.status !== 'queued') confirmed.set(payload.delivery_id, event.received_at)
+        if (event.event_type === 'command.rejected') rejected.add(payload.delivery_id)
+      }
+      const stored = this.db.prepare(`SELECT data FROM chat_messages WHERE owner=? AND profile=? AND session_id=?
+        AND json_extract(data,'$.role')='user'`).all(...scope) as {data: string}[]
+      const candidates = new Map<string, Record<string, unknown>>()
+      for (const message of [...stored.map(row => JSON.parse(row.data)), ...messages]) {
+        const id = serverID(message)
+        if (id && !byServer.has(id)) candidates.set(id, message)
+      }
+      const matches = new Map<string, string[]>(), claims = new Map<string, number>()
+      const submissions = events.filter(event => event.event_type === 'command:prompt.submit' || event.event_type === 'command:session.steer')
+      for (const [index, event] of submissions.entries()) {
+        if (event.event_type !== 'command:prompt.submit') continue
+        const payload = JSON.parse(event.payload), delivery = String(payload._delivery_id ?? '')
+        const client = delivery.replace(/^(web|ios):prompt:/, '')
+        const acceptedAt = confirmed.get(delivery)
+        if (!client || byClient.has(client) || payload.queued || rejected.has(delivery) || acceptedAt === undefined) continue
+        const start = event.received_at / 1000
+        const nextStart = submissions[index + 1]?.received_at ?? Infinity
+        const runStart = events.find(candidate => candidate.event_type === 'message.start'
+          && candidate.received_at >= event.received_at && candidate.received_at < nextStart)
+        // The receipt can precede the gateway's DB insert by a few milliseconds.
+        // A deferred agent build may start later; the next submission always
+        // closes this window, including when users rapidly repeat the same text.
+        const end = Math.min((runStart?.received_at ?? acceptedAt) + 1000, event.received_at + 120_000) / 1000
+        const ids = [...candidates].flatMap(([id, message]) => {
+          const time = firstTimestamp(message, ['timestamp', 'created_at', 'createdAt'])
+          return time !== undefined && time >= start && time <= end && time < nextStart / 1000
+            && message.content === (payload.text ?? payload.input ?? '') ? [id] : []
+        })
+        matches.set(client, ids)
+        for (const id of ids) claims.set(id, (claims.get(id) ?? 0) + 1)
+      }
+      for (const [client, ids] of matches) {
+        if (ids.length === 1 && claims.get(ids[0]!) === 1) bind(ids[0]!, client)
+      }
+    }
+    return messages.map(message => {
+      const client = byServer.get(serverID(message))
+      return client ? { ...message, client_message_id: client } : message
+    })
+  }
+
   messagePage(owner: string, profile: string, sessionID: string, offset: number, limit: number): MessagePage | undefined {
     const session = this.db.prepare(`SELECT data,sync_state,complete,message_total FROM chat_sessions
       WHERE owner=? AND profile=? AND session_id=?`).get(owner, profile, sessionID) as Record<string, unknown> | undefined
@@ -588,8 +679,8 @@ export class ChatCacheStore {
     const start = Math.max(0, end - limit)
     const rows = this.db.prepare(`SELECT data FROM chat_messages WHERE owner=? AND profile=? AND session_id=?
       AND position>=? AND position<? ORDER BY position`).all(owner, profile, sessionID, start, end) as Array<{ data: string }>
-    const messages = rows.map(row => JSON.parse(row.data))
-    if (messages.length !== end - start) return undefined // Fetch only an actually missing page.
+    if (rows.length !== end - start) return undefined // Fetch only an actually missing page.
+    const messages = this.userMessageIdentities(owner, profile, sessionID, rows.map(row => JSON.parse(row.data)), bool(session.complete))
     const body = Buffer.from(JSON.stringify({
       session_id: sessionID,
       session: JSON.parse(this.localDetail(owner,profile,sessionID)!.response.body.toString()),
@@ -675,7 +766,7 @@ export class ChatCacheStore {
     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'])
+      for(const table of ['chat_sessions','chat_messages','chat_events','chat_snapshots','chat_attachments','chat_local_state','chat_event_receipts','chat_recovery_jobs','chat_message_identities'])
         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)
@@ -773,7 +864,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_local_state', 'chat_event_receipts', 'chat_recovery_jobs', 'chat_session_aliases', '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_message_identities', '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')")
@@ -1121,7 +1212,7 @@ export class ChatCacheStore {
     const total = Math.max(offset + messages.length,
       Number(pagination.total ?? payload.total ?? storedSummary.message_count ?? storedSummary.messageCount ?? 0) || 0)
     const start = Math.max(0, total - offset - messages.length)
-    messages.forEach((message, index) => {
+    this.userMessageIdentities(owner, profile, sessionID, messages, false).forEach((message, index) => {
       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 position=excluded.position,

src/server/realtimeBroker.ts

@@ -257,7 +257,7 @@ export class RealtimeBroker {
         .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}})
+          delivery_id:`command-result:${deliveryID}`,payload:{delivery_id:deliveryID,error:response.error?.message,status:response.result?.status}})
         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)

tests/client/chatCrossClientSync.test.ts

@@ -111,6 +111,26 @@ describe('cross-client chat subscriptions', () => {
     expect(api.getMessages).toHaveBeenCalledTimes(1)
   })
 
+  it('reconciles synchronized history before a delayed submit receipt without duplicating or demoting the user row', async () => {
+    chat.sessions = [session('identity-session')]
+    realtime.request.mockResolvedValue({ session_id: 'runtime-1', stored_session_id: 'identity-session', running: false })
+    await chat.selectSession('identity-session', 'alpha')
+    let accept: (value: unknown) => void = () => {}
+    realtime.request.mockImplementation((method: string) => method === 'prompt.submit'
+      ? new Promise(resolve => { accept = resolve }) : Promise.resolve({}))
+    const sending = chat.send('查看服务器情况')
+    await vi.waitFor(() => expect(realtime.request.mock.calls.some(([method]) => method === 'prompt.submit')).toBe(true))
+    const local = chat.messages.find(message => message.role === 'user')!
+    const persisted = { ...local, id: '1155', serverMessageId: '1155', stage: 'settled' as const }
+    api.getMessages.mockResolvedValue({ messages: [persisted], total: 1, returned: 1, hasMore: false })
+    emit('sessions.changed', { reason: 'cache.synced', profile: 'alpha', session_id: 'identity-session' })
+    await vi.waitFor(() => expect(chat.messages[0]?.id).toBe('1155'))
+    accept({ status: 'streaming' })
+    await sending
+    expect(chat.messages.filter(message => message.role === 'user')).toHaveLength(1)
+    expect(chat.messages[0]).toMatchObject({ id: '1155', clientMessageId: local.clientMessageId, stage: 'settled' })
+  })
+
   it('settles a run and clears an interaction completed while the viewer was offline', async () => {
     await chat.selectSession('stored-1', 'alpha')
     emit('message.start')

tests/client/chatStore.test.ts

@@ -45,6 +45,32 @@ describe('chat message reducer', () => {
     expect(merged[0].stage).toBe('settled')
   })
 
+  it.each(['append', 'snapshot', 'prepend'] as const)('repairs both cached echoes when history restores identity (%s)', position => {
+    const local = { ...message('local', '查看服务器情况', 'client-1'), stage: 'accepted' as const }
+    const oldHistory = { ...message('1155', '查看服务器情况'), serverMessageId: '1155', timestamp: 2 }
+    const history = { ...oldHistory, clientMessageId: 'client-1' }
+    const merged = mergeChatMessages([local, oldHistory], [history], position)
+    expect(merged).toHaveLength(1)
+    expect(merged[0]).toMatchObject({ id: '1155', clientMessageId: 'client-1', stage: 'settled', timestamp: 2 })
+  })
+
+  it('keeps persisted identity and delivery state when a peer receipt arrives after history', () => {
+    const current = state()
+    current.messages = [{ ...message('1155', '查看服务器情况', 'client-1'), serverMessageId: '1155' }]
+    const updated = applyChatEvent(current, event('run.peer_user_message', {
+      message_id: 'user:web:prompt:client-1', client_message_id: 'client-1', text: '查看服务器情况', status: 'accepted',
+    }))
+    expect(updated.messages).toHaveLength(1)
+    expect(updated.messages[0]).toMatchObject({ id: '1155', serverMessageId: '1155', stage: 'settled' })
+  })
+
+  it('retains two separate identical submissions while reconciling both echoes', () => {
+    const local = ['first', 'second'].map(id => ({ ...message(id, '查看服务器情况', id), stage: 'accepted' as const }))
+    const history = local.map((row, index) => ({ ...row, id: String(index + 1), serverMessageId: String(index + 1), stage: 'settled' as const }))
+    expect(mergeChatMessages(local, history, 'snapshot').map(row => [row.id, row.clientMessageId, row.stage]))
+      .toEqual([['1', 'first', 'settled'], ['2', 'second', 'settled']])
+  })
+
   it('keeps history and a realtime assistant reply in chronological order', () => {
     const optimistic = { ...message('local-user', '新问题', 'client-1'), timestamp: 30, stage: 'pending' as const }
     const older = { ...message('history-user', '旧问题'), timestamp: 10, sequence: 10 }

tests/server/chatMessageIdentity.test.ts

@@ -0,0 +1,169 @@
+// @vitest-environment node
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { ChatCacheStore } from '../../src/server/chatCache.js'
+import { mergeChatMessages } from '../../src/client/utils/messageReducer'
+import { normalizeChatMessage } from '../../src/client/utils/normalize'
+
+const fixtures: { home: string; store: ChatCacheStore }[] = []
+const now = 1_789_089_547_214
+const text = '查看服务器情况'
+const scope = ['owner', 'server', 'session'] as const
+type Row = Record<string, unknown>
+function fixture() {
+  const home = mkdtempSync(join(tmpdir(), 'chat-identity-'))
+  const f = { home, store: new ChatCacheStore(home) }
+  fixtures.push(f)
+  f.store.recordCommand(...scope, 'session.create', { source: 'web' })
+  return f
+}
+function response(messages: Row[], offset = 0, total = messages.length + offset) {
+  return { status: 200, headers: new Headers(), body: Buffer.from(JSON.stringify({
+    session_id: scope[2], messages, pagination: { total, offset },
+  })) }
+}
+function submit(store: ChatCacheStore, client = 'local', at = now, options: Row = {}, outcome = 'confirmed') {
+  const clock = vi.spyOn(Date, 'now').mockReturnValue(at)
+  store.recordCommand(...scope, 'prompt.submit', { text, _delivery_id: `web:prompt:${client}`, ...options })
+  clock.mockReturnValue(at + 34)
+  if (outcome !== 'pending') store.recordEvent(...scope, {
+    type: `command.${outcome}`, payload: { delivery_id: `web:prompt:${client}` }, delivery_id: `receipt:${client}`,
+  })
+  clock.mockRestore()
+}
+function row(id = '1155', at = now + 39): Row { return { id, role: 'user', content: text, timestamp: at / 1000 } }
+function sync(store: ChatCacheStore, messages: Row[], force = false) {
+  return store.applySync(...scope, store.revision(...scope), [{
+    key: 'messages', kind: 'messages', response: response(messages), startedAt: now + 2000,
+  }], force)
+}
+function read(store: ChatCacheStore, offset = 0, limit = 100): Row[] {
+  return JSON.parse(store.messagePage(...scope, offset, limit)!.response.body.toString()).messages
+}
+afterEach(() => {
+  vi.restoreAllMocks()
+  for (const f of fixtures.splice(0)) {
+    if (!f.store.isClosed) f.store.close()
+    rmSync(f.home, { recursive: true, force: true })
+  }
+})
+
+describe('ordinary chat submission identities', () => {
+  it('reconciles the real receipt-before-persistence shape all the way through the client reducer', () => {
+    const { store } = fixture()
+    submit(store)
+    expect(sync(store, [row()])).toBe(true)
+    const history = read(store).map(value => normalizeChatMessage(value, scope[2], scope[1]))
+    expect(history[0]).toMatchObject({ id: '1155', clientMessageId: 'local' })
+    const optimistic = { ...history[0]!, id: 'local', serverMessageId: undefined, stage: 'accepted' as const }
+    const oldHistory = { ...history[0]!, clientMessageId: undefined }
+    const merged = mergeChatMessages([optimistic, oldHistory], history, 'snapshot')
+    expect(merged).toHaveLength(1)
+    expect(merged[0]).toMatchObject({ id: '1155', stage: 'settled' })
+    expect(read(store)).toHaveLength(1)
+  })
+
+  it.each([100, 3000])('preserves independent identical sends %i ms apart and unrelated older history', gap => {
+    const { store } = fixture()
+    submit(store, 'first')
+    submit(store, 'second', now + gap)
+    sync(store, [row('old', now - 60_000), row('1'), row('2', now + gap + 39)])
+    expect(read(store).map(value => [value.id, value.client_message_id]))
+      .toEqual([['old', undefined], ['1', 'first'], ['2', 'second']])
+  })
+
+  it('retains bindings across restart, forced refresh and subsequent upstream pages without client IDs', () => {
+    const f = fixture()
+    submit(f.store)
+    sync(f.store, [row()])
+    f.store.close()
+    f.store = new ChatCacheStore(f.home)
+    f.store.db.prepare('DELETE FROM chat_events').run()
+    sync(f.store, [row()], true)
+    f.store.putSnapshot(scope[0], 'again', 'messages', scope[1], scope[2], response([row()]))
+    expect(read(f.store)[0]).toMatchObject({ id: '1155', client_message_id: 'local' })
+  })
+
+  it('repairs existing local history from durable commands without fetching upstream or restarting', () => {
+    const { store } = fixture()
+    submit(store)
+    sync(store, [row()])
+    // Simulate a pre-upgrade cache which already replaced its temporary row.
+    store.db.prepare('DELETE FROM chat_message_identities').run()
+    store.db.prepare("UPDATE chat_messages SET data=json_remove(data,'$.client_message_id')").run()
+    expect(read(store)[0]).toMatchObject({ id: '1155', client_message_id: 'local' })
+  })
+
+  it.each(['pending', 'rejected', 'queued', 'gateway-queued', 'ambiguous', 'outside-window'])('does not guess identity for %s submissions', mode => {
+    const { store } = fixture()
+    submit(store, 'local', now, mode === 'queued' ? { queued: true } : {},
+      mode === 'pending' || mode === 'rejected' ? mode : 'confirmed')
+    if (mode === 'gateway-queued') store.db.prepare("UPDATE chat_events SET payload=json_set(payload,'$.status','queued') WHERE event_type='command.confirmed'").run()
+    const rows = mode === 'ambiguous' ? [row('1'), row('2', now + 40)]
+      : [row('1', mode === 'outside-window' ? now + 30_000 : now + 39)]
+    sync(store, rows, true)
+    expect(read(store).map(value => value.client_message_id)).toEqual(rows.map(() => undefined))
+  })
+
+  it('keeps an unconfirmed send when an older identical user message occupies its position', () => {
+    const { store } = fixture()
+    submit(store, 'pending', now, {}, 'pending')
+    expect(() => sync(store, [row('old', now - 60_000)]))
+      .toThrow('Unconfirmed submission must be reconciled')
+    expect(read(store)[0]).toMatchObject({ id: 'user:web:prompt:pending', status: 'pending' })
+  })
+
+  it('uses an explicit upstream identity for queued messages without timestamp heuristics', () => {
+    const { store } = fixture()
+    submit(store, 'queued', now, { queued: true })
+    sync(store, [{ ...row('1', now + 3600_000), client_message_id: 'queued' }])
+    expect(read(store)[0]).toMatchObject({ id: '1', client_message_id: 'queued' })
+  })
+
+  it('includes other cached pages when refusing an ambiguous legacy match', () => {
+    const { store } = fixture()
+    submit(store)
+    sync(store, [row('1'), row('2', now + 40)], true)
+    expect(read(store, 0, 1)[0]).not.toHaveProperty('client_message_id')
+    expect(read(store, 1, 1)[0]).not.toHaveProperty('client_message_id')
+  })
+
+  it('does not bind the first page of an ambiguous multi-page refresh', () => {
+    const { store } = fixture()
+    submit(store)
+    store.applySync(...scope, store.revision(...scope), [
+      { key: 'tail', kind: 'messages', response: response([row('2', now + 40)], 0, 2), startedAt: now + 2000 },
+      { key: 'older', kind: 'messages', response: response([row('1')], 1, 2), startedAt: now + 2000 },
+    ], true)
+    expect(read(store).map(value => value.client_message_id)).toEqual([undefined, undefined])
+  })
+
+  it('supports delayed agent startup using its run-start event', () => {
+    const { store } = fixture()
+    submit(store)
+    vi.spyOn(Date, 'now').mockReturnValue(now + 5000)
+    store.recordEvent(...scope, { type: 'message.start', payload: {}, delivery_id: 'start' })
+    sync(store, [row('1', now + 4999), { id: 'answer', role: 'assistant', content: '完成', timestamp: (now + 6000) / 1000 }])
+    expect(read(store)[0]).toMatchObject({ id: '1', client_message_id: 'local' })
+  })
+
+  it('scopes bindings to owner, profile and session, migrates them with runtime routes and deletes them with the session', () => {
+    const { store } = fixture()
+    store.recordRoute(...scope, 'runtime')
+    submit(store)
+    sync(store, [row()])
+    for (const other of [['other', scope[1], scope[2]], [scope[0], 'other', scope[2]], [scope[0], scope[1], 'other']]) {
+      const [owner, profile, session] = other as [string, string, string]
+      store.recordCommand(owner, profile, session, 'session.create', {})
+      store.putSnapshot(owner, `other-${other.join(':')}`, 'messages', profile, session, response([row()]))
+      const result = JSON.parse(store.messagePage(owner, profile, session, 0, 100)!.response.body.toString())
+      expect(result.messages[0]).not.toHaveProperty('client_message_id')
+    }
+    store.recordRoute(scope[0], scope[1], 'compressed', 'runtime')
+    expect(store.db.prepare('SELECT session_id FROM chat_message_identities').all()).toEqual([{ session_id: 'compressed' }])
+    store.deleteSession(scope[0], scope[1], 'compressed')
+    expect(store.db.prepare('SELECT * FROM chat_message_identities').all()).toEqual([])
+  })
+})