newt-agent 0.8.0

//! `newt dgx` — NVIDIA DGX endpoint management.
//!
//! - `setup`  — first-run DGX configuration wizard; writes `[dgx]` to config.
//! - `route`  — classify a task and recommend a (model, endpoint) formation.
//! - `status` — active-endpoint health + models currently loaded on the DGX.
//! - `models` — list models available on the active endpoint.
//! - `doctor` — probe every configured endpoint flavor + DNS guidance.
//! - `pull`   — pull an Ollama model, OR a HuggingFace GGUF (with automated
//!   sharded-GGUF workaround for ollama/ollama#5245) onto the node.
//! - `rm`     — delete a model from the active Ollama endpoint (`/api/delete`).
//! - `ps`     — list models currently loaded on the active endpoint (`/api/ps`).
//!
//! Later Phase 14 steps add `endpoint`/`formation`/`node`, SSH ops
//! (`run`/`push`/`watch`), and `nim`.
//!
//! ## `pull` and the sharded-GGUF workaround
//!
//! `ollama pull hf.co/<org>/<repo>:<quant>` fails on multi-part (sharded) GGUF
//! repos with `400: The specified tag is a sharded GGUF. Ollama does not
//! support this yet` (ollama/ollama#5245). For those, `pull` downloads each
//! shard onto the node with resumable `curl` and runs `ollama create` from a
//! generated `Modelfile`. Before any of that it runs a **fit pre-flight**
//! (the GLM-5.2 lesson): if the model's on-disk size exceeds the node's RAM it
//! refuses unless `--force`, because such a model is effectively unrunnable
//! (heavy disk paging). The pure logic lives in [`crate::dgx_pull`]; SSH/HF
//! execution is injectable so tests never touch the network or a real node.

use std::path::Path;

use std::io::IsTerminal as _;

use clap::Subcommand;
use newt_core::dgx::{DgxConfig, DgxNode, EndpointKind};
use newt_core::retry::{with_backoff, RetryPolicy};
use newt_core::router::Classification;
use newt_core::{Config, Router, Tier};
use newt_inference::local::LocalVllmBackend;

use crate::dgx_registry::{self, InferenceTool, ModelVariant};
use crate::dgx_status;
use crate::dgx_vllm;

mod ssh;
use ssh::{detect_node_mem, detect_node_mem_available, RealSsh, RealSshCapture, SshExec};
#[cfg(test)]
use ssh::{node_mem_probe, MEM_AVAILABLE_AWK, MEM_TOTAL_AWK};

