pins: icechunk snapshot ids and content fingerprints for directory stores

.claude/skills/ddrs-dev/references/config.md

@@ -52,6 +52,36 @@ is rejected alongside `gridded_network` (the store is already split by DDR).
 
 `aorc_precip` is required whenever `kan_head.disaggregation:` is present — see below.
 
+`pins:` (optional map, source name → icechunk snapshot id) opens a source at
+`VersionInfo::SnapshotId` instead of the `main` branch tip, so a run is
+reproducible after the store is appended to. Only the two icechunk-backed
+sources — `streamflow` and `observations` (`config.rs::ICECHUNK_PINNABLE`) —
+may be pinned; a pin on any other name is a load-time error
+(`config.rs::validate_pins`), and a pin on a path that turns out to be the
+zarr-v2 global product is refused at open (`store/mod.rs::reject_pin_on_non_icechunk`).
+Omitting the block changes nothing: every source opens at the tip, exactly as
+before. Either way the id actually read is recorded — dataset open logs
+`streamflow snapshot: <id> (pinned|main tip)` next to `streamflow resolution:`,
+and `ddrs plan` writes it to `sources.lock` and the run manifest as
+`sources.<name>.snapshot` with `fp = icechunk:<id>`. Because the fingerprint IS
+the snapshot id, appending to an unpinned store shows up as drift on the next
+`plan` (one `field: locked <fp> -> current <fp>` line per source); a pinned
+source never drifts.
+Changing a pin also invalidates the cached summed-Q' baseline:
+`baseline/cache.rs::cache_key` hashes the `pins` block after the window fields,
+so re-pinning recomputes the baseline instead of scoring the model against a
+reference built from a different snapshot. The block is hashed only when present
+and non-empty, so an unpinned config keeps the key it always had and no existing
+cache under `.ddrs/baselines/` is orphaned.
+
+```yaml
+data_sources:
+  streamflow: /mnt/ssd1/data/icechunk/merit_dhbv2_UH_retrospective.ic
+  observations: /mnt/ssd1/data/icechunk/usgs_daily_observations
+  pins:
+    streamflow: E0M3W6W1881H868V2KRG
+```
+
 ## `experiment:`
 
 | Key | Production value | Notes |
@@ -261,6 +291,7 @@ when subdivision is enabled), plus one at dataset open.
 | | `geospatial_fabric_layer` on a non-gpkg | `"geospatial_fabric_layer"` + `".gpkg"` |
 | | `gridded_network` + `geospatial_fabric` | `"gridded_network"` + `"geospatial_fabric"` |
 | | `gridded_network` + explicit adjacency pair | `"gridded_network"` + `"conus_adjacency"` |
+| `validate_pins` (called from `validate_data_sources`) | `pins:` naming anything but `streamflow`/`observations` | `"pins"` + the offending source name |
 | `validate_subdivision` | `subdivision.enabled: true` + `gridded_network` | `"params.subdivision"` + `"gridded_network"` |
 | | nested `validate_subdivision_reaches_the_builder`: `enabled: true` + explicit `conus_adjacency`/`gages_adjacency` pointing at a non-subdivided store | `"params.subdivision"` + `"conflicts with the explicit"` |
 | `validate_geodataset` | `geodataset:` contradicting the adjacency source (`ddm30` with `geospatial_fabric`, `merit` with `gridded_network`) | `"geodataset"` + the source key. Absent ⇒ inferred; explicit adjacency paths ⇒ any label allowed |

.claude/skills/ddrs-dev/references/testing.md

@@ -152,6 +152,8 @@ line citation outright rather than trusting it to stay pinned.
 | Subdivision | `subdivide`, `subdivision_integration`, `gauge_mass_conservation`; `compare_ddr_sandbox` must still report ABSOLUTE MATCH |
 | Adjacency | `adjacency_parity` (managed builder byte-identical to the petgraph engine on `order`/`indices_0`/`indices_1`), `adjacency_build`, `data_zarr_store::conus_adjacency_loads_real_merit_zarr` (invariant 3 on real CONUS data) |
 | CLI / data | `data_dataset`, `data_static`, `cli_manifest`, `cli_lockfile`, `cli_json_contract` |
