pikasTech-HWLAB/tools/hwlab-cli/kafka-regenerate.test.ts

import assert from "node:assert/strict";
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
import os from "node:os";
import path from "node:path";

import { test } from "bun:test";

import { projectAgentRunKafkaEventToHwlabEvent, queryKafkaEventStream } from "../../internal/cloud/kafka-event-bridge.ts";
import { mapAgentRunRecordsToHwlabDebugEvents, projectHwlabRecordsThroughWorkbench, renderKafkaCliText, runKafkaCli } from "../src/hwlab-cli/kafka-regenerate.ts";
import { renderTraceRowsMarkdown, traceDisplayRows, traceToolSummary } 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";
const TRACE_ID = "trc_mretx18t3jl4tg";
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.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)));
  assert.ok(result.payload.commands.some((command: string) => /inspect order.*--session-id ses_/u.test(command)));
});

test("Kafka session order inspection discovers a retained session and proves source through DOM order", async () => {
  const firstTrace = "trc_order_first";
  const secondTrace = "trc_order_second";
  const agentrun = [
    agentrunOrderRecord(1, firstTrace, "user_message", { userMessageId: "msg_order_first_user", text: "first input" }),
    agentrunOrderRecord(2, firstTrace, "backend_status", { phase: "running", message: "first running" }),
    agentrunOrderRecord(3, firstTrace, "assistant_message", { text: "first reply", replyAuthority: true, final: true }),
    agentrunOrderRecord(4, firstTrace, "terminal_status", { terminalStatus: "completed", message: "completed" }),
    agentrunOrderRecord(5, secondTrace, "user_message", { userMessageId: "msg_order_second_user", text: "second input" }),
    agentrunOrderRecord(6, secondTrace, "backend_status", { phase: "running", message: "second running" })
  ];
  const hwlab = agentrun.map((record, index) => hwlabOrderRecord(record, index));
  const reads: any[] = [];
  const result = await runKafkaCli([
    "inspect", "order",
    "--contains-text", "second input",
    "--group-prefix", "hwlab-v03-workbench-isolated-debug-order-test",
    "--json"
  ], {
    env: {},
    now: () => "2026-07-11T12:00:00.000Z",
    async readKafka(input) {
      reads.push(input);
      return completeKafkaRead(input.stream === "agentrun" ? agentrun : hwlab);
    }
  });

  assert.equal(result.exitCode, 0);
  assert.equal(result.payload.scope.resolvedSessionId, HWLAB_SESSION_ID);
  assert.equal(result.payload.scope.discovery.containsTextChars, 12);
  assert.equal(result.payload.scope.discovery.matchedEventCount, 1);
  assert.equal(result.payload.scope.discovery.candidates[0].offset, hwlab[4].offset);
  assert.equal(reads.some((input) => input.stream === "agentrun" && input.sessionId === HWLAB_SESSION_ID), true);
  assert.equal(result.payload.input.refreshHandoff.counts.replayed, 6);
  assert.equal(result.payload.order.eventCount, 6);
  assert.equal(result.payload.order.firstDivergence, null);
  assert.deepEqual(result.payload.order.turnPairs.map((pair: any) => [pair.traceId, pair.dom.user, pair.dom.agent]), [
    [firstTrace, 0, 1],
    [secondTrace, 2, 3]
  ]);
  assert.deepEqual(result.payload.projection.finalMessageOrder.map((row: any) => [row.role, row.traceId]), [
    ["user", firstTrace],
    ["agent", firstTrace],
    ["user", secondTrace],
    ["agent", secondTrace]
  ]);
  assert.equal(result.payload.validation.mapperOneToOne, true);
  assert.equal(result.payload.validation.sourceOrderPreserved, true);
  assert.equal(result.payload.validation.sseOrderPreserved, true);
  assert.equal(result.payload.validation.reducerOrderPreserved, true);
  assert.equal(result.payload.validation.domTurnOrderPreserved, true);
  assert.equal(result.payload.validation.noFirstDivergence, true);
});

