feat: add in-cluster Kafka replay diagnostics

.agents/skills/unidesk-kafka/SKILL.md

@@ -38,6 +38,8 @@ description: UniDesk Kafka 调试与事件链校对技能，覆盖 platform-infr
 
 ## 受控 CLI
 
+本技能遵循 `Skill(cli-spec)`。
+
 平台状态与只读元数据：
 
 ```bash
@@ -50,15 +52,28 @@ bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1
 AgentRun 生产 reducer 重建：
 
 ```bash
-./scripts/agentrun kafka regenerate agentrun \
+bun scripts/cli.ts platform-infra kafka replay agentrun \
+  --node NC01 \
+  --lane nc01-v02 \
   --session-id <ses_agentrun_...> \
   --trace-id <trc_...> \
-  --from kafka \
   --no-publish \
-  --format compact \
-  --json
+  -o json
 ```
 
+- 该入口自动进入 YAML 选中的 AgentRun manager Pod：
+  - 使用运行中 commit 自带的正式 `./scripts/agentrun kafka regenerate agentrun`；
+  - 复用生产 reducer；
+  - 同时读取源 topic 的 partition earliest/end offset；
+  - 不要求 host 解析 cluster DNS；
+  - 不创建 port-forward。
+- 默认输出紧凑文本；机器读取使用 `-o json|yaml`。
+- scan limit、应用读取 timeout 和短连接 grace 由 `config/platform-infra/kafka.yaml#management.replay` 控制。
+- 当前只接受 `--no-publish` 或同义 `--dry-run`：
+  - `mutation=false`；
+  - `topicAppended=false`；
+  - `valuesPrinted=false`。
+
 HWLAB 生产 mapper 重建：
 
 ```bash
@@ -80,7 +95,10 @@ bun scripts/cli.ts web-probe observe status <observerId> \
 ## 网络平面
 
 - Kafka cluster DNS 只在目标 k3s 网络内解析；host workspace 报 `getaddrinfo ENOTFOUND` 只说明网络平面错误，不是 topic 或事件缺失。
-- 优先在包含正式应用 CLI 的目标 Pod/Job 内运行 regenerate。
+- 优先使用 `platform-infra kafka replay agentrun`：
+  - CLI 从 `config/platform-infra/kafka.yaml` 选择 Kafka target；
+  - CLI 从 `config/agentrun.yaml` 选择同 node 的 AgentRun lane；
+  - CLI 在 manager 应用 Pod 内运行正式 replay。
 - 运行镜像未携带应用 CLI 时，允许 operator 侧通过 `trans <node>:k3s kubectl ... port-forward` 建立有界临时诊断通道，再从同一 node 的 source workspace 使用显式 `--brokers 127.0.0.1:<port>` 调用正式 CLI：
   - 先确认端口只监听 loopback；
   - 只用于当前只读或 debug-topic 调试；
@@ -104,6 +122,14 @@ bun scripts/cli.ts web-probe observe status <observerId> \
 
 ## 判定规则
 
+- `classification` 必须使用 typed reason：
+  - `matched`：生产 reducer 已生成调试事件；
+  - `producer-not-written`：源 topic 从未有记录；
+  - `retention-offset-empty-window`：partition 有历史 offset，但当前保留窗口为空；
+  - `filter-mismatch`：读取到记录，但 session/trace/direction/method 不匹配；
+  - `decode-schema-rejection`：记录存在，但 JSON 或 schema 无法解析；
+  - `network-or-runtime-error`：集群网络面、broker Pod、应用 workload 或命令执行失败。
+
 - stdio 有完整 tool/diff，但 regenerate 未产出对应 AgentRun event：归因 reducer/mapping。
 - regenerate 能产出，但 durable `agentrun.event.v1` 缺失：归因 producer/durable append。
 - AgentRun 产品 event 存在，而 HWLAB debug mapper不产出：归因 HWLAB mapper。

config/platform-infra/kafka.yaml

@@ -176,6 +176,12 @@ validation:
   smokeTopic: platform-infra.kafka.smoke.v1
 
 management:
+  replay:
+    defaultLimit: 5000
+    maxLimit: 20000
+    defaultTimeoutMs: 10000
+    maxTimeoutMs: 15000
+    commandGraceMs: 35000
   defaultTailLimit: 10
   maxTailLimit: 50
   defaultGroupListLimit: 25

docs/reference/platform-infra.md

@@ -61,18 +61,46 @@
 - Remote k3s targets that share the already-initialized PK01 Sub2API database should use YAML-controlled `target.runtime.autoSetup=false` and a rendered runtime config initContainer instead of rerunning Sub2API auto setup. The app's migration check path is latency-sensitive because it checks many migration rows one by one; repeated remote auto setup against PK01 can time out without indicating DB corruption. Required PostgreSQL `pg_hba` entries and host firewall forward allowances must be declared in YAML and applied through the controlled platform DB/server paths rather than by changing DB passwords, connection strings, or data.
 - Sub2API account sentinel, public exposure, and HTTPS egress proxy are target-scoped YAML decisions. The active target may run them when YAML enables them; the standby G14 target must stay deployed but inactive until YAML promotion. `sentinel.enabledOnTargets` is the authority for where Codex-pool sentinel image, CronJob, Secret and state resources are expected; disabled targets should report sentinel validation as skipped instead of failing on missing runtime sentinel objects. Do not create a second sentinel, FRP client, public management surface, or edge proxy by hand; enable or move those resources only through the target YAML and the `platform-infra sub2api` / `codex-pool --target` CLI paths.
 
-## Kafka Event Bus Boundary
+## Kafka 事件总线边界
 
-- Kafka for the HWLAB v0.3 / AgentRun v0.2 event-bus POC is a target-scoped UniDesk-operated platform service in namespace `platform-infra`; NC01 is the current validation target for pikasTech/HWLAB#2449. It is not owned by `hwlab-v03`, `agentrun-v02`, a per-lane Kafka namespace, or a service repository deployment file.
-- The canonical source of truth is `config/platform-infra/kafka.yaml`; target, namespace, Strimzi release URL, cluster name, storage class/size, topic list, client declarations, DLQ names, runtime switch and validation smoke topic must stay in that YAML. Current version numbers and retention values belong only in YAML, not in this reference.
-- The canonical entrypoint is `bun scripts/cli.ts platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce --target <node>`; `--node <node>` is an equivalent selector for node-targeted operations. Consumer-group retirement uses `platform-infra kafka groups cleanup --node <node> --policy <id> [--confirm]`. Formal mutation must use that path; raw `kubectl` is bounded diagnosis only.
-- `platform-infra kafka groups` renders a YAML-bounded real group table plus total/shown/omitted counts by default; `--limit` may raise the view only to the YAML maximum, while `--full` and `--raw` are explicit drill-down paths.
-- Consumer-group cleanup is plan-only unless `--confirm` is present and accepts only a policy declared under `config/platform-infra/kafka.yaml#management.consumerGroupCleanup.policies`. A group must match that policy's exact prefix and full name pattern, decode a group-name timestamp old enough for the YAML minimum, and be in a YAML-allowed inactive Kafka state. The candidate limit is also the per-command mutation ceiling; `--full` and `--raw` never expand it. Confirmed execution re-reads broker state after deletion and reports deleted, state-changed and failed groups separately.
-- HWLAB v0.3 and AgentRun v0.2 are client namespaces. Runtime readiness alone does not prove Workbench projection, SSE or AgentRun command ingestion has migrated. App producer/consumer switchover is owned by the corresponding service repo and must be verified from that service's original entrypoint, not by Kafka topic readiness alone.
-- The event-bus authority split is three topics: `codex-stdio.raw.v1` stores AgentRun runner's raw Codex stdio frames; `agentrun.event.v1` stores AgentRun manager durable event mirrors before HWLAB conversion; `hwlab.event.v1` stores HWLAB event stream after HWLAB-side conversion and is the stream that SSE should proxy to Workbench in the event-bus design. Do not mix these schemas into a single topic or let a downstream consumer rewrite the upstream source topic.
-- Shadow produce may write Kafka events for observation only when YAML keeps consumer cutover disabled; it must not replace the current business read path until the corresponding service has an explicit consumer switchover decision and replay/refresh semantics.
-- The first target-scoped POC is a single-node KRaft broker for observability, ordering and replay investigation. It improves auditability and smoke coverage, but it is not a production high-availability claim; replication, backup, min ISR and app-side transactional inbox/outbox are separate decisions.
-- Kafka must stay ClusterIP-only by default. Do not add Ingress, NodePort, LoadBalancer, host networking, public FRP, or browser-facing Kafka access unless a later YAML-controlled platform decision explicitly changes that boundary.
+- HWLAB v0.3 / AgentRun v0.2 事件总线 POC 的 Kafka：
+  - 是 UniDesk 运维的 target-scoped 平台服务；
+  - 位于 `platform-infra` namespace；
+  - 当前验证 target 为 pikasTech/HWLAB#2449 指定的 NC01；
+  - 不属于 `hwlab-v03`、`agentrun-v02`、lane 独立 Kafka namespace 或服务仓库部署文件。
+- 权威配置为 `config/platform-infra/kafka.yaml`：
+  - target、namespace、Strimzi release URL、cluster name、storage、topic、client、DLQ、runtime switch 和 smoke topic 必须保留在该 YAML；
+  - 当前版本号和 retention 值只写 YAML，不写入本参考文档。
+- 平台受控入口为 `bun scripts/cli.ts platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|replay|produce --target <node>`：
+  - node-targeted 操作可等价使用 `--node <node>`；
+  - consumer group 退役使用 `platform-infra kafka groups cleanup --node <node> --policy <id> [--confirm]`；
+  - 正式 mutation 必须使用受控入口；
+  - 原生 `kubectl` 只用于有界诊断。
+- 应用侧 Kafka replay 的受控入口：
+  - 命令为 `bun scripts/cli.ts platform-infra kafka replay agentrun --node <node> --lane <lane> --session-id <id> [--trace-id <id>] --no-publish`；
+  - Kafka target 来自 `config/platform-infra/kafka.yaml`；
+  - AgentRun workload 来自 `config/agentrun.yaml`；
+  - 命令进入应用 Pod 调用运行中 commit 自带的正式 reducer；
+  - 命令同时披露 partition offset window、scanned、accepted、typed rejection、produced 和首个 mismatch identity；
+  - 命令固定 `mutation=false`、`topicAppended=false`、`valuesPrinted=false`；
+  - scan limit、应用读取 timeout 和短连接 grace 由 `config/platform-infra/kafka.yaml#management.replay` 控制；
+  - 机器输出使用 `-o json|yaml`。
+- `platform-infra kafka groups` 默认输出受 YAML 上限约束的真实 group 表及 total/shown/omitted 计数；`--limit` 不能超过 YAML 上限，`--full` 和 `--raw` 是显式下钻入口。
+- consumer group cleanup：
+  - 未提供 `--confirm` 时只生成计划；
+  - 只接受 `config/platform-infra/kafka.yaml#management.consumerGroupCleanup.policies` 声明的 policy；
+  - group 必须匹配 prefix、完整名称、YAML 最小年龄和允许的 inactive state；
+  - candidate limit 同时是单次 mutation 上限，`--full` 和 `--raw` 不扩大该上限；
+  - confirmed 执行在删除后重新读取 broker 状态，分别报告 deleted、state-changed 和 failed。
+- HWLAB v0.3 和 AgentRun v0.2 是 client namespace；runtime ready 不能证明 Workbench projection、SSE 或 AgentRun command ingestion 已迁移，producer/consumer 切换必须由对应服务仓库拥有并从原入口验证。
+- 事件总线按三个 topic 划分权威：
+  - `codex-stdio.raw.v1` 保存 AgentRun runner 的 Codex stdio 原始 frame；
+  - `agentrun.event.v1` 保存 HWLAB 转换前的 AgentRun manager durable event mirror；
+  - `hwlab.event.v1` 保存 HWLAB 转换后的事件流，也是 SSE 向 Workbench 代理的流；
+  - 禁止混合 schema 或由下游 consumer 回写上游 source topic。
+- 只有 YAML 保持 consumer cutover disabled 时，shadow produce 才可写观察事件；在服务仓库明确完成 consumer 切换与 replay/refresh 语义前，不得替代当前业务读取路径。
+- 首个 target-scoped POC 是用于可观测性、顺序和 replay 调查的单节点 KRaft broker；它提高审计与 smoke 覆盖，但不声明生产高可用，replication、backup、min ISR 和应用 transactional inbox/outbox 属于独立决策。
+- Kafka 默认必须保持 ClusterIP-only；除非后续 YAML 控制的平台决策明确改变边界，禁止添加 Ingress、NodePort、LoadBalancer、host networking、public FRP 或浏览器直连 Kafka。
 
 ## LangBot Deployment Boundary
 

