From c935aa530d35b54db30b26bd391b12cfbf1924ca Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jesu=CC=81s=20Pe=CC=81rez?= <jpl@jesusperez.com>
Date: Fri, 10 Jul 2026 06:20:26 +0100
Subject: [PATCH] core: extract cluster-observability slice from components.nu
 (ADR-026)

---
 nulib/cli/cluster.nu            |   3 +-
 nulib/cli/component.nu          |  14 +-
 nulib/cli/components.nu         | 957 +------------------------------
 nulib/cli/components_cluster.nu | 963 ++++++++++++++++++++++++++++++++
 4 files changed, 978 insertions(+), 959 deletions(-)
 create mode 100644 nulib/cli/components_cluster.nu

diff --git a/nulib/cli/cluster.nu b/nulib/cli/cluster.nu
index 1f60a14..25b71b8 100644
--- a/nulib/cli/cluster.nu
+++ b/nulib/cli/cluster.nu
@@ -21,7 +21,8 @@ use platform/user/config.nu [get-workspace-path, get-active-workspace-details]
 use platform/config/accessor.nu [config-get]
 use orchestration/cluster_deploy.nu *
 use orchestration/clusters/discover.nu [discover-clusters, get-cluster-info]
-use cli/components.nu ["main component list", "main cluster top", "main cluster label-nodes", "main cluster plan", "main cluster capacity", "main cluster services"]
+use cli/components.nu ["main component list"]
+use cli/components_cluster.nu ["main cluster top", "main cluster label-nodes", "main cluster plan", "main cluster capacity", "main cluster services"]
 
 def main [
     ...args: string
diff --git a/nulib/cli/component.nu b/nulib/cli/component.nu
index 9af7e62..cb8f888 100644
--- a/nulib/cli/component.nu
+++ b/nulib/cli/component.nu
@@ -39,23 +39,25 @@ use cli/components.nu [
     "main component restore"
     "main component check-updates"
     "main component render"
-    comp-live-check
-    comp-pods-for
-    comp-ssh-pods
-    comp-ssh-k8s-data
-    comp-ssh-describe-pod
     comp-state-server
     comp-kubectl-access
     comp-ws-path
     comp-ws-infra
     comp-load-component-record
+    comp-ncl-export
+]
+use cli/components_cluster.nu [
+    comp-live-check
+    comp-pods-for
+    comp-ssh-pods
+    comp-ssh-k8s-data
+    comp-ssh-describe-pod
     comp-load-server-info
     comp-server-badge
     comp-cluster-summary
     comp-cluster-badge
     comp-cache-last-sync
     comp-settings-k0s-version
-    comp-ncl-export
 ]
 
 def state-color [s: string]: nothing -> string {
diff --git a/nulib/cli/components.nu b/nulib/cli/components.nu
index 0896ed3..b184717 100644
--- a/nulib/cli/components.nu
+++ b/nulib/cli/components.nu
@@ -10,7 +10,7 @@ use primitives/io/service_check.nu [health-gate]
 use primitives/io/logging.nu [is-debug-enabled is-debug-level]
 use platform/integrations/context_assembler.nu [build-cabling-credentials-env]
 
-def cli-err [msg: string] {
+export def cli-err [msg: string] {
     if (is-debug-enabled) {
         error make {msg: $msg}
     } else {
@@ -914,135 +914,6 @@ export def comp-ncl-export [ws_root: string, rel_path: string]: nothing -> recor
     }
 }
 
-# Structured compute capacity for a Hetzner server type code: parsed CPU and RAM
-# for capacity math (cluster top, the placement simulator) plus a human label.
-# Unknown codes yield zero capacity and the code itself as label, so callers can
-# degrade gracefully instead of mis-reporting headroom.
-export def comp-server-capacity [type_code: string]: nothing -> record {
-    let tbl = {
-        cax11: { cpu: 2,  mem_gb: 4,  arch: "ARM" },   cax21: { cpu: 4,  mem_gb: 8,  arch: "ARM" },
-        cax31: { cpu: 8,  mem_gb: 16, arch: "ARM" },   cax41: { cpu: 16, mem_gb: 32, arch: "ARM" },
-        cpx11: { cpu: 2,  mem_gb: 2,  arch: "AMD" },   cpx21: { cpu: 3,  mem_gb: 4,  arch: "AMD" },
-        cpx31: { cpu: 4,  mem_gb: 8,  arch: "AMD" },   cpx41: { cpu: 8,  mem_gb: 16, arch: "AMD" },
-        cpx51: { cpu: 16, mem_gb: 32, arch: "AMD" },
-        cx22:  { cpu: 2,  mem_gb: 4,  arch: "Intel" }, cx32:  { cpu: 4,  mem_gb: 8,  arch: "Intel" },
-        cx42:  { cpu: 8,  mem_gb: 16, arch: "Intel" }, cx52:  { cpu: 16, mem_gb: 32, arch: "Intel" },
-    }
-    let spec = ($tbl | get -o $type_code)
-    if ($spec == null) {
-        { type: $type_code, cpu: 0, cpu_millis: 0, mem_gb: 0, mem_bytes: 0, arch: "", label: $type_code }
-    } else {
-        {
-            type:       $type_code,
-            cpu:        $spec.cpu,
-            cpu_millis: ($spec.cpu * 1000),
-            mem_gb:     $spec.mem_gb,
-            mem_bytes:  ($spec.mem_gb * 1024 * 1024 * 1024),
-            arch:       $spec.arch,
-            label:      $"($spec.cpu)CPU ($spec.mem_gb)GB ($spec.arch)",
-        }
-    }
-}
-
-# Translate Hetzner server type codes to a human-readable CPU/RAM/arch label.
-def hetzner-type-label [type_code: string]: nothing -> string {
-    (comp-server-capacity $type_code).label
-}
-
-def status-dot [status: string]: nothing -> string {
-    match $status {
-        "running"  => "●"
-        "off"      => "○"
-        "starting" => "◐"
-        _          => "○"
-    }
-}
-
-# Load server runtime + declared metadata for all servers in a workspace infra.
-# Cache mode: .servers-state.json (status/public_ip/location) + settings.ncl (type/private_ip).
-# Live mode: hcloud API directly.
-# Returns record keyed by hostname.
-export def comp-load-server-info [
-    ws_root: string
-    infra:   string
-    --live
-]: nothing -> record {
-    let state_path = ($ws_root | path join "infra" $infra ".servers-state.json")
-    let cache = if ($state_path | path exists) { open $state_path } else { {} }
-
-    let declared = do -i {
-        let raw = (comp-ncl-export $ws_root $"infra/($infra)/settings.ncl")
-        $raw | get -o servers | default []
-        | reduce --fold {} {|srv, acc|
-            $acc | insert $srv.hostname {
-                type:       ($srv.server_type? | default ""),
-                private_ip: ($srv | get -o networking.private_ip | default ""),
-                provider:   ($srv.provider? | default "hetzner"),
-            }
-        }
-    } | default {}
-
-    if $live {
-        let res = (do { ^hcloud server list -o json } | complete)
-        if $res.exit_code != 0 { return $declared }
-        ($res.stdout | from json)
-        | reduce --fold {} {|h, acc|
-            let d = ($declared | get -o $h.name | default {})
-            $acc | insert $h.name {
-                status:     $h.status,
-                type:       ($h | get -o server_type.name | default ($d.type? | default "")),
-                location:   ($h | get -o datacenter.location.name | default ""),
-                public_ip:  ($h | get -o public_net.ipv4.ip | default ""),
-                private_ip: ($d.private_ip? | default ""),
-                provider:   ($d.provider? | default "hetzner"),
-            }
-        }
-    } else {
-        $cache | items {|hostname entry|
-            let d = ($declared | get -o $hostname | default {})
-            {
-                key: $hostname,
-                value: {
-                    status:     ($entry.status? | default ""),
-                    type:       ($d.type? | default ""),
-                    location:   ($entry.location? | default ""),
-                    public_ip:  ($entry.public_ip? | default ""),
-                    private_ip: ($d.private_ip? | default ""),
-                    provider:   ($d.provider? | default "hetzner"),
-                },
-            }
-        } | reduce --fold {} {|it, acc| $acc | insert $it.key $it.value}
-    }
-}
-
-# One-line server info badge for the TASKSERV section header.
-export def comp-server-badge [info: record]: nothing -> string {
-    if ($info | is-empty) { return "" }
-    let dot    = (status-dot ($info.status? | default ""))
-    let status = ($info.status? | default "")
-    let type_l = if ($info.type? | default "" | is-not-empty) {
-        hetzner-type-label $info.type
-    } else { "" }
-    let loc    = ($info.location? | default "")
-    let pub    = ($info.public_ip? | default "")
-    let priv   = ($info.private_ip? | default "")
-    let ip_str = if ($pub | is-not-empty) and ($priv | is-not-empty) {
-        $"($pub) / ($priv)"
-    } else if ($pub | is-not-empty) { $pub } else { $priv }
-    let parts = [$"($dot) ($status)" $type_l $loc $ip_str] | where { $in | is-not-empty }
-    if ($parts | is-empty) { "" } else { $parts | str join "  " }
-}
-
-def cluster-cache-path [ws_root: string, infra: string]: nothing -> string {
-    $ws_root | path join "infra" $infra ".cluster-state.json"
-}
-
-# Read the declared k0s version from settings.ncl (settings.k0s.version).
-export def comp-settings-k0s-version [ws_root: string, infra: string]: nothing -> string {
-    let raw = (comp-ncl-export $ws_root $"infra/($infra)/settings.ncl")
-    $raw | get -o settings | default {} | get -o k0s | default {} | get -o version | default ""
-}
-
 # Resolve how to reach the cluster's kube-apiserver for an infra.
 # provisioning NEVER runs kubectl locally: the API server is reachable only from
 # the control-plane node. The SSH host and remote kubectl invocation are declared
@@ -1103,590 +974,6 @@ export def comp-kubectl-ssh [
     comp-ssh-run $info.user $info.key $info.host $remote
 }
 
-# Fetch cluster summary counts via the declared remote kubectl (single SSH call).
-# Returns the same shape as comp-cluster-summary's result record. node/pod totals
-# come from the caller's already-fetched k8s_data; this adds ns/dply/pvc counts.
-def comp-kubectl-summary-counts [ws_root: string, infra: string, version: string, k8s_data: record]: nothing -> record {
-    let nodes = ($k8s_data.nodes? | default [])
-    let pods  = ($k8s_data.pods?  | default [])
-    let nodes_total  = ($nodes | length)
-    let nodes_ready  = ($nodes | where {|n|
-        ($n | get -o status | default {} | get -o conditions | default []) | any { ($in.type? == "Ready") and ($in.status? == "True") }
-    } | length)
-    let pods_total   = ($pods | length)
-    let pods_running = ($pods | where { ($in | get -o status | default {} | get -o phase | default "") == "Running" } | length)
-    let cp_ip = ($nodes | get 0? | default {} | get -o status | default {} | get -o addresses | default [] | where { $in.type? == "InternalIP" } | get 0? | get -o address | default "")
-    let r = (comp-kubectl-ssh $ws_root $infra "get namespaces,deployments,pvc -A --no-headers -o json")
-    let counts = if $r.exit_code == 0 {
-        let items = (try { $r.stdout | from json | get -o items | default [] } catch { [] })
-        {
-            ns:   ($items | where { ($in.kind? | default "") == "Namespace" } | length),
-            dply: ($items | where { ($in.kind? | default "") == "Deployment" } | length),
-            pvc:  ($items | where { (($in.kind? | default "") == "PersistentVolumeClaim") and (($in.status?.phase? | default "") == "Bound") } | length),
-        }
-    } else { { ns: 0, dply: 0, pvc: 0 } }
-    {
-        version:      $version,
-        namespaces:   $counts.ns,
-        deployments:  $counts.dply,
-        pvcs_bound:   $counts.pvc,
-        nodes_ready:  $nodes_ready,
-        nodes_total:  $nodes_total,
-        pods_running: $pods_running,
-        pods_total:   $pods_total,
-        cp_ip:        $cp_ip,
-        last_sync:    (date now | format date "%Y-%m-%dT%H:%M:%SZ"),
-    }
-}
-
-# Fetch (--live) or read cached cluster summary: version, ns/dply/pvc counts, node+pod status.
-# Live: second lightweight SSH (counts only — separate from comp-ssh-k8s-data heavy JSON call).
-# Cache: writes .cluster-state.json on live fetch; reads it on non-live call.
-export def comp-cluster-summary [
-    ws_root:     string
-    infra:       string
-    k8s_data:    record     # node/pod data already fetched by comp-ssh-k8s-data
-    version:     string     # declared k0s/k8s version from settings — caller resolves via comp-settings-k0s-version
-    --live
-]: nothing -> record {
-    let empty = {
-        version: "", namespaces: 0, deployments: 0, pvcs_bound: 0, last_sync: "",
-        nodes_ready: 0, nodes_total: 0, pods_running: 0, pods_total: 0, cp_ip: "",
-    }
-    if not $live {
-        let cache = (cluster-cache-path $ws_root $infra)
-        let base  = if ($cache | path exists) { open $cache } else { $empty }
-        return ($base | merge { version: (if ($version | is-not-empty) { $version } else { $base.version? | default "" }) })
-    }
-    # Live: counts come from the declared remote kubectl (SSH to control-plane).
-    # provisioning never runs kubectl locally — see comp-kubectl-access.
-    let result = (comp-kubectl-summary-counts $ws_root $infra $version $k8s_data)
-    $result | to json | save -f (cluster-cache-path $ws_root $infra)
-    $result
-}
-
-# One-line cluster summary badge for the CLUSTER COMPONENTS header.
-export def comp-cluster-badge [summary: record]: nothing -> string {
-    if ($summary | is-empty) { return "" }
-    let ver  = ($summary.version?   | default "")
-    let ns   = ($summary.namespaces?  | default 0)
-    let pods = ($summary.pods_total?  | default 0)
-    if ($ver | is-empty) and $ns == 0 and $pods == 0 { return "" }
-    let cp_str    = if ($summary.cp_ip?    | default "" | is-not-empty) { $"CP: ($summary.cp_ip)"                          } else { "" }
-    let ver_str   = if ($ver | is-not-empty)                            { $ver                                              } else { "" }
-    let nodes_str = if ($summary.nodes_total? | default 0) > 0         { $"($summary.nodes_ready)/($summary.nodes_total) nodes" } else { "" }
-    let pods_str  = if ($summary.pods_total?  | default 0) > 0         { $"($summary.pods_running)/($summary.pods_total) pods"  } else { "" }
-    let ns_str    = if $ns  > 0 { $"($ns) ns"                       } else { "" }
-    let dply_str  = if ($summary.deployments? | default 0) > 0 { $"($summary.deployments) dply" } else { "" }
-    let pvcs_str  = if ($summary.pvcs_bound?  | default 0) > 0 { $"($summary.pvcs_bound) pvcs"  } else { "" }
-    [$cp_str $ver_str $nodes_str $pods_str $ns_str $dply_str $pvcs_str] | where { $in | is-not-empty } | str join "  "
-}
-
-# Parse a Kubernetes resource quantity (e.g. "529Mi", "1Gi", "1024Ki", "2G") to
-# bytes for numeric sorting and capacity math. Unparseable input yields 0.
-def k8s-quantity-bytes [q: string]: nothing -> int {
-    let s = ($q | str trim)
-    if ($s | is-empty) { return 0 }
-    let m = ($s | parse --regex '(?<num>[0-9.]+)(?<unit>[A-Za-z]*)' | get 0?)
-    if ($m == null) { return 0 }
-    let num  = ($m.num | into float)
-    let mult = (match $m.unit {
-        "Ki" => 1024.0, "Mi" => 1048576.0, "Gi" => 1073741824.0, "Ti" => 1099511627776.0,
-        "k" | "K" => 1000.0, "M" => 1000000.0, "G" => 1000000000.0, "T" => 1000000000000.0,
-        _ => 1.0,
-    })
-    (($num * $mult) | math round | into int)
-}
-
-# Operator-relevant node labels: drop the well-known system/cloud domains so only
-# placement-meaningful labels (node_class, can_use_fip.*, custom) remain. Returns
-# a "k=v k=v" string (valueless labels shown as bare keys).
-def node-operator-labels [labels: record]: nothing -> string {
-    let system = ["kubernetes.io", "k8s.io", "hetzner.cloud", "topology.", "beta.", "csi."]
-    $labels | transpose key value
-    | where {|r| not ($system | any {|p| $r.key | str contains $p }) }
-    | each {|r|
-        # Boolean labels (value "true") and valueless labels show as the bare key —
-        # if it's present it's set. Non-boolean labels keep key=value.
-        if (($r.value | default "" | is-empty) or ($r.value == "true")) { $r.key } else { $"($r.key)=($r.value)" }
-    }
-    | str join " "
-}
-
-# Live per-node resource usage via `kubectl top`, enriched with the declared
-# Hetzner server type and its nominal capacity. provisioning never runs kubectl
-# locally — metrics arrive over the bounded remote-kubectl SSH path
-# (comp-kubectl-ssh). kubectl's own percentages are relative to node *allocatable*
-# (physical RAM minus kernel/kube reservations), which is why the nominal column
-# differs from what the % implies. --pods appends the heaviest pods by memory.
-export def "main cluster top" [
-    mode?: string = ""               # "pods" appends the heaviest pods by memory
-    --workspace (-w): string = ""
-    --infra (-i): string = ""
-    --limit (-l): int = 15
-    --out (-o): string = ""          # json | yaml | table (default terminal view)
-]: nothing -> any {
-    let pods = ($mode in ["pods", "p"])
-    let ws_root = (comp-ws-path $workspace)
-    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
-    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
-    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
-
-    let nodes_res = (comp-kubectl-ssh $ws_root $infra_name "top nodes --no-headers")
-    if $nodes_res.exit_code != 0 {
-        cli-err $"kubectl top nodes failed \(metrics-server reachable?\): ($nodes_res.stderr | str trim)"
-    }
-    let servers = (comp-load-server-info $ws_root $infra_name)
-
-    # Per-node requested/limited memory vs allocatable — the bin-packing signal
-    # kubectl top omits (top reports live usage, not scheduler reservations).
-    # Sourced from pod specs via one SSH JSON fetch; pods in a terminal phase are
-    # excluded to match kubectl's "Allocated resources". no-req counts containers
-    # with no memory request (invisible weight that defeats the scheduler).
-    let access = (comp-kubectl-access $ws_root $infra_name)
-    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
-    let alloc  = ($k8s.nodes | reduce --fold {} {|n, acc|
-        $acc | insert ($n.metadata.name) (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
-    })
-    let labelmap = ($k8s.nodes | reduce --fold {} {|n, acc|
-        $acc | insert ($n.metadata.name) (node-operator-labels ($n.metadata?.labels? | default {}))
-    })
-    let agg = ($k8s.pods
-        | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) }
-        | reduce --fold {} {|p, acc|
-            let node = ($p.spec.nodeName)
-            let cts  = ($p.spec?.containers? | default [])
-            let req  = ($cts | reduce --fold 0 {|c, s| $s + (k8s-quantity-bytes ($c.resources?.requests?.memory? | default "0")) })
-            let lim  = ($cts | reduce --fold 0 {|c, s| $s + (k8s-quantity-bytes ($c.resources?.limits?.memory? | default "0")) })
-            let nor  = ($cts | where { ($in.resources?.requests?.memory? | default "" | is-empty) } | length)
-            let cur  = ($acc | get -o $node | default { req: 0, lim: 0, noreq: 0 })
-            $acc | upsert $node { req: ($cur.req + $req), lim: ($cur.lim + $lim), noreq: ($cur.noreq + $nor) }
-        })
-
-    let node_rows = ($nodes_res.stdout | str replace --all "\r" "" | lines
-        | where { ($in | str trim | is-not-empty) }
-        | each {|ln|
-            let c    = ($ln | str trim | split row --regex '\s+')
-            let name = ($c | get 0? | default "")
-            let type = ($servers | get -o $name | default {} | get -o type | default "")
-            let cap  = (comp-server-capacity $type)
-            let a    = ($alloc | get -o $name | default 0)
-            let g    = ($agg | get -o $name | default { req: 0, lim: 0, noreq: 0 })
-            {
-                node:     $name,
-                type:     (if ($type | is-not-empty) { $type } else { "—" }),
-                nominal:  (if $cap.mem_gb > 0 { $"($cap.cpu)c/($cap.mem_gb)Gi" } else { "—" }),
-                cpu:      ($c | get 1? | default ""),
-                "cpu%":   ($c | get 2? | default ""),
-                mem:      ($c | get 3? | default ""),
-                "mem%":   ($c | get 4? | default ""),
-                "req%":   (if $a > 0 { $"(($g.req * 100 / $a) | math round)%" } else { "—" }),
-                "lim%":   (if $a > 0 { $"(($g.lim * 100 / $a) | math round)%" } else { "—" }),
-                "no-req": $g.noreq,
-                labels:   ($labelmap | get -o $name | default ""),
-            }
-        })
-
-    let pod_rows = if $pods {
-        let pods_res = (comp-kubectl-ssh $ws_root $infra_name "top pods --all-namespaces --no-headers")
-        if $pods_res.exit_code != 0 {
-            print --stderr $"kubectl top pods failed: ($pods_res.stderr | str trim)"
-            []
-        } else {
-            $pods_res.stdout | str replace --all "\r" "" | lines
-            | where { ($in | str trim | is-not-empty) }
-            | each {|ln|
-                let c = ($ln | str trim | split row --regex '\s+')
-                {
-                    namespace: ($c | get 0? | default ""),
-                    pod:       ($c | get 1? | default ""),
-                    cpu:       ($c | get 2? | default ""),
-                    mem:       ($c | get 3? | default ""),
-                    mem_bytes: (k8s-quantity-bytes ($c | get 3? | default "0")),
-                }
-            }
-            | sort-by mem_bytes --reverse
-            | first $limit
-            | reject mem_bytes
-        }
-    } else { [] }
-
-    let data = if $pods { { nodes: $node_rows, pods: $pod_rows } } else { $node_rows }
-    match ($out | str downcase) {
-        "json" => ($data | to json)
-        "yaml" | "yml" => ($data | to yaml)
-        "" | "table" => {
-            if $pods {
-                print "NODES"
-                print ($node_rows | table)
-                cluster-top-legend
-                print ""
-                print $"TOP ($limit) PODS BY MEMORY"
-                $pod_rows
-            } else {
-                print ($node_rows | table)
-                cluster-top-legend
-                null
-            }
-        }
-        _ => { cli-err $"unknown --out format: ($out) \(use json|yaml|table\)" }
-    }
-}
-
-# Interpretive legend for `cluster top` columns — printed under the node table so
-# the numbers can be read without external context. Suppressed in json/yaml output.
-def cluster-top-legend []: nothing -> nothing {
-    print ""
-    print "How to read it:"
-    print "  nominal   physical CPU/RAM of the Hetzner type (cax11=2c/4Gi, cax21=4c/8Gi)."
-    print "  cpu /mem  LIVE usage now (kubectl top). mem% is vs ALLOCATABLE (RAM minus"
-    print "            kernel+kube reservations), so it reads higher than vs nominal."
-    print "  req%      sum of pod memory REQUESTS vs allocatable = what the scheduler"
-    print "            reserved. Healthy < 80%; it is the real 'fullness' for placement."
-    print "  lim%      sum of pod memory LIMITS vs allocatable = overcommit. > 100% means"
-    print "            pods may collectively exceed RAM → OOM-kill risk under pressure."
-    print "  no-req    containers with NO memory request — invisible to the scheduler,"
-    print "            cause blind bin-packing and are first to be OOM-killed."
-    print "  labels    operator node labels (node_class, can_use_fip.*, custom) that drive"
-    print "            placement; system/cloud domains (kubernetes.io, hetzner.cloud) hidden."
-}
-
-# Reconcile declared operator node labels onto the live Kubernetes nodes. Source
-# of truth: servers.ncl node_labels (record) + node_class (list, projected as
-# node-class.<class>=true). Closes the drift where declared labels never reach the
-# nodes (no kubectl label runs at join time). Dry-run by default: prints the plan;
-# --apply runs `kubectl label node ... --overwrite` over the bounded SSH path.
-export def "main cluster label-nodes" [
-    --workspace (-w): string = ""
-    --infra (-i): string = ""
-    --apply (-a)
-]: nothing -> any {
-    let ws_root = (comp-ws-path $workspace)
-    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
-    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
-    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
-
-    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
-    let servers  = ($settings | get -o servers | default []
-        | where { ($in.not_use? | default false) == false })
-
-    # Desired operator labels per node: node_labels record + node_class projection.
-    let desired = ($servers | reduce --fold {} {|s, acc|
-        let base = ($s.node_labels? | default {})
-        let cls  = ($s.node_class? | default [] | reduce --fold {} {|c, m| $m | insert $"node-class.($c)" "true" })
-        $acc | insert ($s.hostname) ($base | merge $cls)
-    })
-
-    let access = (comp-kubectl-access $ws_root $infra_name)
-    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
-    let live   = ($k8s.nodes | reduce --fold {} {|n, acc|
-        $acc | insert ($n.metadata.name) ($n.metadata?.labels? | default {})
-    })
-
-    # Diff via transpose (label keys carry dots/slashes — unsafe for cell-path get).
-    let plan = ($desired | transpose host want | each {|e|
-        let have = (($live | get -o $e.host | default {}) | transpose key value)
-        $e.want | transpose key value | each {|w|
-            let m = ($have | where key == $w.key)
-            let cur = if ($m | is-empty) { null } else { ($m | first | get value) }
-            if $cur != $w.value {
-                { node: $e.host, label: $w.key, current: ($cur | default "—"), desired: $w.value }
-            } else { null }
-        } | compact
-    } | flatten)
-
-    if ($plan | is-empty) {
-        print "All declared node labels already applied — nothing to do."
-        return []
-    }
-    if not $apply {
-        print $"Planned label changes \(($plan | length)\) — dry-run, use --apply to apply:"
-        return $plan
-    }
-    $plan | each {|c|
-        let r = (comp-kubectl-ssh $ws_root $infra_name $"label node ($c.node) ($c.label)=($c.desired) --overwrite")
-        { node: $c.node, label: $c.label, desired: $c.desired, status: (if $r.exit_code == 0 { "applied" } else { ($r.stderr | str trim) }) }
-    }
-}
-
-# Placement simulator: project the per-node impact of (re)deploying a service.
-# Reads the candidate's declared resources_effective (memory request) + placement
-# from settings, the live node snapshot (allocatable, current pod requests, live
-# usage) via the bounded SSH path, then computes projected requests% and
-# working-set% per eligible node (eligibility = node-class label ∩ placement.node_class).
-# Answers "does the next deploy fit, and where?" — the dry-run the ops loop lacked.
-export def "main cluster plan" [
-    service: string
-    --workspace (-w): string = ""
-    --infra (-i): string = ""
-    --out (-o): string = ""
-    --req-threshold: int = 80        # % allocatable requests treated as "full"
-    --use-threshold: int = 90        # % allocatable working-set treated as risky
-]: nothing -> any {
-    let ws_root = (comp-ws-path $workspace)
-    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
-    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
-    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
-
-    # Candidate spec from declared NCL (settings.components.<key>).
-    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
-    let comps = ($settings | get -o components | default {})
-    let key = if (($comps | get -o $service) != null) { $service } else { ($service | str replace --all "-" "_") }
-    let comp = ($comps | get -o $key | default null)
-    if ($comp == null) { cli-err $"component '($service)' not found in settings.components" }
-    let re = ($comp | get -o resources_effective | default {})
-    if ($re | is-empty) { cli-err $"component '($service)' has no resources_effective — declare class/resources first \(step 2c\)" }
-    let cand_req = (k8s-quantity-bytes ($re | get -o memory | default {} | get -o request | default "0"))
-    let cand_lim = (k8s-quantity-bytes ($re | get -o memory | default {} | get -o limit | default "0"))
-    let want_nc  = ($comp | get -o placement | default {} | get -o node_class | default [] | each {|c| $c | into string})
-    let svc_class = ($comp | get -o class | default "")
-
-    let access = (comp-kubectl-access $ws_root $infra_name)
-    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
-    if ($k8s.nodes | is-empty) { cli-err $"could not fetch live nodes \(CP: ($access.host | default 'unresolved'))" }
-    let top    = (comp-kubectl-ssh $ws_root $infra_name "top nodes --no-headers")
-    let usage  = if $top.exit_code == 0 {
-        $top.stdout | str replace --all "\r" "" | lines | where { ($in | str trim | is-not-empty) }
-        | reduce --fold {} {|ln, acc|
-            let c = ($ln | str trim | split row --regex '\s+')
-            $acc | insert ($c | get 0? | default "") (k8s-quantity-bytes ($c | get 3? | default "0"))
-        }
-    } else { {} }
-
-    let live_pods = ($k8s.pods | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) })
-    let rows = ($k8s.nodes | each {|n|
-        let name  = ($n.metadata.name)
-        let alloc = (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
-        let labels = ($n.metadata?.labels? | default {} | transpose k v
-            | where {|r| $r.k | str starts-with "node-class." } | each {|r| $r.k | str replace "node-class." "" })
-        let eligible = if ($want_nc | is-empty) { true } else { ($want_nc | any {|c| $c in $labels }) }
-        let cur_req = ($live_pods | where { ($in.spec.nodeName) == $name }
-            | reduce --fold 0 {|p, s| $s + ($p.spec?.containers? | default [] | reduce --fold 0 {|c, s2| $s2 + (k8s-quantity-bytes ($c.resources?.requests?.memory? | default "0")) }) })
-        let cur_use = ($usage | get -o $name | default 0)
-        let proj_req = ($cur_req + $cand_req)
-        let proj_use = ($cur_use + $cand_req)
-        let req_pct  = (if $alloc > 0 { ($proj_req * 100 / $alloc) | math round } else { 0 })
-        let use_pct  = (if $alloc > 0 { ($proj_use * 100 / $alloc) | math round } else { 0 })
-        {
-            node:      $name,
-            eligible:  $eligible,
-            "req now": (if $alloc > 0 { $"(($cur_req * 100 / $alloc) | math round)%" } else { "—" }),
-            "req→":    (if $alloc > 0 { $"($req_pct)%" } else { "—" }),
-            "ws→":     (if $alloc > 0 { $"($use_pct)%" } else { "—" }),
-            fits:      ($eligible and ($req_pct <= $req_threshold) and ($use_pct <= $use_threshold)),
-            _req_pct:  $req_pct,
-        }
-    })
-
-    let eligible_fit = ($rows | where {|r| $r.eligible and $r.fits } | sort-by _req_pct)
-    let best = ($eligible_fit | get 0? | get -o node | default "")
-    let mem_h = ($re | get -o memory | default {})
-    let summary = {
-        service: $key, class: $svc_class,
-        request: ($mem_h | get -o request | default "?"), limit: ($mem_h | get -o limit | default "?"),
-        node_class: ($want_nc | str join ","),
-        recommendation: (if ($best | is-not-empty) {
-            $"schedules on ($best) \(lowest projected requests among eligible nodes that fit\)"
-        } else {
-            "NO eligible node fits within thresholds — activate another worker (wrk-1) or resize a node"
-        }),
-    }
-
-    let out_rows = ($rows | reject _req_pct)
-    match ($out | str downcase) {
-        "json" => ({ summary: $summary, nodes: $out_rows } | to json)
-        "yaml" | "yml" => ({ summary: $summary, nodes: $out_rows } | to yaml)
-        _ => {
-            print $"plan: ($key)  class=($svc_class)  mem req/lim=($summary.request)/($summary.limit)  prefers node-class=($summary.node_class)"
-            print ($out_rows | table)
-            print $"→ ($summary.recommendation)"
-            null
-        }
-    }
-}
-
-# Format a byte count as a compact Gi/Mi string.
-def fmt-gib [b: int]: nothing -> string {
-    if $b >= 1073741824 { $"(($b / 1073741824 * 10 | math round) / 10)Gi" } else { $"(($b / 1048576) | math round)Mi" }
-}
-
-# Cluster capacity analysis: per-node working-set broken down by category
-# (movable = general app classes, pinned = datastore/storage_engine, infra =
-# kube-system/longhorn/cilium/daemonsets), plus an aggregate capacity check by
-# node-class with what-if projections. Answers "horizontal vs vertical": movable
-# load dominating storage nodes → add a general worker; pinned dominating → resize.
-export def "main cluster capacity" [
-    --workspace (-w): string = ""
-    --infra (-i): string = ""
-    --out (-o): string = ""
-    --add-worker                     # what-if: add one cax11 general worker (~3.4Gi alloc)
-    --resize: string = ""            # what-if: resize <node> cax11→cax21 (+4Gi alloc)
-]: nothing -> any {
-    let ws_root = (comp-ws-path $workspace)
-    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
-    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
-    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
-
-    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
-    let comps = ($settings | get -o components | default {})
-    # pod-name-prefix → pinned?  (component name with _→- is the usual pod prefix)
-    let prefixes = ($comps | items {|k v|
-        { prefix: ($k | str replace --all "_" "-"), pinned: (($v.class? | default "") in ["datastore", "storage_engine"]) }
-    } | sort-by {|c| $c.prefix | str length } --reverse)
-    let infra_ns = ["kube-system", "longhorn-system", "cilium", "cert-manager"]
-
-    let access = (comp-kubectl-access $ws_root $infra_name)
-    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
-    if ($k8s.nodes | is-empty) { cli-err $"could not fetch live nodes \(CP: ($access.host | default 'unresolved'))" }
-    let top = (comp-kubectl-ssh $ws_root $infra_name "top pods --all-namespaces --no-headers")
-    let usage = if $top.exit_code == 0 {
-        $top.stdout | str replace --all "\r" "" | lines | where { ($in | str trim | is-not-empty) }
-        | reduce --fold {} {|ln, acc|
-            let c = ($ln | str trim | split row --regex '\s+')
-            $acc | insert $"($c | get 0? | default '')/($c | get 1? | default '')" (k8s-quantity-bytes ($c | get 3? | default "0"))
-        }
-    } else { {} }
-
-    let live = ($k8s.pods | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) })
-    let per_node = ($k8s.nodes | each {|n|
-        let name  = ($n.metadata.name)
-        let alloc = (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
-        let labels = ($n.metadata?.labels? | default {} | transpose k v
-            | where {|r| $r.k | str starts-with "node-class." } | each {|r| $r.k | str replace "node-class." "" })
-        let pods = ($live | where { ($in.spec.nodeName) == $name })
-        let cats = ($pods | reduce --fold { movable: 0, pinned: 0, infra: 0 } {|p, acc|
-            let ns = ($p.metadata.namespace)
-            let pn = ($p.metadata.name)
-            let use = ($usage | get -o $"($ns)/($pn)" | default 0)
-            let cat = if ($ns in $infra_ns) { "infra" } else {
-                let m = ($prefixes | where {|c| $pn | str starts-with $c.prefix } | get 0? | default null)
-                if ($m == null) { "movable" } else if $m.pinned { "pinned" } else { "movable" }
-            }
-            $acc | upsert $cat (($acc | get $cat) + $use)
-        })
-        { node: $name, class: ($labels | str join ","), alloc: $alloc } | merge $cats
-    })
-
-    # Aggregate by node-class.
-    let general_alloc = ($per_node | where {|r| "general" in ($r.class | split row ",")} | reduce --fold 0 {|r, s| $s + $r.alloc})
-    let storage_alloc = ($per_node | where {|r| "storage" in ($r.class | split row ",")} | reduce --fold 0 {|r, s| $s + $r.alloc})
-    let movable_total = ($per_node | reduce --fold 0 {|r, s| $s + $r.movable})
-    let pinned_total  = ($per_node | reduce --fold 0 {|r, s| $s + $r.pinned})
-
-    let cax11_alloc = 3650722201   # ~3.4Gi typical allocatable on a fresh cax11
-    let resize_node = ($per_node | where node == $resize | get 0? | default null)
-    let resize_is_general = (if ($resize_node != null) { "general" in ($resize_node.class | split row ",") } else { false })
-    let resize_is_storage = (if ($resize_node != null) { "storage" in ($resize_node.class | split row ",") } else { false })
-    let gen_alloc_wi = ($general_alloc
-        + (if $add_worker { $cax11_alloc } else { 0 })
-        + (if $resize_is_general { 4294967296 } else { 0 }))
-    let stg_alloc_wi = ($storage_alloc + (if $resize_is_storage { 4294967296 } else { 0 }))
-
-    let gen_util_now = (if $general_alloc > 0 { ($movable_total * 100 / $general_alloc | math round) } else { 0 })
-    let gen_util_wi  = (if $gen_alloc_wi  > 0 { ($movable_total * 100 / $gen_alloc_wi  | math round) } else { 0 })
-
-    let scenario = if $add_worker { "+1 general worker (cax11)" } else if ($resize | is-not-empty) { $"resize ($resize) → cax21" } else { "current" }
-    let mv_s = (fmt-gib $movable_total)
-    let pn_s = (fmt-gib $pinned_total)
-    let recommendation = if ($movable_total > $pinned_total) {
-        $"movable load ($mv_s) dominates pinned ($pn_s) — HORIZONTAL: add a general worker to relieve the storage nodes; vertical resize would keep apps mislocated"
-    } else {
-        $"pinned/datastore load ($pn_s) dominates movable ($mv_s) — VERTICAL: resize the pressured storage node rather than adding a general worker"
-    }
-
-    let rows = ($per_node | each {|r|
-        {
-            node: $r.node, class: $r.class, alloc: (fmt-gib $r.alloc),
-            movable: (fmt-gib $r.movable), pinned: (fmt-gib $r.pinned), infra: (fmt-gib $r.infra),
-        }
-    })
-    let summary = {
-        movable_total: (fmt-gib $movable_total), pinned_total: (fmt-gib $pinned_total),
-        general_alloc: (fmt-gib $general_alloc), storage_alloc: (fmt-gib $storage_alloc),
-        general_util_now: $"($gen_util_now)%",
-        scenario: $scenario,
-        general_util_scenario: $"($gen_util_wi)%",
-        recommendation: $recommendation,
-    }
-
-    match ($out | str downcase) {
-        "json" => ({ nodes: $rows, summary: $summary } | to json)
-        "yaml" | "yml" => ({ nodes: $rows, summary: $summary } | to yaml)
-        _ => {
-            print "PER-NODE working-set by category (movable=general apps · pinned=datastore · infra=k8s/longhorn)"
-            print ($rows | table)
-            print ""
-            print $"movable total: ($summary.movable_total)   pinned total: ($summary.pinned_total)"
-            print $"general node-class alloc: ($summary.general_alloc)   storage: ($summary.storage_alloc)"
-            print $"general utilization \(movable/general-alloc\): now ($summary.general_util_now)  →  scenario [($scenario)] ($summary.general_util_scenario)"
-            print $"→ ($recommendation)"
-            null
-        }
-    }
-}
-
-# List all Kubernetes services via SSH to the cluster control plane.
-# Matches CLI output of `kubectl get svc --all-namespaces`.
-export def "main cluster services" [
-    --workspace (-w): string = ""
-    --infra (-i): string = ""
-    --namespace (-n): string = ""     # filter to single namespace (default: all)
-    --out (-o): string = ""           # json | yaml | table (default terminal view)
-]: nothing -> any {
-    let ws_root = (comp-ws-path $workspace)
-    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
-    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
-    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
-
-    let ns_flag = if ($namespace | is-not-empty) { $"-n ($namespace)" } else { "--all-namespaces" }
-    let r = (comp-kubectl-ssh $ws_root $infra_name $"get svc ($ns_flag) -o json")
-    if $r.exit_code != 0 {
-        cli-err $"kubectl get svc failed: ($r.stderr | str trim)"
-    }
-
-    let items = (try { $r.stdout | from json | get -o items | default [] } catch { [] })
-    let rows = ($items | each {|svc|
-        let ns       = ($svc.metadata.namespace? | default "")
-        let name     = ($svc.metadata.name? | default "")
-        let svc_type = ($svc.spec.type? | default "ClusterIP")
-        let cip      = ($svc.spec.clusterIP? | default "")
-        let ext_ips  = ($svc.status.loadBalancer?.ingress? | default []
-            | each {|ing| $ing.ip? | default ($ing.hostname? | default "") }
-            | where { $in | is-not-empty } | str join ",")
-        let ports    = ($svc.spec.ports? | default []
-            | each {|p| $"($p.port?):($p.protocol? | default 'TCP')" } | str join " ")
-        let has_https = ($svc.spec.ports? | default [] | any {|p|
-            ($p.port? | default 0) == 443 or (($p.name? | default "") | str downcase | str contains "https")
-        })
-        { namespace: $ns, name: $name, type: $svc_type, cluster_ip: $cip, external_ips: $ext_ips, ports: $ports, https: $has_https }
-    })
-
-    match $out {
-        "json"  => { $rows | to json }
-        "yaml"  => { $rows | to yaml }
-        _       => { $rows }
-    }
-}
-
-# Oldest last_sync across .servers-state.json and .cluster-state.json (= most stale data shown).
-export def comp-cache-last-sync [ws_root: string, infra: string]: nothing -> string {
-    let srv_ts = do {
-        let p = ($ws_root | path join "infra" $infra ".servers-state.json")
-        if ($p | path exists) {
-            open $p | values | each { $in.last_sync? | default "" } | where { $in | is-not-empty }
-        } else { [] }
-    }
-    let cl_ts = do {
-        let p = (cluster-cache-path $ws_root $infra)
-        if ($p | path exists) {
-            let v = ((open $p).last_sync? | default "")
-            if ($v | is-not-empty) { [$v] } else { [] }
-        } else { [] }
-    }
-    let all = ($srv_ts | append $cl_ts) | where { $in | is-not-empty }
-    if ($all | is-empty) { "" } else { $all | sort | first }
-}
-
 # Load DAG formulas for a workspace infra. Returns [] if dag.ncl is absent.
 def comp-load-dag [ws_root: string, infra: string]: nothing -> list {
     let dag_path = ($ws_root | path join "infra" $infra "dag.ncl")
@@ -1818,7 +1105,7 @@ export def comp-state-server [workspace_path: string]: nothing -> string {
 # ─── SSH + k0s kubectl live check infrastructure ─────────────────────────────
 
 # Get SSH connection info for a server (user, key path, host) from servers.ncl.
-def comp-server-ssh-info [ws_root: string, infra: string, server: string]: nothing -> record {
+export def comp-server-ssh-info [ws_root: string, infra: string, server: string]: nothing -> record {
     let servers_path = ($ws_root | path join "infra" $infra "servers.ncl")
     if not ($servers_path | path exists) {
         return { user: "root", key: "", host: $server }
@@ -1853,7 +1140,7 @@ def comp-server-ssh-info [ws_root: string, infra: string, server: string]: nothi
 # ConnectTimeout bounds the connect phase; ServerAlive (5s × 3) tears down a
 # session that goes silent after connect so a stalled remote command can never
 # hang the caller indefinitely. Callers that run kubectl wrap it in `timeout`.
-def comp-ssh-run [user: string, key: string, host: string, cmd: string]: nothing -> record {
+export def comp-ssh-run [user: string, key: string, host: string, cmd: string]: nothing -> record {
     let opts = [
         -o ConnectTimeout=10
         -o ServerAliveInterval=5
@@ -1867,97 +1154,6 @@ def comp-ssh-run [user: string, key: string, host: string, cmd: string]: nothing
     }
 }
 
-# Single SSH session to CP: fetch pods + nodes as JSON in one connection.
-# Returns {pods: list<record>, nodes: list<record>} — parsed from kubectl -o json output.
-# Pods have metadata.namespace/name and status.phase/containerStatuses fields.
-# Nodes have metadata.name and status.conditions fields.
-export def comp-ssh-k8s-data [
-    ws_root:  string,
-    infra:    string,
-    server:   string,
-    --systemd-services: list<string> = [],  # service names to check via systemctl is-active
-]: nothing -> record {
-    let empty = { pods: [], nodes: [], systemd: {} }
-    if ($server | is-empty) {
-        print --stderr "[live] no server resolved — cannot fetch k8s data"
-        return $empty
-    }
-    let info = (comp-server-ssh-info $ws_root $infra $server)
-    if (is-debug-level 2) {
-        let key_part = if ($info.key | is-not-empty) { $" -i ($info.key)" } else { "" }
-        print --stderr $"[live] ssh($key_part) ($info.user)@($info.host) kubectl get pods+nodes -o json"
-    }
-    # PATH: SSH non-interactive sessions may not have /usr/local/bin (where k0s and kubeadm kubectl live).
-    # KUBECONFIG: kubeadm clusters set this in ~/.bashrc (not sourced in BatchMode SSH).
-    # Probe /root/.kube/config then /etc/kubernetes/admin.conf; k0s kubectl ignores KUBECONFIG entirely.
-    let systemd_snippet = if ($systemd_services | is-not-empty) {
-        let checks = ($systemd_services | each {|svc| $"printf '($svc):'; systemctl is-active ($svc) 2>/dev/null || true"} | str join "; ")
-        $"echo '===SYSTEMD==='; ($checks)"
-    } else { "" }
-    let cmd_base = "export PATH=/usr/local/bin:/usr/bin:/bin:$PATH; for _kc in /root/.kube/config /etc/kubernetes/admin.conf; do [ -f \"$_kc\" ] && { export KUBECONFIG=\"$_kc\"; break; }; done; kc() { if command -v kubectl >/dev/null 2>&1; then kubectl \"$@\"; else k0s kubectl \"$@\"; fi; }; echo '===PODS==='; kc get pods --all-namespaces -o json 2>/dev/null || echo '{\"items\":[]}'; echo '===NODES==='; kc get nodes -o json 2>/dev/null || echo '{\"items\":[]}'"
-    let cmd = if ($systemd_snippet | is-not-empty) { $"($cmd_base); ($systemd_snippet)" } else { $cmd_base }
-    let result = (comp-ssh-run $info.user $info.key $info.host $cmd)
-    if (is-debug-level 2) {
-        print --stderr $"[live] SSH exit=($result.exit_code) stdout_len=($result.stdout | str length)"
-    }
-    if $result.exit_code != 0 { return $empty }
-    let all_lines = ($result.stdout | str replace --all "\r" "" | lines)
-    let pods_idx     = ($all_lines | enumerate | where { ($in.item | str trim) == "===PODS===" }     | get 0? | get -o index | default (-1))
-    let nodes_idx    = ($all_lines | enumerate | where { ($in.item | str trim) == "===NODES===" }    | get 0? | get -o index | default (-1))
-    let systemd_idx  = ($all_lines | enumerate | where { ($in.item | str trim) == "===SYSTEMD===" }  | get 0? | get -o index | default (-1))
-    if (is-debug-level 3) {
-        print --stderr $"[live] pods_idx=($pods_idx) nodes_idx=($nodes_idx) systemd_idx=($systemd_idx) total_lines=($all_lines | length)"
-    }
-    let pods = if $pods_idx >= 0 {
-        let from = ($pods_idx + 1)
-        let raw  = if $nodes_idx > $pods_idx { $all_lines | skip $from | first ($nodes_idx - $from) } else { $all_lines | skip $from }
-        try { $raw | str join "\n" | from json | get -o items | default [] } catch {|e|
-            if (is-debug-level 1) {
-                print --stderr $"[live] pods JSON parse error: ($e.msg | str substring 0..120)"
-                print --stderr $"[live] pods first line: ($raw | get 0? | default '')"
-            }
-            []
-        }
-    } else { [] }
-    let nodes = if $nodes_idx >= 0 {
-        let from = ($nodes_idx + 1)
-        let raw  = if $systemd_idx > $nodes_idx { $all_lines | skip $from | first ($systemd_idx - $from) } else { $all_lines | skip $from }
-        try { $raw | str join "\n" | from json | get -o items | default [] } catch {|e|
-            if (is-debug-level 1) {
-                print --stderr $"[live] nodes JSON parse error: ($e.msg | str substring 0..120)"
-            }
-            []
-        }
-    } else { [] }
-    let systemd = if $systemd_idx >= 0 {
-        $all_lines | skip ($systemd_idx + 1)
-        | each {|ln|
-            let parts = ($ln | str trim | split column ":" name status)
-            if ($parts | length) >= 1 and ($parts | get 0 | get name | is-not-empty) {
-                { ($parts | get 0 | get name): ($parts | get 0 | get status | str trim) }
-            } else { {} }
-        }
-        | reduce -f {} {|it, acc| $acc | merge $it }
-    } else { {} }
-    if (is-debug-level 2) {
-        print --stderr $"[live] parsed pods=($pods | length) nodes=($nodes | length) systemd_keys=($systemd | columns | length)"
-    }
-    { pods: $pods, nodes: $nodes, systemd: $systemd }
-}
-
-# Public API: return pod records for a server (used by describe-pod and external callers).
-export def comp-ssh-pods [ws_root: string, infra: string, server: string]: nothing -> list {
-    (comp-ssh-k8s-data $ws_root $infra $server).pods
-}
-
-# Run kubectl describe pod -n <namespace> <pod> via SSH. Returns describe output as string.
-export def comp-ssh-describe-pod [ws_root: string, infra: string, server: string, namespace: string, pod: string]: nothing -> string {
-    let info = (comp-server-ssh-info $ws_root $infra $server)
-    let cmd  = $"if command -v kubectl >/dev/null 2>&1; then kubectl describe pod -n ($namespace) ($pod); else k0s kubectl describe pod -n ($namespace) ($pod); fi"
-    let result = (comp-ssh-run $info.user $info.key $info.host $cmd)
-    if $result.exit_code != 0 { $result.stderr | str trim } else { $result.stdout | str trim }
-}
-
 # Resolve workspace root path (exported for use by component.nu dispatcher).
 export def comp-ws-path [workspace: string]: nothing -> string {
     comp-resolve-ws-path $workspace
@@ -1974,151 +1170,8 @@ export def comp-ws-infra [ws_root: string]: nothing -> string {
     | default ""
 }
 
-# Summarise pod records (from kubectl -o json) into {live, detail}.
-def comp-kubectl-summarise [pods: list]: nothing -> record {
-    if ($pods | is-empty) { return { live: "deployed", detail: "no pods" } }
-    # Exclude Succeeded pods (completed jobs/migrations) — they never return to Running
-    # and would distort the ratio. Failed phase and error-waiting pods are kept and
-    # classified as degraded.
-    let active  = ($pods | where { ($in | get -o status | default {} | get -o phase | default "") != "Succeeded" })
-    let total   = ($active | length)
-    let running = ($active | where { ($in | get -o status | default {} | get -o phase | default "") == "Running" } | length)
-    let pending = ($active | where { ($in | get -o status | default {} | get -o phase | default "") == "Pending"  } | length)
-
-    # Any pod in Failed phase OR any container waiting with an error reason → degraded.
-    let error_reasons = ["CrashLoopBackOff", "ImagePullBackOff", "ErrImagePull",
-                         "Error", "OOMKilled", "CreateContainerError",
-                         "CreateContainerConfigError", "RunContainerError",
-                         "InvalidImageName", "PostStartHookError"]
-    let failing = ($active | where { |p|
-        let phase = ($p | get -o status | default {} | get -o phase | default "")
-        if $phase == "Failed" { true } else {
-            let cs  = ($p | get -o status | default {} | get -o containerStatuses     | default [])
-            let ics = ($p | get -o status | default {} | get -o initContainerStatuses | default [])
-            ($cs | append $ics) | any { |c|
-                ($c | get -o state | default {} | get -o waiting | default {} | get -o reason | default "") in $error_reasons
-            }
-        }
-    } | length)
-
-    let live = if $failing > 0       { "degraded"
-    } else if $pending > 0           { "starting"
-    } else if $total == 0            { "deployed"
-    } else if $running == $total     { "healthy"
-    } else                           { "partial" }
-
-    let detail = if $failing > 0 {
-        $"($running)/($total) pods running  ($failing) failing"
-    } else {
-        $"($running)/($total) pods running"
-    }
-    { live: $live, detail: $detail }
-}
-
-# Filter pod records by namespace (exact match) and name substring.
-def comp-filter-pods [namespace: string, search: string, pods: list]: nothing -> list {
-    let ns_filtered = if ($namespace | is-not-empty) {
-        $pods | where { ($in | get -o metadata | default {} | get -o namespace | default "") == $namespace }
-    } else { $pods }
-    $ns_filtered | where { ($in | get -o metadata | default {} | get -o name | default "") | str contains $search }
-}
-
-# Summarise node records (from kubectl -o json) into {live, detail}.
-def comp-kubectl-nodes-summarise [nodes: list]: nothing -> record {
-    if ($nodes | is-empty) { return { live: "—", detail: "no node data" } }
-    let total = ($nodes | length)
-    let ready = ($nodes | where { |n|
-        $n | get -o status | default {} | get -o conditions | default []
-        | any { ($in | get -o type | default "") == "Ready" and ($in | get -o status | default "") == "True" }
-    } | length)
-    let live = if $ready == $total { "healthy"
-    } else if $ready == 0          { "degraded"
-    } else                         { "partial" }
-    { live: $live, detail: $"($ready)/($total) nodes Ready" }
-}
-
-# Live check for a component using pre-fetched k8s_data {pods, nodes}.
-# Resolves strategy from comp.live_check; falls back to namespace/pod_selector on the component.
-# 'systemd strategy always returns "—" in live (fast) mode — requires prvng c health.
-# Returns {live: string, detail: string}.
-export def comp-live-check [name: string, comp: record, k8s_data: record]: nothing -> record {
-    let lc       = ($comp | get -o live_check | default {})
-    let strategy = ($lc | get -o strategy | default "none" | into string)
-    match $strategy {
-        "k8s_pods" => {
-            if ($k8s_data.pods | is-empty) { return { live: "—", detail: "no pod data" } }
-            let ns  = do {
-                let explicit = ($lc | get -o namespace | default "")
-                if ($explicit | is-not-empty) { $explicit } else { $comp | get -o namespace | default "" }
-            }
-            let sel = do {
-                let explicit = ($lc | get -o selector | default "")
-                if ($explicit | is-not-empty) { $explicit } else {
-                    let ps = ($comp | get -o pod_selector | default "")
-                    if ($ps | is-not-empty) { $ps } else { $name | str replace --all "_" "-" }
-                }
-            }
-            let matching = (comp-filter-pods $ns $sel $k8s_data.pods)
-            if ($matching | is-empty) { return { live: "—", detail: $"no pods: ($sel)" } }
-            comp-kubectl-summarise $matching
-        }
-        "k8s_nodes" => {
-            if ($k8s_data.nodes | is-empty) { return { live: "—", detail: "no node data" } }
-            let sel = ($lc | get -o selector | default "")
-            let filtered = if ($sel | is-not-empty) {
-                $k8s_data.nodes | where { ($in | get -o metadata | default {} | get -o name | default "") | str contains $sel }
-            } else {
-                $k8s_data.nodes
-            }
-            comp-kubectl-nodes-summarise $filtered
-        }
-        "k8s_api" => {
-            # apiserver healthy if kubectl returned any node data
-            if ($k8s_data.nodes | is-not-empty) or ($k8s_data.pods | is-not-empty) {
-                { live: "healthy", detail: "apiserver ok" }
-            } else {
-                { live: "—", detail: "apiserver unreachable" }
-            }
-        }
-        "systemd" => {
-            let svc = ($lc | get -o service | default "")
-            if ($svc | is-empty) { return { live: "—", detail: "health: systemd" } }
-            let units = ($k8s_data | get -o systemd | default {})
-            if ($units | is-empty) { return { live: "—", detail: "health: systemd" } }
-            let active = ($units | get -o $svc | default "")
-            match $active {
-                "active"   => { live: "healthy", detail: $"systemd: ($svc) active" },
-                "inactive" => { live: "degraded", detail: $"systemd: ($svc) inactive" },
-                "failed"   => { live: "degraded", detail: $"systemd: ($svc) failed" },
-                _          => { live: "—", detail: $"systemd: ($svc) ($active)" },
-            }
-        }
-        "on_demand" => { live: "—", detail: "on-demand" }
-        _ => { live: "—", detail: "no live check" }
-    }
-}
-
-# Extract pod summary records for a component from pre-fetched pod JSON records.
-# Returns list of {namespace, pod, ready, status}.
-export def comp-pods-for [namespace: string, name: string, pods: list, pod_selector: string = ""]: nothing -> list {
-    let search = if ($pod_selector | is-not-empty) { $pod_selector } else { $name | str replace --all "_" "-" }
-    (comp-filter-pods $namespace $search $pods)
-    | each {|p|
-        let meta   = ($p | get -o metadata | default {})
-        let status = ($p | get -o status   | default {})
-        let cs     = ($status | get -o containerStatuses | default [])
-        let ready  = ($cs | where { $in.ready? | default false } | length)
-        {
-            namespace: ($meta | get -o namespace | default ""),
-            pod:       ($meta | get -o name      | default ""),
-            ready:     $"($ready)/($cs | length)",
-            status:    ($status | get -o phase   | default ""),
-        }
-    }
-}
-
 # Summarise a list of dag state entries into a single string (for list column).
-def comp-dag-summary [dag: list]: nothing -> string {
+export def comp-dag-summary [dag: list]: nothing -> string {
     if ($dag | is-empty) { return "—" }
     let states = ($dag | each {|e| $e.state })
     if ($states | any { $in == "running"   }) { return "running" }
@@ -2174,7 +1227,7 @@ def project-root-from-prov []: nothing -> string {
 }
 
 # Returns the workspace ROOT PATH (not name) — works for both registered and unregistered workspaces.
-def comp-resolve-ws-path [workspace: string]: nothing -> string {
+export def comp-resolve-ws-path [workspace: string]: nothing -> string {
     # 1. Explicit --workspace → resolve to a path.
     if ($workspace | is-not-empty) {
         # 1a. Absolute or CWD-relative path that is itself a workspace root.
diff --git a/nulib/cli/components_cluster.nu b/nulib/cli/components_cluster.nu
new file mode 100644
index 0000000..d2734d2
--- /dev/null
+++ b/nulib/cli/components_cluster.nu
@@ -0,0 +1,963 @@
+use domain/workspace/mod.nu *
+use platform/user/config.nu [get-active-workspace-details list-workspaces]
+use tools/nickel/process.nu [ncl-eval, ncl-eval-soft, default-ncl-paths]
+use domain/workspace/state.nu [state-read state-write state-node-get state-node-set state-node-start state-node-finish state-node-delete state-node-reset log-trim]
+use platform/clients/orchestrator.nu [orch-url, orch-submit-component, orch-wait-task, orch-health]
+use orchestration/taskservs/run.nu [run_taskserv]
+use platform/config/accessor.nu [get-taskservs-path]
+use domain/utils/templates.nu [on_template_path run_from_template]
+use primitives/io/service_check.nu [health-gate]
+use primitives/io/logging.nu [is-debug-enabled is-debug-level]
+use platform/integrations/context_assembler.nu [build-cabling-credentials-env]
+
+
+# Cluster-observability slice extracted from components.nu (ADR-026 layer hygiene).
+use cli/components.nu [cli-err comp-ncl-export comp-ws-path comp-ws-infra comp-server-ssh-info comp-ssh-run comp-state-server comp-resolve-ws-path comp-dag-summary comp-kubectl-ssh comp-kubectl-access]
+
+# Structured compute capacity for a Hetzner server type code: parsed CPU and RAM
+# for capacity math (cluster top, the placement simulator) plus a human label.
+# Unknown codes yield zero capacity and the code itself as label, so callers can
+# degrade gracefully instead of mis-reporting headroom.
+export def comp-server-capacity [type_code: string]: nothing -> record {
+    let tbl = {
+        cax11: { cpu: 2,  mem_gb: 4,  arch: "ARM" },   cax21: { cpu: 4,  mem_gb: 8,  arch: "ARM" },
+        cax31: { cpu: 8,  mem_gb: 16, arch: "ARM" },   cax41: { cpu: 16, mem_gb: 32, arch: "ARM" },
+        cpx11: { cpu: 2,  mem_gb: 2,  arch: "AMD" },   cpx21: { cpu: 3,  mem_gb: 4,  arch: "AMD" },
+        cpx31: { cpu: 4,  mem_gb: 8,  arch: "AMD" },   cpx41: { cpu: 8,  mem_gb: 16, arch: "AMD" },
+        cpx51: { cpu: 16, mem_gb: 32, arch: "AMD" },
+        cx22:  { cpu: 2,  mem_gb: 4,  arch: "Intel" }, cx32:  { cpu: 4,  mem_gb: 8,  arch: "Intel" },
+        cx42:  { cpu: 8,  mem_gb: 16, arch: "Intel" }, cx52:  { cpu: 16, mem_gb: 32, arch: "Intel" },
+    }
+    let spec = ($tbl | get -o $type_code)
+    if ($spec == null) {
+        { type: $type_code, cpu: 0, cpu_millis: 0, mem_gb: 0, mem_bytes: 0, arch: "", label: $type_code }
+    } else {
+        {
+            type:       $type_code,
+            cpu:        $spec.cpu,
+            cpu_millis: ($spec.cpu * 1000),
+            mem_gb:     $spec.mem_gb,
+            mem_bytes:  ($spec.mem_gb * 1024 * 1024 * 1024),
+            arch:       $spec.arch,
+            label:      $"($spec.cpu)CPU ($spec.mem_gb)GB ($spec.arch)",
+        }
+    }
+}
+
+# Translate Hetzner server type codes to a human-readable CPU/RAM/arch label.
+def hetzner-type-label [type_code: string]: nothing -> string {
+    (comp-server-capacity $type_code).label
+}
+
+def status-dot [status: string]: nothing -> string {
+    match $status {
+        "running"  => "●"
+        "off"      => "○"
+        "starting" => "◐"
+        _          => "○"
+    }
+}
+
+# Load server runtime + declared metadata for all servers in a workspace infra.
+# Cache mode: .servers-state.json (status/public_ip/location) + settings.ncl (type/private_ip).
+# Live mode: hcloud API directly.
+# Returns record keyed by hostname.
+export def comp-load-server-info [
+    ws_root: string
+    infra:   string
+    --live
+]: nothing -> record {
+    let state_path = ($ws_root | path join "infra" $infra ".servers-state.json")
+    let cache = if ($state_path | path exists) { open $state_path } else { {} }
+
+    let declared = do -i {
+        let raw = (comp-ncl-export $ws_root $"infra/($infra)/settings.ncl")
+        $raw | get -o servers | default []
+        | reduce --fold {} {|srv, acc|
+            $acc | insert $srv.hostname {
+                type:       ($srv.server_type? | default ""),
+                private_ip: ($srv | get -o networking.private_ip | default ""),
+                provider:   ($srv.provider? | default "hetzner"),
+            }
+        }
+    } | default {}
+
+    if $live {
+        let res = (do { ^hcloud server list -o json } | complete)
+        if $res.exit_code != 0 { return $declared }
+        ($res.stdout | from json)
+        | reduce --fold {} {|h, acc|
+            let d = ($declared | get -o $h.name | default {})
+            $acc | insert $h.name {
+                status:     $h.status,
+                type:       ($h | get -o server_type.name | default ($d.type? | default "")),
+                location:   ($h | get -o datacenter.location.name | default ""),
+                public_ip:  ($h | get -o public_net.ipv4.ip | default ""),
+                private_ip: ($d.private_ip? | default ""),
+                provider:   ($d.provider? | default "hetzner"),
+            }
+        }
+    } else {
+        $cache | items {|hostname entry|
+            let d = ($declared | get -o $hostname | default {})
+            {
+                key: $hostname,
+                value: {
+                    status:     ($entry.status? | default ""),
+                    type:       ($d.type? | default ""),
+                    location:   ($entry.location? | default ""),
+                    public_ip:  ($entry.public_ip? | default ""),
+                    private_ip: ($d.private_ip? | default ""),
+                    provider:   ($d.provider? | default "hetzner"),
+                },
+            }
+        } | reduce --fold {} {|it, acc| $acc | insert $it.key $it.value}
+    }
+}
+
+# One-line server info badge for the TASKSERV section header.
+export def comp-server-badge [info: record]: nothing -> string {
+    if ($info | is-empty) { return "" }
+    let dot    = (status-dot ($info.status? | default ""))
+    let status = ($info.status? | default "")
+    let type_l = if ($info.type? | default "" | is-not-empty) {
+        hetzner-type-label $info.type
+    } else { "" }
+    let loc    = ($info.location? | default "")
+    let pub    = ($info.public_ip? | default "")
+    let priv   = ($info.private_ip? | default "")
+    let ip_str = if ($pub | is-not-empty) and ($priv | is-not-empty) {
+        $"($pub) / ($priv)"
+    } else if ($pub | is-not-empty) { $pub } else { $priv }
+    let parts = [$"($dot) ($status)" $type_l $loc $ip_str] | where { $in | is-not-empty }
+    if ($parts | is-empty) { "" } else { $parts | str join "  " }
+}
+
+def cluster-cache-path [ws_root: string, infra: string]: nothing -> string {
+    $ws_root | path join "infra" $infra ".cluster-state.json"
+}
+
+# Read the declared k0s version from settings.ncl (settings.k0s.version).
+export def comp-settings-k0s-version [ws_root: string, infra: string]: nothing -> string {
+    let raw = (comp-ncl-export $ws_root $"infra/($infra)/settings.ncl")
+    $raw | get -o settings | default {} | get -o k0s | default {} | get -o version | default ""
+}
+
+# Fetch cluster summary counts via the declared remote kubectl (single SSH call).
+# Returns the same shape as comp-cluster-summary's result record. node/pod totals
+# come from the caller's already-fetched k8s_data; this adds ns/dply/pvc counts.
+def comp-kubectl-summary-counts [ws_root: string, infra: string, version: string, k8s_data: record]: nothing -> record {
+    let nodes = ($k8s_data.nodes? | default [])
+    let pods  = ($k8s_data.pods?  | default [])
+    let nodes_total  = ($nodes | length)
+    let nodes_ready  = ($nodes | where {|n|
+        ($n | get -o status | default {} | get -o conditions | default []) | any { ($in.type? == "Ready") and ($in.status? == "True") }
+    } | length)
+    let pods_total   = ($pods | length)
+    let pods_running = ($pods | where { ($in | get -o status | default {} | get -o phase | default "") == "Running" } | length)
+    let cp_ip = ($nodes | get 0? | default {} | get -o status | default {} | get -o addresses | default [] | where { $in.type? == "InternalIP" } | get 0? | get -o address | default "")
+    let r = (comp-kubectl-ssh $ws_root $infra "get namespaces,deployments,pvc -A --no-headers -o json")
+    let counts = if $r.exit_code == 0 {
+        let items = (try { $r.stdout | from json | get -o items | default [] } catch { [] })
+        {
+            ns:   ($items | where { ($in.kind? | default "") == "Namespace" } | length),
+            dply: ($items | where { ($in.kind? | default "") == "Deployment" } | length),
+            pvc:  ($items | where { (($in.kind? | default "") == "PersistentVolumeClaim") and (($in.status?.phase? | default "") == "Bound") } | length),
+        }
+    } else { { ns: 0, dply: 0, pvc: 0 } }
+    {
+        version:      $version,
+        namespaces:   $counts.ns,
+        deployments:  $counts.dply,
+        pvcs_bound:   $counts.pvc,
+        nodes_ready:  $nodes_ready,
+        nodes_total:  $nodes_total,
+        pods_running: $pods_running,
+        pods_total:   $pods_total,
+        cp_ip:        $cp_ip,
+        last_sync:    (date now | format date "%Y-%m-%dT%H:%M:%SZ"),
+    }
+}
+
+# Fetch (--live) or read cached cluster summary: version, ns/dply/pvc counts, node+pod status.
+# Live: second lightweight SSH (counts only — separate from comp-ssh-k8s-data heavy JSON call).
+# Cache: writes .cluster-state.json on live fetch; reads it on non-live call.
+export def comp-cluster-summary [
+    ws_root:     string
+    infra:       string
+    k8s_data:    record     # node/pod data already fetched by comp-ssh-k8s-data
+    version:     string     # declared k0s/k8s version from settings — caller resolves via comp-settings-k0s-version
+    --live
+]: nothing -> record {
+    let empty = {
+        version: "", namespaces: 0, deployments: 0, pvcs_bound: 0, last_sync: "",
+        nodes_ready: 0, nodes_total: 0, pods_running: 0, pods_total: 0, cp_ip: "",
+    }
+    if not $live {
+        let cache = (cluster-cache-path $ws_root $infra)
+        let base  = if ($cache | path exists) { open $cache } else { $empty }
+        return ($base | merge { version: (if ($version | is-not-empty) { $version } else { $base.version? | default "" }) })
+    }
+    # Live: counts come from the declared remote kubectl (SSH to control-plane).
+    # provisioning never runs kubectl locally — see comp-kubectl-access.
+    let result = (comp-kubectl-summary-counts $ws_root $infra $version $k8s_data)
+    $result | to json | save -f (cluster-cache-path $ws_root $infra)
+    $result
+}
+
+# One-line cluster summary badge for the CLUSTER COMPONENTS header.
+export def comp-cluster-badge [summary: record]: nothing -> string {
+    if ($summary | is-empty) { return "" }
+    let ver  = ($summary.version?   | default "")
+    let ns   = ($summary.namespaces?  | default 0)
+    let pods = ($summary.pods_total?  | default 0)
+    if ($ver | is-empty) and $ns == 0 and $pods == 0 { return "" }
+    let cp_str    = if ($summary.cp_ip?    | default "" | is-not-empty) { $"CP: ($summary.cp_ip)"                          } else { "" }
+    let ver_str   = if ($ver | is-not-empty)                            { $ver                                              } else { "" }
+    let nodes_str = if ($summary.nodes_total? | default 0) > 0         { $"($summary.nodes_ready)/($summary.nodes_total) nodes" } else { "" }
+    let pods_str  = if ($summary.pods_total?  | default 0) > 0         { $"($summary.pods_running)/($summary.pods_total) pods"  } else { "" }
+    let ns_str    = if $ns  > 0 { $"($ns) ns"                       } else { "" }
+    let dply_str  = if ($summary.deployments? | default 0) > 0 { $"($summary.deployments) dply" } else { "" }
+    let pvcs_str  = if ($summary.pvcs_bound?  | default 0) > 0 { $"($summary.pvcs_bound) pvcs"  } else { "" }
+    [$cp_str $ver_str $nodes_str $pods_str $ns_str $dply_str $pvcs_str] | where { $in | is-not-empty } | str join "  "
+}
+
+# Parse a Kubernetes resource quantity (e.g. "529Mi", "1Gi", "1024Ki", "2G") to
+# bytes for numeric sorting and capacity math. Unparseable input yields 0.
+def k8s-quantity-bytes [q: string]: nothing -> int {
+    let s = ($q | str trim)
+    if ($s | is-empty) { return 0 }
+    let m = ($s | parse --regex '(?<num>[0-9.]+)(?<unit>[A-Za-z]*)' | get 0?)
+    if ($m == null) { return 0 }
+    let num  = ($m.num | into float)
+    let mult = (match $m.unit {
+        "Ki" => 1024.0, "Mi" => 1048576.0, "Gi" => 1073741824.0, "Ti" => 1099511627776.0,
+        "k" | "K" => 1000.0, "M" => 1000000.0, "G" => 1000000000.0, "T" => 1000000000000.0,
+        _ => 1.0,
+    })
+    (($num * $mult) | math round | into int)
+}
+
+# Operator-relevant node labels: drop the well-known system/cloud domains so only
+# placement-meaningful labels (node_class, can_use_fip.*, custom) remain. Returns
+# a "k=v k=v" string (valueless labels shown as bare keys).
+def node-operator-labels [labels: record]: nothing -> string {
+    let system = ["kubernetes.io", "k8s.io", "hetzner.cloud", "topology.", "beta.", "csi."]
+    $labels | transpose key value
+    | where {|r| not ($system | any {|p| $r.key | str contains $p }) }
+    | each {|r|
+        # Boolean labels (value "true") and valueless labels show as the bare key —
+        # if it's present it's set. Non-boolean labels keep key=value.
+        if (($r.value | default "" | is-empty) or ($r.value == "true")) { $r.key } else { $"($r.key)=($r.value)" }
+    }
+    | str join " "
+}
+
+# Live per-node resource usage via `kubectl top`, enriched with the declared
+# Hetzner server type and its nominal capacity. provisioning never runs kubectl
+# locally — metrics arrive over the bounded remote-kubectl SSH path
+# (comp-kubectl-ssh). kubectl's own percentages are relative to node *allocatable*
+# (physical RAM minus kernel/kube reservations), which is why the nominal column
+# differs from what the % implies. --pods appends the heaviest pods by memory.
+export def "main cluster top" [
+    mode?: string = ""               # "pods" appends the heaviest pods by memory
+    --workspace (-w): string = ""
+    --infra (-i): string = ""
+    --limit (-l): int = 15
+    --out (-o): string = ""          # json | yaml | table (default terminal view)
+]: nothing -> any {
+    let pods = ($mode in ["pods", "p"])
+    let ws_root = (comp-ws-path $workspace)
+    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
+    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
+    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
+
+    let nodes_res = (comp-kubectl-ssh $ws_root $infra_name "top nodes --no-headers")
+    if $nodes_res.exit_code != 0 {
+        cli-err $"kubectl top nodes failed \(metrics-server reachable?\): ($nodes_res.stderr | str trim)"
+    }
+    let servers = (comp-load-server-info $ws_root $infra_name)
+
+    # Per-node requested/limited memory vs allocatable — the bin-packing signal
+    # kubectl top omits (top reports live usage, not scheduler reservations).
+    # Sourced from pod specs via one SSH JSON fetch; pods in a terminal phase are
+    # excluded to match kubectl's "Allocated resources". no-req counts containers
+    # with no memory request (invisible weight that defeats the scheduler).
+    let access = (comp-kubectl-access $ws_root $infra_name)
+    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
+    let alloc  = ($k8s.nodes | reduce --fold {} {|n, acc|
+        $acc | insert ($n.metadata.name) (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
+    })
+    let labelmap = ($k8s.nodes | reduce --fold {} {|n, acc|
+        $acc | insert ($n.metadata.name) (node-operator-labels ($n.metadata?.labels? | default {}))
+    })
+    let agg = ($k8s.pods
+        | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) }
+        | reduce --fold {} {|p, acc|
+            let node = ($p.spec.nodeName)
+            let cts  = ($p.spec?.containers? | default [])
+            let req  = ($cts | reduce --fold 0 {|c, s| $s + (k8s-quantity-bytes ($c.resources?.requests?.memory? | default "0")) })
+            let lim  = ($cts | reduce --fold 0 {|c, s| $s + (k8s-quantity-bytes ($c.resources?.limits?.memory? | default "0")) })
+            let nor  = ($cts | where { ($in.resources?.requests?.memory? | default "" | is-empty) } | length)
+            let cur  = ($acc | get -o $node | default { req: 0, lim: 0, noreq: 0 })
+            $acc | upsert $node { req: ($cur.req + $req), lim: ($cur.lim + $lim), noreq: ($cur.noreq + $nor) }
+        })
+
+    let node_rows = ($nodes_res.stdout | str replace --all "\r" "" | lines
+        | where { ($in | str trim | is-not-empty) }
+        | each {|ln|
+            let c    = ($ln | str trim | split row --regex '\s+')
+            let name = ($c | get 0? | default "")
+            let type = ($servers | get -o $name | default {} | get -o type | default "")
+            let cap  = (comp-server-capacity $type)
+            let a    = ($alloc | get -o $name | default 0)
+            let g    = ($agg | get -o $name | default { req: 0, lim: 0, noreq: 0 })
+            {
+                node:     $name,
+                type:     (if ($type | is-not-empty) { $type } else { "—" }),
+                nominal:  (if $cap.mem_gb > 0 { $"($cap.cpu)c/($cap.mem_gb)Gi" } else { "—" }),
+                cpu:      ($c | get 1? | default ""),
+                "cpu%":   ($c | get 2? | default ""),
+                mem:      ($c | get 3? | default ""),
+                "mem%":   ($c | get 4? | default ""),
+                "req%":   (if $a > 0 { $"(($g.req * 100 / $a) | math round)%" } else { "—" }),
+                "lim%":   (if $a > 0 { $"(($g.lim * 100 / $a) | math round)%" } else { "—" }),
+                "no-req": $g.noreq,
+                labels:   ($labelmap | get -o $name | default ""),
+            }
+        })
+
+    let pod_rows = if $pods {
+        let pods_res = (comp-kubectl-ssh $ws_root $infra_name "top pods --all-namespaces --no-headers")
+        if $pods_res.exit_code != 0 {
+            print --stderr $"kubectl top pods failed: ($pods_res.stderr | str trim)"
+            []
+        } else {
+            $pods_res.stdout | str replace --all "\r" "" | lines
+            | where { ($in | str trim | is-not-empty) }
+            | each {|ln|
+                let c = ($ln | str trim | split row --regex '\s+')
+                {
+                    namespace: ($c | get 0? | default ""),
+                    pod:       ($c | get 1? | default ""),
+                    cpu:       ($c | get 2? | default ""),
+                    mem:       ($c | get 3? | default ""),
+                    mem_bytes: (k8s-quantity-bytes ($c | get 3? | default "0")),
+                }
+            }
+            | sort-by mem_bytes --reverse
+            | first $limit
+            | reject mem_bytes
+        }
+    } else { [] }
+
+    let data = if $pods { { nodes: $node_rows, pods: $pod_rows } } else { $node_rows }
+    match ($out | str downcase) {
+        "json" => ($data | to json)
+        "yaml" | "yml" => ($data | to yaml)
+        "" | "table" => {
+            if $pods {
+                print "NODES"
+                print ($node_rows | table)
+                cluster-top-legend
+                print ""
+                print $"TOP ($limit) PODS BY MEMORY"
+                $pod_rows
+            } else {
+                print ($node_rows | table)
+                cluster-top-legend
+                null
+            }
+        }
+        _ => { cli-err $"unknown --out format: ($out) \(use json|yaml|table\)" }
+    }
+}
+
+# Interpretive legend for `cluster top` columns — printed under the node table so
+# the numbers can be read without external context. Suppressed in json/yaml output.
+def cluster-top-legend []: nothing -> nothing {
+    print ""
+    print "How to read it:"
+    print "  nominal   physical CPU/RAM of the Hetzner type (cax11=2c/4Gi, cax21=4c/8Gi)."
+    print "  cpu /mem  LIVE usage now (kubectl top). mem% is vs ALLOCATABLE (RAM minus"
+    print "            kernel+kube reservations), so it reads higher than vs nominal."
+    print "  req%      sum of pod memory REQUESTS vs allocatable = what the scheduler"
+    print "            reserved. Healthy < 80%; it is the real 'fullness' for placement."
+    print "  lim%      sum of pod memory LIMITS vs allocatable = overcommit. > 100% means"
+    print "            pods may collectively exceed RAM → OOM-kill risk under pressure."
+    print "  no-req    containers with NO memory request — invisible to the scheduler,"
+    print "            cause blind bin-packing and are first to be OOM-killed."
+    print "  labels    operator node labels (node_class, can_use_fip.*, custom) that drive"
+    print "            placement; system/cloud domains (kubernetes.io, hetzner.cloud) hidden."
+}
+
+# Reconcile declared operator node labels onto the live Kubernetes nodes. Source
+# of truth: servers.ncl node_labels (record) + node_class (list, projected as
+# node-class.<class>=true). Closes the drift where declared labels never reach the
+# nodes (no kubectl label runs at join time). Dry-run by default: prints the plan;
+# --apply runs `kubectl label node ... --overwrite` over the bounded SSH path.
+export def "main cluster label-nodes" [
+    --workspace (-w): string = ""
+    --infra (-i): string = ""
+    --apply (-a)
+]: nothing -> any {
+    let ws_root = (comp-ws-path $workspace)
+    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
+    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
+    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
+
+    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
+    let servers  = ($settings | get -o servers | default []
+        | where { ($in.not_use? | default false) == false })
+
+    # Desired operator labels per node: node_labels record + node_class projection.
+    let desired = ($servers | reduce --fold {} {|s, acc|
+        let base = ($s.node_labels? | default {})
+        let cls  = ($s.node_class? | default [] | reduce --fold {} {|c, m| $m | insert $"node-class.($c)" "true" })
+        $acc | insert ($s.hostname) ($base | merge $cls)
+    })
+
+    let access = (comp-kubectl-access $ws_root $infra_name)
+    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
+    let live   = ($k8s.nodes | reduce --fold {} {|n, acc|
+        $acc | insert ($n.metadata.name) ($n.metadata?.labels? | default {})
+    })
+
+    # Diff via transpose (label keys carry dots/slashes — unsafe for cell-path get).
+    let plan = ($desired | transpose host want | each {|e|
+        let have = (($live | get -o $e.host | default {}) | transpose key value)
+        $e.want | transpose key value | each {|w|
+            let m = ($have | where key == $w.key)
+            let cur = if ($m | is-empty) { null } else { ($m | first | get value) }
+            if $cur != $w.value {
+                { node: $e.host, label: $w.key, current: ($cur | default "—"), desired: $w.value }
+            } else { null }
+        } | compact
+    } | flatten)
+
+    if ($plan | is-empty) {
+        print "All declared node labels already applied — nothing to do."
+        return []
+    }
+    if not $apply {
+        print $"Planned label changes \(($plan | length)\) — dry-run, use --apply to apply:"
+        return $plan
+    }
+    $plan | each {|c|
+        let r = (comp-kubectl-ssh $ws_root $infra_name $"label node ($c.node) ($c.label)=($c.desired) --overwrite")
+        { node: $c.node, label: $c.label, desired: $c.desired, status: (if $r.exit_code == 0 { "applied" } else { ($r.stderr | str trim) }) }
+    }
+}
+
+# Placement simulator: project the per-node impact of (re)deploying a service.
+# Reads the candidate's declared resources_effective (memory request) + placement
+# from settings, the live node snapshot (allocatable, current pod requests, live
+# usage) via the bounded SSH path, then computes projected requests% and
+# working-set% per eligible node (eligibility = node-class label ∩ placement.node_class).
+# Answers "does the next deploy fit, and where?" — the dry-run the ops loop lacked.
+export def "main cluster plan" [
+    service: string
+    --workspace (-w): string = ""
+    --infra (-i): string = ""
+    --out (-o): string = ""
+    --req-threshold: int = 80        # % allocatable requests treated as "full"
+    --use-threshold: int = 90        # % allocatable working-set treated as risky
+]: nothing -> any {
+    let ws_root = (comp-ws-path $workspace)
+    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
+    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
+    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
+
+    # Candidate spec from declared NCL (settings.components.<key>).
+    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
+    let comps = ($settings | get -o components | default {})
+    let key = if (($comps | get -o $service) != null) { $service } else { ($service | str replace --all "-" "_") }
+    let comp = ($comps | get -o $key | default null)
+    if ($comp == null) { cli-err $"component '($service)' not found in settings.components" }
+    let re = ($comp | get -o resources_effective | default {})
+    if ($re | is-empty) { cli-err $"component '($service)' has no resources_effective — declare class/resources first \(step 2c\)" }
+    let cand_req = (k8s-quantity-bytes ($re | get -o memory | default {} | get -o request | default "0"))
+    let cand_lim = (k8s-quantity-bytes ($re | get -o memory | default {} | get -o limit | default "0"))
+    let want_nc  = ($comp | get -o placement | default {} | get -o node_class | default [] | each {|c| $c | into string})
+    let svc_class = ($comp | get -o class | default "")
+
+    let access = (comp-kubectl-access $ws_root $infra_name)
+    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
+    if ($k8s.nodes | is-empty) { cli-err $"could not fetch live nodes \(CP: ($access.host | default 'unresolved'))" }
+    let top    = (comp-kubectl-ssh $ws_root $infra_name "top nodes --no-headers")
+    let usage  = if $top.exit_code == 0 {
+        $top.stdout | str replace --all "\r" "" | lines | where { ($in | str trim | is-not-empty) }
+        | reduce --fold {} {|ln, acc|
+            let c = ($ln | str trim | split row --regex '\s+')
+            $acc | insert ($c | get 0? | default "") (k8s-quantity-bytes ($c | get 3? | default "0"))
+        }
+    } else { {} }
+
+    let live_pods = ($k8s.pods | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) })
+    let rows = ($k8s.nodes | each {|n|
+        let name  = ($n.metadata.name)
+        let alloc = (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
+        let labels = ($n.metadata?.labels? | default {} | transpose k v
+            | where {|r| $r.k | str starts-with "node-class." } | each {|r| $r.k | str replace "node-class." "" })
+        let eligible = if ($want_nc | is-empty) { true } else { ($want_nc | any {|c| $c in $labels }) }
+        let cur_req = ($live_pods | where { ($in.spec.nodeName) == $name }
+            | reduce --fold 0 {|p, s| $s + ($p.spec?.containers? | default [] | reduce --fold 0 {|c, s2| $s2 + (k8s-quantity-bytes ($c.resources?.requests?.memory? | default "0")) }) })
+        let cur_use = ($usage | get -o $name | default 0)
+        let proj_req = ($cur_req + $cand_req)
+        let proj_use = ($cur_use + $cand_req)
+        let req_pct  = (if $alloc > 0 { ($proj_req * 100 / $alloc) | math round } else { 0 })
+        let use_pct  = (if $alloc > 0 { ($proj_use * 100 / $alloc) | math round } else { 0 })
+        {
+            node:      $name,
+            eligible:  $eligible,
+            "req now": (if $alloc > 0 { $"(($cur_req * 100 / $alloc) | math round)%" } else { "—" }),
+            "req→":    (if $alloc > 0 { $"($req_pct)%" } else { "—" }),
+            "ws→":     (if $alloc > 0 { $"($use_pct)%" } else { "—" }),
+            fits:      ($eligible and ($req_pct <= $req_threshold) and ($use_pct <= $use_threshold)),
+            _req_pct:  $req_pct,
+        }
+    })
+
+    let eligible_fit = ($rows | where {|r| $r.eligible and $r.fits } | sort-by _req_pct)
+    let best = ($eligible_fit | get 0? | get -o node | default "")
+    let mem_h = ($re | get -o memory | default {})
+    let summary = {
+        service: $key, class: $svc_class,
+        request: ($mem_h | get -o request | default "?"), limit: ($mem_h | get -o limit | default "?"),
+        node_class: ($want_nc | str join ","),
+        recommendation: (if ($best | is-not-empty) {
+            $"schedules on ($best) \(lowest projected requests among eligible nodes that fit\)"
+        } else {
+            "NO eligible node fits within thresholds — activate another worker (wrk-1) or resize a node"
+        }),
+    }
+
+    let out_rows = ($rows | reject _req_pct)
+    match ($out | str downcase) {
+        "json" => ({ summary: $summary, nodes: $out_rows } | to json)
+        "yaml" | "yml" => ({ summary: $summary, nodes: $out_rows } | to yaml)
+        _ => {
+            print $"plan: ($key)  class=($svc_class)  mem req/lim=($summary.request)/($summary.limit)  prefers node-class=($summary.node_class)"
+            print ($out_rows | table)
+            print $"→ ($summary.recommendation)"
+            null
+        }
+    }
+}
+
+# Format a byte count as a compact Gi/Mi string.
+def fmt-gib [b: int]: nothing -> string {
+    if $b >= 1073741824 { $"(($b / 1073741824 * 10 | math round) / 10)Gi" } else { $"(($b / 1048576) | math round)Mi" }
+}
+
+# Cluster capacity analysis: per-node working-set broken down by category
+# (movable = general app classes, pinned = datastore/storage_engine, infra =
+# kube-system/longhorn/cilium/daemonsets), plus an aggregate capacity check by
+# node-class with what-if projections. Answers "horizontal vs vertical": movable
+# load dominating storage nodes → add a general worker; pinned dominating → resize.
+export def "main cluster capacity" [
+    --workspace (-w): string = ""
+    --infra (-i): string = ""
+    --out (-o): string = ""
+    --add-worker                     # what-if: add one cax11 general worker (~3.4Gi alloc)
+    --resize: string = ""            # what-if: resize <node> cax11→cax21 (+4Gi alloc)
+]: nothing -> any {
+    let ws_root = (comp-ws-path $workspace)
+    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
+    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
+    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
+
+    let settings = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")
+    let comps = ($settings | get -o components | default {})
+    # pod-name-prefix → pinned?  (component name with _→- is the usual pod prefix)
+    let prefixes = ($comps | items {|k v|
+        { prefix: ($k | str replace --all "_" "-"), pinned: (($v.class? | default "") in ["datastore", "storage_engine"]) }
+    } | sort-by {|c| $c.prefix | str length } --reverse)
+    let infra_ns = ["kube-system", "longhorn-system", "cilium", "cert-manager"]
+
+    let access = (comp-kubectl-access $ws_root $infra_name)
+    let k8s    = (comp-ssh-k8s-data $ws_root $infra_name $access.host)
+    if ($k8s.nodes | is-empty) { cli-err $"could not fetch live nodes \(CP: ($access.host | default 'unresolved'))" }
+    let top = (comp-kubectl-ssh $ws_root $infra_name "top pods --all-namespaces --no-headers")
+    let usage = if $top.exit_code == 0 {
+        $top.stdout | str replace --all "\r" "" | lines | where { ($in | str trim | is-not-empty) }
+        | reduce --fold {} {|ln, acc|
+            let c = ($ln | str trim | split row --regex '\s+')
+            $acc | insert $"($c | get 0? | default '')/($c | get 1? | default '')" (k8s-quantity-bytes ($c | get 3? | default "0"))
+        }
+    } else { {} }
+
+    let live = ($k8s.pods | where { (($in.status?.phase? | default "") not-in ["Succeeded", "Failed"]) and (($in.spec?.nodeName? | default "") | is-not-empty) })
+    let per_node = ($k8s.nodes | each {|n|
+        let name  = ($n.metadata.name)
+        let alloc = (k8s-quantity-bytes ($n.status?.allocatable?.memory? | default "0"))
+        let labels = ($n.metadata?.labels? | default {} | transpose k v
+            | where {|r| $r.k | str starts-with "node-class." } | each {|r| $r.k | str replace "node-class." "" })
+        let pods = ($live | where { ($in.spec.nodeName) == $name })
+        let cats = ($pods | reduce --fold { movable: 0, pinned: 0, infra: 0 } {|p, acc|
+            let ns = ($p.metadata.namespace)
+            let pn = ($p.metadata.name)
+            let use = ($usage | get -o $"($ns)/($pn)" | default 0)
+            let cat = if ($ns in $infra_ns) { "infra" } else {
+                let m = ($prefixes | where {|c| $pn | str starts-with $c.prefix } | get 0? | default null)
+                if ($m == null) { "movable" } else if $m.pinned { "pinned" } else { "movable" }
+            }
+            $acc | upsert $cat (($acc | get $cat) + $use)
+        })
+        { node: $name, class: ($labels | str join ","), alloc: $alloc } | merge $cats
+    })
+
+    # Aggregate by node-class.
+    let general_alloc = ($per_node | where {|r| "general" in ($r.class | split row ",")} | reduce --fold 0 {|r, s| $s + $r.alloc})
+    let storage_alloc = ($per_node | where {|r| "storage" in ($r.class | split row ",")} | reduce --fold 0 {|r, s| $s + $r.alloc})
+    let movable_total = ($per_node | reduce --fold 0 {|r, s| $s + $r.movable})
+    let pinned_total  = ($per_node | reduce --fold 0 {|r, s| $s + $r.pinned})
+
+    let cax11_alloc = 3650722201   # ~3.4Gi typical allocatable on a fresh cax11
+    let resize_node = ($per_node | where node == $resize | get 0? | default null)
+    let resize_is_general = (if ($resize_node != null) { "general" in ($resize_node.class | split row ",") } else { false })
+    let resize_is_storage = (if ($resize_node != null) { "storage" in ($resize_node.class | split row ",") } else { false })
+    let gen_alloc_wi = ($general_alloc
+        + (if $add_worker { $cax11_alloc } else { 0 })
+        + (if $resize_is_general { 4294967296 } else { 0 }))
+    let stg_alloc_wi = ($storage_alloc + (if $resize_is_storage { 4294967296 } else { 0 }))
+
+    let gen_util_now = (if $general_alloc > 0 { ($movable_total * 100 / $general_alloc | math round) } else { 0 })
+    let gen_util_wi  = (if $gen_alloc_wi  > 0 { ($movable_total * 100 / $gen_alloc_wi  | math round) } else { 0 })
+
+    let scenario = if $add_worker { "+1 general worker (cax11)" } else if ($resize | is-not-empty) { $"resize ($resize) → cax21" } else { "current" }
+    let mv_s = (fmt-gib $movable_total)
+    let pn_s = (fmt-gib $pinned_total)
+    let recommendation = if ($movable_total > $pinned_total) {
+        $"movable load ($mv_s) dominates pinned ($pn_s) — HORIZONTAL: add a general worker to relieve the storage nodes; vertical resize would keep apps mislocated"
+    } else {
+        $"pinned/datastore load ($pn_s) dominates movable ($mv_s) — VERTICAL: resize the pressured storage node rather than adding a general worker"
+    }
+
+    let rows = ($per_node | each {|r|
+        {
+            node: $r.node, class: $r.class, alloc: (fmt-gib $r.alloc),
+            movable: (fmt-gib $r.movable), pinned: (fmt-gib $r.pinned), infra: (fmt-gib $r.infra),
+        }
+    })
+    let summary = {
+        movable_total: (fmt-gib $movable_total), pinned_total: (fmt-gib $pinned_total),
+        general_alloc: (fmt-gib $general_alloc), storage_alloc: (fmt-gib $storage_alloc),
+        general_util_now: $"($gen_util_now)%",
+        scenario: $scenario,
+        general_util_scenario: $"($gen_util_wi)%",
+        recommendation: $recommendation,
+    }
+
+    match ($out | str downcase) {
+        "json" => ({ nodes: $rows, summary: $summary } | to json)
+        "yaml" | "yml" => ({ nodes: $rows, summary: $summary } | to yaml)
+        _ => {
+            print "PER-NODE working-set by category (movable=general apps · pinned=datastore · infra=k8s/longhorn)"
+            print ($rows | table)
+            print ""
+            print $"movable total: ($summary.movable_total)   pinned total: ($summary.pinned_total)"
+            print $"general node-class alloc: ($summary.general_alloc)   storage: ($summary.storage_alloc)"
+            print $"general utilization \(movable/general-alloc\): now ($summary.general_util_now)  →  scenario [($scenario)] ($summary.general_util_scenario)"
+            print $"→ ($recommendation)"
+            null
+        }
+    }
+}
+
+# List all Kubernetes services via SSH to the cluster control plane.
+# Matches CLI output of `kubectl get svc --all-namespaces`.
+export def "main cluster services" [
+    --workspace (-w): string = ""
+    --infra (-i): string = ""
+    --namespace (-n): string = ""     # filter to single namespace (default: all)
+    --out (-o): string = ""           # json | yaml | table (default terminal view)
+]: nothing -> any {
+    let ws_root = (comp-ws-path $workspace)
+    if ($ws_root | is-empty) { cli-err "no workspace resolved" }
+    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
+    if ($infra_name | is-empty) { cli-err $"no infra found under ($ws_root)/infra" }
+
+    let ns_flag = if ($namespace | is-not-empty) { $"-n ($namespace)" } else { "--all-namespaces" }
+    let r = (comp-kubectl-ssh $ws_root $infra_name $"get svc ($ns_flag) -o json")
+    if $r.exit_code != 0 {
+        cli-err $"kubectl get svc failed: ($r.stderr | str trim)"
+    }
+
+    let items = (try { $r.stdout | from json | get -o items | default [] } catch { [] })
+    let rows = ($items | each {|svc|
+        let ns       = ($svc.metadata.namespace? | default "")
+        let name     = ($svc.metadata.name? | default "")
+        let svc_type = ($svc.spec.type? | default "ClusterIP")
+        let cip      = ($svc.spec.clusterIP? | default "")
+        let ext_ips  = ($svc.status.loadBalancer?.ingress? | default []
+            | each {|ing| $ing.ip? | default ($ing.hostname? | default "") }
+            | where { $in | is-not-empty } | str join ",")
+        let ports    = ($svc.spec.ports? | default []
+            | each {|p| $"($p.port?):($p.protocol? | default 'TCP')" } | str join " ")
+        let has_https = ($svc.spec.ports? | default [] | any {|p|
+            ($p.port? | default 0) == 443 or (($p.name? | default "") | str downcase | str contains "https")
+        })
+        { namespace: $ns, name: $name, type: $svc_type, cluster_ip: $cip, external_ips: $ext_ips, ports: $ports, https: $has_https }
+    })
+
+    match $out {
+        "json"  => { $rows | to json }
+        "yaml"  => { $rows | to yaml }
+        _       => { $rows }
+    }
+}
+
+# Oldest last_sync across .servers-state.json and .cluster-state.json (= most stale data shown).
+export def comp-cache-last-sync [ws_root: string, infra: string]: nothing -> string {
+    let srv_ts = do {
+        let p = ($ws_root | path join "infra" $infra ".servers-state.json")
+        if ($p | path exists) {
+            open $p | values | each { $in.last_sync? | default "" } | where { $in | is-not-empty }
+        } else { [] }
+    }
+    let cl_ts = do {
+        let p = (cluster-cache-path $ws_root $infra)
+        if ($p | path exists) {
+            let v = ((open $p).last_sync? | default "")
+            if ($v | is-not-empty) { [$v] } else { [] }
+        } else { [] }
+    }
+    let all = ($srv_ts | append $cl_ts) | where { $in | is-not-empty }
+    if ($all | is-empty) { "" } else { $all | sort | first }
+}
+
+# Single SSH session to CP: fetch pods + nodes as JSON in one connection.
+# Returns {pods: list<record>, nodes: list<record>} — parsed from kubectl -o json output.
+# Pods have metadata.namespace/name and status.phase/containerStatuses fields.
+# Nodes have metadata.name and status.conditions fields.
+export def comp-ssh-k8s-data [
+    ws_root:  string,
+    infra:    string,
+    server:   string,
+    --systemd-services: list<string> = [],  # service names to check via systemctl is-active
+]: nothing -> record {
+    let empty = { pods: [], nodes: [], systemd: {} }
+    if ($server | is-empty) {
+        print --stderr "[live] no server resolved — cannot fetch k8s data"
+        return $empty
+    }
+    let info = (comp-server-ssh-info $ws_root $infra $server)
+    if (is-debug-level 2) {
+        let key_part = if ($info.key | is-not-empty) { $" -i ($info.key)" } else { "" }
+        print --stderr $"[live] ssh($key_part) ($info.user)@($info.host) kubectl get pods+nodes -o json"
+    }
+    # PATH: SSH non-interactive sessions may not have /usr/local/bin (where k0s and kubeadm kubectl live).
+    # KUBECONFIG: kubeadm clusters set this in ~/.bashrc (not sourced in BatchMode SSH).
+    # Probe /root/.kube/config then /etc/kubernetes/admin.conf; k0s kubectl ignores KUBECONFIG entirely.
+    let systemd_snippet = if ($systemd_services | is-not-empty) {
+        let checks = ($systemd_services | each {|svc| $"printf '($svc):'; systemctl is-active ($svc) 2>/dev/null || true"} | str join "; ")
+        $"echo '===SYSTEMD==='; ($checks)"
+    } else { "" }
+    let cmd_base = "export PATH=/usr/local/bin:/usr/bin:/bin:$PATH; for _kc in /root/.kube/config /etc/kubernetes/admin.conf; do [ -f \"$_kc\" ] && { export KUBECONFIG=\"$_kc\"; break; }; done; kc() { if command -v kubectl >/dev/null 2>&1; then kubectl \"$@\"; else k0s kubectl \"$@\"; fi; }; echo '===PODS==='; kc get pods --all-namespaces -o json 2>/dev/null || echo '{\"items\":[]}'; echo '===NODES==='; kc get nodes -o json 2>/dev/null || echo '{\"items\":[]}'"
+    let cmd = if ($systemd_snippet | is-not-empty) { $"($cmd_base); ($systemd_snippet)" } else { $cmd_base }
+    let result = (comp-ssh-run $info.user $info.key $info.host $cmd)
+    if (is-debug-level 2) {
+        print --stderr $"[live] SSH exit=($result.exit_code) stdout_len=($result.stdout | str length)"
+    }
+    if $result.exit_code != 0 { return $empty }
+    let all_lines = ($result.stdout | str replace --all "\r" "" | lines)
+    let pods_idx     = ($all_lines | enumerate | where { ($in.item | str trim) == "===PODS===" }     | get 0? | get -o index | default (-1))
+    let nodes_idx    = ($all_lines | enumerate | where { ($in.item | str trim) == "===NODES===" }    | get 0? | get -o index | default (-1))
+    let systemd_idx  = ($all_lines | enumerate | where { ($in.item | str trim) == "===SYSTEMD===" }  | get 0? | get -o index | default (-1))
+    if (is-debug-level 3) {
+        print --stderr $"[live] pods_idx=($pods_idx) nodes_idx=($nodes_idx) systemd_idx=($systemd_idx) total_lines=($all_lines | length)"
+    }
+    let pods = if $pods_idx >= 0 {
+        let from = ($pods_idx + 1)
+        let raw  = if $nodes_idx > $pods_idx { $all_lines | skip $from | first ($nodes_idx - $from) } else { $all_lines | skip $from }
+        try { $raw | str join "\n" | from json | get -o items | default [] } catch {|e|
+            if (is-debug-level 1) {
+                print --stderr $"[live] pods JSON parse error: ($e.msg | str substring 0..120)"
+                print --stderr $"[live] pods first line: ($raw | get 0? | default '')"
+            }
+            []
+        }
+    } else { [] }
+    let nodes = if $nodes_idx >= 0 {
+        let from = ($nodes_idx + 1)
+        let raw  = if $systemd_idx > $nodes_idx { $all_lines | skip $from | first ($systemd_idx - $from) } else { $all_lines | skip $from }
+        try { $raw | str join "\n" | from json | get -o items | default [] } catch {|e|
+            if (is-debug-level 1) {
+                print --stderr $"[live] nodes JSON parse error: ($e.msg | str substring 0..120)"
+            }
+            []
+        }
+    } else { [] }
+    let systemd = if $systemd_idx >= 0 {
+        $all_lines | skip ($systemd_idx + 1)
+        | each {|ln|
+            let parts = ($ln | str trim | split column ":" name status)
+            if ($parts | length) >= 1 and ($parts | get 0 | get name | is-not-empty) {
+                { ($parts | get 0 | get name): ($parts | get 0 | get status | str trim) }
+            } else { {} }
+        }
+        | reduce -f {} {|it, acc| $acc | merge $it }
+    } else { {} }
+    if (is-debug-level 2) {
+        print --stderr $"[live] parsed pods=($pods | length) nodes=($nodes | length) systemd_keys=($systemd | columns | length)"
+    }
+    { pods: $pods, nodes: $nodes, systemd: $systemd }
+}
+
+# Public API: return pod records for a server (used by describe-pod and external callers).
+export def comp-ssh-pods [ws_root: string, infra: string, server: string]: nothing -> list {
+    (comp-ssh-k8s-data $ws_root $infra $server).pods
+}
+
+# Run kubectl describe pod -n <namespace> <pod> via SSH. Returns describe output as string.
+export def comp-ssh-describe-pod [ws_root: string, infra: string, server: string, namespace: string, pod: string]: nothing -> string {
+    let info = (comp-server-ssh-info $ws_root $infra $server)
+    let cmd  = $"if command -v kubectl >/dev/null 2>&1; then kubectl describe pod -n ($namespace) ($pod); else k0s kubectl describe pod -n ($namespace) ($pod); fi"
+    let result = (comp-ssh-run $info.user $info.key $info.host $cmd)
+    if $result.exit_code != 0 { $result.stderr | str trim } else { $result.stdout | str trim }
+}
+
+# Summarise pod records (from kubectl -o json) into {live, detail}.
+def comp-kubectl-summarise [pods: list]: nothing -> record {
+    if ($pods | is-empty) { return { live: "deployed", detail: "no pods" } }
+    # Exclude Succeeded pods (completed jobs/migrations) — they never return to Running
+    # and would distort the ratio. Failed phase and error-waiting pods are kept and
+    # classified as degraded.
+    let active  = ($pods | where { ($in | get -o status | default {} | get -o phase | default "") != "Succeeded" })
+    let total   = ($active | length)
+    let running = ($active | where { ($in | get -o status | default {} | get -o phase | default "") == "Running" } | length)
+    let pending = ($active | where { ($in | get -o status | default {} | get -o phase | default "") == "Pending"  } | length)
+
+    # Any pod in Failed phase OR any container waiting with an error reason → degraded.
+    let error_reasons = ["CrashLoopBackOff", "ImagePullBackOff", "ErrImagePull",
+                         "Error", "OOMKilled", "CreateContainerError",
+                         "CreateContainerConfigError", "RunContainerError",
+                         "InvalidImageName", "PostStartHookError"]
+    let failing = ($active | where { |p|
+        let phase = ($p | get -o status | default {} | get -o phase | default "")
+        if $phase == "Failed" { true } else {
+            let cs  = ($p | get -o status | default {} | get -o containerStatuses     | default [])
+            let ics = ($p | get -o status | default {} | get -o initContainerStatuses | default [])
+            ($cs | append $ics) | any { |c|
+                ($c | get -o state | default {} | get -o waiting | default {} | get -o reason | default "") in $error_reasons
+            }
+        }
+    } | length)
+
+    let live = if $failing > 0       { "degraded"
+    } else if $pending > 0           { "starting"
+    } else if $total == 0            { "deployed"
+    } else if $running == $total     { "healthy"
+    } else                           { "partial" }
+
+    let detail = if $failing > 0 {
+        $"($running)/($total) pods running  ($failing) failing"
+    } else {
+        $"($running)/($total) pods running"
+    }
+    { live: $live, detail: $detail }
+}
+
+# Filter pod records by namespace (exact match) and name substring.
+def comp-filter-pods [namespace: string, search: string, pods: list]: nothing -> list {
+    let ns_filtered = if ($namespace | is-not-empty) {
+        $pods | where { ($in | get -o metadata | default {} | get -o namespace | default "") == $namespace }
+    } else { $pods }
+    $ns_filtered | where { ($in | get -o metadata | default {} | get -o name | default "") | str contains $search }
+}
+
+# Summarise node records (from kubectl -o json) into {live, detail}.
+def comp-kubectl-nodes-summarise [nodes: list]: nothing -> record {
+    if ($nodes | is-empty) { return { live: "—", detail: "no node data" } }
+    let total = ($nodes | length)
+    let ready = ($nodes | where { |n|
+        $n | get -o status | default {} | get -o conditions | default []
+        | any { ($in | get -o type | default "") == "Ready" and ($in | get -o status | default "") == "True" }
+    } | length)
+    let live = if $ready == $total { "healthy"
+    } else if $ready == 0          { "degraded"
+    } else                         { "partial" }
+    { live: $live, detail: $"($ready)/($total) nodes Ready" }
+}
+
+# Live check for a component using pre-fetched k8s_data {pods, nodes}.
+# Resolves strategy from comp.live_check; falls back to namespace/pod_selector on the component.
+# 'systemd strategy always returns "—" in live (fast) mode — requires prvng c health.
+# Returns {live: string, detail: string}.
+export def comp-live-check [name: string, comp: record, k8s_data: record]: nothing -> record {
+    let lc       = ($comp | get -o live_check | default {})
+    let strategy = ($lc | get -o strategy | default "none" | into string)
+    match $strategy {
+        "k8s_pods" => {
+            if ($k8s_data.pods | is-empty) { return { live: "—", detail: "no pod data" } }
+            let ns  = do {
+                let explicit = ($lc | get -o namespace | default "")
+                if ($explicit | is-not-empty) { $explicit } else { $comp | get -o namespace | default "" }
+            }
+            let sel = do {
+                let explicit = ($lc | get -o selector | default "")
+                if ($explicit | is-not-empty) { $explicit } else {
+                    let ps = ($comp | get -o pod_selector | default "")
+                    if ($ps | is-not-empty) { $ps } else { $name | str replace --all "_" "-" }
+                }
+            }
+            let matching = (comp-filter-pods $ns $sel $k8s_data.pods)
+            if ($matching | is-empty) { return { live: "—", detail: $"no pods: ($sel)" } }
+            comp-kubectl-summarise $matching
+        }
+        "k8s_nodes" => {
+            if ($k8s_data.nodes | is-empty) { return { live: "—", detail: "no node data" } }
+            let sel = ($lc | get -o selector | default "")
+            let filtered = if ($sel | is-not-empty) {
+                $k8s_data.nodes | where { ($in | get -o metadata | default {} | get -o name | default "") | str contains $sel }
+            } else {
+                $k8s_data.nodes
+            }
+            comp-kubectl-nodes-summarise $filtered
+        }
+        "k8s_api" => {
+            # apiserver healthy if kubectl returned any node data
+            if ($k8s_data.nodes | is-not-empty) or ($k8s_data.pods | is-not-empty) {
+                { live: "healthy", detail: "apiserver ok" }
+            } else {
+                { live: "—", detail: "apiserver unreachable" }
+            }
+        }
+        "systemd" => {
+            let svc = ($lc | get -o service | default "")
+            if ($svc | is-empty) { return { live: "—", detail: "health: systemd" } }
+            let units = ($k8s_data | get -o systemd | default {})
+            if ($units | is-empty) { return { live: "—", detail: "health: systemd" } }
+            let active = ($units | get -o $svc | default "")
+            match $active {
+                "active"   => { live: "healthy", detail: $"systemd: ($svc) active" },
+                "inactive" => { live: "degraded", detail: $"systemd: ($svc) inactive" },
+                "failed"   => { live: "degraded", detail: $"systemd: ($svc) failed" },
+                _          => { live: "—", detail: $"systemd: ($svc) ($active)" },
+            }
+        }
+        "on_demand" => { live: "—", detail: "on-demand" }
+        _ => { live: "—", detail: "no live check" }
+    }
+}
+
+# Extract pod summary records for a component from pre-fetched pod JSON records.
+# Returns list of {namespace, pod, ready, status}.
+export def comp-pods-for [namespace: string, name: string, pods: list, pod_selector: string = ""]: nothing -> list {
+    let search = if ($pod_selector | is-not-empty) { $pod_selector } else { $name | str replace --all "_" "-" }
+    (comp-filter-pods $namespace $search $pods)
+    | each {|p|
+        let meta   = ($p | get -o metadata | default {})
+        let status = ($p | get -o status   | default {})
+        let cs     = ($status | get -o containerStatuses | default [])
+        let ready  = ($cs | where { $in.ready? | default false } | length)
+        {
+            namespace: ($meta | get -o namespace | default ""),
+            pod:       ($meta | get -o name      | default ""),
+            ready:     $"($ready)/($cs | length)",
+            status:    ($status | get -o phase   | default ""),
+        }
+    }
+}
+