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]

export def cli-err [msg: string] {
    if (is-debug-enabled) {
        error make {msg: $msg}
    } else {
        error make --unspanned {msg: $msg}
    }
}

# Resolve the provisioning root for --import-path resolution.
def comp-prov-root []: nothing -> string {
    $env.PROVISIONING? | default "/usr/local/provisioning"
}

# Resolve the deploy directory for a component.
# Checks taskserv/ then cluster/ subdirectories (cluster components use the latter).
# Returns "" if not found.
def comp-find-taskserv-dir [name: string, ...aliases: string]: nothing -> string {
    let prov = (comp-prov-root)
    let base_candidates = [
        $name
        ($name | str replace --all "_" "-")
        ($name | str replace --all "-" "_")
    ]
    let extra = ($aliases | each {|a| [$a ($a | str replace --all "_" "-") ($a | str replace --all "-" "_")] } | flatten)
    for cand in ($base_candidates | append $extra) {
        for subdir in ["taskserv" "cluster"] {
            let p = ($prov | path join "catalog" "components" $cand $subdir)
            if ($p | path exists) { return $p }
        }
    }
    ""
}

# Build a deployable tar.gz bundle for a component using the full run_taskserv pipeline.
# Handles: .j2 template rendering, `prepare` script execution, common.sh injection.
# Returns base64-encoded bundle content ready for the script_compressed payload field.
def comp-build-bundle [
    name: string
    ws_root: string
    infra: string
    target: string
    op: string
    settings_data: record
]: nothing -> string {
    let comp_name_field = ($settings_data | get -o components | default {} | get -o $name | default {} | get -o name | default "")
    let catalog_ref     = ($settings_data | get -o components | default {} | get -o $name | default {} | get -o catalog_ref | default "")
    let taskserv_path = (comp-find-taskserv-dir $name $comp_name_field $catalog_ref)
    if ($taskserv_path | is-empty) {
        cli-err $"No taskserv directory found for component '($name)'"
    }

    let comp_cfg = ($settings_data | get -o components | default {} | get -o $name | default {})

    let settings_file = ($ws_root | path join "infra" $infra "settings.ncl")
    let run_settings = {
        data:       $settings_data,
        src:        ($settings_file | path basename),
        src_path:   ($settings_file | path dirname),
        infra_path: ($ws_root | path join "infra" $infra),
        wk_path:    (^mktemp -d | str trim),
        providers:  ($settings_data | get -o providers | default []),
    }

    # Merge component settings into taskserv record so .j2 templates see taskserv.* vars
    let taskserv_rec = ({
        name:         $name,
        profile:      "default",
        cmd_task:     $op,
        install_mode: "library",
        depends_on:   [],
        on_error:     "Continue",
    } | merge $comp_cfg)

    # Build server record matching the shape that handlers.nu passes to run_taskserv:
    # full servers.ncl entry merged with ip_addresses.{pub,priv}.
    let server_rec = do {
        let servers_path = ($ws_root | path join "infra" $infra "servers.ncl")
        let raw = (ncl-eval-soft $servers_path (default-ncl-paths $ws_root) null)
        let srvr = if ($raw != null) {
            $raw | get -o servers | default [] | where {|s| ($s.hostname? | default "") == $target } | get 0?
        } else { null }
        if ($srvr | is-not-empty) {
            let _state_path = ($ws_root | path join "infra" $infra ".servers-state.json")
            let _srv_state = if ($_state_path | path exists) {
                $_state_path | open | get -o $target | default {}
            } else { {} }
            let pub_ip = do {
                let np = ($srvr | get -o networking.public_ip | default "")
                let fp = ($srvr | get -o floating_ip | default "")
                if ($np | is-not-empty) {
                    $np
                } else if ($fp | is-not-empty) {
                    let fip_addr = ($_srv_state | get -o floating_ip_address | default "")
                    if ($fip_addr | is-not-empty) { $fip_addr } else { $_srv_state | get -o public_ip | default $target }
                } else {
                    let state_ip = ($_srv_state | get -o public_ip | default "")
                    if ($state_ip | is-not-empty) { $state_ip } else { $target }
                }
            }
            let priv_ip = do {
                let np = ($srvr | get -o network_private_ip | default "")
                if ($np | is-not-empty) { $np } else { $srvr | get -o networking.private_ip | default "" }
            }
            $srvr | merge { ip_addresses: { pub: $pub_ip, priv: $priv_ip } }
        } else {
            { hostname: $target, ip_addresses: { pub: $target, priv: "" }, main_domain: "", domains_search: "" }
        }
    }

    # run_taskserv uses $env.NOW for timestamped temp dirs — ensure it's set.
    if ($env.NOW? | is-empty) {
        $env.NOW = (date now | format date "%Y_%m_%d_%H_%M_%S")
    }

    let wk_tmp = (^mktemp -d | str trim)

    let defs = {
        settings:              $run_settings,
        server:                $server_rec,
        taskserv:              $taskserv_rec,
        taskserv_install_mode: "library",
        taskserv_profile:      "default",
        pos:                   { server: "0", taskserv: 0 },
        ip:                    ($server_rec | get -o ip_addresses.pub | default $target),
        check:                 true,
    }

    let ok = (run_taskserv $defs $taskserv_path ($wk_tmp | path join $name))
    rm -rf $wk_tmp

    if not $ok {
        cli-err $"Bundle build failed for component ($name)"
    }

    let tar_path = $"/tmp/($name).tar.gz"
    if not ($tar_path | path exists) {
        # run_taskserv names the tar after taskserv.name (the NCL `name` field, e.g. "kubernetes")
        # which may differ from the settings key (e.g. "kubernetes_worker"). Rename if found.
        let inner_name = ($taskserv_rec | get -o name | default "")
        let alt_tar = $"/tmp/($inner_name).tar.gz"
        if ($inner_name | is-not-empty) and ($inner_name != $name) and ($alt_tar | path exists) {
            mv $alt_tar $tar_path
        } else {
            cli-err $"Bundle tar.gz not found after build: ($tar_path)"
        }
    }

    let b64 = (open --raw $tar_path | encode base64)
    rm -f $tar_path
    $b64
}

