diff --git a/docs/reference/spec-v01-agentrun-mgr.md b/docs/reference/spec-v01-agentrun-mgr.md index 4e6ac15..0858986 100644 --- a/docs/reference/spec-v01-agentrun-mgr.md +++ b/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 接口说明 diff --git a/docs/reference/spec-v01-backend-codex.md b/docs/reference/spec-v01-backend-codex.md index 05d9bd2..c2c57ce 100644 --- a/docs/reference/spec-v01-backend-codex.md +++ b/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。 diff --git a/docs/reference/spec-v01-cli.md b/docs/reference/spec-v01-cli.md index 0694015..479cf30 100644 --- a/docs/reference/spec-v01-cli.md +++ b/docs/reference/spec-v01-cli.md @@ -39,6 +39,8 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交 ./scripts/agentrun runs events --after-seq --limit [--summary|--tail-summary] [--tail ] [--summary-chars ] [--format json|tsv] ./scripts/agentrun runs result [--command-id ] ./scripts/agentrun runs cancel [--reason ] +./scripts/agentrun kafka tail [agentrun|stdio] [--topic ] [--brokers ] [--limit ] [--timeout-ms ] [--run-id ] [--session-id ] [--command-id ] [--trace-id ] [--values] +./scripts/agentrun kafka events [agentrun|stdio] [--topic ] [--brokers ] [--limit ] [--timeout-ms ] ./scripts/agentrun commands create --type turn|steer|interrupt --json-stdin|--json-file ./scripts/agentrun commands show --run-id ./scripts/agentrun commands result --run-id @@ -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 ` 与 `--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。 diff --git a/scripts/src/cli.ts b/scripts/src/cli.ts index 23dd7c6..ee71d20 100644 --- a/scripts/src/cli.ts +++ b/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 { 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 { return client(args).get(`/api/v1/runs/${encodeURIComponent(runId)}/runner-jobs${commandId ? `?commandId=${encodeURIComponent(commandId)}` : ""}`); } +async function kafkaTail(args: ParsedArgs): Promise { + 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 { 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 --after-seq --limit [--summary|--tail-summary] [--tail ] [--summary-chars ] [--format json|tsv]", "runs result [--command-id ]", "runs cancel [--reason ]", + "kafka tail [agentrun|stdio] [--topic ] [--brokers ] [--limit ] [--timeout-ms ] [--run-id ] [--session-id ] [--command-id ] [--trace-id ] [--values]", + "kafka events [agentrun|stdio] [--topic ] [--brokers ] [--limit ] [--timeout-ms ]", "sessions ps [--state default|running|unread|terminal|idle|all] [--profile codex|deepseek|minimax-m3|dsflash-go||M3] [--reader-id ]", "sessions create [sessionId] [--profile codex|deepseek|minimax-m3|dsflash-go||M3] [--expires-in-days ]", "sessions storage ", diff --git a/src/backend/codex-stdio.ts b/src/backend/codex-stdio.ts index 8e9a2a6..d7bc8e5 100644 --- a/src/backend/codex-stdio.ts +++ b/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(); private stderrTailBuffer = Buffer.alloc(0); private stderrBytes = 0; @@ -158,8 +182,9 @@ export class CodexStdioClient { readonly closedPromise: Promise; 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 { 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; diff --git a/src/common/kafka-events.ts b/src/common/kafka-events.ts new file mode 100644 index 0000000..2df1008 --- /dev/null +++ b/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 | 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; +} + +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 { + 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 { + 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((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 { + 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 { + 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)[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)); +} diff --git a/src/mgr/kafka-shadow-producer.ts b/src/mgr/kafka-shadow-producer.ts index 6b1f4d3..45cc541 100644 --- a/src/mgr/kafka-shadow-producer.ts +++ b/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(); -let producerState: { key: string; producer: Producer; connecting: Promise | 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 { return await input.store.getRun(input.runId); } -function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: KafkaShadowConfig }): { key: string; value: string; headers: Record } { +function buildAgentRunShadowMessage(input: ShadowEventInput & { run: RunRecord | null; config: AgentRunKafkaConfig }): { topic: string; key: string; value: JsonRecord; headers: Record } { 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 { - 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 | 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); diff --git a/src/runner/k8s-job.ts b/src/runner/k8s-job.ts index 2107a50..db83920 100644 --- a/src/runner/k8s-job.ts +++ b/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 }] : [];