pikasTech-unidesk/scripts/src/code-queue-execution-plane.ts

import { runCommand, type CommandResult } from "./command";
import { repoRoot } from "./config";
import { classifyD601K3sTarget, d601NativeKubeconfig, d601RequiredNodeName, type D601K3sGuardClassification } from "./d601-k3s-guard";

type PlaneStatus = "ready" | "blocked" | "degraded";
type SignalSeverity = "blocker" | "warning";

interface ExecutionPlaneOptions {
  namespace: string;
  kubeconfig: string;
  worktreePath: string;
  full: boolean;
  raw: boolean;
  timeoutMs: number;
}

interface ProbeResult {
  ok: boolean;
  command: string[];
  exitCode: number | null;
  stdout: string;
  stderr: string;
  timedOut: boolean;
}

interface DeploymentObservation {
  name: string;
  namespace: string;
  observed: boolean;
  expectedRole: "scheduler" | "read" | "write";
  expectedSchedulerEnabled: string;
  labels: {
    app: string | null;
    component: string | null;
    deploymentMode: string | null;
    instanceId: string | null;
  };
  annotations: {
    deployRef: string | null;
    deployCommit: string | null;
    deployRequestedCommit: string | null;
    imageSource: string | null;
  };
  replicas: {
    desired: number | null;
    ready: number;
    available: number;
    updated: number;
  };
  nodeSelector: string | null;
  image: string | null;
  env: {
    serviceRole: string | null;
    schedulerEnabled: string | null;
    unideskDeployCommit: string | null;
    unideskDeployRequestedCommit: string | null;
    codeQueueDeployCommit: string | null;
    codeQueueDeployRequestedCommit: string | null;
  };
  repoHostPath: string | null;
  error: string | null;
}

interface PodObservation {
  name: string;
  instanceId: string | null;
  component: string | null;
  nodeName: string | null;
  phase: string | null;
  ready: boolean;
  imageID: string | null;
  digest: string | null;
}

interface ServiceObservation {
  name: string;
  observed: boolean;
  type: string | null;
  clusterIP: string | null;
  ports: string[];
  selector: {
    app: string | null;
    component: string | null;
    instanceId: string | null;
  };
  error: string | null;
}

interface WorktreeObservation {
  path: string;
  ok: boolean;
  head: string | null;
  error: string | null;
}

interface ResidualObservation {
  composeBackend: {
    ok: boolean;
    present: boolean;
    containers: Array<{ name: string; status: string | null; image: string | null }>;
    error: string | null;
  };
  loopbackPort4222: {
    ok: boolean;
    present: boolean;
    listeners: Array<{ localAddress: string; process: string | null; line: string }>;
    error: string | null;
  };
}

export interface CodeQueueExecutionPlaneObservation {
  checkedAt: string;
  namespace: string;
  kubeconfig: string;
  worktreePath: string;
  guard: D601K3sGuardClassification;
  deployments: DeploymentObservation[];
  pods: PodObservation[];
  services: ServiceObservation[];
  worktree: WorktreeObservation;
  residual: ResidualObservation;
  commandDiagnostics: Record<string, unknown>;
}

interface ExecutionPlaneCollector {
  collect(options: ExecutionPlaneOptions): Promise<CodeQueueExecutionPlaneObservation> | CodeQueueExecutionPlaneObservation;
}

interface DriftSignal {
  code: string;
  severity: SignalSeverity;
  message: string;
  expected?: unknown;
  observed?: unknown;
  field?: string;
}

const expectedNamespace = "unidesk";
const expectedWorktreePath = "/home/ubuntu/cq-deploy";

const expectedDeployments = [
  { name: "code-queue", role: "scheduler" as const, schedulerEnabled: "true" },
  { name: "code-queue-read", role: "read" as const, schedulerEnabled: "false" },
  { name: "code-queue-write", role: "write" as const, schedulerEnabled: "false" },
];

function nowIso(): string {
  return new Date().toISOString();
}

function hasFlag(args: string[], name: string): boolean {
  return args.includes(name);
}

function optionValue(args: string[], names: string[]): string | undefined {
  for (let index = 0; index < args.length; index += 1) {
    if (!names.includes(args[index] ?? "")) continue;
    const value = args[index + 1];
    if (value === undefined || value.startsWith("--")) throw new Error(`${args[index]} requires a value`);
    return value;
  }
  return undefined;
}