# Build a deployable tar.gz bundle for a cluster-mode component.
# Packages catalog/components/<name>/cluster/ + workspace component NCL config.
# The orchestrator SCPs this to the control-plane node and runs install-<name>.sh via SSH.
# Returns {b64: string, work_dir: string}. Caller is responsible for rm -rf work_dir
# unless debug=true, in which case work_dir is printed and preserved for inspection.
def comp-build-cluster-bundle [
    name: string
    ws_root: string
    infra: string
    target: string
    op: string
    settings_data: record
    debug: bool = false
    skip_preflight: bool = false
]: nothing -> record {
    let fs_name = ($name | str replace --all "-" "_")
    let prov = (comp-prov-root)

    # Multi-tenant components (e.g. odoo_jesusperez) share one extension dir (odoo/cluster/).
    # Resolution order: catalog_ref field → comp_cfg.name → settings key.
    let comp_cfg_raw = ($settings_data | get -o components | default {} | get -o $fs_name | default {})
    let catalog_ref_raw = ($comp_cfg_raw | get -o catalog_ref | default "" | into string)
    let ext_name_raw    = ($comp_cfg_raw | get -o name        | default "" | into string)
    let ext_fs_name = if ($catalog_ref_raw | is-not-empty) {
        $catalog_ref_raw | str replace --all "-" "_"
    } else if ($ext_name_raw | is-not-empty) {
        $ext_name_raw | str replace --all "-" "_"
    } else {
        $fs_name
    }

    let cluster_dir = ($prov | path join "catalog" "components" $ext_fs_name "cluster")
    if not ($cluster_dir | path exists) {
        cli-err $"No cluster/ directory found for component '($name)' at ($cluster_dir)"
    }

    let work_dir = (^mktemp -d | str trim)
    let bundle_dir = ($work_dir | path join $name)
    ^mkdir -p $bundle_dir

    # Copy cluster scripts and templates into bundle.
    ^cp -r $"($cluster_dir)/." $bundle_dir

    # Include catalog/lib/ shared scripts so component libs can source shared primitives.
    let catalog_lib_src = (($env.PROVISIONING_CATALOG_PATH? | default ($prov | path join "catalog")) | path join "lib")
    if ($catalog_lib_src | path exists) {
        ^cp -rp $catalog_lib_src ($bundle_dir | path join "lib")
    }

    # Create name aliases so the orchestrator (hyphens) and preflight (underscores) both
    # find install-<name>.sh and <name>-lib.sh regardless of which extension the scripts came from.
    if $ext_fs_name != $fs_name {
        # Multi-tenant / catalog_ref: extension scripts use ext_fs_name as base (catalog dir name).
        # Orchestrator runs install-<name>.sh (hyphens); run-init.sh sources <fs_name>-lib.sh (underscores).
        # Only two aliases needed — no hyphen-lib or underscore-install copies.
        for pair in [
            [$"install-($ext_fs_name).sh" $"install-($name).sh"]
            [$"($ext_fs_name)-lib.sh"     $"($fs_name)-lib.sh"]
        ] {
            let src = ($bundle_dir | path join ($pair | get 0))
            let dst = ($bundle_dir | path join ($pair | get 1))
            if ($src | path exists) and not ($dst | path exists) {
                ^cp $src $dst
            }
        }
    } else if $fs_name != $name {
        # Standard: fs_name uses underscores, name uses hyphens — create hyphen alias.
        for pair in [
            [$"install-($fs_name).sh" $"install-($name).sh"]
            [$"($fs_name)-lib.sh"     $"($name)-lib.sh"]
        ] {
            let src = ($bundle_dir | path join ($pair | get 0))
            let dst = ($bundle_dir | path join ($pair | get 1))
            if ($src | path exists) and not ($dst | path exists) {
                ^cp $src $dst
            }
        }
    }

    # Write resolved component config as JSON so the install script can read it.
    # Inject cluster kubeconfig into comp_cfg so templates access it via {{ name.kubeconfig }}.
    # Source: settings.kubernetes.kubeconfig (kubeadm) or settings.k0s.kubeconfig (k0s).
    let cluster_kubeconfig = (
        $settings_data | get -o kubernetes | default {} | get -o kubeconfig | default (
            $settings_data | get -o k0s | default {} | get -o kubeconfig | default ""
        )
    )
    let comp_cfg_raw = ($settings_data | get -o components | default {} | get -o $fs_name | default {})
    let comp_cfg = if ($cluster_kubeconfig | is-not-empty) and (($comp_cfg_raw | get -o kubeconfig | default "") | is-empty) {
        $comp_cfg_raw | insert kubeconfig $cluster_kubeconfig
    } else {
        $comp_cfg_raw
    }
    let settings_json = ($settings_data | to json)
    $settings_json | save --force ($bundle_dir | path join "settings.json")
    ($comp_cfg | to json) | save --force ($bundle_dir | path join "component.json")

    # Generate config.env from component NCL settings — install script sources this.
    # Maps component record fields → STALWART_* env vars (plus RELAY_* for relay block).
    let prefix = ($fs_name | str upcase)
    mut config_lines: list<string> = []

    # TODO(provider-middleware): resolve lb_ipam_ip via Hetzner provider middleware instead of direct hcloud calls.
    # When the provider query layer is wired, replace this with a provider-aware FIP lookup by namespace/role.
    # let lb_ipam_ip_resolved = (
    #     let v = ($comp_cfg | get -o lb_ipam_ip | default "" | into string)
    #     if ($v | is-not-empty) { $v } else {
    #         let hcloud_secret = ($ws_root | path join "infra" $infra "secrets" "hcloud.sops.yaml")
    #         let kage          = ($ws_root | path join "infra" $infra ".kage")
    #         let sops_c        = ($ws_root | path join "infra" $infra "sops.yaml")
    #         let comp_ns       = ($comp_cfg | get -o namespace | default $name | into string)
    #         if ($hcloud_secret | path exists) and ($kage | path exists) {
    #             let tok_r = do {
    #                 with-env { SOPS_AGE_KEY_FILE: $kage } {
    #                     ^sops --decrypt --config $sops_c --output-type dotenv $hcloud_secret
    #                 }
    #             } | complete
    #             if $tok_r.exit_code == 0 {
    #                 let token = ($tok_r.stdout | lines | where {|l| $l | str starts-with "token=" } | first | str replace "token=" "" | str trim)
    #                 let fips_r = do { with-env { HCLOUD_TOKEN: $token } { ^hcloud floating-ip list -o json } } | complete
    #                 if $fips_r.exit_code == 0 {
    #                     let desc_match = ($comp_ns | str downcase)
    #                     $fips_r.stdout | from json | where {|f| ($f | get -o description | default "" | str downcase) | str contains $desc_match } | get 0? | get -o ip | default ""
    #                 } else { "" }
    #             } else { "" }
    #         } else { "" }
    #     }
    # )
    let lb_ipam_ip_resolved = ($comp_cfg | get -o lb_ipam_ip | default "" | into string)

    # Shared SOPS paths — used here for zone-secret injection and below for TLS/credentials.
    let kage_path = ($ws_root | path join "infra" $infra ".kage")
    let sops_cfg  = ($ws_root | path join "infra" $infra "sops.yaml")

    # Decrypt zone DNS secret if comp_cfg carries a secret_ref (top-level or first cluster_issuer).
    # Injects cf_api_token into comp_cfg so vars.nu receives it via the JSON argument — no env var needed.
    let zone_secret_ref_top = ($comp_cfg | get -o secret_ref | default "" | into string)
    let zone_secret_ref_issuers = (
        $comp_cfg | get -o cluster_issuers | default []
        | where {|i| ($i | get -o secret_ref | default "") | is-not-empty }
        | get 0? | default {} | get -o secret_ref | default ""
    )
    let zone_secret_ref = if ($zone_secret_ref_top | is-not-empty) { $zone_secret_ref_top } else { $zone_secret_ref_issuers }
    let comp_cfg = if ($zone_secret_ref | is-not-empty) and ($kage_path | path exists) {
        let zone_sops = ($ws_root | path join "infra" $infra "secrets" $"($zone_secret_ref).sops.yaml")
        if ($zone_sops | path exists) {
            let dec = (do {
                with-env { SOPS_AGE_KEY_FILE: $kage_path } {
                    ^sops --decrypt --config $sops_cfg --output-type yaml $zone_sops
                }
            } | complete)
            if $dec.exit_code == 0 {
                let token = ($dec.stdout | from yaml | get -o token | default "")
                if ($token | is-not-empty) { $comp_cfg | upsert cf_api_token $token } else { $comp_cfg }
            } else { $comp_cfg }
        } else { $comp_cfg }
    } else { $comp_cfg }

    # Resolve gateway_scope → gateway_ip from cluster networking.gateways when set.
    # Allows components to declare 'private/'public instead of hardcoding IPs.
    # Path: settings.ncl exports { settings: { networking: { gateways: {...} } }, ... }
    let _cluster_gateways = (
        $settings_data
        | get -o settings | default {}
        | get -o networking | default {}
        | get -o gateways | default {}
    )
    let comp_cfg = if not ($comp_cfg | get -o gateway_scope | default "" | is-empty) {
        let _scope = ($comp_cfg | get gateway_scope | into string)
        let _resolved = ($_cluster_gateways | get -o $_scope | default "")
        if ($_resolved | is-not-empty) { $comp_cfg | upsert gateway_ip $_resolved } else { $comp_cfg }
    } else { $comp_cfg }

    # vars.nu: derived vars merged into the component record before tera rendering.
    let vars_nu = ($cluster_dir | path join "vars.nu")
    let derived = if ($vars_nu | path exists) {
        let r = do { ^nu $vars_nu ($comp_cfg | to json) } | complete
        if $r.exit_code != 0 {
            error make { msg: $"vars.nu failed for ($name):\n($r.stderr | str trim)" }
        }
        $r.stdout | str trim | from json
    } else { {} }

    # vars JSON for tera: component config nested under "taskserv" key → {{ taskserv.name }}.
    # vars.nu-derived fields (cf_secret_name, gateway_ip, tls_secret, …) go at the root level.
    # Multi-tenant: odoo_jesusperez and odoo_diegodelgado both resolve to the same "taskserv" key.
    let tera_ctx = ({taskserv: ($comp_cfg | upsert lb_ipam_ip $lb_ipam_ip_resolved)} | merge $derived)
    let vars_json_path = ($work_dir | path join "vars.json")
    ($tera_ctx | to json) | save --force $vars_json_path

    # Render env-<ext_name>.j2 → env-<ext_name> (shell env file sourced by install script).
    # Use direct run_from_template (not on_template_path) to avoid recursing into templates/,
    # which would cause double-render and duplicate content via save --append in run_from_template.
    let env_j2 = ($bundle_dir | path join $"env-($ext_fs_name).j2")
    if ($env_j2 | path exists) {
        let env_out = ($bundle_dir | path join $"env-($ext_fs_name)")
        rm -f $env_out
        run_from_template $env_j2 $vars_json_path $env_out --only_make
        rm -f $env_j2
    }
    # Multi-tenant: run-<op>.sh sources env-($fs_name); alias to the rendered env-($ext_fs_name).
    if $ext_fs_name != $fs_name {
        let env_src = ($bundle_dir | path join $"env-($ext_fs_name)")
        let env_dst = ($bundle_dir | path join $"env-($fs_name)")
        if ($env_src | path exists) and not ($env_dst | path exists) {
            ^cp $env_src $env_dst
        }
    }

    # Render templates/*.j2 → manifests/*.yaml (K8s manifests applied by install script).
    # Layer 1: extension templates (provisioning/catalog/components/<name>/cluster/templates/)
    # Layer 2: workspace overlay (infra/<infra>/components/<name>/templates/) — same name wins.
    let tmpl_dir  = ($bundle_dir | path join "templates")
    let out_dir   = ($bundle_dir | path join "manifests")
    ^mkdir -p $out_dir
    if ($tmpl_dir | path exists) {
        on_template_path $tmpl_dir $vars_json_path true true
        for f in (glob $"($tmpl_dir)/*") { ^mv $f $out_dir }
        ^rm -rf $tmpl_dir
    }
    let ws_tmpl_dir = ($ws_root | path join "infra" $infra "components" $fs_name "templates")
    if ($ws_tmpl_dir | path exists) {
        let ws_tmp = (^mktemp -d | str trim)
        ^cp -r $"($ws_tmpl_dir)/." $ws_tmp
        on_template_path $ws_tmp $vars_json_path true true
        for f in (glob $"($ws_tmp)/*") { ^mv -f $f $out_dir }
        ^rm -rf $ws_tmp
    }

    # Remove all .j2 source files — tera sources must not reach the remote server.
    for j2 in (glob $"($bundle_dir)/**/*.j2") { ^rm -f $j2 }

    # Generate manifest_plan.json: load component plan, merge workspace override.
    let plan_ncl = ($cluster_dir | path join "manifest_plan.ncl")
    let base_plan = if ($plan_ncl | path exists) {
        let r = (ncl-eval-soft $plan_ncl (default-ncl-paths $ws_root) null)
        if $r != null { $r | get -o manifest_plan | default {} } else { {} }
    } else { {} }
    let ws_plan = ($comp_cfg | get -o manifest_plan | default {})
    let merged_plan = if ($ws_plan | is-empty) {
        $base_plan
    } else {
        $base_plan | merge $ws_plan
    }
    ($merged_plan | to json) | save --force ($bundle_dir | path join "manifest_plan.json")

    # Compute op-level hooks from concerns.manifest_hooks (presets) and manifest_plan.hooks (component).
    let effective_hooks = (comp-plan-merge-hooks $merged_plan $comp_cfg)

    # Generate run-<op>.sh scripts from the plan with injected op-level hooks.
    for op in ["init" "update" "delete" "restart"] {
        let steps    = ($merged_plan     | get -o $op | default [])
        let op_hooks = ($effective_hooks | get -o $op | default {})
        let pre      = ($op_hooks | get -o pre  | default [])
        let post     = ($op_hooks | get -o post | default [])
        if ($steps | is-not-empty) or ($pre | is-not-empty) or ($post | is-not-empty) {
            let script = (comp-plan-op-to-bash $fs_name $op $steps $pre $post)
            let run_path = ($bundle_dir | path join $"run-($op).sh")
            $script | save --force $run_path
            ^chmod +x $run_path
        }
    }

    # kage_path and sops_cfg are defined above, before vars.nu invocation.

    # Decrypt TLS secret if present and write tls.crt / tls.key into bundle.
    # The install script creates the K8s TLS secret directly from these files.
    let tls_secret = ($ws_root | path join "infra" $infra "secrets" $"($fs_name)-tls.sops.yaml")
    if ($tls_secret | path exists) and ($kage_path | path exists) {
        let tls_r = do {
            with-env { SOPS_AGE_KEY_FILE: $kage_path } {
                ^sops --decrypt --config $sops_cfg --output-type yaml $tls_secret
            }
        } | complete
        if $tls_r.exit_code == 0 {
            let tls_data = ($tls_r.stdout | from yaml)
            let cert = ($tls_data | get -o certificate | default "")
            let key  = ($tls_data | get -o key         | default "")
            if ($cert | is-not-empty) { $cert | save --force ($bundle_dir | path join "tls.crt") }
            if ($key  | is-not-empty) { $key  | save --force ($bundle_dir | path join "tls.key") }
        }
    }

    # Decrypt component SOPS secret if present and write _credentials.env into bundle.
    # Prefixed with _ — not auto-sourced. Methods that need secrets source it explicitly.
    let secret_path = ($ws_root | path join "infra" $infra "secrets" $"($fs_name).sops.yaml")
    if ($secret_path | path exists) and ($kage_path | path exists) {
        let dec = do {
            with-env { SOPS_AGE_KEY_FILE: $kage_path } {
                ^sops --decrypt --config $sops_cfg --output-type dotenv $secret_path
            }
        } | complete
        if $dec.exit_code == 0 {
            # Single-quote every value so bash `source` doesn't expand $-chars (e.g. bcrypt hashes).
            let quoted = ($dec.stdout | lines | each {|line|
                if ($line | is-empty) or ($line | str starts-with "#") or (not ($line | str contains "=")) {
                    $line
                } else {
                    let kv = ($line | split row "=" | collect)
                    $"(($kv | first))='(($kv | skip 1 | str join "=" | str replace --all "'" "'\\''" ))'"
                }
            } | str join "\n")
            $"($quoted)\n" | save --force ($bundle_dir | path join "_credentials.env")
        } else {
            cli-err $"Failed to decrypt ($secret_path): ($dec.stderr | str trim)"
        }
    }

    # Inject CF tokens into _credentials.env: two sources merged in priority order.
    # 1. Zone SOPS files — derived.cf_secret_name (from vars.nu) → secrets/<name>.sops.yaml → token key.
    #    Produces CF_TOKEN_<ZONE_NAME_UPPER> expected by _method_create-cf-secret.
    # 2. Caller environment CF_TOKEN_* — injected by just recipes (lower priority, overridden by #1).
    let creds_path = ($bundle_dir | path join "_credentials.env")
    mut cf_lines: list<string> = []
    # Captured for the provisioning-side DNS reconcile (returned in .dns below):
    # the cluster nodes never call Cloudflare, so the token stays on provisioning.
    mut dns_token = ""

    let cf_secret_name = ($derived | get -o cf_secret_name | default "" | into string)
    if ($cf_secret_name | is-not-empty) and ($kage_path | path exists) {
        let zone_sops = ($ws_root | path join "infra" $infra "secrets" $"($cf_secret_name).sops.yaml")
        if ($zone_sops | path exists) {
            let dec = do {
                with-env { SOPS_AGE_KEY_FILE: $kage_path } {
                    ^sops --decrypt --config $sops_cfg --output-type dotenv $zone_sops
                }
            } | complete
            if $dec.exit_code == 0 {
                let raw_token = ($dec.stdout | lines
                    | where {|l| ($l | str contains "=") and not ($l | str starts-with "#") }
                    | first | split row "=" | skip 1 | str join "=" | str trim)
                if ($raw_token | is-not-empty) {
                    let env_key = $"CF_TOKEN_($cf_secret_name | str upcase | str replace --all '-' '_')"
                    $cf_lines = ($cf_lines | append $"($env_key)='($raw_token)'")
                    $dns_token = $raw_token
                }
            }
        }
    }

    let cf_env_tokens = ($env | transpose key val | where { |r| $r.key | str starts-with "CF_TOKEN_" })
    for r in $cf_env_tokens {
        if not ($cf_lines | any {|l| $l | str starts-with $"($r.key)="}) {
            $cf_lines = ($cf_lines | append $"($r.key)='($r.val | str replace --all "'" "'\\''")'")
        }
    }

    if ($cf_lines | is-not-empty) {
        let existing = if ($creds_path | path exists) { open --raw $creds_path } else { "" }
        $"($existing)\n($cf_lines | str join "\n")\n" | save --force $creds_path
    }

    # Cabling-based credential injection: if infra/<ws>/integrations/<name>.ncl exists,
    # assemble credentials from its bindings and append to _credentials.env.
    # Takes precedence over what's already written — last writer wins for duplicate keys.
    let cabling_candidates = [
        ($ws_root | path join "infra" $infra "integrations" $"($fs_name).ncl")
        ($ws_root | path join "infra" $infra "integrations" $"($name).ncl")
    ]
    let cabling_path = ($cabling_candidates | where {|p| $p | path exists } | get 0? | default "")
    if ($cabling_path | is-not-empty) {
        let ncl_import = (comp-prov-root)
        let cabling_env = (build-cabling-credentials-env $cabling_path
            --workspace-root $ws_root
            --ncl-import $ncl_import
            --kage-path $kage_path)
        if ($cabling_env | is-not-empty) {
            let creds_path = ($bundle_dir | path join "_credentials.env")
            let existing   = if ($creds_path | path exists) { open --raw $creds_path } else { "" }
            $"($existing)\n# cabling: ($cabling_path | path basename)\n($cabling_env)\n" | save --force $creds_path
        }
    }

    # Preflight var check: collect every _require_env call across all .sh files in the bundle,
    # resolve env-<ext_name> + credentials.env into a flat map, and fail early on any empty var.
    let pf_env_file   = ($bundle_dir | path join $"env-($ext_fs_name)")
    let pf_creds_file = ($bundle_dir | path join "_credentials.env")
    let pf_env_vals = (
        [$pf_env_file $pf_creds_file]
        | where {|f| $f | path exists }
        | each {|f|
            open --raw $f
            | lines
            | where {|l| not ($l | str starts-with "#") and ($l | str contains "=") }
            | each {|l|
                let parts = ($l | split row "=" -n 2)
                {key: ($parts | get 0? | default "" | str trim), val: ($parts | get 1? | default "" | str trim)}
            }
        }
        | flatten
        | where {|r| $r.key | is-not-empty }
        | reduce --fold {} {|r acc| $acc | upsert $r.key $r.val }
    )
    # Only scan run-*.sh (generated op scripts) + *-lib.sh (method implementations).
    # Exclude install-*.sh — it has per-operation _require_env guards (backup, restore)
    # that are not install-time requirements and would produce false positives.
    let pf_sh_candidates = (
        [
            (try { glob $"($bundle_dir)/run-*.sh"   } catch { [] })
            (try { glob $"($bundle_dir)/*-lib.sh"   } catch { [] })
        ] | flatten
    )
    let pf_required = (
        $pf_sh_candidates
        | each {|sh_path|
            open --raw $sh_path
            | lines
            | where {|l| $l | str contains "_require_env " }
            | each {|l|
                $l | str trim | split row " " | get 1? | default "" | str trim
                   | str replace --regex '["\x27]' ""
            }
        }
        | flatten
        | where {|v| $v | is-not-empty }
        | uniq
    )
    let pf_missing = ($pf_required | where {|v| ($pf_env_vals | get -o $v | default "" | str trim) | is-empty })
    if (not $skip_preflight) and ($pf_missing | is-not-empty) {
        for v in $pf_missing { print $"[preflight] ❌  required var not set: ($v)" }
        error make {msg: $"Preflight failed — missing env vars: ($pf_missing | str join ', ')"}
    }

    # DNS admission gate (ADR-051 constraint `preflight-aborts-on-inadmissible`).
    # A component that declares a `domain` cannot participate in the op unless its
    # DNS state is admissible: the zone token must decrypt AND an A-record target
    # (gateway_ip) must resolve. Otherwise the provisioning-side reconcile would
    # silently skip (empty token) or publish nothing (empty target_ip) and the
    # domain would never resolve while the deploy reported green — the exact
    # ontoref failure. Aborting here creates no state (preflight semantics), unlike
    # the postdeploy WARN which fires after the mutation already happened.
    # Additive: components without a `domain` skip the gate entirely.
    let pf_domain = ($comp_cfg | get -o domain | default "" | into string)
    if (not $skip_preflight) and ($pf_domain | is-not-empty) {
        let pf_gw_ip = (
            $derived | get -o gateway_ip
            | default ($comp_cfg | get -o gateway_ip | default "")
            | into string
        )
        mut pf_dns_errs: list<string> = []
        if ($dns_token | is-empty) {
            $pf_dns_errs = ($pf_dns_errs | append $"zone token undecryptable/absent for '($pf_domain)' \(expected secrets/($cf_secret_name).sops.yaml\)")
        }
        if ($pf_gw_ip | is-empty) {
            $pf_dns_errs = ($pf_dns_errs | append $"no A-record target for '($pf_domain)' — gateway_ip empty \(check gateway_scope / hcloud auth\)")
        }
        if ($pf_dns_errs | is-not-empty) {
            for e in $pf_dns_errs { print $"[preflight] ❌  DNS admission: ($e)" }
            error make {msg: $"Preflight failed — DNS state inadmissible for ($name): ($pf_dns_errs | str join '; ')"}
        }
    }

    # Verify naming convention: orchestrator runs install-<name>.sh (hyphens); run-init.sh sources <fs_name>-lib.sh (underscores).
    for pf_required_script in [$"install-($name).sh" $"($fs_name)-lib.sh"] {
        let pf_script_path = ($bundle_dir | path join $pf_required_script)
        if not ($pf_script_path | path exists) {
            print $"[preflight] ❌  required script missing: ($pf_required_script) component=($name)"
            error make {msg: $"Preflight failed — ($pf_required_script) not found in bundle"}
        }
    }

    # Structural contract check — applies to cluster components with a manifest plan.
    # Catches authoring errors before packaging: credential filename, method coverage.
    let pf_plan_json = ($bundle_dir | path join "manifest_plan.json")
    let pf_lib_sh    = ($bundle_dir | path join $"($fs_name)-lib.sh")
    if ($pf_plan_json | path exists) and ($pf_lib_sh | path exists) {
        let pf_install_sh = ($bundle_dir | path join $"install-($name).sh")
        if ($pf_install_sh | path exists) {
            let pf_bad_cred_source = (
                open --raw $pf_install_sh
                | lines
                | where {|l| not ($l | str starts-with "#") }
                | where {|l| $l | str contains "credentials.env" }
                | where {|l| not ($l | str contains "_credentials.env") }
            )
            if ($pf_bad_cred_source | is-not-empty) {
                print $"[preflight] ❌  install-($name).sh sources 'credentials.env' — must be '_credentials.env'"
                error make {msg: $"Preflight failed — install-($name).sh uses legacy credential filename (see adr-007)"}
            }
        }
        let pf_plan     = (open $pf_plan_json)
        let pf_builtins = ["apply" "rollout-restart" "delete" "recreate"]
        let pf_custom_actions = (
            ["init" "update" "delete" "restart"]
            | each {|op|
                $pf_plan | get -o $op | default []
                | each {|step|
                    let main   = ($step | get -o action | default "" | str replace --all "'" "")
                    let posts  = ($step | get -o post | default [] | each {|h| $h | get -o action | default "" | str replace --all "'" "" })
                    let pres   = ($step | get -o pre  | default [] | each {|h| $h | get -o action | default "" | str replace --all "'" "" })
                    [$main] ++ $posts ++ $pres
                }
                | flatten
                | where {|action| ($action | is-not-empty) and not ($action in $pf_builtins) }
            }
            | flatten
            | uniq
        )
        let pf_lib_content      = (open --raw $pf_lib_sh)
        let pf_missing_methods  = (
            $pf_custom_actions
            | where {|action| not ($pf_lib_content | str contains $"_method_($action)") }
        )
        if ($pf_missing_methods | is-not-empty) {
            for m in $pf_missing_methods {
                print $"[preflight] ❌  _method_($m) missing in ($fs_name)-lib.sh"
            }
            error make {msg: $"Preflight failed — missing method implementations: ($pf_missing_methods | str join ', ')"}
        }

        # Secret existence + encrypted_regex coverage.
        # If lib.sh calls _require_env for any var, the SOPS file must exist and its
        # encrypted_regex must cover every required variable name.
        let pf_sops_required_vars = (
            $pf_lib_content
            | lines
            | where {|l| $l | str contains "_require_env "}
            | where {|l| not ($l | str trim | str starts-with "#")}
            | each {|l|
                $l | str trim | split row " " | get 1? | default "" | str trim
                   | str replace --regex '["\x27]' ""
            }
            | where {|v| $v | is-not-empty}
            | uniq
        )
        if (not $skip_preflight) and ($pf_sops_required_vars | is-not-empty) {
            let pf_sops_path = ($ws_root | path join "infra" $infra "secrets" $"($fs_name).sops.yaml")
            if not ($pf_sops_path | path exists) {
                print $"[preflight] ❌  ($fs_name).sops.yaml not found — required by _require_env: ($pf_sops_required_vars | str join ', ')"
                error make {msg: $"Preflight failed — secrets file missing: infra/($infra)/secrets/($fs_name).sops.yaml"}
            }
            let pf_sops_doc         = (open $pf_sops_path)
            let pf_encrypted_regex  = ($pf_sops_doc | get -o sops.encrypted_regex | default "")
            if ($pf_encrypted_regex | str length) > 0 {
                # Only check vars that are actually declared in the SOPS file.
                # _require_env is also used for config vars sourced from env-<name>,
                # which are not secrets and must not be flagged here.
                let pf_sops_keys = ($pf_sops_doc | reject sops | columns)
                let pf_uncovered_vars = (
                    $pf_sops_required_vars
                    | where {|v| $v in $pf_sops_keys}
                    | where {|v| not ($pf_encrypted_regex | str contains $v)}
                )
                if ($pf_uncovered_vars | is-not-empty) {
                    for v in $pf_uncovered_vars {
                        print $"[preflight] ⚠️  ($v) present in ($fs_name).sops.yaml as plaintext — not covered by encrypted_regex"
                    }
                }
            }
        }
    }

    # Pack into tar.gz and base64-encode.
    # Files must be at tar root — orchestrator does: tar xzf <name>.tar.gz -C /tmp/<name> && bash install-<name>.sh
    # Use gtar (GNU tar) if available to strip macOS xattr headers that confuse Linux tar.
    let tar_path = ($work_dir | path join $"($name).tar.gz")
    let tar_bin = if (^which gtar | complete).exit_code == 0 { "gtar" } else { "tar" }
    let tar_extra = if $tar_bin == "gtar" { ["--no-xattrs"] } else { [] }
    do { ^$tar_bin ...$tar_extra -czf $tar_path -C $bundle_dir "." } | complete | ignore
    let b64 = (open --raw $tar_path | encode base64)
    if $debug {
        print $"[preflight] ── Bundle preserved ──────────────────────────────"
        print $"[preflight] dir:       ($bundle_dir)"
        print $"[preflight] manifests: ($bundle_dir)/manifests/"
        print $"[preflight] env:       ($bundle_dir)/env-($name)"
        print $"[preflight] tls:       ($bundle_dir)/tls.crt  ($bundle_dir)/tls.key"
        print $"[preflight] Inspect:   ls ($bundle_dir)"
        print $"[preflight]            cat ($bundle_dir)/out/($name)/configmap.yaml"
        print $"[preflight] ─────────────────────────────────────────────────"
        print $"[preflight] NOTE: _credentials.env contains secrets — inspect locally only"
    } else {
        ^rm -rf $work_dir
    }
    # Multi-site components (e.g. rev_proxy): one A record per declared site, all
    # pointing at the component FIP. Each site's zone token is decrypted from its
    # own SOPS file (dns-<zone>.sops.yaml), mirroring the single-zone block above.
    let dns_sites = do {
        let sites = ($comp_cfg | get -o sites | default [])
        let fip = ($derived | get -o fip_ip | default "")
        if ($sites | is-empty) or ($fip | is-empty) { [] } else {
            $sites | each {|s|
                let domain = ($s | get -o domain | default "")
                let zone   = ($s | get -o dns_zone | default "")
                if ($domain | is-empty) or ($zone | is-empty) { null } else {
                    let secret = $"dns-($zone | str replace --all '.' '-')"
                    let zsops = ($ws_root | path join "infra" $infra "secrets" $"($secret).sops.yaml")
                    let tok = if ($zsops | path exists) and ($kage_path | path exists) {
                        let dec = (do -i { with-env { SOPS_AGE_KEY_FILE: $kage_path } {
                            ^sops --decrypt --config $sops_cfg --output-type dotenv $zsops
                        } } | complete)
                        if $dec.exit_code == 0 {
                            $dec.stdout | lines
                            | where {|l| ($l | str contains "=") and not ($l | str starts-with "#") }
                            | first | default "" | split row "=" | skip 1 | str join "=" | str trim
                        } else { "" }
                    } else { "" }
                    { token: $tok, domain: $domain, target_ip: $fip, zone: $zone }
                }
            } | where {|r| $r != null }
        }
    }

    # DNS reconcile inputs — the provisioning-side deploy step upserts the A record(s)
    # after a successful install (see the $wait/completed path). Empty domain or
    # target_ip → the reconcile skips. Nodes carry no DNS logic.
    let dns_reconcile = {
        token:     $dns_token,
        domain:    ($comp_cfg | get -o domain | default ""),
        target_ip: ($derived | get -o gateway_ip | default ($comp_cfg | get -o gateway_ip | default "")),
        zone:      ($comp_cfg | get -o dns_zone | default ""),
        sites:     $dns_sites,
    }
    {b64: $b64, work_dir: $work_dir, dns: $dns_reconcile}
}