test("Kafka session order inspection identifies AgentRun as the first layer when user_message is committed late", async () => {
  const traceId = "trc_order_late_user";
  const agentrun = [
    agentrunOrderRecord(1, traceId, "backend_status", { phase: "running", message: "running before input" }),
    agentrunOrderRecord(2, traceId, "user_message", { userMessageId: "msg_order_late_user", text: "late input" }),
    agentrunOrderRecord(3, traceId, "assistant_message", { text: "reply", replyAuthority: true, final: true }),
    agentrunOrderRecord(4, traceId, "terminal_status", { terminalStatus: "completed", message: "completed" })
  ];
  const hwlab = agentrun.map((record, index) => hwlabOrderRecord(record, index));
  const result = await runKafkaCli([
    "inspect", "order",
    "--session-id", HWLAB_SESSION_ID,
    "--group-prefix", "hwlab-v03-workbench-isolated-debug-order-test",
    "--json"
  ], {
    env: {},
    now: () => "2026-07-11T12:00:00.000Z",
    async readKafka(input) {
      return completeKafkaRead(input.stream === "agentrun" ? agentrun : hwlab);
    }
  });

  assert.equal(result.exitCode, 0);
  assert.equal(result.payload.order.firstDivergence.layer, "agentrun.event.v1");
  assert.equal(result.payload.order.firstDivergence.code, "user-after-agent-source-event");
  assert.deepEqual(result.payload.order.firstDivergence.evidence.agentrun, { user: 1, agent: 0 });
  assert.deepEqual(result.payload.order.firstDivergence.evidence.dom, { user: 1, agent: 0 });
  assert.equal(result.payload.validation.domTurnOrderPreserved, false);
  assert.equal(result.payload.validation.noFirstDivergence, false);
});

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: "user", eventType: "user", status: "sent", label: "agentrun:user:message", userMessageId: "msg_render_trace_user", messageId: "msg_render_trace_user", targetTraceId: "trc_render_trace_target", text: "hi" }),
    hwlabRecord(2, { 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(3, { 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(4, { 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(5, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: finalText, assistantText: finalText, assistantSource: "completed-agent-message" }),
    hwlabRecord(6, { 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(7, { 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, 7);
  assert.equal(result.payload.input.matchedCount, 7);
  assert.equal(result.payload.projection.decodedCount, 7);
  assert.equal(result.payload.projection.appliedCount, 7);
  assert.equal(result.payload.projection.eventCount, 6, "formal user input must not create an agent trace event");
  assert.equal(result.payload.projection.userMessageCount, 1);
  assert.deepEqual(result.payload.projection.userMessageIds, ["msg_render_trace_user"]);
  assert.deepEqual(result.payload.projection.userTargetTraceIds, ["trc_render_trace_target"]);
  assert.equal(result.payload.projection.terminalObserved, true);
  assert.equal(result.payload.projection.lastEventAt, "2026-07-10T12:00:00.006Z");
  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.deepEqual(result.payload.replayLineage.counts, { source: 7, applied: 7, assistant: 1, tool: 1, terminal: 1, final: 1 });
  assert.deepEqual(result.payload.replayLineage.layers.source, {
    total: 7,
    user: 1,
    assistant: 3,
    tool: 2,
    terminal: 1,
    final: 1,
    other: 1,
    assistantDuplicate: { fingerprintCount: 0, occurrenceCount: 0, first: null, valuesPrinted: false }
  });
  assert.equal(result.payload.replayLineage.layers.final.duplicatedInRender, false);
  assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, null);
  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, /^\d{2}:\d{2}:\d{2}$/u);
  assert.doesNotMatch(toolRow.header, /\b(?:ok|fail|run|running)\b|total=|duration=|exit=/u);
  assert.equal(toolRow.toolState, "success");
  assert.equal(toolRow.statusLabel, "工具调用成功");
  assert.equal(toolRow.preview, "cd /workspace && rg -n target src");
  const markdown = await readFile(outputFile, "utf8");
  assert.equal((markdown.match(new RegExp(finalText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1);
  assert.equal((markdown.match(/^hi$/gmu) ?? []).length, 1);
  assert.match(markdown, /## 用户输入/u);
  assert.match(markdown, /target trace: trc_render_trace_target/u);
  assert.match(markdown, /## 运行记录/u);
  assert.match(markdown, /## 最终回复/u);
  assert.match(markdown, /<summary aria-label="工具调用成功：[^"]+" title="工具调用成功：[^"]+"><code>\d{2}:\d{2}:\d{2}<\/code> cd \/workspace &amp;&amp; rg -n target src<\/summary>/u);
  assert.doesNotMatch(markdown.match(/<summary[^>]*>[\s\S]*?<\/summary>/u)?.[0] ?? "", /total=|duration=|exit=|src\/main\.ts:42:target|总耗时/u);
  assert.equal(result.markdownOutput?.trim(), markdown.trim());
});

test("trace lineage reports the first non-final assistant content duplicate without printing message text", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-duplicate-lineage-"));
  const duplicateText = "I am checking the workspace.";
  const records = [
    hwlabRecord(1, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_first", message: duplicateText, assistantText: duplicateText }),
    hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_between", command: "rg target", exitCode: 0 }),
    hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_repeated", message: duplicateText, assistantText: duplicateText }),
    hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", message: "Done.", assistantText: "Done.", replyAuthority: true, final: true, finalResponse: { text: "Done." } }),
    hwlabRecord(5, { type: "result", eventType: "terminal", status: "completed", label: "agentrun:terminal:completed", terminal: true })
  ];
  const inputFile = path.join(cwd, "hwlab-events.jsonl");
  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,
    "--format", "markdown"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });

  assert.equal(result.exitCode, 0);
  assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 2, tool: 1, terminal: 1, final: 1 });
  assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, "source");
  assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.occurrenceCount, 1);
  assert.equal(result.payload.replayLineage.layers.applied.assistantDuplicate.occurrenceCount, 1);
  assert.equal(result.payload.replayLineage.layers.render.assistantDuplicate.occurrenceCount, 1);
  assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.first.first.identity, "evt_render_1");
  assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.first.duplicate.identity, "evt_render_3");
  assert.doesNotMatch(JSON.stringify(result.payload.replayLineage), new RegExp(duplicateText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "u"));
  assert.match(renderKafkaCliText(result.payload), /assistant=2 tool=1 terminalRows=1 final=1 duplicateLayer=source/u);
});

test("shared row model removes late assistant progress after the same item completed", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-assistant-item-lifecycle-"));
  const inputFile = path.join(cwd, "hwlab-events.jsonl");
  const duplicateText = "The completed assistant item must stay visible once.";
  const records = [
    hwlabRecord(1, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_item", assistantSource: "completed-agent-message", message: duplicateText, assistantText: duplicateText }),
    hwlabRecord(2, { type: "tool", eventType: "tool", status: "completed", label: "agentrun:tool:commandExecution", toolName: "commandExecution", itemId: "tool_between", command: "rg target", exitCode: 0 }),
    hwlabRecord(3, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_item", assistantSource: "agent-message-delta-progress", message: duplicateText, assistantText: duplicateText }),
    hwlabRecord(4, { type: "assistant", eventType: "assistant", status: "running", label: "agentrun:assistant:message", itemId: "assistant_final", assistantSource: "completed-agent-message-final", message: "Done.", assistantText: "Done.", replyAuthority: true, final: true, finalResponse: { text: "Done." } }),
    hwlabRecord(5, { 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,
    "--format", "markdown"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });

  assert.equal(result.exitCode, 0);
  assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 1, tool: 1, terminal: 1, final: 1 });
  assert.equal(result.payload.replayLineage.layers.source.assistantDuplicate.occurrenceCount, 1);
  assert.equal(result.payload.replayLineage.layers.applied.assistantDuplicate.occurrenceCount, 1);
  assert.equal(result.payload.replayLineage.layers.render.assistantDuplicate.occurrenceCount, 0);
  assert.equal(result.payload.replayLineage.firstVisibleDuplicateLayer, "source");
  assert.equal((result.markdownOutput?.match(new RegExp(duplicateText.replace(/[.*+?^${}()|[\]\\]/gu, "\\$&"), "gu")) ?? []).length, 1);
});

test("fixed Kafka assistant lifecycle replays 500/1003 byte progress into one 1190 byte final", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-assistant-progress-contract-"));
  const inputFile = path.join(cwd, "hwlab-events.jsonl");
  const itemId = "msg_0d2ec_assistant_lifecycle";
  const finalText = "x".repeat(1190);
  const progress500 = finalText.slice(0, 500);
  const progress1003 = finalText.slice(0, 1003);
  assert.deepEqual([Buffer.byteLength(progress500), Buffer.byteLength(progress1003), Buffer.byteLength(finalText)], [500, 1003, 1190]);

  const agentrunRecords = [
    agentrunAssistantLifecycleRecord(136, 1, "assistant_progress", {
      text: progress500,
      itemId,
      source: "agent-message-delta-progress",
      progress: true,
      progressFlush: false,
      replyAuthority: false,
      final: false,
      textBytes: 500,
      outputBytes: 500,
      valuesPrinted: false
    }),
    agentrunAssistantLifecycleRecord(137, 2, "assistant_progress", {
      text: progress1003,
      itemId,
      source: "agent-message-delta-progress",
      progress: true,
      progressFlush: false,
      replyAuthority: false,
      final: false,
      textBytes: 1003,
      outputBytes: 1003,
      valuesPrinted: false
    }),
    agentrunAssistantLifecycleRecord(138, 3, "assistant_message", {
      text: finalText,
      itemId,
      source: "completed-agent-message",
      durableText: true,
      replyAuthority: false,
      final: false,
      textBytes: 1190,
      outputBytes: 1190,
      valuesPrinted: false
    }),
    agentrunAssistantLifecycleRecord(142, 4, "backend_status", {
      phase: "assistant-message-final-sealed",
      itemId,
      source: "assistant-message-final-seal",
      finalSeal: true,
      replyAuthority: true,
      final: true,
      replyRef: {
        eventType: "assistant_message",
        itemId,
        source: "completed-agent-message",
        textSha256: "a".repeat(64)
      },
      valuesPrinted: false
    }),
    agentrunAssistantLifecycleRecord(143, 5, "terminal_status", {
      terminalStatus: "completed",
      valuesPrinted: false
    })
  ];
  const hwlabRecords = agentrunRecords.map((record, index) => hwlabOrderRecord(record, index));
  const hwlabEvents = hwlabRecords.map((record) => record.value.event);

  assert.deepEqual(hwlabRecords.map((record) => record.value.context.agentRunEventType), [
    "assistant_progress",
    "assistant_progress",
    "assistant_message",
    "backend_status",
    "terminal_status"
  ]);
  assert.deepEqual(hwlabEvents.slice(0, 3).map((event) => [event.itemId, event.sourceSeq, event.progress, event.assistantText.length]), [
    [itemId, 136, true, 500],
    [itemId, 137, true, 1003],
    [itemId, 138, false, 1190]
  ]);
  assert.equal(hwlabEvents[3]?.finalSeal, true);
  assert.equal(hwlabEvents[3]?.message, "assistant-message-final-sealed");
  assert.equal(hwlabEvents[3]?.assistantText, undefined, "seal must not copy the completed body");
  assert.equal(hwlabEvents[4]?.assistantText, undefined, "terminal must not copy the completed body");

  const runningRows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, hwlabEvents.slice(0, 2));
  assert.equal(runningRows.filter((row) => /助手/u.test(row.header)).length, 1);
  assert.equal(runningRows.find((row) => /助手/u.test(row.header))?.seq, 137);
  assert.equal(runningRows.find((row) => /助手/u.test(row.header))?.body, progress1003);
  const completedRows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, hwlabEvents.slice(0, 3));
  assert.equal(completedRows.filter((row) => /助手/u.test(row.header)).length, 1);
  assert.equal(completedRows.find((row) => /助手/u.test(row.header))?.seq, 138);
  assert.equal(completedRows.find((row) => /助手/u.test(row.header))?.body, finalText);

  const projected = projectHwlabRecordsThroughWorkbench(hwlabRecords, {
    traceId: TRACE_ID,
    sessionId: HWLAB_SESSION_ID,
    observedAt: "2026-07-12T10:01:00.000Z"
  });
  assert.equal(projected.appliedCount, 5);
  assert.equal(projected.sourceEvents.length, 5);
  assert.equal(projected.message?.runnerTrace?.events?.length, 5, "raw/applied observation must retain every unfiltered event");
  assert.equal(projected.message?.text, finalText);
  assert.equal(projected.message?.status, "completed");
  assert.equal((projected.message?.finalResponse as { text?: string } | null)?.text, finalText);

  await writeFile(inputFile, hwlabRecords.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,
    "--jsonl-file", inputFile,
    "--format", "markdown"
  ], { cwd, env: {}, now: () => "2026-07-12T10:02:00.000Z" });

  assert.equal(result.exitCode, 0);
  assert.equal(result.payload.projection.appliedCount, 5);
  assert.equal(result.payload.projection.eventCount, 5);
  assert.equal(result.payload.render.assistantRowCount, 0, "the authoritative outer final replaces the progress lifecycle");
  assert.equal(result.payload.render.finalResponsePresent, true);
  assert.equal(result.payload.render.finalResponseInTrace, false);
  assert.deepEqual(result.payload.replayLineage.counts, { source: 5, applied: 5, assistant: 0, tool: 0, terminal: 1, final: 1 });
  assert.equal(result.payload.replayLineage.layers.source.assistant, 3, "raw source lineage must keep both progress snapshots and the durable completion");
  assert.equal(result.payload.replayLineage.layers.applied.assistant, 3);
  assert.equal(result.payload.validation.allMatchedEventsApplied, true);
  assert.equal(result.payload.validation.finalResponseOutsideTrace, true);
  assert.equal(result.markdownOutput?.split(finalText).length, 2, "the final body must render exactly once");
  const traceMarkdown = (result.markdownOutput ?? "").split("## 运行记录")[1]?.split("## 最终回复")[0] ?? "";
  assert.doesNotMatch(traceMarkdown, /x{500}/u, "progress must not remain as a separate rendered row");
});