function parsePositiveInteger(value: string | undefined, fallback: number, max: number): number {
  if (value === undefined) return fallback;
  const parsed = Number(value);
  if (!Number.isInteger(parsed) || parsed <= 0) throw new Error("--timeout-ms must be a positive integer");
  return Math.min(parsed, max);
}

function parseExecutionPlaneOptions(args: string[]): ExecutionPlaneOptions {
  const knownFlags = new Set(["--full", "--raw"]);
  const knownValues = new Set(["--namespace", "--kubeconfig", "--worktree", "--timeout-ms"]);
  for (let index = 0; index < args.length; index += 1) {
    const arg = args[index] ?? "";
    if (!arg.startsWith("--")) throw new Error(`codex execution-plane does not accept positional argument: ${arg}`);
    if (knownFlags.has(arg)) continue;
    if (knownValues.has(arg)) {
      index += 1;
      continue;
    }
    throw new Error(`unknown codex execution-plane option: ${arg}`);
  }
  const raw = hasFlag(args, "--raw");
  return {
    namespace: optionValue(args, ["--namespace"]) ?? expectedNamespace,
    kubeconfig: optionValue(args, ["--kubeconfig"]) ?? d601NativeKubeconfig,
    worktreePath: optionValue(args, ["--worktree"]) ?? expectedWorktreePath,
    full: raw || hasFlag(args, "--full"),
    raw,
    timeoutMs: parsePositiveInteger(optionValue(args, ["--timeout-ms"]), 15_000, 60_000),
  };
}

function firstLine(text: string): string | null {
  const line = text.split(/\r?\n/u).map((item) => item.trim()).find((item) => item.length > 0);
  return line ?? null;
}

function lines(text: string): string[] {
  return text.split(/\r?\n/u).map((item) => item.trim()).filter(Boolean);
}

function nullable(value: string | undefined): string | null {
  const trimmed = (value ?? "").trim();
  return trimmed.length === 0 ? null : trimmed;
}

function numberOrNull(value: string | undefined): number | null {
  const trimmed = (value ?? "").trim();
  if (trimmed.length === 0) return null;
  const parsed = Number(trimmed);
  return Number.isFinite(parsed) ? parsed : null;
}

function boolFromString(value: string | null): boolean {
  return value === "true";
}

function commandProbe(result: CommandResult): ProbeResult {
  return {
    ok: result.exitCode === 0 && !result.timedOut,
    command: result.command,
    exitCode: result.exitCode,
    stdout: result.stdout,
    stderr: result.stderr,
    timedOut: result.timedOut,
  };
}

function runTran(args: string[], options: ExecutionPlaneOptions, timeoutMs = options.timeoutMs): ProbeResult {
  return commandProbe(runCommand(["./scripts/tran", ...args], repoRoot, {
    timeoutMs,
    env: process.env,
  }));
}

function safeError(probe: ProbeResult): string | null {
  if (probe.ok) return null;
  const text = firstLine(probe.stderr) ?? firstLine(probe.stdout);
  return text ?? `exitCode=${probe.exitCode ?? "null"}`;
}

function runKubectl(args: string[], options: ExecutionPlaneOptions): ProbeResult {
  return runTran(["D601:k3s", "kubectl", ...args], options);
}

function collectGuard(options: ExecutionPlaneOptions): { guard: D601K3sGuardClassification; diagnostics: Record<string, unknown> } {
  const context = runKubectl(["config", "current-context"], options);
  const server = runKubectl(["config", "view", "--minify", "-o", "jsonpath={.clusters[0].cluster.server}"], options);
  const nodes = runKubectl(["get", "nodes", "-o", "jsonpath={range .items[*]}{.metadata.name}{\"\\n\"}{end}"], options);
  const combinedText = [context.stdout, context.stderr, server.stdout, server.stderr, nodes.stdout, nodes.stderr].join("\n");
  const guard = classifyD601K3sTarget({
    kubeconfig: options.kubeconfig,
    expectedKubeconfig: d601NativeKubeconfig,
    currentContext: firstLine(context.stdout),
    apiServer: firstLine(server.stdout),
    nodeNames: lines(nodes.stdout),
    commandsOk: context.ok && server.ok && nodes.ok,
    combinedText,
  });
  return {
    guard,
    diagnostics: {
      context: compactProbe(context),
      server: compactProbe(server),
      nodes: compactProbe(nodes),
    },
  };
}

