diff --git a/docs/reference/spec-v01-cli.md b/docs/reference/spec-v01-cli.md index 7d962fc..c874c07 100644 --- a/docs/reference/spec-v01-cli.md +++ b/docs/reference/spec-v01-cli.md @@ -110,6 +110,9 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交 - 不依赖 manager、runner 或数据库运行面； - 默认发布到 `agentrun.event.debug.v1`； - 默认 consumer group 使用 `agentrun-v02-stdio-regenerate-debug-*` 前缀； + - 调试发布使用命令独占 producer； + - 成功或失败后都必须在返回结果前断开连接； + - 不得关闭或复用 manager 的长驻缓存 producer； - 默认输出 compact text； - 只有显式 `--json` 或 `--format json` 才输出结构化摘要； - 每次输出都必须包含下一条可执行命令提示。 @@ -203,6 +206,10 @@ CLI 官方 TypeScript 入口固定为 `scripts/agentrun-cli.ts`。在 G14 非交 - 显式把 output topic 设为 `agentrun.event.v1`： - 确认命令失败； - 确认没有 producer 写入。 +- 检查调试 producer 生命周期： + - 确认发布完成后独占 producer 已断开； + - 确认 manager 缓存 producer 未被断开且可继续复用； + - 确认命令进程不因 Kafka producer 连接残留而阻塞退出。 ## 规格的实现情况 diff --git a/src/common/kafka-events.ts b/src/common/kafka-events.ts index 1df394e..8e910ef 100644 --- a/src/common/kafka-events.ts +++ b/src/common/kafka-events.ts @@ -82,23 +82,26 @@ export async function publishAgentRunKafkaMessage(config: AgentRunKafkaConfig, m if (config.brokers.length === 0) throw new Error("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS is empty"); const producer = await producerForConfig(config); try { - await producer.send({ - topic: message.topic, - messages: [{ - key: message.key, - value: JSON.stringify(message.value), - headers: { - "x-values-printed": "false", - ...(message.headers ?? {}), - }, - }], - }); + await sendAgentRunKafkaMessage(producer, message); } catch (error) { await invalidateProducer(producer); throw error; } } +export async function publishAgentRunKafkaMessagesOnce(config: AgentRunKafkaConfig, messages: AgentRunKafkaMessage[]): Promise { + if (!config.enabled) return; + if (messages.length === 0) return; + if (config.brokers.length === 0) throw new Error("AGENTRUN_KAFKA_BOOTSTRAP_SERVERS is empty"); + const producer = producerFactory(config); + try { + await producer.connect(); + for (const message of messages) await sendAgentRunKafkaMessage(producer, message); + } finally { + await disconnectProducerQuietly(producer); + } +} + export async function closeAgentRunKafkaProducer(): Promise { const state = producerState; producerState = null; @@ -277,6 +280,20 @@ function defaultProducerFactory(config: AgentRunKafkaConfig): Producer { return kafka.producer({ allowAutoTopicCreation: false }); } +async function sendAgentRunKafkaMessage(producer: Producer, message: AgentRunKafkaMessage): Promise { + await producer.send({ + topic: message.topic, + messages: [{ + key: message.key, + value: JSON.stringify(message.value), + headers: { + "x-values-printed": "false", + ...(message.headers ?? {}), + }, + }], + }); +} + async function invalidateProducer(producer: Producer): Promise { if (producerState?.producer === producer) producerState = null; await disconnectProducerQuietly(producer); diff --git a/src/debug/stdio-event-reconstruction.ts b/src/debug/stdio-event-reconstruction.ts index 8885a96..332c010 100644 --- a/src/debug/stdio-event-reconstruction.ts +++ b/src/debug/stdio-event-reconstruction.ts @@ -5,7 +5,7 @@ import { AgentRunError } from "../common/errors.js"; import { normalizeRunEventPayload } from "../common/events.js"; import { agentRunKafkaConfig, - publishAgentRunKafkaMessage, + publishAgentRunKafkaMessagesOnce, sha256Hex, stableKafkaJson, type AgentRunKafkaConfig, @@ -336,9 +336,12 @@ export async function readStdioKafkaFrames(request: StdioFrameReadRequest): Prom } 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 }); - } + await publishAgentRunKafkaMessagesOnce(config, records.map((record) => ({ + topic: record.topic, + key: record.key, + value: record.value, + headers: record.headers, + }))); } export function parseStdioKafkaFrameJsonl(text: string, defaultTopic = "codex-stdio.raw.v1"): StdioKafkaFrameRecord[] { diff --git a/src/selftest/cases/36-stdio-event-reconstruction.ts b/src/selftest/cases/36-stdio-event-reconstruction.ts index fa0ab25..cb29d8d 100644 --- a/src/selftest/cases/36-stdio-event-reconstruction.ts +++ b/src/selftest/cases/36-stdio-event-reconstruction.ts @@ -3,6 +3,12 @@ import { spawn } from "node:child_process"; import { readFile } from "node:fs/promises"; import path from "node:path"; import { AgentRunError } from "../../common/errors.js"; +import { + closeAgentRunKafkaProducer, + publishAgentRunKafkaMessage, + setAgentRunKafkaProducerFactoryForSelfTest, + type AgentRunKafkaConfig, +} from "../../common/kafka-events.js"; import type { JsonRecord } from "../../common/types.js"; import { DEFAULT_STDIO_RECONSTRUCTION_GROUP_PREFIX, @@ -74,6 +80,8 @@ const selfTest: SelfTestCase = async (context) => { assert.equal(published.length, reconstructed.records.length); assert.match(String(result.nextHint), /kafka tail agentrun --topic agentrun\.event\.debug\.v1/u); + await assertOneShotPublisherOwnership(frames); + let productPublisherCalls = 0; await assert.rejects( () => regenerateAgentRunEventsFromStdio({ sessionId, traceId, brokers: "127.0.0.1:9092", outputTopic: "agentrun.event.v1" }, { @@ -159,6 +167,8 @@ const selfTest: SelfTestCase = async (context) => { "production-reducer-reuse", "jsonl-offline-source", "injected-reader-publisher", + "one-shot-producer-disconnect", + "manager-producer-isolation", "partial-reconstruction-provenance", "debug-topic-group-defaults", "product-topic-permanent-guard", @@ -170,6 +180,66 @@ const selfTest: SelfTestCase = async (context) => { }; }; +async function assertOneShotPublisherOwnership(frames: ReturnType): Promise { + const producers: Array<{ connectCalls: number; sendCalls: number; disconnectCalls: number }> = []; + await setAgentRunKafkaProducerFactoryForSelfTest(() => { + const state = { connectCalls: 0, sendCalls: 0, disconnectCalls: 0 }; + producers.push(state); + return { + connect: async () => { state.connectCalls += 1; }, + send: async () => { state.sendCalls += 1; return []; }, + disconnect: async () => { state.disconnectCalls += 1; }, + } as never; + }); + const managerConfig: AgentRunKafkaConfig = { + brokers: ["127.0.0.1:9092"], + clientId: "agentrun-selftest", + agentrunEventTopic: "agentrun.event.v1", + codexStdioTopic: "codex-stdio.raw.v1", + enabled: true, + valuesPrinted: false, + }; + try { + const managerMessage = { topic: managerConfig.agentrunEventTopic, key: sessionId, value: { schema: "agentrun.event.v1" } }; + await publishAgentRunKafkaMessage(managerConfig, managerMessage); + assert.equal(producers.length, 1); + assert.equal(producers[0]?.disconnectCalls, 0); + + await regenerateAgentRunEventsFromStdio({ sessionId, traceId, brokers: managerConfig.brokers[0] }, { + reader: async () => ({ records: frames, scanned: frames.length, invalidJson: 0, timedOut: false, elapsedMs: 0 }), + }); + assert.equal(producers.length, 2); + assert.deepEqual(producers[1], { connectCalls: 1, sendCalls: 10, disconnectCalls: 1 }); + assert.equal(producers[0]?.disconnectCalls, 0); + + await publishAgentRunKafkaMessage(managerConfig, managerMessage); + assert.equal(producers.length, 2); + assert.equal(producers[0]?.sendCalls, 2); + } finally { + await closeAgentRunKafkaProducer(); + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } + assert.equal(producers[0]?.disconnectCalls, 1); + + const failedProducer = { connectCalls: 0, sendCalls: 0, disconnectCalls: 0 }; + await setAgentRunKafkaProducerFactoryForSelfTest(() => ({ + connect: async () => { failedProducer.connectCalls += 1; }, + send: async () => { failedProducer.sendCalls += 1; throw new Error("debug publish failed"); }, + disconnect: async () => { failedProducer.disconnectCalls += 1; }, + } as never)); + try { + await assert.rejects( + () => regenerateAgentRunEventsFromStdio({ sessionId, traceId, brokers: managerConfig.brokers[0] }, { + reader: async () => ({ records: frames, scanned: frames.length, invalidJson: 0, timedOut: false, elapsedMs: 0 }), + }), + /debug publish failed/u, + ); + } finally { + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } + assert.deepEqual(failedProducer, { connectCalls: 1, sendCalls: 1, disconnectCalls: 1 }); +} + 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 = "";