test("HWLAB compatibility window coalesces legacy assistant_message progress before AgentRun assistant_progress rollout", () => {
  const itemId = "msg_legacy_progress_contract";
  const finalText = "legacy durable body";
  const agentrunRecords = [
    agentrunAssistantLifecycleRecord(201, 1, "assistant_message", {
      text: "legacy progress 1",
      itemId,
      source: "agent-message-delta-progress",
      progress: true,
      progressFlush: false,
      replyAuthority: false,
      final: false
    }),
    agentrunAssistantLifecycleRecord(202, 2, "assistant_message", {
      text: "legacy progress 2",
      itemId,
      source: "agent-message-delta-progress",
      progress: true,
      progressFlush: true,
      replyAuthority: false,
      final: false
    }),
    agentrunAssistantLifecycleRecord(203, 3, "assistant_message", {
      text: finalText,
      itemId,
      source: "completed-agent-message",
      durableText: true,
      replyAuthority: false,
      final: false
    }),
    agentrunAssistantLifecycleRecord(204, 4, "terminal_status", { terminalStatus: "completed" })
  ];
  const hwlabRecords = agentrunRecords.map((record, index) => hwlabOrderRecord(record, index));
  const events = hwlabRecords.map((record) => record.value.event);

  assert.deepEqual(events.slice(0, 2).map((event) => [event.agentRunEventType, event.eventType, event.progress]), [
    ["assistant_message", "assistant", true],
    ["assistant_message", "assistant", true]
  ]);
  const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, events.slice(0, 3));
  const assistantRows = rows.filter((row) => /助手/u.test(row.header));
  assert.equal(assistantRows.length, 1);
  assert.equal(assistantRows[0]?.seq, 203);
  assert.equal(assistantRows[0]?.body, finalText);

  const projected = projectHwlabRecordsThroughWorkbench(hwlabRecords, {
    traceId: TRACE_ID,
    sessionId: HWLAB_SESSION_ID,
    observedAt: "2026-07-12T10:03:00.000Z"
  });
  assert.equal(projected.message?.text, finalText);
  assert.equal(projected.message?.runnerTrace?.events?.length, 4);
  assert.equal((projected.message?.finalResponse as { text?: string } | null)?.text, finalText);
  const finalRows = traceDisplayRows(projected.message?.runnerTrace ?? {}, projected.message?.runnerTrace?.events ?? [], {
    finalResponsePlacement: "outer",
    outerFinalResponseText: finalText
  });
  assert.equal(finalRows.filter((row) => /助手/u.test(row.header)).length, 0);
});