function compactProbe(probe: ProbeResult): Record<string, unknown> {
  return {
    ok: probe.ok,
    exitCode: probe.exitCode,
    timedOut: probe.timedOut,
    stdoutPreview: firstLine(probe.stdout),
    stderrPreview: firstLine(probe.stderr),
  };
}

function deploymentJsonPath(): string {
  return [
    "{.metadata.name}",
    "{\"\\t\"}{.metadata.namespace}",
    "{\"\\t\"}{.metadata.labels.app\\.kubernetes\\.io/name}",
    "{\"\\t\"}{.metadata.labels.app\\.kubernetes\\.io/component}",
    "{\"\\t\"}{.metadata.labels.unidesk\\.ai/deployment-mode}",
    "{\"\\t\"}{.metadata.labels.unidesk\\.ai/instance-id}",
    "{\"\\t\"}{.metadata.annotations.unidesk\\.ai/deploy-ref}",
    "{\"\\t\"}{.metadata.annotations.unidesk\\.ai/deploy-commit}",
    "{\"\\t\"}{.metadata.annotations.unidesk\\.ai/deploy-requested-commit}",
    "{\"\\t\"}{.metadata.annotations.unidesk\\.ai/image-source}",
    "{\"\\t\"}{.status.readyReplicas}",
    "{\"\\t\"}{.status.availableReplicas}",
    "{\"\\t\"}{.status.updatedReplicas}",
    "{\"\\t\"}{.spec.replicas}",
    "{\"\\t\"}{.spec.template.spec.nodeSelector.unidesk\\.ai/node-id}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].image}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"CODE_QUEUE_SERVICE_ROLE\")].value}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"CODE_QUEUE_SCHEDULER_ENABLED\")].value}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"UNIDESK_DEPLOY_COMMIT\")].value}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"UNIDESK_DEPLOY_REQUESTED_COMMIT\")].value}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"CODE_QUEUE_DEPLOY_COMMIT\")].value}",
    "{\"\\t\"}{.spec.template.spec.containers[?(@.name==\"code-queue\")].env[?(@.name==\"CODE_QUEUE_DEPLOY_REQUESTED_COMMIT\")].value}",
    "{\"\\t\"}{.spec.template.spec.volumes[?(@.name==\"repo\")].hostPath.path}",
  ].join("");
}

function parseDeployment(fields: string[], expected: typeof expectedDeployments[number], namespace: string, error: string | null): DeploymentObservation {
  return {
    name: nullable(fields[0]) ?? expected.name,
    namespace: nullable(fields[1]) ?? namespace,
    observed: error === null && nullable(fields[0]) !== null,
    expectedRole: expected.role,
    expectedSchedulerEnabled: expected.schedulerEnabled,
    labels: {
      app: nullable(fields[2]),
      component: nullable(fields[3]),
      deploymentMode: nullable(fields[4]),
      instanceId: nullable(fields[5]),
    },
    annotations: {
      deployRef: nullable(fields[6]),
      deployCommit: nullable(fields[7]),
      deployRequestedCommit: nullable(fields[8]),
      imageSource: nullable(fields[9]),
    },
    replicas: {
      ready: numberOrNull(fields[10]) ?? 0,
      available: numberOrNull(fields[11]) ?? 0,
      updated: numberOrNull(fields[12]) ?? 0,
      desired: numberOrNull(fields[13]),
    },
    nodeSelector: nullable(fields[14]),
    image: nullable(fields[15]),
    env: {
      serviceRole: nullable(fields[16]),
      schedulerEnabled: nullable(fields[17]),
      unideskDeployCommit: nullable(fields[18]),
      unideskDeployRequestedCommit: nullable(fields[19]),
      codeQueueDeployCommit: nullable(fields[20]),
      codeQueueDeployRequestedCommit: nullable(fields[21]),
    },
    repoHostPath: nullable(fields[22]),
    error,
  };
}

