staging: local-view kind — keep a hot side table on local NVMe, gated by a pre-start free-space check

Measured on the 1017 lane: one 250-token completion pulled 258 MB off the
shared NFS mount (Engram rows are streamed at ~1 MB/token), which is why the
vLLM lane's prefill sat at 605 t/s against production's 2,800-3,100.

[[staging]] kind = "local-view" declares the whole thing in the recipe: files
matching  become real node-local copies, everything else becomes a
symlink to the shared cache's container path; serve mounts the view at
/root/.cache/hf-view and rewrites model to it. Content-hash ledger, so a second
serve is verification-only and a hand-staged tree is adopted.

serve now also runs staging::preflight before any copy or container: every
local-view block is measured per node (need vs free, with min_free headroom)
and a short node aborts the serve with the node, repo, size and target named —
instead of dying halfway through a 200 GB copy.

specs/tech-architecture/staging-feature.md

@@ -185,3 +185,54 @@ line** (like `NCCL_IB_GID_INDEX=auto`), with the config override as opt-out.
    recipe migration + A/B bench (NAS vs local engram) to keep the number.
 4. Watermark/LRU + `staging` subcommand.
 5. Probe-and-go consolidation.
+
+## `kind = "local-view"` — keep the hot side table on local NVMe
+
+Measured 2026-10-23 on the 1017 lane: a single 250-token completion made the
+shared NFS mount serve **+258 MB of reads** — the Engram n-gram tables are
+streamed at **~1 MB per token**, straight off the NAS. That is why a vLLM lane
+with the same checkpoint and the same four Sparks showed 605 t/s prefill where
+the production lane (which stages Engram locally) shows 2,800–3,100.
+
+`local-view` makes that staging declarative — no manual copy dance:
+
+```toml
+[[staging]]
+kind = "local-view"
+repo = "deepseek-ai/DeepSeek-V4.1-Flash-1017"        # default: the recipe's model
+dst = "/home/kilork/models/dsv41-1017-local"          # host dir, per node
+local = ["model-00047-of-00048.safetensors",          # REAL local copies
+         "model-00048-of-00048.safetensors"]
+min_free = "230G"                                     # headroom on top of the copies
+verify = "sample-hash"                                # content, never mtime/size
+```
+
+The block materializes `dst` as a *view* of the snapshot: files matching
+`local` become real node-local copies, everything else becomes a symlink to the
+shared cache's **container** path (`/root/.cache/huggingface/…`). `serve` then
+mounts the view at `/root/.cache/hf-view` and rewrites `model` to
+`/root/.cache/hf-view/<repo>`, so the runtime reads the table from NVMe while
+the bulk of the weights still resolves through the shared mount.
+
+Verification is a per-view ledger of sampled content hashes
+(`dst/.sm121-view-ledger.json`), the same discipline as `hub-mirror`: a second
+serve of an unchanged repo costs zero copies, a same-size mutation is detected
+and re-copied, and a hand-staged tree is adopted instead of re-copied.
+
+### Free-space preflight (all staging kinds)
+
+`serve` now runs `staging::preflight` **before** any copy and before the first
+container starts. For every `local-view` block it asks each launch node how
+many bytes the `local` copies need and how much the target filesystem has
+(the node script's `measure` mode, which also honours `min_free` headroom), and
+aborts with a message that names the node, the repo, the size and the target:
+
+```
+not enough free space on the stage target(s) — nothing was started:
+  dgx7.lan: deepseek-ai/DeepSeek-V4.1-Flash-1017 needs 203.0 GiB in
+  /home/kilork/models/dsv41-1017-local, has 167.0 GiB
+```
+
+This is deliberately a pre-start gate: dying halfway through a 200 GB copy (or
+starting four containers that then find a half-populated directory) is much
+worse than a refusal that says exactly what to free.

src/config/staging.rs

@@ -18,6 +18,12 @@ pub enum StagingKind {
     /// Per-rank sparse copy of Engram table rows at the original safetensors
     /// byte offsets (spec P2).
     SparseRows,
+    /// Node-local *view* of one repo snapshot: the files matching `local` are
+    /// real copies (the ones a runtime streams at every token — Engram tables),
+    /// everything else is a symlink into the shared HF cache (container-side
+    /// path, so it resolves inside the container). `serve` mounts the view and
+    /// rewrites `model` to it.
+    LocalView,
 }
 
 /// One `[[staging]]` table in recipe.toml.
@@ -48,6 +54,15 @@ pub struct Staging {
     /// `["engram-page15-v1"]`.
     #[serde(default)]
     pub skip: Vec<String>,
+    /// The repo a `local-view` block mirrors (default: the recipe's model).
+    #[serde(default)]
+    pub repo: Option<String>,
+    /// `local-view`: file globs (relative to the snapshot dir) that must be
+    /// real node-local copies. Everything else is symlinked into the shared
+    /// cache. Keep this to what the runtime actually streams — the whole point
+    /// is to leave the bulk of the weights on the shared storage.
+    #[serde(default)]
+    pub local: Vec<String>,
     /// Materialized exports (hub-mirror): `"name:repo-dir:/container-root"`
     /// — builds `dst/<name>` as a regular-file tree (hardlinks onto staged
     /// blobs) for O_NOFOLLOW runtimes; entries the mirror does not hold are
@@ -95,6 +110,16 @@ pub enum StagingVerify {
     Presence,
 }
 
+impl StagingVerify {
+    /// The value the node-side scripts take (`--verify`).
+    pub fn as_str(self) -> &'static str {
+        match self {
+            StagingVerify::SampleHash => "sample-hash",
+            StagingVerify::Presence => "presence",
+        }
+    }
+}
+
 impl Staging {
     /// The block's ledger key (caller ensures uniqueness within a recipe).
     pub fn ledger_id(&self, index: usize) -> String {
@@ -106,6 +131,32 @@ impl Staging {
 mod tests {
     use super::*;
 
+    #[test]
+    fn parses_local_view_block() {
+        let toml_str = r#"
+[[staging]]
+kind = "local-view"
+repo = "deepseek-ai/DeepSeek-V4.1-Flash-1017"
+dst = "~/models/1017-local"
+local = ["model-00047-of-00048.safetensors", "model-00048-of-00048.safetensors"]
+min_free = "230G"
+"#;
+        #[derive(Deserialize)]
+        struct Wrap {
+            staging: Vec<Staging>,
+        }
+        let w: Wrap = toml::from_str(toml_str).unwrap();
+        let b = &w.staging[0];
+        assert_eq!(b.kind, StagingKind::LocalView);
+        assert_eq!(
+            b.repo.as_deref(),
+            Some("deepseek-ai/DeepSeek-V4.1-Flash-1017")
+        );
+        assert_eq!(b.local.len(), 2);
+        assert_eq!(b.min_free.as_deref(), Some("230G"));
+        assert_eq!(b.verify, StagingVerify::SampleHash);
+    }
+
     #[test]
     fn parses_full_staging_block() {
         let toml_str = r#"

src/docker/launch_node.rs

@@ -98,6 +98,18 @@ pub(crate) async fn launch_node(
             Err(e) => return fail(e),
         }
     }
+    // Node-local staging view (`[[staging]] kind = "local-view"`): the files a
+    // runtime streams at every token live on local NVMe, the rest resolve
+    // through the shared cache mount. `model` points at the view.
+    if let Some((host, container, repo)) = crate::staging::local_view_mount(config, &recipe) {
+        extra_mounts.push(command::ResolvedMount {
+            host: host.display().to_string(),
+            container: container.clone(),
+            read_only: false,
+            env: None,
+        });
+        recipe.model = format!("{container}/{repo}");
+    }
     // Per-node mount manifest (generated by the recipe's before-serve hooks):
     // one `host<TAB>container` line per extra read-only bind. Skipped in
     // dry-run (no SSH); a missing/empty file is a hard error in live runs so

src/staging.rs

@@ -7,6 +7,7 @@
 
 pub mod mirror;
 pub mod rows;
+pub mod view;
 
 use crate::config::{Config, Node, Recipe, StagingKind, expand_tilde};
 use anyhow::{Result, bail};
@@ -35,6 +36,50 @@ impl StageMode {
     }
 }
 
+/// The first `local-view` block's mount: (host view dir, container mount,
+/// repo). `serve` appends the mount and rewrites `model` to
+/// `<container>/<repo>`, so the engine reads the staged files from local NVMe
+/// and everything else through the shared cache's own mount.
+pub fn local_view_mount(config: &Config, recipe: &Recipe) -> Option<(PathBuf, String, String)> {
+    for (i, s) in recipe.staging.iter().enumerate() {
+        if s.kind != StagingKind::LocalView {
+            continue;
+        }
+        let id = s.ledger_id(i);
+        let host = PathBuf::from(view::dst_for(s, &id, &config.ssh_user));
+        let repo = view::repo_for(s, &recipe.model);
+        return Some((host, view::CONTAINER_ROOT.to_string(), repo));
+    }
+    None
+}
+
+/// Free-space preflight for every staging block whose footprint is known.
+/// Runs BEFORE any copy and before any container starts: a node that cannot
+/// hold the staged files fails the serve with the node, the size and the
+/// target directory, instead of dying halfway through a 200 GB copy.
+pub async fn preflight(
+    config: &Config,
+    nodes: &[Node],
+    recipe: &Recipe,
+    shared_source: &Path,
+    dry_run: bool,
+) -> Result<()> {
+    let mut needs: Vec<view::Need> = Vec::new();
+    for (i, block) in recipe.staging.iter().enumerate() {
+        if block.kind != StagingKind::LocalView || dry_run {
+            continue;
+        }
+        needs.extend(view::measure(config, nodes, block, i, &recipe.model, shared_source).await?);
+    }
+    for n in &needs {
+        info!("staging space: {}", n.line());
+    }
+    if let Some(msg) = view::shortfall(&needs) {
+        bail!(msg);
+    }
+    Ok(())
+}
+
 /// True when the recipe stages at least one hub-mirror (the HF mount source
 /// gets swapped to the mirror's dst in that case).
 pub fn has_hub_mirror(recipe: &Recipe) -> bool {
@@ -113,6 +158,7 @@ pub async fn hf_mount(
             Some(p) => PathBuf::from(expand_tilde(p, &config.ssh_user)),
             None => config.resolved_hf_cache_dir(),
         };
+        preflight(config, nodes, recipe, &shared_source, dry_run).await?;
         run(config, nodes, recipe, &shared_source, mode, dry_run).await?;
     }
     if staged {
@@ -182,6 +228,16 @@ pub async fn run(
             for (i, block) in recipe.staging.iter().enumerate() {
                 match block.kind {
                     StagingKind::HubMirror => mirror::stage_node(&ctx, &node, block, i).await?,
+                    StagingKind::LocalView => {
+                        let vctx = view::StageCtx {
+                            config: &config,
+                            recipe: &recipe,
+                            shared_source: &source,
+                            force: mode == StageMode::Force,
+                            dry_run,
+                        };
+                        view::stage_node(&vctx, &node, block, i).await?
+                    }
                     StagingKind::SparseRows => {
                         let rctx = rows::Ctx {
                             config: &config,

src/staging/scripts/local_view.py

@@ -0,0 +1,200 @@
+#!/usr/bin/env python3
+"""Build a node-local *view* of one HF repo snapshot.
+
+The point: some checkpoints stream a large side-table at every token (Engram
+n-gram tables in DeepSeek-V4.1 read ~1 MB of table rows per token). Serving
+those rows over the shared NAS costs an order of magnitude more than serving
+them from local NVMe, while the bulk of the weights is fine on shared storage.
+
+So the view dir contains:
+  * REAL local copies of the files matching `--local` globs, and
+  * SYMLINKS for everything else, pointing at the container-side path of the
+    shared cache (`--container-root`, default /root/.cache/huggingface) — the
+    container mounts the same view and the same shared cache, so the symlinks
+    resolve inside it.
+
+`measure` mode reports the bytes the local copies need (the launcher's
+preflight uses it before starting anything), `apply` mode builds the view.
+Verification is content-based (`--verify sample-hash`, the default) with a
+per-view ledger — never mtime/size, which are unreliable across NFS and rsync.
+"""
+
+import argparse
+import fnmatch
+import hashlib
+import json
+import os
+import shutil
+import sys
+
+LEDGER = ".sm121-view-ledger.json"
+SAMPLE_BYTES = 1 << 20
+
+
+def repo_dir(src_root, repo):
+    return os.path.join(src_root, "models--" + repo.replace("/", "--"))
+
+
+def snapshot_dir(src_root, repo, revision=None):
+    base = repo_dir(src_root, repo)
+    refs = os.path.join(base, "refs", "main")
+    if revision is None:
+        try:
+            with open(refs) as f:
+                revision = f.read().strip()
+        except OSError:
+            revision = None
+    if revision is None:
+        snaps = sorted(os.listdir(os.path.join(base, "snapshots")))
+        if not snaps:
+            raise SystemExit(f"no snapshots under {base}/snapshots")
+        revision = snaps[-1]
+    return os.path.join(base, "snapshots", revision), revision
+
+
+def matches(name, globs):
+    return any(fnmatch.fnmatch(name, g) for g in globs)
+
+
+def list_files(snapshot):
+    """Relative paths of every regular file (or symlink to one) in the snapshot."""
+    out = []
+    for root, dirs, files in os.walk(snapshot):
+        dirs[:] = [d for d in dirs if d not in (".git", "blobs")]
+        for f in files:
+            full = os.path.join(root, f)
+            if os.path.isfile(full):
+                out.append(os.path.relpath(full, snapshot))
+    return sorted(out)
+
+
+def sample_hash(path):
+    h = hashlib.sha256()
+    size = os.path.getsize(path)
+    with open(path, "rb", buffering=0) as f:
+        for off in (0, max(0, size // 2 - SAMPLE_BYTES // 2), max(0, size - SAMPLE_BYTES)):
+            f.seek(off)
+            h.update(f.read(SAMPLE_BYTES))
+    h.update(str(size).encode())
+    return h.hexdigest()[:32]
+
+
+def local_bytes(snapshot, globs):
+    need = 0
+    files = []
+    for rel in list_files(snapshot):
+        if matches(rel, globs):
+            need += os.path.getsize(os.path.join(snapshot, rel))
+            files.append(rel)
+    return need, files
+
+
+def parse_size(text):
+    t = text.strip().upper()
+    mult = 1
+    for suf, m in (("T", 1 << 40), ("G", 1 << 30), ("M", 1 << 20), ("K", 1 << 10)):
+        if t.endswith(suf):
+            t, mult = t[:-1], m
+            break
+    return int(float(t) * mult)
+
+
+def free_bytes(path):
+    probe = path
+    while not os.path.exists(probe):
+        parent = os.path.dirname(probe)
+        if parent == probe:
+            return 0
+        probe = parent
+    st = os.statvfs(probe)
+    return st.f_bavail * st.f_frsize
+
+
+def load_ledger(dst):
+    try:
+        with open(os.path.join(dst, LEDGER)) as f:
+            return json.load(f)
+    except OSError:
+        return {"files": {}}
+
+
+def save_ledger(dst, ledger):
+    tmp = os.path.join(dst, LEDGER + ".tmp")
+    with open(tmp, "w") as f:
+        json.dump(ledger, f, indent=1, sort_keys=True)
+    os.replace(tmp, os.path.join(dst, LEDGER))
+
+
+def container_path(container_root, repo, revision, rel):
+    return os.path.join(
+        container_root, "hub", "models--" + repo.replace("/", "--"), "snapshots", revision, rel
+    )
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("mode", choices=["measure", "apply"])
+    ap.add_argument("--src", required=True, help="host-side shared HF cache root")
+    ap.add_argument("--repo", required=True)
+    ap.add_argument("--dst", required=True)
+    ap.add_argument("--container-root", default="/root/.cache/huggingface")
+    ap.add_argument("--local", action="append", default=[])
+    ap.add_argument("--verify", default="sample-hash", choices=["sample-hash", "presence"])
+    ap.add_argument("--min-free", default=None)
+    ap.add_argument("--force", action="store_true")
+    a = ap.parse_args()
+
+    snapshot, revision = snapshot_dir(a.src, a.repo)
+    need, files = local_bytes(snapshot, a.local)
+    free = free_bytes(a.dst)
+    # A preflight may ask for extra headroom on top of the copies themselves.
+    floor = parse_size(a.min_free) if a.min_free else 0
+    required = max(need + floor, floor)
+
+    if a.mode == "measure":
+        print(json.dumps({"repo": a.repo, "revision": revision, "need": need,
+                          "free": free, "files": len(files)}))
+        return 0
+
+    if free < required:
+        print(f"NEED {required} FREE {free} DST {a.dst}", file=sys.stderr)
+        return 3
+
+    os.makedirs(a.dst, exist_ok=True)
+    ledger = load_ledger(a.dst)
+    seen = ledger.setdefault("files", {})
+    copied = kept = 0
+    for rel in list_files(snapshot):
+        src = os.path.join(snapshot, rel)
+        dst = os.path.join(a.dst, rel)
+        os.makedirs(os.path.dirname(dst) or a.dst, exist_ok=True)
+        if not matches(rel, a.local):
+            target = container_path(a.container_root, a.repo, revision, rel)
+            if os.path.islink(dst) and os.readlink(dst) == target:
+                continue
+            if os.path.exists(dst) or os.path.islink(dst):
+                os.remove(dst)
+            os.symlink(target, dst)
+            seen.pop(rel, None)
+            continue
+        want = None if a.verify == "presence" else sample_hash(src)
+        have = seen.get(rel)
+        if not a.force and os.path.isfile(dst) and not os.path.islink(dst) and have == want:
+            kept += 1
+            continue
+        tmp = dst + ".sm121-tmp"
+        shutil.copyfile(src, tmp)
+        os.replace(tmp, dst)
+        seen[rel] = want if want is not None else sample_hash(dst)
+        copied += 1
+    ledger["repo"] = a.repo
+    ledger["revision"] = revision
+    ledger["local"] = a.local
+    save_ledger(a.dst, ledger)
+    print(f"local-view {a.repo}@{revision[:12]}: {copied} copied, {kept} verified, "
+          f"{need} bytes local, {free} free")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/staging/view.rs

@@ -0,0 +1,285 @@
+//! `local-view` staging block executor.
+//!
+//! Some checkpoints stream a large side table at every token (the DSV4.1
+//! Engram tables read ~1 MB of rows per token). Serving those rows from the
+//! shared NAS costs an order of magnitude more than from local NVMe, while the
+//! bulk of the weights belongs on shared storage. A `local-view` block
+//! materializes, per launch node, a directory holding **real copies** of the
+//! files matching `local` and symlinks (into the shared cache's container
+//! path) for everything else; `serve` mounts it and points `model` at it.
+//!
+//! The node-side work is `scripts/local_view.py`; this module only builds the
+//! command lines, decides the container mount and computes the bytes a node
+//! must have free (used by the pre-start free-space check).
+
+use crate::config::{Config, Node, Staging, expand_tilde};
+use crate::ssh::{run_ssh_long_running, upload_script};
+use anyhow::{Result, bail};
+use std::path::Path;
+use tracing::info;
+
+pub const SCRIPT_BODY: &str = include_str!("scripts/local_view.py");
+
+/// Container-side path the view is mounted at (the model references it).
+pub const CONTAINER_ROOT: &str = "/root/.cache/hf-view";
+
+fn script_path(id: &str) -> String {
+    format!("/tmp/sm121-local-view-{id}.py")
+}
+
+/// The repo a block views: explicit `repo`, else the recipe's model.
+pub fn repo_for(block: &Staging, model: &str) -> String {
+    block.repo.clone().unwrap_or_else(|| model.to_string())
+}
+
+/// Node-side view directory (tilde-expanded).
+pub fn dst_for(block: &Staging, id: &str, ssh_user: &str) -> String {
+    block
+        .dst
+        .as_deref()
+        .map(|d| expand_tilde(d, ssh_user))
+        .unwrap_or_else(|| format!("~/.cache/hf-view/{id}"))
+}
+
+/// `measure` (bytes a node needs) or `apply` (build the view).
+pub struct ViewSpec<'a> {
+    pub repo: &'a str,
+    pub local: &'a [String],
+    pub verify: &'a str,
+    pub min_free: Option<&'a str>,
+    pub force: bool,
+}
+
+/// Command line for `python3 <script> apply --src … --repo … --dst … --local …`.
+pub fn command_for(id: &str, mode: &str, src: &str, dst: &str, spec: &ViewSpec<'_>) -> String {
+    let mut cmd = format!(
+        "python3 {} {} --src {} --repo {} --dst {} --container-root {} --verify {}",
+        crate::docker::shq(&script_path(id)),
+        mode,
+        crate::docker::shq(src),
+        crate::docker::shq(spec.repo),
+        crate::docker::shq(dst),
+        crate::docker::shq(CONTAINER_ROOT),
+        spec.verify,
+    );
+    for g in spec.local {
+        cmd.push_str(&format!(" --local {}", crate::docker::shq(g)));
+    }
+    if let Some(m) = spec.min_free {
+        cmd.push_str(&format!(" --min-free {}", crate::docker::shq(m)));
+    }
+    if spec.force {
+        cmd.push_str(" --force");
+    }
+    cmd
+}
+
+/// What one node needs free for a view, plus what it has (bytes).
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Need {
+    pub node: String,
+    pub repo: String,
+    pub need: u64,
+    pub free: u64,
+    pub dst: String,
+}
+
+impl Need {
+    pub fn short(&self) -> bool {
+        self.free < self.need
+    }
+
+    pub fn line(&self) -> String {
+        format!(
+            "{}: {} → needs {:.1} GiB, has {:.1} GiB{}",
+            self.node,
+            self.repo,
+            self.need as f64 / (1u64 << 30) as f64,
+            self.free as f64 / (1u64 << 30) as f64,
+            if self.short() { "  ⚠ SHORT" } else { "" }
+        )
+    }
+}
+
+/// Parse the `measure` output (`{"need": …, "free": …}`) of the node script.
+pub fn parse_measure(node: &str, repo: &str, dst: &str, stdout: &str) -> Result<Need> {
+    let v: serde_json::Value = serde_json::from_str(stdout.trim()).map_err(|e| {
+        anyhow::anyhow!("local-view measure on {node}: unparsable output {stdout:?}: {e}")
+    })?;
+    let num = |k: &str| v.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0);
+    Ok(Need {
+        node: node.to_string(),
+        repo: repo.to_string(),
+        need: num("need"),
+        free: num("free"),
+        dst: dst.to_string(),
+    })
+}
+
+/// The aggregated, human-facing shortfall message (`None` when every node fits).
+pub fn shortfall(needs: &[Need]) -> Option<String> {
+    let short: Vec<&Need> = needs.iter().filter(|n| n.short()).collect();
+    if short.is_empty() {
+        return None;
+    }
+    let mut msg =
+        String::from("not enough free space on the stage target(s) — nothing was started:\n");
+    for n in &short {
+        msg.push_str(&format!(
+            "  {}: {} needs {:.1} GiB in {}, has {:.1} GiB\n",
+            n.node,
+            n.repo,
+            n.need as f64 / (1u64 << 30) as f64,
+            n.dst,
+            n.free as f64 / (1u64 << 30) as f64
+        ));
+    }
+    msg.push_str("free space (docker rmi / stale staging dirs) or point dst at a bigger volume.");
+    Some(msg)
+}
+
+/// Ask every launch node how much space a block's `local` copies need.
+pub async fn measure(
+    config: &Config,
+    nodes: &[Node],
+    block: &Staging,
+    index: usize,
+    model: &str,
+    shared_source: &Path,
+) -> Result<Vec<Need>> {
+    let id = block.ledger_id(index);
+    let repo = repo_for(block, model);
+    let spec = ViewSpec {
+        repo: &repo,
+        local: &block.local,
+        verify: block.verify.as_str(),
+        min_free: block.min_free.as_deref(),
+        force: false,
+    };
+    let mut out = Vec::new();
+    for node in nodes {
+        let dst = dst_for(block, &id, &config.ssh_user);
+        upload_script(config, node, SCRIPT_BODY, &script_path(&id), false).await?;
+        let cmd = command_for(
+            &id,
+            "measure",
+            &shared_source.display().to_string(),
+            &dst,
+            &spec,
+        );
+        let stdout = crate::ssh::run_ssh(config, node, cmd, false).await?;
+        out.push(parse_measure(&node.name, &repo, &dst, &stdout)?);
+    }
+    Ok(out)
+}
+
+/// Everything one node's `local-view` pass needs.
+pub struct StageCtx<'a> {
+    pub config: &'a Config,
+    pub recipe: &'a crate::config::Recipe,
+    pub shared_source: &'a Path,
+    pub force: bool,
+    pub dry_run: bool,
+}
+
+/// Stage one `local-view` block on one node.
+pub async fn stage_node(
+    ctx: &StageCtx<'_>,
+    node: &Node,
+    block: &Staging,
+    index: usize,
+) -> Result<()> {
+    let (config, model, shared_source, force, dry_run) = (
+        ctx.config,
+        ctx.recipe.model.as_str(),
+        ctx.shared_source,
+        ctx.force,
+        ctx.dry_run,
+    );
+    if block.local.is_empty() {
+        bail!(
+            "[[staging]] local-view block '{}' needs at least one `local` glob (the files that must be real node-local copies)",
+            block.ledger_id(index)
+        );
+    }
+    let id = block.ledger_id(index);
+    let repo = repo_for(block, model);
+    let dst = dst_for(block, &id, &config.ssh_user);
+    let spec = ViewSpec {
+        repo: &repo,
+        local: &block.local,
+        verify: block.verify.as_str(),
+        min_free: block.min_free.as_deref(),
+        force,
+    };
+    let src = shared_source.display().to_string();
+    if dry_run {
+        info!(
+            "dry-run staging[{}] on {}: local-view {} → {} (local: {:?}, min_free {:?})",
+            id, node.name, repo, dst, block.local, block.min_free
+        );
+        return Ok(());
+    }
+    upload_script(config, node, SCRIPT_BODY, &script_path(&id), false).await?;
+    let cmd = command_for(&id, "apply", &src, &dst, &spec);
+    run_ssh_long_running(config, node, cmd, false).await?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn need(node: &str, need: u64, free: u64) -> Need {
+        Need {
+            node: node.into(),
+            repo: "org/model".into(),
+            need,
+            free,
+            dst: "/v".into(),
+        }
+    }
+
+    #[test]
+    fn measure_output_parses() {
+        let n = parse_measure(
+            "dgx5",
+            "org/m",
+            "/v",
+            r#"{"repo":"org/m","need":203,"free":400}"#,
+        )
+        .unwrap();
+        assert_eq!(n.need, 203);
+        assert_eq!(n.free, 400);
+        assert!(!n.short());
+    }
+
+    #[test]
+    fn shortfall_lists_only_the_short_nodes() {
+        let needs = vec![need("a", 100, 200), need("b", 300, 250)];
+        let msg = shortfall(&needs).expect("b is short");
+        assert!(msg.contains("b:"), "{msg}");
+        assert!(!msg.contains("a:"), "{msg}");
+        assert!(msg.contains("nothing was started"), "{msg}");
+        assert!(shortfall(&[need("a", 1, 2)]).is_none());
+    }
+
+    #[test]
+    fn command_carries_globs_and_container_root() {
+        let globs = vec!["model-00047-of-00048.safetensors".to_string()];
+        let spec = ViewSpec {
+            repo: "org/model",
+            local: &globs,
+            verify: "sample-hash",
+            min_free: Some("10G"),
+            force: true,
+        };
+        let cmd = command_for("s0", "apply", "/src", "/dst", &spec);
+        // shq() shell-quotes arguments, so the glob may come back quoted
+        assert!(cmd.contains("model-00047-of-00048.safetensors"), "{cmd}");
+        assert!(cmd.contains("--local"), "{cmd}");
+        assert!(cmd.contains(CONTAINER_ROOT), "{cmd}");
+        assert!(cmd.contains("--min-free '10G'"), "{cmd}");
+        assert!(cmd.ends_with("--force"), "{cmd}");
+    }
+}