scripts/native/platform-infra/kafka-replay-diagnostics.py

@@ -0,0 +1,165 @@
+import json
+import os
+import subprocess
+import sys
+import time
+
+
+def run(argv, timeout):
+    started = time.monotonic()
+    try:
+        completed = subprocess.run(argv, capture_output=True, text=True, timeout=timeout, check=False)
+        return {
+            "exitCode": completed.returncode,
+            "stdout": completed.stdout,
+            "stderr": completed.stderr,
+            "elapsedMs": round((time.monotonic() - started) * 1000),
+            "timedOut": False,
+        }
+    except subprocess.TimeoutExpired as error:
+        stdout = error.stdout.decode("utf-8", errors="replace") if isinstance(error.stdout, bytes) else (error.stdout or "")
+        stderr = error.stderr.decode("utf-8", errors="replace") if isinstance(error.stderr, bytes) else (error.stderr or "")
+        return {
+            "exitCode": 124,
+            "stdout": stdout,
+            "stderr": stderr,
+            "elapsedMs": round((time.monotonic() - started) * 1000),
+            "timedOut": True,
+        }
+
+
+def json_object(text):
+    lines = [line.strip() for line in text.splitlines() if line.strip().startswith("{")]
+    for line in reversed(lines):
+        try:
+            value = json.loads(line)
+            if isinstance(value, dict):
+                return value
+        except json.JSONDecodeError:
+            pass
+    return None
+
+
+def offsets(namespace, cluster, bootstrap, topic, timeout):
+    pod_result = run([
+        "kubectl", "-n", namespace, "get", "pod",
+        "-l", "strimzi.io/cluster=" + cluster + ",strimzi.io/kind=Kafka",
+        "-o", "jsonpath={.items[0].metadata.name}",
+    ], timeout)
+    pod = pod_result["stdout"].strip()
+    if pod_result["exitCode"] != 0 or not pod:
+        return {"ok": False, "partitions": [], "errorClass": "broker-pod-unavailable"}
+    result = run([
+        "kubectl", "-n", namespace, "exec", pod, "--",
+        "bin/kafka-get-offsets.sh", "--bootstrap-server", bootstrap,
+        "--topic", topic, "--time", "-2",
+    ], timeout)
+    end_result = run([
+        "kubectl", "-n", namespace, "exec", pod, "--",
+        "bin/kafka-get-offsets.sh", "--bootstrap-server", bootstrap,
+        "--topic", topic, "--time", "-1",
+    ], timeout)
+    earliest = {}
+    end = {}
+    for output, target in ((result["stdout"], earliest), (end_result["stdout"], end)):
+        for line in output.splitlines():
+            parts = line.rsplit(":", 2)
+            if len(parts) == 3 and parts[1].isdigit() and parts[2].isdigit():
+                target[int(parts[1])] = int(parts[2])
+    partitions = [
+        {"partition": partition, "earliestOffset": earliest.get(partition), "endOffset": end.get(partition), "retainedRecords": max(0, end.get(partition, 0) - earliest.get(partition, 0))}
+        for partition in sorted(set(earliest) | set(end))
+    ]
+    return {
+        "ok": result["exitCode"] == 0 and end_result["exitCode"] == 0,
+        "partitions": partitions,
+        "errorClass": None if result["exitCode"] == 0 and end_result["exitCode"] == 0 else "offset-query-failed",
+    }
+
+
+def classify(replay, window, execution):
+    data = replay.get("data") if isinstance(replay, dict) and isinstance(replay.get("data"), dict) else {}
+    source = data.get("source") if isinstance(data.get("source"), dict) else {}
+    output = data.get("output") if isinstance(data.get("output"), dict) else {}
+    retained = sum(item.get("retainedRecords", 0) for item in window.get("partitions", []))
+    scanned = source.get("scanned", 0)
+    selected = data.get("selectedFrameCount", 0)
+    invalid = source.get("invalidJson", 0)
+    ignored = data.get("ignoredFrames") if isinstance(data.get("ignoredFrames"), dict) else {}
+    if execution.get("timedOut") or window.get("errorClass") or execution.get("exitCode") in (126, 127):
+        return "network-or-runtime-error"
+    if isinstance(replay, dict) and replay.get("ok") is True and output.get("recordCount", 0) > 0:
+        return "matched"
+    if retained == 0 and sum(item.get("endOffset", 0) for item in window.get("partitions", [])) == 0:
+        return "producer-not-written"
+    if retained == 0:
+        return "retention-offset-empty-window"
+    error = replay.get("error") if isinstance(replay.get("error"), dict) else {}
+    details = error.get("details") if isinstance(error.get("details"), dict) else {}
+    if invalid and not selected:
+        return "decode-schema-rejection"
+    if scanned or ignored or details.get("inputRecordCount") == 0:
+        return "filter-mismatch"
+    return "replay-rejected"
+
+
+namespace = os.environ["KAFKA_NAMESPACE"]
+cluster = os.environ["KAFKA_CLUSTER"]
+bootstrap = os.environ["KAFKA_BOOTSTRAP"]
+topic = os.environ["KAFKA_INPUT_TOPIC"]
+manager_namespace = os.environ["AGENTRUN_NAMESPACE"]
+manager_deployment = os.environ["AGENTRUN_MANAGER_DEPLOYMENT"]
+timeout_ms = int(os.environ["KAFKA_REPLAY_TIMEOUT_MS"])
+command_grace_ms = int(os.environ["KAFKA_REPLAY_COMMAND_GRACE_MS"])
+timeout_seconds = max(1, int((timeout_ms + command_grace_ms) / 1000))
+window = offsets(namespace, cluster, bootstrap, topic, timeout_seconds)
+command = [
+    "kubectl", "-n", manager_namespace, "exec", "deployment/" + manager_deployment, "--",
+    "sh", "-c",
+    "cd \"${AGENTRUN_APP_ROOT:-/workspace/agentrun}\" && exec ./scripts/agentrun \"$@\"",
+    "agentrun", "kafka", "regenerate", "agentrun",
+    "--session-id", os.environ["KAFKA_REPLAY_SESSION_ID"],
+    "--input-topic", topic,
+    "--limit", os.environ["KAFKA_REPLAY_LIMIT"],
+    "--timeout-ms", str(timeout_ms),
+    "--no-publish", "--format", "json",
+]
+trace_id = os.environ.get("KAFKA_REPLAY_TRACE_ID", "")
+if trace_id:
+    command.extend(["--trace-id", trace_id])
+execution = run(command, timeout_seconds)
+replay = json_object(execution["stdout"]) or json_object(execution["stderr"]) or {}
+reason = classify(replay, window, execution)
+data = replay.get("data") if isinstance(replay.get("data"), dict) else {}
+source = data.get("source") if isinstance(data.get("source"), dict) else {}
+output = data.get("output") if isinstance(data.get("output"), dict) else {}
+payload = {
+    "ok": reason == "matched",
+    "networkPlane": "k3s-application-pod",
+    "application": "agentrun",
+    "namespace": manager_namespace,
+    "workload": "deployment/" + manager_deployment,
+    "topic": topic,
+    "partitionOffsetWindow": window,
+    "scanned": source.get("scanned", 0),
+    "accepted": data.get("selectedFrameCount", 0),
+    "rejected": {
+        "invalidJson": source.get("invalidJson", 0),
+        "byReason": data.get("ignoredFrames", {}),
+        "classification": reason,
+    },
+    "produced": output.get("recordCount", 0),
+    "firstMismatchIdentity": None if reason == "matched" else {"sessionId": os.environ["KAFKA_REPLAY_SESSION_ID"], "traceId": trace_id or None, "reason": reason},
+    "mutation": False,
+    "topicAppended": False,
+    "valuesPrinted": False,
+    "execution": {
+        "exitCode": execution["exitCode"],
+        "elapsedMs": execution["elapsedMs"],
+        "timedOut": execution["timedOut"],
+        "stderrTail": execution["stderr"][-800:],
+    },
+    "replay": data,
+}
+print(json.dumps(payload, ensure_ascii=False))
+sys.exit(0)