function collectDeployments(options: ExecutionPlaneOptions): DeploymentObservation[] {
  return expectedDeployments.map((expected) => {
    const probe = runKubectl(["-n", options.namespace, "get", "deployment", expected.name, "-o", `jsonpath=${deploymentJsonPath()}`], options);
    const error = safeError(probe);
    const fields = probe.ok ? probe.stdout.trimEnd().split("\t") : [];
    return parseDeployment(fields, expected, options.namespace, error);
  });
}

function collectPods(options: ExecutionPlaneOptions): PodObservation[] {
  const path = [
    "{range .items[*]}",
    "{.metadata.name}{\"\\t\"}",
    "{.metadata.labels.unidesk\\.ai/instance-id}{\"\\t\"}",
    "{.metadata.labels.app\\.kubernetes\\.io/component}{\"\\t\"}",
    "{.spec.nodeName}{\"\\t\"}",
    "{.status.phase}{\"\\t\"}",
    "{.status.containerStatuses[?(@.name==\"code-queue\")].ready}{\"\\t\"}",
    "{.status.containerStatuses[?(@.name==\"code-queue\")].imageID}{\"\\n\"}",
    "{end}",
  ].join("");
  const probe = runKubectl(["-n", options.namespace, "get", "pods", "-l", "app.kubernetes.io/name=code-queue", "-o", `jsonpath=${path}`], options);
  if (!probe.ok) return [];
  return lines(probe.stdout).map((line) => {
    const fields = line.split("\t");
    const imageID = nullable(fields[6]);
    return {
      name: nullable(fields[0]) ?? "unknown",
      instanceId: nullable(fields[1]),
      component: nullable(fields[2]),
      nodeName: nullable(fields[3]),
      phase: nullable(fields[4]),
      ready: boolFromString(nullable(fields[5])),
      imageID,
      digest: digestFromText(imageID),
    };
  });
}

function serviceJsonPath(): string {
  return [
    "{.metadata.name}",
    "{\"\\t\"}{.spec.type}",
    "{\"\\t\"}{.spec.clusterIP}",
    "{\"\\t\"}{range .spec.ports[*]}{.name}:{.port}->{.targetPort}{\",\"}{end}",
    "{\"\\t\"}{.spec.selector.app\\.kubernetes\\.io/name}",
    "{\"\\t\"}{.spec.selector.app\\.kubernetes\\.io/component}",
    "{\"\\t\"}{.spec.selector.unidesk\\.ai/instance-id}",
  ].join("");
}

function collectServices(options: ExecutionPlaneOptions): ServiceObservation[] {
  return expectedDeployments.map((expected) => {
    const probe = runKubectl(["-n", options.namespace, "get", "service", expected.name, "-o", `jsonpath=${serviceJsonPath()}`], options);
    const error = safeError(probe);
    const fields = probe.ok ? probe.stdout.trimEnd().split("\t") : [];
    return {
      name: nullable(fields[0]) ?? expected.name,
      observed: error === null && nullable(fields[0]) !== null,
      type: nullable(fields[1]),
      clusterIP: nullable(fields[2]),
      ports: (nullable(fields[3]) ?? "").split(",").map((item) => item.trim()).filter(Boolean),
      selector: {
        app: nullable(fields[4]),
        component: nullable(fields[5]),
        instanceId: nullable(fields[6]),
      },
      error,
    };
  });
}

function collectWorktree(options: ExecutionPlaneOptions): WorktreeObservation {
  const probe = runTran([`D601:${options.worktreePath}`, "argv", "git", "rev-parse", "HEAD"], options, 10_000);
  return {
    path: options.worktreePath,
    ok: probe.ok,
    head: probe.ok ? firstLine(probe.stdout) : null,
    error: safeError(probe),
  };
}