/// `newt dgx <cmd>` subcommands.
#[derive(Subcommand, Debug)]
pub enum DgxCmd {
    /// First-run DGX setup: write a [dgx] block to ~/.newt/config.toml.
    ///
    /// With no arguments, prints setup instructions. With --host, synthesizes
    /// Ollama / vLLM endpoint URLs from the bare hostname or IP and writes the
    /// config (atomically via Config::save). Use --template to dump the
    /// reference template as TOML without writing anything.
    Setup {
        /// DGX hostname or IP (e.g. REDACTED-IP or REDACTED-HOST).
        #[arg(long)]
        host: Option<String>,

        /// Node name stored in config (default: "dgx").
        #[arg(long, default_value = "dgx")]
        name: String,

        /// Active model id (e.g. qwen2.5-coder:32b, llama3.1:8b).
        #[arg(long)]
        model: Option<String>,

        /// Print the reference template as TOML and exit without writing.
        #[arg(long)]
        template: bool,

        /// Skip the write-confirmation prompt (useful in scripts / tests).
        #[arg(long, short = 'y')]
        yes: bool,
    },
    /// Classify a task and recommend a model + endpoint formation.
    Route {
        /// Task description to classify (quote multi-word tasks).
        task: String,
    },
    /// Show active-endpoint health, the models loaded on the DGX, and the node
    /// memory budget (issue #709 §1): total / used / available memory plus the
    /// running inference workloads (ollama / vllm), read over SSH from
    /// `/proc/meminfo` + `ps`. This replaces the old `nvidia-smi` placeholder.
    Status {
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Emit ONLY the structured memory budget as JSON (issue #709 §1 shape)
        /// — `{ total_gib, used_gib, available_gib, running_workloads[] }`.
        #[arg(long)]
        json: bool,
    },
    /// Select, clear for, and launch the strongest fitting model on the node
    /// (issue #709 §4) — the high-level deploy verb, decoupled from the
    /// low-level `vllm up` knobs.
    ///
    /// With `--strongest`, reads the node's available memory budget over SSH and
    /// picks the strongest registry model that fits (15% headroom, single node).
    /// Otherwise deploys the explicit `<model>` slug (e.g. `ornith-35b-fp16`).
    /// Either way it first evicts competing workloads, then launches — Mode A
    /// invokes the operator's convention script `~/ornith-{size}-{format}.sh`
    /// when present on the node, else Mode B falls back to a generated
    /// `vllm serve`. Finally it polls the endpoint until it is serving.
    Deploy {
        /// Deploy the strongest model that fits the node's available budget.
        #[arg(long, conflicts_with = "model")]
        strongest: bool,
        /// Deploy a specific variant by its convention slug, e.g.
        /// `ornith-35b-fp16` (see `~/ornith-{size}-{format}.sh`). Required
        /// unless `--strongest` is given.
        #[arg(required_unless_present = "strongest")]
        model: Option<String>,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
    },
    /// Evict inference workloads on the node and report the recovered budget
    /// (issue #709 §3).
    ///
    /// Stops any running vLLM server and unloads every resident Ollama model to
    /// free the shared unified-memory pool. With `--restart-ollama`, also
    /// restarts the ollama service over SSH to drop CUDA buffers an idle-but-
    /// loaded runner still pins. Reports the memory freed (budget before/after).
    Clear {
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Also restart the ollama service over SSH to evict CUDA buffers.
        #[arg(long)]
        restart_ollama: bool,
    },
    /// List the models available on the active DGX endpoint.
    Models,
    /// Probe every configured DGX endpoint flavor and report reachability.
    Doctor,
    /// Set the active DGX model and persist it to ~/.newt/config.toml.
    Use {
        /// Model id to activate (e.g. gemma4:e2b, qwen2.5-coder:32b).
        model: String,
    },
    /// Pre-load a model into VRAM on the active endpoint so the first real
    /// request doesn't pay the cold-load latency (which can blow past tight
    /// per-task timeouts, e.g. in `newt-eval`). Uses Ollama's load-only
    /// request — no tokens generated — and pins it resident via `keep_alive`.
    Warm {
        /// Model to warm. Defaults to the active model
        /// (`[dgx].active_model` / `NEWT_DGX_MODEL`).
        model: Option<String>,
        /// How long Ollama keeps the model resident after warming.
        #[arg(long, default_value = "30m")]
        keep_alive: String,
        /// First stop any running vLLM server to free the shared unified pool,
        /// then warm against the Ollama endpoint (the reverse of
        /// `vllm up --evict-ollama`). Implies the Ollama endpoint even if the
        /// active endpoint was vLLM.
        #[arg(long)]
        evict_vllm: bool,
    },
    /// Pull a model onto the DGX node.
    ///
    /// A plain Ollama name (`qwen2.5-coder:32b`) is pulled directly. A
    /// HuggingFace GGUF reference (`unsloth/Repo-GGUF:Q8_0` or
    /// `hf.co/unsloth/Repo-GGUF:Q8_0`) takes the smart path: it queries the HF
    /// API for the quant's GGUF files, runs a fit pre-flight against node RAM,
    /// and — for sharded (multi-part) GGUF — downloads each shard with `curl`
    /// and `ollama create`s from a generated `Modelfile` (the documented
    /// workaround for ollama/ollama#5245).
    Pull {
        /// Model to pull (Ollama name or HuggingFace `<org>/<repo>:<quant>`).
        model: String,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Override the resulting Ollama model name (sharded path only;
        /// default is a sanitized `<repo>-<quant>`).
        #[arg(long)]
        name: Option<String>,
        /// Proceed even when the model is larger than node RAM (the fit
        /// pre-flight would otherwise refuse).
        #[arg(long)]
        force: bool,
        /// Print the resolved plan and the exact remote command/script without
        /// SSHing or executing anything.
        #[arg(long)]
        dry_run: bool,
    },
    /// Delete a model from the active Ollama endpoint (`/api/delete`).
    Rm {
        /// Model id to delete (e.g. qwen2.5-coder:32b).
        model: String,
    },
    /// List models currently loaded on the active Ollama endpoint (`/api/ps`).
    Ps,
    /// Manage a vLLM OpenAI-compatible server on the DGX node.
    ///
    /// Stands up (`up`) / tears down (`down`) a `vllm serve` process over SSH,
    /// reports the served models (`ps`), prints the resolved launch plan
    /// (`config`), or tails the server log (`logs`). Pure planning lives in
    /// [`crate::dgx_vllm`]; SSH/HTTP execution is injectable so tests never
    /// touch the network or a real node.
    Vllm {
        #[command(subcommand)]
        cmd: VllmCmd,
    },
    /// Model cards — every setting to serve one model, as data (#855).
    ///
    /// A card bundles the vLLM / ollama serving profile + tuning + capability
    /// flags for one model. Built-ins ship compiled in (Ornith); a user drops
    /// `~/.newt/models/<name>.toml|.yaml` to override a built-in or teach newt a
    /// new model. `list` / `show` / `validate` are the discover-and-inspect
    /// surface; standing a backend up (`setup`) is the #855 follow-on.
    Card {
        #[command(subcommand)]
        cmd: CardCmd,
    },
    /// Cross-engine GPU residency snapshot: what Ollama and vLLM each hold on
    /// the node right now, plus available memory.
    ///
    /// On the unified-memory GB10 both engines draw from one ~117 GiB pool and
    /// neither negotiates, so this is the operator's window into contention.
    /// Live-queried (Ollama `/api/ps` + vLLM `/v1/models` + `MemAvailable`) —
    /// ground truth, not a cached lease file.
    Gpu,
    /// Flip the node to a target engine + model, evicting the other engine.
    ///
    /// `switch vllm <model>` unloads any resident Ollama models, launches the
    /// vLLM server, and points newt at it. `switch ollama <model>` stops the
    /// vLLM server, verifies the model is pulled (refusing with a hint if not),
    /// points newt at Ollama, and warms it. Composes the `vllm up`/`warm`
    /// eviction swaps into one atomic flip.
    Switch {
        /// Target engine: `ollama` or `vllm`.
        engine: String,
        /// Model to activate on the target engine.
        model: String,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Print the plan without evicting, launching, or persisting.
        #[arg(long)]
        dry_run: bool,
    },
    /// Reconcile newt's configured endpoint/model with what the node is
    /// ACTUALLY running.
    ///
    /// Interrogates the node (vLLM `/v1/models` + Ollama `/api/ps`), derives the
    /// live engine+model, and compares it to `[dgx].active_endpoint`/`active_model`.
    /// On a mismatch you choose: **adopt** the live state into config (point newt
    /// at reality, no eviction) or **enforce** the config onto the node (`switch`
    /// it). Interactive on a TTY; `--adopt`/`--enforce` resolve non-interactively.
    Adopt {
        /// Non-interactively adopt the live state into config.
        #[arg(long, conflicts_with = "enforce")]
        adopt: bool,
        /// Non-interactively enforce the config onto the node (runs `switch`).
        #[arg(long, conflicts_with = "adopt")]
        enforce: bool,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
    },
}

/// Planning knobs shared by `dgx vllm up` and `dgx vllm config`.
#[derive(clap::Args, Debug)]
pub struct VllmPlanArgs {
    /// Override the OpenAI `served-model-name` (default: sanitized model name).
    #[arg(long)]
    pub served_name: Option<String>,
    /// Quant/dtype (auto|nvfp4|fp8|bf16|awq|gptq); inferred from the model name
    /// when omitted.
    #[arg(long)]
    pub dtype: Option<String>,
    /// `--tensor-parallel-size` (number of GPUs). Default 1; >1 is meaningless
    /// on a single unified-memory GB10.
    #[arg(long, default_value_t = 1)]
    pub tensor_parallel: u8,
    /// Cap the context window; otherwise the planner default is used.
    #[arg(long)]
    pub max_model_len: Option<u32>,
    /// `--gpu-memory-utilization` fraction (0.0..=1.0). Default 0.90.
    #[arg(long, default_value_t = 0.90)]
    pub gpu_mem_util: f64,
    /// Listen port on the node. Default 8000.
    #[arg(long, default_value_t = 8000)]
    pub port: u16,
    /// Render the `docker run vllm/vllm-openai` argv instead of native
    /// `vllm serve` (native is the only launcher executed in this step).
    #[arg(long)]
    pub docker: bool,
    /// Extra args appended verbatim to the vLLM command line.
    #[arg(long)]
    pub extra: Vec<String>,
}

/// `newt dgx vllm <cmd>` subcommands.
#[derive(Subcommand, Debug)]
pub enum VllmCmd {
    /// Launch a vLLM server on the node for `model` (HF id or local path).
    ///
    /// Runs a fit pre-flight against the node's *available* memory, renders the
    /// `vllm serve` argv, launches it detached with `nohup`, polls
    /// `/v1/models` until ready, then persists the endpoint + active model.
    Up {
        /// Model to serve (HuggingFace `<org>/<repo>` or an on-node path).
        model: String,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        #[command(flatten)]
        plan: VllmPlanArgs,
        /// Proceed even when the model exceeds the memory budget.
        #[arg(long)]
        force: bool,
        /// Before launching, unload any models resident on the active Ollama
        /// endpoint to free the shared unified-memory pool (the eval-loop swap).
        #[arg(long)]
        evict_ollama: bool,
        /// Print the resolved plan + remote script without SSHing.
        #[arg(long)]
        dry_run: bool,
    },
    /// Stop the vLLM server for `served_name` (default: the active model).
    Down {
        /// Served model name whose pidfile to kill (default: active model).
        served_name: Option<String>,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Print the kill command without SSHing.
        #[arg(long)]
        dry_run: bool,
    },
    /// Probe the configured vLLM endpoint and list its models (`/v1/models`).
    Ps,
    /// Print the resolved launch plan (argv) for `model`. Pure: no SSH/network.
    Config {
        /// Model to plan for.
        model: String,
        #[command(flatten)]
        plan: VllmPlanArgs,
    },
    /// Tail the vLLM server log on the node (`tail -f` over SSH).
    Logs {
        /// Served model name whose log to tail (default: active model).
        served_name: Option<String>,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Number of trailing lines before following. Default 50.
        #[arg(long, default_value_t = 50)]
        lines: u32,
        /// Print the remote command without SSHing.
        #[arg(long)]
        dry_run: bool,
    },
}

/// `newt dgx card <cmd>` — model cards (all settings to serve one model).
///
/// The pure render / validate / resolve logic lives in [`crate::dgx_card`]; this
/// is just the clap surface.
#[derive(Subcommand, Debug)]
pub enum CardCmd {
    /// List the built-in cards plus any drop-ins under `~/.newt/models`.
    List {
        /// Emit a JSON array instead of a table.
        #[arg(long)]
        json: bool,
    },
    /// Print one resolved card as canonical TOML (built-in < drop-in override).
    Show {
        /// Card name, e.g. `Ornith-1.0-35B` (see `card list`).
        name: String,
        /// Emit JSON instead of TOML.
        #[arg(long)]
        json: bool,
    },
    /// Validate a card file (`.toml` / `.yaml`) without installing it — checks
    /// the schema and refuses a card that leaks a host / LAN IP.
    Validate {
        /// Path to a `.toml` / `.yaml` card.
        path: std::path::PathBuf,
    },
    /// Stand a model's backend up from its card + point newt at it.
    ///
    /// vLLM: renders `vllm serve` with the card's parsers / context window /
    /// prefix-caching over SSH to the active DGX node, polls readiness, and
    /// activates the endpoint (reuses the `vllm up` machinery). ollama: NOT yet
    /// implemented — vLLM is the first fully-supported backend.
    Setup {
        /// Card name to set up (e.g. `Ornith-1.0-35B`; see `card list`).
        name: String,
        /// The checkpoint to serve (HF `<org>/<repo>` or an on-node path). The
        /// card carries the serving KNOBS, not WHICH checkpoint; defaults to the
        /// card's `served_name` when omitted.
        #[arg(long)]
        model: Option<String>,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Override the card's backend (`vllm` | `ollama`).
        #[arg(long)]
        backend: Option<String>,
        /// Render the plan + remote script without SSHing or writing config.
        #[arg(long)]
        dry_run: bool,
        /// Proceed even when the model exceeds the node memory budget.
        #[arg(long)]
        force: bool,
    },
    /// Interactively pick a card from the catalog, then stand it up.
    ///
    /// Prints the same catalog as `card list` as a numbered menu, reads a
    /// selection on the TTY, and hands the chosen card straight to the `setup`
    /// path — no new orchestration, just a discovery front-end. Non-interactive
    /// callers (no TTY / piped stdin) should use `card setup <name>` directly.
    Pick {
        /// The checkpoint to serve (HF `<org>/<repo>` or an on-node path);
        /// defaults to the chosen card's `served_name`.
        #[arg(long)]
        model: Option<String>,
        /// SSH node name (defaults to the active node).
        #[arg(long)]
        node: Option<String>,
        /// Override the card's backend (`vllm` | `ollama`).
        #[arg(long)]
        backend: Option<String>,
        /// Render the plan + remote script without SSHing or writing config.
        #[arg(long)]
        dry_run: bool,
        /// Proceed even when the model exceeds the node memory budget.
        #[arg(long)]
        force: bool,
    },
}

/// Dispatch a `newt dgx` subcommand.
pub async fn run(cmd: DgxCmd, config_path: Option<&Path>) -> anyhow::Result<()> {
    match cmd {
        DgxCmd::Setup {
            host,
            name,
            model,
            template,
            yes,
        } => setup(
            config_path,
            host.as_deref(),
            &name,
            model.as_deref(),
            template,
            yes,
        ),
        DgxCmd::Route { task } => route(&task, config_path),
        DgxCmd::Status { node, json } => status(config_path, node.as_deref(), json).await,
        DgxCmd::Deploy {
            strongest,
            model,
            node,
        } => deploy(config_path, strongest, model.as_deref(), node.as_deref()).await,
        DgxCmd::Clear {
            node,
            restart_ollama,
        } => clear(config_path, node.as_deref(), restart_ollama).await,
        DgxCmd::Models => models(config_path).await,
        DgxCmd::Doctor => doctor(config_path).await,
        DgxCmd::Use { model } => use_model(config_path, &model),
        DgxCmd::Warm {
            model,
            keep_alive,
            evict_vllm,
        } => warm(config_path, model, &keep_alive, evict_vllm).await,
        DgxCmd::Pull {
            model,
            node,
            name,
            force,
            dry_run,
        } => {
            pull(
                config_path,
                &model,
                node.as_deref(),
                name.as_deref(),
                force,
                dry_run,
            )
            .await
        }
        DgxCmd::Rm { model } => rm(config_path, &model).await,
        DgxCmd::Ps => ps(config_path).await,
        DgxCmd::Vllm { cmd } => match cmd {
            VllmCmd::Up {
                model,
                node,
                plan,
                force,
                evict_ollama,
                dry_run,
            } => {
                vllm_up(
                    config_path,
                    &model,
                    node.as_deref(),
                    &plan,
                    force,
                    evict_ollama,
                    dry_run,
                )
                .await
            }
            VllmCmd::Down {
                served_name,
                node,
                dry_run,
            } => {
                vllm_down(
                    config_path,
                    served_name.as_deref(),
                    node.as_deref(),
                    dry_run,
                )
                .await
            }
            VllmCmd::Ps => vllm_ps(config_path).await,
            VllmCmd::Config { model, plan } => vllm_config(&model, &plan),
            VllmCmd::Logs {
                served_name,
                node,
                lines,
                dry_run,
            } => {
                vllm_logs(
                    config_path,
                    served_name.as_deref(),
                    node.as_deref(),
                    lines,
                    dry_run,
                )
                .await
            }
        },
        DgxCmd::Card { cmd } => crate::dgx_card::run(cmd, config_path).await,
        DgxCmd::Gpu => gpu(config_path).await,
        DgxCmd::Switch {
            engine,
            model,
            node,
            dry_run,
        } => switch(config_path, &engine, &model, node.as_deref(), dry_run).await,
        DgxCmd::Adopt {
            adopt,
            enforce,
            node,
        } => adopt_cmd(config_path, adopt, enforce, node.as_deref()).await,
    }
}

// ---------------------------------------------------------------------------
// setup
// ---------------------------------------------------------------------------

/// First-run DGX configuration.
///
/// Synthesizes Ollama / vLLM endpoint URLs from a bare `host` (e.g.
/// `REDACTED-IP` or `REDACTED-HOST`) and writes the resulting `[dgx]`
/// block into the resolved config file. Loads the existing config first so
/// non-DGX fields are preserved.
///
/// Confirmation is skipped when `yes = true` (non-interactive / test mode).
fn setup(
    config_path: Option<&Path>,
    host: Option<&str>,
    name: &str,
    model: Option<&str>,
    template: bool,
    yes: bool,
) -> anyhow::Result<()> {
    // --template: dump home_template as TOML then exit without writing.
    if template {
        let tmpl = DgxConfig::home_template();
        let text = toml::to_string_pretty(&tmpl)
            .map_err(|e| anyhow::anyhow!("TOML serialisation failed: {e}"))?;
        println!("# [dgx] reference template");
        println!("# Copy into ~/.newt/config.toml under [dgx]\n");
        print!("{text}");
        return Ok(());
    }

    // No --host: print guidance and exit without error.
    let Some(host) = host else {
        eprintln!("Usage: newt dgx setup --host <hostname-or-ip> [--model <model>] [--yes]");
        eprintln!("       newt dgx setup --template");
        eprintln!("\nExamples:");
        eprintln!("  newt dgx setup --host REDACTED-IP --model qwen2.5-coder:32b --yes");
        eprintln!("  newt dgx setup --host REDACTED-HOST --name home");
        return Ok(());
    };

    // Build the DGX node from the bare host. Ollama default port is 11434;
    // vLLM default port is 8000. The LB and in-cluster URLs require distinct
    // hostnames so they cannot be synthesised from a bare host.
    let node = DgxNode {
        name: name.to_string(),
        ollama: Some(format!("http://{host}:11434")),
        vllm: Some(format!("http://{host}:8000")),
        ssh_host: Some(host.to_string()),
        ..Default::default()
    };
    let dgx = DgxConfig {
        active_node: Some(name.to_string()),
        active_endpoint: EndpointKind::Ollama,
        active_model: model.map(str::to_string),
        nodes: vec![node],
        formations: vec![],
    };

    let text = toml::to_string_pretty(&dgx)
        .map_err(|e| anyhow::anyhow!("TOML serialisation failed: {e}"))?;

    let save_path = config_path
        .map(std::path::PathBuf::from)
        .or_else(newt_core::Config::user_config_path)
        .ok_or_else(|| anyhow::anyhow!("cannot determine config file path (HOME unset?)"))?;

    println!("# [dgx] config to write to {}:\n", save_path.display());
    print!("{text}");

    if !yes {
        // Through the seal (#1909), like `ocap_cmd` and `dock_cmd`: the
        // capability, not a flag. Since #1908 a protocol-mode process cannot
        // obtain a speaking window at all, so `ask` errors here instead of
        // printing a question onto a JSON-RPC wire and blocking for an answer
        // that is never coming.
        let window = newt_core::tty::Terminal::suspend_for_prompt(
            newt_core::tty::TerminalTaker::PlainCliConfirm,
        );
        // EOF is not consent, and it is no longer this site's job to remember
        // that: `confirmed_on_terminal` refuses every outcome that is not an
        // affirmative answer. The `matches!` this replaces was fail-closed by
        // accident of falling through, which #1911 recorded.
        if !newt_core::interaction_terminal::confirmed_on_terminal(
            &window,
            &newt_core::interaction_form::confirm("Write?", "", "yes, write it", "no, stop"),
            false,
        ) {
            window.notice("Aborted.")?;
            return Ok(());
        }
    }

    // Preserve any non-DGX fields already in the config. If the target file
    // doesn't exist yet (first run), start from defaults rather than erroring.
    let mut config = match config_path {
        Some(p) if p.exists() => crate::migration_notices::read(|report| Config::load(p, report))
            .map_err(anyhow::Error::from)?,
        Some(_) => Config::default(),
        None => crate::migration_notices::read(|report| Config::resolve(report))
            .map_err(anyhow::Error::from)?,
    };
    config.dgx = Some(dgx);
    config.save(&save_path)?;
    println!("Saved → {}", save_path.display());
    Ok(())
}

// ---------------------------------------------------------------------------
// route
// ---------------------------------------------------------------------------

fn route(task: &str, config_path: Option<&Path>) -> anyhow::Result<()> {
    let config = load_config(config_path)?;
    let classification = Router::new().classify_detailed(task);
    let rec = recommend(config.dgx.as_ref(), &classification);

    println!(
        "  Complexity:  {}  (confidence {:.2})",
        tier_label(rec.tier),
        rec.confidence
    );
    match &rec.formation {
        Some(name) => println!("  Formation:   {name}"),
        None => println!("  Formation:   (none configured — set NEWT_DGX_HOST=<host>)"),
    }
    match &rec.model {
        Some(model) => println!("  Model:       {model}"),
        None => println!("  Model:       (unset — set [dgx].active_model or NEWT_DGX_MODEL)"),
    }
    println!("  Endpoint:    {}", rec.endpoint);
    println!("  Why:         {}", rec.why);
    Ok(())
}

/// A routing recommendation derived from a [`Classification`].
struct Recommendation {
    tier: Tier,
    confidence: f64,
    formation: Option<String>,
    model: Option<String>,
    endpoint: EndpointKind,
    why: String,
}

/// Recommend a formation for `c`: prefer a configured formation whose name
/// matches the tier convention, else fall back to the active model and a
/// tier-appropriate endpoint.
fn recommend(dgx: Option<&DgxConfig>, c: &Classification) -> Recommendation {
    let why = c
        .reasons
        .first()
        .cloned()
        .unwrap_or_else(|| "no routing signal".to_string());

    if let Some(formation) = dgx.and_then(|d| d.formation(preferred_formation(c.tier))) {
        return Recommendation {
            tier: c.tier,
            confidence: c.confidence,
            formation: Some(formation.name.clone()),
            model: Some(formation.model.clone()),
            endpoint: formation.endpoint,
            why,
        };
    }

    Recommendation {
        tier: c.tier,
        confidence: c.confidence,
        formation: None,
        model: dgx.and_then(|d| d.active_model.clone()),
        endpoint: default_endpoint_for(c.tier),
        why,
    }
}

/// Conventional formation name for a tier (matched against configured
/// formations in [`recommend`]).
fn preferred_formation(tier: Tier) -> &'static str {
    match tier {
        Tier::Review => "review",
        Tier::Complex => "coding",
        Tier::Standard => "standard",
        Tier::Fast => "fast",
    }
}

/// Advisory endpoint when no formation matches: heavier tiers prefer the
/// model-aware in-cluster proxy, lighter tiers the direct / LB endpoints.
fn default_endpoint_for(tier: Tier) -> EndpointKind {
    match tier {
        Tier::Review | Tier::Complex => EndpointKind::InCluster,
        Tier::Standard => EndpointKind::OllamaLb,
        Tier::Fast => EndpointKind::Ollama,
    }
}

fn tier_label(tier: Tier) -> &'static str {
    match tier {
        Tier::Fast => "fast",
        Tier::Standard => "standard",
        Tier::Complex => "complex",
        Tier::Review => "review",
    }
}

