Merge pull request #2478 from pikasTech/fix/2474-debug-replay-diagnostics

修复隔离 Kafka 重放 barrier 与分层诊断

docs/MDTODO/details/pure-kafka-live-single-step-debug/R2.3_Task_Report.md

@@ -0,0 +1,72 @@
+# R2.3 隔离 Kafka 重放分层诊断任务报告
+
+## 上下文
+
+- 主问题：
+  - [HWLAB #2474](https://github.com/pikasTech/HWLAB/issues/2474)
+  - [主问题调查结论](https://github.com/pikasTech/HWLAB/issues/2474#issuecomment-4935408473)
+- 隔离重放专项：
+  - [HWLAB #2476](https://github.com/pikasTech/HWLAB/issues/2476)
+  - [隔离重放根因](https://github.com/pikasTech/HWLAB/issues/2476#issuecomment-4935398125)
+
+## 完成结果
+
+- AgentRun 到 HWLAB 的 debug producer 增加批次关联合同：
+  - 每批生成或接受一个 `replayId`；
+  - 每个 `hwlab.event.debug.v1` envelope 保留 `replayId`、批次序号和原始 AgentRun lineage；
+  - 正常 Kafka 调试默认读取 canonical `agentrun.event.v1`；
+  - `agentrun.event.debug.v1` 仅由调用方显式选择，用于 stdio partial reconstruction；
+  - Kafka 读取只做映射预检，发布必须显式指定 `--publish`。
+- debug producer 形成确定的输出 barrier：
+  - 只向 YAML 指定的隔离 debug topic 发布；
+  - 所有批次记录固定写入 debug topic 的 partition 0；
+  - 由 Kafka publish metadata 返回 `firstOffset`、`lastOffset` 和 `count`；
+  - 发布后无法形成唯一 offset range 时，以 `debug_publish_barrier_missing` 失败关闭。
+- debug SSE 在 JSON parse 和 trace/replay filter 前采集 broker 事实：
+  - 报告 `scanned`、`barrierScanned`、`parsed`、`traceMatched`、`replayMatched`、`matched` 和 `delivered`；
+  - 分别保留 requested、scanned、matched 和 delivered offset range；
+  - completion 保留第一条匹配事件的脱敏 source lineage；
+  - `clientReceived`、`decoded` 和 `applied` 在客户端尚未回报时保持 `null`，不伪造成功计数。
+- 新增分层结果合同：
+  - producer/source：`producer_not_invoked`、`source_trace_missing`；
+  - Kafka/server：`topic_no_append_for_replay`、`record_parse_rejected`、`records_scanned_but_filter_mismatch`、`sse_write_failed`；
+  - client：`transport_failed`、`decoder_rejected`、`reducer_rejected`；
+  - terminal：`terminal_missing`、`terminal_complete`。
+- 输入合同失败关闭：
+  - 不完整或逆序 offset range 返回 `workbench_kafka_debug_offset_range_invalid`；
+  - 非 `rpl_` 安全标识返回 `workbench_kafka_debug_replay_id_invalid`；
+  - product topic 和 product group 继续禁止作为隔离 debug 资源。
+
+## 验证
+
+- 定向命令：
+
+```bash
+bun test \
+  internal/cloud/workbench-kafka-debug-replay-contract.test.ts \
+  internal/cloud/workbench-kafka-sse-debug.test.ts \
+  internal/cloud/kafka-event-bridge.test.ts \
+  tools/hwlab-cli/kafka-regenerate.test.ts
+```
+
+- 结果：
+  - `45 pass`；
+  - `0 fail`。
+- 已覆盖四类 MDTODO 验收结果：
+  - empty：`producer_not_invoked`；
+  - matched-no-terminal：`terminal_missing`；
+  - reducer-rejected：`reducer_rejected`；
+  - terminal-complete：`terminal_complete`。
+- 产品路径回归：
+  - `hwlab` 与 `stdio` diagnostic stream 既有测试通过；
+  - live Kafka direct publish/fanout 测试通过；
+  - 未启用 projector、read model、snapshot、sync、recovery replay 或 polling。
+
+## 边界与后续集成
+
+- 本子任务未修改 `WorkbenchKafkaDebugPanel.vue` 或其他 UI。
+- 本子任务未部署或修改 NC01/v03 运行面。
+- UI 集成时应：
+  - 把 producer 返回的 `replayId` 和 `outputBarrier` 原样传入 debug SSE；
+  - 把客户端实际 `received/decoded/applied` 计数交给同一分类合同；
+  - 不把 debug topic、debug group 或结果合同接入产品实时事实源。

docs/reference/kafka-source-direct-debug.md

@@ -9,15 +9,17 @@
 
 - Kafka 输入适合验证真实 topic 中的 session 事件：
   - 命令为 `hwlab-cli kafka regenerate hwlab --from kafka --session-id <sessionId>`；
-  - 默认输入 topic 为 `agentrun.event.debug.v1`；
-  - 产品 canonical 输入可显式指定 `--input-topic agentrun.event.v1`；
+  - 默认输入 topic 优先读取 owning YAML 注入的 `HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC`；
+  - 未注入时默认读取 canonical `agentrun.event.v1`；
+  - stdio partial reconstruction 必须显式指定 `--input-topic agentrun.event.debug.v1`；
   - output topic 优先读取 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC`；
   - consumer group prefix 优先读取 owning YAML 注入的 `HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX`；
-  - 未注入时使用离线合同值 `hwlab.event.debug.v1` 与 `hwlab-v03-workbench-isolated-debug`；
+  - output topic 未注入时使用隔离合同值 `hwlab.event.debug.v1`；
+  - Kafka 读取或发布缺少 group prefix 时失败关闭，不生成代码内运维默认值；
   - Workbench 消费 debug topic 由 `HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED` 独立控制；
   - CLI 的离线映射不依赖该运行面开关；
-  - Kafka 输入默认发布到独立 HWLAB debug topic；
-  - `--no-publish` 可只生成本地证据。
+  - Kafka 输入默认只做映射预检和本地证据固化；
+  - 只有显式 `--publish` 才发布到独立 HWLAB debug topic。
 
 - JSONL 输入适合完全离线的单步测试：
   - 命令为 `hwlab-cli kafka regenerate hwlab --from jsonl --session-id <sessionId> --jsonl-file <path> --no-publish`；
@@ -35,6 +37,11 @@
 
 - 输出验证采用一对一和原序映射：
   - `--expect-count <n>` 在映射和发布前锁定输入数量；
+  - 每次预检生成 `replayId`，也可用 `--replay-id rpl_...` 显式复用同一批次标识；
+  - `next.command` 保留同一个 `replayId`，用于先预检、再显式发布；
+  - 发布固定使用隔离 debug topic 的 partition 0；
+  - 成功发布必须从 Kafka metadata 形成 `firstOffset`、`lastOffset` 和 `count` barrier；
+  - 无法形成唯一 barrier 时以 `debug_publish_barrier_missing` 失败关闭；
   - 命令验证 source sequence、trace、raw AgentRun session、派生 HWLAB session 和源消息哈希；
   - 默认输出为 compact text；
   - 只有显式 `--json` 才输出结构化 JSON；

internal/cloud/kafka-event-bridge.ts

@@ -12,7 +12,7 @@ import {
 } from "./workbench-realtime-capabilities.ts";
 
 const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
-const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1";
+export const DEFAULT_AGENTRUN_EVENT_TOPIC = "agentrun.event.v1";
 const DEFAULT_HWLAB_EVENT_TOPIC = "hwlab.event.v1";
 const DEFAULT_STDIO_TOPIC = "codex-stdio.raw.v1";
 export const DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC = "agentrun.event.debug.v1";
@@ -733,6 +733,9 @@ export function projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage = {},
     };
   }
   const debugSessionId = decoded.reconstruction ? stringValue(decoded.event.hwlabSessionId) : stringValue(projected.sessionId);
+  const replayId = stringValue(options.replayId);
+  const replayIndex = integerValue(options.replayIndex);
+  const replayCount = integerValue(options.replayCount);
   const debugEnvelope = {
     ...projected,
     schema: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC,
@@ -742,6 +745,15 @@ export function projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage = {},
     event: decoded.reconstruction ? { ...projected.event, sessionId: debugSessionId } : projected.event,
     outboxSeq: null,
     ...(decoded.reconstruction ? { eventId: null, sourceEventId: null } : {}),
+    ...(replayId ? {
+      replayId,
+      debugReplay: {
+        replayId,
+        index: replayIndex,
+        count: replayCount,
+        valuesPrinted: false
+      }
+    } : {}),
     diagnostics: {
       projectedBy: "hwlab-kafka-event-bridge",
       projectionMode: "source-direct-debug",
@@ -759,6 +771,7 @@ export function projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage = {},
       inputRunId: firstText(decoded.event.runId, decoded.event.run?.runId),
       inputCommandId: firstText(decoded.event.commandId, decoded.event.command?.commandId),
       sourceSeq: integerValue(decoded.event.sourceSeq ?? decoded.event.event?.seq),
+      replayId,
       valuesPrinted: false
     },
     ...(decoded.reconstruction ? {
@@ -977,19 +990,23 @@ export async function queryKafkaEventStream({ env = process.env, stream = "hwlab
   };
 }
 
-export async function openKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, groupIdPrefix = null, traceId = null, sessionId = null, runId = null, commandId = null, fromBeginning = false, onEvent, onError = null, kafkaFactory = defaultKafkaFactory } = {}) {
+export async function openKafkaEventStream({ env = process.env, stream = "hwlab", topic = null, groupIdPrefix = null, traceId = null, sessionId = null, runId = null, commandId = null, replayId = null, offsetRange = null, fromBeginning = false, onEvent, onRecord = null, onBarrierReady = null, onBarrierComplete = null, onError = null, kafkaFactory = defaultKafkaFactory } = {}) {
   if (typeof onEvent !== "function") throw new Error("onEvent callback is required for Kafka event streaming.");
   const brokers = csv(env.HWLAB_KAFKA_BOOTSTRAP_SERVERS);
   if (brokers.length === 0) throw new Error("HWLAB_KAFKA_BOOTSTRAP_SERVERS is required for Kafka event streams.");
   const clientId = stringValue(env.HWLAB_KAFKA_CLIENT_ID) || DEFAULT_CLIENT_ID;
   const resolvedTopic = stringValue(topic) || kafkaTopicForStream(stream, env);
-  const filters = compactObject({ traceId, sessionId, runId, commandId });
+  const filters = compactObject({ traceId, sessionId, runId, commandId, replayId });
+  const resolvedOffsetRange = normalizeKafkaOffsetRange(offsetRange);
   const kafka = kafkaFactory({ brokers, clientId: `${clientId}-debug-sse` });
   const resolvedGroupIdPrefix = stringValue(groupIdPrefix) || `${clientId}-debug-sse`;
   const groupId = `${resolvedGroupIdPrefix}-${Date.now()}-${randomUUID().slice(0, 8)}`;
   const consumer = kafka.consumer({ groupId, allowAutoTopicCreation: false });
   let running = false;
   let stopped = false;
+  let seekApplied = false;
+  let barrierCompleted = false;
+  const seenRecordOffsets = new Set();
   await consumer.connect();
   await consumer.subscribe({ topic: resolvedTopic, fromBeginning: fromBeginning === true });
   running = true;
@@ -997,19 +1014,61 @@ export async function openKafkaEventStream({ env = process.env, stream = "hwlab"
     await consumer.run({
       eachMessage: async ({ topic: messageTopic, partition, message }) => {
         if (stopped) return;
+        if (resolvedOffsetRange && !seekApplied) return;
         const valueText = message.value ? Buffer.from(message.value).toString("utf8") : "";
         const value = parseJson(valueText);
-        if (!value || !eventMatchesFilters(value, filters)) return;
-        await onEvent({
-          topic: messageTopic,
-          partition,
-          offset: message.offset,
-          key: message.key ? Buffer.from(message.key).toString("utf8") : null,
-          timestamp: message.timestamp ?? null,
-          value
-        });
+        const parsed = Boolean(value && typeof value === "object" && !Array.isArray(value));
+        const inOffsetRange = kafkaRecordInOffsetRange(partition, message.offset, resolvedOffsetRange);
+        const filterMatches = parsed ? eventFilterResults(value, filters) : {};
+        const offsetKey = `${messageTopic}:${partition}:${message.offset}`;
+        const duplicate = Boolean(resolvedOffsetRange && seenRecordOffsets.has(offsetKey));
+        if (resolvedOffsetRange) seenRecordOffsets.add(offsetKey);
+        const matched = !duplicate && parsed && inOffsetRange && Object.values(filterMatches).every(Boolean);
+        if (typeof onRecord === "function") {
+          await onRecord({
+            topic: messageTopic,
+            partition,
+            offset: message.offset,
+            parsed,
+            duplicate,
+            inOffsetRange,
+            filterMatches,
+            matched,
+            valuesPrinted: false
+          });
+        }
+        if (matched) {
+          await onEvent({
+            topic: messageTopic,
+            partition,
+            offset: message.offset,
+            key: message.key ? Buffer.from(message.key).toString("utf8") : null,
+            timestamp: message.timestamp ?? null,
+            value
+          });
+        }
+        if (!barrierCompleted && kafkaRecordCompletesOffsetRange(partition, message.offset, resolvedOffsetRange)) {
+          barrierCompleted = true;
+          if (typeof onBarrierComplete === "function") {
+            await onBarrierComplete({
+              topic: messageTopic,
+              partition,
+              offset: message.offset,
+              offsetRange: resolvedOffsetRange,
+              valuesPrinted: false
+            });
+          }
+        }
       }
     });
+    if (resolvedOffsetRange) {
+      if (typeof consumer.seek !== "function") throw new Error("Kafka consumer seek is required for correlated replay offset barriers.");
+      await consumer.seek({ topic: resolvedTopic, partition: resolvedOffsetRange.partition, offset: resolvedOffsetRange.firstOffset });
+      seekApplied = true;
+      if (typeof onBarrierReady === "function") {
+        await onBarrierReady({ topic: resolvedTopic, offsetRange: resolvedOffsetRange, seekApplied: true, valuesPrinted: false });
+      }
+    }
   } catch (error) {
     if (typeof onError === "function") onError(error);
     stopped = true;
@@ -1023,6 +1082,8 @@ export async function openKafkaEventStream({ env = process.env, stream = "hwlab"
     topic: resolvedTopic,
     groupId,
     filters,
+    offsetRange: resolvedOffsetRange,
+    seekApplied,
     async stop() {
       stopped = true;
       if (running) await consumer.stop().catch(() => undefined);
@@ -1105,12 +1166,17 @@ function mapAgentRunSourceEventToHwlabEvent({ input, sourceEvent, payload, run,
 }
 
 function eventMatchesFilters(value, filters) {
+  return Object.values(eventFilterResults(value, filters)).every(Boolean);
+}
+
+function eventFilterResults(value, filters) {
+  const results = {};
   for (const [name, expected] of Object.entries(filters)) {
     const needle = stringValue(expected);
     if (!needle) continue;
-    if (!eventFieldCandidates(value, name).includes(needle)) return false;
+    results[name] = eventFieldCandidates(value, name).includes(needle);
   }
-  return true;
+  return results;
 }
 
 function eventFieldCandidates(value, name) {
@@ -1124,15 +1190,47 @@ function eventFieldCandidates(value, name) {
   const ids = objectValue(value.ids);
   const stdio = objectValue(value.stdio ?? value.frame ?? value.codexStdio);
   const sourceEvent = objectValue(value.sourceEvent);
+  const debugReplay = objectValue(value.debugReplay);
+  const debugLineage = objectValue(value.debugLineage);
   const nestedEvent = objectValue(value.event?.sourceEvent ?? value.agentRunEvent ?? payload.event);
   const nestedPayload = objectValue(nestedEvent.payload);
   if (name === "traceId") return textCandidates(value.traceId, value.trace_id, event.traceId, event.trace_id, context.traceId, context.trace_id, sourceEvent.traceId, sourceEvent.trace_id, nestedEvent.traceId, nestedEvent.trace_id, nestedPayload.traceId, nestedPayload.trace_id, payload.traceId, payload.trace_id, metadata.traceId, metadata.trace_id, trace.traceId, trace.trace_id, ids.traceId, ids.trace_id, stdio.traceId, stdio.trace_id);
   if (name === "sessionId") return textCandidates(value.sessionId, value.session_id, event.sessionId, event.session_id, context.sessionId, context.session_id, run.sessionId, run.session_id, nestedPayload.sessionId, nestedPayload.session_id, payload.sessionId, payload.session_id, metadata.sessionId, metadata.session_id, ids.sessionId, ids.session_id, stdio.sessionId, stdio.session_id);
   if (name === "runId") return textCandidates(value.runId, value.run_id, event.runId, event.run_id, context.runId, context.run_id, run.runId, run.run_id, nestedEvent.runId, nestedEvent.run_id, nestedPayload.runId, nestedPayload.run_id, payload.runId, payload.run_id, metadata.runId, metadata.run_id, ids.runId, ids.run_id, stdio.runId, stdio.run_id);
   if (name === "commandId") return textCandidates(value.commandId, value.command_id, event.commandId, event.command_id, context.commandId, context.command_id, command.commandId, command.command_id, nestedEvent.commandId, nestedEvent.command_id, nestedPayload.commandId, nestedPayload.command_id, payload.commandId, payload.command_id, metadata.commandId, metadata.command_id, ids.commandId, ids.command_id, stdio.commandId, stdio.command_id);
+  if (name === "replayId") return textCandidates(value.replayId, value.replay_id, debugReplay.replayId, debugReplay.replay_id, debugLineage.replayId, debugLineage.replay_id);
   return [];
 }
 
+function normalizeKafkaOffsetRange(value) {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+  const partition = integerValue(value.partition);
+  const firstOffset = kafkaOffsetValue(value.firstOffset ?? value.startOffset);
+  const lastOffset = kafkaOffsetValue(value.lastOffset ?? value.endOffset);
+  if (partition === null || firstOffset === null || lastOffset === null || BigInt(firstOffset) > BigInt(lastOffset)) return null;
+  return { partition, firstOffset, lastOffset };
+}
+
+function kafkaRecordInOffsetRange(partitionValue, offsetValue, range) {
+  if (!range) return true;
+  const partition = integerValue(partitionValue);
+  const offset = kafkaOffsetValue(offsetValue);
+  return partition === range.partition && offset !== null && BigInt(offset) >= BigInt(range.firstOffset) && BigInt(offset) <= BigInt(range.lastOffset);
+}
+
+function kafkaRecordCompletesOffsetRange(partitionValue, offsetValue, range) {
+  if (!range) return false;
+  const partition = integerValue(partitionValue);
+  const offset = kafkaOffsetValue(offsetValue);
+  return partition === range.partition && offset !== null && BigInt(offset) === BigInt(range.lastOffset);
+}
+
+function kafkaOffsetValue(value) {
+  const text = stringValue(value);
+  if (!text || !/^\d+$/u.test(text)) return null;
+  try { return BigInt(text).toString(); } catch { return null; }
+}
+
 function kafkaTopicForStream(stream, env) {
   const normalized = stringValue(stream) || "hwlab";
   if (normalized === "stdio" || normalized === "codex-stdio") return stringValue(env.HWLAB_KAFKA_STDIO_TOPIC ?? env.AGENTRUN_KAFKA_STDIO_TOPIC) || DEFAULT_STDIO_TOPIC;
@@ -1178,7 +1276,8 @@ function kafkaHeaders(projected) {
     "x-session-id": projected.sessionId ?? "",
     "x-run-id": projected.context?.runId ?? "",
     "x-command-id": projected.context?.commandId ?? "",
-    "x-values-printed": "false"
+    "x-values-printed": "false",
+    ...(projected.replayId ? { "x-debug-replay-id": projected.replayId } : {})
   };
 }
 

internal/cloud/workbench-kafka-debug-replay-contract.test.ts

@@ -0,0 +1,236 @@
+import assert from "node:assert/strict";
+import { test } from "bun:test";
+
+import {
+  classifyWorkbenchKafkaDebugReplayClientOverlay,
+  classifyWorkbenchKafkaDebugReplayResult,
+  completeWorkbenchKafkaDebugReplay,
+  createWorkbenchKafkaDebugReplayTracker,
+  markWorkbenchKafkaDebugReplayBarrierComplete,
+  markWorkbenchKafkaDebugReplayConsumerReady,
+  observeWorkbenchKafkaDebugReplayRecord,
+  recordWorkbenchKafkaDebugReplayDelivery
+} from "./workbench-kafka-debug-replay-contract.ts";
+
+const readyConsumer = { ready: true, groupId: "hwlab-debug-test" };
+const deliveredCounts = {
+  scanned: 1,
+  barrierScanned: 1,
+  parsed: 1,
+  traceMatched: 1,
+  replayMatched: 1,
+  matched: 1,
+  delivered: 1,
+  clientReceived: 1,
+  decoded: 1,
+  applied: 1
+};
+
+test("isolated replay classifies an empty producer result", () => {
+  const result = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_empty",
+    traceId: "trc_empty",
+    producer: { invoked: false, sourceMatched: null },
+    consumer: readyConsumer,
+    counts: {},
+    terminalObserved: false
+  });
+  assert.deepEqual(result, {
+    ok: false,
+    phase: "producer",
+    code: "producer_not_invoked",
+    message: "No isolated debug producer was invoked for this replay."
+  });
+});
+
+test("isolated replay classifies matching events without terminal", () => {
+  const result = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_no_terminal",
+    traceId: "trc_no_terminal",
+    producer: { invoked: true, sourceMatched: true },
+    consumer: readyConsumer,
+    counts: deliveredCounts,
+    terminalObserved: false
+  });
+  assert.equal(result.code, "terminal_missing");
+  assert.equal(result.phase, "terminal");
+});
+
+test("isolated replay classifies production reducer rejection", () => {
+  const result = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_reducer_rejected",
+    traceId: "trc_reducer_rejected",
+    producer: { invoked: true, sourceMatched: true },
+    consumer: readyConsumer,
+    counts: { ...deliveredCounts, applied: 0 },
+    terminalObserved: true
+  });
+  assert.equal(result.code, "reducer_rejected");
+  assert.equal(result.phase, "client");
+});
+
+test("isolated replay classifies terminal completion", () => {
+  const result = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_complete",
+    traceId: "trc_complete",
+    producer: { invoked: true, sourceMatched: true },
+    consumer: readyConsumer,
+    counts: deliveredCounts,
+    terminalObserved: true
+  });
+  assert.deepEqual(result, {
+    ok: true,
+    phase: "terminal",
+    code: "terminal_complete",
+    message: "The isolated replay reached a terminal event."
+  });
+});
+
+test("isolated replay keeps source, filter, and decoder failures distinct", () => {
+  const sourceMissing = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_source_missing",
+    traceId: "trc_source_missing",
+    producer: { invoked: true, sourceMatched: false },
+    consumer: readyConsumer,
+    counts: { scanned: 2, barrierScanned: 2, parsed: 2, traceMatched: 0 },
+    terminalObserved: false
+  });
+  const filterMismatch = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_filter_mismatch",
+    traceId: "trc_filter_mismatch",
+    producer: { invoked: true, sourceMatched: true },
+    consumer: readyConsumer,
+    counts: { scanned: 2, barrierScanned: 2, parsed: 2, traceMatched: 2, replayMatched: 0 },
+    terminalObserved: false
+  });
+  const decoderRejected = classifyWorkbenchKafkaDebugReplayResult({
+    replayId: "rpl_decoder_rejected",
+    traceId: "trc_decoder_rejected",
+    producer: { invoked: true, sourceMatched: true },
+    consumer: readyConsumer,
+    counts: { ...deliveredCounts, decoded: 0, applied: 0 },
+    terminalObserved: true
+  });
+  assert.equal(sourceMissing.code, "source_trace_missing");
+  assert.equal(filterMismatch.code, "records_scanned_but_filter_mismatch");
+  assert.equal(decoderRejected.code, "decoder_rejected");
+});
+
+test("correlated replay requires seek, lastOffset, and complete unique batch cardinality", () => {
+  const complete = correlatedReplay();
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(complete).code, "terminal_complete");
+
+  const seekMissing = correlatedReplay({ barrier: { seekApplied: false, completed: true, lastOffsetObserved: true } });
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(seekMissing).code, "consumer_seek_not_applied");
+
+  const lastOffsetMissing = correlatedReplay({ barrier: { seekApplied: true, completed: false, lastOffsetObserved: false } });
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(lastOffsetMissing).code, "barrier_not_completed");
+
+  const middleOffsetMissing = correlatedReplay({
+    requestedOffsetRange: { partition: 0, firstOffset: "10", lastOffset: "12" },
+    producer: { invoked: true, sourceMatched: true, publishedCount: 3 }
+  });
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(middleOffsetMissing).code, "barrier_cardinality_mismatch");
+
+  const duplicateOffset = correlatedReplay({ rejectedByReason: { duplicate_offset: 1 } });
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(duplicateOffset).code, "barrier_duplicate_offset");
+});
+
+test("correlated replay fails typed parse, filter, delivery, and producer evidence loss", () => {
+  const parseLost = correlatedReplay({ counts: { ...correlatedCounts(), parsed: 1, traceMatched: 1, replayMatched: 1, matched: 1, delivered: 1 } });
+  const filterLost = correlatedReplay({ counts: { ...correlatedCounts(), traceMatched: 1, replayMatched: 1, matched: 1, delivered: 1 } });
+  const deliveryLost = correlatedReplay({ counts: { ...correlatedCounts(), delivered: 1 } });
+  const producerFalse = correlatedReplay({ producer: { invoked: false, sourceMatched: true, publishedCount: 2 } });
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(parseLost).code, "record_parse_rejected");
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(filterLost).code, "records_scanned_but_filter_mismatch");
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(deliveryLost).code, "sse_write_failed");
+  assert.equal(classifyWorkbenchKafkaDebugReplayResult(producerFalse).code, "producer_not_invoked");
+});
+
+test("server result declares client counts unavailable and client overlay reuses the pure classifier", () => {
+  const serverResult = {
+    ...correlatedReplay(),
+    offsetRange: { requested: { partition: 0, firstOffset: "10", lastOffset: "11" } },
+    completionReason: "barrier",
+    classificationScope: "server",
+    requiresClientOverlay: true,
+    clientCounts: { available: false, received: null, decoded: null, applied: null }
+  };
+  const overlay = classifyWorkbenchKafkaDebugReplayClientOverlay(serverResult, { received: 2, decoded: 2, applied: 1 });
+  assert.equal(overlay.code, "reducer_rejected");
+  assert.equal(overlay.classificationScope, "client");
+  assert.equal(overlay.requiresClientOverlay, false);
+  assert.deepEqual(overlay.clientCounts, { available: true, received: 2, decoded: 2, applied: 1, source: "client-local-overlay" });
+});
+
+test("correlated tracker counts unique offsets and rejects a duplicate even after lastOffset", () => {
+  const tracker = createWorkbenchKafkaDebugReplayTracker({
+    mode: "correlated-v2",
+    replayId: "rpl_unique_offsets",
+    traceId: "trc_unique_offsets",
+    offsetRange: { partition: 0, firstOffset: "10", lastOffset: "11" },
+    producer: { invoked: true, sourceMatched: true, publishedCount: 2 }
+  });
+  markWorkbenchKafkaDebugReplayConsumerReady(tracker, { groupId: "hwlab-debug-unique", seekApplied: true });
+  const observation = (offset: string, duplicate = false) => ({
+    partition: 0,
+    offset,
+    duplicate,
+    inOffsetRange: true,
+    parsed: true,
+    filterMatches: { traceId: true, replayId: true },
+    matched: !duplicate
+  });
+  observeWorkbenchKafkaDebugReplayRecord(tracker, observation("10"));
+  observeWorkbenchKafkaDebugReplayRecord(tracker, observation("10", true));
+  observeWorkbenchKafkaDebugReplayRecord(tracker, observation("11"));
+  recordWorkbenchKafkaDebugReplayDelivery(tracker, { partition: 0, offset: "10", value: {} });
+  recordWorkbenchKafkaDebugReplayDelivery(tracker, { partition: 0, offset: "11", value: {} });
+  markWorkbenchKafkaDebugReplayBarrierComplete(tracker);
+  const result = completeWorkbenchKafkaDebugReplay(tracker, { reason: "barrier", terminalObserved: true });
+  assert.equal(result.counts.scanned, 3);
+  assert.equal(result.counts.barrierScanned, 2);
+  assert.equal(result.barrier.uniqueOffsetCount, 2);
+  assert.equal(result.rejectedByReason.duplicate_offset, 1);
+  assert.equal(result.code, "barrier_duplicate_offset");
+  assert.equal(result.classificationScope, "server");
+  assert.equal(result.requiresClientOverlay, true);
+  assert.deepEqual(result.clientCounts, {
+    available: false,
+    received: null,
+    decoded: null,
+    applied: null,
+    source: "client-local-overlay-required"
+  });
+});
+
+function correlatedReplay(overrides: Record<string, any> = {}) {
+  return {
+    mode: "correlated-v2",
+    replayId: "rpl_correlated",
+    traceId: "trc_correlated",
+    producer: { invoked: true, sourceMatched: true, publishedCount: 2 },
+    consumer: readyConsumer,
+    requestedOffsetRange: { partition: 0, firstOffset: "10", lastOffset: "11" },
+    barrier: { seekApplied: true, completed: true, lastOffsetObserved: true },
+    rejectedByReason: {},
+    counts: correlatedCounts(),
+    terminalObserved: true,
+    ...overrides
+  };
+}
+
+function correlatedCounts() {
+  return {
+    scanned: 2,
+    barrierScanned: 2,
+    parsed: 2,
+    traceMatched: 2,
+    replayMatched: 2,
+    matched: 2,
+    delivered: 2,
+    clientReceived: null,
+    decoded: null,
+    applied: null
+  };
+}

internal/cloud/workbench-kafka-debug-replay-contract.ts

@@ -0,0 +1,375 @@
+// Responsibility: classify isolated Kafka debug replay progress without turning debug state into a product authority.
+
+export const WORKBENCH_KAFKA_DEBUG_REPLAY_CONTRACT_VERSION = "workbench-kafka-debug-replay-v2";
+
+export function createWorkbenchKafkaDebugReplayTracker(input = {}) {
+  return {
+    mode: replayMode(input.mode),
+    replayId: textValue(input.replayId),
+    traceId: textValue(input.traceId),
+    topic: textValue(input.topic),
+    producer: {
+      invoked: booleanOrNull(input.producer?.invoked),
+      sourceMatched: booleanOrNull(input.producer?.sourceMatched),
+      publishedCount: integerOrNull(input.producer?.publishedCount)
+    },
+    consumer: { ready: false, groupId: null },
+    barrier: {
+      seekApplied: false,
+      completed: false,
+      lastOffsetObserved: false,
+      firstOffsetObserved: null,
+      finalOffsetObserved: null,
+      uniqueOffsetCount: 0
+    },
+    requestedOffsetRange: normalizeOffsetRange(input.offsetRange),
+    counts: {
+      scanned: 0,
+      barrierScanned: 0,
+      parsed: 0,
+      traceMatched: 0,
+      replayMatched: 0,
+      matched: 0,
+      delivered: 0,
+      clientReceived: null,
+      decoded: null,
+      applied: null
+    },
+    rejectedByReason: {},
+    offsets: {
+      scanned: new Map(),
+      barrier: new Map(),
+      matched: new Map(),
+      delivered: new Map()
+    },
+    sourceLineage: null
+  };
+}
+
+export function markWorkbenchKafkaDebugReplayConsumerReady(tracker, input = {}) {
+  tracker.consumer.ready = true;
+  tracker.consumer.groupId = textValue(input.groupId);
+  tracker.barrier.seekApplied = input.seekApplied === true;
+  return tracker;
+}
+
+export function markWorkbenchKafkaDebugReplayBarrierComplete(tracker) {
+  tracker.barrier.completed = true;
+  tracker.barrier.lastOffsetObserved = true;
+  return tracker;
+}
+
+export function observeWorkbenchKafkaDebugReplayRecord(tracker, observation = {}) {
+  tracker.counts.scanned += 1;
+  recordOffset(tracker.offsets.scanned, observation.partition, observation.offset);
+  if (observation.duplicate === true) {
+    reject(tracker, "duplicate_offset");
+    return tracker;
+  }
+  if (observation.inOffsetRange === false) {
+    reject(tracker, "offset_range_mismatch");
+    return tracker;
+  }
+  tracker.counts.barrierScanned += 1;
+  recordOffset(tracker.offsets.barrier, observation.partition, observation.offset);
+  const observed = summarizeOffsets(tracker.offsets.barrier);
+  tracker.barrier.firstOffsetObserved = observed[0]?.firstOffset ?? null;
+  tracker.barrier.finalOffsetObserved = observed.at(-1)?.lastOffset ?? null;
+  tracker.barrier.uniqueOffsetCount = tracker.counts.barrierScanned;
+  if (observation.parsed !== true) {
+    reject(tracker, "record_parse_rejected");
+    return tracker;
+  }
+  tracker.counts.parsed += 1;
+  if (observation.filterMatches?.traceId === true) tracker.counts.traceMatched += 1;
+  if (observation.filterMatches?.replayId === true) tracker.counts.replayMatched += 1;
+  if (observation.matched !== true) {
+    reject(tracker, "filter_mismatch");
+    return tracker;
+  }
+  tracker.counts.matched += 1;
+  recordOffset(tracker.offsets.matched, observation.partition, observation.offset);
+  return tracker;
+}
+
+export function recordWorkbenchKafkaDebugReplayDelivery(tracker, record = {}, delivered = true) {
+  if (!delivered) {
+    reject(tracker, "sse_write_failed");
+    return tracker;
+  }
+  tracker.counts.delivered += 1;
+  recordOffset(tracker.offsets.delivered, record.partition, record.offset);
+  if (!tracker.sourceLineage) tracker.sourceLineage = safeSourceLineage(record.value);
+  return tracker;
+}
+
+export function completeWorkbenchKafkaDebugReplay(tracker, input = {}) {
+  const counts = {
+    ...tracker.counts,
+    clientReceived: nullableCount(input.clientReceived, tracker.counts.clientReceived),
+    decoded: nullableCount(input.decoded, tracker.counts.decoded),
+    applied: nullableCount(input.applied, tracker.counts.applied)
+  };
+  const terminalObserved = input.terminalObserved === true;
+  const result = classifyWorkbenchKafkaDebugReplayResult({
+    mode: tracker.mode,
+    replayId: tracker.replayId,
+    traceId: tracker.traceId,
+    producer: tracker.producer,
+    consumer: tracker.consumer,
+    requestedOffsetRange: tracker.requestedOffsetRange,
+    counts,
+    completionReason: textValue(input.reason),
+    terminalObserved,
+    barrier: tracker.barrier,
+    rejectedByReason: tracker.rejectedByReason
+  });
+  return {
+    contractVersion: WORKBENCH_KAFKA_DEBUG_REPLAY_CONTRACT_VERSION,
+    mode: tracker.mode,
+    replayId: tracker.replayId,
+    traceId: tracker.traceId,
+    topic: tracker.topic,
+    producer: tracker.producer,
+    consumer: tracker.consumer,
+    barrier: tracker.barrier,
+    phase: result.phase,
+    code: result.code,
+    ok: result.ok,
+    message: result.message,
+    completionReason: textValue(input.reason),
+    terminalObserved,
+    counts,
+    rejectedByReason: { ...tracker.rejectedByReason },
+    offsetRange: {
+      requested: tracker.requestedOffsetRange,
+      scanned: summarizeOffsets(tracker.offsets.scanned),
+      observed: summarizeOffsets(tracker.offsets.barrier),
+      matched: summarizeOffsets(tracker.offsets.matched),
+      delivered: summarizeOffsets(tracker.offsets.delivered)
+    },
+    sourceLineage: tracker.sourceLineage,
+    classificationScope: "server",
+    requiresClientOverlay: true,
+    clientCounts: {
+      available: false,
+      received: null,
+      decoded: null,
+      applied: null,
+      source: "client-local-overlay-required"
+    },
+    valuesPrinted: false
+  };
+}
+
+export function classifyWorkbenchKafkaDebugReplayResult(input = {}) {
+  const mode = replayMode(input.mode);
+  const counts = normalizeCounts(input.counts);
+  const producer = input.producer ?? {};
+  const consumer = input.consumer ?? {};
+  const hasTraceFilter = Boolean(textValue(input.traceId));
+  const hasReplayFilter = Boolean(textValue(input.replayId));
+  const hasOffsetBarrier = Boolean(normalizeOffsetRange(input.requestedOffsetRange));
+
+  if (producer.invoked === false) return failure("producer", "producer_not_invoked", "No isolated debug producer was invoked for this replay.");
+  if (consumer.ready === false) return failure("consumer", "consumer_not_ready", "The isolated Kafka consumer did not reach its ready barrier.");
+  if (producer.sourceMatched === false) return failure("producer", "source_trace_missing", "The source AgentRun topic did not contain the requested trace lineage.");
+  if (mode === "correlated-v2") {
+    const correlated = classifyCorrelatedReplay(input, counts);
+    if (correlated) return correlated;
+  }
+  if (hasOffsetBarrier && counts.barrierScanned === 0) return failure("kafka", "topic_no_append_for_replay", "The requested replay offset range was not appended or observed.");
+  if (counts.scanned === 0) return failure("producer", "producer_not_invoked", "The isolated debug topic contained no records for this replay.");
+  if (counts.barrierScanned > 0 && counts.parsed === 0) return failure("server", "record_parse_rejected", "Kafka records were scanned but none decoded as JSON objects.");
+  if (hasTraceFilter && counts.traceMatched === 0) {
+    if (producer.sourceMatched === true && producer.invoked === true) return failure("kafka", "topic_no_append_for_replay", "The producer reported success but the debug topic did not expose the requested trace append.");
+    if (producer.sourceMatched === true) return failure("producer", "producer_not_invoked", "The source trace exists but no isolated debug producer append was observed.");
+    return failure("producer", "source_trace_missing", "Kafka records were scanned but none carried the requested trace lineage.");
+  }
+  if (hasReplayFilter && counts.replayMatched === 0) return failure("server", "records_scanned_but_filter_mismatch", "Kafka records were scanned but none matched the requested replayId.");
+  if (counts.matched === 0) return failure("server", "records_scanned_but_filter_mismatch", "Kafka records were scanned but none matched the replay filters.");
+  if (counts.delivered < counts.matched && input.completionReason !== "limit") return failure("server", "sse_write_failed", "Matched Kafka records were not all delivered over SSE.");
+  if (counts.clientReceived !== null && counts.clientReceived < counts.delivered) return failure("client", "transport_failed", "The browser received fewer records than the server delivered.");
+  if (counts.decoded !== null && counts.clientReceived !== null && counts.decoded < counts.clientReceived) return failure("client", "decoder_rejected", "The browser decoder rejected one or more delivered records.");
+  if (counts.applied !== null && counts.decoded !== null && counts.applied < counts.decoded) return failure("client", "reducer_rejected", "The production reducer rejected one or more decoded records.");
+  if (input.terminalObserved !== true) return failure("terminal", "terminal_missing", "Matching events were delivered but no terminal event was observed.");
+  return { ok: true, phase: "terminal", code: "terminal_complete", message: "The isolated replay reached a terminal event." };
+}
+
+export function classifyWorkbenchKafkaDebugReplayClientOverlay(serverResult = {}, clientCounts = {}) {
+  const received = nullableCount(clientCounts.received ?? clientCounts.clientReceived, null);
+  const decoded = nullableCount(clientCounts.decoded, null);
+  const applied = nullableCount(clientCounts.applied, null);
+  const counts = { ...normalizeCounts(serverResult.counts), clientReceived: received, decoded, applied };
+  const complete = received !== null && decoded !== null && applied !== null;
+  const result = complete
+    ? classifyWorkbenchKafkaDebugReplayResult({
+      mode: serverResult.mode,
+      replayId: serverResult.replayId,
+      traceId: serverResult.traceId,
+      producer: serverResult.producer,
+      consumer: serverResult.consumer,
+      requestedOffsetRange: serverResult.offsetRange?.requested,
+      counts,
+      completionReason: serverResult.completionReason,
+      terminalObserved: serverResult.terminalObserved,
+      barrier: serverResult.barrier,
+      rejectedByReason: serverResult.rejectedByReason
+    })
+    : failure("client", "client_counts_incomplete", "Client replay classification requires received, decoded, and applied counts.");
+  return {
+    ...serverResult,
+    ...result,
+    counts,
+    classificationScope: "client",
+    requiresClientOverlay: !complete,
+    clientCounts: {
+      available: complete,
+      received,
+      decoded,
+      applied,
+      source: "client-local-overlay"
+    },
+    valuesPrinted: false
+  };
+}
+
+export function normalizeWorkbenchKafkaDebugOffsetRange(value) {
+  return normalizeOffsetRange(value);
+}
+
+export function workbenchKafkaDebugOffsetRangeWidth(value) {
+  const range = normalizeOffsetRange(value);
+  if (!range) return null;
+  const width = BigInt(range.lastOffset) - BigInt(range.firstOffset) + 1n;
+  return width <= BigInt(Number.MAX_SAFE_INTEGER) ? Number(width) : null;
+}
+
+function classifyCorrelatedReplay(input, counts) {
+  const range = normalizeOffsetRange(input.requestedOffsetRange);
+  const expected = workbenchKafkaDebugOffsetRangeWidth(range);
+  const publishedCount = integerOrNull(input.producer?.publishedCount);
+  if (!textValue(input.replayId) || !range) {
+    return failure("contract", "v2_correlation_incomplete", "Correlated replay requires replayId and a complete offset range.");
+  }
+  if (expected === null || (publishedCount !== null && publishedCount !== expected)) return failure("producer", "producer_cardinality_mismatch", "Producer publishedCount does not equal the requested replay offset range width.");
+  if (input.barrier?.seekApplied !== true) return failure("consumer", "consumer_seek_not_applied", "The correlated replay consumer did not seek to the requested firstOffset.");
+  if (input.barrier?.completed !== true || input.barrier?.lastOffsetObserved !== true) return failure("kafka", "barrier_not_completed", "The correlated replay did not observe its requested lastOffset completion boundary.");
+  if (Number(input.rejectedByReason?.duplicate_offset ?? 0) > 0) return failure("kafka", "barrier_duplicate_offset", "The correlated replay observed a duplicate Kafka offset.");
+  if (counts.barrierScanned !== expected) return failure("kafka", "barrier_cardinality_mismatch", "The consumer did not observe every offset in the requested replay barrier.");
+  if (counts.parsed !== expected) return failure("server", "record_parse_rejected", "One or more records in the replay barrier failed JSON object decoding.");
+  if (counts.traceMatched !== expected || counts.replayMatched !== expected || counts.matched !== expected) {
+    return failure("server", "records_scanned_but_filter_mismatch", "One or more records in the replay barrier failed traceId or replayId correlation.");
+  }
+  if (counts.delivered !== expected) return failure("server", "sse_write_failed", "One or more correlated replay records were not delivered over SSE.");
+  return null;
+}
+
+function failure(phase, code, message) {
+  return { ok: false, phase, code, message };
+}
+
+function replayMode(value) {
+  return value === "correlated-v2" ? "correlated-v2" : "trace-only-v1";
+}
+
+function reject(tracker, reason) {
+  tracker.rejectedByReason[reason] = (tracker.rejectedByReason[reason] ?? 0) + 1;
+}
+
+function safeSourceLineage(value) {
+  const lineage = objectValue(value?.debugLineage);
+  const sourceEvent = objectValue(value?.sourceEvent);
+  if (Object.keys(lineage).length === 0) return null;
+  return {
+    inputSchema: textValue(lineage.inputSchema),
+    inputEventType: textValue(lineage.inputEventType),
+    inputTraceId: textValue(lineage.inputTraceId),
+    inputSessionId: textValue(lineage.inputSessionId),
+    inputHwlabSessionId: textValue(lineage.inputHwlabSessionId),
+    inputRunId: textValue(lineage.inputRunId),
+    inputCommandId: textValue(lineage.inputCommandId),
+    sourceSeq: integerOrNull(lineage.sourceSeq),
+    sourceTopic: textValue(sourceEvent.topic),
+    sourcePartition: integerOrNull(sourceEvent.partition),
+    sourceOffset: kafkaOffset(sourceEvent.offset),
+    valuesPrinted: false
+  };
+}
+
+function recordOffset(target, partitionValue, offsetValue) {
+  const partition = integerOrNull(partitionValue);
+  const offset = kafkaOffset(offsetValue);
+  if (partition === null || offset === null) return;
+  const current = target.get(partition);
+  if (!current) {
+    target.set(partition, { partition, firstOffset: offset, lastOffset: offset, count: 1 });
+    return;
+  }
+  if (BigInt(offset) < BigInt(current.firstOffset)) current.firstOffset = offset;
+  if (BigInt(offset) > BigInt(current.lastOffset)) current.lastOffset = offset;
+  current.count += 1;
+}
+
+function summarizeOffsets(target) {
+  return [...target.values()].sort((left, right) => left.partition - right.partition).map((entry) => ({ ...entry }));
+}
+
+function normalizeOffsetRange(value) {
+  if (!value || typeof value !== "object" || Array.isArray(value)) return null;
+  const partition = integerOrNull(value.partition);
+  const firstOffset = kafkaOffset(value.firstOffset ?? value.startOffset);
+  const lastOffset = kafkaOffset(value.lastOffset ?? value.endOffset);
+  if (partition === null || firstOffset === null || lastOffset === null || BigInt(firstOffset) > BigInt(lastOffset)) return null;
+  return { partition, firstOffset, lastOffset };
+}
+
+function normalizeCounts(value = {}) {
+  const nullable = (name) => nullableCount(value[name], null);
+  return {
+    scanned: count(value.scanned),
+    barrierScanned: count(value.barrierScanned),
+    parsed: count(value.parsed),
+    traceMatched: count(value.traceMatched),
+    replayMatched: count(value.replayMatched),
+    matched: count(value.matched),
+    delivered: count(value.delivered),
+    clientReceived: nullable("clientReceived"),
+    decoded: nullable("decoded"),
+    applied: nullable("applied")
+  };
+}
+
+function nullableCount(value, fallback) {
+  const parsed = integerOrNull(value);
+  return parsed === null || parsed < 0 ? fallback : parsed;
+}
+
+function count(value) {
+  return nullableCount(value, 0);
+}
+
+function kafkaOffset(value) {
+  const text = textValue(value);
+  if (!text || !/^\d+$/u.test(text)) return null;
+  try { return BigInt(text).toString(); } catch { return null; }
+}
+
+function integerOrNull(value) {
+  if (value === null || value === undefined || (typeof value === "string" && value.trim() === "")) return null;
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : null;
+}
+
+function booleanOrNull(value) {
+  return value === true ? true : value === false ? false : null;
+}
+
+function objectValue(value) {
+  return value && typeof value === "object" && !Array.isArray(value) ? value : {};
+}
+
+function textValue(value) {
+  const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
+  return text.length > 0 ? text : null;
+}

internal/cloud/workbench-kafka-sse-debug.test.ts

@@ -88,6 +88,55 @@ test("isolated Workbench Kafka replay uses its YAML-owned topic, unique group, a
   }
 });
 
+test("isolated Workbench Kafka replay applies replayId and offset barriers with bounded counts", async () => {
+  const fakeKafka = createFakeKafkaFactory();
+  const env = {
+    HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
+    HWLAB_KAFKA_CLIENT_ID: "test-hwlab-cloud-api",
+    HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true",
+    HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.debug.v1",
+    HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-test-workbench-isolated-debug",
+    HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_LIMIT: "10",
+    HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_TIMEOUT_MS: "3000"
+  };
+  const server = createServer((request, response) => {
+    const url = new URL(request.url ?? "/", "http://127.0.0.1");
+    void handleWorkbenchKafkaSseDebugHttp(request, response, url, { env, kafkaFactory: fakeKafka.factory, accessController: fakeAccessController(), logger: null });
+  });
+  await listen(server);
+  try {
+    const query = "stream=hwlab-debug&traceId=trc_barrier&replayId=rpl_barrier&fromBeginning=true&partition=0&firstOffset=10&lastOffset=11";
+    const response = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?${query}`);
+    const reader = response.body?.getReader();
+    assert.ok(reader);
+    const connected = await readUntil(reader, "hwlab.kafka.connected");
+    assert.match(connected, /"replayId":"rpl_barrier"/u);
+    assert.match(connected, /"firstOffset":"10"/u);
+    assert.match(connected, /"replayMode":"correlated-v2"/u);
+    assert.match(connected, /"seekApplied":true/u);
+    assert.equal(fakeKafka.fromBeginning, false);
+    assert.deepEqual(fakeKafka.seekCalls, [{ topic: "hwlab.event.debug.v1", partition: 0, offset: "10" }]);
+
+    await fakeKafka.emit({ schema: "hwlab.event.debug.v1", traceId: "trc_barrier", replayId: "rpl_barrier", debugLineage: { inputTraceId: "trc_barrier", inputSessionId: "ses_agentrun_barrier", sourceSeq: 1 }, event: { type: "assistant" } }, { offset: "10" });
+    await fakeKafka.emit({ schema: "hwlab.event.debug.v1", traceId: "trc_barrier", replayId: "rpl_barrier", debugLineage: { inputTraceId: "trc_barrier", inputSessionId: "ses_agentrun_barrier", sourceSeq: 2 }, eventType: "terminal", event: { type: "result", terminal: true } }, { offset: "11" });
+    const replay = await readUntil(reader, "hwlab.kafka.replay-complete");
+    assert.match(replay, /"code":"terminal_complete"/u);
+    assert.match(replay, /"reason":"barrier"/u);
+    assert.match(replay, /"scanned":2/u);
+    assert.match(replay, /"barrierScanned":2/u);
+    assert.match(replay, /"parsed":2/u);
+    assert.match(replay, /"matched":2/u);
+    assert.match(replay, /"delivered":2/u);
+    assert.match(replay, /"firstOffset":"10","lastOffset":"11","count":2/u);
+    assert.match(replay, /"inputSessionId":"ses_agentrun_barrier"/u);
+    assert.match(replay, /"barrierComplete":true/u);
+    assert.match(replay, /"clientCountsAvailable":false/u);
+    await waitUntil(() => fakeKafka.stopCalls > 0);
+  } finally {
+    await close(server);
+  }
+});
+
 test("isolated Workbench Kafka capability rejects product topic and group identities", () => {
   const env = {
     HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_ENABLED: "true",
@@ -137,12 +186,42 @@ test("isolated Workbench Kafka debug fails closed when disabled and rejects repl
     assert.equal(replay.status, 400);
     assert.match(await replay.text(), /workbench_kafka_debug_replay_required/u);
     assert.equal(fakeKafka.consumerCreated, false);
+
+    const invalidRange = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&partition=0&firstOffset=20`);
+    assert.equal(invalidRange.status, 400);
+    assert.match(await invalidRange.text(), /workbench_kafka_debug_offset_range_invalid/u);
+    assert.equal(fakeKafka.consumerCreated, false);
+
+    const invalidReplayId = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&replayId=not-a-replay`);
+    assert.equal(invalidReplayId.status, 400);
+    assert.match(await invalidReplayId.text(), /workbench_kafka_debug_replay_id_invalid/u);
+    assert.equal(fakeKafka.consumerCreated, false);
+
+    const replayOnly = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&replayId=rpl_incomplete`);
+    assert.equal(replayOnly.status, 400);
+    assert.match(await replayOnly.text(), /workbench_kafka_debug_v2_correlation_incomplete/u);
+    assert.equal(fakeKafka.consumerCreated, false);
+
+    const rangeOnly = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&partition=0&firstOffset=10&lastOffset=11`);
+    assert.equal(rangeOnly.status, 400);
+    assert.match(await rangeOnly.text(), /workbench_kafka_debug_v2_correlation_incomplete/u);
+    assert.equal(fakeKafka.consumerCreated, false);
+
+    const producerNotInvoked = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&replayId=rpl_not_invoked&partition=0&firstOffset=10&lastOffset=11&producerInvoked=false`);
+    assert.equal(producerNotInvoked.status, 409);
+    assert.match(await producerNotInvoked.text(), /workbench_kafka_debug_producer_not_invoked/u);
+    assert.equal(fakeKafka.consumerCreated, false);
+
+    const cardinalityMismatch = await fetch(`${serverUrl(server)}/v1/workbench/debug/kafka-sse/events?stream=hwlab-debug&traceId=trc_replay&fromBeginning=true&replayId=rpl_bad_count&partition=0&firstOffset=10&lastOffset=11&publishedCount=3`);
+    assert.equal(cardinalityMismatch.status, 400);
+    assert.match(await cardinalityMismatch.text(), /workbench_kafka_debug_producer_cardinality_mismatch/u);
+    assert.equal(fakeKafka.consumerCreated, false);
   } finally {
     await close(server);
   }
 });
 
-test("isolated Workbench Kafka replay closes as incomplete when terminal is absent", async () => {
+test("isolated Workbench Kafka replay reports matched events without terminal", async () => {
   const fakeKafka = createFakeKafkaFactory();
   const env = {
     HWLAB_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092",
@@ -150,7 +229,7 @@ test("isolated Workbench Kafka replay closes as incomplete when terminal is abse
     HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC: "hwlab.event.debug.v1",
     HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX: "hwlab-test-workbench-isolated-debug",
     HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_LIMIT: "10",
-    HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_TIMEOUT_MS: "30"
+    HWLAB_WORKBENCH_KAFKA_DEBUG_REPLAY_TIMEOUT_MS: "100"
   };
   const server = createServer((request, response) => {
     const url = new URL(request.url ?? "/", "http://127.0.0.1");
@@ -162,9 +241,12 @@ test("isolated Workbench Kafka replay closes as incomplete when terminal is abse
     assert.equal(response.status, 200);
     const reader = response.body?.getReader();
     assert.ok(reader);
+    await readUntil(reader, "hwlab.kafka.connected");
+    await fakeKafka.emit({ schema: "hwlab.event.debug.v1", traceId: "trc_incomplete", eventType: "assistant", event: { traceId: "trc_incomplete", type: "assistant" } }, { offset: "5" });
     const replay = await readUntil(reader, "hwlab.kafka.replay-complete");
     assert.match(replay, /"reason":"timeout"/u);
-    assert.match(replay, /"count":0/u);
+    assert.match(replay, /"count":1/u);
+    assert.match(replay, /"code":"terminal_missing"/u);
     assert.match(replay, /"terminalObserved":false/u);
     await waitUntil(() => fakeKafka.stopCalls > 0);
   } finally {
@@ -352,6 +434,8 @@ function createFakeKafkaFactory(options: { deferRun?: boolean } = {}) {
   let stopCalls = 0;
   let disconnectCalls = 0;
   let consumerCreated = false;
+  let runStarted = false;
+  const seekCalls: Array<{ topic: string; partition: number; offset: string }> = [];
   const consumer = {
     connect: async () => undefined,
     subscribe: async (input: { topic?: string; fromBeginning?: boolean }) => {
@@ -360,8 +444,13 @@ function createFakeKafkaFactory(options: { deferRun?: boolean } = {}) {
     },
     run: async (input: any) => {
       if (runGate) await runGate;
+      runStarted = true;
       eachMessage = input.eachMessage;
     },
+    seek: async (input: { topic: string; partition: number; offset: string }) => {
+      assert.equal(runStarted, true, "seek must happen after consumer.run assignment");
+      seekCalls.push(input);
+    },
     stop: async () => { stopCalls += 1; },
     disconnect: async () => { disconnectCalls += 1; }
   };
@@ -379,14 +468,15 @@ function createFakeKafkaFactory(options: { deferRun?: boolean } = {}) {
     get stopCalls() { return stopCalls; },
     get disconnectCalls() { return disconnectCalls; },
     get consumerCreated() { return consumerCreated; },
+    get seekCalls() { return seekCalls; },
     releaseRun: () => releaseRun?.(),
-    emit: async (value: Record<string, unknown>) => {
+    emit: async (value: Record<string, unknown>, transport: { offset?: string; partition?: number } = {}) => {
       assert.ok(eachMessage, "consumer.run must be called before emitting fake Kafka events");
       await eachMessage({
         topic: subscribedTopic,
-        partition: 0,
+        partition: transport.partition ?? 0,
         message: {
-          offset: String(value.sessionId === "ses_other" ? 1 : 2),
+          offset: transport.offset ?? String(value.sessionId === "ses_other" ? 1 : 2),
           key: Buffer.from(String(value.sessionId ?? "unknown")),
           timestamp: "2026-07-09T18:30:00.000Z",
           value: Buffer.from(JSON.stringify(value))

internal/cloud/workbench-kafka-sse-debug.ts

@@ -7,6 +7,16 @@ import { openKafkaEventStream } from "./kafka-event-bridge.ts";
 import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
 import { authenticateWorkbenchRead } from "./server-workbench-read-http.ts";
 import { workbenchKafkaDebugCapability } from "./workbench-kafka-debug-capability.ts";
+import {
+  completeWorkbenchKafkaDebugReplay,
+  createWorkbenchKafkaDebugReplayTracker,
+  markWorkbenchKafkaDebugReplayBarrierComplete,
+  markWorkbenchKafkaDebugReplayConsumerReady,
+  normalizeWorkbenchKafkaDebugOffsetRange,
+  observeWorkbenchKafkaDebugReplayRecord,
+  recordWorkbenchKafkaDebugReplayDelivery,
+  workbenchKafkaDebugOffsetRangeWidth
+} from "./workbench-kafka-debug-replay-contract.ts";
 
 const CONTRACT_VERSION = "workbench-debug-kafka-sse-v1";
 const DEFAULT_STREAM = "hwlab";
@@ -48,7 +58,7 @@ export async function handleWorkbenchKafkaSseDebugHttp(request, response, url, o
 function describeKafkaSseDebug(url, env, isolatedDebug = null) {
   const stream = streamFromUrl(url);
   const requestedFilters = filtersFromUrl(url);
-  const resolvedFilters = stream === "hwlab-debug" ? compactObject({ traceId: requestedFilters.traceId }) : resolveKafkaDebugFilters(requestedFilters, { traceStore: defaultCodeAgentTraceStore });
+  const resolvedFilters = stream === "hwlab-debug" ? compactObject({ traceId: requestedFilters.traceId, replayId: requestedFilters.replayId }) : resolveKafkaDebugFilters(requestedFilters, { traceStore: defaultCodeAgentTraceStore });
   return {
     ok: true,
     contractVersion: CONTRACT_VERSION,
@@ -60,6 +70,7 @@ function describeKafkaSseDebug(url, env, isolatedDebug = null) {
     replayLimit: isolatedDebug?.replayLimit ?? null,
     replayTimeoutMs: isolatedDebug?.replayTimeoutMs ?? null,
     groupIdPrefix: isolatedDebug?.groupIdPrefix ?? null,
+    replayRequest: isolatedDebug?.replayRequest ?? null,
     filters: requestedFilters,
     resolvedFilters,
     eventsRoute: `/v1/workbench/debug/kafka-sse/events${url.search || ""}`,
@@ -72,7 +83,17 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
   const stream = streamFromUrl(url);
   const filters = filtersFromUrl(url);
   const isolatedReplay = stream === "hwlab-debug";
-  const resolvedFilters = isolatedReplay ? compactObject({ traceId: filters.traceId }) : resolveKafkaDebugFilters(filters, options);
+  const resolvedFilters = isolatedReplay ? compactObject({ traceId: filters.traceId, replayId: filters.replayId }) : resolveKafkaDebugFilters(filters, options);
+  const replayRequest = isolatedDebug?.replayRequest ?? replayRequestFromUrl(url);
+  const correlatedReplay = isolatedReplay && replayRequest.mode === "correlated-v2";
+  const replayTracker = isolatedReplay ? createWorkbenchKafkaDebugReplayTracker({
+    mode: replayRequest.mode,
+    replayId: resolvedFilters.replayId,
+    traceId: resolvedFilters.traceId,
+    topic: isolatedDebug.topic,
+    offsetRange: replayRequest.offsetRange,
+    producer: replayRequest.producer
+  }) : null;
   response.writeHead(200, {
     "content-type": "text/event-stream; charset=utf-8",
     "cache-control": "no-store, no-transform",
@@ -88,6 +109,7 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
   let replayTimer = null;
   let replayCount = 0;
   let replayTerminalObserved = false;
+  let barrierCompletionPending = false;
   const bufferedRecords = [];
   const close = () => {
     if (closed) return;
@@ -102,6 +124,10 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
     if (!isolatedReplay || replayFinished || closed) return;
     replayFinished = true;
     if (replayTimer) clearTimeout(replayTimer);
+    const replayResult = completeWorkbenchKafkaDebugReplay(replayTracker, {
+      reason,
+      terminalObserved: replayTerminalObserved
+    });
     writeSse(response, "hwlab.kafka.replay-complete", {
       ok: true,
       contractVersion: CONTRACT_VERSION,
@@ -109,18 +135,30 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       topic: isolatedDebug.topic,
       traceId: resolvedFilters.traceId,
       replayComplete: true,
+      replayMode: replayResult.mode,
+      replayId: replayResult.replayId,
+      phase: replayResult.phase,
+      code: replayResult.code,
+      result: replayResult,
       reason,
       count: replayCount,
+      counts: replayResult.counts,
+      rejectedByReason: replayResult.rejectedByReason,
+      offsetRange: replayResult.offsetRange,
+      sourceLineage: replayResult.sourceLineage,
       limit: isolatedDebug.replayLimit,
       timeoutMs: isolatedDebug.replayTimeoutMs,
       terminalObserved: replayTerminalObserved,
+      barrierComplete: replayResult.barrier.completed,
+      stopBoundary: "sse-end",
+      clientCountsAvailable: false,
       valuesPrinted: false
     });
     if (!response.writableEnded) response.end();
   };
   const writeRecord = (record) => {
     if (closed || replayFinished) return;
-    writeSse(response, "hwlab.kafka.event", {
+    const delivered = writeSse(response, "hwlab.kafka.event", {
       ok: true,
       contractVersion: CONTRACT_VERSION,
       stream,
@@ -134,10 +172,11 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       valuesPrinted: false
     });
     if (!isolatedReplay) return;
-    replayCount += 1;
-    replayTerminalObserved = debugRecordIsTerminal(record);
-    if (replayTerminalObserved) finishReplay("terminal");
-    else if (replayCount >= isolatedDebug.replayLimit) finishReplay("limit");
+    recordWorkbenchKafkaDebugReplayDelivery(replayTracker, record, delivered);
+    replayCount = replayTracker.counts.delivered;
+    replayTerminalObserved ||= debugRecordIsTerminal(record);
+    if (!correlatedReplay && replayTerminalObserved) finishReplay("terminal");
+    else if (!correlatedReplay && replayCount >= isolatedDebug.replayLimit) finishReplay("limit");
   };
   if (isolatedReplay) {
     replayTimer = setTimeout(() => finishReplay("timeout"), isolatedDebug.replayTimeoutMs);
@@ -149,9 +188,18 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       stream,
       topic: isolatedDebug?.topic ?? null,
       groupIdPrefix: isolatedDebug?.groupIdPrefix ?? null,
-      fromBeginning: isolatedReplay || url.searchParams.get("fromBeginning") === "1" || url.searchParams.get("fromBeginning") === "true",
+      fromBeginning: isolatedReplay ? !correlatedReplay : url.searchParams.get("fromBeginning") === "1" || url.searchParams.get("fromBeginning") === "true",
       ...resolvedFilters,
+      offsetRange: replayRequest.offsetRange,
       kafkaFactory: options.kafkaFactory,
+      onRecord: isolatedReplay ? async (observation) => {
+        observeWorkbenchKafkaDebugReplayRecord(replayTracker, observation);
+      } : null,
+      onBarrierComplete: correlatedReplay ? async () => {
+        markWorkbenchKafkaDebugReplayBarrierComplete(replayTracker);
+        if (consumerReady) finishReplay("barrier");
+        else barrierCompletionPending = true;
+      } : null,
       onEvent: async (record) => {
         if (isolatedReplay && !consumerReady) {
           if (bufferedRecords.length < isolatedDebug.replayLimit) bufferedRecords.push(record);
@@ -164,6 +212,7 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       await kafkaStream.stop?.();
       return;
     }
+    if (isolatedReplay) markWorkbenchKafkaDebugReplayConsumerReady(replayTracker, { groupId: kafkaStream.groupId, seekApplied: kafkaStream.seekApplied });
     writeSse(response, "hwlab.kafka.connected", {
       ok: true,
       contractVersion: CONTRACT_VERSION,
@@ -177,6 +226,11 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       replay: isolatedReplay,
       replayLimit: isolatedDebug?.replayLimit ?? null,
       replayTimeoutMs: isolatedDebug?.replayTimeoutMs ?? null,
+      replayId: replayTracker?.replayId ?? null,
+      replayMode: replayTracker?.mode ?? null,
+      offsetRange: replayTracker?.requestedOffsetRange ?? null,
+      replayContractVersion: replayTracker?.mode === "correlated-v2" ? "workbench-kafka-debug-replay-v2" : CONTRACT_VERSION,
+      seekApplied: kafkaStream.seekApplied === true,
       filters,
       resolvedFilters,
       serverSentAt: new Date().toISOString(),
@@ -187,10 +241,11 @@ async function openKafkaDebugSse(request, response, url, options, isolatedDebug
       if (replayFinished || closed) break;
       writeRecord(record);
     }
+    if (barrierCompletionPending && !replayFinished && !closed) finishReplay("barrier");
   } catch (error) {
     if (replayTimer) clearTimeout(replayTimer);
     writeSse(response, "hwlab.kafka.error", { ok: false, error: errorMessagePayload(error), valuesPrinted: false });
-    if (isolatedReplay && !response.writableEnded) response.end();
+    if (isolatedReplay) finishReplay("error");
   }
 }
 
@@ -254,7 +309,8 @@ function filtersFromUrl(url) {
     traceId: safeId(url.searchParams.get("traceId") || url.searchParams.get("trace-id")),
     sessionId: safeId(url.searchParams.get("sessionId") || url.searchParams.get("session-id")),
     runId: safeId(url.searchParams.get("runId") || url.searchParams.get("run-id")),
-    commandId: safeId(url.searchParams.get("commandId") || url.searchParams.get("command-id"))
+    commandId: safeId(url.searchParams.get("commandId") || url.searchParams.get("command-id")),
+    replayId: safeReplayId(url.searchParams.get("replayId") || url.searchParams.get("replay-id"))
   });
 }
 
@@ -285,7 +341,79 @@ function isolatedDebugConfig(response, url, env, options = {}) {
     sendJson(response, 400, debugError("workbench_kafka_debug_replay_required", "Isolated Workbench Kafka debug requires explicit fromBeginning=true."));
     return false;
   }
-  return config;
+  const rawReplayId = url.searchParams.get("replayId") || url.searchParams.get("replay-id");
+  if (textValue(rawReplayId) && !safeReplayId(rawReplayId)) {
+    sendJson(response, 400, debugError("workbench_kafka_debug_replay_id_invalid", "Replay correlation requires replayId to start with rpl_ and contain only safe identifier characters."));
+    return false;
+  }
+  const replayRequest = replayRequestFromUrl(url);
+  if (options.requireReplay && replayRequest.offsetRangeInvalid) {
+    sendJson(response, 400, debugError("workbench_kafka_debug_offset_range_invalid", "Replay offset barrier requires partition, firstOffset, and lastOffset with firstOffset <= lastOffset."));
+    return false;
+  }
+  if (options.requireReplay && replayRequest.mode === "correlated-v2") {
+    if (!replayRequest.replayId || !replayRequest.offsetRange) {
+      sendJson(response, 400, debugError("workbench_kafka_debug_v2_correlation_incomplete", "Correlated V2 replay requires replayId and partition/firstOffset/lastOffset."));
+      return false;
+    }
+    if (replayRequest.producerMetadataInvalid) {
+      sendJson(response, 400, debugError("workbench_kafka_debug_v2_producer_metadata_invalid", "Optional producerInvoked, sourceMatched, and publishedCount must use valid boolean/integer values."));
+      return false;
+    }
+    if (replayRequest.producer.invoked === false) {
+      sendJson(response, 409, debugError("workbench_kafka_debug_producer_not_invoked", "Correlated V2 replay cannot use producerInvoked=false."));
+      return false;
+    }
+    if (replayRequest.producer.sourceMatched === false) {
+      sendJson(response, 409, debugError("workbench_kafka_debug_source_trace_missing", "Correlated V2 replay cannot start because the producer did not match the requested source trace."));
+      return false;
+    }
+    const rangeWidth = workbenchKafkaDebugOffsetRangeWidth(replayRequest.offsetRange);
+    if (!rangeWidth || (replayRequest.producer.publishedCount !== null && replayRequest.producer.publishedCount !== rangeWidth)) {
+      sendJson(response, 400, debugError("workbench_kafka_debug_producer_cardinality_mismatch", "publishedCount must equal the requested offset range width."));
+      return false;
+    }
+    if (rangeWidth > config.replayLimit) {
+      sendJson(response, 400, debugError("workbench_kafka_debug_replay_limit_exceeded", "The correlated replay batch exceeds the YAML-owned replay limit."));
+      return false;
+    }
+  }
+  return { ...config, replayRequest };
+}
+
+function replayRequestFromUrl(url) {
+  const rawReplayId = url.searchParams.get("replayId") || url.searchParams.get("replay-id");
+  const rawProducerInvoked = url.searchParams.get("producerInvoked") || url.searchParams.get("producer-invoked");
+  const rawSourceMatched = url.searchParams.get("sourceMatched") || url.searchParams.get("source-matched");
+  const rawPublishedCount = url.searchParams.get("publishedCount") || url.searchParams.get("published-count");
+  const rawRange = {
+    partition: url.searchParams.get("partition"),
+    firstOffset: url.searchParams.get("firstOffset") || url.searchParams.get("first-offset") || url.searchParams.get("startOffset") || url.searchParams.get("start-offset"),
+    lastOffset: url.searchParams.get("lastOffset") || url.searchParams.get("last-offset") || url.searchParams.get("endOffset") || url.searchParams.get("end-offset")
+  };
+  const offsetRangeProvided = Object.values(rawRange).some((entry) => textValue(entry));
+  const offsetRange = normalizeWorkbenchKafkaDebugOffsetRange(rawRange);
+  const correlationProvided = Boolean(textValue(rawReplayId) || offsetRangeProvided || textValue(rawProducerInvoked) || textValue(rawSourceMatched) || textValue(rawPublishedCount));
+  const producerInvoked = optionalBoolean(rawProducerInvoked);
+  const sourceMatched = optionalBoolean(rawSourceMatched);
+  const publishedCount = optionalNonNegativeInteger(rawPublishedCount);
+  return {
+    mode: correlationProvided ? "correlated-v2" : "trace-only-v1",
+    replayId: safeReplayId(rawReplayId),
+    offsetRange,
+    offsetRangeInvalid: offsetRangeProvided && !offsetRange,
+    producerMetadataInvalid: Boolean(
+      (textValue(rawProducerInvoked) && producerInvoked === null)
+      || (textValue(rawSourceMatched) && sourceMatched === null)
+      || (textValue(rawPublishedCount) && publishedCount === null)
+    ),
+    producer: {
+      invoked: producerInvoked,
+      sourceMatched,
+      publishedCount
+    },
+    valuesPrinted: false
+  };
 }
 
 function methodNotAllowed(response, allow) {
@@ -313,6 +441,24 @@ function safeId(value) {
   return text && /^[A-Za-z0-9_.:-]{3,220}$/u.test(text) ? text : null;
 }
 
+function safeReplayId(value) {
+  const text = safeId(value);
+  return text && /^rpl_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
+}
+
+function optionalBoolean(value) {
+  const text = textValue(value)?.toLowerCase();
+  if (text === "true" || text === "1") return true;
+  if (text === "false" || text === "0") return false;
+  return null;
+}
+
+function optionalNonNegativeInteger(value) {
+  if (!textValue(value)) return null;
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : null;
+}
+
 function textValue(value) {
   const text = typeof value === "string" ? value.trim() : value === null || value === undefined ? "" : String(value).trim();
   return text.length > 0 ? text : null;

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

@@ -13,6 +13,13 @@ 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.match(result.payload.commands[0], /--input-topic agentrun\.event\.v1/u);
+  assert.match(result.payload.commands[1], /--input-topic agentrun\.event\.debug\.v1/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");
@@ -25,6 +32,7 @@ test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug even
     "--from", "jsonl",
     "--session-id", SESSION_ID,
     "--trace-id", TRACE_ID,
+    "--replay-id", "rpl_offline_35",
     "--jsonl-file", inputFile,
     "--output-jsonl", outputFile,
     "--expect-count", "35",
@@ -62,6 +70,9 @@ test("offline JSONL maps 35 stdio reconstructions to 35 ordered HWLAB debug even
   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));
@@ -86,7 +97,9 @@ test("Kafka reader and producer are injected while canonical AgentRun events use
     "--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"
@@ -121,7 +134,10 @@ test("Kafka reader and producer are injected while canonical AgentRun events use
   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));
@@ -132,6 +148,101 @@ test("Kafka reader and producer are injected while canonical AgentRun events use
   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)] }; },
+    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("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)] }; },
+    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 () => {
@@ -161,6 +272,13 @@ test("reconstruction identity fabrication and product output topic fail closed",
   ], { 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 () => {

tools/src/hwlab-cli/kafka-regenerate.ts

@@ -9,15 +9,17 @@ import { Kafka, logLevel } from "kafkajs";
 import {
   buildHwlabKafkaProducerMessage,
   DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC,
+  DEFAULT_AGENTRUN_EVENT_TOPIC,
   DEFAULT_HWLAB_DEBUG_EVENT_TOPIC,
   projectAgentRunKafkaMessageToHwlabDebugEvent,
   queryKafkaEventStream
 } from "../../../internal/cloud/kafka-event-bridge.ts";
 
 const CLI_NAME = "hwlab-cli";
-const VERSION = "0.3.0-kafka-debug";
+const VERSION = "0.3.1-kafka-debug-replay";
 const DEFAULT_LIMIT = 500;
 const DEFAULT_TIMEOUT_MS = 5000;
+const DEBUG_OUTPUT_PARTITION = 0;
 
 type JsonRecord = Record<string, any>;
 type EnvLike = Record<string, string | undefined>;
@@ -66,8 +68,11 @@ export async function runKafkaCli(argv: string[], options: KafkaCliOptions = {})
 export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencies: Required<KafkaCliOptions>) {
   const sessionId = requiredSessionId(parsed.sessionId);
   const traceId = optionalTraceId(parsed.traceId);
+  const replayId = optionalReplayId(parsed.replayId) || `rpl_${randomUUID().replaceAll("-", "")}`;
   const sourceMode = enumValue(parsed.from ?? "kafka", "from", ["kafka", "jsonl"]);
-  const inputTopicResolution = resolveConfig(parsed.inputTopic, undefined, DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, "--input-topic", "contract-default");
+  const inputTopicResolution = sourceMode === "kafka"
+    ? resolveConfig(parsed.inputTopic, dependencies.env.HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC, DEFAULT_AGENTRUN_EVENT_TOPIC, "--input-topic", "env:HWLAB_KAFKA_AGENTRUN_EVENT_TOPIC")
+    : resolveConfig(parsed.inputTopic, undefined, DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC, "--input-topic", "contract-default");
   const outputTopicResolution = resolveConfig(parsed.outputTopic, dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC, DEFAULT_HWLAB_DEBUG_EVENT_TOPIC, "--output-topic", "env:HWLAB_KAFKA_HWLAB_DEBUG_EVENT_TOPIC");
   const inputTopic = inputTopicResolution.value;
   const outputTopic = outputTopicResolution.value;
@@ -76,7 +81,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
   const timeoutMs = boundedInteger(parsed.timeoutMs, DEFAULT_TIMEOUT_MS, 250, 60000, "timeoutMs");
   const expectedCount = optionalPositiveInteger(parsed.expectCount, "expectCount");
   const readLimit = expectedCount === undefined ? limit : Math.min(5000, expectedCount + 1);
-  const shouldPublish = parsed.noPublish === true ? false : parsed.publish === true ? true : sourceMode === "kafka";
+  const shouldPublish = parsed.publish === true && parsed.noPublish !== true;
   const groupPrefixResolution = resolveKafkaGroupConfig(
     parsed.groupPrefix,
     dependencies.env.HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX,
@@ -103,7 +108,8 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
     sessionId,
     traceId,
     expectedCount,
-    producedAt: dependencies.now()
+    producedAt: dependencies.now(),
+    replayId
   });
 
   const stateId = `hwlab-debug-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`;
@@ -115,7 +121,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
     const message = buildHwlabKafkaProducerMessage(envelope);
     return {
       topic: outputTopic,
-      partition: null,
+      partition: DEBUG_OUTPUT_PARTITION,
       offset: null,
       key: message.key,
       value: envelope,
@@ -134,7 +140,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
       await producer.connect?.();
       publishMetadata = await producer.send({
         topic: outputTopic,
-        messages: mapped.events.map(buildHwlabKafkaProducerMessage)
+        messages: mapped.events.map((event) => ({ ...buildHwlabKafkaProducerMessage(event), partition: DEBUG_OUTPUT_PARTITION }))
       });
       publishedCount = mapped.events.length;
     } catch (error) {
@@ -152,11 +158,22 @@ 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) {
+    throw cliError("debug_publish_barrier_missing", "Kafka publish succeeded without a usable isolated replay offset barrier", {
+      replayId,
+      topic: outputTopic,
+      partition: DEBUG_OUTPUT_PARTITION,
+      publishedCount,
+      metadataCount: Array.isArray(publishMetadata) ? publishMetadata.length : 0
+    });
+  }
   const sourceFile = text(readResult.sourceFile);
   const traceArgument = traceId ? ` --trace-id ${traceId}` : "";
+  const groupArgument = groupPrefix ? ` --group-prefix ${groupPrefix}` : "";
   const baseCommand = sourceMode === "jsonl"
-    ? `hwlab-cli kafka regenerate hwlab --from jsonl --session-id ${sessionId}${traceArgument} --jsonl-file ${shellArg(sourceFile || String(parsed.jsonlFile))}`
-    : `hwlab-cli kafka regenerate hwlab --from kafka --session-id ${sessionId}${traceArgument} --input-topic ${inputTopic}`;
+    ? `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
     ? `${baseCommand} --no-publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json`
     : `${baseCommand} --publish --output-topic ${outputTopic} --expect-count ${mapped.events.length} --json`;
@@ -164,6 +181,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
     ok: true,
     action: "kafka.regenerate.hwlab",
     status: "succeeded",
+    replayId,
     sourceMode,
     config: {
       inputTopic: inputTopicResolution,
@@ -185,6 +203,7 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
       published: shouldPublish,
       publishedCount,
       metadataCount: Array.isArray(publishMetadata) ? publishMetadata.length : 0,
+      offsetRanges,
       artifact: {
         path: artifactPath,
         bytes: Buffer.byteLength(artifactText),
@@ -200,6 +219,27 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
       lastSourceSeq: last?.debugLineage?.sourceSeq ?? null,
       inputKinds: [...new Set(mapped.inputKinds)]
     },
+    replay: {
+      replayId,
+      producerInvoked: shouldPublish,
+      sourceMatched: true,
+      sourceLineage: {
+        traceId: first?.debugLineage?.inputTraceId ?? traceId ?? null,
+        sessionId: first?.debugLineage?.inputSessionId ?? sessionId,
+        hwlabSessionId: first?.debugLineage?.inputHwlabSessionId ?? null,
+        runId: first?.debugLineage?.inputRunId ?? null,
+        commandId: first?.debugLineage?.inputCommandId ?? null,
+        inputTopic,
+        sourcePartition: first?.sourceEvent?.partition ?? null,
+        firstSourceOffset: first?.sourceEvent?.offset ?? null,
+        lastSourceOffset: last?.sourceEvent?.offset ?? null,
+        firstSourceSeq: first?.debugLineage?.sourceSeq ?? null,
+        lastSourceSeq: last?.debugLineage?.sourceSeq ?? null,
+        valuesPrinted: false
+      },
+      outputBarrier: offsetRanges.length === 1 ? offsetRanges[0] : null,
+      valuesPrinted: false
+    },
     runtimeDependencies: {
       cloudApi: false,
       database: false,
@@ -216,10 +256,10 @@ export async function regenerateHwlabDebugEvents(parsed: ParsedArgs, dependencie
   };
 }
 
-export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], options: { sessionId: string; traceId?: string; expectedCount?: number; producedAt?: string }) {
+export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], options: { sessionId: string; traceId?: string; expectedCount?: number; producedAt?: string; replayId?: string }) {
   const matchingRecords = records.filter((record) => recordMatches(record, options.sessionId, options.traceId));
   if (matchingRecords.length === 0) {
-    throw cliError("agentrun_debug_events_missing", "no AgentRun events matched the requested sessionId/traceId", {
+    throw cliError("source_trace_missing", "no AgentRun events matched the requested sessionId/traceId", {
       sessionId: options.sessionId,
       traceId: options.traceId || null,
       readCount: records.length
@@ -237,7 +277,12 @@ export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], opti
   let lineagePreserved = true;
   for (const [index, record] of matchingRecords.entries()) {
     const kafkaMessage = kafkaMessageFromRecord(record, index);
-    const projected = projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage, { producedAt: options.producedAt });
+    const projected = projectAgentRunKafkaMessageToHwlabDebugEvent(kafkaMessage, {
+      producedAt: options.producedAt,
+      replayId: options.replayId,
+      replayIndex: index + 1,
+      replayCount: matchingRecords.length
+    });
     if (!projected.ok) {
       throw cliError(projected.error?.code || "hwlab_debug_projection_failed", projected.error?.message || "AgentRun debug projection failed", {
         index,
@@ -251,7 +296,8 @@ export function mapAgentRunRecordsToHwlabDebugEvents(records: JsonRecord[], opti
     orderPreserved &&= projected.envelope.debugLineage?.sourceSeq === sourceSeq;
     lineagePreserved &&= projected.envelope.debugLineage?.inputTraceId === text(input.traceId ?? input.event?.payload?.traceId)
       && sessionCandidates(input).includes(options.sessionId)
-      && projected.envelope.sourceEvent?.sha256 === kafkaMessage.inputSha256;
+      && projected.envelope.sourceEvent?.sha256 === kafkaMessage.inputSha256
+      && (!options.replayId || projected.envelope.replayId === options.replayId);
     events.push(projected.envelope);
     inputKinds.push(projected.inputKind);
   }
@@ -272,19 +318,22 @@ function kafkaHelp() {
     action: "kafka.help",
     status: "succeeded",
     commands: [
-      "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... [--trace-id trc_...] [--input-topic agentrun.event.debug.v1] [--expect-count 35] [--json]",
-      "hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... --jsonl-file agentrun-events.jsonl --no-publish [--output-jsonl hwlab-events.jsonl] [--json]"
+      "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... [--trace-id trc_...] [--replay-id rpl_...] [--input-topic agentrun.event.v1] [--group-prefix hwlab-...-debug] [--expect-count 35] [--json]",
+      "hwlab-cli kafka regenerate hwlab --from kafka --session-id ses_... --input-topic agentrun.event.debug.v1 --group-prefix hwlab-...-debug [--replay-id rpl_...] [--json]",
+      "hwlab-cli kafka regenerate hwlab --from jsonl --session-id ses_... [--replay-id rpl_...] --jsonl-file agentrun-events.jsonl --no-publish [--output-jsonl hwlab-events.jsonl] [--json]"
     ],
     defaults: {
-      inputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC,
+      kafkaInputTopic: DEFAULT_AGENTRUN_EVENT_TOPIC,
+      jsonlInputTopic: DEFAULT_AGENTRUN_DEBUG_EVENT_TOPIC,
       outputTopic: DEFAULT_HWLAB_DEBUG_EVENT_TOPIC,
       groupPrefix: null,
       groupPrefixAuthority: "--group-prefix or env:HWLAB_KAFKA_HWLAB_DEBUG_GROUP_PREFIX when Kafka is used",
-      kafkaInputPublishes: true,
+      kafkaInputPublishes: false,
       jsonlInputPublishes: false
     },
     boundaries: {
       productTopicAllowed: false,
+      explicitPublishRequired: true,
       cloudApi: false,
       database: false,
       transactionalProjector: false,
@@ -462,6 +511,33 @@ function optionalTraceId(value: unknown) {
   return result;
 }
 
+function optionalReplayId(value: unknown) {
+  const result = text(value);
+  if (result && !/^rpl_[A-Za-z0-9_.:-]+$/u.test(result)) throw cliError("invalid_replay_id", "replayId must start with rpl_", { replayId: result });
+  return result;
+}
+
+function publishOffsetRanges(metadata: unknown, publishedCount: number, partition: number) {
+  if (!Array.isArray(metadata) || publishedCount <= 0) return [];
+  const item = metadata.find((entry: any) => Number(entry?.partition) === partition);
+  const firstOffset = kafkaOffset(item?.baseOffset);
+  if (!item || firstOffset === null) return [];
+  return [{
+    topic: text(item.topicName) || null,
+    partition,
+    firstOffset,
+    lastOffset: (BigInt(firstOffset) + BigInt(publishedCount - 1)).toString(),
+    count: publishedCount,
+    valuesPrinted: false
+  }];
+}
+
+function kafkaOffset(value: unknown) {
+  const result = text(value);
+  if (!result || !/^\d+$/u.test(result)) return null;
+  try { return BigInt(result).toString(); } catch { return null; }
+}
+
 function enumValue(value: unknown, field: string, allowed: string[]) {
   const result = text(value);
   if (!allowed.includes(result)) throw cliError("invalid_option", `${field} must be one of: ${allowed.join(", ")}`, { field, value: result });