// SPEC: PJ2026-01060309 cross-node-proxy-ci-benchmark draft-2026-06-26-k3s-build-benchmark.
// Responsibility: Generic platform-infra k3s build benchmark coordinator with proxyserver traffic evidence.
import { readFileSync } from "node:fs";
import { rootPath } from "./config";
import { runCommand, type CommandResult } from "./command";
import type { RenderedCliResult } from "./output";
import { egressProxyTrafficScript, type EgressProxyTrafficSpec } from "./egress-proxy-traffic";
import { readSub2ApiConfig } from "./platform-infra/config";
import type { Sub2ApiTargetConfig } from "./platform-infra/entry";
import { resolveTarget } from "./platform-infra/manifest";

const BENCHMARK_CONFIG_PATH = "config/platform-infra/egress-proxy-benchmarks.yaml";
const BENCHMARK_APP = "unidesk-k3s-build-benchmark";
const STAGE_PROXY_EVIDENCE_ANNOTATION = "unidesk.ai/stage-proxy-evidence";
const REAL_DEPS_STAGE_ORDER = ["apk", "npm", "go", "git-mirror"];

type K3sBuildAction = "start" | "status" | "logs" | "cleanup";
type ImagePullPolicy = "Always" | "IfNotPresent" | "Never";
type K3sBuildWorkload = "k3s-build" | "k3s-real-deps";

interface K3sBuildBenchmarkOptions {
  action: K3sBuildAction;
  targets: readonly string[];
  profile: string;
  confirm: boolean;
  dryRun: boolean;
  timeoutSeconds: number;
  tailLines: number;
  trafficSampleSeconds: number;
}

interface K3sBuildBenchmarkProfile {
  id: string;
  enabled: boolean;
  workload: K3sBuildWorkload;
  description: string;
  image: string;
  imagePullPolicy: ImagePullPolicy;
  targetOverrides: Record<string, K3sBuildBenchmarkTargetOverride>;
  payloadMiB: number;
  timeoutSeconds: number;
  ttlSecondsAfterFinished: number;
  noMirror: {
    apt: string;
    npmRegistry: string;
    pipIndexUrl: string;
    registryMirror: "forbidden";
  };
  aptPackages: readonly string[];
  dependencyDownload: {
    enabled: boolean;
    url: string;
    chunks: number;
    expectedMiB: number;
  };
  realDeps?: K3sRealDepsSpec;
}

interface K3sRealDepsSpec {
  minProxyMiB: number;
  imagePullPolicy: ImagePullPolicy;
  apk: {
    image: string;
    packages: readonly string[];
    expectedMiB: number;
  };
  npm: {
    image: string;
    registry: string;
    packages: Record<string, string>;
    expectedMiB: number;
  };
  go: {
    image: string;
    goProxy: string;
    modules: readonly string[];
    expectedMiB: number;
  };
  gitMirror: {
    image: string;
    remote: string;
    expectedMiB: number;
  };
}

interface K3sBuildBenchmarkTargetOverride {
  image?: string;
  imagePullPolicy?: ImagePullPolicy;
}

interface K3sBuildBenchmarkConfig {
  version: number;
  kind: string;
  metadata: { owner: string; relatedIssues: readonly number[] };
  profiles: Record<string, K3sBuildBenchmarkProfile>;
}

interface TargetPlan {
  ok: boolean;
  targetId: string;
  target?: Sub2ApiTargetConfig;
  profile?: K3sBuildBenchmarkProfile;
  blocker?: string;
  detail?: string;
}

interface TargetStatus {
  ok: boolean;
  targetId: string;
  profile: string;
  state: string;
  jobName: string;
  runId: string;
  startedAt: string;
  completedAt: string;
  durationSeconds: number | null;
  outputMiB: number | null;
  downloadMiB: number | null;
  payloadMiB: number | null;
  apkMiB: number | null;
  npmMiB: number | null;
  goMiB: number | null;
  gitMirrorMiB: number | null;
  realDepsMiB: number | null;
  currentStage: string;
  stageEvidence: readonly StageProxyEvidence[];
  failureFamily: string;
  logTail: string;
  traffic?: TrafficSummary;
}

interface StageProxyEvidence {
  stage: string;
  topDestination: string;
  topClient: string;
  windowBytes: number;
  maxRateBps: number;
  proxyCumulativeBytes: number;
  firstObservedAt: string;
  lastObservedAt: string;
  samples: number;
}

interface TrafficSummary {
  ok: boolean;
  reason: string;
  windowBytes: number;
  rateBps: number;
  processTotalBytes: number;
  topClient: string;
  topDestination: string;
}

export function runK3sBuildBenchmarkCommand(args: string[]): RenderedCliResult {
  const options = parseK3sBuildBenchmarkOptions(args);
  const config = readK3sBuildBenchmarkConfig();
  const plans = resolvePlans(config, options);
  if (options.action === "start" && options.dryRun) return renderDryRun(plans, options);
  if (options.action === "start") return startBenchmarks(plans, options);
  if (options.action === "cleanup") return cleanupBenchmarks(plans, options);
  return statusBenchmarks(plans, options);
}

function parseK3sBuildBenchmarkOptions(args: string[]): K3sBuildBenchmarkOptions {
  const first = args[0];
  const action: K3sBuildAction = first === "status" || first === "logs" || first === "cleanup" ? first : "start";
  const rest = action === "start" ? args : args.slice(1);
  if (first === "--help" || first === "-h" || first === "help") {
    throw new Error("platform-infra egress-proxy k3s-build-benchmark usage: k3s-build-benchmark [status|logs|cleanup] --targets D601,D518 --profile real-deps-500m [--dry-run|--confirm]");
  }
  const confirm = rest.includes("--confirm");
  const explicitDryRun = rest.includes("--dry-run");
  if (confirm && explicitDryRun) throw new Error("k3s-build-benchmark accepts only one of --confirm or --dry-run");
  return {
    action,
    targets: parseTargetsOption(rest),
    profile: option(rest, "--profile") ?? "no-mirror-600m",
    confirm,
    dryRun: action === "start" ? explicitDryRun || !confirm : false,
    timeoutSeconds: positiveIntOption(rest, "--timeout-seconds", 60, 60),
    tailLines: positiveIntOption(rest, "--tail-lines", action === "logs" ? 160 : 40, 400),
    trafficSampleSeconds: positiveIntOption(rest, "--traffic-sample-seconds", 0, 45),
  };
}

function parseTargetsOption(args: string[]): readonly string[] {
  const raw = option(args, "--targets") ?? option(args, "--target") ?? "D601";
  if (raw === "all") {
    const sub2api = readSub2ApiConfig();
    return sub2api.targets
      .filter((target) => target.enabled && target.egressProxy?.enabled === true)
      .map((target) => target.id);
  }
  const targets = raw.split(",").map((item) => item.trim()).filter(Boolean);
  if (targets.length === 0) throw new Error("--targets must contain at least one target id");
  return [...new Set(targets)];
}

function resolvePlans(config: K3sBuildBenchmarkConfig, options: K3sBuildBenchmarkOptions): readonly TargetPlan[] {
  const sub2api = readSub2ApiConfig();
  const profile = config.profiles[options.profile];
  return options.targets.map((targetId): TargetPlan => {
    if (profile === undefined) return { ok: false, targetId, blocker: "profile-missing", detail: `${BENCHMARK_CONFIG_PATH}.profiles.${options.profile} is missing` };
    if (!profile.enabled) return { ok: false, targetId, profile, blocker: "profile-disabled", detail: `${BENCHMARK_CONFIG_PATH}.profiles.${profile.id}.enabled=false` };
    if (profile.payloadMiB < 500) return { ok: false, targetId, profile, blocker: "payload-too-small", detail: `${BENCHMARK_CONFIG_PATH}.profiles.${profile.id}.payloadMiB must be >= 500` };
    if (profile.workload === "k3s-real-deps" && profile.realDeps === undefined) return { ok: false, targetId, profile, blocker: "real-deps-missing", detail: `${BENCHMARK_CONFIG_PATH}.profiles.${profile.id}.realDeps is missing` };
    try {
      const target = resolveTarget(sub2api, targetId);
      if (target.egressProxy === null || !target.egressProxy.enabled) return { ok: false, targetId: target.id, target, profile, blocker: "egress-proxy-disabled", detail: `config/platform-infra/sub2api.yaml target ${target.id} has no enabled egressProxy` };
      return { ok: true, targetId: target.id, target, profile };
    } catch (error) {
      return { ok: false, targetId, profile, blocker: "target-missing", detail: error instanceof Error ? error.message : String(error) };
    }
  });
}