// ---------------------------------------------------------------------------
// status / models / doctor
// ---------------------------------------------------------------------------

async fn models(config_path: Option<&Path>) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    let kind = dgx.active_endpoint;
    let base = dgx.resolve_endpoint()?;
    println!("Models on {kind} endpoint ({base}):");

    let names = if kind.is_openai_compatible() {
        LocalVllmBackend::new(base.as_str(), "")
            .list_models()
            .await?
            .into_iter()
            .map(|m| m.id)
            .collect::<Vec<_>>()
    } else {
        fetch_ollama_models(&http_client(), &base).await?
    };

    if names.is_empty() {
        println!("  (none)");
    }
    for name in &names {
        println!("  {name}");
    }
    Ok(())
}

async fn status(config_path: Option<&Path>, node: Option<&str>, json: bool) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }

    // --json: emit ONLY the structured memory budget (issue #709 §1). The SSH
    // host is required for this form, so a missing host is a hard error here.
    if json {
        let host = dgx.ssh_host()?;
        let user = dgx.ssh_user();
        let budget = dgx_status::gather_budget(&RealSshCapture, &user, &host, None)?;
        println!(
            "{}",
            serde_json::to_string_pretty(&dgx_status::budget_json(&budget))?
        );
        return Ok(());
    }

    // Human form: endpoint health + loaded models (existing), then the SSH
    // memory budget. The budget is best-effort — never block the health view on
    // SSH reachability.
    let kind = dgx.active_endpoint;
    let base = dgx.resolve_endpoint()?;
    let client = http_client();
    let health_path = if kind.is_openai_compatible() {
        "/v1/models"
    } else {
        "/api/tags"
    };

    println!("DGX status — {kind} endpoint ({base})");
    println!("  Health:   {}", probe(&client, &base, health_path).await);

    if kind.is_openai_compatible() {
        println!("  Running:  (vLLM exposes no running-models endpoint)");
    } else {
        match fetch_ollama_running(&client, &base).await {
            Ok(names) if names.is_empty() => println!("  Running:  (no models loaded)"),
            Ok(names) => println!("  Running:  {}", names.join(", ")),
            Err(e) => println!("  Running:  (unavailable: {e})"),
        }
    }

    // Node memory budget over SSH (replaces the old nvidia-smi placeholder).
    match dgx.ssh_host() {
        Ok(host) => {
            let user = dgx.ssh_user();
            match dgx_status::gather_budget(&RealSshCapture, &user, &host, None) {
                Ok(budget) => print!("{}", dgx_status::render_budget_human(&host, &budget)),
                Err(e) => println!("  Memory:   (unavailable over SSH: {e})"),
            }
        }
        Err(_) => {
            println!("  Memory:   (no ssh_host configured — set [dgx].nodes[].ssh_host)");
        }
    }
    Ok(())
}

async fn doctor(config_path: Option<&Path>) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    let client = http_client();
    println!("newt dgx doctor — probing configured endpoints\n");

    let mut any = false;
    for kind in EndpointKind::ALL {
        let label = kind.as_str();
        match dgx.resolve_endpoint_for(kind) {
            Ok(base) => {
                any = true;
                let path = if kind.is_openai_compatible() {
                    "/v1/models"
                } else {
                    "/api/tags"
                };
                let health = probe(&client, &base, path).await;
                println!("  {label:<10} {base}  —  {health}");
            }
            Err(_) => println!("  {label:<10} (not set)"),
        }
    }

    if !any {
        println!("\n  No DGX endpoints configured. Set:");
        println!("    NEWT_DGX_HOST=<host>          (synthesizes ollama + vllm URLs from a bare hostname)");
        println!("    NEWT_DGX_OLLAMA_URL=<url>     (direct URL, e.g. https://REDACTED-HOST)");
    }
    println!("\n  DNS note: some mesh routers intercept certain local TLDs (e.g. .lan), so a");
    println!(
        "  custom DNS suffix may resolve from some hosts but not others; prefer an IP. Inside"
    );
    println!(
        "  k3s use the in_cluster proxy (http://ollama-proxy.inference.svc.cluster.local:11434)."
    );
    Ok(())
}

// ---------------------------------------------------------------------------
// helpers
// ---------------------------------------------------------------------------

/// Load the `[dgx]` sub-table, defaulting to an empty config so that
/// `NEWT_DGX_*` env overrides still work without a config file.
fn dgx_config(config_path: Option<&Path>) -> anyhow::Result<DgxConfig> {
    Ok(load_config(config_path)?.dgx.unwrap_or_default())
}

fn load_config(config_path: Option<&Path>) -> anyhow::Result<Config> {
    let config = match config_path {
        Some(p) => crate::migration_notices::read(|report| Config::load(p, report))?,
        None => crate::migration_notices::read(|report| Config::resolve(report))?,
    };
    Ok(config)
}

fn http_client() -> reqwest::Client {
    reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(5))
        .build()
        .expect("build reqwest client")
}

/// GET `{base}/api/tags` → installed Ollama model names.
async fn fetch_ollama_models(client: &reqwest::Client, base: &str) -> anyhow::Result<Vec<String>> {
    fetch_names(client, base, "/api/tags").await
}

/// GET `{base}/api/ps` → running Ollama model names. Routed through
/// `newt_core::backend_probe::fetch_ollama_ps` — the ONE `/api/ps` home
/// (#1312 discipline).
async fn fetch_ollama_running(client: &reqwest::Client, base: &str) -> anyhow::Result<Vec<String>> {
    Ok(
        newt_core::backend_probe::fetch_ollama_ps(client, base, None)
            .await?
            .into_iter()
            .map(|m| m.name)
            .collect(),
    )
}

async fn fetch_names(
    client: &reqwest::Client,
    base: &str,
    path: &str,
) -> anyhow::Result<Vec<String>> {
    let url = format!("{}{path}", base.trim_end_matches('/'));
    let resp = client
        .get(&url)
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HTTP {}", resp.status());
    }
    let json: serde_json::Value = resp.json().await?;
    Ok(extract_names(&json["models"]))
}

/// Pull `name` fields out of a JSON array (`[{ "name": "..." }, ...]`).
fn extract_names(models: &serde_json::Value) -> Vec<String> {
    models
        .as_array()
        .map(|arr| {
            arr.iter()
                .filter_map(|m| m["name"].as_str().map(str::to_string))
                .collect()
        })
        .unwrap_or_default()
}

/// Reachability probe: GET `{base}{path}`, summarized as a status string.
async fn probe(client: &reqwest::Client, base: &str, path: &str) -> String {
    let url = format!("{}{path}", base.trim_end_matches('/'));
    match client.get(&url).send().await {
        Ok(resp) if resp.status().is_success() => "OK".to_string(),
        Ok(resp) => format!("HTTP {}", resp.status()),
        Err(e) => format!("unreachable: {e}"),
    }
}

// ---------------------------------------------------------------------------
// use — persist active model to config
// ---------------------------------------------------------------------------

