diff --git a/deploy/deploy.schema.json b/deploy/deploy.schema.json index b77b0852..d09c0da3 100644 --- a/deploy/deploy.schema.json +++ b/deploy/deploy.schema.json @@ -587,7 +587,8 @@ "workbenchUiPolicy": { "type": "object", "properties": { - "traceTimeline": { "$ref": "#/$defs/workbenchTraceTimelinePolicy" } + "traceTimeline": { "$ref": "#/$defs/workbenchTraceTimelinePolicy" }, + "rawHwlabEventWindow": { "$ref": "#/$defs/workbenchRawHwlabEventWindowPolicy" } }, "additionalProperties": false }, @@ -599,6 +600,16 @@ }, "additionalProperties": false }, + "workbenchRawHwlabEventWindowPolicy": { + "type": "object", + "required": ["enabled", "maxEntries", "maxRetainedBytes"], + "properties": { + "enabled": { "type": "boolean" }, + "maxEntries": { "type": "integer", "minimum": 1 }, + "maxRetainedBytes": { "type": "integer", "minimum": 1 } + }, + "additionalProperties": false + }, "externalPostgres": { "type": "object", "required": ["enabled", "serviceName", "endpointAddress", "port"], diff --git a/deploy/deploy.yaml b/deploy/deploy.yaml index 59f8c88b..77725898 100644 --- a/deploy/deploy.yaml +++ b/deploy/deploy.yaml @@ -374,6 +374,10 @@ lanes: traceTimeline: autoExpandRunning: false autoCollapseTerminal: false + rawHwlabEventWindow: + enabled: true + maxEntries: 200 + maxRetainedBytes: 1048576 observability: traceExplorerUrlTemplate: /v1/workbench/traces/{trace_id}/events prometheusOperatorResources: false diff --git a/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.1_Task_Report.md b/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.1_Task_Report.md new file mode 100644 index 00000000..3a27a277 --- /dev/null +++ b/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.1_Task_Report.md @@ -0,0 +1,47 @@ +# R2.1 纯 Kafka Trace 序列权威修复报告 + +## 上下文 + +- 前端根因: + - [HWLAB #2475](https://github.com/pikasTech/HWLAB/issues/2475#issuecomment-4935381417) +- 运行面证据: + - [HWLAB #2477](https://github.com/pikasTech/HWLAB/issues/2477#issuecomment-4935385934) + +## 完成结果 + +- 共享 Trace renderer 按 `runnerTrace.eventSource` 选择序列权威: + - `hwlab-kafka-sse` 使用 `sourceSeq`; + - `trace-api` 和其他 durable 路径继续只使用 `projectedSeq`。 +- live Kafka 事件保持前端 ingress 数组顺序: + - 不按 `sourceSeq` 重新排序; + - 优先以 `sourceEventId` 去重; + - 缺少该字段时使用 `runId + sourceSeq`; + - 不伪造 `projectedSeq`。 +- Trace 行显式携带 `sequenceAuthority`: + - DOM 通用属性为 `data-event-seq` 与 `data-event-seq-authority`; + - 只有 durable 行继续暴露 `data-projected-seq`。 +- live Trace 合并保持: + - source identity 去重; + - terminal 事件不清空既有事件; + - `eventCount` 等于已保存事件数。 + +## 验证 + +- production mapper 验证: + - inner event 携带 `sourceSeq`; + - inner event 不产生 `projectedSeq`。 +- renderer 与状态验证: + - source-only assistant、tool 和 terminal 能生成可读行; + - 行顺序与 ingress 顺序一致; + - durable Trace 缺少 `projectedSeq` 时仍拒绝渲染; + - 重复 `sourceEventId` 不产生重复行。 +- 定向测试结果: + - Bun 后端、renderer、merge 和 SSE API:`50 pass`; + - TraceTimeline 与调试面板组件:`4 pass`; + - v03 GitOps render 目标用例:`1 pass`。 + +## 边界 + +- 未启用或调用 projector、read model、snapshot、sync、recovery replay、gap-fill 或 polling。 +- 未部署运行面。 +- 未修改父 issue [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474)。 diff --git a/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.2_Task_Report.md b/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.2_Task_Report.md new file mode 100644 index 00000000..e98f0067 --- /dev/null +++ b/docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.2_Task_Report.md @@ -0,0 +1,76 @@ +# R2.2 原始 HWLAB Event 观察窗报告 + +## 上下文 + +- 主问题: + - [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474) +- 前端设计与最小 seam: + - [HWLAB #2475](https://github.com/pikasTech/HWLAB/issues/2475#issuecomment-4935381417) + +## 完成结果 + +- 现有产品 EventSource 增加单一 ingress tee: + - 每个命名 `hwlab.event.v1` frame 在 decode 和 coalesce 前记录一次; + - 同一次 decode 结果继续进入原生产 reducer; + - 未增加产品 EventSource、Kafka consumer 或第二事实源。 +- 原始观察状态为 bounded 内存: + - 按到达顺序保留完整 `MessageEvent.data`; + - 不按 trace、event type 或 reducer 结果做客户端过滤; + - invalid JSON 和非对象 JSON 保留原文并显示拒绝原因; + - 淘汰完整 envelope,不截断仍被保留的 envelope; + - 不写 localStorage、数据库或 canonical Workbench store。 +- 隔离调试卡片增加独立子标签: + - `Trace 卡片` 保留隔离 debug replay 和生产 reducer 卡片; + - `原始 HWLAB Event` 使用只读纯文本框显示同一产品 ingress; + - 切换标签不创建或关闭网络连接; + - 清空观察窗只清原始内存 buffer。 +- R2.3 v2 UI 集成点已预留: + - 支持 `replayId` 和 output barrier query 参数; + - 接收 `replay-complete.result`; + - 展示 server `scanned/matched/delivered`; + - 用明确标注的 client local overlay 展示 `received/decoded/applied`。 + +## YAML-first + +- owning YAML: + - `deploy/deploy.yaml#lanes.v03.workbench.rawHwlabEventWindow`。 +- 独立字段: + - `enabled`; + - `maxEntries`; + - `maxRetainedBytes`。 +- 渲染链: + - deploy schema 校验字段形状; + - GitOps renderer 生成 Cloud Web env; + - Cloud Web runtime 注入浏览器配置; + - 前端只读取显式值,不设置容量默认值。 +- 特例分类: + - `keep-domain-special`:观察窗保持管理员可见、内存只读; + - `remove-code-default`:容量与开关没有代码默认值; + - `legacy-retire`:本任务没有新增或保留旧入口。 + +## 验证 + +- ingress 与 buffer: + - 每个产品 frame 恰好产生一条 raw ingress 和一次 reducer delivery; + - invalid JSON 可见但不进入 reducer; + - entry/byte eviction 均保持完整 envelope。 +- UI: + - 两个子标签、局部清空和 raw 文本顺序通过; + - 标签切换期间 EventSource 实例数保持不变。 +- 配置与 runtime: + - Cloud Web runtime:`18 pass`; + - v03 GitOps render 目标用例:`1 pass`; + - deploy 合同:`15` 个 JSON 文件验证通过。 +- 定向功能测试: + - Bun 后端、renderer、merge 和 SSE API:`50 pass`; + - TraceTimeline 与调试面板组件:`4 pass`。 +- 已知基线失败: + - renderer 全套仍有两个与本变更无关的旧断言失败; + - Web check 的 source-shape、显式 any 扫描和 Vite build 通过; + - `workbench-isolated-kafka-debug.test.ts:61` 仍有 R2.3 分支已有的类型断言错误。 + +## 边界 + +- 未部署 NC01/v03。 +- 未修改父 issue [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474)。 +- 产品链继续保持纯 Kafka 实时,不增加补洞或投影路径。 diff --git a/docs/MDTODO/pure-kafka-live-single-step-debug.md b/docs/MDTODO/pure-kafka-live-single-step-debug.md index d014ba07..13727bbc 100644 --- a/docs/MDTODO/pure-kafka-live-single-step-debug.md +++ b/docs/MDTODO/pure-kafka-live-single-step-debug.md @@ -19,10 +19,10 @@ ## R2 [in_progress] 修复纯 Kafka 实时 Trace 可读性、隔离调试分层诊断与同一 SSE ingress 原始 HWLAB Event 观察窗;保持产品链无 projector、read model、snapshot、sync、recovery replay 或 polling。上下文:[HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474) 与 [调查结论](https://github.com/pikasTech/HWLAB/issues/2474#issuecomment-4935408473),完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2_Task_Report.md)。 -### R2.1 [in_progress] +### R2.1 [completed] 按 eventSource 修复 Trace 序列权威:live Kafka 保留 ingress 顺序并以 sourceEventId 或 runId+sourceSeq 识别事件,不伪造 projectedSeq;补 sourceSeq-only running/assistant/tool/terminal、去重与 terminal 保留合同。上下文:[前端根因 #2475](https://github.com/pikasTech/HWLAB/issues/2475#issuecomment-4935381417)、[运行面证据 #2477](https://github.com/pikasTech/HWLAB/issues/2477#issuecomment-4935385934)。依赖:无。验证:真实 mapper envelope 的 eventCount 等于 stored events、readableRows 大于 0,projection 保持关闭,完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.1_Task_Report.md)。 -### R2.2 +### R2.2 [completed] 在同一产品 EventSource 的 decode/coalesce 前 tee 原始 frame,并在隔离调试卡片增加 Trace 卡片/原始 HWLAB Event 子标签;raw 按到达顺序写入 bounded 内存、暴露 invalid JSON,YAML-first 开关和容量独立配置,不新增 SSE 或 Kafka consumer。上下文:[主 issue #2474](https://github.com/pikasTech/HWLAB/issues/2474)、[前端设计 #2475](https://github.com/pikasTech/HWLAB/issues/2475#issuecomment-4935381417)。依赖:无,可与 Trace 序列权威修复并行。验证:每个产品 frame 恰好 raw+1 且 reducer 一次,切换标签不新增 EventSource,eviction/invalid-json/runtime contract 通过,完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.2_Task_Report.md)。 ### R2.3 diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 8b4aa3e3..9634eee8 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -88,6 +88,8 @@ test("projects AgentRun assistant_message Kafka event into HWLAB trace event", ( assert.equal(projected.context.sourceSeq, 7); assert.equal(projected.context.agentRunEventType, "assistant_message"); assert.equal(projected.event.type, "assistant"); + assert.equal(projected.event.sourceSeq, 7); + assert.equal(Object.prototype.hasOwnProperty.call(projected.event, "projectedSeq"), false); assert.equal(projected.event.label, "agentrun:assistant:message"); assert.equal(projected.event.status, "running"); assert.equal(projected.event.text, "partial answer"); diff --git a/internal/dev-entrypoint/cloud-web-runtime.mjs b/internal/dev-entrypoint/cloud-web-runtime.mjs index e63230a0..224d6ca4 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.mjs @@ -396,7 +396,12 @@ function workbenchRuntimeConfigFromEnv() { projectionRealtime: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED, "HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED") }, debugCapabilities: { - isolatedKafka: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED, "HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED") + isolatedKafka: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED, "HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED"), + rawHwlabEventWindow: { + enabled: requiredWorkbenchRealtimeFeature(process.env.HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED"), + maxEntries: requiredPositiveWorkbenchInteger(process.env.HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES"), + maxRetainedBytes: requiredPositiveWorkbenchInteger(process.env.HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES") + } } }; const autoExpandRunning = parseEnvBoolean(process.env.HWLAB_WORKBENCH_TRACE_AUTO_EXPAND_RUNNING); @@ -416,6 +421,12 @@ function requiredWorkbenchRealtimeFeature(value, name) { throw new Error(`${name} is required and must be explicitly true or false`); } +function requiredPositiveWorkbenchInteger(value, name) { + const number = Number(value); + if (!Number.isInteger(number) || number <= 0) throw new Error(`${name} is required and must be a positive integer`); + return number; +} + function traceExplorerUrlTemplateFromEnv(value) { if (value === undefined || value === null || value === "") return null; const template = String(value).trim(); diff --git a/internal/dev-entrypoint/cloud-web-runtime.test.mjs b/internal/dev-entrypoint/cloud-web-runtime.test.mjs index 5a78aa2b..afd75ac9 100644 --- a/internal/dev-entrypoint/cloud-web-runtime.test.mjs +++ b/internal/dev-entrypoint/cloud-web-runtime.test.mjs @@ -432,7 +432,10 @@ test("cloud web serves client deep links through the Vue shell", async () => { HWLAB_CLOUD_WEB_DISPLAY_TIME_LABEL: "北京时间", HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false", - HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES: "200", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES: "1048576" }); const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-cloud-web-runtime-")); await writeFile(path.join(root, "index.html"), "
\n", "utf8"); @@ -463,7 +466,7 @@ test("cloud web serves client deep links through the Vue shell", async () => { const html = await response.text(); assert.match(html, /
<\/div>/u, route); assert.match(html, /HWLAB_CLOUD_WEB_CONFIG/u, route); - assert.match(html, /"debugCapabilities":\{"isolatedKafka":true\}/u, route); + assert.match(html, /"debugCapabilities":\{"isolatedKafka":true,"rawHwlabEventWindow":\{"enabled":true,"maxEntries":200,"maxRetainedBytes":1048576\}\}/u, route); } const assetResponse = await fetch(`${serverUrl(cloudWeb)}/asset.txt`, { @@ -491,7 +494,10 @@ test("cloud web injects trace explorer runtime config from env", async () => { HWLAB_WORKBENCH_TRACE_EXPLORER_URL_TEMPLATE: "/v1/workbench/traces/{trace_id}/events", HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false", - HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES: "200", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES: "1048576" }); const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-cloud-web-runtime-")); await writeFile(path.join(root, "index.html"), "
\n", "utf8"); @@ -598,7 +604,10 @@ test("cloud web OpenCode frame-url endpoint mints traced tickets", async () => { HWLAB_CLOUD_WEB_OPENCODE_URL: "https://opencode.example.test", HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false", - HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES: "200", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES: "1048576" }); const cloudApiRequests = []; const opencodeRequests = []; @@ -1041,7 +1050,10 @@ test("cloud web OpenCode proxy accepts short-lived tickets minted by the shell", HWLAB_CLOUD_WEB_OPENCODE_URL: "https://opencode.example.test", HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED: "true", HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED: "false", - HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true" + HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED: "true", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES: "200", + HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES: "1048576" }); const root = await mkdtemp(path.join(os.tmpdir(), "hwlab-cloud-web-runtime-")); await writeFile(path.join(root, "index.html"), "
\n", "utf8"); diff --git a/scripts/gitops-render.test.ts b/scripts/gitops-render.test.ts index e9ffcce3..62e7ada9 100644 --- a/scripts/gitops-render.test.ts +++ b/scripts/gitops-render.test.ts @@ -660,6 +660,9 @@ test("v03 render keeps node identity as data instead of generated structure", as assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_LIVE_KAFKA_SSE_ENABLED"), "true"); assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_PROJECTION_REALTIME_ENABLED"), "false"); assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED"), "true"); + assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED"), "1"); + assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES"), "200"); + assert.equal(cloudWebEnv.get("HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES"), "1048576"); assert.match(cloudApiEnv.get("HWLAB_ENVIRONMENT_IMAGE") ?? "", /hwlab-cloud-api-env:env-stale/u); assert.deepEqual(cloudApiEnvEntries.get("HWLAB_SECRET_PLANE_SMOKE")?.valueFrom?.secretKeyRef, { name: "hwlab-secret-plane-smoke", key: "password", optional: true }); assert.ok((cloudApi?.spec?.template?.spec?.volumes ?? []).some((volume) => volume.name === "hwpod-preinstalled-specs" && volume.configMap?.name === "hwlab-v03-hwpod-preinstalled-specs")); diff --git a/scripts/src/gitops-render/core.mjs b/scripts/src/gitops-render/core.mjs index eccbf06a..3285aaf0 100644 --- a/scripts/src/gitops-render/core.mjs +++ b/scripts/src/gitops-render/core.mjs @@ -1136,6 +1136,17 @@ function runtimeWorkbenchTraceTimelinePolicy(deploy, profile) { return Object.keys(result).length > 0 ? result : null; } +function runtimeWorkbenchRawHwlabEventWindowPolicy(deploy, profile) { + if (!isRuntimeLane(profile)) return null; + const policy = runtimeLaneConfig(deploy, profile)?.workbench?.rawHwlabEventWindow; + if (!policy || typeof policy !== "object" || Array.isArray(policy)) return null; + return { + enabled: policy.enabled, + maxEntries: policy.maxEntries, + maxRetainedBytes: policy.maxRetainedBytes + }; +} + function runtimeTraceExplorerUrlTemplate(deploy, profile) { if (!isRuntimeLane(profile)) return null; const template = runtimeLaneConfig(deploy, profile)?.observability?.traceExplorerUrlTemplate; @@ -1586,6 +1597,12 @@ function transformWorkloads({ workloads, deploy, catalog, source, sourceBranch = if (typeof traceTimelinePolicy?.autoCollapseTerminal === "boolean") { upsertEnv(container.env, "HWLAB_WORKBENCH_TRACE_AUTO_COLLAPSE_TERMINAL", booleanEnv(traceTimelinePolicy.autoCollapseTerminal)); } + const rawHwlabEventWindowPolicy = runtimeWorkbenchRawHwlabEventWindowPolicy(deploy, profile); + if (rawHwlabEventWindowPolicy) { + upsertEnv(container.env, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_ENABLED", booleanEnv(rawHwlabEventWindowPolicy.enabled)); + upsertEnv(container.env, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_ENTRIES", String(rawHwlabEventWindowPolicy.maxEntries)); + upsertEnv(container.env, "HWLAB_WORKBENCH_RAW_HWLAB_EVENT_WINDOW_MAX_RETAINED_BYTES", String(rawHwlabEventWindowPolicy.maxRetainedBytes)); + } const traceExplorerUrlTemplate = runtimeTraceExplorerUrlTemplate(deploy, profile); if (traceExplorerUrlTemplate) { upsertEnv(container.env, "HWLAB_WORKBENCH_TRACE_EXPLORER_URL_TEMPLATE", traceExplorerUrlTemplate); diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index 3df7857e..3df3f248 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -1,6 +1,7 @@ export interface TraceEventRow { rowId: string; seq: number | null; + sequenceAuthority: TraceSequenceAuthority; tone: "ok" | "blocked" | "warn" | "source"; header: string; body: string | null; @@ -11,13 +12,25 @@ export interface TraceEventRow { type TraceEvent = Record; type TraceClockFormatter = (value: string) => string; +export type TraceSequenceAuthority = "projected" | "source"; export interface TraceDisplayRowsOptions { formatClock?: TraceClockFormatter; } +interface ResolvedTraceDisplayRowsOptions extends TraceDisplayRowsOptions { + sequenceAuthority: TraceSequenceAuthority; +} + export function traceDisplayRows(trace: Record = {}, events: TraceEvent[] = [], options: TraceDisplayRowsOptions = {}): TraceEventRow[] { - const orderedEvents = traceEventsForDisplay(events.filter(hasProjectedTraceSequence)); + const displayOptions: ResolvedTraceDisplayRowsOptions = { + ...options, + sequenceAuthority: traceSequenceAuthority(trace) + }; + const orderedEvents = traceEventsForDisplay( + events.filter((event) => hasTraceSequence(event, displayOptions.sequenceAuthority)), + displayOptions.sequenceAuthority + ); const effectiveTrace = traceWithInferredStart(trace, orderedEvents); const finalResponseText = traceFinalResponseText(effectiveTrace); const finalResponseFingerprint = traceMessageFingerprint(finalResponseText); @@ -41,7 +54,7 @@ export function traceDisplayRows(trace: Record = {}, events: Tr if (renderedToolIdentities.has(identity)) continue; renderedToolIdentities.add(identity); } - rows.push(traceToolCallRow(effectiveTrace, event, options)); + rows.push(traceToolCallRow(effectiveTrace, event, displayOptions)); continue; } if (isRequestTraceEvent(event)) { @@ -53,33 +66,37 @@ export function traceDisplayRows(trace: Record = {}, events: Tr if (isTerminalAssistantTraceEvent(event)) { if (hasCompletionEvent) continue; if (finalResponseText) { - rows.push(traceFinalResponseRow(event, finalResponseText, options)); + rows.push(traceFinalResponseRow(event, finalResponseText, displayOptions)); } else { - rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true }, options)); + rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: true }, displayOptions)); } continue; } if (isAssistantTraceEvent(event)) { if (finalResponseFingerprint && traceAssistantEventFingerprint(effectiveTrace, event) === finalResponseFingerprint) continue; - rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, options)); + rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, displayOptions)); continue; } if (isCompletionTraceEvent(event)) { completionEvent ??= event; continue; } - rows.push(traceDisplayRow(effectiveTrace, event, options)); + rows.push(traceDisplayRow(effectiveTrace, event, displayOptions)); } - if (completionEvent) rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent, options)); - const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, orderedEvents, options); + if (completionEvent) rows.push(traceCompletionSummaryRow(effectiveTrace, completionEvent, displayOptions)); + const progressRow = traceBackendProgressSummaryRow(effectiveTrace, rows, orderedEvents, displayOptions); if (progressRow) rows.push(progressRow); if (rows.length > 0) return rows; if (orderedEvents.length === 0) return []; - return orderedEvents.filter((event) => !isSuppressedTraceEvent(event)).map((event) => traceDisplayRow(effectiveTrace, event, options)); + return orderedEvents.filter((event) => !isSuppressedTraceEvent(event)).map((event) => traceDisplayRow(effectiveTrace, event, displayOptions)); } -function hasProjectedTraceSequence(event: TraceEvent | null | undefined): boolean { - return traceEventDisplaySeq(event) !== null; +function traceSequenceAuthority(trace: Record): TraceSequenceAuthority { + return nonEmptyString(trace.eventSource) === "hwlab-kafka-sse" ? "source" : "projected"; +} + +function hasTraceSequence(event: TraceEvent | null | undefined, authority: TraceSequenceAuthority): boolean { + return traceEventDisplaySeq(event, authority) !== null; } export function renderTraceRowsMarkdown(rows: TraceEventRow[] = []): string { @@ -106,10 +123,11 @@ function renderTraceMessageRowMarkdown(row: TraceEventRow): string { return traceBoldMarkdown(row.header || `_No readable trace row body._`); } -function traceEventsForDisplay(events: TraceEvent[] = []): TraceEvent[] { +function traceEventsForDisplay(events: TraceEvent[] = [], authority: TraceSequenceAuthority): TraceEvent[] { + if (authority === "source") return [...events]; return [...events].sort((left, right) => { - const leftSeq = traceEventDisplaySeq(left); - const rightSeq = traceEventDisplaySeq(right); + const leftSeq = traceEventDisplaySeq(left, authority); + const rightSeq = traceEventDisplaySeq(right, authority); if (leftSeq !== null || rightSeq !== null) { if (leftSeq === null) return 1; if (rightSeq === null) return -1; @@ -126,12 +144,20 @@ function traceEventsForDisplay(events: TraceEvent[] = []): TraceEvent[] { }); } -function traceEventDisplaySeq(event: TraceEvent | null | undefined): number | null { - return numberOrNull(event?.projectedSeq); +function traceEventDisplaySeq(event: TraceEvent | null | undefined, authority: TraceSequenceAuthority): number | null { + return numberOrNull(authority === "source" ? event?.sourceSeq : event?.projectedSeq); } -function traceEventIdentityToken(event: TraceEvent): string | number | null { - const seq = traceEventDisplaySeq(event); +function traceEventIdentityToken(event: TraceEvent, authority: TraceSequenceAuthority): string | number | null { + if (authority === "source") { + const sourceEventId = nonEmptyString(event.sourceEventId); + if (sourceEventId) return sourceEventId; + const runId = nonEmptyString(event.runId); + const sourceSeq = traceEventDisplaySeq(event, authority); + if (runId && sourceSeq !== null) return `${runId}:${sourceSeq}`; + return sourceSeq; + } + const seq = traceEventDisplaySeq(event, authority); return seq ?? nonEmptyString(event.projectedSeq); } @@ -149,15 +175,16 @@ function stripTraceToolOutputFromSummary(value: string): string { return value.replace(/\s+(stdout:|stderr:|exitCode=).*$/isu, "").trim(); } -function traceToolCallRow(trace: Record, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow { +function traceToolCallRow(trace: Record, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const command = cleanShellCommand(event.command); const status = traceStatusToken(event); const toolName = traceToolName(event, command); const body = traceToolCallBody(event, command); - const eventKey = traceEventIdentityToken(event); + const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { rowId: `tool:${event.itemId ?? eventKey ?? `${event.label ?? event.type ?? "tool"}:${event.createdAt ?? "unknown"}`}`, - seq: traceEventDisplaySeq(event), + seq: traceEventDisplaySeq(event, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: traceEventTone(event), header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${status} ${toolName}`.trim(), body, @@ -221,14 +248,15 @@ export function traceNoiseEventCount(events: TraceEvent[] = []): number { return Array.isArray(events) ? events.filter((event) => isNoisyTraceEvent(event)).length : 0; } -function traceAssistantMessageRow(trace: Record, event: TraceEvent, { terminal }: { terminal: boolean }, options: TraceDisplayRowsOptions): TraceEventRow { +function traceAssistantMessageRow(trace: Record, event: TraceEvent, { terminal }: { terminal: boolean }, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const text = traceAssistantEventText(trace, event); const index = Number.isInteger(event.messageIndex) ? Number(event.messageIndex) : null; const count = Number.isInteger(event.messageCount) ? Number(event.messageCount) : null; - const eventKey = traceEventIdentityToken(event); + const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { rowId: `event:${eventKey ?? `${event.label ?? event.type ?? "assistant"}:${event.createdAt ?? "unknown"}`}`, - seq: traceEventDisplaySeq(event), + seq: traceEventDisplaySeq(event, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: "ok", header: `${traceEventClock(event, options)} ${terminal ? "助手最终消息" : "助手消息"}${index ? ` ${index}${count ? `/${count}` : ""}` : ""}`, terminal: terminal ? true : undefined, @@ -250,11 +278,12 @@ function traceMessageFingerprint(value: unknown): string | null { return text || null; } -function traceFinalResponseRow(event: TraceEvent, finalText: string, options: TraceDisplayRowsOptions): TraceEventRow { - const eventKey = traceEventIdentityToken(event); +function traceFinalResponseRow(event: TraceEvent, finalText: string, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { + const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { rowId: `trace-final-response:${eventKey ?? "completed"}`, - seq: traceEventDisplaySeq(event), + seq: traceEventDisplaySeq(event, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: "ok", header: `${traceEventClock(event, options)} 助手最终消息`, terminal: true, @@ -268,24 +297,26 @@ function traceFinalResponseText(trace: Record): string | null { return nonEmptyString(response?.text ?? response?.content ?? response?.message); } -function traceCompletionSummaryRow(trace: Record, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow { - const eventKey = traceEventIdentityToken(event); +function traceCompletionSummaryRow(trace: Record, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { + const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { rowId: `trace-completion:${eventKey ?? "turn"}`, - seq: traceEventDisplaySeq(event), + seq: traceEventDisplaySeq(event, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: traceEventTone(event), header: `${traceEventClock(event, options)} 轮次完成(总耗时 ${formatTraceDuration(traceRelativeMs(trace, event))})`, body: null }; } -function traceNoiseSummaryRow(trace: Record, events: TraceEvent[], options: TraceDisplayRowsOptions): TraceEventRow { +function traceNoiseSummaryRow(trace: Record, events: TraceEvent[], options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const lastEvent = events.at(-1); const status = nonEmptyString(trace.status ?? trace.traceStatus) ?? "running"; const lastLabel = lastEvent ? readableTraceLabel(lastEvent) : "未观测"; return { rowId: "trace-noise-summary", - seq: traceEventDisplaySeq(lastEvent), + seq: traceEventDisplaySeq(lastEvent, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: lastEvent ? traceEventTone(lastEvent) : "source", header: `Trace ${status},等待可读事件`, body: `已隐藏 ${events.length} 条 AgentRun backend 状态事件。最新原始事件:${lastLabel}。`, @@ -293,23 +324,24 @@ function traceNoiseSummaryRow(trace: Record, events: TraceEvent }; } -function traceBackendProgressSummaryRow(trace: Record, rows: TraceEventRow[], events: TraceEvent[], options: TraceDisplayRowsOptions): TraceEventRow | null { +function traceBackendProgressSummaryRow(trace: Record, rows: TraceEventRow[], events: TraceEvent[], options: ResolvedTraceDisplayRowsOptions): TraceEventRow | null { if (!Array.isArray(events) || events.length === 0) return null; if (rows.some((row) => row.rowId.startsWith("tool:") || /助手|轮次完成/u.test(row.header))) return null; const lastRenderedSeq = Math.max(0, ...rows.map((row) => row.seq ?? 0).filter(Number.isFinite)); const hiddenProgress = events.filter((event) => { - const seq = traceEventDisplaySeq(event) ?? 0; + const seq = traceEventDisplaySeq(event, options.sequenceAuthority) ?? 0; return seq > lastRenderedSeq && (isRequestTraceEvent(event) || isSetupTraceEvent(event) || isNoisyTraceEvent(event)); }); return hiddenProgress.length > 0 ? traceNoiseSummaryRow(trace, hiddenProgress, options) : null; } -function traceDisplayRow(trace: Record, event: TraceEvent, options: TraceDisplayRowsOptions): TraceEventRow { +function traceDisplayRow(trace: Record, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const label = readableTraceLabel(event); - const eventKey = traceEventIdentityToken(event); + const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); return { rowId: `event:${eventKey ?? `${event.label ?? event.type ?? "event"}:${event.createdAt ?? "unknown"}`}`, - seq: traceEventDisplaySeq(event), + seq: traceEventDisplaySeq(event, options.sequenceAuthority), + sequenceAuthority: options.sequenceAuthority, tone: traceEventTone(event), header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${traceStatusToken(event)} ${label}`.trim(), body: traceDisplayBody(event) @@ -380,8 +412,12 @@ function toolIdentity(event: TraceEvent): string | null { function traceSourceEventKey(event: TraceEvent | null | undefined): string | null { if (!event || typeof event !== "object") return null; - const source = nonEmptyString(event.source); + const sourceEventId = nonEmptyString(event.sourceEventId); + if (sourceEventId) return `source-event:${sourceEventId}`; + const runId = nonEmptyString(event.runId); const sourceSeq = nonEmptyString(event.sourceSeq); + if (runId && sourceSeq) return `run-source:${runId}:${sourceSeq}`; + const source = nonEmptyString(event.source); if (!source || !sourceSeq) return null; return `${source}:${sourceSeq}:${nonEmptyString(event.label ?? event.type) ?? "event"}`; } diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index 924f0a75..2ce7084c 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts @@ -102,7 +102,7 @@ const terminalAssistantFinalText = [ ].join("\n"); const canonicalTraceFinalText = "messageProjection 的 sealed final response 才是主消息正文。"; const traceDetailConflictText = "TRACE_DETAIL_CONFLICT_SHOULD_NOT_REPLACE_MESSAGE"; -const runtimeConfigScript = ``; +const runtimeConfigScript = ``; let state = createScenarioState("baseline"); const sseClients = new Set(); diff --git a/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts b/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts index 3a4c7c73..014747ec 100644 --- a/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-r1-parity.test.ts @@ -117,6 +117,24 @@ test("R1 shared trace renderer keeps UTC by default and lets Web supply display assert.match(displayRows[0]?.header ?? "", /^14:34:14 /u); }); +test("pure Kafka Trace renders sourceSeq-only rows in ingress order without projectedSeq", () => { + const events = [ + { sourceEventId: "evt_tool", runId: "run_live", source: "agentrun.kafka", sourceSeq: 8, label: "agentrun:tool:commandExecution", type: "tool_call", toolName: "commandExecution", status: "completed", command: "make test" }, + { sourceEventId: "evt_assistant", runId: "run_live", source: "agentrun.kafka", sourceSeq: 7, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "live answer" }, + { sourceEventId: "evt_assistant", runId: "run_live", source: "agentrun.kafka", sourceSeq: 70, label: "agentrun:assistant:message", type: "assistant", status: "running", message: "duplicate should not render" }, + { sourceEventId: "evt_terminal", runId: "run_live", source: "agentrun.kafka", sourceSeq: 9, label: "agentrun:terminal:completed", type: "result", status: "completed", terminal: true } + ]; + + const liveRows = traceDisplayRows({ traceId: "trc_live_source", eventSource: "hwlab-kafka-sse", status: "completed" }, events); + assert.deepEqual(liveRows.map((row) => row.seq), [8, 7, 9]); + assert.ok(liveRows.every((row) => row.sequenceAuthority === "source")); + assert.equal(liveRows.some((row) => row.body === "duplicate should not render"), false); + assert.equal(events.some((event) => "projectedSeq" in event), false); + + const durableRows = traceDisplayRows({ traceId: "trc_durable", eventSource: "trace-api", status: "completed" }, events); + assert.deepEqual(durableRows, []); +}); + test("R1 trace API snapshots stay authoritative over compact result traces", () => { const previous = { traceId: "trc_merge", diff --git a/web/hwlab-cloud-web/src/api/workbench-debug.test.ts b/web/hwlab-cloud-web/src/api/workbench-debug.test.ts index 4e8cf354..f63d7895 100644 --- a/web/hwlab-cloud-web/src/api/workbench-debug.test.ts +++ b/web/hwlab-cloud-web/src/api/workbench-debug.test.ts @@ -28,6 +28,19 @@ test("isolated Kafka debug replay path is trace-scoped and explicit", () => { ); }); +test("isolated Kafka debug v2 path carries replay correlation and output barrier", () => { + assert.equal( + workbenchKafkaSseDebugPath({ + stream: "hwlab-debug", + fromBeginning: true, + traceId: "trc_isolated_v2", + replayId: "rpl_isolated_v2", + outputBarrier: { partition: 0, firstOffset: "42", lastOffset: "51" } + }), + "/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&fromBeginning=true&traceId=trc_isolated_v2&replayId=rpl_isolated_v2&partition=0&firstOffset=42&lastOffset=51" + ); +}); + test("Kafka SSE debug correlation ids normalize stdio snake_case metadata", () => { assert.deepEqual( workbenchKafkaSseDebugCorrelationIds({ diff --git a/web/hwlab-cloud-web/src/api/workbench-debug.ts b/web/hwlab-cloud-web/src/api/workbench-debug.ts index 35e91b5d..1f31db25 100644 --- a/web/hwlab-cloud-web/src/api/workbench-debug.ts +++ b/web/hwlab-cloud-web/src/api/workbench-debug.ts @@ -52,6 +52,44 @@ export interface WorkbenchDebugFakeSseStream { export type WorkbenchKafkaSseDebugStreamName = "stdio" | "agentrun" | "hwlab" | "hwlab-debug"; +export interface WorkbenchKafkaDebugOffsetBarrier { + partition: number; + firstOffset: string; + lastOffset: string; +} + +export interface WorkbenchKafkaDebugReplayCounts { + scanned?: number; + barrierScanned?: number; + parsed?: number; + traceMatched?: number; + replayMatched?: number; + matched?: number; + delivered?: number; + clientReceived?: number | null; + decoded?: number | null; + applied?: number | null; +} + +export interface WorkbenchKafkaDebugReplayResult { + contractVersion?: string; + replayId?: string | null; + traceId?: string | null; + phase?: string; + code?: string; + ok?: boolean; + message?: string; + terminalObserved?: boolean; + counts?: WorkbenchKafkaDebugReplayCounts; + offsetRange?: Record | null; + [key: string]: unknown; +} + +export interface WorkbenchKafkaSseDebugIngressFrame { + eventName: string; + decodeStatus: "accepted" | "invalid-json" | "not-object"; +} + export interface WorkbenchKafkaSseDebugEvent { ok?: boolean; contractVersion?: string; @@ -66,6 +104,12 @@ export interface WorkbenchKafkaSseDebugEvent { limit?: number; timeoutMs?: number; terminalObserved?: boolean; + replayId?: string | null; + phase?: string; + code?: string; + result?: WorkbenchKafkaDebugReplayResult | null; + counts?: WorkbenchKafkaDebugReplayCounts; + offsetRange?: Record | null; traceId?: string | null; groupId?: string; stream?: WorkbenchKafkaSseDebugStreamName; @@ -89,6 +133,9 @@ export interface WorkbenchKafkaSseDebugStreamOptions { sessionId?: string | null; runId?: string | null; commandId?: string | null; + replayId?: string | null; + outputBarrier?: WorkbenchKafkaDebugOffsetBarrier | null; + onIngress?: (frame: WorkbenchKafkaSseDebugIngressFrame) => void; onEvent: (event: WorkbenchKafkaSseDebugEvent, eventName: string) => void; onOpen?: () => void; onError?: (event: Event) => void; @@ -149,18 +196,13 @@ export function connectWorkbenchKafkaSseDebug(options: WorkbenchKafkaSseDebugStr source.onerror = (event) => options.onError?.(event); const listeners = DEBUG_KAFKA_SSE_EVENTS.map((name) => { const listener = (event: MessageEvent) => { - const payload = parseKafkaSseDebugEvent(event.data); - if (!payload) return; - options.onEvent(payload, name); - if (name === "hwlab.kafka.connected" && payload.consumerReady === true) options.onOpen?.(); - if (name === "hwlab.kafka.error") options.onError?.(event); + deliverKafkaSseDebugFrame(event.data, name, event, options); }; source.addEventListener(name, listener); return { name, listener }; }); source.onmessage = (event) => { - const payload = parseKafkaSseDebugEvent(event.data); - if (payload) options.onEvent(payload, "message"); + deliverKafkaSseDebugFrame(event.data, "message", event, options); }; return { close() { @@ -170,13 +212,19 @@ export function connectWorkbenchKafkaSseDebug(options: WorkbenchKafkaSseDebugStr }; } -export function workbenchKafkaSseDebugPath(options: Pick): string { +export function workbenchKafkaSseDebugPath(options: Pick): string { const query = new URLSearchParams({ stream: options.stream }); if (options.fromBeginning === true) query.set("fromBeginning", "true"); appendDebugFilter(query, "traceId", options.traceId); appendDebugFilter(query, "sessionId", options.sessionId); appendDebugFilter(query, "runId", options.runId); appendDebugFilter(query, "commandId", options.commandId); + appendDebugFilter(query, "replayId", options.replayId); + if (options.outputBarrier) { + query.set("partition", String(options.outputBarrier.partition)); + query.set("firstOffset", options.outputBarrier.firstOffset); + query.set("lastOffset", options.outputBarrier.lastOffset); + } return `/v1/workbench/debug/kafka-sse/events?${query.toString()}`; } @@ -214,12 +262,23 @@ function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { } } -function parseKafkaSseDebugEvent(raw: string): WorkbenchKafkaSseDebugEvent | null { +function deliverKafkaSseDebugFrame(raw: unknown, eventName: string, transportEvent: MessageEvent, options: WorkbenchKafkaSseDebugStreamOptions): void { + const decoded = parseKafkaSseDebugEvent(typeof raw === "string" ? raw : String(raw ?? "")); + options.onIngress?.({ eventName, decodeStatus: decoded.status }); + if (!decoded.payload) return; + options.onEvent(decoded.payload, eventName); + if (eventName === "hwlab.kafka.connected" && decoded.payload.consumerReady === true) options.onOpen?.(); + if (eventName === "hwlab.kafka.error") options.onError?.(transportEvent); +} + +function parseKafkaSseDebugEvent(raw: string): { payload: WorkbenchKafkaSseDebugEvent | null; status: WorkbenchKafkaSseDebugIngressFrame["decodeStatus"] } { try { const value = JSON.parse(raw); - return value && typeof value === "object" && !Array.isArray(value) ? value as WorkbenchKafkaSseDebugEvent : null; + return value && typeof value === "object" && !Array.isArray(value) + ? { payload: value as WorkbenchKafkaSseDebugEvent, status: "accepted" } + : { payload: null, status: "not-object" }; } catch { - return null; + return { payload: null, status: "invalid-json" }; } } diff --git a/web/hwlab-cloud-web/src/api/workbench-events.test.ts b/web/hwlab-cloud-web/src/api/workbench-events.test.ts index 7d5af779..320a25a9 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.test.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; -import { realtimeCoalesceKey, type WorkbenchRealtimeEvent } from "./workbench-events"; +import { connectWorkbenchEvents, realtimeCoalesceKey, type WorkbenchRealtimeEvent, type WorkbenchSseIngressFrame } from "./workbench-events"; import { createCoalescedEventQueue } from "../utils/scheduler/coalesced-event-queue"; test("turn snapshot coalescing keys by trace instead of per sequence", () => { @@ -39,3 +39,66 @@ test("live Kafka envelopes from one trace retain every assistant, tool, output, assert.deepEqual(delivered.map((event) => event.sourceEventId), ["evt_assistant", "evt_tool", "evt_output", "evt_terminal"]); }); + +test("product EventSource tees every raw HWLAB frame once before decode and reducer delivery", async () => { + const originalEventSource = globalThis.EventSource; + const ingress: WorkbenchSseIngressFrame[] = []; + const delivered: WorkbenchRealtimeEvent[] = []; + FakeProductEventSource.instances = []; + globalThis.EventSource = FakeProductEventSource as unknown as typeof EventSource; + try { + const stream = connectWorkbenchEvents({ + realtimeCapabilities: { liveKafkaSse: true, projectionRealtime: false }, + sessionId: "ses_raw_tee", + flushYieldMs: 1, + onIngress: (frame) => ingress.push(frame), + onEvent: (event) => delivered.push(event) + }); + assert.ok(stream); + const source = FakeProductEventSource.instances[0]; + assert.ok(source); + source.emitRaw("hwlab.event.v1", '{"schema":"hwlab.event.v1","eventId":"evt_raw_tee"}'); + source.emitRaw("hwlab.event.v1", "{invalid-json"); + await new Promise((resolve) => setTimeout(resolve, 10)); + + assert.equal(FakeProductEventSource.instances.length, 1); + assert.deepEqual(ingress.map((frame) => [frame.eventName, frame.decodeStatus]), [ + ["hwlab.event.v1", "accepted"], + ["hwlab.event.v1", "invalid-json"] + ]); + assert.equal(delivered.length, 1); + assert.equal(delivered[0]?.eventId, "evt_raw_tee"); + stream.close(); + } finally { + globalThis.EventSource = originalEventSource; + } +}); + +class FakeProductEventSource { + static instances: FakeProductEventSource[] = []; + onopen: ((event: Event) => void) | null = null; + onerror: ((event: Event) => void) | null = null; + onmessage: ((event: MessageEvent) => void) | null = null; + private readonly listeners = new Map void>>(); + + constructor(readonly url: string | URL) { + FakeProductEventSource.instances.push(this); + } + + addEventListener(name: string, listener: (event: MessageEvent) => void): void { + const listeners = this.listeners.get(name) ?? new Set(); + listeners.add(listener); + this.listeners.set(name, listeners); + } + + removeEventListener(name: string, listener: (event: MessageEvent) => void): void { + this.listeners.get(name)?.delete(listener); + } + + emitRaw(name: string, rawText: string): void { + const event = { data: rawText } as MessageEvent; + for (const listener of this.listeners.get(name) ?? []) listener(event); + } + + close(): void {} +} diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 12ba9def..087fb3d7 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -107,6 +107,7 @@ export interface WorkbenchEventStreamOptions { flushMaxItemsPerChunk?: number | null; flushMaxChunkMs?: number | null; flushYieldMs?: number | null; + onIngress?: (frame: WorkbenchSseIngressFrame) => void; onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void; onOpen?: () => void; onError?: (event: Event) => void; @@ -116,6 +117,12 @@ export interface WorkbenchEventStream { close: () => void; } +export interface WorkbenchSseIngressFrame { + eventName: string; + rawText: string; + decodeStatus: "accepted" | "invalid-json" | "not-object"; +} + interface QueuedRealtimeEvent { payload: WorkbenchRealtimeEvent; eventName: string; @@ -182,15 +189,13 @@ export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): Wo }; const listeners = WORKBENCH_EVENT_NAMES.map((name) => { const listener = (event: MessageEvent) => { - const payload = parseRealtimeEvent(event.data); - if (payload) queue.push({ payload, eventName: name }); + enqueueRealtimeFrame(event.data, name, options, queue); }; source.addEventListener(name, listener); return { name, listener }; }); source.onmessage = (event) => { - const payload = parseRealtimeEvent(event.data); - if (payload) queue.push({ payload, eventName: "message" }); + enqueueRealtimeFrame(event.data, "message", options, queue); }; return { close() { @@ -244,12 +249,21 @@ function appendNumberParam(params: URLSearchParams, key: string, value: number | params.set(key, String(Math.trunc(parsed))); } -function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null { +function enqueueRealtimeFrame(raw: unknown, eventName: string, options: WorkbenchEventStreamOptions, queue: ReturnType>): void { + const rawText = typeof raw === "string" ? raw : String(raw ?? ""); + const decoded = decodeRealtimeEvent(rawText); + options.onIngress?.({ eventName, rawText, decodeStatus: decoded.status }); + if (decoded.payload) queue.push({ payload: decoded.payload, eventName }); +} + +function decodeRealtimeEvent(raw: string): { payload: WorkbenchRealtimeEvent | null; status: WorkbenchSseIngressFrame["decodeStatus"] } { try { const value = JSON.parse(raw); - return value && typeof value === "object" ? value as WorkbenchRealtimeEvent : null; + return value && typeof value === "object" && !Array.isArray(value) + ? { payload: value as WorkbenchRealtimeEvent, status: "accepted" } + : { payload: null, status: "not-object" }; } catch { - return null; + return { payload: null, status: "invalid-json" }; } } diff --git a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.test.ts b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.test.ts new file mode 100644 index 00000000..6bf5a5de --- /dev/null +++ b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.test.ts @@ -0,0 +1,77 @@ +// @vitest-environment jsdom + +import { beforeEach, describe, expect, it } from "vitest"; +import { mount } from "@vue/test-utils"; +import { nextTick } from "vue"; + +import TraceTimeline from "./TraceTimeline.vue"; + +describe("TraceTimeline sequence authority", () => { + beforeEach(() => { + window.HWLAB_CLOUD_WEB_CONFIG = { + displayTime: { timeZone: "Asia/Shanghai", locale: "zh-CN", label: "北京时间" } + }; + window.localStorage.clear(); + globalThis.ResizeObserver = FakeResizeObserver as unknown as typeof ResizeObserver; + Object.defineProperty(HTMLElement.prototype, "scrollTo", { configurable: true, value: () => undefined }); + }); + + it("renders sourceSeq-only live Kafka rows while terminal and does not label them projected", async () => { + const wrapper = mount(TraceTimeline, { + props: { + autoExpanded: true, + trace: { + traceId: "trc_live_timeline", + eventSource: "hwlab-kafka-sse", + status: "completed", + eventCount: 2, + events: [ + { sourceEventId: "evt_live_answer", runId: "run_live", sourceSeq: 4, source: "agentrun.kafka", type: "assistant", label: "agentrun:assistant:message", message: "live answer" }, + { sourceEventId: "evt_live_terminal", runId: "run_live", sourceSeq: 5, source: "agentrun.kafka", type: "result", label: "agentrun:terminal:completed", status: "completed", terminal: true } + ] + } + } + }); + await nextTick(); + + const disclosure = wrapper.get(".trace-disclosure"); + expect(disclosure.attributes("data-event-count")).toBe("2"); + expect(Number(disclosure.attributes("data-readable-row-count"))).toBeGreaterThan(0); + const rows = wrapper.findAll('[data-testid="trace-render-row"]'); + expect(rows.length).toBeGreaterThan(0); + expect(rows[0]?.attributes("data-event-seq-authority")).toBe("source"); + expect(rows[0]?.attributes("data-event-seq")).toBe("4"); + expect(rows[0]?.attributes("data-projected-seq")).toBeUndefined(); + expect(wrapper.text()).toContain("live answer"); + expect(wrapper.text()).not.toContain("运行已完成,暂无可读 Trace"); + wrapper.unmount(); + }); + + it("keeps durable Trace rows on projected sequence authority", async () => { + const wrapper = mount(TraceTimeline, { + props: { + autoExpanded: true, + trace: { + traceId: "trc_projected_timeline", + eventSource: "trace-api", + status: "running", + eventCount: 1, + events: [{ projectedSeq: 8, type: "assistant", label: "agentrun:assistant:message", message: "projected answer" }] + } + } + }); + await nextTick(); + + const row = wrapper.get('[data-testid="trace-render-row"]'); + expect(row.attributes("data-event-seq-authority")).toBe("projected"); + expect(row.attributes("data-event-seq")).toBe("8"); + expect(row.attributes("data-projected-seq")).toBe("8"); + wrapper.unmount(); + }); +}); + +class FakeResizeObserver { + observe(): void {} + unobserve(): void {} + disconnect(): void {} +} diff --git a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue index d2b6ba80..75d661a9 100644 --- a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue +++ b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue @@ -51,7 +51,7 @@ watch(() => [props.trace?.traceId, props.storageKey, props.defaultExpanded, prop }, { immediate: true }); function rowKey(row: TraceEventRow, index: number): string { - return `${row.rowId}-${row.seq ?? "no-seq"}-${index}`; + return `${row.rowId}-${row.sequenceAuthority}-${row.seq ?? "no-seq"}-${index}`; } function onDisclosureToggle(event: Event): void { @@ -144,7 +144,7 @@ function firstNonEmptyString(...values: unknown[]): string | null {

显示最近 {{ renderedRows.length }} / {{ readableRows.length }} 行

    -
  1. +
  2. {{ row.header }} diff --git a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts index cf36cb9e..cb822efc 100644 --- a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts +++ b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.test.ts @@ -1,6 +1,6 @@ // @vitest-environment jsdom -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { mount } from "@vue/test-utils"; import { nextTick } from "vue"; @@ -19,23 +19,46 @@ describe("WorkbenchKafkaDebugPanel", () => { }); it("replays only after an explicit click, completes on terminal, and clears locally", async () => { - const wrapper = mount(WorkbenchKafkaDebugPanel, { props: { traceId: "trc_component_debug" } }); + const wrapper = mount(WorkbenchKafkaDebugPanel, { + props: { + traceId: "trc_component_debug", + replayId: "rpl_component_debug", + outputBarrier: { partition: 0, firstOffset: "10", lastOffset: "11" } + } + }); expect(FakeEventSource.instances).toHaveLength(0); await wrapper.get('[data-testid="workbench-kafka-debug-replay"]').trigger("click"); - const source = FakeEventSource.instances[0]; + const source = FakeEventSource.instances[0]!; expect(source.url).toContain("stream=hwlab-debug"); expect(source.url).toContain("fromBeginning=true"); expect(source.url).toContain("traceId=trc_component_debug"); + expect(source.url).toContain("replayId=rpl_component_debug"); + expect(source.url).toContain("partition=0"); + expect(source.url).toContain("firstOffset=10"); + expect(source.url).toContain("lastOffset=11"); source.emit("hwlab.kafka.connected", { consumerReady: true, debugIsolation: true, deliverySemantics: "debug-replay", liveOnly: false, replay: true, topic: "hwlab.event.debug.v1", groupId: "debug-group-unique" }); source.emit("hwlab.kafka.event", debugRecord("assistant", { type: "assistant", assistantText: "debug answer", status: "running" })); source.emit("hwlab.kafka.event", debugRecord("terminal", { type: "result", eventType: "terminal", terminal: true, status: "completed" })); - source.emit("hwlab.kafka.replay-complete", { reason: "terminal", terminalObserved: true, count: 2 }); + source.emit("hwlab.kafka.replay-complete", { + reason: "terminal", + terminalObserved: true, + count: 2, + result: { + contractVersion: "workbench-kafka-debug-replay-v2", + replayId: "rpl_component_debug", + code: "terminal_complete", + terminalObserved: true, + counts: { scanned: 2, matched: 2, delivered: 2, clientReceived: null, decoded: null, applied: null } + } + }); await nextTick(); expect(wrapper.get('[data-testid="workbench-kafka-debug-panel"]').attributes("data-replay-status")).toBe("completed"); expect(wrapper.get('[data-testid="workbench-kafka-debug-counts"]').text()).toContain("2/2"); + expect(wrapper.get('[data-testid="workbench-kafka-debug-layer-counts"]').text()).toContain("scanned=2"); + expect(wrapper.get('[data-testid="workbench-kafka-debug-layer-counts"]').text()).toContain("received=2 · decoded=2 · applied=2"); expect(wrapper.get(".message-card").attributes("data-status")).toBe("completed"); expect(wrapper.get(".message-card").text()).toContain("debug answer"); expect(source.closed).toBe(true); @@ -45,7 +68,7 @@ describe("WorkbenchKafkaDebugPanel", () => { expect(wrapper.get('[data-testid="workbench-kafka-debug-counts"]').text()).toContain("0/0"); await wrapper.get('[data-testid="workbench-kafka-debug-replay"]').trigger("click"); - const incompleteSource = FakeEventSource.instances[1]; + const incompleteSource = FakeEventSource.instances[1]!; incompleteSource.emit("hwlab.kafka.connected", { consumerReady: true, debugIsolation: true, deliverySemantics: "debug-replay", liveOnly: false, replay: true }); incompleteSource.emit("hwlab.kafka.replay-complete", { reason: "timeout", terminalObserved: false, count: 0 }); await nextTick(); @@ -53,6 +76,42 @@ describe("WorkbenchKafkaDebugPanel", () => { expect(wrapper.text()).toContain("重放未观测到 terminal(timeout)"); expect(incompleteSource.closed).toBe(true); }); + + it("shows the product raw HWLAB Event buffer without opening another EventSource", async () => { + const firstRaw = '{"schema":"hwlab.event.v1","eventId":"evt_raw_1"}'; + const invalidRaw = "{invalid-json"; + const clearRawIngress = vi.fn(); + const wrapper = mount(WorkbenchKafkaDebugPanel, { + props: { + traceId: "trc_component_raw", + replayEnabled: true, + rawEnabled: true, + onClearRawIngress: clearRawIngress, + rawIngress: { + scopeKey: "ses_component_raw", + receivedCount: 2, + decodedCount: 1, + rejectedCount: 1, + evictedCount: 0, + retainedBytes: firstRaw.length + invalidRaw.length, + rows: [ + { arrivalSeq: 1, receivedAt: "2026-07-10T00:00:00.000Z", eventName: "hwlab.event.v1", rawText: firstRaw, decodeStatus: "accepted", byteLength: firstRaw.length }, + { arrivalSeq: 2, receivedAt: "2026-07-10T00:00:01.000Z", eventName: "hwlab.event.v1", rawText: invalidRaw, decodeStatus: "invalid-json", byteLength: invalidRaw.length } + ] + } + } + }); + + expect(FakeEventSource.instances).toHaveLength(0); + await wrapper.get('[data-testid="workbench-kafka-debug-tab-raw"]').trigger("click"); + expect(FakeEventSource.instances).toHaveLength(0); + expect(wrapper.get('[data-testid="workbench-raw-hwlab-counts"]').text()).toContain("received=2"); + expect((wrapper.get('[data-testid="workbench-raw-hwlab-text"]').element as HTMLTextAreaElement).value).toBe(`${firstRaw}\n${invalidRaw}`); + + await wrapper.get('[data-testid="workbench-raw-hwlab-clear"]').trigger("click"); + expect(clearRawIngress).toHaveBeenCalledTimes(1); + expect(FakeEventSource.instances).toHaveLength(0); + }); }); class FakeEventSource { diff --git a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.vue b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.vue index 3dd7ba20..1a16abca 100644 --- a/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.vue +++ b/web/hwlab-cloud-web/src/components/workbench/WorkbenchKafkaDebugPanel.vue @@ -2,31 +2,66 @@ // Responsibility: admin-only, local-state view of the YAML-gated hwlab.event.debug.v1 SSE stream. import { computed, onBeforeUnmount, ref, watch } from "vue"; -import { connectWorkbenchKafkaSseDebug, type WorkbenchKafkaSseDebugEvent, type WorkbenchKafkaSseDebugStream } from "@/api/workbench-debug"; +import { connectWorkbenchKafkaSseDebug, type WorkbenchKafkaDebugOffsetBarrier, type WorkbenchKafkaDebugReplayCounts, type WorkbenchKafkaDebugReplayResult, type WorkbenchKafkaSseDebugEvent, type WorkbenchKafkaSseDebugStream } from "@/api/workbench-debug"; import StatusBadge from "@/components/common/StatusBadge.vue"; import { applyWorkbenchIsolatedKafkaDebugEvent, createWorkbenchIsolatedKafkaDebugState } from "@/stores/workbench-isolated-kafka-debug"; +import { createRawHwlabIngressState, type RawHwlabIngressState } from "@/stores/workbench-raw-hwlab-ingress"; import WorkbenchMessageCard from "./WorkbenchMessageCard.vue"; -const props = defineProps<{ traceId: string | null }>(); +const props = withDefaults(defineProps<{ + traceId: string | null; + replayEnabled?: boolean; + rawEnabled?: boolean; + rawIngress?: RawHwlabIngressState; + onClearRawIngress?: () => void; + replayId?: string | null; + outputBarrier?: WorkbenchKafkaDebugOffsetBarrier | null; +}>(), { + replayEnabled: true, + rawEnabled: false, + rawIngress: () => createRawHwlabIngressState(), + onClearRawIngress: () => undefined, + replayId: null, + outputBarrier: null +}); +type InspectorTab = "trace" | "raw"; const state = ref(createWorkbenchIsolatedKafkaDebugState()); const connection = ref(null); const status = ref<"idle" | "connecting" | "replaying" | "completed" | "incomplete" | "error" | "closed">("idle"); const transportError = ref(null); +const activeTab = ref(props.replayEnabled ? "trace" : "raw"); +const clientReceived = ref(0); +const clientDecoded = ref(0); +const replayResult = ref(null); let stream: WorkbenchKafkaSseDebugStream | null = null; const badgeStatus = computed(() => status.value === "completed" ? "completed" : ["incomplete", "error"].includes(status.value) ? "failed" : "running"); const statusLabel = computed(() => status.value === "completed" ? "重放完成" : status.value === "incomplete" ? "重放不完整" : status.value === "replaying" ? "重放中" : status.value === "idle" ? "等待重放" : `debug ${status.value}`); const visibleError = computed(() => transportError.value ?? state.value.error); const replayBusy = computed(() => ["connecting", "replaying"].includes(status.value)); +const rawText = computed(() => props.rawIngress.rows.map((row) => row.rawText).join("\n")); +const replayIdLabel = computed(() => replayResult.value?.replayId ?? connection.value?.replayId ?? props.replayId ?? "-"); +const outputBarrierLabel = computed(() => props.outputBarrier + ? `${props.outputBarrier.partition}:${props.outputBarrier.firstOffset}-${props.outputBarrier.lastOffset}` + : "-"); +const serverCounts = computed(() => replayResult.value?.counts ?? {}); watch(() => props.traceId, resetForTrace); +watch(() => [props.replayEnabled, props.rawEnabled] as const, ([replayEnabled, rawEnabled]) => { + if (activeTab.value === "trace" && !replayEnabled && rawEnabled) activeTab.value = "raw"; + if (activeTab.value === "raw" && !rawEnabled && replayEnabled) activeTab.value = "trace"; +}, { immediate: true }); onBeforeUnmount(closeStream); function replayCurrentTrace(): void { + if (!props.replayEnabled) return; closeStream(false); state.value = createWorkbenchIsolatedKafkaDebugState(); connection.value = null; transportError.value = null; + clientReceived.value = 0; + clientDecoded.value = 0; + replayResult.value = null; const traceId = cleanTraceId(props.traceId); if (!traceId) { status.value = "idle"; @@ -37,6 +72,13 @@ function replayCurrentTrace(): void { stream: "hwlab-debug", traceId, fromBeginning: true, + replayId: props.replayId, + outputBarrier: props.outputBarrier, + onIngress: (frame) => { + if (frame.eventName !== "hwlab.kafka.event") return; + clientReceived.value += 1; + if (frame.decodeStatus === "accepted") clientDecoded.value += 1; + }, onOpen: () => { status.value = "replaying"; }, onError: () => { if (["completed", "incomplete"].includes(status.value)) return; @@ -62,9 +104,16 @@ function replayCurrentTrace(): void { if (eventName === "hwlab.kafka.replay-complete") { stream?.close(); stream = null; - const complete = event.reason === "terminal" && event.terminalObserved === true; + replayResult.value = event.result ?? null; + const complete = event.result + ? event.result.code === "terminal_complete" && event.result.terminalObserved === true + : event.reason === "terminal" && event.terminalObserved === true; status.value = complete ? "completed" : "incomplete"; - transportError.value = complete ? null : `重放未观测到 terminal(${event.reason || "unknown"})`; + transportError.value = complete + ? null + : event.result?.code + ? `${event.result.code}: ${event.result.message || "隔离重放未完成"}` + : `重放未观测到 terminal(${event.reason || "unknown"})`; return; } if (eventName !== "hwlab.kafka.event") return; @@ -88,6 +137,9 @@ function clearPanel(): void { state.value = createWorkbenchIsolatedKafkaDebugState(); connection.value = null; transportError.value = null; + clientReceived.value = 0; + clientDecoded.value = 0; + replayResult.value = null; status.value = "idle"; } @@ -106,25 +158,44 @@ function cleanTraceId(value: string | null): string | null {
    Kafka 卡片隔离调试 - 只读独立 debug topic;不写产品会话 + 隔离重放与产品 SSE 原始事件共用一个只读调试卡片
    - - - + + {{ rawIngress.receivedCount }} frames + + +
    -
    + + +
    traceId {{ traceId || "等待当前 turn" }} topic {{ connection?.topic || "等待 consumer-ready" }} group {{ connection?.groupId || "-" }} + replayId {{ replayIdLabel }} + barrier {{ outputBarrierLabel }} delivery debug-replay · fromBeginning=true
    +
    + source 现有产品 EventSource + scope {{ rawIngress.scopeKey || "等待产品 SSE scope" }} + delivery hwlab.event.v1 · memory-only · no client event filter +
    -

    {{ visibleError }}

    +

    {{ visibleError }}

    -
    +
    + server scanned={{ serverCounts.scanned ?? "-" }} · matched={{ serverCounts.matched ?? "-" }} · delivered={{ serverCounts.delivered ?? "-" }} + client local overlay received={{ clientReceived }} · decoded={{ clientDecoded }} · applied={{ state.appliedCount }} +
    + +
    尚未收到 debug event。
    +
    +
    + 产品 SSE 原始帧 + + received={{ rawIngress.receivedCount }} · retained={{ rawIngress.rows.length }} · decoded={{ rawIngress.decodedCount }} · rejected={{ rawIngress.rejectedCount }} · evicted={{ rawIngress.evictedCount }} · bytes={{ rawIngress.retainedBytes }} + +
    +