test("shared tool summary model keeps status semantic and visible content status-free", () => {
  const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [
    {
      sourceSeq: 1,
      sourceEventId: "evt_tool_running",
      type: "tool_call",
      status: "running",
      label: "agentrun:tool:commandExecution:started",
      toolName: "commandExecution",
      itemId: "tool_running",
      command: "bun test active-fixture",
      createdAt: "2026-07-10T11:00:01.000Z"
    },
    {
      sourceSeq: 2,
      sourceEventId: "evt_tool_success",
      type: "tool_call",
      status: "completed",
      label: "agentrun:tool:commandExecution:completed",
      toolName: "commandExecution",
      itemId: "tool_success",
      command: "find . -maxdepth 3 -name package.json -o -name tsconfig.json -o -name vitest.config.ts",
      durationMs: 1234,
      exitCode: 0,
      stdoutSummary: "src/main.ts:1:success",
      createdAt: "2026-07-10T11:00:02.000Z"
    },
    {
      sourceSeq: 3,
      sourceEventId: "evt_tool_failure",
      type: "tool_call",
      status: "completed",
      label: "agentrun:tool:commandExecution:completed",
      toolName: "commandExecution",
      itemId: "tool_failure",
      command: "bun test failure-fixture",
      durationMs: 2345,
      exitCode: 1,
      stderrSummary: "failure output",
      createdAt: "2026-07-10T11:00:03.000Z"
    },
    {
      sourceSeq: 4,
      sourceEventId: "evt_tool_output_chunk",
      type: "tool_call",
      status: "output_chunk",
      label: "item/commandExecution/outputDelta",
      outputSummary: "streamed output",
      createdAt: "2026-07-10T11:00:04.000Z"
    }
  ]).filter((row) => row.rowId.startsWith("tool:"));

  assert.deepEqual(rows.map((row) => row.toolState), ["running", "success", "failure", "running"]);
  assert.deepEqual(rows.map((row) => row.tone), ["warn", "ok", "blocked", "warn"]);
  assert.deepEqual(rows.map((row) => row.statusLabel), ["工具调用中", "工具调用成功", "工具调用失败", "工具调用中"]);
  assert.ok(rows.every((row) => /^\d{2}:\d{2}:\d{2}$/u.test(row.header)));
  assert.equal(rows[1]!.preview, "find . -maxdepth 3 -name package.json -o -name tsconfig.json -o -name vitest.config.ts");
  assert.equal(rows[2]!.preview, "bun test failure-fixture");
  assert.doesNotMatch(rows[3]!.header, /commandExecution|outputDelta/u);

  const markdown = renderTraceRowsMarkdown(rows);
  const visibleSummaries = [...markdown.matchAll(/<summary[^>]*>([\s\S]*?)<\/summary>/gu)].map((match) => match[1]);
  assert.equal(visibleSummaries.length, 4);
  for (const [index, summary] of visibleSummaries.entries()) {
    assert.equal(summary, `<code>${rows[index]!.header}</code> ${traceToolSummary(rows[index]!)}`);
    assert.doesNotMatch(summary, /\b(?:ok|fail|run|running)\b|total=|duration=|exit=/u);
  }
  assert.match(visibleSummaries[1]!, /^<code>\d{2}:\d{2}:\d{2}<\/code> find \. -maxdepth 3/u);
  assert.match(markdown, /aria-label="工具调用中：/u);
  assert.match(markdown, /aria-label="工具调用成功：/u);
  assert.match(markdown, /aria-label="工具调用失败：/u);
});

test("fileChange and turn diff render real paths and unified diff as valid Markdown cards", () => {
  const fileDiff = "@@ -1 +1 @@\n-export const value = 1;\n+export const value = 2;";
  const aggregateDiff = `${fileDiff}\n@@ -0,0 +1 @@\n+# 文件编辑可见`;
  const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [
    {
      sourceSeq: 21,
      sourceEventId: "evt_file_change_completed",
      type: "tool_call",
      eventType: "tool_call",
      status: "completed",
      label: "agentrun:tool:fileChange",
      toolName: "fileChange",
      toolType: "fileChange",
      itemId: "item_file_change",
      changes: [
        { path: "src/example.ts", kind: "update", diff: fileDiff },
        { path: "docs/说明.md", kind: "add", diff: "@@ -0,0 +1 @@\n+# 文件编辑可见" }
      ],
      createdAt: "2026-07-12T15:00:01.000Z"
    },
    {
      sourceSeq: 22,
      sourceEventId: "evt_turn_diff_updated",
      type: "diff",
      eventType: "diff",
      status: "updated",
      label: "agentrun:diff",
      diff: aggregateDiff,
      message: aggregateDiff,
      createdAt: "2026-07-12T15:00:02.000Z"
    }
  ]);
  const markdown = renderTraceRowsMarkdown(rows);
  assert.equal(rows.length, 2);
  assert.match(markdown, /2 个文件 · \+2 -1 · src\/example\.ts, docs\/说明\.md/u);
  assert.match(markdown, /#### `src\/example\.ts`/u);
  assert.match(markdown, /#### `docs\/说明\.md`/u);
  assert.match(markdown, /\+export const value = 2;/u);
  assert.match(markdown, /文件 diff 已更新/u);
  assert.match(markdown, /\+# 文件编辑可见/u);
  assert.equal((markdown.match(/<details>/gu) ?? []).length, 2);
  assert.equal((markdown.match(/<\/details>/gu) ?? []).length, 2);
  assert.ok(rows.every((row) => row.bodyFormat === "markdown"));
});

test("fileChange rendering uses preserved stats and ignores historical object coercion", () => {
  const rows = traceDisplayRows({ eventSource: "hwlab-kafka-sse" }, [
    {
      sourceSeq: 53,
      type: "tool",
      eventType: "tool",
      status: "completed",
      label: "agentrun:tool:fileChange",
      toolName: "fileChange",
      itemId: "item_large_file_change",
      changes: [{ path: "/agentrun-workspace/src/large.ts", kind: "add", additions: 400, deletions: 0, diffTruncated: true }],
      message: "bounded file change metadata",
      createdAt: "2026-07-21T08:04:30.452Z"
    },
    {
      sourceSeq: 54,
      type: "tool",
      eventType: "tool",
      status: "completed",
      label: "agentrun:tool:fileChange",
      toolName: "fileChange",
      itemId: "item_historical_file_change",
      message: "[object Object]",
      createdAt: "2026-07-21T08:04:31.452Z"
    }
  ]);
  assert.equal(rows[0]?.preview, "src/large.ts · +400 -0");
  assert.match(rows[0]?.body ?? "", /新增 · \+400 -0/u);
  assert.doesNotMatch(JSON.stringify(rows), /\[object Object\]/u);
  assert.equal(rows[1]?.body, null);
  assert.equal(rows[1]?.preview, "fileChange");
});

test("outer final response suppression follows formal assistant identity instead of equal text", () => {
  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.equal(rows.length, 1);
  assert.equal(rows[0]?.rowId, "event:evt_same_text");
  assert.equal(rows[0]?.body, finalText, "a distinct itemId with equal text must remain visible");
});

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;
      const events = [exactTool, exactProgress, unscopedFinal, unscopedTerminal, runnerShutdown, unscopedBackend];
      return {
        ...completeKafkaRead(events),
        scannedCount: 6,
        parsedCount: 6,
        matchedCount: 6,
        filterRejectedCount: 0
      };
    }
  });

  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.equal(result.payload.input.refreshHandoff.status, "succeeded");
  assert.equal(result.payload.input.refreshHandoff.liveSubscriptionInstalled, true);
  assert.equal(result.payload.input.refreshHandoff.counts.replayed, 6);
  assert.deepEqual(result.payload.input.refreshHandoff.barrier, [{ partition: 0, endOffset: "1907" }]);
  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.projection.pipeline[0], "createWorkbenchKafkaRefreshHandoff");
  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() {
      const events = [otherCommand, commandlessFinal, commandlessTerminal];
      return {
        ...completeKafkaRead(events),
        scannedCount: 3,
        parsedCount: 3,
        matchedCount: 3,
        filterRejectedCount: 0
      };
    }
  });

  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("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: false,
        endOffsetsError: { code: "ENOTFOUND", message: "broker metadata hostname unresolved", valuesRedacted: true },
        endOffsets: [],
        completion: { reason: "timeout", complete: false, barrierReached: false, retentionStartVerified: true },
        lastScannedOffsets: [{ partition: 0, offset: "0" }]
      };
    }
  });

  assert.equal(result.exitCode, 2);
  assert.equal(result.payload.input.endOffsetsError.code, "ENOTFOUND");
  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.input.refreshHandoff.status, "failed");
  assert.equal(result.payload.input.refreshHandoff.error.code, "workbench_kafka_refresh_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("incomplete zero-match Kafka scan stays partial and never claims the trace is missing", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-trace-empty-partial-"));
  const result = await runKafkaCli([
    "render", "trace",
    "--from", "kafka",
    "--trace-id", TRACE_ID,
    "--group-prefix", "hwlab-v03-workbench-isolated-debug",
    "--format", "markdown"
  ], {
    cwd,
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka() {
      return {
        events: [],
        scannedCount: 0,
        parsedCount: 0,
        matchedCount: 0,
        filterRejectedCount: 0,
        completionReason: "timeout",
        reachedEndOffsets: false,
        endOffsetsAvailable: true,
        endOffsets: [{ partition: 0, startOffset: "0", endOffset: "12" }],
        completion: { reason: "timeout", complete: false, barrierReached: false, retentionStartVerified: true },
        lastScannedOffsets: []
      };
    }
  });

  assert.equal(result.exitCode, 2);
  assert.equal(result.payload.status, "partial");
  assert.equal(result.payload.error.code, "source_scan_incomplete");
  assert.notEqual(result.payload.error.code, "source_trace_missing");
  assert.equal(result.payload.input.scannedCount, 0);
  assert.equal(result.payload.input.matchedCount, 0);
  assert.equal(result.payload.input.scanComplete, false);
  assert.equal(result.payload.input.refreshHandoff.status, "failed");
  assert.equal(result.payload.input.refreshHandoff.error.code, "workbench_kafka_refresh_timeout");
  assert.equal(result.payload.validation.refreshHandoffComplete, 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 scanned=0 matched=0 applied=0/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");
  const outputFile = path.join(cwd, "hwlab-events.jsonl");
  const records = Array.from({ length: 35 }, (_, index) => reconstructionRecord(index + 1));
  await writeFile(inputFile, records.map((record) => JSON.stringify(record)).join("\n") + "\n", "utf8");

  const result = await runKafkaCli([
    "regenerate", "hwlab",
    "--from", "jsonl",
    "--session-id", SESSION_ID,
    "--trace-id", TRACE_ID,
    "--replay-id", "rpl_offline_35",
    "--jsonl-file", inputFile,
    "--output-jsonl", outputFile,
    "--expect-count", "35",
    "--no-publish",
    "--json"
  ], {
    cwd,
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async createProducer() { throw new Error("offline JSONL must not create a producer"); }
  });

  assert.equal(result.exitCode, 0);
  assert.equal(result.payload.input.readCount, 35);
  assert.equal(result.payload.output.count, 35);
  assert.equal(result.payload.output.publishedCount, 0);
  assert.deepEqual(result.payload.validation, {
    expectedCount: 35,
    sourceScanComplete: true,
    oneToOne: true,
    orderPreserved: true,
    lineagePreserved: true,
    firstSourceSeq: 1,
    lastSourceSeq: 35,
    inputKinds: ["stdio-derived partial reconstruction"]
  });
  assert.deepEqual(result.payload.runtimeDependencies, {
    cloudApi: false,
    database: false,
    transactionalProjector: false,
    kafka: false
  });

  const output = (await readFile(outputFile, "utf8")).trim().split("\n").map(JSON.parse);
  assert.equal(output.length, 35);
  assert.deepEqual(output.map((row) => row.value.debugLineage.sourceSeq), Array.from({ length: 35 }, (_, index) => index + 1));
  assert.ok(output.every((row) => row.topic === "hwlab.event.debug.v1"));
  assert.ok(output.every((row) => row.value.schema === "hwlab.event.debug.v1"));
  assert.ok(output.every((row) => row.partition === 0));
  assert.ok(output.every((row) => row.value.replayId === "rpl_offline_35"));
  assert.deepEqual(output.map((row) => row.value.debugReplay.index), Array.from({ length: 35 }, (_, index) => index + 1));
  assert.ok(output.every((row) => row.value.eventId === null && row.value.sourceEventId === null && row.value.outboxSeq === null));
  assert.ok(output.every((row) => row.value.debugLineage.inputEventId === null && row.value.debugLineage.inputOutboxSeq === null));
  assert.ok(output.every((row) => row.value.sourceEvent.partition === null && row.value.sourceEvent.offset === null));
  assert.ok(output.every((row) => row.value.traceId === TRACE_ID && row.value.sessionId === HWLAB_SESSION_ID));
  assert.ok(output.every((row) => row.value.debugLineage.inputSessionId === SESSION_ID));
  assert.ok(output.every((row) => row.value.debugLineage.inputHwlabSessionId === HWLAB_SESSION_ID));
  assert.ok(output.every((row) => row.value.reconstruction.kind === "stdio-derived partial reconstruction"));
  assert.ok(output.every((row) => row.value.reconstruction.originalEventId === null && row.value.reconstruction.originalOutboxSeq === null));
  assert.ok(output.every((row) => row.value.reconstruction.identityDerivation.kind === "agentrun-session-uuid-to-hwlab-session"));
  assert.ok(output.every((row) => row.value.reconstruction.identityDerivation.reversible === true));
});

