From 61833a036c6d7a1fdd68b1c8b5e1f1c43e0f4ae6 Mon Sep 17 00:00:00 2001 From: root Date: Fri, 10 Jul 2026 13:30:57 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=A2=9E=E5=8A=A0=20stdio=20Kafka=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E9=83=A8=E5=88=86=E9=87=8D=E5=BB=BA=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/reference/cli.md | 12 + docs/reference/spec-v01-cli.md | 40 ++ scripts/src/cli.ts | 135 +++- src/backend/codex-stdio.ts | 73 ++- src/debug/stdio-event-reconstruction.ts | 590 ++++++++++++++++++ .../cases/36-stdio-event-reconstruction.ts | 186 ++++++ .../fixtures/stdio-reconstruction.jsonl | 10 + 7 files changed, 1021 insertions(+), 25 deletions(-) create mode 100644 src/debug/stdio-event-reconstruction.ts create mode 100644 src/selftest/cases/36-stdio-event-reconstruction.ts create mode 100644 src/selftest/fixtures/stdio-reconstruction.jsonl diff --git a/docs/reference/cli.md b/docs/reference/cli.md index a38e00a..70797bd 100644 --- a/docs/reference/cli.md +++ b/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 --after-seq --limit ./scripts/agentrun runs result [--command-id ] ./scripts/agentrun runs cancel [--reason ] +./scripts/agentrun kafka regenerate agentrun --session-id [--trace-id ] [--from kafka|jsonl] [--jsonl-file ] [--no-publish|--dry-run] [--output-jsonl ] [--json|--format json] ./scripts/agentrun commands create --type turn|steer|interrupt --json-stdin|--json-file ./scripts/agentrun commands show --run-id ./scripts/agentrun commands result --run-id @@ -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。 diff --git a/docs/reference/spec-v01-cli.md b/docs/reference/spec-v01-cli.md index 479cf30..7d962fc 100644 --- a/docs/reference/spec-v01-cli.md +++ b/docs/reference/spec-v01-cli.md @@ -41,6 +41,7 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交 ./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 kafka regenerate agentrun --session-id [--trace-id ] [--from kafka|jsonl] [--jsonl-file ] [--no-publish|--dry-run] [--output-jsonl ] [--json|--format json] ./scripts/agentrun commands create --type turn|steer|interrupt --json-stdin|--json-file ./scripts/agentrun commands show --run-id ./scripts/agentrun commands result --run-id @@ -103,6 +104,27 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交 - `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。 +- `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_` 可逆转换时生成 `hwlabSessionId=ses_`； + - 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` 切换主闭环。 | diff --git a/scripts/src/cli.ts b/scripts/src/cli.ts index b198487..f928153 100644 --- a/scripts/src/cli.ts +++ b/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 { + 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 { 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 { 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 { }) as JsonValue; } +async function kafkaRegenerateAgentRun(args: ParsedArgs): Promise { + 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 [--trace-id ] [--from kafka|jsonl] [--jsonl-file ] [--no-publish|--dry-run] [--output-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 { 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 [--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 ]", + "kafka regenerate agentrun --session-id [--trace-id ] [--from kafka|jsonl] [--jsonl-file ] [--input-topic codex-stdio.raw.v1] [--output-topic agentrun.event.debug.v1] [--group-id ] [--no-publish|--dry-run] [--output-jsonl ] [--format json|compact] [--json]", "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 a40762f..dc7d812 100644 --- a/src/backend/codex-stdio.ts +++ b/src/backend/codex-stdio.ts @@ -145,6 +145,53 @@ interface SuppressedNotificationSummary { byItemType: Record; } +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; 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 | null = interruptInFlight as Promise | 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 } : {}) }; } diff --git a/src/debug/stdio-event-reconstruction.ts b/src/debug/stdio-event-reconstruction.ts new file mode 100644 index 0000000..8885a96 --- /dev/null +++ b/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; +} + +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; +export type ReconstructionPublisher = (records: ReconstructedAgentRunKafkaRecord[], config: AgentRunKafkaConfig) => Promise; + +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 { + 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 = {}; + 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(); + 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 = {}; + 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 { + 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((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 { + 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_", + 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 { + 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, `'"'"'`)}'`; +} diff --git a/src/selftest/cases/36-stdio-event-reconstruction.ts b/src/selftest/cases/36-stdio-event-reconstruction.ts new file mode 100644 index 0000000..fa0ab25 --- /dev/null +++ b/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((resolve, reject) => { + proc.once("error", reject); + proc.once("close", (value) => resolve(value ?? 1)); + }); + return { code, stdout: stdout.trim(), stderr: stderr.trim() }; +} + +export default selfTest; diff --git a/src/selftest/fixtures/stdio-reconstruction.jsonl b/src/selftest/fixtures/stdio-reconstruction.jsonl new file mode 100644 index 0000000..768a577 --- /dev/null +++ b/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}}