Merge pull request #278 from pikasTech/feat/stdio-kafka-event-regenerate

feat: 增加 stdio Kafka 事件部分重建 CLI

docs/reference/cli.md

@@ -15,6 +15,12 @@ scripts/src/*.ts
 
 CLI 默认输出 JSON。空 stdout 是失败，不是成功。每个命令都必须返回足够继续排障的结构化信息，包括 id、status、log path 或后续命令。
 
+`kafka regenerate agentrun` 是高噪声调试命令的显式例外：
+
+- 默认输出 compact text；
+- 显式 `--json` 或 `--format json` 才输出结构化摘要；
+- 两种输出都必须包含下一条可执行命令提示。
+
 一次性 JSON body 优先通过 `--json-stdin` 和 quoted heredoc 输入；不要为了把 heredoc/stdin 内容传给 CLI 而先落临时 JSON dump 文件。`--json-file` 只用于可复用、已受控的输入文件。
 
 长操作必须是 fire-and-forget 或短异步资源操作。CLI 调用应在 60 秒内返回。创建 run 或启动 runner 的命令返回创建出的资源和轮询命令，不等待模型 turn 完成。
@@ -29,6 +35,7 @@ CLI 默认输出 JSON。空 stdout 是失败，不是成功。每个命令都必
 ./scripts/agentrun runs events <runId> --after-seq <n> --limit <n>
 ./scripts/agentrun runs result <runId> [--command-id <commandId>]
 ./scripts/agentrun runs cancel <runId> [--reason <text>]
+./scripts/agentrun kafka regenerate agentrun --session-id <id> [--trace-id <id>] [--from kafka|jsonl] [--jsonl-file <stdio.jsonl>] [--no-publish|--dry-run] [--output-jsonl <events.jsonl>] [--json|--format json]
 ./scripts/agentrun commands create <runId> --type turn|steer|interrupt --json-stdin|--json-file <payload.json>
 ./scripts/agentrun commands show <commandId> --run-id <runId>
 ./scripts/agentrun commands result <commandId> --run-id <runId>
@@ -48,6 +55,11 @@ CLI 默认输出 JSON。空 stdout 是失败，不是成功。每个命令都必
 - `commands create` 创建 durable command；`turn` 启动一轮对话，`steer` 只在同 run 有 active turn 时由 runner 转发到 Codex `turn/steer`，迟到 steer 返回结构化 blocked，不终结 run。
 - `runner start` 启动本地进程或 Kubernetes Job，并返回 process/job identity、log path 和 poll commands。
 - `events` 默认分页且有界。
+- `kafka regenerate agentrun` 只生成 `stdio-derived partial reconstruction`：
+  - 默认 topic 为 `agentrun.event.debug.v1`；
+  - 永久禁止输出到 `agentrun.event.v1`；
+  - 不恢复或伪造原始 `eventId`、`outboxSeq` 与完整 lifecycle；
+  - 可用 JSONL 固件在无 manager、runner 和数据库的环境中执行。
 - `server logs` 返回有界日志，并指向完整日志文件。
 - `server start` 默认后台启动本地 manager，短返回 pid、pidFile、logPath 和后续 status/stop 命令；`--foreground` 仅用于容器入口或显式调试。
 - `server status` 暴露 port、process id、health 和 log paths；health 不通时仍返回结构化 JSON。

docs/reference/spec-v01-cli.md

@@ -41,6 +41,7 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 ./scripts/agentrun runs cancel <runId> [--reason <text>]
 ./scripts/agentrun kafka tail [agentrun|stdio] [--topic <topic>] [--brokers <host:port,...>] [--limit <n>] [--timeout-ms <ms>] [--run-id <id>] [--session-id <id>] [--command-id <id>] [--trace-id <id>] [--values]
 ./scripts/agentrun kafka events [agentrun|stdio] [--topic <topic>] [--brokers <host:port,...>] [--limit <n>] [--timeout-ms <ms>]
+./scripts/agentrun kafka regenerate agentrun --session-id <id> [--trace-id <id>] [--from kafka|jsonl] [--jsonl-file <stdio.jsonl>] [--no-publish|--dry-run] [--output-jsonl <events.jsonl>] [--json|--format json]
 ./scripts/agentrun commands create <runId> --type turn|steer|interrupt --json-stdin|--json-file <payload.json>
 ./scripts/agentrun commands show <commandId> --run-id <runId>
 ./scripts/agentrun commands result <commandId> --run-id <runId>
@@ -103,6 +104,27 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 - `events --tail <n>` 与 `--after-seq`/`--limit` 组合时，只显示本次分页结果最后 `n` 条 summary；`--tail-summary` 是默认取最多 20 条 tail 的低噪声快捷入口。
 - `events --format tsv` 只在显式请求时输出 TSV 文本，用于人工现场跟踪和上层 trace 压缩；默认 JSON envelope 行为不得改变。
 - `kafka tail/events` 是 AgentRun Kafka 事件流的受控查询入口，默认读取 `agentrun.event.v1`，传 `stdio` 时读取 `codex-stdio.raw.v1`；默认只输出 topic/offset/key hash/value hash/bytes/schema/eventType/runId/sessionId/commandId/direction/method，不输出 Kafka value。只有显式 `--values` 或 `--raw-values` 才返回有界 value 预览，用于针对性调试裸 stdio 或 manager event。
+- `kafka regenerate agentrun` 是 stdio 单步调试入口：
+  - 按已有 `sessionId` 从 `codex-stdio.raw.v1` 或 JSONL 固件读取帧；
+  - 直接复用生产 Codex stdio stateful reducer；
+  - 不依赖 manager、runner 或数据库运行面；
+  - 默认发布到 `agentrun.event.debug.v1`；
+  - 默认 consumer group 使用 `agentrun-v02-stdio-regenerate-debug-*` 前缀；
+  - 默认输出 compact text；
+  - 只有显式 `--json` 或 `--format json` 才输出结构化摘要；
+  - 每次输出都必须包含下一条可执行命令提示。
+- stdio 再生结果必须声明 `stdio-derived partial reconstruction`：
+  - schema 固定为 `agentrun.event.reconstruction.v1`；
+  - event type 固定为 `agentrun.event.reconstructed`；
+  - `eventId`、`outboxSeq` 与 `event.id` 固定为 `null`；
+  - 不得声称恢复完整 lifecycle 或 durable commit 顺序；
+  - 原始 `sessionId` 必须保留；
+  - 仅当 `ses_agentrun_<uuid-with-underscores>` 可逆转换时生成 `hwlabSessionId=ses_<uuid-with-hyphens>`；
+  - identity 转换依据写入 `reconstruction.identityDerivation`。
+- stdio 再生结果禁止写入 `agentrun.event.v1`：
+  - CLI 不提供产品 topic override；
+  - 即使显式传入产品 topic，也必须返回 `tenant-policy-denied`；
+  - JSONL 输出使用每行 `{topic,partition,offset,key,value,headers}` 的 Kafka record 形状。
 - `server start` 默认以本地后台进程启动 manager，立刻返回 pid、pidFile、logPath、baseUrl 和后续 `status/stop` 命令；只有显式 `--foreground` 才允许占用当前终端。启动前必须检查 pidFile 与端口占用，避免同一端口上堆叠临时 manager。
 - `server status` 必须同时返回本地 pid/port/logPath 状态和 `/health/readiness` 结果；即使 readiness 失败，也要输出结构化 JSON 和 failure details。
 - `server logs` 必须返回有界日志尾部、bytes、truncated 和 logPath；找不到日志文件时也必须返回非空 JSON。
@@ -165,6 +187,23 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 
 阅读本文和 [spec-v01-queue.md](spec-v01-queue.md)，然后用正式 CLI 创建一个 Queue task，轮询 `queue list/show/stats/commander/read`，再使用 `queue show` 返回的 `sessionPath` 调用 `sessions output/trace`。确认 Queue 命令只返回 summary/stats/read/sessionPath，不代理 output/trace；Session 命令能查询输出和 trace。最终交互验收必须使用真实 runtime，不使用 mock，不写自动交互脚本。
 
+### T7 stdio 事件部分重建
+
+- 使用固定 JSONL 固件执行 `kafka regenerate agentrun`：
+  - 不启动 manager、runner 或数据库；
+  - 确认生产 reducer 与 CLI 共用同一实现；
+  - 确认默认输出 topic/group 均为 debug 命名空间；
+  - 确认默认 compact text 与显式 `--json` 都非空；
+  - 确认输出包含可执行的 `nextHint`。
+- 检查每条 reconstruction event：
+  - `eventId=null`；
+  - `outboxSeq=null`；
+  - `completeLifecycle=false`；
+  - identity derivation 仅对严格 UUID 形态生效。
+- 显式把 output topic 设为 `agentrun.event.v1`：
+  - 确认命令失败；
+  - 确认没有 producer 写入。
+
 ## 规格的实现情况
 
 | 规格项 | 状态 | 说明 |
@@ -178,6 +217,7 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 | Queue CLI | 已实现/Q1 | 已提供 `queue submit/list/show/stats/commander/read/cancel`，通过 manager REST 访问 Queue task 和 stats，不直连 Postgres。 |
 | Queue dispatch/refresh CLI | 已实现/Q2 | `queue dispatch` 受控创建 Core run/command/runner job；`queue refresh` 从 Core run/command 终态回写 Queue task/latestAttempt。 |
 | 本地 server 生命周期 CLI | 已实现/Q2 hardening | `server start` 默认后台短返回，`server status/stop` 提供 pid、port、logPath 和 readiness 可见性；`--foreground` 保留给容器/显式调试。 |
+| stdio 事件部分重建 CLI | 已实现/debug | 按 `sessionId` 从 Kafka 或 JSONL 调用生产 reducer，默认写独立 debug topic/group，永久禁止写产品 topic，并显式披露不可恢复的 durable identity 与 lifecycle。 |
 | Session CLI | 已实现/Q3 | 已提供 `sessions ps/show/send/cancel/trace/output/read`；默认 ps 只显示 running/unread，terminal 后自动 unread，read cursor 由 CLI 标记。用户级 CLI 只保留 `send`；`turn/steer` 只作为 manager 内部 command type，不再作为 CLI 入口或兼容 alias。 |
 | CLI 测试规格 | 已定义/已验证主闭环 | 综合联调见 [spec-v01-validation.md](spec-v01-validation.md)；每次发布仍按手动交互验收复跑。 |
 | `deepseek` profile CLI | 已实现/已通过主闭环 | `secrets codex render --profile deepseek`、`backends list`、`runner start --backend`、`runner job` 和 JSON 错误可见性已实现；真实 CLI/RESTful 联调已通过 `codex -> deepseek -> codex` 切换主闭环。 |

scripts/src/cli.ts

@@ -17,6 +17,12 @@ import { backendProfileSpec, isBackendProfile } from "../../src/common/backend-p
 import { outputBytesFromPayload, outputTruncatedFromPayload } from "../../src/common/output.js";
 import { redactJson, redactText } from "../../src/common/redaction.js";
 import { tailAgentRunKafka } from "../../src/common/kafka-events.js";
+import {
+  parseStdioKafkaFrameJsonl,
+  reconstructionRecordsJsonl,
+  regenerateAgentRunEventsFromStdio,
+  type StdioFrameReader,
+} from "../../src/debug/stdio-event-reconstruction.js";
 
 interface ParsedArgs {
   positional: string[];
@@ -33,8 +39,9 @@ interface PlainTextOutput {
 type CliResult = JsonValue | PlainTextOutput;
 
 export async function runCli(argv: string[]): Promise<void> {
+  const parsed = parseArgs(argv);
   try {
-    const result = await dispatch(parseArgs(argv));
+    const result = await dispatch(parsed);
     if (isPlainTextOutput(result)) {
       process.stdout.write(result.text.endsWith("\n") ? result.text : `${result.text}\n`);
       return;
@@ -42,18 +49,31 @@ export async function runCli(argv: string[]): Promise<void> {
     print({ ok: true, data: result });
   } catch (error) {
     const status = error instanceof AgentRunError ? error.httpStatus : 1;
-    print({ ok: false, ...(error instanceof AgentRunError ? { failureKind: error.failureKind, message: error.message } : { failureKind: "infra-failed", message: error instanceof Error ? error.message : String(error) }), error: errorToJson(error) });
+    const failureKind = error instanceof AgentRunError ? error.failureKind : "infra-failed";
+    const message = error instanceof Error ? error.message : String(error);
+    if (isDefaultCompactStdioReconstruction(parsed)) {
+      process.stdout.write(`stdio 部分重建失败 failureKind=${failureKind} message=${message}\n下一步: ./scripts/agentrun kafka regenerate agentrun --help\n`);
+    } else {
+      print({ ok: false, failureKind, message, error: errorToJson(error) });
+    }
     process.exitCode = status === 0 ? 1 : status > 255 ? 1 : status;
   }
 }
 
+function isDefaultCompactStdioReconstruction(args: ParsedArgs): boolean {
+  if (args.positional[0] !== "kafka" || args.positional[1] !== "regenerate" || args.positional[2] !== "agentrun") return false;
+  return args.flags.get("json") !== true && optionalFlag(args, "format") !== "json";
+}
+
 async function dispatch(args: ParsedArgs): Promise<CliResult> {
   const [group, command, id] = args.positional;
   if (!group || group === "help" || group === "--help") return help(args);
+  if (group === "kafka" && command === "regenerate" && id === "agentrun" && args.flags.get("help") === true) return kafkaRegenerateAgentRunHelp(args);
   if (args.flags.get("help") === true) return help(args, group);
   if (command === "help" || command === "--help") return help(args, group);
   if (group === "manager" && (command === "url" || command === "resolve-url" || command === "status")) return managerEndpoint(args);
   if (group === "kafka" && (command === "tail" || command === "events")) return kafkaTail(args);
+  if (group === "kafka" && command === "regenerate" && id === "agentrun") return kafkaRegenerateAgentRun(args);
   if (group === "server" && command === "start") return startServer(args);
   if (group === "server" && command === "status") return serverStatus(args);
   if (group === "server" && command === "logs") return serverLogs(args);
@@ -184,6 +204,116 @@ async function kafkaTail(args: ParsedArgs): Promise<JsonValue> {
   }) as JsonValue;
 }
 
+async function kafkaRegenerateAgentRun(args: ParsedArgs): Promise<CliResult> {
+  if (args.flags.has("allow-product-topic")) {
+    throw new AgentRunError("tenant-policy-denied", "--allow-product-topic is not supported; stdio reconstruction can never write a product topic", { httpStatus: 2 });
+  }
+  const sessionId = flag(args, "session-id", "");
+  if (!sessionId) throw new AgentRunError("schema-invalid", "kafka regenerate agentrun requires --session-id", { httpStatus: 2 });
+  const jsonlFile = optionalFlag(args, "jsonl-file");
+  const sourceKind = flag(args, "from", jsonlFile ? "jsonl" : "kafka");
+  if (sourceKind !== "kafka" && sourceKind !== "jsonl") {
+    throw new AgentRunError("schema-invalid", "kafka regenerate agentrun --from must be kafka|jsonl", { httpStatus: 2 });
+  }
+  if (sourceKind === "jsonl" && !jsonlFile) {
+    throw new AgentRunError("schema-invalid", "kafka regenerate agentrun --from jsonl requires --jsonl-file", { httpStatus: 2 });
+  }
+  if (sourceKind === "kafka" && jsonlFile) {
+    throw new AgentRunError("schema-invalid", "--jsonl-file requires --from jsonl", { httpStatus: 2 });
+  }
+  const dryRun = args.flags.get("dry-run") === true;
+  const noPublish = args.flags.get("no-publish") === true;
+  const outputJsonl = optionalFlag(args, "output-jsonl");
+  if (dryRun && outputJsonl) {
+    throw new AgentRunError("schema-invalid", "--dry-run cannot write --output-jsonl; use --no-publish for an offline file transform", { httpStatus: 2 });
+  }
+  let reader: StdioFrameReader | undefined;
+  if (jsonlFile) {
+    const inputText = await readFile(path.resolve(jsonlFile), "utf8");
+    const records = parseStdioKafkaFrameJsonl(inputText, optionalFlag(args, "input-topic") ?? "codex-stdio.raw.v1");
+    reader = async () => ({ sourceKind: "jsonl", records, scanned: records.length, invalidJson: 0, timedOut: false, elapsedMs: 0 });
+  }
+  const dependencies = reader ? { reader } : {};
+  const result = await regenerateAgentRunEventsFromStdio({
+    sessionId,
+    traceId: optionalFlag(args, "trace-id"),
+    ...(optionalFlag(args, "brokers") ? { brokers: optionalFlag(args, "brokers")! } : {}),
+    ...(optionalFlag(args, "input-topic") ? { inputTopic: optionalFlag(args, "input-topic")! } : {}),
+    ...(optionalFlag(args, "output-topic") ? { outputTopic: optionalFlag(args, "output-topic")! } : {}),
+    ...(optionalFlag(args, "group-id") ? { groupId: optionalFlag(args, "group-id")! } : {}),
+    ...(optionalNumberFlag(args, "limit") !== null ? { limit: optionalNumberFlag(args, "limit")! } : {}),
+    ...(optionalNumberFlag(args, "timeout-ms") !== null ? { timeoutMs: optionalNumberFlag(args, "timeout-ms")! } : {}),
+    publish: !dryRun && !noPublish,
+  }, dependencies);
+  if (dryRun || noPublish) result.nextHint = kafkaRegeneratePublishHint(args, sessionId, sourceKind, jsonlFile);
+  if (outputJsonl) {
+    const absolute = path.resolve(outputJsonl);
+    await mkdir(path.dirname(absolute), { recursive: true });
+    await writeFile(absolute, reconstructionRecordsJsonl(result.records), "utf8");
+  }
+  const { records, ...compact } = result;
+  const summary = {
+    ...compact,
+    dryRun,
+    outputJsonl: outputJsonl ? path.resolve(outputJsonl) : null,
+    outputJsonlRecords: outputJsonl ? records.length : 0,
+  } as unknown as JsonRecord;
+  const format = optionalFlag(args, "format") ?? (args.flags.get("json") === true ? "json" : "compact");
+  if (args.flags.get("json") === true && format !== "json") {
+    throw new AgentRunError("schema-invalid", "--json cannot be combined with non-JSON --format", { httpStatus: 2 });
+  }
+  if (format === "compact" || format === "text") return plainTextOutput(formatStdioReconstructionCompact(summary));
+  if (format !== "json") throw new AgentRunError("schema-invalid", "--format must be json|compact", { httpStatus: 2 });
+  return summary;
+}
+
+function formatStdioReconstructionCompact(result: JsonRecord): string {
+  const source = jsonRecordValue(result.source) ?? {};
+  const output = jsonRecordValue(result.output) ?? {};
+  return [
+    `stdio 部分重建 session=${String(source.sessionId ?? "-")} frames=${String(result.selectedFrameCount ?? 0)} events=${String(output.recordCount ?? 0)} publish=${String(output.publish === true)} topic=${String(output.topic ?? "-")}`,
+    `边界: ${String(result.reconstructionKind ?? "-")}；originalEventId/outboxSeq 不可恢复；完整 lifecycle 不可恢复`,
+    `下一步: ${String(result.nextHint ?? "-")}`,
+  ].join("\n");
+}
+
+function kafkaRegenerateAgentRunHelp(args: ParsedArgs): CliResult {
+  const data: JsonRecord = {
+    command: "kafka regenerate agentrun",
+    purpose: "按 stdio sessionId 调用生产 reducer 生成部分重建调试事件",
+    usage: "./scripts/agentrun kafka regenerate agentrun --session-id <id> [--trace-id <id>] [--from kafka|jsonl] [--jsonl-file <stdio.jsonl>] [--no-publish|--dry-run] [--output-jsonl <events.jsonl>] [--json|--format json]",
+    defaultInputTopic: "codex-stdio.raw.v1",
+    defaultOutputTopic: "agentrun.event.debug.v1",
+    productTopicAllowed: false,
+    reconstructionKind: "stdio-derived partial reconstruction",
+    valuesPrinted: false,
+  };
+  const format = optionalFlag(args, "format") ?? (args.flags.get("json") === true ? "json" : "compact");
+  if (format === "json") return data;
+  if (format !== "compact" && format !== "text") throw new AgentRunError("schema-invalid", "--format must be json|compact", { httpStatus: 2 });
+  return plainTextOutput([
+    String(data.purpose),
+    `用法: ${String(data.usage)}`,
+    `默认: codex-stdio.raw.v1 -> agentrun.event.debug.v1；产品 topic 永久禁止`,
+  ].join("\n"));
+}
+
+function kafkaRegeneratePublishHint(args: ParsedArgs, sessionId: string, sourceKind: "kafka" | "jsonl", jsonlFile: string | null): string {
+  const parts = ["./scripts/agentrun", "kafka", "regenerate", "agentrun", "--from", sourceKind];
+  if (jsonlFile) parts.push("--jsonl-file", path.resolve(jsonlFile));
+  parts.push("--session-id", sessionId);
+  for (const name of ["trace-id", "brokers", "input-topic", "output-topic", "group-id", "limit", "timeout-ms"] as const) {
+    const value = optionalFlag(args, name);
+    if (value) parts.push(`--${name}`, value);
+  }
+  parts.push("--json");
+  return parts.map(cliShellToken).join(" ");
+}
+
+function cliShellToken(value: string): string {
+  return /^[A-Za-z0-9._:/=-]+$/u.test(value) ? value : `'${value.replace(/'/gu, `'"'"'`)}'`;
+}
+
 async function showRun(args: ParsedArgs, runId: string): Promise<JsonValue> {
   const run = await client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}`);
   if (wantsExpandedOutput(args)) return run;
@@ -2086,6 +2216,7 @@ function help(args: ParsedArgs, group?: string): JsonRecord {
       "runs cancel <runId> [--reason <text>]",
       "kafka tail [agentrun|stdio] [--topic <topic>] [--brokers <host:port,...>] [--limit <n>] [--timeout-ms <ms>] [--run-id <id>] [--session-id <id>] [--command-id <id>] [--trace-id <id>] [--values]",
       "kafka events [agentrun|stdio] [--topic <topic>] [--brokers <host:port,...>] [--limit <n>] [--timeout-ms <ms>]",
+      "kafka regenerate agentrun --session-id <id> [--trace-id <id>] [--from kafka|jsonl] [--jsonl-file <stdio.jsonl>] [--input-topic codex-stdio.raw.v1] [--output-topic agentrun.event.debug.v1] [--group-id <debug-group>] [--no-publish|--dry-run] [--output-jsonl <events.jsonl>] [--format json|compact] [--json]",
       "sessions ps [--state default|running|unread|terminal|idle|all] [--profile codex|deepseek|minimax-m3|dsflash-go|<dynamic-profile>|M3] [--reader-id <reader>]",
       "sessions create [sessionId] [--profile codex|deepseek|minimax-m3|dsflash-go|<dynamic-profile>|M3] [--expires-in-days <n>]",
       "sessions storage <sessionId>",

src/backend/codex-stdio.ts

@@ -145,6 +145,53 @@ interface SuppressedNotificationSummary {
   byItemType: Record<string, number>;
 }
 
+export interface CodexStdioReducedNotification {
+  events: BackendEvent[];
+  threadId?: string;
+  turnId?: string;
+  terminal?: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null };
+}
+
+export class CodexStdioEventReducer {
+  private assistantText = "";
+  private readonly assistantDeltaProgress = createAssistantDeltaProgressState();
+  private readonly completedAssistantMessages: CompletedAssistantMessage[] = [];
+  private readonly deferredTerminalEvents: BackendEvent[] = [];
+  private readonly suppressedNotifications = createSuppressedNotificationSummary();
+
+  reduce(message: JsonRecord): CodexStdioReducedNotification {
+    const normalized = normalizeCodexNotification(message, this.suppressedNotifications);
+    const events = [...normalized.events];
+    if (normalized.assistantDelta) {
+      this.assistantText += normalized.assistantDelta.text;
+      const progress = recordAssistantDeltaProgress(this.assistantDeltaProgress, normalized.assistantDelta);
+      if (progress) events.push(progress);
+    }
+    if (normalized.completedAssistantMessage) {
+      this.completedAssistantMessages.push(normalized.completedAssistantMessage);
+      events.push(assistantMessageEventForCompleted(normalized.completedAssistantMessage, this.completedAssistantMessages.length));
+    }
+    if (normalized.terminalEvents) this.deferredTerminalEvents.push(...normalized.terminalEvents);
+    return {
+      events,
+      ...(normalized.threadId ? { threadId: normalized.threadId } : {}),
+      ...(normalized.turnId ? { turnId: normalized.turnId } : {}),
+      ...(normalized.terminal ? { terminal: normalized.terminal } : {}),
+    };
+  }
+
+  finalize(terminal: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } | null): BackendEvent[] {
+    const events = flushAssistantDeltaProgress(this.assistantDeltaProgress);
+    const finalAssistant = terminal?.status === "completed" ? finalAssistantMessageForTurn(this.completedAssistantMessages, this.assistantText) : null;
+    if (finalAssistant) events.push(assistantFinalResponseEvent(finalAssistant));
+    else if (this.completedAssistantMessages.length === 0) events.push(...assistantMessageEventsForTurn(this.assistantText, false));
+    events.push(...suppressedNotificationEvents(this.suppressedNotifications));
+    events.push(...this.deferredTerminalEvents);
+    if (terminal) events.push({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
+    return events;
+  }
+}
+
 type AssistantDeltaProgressState = Map<string, AssistantDeltaProgressItem>;
 
 interface CodexStdioCloseInfo extends JsonRecord {
@@ -622,11 +669,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
     events.push({ type: "terminal_status", payload: { terminalStatus: cancelled.status, failureKind: cancelled.failureKind, message: cancelled.message } });
     return { terminalStatus: cancelled.status, failureKind: cancelled.failureKind, failureMessage: cancelled.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })) };
   }
-  let assistantText = "";
-  const assistantDeltaProgress = createAssistantDeltaProgressState();
-  const completedAssistantMessages: CompletedAssistantMessage[] = [];
-  const deferredTerminalEvents: BackendEvent[] = [];
-  const suppressedNotifications = createSuppressedNotificationSummary();
+  const eventReducer = new CodexStdioEventReducer();
   let waitingFor = "codex-app-server";
   let lastNotificationMethod: string | null = null;
   const turnStartedAt = Date.now();
@@ -812,7 +855,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
     refreshTurnActivity();
     lastNotificationMethod = typeof message.method === "string" ? message.method : "unknown";
     emitCodexNotificationOtel(options, env, message, { threadId: threadId ?? null, turnId: turnId ?? null, waitingFor });
-    const normalized = normalizeCodexNotification(message, suppressedNotifications);
+    const normalized = eventReducer.reduce(message);
     if (normalized.threadId) threadId = normalized.threadId;
     if (normalized.turnId) turnId = normalized.turnId;
     waitingFor = waitingForAfterNotification(message, normalized.terminal !== undefined);
@@ -823,16 +866,6 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
     }
     exposeActiveTurn(normalized.turnId ? "turn-notification" : "notification");
     emitEvents(normalized.events);
-    if (normalized.assistantDelta) {
-      assistantText += normalized.assistantDelta.text;
-      const progress = recordAssistantDeltaProgress(assistantDeltaProgress, normalized.assistantDelta);
-      if (progress) emitEvent(progress);
-    }
-    if (normalized.completedAssistantMessage) {
-      completedAssistantMessages.push(normalized.completedAssistantMessage);
-      emitEvent(assistantMessageEventForCompleted(normalized.completedAssistantMessage, completedAssistantMessages.length));
-    }
-    if (normalized.terminalEvents) deferredTerminalEvents.push(...normalized.terminalEvents);
     if (normalized.terminal && !terminal) {
       terminal = normalized.terminal;
       resolveTerminalAfterNotificationDrain();
@@ -956,13 +989,7 @@ async function runCodexStdioTurnWithSession(options: CodexStdioTurnOptions, sess
   const pendingInterrupt: Promise<void> | null = interruptInFlight as Promise<void> | null;
   if (pendingInterrupt) await pendingInterrupt.catch(() => undefined);
   if (terminal.status !== "completed") appendResultEvents(await session.close());
-  appendResultEvents(flushAssistantDeltaProgress(assistantDeltaProgress));
-  const finalAssistant = terminal.status === "completed" ? finalAssistantMessageForTurn(completedAssistantMessages, assistantText) : null;
-  if (finalAssistant) appendResultEvent(assistantFinalResponseEvent(finalAssistant));
-  else if (completedAssistantMessages.length === 0) appendResultEvents(assistantMessageEventsForTurn(assistantText, false));
-  appendResultEvents(suppressedNotificationEvents(suppressedNotifications));
-  appendResultEvents(deferredTerminalEvents);
-  appendResultEvent({ type: "terminal_status", payload: { terminalStatus: terminal.status, failureKind: terminal.failureKind, message: terminal.message } });
+  appendResultEvents(eventReducer.finalize(terminal));
   await liveEventWrite;
   return { terminalStatus: terminal.status, failureKind: terminal.failureKind, failureMessage: terminal.message, events: events.map((event) => ({ ...event, payload: redactJson(event.payload) })), ...(threadId ? { threadId } : {}), ...(turnId ? { turnId } : {}) };
 }

src/debug/stdio-event-reconstruction.ts

@@ -0,0 +1,590 @@
+import { randomUUID } from "node:crypto";
+import { Kafka, logLevel, type Consumer } from "kafkajs";
+import { CodexStdioEventReducer } from "../backend/codex-stdio.js";
+import { AgentRunError } from "../common/errors.js";
+import { normalizeRunEventPayload } from "../common/events.js";
+import {
+  agentRunKafkaConfig,
+  publishAgentRunKafkaMessage,
+  sha256Hex,
+  stableKafkaJson,
+  type AgentRunKafkaConfig,
+} from "../common/kafka-events.js";
+import { redactJson } from "../common/redaction.js";
+import type { BackendEvent, FailureKind, JsonRecord, TerminalStatus } from "../common/types.js";
+
+export const STDIO_RECONSTRUCTION_KIND = "stdio-derived partial reconstruction";
+export const STDIO_RECONSTRUCTION_SCHEMA = "agentrun.event.reconstruction.v1";
+export const STDIO_RECONSTRUCTION_EVENT_TYPE = "agentrun.event.reconstructed";
+export const DEFAULT_STDIO_RECONSTRUCTION_TOPIC = "agentrun.event.debug.v1";
+export const DEFAULT_STDIO_RECONSTRUCTION_GROUP_PREFIX = "agentrun-v02-stdio-regenerate-debug";
+
+export interface StdioKafkaFrameRecord {
+  topic: string;
+  partition: number | null;
+  offset: string | null;
+  key: string | null;
+  timestamp: string | null;
+  value: JsonRecord;
+}
+
+export interface ReconstructedAgentRunKafkaRecord {
+  topic: string;
+  partition: null;
+  offset: null;
+  key: string;
+  value: JsonRecord;
+  headers: Record<string, string>;
+}
+
+export interface StdioFrameReadRequest {
+  config: AgentRunKafkaConfig;
+  topic: string;
+  groupId: string;
+  sessionId: string;
+  traceId: string | null;
+  limit: number;
+  timeoutMs: number;
+}
+
+export interface StdioFrameReadResult {
+  sourceKind?: "kafka" | "jsonl" | "injected";
+  records: StdioKafkaFrameRecord[];
+  scanned: number;
+  invalidJson: number;
+  timedOut: boolean;
+  elapsedMs: number;
+}
+
+export type StdioFrameReader = (request: StdioFrameReadRequest) => Promise<StdioFrameReadResult>;
+export type ReconstructionPublisher = (records: ReconstructedAgentRunKafkaRecord[], config: AgentRunKafkaConfig) => Promise<void>;
+
+export interface RegenerateAgentRunEventsInput {
+  sessionId: string;
+  traceId?: string | null;
+  brokers?: string;
+  inputTopic?: string;
+  outputTopic?: string;
+  groupId?: string;
+  limit?: number;
+  timeoutMs?: number;
+  publish?: boolean;
+  env?: NodeJS.ProcessEnv;
+}
+
+export interface RegenerateAgentRunEventsDependencies {
+  reader?: StdioFrameReader;
+  publisher?: ReconstructionPublisher;
+}
+
+export interface RegenerateAgentRunEventsResult {
+  action: string;
+  reconstructionKind: string;
+  completeLifecycle: false;
+  source: JsonRecord;
+  output: JsonRecord;
+  selectedFrameCount: number;
+  reducerGroupCount: number;
+  ignoredFrames: JsonRecord;
+  originalEventIdAvailable: false;
+  originalOutboxSeqAvailable: false;
+  nextHint: string;
+  records: ReconstructedAgentRunKafkaRecord[];
+  valuesPrinted: false;
+}
+
+interface SelectedFrame {
+  record: StdioKafkaFrameRecord;
+  context: JsonRecord;
+  notification: JsonRecord;
+  frameSeq: number | null;
+  observedAt: string;
+  sourceValueSha256: string;
+  inputIndex: number;
+}
+
+interface ReducerGroup {
+  key: string;
+  firstInputIndex: number;
+  reducer: CodexStdioEventReducer;
+  context: JsonRecord;
+  lastFrame: SelectedFrame;
+  terminal: { status: TerminalStatus; failureKind: FailureKind | null; message: string | null } | null;
+  threadId: string | null;
+  turnId: string | null;
+}
+
+interface EventCandidate {
+  group: ReducerGroup;
+  frame: SelectedFrame;
+  event: BackendEvent;
+  stage: number;
+  eventIndex: number;
+}
+
+export async function regenerateAgentRunEventsFromStdio(
+  input: RegenerateAgentRunEventsInput,
+  dependencies: RegenerateAgentRunEventsDependencies = {},
+): Promise<RegenerateAgentRunEventsResult> {
+  const sessionId = requiredText(input.sessionId, "sessionId");
+  const traceId = optionalText(input.traceId);
+  const limit = boundedInteger(input.limit ?? 5_000, 1, 20_000, "limit");
+  const timeoutMs = boundedInteger(input.timeoutMs ?? 5_000, 250, 60_000, "timeoutMs");
+  const config = agentRunKafkaConfig(input.env ?? process.env, { ...(input.brokers ? { brokers: input.brokers } : {}) });
+  if (config.brokers.length === 0) throw new AgentRunError("schema-invalid", "Kafka brokers are required", { httpStatus: 2 });
+  const inputTopic = optionalText(input.inputTopic) ?? config.codexStdioTopic;
+  const outputTopic = optionalText(input.outputTopic) ?? DEFAULT_STDIO_RECONSTRUCTION_TOPIC;
+  const productTopics = new Set(["agentrun.event.v1", config.agentrunEventTopic]);
+  const productTopicOverride = productTopics.has(outputTopic);
+  if (productTopicOverride) {
+    throw new AgentRunError(
+      "tenant-policy-denied",
+      `stdio reconstruction output is forbidden on product topic ${outputTopic}`,
+      { httpStatus: 2, details: { outputTopic, reconstructionKind: STDIO_RECONSTRUCTION_KIND, valuesPrinted: false } },
+    );
+  }
+  const groupId = optionalText(input.groupId) ?? debugGroupId(sessionId);
+  const read = await (dependencies.reader ?? readStdioKafkaFrames)({ config, topic: inputTopic, groupId, sessionId, traceId, limit, timeoutMs });
+  const reconstruction = reconstructAgentRunEventsFromStdioFrames(read.records, { sessionId, traceId, outputTopic });
+  const publish = input.publish !== false;
+  if (publish) await (dependencies.publisher ?? publishReconstructedAgentRunRecords)(reconstruction.records, { ...config, enabled: true });
+  const nextHint = `./scripts/agentrun kafka tail agentrun --topic ${shellToken(outputTopic)} --session-id ${shellToken(sessionId)} --values`;
+  return {
+    action: "kafka-regenerate-agentrun-from-stdio",
+    reconstructionKind: STDIO_RECONSTRUCTION_KIND,
+    completeLifecycle: false,
+    source: {
+      kind: read.sourceKind ?? (dependencies.reader ? "injected" : "kafka"),
+      topic: inputTopic,
+      groupId,
+      brokers: config.brokers,
+      sessionId,
+      traceId,
+      scanned: read.scanned,
+      invalidJson: read.invalidJson,
+      timedOut: read.timedOut,
+      elapsedMs: read.elapsedMs,
+      valuesPrinted: false,
+    },
+    output: {
+      topic: outputTopic,
+      publish,
+      productTopicOverride,
+      recordCount: reconstruction.records.length,
+      byEventType: reconstruction.byEventType,
+      valuesPrinted: false,
+    },
+    selectedFrameCount: reconstruction.selectedFrameCount,
+    reducerGroupCount: reconstruction.reducerGroupCount,
+    ignoredFrames: reconstruction.ignoredFrames,
+    originalEventIdAvailable: false,
+    originalOutboxSeqAvailable: false,
+    nextHint,
+    records: reconstruction.records,
+    valuesPrinted: false,
+  };
+}
+
+export function reconstructAgentRunEventsFromStdioFrames(
+  records: StdioKafkaFrameRecord[],
+  input: { sessionId: string; traceId?: string | null; outputTopic?: string },
+): {
+  records: ReconstructedAgentRunKafkaRecord[];
+  selectedFrameCount: number;
+  reducerGroupCount: number;
+  ignoredFrames: JsonRecord;
+  byEventType: JsonRecord;
+} {
+  const sessionId = requiredText(input.sessionId, "sessionId");
+  const traceId = optionalText(input.traceId);
+  const outputTopic = optionalText(input.outputTopic) ?? DEFAULT_STDIO_RECONSTRUCTION_TOPIC;
+  const ignored: Record<string, number> = {};
+  const selected: SelectedFrame[] = [];
+  records.forEach((record, inputIndex) => {
+    const reason = ignoredFrameReason(record, sessionId, traceId);
+    if (reason) {
+      ignored[reason] = (ignored[reason] ?? 0) + 1;
+      return;
+    }
+    const context = asRecord(record.value.context);
+    const stdio = asRecord(record.value.stdio);
+    const notification = asRecord(stdio.rawJson);
+    selected.push({
+      record,
+      context,
+      notification,
+      frameSeq: finiteInteger(record.value.frameSeq) ?? finiteInteger(stdio.frameSeq),
+      observedAt: observedAt(record),
+      sourceValueSha256: sha256Hex(stableKafkaJson(record.value)),
+      inputIndex,
+    });
+  });
+  if (selected.length === 0) {
+    throw new AgentRunError("schema-invalid", `no reconstructable stdout notifications found for sessionId=${sessionId}`, {
+      httpStatus: 2,
+      details: { sessionId, traceId, inputRecordCount: records.length, ignoredFrames: ignored, reconstructionKind: STDIO_RECONSTRUCTION_KIND, valuesPrinted: false },
+    });
+  }
+
+  const groups = new Map<string, ReducerGroup>();
+  const candidates: EventCandidate[] = [];
+  for (const frame of selected) {
+    const key = reducerGroupKey(frame.context);
+    let group = groups.get(key);
+    if (!group) {
+      group = {
+        key,
+        firstInputIndex: frame.inputIndex,
+        reducer: new CodexStdioEventReducer(),
+        context: frame.context,
+        lastFrame: frame,
+        terminal: null,
+        threadId: optionalText(frame.context.threadId),
+        turnId: null,
+      };
+      groups.set(key, group);
+    }
+    group.context = frame.context;
+    group.lastFrame = frame;
+    const reduced = group.reducer.reduce(frame.notification);
+    if (reduced.threadId) group.threadId = reduced.threadId;
+    if (reduced.turnId) group.turnId = reduced.turnId;
+    if (reduced.terminal) group.terminal = reduced.terminal;
+    reduced.events.forEach((event, eventIndex) => candidates.push({ group: group!, frame, event, stage: 0, eventIndex }));
+  }
+  for (const group of groups.values()) {
+    group.reducer.finalize(group.terminal).forEach((event, eventIndex) => {
+      candidates.push({ group, frame: group.lastFrame, event, stage: 1, eventIndex });
+    });
+  }
+  candidates.sort((left, right) =>
+    left.frame.inputIndex - right.frame.inputIndex
+    || left.stage - right.stage
+    || left.eventIndex - right.eventIndex
+    || left.group.firstInputIndex - right.group.firstInputIndex,
+  );
+
+  const byEventType: Record<string, number> = {};
+  const output = candidates.map((candidate, index) => {
+    byEventType[candidate.event.type] = (byEventType[candidate.event.type] ?? 0) + 1;
+    return reconstructionRecord(candidate, index + 1, sessionId, traceId, outputTopic);
+  });
+  return {
+    records: output,
+    selectedFrameCount: selected.length,
+    reducerGroupCount: groups.size,
+    ignoredFrames: ignored,
+    byEventType,
+  };
+}
+
+export async function readStdioKafkaFrames(request: StdioFrameReadRequest): Promise<StdioFrameReadResult> {
+  const startedAt = Date.now();
+  const kafka = new Kafka({ clientId: `${request.config.clientId}-stdio-regenerate`, brokers: request.config.brokers, logLevel: logLevel.NOTHING });
+  const consumer = kafka.consumer({ groupId: request.groupId, allowAutoTopicCreation: false });
+  const records: StdioKafkaFrameRecord[] = [];
+  let scanned = 0;
+  let invalidJson = 0;
+  let timedOut = false;
+  await consumer.connect();
+  try {
+    await consumer.subscribe({ topic: request.topic, fromBeginning: true });
+    await new Promise<void>((resolve, reject) => {
+      let finished = false;
+      const finish = (): void => {
+        if (finished) return;
+        finished = true;
+        resolve();
+      };
+      const timer = setTimeout(() => {
+        timedOut = true;
+        finish();
+      }, request.timeoutMs);
+      timer.unref?.();
+      consumer.run({
+        eachMessage: async ({ topic, partition, message }) => {
+          scanned += 1;
+          const valueText = message.value?.toString("utf8") ?? "";
+          const value = parseJsonRecord(valueText);
+          if (!value) {
+            invalidJson += 1;
+            return;
+          }
+          if (!frameIdentityMatches(value, request.sessionId, request.traceId)) return;
+          records.push({
+            topic,
+            partition,
+            offset: message.offset,
+            key: message.key?.toString("utf8") ?? null,
+            timestamp: message.timestamp ?? null,
+            value,
+          });
+          if (records.length >= request.limit) {
+            clearTimeout(timer);
+            finish();
+          }
+        },
+      }).catch((error) => {
+        clearTimeout(timer);
+        if (!finished) reject(error);
+      });
+    });
+  } finally {
+    await disconnectQuietly(consumer);
+  }
+  return { sourceKind: "kafka", records, scanned, invalidJson, timedOut, elapsedMs: Date.now() - startedAt };
+}
+
+export async function publishReconstructedAgentRunRecords(records: ReconstructedAgentRunKafkaRecord[], config: AgentRunKafkaConfig): Promise<void> {
+  for (const record of records) {
+    await publishAgentRunKafkaMessage(config, { topic: record.topic, key: record.key, value: record.value, headers: record.headers });
+  }
+}
+
+export function parseStdioKafkaFrameJsonl(text: string, defaultTopic = "codex-stdio.raw.v1"): StdioKafkaFrameRecord[] {
+  const records: StdioKafkaFrameRecord[] = [];
+  const lines = text.split(/\r?\n/u);
+  for (let index = 0; index < lines.length; index += 1) {
+    const line = lines[index]?.trim() ?? "";
+    if (!line) continue;
+    let parsed: JsonRecord;
+    try {
+      parsed = requireRecord(JSON.parse(line) as unknown, `JSONL line ${index + 1}`);
+    } catch (error) {
+      throw new AgentRunError("schema-invalid", `invalid stdio JSONL line ${index + 1}: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 2 });
+    }
+    const message = asRecord(parsed.message);
+    const wrappedValue = parsed.value ?? message.value;
+    const value = wrappedValue === undefined ? parsed : typeof wrappedValue === "string" ? parseJsonRecord(wrappedValue) : asNullableRecord(wrappedValue);
+    if (!value) throw new AgentRunError("schema-invalid", `stdio JSONL line ${index + 1} has no object value`, { httpStatus: 2 });
+    records.push({
+      topic: optionalText(parsed.topic) ?? defaultTopic,
+      partition: finiteInteger(parsed.partition),
+      offset: optionalText(parsed.offset ?? message.offset),
+      key: optionalText(parsed.key ?? message.key),
+      timestamp: optionalText(parsed.timestamp ?? message.timestamp),
+      value,
+    });
+  }
+  return records;
+}
+
+export function reconstructionRecordsJsonl(records: ReconstructedAgentRunKafkaRecord[]): string {
+  return records.map((record) => JSON.stringify(record)).join("\n") + (records.length > 0 ? "\n" : "");
+}
+
+function reconstructionRecord(candidate: EventCandidate, sourceSeq: number, sessionId: string, requestedTraceId: string | null, outputTopic: string): ReconstructedAgentRunKafkaRecord {
+  const { frame, group, event } = candidate;
+  const context = group.context;
+  const traceId = requestedTraceId ?? optionalText(context.traceId);
+  const runId = optionalText(context.runId);
+  const commandId = optionalText(context.commandId);
+  const identity = deriveHwlabSessionIdentity(sessionId);
+  const reconstruction: JsonRecord = {
+    kind: STDIO_RECONSTRUCTION_KIND,
+    completeLifecycle: false,
+    originalEventId: null,
+    originalOutboxSeq: null,
+    sourceTopic: frame.record.topic,
+    sourcePartition: frame.record.partition,
+    sourceOffset: frame.record.offset,
+    sourceFrameSeq: frame.frameSeq,
+    sourceValueSha256: frame.sourceValueSha256,
+    identityDerivation: identity.derivation,
+    valuesPrinted: false,
+  };
+  const payload = redactJson(normalizeRunEventPayload(event.type, {
+    ...event.payload,
+    commandId,
+    traceId,
+    sessionId,
+    hwlabSessionId: identity.hwlabSessionId,
+    reconstructed: true,
+    reconstructionKind: STDIO_RECONSTRUCTION_KIND,
+    reconstruction,
+  })) as JsonRecord;
+  const value: JsonRecord = {
+    schema: STDIO_RECONSTRUCTION_SCHEMA,
+    eventType: STDIO_RECONSTRUCTION_EVENT_TYPE,
+    source: "agentrun-cli-stdio-reconstruction",
+    eventId: null,
+    outboxSeq: null,
+    sourceSeq,
+    observedAt: frame.observedAt,
+    traceId,
+    sessionId,
+    hwlabSessionId: identity.hwlabSessionId,
+    runId,
+    commandId,
+    run: {
+      runId,
+      status: "reconstructed-partial",
+      terminalStatus: group.terminal?.status ?? null,
+      failureKind: group.terminal?.failureKind ?? null,
+      tenantId: null,
+      projectId: null,
+      providerId: null,
+      backendProfile: optionalText(context.backendProfile),
+      sessionId,
+      hwlabSessionId: identity.hwlabSessionId,
+      conversationId: optionalText(context.conversationId),
+      threadId: group.threadId ?? optionalText(context.threadId),
+      valuesPrinted: false,
+    },
+    command: commandId ? {
+      commandId,
+      runId,
+      seq: null,
+      type: null,
+      state: "reconstructed-partial",
+      payloadHash: null,
+      valuesPrinted: false,
+    } : null,
+    event: {
+      id: null,
+      runId,
+      seq: sourceSeq,
+      type: event.type,
+      payload,
+      createdAt: frame.observedAt,
+    },
+    reconstruction,
+    valuesPrinted: false,
+  };
+  return {
+    topic: outputTopic,
+    partition: null,
+    offset: null,
+    key: sessionId,
+    value,
+    headers: {
+      "x-event-schema": STDIO_RECONSTRUCTION_SCHEMA,
+      "x-event-type": STDIO_RECONSTRUCTION_EVENT_TYPE,
+      "x-reconstruction-kind": STDIO_RECONSTRUCTION_KIND,
+      "x-source-seq": String(sourceSeq),
+      "x-values-printed": "false",
+    },
+  };
+}
+
+function ignoredFrameReason(record: StdioKafkaFrameRecord, sessionId: string, traceId: string | null): string | null {
+  if (record.value.schema !== "codex-stdio.raw.v1") return "schema-not-stdio-raw-v1";
+  if (!frameIdentityMatches(record.value, sessionId, traceId)) return "identity-mismatch";
+  const stdio = asRecord(record.value.stdio);
+  if (stdio.direction !== "stdout") return "direction-not-stdout";
+  const rawJson = asNullableRecord(stdio.rawJson);
+  if (!rawJson) return "raw-json-missing";
+  if (!optionalText(rawJson.method)) return "jsonrpc-method-missing";
+  return null;
+}
+
+function frameIdentityMatches(value: JsonRecord, sessionId: string, traceId: string | null): boolean {
+  const context = asRecord(value.context);
+  if (optionalText(context.sessionId) !== sessionId) return false;
+  if (traceId && optionalText(context.traceId) !== traceId) return false;
+  return true;
+}
+
+function reducerGroupKey(context: JsonRecord): string {
+  return optionalText(context.commandId) ?? optionalText(context.runId) ?? optionalText(context.sessionId) ?? "stdio-reconstruction-unknown";
+}
+
+function observedAt(record: StdioKafkaFrameRecord): string {
+  const producedAt = optionalText(record.value.producedAt);
+  if (producedAt) return producedAt;
+  const timestamp = optionalText(record.timestamp);
+  if (timestamp && /^\d+$/u.test(timestamp)) return new Date(Number(timestamp)).toISOString();
+  return timestamp ?? new Date(0).toISOString();
+}
+
+function debugGroupId(sessionId: string): string {
+  return `${DEFAULT_STDIO_RECONSTRUCTION_GROUP_PREFIX}-${sha256Hex(sessionId).slice(0, 10)}-${Date.now().toString(36)}-${randomUUID().slice(0, 8)}`;
+}
+
+export function deriveHwlabSessionIdentity(sessionId: string): { hwlabSessionId: string | null; derivation: JsonRecord } {
+  const match = sessionId.match(/^ses_agentrun_([0-9a-fA-F]{8})_([0-9a-fA-F]{4})_([0-9a-fA-F]{4})_([0-9a-fA-F]{4})_([0-9a-fA-F]{12})$/u);
+  if (!match) {
+    return {
+      hwlabSessionId: null,
+      derivation: {
+        kind: "none",
+        sourceField: "sessionId",
+        targetField: "hwlabSessionId",
+        reversible: false,
+        reason: "source sessionId is not ses_agentrun_<uuid-with-underscores>",
+        valuesPrinted: false,
+      },
+    };
+  }
+  const uuid = match.slice(1).join("-").toLowerCase();
+  const hwlabSessionId = `ses_${uuid}`;
+  return {
+    hwlabSessionId,
+    derivation: {
+      kind: "agentrun-session-uuid-to-hwlab-session",
+      sourceField: "sessionId",
+      targetField: "hwlabSessionId",
+      reversible: true,
+      sourcePrefix: "ses_agentrun_",
+      targetPrefix: "ses_",
+      separatorTransform: "underscore-to-hyphen",
+      valuesPrinted: false,
+    },
+  };
+}
+
+function parseJsonRecord(value: string): JsonRecord | null {
+  try {
+    return asNullableRecord(JSON.parse(value) as unknown);
+  } catch {
+    return null;
+  }
+}
+
+function requireRecord(value: unknown, label: string): JsonRecord {
+  const record = asNullableRecord(value);
+  if (!record) throw new Error(`${label} must be a JSON object`);
+  return record;
+}
+
+function asNullableRecord(value: unknown): JsonRecord | null {
+  return value && typeof value === "object" && !Array.isArray(value) ? value as JsonRecord : null;
+}
+
+function asRecord(value: unknown): JsonRecord {
+  return asNullableRecord(value) ?? {};
+}
+
+function requiredText(value: unknown, label: string): string {
+  const text = optionalText(value);
+  if (!text) throw new AgentRunError("schema-invalid", `${label} is required`, { httpStatus: 2 });
+  return text;
+}
+
+function optionalText(value: unknown): string | null {
+  return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
+}
+
+function finiteInteger(value: unknown): number | null {
+  const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim().length > 0 ? Number(value) : Number.NaN;
+  return Number.isInteger(parsed) ? parsed : null;
+}
+
+function boundedInteger(value: number, min: number, max: number, label: string): number {
+  if (!Number.isFinite(value)) throw new AgentRunError("schema-invalid", `${label} must be finite`, { httpStatus: 2 });
+  const parsed = Math.floor(value);
+  if (parsed < min || parsed > max) throw new AgentRunError("schema-invalid", `${label} must be between ${min} and ${max}`, { httpStatus: 2 });
+  return parsed;
+}
+
+async function disconnectQuietly(consumer: Consumer): Promise<void> {
+  try {
+    await consumer.disconnect();
+  } catch {
+    // The read result or original Kafka error is more useful than a disconnect race.
+  }
+}
+
+function shellToken(value: string): string {
+  return /^[A-Za-z0-9._:/=-]+$/u.test(value) ? value : `'${value.replace(/'/gu, `'"'"'`)}'`;
+}

src/selftest/cases/36-stdio-event-reconstruction.ts

@@ -0,0 +1,186 @@
+import assert from "node:assert/strict";
+import { spawn } from "node:child_process";
+import { readFile } from "node:fs/promises";
+import path from "node:path";
+import { AgentRunError } from "../../common/errors.js";
+import type { JsonRecord } from "../../common/types.js";
+import {
+  DEFAULT_STDIO_RECONSTRUCTION_GROUP_PREFIX,
+  DEFAULT_STDIO_RECONSTRUCTION_TOPIC,
+  STDIO_RECONSTRUCTION_KIND,
+  STDIO_RECONSTRUCTION_SCHEMA,
+  deriveHwlabSessionIdentity,
+  parseStdioKafkaFrameJsonl,
+  reconstructAgentRunEventsFromStdioFrames,
+  regenerateAgentRunEventsFromStdio,
+  type ReconstructedAgentRunKafkaRecord,
+} from "../../debug/stdio-event-reconstruction.js";
+import type { SelfTestCase } from "../harness.js";
+
+const sessionId = "ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105";
+const traceId = "trc_mretx18t3jl4tg";
+
+const selfTest: SelfTestCase = async (context) => {
+  const fixture = path.join(context.root, "src/selftest/fixtures/stdio-reconstruction.jsonl");
+  const fixtureText = await readFile(fixture, "utf8");
+  const frames = parseStdioKafkaFrameJsonl(fixtureText);
+  assert.equal(frames.length, 10);
+
+  const reconstructed = reconstructAgentRunEventsFromStdioFrames(frames, { sessionId, traceId });
+  assert.equal(reconstructed.selectedFrameCount, 8);
+  assert.equal(reconstructed.reducerGroupCount, 1);
+  assert.equal(reconstructed.records.length, 10);
+  assert.equal((reconstructed.ignoredFrames as JsonRecord)["direction-not-stdout"], 1);
+  assert.equal((reconstructed.ignoredFrames as JsonRecord)["identity-mismatch"], 1);
+  assert.deepEqual(reconstructed.records.map((record) => record.value.sourceSeq), Array.from({ length: 10 }, (_, index) => index + 1));
+  for (const record of reconstructed.records) {
+    assert.equal(record.topic, DEFAULT_STDIO_RECONSTRUCTION_TOPIC);
+    assert.equal(record.key, sessionId);
+    assert.equal(record.value.schema, STDIO_RECONSTRUCTION_SCHEMA);
+    assert.equal(record.value.eventId, null);
+    assert.equal(record.value.outboxSeq, null);
+    assert.equal(record.value.hwlabSessionId, "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105");
+    assert.equal((record.value.event as JsonRecord).id, null);
+    assert.equal((record.value.reconstruction as JsonRecord).kind, STDIO_RECONSTRUCTION_KIND);
+    assert.equal((record.value.reconstruction as JsonRecord).completeLifecycle, false);
+    assert.equal((record.value.reconstruction as JsonRecord).originalEventId, null);
+    assert.equal((record.value.reconstruction as JsonRecord).originalOutboxSeq, null);
+    assert.equal(((record.value.reconstruction as JsonRecord).identityDerivation as JsonRecord).reversible, true);
+  }
+  const finalAssistant = reconstructed.records.find((record) => {
+    const event = record.value.event as JsonRecord;
+    const payload = event.payload as JsonRecord;
+    return event.type === "assistant_message" && payload.replyAuthority === true;
+  });
+  assert.equal(((finalAssistant?.value.event as JsonRecord).payload as JsonRecord).text, "stdio reconstruction fixture answer");
+  assert.equal((reconstructed.records.at(-1)?.value.event as JsonRecord).type, "terminal_status");
+
+  let readGroupId = "";
+  const published: ReconstructedAgentRunKafkaRecord[] = [];
+  const result = await regenerateAgentRunEventsFromStdio({
+    sessionId,
+    traceId,
+    brokers: "127.0.0.1:9092",
+  }, {
+    reader: async (request) => {
+      readGroupId = request.groupId;
+      return { records: frames, scanned: frames.length, invalidJson: 0, timedOut: false, elapsedMs: 1 };
+    },
+    publisher: async (records) => { published.push(...records); },
+  });
+  assert.match(readGroupId, new RegExp(`^${DEFAULT_STDIO_RECONSTRUCTION_GROUP_PREFIX}-`, "u"));
+  assert.equal((result.output as JsonRecord).topic, DEFAULT_STDIO_RECONSTRUCTION_TOPIC);
+  assert.equal((result.output as JsonRecord).publish, true);
+  assert.equal(published.length, reconstructed.records.length);
+  assert.match(String(result.nextHint), /kafka tail agentrun --topic agentrun\.event\.debug\.v1/u);
+
+  let productPublisherCalls = 0;
+  await assert.rejects(
+    () => regenerateAgentRunEventsFromStdio({ sessionId, traceId, brokers: "127.0.0.1:9092", outputTopic: "agentrun.event.v1" }, {
+      reader: async () => ({ records: frames, scanned: frames.length, invalidJson: 0, timedOut: false, elapsedMs: 0 }),
+      publisher: async () => { productPublisherCalls += 1; },
+    }),
+    (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.message.includes("forbidden on product topic"),
+  );
+  assert.equal(productPublisherCalls, 0);
+  assert.equal(deriveHwlabSessionIdentity(sessionId).hwlabSessionId, "ses_5ec4e141-6abc-466d-9afe-049f7c0ac105");
+  assert.equal(deriveHwlabSessionIdentity("ses_not_uuid").hwlabSessionId, null);
+
+  const outputJsonl = path.join(context.tmp, "agentrun-reconstructed-events.jsonl");
+  const jsonCli = await runCli(context.root, [
+    "kafka", "regenerate", "agentrun",
+    "--from", "jsonl",
+    "--jsonl-file", fixture,
+    "--session-id", sessionId,
+    "--trace-id", traceId,
+    "--no-publish",
+    "--output-jsonl", outputJsonl,
+    "--json",
+  ]);
+  assert.equal(jsonCli.code, 0, jsonCli.stderr);
+  const jsonEnvelope = JSON.parse(jsonCli.stdout) as JsonRecord;
+  assert.equal(jsonEnvelope.ok, true);
+  const jsonData = jsonEnvelope.data as JsonRecord;
+  assert.equal(jsonData.reconstructionKind, STDIO_RECONSTRUCTION_KIND);
+  assert.equal((jsonData.source as JsonRecord).kind, "jsonl");
+  assert.equal(jsonData.outputJsonlRecords, reconstructed.records.length);
+  assert.equal(String(jsonData.nextHint).includes("kafka regenerate agentrun"), true);
+  assert.equal((await readFile(outputJsonl, "utf8")).trim().split(/\r?\n/u).length, reconstructed.records.length);
+
+  const compactCli = await runCli(context.root, [
+    "kafka", "regenerate", "agentrun",
+    "--from", "jsonl",
+    "--jsonl-file", fixture,
+    "--session-id", sessionId,
+    "--trace-id", traceId,
+    "--no-publish",
+  ]);
+  assert.equal(compactCli.code, 0, compactCli.stderr);
+  assert.match(compactCli.stdout, /stdio 部分重建/u);
+  assert.match(compactCli.stdout, /originalEventId\/outboxSeq 不可恢复/u);
+  assert.match(compactCli.stdout, /下一步: \.\/scripts\/agentrun kafka regenerate agentrun/u);
+
+  const removedOverrideCli = await runCli(context.root, [
+    "kafka", "regenerate", "agentrun",
+    "--from", "jsonl",
+    "--jsonl-file", fixture,
+    "--session-id", sessionId,
+    "--output-topic", "agentrun.event.v1",
+    "--allow-product-topic",
+    "--no-publish",
+  ]);
+  assert.equal(removedOverrideCli.code, 2);
+  assert.match(removedOverrideCli.stdout, /^stdio 部分重建失败/u);
+  assert.match(removedOverrideCli.stdout, /--allow-product-topic is not supported/u);
+  assert.equal(removedOverrideCli.stdout.startsWith("{"), false);
+
+  const jsonGuardCli = await runCli(context.root, [
+    "kafka", "regenerate", "agentrun",
+    "--from", "jsonl",
+    "--jsonl-file", fixture,
+    "--session-id", sessionId,
+    "--output-topic", "agentrun.event.v1",
+    "--no-publish",
+    "--json",
+  ]);
+  assert.equal(jsonGuardCli.code, 2);
+  const guardEnvelope = JSON.parse(jsonGuardCli.stdout) as JsonRecord;
+  assert.equal(guardEnvelope.ok, false);
+  assert.equal(guardEnvelope.failureKind, "tenant-policy-denied");
+
+  const compactHelp = await runCli(context.root, ["kafka", "regenerate", "agentrun", "--help"]);
+  assert.equal(compactHelp.code, 0);
+  assert.match(compactHelp.stdout, /^按 stdio sessionId/u);
+  assert.equal(compactHelp.stdout.startsWith("{"), false);
+
+  return {
+    name: "stdio-event-reconstruction",
+    tests: [
+      "production-reducer-reuse",
+      "jsonl-offline-source",
+      "injected-reader-publisher",
+      "partial-reconstruction-provenance",
+      "debug-topic-group-defaults",
+      "product-topic-permanent-guard",
+      "compact-and-json-error-contract",
+      "reversible-hwlab-session-identity",
+      "compact-and-json-cli",
+      "semantic-next-hint",
+    ],
+  };
+};
+
+async function runCli(root: string, args: string[]): Promise<{ code: number; stdout: string; stderr: string }> {
+  const proc = spawn(process.execPath, [path.join(root, "scripts/agentrun-cli.ts"), ...args], { stdio: ["ignore", "pipe", "pipe"] });
+  let stdout = "";
+  let stderr = "";
+  proc.stdout.on("data", (chunk) => { stdout += String(chunk); });
+  proc.stderr.on("data", (chunk) => { stderr += String(chunk); });
+  const code = await new Promise<number>((resolve, reject) => {
+    proc.once("error", reject);
+    proc.once("close", (value) => resolve(value ?? 1));
+  });
+  return { code, stdout: stdout.trim(), stderr: stderr.trim() };
+}
+
+export default selfTest;

src/selftest/fixtures/stdio-reconstruction.jsonl

@@ -0,0 +1,10 @@
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"1","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600001","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdin","source":"agentrun-runner","frameSeq":1,"producedAt":"2026-07-10T10:00:00.001Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":1,"direction":"stdin","method":"turn/start","phase":"request","rpcId":3,"rawJson":{"jsonrpc":"2.0","id":3,"method":"turn/start","params":{}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"2","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600002","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":2,"producedAt":"2026-07-10T10:00:00.002Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":2,"direction":"stdout","method":"thread/started","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"thread/started","params":{"thread":{"id":"thr_stdio_reconstruction_fixture"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"3","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600003","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":3,"producedAt":"2026-07-10T10:00:00.003Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":3,"direction":"stdout","method":"turn/started","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"turn/started","params":{"turn":{"id":"turn_stdio_reconstruction_fixture"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"4","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600004","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":4,"producedAt":"2026-07-10T10:00:00.004Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":4,"direction":"stdout","method":"item/started","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"item/started","params":{"item":{"id":"item_stdio_reconstruction_tool","type":"commandExecution","command":"printf fixture","status":"inProgress"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"5","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600005","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":5,"producedAt":"2026-07-10T10:00:00.005Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":5,"direction":"stdout","method":"item/commandExecution/outputDelta","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"item/commandExecution/outputDelta","params":{"itemId":"item_stdio_reconstruction_tool","delta":"fixture-output\n"}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"6","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600006","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":6,"producedAt":"2026-07-10T10:00:00.006Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":6,"direction":"stdout","method":"item/completed","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"item/completed","params":{"item":{"id":"item_stdio_reconstruction_tool","type":"commandExecution","command":"printf fixture","status":"completed","exitCode":0,"durationMs":12,"aggregatedOutput":"fixture-output\n"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"7","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600007","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":7,"producedAt":"2026-07-10T10:00:00.007Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":7,"direction":"stdout","method":"item/agentMessage/delta","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"item/agentMessage/delta","params":{"itemId":"item_stdio_reconstruction_answer","delta":"stdio reconstruction fixture answer"}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"8","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600008","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":8,"producedAt":"2026-07-10T10:00:00.008Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":8,"direction":"stdout","method":"item/completed","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"item/completed","params":{"item":{"id":"item_stdio_reconstruction_answer","type":"agentMessage","text":"stdio reconstruction fixture answer"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"9","key":"cmd_stdio_reconstruction_fixture","timestamp":"1783677600009","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":9,"producedAt":"2026-07-10T10:00:00.009Z","context":{"runId":"run_stdio_reconstruction_fixture","commandId":"cmd_stdio_reconstruction_fixture","sessionId":"ses_agentrun_5ec4e141_6abc_466d_9afe_049f7c0ac105","traceId":"trc_mretx18t3jl4tg","conversationId":"conv_stdio_reconstruction_fixture","threadId":null,"backendProfile":"codex","valuesPrinted":false},"stdio":{"frameSeq":9,"direction":"stdout","method":"turn/completed","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"turn/completed","params":{"turn":{"id":"turn_stdio_reconstruction_fixture","status":"completed"}}},"valuesPrinted":true},"valuesPrinted":true}}
+{"topic":"codex-stdio.raw.v1","partition":0,"offset":"10","key":"cmd_other_session","timestamp":"1783677600010","value":{"schema":"codex-stdio.raw.v1","eventType":"codex.stdio.stdout","source":"agentrun-runner","frameSeq":1,"producedAt":"2026-07-10T10:00:00.010Z","context":{"runId":"run_other_session","commandId":"cmd_other_session","sessionId":"ses_other_session","traceId":"trc_other_session","valuesPrinted":false},"stdio":{"frameSeq":1,"direction":"stdout","method":"turn/completed","phase":"notification","rpcId":null,"rawJson":{"jsonrpc":"2.0","method":"turn/completed","params":{"turn":{"id":"turn_other_session","status":"completed"}}},"valuesPrinted":true},"valuesPrinted":true}}