function collectResidual(): ResidualObservation {
  const remoteOptions: ExecutionPlaneOptions = {
    namespace: expectedNamespace,
    kubeconfig: d601NativeKubeconfig,
    worktreePath: expectedWorktreePath,
    full: false,
    raw: false,
    timeoutMs: 15_000,
  };
  const docker = runTran(["D601", "argv", "docker", "ps", "-a", "--filter", "name=code-queue-backend", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"], remoteOptions, 15_000);
  const containers = docker.ok
    ? lines(docker.stdout).map((line) => {
        const fields = line.split("\t");
        return { name: fields[0] ?? "unknown", status: nullable(fields[1]), image: nullable(fields[2]) };
      }).filter((item) => item.name === "code-queue-backend" || item.name.includes("code-queue-backend"))
    : [];

  const ss = runTran(["D601", "argv", "ss", "-H", "-ltnp"], remoteOptions, 15_000);
  const listeners = ss.ok
    ? (lines(ss.stdout)
        .filter((line) => line.includes(":4222"))
        .map((line) => {
          const fields = line.split(/\s+/u);
          const localAddress = fields[3] ?? fields[2] ?? "";
          const processText = fields.find((field) => field.includes("users:")) ?? null;
          return { localAddress, process: processText, line };
        })
        .filter((item) => item.localAddress.startsWith("127.0.0.1:4222") || item.localAddress.startsWith("[::1]:4222")))
    : [];

  return {
    composeBackend: {
      ok: docker.ok,
      present: containers.length > 0,
      containers,
      error: safeError(docker),
    },
    loopbackPort4222: {
      ok: ss.ok,
      present: listeners.length > 0,
      listeners,
      error: safeError(ss),
    },
  };
}

function digestFromText(value: string | null): string | null {
  if (value === null) return null;
  const match = value.match(/sha256:[0-9a-f]{64}/iu);
  return match?.[0].toLowerCase() ?? null;
}

class LiveExecutionPlaneCollector implements ExecutionPlaneCollector {
  collect(options: ExecutionPlaneOptions): CodeQueueExecutionPlaneObservation {
    const { guard, diagnostics } = collectGuard(options);
    const residual = collectResidual();
    const worktree = collectWorktree(options);
    if (guard.status !== "pass") {
      return {
        checkedAt: nowIso(),
        namespace: options.namespace,
        kubeconfig: options.kubeconfig,
        worktreePath: options.worktreePath,
        guard,
        deployments: [],
        pods: [],
        services: [],
        worktree,
        residual,
        commandDiagnostics: { guard: diagnostics },
      };
    }
    return {
      checkedAt: nowIso(),
      namespace: options.namespace,
      kubeconfig: options.kubeconfig,
      worktreePath: options.worktreePath,
      guard,
      deployments: collectDeployments(options),
      pods: collectPods(options),
      services: collectServices(options),
      worktree,
      residual,
      commandDiagnostics: { guard: diagnostics },
    };
  }
}

function safeCommit(value: string | null): string | null {
  if (value === null) return null;
  const trimmed = value.trim().toLowerCase();
  if (!/^[0-9a-f]{7,40}$/u.test(trimmed)) return null;
  return trimmed;
}

function commitMatches(left: string, right: string): boolean {
  const a = left.toLowerCase();
  const b = right.toLowerCase();
  return a === b || (a.length >= 7 && b.startsWith(a)) || (b.length >= 7 && a.startsWith(b));
}

function uniqueStrings(values: Array<string | null>): string[] {
  return [...new Set(values.filter((value): value is string => value !== null && value.length > 0))];
}

function deploymentCommitMarkers(deployment: DeploymentObservation): Record<string, string | null> {
  return {
    annotationDeployCommit: safeCommit(deployment.annotations.deployCommit),
    annotationDeployRequestedCommit: safeCommit(deployment.annotations.deployRequestedCommit),
    unideskDeployCommit: safeCommit(deployment.env.unideskDeployCommit),
    unideskDeployRequestedCommit: safeCommit(deployment.env.unideskDeployRequestedCommit),
    codeQueueDeployCommit: safeCommit(deployment.env.codeQueueDeployCommit),
    codeQueueDeployRequestedCommit: safeCommit(deployment.env.codeQueueDeployRequestedCommit),
  };
}

function deploymentFormalSignals(deployment: DeploymentObservation): DriftSignal[] {
  const signals: DriftSignal[] = [];
  if (!deployment.observed) {
    signals.push({ code: "execution-plane-deployment-missing", severity: "blocker", message: `${deployment.name} deployment is missing`, expected: deployment.name, observed: deployment.error });
    return signals;
  }
  if (deployment.namespace !== expectedNamespace) {
    signals.push({ code: "execution-plane-namespace-mismatch", severity: "blocker", message: `${deployment.name} is not in ${expectedNamespace}`, expected: expectedNamespace, observed: deployment.namespace });
  }
  if (deployment.labels.app !== "code-queue") {
    signals.push({ code: "execution-plane-app-label-mismatch", severity: "blocker", message: `${deployment.name} app label is not code-queue`, expected: "code-queue", observed: deployment.labels.app });
  }
  if (deployment.labels.deploymentMode !== "k3sctl-managed") {
    signals.push({ code: "execution-plane-mode-mismatch", severity: "blocker", message: `${deployment.name} is not k3sctl-managed`, expected: "k3sctl-managed", observed: deployment.labels.deploymentMode });
  }
  if (deployment.expectedRole !== "scheduler" && deployment.labels.component !== deployment.expectedRole) {
    signals.push({ code: "execution-plane-component-mismatch", severity: "blocker", message: `${deployment.name} component label does not match role`, expected: deployment.expectedRole, observed: deployment.labels.component });
  }
  if (deployment.nodeSelector !== "D601" && deployment.nodeSelector !== "d601") {
    signals.push({ code: "execution-plane-node-selector-mismatch", severity: "blocker", message: `${deployment.name} is not pinned to D601`, expected: "D601", observed: deployment.nodeSelector });
  }
  if (deployment.env.serviceRole !== deployment.expectedRole) {
    signals.push({ code: "execution-plane-role-mismatch", severity: "blocker", message: `${deployment.name} CODE_QUEUE_SERVICE_ROLE mismatch`, expected: deployment.expectedRole, observed: deployment.env.serviceRole });
  }
  if (deployment.env.schedulerEnabled !== deployment.expectedSchedulerEnabled) {
    signals.push({ code: "execution-plane-scheduler-flag-mismatch", severity: "blocker", message: `${deployment.name} CODE_QUEUE_SCHEDULER_ENABLED mismatch`, expected: deployment.expectedSchedulerEnabled, observed: deployment.env.schedulerEnabled });
  }
  if (deployment.image === null || !deployment.image.includes("unidesk-code-queue")) {
    signals.push({ code: "execution-plane-image-mismatch", severity: "blocker", message: `${deployment.name} image is not a Code Queue image`, expected: "unidesk-code-queue", observed: deployment.image });
  }
  if (deployment.replicas.ready <= 0) {
    signals.push({ code: "execution-plane-no-ready-replica", severity: "blocker", message: `${deployment.name} has no ready replicas`, expected: "readyReplicas > 0", observed: deployment.replicas.ready });
  }
  return signals;
}

function driftSignals(observation: CodeQueueExecutionPlaneObservation): DriftSignal[] {
  const signals: DriftSignal[] = [];
  const commitValues = observation.deployments.flatMap((deployment) => Object.values(deploymentCommitMarkers(deployment)));
  const uniqueCommits = uniqueStrings(commitValues);
  const worktreeHead = safeCommit(observation.worktree.head);

  if (uniqueCommits.length === 0) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "deployment.commitMarkers",
      message: "Code Queue deployments do not expose comparable deploy commit markers",
      expected: "UNIDESK_DEPLOY_COMMIT or CODE_QUEUE_DEPLOY_COMMIT",
      observed: "missing",
    });
  }
  if (uniqueCommits.length > 1) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "deployment.commitMarkers",
      message: "Code Queue deployment commit markers disagree",
      observed: uniqueCommits,
    });
  }
  for (const deployment of observation.deployments) {
    if (uniqueStrings(Object.values(deploymentCommitMarkers(deployment))).length === 0) {
      signals.push({
        code: "deployment-drift",
        severity: "blocker",
        field: `deployment.${deployment.name}.commitMarkers`,
        message: `${deployment.name} does not expose comparable deploy commit markers`,
        expected: "UNIDESK_DEPLOY_COMMIT or CODE_QUEUE_DEPLOY_COMMIT",
        observed: "missing",
      });
    }
  }
  if (worktreeHead === null) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "mountedWorktree.head",
      message: "Mounted Code Queue worktree HEAD could not be observed",
      expected: observation.worktreePath,
      observed: observation.worktree.error,
    });
  } else if (uniqueCommits.length > 0 && !uniqueCommits.some((commit) => commitMatches(commit, worktreeHead))) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "mountedWorktree.head",
      message: "Mounted worktree HEAD does not match deployment commit markers",
      expected: uniqueCommits,
      observed: worktreeHead,
    });
  }

  const repoHostPaths = uniqueStrings(observation.deployments.map((deployment) => deployment.repoHostPath));
  if (repoHostPaths.length !== 1 || repoHostPaths[0] !== observation.worktreePath) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "deployment.repoHostPath",
      message: "Code Queue deployments do not all mount only the expected worktree path",
      expected: observation.worktreePath,
      observed: repoHostPaths,
    });
  }

  const podDigests = uniqueStrings(observation.pods.map((pod) => pod.digest));
  if (podDigests.length === 0) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "artifact.digest",
      message: "No running Code Queue pod image digest was observed",
      expected: "sha256 digest from pod status imageID",
      observed: "missing",
    });
  } else if (podDigests.length > 1) {
    signals.push({
      code: "deployment-drift",
      severity: "blocker",
      field: "artifact.digest",
      message: "Code Queue pod artifact digests disagree",
      observed: podDigests,
    });
  }

  return signals;
}

