Workflow LLM Stream Chunk SSE 投影方案

[jason-aelf]

Related: #2915

背景

在本地 mainnet host 验证 workflow chat / service chat stream 时，前端没有观察到预期的 LLM chunk。进一步看下来，问题不在 SSE writer，也不在 mapper 是否能把 chunk 转成前端 frame，而在 producer 产出的事件契约和 projection session relay 的输入不一致。

当前 WorkflowRoleGAgent 会把 LLM chunk 作为 transient topology publication 发给 parent：

await PublishAsync(new WorkflowLlmStreamChunkEvent { ... }, TopologyAudience.Parent);

对应 proto：

message WorkflowLlmStreamChunkEvent
{
  string run_id = 1;
  string step_id = 2;
  string session_id = 3;
  string role_actor_id = 4;
  string delta_content = 5;
  string delta_reasoning_content = 6;
}

EventEnvelopeToWorkflowRunEventMapper 已经知道如何把 WorkflowLlmStreamChunkEvent.delta_content 映射成 WorkflowRunEventEnvelope.TextMessageContent，也能把 delta_reasoning_content 映射成 aevatar.llm.reasoning custom event。

但是生产 projection session relay 只消费 committed observation envelope，过滤的是 CommittedStateEventPublished。raw topology WorkflowLlmStreamChunkEvent 不会进入这条链路，所以 mapper 虽然支持该类型，live host 的 workflow SSE 仍然可能看不到 chunk。

设计结论

Workflow LLM stream chunk 应继续走现有 Workflow Execution Projection Session 链路：

WorkflowRoleGAgent
  -> commit typed/coalesced LLM stream observation
  -> committed-state publication
  -> workflow execution session projector
  -> ProjectionSessionEventHub<WorkflowRunEventEnvelope>
  -> WorkflowExecutionProjectionPort live sink
  -> ChatSseResponseWriter
  -> frontend SSE

推荐让 workflow/role actor 产出 typed、run-scoped、coalesced 的 committed LLM stream observation，再由统一 projection session 链路转发。raw provider chunk 不应逐条写入 event store。

这个方向保持了统一 CQRS / AGUI Projection Pipeline，也避免新增 Host 旁路订阅、进程内 sessionId -> sink registry 或第二条 observability path。

ProjectionSessionEventHub 的边界

ProjectionSessionEventHub<TEvent> 应理解为按 rootActorId + sessionId 路由的 live fan-out transport。它不写 source actor event store，不写业务 actor state，也不写 document/read model。

是否进入 event store，取决于 hub 之前的 source fact 是否已经由 actor PersistDomainEventAsync 提交；是否进入 projection state，取决于 projection scope 是否为了 attachment、watermark、failure replay 记录自己的 projection 运行态。这两者都不是 hub 自身的数据持久化。

所以项目里可以存在不进入存储的 live-only 路径，例如 voice realtime frame。但 workflow LLM chunk 如果选择 live-only，就不能同时声称具备 committed projection、replay、dedupe 和 durable observability 语义；如果选择 committed observation，就应先 coalescing，再提交合并后的 progress。

路径对照

路径	当前形态	是否同类缺口	可参考点
Workflow chat / published workflow service chat stream	WorkflowChatRunInteractionService 通过 WorkflowExecutionProjectionPort 绑定 live sink，SSE 写出 WorkflowRunEventEnvelope	是。如果 LLM 增量只来自 raw parent topology publication，projection relay 看不到	修 producer contract，让 workflow/role actor 产出 committed、run-scoped stream observation
Team workflow dry run	同样依赖 command interaction + projection session live sink 形态	是，前提是期望看到 workflow role 的 LLM token 级 live chunk	不应在 dry run endpoint 旁路订阅 role actor raw stream
Team service chat stream / script service AGUI stream	使用 DefaultCommandInteractionService、AGUI session projector 和 ProjectionSessionEventHub<AGUIEvent>	对 script/static GAgent committed/explicit observation 不一定有该缺口；如果底层调用 workflow role raw chunk，则问题相同	复用 command interaction 骨架，修 producer contract
NyxIdChat 直连 SSE	RoleGAgent 提交 RoleChatSessionProgressedEvent(session_id, sequence, typed payload)，NyxIdChatSessionEventProjector 只消费 committed envelope	不是同一个缺口，已经是可参考实现	typed、turn-scoped、committed progress；projection 不读 transient TextMessage*
普通 NyxID relay 平台回复	Channel.NyxIdRelay 进入 ConversationGAgent，通过平台 reply/update carrier 发回外部平台，不是浏览器 SSE	不是同一路径	可参考 actor-owned 去重、流式合并、terminal fact 和最终回复发送的权责划分
Responses / Messages / Chat Completions direct	provider stream 由执行 worker 消费，通过 typed Record* command 回投 LlmSessionGAgent，再提交 LlmStreamChunkObserved 等 recorder event	不是 workflow SSE 缺口；它选择把 run observation 持久化	可参考“执行侧不写 state，session actor 拥有 sequence/terminal”的模式，但不应照搬为 workflow raw chunk 逐条落库