fn use_model(config_path: Option<&Path>, model: &str) -> anyhow::Result<()> {
    let mut config = load_config(config_path)?;

    // Update or create the [dgx] section with the chosen model.
    let dgx = config.dgx.get_or_insert_with(Default::default);
    dgx.active_model = Some(model.to_string());

    let save_path = config_path
        .map(std::path::PathBuf::from)
        .or_else(newt_core::Config::user_config_path)
        .ok_or_else(|| anyhow::anyhow!("cannot determine config file path"))?;

    config.save(&save_path)?;
    println!("Active model set to {model}");
    println!("Saved → {}", save_path.display());
    Ok(())
}

// ---------------------------------------------------------------------------
// warm
// ---------------------------------------------------------------------------

/// Ollama load-only request body: a `/api/generate` call with no `prompt`
/// and a `keep_alive` window loads the model into VRAM (and pins it resident)
/// without generating any tokens.
fn warm_body(model: &str, keep_alive: &str) -> serde_json::Value {
    serde_json::json!({ "model": model, "keep_alive": keep_alive, "stream": false })
}

/// POST the load-only request and return the load time in seconds when Ollama
/// actually had to load the model (absent / `None` when it was already
/// resident — a warm hit returns near-instantly with no `load_duration`).
async fn warm_model(
    client: &reqwest::Client,
    base: &str,
    model: &str,
    keep_alive: &str,
) -> anyhow::Result<Option<f64>> {
    let url = format!("{}/api/generate", base.trim_end_matches('/'));
    let resp = client
        .post(&url)
        .json(&warm_body(model, keep_alive))
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HTTP {}", resp.status());
    }
    let json: serde_json::Value = resp.json().await?;
    Ok(json["load_duration"].as_u64().map(|ns| ns as f64 / 1e9))
}

/// Resolve which model `warm` should load. An explicit arg always wins. With no
/// arg, `--evict-vllm` requires an explicit Ollama model — the persisted active
/// model is the just-stopped vLLM served name, which Ollama doesn't have (it
/// would 404), so refuse with an actionable message rather than that confusion.
/// Otherwise fall back to the active model. Pure → unit-tested.
fn resolve_warm_model(
    model: Option<String>,
    evict_vllm: bool,
    dgx: &DgxConfig,
) -> anyhow::Result<String> {
    resolve_warm_model_with(model, evict_vllm, dgx, &|k| std::env::var(k).ok())
}

/// [`resolve_warm_model`] with injectable env so tests need not mutate the
/// process-global environment (which is unsound under parallel test threads).
fn resolve_warm_model_with(
    model: Option<String>,
    evict_vllm: bool,
    dgx: &DgxConfig,
    get_env: &dyn Fn(&str) -> Option<String>,
) -> anyhow::Result<String> {
    if let Some(m) = model {
        return Ok(m);
    }
    if evict_vllm {
        anyhow::bail!(
            "--evict-vllm needs an explicit Ollama model to warm (the active model {:?} \
             is the vLLM served name, now stopped) — e.g. \
             `dgx warm --evict-vllm qwen2.5-coder:32b`",
            dgx.active_model.as_deref().unwrap_or("<none>")
        );
    }
    Ok(dgx.resolve_active_model_with(get_env)?)
}

async fn warm(
    config_path: Option<&Path>,
    model: Option<String>,
    keep_alive: &str,
    evict_vllm: bool,
) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    // `--evict-vllm`: free the shared pool first, then target the Ollama
    // endpoint explicitly — the flag is an explicit "switch to Ollama" intent,
    // so it bypasses the active-endpoint guard below.
    let (kind, base) = if evict_vllm {
        evict_vllm_server(&RealSsh, &dgx).await?;
        (
            EndpointKind::Ollama,
            dgx.resolve_endpoint_for(EndpointKind::Ollama)?,
        )
    } else {
        let kind = dgx.active_endpoint;
        if kind.is_openai_compatible() {
            anyhow::bail!(
                "`newt dgx warm` targets Ollama endpoints; the active endpoint is vLLM \
                 (vLLM keeps its served model resident already). Pass --evict-vllm to stop \
                 the vLLM server and warm an Ollama model instead."
            );
        }
        (kind, dgx.resolve_endpoint()?)
    };
    let model = resolve_warm_model(model, evict_vllm, &dgx)?;
    println!("Warming {model} on {kind} endpoint ({base}) — keep_alive={keep_alive}");

    // Cold loads of large models can take tens of seconds; give warm its own
    // generous timeout rather than the short probe client.
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()
        .expect("build reqwest client");
    match warm_model(&client, &base, &model, keep_alive).await? {
        Some(secs) => println!("  loaded in {secs:.1}s — now resident"),
        None => println!("  ready (already resident)"),
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// pull / rm / ps
// ---------------------------------------------------------------------------

use crate::dgx_pull::{self, fit_check, FitVerdict, GgufFile, ModelRef, PullPlan};

/// GET `<hf_base>/api/models/<org>/<repo>?blobs=true` → the `.gguf` siblings.
async fn fetch_hf_siblings(
    client: &reqwest::Client,
    hf_base: &str,
    org: &str,
    repo: &str,
) -> anyhow::Result<Vec<GgufFile>> {
    let url = format!(
        "{}/api/models/{org}/{repo}?blobs=true",
        hf_base.trim_end_matches('/')
    );
    let resp = client
        .get(&url)
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("HF API request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HF API returned HTTP {}", resp.status());
    }
    let json: serde_json::Value = resp.json().await?;
    Ok(dgx_pull::parse_gguf_siblings(&json))
}

/// Top-level `pull` handler: resolve config, dispatch on the arg shape.
async fn pull(
    config_path: Option<&Path>,
    model: &str,
    node: Option<&str>,
    name: Option<&str>,
    force: bool,
    dry_run: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(node) = node {
        dgx.active_node = Some(node.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;
    let has_token = std::env::var("HF_TOKEN").is_ok_and(|t| !t.trim().is_empty());

    match ModelRef::parse(model) {
        ModelRef::Ollama { name: tag } => {
            // A plain Ollama name needs no HF metadata or fit pre-flight.
            execute_hf_plan(
                &RealSsh,
                &user,
                &host,
                &PullPlan::OllamaNative { tag },
                has_token,
                dry_run,
            )
        }
        ModelRef::Hf { org, repo, quant } => {
            let client = http_client_long();
            let all = fetch_hf_siblings(&client, &hf_api_base(), &org, &repo).await?;
            let matched: Vec<GgufFile> = all
                .into_iter()
                .filter(|f| dgx_pull::file_matches_quant(&f.path, &quant))
                .collect();
            let plan = dgx_pull::plan_hf(&org, &repo, &quant, &matched, name)
                .map_err(|e| anyhow::anyhow!(e))?;

            // Fit pre-flight (the GLM-5.2 lesson).
            let model_bytes = dgx_pull::total_bytes(&matched);
            let mem = if dry_run {
                None
            } else {
                detect_node_mem(&user, &host, None)
            };
            report_fit(fit_check(model_bytes, mem), force)?;

            execute_hf_plan(&RealSsh, &user, &host, &plan, has_token, dry_run)
        }
    }
}

/// HF API base URL (overridable via `NEWT_HF_API_BASE` for tests).
fn hf_api_base() -> String {
    std::env::var("NEWT_HF_API_BASE").unwrap_or_else(|_| "https://huggingface.co".to_string())
}

/// A reqwest client with a generous timeout (HF downloads metadata can be slow).
fn http_client_long() -> reqwest::Client {
    reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(60))
        .build()
        .expect("build reqwest client")
}

/// Print the fit verdict and either proceed or refuse (unless `--force`).
fn report_fit(verdict: FitVerdict, force: bool) -> anyhow::Result<()> {
    let refuse = verdict.should_refuse();
    match &verdict {
        FitVerdict::Fits {
            model_bytes,
            mem_bytes,
        } => println!(
            "  Fit: model {:.1} GB fits in node memory {:.1} GB",
            dgx_pull::bytes_to_gib(*model_bytes),
            dgx_pull::bytes_to_gib(*mem_bytes)
        ),
        FitVerdict::Undetectable { model_bytes } => eprintln!(
            "  WARNING: could not detect node memory; model is {:.1} GB. Proceeding best-effort.",
            dgx_pull::bytes_to_gib(*model_bytes)
        ),
        FitVerdict::Exceeds {
            model_bytes,
            mem_bytes,
        } => eprintln!(
            "  WARNING: model {:.1} GB exceeds node memory {:.1} GB — \
             will be unrunnable / heavy disk paging.",
            dgx_pull::bytes_to_gib(*model_bytes),
            dgx_pull::bytes_to_gib(*mem_bytes)
        ),
    }
    if refuse && !force {
        anyhow::bail!("refusing to pull a model larger than node memory; pass --force to override");
    }
    if refuse {
        eprintln!("  --force given: proceeding anyway.");
    }
    Ok(())
}

/// Execute (or dry-run) an HF [`PullPlan`].
fn execute_hf_plan(
    ssh: &dyn SshExec,
    user: &str,
    host: &str,
    plan: &PullPlan,
    has_token: bool,
    dry_run: bool,
) -> anyhow::Result<()> {
    match plan {
        PullPlan::OllamaNative { tag } => {
            let command = dgx_pull::ollama_native_remote_command(tag);
            run_or_dryrun(
                ssh,
                user,
                host,
                None,
                &command,
                dry_run,
                &format!("PullPlan: OllamaNative {{ tag: {tag:?} }}"),
            )
        }
        PullPlan::SingleFileHf { org, repo, quant } => {
            let command = dgx_pull::single_file_remote_command(org, repo, quant);
            run_or_dryrun(
                ssh,
                user,
                host,
                None,
                &command,
                dry_run,
                "PullPlan: SingleFileHf (ollama pull hf.co/...)",
            )
        }
        PullPlan::ShardedHf {
            org,
            repo,
            quant,
            parts,
            modelfile,
            name,
        } => {
            let script =
                dgx_pull::sharded_remote_script(org, repo, parts, modelfile, name, has_token);
            let summary = format!(
                "PullPlan: ShardedHf {{ quant: {quant:?}, parts: {}, name: {name:?} }}",
                parts.len()
            );
            run_or_dryrun(ssh, user, host, None, &script, dry_run, &summary)
        }
    }
}

/// Either print the plan + exact remote command (dry-run) or SSH-execute it.
fn run_or_dryrun(
    ssh: &dyn SshExec,
    user: &str,
    host: &str,
    port: Option<u16>,
    command: &str,
    dry_run: bool,
    summary: &str,
) -> anyhow::Result<()> {
    if dry_run {
        println!("{summary}");
        let argv = dgx_pull::ssh_argv(user, host, port, command);
        println!("Would run: {}", argv.join(" "));
        println!("--- remote command ---");
        println!("{command}");
        return Ok(());
    }
    println!("{summary}");
    eprintln!("→ ssh {user}@{host}: executing pull (output below)");
    ssh.run(user, host, port, command)
}

/// `rm <model>` — DELETE `/api/delete` on the active Ollama endpoint.
async fn rm(config_path: Option<&Path>, model: &str) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    if dgx.active_endpoint.is_openai_compatible() {
        anyhow::bail!("`newt dgx rm` targets Ollama endpoints; the active endpoint is vLLM");
    }
    let base = dgx.resolve_endpoint()?;
    delete_ollama_model(&http_client(), &base, model).await?;
    println!("Deleted {model}");
    Ok(())
}