scripts/src/platform-infra-kafka-replay.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, test } from "bun:test";
+import { parseKafkaReplayOptions, renderKafkaReplay } from "./platform-infra-kafka";
+import { renderMachine } from "./agentrun/render";
+
+describe("platform-infra Kafka replay", () => {
+  test("accepts the bounded no-publish application replay contract", () => {
+    expect(parseKafkaReplayOptions([
+      "agentrun", "--node", "NC01", "--lane", "nc01-v02",
+      "--session-id", "ses_example", "--trace-id", "trc_example",
+      "--no-publish", "-o", "yaml",
+    ])).toMatchObject({
+      targetId: "NC01",
+      lane: "nc01-v02",
+      sessionId: "ses_example",
+      traceId: "trc_example",
+      output: "yaml",
+    });
+  });
+
+  test("rejects replay without the explicit no-publish boundary", () => {
+    expect(() => parseKafkaReplayOptions(["agentrun", "--session-id", "ses_example"])).toThrow("requires --no-publish");
+  });
+
+  test("renders matched replay counts and mutation disclosure", () => {
+    const rendered = renderKafkaReplay({
+      ok: true,
+      diagnosis: {
+        networkPlane: "k3s-application-pod",
+        application: "agentrun",
+        namespace: "agentrun-v02",
+        workload: "deployment/agentrun-mgr",
+        topic: "codex-stdio.raw.v1",
+        scanned: 30,
+        accepted: 12,
+        produced: 8,
+        rejected: { classification: "matched" },
+        mutation: false,
+        topicAppended: false,
+        valuesPrinted: false,
+        firstMismatchIdentity: null,
+      },
+    });
+    expect(rendered.renderedText).toContain("k3s-application-pod");
+    expect(rendered.renderedText).toContain("30");
+    expect(rendered.renderedText).toContain("matched");
+    expect(rendered.renderedText).toContain("no              no");
+  });
+
+  test("keeps JSON and YAML machine projections isomorphic", () => {
+    const payload = {
+      ok: true,
+      diagnosis: {
+        networkPlane: "k3s-application-pod",
+        scanned: 30,
+        accepted: 12,
+        produced: 8,
+        rejected: { classification: "matched", byReason: {} },
+        mutation: false,
+        topicAppended: false,
+        valuesPrinted: false,
+      },
+    };
+    const json = JSON.parse(renderMachine("replay", payload, "json").renderedText);
+    const yaml = Bun.YAML.parse(renderMachine("replay", payload, "yaml").renderedText);
+    expect(yaml).toEqual(json);
+  });
+});

