优化 Bot 发送回执与确认反馈并提前准备安全令牌

docs/verification/2026-09-13-bot-message-optimization/README.md

@@ -0,0 +1,14 @@
+# Bot 模式消息收发优化
+
+开发分支：三个仓库均为 `codex/bot-message-latency`，使用 `/Users/samien/git/bot-message-latency/` 下独立 worktree。
+基线：Web/服务端 `bed10cb`，Android `8ed6079`。iOS 从 main `52e9f12` 建分支后快进至已提交的最新客户端 `aceb477`；没有修改原 main 或带入原工作区的暂存头像文件。
+
+## 第一阶段
+
+- capabilities 在既有授权和 CSRF 签发器下返回 csrfToken；发送回执包含 requestId、完整用户消息、run、conversation、task 和 cursor。
+- 三端提交确认后解除发送状态；回执按实体事件版本合并，不推进全局重放游标，不覆盖更新的 SSE。
+- 发送补偿在后台按会话合并，正常 SSE 追上回执后不请求详情。
+- iOS、Android 提前准备 CSRF，保留旧服务端 bootstrap 兼容；CSRF 失效仅重试一次并沿用原请求 ID。Web 保留既有登录预热。
+- 捕获提交时的草稿与账号/会话上下文，晚到请求不清除后续输入。
+
+验证：Web/服务端 34 项针对性测试与 TypeScript 检查通过；Android core:model、core:network 测试及 App Kotlin 编译通过；iOS WorkspaceChatTests 在签名模拟器上通过，包含 160 ms 模拟响应下 250 ms 内恢复发送状态的验收。未签名模拟器最初出现 Keychain -34018，改为 ad-hoc 签名后通过。一次服务端旧上下文测试短暂失败，主线独立复核及分支完整重跑均通过。

docs/verification/2026-09-13-mobile-latency/README.md

@@ -56,7 +56,7 @@
 | 增加 160 ms 往返 | 发送恢复 | 844.89 | 214.75 |
 | 增加 160 ms 往返 | 首段进入界面 | 611.76 | 542.45 |
 
-请求记录显示夭夭首次提交前多一次 `/api/realtime/capabilities`，用于初始化 cookie 会话的 CSRF 状态。随后提交确认、发送后刷新照常执行。这解释了为什么首次发送时首段顺序会反转。它不是每条消息都有的 160 ms 固定等待。
+请求记录显示夭夭首次提交前多一次 `/api/app/bootstrap`，用于初始化 cookie 会话的 CSRF 状态。随后提交确认、发送后刷新照常执行。这解释了为什么首次发送时首段顺序会反转。它不是每条消息都有的 160 ms 固定等待。
 
 ## 实际请求链与判断
 

src/client/components/workspace/transcriptStore.ts

@@ -1,6 +1,6 @@
 import type { AgentAssignment, AgentGoal } from '@shared/agentTasks'
 import { compareWorkspaceConversations } from '@shared/workspace'