# Generate bash lines for a single hook or method-action step.
def comp-plan-hook-lines [hook: record]: nothing -> list<string> {
    let action = ($hook | get -o action | default "noop" | str replace "'" "")
    let params  = ($hook | get -o params | default {})
    let delay   = ($hook | get -o delay  | default 0)
    mut lines: list<string> = []
    for kv in ($params | items {|k v| {k: $k, v: $v}}) {
        $lines = ($lines | append $"export PLAN_PARAM_($kv.k | str upcase)=\"($kv.v)\"")
    }
    $lines = ($lines | append $"_method_($action)")
    for k in ($params | columns) {
        $lines = ($lines | append $"unset PLAN_PARAM_($k | str upcase)")
    }
    if $delay > 0 { $lines = ($lines | append $"sleep ($delay)") }
    $lines
}

# Generate a bash run-<op>.sh script from a list of manifest plan steps.
# Merge op-level hooks from concerns.manifest_hooks (preset) and manifest_plan.hooks (component).
# Concerns hooks run first in pre (infrastructure before app logic).
# Plan hooks run first in post (app cleanup before infrastructure cleanup).
# Known limitation: ws_plan merge is shallow — workspace hooks replace component hooks entirely.
def comp-plan-merge-hooks [plan: record, comp_cfg: record]: nothing -> record {
    let ch = ($comp_cfg | get -o concerns | default {} | get -o manifest_hooks | default {})
    let ph = ($plan     | get -o hooks    | default {})

    let merge_op = {|op|
        let c = ($ch | get -o $op | default {})
        let p = ($ph | get -o $op | default {})
        {
            pre:  (($c | get -o pre  | default []) | append ($p | get -o pre  | default [])),
            post: (($p | get -o post | default []) | append ($c | get -o post | default [])),
        }
    }
    {
        init:    (do $merge_op "init"),
        update:  (do $merge_op "update"),
        delete:  (do $merge_op "delete"),
        restart: (do $merge_op "restart"),
    }
}

