mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 释放 stdio 重建调试 producer

Browse Source
This commit is contained in:
root
2026-07-10 13:48:57 +02:00
parent 021db77f6e
commit 48a5fcd3e8
4 changed files with 112 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/reference/spec-v01-cli.md
+7
View File
@@ -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 连接残留而阻塞退出。
## 规格的实现情况
src/common/kafka-events.ts
+28 -11
View File
@@ -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<void> {
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<void> {
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<void> {
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<void> {
if (producerState?.producer === producer) producerState = null;
await disconnectProducerQuietly(producer);
src/debug/stdio-event-reconstruction.ts
+7 -4
View File
@@ -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<void> {
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[] {
src/selftest/cases/36-stdio-event-reconstruction.ts
+70
View File
@@ -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<typeof parseStdioKafkaFrameJsonl>): Promise<void> {
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 = "";