Merge pull request #2484 from pikasTech/fix/2474-kafka-trace-visibility

fix: 保持纯 Kafka 终态与扫描完整性

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 输入适合完全离线的单步测试：

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);
     });

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;

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<string, any>) {
+export function renderKafkaCliText(payload: Record<string, any>) {
+  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 [

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 {

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<string, unknown> | null {
   return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
 }

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;
     }