/// DELETE `{base}/api/delete` with a `{ "model": <name> }` body.
async fn delete_ollama_model(
    client: &reqwest::Client,
    base: &str,
    model: &str,
) -> anyhow::Result<()> {
    let url = format!("{}/api/delete", base.trim_end_matches('/'));
    let resp = client
        .delete(&url)
        .json(&serde_json::json!({ "model": model }))
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HTTP {}", resp.status());
    }
    Ok(())
}

/// `ps` — GET `/api/ps`, print each loaded model with its size.
async fn ps(config_path: Option<&Path>) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    if dgx.active_endpoint.is_openai_compatible() {
        anyhow::bail!("`newt dgx ps` targets Ollama endpoints; the active endpoint is vLLM");
    }
    let base = dgx.resolve_endpoint()?;
    let loaded = fetch_ollama_ps(&http_client(), &base).await?;
    println!("Loaded models on {} ({base}):", dgx.active_endpoint);
    if loaded.is_empty() {
        println!("  (none)");
    }
    for (name, size) in &loaded {
        match size {
            Some(bytes) => println!("  {name}  ({:.1} GB)", dgx_pull::bytes_to_gib(*bytes)),
            None => println!("  {name}"),
        }
    }
    Ok(())
}

/// GET `{base}/api/ps` → `(name, size_bytes)` for each loaded model.
/// Routed through `newt_core::backend_probe::fetch_ollama_ps` — the ONE
/// `/api/ps` home (#1312 discipline) — projected onto the CLI table shape.
async fn fetch_ollama_ps(
    client: &reqwest::Client,
    base: &str,
) -> anyhow::Result<Vec<(String, Option<u64>)>> {
    Ok(
        newt_core::backend_probe::fetch_ollama_ps(client, base, None)
            .await?
            .into_iter()
            .map(|m| (m.name, m.size_bytes))
            .collect(),
    )
}

// ---------------------------------------------------------------------------
// vllm — stand up / tear down a vLLM server on the node (Step 14.11)
// ---------------------------------------------------------------------------

/// GET `<hf_base>/api/models/<org>/<repo>?blobs=true` → raw JSON (for sizing
/// safetensors weights; same endpoint as the GGUF sibling fetch).
async fn fetch_hf_model_json(
    client: &reqwest::Client,
    hf_base: &str,
    org: &str,
    repo: &str,
) -> anyhow::Result<serde_json::Value> {
    let url = format!(
        "{}/api/models/{org}/{repo}?blobs=true",
        hf_base.trim_end_matches('/')
    );
    let resp = client
        .get(&url)
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("HF API request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HF API returned HTTP {}", resp.status());
    }
    Ok(resp.json().await?)
}

/// Best-effort vLLM weight size: sum the repo's weight files (`.safetensors` /
/// `.bin`) when `model` is an HF id; `None` for a local path, a fetch/parse
/// failure, or a repo with no sized weights.
async fn fetch_vllm_weight_bytes(model: &str) -> Option<u64> {
    let (org, repo) = dgx_vllm::hf_repo_parts(model)?;
    let client = http_client_long();
    let json = fetch_hf_model_json(&client, &hf_api_base(), &org, &repo)
        .await
        .ok()?;
    let bytes = dgx_vllm::sum_weight_bytes(&json);
    (bytes > 0).then_some(bytes)
}

/// Docker execution isn't wired yet (the remote script wraps native
/// `vllm serve` only), so `up --docker` must refuse clearly rather than silently
/// launch native. `dgx vllm config --docker` still previews the docker argv.
fn ensure_executable_runtime(runtime: dgx_vllm::VllmRuntime) -> anyhow::Result<()> {
    if matches!(runtime, dgx_vllm::VllmRuntime::Docker) {
        anyhow::bail!(
            "--docker is preview-only here: `dgx vllm config --docker` prints the docker \
             argv, but native `vllm serve` is the only launcher `up` executes in this step. \
             Omit --docker to launch (docker execution is a follow-up)."
        );
    }
    Ok(())
}

/// Build a resolved [`dgx_vllm::VllmPlan`] from CLI args (shared by `up`/`config`).
fn build_plan_from_args(model: &str, a: &VllmPlanArgs) -> anyhow::Result<dgx_vllm::VllmPlan> {
    let dtype = match a.dtype.as_deref() {
        Some(s) => Some(dgx_vllm::Dtype::parse(s).ok_or_else(|| {
            anyhow::anyhow!(
                "unknown --dtype {s:?}; expected one of: auto, nvfp4, fp8, bf16, awq, gptq"
            )
        })?),
        None => None,
    };
    let runtime = if a.docker {
        dgx_vllm::VllmRuntime::Docker
    } else {
        dgx_vllm::VllmRuntime::Native
    };
    Ok(dgx_vllm::resolve_plan(dgx_vllm::PlanInputs {
        model,
        served_name: a.served_name.as_deref(),
        dtype,
        tensor_parallel: a.tensor_parallel,
        max_model_len: a.max_model_len,
        gpu_mem_util: a.gpu_mem_util,
        port: a.port,
        runtime,
        extra: a.extra.clone(),
    }))
}

/// Execute (or dry-run) a vLLM launch over SSH. Injection seam: `ssh` is
/// `&RealSsh` in prod, `&RecordingSsh` in tests.
fn execute_vllm_plan(
    ssh: &dyn SshExec,
    user: &str,
    host: &str,
    plan: &dgx_vllm::VllmPlan,
    dry_run: bool,
) -> anyhow::Result<()> {
    let script = dgx_vllm::vllm_remote_script(plan);
    let summary = format!(
        "VllmPlan: serve {:?} as {:?} ({:?}, tp={}, max_len={}, util={:.2}, port={})",
        plan.model,
        plan.served_name,
        plan.dtype,
        plan.tensor_parallel,
        plan.max_model_len,
        plan.gpu_mem_util(),
        plan.port,
    );
    run_or_dryrun(ssh, user, host, None, &script, dry_run, &summary)
}

/// Poll the freshly-launched server's `/v1/models` with bounded backoff. Cold
/// model loads can take minutes, hence the generous `for_local_inference`
/// policy in production; tests inject `RetryPolicy::immediate`.
async fn poll_vllm_ready(endpoint: &str, policy: &RetryPolicy) -> anyhow::Result<()> {
    let backend = LocalVllmBackend::new(endpoint, "");
    with_backoff(policy, || async { backend.list_models().await.map(|_| ()) })
        .await
        .map(|_| ())
        .map_err(|e| anyhow::anyhow!("vLLM did not become ready at {endpoint}: {e}"))
}

/// Pure in-memory mutation for persist-on-`up`: point the active endpoint at the
/// freshly-launched vLLM server and write its URL onto the target node. No IO —
/// the disk write lives in [`persist_vllm_endpoint`].
///
/// Returns `true` when the URL was written onto a matching node. `false` means
/// no node matched (the target name isn't in `nodes[]`) — the active endpoint is
/// still flipped to vLLM, but the URL wasn't recorded, so the caller warns
/// rather than leaving the user with a silently incomplete config.
#[must_use]
fn apply_vllm_persist(
    config: &mut Config,
    node_name: Option<&str>,
    endpoint_url: &str,
    served_name: &str,
) -> bool {
    let dgx = config.dgx.get_or_insert_with(Default::default);
    dgx.active_endpoint = EndpointKind::Vllm;
    dgx.active_model = Some(served_name.to_string());
    let target = node_name
        .map(str::to_string)
        .or_else(|| dgx.active_node.clone());
    if let Some(name) = target {
        if let Some(node) = dgx.nodes.iter_mut().find(|n| n.name == name) {
            node.vllm = Some(endpoint_url.to_string());
            return true;
        }
    }
    false
}

/// Persist the vLLM endpoint to config (load → [`apply_vllm_persist`] → save).
fn persist_vllm_endpoint(
    config_path: Option<&Path>,
    node_name: Option<&str>,
    endpoint_url: &str,
    served_name: &str,
) -> anyhow::Result<()> {
    let mut config = load_config(config_path)?;
    let node_recorded = apply_vllm_persist(&mut config, node_name, endpoint_url, served_name);
    let save_path = config_path
        .map(std::path::PathBuf::from)
        .or_else(newt_core::Config::user_config_path)
        .ok_or_else(|| anyhow::anyhow!("cannot determine config file path"))?;
    config.save(&save_path)?;
    println!("vLLM endpoint {endpoint_url} active; model {served_name}");
    println!("Saved → {}", save_path.display());
    if !node_recorded {
        eprintln!(
            "  WARNING: active endpoint set to vLLM, but no matching node was found to \
             record {endpoint_url} on — configure the node (`dgx setup` / `dgx node`) so \
             the URL resolves without env fallback."
        );
    }
    Ok(())
}

/// The served name to act on: an explicit arg, else the active model.
fn resolve_served_name(dgx: &DgxConfig, served_name: Option<&str>) -> anyhow::Result<String> {
    match served_name {
        Some(s) => Ok(s.to_string()),
        None => Ok(dgx.resolve_active_model()?),
    }
}

/// `dgx vllm up <model>` — fit pre-flight, launch, wait for readiness, persist.
///
/// `pub(crate)` so `card setup` (dgx_card.rs) reuses the same stand-up after
/// mapping a card onto [`VllmPlanArgs`].
pub(crate) async fn vllm_up(
    config_path: Option<&Path>,
    model: &str,
    node: Option<&str>,
    plan_args: &VllmPlanArgs,
    force: bool,
    evict_ollama: bool,
    dry_run: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;

    let plan = build_plan_from_args(model, plan_args)?;
    ensure_executable_runtime(plan.runtime)?;

    // Fit pre-flight (the GLM-5.2 lesson, ported). Skipped on dry-run — no
    // network: dry-run only shows the plan + remote script.
    if !dry_run {
        // The eval-loop swap: free the shared pool first so the fit probe below
        // sees the reclaimed memory.
        if evict_ollama {
            evict_ollama_models(&dgx).await?;
        }
        let mem = detect_node_mem_available(&user, &host, None);
        match fetch_vllm_weight_bytes(model).await {
            Some(weight) => report_fit(
                dgx_vllm::vllm_fit_check(weight, mem, plan_args.gpu_mem_util),
                force,
            )?,
            None => eprintln!(
                "  WARNING: could not size weights for {model:?} (local path, HF lookup \
                 failed, or no sized .safetensors/.bin) — skipping the fit pre-flight; the \
                 model may exceed node memory."
            ),
        }
    }

    execute_vllm_plan(&RealSsh, &user, &host, &plan, dry_run)?;

    if !dry_run {
        let endpoint = format!("http://{host}:{}", plan.port);
        println!("Waiting for vLLM at {endpoint}/v1/models (cold load can take minutes) …");
        poll_vllm_ready(&endpoint, &RetryPolicy::for_local_inference()).await?;
        persist_vllm_endpoint(config_path, node, &endpoint, &plan.served_name)?;
    }
    Ok(())
}

/// `dgx vllm down [served_name]` — kill the recorded server PID on the node.
async fn vllm_down(
    config_path: Option<&Path>,
    served_name: Option<&str>,
    node: Option<&str>,
    dry_run: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;
    let served = resolve_served_name(&dgx, served_name)?;
    let command = dgx_vllm::vllm_stop_command(&served);
    run_or_dryrun(
        &RealSsh,
        &user,
        &host,
        None,
        &command,
        dry_run,
        &format!("vLLM down: {served:?}"),
    )
}

/// `dgx vllm logs [served_name]` — tail and follow the server log on the node.
async fn vllm_logs(
    config_path: Option<&Path>,
    served_name: Option<&str>,
    node: Option<&str>,
    lines: u32,
    dry_run: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;
    let served = resolve_served_name(&dgx, served_name)?;
    let command = dgx_vllm::vllm_logs_command(&served, lines);
    run_or_dryrun(
        &RealSsh,
        &user,
        &host,
        None,
        &command,
        dry_run,
        &format!("vLLM logs: {served:?} (tail -f)"),
    )
}

