feat: publish AgentRun Kafka event streams

docs/reference/spec-v01-agentrun-mgr.md

@@ -11,6 +11,7 @@
 - 校验并持久化 `tenantId`、`projectId`、`workspaceRef`、`providerId`、`backendProfile`、`executionPolicy` 和 `traceSink`。
 - 执行最小 tenant policy boundary：只做 schema、allowlist、idempotency、secret scope 和 executionPolicy 范围检查；不内建 UniDesk/HWLAB 的业务授权。
 - 使用 Postgres 保存 runs、commands、events、runners、backends、leases 和 migration ledger。
+- 以 Kafka topic `agentrun.event.v1` 发布 manager durable event 流，作为 HWLAB 转换前的 AgentRun event 权威源；Postgres 仍是 manager 在线读写的 durable store。
 - 输出结构化 health/readiness、failureKind、redacted SecretRef 和 trace correlation。
 - 可观测性只能定位和验证状态，不能替代缺失能力实现；如果 HWLAB canary 需要的 final reply、command result、runner 多 turn、SessionRef 或 cancel 能力缺失，manager 必须补 durable API/状态机，而不是只补 trace 文案。
 
@@ -23,6 +24,7 @@
 - Migration 必须在 readiness 前完成或显式 fail fast，不能以空 schema 静默启动。
 - Provider credential、Codex auth/config、Postgres DSN 明文不进数据库、event、trace、日志或 CLI 输出。
 - Manager 可以保存 SecretRef 和 credential source reference，但不得读取 provider Secret 值后存库。
+- Kafka 发布在本阶段是 produce-only 可见性链路：manager 将 `createRun`、`createCommand`、`appendEvent`、`finishCommand` 和 `finishRun` 等 durable store 写入镜像到 `agentrun.event.v1`，schema 为 `agentrun.event.v1`。不得让 Kafka consumer 反向驱动 manager 状态，也不得把 HWLAB event 或 Codex stdio raw 混入该 topic。
 
 ## API 接口说明
 

docs/reference/spec-v01-backend-codex.md

@@ -20,6 +20,12 @@ codex app-server --listen stdio://
 
 Adapter 通过 stdin 写入换行分隔 JSON-RPC 请求，通过 stdout 逐行读取 JSON-RPC response 和 notification，stderr 只作为有界诊断日志。最小请求序列是 `initialize`、`thread/start` 或 `thread/resume`、`turn/start`；response 中必须提取 thread/turn identity，notification 和后续输出必须归一化为 `backend_status`、`assistant_message`、`tool_call`、`command_output`、`error` 和 `terminal_status` events。运行中 steer 使用同一 app-server 进程的 `turn/steer` JSON-RPC 方法，参数为 `threadId`、`expectedTurnId` 和文本 `input` 数组；取消/中断使用 `turn/interrupt`，参数为 `threadId` 和 `turnId`。已有 `SessionRef.threadId` 时只能执行 Codex stdio 原生 `thread/resume` 后接 `turn/start`；当 `thread/resume` 返回 `no rollout found for thread id` 或任何其他协议错误时，adapter 必须输出 `thread-resume-failed` 并终止当前 turn。adapter 不得启动替代 `thread/start`、拼接历史 prompt、回写新 threadId 或用其他上下文模拟继续会话。
 
+### Codex stdio Kafka 原始流
+
+AgentRun v0.2 的 Codex stdio 原始流权威源是 Kafka topic `codex-stdio.raw.v1`。该流由 runner 内的 `CodexStdioClient` 在真实 stdio 边界发布，覆盖写入 child stdin 的 JSON-RPC line、从 child stdout 读取的每一行、stderr chunk 和 process close lifecycle。消息 schema 固定为 `codex-stdio.raw.v1`，携带 `runId`、`commandId`、`sessionId`、`threadId`、`runnerJobId`、`podName`、direction、method、phase、raw bytes/hash 和原始文本/JSON。Kafka value 是分析用的裸 stdio 事实，生产前不得做 redaction；CLI、日志和 UI 默认只显示 hash/bytes，只有显式 raw/value 查询才展示 value。
+
+启用条件来自 runner env：`AGENTRUN_KAFKA_ENABLED=true` 或 `AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED=true`，并声明 `AGENTRUN_KAFKA_BOOTSTRAP_SERVERS`、`AGENTRUN_KAFKA_STDIO_TOPIC` 和 `AGENTRUN_KAFKA_STDIO_CLIENT_ID`。这些 env 由 manager Deployment 通过 runner Job manifest 透传，配置 authority 在 UniDesk `config/agentrun.yaml`，不由 HWLAB workbench 或 AgentRun service repo 的 legacy `deploy.json` 决定。
+
 若 run 的 `ResourceBundleRef` 包含 `promptRefs` 或 gitbundle `.agents/skills`，Codex adapter 只能消费 runner 已装配好的有界 `initialPrompt`、skill summary 和 skill registry path。对新 thread，adapter 在首个 `turn/start` 中把 `initialPrompt` 和 skill facts 放在用户 message 之前；对 `thread/resume`，adapter 不重复注入 `initialPrompt`，只发送当前 command 的用户 message。当前 Codex app-server 若只有 `input: [{ type: "text", text }]`，允许使用结构化文本前缀承载 initial prompt；若后续 app-server 支持 developer/runtime instruction item，优先映射到该标准 item。无论哪种 wire shape，events 只记录 prompt/skill 的 path/hash/bytes/injected 状态，不输出全文。
 
 不得把以下路径作为 `v0.1` Codex stdio backend 的正式实现或综合联调通过证据：直接 Responses HTTP 代理、OpenAI SDK wrapper、`codex exec` 一次性命令输出、fake provider、固定文本回复、只读 shortcut 或本地 shell 模拟。裸 HTTP 或 `codex exec --json` 可以作为 provider/upstream 诊断，但最终通过必须来自 app-server stdio turn。

docs/reference/spec-v01-cli.md

