12 · 运行时事件归一化

Adapter 统一形状，Ingestion 统一含义

Provider adapter 已经输出 canonical event，为什么还需要 2,000 多行的 ProviderRuntimeIngestion.ts？

因为“字段形状一致”不代表“事件具有完整领域上下文”。一条文本 delta 可能没有 thread ID；tool update 可能先于 UI 见到 start； Provider 可能结束 stream 却没发规范的 turn completed；同一个 assistant segment 可能被拆成数百碎片。 Ingestion 是协议时间轴与领域时间轴之间的防腐层。

输入、缓存、输出

缓存不是权威存储；它保存翻译当前 stream 所需的短期上下文，例如 provider session 到 T3 thread 的关联、当前 turn、 assistant segment 是否已经创建和待 flush 文本。权威生命周期仍由 engine/projector 决定。

为什么不能让 Adapter 直接 dispatch

把 dispatch 分散到五个 adapter 会导致：

• 每个 Provider 各自实现 segment 合并和容量限制；
• 生命周期 invariant 有五个版本；
• session binding 与 thread lookup 重复；
• 活动、审批和错误的 fallback 不一致；
• 测试难以用同一事件夹具验证所有 Provider。

Adapter 负责忠实解释原生协议，Ingestion 负责执行 T3 的产品规则。两层看似多一道跳转，实际减少的是组合爆炸。

生命周期守卫

一个理想 Provider 会按 session started → turn started → content → turn completed → idle 发送；真实世界会出现：

• 重复 turn started；
• stream reconnect 后重发一部分 notification；
• completion 先于最后一段文本被消费；
• tool completion 没有对应 start；
• session stop 发生在 turn running；
• session ID 已被 reaper 清理；
• 旧实例的迟到事件撞上新实例 session。

Ingestion 根据当前 projection/runtime cache 判断事件是否可接受、需要补充上下文、可以幂等忽略，还是必须转成错误。 严格不是为了“协议洁癖”，而是防止一个迟到 delta 被追加到下一轮消息。

Assistant segment 缓冲

文本流通常是最高频路径。Ingestion 维护每个 assistant segment 的 buffer，合并碎片后 dispatch。单个缓冲块限制约 24,000 字符； 触及上限就 flush，而非继续等待 Provider 的自然边界。

缓冲策略需要在三个目标间平衡：

T3 Code 选择“有界聚合”，而不是只按时间 debounce。字符上限提供确定内存边界，stream end/segment change/lifecycle event 则触发语义 flush。

消息与 Activity 分离

Provider 流中至少有两类内容：

• Message：用户/assistant 可阅读的长期对话，按 segment 组织文本和附件；
• Activity：工具调用、命令、文件操作、搜索、状态与诊断，具有 start/update/complete/error 生命周期。

若把工具 JSON 全塞进 assistant message，移动端无法做折叠、审批关联和进度显示；若把所有文本都当 activity，又会破坏对话阅读和复制。 Ingestion 根据 canonical event 类型创建/更新正确对象，并保留 provider item ID 作为去重/关联线索。

Approval 与 User Input

两者都代表 Provider 暂停等待人，但语义不同：

• approval 是允许/拒绝一个动作，常与工具 activity 关联；
• user input 是 Provider 主动提问，可能是选择或自由文本。

Ingestion 将 request ID、session、turn、提示和选项写入相应 pending projection。客户端响应后，command reactor 再路由到 adapter。 必须保持 request ID 稳定，否则多个设备同时响应时无法判定哪一个已结算。

重复响应由 decider/invariant 拒绝或幂等处理，而不是把两个答案都送给 Provider。

Plan、Reasoning 与 Provider 特有内容

不同 Provider 对 plan/reasoning 的暴露程度不同。归一化层遵循“可通用则进入通用 segment/activity，不可通用则保留可选 metadata”的原则：

• Plan 可以成为 proposed plan 或 plan activity；
• Reasoning summary 可作为明确标识的 segment，而非伪装成最终回答；
• Provider 私有字段不应污染所有客户端的必填 Schema；
• 敏感内部链路不能因为原生 event 存在就无条件展示。

Stream 结束时的清理顺序

Provider stream 正常/异常结束时，正确顺序通常是：

1. flush 尚未提交的文本；
2. 结算仍打开的 segment/activity，或标记 interrupted/error；
3. dispatch session/turn lifecycle 命令；
4. 清除 session/turn/segment caches；
5. 让等待 drain 的测试/关闭流程完成。

若先清 cache 再 flush，会丢关联；若只关闭 stream 不结算，UI 会永久 running。

DrainableWorker 的确定性

Ingestion 使用 makeDrainableWorker 串行处理输入，并暴露 drain()。它内部跟踪 outstanding 工作而不是靠 queue size 猜测：队列暂时为空不代表当前 handler 已完成。

这让测试可以：push events → drain → 断言 projector，而无需等待 50ms。生产 shutdown 也可在 scope 关闭前等待已接受输入处理完。

观测与诊断

Provider 原生事件可选写入 NDJSON logger (EventNdjsonLogger.ts)，用于比较“上游发了什么”和“领域投影成什么”。 这类日志要注意：

• 可能包含代码、提示和工具参数，不能默认上传；
• 应标记 provider instance/session/sequence，但避免长期凭据；
• decode/unknown event 应可见；
• 用户可控的调试开关与保留策略要清楚。

测试矩阵

源码将普通、activity 和 approval 测试拆开：

• ProviderRuntimeIngestion.test.ts
• ProviderRuntimeIngestion.activity.test.ts
• ProviderRuntimeIngestion.approval.test.ts

一个完整矩阵还应覆盖：文本碎片/flush、重复事件、非法顺序、未知 session、两个实例 ID 相同 session 名、interrupt、stream error、 completion 前最后 delta、pending approval 在 disconnect 后响应，以及 buffer 上限。

可迁移经验

• Protocol canonicalization 与 domain ingestion 是两层：前者统一形状，后者统一时序和产品语义。
• 高频 delta 必须有语义 flush 与硬容量上限。
• 短期缓存只能补全翻译上下文，不能成为生命周期权威。
• 所有 stream 结束路径都必须结算打开对象并 flush 数据。
• 为后台消费器提供 drain，让测试和 shutdown 拥有明确完成条件。