staging P2: sparse-rows packed Engram — per-rank local shards, ledger-verified

Second staging kind: [[staging]] kind="sparse-rows" packs the launch
rank's slice of the Engram tables onto node-local NVMe before launch, so
the engine's row store reads its partition from local disk instead of two
strided NFS reads per cache miss (the +3-11 ms/step that caps dsv41 lanes
on NAS rows).

- rows split derived: lo=rows*rank//tp, hi=rows*(rank+1)//tp — rank is the
  cluster launch rank, tp the recipe's tensor_parallel_size (else node
  count); row counts come from the checkpoint header, never hardcoded
- packed-bin format (the runtime's read contract, clean-room impl):
  4096 B header (magic/layer/lo/hi/rows/264) + 264 B interleaved records,
  streamed in 64K-row chunks with fsync+rename
- ledger per block in dst: exact expected size + sampled combined-row
  hashes checked against BOTH source and staged records — a corrupted
  packed record is detected and repacked on the next serve
- {{hf}} placeholder in src resolves against the serve's shared cache
- dsv41-flash-4x-mia adopted: staging block + DSV41_PACKED_DIR=/state/engram
  (dst under the mia-state bind); dry-run shows ranks 0..3/tp4

E2E on dgx1 with a synthetic safetensors fixture: pack, verify-only re-
pass, corrupt-record detect+repack, row content round-trip. 350 tests,
all gates.

AGENTS.md

@@ -139,8 +139,14 @@ 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).
+`kind = "sparse-rows"` (spec P2) packs each launch rank's slice of the
+Engram tables onto node-local NVMe (`dst`, ledger of sampled row hashes in
+the dst dir); rank/tp derive from the cluster and recipe — row ranges are
+never hardcoded. The runtime-visible dir must be bind-mounted by the
+recipe's `extra_mounts` (`dst` is host-side). The first staging serve packs
+(~15 min over the NAS source); combine with `--no-wait` if a cold weight
+load also pushes launch past the readiness budget. The aiden16 manifest
+flavor (`SPARK_ENGRAM_DIR`) is pending.
 
 **Cluster health is cluster-level** (`health`, web dashboard): a cluster is
 UP only when *all* its nodes are reachable AND its **blocking**

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

@@ -150,3 +150,23 @@ directive: recipes declare staging, plain `serve` does everything):
   and re-copied, `--skip` honored, dangling source link aborted. 345 tests.
 
 Next: migrate EXL3's recipe hook to the declarative block and re-verify.
+
+## Staging P2 — sparse-rows (packed Engram)
+
+- `src/staging/rows.rs` + `scripts/sparse_rows.py` (clean-room, format =
+  runtime read contract): per-rank packed shards `engram-l<L>-r<r>of<tp>.bin`
+  (4096 B header + 264 B records), rank from cluster launch rank, tp from
+  recipe, ranges derived from the checkpoint header. Ledger of sampled row
+  hashes per block in dst; corrupt-record detection and repack verified.
+- First adoption: `dsv41-flash-4x-mia` — `[[staging]]` block (dst under the
+  mia-state bind, `min_free 55G`) + `DSV41_PACKED_DIR=/state/engram`.
+  Dry-run prints the per-node rank plan correctly (rank 0..3 / tp 4).
+- E2E on dgx1 (synthetic safetensors fixture): pack pass, verify-only
+  second pass, corruption of a packed record detected + repacked; row
+  content round-trip matches the source interleave.
+
+Gotchas noted: first staging serve adds ~15 min packing (NAS source) —
+beyond the 30-min readiness budget if the weight load is also cold, so the
+first live serve should use `--no-wait`; the aiden16 manifest flavor is a
+pending format; packed shards cost ~48 GiB/node (fits, but P3's LRU needs
+to know about them).

recipes/dsv41-flash-4x-mia/recipe.toml

@@ -93,6 +93,12 @@ DSV41_SOURCE = "/root/.cache/huggingface/hub/models--deepseek-ai--DeepSeek-V4.1-
 SGLANG_FLASHINFER_MOE_FUSED_FINALIZE = "0"   # determinism gate
 SGLANG_DSV41_REASONING_EFFORT = "100"
 PYTORCH_CUDA_ALLOC_CONF = "expandable_segments:False"   # True => NaN logits >64-token prefills
