pikasTech-unidesk/docs/reference/oa-event-flow.md

Unified OA Event Flow

Unified OA Event Flow 是 UniDesk 的事件流与统计投影基础设施。它独立部署为主 server 用户服务 id=oa-event-flow，为 Pipeline、Code Queue 以及后续 agent runtime 提供统一的事实事件写入、按 tag 订阅、事件表查询和统计中心读接口。Pipeline 专有的流程推进规则仍以 docs/reference/pipeline-oa-event-flow.md 为准；本文件只定义跨业务共享的事件存储、订阅和统计中心契约。

Architecture Boundary

• oa-event-flow 必须是独立微服务，运行在主 server Compose 中，只暴露 Docker 内网端口，不开放公网端口；浏览器只能通过 UniDesk frontend 同源 /api/microservices/oa-event-flow/... 访问。
• 业务服务只发布事实事件或控制意图，不直接修改其他服务状态；统计中心只消费事件流更新数据库投影，并通过读接口和派生事件暴露统计变化。
• 前端页面不得轮询业务 transcript、JSONL、runner 私有文件或旧 SSE 来推断 STEP/Trace 统计；应订阅 trace-stats-updated 等统计变化事件，并按需拉取 oa_trace_stats 投影。
• 事件表、投影表和 projection offset 必须写入主 PostgreSQL；.state/ 只能保存日志、缓存或可重建工件，不能作为事件或统计权威状态。
• 兼容迁移只能在同一重构闭环内存在；交付态不得保留两套互相竞争的 STEP 统计、Trace 刷新或 Pipeline 推进权威来源。

Event Envelope

所有写入 oa_events 的事件都必须使用统一 envelope：

• eventId：全局稳定幂等键；重复发布同一事实必须命中同一 eventId，不能造成重复统计。
• type：事实事件或派生事件类型，例如 trace-step-created、task-updated、node-finished、trace-stats-updated。
• createdAt：事件发生时间；服务端可补齐但不得覆盖调用方提供的真实事实时间。
• sourceKind 与 sourceId：事件来源类型和来源实例，例如 service/code-queue、service/pipeline、projection/trace-stats。
• aggregateType 与 aggregateId：事件所属聚合，例如 task/<taskId>、pipeline/<pipelineId>、epoch/<runId>。
• correlationId 与可选 causationId：跨服务追踪和派生事件回溯链路。
• tags：字符串数组，用于 scope、业务标签、订阅和投影筛选。
• payload：事件事实数据；只放事实和必要摘要，不放未分页的大 transcript、完整日志或 secret。

Tag Contract

Tag 是 OA 事件流的订阅和投影索引。所有 tag 必须是稳定字符串，不依赖 UI 文案。

• 共享 tag：service:<id>、trace、stats、control、diagnostics。
• Code Queue scope：service:code-queue、task:<taskId>、queue:<queueId>、attempt:<index>。
• Pipeline scope：service:pipeline、pipeline:<pipelineId>、epoch:<runId>、node:<nodeId>、attempt:<attemptId>、procedure:<procedureRunId>。
• 订阅接口按 tag 取交集：tags=service:code-queue,task:<taskId> 只返回同时带有这两个 tag 的事件。
• 事件生产者必须自动附加业务 scope tag；前端只传当前页面或当前选择的 tag，不在浏览器里拼接跨业务控制语义。

Event API

oa-event-flow 至少提供以下后端接口：

• GET /health：返回数据库 ready、事件数、投影数、SSE client 数和服务启动时间。
• POST /api/events：批量或单条写入 OA 事件；服务端按 eventId 幂等去重，成功写入后同步推进内置投影。
• GET /api/events?tags=<tag,...>&afterSeq=<n>&limit=<n>：查询事件表，默认按 sequence ASC 返回有界结果。
• GET /api/events/stream?tags=<tag,...>&afterSeq=<n>：SSE 订阅事件表；连接建立后先补发 afterSeq 之后的有界 backlog，再推送 live event。
• GET /api/stats/trace?scopeId=<scope> 或 scopeIds=<scope,...>：读取 Trace/STEP 统计投影，scope 形如 task:<taskId>、task:<taskId>:attempt:<index> 或 pipeline:<pipelineId>。
• GET /api/diagnostics：返回事件表、投影表、近期事件类型、近期统计变化和订阅客户端摘要。

