staging P1: declarative [[staging]] hub-mirror, ledger-verified

Recipes declare [[staging]] kind="hub-mirror" blocks; serve runs the
hub_mirror.py engine on every launch node before containers start and
mounts the node-local mirror instead of the shared HF cache (dst swaps
the mount source; --mirror combos are rejected with a hint).

Verification is content-based: per-root ledger (dst/hub/.sm121-staging-
ledger.json) holds sampled 1MiB-region hashes; never mtime/size — the
stale-NFS failure class that drove the EXL3 six-generation staging saga
is caught and re-copied (verified: same-size in-place blob mutation was
detected and restored on the next pass). Snapshot symlinks are rewritten
into the mirror; out-of-source link targets materialize as files;
dangling source links abort the pass.

- src/config/staging.rs: the block (kind/src/dst/repos/skip/verify/
  min_free), ledger_id derivation; Recipe.staging field
- src/staging{.rs,/mirror.rs}: orchestration, per-node upload+run,
  StageMode (auto|skip|force), hf_mount serve entry (validates combos,
  runs blocks, returns the mount source)
- src/staging/scripts/hub_mirror.py: the engine (STAGED/DONE/FAIL markers)
- --stage=auto|skip|force in cli; staging.rs serve pass before launch
- docker/mod.rs: hf_mount replaces inline mirror::resolve (net shorter)
- spec + AGENTS.md updated: P1 implemented, P2 declared-but-aborted

E2E on dgx1: copy pass, idempotent verify-only pass, mutation detect,
skip honored, dangling-link abort. 345 tests, all gates.

AGENTS.md

@@ -124,6 +124,24 @@ The mount target stays `/root/.cache/huggingface`, so model path refs and
 repo mounts resolve as usual. Mutually exclusive with `--mirror` (both pick
 the HF mount source — the error says so).
 
+**Declarative staging (`[[staging]]`, spec
+[`specs/tech-architecture/staging-feature.md`](specs/tech-architecture/staging-feature.md)):**
+a recipe declares `[[staging]] kind = "hub-mirror"` blocks and the operator
+runs plain `serve` — the launcher materializes the recipe's model data on
+every launch node **before** containers start, then mounts the node-local
+mirror instead of the shared HF cache (`dst` replaces the `--hf-cache` /
+config source; `--mirror` is rejected when a block declares a mirror).
+Verification is **content-based**: a per-root ledger (`<dst>/hub/.sm121-
+staging-ledger.json`) stores sampled content hashes — never mtime/size (the
+stale-NFS failure class) — so an unchanged repo costs zero copies, a
+same-size mutation in the mirror is detected and re-copied, and the second
+serve is verification-only. `repos` defaults to the recipe's model + repo
+path-refs + composed sources; `skip` leaves external/managed repos alone;
+`min_free` gates the whole pass per node (`G`/`T`/`%`). `--stage=skip`
+assumes staged (CI/known-good), `--stage=force` re-verifies everything.
+`kind = "sparse-rows"` (per-rank Engram staging) is spec P2 — declared
+but not yet executed (serve aborts with the hint).
+
 **Cluster health is cluster-level** (`health`, web dashboard): a cluster is
 UP only when *all* its nodes are reachable AND its **blocking**
 `extra_entities` report `on`. HA entities are probed via HA REST; a

history/night-shifts/2026-09-27.md

@@ -126,3 +126,27 @@ spec — the localization story and this lane converge.
   384,006,168 / 384,016,682 rows, contiguous quarter-partitions per rank,
   ~47.2 GiB/rank per Aiden docs); needs dgx3/4 powered and dgx1/2 freed.
 - Registry go-live + r39 prod cut — operator decisions.
+
+## Staging P1 — hub-mirror in the launcher
+
+Implemented per `specs/tech-architecture/staging-feature.md` (operator
+directive: recipes declare staging, plain `serve` does everything):
+
+- `[[staging]]` recipe block (`src/config/staging.rs`): kind
+  `hub-mirror` (P1) / `sparse-rows` (P2 placeholder, serve aborts with the
+  hint), `id/src/dst/repos/skip/verify/min_free`.
+- Engine `src/staging/scripts/hub_mirror.py`: walks the repo tree, copies
+  real files atomically, materializes out-of-source link targets as regular
+  files, rewrites snapshot symlinks to the mirror, verifies by **sampled
+  content hashes** in a ledger inside the mirror
+  (`dst/hub/.sm121-staging-ledger.json`) — never mtime/size (the stale-NFS
+  class that drove the EXL3 six-generation saga).
+- `src/staging/{rs,mirror.rs}`: per-node parallel upload+run, markers,
+  min-free preflight, dry-run plan; serve integration mounts the mirror's
+  `dst` instead of the shared cache (spec swap) and rejects `--mirror`
+  combination; `--stage=auto|skip|force`.
+- E2E on dgx1 (read-only fixture, Mia untouched): first pass copies, second
+  verifies only (copied=0), same-size mutation of a staged blob was detected
+  and re-copied, `--skip` honored, dangling source link aborted. 345 tests.
+
+Next: migrate EXL3's recipe hook to the declarative block and re-verify.

specs/tech-architecture/staging-feature.md

@@ -1,6 +1,8 @@
 # Staging as a first-class sm121 feature — design & plan
 
-Status: **planned** (spec cut 2026-09-27; nothing implemented yet).
+Status: **P1 implemented** (2026-09-27, `src/staging/` — hub-mirror +
+ledger + `--stage`; P2 sparse-rows is declared-but-aborted; P3/P4 planned).
+Spec cut 2026-09-27.
 Motivation: weight/engram localization of model data onto node-local NVMe is
 now what makes the measured difference (EXL3 2x: 13.3 → 35 t/s decode, pp
 112 → 795 t/s; Aiden SGLang docs: NFS engram reads cost 6–8 ms/step against
@@ -59,6 +61,14 @@ health/web/serve).
 ## Implementation plan
 
 ### P1 — `hub-mirror` in the launcher (`src/staging/`)