+# Packed Engram read path (staging P2): [[staging]] below materializes this
+# node's rank slice at /home/kilork/mia-state/engram, bind-mounted into the
+# container at /state/engram (the mia-state mount). With the files present
+# the row store attaches packed; DSV41_SOURCE stays as the unpacked
+# fallback if a slice is ever missing.
+DSV41_PACKED_DIR = "/state/engram"
 # ---- production-line adapter gates (docs/tp4.md "What runs" / EXTRA_CONTAINER_ENV) ----
 DSV41_INDEXER_CHUNKED = "1"
 SGLANG_DSPARK_FOLDED_SAMPLING = "2"
@@ -160,3 +166,16 @@ NCCL_MAX_NCHANNELS = "8"
 # boot.py's state dir: dspark tables (future), b12x jit/roce caches; per-node host dir
 host = "/home/kilork/mia-state"
 container = "/state"
+
+# ---- staging (spec: specs/tech-architecture/staging-feature.md) ----
+# Per-rank Engram rows packed onto node-local NVMe (staging P2, packed-bin):
+# rows split by launch rank (rank r owns rows r*rows/tp .. (r+1)*rows/tp),
+# derived from the checkpoint header at stage time. First serve packs from
+# the NAS snapshot (~7 min/layer/node, all nodes in parallel); later serves
+# verify sampled row hashes against the ledger and pack only what changed.
+[[staging]]
+kind = "sparse-rows"
+id = "engram"
+src = "{{hf}}/hub/models--deepseek-ai--DeepSeek-V4.1-Flash/snapshots/dba1be0a40aa45a94ad051997016db3960a90277"
+dst = "/home/kilork/mia-state/engram"
+min_free = "55G"

specs/tech-architecture/staging-feature.md

@@ -1,8 +1,8 @@
 # Staging as a first-class sm121 feature — design & plan
 
