From 9e47e0f8a70cb5c86905193f79da2c1525c893d2 Mon Sep 17 00:00:00 2001
From: Tadd Bindas <taddbindas@gmail.com>
Date: Tue, 15 Sep 2026 22:22:55 -0400
Subject: [PATCH 1/4] cli: content fingerprints for directory data sources
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`compute_fp` fingerprinted a directory store by hashing the sorted names of
its top-level entries, so any two icechunk repositories were identical to
`sources.lock`: the committed Juniata Q' and observations stores both hashed
to blake3:b30974f8... and drift between them was invisible.

Two directory cases now:

* icechunk repository (`snapshots/` + `manifests/`, or `refs/`) —
  `icechunk:<main branch snapshot id>`, via the new
  `src/data/store/icechunk.rs::main_branch_snapshot`, which opens the repo
  read-only the way `open_session` does and stops at `lookup_branch`. The
  snapshot id is a content hash by construction, so there is no walk.
* any other directory — blake3 over the recursive, sorted
  `<relative path>\n<size>\n` listing plus the bytes of the root metadata
  file (`zarr.json` / `.zarray` / `.zgroup`).

`Fingerprint` gains `snapshot: Option<String>`, serde-default and skipped
when absent, so existing locks and manifests still parse.

`reuse_if_unchanged` now short-circuits on stat for regular files only: a
directory's own mtime does not move when a chunk deep inside it does, so
directories always recompute. Cost is noted on the function.

Known limit, pinned by a test comment: the walk hashes sizes, not bytes, so
an in-place rewrite that preserves a file's length is not detected. Hashing
every byte of the CONUS and global stores on each `ddrs plan` was judged too
expensive; icechunk sources escape the tradeoff entirely.

`plan.rs` and `tests/cli_lockfile.rs` each gain one line for the new field.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GbK4a8dyT5eNDU97CUWHYZ
---
 src/cli/fingerprint.rs     | 129 ++++++++++++++++++++++++++++++-------
 src/cli/plan.rs            |   1 +
 src/data/store/icechunk.rs |  37 +++++++++++
 tests/cli_fingerprint.rs   | 122 +++++++++++++++++++++++++++++++++++
 tests/cli_lockfile.rs      |   2 +-
 5 files changed, 265 insertions(+), 26 deletions(-)

diff --git a/src/cli/fingerprint.rs b/src/cli/fingerprint.rs
index 73bab566..a321f74f 100644
--- a/src/cli/fingerprint.rs
+++ b/src/cli/fingerprint.rs
@@ -4,15 +4,23 @@ 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)]
@@ -20,6 +28,7 @@ pub struct ReuseResult {
     pub fp: String,
     pub mtime: String,
     pub size: u64,
+    pub snapshot: Option<String>,
     pub reused: bool,
 }
 
@@ -27,43 +36,113 @@ pub fn fingerprint_path(path: &Path) -> 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)?;
+    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> {
     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 = compute_fp(path, &md)?;
-    Ok(ReuseResult { fp, mtime, size, reused: false })
+    let (fp, snapshot) = compute_fp(path, &md)?;
+    Ok(ReuseResult { fp, mtime, size, snapshot, reused: false })
 }
 
-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()));
-            }
+/// Returns `(fp, snapshot)`.
+///
+/// * regular file (CSV, NetCDF, gpkg) — blake3 over the full content.
+/// * icechunk repository — `icechunk:<main branch snapshot id>`. The 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) -> 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) {
+        let snapshot = 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());
+    }
+    Ok((format!("blake3:{hex}"), None))
+}
+
+/// 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 {
diff --git a/src/cli/plan.rs b/src/cli/plan.rs
index 9fefccbf..39b6844e 100644
--- a/src/cli/plan.rs
+++ b/src/cli/plan.rs
@@ -218,6 +218,7 @@ pub fn plan(input: PlanInput, workspace: &Workspace) -> Result<PlanResult, CliEr
                     mtime: r.mtime,
                     size: r.size,
                     fp: r.fp,
+                    snapshot: r.snapshot,
                 }
             }
             None => fingerprint_path(&path)?,
diff --git a/src/data/store/icechunk.rs b/src/data/store/icechunk.rs
index e4755114..b4108aed 100644
--- a/src/data/store/icechunk.rs
+++ b/src/data/store/icechunk.rs
@@ -78,6 +78,43 @@ pub(crate) fn open_session(path: &Path) -> Result<IcSession> {
     Ok(IcSession { runtime, store })
 }
 
+/// Resolve the `main` branch tip of the icechunk repository at `path` to its
+/// snapshot id (Crockford base32, as icechunk prints it).
+///
+/// Opens the repository read-only exactly the way [`open_session`] does — a
+/// private 2-worker tokio runtime, `new_local_filesystem_storage` +
+/// `Repository::open` — but stops at `lookup_branch`, so no `Session` and no
+/// `Store` are built. Used by `src/cli/fingerprint.rs::compute_fp` to give an
+/// icechunk store a content fingerprint: the snapshot id is a content hash by
+/// construction, so no file walk is needed.
+pub fn main_branch_snapshot(path: &Path) -> Result<String> {
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .map_err(|e| DataError::Io {
+            path: path.to_path_buf(),
+            source: e,
+        })?;
+
+    runtime.block_on(async {
+        let storage = new_local_filesystem_storage(path)
+            .await
+            .map_err(|e| ic_err(path, e))?;
+
+        let repo = Repository::open(None, storage, std::collections::HashMap::new())
+            .await
+            .map_err(|e| ic_err(path, e))?;
+
+        let snapshot = repo
+            .lookup_branch("main")
+            .await
+            .map_err(|e| ic_err(path, e))?;
+
+        Ok(snapshot.to_string())
+    })
+}
+
 #[allow(dead_code)] // Used by the helpers in this module once filled in.
 pub(crate) fn ic_err<E: std::error::Error + Send + Sync + 'static>(
     path: &Path,
diff --git a/tests/cli_fingerprint.rs b/tests/cli_fingerprint.rs
index bb167c13..a79f9fd0 100644
--- a/tests/cli_fingerprint.rs
+++ b/tests/cli_fingerprint.rs
@@ -50,3 +50,125 @@ fn reuse_recomputes_when_size_changes() {
     assert_ne!(reused.fp, fp.fp);
     assert!(!reused.reused);
 }