function startBenchmarks(plans: readonly TargetPlan[], options: K3sBuildBenchmarkOptions): RenderedCliResult {
  const rows = plans.map((plan) => {
    if (!plan.ok || plan.target === undefined || plan.profile === undefined) {
      return { ...plan, started: false, state: "blocked", jobName: "-", runId: "-", result: null };
    }
    const runId = `k3sbuild-${Date.now().toString(36)}-${plan.target.id.toLowerCase()}`;
    const jobName = benchmarkJobName(plan.target, plan.profile, runId);
    const manifest = benchmarkJobManifest(plan.target, plan.profile, runId, jobName);
    const result = runTrans(plan.target.route, startScript(manifest, plan.target, plan.profile, runId, jobName), options.timeoutSeconds);
    const parsed = parseJson(result.stdout);
    return {
      ...plan,
      started: result.exitCode === 0,
      state: result.exitCode === 0 ? "started" : "failed",
      jobName,
      runId,
      result: typeof parsed === "object" && parsed !== null ? parsed : { stdoutPreview: result.stdout.slice(0, 1200), stderrPreview: result.stderr.slice(-1200), exitCode: result.exitCode },
    };
  });
  const ok = rows.every((row) => row.ok && row.started);
  const tableRows = rows.map((row) => [
    row.targetId,
    row.profile?.id ?? options.profile,
    row.state,
    row.jobName,
    row.runId,
    startRowDetail(row),
  ]);
  return {
    ok,
    command: "platform-infra egress-proxy k3s-build-benchmark",
    contentType: "text/plain",
    renderedText: [
      "PLATFORM-INFRA K3S BUILD BENCHMARK START",
      "",
      ...table(["TARGET", "PROFILE", "STATE", "JOB", "RUN", "NEXT"], tableRows),
      "",
      "NEXT",
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark status --targets ${rows.map((row) => row.targetId).join(",")} --profile ${options.profile}`,
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark status --targets ${rows.map((row) => row.targetId).join(",")} --profile ${options.profile} --traffic-sample-seconds 15`,
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark logs --targets ${rows.map((row) => row.targetId).join(",")} --profile ${options.profile}`,
      "",
      "Disclosure: start is fire-and-forget; the k3s Job uses emptyDir and a unique Job name, so benchmark work is not reused.",
    ].join("\n"),
  };
}

function startRowDetail(row: TargetPlan & { started: boolean; state: string; jobName: string; runId: string; result: unknown }): string {
  if (!row.ok) return `${row.blocker}: ${row.detail}`;
  if (row.started) {
    const replaced = record(row.result).replaced;
    const recorder = text(record(row.result).stageRecorder, "");
    const prefix = replaced === undefined || replaced === 0 ? "status/logs/traffic" : `status/logs/traffic replaced=${replaced}`;
    return recorder === "" || recorder === "disabled" ? prefix : `${prefix} stage-recorder=${recorder}`;
  }
  const result = record(row.result);
  const stderr = text(result.stderrPreview, "");
  const stdout = text(result.stdoutPreview, "");
  const exitCode = text(result.exitCode, "-");
  const detail = stderr || stdout || "start-failed";
  return `exit=${exitCode} ${detail.replace(/\s+/gu, " ").slice(0, 180)}`;
}

function statusBenchmarks(plans: readonly TargetPlan[], options: K3sBuildBenchmarkOptions): RenderedCliResult {
  const statuses = plans.map((plan): TargetStatus => {
    if (!plan.ok || plan.target === undefined || plan.profile === undefined) {
      return blockedStatus(plan, options.profile);
    }
    const result = runTrans(plan.target.route, statusScript(plan.target, plan.profile, options.tailLines), options.timeoutSeconds);
    const parsed = parseJson(result.stdout);
    const status = normalizeStatus(plan, parsed, result);
    if (options.trafficSampleSeconds > 0) status.traffic = sampleTraffic(plan.target, options.trafficSampleSeconds, options.timeoutSeconds);
    return status;
  });
  const ok = statuses.every((status) => status.ok || options.action === "logs");
  const rows = statuses.map((status) => [
    status.targetId,
    status.profile,
    status.state,
    status.currentStage,
    status.jobName,
    status.durationSeconds === null ? "-" : `${status.durationSeconds}s`,
    status.outputMiB === null ? "-" : `${status.outputMiB}MiB`,
    status.downloadMiB === null ? "-" : `${status.downloadMiB}MiB`,
    status.apkMiB === null ? "-" : `${status.apkMiB}MiB`,
    status.npmMiB === null ? "-" : `${status.npmMiB}MiB`,
    status.goMiB === null ? "-" : `${status.goMiB}MiB`,
    status.gitMirrorMiB === null ? "-" : `${status.gitMirrorMiB}MiB`,
    status.realDepsMiB === null ? "-" : `${status.realDepsMiB}MiB`,
    stageEvidenceSummary(status.stageEvidence),
    status.traffic === undefined ? "-" : bytes(status.traffic.windowBytes),
    status.traffic === undefined ? "-" : rate(status.traffic.rateBps),
    status.traffic === undefined ? "-" : bytes(status.traffic.processTotalBytes),
    status.traffic === undefined ? "-" : status.traffic.topClient,
    status.traffic === undefined ? "-" : status.traffic.topDestination,
    status.failureFamily,
  ]);
  const logSections = options.action === "logs"
    ? statuses.flatMap((status) => ["", `LOG ${status.targetId} ${status.jobName}`, status.logTail || "-"])
    : [];
  return {
    ok,
    command: `platform-infra egress-proxy k3s-build-benchmark ${options.action}`,
    contentType: "text/plain",
    renderedText: [
      "PLATFORM-INFRA K3S BUILD BENCHMARK STATUS",
      "",
      ...table(["TARGET", "PROFILE", "STATE", "STAGE", "JOB", "DURATION", "OUTPUT", "DOWNLOAD", "APK", "NPM", "GO", "GIT_MIRROR", "REAL_DEPS", "STAGE_PROXY", "TRAFFIC_WINDOW", "TRAFFIC_RATE", "PROXY_CUM", "TOP_CLIENT", "TOP_DEST", "FAILURE"], rows),
      ...stageEvidenceSections(statuses),
      ...logSections,
      "",
      "NEXT",
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark status --targets ${statuses.map((status) => status.targetId).join(",")} --profile ${options.profile} --traffic-sample-seconds 15`,
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark logs --targets ${statuses.map((status) => status.targetId).join(",")} --profile ${options.profile}`,
      "",
      "Disclosure: traffic columns are proxyserver-side samples only when --traffic-sample-seconds is set; STAGE_PROXY_EVIDENCE is persisted by the benchmark stage recorder in Job annotations; Secret/proxy values are redacted.",
    ].join("\n"),
  };
}

function cleanupBenchmarks(plans: readonly TargetPlan[], options: K3sBuildBenchmarkOptions): RenderedCliResult {
  const rows = plans.map((plan) => {
    if (!plan.ok || plan.target === undefined || plan.profile === undefined) {
      return { targetId: plan.targetId, profile: options.profile, state: "blocked", deleted: "-", detail: plan.detail ?? plan.blocker ?? "blocked" };
    }
    if (!options.confirm) return { targetId: plan.targetId, profile: plan.profile.id, state: "dry-run", deleted: "-", detail: "pass --confirm to delete matching benchmark Jobs" };
    const result = runTrans(plan.target.route, cleanupScript(plan.target, plan.profile), options.timeoutSeconds);
    const parsed = parseJson(result.stdout);
    const data = record(parsed);
    return {
      targetId: plan.targetId,
      profile: plan.profile.id,
      state: result.exitCode === 0 && data.ok !== false ? "deleted" : "failed",
      deleted: text(data.deleted, "-"),
      detail: text(data.detail, result.stderr.slice(-1000) || result.stdout.slice(-1000)),
    };
  });
  return {
    ok: rows.every((row) => row.state === "deleted" || row.state === "dry-run"),
    command: "platform-infra egress-proxy k3s-build-benchmark cleanup",
    contentType: "text/plain",
    renderedText: [
      "PLATFORM-INFRA K3S BUILD BENCHMARK CLEANUP",
      "",
      ...table(["TARGET", "PROFILE", "STATE", "DELETED", "DETAIL"], rows.map((row) => [row.targetId, row.profile, row.state, row.deleted, row.detail])),
      "",
      "NEXT",
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark --targets ${rows.map((row) => row.targetId).join(",")} --profile ${options.profile} --confirm`,
    ].join("\n"),
  };
}

function renderDryRun(plans: readonly TargetPlan[], options: K3sBuildBenchmarkOptions): RenderedCliResult {
  const rows = plans.map((plan) => [
    plan.targetId,
    plan.profile?.id ?? options.profile,
    plan.ok ? "ok" : `blocked:${plan.blocker}`,
    plan.target?.route ?? "-",
    plan.target?.namespace ?? "-",
    plan.target?.egressProxy?.serviceName ?? "-",
    plan.profile !== undefined ? dryRunImages(plan.profile) : "-",
    plan.profile !== undefined ? dryRunPullPolicy(plan.profile) : "-",
    plan.profile === undefined ? "-" : `${plan.profile.payloadMiB}MiB`,
    plan.profile === undefined ? "-" : dependencySummary(plan.profile),
    plan.detail ?? dryRunDetail(plan.profile),
  ]);
  return {
    ok: plans.every((plan) => plan.ok),
    command: "platform-infra egress-proxy k3s-build-benchmark",
    contentType: "text/plain",
    renderedText: [
      "PLATFORM-INFRA K3S BUILD BENCHMARK DRY-RUN",
      "",
      ...table(["TARGET", "PROFILE", "STATUS", "ROUTE", "NAMESPACE", "PROXY", "IMAGES", "PULL", "MIN_PROXY", "DEPS", "DETAIL"], rows),
      "",
      "NEXT",
      `  bun scripts/cli.ts platform-infra egress-proxy k3s-build-benchmark --targets ${plans.map((plan) => plan.targetId).join(",")} --profile ${options.profile} --confirm`,
      "",
      `Config: ${BENCHMARK_CONFIG_PATH}`,
    ].join("\n"),
  };
}

function dryRunImages(profile: K3sBuildBenchmarkProfile): string {
  if (profile.workload === "k3s-real-deps" && profile.realDeps !== undefined) {
    return [profile.realDeps.apk.image, profile.realDeps.npm.image, profile.realDeps.go.image, profile.realDeps.gitMirror.image].join(",");
  }
  return profile.image;
}

function dryRunPullPolicy(profile: K3sBuildBenchmarkProfile): string {
  if (profile.workload === "k3s-real-deps" && profile.realDeps !== undefined) return profile.realDeps.imagePullPolicy;
  return profile.imagePullPolicy;
}

function dependencySummary(profile: K3sBuildBenchmarkProfile): string {
  if (profile.workload === "k3s-real-deps" && profile.realDeps !== undefined) {
    return `apk~${profile.realDeps.apk.expectedMiB} npm~${profile.realDeps.npm.expectedMiB} go~${profile.realDeps.go.expectedMiB} gitMirror~${profile.realDeps.gitMirror.expectedMiB}`;
  }
  return `${profile.dependencyDownload.expectedMiB}MiB`;
}

function dryRunDetail(profile: K3sBuildBenchmarkProfile | undefined): string {
  if (profile?.workload === "k3s-real-deps") return "kubelet-image-pull + apk/npm/go/git-mirror through proxy";
  return "no-mirror emptyDir unique-job";
}