test("Kafka reader and producer are injected while canonical AgentRun events use strict decode and isolated output", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-canonical-"));
  const records = Array.from({ length: 35 }, (_, index) => canonicalRecord(index + 1));
  const readerCalls: any[] = [];
  const producerCalls: any[] = [];
  let connected = 0;
  let disconnected = 0;
  const result = await runKafkaCli([
    "regenerate", "hwlab",
    "--from", "kafka",
    "--session-id", SESSION_ID,
    "--trace-id", TRACE_ID,
    "--replay-id", "rpl_canonical_35",
    "--input-topic", "agentrun.event.v1",
    "--publish",
    "--expect-count", "35",
    "--output-jsonl", path.join(cwd, "canonical-output.jsonl"),
    "--json"
  ], {
    cwd,
    env: {
      HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092",
      HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.debug.v1",
      HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-v03-workbench-isolated-debug"
    },
    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, completionReason: "end-offset", reachedEndOffsets: true };
    },
    async createProducer() {
      return {
        async connect() { connected += 1; },
        async send(input: any) { producerCalls.push(input); return [{ topicName: input.topic, partition: 0, baseOffset: "100" }]; },
        async disconnect() { disconnected += 1; }
      };
    }
  });

  assert.equal(result.exitCode, 0);
  assert.equal(readerCalls.length, 1);
  assert.equal(readerCalls[0].groupIdPrefix, "hwlab-v03-workbench-isolated-debug");
  assert.equal(readerCalls[0].sessionId, SESSION_ID);
  assert.equal(readerCalls[0].limit, 36, "exact count verification must leave room to detect one extra event");
  assert.equal(connected, 1);
  assert.equal(disconnected, 1);
  assert.equal(producerCalls.length, 1);
  assert.equal(producerCalls[0].topic, "hwlab.event.debug.v1");
  assert.equal(producerCalls[0].messages.length, 35);
  assert.ok(producerCalls[0].messages.every((message: any) => message.partition === 0));
  const envelopes = producerCalls[0].messages.map((message: any) => JSON.parse(message.value));
  assert.ok(envelopes.every((event: any) => event.replayId === "rpl_canonical_35"));
  assert.ok(producerCalls[0].messages.every((message: any) => message.headers["x-debug-replay-id"] === "rpl_canonical_35"));
  assert.deepEqual(envelopes.map((event: any) => event.debugLineage.sourceSeq), Array.from({ length: 35 }, (_, index) => index + 1));
  assert.deepEqual(envelopes.map((event: any) => event.eventId), Array.from({ length: 35 }, (_, index) => `hwlab:evt_${index + 1}`));
  assert.deepEqual(envelopes.map((event: any) => event.debugLineage.inputOutboxSeq), Array.from({ length: 35 }, (_, index) => index + 1));
  assert.ok(envelopes.every((event: any) => event.diagnostics.inputKind === "canonical-durable"));
  assert.ok(envelopes.every((event: any) => event.schema === "hwlab.event.debug.v1"));
  assert.equal(result.payload.output.publishedCount, 35);
  assert.deepEqual(result.payload.config.outputTopic, { value: "hwlab.event.debug.v1", source: "env:HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC" });
  assert.deepEqual(result.payload.config.groupPrefix, { value: "hwlab-v03-workbench-isolated-debug", source: "env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX" });
  assert.equal(result.payload.validation.orderPreserved, true);
  assert.equal(result.payload.validation.lineagePreserved, true);
  assert.deepEqual(result.payload.output.offsetRanges, [{
    topic: "hwlab.event.debug.v1",
    partition: 0,
    firstOffset: "100",
    lastOffset: "134",
    count: 35,
    valuesPrinted: false
  }]);
  assert.equal(result.payload.replay.replayId, "rpl_canonical_35");
  assert.equal(result.payload.replay.producerInvoked, true);
  assert.equal(result.payload.replay.sourceMatched, true);
  assert.deepEqual(result.payload.replay.outputBarrier, result.payload.output.offsetRanges[0]);
});

