Merge pull request #2712 from pikasTech/fix/kafka-replay-classification

fix: 正确分类 Kafka replay 运行时失败

scripts/native/platform-infra/kafka-replay-diagnostics.py

@@ -86,7 +86,7 @@ def classify(replay, window, execution):
     selected = data.get("selectedFrameCount", 0)
     invalid = source.get("invalidJson", 0)
     ignored = data.get("ignoredFrames") if isinstance(data.get("ignoredFrames"), dict) else {}
-    if execution.get("timedOut") or window.get("errorClass") or execution.get("exitCode") in (126, 127):
+    if execution.get("timedOut") or window.get("errorClass") or execution.get("exitCode") != 0:
         return "network-or-runtime-error"
     if isinstance(replay, dict) and replay.get("ok") is True and output.get("recordCount", 0) > 0:
         return "matched"

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

@@ -1,4 +1,8 @@
 import { describe, expect, test } from "bun:test";
+import { chmodSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { spawnSync } from "node:child_process";
 import { parseKafkaReplayOptions, renderKafkaReplay } from "./platform-infra-kafka";
 import { renderMachine } from "./agentrun/render";
 
@@ -64,4 +68,49 @@ describe("platform-infra Kafka replay", () => {
     const yaml = Bun.YAML.parse(renderMachine("replay", payload, "yaml").renderedText);
     expect(yaml).toEqual(json);
   });
+
+  test("classifies a nonzero application replay as a runtime error", () => {
+    const directory = mkdtempSync(join(tmpdir(), "unidesk-kafka-replay-"));
+    const kubectl = join(directory, "kubectl");
+    writeFileSync(kubectl, `#!/bin/sh
+case "$*" in
+  *"get pod"*) printf 'kafka-0' ;;
+  *"--time -2"*) printf 'codex-stdio.raw.v1:0:0\\n' ;;
+  *"--time -1"*) printf 'codex-stdio.raw.v1:0:4\\n' ;;
+  *"deployment/agentrun-mgr"*) printf '%s\\n' 'KafkaJS error: timeout must not be negative: -1784589809970' >&2; exit 1 ;;
+  *) exit 2 ;;
+esac
+`);
+    chmodSync(kubectl, 0o755);
+    try {
+      const result = spawnSync("python3", ["scripts/native/platform-infra/kafka-replay-diagnostics.py"], {
+        cwd: process.cwd(),
+        encoding: "utf8",
+        env: {
+          ...process.env,
+          PATH: `${directory}:${process.env.PATH ?? ""}`,
+          KAFKA_NAMESPACE: "platform-infra",
+          KAFKA_CLUSTER: "platform-kafka",
+          KAFKA_BOOTSTRAP: "platform-kafka-kafka-bootstrap:9092",
+          KAFKA_INPUT_TOPIC: "codex-stdio.raw.v1",
+          KAFKA_REPLAY_SESSION_ID: "ses_example",
+          KAFKA_REPLAY_TRACE_ID: "trc_example",
+          KAFKA_REPLAY_LIMIT: "5000",
+          KAFKA_REPLAY_TIMEOUT_MS: "10000",
+          KAFKA_REPLAY_COMMAND_GRACE_MS: "35000",
+          AGENTRUN_NAMESPACE: "agentrun-v02",
+          AGENTRUN_MANAGER_DEPLOYMENT: "agentrun-mgr",
+        },
+      });
+      expect(result.status).toBe(0);
+      const payload = JSON.parse(result.stdout);
+      expect(payload.ok).toBe(false);
+      expect(payload.scanned).toBe(0);
+      expect(payload.rejected.classification).toBe("network-or-runtime-error");
+      expect(payload.execution.exitCode).toBe(1);
+      expect(payload.execution.stderrTail).toContain("timeout must not be negative");
+    } finally {
+      rmSync(directory, { recursive: true, force: true });
+    }
+  });
 });