/// `dgx vllm config <model>` — print the resolved launch argv. Pure (no SSH).
fn vllm_config(model: &str, plan_args: &VllmPlanArgs) -> anyhow::Result<()> {
    let plan = build_plan_from_args(model, plan_args)?;
    let argv = if matches!(plan.runtime, dgx_vllm::VllmRuntime::Docker) {
        dgx_vllm::vllm_docker_argv(&plan)
    } else {
        dgx_vllm::render_vllm_argv(&plan)
    };
    println!("Resolved vLLM launch plan for {model:?}:");
    println!("  served-model-name: {}", plan.served_name);
    println!(
        "  dtype={:?}  tensor-parallel={}  max-model-len={}  gpu-mem-util={:.2}  port={}",
        plan.dtype,
        plan.tensor_parallel,
        plan.max_model_len,
        plan.gpu_mem_util(),
        plan.port,
    );
    println!("  argv: {}", argv.join(" "));
    Ok(())
}

/// `dgx vllm ps` — GET the configured vLLM endpoint's `/v1/models`.
async fn vllm_ps(config_path: Option<&Path>) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    let base = dgx.resolve_endpoint_for(EndpointKind::Vllm)?;
    let models = fetch_vllm_models(&base).await?;
    println!("vLLM models on {base}:");
    if models.is_empty() {
        println!("  (none)");
    }
    for m in &models {
        println!("  {m}");
    }
    Ok(())
}

/// GET `<base>/v1/models` → served model ids (OpenAI-compatible). `base` is the
/// injection seam: tests pass a `wiremock` `MockServer::uri()`.
async fn fetch_vllm_models(base: &str) -> anyhow::Result<Vec<String>> {
    let backend = LocalVllmBackend::new(base, "");
    Ok(backend
        .list_models()
        .await?
        .into_iter()
        .map(|m| m.id)
        .collect())
}

// ---------------------------------------------------------------------------
// gpu — cross-engine residency view + eviction (Step 14.12)
// ---------------------------------------------------------------------------

/// A ground-truth snapshot of what each engine holds on the node, plus the
/// headroom. Assembled from live probes (`/api/ps`, `/v1/models`,
/// `MemAvailable`) rather than a cached lease file.
struct Residency {
    /// Ollama resident models: `(name, size_bytes)` from `/api/ps`.
    ollama: Vec<(String, Option<u64>)>,
    /// vLLM served model ids from `/v1/models`.
    vllm: Vec<String>,
    /// `MemAvailable` in bytes, or `None` when the probe failed.
    mem_available: Option<u64>,
}

impl Residency {
    /// Both engines hold something → they're sharing the one unified pool.
    fn is_contended(&self) -> bool {
        !self.ollama.is_empty() && !self.vllm.is_empty()
    }
}

/// The Ollama models to unload to free the GPU (every currently-resident one).
fn ollama_evict_targets(res: &Residency) -> Vec<String> {
    res.ollama.iter().map(|(name, _)| name.clone()).collect()
}

/// Ollama unload request body: `keep_alive: 0` unloads the model immediately
/// (the inverse of `warm`'s positive keep-alive). Pure.
fn ollama_unload_body(model: &str) -> serde_json::Value {
    serde_json::json!({ "model": model, "keep_alive": 0 })
}

/// Render the residency snapshot for `dgx gpu`. Pure → unit-tested directly.
fn render_residency(res: &Residency) -> String {
    let mut s = String::new();
    match res.mem_available {
        // bytes_to_gib divides by 1024^3 (gibibytes); the codebase labels these
        // "GB" throughout (fit verdict, `ps`), so match that convention here.
        Some(b) => s.push_str(&format!(
            "  MemAvailable: {:.1} GB\n",
            dgx_pull::bytes_to_gib(b)
        )),
        None => s.push_str("  MemAvailable: (undetected)\n"),
    }
    s.push_str("  Ollama resident:\n");
    if res.ollama.is_empty() {
        s.push_str("    (none)\n");
    }
    for (name, size) in &res.ollama {
        match size {
            Some(b) => s.push_str(&format!(
                "    {name}  ({:.1} GB)\n",
                dgx_pull::bytes_to_gib(*b)
            )),
            None => s.push_str(&format!("    {name}\n")),
        }
    }
    s.push_str("  vLLM served:\n");
    if res.vllm.is_empty() {
        s.push_str("    (none)\n");
    }
    for m in &res.vllm {
        s.push_str(&format!("    {m}\n"));
    }
    if res.is_contended() {
        s.push_str(
            "  ⚠ both engines are resident and share one unified pool — \
             use `dgx vllm up --evict-ollama` to free it before a large vLLM serve.\n",
        );
    }
    s
}

/// `dgx gpu` — print the cross-engine residency snapshot. Each side is
/// best-effort: an unconfigured/unreachable engine simply shows nothing.
async fn gpu(config_path: Option<&Path>) -> anyhow::Result<()> {
    let dgx = dgx_config(config_path)?;
    let client = http_client();
    let ollama = match dgx.resolve_endpoint_for(EndpointKind::Ollama) {
        Ok(base) => fetch_ollama_ps(&client, &base).await.unwrap_or_default(),
        Err(_) => Vec::new(),
    };
    let vllm = match dgx.resolve_endpoint_for(EndpointKind::Vllm) {
        Ok(base) => fetch_vllm_models(&base).await.unwrap_or_default(),
        Err(_) => Vec::new(),
    };
    let mem_available = match dgx.ssh_host() {
        Ok(host) => detect_node_mem_available(&dgx.ssh_user(), &host, None),
        Err(_) => None,
    };
    let res = Residency {
        ollama,
        vllm,
        mem_available,
    };
    println!("GPU residency:");
    print!("{}", render_residency(&res));
    Ok(())
}

/// POST an `/api/generate` unload (`keep_alive: 0`) to the Ollama endpoint.
async fn unload_ollama_model(
    client: &reqwest::Client,
    base: &str,
    model: &str,
) -> anyhow::Result<()> {
    let url = format!("{}/api/generate", base.trim_end_matches('/'));
    let resp = client
        .post(&url)
        .json(&ollama_unload_body(model))
        .send()
        .await
        .map_err(|e| anyhow::anyhow!("request failed: {e}"))?;
    if !resp.status().is_success() {
        anyhow::bail!("HTTP {}", resp.status());
    }
    Ok(())
}

/// Unload every model resident on the active Ollama endpoint (the
/// `--evict-ollama` swap). Best-effort: a missing endpoint or a single failed
/// unload warns rather than aborting the launch.
async fn evict_ollama_models(dgx: &DgxConfig) -> anyhow::Result<()> {
    let base = match dgx.resolve_endpoint_for(EndpointKind::Ollama) {
        Ok(b) => b,
        Err(_) => {
            eprintln!("  --evict-ollama: no Ollama endpoint configured; nothing to evict");
            return Ok(());
        }
    };
    let client = http_client();
    let resident = fetch_ollama_ps(&client, &base).await.unwrap_or_default();
    let targets = ollama_evict_targets(&Residency {
        ollama: resident,
        vllm: Vec::new(),
        mem_available: None,
    });
    if targets.is_empty() {
        println!("  --evict-ollama: no resident Ollama models to free");
        return Ok(());
    }
    for m in &targets {
        match unload_ollama_model(&client, &base, m).await {
            Ok(()) => println!("  evicted Ollama model {m}"),
            Err(e) => eprintln!("  WARNING: failed to evict Ollama model {m}: {e}"),
        }
    }
    Ok(())
}

/// Stop the running vLLM server(s) to free the unified pool (the `--evict-vllm`
/// swap, the mirror of [`evict_ollama_models`]). Queries `/v1/models` for the
/// served id(s) and kills each via the pidfile-based stop command over SSH —
/// vLLM has no HTTP unload, so the server *is* the residency. Best-effort: a
/// missing endpoint, no running server, or an unresolvable SSH host is a no-op,
/// not an error. `ssh` is injectable so tests never touch a real node.
async fn evict_vllm_server(ssh: &dyn SshExec, dgx: &DgxConfig) -> anyhow::Result<()> {
    let base = match dgx.resolve_endpoint_for(EndpointKind::Vllm) {
        Ok(b) => b,
        Err(_) => {
            eprintln!("  --evict-vllm: no vLLM endpoint configured; nothing to evict");
            return Ok(());
        }
    };
    let served = fetch_vllm_models(&base).await.unwrap_or_default();
    if served.is_empty() {
        println!("  --evict-vllm: no vLLM server running");
        return Ok(());
    }
    let user = dgx.ssh_user();
    let host = match dgx.ssh_host() {
        Ok(h) => h,
        Err(e) => {
            eprintln!("  --evict-vllm: cannot resolve SSH host ({e}); skipping eviction");
            return Ok(());
        }
    };
    for id in &served {
        let cmd = dgx_vllm::vllm_stop_command(id);
        match ssh.run(&user, &host, None, &cmd) {
            Ok(()) => println!("  stopped vLLM server {id}"),
            Err(e) => eprintln!("  WARNING: failed to stop vLLM server {id}: {e}"),
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// deploy / clear — hardware-aware inference deployment (issue #709, PR2)
// ---------------------------------------------------------------------------

/// The OpenAI-compatible serve port a deployed variant listens on, by tool: the
/// vLLM / llama.cpp convention is 8000, Ollama's is 11434. The convention deploy
/// scripts follow this (e.g. `vllm serve --port 8000`). The health gate polls
/// `<host>:<port>/v1/models` — which Ollama also exposes — so one port rule
/// covers both. Pure.
fn serve_port(tool: InferenceTool) -> u16 {
    match tool {
        InferenceTool::Ollama => 11434,
        InferenceTool::Vllm | InferenceTool::LlamaCpp => 8000,
    }
}

/// Resolve which variant `deploy` should launch (issue #709 §6 / §4). With
/// `strongest`, the strongest registry model fitting `available_gib` on a single
/// node; otherwise the explicit `model` slug looked up in the registry.
/// `available_gib` is ignored in the explicit path (the caller skips the budget
/// probe then). Pure — the budget is gathered by the caller.
fn select_deploy_variant(
    strongest: bool,
    model: Option<&str>,
    available_gib: f64,
) -> anyhow::Result<&'static ModelVariant> {
    if strongest {
        return dgx_registry::strongest_model(available_gib, 1).ok_or_else(|| {
            anyhow::anyhow!(
                "no registry model fits the available budget ({available_gib:.1} GiB, before \
                 the 15% headroom) — free memory (`newt dgx clear`) or see `newt dgx status`"
            )
        });
    }
    let slug = model.ok_or_else(|| {
        anyhow::anyhow!("specify a model slug (e.g. ornith-35b-fp16) or pass --strongest")
    })?;
    dgx_registry::find_variant(slug).ok_or_else(|| {
        anyhow::anyhow!(
            "unknown model {slug:?}; expected a registry slug like ornith-35b-fp16 \
             (the `~/ornith-{{size}}-{{format}}.sh` convention)"
        )
    })
}

/// How `deploy` launches a selected variant (issue #709 §4), decided purely from
/// whether the variant's convention script is present on the node.
#[derive(Debug, PartialEq, Eq)]
enum DeployRoute {
    /// Mode A — invoke the operator's pre-written convention script.
    Script {
        /// The `~/ornith-{size}-{format}.sh` path (tilde expands on the node).
        path: String,
        /// The shell-safe convention slug (used for the log/pid file names).
        slug: String,
        /// Port the launched server is expected to serve on.
        port: u16,
    },
    /// Mode B — no convention script: serve the model id via generated `vllm`.
    VllmFallback {
        /// The model id passed to the generated `vllm serve` (the variant name).
        model: String,
        /// Port the generated server serves on.
        port: u16,
    },
}

/// Choose the deploy route for `v` given whether its convention script exists on
/// the node (issue #709 §4): Mode A invokes the script, Mode B falls back to a
/// generated `vllm serve`. Pure.
fn deploy_route(v: &ModelVariant, script_present: bool) -> DeployRoute {
    let port = serve_port(v.tool);
    if script_present {
        DeployRoute::Script {
            path: dgx_registry::script_name(v),
            slug: dgx_registry::variant_slug(v),
            port,
        }
    } else {
        DeployRoute::VllmFallback {
            model: v.name.to_string(),
            port,
        }
    }
}

/// Remote command that prints `FOUND`/`MISSING` for the convention script and
/// always exits 0 (so a *missing* script is distinguishable from an SSH/host
/// error, which surfaces as a capture failure). Pure. The leading `~` in
/// `script_path` is left unquoted so the remote shell expands it.
fn script_probe_command(script_path: &str) -> String {
    format!("test -f {script_path} && echo FOUND || echo MISSING")
}

/// True iff the [`script_probe_command`] stdout reports the script present. Pure.
fn script_is_present(probe_stdout: &str) -> bool {
    probe_stdout.lines().any(|l| l.trim() == "FOUND")
}

/// On-node state dir for deploy launch bookkeeping (log + pidfile).
fn deploy_state_dir() -> &'static str {
    "$HOME/.newt/dgx/deploy"
}

/// Remote script that launches the convention deploy script detached with
/// `nohup`, logging under [`deploy_state_dir`] keyed by the (shell-safe) `slug`,
/// and recording its PID — mirroring [`dgx_vllm::vllm_remote_script`]. Pure.
/// `script_path`'s leading `~` is left unquoted so it expands on the node.
fn deploy_launch_command(script_path: &str, slug: &str) -> String {
    let dir = deploy_state_dir();
    format!(
        "set -eu\n\
         mkdir -p \"{dir}\"\n\
         nohup {script_path} > \"{dir}/{slug}.log\" 2>&1 &\n\
         echo $! > \"{dir}/{slug}.pid\"\n"
    )
}

/// The remote command that restarts the ollama service to drop CUDA buffers an
/// idle-but-loaded runner still pins (the `--restart-ollama` step). Tries the
/// user-scope unit first, then the system unit, covering both common install
/// styles; names no host. Pure.
fn restart_ollama_command() -> &'static str {
    "systemctl --user restart ollama 2>/dev/null || systemctl restart ollama"
}

