Merge pull request #1858 from pikasTech/fix/kafka-replay-diagnostics

Kafka 调试 CLI 增加集群内应用 replay 诊断

scripts/src/platform-infra-kafka-replay.test.ts

@@ -0,0 +1,67 @@
+import { describe, expect, test } from "bun:test";
+import { parseKafkaReplayOptions, renderKafkaReplay } from "./platform-infra-kafka";
+import { renderMachine } from "./agentrun/render";
+
+describe("platform-infra Kafka replay", () => {
+  test("accepts the bounded no-publish application replay contract", () => {
+    expect(parseKafkaReplayOptions([
+      "agentrun", "--node", "NC01", "--lane", "nc01-v02",
+      "--session-id", "ses_example", "--trace-id", "trc_example",
+      "--no-publish", "-o", "yaml",
+    ])).toMatchObject({
+      targetId: "NC01",
+      lane: "nc01-v02",
+      sessionId: "ses_example",
+      traceId: "trc_example",
+      output: "yaml",
+    });
+  });
+
+  test("rejects replay without the explicit no-publish boundary", () => {
+    expect(() => parseKafkaReplayOptions(["agentrun", "--session-id", "ses_example"])).toThrow("requires --no-publish");
+  });
+
+  test("renders matched replay counts and mutation disclosure", () => {
+    const rendered = renderKafkaReplay({
+      ok: true,
+      diagnosis: {
+        networkPlane: "k3s-application-pod",
+        application: "agentrun",
+        namespace: "agentrun-v02",
+        workload: "deployment/agentrun-mgr",
+        topic: "codex-stdio.raw.v1",
+        scanned: 30,
+        accepted: 12,
+        produced: 8,
+        rejected: { classification: "matched" },
+        mutation: false,
+        topicAppended: false,
+        valuesPrinted: false,
+        firstMismatchIdentity: null,
+      },
+    });
+    expect(rendered.renderedText).toContain("k3s-application-pod");
+    expect(rendered.renderedText).toContain("30");
+    expect(rendered.renderedText).toContain("matched");
+    expect(rendered.renderedText).toContain("no              no");
+  });
+
+  test("keeps JSON and YAML machine projections isomorphic", () => {
+    const payload = {
+      ok: true,
+      diagnosis: {
+        networkPlane: "k3s-application-pod",
+        scanned: 30,
+        accepted: 12,
+        produced: 8,
+        rejected: { classification: "matched", byReason: {} },
+        mutation: false,
+        topicAppended: false,
+        valuesPrinted: false,
+      },
+    };
+    const json = JSON.parse(renderMachine("replay", payload, "json").renderedText);
+    const yaml = Bun.YAML.parse(renderMachine("replay", payload, "yaml").renderedText);
+    expect(yaml).toEqual(json);
+  });
+});

scripts/src/platform-infra-kafka.ts

@@ -2,6 +2,8 @@ import { readFileSync } from "node:fs";
 import type { UniDeskConfig } from "./config";
 import { rootPath } from "./config";
 import type { RenderedCliResult } from "./output";