test("Kafka regeneration reports source trace missing before creating a producer", async () => {
  let producerCreated = false;
  let readerInput: any = null;
  const result = await runKafkaCli([
    "regenerate", "hwlab",
    "--from", "kafka",
    "--session-id", SESSION_ID,
    "--trace-id", "trc_missing_debug_source",
    "--replay-id", "rpl_missing_debug_source",
    "--group-prefix", "hwlab-v03-workbench-isolated-debug"
  ], {
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka(input) { readerInput = input; return { events: [canonicalRecord(1)] }; },
    async createProducer() { producerCreated = true; throw new Error("must not create producer"); }
  });

  assert.equal(result.exitCode, 1);
  assert.equal(result.payload.error.code, "source_trace_missing");
  assert.equal(readerInput.topic, "agentrun.event.v1");
  assert.equal(producerCreated, false);
});

test("canonical Kafka regeneration preflights without publishing and preserves replayId in the publish hint", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-preflight-"));
  let producerCreated = false;
  const result = await runKafkaCli([
    "regenerate", "hwlab",
    "--from", "kafka",
    "--session-id", SESSION_ID,
    "--trace-id", TRACE_ID,
    "--replay-id", "rpl_canonical_preflight",
    "--group-prefix", "hwlab-v03-workbench-isolated-debug",
    "--output-jsonl", path.join(cwd, "output.jsonl")
  ], {
    cwd,
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka() { return { events: [canonicalRecord(1)], completionReason: "end-offset", reachedEndOffsets: true }; },
    async createProducer() { producerCreated = true; throw new Error("preflight must not create a producer"); }
  });

  assert.equal(result.exitCode, 0);
  assert.equal(result.payload.config.inputTopic.value, "agentrun.event.v1");
  assert.equal(result.payload.output.published, false);
  assert.equal(result.payload.output.publishedCount, 0);
  assert.equal(result.payload.replay.producerInvoked, false);
  assert.equal(result.payload.replay.outputBarrier, null);
  assert.match(result.payload.next.command, /--replay-id rpl_canonical_preflight/u);
  assert.match(result.payload.next.command, /--group-prefix hwlab-v03-workbench-isolated-debug/u);
  assert.match(result.payload.next.command, /--publish/u);
  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([
    "regenerate", "hwlab",
    "--from", "kafka",
    "--session-id", SESSION_ID,
    "--trace-id", TRACE_ID,
    "--replay-id", "rpl_missing_offset_barrier",
    "--group-prefix", "hwlab-v03-workbench-isolated-debug",
    "--publish",
    "--output-jsonl", path.join(cwd, "output.jsonl")
  ], {
    cwd,
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka() { return { events: [canonicalRecord(1)], completionReason: "end-offset", reachedEndOffsets: true }; },
    async createProducer() {
      return {
        async connect() {},
        async send() { return []; },
        async disconnect() {}
      };
    }
  });

  assert.equal(result.exitCode, 1);
  assert.equal(result.payload.error.code, "debug_publish_barrier_missing");
});

