From 65f30389bc14ae0b53037e2831f3168486d699b3 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:30:05 -0700
Subject: [PATCH 01/15] perf(dynacell/eval): read and write cache manifests
 with libyaml

The manifest is saved and reloaded after every FOV, and #511's
per-position sources made it several times larger. OmegaConf round-trips
it through its own node tree plus the pure-Python YAML emitter.

PyYAML's C loader and dumper produce byte-identical files and equal
dicts. On a real 6-family x 100-position manifest with 600 sources,
save+load drops from 0.299 s to 0.022 s. Interpolation resolution is
dropped with OmegaConf; no production manifest contains `${`. The
pure-Python classes remain the fallback for a PyYAML without libyaml.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py     | 15 ++++++++++++---
 1 file changed, 12 insertions(+), 3 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index 124266e4f..1595cecc2 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -27,14 +27,21 @@
 
 import numpy as np
 import numpy.typing as npt
+import yaml
 import zarr
 from iohub.ngff import ImageArray, Position, open_ome_zarr
-from omegaconf import OmegaConf
 
 from viscy_utils.prediction_metadata import PREDICTION_COMPLETE_KEY, marker_identity
 
 FeatureKind = Literal["cp", "dinov3", "dynaclr", "celldino", "morphem"]
 
+# The manifest is read and written once per FOV, and its per-position ``sources``
+# make it large; libyaml's C loader and dumper are an order of magnitude faster than
+# the pure-Python ones. They are absent only from a PyYAML built without libyaml,
+# which then falls back to the pure-Python classes.
+_YAML_LOADER = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
+_YAML_DUMPER = getattr(yaml, "CSafeDumper", yaml.SafeDumper)
+
 CACHE_SCHEMA_VERSION = 1
 
 _MASK_CHANNEL = "target_seg"
@@ -118,7 +125,8 @@ def load_manifest(paths: CachePaths) -> dict[str, Any]:
             "cell_segmentation": None,
             "artifacts": {},
         }
-    raw = OmegaConf.to_container(OmegaConf.load(paths.manifest), resolve=True)
+    with open(paths.manifest) as f:
+        raw = yaml.load(f, Loader=_YAML_LOADER)
     if not isinstance(raw, dict):
         raise StaleCacheError(f"Manifest at {paths.manifest} is not a mapping")
     raw.setdefault("gt", None)
@@ -131,7 +139,8 @@ def load_manifest(paths: CachePaths) -> dict[str, Any]:
 def save_manifest(paths: CachePaths, manifest: dict[str, Any]) -> None:
     """Persist *manifest* as YAML under *paths.manifest*, creating parents."""
     paths.root.mkdir(parents=True, exist_ok=True)