+| Source fingerprints (`src/cli/fingerprint.rs`) | `cli_fingerprint` (icechunk snapshot-id form, the two Juniata stores staying distinct, nested-content drift, old lock JSON without `snapshot`), then `cli_lockfile`, `cli_plan`, `cli_manifest` because all three serialize a `Fingerprint` |
+| Icechunk reader / snapshot pins (`src/data/store/icechunk.rs`, `data_sources.pins`) | `icechunk_pins` (pinned open equals unpinned at the tip; absent and malformed ids error naming path and id), `cli_lockfile`, `cargo test --lib config::tests` (pin name validation) and `cargo test --lib cache_key` (baseline key hashes the pins), plus `juniata_bundle`, `gridded_bundle`, `hourly_streamflow` as the read-path regression |
 | Checkpointing | `checkpoint_resume` (**not** `cargo test --lib training::checkpoint` — that module has zero tests) |
 | Eval robustness | `cargo test --lib training::eval::tests` |
 | Disagg freeze | `disagg_freeze` |

CLAUDE.md

@@ -366,6 +366,21 @@ no `da_ratio` output correction) are in
 DDR's gridded Q′ stores are `Qr(time, divide_id)`; the icechunk reader sniffs
 the axis order (`detect_time_major`) — see the ddrs-dev skill's trap T11.
 
+**Snapshot pinning and source fingerprints** (2026-09-16): `ddrs plan`
+fingerprints an icechunk repository by its `main` branch snapshot id
+(`fp = "icechunk:<id>"`) and any other directory store by a recursive walk of
+relative paths and sizes plus the root metadata bytes
+(`src/cli/fingerprint.rs::compute_fp`); the id actually read is recorded as
+`sources.<name>.snapshot` in `.ddrs/sources.lock` and in the run manifest,
+pinned or not, so every run names the exact data version it saw. The optional
+`data_sources.pins` map (source name to snapshot id) opens `streamflow` and
+`observations` at that snapshot instead of the branch tip; only those two names
+may be pinned (`src/config.rs::validate_pins`), and re-pinning recomputes the
+summed-Q' baseline rather than reusing the previous snapshot's, because
+`src/baseline/cache.rs::cache_key` hashes the pins. Because the directory
+fingerprint format changed, the first `ddrs plan` after this landed reports
+drift once and relocks.
+
 The `observations` and `streamflow` data sources auto-detect their format
 (`ObservationsStore::open` / `StreamflowSource::open`,
 `src/data/store/mod.rs`); the icechunk CONUS path is the fallback when the

src/baseline/cache.rs

@@ -81,6 +81,29 @@ pub fn cache_key(test_cfg: &Config) -> Result<String, BaselineError> {
     h.update(b"\n");
     h.update(exp.end_time.as_bytes());
 
+    // `data_sources.pins` changes which snapshot `summed_q_prime` READS, so it
+    // changes the baseline's inputs and must change the key — otherwise
+    // re-pinning silently reuses the previous snapshot's cached predictions,
+    // and the trained model is judged against the wrong reference.
+    //
+    // Hashed only when there is at least one pin, so an unpinned config yields
+    // the byte-identical key it did before pins existed and no cached baseline
+    // is invalidated. `BASELINE_ALGO_VERSION` is deliberately NOT bumped: the
+    // baseline ALGORITHM is unchanged (that constant means "recompute stale
+    // caches from older binaries"), this is a new INPUT, and bumping it would
+    // throw away every existing cache to no purpose.
+    if let Some(pins) = ds.pins.as_ref().filter(|p| !p.is_empty()) {
+        h.update(b"\npins\n");
+        // BTreeMap iterates in sorted key order, so the digest is independent
+        // of the order the keys appear in the YAML.
+        for (source, snapshot) in pins {
+            h.update(source.as_bytes());
+            h.update(b"=");
+            h.update(snapshot.as_bytes());
+            h.update(b"\n");
+        }
+    }
+
     let hex = h.finalize().to_hex();
     Ok(hex.as_str()[..16].to_string())
 }
