[!info] related notes

• 所属 MOC: bodysense-moc、agent-runtime-moc
• Trust boundary: bodysense-stream-event-trust-boundary
• Client projection: bodysense-active-turn-state-machine
• Durable recovery: durable-sse-recovery-with-after-seq
• Interrupt / Resume: agent-interrupt-resume-lifecycle
• Runtime layers: agent-runtime-command-event-replay-resume-state-layers
• Query boundary: tanstack-query-with-sse-streaming
• Cancellation: cancellation

BodySense Consultation Streaming Architecture

一句话总览

BodySense 把 Consultation 流式交互拆成：Command/Execution → Durable Public Events → Live SSE / Replay → Runtime Validation → ActiveTurn Projection → Durable Query/Read-Model Reconciliation。Transport 可以失败，但 Run 和最终业务事实不依赖单条连接存活。

当前端到端结构

User Intent
   ↓
Command
(send / answer interaction / cancel run)
   ↓
Go Consultation Runtime
   ↓
Python Agent Runtime / Checkpoint
   ↓ semantic events
Go normalizes public StreamEvent
   ↓
Durable Runtime Event Log
   ├──────────────→ Live POST/SSE
   │                    ↓
   │              fetch + ReadableStream
   │                    ↓
   │               SSE parser
   │                    ↓
   │              parseStreamEvent
   │                    ↓
   │           Trusted StreamEvent
   │                    ↓
   │           run-aware reducer
   │                    ↓
   │             ActiveTurnState
   │                    ↓
   │             Streaming UI
   │
   └──────────────→ after_seq Replay
                        ↓
                   same validation
                        ↓
                   same reducer

terminal / durable domain change
   ↓
setQueryData or invalidate/refetch
   ↓
TanStack Query read-model projection

核心分层：

Transport          ≠ Run lifetime
Event Log          ≠ Checkpoint ≠ Read Model
ActiveTurn         ≠ Durable Truth
Replay             ≠ Resume
Abort transport    ≠ Cancel Run
Command            ≠ Event

1. Trust Boundary：网络 JSON 必须先证明自己

TypeScript StreamEvent union 只在编译期存在。真正从网络来的 JSON 必须经历：

bytes
→ text
→ SSE frame
→ JSON unknown
→ parseStreamEvent(...)
→ trusted StreamEvent
→ reducer

当前 BodySense 已经不是旧的 JSON.parse(...) as StreamEvent 直接信任路径：useSSEProcessor.ts 和 listRunEvents() 都通过共享 contracts 的 parseStreamEvent 建立 runtime trust boundary。

所以要区分：

Type annotation / as T
= compile-time claim

Runtime parser/schema
= runtime evidence

主要代码：

packages/contracts/src/stream-events.ts
packages/contracts/src/stream-event-parser.ts
apps/web/src/features/consultation/hooks/useSSEProcessor.ts
apps/web/src/features/consultation/services/consultationService.ts

详见 bodysense-stream-event-trust-boundary。

2. ActiveTurn：Current Run 的 Transient Projection

当前流式 turn 需要即时呈现：

text delta
tool lifecycle
citation / knowledge gap
red flag
pending interaction
status

这些状态进入一个独立 ActiveTurnState，由纯 reducer 投影：

Current State + StreamEvent
→ Next State + declared effects

当前关键字段包括：

runId
conversationId
status
text
toolCallsById
citationsByKey
pendingInteraction
sequenceRunId
lastSeq
...

这里已采用 run-aware sequence guard：当前实现使用 sequenceRunId + lastSeq，而不是旧的 lastSeqByType compatibility 方案。

详见 bodysense-active-turn-state-machine。

3. Public Sequence：(run_id, seq) 是 Event Identity 的核心

seq 不是 Conversation-global sequence，而是 run-local ordering domain。

例如 HITL Resume：

Conversation C1
├─ Run R1: seq 1...99
│     ↓ interaction.required
│
└─ Run R2: seq 1...
      ↑ resume same logical Agent thread/checkpoint

因此：

(R1, 99)
(R2, 1)

是两个合法且有序于各自 domain 的事件。

如果只记 lastSeq=99 而不记 run_id，R2 的低序号合法事件会被误判为 duplicate，造成 UI 数据丢失。

4. SSE 是 Transport，Runtime Event Log 才提供 Durable Recovery

实时 SSE 可能因为网络/proxy/browser 生命周期中断，但 Run 不应因此自动消失。

服务端公共事件进入 durable event log，客户端断线后用：

(conversation_id, run_id, after_seq)

恢复：

GET .../runs/:runId/events?after_seq=N
→ WHERE seq > N ORDER BY seq ASC

恢复 helper 会：

empty page
→ wait/poll, not done

hasMore
→ continue pagination

replayed seq <= current cursor
→ dedup

explicit durable terminal
→ finish recovery

详见 durable-sse-recovery-with-after-seq。

5. Persist Before Delivery

当前 Go runtime 的公共事件发送顺序是：

enrich event
→ recordPublicEvent
→ WriteEvent over SSE

这样即使：

persist seq=8 succeeded
SSE delivery failed

客户端后续仍可以通过 after_seq=7 找回 8。

这避免把 live transport 当唯一历史来源。

6. Live 与 Replay 共用同一 Event Semantics

Live SSE ─────┐
              ↓
          validator
              ↓
           handlers
              ↓
           reducer
              ↓
         ActiveTurn
              ↑
Durable Replay┘

正确性目标：

projection(live/replay delivery)
== projection(canonical ordered unique event history)

所以系统更适合按：

at-least-once delivery
+ stable identity
+ idempotent projection

设计，而不是假设 replay 永不重复。

7. State Ownership：Layer 不等于 Authority

