feat: update message synchronization flow

This commit is contained in:
YBF
2026-09-14 10:02:44 +08:00
parent 211816c16b
commit 7012bf9547
39 changed files with 1607 additions and 108 deletions
@@ -0,0 +1,8 @@
{"file": ".trellis/spec/project/architecture.md", "reason": "跨包所有权与边界验收。"}
{"file": ".trellis/spec/project/async-state-boundaries.md", "reason": "异步命令、ACK 与状态转换检查。"}
{"file": ".trellis/spec/project/database-query-composition.md", "reason": "事务范围与无 JOIN 规则。"}
{"file": ".trellis/spec/server/backend/quality-guidelines.md", "reason": "服务端检查命令和测试要求。"}
{"file": ".trellis/spec/server/backend/error-handling.md", "reason": "HTTP/WS 错误映射验收。"}
{"file": ".trellis/spec/chrome-extension/frontend/quality-guidelines.md", "reason": "扩展检查命令和测试要求。"}
{"file": ".trellis/spec/chrome-extension/frontend/onetalk/durable-sync.md", "reason": "plugin durable reset 与 completion fence 验收。"}
{"file": ".trellis/tasks/09-12-mind-rebuild-conversation-history/research/rebuild-reset-protocol.md", "reason": "重建边界、race matrix 与运行时验证前提。"}
@@ -0,0 +1,81 @@
# 单会话历史重建设计
## 边界与成功语义
Mind 通过 `POST /api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history/rebuild` 请求重建。该接口的成功仅表示 Bright 已在一个数据库事务内删除目标消息/会话级异常并重置目标会话的消息派生索引;它不表示插件已经拉完历史,也不表示 Mind 已经看到新消息。
请求必须具备新的 `rebuild` permission、合法 Mind Origin/Cookie、当前 cutover admission 和精确的 `channelAccountId` scope。Bright 只选择一个同时匹配 Mind scope、binding、authorization version、`rebuild` permission 且 heartbeat lease 新鲜的 canonical plugin。`isPluginOnline()` 仅说明存在连接,不能作为该判断的证据。
操作状态仅在进程内存在,用于同会话互斥、回执等待和连接/策略代际检查;不新增可恢复的 rebuild operation 表。进程重启、插件断线、命令超时或数据库失败会留下稳定原因,Mind 可重新提交同一会话的重建请求。重试是幂等的 scoped clear/reset,不依赖上一次操作仍在内存中。
## HTTP 契约
成功响应为 HTTP 200
```json
{
"scope": { "mindUserId": "…", "workspaceId": "…", "channelAccountId": "…" },
"conversationId": "…",
"rebuildId": "…",
"status": "server_reset_committed",
"resync": { "status": "started" }
}
```
若 Bright 已提交删除、但后续同步命令未确认,仍返回 HTTP 200,`resync` 改为 `{ "status": "failed", "reason": "…" }`。这符合“服务端删除是唯一成功依据”。在 Bright transaction 提交前失败则不返回成功,使用稳定错误:既有授权/CORS 错误、`conversation_not_found``rebuild_in_progress``plugin_offline``plugin_heartbeat_stale``plugin_reset_rejected``plugin_reset_timeout``database_unavailable``internal_error`
public CORS 需显式允许这个 POST 和 `content-type`;不能把 GET-only read middleware 偷换为状态变更授权。
## WebSocket 协议
增加受限的、可复用的 plugin local-storage 命令,不让 Bright 传递任意 IndexedDB store 或 key
| 帧 | 方向 | 核心 payload | 职责 |
| --- | --- | --- | --- |
| `storage.delete.command` | Bright → plugin | `rebuildId`, `{ kind: "conversation_history", conversationId }` | 让插件停止目标会话旧 generation,并删除限定的 durable ledger。 |
| `storage.delete.ack` | plugin → Bright | `rebuildId`, `conversationId`, `status`, `reason?` | 只在 IndexedDB transaction `oncomplete` 后确认。 |
| `history.sync.command` | Bright → plugin | `rebuildId`, `historyGeneration`, `conversationId`, `mode: "full"` | 在 Bright reset commit 后释放目标会话,并启动 full sync。 |
| `history.sync.ack` | plugin → Bright | `rebuildId`, `conversationId`, `status`, `reason?` | 回报已启动或立即失败,供 HTTP 的 `resync` 字段和 Mind status 使用。 |
| `rebuild.status` | Bright → Mind | `rebuildId`, `conversationId`, `stage`, `reason?` | 发布 pre-commit 失败、local cleared、server committed、sync started/completed/failed。 |
这些帧是新领域协议,不能复用 `send.command`、pending-send coordinator 或 `sendRequestId`。协议 decoder 使用精确 key/方向校验。为了让旧页面观测无法在 reset 后重写新事实,`historyGeneration` 必须成为消息观察、消息 ACK、`sync.complete` 和 anchor snapshot 的强制关联字段;这是一项 breaking wire change,应将共享协议从 v6 升到 v7,而不是把该字段做成可选 fallback。
## 删除和 generation 栅栏
`storage.delete.command` 的插件实现只在 `trade-message-center` IDB 的一个 readwrite transaction 中,按 `(channelAccountId, conversationId)` 删除:
- `onetalk_messages`
- `onetalk_sync_checkpoints`
- `onetalk_sync_candidates`
- `onetalk_sync_anomalies`(只删除有完全相同 conversationId 的行;账号级 anomaly 保留)
不得删除 `onetalk_contact_profiles`、独立数据库 `trade-message-center-onetalk-buyer-facts``onetalk_conversation_bootstraps`、Chrome 配置或其它会话的记录。bootstrap marker 是 account/migration 级调度状态,不是单会话 history ledger。
插件收到 delete command 后把该会话置为 quiesced:等候该会话 queue 和 observation write chain 收敛、取消目标 pending ACK/completion memory,并拒绝旧 generation 的迟到页面观察、ACK 和 page result。删除 ACK 后继续保持 quiesced,不能立即启动 full sync;否则新消息可能在 Bright 随后的 DELETE 前入库。
Bright 收到匹配 ACK 后重新验证 socket canonical generation、lease、binding、authorization version 和 cutover epoch,并在一个 transaction 内:
1. 删除目标 `onetalk_message` 和目标 `onetalk_message_anomaly`
2. 保留 `onetalk_conversation` identity/discovery/list-display 字段、`onetalk_contact_profile``onetalk_buyer_fact`
3. 重置 `lastMessageAtMs``latestMessageId``historyComplete``messageCount``anchorUpdatedAt``syncPhase``syncResult`
4. 写入新的 opaque `historyGeneration`,作为以后该会话所有消息/完成帧的唯一代际。
`historyGeneration` 是持久化的会话事实边界,不是可恢复的 rebuild operation:它让服务端在 reset 后拒绝 pre-reset frame,并让重启后的 Bright 仍能区分旧 generation。全量同步命令把它传回插件;Service Worker 对目标会话的后续 history 和 live observation 均附带该 generation。服务端只接受与 conversation 当前 generation 相等的观察与 completion。
## 重建触发与完成
当前 `maybeBootstrap()` 不能作为触发机制:已完成的 `onetalk_conversation_bootstraps` marker 会抑制 page-ready/reconnect 下的历史 bootstrap,清掉该 marker 又会触发账户级 batch,违反单会话范围。
Bright commit 后发送 `history.sync.command`。插件先刷新 MAIN-world direct-conversation cache,再复用现有 `startSync({ mode: "full" })` 和精确 `onetalk.sync.conversation` page command;不新增第二套 OneTalk 拉取逻辑。`history.sync.ack` 若超时、断线或页面 cache 缺失,属于 post-commit best-effort 失败并带稳定 reasonMind 可安全重试整个 HTTP rebuild。
当同 generation 的 `sync.complete` 已由 Bright 持久化并产生 successful sync status 后,Bright 向插件回发当前 `anchor.snapshot`。这复用现有 plugin completion coordinator,使 checkpoint 从 uploading 收敛;仅收到 plugin `sync.complete` 不足以表示本地 durable completion。`rebuild.status` 会带同一 rebuildId 通知 Mind;它是观察重建是否已拉回的信号,而不是 HTTP 成功判定。
## 缓存错位处理
现有模型中,插件是 durable upload ledger`accepted`/`duplicate` ACK 令本地 candidate 终态,anchor snapshot 只给增量 stop boundary。Bright 单方删除既不会使 confirmed candidate 重传,也不会绕过 completed bootstrap marker 重新拉历史。因此不存在可依赖的“等待自动同步自行修复”机制。
本设计用 plugin clear -> ACK -> Bright transaction -> explicit release 的顺序消除成功路径错位。若 plugin clear 后 Bright transaction 失败,插件保持目标会话 quiesced,HTTP 返回数据库原因;重试会再次 clear 并尝试 transaction。若 Bright 已提交但 sync command 失败,则两端该会话历史可为空;HTTP 仍成功,Mind 收到明确的 `rebuild.status` reason,并以新请求触发下一次 full sync。
## 回滚与观测
提交前失败不删除 Bright 消息;提交后不回滚服务器状态。因为 plugin IndexedDB 与 PostgreSQL 没有跨系统事务,日志/状态只记录 `rebuildId`、conversationId、阶段、稳定 reason,绝不记录消息内容、cookie、binding 或原始 SDK payload。测试必须用可控 Promise 验证每个 await 后的 connection/policy/generation 重校验,且不能把 200 或 `conversation.updated` 误判为 history rebuild completed。
@@ -0,0 +1,10 @@
{"file": ".trellis/spec/project/architecture.md", "reason": "跨包契约、数据边界与模块所有权。"}
{"file": ".trellis/spec/project/async-state-boundaries.md", "reason": "重建单飞、ACK 与跨 await 重校验。"}
{"file": ".trellis/spec/project/structured-value-equality.md", "reason": "rebuildId、generation 与消息事实身份。"}
{"file": ".trellis/spec/project/database-query-composition.md", "reason": "scoped reset 事务保持无 SQL JOIN。"}
{"file": ".trellis/spec/server/backend/error-handling.md", "reason": "HTTP/WS 稳定失败语义。"}
{"file": ".trellis/spec/server/backend/mind-authorization.md", "reason": "Mind session、Origin、CORS 与授权边界。"}
{"file": ".trellis/spec/chrome-extension/frontend/onetalk/runtime-sync.md", "reason": "插件到 Bright 的同步链路与数据边界。"}
{"file": ".trellis/spec/chrome-extension/frontend/onetalk/page-bridge.md", "reason": "页面命令与定向路由契约。"}
{"file": ".trellis/spec/chrome-extension/frontend/onetalk/durable-sync.md", "reason": "IndexedDB、ACK、checkpoint 和恢复语义。"}
{"file": ".trellis/tasks/09-12-mind-rebuild-conversation-history/research/rebuild-reset-protocol.md", "reason": "当前源码证据、reset 序列和竞态风险。"}
@@ -0,0 +1,36 @@
# 单会话历史重建实施计划
## 1. Shared contract and generation fence
1. 将 protocol version 从 v6 升到 v7;在 shared contract 中加入 `rebuild` permission、HTTP rebuild route 常量、稳定 HTTP/action errors、storage delete/sync command/ACK/rebuild status frame 类型、decoder/encoder 和 strict payload tests。
2. 为 target conversation 的 `message.observed``messages.observed`、对应 ACK、`sync.complete``sync.status`/anchor projection 定义必填 `historyGeneration` 关联;更新 wire direction、authenticated route table 和 version-upgrade tests。
3.`onetalk_conversation` schema/migration 和 OneTalk model/repository 中加入 current `historyGeneration`,确保新发现会话和 reset 后会话都拥有明确 generation;禁止 optional/legacy fallback。
4. 为服务端建立一个 rebuild coordinator owner:同 `(channelAccountId, conversationId)` in-process single-flight、fresh canonical-plugin selector、ACK awaiter 和每个 await 后的 connection/policy/authorization revalidation。它不承担跨重启恢复。
## 2. Bright reset and HTTP boundary
1. 在 OneTalk repository/service 添加一个 scoped reset domain operation:确认 direct conversation 存在,在单 PostgreSQL transaction 中删除 target message/anomaly 行,重置所有消息派生字段,写新 generation,保留会话/profile/buyer/list-display字段;不使用 SQL JOIN。
2. 新建 rebuild HTTP route,复用 public origin/cutover/session boundary但不复用 read-only permission helper;显式授权 `rebuild`,更新 CORS preflight 允许 POST。
3. HTTP flow:选 fresh plugin -> send `storage.delete.command` -> wait matching ACK -> revalidate -> commit server reset -> send `history.sync.command` -> bounded wait `history.sync.ack` -> 返回 server-reset-committed + resync result。commit 前失败为 HTTP errorcommit 后 command failure 为 HTTP 200 的 resync failure。
4. 向授权 Mind connections 发布 correlated `rebuild.status`。本 operation 的 success stage 固定为 `server_reset_committed`sync terminal stage是独立观测信息。
## 3. Plugin scoped cleanup and full-sync reuse
1. 扩展 `OneTalkSyncStore`,实现按 conversation 扫描/删除四个 target stores 的单一 IDB transaction;添加 targeted test fixture,覆盖非目标会话、account-level anomaly、profile/buyer/bootstrap 保留和 transaction abort 无 ACK。
2. 在 sync engine/ack coordinator/queue 中实现 per-conversation quiesce and generation owner;清理仅该会话的 active maps、pending ACK/completion,并阻断旧 generation 的迟到结果,不调用全连接 `resetConnection()`
3. 在 worker frame router 新增 storage delete 和 history sync command handlers。delete ACK 只能在 durable clear 后发出;sync command 先刷新 target page cache,再调用现有 `startSync(full)`,为所有 target observation/completion 绑定 generation,并回报 start outcome。
4. 复用 existing `anchor.snapshot` completion path:同 generation `sync.complete` 成功持久化后,接受 Bright refreshed anchor snapshot 才结束 local checkpoint。不要修改为在插件发送 `sync.complete` 时自行宣告成功。
## 4. Tests and validation
1. Contract focused testsv7 decoder directions/exact keys、permission、generation required、unknown/old frame rejection、rebuild status correlation。
2. Extension focused testsIDB scoped clear、quiesce/generation matrix、old page observation/ACK/completion no-op、fresh full start uses exact conversation command、bootstrap marker never triggers account batch、sync-start failure reason。
3. Server focused testsHTTP auth/origin/CORS/errorsfresh-heartbeat exact plugin selectionACK-before-delete ordertransaction scope/preservation; post-commit sync failure still 200duplicate retryMind status correlationplugin replace/revoke/pause/timeout database-failure matrices。
4. Run targeted tests first, then `pnpm typecheck`, relevant package builds, root format/diff checks and GitNexus `detect_changes()` before commit. Backend tests use a 60-second timeout; PostgreSQL integration uses only an isolated test DB and is reported as skipped if unavailable.
5. Runtime smoke, if dedicated test account/plugin are available: POST one conversation rebuild, capture delete command/ACK, inspect four local stores and Bright rows, verify post-commit full-sync command, refreshed anchor snapshot and Mind status. Never exercise a real customer conversation.
## Review gates
- Before code edits: run GitNexus upstream impact on every target handler/function and report any HIGH/CRITICAL blast radius.
- Before integrating server/plugin changes: verify one contract owner, no raw IndexedDB store-name payload, no raw OneTalk SDK data crossing the bridge, and no secondary history source.
- Before start/commit: verify all generation checks cross each await, `server_reset_committed` is not confused with `sync_completed`, and the diff does not alter send or account-wide bootstrap behavior.
@@ -0,0 +1,68 @@
# Expose HTTP endpoint for Mind to rebuild one conversation's history
## Goal
Allow Mind to start an unconditional, full reconstruction for exactly one authorized OneTalk conversation through a new Bright HTTP endpoint. Reconstruction first removes the targeted message and history-sync state from both Bright and the connected plugin, retains the conversation record and customer-profile facts, then starts a fresh full OneTalk history collection.
## Confirmed facts
- Bright already exposes `GET /api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/messages`; it authorizes the requesting Mind session for the path account and returns one page of persisted messages plus an opaque cursor. [apps/server/src/http/onetalk/messages.ts:15-51; apps/server/src/http/onetalk/read.ts:72-121]
- This public route uses the Mind session authorization `read` permission, exact configured origin/CORS handling, and a cutover-policy admission epoch. [apps/server/src/http/onetalk/public-context.ts:65-120; apps/server/src/http/onetalk/public-context.ts:143-215]
- The only current `historyComplete` gate is the internal summary listener. It requires an explicit time window and returns `503 history_incomplete` with `Retry-After: 30` until the stored conversation is complete. [apps/server/src/http/onetalk/summary.ts:37-84; apps/server/src/http/onetalk/read.ts:83-114; apps/server/src/onetalk/read-service.ts:160-176]
- Mind's current WebSocket command surface accepts only `send.request` and `heartbeat`; OneTalk collection/synchronization is not currently a command exposed to Mind. [apps/server/src/websocket/mind/index.ts:50-91]
- Stored facts remain scoped by `channelAccountId + conversationId + messageId`; the read service creates opaque cursors bound to both the account and conversation. [apps/server/src/onetalk/read-service.ts:54-76; apps/server/src/onetalk/read-service.ts:186-216]
- Bright has no per-conversation destructive history operation. The message table is separate from `onetalk_conversation`; profile and buyer-fact tables deliberately have no foreign key to the conversation row. [apps/server/src/database/schema/onetalk.ts:60-180; apps/server/src/database/schema/onetalk.ts:182-315]
- Plugin IndexedDB stores per-conversation messages, checkpoints, candidates, and scoped anomalies separately from contact-profile ledger records. Its current public store exposes no per-conversation deletion operation. [apps/chrome-extension/src/onetalk/service-worker/storage.ts:8-177; apps/chrome-extension/src/onetalk/service-worker/storage.ts:574-744]
- The Bright-to-plugin protocol has no history-rebuild command or acknowledgement today; the only server-initiated plugin command is the separately coordinated send command. [packages/onetalk-contract/src/wire.ts:43-164; apps/server/src/websocket/registry.ts:34-119; apps/server/src/websocket/pending-send-coordinator.ts:123-255]
## Agreed product decisions
- Add a **new** HTTP endpoint rather than changing the existing paginated message-read endpoint.
- The action is an unconditional reconstruction: it must not reject merely because `historyComplete` is already true, false, or the stored history is stale.
- Scope is one `channelAccountId` and one `conversationId` per request.
- Preserve the existing Bright `onetalk_conversation` row and customer-profile facts on both Bright and the plugin.
- Clear the targeted message facts and history indexes on both Bright and the plugin before starting a fresh full collection.
- Add a reusable Bright-to-plugin WebSocket storage-deletion command with a correlated acknowledgement. The concrete rebuild operation uses its `conversation_history` target rather than allowing Bright to name arbitrary IndexedDB stores.
- The required orchestration order is: Mind request -> Bright verifies one fresh, authorized plugin heartbeat -> plugin durable deletion acknowledgement -> Bright deletes/reset its scoped facts -> Bright sends an explicit full single-conversation synchronization command -> Mind receives correlated follow-up status.
- Bright's committed deletion/reset of the target conversation's server-side message facts and indexes is the sole success criterion for the HTTP rebuild operation. Plugin resynchronization is a best-effort follow-up: its failure does not roll back or change a committed server reset.
- Do not persist rebuild operations for crash recovery. A retry is allowed to repeat the idempotent plugin clear and Bright reset; a pre-commit failure returns its reason, while a post-commit resynchronization failure is reported separately as a non-successful follow-up.
## Requirements
- The endpoint is a state-changing HTTP operation, not a history read. It must use the existing Mind session/origin/cutover boundary and require a new explicit `rebuild` permission; the existing `read` permission must not authorize deletion.
- The endpoint is expected to be `POST /api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history/rebuild`. It returns success only after the Bright delete/reset transaction commits, together with a request/rebuild ID and the best-effort resynchronization dispatch outcome; it must not return success merely because the request was parsed.
- The generic plugin command has a semantic target such as `conversation_history` and a caller-generated operation ID; it does not expose implementation store names over the wire. Its acknowledgement proves the scoped IndexedDB transaction committed before Bright removes its own facts.
- The server reset must delete only target-scope `onetalk_message` and `onetalk_message_anomaly` facts. In the same transaction it must reset the conversation's message-derived state (`lastMessageAtMs`, `latestMessageId`, `historyComplete`, `messageCount`, anchor state, and sync result), without deleting the conversation row or customer-profile/buyer facts. It retains the account-level anomalies that have no `conversationId`.
- The plugin reset must delete records for the target `(channelAccountId, conversationId)` from `onetalk_messages`, `onetalk_sync_checkpoints`, `onetalk_sync_candidates`, and `onetalk_sync_anomalies`. It must leave `onetalk_contact_profiles` and the separate `trade-message-center-onetalk-buyer-facts` database intact. The account-wide `onetalk_conversation_bootstraps` marker is not a per-conversation history index and must not be deleted or rewritten by this command.
- The rebuild coordinator must choose exactly one canonical plugin whose heartbeat lease is fresh and whose scope, binding, authorization version, and `rebuild` permission match the Mind request. Existing `isPluginOnline` is insufficient because it only proves a connection exists, not a fresh heartbeat.
- An explicit server-to-plugin full-sync command must refresh the target page cache and reuse the existing plugin `startSync({ mode: "full" })` / exact `onetalk.sync.conversation` page-command path. Waiting for `maybeBootstrap()` is incorrect: a completed account-wide bootstrap marker suppresses reconnect/page-ready bootstrap work.
- The storage deletion command leaves the target conversation quiesced after its ACK. Only after Bright commits the server reset may a correlated single-conversation full-sync command release it. This prevents a newly observed full-history batch from being deleted by the server transaction that follows the plugin ACK.
- The storage deletion and resync commands need a single per-conversation operation/generation correlation. It must flow through targeted page observations/progress, plugin ACK/completion, and Bright's observation writer so that late pre-reset messages, ACKs, or page results are discarded instead of being written into the reconstructed generation.
- After Bright accepts the reset-triggered `sync.complete`, it must send the existing refreshed `anchor.snapshot` evidence back to the plugin so the plugin's existing completion coordinator can close the local checkpoint; a plugin-originated `sync.complete` alone is not its durable local completion signal.
- Mind receives a correlated `rebuild.status` WebSocket event for pre-commit rejection, storage-cleared, server-reset-committed, sync-started, sync-completed, and best-effort sync failure stages. HTTP success means `server-reset-committed`; it does not claim that full history has already reappeared. A bare uncorrelated `sync.status` or `conversation.updated` must never be treated as reconstruction completion.
- Mind must not receive unscoped, account-wide, or raw OneTalk SDK payloads.
- Authorization, origin handling, cutover admission, and error semantics must stay fail-closed.
## Acceptance criteria
- [ ] An authorized Mind caller can POST one exact direct conversation; the route rejects a missing/invalid `rebuild` permission, mismatched account scope/origin, unavailable cutover, unknown conversation, concurrent same-conversation operation, missing plugin, stale heartbeat, reset rejection, timeout, and database failure with stable reasons and no unintended server deletion.
- [ ] Before Bright deletes any server facts, exactly one matching fresh plugin receives a semantic `conversation_history` storage-delete command and returns a correlated durable ACK; a stale/replaced/revoked connection cannot advance the operation.
- [ ] The plugin clear changes only target-scope message/checkpoint/candidate/conversation anomaly records. It preserves target profile/buyer records, account bootstrap marker, account-level anomalies, configuration, and every other conversation's state.
- [ ] On a matching ACK, one Bright transaction deletes only target message/conversation-anomaly rows and resets the stated derived conversation fields while preserving the conversation, profile, buyer facts, and list-display fields.
- [ ] A committed Bright reset returns HTTP success even if the subsequent sync dispatch/start fails; the response and correlated Mind status expose the exact post-commit failure reason. A pre-commit failure returns no success.
- [ ] A post-commit release invokes only the target full-sync path, never account-wide bootstrap/batch collection; a matching generation is required for all resulting observations and completion. Late old-generation page results, observations, ACKs, completions, replacement sockets, and policy changes cannot rewrite the reset conversation.
- [ ] Bright sends refreshed anchor evidence after accepting the reset generation's sync completion; Mind can distinguish `server_reset_committed`, `sync_started`, `sync_completed`, and best-effort sync failure by `rebuildId`, without treating HTTP success or a generic conversation update as full-history completion.
## Out of scope
- Reconstructing multiple conversations or the whole account in one request.
- Duplicating messages into Mind's database or making Mind read Bright's database directly.
- Deleting the conversation row, contact-profile facts, buyer facts, or account-wide bootstrap state.
- Changing the OneTalk fact identity, live WebSocket send protocol, or unrelated history reads.
- Clearing the account-wide discovery bootstrap marker, the contact-profile ledger, buyer-fact ledger, or unrelated conversations' IndexedDB records.
## Cache-mismatch behavior
- The plugin's IndexedDB is a durable upload ledger, not a server-history cache. `accepted`/`duplicate` ACKs mark local candidates as delivered, and `anchor.snapshot` supplies only the server's newest stop boundary for later incremental reads.
- Bright deleting a conversation does not notify an existing plugin. A plugin with confirmed local candidates will not replay them; an account bootstrap marker already in `complete` also suppresses automatic historical bootstrap. Therefore neither side can repair this deletion mismatch by waiting for current automatic sync alone.
- The rebuild flow repairs the mismatch by deleting the plugin's target ledger first, then committing Bright's target reset, then explicitly releasing the plugin's existing one-conversation full-sync path. If the server transaction fails after plugin deletion, the request fails with its database reason and retry repeats the safe clear/reset sequence. If the post-commit sync dispatch/start fails, the HTTP request remains successful and Mind receives the specific follow-up reason; retry can initiate another full reconstruction.
@@ -0,0 +1,118 @@
# 研究:单会话重建 reset 协议与执行序列
- 查询:核对 Mind HTTP → Bright 授权/插件 presence → Bright→插件 IndexedDB 清理 → ACK → Bright 数据库清理 → 单会话 full sync → Mind 完成通知;确认现有 store/key、服务端表/字段、`maybeBootstrap`/`startSync` 复用性及 reset 竞态。
- 范围:internal
- 日期:2026-09-12
- 协调类别:cross-cutting
- 阻塞:HTTP rebuild 路由与授权、shared WebSocket contract/decoder、Bright plugin command routing/registry、插件 IndexedDB reset 与 sync engine、Bright OneTalk repository/状态通知;这些 writer 共享 resetId/generation 与 ACK/完成语义,不能各自先写。
- 共享边界:`packages/onetalk-contract/src/{wire,decoder,conversation-sync,messages}.ts``apps/server/src/{http/onetalk/public-context.ts,websocket/{connection-store,registry}.ts,websocket/plugin/index.ts,onetalk/{repository,model,service}.ts}``apps/chrome-extension/src/onetalk/service-worker/{storage.ts,routing/frame-router.ts,sync-engine.ts,sync-engine/bootstrap-coordinator.ts,sync-engine/ack-completion.ts}``apps/chrome-extension/src/onetalk/main-page/current-conversation-history/{all-conversations.ts,page-command.ts}.ts`
- 证据基线:checkout `/Users/ybf/code/trade-message-center-worktree`HEAD `ce61b4cc72f72ba96167c59b9791241856d4c3cc``git status --short --branch` 仅显示任务目录未跟踪文件,无产品代码 dirty path。GitNexus worktree index 记录同一 HEAD2026-09-12 08:34 UTC),未运行服务、浏览器、真实 WebSocket 或 PostgreSQL runtime probe。
- 复用证据与缺口:既有任务 `prd.md` 已覆盖单会话、保留会话/profile、清理消息/同步状态的目标;本文件刷新了当前源码锚点。仓库没有 reset/delete command、reset ACK、重建任务持久化、服务端主动选 plugin/send command 或 Mind rebuild completion contract;这些是实现前缺口。
## 发现
### 1. 插件 IndexedDB 的精确清理边界
`apps/chrome-extension/src/onetalk/service-worker/storage.ts:15-23` 现有数据库是 `trade-message-center` v8,固定 object stores 为:
- `onetalk_messages`:记录含 `channelAccountId``conversationId``messageId`key 为 `JSON.stringify([channelAccountId, conversationId, messageId])``storage.ts:54-59,253-270`)。只删目标 account+conversation,保留其它会话。
- `onetalk_sync_candidates`:同一三元业务 key,保存页面事实和 `pending_ack/confirmed/anomaly/rejected/awaiting_anchor` 状态(`storage.ts:81-99,467-496`)。必须与 messages 一起删,否则旧候选会重新上传。
- `onetalk_sync_checkpoints`key 为 `JSON.stringify([channelAccountId, conversationId])`,保存 mode、phase、anchor、分页位置、`historyComplete``latestMessageId``completionSent` 等(`storage.ts:61-79,206-216,589-617`)。这是当前所谓历史/同步 index 的主要持久化记录,应删除后由 full run 新建。
- `onetalk_sync_anomalies`key 为 account、可选 conversation、code、排序后的 fields、source;记录按 `channelAccountId` 和可选 `conversationId` 过滤(`storage.ts:101-111,237-250,718-729`)。目标会话 anomaly 应删;`conversationId` 缺失的 account-level diagnostic 不能误删。
`openSyncDatabase` 只以 `keyPath: "key"` 建 store,没有 `createIndex``storage.ts:273-317`);因此“删除 history index”不是删一个 IDB index,而是按完整字段遍历/删除上述记录。清理要在同一 readwrite transaction 中完成,并以 transaction `oncomplete` 作为 ACK 前置证据。现有 profile ledger 使用同一 pair key,但独立于消息 pipeline`storage.ts:113-131,775-881`),必须保留 `onetalk_contact_profiles``onetalk_conversation_bootstraps` 的 key 是 `[channelAccountId, migrationId]`,是 account-wide discovery marker,不是单会话消息 state;单会话 reset 不应清除它(`storage.ts:25-52,222-223,319-363`)。Chrome 配置、deviceId 和其它渠道数据也不在删除范围。
### 2. Bright 数据库的清理/保留边界
当前 OneTalk schema 只有四类相关表:
- 删除 `onetalk_message` 中完整 scope `channel_account_id + conversation_id` 的全部行。表主键和查询 key 是三元组,时间读取索引为 `(channel_account_id, conversation_id, sent_at_ms, message_id)``apps/server/src/database/schema/onetalk.ts:60-119``apps/server/src/onetalk/repository.ts:83-93,333-361`)。物理 B-tree 会由 PostgreSQL 随 DELETE 维护,不应把索引对象本身当业务数据删除。
- 清除 `onetalk_message_anomaly` 中同一 account+conversation 的诊断行,保留 conversationId 为空的账号级诊断(`schema/onetalk.ts:255-301`)。不清理会留下旧同步异常,污染重建后的诊断/结果。
- 保留 `onetalk_conversation` 主记录及发现元数据:主键 `(channel_account_id, conversation_id)`,并保留 `participantIds``bizType``conversationType``conversationKind``joinTimeMs``modifyTimeMs``firstDiscoveredAt`,以及面向会话列表的 `lastContactTimeMs`/`messagePreview``schema/onetalk.ts:121-180`)。更新 `lastObservedAt` 为 reset 时间是合理的审计副作用。
- 在同一 server transaction 中重置消息派生同步字段:`syncPhase='initial'``syncResult='incomplete'``latestMessageId=NULL``historyComplete=false``messageCount=0``anchorUpdatedAt=NULL`;建议同时把 `lastMessageAtMs=NULL`,因为它是消息事实派生值,而现有写入只做 monotonic max`repository.ts:99-106,650-687`),不清空会在新历史比旧数据更旧/不完整时保留陈旧活动时间。`lastContactTimeMs`/`messagePreview` 是会话列表活动摘要,不属于消息事实,建议保留并让后续 discovery 继续按既有 monotonic 规则更新。
- 保留 `onetalk_contact_profile` 整行(`schema/onetalk.ts:182-218`)以及 `onetalk_buyer_fact` 整行(`schema/onetalk.ts:220-253`)。两者均以 account+conversation pair 独立存储,读取时由 `read-repository.ts:122-180,254-329` 受限查询后以内存组合;它们是“对应客户资料/买家资料”,不能随消息 reset 删除。
服务端 reset repository 需要以 account+conversation 条件包住 DELETE/UPDATE,不能按 `conversationId` 单独删,也不能借用公共 read path。对 HTTP 触发的 destructive operation,应在插件 ACK 后开启事务,并在事务写入前/后使用同一连接/policy guard;数据库事务 rollback 只覆盖 Bright,无法回滚已提交的插件 IndexedDB 清理。
### 3. 现有授权、presence 和 WS 结构不能直接完成该序列
当前 public HTTP 上下文只实现 `authorizeRead`:读取 Cookie、校验精确 `channelAccountId`、要求 `read`,并在请求前后检查 cutover epoch`apps/server/src/http/onetalk/public-context.ts:65-94,158-204`)。新路由必须使用同一 Origin/Cookie/cutover 边界;CORS preflight 当前只允许 GET`public-context.ts:15-16,132-155`),POST rebuild 需要显式加入,不能复用 GET-only 语义。contract 的授权 operation 已有 `sync`,权限集合目前只有 `read/send``packages/onetalk-contract/src/connection.ts:7-14`);最小兼容方案是按 `operation: "sync"` 授权并要求 `read`,是否增加独立 destructive permission 是产品决策。
`OneTalkConnectionStore.isPluginLeaseFresh` 已按 canonical registered plugin 和 `lastHeartbeatAtMs > now - heartbeatTimeoutMs` 判断新鲜度(`apps/server/src/websocket/connection-store.ts:60-78,246-253`),默认 heartbeat interval/timeout 为 25s/75s`connection-store.ts:104-116`)。但 `OneTalkConnectionRegistry` 对外只暴露 `isPluginOnline`,它只代表存在同 scope plugin,不检查 lease 新鲜度(`registry.ts:30-41,106-119`)。新 flow 需要一个 registry-owned “唯一 canonical plugin + fresh lease + binding/scope/permission” 选择/guard;不能把 `isPluginOnline` 当 heartbeat 证明,也不能按 account 广播或跨 binding fallback。
当前 Bright→插件只有 `send.command`,且它通过 pending-send coordinator 路由页面发送并要求 `sendRequestId``packages/onetalk-contract/src/wire.ts:43-70,88-105``apps/chrome-extension/src/onetalk/service-worker/routing/frame-router.ts:40-59``send-command-flow.ts:36-97`)。复用 `send.command` 做删除或同步会把 destructive action 错接入发送/消息事实 pipeline,结论是不可复用。应增加 domain-specific server→plugin command(建议 `conversation.reset`/`sync.reset`payload 只有目标 conversation 与 `rebuildId`,不接受任意 objectStore/key),以及明确的 plugin→server ACK。若“普适”要求可复用,也只能让 payload 的 target 是固定白名单的 conversation message-state,不应暴露任意 IndexedDB store 名称/键。
### 4. `maybeBootstrap`、batch bootstrap 和 `startSync` 的复用结论
`BootstrapCoordinator.maybeBootstrap` 先要求 Bright 已认证、收到 anchor snapshot、页面 ready`apps/chrome-extension/src/onetalk/service-worker/sync-engine/bootstrap-coordinator.ts:326-344`),然后读取 account-wide `onetalk_conversation_bootstraps` markermarker 已经是 `complete`/`complete_with_skips` 时直接返回,不会重新同步目标(`bootstrap-coordinator.ts:346-374`)。若强行清 marker`runDiscoveryBootstrap` 会发全 account `conversations.discovered` 并逐个 `startSync``bootstrap-coordinator.ts:376-467`),违反单会话闭环并可能重建其它会话。因此不要用清 marker 或 account batch 作为 reset 后启动手段。
现有 `startSync` 是可复用的 full history engine,但它只在页面 cache 已有目标 SDK conversation 时才能调用 `syncCollectedConversation`;页面 cache 未命中会返回 `conversationCacheMissing``apps/chrome-extension/src/onetalk/main-page/current-conversation-history/all-conversations.ts:205-214,324-354`)。其内部 full path 会先写扫描 checkpoint、路由 `onetalk.sync.conversation`、等待观察写入,再 apply page result、drain ACK candidates、发送 `sync.complete``bootstrap-coordinator.ts:148-275,555-614`)。最小复用方式是新增明确的 Bright→plugin `sync.start`(或 reset flow 的第二阶段):插件收到后先调用已有 `onetalk.discover-conversations` 刷新 MAIN 的 page cache,再调用 `engine.startSync({ channelAccountId, conversationId, mode: "full" })`;不要直接调用 `maybeBootstrap`,也不要发 account-wide `onetalk.sync`
当前 `sync.complete` 是 plugin→Bright 的完成声明,不是成功 ACK;插件发送后 checkpoint 仍为 `uploading`,必须等待服务端下一次 `anchor.snapshot` 与 latest message 精确匹配才进入 `completed``durable-sync.md:100-109``ack-completion.ts:241-294,524-556`)。`anchor.snapshot` 目前只在 plugin hello 后由 server `listAnchors` 发送(`apps/server/src/websocket/plugin/index.ts:464-488`),不能假设 reset 后会自动再发一次;显式单会话 sync command 需要沿用该 checkpoint/ACK 机制,并在适当时请求或复用既有 anchor snapshot 证据。
### 5. 推荐的最小闭环与完成通知
推荐的逻辑顺序是:
1. Mind `POST` 新 rebuild routeBright 使用精确 Mind session/origin/cutover 授权,校验 direct conversation 存在,并选择同 scope/binding 的唯一 fresh plugin。未找到 fresh plugin 时不做任何删除,返回稳定的 `503 plugin_offline`/`plugin_heartbeat_stale`(最终错误码需统一)。
2. Bright 发送 `conversation.reset`(携带 `rebuildId`,在 plugin WS 上只投递到该 canonical plugin),等待 plugin ACK。ACK 只能在四个目标 IDB state store 的 readwrite transaction `oncomplete` 后产生,且应回显同一 `rebuildId`/conversation。
3. 收到 ACK 后重新检查 canonical plugin、lease、binding/authorization version、scope、policy epoch,随后在一个 Bright DB transaction 中删除目标 message/anomaly、重置 conversation 派生字段,提交后才能进入下一步。
4. DB commit 后发送 `conversation.sync.start`(或 reset coordinator 的 second-phase command),plugin 复用 `startSync(full)` 的历史获取、durable candidate、逐条 ACK 和 `sync.complete`,但必须先刷新目标 page cache;不能复用 account-wide `maybeBootstrap`/batch。
5. HTTP 建议返回 `202 Accepted` + `{ conversationId, rebuildId, status: "accepted" }`,不持有请求等待历史页数。Mind 以 rebuildId 订阅 completion;现有 `sync.status`/`conversation.updated` 可复用传输,但当前 status payload 没有 rebuild correlation,至少需在 reset-triggered flow 的 status(以及失败终态)增加可匹配 `rebuildId`,或新增 `rebuild.status`。仅凭 `conversation.updated` 不足以证明重建完成;成功条件应为同 rebuildId、同 conversation 的 `sync.status``historyComplete=true``syncResult=succeeded|succeeded_with_anomalies``anchorAdvanced=true`。失败/断线应有明确终态,不能把 HTTP accepted 当完成。
### 6. 竞态与最小 generation/correlation 边界
现有 `ConversationQueue` 只串行调用者对同一 `(account,conversation)` 的 operation`sync-engine/conversation-queue.ts:7-49`);`ObservationPipeline` 另有独立 observation write chain`observation-pipeline.ts:65-75,227-232,309-347`),page command 的历史 fetch 还通过 MAIN `postMessage`,页面观察 envelope 没有 command requestId/generation`page-bridge/model.ts:81-106,406-438`)。所以仅“清 store + 调 `startSync`”不能阻止旧 history/live observation 或 ACK 迟到回写。
必须至少有一个 reset-owned per-conversation generation/correlation boundary
- `rebuildId` 贯穿 HTTP response、reset command/ACK、sync-start command、plugin checkpoint/active memory、post-reset sync completion/status;服务端只接受当前 rebuild 的 completion/observation,旧 requestId/旧 rebuild 的结果 no-op。
- Plugin reset handler 先让目标进入 quiescing,串行等待该会话现有 `ConversationQueue``ObservationPipeline` 写入/页面同步请求收敛,清理 `AckCompletionCoordinator` 的目标 `requestCandidates``explicitlySentCandidates``pendingDiscoveries``pendingIncompletes``CheckpointCoordinator` active maps,然后在同一 IDB transaction 清理。不能调用全连接 `resetConnection()`,否则会影响其它会话。
- 页面历史 command 在每个跨 await 后需要检查当前 rebuild token;旧 command 的观察不能在 reset 后以新 generation 上传。由于当前 page observed 无 correlation,若不扩展 page command/observation envelope,就无法证明“reset 前已产生、reset 后才到达”的 live/history 事件属于哪一代;这是实现前必须选择的 contract 设计点。最小安全选择是为 targeted history observation/progress 加 `rebuildId`,并在 reset quiescing 期间丢弃未带当前 token 的目标观察;或者取消/失效旧页面 command 并让所有旧结果明确 no-op。
- Server 的 `ObservationBatcher` 会在 `handleObservation` 返回后异步 flush`apps/server/src/websocket/plugin/flows/sync-flows.ts:318-395`);reset 必须与该 conversation 的 batch flush/commit 有同一 writer lock 或 generation guard。否则 pre-reset message batch 可能在 DB DELETE 后才入库。
- ACK/complete 还有跨 await 的 canonical connection/policy guardplugin replaced、heartbeat 过期、Mind revoke、cutover pause、socket close 都必须让旧 ACK/commit/publish no-op。现有 registry generation/commit guard 可作为底层机制(`connection-store.ts:142-159,205-243`),但需要扩展到 rebuild coordinator。
分布式顺序不可原子回滚:plugin IDB ACK 后 Bright DB DELETE 失败,会留下“插件已清、server 仍有旧消息”;DB DELETE 成功后 Bright 在发 sync command 前崩溃,也会留下空但 incomplete 的会话。`rebuildId` 必须可重试且 reset 幂等;若要求进程重启后自动恢复,则应把 pending rebuild 状态持久化(独立 operation 表或 conversation 派生状态),不能只放内存 Map。若产品选择“只允许在线且调用失败即终止”,仍需明确这些 ACK 后故障的重试/人工恢复语义。
## 候选 Scope 与依赖
1. **Cross-cutting contract/operation fence(前置)**:定义 reset/sync-start/ACK 以及 completion correlation 的 exact payload、方向、`rebuildId`/generation 和稳定错误码;拥有 `wire.ts``decoder.ts`、相关 contract 测试。必须先于 server/plugin writer;不允许任意 store/key 删除。
2. **Bright HTTP + rebuild coordinator(依赖 1**:新增单会话 POST 路由、Mind auth/CORS/cutover、fresh plugin 选择、ACK 等待、DB transaction、202 response 和 retry/status ownership。拥有 server HTTP/rebuild service/repository/registry adapter;排除 plugin IDB 实现和 Mind UI。实际共享写入是 registry/plugin command send、`onetalk_message`/`onetalk_message_anomaly`/`onetalk_conversation`
3. **Plugin reset/sync command flow(依赖 1,必须等 2 的 wire/sequence contract**Service Worker 收到 Bright command,按会话 quiesce/generation,清四个消息 state stores,保留 profile/buyer/bootstrap/configACK oncomplete,并触发 page-cache refresh + `startSync(full)`。拥有 storage API、targeted memory invalidation、frame-router、sync-engine/page command;排除 server DB deletion。
4. **Mind completion consumer(依赖 1/2**:仅消费同 account/conversation/rebuildId 的 accepted/status/final event;不能用旧 `sync.status``conversation.updated` 猜新 rebuild。若 Mind 代码不在本仓库,则只交付 contract/fixture 及联调说明。
就“可并发”而言,contract/operation fence 必须先完成;其后 Bright coordinator 与 plugin reset/sync 可在 contract 稳定且各自 write-disjoint、shared-write-free、runtime-isolated 时并发实现,但两者都不能在 fence 定义前开始。Mind consumer 需等 server emitted status shape 稳定后开始。目录不同本身不构成并发依据。
## Invariant 与验收探针
结构性 invariant
- 删除范围严格是 `(channelAccountId, conversationId)`;不得删除 conversation/profile/buyer/bootstrap/account-level anomaly/其它会话。
- plugin ACK 只代表 IDB transaction 已完成;Bright DB deletion 发生在 ACK 之后;DB commit 之后才发 sync-startcompletion 只匹配同 conversation+rebuildId/generation。
- 同一 conversation 同时最多一个 active rebuild;同 rebuildId retry 幂等;旧 connection generation、旧 ACK、旧 page command、旧 observation 不得改变新 generation state。
- full rebuild 最终只能产生目标会话事实;其它会话 checkpoint/候选/消息、profile/buyer ledger 和 account bootstrap marker 不变。
Static/unit probes(不启动服务,不使用端口/DB/浏览器):
- `rg -n 'ONE_TALK_(MESSAGE|CHECKPOINT|CANDIDATE|ANOMALY|CONTACT_PROFILE|CONVERSATION_BOOTSTRAP)_STORE_NAME|createIndex|syncCollectedConversation|maybeBootstrap|startSync|isPluginLeaseFresh|isPluginOnline' apps/chrome-extension/src apps/server/src packages/onetalk-contract/src`;预期能审计固定 store/key、无现有 reset command、`maybeBootstrap` account-wide、`startSync`/page cache 依赖和 fresh lease 差异。
- 用 fake IndexedDB unit test 预置目标/非目标 message、candidate、checkpoint、anomaly、profile、bootstrap 记录;assert 目标四 store 删除、非目标/`contact_profiles`/bootstrap 保留,transaction abort 时无 ACK。覆盖 anomaly `conversationId=undefined` 不被删。
- server repository test 预置四表事实;assert single DB transaction 删除目标 message/targeted anomaly、重置规定 conversation 字段、保留 identity/activity/profile/buyer 和其它会话。验证 `lastMessageAtMs` 的 null 决策及 no JOIN/完整 scope 条件。
- protocol/router test 覆盖 reset/sync-start/ACK exact keys、plugin-only direction、错误 scope、duplicate rebuildId、旧 generation no-opassert 不进入 `send.command`/pending-send。
- controllable Promise sequence testfresh lease/授权失败无任何删除;IDB ACK 前 DB 无变化;ACK 后 DB fail 的 partial boundary 可重试;DB commit 前 policy/socket/generation 失效不发 sync-start;发送失败保持 incomplete。
- generation mutation matrix:旧 observation、旧 message ACK、旧 sync.complete、plugin replacement、heartbeat expiry、policy pause、Mind revoke、page identity/Port disconnect 均不改变新 rebuild;同 rebuildId retry 可重复得到确定结果。
- completion testMind 只接受同 `rebuildId``sync.status` terminal;旧 status/只含 `conversation.updated`/HTTP 202 均不视为完成;success、anomaly、failure、offline 都有可区分终态。
Runtime probes(本研究未执行,需隔离的 test DB、fake/real IndexedDB、独立插件/Mind 账号和可控 WebSocket):
- `pnpm --filter @trade-message-center/server test`:在新增 focused tests 后验证 HTTP/WS/auth/repository sequence,后端测试 hard timeout 60s;不得使用生产 DB。
- `pnpm --filter @trade-message-center/chrome-extension test`:验证 fake IndexedDB、page bridge、sync engine 和 SW frame router;不要仅以构建成功代替 ACK/transaction 证据。
- 真实 probe 需先向测试 Bright 发 POST,再捕获 plugin WS reset command/ACK、检查四个 store 与 server 四表快照、捕获 sync-start 和 Mind 同 rebuildId terminal status;一次测试只使用独立 account/conversation,避免污染其它会话。当前未启动 server、数据库、Chrome 或服务端口,故不能声称 runtime 已证明。
## 注意事项 / 未找到内容
- 当前协议版本是 `ONETALK_PROTOCOL_VERSION=6``packages/onetalk-contract/src/wire.ts:41`);spec 中部分旧文字仍称 v5,新增 frame 应以源码 v6 为准并同步测试/规范。
- `isPluginOnline` 不是 fresh heartbeat 证明;当前没有 Bright 主动 ping plugin 的 request/response。需决定使用新鲜 lease 作为“检查心跳”,还是另加一次性 heartbeat probe。
- 当前 server 无 rebuild operation table/outboxplugin reset 后的 DB failure、server crash、WS ACK 丢失和 sync-start 丢失都没有自动恢复来源;若要求“无条件”包含断线/重启,必须把 operation/generation 持久化或明确 deferred retry。
- 当前页面历史 fetch 的 observation envelope 没有 requestId/rebuildId,迟到 pre-reset history/live observation 的来源无法静态区分;不能只靠现有 `ConversationQueue`/`observationWrites` 宣称已解决。
- 当前 `sync.status``OneTalkSyncStatusPayload` 只有 conversation/mode/result/count/anchor 字段(`packages/onetalk-contract/src/conversation-sync.ts:93-107`),`conversation.updated` 也只有投影会话和 moveToTop(`conversation-sync.ts:108-112`);Mind completion correlation 必须新增字段或独立 status frame。
- 没有可引用的外部文档或生产版本参考;本研究仅依据当前 checkout 的源码、migration、spec 和测试命名,未做真实 runtime 验证。
@@ -0,0 +1,26 @@
{
"id": "mind-rebuild-conversation-history",
"name": "mind-rebuild-conversation-history",
"title": "Expose HTTP endpoint for Mind to rebuild conversation history",
"description": "Expose a Bright HTTP endpoint that enables Mind to request reconstruction of a OneTalk conversation's stored history under the existing authorization and data-boundary rules.",
"status": "in_progress",
"dev_type": null,
"scope": null,
"package": null,
"priority": "P2",
"creator": "ybf",
"assignee": "ybf",
"createdAt": "2026-09-12",
"completedAt": null,
"branch": "09-12-mind-rebuild-conversation-history",
"base_branch": "main",
"worktree_path": null,
"commit": null,
"pr_url": null,
"subtasks": [],
"children": [],
"parent": null,
"relatedFiles": [],
"notes": "",
"meta": {}
}
@@ -29,6 +29,7 @@ type WriterOptions = {
scope: OneTalkPluginScope;
send: OneTalkFrameSend;
createRequestId: (kind: string) => string;
historyGeneration?: (conversationId: string) => string | undefined;
};
const isNonEmptyString = (value: string): boolean => value.length > 0;
@@ -144,14 +145,19 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
const sendMessageObserved: OneTalkSyncFrameWriter["sendMessageObserved"] = (input) => {
const requestId = input.requestId ?? options.createRequestId("message");
if (!isNonEmptyString(requestId)) return null;
const historyGeneration = options.historyGeneration?.(input.message.conversationId);
if (!isNonEmptyString(requestId) || !historyGeneration) return null;
const frame: Extract<OneTalkFrame, { type: "message.observed" }> = {
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "message.observed",
requestId,
scope: options.scope,
payload: { observationSource: input.observationSource, message: input.message },
payload: {
historyGeneration,
observationSource: input.observationSource,
message: input.message,
},
};
return options.send(frame) ? requestId : null;
};
@@ -165,10 +171,21 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
) {
return null;
}
const conversationId = input.messages[0]?.conversationId;
const historyGeneration = conversationId
? options.historyGeneration?.(conversationId)
: undefined;
if (
!historyGeneration ||
input.messages.some((message) => message.conversationId !== conversationId)
) {
return null;
}
const frame = createOneTalkMessagesObservedFrame(
{ connectionType: "plugin", requestId, scope: options.scope },
input.observationSource,
input.messages,
historyGeneration,
);
return byteLength(frame) <= ONETALK_MESSAGE_BATCH_MAX_FRAME_BYTES && options.send(frame)
? requestId
@@ -176,8 +193,10 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
};
const sendSyncComplete: OneTalkSyncFrameWriter["sendSyncComplete"] = (input) => {
const historyGeneration = options.historyGeneration?.(input.conversationId);
if (
!isNonEmptyString(input.conversationId) ||
!historyGeneration ||
(input.latestMessageId !== undefined &&
input.latestMessageId !== null &&
!isNonEmptyString(input.latestMessageId)) ||
@@ -195,6 +214,7 @@ export const createOneTalkSyncFrameWriter = (options: WriterOptions): OneTalkSyn
scope: options.scope,
payload: {
conversationId: input.conversationId,
historyGeneration,
mode: input.mode,
historyComplete: input.historyComplete,
result: input.result,
@@ -59,6 +59,8 @@ class OneTalkSyncEngineImpl implements OneTalkSyncEngine {
scope: options.scope,
send: (frame) => options.bright.send(frame),
createRequestId: this.createRequestId,
historyGeneration: (conversationId) =>
this.lifecycle?.getHistoryGeneration(conversationId),
});
let notifyQueueChange = (): void => undefined;
this.queue = new ConversationQueue(() => notifyQueueChange());
@@ -240,6 +242,7 @@ class OneTalkSyncEngineImpl implements OneTalkSyncEngine {
if (frame.type === "anchor.snapshot") {
const epoch = this.lifecycle.acceptAnchorSnapshot(frame.payload.anchors);
if (epoch === undefined) return;
this.ack.acceptAnchorSnapshot();
this.bootstrap.acceptAnchorSnapshot();
for (const anchor of frame.payload.anchors) {
void this.ack
@@ -40,6 +40,12 @@ type AckCompletionCoordinatorOptions = {
isBootstrapRunning: () => boolean;
};
type CandidateRequest = {
candidateKeys: string[];
conversationId: string;
historyGeneration: string;
};
/** 管理候选投递、ACK、同步完成和可恢复发送。 */
export class AckCompletionCoordinator {
private readonly scope: OneTalkPluginScope;
@@ -52,12 +58,12 @@ export class AckCompletionCoordinator {
private readonly queue: ConversationQueue;
private readonly checkpoints: CheckpointCoordinator;
private readonly isBootstrapRunning: () => boolean;
private readonly requestCandidates = new Map<string, string[]>();
private readonly requestCandidates = new Map<string, CandidateRequest>();
private readonly explicitlySentCandidates = new Set<string>();
private readonly discovered = new Set<string>();
private readonly pendingDiscoveries = new Map<
string,
{ key: string; conversationId: string; explicitStart: boolean }
{ key: string; conversationId: string; explicitStart: boolean; epoch: string }
>();
private readonly pendingIncompletes = new Map<
string,
@@ -88,11 +94,36 @@ export class AckCompletionCoordinator {
this.pendingIncompletes.clear();
}
/** 新 snapshot 使旧 discovery request 的 generation 失去写入资格。 */
public acceptAnchorSnapshot(): void {
this.pendingDiscoveries.clear();
}
/** Batch ACK 是 initial direct 会话已提交的唯一证据。 */
public confirmBatchConversations(conversationIds: readonly string[]): void {
for (const conversationId of conversationIds) {
this.discovered.add(scopedKey(this.scope.channelAccountId, conversationId));
public confirmBatchConversations(
conversations: readonly { conversationId: string; historyGeneration: string }[],
epoch: string,
): boolean {
if (!this.lifecycle.epochIsCurrent(epoch)) return false;
if (
conversations.some((conversation) => {
const current = this.lifecycle.getHistoryGeneration(conversation.conversationId);
return current !== undefined && current !== conversation.historyGeneration;
})
) {
return false;
}
for (const conversation of conversations) {
this.lifecycle.acceptHistoryGeneration(
conversation.conversationId,
conversation.historyGeneration,
epoch,
);
this.discovered.add(
scopedKey(this.scope.channelAccountId, conversation.conversationId),
);
}
return true;
}
public async activateAwaitingCandidates(
@@ -141,7 +172,9 @@ export class AckCompletionCoordinator {
return false;
}
const key = scopedKey(this.scope.channelAccountId, conversationId);
if (this.discovered.has(key)) return true;
if (this.discovered.has(key) && this.lifecycle.getHistoryGeneration(conversationId))
return true;
this.discovered.delete(key);
if ([...this.pendingDiscoveries.values()].some((value) => value.key === key)) return false;
if (!this.bright.isOnline()) {
this.lifecycle.emitDiagnostic({
@@ -152,7 +185,12 @@ export class AckCompletionCoordinator {
return false;
}
const requestId = this.createRequestId("conversation-discovery");
this.pendingDiscoveries.set(requestId, { key, conversationId, explicitStart });
this.pendingDiscoveries.set(requestId, {
key,
conversationId,
explicitStart,
epoch: this.lifecycle.currentEpoch(),
});
const sentRequestId = this.writer.sendConversationDiscovery({
conversationId,
lastContactTimeLong: null,
@@ -176,6 +214,15 @@ export class AckCompletionCoordinator {
const pending = this.pendingDiscoveries.get(frame.requestId);
if (!pending || pending.conversationId !== frame.payload.conversationId) return;
this.pendingDiscoveries.delete(frame.requestId);
if (
!this.lifecycle.acceptHistoryGeneration(
frame.payload.conversationId,
frame.payload.historyGeneration,
pending.epoch,
)
) {
return;
}
this.discovered.add(pending.key);
void this.queue.enqueue(this.scope.channelAccountId, pending.conversationId, async () => {
this.sendPendingIncomplete(pending.key);
@@ -217,6 +264,14 @@ export class AckCompletionCoordinator {
return;
}
if (!this.ensureDiscovered(conversationId, explicitStart)) return;
if (!this.lifecycle.getHistoryGeneration(conversationId)) {
this.lifecycle.emitDiagnostic({
event: "sync_guard",
status: "blocked",
code: "history_generation_missing",
});
return;
}
const candidates = (await this.store.listCandidates(channelAccountId, conversationId))
.filter(pendingCandidate)
.sort(candidateOrder);
@@ -245,6 +300,7 @@ export class AckCompletionCoordinator {
explicitStart = false,
): Promise<void> {
if (this.lifecycle.isDisposed()) return;
if (!this.lifecycle.getHistoryGeneration(conversationId)) return;
const checkpoint = await this.checkpoints.getCheckpoint(channelAccountId, conversationId);
if (!checkpoint || !checkpoint.historyComplete || checkpoint.phase !== "uploading") return;
const candidates = await this.store.listCandidates(channelAccountId, conversationId);
@@ -297,11 +353,21 @@ export class AckCompletionCoordinator {
frame: Extract<OneTalkFrame, { type: "message.ack" | "messages.ack" }>,
): Promise<void> {
if (this.lifecycle.isDisposed()) return;
const requestKeys = this.requestCandidates.get(frame.requestId) ?? [];
const request = this.requestCandidates.get(frame.requestId);
this.requestCandidates.delete(frame.requestId);
if (!request) return;
const results = frame.type === "messages.ack" ? frame.payload.results : [frame.payload];
for (const [index, result] of results.entries()) {
await this.handleAcknowledgementResult(requestKeys[index], result);
if (
result.historyGeneration !== request.historyGeneration ||
this.lifecycle.getHistoryGeneration(request.conversationId) !==
request.historyGeneration ||
(isNonEmptyString(result.conversationId) &&
result.conversationId !== request.conversationId)
) {
continue;
}
await this.handleAcknowledgementResult(request.candidateKeys[index], result);
}
}
@@ -391,7 +457,9 @@ export class AckCompletionCoordinator {
}
private candidateIsInFlight(key: string): boolean {
return [...this.requestCandidates.values()].some((keys) => keys.includes(key));
return [...this.requestCandidates.values()].some((request) =>
request.candidateKeys.includes(key),
);
}
private isBatchableCandidate(
@@ -408,6 +476,8 @@ export class AckCompletionCoordinator {
first: OneTalkSyncCandidate & { observationSource: "history" | "incremental" },
): OneTalkSyncCandidate[] {
const selected: OneTalkSyncCandidate[] = [];
const historyGeneration = this.lifecycle.getHistoryGeneration(first.conversationId);
if (!historyGeneration) return selected;
const start = candidates.indexOf(first);
for (const candidate of candidates.slice(start)) {
if (
@@ -422,6 +492,7 @@ export class AckCompletionCoordinator {
{ connectionType: "plugin", requestId: "message-size-check", scope: this.scope },
candidate.observationSource,
[...selected.map((item) => item.message), candidate.message],
historyGeneration,
);
if (
new TextEncoder().encode(JSON.stringify(frame)).byteLength >
@@ -438,19 +509,31 @@ export class AckCompletionCoordinator {
candidates: OneTalkSyncCandidate[],
explicitStart: boolean,
): Promise<boolean> {
const first = candidates[0];
if (!first) return false;
const historyGeneration = this.lifecycle.getHistoryGeneration(first.conversationId);
if (
!historyGeneration ||
candidates.some((candidate) => candidate.conversationId !== first.conversationId)
) {
return false;
}
const requestId = this.createRequestId(candidates.length === 1 ? "message" : "messages");
this.requestCandidates.set(
requestId,
candidates.map((candidate) => candidate.key),
);
this.requestCandidates.set(requestId, {
candidateKeys: candidates.map((candidate) => candidate.key),
conversationId: first.conversationId,
historyGeneration,
});
for (const candidate of candidates) {
await this.store.updateCandidate(candidate, candidate.status, requestId);
}
if (this.lifecycle.isDisposed()) {
if (
this.lifecycle.isDisposed() ||
this.lifecycle.getHistoryGeneration(first.conversationId) !== historyGeneration
) {
this.requestCandidates.delete(requestId);
return false;
}
const first = candidates[0];
let sent: string | null;
if (first.observationSource === "history" || first.observationSource === "incremental") {
sent = this.writer.sendMessagesObserved({
@@ -522,10 +605,16 @@ export class AckCompletionCoordinator {
}
public async confirmCompletionFromAnchor(
anchor: { conversationId: string; latestMessageId: string | null },
anchor: {
conversationId: string;
latestMessageId: string | null;
historyGeneration: string;
},
epoch: string,
): Promise<void> {
if (!this.lifecycle.epochIsCurrent(epoch)) return;
if (this.lifecycle.getHistoryGeneration(anchor.conversationId) !== anchor.historyGeneration)
return;
await this.queue.enqueue(this.scope.channelAccountId, anchor.conversationId, async () => {
if (!this.lifecycle.epochIsCurrent(epoch)) return;
const checkpoint = await this.checkpoints.getCheckpoint(
@@ -88,7 +88,12 @@ export class BootstrapCoordinator {
private readonly bootstrapStore: OneTalkConversationBootstrapStore;
private readonly batchAcknowledgements = new Map<
string,
{ batchId: string; settle: (accepted: boolean) => void }
{
batchId: string;
conversationIds: readonly string[];
epoch: string;
settle: (accepted: boolean) => void;
}
>();
private bootstrapPromise: Promise<void> | null = null;
@@ -114,6 +119,8 @@ export class BootstrapCoordinator {
}
public acceptAnchorSnapshot(): void {
for (const pending of this.batchAcknowledgements.values()) pending.settle(false);
this.batchAcknowledgements.clear();
this.checkpoints.resetActive();
this.resetProgress();
}
@@ -137,7 +144,17 @@ export class BootstrapCoordinator {
const pending = this.batchAcknowledgements.get(frame.requestId);
if (!pending || pending.batchId !== frame.payload.batchId) return;
this.batchAcknowledgements.delete(frame.requestId);
pending.settle(true);
const receivedIds = frame.payload.conversations.map(
(conversation) => conversation.conversationId,
);
const expectedIds = new Set(pending.conversationIds);
const acknowledged =
this.lifecycle.epochIsCurrent(pending.epoch) &&
receivedIds.length === expectedIds.size &&
new Set(receivedIds).size === receivedIds.length &&
receivedIds.every((conversationId) => expectedIds.has(conversationId)) &&
this.ack.confirmBatchConversations(frame.payload.conversations, pending.epoch);
pending.settle(acknowledged);
}
public async handlePageReady(conversationId?: string): Promise<void> {
@@ -164,6 +181,15 @@ export class BootstrapCoordinator {
let result: OneTalkSyncStartResult = failedResult(input.conversationId, "sync_not_started");
try {
await this.queue.enqueue(input.channelAccountId, input.conversationId, async () => {
if (!this.lifecycle.getHistoryGeneration(input.conversationId)) {
this.lifecycle.emitDiagnostic({
event: "sync_guard",
status: "blocked",
code: "history_generation_missing",
});
result = failedResult(input.conversationId, "history_generation_missing");
return;
}
const existingCheckpoint = await this.checkpoints.getCheckpoint(
input.channelAccountId,
input.conversationId,
@@ -287,6 +313,9 @@ export class BootstrapCoordinator {
);
return Promise.all(
resumable.map(async (checkpoint) => {
if (!this.lifecycle.getHistoryGeneration(checkpoint.conversationId)) {
return failedResult(checkpoint.conversationId, "history_generation_missing");
}
if (checkpoint.phase === "uploading" && checkpoint.historyComplete) {
await this.queue.enqueue(
checkpoint.channelAccountId,
@@ -390,7 +419,12 @@ export class BootstrapCoordinator {
const batchId = this.createRequestId("discovery-batch");
const requestId = this.createRequestId("discovery-frame");
const acknowledgement = new Promise<boolean>((resolve) => {
this.batchAcknowledgements.set(requestId, { batchId, settle: resolve });
this.batchAcknowledgements.set(requestId, {
batchId,
conversationIds: entries.map((entry) => entry.conversationId),
epoch,
settle: resolve,
});
});
const sent = this.writer.sendConversationsDiscovered({ batchId, entries, requestId });
if (!sent) {
@@ -402,7 +436,6 @@ export class BootstrapCoordinator {
const marker = this.mergeMarker(existing, batchId, entries);
await this.bootstrapStore.putConfirmed(marker);
if (!this.lifecycle.epochIsCurrent(epoch)) return;
this.ack.confirmBatchConversations(marker.discoveryOrder);
for (const conversationId of marker.discoveryOrder) {
const current = await this.bootstrapStore.get(
this.scope.channelAccountId,
@@ -1,5 +1,7 @@
// 管理同步生命周期和业务门禁
import type { OneTalkAnchor } from "@trade-message-center/onetalk-contract";
import type {
OneTalkBrightClient,
OneTalkBrightConnectionStatus,
@@ -46,7 +48,7 @@ export class OneTalkSyncLifecycle {
private connectionEpoch = 0;
private snapshotEpoch = 0;
private snapshotFingerprint: string | undefined;
private readonly anchors = new Map<string, string | null>();
private readonly anchors = new Map<string, OneTalkAnchor>();
private pageEpoch = 0;
private pageConversationId: string | undefined;
private lastError: string | undefined;
@@ -88,11 +90,38 @@ export class OneTalkSyncLifecycle {
}
public getAnchor(conversationId: string): string | null | undefined {
return this.anchors.get(conversationId);
return this.anchors.get(conversationId)?.latestMessageId;
}
public getAnchors(): ReadonlyMap<string, string | null> {
return this.anchors;
return new Map(
[...this.anchors.entries()].map(([conversationId, anchor]) => [
conversationId,
anchor.latestMessageId,
]),
);
}
/** 当前 server snapshot 是插件发送目标会话事实的唯一 history generation 来源。 */
public getHistoryGeneration(conversationId: string): string | undefined {
return this.anchors.get(conversationId)?.historyGeneration;
}
/** Discovery ACK 只能为当前 epoch 中尚无 snapshot 的会话补齐 generation,绝不覆盖它。 */
public acceptHistoryGeneration(
conversationId: string,
historyGeneration: string,
expectedEpoch?: string,
): boolean {
if (expectedEpoch !== undefined && !this.epochIsCurrent(expectedEpoch)) return false;
const current = this.anchors.get(conversationId);
if (current) return current.historyGeneration === historyGeneration;
this.anchors.set(conversationId, {
conversationId,
historyGeneration,
latestMessageId: null,
});
return true;
}
public isPageReady(): boolean {
@@ -166,9 +195,7 @@ export class OneTalkSyncLifecycle {
this.lastError = code;
}
public acceptAnchorSnapshot(
anchors: Array<{ conversationId: string; latestMessageId: string | null }>,
): string | undefined {
public acceptAnchorSnapshot(anchors: OneTalkAnchor[]): string | undefined {
if (!this.brightAuthenticated) {
this.emitDiagnostic({
event: "sync_guard",
@@ -185,7 +212,7 @@ export class OneTalkSyncLifecycle {
this.snapshotEpoch += 1;
this.anchors.clear();
for (const anchor of anchors) {
this.anchors.set(anchor.conversationId, anchor.latestMessageId);
this.anchors.set(anchor.conversationId, { ...anchor });
}
this.anchorSnapshotReceived = true;
this.emitDiagnostic({
@@ -116,7 +116,15 @@ test("authenticates, receives anchors, uploads observations, and heartbeats", as
type: "anchor.snapshot",
requestId: "anchors-1",
scope,
payload: { anchors: [{ conversationId: "conversation-1", latestMessageId: "m-1" }] },
payload: {
anchors: [
{
conversationId: "conversation-1",
historyGeneration: "generation-1",
latestMessageId: "m-1",
},
],
},
});
assert.equal(client.isOnline(), true);
assert.equal(frames.at(-1).type, "anchor.snapshot");
@@ -125,6 +133,7 @@ test("authenticates, receives anchors, uploads observations, and heartbeats", as
scope,
send: client.send,
createRequestId: (kind) => `${kind}-1`,
historyGeneration: () => "generation-1",
});
assert.equal(
sync.sendConversationDiscovery({
@@ -166,6 +175,7 @@ test("authenticates, receives anchors, uploads observations, and heartbeats", as
);
assert.deepEqual(sockets[0].sent.at(-1).payload, {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -195,7 +205,7 @@ test("authenticates, receives anchors, uploads observations, and heartbeats", as
assert.equal(client.getState().status, "closed");
});
test("sends only v5 direct discovery entries through the strict contract", () => {
test("sends only v7 direct discovery entries through the strict contract", () => {
const socket = new FakeSocket("wss://bright.example/ws");
const client = createOneTalkBrightClient({
url: socket.url,
@@ -7,6 +7,7 @@ import { createOneTalkSyncEngine as createSyncEngine } from "../src/onetalk/serv
import { createOneTalkServiceWorkerFrameRouter } from "../src/onetalk/service-worker/routing/frame-router.ts";
import { ONE_TALK_PAGE_BRIDGE_VERSION } from "../src/onetalk/page-bridge/model.ts";
import { pageMessageToObserved } from "../src/onetalk/service-worker/sync-engine/helpers.ts";
import { OneTalkSyncLifecycle } from "../src/onetalk/service-worker/sync-engine/lifecycle.ts";
const scope = {
channelAccountId: "account-1",
@@ -283,7 +284,10 @@ class FakeBright {
this.emit({
type: "conversation.ack",
requestId,
payload: { conversationId: input.conversationId },
payload: {
conversationId: input.conversationId,
historyGeneration: "generation-1",
},
});
return requestId;
}
@@ -297,7 +301,13 @@ class FakeBright {
this.emit({
type: "conversations.ack",
requestId,
payload: { batchId: input.batchId },
payload: {
batchId: input.batchId,
conversations: input.entries.map((entry) => ({
conversationId: entry.conversationId,
historyGeneration: "generation-1",
})),
},
}),
);
}
@@ -365,7 +375,58 @@ class FakeBright {
}
emit(frame) {
for (const listener of this.listeners) listener(frame);
const historyGeneration = "generation-1";
const normalized =
frame.type === "anchor.snapshot"
? {
...frame,
payload: {
...frame.payload,
anchors: frame.payload.anchors.map((anchor) => ({
historyGeneration,
...anchor,
})),
},
}
: frame.type === "conversation.ack"
? {
...frame,
payload: { historyGeneration, ...frame.payload },
}
: frame.type === "conversations.ack"
? {
...frame,
payload: {
...frame.payload,
conversations:
frame.payload.conversations ??
this.batches
.find((batch) => batch.requestId === frame.requestId)
?.entries.map((entry) => ({
conversationId: entry.conversationId,
historyGeneration,
})) ??
[],
},
}
: frame.type === "message.ack"
? {
...frame,
payload: { historyGeneration, ...frame.payload },
}
: frame.type === "messages.ack"
? {
...frame,
payload: {
...frame.payload,
results: frame.payload.results.map((result) => ({
historyGeneration,
...result,
})),
},
}
: frame;
for (const listener of this.listeners) listener(normalized);
}
emitStatus(status) {
@@ -407,6 +468,112 @@ const scanningCheckpoint = (conversationId) => ({
updatedAt: 500,
});
test("does not start page sync without a server-provided history generation", async () => {
const store = new MemoryStore();
const bright = new FakeBright();
let routeCalls = 0;
const engine = createOneTalkSyncEngine({
scope,
bright,
store,
pageRuntime: {
routePageCommand: async () => {
routeCalls += 1;
return {
status: "completed",
historyComplete: true,
conversations: [
{
cid: "conversation-1",
status: "completed",
pages: 1,
uniqueMessages: 0,
mode: "full",
anchorFound: false,
latestMessageId: null,
},
],
};
},
},
now: () => 500,
createRequestId: (kind) => `${kind}-missing-generation`,
});
assert.deepEqual(
await engine.startSync({
channelAccountId: scope.channelAccountId,
conversationId: "conversation-1",
mode: "full",
}),
{
status: "failed",
conversationId: "conversation-1",
reason: "history_generation_missing",
},
);
assert.equal(routeCalls, 0);
assert.equal(await store.getCheckpoint(scope.channelAccountId, "conversation-1"), null);
});
test("ignores an ACK with a stale generation even when its request has no echoed scope", async () => {
const store = new MemoryStore();
const bright = new FakeBright();
const engine = createOneTalkSyncEngine({
scope,
bright,
store,
pageRuntime: { routePageCommand: async () => ({ status: "completed" }) },
now: () => 500,
createRequestId: (kind) => `${kind}-stale-ack`,
});
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: null }]);
await engine.ingestObservedBatch({
channelAccountId: scope.channelAccountId,
conversationId: "conversation-1",
observationSource: "live",
messages: [rawMessage("stale-ack-candidate", 500)],
});
await waitFor(() => bright.messages.length === 1, "candidate was not sent");
const requestId = bright.messages[0].requestId;
bright.emit({
type: "message.ack",
requestId,
payload: { historyGeneration: "stale-generation", status: "accepted" },
});
await tick();
await tick();
assert.equal(
(await store.getCandidate(scope.channelAccountId, "conversation-1", "stale-ack-candidate"))
.status,
"pending_ack",
);
});
test("does not replace an anchor generation with a late discovery generation", () => {
const bright = new FakeBright();
const lifecycle = new OneTalkSyncLifecycle(
{ bright, getActiveConversationCount: () => 0 },
{ onBrightStatusChanged: () => undefined },
);
lifecycle.start();
bright.emitStatus("authenticated");
lifecycle.acceptAnchorSnapshot([
{
conversationId: "conversation-1",
historyGeneration: "generation-2",
latestMessageId: null,
},
]);
lifecycle.acceptHistoryGeneration("conversation-1", "stale-generation");
assert.equal(lifecycle.getHistoryGeneration("conversation-1"), "generation-2");
});
test("holds incremental candidates before the anchor and completes after ACK", async () => {
const store = new MemoryStore();
const bright = new FakeBright();
@@ -498,6 +665,7 @@ test("holds incremental candidates before the anchor and completes after ACK", a
assert.equal(bright.completions.length, 1);
assert.deepEqual(bright.completions[0], {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "incremental",
historyComplete: true,
result: "succeeded",
@@ -578,7 +746,7 @@ test("records a missing message ID while continuing the valid message", async ()
pageRuntime: { routePageCommand: async () => ({ status: "completed" }) },
now: () => 700,
});
authenticate(bright);
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: null }]);
assert.equal(engine.getStatus().anchorSnapshotReceived, true);
assert.equal(engine.getStatus().brightStatus, "authenticated");
@@ -825,6 +993,7 @@ test("rejects an oversized history item durably and reports an incomplete sync",
await waitFor(() => bright.completions.length === 1);
assert.deepEqual(bright.completions[0], {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
historyComplete: true,
result: "incomplete",
@@ -956,7 +1125,7 @@ test("re-discovers a conversation before resuming pending facts after reconnect"
},
now: () => 750,
});
authenticate(bright);
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: null }]);
await engine.startSync({
channelAccountId: scope.channelAccountId,
@@ -1706,6 +1875,7 @@ test("holds live observations while a full history scan is still running", async
},
now: () => 758,
});
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: null }]);
const start = engine.startSync({
channelAccountId: scope.channelAccountId,
@@ -1810,6 +1980,7 @@ test("persists page progress while the page command is still scanning", async ()
},
now: () => 762,
});
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: "anchor" }]);
const start = engine.startSync({
channelAccountId: scope.channelAccountId,
@@ -1931,6 +2102,7 @@ test("holds live observations during incremental scan and uploads them after the
},
now: () => 770,
});
authenticate(bright, [{ conversationId: "conversation-1", latestMessageId: "anchor" }]);
const start = engine.startSync({
channelAccountId: scope.channelAccountId,
@@ -2093,6 +2265,7 @@ test("converts an anchor miss to full mode before advancing the new anchor", asy
);
assert.deepEqual(bright.completions[0], {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "incremental",
historyComplete: true,
result: "incomplete",
@@ -2109,6 +2282,7 @@ test("converts an anchor miss to full mode before advancing the new anchor", asy
assert.deepEqual(bright.completions[1], {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
historyComplete: true,
result: "succeeded_with_anomalies",
@@ -0,0 +1 @@
ALTER TABLE "onetalk_conversation" ADD COLUMN "history_generation" text NOT NULL DEFAULT 'initial';
+7
View File
@@ -85,6 +85,13 @@
"when": 1789130586184,
"tag": "0011_mushy_baron_strucker",
"breakpoints": true
},
{
"idx": 12,
"version": "7",
"when": 1789190400000,
"tag": "0012_onetalk_history_generation",
"breakpoints": true
}
]
}
+8 -1
View File
@@ -17,7 +17,10 @@ import {
uuid,
uniqueIndex,
} from "drizzle-orm/pg-core";
import type { OneTalkMessageContent } from "@trade-message-center/onetalk-contract";
import {
ONETALK_INITIAL_HISTORY_GENERATION,
type OneTalkMessageContent,
} from "@trade-message-center/onetalk-contract";
export type JsonValue =
| null
@@ -150,6 +153,10 @@ export const onetalkConversation = pgTable(
syncResult: onetalkSyncResult("sync_result").notNull().default("incomplete"),
/** 增量同步使用的共享停止边界,不用于消息排序或幂等。 */
latestMessageId: text("latest_message_id"),
/** 服务端持久化的单会话历史代际;所有消息和同步回执必须精确匹配。 */
historyGeneration: text("history_generation")
.notNull()
.default(ONETALK_INITIAL_HISTORY_GENERATION),
/** 页面历史是否结束且有效消息已全部确认。 */
historyComplete: boolean("history_complete").notNull().default(false),
/** 该技术会话已提交的消息事实数量。 */
+11 -3
View File
@@ -9,6 +9,7 @@ import type {
OneTalkObservedMessage,
OneTalkObservationSource,
OneTalkMindScope,
OneTalkHistoryGeneration,
OneTalkSyncAnomalyCode,
OneTalkSyncMode,
OneTalkSyncResult,
@@ -60,12 +61,13 @@ export type OneTalkConversationState = {
latestMessageId: string | null;
historyComplete: boolean;
messageCount: number;
historyGeneration: OneTalkHistoryGeneration;
};
export type OneTalkMessageInsertResult =
| { status: "accepted"; message: OneTalkMessage }
| { status: "duplicate"; message: OneTalkMessage }
| { status: "rejected"; reason: "conversation_not_discovered" };
| { status: "rejected"; reason: "conversation_not_discovered" | "history_generation_mismatch" };
export type OneTalkAnomalyInput = {
fingerprint: string;
@@ -80,6 +82,7 @@ export type OneTalkAnomalyInput = {
export type OneTalkSyncCompletionInput = {
conversationId: string;
historyGeneration: OneTalkHistoryGeneration;
mode: OneTalkSyncMode;
historyComplete: boolean;
result: OneTalkSyncResult;
@@ -89,6 +92,7 @@ export type OneTalkSyncCompletionInput = {
};
export type OneTalkSyncStateUpdate = {
historyGeneration: OneTalkHistoryGeneration;
mode: OneTalkSyncMode;
result: OneTalkSyncResult;
historyComplete: boolean;
@@ -103,7 +107,7 @@ export type OneTalkSyncCompletionResult =
conversation: OneTalkConversationState;
anchorAdvanced: boolean;
}
| { status: "rejected"; reason: "conversation_not_discovered" };
| { status: "rejected"; reason: "conversation_not_discovered" | "history_generation_mismatch" };
export type OneTalkRepository = {
findConversation: (
@@ -163,11 +167,13 @@ export type OneTalkRepository = {
) => Promise<boolean>;
insertMessage: (
context: OneTalkSourceContext,
historyGeneration: OneTalkHistoryGeneration,
observationSource: OneTalkObservationSource,
message: OneTalkMessage,
) => Promise<OneTalkMessageInsertResult>;
guardedInsertMessage: (
context: OneTalkSourceContext,
historyGeneration: OneTalkHistoryGeneration,
observationSource: OneTalkObservationSource,
message: OneTalkMessage,
guard: OneTalkCommitGuard,
@@ -199,14 +205,16 @@ export type OneTalkMessageNormalization =
export type OneTalkObservationResult =
| { status: "accepted" | "duplicate"; message: OneTalkMessage }
| { status: "anomaly"; anomalyCode: string }
| { status: "rejected"; reason: "conversation_not_discovered" };
| { status: "rejected"; reason: "conversation_not_discovered" | "history_generation_mismatch" };
export type OneTalkObservationInput = {
historyGeneration: OneTalkHistoryGeneration;
observationSource: OneTalkObservationSource;
message: OneTalkObservedMessage;
};
export type OneTalkNormalizedObservation = {
historyGeneration: OneTalkHistoryGeneration;
observationSource: OneTalkObservationSource;
normalization: OneTalkMessageNormalization;
};
+46 -9
View File
@@ -5,6 +5,7 @@ import type {
OneTalkConversationDiscoveryEntry,
OneTalkConversationType,
OneTalkJsonValue,
OneTalkHistoryGeneration,
OneTalkMessage,
OneTalkObservationSource,
} from "@trade-message-center/onetalk-contract";
@@ -63,6 +64,7 @@ const toConversation = (row: ConversationRow): OneTalkConversationState => {
latestMessageId: row.latestMessageId,
historyComplete: row.historyComplete,
messageCount: row.messageCount,
historyGeneration: row.historyGeneration,
};
};
@@ -202,6 +204,7 @@ const discoverConversation = async (
channelAccountId: context.channelAccountId,
conversationId,
conversationKind,
historyGeneration: "initial",
...(discovery?.lastContactTimeLong === undefined ||
discovery.lastContactTimeLong === null
? {}
@@ -251,6 +254,7 @@ const discoverConversations = async (
channelAccountId: context.channelAccountId,
conversationId: entry.conversationId,
conversationKind: "direct",
historyGeneration: "initial",
lastContactTimeMs: entry.lastContactTimeLong,
messagePreview: entry.messagePreview,
})
@@ -276,6 +280,7 @@ const listAnchors = async (database: Database, context: OneTalkSourceContext) =>
.select({
conversationId: onetalkConversation.conversationId,
latestMessageId: onetalkConversation.latestMessageId,
historyGeneration: onetalkConversation.historyGeneration,
})
.from(onetalkConversation)
.where(eq(onetalkConversation.channelAccountId, context.channelAccountId));
@@ -397,6 +402,7 @@ const guardedHasMessage = async (
const insertMessage = async (
database: Database,
context: OneTalkSourceContext,
historyGeneration: OneTalkHistoryGeneration,
observationSource: OneTalkObservationSource,
message: OneTalkMessage,
guard?: OneTalkCommitGuard,
@@ -408,15 +414,20 @@ const insertMessage = async (
.select({
conversationId: onetalkConversation.conversationId,
conversationKind: onetalkConversation.conversationKind,
historyGeneration: onetalkConversation.historyGeneration,
})
.from(onetalkConversation)
.where(conversationCondition(context, message.conversationId))
.limit(1);
.limit(1)
.for("update");
guard?.assertValid();
if (conversationRows.length === 0 || conversationRows[0]?.conversationKind !== "direct") {
guard?.assertValid();
return { status: "rejected", reason: "conversation_not_discovered" };
}
if (conversationRows[0].historyGeneration !== historyGeneration) {
return { status: "rejected", reason: "history_generation_mismatch" };
}
guard?.assertValid();
const insertedRows = await transaction
@@ -474,11 +485,12 @@ const insertMessage = async (
const guardedInsertMessage = (
database: Database,
context: OneTalkSourceContext,
historyGeneration: OneTalkHistoryGeneration,
observationSource: OneTalkObservationSource,
message: OneTalkMessage,
guard: OneTalkCommitGuard,
): Promise<OneTalkMessageInsertResult> =>
insertMessage(database, context, observationSource, message, guard);
insertMessage(database, context, historyGeneration, observationSource, message, guard);
const guardedObserveMessages = async (
database: Database,
@@ -522,15 +534,21 @@ const guardedObserveMessages = async (
.select({
conversationId: onetalkConversation.conversationId,
conversationKind: onetalkConversation.conversationKind,
historyGeneration: onetalkConversation.historyGeneration,
})
.from(onetalkConversation)
.where(conversationCondition(context, message.conversationId))
.limit(1);
.limit(1)
.for("update");
guard.assertValid();
if (conversation.length === 0 || conversation[0]?.conversationKind !== "direct") {
results.push({ status: "rejected", reason: "conversation_not_discovered" });
continue;
}
if (conversation[0].historyGeneration !== observation.historyGeneration) {
results.push({ status: "rejected", reason: "history_generation_mismatch" });
continue;
}
const inserted = await transaction
.insert(onetalkMessage)
@@ -674,12 +692,22 @@ const updateSyncState = async (
latestMessageId: update.latestMessageId,
anchorUpdatedAt: now,
})
.where(conversationCondition(context, conversationId))
.where(
and(
conversationCondition(context, conversationId),
eq(onetalkConversation.historyGeneration, update.historyGeneration),
),
)
.returning()
: await transaction
.update(onetalkConversation)
.set(baseValues)
.where(conversationCondition(context, conversationId))
.where(
and(
conversationCondition(context, conversationId),
eq(onetalkConversation.historyGeneration, update.historyGeneration),
),
)
.returning();
const row = rows[0];
guard?.assertValid();
@@ -759,11 +787,20 @@ export const createOneTalkRepository = (database: Database): OneTalkRepository =
withDatabaseError(() =>
guardedHasMessage(database, context, conversationId, messageId, guard),
),
insertMessage: (context, observationSource, message) =>
withDatabaseError(() => insertMessage(database, context, observationSource, message)),
guardedInsertMessage: (context, observationSource, message, guard) =>
insertMessage: (context, historyGeneration, observationSource, message) =>
withDatabaseError(() =>
guardedInsertMessage(database, context, observationSource, message, guard),
insertMessage(database, context, historyGeneration, observationSource, message),
),
guardedInsertMessage: (context, historyGeneration, observationSource, message, guard) =>
withDatabaseError(() =>
guardedInsertMessage(
database,
context,
historyGeneration,
observationSource,
message,
guard,
),
),
guardedObserveMessages: (context, observations, guard) =>
withDatabaseError(() => guardedObserveMessages(database, context, observations, guard)),
+18 -1
View File
@@ -143,6 +143,9 @@ const completeSync = async (
const conversation = await repository.findConversation(context, completion.conversationId);
commitGuard?.assertValid();
if (!conversation) return { status: "rejected", reason: "conversation_not_discovered" };
if (conversation.historyGeneration !== completion.historyGeneration) {
return { status: "rejected", reason: "history_generation_mismatch" };
}
const syncAnomaly = syncAnomalyFor(context, completion);
if (syncAnomaly) {
@@ -218,6 +221,7 @@ const completeSync = async (
}
const syncUpdate = {
historyGeneration: completion.historyGeneration,
mode: completion.mode,
result,
historyComplete,
@@ -326,15 +330,27 @@ export const createOneTalkService = (repository: OneTalkRepository): OneTalkServ
else await repository.recordAnomaly(normalized.anomaly);
return { status: "anomaly", anomalyCode: normalized.anomaly.anomalyCode };
}
const conversation = await repository.findConversation(
context,
normalized.message.conversationId,
);
commitGuard?.assertValid();
if (!conversation) return { status: "rejected", reason: "conversation_not_discovered" };
const result = commitGuard
? await repository.guardedInsertMessage(
context,
conversation.historyGeneration,
observationSource,
normalized.message,
commitGuard,
)
: await repository.insertMessage(context, observationSource, normalized.message);
: await repository.insertMessage(
context,
conversation.historyGeneration,
observationSource,
normalized.message,
);
if (result.status === "rejected") return result;
return result;
};
@@ -348,6 +364,7 @@ export const createOneTalkService = (repository: OneTalkRepository): OneTalkServ
return repository.guardedObserveMessages(
context,
observations.map((observation) => ({
historyGeneration: observation.historyGeneration,
observationSource: observation.observationSource,
normalization: normalizeMessage(
context,
@@ -37,7 +37,7 @@ export type OneTalkRegisteredConnection = {
export type OneTalkMindSessionCapability = {
authorize: (
scope: OneTalkMindScope,
operation: "read" | "send" | "heartbeat",
operation: "read" | "send" | "heartbeat" | "rebuild",
) => ReturnType<OneTalkAuthorizationReader["authorize"]>;
clear: () => void;
};
@@ -21,7 +21,7 @@ export class OneTalkMindSessionAuthorization implements OneTalkMindSessionCapabi
public authorize = (
scope: OneTalkMindScope,
operation: "read" | "send" | "heartbeat",
operation: "read" | "send" | "heartbeat" | "rebuild",
): ReturnType<OneTalkAuthorizationReader["authorize"]> => {
if (this.#revoked) {
return Promise.resolve({
@@ -99,6 +99,7 @@ const syncStatusFor = (
anchorAdvanced: boolean,
): OneTalkSyncStatusPayload => ({
conversationId: conversation.conversationId,
historyGeneration: conversation.historyGeneration,
mode,
syncPhase: conversation.syncPhase,
syncResult: conversation.syncResult,
@@ -114,10 +115,12 @@ const safeIdentifier = (value: unknown): string | undefined =>
const acknowledgementFor = (
result: OneTalkObservationResult,
message: { conversationId: unknown; messageId: unknown },
historyGeneration: string,
) => {
if (result.status === "accepted" || result.status === "duplicate") {
return {
status: result.status,
historyGeneration,
conversationId: result.message.conversationId,
messageId: result.message.messageId,
} as const;
@@ -125,6 +128,7 @@ const acknowledgementFor = (
if (result.status === "anomaly") {
return {
status: result.status,
historyGeneration,
conversationId: safeIdentifier(message.conversationId),
messageId: safeIdentifier(message.messageId),
anomalyCode: result.anomalyCode,
@@ -132,6 +136,7 @@ const acknowledgementFor = (
}
return {
status: result.status,
historyGeneration,
conversationId: safeIdentifier(message.conversationId),
messageId: safeIdentifier(message.messageId),
} as const;
@@ -140,10 +145,15 @@ const acknowledgementFor = (
const observationInputsFor = (frame: ObservationFrame): OneTalkObservationInput[] => {
if (frame.type === "message.observed") {
return [
{ observationSource: frame.payload.observationSource, message: frame.payload.message },
{
historyGeneration: frame.payload.historyGeneration,
observationSource: frame.payload.observationSource,
message: frame.payload.message,
},
];
}
return frame.payload.messages.map((message) => ({
historyGeneration: frame.payload.historyGeneration,
observationSource: frame.payload.observationSource,
message,
}));
@@ -186,7 +196,14 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
options.closeForPause();
return;
}
if (!options.sendFrame(createOneTalkConversationAckFrame(frame))) {
if (
!options.sendFrame(
createOneTalkConversationAckFrame(frame, {
conversationId: conversation.conversationId,
historyGeneration: conversation.historyGeneration,
}),
)
) {
options.closeForAcknowledgementFailure();
return;
}
@@ -219,7 +236,7 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
entries: OneTalkConversationsDiscoveredFrame["payload"]["entries"],
): Promise<void> => {
try {
await options.service.discoverConversations(
const conversations = await options.service.discoverConversations(
request.context,
entries,
request.guard,
@@ -229,7 +246,17 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
options.closeForPause();
return;
}
if (!options.sendFrame(createOneTalkConversationsAckFrame(frame))) {
if (
!options.sendFrame(
createOneTalkConversationsAckFrame(frame, {
batchId: frame.payload.batchId,
conversations: conversations.map((conversation) => ({
conversationId: conversation.conversationId,
historyGeneration: conversation.historyGeneration,
})),
}),
)
) {
options.closeForAcknowledgementFailure();
}
} catch (error: unknown) {
@@ -331,7 +358,11 @@ export const createOneTalkSyncFlows = (options: OneTalkSyncFlowsOptions): OneTal
}
offset += queued.items.length;
const acknowledgements = frameResults.map((result, index) =>
acknowledgementFor(result, queued.items[index].message),
acknowledgementFor(
result,
queued.items[index].message,
queued.frame.payload.historyGeneration,
),
);
const acknowledgementFrame =
queued.frame.type === "message.observed"
+10 -3
View File
@@ -57,7 +57,7 @@ const singleFrame = (requestId: string, observed: OneTalkMessage): OneTalkMessag
type: "message.observed",
requestId,
scope: pluginScope,
payload: { observationSource: "live", message: observed },
payload: { historyGeneration: "generation-1", observationSource: "live", message: observed },
});
const arrayFrame = (
@@ -69,7 +69,7 @@ const arrayFrame = (
type: "messages.observed",
requestId,
scope: pluginScope,
payload: { observationSource: "history", messages },
payload: { historyGeneration: "generation-1", observationSource: "history", messages },
});
const queued = (
@@ -78,8 +78,15 @@ const queued = (
frame,
items:
frame.type === "message.observed"
? [{ observationSource: "live", message: frame.payload.message }]
? [
{
historyGeneration: frame.payload.historyGeneration,
observationSource: "live",
message: frame.payload.message,
},
]
: frame.payload.messages.map((observed) => ({
historyGeneration: frame.payload.historyGeneration,
observationSource: "history",
message: observed,
})),
@@ -121,6 +121,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
}),
discoverConversations: async (_context, entries) =>
entries.map((entry) => ({
@@ -132,6 +133,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
})),
listAnchors: async () => [],
listConversations: async () => [],
@@ -153,6 +155,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
},
anchorAdvanced: false,
}),
+58 -3
View File
@@ -71,6 +71,7 @@ const createRepositoryHarness = () => {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
};
conversations.set(key, created);
return created;
@@ -108,9 +109,10 @@ const createRepositoryHarness = () => {
(conversation) =>
conversation.channelAccountId === sourceContext.channelAccountId,
)
.map(({ conversationId, latestMessageId }) => ({
.map(({ conversationId, latestMessageId, historyGeneration }) => ({
conversationId,
latestMessageId,
historyGeneration,
}));
},
listConversations: async (sourceScope) => {
@@ -171,10 +173,17 @@ const createRepositoryHarness = () => {
guard.assertValid();
return result;
},
insertMessage: async (sourceContext, observationSource, observedMessage) => {
insertMessage: async (
sourceContext,
historyGeneration,
observationSource,
observedMessage,
) => {
const conversationKey = `${sourceContext.channelAccountId}:${observedMessage.conversationId}`;
const conversation = conversations.get(conversationKey);
if (!conversation) return { status: "rejected", reason: "conversation_not_discovered" };
if (conversation.historyGeneration !== historyGeneration)
return { status: "rejected", reason: "history_generation_mismatch" };
const key = observationKey(observedMessage.conversationId, observedMessage.messageId);
const existing = messages.get(key);
@@ -189,10 +198,17 @@ const createRepositoryHarness = () => {
assert.ok(observationSource);
return { status: "accepted", message: stored };
},
guardedInsertMessage: async (sourceContext, observationSource, observedMessage, guard) => {
guardedInsertMessage: async (
sourceContext,
historyGeneration,
observationSource,
observedMessage,
guard,
) => {
guard.assertValid();
const result = await repository.insertMessage(
sourceContext,
historyGeneration,
observationSource,
observedMessage,
);
@@ -214,6 +230,7 @@ const createRepositoryHarness = () => {
results.push(
await repository.guardedInsertMessage(
sourceContext,
observation.historyGeneration,
observation.observationSource,
observation.normalization.message,
guard,
@@ -272,6 +289,7 @@ test("persists direct discovery and upgrades an explicit legacy rediscovery", as
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
};
harness.conversations.set(
`${context.channelAccountId}:legacy-conversation`,
@@ -400,6 +418,7 @@ test("advances only valid shared anchors and preserves incomplete outcomes", asy
await emptyService.discoverConversation(context, "empty-conversation", undefined, "direct");
const emptyResult = await emptyService.completeSync(context, {
conversationId: "empty-conversation",
historyGeneration: "initial",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -419,6 +438,7 @@ test("advances only valid shared anchors and preserves incomplete outcomes", asy
const valid = await service.completeSync(context, {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "incremental",
historyComplete: true,
result: "succeeded_with_anomalies",
@@ -435,6 +455,7 @@ test("advances only valid shared anchors and preserves incomplete outcomes", asy
const missing = await service.completeSync(context, {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "incremental",
historyComplete: true,
result: "succeeded",
@@ -449,6 +470,7 @@ test("advances only valid shared anchors and preserves incomplete outcomes", asy
const anchorNotFound = await service.completeSync(context, {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "incremental",
historyComplete: true,
result: "succeeded",
@@ -464,6 +486,7 @@ test("advances only valid shared anchors and preserves incomplete outcomes", asy
const failed = await service.completeSync(context, {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "incremental",
historyComplete: true,
result: "failed",
@@ -494,6 +517,7 @@ test("does not accept a sync completion for an unknown conversation", async () =
const result = await service.completeSync(context, {
conversationId: "unknown-conversation",
historyGeneration: "initial",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -505,3 +529,34 @@ test("does not accept a sync completion for an unknown conversation", async () =
reason: "conversation_not_discovered",
});
});
test("rejects stale history generation for observations and sync completion", async () => {
const harness = createRepositoryHarness();
const service = createOneTalkService(harness.repository);
await service.discoverConversation(context, "conversation-1", undefined, "direct");
const observation = await harness.repository.insertMessage(
context,
"stale-generation",
"history",
message(),
);
assert.deepEqual(observation, {
status: "rejected",
reason: "history_generation_mismatch",
});
const completion = await service.completeSync(context, {
conversationId: "conversation-1",
historyGeneration: "stale-generation",
mode: "full",
historyComplete: true,
result: "succeeded",
latestMessageId: null,
});
assert.deepEqual(completion, {
status: "rejected",
reason: "history_generation_mismatch",
});
assert.equal(harness.syncUpdates.length, 0);
});
@@ -84,6 +84,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
}),
discoverConversations: async (_context, entries, _commitGuard) =>
entries.map((entry) => ({
@@ -95,6 +96,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
})),
listAnchors: async () => [],
listConversations: async () => [],
@@ -116,6 +118,7 @@ const messageService: OneTalkService = {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
},
anchorAdvanced: false,
}),
+18 -2
View File
@@ -126,6 +126,7 @@ const createService = (observeMessage: OneTalkService["observeMessage"]): OneTal
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
}),
discoverConversations: async (_context, entries, _guard) =>
entries.map((entry) => ({
@@ -137,6 +138,7 @@ const createService = (observeMessage: OneTalkService["observeMessage"]): OneTal
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
})),
listAnchors: async () => [],
listConversations: async () => [],
@@ -165,6 +167,7 @@ const createService = (observeMessage: OneTalkService["observeMessage"]): OneTal
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
},
anchorAdvanced: false,
}),
@@ -382,7 +385,11 @@ const observedFrame = (
type: "message.observed",
requestId,
scope: pluginScope,
payload: { observationSource: "live", message: observedMessage },
payload: {
historyGeneration: "initial",
observationSource: "live",
message: observedMessage,
},
};
};
@@ -397,7 +404,7 @@ const observedBatchFrame = (
type: "messages.observed",
requestId,
scope: pluginScope,
payload: { observationSource, messages },
payload: { historyGeneration: "initial", observationSource, messages },
};
};
@@ -604,6 +611,7 @@ test("executes a wire assertion for every declared OneTalk client frame", async
scope: pluginScope,
payload: {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -991,6 +999,7 @@ test("isolates diagnostics sink failures from handshake, business frames, and cl
socket.send(JSON.stringify(observedFrame("diagnostics-business-frame")));
assert.deepEqual((await ack).payload, {
status: "duplicate",
historyGeneration: "initial",
conversationId: "conversation-1",
messageId: "message-1",
});
@@ -1311,6 +1320,7 @@ test("continues after anomaly and rejection with independent ACKs", async () =>
plugin.send(JSON.stringify(observedFrame("observe-anomaly")));
assert.deepEqual((await anomalyAck).payload, {
status: "anomaly",
historyGeneration: "initial",
conversationId: "conversation-1",
messageId: "message-1",
anomalyCode: "invalid_message_observation",
@@ -1320,6 +1330,7 @@ test("continues after anomaly and rejection with independent ACKs", async () =>
plugin.send(JSON.stringify(observedFrame("observe-rejected")));
assert.deepEqual((await rejectedAck).payload, {
status: "rejected",
historyGeneration: "initial",
conversationId: "conversation-1",
messageId: "message-1",
});
@@ -1328,6 +1339,7 @@ test("continues after anomaly and rejection with independent ACKs", async () =>
plugin.send(JSON.stringify(observedFrame("observe-accepted", message("message-3"))));
assert.deepEqual((await acceptedAck).payload, {
status: "accepted",
historyGeneration: "initial",
conversationId: "conversation-1",
messageId: "message-3",
});
@@ -1353,6 +1365,7 @@ test("prevents Mind pages from modifying technical conversations", async () => {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
};
};
const authorization = createMockAuthorizationReader([authorizationRecord]);
@@ -1509,6 +1522,7 @@ test("flushes observations before completing sync and publishes accepted live fa
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
},
anchorAdvanced: false,
};
@@ -1539,6 +1553,7 @@ test("flushes observations before completing sync and publishes accepted live fa
scope: pluginScope,
payload: {
conversationId: "conversation-1",
historyGeneration: "initial",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -1585,6 +1600,7 @@ test("does not publish duplicate live observations", async () => {
plugin.send(JSON.stringify(observedFrame("duplicate-observation")));
assert.deepEqual((await acknowledgement).payload, {
status: "duplicate",
historyGeneration: "initial",
conversationId: "conversation-1",
messageId: "message-1",
});
+2
View File
@@ -84,6 +84,7 @@ const createServiceStub = (): OneTalkService => {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
}),
discoverConversations: async (_context, entries, _commitGuard) =>
entries.map((entry) => ({
@@ -95,6 +96,7 @@ const createServiceStub = (): OneTalkService => {
latestMessageId: null,
historyComplete: false,
messageCount: 0,
historyGeneration: "initial",
})),
listAnchors: async () => [],
listConversations: async () => [],
+10 -2
View File
@@ -31,7 +31,7 @@ export type MindAuthorizationDecodeResult =
| { ok: false; rejection?: MindAuthorizationRejection };
const isPermission = (value: unknown): value is OneTalkPermission =>
value === "read" || value === "send";
value === "read" || value === "send" || value === "rebuild";
/** 严格解码 Mind 成功/稳定拒绝响应;未知字段和缺失字段均失败关闭。 */
export const decodeMindAuthorizationResponse = (value: unknown): MindAuthorizationDecodeResult => {
@@ -199,7 +199,15 @@ const hasPermission = (
permissions: OneTalkPermission[],
): boolean => {
if (operation === "connect" || operation === "heartbeat") return true;
return permissions.includes(operation === "send" ? "send" : "read");
const requiredPermission =
operation === "sync" || operation === "read"
? "read"
: operation === "send"
? "send"
: operation === "rebuild"
? "rebuild"
: undefined;
return requiredPermission !== undefined && permissions.includes(requiredPermission);
};
const decisionForRecord = (
+9 -2
View File
@@ -7,10 +7,17 @@ import type { OneTalkBaseFrame, OneTalkErrorCode, OneTalkFrameContext } from "./
export const ONETALK_CONNECTION_TYPES = ["plugin", "mind_page"] as const;
export type OneTalkConnectionType = (typeof ONETALK_CONNECTION_TYPES)[number];
export const ONETALK_PERMISSIONS = ["read", "send"] as const;
export const ONETALK_PERMISSIONS = ["read", "send", "rebuild"] as const;
export type OneTalkPermission = (typeof ONETALK_PERMISSIONS)[number];
export const ONETALK_AUTH_OPERATIONS = ["connect", "heartbeat", "read", "sync", "send"] as const;
export const ONETALK_AUTH_OPERATIONS = [
"connect",
"heartbeat",
"read",
"sync",
"send",
"rebuild",
] as const;
export type OneTalkAuthorizationOperation = (typeof ONETALK_AUTH_OPERATIONS)[number];
export const ONETALK_PLUGIN_STATUSES = ["online", "offline"] as const;
@@ -10,6 +10,7 @@ import { isPlainRecord } from "./guards.ts";
import { ONETALK_MESSAGE_BATCH_MAX_FRAME_BYTES } from "./messages.ts";
import { ONETALK_PROTOCOL_VERSION, oneTalkFrameWireByteLength } from "./wire.ts";
import type { OneTalkBaseFrame, OneTalkFrameContext } from "./wire.ts";
import { isOneTalkHistoryGeneration, type OneTalkHistoryGeneration } from "./rebuild.ts";
export const ONETALK_SYNC_MODES = ["full", "incremental"] as const;
export type OneTalkSyncMode = (typeof ONETALK_SYNC_MODES)[number];
@@ -33,7 +34,11 @@ export const ONETALK_CONVERSATIONS_DISCOVERED_MAX_LOGICAL_ENTRIES = 100_000;
export const ONETALK_CONVERSATIONS_DISCOVERED_MAX_LOGICAL_FRAME_BYTES =
ONETALK_MESSAGE_BATCH_MAX_FRAME_BYTES * ONETALK_CONVERSATIONS_DISCOVERED_MAX_FRAGMENTS;
export type OneTalkAnchor = { conversationId: string; latestMessageId: string | null };
export type OneTalkAnchor = {
conversationId: string;
latestMessageId: string | null;
historyGeneration: OneTalkHistoryGeneration;
};
export type OneTalkConversationDiscoveredFrame = OneTalkBaseFrame<
"conversation.discovered",
{
@@ -46,7 +51,7 @@ export type OneTalkConversationDiscoveredFrame = OneTalkBaseFrame<
>;
export type OneTalkConversationAckFrame = OneTalkBaseFrame<
"conversation.ack",
{ conversationId: string },
{ conversationId: string; historyGeneration: OneTalkHistoryGeneration },
"plugin"
>;
export type OneTalkConversationDiscoveryEntry = {
@@ -69,7 +74,13 @@ export type OneTalkConversationsDiscoveredFrame = OneTalkBaseFrame<
>;
export type OneTalkConversationsAckFrame = OneTalkBaseFrame<
"conversations.ack",
{ batchId: string },
{
batchId: string;
conversations: Array<{
conversationId: string;
historyGeneration: OneTalkHistoryGeneration;
}>;
},
"plugin"
>;
export type OneTalkAnchorSnapshotFrame = OneTalkBaseFrame<
@@ -81,6 +92,7 @@ export type OneTalkSyncCompleteFrame = OneTalkBaseFrame<
"sync.complete",
{
conversationId: string;
historyGeneration: OneTalkHistoryGeneration;
mode: OneTalkSyncMode;
historyComplete: boolean;
result: OneTalkSyncResult;
@@ -92,6 +104,7 @@ export type OneTalkSyncCompleteFrame = OneTalkBaseFrame<
>;
export type OneTalkSyncStatusPayload = {
conversationId: string;
historyGeneration: OneTalkHistoryGeneration;
mode: OneTalkSyncMode;
syncPhase: "initial" | "incremental";
syncResult: OneTalkSyncResult;
@@ -151,7 +164,9 @@ const hasUniqueDiscoveryEntries = (entries: unknown[]): boolean => {
};
const isAnchor = (value: unknown): value is OneTalkAnchor =>
isPlainRecord(value) &&
hasExactKeys(value, ["conversationId", "historyGeneration", "latestMessageId"]) &&
isNonEmptyString(value.conversationId) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
(value.latestMessageId === null || isNonEmptyString(value.latestMessageId));
/** 严格验证会话发现和同步帧的业务 payload。 */
@@ -180,8 +195,9 @@ export const isValidOneTalkConversationSyncPayload = (
case "conversation.ack":
return (
hasExactDiscoveryFrameKeys(frame) &&
hasExactKeys(value, ["conversationId"]) &&
isNonEmptyString(value.conversationId)
hasExactKeys(value, ["conversationId", "historyGeneration"]) &&
isNonEmptyString(value.conversationId) &&
isOneTalkHistoryGeneration(value.historyGeneration)
);
case "conversations.discovered": {
if (
@@ -213,13 +229,36 @@ export const isValidOneTalkConversationSyncPayload = (
case "conversations.ack":
return (
hasExactDiscoveryFrameKeys(frame) &&
hasExactKeys(value, ["batchId"]) &&
isNonEmptyString(value.batchId)
hasExactKeys(value, ["batchId", "conversations"]) &&
isNonEmptyString(value.batchId) &&
Array.isArray(value.conversations) &&
value.conversations.every(
(conversation) =>
isPlainRecord(conversation) &&
hasExactKeys(conversation, ["conversationId", "historyGeneration"]) &&
isNonEmptyString(conversation.conversationId) &&
isOneTalkHistoryGeneration(conversation.historyGeneration),
)
);
case "anchor.snapshot":
return Array.isArray(value.anchors) && value.anchors.every(isAnchor);
return (
hasExactKeys(value, ["anchors"]) &&
Array.isArray(value.anchors) &&
value.anchors.every(isAnchor)
);
case "sync.complete":
return (
hasExactKeys(value, [
"conversationId",
"historyGeneration",
"mode",
"historyComplete",
"result",
...(value.latestMessageId === undefined ? [] : ["latestMessageId"]),
...(value.latestMessageAtMs === undefined ? [] : ["latestMessageAtMs"]),
...(value.anomalyCode === undefined ? [] : ["anomalyCode"]),
]) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
isNonEmptyString(value.conversationId) &&
typeof value.mode === "string" &&
ONETALK_SYNC_MODES.includes(value.mode as never) &&
@@ -238,6 +277,18 @@ export const isValidOneTalkConversationSyncPayload = (
);
case "sync.status":
return (
hasExactKeys(value, [
"conversationId",
"historyGeneration",
"mode",
"syncPhase",
"syncResult",
"latestMessageId",
"historyComplete",
"messageCount",
"anchorAdvanced",
]) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
isNonEmptyString(value.conversationId) &&
typeof value.mode === "string" &&
ONETALK_SYNC_MODES.includes(value.mode as never) &&
@@ -273,13 +324,14 @@ export const createOneTalkConversationDiscoveredFrame = (
});
export const createOneTalkConversationAckFrame = (
frame: OneTalkConversationDiscoveredFrame,
payload: OneTalkConversationAckFrame["payload"],
): OneTalkConversationAckFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "conversation.ack",
requestId: frame.requestId,
scope: frame.scope,
payload: { conversationId: frame.payload.conversationId },
payload: { ...payload },
});
export const createOneTalkConversationsDiscoveredFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
@@ -298,13 +350,17 @@ export const createOneTalkConversationsDiscoveredFrame = (
});
export const createOneTalkConversationsAckFrame = (
frame: OneTalkConversationsDiscoveredFrame,
payload: OneTalkConversationsAckFrame["payload"],
): OneTalkConversationsAckFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "conversations.ack",
requestId: frame.requestId,
scope: frame.scope,
payload: { batchId: frame.payload.batchId },
payload: {
batchId: payload.batchId,
conversations: payload.conversations.map((conversation) => ({ ...conversation })),
},
});
export const isNextOneTalkConversationsDiscoveredFragment = (
previous: OneTalkConversationsDiscoveredFrame,
+17
View File
@@ -12,6 +12,7 @@ import { isValidOneTalkBuyerFactPayload } from "./buyer-facts.ts";
import { isValidOneTalkConversationSyncPayload } from "./conversation-sync.ts";
import { isValidOneTalkMessagePayload } from "./messages.ts";
import { isValidOneTalkSendingPayload } from "./sending.ts";
import { isValidOneTalkHistoryRebuildPayload } from "./rebuild.ts";
import { isPlainRecord } from "./guards.ts";
import { ONETALK_FRAME_TYPES, ONETALK_PROTOCOL_VERSION } from "./wire.ts";
import type { OneTalkFrame, OneTalkFrameContext, OneTalkFrameType } from "./wire.ts";
@@ -87,6 +88,10 @@ const hasValidFrameDirection = (
"messages.observed",
"message.ack",
"messages.ack",
"storage.delete.command",
"storage.delete.ack",
"history.sync.command",
"history.sync.ack",
].includes(type)
)
return connectionType === "plugin";
@@ -103,6 +108,7 @@ const hasValidFrameDirection = (
"conversation.updated",
"send.request",
"send.result",
"rebuild.status",
].includes(type)
) {
return connectionType === "mind_page";
@@ -161,6 +167,17 @@ const isValidPayload = (
) {
return isValidOneTalkMessagePayload(type, value);
}
if (
[
"storage.delete.command",
"storage.delete.ack",
"history.sync.command",
"history.sync.ack",
"rebuild.status",
].includes(type)
) {
return isValidOneTalkHistoryRebuildPayload(type, value);
}
return isValidOneTalkSendingPayload(type, value);
};
+24
View File
@@ -198,5 +198,29 @@ export {
ONETALK_CONVERSATION_ROUTE,
ONETALK_MESSAGES_ROUTE,
ONETALK_HISTORY_ROUTE,
ONETALK_HISTORY_REBUILD_ROUTE,
} from "./read-api.ts";
export type { OneTalkHttpConversation, OneTalkHttpCustomerProfile } from "./read-api.ts";
export {
ONETALK_HISTORY_REBUILD_STAGES,
ONETALK_HISTORY_REBUILD_REASONS,
ONETALK_INITIAL_HISTORY_GENERATION,
isOneTalkHistoryGeneration,
isValidOneTalkHistoryRebuildPayload,
createOneTalkStorageDeleteCommandFrame,
createOneTalkStorageDeleteAckFrame,
createOneTalkHistorySyncCommandFrame,
createOneTalkHistorySyncAckFrame,
createOneTalkRebuildStatusFrame,
} from "./rebuild.ts";
export type {
OneTalkHistoryGeneration,
OneTalkHistoryRebuildStage,
OneTalkHistoryRebuildReason,
OneTalkStorageDeleteCommandFrame,
OneTalkStorageDeleteAckFrame,
OneTalkHistorySyncCommandFrame,
OneTalkHistorySyncAckFrame,
OneTalkRebuildStatusFrame,
} from "./rebuild.ts";
+56 -10
View File
@@ -10,6 +10,7 @@ import type { OneTalkMindScope, OneTalkPluginScope } from "./connection.ts";
import { isPlainRecord } from "./guards.ts";
import { ONETALK_PROTOCOL_VERSION } from "./wire.ts";
import type { OneTalkBaseFrame, OneTalkFrameContext } from "./wire.ts";
import { isOneTalkHistoryGeneration, type OneTalkHistoryGeneration } from "./rebuild.ts";
export const ONETALK_OBSERVATION_SOURCES = [
"history",
@@ -45,15 +46,24 @@ export type OneTalkObservedMessage = OneTalkMessage;
export type OneTalkBatchObservationSource = "history" | "incremental";
export type OneTalkMessageObservedFrame = OneTalkBaseFrame<
"message.observed",
{ observationSource: OneTalkObservationSource; message: OneTalkObservedMessage },
{
historyGeneration: OneTalkHistoryGeneration;
observationSource: OneTalkObservationSource;
message: OneTalkObservedMessage;
},
"plugin"
>;
export type OneTalkMessagesObservedFrame = OneTalkBaseFrame<
"messages.observed",
{ observationSource: OneTalkBatchObservationSource; messages: OneTalkObservedMessage[] },
{
historyGeneration: OneTalkHistoryGeneration;
observationSource: OneTalkBatchObservationSource;
messages: OneTalkObservedMessage[];
},
"plugin"
>;
export type OneTalkMessageAckPayload = {
historyGeneration: OneTalkHistoryGeneration;
status: OneTalkMessageAckStatus;
conversationId?: string;
messageId?: string;
@@ -118,31 +128,45 @@ export const isValidOneTalkMessagePayload = (type: string, value: unknown): bool
const isAck = (item: unknown): boolean =>
isPlainRecord(item) &&
Object.keys(item).every((key) =>
["status", "conversationId", "messageId", "anomalyCode"].includes(key),
["historyGeneration", "status", "conversationId", "messageId", "anomalyCode"].includes(
key,
),
) &&
Object.keys(item).includes("status") &&
isOneTalkHistoryGeneration(item.historyGeneration) &&
typeof item.status === "string" &&
ONETALK_MESSAGE_ACK_STATUSES.includes(item.status as never) &&
(item.conversationId === undefined || isNonEmptyString(item.conversationId)) &&
(item.messageId === undefined || isNonEmptyString(item.messageId)) &&
(item.anomalyCode === undefined || isNonEmptyString(item.anomalyCode));
const hasOneConversation = (messages: unknown[]): boolean => {
const first = messages[0];
return (
isMessage(first) &&
messages.every(
(message) => isMessage(message) && message.conversationId === first.conversationId,
)
);
};
switch (type) {
case "message.observed":
return (
hasExactKeys(value, ["observationSource", "message"]) &&
hasExactKeys(value, ["historyGeneration", "observationSource", "message"]) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
typeof value.observationSource === "string" &&
ONETALK_OBSERVATION_SOURCES.includes(value.observationSource as never) &&
isMessage(value.message)
);
case "messages.observed":
return (
hasExactKeys(value, ["observationSource", "messages"]) &&
hasExactKeys(value, ["historyGeneration", "observationSource", "messages"]) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
(value.observationSource === "history" ||
value.observationSource === "incremental") &&
Array.isArray(value.messages) &&
value.messages.length > 0 &&
value.messages.length <= ONETALK_MESSAGE_BATCH_MAX_ITEMS &&
value.messages.every(isMessage)
hasOneConversation(value.messages)
);
case "message.ack":
return isAck(value);
@@ -162,10 +186,30 @@ export const isValidOneTalkMessagePayload = (type: string, value: unknown): bool
};
export const isOneTalkMessage = (value: unknown): value is OneTalkMessage =>
isValidOneTalkMessagePayload("message.observed", {
observationSource: "history",
message: value,
});
isPlainRecord(value) &&
hasExactKeys(value, [
"messageId",
"conversationId",
"senderId",
"direction",
"sentAtMs",
"content",
"participantIds",
"readStatus",
"messageStatus",
"unreadCount",
]) &&
isNonEmptyString(value.messageId) &&
isNonEmptyString(value.conversationId) &&
isNonEmptyString(value.senderId) &&
typeof value.direction === "string" &&
ONETALK_DIRECTIONS.includes(value.direction as never) &&
isNonNegativeSafeInteger(value.sentAtMs) &&
isOneTalkMessageContent(value.content) &&
isOneTalkDirectParticipantSet(value.participantIds, value.senderId) &&
isNonNegativeSafeInteger(value.readStatus) &&
isNonNegativeSafeInteger(value.messageStatus) &&
isNonNegativeSafeInteger(value.unreadCount);
export const createOneTalkMessageAckFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
payload: OneTalkMessageAckPayload,
@@ -182,6 +226,7 @@ export const createOneTalkMessagesObservedFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
observationSource: OneTalkBatchObservationSource,
messages: OneTalkObservedMessage[],
historyGeneration: OneTalkHistoryGeneration,
): OneTalkMessagesObservedFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
@@ -189,6 +234,7 @@ export const createOneTalkMessagesObservedFrame = (
requestId: frame.requestId,
scope: frame.scope,
payload: {
historyGeneration,
observationSource,
messages: messages.map((message) => ({
...message,
@@ -8,6 +8,8 @@ export const ONETALK_MESSAGES_ROUTE =
"/api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/messages";
export const ONETALK_HISTORY_ROUTE =
"/api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history";
export const ONETALK_HISTORY_REBUILD_ROUTE =
"/api/bright/onetalk/accounts/:channelAccountId/conversations/:conversationId/history/rebuild";
/** Bright HTTP 会话读取返回的客户资料快照。 */
export type OneTalkHttpCustomerProfile = {
+196
View File
@@ -0,0 +1,196 @@
// 定义单会话历史重建的受限控制帧。
import type { OneTalkMindScope, OneTalkPluginScope } from "./connection.ts";
import { isPlainRecord } from "./guards.ts";
import { ONETALK_PROTOCOL_VERSION } from "./wire.ts";
import type { OneTalkBaseFrame, OneTalkFrameContext } from "./wire.ts";
export const ONETALK_HISTORY_REBUILD_STAGES = [
"rejected",
"storage_cleared",
"server_reset_committed",
"sync_started",
"sync_completed",
"sync_failed",
] as const;
export type OneTalkHistoryRebuildStage = (typeof ONETALK_HISTORY_REBUILD_STAGES)[number];
export const ONETALK_HISTORY_REBUILD_REASONS = [
"conversation_not_found",
"rebuild_in_progress",
"plugin_offline",
"plugin_heartbeat_stale",
"plugin_reset_rejected",
"plugin_reset_timeout",
"database_unavailable",
"sync_start_failed",
] as const;
export type OneTalkHistoryRebuildReason = (typeof ONETALK_HISTORY_REBUILD_REASONS)[number];
export type OneTalkHistoryGeneration = string;
export const ONETALK_INITIAL_HISTORY_GENERATION = "initial" as const;
export type OneTalkStorageDeleteCommandFrame = OneTalkBaseFrame<
"storage.delete.command",
{
rebuildId: string;
target: { kind: "conversation_history"; conversationId: string };
},
"plugin"
>;
export type OneTalkStorageDeleteAckFrame = OneTalkBaseFrame<
"storage.delete.ack",
| { rebuildId: string; conversationId: string; status: "cleared"; reason?: never }
| { rebuildId: string; conversationId: string; status: "rejected"; reason: string },
"plugin"
>;
export type OneTalkHistorySyncCommandFrame = OneTalkBaseFrame<
"history.sync.command",
{
rebuildId: string;
historyGeneration: OneTalkHistoryGeneration;
conversationId: string;
mode: "full";
},
"plugin"
>;
export type OneTalkHistorySyncAckFrame = OneTalkBaseFrame<
"history.sync.ack",
| { rebuildId: string; conversationId: string; status: "started"; reason?: never }
| { rebuildId: string; conversationId: string; status: "rejected"; reason: string },
"plugin"
>;
export type OneTalkRebuildStatusFrame = OneTalkBaseFrame<
"rebuild.status",
{
rebuildId: string;
conversationId: string;
stage: OneTalkHistoryRebuildStage;
reason?: OneTalkHistoryRebuildReason;
},
"mind_page"
>;
const isNonEmptyString = (value: unknown): value is string =>
typeof value === "string" && value.trim().length > 0;
const hasExactKeys = (value: Record<string, unknown>, keys: readonly string[]): boolean => {
const actual = Object.keys(value).sort();
const expected = [...keys].sort();
return (
actual.length === expected.length && actual.every((key, index) => key === expected[index])
);
};
/** 历史代际是服务端持久化的 opaque token,空值或无界输入都不可上 wire。 */
export const isOneTalkHistoryGeneration = (value: unknown): value is OneTalkHistoryGeneration =>
isNonEmptyString(value) && value.length <= 256;
/** 严格验证仅包含固定 semantic target 的历史重建控制帧。 */
export const isValidOneTalkHistoryRebuildPayload = (type: string, value: unknown): boolean => {
if (!isPlainRecord(value)) return false;
if (type === "storage.delete.command") {
return (
hasExactKeys(value, ["rebuildId", "target"]) &&
isNonEmptyString(value.rebuildId) &&
isPlainRecord(value.target) &&
hasExactKeys(value.target, ["conversationId", "kind"]) &&
value.target.kind === "conversation_history" &&
isNonEmptyString(value.target.conversationId)
);
}
if (type === "history.sync.command") {
return (
hasExactKeys(value, ["conversationId", "historyGeneration", "mode", "rebuildId"]) &&
isNonEmptyString(value.rebuildId) &&
isOneTalkHistoryGeneration(value.historyGeneration) &&
isNonEmptyString(value.conversationId) &&
value.mode === "full"
);
}
if (type === "storage.delete.ack" || type === "history.sync.ack") {
const successfulStatus = type === "storage.delete.ack" ? "cleared" : "started";
return (
isNonEmptyString(value.rebuildId) &&
isNonEmptyString(value.conversationId) &&
((value.status === successfulStatus &&
hasExactKeys(value, ["conversationId", "rebuildId", "status"])) ||
(value.status === "rejected" &&
hasExactKeys(value, ["conversationId", "reason", "rebuildId", "status"]) &&
isNonEmptyString(value.reason)))
);
}
if (type === "rebuild.status") {
return (
isNonEmptyString(value.rebuildId) &&
isNonEmptyString(value.conversationId) &&
typeof value.stage === "string" &&
ONETALK_HISTORY_REBUILD_STAGES.includes(value.stage as never) &&
((value.reason === undefined &&
hasExactKeys(value, ["conversationId", "rebuildId", "stage"])) ||
(typeof value.reason === "string" &&
ONETALK_HISTORY_REBUILD_REASONS.includes(value.reason as never) &&
hasExactKeys(value, ["conversationId", "reason", "rebuildId", "stage"])))
);
}
return false;
};
export const createOneTalkStorageDeleteCommandFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
payload: OneTalkStorageDeleteCommandFrame["payload"],
): OneTalkStorageDeleteCommandFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "storage.delete.command",
requestId: frame.requestId,
scope: frame.scope,
payload: { rebuildId: payload.rebuildId, target: { ...payload.target } },
});
export const createOneTalkHistorySyncCommandFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
payload: OneTalkHistorySyncCommandFrame["payload"],
): OneTalkHistorySyncCommandFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "history.sync.command",
requestId: frame.requestId,
scope: frame.scope,
payload: { ...payload },
});
export const createOneTalkStorageDeleteAckFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
payload: OneTalkStorageDeleteAckFrame["payload"],
): OneTalkStorageDeleteAckFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "storage.delete.ack",
requestId: frame.requestId,
scope: frame.scope,
payload: { ...payload },
});
export const createOneTalkHistorySyncAckFrame = (
frame: OneTalkFrameContext & { connectionType: "plugin"; scope: OneTalkPluginScope },
payload: OneTalkHistorySyncAckFrame["payload"],
): OneTalkHistorySyncAckFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "plugin",
type: "history.sync.ack",
requestId: frame.requestId,
scope: frame.scope,
payload: { ...payload },
});
export const createOneTalkRebuildStatusFrame = (
frame: OneTalkFrameContext & { connectionType: "mind_page"; scope: OneTalkMindScope },
payload: OneTalkRebuildStatusFrame["payload"],
): OneTalkRebuildStatusFrame => ({
protocolVersion: ONETALK_PROTOCOL_VERSION,
connectionType: "mind_page",
type: "rebuild.status",
requestId: frame.requestId,
scope: frame.scope,
payload: { ...payload },
});
+22 -2
View File
@@ -37,8 +37,15 @@ import type {
OneTalkSendRequestFrame,
OneTalkSendResultFrame,
} from "./sending.ts";
import type {
OneTalkHistorySyncAckFrame,
OneTalkHistorySyncCommandFrame,
OneTalkRebuildStatusFrame,
OneTalkStorageDeleteAckFrame,
OneTalkStorageDeleteCommandFrame,
} from "./rebuild.ts";
export const ONETALK_PROTOCOL_VERSION = 6 as const;
export const ONETALK_PROTOCOL_VERSION = 7 as const;
export const ONETALK_FRAME_TYPES = [
"ws.hello",
@@ -68,6 +75,11 @@ export const ONETALK_FRAME_TYPES = [
"send.command",
"send.confirmation",
"send.result",
"storage.delete.command",
"storage.delete.ack",
"history.sync.command",
"history.sync.ack",
"rebuild.status",
] as const;
export type OneTalkFrameType = (typeof ONETALK_FRAME_TYPES)[number];
@@ -102,6 +114,9 @@ export const ONETALK_SERVER_FRAME_TYPES = [
"message.created",
"send.command",
"send.result",
"storage.delete.command",
"history.sync.command",
"rebuild.status",
] as const satisfies readonly OneTalkFrameType[];
export const ONETALK_ERROR_CODES = {
@@ -161,7 +176,12 @@ export type OneTalkFrame =
| OneTalkSendRequestFrame
| OneTalkSendCommandFrame
| OneTalkSendConfirmationFrame
| OneTalkSendResultFrame;
| OneTalkSendResultFrame
| OneTalkStorageDeleteCommandFrame
| OneTalkStorageDeleteAckFrame
| OneTalkHistorySyncCommandFrame
| OneTalkHistorySyncAckFrame
| OneTalkRebuildStatusFrame;
export type OneTalkFrameContext = Pick<
OneTalkFrame,
+242 -16
View File
@@ -15,6 +15,7 @@ import {
ONETALK_MAX_MEDIA_SIZE_BYTES,
ONETALK_MESSAGES_ROUTE,
ONETALK_HISTORY_ROUTE,
ONETALK_HISTORY_REBUILD_ROUTE,
ONETALK_PROTOCOL_VERSION,
cloneOneTalkBuyerFact,
createMockAuthorizationReader,
@@ -31,6 +32,11 @@ import {
createOneTalkMessagesObservedFrame,
createOneTalkPluginStatusFrame,
createOneTalkSyncStatusFrame,
createOneTalkStorageDeleteAckFrame,
createOneTalkStorageDeleteCommandFrame,
createOneTalkHistorySyncAckFrame,
createOneTalkHistorySyncCommandFrame,
createOneTalkRebuildStatusFrame,
decodeMindAuthorizationResponse,
decodeOneTalkFrame,
decodeOneTalkMessageContent,
@@ -275,8 +281,8 @@ test("rejects an unknown protocol version with the upgrade error", () => {
});
});
test("hard-rejects v5 and all other old protocol versions with the upgrade error", () => {
for (const protocolVersion of [0, 1, 2, 3, 4, 5, 99]) {
test("hard-rejects v6 and all other old protocol versions with the upgrade error", () => {
for (const protocolVersion of [0, 1, 2, 3, 4, 5, 6, 99]) {
const result = decodeOneTalkFrame({
...frameBase,
protocolVersion,
@@ -291,6 +297,101 @@ test("hard-rejects v5 and all other old protocol versions with the upgrade error
}
});
test("strictly decodes restricted rebuild control frames without exposing store names", () => {
const pluginFrame = { ...frameBase, connectionType: "plugin" as const, scope: pluginScope };
const mindFrame = { ...frameBase, connectionType: "mind_page" as const, scope: mindScope };
const deleteCommand = createOneTalkStorageDeleteCommandFrame(pluginFrame, {
rebuildId: "rebuild-1",
target: { kind: "conversation_history", conversationId: "conversation-1" },
});
const deleteAck = createOneTalkStorageDeleteAckFrame(pluginFrame, {
rebuildId: "rebuild-1",
conversationId: "conversation-1",
status: "cleared",
});
const syncCommand = createOneTalkHistorySyncCommandFrame(pluginFrame, {
rebuildId: "rebuild-1",
historyGeneration: "generation-2",
conversationId: "conversation-1",
mode: "full",
});
const syncAck = createOneTalkHistorySyncAckFrame(pluginFrame, {
rebuildId: "rebuild-1",
conversationId: "conversation-1",
status: "started",
});
const status = createOneTalkRebuildStatusFrame(mindFrame, {
rebuildId: "rebuild-1",
conversationId: "conversation-1",
stage: "server_reset_committed",
});
for (const frame of [deleteCommand, deleteAck, syncCommand, syncAck, status]) {
assert.equal(decodeOneTalkFrame(frame).ok, true);
}
assert.equal(ONETALK_HISTORY_REBUILD_ROUTE.endsWith("/history/rebuild"), true);
assert.deepEqual(
decodeOneTalkFrame({
...deleteCommand,
payload: {
rebuildId: "rebuild-1",
target: {
kind: "conversation_history",
conversationId: "conversation-1",
store: "x",
},
},
}),
{ ok: false, code: ONETALK_ERROR_CODES.invalidMessage },
);
});
test("rejects unknown fields in generation-bearing synchronization payloads", () => {
const pluginSyncFrame = { ...frameBase, connectionType: "plugin" as const, scope: pluginScope };
const mindSyncFrame = { ...frameBase, connectionType: "mind_page" as const, scope: mindScope };
const invalidFrames = [
{
...pluginSyncFrame,
type: "anchor.snapshot" as const,
payload: { anchors: [], extra: true },
},
{
...pluginSyncFrame,
type: "sync.complete" as const,
payload: {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
historyComplete: true,
result: "succeeded",
latestMessageId: null,
extra: true,
},
},
{
...mindSyncFrame,
type: "sync.status" as const,
payload: {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
syncPhase: "initial",
syncResult: "succeeded",
latestMessageId: null,
historyComplete: true,
messageCount: 0,
anchorAdvanced: true,
extra: true,
},
},
];
for (const frame of invalidFrames) {
assert.deepEqual(decodeOneTalkFrame(frame), {
ok: false,
code: ONETALK_ERROR_CODES.invalidMessage,
});
}
});
test("rejects malformed scope and required send correlation", () => {
const missingDevice = decodeOneTalkFrame({
...frameBase,
@@ -332,7 +433,10 @@ test("decodes only exact direct discovery with its commit ACK and the profile fu
assert.equal(direct.type, "conversation.discovered");
assert.equal(direct.payload.conversationType, "direct");
const acknowledgement = createOneTalkConversationAckFrame(discovered);
const acknowledgement = createOneTalkConversationAckFrame(discovered, {
conversationId: "conversation-1",
historyGeneration: "generation-1",
});
assert.deepEqual(decodeOneTalkFrame(acknowledgement), { ok: true, frame: acknowledgement });
assert.equal(acknowledgement.requestId, discovered.requestId);
assert.deepEqual(acknowledgement.scope, discovered.scope);
@@ -393,13 +497,22 @@ test("strictly decodes logical conversation discovery batches, fragments, and co
entries: [],
});
assert.deepEqual(decodeOneTalkFrame(emptyDiscovered), { ok: true, frame: emptyDiscovered });
const emptyAcknowledgement = createOneTalkConversationsAckFrame(emptyDiscovered);
const emptyAcknowledgement = createOneTalkConversationsAckFrame(emptyDiscovered, {
batchId: "batch-empty",
conversations: [],
});
assert.deepEqual(decodeOneTalkFrame(emptyAcknowledgement), {
ok: true,
frame: emptyAcknowledgement,
});
const acknowledgement = createOneTalkConversationsAckFrame(discovered);
const acknowledgement = createOneTalkConversationsAckFrame(discovered, {
batchId: "batch-1",
conversations: entries.map((entry) => ({
conversationId: entry.conversationId,
historyGeneration: "generation-1",
})),
});
assert.deepEqual(decodeOneTalkFrame(acknowledgement), { ok: true, frame: acknowledgement });
assert.equal(acknowledgement.requestId, discovered.requestId);
assert.deepEqual(acknowledgement.scope, discovered.scope);
@@ -562,7 +675,11 @@ test("validates observed messages for history, incremental, live, and send confi
const frame = decode({
...frameBase,
type: "message.observed",
payload: { observationSource, message: observedMessage },
payload: {
historyGeneration: "generation-1",
observationSource,
message: observedMessage,
},
});
assert.equal(frame.type, "message.observed");
}
@@ -573,20 +690,38 @@ test("accepts bounded history batches and rejects invalid batch sources, counts,
{ ...frameBase, connectionType: "plugin", scope: pluginScope },
"history",
[observedMessage],
"generation-1",
);
assert.equal(decodeOneTalkFrame(observed).ok, true);
for (const payload of [
{ observationSource: "live", messages: [observedMessage] },
{ observationSource: "send_confirmation", messages: [observedMessage] },
{ observationSource: "history", messages: [] },
{
historyGeneration: "generation-1",
observationSource: "live",
messages: [observedMessage],
},
{
historyGeneration: "generation-1",
observationSource: "send_confirmation",
messages: [observedMessage],
},
{ historyGeneration: "generation-1", observationSource: "history", messages: [] },
{
historyGeneration: "generation-1",
observationSource: "incremental",
messages: Array.from(
{ length: ONETALK_MESSAGE_BATCH_MAX_ITEMS + 1 },
() => observedMessage,
),
},
{
historyGeneration: "generation-1",
observationSource: "history",
messages: [
observedMessage,
{ ...observedMessage, conversationId: "conversation-2", messageId: "message-2" },
],
},
]) {
assert.deepEqual(decodeOneTalkFrame({ ...observed, payload }), {
ok: false,
@@ -596,7 +731,14 @@ test("accepts bounded history batches and rejects invalid batch sources, counts,
const acknowledgement = createOneTalkMessagesAckFrame(
{ ...frameBase, connectionType: "plugin", scope: pluginScope },
[{ status: "accepted", conversationId: "conversation-1", messageId: "message-1" }],
[
{
status: "accepted",
historyGeneration: "generation-1",
conversationId: "conversation-1",
messageId: "message-1",
},
],
);
assert.equal(decodeOneTalkFrame(acknowledgement).ok, true);
assert.deepEqual(
@@ -1038,7 +1180,11 @@ test("requires exact normalized message observations and excludes raw message fi
const observed = decode({
...frameBase,
type: "message.observed",
payload: { observationSource: "history", message: observedMessage },
payload: {
historyGeneration: "generation-1",
observationSource: "history",
message: observedMessage,
},
});
assert.equal(observed.type, "message.observed");
assert.deepEqual(observed.payload.message, observedMessage);
@@ -1306,7 +1452,11 @@ test("requires exactly two unique direct participants including the sender", ()
decodeOneTalkFrame({
...frameBase,
type: "message.observed",
payload: { observationSource: "history", message },
payload: {
historyGeneration: "generation-1",
observationSource: "history",
message,
},
}),
{ ok: false, code: ONETALK_ERROR_CODES.invalidMessage },
);
@@ -1336,7 +1486,11 @@ test("requires exactly two unique direct participants including the sender", ()
decodeOneTalkFrame({
...frameBase,
type: "message.observed",
payload: { observationSource: "history", message },
payload: {
historyGeneration: "generation-1",
observationSource: "history",
message,
},
}).ok,
true,
);
@@ -1536,16 +1690,30 @@ test("covers anchors, sync completion, acknowledgements, and all send results",
...frameBase,
type: "anchor.snapshot",
payload: {
anchors: [{ conversationId: "conversation-1", latestMessageId: "message-1" }],
anchors: [
{
conversationId: "conversation-1",
historyGeneration: "generation-1",
latestMessageId: "message-1",
},
],
},
});
assert.equal(anchor.type, "anchor.snapshot");
const anchorSnapshot = createOneTalkAnchorSnapshotFrame(frameBase, [
{ conversationId: "conversation-1", latestMessageId: "message-1" },
{
conversationId: "conversation-1",
historyGeneration: "generation-1",
latestMessageId: "message-1",
},
]);
assert.equal(anchorSnapshot.requestId, frameBase.requestId);
assert.deepEqual(anchorSnapshot.payload.anchors, [
{ conversationId: "conversation-1", latestMessageId: "message-1" },
{
conversationId: "conversation-1",
historyGeneration: "generation-1",
latestMessageId: "message-1",
},
]);
const completion = decode({
@@ -1553,6 +1721,7 @@ test("covers anchors, sync completion, acknowledgements, and all send results",
type: "sync.complete",
payload: {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "incremental",
historyComplete: true,
result: "succeeded_with_anomalies",
@@ -1571,6 +1740,7 @@ test("covers anchors, sync completion, acknowledgements, and all send results",
type: "sync.complete",
payload: {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "full",
historyComplete: true,
result: "succeeded",
@@ -1586,6 +1756,7 @@ test("covers anchors, sync completion, acknowledgements, and all send results",
type: "message.ack",
payload: {
status: "anomaly",
historyGeneration: "generation-1",
conversationId: "conversation-1",
messageId: "message-3",
anomalyCode: "missing_sender_id",
@@ -1595,6 +1766,7 @@ test("covers anchors, sync completion, acknowledgements, and all send results",
const acknowledgementFromObservation = createOneTalkMessageAckFrame(frameBase, {
status: "accepted",
historyGeneration: "generation-1",
conversationId: "conversation-1",
messageId: "message-1",
});
@@ -1745,6 +1917,7 @@ test("decodes Mind page plugin and sync status frames", () => {
const syncStatus = createOneTalkSyncStatusFrame(mindFrameBase, {
conversationId: "conversation-1",
historyGeneration: "generation-1",
mode: "incremental",
syncPhase: "incremental",
syncResult: "succeeded",
@@ -1923,6 +2096,59 @@ test("mock authorization separates scope, permission, revoke, and dependency fai
assert.equal(await reader.readAuthorizationVersion(pluginScope, "binding-1"), "version-2");
});
test("keeps rebuild permission distinct from read permission", async () => {
const decoded = decodeMindAuthorizationResponse({
binding: "binding-1",
authorizationVersion: "version-1",
permissions: ["rebuild"],
mindScope,
});
assert.equal(decoded.ok, true);
const reader = createMockAuthorizationReader([
{ ...authorizationRecord, permissions: ["rebuild"] },
]);
assert.deepEqual(
await reader.authorize({
connectionType: "mind_page",
operation: "rebuild",
scope: mindScope,
}),
{
allowed: true,
binding: "binding-1",
authorizationVersion: "version-1",
permissions: ["rebuild"],
mindScope,
},
);
const readDenied = await reader.authorize({
connectionType: "mind_page",
operation: "read",
scope: mindScope,
});
assert.deepEqual(readDenied, {
allowed: false,
code: ONETALK_ERROR_CODES.authorizationRejected,
});
const readOnly = createMockAuthorizationReader([
{ ...authorizationRecord, permissions: ["read"] },
]);
assert.deepEqual(
await readOnly.authorize({
connectionType: "mind_page",
operation: "rebuild",
scope: mindScope,
}),
{
allowed: false,
code: ONETALK_ERROR_CODES.authorizationRejected,
},
);
});
test("does not use deviceId as an authorization decision input", async () => {
const reader = createMockAuthorizationReader([authorizationRecord]);
const decision = await reader.authorize({