+
+// ---------------------------------------------------------------------------
+// Directory stores: content fingerprints (Task 1 of the pins plan).
+// ---------------------------------------------------------------------------
+
+/// Build a zarr-v3-shaped store: root `zarr.json`, a nested array with two
+/// chunk files.
+fn zarr_like(root: &std::path::Path) {
+    fs::create_dir_all(root.join("Qr/c/0")).unwrap();
+    fs::write(root.join("zarr.json"), br#"{"zarr_format":3,"node_type":"group"}"#).unwrap();
+    fs::write(root.join("Qr/zarr.json"), br#"{"zarr_format":3,"node_type":"array"}"#).unwrap();
+    fs::write(root.join("Qr/c/0/0"), b"chunk-zero").unwrap();
+    fs::write(root.join("Qr/c/0/1"), b"chunk-one!").unwrap();
+}
+
+#[test]
+fn directory_fp_changes_when_a_nested_chunk_changes() {
+    let d = tempfile::tempdir().unwrap();
+    let root = d.path().join("store.zarr");
+    zarr_like(&root);
+    let before = fingerprint_path(&root).unwrap();
+
+    // Edit a chunk two levels down. The fingerprint hashes the recursive
+    // (relative path, size) listing, so the edit is seen through the size.
+    // KNOWN LIMIT: an in-place rewrite that preserves length is NOT detected
+    // — hashing every byte of the CONUS/global stores on each `ddrs plan` was
+    // judged too expensive. See `fingerprint.rs::hash_dir`.
+    fs::write(root.join("Qr/c/0/1"), b"chunk-one-rewritten").unwrap();
+    let after = fingerprint_path(&root).unwrap();
+    assert_ne!(
+        before.fp, after.fp,
+        "nested chunk edit must change the directory fingerprint"
+    );
+}
+
+#[test]
+fn directory_fp_stable_when_only_mtimes_change() {
+    let d = tempfile::tempdir().unwrap();
+    let root = d.path().join("store.zarr");
+    zarr_like(&root);
+    let before = fingerprint_path(&root).unwrap();
+
+    std::thread::sleep(std::time::Duration::from_millis(50));
+    for rel in ["zarr.json", "Qr/zarr.json", "Qr/c/0/0", "Qr/c/0/1"] {
+        let p = root.join(rel);
+        let bytes = fs::read(&p).unwrap();
+        fs::write(&p, bytes).unwrap(); // rewrite identical content, new mtime
+    }
+    let after = fingerprint_path(&root).unwrap();
+    assert_eq!(
+        before.fp, after.fp,
+        "rewriting identical bytes must not change the fingerprint"
+    );
+}
+
+#[test]
+fn the_two_juniata_icechunk_stores_have_distinct_fingerprints() {
+    let data = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("examples/juniata/data");
+    let qprime = fingerprint_path(&data.join("juniata_qprime.ic")).unwrap();
+    let obs = fingerprint_path(&data.join("juniata_obs.ic")).unwrap();
+    assert_ne!(
+        qprime.fp, obs.fp,
+        "two different icechunk stores must not share a fingerprint"
+    );
+}
+
+/// Open the repo the way `src/data/store/icechunk.rs::open_session` does and
+/// read the `main` branch tip, so the assertion below is independent of the
+/// production helper.
+fn main_branch_tip(path: &std::path::Path) -> String {
+    let rt = tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(2)
+        .enable_all()
+        .build()
+        .unwrap();
+    rt.block_on(async {
+        let storage = icechunk::new_local_filesystem_storage(path).await.unwrap();
+        let repo = icechunk::Repository::open(None, storage, std::collections::HashMap::new())
+            .await
+            .unwrap();
+        repo.lookup_branch("main").await.unwrap().to_string()
+    })
+}
+
+#[test]
+fn icechunk_fp_is_the_main_branch_snapshot_id() {
+    let data = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("examples/juniata/data");
+    for name in ["juniata_qprime.ic", "juniata_obs.ic"] {
+        let p = data.join(name);
+        let fp = fingerprint_path(&p).unwrap();
+        let tip = main_branch_tip(&p);
+        assert_eq!(fp.fp, format!("icechunk:{tip}"), "{name}");
+        assert_eq!(fp.snapshot.as_deref(), Some(tip.as_str()), "{name}");
+    }
+}
+
+#[test]
+fn directory_reuse_always_recomputes() {
+    let d = tempfile::tempdir().unwrap();
+    let root = d.path().join("store.zarr");
+    zarr_like(&root);
+    let locked = fingerprint_path(&root).unwrap();
+
+    // Edit a nested chunk in place: the root directory's own mtime and size
+    // are untouched, so a stat short-circuit would wrongly report "reused".
+    fs::write(root.join("Qr/c/0/1"), b"chunk-one-rewritten").unwrap();
+    let r = reuse_if_unchanged(&root, &locked).unwrap();
+    assert!(!r.reused, "directories must never short-circuit on stat");
+    assert_ne!(r.fp, locked.fp);
+}
+
+#[test]
+fn fingerprint_json_without_snapshot_still_parses() {
+    let json = r#"{"path":"/a/b.csv","mtime":"2026-01-01T00:00:00Z","size":5,"fp":"blake3:dead"}"#;
+    let fp: Fingerprint = serde_json::from_str(json).unwrap();
+    assert_eq!(fp.snapshot, None);
+    assert_eq!(fp.fp, "blake3:dead");
+    // And a snapshot-less fingerprint round-trips without emitting the key.
+    let back = serde_json::to_string(&fp).unwrap();
+    assert!(!back.contains("snapshot"), "{back}");
+}
+
diff --git a/tests/cli_lockfile.rs b/tests/cli_lockfile.rs
index 38f61857..bad6fc2c 100644
--- a/tests/cli_lockfile.rs
+++ b/tests/cli_lockfile.rs
@@ -6,7 +6,7 @@ use std::fs;
 fn fp(path: &str, fp: &str) -> Fingerprint {
     Fingerprint {
         path: path.into(), mtime: "2026-05-30T00:00:00Z".into(),
-        size: 1, fp: fp.into(),
+        size: 1, fp: fp.into(), snapshot: None,
     }
 }
 

From 0939111b35aaa22916d93f283512c412e71ac3af Mon Sep 17 00:00:00 2001
From: Tadd Bindas <taddbindas@gmail.com>
Date: Tue, 15 Sep 2026 22:57:35 -0400
Subject: [PATCH 2/4] data: pin icechunk sources to a snapshot via
 data_sources.pins
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`data_sources.pins` (optional map, source name -> icechunk snapshot id)
opens `streamflow` / `observations` at `VersionInfo::SnapshotId` instead of
the `main` branch tip, so a run is reproducible after the store is appended
to. A pin on any other source name is rejected at load
(`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`).

Both icechunk stores gain `open_at(path, Option<&str>)` (`open` delegates
with `None`) and a `snapshot` field carrying the id they actually read —
the pin, or the `main` tip resolved via `Session::snapshot_id()`. A
malformed or absent id is a `DataError::Malformed` naming the store path
and the id. Dataset open logs `streamflow snapshot: <id> (pinned|main tip)`
beside the existing `streamflow resolution:` line, same for observations.

The fingerprint path takes the pin too (`fingerprint_path_at` /
`reuse_if_unchanged_at`), so `sources.lock` and the run manifest record the
version the run will read rather than a tip it never touches. `ddrs plan`'s
drift report now prints one `source: locked <fp> -> current <fp>` line per
drifted source, in both the warn and the `--strict` abort path.

Omitting `pins:` changes nothing except that icechunk sources now carry
their resolved snapshot id in the lock and manifest.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GbK4a8dyT5eNDU97CUWHYZ
---
 .claude/skills/ddrs-dev/references/config.md |  25 ++++
 src/baseline/cache.rs                        |   2 +
 src/baseline/summed_q_prime.rs               |   4 +-
 src/cli/fingerprint.rs                       |  48 +++++-
 src/cli/plan.rs                              |  24 ++-
 src/cli/sources.rs                           |   6 +-
 src/config.rs                                | 108 +++++++++++++
 src/data/dataset.rs                          |  29 +++-
 src/data/store/icechunk.rs                   |  91 +++++++++--
 src/data/store/mod.rs                        |  65 +++++++-
 tests/cli_lockfile.rs                        |  85 +++++++++++
 tests/icechunk_pins.rs                       | 150 +++++++++++++++++++
 12 files changed, 601 insertions(+), 36 deletions(-)
 create mode 100644 tests/icechunk_pins.rs

diff --git a/.claude/skills/ddrs-dev/references/config.md b/.claude/skills/ddrs-dev/references/config.md
index 70ee42ae..6021deb0 100644
--- a/.claude/skills/ddrs-dev/references/config.md
+++ b/.claude/skills/ddrs-dev/references/config.md
@@ -52,6 +52,30 @@ 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.
+
+```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 +285,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 |
diff --git a/src/baseline/cache.rs b/src/baseline/cache.rs
index 83eb20f5..944a2596 100644
--- a/src/baseline/cache.rs
+++ b/src/baseline/cache.rs
@@ -374,6 +374,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,
@@ -502,6 +503,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,
diff --git a/src/baseline/summed_q_prime.rs b/src/baseline/summed_q_prime.rs
index 45091b0e..91f8823e 100644
--- a/src/baseline/summed_q_prime.rs
+++ b/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)?;
diff --git a/src/cli/fingerprint.rs b/src/cli/fingerprint.rs
index a321f74f..c534e55e 100644
--- a/src/cli/fingerprint.rs
+++ b/src/cli/fingerprint.rs
@@ -33,10 +33,18 @@ pub struct ReuseResult {
 }
 
 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, snapshot) = compute_fp(path, &md)?;
+    let (fp, snapshot) = compute_fp(path, &md, pin)?;
     Ok(Fingerprint { path: path.into(), mtime, size, fp, snapshot })
 }
 
@@ -51,6 +59,15 @@ pub fn fingerprint_path(path: &Path) -> Result<Fingerprint, CliError> {
 /// 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()?);
@@ -63,26 +80,41 @@ pub fn reuse_if_unchanged(path: &Path, locked: &Fingerprint) -> Result<ReuseResu
             reused: true,
         });
     }
-    let (fp, snapshot) = compute_fp(path, &md)?;
+    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:<main branch snapshot id>`. The snapshot
-///   id is a content hash by construction, so no walk is needed.
+/// * 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) -> Result<(String, Option<String>), CliError> {
+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) {
-        let snapshot = main_branch_snapshot(path).map_err(|e| {
-            CliError::Runtime(format!("icechunk fingerprint failed for {}: {e}", path.display()))
-        })?;
+        // 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)?;
diff --git a/src/cli/plan.rs b/src/cli/plan.rs
index 39b6844e..d8b7633a 100644
--- a/src/cli/plan.rs
+++ b/src/cli/plan.rs
@@ -5,7 +5,7 @@ use serde::Serialize;
 use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 
-use crate::cli::fingerprint::{Fingerprint, fingerprint_path, reuse_if_unchanged};
+use crate::cli::fingerprint::{Fingerprint, fingerprint_path_at, reuse_if_unchanged_at};
 use crate::cli::lockfile::{Lockfile, diff_against_live};
 use crate::cli::types::Workflow;
 use crate::cli::workspace::Workspace;
@@ -210,9 +210,12 @@ pub fn plan(input: PlanInput, workspace: &Workspace) -> Result<PlanResult, CliEr
     }
     let mut sources = BTreeMap::new();
     for (key, path) in pairs {
+        // `data_sources.pins` (icechunk sources only) decides which snapshot
+        // the run will read, so the lock records the pin rather than the tip.
+        let pin = data_sources.pin_for(&key);
         let live = match prior_lock.as_ref().and_then(|l| l.sources.get(&key)) {
             Some(locked) => {
-                let r = reuse_if_unchanged(&path, locked)?;
+                let r = reuse_if_unchanged_at(&path, locked, pin)?;
                 Fingerprint {
                     path: path.clone(),
                     mtime: r.mtime,
@@ -221,7 +224,7 @@ pub fn plan(input: PlanInput, workspace: &Workspace) -> Result<PlanResult, CliEr
                     snapshot: r.snapshot,
                 }
             }
-            None => fingerprint_path(&path)?,
+            None => fingerprint_path_at(&path, pin)?,
         };
         sources.insert(key, live);
     }
@@ -234,10 +237,23 @@ pub fn plan(input: PlanInput, workspace: &Workspace) -> Result<PlanResult, CliEr
     // Drift policy + auto-relock. Strict callers (run --strict) abort
     // BEFORE the lock is refreshed so the drift evidence survives.
     if !drift.is_empty() {
+        // One line per drifted source, naming both fingerprints. For an
+        // icechunk source the fp IS the snapshot id, so this reads as
+        // "the store was appended to since the lock" without further digging.
+        eprintln!("data source drift since last plan:");
+        for field in &drift {
+            let locked = prior_lock
+                .as_ref()
+                .and_then(|l| l.sources.get(field))
+                .map(|f| f.fp.as_str())
+                .unwrap_or("<absent>");
+            let current = sources.get(field).map(|f| f.fp.as_str()).unwrap_or("<absent>");
+            eprintln!("  {field}: locked {locked} -> current {current}");
+        }
         if input.strict {
             return Err(CliError::LockDrift { fields: drift });
         }
-        eprintln!("warning: data source drift since last plan: {drift:?} — relocking");
+        eprintln!("warning: relocking (pass --strict to abort instead)");
     }
     // Rewrite only when something actually changed, so an unchanged re-plan
     // leaves the lock byte-identical. Note: intentionally stat-sensitive
diff --git a/src/cli/sources.rs b/src/cli/sources.rs
index 90938edd..f63089cd 100644
--- a/src/cli/sources.rs
+++ b/src/cli/sources.rs
@@ -25,7 +25,7 @@ use std::path::{Path, PathBuf};
 
 use serde::Deserialize;
 
-use crate::cli::fingerprint::fingerprint_path;
+use crate::cli::fingerprint::fingerprint_path_at;
 use crate::cli::lockfile::Lockfile;
 use crate::cli::workspace::Workspace;
 use crate::config::{Config, ConfigMode, DataSources};
@@ -251,7 +251,9 @@ fn lock_sources_from_config(cfg_path: &Path, ws: &Workspace) -> Result<(), CliEr
             .map(|(k, p)| {
                 let p = p.clone();
                 let k = k.to_string();
-                s.spawn(move || (k, fingerprint_path(&p)))
+                // Same rule as `plan`: a pinned icechunk source locks its pin.
+                let pin = ds.pin_for(&k).map(str::to_owned);
+                s.spawn(move || (k, fingerprint_path_at(&p, pin.as_deref())))
             })
             .collect();
         handles
diff --git a/src/config.rs b/src/config.rs
index e4c94e7a..98b69718 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -138,6 +138,30 @@ pub struct DataSources {
     /// `research/specs/2026-09-08-ddrs-gridded-routing-design.md`.
     #[serde(default)]
     pub gridded_network: Option<std::path::PathBuf>,
+    /// Optional per-source icechunk snapshot pins: source name → snapshot id
+    /// (Crockford base32, as icechunk prints it). A pinned source is opened at
+    /// `VersionInfo::SnapshotId` instead of the `main` branch tip, so a run is
+    /// reproducible even after the store is appended to. Only the two
+    /// icechunk-backed sources can be pinned (`ICECHUNK_PINNABLE`); a pin on
+    /// any other name is rejected at load rather than silently ignored.
+    /// Absent (the default) ⇒ every source opens at the `main` tip, exactly as
+    /// before.
+    #[serde(default)]
+    pub pins: Option<BTreeMap<String, String>>,
+}
+
+/// The `data_sources` keys backed by an icechunk repository, and therefore the
+/// only ones `data_sources.pins` can name.
+pub const ICECHUNK_PINNABLE: [&str; 2] = ["streamflow", "observations"];
+
+impl DataSources {
+    /// The pinned icechunk snapshot id for `source`, if one is configured.
+    pub fn pin_for(&self, source: &str) -> Option<&str> {
+        self.pins
+            .as_ref()
+            .and_then(|p| p.get(source))
+            .map(String::as_str)
+    }
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -1124,11 +1148,33 @@ impl Config {
 /// - Exactly one adjacency key → error: partial adjacency.
 /// - `geospatial_fabric_layer` set while the fabric is not a `.gpkg` → error
 ///   (the layer concept only exists for GeoPackage).
+/// `data_sources.pins` may only name an icechunk-backed source.
+///
+/// A pin is a snapshot id handed to `VersionInfo::SnapshotId` at open. Only
+/// `streamflow` and `observations` go through that path; a pin on `attributes`,
+/// an adjacency zarr or the gage CSV would be read by nothing, so it is a
+/// load-time error rather than a silently ignored key.
+fn validate_pins(ds: &DataSources) -> std::result::Result<(), String> {
+    let Some(pins) = ds.pins.as_ref() else { return Ok(()) };
+    for name in pins.keys() {
+        if !ICECHUNK_PINNABLE.contains(&name.as_str()) {
+            return Err(format!(
+                "data_sources: `pins` names `{name}`, which is not an icechunk-backed \
+                 source; only {} can be pinned to a snapshot id.",
+                ICECHUNK_PINNABLE.join(" and ")
+            ));
+        }
+    }
+    Ok(())
+}
+
 fn validate_data_sources(cfg: &Config) -> std::result::Result<(), String> {
     let ds = match cfg.data_sources.as_ref() {
         None => return Ok(()), // no data_sources section at all — allowed for default/test configs
         Some(ds) => ds,
     };
+    validate_pins(ds)?;
+
     let has_conus = ds.conus_adjacency.is_some();
     let has_gages = ds.gages_adjacency.is_some();
     let has_fabric = ds.geospatial_fabric.is_some();
@@ -2048,6 +2094,68 @@ data_sources:
         );
     }
 
+    #[test]
+    fn pins_on_a_non_icechunk_source_are_rejected_by_name() {
+        // `attributes` is a NetCDF file, not an icechunk repo — a pin there
+        // would be silently ignored, so it is a load-time error.
+        let ds_block = r#"
+data_sources:
+  attributes: /dev/null/attrs.nc
+  conus_adjacency: /dev/null/conus.zarr
+  gages_adjacency: /dev/null/gages.zarr
+  streamflow: /dev/null/sf.ic
+  observations: /dev/null/obs.ic
+  gages: /dev/null/gages.csv
+  pins:
+    attributes: abc
+"#;
+        let path = write_yaml_with_data_sources("ddrs_ds_pin_bad_source.yaml", ds_block);
+        let err = Config::from_yaml_file(&path).unwrap_err();
+        let msg = format!("{}", err);
+        assert!(msg.contains("pins"), "expected a pins error, got: {msg}");
+        assert!(msg.contains("attributes"), "error must name the source, got: {msg}");
+    }
+
+    #[test]
+    fn pins_on_the_icechunk_sources_parse() {
+        let ds_block = r#"
+data_sources:
+  attributes: /dev/null/attrs.nc
+  conus_adjacency: /dev/null/conus.zarr
+  gages_adjacency: /dev/null/gages.zarr
+  streamflow: /dev/null/sf.ic
+  observations: /dev/null/obs.ic
+  gages: /dev/null/gages.csv
+  pins:
+    streamflow: E0M3W6W1881H868V2KRG
+    observations: BPY4SW34FJGXWC3T806G
+"#;
+        let path = write_yaml_with_data_sources("ddrs_ds_pin_ok.yaml", ds_block);
+        let cfg = Config::from_yaml_file(&path).expect("pins on icechunk sources are valid");
+        let ds = cfg.data_sources.as_ref().unwrap();
+        assert_eq!(ds.pin_for("streamflow"), Some("E0M3W6W1881H868V2KRG"));
+        assert_eq!(ds.pin_for("observations"), Some("BPY4SW34FJGXWC3T806G"));
+        assert_eq!(ds.pin_for("attributes"), None);
+    }
+
+    #[test]
+    fn absent_pins_block_leaves_every_source_unpinned() {
+        let ds_block = r#"
+data_sources:
+  attributes: /dev/null/attrs.nc
+  conus_adjacency: /dev/null/conus.zarr
+  gages_adjacency: /dev/null/gages.zarr
+  streamflow: /dev/null/sf.ic
+  observations: /dev/null/obs.ic
+  gages: /dev/null/gages.csv
+"#;
+        let path = write_yaml_with_data_sources("ddrs_ds_pin_absent.yaml", ds_block);
+        let cfg = Config::from_yaml_file(&path).expect("no pins block is valid");
+        let ds = cfg.data_sources.as_ref().unwrap();
+        assert!(ds.pins.is_none());
+        assert_eq!(ds.pin_for("streamflow"), None);
+    }
+
     #[test]
     fn gridded_network_alone_is_a_valid_managed_build() {
         let ds_block = r#"
diff --git a/src/data/dataset.rs b/src/data/dataset.rs
index 4ee379e8..a422d353 100644
--- a/src/data/dataset.rs
+++ b/src/data/dataset.rs
@@ -360,6 +360,20 @@ pub struct MeritGagesDataset {
     leakance_impervious_threshold: Option<f32>,
 }
 
+/// Log the data version a store opened at, as a sibling of the
+/// `streamflow resolution:` self-check line. `(pinned)` means
+/// `data_sources.pins.<source>` decided it; `(main tip)` means the branch tip
+/// was resolved at open. Non-icechunk sources carry no snapshot and say so.
+fn log_snapshot(source: &str, snapshot: Option<&str>, pin: Option<&str>) {
+    match snapshot {
+        Some(id) => {
+            let origin = if pin.is_some() { "pinned" } else { "main tip" };
+            eprintln!("{source} snapshot: {id} ({origin})");
+        }
+        None => eprintln!("{source} snapshot: n/a (not an icechunk store)"),
+    }
+}
+
 /// Reject the disaggregation head when the streamflow store is hourly-native:
 /// disaggregating an already-hourly signal is a config contradiction, and
 /// after the 2026-07-01 stale-binary incident nothing in the forcing path is
@@ -485,15 +499,24 @@ impl MeritGagesDataset {
         };
 
         // ---------- 3. Icechunk stores ----------
-        let streamflow = Arc::new(StreamflowSource::open(&ds.streamflow)?);
-        // The smoke-train self-check line: proves which read path executed.
+        let streamflow = Arc::new(StreamflowSource::open_at(
+            &ds.streamflow,
+            ds.pin_for("streamflow"),
+        )?);
+        // The smoke-train self-check lines: prove which read path executed,
+        // and which data version it read (see `data_sources.pins`).
         eprintln!("streamflow resolution: {:?}", streamflow.resolution());
+        log_snapshot("streamflow", streamflow.snapshot(), ds.pin_for("streamflow"));
         validate_disagg_vs_resolution(
             streamflow.resolution(),
             head_cfg.disaggregation.is_some(),
             &ds.streamflow,
         )?;
-        let observations = Arc::new(ObservationsStore::open(&ds.observations)?);
+        let observations = Arc::new(ObservationsStore::open_at(
+            &ds.observations,
+            ds.pin_for("observations"),
+        )?);
+        log_snapshot("observations", observations.snapshot(), ds.pin_for("observations"));
 
         // Optional hourly precip store for the disaggregation head. The head
         // always requires precip when enabled — no separate toggle. (`temp`
diff --git a/src/data/store/icechunk.rs b/src/data/store/icechunk.rs
index b4108aed..cfa6d2bd 100644
--- a/src/data/store/icechunk.rs
+++ b/src/data/store/icechunk.rs
@@ -23,6 +23,7 @@ use tokio::runtime::Runtime;
 use tokio::sync::RwLock;
 
 use icechunk::{Repository, Store, new_local_filesystem_storage};
+use icechunk::format::SnapshotId;
 use icechunk::repository::VersionInfo;
 use zarrs::array::Array as ZarrArray;
 use zarrs::storage::{
@@ -40,12 +41,24 @@ use crate::data::ids::{Comid, IdIndex, Staid};
 pub(crate) struct IcSession {
     pub(crate) runtime: Arc<Runtime>,
     pub(crate) store: Arc<Store>,
+    /// The snapshot the session actually reads — the pinned id when `snapshot`
+    /// was `Some`, otherwise the resolved `main` branch tip.
+    pub(crate) snapshot: String,
 }
 
-/// Open an icechunk repo at `path` and start a read-only session on the
-/// `main` branch.
+/// Open an icechunk repo at `path` read-only.
+///
+/// `snapshot` is `data_sources.pins`' value for this source: `None` opens the
+/// `main` branch tip (the historical behavior), `Some(id)` opens exactly that
+/// snapshot via `VersionInfo::SnapshotId`. Either way the resolved id is
+/// returned on [`IcSession::snapshot`], so the run manifest can record the data
+/// version that was actually read.
+///
+/// A malformed id (not Crockford base32 of 12 bytes) and an id no snapshot in
+/// this repo carries are both `DataError::Malformed`, naming the store path and
+/// the id — icechunk's own errors name neither reliably.
 #[allow(dead_code)] // Tasks 2-4 will call this.
-pub(crate) fn open_session(path: &Path) -> Result<IcSession> {
+pub(crate) fn open_session(path: &Path, snapshot: Option<&str>) -> Result<IcSession> {
     let runtime = Arc::new(
         tokio::runtime::Builder::new_multi_thread()
             .worker_threads(2)
@@ -57,7 +70,7 @@ pub(crate) fn open_session(path: &Path) -> Result<IcSession> {
             })?,
     );
 
-    let store = runtime.block_on(async {
+    let (store, resolved) = runtime.block_on(async {
         let storage = new_local_filesystem_storage(path)
             .await
             .map_err(|e| ic_err(path, e))?;
@@ -66,16 +79,35 @@ pub(crate) fn open_session(path: &Path) -> Result<IcSession> {
             .await
             .map_err(|e| ic_err(path, e))?;
 
-        let session = repo
-            .readonly_session(&VersionInfo::BranchTipRef("main".into()))
-            .await
-            .map_err(|e| ic_err(path, e))?;
+        let version = match snapshot {
+            None => VersionInfo::BranchTipRef("main".into()),
+            Some(id) => VersionInfo::SnapshotId(parse_snapshot_id(path, id)?),
+        };
+
+        let session = repo.readonly_session(&version).await.map_err(|e| match snapshot {
+            // A pinned open that fails is almost always "no such snapshot";
+            // re-wrap so the message names the id the user wrote.
+            Some(id) => DataError::Malformed {
+                path: path.to_path_buf(),
+                message: format!("cannot open pinned icechunk snapshot `{id}`: {e}"),
+            },
+            None => ic_err(path, e),
+        })?;
+        let resolved = session.snapshot_id().to_string();
 
         let store = Store::from_session(Arc::new(RwLock::new(session))).await;
-        Ok::<Arc<Store>, DataError>(Arc::new(store))
+        Ok::<(Arc<Store>, String), DataError>((Arc::new(store), resolved))
     })?;
 
-    Ok(IcSession { runtime, store })
+    Ok(IcSession { runtime, store, snapshot: resolved })
+}
+
+/// Parse a configured pin into an icechunk `SnapshotId`.
+fn parse_snapshot_id(path: &Path, id: &str) -> Result<SnapshotId> {
+    SnapshotId::try_from(id).map_err(|e| DataError::Malformed {
+        path: path.to_path_buf(),
+        message: format!("`{id}` is not a valid icechunk snapshot id: {e}"),
+    })
 }
 
 /// Resolve the `main` branch tip of the icechunk repository at `path` to its
@@ -222,6 +254,9 @@ pub struct StreamflowStore {
     pub n_time: usize,
     /// Native axis resolution, sniffed from the CF `units` attribute.
     pub resolution: Frequency,
+    /// The icechunk snapshot this store reads: the configured pin, or the
+    /// `main` branch tip resolved at open. Recorded in the run manifest.
+    pub snapshot: String,
     // SP-3 may consolidate to a shared runtime; keep the Arc alive so the
     // icechunk Store is not dropped while `qr` is in use.
     #[allow(dead_code)]
@@ -278,8 +313,15 @@ fn detect_time_major(
 
 impl StreamflowStore {
     pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
+        Self::open_at(path, None)
+    }
+
+    /// Open at a specific icechunk snapshot. `snapshot: None` is [`Self::open`]
+    /// — the `main` branch tip. See `data_sources.pins` in `src/config.rs`.
+    pub fn open_at(path: impl Into<PathBuf>, snapshot: Option<&str>) -> Result<Self> {
         let path = path.into();
-        let session = open_session(&path)?;
+        let session = open_session(&path, snapshot)?;
+        let snapshot = session.snapshot.clone();
         let storage: Arc<IcZarrStorage> = IcZarrStorage::shared(&session);
         // zarrs Array::open takes Arc<dyn ReadableStorageTraits> — cast via type alias
         let readable: ReadableStorage = storage.clone();
@@ -325,7 +367,17 @@ impl StreamflowStore {
             .map_err(|e| ic_err(&path, e))?;
         let time_major = detect_time_major(&qr, index.len(), n_time, &path)?;
 
-        Ok(Self { path, index, time_start, n_time, resolution, storage, qr, time_major })
+        Ok(Self {
+            path,
+            index,
+            time_start,
+            n_time,
+            resolution,
+            snapshot,
+            storage,
+            qr,
+            time_major,
+        })
     }
 }
 
@@ -783,6 +835,9 @@ pub struct UsgsObservationsStore {
     pub index: IdIndex<Staid>,
     pub time_start: NaiveDate,
     pub n_time: usize,
+    /// The icechunk snapshot this store reads (pin, or the resolved `main`
+    /// tip). Recorded in the run manifest.
+    pub snapshot: String,
     #[allow(dead_code)]
     storage: Arc<IcZarrStorage>,
     streamflow: ZarrArray<dyn ReadableStorageTraits>,
@@ -790,8 +845,15 @@ pub struct UsgsObservationsStore {
 
 impl UsgsObservationsStore {
     pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
+        Self::open_at(path, None)
+    }
+
+    /// Open at a specific icechunk snapshot. `snapshot: None` is [`Self::open`]
+    /// — the `main` branch tip. See `data_sources.pins` in `src/config.rs`.
+    pub fn open_at(path: impl Into<PathBuf>, snapshot: Option<&str>) -> Result<Self> {
         let path = path.into();
-        let session = open_session(&path)?;
+        let session = open_session(&path, snapshot)?;
+        let snapshot = session.snapshot.clone();
         let storage: Arc<IcZarrStorage> = IcZarrStorage::shared(&session);
         let readable: ReadableStorage = storage.clone();
 
@@ -825,6 +887,7 @@ impl UsgsObservationsStore {
             index,
             time_start,
             n_time,
+            snapshot,
             storage,
             streamflow,
         })
@@ -1095,7 +1158,7 @@ mod tests {
         if !p.exists() {
             return;
         }
-        assert!(open_session(p).is_ok());
+        assert!(open_session(p, None).is_ok());
     }
 
     #[test]
diff --git a/src/data/store/mod.rs b/src/data/store/mod.rs
index 68f35f78..ee01f8d7 100644
--- a/src/data/store/mod.rs
+++ b/src/data/store/mod.rs
@@ -34,7 +34,7 @@ pub use zarr_qprime::GlobalStreamflowStore;
 use ndarray::Array2;
 
 use crate::data::dates::{Frequency, RhoWindow};
-use crate::data::error::Result;
+use crate::data::error::{DataError, Result};
 use crate::data::ids::{Comid, Staid};
 
 /// Format-dispatching observations reader. The `observations` data source is
@@ -53,11 +53,30 @@ impl ObservationsStore {
     /// Open `path`, sniffing the format: a `.zgroup` at the root means a
     /// plain zarr v2 group; anything else is treated as an icechunk repo.
     pub fn open(path: impl Into<std::path::PathBuf>) -> Result<Self> {
+        Self::open_at(path, None)
+    }
+
+    /// Open at a configured icechunk snapshot pin
+    /// (`data_sources.pins.observations`). `None` is [`Self::open`].
+    pub fn open_at(
+        path: impl Into<std::path::PathBuf>,
+        snapshot: Option<&str>,
+    ) -> Result<Self> {
         let path = path.into();
         if GlobalObservationsStore::sniff(&path) {
+            reject_pin_on_non_icechunk(&path, snapshot, "observations")?;
             Ok(Self::Global(GlobalObservationsStore::open(path)?))
         } else {
-            Ok(Self::Usgs(UsgsObservationsStore::open(path)?))
+            Ok(Self::Usgs(UsgsObservationsStore::open_at(path, snapshot)?))
+        }
+    }
+
+    /// The icechunk snapshot this store reads, when it is icechunk-backed.
+    /// The zarr-v2 global product has no versioning, hence `None`.
+    pub fn snapshot(&self) -> Option<&str> {
+        match self {
+            Self::Usgs(s) => Some(s.snapshot.as_str()),
+            Self::Global(_) => None,
         }
     }
 
@@ -108,11 +127,30 @@ impl StreamflowSource {
     /// `streamflow/.zarray`, at the root or one level down) mean the
     /// global zarr v2 layout; anything else is treated as an icechunk repo.
     pub fn open(path: impl Into<std::path::PathBuf>) -> Result<Self> {
+        Self::open_at(path, None)
+    }
+
+    /// Open at a configured icechunk snapshot pin
+    /// (`data_sources.pins.streamflow`). `None` is [`Self::open`].
+    pub fn open_at(
+        path: impl Into<std::path::PathBuf>,
+        snapshot: Option<&str>,
+    ) -> Result<Self> {
         let path = path.into();
         if GlobalStreamflowStore::sniff(&path) {
+            reject_pin_on_non_icechunk(&path, snapshot, "streamflow")?;
             Ok(Self::GlobalZarr(GlobalStreamflowStore::open(path)?))
         } else {
-            Ok(Self::Icechunk(StreamflowStore::open(path)?))
+            Ok(Self::Icechunk(StreamflowStore::open_at(path, snapshot)?))
+        }
+    }
+
+    /// The icechunk snapshot this store reads, when it is icechunk-backed.
+    /// The zarr-v2 global product has no versioning, hence `None`.
+    pub fn snapshot(&self) -> Option<&str> {
+        match self {
+            Self::Icechunk(s) => Some(s.snapshot.as_str()),
+            Self::GlobalZarr(_) => None,
         }
     }
 
@@ -165,3 +203,24 @@ impl std::fmt::Debug for StreamflowSource {
         }
     }
 }
+
+/// `data_sources.pins` names a source, and config validation only knows names —
+/// it cannot tell that the configured `streamflow` path happens to be the
+/// zarr-v2 global product rather than an icechunk repo. Catch that here instead
+/// of opening the store and silently ignoring the pin.
+fn reject_pin_on_non_icechunk(
+    path: &std::path::Path,
+    snapshot: Option<&str>,
+    source: &str,
+) -> Result<()> {
+    match snapshot {
+        None => Ok(()),
+        Some(id) => Err(DataError::Malformed {
+            path: path.to_path_buf(),
+            message: format!(
+                "data_sources.pins.{source} is set to `{id}`, but this store is not an \
+                 icechunk repository — only icechunk sources carry snapshots"
+            ),
+        }),
+    }
+}
diff --git a/tests/cli_lockfile.rs b/tests/cli_lockfile.rs
index bad6fc2c..106e9be5 100644
--- a/tests/cli_lockfile.rs
+++ b/tests/cli_lockfile.rs
@@ -36,3 +36,88 @@ fn diff_lists_drifted_keys() {
     let drift = diff_against_live(&lock, &live);
     assert_eq!(drift, vec!["conus_adjacency".to_string()]);
 }
+
+// ── icechunk snapshots in the lock (pins plan, Task 2) ──────────────────────
+
+fn repo_path(rel: &str) -> std::path::PathBuf {
+    std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join(rel)
+}
+
+/// Juniata-bundle config with absolute paths, so `plan` needs no chdir.
+fn write_juniata_config(dir: &std::path::Path, pins: &str) -> std::path::PathBuf {
+    let b = |n: &str| repo_path(&format!("examples/juniata/data/{n}"));
+    let cfg = dir.join("ddrs.yaml");
+    fs::write(
+        &cfg,
+        format!(
+            "mode: training\nworkflow: train\ngeodataset: merit\nseed: 1\nnp_seed: 1\n\
+             data_sources:\n\
+             \x20 attributes: {}\n\
+             \x20 conus_adjacency: {}\n\
+             \x20 gages_adjacency: {}\n\
+             \x20 streamflow: {}\n\
+             \x20 observations: {}\n\
+             \x20 gages: {}\n{pins}\
+             experiment:\n\
+             \x20 batch_size: 1\n  start_time: 1981/10/01\n  end_time: 1981/12/31\n\
+             \x20 epochs: 1\n  rho: 10\n  warmup: 1\n",
+            b("juniata_attributes.nc").display(),
+            b("juniata_conus_adjacency.zarr").display(),
+            b("juniata_gages_adjacency.zarr").display(),
+            b("juniata_qprime.ic").display(),
+            b("juniata_obs.ic").display(),
+            b("juniata_gage.csv").display(),
+        ),
+    )
+    .unwrap();
+    cfg
+}
+
+fn plan_into_tmp(pins: &str) -> (tempfile::TempDir, Lockfile) {
+    let d = tempfile::tempdir().unwrap();
+    let cfg = write_juniata_config(d.path(), pins);
+    let ws = ddrs::cli::workspace::Workspace::with_root(d.path().join(".ddrs"));
+    ddrs::cli::plan::plan(
+        ddrs::cli::plan::PlanInput {
+            config_path: Some(cfg),
+            skip_smoke: true,
+            ..Default::default()
+        },
+        &ws,
+    )
+    .expect("plan on the Juniata bundle must succeed");
+    let lock = Lockfile::read(&ws.lockfile()).unwrap();
+    (d, lock)
+}
+
+#[test]
+fn plan_locks_icechunk_sources_with_their_snapshot_ids() {
+    let (_d, lock) = plan_into_tmp("");
+
+    for (key, store) in [
+        ("streamflow", "juniata_qprime.ic"),
+        ("observations", "juniata_obs.ic"),
+    ] {
+        let tip = ddrs::data::store::icechunk::main_branch_snapshot(&repo_path(&format!(
+            "examples/juniata/data/{store}"
+        )))
+        .unwrap();
+        let e = lock.sources.get(key).expect("locked");
+        assert_eq!(e.snapshot.as_deref(), Some(tip.as_str()), "{key} snapshot");
+        assert_eq!(e.fp, format!("icechunk:{tip}"), "{key} fp");
+    }
+
+    // Non-icechunk sources stay snapshot-less.
+    assert!(lock.sources["attributes"].snapshot.is_none());
+    assert!(lock.sources["conus_adjacency"].snapshot.is_none());
+}
+
+#[test]
+fn a_pin_is_what_gets_locked() {
+    let qp = repo_path("examples/juniata/data/juniata_qprime.ic");
+    let tip = ddrs::data::store::icechunk::main_branch_snapshot(&qp).unwrap();
+    let (_d, lock) = plan_into_tmp(&format!("  pins:\n    streamflow: {tip}\n"));
+    let e = &lock.sources["streamflow"];
+    assert_eq!(e.snapshot.as_deref(), Some(tip.as_str()));
+    assert_eq!(e.fp, format!("icechunk:{tip}"));
+}
diff --git a/tests/icechunk_pins.rs b/tests/icechunk_pins.rs
new file mode 100644
index 00000000..d7172bf3
--- /dev/null
+++ b/tests/icechunk_pins.rs
@@ -0,0 +1,150 @@
+//! Pinned-snapshot opens of the two icechunk-backed stores
+//! (`data_sources.pins`, Task 2 of the pins plan).
+//!
+//! Uses the committed Juniata bundle (`examples/juniata/data/`), so this
+//! test never skips — the same reason `tests/juniata_bundle.rs` uses it.
+
+use ddrs::data::store::icechunk::main_branch_snapshot;
+use ddrs::data::store::{ObservationsStore, StreamflowSource, StreamflowStore, UsgsObservationsStore};
+
+/// A syntactically valid `SnapshotId` (12 bytes → 20 Crockford base32 chars)
+/// that no repository will ever hold: icechunk's all-zero `FAKE` id.
+const ABSENT_SNAPSHOT: &str = "00000000000000000000";
+
+fn bundle(name: &str) -> std::path::PathBuf {
+    std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("examples/juniata/data")
+        .join(name)
+}
+
+#[test]
+fn streamflow_pinned_at_main_tip_reads_the_same_values_as_unpinned() {
+    let p = bundle("juniata_qprime.ic");
+    let tip = main_branch_snapshot(&p).expect("resolve main tip");
+
+    let unpinned = StreamflowStore::open(&p).expect("open");
+    let pinned = StreamflowStore::open_at(&p, Some(&tip)).expect("open_at main tip");
+
+    assert_eq!(pinned.snapshot, tip, "store records the snapshot it opened");
+    assert_eq!(unpinned.snapshot, tip, "unpinned open resolves main to the same tip");
+
+    let comids: Vec<_> = unpinned.index.ids()[..4].to_vec();
+    let a = unpinned
+        .read_window_daily(unpinned.time_start, 5, &comids)
+        .expect("read unpinned");
+    let b = pinned
+        .read_window_daily(pinned.time_start, 5, &comids)
+        .expect("read pinned");
+    assert_eq!(a, b, "a pin at the main tip must read byte-identical values");
+}
+
+#[test]
+fn streamflow_pinned_at_an_absent_snapshot_errors_naming_path_and_id() {
+    let p = bundle("juniata_qprime.ic");
+    let err = match StreamflowStore::open_at(&p, Some(ABSENT_SNAPSHOT)) {
+        Err(e) => e,
+        Ok(_) => panic!("an absent snapshot id must not open"),
+    };
+    let msg = format!("{err}");
+    assert!(msg.contains("juniata_qprime.ic"), "error must name the source path: {msg}");
+    assert!(msg.contains(ABSENT_SNAPSHOT), "error must name the snapshot id: {msg}");
+}
+
+#[test]
+fn streamflow_pinned_at_a_malformed_snapshot_errors_naming_path_and_id() {
+    let p = bundle("juniata_qprime.ic");
+    let err = match StreamflowStore::open_at(&p, Some("not-a-snapshot-id")) {
+        Err(e) => e,
+        Ok(_) => panic!("a malformed snapshot id must not open"),
+    };
+    let msg = format!("{err}");
+    assert!(msg.contains("juniata_qprime.ic"), "error must name the source path: {msg}");
+    assert!(msg.contains("not-a-snapshot-id"), "error must name the snapshot id: {msg}");
+}
+
+#[test]
+fn observations_pinned_at_main_tip_reads_the_same_values_as_unpinned() {
+    let p = bundle("juniata_obs.ic");
+    let tip = main_branch_snapshot(&p).expect("resolve main tip");
+
+    let unpinned = UsgsObservationsStore::open(&p).expect("open");
+    let pinned = UsgsObservationsStore::open_at(&p, Some(&tip)).expect("open_at main tip");
+    assert_eq!(pinned.snapshot, tip);
+
+    let staids: Vec<_> = unpinned.index.ids().to_vec();
+    let a = unpinned
+        .read_window_daily(unpinned.time_start, 5, &staids)
+        .expect("read unpinned");
+    let b = pinned
+        .read_window_daily(pinned.time_start, 5, &staids)
+        .expect("read pinned");
+    assert_eq!(a, b);
+}
+
+#[test]
+fn observations_pinned_at_an_absent_snapshot_errors_naming_path_and_id() {
+    let p = bundle("juniata_obs.ic");
+    let err = match UsgsObservationsStore::open_at(&p, Some(ABSENT_SNAPSHOT)) {
+        Err(e) => e,
+        Ok(_) => panic!("an absent snapshot id must not open"),
+    };
+    let msg = format!("{err}");
+    assert!(msg.contains("juniata_obs.ic"), "error must name the source path: {msg}");
+    assert!(msg.contains(ABSENT_SNAPSHOT), "error must name the snapshot id: {msg}");
+}
+
+#[test]
+fn format_dispatching_wrappers_thread_the_pin_and_report_the_snapshot() {
+    let qp = bundle("juniata_qprime.ic");
+    let ob = bundle("juniata_obs.ic");
+    let qp_tip = main_branch_snapshot(&qp).unwrap();
+    let ob_tip = main_branch_snapshot(&ob).unwrap();
+
+    let sf = StreamflowSource::open_at(&qp, Some(&qp_tip)).expect("open_at streamflow");
+    assert_eq!(sf.snapshot(), Some(qp_tip.as_str()));
+
+    let obs = ObservationsStore::open_at(&ob, Some(&ob_tip)).expect("open_at observations");
+    assert_eq!(obs.snapshot(), Some(ob_tip.as_str()));
+
+    // Unpinned goes through the same accessor and still reports the tip.
+    assert_eq!(
+        StreamflowSource::open(&qp).unwrap().snapshot(),
+        Some(qp_tip.as_str())
+    );
+
+    let err = match StreamflowSource::open_at(&qp, Some(ABSENT_SNAPSHOT)) {
+        Err(e) => e,
+        Ok(_) => panic!("an absent snapshot id must not open"),
+    };
+    assert!(format!("{err}").contains(ABSENT_SNAPSHOT));
+}
+
+/// End-to-end: `data_sources.pins` reaches the two stores through
+/// `Config` → `MeritGagesDataset::open`. Run with `--nocapture` to see the
+/// `streamflow snapshot: <id> (pinned)` line this exercises.
+#[test]
+fn a_pinned_config_opens_the_juniata_dataset() {
+    use ddrs::config::Config;
+    use ddrs::data::MeritGagesDataset;
+
+    let tip_q = main_branch_snapshot(&bundle("juniata_qprime.ic")).unwrap();
+    let tip_o = main_branch_snapshot(&bundle("juniata_obs.ic")).unwrap();
+
+    // Integration tests run with the package root as cwd, so the example
+    // config's repo-root-relative data paths resolve unchanged.
+    let anchor = "  gages: examples/juniata/data/juniata_gage.csv";
+    let src = std::fs::read_to_string("examples/juniata/ddrs.yaml").unwrap();
+    assert!(src.contains(anchor), "example config layout changed");
+    let pinned = src.replace(
+        anchor,
+        &format!("{anchor}\n  pins:\n    streamflow: {tip_q}\n    observations: {tip_o}"),
+    );
+
+    let d = tempfile::tempdir().unwrap();
+    let p = d.path().join("pinned.yaml");
+    std::fs::write(&p, pinned).unwrap();
+
+    let cfg = Config::from_yaml_file(&p).expect("pinned config loads");
+    let ds = MeritGagesDataset::open(&cfg).expect("open the dataset at pinned snapshots");
+    assert_eq!(ds.len(), 1);
+}

From a65be9d114fbefd3ff084dee151556aa53cbc646 Mon Sep 17 00:00:00 2001
From: Tadd Bindas <taddbindas@gmail.com>
Date: Wed, 16 Sep 2026 09:17:08 -0400
Subject: [PATCH 3/4] baseline: invalidate the summed-Q' cache when a pin
 changes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`summed_q_prime` now reads `data_sources.pins`' snapshot, so changing a pin
changes the baseline's inputs. `cache_key` did not hash the pins, so the
second run silently reused the first snapshot's cached predictions and
observations — and the baseline is what a trained model is judged against,
so a stale one invalidates the comparison the run exists to make.

The pins block is hashed only when at least one pin is present, so an
unpinned config yields the byte-identical key it did before pins existed and
none of the existing cached baselines is invalidated. `BASELINE_ALGO_VERSION`
is deliberately NOT bumped: the algorithm is unchanged (that constant means
"recompute stale caches from older binaries"), this is a new input.

Three tests in `cache.rs`'s module: the unpinned key is pinned to its
literal `43843300d30ff837` (measured before the change) so any future change
to the key's inputs fails loudly; two pin values give different keys; a
pinned and an unpinned config give different keys, while an EMPTY pins map
hashes as unpinned.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GbK4a8dyT5eNDU97CUWHYZ
---
 src/baseline/cache.rs | 68 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 68 insertions(+)

diff --git a/src/baseline/cache.rs b/src/baseline/cache.rs
index 944a2596..c2c9006c 100644
--- a/src/baseline/cache.rs
+++ b/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())
 }
@@ -432,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();

From 75df2e764746afc917e7f3e681a5de76783eeccf Mon Sep 17 00:00:00 2001
From: Tadd Bindas <taddbindas@gmail.com>
Date: Wed, 16 Sep 2026 09:40:43 -0400
Subject: [PATCH 4/4] docs: record snapshot pins and content fingerprints
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CLAUDE.md §Data sources gains a paragraph on the new fingerprint scheme
(icechunk repo -> main-branch snapshot id, other directories -> recursive
path+size walk), the `sources.<name>.snapshot` field now written to both
`sources.lock` and the run manifest, the optional `data_sources.pins` map,
and the one-off drift every existing lock will report.

The ddrs-dev skill's config.md records that the summed-Q' cache key hashes
the pins, so re-pinning recomputes the baseline instead of reusing the
previous snapshot's; testing.md gains change-to-gate rows for
`src/cli/fingerprint.rs` and the icechunk reader / pins.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GbK4a8dyT5eNDU97CUWHYZ
---
 .claude/skills/ddrs-dev/references/config.md  |  6 ++++++
 .claude/skills/ddrs-dev/references/testing.md |  2 ++
 CLAUDE.md                                     | 15 +++++++++++++++
 3 files changed, 23 insertions(+)

diff --git a/.claude/skills/ddrs-dev/references/config.md b/.claude/skills/ddrs-dev/references/config.md
index 6021deb0..60e4e7ff 100644
--- a/.claude/skills/ddrs-dev/references/config.md
+++ b/.claude/skills/ddrs-dev/references/config.md
@@ -67,6 +67,12 @@ and `ddrs plan` writes it to `sources.lock` and the run manifest as
 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:
diff --git a/.claude/skills/ddrs-dev/references/testing.md b/.claude/skills/ddrs-dev/references/testing.md
index 0f7f5928..d6d1f0e2 100644
--- a/.claude/skills/ddrs-dev/references/testing.md
+++ b/.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` |
diff --git a/CLAUDE.md b/CLAUDE.md
index a8a7feda..d64a14e3 100644
--- a/CLAUDE.md
+++ b/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