def comp-plan-op-to-bash [name: string, op: string, steps: list, hooks_pre: list = [], hooks_post: list = []]: nothing -> string {
    mut lines: list<string> = [
        "#!/bin/bash",
        "set -euo pipefail",
        $"# Generated by provisioning bundle builder for ($name):($op) — do not edit",
        "SCRIPT_DIR=\"$(cd \"$(dirname \"${BASH_SOURCE[0]}\")\" && pwd)\"",
        "# shellcheck source=/dev/null",
        "set -a",
        $"[ -f \"$SCRIPT_DIR/env-($name)\" ]    && source \"$SCRIPT_DIR/env-($name)\"",
        "# _credentials.env is NOT auto-sourced — methods source it explicitly when needed",
        "set +a",
        $"source \"$SCRIPT_DIR/($name)-lib.sh\"",
        "",
    ]
    if ($hooks_pre | is-not-empty) or ($hooks_post | is-not-empty) {
        $lines = ($lines | append
            "[ -f \"$SCRIPT_DIR/lib/tls-hook-methods.sh\" ] && source \"$SCRIPT_DIR/lib/tls-hook-methods.sh\"")
        $lines = ($lines | append "")
    }
    for step in ($hooks_pre ++ $steps ++ $hooks_post) {
        let action = ($step | get -o action | default "apply" | str replace "'" "")
        let file    = ($step | get -o file   | default "")
        let skip    = ($step | get -o skip_if_exists | default false)
        let delay   = ($step | get -o delay  | default 0)
        let params  = ($step | get -o params | default {})
        let pre     = ($step | get -o pre    | default [])
        let post    = ($step | get -o post   | default [])

        for hook in $pre  { $lines = ($lines | append (comp-plan-hook-lines $hook)) }

        let label = if ($file | is-not-empty) { $"($action) ($file)" } else { $action }
        $lines = ($lines | append $"echo \"  [($op)] ($label)\"")

        match $action {
            "apply" => {
                if ($file | is-not-empty) {
                    if $skip {
                        $lines = ($lines | append $"_plan_apply_skip \"($file)\"")
                    } else {
                        $lines = ($lines | append $"_plan_apply \"($file)\"")
                    }
                }
            }
            "rollout-restart" => {
                if ($file | is-not-empty) {
                    $lines = ($lines | append $"_plan_rollout_restart \"($file)\"")
                }
            }
            "delete" => {
                if ($file | is-not-empty) {
                    $lines = ($lines | append $"_plan_delete \"($file)\"")
                }
            }
            "recreate" => {
                if ($file | is-not-empty) {
                    $lines = ($lines | append $"_plan_recreate \"($file)\"")
                }
            }
            _ => {
                for kv in ($params | items {|k v| {k: $k, v: $v}}) {
                    $lines = ($lines | append $"export PLAN_PARAM_($kv.k | str upcase)=\"($kv.v)\"")
                }
                $lines = ($lines | append $"_method_($action)")
                for k in ($params | columns) {
                    $lines = ($lines | append $"unset PLAN_PARAM_($k | str upcase)")
                }
            }
        }

        if $delay > 0 { $lines = ($lines | append $"sleep ($delay)") }
        for hook in $post { $lines = ($lines | append (comp-plan-hook-lines $hook)) }
        $lines = ($lines | append "")
    }
    $lines | str join "\n"
}

# Export a Nickel file as parsed JSON.
# Always calls nickel directly (no plugin cache) — deploy/ops paths require fresh config.
# Set PROVISIONING_USE_CACHE=true to opt into the plugin cache (read-only queries only).
export def comp-ncl-export [ws_root: string, rel_path: string]: nothing -> record {
    let full_path = ($ws_root | path join $rel_path)
    let use_cache = ($env.PROVISIONING_USE_CACHE? | default "false") == "true"
    if $use_cache {
        ncl-eval $full_path (default-ncl-paths $ws_root)
    } else {
        let import_paths = (default-ncl-paths $ws_root)
        let import_args = ($import_paths | each {|p| ["--import-path" $p] } | flatten)
        let r = (do { ^nickel export --format json $full_path ...$import_args } | complete)
        if $r.exit_code != 0 {
            cli-err $"NCL evaluation failed for ($full_path): ($r.stderr | str trim)"
        }
        $r.stdout | from json
    }
}

# 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
# in settings.cluster_access (ssh_host + remote_kubectl). When unset, fall back to
# the first server present in the state file and `k0s kubectl`.
export def comp-kubectl-access [ws_root: string, infra: string]: nothing -> record {
    let raw = (comp-ncl-export $ws_root $"infra/($infra)/settings.ncl")
    let s   = ($raw | get -o settings | default {})
    let ca  = ($s | get -o cluster_access | default {})
    # Operator-delegated workspaces (adr-002) own no cluster — they deploy onto a
    # target cluster. With no local cluster_access, resolve the target workspace's
    # access so kubectl reads hit the operator cluster's control-plane, not a CP
    # this workspace does not have. Guarded: a failed cross-workspace resolve
    # falls through to the local default rather than aborting the caller.
    let target_ws = ($s | get -o deployment_target | default {}
        | get -o target_cluster | default {} | get -o workspace | default "")
    if ($ca | is-empty) and ($target_ws | is-not-empty) {
        let delegated = (try {
            let tgt_root = (comp-ws-path $target_ws)
            if ($tgt_root | is-not-empty) and ($tgt_root != $ws_root) and ($tgt_root | path exists) {
                comp-kubectl-access $tgt_root (comp-ws-infra $tgt_root)
            } else { null }
        } catch { null })
        if ($delegated != null) { return $delegated }
    }
    let host = ($ca | get -o ssh_host | default "")
    let kc   = ($ca | get -o remote_kubectl | default "k0s kubectl")
    let resolved_host = if ($host | is-not-empty) { $host } else {
        let state_root = if (($ws_root | path join "infra" $infra ".provisioning-state.ncl") | path exists) {
            $ws_root | path join "infra" $infra
        } else { $ws_root }
        comp-state-server $state_root
    }
    # ws_root/infra identify the workspace that OWNS the resolved host, so callers
    # resolve SSH connection details (ip/user/key) from the right servers.ncl even
    # when access was delegated from an operator-delegated workspace.
    { host: $resolved_host, kubectl: $kc, ws_root: $ws_root, infra: $infra }
}

# Run a remote kubectl command via SSH using the infra's declared cluster_access.
# Hard-bounded so it can never hang: ConnectTimeout caps the connect phase,
# ServerAlive kills a half-dead session, and a remote `timeout` caps the kubectl
# call itself. Returns the {exit_code, stdout, stderr} record from `complete`.
export def comp-kubectl-ssh [
    ws_root: string,
    infra:   string,
    kargs:   string,          # kubectl arguments, e.g. "get pods -A -o json"
    --cmd-timeout: int = 20,  # remote wall-clock cap on the kubectl invocation
]: nothing -> record {
    let access = (comp-kubectl-access $ws_root $infra)
    if ($access.host | is-empty) {
        return { exit_code: 1, stdout: "", stderr: "cluster_access.ssh_host unset and no server in state file" }
    }
    # Resolve SSH details from the workspace that owns the host (the operator
    # cluster when access was delegated), not necessarily the calling workspace.
    let info   = (comp-server-ssh-info $access.ws_root $access.infra $access.host)
    let remote = $"export PATH=/usr/local/bin:/usr/bin:/bin:$PATH; timeout ($cmd_timeout) ($access.kubectl) ($kargs)"
    comp-ssh-run $info.user $info.key $info.host $remote
}

# 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")
    if not ($dag_path | path exists) { return [] }
    let full_path = ($ws_root | path join $"infra/($infra)/dag.ncl")
    let dag = (ncl-eval-soft $full_path (default-ncl-paths $ws_root) null)
    if ($dag == null) { return [] }
    $dag | get -o formulas | default []
}

# Extract name from a DAG node — handles both taskserv.name and component.name shapes.
def comp-node-name [n: record]: nothing -> string {
    let ts_name = ($n | get -o taskserv | default null | get -o name | default "")
    if ($ts_name | is-not-empty) { return $ts_name }
    $n | get -o component | default null | get -o name | default ""
}

# Locate the directory that holds .provisioning-state.ncl for a workspace+infra pair.
# The dag-executor writes state to src_path (= infra dir) when called via comp-build-bundle,
# but some workspaces have state at the workspace root. Probe both.
def comp-state-root [ws_root: string, infra: string]: nothing -> string {
    let infra_path = ($ws_root | path join "infra" $infra)
    if (($infra_path | path join ".provisioning-state.ncl") | path exists) {
        $infra_path
    } else {
        $ws_root
    }
}

# Query DAG execution state for a component across all formulas.
# workspace_path: directory containing .provisioning-state.ncl — use comp-state-root to resolve.
# Normalizes underscore→hyphen: settings use vol_prepare, DAG/state use vol-prepare.
# Returns list<{formula, server, state, started_at, ended_at, blocker}>.
# Entries for template-formula servers that were never initialized are excluded so that
# worker template formulas (server never deployed) do not dilute the aggregate state
# of components that share a name (e.g., os, k0s) with the template.
def comp-dag-state-for [workspace_path: string, formulas: list, comp_name: string]: nothing -> list {
    # Canonical form is always kebab — DAG formula taskserv.name convention.
    # State is normalized to kebab by component_workflow and state-normalize.
    let normalized = ($comp_name | str replace --all "_" "-")
    $formulas | each {|f|
        $f.nodes | where {|n| (comp-node-name $n) == $normalized } | each {|node|
            let ns = (state-node-get $workspace_path $f.server $normalized)
            # A node is "initialized" if dag-executor has ever touched it:
            # either it has log entries, or actor.identity is non-empty.
            # Pure defaults (log=[], identity="") mean the server template was never deployed.
            let initialized = (
                ($ns.log? | default [] | is-not-empty) or
                ($ns.actor?.identity? | default "" | is-not-empty)
            )
            {
                _init:      $initialized,
                formula:    $f.id,
                server:     $f.server,
                state:      ($ns.state? | default "pending"),
                started_at: ($ns.started_at? | default ""),
                ended_at:   ($ns.ended_at? | default ""),
                blocker:    ($ns.blocker? | default ""),
                log:        ($ns.log? | default []),
            }
        }
    } | flatten
    | where { |e| $e._init }
    | each  { |e| $e | reject _init }
}

# Lookup component state directly from the state file (no DAG formula required).
# Used as fallback for cluster-mode components not registered in any DAG formula.
# Returns the same list shape as comp-dag-state-for.
def comp-direct-state-for [workspace_path: string, comp_name: string]: nothing -> list {
    # Try kebab form first (canonical after state-normalize), fall back to
    # the original underscore form — state files written by older orchestrator
    # versions use kebab; newer ones may use underscore. Both must resolve.
    let kebab = ($comp_name | str replace --all "_" "-")
    let snake = ($comp_name | str replace --all "-" "_")
    let st = (state-read $workspace_path)
    $st.servers
    | transpose hostname srv
    | each {|row|
        let taskservs = ($row.srv.taskservs? | default {})
        # The same taskserv may be recorded under both kebab and snake forms
        # (orphan duplicate from a node-name convention change). Pick the
        # genuinely-active entry — a stale pending orphan must not shadow the
        # completed one — by preferring non-pending state, then most-recent activity.
        let cands = ([$kebab $comp_name $snake] | uniq
            | each {|k| $taskservs | get -o $k }
            | where {|v| ($v != null) and ($v | is-not-empty) })
        let ts = if ($cands | is-empty) { {} } else if ($cands | length) == 1 {
            $cands | first
        } else {
            ($cands | each {|c|
                let advanced = (if (($c.state? | default "pending" | into string) != "pending") { "1" } else { "0" })
                let tstamp = (if (($c.ended_at? | default "") != "") { ($c.ended_at? | default "") } else { ($c.started_at? | default "") })
                { node: $c, key: $"($advanced)|($tstamp)" }
            } | sort-by key | last | get node)
        }
        if ($ts | is-empty) { [] } else {
            let is_init = (
                ($ts.log? | default [] | is-not-empty) or
                ($ts.actor?.identity? | default "" | is-not-empty)
            )
            if $is_init {
                [{
                    formula:    "direct",
                    server:     $row.hostname,
                    state:      ($ts.state? | default "pending"),
                    started_at: ($ts.started_at? | default ""),
                    ended_at:   ($ts.ended_at? | default ""),
                    blocker:    ($ts.blocker? | default ""),
                    log:        ($ts.log? | default []),
                }]
            } else { [] }
        }
    } | flatten
}

# Resolve the first server hostname present in the state file for a workspace.
# Used by --live to find which host to SSH into when the DAG server is unknown.
export def comp-state-server [workspace_path: string]: nothing -> string {
    let st = (state-read $workspace_path)
    $st.servers
    | transpose hostname srv
    | where {|row| ($row.srv.taskservs? | default {} | is-not-empty) }
    | get 0?
    | get -o hostname
    | default ""
}

# ─── SSH + k0s kubectl live check infrastructure ─────────────────────────────

# Get SSH connection info for a server (user, key path, host) from servers.ncl.
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 }
    }
    let raw = (ncl-eval-soft $servers_path (default-ncl-paths $ws_root) null)
    if ($raw == null) {
        return { user: "root", key: "", host: $server }
    }
    let srv = ($raw | get -o servers | default [] | where {|s| ($s.hostname? | default "") == $server } | get 0?)
    let user = if ($srv | is-not-empty) { ($srv | get -o installer_user | default "root") } else { "root" }
    let key_raw = if ($srv | is-not-empty) { ($srv | get -o ssh_key_path | default "") } else { "" }
    let key = if ($key_raw | is-not-empty) { $key_raw | path expand } else { "" }
    let state_path = ($ws_root | path join "infra" $infra ".servers-state.json")
    let srv_state = if ($state_path | path exists) {
        $state_path | open | get -o $server | default {}
    } else { {} }
    let host = if ($srv | is-not-empty) {
        let np = ($srv | get -o networking.public_ip | default "")
        if ($np | is-not-empty) {
            $np
        } else {
            # floating_ip is a Hetzner API assignment name, not an SSH target.
            # Always use public_ip from state for SSH — the server's primary reachable address.
            let state_ip = ($srv_state | get -o public_ip | default "")
            if ($state_ip | is-not-empty) { $state_ip } else { $server }
        }
    } else { $server }
    { user: $user, key: $key, host: $host }
}

