Merge pull request #300 from pikasTech/fix/299-stdio-kafka-frame-order

修复 stdio Kafka 帧发布顺序

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<void>;
 }
 
 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<CodexStdioKafkaPublisher>();
   private readonly pending = new Map<number, PendingRequest>();
   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<void> {
+    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) } }];

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<void> {
+  return publishAgentRunKafkaMessage(config, message)
     .catch((error) => {
       console.warn(JSON.stringify({
         code: "agentrun-kafka-produce-failed",

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<void> {
+  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<string, ObservedFrame[]>();
+  const completedByKey = new Map<string, ObservedFrame[]>();
+  const inFlightByKey = new Map<string, number>();
+  const maxInFlightByKey = new Map<string, number>();
+  let releaseFirst!: () => void;
+  let observeFirstStarted!: () => void;
+  const firstGate = new Promise<void>((resolve) => { releaseFirst = resolve; });
+  const firstStarted = new Promise<void>((resolve) => { observeFirstStarted = resolve; });
+  let firstFrameGated = false;
+  let closePromise: ReturnType<CodexStdioBackendSession["close"]> | 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<void> {
+  let releaseKafka!: () => void;
+  let observeKafkaStarted!: () => void;
+  const kafkaGate = new Promise<void>((resolve) => { releaseKafka = resolve; });
+  const kafkaStarted = new Promise<void>((resolve) => { observeKafkaStarted = resolve; });
+  let client: CodexStdioClient | null = null;
+  let drainPromise: Promise<void> | 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<void> {
+  const store = new MemoryAgentRunStore();
+  const first = createTurn(store, context, "parallel-a");
+  const second = createTurn(store, context, "parallel-b");
+  const inFlightByKey = new Map<string, number>();
+  const maxInFlightByKey = new Map<string, number>();
+  let globalInFlight = 0;
+  let globalMaxInFlight = 0;
+  let producerCount = 0;
+  let releaseFirst!: () => void;
+  let observeFirstStarted!: () => void;
+  const firstGate = new Promise<void>((resolve) => { releaseFirst = resolve; });
+  const firstStarted = new Promise<void>((resolve) => { observeFirstStarted = resolve; });
+  let firstFrameGated = false;
+  let firstResultPromise: ReturnType<typeof runTurn> | 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<T>(promise: Promise<T>, timeoutMs: number, label: string): Promise<T> {
+  let timer: NodeJS.Timeout | null = null;
+  try {
+    return await Promise.race([
+      promise,
+      new Promise<never>((_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<PublishObservation> {
+  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<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+export default selfTest;