-    OmegaConf.save(OmegaConf.create(manifest), paths.manifest)
+    with open(paths.manifest, "w") as f:
+        yaml.dump(manifest, f, Dumper=_YAML_DUMPER, sort_keys=False)
 
 
 def check_cache_identity(

From 7cb60703ee5dad5dd5728d3afc28aa727fcc5772 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:31:14 -0700
Subject: [PATCH 02/15] refactor(dynacell/eval): share one JSON sha256_12
 helper in the cache

The marker digest, the prediction-sources digest and the encoder-config
hash each built the same key-sorted JSON payload and truncated sha256
by hand. json_sha256_12 does it once; digests are unchanged (checked on
a real marker, a real 100-position source snapshot and a non-JSON
encoder value, which default=str still serializes).

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py | 21 ++++++++++++-------
 1 file changed, 13 insertions(+), 8 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index 1595cecc2..97588099d 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -281,6 +281,16 @@ def built_at_now() -> str:
     return datetime.now(timezone.utc).isoformat(timespec="seconds")
 
 
+def json_sha256_12(obj: Any) -> str:
+    """Return the first 12 hex chars of the sha256 of *obj* serialized as JSON.
+
+    Keys are sorted, so representation-equivalent mappings hash alike; values JSON
+    cannot encode are serialized with ``str``.
+    """
+    payload = json.dumps(obj, sort_keys=True, default=str).encode("utf-8")
+    return hashlib.sha256(payload).hexdigest()[:12]
+
+
 def _chunk_files(directory: Path) -> Iterator[os.DirEntry]:
     """Yield the chunk files under *directory*, depth first and lazily.
 
@@ -353,10 +363,7 @@ def _first_channel_chunk_mtime_ns(plate_path: Path, array: ImageArray, channel_i
 def _position_source(plate_path: Path, position: Position, channel_name: str, archive_ns: int | None) -> dict[str, Any]:
     """Return one position's source; see :func:`prediction_sources`."""
     marker = position.zattrs.get(PREDICTION_COMPLETE_KEY, {}).get(channel_name)
-    digest = None
-    if marker is not None:
-        payload = json.dumps(marker_identity(marker), sort_keys=True).encode("utf-8")
-        digest = hashlib.sha256(payload).hexdigest()[:12]
+    digest = None if marker is None else json_sha256_12(marker_identity(marker))
     if archive_ns is not None:
         written_ns = archive_ns
     else:
@@ -448,8 +455,7 @@ def prediction_sources_sha256_12(sources: dict[str, dict[str, Any]]) -> str:
     str
         First 12 hex characters of the digest.
     """
-    payload = json.dumps(sorted(sources.items()), sort_keys=True).encode("utf-8")
-    return hashlib.sha256(payload).hexdigest()[:12]
+    return json_sha256_12(sorted(sources.items()))
 
 
 def _read_position_channel0(plate_path: Path, pos_name: str, dtype: npt.DTypeLike) -> np.ndarray | None:
@@ -778,8 +784,7 @@ def encoder_config_sha256_12(encoder_cfg: dict[str, Any]) -> str:
 
     Keys are sorted so representation-equivalent configs produce the same hash.
     """
-    payload = json.dumps(encoder_cfg, sort_keys=True, default=str).encode("utf-8")
-    return hashlib.sha256(payload).hexdigest()[:12]
+    return json_sha256_12(encoder_cfg)
 
 
 def feature_slug(name: str) -> str:

From c7b359836546f5e8335a3f56457ce61130797190 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:32:38 -0700
Subject: [PATCH 03/15] refactor(dynacell/eval): drop the flat zarr v2
 chunk-key fallback scan

The bounded scandir for `.`-separated keys served only flat zarr v2
layouts, and every evaluated non-v3 prediction store uses `/`
separators, so nothing reaches it. A flat store now resolves only its
exact chunk key; when that chunk is absent the position reads as
unwritten, which rebuilds a legacy cache entry over it once. The `/`
prefix fallback and the exact-key lookup for every layout stay.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../src/dynacell/evaluation/README.md         |  2 +-
 .../dynacell/src/dynacell/evaluation/cache.py | 27 +++++++------------
 .../test_evaluation_pred_cache_identity.py    | 25 ++---------------
 3 files changed, 12 insertions(+), 42 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/README.md b/applications/dynacell/src/dynacell/evaluation/README.md
index 7cd86544a..45f10fc1f 100644
--- a/applications/dynacell/src/dynacell/evaluation/README.md
+++ b/applications/dynacell/src/dynacell/evaluation/README.md
@@ -219,7 +219,7 @@ Each per-artifact flag invalidates that family for its side only:
 
 Without `io.gt_cache_dir` / `io.pred_cache_dir`, only `force_recompute.{final_metrics, all}` matter.
 
-A re-predict into the same `io.pred_path` needs none of these flags. The prediction side records, per cached position, the source it was built from: a hash of the position's writer marker (`viscy_prediction_complete` for `io.pred_channel_name`: checkpoint content hash, settings hash, depth handling, source shape) and the mtime of the first chunk of that channel at `t=0`. A position whose source no longer matches the store is an ordinary cache miss, recomputed and re-recorded (a `StaleCacheError` under `io.require_complete_cache=true`), so an interrupted rebuild resumes where it stopped and a code-only re-predict (same checkpoint and settings, new chunks) is caught. Writes to other channels or to position metadata do not count. Entries written before sources existed are upgraded on the first run, per position: a position is current iff its first stored chunk is no newer than the entry's `built_at`; a position with no stored chunk (a blank prediction) cannot be dated and is rebuilt once. The upgrade rewrites the manifest, so the cache directory must be writable even under `io.require_complete_cache=true`. `metrics_provenance.json` records a digest of all positions' sources, taken before scoring, and `_final_metrics_cache_valid` rejects saved metrics when it differs (an older sidecar: when any chunk is newer than the sidecar, or a position has no stored chunk). Metrics saved after a re-predict but scored from caches it had left stale are the one legacy case this cannot see. A missing prediction store raises `FileNotFoundError` wherever its sources are read, the final-metrics gate included. A zarr v2 store with `.`-separated chunk keys is scanned for its first chunk over at most 4096 directory entries and can read as unwritten beyond that; the current writer only writes zarr v3. The GT side is not tracked.
+A re-predict into the same `io.pred_path` needs none of these flags. The prediction side records, per cached position, the source it was built from: a hash of the position's writer marker (`viscy_prediction_complete` for `io.pred_channel_name`: checkpoint content hash, settings hash, depth handling, source shape) and the mtime of the first chunk of that channel at `t=0`. A position whose source no longer matches the store is an ordinary cache miss, recomputed and re-recorded (a `StaleCacheError` under `io.require_complete_cache=true`), so an interrupted rebuild resumes where it stopped and a code-only re-predict (same checkpoint and settings, new chunks) is caught. Writes to other channels or to position metadata do not count. Entries written before sources existed are upgraded on the first run, per position: a position is current iff its first stored chunk is no newer than the entry's `built_at`; a position with no stored chunk (a blank prediction) cannot be dated and is rebuilt once. The upgrade rewrites the manifest, so the cache directory must be writable even under `io.require_complete_cache=true`. `metrics_provenance.json` records a digest of all positions' sources, taken before scoring, and `_final_metrics_cache_valid` rejects saved metrics when it differs (an older sidecar: when any chunk is newer than the sidecar, or a position has no stored chunk). Metrics saved after a re-predict but scored from caches it had left stale are the one legacy case this cannot see. A missing prediction store raises `FileNotFoundError` wherever its sources are read, the final-metrics gate included. The GT side is not tracked.
 
 ### Invalidation
 
diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index 97588099d..fbd441c40 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -14,7 +14,6 @@
 from __future__ import annotations
 
 import hashlib
-import itertools
 import json
 import os
 import shutil
@@ -306,32 +305,24 @@ def _chunk_files(directory: Path) -> Iterator[os.DirEntry]:
                 yield entry
 
 
-#: Directory entries a ``.``-separated chunk-key fallback scans before giving up. Only
-#: zarr v2 flat layouts reach it, and a store holding more entries than this ahead of
-#: the channel's chunks reads as having none (``written_ns=None``); the current writer
-#: only writes zarr v3, so those stores are not re-predicted.
-_FLAT_CHUNK_SCAN_LIMIT = 4096
-
-
 def _stored_chunk(array_dir: Path, key: str, spatial: int) -> os.DirEntry | Path | None:
     """Return the chunk stored at *key*, else the first stored chunk sharing its ``(t, c)`` prefix.
 
     A chunk never written, or reset to the fill value (zarr deletes it), is absent. The
-    prefix scan is bounded: one directory batch per level for ``/``-separated keys, at
-    most :data:`_FLAT_CHUNK_SCAN_LIMIT` entries for ``.``-separated ones.
+    prefix fallback exists only for ``/``-separated keys, whose ``(t, c)`` chunks share
+    one directory subtree, read one directory batch per level. A ``.``-separated key (a
+    flat zarr v2 layout) resolves only exactly; when that chunk is absent the position
+    reads as unwritten (``written_ns=None``), which rebuilds a legacy entry over it once.
+    Every evaluated non-v3 store uses ``/`` separators.
     """
     chunk = array_dir / key
     if chunk.is_file():
         return chunk
     parts = key.split("/")
-    if len(parts) > spatial:
-        # "/"-separated coordinates: the (t, c) chunks share one directory subtree.
-        prefix = array_dir.joinpath(*parts[:-spatial])
-        return next(_chunk_files(prefix), None) if prefix.is_dir() else None
-    # "."-separated coordinates: the (t, c) chunks share a name prefix.
-    stem = key.rsplit(".", spatial)[0] + "."
-    with os.scandir(array_dir) as entries:
-        return next((e for e in itertools.islice(entries, _FLAT_CHUNK_SCAN_LIMIT) if e.name.startswith(stem)), None)
+    if len(parts) <= spatial:
+        return None
+    prefix = array_dir.joinpath(*parts[:-spatial])
+    return next(_chunk_files(prefix), None) if prefix.is_dir() else None
 
 
 def _first_channel_chunk_mtime_ns(plate_path: Path, array: ImageArray, channel_index: int) -> int | None:
diff --git a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
index 7bead3d4b..8e93d5b48 100644
--- a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
+++ b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
@@ -14,14 +14,12 @@
 
 import json
 import os
-import shutil
 import warnings
 from datetime import datetime
 from pathlib import Path
 
 import numpy as np
 import pytest
-import zarr
 from iohub.ngff import open_ome_zarr
 from omegaconf import OmegaConf
 
@@ -599,24 +597,7 @@ def test_sources_track_marker_and_chunk_not_provenance(tmp_path: Path, version:
     assert prediction_sources(pred, _CHANNEL)["A/1/1"]["written_ns"] > first["A/1/1"]["written_ns"]
 
 
-def _flat_v2(path: Path, data: np.ndarray) -> None:
-    """Replace array ``0`` of every position with a zarr v2 array using ``.``-separated chunk keys."""
-    for name in _POSITIONS:
-        shutil.rmtree(path / name / "0")
-        array = zarr.create_array(
-            store=str(path / name),
-            name="0",
-            shape=data.shape,
-            chunks=(1, 1, 1, _H, _W),
-            dtype="f4",
-            zarr_format=2,
-            chunk_key_encoding={"name": "v2", "separator": "."},
-            fill_value=0,
-        )
-        array[:] = data
-
-
-@pytest.mark.parametrize("layout", ["zarr-v3", "zarr-v3-sharded", "zarr-v2", "zarr-v2-flat"])
+@pytest.mark.parametrize("layout", ["zarr-v3", "zarr-v3-sharded", "zarr-v2"])
 def test_source_chunk_is_found_in_every_layout(tmp_path: Path, layout: str) -> None:
     """The channel's ``t=0`` chunk is found through the array's own key encoding, past a fill-value first chunk."""
     pred = tmp_path / "pred.zarr"
@@ -627,8 +608,6 @@ def test_source_chunk_is_found_in_every_layout(tmp_path: Path, layout: str) -> N
     with open_ome_zarr(pred, mode="w", layout="hcs", channel_names=[_CHANNEL, _OTHER], version=version) as plate:
         for name in _POSITIONS:
             plate.create_position(*name.split("/")).create_image("0", data, chunks=(1, 1, 1, _H, _W), **extra)
-    if layout == "zarr-v2-flat":
-        _flat_v2(pred, data)
     chunks = sorted(
         f for f in (pred / "A/1/0/0").rglob("*") if f.is_file() and f.name != "zarr.json" and f.name[0] != "."
     )
@@ -637,7 +616,7 @@ def test_source_chunk_is_found_in_every_layout(tmp_path: Path, layout: str) -> N
 
     written = prediction_sources(pred, _OTHER)["A/1/0"]["written_ns"]
     (match,) = [chunk for chunk in chunks if chunk.stat().st_mtime_ns == written]
-    key = str(match.relative_to(pred / "A/1/0/0")).replace(".", "/").removeprefix("c/")
+    key = str(match.relative_to(pred / "A/1/0/0")).removeprefix("c/")
     assert key.split("/")[:2] == ["0", "1"], key  # t=0, channel 1
 
 

From d879bd2258a7c24a780e54e8c2e2fc59ca2f4a88 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:33:50 -0700
Subject: [PATCH 04/15] refactor(dynacell/eval): read single positions through
 prediction_sources

prediction_source duplicated prediction_sources for one position; a
positions= argument covers it. The on-demand read for a position that
appeared after the snapshot now takes the store and channel from the
manifest's pred identity, which init_cache_context already seeds and
validates, so _CacheContext no longer carries its own copy of them.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py | 34 ++++++-------------
 .../src/dynacell/evaluation/pipeline_cache.py | 13 +++----
 2 files changed, 16 insertions(+), 31 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index fbd441c40..a304e932a 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -17,7 +17,7 @@
 import json
 import os
 import shutil
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timezone
@@ -368,8 +368,10 @@ def _archive_mtime_ns(path: Path) -> int | None:
     return path.stat().st_mtime_ns if path.is_file() else None
 
 
-def prediction_sources(plate_path: Path | str, channel_name: str) -> dict[str, dict[str, Any]]:
-    """Return the source identity of every position of one prediction channel.
+def prediction_sources(
+    plate_path: Path | str, channel_name: str, positions: Iterable[str] | None = None
+) -> dict[str, dict[str, Any]]:
+    """Return the source identity of every position (or of *positions*) of one prediction channel.
 
     A position's source is what a cached artifact built from it must still match:
 
@@ -402,6 +404,10 @@ def prediction_sources(plate_path: Path | str, channel_name: str) -> dict[str, d
         HCS prediction store; a missing store raises ``FileNotFoundError``.
     channel_name : str
         Prediction channel the eval scores.
+    positions : iterable of str, optional
+        Read only these positions, e.g. one a running predict finished after the
+        store-wide snapshot; each must exist (``KeyError`` otherwise). ``None`` reads
+        every position.
 
     Returns
     -------
@@ -411,26 +417,8 @@ def prediction_sources(plate_path: Path | str, channel_name: str) -> dict[str, d
     path = Path(plate_path)
     archive_ns = _archive_mtime_ns(path)
     with open_ome_zarr(path, mode="r") as plate:
-        return {
-            name: _position_source(path, position, channel_name, archive_ns) for name, position in plate.positions()
-        }
-
-
-def prediction_source(plate_path: Path | str, position_name: str, channel_name: str) -> dict[str, Any]:
-    """Return the :func:`prediction_sources` entry of one position, read on its own.
-
-    For a position that appeared after the store-wide snapshot, e.g. one a running
-    predict finished while an ``io.exclude_fov_names`` eval was already open.
-
-    Raises
-    ------
-    KeyError
-        If the store has no such position.
-    """
-    path = Path(plate_path)
-    archive_ns = _archive_mtime_ns(path)
-    with open_ome_zarr(path, mode="r") as plate:
-        return _position_source(path, plate[position_name], channel_name, archive_ns)
+        items = plate.positions() if positions is None else ((name, plate[name]) for name in positions)
+        return {name: _position_source(path, position, channel_name, archive_ns) for name, position in items}
 
 
 def prediction_sources_sha256_12(sources: dict[str, dict[str, Any]]) -> str:
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index aaf30300b..1665715d6 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -34,7 +34,6 @@
     feature_slug,
     load_manifest,
     open_features_group,
-    prediction_source,
     prediction_sources,
     read_features_from_group,
     read_instance_mask,
@@ -115,9 +114,6 @@ class _CacheContext:
     celldino_preprocess_version: str | None = None
     morphem_preprocess_version: str | None = None
     prediction_sources: dict[str, dict[str, Any]] | None = None
-    # Store and channel the sources come from, to read a position added after the snapshot.
-    prediction_plate_path: str | None = None
-    prediction_channel_name: str | None = None
     _manifest_dirty: bool = field(default=False, init=False, repr=False)
     # Positions this process recorded a source for, per manifest path, since the last flush.
     _written_sources: dict[tuple[str, ...], set[str]] = field(default_factory=dict, init=False, repr=False)
@@ -397,8 +393,6 @@ def init_cache_context(
         # Read once, before any artifact is built: a position written by this run is
         # recorded with its source as it stood when the run started.
         prediction_sources=prediction_sources(plate_path, channel_name) if side == "pred" else None,
-        prediction_plate_path=plate_path if side == "pred" else None,
-        prediction_channel_name=channel_name if side == "pred" else None,
         **base_kwargs,
     )
     _auto_invalidate_on_artifact_param_mismatch(ctx)
@@ -715,11 +709,14 @@ def _prediction_source(ctx: _CacheContext, pos_name: str) -> dict[str, Any]:
 
     A running predict can finish a position between :func:`init_cache_context` and the
     walk that lists it (an ``io.exclude_fov_names`` eval over a partial store). The
-    on-demand read is cached; a position the store does not have raises ``KeyError``.
+    on-demand read comes from the store and channel the manifest's ``pred`` identity
+    records (seeded and checked by :func:`init_cache_context`) and is cached; a
+    position the store does not have raises ``KeyError``.
     """
     source = ctx.prediction_sources.get(pos_name)
     if source is None:
-        source = prediction_source(ctx.prediction_plate_path, pos_name, ctx.prediction_channel_name)
+        pred = ctx.manifest["pred"]
+        source = prediction_sources(pred["plate_path"], pred["channel_name"], [pos_name])[pos_name]
         ctx.prediction_sources[pos_name] = source
     return source
 

From 683d2a405f52e15fcb70586637cdf29f36bc3966 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:37:19 -0700
Subject: [PATCH 05/15] refactor(dynacell/eval): date legacy caches and
 sidecars by one rule

The legacy manifest upgrade and the final-metrics gate each spelled out
"a position is verified iff its first stored chunk exists and is no
newer than the legacy record's time". source_predates states it once;
the upgrade checks the missing-built_at case once instead of per
position, and drops an isinstance guard on the dict it just wrote.

The upgrade also no longer builds per-family identities it discards:
_artifact_checks splits into _artifact_entries (kind, label, manifest
keys) and _artifact_identity, which only the param check needs.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py |  23 +++
 .../src/dynacell/evaluation/pipeline.py       |   6 +-
 .../src/dynacell/evaluation/pipeline_cache.py | 156 +++++++++---------
 3 files changed, 101 insertions(+), 84 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index a304e932a..f969d42aa 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -421,6 +421,29 @@ def prediction_sources(
         return {name: _position_source(path, position, channel_name, archive_ns) for name, position in items}
 
 
+def source_predates(source: dict[str, Any], horizon_ns: int) -> bool:
+    """Return whether a :func:`prediction_sources` entry was written no later than *horizon_ns*.
+
+    Dates caches recorded before sources existed, which carry only a time: a manifest
+    entry's ``built_at``, or a metrics sidecar's mtime. A position whose first stored
+    chunk (``written_ns``) is no newer than that time is verified; one with no stored
+    chunk cannot be dated and never is.
+
+    Parameters
+    ----------
+    source : dict
+        One position's ``{"marker", "written_ns"}``.
+    horizon_ns : int
+        The legacy record's time, in ns since the epoch.
+
+    Returns
+    -------
+    bool
+        ``written_ns is not None and written_ns <= horizon_ns``.
+    """
+    return source["written_ns"] is not None and source["written_ns"] <= horizon_ns
+
+
 def prediction_sources_sha256_12(sources: dict[str, dict[str, Any]]) -> str:
     """Return the sha256_12 over the name-sorted ``(position, source)`` pairs of :func:`prediction_sources`.
 
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline.py b/applications/dynacell/src/dynacell/evaluation/pipeline.py
index 3b62b7c80..1db0320eb 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline.py
@@ -17,7 +17,7 @@
 from tqdm import tqdm
 
 from dynacell.evaluation._ref_hook import apply_dataset_ref
-from dynacell.evaluation.cache import FeatureKind, prediction_sources, prediction_sources_sha256_12
+from dynacell.evaluation.cache import FeatureKind, prediction_sources, prediction_sources_sha256_12, source_predates
 from dynacell.evaluation.cp_reference import (
     CP_SIDECAR_FILENAME,
     DatasetCPSpace,
@@ -2106,8 +2106,8 @@ def _metrics_scored_current_prediction(config: DictConfig, save_dir: Path) -> bo
     recorded = json.loads(stamp.read_text()).get("prediction_sources_sha256_12")
     if recorded is not None:
         return recorded == prediction_sources_sha256_12(sources)
-    written = [source["written_ns"] for source in sources.values()]
-    return None not in written and max(written, default=0) <= stamp.stat().st_mtime_ns
+    saved_ns = stamp.stat().st_mtime_ns
+    return all(source_predates(source, saved_ns) for source in sources.values())
 
 
 def _load_cached_final_metrics(config: DictConfig) -> tuple[list, list, list]:
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index 1665715d6..b80fd0443 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -40,6 +40,7 @@
     read_mask,
     save_manifest,
     seed_cache_identity,
+    source_predates,
     write_features_to_group,
     write_instance_mask,
     write_mask,
@@ -515,8 +516,9 @@ def _auto_invalidate_on_artifact_param_mismatch(ctx: _CacheContext) -> None:
     """
     if not ctx.enabled:
         return
-    for kind, artifact_label, entry, current, numeric_keys in _artifact_checks(ctx):
-        mismatches = diff_artifact_params(entry, current, numeric_keys=numeric_keys)
+    for kind, artifact_label, keys in _artifact_entries(ctx):
+        current, numeric_keys = _artifact_identity(ctx, kind)
+        mismatches = diff_artifact_params(_manifest_entry(ctx, keys), current, numeric_keys=numeric_keys)
         if not mismatches:
             continue
         details = ", ".join(f"{key}: cached={cached!r}, current={cur!r}" for key, cached, cur in mismatches)
@@ -537,103 +539,92 @@ def _auto_invalidate_on_artifact_param_mismatch(ctx: _CacheContext) -> None:
         )
 
 
-def _artifact_checks(
-    ctx: _CacheContext,
-) -> list[tuple[str, str, dict[str, Any] | None, dict[str, Any], tuple[str, ...]]]:
-    """Return one ``(kind, label, entry, current identity, numeric keys)`` per artifact family this run reads.
-
-    ``entry`` is the family's manifest entry under its current key (``None`` when
-    absent); ``current`` is the identity this run would write, compared by
-    :func:`diff_artifact_params` over ``numeric keys`` with a tolerance.
-    """
-    artifacts = ctx.manifest.get("artifacts", {})
-
-    checks: list[tuple[str, str, dict[str, Any] | None, dict[str, Any], tuple[str, ...]]] = [
-        (
-            # Key by ``mask_stem`` (= ``{target}__{backend}`` for non-default
-            # backends), the same stem as the cache plate path, so masks from
-            # different backends get separate manifest entries instead of
-            # clobbering one ``organelle_masks[target_name]`` slot. ``backend`` is
-            # encoded in the stem itself, so it is intentionally NOT a param in
-            # the identity dict below (a backend switch lands on a different
-            # entry/plate — there is nothing stale to invalidate, and adding it
-            # would spuriously invalidate pre-existing caches that predate the
-            # field).
-            "masks",
-            f"{ctx.label_prefix}organelle_masks[{ctx.mask_stem}]",
-            artifacts.get("organelle_masks", {}).get(ctx.mask_stem),
-            {"target_name": ctx.target_name, **ctx.source_tag},
-            (),
-        ),
-        (
-            "cp",
-            f"{ctx.label_prefix}cp_features",
-            artifacts.get("cp_features"),
-            _cp_identity(ctx),
-            ("spacing", "cp_norm_p_lo", "cp_norm_p_hi"),
-        ),
+def _artifact_entries(ctx: _CacheContext) -> list[tuple[str, str, list[str]]]:
+    """Return one ``(kind, label, manifest keys)`` per artifact family this run reads."""
+    families = [
+        # Key by ``mask_stem`` (= ``{target}__{backend}`` for non-default
+        # backends), the same stem as the cache plate path, so masks from
+        # different backends get separate manifest entries instead of
+        # clobbering one ``organelle_masks[target_name]`` slot.
+        ("masks", f"{ctx.label_prefix}organelle_masks[{ctx.mask_stem}]", ["organelle_masks", ctx.mask_stem]),
+        ("cp", f"{ctx.label_prefix}cp_features", ["cp_features"]),
     ]
     if ctx.compute_instance_ap:
         stem = f"{ctx.target_name}__{ctx.backend}"
-        checks.append(
-            (
-                "instances",
-                f"{ctx.label_prefix}instance_masks[{stem}]",
-                artifacts.get("instance_masks", {}).get(stem),
-                _instance_identity(ctx),
-                (),
-            )
-        )
+        families.append(("instances", f"{ctx.label_prefix}instance_masks[{stem}]", ["instance_masks", stem]))
     if ctx.dinov3_model_name is not None:
-        checks.append(
+        families.append(
             (
                 "dinov3",
                 f"{ctx.label_prefix}dinov3_features[{ctx.dinov3_model_name}]",
-                artifacts.get("dinov3_features", {}).get(feature_slug(ctx.dinov3_model_name)),
-                {"model_name": ctx.dinov3_model_name, "patch_size": ctx.patch_size, **ctx.source_tag},
-                (),
+                ["dinov3_features", feature_slug(ctx.dinov3_model_name)],
             )
         )
     if ctx.dynaclr_ckpt_sha12 is not None:
-        checks.append(
+        families.append(
             (
                 "dynaclr",
                 f"{ctx.label_prefix}dynaclr_features[{ctx.dynaclr_ckpt_sha12}]",
-                artifacts.get("dynaclr_features", {}).get(ctx.dynaclr_ckpt_sha12),
-                {
-                    "checkpoint_sha256_12": ctx.dynaclr_ckpt_sha12,
-                    "encoder_config_sha256_12": ctx.dynaclr_encoder_sha12,
-                    "patch_size": ctx.patch_size,
-                    **ctx.source_tag,
-                },
-                (),
+                ["dynaclr_features", ctx.dynaclr_ckpt_sha12],
             )
         )
     if ctx.celldino_weights_sha12 is not None:
-        checks.append(
+        families.append(
             (
                 "celldino",
                 f"{ctx.label_prefix}celldino_features[{ctx.celldino_weights_sha12}]",
-                artifacts.get("celldino_features", {}).get(ctx.celldino_weights_sha12),
-                {
-                    "weights_sha256_12": ctx.celldino_weights_sha12,
-                    "patch_size": ctx.patch_size,
-                    **ctx.source_tag,
-                },
-                (),
+                ["celldino_features", ctx.celldino_weights_sha12],
             )
         )
     if ctx.morphem_model_name is not None:
-        checks.append(
+        families.append(
             (
                 "morphem",
                 f"{ctx.label_prefix}morphem_features[{ctx.morphem_model_name}]",
-                artifacts.get("morphem_features", {}).get(feature_slug(ctx.morphem_model_name)),
-                {"model_name": ctx.morphem_model_name, "patch_size": ctx.patch_size, **ctx.source_tag},
-                (),
+                ["morphem_features", feature_slug(ctx.morphem_model_name)],
             )
         )
-    return checks
+    return families
+
+
+def _artifact_identity(ctx: _CacheContext, kind: str) -> tuple[dict[str, Any], tuple[str, ...]]:
+    """Return the identity a *kind* entry must record, and the keys compared with a numeric tolerance.
+
+    The masks identity leaves out ``backend``: it is encoded in the entry's stem
+    (see :func:`_artifact_entries`), so a backend switch lands on a different
+    entry and plate -- there is nothing stale to invalidate, and adding it would
+    spuriously invalidate pre-existing caches that predate the field.
+    """
+    if kind == "masks":
+        return {"target_name": ctx.target_name, **ctx.source_tag}, ()
+    if kind == "cp":
+        return _cp_identity(ctx), ("spacing", "cp_norm_p_lo", "cp_norm_p_hi")
+    if kind == "instances":
+        return _instance_identity(ctx), ()
+    if kind == "dinov3":
+        return {"model_name": ctx.dinov3_model_name, "patch_size": ctx.patch_size, **ctx.source_tag}, ()
+    if kind == "dynaclr":
+        return {
+            "checkpoint_sha256_12": ctx.dynaclr_ckpt_sha12,
+            "encoder_config_sha256_12": ctx.dynaclr_encoder_sha12,
+            "patch_size": ctx.patch_size,
+            **ctx.source_tag,
+        }, ()
+    if kind == "celldino":
+        return {"weights_sha256_12": ctx.celldino_weights_sha12, "patch_size": ctx.patch_size, **ctx.source_tag}, ()
+    if kind == "morphem":
+        return {"model_name": ctx.morphem_model_name, "patch_size": ctx.patch_size, **ctx.source_tag}, ()
+    raise ValueError(f"Unknown artifact kind: {kind!r}")
+
+
+def _manifest_entry(ctx: _CacheContext, keys: list[str]) -> Any:
+    """Return the manifest value at *keys* under ``artifacts``, or ``None`` when absent."""
+    entry: Any = ctx.manifest.get("artifacts", {})
+    for key in keys:
+        entry = entry.get(key)
+        if entry is None:
+            return None
+    return entry
 
 
 def _check_prediction_sources(ctx: _CacheContext) -> None:
@@ -664,24 +655,27 @@ def _check_prediction_sources(ctx: _CacheContext) -> None:
     audit (2026-09-30; 1304 grouped store/cache pairs, dated per chunk) found every
     legacy cache fresh except the 31 already being recomputed.
 
-    Checks the families :func:`_artifact_checks` lists for this run. Entries with no
+    Checks the families :func:`_artifact_entries` lists for this run. Entries with no
     identity yet (absent, or bookkeeping only) are left to the param check.
     """
     if not ctx.enabled or ctx.prediction_sources is None:
         return
-    for _, artifact_label, entry, _, _ in _artifact_checks(ctx):
+    for _, artifact_label, keys in _artifact_entries(ctx):
+        entry = _manifest_entry(ctx, keys)
         if not isinstance(entry, dict) or not (entry.keys() - _BOOKKEEPING_KEYS):
             continue
         if "sources" not in entry:
             built_at = entry.get("built_at")
-            horizon_ns = None if built_at is None else round(datetime.fromisoformat(str(built_at)).timestamp() * 1e9)
-            entry["sources"] = {
-                pos: dict(source)
-                for pos, source in ctx.prediction_sources.items()
-                if source["written_ns"] is not None and horizon_ns is not None and source["written_ns"] <= horizon_ns
-            }
+            entry["sources"] = {}
+            if built_at is not None:
+                horizon_ns = round(datetime.fromisoformat(str(built_at)).timestamp() * 1e9)
+                entry["sources"] = {
+                    pos: dict(source)
+                    for pos, source in ctx.prediction_sources.items()
+                    if source_predates(source, horizon_ns)
+                }
             ctx.mark_manifest_dirty()
-        recorded = entry["sources"] if isinstance(entry["sources"], dict) else {}
+        recorded = entry["sources"]
         stale = [pos for pos in entry.get("positions", []) if recorded.get(pos) != ctx.prediction_sources.get(pos)]
         if stale:
             warnings.warn(

From 488b3c12db8e3ad08d86c675bb18e5d69dcc209e Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:41:14 -0700
Subject: [PATCH 06/15] refactor(dynacell/eval): check the prediction stamp in
 the sidecar reader

metrics_provenance_matches already reads metrics_provenance.json; the
final-metrics gate then read it a second time, without the is_file()
guard, to compare the prediction digest. It now takes the current
prediction_sources and does both checks from one read, with the same
semantics: the stamped digest must match, and an unstamped sidecar is
judged by date through source_predates against its mtime.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../src/dynacell/evaluation/pipeline.py       | 45 ++----------
 .../src/dynacell/evaluation/provenance.py     | 34 +++++++--
 .../dynacell/evaluation/provenance_test.py    | 70 ++++++++++++-------
 3 files changed, 77 insertions(+), 72 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline.py b/applications/dynacell/src/dynacell/evaluation/pipeline.py
index 1db0320eb..6f13004ac 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline.py
@@ -17,7 +17,7 @@
 from tqdm import tqdm
 
 from dynacell.evaluation._ref_hook import apply_dataset_ref
-from dynacell.evaluation.cache import FeatureKind, prediction_sources, prediction_sources_sha256_12, source_predates
+from dynacell.evaluation.cache import FeatureKind, prediction_sources, prediction_sources_sha256_12
 from dynacell.evaluation.cp_reference import (
     CP_SIDECAR_FILENAME,
     DatasetCPSpace,
@@ -2007,9 +2007,9 @@ def _final_metrics_cache_valid(config: DictConfig) -> bool:
         with open_ome_zarr(Path(config.io.gt_path), mode="r") as gt_plate:
             space.check_positions([name for name, _ in gt_plate.positions()])
         current_sha256 = space.binding_sha256
-    if not metrics_provenance_matches(save_dir, cp_space_sha256=current_sha256):
-        return False
-    if not _metrics_scored_current_prediction(config, save_dir):
+    # A missing prediction store raises FileNotFoundError here, as it would when scoring.
+    sources = prediction_sources(config.io.pred_path, config.io.pred_channel_name)
+    if not metrics_provenance_matches(save_dir, cp_space_sha256=current_sha256, prediction_sources=sources):
         return False
     pixel_ok = (save_dir / config.save.pixel_metrics_filename).exists()
     mask_path = save_dir / config.save.mask_metrics_filename
@@ -2073,43 +2073,6 @@ def _final_metrics_cache_valid(config: DictConfig) -> bool:
     return pixel_ok and mask_ok and feature_ok
 
 
-def _metrics_scored_current_prediction(config: DictConfig, save_dir: Path) -> bool:
-    """Return whether ``save_dir``'s metrics were scored on the prediction now at ``io.pred_path``.
-
-    A re-predict writes into the same path, so compare the
-    ``prediction_sources_sha256_12`` that :func:`save_metrics` stamped with the store's
-    current one (:func:`~dynacell.evaluation.cache.prediction_sources`). A stamp written
-    before that field existed is dated instead: the rows are reusable only when every
-    position's chunks are dated and none was written after the stamp file; a blank
-    position (no stored chunk) cannot be dated and forces one recompute. A missing
-    prediction store raises ``FileNotFoundError`` here, as it would when scoring.
-    That misses one legacy case:
-    metrics saved after a re-predict but scored from pred caches it had left stale. The
-    grouped leaves' ``force_recompute.final_metrics: true`` and the 2026-09-30 recompute
-    of the stale caches cover it.
-
-    Parameters
-    ----------
-    config : DictConfig
-        Eval config (``io.pred_path``, ``io.pred_channel_name``).
-    save_dir : pathlib.Path
-        Directory holding ``metrics_provenance.json``.
-
-    Returns
-    -------
-    bool
-        True when the stamped fingerprint matches, or, for an unstamped sidecar, when
-        the store is not newer than it.
-    """
-    stamp = save_dir / PROVENANCE_FILENAME
-    sources = prediction_sources(config.io.pred_path, config.io.pred_channel_name)
-    recorded = json.loads(stamp.read_text()).get("prediction_sources_sha256_12")
-    if recorded is not None:
-        return recorded == prediction_sources_sha256_12(sources)
-    saved_ns = stamp.stat().st_mtime_ns
-    return all(source_predates(source, saved_ns) for source in sources.values())
-
-
 def _load_cached_final_metrics(config: DictConfig) -> tuple[list, list, list]:
     """Reload the (pixel, mask, feature) NPY caches saved by ``save_metrics``."""
     save_dir = Path(config.save.save_dir)
diff --git a/applications/dynacell/src/dynacell/evaluation/provenance.py b/applications/dynacell/src/dynacell/evaluation/provenance.py
index 2dff0cf4e..21e3cec02 100644
--- a/applications/dynacell/src/dynacell/evaluation/provenance.py
+++ b/applications/dynacell/src/dynacell/evaluation/provenance.py
@@ -49,6 +49,9 @@
 import json
 from importlib.metadata import version
 from pathlib import Path
+from typing import Any
+
+from dynacell.evaluation.cache import prediction_sources_sha256_12, source_predates
 
 #: The ``cubic`` version this repo is built against. Must equal the pin in
 #: ``applications/dynacell/pyproject.toml``; ``provenance_test.py`` asserts
@@ -158,8 +161,10 @@ def write_metrics_provenance(
     (save_dir / PROVENANCE_FILENAME).write_text(json.dumps(payload, indent=1, sort_keys=True) + "\n")
 
 
-def metrics_provenance_matches(save_dir: Path, *, cp_space_sha256: str | None) -> bool:
-    """Return True when ``save_dir``'s metrics were built by the running ``cubic`` and CP reference.
+def metrics_provenance_matches(
+    save_dir: Path, *, cp_space_sha256: str | None, prediction_sources: dict[str, dict[str, Any]]
+) -> bool:
+    """Return True when ``save_dir``'s metrics were built by the running ``cubic`` and CP reference, from this prediction.
 
     A missing sidecar returns False. Every cache written before this stamp
     existed predates the fix and is exactly the ambiguous case that has to be
@@ -177,14 +182,25 @@ def metrics_provenance_matches(save_dir: Path, *, cp_space_sha256: str | None) -
         reused and the recorded hash (or its absence) is irrelevant. When given, a
         sidecar without a ``cp_space_sha256`` (written before the binding existed)
         never matches.
+    prediction_sources : dict
+        :func:`~dynacell.evaluation.cache.prediction_sources` of the store the current
+        run would score. A re-predict writes into the same path, so the sidecar's
+        ``prediction_sources_sha256_12`` must equal their digest. A sidecar written
+        before that field existed is dated instead: reusable only when every position
+        predates it (:func:`~dynacell.evaluation.cache.source_predates` against its
+        mtime); a blank position (no stored chunk) cannot be dated and forces one
+        recompute. That misses one legacy case: metrics saved after a re-predict but
+        scored from pred caches it had left stale. The grouped leaves'
+        ``force_recompute.final_metrics: true`` and the 2026-09-30 recompute of the
+        stale caches cover it.
 
     Returns
     -------
     bool
         True when the recorded ``cubic`` version equals the installed one (or is
         in :data:`CUBIC_VERSIONS_EQUIVALENT_TO` for the declared pin while the installed one is
-        the declared pin) and, if ``cp_space_sha256`` is given, the recorded binding
-        equals it.
+        the declared pin), if ``cp_space_sha256`` is given the recorded binding
+        equals it, and the prediction check above passes.
     """
     path = save_dir / PROVENANCE_FILENAME
     if not path.is_file():
@@ -196,6 +212,10 @@ def metrics_provenance_matches(save_dir: Path, *, cp_space_sha256: str | None) -
     reproduced = installed == REQUIRED_CUBIC_VERSION and recorded in equivalents
     if recorded != installed and not reproduced:
         return False
-    if cp_space_sha256 is None:
-        return True
-    return "cp_space_sha256" in payload and payload["cp_space_sha256"] == cp_space_sha256
+    if cp_space_sha256 is not None and payload.get("cp_space_sha256") != cp_space_sha256:
+        return False
+    digest = payload.get("prediction_sources_sha256_12")
+    if digest is not None:
+        return digest == prediction_sources_sha256_12(prediction_sources)
+    saved_ns = path.stat().st_mtime_ns
+    return all(source_predates(source, saved_ns) for source in prediction_sources.values())
diff --git a/applications/dynacell/src/dynacell/evaluation/provenance_test.py b/applications/dynacell/src/dynacell/evaluation/provenance_test.py
index 86cbce32d..621a33c92 100644
--- a/applications/dynacell/src/dynacell/evaluation/provenance_test.py
+++ b/applications/dynacell/src/dynacell/evaluation/provenance_test.py
@@ -7,6 +7,7 @@
 
 import pytest
 
+from dynacell.evaluation.cache import prediction_sources_sha256_12
 from dynacell.evaluation.provenance import (
     CUBIC_VERSIONS_EQUIVALENT_TO,
     PROVENANCE_FILENAME,
@@ -18,6 +19,10 @@
 )
 
 _PYPROJECT = Path(__file__).resolve().parents[3] / "pyproject.toml"
+# One dated position: a sidecar stamped with _DIGEST matches it, one without the
+# digest is judged by date, and any sidecar written now postdates written_ns=1.
+_SOURCES = {"A/1/0": {"marker": None, "written_ns": 1}}
+_DIGEST = prediction_sources_sha256_12(_SOURCES)
 
 
 def test_required_version_matches_every_pyproject_pin():
@@ -40,47 +45,47 @@ def test_installed_versions_records_cubic():
 
 def test_roundtrip_matches_running_environment(tmp_path):
     write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12="pred-abc"
+        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12=_DIGEST
     )
     payload = json.loads((tmp_path / PROVENANCE_FILENAME).read_text())
     assert payload["versions"]["cubic"] == version("cubic")
     assert payload["cp_reference_sha256"] == "ref-abc"
     assert payload["cp_space_sha256"] == "abc"
-    assert payload["prediction_sources_sha256_12"] == "pred-abc"
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
+    assert payload["prediction_sources_sha256_12"] == _DIGEST
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
 
 def test_feature_less_stamp_matches_only_a_feature_less_run(tmp_path):
     """A run without feature metrics stamps ``None``; a CP-scoring run must not reuse it."""
     write_metrics_provenance(
-        tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12="pred-abc"
+        tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
     )
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
 
 def test_feature_less_run_ignores_the_cp_key(tmp_path):
     """compute_feature_metrics=false reuses no CP value, so a missing or foreign hash is irrelevant."""
     (tmp_path / PROVENANCE_FILENAME).write_text(json.dumps({"versions": {"cubic": version("cubic")}}))
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
     write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12="pred-abc"
+        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12=_DIGEST
     )
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_other_cp_reference_is_not_a_match(tmp_path):
     """CP values scored in another reference are not reusable after a rebuild."""
     write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-old", cp_space_sha256="old", prediction_sources_sha256_12="pred-abc"
+        tmp_path, cp_reference_sha256="ref-old", cp_space_sha256="old", prediction_sources_sha256_12=_DIGEST
     )
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="new")
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="new", prediction_sources=_SOURCES)
 
 
 def test_stamp_predating_cp_reference_is_not_a_match(tmp_path):
     """A sidecar written before the CP reference existed carries no hash; a CP-scoring run refuses it."""
     (tmp_path / PROVENANCE_FILENAME).write_text(json.dumps({"versions": {"cubic": version("cubic")}}))
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
 
 def test_missing_sidecar_is_not_a_match(tmp_path):
@@ -89,7 +94,7 @@ def test_missing_sidecar_is_not_a_match(tmp_path):
     Every cache written before the stamp existed is exactly the ambiguous
     0.8.0a2-or-0.9.0a1 case that has to be recomputed.
     """
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def _stamp_cubic(save_dir, cubic, *, cp_space_sha256):
@@ -99,7 +104,7 @@ def _stamp_cubic(save_dir, cubic, *, cp_space_sha256):
         save_dir,
         cp_reference_sha256=cp_reference_sha256,
         cp_space_sha256=cp_space_sha256,
-        prediction_sources_sha256_12="pred-abc",
+        prediction_sources_sha256_12=_DIGEST,
     )
     path = save_dir / PROVENANCE_FILENAME
     payload = json.loads(path.read_text())
@@ -109,7 +114,7 @@ def _stamp_cubic(save_dir, cubic, *, cp_space_sha256):
 
 def test_foreign_cubic_version_is_not_a_match(tmp_path):
     _stamp_cubic(tmp_path, "0.8.0a2", cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 @pytest.fixture
@@ -127,29 +132,29 @@ def test_equivalent_versions_exclude_the_declared_one():
 def test_equivalent_cubic_version_is_a_match(tmp_path, running_the_declared_pin):
     """A 0.9.0a1 cache holds the values the declared pin reproduces, so it is reused."""
     _stamp_cubic(tmp_path, "0.9.0a1", cp_space_sha256=None)
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_older_cubic_version_is_still_not_a_match(tmp_path, running_the_declared_pin):
     """0.8.0a2 moved FSC/FRC/Spectral_PCC, so its caches stay refused."""
     _stamp_cubic(tmp_path, "0.8.0a2", cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_equivalent_cubic_version_keeps_the_cp_rules(tmp_path, running_the_declared_pin):
     """Accepting the cubic stamp does not relax the CP-space binding check."""
     _stamp_cubic(tmp_path, "0.9.0a1", cp_space_sha256="abc")
-    assert metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="other")
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="other", prediction_sources=_SOURCES)
     _stamp_cubic(tmp_path, "0.9.0a1", cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
 
 def test_equivalence_only_applies_under_the_declared_pin(tmp_path, monkeypatch):
     """An environment off the pin gets no equivalence: only its own version matches."""
     monkeypatch.setattr("dynacell.evaluation.provenance.version", lambda _name: "0.8.0a2")
     _stamp_cubic(tmp_path, "0.9.0a1", cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_unmeasured_pin_has_no_equivalents(tmp_path, monkeypatch):
@@ -157,7 +162,7 @@ def test_unmeasured_pin_has_no_equivalents(tmp_path, monkeypatch):
     monkeypatch.setattr("dynacell.evaluation.provenance.REQUIRED_CUBIC_VERSION", "0.9.0a3")
     monkeypatch.setattr("dynacell.evaluation.provenance.version", lambda _name: "0.9.0a3")
     _stamp_cubic(tmp_path, "0.9.0a1", cp_space_sha256=None)
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_check_cubic_pin_accepts_the_declared_version():
@@ -177,11 +182,28 @@ def test_stamp_without_a_cp_space_binding_is_not_a_match(tmp_path):
     """A stamp carrying only the whole-reference hash (no dataset binding) is refused by a CP-scoring run."""
     payload = {"versions": {"cubic": version("cubic")}, "cp_reference_sha256": "abc"}
     (tmp_path / PROVENANCE_FILENAME).write_text(json.dumps(payload))
-    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc")
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
 
 def test_half_a_cp_stamp_is_refused(tmp_path):
     with pytest.raises(ValueError, match="both be given or both be None"):
         write_metrics_provenance(
-            tmp_path, cp_reference_sha256="abc", cp_space_sha256=None, prediction_sources_sha256_12="pred-abc"
+            tmp_path, cp_reference_sha256="abc", cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
         )
+
+
+def test_prediction_sources_must_match_the_stamp_or_predate_the_sidecar(tmp_path):
+    """The stamped digest is compared; a sidecar without one is reusable only if every position predates it."""
+    write_metrics_provenance(
+        tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
+    )
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
+    rewritten = {"A/1/0": {"marker": None, "written_ns": 2}}
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=rewritten)
+
+    (tmp_path / PROVENANCE_FILENAME).write_text(json.dumps({"versions": {"cubic": version("cubic")}}))
+    saved_ns = (tmp_path / PROVENANCE_FILENAME).stat().st_mtime_ns
+    older, newer, blank = ({"A/1/0": {"marker": None, "written_ns": w}} for w in (saved_ns, saved_ns + 1, None))
+    assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=older)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=newer)
+    assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=blank)

From 2ff1f2c32915731d98df2590452ed657835e216d Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:45:06 -0700
Subject: [PATCH 07/15] refactor(dynacell/eval): walk manifest entries through
 one helper

The five cache writers each repeated _update_manifest_entry, then
_record_position, then mark_manifest_dirty; _record_write does the three.
_record_position absorbs _add_position's separate manifest walk and uses
_manifest_leaf, and _source_current takes the entry's keys instead of
every caller walking to the entry first, which drops its isinstance
guards on values that are always dicts. _instance_keys names the
instance-label entry once for the writer, instance_cache_hit and the
family list.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../src/dynacell/evaluation/pipeline_cache.py | 95 ++++++++-----------
 1 file changed, 40 insertions(+), 55 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index b80fd0443..dc078c9b1 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -550,8 +550,8 @@ def _artifact_entries(ctx: _CacheContext) -> list[tuple[str, str, list[str]]]:
         ("cp", f"{ctx.label_prefix}cp_features", ["cp_features"]),
     ]
     if ctx.compute_instance_ap:
-        stem = f"{ctx.target_name}__{ctx.backend}"
-        families.append(("instances", f"{ctx.label_prefix}instance_masks[{stem}]", ["instance_masks", stem]))
+        keys = _instance_keys(ctx)
+        families.append(("instances", f"{ctx.label_prefix}instance_masks[{keys[1]}]", keys))
     if ctx.dinov3_model_name is not None:
         families.append(
             (
@@ -685,16 +685,15 @@ def _check_prediction_sources(ctx: _CacheContext) -> None:
             )
 
 
-def _source_current(ctx: _CacheContext, entry: dict[str, Any], pos_name: str) -> bool:
-    """Return whether *entry*'s cached *pos_name* was built from the prediction the store holds now.
+def _source_current(ctx: _CacheContext, keys: list[str], pos_name: str) -> bool:
+    """Return whether the entry at *keys* caches *pos_name* from the prediction the store holds now.
 
     Always true on the GT side. On the prediction side a position with no recorded
     source is not current.
     """
     if ctx.prediction_sources is None:
         return True
-    sources = entry.get("sources") if isinstance(entry, dict) else None
-    recorded = sources.get(pos_name) if isinstance(sources, dict) else None
+    recorded = _manifest_leaf(ctx, keys).get("sources", {}).get(pos_name)
     return recorded is not None and recorded == _prediction_source(ctx, pos_name)
 
 
@@ -779,23 +778,30 @@ def _record_position(ctx: _CacheContext, keys: list[str], pos_name: str) -> None
     The source (prediction side only) is what :func:`_source_current` compares on the
     next read, so it is recorded for exactly the position written.
     """
-    _add_position(ctx.manifest, keys, pos_name)
+    leaf = _manifest_leaf(ctx, keys)
+    positions = leaf.setdefault("positions", [])
+    if pos_name not in positions:
+        positions.append(pos_name)
     if ctx.prediction_sources is None:
         return
-    _manifest_leaf(ctx, keys).setdefault("sources", {})[pos_name] = dict(_prediction_source(ctx, pos_name))
+    leaf.setdefault("sources", {})[pos_name] = dict(_prediction_source(ctx, pos_name))
     ctx._written_sources.setdefault(tuple(keys), set()).add(pos_name)
 
 
-def _add_position(manifest: dict, keys: list[str], pos_name: str) -> None:
-    """Append *pos_name* to an artifact entry's ``positions`` list if absent."""
-    current = manifest.get("artifacts", {})
-    for key in keys:
-        current = current.get(key, {})
-        if not isinstance(current, dict):
-            return
-    positions = current.setdefault("positions", [])
-    if pos_name not in positions:
-        positions.append(pos_name)
+def _record_write(ctx: _CacheContext, keys: list[str], entry: dict[str, Any], pos_name: str) -> None:
+    """Record this run's write of *pos_name* to the artifact at *keys*: its identity *entry*, the position, its source.
+
+    The identity follows :func:`_update_manifest_entry` (kept as recorded on an excluded
+    walk); the position and its source are recorded by :func:`_record_position`.
+    """
+    _update_manifest_entry(ctx.manifest, keys, entry, preserve_identity=ctx.excluded_walk)
+    _record_position(ctx, keys, pos_name)
+    ctx.mark_manifest_dirty()
+
+
+def _instance_keys(ctx: _CacheContext) -> list[str]:
+    """Manifest keys of this run's instance-label entry: ``instance_masks`` / ``{target}__{backend}``."""
+    return ["instance_masks", f"{ctx.target_name}__{ctx.backend}"]
 
 
 def _check_identity_bootstrap(
@@ -954,6 +960,7 @@ def _fov_masks(
 
     t_count = image_arr.shape[0]
     channel_name = _MASK_CHANNEL_BY_SIDE[ctx.side]
+    keys = ["organelle_masks", manifest_key]
 
     def _compute_masks() -> np.ndarray:
         return np.stack(
@@ -978,23 +985,13 @@ def _validate_cached_shape(arr: np.ndarray) -> np.ndarray:
             )
         return arr
 
-    def _record_write() -> None:
-        _update_manifest_entry(
-            ctx.manifest,
-            ["organelle_masks", manifest_key],
-            manifest_entry,
-            preserve_identity=ctx.excluded_walk,
-        )
-        _record_position(ctx, ["organelle_masks", manifest_key], pos_name)
-        ctx.mark_manifest_dirty()
-
     def _write_masks(masks: np.ndarray) -> None:
         write_mask(ctx.paths, ctx.target_name, pos_name, masks, channel_name=channel_name, backend=ctx.backend)
 
     def _check_bootstrap() -> None:
         _check_identity_bootstrap(
             ctx,
-            ["organelle_masks", manifest_key],
+            keys,
             artifact_label=artifact_label,
             force_key=force_key,
             writing={pos_name},
@@ -1012,11 +1009,10 @@ def _check_bootstrap() -> None:
             _check_bootstrap()
             masks = _compute_masks()
             _write_masks(masks)
-            _record_write()
+            _record_write(ctx, keys, manifest_entry, pos_name)
         return masks
 
-    # A cached prediction-side position built from another prediction is a miss.
-    current = _source_current(ctx, _manifest_leaf(ctx, ["organelle_masks", manifest_key]), pos_name)
+    current = _source_current(ctx, keys, pos_name)
 
     # Fast-path: cache hit without taking a lock.
     cached = read_mask(ctx.paths, ctx.target_name, pos_name, backend=ctx.backend) if current else None
@@ -1034,7 +1030,7 @@ def _check_bootstrap() -> None:
         _check_bootstrap()
         masks = _compute_masks()
         _write_masks(masks)
-        _record_write()
+        _record_write(ctx, keys, manifest_entry, pos_name)
     return masks
 
 
@@ -1185,8 +1181,8 @@ def _fov_instances(
     """
     t_count = ref_stack.shape[0]
     force_key = f"{ctx.side}_instances"
-    stem = f"{ctx.target_name}__{ctx.backend}"
-    manifest_keys = ["instance_masks", stem]
+    manifest_keys = _instance_keys(ctx)
+    stem = manifest_keys[1]
     artifact_label = f"{ctx.label_prefix}instance_masks[{stem}]"
     manifest_entry = {
         "path": f"instance_masks/{stem}.zarr",
@@ -1229,11 +1225,6 @@ def _check_bootstrap() -> None:
             stored=lambda: _plate_positions(ctx.paths.instance_mask_plate(ctx.target_name, ctx.backend)),
         )
 
-    def _record_write() -> None:
-        _update_manifest_entry(ctx.manifest, manifest_keys, manifest_entry, preserve_identity=ctx.excluded_walk)
-        _record_position(ctx, manifest_keys, pos_name)
-        ctx.mark_manifest_dirty()
-
     if not ctx.enabled:
         return _compute()
 
@@ -1243,10 +1234,10 @@ def _record_write() -> None:
             _check_bootstrap()
             labels = _compute()
             _write(labels)
-            _record_write()
+            _record_write(ctx, manifest_keys, manifest_entry, pos_name)
         return labels
 
-    current = _source_current(ctx, _manifest_leaf(ctx, manifest_keys), pos_name)
+    current = _source_current(ctx, manifest_keys, pos_name)
     cached = read_instance_mask(ctx.paths, ctx.target_name, pos_name, backend=ctx.backend) if current else None
     if cached is not None:
         return _validate_cached_shape(cached)
@@ -1259,7 +1250,7 @@ def _record_write() -> None:
         _check_bootstrap()
         labels = _compute()
         _write(labels)
-        _record_write()
+        _record_write(ctx, manifest_keys, manifest_entry, pos_name)
     return labels
 
 
@@ -1274,7 +1265,7 @@ def instance_cache_hit(ctx: _CacheContext, pos_name: str) -> bool:
     """
     if not ctx.enabled or ctx.force[f"{ctx.side}_instances"]:
         return False
-    if not _source_current(ctx, _manifest_leaf(ctx, ["instance_masks", f"{ctx.target_name}__{ctx.backend}"]), pos_name):
+    if not _source_current(ctx, _instance_keys(ctx), pos_name):
         return False
     return read_instance_mask(ctx.paths, ctx.target_name, pos_name, backend=ctx.backend) is not None
 
@@ -1444,7 +1435,7 @@ def _load_or_compute_feature_timepoints(
 
     force_recompute = ctx.force[force_key]
     # A cached prediction-side position built from another prediction is read as a miss.
-    reuse = not force_recompute and _source_current(ctx, _manifest_leaf(ctx, manifest_keys), pos_name)
+    reuse = not force_recompute and _source_current(ctx, manifest_keys, pos_name)
     per_t: list[np.ndarray | None] = [None] * t_count
 
     # Lockless prefetch pass. Skipped under force_recompute because we'll
@@ -1618,9 +1609,7 @@ def fov_cp_features(
             # have their names read back by cached_cp_feature_names.
             "cp_feature_names": list(names),
         }
-        _update_manifest_entry(ctx.manifest, ["cp_features"], entry, preserve_identity=ctx.excluded_walk)
-        _record_position(ctx, ["cp_features"], pos_name)
-        ctx.mark_manifest_dirty()
+        _record_write(ctx, ["cp_features"], entry, pos_name)
 
     return per_t
 
@@ -1758,9 +1747,7 @@ def _fov_deep_features(
     )
 
     if ctx.enabled and manifest_updated:
-        _update_manifest_entry(ctx.manifest, manifest_keys, entry, preserve_identity=ctx.excluded_walk)
-        _record_position(ctx, manifest_keys, pos_name)
-        ctx.mark_manifest_dirty()
+        _record_write(ctx, manifest_keys, entry, pos_name)
 
     return per_t
 
@@ -1897,7 +1884,7 @@ def pending_kinds_per_t(self, pos_name: str, t_count: int) -> dict[FeatureKind,
         for kind in self.extractors:
             manifest_keys = _deep_feature_cache_metadata(self.ctx, kind)[3]
             # A prediction-side position built from another prediction is rebuilt whole.
-            stale = not _source_current(self.ctx, _manifest_leaf(self.ctx, manifest_keys), pos_name)
+            stale = not _source_current(self.ctx, manifest_keys, pos_name)
             if self._force_snapshot[kind] or stale:
                 out[kind] = list(range(t_count))
                 continue
@@ -2015,9 +2002,7 @@ def _flush_kind(
         ):
             for t, chunk in ts_chunks:
                 write_features_to_group(group, pos_name, t, chunk)
-        _update_manifest_entry(ctx.manifest, manifest_keys, entry, preserve_identity=ctx.excluded_walk)
-        _record_position(ctx, manifest_keys, pos_name)
-        ctx.mark_manifest_dirty()
+        _record_write(ctx, manifest_keys, entry, pos_name)
     # Manifest persistence is deferred to the caller (after batcher.drain()).
     # Zarr slot writes above are durable per-slot; on resume the lockless
     # prefetch finds them whether or not the manifest is up to date.

From 0fdf319128ad3f5fb3fc2e69d151836f832830f6 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:46:02 -0700
Subject: [PATCH 08/15] refactor(dynacell/eval): require the written set when
 merging manifests

flush_manifest, the only caller, always passes the positions this
process wrote, and the source merge is only correct with them: a
defaulted empty set would let the on-disk copy win for positions this
process just rebuilt. Making the argument required keeps a future
caller from dropping it silently.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/pipeline_cache.py    | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index dc078c9b1..7d3debe30 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -1765,7 +1765,7 @@ def flush_manifest(ctx: _CacheContext) -> None:
         return
     with _pos_write_lock(ctx, "manifest", "global"):
         on_disk = load_manifest(ctx.paths)
-        merged = _merge_manifests(on_disk, ctx.manifest, written=ctx._written_sources)
+        merged = _merge_manifests(on_disk, ctx.manifest, ctx._written_sources)
         save_manifest(ctx.paths, merged)
         ctx.manifest = merged
         ctx._written_sources = {}
@@ -1798,10 +1798,7 @@ def _merge_entry(on_disk: dict[str, Any], in_memory: dict[str, Any], written: se
 
 
 def _merge_manifests(
-    on_disk: dict[str, Any],
-    in_memory: dict[str, Any],
-    *,
-    written: dict[tuple[str, ...], set[str]] | None = None,
+    on_disk: dict[str, Any], in_memory: dict[str, Any], written: dict[tuple[str, ...], set[str]]
 ) -> dict[str, Any]:
     """Merge an in-memory manifest on top of the on-disk one, unioning per-position bookkeeping.
 
@@ -1811,7 +1808,6 @@ def _merge_manifests(
     (see :func:`_merge_entry`); *written* maps each entry's manifest path to the
     positions this process wrote since its last flush.
     """
-    written = written or {}
     merged = {**on_disk, **{k: v for k, v in in_memory.items() if k != "artifacts"}}
     merged["artifacts"] = dict(on_disk.get("artifacts", {}))
     in_artifacts = in_memory.get("artifacts", {})

From 639338554002b3102759df6e1523d3ddad2f017c Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:52:32 -0700
Subject: [PATCH 09/15] perf(dynacell/eval): read the prediction sources once
 per condition

evaluate_model and the grouped driver read every position's sources to
stamp the metrics, then init_cache_context read them all again for the
pred cache. The caller's snapshot now flows through evaluate_predictions
and init_cache_contexts into the pred context, which copies it (it adds
positions read on demand); the stamp hashes the same snapshot. The
final-metrics gate and process-executor workers still read their own.

The save_metrics / write_metrics_provenance kwarg becomes
prediction_digest: prediction_sources_sha256_12 shadowed the function
of that name. The sidecar key is unchanged.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../evaluation/instance_metrics_test.py       |  2 +-
 .../src/dynacell/evaluation/model_loader.py   |  9 ++--
 .../src/dynacell/evaluation/pipeline.py       | 43 +++++++++++--------
 .../src/dynacell/evaluation/pipeline_cache.py | 20 +++++++--
 .../src/dynacell/evaluation/provenance.py     | 13 +++---
 .../dynacell/evaluation/provenance_test.py    | 26 +++--------
 .../dynacell/tests/test_evaluate_compose.py   |  2 +-
 .../test_evaluation_cp_scaler_selection.py    | 11 +++--
 .../tests/test_evaluation_pipeline.py         |  2 +-
 .../test_evaluation_pred_cache_identity.py    | 13 +++++-
 .../tests/test_evaluation_timepoint_guard.py  |  4 +-
 .../dynacell/tests/test_pipeline_cache.py     |  4 +-
 .../tools/scan_nan_microssim_caches_test.py   |  4 +-
 13 files changed, 89 insertions(+), 64 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/instance_metrics_test.py b/applications/dynacell/src/dynacell/evaluation/instance_metrics_test.py
index a069feae6..24cf49b66 100644
--- a/applications/dynacell/src/dynacell/evaluation/instance_metrics_test.py
+++ b/applications/dynacell/src/dynacell/evaluation/instance_metrics_test.py
@@ -152,7 +152,7 @@ def test_instance_ap_is_written_to_mask_metrics_csv(tmp_path):
         config,
         mask_metrics=[{"FOV": "0/0/0", "Timepoint": 0, **row}],
         cp_space=None,
-        prediction_sources_sha256_12="pred-abc",
+        prediction_digest="pred-abc",
     )
 
     assert not list(tmp_path.glob("instance_ap*.csv"))
diff --git a/applications/dynacell/src/dynacell/evaluation/model_loader.py b/applications/dynacell/src/dynacell/evaluation/model_loader.py
index 98c0abaa5..97a68850a 100644
--- a/applications/dynacell/src/dynacell/evaluation/model_loader.py
+++ b/applications/dynacell/src/dynacell/evaluation/model_loader.py
@@ -200,19 +200,22 @@ def _identity_kwargs(models: EvalModels) -> dict[str, Any]:
     }
 
 
-def init_cache_contexts(config: DictConfig, models: EvalModels) -> tuple[Any, Any]:
+def init_cache_contexts(
+    config: DictConfig, models: EvalModels, *, prediction_snapshot: dict[str, dict[str, Any]] | None = None
+) -> tuple[Any, Any]:
     """Build ``(gt_ctx, pred_ctx)`` cache contexts using ``models``'s identity tags.
 
     Replaces the four-call-site duplication in ``evaluate_predictions``
     and ``_worker_setup`` where the same six kwargs flowed into
     ``init_cache_context(..., side="gt")`` then again into
-    ``init_cache_context(..., side="pred")``.
+    ``init_cache_context(..., side="pred")``. *prediction_snapshot* goes to the
+    prediction side only (see :func:`~dynacell.evaluation.pipeline_cache.init_cache_context`).
     """
     from dynacell.evaluation.pipeline_cache import init_cache_context
 
     kwargs = _identity_kwargs(models)
     gt_ctx = init_cache_context(config, side="gt", **kwargs)
-    pred_ctx = init_cache_context(config, side="pred", **kwargs)
+    pred_ctx = init_cache_context(config, side="pred", prediction_snapshot=prediction_snapshot, **kwargs)
     return gt_ctx, pred_ctx
 
 
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline.py b/applications/dynacell/src/dynacell/evaluation/pipeline.py
index 6f13004ac..133972d4f 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline.py
@@ -1330,7 +1330,11 @@ def _worker_run_fov(
 
 
 def evaluate_predictions(
-    config: DictConfig, *, models: EvalModels | None = None, cp_space: DatasetCPSpace | None = None
+    config: DictConfig,
+    *,
+    models: EvalModels | None = None,
+    cp_space: DatasetCPSpace | None = None,
+    prediction_snapshot: dict[str, dict[str, Any]] | None = None,
 ):
     """Evaluate predictions on all test images.
 
@@ -1350,6 +1354,11 @@ def evaluate_predictions(
         The CP reference bound to this eval's dataset (:func:`eval_cp_space`).
         Required exactly when ``compute_feature_metrics=true``; the caller loads it
         so the same verified reference is stamped by :func:`save_metrics`.
+    prediction_snapshot : dict or None, optional
+        :func:`~dynacell.evaluation.cache.prediction_sources` of ``io.pred_path``, read
+        by the caller for the stamp it hashes; the prediction-side cache then uses the
+        same snapshot instead of reading the store again. ``None`` reads it here.
+        Process-executor workers read their own.
     """
     # Phase 1 runtime resolution: lock in executor + thread caps before any
     # heavy work. fov_workers may be provisional when "auto"; re-resolved in
@@ -1402,7 +1411,7 @@ def evaluate_predictions(
         celldino_feature_extractor = models.celldino
         morphem_feature_extractor = models.morphem
 
-        cache_ctx, pred_cache_ctx = init_cache_contexts(config, models)
+        cache_ctx, pred_cache_ctx = init_cache_contexts(config, models, prediction_snapshot=prediction_snapshot)
     # The CP reference masks and scales by column position: refuse a GT or pred CP
     # cache whose columns are not the reference's, by name and order.
     if cp_space is not None:
@@ -1904,7 +1913,7 @@ def save_metrics(
     feature_metrics=None,
     *,
     cp_space: DatasetCPSpace | None,
-    prediction_sources_sha256_12: str,
+    prediction_digest: str,
 ):
     """Save metric rows as CSV + NPY (and plots), then stamp ``metrics_provenance.json``.
 
@@ -1920,11 +1929,11 @@ def save_metrics(
         Its ``reference_sha256`` (audit) and ``binding_sha256`` (compared on cache
         reuse) are stamped. It is passed in rather than re-read here, so a reference
         rebuilt mid-run cannot be stamped on values it did not produce.
-    prediction_sources_sha256_12 : str
-        :func:`~dynacell.evaluation.cache.prediction_sources_sha256_12` of
-        ``io.pred_path``, taken before scoring for the same reason: a re-predict landing
-        mid-run then leaves a stamp that no longer matches the store, and the rows are
-        recomputed.
+    prediction_digest : str
+        :func:`~dynacell.evaluation.cache.prediction_sources_sha256_12` of the
+        ``io.pred_path`` snapshot scored, taken before scoring for the same reason: a
+        re-predict landing mid-run then leaves a stamp that no longer matches the
+        store, and the rows are recomputed.
 
     Raises
     ------
@@ -1962,7 +1971,7 @@ def save_metrics(
         save_dir,
         cp_reference_sha256=cp_space.reference_sha256 if cp_space is not None else None,
         cp_space_sha256=cp_space.binding_sha256 if cp_space is not None else None,
-        prediction_sources_sha256_12=prediction_sources_sha256_12,
+        prediction_digest=prediction_digest,
     )
 
 
@@ -2326,11 +2335,9 @@ def get_models() -> EvalModels:
             pixel_metrics, mask_metrics, feature_metrics = _load_cached_final_metrics(merged)
         else:
             cp_space = eval_cp_space(merged) if merged.compute_feature_metrics else None
-            scored_prediction = prediction_sources_sha256_12(
-                prediction_sources(merged.io.pred_path, merged.io.pred_channel_name)
-            )
+            snapshot = prediction_sources(merged.io.pred_path, merged.io.pred_channel_name)
             pixel_metrics, mask_metrics, feature_metrics = evaluate_predictions(
-                merged, models=get_models(), cp_space=cp_space
+                merged, models=get_models(), cp_space=cp_space, prediction_snapshot=snapshot
             )
             save_metrics(
                 merged,
@@ -2338,7 +2345,7 @@ def get_models() -> EvalModels:
                 mask_metrics=mask_metrics,
                 feature_metrics=feature_metrics,
                 cp_space=cp_space,
-                prediction_sources_sha256_12=scored_prediction,
+                prediction_digest=prediction_sources_sha256_12(snapshot),
             )
         results.append((name, (pixel_metrics, mask_metrics, feature_metrics)))
         condition_save_dirs.append(Path(merged.save.save_dir))
@@ -2382,10 +2389,10 @@ def evaluate_model(config: DictConfig):
         # stamped by save_metrics is the reference the metrics were scored with. The
         # prediction fingerprint is taken before scoring for the same reason.
         cp_space = eval_cp_space(config) if config.compute_feature_metrics else None
-        scored_prediction = prediction_sources_sha256_12(
-            prediction_sources(config.io.pred_path, config.io.pred_channel_name)
+        snapshot = prediction_sources(config.io.pred_path, config.io.pred_channel_name)
+        pixel_metrics, mask_metrics, feature_metrics = evaluate_predictions(
+            config, cp_space=cp_space, prediction_snapshot=snapshot
         )
-        pixel_metrics, mask_metrics, feature_metrics = evaluate_predictions(config, cp_space=cp_space)
         with region_timer("save_metrics_csvs", "<parent>"):
             save_metrics(
                 config,
@@ -2393,7 +2400,7 @@ def evaluate_model(config: DictConfig):
                 mask_metrics=mask_metrics,
                 feature_metrics=feature_metrics,
                 cp_space=cp_space,
-                prediction_sources_sha256_12=scored_prediction,
+                prediction_digest=prediction_sources_sha256_12(snapshot),
             )
         # Re-dump so save_metrics_csvs lands in eval_timing.csv. evaluate_predictions
         # dumps once before save_metrics runs; this second dump overwrites with the
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index 7d3debe30..062bb1534 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -197,6 +197,7 @@ def init_cache_context(
     dynaclr_preprocess_version: str | None = None,
     celldino_preprocess_version: str | None = None,
     morphem_preprocess_version: str | None = None,
+    prediction_snapshot: dict[str, dict[str, Any]] | None = None,
 ) -> _CacheContext:
     """Open and validate the *side*-specific artifact cache for the run.
 
@@ -231,6 +232,11 @@ def init_cache_context(
         recomputed with the current preprocessing. Missing values (e.g.
         the kind isn't loaded, or the cached manifest pre-dates version
         tracking) are treated as "no constraint" — no auto-invalidation.
+    prediction_snapshot
+        :func:`~dynacell.evaluation.cache.prediction_sources` of ``io.pred_path`` the
+        caller already read (and hashed for the metrics stamp); the prediction side
+        copies it instead of reading the store again. ``None`` reads it here. Ignored
+        on the GT side.
 
     Notes
     -----
@@ -386,14 +392,22 @@ def init_cache_context(
         cell_segmentation_path=cell_seg_path,
     )
 
+    # Read once, before any artifact is built: a position written by this run is
+    # recorded with its source as it stood when the run started. A caller's snapshot is
+    # copied, since positions read on demand are added to it.
+    sources = None
+    if side == "pred":
+        sources = (
+            dict(prediction_snapshot)
+            if prediction_snapshot is not None
+            else prediction_sources(plate_path, channel_name)
+        )
     ctx = _CacheContext(
         paths=paths,
         manifest=manifest,
         require_complete=require_complete_requested,
         side=side,
-        # Read once, before any artifact is built: a position written by this run is
-        # recorded with its source as it stood when the run started.
-        prediction_sources=prediction_sources(plate_path, channel_name) if side == "pred" else None,
+        prediction_sources=sources,
         **base_kwargs,
     )
     _auto_invalidate_on_artifact_param_mismatch(ctx)
diff --git a/applications/dynacell/src/dynacell/evaluation/provenance.py b/applications/dynacell/src/dynacell/evaluation/provenance.py
index 21e3cec02..ca9d8f61d 100644
--- a/applications/dynacell/src/dynacell/evaluation/provenance.py
+++ b/applications/dynacell/src/dynacell/evaluation/provenance.py
@@ -125,7 +125,7 @@ def write_metrics_provenance(
     *,
     cp_reference_sha256: str | None,
     cp_space_sha256: str | None,
-    prediction_sources_sha256_12: str,
+    prediction_digest: str,
 ) -> None:
     """Write the numeric-provenance sidecar into ``save_dir``.
 
@@ -140,10 +140,11 @@ def write_metrics_provenance(
         ``DatasetCPSpace.binding_sha256`` of the space the run scored in: the
         reference bound to one dataset and the GT cells it was fit on. This is the
         value cache reuse compares. ``None`` exactly when ``cp_reference_sha256`` is.
-    prediction_sources_sha256_12 : str
-        Digest of the per-position sources of the prediction the metrics were scored
-        on, taken before scoring; the final-metrics cache gate compares it with the
-        store's current one.
+    prediction_digest : str
+        :func:`~dynacell.evaluation.cache.prediction_sources_sha256_12` of the
+        prediction the metrics were scored on, taken before scoring and stored as
+        ``prediction_sources_sha256_12``; the final-metrics cache gate compares it with
+        the store's current one.
 
     Raises
     ------
@@ -156,7 +157,7 @@ def write_metrics_provenance(
         "versions": installed_versions(),
         "cp_reference_sha256": cp_reference_sha256,
         "cp_space_sha256": cp_space_sha256,
-        "prediction_sources_sha256_12": prediction_sources_sha256_12,
+        "prediction_sources_sha256_12": prediction_digest,
     }
     (save_dir / PROVENANCE_FILENAME).write_text(json.dumps(payload, indent=1, sort_keys=True) + "\n")
 
diff --git a/applications/dynacell/src/dynacell/evaluation/provenance_test.py b/applications/dynacell/src/dynacell/evaluation/provenance_test.py
index 621a33c92..a7b8d2feb 100644
--- a/applications/dynacell/src/dynacell/evaluation/provenance_test.py
+++ b/applications/dynacell/src/dynacell/evaluation/provenance_test.py
@@ -44,9 +44,7 @@ def test_installed_versions_records_cubic():
 
 
 def test_roundtrip_matches_running_environment(tmp_path):
-    write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12=_DIGEST
-    )
+    write_metrics_provenance(tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_digest=_DIGEST)
     payload = json.loads((tmp_path / PROVENANCE_FILENAME).read_text())
     assert payload["versions"]["cubic"] == version("cubic")
     assert payload["cp_reference_sha256"] == "ref-abc"
@@ -57,9 +55,7 @@ def test_roundtrip_matches_running_environment(tmp_path):
 
 def test_feature_less_stamp_matches_only_a_feature_less_run(tmp_path):
     """A run without feature metrics stamps ``None``; a CP-scoring run must not reuse it."""
-    write_metrics_provenance(
-        tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
-    )
+    write_metrics_provenance(tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_digest=_DIGEST)
     assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
     assert not metrics_provenance_matches(tmp_path, cp_space_sha256="abc", prediction_sources=_SOURCES)
 
@@ -68,17 +64,13 @@ def test_feature_less_run_ignores_the_cp_key(tmp_path):
     """compute_feature_metrics=false reuses no CP value, so a missing or foreign hash is irrelevant."""
     (tmp_path / PROVENANCE_FILENAME).write_text(json.dumps({"versions": {"cubic": version("cubic")}}))
     assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
-    write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_sources_sha256_12=_DIGEST
-    )
+    write_metrics_provenance(tmp_path, cp_reference_sha256="ref-abc", cp_space_sha256="abc", prediction_digest=_DIGEST)
     assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
 
 
 def test_other_cp_reference_is_not_a_match(tmp_path):
     """CP values scored in another reference are not reusable after a rebuild."""
-    write_metrics_provenance(
-        tmp_path, cp_reference_sha256="ref-old", cp_space_sha256="old", prediction_sources_sha256_12=_DIGEST
-    )
+    write_metrics_provenance(tmp_path, cp_reference_sha256="ref-old", cp_space_sha256="old", prediction_digest=_DIGEST)
     assert not metrics_provenance_matches(tmp_path, cp_space_sha256="new", prediction_sources=_SOURCES)
 
 
@@ -104,7 +96,7 @@ def _stamp_cubic(save_dir, cubic, *, cp_space_sha256):
         save_dir,
         cp_reference_sha256=cp_reference_sha256,
         cp_space_sha256=cp_space_sha256,
-        prediction_sources_sha256_12=_DIGEST,
+        prediction_digest=_DIGEST,
     )
     path = save_dir / PROVENANCE_FILENAME
     payload = json.loads(path.read_text())
@@ -187,16 +179,12 @@ def test_stamp_without_a_cp_space_binding_is_not_a_match(tmp_path):
 
 def test_half_a_cp_stamp_is_refused(tmp_path):
     with pytest.raises(ValueError, match="both be given or both be None"):
-        write_metrics_provenance(
-            tmp_path, cp_reference_sha256="abc", cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
-        )
+        write_metrics_provenance(tmp_path, cp_reference_sha256="abc", cp_space_sha256=None, prediction_digest=_DIGEST)
 
 
 def test_prediction_sources_must_match_the_stamp_or_predate_the_sidecar(tmp_path):
     """The stamped digest is compared; a sidecar without one is reusable only if every position predates it."""
-    write_metrics_provenance(
-        tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12=_DIGEST
-    )
+    write_metrics_provenance(tmp_path, cp_reference_sha256=None, cp_space_sha256=None, prediction_digest=_DIGEST)
     assert metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=_SOURCES)
     rewritten = {"A/1/0": {"marker": None, "written_ns": 2}}
     assert not metrics_provenance_matches(tmp_path, cp_space_sha256=None, prediction_sources=rewritten)
diff --git a/applications/dynacell/tests/test_evaluate_compose.py b/applications/dynacell/tests/test_evaluate_compose.py
index 4beaf6002..0bd679b4e 100644
--- a/applications/dynacell/tests/test_evaluate_compose.py
+++ b/applications/dynacell/tests/test_evaluate_compose.py
@@ -180,7 +180,7 @@ def test_evaluate_model_wires_hook(monkeypatch, tmp_path) -> None:
     """``evaluate_model`` runs ``apply_dataset_ref`` before ``evaluate_predictions``."""
     captured: list[DictConfig] = []
 
-    def _fake_evaluate_predictions(cfg: DictConfig, *, cp_space):
+    def _fake_evaluate_predictions(cfg: DictConfig, *, cp_space, prediction_snapshot):
         captured.append(cfg)
         return ([], [], [])
 
diff --git a/applications/dynacell/tests/test_evaluation_cp_scaler_selection.py b/applications/dynacell/tests/test_evaluation_cp_scaler_selection.py
index c22eb3131..4ad8976d4 100644
--- a/applications/dynacell/tests/test_evaluation_cp_scaler_selection.py
+++ b/applications/dynacell/tests/test_evaluation_cp_scaler_selection.py
@@ -80,7 +80,7 @@ def test_grouped_conditions_each_get_their_own_scaler(tmp_path: Path, monkeypatc
     seen: dict[str, object] = {}
     stamped: dict[str, str | None] = {}
 
-    def _fake_evaluate_predictions(cfg, *, models, cp_space):
+    def _fake_evaluate_predictions(cfg, *, models, cp_space, prediction_snapshot):
         seen[cfg.benchmark.dataset_ref.dataset] = cp_space
         return [], [], []
 
@@ -139,7 +139,9 @@ def test_grouped_probe_failure_on_the_cp_reference_propagates(tmp_path: Path, mo
     )
     monkeypatch.setattr(pipeline, "apply_dataset_ref", lambda cfg: None)
     monkeypatch.setattr(pipeline, "load_eval_models", lambda cfg: object())
-    monkeypatch.setattr(pipeline, "evaluate_predictions", lambda cfg, *, models, cp_space: ([], [], []))
+    monkeypatch.setattr(
+        pipeline, "evaluate_predictions", lambda cfg, *, models, cp_space, prediction_snapshot: ([], [], [])
+    )
     monkeypatch.setattr(pipeline, "save_metrics", lambda *a, **k: None)
     with pytest.raises(KeyError, match="reference_path"):
         pipeline.evaluate_predictions_grouped(config)
@@ -198,7 +200,10 @@ def test_real_dataset_names_are_not_swapped(tmp_path: Path, monkeypatch) -> None
     monkeypatch.setattr(
         pipeline,
         "evaluate_predictions",
-        lambda cfg, *, models, cp_space: seen.__setitem__(cfg.benchmark.dataset_ref.dataset, cp_space) or ([], [], []),
+        lambda cfg, *, models, cp_space, prediction_snapshot: seen.__setitem__(
+            cfg.benchmark.dataset_ref.dataset, cp_space
+        )
+        or ([], [], []),
     )
     monkeypatch.setattr(pipeline, "save_metrics", lambda *a, **k: None)
     pipeline.evaluate_predictions_grouped(grouped)
diff --git a/applications/dynacell/tests/test_evaluation_pipeline.py b/applications/dynacell/tests/test_evaluation_pipeline.py
index 25eb5c62e..a964f4ed6 100644
--- a/applications/dynacell/tests/test_evaluation_pipeline.py
+++ b/applications/dynacell/tests/test_evaluation_pipeline.py
@@ -126,7 +126,7 @@ def test_evaluate_model_reuses_cache_without_feature_metrics(
         tmp_path,
         cp_reference_sha256=None,
         cp_space_sha256=None,
-        prediction_sources_sha256_12=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
+        prediction_digest=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
     )
 
     def fail_if_recomputed(_config):
diff --git a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
index 8e93d5b48..5e07a3b7f 100644
--- a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
+++ b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
@@ -449,6 +449,15 @@ def test_a_position_added_after_init_is_recorded_and_reused(tmp_path: Path, monk
     assert _masks(_open_quietly(lambda: init_cache_context(config, side="pred")), monkeypatch, everything) == {}
 
 
+def test_pred_context_uses_the_callers_snapshot(tmp_path: Path) -> None:
+    """A snapshot the caller already read (and hashed) is used as is, not re-read, and copied."""
+    _write_prediction(tmp_path / "pred.zarr", "run-1")
+    snapshot = {name: {"marker": "caller", "written_ns": 1} for name in _POSITIONS}
+    ctx = init_cache_context(_config(tmp_path), side="pred", prediction_snapshot=snapshot)
+    assert ctx.prediction_sources == snapshot
+    assert ctx.prediction_sources is not snapshot
+
+
 def test_multichannel_repredict_invalidates_only_its_channel(tmp_path: Path, monkeypatch) -> None:
     """Rewriting channel 1 of a two-channel store misses a channel-1 cache and leaves a channel-0 cache hit."""
     pred = tmp_path / "pred.zarr"
@@ -639,7 +648,7 @@ def test_a_store_without_chunks_has_no_written_time(tmp_path: Path, version: str
 def _evaluate_and_save(pipeline, config, monkeypatch, *, during_scoring=None) -> None:
     """Run the real ``evaluate_model`` save path over stubbed scoring rows."""
 
-    def _fake_evaluate_predictions(cfg, *, cp_space):
+    def _fake_evaluate_predictions(cfg, *, cp_space, prediction_snapshot):
         if during_scoring is not None:
             during_scoring()
         return _PIXEL, _MASK, []
@@ -699,7 +708,7 @@ def test_grouped_final_metrics_stamp_the_prediction_scored(tmp_path: Path, monke
         },
     )
 
-    def _fake_evaluate_predictions(cfg, *, models, cp_space):
+    def _fake_evaluate_predictions(cfg, *, models, cp_space, prediction_snapshot):
         _repredict(pred, "run-2")
         return _PIXEL, _MASK, []
 
diff --git a/applications/dynacell/tests/test_evaluation_timepoint_guard.py b/applications/dynacell/tests/test_evaluation_timepoint_guard.py
index cacc4df8a..5a8db24b0 100644
--- a/applications/dynacell/tests/test_evaluation_timepoint_guard.py
+++ b/applications/dynacell/tests/test_evaluation_timepoint_guard.py
@@ -127,7 +127,7 @@ def _write_final_caches(pipeline, save_dir: Path, feature_row: dict, config) ->
         save_dir,
         cp_reference_sha256=space.reference_sha256,
         cp_space_sha256=space.binding_sha256,
-        prediction_sources_sha256_12=prediction_sources_sha256_12(
+        prediction_digest=prediction_sources_sha256_12(
             prediction_sources(config.io.pred_path, config.io.pred_channel_name)
         ),
     )
@@ -331,7 +331,7 @@ def test_evaluate_model_stamps_the_reference_it_scored_with(tmp_path: Path, monk
     config.force_recompute.final_metrics = True
     seen = {}
 
-    def _fake_evaluate_predictions(cfg, *, cp_space):
+    def _fake_evaluate_predictions(cfg, *, cp_space, prediction_snapshot):
         seen["cp_space"] = cp_space
         make_cp_reference(cfg, tmp_path / "cp_reference.json", seed=7)  # rebuilt mid-run
         return [{"FOV": "A/1/0", "Timepoint": 0, "PCC": 0.5}], [], []
diff --git a/applications/dynacell/tests/test_pipeline_cache.py b/applications/dynacell/tests/test_pipeline_cache.py
index 37cbc918c..10b4211ce 100644
--- a/applications/dynacell/tests/test_pipeline_cache.py
+++ b/applications/dynacell/tests/test_pipeline_cache.py
@@ -2184,7 +2184,7 @@ def test_final_metrics_cache_gate_requires_ap_columns(tmp_path: Path) -> None:
         save_dir,
         cp_reference_sha256=None,
         cp_space_sha256=None,
-        prediction_sources_sha256_12=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
+        prediction_digest=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
     )
     np.save(save_dir / "pixel_metrics.npy", np.array([dict(_DUAL_SCALING_PIXEL_ROW)], dtype=object))
     cfg = _make_config(
@@ -2230,7 +2230,7 @@ def test_final_metrics_cache_gate_requires_both_pixel_scalings(tmp_path: Path) -
         save_dir,
         cp_reference_sha256=None,
         cp_space_sha256=None,
-        prediction_sources_sha256_12=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
+        prediction_digest=prediction_sources_sha256_12(prediction_sources(pred_path, "prediction")),
     )
     cfg = _make_config(
         **{
diff --git a/applications/dynacell/tools/scan_nan_microssim_caches_test.py b/applications/dynacell/tools/scan_nan_microssim_caches_test.py
index c65b7fed3..4804199c9 100644
--- a/applications/dynacell/tools/scan_nan_microssim_caches_test.py
+++ b/applications/dynacell/tools/scan_nan_microssim_caches_test.py
@@ -19,9 +19,7 @@ def _save_dir(root, rel, microssim, cubic="0.9.0a1"):
     save_dir = root / rel
     save_dir.mkdir(parents=True)
     # The scan reads only the cubic stamp; the prediction digest is a placeholder.
-    write_metrics_provenance(
-        save_dir, cp_reference_sha256=None, cp_space_sha256=None, prediction_sources_sha256_12="000000000000"
-    )
+    write_metrics_provenance(save_dir, cp_reference_sha256=None, cp_space_sha256=None, prediction_digest="000000000000")
     sidecar = save_dir / PROVENANCE_FILENAME
     payload = json.loads(sidecar.read_text())
     payload["versions"]["cubic"] = cubic

From be033c17f55a4475032cca5120c4792ed0ea3c69 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:53:48 -0700
Subject: [PATCH 10/15] docs(dynacell/eval): correct the prediction-source
 notes

The README still dated a position by its t=0 chunk; it is the earliest
stored timepoint since blank outputs store none at t=0. The feature
prefetch comment named force_recompute, but the guard is now any
position that is rewritten anyway. Comments that restated the
_source_current call go, and the legacy-dating docstrings keep their
reasoning without the campaign's dates and counts.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/README.md       |  2 +-
 .../src/dynacell/evaluation/pipeline_cache.py        | 12 +++++-------
 .../dynacell/src/dynacell/evaluation/provenance.py   |  4 ++--
 3 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/README.md b/applications/dynacell/src/dynacell/evaluation/README.md
index 45f10fc1f..dbab4657a 100644
--- a/applications/dynacell/src/dynacell/evaluation/README.md
+++ b/applications/dynacell/src/dynacell/evaluation/README.md
@@ -219,7 +219,7 @@ Each per-artifact flag invalidates that family for its side only:
 
 Without `io.gt_cache_dir` / `io.pred_cache_dir`, only `force_recompute.{final_metrics, all}` matter.
 
-A re-predict into the same `io.pred_path` needs none of these flags. The prediction side records, per cached position, the source it was built from: a hash of the position's writer marker (`viscy_prediction_complete` for `io.pred_channel_name`: checkpoint content hash, settings hash, depth handling, source shape) and the mtime of the first chunk of that channel at `t=0`. A position whose source no longer matches the store is an ordinary cache miss, recomputed and re-recorded (a `StaleCacheError` under `io.require_complete_cache=true`), so an interrupted rebuild resumes where it stopped and a code-only re-predict (same checkpoint and settings, new chunks) is caught. Writes to other channels or to position metadata do not count. Entries written before sources existed are upgraded on the first run, per position: a position is current iff its first stored chunk is no newer than the entry's `built_at`; a position with no stored chunk (a blank prediction) cannot be dated and is rebuilt once. The upgrade rewrites the manifest, so the cache directory must be writable even under `io.require_complete_cache=true`. `metrics_provenance.json` records a digest of all positions' sources, taken before scoring, and `_final_metrics_cache_valid` rejects saved metrics when it differs (an older sidecar: when any chunk is newer than the sidecar, or a position has no stored chunk). Metrics saved after a re-predict but scored from caches it had left stale are the one legacy case this cannot see. A missing prediction store raises `FileNotFoundError` wherever its sources are read, the final-metrics gate included. The GT side is not tracked.
+A re-predict into the same `io.pred_path` needs none of these flags. The prediction side records, per cached position, the source it was built from: a hash of the position's writer marker (`viscy_prediction_complete` for `io.pred_channel_name`: checkpoint content hash, settings hash, depth handling, source shape) and the mtime of that channel's first stored chunk, at its earliest stored timepoint. A position whose source no longer matches the store is an ordinary cache miss, recomputed and re-recorded (a `StaleCacheError` under `io.require_complete_cache=true`), so an interrupted rebuild resumes where it stopped and a code-only re-predict (same checkpoint and settings, new chunks) is caught. Writes to other channels or to position metadata do not count. Entries written before sources existed are upgraded on the first run, per position: a position is current iff its first stored chunk is no newer than the entry's `built_at`; a position with no stored chunk (a blank prediction) cannot be dated and is rebuilt once. The upgrade rewrites the manifest, so the cache directory must be writable even under `io.require_complete_cache=true`. `metrics_provenance.json` records a digest of all positions' sources, taken before scoring, and `_final_metrics_cache_valid` rejects saved metrics when it differs (an older sidecar: when any chunk is newer than the sidecar, or a position has no stored chunk). Metrics saved after a re-predict but scored from caches it had left stale are the one legacy case this cannot see. A missing prediction store raises `FileNotFoundError` wherever its sources are read, the final-metrics gate included. The GT side is not tracked.
 
 ### Invalidation
 
diff --git a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
index 062bb1534..b47eb6682 100644
--- a/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/pipeline_cache.py
@@ -665,9 +665,9 @@ def _check_prediction_sources(ctx: _CacheContext) -> None:
     ``built_at`` is the time of the family's LAST write, so upgrading after a partial
     write of this run would move it past the store's date. The same imprecision remains
     inside the legacy entry -- a position rebuilt after a re-predict moves ``built_at``
-    past positions still holding the old prediction -- and is accepted: a production
-    audit (2026-09-30; 1304 grouped store/cache pairs, dated per chunk) found every
-    legacy cache fresh except the 31 already being recomputed.
+    past positions still holding the old prediction -- so legacy dating is
+    approximate. It is accepted because legacy caches were audited against their
+    stores' chunks and the stale ones recomputed.
 
     Checks the families :func:`_artifact_entries` lists for this run. Entries with no
     identity yet (absent, or bookkeeping only) are left to the param check.
@@ -1448,12 +1448,11 @@ def _load_or_compute_feature_timepoints(
         return [np.asarray(compute_fn(t)) for t in range(t_count)], False
 
     force_recompute = ctx.force[force_key]
-    # A cached prediction-side position built from another prediction is read as a miss.
     reuse = not force_recompute and _source_current(ctx, manifest_keys, pos_name)
     per_t: list[np.ndarray | None] = [None] * t_count
 
-    # Lockless prefetch pass. Skipped under force_recompute because we'll
-    # rewrite every timepoint anyway.
+    # Lockless prefetch pass. Skipped when the position is rewritten anyway: forced,
+    # or cached from another prediction.
     if reuse:
         with open_features_group(ctx.paths, kind, mode="r", **cache_kwargs) as group:
             if group is not None:
@@ -1893,7 +1892,6 @@ def pending_kinds_per_t(self, pos_name: str, t_count: int) -> dict[FeatureKind,
         out: dict[FeatureKind, list[int]] = {}
         for kind in self.extractors:
             manifest_keys = _deep_feature_cache_metadata(self.ctx, kind)[3]
-            # A prediction-side position built from another prediction is rebuilt whole.
             stale = not _source_current(self.ctx, manifest_keys, pos_name)
             if self._force_snapshot[kind] or stale:
                 out[kind] = list(range(t_count))
diff --git a/applications/dynacell/src/dynacell/evaluation/provenance.py b/applications/dynacell/src/dynacell/evaluation/provenance.py
index ca9d8f61d..8fec02d1e 100644
--- a/applications/dynacell/src/dynacell/evaluation/provenance.py
+++ b/applications/dynacell/src/dynacell/evaluation/provenance.py
@@ -192,8 +192,8 @@ def metrics_provenance_matches(
         mtime); a blank position (no stored chunk) cannot be dated and forces one
         recompute. That misses one legacy case: metrics saved after a re-predict but
         scored from pred caches it had left stale. The grouped leaves'
-        ``force_recompute.final_metrics: true`` and the 2026-09-30 recompute of the
-        stale caches cover it.
+        ``force_recompute.final_metrics: true`` covers it, and stale legacy caches were
+        audited and recomputed.
 
     Returns
     -------

From 5e75398be618bef1a1f0e5b64901cec33fb59a78 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 12:54:56 -0700
Subject: [PATCH 11/15] refactor(dynacell/eval): share the store-file
 predicates in cache tests

Four places in the prediction-identity tests re-derived "zarr metadata
vs chunk file" by hand; _is_metadata and _chunks now define it once. The
pipeline cache-reuse test builds its prediction store with
make_hcs_plate like the other eval tests, dropping its own iohub setup.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../tests/test_evaluation_pipeline.py         |  6 ++---
 .../test_evaluation_pred_cache_identity.py    | 24 +++++++++----------
 2 files changed, 15 insertions(+), 15 deletions(-)

diff --git a/applications/dynacell/tests/test_evaluation_pipeline.py b/applications/dynacell/tests/test_evaluation_pipeline.py
index a964f4ed6..fb72ccb02 100644
--- a/applications/dynacell/tests/test_evaluation_pipeline.py
+++ b/applications/dynacell/tests/test_evaluation_pipeline.py
@@ -6,12 +6,13 @@
 from pathlib import Path
 
 import numpy as np
-from iohub.ngff import open_ome_zarr
 from omegaconf import OmegaConf
 
 from dynacell.evaluation.cache import prediction_sources, prediction_sources_sha256_12
 from dynacell.evaluation.provenance import write_metrics_provenance
 
+from ._eval_fixtures import make_hcs_plate
+
 
 def _write_metrics(path: Path, payload: list[dict[str, str]]) -> None:
     """Write an object-array metrics cache file."""
@@ -91,8 +92,7 @@ def test_evaluate_model_reuses_cache_without_feature_metrics(
     """Reuse pixel and mask caches when feature metrics are disabled."""
     pipeline = _import_pipeline_with_stubs(monkeypatch)
     pred_path = tmp_path / "pred.zarr"
-    with open_ome_zarr(pred_path, mode="w", layout="hcs", channel_names=["prediction"], version="0.5") as plate:
-        plate.create_position("A", "1", "0").create_image("0", np.zeros((1, 1, 1, 2, 2), dtype=np.float32))
+    make_hcs_plate(pred_path, "prediction", seed=0)
     config = OmegaConf.create(
         {
             "compute_feature_metrics": False,
diff --git a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
index 5e07a3b7f..f2ac51604 100644
--- a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
+++ b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
@@ -100,10 +100,17 @@ def _write_prediction(
     _date_store(path, attributes=_WRITTEN_AT, chunks=_WRITTEN_AT)
 
 
+def _is_metadata(f: Path) -> bool:
+    """Whether *f* is zarr metadata (v3 ``zarr.json``, v2 ``.zattrs``/``.zgroup``/``.zarray``), not a chunk."""
+    return f.name == "zarr.json" or f.name.startswith(".")
+
+
+def _chunks(path: Path) -> list[Path]:
+    return sorted(f for f in path.rglob("*") if f.is_file() and not _is_metadata(f))
+
+
 def _metadata_mtimes(path: Path) -> dict[Path, float]:
-    return {
-        f: f.stat().st_mtime for f in path.rglob("*") if f.is_file() and (f.name == "zarr.json" or f.name[0] == ".")
-    }
+    return {f: f.stat().st_mtime for f in path.rglob("*") if f.is_file() and _is_metadata(f)}
 
 
 def _repredict(
@@ -140,8 +147,7 @@ def _date_store(path: Path, *, attributes: float, chunks: float) -> None:
         for file in position.rglob("*"):
             if not file.is_file():
                 continue
-            is_metadata = file.name == "zarr.json" or file.name.startswith(".")
-            mtime = attributes if is_metadata else chunks
+            mtime = attributes if _is_metadata(file) else chunks
             os.utime(file, (mtime, mtime))
 
 
@@ -389,10 +395,6 @@ def _blank(path: Path, timepoints: slice) -> None:
             position["0"][timepoints] = 0.0
 
 
-def _chunks(path: Path) -> list[Path]:
-    return sorted(f for f in path.rglob("*") if f.is_file() and f.name != "zarr.json" and not f.name.startswith("."))
-
-
 @pytest.mark.parametrize(
     ("chunks_at", "rebuilt"),
     [(_BUILT_AT_S - _DAY, set()), (_BUILT_AT_S + _DAY, set(_POSITIONS))],
@@ -617,9 +619,7 @@ def test_source_chunk_is_found_in_every_layout(tmp_path: Path, layout: str) -> N
     with open_ome_zarr(pred, mode="w", layout="hcs", channel_names=[_CHANNEL, _OTHER], version=version) as plate:
         for name in _POSITIONS:
             plate.create_position(*name.split("/")).create_image("0", data, chunks=(1, 1, 1, _H, _W), **extra)
-    chunks = sorted(
-        f for f in (pred / "A/1/0/0").rglob("*") if f.is_file() and f.name != "zarr.json" and f.name[0] != "."
-    )
+    chunks = _chunks(pred / "A/1/0/0")
     for i, chunk in enumerate(chunks):  # every stored chunk gets its own date
         os.utime(chunk, (_WRITTEN_AT + i, _WRITTEN_AT + i))
 

From 22df884b1e45b479002d7a2d260dd2867598b7ab Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 13:32:35 -0700
Subject: [PATCH 12/15] fix(dynacell/eval): write cache manifests exactly as
 OmegaConf did

The plain libyaml SafeDumper differs from OmegaConf.save in three
ways that matter here: it leaves a str such as the hex sha12 marker
4e1234567890 unquoted, which OmegaConf's loader (and any tool that
reads a manifest with it) parses as a float; it emits &id001 aliases
when two entries share one object, as ctx.spacing does; and it
escapes non-ASCII text. A dumper subclass now ignores aliases and
single-quotes any str that parses as a bool, int or float, as
OmegaConf's own representer does, and allow_unicode is set. A new
test pins the output byte-for-byte to OmegaConf.save on those cases.

Found in the internal review of this branch; all 1224 production
manifests were already byte-identical, since none holds such a str yet.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py | 33 +++++++++++++++++--
 .../dynacell/tests/test_evaluation_cache.py   | 33 +++++++++++++++++++
 2 files changed, 64 insertions(+), 2 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index f969d42aa..a31382cdf 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -39,7 +39,36 @@
 # the pure-Python ones. They are absent only from a PyYAML built without libyaml,
 # which then falls back to the pure-Python classes.
 _YAML_LOADER = getattr(yaml, "CSafeLoader", yaml.SafeLoader)
-_YAML_DUMPER = getattr(yaml, "CSafeDumper", yaml.SafeDumper)
+
+#: YAML 1.1 bool spellings that ``OmegaConf.save`` quotes.
+_OMEGACONF_BOOLS = frozenset(
+    "y Y yes Yes YES n N no No NO true True TRUE false False FALSE on On ON off Off OFF".split()
+)
+
+
+class _ManifestDumper(getattr(yaml, "CSafeDumper", yaml.SafeDumper)):
+    """libyaml dumper that writes what ``OmegaConf.save`` wrote: no anchors, str quoted like OmegaConf.
+
+    OmegaConf's loader reads an unquoted ``4e1234567890`` (a possible hex sha12 marker)
+    as a float, so a str that parses as a bool, int or float is single-quoted, exactly
+    as OmegaConf's own representer does. Entries sharing one object (``ctx.spacing``)
+    are written out in full rather than as ``&id001`` aliases.
+    """
+
+    def ignore_aliases(self, data: Any) -> bool:
+        return True
+
+
+def _represent_str(dumper: yaml.SafeDumper, data: str) -> yaml.ScalarNode:
+    try:
+        float(data)
+        quoted = True
+    except ValueError:
+        quoted = data in _OMEGACONF_BOOLS
+    return dumper.represent_scalar("tag:yaml.org,2002:str", data, style="'" if quoted else None)
+
+
+_ManifestDumper.add_representer(str, _represent_str)
 
 CACHE_SCHEMA_VERSION = 1
 
@@ -139,7 +168,7 @@ def save_manifest(paths: CachePaths, manifest: dict[str, Any]) -> None:
     """Persist *manifest* as YAML under *paths.manifest*, creating parents."""
     paths.root.mkdir(parents=True, exist_ok=True)
     with open(paths.manifest, "w") as f:
-        yaml.dump(manifest, f, Dumper=_YAML_DUMPER, sort_keys=False)
+        yaml.dump(manifest, f, Dumper=_ManifestDumper, sort_keys=False, allow_unicode=True)
 
 
 def check_cache_identity(
diff --git a/applications/dynacell/tests/test_evaluation_cache.py b/applications/dynacell/tests/test_evaluation_cache.py
index 8a0b4bef4..ec5c30974 100644
--- a/applications/dynacell/tests/test_evaluation_cache.py
+++ b/applications/dynacell/tests/test_evaluation_cache.py
@@ -11,6 +11,8 @@
 pytest.importorskip("iohub")
 pytest.importorskip("omegaconf")
 
+from omegaconf import OmegaConf  # noqa: E402
+
 from dynacell.evaluation.cache import (  # noqa: E402
     CACHE_SCHEMA_VERSION,
     StaleCacheError,
@@ -74,6 +76,37 @@ def test_save_and_load_manifest_roundtrip(tmp_path: Path) -> None:
     assert loaded == manifest
 
 
+def test_save_manifest_writes_what_omegaconf_wrote(tmp_path: Path) -> None:
+    """The libyaml writer matches ``OmegaConf.save`` byte for byte, and OmegaConf reads its output back.
+
+    The cases are the ones libyaml's plain SafeDumper gets wrong: a hex sha12 marker that
+    OmegaConf's loader parses as a float, bool and int spellings, non-ASCII text, a
+    timestamp-like str, and one list object shared by two entries (aliases).
+    """
+    spacing = [0.29, 0.108, 0.108]
+    manifest = {
+        "cache_schema_version": CACHE_SCHEMA_VERSION,
+        "pred": {"plate_path": "/data/café/pred.zarr", "channel_name": "y"},
+        "artifacts": {
+            "cp_features": {"spacing": spacing, "built_at": "2026-09-30T19:44:54+00:00"},
+            "organelle_masks": {"er": {"spacing": spacing, "threshold": "1e-3"}},
+            "dinov3_features": {
+                "m": {
+                    "sources": {
+                        "A/1/0": {"marker": "4e1234567890", "written_ns": 1},
+                        "A/1/1": {"marker": "012345678901"},
+                    }
+                }
+            },
+        },
+    }
+    save_manifest(paths := cache_paths(tmp_path / "libyaml"), manifest)
+    OmegaConf.save(OmegaConf.create(manifest), reference := tmp_path / "omegaconf.yaml")
+    assert paths.manifest.read_bytes() == reference.read_bytes()
+    assert OmegaConf.to_container(OmegaConf.load(paths.manifest)) == manifest
+    assert load_manifest(paths) == {**manifest, "gt": None, "cell_segmentation": None}
+
+
 def test_check_cache_identity_version_mismatch(tmp_path: Path) -> None:
     """Wrong cache_schema_version raises with a clear message."""
     manifest = {"cache_schema_version": CACHE_SCHEMA_VERSION + 99, "gt": None, "cell_segmentation": None}

From 866996bea36fe3609df37e991f6e453033fab2be Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 13:33:54 -0700
Subject: [PATCH 13/15] fix(dynacell/eval): replace cache manifests atomically

save_manifest truncated the manifest in place while
init_cache_context reads it without the manifest lock, so an eval
starting during another job's flush, or after a writer died mid-dump,
read a torn file. OmegaConf turned an empty read into an empty cache;
the libyaml loader on this branch raises StaleCacheError instead.
Writing to a sibling temp file and renaming it over the manifest
leaves readers the old file or the new one. The lock files live under
.locks/, so the rename does not disturb locking, and cache dirs are
setgid with default ACLs, so the replacement keeps the group and
permissions.

Found in the internal review of this branch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py  | 18 +++++++++++++++---
 .../dynacell/tests/test_evaluation_cache.py    | 17 +++++++++++++++++
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index a31382cdf..75a3811f1 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -17,6 +17,7 @@
 import json
 import os
 import shutil
+import socket
 from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -165,10 +166,21 @@ def load_manifest(paths: CachePaths) -> dict[str, Any]:
 
 
 def save_manifest(paths: CachePaths, manifest: dict[str, Any]) -> None:
-    """Persist *manifest* as YAML under *paths.manifest*, creating parents."""
+    """Persist *manifest* as YAML under *paths.manifest*, creating parents.
+
+    The YAML goes to a sibling temp file that is then renamed over the manifest, so a
+    reader that does not hold the manifest lock (``init_cache_context``) sees the old
+    file or the new one, never a truncated one.
+    """
     paths.root.mkdir(parents=True, exist_ok=True)
-    with open(paths.manifest, "w") as f:
-        yaml.dump(manifest, f, Dumper=_ManifestDumper, sort_keys=False, allow_unicode=True)
+    tmp = paths.manifest.with_name(f".{paths.manifest.name}.{socket.gethostname()}.{os.getpid()}.tmp")
+    try:
+        with open(tmp, "w") as f:
+            yaml.dump(manifest, f, Dumper=_ManifestDumper, sort_keys=False, allow_unicode=True)
+        os.replace(tmp, paths.manifest)
+    except BaseException:
+        tmp.unlink(missing_ok=True)
+        raise
 
 
 def check_cache_identity(
diff --git a/applications/dynacell/tests/test_evaluation_cache.py b/applications/dynacell/tests/test_evaluation_cache.py
index ec5c30974..99a0f5799 100644
--- a/applications/dynacell/tests/test_evaluation_cache.py
+++ b/applications/dynacell/tests/test_evaluation_cache.py
@@ -107,6 +107,23 @@ def test_save_manifest_writes_what_omegaconf_wrote(tmp_path: Path) -> None:
     assert load_manifest(paths) == {**manifest, "gt": None, "cell_segmentation": None}
 
 
+def test_save_manifest_interrupted_leaves_the_old_file(tmp_path: Path, monkeypatch) -> None:
+    """A writer that dies mid-dump leaves the previous manifest intact and no temp file behind."""
+    paths = cache_paths(tmp_path)
+    save_manifest(paths, {"cache_schema_version": CACHE_SCHEMA_VERSION, "artifacts": {"a": 1}})
+    before = paths.manifest.read_bytes()
+
+    def dump_then_die(data, stream, **kwargs):
+        stream.write("cache_schema_version: 1\nartif")
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr("dynacell.evaluation.cache.yaml.dump", dump_then_die)
+    with pytest.raises(KeyboardInterrupt):
+        save_manifest(paths, {"cache_schema_version": CACHE_SCHEMA_VERSION, "artifacts": {"b": 2}})
+    assert paths.manifest.read_bytes() == before
+    assert sorted(p.name for p in tmp_path.iterdir()) == ["manifest.yaml"]
+
+
 def test_check_cache_identity_version_mismatch(tmp_path: Path) -> None:
     """Wrong cache_schema_version raises with a clear message."""
     manifest = {"cache_schema_version": CACHE_SCHEMA_VERSION + 99, "gt": None, "cell_segmentation": None}

From 444b34d9a2883ddc57adb27277c479d81e5fc0fa Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 13:35:05 -0700
Subject: [PATCH 14/15] test(dynacell/eval): pin that on-demand positions stay
 out of the caller's snapshot

The metrics digest is computed from the snapshot the caller read before
scoring, and the pred context adds positions that appear later to its
own copy. The old test only checked identity (is not); it now adds a
position on demand and checks the caller's dict is unchanged. Dropping
the dict() copy in init_cache_context makes it fail.

Found in the internal review of this branch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../tests/test_evaluation_pred_cache_identity.py    | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

diff --git a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
index f2ac51604..4fc1282bd 100644
--- a/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
+++ b/applications/dynacell/tests/test_evaluation_pred_cache_identity.py
@@ -451,13 +451,18 @@ def test_a_position_added_after_init_is_recorded_and_reused(tmp_path: Path, monk
     assert _masks(_open_quietly(lambda: init_cache_context(config, side="pred")), monkeypatch, everything) == {}
 
 
-def test_pred_context_uses_the_callers_snapshot(tmp_path: Path) -> None:
-    """A snapshot the caller already read (and hashed) is used as is, not re-read, and copied."""
+def test_pred_context_uses_the_callers_snapshot(tmp_path: Path, monkeypatch) -> None:
+    """A snapshot the caller already read (and hashed) is used as is, and positions read on demand stay out of it."""
     _write_prediction(tmp_path / "pred.zarr", "run-1")
-    snapshot = {name: {"marker": "caller", "written_ns": 1} for name in _POSITIONS}
+    first, late = _POSITIONS
+    snapshot = {first: {"marker": "caller", "written_ns": 1}}
     ctx = init_cache_context(_config(tmp_path), side="pred", prediction_snapshot=snapshot)
     assert ctx.prediction_sources == snapshot
-    assert ctx.prediction_sources is not snapshot
+
+    _masks(ctx, monkeypatch)
+    assert ctx.prediction_sources[first] == {"marker": "caller", "written_ns": 1}
+    assert late in ctx.prediction_sources
+    assert snapshot == {first: {"marker": "caller", "written_ns": 1}}
 
 
 def test_multichannel_repredict_invalidates_only_its_channel(tmp_path: Path, monkeypatch) -> None:

From 37e6b412ff7843f88d4d9bdd00e019f1c48cda48 Mon Sep 17 00:00:00 2001
From: Alexandr Kalinin <alxndrkalinin@users.noreply.github.com>
Date: Wed, 30 Sep 2026 13:45:08 -0700
Subject: [PATCH 15/15] fix(dynacell/eval): create the manifest temp file
 exclusively under a random name

The temp name was predictable (host + pid) and opened with "w", which
truncates and follows whatever already sits at that path. Cache dirs
are group-writable, so another process could plant a symlink there
and have the writer truncate a file it points to; a stale temp could
also collide after pid reuse. The name is now a uuid4 and the file is
opened with "x" (O_CREAT | O_EXCL), which refuses any existing path,
symlinks included. On that refusal the path is left alone rather than
unlinked. Permissions still come from umask plus the directory's
default ACL, unlike mkstemp's fixed 0600.

A test plants a symlink at the temp name and checks that save_manifest
raises, the target is untouched and the symlink remains; with "w" it
does not raise.

Addresses Copilot review on #514.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XvkM1gKifWmoDPq8Drd4JF
---
 .../dynacell/src/dynacell/evaluation/cache.py   | 12 ++++++++----
 .../dynacell/tests/test_evaluation_cache.py     | 17 +++++++++++++++++
 2 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/applications/dynacell/src/dynacell/evaluation/cache.py b/applications/dynacell/src/dynacell/evaluation/cache.py
index 75a3811f1..fb142547f 100644
--- a/applications/dynacell/src/dynacell/evaluation/cache.py
+++ b/applications/dynacell/src/dynacell/evaluation/cache.py
@@ -17,7 +17,7 @@
 import json
 import os
 import shutil
-import socket
+import uuid
 from collections.abc import Iterable, Iterator
 from contextlib import contextmanager
 from dataclasses import dataclass
@@ -170,14 +170,18 @@ def save_manifest(paths: CachePaths, manifest: dict[str, Any]) -> None:
 
     The YAML goes to a sibling temp file that is then renamed over the manifest, so a
     reader that does not hold the manifest lock (``init_cache_context``) sees the old
-    file or the new one, never a truncated one.
+    file or the new one, never a truncated one. The temp name is unguessable and created
+    exclusively, so in a group-writable cache dir it never truncates or writes through a
+    path (or symlink) that another process made.
     """
     paths.root.mkdir(parents=True, exist_ok=True)
-    tmp = paths.manifest.with_name(f".{paths.manifest.name}.{socket.gethostname()}.{os.getpid()}.tmp")
+    tmp = paths.manifest.with_name(f".{paths.manifest.name}.{uuid.uuid4().hex}.tmp")
     try:
-        with open(tmp, "w") as f:
+        with open(tmp, "x") as f:
             yaml.dump(manifest, f, Dumper=_ManifestDumper, sort_keys=False, allow_unicode=True)
         os.replace(tmp, paths.manifest)
+    except FileExistsError:
+        raise  # only open(tmp, "x") raises this; the path is not ours to remove
     except BaseException:
         tmp.unlink(missing_ok=True)
         raise
diff --git a/applications/dynacell/tests/test_evaluation_cache.py b/applications/dynacell/tests/test_evaluation_cache.py
index 99a0f5799..5e488cdaa 100644
--- a/applications/dynacell/tests/test_evaluation_cache.py
+++ b/applications/dynacell/tests/test_evaluation_cache.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import uuid
 from pathlib import Path
 
 import numpy as np
@@ -124,6 +125,22 @@ def dump_then_die(data, stream, **kwargs):
     assert sorted(p.name for p in tmp_path.iterdir()) == ["manifest.yaml"]
 
 
+def test_save_manifest_never_writes_through_a_planted_temp_path(tmp_path: Path, monkeypatch) -> None:
+    """A path already at the temp name (here a symlink to another file) is refused, not followed or removed."""
+    paths = cache_paths(tmp_path / "cache")
+    paths.root.mkdir()
+    victim = tmp_path / "victim.txt"
+    victim.write_text("keep me")
+    monkeypatch.setattr("dynacell.evaluation.cache.uuid.uuid4", lambda: uuid.UUID(int=0))
+    planted = paths.root / f".manifest.yaml.{uuid.UUID(int=0).hex}.tmp"
+    planted.symlink_to(victim)
+    with pytest.raises(FileExistsError):
+        save_manifest(paths, {"cache_schema_version": CACHE_SCHEMA_VERSION, "artifacts": {}})
+    assert victim.read_text() == "keep me"
+    assert planted.is_symlink()
+    assert not paths.manifest.exists()
+
+
 def test_check_cache_identity_version_mismatch(tmp_path: Path) -> None:
     """Wrong cache_schema_version raises with a clear message."""
     manifest = {"cache_schema_version": CACHE_SCHEMA_VERSION + 99, "gt": None, "cell_segmentation": None}