/// Restart the ollama service on the node (best-effort: a warning, not an abort,
/// on failure). `ssh` is injectable so tests never touch a real node.
fn restart_ollama_service(ssh: &dyn SshExec, user: &str, host: &str) {
    match ssh.run(user, host, None, restart_ollama_command()) {
        Ok(()) => println!("  restarted ollama service"),
        Err(e) => eprintln!("  WARNING: failed to restart ollama: {e}"),
    }
}

/// Memory freed (GiB) between two budget snapshots: how much `available` grew.
/// Saturating — a workload that started in between can never report negative.
/// Pure.
fn freed_gib(before: &dgx_status::MemBudget, after: &dgx_status::MemBudget) -> f64 {
    let freed = after.available_bytes.saturating_sub(before.available_bytes);
    dgx_pull::bytes_to_gib(freed)
}

/// `dgx deploy [--strongest] [<model>]` — select, clear for, launch, health-gate.
async fn deploy(
    config_path: Option<&Path>,
    strongest: bool,
    model: Option<&str>,
    node: Option<&str>,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;

    // 1. Select the variant (the budget probe is only needed for --strongest).
    let variant = if strongest {
        let budget = dgx_status::gather_budget(&RealSshCapture, &user, &host, None)?;
        let available_gib = dgx_pull::bytes_to_gib(budget.available_bytes);
        println!("Available budget: {available_gib:.1} GiB");
        select_deploy_variant(true, None, available_gib)?
    } else {
        select_deploy_variant(false, model, 0.0)?
    };
    println!(
        "Selected {} {} (~{:.0} GiB, {})",
        variant.name,
        variant.format,
        variant.gib,
        variant.tool.as_str()
    );

    // 2. Evict competing workloads to make room before launching.
    println!("Clearing competing workloads…");
    evict_vllm_server(&RealSsh, &dgx).await?;
    evict_ollama_models(&dgx).await?;

    // 3. Mode A (convention script present) or Mode B (generated vLLM fallback).
    use crate::dgx_status::SshCapture as _;
    let script_path = dgx_registry::script_name(variant);
    let probe = RealSshCapture.capture(&user, &host, None, &script_probe_command(&script_path))?;
    let route = deploy_route(variant, script_is_present(&probe));
    let port = match &route {
        DeployRoute::Script { path, slug, port } => {
            println!("Mode A: invoking convention script {path}");
            RealSsh.run(&user, &host, None, &deploy_launch_command(path, slug))?;
            *port
        }
        DeployRoute::VllmFallback { model, port } => {
            println!(
                "Mode B: no convention script {script_path} on node — \
                 generating a vLLM launch for {model}"
            );
            let plan = build_plan_from_args(model, &default_vllm_plan_args())?;
            execute_vllm_plan(&RealSsh, &user, &host, &plan, false)?;
            *port
        }
    };

    // 4. Health gate: poll until the endpoint is serving, then report its URL.
    let endpoint = format!("http://{host}:{port}");
    println!("Waiting for {endpoint}/v1/models (cold load can take minutes)…");
    poll_vllm_ready(&endpoint, &RetryPolicy::for_local_inference()).await?;
    println!("Serving at {endpoint}");
    Ok(())
}

