diff --git a/docs/MDTODO/pure-kafka-live-single-step-debug.md b/docs/MDTODO/pure-kafka-live-single-step-debug.md index d4f8d552..7988fd5d 100644 --- a/docs/MDTODO/pure-kafka-live-single-step-debug.md +++ b/docs/MDTODO/pure-kafka-live-single-step-debug.md @@ -30,4 +30,19 @@ 修复隔离 debug replay 的 producer/barrier 与分层结果合同：保留 source lineage、replayId、offset range，报告 scanned/matched/delivered/decoded/applied/terminal，区分 producer-not-invoked、trace missing、filter mismatch、decoder/reducer rejection 与 terminal missing；仅写 debug topic。上下文：[隔离根因 #2476](https://github.com/pikasTech/HWLAB/issues/2476#issuecomment-4935398125)、[主方案 #2474](https://github.com/pikasTech/HWLAB/issues/2474#issuecomment-4935408473)。依赖：无。验证：empty、matched-no-terminal、reducer-rejected、terminal-complete 四类 typed 结果及 semantic internal web-probe command，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.3_Task_Report.md)。 ### R2.4 [in_progress] -集成前三个子任务，受控部署到 NC01/v03 并用原入口验证运行中及结束后 Trace 可读、同一 ingress raw/reducer 计数可解释、隔离 debug 命中 terminal；保持 OTel 三服务完整、error=0、projection=0 且产品链无补洞路径。上下文：[验收 #2474](https://github.com/pikasTech/HWLAB/issues/2474)、[OTel 基线 #2477](https://github.com/pikasTech/HWLAB/issues/2477#issuecomment-4935385934)。依赖：Trace 序列权威修复、单一 ingress raw 观察窗、隔离 debug replay 分层诊断。验证：定向测试、YAML/render/runtime contract、受控 CI/CD、semantic --origin internal web-probe 与 Kafka/SSE/DOM/OTel bounded counts，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.4_Task_Report.md)。 \ No newline at end of file +以 [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474) 和固定 trace trc_fb6031c28d79495e 完成纯 Kafka 集成闭环：先增强 CLI 直接读取持久 hwlab.event 并复用前端 decode、merge、row model、summary 输出同构 Markdown，再修复 Trace 内 final response 重复、tool-call summary 可见性和每个 HWLAB SSE ingress 刷新 lastEventAt；最后受控部署 NC01/v03，以 Kafka 重放、semantic internal web-probe、DOM 与 OTel 验收 projection=0 且无补洞路径，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R2.4_Task_Report.md)。 +## R3 + +在纯 Kafka SSE 权威链内实现页面刷新后的可重放引导，明确区别于 projector/read model/snapshot/sync/gap-fill/finalizer 补洞；承接上下文：[HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474)，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3_Task_Report.md)。 +### R3.1 + +深入分析 Kafka retention、SSE 刷新连接、offset barrier、live handoff 与去重身份，形成不引入第二事实源的方案和失败可见性，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.1_Task_Report.md)。 +### R3.2 + +实现 trace/session scoped Kafka refresh bootstrap 与 live fanout 无缝交接，全部开关、group、容量和时限保持 YAML-first，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.2_Task_Report.md)。 +### R3.3 + +优先使用既有 trace 通过 CLI 单步重放验证顺序、terminal、final 唯一性、tool summary 和 timing，不启动新真实运行，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.3_Task_Report.md)。 +### R3.4 + +完成定向测试、受控 CI/CD 与 semantic internal web-probe 页面刷新验收，确认 projection=0 且无 snapshot/sync/gap-fill/finalizer 路径，完成任务后将详细报告写入[任务报告](./details/pure-kafka-live-single-step-debug/R3.4_Task_Report.md)。 \ No newline at end of file diff --git a/docs/reference/kafka-source-direct-debug.md b/docs/reference/kafka-source-direct-debug.md index 8e19ae7c..4dbe81cf 100644 --- a/docs/reference/kafka-source-direct-debug.md +++ b/docs/reference/kafka-source-direct-debug.md @@ -1,5 +1,50 @@ # Kafka 源码直连调试 +- `hwlab-cli kafka render trace` 用于把持久 HWLAB Kafka 事件直接渲染为与 Web 同构的 Markdown Trace： + - 命令为 `hwlab-cli kafka render trace --from kafka --trace-id --format markdown`； + - 默认输入 topic 优先读取 owning YAML 注入的 `HWLAB_KAFKA_EVENT_TOPIC`； + - 未注入时使用 canonical `hwlab.event.v1`； + - Kafka 读取必须使用 `--group-prefix` 或 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX`； + - group prefix 必须标识独立 debug group，不复用产品 consumer group； + - 命令只读取 Kafka，不发布事件，不启动 Cloud API，不访问数据库，也不依赖 projector。 + +- Trace 渲染与 Web 共享最后一个展示分叉之前的生产管线： + - JSON frame 使用 `decodeWorkbenchRealtimeEventFrame` 解码； + - 事件使用 `reduceWorkbenchRealtimeEvent` 与 `planWorkbenchRealtimeApply` 分类； + - 卡片使用 `projectWorkbenchLiveKafkaMessage` 增量归约； + - Trace 行使用 `traceDisplayRows` 生成同一 row model； + - Web 最后渲染 HTML，CLI 最后使用 `renderTraceRowsMarkdown` 渲染 Markdown； + - Final Response 只放在 Trace 外层，Trace 内保留非最终的助手进展消息。 + +- Kafka Trace 查询必须披露有界扫描事实： + - `--run-id` 与 `--command-id` 用于在复用或异常混合的 traceId 内精确隔离一轮； + - `--command-id` 先按 trace、session 与 run 读取基础流，再在 CLI 内形成 command 生命周期； + - 只有基础流中至少存在一条精确 commandId 事件，才允许吸附同一 run 的无 commandId final/terminal； + - 精确 commandId 不存在时必须返回 `source_command_missing`，不得由 commandless 生命周期冒充成功； + - 精确 command 事件保留； + - 同一 run 中缺少 commandId 的 final/terminal 生命周期事件保留； + - 其他 command 和无法归属的 commandless 事件排除； + - 禁止把 commandId 直接下推后丢失 run-scoped terminal/final； + - identity 输出必须同时披露观测到的 session、run、command 和 source event 去重计数； + - command scope 输出必须披露每个 commandId 的计数、缺失 commandId 数、可归属生命周期数、歧义数与排除数； + - command 生命周期只有在 `scanComplete=true` 时才可判定完整； + - `limit` 或 `timeout` 只能作为部分流证据，`commandLifecyclePreserved` 必须保持 false； + - 输出 `scannedCount`、`parsedCount`、`matchedCount`、`invalidJsonCount` 与 `filterRejectedCount`； + - 输出 topic end offsets、最后扫描 offsets 与 `completionReason`； + - `completionReason=end-offset` 表示已经扫描到命令启动时捕获的 Kafka 上界； + - `completionReason=limit` 表示达到显式事件上限，不能冒充完整 trace； + - `completionReason=timeout` 表示未在 YAML 或 CLI 时限内证明到达上界； + - 没有匹配事件时以 `source_trace_missing` 失败，并保留上述扫描计数。 + +- Trace Markdown artifact 默认写入 `.state/hwlab-cli/kafka-render/`： + - `--output-markdown` 可显式选择路径； + - `--from jsonl --jsonl-file ` 可完全离线复用同一管线； + - 默认文本输出保持 compact； + - 默认 JSON 只返回 `--row-limit` 指定的尾部行，并披露 returned/omitted/window； + - `--full` 仅用于显式下钻完整 row payload； + - `--format markdown` 输出 Markdown 正文； + - `--json` 输出身份、投影、row model、终态、计数和 artifact SHA 的结构化证据。 + - `hwlab-cli kafka regenerate hwlab` 用于把指定 AgentRun session 的事件直接映射为 HWLAB debug 事件： - 调用生产路径同源的 AgentRun 解码与 HWLAB 事件映射函数； - 不启动 Cloud API； diff --git a/internal/cloud/kafka-event-bridge.test.ts b/internal/cloud/kafka-event-bridge.test.ts index 9634eee8..dab70c94 100644 --- a/internal/cloud/kafka-event-bridge.test.ts +++ b/internal/cloud/kafka-event-bridge.test.ts @@ -117,6 +117,50 @@ test("projects final AgentRun assistant text without sealing before terminal_sta assert.equal(projected.event.finalResponse.traceId, "trc_final"); }); +test("projects structured AgentRun commandExecution visibility into HWLAB tool events", () => { + const projected = projectAgentRunKafkaEventToHwlabEvent({ + schema: "agentrun.event.v1", + eventType: "agentrun.run.event", + run: { runId: "run_tool", sessionId: "ses_tool" }, + command: { commandId: "cmd_tool" }, + event: { + seq: 10, + type: "tool_call", + payload: { + traceId: "trc_tool", + type: "commandExecution", + toolName: "commandExecution", + method: "item/completed", + itemId: "tool_rg", + status: "completed", + command: "/bin/bash -lc 'cd /workspace && rg -n target src'", + cwd: "/workspace", + exitCode: 0, + durationMs: 1234, + stdoutSummary: "src/main.ts:42:target", + stderrSummary: "warning summary", + outputSummary: "1 match", + summary: { + outputBytes: 24, + outputTruncated: true + } + } + } + }, { source: "hwlab-test" }); + + assert.equal(projected.event.toolName, "commandExecution"); + assert.equal(projected.event.toolType, "commandExecution"); + assert.equal(projected.event.command, "/bin/bash -lc 'cd /workspace && rg -n target src'"); + assert.equal(projected.event.cwd, "/workspace"); + assert.equal(projected.event.exitCode, 0); + assert.equal(projected.event.durationMs, 1234); + assert.equal(projected.event.stdoutSummary, "src/main.ts:42:target"); + assert.equal(projected.event.stderrSummary, "warning summary"); + assert.equal(projected.event.outputSummary, "1 match"); + assert.equal(projected.event.outputBytes, 24); + assert.equal(projected.event.outputTruncated, true); +}); + test("projects AgentRun terminal_status Kafka event into terminal HWLAB event", () => { const projected = projectAgentRunKafkaEventToHwlabEvent({ schema: "agentrun.event.v1", diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index 58cb861a..af6147f4 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -13,7 +13,7 @@ import { const TRUE_VALUES = new Set(["1", "true", "yes", "on"]); export const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1"; -const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1"; +export const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1"; const DEFAULT_STDIO_TOPIC = "codex-stdio.raw.v1"; export const DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC = "agentrun.event.debug.v1"; export const DEFAULT_HWLAB_DEBUG_EVENT_TOPIC = "hwlab.event.debug.v1"; @@ -933,61 +933,166 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env); const maxEvents = Math.max(1, integerValue(limit) || DEFAULT_QUERY_LIMIT); const budgetMs = Math.max(250, integerValue(timeoutMs) || DEFAULT_QUERY_TIMEOUT_MS); + const deadlineMs = Date.now() + budgetMs; const kafka = kafkaFactory({ brokers, clientId: `${clientId}-query` }); + const endOffsetSnapshot = await kafkaTopicEndOffsetSnapshot(kafka, resolvedTopic, deadlineMs); + const endOffsetByPartition = new Map(endOffsetSnapshot.offsets.map((entry) => [entry.partition, entry.endOffset])); + const reachedEndPartitions = new Set(endOffsetSnapshot.offsets.filter(kafkaOffsetPartitionIsEmpty).map((entry) => entry.partition)); const resolvedGroupIdPrefix = stringValue(groupIdPrefix) || `${clientId}-query`; const groupId = `${resolvedGroupIdPrefix}-${Date.now()}-${randomUUID().slice(0, 8)}`; const consumer = kafka.consumer({ groupId, allowAutoTopicCreation: false }); const events = []; + const lastScannedOffsetByPartition = new Map(); + let scannedCount = 0; + let parsedCount = 0; + let invalidJsonCount = 0; + let filterRejectedCount = 0; + let completionReason = "timeout"; let timer = null; let running = false; - await consumer.connect(); - await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false }); try { + if (remainingKafkaQueryBudget(deadlineMs) <= 0) return kafkaQueryResult(); + await withinKafkaQueryBudget(consumer.connect(), deadlineMs, "consumer-connect"); + await withinKafkaQueryBudget(consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning !== false }), deadlineMs, "consumer-subscribe"); + const alreadyAtBoundedEnd = fromBeginning !== false && endOffsetSnapshot.available && reachedEndPartitions.size === endOffsetByPartition.size; + if (alreadyAtBoundedEnd) completionReason = "end-offset"; + if (alreadyAtBoundedEnd) return kafkaQueryResult(); await new Promise((resolve, reject) => { let finished = false; - const finish = (value = null) => { + const finish = (reason = "timeout") => { if (finished) return; finished = true; - resolve(value); + completionReason = reason; + resolve(reason); }; - timer = setTimeout(finish, budgetMs); + timer = setTimeout(() => finish("timeout"), Math.max(1, remainingKafkaQueryBudget(deadlineMs))); timer.unref?.(); running = true; consumer.run({ eachMessage: async ({ topic: messageTopic, partition, message }) => { + scannedCount += 1; + lastScannedOffsetByPartition.set(partition, stringValue(message.offset)); const valueText = message.value ? Buffer.from(message.value).toString("utf8") : ""; const value = parseJson(valueText); - if (!value || !eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) return; - events.push({ - topic: messageTopic, - partition, - offset: message.offset, - key: message.key ? Buffer.from(message.key).toString("utf8") : null, - timestamp: message.timestamp ?? null, - valueSha256: sha256(valueText), - value - }); - if (events.length >= maxEvents) finish(); + if (!value || typeof value !== "object" || Array.isArray(value)) invalidJsonCount += 1; + else { + parsedCount += 1; + if (!eventMatchesFilters(value, { traceId, sessionId, runId, commandId })) filterRejectedCount += 1; + else { + events.push({ + topic: messageTopic, + partition, + offset: message.offset, + key: message.key ? Buffer.from(message.key).toString("utf8") : null, + timestamp: message.timestamp ?? null, + valueSha256: sha256(valueText), + value + }); + } + } + if (events.length >= maxEvents) return finish("limit"); + const endOffset = endOffsetByPartition.get(partition); + if (fromBeginning !== false && endOffset && kafkaOffsetReached(message.offset, endOffset)) reachedEndPartitions.add(partition); + if (endOffsetSnapshot.available && reachedEndPartitions.size === endOffsetByPartition.size) finish("end-offset"); } }).catch(reject); }); } finally { if (timer) clearTimeout(timer); - if (running) await consumer.stop().catch(() => undefined); - await boundedDisconnect(consumer); + if (running) await boundedKafkaPromise(consumer.stop(), remainingKafkaQueryCleanupBudget(deadlineMs)).catch(() => undefined); + await boundedDisconnect(consumer, remainingKafkaQueryCleanupBudget(deadlineMs)); + } + return kafkaQueryResult(); + + function kafkaQueryResult() { + return { + ok: true, + stream, + topic: resolvedTopic, + groupId, + count: events.length, + scannedCount, + parsedCount, + matchedCount: events.length, + invalidJsonCount, + filterRejectedCount, + completionReason, + reachedEndOffsets: completionReason === "end-offset", + endOffsets: endOffsetSnapshot.offsets, + endOffsetsAvailable: endOffsetSnapshot.available, + endOffsetsError: endOffsetSnapshot.error, + lastScannedOffsets: [...lastScannedOffsetByPartition.entries()].map(([partition, offset]) => ({ partition, offset })), + limit: maxEvents, + timeoutMs: budgetMs, + filters: compactObject({ traceId, sessionId, runId, commandId }), + events, + valuesPrinted: false + }; + } +} + +async function kafkaTopicEndOffsetSnapshot(kafka, topic, deadlineMs) { + if (typeof kafka?.admin !== "function") return { available: false, offsets: [], error: { code: "kafka_admin_unavailable", valuesRedacted: true } }; + const admin = kafka.admin(); + try { + await withinKafkaQueryBudget(admin.connect(), deadlineMs, "admin-connect"); + const fetched = await withinKafkaQueryBudget(admin.fetchTopicOffsets(topic), deadlineMs, "fetch-topic-offsets"); + const offsets = Array.isArray(fetched) + ? fetched.map((entry) => ({ + partition: integerValue(entry.partition), + startOffset: kafkaOffsetValue(entry.low), + endOffset: kafkaOffsetValue(entry.high ?? entry.offset) + })).filter((entry) => entry.partition !== null && entry.endOffset !== null) + : []; + if (offsets.length === 0) return { available: false, offsets: [], error: { code: "kafka_end_offsets_empty", valuesRedacted: true } }; + return { + available: true, + offsets, + error: null + }; + } catch (error) { + return { available: false, offsets: [], error: { code: error?.code ?? "kafka_end_offsets_failed", message: errorMessage(error), valuesRedacted: true } }; + } finally { + await boundedDisconnect(admin, remainingKafkaQueryCleanupBudget(deadlineMs)); + } +} + +function kafkaOffsetPartitionIsEmpty(entry) { + if (entry.startOffset !== null) return entry.startOffset === entry.endOffset; + return entry.endOffset === "0"; +} + +function remainingKafkaQueryBudget(deadlineMs) { + return Math.max(0, deadlineMs - Date.now()); +} + +function remainingKafkaQueryCleanupBudget(deadlineMs) { + return Math.max(1, remainingKafkaQueryBudget(deadlineMs)); +} + +async function withinKafkaQueryBudget(promise, deadlineMs, stage) { + const remainingMs = remainingKafkaQueryBudget(deadlineMs); + if (remainingMs <= 0) throw contractError("kafka_query_timeout", `Kafka query timed out before ${stage}.`); + let timer = null; + try { + return await Promise.race([ + promise, + new Promise((_, reject) => { + timer = setTimeout(() => reject(contractError("kafka_query_timeout", `Kafka query timed out during ${stage}.`)), remainingMs); + timer.unref?.(); + }) + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +function kafkaOffsetReached(offset, endOffset) { + try { + return BigInt(String(offset)) + 1n >= BigInt(String(endOffset)); + } catch { + return false; } - return { - ok: true, - stream, - topic: resolvedTopic, - groupId, - count: events.length, - limit: maxEvents, - timeoutMs: budgetMs, - filters: compactObject({ traceId, sessionId, runId, commandId }), - events, - valuesPrinted: false - }; } export async function openKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, groupIdPrefix = null, traceId = null, sessionId = null, runId = null, commandId = null, replayId = null, offsetRange = null, fromBeginning = false, onEvent, onRecord = null, onBarrierReady = null, onBarrierComplete = null, onError = null, kafkaFactory = defaultKafkaFactory } = {}) { @@ -1122,8 +1227,13 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, assistantText, finalResponse: authoritative ? { text: assistantText, status: "completed", traceId, source: "agentrun.kafka", valuesPrinted: false } : null, itemId: firstText(payload.itemId), + assistantSource: firstText(payload.source), + messageIndex: integerValue(payload.messageIndex), + messageCount: integerValue(payload.messageCount), replyAuthority: payload.replyAuthority === true, final: payload.final === true, + textTruncated: payload.textTruncated === true, + outputTruncated: payload.outputTruncated === true, terminal: false }; } @@ -1146,11 +1256,20 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run, label: `agentrun:tool:${toolName}`, message: textPayload(payload, `AgentRun tool ${toolName}`), toolName, + toolType: firstText(payload.type), method: firstText(payload.method), itemId: firstText(payload.itemId), + command: firstText(payload.command, payload.commandLine), + cwd: firstText(payload.cwd), + processId: firstText(payload.processId), exitCode: integerValue(payload.exitCode), durationMs: integerValue(payload.durationMs), outputSummary: firstText(payload.outputSummary, payload.summary?.text), + stdoutSummary: firstText(payload.stdoutSummary), + stderrSummary: firstText(payload.stderrSummary), + commandTruncated: payload.commandTruncated === true, + outputBytes: integerValue(payload.outputBytes ?? payload.summary?.outputBytes), + outputTruncated: payload.outputTruncated === true || payload.summary?.outputTruncated === true, terminal: false }; } @@ -1244,12 +1363,16 @@ function defaultKafkaFactory(config) { } async function boundedDisconnect(consumer, timeoutMs = 2000) { + return boundedKafkaPromise(consumer.disconnect(), timeoutMs); +} + +async function boundedKafkaPromise(promise, timeoutMs) { let timer = null; try { await Promise.race([ - consumer.disconnect(), + promise, new Promise((resolve) => { - timer = setTimeout(resolve, timeoutMs); + timer = setTimeout(resolve, Math.max(1, timeoutMs)); timer.unref?.(); }) ]); diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index a07db059..ed6473fe 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -7,6 +7,7 @@ import { test } from "bun:test"; import { queryKafkaEventStream } from "../../internal/cloud/kafka-event-bridge.ts"; import { mapAgentRunRecordsToHwlabDebugEvents, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts"; +import { traceDisplayRows } from "../src/hwlab-cli/trace-renderer.ts"; const SESSION_ID = "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105"; const HWLAB_SESSION_ID = "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105"; @@ -16,8 +17,294 @@ const RUN_ID = "run_a78cbcb05f2c4afaa748a3158db44998"; test("Kafka help shows canonical default input and explicit reconstruction debug input", async () => { const result = await runKafkaCli(["help", "--json"], { env: {}, now: () => "2026-07-10T12:00:00.000Z" }); assert.equal(result.exitCode, 0); - assert.match(result.payload.commands[0], /--input-topic agentrun\.event\.v1/u); - assert.match(result.payload.commands[1], /--input-topic agentrun\.event\.debug\.v1/u); + assert.ok(result.payload.commands.some((command: string) => /regenerate hwlab.*--input-topic agentrun\.event\.v1/u.test(command))); + assert.ok(result.payload.commands.some((command: string) => /regenerate hwlab.*--input-topic agentrun\.event\.debug\.v1/u.test(command))); + assert.ok(result.payload.commands.some((command: string) => /render trace.*--input-topic hwlab\.event\.v1/u.test(command))); +}); + +test("offline HWLAB JSONL uses the shared Web pipeline and renders final response only outside Trace", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-render-")); + const inputFile = path.join(cwd, "hwlab-events.jsonl"); + const outputFile = path.join(cwd, "trace.md"); + const finalText = "Hi. What do you want to work on?"; + const records = [ + hwlabRecord(1, { type: "tool", eventType: "tool", status: "started", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_rg", command: "/bin/bash -lc 'cd /workspace && rg -n target src'" }), + hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_rg", command: "/bin/bash -lc 'cd /workspace && rg -n target src'", durationMs: 1234, exitCode: 0, stdoutSummary: "src/main.ts:42:target" }), + hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_progress", message: "I am checking the workspace.", assistantText: "I am checking the workspace.", assistantSource: "agent-message" }), + hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message" }), + hwlabRecord(5, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message-final", replyAuthority: true, final: true, finalResponse: { text: finalText } }), + hwlabRecord(6, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true, message: "AgentRun completed" }) + ]; + await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + + const result = await runKafkaCli([ + "render", "trace", + "--from", "jsonl", + "--trace-id", TRACE_ID, + "--session-id", HWLAB_SESSION_ID, + "--run-id", RUN_ID, + "--command-id", "cmd_render_trace", + "--jsonl-file", inputFile, + "--output-markdown", outputFile, + "--format", "markdown" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + + assert.equal(result.exitCode, 0); + assert.equal(result.payload.input.scannedCount, 6); + assert.equal(result.payload.input.matchedCount, 6); + assert.equal(result.payload.projection.decodedCount, 6); + assert.equal(result.payload.projection.appliedCount, 6); + assert.equal(result.payload.projection.terminalObserved, true); + assert.equal(result.payload.projection.lastEventAt, "2026-07-10T12:00:00.005Z"); + assert.deepEqual(result.payload.identity.observedRunIds, [RUN_ID]); + assert.deepEqual(result.payload.identity.observedCommandIds, ["cmd_render_trace"]); + assert.equal(result.payload.identity.sourceEventIdentityDuplicates, 0); + assert.equal(result.payload.render.toolRowCount, 1); + assert.equal(result.payload.render.assistantRowCount, 1); + assert.equal(result.payload.render.finalResponsePresent, true); + assert.equal(result.payload.render.finalResponseInTrace, false); + assert.equal(result.payload.render.rowsReturned, result.payload.render.rowCount); + assert.equal(result.payload.render.rowsOmitted, 0); + assert.equal(result.payload.validation.allMatchedEventsApplied, true); + assert.equal(result.payload.validation.finalResponseOutsideTrace, true); + const toolRow = result.payload.render.rows.find((row: any) => row.rowId.startsWith("tool:")); + assert.match(toolRow.header, /ok rg duration=00:00:01 exit=0/u); + assert.match(toolRow.preview, /rg -n target src/u); + assert.match(toolRow.preview, /src\/main\.ts:42:target/u); + const markdown = await readFile(outputFile, "utf8"); + assert.equal((markdown.match(new RegExp(finalText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1); + assert.match(markdown, /## 运行记录/u); + assert.match(markdown, /## 最终回复/u); + assert.equal(result.markdownOutput?.trim(), markdown.trim()); +}); + +test("outer final response stays out of the Trace fallback when no completion event exists", () => { + const finalText = "Hi. What do you want to work on?"; + const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [ + { + sourceSeq: 1, + sourceEventId: "evt_same_text", + type: "assistant", + eventType: "assistant", + status: "running", + label: "agentrun:assistant:message", + itemId: "assistant_same_text", + message: finalText, + assistantText: finalText, + assistantSource: "completed-agent-message" + }, + { + sourceSeq: 2, + sourceEventId: "evt_final", + type: "assistant", + eventType: "assistant", + status: "running", + label: "agentrun:assistant:message", + itemId: "assistant_final", + message: finalText, + assistantText: finalText, + assistantSource: "completed-agent-message-final", + finalResponse: { text: finalText }, + final: true, + replyAuthority: true + } + ], { + finalResponsePlacement: "outer", + outerFinalResponseText: finalText + }); + + assert.deepEqual(rows, []); +}); + +test("trace JSON defaults to a bounded tail while the Markdown artifact stays complete", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-window-")); + const inputFile = path.join(cwd, "hwlab-events.jsonl"); + const records = Array.from({ length: 25 }, (_, index) => hwlabRecord(index + 1, { + type: "assistant", + eventType: "assistant", + status: "running", + label: "agentrun:assistant:message", + itemId: `assistant_${index + 1}`, + message: `progress ${index + 1}`, + assistantText: `progress ${index + 1}` + })); + records.push(hwlabRecord(26, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true })); + await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8"); + + const result = await runKafkaCli([ + "render", "trace", + "--from", "jsonl", + "--trace-id", TRACE_ID, + "--jsonl-file", inputFile, + "--row-limit", "5", + "--format", "markdown" + ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" }); + + assert.equal(result.exitCode, 0); + assert.equal(result.payload.render.rowCount, 25); + assert.equal(result.payload.render.rowsReturned, 5); + assert.equal(result.payload.render.rowsOmitted, 20); + assert.equal(result.payload.render.rowWindow, "tail"); + assert.match(result.markdownOutput, /progress 1/u); + assert.match(result.markdownOutput, /progress 24/u); +}); + +test("command-scoped trace keeps attributable commandless final and terminal lifecycle events", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-command-scope-")); + const finalText = "Hi. What do you want to work on?"; + const exactTool = hwlabRecord(1, { + type: "tool", + eventType: "tool", + status: "completed", + label: "agentrun:tool:commandExecution", + toolName: "commandExecution", + itemId: "tool_command_scope", + command: "rg -n target src", + exitCode: 0 + }); + const exactProgress = hwlabRecord(2, { + type: "assistant", + eventType: "assistant", + status: "running", + itemId: "assistant_progress", + assistantText: "Checking the workspace." + }); + const unscopedFinal = withoutCommandId(hwlabRecord(3, { + type: "assistant", + eventType: "assistant", + status: "running", + itemId: "assistant_final", + assistantText: finalText, + finalResponse: { text: finalText }, + final: true, + replyAuthority: true + })); + const unscopedTerminal = withoutCommandId(hwlabRecord(4, { + type: "result", + eventType: "terminal", + status: "completed", + terminal: true, + message: "AgentRun completed" + })); + const runnerShutdown = withCommandId(hwlabRecord(5, { + type: "backend", + eventType: "backend", + status: "running", + message: "runner shutdown" + }), "runner-shutdown"); + const unscopedBackend = withoutCommandId(hwlabRecord(6, { + type: "backend", + eventType: "backend", + status: "running", + message: "unscoped heartbeat" + })); + let readerInput: any = null; + const result = await runKafkaCli([ + "render", "trace", + "--from", "kafka", + "--trace-id", TRACE_ID, + "--session-id", HWLAB_SESSION_ID, + "--run-id", RUN_ID, + "--command-id", "cmd_render_trace", + "--group-prefix", "hwlab-v03-workbench-isolated-debug", + "--row-limit", "2", + "--format", "markdown" + ], { + cwd, + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka(input) { + readerInput = input; + return { + events: [exactTool, exactProgress, unscopedFinal, unscopedTerminal, runnerShutdown, unscopedBackend], + scannedCount: 6, + parsedCount: 6, + matchedCount: 6, + filterRejectedCount: 0, + completionReason: "end-offset", + reachedEndOffsets: true + }; + } + }); + + assert.equal(result.exitCode, 0); + assert.equal(readerInput.commandId, undefined, "commandId must not be pushed down before lifecycle attribution"); + assert.equal(readerInput.runId, RUN_ID); + assert.equal(result.payload.input.queryMatchedCount, 6); + assert.equal(result.payload.input.matchedCount, 4); + assert.deepEqual(result.payload.input.commandScope.commandIdCounts, [ + { commandId: "cmd_render_trace", count: 2 }, + { commandId: "runner-shutdown", count: 1 } + ]); + assert.equal(result.payload.input.commandScope.missingCommandIdCount, 3); + assert.equal(result.payload.input.commandScope.unscopedLifecycleCount, 2); + assert.equal(result.payload.input.commandScope.selectedUnscopedLifecycleCount, 2); + assert.equal(result.payload.input.commandScope.unscopedNonLifecycleCount, 1); + assert.equal(result.payload.input.commandScope.otherCommandRecordCount, 1); + assert.equal(result.payload.projection.eventCount, 4); + assert.equal(result.payload.projection.terminalObserved, true); + assert.equal(result.payload.render.finalResponseInTrace, false); + assert.equal(result.payload.validation.commandLifecyclePreserved, true); + assert.equal((result.markdownOutput?.match(/Hi\. What do you want to work on\?/gu) ?? []).length, 1); +}); + +test("missing commandId cannot be satisfied by commandless lifecycle events from the same run", async () => { + const otherCommand = withCommandId(hwlabRecord(1, { + type: "tool", + eventType: "tool", + status: "completed", + label: "agentrun:tool:commandExecution", + itemId: "tool_other_command", + command: "rg -n target src", + exitCode: 0 + }), "cmd_other"); + const commandlessFinal = withoutCommandId(hwlabRecord(2, { + type: "assistant", + eventType: "assistant", + status: "running", + itemId: "assistant_final", + assistantText: "This belongs to the existing command.", + finalResponse: { text: "This belongs to the existing command." }, + final: true, + replyAuthority: true + })); + const commandlessTerminal = withoutCommandId(hwlabRecord(3, { + type: "result", + eventType: "terminal", + status: "completed", + terminal: true + })); + + const result = await runKafkaCli([ + "render", "trace", + "--from", "kafka", + "--trace-id", TRACE_ID, + "--session-id", HWLAB_SESSION_ID, + "--run-id", RUN_ID, + "--command-id", "cmd_missing", + "--group-prefix", "hwlab-v03-workbench-isolated-debug", + "--format", "markdown" + ], { + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka() { + return { + events: [otherCommand, commandlessFinal, commandlessTerminal], + scannedCount: 3, + parsedCount: 3, + matchedCount: 3, + filterRejectedCount: 0, + completionReason: "end-offset", + reachedEndOffsets: true + }; + } + }); + + assert.equal(result.exitCode, 1); + assert.equal(result.payload.error.code, "source_command_missing"); + assert.equal(result.payload.error.details.commandScope.exactCommandCount, 0); + assert.equal(result.payload.error.details.commandScope.selectedCount, 0); + assert.equal(result.payload.error.details.commandScope.selectedUnscopedLifecycleCount, 0); }); test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug events without runtime dependencies", async () => { @@ -325,6 +612,170 @@ test("Kafka query exposes the isolated generated group and original value hash", assert.match(queried.groupId, /^hwlab-v03-workbench-isolated-debug-\d+-[a-f0-9]{8}$/u); assert.equal(consumerConfigs[0].groupId, queried.groupId); assert.equal(queried.events[0].valueSha256, createSha256(valueText)); + assert.equal(queried.scannedCount, 1); + assert.equal(queried.parsedCount, 1); + assert.equal(queried.matchedCount, 1); + assert.equal(queried.invalidJsonCount, 0); + assert.equal(queried.completionReason, "limit"); + assert.equal(queried.endOffsetsAvailable, false); +}); + +test("Kafka query reports invalid JSON, filter rejection, and bounded end-offset completion", async () => { + const matching = canonicalRecord(1).value; + const mismatch = structuredClone(matching); + mismatch.sessionId = "ses_agentrun_other"; + mismatch.run.sessionId = "ses_agentrun_other"; + mismatch.event.payload.sessionId = "ses_agentrun_other"; + const consumer = { + async connect() {}, + async subscribe() {}, + async run({ eachMessage }: any) { + await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "0", value: Buffer.from("{invalid") } }); + await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "1", value: Buffer.from(JSON.stringify(mismatch)) } }); + await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "2", value: Buffer.from(JSON.stringify(matching)) } }); + }, + async stop() {}, + async disconnect() {} + }; + const queried = await queryKafkaEventStream({ + env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" }, + topic: "agentrun.event.v1", + sessionId: SESSION_ID, + limit: 2, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "3" }]; }, async disconnect() {} }; }, + consumer() { return consumer; } + }; + } + }); + assert.equal(queried.scannedCount, 3); + assert.equal(queried.parsedCount, 2); + assert.equal(queried.invalidJsonCount, 1); + assert.equal(queried.filterRejectedCount, 1); + assert.equal(queried.matchedCount, 1); + assert.equal(queried.completionReason, "end-offset"); + assert.equal(queried.reachedEndOffsets, true); + assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "0", endOffset: "3" }]); + assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "2" }]); +}); + +test("Kafka query treats a retained empty partition as a completed bounded scan", async () => { + let runCalled = false; + let disconnected = false; + const queried = await queryKafkaEventStream({ + env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" }, + topic: "hwlab.event.v1", + traceId: TRACE_ID, + timeoutMs: 250, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "7", high: "7" }]; }, async disconnect() {} }; }, + consumer() { + return { + async connect() {}, + async subscribe() {}, + async run() { runCalled = true; }, + async stop() {}, + async disconnect() { disconnected = true; } + }; + } + }; + } + }); + assert.equal(queried.completionReason, "end-offset"); + assert.equal(queried.reachedEndOffsets, true); + assert.equal(queried.scannedCount, 0); + assert.equal(runCalled, false); + assert.equal(disconnected, true); + assert.deepEqual(queried.endOffsets, [{ partition: 0, startOffset: "7", endOffset: "7" }]); +}); + +test("Kafka query does not report an empty offset response as a completed scan", async () => { + const queried = await queryKafkaEventStream({ + env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" }, + topic: "hwlab.event.v1", + traceId: TRACE_ID, + timeoutMs: 250, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + admin() { return { async connect() {}, async fetchTopicOffsets() { return []; }, async disconnect() {} }; }, + consumer() { return { async connect() {}, async subscribe() {}, async run() {}, async stop() {}, async disconnect() {} }; } + }; + } + }); + assert.equal(queried.completionReason, "timeout"); + assert.equal(queried.reachedEndOffsets, false); + assert.equal(queried.endOffsetsAvailable, false); + assert.equal(queried.endOffsetsError?.code, "kafka_end_offsets_empty"); +}); + +test("Kafka query disconnects a connected consumer when subscribe fails", async () => { + let disconnected = 0; + await assert.rejects(queryKafkaEventStream({ + env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" }, + topic: "hwlab.event.v1", + traceId: TRACE_ID, + timeoutMs: 250, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + consumer() { + return { + async connect() {}, + async subscribe() { throw new Error("subscribe failed"); }, + async run() {}, + async stop() {}, + async disconnect() { disconnected += 1; } + }; + } + }; + } + }), /subscribe failed/u); + assert.equal(disconnected, 1); +}); + +test("Kafka query setup and cleanup stay inside the command budget", async () => { + let consumerConnectCalled = false; + let consumerDisconnected = false; + const startedAt = Date.now(); + const queried = await queryKafkaEventStream({ + env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" }, + topic: "hwlab.event.v1", + traceId: TRACE_ID, + timeoutMs: 250, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + admin() { + return { + async connect() { await new Promise(() => undefined); }, + async fetchTopicOffsets() { throw new Error("must not fetch after setup timeout"); }, + async disconnect() {} + }; + }, + consumer() { + return { + async connect() { consumerConnectCalled = true; }, + async subscribe() {}, + async run() {}, + async stop() {}, + async disconnect() { consumerDisconnected = true; } + }; + } + }; + } + }); + const elapsedMs = Date.now() - startedAt; + assert.equal(queried.completionReason, "timeout"); + assert.equal(queried.endOffsetsAvailable, false); + assert.equal(queried.endOffsetsError?.code, "kafka_query_timeout"); + assert.equal(consumerConnectCalled, false); + assert.equal(consumerDisconnected, true); + assert.ok(elapsedMs < 500, `query exceeded bounded setup budget: ${elapsedMs}ms`); }); test("non-reversible AgentRun session identity stays only in debug lineage", () => { @@ -465,6 +916,51 @@ function canonicalRecord(seq: number): any { return { topic: "agentrun.event.v1", partition: 0, offset: String(1600 + seq), key: SESSION_ID, value }; } +function hwlabRecord(sourceSeq: number, event: Record): any { + return { + topic: "hwlab.event.v1", + partition: 0, + offset: String(1900 + sourceSeq), + key: HWLAB_SESSION_ID, + value: { + schema: "hwlab.event.v1", + eventType: "hwlab.trace.event.projected", + eventId: `hwlab:evt_render_${sourceSeq}`, + sourceEventId: `evt_render_${sourceSeq}`, + traceId: TRACE_ID, + hwlabSessionId: HWLAB_SESSION_ID, + sessionId: HWLAB_SESSION_ID, + runId: RUN_ID, + commandId: "cmd_render_trace", + serverSentAt: `2026-07-10T11:00:0${sourceSeq}.000Z`, + event: { + traceId: TRACE_ID, + sessionId: HWLAB_SESSION_ID, + source: "agentrun.kafka", + sourceSeq, + sourceEventId: `evt_render_${sourceSeq}`, + runId: RUN_ID, + commandId: "cmd_render_trace", + createdAt: `2026-07-10T11:00:0${sourceSeq}.000Z`, + ...event + }, + valuesPrinted: false + } + }; +} + +function withoutCommandId(record: any) { + delete record.value.commandId; + delete record.value.event.commandId; + return record; +} + +function withCommandId(record: any, commandId: string) { + record.value.commandId = commandId; + record.value.event.commandId = commandId; + return record; +} + function createSha256(value: string) { return new Bun.CryptoHasher("sha256").update(value).digest("hex"); } diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts index 5cca5b71..049125fe 100644 --- a/tools/src/hwlab-cli/kafka-regenerate.ts +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -11,12 +11,18 @@ import { DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, DEFAULT_AGENTRUN_EVENT_TOPIC, DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, + DEFAULT_HWLAB_EVENT_TOPIC, projectAgentRunKafkaMessageToHwlabDebugEvent, queryKafkaEventStream } from "../../../internal/cloud/kafka-event-bridge.ts"; +import { decodeWorkbenchRealtimeEventFrame } from "../../../web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts"; +import { reduceWorkbenchRealtimeEvent } from "../../../web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts"; +import { projectWorkbenchLiveKafkaMessage } from "../../../web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts"; +import { planWorkbenchRealtimeApply } from "../../../web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts"; +import { renderTraceRowsMarkdown, traceDisplayRows } from "./trace-renderer.ts"; const CLI_NAME = "hwlab-cli"; -const VERSION = "0.3.1-kafka-debug-replay"; +const VERSION = "0.3.3-kafka-command-lifecycle-scope"; const DEFAULT_LIMIT = 500; const DEFAULT_TIMEOUT_MS = 5000; const DEBUG_OUTPUT_PARTITION = 0; @@ -36,12 +42,14 @@ type KafkaCliOptions = { export async function mainKafkaCli(argv = process.argv.slice(3), options: KafkaCliOptions = {}) { const result = await runKafkaCli(argv, options); const json = argv.includes("--json"); - process.stdout.write(json ? `${JSON.stringify(result.payload, null, 2)}\n` : `${renderKafkaCliText(result.payload)}\n`); + const markdown = requestedFormat(argv) === "markdown" && !json ? result.markdownOutput : null; + process.stdout.write(json ? `${JSON.stringify(result.payload, null, 2)}\n` : markdown ? `${markdown.trimEnd()}\n` : `${renderKafkaCliText(result.payload)}\n`); process.exitCode = result.exitCode; } export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) { const now = options.now ?? (() => new Date().toISOString()); + let action = "kafka.help"; try { const parsed = parseOptions(argv); const command = parsed._[0] || "help"; @@ -49,9 +57,19 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) if (["help", "--help", "-h"].includes(command) || parsed.help === true) { return result(0, kafkaHelp(), now); } - if (command !== "regenerate" || resource !== "hwlab") { - throw cliError("unsupported_kafka_command", "supported command: kafka regenerate hwlab", { command, resource }); + if (command === "render" && resource === "trace") { + action = "kafka.render.trace"; + const rendered = await renderHwlabKafkaTrace(parsed, { + cwd: options.cwd ?? process.cwd(), + env: options.env ?? process.env, + now, + readKafka: options.readKafka ?? defaultKafkaReader, + createProducer: options.createProducer ?? defaultProducerFactory + }); + return { ...result(0, rendered.payload, now), markdownOutput: rendered.markdown }; } + if (command !== "regenerate" || resource !== "hwlab") throw cliError("unsupported_kafka_command", "supported commands: kafka regenerate hwlab | kafka render trace", { command, resource }); + action = "kafka.regenerate.hwlab"; const payload = await regenerateHwlabDebugEvents(parsed, { cwd: options.cwd ?? process.cwd(), env: options.env ?? process.env, @@ -61,7 +79,7 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) }); return result(0, payload, now); } catch (error) { - return result(1, failure(error), now); + return result(1, failure(error, action), now); } } @@ -256,6 +274,321 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie }; } +export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Required) { + const traceId = requiredTraceId(parsed.traceId); + const sessionId = optionalSessionId(parsed.sessionId); + const runId = optionalRunId(parsed.runId); + const commandId = optionalCommandId(parsed.commandId); + const sourceMode = enumValue(parsed.from ?? "kafka", "from", ["kafka", "jsonl"]); + const format = enumValue(parsed.format ?? "text", "format", ["text", "markdown"]); + const inputTopicResolution = resolveConfig( + parsed.inputTopic, + dependencies.env.HWLAB_KAFKA_EVENT_TOPIC, + DEFAULT_HWLAB_EVENT_TOPIC, + "--input-topic", + "env:HWLAB_KAFKA_EVENT_TOPIC" + ); + const inputTopic = inputTopicResolution.value; + const limit = boundedInteger(parsed.limit, DEFAULT_LIMIT, 1, 5000, "limit"); + const timeoutMs = boundedInteger(parsed.timeoutMs, DEFAULT_TIMEOUT_MS, 250, 60000, "timeoutMs"); + const expectedCount = optionalPositiveInteger(parsed.expectCount, "expectCount"); + const readLimit = commandId || expectedCount === undefined ? limit : Math.min(5000, expectedCount + 1); + const groupPrefixResolution = resolveKafkaGroupConfig( + parsed.groupPrefix, + dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX, + sourceMode === "kafka" + ); + const groupPrefix = text(groupPrefixResolution.value); + if (groupPrefix) assertDebugGroup(groupPrefix); + + const readResult = sourceMode === "jsonl" + ? await readJsonlInput(requiredText(parsed.jsonlFile, "jsonlFile"), dependencies.cwd, inputTopic) + : await dependencies.readKafka({ + env: dependencies.env, + stream: "hwlab", + topic: inputTopic, + traceId, + sessionId: sessionId || null, + runId: runId || null, + limit: readLimit, + timeoutMs, + fromBeginning: parsed.fromEnd !== true, + groupIdPrefix: groupPrefix + }); + const records = Array.isArray(readResult.events) ? readResult.events : []; + const scoped = scopeHwlabTraceRecords(records, { traceId, sessionId, runId, commandId }); + const commandScopeScanComplete = sourceMode === "jsonl" || readResult.completionReason === "end-offset"; + const matchingRecords = scoped.records; + if (matchingRecords.length === 0) { + throw cliError(commandId && scoped.summary.baseMatchedCount > 0 ? "source_command_missing" : "source_trace_missing", commandId + ? "no HWLAB Kafka events matched the requested command lifecycle" + : "no HWLAB Kafka events matched the requested traceId/sessionId", { + traceId, + sessionId: sessionId || null, + runId: runId || null, + commandId: commandId || null, + scannedCount: readResult.scannedCount ?? records.length, + parsedCount: readResult.parsedCount ?? records.length, + queryMatchedCount: readResult.matchedCount ?? records.length, + commandScope: { ...scoped.summary, scanComplete: commandScopeScanComplete }, + completionReason: readResult.completionReason ?? null, + reachedEndOffsets: readResult.reachedEndOffsets ?? null + }); + } + if (expectedCount !== undefined && matchingRecords.length !== expectedCount) { + throw cliError("hwlab_trace_event_count_mismatch", "matched HWLAB event count does not equal --expect-count", { + expectedCount, + actualCount: matchingRecords.length, + traceId, + sessionId: sessionId || null + }); + } + + const projection = projectHwlabRecordsThroughWorkbench(matchingRecords, { + traceId, + sessionId, + observedAt: dependencies.now() + }); + if (!projection.message?.runnerTrace) { + throw cliError("workbench_trace_projection_empty", "HWLAB events did not produce a Workbench runner trace", { + traceId, + decodedCount: projection.decodedCount, + plannedCount: projection.plannedCount, + appliedCount: projection.appliedCount, + rejectedCount: projection.rejections.length, + firstRejection: projection.rejections[0] ?? null + }); + } + const finalResponseText = projection.message.traceAutoLifecycle === "terminal" ? text(projection.message.text) : ""; + const trace = projection.message.runnerTrace as Record; + const events = Array.isArray(trace.events) ? trace.events : []; + const rows = traceDisplayRows(trace, events, { + finalResponsePlacement: "outer", + outerFinalResponseText: finalResponseText || null + }); + const markdown = renderWorkbenchKafkaTraceMarkdown({ + traceId, + sessionId: text(projection.message.sessionId), + status: text(projection.message.status), + sourceEventCount: events.length, + rows, + finalResponseText: finalResponseText || null + }); + const stateId = `hwlab-trace-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`; + const artifactPath = path.resolve( + dependencies.cwd, + text(parsed.outputMarkdown) || path.join(".state", "hwlab-cli", "kafka-render", stateId, `${traceId}.md`) + ); + await mkdir(path.dirname(artifactPath), { recursive: true }); + await writeFile(artifactPath, `${markdown.trimEnd()}\n`, "utf8"); + const fullRows = parsed.full === true; + const rowPreviewLimit = boundedInteger(parsed.rowLimit, 20, 1, 200, "rowLimit"); + const selectedRows = fullRows ? rows : rows.slice(-rowPreviewLimit); + const sessionIds = uniqueText(matchingRecords.flatMap((record) => hwlabSessionCandidates(valueFromRecord(record)))); + const runIds = uniqueText(matchingRecords.flatMap((record) => runCandidates(valueFromRecord(record)))); + const commandIds = uniqueText(matchingRecords.flatMap((record) => commandCandidates(valueFromRecord(record)))); + const sourceEventIdentityValues = matchingRecords.map((record) => text(valueFromRecord(record)?.sourceEventId ?? valueFromRecord(record)?.event?.sourceEventId)); + const sourceEventIds = uniqueText(sourceEventIdentityValues); + const sourceEventIdentityPresentCount = sourceEventIdentityValues.filter(Boolean).length; + const rowPayload = selectedRows.map((row) => ({ + rowId: row.rowId, + seq: row.seq, + sequenceAuthority: row.sequenceAuthority, + tone: row.tone, + header: row.header, + preview: row.preview ?? null, + body: fullRows ? row.body : boundedText(row.body, 500), + bodyTruncated: !fullRows && text(row.body).length > 500, + terminal: row.terminal === true, + bodyFormat: row.bodyFormat ?? null + })); + const toolRowCount = rows.filter((row) => row.rowId.startsWith("tool:")).length; + const assistantRowCount = rows.filter((row) => /助手/u.test(row.header)).length; + const finalResponseFingerprint = traceTextFingerprint(finalResponseText); + const finalResponseInTrace = Boolean(finalResponseFingerprint && rows.some((row) => + /助手/u.test(row.header) && traceTextFingerprint(row.body) === finalResponseFingerprint + )); + return { + markdown, + payload: { + ok: true, + action: "kafka.render.trace", + status: "succeeded", + format, + input: { + mode: sourceMode, + topic: inputTopic, + topicSource: inputTopicResolution.source, + groupId: readResult.groupId ?? null, + groupPrefix: groupPrefix || null, + groupPrefixSource: groupPrefixResolution.source, + scannedCount: readResult.scannedCount ?? records.length, + parsedCount: readResult.parsedCount ?? records.length, + queryMatchedCount: readResult.matchedCount ?? records.length, + matchedCount: matchingRecords.length, + invalidJsonCount: (readResult.invalidJsonCount ?? 0) + scoped.summary.invalidTransportRecordCount, + filterRejectedCount: readResult.filterRejectedCount ?? 0, + scopeRejectedCount: scoped.summary.baseRejectedCount + scoped.summary.commandScopeRejectedCount, + commandScope: { ...scoped.summary, scanComplete: commandScopeScanComplete }, + completionReason: readResult.completionReason ?? (sourceMode === "jsonl" ? "file-end" : null), + reachedEndOffsets: readResult.reachedEndOffsets ?? (sourceMode === "jsonl"), + endOffsetsAvailable: readResult.endOffsetsAvailable ?? false, + endOffsets: readResult.endOffsets ?? [], + lastScannedOffsets: readResult.lastScannedOffsets ?? [], + sourceFile: readResult.sourceFile ?? null, + valuesPrinted: false + }, + identity: { + requestedTraceId: traceId, + requestedSessionId: sessionId || null, + requestedRunId: runId || null, + requestedCommandId: commandId || null, + observedTraceId: text(projection.message.traceId), + observedSessionIds: sessionIds, + observedRunIds: runIds, + observedCommandIds: commandIds, + baseObservedRunIds: scoped.summary.baseObservedRunIds, + baseCommandIdCounts: scoped.summary.commandIdCounts, + missingCommandIdCount: scoped.summary.missingCommandIdCount, + selectedSessionId: text(projection.message.sessionId), + sourceEventIdentityCount: sourceEventIds.length, + sourceEventIdentityMissingCount: matchingRecords.length - sourceEventIdentityPresentCount, + sourceEventIdentityDuplicates: Math.max(0, sourceEventIdentityPresentCount - sourceEventIds.length), + valuesPrinted: false + }, + projection: { + pipeline: ["decodeWorkbenchRealtimeEventFrame", "reduceWorkbenchRealtimeEvent", "planWorkbenchRealtimeApply", "projectWorkbenchLiveKafkaMessage", "traceDisplayRows"], + decodedCount: projection.decodedCount, + plannedCount: projection.plannedCount, + appliedCount: projection.appliedCount, + rejectedCount: projection.rejections.length, + rejections: projection.rejections.slice(0, 20), + eventCount: events.length, + terminalObserved: projection.message.traceAutoLifecycle === "terminal", + status: projection.message.status, + lastEventAt: projection.message.timing?.lastEventAt ?? null, + valuesPrinted: false + }, + render: { + rowModel: "tools/src/hwlab-cli/trace-renderer:traceDisplayRows", + markdownRenderer: "tools/src/hwlab-cli/trace-renderer:renderTraceRowsMarkdown", + finalResponsePlacement: "outer", + rowCount: rows.length, + rowsReturned: rowPayload.length, + rowsOmitted: Math.max(0, rows.length - rowPayload.length), + rowWindow: fullRows ? "full" : "tail", + rowPreviewLimit: fullRows ? null : rowPreviewLimit, + toolRowCount, + assistantRowCount, + finalResponsePresent: Boolean(finalResponseText), + finalResponseInTrace, + rowModelSha256: sha256(JSON.stringify(rows)), + rows: rowPayload, + valuesPrinted: false + }, + output: { + artifact: { path: artifactPath, byteCount: Buffer.byteLength(`${markdown.trimEnd()}\n`, "utf8"), sha256: sha256(`${markdown.trimEnd()}\n`) }, + valuesPrinted: false + }, + validation: { + expectedCount: expectedCount ?? null, + allMatchedEventsApplied: projection.appliedCount === matchingRecords.length, + commandLifecyclePreserved: !commandId || (commandScopeScanComplete && scoped.summary.exactCommandCount > 0 && scoped.summary.ambiguousUnscopedLifecycleCount === 0), + finalResponseOutsideTrace: !finalResponseInTrace && Boolean(finalResponseText), + sharedRowModel: true, + valuesPrinted: false + }, + runtimeDependencies: { + cloudApi: false, + database: false, + transactionalProjector: false, + kafka: sourceMode === "kafka" + }, + next: { + command: `hwlab-cli kafka render trace --from ${sourceMode} --trace-id ${traceId}${sessionId ? ` --session-id ${sessionId}` : ""}${runId ? ` --run-id ${runId}` : ""}${commandId ? ` --command-id ${commandId}` : ""}${sourceMode === "jsonl" ? ` --jsonl-file ${shellArg(readResult.sourceFile)}` : ""} --format markdown`, + reason: "使用同一持久事件和共享 row model 直接检查 Markdown Trace；浏览器验收只需核对最后的 HTML 分叉。" + }, + valuesPrinted: false + } + }; +} + +export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], options: { traceId: string; sessionId?: string; observedAt: string }) { + let message: any = null; + let decodedCount = 0; + let plannedCount = 0; + let appliedCount = 0; + const rejections: Array<{ index: number; code: string; detail?: string }> = []; + const observedBaseMs = Date.parse(options.observedAt); + const baseMs = Number.isFinite(observedBaseMs) ? observedBaseMs : 0; + for (const [index, record] of records.entries()) { + const value = valueFromRecord(record); + const decoded = decodeWorkbenchRealtimeEventFrame(JSON.stringify(value)); + if (!decoded.payload) { + rejections.push({ index, code: `decode-${decoded.status}` }); + continue; + } + decodedCount += 1; + const schema = text(decoded.payload.schema); + const realtimeEvent = schema === "hwlab.event.debug.v1" + ? { ...decoded.payload, schema: "hwlab.event.v1" } + : decoded.payload; + if (realtimeEvent.schema !== "hwlab.event.v1") { + rejections.push({ index, code: "schema-mismatch", detail: text(realtimeEvent.schema) }); + continue; + } + const reduced = reduceWorkbenchRealtimeEvent(realtimeEvent as any, "hwlab.event.v1"); + const plan = planWorkbenchRealtimeApply(reduced.action); + const traceStep = plan.steps.find((step) => step.type === "apply-trace-event"); + if (!traceStep || traceStep.type !== "apply-trace-event" || !traceStep.event) { + rejections.push({ index, code: reduced.action.type === "ignore" ? reduced.action.reason : "trace-step-missing" }); + continue; + } + plannedCount += 1; + const eventTraceId = text(realtimeEvent.traceId ?? traceStep.event.traceId); + if (eventTraceId !== options.traceId) { + rejections.push({ index, code: "trace-mismatch", detail: eventTraceId }); + continue; + } + const eventSessionId = text(realtimeEvent.hwlabSessionId ?? realtimeEvent.sessionId ?? traceStep.event.sessionId ?? options.sessionId); + if (!eventSessionId) { + rejections.push({ index, code: "session-missing" }); + continue; + } + if (options.sessionId && !hwlabSessionCandidates(realtimeEvent).includes(options.sessionId)) { + rejections.push({ index, code: "session-mismatch", detail: eventSessionId }); + continue; + } + message = projectWorkbenchLiveKafkaMessage({ + previous: message, + traceId: options.traceId, + sessionId: eventSessionId, + event: traceStep.event as any, + receivedAt: new Date(baseMs + index).toISOString() + }); + appliedCount += 1; + } + return { message, decodedCount, plannedCount, appliedCount, rejections }; +} + +function renderWorkbenchKafkaTraceMarkdown(input: { traceId: string; sessionId: string; status: string; sourceEventCount: number; rows: any[]; finalResponseText: string | null }) { + const lines = [ + `# Workbench Trace ${input.traceId}`, + "", + `- session: ${input.sessionId || "unknown"}`, + `- status: ${input.status || "unknown"}`, + `- source events: ${input.sourceEventCount}`, + `- rendered rows: ${input.rows.length}`, + `- final response placement: outer`, + "", + "## 运行记录", + "", + renderTraceRowsMarkdown(input.rows) + ]; + if (input.finalResponseText) lines.push("", "## 最终回复", "", input.finalResponseText); + return lines.join("\n"); +} + export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], options: { sessionId: string; traceId?: string; expectedCount?: number; producedAt?: string; replayId?: string }) { const matchingRecords = records.filter((record) => recordMatches(record, options.sessionId, options.traceId)); if (matchingRecords.length === 0) { @@ -318,12 +651,15 @@ function kafkaHelp() { action: "kafka.help", status: "succeeded", commands: [ + "hwlab-cli kafka render trace --from kafka --trace-id trc_... [--session-id ses_...] [--run-id run_...] [--command-id cmd_...] [--input-topic hwlab.event.v1] [--group-prefix hwlab-...-debug] [--format markdown] [--row-limit 20] [--json]", + "hwlab-cli kafka render trace --from jsonl --trace-id trc_... [--run-id run_...] [--command-id cmd_...] --jsonl-file hwlab-events.jsonl [--format markdown] [--output-markdown trace.md] [--row-limit 20] [--json]", "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... [--trace-id trc_...] [--replay-id rpl_...] [--input-topic agentrun.event.v1] [--group-prefix hwlab-...-debug] [--expect-count 35] [--json]", "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... --input-topic agentrun.event.debug.v1 --group-prefix hwlab-...-debug [--replay-id rpl_...] [--json]", "hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... [--replay-id rpl_...] --jsonl-file agentrun-events.jsonl --no-publish [--output-jsonl hwlab-events.jsonl] [--json]" ], defaults: { kafkaInputTopic: DEFAULT_AGENTRUN_EVENT_TOPIC, + kafkaTraceInputTopic: DEFAULT_HWLAB_EVENT_TOPIC, jsonlInputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, outputTopic: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, groupPrefix: null, @@ -348,10 +684,27 @@ function kafkaHelp() { } async function defaultKafkaReader(input: Record) { - return queryKafkaEventStream({ + return withBunKafkaTimerClamp(() => queryKafkaEventStream({ ...input, kafkaFactory(config: { brokers: string[]; clientId: string }) { return boundedKafka(config); } - }); + })); +} + +async function withBunKafkaTimerClamp(task: () => Promise) { + if (!process.versions.bun) return task(); + const originalSetTimeout = globalThis.setTimeout; + // KafkaJS 2.2.4 schedules an idle queue check at 0 - Date.now(); Bun clamps it but emits an unbounded warning stack. + const clampedSetTimeout = ((handler: Parameters[0], timeout?: number, ...args: any[]) => originalSetTimeout( + handler, + typeof timeout === "number" && timeout < 0 ? 1 : timeout, + ...args + )) as typeof globalThis.setTimeout; + globalThis.setTimeout = clampedSetTimeout; + try { + return await task(); + } finally { + if (globalThis.setTimeout === clampedSetTimeout) globalThis.setTimeout = originalSetTimeout; + } } async function defaultProducerFactory({ env, clientId }: { env: EnvLike; clientId: string }) { @@ -383,7 +736,21 @@ async function readJsonlInput(fileValue: string, cwd: string, defaultTopic: stri } events.push(normalizeRecord(value, defaultTopic)); } - return { ok: true, topic: defaultTopic, groupId: null, events, sourceFile: file }; + return { + ok: true, + topic: defaultTopic, + groupId: null, + count: events.length, + scannedCount: events.length, + parsedCount: events.length, + matchedCount: events.length, + invalidJsonCount: 0, + filterRejectedCount: 0, + completionReason: "file-end", + reachedEndOffsets: true, + events, + sourceFile: file + }; } function normalizeRecord(record: any, defaultTopic: string) { @@ -432,12 +799,145 @@ function sessionCandidates(value: any) { value.run?.sessionId, value.run?.hwlabSessionId, value.event?.payload?.sessionId, - value.event?.payload?.hwlabSessionId + value.event?.payload?.hwlabSessionId, + value.event?.sessionId, + value.event?.hwlabSessionId ]); } function traceCandidates(value: any) { - return uniqueText([value.traceId, value.run?.traceId, value.command?.traceId, value.event?.payload?.traceId, value.event?.payload?.hwlabTraceId]); + return uniqueText([value.traceId, value.run?.traceId, value.command?.traceId, value.event?.traceId, value.event?.payload?.traceId, value.event?.payload?.hwlabTraceId]); +} + +function scopeHwlabTraceRecords(records: JsonRecord[], filters: { traceId: string; sessionId?: string; runId?: string; commandId?: string }) { + const parsed: Array<{ record: JsonRecord; value: Record }> = []; + let invalidTransportRecordCount = 0; + for (const record of records) { + try { + parsed.push({ record, value: valueFromRecord(record) }); + } catch { + invalidTransportRecordCount += 1; + } + } + const base = parsed.filter(({ value }) => { + if (!traceCandidates(value).includes(filters.traceId)) return false; + if (filters.sessionId && !hwlabSessionCandidates(value).includes(filters.sessionId)) return false; + return !filters.runId || runCandidates(value).includes(filters.runId); + }); + const baseObservedRunIds = uniqueText(base.flatMap(({ value }) => runCandidates(value))); + const commandIdCountMap = new Map(); + let missingCommandIdCount = 0; + let unscopedLifecycleCount = 0; + let unscopedNonLifecycleCount = 0; + for (const { value } of base) { + const commands = commandCandidates(value); + if (commands.length === 0) { + missingCommandIdCount += 1; + if (hwlabCommandLifecycleEvent(value)) unscopedLifecycleCount += 1; + else unscopedNonLifecycleCount += 1; + continue; + } + for (const candidate of commands) commandIdCountMap.set(candidate, (commandIdCountMap.get(candidate) ?? 0) + 1); + } + const exactCommandRecords = filters.commandId + ? base.filter(({ value }) => commandCandidates(value).includes(filters.commandId as string)) + : []; + const soleBaseRunId = baseObservedRunIds.length === 1 ? baseObservedRunIds[0] : null; + const explicitTargetRunIds = uniqueText([ + filters.runId, + ...exactCommandRecords.flatMap(({ value }) => runCandidates(value)) + ]); + const targetRunIds = explicitTargetRunIds.length > 0 ? explicitTargetRunIds : uniqueText([soleBaseRunId]); + let selectedUnscopedLifecycleCount = 0; + let attributableUnscopedLifecycleCount = 0; + let ambiguousUnscopedLifecycleCount = 0; + let otherRunUnscopedLifecycleCount = 0; + let otherCommandRecordCount = 0; + const selected = !filters.commandId + ? base + : exactCommandRecords.length === 0 + ? [] + : base.filter(({ value }) => { + const commands = commandCandidates(value); + if (commands.includes(filters.commandId as string)) return true; + if (commands.length > 0) { + otherCommandRecordCount += 1; + return false; + } + if (!hwlabCommandLifecycleEvent(value)) return false; + const runs = runCandidates(value); + const attributable = runs.some((candidate) => targetRunIds.includes(candidate)) + || (runs.length === 0 && Boolean(soleBaseRunId) && targetRunIds.includes(soleBaseRunId as string)); + if (!attributable) { + if (runs.length > 0 && targetRunIds.length > 0) otherRunUnscopedLifecycleCount += 1; + else ambiguousUnscopedLifecycleCount += 1; + return false; + } + attributableUnscopedLifecycleCount += 1; + selectedUnscopedLifecycleCount += 1; + return true; + }); + return { + records: selected.map(({ record }) => record), + summary: { + mode: filters.commandId ? "command-lifecycle" : "trace-run", + requestedCommandId: filters.commandId || null, + baseMatchedCount: base.length, + selectedCount: selected.length, + exactCommandCount: exactCommandRecords.length, + missingCommandIdCount, + unscopedLifecycleCount, + attributableUnscopedLifecycleCount, + selectedUnscopedLifecycleCount, + ambiguousUnscopedLifecycleCount, + otherRunUnscopedLifecycleCount, + unscopedNonLifecycleCount, + otherCommandRecordCount, + commandScopeRejectedCount: Math.max(0, base.length - selected.length), + baseRejectedCount: Math.max(0, records.length - invalidTransportRecordCount - base.length), + invalidTransportRecordCount, + baseObservedRunIds, + targetRunIds, + commandIdCounts: [...commandIdCountMap.entries()] + .sort(([left], [right]) => left.localeCompare(right)) + .map(([commandId, count]) => ({ commandId, count })), + valuesPrinted: false + } + }; +} + +function hwlabCommandLifecycleEvent(value: Record) { + const event = recordObject(value.event) ?? value; + const finalResponse = recordObject(event.finalResponse); + const types = uniqueText([event.type, event.eventType, value.type, value.eventType]).map((candidate) => candidate.toLowerCase()); + return event.terminal === true + || event.final === true + || event.replyAuthority === true + || Boolean(text(finalResponse?.text)) + || types.some((candidate) => ["terminal", "terminal_status", "result"].includes(candidate)); +} + +function hwlabSessionCandidates(value: any) { + return uniqueText([ + value?.hwlabSessionId, + value?.sessionId, + value?.event?.hwlabSessionId, + value?.event?.sessionId, + value?.context?.hwlabSessionId, + value?.context?.sessionId + ]); +} + +function runCandidates(value: any) { + return uniqueText([value?.runId, value?.run_id, value?.event?.runId, value?.event?.run_id, value?.context?.runId, value?.context?.run_id]); +} + +function commandCandidates(value: any) { + return uniqueText([value?.commandId, value?.command_id, value?.event?.commandId, value?.event?.command_id, value?.context?.commandId, value?.context?.command_id]); +} + +function recordObject(value: unknown): Record | null { + return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } function parseOptions(argv: string[]): ParsedArgs { @@ -462,7 +962,17 @@ function renderKafkaCliText(payload: Record) { return [ "ok", `action=${payload.action}`, - payload.input ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null, + payload.input?.scannedCount !== undefined ? `scanned=${payload.input.scannedCount}` : null, + payload.input?.parsedCount !== undefined ? `parsed=${payload.input.parsedCount}` : null, + payload.input?.queryMatchedCount !== undefined ? `queryMatched=${payload.input.queryMatchedCount}` : null, + payload.input?.matchedCount !== undefined ? `matched=${payload.input.matchedCount}` : null, + payload.input?.commandScope?.exactCommandCount !== undefined ? `commandExact=${payload.input.commandScope.exactCommandCount}` : null, + payload.input?.commandScope?.selectedUnscopedLifecycleCount !== undefined ? `commandLifecycle=${payload.input.commandScope.selectedUnscopedLifecycleCount}` : null, + payload.input?.commandScope?.scanComplete !== undefined ? `scopeComplete=${payload.input.commandScope.scanComplete}` : null, + payload.input?.readCount !== undefined ? `input=${payload.input.matchedCount}/${payload.input.readCount}` : null, + payload.projection?.appliedCount !== undefined ? `applied=${payload.projection.appliedCount}` : null, + payload.render?.rowCount !== undefined ? `rows=${payload.render.rowCount}` : null, + payload.projection?.terminalObserved !== undefined ? `terminal=${payload.projection.terminalObserved}` : null, payload.output ? `output=${payload.output.count}` : null, payload.output ? `published=${payload.output.publishedCount}` : null, validation.orderPreserved !== undefined ? `order=${validation.orderPreserved ? "preserved" : "failed"}` : null, @@ -478,13 +988,13 @@ function result(exitCode: number, payload: Record, now: () => strin return { exitCode, payload: { generatedAt: now(), cli: CLI_NAME, version: VERSION, ...payload } }; } -function failure(error: any) { +function failure(error: any, action = "kafka.regenerate.hwlab") { return { ok: false, - action: "kafka.regenerate.hwlab", + action, status: "failed", error: { code: error?.code ?? "hwlab_kafka_debug_cli_failed", message: error instanceof Error ? error.message : String(error), details: error?.details ?? undefined }, - next: { command: "hwlab-cli kafka help --json", reason: "检查独立 debug topic、sessionId、JSONL 合同和 Kafka YAML 注入后重试。" }, + next: { command: "hwlab-cli kafka help --json", reason: "检查 Kafka topic/group 的 YAML 注入、trace/session identity、JSONL 合同和扫描可见性后重试。" }, valuesPrinted: false }; } @@ -505,6 +1015,30 @@ function requiredSessionId(value: unknown) { return result; } +function optionalSessionId(value: unknown) { + const result = text(value); + if (result && !/^ses_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_session_id", "sessionId must start with ses_", { sessionId: result }); + return result; +} + +function optionalRunId(value: unknown) { + const result = text(value); + if (result && !/^run_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_run_id", "runId must start with run_", { runId: result }); + return result; +} + +function optionalCommandId(value: unknown) { + const result = text(value); + if (result && !/^cmd_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_command_id", "commandId must start with cmd_", { commandId: result }); + return result; +} + +function requiredTraceId(value: unknown) { + const result = optionalTraceId(value); + if (!result) throw cliError("trace_id_required", "--trace-id is required", { field: "traceId" }); + return result; +} + function optionalTraceId(value: unknown) { const result = text(value); if (result && !/^trc_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_trace_id", "traceId must start with trc_", { traceId: result }); @@ -597,6 +1131,9 @@ function cliError(code: string, message: string, details: Record item.trim()).filter(Boolean); } function sha256(value: string) { return createHash("sha256").update(value).digest("hex"); } function shellArg(value: unknown) { const result = String(value ?? ""); return /^[A-Za-z0-9_./:@-]+$/u.test(result) ? result : `'${result.replace(/'/gu, `'"'"'`)}'`; } +function requestedFormat(argv: string[]) { const index = argv.findIndex((item) => item === "--format"); if (index >= 0) return text(argv[index + 1]); const inline = argv.find((item) => item.startsWith("--format=")); return inline ? text(inline.slice("--format=".length)) : "text"; } diff --git a/tools/src/hwlab-cli/trace-renderer.ts b/tools/src/hwlab-cli/trace-renderer.ts index 3df3f248..1fa2205b 100644 --- a/tools/src/hwlab-cli/trace-renderer.ts +++ b/tools/src/hwlab-cli/trace-renderer.ts @@ -13,27 +13,37 @@ export interface TraceEventRow { type TraceEvent = Record; type TraceClockFormatter = (value: string) => string; export type TraceSequenceAuthority = "projected" | "source"; +export type TraceFinalResponsePlacement = "trace" | "outer"; export interface TraceDisplayRowsOptions { formatClock?: TraceClockFormatter; + finalResponsePlacement?: TraceFinalResponsePlacement; + outerFinalResponseText?: string | null; } interface ResolvedTraceDisplayRowsOptions extends TraceDisplayRowsOptions { sequenceAuthority: TraceSequenceAuthority; + finalResponsePlacement: TraceFinalResponsePlacement; } export function traceDisplayRows(trace: Record = {}, events: TraceEvent[] = [], options: TraceDisplayRowsOptions = {}): TraceEventRow[] { const displayOptions: ResolvedTraceDisplayRowsOptions = { ...options, - sequenceAuthority: traceSequenceAuthority(trace) + sequenceAuthority: traceSequenceAuthority(trace), + finalResponsePlacement: options.finalResponsePlacement ?? "trace" }; const orderedEvents = traceEventsForDisplay( events.filter((event) => hasTraceSequence(event, displayOptions.sequenceAuthority)), displayOptions.sequenceAuthority ); const effectiveTrace = traceWithInferredStart(trace, orderedEvents); - const finalResponseText = traceFinalResponseText(effectiveTrace); + const finalResponseText = nonEmptyString(options.outerFinalResponseText) ?? traceFinalResponseText(effectiveTrace); const finalResponseFingerprint = traceMessageFingerprint(finalResponseText); + const terminalAssistantFingerprints = new Set(orderedEvents + .filter(isTerminalAssistantTraceEvent) + .map((event) => traceAssistantEventFingerprint(effectiveTrace, event)) + .filter((value): value is string => Boolean(value))); + if (finalResponseFingerprint) terminalAssistantFingerprints.add(finalResponseFingerprint); const hasCompletionEvent = orderedEvents.some(isCompletionTraceEvent); const rows: TraceEventRow[] = []; let completionEvent: TraceEvent | null = null; @@ -64,6 +74,7 @@ export function traceDisplayRows(trace: Record = {}, events: Tr continue; } if (isTerminalAssistantTraceEvent(event)) { + if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue; if (hasCompletionEvent) continue; if (finalResponseText) { rows.push(traceFinalResponseRow(event, finalResponseText, displayOptions)); @@ -73,7 +84,9 @@ export function traceDisplayRows(trace: Record = {}, events: Tr continue; } if (isAssistantTraceEvent(event)) { - if (finalResponseFingerprint && traceAssistantEventFingerprint(effectiveTrace, event) === finalResponseFingerprint) continue; + const fingerprint = traceAssistantEventFingerprint(effectiveTrace, event); + if (isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) continue; + if (finalResponseFingerprint && fingerprint === finalResponseFingerprint) continue; rows.push(traceAssistantMessageRow(effectiveTrace, event, { terminal: false }, displayOptions)); continue; } @@ -88,7 +101,23 @@ export function traceDisplayRows(trace: Record = {}, events: Tr 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, displayOptions)); + return orderedEvents + .filter((event) => !isSuppressedTraceEvent(event)) + .filter((event) => !isOuterFinalResponseTraceEvent(effectiveTrace, event, displayOptions, terminalAssistantFingerprints)) + .map((event) => traceDisplayRow(effectiveTrace, event, displayOptions)); +} + +function isOuterFinalResponseTraceEvent( + trace: Record, + event: TraceEvent, + options: ResolvedTraceDisplayRowsOptions, + terminalAssistantFingerprints: Set +): boolean { + if (options.finalResponsePlacement !== "outer") return false; + if (isTerminalAssistantTraceEvent(event)) return true; + if (!isAssistantTraceEvent(event)) return false; + const fingerprint = traceAssistantEventFingerprint(trace, event); + return Boolean(fingerprint && terminalAssistantFingerprints.has(fingerprint)); } function traceSequenceAuthority(trace: Record): TraceSequenceAuthority { @@ -177,42 +206,71 @@ function stripTraceToolOutputFromSummary(value: string): string { function traceToolCallRow(trace: Record, event: TraceEvent, options: ResolvedTraceDisplayRowsOptions): TraceEventRow { const command = cleanShellCommand(event.command); + const outputPreview = traceEventText(event, ["outputSummary", "stdoutSummary", "stderrSummary", "output", "outputText"]); const status = traceStatusToken(event); const toolName = traceToolName(event, command); const body = traceToolCallBody(event, command); const eventKey = traceEventIdentityToken(event, options.sequenceAuthority); + const duration = numberOrNull(event.durationMs); + const exitCode = numberOrNull(event.exitCode); return { rowId: `tool:${event.itemId ?? eventKey ?? `${event.label ?? event.type ?? "tool"}:${event.createdAt ?? "unknown"}`}`, seq: traceEventDisplaySeq(event, options.sequenceAuthority), sequenceAuthority: options.sequenceAuthority, tone: traceEventTone(event), - header: `${traceEventClock(event, options)} total=${formatTraceDuration(traceRelativeMs(trace, event))} ${status} ${toolName}`.trim(), + header: [ + traceEventClock(event, options), + `total=${formatTraceDuration(traceRelativeMs(trace, event))}`, + status, + toolName, + duration !== null ? `duration=${formatTraceDuration(duration)}` : null, + exitCode !== null ? `exit=${exitCode}` : null + ].filter(Boolean).join(" "), body, - preview: command ? compactTraceOneLine(command, 2000) : null, + preview: traceToolPreview(command, outputPreview), bodyFormat: "text" }; } +function traceToolPreview(command: string | null, output: string | null): string | null { + const parts = [ + command ? compactTraceOneLine(command, 92) : null, + output ? `output: ${compactTraceOneLine(output, 92)}` : null + ].filter((value): value is string => Boolean(value)); + return parts.length > 0 ? parts.join(" | ") : null; +} + function traceToolName(event: TraceEvent, command: string | null): string { const direct = nonEmptyString(event.toolName); - if (direct && !isProtocolToolLifecycleName(direct)) return direct; + if (direct && !isProtocolToolLifecycleName(direct) && !isGenericToolName(direct)) return direct; const label = readableTraceLabel(event).replace(/^agentrun:tool:/u, "").replace(/^tool:/u, "").replace(/^item\//u, ""); - if (label && !isProtocolToolLifecycleName(label)) return label; + if (label && !isProtocolToolLifecycleName(label) && !isGenericToolName(label)) return label; const commandName = traceToolNameFromCommand(command); if (commandName) return commandName; + if (direct && !isProtocolToolLifecycleName(direct)) return direct; const previewName = traceToolNameFromPreview(event); if (previewName) return previewName; return "tool call"; } +function isGenericToolName(value: unknown): boolean { + return /^(commandExecution|tool|shell|command|unknown)$/iu.test(String(value ?? "").trim()); +} + function isProtocolToolLifecycleName(value: unknown): boolean { return /^(?:item\/)?(?:started|completed)$/iu.test(String(value ?? "").trim()) || /^item\/(?:started|completed)$/iu.test(String(value ?? "").trim()); } function traceToolNameFromCommand(command: string | null): string | null { const text = cleanTraceText(command); - const match = text.match(/^([A-Za-z][A-Za-z0-9_.:-]{1,80})(?:\s|$)/u); - return match?.[1] ?? null; + const segments = text.split(/\s*(?:&&|\|\||;)\s*/u).map((value) => value.trim()).filter(Boolean); + const candidate = segments.find((value) => !/^(?:cd|export|set|source|umask)\b/u.test(value)) ?? segments[0] ?? text; + const normalized = candidate + .replace(/^(?:[A-Za-z_][A-Za-z0-9_]*=[^\s]+\s+)+/u, "") + .replace(/^(?:sudo\s+|env\s+|command\s+)+/u, "") + .replace(/^timeout\s+\S+\s+/u, ""); + const match = normalized.match(/^([A-Za-z./][A-Za-z0-9_./:-]{0,80})(?:\s|$)/u); + return match?.[1]?.split("/").filter(Boolean).at(-1) ?? null; } function traceToolNameFromPreview(event: TraceEvent): string | null { @@ -234,11 +292,19 @@ function jsonPreviewStringField(text: string | null, key: string): string | null } function traceToolCallBody(event: TraceEvent, command: string | null): string | null { + const stdout = traceEventText(event, ["stdout", "stdoutText", "stdoutSummary"]); + const stderr = traceEventText(event, ["stderr", "stderrText", "stderrSummary"]); + const output = traceEventText(event, ["output", "outputText", "outputSummary"]); + const distinctOutput = output && output !== stdout && output !== stderr ? output : null; + const outputBytes = numberOrNull(event.outputBytes); const lines = [ command, - traceDetailBlock("stdout", traceEventText(event, ["stdout", "stdoutText", "stdoutSummary"])), - traceDetailBlock("stderr", traceEventText(event, ["stderr", "stderrText", "stderrSummary"])), - command ? null : traceEventText(event, ["output", "outputText", "outputSummary"]), + traceDetailBlock("stdout", stdout), + traceDetailBlock("stderr", stderr), + traceDetailBlock("output", distinctOutput), + event.commandTruncated === true ? "commandTruncated=true" : null, + outputBytes !== null ? `outputBytes=${outputBytes}` : null, + event.outputTruncated === true ? "outputTruncated=true" : null, event.exitCode !== undefined ? `exitCode=${event.exitCode}` : null ].map(cleanTraceDetailText).filter(Boolean); return lines.length > 0 ? lines.join("\n") : null; @@ -634,6 +700,7 @@ function compactTraceOneLine(value: unknown, limit = 220): string { } function numberOrNull(value: unknown): number | null { + if (value === null || value === undefined || value === "") return null; const number = Number(value); return Number.isInteger(number) ? number : null; } diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 087fb3d7..2f731b4b 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -7,6 +7,7 @@ import { createCoalescedEventQueue } from "@/utils/scheduler/coalesced-event-que import { composeWorkbenchScopedKey, firstScopePart } from "@/utils/workbench-key"; import { recordWorkbenchRuntimeDiagnostic, recordWorkbenchSseLifecycle } from "@/utils/workbench-performance"; import type { WorkbenchRealtimeCapabilities } from "@/config/runtime"; +import { decodeWorkbenchRealtimeEventFrame } from "./workbench-realtime-codec"; export interface WorkbenchRealtimeTraceSnapshot { traceId?: string | null; @@ -120,7 +121,7 @@ export interface WorkbenchEventStream { export interface WorkbenchSseIngressFrame { eventName: string; rawText: string; - decodeStatus: "accepted" | "invalid-json" | "not-object"; + decodeStatus: import("./workbench-realtime-codec").WorkbenchRealtimeDecodeStatus; } interface QueuedRealtimeEvent { @@ -251,20 +252,9 @@ function appendNumberParam(params: URLSearchParams, key: string, value: number | function enqueueRealtimeFrame(raw: unknown, eventName: string, options: WorkbenchEventStreamOptions, queue: ReturnType>): void { const rawText = typeof raw === "string" ? raw : String(raw ?? ""); - const decoded = decodeRealtimeEvent(rawText); + const decoded = decodeWorkbenchRealtimeEventFrame(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" && !Array.isArray(value) - ? { payload: value as WorkbenchRealtimeEvent, status: "accepted" } - : { payload: null, status: "not-object" }; - } catch { - return { payload: null, status: "invalid-json" }; - } + if (decoded.payload) queue.push({ payload: decoded.payload as WorkbenchRealtimeEvent, eventName }); } function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void { diff --git a/web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts b/web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts new file mode 100644 index 00000000..cb139581 --- /dev/null +++ b/web/hwlab-cloud-web/src/api/workbench-realtime-codec.ts @@ -0,0 +1,17 @@ +export type WorkbenchRealtimeDecodeStatus = "accepted" | "invalid-json" | "not-object"; + +export interface WorkbenchRealtimeDecodedFrame { + payload: Record | null; + status: WorkbenchRealtimeDecodeStatus; +} + +export function decodeWorkbenchRealtimeEventFrame(raw: string): WorkbenchRealtimeDecodedFrame { + try { + const value = JSON.parse(raw); + return value && typeof value === "object" && !Array.isArray(value) + ? { payload: value as Record, status: "accepted" } + : { payload: null, status: "not-object" }; + } catch { + return { payload: null, status: "invalid-json" }; + } +} diff --git a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue index 5e3387f0..018c9c53 100644 --- a/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue +++ b/web/hwlab-cloud-web/src/components/agent/TraceTimeline.vue @@ -6,9 +6,16 @@ import MessageMarkdown from "@/components/workbench/MessageMarkdown.vue"; import { useWorkbenchBottomFollowScroll } from "@/composables/useWorkbenchScrollRuntime"; import { formatDisplayClock } from "@/config/runtime"; import { readWorkbenchString, writeWorkbenchString } from "@/utils/workbench-storage-runtime"; -import { traceDisplayRows, type TraceEventRow } from "../../../../../tools/src/hwlab-cli/trace-renderer.ts"; +import { traceDisplayRows, type TraceEventRow, type TraceFinalResponsePlacement } from "../../../../../tools/src/hwlab-cli/trace-renderer.ts"; -const props = defineProps<{ trace?: RunnerTrace | null; defaultExpanded?: boolean; storageKey?: string; autoExpanded?: boolean | null }>(); +const props = defineProps<{ + trace?: RunnerTrace | null; + defaultExpanded?: boolean; + storageKey?: string; + autoExpanded?: boolean | null; + finalResponsePlacement?: TraceFinalResponsePlacement; + outerFinalResponseText?: string | null; +}>(); const TRACE_RENDER_ROW_WINDOW = 120; const listRef = ref(null); @@ -17,7 +24,11 @@ const { following, keepBottomAfterUpdate, onScroll, scrollToBottom } = useWorkbe const events = computed(() => Array.isArray(props.trace?.events) ? props.trace.events : []); const eventCount = computed(() => props.trace?.eventCount ?? events.value.length); -const rawReadableRows = computed(() => expanded.value ? traceDisplayRows(traceRecord(props.trace), events.value as Record[], { formatClock: formatDisplayClock }) : []); +const rawReadableRows = computed(() => expanded.value ? traceDisplayRows(traceRecord(props.trace), events.value as Record[], { + formatClock: formatDisplayClock, + finalResponsePlacement: props.finalResponsePlacement, + outerFinalResponseText: props.outerFinalResponseText +}) : []); const readableRows = computed(() => rawReadableRows.value); const hiddenRowCount = computed(() => Math.max(0, readableRows.value.length - TRACE_RENDER_ROW_WINDOW)); const renderedRows = computed(() => hiddenRowCount.value > 0 ? readableRows.value.slice(-TRACE_RENDER_ROW_WINDOW) : readableRows.value); diff --git a/web/hwlab-cloud-web/src/components/workbench/MessageTraceDebugPanel.vue b/web/hwlab-cloud-web/src/components/workbench/MessageTraceDebugPanel.vue index fa268c33..455bba34 100644 --- a/web/hwlab-cloud-web/src/components/workbench/MessageTraceDebugPanel.vue +++ b/web/hwlab-cloud-web/src/components/workbench/MessageTraceDebugPanel.vue @@ -16,7 +16,11 @@ const { following, keepBottomAfterUpdate, onScroll, scrollToBottom, toggleFollow const trace = computed(() => props.message.runnerTrace ?? null); const events = computed(() => Array.isArray(trace.value?.events) ? trace.value.events : []); const eventCount = computed(() => trace.value?.eventCount ?? events.value.length); -const readableRows = computed(() => traceDisplayRows(traceRecord(trace.value), events.value as Record[], { formatClock: formatDisplayClock })); +const readableRows = computed(() => traceDisplayRows(traceRecord(trace.value), events.value as Record[], { + formatClock: formatDisplayClock, + finalResponsePlacement: "outer", + outerFinalResponseText: firstNonEmptyString(props.message.text) +})); const noiseCount = computed(() => traceNoiseEventCount(events.value as Record[])); const lastEvent = computed(() => events.value.at(-1) ?? null); const lastReadableRow = computed(() => readableRows.value.at(-1) ?? null); diff --git a/web/hwlab-cloud-web/src/components/workbench/WorkbenchMessageCard.vue b/web/hwlab-cloud-web/src/components/workbench/WorkbenchMessageCard.vue index b887d267..07fa0beb 100644 --- a/web/hwlab-cloud-web/src/components/workbench/WorkbenchMessageCard.vue +++ b/web/hwlab-cloud-web/src/components/workbench/WorkbenchMessageCard.vue @@ -145,7 +145,7 @@ function traceStorageKey(message: ChatMessage): string { - + diff --git a/web/hwlab-cloud-web/src/composables/workbench-trace-snapshot.ts b/web/hwlab-cloud-web/src/composables/workbench-trace-snapshot.ts index 47c14a4b..86f42ef5 100644 --- a/web/hwlab-cloud-web/src/composables/workbench-trace-snapshot.ts +++ b/web/hwlab-cloud-web/src/composables/workbench-trace-snapshot.ts @@ -1,8 +1,8 @@ // SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010401 Web工作台 draft-2026-06-18-r1. // Responsibility: Trace snapshot helpers for Workbench realtime authority reducers. -import type { AgentChatResultResponse, AgentRunProvenance, ChatMessage, ProjectionDiagnostic, TraceEvent, WorkbenchTurnTimingProjection } from "@/types"; -import { firstNonEmptyString } from "@/utils"; +import type { AgentChatResultResponse, AgentRunProvenance, ChatMessage, ProjectionDiagnostic, TraceEvent, WorkbenchTurnTimingProjection } from "../types"; +import { firstNonEmptyString } from "../utils"; export interface TraceSnapshot { traceId?: string; diff --git a/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts b/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts index 2c81c088..e9984b5c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-event-reducer.ts @@ -3,8 +3,8 @@ // Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:21-48 global event classification, :93-270 session/message reducer action boundary. // Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes. -import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; -import { firstNonEmptyString } from "@/utils"; +import type { WorkbenchRealtimeEvent } from "../api/workbench-events"; +import { firstNonEmptyString } from "../utils"; import { workbenchRealtimePrimaryAuthorityDecision } from "./workbench-realtime-authority"; export type WorkbenchRealtimeAction = diff --git a/web/hwlab-cloud-web/src/stores/workbench-isolated-kafka-debug.ts b/web/hwlab-cloud-web/src/stores/workbench-isolated-kafka-debug.ts index 1fa059c7..1fdce381 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-isolated-kafka-debug.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-isolated-kafka-debug.ts @@ -3,13 +3,11 @@ import type { WorkbenchKafkaSseDebugEvent } from "@/api/workbench-debug"; import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; -import { mergeRunnerTrace, snapshotToRunnerTrace } from "@/composables/workbench-trace-snapshot"; -import type { ChatMessage, TraceEvent } from "@/types"; +import type { ChatMessage } from "@/types"; import { firstNonEmptyString } from "@/utils"; import { reduceWorkbenchRealtimeEvent } from "./workbench-event-reducer"; -import { reduceWorkbenchLiveKafkaMessageState, workbenchLiveKafkaMessageId } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event"; import { planWorkbenchRealtimeApply } from "./workbench-realtime-plan"; -import { realtimeSnapshotToTraceSnapshot } from "./workbench-trace-detail"; export interface WorkbenchIsolatedKafkaDebugLog { id: string; @@ -57,7 +55,14 @@ export function applyWorkbenchIsolatedKafkaDebugEvent( if (!traceStep || traceStep.type !== "apply-trace-event" || !traceStep.event) { return withLog(base, event, traceId, reduced.action.type, plan.steps.map((step) => step.type), false, reduced.action.type === "ignore" ? reduced.action.reason : "debug-envelope-not-a-trace-event"); } - const message = projectDebugMessage(state.message, traceId, event, traceStep.event); + const message = projectWorkbenchLiveKafkaMessage({ + previous: state.message, + traceId, + sessionId: firstNonEmptyString(event.hwlabSessionId, event.sessionId, traceStep.event.sessionId, state.message?.sessionId) ?? "ses_workbench_isolated_debug", + event: traceStep.event, + receivedAt: new Date().toISOString(), + title: "Code Agent · 隔离调试" + }); return withLog({ ...base, message, appliedCount: state.appliedCount + 1, error: null }, event, traceId, reduced.action.type, plan.steps.map((step) => step.type), true, null); } @@ -72,54 +77,6 @@ export function workbenchCurrentDebugTraceId(messages: ChatMessage[], sessionLas ) ?? null; } -function projectDebugMessage(previous: ChatMessage | null, traceId: string, realtimeEvent: WorkbenchRealtimeEvent, event: TraceEvent): ChatMessage { - const liveState = reduceWorkbenchLiveKafkaMessageState({ - text: previous?.text ?? "", - status: previous?.status ?? "running", - terminal: false - }, event); - const sessionId = firstNonEmptyString(realtimeEvent.hwlabSessionId, realtimeEvent.sessionId, event.sessionId, previous?.sessionId) ?? "ses_workbench_isolated_debug"; - const updatedAt = firstNonEmptyString(event.createdAt, event.updatedAt, realtimeEvent.eventCreatedAt, realtimeEvent.serverSentAt) ?? new Date().toISOString(); - const startedAt = firstNonEmptyString(previous?.timing?.startedAt, updatedAt) ?? updatedAt; - const snapshot = realtimeSnapshotToTraceSnapshot(traceId, { - traceId, - sessionId, - status: liveState.status, - events: [event], - eventCount: 1, - startedAt, - lastEventAt: updatedAt, - finishedAt: liveState.terminal ? updatedAt : null, - durationMs: finiteNumber(event.elapsedMs) - }, [event], { eventSource: "hwlab-kafka-sse" }); - const runnerTrace = mergeRunnerTrace(previous?.runnerTrace ?? null, snapshotToRunnerTrace(snapshot)); - const messageId = workbenchLiveKafkaMessageId(traceId); - return { - ...(previous ?? {}), - id: previous?.id ?? messageId, - messageId: previous?.messageId ?? messageId, - role: "agent", - title: "Code Agent · 隔离调试", - text: liveState.text, - status: liveState.status as ChatMessage["status"], - traceId, - turnId: traceId, - sessionId, - createdAt: previous?.createdAt ?? startedAt, - updatedAt, - runnerTrace, - traceAutoLifecycle: liveState.terminal ? "terminal" : "running", - timing: { - startedAt, - lastEventAt: updatedAt, - finishedAt: liveState.terminal ? updatedAt : null, - durationMs: liveState.terminal ? finiteNumber(event.elapsedMs) : null, - valuesRedacted: true - }, - ...(liveState.terminal && liveState.text ? { finalResponse: { text: liveState.text, status: liveState.status, traceId } } : {}) - } as ChatMessage; -} - function rejected(state: WorkbenchIsolatedKafkaDebugState, event: WorkbenchRealtimeEvent, traceId: string | null, reason: string): WorkbenchIsolatedKafkaDebugState { return withLog(state, event, traceId, "ignore", [], false, reason); } @@ -154,8 +111,3 @@ function realtimeTraceId(event: WorkbenchRealtimeEvent): string | null { function recordValue(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } - -function finiteNumber(value: unknown): number | null { - const number = Number(value); - return Number.isFinite(number) && number >= 0 ? Math.trunc(number) : null; -} diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts new file mode 100644 index 00000000..a0d54b38 --- /dev/null +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts @@ -0,0 +1,91 @@ +import assert from "node:assert/strict"; +import { test } from "bun:test"; + +import type { ChatMessage, TraceEvent } from "../types"; +import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event"; + +test("live Kafka projection refreshes lastEventAt from each ingress receipt", () => { + const traceId = "trc_live_timing"; + const sessionId = "ses_live_timing"; + const firstReceivedAt = "2026-07-10T10:00:05.000Z"; + const secondReceivedAt = "2026-07-10T10:00:09.000Z"; + const first = projectWorkbenchLiveKafkaMessage({ + previous: null, + traceId, + sessionId, + receivedAt: firstReceivedAt, + event: traceEvent(1, "tool") + }); + const second = projectWorkbenchLiveKafkaMessage({ + previous: first, + traceId, + sessionId, + receivedAt: secondReceivedAt, + event: traceEvent(2, "assistant", { assistantText: "visible response" }) + }); + + assert.equal(first.lastEventAt, firstReceivedAt); + assert.equal(second.lastEventAt, secondReceivedAt); + assert.equal(second.timing?.lastEventAt, secondReceivedAt); + assert.equal(second.timing?.lastEventAgeMs, 0); + assert.equal(second.runnerTrace?.lastEventAt, secondReceivedAt); + assert.equal(second.runnerTrace?.timing?.lastEventAt, secondReceivedAt); + assert.equal(second.runnerTrace?.eventCount, 2); +}); + +test("live Kafka projection clears stale terminal timing when a running event is accepted", () => { + const traceId = "trc_live_reopened"; + const sessionId = "ses_live_reopened"; + const terminalAt = "2026-07-10T10:00:10.000Z"; + const terminal = projectWorkbenchLiveKafkaMessage({ + previous: seededMessage(traceId, sessionId), + traceId, + sessionId, + receivedAt: terminalAt, + event: traceEvent(2, "result", { terminal: true, status: "completed" }) + }); + const acceptedAt = "2026-07-10T10:00:12.000Z"; + const running = projectWorkbenchLiveKafkaMessage({ + previous: terminal, + traceId, + sessionId, + receivedAt: acceptedAt, + event: traceEvent(3, "tool") + }); + + assert.equal(terminal.finishedAt, terminalAt); + const terminalFinalResponse = terminal.finalResponse as { text?: string } | null | undefined; + assert.equal(terminalFinalResponse?.text, "final answer"); + assert.equal(running.status, "running"); + assert.equal(running.lastEventAt, acceptedAt); + assert.equal(running.finishedAt, null); + assert.equal(running.durationMs, null); + assert.equal(running.finalResponse, null); + assert.equal(running.runnerTrace?.finishedAt, null); + assert.equal(running.runnerTrace?.durationMs, null); +}); + +function seededMessage(traceId: string, sessionId: string): ChatMessage { + return projectWorkbenchLiveKafkaMessage({ + previous: null, + traceId, + sessionId, + receivedAt: "2026-07-10T10:00:01.000Z", + event: traceEvent(1, "assistant", { assistantText: "final answer" }) + }); +} + +function traceEvent(sourceSeq: number, type: string, extra: Record = {}): TraceEvent { + return { + traceId: "trc_live_timing", + sessionId: "ses_live_timing", + source: "agentrun.kafka", + sourceEventId: `evt_live_${sourceSeq}`, + sourceSeq, + type, + eventType: type, + status: "running", + createdAt: `2026-07-10T10:00:0${sourceSeq}.000Z`, + ...extra + }; +} diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts index baf2343c..7a1a1d81 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts @@ -1,8 +1,9 @@ // SPEC: PJ2026-0104010803 Workbench live Kafka SSE. // Responsibility: project one transparent hwlab.event.v1 business event into visible turn state without replay/finalizer/polling. -import type { TraceEvent } from "@/types"; -import { firstNonEmptyString } from "@/utils"; +import { mergeRunnerTrace } from "../composables/workbench-trace-snapshot"; +import type { ChatMessage, TraceEvent, WorkbenchTurnTimingProjection } from "../types"; +import { firstNonEmptyString } from "../utils"; export interface WorkbenchLiveMessageState { text: string; @@ -10,6 +11,15 @@ export interface WorkbenchLiveMessageState { terminal: boolean; } +export interface WorkbenchLiveKafkaProjectionInput { + previous: ChatMessage | null; + traceId: string; + sessionId: string; + event: TraceEvent; + receivedAt: string; + title?: string; +} + export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent): WorkbenchLiveMessageState { const terminal = workbenchLiveKafkaEventIsTerminal(event); const status = terminal ? firstNonEmptyString(event.status, "completed") ?? "completed" : "running"; @@ -45,6 +55,87 @@ export function workbenchLiveKafkaTerminalText(event: TraceEvent, previousText: return text ?? `AgentRun ${status}`; } +export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjectionInput): ChatMessage { + const receivedAt = timestamp(input.receivedAt); + if (!receivedAt) throw new Error("receivedAt must be a valid timestamp for live Kafka projection"); + const previous = input.previous; + const liveState = reduceWorkbenchLiveKafkaMessageState({ + text: previous?.text ?? "", + status: previous?.status ?? "running", + terminal: false + }, input.event); + const startedAt = timestamp(previous?.timing?.startedAt) + ?? timestamp(previous?.startedAt) + ?? timestamp(input.event.createdAt) + ?? receivedAt; + const durationMs = liveState.terminal ? Math.max(0, Date.parse(receivedAt) - Date.parse(startedAt)) : null; + const timing: WorkbenchTurnTimingProjection = { + startedAt, + lastEventAt: receivedAt, + finishedAt: liveState.terminal ? receivedAt : null, + durationMs, + observedAt: receivedAt, + lastEventAgeMs: 0, + valuesRedacted: true + }; + const mergedRunnerTrace = mergeRunnerTrace(previous?.runnerTrace ?? null, { + traceId: input.traceId, + sessionId: input.sessionId, + status: liveState.status, + traceStatus: liveState.status, + events: [input.event], + eventCount: 1, + eventSource: "hwlab-kafka-sse", + timing, + startedAt, + lastEventAt: receivedAt, + finishedAt: liveState.terminal ? receivedAt : null, + durationMs, + updatedAt: receivedAt + }); + const runnerTrace = { + ...mergedRunnerTrace, + status: liveState.status, + traceStatus: liveState.status, + timing, + startedAt, + lastEventAt: receivedAt, + finishedAt: liveState.terminal ? receivedAt : null, + durationMs, + updatedAt: receivedAt + }; + const messageId = previous?.messageId ?? previous?.id ?? workbenchLiveKafkaMessageId(input.traceId); + return { + ...(previous ?? {}), + id: previous?.id ?? messageId, + messageId, + role: "agent", + title: input.title ?? previous?.title ?? "Code Agent", + text: liveState.text, + status: liveState.status as ChatMessage["status"], + traceId: input.traceId, + turnId: previous?.turnId ?? input.traceId, + sessionId: input.sessionId, + createdAt: previous?.createdAt ?? startedAt, + updatedAt: receivedAt, + startedAt, + lastEventAt: receivedAt, + finishedAt: liveState.terminal ? receivedAt : null, + durationMs, + runnerTrace, + traceAutoLifecycle: liveState.terminal ? "terminal" : "running", + timing, + finalResponse: liveState.terminal && liveState.text + ? { text: liveState.text, status: liveState.status, traceId: input.traceId } + : null + } as ChatMessage; +} + +function timestamp(value: unknown): string | null { + const text = firstNonEmptyString(value); + return text && Number.isFinite(Date.parse(text)) ? text : null; +} + function recordValue(value: unknown): Record | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : null; } diff --git a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts index 2041c489..03f00d50 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-realtime-plan.ts @@ -3,9 +3,9 @@ // Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:93-270 session/message reducer action boundary. // Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes. -import type { WorkbenchRealtimeEvent } from "@/api/workbench-events"; -import type { WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime"; -import { firstNonEmptyString, normalizeWorkbenchSessionId } from "@/utils"; +import type { WorkbenchRealtimeEvent } from "../api/workbench-events"; +import { firstNonEmptyString, normalizeWorkbenchSessionId } from "../utils"; +import type { WorkbenchStreamTransportRecovery } from "../utils/workbench-realtime-runtime"; import type { WorkbenchRealtimeAction } from "./workbench-event-reducer"; export type WorkbenchRealtimeApplyStep = diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 8f54590f..831a7e8c 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -22,7 +22,7 @@ import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection"; import { cleanupWorkbenchServerStateSessions, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache"; import { reduceWorkbenchRealtimeEvent, workbenchRealtimeEventIsBusinessActivity, type WorkbenchRealtimeAction } from "./workbench-event-reducer"; -import { reduceWorkbenchLiveKafkaMessageState, workbenchLiveKafkaEventIsTerminal, workbenchLiveKafkaMessageId } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event"; import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority"; import { boundedProjectionMessageLimit, mergeBoundedProjectionMessages, selectProjectionMessageWindow, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget"; import { @@ -1360,56 +1360,52 @@ export const useWorkbenchStore = defineStore("workbench", () => { } const events = event ? [event] : Array.isArray(snapshot?.events) ? snapshot.events : []; markWorkbenchTraceEventsReceived({ traceId: id, events, transport: "sse", serverSentAt: realtimeEvent?.serverSentAt, eventCreatedAt: realtimeEvent?.eventCreatedAt, traceSeq: realtimeEvent?.traceSeq ?? realtimeEvent?.cursor?.traceSeq }); - const liveTerminal = liveKafkaEnvelope && workbenchLiveKafkaEventIsTerminal(event); - const eventSnapshot = snapshot ?? { traceId: id, status: liveKafkaEnvelope && !liveTerminal ? "running" : event?.status, events }; - if (liveKafkaEnvelope && event) applyLiveKafkaBusinessEvent(id, sessionId, event); - applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, eventSnapshot, events, liveKafkaEnvelope ? { eventSource: "hwlab-kafka-sse" } : undefined), liveKafkaEnvelope); + if (liveKafkaEnvelope && event) { + applyLiveKafkaBusinessEvent(id, sessionId, event, new Date().toISOString()); + return; + } + const eventSnapshot = snapshot ?? { traceId: id, status: event?.status, events }; + applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, eventSnapshot, events)); } - function applyLiveKafkaBusinessEvent(traceId: string, authoritySessionId: string | null, event: TraceEvent): void { + function applyLiveKafkaBusinessEvent(traceId: string, authoritySessionId: string | null, event: TraceEvent, receivedAt: string): void { const eventSessionId = firstNonEmptyString(event.sessionId); const ownerSessionId = traceOwnerSessionId(traceId, authoritySessionId ?? eventSessionId, true); if (!ownerSessionId) return; const previousMessage = (serverState.value.messagesBySessionId[ownerSessionId] ?? []).find((message) => messageMatchesTraceAuthority(message, traceId, authoritySessionId ?? eventSessionId, ownerSessionId, true)) ?? null; - const liveState = reduceWorkbenchLiveKafkaMessageState({ text: previousMessage?.text ?? "", status: previousMessage?.status ?? "running", terminal: false }, event); - const terminal = liveState.terminal; - const status = liveState.status; - const assistantText = liveState.text; - const updatedAt = firstNonEmptyString(event.createdAt, event.updatedAt) ?? new Date().toISOString(); - const messageId = workbenchLiveKafkaMessageId(traceId); - const message = previousMessage ?? makeMessage("agent", "", "running", { - id: messageId, - messageId, + const message = projectWorkbenchLiveKafkaMessage({ + previous: previousMessage, traceId, sessionId: ownerSessionId, - title: "Code Agent", - traceAutoLifecycle: "running", - createdAt: updatedAt, - updatedAt + event, + receivedAt }); reduceServerState({ type: "message.upsert", sessionId: ownerSessionId, - message: { - ...message, - ...(assistantText ? { text: assistantText } : {}), - status: status as ChatMessage["status"], - traceAutoLifecycle: terminal ? "terminal" : "running", - ...(terminal && assistantText ? { finalResponse: { text: assistantText, status, traceId } } : {}), - updatedAt - } + message }); + if (message.runnerTrace) rememberTraceAuthority(message.runnerTrace); + markWorkbenchTraceProjected(traceId); + const terminal = message.traceAutoLifecycle === "terminal"; + const status = firstNonEmptyString(message.status) ?? "running"; + const finalResponse = terminal && message.text ? { text: message.text, status, traceId } : undefined; rememberTurnStatus(traceId, { traceId, sessionId: ownerSessionId, status, running: !terminal, terminal, - finalResponse: terminal && assistantText ? { text: assistantText, status, traceId } : undefined, - updatedAt + finalResponse, + timing: message.timing, + startedAt: message.timing?.startedAt, + lastEventAt: message.timing?.lastEventAt, + finishedAt: message.timing?.finishedAt, + durationMs: message.timing?.durationMs, + updatedAt: receivedAt } as AgentChatResultResponse); const existing = sessions.value.find((session) => session.sessionId === ownerSessionId) ?? null; - if (existing) rememberSessionList(mergeSessionIntoList(sessions.value, { ...existing, status, lastTraceId: traceId, updatedAt: firstNonEmptyString(event.createdAt, event.updatedAt) ?? new Date().toISOString() })); + if (existing) rememberSessionList(mergeSessionIntoList(sessions.value, { ...existing, status, lastTraceId: traceId, updatedAt: receivedAt })); if (terminal && currentRequest.value?.traceId === traceId) { chatPending.value = false; currentRequest.value = null;