pikasTech-agentrun/src/selftest/cases/36-stdio-event-reconstruction.ts

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 {
  closeAgentRunKafkaProducer,
  publishAgentRunKafkaMessage,
  setAgentRunKafkaProducerFactoryForSelfTest,
  sha256Hex,
  type AgentRunKafkaConfig,
} from "../../common/kafka-events.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,
  readStdioKafkaFrames,
  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, 9);
  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: 9 }, (_, 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 finalSeal = reconstructed.records.find((record) => {
    const event = record.value.event as JsonRecord;
    const payload = event.payload as JsonRecord;
    return event.type === "backend_status" && payload.finalSeal === true;
  });
  const finalSealPayload = (finalSeal?.value.event as JsonRecord).payload as JsonRecord;
  assert.equal(finalSealPayload.finalSeal, true);
  assert.equal(finalSealPayload.phase, "assistant-message-final-sealed");
  assert.equal(Object.hasOwn(finalSealPayload, "text"), false);
  const replyRef = finalSealPayload.replyRef as JsonRecord;
  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.itemId === replyRef.itemId && payload.source === replyRef.source;
  });
  const finalAssistantText = String(((finalAssistant?.value.event as JsonRecord).payload as JsonRecord).text ?? "");
  assert.equal(finalAssistantText, "stdio reconstruction fixture answer");
  assert.equal(replyRef.textSha256, sha256Hex(finalAssistantText));
  assert.equal((reconstructed.records.at(-1)?.value.event as JsonRecord).type, "terminal_status");

  const fileChangeFixture = await readFile(path.join(context.root, "src/selftest/fixtures/stdio-file-change-reconstruction.jsonl"), "utf8");
  const fileChangeFrames = parseStdioKafkaFrameJsonl(fileChangeFixture);
  const fileChangeReconstruction = reconstructAgentRunEventsFromStdioFrames(fileChangeFrames, {
    sessionId: "ses_agentrun_d4473d6c_4d7a_4612_a102_254977698bc2",
    traceId: "trc_9352107b5e8a4125",
  });
  assert.equal(fileChangeReconstruction.selectedFrameCount, 8);
  assert.deepEqual(fileChangeReconstruction.byEventType, { backend_status: 4, tool_call: 3, diff: 1, assistant_message: 1, terminal_status: 1 });
  const fileChangeEvents = fileChangeReconstruction.records.map((record) => record.value.event as JsonRecord);
  const completedFileChange = fileChangeEvents.find((event) => event.type === "tool_call" && (event.payload as JsonRecord).method === "item/completed");
  assert.equal(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[0]?.path, "src/example.ts");
  assert.equal(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[1]?.path, "docs/说明.md");
  assert.match(String(((completedFileChange?.payload as JsonRecord).changes as JsonRecord[])[0]?.diff), /\+export const value = 2;/u);
  const aggregateDiff = fileChangeEvents.find((event) => event.type === "diff");
  assert.match(String((aggregateDiff?.payload as JsonRecord).diff), /\+# 文件编辑可见/u);

  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);

  await assertOneShotPublisherOwnership(frames);
  await assertKafkaReaderStopsBeforeDisconnect(frames[0]!);

  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",
      "one-shot-producer-disconnect",
      "kafka-reader-stop-before-disconnect",
      "manager-producer-isolation",
      "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 assertKafkaReaderStopsBeforeDisconnect(frame: ReturnType<typeof parseStdioKafkaFrameJsonl>[number]): Promise<void> {
  const lifecycle: string[] = [];
  let stopped = false;
  let releaseRun: (() => void) | null = null;
  const runFinished = new Promise<void>((resolve) => { releaseRun = resolve; });
  const consumer = {
    connect: async () => { lifecycle.push("connect"); },
    subscribe: async () => { lifecycle.push("subscribe"); },
    run: async ({ eachMessage }: { eachMessage: (input: JsonRecord) => Promise<void> }) => {
      lifecycle.push("run");
      await eachMessage({
        topic: frame.topic,
        partition: frame.partition ?? 0,
        message: {
          offset: frame.offset ?? "0",
          key: frame.key === null ? null : Buffer.from(frame.key),
          timestamp: frame.timestamp,
          value: Buffer.from(JSON.stringify(frame.value)),
        },
      });
      await runFinished;
      assert.equal(stopped, true, "consumer.run must only settle after consumer.stop");
    },
    stop: async () => { lifecycle.push("stop"); stopped = true; releaseRun?.(); },
    disconnect: async () => { lifecycle.push("disconnect"); },
  };
  const result = await readStdioKafkaFrames({
    config: {
      brokers: ["127.0.0.1:9092"],
      clientId: "agentrun-selftest",
      agentrunEventTopic: "agentrun.event.v1",
      codexStdioTopic: "codex-stdio.raw.v1",
      dnsServers: [],
      dnsSearchDomains: [],
      enabled: true,
      valuesPrinted: false,
    },
    topic: frame.topic,
    groupId: "agentrun-selftest-reader",
    sessionId,
    traceId,
    limit: 1,
    timeoutMs: 250,
  }, { consumer: consumer as never });
  assert.equal(result.records.length, 1);
  assert.deepEqual(lifecycle, ["connect", "subscribe", "run", "stop", "disconnect"]);
}

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",
    dnsServers: [],
    dnsSearchDomains: [],
    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: 9, 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 = "";
  let stderr = "";
  proc.stdout.on("data", (chunk) => { stdout += String(chunk); });
  proc.stderr.on("data", (chunk) => { stderr += String(chunk); });
  const code = await new Promise<number>((resolve, reject) => {
    proc.once("error", reject);
    proc.once("close", (value) => resolve(value ?? 1));
  });
  return { code, stdout: stdout.trim(), stderr: stderr.trim() };
}

export default selfTest;