@@ -374,6 +397,7 @@ mod tests {
             observations: PathBuf::from("/dev/null/obs.ic"),
             gages: PathBuf::from("/dev/null/gages.csv"),
             aorc_precip: None,
+            pins: None,
         });
         cfg.experiment = Some(Experiment {
             batch_size: 1,
@@ -431,6 +455,51 @@ mod tests {
         assert_ne!(cache_key(&cfg).unwrap(), legacy_key);
     }
 
+    /// An UNPINNED config must keep producing the key it produced before
+    /// `data_sources.pins` existed, or every cached baseline on every
+    /// workstation is invalidated for nothing. The literal is the value
+    /// `fake_config()` hashed to before the pins block was added to
+    /// `cache_key`; it is machine-independent because the fixture's
+    /// `/dev/null/...` paths never canonicalize. If this assertion fails, a
+    /// change altered the key's inputs — decide deliberately whether that is
+    /// intended before updating the literal.
+    #[test]
+    fn unpinned_cache_key_is_unchanged_by_the_pins_support() {
+        assert_eq!(cache_key(&fake_config()).unwrap(), "43843300d30ff837");
+    }
+
+    #[test]
+    fn cache_key_invalidates_on_pin_change() {
+        // `summed_q_prime` reads the pinned snapshot, so two pins are two
+        // different sets of baseline inputs and must not share a cache entry.
+        let pinned = |id: &str| {
+            let mut c = fake_config();
+            c.data_sources.as_mut().unwrap().pins =
+                Some([("streamflow".to_string(), id.to_string())].into_iter().collect());
+            c
+        };
+        assert_ne!(
+            cache_key(&pinned("E0M3W6W1881H868V2KRG")).unwrap(),
+            cache_key(&pinned("BPY4SW34FJGXWC3T806G")).unwrap(),
+        );
+    }
+
+    #[test]
+    fn cache_key_invalidates_when_a_pin_is_added() {
+        let mut pinned = fake_config();
+        pinned.data_sources.as_mut().unwrap().pins = Some(
+            [("streamflow".to_string(), "E0M3W6W1881H868V2KRG".to_string())]
+                .into_iter()
+                .collect(),
+        );
+        assert_ne!(cache_key(&fake_config()).unwrap(), cache_key(&pinned).unwrap());
+
+        // An EMPTY pins map is not a pin — it must hash as unpinned.
+        let mut empty = fake_config();
+        empty.data_sources.as_mut().unwrap().pins = Some(Default::default());
+        assert_eq!(cache_key(&fake_config()).unwrap(), cache_key(&empty).unwrap());
+    }
+
     #[test]
     fn cache_key_invalidates_on_window_change() {
         let mut a = fake_config();
@@ -502,6 +571,7 @@ mod tests {
             observations: PathBuf::from("/dev/null/obs.ic"),
             gages: PathBuf::from("/dev/null/gages.csv"),
             aorc_precip: None,
+            pins: None,
         });
         cfg.experiment = Some(Experiment {
             batch_size: 1,

src/baseline/summed_q_prime.rs

@@ -92,7 +92,7 @@ pub fn compute(test_cfg: &Config) -> Result<SummedQPrime, BaselineError> {
     let all_staids: Vec<Staid> = gage_meta.rows.iter().map(|r| r.staid.clone()).collect();
 
     let gages_adj = GagesAdjacencyStore::open(gages_adj_path, &all_staids)?;
-    let observations = ObservationsStore::open(&ds.observations)?;
+    let observations = ObservationsStore::open_at(&ds.observations, ds.pin_for("observations"))?;
     let valid_staids = valid_gauges(&all_staids, &gages_adj, |s| observations.contains(s));
     if valid_staids.is_empty() {
         return Err(BaselineError::NoGauges {
@@ -121,7 +121,7 @@ pub fn compute(test_cfg: &Config) -> Result<SummedQPrime, BaselineError> {
         n_days,
     );
 
-    let streamflow = StreamflowSource::open(&ds.streamflow)?;
+    let streamflow = StreamflowSource::open_at(&ds.streamflow, ds.pin_for("streamflow"))?;
 
     let qr_daily = streamflow.read_window_daily(start, n_days, &all_needed_sorted)?;
     let obs_daily = observations.read_window_daily(start, n_days, &valid_staids)?;

src/cli/fingerprint.rs

@@ -4,66 +4,177 @@ use std::path::{Path, PathBuf};
 use std::time::SystemTime;
 
 use crate::cli::CliError;
+use crate::data::store::icechunk::main_branch_snapshot;
 
 /// Stat-and-content fingerprint stored in `sources.lock` and the per-run
 /// manifest. `fp` is opaque — see spec § Schemas.
+///
+/// `snapshot` is set only for icechunk repositories, where it carries the
+/// `main` branch tip that `fp` is derived from. It is optional and skipped
+/// when absent so locks and manifests written before content fingerprints
+/// still parse.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Fingerprint {
     pub path: PathBuf,
     pub mtime: String,
     pub size: u64,
     pub fp: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub snapshot: Option<String>,
 }
 
 #[derive(Debug)]
 pub struct ReuseResult {
     pub fp: String,
     pub mtime: String,
     pub size: u64,
+    pub snapshot: Option<String>,
     pub reused: bool,
 }
 
 pub fn fingerprint_path(path: &Path) -> Result<Fingerprint, CliError> {
+    fingerprint_path_at(path, None)
+}
+
+/// As [`fingerprint_path`], but for a source pinned to an icechunk snapshot
+/// (`data_sources.pins`): the fingerprint then records the PIN, because that is
+/// the data version the run will actually read. Unpinned icechunk sources still
+/// record the resolved `main` tip.
+pub fn fingerprint_path_at(path: &Path, pin: Option<&str>) -> Result<Fingerprint, CliError> {
     let md = fs::metadata(path).map_err(|_| CliError::DataSourceMissing { path: path.into() })?;
     let size = md.len();
     let mtime = systime_to_iso(md.modified()?);
-    let fp = compute_fp(path, &md)?;
-    Ok(Fingerprint { path: path.into(), mtime, size, fp })
+    let (fp, snapshot) = compute_fp(path, &md, pin)?;
+    Ok(Fingerprint { path: path.into(), mtime, size, fp, snapshot })
 }
 
 /// Reuse the locked `fp` when `(path, mtime, size)` is unchanged; otherwise
 /// re-hash.
+///
+/// The stat short-circuit applies to **regular files only**. A directory's own
+/// mtime and size do not change when a chunk deep inside it is rewritten, so
+/// reusing on stat would make a mutated store look pristine. Directories are
+/// therefore always recomputed, and the cost is real: one full recursive walk
+/// per `ddrs plan` of every directory source, which for the CONUS adjacency
+/// and Q' stores is on the order of the store's file count (icechunk repos
+/// escape this — their fingerprint is one ref read, no walk).
 pub fn reuse_if_unchanged(path: &Path, locked: &Fingerprint) -> Result<ReuseResult, CliError> {
+    reuse_if_unchanged_at(path, locked, None)
+}
+
+/// As [`reuse_if_unchanged`], with the source's `data_sources.pins` value.
+pub fn reuse_if_unchanged_at(
+    path: &Path,
+    locked: &Fingerprint,
+    pin: Option<&str>,
+) -> Result<ReuseResult, CliError> {
     let md = fs::metadata(path).map_err(|_| CliError::DataSourceMissing { path: path.into() })?;
     let size = md.len();
     let mtime = systime_to_iso(md.modified()?);
-    if size == locked.size && mtime == locked.mtime {
-        return Ok(ReuseResult { fp: locked.fp.clone(), mtime, size, reused: true });
+    if !md.is_dir() && size == locked.size && mtime == locked.mtime {
+        return Ok(ReuseResult {
+            fp: locked.fp.clone(),
+            mtime,
+            size,
+            snapshot: locked.snapshot.clone(),
+            reused: true,
+        });
+    }
+    let (fp, snapshot) = compute_fp(path, &md, pin)?;
+    Ok(ReuseResult { fp, mtime, size, snapshot, reused: false })
+}
+
+/// Returns `(fp, snapshot)`.
+///
+/// * regular file (CSV, NetCDF, gpkg) — blake3 over the full content.
+/// * icechunk repository — `icechunk:<snapshot id>`: the configured `pin` when
+///   there is one, else the `main` branch tip. Either way it is the version the
+///   run will read, and a snapshot id is a content hash by construction, so no
+///   walk is needed.
+/// * any other directory (zarr v2/v3 stores, unpacked fabrics) — blake3 over
+///   the recursive `<relative path>\n<size>\n` listing plus the bytes of the
+///   root metadata file, if any.
+fn compute_fp(
+    path: &Path,
+    md: &fs::Metadata,
+    pin: Option<&str>,
+) -> Result<(String, Option<String>), CliError> {
+    if !md.is_dir() {
+        return Ok((format!("blake3:{}", blake3::hash(&fs::read(path)?).to_hex()), None));
+    }
+    if is_icechunk_repo(path) {
+        // A pin is already an immutable snapshot id — no ref read needed, and
+        // recording the tip here would name a version the run never touches.
+        // An id that does not resolve is caught when the store is opened.
+        let snapshot = match pin {
+            Some(id) => id.to_string(),
+            None => main_branch_snapshot(path).map_err(|e| {
+                CliError::Runtime(format!(
+                    "icechunk fingerprint failed for {}: {e}",
+                    path.display()
+                ))
+            })?,
+        };
+        return Ok((format!("icechunk:{snapshot}"), Some(snapshot)));
+    }
+    let (hex, files) = hash_dir(path)?;
+    if std::env::var_os("DDRS_DEBUG").is_some() {
+        eprintln!("fingerprint: {} files walked under {}", files, path.display());
     }
-    let fp = compute_fp(path, &md)?;
-    Ok(ReuseResult { fp, mtime, size, reused: false })
+    Ok((format!("blake3:{hex}"), None))
 }
 
-fn compute_fp(path: &Path, md: &fs::Metadata) -> Result<String, CliError> {
-    // For directories (zarr / icechunk stores), hash the root metadata file.
-    // For regular files (CSV, NetCDF), hash full content.
-    let bytes = if md.is_dir() {
-        for candidate in ["zarr.json", ".zarray", ".zgroup"] {
-            let p = path.join(candidate);
-            if p.is_file() {
-                return Ok(format!("blake3:{}", blake3::hash(&fs::read(p)?).to_hex()));
-            }
+/// An icechunk repository keeps its snapshots and chunk manifests in fixed
+/// top-level directories (`refs/` in the v1 layout).
+fn is_icechunk_repo(path: &Path) -> bool {
+    (path.join("snapshots").is_dir() && path.join("manifests").is_dir())
+        || path.join("refs").is_dir()
+}
+
+/// Hash the recursive contents of `root`. Returns `(hex digest, file count)`.
+///
+/// Paths are relative to `root` and sorted, so the digest is independent of
+/// readdir order. Sizes stand in for chunk bytes: a zarr chunk rewrite that
+/// preserves length is caught by the root metadata bytes only if the metadata
+/// changed, so this is a cheap-but-not-cryptographic content check — the same
+/// tradeoff the mtime/size stat check makes, applied one level deeper.
+/// Symlinks are recorded by their own size and never followed, so a cycle
+/// cannot hang the walk.
+fn hash_dir(root: &Path) -> Result<(String, usize), CliError> {
+    let mut entries: Vec<(String, u64)> = Vec::new();
+    collect(root, root, &mut entries)?;
+    entries.sort();
+
+    let mut hasher = blake3::Hasher::new();
+    for (rel, size) in &entries {
+        hasher.update(rel.as_bytes());
+        hasher.update(b"\n");
+        hasher.update(size.to_string().as_bytes());
+        hasher.update(b"\n");
+    }
+    for candidate in ["zarr.json", ".zarray", ".zgroup"] {
+        let p = root.join(candidate);
+        if p.is_file() {
+            hasher.update(&fs::read(p)?);
+            break;
         }
-        let mut names: Vec<_> = fs::read_dir(path)?
-            .filter_map(Result::ok)
-            .map(|e| e.file_name().to_string_lossy().into_owned())
-            .collect();
-        names.sort();
-        names.join("\n").into_bytes()
-    } else {
-        fs::read(path)?
-    };
-    Ok(format!("blake3:{}", blake3::hash(&bytes).to_hex()))
+    }
+    Ok((hasher.finalize().to_hex().to_string(), entries.len()))
+}
+
+fn collect(root: &Path, dir: &Path, out: &mut Vec<(String, u64)>) -> Result<(), CliError> {
+    for entry in fs::read_dir(dir)? {
+        let entry = entry?;
+        let ft = entry.file_type()?;
+        let p = entry.path();
+        if ft.is_dir() {
+            collect(root, &p, out)?;
+        } else {
+            let rel = p.strip_prefix(root).unwrap_or(&p).to_string_lossy().into_owned();
+            out.push((rel, entry.metadata()?.len()));
+        }
+    }
+    Ok(())
 }
 
 fn systime_to_iso(t: SystemTime) -> String {