客户端有多个状态容器，但 business authority 不会因此转移：

URL
→ current resource identity

TanStack Query
→ durable server read-model 的 client cache/projection

ActiveTurn
→ current run transient projection

useState / Zustand
→ local/shared client presentation/runtime state

Server
→ durable business authority

如果 Query Cache 的 BodyState revision=42，而 Server=43，42 必须被视为 stale projection。

因此最重要的问题是：

两份副本冲突时，谁裁决？

见 four-layer-state-architecture 与 tanstack-query-server-state。

8. Mutation / Query Reconciliation

服务器事实被 Mutation 改变后：

有完整 authoritative response
→ 可以 setQueryData

只知道 server changed / response 不完整
→ invalidate + refetch

409 stale revision
→ 本地 projection 已被证明过期
→ invalidate/refetch

ActiveTurn terminal 后也要让 durable Query/read model 收敛，不能把 transient streaming state 永久当成历史来源。

9. Interrupt / Resume：不是普通新消息

ask_user 是 Agent 控制流：

running
→ interaction.required
→ waiting_user / interrupted
→ user answer durably persisted
→ source interaction/run validated
→ new public Run created
→ run.resumed
→ same logical Agent checkpoint/thread continues

当前 BodySense Resume 的关键 identity：

Conversation stays the same
Logical Agent thread/checkpoint continues
Public Run becomes a new Run
New Run has a new run-local seq domain
Pinned source configuration remains explicit

这也是为什么 interaction.answered 和 run.resumed 不能被压成一个无持久化边界的瞬时动作。

详见 agent-interrupt-resume-lifecycle。

10. Replay ≠ Resume

Replay
= 读取已经发生过的 durable events
→ 重建 projection
→ 不创造新现实

Resume
= 给未完成 checkpoint 新合法输入
→ 继续执行
→ 产生新 events

因此 replay 一个 tool.call / interaction.answered / run.resumed event 时，绝不能重新执行 tool、重新提交答案或重新 Resume。

见 agent-runtime-command-event-replay-resume-state-layers。

11. Disconnect ≠ Cancellation

Network / transport disconnect
→ recover

真正的用户 Stop：

CancelRun(run_id)
→ server authorization/state transition
→ active execution cancellation propagation
→ durable run.cancelled

当前 BodySense 已经有显式 cancel path，而不是旧笔记所说的“尚未完成”：

POST /consultation-runs/:id/cancel
→ Runtime.CancelRun
→ RunService / RunRepository
→ running|waiting_user → cancelled
→ durable run.cancelled(reason=cancelled_by_user)

AbortController.abort() 只代表客户端 transport/async work 被中止，不是业务终态 authority。

详见 cancellation。

12. Terminal Fact 必须显式

reader EOF
network error
empty recovery page

都不够证明 Run 已完成。

业务/runtime terminal 应由明确事实表达：

run.completed
run.failed
run.cancelled
stream.done / stream.error（按协议语义）

并保证 terminal state 唯一、不可被迟到普通事件 reopen。

三个 Durable State 层不要混淆

Runtime Event Log
→ what happened during this Run?

Agent Checkpoint
→ where should unfinished computation resume?

Read Model / persisted messages
→ what should normal page load display now?

三者可以都持久化，但职责不同。

当前关键代码走廊

前端：

packages/contracts/src/stream-event-parser.ts
apps/web/src/features/consultation/hooks/useSSEProcessor.ts
apps/web/src/features/consultation/runtime/activeTurnReducer.ts
apps/web/src/features/consultation/runtime/durableRunRecovery.ts
apps/web/src/features/consultation/hooks/useAssistantChatRuntime.ts
apps/web/src/features/consultation/hooks/useThreadProjectionActions.ts

服务端：

apps/api/internal/consultation/runtime.go
apps/api/internal/repository/runtime_event_repository.go
apps/api/internal/handler/runtime_event_handler.go
apps/api/internal/service/run_service.go
apps/api/internal/repository/run_repository.go

一套完整的故障推理

假设：

R1 live receives seq 1..5
network drops
server persists 6..10

正确推理：

network drop
≠ CancelRun

client cursor = (R1, 5)
→ after_seq=5
→ replay 6..10
→ same reducer

如果恢复和 live overlap：

replay 6..8
live 7..10

则：

(R1,7)/(R1,8) duplicate
→ ignore/idempotent
→ final projection = unique ordered 1..10

随后 Agent ask_user：

R1 interrupted
user answers
R1 source segment closes
R2 resumes same logical thread
R2 seq starts at 1

客户端必须切换 sequence domain：

sequenceRunId: R1 → R2
lastSeq: 99 → 1

而不是用 R1 的 cursor 过滤 R2。

面试表达

“我会把 AI streaming 分成 transport、durable event、client projection 和 execution checkpoint。Live 用 SSE 获得低延迟，但 public event 会有 (run_id, seq) 并进入 durable event log；浏览器断线后用 after_seq catch up，Live/Replay 走同一 runtime validation 和 reducer。当前 streaming UI 是 ActiveTurn projection，不是 durable truth，终态后再和 TanStack Query/read model 收敛。网络断线不等于 Cancel；显式 Stop 是 CancelRun command，由服务端形成 durable run.cancelled。HITL Resume 则继续同一 logical checkpoint，但可以创建新的 public Run 和新的 seq namespace。”

最短记忆

unknown network data → runtime validation
StreamEvent → pure/run-aware projection
SSE → immediacy
Event Log → durability/recovery
(run_id, seq) → event identity
Query → server read-model cache
Checkpoint → execution resume
Disconnect → recover
Stop → CancelRun
Replay → rebuild past
Resume → continue unfinished computation