function benchmarkJobManifest(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, runId: string, jobName: string): Record<string, unknown> {
  if (profile.workload === "k3s-real-deps") return realDepsJobManifest(target, profile, runId, jobName);
  const proxy = target.egressProxy;
  if (proxy === null) throw new Error(`target ${target.id} has no egressProxy`);
  const proxyUrl = `http://${proxy.serviceName}.${target.namespace}.svc.cluster.local:${proxy.listenPort}`;
  const noProxy = proxy.noProxy.join(",");
  const image = effectiveImage(target, profile);
  return {
    apiVersion: "batch/v1",
    kind: "Job",
    metadata: {
      name: jobName,
      namespace: target.namespace,
      labels: benchmarkLabels(target, profile, runId),
      annotations: {
        "unidesk.ai/no-mirror": JSON.stringify(profile.noMirror),
        "unidesk.ai/payload-mib": String(profile.payloadMiB),
        "unidesk.ai/dependency-download-mib": String(profile.dependencyDownload.expectedMiB),
        "unidesk.ai/bootstrap-image": image.image,
        "unidesk.ai/bootstrap-image-pull-policy": image.imagePullPolicy,
      },
    },
    spec: {
      backoffLimit: 0,
      activeDeadlineSeconds: profile.timeoutSeconds,
      ttlSecondsAfterFinished: profile.ttlSecondsAfterFinished,
      template: {
        metadata: { labels: benchmarkLabels(target, profile, runId) },
        spec: {
          restartPolicy: "Never",
          containers: [{
            name: "build",
            image: image.image,
            imagePullPolicy: image.imagePullPolicy,
            command: ["/bin/sh", "-lc"],
            args: [workloadScript(profile)],
            env: [
              { name: "HTTP_PROXY", value: proxyUrl },
              { name: "HTTPS_PROXY", value: proxyUrl },
              { name: "ALL_PROXY", value: proxyUrl },
              { name: "http_proxy", value: proxyUrl },
              { name: "https_proxy", value: proxyUrl },
              { name: "all_proxy", value: proxyUrl },
              { name: "NO_PROXY", value: noProxy },
              { name: "no_proxy", value: noProxy },
              { name: "DEBIAN_FRONTEND", value: "noninteractive" },
              { name: "NPM_CONFIG_REGISTRY", value: profile.noMirror.npmRegistry },
              { name: "PIP_INDEX_URL", value: profile.noMirror.pipIndexUrl },
              { name: "BENCHMARK_TARGET", value: target.id },
              { name: "BENCHMARK_PROFILE", value: profile.id },
              { name: "BENCHMARK_RUN_ID", value: runId },
              { name: "PAYLOAD_MIB", value: String(profile.payloadMiB) },
              { name: "DOWNLOAD_URL", value: profile.dependencyDownload.url },
              { name: "DOWNLOAD_CHUNKS", value: String(profile.dependencyDownload.chunks) },
              { name: "DOWNLOAD_EXPECTED_MIB", value: String(profile.dependencyDownload.expectedMiB) },
              { name: "APT_PACKAGES", value: profile.aptPackages.join(" ") },
            ],
            volumeMounts: [{ name: "work", mountPath: "/work" }],
          }],
          volumes: [{ name: "work", emptyDir: { sizeLimit: "3Gi" } }],
        },
      },
    },
  };
}

function realDepsJobManifest(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, runId: string, jobName: string): Record<string, unknown> {
  const proxy = target.egressProxy;
  if (proxy === null) throw new Error(`target ${target.id} has no egressProxy`);
  const realDeps = requireRealDeps(profile);
  const proxyUrl = `http://${proxy.serviceName}.${target.namespace}.svc.cluster.local:${proxy.listenPort}`;
  const noProxy = proxy.noProxy.join(",");
  const labels = benchmarkLabels(target, profile, runId);
  const commonEnv = [
    { name: "HTTP_PROXY", value: proxyUrl },
    { name: "HTTPS_PROXY", value: proxyUrl },
    { name: "ALL_PROXY", value: proxyUrl },
    { name: "http_proxy", value: proxyUrl },
    { name: "https_proxy", value: proxyUrl },
    { name: "all_proxy", value: proxyUrl },
    { name: "NO_PROXY", value: noProxy },
    { name: "no_proxy", value: noProxy },
    { name: "BENCHMARK_TARGET", value: target.id },
    { name: "BENCHMARK_PROFILE", value: profile.id },
    { name: "BENCHMARK_RUN_ID", value: runId },
    { name: "MIN_PROXY_MIB", value: String(realDeps.minProxyMiB) },
  ];
  return {
    apiVersion: "batch/v1",
    kind: "Job",
    metadata: {
      name: jobName,
      namespace: target.namespace,
      labels,
      annotations: {
        "unidesk.ai/workload": profile.workload,
        "unidesk.ai/min-proxy-mib": String(realDeps.minProxyMiB),
        "unidesk.ai/image-pull-mode": "kubelet-containerd",
        "unidesk.ai/remote-images": [realDeps.apk.image, realDeps.npm.image, realDeps.go.image, realDeps.gitMirror.image].join(","),
        "unidesk.ai/git-mirror-remote": realDeps.gitMirror.remote,
      },
    },
    spec: {
      backoffLimit: 0,
      activeDeadlineSeconds: profile.timeoutSeconds,
      ttlSecondsAfterFinished: profile.ttlSecondsAfterFinished,
      template: {
        metadata: { labels },
        spec: {
          restartPolicy: "Never",
          initContainers: [
            {
              name: "apk-add",
              image: realDeps.apk.image,
              imagePullPolicy: realDeps.imagePullPolicy,
              command: ["/bin/sh", "-lc"],
              args: [realDepsApkScript()],
              env: [
                ...commonEnv,
                { name: "APK_PACKAGES", value: realDeps.apk.packages.join(" ") },
                { name: "APK_IMAGE", value: realDeps.apk.image },
              ],
              volumeMounts: [{ name: "work", mountPath: "/work" }],
            },
            {
              name: "npm-install",
              image: realDeps.npm.image,
              imagePullPolicy: realDeps.imagePullPolicy,
              command: ["/bin/sh", "-lc"],
              args: [realDepsNpmScript(realDeps)],
              env: [
                ...commonEnv,
                { name: "NPM_CONFIG_REGISTRY", value: realDeps.npm.registry },
                { name: "NPM_IMAGE", value: realDeps.npm.image },
              ],
              volumeMounts: [{ name: "work", mountPath: "/work" }],
            },
            {
              name: "go-download",
              image: realDeps.go.image,
              imagePullPolicy: realDeps.imagePullPolicy,
              command: ["/bin/sh", "-lc"],
              args: [realDepsGoScript()],
              env: [
                ...commonEnv,
                { name: "GOPROXY", value: realDeps.go.goProxy },
                { name: "GO_MODULES", value: realDeps.go.modules.join(" ") },
                { name: "GO_IMAGE", value: realDeps.go.image },
              ],
              volumeMounts: [{ name: "work", mountPath: "/work" }],
            },
            {
              name: "git-mirror",
              image: realDeps.gitMirror.image,
              imagePullPolicy: realDeps.imagePullPolicy,
              command: ["/bin/sh", "-lc"],
              args: [realDepsGitMirrorScript()],
              env: [
                ...commonEnv,
                { name: "GIT_MIRROR_IMAGE", value: realDeps.gitMirror.image },
                { name: "GIT_MIRROR_REMOTE", value: realDeps.gitMirror.remote },
              ],
              volumeMounts: [{ name: "work", mountPath: "/work" }],
            },
          ],
          containers: [{
            name: "summary",
            image: realDeps.apk.image,
            imagePullPolicy: "IfNotPresent",
            command: ["/bin/sh", "-lc"],
            args: [realDepsSummaryScript(realDeps)],
            env: commonEnv,
            volumeMounts: [{ name: "work", mountPath: "/work" }],
          }],
          volumes: [{ name: "work", emptyDir: { sizeLimit: "6Gi" } }],
        },
      },
    },
  };
}

function requireRealDeps(profile: K3sBuildBenchmarkProfile): K3sRealDepsSpec {
  if (profile.realDeps === undefined) throw new Error(`profiles.${profile.id}.realDeps is required for workload=${profile.workload}`);
  return profile.realDeps;
}

function realDepsApkScript(): string {
  return `set -eu
mkdir -p /work/stages
started_epoch="$(date +%s)"
started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
printf 'UNIDESK_K3S_REAL_DEPS_EVENT stage=apk target=%s profile=%s run=%s image=%s packages="%s"\\n' "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$APK_IMAGE" "$APK_PACKAGES"
if grep -R -E 'npmmirror|daocloud|aliyun|tuna|ustc|huaweicloud' /etc/apk/repositories >/tmp/apk-mirror-check.out 2>/dev/null; then
  cat /tmp/apk-mirror-check.out >&2
  echo "unexpected apk mirror in base image" >&2
  exit 42
fi
apk update
apk add --no-cache $APK_PACKAGES
apk_mib="$(du -sk /usr/bin /usr/lib /usr/include /usr/local 2>/dev/null | awk '{s+=$1} END {printf "%d", int((s+1023)/1024)}')"
{
  printf 'realDepsStartedEpoch=%s\\n' "$started_epoch"
  printf 'realDepsStartedAt=%s\\n' "$started_at"
  printf 'apkMiB=%s\\n' "$apk_mib"
} > /work/stages/apk.env
printf 'UNIDESK_K3S_REAL_DEPS_STAGE {"stage":"apk","ok":true,"image":"%s","installedMiB":%s}\\n' "$APK_IMAGE" "$apk_mib"
`;
}

function realDepsNpmScript(realDeps: K3sRealDepsSpec): string {
  const packageJson = JSON.stringify({ private: true, dependencies: realDeps.npm.packages }, null, 2);
  return `set -eu
mkdir -p /work/stages /work/npm/project /work/npm/cache
cd /work/npm/project
cat > package.json <<'JSON'
${packageJson}
JSON
printf 'UNIDESK_K3S_REAL_DEPS_EVENT stage=npm target=%s profile=%s run=%s image=%s registry=%s\\n' "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$NPM_IMAGE" "$NPM_CONFIG_REGISTRY"
npm install --ignore-scripts --no-audit --no-fund --cache /work/npm/cache --registry "$NPM_CONFIG_REGISTRY"
npm_mib="$(du -sk /work/npm/cache /work/npm/project/node_modules 2>/dev/null | awk '{s+=$1} END {printf "%d", int((s+1023)/1024)}')"
printf 'npmMiB=%s\\n' "$npm_mib" > /work/stages/npm.env
printf 'UNIDESK_K3S_REAL_DEPS_STAGE {"stage":"npm","ok":true,"image":"%s","installedMiB":%s}\\n' "$NPM_IMAGE" "$npm_mib"
`;
}

function realDepsGoScript(): string {
  return `set -eu
mkdir -p /work/stages /work/go/module /work/go/gopath /work/go/gomodcache
cd /work/go/module
if ! command -v go >/dev/null 2>&1 && [ -x /usr/local/go/bin/go ]; then
  export PATH="/usr/local/go/bin:\${PATH:-/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin}"
fi
if ! command -v go >/dev/null 2>&1; then
  echo "go-runtime-missing PATH=\${PATH:-}" >&2
  exit 127
fi
go version
go mod init unidesk.local/proxy-benchmark
printf 'UNIDESK_K3S_REAL_DEPS_EVENT stage=go target=%s profile=%s run=%s image=%s goproxy=%s modules="%s"\\n' "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$GO_IMAGE" "$GOPROXY" "$GO_MODULES"
export GOPATH=/work/go/gopath
export GOMODCACHE=/work/go/gomodcache
for module in $GO_MODULES; do
  go get "$module"
done
go mod download -x all
go_mib="$(du -sk /work/go/gomodcache /work/go/gopath/pkg 2>/dev/null | awk '{s+=$1} END {printf "%d", int((s+1023)/1024)}')"
printf 'goMiB=%s\\n' "$go_mib" > /work/stages/go.env
printf 'UNIDESK_K3S_REAL_DEPS_STAGE {"stage":"go","ok":true,"image":"%s","downloadedMiB":%s}\\n' "$GO_IMAGE" "$go_mib"
`;
}