存储语义

需要区分三类 payload：

类别	示例	Event store	Actor state	SSE
Raw provider chunk	一个 token 或 provider frame	否	否	可以，经过明确 live-only mapping 后发送
Committed stream observation	带 run identity 和 sequence 的合并 text/reasoning/tool delta	是	只保留最小 watermark 或合并摘要	是
Terminal result	最终内容、tool calls、tool results、usage、status	是	是，作为稳定 run/session state	是

如果把 raw provider chunk 逐条写入 event store，系统会重复承担多层成本：

• 每个 chunk 都追加一次 source actor event store。
• 每个 chunk 都触发一次 source actor state transition。
• 每个 chunk 都触发一次 committed-state publication。
• 每个 handled observation 都触发 projection scope watermark event。
• replay 和 activation 成本与 chunk 数量成正比。
• 如果 reducer 把内容合并进 state，state root 会变大。

现有 LlmSessionGAgent 的 LlmStreamChunkObserved 模型适合专门的 LLM session observation actor，但不应不加区分地复制到 workflow run state，用来承接每个 provider frame。workflow 更适合参考 RoleGAgent 的 progress 模型：typed progress event 可以提交，但 state 只保存单调 progress watermark 和 terminal session facts。

Coalescing 成本边界

Coalescing 必须发生在 PersistDomainEventAsync 之前。

如果 raw provider chunk 已经逐条 committed，再由 projector 或 SSE writer 合并，这种后置 coalescing 只能减少前端 frame 数，不能减少 event store 写入、source actor state transition、committed-state publication、projection scope watermark 或 replay/activation 成本。

推荐链路：

provider chunk
  -> actor 内 coalescer
  -> PersistDomainEventAsync(coalesced progress)
  -> projector 映射
  -> SSE frame

不推荐链路：

provider chunk
  -> PersistDomainEventAsync(raw chunk)
  -> projector / SSE writer 再合并

Actor 内 coalescer 可以临时保存尚未 flush 的 chunk 信息，但它应是当前 execution flow 内的 transient buffer，不是跨节点事实源。推荐使用 handler 局部 StringBuilder / coalescer；不推荐把 pending raw chunk buffer 放入 actor State，也不应在 Host/Application/projection service 中维护 sessionId -> buffer dictionary。

如果 actor 或进程在 flush 前崩溃，未 flush 的 raw delta 可能丢失。这个前提只有在 raw provider chunk 不被定义为 durable fact 时才成立。flush 间隔越短，写入次数越多但丢失窗口越小；flush 间隔越长，写入次数越少但前端粒度更粗。

建议实现

1. 定义 committed stream progress contract

新增或演进 workflow execution proto，不复用 raw topology 语义。事件应包含：

• run_id
• step_id
• session_id
• role_actor_id
• sequence
• observed_at
• typed oneof payload，例如 text delta、reasoning delta、usage、text ended，以及后续 tool lifecycle payload。

这些字段是核心业务语义，不应放入 generic metadata bag。

2. 从 producer 提交 coalesced progress

在 WorkflowRoleGAgent 或明确选择的 producer actor 内做 actor-owned coalescing，然后提交合并后的 progress。保留现有 terminal WorkflowLlmInvocationCompletedEvent 和 RoleChatSessionCompletedEvent 路径，继续由它们作为最终 output、tool calls、receipts、usage 和 failure 的权威事实。

Flush 边界建议：

• 累积文本达到 50-100 ms 后 flush。
• 累积文本达到配置的字符阈值后 flush。
• tool execution 开始前立即 flush。
• reasoning/text channel 切换时立即 flush。
• terminal completion、failure、cancellation、timeout 时 flush。

3. 保持 producer state 小而稳定

Reducer 只记录 latest accepted stream sequence 和幂等所需的最小 active stream identity，不把每个 chunk 追加进 actor state，不在 query 时通过 replay chunk event 重建最终 workflow result。

4. 把 committed progress 映射成 workflow SSE frame

扩展 EventEnvelopeToWorkflowRunEventMapper，继续基于 committed observed envelope 映射：