# Run a command on a remote host via SSH (with or without explicit identity file).
# 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`.
export def comp-ssh-run [user: string, key: string, host: string, cmd: string]: nothing -> record {
    let opts = [
        -o ConnectTimeout=10
        -o ServerAliveInterval=5
        -o ServerAliveCountMax=3
        -o BatchMode=yes
    ]
    if ($key | is-not-empty) {
        do { ^ssh -i $key ...$opts $"($user)@($host)" $cmd } | complete
    } else {
        do { ^ssh ...$opts $"($user)@($host)" $cmd } | complete
    }
}

# 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
}

# Auto-detect infra name from workspace root (exported for use by component.nu dispatcher).
export def comp-ws-infra [ws_root: string]: nothing -> string {
    ls ($ws_root | path join "infra")
    | where type == dir
    | get name
    | each {|p| $p | path basename }
    | where {|n| ($ws_root | path join "infra" $n "settings.ncl" | path exists) }
    | get 0?
    | default ""
}

# Summarise a list of dag state entries into a single string (for list column).
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" }
    if ($states | any { $in == "failed"    }) { return "failed" }
    if ($states | any { $in == "blocked"   }) { return "blocked" }
    if ($states | any { $in == "unknown"   }) { return "unknown" }
    if ($states | all { $in == "completed" }) { return "completed" }
    if ($states | any { $in == "completed" }) { return "partial" }
    "pending"
}

# Resolve workspace name: explicit --workspace flag or active workspace.
# A path counts as a workspace ONLY if it has real workspace content — not just
# any directory that happens to be called "infra/". Otherwise $HOME/infra/
# (or similar incidental dirs) hijack the walk-up and shadow the active workspace.
def is-workspace-root [path: string]: nothing -> bool {
    if (($path | path join ".ontology") | path exists) { return true }
    if (($path | path join "config" "provisioning.ncl") | path exists) { return true }
    let infra_dir = ($path | path join "infra")
    if not ($infra_dir | path exists) { return false }
    # Require at least one infra/*/settings.ncl to treat this as a workspace root.
    (ls $infra_dir | where type == dir | get name
        | any { |p| ($p | path join "settings.ncl" | path exists) })
}

def find-workspace-root-up [path: string]: nothing -> string {
    if ($path | is-empty) or $path == "/" { return "" }
    if (is-workspace-root $path) { return $path }
    let parent = ($path | path dirname)
    if $parent == $path { return "" }
    find-workspace-root-up $parent
}

# Walk up from $path looking for a sibling `workspaces/<name>/` that qualifies as
# a workspace root. Returns the full path to the workspace, or "" if not found.
def find-named-workspace-up [path: string, name: string]: nothing -> string {
    if ($path | is-empty) or $path == "/" { return "" }
    let candidate = ($path | path join "workspaces" $name)
    if ($candidate | path exists) and (is-workspace-root $candidate) {
        return $candidate
    }
    let parent = ($path | path dirname)
    if $parent == $path { return "" }
    find-named-workspace-up $parent $name
}

# Derive the project root from $env.PROVISIONING (typically
# `<project-root>/provisioning`). Returns "" if PROVISIONING is unset.
def project-root-from-prov []: nothing -> string {
    let prov = ($env.PROVISIONING? | default "")
    if ($prov | is-empty) { return "" }
    $prov | path dirname
}

# Returns the workspace ROOT PATH (not name) — works for both registered and unregistered workspaces.
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.
        let as_path = ($workspace | path expand)
        if ($as_path | path exists) and (is-workspace-root $as_path) {
            return $as_path
        }
        # 1b. Registered in the user config.
        let registered = (
            list-workspaces
            | where { |ws| $ws.name == $workspace }
            | get -o 0
        )
        if ($registered | is-not-empty) {
            return $registered.path
        }
        # 1c. Walk up from CWD looking for a sibling `workspaces/<name>/`.
        let from_cwd = (find-named-workspace-up $env.PWD $workspace)
        if ($from_cwd | is-not-empty) { return $from_cwd }
        # 1d. Derive from $env.PROVISIONING — the canonical project-root anchor.
        let project_root = (project-root-from-prov)
        if ($project_root | is-not-empty) {
            let from_prov = ($project_root | path join "workspaces" $workspace)
            if ($from_prov | path exists) and (is-workspace-root $from_prov) {
                return $from_prov
            }
        }
        cli-err $"Workspace '($workspace)' not found in registry, CWD ancestors, or under ($project_root)/workspaces/"
    }
    # 2. Detect from CWD: registered workspace whose path is prefix of CWD
    let pwd = $env.PWD
    let from_registry = (
        list-workspaces
        | where { |ws| ($ws.path | is-not-empty) and ($pwd | str starts-with $ws.path) }
        | sort-by { |ws| ($ws.path | str length) }
        | last  # longest match wins
    )
    if not ($from_registry | is-empty) {
        return $from_registry.path
    }
    # 3. Unregistered: walk up CWD until workspace markers found
    let ws_root = (find-workspace-root-up $pwd)
    if ($ws_root | is-not-empty) {
        return $ws_root
    }
    # 4. Fall back to active workspace
    let details = (get-active-workspace-details)
    if ($details == null) or ($details.path? | default "" | is-empty) {
        cli-err "No active workspace — pass --workspace or activate one first"
    }
    $details.path
}

# Validate cluster capabilities against real infrastructure state.
#
# Exports infra/{infra}/capabilities.ncl from the workspace and compares declared
# capabilities (storage_classes, ingress_class) against live kubectl output.
# Returns a table of check / expected / actual / status rows.
#
# Usage:
#   provisioning validate capabilities --workspace libre-daoshi --infra wuji
export def "main validate capabilities" [
    --workspace (-w): string = ""  # Workspace name or path (default: auto-detect)
    --infra (-i): string = ""  # Infra name (default: auto-detect)
]: nothing -> table<check: string, expected: string, actual: string, status: string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    if ($ws_root | is-empty) {
        cli-err "Could not resolve workspace root"
    }
    let infra = if ($infra | is-not-empty) { $infra } else {
        ls ($ws_root | path join "infra") | where type == dir | get name
        | each { |p| $p | path basename }
        | where { |n| ($ws_root | path join "infra" $n "settings.ncl" | path exists) }
        | get 0? | default ""
    }

    let caps_path = ($ws_root | path join "infra" $infra "capabilities.ncl")
    if not ($caps_path | path exists) {
        cli-err $"capabilities.ncl not found at ($caps_path)"
    }

    let caps = (comp-ncl-export $ws_root ($"infra/($infra)/capabilities.ncl"))
    mut rows: list<record<check: string, expected: string, actual: string, status: string>> = []

    # Check storage classes
    let declared_sc = ($caps | get -o provides | default {} | get -o storage_classes | default [] | each { $in | into string })
    if ($declared_sc | is-not-empty) {
        let sc_result = (comp-kubectl-ssh $ws_root $infra "get sc --no-headers -o custom-columns=NAME:.metadata.name")
        let actual_sc = if $sc_result.exit_code == 0 {
            $sc_result.stdout | lines | where { $in | is-not-empty }
        } else {
            []
        }
        for sc in $declared_sc {
            let found = ($actual_sc | any { $in == $sc })
            $rows = ($rows | append {
                check: "storage_class",
                expected: $sc,
                actual: (if $found { $sc } else { "<not found>" }),
                status: (if $found { "ok" } else { "MISSING" }),
            })
        }
    }

    # Check ingress class
    let declared_ic = ($caps | get -o provides | default {} | get -o ingress_class | default "")
    if ($declared_ic | is-not-empty) {
        let ic_result = (comp-kubectl-ssh $ws_root $infra "get ingressclass --no-headers -o custom-columns=NAME:.metadata.name")
        let actual_ic = if $ic_result.exit_code == 0 {
            $ic_result.stdout | lines | where { $in | is-not-empty }
        } else {
            []
        }
        let found = ($actual_ic | any { $in == $declared_ic })
        $rows = ($rows | append {
            check: "ingress_class",
            expected: $declared_ic,
            actual: (if $found { $declared_ic } else { "<not found>" }),
            status: (if $found { "ok" } else { "MISSING" }),
        })
    }

    $rows
}

# Validate component configuration against workspace capabilities and server inventory.
#
# Exports infra/{infra}/settings.ncl and checks each component:
#   - taskserv mode: verifies the target server exists in the servers map.
#   - cluster mode:  verifies the storage_class (if declared) is in capabilities.storage_classes.
# Returns a table of component / check / status / detail rows.
#
# Usage:
#   provisioning validate components --workspace libre-daoshi --infra wuji
export def "main validate components" [
    --workspace (-w): string = ""  # Workspace name or path (default: auto-detect)
    --infra (-i): string = ""  # Infra name (default: auto-detect)
]: nothing -> table<component: string, check: string, status: string, detail: string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    if ($ws_root | is-empty) {
        cli-err "Could not resolve workspace root"
    }
    let infra = if ($infra | is-not-empty) { $infra } else {
        ls ($ws_root | path join "infra") | where type == dir | get name
        | each { |p| $p | path basename }
        | where { |n| ($ws_root | path join "infra" $n "settings.ncl" | path exists) }
        | get 0? | default ""
    }

    let settings = (comp-ncl-export $ws_root ($"infra/($infra)/settings.ncl"))
    let components = ($settings | get -o components | default {})

    # Load capabilities for storage_class cross-check (best-effort: skip if absent).
    let caps_path = ($ws_root | path join "infra" $infra "capabilities.ncl")
    let caps_sc: list<string> = if ($caps_path | path exists) {
        let c = (comp-ncl-export $ws_root ($"infra/($infra)/capabilities.ncl"))
        $c | get -o provides | default {} | get -o storage_classes | default [] | each { $in | into string }
    } else {
        []
    }

    # Load servers for taskserv target validation (best-effort).
    let servers_path = ($ws_root | path join "infra" $infra "servers.ncl")
    let server_names: list<string> = if ($servers_path | path exists) {
        ncl-eval-soft $servers_path (default-ncl-paths $ws_root) {} | get -o servers | default {} | columns
    } else {
        []
    }

    mut rows: list<record<component: string, check: string, status: string, detail: string>> = []

    let comp_names = ($components | columns)
    for comp_name in $comp_names {
        let comp = ($components | get $comp_name)
        let mode = ($comp | get -o mode | default "cluster")

        if $mode == "taskserv" {
            let target = ($comp | get -o target | default "")
            if ($target | is-empty) {
                $rows = ($rows | append { component: $comp_name, check: "target_server", status: "WARN", detail: "mode=taskserv but no target specified" })
            } else if ($server_names | is-empty) {
                $rows = ($rows | append { component: $comp_name, check: "target_server", status: "SKIP", detail: $"servers.ncl not available — cannot verify '($target)'" })
            } else {
                let found = ($server_names | any { $in == $target })
                $rows = ($rows | append {
                    component: $comp_name,
                    check: "target_server",
                    status: (if $found { "ok" } else { "MISSING" }),
                    detail: (if $found { $"target '($target)' exists" } else { $"target '($target)' not found in servers" }),
                })
            }
        } else if $mode == "cluster" {
            let sc = ($comp | get -o storage_class | default "")
            if ($sc | is-not-empty) {
                if ($caps_sc | is-empty) {
                    $rows = ($rows | append { component: $comp_name, check: "storage_class", status: "SKIP", detail: "capabilities.ncl not available" })
                } else {
                    let found = ($caps_sc | any { $in == $sc })
                    $rows = ($rows | append {
                        component: $comp_name,
                        check: "storage_class",
                        status: (if $found { "ok" } else { "MISSING" }),
                        detail: (if $found { $"storage_class '($sc)' available" } else { $"storage_class '($sc)' not in capabilities" }),
                    })
                }
            }
        }

        # Always emit a baseline row even when no sub-checks apply.
        if ($rows | where component == $comp_name | is-empty) {
            $rows = ($rows | append { component: $comp_name, check: "declared", status: "ok", detail: $"mode=($mode)" })
        }
    }

    $rows
}