function realDepsGitMirrorScript(): string {
  return `set -eu
mkdir -p /work/stages /work/git-mirror
printf 'UNIDESK_K3S_REAL_DEPS_EVENT stage=git-mirror target=%s profile=%s run=%s image=%s remote=%s\\n' "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$GIT_MIRROR_IMAGE" "$GIT_MIRROR_REMOTE"
if ! command -v git >/dev/null 2>&1; then
  if ! command -v apk >/dev/null 2>&1; then
    echo "git-runtime-missing image=$GIT_MIRROR_IMAGE" >&2
    exit 127
  fi
  if grep -R -E 'npmmirror|daocloud|aliyun|tuna|ustc|huaweicloud' /etc/apk/repositories >/tmp/git-apk-mirror-check.out 2>/dev/null; then
    cat /tmp/git-apk-mirror-check.out >&2
    echo "unexpected apk mirror in git mirror base image" >&2
    exit 42
  fi
  apk update
  apk add --no-cache git ca-certificates
fi
git --version
git_proxy() {
  git -c "http.proxy=\${HTTP_PROXY:-}" -c "https.proxy=\${HTTPS_PROXY:-}" -c http.version=HTTP/1.1 -c http.lowSpeedLimit=1024 -c http.lowSpeedTime=120 "$@"
}
attempt=1
git_ok=0
while [ "$attempt" -le 3 ]; do
  rm -rf /work/git-mirror/repo.git /work/git-mirror/ls-remote-heads.txt
  if git_proxy clone --mirror --filter=blob:none "$GIT_MIRROR_REMOTE" /work/git-mirror/repo.git \\
    && git_proxy --git-dir=/work/git-mirror/repo.git remote update --prune \\
    && git_proxy --git-dir=/work/git-mirror/repo.git remote -v \\
    && git_proxy ls-remote --heads "$GIT_MIRROR_REMOTE" >/work/git-mirror/ls-remote-heads.txt; then
    git_ok=1
    break
  fi
  echo "git-mirror-attempt-failed attempt=$attempt remote=$GIT_MIRROR_REMOTE" >&2
  attempt=$((attempt + 1))
  sleep $((attempt * 5))
done
if [ "$git_ok" != "1" ]; then
  echo "git-mirror-failed-after-retries remote=$GIT_MIRROR_REMOTE" >&2
  exit 44
fi
git_mib="$(du -sk /work/git-mirror/repo.git /work/git-mirror/ls-remote-heads.txt 2>/dev/null | awk '{s+=$1} END {printf "%d", int((s+1023)/1024)}')"
printf 'gitMirrorMiB=%s\\n' "$git_mib" > /work/stages/git-mirror.env
printf 'UNIDESK_K3S_REAL_DEPS_STAGE {"stage":"git-mirror","ok":true,"image":"%s","remote":"%s","mirrorMiB":%s}\\n' "$GIT_MIRROR_IMAGE" "$GIT_MIRROR_REMOTE" "$git_mib"
`;
}

function realDepsSummaryScript(realDeps: K3sRealDepsSpec): string {
  return `set -eu
apkMiB=0
npmMiB=0
goMiB=0
gitMirrorMiB=0
realDepsStartedEpoch="$(date +%s)"
realDepsStartedAt="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
[ -f /work/stages/apk.env ] && . /work/stages/apk.env
[ -f /work/stages/npm.env ] && . /work/stages/npm.env
[ -f /work/stages/go.env ] && . /work/stages/go.env
[ -f /work/stages/git-mirror.env ] && . /work/stages/git-mirror.env
completed_epoch="$(date +%s)"
completed_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
real_deps_mib=$((apkMiB + npmMiB + goMiB + gitMirrorMiB))
duration=$((completed_epoch - realDepsStartedEpoch))
pod_ip="$(hostname -i 2>/dev/null | awk '{print $1}')"
printf 'UNIDESK_K3S_REAL_DEPS_RESULT {"ok":true,"target":"%s","profile":"%s","runId":"%s","startedAt":"%s","completedAt":"%s","durationSeconds":%s,"podIp":"%s","apkMiB":%s,"npmMiB":%s,"goMiB":%s,"gitMirrorMiB":%s,"realDepsMiB":%s,"minProxyMiB":%s,"imagePullMode":"kubelet-containerd","apkImage":"%s","npmImage":"%s","goImage":"%s","gitMirrorImage":"%s","gitMirrorRemote":"%s"}\\n' \\
  "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$realDepsStartedAt" "$completed_at" "$duration" "$pod_ip" "$apkMiB" "$npmMiB" "$goMiB" "$gitMirrorMiB" "$real_deps_mib" "$MIN_PROXY_MIB" ${shQuote(realDeps.apk.image)} ${shQuote(realDeps.npm.image)} ${shQuote(realDeps.go.image)} ${shQuote(realDeps.gitMirror.image)} ${shQuote(realDeps.gitMirror.remote)}
`;
}

function workloadScript(profile: K3sBuildBenchmarkProfile): string {
  return `set -eu
started_epoch="$(date +%s)"
started_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
export BENCHMARK_STARTED_EPOCH="$started_epoch"
export BENCHMARK_STARTED_AT="$started_at"
work=/work/k3s-build-benchmark
download_dir="$work/download"
build_dir="$work/build"
output_dir="$work/output"
mkdir -p "$download_dir" "$build_dir" "$output_dir"
printf 'UNIDESK_K3S_BUILD_BENCHMARK_EVENT target=%s profile=%s run=%s payloadMiB=%s expectedDownloadMiB=%s noMirror=true\\n' "$BENCHMARK_TARGET" "$BENCHMARK_PROFILE" "$BENCHMARK_RUN_ID" "$PAYLOAD_MIB" "$DOWNLOAD_EXPECTED_MIB"
if command -v apt-get >/dev/null 2>&1 && [ -n "$APT_PACKAGES" ]; then
  if grep -R -E 'npmmirror|daocloud|aliyun|tuna|ustc' /etc/apt/sources.list /etc/apt/sources.list.d >/tmp/mirror-check.out 2>/dev/null; then
    cat /tmp/mirror-check.out >&2
    echo "unexpected apt mirror in base image" >&2
    exit 42
  fi
  apt-get -o Acquire::http::No-Cache=true -o Acquire::https::No-Cache=true update
  apt-get -o Acquire::http::No-Cache=true -o Acquire::https::No-Cache=true install -y --no-install-recommends $APT_PACKAGES
  cat > "$build_dir/bench.c" <<'C'
#include <stdint.h>
#include <stdio.h>
int main(void) {
  uint64_t x = 1469598103934665603ULL;
  for (uint64_t i = 0; i < 12000000ULL; ++i) {
    x ^= i;
    x *= 1099511628211ULL;
  }
  printf("%llu\\n", (unsigned long long)x);
  return 0;
}
C
  cc -O2 "$build_dir/bench.c" -o "$build_dir/bench"
  "$build_dir/bench" > "$output_dir/compile-result.txt"
fi
pybin=""
if command -v python3 >/dev/null 2>&1; then pybin=python3; elif command -v python >/dev/null 2>&1; then pybin=python; fi
if [ -z "$pybin" ]; then
  echo "python-runtime-missing" >&2
  exit 44
fi
"$pybin" - <<'PY'
import hashlib
import json
import math
import os
import pathlib
import shutil
import sys
import time
import urllib.request

started_epoch = int(os.environ.get("BENCHMARK_STARTED_EPOCH", "0") or "0") or int(time.time())
started_at = os.environ.get("BENCHMARK_STARTED_AT") or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started_epoch))
payload_mib = int(os.environ["PAYLOAD_MIB"])
download_chunks = int(os.environ["DOWNLOAD_CHUNKS"])
download_expected_mib = int(os.environ["DOWNLOAD_EXPECTED_MIB"])
download_url = os.environ["DOWNLOAD_URL"]
download_enabled = "${profile.dependencyDownload.enabled ? "1" : "0"}" == "1"
download_dir = pathlib.Path("/work/k3s-build-benchmark/download")
build_dir = pathlib.Path("/work/k3s-build-benchmark/build")
output_dir = pathlib.Path("/work/k3s-build-benchmark/output")
download_dir.mkdir(parents=True, exist_ok=True)
build_dir.mkdir(parents=True, exist_ok=True)
output_dir.mkdir(parents=True, exist_ok=True)

def mib_from_bytes(value):
    return int(math.ceil(value / 1048576))

download_bytes = 0
if download_enabled:
    for index in range(1, download_chunks + 1):
        destination = download_dir / f"chunk-{index}.bin"
        request = urllib.request.Request(download_url, headers={"User-Agent": "curl/8.5.0", "Accept": "*/*"})
        with urllib.request.urlopen(request, timeout=240) as response, destination.open("wb") as handle:
            while True:
                block = response.read(1024 * 1024)
                if not block:
                    break
                handle.write(block)
                download_bytes += len(block)
        print(f"UNIDESK_K3S_BUILD_BENCHMARK_DOWNLOAD chunk={index} bytes={destination.stat().st_size}", flush=True)

source_file = build_dir / "portable_build_source.py"
source_file.write_text("VALUE = 'unidesk-k3s-build-benchmark'\\n", encoding="utf-8")
__import__("py_compile").compile(str(source_file), cfile=str(build_dir / "portable_build_source.pyc"), doraise=True)

payload = output_dir / "payload.bin"
digest = hashlib.sha256()
seed = hashlib.sha256(f"{os.environ['BENCHMARK_TARGET']}:{os.environ['BENCHMARK_RUN_ID']}".encode("utf-8")).digest()
block = (seed * ((1024 * 1024 // len(seed)) + 1))[:1024 * 1024]
with payload.open("wb") as handle:
    for index in range(payload_mib):
        digest.update(block)
        handle.write(block)
        if index % 64 == 0:
            handle.flush()

(output_dir / "payload.sha256").write_text(digest.hexdigest() + "  payload.bin\\n", encoding="utf-8")
shutil.rmtree(download_dir, ignore_errors=True)
output_bytes = sum(path.stat().st_size for path in output_dir.rglob("*") if path.is_file())
output_mib = mib_from_bytes(output_bytes)
download_mib = mib_from_bytes(download_bytes)
if output_mib < 500:
    print(f"payload-too-small outputMiB={output_mib}", file=sys.stderr)
    sys.exit(43)
completed_epoch = int(time.time())
result = {
    "ok": True,
    "target": os.environ["BENCHMARK_TARGET"],
    "profile": os.environ["BENCHMARK_PROFILE"],
    "runId": os.environ["BENCHMARK_RUN_ID"],
    "startedAt": started_at,
    "completedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(completed_epoch)),
    "durationSeconds": completed_epoch - started_epoch,
    "payloadMiB": payload_mib,
    "downloadMiB": download_mib,
    "downloadExpectedMiB": download_expected_mib,
    "outputMiB": output_mib,
    "noMirror": True,
    "packageManager": "none" if not os.environ.get("APT_PACKAGES") else "apt",
    "aptMirror": "not-used" if not os.environ.get("APT_PACKAGES") else "system-default",
    "npmRegistry": os.environ["NPM_CONFIG_REGISTRY"],
    "pipIndexUrl": os.environ["PIP_INDEX_URL"],
}
print("UNIDESK_K3S_BUILD_BENCHMARK_RESULT " + json.dumps(result, sort_keys=True), flush=True)
PY
`;
}