@@ -39,6 +39,8 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 ./scripts/agentrun runs events <runId> --after-seq <n> --limit <n> [--summary|--tail-summary] [--tail <n>] [--summary-chars <n>] [--format json|tsv]
 ./scripts/agentrun runs result <runId> [--command-id <commandId>]
 ./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 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>
@@ -100,6 +102,7 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交
 - `events --summary` 返回低噪声 JSON summary，单条至少包含 `seq`、`type`、`method`、`status`、`command`、`text`、`exitCode`、`durationMs`、`outputTruncated`、`outputBytes`、`outputSummary` 和 `summary`；summary 文本必须压缩换行并继续沿用 redaction，不能泄漏 Secret/env value。
 - `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。
 - `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。

scripts/src/cli.ts

@@ -16,6 +16,7 @@ import type { RunnerOnceOptions } from "../../src/runner/run-once.js";
 import { backendProfileSpec, isBackendProfile } from "../../src/common/backend-profiles.js";
 import { outputBytesFromPayload, outputTruncatedFromPayload } from "../../src/common/output.js";
 import { redactJson, redactText } from "../../src/common/redaction.js";
+import { tailAgentRunKafka } from "../../src/common/kafka-events.js";
 
 interface ParsedArgs {
   positional: string[];
@@ -52,6 +53,7 @@ async function dispatch(args: ParsedArgs): Promise<CliResult> {
   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 === "server" && command === "start") return startServer(args);
   if (group === "server" && command === "status") return serverStatus(args);
   if (group === "server" && command === "logs") return serverLogs(args);
@@ -159,6 +161,29 @@ async function listRunnerJobs(args: ParsedArgs): Promise<JsonValue> {
   return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/runner-jobs${commandId ? `?commandId=${encodeURIComponent(commandId)}` : ""}`);
 }
 
+async function kafkaTail(args: ParsedArgs): Promise<JsonValue> {
+  const stream = optionalFlag(args, "stream") ?? args.positional[2] ?? null;
+  const topic = optionalFlag(args, "topic");
+  const limit = optionalNumberFlag(args, "limit");
+  const timeoutMs = optionalNumberFlag(args, "timeout-ms");
+  return await tailAgentRunKafka({
+    brokers: optionalFlag(args, "brokers") ?? undefined,
+    stream: stream ?? undefined,
+    topic: topic ?? undefined,
+    groupId: optionalFlag(args, "group-id") ?? undefined,
+    limit: limit ?? undefined,
+    timeoutMs: timeoutMs ?? undefined,
+    fromBeginning: args.flags.get("latest") === true ? false : true,
+    values: args.flags.get("values") === true || args.flags.get("raw-values") === true,
+    filter: {
+      traceId: optionalFlag(args, "trace-id"),
+      sessionId: optionalFlag(args, "session-id"),
+      runId: optionalFlag(args, "run-id"),
+      commandId: optionalFlag(args, "command-id"),
+    },
+  }) as JsonValue;
+}
+
 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;
@@ -2039,6 +2064,14 @@ function optionalFlag(args: ParsedArgs, name: string): string | null {
   return typeof value === "string" && value.length > 0 ? value : null;
 }
 
+function optionalNumberFlag(args: ParsedArgs, name: string): number | null {
+  const value = optionalFlag(args, name);
+  if (value === null) return null;
+  const parsed = Number(value);
+  if (!Number.isFinite(parsed)) throw new AgentRunError("schema-invalid", `--${name} must be a finite number`, { httpStatus: 2 });
+  return parsed;
+}
+
 function cancelBody(args: ParsedArgs): JsonRecord {
   const reason = optionalFlag(args, "reason");
   return reason ? { reason } : {};
@@ -2051,6 +2084,8 @@ function help(args: ParsedArgs, group?: string): JsonRecord {
       "runs events <runId> --after-seq <n> --limit <n> [--summary|--tail-summary] [--tail <n>] [--summary-chars <n>] [--format json|tsv]",
       "runs result <runId> [--command-id <commandId>]",
       "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>]",
       "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

@@ -9,6 +9,7 @@ import { redactJson, redactText } from "../common/redaction.js";
 import { backendProfileSpec } from "../common/backend-profiles.js";
 import { boundedTextSummary, commandOutputPayload } from "../common/output.js";
 import { emitAgentRunOtelSpan } from "../common/otel-trace.js";
+import { agentRunKafkaConfig, publishAgentRunKafkaMessageBestEffort, rawFrameValue, safeText, sha256Hex, type AgentRunKafkaConfig } from "../common/kafka-events.js";
 
 const codexProtocol = "codex-app-server-jsonrpc-stdio";
 const defaultCodexArgs = ["app-server", "--listen", "stdio://"];
@@ -81,6 +82,28 @@ interface CodexLifecycleOtelContext {
   attributes: JsonRecord;
 }
 
+interface CodexStdioKafkaContext extends JsonRecord {
+  runId: string | null;
+  commandId: string | null;
+  attemptId: string | null;
+  runnerId: string | null;
+  runnerJobId: string | null;
+  jobName: string | null;
+  podName: string | null;
+  sessionId: string | null;
+  conversationId: string | null;
+  threadId: string | null;
+  backendProfile: string | null;
+  sourceCommit: string | null;
+  valuesPrinted: false;
+}
+
+interface CodexStdioKafkaPublisher {
+  config: AgentRunKafkaConfig;
+  context: CodexStdioKafkaContext;
+  source: string;
+}
+
 export interface CodexActiveTurnControl {
   threadId: string;
   turnId: string;
@@ -149,6 +172,7 @@ class CodexStdioFailure extends Error {
 
 export class CodexStdioClient {
   private readonly child: ChildProcessWithoutNullStreams;
+  private stdioKafka: CodexStdioKafkaPublisher | null;
   private readonly pending = new Map<number, PendingRequest>();
   private stderrTailBuffer = Buffer.alloc(0);
   private stderrBytes = 0;
@@ -158,8 +182,9 @@ export class CodexStdioClient {
   readonly closedPromise: Promise<CodexStdioCloseInfo>;
   private closeResolve!: (value: CodexStdioCloseInfo) => void;
 
-  constructor(options: { command?: string; args?: string[]; cwd: string; env?: NodeJS.ProcessEnv; onNotification: (message: JsonRecord) => void }) {
+  constructor(options: { command?: string; args?: string[]; cwd: string; env?: NodeJS.ProcessEnv; onNotification: (message: JsonRecord) => void; stdioKafka?: CodexStdioKafkaPublisher | null }) {
     this.closedPromise = new Promise((resolve) => { this.closeResolve = resolve; });
+    this.stdioKafka = options.stdioKafka ?? null;
     const command = options.command ?? "codex";
     const args = options.args ?? defaultCodexArgs;
     try {
@@ -172,7 +197,14 @@ export class CodexStdioClient {
     } catch (error) {
       throw spawnFailure(command, error);
     }
-    this.child.stderr.on("data", (chunk: Buffer) => this.appendStderr(chunk));
+    this.child.stderr.on("data", (chunk: Buffer) => {
+      this.appendStderr(chunk);
+      publishCodexStdioFrame(this.stdioKafka, "stderr", {
+        rawText: chunk.toString("utf8"),
+        method: "process.stderr",
+        phase: "stderr:data",
+      });
+    });
     const rl = readline.createInterface({ input: this.child.stdout, crlfDelay: Infinity });
     void this.readLines(rl, options.onNotification);
     this.child.on("close", (code, signal) => this.handleClose(code, signal));
@@ -183,6 +215,10 @@ export class CodexStdioClient {
     return this.closed;
   }
 
+  setStdioKafka(publisher: CodexStdioKafkaPublisher | null): void {
+    this.stdioKafka = publisher;
+  }
+
   request(method: string, params: JsonRecord, timeoutMs = requestTimeoutCapMs): Promise<unknown> {
     if (this.closed) return Promise.reject(this.closeFailure ?? new CodexStdioFailure("backend-failed", "codex app-server is closed", `request:${method}`));
     const id = this.nextId++;
@@ -193,7 +229,15 @@ export class CodexStdioClient {
         this.rejectRequest(id, new CodexStdioFailure("backend-timeout", `Codex stdio request ${method} timed out after ${effectiveTimeoutMs}ms`, `request:${method}`, { method, timeoutMs: effectiveTimeoutMs }));
       }, effectiveTimeoutMs);
       this.pending.set(id, { method, timer, resolve, reject });
-      this.child.stdin.write(`${JSON.stringify(message)}\n`, "utf8", (error: Error | null | undefined) => {
+      const rawLine = `${JSON.stringify(message)}\n`;
+      publishCodexStdioFrame(this.stdioKafka, "stdin", {
+        rawText: rawLine,
+        rawJson: message,
+        method,
+        rpcId: id,
+        phase: "client.request",
+      });
+      this.child.stdin.write(rawLine, "utf8", (error: Error | null | undefined) => {
         if (!error) return;
         this.rejectRequest(id, new CodexStdioFailure("backend-failed", `failed to write Codex stdio request ${method}: ${error.message}`, `request:${method}`, { method }));
       });
@@ -202,7 +246,15 @@ export class CodexStdioClient {
 
   notify(method: string, params: JsonRecord = {}): void {
     if (this.closed) return;
-    this.child.stdin.write(`${JSON.stringify({ method, params })}\n`, "utf8", () => undefined);
+    const message = { method, params };
+    const rawLine = `${JSON.stringify(message)}\n`;
+    publishCodexStdioFrame(this.stdioKafka, "stdin", {
+      rawText: rawLine,
+      rawJson: message,
+      method,
+      phase: "client.notify",
+    });
+    this.child.stdin.write(rawLine, "utf8", () => undefined);
   }
 
   stop(): void {
@@ -241,9 +293,21 @@ export class CodexStdioClient {
         try {
           message = JSON.parse(trimmed) as JsonRecord;
         } catch {
+          publishCodexStdioFrame(this.stdioKafka, "stdout", {
+            rawText: line,
+            method: "stdout.parse-failed",
+            phase: "stdout:parse",
+          });
           this.handleProtocolFailure(new CodexStdioFailure("backend-json-parse-error", "codex app-server emitted invalid JSON on stdout", "stdout:parse", { linePreview: redactText(trimmed.slice(0, 800)), lineChars: trimmed.length }));
           break;
         }
+        publishCodexStdioFrame(this.stdioKafka, "stdout", {
+          rawText: line,
+          rawJson: message,
+          method: typeof message.method === "string" ? message.method : "response",
+          rpcId: typeof message.id === "number" ? message.id : null,
+          phase: "stdout:message",
+        });
         this.handleMessage(message, onNotification);
       }
     } catch (error) {
@@ -287,10 +351,28 @@ export class CodexStdioClient {
 
   private handleServerRequest(id: number, method: string): void {
     if (method === "item/commandExecution/requestApproval" || method === "item/fileChange/requestApproval") {
-      this.child.stdin.write(`${JSON.stringify({ id, result: { decision: "decline" } })}\n`);
+      const message = { id, result: { decision: "decline" } };
+      const rawLine = `${JSON.stringify(message)}\n`;
+      publishCodexStdioFrame(this.stdioKafka, "stdin", {
+        rawText: rawLine,
+        rawJson: message,
+        method,
+        rpcId: id,
+        phase: "client.response",
+      });
+      this.child.stdin.write(rawLine);
       return;
     }
-    this.child.stdin.write(`${JSON.stringify({ id, error: { code: -32601, message: `Unsupported client-side request: ${method}` } })}\n`);
+    const message = { id, error: { code: -32601, message: `Unsupported client-side request: ${method}` } };
+    const rawLine = `${JSON.stringify(message)}\n`;
+    publishCodexStdioFrame(this.stdioKafka, "stdin", {
+      rawText: rawLine,
+      rawJson: message,
+      method,
+      rpcId: id,
+      phase: "client.response",
+    });
+    this.child.stdin.write(rawLine);
   }
 
   private rejectRequest(id: number, error: CodexStdioFailure): void {
@@ -330,6 +412,11 @@ export class CodexStdioClient {
       failureKind: this.closeFailure?.failureKind ?? null,
       message: this.closeFailure?.message ?? null,
     };
+    publishCodexStdioFrame(this.stdioKafka, "lifecycle", {
+      rawJson: closeInfo,
+      method: "process.close",
+      phase: "process:close",
+    });
     this.rejectAll(this.closeFailure ?? new CodexStdioFailure("backend-failed", `codex app-server closed code=${code} signal=${signal}`, "process:close", closeInfo));
     this.closeResolve(closeInfo);
   }
@@ -351,6 +438,71 @@ export async function runCodexStdioTurn(options: CodexStdioTurnOptions): Promise
   return { ...result, events: [...result.events, ...closeEvents] };
 }
 
+function codexStdioKafkaPublisher(options: CodexStdioTurnOptions, env: NodeJS.ProcessEnv): CodexStdioKafkaPublisher | null {
+  if (!truthy(env.AGENTRUN_KAFKA_ENABLED) && !truthy(env.AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED)) return null;
+  const config = agentRunKafkaConfig(env, { clientId: env.AGENTRUN_KAFKA_STDIO_CLIENT_ID ?? env.AGENTRUN_KAFKA_CLIENT_ID ?? "agentrun-v02-runner", stream: "stdio" });
+  if (config.brokers.length === 0) return null;
+  return {
+    config,
+    source: config.clientId,
+    context: codexStdioKafkaContext(options),
+  };
+}
+
+function truthy(value: string | undefined): boolean {
+  return new Set(["1", "true", "yes", "on"]).has(String(value ?? "").trim().toLowerCase());
+}
+
+function codexStdioKafkaContext(options: CodexStdioTurnOptions): CodexStdioKafkaContext {
+  const run = options.otelContext?.run;
+  const command = options.otelContext?.command;
+  return {
+    runId: run?.id ?? null,
+    commandId: command?.id ?? null,
+    attemptId: safeText(options.otelContext?.attemptId) ?? null,
+    runnerId: safeText(options.otelContext?.runnerId) ?? null,
+    runnerJobId: safeText(options.otelContext?.runnerJobId) ?? null,
+    jobName: safeText(options.otelContext?.jobName) ?? null,
+    podName: safeText(options.otelContext?.podName) ?? null,
+    sessionId: run?.sessionRef?.sessionId ?? null,
+    conversationId: run?.sessionRef?.conversationId ?? null,
+    threadId: run?.sessionRef?.threadId ?? options.threadId ?? null,
+    backendProfile: run?.backendProfile ?? options.backendProfile ?? null,
+    sourceCommit: safeText(options.otelContext?.sourceCommit) ?? null,
+    valuesPrinted: false,
+  };
+}
+
+function publishCodexStdioFrame(publisher: CodexStdioKafkaPublisher | null | undefined, direction: "stdin" | "stdout" | "stderr" | "lifecycle", input: { rawText?: string; rawJson?: JsonRecord; method: string; phase: string; rpcId?: number | null }): void {
+  if (!publisher) return;
+  const rawText = input.rawText ?? (input.rawJson ? JSON.stringify(input.rawJson) : "");
+  const context = publisher.context;
+  const keySource = `${context.runId ?? "run-unknown"}|${context.commandId ?? "command-unknown"}|${direction}|${input.method}|${Date.now()}|${sha256Hex(rawText).slice(0, 12)}`;
+  publishAgentRunKafkaMessageBestEffort(publisher.config, {
+    topic: publisher.config.codexStdioTopic,
+    key: keySource,
+    value: rawFrameValue({
+      schema: "codex-stdio.raw.v1",
+      eventType: `codex.stdio.${direction}`,
+      source: publisher.source,
+      context,
+      stdio: {
+        direction,
+        method: input.method,
+        phase: input.phase,
+        rpcId: input.rpcId ?? null,
+      },
+      rawText,
+      rawJson: input.rawJson,
+    }),
+    headers: {
+      "x-event-type": `codex.stdio.${direction}`,
+      "x-stdio-direction": direction,
+      "x-values-printed": "true",
+    },
+  }, { eventType: `codex.stdio.${direction}`, direction, method: input.method });
+}
+
 export class CodexStdioBackendSession {
   private client: CodexStdioClient | null = null;
   private clientKey: string | null = null;
@@ -376,6 +528,7 @@ export class CodexStdioBackendSession {
     const key = codexClientKey(options, env);
     this.lifecycleOtel = codexLifecycleOtelContext(options, env);
     if (this.client && !this.client.isClosed && this.clientKey === key) {
+      this.client.setStdioKafka(codexStdioKafkaPublisher(options, env));
       emitEvent({ type: "backend_status", payload: { phase: "codex-app-server:reused", ...backendMetadata(options), protocol: codexProtocol } });
       emitCodexOtelSpan("codex_app_server.reused", options, env, { command: options.command ?? "codex", argsFingerprint: shortHash(JSON.stringify(options.args ?? defaultCodexArgs)) });
       return this.client;
@@ -398,6 +551,7 @@ export class CodexStdioBackendSession {
       cwd: options.cwd,
       env,
       onNotification: (message) => this.onNotification(message),
+      stdioKafka: codexStdioKafkaPublisher(options, env),
     };
     if (options.command) clientOptions.command = options.command;
     if (options.args) clientOptions.args = options.args;

src/common/kafka-events.ts

@@ -0,0 +1,325 @@
+import { Kafka, logLevel, type Consumer, type EachMessagePayload, type Producer } from "kafkajs";
+import { createHash } from "node:crypto";
+import type { JsonRecord, JsonValue } from "./types.js";
+import { redactJson, redactText } from "./redaction.js";
+
+const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
+const DEFAULT_BROKERS = "platform-infra-kafka-kafka-bootstrap.platform-infra.svc.cluster.local:9092";
+const DEFAULT_AGENTS_TOPIC = "agentrun.event.v1";
+const DEFAULT_STDIO_TOPIC = "codex-stdio.raw.v1";
+const MAX_VALUE_PREVIEW_CHARS = 4_000;
+
+let producerState: { key: string; producer: Producer; connecting: Promise<Producer> | null } | null = null;
+
+export interface AgentRunKafkaConfig {
+  brokers: string[];
+  clientId: string;
+  agentrunEventTopic: string;
+  codexStdioTopic: string;
+  enabled: boolean;
+  valuesPrinted: false;
+}
+
+export interface AgentRunKafkaMessage {
+  topic: string;
+  key: string;
+  value: JsonRecord;
+  headers?: Record<string, string>;
+}
+
+export interface KafkaTailOptions {
+  env?: NodeJS.ProcessEnv;
+  brokers?: string;
+  topic?: string;
+  stream?: string;
+  groupId?: string;
+  limit?: number;
+  timeoutMs?: number;
+  fromBeginning?: boolean;
+  values?: boolean;
+  filter?: {
+    traceId?: string | null;
+    sessionId?: string | null;
+    runId?: string | null;
+    commandId?: string | null;
+  };
+}
+
+export function agentRunKafkaConfig(env: NodeJS.ProcessEnv = process.env, overrides: { brokers?: string; clientId?: string; topic?: string; stream?: string } = {}): AgentRunKafkaConfig {
+  const brokers = csv(overrides.brokers ?? env.AGENTRUN_KAFKA_BOOTSTRAP_SERVERS ?? DEFAULT_BROKERS);
+  const clientId = safeText(overrides.clientId ?? env.AGENTRUN_KAFKA_CLIENT_ID) ?? "agentrun-v02";
+  const agentrunEventTopic = safeText(env.AGENTRUN_KAFKA_EVENT_TOPIC) ?? DEFAULT_AGENTS_TOPIC;
+  const codexStdioTopic = safeText(env.AGENTRUN_KAFKA_STDIO_TOPIC) ?? DEFAULT_STDIO_TOPIC;
+  const topicOverride = safeText(overrides.topic);
+  return {
+    brokers,
+    clientId,
+    agentrunEventTopic: topicOverride && overrides.stream !== "stdio" ? topicOverride : agentrunEventTopic,
+    codexStdioTopic: topicOverride && overrides.stream === "stdio" ? topicOverride : codexStdioTopic,
+    enabled: truthy(env.AGENTRUN_KAFKA_ENABLED) || truthy(env.AGENTRUN_KAFKA_SHADOW_PRODUCE_ENABLED) || truthy(env.AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED),
+    valuesPrinted: false,
+  };
+}
+
+export function kafkaTopicForStream(config: AgentRunKafkaConfig, streamOrTopic: string | null | undefined): string {
+  const value = safeText(streamOrTopic);
+  if (!value || value === "agentrun" || value === "event" || value === "events") return config.agentrunEventTopic;
+  if (value === "stdio" || value === "codex-stdio" || value === "raw") return config.codexStdioTopic;
+  return value;
+}
+
+export function kafkaStreamForTopic(config: AgentRunKafkaConfig, topic: string): string {
+  if (topic === config.codexStdioTopic) return "stdio";
+  if (topic === config.agentrunEventTopic) return "agentrun";
+  return "custom";
+}
+
+export async function publishAgentRunKafkaMessage(config: AgentRunKafkaConfig, message: AgentRunKafkaMessage): Promise<void> {
+  if (!config.enabled) return;
+  if (config.brokers.length === 0) throw new Error("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS is empty");
+  const producer = await producerForConfig(config);
+  await producer.send({
+    topic: message.topic,
+    messages: [{
+      key: message.key,
+      value: JSON.stringify(message.value),
+      headers: {
+        "x-values-printed": "false",
+        ...(message.headers ?? {}),
+      },
+    }],
+  });
+}
+
+export function publishAgentRunKafkaMessageBestEffort(config: AgentRunKafkaConfig, message: AgentRunKafkaMessage, context: JsonRecord = {}): void {
+  void publishAgentRunKafkaMessage(config, message)
+    .catch((error) => {
+      console.warn(JSON.stringify({
+        code: "agentrun-kafka-produce-failed",
+        component: "agentrun-kafka",
+        message: error instanceof Error ? error.message : String(error),
+        topic: message.topic,
+        keySha256: sha256Hex(message.key),
+        ...context,
+        valuesPrinted: false,
+      }));
+    });
+}
+
+export async function tailAgentRunKafka(options: KafkaTailOptions): Promise<JsonRecord> {
+  const env = options.env ?? process.env;
+  const config = agentRunKafkaConfig(env, { brokers: options.brokers, stream: options.stream, topic: options.topic });
+  const topic = kafkaTopicForStream(config, options.topic ?? options.stream);
+  if (config.brokers.length === 0) throw new Error("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS is empty; pass --brokers or set env");
+  const limit = boundedInteger(options.limit ?? 20, 1, 200, "limit");
+  const timeoutMs = boundedInteger(options.timeoutMs ?? 5_000, 500, 60_000, "timeout-ms");
+  const groupId = safeText(options.groupId) ?? `agentrun-cli-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
+  const kafka = new Kafka({ clientId: `${config.clientId}-cli`, brokers: config.brokers, logLevel: logLevel.NOTHING });
+  const consumer = kafka.consumer({ groupId });
+  const messages: JsonRecord[] = [];
+  const startedAt = Date.now();
+  let timedOut = false;
+  await consumer.connect();
+  try {
+    await consumer.subscribe({ topic, fromBeginning: options.fromBeginning !== false });
+    await new Promise<void>((resolve, reject) => {
+      const timer = setTimeout(() => {
+        timedOut = true;
+        resolve();
+      }, timeoutMs);
+      timer.unref?.();
+      consumer.run({
+        eachMessage: async (payload) => {
+          const item = kafkaMessageSummary(topic, payload, Boolean(options.values));
+          if (matchesKafkaFilter(item, options.filter)) messages.push(item);
+          if (messages.length >= limit) {
+            clearTimeout(timer);
+            resolve();
+          }
+        },
+      }).catch((error) => {
+        clearTimeout(timer);
+        reject(error);
+      });
+    });
+  } finally {
+    await disconnectQuietly(consumer);
+  }
+  return {
+    action: "agentrun-kafka-tail",
+    topic,
+    stream: kafkaStreamForTopic(config, topic),
+    brokers: config.brokers,
+    groupId,
+    fromBeginning: options.fromBeginning !== false,
+    limit,
+    timeoutMs,
+    elapsedMs: Date.now() - startedAt,
+    timedOut,
+    matched: messages.length,
+    messages,
+    valuesPrinted: Boolean(options.values),
+  };
+}
+
+export function sha256Hex(value: string | Buffer): string {
+  return createHash("sha256").update(value).digest("hex");
+}
+
+export function stableKafkaJson(value: JsonValue): string {
+  return JSON.stringify(sortJson(value));
+}
+
+export function jsonBytes(value: JsonValue): number {
+  return Buffer.byteLength(JSON.stringify(value), "utf8");
+}
+
+export function safeText(value: unknown, max = 240): string | null {
+  const text = typeof value === "string" ? value.trim() : value === undefined || value === null ? "" : String(value).trim();
+  if (text.length === 0) return null;
+  return text.length > max ? `${text.slice(0, max)}...` : text;
+}
+
+function kafkaMessageSummary(topic: string, payload: EachMessagePayload, includeValue: boolean): JsonRecord {
+  const raw = payload.message.value?.toString("utf8") ?? "";
+  const parsed = parseJsonRecord(raw);
+  const key = payload.message.key?.toString("utf8") ?? "";
+  const base: JsonRecord = {
+    topic,
+    partition: payload.partition,
+    offset: payload.message.offset,
+    timestamp: payload.message.timestamp ?? null,
+    keySha256: key ? sha256Hex(key) : null,
+    valueSha256: sha256Hex(raw),
+    valueBytes: Buffer.byteLength(raw, "utf8"),
+    schema: typeof parsed?.schema === "string" ? parsed.schema : null,
+    eventType: typeof parsed?.eventType === "string" ? parsed.eventType : null,
+    producedAt: typeof parsed?.producedAt === "string" ? parsed.producedAt : null,
+    runId: nestedString(parsed, ["run", "runId"]) ?? nestedString(parsed, ["context", "runId"]),
+    commandId: nestedString(parsed, ["command", "commandId"]) ?? nestedString(parsed, ["context", "commandId"]) ?? nestedString(parsed, ["event", "commandId"]),
+    sessionId: nestedString(parsed, ["run", "sessionId"]) ?? nestedString(parsed, ["context", "sessionId"]),
+    traceId: nestedString(parsed, ["context", "traceId"]) ?? nestedString(parsed, ["traceId"]),
+    method: nestedString(parsed, ["stdio", "method"]),
+    direction: nestedString(parsed, ["stdio", "direction"]),
+    valuesPrinted: includeValue,
+  };
+  if (includeValue) {
+    base.value = raw.length > MAX_VALUE_PREVIEW_CHARS ? `${raw.slice(0, MAX_VALUE_PREVIEW_CHARS)}...` : raw;
+    base.valueTruncated = raw.length > MAX_VALUE_PREVIEW_CHARS;
+  }
+  return base;
+}
+
+function matchesKafkaFilter(message: JsonRecord, filter: KafkaTailOptions["filter"]): boolean {
+  if (!filter) return true;
+  if (filter.traceId && message.traceId !== filter.traceId) return false;
+  if (filter.sessionId && message.sessionId !== filter.sessionId) return false;
+  if (filter.runId && message.runId !== filter.runId) return false;
+  if (filter.commandId && message.commandId !== filter.commandId) return false;
+  return true;
+}
+
+async function producerForConfig(config: AgentRunKafkaConfig): Promise<Producer> {
+  const key = `${config.clientId}|${config.brokers.join(",")}`;
+  if (producerState?.key === key && producerState.producer) return producerState.producer;
+  if (producerState?.key === key && producerState.connecting) return producerState.connecting;
+  const kafka = new Kafka({ clientId: config.clientId, brokers: config.brokers, logLevel: logLevel.NOTHING });
+  const producer = kafka.producer({ allowAutoTopicCreation: false });
+  const connecting = producer.connect().then(() => {
+    if (producerState?.producer === producer) producerState.connecting = null;
+    return producer;
+  }).catch((error) => {
+    if (producerState?.producer === producer) producerState = null;
+    throw error;
+  });
+  producerState = { key, producer, connecting };
+  return await connecting;
+}
+
+async function disconnectQuietly(consumer: Consumer): Promise<void> {
+  try {
+    await consumer.disconnect();
+  } catch {
+    // CLI tail should report already captured messages even if disconnect races with timeout.
+  }
+}
+
+function parseJsonRecord(value: string): JsonRecord | null {
+  try {
+    const parsed = JSON.parse(value) as unknown;
+    return parsed && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as JsonRecord : null;
+  } catch {
+    return null;
+  }
+}
+
+function nestedString(record: JsonRecord | null, path: string[]): string | null {
+  let current: unknown = record;
+  for (const key of path) {
+    if (!current || typeof current !== "object" || Array.isArray(current)) return null;
+    current = (current as Record<string, unknown>)[key];
+  }
+  return typeof current === "string" && current.length > 0 ? current : null;
+}
+
+function sortJson(value: JsonValue): JsonValue {
+  if (Array.isArray(value)) return value.map(sortJson);
+  if (!value || typeof value !== "object") return value;
+  const sorted: JsonRecord = {};
+  for (const key of Object.keys(value).sort()) sorted[key] = sortJson((value as JsonRecord)[key] as JsonValue);
+  return sorted;
+}
+
+function boundedInteger(value: number, min: number, max: number, label: string): number {
+  if (!Number.isFinite(value)) throw new Error(`${label} must be a finite number`);
+  const next = Math.floor(value);
+  if (next < min || next > max) throw new Error(`${label} must be between ${min} and ${max}`);
+  return next;
+}
+
+function truthy(value: string | undefined): boolean {
+  return TRUE_VALUES.has(String(value ?? "").trim().toLowerCase());
+}
+
+function csv(value: string | undefined): string[] {
+  return String(value ?? "").split(",").map((entry) => entry.trim()).filter(Boolean);
+}
+
+export function rawFrameValue(input: {
+  schema: string;
+  eventType: string;
+  source: string;
+  context: JsonRecord;
+  stdio: JsonRecord;
+  rawText?: string;
+  rawJson?: JsonRecord;
+}): JsonRecord {
+  const raw = input.rawText ?? (input.rawJson ? stableKafkaJson(input.rawJson) : "");
+  return {
+    schema: input.schema,
+    eventType: input.eventType,
+    source: input.source,
+    producedAt: new Date().toISOString(),
+    context: input.context,
+    stdio: {
+      ...input.stdio,
+      rawSha256: sha256Hex(raw),
+      rawBytes: Buffer.byteLength(raw, "utf8"),
+      rawText: input.rawText ?? null,
+      rawJson: input.rawJson ?? null,
+      valuesPrinted: true,
+    },
+    diagnostics: {
+      valuesPrinted: true,
+    },
+    valuesPrinted: true,
+  };
+}
+
+export function redactedKafkaValue(value: JsonRecord): JsonRecord {
+  return redactJson(value) as JsonRecord;
+}
+
+export function kafkaErrorMessage(error: unknown): string {
+  return error instanceof Error ? redactText(error.message) : redactText(String(error));
+}

src/mgr/kafka-shadow-producer.ts

@@ -1,18 +1,9 @@
-import { Kafka, logLevel, type Producer } from "kafkajs";
 import type { AgentRunStore } from "./store.js";
 import type { CommandRecord, JsonRecord, JsonValue, RunEvent, RunRecord } from "../common/types.js";
-import { stableHash } from "../common/validation.js";
+import { agentRunKafkaConfig, jsonBytes, publishAgentRunKafkaMessageBestEffort, safeText, sha256Hex, stableKafkaJson, type AgentRunKafkaConfig } from "../common/kafka-events.js";
 
 const TRUE_VALUES = new Set(["1", "true", "yes", "on"]);
 const warnedCodes = new Set<string>();
-let producerState: { key: string; producer: Producer; connecting: Promise<Producer> | null } | null = null;
-
-interface KafkaShadowConfig {
-  brokers: string[];
-  topic: string;
-  clientId: string;
-  key: string;
-}
 
 interface ShadowEventInput {
   env: NodeJS.ProcessEnv;
@@ -28,7 +19,7 @@ interface ShadowEventInput {
 type StoreFunction = (...args: never[]) => unknown;
 
 export function withKafkaShadowProducer(store: AgentRunStore, env: NodeJS.ProcessEnv = process.env): AgentRunStore {
-  if (!kafkaShadowConfig(env)) return store;
+  if (!kafkaEventConfig(env)) return store;
   return new Proxy(store, {
     get(target, prop, receiver) {
       if (prop === "createRun") {
@@ -74,17 +65,17 @@ export function withKafkaShadowProducer(store: AgentRunStore, env: NodeJS.Proces
 }
 
 function publishAgentRunShadow(input: ShadowEventInput): void {
-  const config = kafkaShadowConfig(input.env);
+  const config = kafkaEventConfig(input.env);
   if (!config) return;
   void Promise.resolve(resolveRun(input))
     .then((run) => {
       const message = buildAgentRunShadowMessage({ ...input, run, config });
-      return producerForConfig(config).then((producer) => producer.send({ topic: config.topic, messages: [message] }));
+      publishAgentRunKafkaMessageBestEffort(config, message, { eventType: input.eventType });
     })
     .catch((error) => warnShadowProducer("agentrun-kafka-shadow-produce-failed", {
       message: error instanceof Error ? error.message : String(error),
       eventType: input.eventType,
-      topic: config.topic,
+      topic: config.agentrunEventTopic,
       clientId: config.clientId,
     }));
 }
@@ -95,10 +86,10 @@ async function resolveRun(input: ShadowEventInput): Promise<RunRecord | null> {
   return await input.store.getRun(input.runId);
 }
 
-function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: KafkaShadowConfig }): { key: string; value: string; headers: Record<string, string> } {
+function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record<string, string> } {
   const payload = input.eventPayload ?? eventPayload(input.event);
   const event = {
-    schema: "agentrun.hwlab.event.shadow.v1",
+    schema: "agentrun.event.v1",
     eventType: input.eventType,
     source: input.config.clientId,
     producedAt: new Date().toISOString(),
@@ -107,7 +98,7 @@ function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord |
     command: commandSummary(input.command),
     event: runEventSummary(input.event, payload),
     diagnostics: {
-      payloadSha256: stableHash((payload ?? {}) as JsonValue),
+      payloadSha256: sha256Hex(stableKafkaJson((payload ?? {}) as JsonValue)),
       payloadBytes: jsonBytes(payload ?? {}),
       valuesPrinted: false,
       shadowConsumeEnabled: false,
@@ -115,8 +106,9 @@ function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord |
     valuesPrinted: false,
   };
   return {
+    topic: input.config.agentrunEventTopic,
     key: safeText(input.command?.id ?? input.eventPayload?.commandId ?? input.run?.id ?? input.runId) ?? input.config.clientId,
-    value: JSON.stringify(event),
+    value: event as JsonRecord,
     headers: {
       "x-shadow-mode": "produce-only",
       "x-values-printed": "false",
@@ -125,47 +117,24 @@ function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord |
   };
 }
 
-function kafkaShadowConfig(env: NodeJS.ProcessEnv): KafkaShadowConfig | null {
-  if (!truthy(env.AGENTRUN_KAFKA_SHADOW_PRODUCE_ENABLED)) return null;
+function kafkaEventConfig(env: NodeJS.ProcessEnv): AgentRunKafkaConfig | null {
+  if (!truthy(env.AGENTRUN_KAFKA_ENABLED) && !truthy(env.AGENTRUN_KAFKA_SHADOW_PRODUCE_ENABLED)) return null;
   if (truthy(env.AGENTRUN_KAFKA_SHADOW_CONSUME_ENABLED)) {
     warnOnce("agentrun-kafka-shadow-consume-ignored", {
       message: "AgentRun Kafka consumer cutover is disabled in this stage; producer continues in shadow mode.",
       valuesPrinted: false,
     });
   }
-  const brokers = csv(env.AGENTRUN_KAFKA_BOOTSTRAP_SERVERS);
-  const topic = safeText(env.AGENTRUN_KAFKA_EVENT_TOPIC);
-  const clientId = safeText(env.AGENTRUN_KAFKA_CLIENT_ID);
+  const config = agentRunKafkaConfig(env, { clientId: env.AGENTRUN_KAFKA_CLIENT_ID ?? "agentrun-v02-manager" });
   const missing: string[] = [];
-  if (brokers.length === 0) missing.push("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS");
-  if (!topic) missing.push("AGENTRUN_KAFKA_EVENT_TOPIC");
-  if (!clientId) missing.push("AGENTRUN_KAFKA_CLIENT_ID");
-  if (missing.length > 0 || !topic || !clientId) {
+  if (config.brokers.length === 0) missing.push("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS");
+  if (!config.agentrunEventTopic) missing.push("AGENTRUN_KAFKA_EVENT_TOPIC");
+  if (!config.clientId) missing.push("AGENTRUN_KAFKA_CLIENT_ID");
+  if (missing.length > 0) {
     warnOnce("agentrun-kafka-shadow-config-missing", { missing, valuesPrinted: false });
     return null;
   }
-  return { brokers, topic, clientId, key: `${clientId}|${topic}|${brokers.join(",")}` };
-}
-
-async function producerForConfig(config: KafkaShadowConfig): Promise<Producer> {
-  if (producerState?.key === config.key && producerState.producer) return producerState.producer;
-  if (producerState?.key === config.key && producerState.connecting) return producerState.connecting;
-  const kafka = new Kafka({ clientId: config.clientId, brokers: config.brokers, logLevel: logLevel.NOTHING });
-  const producer = kafka.producer({ allowAutoTopicCreation: false });
-  const connecting = producer.connect().then(() => {
-    if (producerState?.producer === producer) producerState.connecting = null;
-    return producer;
-  }).catch((error) => {
-    if (producerState?.producer === producer) producerState = null;
-    throw error;
-  });
-  const state: { key: string; producer: Producer; connecting: Promise<Producer> | null } = {
-    key: config.key,
-    producer,
-    connecting,
-  };
-  producerState = state;
-  return await connecting;
+  return config;
 }
 
 function runSummary(run: RunRecord | null): JsonRecord | null {
@@ -244,16 +213,6 @@ function csv(value: string | undefined): string[] {
   return String(value ?? "").split(",").map((entry) => entry.trim()).filter(Boolean);
 }
 
-function safeText(value: unknown, max = 240): string | null {
-  const text = typeof value === "string" ? value.trim() : value === undefined || value === null ? "" : String(value).trim();
-  if (text.length === 0) return null;
-  return text.length > max ? `${text.slice(0, max)}...` : text;
-}
-
-function jsonBytes(value: JsonRecord): number {
-  return Buffer.byteLength(JSON.stringify(value), "utf8");
-}
-
 function warnOnce(code: string, details: JsonRecord): void {
   if (warnedCodes.has(code)) return;
   warnedCodes.add(code);

src/runner/k8s-job.ts

@@ -284,6 +284,7 @@ function runnerEnv(options: RunnerJobRenderOptions, context: { namespace: string
     { name: "CODEX_HOME", value: codexHome },
     { name: "AGENTRUN_CODEX_HOME_STORAGE_KIND", value: context.sessionPvc ? "session-pvc" : "runner-home" },
     ...runnerOtelEnvVars(process.env),
+    ...runnerKafkaEnvVars(process.env),
     ...(selectedSecret ? [{ name: "AGENTRUN_CODEX_SECRET_HOME", value: selectedSecret.projectionMountPath }] : []),
     ...(context.sessionPvc ? [
       { name: "AGENTRUN_SESSION_PVC_NAME", value: context.sessionPvc.pvcName },
@@ -435,6 +436,16 @@ function runnerOtelEnvVars(env: NodeJS.ProcessEnv): JsonRecord[] {
   ];
 }
 
+function runnerKafkaEnvVars(env: NodeJS.ProcessEnv): JsonRecord[] {
+  return [
+    ...optionalEnvVar("AGENTRUN_KAFKA_ENABLED", env.AGENTRUN_KAFKA_ENABLED),
+    ...optionalEnvVar("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS", env.AGENTRUN_KAFKA_BOOTSTRAP_SERVERS),
+    ...optionalEnvVar("AGENTRUN_KAFKA_STDIO_TOPIC", env.AGENTRUN_KAFKA_STDIO_TOPIC),
+    ...optionalEnvVar("AGENTRUN_KAFKA_STDIO_CLIENT_ID", env.AGENTRUN_KAFKA_STDIO_CLIENT_ID),
+    ...optionalEnvVar("AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED", env.AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED),
+  ];
+}
+
 function optionalEnvVar(name: string, value: string | undefined): JsonRecord[] {
   const normalized = value?.trim();
   return normalized ? [{ name, value: normalized }] : [];