+import { renderMachine } from "./agentrun/render";
+import { resolveAgentRunLaneTarget } from "./agentrun-lanes";
 import {
   capture,
   compactCapture,
@@ -17,6 +19,7 @@ const configLabel = "config/platform-infra/kafka.yaml";
 const fieldManager = "unidesk-platform-infra-kafka";
 const y = createYamlFieldReader(configLabel);
 const kafkaConsumerGroupsNativeScript = readFileSync(rootPath("scripts", "native", "platform-infra", "kafka-consumer-groups.py"), "utf8");
+const kafkaReplayDiagnosticsNativeScript = readFileSync(rootPath("scripts", "native", "platform-infra", "kafka-replay-diagnostics.py"), "utf8");
 
 interface KafkaTarget {
   id: string;
@@ -116,6 +119,13 @@ interface PlatformKafkaConfig {
     smokeTopic: string;
   };
   management: {
+    replay: {
+      defaultLimit: number;
+      maxLimit: number;
+      defaultTimeoutMs: number;
+      maxTimeoutMs: number;
+      commandGraceMs: number;
+    };
     defaultTailLimit: number;
     maxTailLimit: number;
     defaultGroupListLimit: number;
@@ -164,6 +174,15 @@ interface KafkaGroupCleanupOptions extends CommonOptions {
   limit: number | null;
 }
 
+export interface KafkaReplayOptions extends CommonOptions {
+  sessionId: string;
+  traceId: string | null;
+  lane: string | null;
+  limit: number | null;
+  timeoutMs: number | null;
+  output: "text" | "json" | "yaml";
+}
+
 export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown> | RenderedCliResult> {
   const [action = "plan"] = args;
   if (action === "plan") {
@@ -203,6 +222,12 @@ export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args:
     const result = await tail(config, options);
     return options.full || options.raw ? result : renderTail(result);
   }
+  if (action === "replay") {
+    const options = parseKafkaReplayOptions(args.slice(1));
+    const result = await replayAgentRun(config, options);
+    if (options.output === "json" || options.output === "yaml") return renderMachine("platform-infra kafka replay agentrun", result, options.output, result.ok !== false);
+    return renderKafkaReplay(result);
+  }
   if (action === "produce") return await produce(config, await parseProduceOptions(args.slice(1)));
   return {
     ok: false,
@@ -214,7 +239,7 @@ export async function runPlatformInfraKafkaCommand(config: UniDeskConfig, args:
 
 export function kafkaHelp(): Record<string, unknown> {
   return {
-    command: "platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|produce",
+    command: "platform-infra kafka plan|apply|status|validate|topics|groups|offsets|tail|replay|produce",
     configTruth: configLabel,
     usage: [
       "bun scripts/cli.ts platform-infra kafka plan --target NC01",
@@ -228,6 +253,7 @@ export function kafkaHelp(): Record<string, unknown> {
       "bun scripts/cli.ts platform-infra kafka groups cleanup --node NC01 --policy hwlab-cloud-api-sse [--confirm]",
       "bun scripts/cli.ts platform-infra kafka offsets --node NC01 [--topic hwlab.event.v1]",
       "bun scripts/cli.ts platform-infra kafka tail --node NC01 --topic hwlab.event.v1 --limit 5",
+      "bun scripts/cli.ts platform-infra kafka replay agentrun --node NC01 --lane nc01-v02 --session-id <id> --trace-id <id> --no-publish [-o json|yaml]",
       "bun scripts/cli.ts platform-infra kafka produce --node NC01 --topic hwlab.event.v1 --key <trace-or-session>",
     ],
     boundary: "Kafka runtime belongs to platform-infra. Durable producers and fixed-group consumers are enabled; browser delivery remains a database-projected SSE concern.",
@@ -350,8 +376,16 @@ function parseClient(record: Record<string, unknown>, index: number): KafkaClien
 
 function parseManagement(root: Record<string, unknown>): PlatformKafkaConfig["management"] {
   const management = y.objectField(root, "management", "");
+  const replay = y.objectField(management, "replay", "management");
   const cleanup = y.objectField(management, "consumerGroupCleanup", "management");
   return {
+    replay: {
+      defaultLimit: positiveInteger(replay, "defaultLimit", "management.replay"),
+      maxLimit: positiveInteger(replay, "maxLimit", "management.replay"),
+      defaultTimeoutMs: positiveInteger(replay, "defaultTimeoutMs", "management.replay"),
+      maxTimeoutMs: positiveInteger(replay, "maxTimeoutMs", "management.replay"),
+      commandGraceMs: positiveInteger(replay, "commandGraceMs", "management.replay"),
+    },
     defaultTailLimit: positiveInteger(management, "defaultTailLimit", "management"),
     maxTailLimit: positiveInteger(management, "maxTailLimit", "management"),
     defaultGroupListLimit: positiveInteger(management, "defaultGroupListLimit", "management"),
@@ -407,6 +441,9 @@ function validateConfig(kafka: PlatformKafkaConfig): void {
   if (!/^[0-9]+Gi$/u.test(kafka.cluster.storage.size)) throw new Error(`${configLabel}.cluster.storage.size must be a Gi quantity`);
   if (!kafka.topics.some((topic) => topic.name === kafka.validation.smokeTopic)) throw new Error(`${configLabel}.validation.smokeTopic must reference one of topics[].name`);
   if (!kafka.topics.some((topic) => topic.name === kafka.management.defaultTopic)) throw new Error(`${configLabel}.management.defaultTopic must reference one of topics[].name`);
+  if (kafka.management.replay.defaultLimit > kafka.management.replay.maxLimit) throw new Error(`${configLabel}.management.replay.defaultLimit must be <= maxLimit`);
+  if (kafka.management.replay.defaultTimeoutMs > kafka.management.replay.maxTimeoutMs) throw new Error(`${configLabel}.management.replay.defaultTimeoutMs must be <= maxTimeoutMs`);
+  if (kafka.management.replay.maxTimeoutMs + kafka.management.replay.commandGraceMs >= 60_000) throw new Error(`${configLabel}.management.replay timeout plus command grace must stay below the trans short-connection budget`);
   if (kafka.management.defaultTailLimit > kafka.management.maxTailLimit) throw new Error(`${configLabel}.management.defaultTailLimit must be <= management.maxTailLimit`);
   if (kafka.management.defaultGroupListLimit > kafka.management.maxGroupListLimit) throw new Error(`${configLabel}.management.defaultGroupListLimit must be <= management.maxGroupListLimit`);
   if (kafka.management.consumerGroupCleanup.defaultCandidateLimit > kafka.management.consumerGroupCleanup.maxCandidateLimit) throw new Error(`${configLabel}.management.consumerGroupCleanup.defaultCandidateLimit must be <= maxCandidateLimit`);
@@ -661,6 +698,50 @@ async function tail(config: UniDeskConfig, options: KafkaInspectOptions): Promis
   };
 }
 
+async function replayAgentRun(config: UniDeskConfig, options: KafkaReplayOptions): Promise<Record<string, unknown>> {
+  const kafka = readKafkaConfig();
+  const target = resolveTarget(kafka, options.targetId);
+  const lane = resolveAgentRunLaneTarget({ node: target.id, lane: options.lane }).spec;
+  if (lane.nodeId !== target.id) throw new Error(`AgentRun lane ${lane.lane} belongs to ${lane.nodeId}, not Kafka target ${target.id}`);
+  const inputTopic = resolveTopic(kafka, "codex-stdio.raw.v1", null);
+  if (inputTopic === null) throw new Error("Kafka replay input topic codex-stdio.raw.v1 is not declared");
+  const limit = options.limit ?? kafka.management.replay.defaultLimit;
+  if (!Number.isInteger(limit) || limit < 1 || limit > kafka.management.replay.maxLimit) throw new Error(`Kafka replay --limit must be an integer from 1 to ${kafka.management.replay.maxLimit}`);
+  const timeoutMs = options.timeoutMs ?? kafka.management.replay.defaultTimeoutMs;
+  if (!Number.isInteger(timeoutMs) || timeoutMs < 250 || timeoutMs > kafka.management.replay.maxTimeoutMs) throw new Error(`Kafka replay --timeout-ms must be an integer from 250 to ${kafka.management.replay.maxTimeoutMs}`);
+  const runtimeEnvironment = {
+    KAFKA_NAMESPACE: target.namespace,
+    KAFKA_CLUSTER: kafka.cluster.name,
+    KAFKA_BOOTSTRAP: brokerBootstrap(kafka),
+    KAFKA_INPUT_TOPIC: inputTopic,
+    KAFKA_REPLAY_SESSION_ID: options.sessionId,
+    KAFKA_REPLAY_TRACE_ID: options.traceId ?? "",
+    KAFKA_REPLAY_LIMIT: String(limit),
+    KAFKA_REPLAY_TIMEOUT_MS: String(timeoutMs),
+    KAFKA_REPLAY_COMMAND_GRACE_MS: String(kafka.management.replay.commandGraceMs),
+    AGENTRUN_NAMESPACE: lane.runtime.namespace,
+    AGENTRUN_MANAGER_DEPLOYMENT: lane.runtime.managerDeployment,
+  };
+  const remoteScript = [
+    ...Object.entries(runtimeEnvironment).map(([key, value]) => `export ${key}=${shQuote(value)}`),
+    "python3 - <<'PY'",
+    kafkaReplayDiagnosticsNativeScript.trimEnd(),
+    "PY",
+  ].join("\n");
+  const result = await capture(config, target.route, ["sh"], remoteScript);
+  const parsed = parseJsonOutput(result.stdout);
+  return {
+    ok: result.exitCode === 0 && parsed?.ok === true,
+    action: "platform-infra-kafka-replay-agentrun",
+    mode: "no-publish",
+    mutation: false,
+    target: targetSummary(target),
+    lane: { node: lane.nodeId, lane: lane.lane, namespace: lane.runtime.namespace, managerDeployment: lane.runtime.managerDeployment },
+    diagnosis: parsed,
+    remote: options.raw && parsed !== null ? parsed : compactCapture(result, { full: options.full || result.exitCode !== 0 }),
+  };
+}
+
 async function produce(config: UniDeskConfig, options: KafkaProduceOptions): Promise<Record<string, unknown>> {
   const kafka = readKafkaConfig();
   const target = resolveTarget(kafka, options.targetId);
@@ -1840,6 +1921,22 @@ function renderTail(result: Record<string, unknown>): RenderedCliResult {
   ]);
 }
 
+export function renderKafkaReplay(result: Record<string, unknown>): RenderedCliResult {
+  const diagnosis = record(result.diagnosis);
+  const rejected = record(diagnosis.rejected);
+  const mismatch = record(diagnosis.firstMismatchIdentity);
+  return rendered(result, "platform-infra kafka replay agentrun", [
+    "PLATFORM-INFRA KAFKA REPLAY",
+    ...table(["PLANE", "APPLICATION", "NAMESPACE", "WORKLOAD"], [[stringValue(diagnosis.networkPlane), stringValue(diagnosis.application), stringValue(diagnosis.namespace), stringValue(diagnosis.workload)]]),
+    "",
+    ...table(["TOPIC", "SCANNED", "ACCEPTED", "PRODUCED", "CLASSIFICATION"], [[stringValue(diagnosis.topic), stringValue(diagnosis.scanned), stringValue(diagnosis.accepted), stringValue(diagnosis.produced), stringValue(rejected.classification)]]),
+    "",
+    ...table(["MUTATION", "TOPIC_APPENDED", "VALUES_PRINTED"], [[boolText(diagnosis.mutation), boolText(diagnosis.topicAppended), boolText(diagnosis.valuesPrinted)]]),
+    ...(diagnosis.firstMismatchIdentity === null ? [] : ["", `MISMATCH  session=${stringValue(mismatch.sessionId)} trace=${stringValue(mismatch.traceId)} reason=${stringValue(mismatch.reason)}`]),
+    ...remoteErrorLines(result),
+  ]);
+}
+
 function remoteErrorLines(result: Record<string, unknown>): string[] {
   if (result.ok !== false) return [];
   const remote = record(result.remote);
@@ -1930,6 +2027,42 @@ function parseInspectOptions(args: string[]): KafkaInspectOptions {
   return { ...parseCommonOptions(commonArgs), topic, group, limit };
 }
 
+export function parseKafkaReplayOptions(args: string[]): KafkaReplayOptions {
+  const commonArgs: string[] = [];
+  let application: string | null = null;
+  let sessionId: string | null = null;
+  let traceId: string | null = null;
+  let lane: string | null = null;
+  let limit: number | null = null;
+  let timeoutMs: number | null = null;
+  let output: KafkaReplayOptions["output"] = "text";
+  let noPublish = false;
+  for (let index = 0; index < args.length; index += 1) {
+    const arg = args[index];
+    if (!arg.startsWith("--") && application === null) application = arg;
+    else if (arg === "--session-id") { sessionId = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--trace-id") { traceId = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--lane") { lane = parseSimpleOption(args[index + 1], arg); index += 1; }
+    else if (arg === "--limit") { limit = Number(parseSimpleOption(args[index + 1], arg)); index += 1; }
+    else if (arg === "--timeout-ms") { timeoutMs = Number(parseSimpleOption(args[index + 1], arg)); index += 1; }
+    else if (arg === "--no-publish" || arg === "--dry-run") noPublish = true;
+    else if (arg === "--json") output = "json";
+    else if (arg === "-o" || arg === "--output") {
+      const value = parseSimpleOption(args[index + 1], arg);
+      if (value !== "text" && value !== "json" && value !== "yaml") throw new Error(`${arg} must be text|json|yaml`);
+      output = value;
+      index += 1;
+    } else {
+      commonArgs.push(arg);
+      if (arg === "--target" || arg === "--node") { commonArgs.push(args[index + 1] ?? ""); index += 1; }
+    }
+  }
+  if (application !== "agentrun") throw new Error("Kafka replay requires application 'agentrun'");
+  if (sessionId === null) throw new Error("Kafka replay requires --session-id <id>");
+  if (!noPublish) throw new Error("Kafka replay currently requires --no-publish; product topic replay is unsupported");
+  return { ...parseCommonOptions(commonArgs), sessionId, traceId, lane, limit, timeoutMs, output };
+}
+
 function parseGroupCleanupOptions(args: string[]): KafkaGroupCleanupOptions {
   const commonArgs: string[] = [];
   let policyId: string | null = null;