function effectiveImage(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile): { image: string; imagePullPolicy: ImagePullPolicy } {
  const override = profile.targetOverrides[target.id] ?? profile.targetOverrides[target.id.toLowerCase()] ?? profile.targetOverrides[target.id.toUpperCase()];
  return {
    image: override?.image ?? profile.image,
    imagePullPolicy: override?.imagePullPolicy ?? profile.imagePullPolicy,
  };
}

function benchmarkLabels(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, runId: string): Record<string, string> {
  return {
    "app.kubernetes.io/name": BENCHMARK_APP,
    "app.kubernetes.io/part-of": "platform-infra",
    "app.kubernetes.io/managed-by": "unidesk",
    "unidesk.ai/benchmark": "k3s-build",
    "unidesk.ai/benchmark-profile": profile.id,
    "unidesk.ai/runtime-node": target.id.toLowerCase(),
    "unidesk.ai/run-id": runId,
  };
}

function startScript(manifest: Record<string, unknown>, target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, runId: string, jobName: string): string {
  const yaml = `${JSON.stringify(manifest, null, 2)}\n`;
  const encoded = Buffer.from(yaml, "utf8").toString("base64");
  const proxy = target.egressProxy;
  const proxySelector = proxy === null ? "" : `app.kubernetes.io/name=${proxy.deploymentName},app.kubernetes.io/component=egress-proxy`;
  const recorderEnabled = profile.workload === "k3s-real-deps" && proxy !== null;
  return `
set -eu
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
manifest="$tmp/k3s-build-benchmark.yaml"
printf '%s' '${encoded}' | base64 -d > "$manifest"
selector="app.kubernetes.io/name=${BENCHMARK_APP},unidesk.ai/benchmark-profile=${profile.id},unidesk.ai/runtime-node=${target.id.toLowerCase()}"
# Single-active guard: delete any prior benchmark Jobs for this target+profile before applying the new
# one, so a new run cannot race with a stale/slow prior run and make status report the wrong Job.
# The benchmark is fire-and-forget and uses emptyDir, so deleting the prior Job is the safe way to keep
# exactly one active run per target.
replaced="$(kubectl -n ${shQuote(target.namespace)} get jobs -l "$selector" --no-headers 2>/dev/null | wc -l | tr -d ' ')"
[ -z "$replaced" ] && replaced=0
if [ "$replaced" != "0" ]; then
  kubectl -n ${shQuote(target.namespace)} delete jobs -l "$selector" --ignore-not-found >/dev/null 2>&1
fi
kubectl apply -f "$manifest" >/dev/null
stage_recorder="disabled"
recorder_pid=""
if [ ${recorderEnabled ? "1" : "0"} = 1 ]; then
  recorder_dir="/tmp/unidesk-k3s-build-benchmark"
  mkdir -p "$recorder_dir"
  recorder_script="$recorder_dir/${jobName}.stage-recorder.py"
  recorder_log="$recorder_dir/${jobName}.stage-recorder.log"
  cat > "$recorder_script" <<'PY'
${stageProxyRecorderPython()}
PY
  nohup python3 "$recorder_script" ${shQuote(target.namespace)} ${shQuote(jobName)} ${shQuote(proxySelector)} ${shQuote(String(profile.timeoutSeconds))} 1 >"$recorder_log" 2>&1 &
  recorder_pid="$!"
  stage_recorder="started"
fi
python3 - ${shQuote(jobName)} ${shQuote(target.namespace)} ${shQuote(target.id)} ${shQuote(runId)} ${shQuote(profile.id)} "$replaced" "$stage_recorder" "$recorder_pid" <<'PY'
import json, sys
job_name, namespace, target, run_id, profile, replaced, stage_recorder, recorder_pid = sys.argv[1:9]
print(json.dumps({
  "ok": True,
  "jobName": job_name,
  "namespace": namespace,
  "target": target,
  "runId": run_id,
  "profile": profile,
  "replaced": int(replaced),
  "stageRecorder": stage_recorder,
  "stageRecorderPid": recorder_pid or None,
}, ensure_ascii=False))
PY
`;
}

function stageProxyRecorderPython(): string {
  return String.raw`from collections import Counter, defaultdict
import json
import subprocess
import sys
import time

namespace, job_name, proxy_selector, timeout_raw, interval_raw = sys.argv[1:6]
deadline = time.time() + int(timeout_raw) + 180
interval = max(1, int(interval_raw))
annotation_key = "unidesk.ai/stage-proxy-evidence"
stage_by_container = {
    "apk-add": "apk",
    "npm-install": "npm",
    "go-download": "go",
    "git-mirror": "git-mirror",
}
container_by_stage = [
    ("apk-add", "apk"),
    ("npm-install", "npm"),
    ("go-download", "go"),
    ("git-mirror", "git-mirror"),
]
stage_order = ["apk", "npm", "go", "git-mirror"]
previous = {}
evidence = {}
last_stage = None

def utc_now():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

def run(args, timeout=10):
    return subprocess.run(["kubectl", "-n", namespace, *args], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout)

def get_json(args, timeout=10):
    result = run(args, timeout=timeout)
    if result.returncode != 0:
        return None
    try:
        return json.loads(result.stdout or "{}")
    except Exception:
        return None

def number(value):
    try:
        return int(value)
    except Exception:
        return 0

def field(record, names, default=None):
    if not isinstance(record, dict):
        return default
    for name in names:
        value = record.get(name)
        if value not in (None, ""):
            return value
    return default

def metadata(conn):
    value = conn.get("metadata") if isinstance(conn, dict) else None
    return value if isinstance(value, dict) else {}

def client_of(conn):
    meta = metadata(conn)
    value = field(meta, ["sourceIP", "source_ip", "source", "clientIP", "client_ip", "srcIP", "src_ip"])
    if value is None:
        value = field(conn, ["sourceIP", "source_ip", "source"])
    text = str(value or "unknown")
    if text.count(":") == 1 and text.rsplit(":", 1)[1].isdigit():
        text = text.rsplit(":", 1)[0]
    return text

def destination_of(conn):
    meta = metadata(conn)
    host = field(meta, ["host", "destinationIP", "destination_ip", "destination", "dstIP", "dst_ip"])
    port = field(meta, ["destinationPort", "destination_port", "dstPort", "dst_port"])
    if host is None:
        host = field(conn, ["host", "destination"])
    if host is None:
        return "-"
    host_text = str(host)
    return f"{host_text}:{port}" if port not in (None, "") else host_text

def conn_id(conn):
    value = field(conn, ["id", "uuid", "connId", "connectionId"])
    if value not in (None, ""):
        return str(value)
    meta = metadata(conn)
    return "|".join([
        client_of(conn),
        str(field(meta, ["sourcePort", "source_port"], "")),
        destination_of(conn),
        str(field(meta, ["network", "type"], "")),
    ])

def transfer_total(conn):
    upload = number(field(conn, ["upload", "up", "uploadBytes", "upload_bytes"], 0))
    download = number(field(conn, ["download", "down", "downloadBytes", "download_bytes"], 0))
    return upload + download

def process_total(data):
    if not isinstance(data, dict):
        return 0
    upload = number(field(data, ["uploadTotal", "upload_total", "upload"], 0))
    download = number(field(data, ["downloadTotal", "download_total", "download"], 0))
    return upload + download

def current_stage_and_pod():
    job = get_json(["get", "job", job_name, "-o", "json"])
    if not isinstance(job, dict):
        return None, None, True
    status = job.get("status") or {}
    done = number(status.get("succeeded")) > 0 or number(status.get("failed")) > 0
    pods = get_json(["get", "pods", "-l", "job-name=" + job_name, "-o", "json"])
    items = pods.get("items", []) if isinstance(pods, dict) else []
    items.sort(key=lambda item: item.get("metadata", {}).get("creationTimestamp", ""))
    if not items:
        return None, None, done
    pod = items[-1]
    pod_ip = (pod.get("status") or {}).get("podIP")
    raw_statuses = (pod.get("status") or {}).get("initContainerStatuses") or []
    statuses = {item.get("name"): item for item in raw_statuses if isinstance(item, dict)}
    for container_name, stage in container_by_stage:
        item = statuses.get(container_name)
        if item is None:
            return stage, pod_ip, done
        state = item.get("state") or {}
        terminated = state.get("terminated") or {}
        if terminated and number(terminated.get("exitCode")) == 0:
            continue
        return stage, pod_ip, done
    return None, pod_ip, done

def proxy_connections():
    pods = get_json(["get", "pod", "-l", proxy_selector, "-o", "json"])
    items = pods.get("items", []) if isinstance(pods, dict) else []
    if not items:
        return None
    items.sort(key=lambda item: item.get("metadata", {}).get("creationTimestamp", ""))
    pod_name = items[-1].get("metadata", {}).get("name")
    if not pod_name:
        return None
    result = run(["exec", pod_name, "--", "sh", "-c", "if command -v wget >/dev/null 2>&1; then wget -qO- http://127.0.0.1:9090/connections; elif command -v curl >/dev/null 2>&1; then curl -fsS --max-time 3 http://127.0.0.1:9090/connections; else exit 127; fi"], timeout=8)
    if result.returncode != 0:
        return None
    try:
        parsed = json.loads(result.stdout or "{}")
    except Exception:
        return None
    connections = parsed.get("connections") if isinstance(parsed, dict) else None
    return {"data": parsed, "connections": connections if isinstance(connections, list) else []}

def ensure_stage(stage):
    if stage not in evidence:
        evidence[stage] = {
            "stage": stage,
            "firstObservedAt": utc_now(),
            "lastObservedAt": utc_now(),
            "samples": 0,
            "windowBytes": 0,
            "maxRateBps": 0,
            "proxyCumulativeBytes": 0,
            "destBytes": Counter(),
            "clientBytes": Counter(),
        }
    return evidence[stage]

def compact_evidence(state):
    stages = {}
    for stage in stage_order:
        item = evidence.get(stage)
        if not item:
            continue
        top_destinations = [{"destination": name, "bytes": count} for name, count in item["destBytes"].most_common(3)]
        top_clients = [{"client": name, "bytes": count} for name, count in item["clientBytes"].most_common(2)]
        stages[stage] = {
            "stage": stage,
            "firstObservedAt": item["firstObservedAt"],
            "lastObservedAt": item["lastObservedAt"],
            "samples": item["samples"],
            "windowBytes": item["windowBytes"],
            "maxRateBps": item["maxRateBps"],
            "proxyCumulativeBytes": item["proxyCumulativeBytes"],
            "topDestination": top_destinations[0]["destination"] if top_destinations else "-",
            "topClient": top_clients[0]["client"] if top_clients else "-",
            "topDestinations": top_destinations,
            "topClients": top_clients,
        }
    return {"version": 1, "state": state, "updatedAt": utc_now(), "stages": stages}

def patch_evidence(state):
    payload = json.dumps(compact_evidence(state), ensure_ascii=False, separators=(",", ":"), sort_keys=True)
    result = run(["annotate", "job", job_name, f"{annotation_key}={payload}", "--overwrite"], timeout=8)
    return result.returncode == 0

def sample_stage(stage, pod_ip, snapshot, elapsed):
    global previous
    current = {}
    for conn in snapshot["connections"]:
        current[conn_id(conn)] = {
            "client": client_of(conn),
            "destination": destination_of(conn),
            "total": transfer_total(conn),
        }
    total_delta = 0
    stage_item = ensure_stage(stage)
    for cid, curr in current.items():
        if pod_ip and curr["client"] != pod_ip:
            continue
        prev = previous.get(cid)
        delta = max(0, curr["total"] - (prev["total"] if prev else curr["total"]))
        if delta <= 0:
            continue
        total_delta += delta
        stage_item["destBytes"][curr["destination"]] += delta
        stage_item["clientBytes"][curr["client"]] += delta
    if total_delta > 0:
        stage_item["samples"] += 1
        stage_item["lastObservedAt"] = utc_now()
        stage_item["windowBytes"] += total_delta
        stage_item["maxRateBps"] = max(stage_item["maxRateBps"], int(total_delta / max(elapsed, interval, 1)))
        stage_item["proxyCumulativeBytes"] = process_total(snapshot["data"])
    previous = current

last_time = time.time()
last_patch = 0
while time.time() < deadline:
    stage, pod_ip, done = current_stage_and_pod()
    snapshot = proxy_connections()
    now = time.time()
    elapsed = now - last_time
    if stage != last_stage:
        previous = {}
        last_stage = stage
        last_time = now
        if done:
            break
        time.sleep(interval)
        continue
    if stage and snapshot:
        sample_stage(stage, pod_ip, snapshot, elapsed)
    last_time = now
    if evidence and now - last_patch >= max(interval, 3):
        patch_evidence("running" if not done else "completed")
        last_patch = now
    if done:
        break
    time.sleep(interval)

if evidence:
    patch_evidence("completed")
`;
}

