From 240a53ff253878f91db7e9346a98218b3027bbb8 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 11 Jul 2026 21:26:56 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E6=8C=89=20frameSeq=20=E4=B8=B2?= =?UTF-8?q?=E8=A1=8C=E5=8F=91=E5=B8=83=20stdio=20Kafka=20=E5=B8=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/backend/codex-stdio.ts | 35 +- src/common/kafka-events.ts | 4 +- .../cases/38-stdio-kafka-frame-order.ts | 432 ++++++++++++++++++ 3 files changed, 465 insertions(+), 6 deletions(-) create mode 100644 src/selftest/cases/38-stdio-kafka-frame-order.ts diff --git a/src/backend/codex-stdio.ts b/src/backend/codex-stdio.ts index dc7d812..43b98b5 100644 --- a/src/backend/codex-stdio.ts +++ b/src/backend/codex-stdio.ts @@ -9,7 +9,7 @@ import { redactJson, redactText } from "../common/redaction.js"; import { backendProfileSpec } from "../common/backend-profiles.js"; import { boundedTextSummary, commandOutputPayload } from "../common/output.js"; import { agentRunBusinessTraceId, emitAgentRunOtelSpan } from "../common/otel-trace.js"; -import { agentRunKafkaConfig, publishAgentRunKafkaMessageBestEffort, rawFrameValue, safeText, type AgentRunKafkaConfig } from "../common/kafka-events.js"; +import { agentRunKafkaConfig, publishAgentRunKafkaMessageBestEffort, rawFrameValue, safeText, type AgentRunKafkaConfig, type AgentRunKafkaMessage } from "../common/kafka-events.js"; const codexProtocol = "codex-app-server-jsonrpc-stdio"; const defaultCodexArgs = ["app-server", "--listen", "stdio://"]; @@ -103,6 +103,7 @@ interface CodexStdioKafkaPublisher { context: CodexStdioKafkaContext; source: string; frameSeq: number; + publishTail: Promise; } export interface CodexActiveTurnControl { @@ -221,6 +222,7 @@ class CodexStdioFailure extends Error { export class CodexStdioClient { private readonly child: ChildProcessWithoutNullStreams; private stdioKafka: CodexStdioKafkaPublisher | null; + private readonly retiredStdioKafka = new Set(); private readonly pending = new Map(); private stderrTailBuffer = Buffer.alloc(0); private stderrBytes = 0; @@ -264,6 +266,14 @@ export class CodexStdioClient { } setStdioKafka(publisher: CodexStdioKafkaPublisher | null): void { + const previous = this.stdioKafka; + if (previous && previous !== publisher) { + this.retiredStdioKafka.add(previous); + void previous.publishTail.then( + () => this.retiredStdioKafka.delete(previous), + () => this.retiredStdioKafka.delete(previous), + ); + } this.stdioKafka = publisher; } @@ -469,6 +479,12 @@ export class CodexStdioClient { this.closeResolve(closeInfo); } + async drainStdioKafkaPublishers(): Promise { + const publishers = new Set(this.retiredStdioKafka); + if (this.stdioKafka) publishers.add(this.stdioKafka); + await Promise.all([...publishers].map(async (publisher) => await publisher.publishTail)); + } + private stderrInfo(): { stderrTail: string; stderrTruncated: boolean } { const buffered = this.stderrTailBuffer.toString("utf8"); const tail = buffered.slice(-8000); @@ -495,6 +511,7 @@ function codexStdioKafkaPublisher(options: CodexStdioTurnOptions, env: NodeJS.Pr source: config.clientId, context: codexStdioKafkaContext(options), frameSeq: 0, + publishTail: Promise.resolve(), }; } @@ -529,7 +546,7 @@ function publishCodexStdioFrame(publisher: CodexStdioKafkaPublisher | null | und const context = publisher.context; const frameSeq = ++publisher.frameSeq; const keySource = context.commandId ?? context.runId ?? publisher.source; - publishAgentRunKafkaMessageBestEffort(publisher.config, { + const message: AgentRunKafkaMessage = { topic: publisher.config.codexStdioTopic, key: keySource, value: rawFrameValue({ @@ -546,7 +563,7 @@ function publishCodexStdioFrame(publisher: CodexStdioKafkaPublisher | null | und rpcId: input.rpcId ?? null, }, rawText, - rawJson: input.rawJson, + ...(input.rawJson ? { rawJson: input.rawJson } : {}), }), headers: { "x-event-type": `codex.stdio.${direction}`, @@ -554,7 +571,16 @@ function publishCodexStdioFrame(publisher: CodexStdioKafkaPublisher | null | und "x-frame-seq": String(frameSeq), "x-values-printed": "true", }, - }, { eventType: `codex.stdio.${direction}`, direction, method: input.method }); + }; + // frameSeq is assigned at capture time; a per-publisher tail makes Kafka offset + // assignment follow that same formal order without imposing global serialization. + publisher.publishTail = publisher.publishTail.then(async () => { + await publishAgentRunKafkaMessageBestEffort( + publisher.config, + message, + { eventType: `codex.stdio.${direction}`, direction, method: input.method, frameSeq }, + ); + }); } export class CodexStdioBackendSession { @@ -573,6 +599,7 @@ export class CodexStdioBackendSession { this.clientKey = null; client.stop(); const closeInfo = await client.closedPromise; + await client.drainStdioKafkaPublishers(); emitCodexOtelSpanFromLifecycle("codex_app_server.exit", this.lifecycleOtel, { status: closeInfo.failureKind ? "error" : "ok", error: closeInfo.message ?? closeInfo.failureKind ?? undefined, attributes: { ...closeEventAttributes(closeInfo), failureKind: closeInfo.failureKind } }); this.lifecycleOtel = null; return [{ type: "backend_status", payload: { phase: "codex-app-server-closed", appServerExit: closeEvent(closeInfo) } }]; diff --git a/src/common/kafka-events.ts b/src/common/kafka-events.ts index 8e910ef..f5492b5 100644 --- a/src/common/kafka-events.ts +++ b/src/common/kafka-events.ts @@ -119,8 +119,8 @@ export async function setAgentRunKafkaProducerFactoryForSelfTest(factory: KafkaP producerFactory = factory ?? defaultProducerFactory; } -export function publishAgentRunKafkaMessageBestEffort(config: AgentRunKafkaConfig, message: AgentRunKafkaMessage, context: JsonRecord = {}): void { - void publishAgentRunKafkaMessage(config, message) +export function publishAgentRunKafkaMessageBestEffort(config: AgentRunKafkaConfig, message: AgentRunKafkaMessage, context: JsonRecord = {}): Promise { + return publishAgentRunKafkaMessage(config, message) .catch((error) => { console.warn(JSON.stringify({ code: "agentrun-kafka-produce-failed", diff --git a/src/selftest/cases/38-stdio-kafka-frame-order.ts b/src/selftest/cases/38-stdio-kafka-frame-order.ts new file mode 100644 index 0000000..ab974b2 --- /dev/null +++ b/src/selftest/cases/38-stdio-kafka-frame-order.ts @@ -0,0 +1,432 @@ +import assert from "node:assert/strict"; +import { CodexStdioBackendSession, CodexStdioClient, runCodexStdioTurn, type CodexStdioTurnOptions } from "../../backend/codex-stdio.js"; +import { + agentRunKafkaConfig, + closeAgentRunKafkaProducer, + setAgentRunKafkaProducerFactoryForSelfTest, +} from "../../common/kafka-events.js"; +import type { CommandRecord, CreateRunInput, JsonRecord, RunRecord } from "../../common/types.js"; +import { validateCreateCommand } from "../../common/validation.js"; +import { MemoryAgentRunStore } from "../../mgr/store.js"; +import type { SelfTestCase, SelfTestContext } from "../harness.js"; + +interface ObservedFrame { + frameSeq: number; + method: string; +} + +interface PublishObservation { + started: ObservedFrame[]; + completed: ObservedFrame[]; + failed: ObservedFrame[]; + maxInFlight: number; + producerCount: number; +} + +const selfTest: SelfTestCase = async (context) => { + const ordered = await runPublishScenario(context, null); + assert.equal(ordered.maxInFlight, 1, "codex stdio Kafka publisher must have only one frame publish in flight"); + assert.deepEqual( + ordered.started.map((frame) => frame.frameSeq), + ordered.completed.map((frame) => frame.frameSeq), + "Kafka publish completion order must match formal frameSeq order", + ); + assertStrictFrameSequence(ordered.started); + assert.equal(ordered.completed.at(-1)?.method, "process.close", "shutdown must drain through the lifecycle frame"); + + const recovered = await runPublishScenario(context, 2); + assert.equal(recovered.maxInFlight, 1, "a failed frame must not introduce concurrent recovery publishes"); + assert.deepEqual(recovered.failed.map((frame) => frame.frameSeq), [2]); + assertStrictFrameSequence(recovered.started); + assert.equal(recovered.started.at(-1)?.method, "process.close", "a failed frame must not poison the serial tail"); + assert.equal(recovered.completed.at(-1)?.method, "process.close", "shutdown must drain later frames after one best-effort failure"); + assert.ok(recovered.producerCount >= 2, "a failed producer must be replaced for later frames"); + + await assertPublisherIsolation(context); + await assertWarmSessionPublisherSwapDrainsRetired(context); + await assertTransportCloseIndependentFromKafkaDrain(context); + + return { + name: "stdio-kafka-frame-order", + tests: [ + "per-publisher-single-inflight", + "frame-seq-completion-order", + "best-effort-failure-tail-recovery", + "shutdown-drains-lifecycle-frame", + "parallel-publisher-isolation", + "warm-session-retired-publisher-drain", + "transport-close-not-delayed-by-kafka-gate", + ], + }; +}; + +async function assertWarmSessionPublisherSwapDrainsRetired(context: SelfTestContext): Promise { + const store = new MemoryAgentRunStore(); + const first = createTurn(store, context, "warm-first"); + const second = createTurn(store, context, "warm-second"); + const session = new CodexStdioBackendSession(); + const startedByKey = new Map(); + const completedByKey = new Map(); + const inFlightByKey = new Map(); + const maxInFlightByKey = new Map(); + let releaseFirst!: () => void; + let observeFirstStarted!: () => void; + const firstGate = new Promise((resolve) => { releaseFirst = resolve; }); + const firstStarted = new Promise((resolve) => { observeFirstStarted = resolve; }); + let firstFrameGated = false; + let closePromise: ReturnType | null = null; + + await setAgentRunKafkaProducerFactoryForSelfTest(() => ({ + connect: async () => undefined, + send: async (request: { messages: Array<{ key?: string | Buffer | null; value: string | Buffer }> }) => { + const message = request.messages[0]; + const key = String(message?.key ?? "unknown"); + const value = JSON.parse(String(message?.value ?? "{}")) as JsonRecord; + const stdio = value.stdio as JsonRecord; + const frame = { frameSeq: Number(value.frameSeq), method: String(stdio.method ?? "unknown") }; + const started = startedByKey.get(key) ?? []; + started.push(frame); + startedByKey.set(key, started); + const inFlight = (inFlightByKey.get(key) ?? 0) + 1; + inFlightByKey.set(key, inFlight); + maxInFlightByKey.set(key, Math.max(maxInFlightByKey.get(key) ?? 0, inFlight)); + try { + if (key === first.command.id && !firstFrameGated) { + firstFrameGated = true; + observeFirstStarted(); + await firstGate; + } + const completed = completedByKey.get(key) ?? []; + completed.push(frame); + completedByKey.set(key, completed); + return []; + } finally { + inFlightByKey.set(key, (inFlightByKey.get(key) ?? 1) - 1); + } + }, + disconnect: async () => undefined, + } as never)); + + try { + const firstResultPromise = session.runTurn(turnOptions(context, first.run, first.command)); + await withTimeout(firstStarted, 2_000, "warm first publisher gated frame"); + const firstResult = await withTimeout(firstResultPromise, 2_000, "warm first turn result"); + assert.equal(firstResult.terminalStatus, "completed", "runTurn must not wait for best-effort Kafka drain"); + + const secondResult = await withTimeout( + session.runTurn(turnOptions(context, second.run, second.command)), + 2_000, + "warm second turn after publisher swap", + ); + assert.equal(secondResult.terminalStatus, "completed", "replacement publisher must progress while retired publisher is gated"); + assert.ok((completedByKey.get(second.command.id)?.length ?? 0) > 0); + assert.equal(maxInFlightByKey.get(first.command.id), 1); + assert.equal(maxInFlightByKey.get(second.command.id), 1); + + let closeSettled = false; + closePromise = session.close().finally(() => { closeSettled = true; }); + await delay(30); + assert.equal(closeSettled, false, "session.close must wait for the gated retired publisher tail"); + releaseFirst(); + await withTimeout(closePromise, 2_000, "warm session retired publisher drain"); + + assert.deepEqual(startedByKey.get(first.command.id), completedByKey.get(first.command.id)); + assert.deepEqual(startedByKey.get(second.command.id), completedByKey.get(second.command.id)); + assert.equal(completedByKey.get(second.command.id)?.at(-1)?.method, "process.close"); + } finally { + releaseFirst(); + closePromise ??= session.close(); + await closePromise.catch(() => undefined); + await closeAgentRunKafkaProducer(); + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } +} + +async function assertTransportCloseIndependentFromKafkaDrain(context: SelfTestContext): Promise { + let releaseKafka!: () => void; + let observeKafkaStarted!: () => void; + const kafkaGate = new Promise((resolve) => { releaseKafka = resolve; }); + const kafkaStarted = new Promise((resolve) => { observeKafkaStarted = resolve; }); + let client: CodexStdioClient | null = null; + let drainPromise: Promise | null = null; + + await setAgentRunKafkaProducerFactoryForSelfTest(() => ({ + connect: async () => undefined, + send: async () => { + observeKafkaStarted(); + await kafkaGate; + return []; + }, + disconnect: async () => undefined, + } as never)); + + try { + const env = stdioKafkaEnv(context); + const config = agentRunKafkaConfig(env, { stream: "stdio", clientId: "agentrun-selftest-transport-close" }); + client = new CodexStdioClient({ + command: process.execPath, + args: ["-e", "process.exit(0)"], + cwd: context.workspace, + env, + onNotification: () => undefined, + stdioKafka: { + config, + source: config.clientId, + context: { + runId: null, + commandId: null, + attemptId: null, + runnerId: null, + runnerJobId: null, + jobName: null, + podName: null, + sessionId: null, + conversationId: null, + threadId: null, + backendProfile: "codex", + sourceCommit: null, + valuesPrinted: false, + }, + frameSeq: 0, + publishTail: Promise.resolve(), + } as never, + }); + await withTimeout(kafkaStarted, 2_000, "gated lifecycle Kafka publish"); + const closeInfo = await withTimeout(client.closedPromise, 200, "transport close independent from Kafka"); + assert.equal(closeInfo.code, 0); + + let drainSettled = false; + drainPromise = client.drainStdioKafkaPublishers().finally(() => { drainSettled = true; }); + await delay(20); + assert.equal(drainSettled, false, "explicit stdio Kafka drain must still wait for the gated lifecycle publish"); + releaseKafka(); + await withTimeout(drainPromise, 2_000, "stdio Kafka drain after gate release"); + assert.equal(drainSettled, true); + } finally { + releaseKafka(); + client?.stop(); + await drainPromise?.catch(() => undefined); + await closeAgentRunKafkaProducer(); + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } +} + +async function assertPublisherIsolation(context: SelfTestContext): Promise { + const store = new MemoryAgentRunStore(); + const first = createTurn(store, context, "parallel-a"); + const second = createTurn(store, context, "parallel-b"); + const inFlightByKey = new Map(); + const maxInFlightByKey = new Map(); + let globalInFlight = 0; + let globalMaxInFlight = 0; + let producerCount = 0; + let releaseFirst!: () => void; + let observeFirstStarted!: () => void; + const firstGate = new Promise((resolve) => { releaseFirst = resolve; }); + const firstStarted = new Promise((resolve) => { observeFirstStarted = resolve; }); + let firstFrameGated = false; + let firstResultPromise: ReturnType | null = null; + + await setAgentRunKafkaProducerFactoryForSelfTest(() => { + producerCount += 1; + return { + connect: async () => undefined, + send: async (request: { messages: Array<{ key?: string | Buffer | null }> }) => { + const key = String(request.messages[0]?.key ?? "unknown"); + const keyInFlight = (inFlightByKey.get(key) ?? 0) + 1; + inFlightByKey.set(key, keyInFlight); + maxInFlightByKey.set(key, Math.max(maxInFlightByKey.get(key) ?? 0, keyInFlight)); + globalInFlight += 1; + globalMaxInFlight = Math.max(globalMaxInFlight, globalInFlight); + try { + if (key === first.command.id && !firstFrameGated) { + firstFrameGated = true; + observeFirstStarted(); + await firstGate; + } + return []; + } finally { + inFlightByKey.set(key, (inFlightByKey.get(key) ?? 1) - 1); + globalInFlight -= 1; + } + }, + disconnect: async () => undefined, + } as never; + }); + + try { + let firstSettled = false; + firstResultPromise = runTurn(context, first.run, first.command).finally(() => { firstSettled = true; }); + await withTimeout(firstStarted, 2_000, "first publisher gated frame"); + const secondResult = await withTimeout(runTurn(context, second.run, second.command), 2_000, "second publisher completion"); + assert.equal(secondResult.terminalStatus, "completed"); + assert.equal(firstSettled, false, "publisher B must complete while publisher A remains gated"); + assert.ok(globalMaxInFlight >= 2, "different stdio publishers must be able to publish concurrently"); + assert.equal(maxInFlightByKey.get(first.command.id), 1); + assert.equal(maxInFlightByKey.get(second.command.id), 1); + assert.equal(producerCount, 1, "parallel publishers should share the configured Kafka producer without sharing a serial tail"); + releaseFirst(); + const firstResult = await withTimeout(firstResultPromise, 2_000, "first publisher completion after release"); + assert.equal(firstResult.terminalStatus, "completed"); + } finally { + releaseFirst(); + await firstResultPromise?.catch(() => undefined); + await closeAgentRunKafkaProducer(); + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } +} + +function createTurn(store: MemoryAgentRunStore, context: SelfTestContext, suffix: string): { run: RunRecord; command: CommandRecord } { + const run = store.createRun(runInput(context, suffix)); + const command = store.createCommand(run.id, validateCreateCommand({ + type: "turn", + idempotencyKey: `stdio-order-${suffix}`, + payload: { prompt: `verify ${suffix}`, traceId: `trc_stdio_order_${suffix}` }, + })); + return { run, command }; +} + +function runInput(context: SelfTestContext, suffix: string): CreateRunInput { + return { + tenantId: "hwlab", + projectId: "pikasTech/HWLAB", + workspaceRef: { kind: "host-path", path: context.workspace }, + sessionRef: { sessionId: `ses_stdio_order_${suffix}` }, + resourceBundleRef: null, + providerId: "gpt.pika", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 5_000, + network: "default", + secretScope: { providerCredentials: [], toolCredentials: [] }, + }, + traceSink: { kind: "hwlab", traceId: `trc_stdio_order_${suffix}` }, + }; +} + +function runTurn(context: SelfTestContext, run: RunRecord, command: CommandRecord) { + return runCodexStdioTurn(turnOptions(context, run, command)); +} + +function turnOptions(context: SelfTestContext, run: RunRecord, command: CommandRecord): CodexStdioTurnOptions { + return { + backendProfile: "codex", + prompt: String(command.payload.prompt ?? "verify stdio order"), + cwd: context.workspace, + approvalPolicy: "never", + sandbox: "workspace-write", + timeoutMs: 5_000, + command: context.fakeCodexCommand, + args: context.fakeCodexArgs, + codexHome: context.codexHome, + otelContext: { run, command }, + env: stdioKafkaEnv(context), + }; +} + +async function withTimeout(promise: Promise, timeoutMs: number, label: string): Promise { + let timer: NodeJS.Timeout | null = null; + try { + return await Promise.race([ + promise, + new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new Error(`timed out waiting for ${label}`)), timeoutMs); + }), + ]); + } finally { + if (timer) clearTimeout(timer); + } +} + +async function runPublishScenario(context: SelfTestContext, failFrameSeq: number | null): Promise { + const started: ObservedFrame[] = []; + const completed: ObservedFrame[] = []; + const failed: ObservedFrame[] = []; + let inFlight = 0; + let maxInFlight = 0; + let producerCount = 0; + let failureInjected = false; + + await setAgentRunKafkaProducerFactoryForSelfTest(() => { + producerCount += 1; + return { + connect: async () => undefined, + send: async (request: { messages: Array<{ value: string | Buffer }> }) => { + const value = JSON.parse(String(request.messages[0]?.value ?? "{}")) as JsonRecord; + const stdio = value.stdio as JsonRecord; + const frame = { + frameSeq: Number(value.frameSeq), + method: String(stdio.method ?? "unknown"), + }; + started.push(frame); + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + try { + if (failFrameSeq === frame.frameSeq && !failureInjected) { + failureInjected = true; + failed.push(frame); + throw new Error(`injected stdio Kafka failure frameSeq=${frame.frameSeq}`); + } + await delay(frame.frameSeq % 2 === 1 ? 12 : 0); + completed.push(frame); + return []; + } finally { + inFlight -= 1; + } + }, + disconnect: async () => undefined, + } as never; + }); + + try { + const result = await runCodexStdioTurn({ + backendProfile: "codex", + prompt: "verify stdio Kafka frame order", + cwd: context.workspace, + approvalPolicy: "never", + sandbox: "workspace-write", + timeoutMs: 5_000, + command: context.fakeCodexCommand, + args: context.fakeCodexArgs, + codexHome: context.codexHome, + env: stdioKafkaEnv(context), + }); + assert.equal(result.terminalStatus, "completed"); + assert.equal(inFlight, 0, "runCodexStdioTurn close must wait for all stdio Kafka publishes"); + assert.equal(completed.some((frame) => frame.method === "process.close"), true, "close must wait for the lifecycle frame publish"); + } finally { + await closeAgentRunKafkaProducer(); + await setAgentRunKafkaProducerFactoryForSelfTest(null); + } + + return { started, completed, failed, maxInFlight, producerCount }; +} + +function stdioKafkaEnv(context: SelfTestContext): NodeJS.ProcessEnv { + return { + CODEX_HOME: context.codexHome, + AGENTRUN_FAKE_CODEX_MODE: "success", + AGENTRUN_CODEX_TERMINAL_NOTIFICATION_DRAIN_MS: "0", + AGENTRUN_KAFKA_STDIO_PRODUCE_ENABLED: "true", + AGENTRUN_KAFKA_BOOTSTRAP_SERVERS: "127.0.0.1:9092", + AGENTRUN_KAFKA_STDIO_TOPIC: "codex-stdio.raw.v1", + AGENTRUN_KAFKA_STDIO_CLIENT_ID: "agentrun-selftest-stdio-order", + }; +} + +function assertStrictFrameSequence(frames: ObservedFrame[]): void { + assert.ok(frames.length > 2, "test must observe multiple stdio frames"); + assert.deepEqual( + frames.map((frame) => frame.frameSeq), + Array.from({ length: frames.length }, (_, index) => index + 1), + "stdio publisher must start Kafka sends in contiguous formal frameSeq order", + ); +} + +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export default selfTest;