test("reconstruction identity fabrication and product output topic fail closed", async () => {
  const cwd = await mkdtemp(path.join(os.tmpdir(), "hwlab-kafka-debug-invalid-"));
  const forged = reconstructionRecord(1);
  forged.value.eventId = "evt_forged";
  const inputFile = path.join(cwd, "forged.jsonl");
  await writeFile(inputFile, `${JSON.stringify(forged)}\n`, "utf8");

  const forgedResult = await runKafkaCli([
    "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID,
    "--jsonl-file", inputFile, "--no-publish"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
  assert.equal(forgedResult.exitCode, 1);
  assert.equal(forgedResult.payload.error.code, "hwlab_kafka_debug_identity_invalid");

  const productResult = await runKafkaCli([
    "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID,
    "--jsonl-file", inputFile, "--output-topic", "hwlab.event.v1", "--no-publish"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
  assert.equal(productResult.exitCode, 1);
  assert.equal(productResult.payload.error.code, "product_topic_forbidden");

  const productGroupResult = await runKafkaCli([
    "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID,
    "--jsonl-file", inputFile, "--group-prefix", "hwlab-v03-workbench-live-sse", "--no-publish"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
  assert.equal(productGroupResult.exitCode, 1);
  assert.equal(productGroupResult.payload.error.code, "debug_group_required");

  const replayIdResult = await runKafkaCli([
    "regenerate", "hwlab", "--from", "jsonl", "--session-id", SESSION_ID,
    "--replay-id", "not-a-replay", "--jsonl-file", inputFile, "--no-publish"
  ], { cwd, env: {}, now: () => "2026-07-10T12:00:00.000Z" });
  assert.equal(replayIdResult.exitCode, 1);
  assert.equal(replayIdResult.payload.error.code, "invalid_replay_id");
});

test("Kafka mode requires a CLI or YAML-owned isolated group prefix", async () => {
  let readCalled = false;
  const result = await runKafkaCli([
    "regenerate", "hwlab", "--from", "kafka", "--session-id", SESSION_ID, "--no-publish"
  ], {
    env: {},
    now: () => "2026-07-10T12:00:00.000Z",
    async readKafka() { readCalled = true; return { events: [] }; }
  });

  assert.equal(result.exitCode, 1);
  assert.equal(result.payload.error.code, "kafka_debug_group_missing");
  assert.equal(readCalled, false);
});

test("Kafka query exposes the isolated generated group and original value hash", async () => {
  const valueText = JSON.stringify(canonicalRecord(1).value);
  const consumerConfigs: any[] = [];
  const consumer = {
    async connect() {},
    async subscribe() {},
    async run({ eachMessage }: any) {
      await eachMessage({
        topic: "agentrun.event.v1",
        partition: 0,
        message: { offset: "1601", key: Buffer.from(SESSION_ID), value: Buffer.from(valueText), timestamp: "1783681200000" }
      });
    },
    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: 1,
    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
    kafkaFactory() {
      return { consumer(config: any) { consumerConfigs.push(config); return consumer; } };
    }
  });
  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.firstScannedOffsets, [{ partition: 0, offset: "0" }]);
  assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "2" }]);
});

test("Kafka query fails closed when retention advances between barrier capture and consumer assignment", 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: "hwlab.event.v1",
    sessionId: SESSION_ID,
    limit: 10,
    scanLimit: 10,
    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 {
            async connect() {},
            async subscribe() {},
            async run({ eachMessage }: any) {
              await eachMessage({ topic: "hwlab.event.v1", partition: 0, message: { offset: "2", value: Buffer.from(JSON.stringify(matching)) } });
            },
            async stop() {},
            async disconnect() {}
          };
        }
      };
    }
  });

  assert.equal(queried.completionReason, "retention-start-moved");
  assert.equal(queried.completion.complete, false);
  assert.equal(queried.completion.retentionStartVerified, false);
  assert.equal(queried.completion.retentionStartMoved, true);
  assert.equal(queried.reachedEndOffsets, false);
  assert.equal(queried.scannedCount, 0);
  assert.deepEqual(queried.firstScannedOffsets, [{ 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 session filter accepts canonical hwlabSessionId across AgentRun and HWLAB topics", 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: HWLAB_SESSION_ID,
    limit: 10,
    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.filterRejectedCount, 0);
  assert.equal(queried.completionReason, "end-offset");
});

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 stops every partition at the captured exclusive barrier and excludes later records", 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 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: 10,
    scanLimit: 10,
    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
    kafkaFactory() {
      return {
        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "2" }, { partition: 1, low: "0", high: "2" }]; }, 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)) } });
              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)) } });
              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "0", value: Buffer.from(JSON.stringify(mismatch)) } });
              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "1", value: Buffer.from(JSON.stringify(matching)) } });
              await eachMessage({ topic: "agentrun.event.v1", partition: 1, message: { offset: "3", value: Buffer.from(JSON.stringify(matching)) } });
            },
            async stop() {},
            async disconnect() {}
          };
        }
      };
    }
  });

  assert.equal(queried.completionReason, "end-offset");
  assert.equal(queried.completion.complete, true);
  assert.equal(queried.completion.barrierReached, true);
  assert.equal(queried.scannedCount, 4);
  assert.equal(queried.excludedPostBarrierCount, 1);
  assert.deepEqual(queried.events.map((event: any) => [event.partition, event.offset]), [[0, "0"], [1, "1"]]);
  assert.deepEqual(queried.lastScannedOffsets, [{ partition: 0, offset: "1" }, { partition: 1, offset: "1" }]);
});

test("Kafka query reports scan-limit and abort as typed incomplete completion", async () => {
  const matching = canonicalRecord(1).value;
  const limited = await queryKafkaEventStream({
    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
    topic: "agentrun.event.v1",
    sessionId: SESSION_ID,
    limit: 10,
    scanLimit: 1,
    timeoutMs: 500,
    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
    kafkaFactory() {
      return {
        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "5" }]; }, 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.deepEqual(limited.completion, {
    reason: "scan-limit",
    complete: false,
    barrierReached: false,
    retentionStartVerified: true,
    retentionStartMoved: false,
    timeout: false,
    matchedEventLimitReached: false,
    scanLimitReached: true,
    aborted: false,
    valuesRedacted: true
  });

  const controller = new AbortController();
  let disconnected = false;
  const abortedPromise = queryKafkaEventStream({
    env: { HWLAB_KAFKA_BOOTSTRAP_SERVERS: "kafka.test:9092", HWLAB_KAFKA_CLIENT_ID: "hwlab-test" },
    topic: "agentrun.event.v1",
    sessionId: SESSION_ID,
    limit: 10,
    scanLimit: 10,
    timeoutMs: 500,
    signal: controller.signal,
    groupIdPrefix: "hwlab-v03-workbench-isolated-debug",
    kafkaFactory() {
      return {
        admin() { return { async connect() {}, async fetchTopicOffsets() { return [{ partition: 0, low: "0", high: "5" }]; }, async disconnect() {} }; },
        consumer() { return { async connect() {}, async subscribe() {}, async run() { await new Promise(() => undefined); }, async stop() {}, async disconnect() { disconnected = true; } }; }
      };
    }
  });
  setTimeout(() => controller.abort(), 5);
  const aborted = await abortedPromise;
  assert.equal(aborted.completionReason, "aborted");
  assert.equal(aborted.completion.aborted, true);
  assert.equal(aborted.completion.complete, false);
  assert.equal(disconnected, true);
});

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", () => {
  const rawSessionId = "ses_agentrun_not_uuid";
  const record = reconstructionRecord(1);
  record.key = rawSessionId;
  record.value.sessionId = rawSessionId;
  record.value.hwlabSessionId = null;
  record.value.run.sessionId = rawSessionId;
  record.value.run.hwlabSessionId = null;
  record.value.event.payload.sessionId = rawSessionId;
  record.value.reconstruction.identityDerivation = {
    kind: "none",
    sourceField: "sessionId",
    targetField: "hwlabSessionId",
    reversible: false,
    reason: "source sessionId is not ses_agentrun_<uuid-with-underscores>",
    valuesPrinted: false
  };
  const mapped = mapAgentRunRecordsToHwlabDebugEvents([record], {
    sessionId: rawSessionId,
    traceId: TRACE_ID,
    expectedCount: 1,
    producedAt: "2026-07-10T12:00:00.000Z"
  });
  assert.equal(mapped.events[0].sessionId, null);
  assert.equal(mapped.events[0].hwlabSessionId, null);
  assert.equal(mapped.events[0].event.sessionId, null);
  assert.equal(mapped.events[0].debugLineage.inputSessionId, rawSessionId);
  assert.equal(mapped.events[0].reconstruction.identityDerivation.kind, "none");
});