+> Implemented 2026-09-27: `src/config/staging.rs` (block), `src/staging.rs`
+> (+ `mirror.rs`, `scripts/hub_mirror.py`), serve integration in
+> `src/docker/mod.rs` (staging pass + cache_dir swap), `--stage` in cli.
+> The script resolves link targets through the hub store and dedups by
+> content sample; ledger lives in the mirror itself (`dst/hub/.sm121-
+staging-ledger.json`) so it travels with the mirror and survives node
+> reinstall of /tmp. EXL3 hook migration is the first real adoption (next
+> session, with a verify re-run).
 - Tree walk (links stay links, real files copied, `skip` pruned), then
   materialize every snapshot link target **resolved through the hub-level
   store** (the lesson that cost the whole six-generation staging saga:

src/after_stop.rs

@@ -222,6 +222,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/before_serve.rs

@@ -390,6 +390,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/cli.rs

@@ -93,6 +93,12 @@ pub enum Commands {
         #[arg(long)]
         hf_cache: Option<String>,
 
+        /// How hard to run the recipe's `[[staging]]` blocks: `auto`
+        /// (ledger-verified, copy what is new), `skip` (assume staged),
+        /// `force` (re-verify everything).
+        #[arg(long)]
+        stage: Option<String>,
+
         /// Scale the recipe to N nodes (`--fit=N`; bare = whole cluster, TP
         /// tracks it; incompatible with --solo).
         #[arg(long, num_args = 0..=1, default_missing_value = "")]

src/config.rs

@@ -52,6 +52,8 @@ impl ExtraEntity {
 mod recipe;
 pub use crate::pre_serve::{BeforeServeTable, DistributedBackend, OnFailure, PreServeCommand};
 pub use recipe::*;
+mod staging;
+pub use staging::{Staging, StagingKind, StagingVerify};
 mod types;
 pub use types::*;
 

src/config/recipe.rs

@@ -2,6 +2,7 @@
 //! [`EnvValue`] — the `[env]` value type: a literal string, or a reference
 //! into a recipe field (`{ from = "port" }`) resolved at serve time.
 
+use crate::config::staging::Staging;
 use crate::config::types::{
     ComposedModel, DeployTarget, ExtraMount, RecipeBuild, default_engine, default_port,
     default_shm_size, default_sync_image, default_tensor_parallel_size,
@@ -200,6 +201,13 @@ pub struct Recipe {
     /// files where the engine opens with O_NOFOLLOW). Docker-only.
     #[serde(default)]
     pub extra_mounts_manifest: Option<String>,
+    /// Declarative data staging executed on the launch nodes before the
+    /// containers start (spec: specs/tech-architecture/staging-feature.md).
+    /// A `[[staging]]` block declares what must exist on node-local storage
+    /// (hub mirror, engram rows); serve stages, verifies, and mounts it.
+    /// See [`Staging`].
+    #[serde(default)]
+    pub staging: Vec<Staging>,
 }
 
 impl Recipe {

src/config/staging.rs

@@ -0,0 +1,130 @@
+//! `[[staging]]` — declarative, per-recipe data staging (spec:
+//! specs/tech-architecture/staging-feature.md). A recipe declares what must
+//! exist on the launch nodes' local storage before the engine starts; the
+//! serve pipeline stages, verifies, and mounts it, so plain `sm121 serve`
+//! is the whole interface. No manual per-node prep steps.
+
+use serde::{Deserialize, Serialize};
+
+/// What kind of staging a block performs.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum StagingKind {
+    /// Mirror an HF hub layout (repo trees + hub blobs) from the shared
+    /// source cache to a node-local mirror. Idempotent by content
+    /// (sample-hash ledger), never by mtime or size.
+    #[default]
+    HubMirror,
+    /// Per-rank sparse copy of Engram table rows at the original safetensors
+    /// byte offsets (spec P2).
+    SparseRows,
+}
+
+/// One `[[staging]]` table in recipe.toml.
+#[derive(Debug, Clone, PartialEq, Default, Deserialize, Serialize)]
+#[serde(default)]
+pub struct Staging {
+    /// Ledger key; must be unique within the recipe. Defaults to the index
+    /// (`s0`, `s1`, ...) when omitted.
+    #[serde(default)]
+    pub id: Option<String>,
+    #[serde(default)]
+    pub kind: StagingKind,
+    /// Source root (hub root for `hub-mirror`, checkpoint dir for
+    /// `sparse-rows`). For hub-mirror it defaults to the serve's shared HF
+    /// cache source (the `--hf-cache` / config dir).
+    #[serde(default)]
+    pub src: Option<String>,
+    /// Destination root on every launch node. `~` expands to the SSH user's
+    /// home. A hub-mirror's dst replaces the shared HF cache mount for the
+    /// duration of the serve.
+    #[serde(default)]
+    pub dst: Option<String>,
+    /// Repos to stage (hub-mirror). Default: everything the recipe needs
+    /// (main model + repo path refs + composed-model sources).
+    #[serde(default)]
+    pub repos: Vec<String>,
+    /// Path components to exclude at any depth (hub-mirror), e.g.
+    /// `["engram-page15-v1"]`.
+    #[serde(default)]
+    pub skip: Vec<String>,
+    /// Verification after staging: `sample-hash` (default) verifies every
+    /// staged file over sampled regions; `presence` checks existence only.
+    #[serde(default)]
+    pub verify: StagingVerify,
+    /// Minimum free space required on the dst filesystem, e.g. `"450G"`.
+    /// `0` = no check.
+    #[serde(default)]
+    pub min_free: Option<String>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum StagingVerify {
+    #[default]
+    SampleHash,
+    Presence,
+}
+
+impl Staging {
+    /// The block's ledger key (caller ensures uniqueness within a recipe).
+    pub fn ledger_id(&self, index: usize) -> String {
+        self.id.clone().unwrap_or_else(|| format!("s{index}"))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parses_full_staging_block() {
+        let toml_str = r#"
+[[staging]]
+id = "exl3-weights"
+kind = "hub-mirror"
+src = "/mnt/nas/ai/huggingface"
+dst = "~/.cache/hf-local"
+repos = ["org/model"]
+skip = ["engram-page15-v1"]
+verify = "sample-hash"
+min_free = "450G"
+"#;
+        let v: toml::Table = toml::from_str(toml_str).unwrap();
+        let blocks: Vec<Staging> = v["staging"].clone().try_into().unwrap();
+        assert_eq!(blocks.len(), 1);
+        let s = &blocks[0];
+        assert_eq!(s.id.as_deref(), Some("exl3-weights"));
+        assert_eq!(s.kind, StagingKind::HubMirror);
+        assert_eq!(s.src.as_deref(), Some("/mnt/nas/ai/huggingface"));
+        assert_eq!(s.dst.as_deref(), Some("~/.cache/hf-local"));
+        assert_eq!(s.repos, vec!["org/model"]);
+        assert_eq!(s.skip, vec!["engram-page15-v1"]);
+        assert_eq!(s.verify, StagingVerify::SampleHash);
+        assert_eq!(s.min_free.as_deref(), Some("450G"));
+    }
+
+    #[test]
+    fn defaults_round_trip() {
+        let toml_str = r#"
+[[staging]]
+kind = "sparse-rows"
+dst = "/var/tmp/engram-local"
+"#;
+        let v: toml::Table = toml::from_str(toml_str).unwrap();
+        let blocks: Vec<Staging> = v["staging"].clone().try_into().unwrap();
+        let s = &blocks[0];
+        assert_eq!(s.kind, StagingKind::SparseRows);
+        assert!(s.id.is_none());
+        assert_eq!(s.ledger_id(0), "s0");
+        assert_eq!(s.verify, StagingVerify::SampleHash);
+        assert!(s.repos.is_empty() && s.skip.is_empty());
+    }
+
+    #[test]
+    fn presence_verify_parses() {
+        let s: Staging =
+            toml::from_str("kind = \"hub-mirror\"\ndst = \"/x\"\nverify = \"presence\"\n").unwrap();
+        assert_eq!(s.verify, StagingVerify::Presence);
+    }
+}

src/docker/build.rs

@@ -348,6 +348,7 @@ pub async fn handle_build_cmd(
         solo: None,
         mirror: None,
         hf_cache: None,
+        stage: None,
         fit: None,
         clear: false,
     };

src/docker/command_tests.rs

@@ -111,9 +111,10 @@ fn base_recipe() -> Recipe {
         sync_image: true,
         docker_image_dir: None,
         extra_mounts: vec![],
-            extra_mounts_manifest: None,
+        extra_mounts_manifest: None,
         build: Default::default(),
         composed_model: None,
+        staging: vec![],
     }
 }
 

src/docker/launch_node.rs

@@ -104,22 +104,30 @@ pub(crate) async fn launch_node(
     // a half-staged node fails loudly instead of starting with half the model.
     if let Some(manifest_path) = recipe.extra_mounts_manifest.clone() {
         if dry {
-            tracing::info!("[dry-run] would read mount manifest {} on {}", manifest_path, node.name);
+            tracing::info!(
+                "[dry-run] would read mount manifest {} on {}",
+                manifest_path,
+                node.name
+            );
         } else {
-            let out =
-                match crate::ssh::run_ssh(config, node, format!("cat {}", shq(&manifest_path)), false)
-                    .await
-                {
-                    Ok(o) => o,
-                    Err(e) => {
-                        return fail(anyhow::anyhow!(
-                            "mount manifest {} not readable on {}: {}",
-                            manifest_path,
-                            node.name,
-                            e
-                        ))
-                    }
-                };
+            let out = match crate::ssh::run_ssh(
+                config,
+                node,
+                format!("cat {}", shq(&manifest_path)),
+                false,
+            )
+            .await
+            {
+                Ok(o) => o,
+                Err(e) => {
+                    return fail(anyhow::anyhow!(
+                        "mount manifest {} not readable on {}: {}",
+                        manifest_path,
+                        node.name,
+                        e
+                    ));
+                }
+            };
             let mut count = 0;
             for line in out.lines() {
                 let line = line.trim();
@@ -133,7 +141,7 @@ pub(crate) async fn launch_node(
                             "bad mount manifest line on {} (expected host<TAB>container): {}",
                             node.name,
                             line
-                        ))
+                        ));
                     }
                 };
                 count += 1;
@@ -151,7 +159,12 @@ pub(crate) async fn launch_node(
                     node.name
                 ));
             }
-            tracing::info!("mount manifest {} on {}: {} mounts", manifest_path, node.name, count);
+            tracing::info!(
+                "mount manifest {} on {}: {} mounts",
+                manifest_path,
+                node.name,
+                count
+            );
         }
     }
     let ctx = command::DockerCommandCtx {

src/docker/mirror.rs

@@ -212,6 +212,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/docker/mod.rs

@@ -51,6 +51,9 @@ pub struct ServeOpts {
     /// the config `hf_cache_dir` default. Mutually exclusive with `mirror`
     /// (both pick the HF mount source).
     pub hf_cache: Option<String>,
+    /// Staging aggressiveness (`--stage=auto|skip|force`); applies to the
+    /// recipe's `[[staging]]` blocks (see `crate::staging`).
+    pub stage: Option<String>,
     /// CLI override for recipe scaling: `Some("")` = bare `--fit` → scale to
     /// the whole selected cluster; `Some("N")` = scale to N nodes;
     /// `None` = unset. See `docker::serve`.
@@ -269,22 +272,18 @@ pub async fn serve(
         }
     }
 
-    // `--mirror`: mount a per-node local mirror of the HF cache instead of
-    // the shared one (first serve copies the model per node; later serves
-    // load it locally). See mirror.rs.
-    if opts.mirror.is_some() && opts.hf_cache.is_some() {
-        anyhow::bail!(
-            "--hf-cache and --mirror are mutually exclusive (both pick the HF mount source)"
-        );
-    }
-    let (mirror, cache_dir) =
-        mirror::resolve(config, opts.mirror.as_deref(), opts.hf_cache.as_deref());
-    if let Some(ref m) = mirror {
-        info!(
-            "serving from local HF mirror {} (models copied per node on first use)",
-            m.display()
-        );
-    }
+    // HF mount source: --mirror/--hf-cache resolution with the declarative
+    // [[staging]] pass layered on (a hub-mirror block takes over the mount).
+    let (mirror, cache_dir) = crate::staging::hf_mount(
+        config,
+        &cluster.nodes,
+        &recipe,
+        opts.mirror.as_deref(),
+        opts.hf_cache.as_deref(),
+        opts.stage.as_deref(),
+        opts.dry_run,
+    )
+    .await?;
     start_cluster(
         config,
         &cluster,

src/docker/sglang_tests.rs

@@ -37,9 +37,10 @@ fn sglang_recipe() -> Recipe {
         sync_image: true,
         docker_image_dir: None,
         extra_mounts: vec![],
-            extra_mounts_manifest: None,
+        extra_mounts_manifest: None,
         build: Default::default(),
         composed_model: None,
+        staging: vec![],
     }
 }
 

src/docker/stop.rs

@@ -49,20 +49,14 @@ pub fn remote_command(recipe_name: &str, cluster_name: &str, rank: u32, wait: bo
 /// `--cluster`): the only cluster whose node set is a superset of the given
 /// names. Ambiguous (nodes spread across several clusters) → error naming
 /// the candidates.
-fn cluster_for_node(
-    config: &Config,
-    nodes: &[String],
-) -> anyhow::Result<Cluster> {
+fn cluster_for_node(config: &Config, nodes: &[String]) -> anyhow::Result<Cluster> {
     let mut matches: Vec<&Cluster> = config
         .clusters
         .iter()
         .filter(|c| nodes.iter().all(|n| c.nodes.iter().any(|cn| cn.name == *n)))
         .collect();
     match matches.len() {
-        0 => anyhow::bail!(
-            "Node(s) {} not found in any cluster",
-            nodes.join(", ")
-        ),
+        0 => anyhow::bail!("Node(s) {} not found in any cluster", nodes.join(", ")),
         1 => Ok(matches.remove(0).clone()),
         n => anyhow::bail!(
             "Nodes {} are in {} clusters ({}); add --cluster",
@@ -154,32 +148,33 @@ pub async fn handle(
     power_off: bool,
     dry_run: bool,
 ) -> Result<(), anyhow::Error> {
-    let (cluster, nodes): (crate::config::Cluster, Vec<String>) = if !cluster_name.is_empty() && cluster_name != "auto" {
-        // Explicit cluster selection (exact or prefix): stop that cluster's
-        // deployment directly — no discovery, no multi-deployment ambiguity.
-        let cluster = crate::config::select_cluster_by_name(config, cluster_name)?;
-        (cluster, nodes.to_vec())
-    } else if !nodes.is_empty() {
-        // --node without --cluster: find the cluster that contains the node.
-        let cluster = cluster_for_node(config, nodes)?;
-        (cluster, nodes.to_vec())
-    } else {
-        let variants =
-            crate::docker::deploy::discover_recipe_deployments(config, recipe_name).await?;
-        if variants.is_empty() {
-            anyhow::bail!("No running containers found for recipe '{}'", recipe_name);
-        }
-        if variants.len() == 1 {
-            let v = &variants[0];
-            let cluster = crate::config::select_cluster_by_name(config, &v.cluster)?;
-            (cluster, v.nodes.clone())
+    let (cluster, nodes): (crate::config::Cluster, Vec<String>) =
+        if !cluster_name.is_empty() && cluster_name != "auto" {
+            // Explicit cluster selection (exact or prefix): stop that cluster's
+            // deployment directly — no discovery, no multi-deployment ambiguity.
+            let cluster = crate::config::select_cluster_by_name(config, cluster_name)?;
+            (cluster, nodes.to_vec())
+        } else if !nodes.is_empty() {
+            // --node without --cluster: find the cluster that contains the node.
+            let cluster = cluster_for_node(config, nodes)?;
+            (cluster, nodes.to_vec())
         } else {
-            anyhow::bail!(crate::docker::deploy::multi_deployment_error(
-                recipe_name,
-                &variants
-            ));
-        }
-    };
+            let variants =
+                crate::docker::deploy::discover_recipe_deployments(config, recipe_name).await?;
+            if variants.is_empty() {
+                anyhow::bail!("No running containers found for recipe '{}'", recipe_name);
+            }
+            if variants.len() == 1 {
+                let v = &variants[0];
+                let cluster = crate::config::select_cluster_by_name(config, &v.cluster)?;
+                (cluster, v.nodes.clone())
+            } else {
+                anyhow::bail!(crate::docker::deploy::multi_deployment_error(
+                    recipe_name,
+                    &variants
+                ));
+            }
+        };
     stop_cluster(config, &cluster, recipe_name, &nodes, wait, dry_run).await?;
     // the stop is verified (every node ran the verifying command) — close
     // the recipe's open run in the ledger

src/docker/sync.rs

@@ -46,11 +46,7 @@ async fn get_image_created(config: &Config, node: &Node, image: &str) -> Option<
 /// from the registry over the internet; the SSH timeout then kills the pull
 /// while the serve still reports "started" and waits out the readiness
 /// timeout for a container that never loads.
-pub async fn ensure_prebuilt_image(
-    config: &Config,
-    cluster: &Cluster,
-    image: &str,
-) -> Result<()> {
+pub async fn ensure_prebuilt_image(config: &Config, cluster: &Cluster, image: &str) -> Result<()> {
     sync_image_to_nodes(config, cluster, image, None).await.map_err(|e| {
         anyhow::anyhow!(
             "image '{}' not available on the launch nodes: {e} \u{2014} pull it on one node first (or build it with --build-node)",

src/docker/tests.rs

@@ -77,9 +77,10 @@ fn trec() -> Recipe {
         sync_image: true,
         docker_image_dir: None,
         extra_mounts: vec![],
-            extra_mounts_manifest: None,
+        extra_mounts_manifest: None,
         build: Default::default(),
         composed_model: None,
+        staging: vec![],
     }
 }
 

src/doctor.rs

@@ -390,6 +390,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/engine_flags.rs

@@ -144,6 +144,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/main.rs

@@ -37,6 +37,7 @@ mod registry;
 mod report;
 mod runlock;
 mod ssh;
+mod staging;
 mod stamp;
 mod tmux;
 mod web;
@@ -92,6 +93,7 @@ async fn main() -> Result<()> {
             solo,
             mirror,
             hf_cache,
+            stage,
             fit,
             clear,
             no_push,
@@ -116,6 +118,7 @@ async fn main() -> Result<()> {
                     solo,
                     mirror,
                     hf_cache,
+                    stage,
                     fit,
                     clear,
                 },

src/staging.rs

@@ -0,0 +1,287 @@
+//! Declarative data staging — see specs/tech-architecture/staging-feature.md.
+//!
+//! A recipe declares `[[staging]]` blocks; `serve` executes them on every
+//! launch node before containers start, then (for hub-mirrors) serves from
+//! the local mirror instead of the shared HF cache. No manual per-node
+//! preparation is required — the operator runs plain `sm121 serve`.
+
+pub mod mirror;
+
+use crate::config::{Config, Node, Recipe, StagingKind, expand_tilde};
+use anyhow::{Result, bail};
+use std::path::{Path, PathBuf};
+use tracing::{info, warn};
+
+/// How aggressively to stage (`--stage=auto|skip|force`).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum StageMode {
+    /// Verify staged content against the ledger, copy what is new (default).
+    Auto,
+    /// Assume staging is already done; only print the plan (CI / known-good).
+    Skip,
+    /// Re-verify everything regardless of the ledger.
+    Force,
+}
+
+impl StageMode {
+    pub fn parse(s: &str) -> Result<Self> {
+        match s.trim().to_ascii_lowercase().as_str() {
+            "" | "auto" => Ok(StageMode::Auto),
+            "skip" => Ok(StageMode::Skip),
+            "force" => Ok(StageMode::Force),
+            other => bail!("unknown --stage value '{other}' (auto|skip|force)"),
+        }
+    }
+}
+
+/// 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 {
+    recipe
+        .staging
+        .iter()
+        .any(|s| s.kind == StagingKind::HubMirror)
+}
+
+/// The hub-mirror dst (tilde-expanded) serve must mount instead of the
+/// shared HF cache. Multiple hub-mirror blocks must share one dst; blocks
+/// without an explicit `dst` use `~/.cache/hf-staging/<id>` — identical to
+/// the default `stage_node` picks.
+pub fn hub_mirror_dst(config: &Config, recipe: &Recipe) -> Option<PathBuf> {
+    let mut out: Option<PathBuf> = None;
+    for (i, s) in recipe.staging.iter().enumerate() {
+        if s.kind != StagingKind::HubMirror {
+            continue;
+        }
+        let d = s
+            .dst
+            .clone()
+            .unwrap_or_else(|| format!("~/.cache/hf-staging/{}", s.ledger_id(i)));
+        let p = PathBuf::from(expand_tilde(&d, &config.ssh_user));
+        if let Some(prev) = &out
+            && *prev != p
+        {
+            warn!(
+                "multiple hub-mirror staging blocks with different dst — mounting {}",
+                prev.display()
+            );
+        }
+        if out.is_none() {
+            out = Some(p);
+        }
+    }
+    out
+}
+
+/// Serve-side staging pass: validate the CLI knobs, run the recipe's
+/// `[[staging]]` blocks on every launch node, and return the HF mount
+/// source for this serve: `(mirror, cache_dir)`. A hub-mirror block takes
+/// over the mount (`mirror=None`, `cache_dir=dst`); otherwise the standard
+/// `mirror::resolve` applies (`--mirror` / `--hf-cache`).
+pub async fn hf_mount(
+    config: &Config,
+    nodes: &[Node],
+    recipe: &Recipe,
+    mirror_cli: Option<&str>,
+    hf_cache_cli: Option<&str>,
+    stage_cli: Option<&str>,
+    dry_run: bool,
+) -> Result<(Option<PathBuf>, PathBuf)> {
+    if mirror_cli.is_some() && hf_cache_cli.is_some() {
+        anyhow::bail!(
+            "--hf-cache and --mirror are mutually exclusive (both pick the HF mount source)"
+        );
+    }
+    let staged = has_hub_mirror(recipe);
+    if staged && mirror_cli.is_some() {
+        anyhow::bail!(
+            "--mirror and a recipe [[staging]] hub-mirror cannot be combined — declare the mirror in the recipe"
+        );
+    }
+    if staged && hf_cache_cli.is_some() {
+        anyhow::bail!(
+            "--hf-cache is ignored for a hub-mirror staging recipe (dst wins); drop one of them"
+        );
+    }
+    if !recipe.staging.is_empty() {
+        let mode = stage_cli
+            .map(StageMode::parse)
+            .transpose()?
+            .unwrap_or(StageMode::Auto);
+        let shared_source = match hf_cache_cli {
+            Some(p) => PathBuf::from(expand_tilde(p, &config.ssh_user)),
+            None => config.resolved_hf_cache_dir(),
+        };
+        run(config, nodes, recipe, &shared_source, mode, dry_run).await?;
+    }
+    if staged {
+        let dst = hub_mirror_dst(config, recipe)
+            .ok_or_else(|| anyhow::anyhow!("hub-mirror staging block has no dst"))?;
+        return Ok((None, dst));
+    }
+    let (mirror, cache_dir) = crate::docker::mirror::resolve(config, mirror_cli, hf_cache_cli);
+    if let Some(ref m) = mirror {
+        info!(
+            "serving from local HF mirror {} (models copied per node on first use)",
+            m.display()
+        );
+    }
+    Ok((mirror, cache_dir))
+}
+
+/// Run every `[[staging]]` block on every launch node (blocks per node run
+/// in sequence, nodes in parallel). `shared_source` is the host-side HF cache
+/// a hub-mirror defaults to copy FROM (the `--hf-cache` / config dir).
+pub async fn run(
+    config: &Config,
+    nodes: &[Node],
+    recipe: &Recipe,
+    shared_source: &Path,
+    mode: StageMode,
+    dry_run: bool,
+) -> Result<()> {
+    if recipe.staging.is_empty() {
+        return Ok(());
+    }
+    if mode == StageMode::Skip {
+        info!("staging: --stage=skip — assuming staged");
+        return Ok(());
+    }
+    // Validate ids before touching any node.
+    let mut seen: Vec<String> = Vec::new();
+    for (i, s) in recipe.staging.iter().enumerate() {
+        let id = s.ledger_id(i);
+        if seen.contains(&id) {
+            bail!("duplicate [[staging]] id '{id}' in recipe");
+        }
+        seen.push(id);
+    }
+
+    let mut handles = Vec::new();
+    for node in nodes {
+        let config = config.clone();
+        let node = node.clone();
+        let recipe = recipe.clone();
+        let source = shared_source.to_path_buf();
+        handles.push(crate::locale::spawn_with_locale(async move {
+            let ctx = mirror::Ctx {
+                config: &config,
+                recipe: &recipe,
+                shared_source: &source,
+                mode,
+                dry_run,
+            };
+            for (i, block) in recipe.staging.iter().enumerate() {
+                match block.kind {
+                    StagingKind::HubMirror => mirror::stage_node(&ctx, &node, block, i).await?,
+                    StagingKind::SparseRows => bail!(
+                        "[[staging]] kind=\"sparse-rows\" is not implemented yet \
+                         (staging spec P2) — id '{}'",
+                        block.ledger_id(i)
+                    ),
+                }
+            }
+            Ok(())
+        }));
+    }
+    for h in handles {
+        h.await??;
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::{DeployTarget, DistributedBackend, Staging};
+    use crate::pre_serve::{RecipeArgs, VenvSetup};
+    use std::collections::HashMap;
+
+    fn cfg() -> Config {
+        Config {
+            ssh_user: "u".into(),
+            ssh_key_path: "/k".into(),
+            clusters: vec![],
+            nccl: HashMap::new(),
+            hf_cache_dir: None,
+            init_port: 25000,
+            before_serve: HashMap::new(),
+            exec_aliases: HashMap::new(),
+            home_assistant: None,
+            mikrotik: None,
+            global_mods_dir: None,
+            drop_caches: true,
+            mount_aliases: HashMap::new(),
+            recipe_sources: Vec::new(),
+            docker_image_sources: Vec::new(),
+            cluster_sources: Vec::new(),
+            registry: None,
+        }
+    }
+
+    fn trec() -> Recipe {
+        Recipe {
+            build_profile: None,
+            parent_model: None,
+            archived: false,
+            engine: "vllm".into(),
+            name: "T".into(),
+            docker_image: "img".into(),
+            model: "org/main".into(),
+            port: 8000,
+            required_nodes: Some(1),
+            pipeline_parallel_size: Some(1),
+            tensor_parallel_size: 1,
+            args: RecipeArgs::List(vec![]),
+            env: HashMap::new(),
+            cluster: None,
+            entrypoint: vec![],
+            distributed_backend: Some(DistributedBackend::Mp),
+            before_serve: vec![],
+            after_stop: vec![],
+            solo: None,
+            deploy: DeployTarget::Docker,
+            venv: None,
+            venv_setup: VenvSetup::default(),
+            venv_setup_script: None,
+            mods: Default::default(),
+            shm_size: String::new(),
+            devices: vec![],
+            memory_limit: None,
+            cpus: None,
+            sync_image: true,
+            docker_image_dir: None,
+            extra_mounts: vec![],
+            extra_mounts_manifest: None,
+            build: Default::default(),
+            composed_model: None,
+            staging: vec![],
+        }
+    }
+
+    #[test]
+    fn stage_mode_parse() {
+        assert_eq!(StageMode::parse("").unwrap(), StageMode::Auto);
+        assert_eq!(StageMode::parse("auto").unwrap(), StageMode::Auto);
+        assert_eq!(StageMode::parse("SKIP").unwrap(), StageMode::Skip);
+        assert_eq!(StageMode::parse("force").unwrap(), StageMode::Force);
+        assert!(StageMode::parse("nope").is_err());
+    }
+
+    #[test]
+    fn hub_mirror_dst_tilde_and_detection() {
+        let mut r = trec();
+        assert!(!has_hub_mirror(&r));
+        assert!(hub_mirror_dst(&cfg(), &r).is_none());
+        r.staging.push(Staging {
+            kind: StagingKind::HubMirror,
+            dst: Some("~/.cache/hf-local".into()),
+            ..Default::default()
+        });
+        assert!(has_hub_mirror(&r));
+        assert_eq!(
+            hub_mirror_dst(&cfg(), &r),
+            Some(PathBuf::from("/home/u/.cache/hf-local"))
+        );
+    }
+}

src/staging/mirror.rs

@@ -0,0 +1,196 @@
+//! `hub-mirror` staging block executor: uploads `scripts/hub_mirror.py` to
+//! the node and runs it against the shared source. See ../staging.rs for
+//! the orchestration and specs/tech-architecture/staging-feature.md for the
+//! design.
+
+use crate::config::{Config, Node, Recipe, Staging, expand_tilde};
+use crate::docker::mirror::repos_for_recipe;
+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/hub_mirror.py");
+
+/// Remote path the script is uploaded to (per ledger id, so concurrent
+/// recipes cannot clobber each other's copy).
+pub fn script_path(id: &str) -> String {
+    format!("/tmp/sm121-staging-{id}.py")
+}
+
+/// Fully-built command line for `script_path(id)`. Paths must already be
+/// host-resolved (`~` expanded); they are shell-quoted here.
+pub fn command_for(id: &str, src: &str, dst: &str, spec: &MirrorSpec) -> String {
+    let script = script_path(id);
+    let mut cmd = format!(
+        "python3 {} --src {} --dst {} --verify {}",
+        crate::docker::shq(&script),
+        crate::docker::shq(src),
+        crate::docker::shq(dst),
+        spec.verify
+    );
+    for r in spec.repos {
+        cmd.push_str(&format!(" --repo {}", crate::docker::shq(r)));
+    }
+    for s in spec.skip {
+        cmd.push_str(&format!(" --skip {}", crate::docker::shq(s)));
+    }
+    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
+}
+
+/// Script parameters for one hub-mirror run (borrowed view of the block).
+pub struct MirrorSpec<'a> {
+    pub repos: &'a [String],
+    pub skip: &'a [String],
+    pub verify: &'a str,
+    pub min_free: Option<&'a str>,
+    pub force: bool,
+}
+
+/// Per-serve context shared by every staging block run (built once by
+/// `crate::staging::run`).
+pub struct Ctx<'a> {
+    pub config: &'a Config,
+    pub recipe: &'a Recipe,
+    /// Host-side HF cache the mirrors copy from (`--hf-cache`/config).
+    pub shared_source: &'a Path,
+    pub mode: crate::staging::StageMode,
+    pub dry_run: bool,
+}
+
+/// Stage one block on one node (upload + run, or print the plan in dry-run).
+pub async fn stage_node(ctx: &Ctx<'_>, node: &Node, block: &Staging, index: usize) -> Result<()> {
+    let config = ctx.config;
+    let mode = ctx.mode;
+    let dry_run = ctx.dry_run;
+    let recipe = ctx.recipe;
+    let id = block.ledger_id(index);
+    let src = match &block.src {
+        Some(s) => expand_tilde(s, &config.ssh_user),
+        None => ctx.shared_source.display().to_string(),
+    };
+    let dst = block
+        .dst
+        .as_deref()
+        .map(|d| expand_tilde(d, &config.ssh_user))
+        .unwrap_or_else(|| format!("~/.cache/hf-staging/{id}"));
+    let repos: Vec<String> = if block.repos.is_empty() {
+        repos_for_recipe(recipe)
+    } else {
+        block.repos.clone()
+    };
+    let force = mode == crate::staging::StageMode::Force;
+
+    if dry_run {
+        info!(
+            "dry-run staging[{}] on {}: mirror {} repo(s) {} -> {} (skip {:?}, verify {:?}, min_free {:?})",
+            id,
+            node.name,
+            repos.len(),
+            repos.join(", "),
+            dst,
+            block.skip,
+            block.verify,
+            block.min_free
+        );
+        return Ok(());
+    }
+
+    let remote = script_path(&id);
+    upload_script(config, node, SCRIPT_BODY, &remote, false).await?;
+    let spec = MirrorSpec {
+        repos: &repos,
+        skip: &block.skip,
+        verify: verify_str(block),
+        min_free: block.min_free.as_deref(),
+        force,
+    };
+    let cmd = command_for(&id, &src, &dst, &spec);
+    let out = run_ssh_long_running(config, node, cmd, false).await?;
+    if !out.contains("STAGING_DONE") {
+        let tail: String = out
+            .lines()
+            .rev()
+            .take(3)
+            .collect::<Vec<_>>()
+            .join(" | ")
+            .chars()
+            .take(300)
+            .collect();
+        bail!("staging[{}] failed on {}: {}", id, node.name, tail);
+    }
+    for line in out.lines().filter(|l| l.starts_with("STAGED")) {
+        info!("{}: staging[{}] {}", node.name, id, line);
+    }
+    Ok(())
+}
+
+fn verify_str(block: &Staging) -> &'static str {
+    match block.verify {
+        crate::config::StagingVerify::SampleHash => "sample-hash",
+        crate::config::StagingVerify::Presence => "presence",
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn command_assembles_all_flags() {
+        let repos = ["models--org--m".to_string()];
+        let skip = ["engram-page15-v1".to_string()];
+        let c = command_for(
+            "weights",
+            "/mnt/nas/hf",
+            "/home/k/hf-local",
+            &MirrorSpec {
+                repos: &repos,
+                skip: &skip,
+                verify: "sample-hash",
+                min_free: Some("450G"),
+                force: true,
+            },
+        );
+        assert!(c.starts_with("python3 '/tmp/sm121-staging-weights.py'"));
+        assert!(c.contains("--src '/mnt/nas/hf'"));
+        assert!(c.contains("--dst '/home/k/hf-local'"));
+        assert!(c.contains("--repo 'models--org--m'"));
+        assert!(c.contains("--skip 'engram-page15-v1'"));
+        assert!(c.contains("--min-free '450G'"));
+        assert!(c.contains("--verify sample-hash"));
+        assert!(c.ends_with("--force"));
+    }
+
+    #[test]
+    fn command_minimal_no_force_no_skip() {
+        let repos = ["r".to_string()];
+        let c = command_for(
+            "s0",
+            "/s",
+            "/d",
+            &MirrorSpec {
+                repos: &repos,
+                skip: &[],
+                verify: "presence",
+                min_free: None,
+                force: false,
+            },
+        );
+        assert!(!c.contains("--force"));
+        assert!(!c.contains("--min-free"));
+        assert!(!c.contains("--skip"));
+        assert!(!c.contains("--src '/tmp"));
+    }
+
+    #[test]
+    fn script_path_is_id_scoped() {
+        assert_eq!(script_path("abc"), "/tmp/sm121-staging-abc.py");
+    }
+}

src/staging/scripts/hub_mirror.py

@@ -0,0 +1,271 @@
+#!/usr/bin/env python3
+"""sm121 hub-mirror staging engine (spec: staging-feature.md, P1).
+
+Mirror the parts of a shared HF hub cache a recipe needs (repo trees + hub
+blobs) onto a node-local hub root, keeping the canonical layout so refs/main,
+snapshot trees, and composed-model symlinks resolve unchanged:
+
+    <src>/hub/models--org--model/...  ->  <dst>/hub/models--org--model/...
+
+Idempotency is by CONTENT — never by mtime or by size alone. Stale NFS
+`st_mtime` and size-only "already copied" checks silently skip real copies
+(the failure class of the 2026-09-27 staging saga), so a per-root ledger
+records a sampled content hash per staged file and re-checks the sample
+cheaply. Files whose relative path contains a `--skip` component at any depth
+are left alone (the caller's own binds cover them).
+
+Snapshot symlinks are recreated inside the mirror: link targets under `<src>`
+are staged into their mapped `<dst>` location and the link is rewritten with
+the equivalent relative target; targets outside `<src>` are materialized as
+regular files so the mirror is self-contained.
+
+Markers on stdout (consumed by the launcher):
+    STAGED <repo> links=<n> files=<n> verified=<n> copied=<n> secs=<n>
+    STAGING_DONE | STAGING_FAIL <reason> (exit != 0)
+"""
+import argparse
+import hashlib
+import json
+import os
+import stat
+import sys
+import time
+
+SAMPLE_BLOCK = 1 << 20    # 1 MiB per sampled region
+SAMPLE_REGIONS = 16       # spread over the file -> 8 MiB hashed per big file
+
+
+def fail(msg):
+    print(f"STAGING_FAIL {msg}", flush=True)
+    sys.exit(1)
+
+
+def sample_hash(path):
+    """Digest of 16 spread-out 1 MiB reads (size-salted). Small files hash
+    their tail block; this is exactly the check a ledger skip relies on."""
+    size = os.path.getsize(path)
+    h = hashlib.sha256(str(size).encode())
+    if size <= SAMPLE_BLOCK * SAMPLE_REGIONS:
+        off = 0 if size <= SAMPLE_BLOCK else size - SAMPLE_BLOCK
+        if size:
+            with open(path, "rb") as f:
+                f.seek(off)
+                h.update(f.read(SAMPLE_BLOCK))
+    else:
+        step = (size - SAMPLE_BLOCK) / (SAMPLE_REGIONS - 1)
+        off = 0.0
+        with open(path, "rb") as f:
+            for _ in range(SAMPLE_REGIONS):
+                f.seek(int(off))
+                h.update(f.read(SAMPLE_BLOCK))
+                off += step
+    return h.hexdigest()
+
+
+class Ledger:
+    """Content-addressed check ledger: {dst_path: [size, sample_sha256]}."""
+
+    def __init__(self, path):
+        self.path = path
+        self.data = {}
+        try:
+            with open(path) as f:
+                self.data = json.load(f).get("files", {})
+        except Exception:
+            pass
+
+    def get(self, key):
+        v = self.data.get(key)
+        return tuple(v) if v else None
+
+    def put(self, key, size, sample):
+        self.data[key] = [size, sample]
+
+    def save(self):
+        tmp = self.path + ".tmp"
+        os.makedirs(os.path.dirname(self.path), exist_ok=True)
+        with open(tmp, "w") as f:
+            json.dump({"files": self.data, "ts": int(time.time())}, f)
+        os.replace(tmp, self.path)
+
+
+class Stager:
+    def __init__(self, src, dst, skip, verify, force, ledger):
+        self.src = src
+        self.dst = dst
+        self.skip = set(skip)
+        self.verify = verify
+        self.force = force
+        self.ledger = ledger
+        self.links = self.files = self.verified = self.copied = 0
+
+    def skipped(self, rel_parts):
+        return any(p in self.skip for p in rel_parts)
+
+    def copy_into(self, src_path, dst_path):
+        """Copy src_path -> dst_path via tmp, verify the copy, publish."""
+        size = os.path.getsize(src_path)
+        tmp = dst_path + ".sm121tmp"
+        try:
+            if os.path.exists(tmp):
+                os.remove(tmp)
+            buf = memoryview(bytearray(1 << 22))
+            with open(src_path, "rb") as fin, open(tmp, "wb") as fout:
+                while True:
+                    n = fin.readinto(buf)
+                    if not n:
+                        break
+                    fout.write(buf[:n])
+            if os.path.getsize(tmp) != size:
+                raise RuntimeError("short copy")
+            if self.verify == "sample-hash" and sample_hash(tmp) != sample_hash(src_path):
+                raise RuntimeError("verify failed after copy")
+            os.replace(tmp, dst_path)
+        finally:
+            if os.path.exists(tmp):
+                os.remove(tmp)
+        self.copied += 1
+
+    def ensure_file(self, src_path, dst_path):
+        """Stage src_path into dst_path unless its content is already proven.
+        Returns True if bytes were copied."""
+        size = os.path.getsize(src_path)
+        if not self.force and os.path.exists(dst_path):
+            entry = self.ledger.get(dst_path)
+            if entry and entry[0] == size and entry[1] == sample_hash(dst_path):
+                self.verified += 1
+                return False
+            if entry is None and os.path.getsize(dst_path) == size \
+                    and sample_hash(dst_path) == sample_hash(src_path):
+                # untracked but currently matching source: adopt into ledger
+                self.ledger.put(dst_path, size, sample_hash(dst_path))
+                self.verified += 1
+                return False
+        os.makedirs(os.path.dirname(dst_path), exist_ok=True)
+        self.copy_into(src_path, dst_path)
+        self.ledger.put(dst_path, size, sample_hash(dst_path))
+        self.files += 1
+        return True
+
+    def stage_tree(self, repo_src, repo_dst):
+        """Walk the repo dir (no symlink following); copy files, make dirs.
+        Returns the list of (link_path, rel_from_src_root) entries."""
+        links = []
+        stack = [repo_src]
+        while stack:
+            root = stack.pop()
+            try:
+                names = sorted(os.listdir(root))
+            except OSError as e:
+                fail(f"cannot list {root}: {e}")
+            for name in names:
+                p = os.path.join(root, name)
+                rel = os.path.relpath(p, self.src)
+                parts = rel.split(os.sep)
+                if self.skipped(parts[:-1]):
+                    continue
+                st = os.lstat(p)
+                if stat.S_ISLNK(st.st_mode):
+                    links.append((p, rel))
+                elif stat.S_ISDIR(st.st_mode):
+                    os.makedirs(os.path.join(self.dst, rel), exist_ok=True)
+                    stack.append(p)
+                elif stat.S_ISREG(st.st_mode):
+                    self.ensure_file(p, os.path.join(self.dst, rel))
+        return links
+
+    def stage_links(self, links):
+        for p, rel in links:
+            target = os.path.realpath(p)
+            if not os.path.exists(target):
+                fail(f"dangling link in source: {p}")
+            mapped_rel = os.path.relpath(target, self.src)
+            dst_entry = os.path.join(self.dst, rel)
+            if mapped_rel.startswith(".."):
+                # target outside the source tree: materialize as a file
+                os.makedirs(os.path.dirname(dst_entry), exist_ok=True)
+                if os.path.islink(dst_entry):
+                    os.remove(dst_entry)
+                self.ensure_file(target, dst_entry)
+                continue
+            mapped_dst = os.path.normpath(os.path.join(self.dst, mapped_rel))
+            if os.path.isdir(target):
+                os.makedirs(mapped_dst, exist_ok=True)
+            else:
+                if os.path.abspath(mapped_dst) != os.path.abspath(dst_entry):
+                    self.ensure_file(target, mapped_dst)
+                else:
+                    self.ensure_file(target, mapped_dst)
+            rel_target = os.path.relpath(mapped_dst, os.path.dirname(dst_entry))
+            os.makedirs(os.path.dirname(dst_entry), exist_ok=True)
+            if os.path.islink(dst_entry):
+                if os.readlink(dst_entry) == rel_target:
+                    self.links += 1
+                    continue
+                os.remove(dst_entry)
+            elif os.path.exists(dst_entry):
+                os.remove(dst_entry)
+            try:
+                os.symlink(rel_target, dst_entry)
+            except OSError as e:
+                fail(f"link {dst_entry} -> {rel_target}: {e}")
+            self.links += 1
+
+    def stage_repo(self, repo_dir):
+        repo_src = os.path.join(self.src, "hub", repo_dir)
+        repo_dst = os.path.join(self.dst, "hub", repo_dir)
+        if not os.path.isdir(repo_src):
+            fail(f"source repo missing: {repo_src}")
+        t0 = time.monotonic()
+        self.links = self.files = self.verified = self.copied = 0
+        links = self.stage_tree(repo_src, repo_dst)
+        self.stage_links(links)
+        elapsed = time.monotonic() - t0
+        print(f"STAGED {repo_dir} links={self.links} files={self.files} "
+              f"verified={self.verified} copied={self.copied} secs={elapsed:.0f}", flush=True)
+
+
+def parse_min_free(s):
+    s = s.strip().lower().rstrip("b")
+    mult = 1
+    if s.endswith("g"):
+        mult = 1 << 30
+    elif s.endswith("m"):
+        mult = 1 << 20
+    elif s.endswith("k"):
+        mult = 1 << 10
+    return int(float(s.rstrip("gkm")) * mult)
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--src", required=True, help="shared hub root (…/huggingface)")
+    ap.add_argument("--dst", required=True, help="node-local mirror root")
+    ap.add_argument("--repo", action="append", required=True,
+                    help="models--org--name directory (repeatable)")
+    ap.add_argument("--skip", action="append", default=[],
+                    help="path component ignored at any depth (repeatable)")
+    ap.add_argument("--verify", default="sample-hash", choices=["sample-hash", "presence"])
+    ap.add_argument("--min-free", default="0", help="required free space at dst")
+    ap.add_argument("--force", action="store_true", help="re-verify everything")
+    args = ap.parse_args()
+
+    if not os.path.isdir(args.src):
+        fail(f"source root not readable: {args.src}")
+    os.makedirs(args.dst, exist_ok=True)
+    need = parse_min_free(args.min_free)
+    free = os.statvfs(args.dst).f_bavail * os.statvfs(args.dst).f_frsize
+    if need and free < need:
+        fail(f"insufficient local space at {args.dst}: {free / 2**30:.0f} GiB free, "
+             f"need {need / 2**30:.0f} GiB")
+
+    ledger = Ledger(os.path.join(args.dst, "hub", ".sm121-staging-ledger.json"))
+    stager = Stager(args.src, args.dst, args.skip, args.verify, args.force, ledger)
+    for repo in args.repo:
+        stager.stage_repo(repo)
+    ledger.save()
+    print("STAGING_DONE", flush=True)
+
+
+if __name__ == "__main__":
+    main()

src/tmux/commands.rs

@@ -178,6 +178,7 @@ mod tests {
             extra_mounts_manifest: None,
             build: Default::default(),
             composed_model: None,
+            staging: vec![],
         }
     }
 

src/web/api.rs

@@ -75,6 +75,7 @@ pub(crate) async fn serve(
         solo: req.solo,
         mirror: req.mirror,
         hf_cache: None,
+        stage: None,
         fit: req.fit,
         clear: req.clear,
     };