Statistics Center

统计中心是事件流内置投影，不由前端或业务页面临时重算：

• oa_trace_stats 是 Trace/STEP 统计权威表，按 scopeId 保存 statsRevision、stepCount、llmStepCount、readCount、editCount、runCount、errorCount、traceLineCount、outputMaxSeq、attempt 摘要和 updatedAt；其中 stepCount 表示 TraceView 可见执行行数，不等同于工具调用数，工具调用数由 readCount+editCount+runCount 表达。
• Code Queue 任务级 scope task:<taskId> 只表达整个任务的累计统计；每个执行过程摘要必须使用独立 attempt scope task:<taskId>:attempt:<index>。同一个 trace-step-created 事实可由统计投影同时写入任务级和 attempt 级 oa_trace_steps/oa_trace_stats，但前端 attempt 卡片只能读取自己的 attempt scope，不得复用任务级累计统计。
• 投影只消费已经写入 oa_events 的事实事件；不得直接读取 Code Queue transcript、Pipeline .state 或 runner JSONL 作为实时统计来源。
• trace-stats-snapshot 可以作为迁移、服务重启或事件乱序时的权威种子事件；它仍必须进入事件表，由统计投影消费，不能绕过事件流直接写投影。投影消费 trace-step-created 时必须以 oa_trace_steps 的幂等行集合更新统计，不能把 snapshot 和 step 事件对同一个 step 重复累加。
• 投影更新后必须发布派生事件 trace-stats-updated，带 stats payload 和 stats tag；前端订阅该事件来更新左侧 STEP、Trace 摘要指标和 OA 可视化页面。
• 统计投影必须按 eventId 幂等；重复事实不能重复增加 STEP 或 read/edit/run 计数。

Code Queue Integration

• Code Queue 不再维护独立 SSE 作为前端 Trace/STEP 刷新权威；任务输出、状态变更、queue 变更和统计快照必须发布到 oa-event-flow。
• Code Queue 左侧 task card 的 STEP、选中 task 的 Trace Summary 和全局统计只能读取 oa_trace_stats；本地 task JSON 中的历史字段只能作为发布 snapshot 事件的输入，不作为前端权威统计来源。统计尚未投影完成时必须明确显示 statsSource=unavailable 或 STEP --，不得回退到 transcript、本地 stepCount 或前端重算。
• 运行中每个新的 TraceView 可见执行行都必须发布 trace-step-created，并带 task:<taskId>、queue:<queueId>、attempt:<index>、service:code-queue、trace tag，以及 payload 中的 scopeId=task:<taskId>、attemptIndex 和 attemptScopeId=task:<taskId>:attempt:<index>；统计中心据此幂等更新任务级累计统计和 attempt 级独立统计，message/error 行增加 STEP 或 error，system 行默认仅保留在任务原始输出/数据库中，不进入 STEP 计数且不伪装为工具调用。
• Trace Summary 顶部执行摘要读取任务级 task:<taskId>；执行过程摘要 #<index> 读取 task:<taskId>:attempt:<index>。如果 attempt scope 尚未投影完成，必须显示 statsSource=unavailable 或 --，不得回退到任务级累计统计、transcript 重算或旧本地字段。
• 任务入队、开始、终态、移动 queue、标记已读等状态事实必须发布 task-updated 或更具体的事实事件，供事件表和后续审计使用。
• Code Queue judge 每次真正发起 MiniMax LLM 请求前必须发布 judge-llm-request 诊断事件，tag 至少包含 service:code-queue、task:<taskId>、attempt:<index>、judge、diagnostics，payload 必须包含不带 API key 的最终 request payload/messages、prompt/payload 尺寸和 repairAttempt；随后发布 judge-llm-response、judge-json-parse-error（如有）和 judge-result，用于追溯“最终发给 judge LLM 的 prompt”和安全覆盖结果。
• Code Queue 服务启动后可对 PostgreSQL 中已有任务回放每个 TraceView 可见执行行的 trace-step-created，并发布 trace-stats-snapshot 事件完成统计中心种子同步；回放必须使用相同 eventId 保持幂等，不得阻塞队列恢复。历史回放必须按 attempt start 行推导 attemptIndex，确保重建投影时 attempt scope 可独立恢复。