# List all components declared in the workspace infra settings.
#
# Cluster components: one row per component (dag-summary state).
# Taskserv components: one row per (component × server) from the state file.
#
# Usage:
#   provisioning component list --workspace libre-daoshi
#   provisioning component list --server libre-wuji-wrk-0
export def "main component list" [
    --workspace (-w): string = ""  # Workspace name or path (default: auto-detect from CWD)
    --infra (-i): string = ""      # Infra name (default: auto-detect from workspace)
    --mode (-m): string = ""       # Filter by deploy mode: cluster | taskserv (default: all)
    --server (-s): string = ""     # Filter taskserv rows to a specific server hostname
]: nothing -> table<name: string, mode: string, server: string, namespace: string, version: string, state: string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    if ($ws_root | is-empty) {
        cli-err "Could not resolve workspace root"
    }
    let infra = if ($infra | is-not-empty) { $infra } else {
        # Auto-detect: first infra dir that has settings.ncl
        ls ($ws_root | path join "infra")
        | where type == dir
        | get name
        | each { |p| $p | path basename }
        | where { |n| ($ws_root | path join "infra" $n "settings.ncl" | path exists) }
        | get 0?
        | default ""
    }
    if ($infra | is-empty) {
        cli-err $"No infra with settings.ncl found in ($ws_root)/infra/"
    }

    let settings = (comp-ncl-export $ws_root ($"infra/($infra)/settings.ncl"))
    let components = ($settings | get -o components | default {})
    let dag_formulas = (comp-load-dag $ws_root $infra)

    let all_names = ($components | columns)
    let filtered_names = if ($mode | is-not-empty) {
        $all_names | where {|n|
            ($components | get $n | get -o mode | default "cluster") == $mode
        }
    } else {
        $all_names
    }

    let state_root = (comp-state-root $ws_root $infra)
    $filtered_names | each { |comp_name|
        let comp = ($components | get $comp_name)
        let comp_mode = ($comp | get -o mode | default "cluster")
        # Service-class contract surface (step 2): class, effective resources
        # (cpu·mem req/lim) and placement (node_class + mode). "—" when undeclared.
        let svc_class = ($comp | get -o class | default "")
        let re = ($comp | get -o resources_effective | default {})
        let resources = if ($re | is-empty) { "—" } else {
            let c = ($re | get -o cpu | default {})
            let m = ($re | get -o memory | default {})
            $"($c | get -o request | default '?')/($c | get -o limit | default '?') · ($m | get -o request | default '?')/($m | get -o limit | default '?')"
        }
        let pl = ($comp | get -o placement | default {})
        let placement = if ($pl | is-empty) { "—" } else {
            let ncs = ($pl | get -o node_class | default [] | str join ",")
            if ($ncs | is-empty) { "—" } else { $"($ncs) \(($pl | get -o mode | default 'prefer')\)" }
        }
        if $comp_mode == "taskserv" {
            # Expand: one row per server. If the declaration has an explicit target,
            # restrict to that server only — prevents node-join state (same taskserv
            # key, different server) from polluting the component's row.
            let declared_target = ($comp | get -o target | default "")
            let per_server_all  = (comp-direct-state-for $state_root $comp_name)
            let per_server = if ($declared_target | is-not-empty) {
                $per_server_all | where server == $declared_target
            } else {
                $per_server_all
            }
            if ($per_server | is-empty) {
                [{
                    name:         $comp_name,
                    mode:         "taskserv",
                    server:       $declared_target,
                    namespace:    "",
                    version:      ($comp | get -o version      | default ""),
                    class:        $svc_class,
                    resources:    $resources,
                    placement:    $placement,
                    pod_selector: ($comp | get -o pod_selector | default ""),
                    live_check:   ($comp | get -o live_check   | default {}),
                    state:        "pending",
                }]
            } else {
                $per_server | each {|entry|
                    {
                        name:         $comp_name,
                        mode:         "taskserv",
                        server:       $entry.server,
                        namespace:    "",
                        version:      ($comp | get -o version      | default ""),
                        class:        $svc_class,
                        resources:    $resources,
                        placement:    $placement,
                        pod_selector: ($comp | get -o pod_selector | default ""),
                        live_check:   ($comp | get -o live_check   | default {}),
                        state:        $entry.state,
                    }
                }
            }
        } else {
            # Cluster: one row, dag-summary state.
            let dag_raw = (comp-dag-state-for $state_root $dag_formulas $comp_name)
            let dag = if ($dag_raw | is-empty) {
                comp-direct-state-for $state_root $comp_name
            } else {
                $dag_raw
            }
            [{
                name:         $comp_name,
                mode:         "cluster",
                server:       "",
                namespace:    ($comp | get -o namespace    | default ""),
                version:      ($comp | get -o version      | default ""),
                class:        $svc_class,
                resources:    $resources,
                placement:    $placement,
                pod_selector: ($comp | get -o pod_selector | default ""),
                live_check:   ($comp | get -o live_check   | default {}),
                state:        (comp-dag-summary $dag),
            }]
        }
    } | flatten
    | if ($server | is-not-empty) {
        where {|r| ($r.mode == "cluster") or ($r.server == $server) }
    } else { $in }
}

# Show the full unified view of a single component declaration.
#
# Exports infra/{infra}/components/{name}.ncl from the workspace. If that file
# does not exist, falls back to the component entry in settings.ncl.
# Returns a record with mode, target, namespace, requires, provides, and operations.
#
# Usage:
#   provisioning component info postgresql --workspace libre-daoshi --infra wuji
# Load a component record + DAG state for use by the status subcommand and info printer.
# Returns a record with: mode, target, namespace, version, pod_selector, live_check,
# requires (normalized list), operations (enabled list), provides, dag (state entries).
export def comp-load-component-record [
    name: string
    ws_root: string
    infra: string
]: nothing -> record {
    let comp_ncl_path = ($ws_root | path join "infra" $infra "components" $"($name).ncl")
    let comp = if ($comp_ncl_path | path exists) {
        let raw = (comp-ncl-export $ws_root ($"infra/($infra)/components/($name).ncl"))
        $raw | get -o $name | default $raw
    } else {
        let settings = (comp-ncl-export $ws_root ($"infra/($infra)/settings.ncl"))
        let components = ($settings | get -o components | default {})
        if not ($name in ($components | columns)) {
            cli-err $"Component '($name)' not found in settings.ncl or components/$name.ncl"
        }
        $components | get $name
    }

    let dag_formulas = (comp-load-dag $ws_root $infra)
    let state_root   = (comp-state-root $ws_root $infra)
    let dag_raw      = (comp-dag-state-for $state_root $dag_formulas $name)
    let dag          = if ($dag_raw | is-empty) { comp-direct-state-for $state_root $name } else { $dag_raw }

    let req_raw  = ($comp | get -o requires | default {})
    let req_norm = if ($req_raw | describe | str starts-with "record") {
        [
            (if ($req_raw | get -o storage | default null | is-not-empty) {
                let s = ($req_raw.storage? | default {})
                [$"storage:($s | get -o size | default '?')"]
            } else { [] }),
            (if ($req_raw | get -o ports | default [] | is-not-empty) {
                let ports = ($req_raw.ports? | default [] | each {|p| $"($p | get -o port | default '?')"} | str join "/")
                [$"ports:($ports)"]
            } else { [] }),
            ($req_raw | get -o credentials | default []),
        ] | flatten
    } else if ($req_raw | describe | str starts-with "list") { $req_raw } else { [] }

    let ops_raw  = ($comp | get -o operations | default {})
    let ops_norm = if ($ops_raw | describe | str starts-with "record") {
        $ops_raw | transpose k v | where {|r| $r.v == true } | each {|r| $r.k}
    } else if ($ops_raw | describe | str starts-with "list") { $ops_raw } else { [] }

    {
        mode:         ($comp | get -o mode         | default "cluster"),
        target:       ($comp | get -o target       | default ""),
        namespace:    ($comp | get -o namespace    | default ""),
        version:      ($comp | get -o version      | default ""),
        pod_selector: ($comp | get -o pod_selector | default ""),
        live_check:   ($comp | get -o live_check   | default {}),
        requires:     $req_norm,
        operations:   $ops_norm,
        provides:     ($comp | get -o provides     | default {}),
        dag:          $dag,
    }
}

export def "main component info" [
    name: string                   # Component name
    --workspace (-w): string = ""  # Workspace name (default: active)
    --infra (-i): string = ""  # Infra name (default: auto-detect)
]: nothing -> nothing {
    let ws_root = (comp-resolve-ws-path $workspace)
    if ($ws_root | is-empty) {
        cli-err "Could not resolve workspace root"
    }
    let infra = if ($infra | is-not-empty) { $infra } else {
        ls ($ws_root | path join "infra")
        | where type == dir
        | get name
        | each { |p| $p | path basename }
        | where { |n| ($ws_root | path join "infra" $n "settings.ncl" | path exists) }
        | get 0?
        | default ""
    }
    if ($infra | is-empty) {
        cli-err $"No infra with settings.ncl found in ($ws_root)/infra/"
    }

    let info = (comp-load-component-record $name $ws_root $infra)
    let comp = do {
        let comp_ncl_path = ($ws_root | path join "infra" $infra "components" $"($name).ncl")
        if ($comp_ncl_path | path exists) {
            let raw = (comp-ncl-export $ws_root ($"infra/($infra)/components/($name).ncl"))
            $raw | get -o $name | default $raw
        } else {
            let settings = (comp-ncl-export $ws_root ($"infra/($infra)/settings.ncl"))
            ($settings | get -o components | default {}) | get -o $name | default {}
        }
    }
    let dag          = $info.dag
    let req_norm     = $info.requires
    let ops_norm     = $info.operations
    let provides_raw = $info.provides

    print $"component  ($name)"
    print $"infra      ($infra)"
    print $"mode       ($info.mode)"
    if ($info.version | is-not-empty)      { print $"version    ($info.version)" }
    if ($info.namespace | is-not-empty)    { print $"namespace  ($info.namespace)" }
    if ($info.target | is-not-empty)       { print $"target     ($info.target)" }
    if ($info.pod_selector | is-not-empty) { print $"pod_selector  ($info.pod_selector)" }

    print ""
    print "requires"
    for $r in $req_norm { print $"  ($r)" }
    if ($req_norm | is-empty) { print "  (none)" }

    print ""
    print "provides"
    if ($provides_raw | columns | is-not-empty) {
        $provides_raw | transpose k v | each {|r| print $"  ($r.k)  ($r.v)"} | ignore
    } else {
        print "  (none)"
    }

    print ""
    print "operations"
    for $op in $ops_norm { print $"  ($op)" }
    if ($ops_norm | is-empty) { print "  (none)" }

    if ($dag | is-not-empty) {
        print ""
        print $"DAG execution — ($dag | length) formulas:"
        for $step in $dag {
            let state_icon = match $step.state {
                "completed" => "✅", "running" => "🔄",
                "failed"    => "❌", "blocked" => "⊘", _ => "⏳",
            }
            let blocker = if ($step.blocker | is-not-empty) { $"  ← blocked by ($step.blocker)" } else { "" }
            print $"  ($state_icon) ($step.formula | fill -a l -w 28) server: ($step.server)  ($step.state)($blocker)"
            let entries = ($step.log? | default [])
            if ($entries | is-empty) {
                print "       history: —"
            } else {
                for $ev in $entries {
                    let ev_icon = match $ev.event {
                        "completed"         => "✅",
                        "started"           => "▶",
                        "failed"            => "❌",
                        "reset-reprovision" => "↺",
                        _                   => "·",
                    }
                    let ts = ($ev.ts | str replace "T" " " | str replace "Z" "")
                    print $"       ($ev_icon) ($ts)  ($ev.event)"
                }
            }
        }
    }
}

# ── Lifecycle operation dispatcher ───────────────────────────────────────────

# Fan-out dispatcher: when server is empty, expands to all active servers from state.
# Falls back to a single call with an empty target when no active servers are found
# (preserves existing resolution logic in component_workflow for new installs).
def component_dispatch [
    ws_root: string
    infra: string
    name: string
    op: string
    server: string
    check: bool
    wait: bool
    timeout: int
    orchestrator: string
    no_orch: bool
    debug: bool
]: nothing -> list<string> {
    use domain/workspace/state.nu [state-component-active-servers]
    let infra_name = ($infra | path basename)
    let targets = if ($server | is-not-empty) {
        [$server]
    } else {
        let active = (state-component-active-servers $ws_root $name)
        if ($active | is-not-empty) { $active } else { [""] }
    }
    mut results = []
    for t in $targets {
        $results = ($results | append (component_workflow $ws_root $infra_name $name $op $t $check $wait $timeout $orchestrator $no_orch $debug))
    }
    $results
}