function cleanupScript(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile): string {
  const selector = `app.kubernetes.io/name=${BENCHMARK_APP},unidesk.ai/benchmark-profile=${profile.id},unidesk.ai/runtime-node=${target.id.toLowerCase()}`;
  return `
set -eu
namespace=${shQuote(target.namespace)}
selector=${shQuote(selector)}
before="$(kubectl -n "$namespace" get jobs -l "$selector" --no-headers 2>/dev/null | wc -l | tr -d ' ')"
kubectl -n "$namespace" delete jobs -l "$selector" --ignore-not-found >/tmp/k3s-build-benchmark-cleanup.out 2>/tmp/k3s-build-benchmark-cleanup.err
after="$(kubectl -n "$namespace" get jobs -l "$selector" --no-headers 2>/dev/null | wc -l | tr -d ' ')"
deleted=$((before - after))
python3 - "$before" "$after" "$deleted" <<'PY'
import json, sys
before, after, deleted = sys.argv[1:4]
print(json.dumps({"ok": True, "before": int(before), "after": int(after), "deleted": int(deleted), "detail": "matching jobs deleted"}, ensure_ascii=False))
PY
`;
}

function statusScript(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, tailLines: number): string {
  const selector = `app.kubernetes.io/name=${BENCHMARK_APP},unidesk.ai/benchmark-profile=${profile.id},unidesk.ai/runtime-node=${target.id.toLowerCase()}`;
  return `
set -eu
python3 - ${shQuote(target.namespace)} ${shQuote(selector)} ${shQuote(String(tailLines))} <<'PY'
import json, re, subprocess, sys
namespace, selector, tail_lines_raw = sys.argv[1:4]
tail_lines = int(tail_lines_raw)

def kubectl(args):
    return subprocess.run(["kubectl", "-n", namespace, *args], text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

jobs_result = kubectl(["get", "jobs", "-l", selector, "-o", "json"])
if jobs_result.returncode != 0:
    print(json.dumps({"ok": False, "reason": "kubectl-jobs-failed", "stderr": jobs_result.stderr[-2000:]}, ensure_ascii=False))
    sys.exit(0)
jobs = json.loads(jobs_result.stdout or "{}").get("items", [])
if not jobs:
    print(json.dumps({"ok": False, "reason": "job-missing", "state": "missing"}, ensure_ascii=False))
    sys.exit(0)
jobs.sort(key=lambda item: item.get("metadata", {}).get("creationTimestamp", ""))
job = jobs[-1]
meta = job.get("metadata", {})
status = job.get("status", {})
job_name = meta.get("name") or "-"
labels = meta.get("labels", {})
annotations = meta.get("annotations", {}) or {}
pods_result = kubectl(["get", "pods", "-l", "job-name=" + job_name, "-o", "json"])
pods = json.loads(pods_result.stdout or "{}").get("items", []) if pods_result.returncode == 0 else []
pods.sort(key=lambda item: item.get("metadata", {}).get("creationTimestamp", ""))
pod_name = pods[-1].get("metadata", {}).get("name") if pods else None
waiting_reasons = []
container_names = []
stage_by_container = {"apk-add": "apk", "npm-install": "npm", "go-download": "go", "git-mirror": "git-mirror"}
container_by_stage = [("apk-add", "apk"), ("npm-install", "npm"), ("go-download", "go"), ("git-mirror", "git-mirror")]
current_stage = "-"
if pods:
    pod_status = pods[-1].get("status", {}) or {}
    status_groups = []
    status_groups.extend((pod_status.get("initContainerStatuses") or []))
    status_groups.extend((pod_status.get("containerStatuses") or []))
    init_statuses = {item.get("name"): item for item in (pod_status.get("initContainerStatuses") or []) if isinstance(item, dict)}
    for container_name, stage_name in container_by_stage:
        item = init_statuses.get(container_name)
        if item is None:
            current_stage = stage_name
            break
        state_record = item.get("state") or {}
        terminated = state_record.get("terminated") or {}
        if terminated and int(terminated.get("exitCode") or 0) == 0:
            continue
        current_stage = stage_name
        break
    for container_status in status_groups:
        container_name = container_status.get("name") or "container"
        container_names.append(container_name)
        stage_name = stage_by_container.get(container_name)
        image_name = container_status.get("image") or "-"
        state_record = container_status.get("state") or {}
        waiting = (state_record.get("waiting") or {})
        if waiting:
            reason = waiting.get("reason") or "waiting"
            message = waiting.get("message") or ""
            waiting_reasons.append((container_name + " " + image_name + " " + reason + " " + message).strip())
        terminated = (state_record.get("terminated") or {})
        if terminated and int(terminated.get("exitCode") or 0) != 0:
            reason = terminated.get("reason") or "terminated"
            message = terminated.get("message") or ""
            waiting_reasons.append(f"{container_name} {image_name} terminated exit={terminated.get('exitCode')} reason={reason} {message}".strip())

def collect_logs(lines):
    if not pod_name:
        return ""
    chunks = []
    seen = set()
    for name in container_names:
        if name in seen:
            continue
        seen.add(name)
        log_result = kubectl(["logs", pod_name, "-c", name, "--tail", str(lines)])
        text = log_result.stdout if log_result.returncode == 0 else log_result.stderr
        if text:
            chunks.append(f"[{name}]\\n{text}")
    return "\\n".join(chunks)

logs = collect_logs(tail_lines)
full_logs = collect_logs(800)
match = None
for line in reversed(full_logs.splitlines()):
    marker = None
    if "UNIDESK_K3S_BUILD_BENCHMARK_RESULT " in line:
        marker = "UNIDESK_K3S_BUILD_BENCHMARK_RESULT "
    elif "UNIDESK_K3S_REAL_DEPS_RESULT " in line:
        marker = "UNIDESK_K3S_REAL_DEPS_RESULT "
    if marker is not None:
        try:
            match = json.loads(line.split(marker, 1)[1])
        except Exception:
            match = None
        break

def parse_stage_evidence(raw):
    if not raw:
        return []
    try:
        parsed = json.loads(raw)
    except Exception:
        return []
    stages = parsed.get("stages") if isinstance(parsed, dict) else None
    if not isinstance(stages, dict):
        return []
    order = ["apk", "npm", "go", "git-mirror"]
    rows = []
    for stage in order:
        item = stages.get(stage)
        if not isinstance(item, dict):
            continue
        rows.append({
            "stage": stage,
            "topDestination": item.get("topDestination") or "-",
            "topClient": item.get("topClient") or "-",
            "windowBytes": int(item.get("windowBytes") or 0),
            "maxRateBps": int(item.get("maxRateBps") or 0),
            "proxyCumulativeBytes": int(item.get("proxyCumulativeBytes") or 0),
            "firstObservedAt": item.get("firstObservedAt") or "-",
            "lastObservedAt": item.get("lastObservedAt") or "-",
            "samples": int(item.get("samples") or 0),
        })
    return rows

conditions = status.get("conditions") or []
failed = any(item.get("type") == "Failed" and item.get("status") == "True" for item in conditions)
succeeded = status.get("succeeded", 0) > 0
active = status.get("active", 0) > 0
if succeeded:
    state = "succeeded"
elif failed:
    state = "failed"
elif active:
    state = "running"
else:
    state = "pending"
failure_family = "none" if state == "succeeded" else ("in-progress" if state in ("running", "pending") else "unknown")
tail_text = "\\n".join(waiting_reasons + [(full_logs or logs)[-4000:]])
if state == "missing":
    failure_family = "job-missing"
elif "ImagePullBackOff" in tail_text or "ErrImagePull" in tail_text:
    failure_family = "image-pull"
elif "apt-get" in tail_text and ("Failed" in tail_text or "Unable to" in tail_text):
    failure_family = "apt-download"
elif "apk-add" in tail_text and ("ERROR:" in tail_text or "temporary error" in tail_text or "Permission denied" in tail_text):
    failure_family = "apk-download"
elif "npm-install" in tail_text and ("npm ERR!" in tail_text or "EAI_AGAIN" in tail_text or "ETIMEDOUT" in tail_text):
    failure_family = "npm-download"
elif "go-download" in tail_text and ("terminated exit=" in tail_text or "i/o timeout" in tail_text or "connection refused" in tail_text or "no such host" in tail_text or "proxyconnect tcp" in tail_text or "TLS handshake timeout" in tail_text or "go-runtime-missing" in tail_text):
    failure_family = "go-download"
elif "git-mirror" in tail_text and ("terminated exit=" in tail_text or "fatal:" in tail_text or "Could not resolve host" in tail_text or "Failed to connect" in tail_text or "connection refused" in tail_text or "proxyconnect tcp" in tail_text or "TLS handshake timeout" in tail_text or "git-runtime-missing" in tail_text):
    failure_family = "git-mirror"
elif "curl:" in tail_text or "urllib.error.HTTPError" in tail_text or "urllib.error.URLError" in tail_text:
    failure_family = "dependency-download"
elif "payload-too-small" in tail_text:
    failure_family = "payload-too-small"
elif failed:
    failure_family = "failed"
payload = {
    "ok": state == "succeeded",
    "state": state,
    "jobName": job_name,
    "runId": labels.get("unidesk.ai/run-id") or "-",
    "profile": labels.get("unidesk.ai/benchmark-profile") or "-",
    "currentStage": current_stage,
    "startedAt": status.get("startTime") or (match or {}).get("startedAt"),
    "completedAt": status.get("completionTime") or (match or {}).get("completedAt"),
    "result": match,
    "stageEvidence": parse_stage_evidence(annotations.get(${shQuote(STAGE_PROXY_EVIDENCE_ANNOTATION)})),
    "failureFamily": failure_family,
    "logTail": "\\n".join(waiting_reasons + [logs])[-4000:],
}
print(json.dumps(payload, ensure_ascii=False))
PY
`;
}