Pipeline Integration

• Pipeline 运行时的 OA 事实、控制事件和 Gantt 取证事件应写入同一个 oa-event-flow，并带 service:pipeline 与 Pipeline scope tag。
• 当 Pipeline 后端部署在远端 provider 且尚未直接写入主 server oa-event-flow 时，主 server oa-event-flow 必须通过 backend-core 内网代理启用单一路径 Pipeline snapshot bridge：只读取 Pipeline 的 /api/snapshot 与 /api/oa-event-flow/diagnostics 结构化接口，并发布稳定的 pipeline-run-snapshot 迁移事件，确保统一事件表中可见 service:pipeline 与 epoch:<runId> 事实。bridge 只做迁移期事件可见性接入，不得再实现“先读 ledger，失败后 snapshot”的双路径 fallback；Pipeline 直接发布到统一事件流后，应删除该 bridge。
• Pipeline 后端可继续向 UniDesk 暴露 Pipeline 专用 snapshot、Gantt DTO、node detail 和 control API，但这些 DTO 的权威来源必须是统一 OA 事件流，而不是旧文件通道或旧 batch 推进状态。
• UniDesk Pipeline 页面可以保留结构化 Pipeline 诊断面板；跨业务事件表、tag 订阅和原始 OA 事件可见性统一进入 用户服务 / OA Event Flow 页面。
• Pipeline 接入完成态必须同时满足本文件的共享事件流契约和 docs/reference/pipeline-oa-event-flow.md 的 Pipeline 控制流去残留门禁。

Frontend Visibility

用户服务 / OA Event Flow 是事件表和统计中心的可视化入口：

• 页面必须显示服务健康、事件总数、最新 sequence、Trace stats scope 数和 live stream 状态。
• 页面必须提供 tag 过滤输入，展示事件表列：sequence、type、source、aggregate、tags、createdAt、payload 摘要和显式 查看原始JSON。
• 页面必须展示 Trace stats 表，至少包含 scope、service、subject、STEP、read/edit/run、error、outputMaxSeq、statsRevision 和更新时间。
• 默认界面不得裸铺完整 JSON；完整事件和统计记录只能通过显式原始 JSON 按钮打开。

Completion Gates

• config.json、docker-compose.yml、CLI server rebuild、server status/logs 和 E2E 必须认识 oa-event-flow。
• Code Queue 前端事件订阅必须指向 oa-event-flow 的 tag stream；不得再依赖 Code Queue 私有 /api/events SSE 作为刷新权威。
• Code Queue Trace Summary 和 task list 响应必须带 traceStats.source=oa-event-flow 或明确的 statsSource=unavailable 状态；unavailable 时不得返回本地重算 STEP 伪装为权威统计。
• Code Queue TraceView 的 attempt 执行摘要必须带 attemptScopeId、attempt 级 traceStats.source=oa-event-flow 或明确的 statsSource=unavailable；前端收到 task-attempt 的 trace-stats-updated 时只能刷新选中任务 Trace Summary，不能把 attempt 统计覆盖到左侧 task 级 STEP。
• Pipeline bridge 启用时，/api/diagnostics 必须暴露 pipelineBridge.mode=snapshot 或各 run 的 bridgeMode=snapshot 状态，GET /api/events?tags=service:pipeline 必须能看到带 epoch:<runId> 的 Pipeline 事件；远端 Pipeline 后端直接发布到统一服务后，应删除 bridge 作为完成态清理项。
• 新增或修改 OA 事件流页面、Code Queue TraceView 或 Pipeline OA 可视化后，必须重建并验证 oa-event-flow、相关业务后端和 frontend。