From 55f85a164d0e219b648c3639f6a1d3fa1463bf49 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 11 Jul 2026 00:27:24 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=9D=E6=8C=81=E7=BA=AF=20Kafka=20?= =?UTF-8?q?=E7=BB=88=E6=80=81=E4=B8=8E=E6=89=AB=E6=8F=8F=E5=AE=8C=E6=95=B4?= =?UTF-8?q?=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/reference/kafka-source-direct-debug.md | 3 + internal/cloud/kafka-event-bridge.ts | 4 +- tools/hwlab-cli/kafka-regenerate.test.ts | 132 +++++++++++++++++- tools/src/hwlab-cli/kafka-regenerate.ts | 91 +++++++++--- .../stores/workbench-live-kafka-event.test.ts | 25 ++-- .../src/stores/workbench-live-kafka-event.ts | 28 +++- web/hwlab-cloud-web/src/stores/workbench.ts | 6 +- 7 files changed, 248 insertions(+), 41 deletions(-) diff --git a/docs/reference/kafka-source-direct-debug.md b/docs/reference/kafka-source-direct-debug.md index 4dbe81cf..a7619c62 100644 --- a/docs/reference/kafka-source-direct-debug.md +++ b/docs/reference/kafka-source-direct-debug.md @@ -34,6 +34,8 @@ - `completionReason=end-offset` 表示已经扫描到命令启动时捕获的 Kafka 上界； - `completionReason=limit` 表示达到显式事件上限，不能冒充完整 trace； - `completionReason=timeout` 表示未在 YAML 或 CLI 时限内证明到达上界； + - 未到捕获上界时，Trace 命令必须返回 `status=partial`、`ok=false`、`source_scan_incomplete` 和非零退出码，同时保留有界 Markdown artifact 供下钻； + - 默认文本首词必须是 `partial`，并披露 `completionReason` 与 `reachedEndOffsets`，禁止用 `ok` 或 `succeeded` 表示部分扫描； - 没有匹配事件时以 `source_trace_missing` 失败，并保留上述扫描计数。 - Trace Markdown artifact 默认写入 `.state/hwlab-cli/kafka-render/`： @@ -64,6 +66,7 @@ - Workbench 消费 debug topic 由 `HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED` 独立控制； - CLI 的离线映射不依赖该运行面开关； - Kafka 输入默认只做映射预检和本地证据固化； + - Kafka 输入未扫描到命令启动时捕获的 topic 上界时返回 `partial/source_scan_incomplete`，保留 artifact，但即使指定 `--publish` 也不得调用 producer； - 只有显式 `--publish` 才发布到独立 HWLAB debug topic。 - JSONL 输入适合完全离线的单步测试： diff --git a/internal/cloud/kafka-event-bridge.ts b/internal/cloud/kafka-event-bridge.ts index af6147f4..0db463ba 100644 --- a/internal/cloud/kafka-event-bridge.ts +++ b/internal/cloud/kafka-event-bridge.ts @@ -990,10 +990,10 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab }); } } - 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"); + if (endOffsetSnapshot.available && reachedEndPartitions.size === endOffsetByPartition.size) return finish("end-offset"); + if (events.length >= maxEvents) finish("limit"); } }).catch(reject); }); diff --git a/tools/hwlab-cli/kafka-regenerate.test.ts b/tools/hwlab-cli/kafka-regenerate.test.ts index ed6473fe..0b0ddce8 100644 --- a/tools/hwlab-cli/kafka-regenerate.test.ts +++ b/tools/hwlab-cli/kafka-regenerate.test.ts @@ -6,7 +6,7 @@ import path from "node:path"; 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 { mapAgentRunRecordsToHwlabDebugEvents, renderKafkaCliText, 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"; @@ -307,6 +307,57 @@ test("missing commandId cannot be satisfied by commandless lifecycle events from assert.equal(result.payload.error.details.commandScope.selectedUnscopedLifecycleCount, 0); }); +test("Kafka Trace render returns a typed partial result when the captured end offsets were not reached", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-partial-")); + const record = hwlabRecord(1, { + type: "assistant", + eventType: "assistant", + status: "running", + itemId: "assistant_partial", + assistantText: "Partial trace row" + }); + 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", + "--format", "markdown" + ], { + cwd, + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka() { + return { + events: [record], + scannedCount: 1, + parsedCount: 1, + matchedCount: 1, + filterRejectedCount: 0, + completionReason: "timeout", + reachedEndOffsets: false, + endOffsetsAvailable: true, + endOffsets: [{ partition: 0, startOffset: "0", endOffset: "2" }], + lastScannedOffsets: [{ partition: 0, offset: "0" }] + }; + } + }); + + assert.equal(result.exitCode, 2); + assert.equal(result.payload.ok, false); + assert.equal(result.payload.status, "partial"); + assert.equal(result.payload.error.code, "source_scan_incomplete"); + assert.equal(result.payload.input.scanComplete, false); + assert.equal(result.payload.input.completionReason, "timeout"); + assert.equal(result.payload.validation.sourceScanComplete, false); + assert.equal(result.payload.validation.commandLifecyclePreserved, false); + assert.match(result.markdownOutput ?? "", /source scan: partial \(timeout\)/u); + assert.equal(typeof result.payload.output.artifact.path, "string"); + assert.match(renderKafkaCliText(result.payload), /^partial action=kafka\.render\.trace code=source_scan_incomplete completion=timeout reachedEnd=false/u); +}); + test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug events without runtime dependencies", async () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-jsonl-")); const inputFile = path.join(cwd, "agentrun-events.jsonl"); @@ -338,6 +389,7 @@ test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug even assert.equal(result.payload.output.publishedCount, 0); assert.deepEqual(result.payload.validation, { expectedCount: 35, + sourceScanComplete: true, oneToOne: true, orderPreserved: true, lineagePreserved: true, @@ -400,7 +452,7 @@ test("Kafka reader and producer are injected while canonical AgentRun events use now: () => "2026-07-10T12:00:00.000Z", async readKafka(input) { readerCalls.push(input); - return { ok: true, groupId: "hwlab-v03-agentrun-event-debug-regenerate-test", events: records }; + return { ok: true, groupId: "hwlab-v03-agentrun-event-debug-regenerate-test", events: records, completionReason: "end-offset", reachedEndOffsets: true }; }, async createProducer() { return { @@ -487,7 +539,7 @@ test("canonical Kafka regeneration preflights without publishing and preserves r cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z", - async readKafka() { return { events: [canonicalRecord(1)] }; }, + async readKafka() { return { events: [canonicalRecord(1)], completionReason: "end-offset", reachedEndOffsets: true }; }, async createProducer() { producerCreated = true; throw new Error("preflight must not create a producer"); } }); @@ -503,6 +555,47 @@ test("canonical Kafka regeneration preflights without publishing and preserves r assert.equal(producerCreated, false); }); +test("partial AgentRun Kafka regeneration preserves the artifact and never invokes the producer", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-partial-")); + let producerCreated = false; + const result = await runKafkaCli([ + "regenerate", "hwlab", + "--from", "kafka", + "--session-id", SESSION_ID, + "--trace-id", TRACE_ID, + "--replay-id", "rpl_partial_no_publish", + "--group-prefix", "hwlab-v03-workbench-isolated-debug", + "--publish", + "--output-jsonl", path.join(cwd, "partial.jsonl") + ], { + cwd, + env: {}, + now: () => "2026-07-10T12:00:00.000Z", + async readKafka() { + return { + events: [canonicalRecord(1)], + scannedCount: 1, + parsedCount: 1, + matchedCount: 1, + completionReason: "limit", + reachedEndOffsets: false + }; + }, + async createProducer() { producerCreated = true; throw new Error("partial scan must not create producer"); } + }); + + assert.equal(result.exitCode, 2); + assert.equal(result.payload.status, "partial"); + assert.equal(result.payload.error.code, "source_scan_incomplete"); + assert.equal(result.payload.input.scanComplete, false); + assert.equal(result.payload.output.published, false); + assert.equal(result.payload.output.publishedCount, 0); + assert.equal(result.payload.replay.producerInvoked, false); + assert.equal(producerCreated, false); + assert.equal(typeof result.payload.output.artifact.path, "string"); + assert.match(renderKafkaCliText(result.payload), /^partial action=kafka\.regenerate\.hwlab code=source_scan_incomplete completion=limit reachedEnd=false/u); +}); + test("Kafka regeneration fails closed when publish metadata cannot form an offset barrier", async () => { const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-barrier-")); const result = await runKafkaCli([ @@ -518,7 +611,7 @@ test("Kafka regeneration fails closed when publish metadata cannot form an offse cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z", - async readKafka() { return { events: [canonicalRecord(1)] }; }, + async readKafka() { return { events: [canonicalRecord(1)], completionReason: "end-offset", reachedEndOffsets: true }; }, async createProducer() { return { async connect() {}, @@ -661,6 +754,37 @@ test("Kafka query reports invalid JSON, filter rejection, and bounded end-offset assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "2" }]); }); +test("Kafka query prefers end-offset when the final record also reaches the event limit", async () => { + const matching = canonicalRecord(1).value; + 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: 1, + groupIdPrefix: "hwlab-v03-workbench-isolated-debug", + kafkaFactory() { + return { + admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "1" }]; }, async disconnect() {} }; }, + consumer() { + return { + async connect() {}, + async subscribe() {}, + async run({ eachMessage }: any) { + await eachMessage({ topic: "agentrun.event.v1", partition: 0, message: { offset: "0", value: Buffer.from(JSON.stringify(matching)) } }); + }, + async stop() {}, + async disconnect() {} + }; + } + }; + } + }); + + assert.equal(queried.matchedCount, 1); + assert.equal(queried.completionReason, "end-offset"); + assert.equal(queried.reachedEndOffsets, true); +}); + test("Kafka query treats a retained empty partition as a completed bounded scan", async () => { let runCalled = false; let disconnected = false; diff --git a/tools/src/hwlab-cli/kafka-regenerate.ts b/tools/src/hwlab-cli/kafka-regenerate.ts index 049125fe..ef26e42e 100644 --- a/tools/src/hwlab-cli/kafka-regenerate.ts +++ b/tools/src/hwlab-cli/kafka-regenerate.ts @@ -22,7 +22,7 @@ import { planWorkbenchRealtimeApply } from "../../../web/hwlab-cloud-web/src/sto import { renderTraceRowsMarkdown, traceDisplayRows } from "./trace-renderer.ts"; const CLI_NAME = "hwlab-cli"; -const VERSION = "0.3.3-kafka-command-lifecycle-scope"; +const VERSION = "0.3.4-kafka-partial-scan-terminal-monotonic"; const DEFAULT_LIMIT = 500; const DEFAULT_TIMEOUT_MS = 5000; const DEBUG_OUTPUT_PARTITION = 0; @@ -66,7 +66,7 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) readKafka: options.readKafka ?? defaultKafkaReader, createProducer: options.createProducer ?? defaultProducerFactory }); - return { ...result(0, rendered.payload, now), markdownOutput: rendered.markdown }; + return { ...result(rendered.payload.status === "partial" ? 2 : 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"; @@ -77,7 +77,7 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {}) readKafka: options.readKafka ?? defaultKafkaReader, createProducer: options.createProducer ?? defaultProducerFactory }); - return result(0, payload, now); + return result(payload.status === "partial" ? 2 : 0, payload, now); } catch (error) { return result(1, failure(error, action), now); } @@ -122,6 +122,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie groupIdPrefix: groupPrefix }); const records = Array.isArray(readResult.events) ? readResult.events : []; + const sourceScanComplete = sourceMode === "jsonl" || readResult.completionReason === "end-offset"; const mapped = mapAgentRunRecordsToHwlabDebugEvents(records, { sessionId, traceId, @@ -152,7 +153,8 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie let publishedCount = 0; let publishMetadata: unknown[] = []; - if (shouldPublish) { + const publishAllowed = shouldPublish && sourceScanComplete; + if (publishAllowed) { const producer = await dependencies.createProducer({ env: dependencies.env, clientId: `${groupPrefix}-producer` }); try { await producer.connect?.(); @@ -177,7 +179,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie const first = mapped.events[0]; const last = mapped.events.at(-1); const offsetRanges = publishOffsetRanges(publishMetadata, publishedCount, DEBUG_OUTPUT_PARTITION); - if (shouldPublish && offsetRanges.length !== 1) { + if (publishAllowed && offsetRanges.length !== 1) { throw cliError("debug_publish_barrier_missing", "Kafka publish succeeded without a usable isolated replay offset barrier", { replayId, topic: outputTopic, @@ -192,13 +194,26 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie const baseCommand = sourceMode === "jsonl" ? `hwlab-cli kafka regenerate hwlab --from jsonl --session-id ${sessionId}${traceArgument} --replay-id ${replayId}${groupArgument} --jsonl-file ${shellArg(sourceFile || String(parsed.jsonlFile))}` : `hwlab-cli kafka regenerate hwlab --from kafka --session-id ${sessionId}${traceArgument} --replay-id ${replayId}${groupArgument} --input-topic ${inputTopic}`; - const nextCommand = shouldPublish + const nextCommand = !sourceScanComplete + ? `${baseCommand}${shouldPublish ? " --publish" : " --no-publish"} --output-topic ${outputTopic} --json` + : shouldPublish ? `${baseCommand} --no-publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json` : `${baseCommand} --publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json`; return { - ok: true, + ok: sourceScanComplete, action: "kafka.regenerate.hwlab", - status: "succeeded", + status: sourceScanComplete ? "succeeded" : "partial", + error: sourceScanComplete ? undefined : { + code: "source_scan_incomplete", + message: "Kafka scan ended before the captured AgentRun topic end offsets were reached; debug publish was not attempted", + details: { + completionReason: readResult.completionReason ?? null, + reachedEndOffsets: readResult.reachedEndOffsets ?? false, + publishRequested: shouldPublish, + artifactPath, + valuesPrinted: false + } + }, replayId, sourceMode, config: { @@ -211,6 +226,15 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie groupId: readResult.groupId ?? null, readCount: records.length, matchedCount: mapped.events.length, + scannedCount: readResult.scannedCount ?? records.length, + parsedCount: readResult.parsedCount ?? records.length, + queryMatchedCount: readResult.matchedCount ?? records.length, + scanComplete: sourceScanComplete, + completionReason: readResult.completionReason ?? (sourceMode === "jsonl" ? "file-end" : null), + reachedEndOffsets: readResult.reachedEndOffsets ?? (sourceMode === "jsonl"), + endOffsetsAvailable: readResult.endOffsetsAvailable ?? false, + endOffsets: readResult.endOffsets ?? [], + lastScannedOffsets: readResult.lastScannedOffsets ?? [], sessionId, traceId: traceId || null }, @@ -218,7 +242,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie schema: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, topic: outputTopic, count: mapped.events.length, - published: shouldPublish, + published: publishAllowed, publishedCount, metadataCount: Array.isArray(publishMetadata) ? publishMetadata.length : 0, offsetRanges, @@ -230,6 +254,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie }, validation: { expectedCount: expectedCount ?? null, + sourceScanComplete, oneToOne: mapped.inputCount === mapped.events.length, orderPreserved: mapped.orderPreserved, lineagePreserved: mapped.lineagePreserved, @@ -239,7 +264,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie }, replay: { replayId, - producerInvoked: shouldPublish, + producerInvoked: publishAllowed, sourceMatched: true, sourceLineage: { traceId: first?.debugLineage?.inputTraceId ?? traceId ?? null, @@ -262,11 +287,13 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie cloudApi: false, database: false, transactionalProjector: false, - kafka: sourceMode === "kafka" || shouldPublish + kafka: sourceMode === "kafka" || publishAllowed }, next: { command: nextCommand, - reason: shouldPublish + reason: !sourceScanComplete + ? "源 Kafka 扫描不完整，已保留 artifact 且未调用 producer；先扩大有界扫描并到达捕获上界，再重试发布。" + : shouldPublish ? "用同一源码映射做一次无 Kafka 写入复验；随后 Workbench 调试模式按 traceId 消费独立 debug topic。" : "离线映射已固化为 JSONL；确认内容后可显式 --publish 写入独立 debug topic。" }, @@ -317,7 +344,8 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re }); 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 sourceScanComplete = sourceMode === "jsonl" || readResult.completionReason === "end-offset"; + const commandScopeScanComplete = sourceScanComplete; const matchingRecords = scoped.records; if (matchingRecords.length === 0) { throw cliError(commandId && scoped.summary.baseMatchedCount > 0 ? "source_command_missing" : "source_trace_missing", commandId @@ -372,7 +400,9 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re status: text(projection.message.status), sourceEventCount: events.length, rows, - finalResponseText: finalResponseText || null + finalResponseText: finalResponseText || null, + sourceScanComplete, + completionReason: readResult.completionReason ?? (sourceMode === "jsonl" ? "file-end" : null) }); const stateId = `hwlab-trace-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`; const artifactPath = path.resolve( @@ -411,9 +441,19 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re return { markdown, payload: { - ok: true, + ok: sourceScanComplete, action: "kafka.render.trace", - status: "succeeded", + status: sourceScanComplete ? "succeeded" : "partial", + error: sourceScanComplete ? undefined : { + code: "source_scan_incomplete", + message: "Kafka scan ended before the captured topic end offsets were reached", + details: { + completionReason: readResult.completionReason ?? null, + reachedEndOffsets: readResult.reachedEndOffsets ?? false, + artifactPath, + valuesPrinted: false + } + }, format, input: { mode: sourceMode, @@ -430,6 +470,7 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re filterRejectedCount: readResult.filterRejectedCount ?? 0, scopeRejectedCount: scoped.summary.baseRejectedCount + scoped.summary.commandScopeRejectedCount, commandScope: { ...scoped.summary, scanComplete: commandScopeScanComplete }, + scanComplete: sourceScanComplete, completionReason: readResult.completionReason ?? (sourceMode === "jsonl" ? "file-end" : null), reachedEndOffsets: readResult.reachedEndOffsets ?? (sourceMode === "jsonl"), endOffsetsAvailable: readResult.endOffsetsAvailable ?? false, @@ -494,6 +535,7 @@ export async function renderHwlabKafkaTrace(parsed: ParsedArgs, dependencies: Re expectedCount: expectedCount ?? null, allMatchedEventsApplied: projection.appliedCount === matchingRecords.length, commandLifecyclePreserved: !commandId || (commandScopeScanComplete && scoped.summary.exactCommandCount > 0 && scoped.summary.ambiguousUnscopedLifecycleCount === 0), + sourceScanComplete, finalResponseOutsideTrace: !finalResponseInTrace && Boolean(finalResponseText), sharedRowModel: true, valuesPrinted: false @@ -571,12 +613,13 @@ export function projectHwlabRecordsThroughWorkbench(records: JsonRecord[], optio return { message, decodedCount, plannedCount, appliedCount, rejections }; } -function renderWorkbenchKafkaTraceMarkdown(input: { traceId: string; sessionId: string; status: string; sourceEventCount: number; rows: any[]; finalResponseText: string | null }) { +function renderWorkbenchKafkaTraceMarkdown(input: { traceId: string; sessionId: string; status: string; sourceEventCount: number; rows: any[]; finalResponseText: string | null; sourceScanComplete: boolean; completionReason: string | null }) { const lines = [ `# Workbench Trace ${input.traceId}`, "", `- session: ${input.sessionId || "unknown"}`, `- status: ${input.status || "unknown"}`, + `- source scan: ${input.sourceScanComplete ? "complete" : `partial (${input.completionReason ?? "unknown"})`}`, `- source events: ${input.sourceEventCount}`, `- rendered rows: ${input.rows.length}`, `- final response placement: outer`, @@ -956,7 +999,19 @@ function parseOptions(argv: string[]): ParsedArgs { return out; } -function renderKafkaCliText(payload: Record) { +export function renderKafkaCliText(payload: Record) { + if (payload.status === "partial") return [ + "partial", + `action=${payload.action}`, + `code=${payload.error?.code ?? "source_scan_incomplete"}`, + payload.input?.completionReason ? `completion=${payload.input.completionReason}` : null, + payload.input?.reachedEndOffsets !== undefined ? `reachedEnd=${payload.input.reachedEndOffsets}` : null, + payload.input?.scannedCount !== undefined ? `scanned=${payload.input.scannedCount}` : null, + payload.input?.matchedCount !== undefined ? `matched=${payload.input.matchedCount}` : null, + payload.projection?.appliedCount !== undefined ? `applied=${payload.projection.appliedCount}` : null, + payload.output?.artifact?.path ? `artifact=${shellArg(payload.output.artifact.path)}` : null, + payload.next?.command ? `next=${JSON.stringify(payload.next.command)}` : null + ].filter(Boolean).join(" "); if (payload.ok === false) return `failed action=${payload.action} code=${payload.error?.code} message=${JSON.stringify(payload.error?.message)} next=${JSON.stringify(payload.next?.command)}`; const validation = payload.validation ?? {}; return [ 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 index a0d54b38..1d0d2435 100644 --- 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 @@ -2,7 +2,13 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage, TraceEvent } from "../types"; -import { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, workbenchLiveKafkaEnvelope } from "./workbench-live-kafka-event"; + +test("only canonical HWLAB Kafka envelopes bypass terminal priority", () => { + assert.equal(workbenchLiveKafkaEnvelope("hwlab.event.v1"), true); + assert.equal(workbenchLiveKafkaEnvelope("workbench.trace.event.v1"), false); + assert.equal(workbenchLiveKafkaEnvelope(null), false); +}); test("live Kafka projection refreshes lastEventAt from each ingress receipt", () => { const traceId = "trc_live_timing"; @@ -33,7 +39,7 @@ test("live Kafka projection refreshes lastEventAt from each ingress receipt", () assert.equal(second.runnerTrace?.eventCount, 2); }); -test("live Kafka projection clears stale terminal timing when a running event is accepted", () => { +test("live Kafka projection keeps terminal lifecycle while accepting a later event", () => { const traceId = "trc_live_reopened"; const sessionId = "ses_live_reopened"; const terminalAt = "2026-07-10T10:00:10.000Z"; @@ -56,13 +62,16 @@ test("live Kafka projection clears stale terminal timing when a running event is 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.status, "completed"); 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); + assert.equal(running.timing?.lastEventAgeMs, 0); + assert.equal(running.finishedAt, terminalAt); + assert.equal(running.durationMs, terminal.durationMs); + assert.equal((running.finalResponse as { text?: string } | null)?.text, "final answer"); + assert.equal(running.runnerTrace?.finishedAt, terminalAt); + assert.equal(running.runnerTrace?.durationMs, terminal.durationMs); + assert.equal(running.runnerTrace?.eventCount, 3); + assert.equal(running.traceAutoLifecycle, "terminal"); }); function seededMessage(traceId: string, sessionId: string): ChatMessage { 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 7a1a1d81..46f6a9cf 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 @@ -21,6 +21,7 @@ export interface WorkbenchLiveKafkaProjectionInput { } export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMessageState, event: TraceEvent): WorkbenchLiveMessageState { + if (previous.terminal) return previous; const terminal = workbenchLiveKafkaEventIsTerminal(event); const status = terminal ? firstNonEmptyString(event.status, "completed") ?? "completed" : "running"; const assistantText = workbenchLiveKafkaAssistantText(event); @@ -35,6 +36,10 @@ export function workbenchLiveKafkaMessageId(traceId: string): string { return `msg_live_${traceId}`; } +export function workbenchLiveKafkaEnvelope(schema: unknown): boolean { + return firstNonEmptyString(schema) === "hwlab.event.v1"; +} + export function workbenchLiveKafkaEventIsTerminal(event: TraceEvent | null | undefined): boolean { return Boolean(event?.terminal === true || firstNonEmptyString(event?.eventType) === "terminal" || firstNonEmptyString(event?.type) === "result"); } @@ -62,17 +67,24 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec const liveState = reduceWorkbenchLiveKafkaMessageState({ text: previous?.text ?? "", status: previous?.status ?? "running", - terminal: false + terminal: previous?.traceAutoLifecycle === "terminal" }, 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 previousFinishedAt = timestamp(previous?.timing?.finishedAt) ?? timestamp(previous?.finishedAt); + const finishedAt = liveState.terminal ? previousFinishedAt ?? receivedAt : null; + const previousDurationMs = nonNegativeNumber(previous?.timing?.durationMs) ?? nonNegativeNumber(previous?.durationMs); + const durationMs = liveState.terminal + ? previousFinishedAt && previousDurationMs !== null + ? previousDurationMs + : Math.max(0, Date.parse(finishedAt ?? receivedAt) - Date.parse(startedAt)) + : null; const timing: WorkbenchTurnTimingProjection = { startedAt, lastEventAt: receivedAt, - finishedAt: liveState.terminal ? receivedAt : null, + finishedAt, durationMs, observedAt: receivedAt, lastEventAgeMs: 0, @@ -89,7 +101,7 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec timing, startedAt, lastEventAt: receivedAt, - finishedAt: liveState.terminal ? receivedAt : null, + finishedAt, durationMs, updatedAt: receivedAt }); @@ -100,7 +112,7 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec timing, startedAt, lastEventAt: receivedAt, - finishedAt: liveState.terminal ? receivedAt : null, + finishedAt, durationMs, updatedAt: receivedAt }; @@ -120,7 +132,7 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec updatedAt: receivedAt, startedAt, lastEventAt: receivedAt, - finishedAt: liveState.terminal ? receivedAt : null, + finishedAt, durationMs, runnerTrace, traceAutoLifecycle: liveState.terminal ? "terminal" : "running", @@ -136,6 +148,10 @@ function timestamp(value: unknown): string | null { return text && Number.isFinite(Date.parse(text)) ? text : null; } +function nonNegativeNumber(value: unknown): number | null { + return typeof value === "number" && Number.isFinite(value) && value >= 0 ? value : 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.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 831a7e8c..456c5cd4 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 { projectWorkbenchLiveKafkaMessage } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, workbenchLiveKafkaEnvelope } from "./workbench-live-kafka-event"; import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority"; import { boundedProjectionMessageLimit, mergeBoundedProjectionMessages, selectProjectionMessageWindow, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget"; import { @@ -1351,10 +1351,10 @@ export const useWorkbenchStore = defineStore("workbench", () => { function applyRealtimeTraceEvent(traceId: string | null | undefined, event: WorkbenchRealtimeEvent["event"], snapshot: WorkbenchRealtimeEvent["snapshot"], realtimeEvent?: WorkbenchRealtimeEvent | null): void { const id = firstNonEmptyString(traceId, event?.traceId, snapshot?.traceId); if (!id) return; - const liveKafkaEnvelope = realtimeEvent?.schema === "hwlab.event.v1"; + const liveKafkaEnvelope = workbenchLiveKafkaEnvelope(realtimeEvent?.schema); const sessionId = realtimeEvent ? realtimeEventSessionId(realtimeEvent) : traceResultSessionId(snapshot ?? event ?? null); if (!shouldApplyActiveTraceAuthority(id, sessionId, liveKafkaEnvelope)) return; - if (traceTerminalBodyIsVisible(id, sessionId)) { + if (traceTerminalBodyIsVisible(id, sessionId) && !liveKafkaEnvelope) { recordWorkbenchRuntimeDiagnostic({ module: "workbench-terminal-priority", sessionId, traceId: id, outcome: "ok", diagnostic: { code: "terminal_low_priority_sse_trace_skip", source: "realtime-trace-event", valuesRedacted: true } }); return; }