function sampleTraffic(target: Sub2ApiTargetConfig, sampleSeconds: number, timeoutSeconds: number): TrafficSummary {
  const spec = trafficSpec(target);
  const result = runTrans(spec.route, egressProxyTrafficScript(spec, sampleSeconds), timeoutSeconds);
  const parsed = parseJson(result.stdout);
  const data = typeof parsed === "object" && parsed !== null ? parsed as Record<string, unknown> : {};
  const totals = record(data.totals);
  const clients = arrayRecords(data.clients);
  const topClient = clients[0] ?? {};
  const destinations = arrayRecords(topClient.topDestinations);
  return {
    ok: result.exitCode === 0 && data.ok !== false,
    reason: text(data.reason, result.exitCode === 0 ? "ok" : "traffic-failed"),
    windowBytes: number(totals.clientWindowTotalBytes),
    rateBps: number(totals.clientTotalBps),
    processTotalBytes: number(totals.processTotalBytes),
    topClient: text(topClient.client),
    topDestination: text(destinations[0]?.destination),
  };
}

function trafficSpec(target: Sub2ApiTargetConfig): EgressProxyTrafficSpec {
  const proxy = target.egressProxy;
  if (proxy === null) throw new Error(`target ${target.id} has no egressProxy`);
  return {
    scope: "platform-infra",
    targetId: target.id,
    route: target.route,
    namespace: target.namespace,
    deploymentName: proxy.deploymentName,
    serviceName: proxy.serviceName,
    port: proxy.listenPort,
    sourceType: proxy.sourceType,
    sourceRef: proxy.sourceRef,
    sourceConfigRef: proxy.sourceConfigRef,
    sourceFingerprint: proxy.sourceFingerprint,
  };
}

function stageEvidenceList(value: unknown): StageProxyEvidence[] {
  return arrayRecords(value).map((item) => ({
    stage: text(item.stage),
    topDestination: text(item.topDestination),
    topClient: text(item.topClient),
    windowBytes: number(item.windowBytes),
    maxRateBps: number(item.maxRateBps),
    proxyCumulativeBytes: number(item.proxyCumulativeBytes),
    firstObservedAt: text(item.firstObservedAt),
    lastObservedAt: text(item.lastObservedAt),
    samples: number(item.samples),
  })).filter((item) => item.stage !== "-");
}

function stageEvidenceSummary(items: readonly StageProxyEvidence[]): string {
  if (items.length === 0) return "-";
  const byStage = new Map(items.map((item) => [item.stage, item]));
  const visible = REAL_DEPS_STAGE_ORDER
    .map((stage) => byStage.get(stage))
    .filter((item): item is StageProxyEvidence => item !== undefined);
  const git = byStage.get("git-mirror");
  const suffix = git === undefined ? "" : ` git=${compactDestination(git.topDestination)} ${bytes(git.windowBytes)}`;
  return `${visible.length}/${REAL_DEPS_STAGE_ORDER.length}${suffix}`;
}

function stageEvidenceSections(statuses: readonly TargetStatus[]): string[] {
  const rows = statuses.flatMap((status) => status.stageEvidence.map((item) => [
    status.targetId,
    item.stage,
    item.topDestination,
    item.topClient,
    bytes(item.windowBytes),
    rate(item.maxRateBps),
    bytes(item.proxyCumulativeBytes),
    String(item.samples),
    item.lastObservedAt,
  ]));
  if (rows.length === 0) return [];
  return [
    "",
    "STAGE_PROXY_EVIDENCE",
    ...table(["TARGET", "STAGE", "TOP_DEST", "TOP_CLIENT", "WINDOW", "MAX_RATE", "PROXY_CUM", "SAMPLES", "LAST_OBSERVED"], rows),
  ];
}

function compactDestination(value: string): string {
  if (value === "-" || value.length <= 28) return value;
  const withoutPort = value.replace(/:443$/u, "");
  if (withoutPort.length <= 28) return withoutPort;
  return `${withoutPort.slice(0, 25)}...`;
}

function normalizeStatus(plan: TargetPlan, parsed: unknown, result: CommandResult): TargetStatus {
  if (typeof parsed !== "object" || parsed === null) {
    const state = result.exitCode === 0 ? "transport-unparseable" : "transport-failed";
    return {
      ok: false,
      targetId: plan.targetId,
      profile: plan.profile?.id ?? "-",
      state,
      jobName: "-",
      runId: "-",
      startedAt: "-",
      completedAt: "-",
      durationSeconds: null,
      outputMiB: null,
      downloadMiB: null,
      payloadMiB: plan.profile?.payloadMiB ?? null,
      apkMiB: null,
      npmMiB: null,
      goMiB: null,
      gitMirrorMiB: null,
      realDepsMiB: null,
      currentStage: "-",
      stageEvidence: [],
      failureFamily: result.timedOut ? "transport-timeout" : state,
      logTail: (result.stderr || result.stdout).slice(-4000),
    };
  }
  const data = typeof parsed === "object" && parsed !== null ? parsed as Record<string, unknown> : {};
  const jobResult = record(data.result);
  const state = text(data.state, result.exitCode === 0 ? "unknown" : "failed");
  const commandOk = result.exitCode === 0 && state !== "failed" && state !== "missing" && text(data.reason, "") !== "kubectl-jobs-failed";
  const status: TargetStatus = {
    ok: commandOk,
    targetId: plan.targetId,
    profile: text(data.profile, plan.profile?.id ?? "-"),
    state,
    jobName: text(data.jobName),
    runId: text(data.runId),
    startedAt: text(data.startedAt),
    completedAt: text(data.completedAt),
    durationSeconds: nullableNumber(jobResult.durationSeconds),
    outputMiB: nullableNumber(jobResult.outputMiB),
    downloadMiB: nullableNumber(jobResult.downloadMiB),
    payloadMiB: nullableNumber(jobResult.payloadMiB),
    apkMiB: nullableNumber(jobResult.apkMiB),
    npmMiB: nullableNumber(jobResult.npmMiB),
    goMiB: nullableNumber(jobResult.goMiB),
    gitMirrorMiB: nullableNumber(jobResult.gitMirrorMiB),
    realDepsMiB: nullableNumber(jobResult.realDepsMiB),
    currentStage: text(data.currentStage),
    stageEvidence: stageEvidenceList(data.stageEvidence),
    failureFamily: text(data.failureFamily, data.ok === true ? "none" : state === "running" || state === "pending" ? "in-progress" : text(data.reason, "unknown")),
    logTail: text(data.logTail, result.stderr.slice(-2000)),
  };
  return status;
}

function blockedStatus(plan: TargetPlan, profile: string): TargetStatus {
  return {
    ok: false,
    targetId: plan.targetId,
    profile,
    state: "blocked",
    jobName: "-",
    runId: "-",
    startedAt: "-",
    completedAt: "-",
    durationSeconds: null,
    outputMiB: null,
    downloadMiB: null,
    payloadMiB: plan.profile?.payloadMiB ?? null,
    apkMiB: null,
    npmMiB: null,
    goMiB: null,
    gitMirrorMiB: null,
    realDepsMiB: null,
    currentStage: "-",
    stageEvidence: [],
    failureFamily: plan.blocker ?? "blocked",
    logTail: plan.detail ?? "",
  };
}

function benchmarkJobName(target: Sub2ApiTargetConfig, profile: K3sBuildBenchmarkProfile, runId: string): string {
  const base = `k3s-build-${target.id.toLowerCase()}-${profile.id}-${runId.replace(/^k3sbuild-/u, "")}`.toLowerCase().replace(/[^a-z0-9-]/gu, "-");
  return base.slice(0, 63).replace(/-+$/u, "");
}