scripts/src/platform-infra-kafka.ts

@@ -2,6 +2,8 @@ import { readFileSync } from "node:fs";
 import type { UniDeskConfig } from "./config";
 import { rootPath } from "./config";
 import type { RenderedCliResult } from "./output";
+import { renderMachine } from "./agentrun/render";
+import { resolveAgentRunLaneTarget } from "./agentrun-lanes";
 import {
   capture,
   compactCapture,
@@ -17,6 +19,7 @@ const configLabel = "config/platform-infra/kafka.yaml";
 const fieldManager = "unidesk-platform-infra-kafka";
 const y = createYamlFieldReader(configLabel);
 const kafkaConsumerGroupsNativeScript = readFileSync(rootPath("scripts", "native", "platform-infra", "kafka-consumer-groups.py"), "utf8");
+const kafkaReplayDiagnosticsNativeScript = readFileSync(rootPath("scripts", "native", "platform-infra", "kafka-replay-diagnostics.py"), "utf8");
 
 interface KafkaTarget {
   id: string;
@@ -116,6 +119,13 @@ interface PlatformKafkaConfig {
     smokeTopic: string;
   };
   management: {
+    replay: {
+      defaultLimit: number;
+      maxLimit: number;
+      defaultTimeoutMs: number;
+      maxTimeoutMs: number;
+      commandGraceMs: number;
+    };
     defaultTailLimit: number;
     maxTailLimit: number;
     defaultGroupListLimit: number;
@@ -164,6 +174,15 @@ interface KafkaGroupCleanupOptions extends CommonOptions {
   limit: number | null;
 }
 
+export interface KafkaReplayOptions extends CommonOptions {
+  sessionId: string;
+  traceId: string | null;
+  lane: string | null;
+  limit: number | null;
+  timeoutMs: number | null;
+  output: "text" | "json" | "yaml";
+}
+
 export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown> | RenderedCliResult> {
   const [action = "plan"] = args;
   if (action === "plan") {
@@ -203,6 +222,12 @@ export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args:
     const result = await tail(config, options);
     return options.full || options.raw ? result : renderTail(result);
   }
+  if (action === "replay") {
+    const options = parseKafkaReplayOptions(args.slice(1));
+    const result = await replayAgentRun(config, options);
+    if (options.output === "json" || options.output === "yaml") return renderMachine("platform-infra kafka replay agentrun", result, options.output, result.ok !== false);
+    return renderKafkaReplay(result);
+  }
   if (action === "produce") return await produce(config, await parseProduceOptions(args.slice(1)));
   return {
     ok: false,
@@ -214,7 +239,7 @@ export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args:
 
 export function kafkaHelp(): Record<string, unknown> {
   return {
-    command: "platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce",
+    command: "platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|replay|produce",
     configTruth: configLabel,
     usage: [
       "bun scripts/cli.ts platform-infra kafka plan --target NC01",
@@ -228,6 +253,7 @@ export function kafkaHelp(): Record<string, unknown> {
       "bun scripts/cli.ts platform-infra kafka groups cleanup --node NC01 --policy hwlab-cloud-api-sse [--confirm]",
       "bun scripts/cli.ts platform-infra kafka offsets --node NC01 [--topic hwlab.event.v1]",
       "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5",
+      "bun scripts/cli.ts platform-infra kafka replay agentrun --node NC01 --lane nc01-v02 --session-id <id> --trace-id <id> --no-publish [-o json|yaml]",
       "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key <trace-or-session>",
     ],
     boundary: "Kafka runtime belongs to platform-infra. Durable producers and fixed-group consumers are enabled; browser delivery remains a database-projected SSE concern.",
@@ -350,8 +376,16 @@ function parseClient(record: Record<string, unknown>, index: number): KafkaClien
 
 function parseManagement(root: Record<string, unknown>): PlatformKafkaConfig["management"] {
   const management = y.objectField(root, "management", "");
+  const replay = y.objectField(management, "replay", "management");
   const cleanup = y.objectField(management, "consumerGroupCleanup", "management");
   return {
+    replay: {
+      defaultLimit: positiveInteger(replay, "defaultLimit", "management.replay"),
+      maxLimit: positiveInteger(replay, "maxLimit", "management.replay"),
+      defaultTimeoutMs: positiveInteger(replay, "defaultTimeoutMs", "management.replay"),
+      maxTimeoutMs: positiveInteger(replay, "maxTimeoutMs", "management.replay"),
+      commandGraceMs: positiveInteger(replay, "commandGraceMs", "management.replay"),
+    },
     defaultTailLimit: positiveInteger(management, "defaultTailLimit", "management"),
     maxTailLimit: positiveInteger(management, "maxTailLimit", "management"),
     defaultGroupListLimit: positiveInteger(management, "defaultGroupListLimit", "management"),
@@ -407,6 +441,9 @@ function validateConfig(kafka: PlatformKafkaConfig): void {
   if (!/^[0-9]+Gi$/u.test(kafka.cluster.storage.size)) throw new Error(`${configLabel}.cluster.storage.size must be a Gi quantity`);
   if (!kafka.topics.some((topic) => topic.name === kafka.validation.smokeTopic)) throw new Error(`${configLabel}.validation.smokeTopic must reference one of topics[].name`);
   if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultTopic)) throw new Error(`${configLabel}.management.defaultTopic must reference one of topics[].name`);
+  if (kafka.management.replay.defaultLimit > kafka.management.replay.maxLimit) throw new Error(`${configLabel}.management.replay.defaultLimit must be <= maxLimit`);
+  if (kafka.management.replay.defaultTimeoutMs > kafka.management.replay.maxTimeoutMs) throw new Error(`${configLabel}.management.replay.defaultTimeoutMs must be <= maxTimeoutMs`);
+  if (kafka.management.replay.maxTimeoutMs + kafka.management.replay.commandGraceMs >= 60_000) throw new Error(`${configLabel}.management.replay timeout plus command grace must stay below the trans short-connection budget`);
   if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`);
   if (kafka.management.defaultGroupListLimit > kafka.management.maxGroupListLimit) throw new Error(`${configLabel}.management.defaultGroupListLimit must be <= management.maxGroupListLimit`);
   if (kafka.management.consumerGroupCleanup.defaultCandidateLimit > kafka.management.consumerGroupCleanup.maxCandidateLimit) throw new Error(`${configLabel}.management.consumerGroupCleanup.defaultCandidateLimit must be <= maxCandidateLimit`);
@@ -661,6 +698,50 @@ async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promis
   };
 }
 
+async function replayAgentRun(config: UniDeskConfig, options: KafkaReplayOptions): Promise<Record<string, unknown>> {
+  const kafka = readKafkaConfig();
+  const target = resolveTarget(kafka, options.targetId);
+  const lane = resolveAgentRunLaneTarget({ node: target.id, lane: options.lane }).spec;
+  if (lane.nodeId !== target.id) throw new Error(`AgentRun lane ${lane.lane} belongs to ${lane.nodeId}, not Kafka target ${target.id}`);
+  const inputTopic = resolveTopic(kafka, "codex-stdio.raw.v1", null);
+  if (inputTopic === null) throw new Error("Kafka replay input topic codex-stdio.raw.v1 is not declared");
+  const limit = options.limit ?? kafka.management.replay.defaultLimit;
+  if (!Number.isInteger(limit) || limit < 1 || limit > kafka.management.replay.maxLimit) throw new Error(`Kafka replay --limit must be an integer from 1 to ${kafka.management.replay.maxLimit}`);
+  const timeoutMs = options.timeoutMs ?? kafka.management.replay.defaultTimeoutMs;
+  if (!Number.isInteger(timeoutMs) || timeoutMs < 250 || timeoutMs > kafka.management.replay.maxTimeoutMs) throw new Error(`Kafka replay --timeout-ms must be an integer from 250 to ${kafka.management.replay.maxTimeoutMs}`);
+  const runtimeEnvironment = {
+    KAFKA_NAMESPACE: target.namespace,
+    KAFKA_CLUSTER: kafka.cluster.name,
+    KAFKA_BOOTSTRAP: brokerBootstrap(kafka),
+    KAFKA_INPUT_TOPIC: inputTopic,
+    KAFKA_REPLAY_SESSION_ID: options.sessionId,
+    KAFKA_REPLAY_TRACE_ID: options.traceId ?? "",
+    KAFKA_REPLAY_LIMIT: String(limit),
+    KAFKA_REPLAY_TIMEOUT_MS: String(timeoutMs),
+    KAFKA_REPLAY_COMMAND_GRACE_MS: String(kafka.management.replay.commandGraceMs),
+    AGENTRUN_NAMESPACE: lane.runtime.namespace,
+    AGENTRUN_MANAGER_DEPLOYMENT: lane.runtime.managerDeployment,
+  };
+  const remoteScript = [
+    ...Object.entries(runtimeEnvironment).map(([key, value]) => `export ${key}=${shQuote(value)}`),
+    "python3 - <<'PY'",
+    kafkaReplayDiagnosticsNativeScript.trimEnd(),
+    "PY",
+  ].join("\n");
+  const result = await capture(config, target.route, ["sh"], remoteScript);
+  const parsed = parseJsonOutput(result.stdout);
+  return {
+    ok: result.exitCode === 0 && parsed?.ok === true,
+    action: "platform-infra-kafka-replay-agentrun",
+    mode: "no-publish",
+    mutation: false,
+    target: targetSummary(target),
+    lane: { node: lane.nodeId, lane: lane.lane, namespace: lane.runtime.namespace, managerDeployment: lane.runtime.managerDeployment },
+    diagnosis: parsed,
+    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
+  };
+}
+
 async function produce(config: UniDeskConfig, options: KafkaProduceOptions): Promise<Record<string, unknown>> {
   const kafka = readKafkaConfig();
   const target = resolveTarget(kafka, options.targetId);
@@ -1840,6 +1921,22 @@ function renderTail(result: Record<string, unknown>): RenderedCliResult {
   ]);
 }
 
+export function renderKafkaReplay(result: Record<string, unknown>): RenderedCliResult {
+  const diagnosis = record(result.diagnosis);
+  const rejected = record(diagnosis.rejected);
+  const mismatch = record(diagnosis.firstMismatchIdentity);
+  return rendered(result, "platform-infra kafka replay agentrun", [
+    "PLATFORM-INFRA KAFKA REPLAY",
+    ...table(["PLANE", "APPLICATION", "NAMESPACE", "WORKLOAD"], [[stringValue(diagnosis.networkPlane), stringValue(diagnosis.application), stringValue(diagnosis.namespace), stringValue(diagnosis.workload)]]),
+    "",
+    ...table(["TOPIC", "SCANNED", "ACCEPTED", "PRODUCED", "CLASSIFICATION"], [[stringValue(diagnosis.topic), stringValue(diagnosis.scanned), stringValue(diagnosis.accepted), stringValue(diagnosis.produced), stringValue(rejected.classification)]]),
+    "",
+    ...table(["MUTATION", "TOPIC_APPENDED", "VALUES_PRINTED"], [[boolText(diagnosis.mutation), boolText(diagnosis.topicAppended), boolText(diagnosis.valuesPrinted)]]),
+    ...(diagnosis.firstMismatchIdentity === null ? [] : ["", `MISMATCH  session=${stringValue(mismatch.sessionId)} trace=${stringValue(mismatch.traceId)} reason=${stringValue(mismatch.reason)}`]),
+    ...remoteErrorLines(result),
+  ]);
+}
+
 function remoteErrorLines(result: Record<string, unknown>): string[] {
   if (result.ok !== false) return [];
   const remote = record(result.remote);
@@ -1930,6 +2027,42 @@ function parseInspectOptions(args: string[]): KafkaInspectOptions {
   return { ...parseCommonOptions(commonArgs), topic, group, limit };
 }
 
+export function parseKafkaReplayOptions(args: string[]): KafkaReplayOptions {
+  const commonArgs: string[] = [];
+  let application: string | null = null;
+  let sessionId: string | null = null;
+  let traceId: string | null = null;
+  let lane: string | null = null;
+  let limit: number | null = null;
+  let timeoutMs: number | null = null;
+  let output: KafkaReplayOptions["output"] = "text";
+  let noPublish = false;
+  for (let index = 0; index < args.length; index += 1) {
+    const arg = args[index];
+    if (!arg.startsWith("--") && application === null) application = arg;
+    else if (arg === "--session-id") { sessionId = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--trace-id") { traceId = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--lane") { lane = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--limit") { limit = Number(parseSimpleOption(args[index + 1], arg)); index += 1; }
+    else if (arg === "--timeout-ms") { timeoutMs = Number(parseSimpleOption(args[index + 1], arg)); index += 1; }
+    else if (arg === "--no-publish" || arg === "--dry-run") noPublish = true;
+    else if (arg === "--json") output = "json";
+    else if (arg === "-o" || arg === "--output") {
+      const value = parseSimpleOption(args[index + 1], arg);
+      if (value !== "text" && value !== "json" && value !== "yaml") throw new Error(`${arg} must be text|json|yaml`);
+      output = value;
+      index += 1;
+    } else {
+      commonArgs.push(arg);
+      if (arg === "--target" || arg === "--node") { commonArgs.push(args[index + 1] ?? ""); index += 1; }
+    }
+  }
+  if (application !== "agentrun") throw new Error("Kafka replay requires application 'agentrun'");
+  if (sessionId === null) throw new Error("Kafka replay requires --session-id <id>");
+  if (!noPublish) throw new Error("Kafka replay currently requires --no-publish; product topic replay is unsupported");
+  return { ...parseCommonOptions(commonArgs), sessionId, traceId, lane, limit, timeoutMs, output };
+}
+
 function parseGroupCleanupOptions(args: string[]): KafkaGroupCleanupOptions {
   const commonArgs: string[] = [];
   let policyId: string | null = null;