/// `dgx clear [--restart-ollama]` — evict inference workloads, report freed mem.
async fn clear(
    config_path: Option<&Path>,
    node: Option<&str>,
    restart_ollama: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    let user = dgx.ssh_user();
    let host = dgx.ssh_host()?;

    // Budget before (best-effort — never block the eviction on an SSH read).
    let before = dgx_status::gather_budget(&RealSshCapture, &user, &host, None).ok();

    println!("Clearing inference workloads on {host}…");
    evict_vllm_server(&RealSsh, &dgx).await?;
    evict_ollama_models(&dgx).await?;
    if restart_ollama {
        restart_ollama_service(&RealSsh, &user, &host);
    }

    // Budget after + freed report (best-effort).
    let after = dgx_status::gather_budget(&RealSshCapture, &user, &host, None).ok();
    match (before, after) {
        (Some(b), Some(a)) => {
            print!("{}", dgx_status::render_budget_human(&host, &a));
            println!("  Freed:      {:.1} GiB", freed_gib(&b, &a));
        }
        (None, Some(a)) => print!("{}", dgx_status::render_budget_human(&host, &a)),
        _ => println!("  (memory budget unavailable over SSH)"),
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// switch — flip the node between engines + models (Step 14.14)
// ---------------------------------------------------------------------------

/// Parse the `switch <engine>` token. Only `ollama` and `vllm` are switchable
/// node-local engines (the load-balancer / in-cluster kinds aren't).
fn parse_switch_engine(engine: &str) -> anyhow::Result<EndpointKind> {
    match engine.to_ascii_lowercase().as_str() {
        "ollama" => Ok(EndpointKind::Ollama),
        "vllm" => Ok(EndpointKind::Vllm),
        other => anyhow::bail!("`dgx switch` targets `ollama` or `vllm`, not {other:?}"),
    }
}

/// Default vLLM launch knobs for `switch vllm <model>` (use `dgx vllm up` for
/// fine control over dtype / port / context / tensor-parallel).
fn default_vllm_plan_args() -> VllmPlanArgs {
    VllmPlanArgs {
        served_name: None,
        dtype: None,
        tensor_parallel: 1,
        max_model_len: None,
        gpu_mem_util: 0.90,
        port: 8000,
        docker: false,
        extra: Vec::new(),
    }
}

/// Is `model` present in the Ollama endpoint's installed list? Pure.
fn ollama_has_model(installed: &[String], model: &str) -> bool {
    installed.iter().any(|m| m == model)
}

/// Pure: set the active endpoint + model in config (no URL — the endpoint's URL
/// is already configured on its node).
fn apply_active(config: &mut Config, kind: EndpointKind, model: &str) {
    let dgx = config.dgx.get_or_insert_with(Default::default);
    dgx.active_endpoint = kind;
    dgx.active_model = Some(model.to_string());
}

/// Persist the active endpoint + model to config (load → mutate → save).
fn persist_active(
    config_path: Option<&Path>,
    kind: EndpointKind,
    model: &str,
) -> anyhow::Result<()> {
    let mut config = load_config(config_path)?;
    apply_active(&mut config, kind, model);
    let save_path = config_path
        .map(std::path::PathBuf::from)
        .or_else(newt_core::Config::user_config_path)
        .ok_or_else(|| anyhow::anyhow!("cannot determine config file path"))?;
    config.save(&save_path)?;
    println!("Active endpoint {kind}; model {model}");
    println!("Saved → {}", save_path.display());
    Ok(())
}

/// `dgx switch <engine> <model>` — atomic engine+model flip, evicting the other.
async fn switch(
    config_path: Option<&Path>,
    engine: &str,
    model: &str,
    node: Option<&str>,
    dry_run: bool,
) -> anyhow::Result<()> {
    match parse_switch_engine(engine)? {
        EndpointKind::Vllm => {
            // Evict Ollama, launch vLLM, persist — exactly `vllm up --evict-ollama`.
            println!("Switching → vLLM serving {model:?} (evicting Ollama)…");
            vllm_up(
                config_path,
                model,
                node,
                &default_vllm_plan_args(),
                false,
                true,
                dry_run,
            )
            .await
        }
        EndpointKind::Ollama => switch_to_ollama(config_path, model, node, dry_run).await,
        // parse_switch_engine only yields Ollama / Vllm.
        other => anyhow::bail!("cannot switch to endpoint kind {other}"),
    }
}

/// The `switch ollama <model>` direction: stop vLLM, verify the model is pulled,
/// point newt at Ollama, and warm it.
async fn switch_to_ollama(
    config_path: Option<&Path>,
    model: &str,
    node: Option<&str>,
    dry_run: bool,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    println!("Switching → Ollama serving {model:?} (stopping any vLLM server)…");
    if dry_run {
        println!(
            "  [dry-run] would: stop the vLLM server, verify {model:?} is pulled, \
             set active=ollama:{model}, and warm it"
        );
        return Ok(());
    }
    // 1. Stop any running vLLM server (best-effort — frees the unified pool).
    evict_vllm_server(&RealSsh, &dgx).await?;
    // 2. Refuse with a hint when the model isn't pulled (the GLM-5.2 lesson —
    //    never silently pull multi-GB onto the node).
    let base = dgx.resolve_endpoint_for(EndpointKind::Ollama)?;
    let installed = fetch_ollama_models(&http_client(), &base)
        .await
        .unwrap_or_default();
    if !ollama_has_model(&installed, model) {
        anyhow::bail!(
            "{model:?} is not pulled on the Ollama endpoint — run `newt dgx pull {model}` first"
        );
    }
    // 3. Point newt at Ollama + the model.
    persist_active(config_path, EndpointKind::Ollama, model)?;
    // 4. Warm it (best-effort; a generous timeout for a cold load).
    let warm_client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(600))
        .build()
        .expect("build reqwest client");
    match warm_model(&warm_client, &base, model, "30m").await {
        Ok(Some(secs)) => println!("  warmed {model} in {secs:.1}s — now resident"),
        Ok(None) => println!("  {model} ready (already resident)"),
        Err(e) => eprintln!("  WARNING: warm failed: {e}"),
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// adopt — reconcile config vs the node's live state (Step 14.15)
// ---------------------------------------------------------------------------

/// The node's actual active engine + model from live probes: a running vLLM
/// server wins (it holds the pool), else a resident Ollama model, else nothing.
/// Pure.
fn derive_live_state(
    vllm: &[String],
    ollama: &[(String, Option<u64>)],
) -> Option<(EndpointKind, String)> {
    if let Some(m) = vllm.first() {
        Some((EndpointKind::Vllm, m.clone()))
    } else {
        ollama
            .first()
            .map(|(m, _)| (EndpointKind::Ollama, m.clone()))
    }
}

/// The reconcile verdict between newt's config and the node's live state.
#[derive(Debug, PartialEq, Eq)]
enum ReconcileVerdict {
    InSync {
        kind: EndpointKind,
        model: String,
    },
    Mismatch {
        cfg_kind: EndpointKind,
        cfg_model: Option<String>,
        live_kind: EndpointKind,
        live_model: String,
    },
    NoLiveState,
}

/// Compare configured (endpoint, model) against the node's live residency. In
/// sync when the configured engine+model is *actually serving* — even if the
/// other engine is also resident (a contended node is not a mismatch as long as
/// newt's configured target is live). Otherwise a mismatch against the primary
/// live engine (a running vLLM server, else a resident Ollama model), or
/// no-live when nothing runs. Pure.
fn reconcile(
    cfg_kind: EndpointKind,
    cfg_model: Option<&str>,
    vllm: &[String],
    ollama: &[(String, Option<u64>)],
) -> ReconcileVerdict {
    // Is the configured engine+model genuinely serving right now?
    let configured_is_live = match (cfg_kind, cfg_model) {
        (EndpointKind::Vllm, Some(m)) => vllm.iter().any(|x| x == m),
        (EndpointKind::Ollama, Some(m)) => ollama.iter().any(|(x, _)| x == m),
        // OllamaLb/InCluster or no configured model → can't confirm against the
        // node-local probes; fall through to the primary-live comparison.
        _ => false,
    };
    if let (true, Some(m)) = (configured_is_live, cfg_model) {
        return ReconcileVerdict::InSync {
            kind: cfg_kind,
            model: m.to_string(),
        };
    }
    match derive_live_state(vllm, ollama) {
        None => ReconcileVerdict::NoLiveState,
        Some((live_kind, live_model)) => ReconcileVerdict::Mismatch {
            cfg_kind,
            cfg_model: cfg_model.map(str::to_string),
            live_kind,
            live_model,
        },
    }
}

/// How a reconcile mismatch is resolved.
#[derive(Debug, PartialEq, Eq)]
enum ReconcileAction {
    Adopt,
    Enforce,
    Report,
}

/// Pick the action from the flags + terminal state. `--adopt`/`--enforce` are
/// explicit; otherwise the TTY answer decides; a piped/headless run with no flag
/// only reports (never changes state silently). Pure.
/// The three-way drift choice, as options.
///
/// `a`/`e`/`c` are accelerators on `adopt`/`enforce`/`cancel`, which is what
/// they always were — the private `match` this replaces spelled each of them
/// twice ("a" and "adopt") and could not advertise either.
fn reconcile_menu() -> newt_core::InteractionDefinition {
    newt_core::interaction_form::menu(
        "config and the node disagree",
        "",
        &[
            ("a", "adopt the node's state into config"),
            ("e", "enforce config onto the node"),
            ("c", "cancel, change nothing"),
        ],
    )
}

/// Which action the drift prompt resolved to.
///
/// `picked` is an OPTION ID (`a` / `e` / `c`), not a raw line: the parsing
/// happened in `resolve_typed` before it got here. What stays is the
/// PRECEDENCE, which is policy rather than parsing — explicit flags win over
/// any answer, and a non-TTY reports rather than changing state.
fn reconcile_action(
    adopt: bool,
    enforce: bool,
    is_tty: bool,
    tty_answer: Option<&str>,
) -> ReconcileAction {
    if adopt {
        return ReconcileAction::Adopt;
    }
    if enforce {
        return ReconcileAction::Enforce;
    }
    if !is_tty {
        return ReconcileAction::Report;
    }
    match tty_answer {
        Some("a") => ReconcileAction::Adopt,
        Some("e") => ReconcileAction::Enforce,
        _ => ReconcileAction::Report,
    }
}

/// `dgx adopt` — interrogate the node, compare to config, and reconcile.
async fn adopt_cmd(
    config_path: Option<&Path>,
    adopt: bool,
    enforce: bool,
    node: Option<&str>,
) -> anyhow::Result<()> {
    let mut dgx = dgx_config(config_path)?;
    if let Some(n) = node {
        dgx.active_node = Some(n.to_string());
    }
    // Interrogate the node (best-effort, like `dgx gpu`).
    let client = http_client();
    let ollama = match dgx.resolve_endpoint_for(EndpointKind::Ollama) {
        Ok(base) => fetch_ollama_ps(&client, &base).await.unwrap_or_default(),
        Err(_) => Vec::new(),
    };
    let vllm = match dgx.resolve_endpoint_for(EndpointKind::Vllm) {
        Ok(base) => fetch_vllm_models(&base).await.unwrap_or_default(),
        Err(_) => Vec::new(),
    };
    match reconcile(
        dgx.active_endpoint,
        dgx.active_model.as_deref(),
        &vllm,
        &ollama,
    ) {
        ReconcileVerdict::InSync { kind, model } => {
            println!("dgx1 in sync: the configured {kind} model {model:?} is serving on the node.");
            Ok(())
        }
        ReconcileVerdict::NoLiveState => {
            println!(
                "Nothing is running on the node (no vLLM server, no resident Ollama model). \
                 Config wants {} serving {:?}; start it with `dgx switch` / `vllm up` / `warm`.",
                dgx.active_endpoint,
                dgx.active_model.as_deref().unwrap_or("<none>")
            );
            Ok(())
        }
        ReconcileVerdict::Mismatch {
            cfg_kind,
            cfg_model,
            live_kind,
            live_model,
        } => {
            println!("dgx1 mismatch — config and the node disagree:");
            println!(
                "  config: {cfg_kind} : {}",
                cfg_model.as_deref().unwrap_or("<none>")
            );
            println!("  node:   {live_kind} : {live_model}");
            let is_tty = std::io::stdin().is_terminal();
            let answer = if !adopt && !enforce && is_tty {
                // NOT a "press enter to continue" pause: this is a three-way
                // choice whose answer `reconcile_action` consumes below. The
                // `.ok()` this replaces discarded the io::Result, not the
                // response — so a read ERROR arrived at the resolver as an
                // empty line, indistinguishable from the operator pressing
                // enter. Both land on `Report` (change nothing), so it was
                // fail-closed but not honest. `None` now means "no answer",
                // for either reason, and says so in one place.
                let window = newt_core::tty::Terminal::suspend_for_prompt(
                    newt_core::tty::TerminalTaker::PlainCliConfirm,
                );
                // The three choices are OPTIONS now, so `a`/`adopt` are an
                // accelerator and an id rather than two strings a private
                // match had to know about. `resolve_on_terminal` folds EOF, a
                // cancel and an unrecognised answer into one `None`, which is
                // the `Report` (change nothing) arm below.
                newt_core::interaction_terminal::resolve_on_terminal(&window, &reconcile_menu())
                    .map(|picked| picked.as_str().to_string())
            } else {
                None
            };
            match reconcile_action(adopt, enforce, is_tty, answer.as_deref()) {
                ReconcileAction::Adopt => {
                    println!("Adopting: pointing newt at {live_kind} : {live_model}.");
                    persist_active(config_path, live_kind, &live_model)
                }
                ReconcileAction::Enforce => {
                    // Validate switchability BEFORE promising anything — `switch`
                    // only manages local ollama/vllm, not the lb/in-cluster kinds.
                    if !matches!(cfg_kind, EndpointKind::Ollama | EndpointKind::Vllm) {
                        anyhow::bail!(
                            "cannot enforce a non-local endpoint ({cfg_kind}) — `dgx switch` only \
                             manages ollama/vllm on the node; adopt the node's state instead"
                        );
                    }
                    let model = cfg_model.ok_or_else(|| {
                        anyhow::anyhow!(
                            "cannot enforce: no active_model configured — set one (`dgx use`) \
                             or adopt the node's state instead"
                        )
                    })?;
                    println!("Enforcing: switching the node to {cfg_kind} : {model}.");
                    switch(config_path, cfg_kind.as_str(), &model, node, false).await
                }
                ReconcileAction::Report => {
                    println!(
                        "No change. Re-run with --adopt (point newt at the node) or \
                         --enforce (switch the node to the config)."
                    );
                    Ok(())
                }
            }
        }
    }
}

/// The one-line session-start drift notice for a reconcile verdict — `Some` only
/// on a mismatch (in-sync / no-live / unreachable stay silent at startup). Pure.
fn drift_notice_line(verdict: &ReconcileVerdict) -> Option<String> {
    match verdict {
        ReconcileVerdict::Mismatch {
            cfg_kind,
            cfg_model,
            live_kind,
            live_model,
        } => Some(format!(
            "dgx note: the node is running {live_kind}:{live_model}, but config expects \
             {cfg_kind}:{} — run `newt dgx adopt` to reconcile.",
            cfg_model.as_deref().unwrap_or("<none>")
        )),
        _ => None,
    }
}

/// Best-effort session-start notice: if the node's live engine/model has drifted
/// from `[dgx]` config, print one line (to stderr) pointing at `dgx adopt`.
/// Never errors and never blocks startup — it's skipped when dgx isn't in use,
/// and the whole probe is bounded by a short timeout so a slow/unreachable node
/// is a silent no-op. The chosen "startup notice + `dgx adopt`" UX (Step 14.16).
///
/// Takes an ALREADY-RESOLVED config (#1951) rather than a path to resolve
/// itself: this is one of several startup checks the `Command::Code` dispatch
/// arm makes off a single shared resolution, so a config that fails to
/// resolve is not re-read from disk (and re-warned about) once per check.
pub(crate) async fn startup_drift_notice(config: Option<&Config>) {
    let Some(dgx) = config.and_then(|c| c.dgx.clone()) else {
        return;
    };
    // Only when dgx is actually configured/in use — never probe otherwise.
    if dgx.nodes.is_empty() && dgx.active_model.is_none() {
        return;
    }
    let client = http_client();
    let probe = async {
        let ollama = match dgx.resolve_endpoint_for(EndpointKind::Ollama) {
            Ok(base) => fetch_ollama_ps(&client, &base).await.unwrap_or_default(),
            Err(_) => Vec::new(),
        };
        let vllm = match dgx.resolve_endpoint_for(EndpointKind::Vllm) {
            Ok(base) => fetch_vllm_models(&base).await.unwrap_or_default(),
            Err(_) => Vec::new(),
        };
        (ollama, vllm)
    };
    // Bound the whole interrogation so a down node never delays the session.
    let (ollama, vllm) =
        match tokio::time::timeout(std::time::Duration::from_millis(2500), probe).await {
            Ok(r) => r,
            Err(_) => return,
        };
    let verdict = reconcile(
        dgx.active_endpoint,
        dgx.active_model.as_deref(),
        &vllm,
        &ollama,
    );
    if let Some(line) = drift_notice_line(&verdict) {
        eprintln!("{line}");
    }
}

#[cfg(test)]
#[path = "dgx/tests.rs"]
mod tests;