-import type { WorkspaceAgent, WorkspaceConversation, WorkspaceEvent, WorkspaceInteraction, WorkspaceMessage, WorkspaceRun, WorkspaceTask } from '@shared/workspace'
+import type { WorkspaceAgent, WorkspaceConversation, WorkspaceEvent, WorkspaceInteraction, WorkspaceMessage, WorkspaceRun, WorkspaceTask, WorkspaceSendReceipt } from '@shared/workspace'
 
 export interface WorkspaceDetail {
   conversation: WorkspaceConversation
@@ -70,6 +70,15 @@ export class WorkspaceTranscriptStore {
   cursor = 0
   readonly selectedTasks = new Map<string, string>()
   private reads = new Map<symbol, WorkspaceEvent[]>()
+  private entityVersions = new Map<string, number>()
+  private snapshotFloor = 0
+  acceptReceipt(receipt: WorkspaceSendReceipt): void {
+    if (receipt.cursor === undefined) return
+    for (const [type, data] of [['message.changed', receipt.message], ['run.changed', receipt.run],
+      ['conversation.changed', receipt.conversation], ['task.changed', receipt.task]] as const) {
+      if (data) this.accept({ seq: receipt.cursor, type, conversationId: receipt.run.conversationId, data })
+    }
+  }
   beginRead(): symbol { const token = Symbol(); this.reads.set(token, []); return token }
   cancelRead(token: symbol): void { this.reads.delete(token) }
   finishRead(token: symbol, incoming: WorkspaceDetail, preserveHistory = true): WorkspaceDetail {
@@ -98,6 +107,7 @@ export class WorkspaceTranscriptStore {
     return detail
   }
   hydrate(snapshot: WorkspaceSnapshot): void {
+    this.entityVersions.clear(); this.snapshotFloor = snapshot.cursor
     this.agents = snapshot.agents; this.conversations = snapshot.conversations; this.cursor = snapshot.cursor
     this.details.clear()
     for (const detail of snapshot.details) this.details.set(transcriptKey(detail.conversation.id, detail.task?.id), detail)
@@ -110,8 +120,15 @@ export class WorkspaceTranscriptStore {
   }
   apply(event: WorkspaceEvent): void {
     if (event.seq <= this.cursor) return
-    for (const events of this.reads.values()) events.push(event)
+    this.accept(event)
+    this.cursor = event.seq
+  }
+  private accept(event: WorkspaceEvent): void {
     const data = event.data as { id: string }
+    const key = `${event.type.split('.')[0]}:${data.id ?? event.conversationId}`
+    if (event.seq < (this.entityVersions.get(key) ?? this.snapshotFloor)) return
+    this.entityVersions.set(key, event.seq)
+    for (const events of this.reads.values()) events.push(event)
     if (event.type === 'agent.changed') this.agents = upsert(this.agents, event.data as WorkspaceAgent)
     if (event.type === 'agent.deleted') this.agents = this.agents.filter(a => a.id !== data.id)
     if (event.type === 'conversation.changed') this.conversations = upsert(this.conversations, event.data as WorkspaceConversation)
@@ -121,6 +138,5 @@ export class WorkspaceTranscriptStore {
         || (event.type === 'task.deleted' && detail.task?.id === data.id)) this.details.delete(key)
       else this.details.set(key, foldDetail(detail, event))
     }
-    this.cursor = event.seq
   }
 }

src/client/views/ConversationsView.vue

@@ -11,7 +11,7 @@ import LocalVmWorkspace from '@/components/workspace/LocalVmWorkspace.vue'
 import ComputerPanel from '@/components/workspace/ComputerPanel.vue'
 import TaskPlan from '@/components/workspace/TaskPlan.vue'
 import type { AgentAssignment } from '@shared/agentTasks'
-import type { WorkspaceTask } from '@shared/workspace'
+import type { WorkspaceSendReceipt, WorkspaceTask } from '@shared/workspace'
 import { workspaceHasUnread } from '@shared/workspace'
 import type { WorkspaceLifecycleAction, WorkspaceLifecyclePreview } from '@shared/workspaceLifecycle'
 import FloatingResourceSearch from '@/components/app/FloatingResourceSearch.vue'
@@ -643,11 +643,25 @@ async function upload(event: Event) {
     ;(event.target as HTMLInputElement).value = ''
   }
 }
+let draftRevision = 0, accountEpoch = 0
+watch([text, files, mentions], () => { draftRevision++ }, { deep: true, flush: 'sync' })
+const sendRefreshes = new Map<string, ReturnType<typeof setTimeout>>()
+function reconcileSend(receipt: WorkspaceSendReceipt, scope: string, id: string, owner: string | undefined) {
+  clearTimeout(sendRefreshes.get(scope))
+  sendRefreshes.set(scope, setTimeout(() => {
+    sendRefreshes.delete(scope)
+    if (disposed || auth.user?.id !== owner || composerKey.value !== scope) return
+    if (receipt.cursor !== undefined && transcriptStore.cursor >= receipt.cursor && eventSource?.readyState === EventSource.OPEN) return
+    void load(id, true).catch(() => { error.value = '消息已发送，正在恢复同步' })
+  }, receipt.cursor === undefined ? 0 : 2000))
+}
 async function send() {
   const c = active.value
   if (!c || busy.value || (!text.value.trim() && !files.value.length)) return
   busy.value = true
   error.value = ''
+  const owner = auth.user?.id, epoch = accountEpoch, revision = draftRevision, scope = composerKey.value, taskId = activeTask.value?.id
+  const submittedText = text.value, submittedFiles = files.value, submittedMentions = [...mentions.value]
   const mode = deliveryMode.value ? 'goal' : 'chat'
   const fingerprint = JSON.stringify([c.id, activeTask.value?.id, text.value, mentions.value, files.value.map(f => f.id), mode])
   if (!pendingRequestId || pendingFingerprint !== fingerprint) {
@@ -655,7 +669,7 @@ async function send() {
     pendingFingerprint = fingerprint
   }
   try {
-    await apiRequest(`/api/app/conversations/${c.id}/messages`, {
+    const receipt = await apiRequest<WorkspaceSendReceipt>(`/api/app/conversations/${c.id}/messages`, {
       method: 'POST',
       body: {
         requestId: pendingRequestId,
@@ -666,19 +680,24 @@ async function send() {
         fileIds: files.value.map((f) => f.id),
       },
     })
-    text.value = ''
-    files.value = []
-    mentions.value = []
-    pendingRequestId = undefined
-    pendingFingerprint = ''
-    deliveryMode.value = false
-    await load(c.id, true)
-    await refresh()
+    if (disposed || accountEpoch !== epoch || auth.user?.id !== owner) return true
+    busy.value = false
+    if (draftRevision === revision && composerKey.value === scope && text.value === submittedText && files.value === submittedFiles
+      && JSON.stringify(mentions.value) === JSON.stringify(submittedMentions)) {
+      text.value = ''; files.value = []; mentions.value = []
+      pendingRequestId = undefined; pendingFingerprint = ''; deliveryMode.value = false
+    }
+    flushEvents()
+    transcriptStore.acceptReceipt(receipt)
+    conversations.value = transcriptStore.conversations
+    const accepted = transcriptStore.get(c.id, taskId)
+    if (accepted && composerKey.value === scope) presentDetail(accepted)
+    reconcileSend(receipt, scope, c.id, owner)
     return true
   } catch (e) {
-    error.value = e instanceof Error ? e.message : '发送失败'
+    if (accountEpoch === epoch) error.value = e instanceof Error ? e.message : '发送失败'
   } finally {
-    busy.value = false
+    if (accountEpoch === epoch) busy.value = false
   }
 }
 
@@ -804,6 +823,7 @@ watch([selected,selectedTask], () => {
 watch(selected,()=>{dockView.value=undefined})
 watch(() => auth.user?.id, (owner, previous) => {
   if (owner === previous) return
+  accountEpoch++; busy.value = false
   generation++
   eventSource?.close(); eventSource = undefined; eventQueue = []
   if (eventFrame !== undefined) cancelAnimationFrame(eventFrame)
@@ -826,6 +846,8 @@ onMounted(async () => {
   }
 })
 onBeforeUnmount(() => {
+  for (const timer of sendRefreshes.values()) clearTimeout(timer)
+  sendRefreshes.clear()
   disposed = true
   eventSource?.close()
   if (eventFrame !== undefined) cancelAnimationFrame(eventFrame)

src/server/app.ts

@@ -424,7 +424,7 @@ export function createApplication(options: ApplicationOptions = {}): Application
   const sharedComputerRouter=sharedComputers.router();app.use(sharedComputerRouter.routes());app.use(sharedComputerRouter.allowedMethods())
   const localVmRouter=localVm.router();app.use(localVmRouter.routes());app.use(localVmRouter.allowedMethods())
   const computerRouter=computerControls.router();app.use(computerRouter.routes());app.use(computerRouter.allowedMethods())
-  const applicationRouter = workspaceRouter(workspace, workspaceRuntime, workspaceNodes, workspaceAssets, uploads, auth, push)
+  const applicationRouter = workspaceRouter(workspace, workspaceRuntime, workspaceNodes, workspaceAssets, uploads, auth, push, csrf)
   app.use(runners.adminRouter().routes())
   app.use(applicationRouter.routes())
   app.use(async (ctx, next) => {

src/server/workspaceRoutes.ts

@@ -18,6 +18,7 @@ import { receiveGroupUploads, type UploadStore } from './uploads.js'
 import { HttpError } from './errors.js'
 import type { PushCoordinator } from './pushCoordinator.js'
 import type { LocalAuthStore } from './localAuth.js'
+import type { CsrfProtection } from './security.js'
 import { compareWorkspaceConversations } from '../shared/workspace.js'
 import type {
   WorkspaceMessage,
@@ -38,6 +39,7 @@ export function workspaceRouter(
   uploads: UploadStore,
   auth: LocalAuthStore,
   push: PushCoordinator,
+  csrf?: CsrfProtection,
 ): Router {
   const router = new Router(),
     owner = (ctx: Koa.Context) => auth.require(ctx).id
@@ -60,7 +62,9 @@ export function workspaceRouter(
   }
   router.get('/api/app/capabilities', (ctx) => {
     owner(ctx)
+    ctx.set('Cache-Control', 'no-store')
     ctx.body = {
+      csrfToken: csrf?.issue(ctx),
       protocolVersion: 1,
       serverKind: 'yaoyao-web',
       features: [
@@ -277,7 +281,15 @@ export function workspaceRouter(
       await runtime.teamTools.requireAvailable(user, store.require<WorkspaceAgent>(user, 'agent', conversation.administratorId))
       if (auth.pushAuthorizationVersion(user) !== version) throw new HttpError(401, '账号授权已变化，请重新登录', 'session_revoked')
     }
-    ctx.body = { run: runtime.send(user, ctx.params.id, input) }
+    const accepted = runtime.send(user, ctx.params.id, input)
+    const run = store.require<WorkspaceRun>(user, 'run', accepted.id)
+    ctx.body = {
+      requestId: input.requestId, run,
+      message: store.messageForDisplay(user, store.require<WorkspaceMessage>(user, 'message', run.messageId)),
+      conversation: store.conversationSummary(user, store.require<WorkspaceConversation>(user, 'conversation', run.conversationId)),
+      task: run.conversationTaskId ? store.requireTask(user, run.conversationId, run.conversationTaskId) : null,
+      cursor: store.cursor(user),
+    }
     ctx.status = 202
   })
   router.put('/api/app/conversations/:id/read', (ctx) => {

src/shared/workspace.ts

@@ -175,6 +175,15 @@ export interface WorkspaceEvent {
   conversationId?: string
   data: unknown
 }
+/** Additional fields are optional while clients can still connect to older servers. */
+export interface WorkspaceSendReceipt {
+  requestId?: string
+  run: WorkspaceRun
+  message?: WorkspaceMessage
+  conversation?: WorkspaceConversation
+  task?: WorkspaceTask | null
+  cursor?: number
+}
 export interface WorkspaceSource {
   nodeId: string
   profile: string

tests/client/workspaceTranscript.test.ts

@@ -86,3 +86,21 @@ describe('Bot transcript cache', () => {
     expect(foldDetail(pending, { seq: 13, conversationId: 'c', type: 'interaction.changed', data: { id: 'approval', resolved: true } }).interactions).toHaveLength(0)
   })
 })
+
+it('merges send receipts without advancing replay or regressing newer events', () => {
+  const store = new WorkspaceTranscriptStore()
+  store.hydrate({ agents: [], conversations: [conversation], details: [detail()], cursor: 10 })
+  const run = { id: 'r', conversationId: 'c', messageId: 'sent', status: 'queued' as const, mentionIds: [], round: 0, createdAt: 1, updatedAt: 1 }
+  const receipt = { run, message: message('sent', 3, 'accepted'), conversation, cursor: 15 }
+  const pending = store.beginRead()
+  store.acceptReceipt(receipt)
+  expect(store.cursor).toBe(10)
+  store.apply(event(12, message('sent', 3, 'stale')))
+  expect(store.get('c')!.messages.at(-1)!.content).toBe('accepted')
+  store.apply(event(16, message('sent', 3, 'newer')))
+  store.acceptReceipt(receipt)
+  expect(store.get('c')!.messages.at(-1)!.content).toBe('newer')
+  expect(store.get('c')!.messages.filter(m => m.id === 'sent')).toHaveLength(1)
+  expect(store.finishRead(pending, detail()).messages.at(-1)!.content).toBe('newer')
+  expect(store.cursor).toBe(16)
+})

tests/server/workspaceRoutes.test.ts

@@ -560,3 +560,22 @@ describe('application workspace HTTP contract', () => {
     expect(upstream.some((p) => p.includes('internal-session') || p.includes('old-id'))).toBe(false)
   })
 })
+
+it('prepares CSRF with capabilities and returns an idempotent committed send receipt', async () => {
+  const caps = await req('get', '/api/app/capabilities').expect(200)
+  expect(caps.body.csrfToken).toBe(csrf)
+  expect(caps.headers['cache-control']).toBe('no-store')
+  const store = runtime.workspace
+  store.createAgent('first', { name: '回执测试', profile: 'default' })
+  const conversation = store.list<any>('first', 'conversation')[0]
+  const input = { requestId: randomUUID(), content: '确认后立即恢复' }
+  const send = () => req('post', `/api/app/conversations/${conversation.id}/messages`).send(input).expect(202)
+  const a = (await send()).body, b = (await send()).body
+  expect(a.requestId).toBe(input.requestId)
+  expect(a.message.id).toBe(a.run.messageId)
+  expect(a.conversation.id).toBe(conversation.id)
+  expect(a.cursor).toBeGreaterThan(0)
+  expect(b.message.id).toBe(a.message.id)
+  expect(b.run.id).toBe(a.run.id)
+  expect(store.messages('first', conversation.id).filter(m => m.role === 'user')).toHaveLength(1)
+})