# Build the ComponentWorkflow payload and submit it via the orchestrator client.
# Returns the task_id. --wait blocks until the task reaches a terminal state.
# no_orch = true: bypass orchestrator, execute directly via the taskserv runner.
# Used by the orchestrator itself to avoid the component→orchestrator loop.
def component_workflow [
    ws_root: string
    infra: string
    name: string
    op: string
    server: string
    check: bool
    wait: bool
    timeout: int
    orchestrator: string
    no_orch: bool = false
    debug: bool = false
]: nothing -> string {
    let fs_name   = ($name | str replace --all "-" "_")
    # State key is always kebab — DAG formula taskserv.name convention. Using a single canonical
    # form prevents duplicate entries when CLI passes snake and orchestrator passes kebab for the
    # same component.
    let state_key = ($name | str replace --all "_" "-")
    let settings = (comp-ncl-export $ws_root ($"infra/($infra)/settings.ncl"))
    let comp = ($settings | get -o components | default {} | get -o $fs_name | default {})
    let mode = ($comp | get -o mode | default "cluster")
    let ns = ($comp | get -o namespace | default "")

    let target = if ($server | is-not-empty) {
        $server
    } else if $mode == "taskserv" {
        let explicit = ($comp | get -o target | default "")
        if ($explicit | is-not-empty) {
            $explicit
        } else {
            $settings | get -o servers | default [] | get 0? | default {} | get -o hostname | default ""
        }
    } else {
        let formulas = (comp-load-dag $ws_root $infra)
        if ($formulas | is-not-empty) {
            $formulas | get 0? | default {} | get -o server | default ""
        } else {
            $settings | get -o servers | default [] | get 0? | default {} | get -o hostname | default ""
        }
    }

    # Auto-redirect: if the component declares a target and the install target differs,
    # look for a {name}_worker variant in settings and use it instead.
    let comp_target = ($comp | get -o target | default "")
    let name = if ($comp_target | is-not-empty) and ($comp_target != $target) {
        let worker_key = $"($fs_name)_worker"
        let has_worker = ($settings | get -o components | default {} | get -o $worker_key | default null) != null
        if $has_worker {
            print $"↪ component ($name) target=($comp_target) ≠ ($target) — using ($worker_key)"
            $worker_key
        } else { $name }
    } else { $name }

    if $no_orch {
        # Direct execution path: fallback when orchestrator is unavailable.
        print $"Executing component ($op) directly: ($name) → ($target)"
        match $mode {
            "taskserv" => {
                use orchestration/taskservs/create.nu *
                let infra_path = ($ws_root | path join "infra" $infra)
                main create $name $target --infra $infra_path --check=$check
            }
            _ => {
                cli-err $"Direct execution not yet supported for deploy mode '($mode)' \(component: ($name)\)"
            }
        }
        return "direct"
    }

    if ($target | is-empty) {
        cli-err $"Cannot resolve target server for component '($name)' — set a target in settings or pass --server"
    }

    let ssh_info = (comp-server-ssh-info $ws_root $infra $target)
    let prov_root = (comp-prov-root)

    # Pre-flight: build + validate the bundle BEFORE any op tracking or state changes.
    # Errors here are clean exits — no jj/rad/op-log/state-node noise.
    print $"[preflight] Validating component bundle: ($name) → ($target)"
    let bundle_result = if $mode == "cluster" {
        comp-build-cluster-bundle $name $ws_root $infra $target $op $settings $debug
    } else {
        let b64 = (comp-build-bundle $name $ws_root $infra $target $op $settings)
        {b64: $b64, work_dir: ""}
    }

    if $check {
        print $"[preflight] OK — bundle validated. check_mode=true, not submitting."
        if $debug { print $"[preflight] work_dir preserved: ($bundle_result.work_dir)" }
        if ($bundle_result.work_dir | is-not-empty) and (not $debug) { ^rm -rf $bundle_result.work_dir }
        return "check"
    }

    let payload = {
        workspace:         $ws_root,
        infra:             $infra,
        component:         $name,
        server:            $ssh_info.host,
        namespace:         (if ($ns | is-not-empty) { $ns } else { null }),
        ssh_user:          $ssh_info.user,
        ssh_key_path:      (if ($ssh_info.key | is-not-empty) { $ssh_info.key } else { null }),
        settings:          ($settings | to json),
        check_mode:        $check,
        debug:             $debug,
        provisioning:      $prov_root,
        script_compressed: $bundle_result.b64,
        mode:              $mode,
    }

    # Service health gates BEFORE any state mutation — clean exit with no side effects on failure.
    health-gate $"component ($name) ($op)"
    let orch_url = (orch-url --orchestrator $orchestrator)
    if not (orch-health --orchestrator $orchestrator) {
        cli-err $"Orchestrator is not running at ($orch_url)\nStart it with: nu provisioning/platform/scripts/start-provisioning-daemon.nu"
    }

    # All gates passed — safe to record op start now.
    let actor = ($env.USER? | default "system")
    let effective_op = match $op {
        "install" | "reinstall" => "create",
        $o => $o,
    }
    let state_root = (comp-state-root $ws_root $infra)
    state-node-start $state_root $target $state_key --actor $actor --source "cli" --operation $effective_op

    print $"Submitting component ($op): ($name) → ($target)"
    let task_id = (orch-submit-component $payload $op --orchestrator $orchestrator)

    if $wait {
        let final = (orch-wait-task $task_id --timeout $timeout --orchestrator $orchestrator)
        let status = ($final | get -o status | default "unknown" | str downcase)

        match $status {
            "completed" => {
                match $op {
                    "delete" => {
                        state-node-delete $state_root $target $state_key
                    },
                    "reinstall" => {
                        state-node-reset $state_root $target $state_key --source "cli" --actor $actor
                    },
                    _ => {
                        state-node-finish $state_root $target $state_key --success --source "cli"
                    },
                }
                # Provisioning-side DNS reconcile — nodes carry no DNS tooling, so the
                # A record(s) are upserted here with native http/json after a successful
                # install. Never CRASHES the deploy over DNS, but the outcome is captured
                # and a miss is surfaced loudly (not the old silent `do -i`): cloudflare-dns.nu
                # verifies the record and exits non-zero when it is not confirmed.
                # DNS-degraded tracking (ADR-051 postop-degrades-and-notifies): the
                # WARN sites below no longer just print — a failed verify degrades the
                # op. component_dispatch owns no op record (op start/finish are separate
                # operator calls), so the degrade surfaces as: a [DEGRADED] log line
                # (always), an op.json field (when PROVISIONING_OP_ID is exported), and
                # a non-zero exit when ops.dns.strict = true (default false = WARN-only).
                mut dns_degraded = false
                mut dns_degraded_detail = ""
                if $mode == "cluster" and $op != "delete" {
                    let dns = ($bundle_result | get -o dns | default {})
                    let dns_mod = ($prov_root | path join "catalog" "lib" "cloudflare-dns.nu")
                    let dns_domain = ($dns | get -o domain | default "")
                    let dns_ip     = ($dns | get -o target_ip | default "")
                    # Domain declared but no resolved target — the reconcile cannot publish.
                    # Do not hide it (this was the ontoref silent skip: gateway_ip empty).
                    if ($dns_domain | is-not-empty) and ($dns_ip | is-empty) {
                        print $"  [WARN] DNS: ($dns_domain) has no resolved target_ip (gateway_ip empty) — A record NOT published; the domain will not resolve. Check gateway_fip / hcloud auth."
                        $dns_degraded = true
                        $dns_degraded_detail = $"($dns_domain): no resolved target_ip (gateway_ip empty)"
                    }
                    if ($dns_domain | is-not-empty) and ($dns_ip | is-not-empty) {
                        let dns_r = (do {
                            with-env { CF_API_TOKEN: ($dns | get -o token | default "") } {
                                ^nu $dns_mod $dns_domain $dns_ip ($dns | get -o zone | default "")
                            }
                        } | complete)
                        if not ($dns_r.stdout | str trim | is-empty) { print ($dns_r.stdout | str trim) }
                        if $dns_r.exit_code != 0 {
                            print $"  [WARN] DNS reconcile NOT confirmed for ($dns_domain) → ($dns_ip); deploy continued but the domain may not resolve — publish it or re-run the op."
                            $dns_degraded = true
                            $dns_degraded_detail = $"($dns_domain) → ($dns_ip): reconcile not confirmed"
                        }
                    }
                    let dns_sites = ($dns | get -o sites | default [])
                    if ($dns_sites | is-not-empty) {
                        let tmp = (mktemp --suffix ".json")
                        $dns_sites | to json | save --force $tmp
                        let many_r = (do { ^nu $dns_mod many $tmp } | complete)
                        if not ($many_r.stdout | str trim | is-empty) { print ($many_r.stdout | str trim) }
                        if $many_r.exit_code != 0 {
                            print "  [WARN] DNS multi-site reconcile reported unconfirmed records — see output above."
                            $dns_degraded = true
                            $dns_degraded_detail = "multi-site reconcile: unconfirmed records"
                        }
                        rm -f $tmp
                    }
                }
                # Delete-path DNS inversion (ADR-051 delete-inverts-cleanup): the
                # install/update reconcile above publishes the A record, so delete
                # must remove it or the component leaves an orphaned record. Runs on
                # the provisioner (token locality), symmetric with the upsert path.
                if $mode == "cluster" and $op == "delete" {
                    let dns = ($bundle_result | get -o dns | default {})
                    let dns_domain = ($dns | get -o domain | default "")
                    if ($dns_domain | is-not-empty) {
                        let dns_mod = ($prov_root | path join "catalog" "lib" "cloudflare-dns.nu")
                        let del_r = (do {
                            with-env { CF_API_TOKEN: ($dns | get -o token | default "") } {
                                ^nu $dns_mod delete $dns_domain ($dns | get -o zone | default "")
                            }
                        } | complete)
                        if not ($del_r.stdout | str trim | is-empty) { print ($del_r.stdout | str trim) }
                        if $del_r.exit_code != 0 {
                            print $"  [WARN] DNS delete NOT confirmed for ($dns_domain); the A record may be orphaned — remove it manually or re-run the delete."
                            $dns_degraded = true
                            $dns_degraded_detail = $"($dns_domain): delete not confirmed (A record may be orphaned)"
                        }
                    }
                }
                # Postdeploy escalation (ADR-051 postop-degrades-and-notifies): a
                # degraded DNS state must be observable + recorded, never silently green.
                if $dns_degraded {
                    print $"  [DEGRADED] dns: ($dns_degraded_detail)"
                    # Persist to the op record only when the wrapping op exported its id
                    # (dispatch has no op context of its own). No-op when absent/missing.
                    let op_id_env = ($env.PROVISIONING_OP_ID? | default "")
                    if ($op_id_env | is-not-empty) {
                        let op_json = ($ws_root | path join "infra" $infra "ops" $op_id_env "op.json")
                        if ($op_json | path exists) {
                            let rec = (open $op_json)
                            $rec | merge { degraded: { dns: $dns_degraded_detail } } | to json --indent 2 | save --force $op_json
                            print $"  [DEGRADED] recorded in op ($op_id_env)"
                        }
                    }
                    # ops.dns.strict = true → fail the process so the wrapping op is
                    # marked failed. Default false preserves today's WARN-only behavior
                    # for all workspaces that have not opted in.
                    let prov_cfg = (try { comp-ncl-export $ws_root "config/provisioning.ncl" } catch { {} })
                    let dns_strict = ($prov_cfg | get -o ops.dns.strict | default false)
                    if $dns_strict {
                        error make {msg: $"Op degraded — DNS not confirmed: ($dns_degraded_detail) \(ops.dns.strict=true\)"}
                    }
                }
                print $"Done: ($status)"
            },
            _ => {
                state-node-finish $state_root $target $state_key --source "cli"
                cli-err $"Component ($op) failed: ($final | get -o error | default 'no detail')"
            },
        }
    }
    $task_id
}

# Install a component on the target server.
export def "main component install" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --check
    --wait
    --timeout: int = 300
    --orchestrator: string = ""
    --no-orchestrator    # bypass orchestrator, execute directly
    --debug (-x)         # bash -x tracing on remote; leaves /tmp/{name} intact for inspection
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "install" $server $check $wait $timeout $orchestrator $no_orchestrator $debug
}

# Delete a component from the target server.
export def "main component delete" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --check
    --wait
    --timeout: int = 300
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "delete" $server $check $wait $timeout $orchestrator $no_orchestrator $debug
}

# Update a component to the latest declared version.
export def "main component update" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --check
    --wait
    --timeout: int = 300
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "update" $server $check $wait $timeout $orchestrator $no_orchestrator $debug
}

# Reinstall a component (delete + install in one operation).
export def "main component reinstall" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --check
    --wait
    --timeout: int = 300
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "reinstall" $server $check $wait $timeout $orchestrator $no_orchestrator $debug
}

# Restart a running component.
export def "main component restart" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --wait
    --timeout: int = 120
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "restart" $server false $wait $timeout $orchestrator $no_orchestrator $debug
}

# Run a backup of a component's persistent state.
export def "main component backup" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --wait
    --timeout: int = 600
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "backup" $server false $wait $timeout $orchestrator $no_orchestrator $debug
}

# Restore a component from its most recent backup.
export def "main component restore" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --check
    --wait
    --timeout: int = 600
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "restore" $server $check $wait $timeout $orchestrator $no_orchestrator $debug
}

# Render a component bundle locally for audit — no ops, no SSH, no deployment.
# Outputs rendered manifests and generated run-*.sh scripts to <out_dir>/<component>/.
# Useful for reviewing what would be deployed before committing to an operation.
export def "main component render" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --out (-o): string = "audit"    # output directory (relative to cwd)
    --op: string = "init"           # which op's plan scripts to include (init|update|delete|restart)
    --bundle                        # also write <name>.tar.gz.b64 alongside rendered files
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let settings   = (comp-ncl-export $ws_root $"infra/($infra_name)/settings.ncl")

    let fs_name  = ($name | str replace --all "-" "_")
    let mode     = ($settings | get -o components | default {} | get -o $fs_name | default {} | get -o mode | default "cluster")
    let bundle_result = if $mode == "cluster" {
        comp-build-cluster-bundle $name $ws_root $infra_name "" $op $settings false true
    } else {
        let b64 = (comp-build-bundle $name $ws_root $infra_name "" $op $settings)
        {b64: $b64, work_dir: ""}
    }
    let bundle_b64 = $bundle_result.b64
    if ($bundle_result.work_dir | is-not-empty) { ^rm -rf $bundle_result.work_dir }

    let out_dir = ($out | path expand | path join $name)
    ^mkdir -p $out_dir

    let tmp_tar = (^mktemp | str trim)
    $bundle_b64 | ^base64 -d | save --force $tmp_tar
    ^tar -xzf $tmp_tar -C $out_dir
    ^rm -f $tmp_tar

    if $bundle {
        let b64_path = ($out_dir | path join $"($name).tar.gz.b64")
        $bundle_b64 | save --force $b64_path
        print $"Bundle archive → ($b64_path)"
    }

    print $"Rendered bundle → ($out_dir)"
    print $"Contents:"
    ^find $out_dir -type f | lines | sort | each {|f| print $"  ($f)" }
}

# Check for available updates for a component.
export def "main component check-updates" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --orchestrator: string = ""
    --no-orchestrator
    --debug (-x)
]: nothing -> list<string> {
    let ws_root = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    component_dispatch $ws_root $infra_name $name "check-updates" $server false false 60 $orchestrator $no_orchestrator $debug
}

