staging: sparse-copy format (aiden16 DSV41_ENGRAM_DIR contract) + packed-Mia bench notes

AGENTS.md

@@ -145,8 +145,12 @@ 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.
+load also pushes launch past the readiness budget. `format = "sparse-copy"`
+switches the output from the packed-bin shards to safetensors-pread layout
+— sparse node-local copies of the original shards containing exactly the
+rank's row ranges (holes are free) + `engram-local.json` row-range
+manifest: the aiden16/DSV41_ENGRAM_DIR contract. Sample-hash verification
+also **adopts** a manually staged tree without re-copying.
 
 **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

@@ -215,3 +215,30 @@ filled the node, both passes cleaned up, disk back to 795G/75G).
 Staging verified per node (dgx1/dgx2): 0 copies of page15, 133+10 export
 hardlinks, 28 manifest lines. Serve verification is for the lane cutover
 (Mia holds all four nodes during the bench).
+
+## Bench packed-Mia — 161/184 (88/100, ★★★★)
+
+Full `tool-eval-bench` (seed 42, hardmode, perf conc 1,2, harness
+v2.7.1.dev7) against the packed-Engram lane:
+`recipes/dsv41-flash-4x-mia/reports/2026-09-27_114508_packed-engram.*`
+(md + json + recipe copy + manifest). Score 161/184 = 88/100 vs the EXL3
+localized 169/184 (92/100) — different harness era, so the delta is not
+pure-Engram (the quality-gate pool changed between dev65-era and v2.7.x;
+re-baselining the EXL3 anchor under v2.7.1 would be the clean read).
+Perf confirms the staging thesis: **tg c1 59-88 t/s** (unpacked lane
+floored at 29; the packed decode probe read 71.9-74.0), pp 3.2-4.8K t/s.
+Responsiveness median 1.4 s, Deployability 84, two safety-critical failures
+to eyeball on the next review pass.
+
+## Staging P2.5 — sparse-copy format (aiden16 `DSV41_ENGRAM_DIR`)
+
+`[[staging]] kind="sparse-rows", format="sparse-copy"` — same ledger/rank
+machinery, different output: sparse node-local copies of the original
+safetensors shards containing exactly the rank's row ranges (holes free,
+size must match source — the loader mmaps and preads at original offsets)
++ `engram-local.json` row-range manifest + copied index. Sample-hash
+verification adopts existing trees (manual `tools/engram_local.py` staging)
+without re-copying, and rebuilds ranges when they don't verify. E2E on
+dgx1 (synthetic fixture, rank1/tp4): pack, verify-only, adoption,
+corrupt-one-row rebuild — all green; sparse-ness confirmed (528 K logical,
+147 K allocated).

specs/tech-architecture/staging-feature.md

@@ -1,8 +1,9 @@
 # Staging as a first-class sm121 feature — design & plan
 