• text delta -> WorkflowRunEventEnvelope.TextMessageContent
• reasoning delta -> custom event aevatar.llm.reasoning
• usage -> WorkflowRunEventEnvelope.Usage
• text ended -> WorkflowRunEventEnvelope.TextMessageEnd

Host 不应新增特殊订阅 raw actor stream 的代码。

5. 验证 live SSE

建议补充测试覆盖：

• SSE 在 terminal completion 前至少收到一个 text content frame。
• reasoning chunks 以 aevatar.llm.reasoning 发送。
• terminal event 仍来自现有 terminal path。
• 没有 Host endpoint 直接订阅 raw actor stream。
• ProjectionSessionEventHub 仍然不承担存储职责。

待决策点

实现前需要明确 stream observation authority：

1. 推荐：在 WorkflowRoleGAgent 中提交 progress，并让 role actor 继续作为 producer authority。
2. 备选：在 parent WorkflowRunGAgent 中提交 run-scoped observation，把 workflow run actor 定义为 stream observation authority。

默认推荐第一种，因为它遵循现有 RoleGAgent committed progress 模型，并避免把 role output 重新归属到 parent actor。只有当 workflow 产品语义明确要求所有 live stream facts 都由 workflow run actor 直接拥有时，第二种才适合采用。

非目标

• 不把每个 LLM provider token 都变成 durable fact。
• 不创建可查询的 chunk history read model。
• 不让 ProjectionSessionEventHub 写 actor state、event store、document store 或任何 read model。
• 不在 Host、Application 或 projection services 中引入进程内 session sink registry。

[eanz17]

总体上我赞成这个方向：Workflow LLM stream progress 应继续进入统一的 committed observation / Projection Pipeline，由 role actor 保持 authority，并在 commit 前 coalescing；不应在 Host 增加 raw stream 旁路或进程内 session registry。

不过我建议把当前结论标记为 direction approved, changes required before implementation，至少需要补齐以下几点：

1. 不要把几种可靠性保证混在一起

committed progress 只能说明 source fact 已进入 role actor event store，并允许既有 projection failure 机制处理已记录的失败。ProjectionSessionEventHub 仍然是 live-only transport：late subscriber 或 SSE 断线重连不能自动补回错过的 progress，也不是 end-to-end exactly-once。

建议明确列出 guarantee matrix，分别说明：

• source durability
• normal projection attach 是否 replay
• projection failure replay
• SSE reconnect replay
• presentation dedupe

否则“committed / replay / dedupe / durable observability”容易被误解为同一层保证。

2. sequence 必须穿透 presentation contract

当前 WorkflowRunEventEnvelope 没有 source sequence。若方案声称支持 ordering / dedupe，应把 source position 一直传到 SSE contract，并明确唯一键，例如：

(role_actor_id, role_session_id, sequence)

该 sequence 只能保证单个 role invocation 内有序，不能被描述为整个 workflow run 的全局序列。run_id、step_id、role_actor_id、role_session_id 都应是显式强类型字段，不能通过 session ID、actor ID 的字符串规则推导。

这里还需要区分两个当前都叫 sessionId 的概念：source event 的 role/LLM turn identity，以及 ProjectionSessionEventHub 使用的 live interaction / command identity。二者不能互相冒充。

3. 当前 actor turn 无法直接承诺硬性的 50–100ms flush

WorkflowRoleGAgent 在同一个 handler turn 内执行完整的 await foreach。局部 coalescer 可以在下一条 chunk、字符阈值、reasoning/text channel 切换、tool boundary 或 terminal 时 flush，但 provider 静默期间无法保证墙钟 100ms 内 flush；timer callback 也不能直接修改 actor state，而 self-message 在当前长 turn 结束前无法消费。

因此需要二选一：

• 把时间阈值定义为 opportunistic threshold，在下一次 provider signal 到来时检查；或
• 若 50–100ms 是硬 SLA，则把 stream consumption continuation 化，通过 typed internal signal 让 actor 在独立 turn 中处理/flush，并携带 run_id + step_id + role_session_id 做陈旧事件对账。

字符阈值和时间阈值也建议做成 typed options，并用 event-store write QPS、SSE latency 和 state-root publication 成本来确定默认值。

4. role actor 做 authority 是合理的，但不要产生两套 persisted progress

Local/Orleans runtime 在 parent-child link 后已经会把 child committed observations relay 到 parent，技术上不需要让 WorkflowRunGAgent 再提交一份相同事实。

Workflow 确实需要 generic RoleChatSessionProgressedEvent 目前没有的 run_id / step_id。可以新增 workflow-specific WorkflowLLMStreamProgressedEvent，但建议复用共享 typed payload/coalescing 语义，并确保同一次调用只有一套 persisted progress source；不要同时提交 generic role progress 和 workflow progress 两套状态机。

