fix: 统一 HWLAB 全局 Kafka 投影消费组

This commit is contained in:
pikastech
2026-07-19 19:06:56 +02:00
parent e36aec5843
commit 66a655f29e
3 changed files with 22 additions and 6 deletions
+16 -4
View File
@@ -16,6 +16,10 @@ description: UniDesk Kafka 调试与事件链校对技能,覆盖 platform-infr
- `agentrun.event.debug.v1`
- `hwlab.event.debug.v1`
- replay/regenerate 不得向产品 topic 追加事件,不得重置产品 consumer group,不得把 debug group 冒充产品消费证据。
- 同一个全局输入 topic 到同一个全局输出 topic 的 direct projector
- 跨 lane 实例必须使用同一个 consumer group
- 各 lane 面向自身客户端的 SSE fanout 必须使用彼此独立的 consumer group
- 禁止用 lane 专属 direct-projector group 重复消费全局输入并向全局输出写入同一事件。
- 优先使用应用生产 reducer/mapper 的 regenerate CLI;平台 `kafka tail` 只返回 hash/bytes,不能证明 payload 语义。
- 调查结论必须区分:
- 源 stdio 未产生;
@@ -56,10 +60,14 @@ description: UniDesk Kafka 调试与事件链校对技能,覆盖 platform-infr
2. 查 OTel stdio:统计 `codex_stdio.notification`、`tool_call.started|completed`、assistant delta 和 diff notification。
3. 查 AgentRun durable events:按 seq 分页,精确下钻 tool/diff/terminal event;不得只看默认前 20 条。
4. 查 Kafka control plane:确认 broker、topic ready、end offset 和相关 fixed group lag。
5. 先执行 `--no-publish`/`--dry-run` regenerate,对照源 frame 数、解析数、产出 event 数、拒绝原因和 lineage。
6. 需要跨 mapper 校对时,才把结果写入 debug topic,再由下一层从 debug topic regenerate;每层保存 count/hash/seq/type/tool identity。
7. 需要浏览器产品 reducer 校对时,使用 Workbench 隔离 debug replay;不得直接操作产品 SSE group。
8. 输出首个发生差异的层级和缺失 identity,不用后层缺失反推前层未产生
5. 怀疑重复投影时,先按同一 `sourceEventId` 聚合 `hwlab.event.v1`
- 一条 AgentRun durable event 对应多个 HWLAB envelope,且 `source`/producer 不同,直接判为多 projector 重复消费;
- 同一 producer 重复出现时,再调查 offset commit、重试和幂等;
- 上游 `agentrun.event.v1` 的 event id 唯一时,不再重复调查 provider、runner 或前端 reducer
6. 先执行 `--no-publish`/`--dry-run` regenerate,对照源 frame 数、解析数、产出 event 数、拒绝原因和 lineage。
7. 需要跨 mapper 校对时,才把结果写入 debug topic,再由下一层从 debug topic regenerate;每层保存 count/hash/seq/type/tool identity。
8. 需要浏览器产品 reducer 校对时,使用 Workbench 隔离 debug replay;不得直接操作产品 SSE group。
9. 输出首个发生差异的层级和缺失 identity,不用后层缺失反推前层未产生。
## 受控 CLI
@@ -158,6 +166,10 @@ bun scripts/cli.ts web-probe observe status <observerId> \
- stdio 有完整 tool/diff,但 regenerate 未产出对应 AgentRun event:归因 reducer/mapping。
- regenerate 能产出,但 durable `agentrun.event.v1` 缺失:归因 producer/durable append。
- AgentRun 产品 event 存在,而 HWLAB debug mapper不产出:归因 HWLAB mapper。
- 同一 AgentRun `eventId` 在 `hwlab.event.v1` 对应多个不同 producer 的 envelope
- 归因 direct projector consumer group 所有权错误;
- owning YAML 中共享 projector group 是修复点;
- 前端按 event id 去重只能作为显示层防御,不能作为根因修复。
- `hwlab.event.v1` 存在且 direct publish/fanout 数相等,但 UI/read model 缺失:归因 SSE/reducer/projection。
- `turn/diff/updated` 只保留 phase/correlation、不保留文件或变更摘要时,应报告为源映射降维;不能把它描述成“后续展示偶然漏了一条”。
- 直接答复没有工具调用时,不要求合成 tool event。
+2 -2
View File
@@ -344,7 +344,7 @@ lanes:
agentRunEventTopic: agentrun.event.v1
hwlabEventTopic: hwlab.event.v1
clientId: hwlab-v03-cloud-api
directPublishConsumerGroupId: hwlab-v03-agentrun-event-direct-publish
directPublishConsumerGroupId: hwlab-agentrun-event-direct-publish
transactionalProjectorConsumerGroupId: hwlab-v03-agentrun-event-projector
hwlabEventConsumerGroupId: hwlab-v03-workbench-live-sse
externalPostgres:
@@ -475,7 +475,7 @@ lanes:
groupIdPrefix: hwlab-production-cloud-api-sse
configRef: config/platform-infra/kafka.yaml#clients.hwlab-production-cloud-api
clientId: hwlab-production-cloud-api
directPublishConsumerGroupId: hwlab-production-agentrun-event-direct-publish
directPublishConsumerGroupId: hwlab-agentrun-event-direct-publish
transactionalProjectorConsumerGroupId: hwlab-production-agentrun-event-projector
hwlabEventConsumerGroupId: hwlab-production-workbench-live-sse
externalPostgres:
@@ -785,11 +785,15 @@ describe("runtime source-commit selection", () => {
});
expect(spec.codeAgentRuntime?.kafkaEventBridge).toMatchObject({
enabled: true,
directPublishConsumerGroupId: "hwlab-agentrun-event-direct-publish",
transactionalProjectorConsumerGroupId: "hwlab-v03-agentrun-event-projector",
hwlabEventConsumerGroupId: "hwlab-v03-workbench-live-sse",
features: expect.any(Object),
refreshReplay: expect.any(Object),
});
expect(
hwlabRuntimeLaneSpecForNode("production", "NC01").codeAgentRuntime?.kafkaEventBridge?.directPublishConsumerGroupId,
).toBe(spec.codeAgentRuntime?.kafkaEventBridge?.directPublishConsumerGroupId);
expect(remoteAnnotations).toEqual(pipelineProvenanceAnnotations(remoteProvenance));
expect(Object.keys(remoteAnnotations).sort()).toEqual(Object.values(pipelineProvenanceAnnotationKeys).sort());
expect(readFileSync(rootPath("scripts", "src", "hwlab-node", "render.ts"), "utf8")).not.toContain("overlay.sourceArtifactProvenance");