function residualSignals(observation: CodeQueueExecutionPlaneObservation): DriftSignal[] {
  const signals: DriftSignal[] = [];
  if (observation.residual.composeBackend.present || observation.residual.loopbackPort4222.present) {
    signals.push({
      code: "deprecated-compose-residual",
      severity: "blocker",
      message: "Deprecated Docker Compose code-queue-backend or old 127.0.0.1:4222 listener is still present",
      observed: {
        composeBackendPresent: observation.residual.composeBackend.present,
        loopbackPort4222Present: observation.residual.loopbackPort4222.present,
      },
    });
  }
  if (!observation.residual.composeBackend.ok) {
    signals.push({
      code: "deprecated-compose-residual-observation-gap",
      severity: "warning",
      message: "Could not observe Docker Compose residual state",
      observed: observation.residual.composeBackend.error,
    });
  }
  if (!observation.residual.loopbackPort4222.ok) {
    signals.push({
      code: "deprecated-port-observation-gap",
      severity: "warning",
      message: "Could not observe local listener state for 127.0.0.1:4222",
      observed: observation.residual.loopbackPort4222.error,
    });
  }
  return signals;
}

function compactDeployment(deployment: DeploymentObservation): Record<string, unknown> {
  return {
    name: deployment.name,
    observed: deployment.observed,
    role: deployment.env.serviceRole,
    expectedRole: deployment.expectedRole,
    schedulerEnabled: deployment.env.schedulerEnabled,
    expectedSchedulerEnabled: deployment.expectedSchedulerEnabled,
    readyReplicas: deployment.replicas.ready,
    image: deployment.image,
    nodeSelector: deployment.nodeSelector,
    deployCommit: deployment.env.codeQueueDeployCommit ?? deployment.env.unideskDeployCommit ?? deployment.annotations.deployCommit,
    requestedCommit: deployment.env.codeQueueDeployRequestedCommit ?? deployment.env.unideskDeployRequestedCommit ?? deployment.annotations.deployRequestedCommit,
    repoHostPath: deployment.repoHostPath,
    error: deployment.error,
  };
}