另外，既然这是 coalesced committed fact，建议使用 Progressed 而不是继续沿用 Chunk 命名。

5. terminal ordering 应成为显式不变量

必须要求所有 success / failure / cancellation / timeout 路径都满足：

final progress flush + commit
  -> usage / text-ended / terminal progress commit
  -> parent-visible workflow completion
  -> step/run terminal

否则最后一段 text/reasoning、usage 或 text-end 仍可能出现在 RunFinished 之后。现有 RoleGAgent 将 terminal presentation tail 与 completion authority 一起提交的原则可以直接参考。

6. 当前回归测试绕过了真正的缺口

WorkflowLlmChunkSseProjectionTests 直接订阅 root stream 并调用 projector，因此绕过了生产 ProjectionScopeObservationRelayBinding 的 committed-only filter。它能证明 mapper/SSE writer 工作，但不能证明 live host production path 工作，属于这个问题上的 false positive。

替换测试应经过真实 wiring：

role committed event
  -> child-to-parent committed relay
  -> production projection scope relay
  -> workflow execution projector
  -> ProjectionSessionEventHub
  -> SSE

并断言：raw topology chunk 不可见、committed progress 可见、sequence 在 mapping 后仍存在、progress 严格先于 terminal。等待建议使用 TaskCompletionSource / Channel，不要引入 polling delay，并执行 tools/ci/test_stability_guards.sh。

其他小点

StateEvent 已经带 committed timestamp。只有当 observed_at 明确表示 provider observation time 时才建议加到 payload，并最好区分 first_observed_at / last_observed_at；否则它是重复语义。

总结：保留本文的主干方案，但在编码前收紧 guarantee matrix、identity/sequence contract、flush semantics、terminal ordering 和 production-relay integration test。这些补齐后，我认为方案与仓库的 actor authority、强类型内核和统一 Projection Pipeline 原则是一致的。

[louis4li]

我按最新 origin/feature/integrate 又核了一遍代码，结论基本支持这个方向，但需要修正一个实现层面的判断：当前最新分支里 mapper 已经有 WorkflowLlmStreamChunkEvent 映射，不是 mapper 缺失。

现状证据：

• WorkflowRoleGAgent 仍然在 stream loop 内通过 PublishAsync(new WorkflowLlmStreamChunkEvent, TopologyAudience.Parent) 发 raw topology publication。
• EventEnvelopeToWorkflowRunEventMapper 已能把 WorkflowLlmStreamChunkEvent.DeltaContent 映射成 TextMessageContent，也能把 DeltaReasoningContent 映射成 aevatar.llm.reasoning custom event。
• ProjectionScopeObservationRelayBinding 的 EventTypeFilter 只绑定 CommittedStateEventPublished。
• ProjectionMaterializationScopeGAgentBase 也会跳过非 committed observation envelope。

所以缺口不是 “chunk 到 SSE frame 的 mapper 不存在”，而是 producer 产出的 raw topology event 不会进入生产 workflow projection session relay。mapper 支持 raw payload 只说明“如果这个 payload 到了 mapper，它能转”，但生产链路的入口仍然是 committed observation。

我建议把实现边界收敛成：

1. 不新增 Host 旁路订阅 raw actor stream，也不引入 sessionId -> sink/buffer 进程内 registry。
2. 不把 provider token 逐条写 event store；coalescing 必须发生在 PersistDomainEventAsync 前。
3. 新增/演进强类型 committed progress contract，至少包含 run_id、step_id、session_id、role_actor_id、sequence、observed_at 和 typed oneof payload。
4. reducer/state 只记录最小 watermark / last sequence / active stream identity；terminal output、tool calls/results、usage、failure 继续走现有 terminal facts。
5. 实现前先定 authority：
   • 若 child committed envelope 能诚实路由到 workflow projection session，倾向让 WorkflowRoleGAgent 作为 progress producer authority，负责 coalescing + commit；WorkflowRunGAgent 继续拥有 terminal run facts。
   • 若 ordering / cancellation / retry / reconciliation 必须由 run actor 统一保证，则让 WorkflowRoleGAgent 发 typed progress continuation 给 WorkflowRunGAgent，由 WorkflowRunGAgent commit。这个选择要明确成 actor protocol，不能靠 topology 或字符串推断。

因此我会把本 discussion 的落点表述为：mapper 已具备部分 frame 转换能力；仍需要把 workflow LLM stream progress 从 transient topology publication 升级为 actor-owned、typed、coalesced committed observation，并继续走现有 Workflow Execution Projection Session 链路。