function readK3sBuildBenchmarkConfig(): K3sBuildBenchmarkConfig {
  const raw = asRecord(Bun.YAML.parse(readFileSync(rootPath(BENCHMARK_CONFIG_PATH), "utf8")) as unknown, BENCHMARK_CONFIG_PATH);
  const version = integerField(raw, "version", BENCHMARK_CONFIG_PATH);
  const kind = stringField(raw, "kind", BENCHMARK_CONFIG_PATH);
  if (kind !== "platform-infra-egress-proxy-benchmarks") throw new Error(`${BENCHMARK_CONFIG_PATH}.kind must be platform-infra-egress-proxy-benchmarks`);
  const metadataRaw = asRecord(raw.metadata, "metadata");
  const profilesRaw = asRecord(raw.profiles, "profiles");
  const profiles = Object.fromEntries(Object.entries(profilesRaw).map(([id, value]) => [id, profileSpec(id, asRecord(value, `profiles.${id}`))]));
  return {
    version,
    kind,
    metadata: {
      owner: stringField(metadataRaw, "owner", "metadata"),
      relatedIssues: integerArrayField(metadataRaw, "relatedIssues", "metadata"),
    },
    profiles,
  };
}

function profileSpec(id: string, raw: Record<string, unknown>): K3sBuildBenchmarkProfile {
  const workload = stringField(raw, "workload", `profiles.${id}`);
  if (workload !== "k3s-build" && workload !== "k3s-real-deps") throw new Error(`profiles.${id}.workload must be k3s-build or k3s-real-deps`);
  const imagePullPolicy = imagePullPolicyField(raw, "imagePullPolicy", `profiles.${id}`);
  const noMirror = asRecord(raw.noMirror, `profiles.${id}.noMirror`);
  const registryMirror = stringField(noMirror, "registryMirror", `profiles.${id}.noMirror`);
  if (registryMirror !== "forbidden") throw new Error(`profiles.${id}.noMirror.registryMirror must be forbidden`);
  const dependencyDownload = asRecord(raw.dependencyDownload, `profiles.${id}.dependencyDownload`);
  return {
    id,
    enabled: booleanField(raw, "enabled", `profiles.${id}`),
    workload,
    description: stringField(raw, "description", `profiles.${id}`),
    image: stringField(raw, "image", `profiles.${id}`),
    imagePullPolicy,
    targetOverrides: targetOverridesField(raw, `profiles.${id}`),
    payloadMiB: integerField(raw, "payloadMiB", `profiles.${id}`),
    timeoutSeconds: integerField(raw, "timeoutSeconds", `profiles.${id}`),
    ttlSecondsAfterFinished: integerField(raw, "ttlSecondsAfterFinished", `profiles.${id}`),
    noMirror: {
      apt: stringField(noMirror, "apt", `profiles.${id}.noMirror`),
      npmRegistry: stringField(noMirror, "npmRegistry", `profiles.${id}.noMirror`),
      pipIndexUrl: stringField(noMirror, "pipIndexUrl", `profiles.${id}.noMirror`),
      registryMirror,
    },
    aptPackages: stringArrayField(raw, "aptPackages", `profiles.${id}`),
    dependencyDownload: {
      enabled: booleanField(dependencyDownload, "enabled", `profiles.${id}.dependencyDownload`),
      url: stringField(dependencyDownload, "url", `profiles.${id}.dependencyDownload`),
      chunks: integerField(dependencyDownload, "chunks", `profiles.${id}.dependencyDownload`),
      expectedMiB: integerField(dependencyDownload, "expectedMiB", `profiles.${id}.dependencyDownload`),
    },
    realDeps: raw.realDeps === undefined ? undefined : realDepsSpec(asRecord(raw.realDeps, `profiles.${id}.realDeps`), `profiles.${id}.realDeps`),
  };
}

function realDepsSpec(raw: Record<string, unknown>, path: string): K3sRealDepsSpec {
  const apk = asRecord(raw.apk, `${path}.apk`);
  const npm = asRecord(raw.npm, `${path}.npm`);
  const go = asRecord(raw.go, `${path}.go`);
  const gitMirror = asRecord(raw.gitMirror, `${path}.gitMirror`);
  return {
    minProxyMiB: integerField(raw, "minProxyMiB", path),
    imagePullPolicy: imagePullPolicyField(raw, "imagePullPolicy", path),
    apk: {
      image: stringField(apk, "image", `${path}.apk`),
      packages: stringArrayField(apk, "packages", `${path}.apk`),
      expectedMiB: integerField(apk, "expectedMiB", `${path}.apk`),
    },
    npm: {
      image: stringField(npm, "image", `${path}.npm`),
      registry: stringField(npm, "registry", `${path}.npm`),
      packages: stringRecordField(npm, "packages", `${path}.npm`),
      expectedMiB: integerField(npm, "expectedMiB", `${path}.npm`),
    },
    go: {
      image: stringField(go, "image", `${path}.go`),
      goProxy: stringField(go, "goProxy", `${path}.go`),
      modules: stringArrayField(go, "modules", `${path}.go`),
      expectedMiB: integerField(go, "expectedMiB", `${path}.go`),
    },
    gitMirror: {
      image: stringField(gitMirror, "image", `${path}.gitMirror`),
      remote: stringField(gitMirror, "remote", `${path}.gitMirror`),
      expectedMiB: integerField(gitMirror, "expectedMiB", `${path}.gitMirror`),
    },
  };
}

function targetOverridesField(raw: Record<string, unknown>, path: string): Record<string, K3sBuildBenchmarkTargetOverride> {
  if (raw.targetOverrides === undefined) return {};
  const overrides = asRecord(raw.targetOverrides, `${path}.targetOverrides`);
  return Object.fromEntries(Object.entries(overrides).map(([targetId, value]) => {
    const recordValue = asRecord(value, `${path}.targetOverrides.${targetId}`);
    const override: K3sBuildBenchmarkTargetOverride = {};
    if (recordValue.image !== undefined) override.image = stringField(recordValue, "image", `${path}.targetOverrides.${targetId}`);
    if (recordValue.imagePullPolicy !== undefined) {
      override.imagePullPolicy = imagePullPolicyField(recordValue, "imagePullPolicy", `${path}.targetOverrides.${targetId}`);
    }
    return [targetId, override];
  }));
}

function runTrans(route: string, script: string, timeoutSeconds: number): CommandResult {
  return runCommand(["/root/.local/bin/trans", route, "sh", "--", script], rootPath(), { timeoutMs: timeoutSeconds * 1000 });
}

function shQuote(value: string): string {
  return `'${value.replace(/'/gu, `'\\''`)}'`;
}

function option(args: string[], name: string): string | null {
  const index = args.indexOf(name);
  if (index === -1) return null;
  const value = args[index + 1];
  if (value === undefined || value.startsWith("--")) throw new Error(`${name} requires a value`);
  return value;
}

function positiveIntOption(args: string[], name: string, defaultValue: number, maxValue: number): number {
  const raw = option(args, name);
  if (raw === null) return defaultValue;
  const value = Number.parseInt(raw, 10);
  if (!Number.isInteger(value) || value < 0 || value > maxValue) throw new Error(`${name} must be an integer from 0 to ${maxValue}`);
  return value;
}

function parseJson(textValue: string): unknown {
  const trimmed = textValue.trim();
  if (trimmed.length === 0) return null;
  try { return JSON.parse(trimmed); } catch {
    const start = trimmed.indexOf("{");
    const end = trimmed.lastIndexOf("}");
    if (start >= 0 && end > start) {
      try { return JSON.parse(trimmed.slice(start, end + 1)); } catch {}
    }
  }
  return null;
}

function asRecord(value: unknown, path: string): Record<string, unknown> {
  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${path} must be an object`);
  return value as Record<string, unknown>;
}

function stringField(obj: Record<string, unknown>, key: string, path: string): string {
  const value = obj[key];
  if (typeof value !== "string" || value.length === 0) throw new Error(`${path}.${key} must be a non-empty string`);
  return value;
}

function booleanField(obj: Record<string, unknown>, key: string, path: string): boolean {
  const value = obj[key];
  if (typeof value !== "boolean") throw new Error(`${path}.${key} must be a boolean`);
  return value;
}

function integerField(obj: Record<string, unknown>, key: string, path: string): number {
  const value = obj[key];
  if (typeof value !== "number" || !Number.isInteger(value) || value < 0) throw new Error(`${path}.${key} must be a non-negative integer`);
  return value;
}

function integerArrayField(obj: Record<string, unknown>, key: string, path: string): number[] {
  const value = obj[key];
  if (!Array.isArray(value) || value.some((item) => typeof item !== "number" || !Number.isInteger(item))) throw new Error(`${path}.${key} must be an integer array`);
  return [...value] as number[];
}

function stringArrayField(obj: Record<string, unknown>, key: string, path: string): string[] {
  const value = obj[key];
  if (!Array.isArray(value) || value.some((item) => typeof item !== "string" || item.length === 0)) throw new Error(`${path}.${key} must be a non-empty string array`);
  return [...value] as string[];
}

function stringRecordField(obj: Record<string, unknown>, key: string, path: string): Record<string, string> {
  const value = obj[key];
  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${path}.${key} must be an object`);
  const entries = Object.entries(value as Record<string, unknown>);
  if (entries.some(([name, item]) => name.length === 0 || typeof item !== "string" || item.length === 0)) throw new Error(`${path}.${key} must contain string values`);
  return Object.fromEntries(entries) as Record<string, string>;
}

function imagePullPolicyField(obj: Record<string, unknown>, key: string, path: string): ImagePullPolicy {
  const value = stringField(obj, key, path);
  if (value !== "Always" && value !== "IfNotPresent" && value !== "Never") throw new Error(`${path}.${key} must be Always, IfNotPresent, or Never`);
  return value;
}

function table(headers: string[], rows: string[][]): string[] {
  const widths = headers.map((header, index) => Math.max(header.length, ...rows.map((row) => row[index]?.length ?? 0)));
  const render = (row: string[]) => row.map((cell, index) => cell.padEnd(widths[index] ?? cell.length)).join("  ").trimEnd();
  return [render(headers), ...rows.map(render)];
}

function record(value: unknown): Record<string, unknown> {
  return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
}

function arrayRecords(value: unknown): Array<Record<string, unknown>> {
  return Array.isArray(value) ? value.map(record) : [];
}

function text(value: unknown, fallback = "-"): string {
  if (value === undefined || value === null || value === "") return fallback;
  return String(value);
}

function number(value: unknown): number {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : 0;
}

function nullableNumber(value: unknown): number | null {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : null;
}

function bytes(value: unknown): string {
  const parsed = number(value);
  if (parsed <= 0) return "0 B";
  const units = ["B", "KiB", "MiB", "GiB", "TiB"];
  let scaled = parsed;
  let unit = 0;
  while (scaled >= 1024 && unit < units.length - 1) {
    scaled /= 1024;
    unit += 1;
  }
  return `${scaled >= 10 || unit === 0 ? scaled.toFixed(0) : scaled.toFixed(1)} ${units[unit]}`;
}

function rate(value: unknown): string {
  return `${bytes(value)}/s`;
}