# Mark a component as completed in the provisioning state, with full attribution.
#
# Use when a component was deployed manually, by an agent, or when the orchestrator
# lost the completion callback (state stuck in 'running with ended_at="").
# Records actor identity, timestamp, and an optional verification note in the log.
#
# Usage:
#   provisioning component mark-complete postgresql --note "verified via kubectl get sts -n data"
export def "main component mark-complete" [
    name: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
    --note: string = ""
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)
    let actor      = ($env.USER? | default "system")
    let normalized = ($name | str replace --all "_" "-")

    # Resolve server: explicit flag → state-file scan → first DAG formula server.
    let target = if ($server | is-not-empty) {
        $server
    } else {
        # Scan state file for any server that has this component registered.
        let st = (state-read $state_root)
        let from_state = (
            $st.servers
            | transpose hostname srv
            | where {|row|
                let ts = ($row.srv.taskservs? | default {})
                ($ts | get -o $name | default {} | is-not-empty) or ($ts | get -o $normalized | default {} | is-not-empty)
            }
            | get 0?
            | get -o hostname
            | default ""
        )
        if ($from_state | is-not-empty) {
            $from_state
        } else {
            # Fall back to first DAG formula server that contains this component node.
            let dag_server = (
                comp-load-dag $ws_root $infra_name
                | each {|f|
                    let matched = ($f.nodes | where {|n| (comp-node-name $n) == $normalized })
                    if ($matched | is-not-empty) { [$f.server] } else { [] }
                }
                | flatten | get 0? | default ""
            )
            $dag_server
        }
    }

    if ($target | is-empty) {
        cli-err $"Cannot resolve server for component '($name)' — pass --server"
    }

    # Canonical key is kebab (matches DAG formula taskserv.name convention and the new
    # state_key normalization in component_workflow). Prefer $normalized (kebab) when both
    # forms are present — avoids leaving the DAG-tracked key stale after mark-complete.
    let server_ts = (state-read $state_root | get -o servers | default {} | get -o $target | default {} | get -o taskservs | default {})
    let active_key = if ($server_ts | get -o $normalized | default {} | is-not-empty) {
        $normalized
    } else {
        $name
    }

    let ts = ((date now) | format date "%Y-%m-%dT%H:%M:%SZ")
    let existing = (state-node-get $state_root $target $active_key)
    let event_label = if ($note | is-not-empty) { $"mark-complete: ($note)" } else { "mark-complete" }
    let updated_log = (log-trim ($existing.log? | default [] | append {
        ts: $ts, event: $event_label, source: "cli"
    }))

    state-node-set $state_root $target $active_key {
        state:    "completed",
        ended_at: $ts,
        actor:    { identity: $actor, source: "cli" },
        log:      $updated_log,
    }

    print $"component  ($name)"
    print $"server     ($target)"
    print $"actor      ($actor)"
    print $"timestamp  ($ts)"
    if ($note | is-not-empty) { print $"note       ($note)" }
    print "state      completed"
}

# Sweep all stale components (state=running with empty ended_at) to completed in one write.
#
# Use after an orchestrator crash or manual deploy session where callbacks were lost.
# Pass --include-failed to also sweep 'failed' entries (e.g. cilium/longhorn that failed
# on first script attempt but are actually running — verify pods before using this flag).
# Pass --dry-run to preview without writing.
export def "main component bulk-mark-complete" [
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --note: string = ""
    --include-failed  # Also sweep state=failed entries
    --dry-run         # Print what would change, no writes
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)
    let actor      = ($env.USER? | default "system")
    let ts         = ((date now) | format date "%Y-%m-%dT%H:%M:%SZ")

    mut st = (state-read $state_root)

    # Collect all stale (server, key) pairs from the in-memory record.
    let stale = (
        $st.servers
        | transpose hostname srv
        | each {|row|
            $row.srv.taskservs?
            | default {}
            | transpose name node
            | where {|e|
                let s = ($e.node | get -o state | default "pending")
                let ended = ($e.node | get -o ended_at | default "")
                if $include_failed {
                    ($s == "running" and $ended == "") or $s == "failed"
                } else {
                    $s == "running" and $ended == ""
                }
            }
            | each {|e| { hostname: $row.hostname, name: $e.name, state: ($e.node | get -o state | default "") } }
        }
        | flatten
    )

    if ($stale | is-empty) {
        print "No stale entries found."
        return
    }

    print $"Found ($stale | length) stale entries:"
    $stale | each {|e| print $"  ($e.hostname)  ($e.name)  [($e.state)]" }

    if $dry_run { return }

    let event_label = if ($note | is-not-empty) { $"mark-complete: ($note)" } else { "bulk-mark-complete" }

    # Mutate in memory, single write at the end.
    # canonical_key is always kebab — matches DAG formula convention.
    for e in $stale {
        let canonical_key = ($e.name | str replace --all "_" "-")
        let current_server = (
            $st.servers
            | transpose k v
            | where { |r| $r.k == $e.hostname }
            | get -o v.0
            | default { provider_id: "", provider_state: "unknown", last_sync: "", taskservs: {} }
        )
        # Merge logs from both the original key and the canonical key (if they differ)
        # so no history is lost when normalizing snake → kebab.
        let original_node = (
            $current_server.taskservs?
            | default {}
            | transpose k v
            | where { |r| $r.k == $e.name }
            | get -o v.0
            | default { state: "pending", operation: "create", profile: "", started_at: "", ended_at: "", blocker: "", actor: { identity: "", source: "cli" }, log: [] }
        )
        let canonical_node = if $canonical_key != $e.name {
            $current_server.taskservs?
            | default {}
            | transpose k v
            | where { |r| $r.k == $canonical_key }
            | get -o v.0
            | default {}
        } else { {} }
        let merged_log = (
            ($original_node.log? | default [])
            | append ($canonical_node | get -o log | default [])
            | uniq-by ts
            | sort-by ts
        )
        let base_node = if ($canonical_node | is-not-empty) { $canonical_node } else { $original_node }
        let updated_log = (log-trim ($merged_log | append { ts: $ts, event: $event_label, source: "cli" }))
        let merged = ($base_node | merge {
            state:    "completed",
            ended_at: $ts,
            actor:    { identity: $actor, source: "cli" },
            log:      $updated_log,
        })
        # Write to canonical key, delete snake duplicate if it differed.
        mut new_ts = ($current_server.taskservs? | default {} | upsert $canonical_key $merged)
        if $canonical_key != $e.name and ($new_ts | columns | any { $in == $e.name }) {
            $new_ts = ($new_ts | reject $e.name)
        }
        let new_server = ($current_server | upsert taskservs $new_ts)
        let new_servers = (
            $st.servers
            | transpose k v
            | each { |r| if $r.k == $e.hostname { { k: $r.k, v: $new_server } } else { $r } }
            | transpose -r -d
        )
        $st.servers = $new_servers
    }

    state-write $state_root $st

    print ""
    $stale | each {|e|
        let canonical_key = ($e.name | str replace --all "_" "-")
        if $canonical_key != $e.name {
            print $"✓ ($e.hostname)  ($e.name) → ($canonical_key)  [normalized]"
        } else {
            print $"✓ ($e.hostname)  ($e.name)"
        }
    }
    print $"\nMarked ($stale | length) components as completed."
}

# Drop a specific state key from a server's taskservs — prunes duplicate or orphaned entries.
#
# Use when the same component exists under two keys (e.g. both 'odoo-jesusperez' and
# 'odoo_jesusperez' after a naming migration). Identify the stale key with 'prvng c ll'
# then drop it here. The surviving key retains its full history.
export def "main component drop-key" [
    name: string          # Exact key to delete from state
    --server (-s): string # Target server hostname (required — no auto-resolution)
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --dry-run
]: nothing -> nothing {
    if ($server | is-empty) {
        cli-err "Pass --server <hostname> — no auto-resolution for drop-key (safety measure)"
    }
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)

    let st = (state-read $state_root)
    let existing = ($st.servers | get -o $server | default {} | get -o taskservs | default {} | get -o $name | default null)
    if ($existing | is-empty) {
        cli-err $"Key '($name)' not found on server '($server)'"
    }

    print $"Will drop:  ($server)  ($name)  [state=($existing | get -o state | default '?')]"
    if $dry_run { return }

    state-node-delete $state_root $server $name
    print "Dropped."
}

# Remove an entire server block from the state file.
#
# Use when a server has been decommissioned and its state entries are orphaned.
# All component states for that server are deleted in one operation.
export def "main component drop-server" [
    hostname: string
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --dry-run
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)

    let st = (state-read $state_root)
    let srv = ($st.servers | get -o $hostname | default null)
    if ($srv | is-empty) {
        cli-err $"Server '($hostname)' not found in state"
    }

    let taskserv_count = ($srv | get -o taskservs | default {} | columns | length)
    print $"Will drop server: ($hostname)  [($taskserv_count) taskserv entries]"
    if $dry_run { return }

    let new_servers = (
        $st.servers
        | transpose k v
        | where { $in.k != $hostname }
        | each {|r| { ($r.k): $r.v } }
        | reduce --fold {} {|a, acc| $acc | merge $a }
    )
    let new_st = ($st | upsert servers $new_servers)
    state-write $state_root $new_st
    print $"Dropped server ($hostname) with ($taskserv_count) entries."
}

# Diagnose state coherence: lists components whose provisioning state doesn't match
# live pod status. Outputs a table of incoherent entries so you know what to fix.
#
# Incoherent cases:
#   stale-running  — state=running with empty ended_at (callback lost)
#   failed-but-up  — state=failed but live pods found (transient failure, manually recovered)
#   duplicate-keys — both kebab and snake_case key present for the same component
export def "main component state-check" [
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --server (-s): string = ""
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)

    let st = (state-read $state_root)

    let entries = (
        $st.servers
        | transpose hostname srv
        | where {|row| if ($server | is-not-empty) { $row.hostname == $server } else { true } }
        | each {|row|
            $row.srv.taskservs?
            | default {}
            | transpose name node
            | each {|e| {
                hostname: $row.hostname,
                name:     $e.name,
                state:    ($e.node | get -o state    | default "pending"),
                ended_at: ($e.node | get -o ended_at | default ""),
            }}
        }
        | flatten
    )

    # stale-running: state=running with empty ended_at
    let stale_running = ($entries | where { $in.state == "running" and $in.ended_at == "" })

    # failed: state=failed (may be recovered)
    let failed = ($entries | where { $in.state == "failed" })

    # duplicate keys: a kebab entry AND a snake entry exist for the same component on the same server
    let duplicates = (
        $entries
        | group-by { |e| $"($e.hostname)/($e.name | str replace --all '-' '_')" }
        | transpose normalized group
        | where { ($in.group | length) > 1 }
        | each {|row|
            $row.group | each {|e| $e | merge { issue: "duplicate-key", normalized: $row.normalized } }
        }
        | flatten
    )

    print "── STALE RUNNING (callback lost) ──────────────────────────────────────"
    if ($stale_running | is-empty) {
        print "  none"
    } else {
        $stale_running | select hostname name state ended_at | print
        print $"\n  Fix: prvng c bulk-mark-complete"
    }

    print "\n── FAILED (may be recovered) ──────────────────────────────────────────"
    if ($failed | is-empty) {
        print "  none"
    } else {
        $failed | select hostname name state ended_at | print
        print "\n  Fix: prvng c mark-complete <name> --server <hostname>"
    }

    print "\n── DUPLICATE KEYS (kebab+snake coexist) ───────────────────────────────"
    if ($duplicates | is-empty) {
        print "  none"
    } else {
        $duplicates | select hostname name state | print
        print "\n  Fix: prvng c drop-key <stale-key> --server <hostname>"
    }

    let total = (($stale_running | length) + ($failed | length) + ($duplicates | length))
    print $"\nTotal issues: ($total)"
    if $total > 0 {
        print "\nRun 'prvng c state-normalize' to merge snake→kebab duplicates in one pass."
    }
}

# Pure helper: given a taskservs record, return a new record with all snake_case keys
# merged into their kebab equivalents. No mutation — safe in closures.
def state-normalize-taskservs [ts: record]: nothing -> record {
    let snake_keys = ($ts | columns | where { ($in | str contains "_") and ($in | str replace --all "_" "-") != $in })
    if ($snake_keys | is-empty) { return $ts }
    $snake_keys | reduce --fold $ts {|sk, acc|
        let kk          = ($sk | str replace --all "_" "-")
        let snake_node  = ($acc | get $sk)
        let kebab_node  = ($acc | get -o $kk | default {})
        let merged_log  = (
            ($snake_node.log? | default [])
            | append ($kebab_node | get -o log | default [])
            | uniq-by ts
            | sort-by ts
        )
        let base        = if ($kebab_node | is-not-empty) { $kebab_node } else { $snake_node }
        let final_node  = ($base | upsert log $merged_log)
        $acc | upsert $kk $final_node | reject $sk
    }
}

# One-shot migration: normalize all snake_case state keys to kebab-case.
#
# Merges each snake_case entry into its kebab counterpart (creating it if absent),
# preserving the full log union. After this command, every key in the state file uses
# kebab-case — consistent with the DAG formula convention and the new state_key write path.
# Run once after upgrading, then verify with 'prvng c sc'.
export def "main component state-normalize" [
    --workspace (-w): string = ""
    --infra (-i): string = ""
    --dry-run
]: nothing -> nothing {
    let ws_root    = (comp-resolve-ws-path $workspace)
    let infra_name = if ($infra | is-not-empty) { $infra } else { comp-ws-infra $ws_root }
    let state_root = (comp-state-root $ws_root $infra_name)

    let st = (state-read $state_root)

    # Compute migration plan (pure — no writes).
    let plan = (
        $st.servers
        | transpose hostname srv
        | each {|row|
            let ts = ($row.srv.taskservs? | default {})
            $ts | columns
            | where { ($in | str contains "_") and ($in | str replace --all "_" "-") != $in }
            | each {|sk| { hostname: $row.hostname, from: $sk, to: ($sk | str replace --all "_" "-") } }
        }
        | flatten
    )

    if ($plan | is-empty) {
        print "No snake_case keys found — state is already normalized."
        return
    }

    print $"($plan | length) keys to normalize:"
    $plan | each {|m| print $"  ($m.hostname)  ($m.from) → ($m.to)" }

    if $dry_run { return }

    # Build new servers record purely: normalize taskservs per server, then reassemble.
    let new_servers = (
        $st.servers
        | transpose hostname srv
        | each {|row|
            let new_ts = (state-normalize-taskservs ($row.srv.taskservs? | default {}))
            { ($row.hostname): ($row.srv | upsert taskservs $new_ts) }
        }
        | reduce --fold {} {|a, acc| $acc | merge $a }
    )

    let new_st = ($st | upsert servers $new_servers)
    state-write $state_root $new_st

    print "\nNormalized. Run 'prvng c sc' to verify."
}