function evaluateObservation(observation: CodeQueueExecutionPlaneObservation, options: ExecutionPlaneOptions): Record<string, unknown> {
  const formalSignals = observation.deployments.flatMap(deploymentFormalSignals);
  const deploymentDriftSignals = driftSignals(observation);
  const deprecatedResidualSignals = residualSignals(observation);
  const guardSignals: DriftSignal[] = observation.guard.status === "pass"
    ? []
    : [{
        code: "d601-k3s-guard-blocked",
        severity: "blocker",
        message: observation.guard.summary,
        expected: { kubeconfig: d601NativeKubeconfig, node: d601RequiredNodeName },
        observed: { status: observation.guard.status, nodes: observation.guard.nodeNames },
      }];
  const signals = [...guardSignals, ...formalSignals, ...deploymentDriftSignals, ...deprecatedResidualSignals];
  const blockers = signals.filter((signal) => signal.severity === "blocker");
  const warnings = signals.filter((signal) => signal.severity === "warning");
  const status: PlaneStatus = blockers.length > 0 ? "blocked" : warnings.length > 0 ? "degraded" : "ready";
  const formal = formalSignals.length === 0 && observation.deployments.length === expectedDeployments.length;
  const uniqueDigests = uniqueStrings(observation.pods.map((pod) => pod.digest));
  const uniqueCommits = uniqueStrings(observation.deployments.flatMap((deployment) => Object.values(deploymentCommitMarkers(deployment))));
  const result: Record<string, unknown> = {
    ok: status === "ready",
    surface: "code-queue-execution-plane",
    checkedAt: observation.checkedAt,
    devTestClass: "live-read",
    mutation: false,
    namespace: observation.namespace,
    kubeconfig: observation.kubeconfig,
    status,
    runnerDisposition: status === "ready" ? "ready" : "infra-blocked",
    summary: {
      formalExecutionPlane: formal,
      deploymentDrift: deploymentDriftSignals.some((signal) => signal.severity === "blocker"),
      deprecatedComposeResidual: deprecatedResidualSignals.some((signal) => signal.code === "deprecated-compose-residual"),
      blockerCount: blockers.length,
      warningCount: warnings.length,
      deploymentsObserved: observation.deployments.filter((deployment) => deployment.observed).length,
      readyPodCount: observation.pods.filter((pod) => pod.ready).length,
      artifactDigest: uniqueDigests.length === 1 ? uniqueDigests[0] : null,
      deploymentCommit: uniqueCommits.length === 1 ? uniqueCommits[0] : null,
      mountedWorktreeHead: observation.worktree.head,
    },
    guard: {
      status: observation.guard.status,
      kubeconfig: observation.guard.kubeconfig,
      nodeNames: observation.guard.nodeNames,
      requiredNodeName: observation.guard.requiredNodeName,
      requiredNodePresent: observation.guard.requiredNodePresent,
      summary: observation.guard.summary,
    },
    executionPlane: {
      formal,
      expectedDeployments: expectedDeployments.map((deployment) => deployment.name),
      deployments: observation.deployments.map(compactDeployment),
    },
    drift: {
      status: deploymentDriftSignals.some((signal) => signal.severity === "blocker") ? "deployment-drift" : "none",
      signalCount: deploymentDriftSignals.length,
      signals: deploymentDriftSignals.slice(0, options.full ? undefined : 5),
      omittedSignals: options.full ? 0 : Math.max(0, deploymentDriftSignals.length - 5),
      comparedFields: ["deployment env/annotation commit", "artifact digest", "mounted worktree HEAD"],
    },
    residual: {
      status: deprecatedResidualSignals.some((signal) => signal.code === "deprecated-compose-residual") ? "deprecated-compose-residual" : "none",
      composeBackendPresent: observation.residual.composeBackend.present,
      loopbackPort4222Present: observation.residual.loopbackPort4222.present,
      containers: observation.residual.composeBackend.containers,
      listeners: observation.residual.loopbackPort4222.listeners,
    },
    blockers,
    warnings,
    commands: {
      refresh: "bun scripts/cli.ts codex execution-plane",
      full: "bun scripts/cli.ts codex execution-plane --full",
      raw: "bun scripts/cli.ts codex execution-plane --raw",
    },
    disclosure: {
      defaultView: "compact-low-noise",
      fullDetailOmitted: !options.full,
      rawOmitted: !options.raw,
      expandWith: "bun scripts/cli.ts codex execution-plane --full",
      rawWith: "bun scripts/cli.ts codex execution-plane --raw",
      secretValuesPrinted: false,
      rawKubernetesDeploymentJsonPrinted: false,
    },
  };
  if (options.full) {
    result.details = {
      deployments: observation.deployments,
      pods: observation.pods,
      services: observation.services,
      worktree: observation.worktree,
      commandDiagnostics: observation.commandDiagnostics,
    };
  }
  if (options.raw) result.rawObservation = observation;
  return result;
}

export async function runCodeQueueExecutionPlane(args: string[], collector: ExecutionPlaneCollector = new LiveExecutionPlaneCollector()): Promise<Record<string, unknown>> {
  const options = parseExecutionPlaneOptions(args);
  const observation = await collector.collect(options);
  return evaluateObservation(observation, options);
}

export async function runCodeQueueExecutionPlaneForTest(args: string[], observation: CodeQueueExecutionPlaneObservation): Promise<Record<string, unknown>> {
  return runCodeQueueExecutionPlane(args, { collect: () => observation });
}