function completeKafkaRead(events: any[]) {
  const byPartition = new Map<number, bigint[]>();
  for (const event of events) {
    const partition = Number(event.partition);
    const offset = BigInt(String(event.offset));
    const offsets = byPartition.get(partition) ?? [];
    offsets.push(offset);
    byPartition.set(partition, offsets);
  }
  const endOffsets = [...byPartition.entries()].map(([partition, offsets]) => ({
    partition,
    startOffset: String(offsets.reduce((left, right) => left < right ? left : right)),
    endOffset: String(offsets.reduce((left, right) => left > right ? left : right) + 1n)
  })).sort((left, right) => left.partition - right.partition);
  return {
    topic: events[0]?.topic ?? "hwlab.event.v1",
    events,
    completionReason: "end-offset",
    reachedEndOffsets: true,
    endOffsetsAvailable: true,
    endOffsets,
    completion: { reason: "end-offset", complete: true, barrierReached: true, retentionStartVerified: true }
  };
}

function reconstructionRecord(seq: number): any {
  const value = {
    schema: "agentrun.event.reconstruction.v1",
    eventType: "agentrun.event.reconstructed",
    source: "agentrun-cli-stdio-reconstruction",
    eventId: null,
    outboxSeq: null,
    sourceSeq: seq,
    observedAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`,
    traceId: TRACE_ID,
    sessionId: SESSION_ID,
    hwlabSessionId: HWLAB_SESSION_ID,
    runId: RUN_ID,
    commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e",
    run: { runId: RUN_ID, sessionId: SESSION_ID, hwlabSessionId: HWLAB_SESSION_ID, status: "running", valuesPrinted: false },
    command: { commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e", runId: RUN_ID, seq: 1, type: "run", state: "running", valuesPrinted: false },
    event: {
      id: null,
      runId: RUN_ID,
      seq,
      type: "backend_status",
      payload: {
        traceId: TRACE_ID,
        sessionId: SESSION_ID,
        phase: `reconstructed-${seq}`,
        reconstructed: true,
        reconstructionKind: "stdio-derived partial reconstruction"
      },
      createdAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`
    },
    reconstruction: {
      kind: "stdio-derived partial reconstruction",
      completeLifecycle: false,
      originalEventId: null,
      originalOutboxSeq: null,
      sourceTopic: "codex-stdio.raw.v1",
      sourcePartition: 0,
      sourceOffset: String(3500 + seq),
      sourceFrameSeq: seq,
      sourceValueSha256: `sha256:${String(seq).padStart(64, "0")}`,
      identityDerivation: {
        kind: "agentrun-session-uuid-to-hwlab-session",
        sourceField: "sessionId",
        targetField: "hwlabSessionId",
        reversible: true,
        sourcePrefix: "ses_agentrun_",
        targetPrefix: "ses_",
        separatorTransform: "underscore-to-hyphen",
        valuesPrinted: false
      }
    },
    valuesPrinted: false
  };
  return { topic: "agentrun.event.debug.v1", partition: null, offset: null, key: SESSION_ID, value, headers: { "x-trace-id": TRACE_ID } };
}

function agentrunOrderRecord(seq: number, traceId: string, type: string, payload: Record<string, unknown>): any {
  const record = canonicalRecord(seq);
  const eventId = `evt_order_${seq}`;
  record.offset = String(1600 + seq);
  record.value.eventId = eventId;
  record.value.sourceSeq = seq;
  record.value.traceId = traceId;
  record.value.hwlabSessionId = HWLAB_SESSION_ID;
  record.value.run.hwlabSessionId = HWLAB_SESSION_ID;
  record.value.event = {
    id: eventId,
    runId: RUN_ID,
    seq,
    type,
    payload: {
      traceId,
      sessionId: SESSION_ID,
      hwlabSessionId: HWLAB_SESSION_ID,
      ...payload
    },
    createdAt: `2026-07-11T11:00:${String(seq).padStart(2, "0")}.000Z`
  };
  return record;
}

function agentrunAssistantLifecycleRecord(seq: number, ordinal: number, type: string, payload: Record<string, unknown>): any {
  const record = agentrunOrderRecord(seq, TRACE_ID, type, payload);
  const createdAt = `2026-07-12T10:00:0${ordinal}.000Z`;
  record.value.committedAt = createdAt;
  record.value.event.createdAt = createdAt;
  return record;
}

function hwlabOrderRecord(source: any, index: number): any {
  const valueText = JSON.stringify(source.value);
  const projected = projectAgentRunKafkaEventToHwlabEvent(source.value, {
    producedAt: `2026-07-11T11:01:${String(index).padStart(2, "0")}.000Z`,
    sourceTopic: source.topic,
    sourcePartition: source.partition,
    sourceOffset: source.offset,
    sourceKey: source.key,
    inputSha256: createSha256(valueText)
  });
  return {
    topic: "hwlab.event.v1",
    partition: 0,
    offset: String(1901 + index),
    key: HWLAB_SESSION_ID,
    valueSha256: createSha256(JSON.stringify(projected)),
    value: projected
  };
}

function canonicalRecord(seq: number): any {
  const eventId = `evt_${seq}`;
  const value = {
    schema: "agentrun.event.v1",
    eventType: "agentrun.event.committed",
    source: "agentrun-manager",
    eventId,
    outboxSeq: seq,
    sourceSeq: seq,
    committedAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`,
    traceId: TRACE_ID,
    sessionId: SESSION_ID,
    hwlabSessionId: HWLAB_SESSION_ID,
    runId: RUN_ID,
    commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e",
    run: {
      runId: RUN_ID,
      status: "running",
      terminalStatus: null,
      failureKind: null,
      tenantId: "tenant-test",
      projectId: "project-test",
      providerId: "gpt.pika",
      backendProfile: "gpt.pika",
      sessionId: SESSION_ID,
      hwlabSessionId: HWLAB_SESSION_ID,
      conversationId: null,
      threadId: null,
      valuesPrinted: false
    },
    command: {
      commandId: "cmd_9fff03fc124b4203b065d04bbb7c4f1e",
      runId: RUN_ID,
      seq: 1,
      type: "run",
      state: "running",
      payloadHash: "hash",
      valuesPrinted: false
    },
    event: {
      id: eventId,
      runId: RUN_ID,
      seq,
      type: "backend_status",
      payload: { traceId: TRACE_ID, sessionId: SESSION_ID, phase: `canonical-${seq}` },
      createdAt: `2026-07-10T11:00:${String(seq).padStart(2, "0")}.000Z`
    },
    valuesPrinted: false
  };
  return { topic: "agentrun.event.v1", partition: 0, offset: String(1600 + seq), key: SESSION_ID, value };
}

function hwlabRecord(sourceSeq: number, event: Record<string, unknown>): 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");
}