-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.
+Status: **P1 + P2 implemented (packed-bin and sparse-copy)** (2026-09-27,
+`src/staging/` — hub-mirror + sparse-rows (+`format = "sparse-copy"` for the
+aiden16 `DSV41_ENGRAM_DIR` safetensors-pread layout) + ledger + `--stage`;
+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
@@ -94,7 +95,15 @@ staging-ledger.json`) so it travels with the mirror and survives node
 > 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.
+> **Implemented 2026-09-27:** `format = "sparse-copy"` produces the aiden16
+> `DSV41_ENGRAM_DIR` layout — sparse node-local shard copies at the original
+> byte offsets holding exactly the rank's row ranges, plus
+> `engram-local.json` (`{"layers": {"<L>": [lo, hi]}}`). Sample-hash
+> verification adopts a manually staged tree (`tools/engram_local.py`
+> output) into the ledger instead of re-copying it; corrupted rows trigger
+> a range rebuild; untracked manual trees that don't verify are rebuilt
+> into the declarative format. E2E (fixture, rank1/tp4): pack, verify-only,
+> adoption, corrupt-rebuild all green.
 - 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.rs

@@ -53,7 +53,7 @@ mod recipe;
 pub use crate::pre_serve::{BeforeServeTable, DistributedBackend, OnFailure, PreServeCommand};
 pub use recipe::*;
 mod staging;
-pub use staging::{Staging, StagingKind, StagingVerify};
+pub use staging::{Staging, StagingFormat, StagingKind, StagingVerify};
 mod types;
 pub use types::*;
 

src/config/staging.rs

@@ -67,6 +67,20 @@ pub struct Staging {
     /// header, never hardcoded.
     #[serde(default = "default_engram_layers")]
     pub layers: Vec<u32>,
+    /// Output format (`sparse-rows`): `packed` (264 B record shards, the
+    /// row-store `DSV41_PACKED_DIR` contract) or `sparse-copy` (node-local
+    /// dir with `engram-local.json` + sparse shard copies at the original
+    /// byte offsets — the `DSV41_ENGRAM_DIR`/aiden16 contract).
+    #[serde(default)]
+    pub format: StagingFormat,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub enum StagingFormat {
+    #[default]
+    Packed,
+    SparseCopy,
 }
 
 fn default_engram_layers() -> Vec<u32> {
@@ -163,5 +177,10 @@ dst = "/var/tmp/engram-local"
         let custom: Staging =
             toml::from_str("kind = \"sparse-rows\"\nlayers = [1, 7, 14]\n").unwrap();
         assert_eq!(custom.layers, vec![1, 7, 14]);
+        assert_eq!(custom.format, StagingFormat::Packed);
+        let sc: Staging =
+            toml::from_str("kind = \"sparse-rows\"\nformat = \"sparse-copy\"\nsrc = \"/s\"\n")
+                .unwrap();
+        assert_eq!(sc.format, StagingFormat::SparseCopy);
     }
 }

src/staging/rows.rs

@@ -8,7 +8,7 @@
 //! 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::config::{Config, Node, Staging, StagingFormat, expand_tilde};
 use crate::ssh::{run_ssh_long_running, upload_script};
 use anyhow::{Result, bail};
 use std::path::Path;
@@ -62,6 +62,9 @@ pub fn command_for(id: &str, src: &str, dst: &str, rank: u32, tp: u32, spec: &Ro
             .collect::<Vec<_>>()
             .join(",")
     );
+    if spec.sparse_copy {
+        cmd.push_str(" --format sparse-copy");
+    }
     if let Some(m) = spec.min_free {
         cmd.push_str(&format!(" --min-free {}", crate::docker::shq(m)));
     }
@@ -75,6 +78,9 @@ pub struct RowsSpec<'a> {
     pub layers: &'a [u32],
     pub min_free: Option<&'a str>,
     pub force: bool,
+    /// `sparse-copy` appends `--format sparse-copy` (safetensors-pread
+    /// contract); `packed` stays the script default.
+    pub sparse_copy: bool,
 }
 
 /// Stage one sparse-rows block on one node.
@@ -95,8 +101,19 @@ pub async fn stage_node(ctx: &Ctx<'_>, node: &Node, block: &Staging, index: usiz
 
     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
+            "dry-run staging[{}] on {}: pack layers {:?} ({}) rank {}/{} -> {} (min_free {:?})",
+            id,
+            node.name,
+            block.layers,
+            if block.format == StagingFormat::SparseCopy {
+                "sparse-copy"
+            } else {
+                "packed"
+            },
+            ctx.rank,
+            ctx.tp,
+            dst,
+            block.min_free
         );
         return Ok(());
     }
@@ -106,6 +123,7 @@ pub async fn stage_node(ctx: &Ctx<'_>, node: &Node, block: &Staging, index: usiz
         layers: &block.layers,
         min_free: block.min_free.as_deref(),
         force,
+        sparse_copy: block.format == StagingFormat::SparseCopy,
     };
     let cmd = command_for(&id, &src, &dst, ctx.rank, ctx.tp, &spec);
     let out = run_ssh_long_running(ctx.config, node, cmd, false).await?;
@@ -137,6 +155,7 @@ mod tests {
             layers: &[1, 14],
             min_free: Some("150G"),
             force: true,
+            sparse_copy: false,
         };
         let c = command_for("engram", "/mnt/nas/snap", "/home/k/engram", 2, 4, &spec);
         assert!(c.starts_with("python3 '/tmp/sm121-staging-engram.py'"));
@@ -147,6 +166,19 @@ mod tests {
         assert!(c.contains("--layers 1,14"));
         assert!(c.contains("--min-free '150G'"));
         assert!(c.ends_with("--force"));
+        assert!(!c.contains("--format"));
+    }
+
+    #[test]
+    fn command_sparse_copy_flag() {
+        let spec = RowsSpec {
+            layers: &[1, 14],
+            min_free: None,
+            force: false,
+            sparse_copy: true,
+        };
+        let c = command_for("s0", "/s", "/d", 0, 4, &spec);
+        assert!(c.contains("--format sparse-copy"));
     }
 
     #[test]
@@ -155,6 +187,7 @@ mod tests {
             layers: &[1, 14],
             min_free: None,
             force: false,
+            sparse_copy: false,
         };
         let c = command_for("s0", "/s", "/d", 0, 1, &spec);
         assert!(c.contains("--layers 1,14"));

src/staging/scripts/sparse_rows.py

@@ -119,6 +119,9 @@ class Ledger:
 
     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.save()
+
+    def save(self) -> None:
         self.path.parent.mkdir(parents=True, exist_ok=True)
         self.path.write_text(json.dumps({"files": self.files}))
         self.path.chmod(0o644)
@@ -132,6 +135,125 @@ def sample_rows(lo: int, hi: int, count: int) -> list:
     return [lo + int(i * step) for i in range(count)]
 
 
+def sparse_recs(root: Path, layer: int):
+    """{kind: (src_path, rel_name, base_off, rows)} for the layer's engram
+    tensors — base_off is the tensor's first byte inside the shard."""
+    index = json.loads((root / "model.safetensors.index.json").read_text())["weight_map"]
+    recs = {}
+    for kind, width in (("weight", WEIGHT_BYTES), ("scale", SCALE_BYTES)):
+        tname = f"layers.{layer}.engram.embed.{kind}"
+        rel = index[tname]
+        path = root / rel
+        with path.open("rb") as f:
+            length = struct.unpack("<Q", f.read(8))[0]
+            header = json.loads(f.read(length))
+        info = header[tname]
+        assert info["dtype"] in ("F8_E4M3", "F8_E8M0"), f"unexpected dtype for {tname}"
+        assert info["shape"][1] == width
+        recs[kind] = (path, rel, 8 + length + info["data_offsets"][0], info["shape"][0])
+    return recs
+
+
+def row_hash_src(recs, row: int) -> str:
+    wpath, _, woff, _ = recs["weight"]
+    spath, _, soff, _ = recs["scale"]
+    return hashlib.sha256(read_row(wpath, woff, soff, row)).digest()[:16].hex()
+
+
+def row_hash_dst(wpath, woff, spath, soff, row: int) -> str:
+    return hashlib.sha256(read_row(wpath, woff, soff, row)).digest()[:16].hex()
+
+
+def copy_range(src_f, dst_f, off: int, length: int, started: float, tag: str) -> None:
+    """Stream one byte range src->dst at the same offset (sparse-friendly)."""
+    src_f.seek(off)
+    dst_f.seek(off)
+    done = 0
+    chunk = 1 << 26
+    while done < length:
+        n = min(chunk, length - done)
+        buf = src_f.read(n)
+        if len(buf) != n:
+            raise SystemExit(f"STAGING_FAIL short read at {tag} offset {off + done}")
+        dst_f.write(buf)
+        done += n
+        rate = done / max(time.monotonic() - started, 1e-6)
+        if done == length or done % (1 << 30) < chunk:
+            print(f"PACK sparse {tag}: {done/2**30:6.1f} GiB {rate/2**20:.0f} MiB/s",
+                  flush=True)
+
+
+def stage_sparse_layer(out: Path, layer: int, recs, lo: int, hi: int, ledger: Ledger,
+                       samples_n: int, force: bool, verify: str) -> None:
+    """One rank slice into node-local sparse shard copies at the ORIGINAL
+    byte offsets (the aiden16 `DSV41_ENGRAM_DIR` contract): the loader
+    mmaps the copied shard and preads rows, so holes are free and the
+    file size must match the source exactly. Verification is again
+    content-based: size + sampled row hashes vs source — a manually
+    staged tree (the old tools/engram_local.py output) is adopted into
+    the ledger instead of re-copied."""
+    key = f"sparse-l{layer}"
+    wpath, wrel, woff, _ = recs["weight"]
+    spath, srel, soff, _ = recs["scale"]
+    dst_w, dst_s = out / wrel, out / srel
+    size_w, size_s = wpath.stat().st_size, spath.stat().st_size
+    rec = ledger.files.get(key)
+
+    def files_ok() -> bool:
+        return (dst_w.exists() and dst_w.stat().st_size == size_w
+                and dst_s.exists() and dst_s.stat().st_size == size_s)
+
+    def check(samples) -> bool:
+        for row, want in samples:
+            if row_hash_src(recs, row) != want:
+                return False  # source row changed: rebuild
+            if row_hash_dst(dst_w, woff, dst_s, soff, row) != want:
+                return False  # staged row corrupted: rebuild
+        return True
+
+    t0 = time.monotonic()
+    verified_only = False
+    if not force and files_ok():
+        if rec is not None and rec.get("lo") == lo and rec.get("hi") == hi:
+            if verify == "presence" or check([(int(k), v) for k, v in rec["sample"].items()]):
+                verified_only = True
+        else:
+            # untracked tree (manual staging): adopt when samples match
+            if verify != "presence":
+                samples = [(r, row_hash_src(recs, r)) for r in sample_rows(lo, hi, samples_n)]
+                if check(samples):
+                    ledger.files[key] = {"lo": lo, "hi": hi,
+                                         "sample": {str(r): h for r, h in samples}}
+                    verified_only = True
+    if verified_only:
+        ledger.save()
+        print(f"STAGED sparse l{layer} rows={hi - lo} verified "
+              f"secs={time.monotonic() - t0:.0f}", flush=True)
+        return
+
+    # (re)build: ensure sparse-sized files, then stream our ranges in
+    for dst, size in ((dst_w, size_w), (dst_s, size_s)):
+        dst.parent.mkdir(parents=True, exist_ok=True)
+        if not dst.exists():
+            with open(dst, "xb"):
+                os.truncate(dst, size)
+        elif dst.stat().st_size != size:
+            os.truncate(dst, size)
+    with open(wpath, "rb", buffering=0) as sf, open(dst_w, "r+b", buffering=0) as df:
+        copy_range(sf, df, woff + lo * WEIGHT_BYTES, (hi - lo) * WEIGHT_BYTES,
+                   t0, f"l{layer}-weight")
+    with open(spath, "rb", buffering=0) as sf, open(dst_s, "r+b", buffering=0) as df:
+        copy_range(sf, df, soff + lo * SCALE_BYTES, (hi - lo) * SCALE_BYTES,
+                   t0, f"l{layer}-scale")
+    ledger.files[key] = {"lo": lo, "hi": hi,
+                         "sample": {str(r): row_hash_src(recs, r)
+                                    for r in sample_rows(lo, hi, samples_n)}}
+    ledger.save()
+    print(f"STAGED sparse l{layer} rows={hi - lo} packed "
+          f"secs={time.monotonic() - t0:.0f}", flush=True)
+
+
+
 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)."""
@@ -161,6 +283,9 @@ def main() -> int:
     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("--format", default="packed", choices=["packed", "sparse-copy"],
+                    help="packed shards (row-store) or sparse shard copies at the "
+                         "original offsets (safetensors-pread / aiden16 contract)")
     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"])
@@ -199,6 +324,29 @@ def main() -> int:
         print(f"STAGING_FAIL need {need/2**30:.1f} GiB, only {free/2**30:.1f} GiB at {out_dir}")
         return 2
 
+    if args.format == "sparse-copy":
+        # the loader resolves tensors through the index, so it travels too
+        idx_src = root / "model.safetensors.index.json"
+        idx_dst = out_dir / "model.safetensors.index.json"
+        want = hashlib.sha256(idx_src.read_bytes()).hexdigest()
+        if not idx_dst.exists() or hashlib.sha256(idx_dst.read_bytes()).hexdigest() != want:
+            tmp = idx_dst.with_name(idx_dst.name + ".sm121tmp")
+            tmp.write_bytes(idx_src.read_bytes())
+            os.replace(tmp, idx_dst)
+        manifest = {}
+        for layer in layers:
+            recs = sparse_recs(root, layer)
+            rows = recs["weight"][3]
+            lo, hi = rows * args.rank // args.tp, rows * (args.rank + 1) // args.tp
+            stage_sparse_layer(out_dir, layer, recs, lo, hi, ledger,
+                               args.samples, args.force, args.verify)
+            manifest[str(layer)] = [lo, hi]
+        tmp = out_dir / "engram-local.json.sm121tmp"
+        tmp.write_text(json.dumps({"layers": manifest}))
+        os.replace(tmp, out_dir / "engram-local.json")
+        print("STAGING_DONE", flush=True)
+        return 0
+
     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"