-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.
+Status: **P1 + P2 packed-bin implemented** (2026-09-27, `src/staging/` —
+hub-mirror + sparse-rows + ledger + `--stage`; manifest-format sparse-rows
+and 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
@@ -84,6 +84,17 @@ staging-ledger.json`) so it travels with the mirror and survives node
   `recipes/dsv41-flash-exl3-2x` keeps `verify = "sample-hash"`.
 
 ### P2 — `sparse-rows` (engram staging for Aiden)
+> Implemented 2026-09-27 in packed-bin form: `src/staging/rows.rs` +
+> `scripts/sparse_rows.py` (clean-room; format = runtime read contract:
+> 4096 B header `<Q magic, layer, lo, hi, rows, 264>` + 264 B interleaved
+> records). Per-node rank comes from the cluster launch rank, tp from the
+> recipe (else node count); row splits derive from the checkpoint header.
+> Ledger per block in the dst dir (size + sampled row hashes vs source).
+> First adoption: `dsv41-flash-4x-mia` (`DSV41_PACKED_DIR=/state/engram`,
+> dst under the mia-state bind). NOTE: first staging serve adds ~15 min of
+> packing to launch — beyond the 30-min readiness budget when combined
+> with a cold NAS weight load; use `--no-wait` on the very first run.
+> The aiden16 manifest flavor (`SPARK_ENGRAM_DIR`) is a pending format.
 - Port of `tools/engram_local.py` semantics to Rust (pread/pwrite ranges,
   fadvise-DONTNEED, sparse files, row-verify vs source).
 - Per-rank partition is **derived by the launcher** (it knows each node's

src/config/staging.rs

@@ -56,6 +56,15 @@ pub struct Staging {
     /// `0` = no check.
     #[serde(default)]
     pub min_free: Option<String>,
+    /// Engram layers to pack (`sparse-rows`; default `1,14` — the DSV4.1
+    /// Engram tables). Row ranges are derived per rank from the checkpoint
+    /// header, never hardcoded.
+    #[serde(default = "default_engram_layers")]
+    pub layers: Vec<u32>,
+}
+
+fn default_engram_layers() -> Vec<u32> {
+    vec![1, 14]
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize, Serialize)]
@@ -127,4 +136,17 @@ dst = "/var/tmp/engram-local"
             toml::from_str("kind = \"hub-mirror\"\ndst = \"/x\"\nverify = \"presence\"\n").unwrap();
         assert_eq!(s.verify, StagingVerify::Presence);
     }
+
+    #[test]
+    fn sparse_rows_defaults_layers() {
+        let s: Staging = toml::from_str(
+            "kind = \"sparse-rows\"\nsrc = \"{{hf}}/snap\"\ndst = \"~/.cache/engram\"\n",
+        )
+        .unwrap();
+        assert_eq!(s.kind, StagingKind::SparseRows);
+        assert_eq!(s.layers, vec![1, 14]);
+        let custom: Staging =
+            toml::from_str("kind = \"sparse-rows\"\nlayers = [1, 7, 14]\n").unwrap();
+        assert_eq!(custom.layers, vec![1, 7, 14]);
+    }
 }

src/staging.rs

@@ -6,6 +6,7 @@
 //! preparation is required — the operator runs plain `sm121 serve`.
 
 pub mod mirror;
+pub mod rows;
 
 use crate::config::{Config, Node, Recipe, StagingKind, expand_tilde};
 use anyhow::{Result, bail};
@@ -157,7 +158,14 @@ pub async fn run(
         seen.push(id);
     }
 
-    let mut handles = Vec::new();
+    let mut handles: Vec<tokio::task::JoinHandle<Result<()>>> = Vec::new();
+    // sparse-rows splits the table across the engine's tensor-parallel
+    // ranks: recipe TP wins, else one rank per launch node.
+    let tp = if recipe.tensor_parallel_size > 0 {
+        recipe.tensor_parallel_size
+    } else {
+        nodes.len() as u32
+    };
     for node in nodes {
         let config = config.clone();
         let node = node.clone();
@@ -174,11 +182,17 @@ 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::SparseRows => bail!(
-                        "[[staging]] kind=\"sparse-rows\" is not implemented yet \
-                         (staging spec P2) — id '{}'",
-                        block.ledger_id(i)
-                    ),
+                    StagingKind::SparseRows => {
+                        let rctx = rows::Ctx {
+                            config: &config,
+                            shared_source: &source,
+                            mode,
+                            dry_run,
+                            rank: node.rank,
+                            tp,
+                        };
+                        rows::stage_node(&rctx, &node, block, i).await?
+                    }
                 }
             }
             Ok(())
@@ -193,6 +207,7 @@ pub async fn run(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::config::Node;
     use crate::config::{DeployTarget, DistributedBackend, Staging};
     use crate::pre_serve::{RecipeArgs, VenvSetup};
     use std::collections::HashMap;
@@ -268,6 +283,48 @@ mod tests {
         assert!(StageMode::parse("nope").is_err());
     }
 
+    #[tokio::test]
+    async fn sparse_rows_dry_run_dispatches() {
+        // dry-run touches no SSH: both nodes just print their plan (rank
+        // 0 and 1 of a 2-rank split) and the pass succeeds — this exercises
+        // the sparse-rows dispatch arm and the tp derivation.
+        let mut r = trec();
+        r.staging.push(Staging {
+            kind: crate::config::StagingKind::SparseRows,
+            src: Some("{{hf}}/hub/snap".into()),
+            dst: Some("~/.cache/engram".into()),
+            ..Default::default()
+        });
+        let nodes = vec![
+            Node {
+                name: "n0".into(),
+                ssh_ip: "10.0.0.1".into(),
+                ib_ip: None,
+                rank: 0,
+                host_key: None,
+                entity: None,
+            },
+            Node {
+                name: "n1".into(),
+                ssh_ip: "10.0.0.2".into(),
+                ib_ip: None,
+                rank: 1,
+                host_key: None,
+                entity: None,
+            },
+        ];
+        crate::staging::run(
+            &cfg(),
+            &nodes,
+            &r,
+            std::path::Path::new("/mnt/nas/hf"),
+            StageMode::Auto,
+            true,
+        )
+        .await
+        .unwrap();
+    }
+
     #[test]
     fn hub_mirror_dst_tilde_and_detection() {
         let mut r = trec();

src/staging/rows.rs

@@ -0,0 +1,185 @@
+//! `sparse-rows` staging block executor (spec P2): packs the per-rank
+//! Engram table rows into a node-local contiguous shard directory so the
+//! engine reads its Engram rows from local NVMe instead of streaming the
+//! two strided tensors over NFS at cache-miss time.
+//!
+//! The row split is derived per node from its launch rank and the engine TP
+//! (`ctx.rank`, `ctx.tp`) — the launcher knows both, so the recipe never
+//! hardcodes ranges. The packed file format (4096 B header + 264 B records)
+//! is the runtime's read contract; see `scripts/sparse_rows.py`.
+
+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/sparse_rows.py");
+
+/// Per-serve context for one node's sparse-rows pass.
+pub struct Ctx<'a> {
+    pub config: &'a Config,
+    /// Host-side HF cache root used to expand `{{hf}}` in `src`.
+    pub shared_source: &'a Path,
+    pub mode: crate::staging::StageMode,
+    pub dry_run: bool,
+    /// This node's launch rank (the table slice it owns).
+    pub rank: u32,
+    /// Engine tensor-parallel size (number of row slices).
+    pub tp: u32,
+}
+
+fn script_path(id: &str) -> String {
+    format!("/tmp/sm121-staging-{id}.py")
+}
+
+/// Resolve a `sparse-rows` `src`: defaults to `{{hf}}/hub/<main-model-snapshot>`
+/// resolution is left to the caller passing the snapshot dir; supports the
+/// `{{hf}}` placeholder for the shared cache root.
+pub fn resolve_src(block: &Staging, ssh_user: &str, shared_source: &Path) -> Option<String> {
+    match &block.src {
+        Some(s) if s.starts_with("{{hf}}") => {
+            Some(format!("{}{}", shared_source.display(), &s[6..]))
+        }
+        Some(s) => Some(expand_tilde(s, ssh_user)),
+        None => None,
+    }
+}
+
+/// Build the `python3 ...` command line for one sparse-rows block.
+pub fn command_for(id: &str, src: &str, dst: &str, rank: u32, tp: u32, spec: &RowsSpec) -> String {
+    let mut cmd = format!(
+        "python3 {} --id {} --model {} --out {} --rank {} --tp {} --layers {}",
+        crate::docker::shq(&script_path(id)),
+        crate::docker::shq(id),
+        crate::docker::shq(src),
+        crate::docker::shq(dst),
+        rank,
+        tp,
+        spec.layers
+            .iter()
+            .map(u32::to_string)
+            .collect::<Vec<_>>()
+            .join(",")
+    );
+    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
+}
+
+pub struct RowsSpec<'a> {
+    pub layers: &'a [u32],
+    pub min_free: Option<&'a str>,
+    pub force: bool,
+}
+
+/// Stage one sparse-rows block on one node.
+pub async fn stage_node(ctx: &Ctx<'_>, node: &Node, block: &Staging, index: usize) -> Result<()> {
+    let id = block.ledger_id(index);
+    let Some(src) = resolve_src(block, &ctx.config.ssh_user, ctx.shared_source) else {
+        bail!(
+            "[[staging]] sparse-rows block '{}' needs an explicit src (checkpoint snapshot dir; {{hf}} expands to the shared cache)",
+            id
+        );
+    };
+    let dst = block
+        .dst
+        .as_deref()
+        .map(|d| expand_tilde(d, &ctx.config.ssh_user))
+        .unwrap_or_else(|| format!("~/.cache/engram/{id}"));
+    let force = ctx.mode == crate::staging::StageMode::Force;
+
+    if ctx.dry_run {
+        info!(
+            "dry-run staging[{}] on {}: pack layers {:?} rank {}/{} -> {} (min_free {:?})",
+            id, node.name, block.layers, ctx.rank, ctx.tp, dst, block.min_free
+        );
+        return Ok(());
+    }
+
+    upload_script(ctx.config, node, SCRIPT_BODY, &script_path(&id), false).await?;
+    let spec = RowsSpec {
+        layers: &block.layers,
+        min_free: block.min_free.as_deref(),
+        force,
+    };
+    let cmd = command_for(&id, &src, &dst, ctx.rank, ctx.tp, &spec);
+    let out = run_ssh_long_running(ctx.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(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn command_assembles_args() {
+        let spec = RowsSpec {
+            layers: &[1, 14],
+            min_free: Some("150G"),
+            force: true,
+        };
+        let c = command_for("engram", "/mnt/nas/snap", "/home/k/engram", 2, 4, &spec);
+        assert!(c.starts_with("python3 '/tmp/sm121-staging-engram.py'"));
+        assert!(c.contains("--model '/mnt/nas/snap'"));
+        assert!(c.contains("--out '/home/k/engram'"));
+        assert!(c.contains("--rank 2"));
+        assert!(c.contains("--tp 4"));
+        assert!(c.contains("--layers 1,14"));
+        assert!(c.contains("--min-free '150G'"));
+        assert!(c.ends_with("--force"));
+    }
+
+    #[test]
+    fn command_defaults_layers_no_minfree() {
+        let spec = RowsSpec {
+            layers: &[1, 14],
+            min_free: None,
+            force: false,
+        };
+        let c = command_for("s0", "/s", "/d", 0, 1, &spec);
+        assert!(c.contains("--layers 1,14"));
+        assert!(!c.contains("--min-free"));
+        assert!(!c.contains("--force"));
+    }
+
+    #[test]
+    fn hf_placeholder_in_src() {
+        let block = Staging {
+            src: Some("{{hf}}/hub/models--x/snapshots/abc".into()),
+            ..Default::default()
+        };
+        assert_eq!(
+            resolve_src(&block, "kilork", Path::new("/mnt/nas/hf")).unwrap(),
+            "/mnt/nas/hf/hub/models--x/snapshots/abc"
+        );
+        let tilde = Staging {
+            src: Some("~/snap".into()),
+            ..Default::default()
+        };
+        assert_eq!(
+            resolve_src(&tilde, "kilork", Path::new("/x")).unwrap(),
+            "/home/kilork/snap"
+        );
+        assert!(resolve_src(&Staging::default(), "kilork", Path::new("/x")).is_none());
+    }
+}

src/staging/scripts/sparse_rows.py

@@ -0,0 +1,256 @@
+#!/usr/bin/env python3
+"""sm121 sparse-rows staging engine (spec: staging-feature.md, P2).
+
+Pack one launch rank's slice of the Engram tables into a node-local
+contiguous shard dir, so the engine reads its rows from local NVMe instead
+of two strided NFS reads per cache miss.
+
+Packed layout (the runtime's read contract):
+  engram-l<layer>-r<rank>of<tp>.bin
+    4096 B header: <Q magic, layer, lo, hi, rows, record_bytes>
+    rows*record_bytes of interleaved records (weight + scale per row)
+
+The row split (lo = rows*rank//tp, hi = rows*(rank+1)//tp) is derived from
+the checkpoint header, never hardcoded. Idempotency: exact expected size
+plus a per-file ledger of sampled combined (weight+scale) row hashes
+checked against the source — mtime/size alone never count.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import struct
+import sys
+import time
+
+MAGIC = 0x31344E4531565344  # "DSV41EN1"
+HEADER_BYTES = 4096
+try:
+    import numpy as np
+
+    WEIGHT_BYTES, SCALE_BYTES = 256, 8
+    RECORD_BYTES = WEIGHT_BYTES + SCALE_BYTES
+except ImportError:  # tiny fallback, numpy not required
+    np = None
+    WEIGHT_BYTES, SCALE_BYTES, RECORD_BYTES = 256, 8, 264
+
+
+def parse_size(s: str) -> int:
+    s = s.strip().upper()
+    mult = 1
+    if s.endswith("G"):
+        mult = 2**30
+    elif s.endswith("T"):
+        mult = 2**40
+    elif s.endswith("M"):
+        mult = 2**20
+    return int(float(s.rstrip("GMT")) * mult)
+
+
+def tensor_span(root: Path, layer: int):
+    """(shard_path, rows, weight_offset, scale_offset) for one Engram layer."""
+    index = json.loads((root / "model.safetensors.index.json").read_text())["weight_map"]
+    prefix = f"layers.{layer}.engram.embed."
+    w_name = index[prefix + "weight"]
+    assert index[prefix + "scale"] == w_name, "weight and scale must share a shard"
+    path = root / w_name
+    with path.open("rb") as f:
+        length = struct.unpack("<Q", f.read(8))[0]
+        header = json.loads(f.read(length))
+    w, s = header[prefix + "weight"], header[prefix + "scale"]
+    assert w["dtype"] == "F8_E4M3" and s["dtype"] == "F8_E8M0", "unexpected dtype"
+    assert w["shape"][1] == WEIGHT_BYTES and s["shape"][1] == SCALE_BYTES
+    base = 8 + length
+    return path, w["shape"][0], base + w["data_offsets"][0], base + s["data_offsets"][0]
+
+
+def read_row(path: Path, w_off: int, s_off: int, row: int) -> bytes:
+    with path.open("rb", buffering=0) as f:
+        f.seek(w_off + row * WEIGHT_BYTES)
+        w = f.read(WEIGHT_BYTES)
+        f.seek(s_off + row * SCALE_BYTES)
+        s = f.read(SCALE_BYTES)
+    return w + s
+
+
+def interleave(weights: bytes, scales: bytes, n: int) -> bytes:
+    if np is not None:
+        buf = np.empty((n, RECORD_BYTES), dtype=np.uint8)
+        buf[:, :WEIGHT_BYTES] = np.frombuffer(weights, dtype=np.uint8).reshape(n, WEIGHT_BYTES)
+        buf[:, WEIGHT_BYTES:] = np.frombuffer(scales, dtype=np.uint8).reshape(n, SCALE_BYTES)
+        return buf.tobytes()
+    out = bytearray(n * RECORD_BYTES)
+    w = memoryview(weights)
+    s = memoryview(scales)
+    for i in range(n):
+        k = i * RECORD_BYTES
+        out[k : k + WEIGHT_BYTES] = w[i * WEIGHT_BYTES : (i + 1) * WEIGHT_BYTES]
+        out[k + WEIGHT_BYTES : k + SCALE_BYTES + WEIGHT_BYTES] = s[i * SCALE_BYTES : (i + 1) * SCALE_BYTES]
+    return bytes(out)
+
+
+def expected_size(rows: int, lo: int, hi: int) -> int:
+    return HEADER_BYTES + (hi - lo) * RECORD_BYTES
+
+
+class Ledger:
+    """Per-block ledger in the output dir: per-file {size, lo, hi, rows,
+    sample: {row: sha256(128B-slice of combined row)}} — content, not mtime."""
+
+    def __init__(self, path: Path):
+        self.path = path
+        self.files = {}
+        if path.exists():
+            self.data = json.loads(path.read_text())
+            self.files = self.data.get("files", {})
+
+    def known(self, name: str, size: int, lo: int, hi: int, rows: int) -> bool:
+        rec = self.files.get(name)
+        return (
+            rec is not None
+            and rec["size"] == size
+            and rec["lo"] == lo
+            and rec["hi"] == hi
+            and rec["rows"] == rows
+        )
+
+    def set(self, name: str, size: int, lo: int, hi: int, rows: int, samples: dict):
+        self.files[name] = {"size": size, "lo": lo, "hi": hi, "rows": rows, "sample": samples}
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.path.write_text(json.dumps({"files": self.files}))
+        self.path.chmod(0o644)
+
+
+def sample_rows(lo: int, hi: int, count: int) -> list:
+    n = hi - lo
+    if n <= count:
+        return list(range(lo, hi))
+    step = n / count
+    return [lo + int(i * step) for i in range(count)]
+
+
+def verify_file(path: Path, src: Path, w_off: int, s_off: int, lo: int, hi: int,
+                size: int, ledger: Ledger, name: str) -> "bool | None":
+    """True staged-ok, False needs repack, None size-only trust (presence)."""
+    if not path.exists() or path.stat().st_size != size:
+        return False
+    rec = ledger.files.get(name)
+    if rec is None or not rec.get("sample"):
+        return None  # size ok, no content samples — presence pass
+    with path.open("rb", buffering=0) as f:
+        for row in sorted(int(k) for k in rec["sample"]):
+            want = rec["sample"][str(row)]
+            h = hashlib.sha256(read_row(src, w_off, s_off, row)).digest()[:16].hex()
+            if want != h:
+                return False  # source row changed
+            f.seek(HEADER_BYTES + (row - lo) * RECORD_BYTES)
+            h2 = hashlib.sha256(f.read(RECORD_BYTES)).digest()[:16].hex()
+            if h != h2:
+                return False  # staged record corrupted
+    return True
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--id", required=True)
+    ap.add_argument("--model", required=True, help="checkpoint snapshot dir")
+    ap.add_argument("--out", required=True, help="node-local output dir")
+    ap.add_argument("--rank", type=int, required=True)
+    ap.add_argument("--tp", type=int, required=True)
+    ap.add_argument("--layers", default="1,14")
+    ap.add_argument("--min-free", default=None)
+    ap.add_argument("--force", action="store_true")
+    ap.add_argument("--verify", default="sample-hash", choices=["sample-hash", "presence"])
+    ap.add_argument("--chunk-rows", type=int, default=1 << 16)
+    ap.add_argument("--samples", type=int, default=128)
+    args = ap.parse_args()
+
+    root, out_dir = Path(args.model), Path(args.out)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    ledger = Ledger(out_dir / f".sm121-staging-{args.id}.json")
+    layers = [int(x) for x in args.layers.split(",") if x.strip()]
+    if args.tp < 1 or not (0 <= args.rank < args.tp):
+        print(f"STAGING_FAIL invalid rank/tp {args.rank}/{args.tp}")
+        return 2
+
+    # min-free gate before any write (sum across layers)
+    need = 0
+    spans = {}
+    for layer in layers:
+        try:
+            path, rows, w_off, s_off = tensor_span(root, layer)
+        except Exception as e:  # noqa: BLE001
+            print(f"STAGING_FAIL cannot read engram layer {layer} of {root}: {e}")
+            return 2
+        lo, hi = rows * args.rank // args.tp, rows * (args.rank + 1) // args.tp
+        spans[layer] = (path, rows, w_off, s_off, lo, hi)
+        need += expected_size(rows, lo, hi)
+    if args.min_free:
+        need_free = parse_size(args.min_free)
+        free = os.statvfs(out_dir).f_bavail * os.statvfs(out_dir).f_frsize
+        if free < need_free:
+            print(f"STAGING_FAIL {out_dir} free {free/2**30:.0f} GiB < min_free {args.min_free}")
+            return 2
+    free = os.statvfs(out_dir).f_bavail * os.statvfs(out_dir).f_frsize
+    if free < need:
+        print(f"STAGING_FAIL need {need/2**30:.1f} GiB, only {free/2**30:.1f} GiB at {out_dir}")
+        return 2
+
+    for layer in layers:
+        path, rows, w_off, s_off, lo, hi = spans[layer]
+        name = f"engram-l{layer}-r{args.rank}of{args.tp}.bin"
+        target = out_dir / name
+        size = expected_size(rows, lo, hi)
+        if not args.force and target.exists():
+            ok = verify_file(target, path, w_off, s_off, lo, hi, size, ledger, name)
+            if ok is True:
+                print(f"STAGED {name} rows={hi-lo} verified")
+                continue
+            if ok is None:
+                print(f"STAGED {name} rows={hi-lo} presence-ok")
+                continue
+        started = time.monotonic()
+        header = bytearray(HEADER_BYTES)
+        struct.pack_into("<6Q", header, 0, MAGIC, layer, lo, hi, rows, RECORD_BYTES)
+        partial = target.with_name(target.name + ".partial")
+        samples: dict = {}
+        rows_written = 0
+        with open(path, "rb", buffering=0) as src, open(partial, "wb", buffering=0) as dst:
+            dst.write(header)
+            for start in range(lo, hi, args.chunk_rows):
+                stop = min(start + args.chunk_rows, hi)
+                n = stop - start
+                src.seek(w_off + start * WEIGHT_BYTES)
+                weights = src.read(n * WEIGHT_BYTES)
+                src.seek(s_off + start * SCALE_BYTES)
+                scales = src.read(n * SCALE_BYTES)
+                if len(weights) != n * WEIGHT_BYTES or len(scales) != n * SCALE_BYTES:
+                    print(f"STAGING_FAIL short read from {path} at row {start}")
+                    partial.unlink(missing_ok=True)
+                    return 2
+                dst.write(interleave(weights, scales, n))
+                rows_written += n
+                if rows_written % (16 * args.chunk_rows) == 0:
+                    rate = rows_written * RECORD_BYTES / max(time.monotonic() - started, 1e-6)
+                    print(f"PACK {name}: {100.0*rows_written/(hi-lo):5.1f}% {rate/2**20:.0f} MiB/s",
+                          flush=True)
+            dst.flush()
+            os.fsync(dst.fileno())
+        partial.rename(target)
+        if args.verify == "sample-hash":
+            for row in sample_rows(lo, hi, args.samples):
+                h = hashlib.sha256(read_row(path, w_off, s_off, row)).digest()[:16].hex()
+                samples[row] = h
+        ledger.set(name, size, lo, hi, rows, samples)
+        secs = time.monotonic() - started
+        print(f"STAGED {name} rows={hi-lo} packed {size/2**30:.1f} GiB in {secs:.0f}s",
+              flush=True)
+    print("STAGING_DONE")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())