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; skipProbe: 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; }; } interface JudgeProbeObservation { ok: boolean; attempted: boolean; behaviorVersion: string | null; expectedBehaviorVersion: string; configured: boolean | null; model: string | null; hits: number | null; total: number | null; hitRate: number | null; serviceProxyPath: string | null; error: string | null; raw: unknown; } export interface CodeQueueExecutionPlaneObservation { checkedAt: string; namespace: string; kubeconfig: string; worktreePath: string; guard: D601K3sGuardClassification; deployments: DeploymentObservation[]; pods: PodObservation[]; services: ServiceObservation[]; worktree: WorktreeObservation; residual: ResidualObservation; judgeProbe: JudgeProbeObservation; commandDiagnostics: Record; } interface ExecutionPlaneCollector { collect(options: ExecutionPlaneOptions): Promise | CodeQueueExecutionPlaneObservation; } interface DriftSignal { code: string; severity: SignalSeverity; message: string; expected?: unknown; observed?: unknown; field?: string; } const expectedNamespace = "unidesk"; const expectedWorktreePath = "/home/ubuntu/cq-deploy"; export const expectedJudgeProbeBehaviorVersion = "code-queue-judge-probe:v1"; 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", "--skip-probe"]); 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, skipProbe: hasFlag(args, "--skip-probe"), 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 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 commandProbe(runCommand(["kubectl", ...args], repoRoot, { timeoutMs: options.timeoutMs, env: { ...process.env, KUBECONFIG: options.kubeconfig }, })); } function collectGuard(options: ExecutionPlaneOptions): { guard: D601K3sGuardClassification; diagnostics: Record } { 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 { 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 = commandProbe(runCommand(["git", "-C", options.worktreePath, "rev-parse", "HEAD"], repoRoot, { timeoutMs: 5_000 })); return { path: options.worktreePath, ok: probe.ok, head: probe.ok ? firstLine(probe.stdout) : null, error: safeError(probe), }; } function collectResidual(): ResidualObservation { const docker = commandProbe(runCommand(["docker", "ps", "-a", "--filter", "name=code-queue-backend", "--format", "{{.Names}}\t{{.Status}}\t{{.Image}}"], repoRoot, { timeoutMs: 8_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 = commandProbe(runCommand(["ss", "-H", "-ltnp"], repoRoot, { timeoutMs: 8_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 parseJson(text: string): unknown { try { return JSON.parse(text) as unknown; } catch { return null; } } function asRecord(value: unknown): Record | null { return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record : null; } function stringValue(value: unknown): string | null { return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; } function numberValue(value: unknown): number | null { return typeof value === "number" && Number.isFinite(value) ? value : null; } function collectJudgeProbe(options: ExecutionPlaneOptions): JudgeProbeObservation { if (options.skipProbe) { return { ok: false, attempted: false, behaviorVersion: null, expectedBehaviorVersion: expectedJudgeProbeBehaviorVersion, configured: null, model: null, hits: null, total: null, hitRate: null, serviceProxyPath: null, error: "--skip-probe requested", raw: null, }; } const paths = [ `/api/v1/namespaces/${options.namespace}/services/code-queue/proxy/api/judge/probe`, `/api/v1/namespaces/${options.namespace}/services/code-queue:4222/proxy/api/judge/probe`, `/api/v1/namespaces/${options.namespace}/services/code-queue:http/proxy/api/judge/probe`, ]; for (const path of paths) { const probe = runKubectl(["--request-timeout=15s", "get", "--raw", path], options); if (!probe.ok) continue; const raw = parseJson(probe.stdout); const record = asRecord(raw); if (record === null) { return { ok: false, attempted: true, behaviorVersion: null, expectedBehaviorVersion: expectedJudgeProbeBehaviorVersion, configured: null, model: null, hits: null, total: null, hitRate: null, serviceProxyPath: path, error: "judge probe returned non-JSON response", raw: probe.stdout.slice(0, 2000), }; } return { ok: record.ok === true, attempted: true, behaviorVersion: stringValue(record.behaviorVersion ?? asRecord(record.behavior)?.version ?? record.version), expectedBehaviorVersion: expectedJudgeProbeBehaviorVersion, configured: typeof record.configured === "boolean" ? record.configured : null, model: stringValue(record.model), hits: numberValue(record.hits), total: numberValue(record.total), hitRate: numberValue(record.hitRate), serviceProxyPath: path, error: record.ok === true ? null : stringValue(record.error) ?? "judge probe returned ok=false", raw, }; } return { ok: false, attempted: true, behaviorVersion: null, expectedBehaviorVersion: expectedJudgeProbeBehaviorVersion, configured: null, model: null, hits: null, total: null, hitRate: null, serviceProxyPath: null, error: "judge probe could not be read through Kubernetes API service proxy", raw: null, }; } 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, judgeProbe: { ok: false, attempted: false, behaviorVersion: null, expectedBehaviorVersion: expectedJudgeProbeBehaviorVersion, configured: null, model: null, hits: null, total: null, hitRate: null, serviceProxyPath: null, error: "D601 k3s guard did not pass; runtime probe skipped", raw: null, }, 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, judgeProbe: collectJudgeProbe(options), 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[] { return [...new Set(values.filter((value): value is string => value !== null && value.length > 0))]; } function deploymentCommitMarkers(deployment: DeploymentObservation): Record { 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, }); } if (!observation.judgeProbe.attempted) { signals.push({ code: "deployment-drift", severity: "warning", field: "judgeProbe.behaviorVersion", message: "/api/judge/probe behavior version was not checked", expected: expectedJudgeProbeBehaviorVersion, observed: observation.judgeProbe.error, }); } else if (observation.judgeProbe.behaviorVersion !== expectedJudgeProbeBehaviorVersion) { signals.push({ code: "deployment-drift", severity: "blocker", field: "judgeProbe.behaviorVersion", message: "/api/judge/probe behavior version does not match the repo contract", expected: expectedJudgeProbeBehaviorVersion, observed: observation.judgeProbe.behaviorVersion, }); } else if (!observation.judgeProbe.ok) { signals.push({ code: "deployment-drift", severity: "blocker", field: "judgeProbe.ok", message: "/api/judge/probe returned a non-ready result", expected: true, observed: observation.judgeProbe.error, }); } 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 { 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 compactJudgeProbe(probe: JudgeProbeObservation, full: boolean): Record { return { ok: probe.ok, attempted: probe.attempted, behaviorVersion: probe.behaviorVersion, expectedBehaviorVersion: probe.expectedBehaviorVersion, configured: probe.configured, model: probe.model, hits: probe.hits, total: probe.total, hitRate: probe.hitRate, serviceProxyPath: probe.serviceProxyPath, error: probe.error, ...(full ? { raw: probe.raw } : {}), }; } function evaluateObservation(observation: CodeQueueExecutionPlaneObservation, options: ExecutionPlaneOptions): Record { 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 = { 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, judgeProbeBehaviorVersion: observation.judgeProbe.behaviorVersion, }, 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", "/api/judge/probe behaviorVersion"], }, 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, }, judgeProbe: compactJudgeProbe(observation.judgeProbe, options.full), 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> { const options = parseExecutionPlaneOptions(args); const observation = await collector.collect(options); return evaluateObservation(observation, options); } export async function runCodeQueueExecutionPlaneForTest(args: string[], observation: CodeQueueExecutionPlaneObservation): Promise> { return runCodeQueueExecutionPlane(args, { collect: () => observation }); }