diff --git a/changelog.d/graph-verify-once.changed.md b/changelog.d/graph-verify-once.changed.md
new file mode 100644
index 0000000000..d78631b6b9
--- /dev/null
+++ b/changelog.d/graph-verify-once.changed.md
@@ -0,0 +1 @@
+Derive each graph source's content key once per run instead of once per executed node, re-deriving every source in full before the manifest is built, and build the per-column context-digest stream with numpy. A run that re-derivation refuses now evicts every store object it published, so nothing it wrote survives under the source identity it started with; a run that leaves by any other exception gets the same re-derivation on its way out, keeping its work when its sources held still and evicting it otherwise. A process that stops without unwinding, after a rewrite that left every stat field identical, is not covered. No digest, key, seal or receipt value changes.
diff --git a/packages/microcosm-graph/src/microcosm/graph/codecs.py b/packages/microcosm-graph/src/microcosm/graph/codecs.py
index 3ebc9d8e93..882633e895 100644
--- a/packages/microcosm-graph/src/microcosm/graph/codecs.py
+++ b/packages/microcosm-graph/src/microcosm/graph/codecs.py
@@ -541,8 +541,10 @@ def load_raw_bytes(path: Path, *, store: ContentStore | None = None) -> bytes:
     itself opened, and the open is non-blocking, so the codec cannot be made
     to wait on a pipe or stream a device. The read is bounded rather than
     trusted to ``st_size``, because a file may grow after it is measured.
-    Detecting that a source moved is the executor's post-run content check;
-    this bound only keeps the codec from reading an unbounded amount first.
+    Detecting that a source moved is the executor's own content check -- after
+    every cold node that declares the source, and again over every source in
+    full before the run manifest is built; this bound only keeps the codec from
+    reading an unbounded amount first.
     """
 
     del store  # raw bytes are self-describing; no content store is consulted
diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py
index b61a678d4c..ff6f0e9fd5 100644
--- a/packages/microcosm-graph/src/microcosm/graph/executor.py
+++ b/packages/microcosm-graph/src/microcosm/graph/executor.py
@@ -6,10 +6,11 @@
 import json
 import pickle
 import socket
+import stat
 import time
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Iterable, Mapping
 from copy import deepcopy
-from dataclasses import fields, replace
+from dataclasses import dataclass, fields, replace
 from datetime import UTC, datetime
 from pathlib import Path
 from types import MappingProxyType
@@ -442,6 +443,78 @@ def _update_array(digest: hashlib._Hash, values: object) -> None:
         digest.update(np.ascontiguousarray(array).tobytes())
 
 
+def _framed(payloads: np.ndarray) -> bytes:
+    """Frame equal-width payload rows exactly as ``_update_scalar`` frames one."""
+
+    rows, width = payloads.shape
+    prefix = np.frombuffer(
+        np.full(rows, width, dtype="<u8").tobytes(), dtype=np.uint8
+    ).reshape(rows, 8)
+    return np.concatenate([prefix, payloads], axis=1).tobytes()
+
+
+def _object_stream(series: pd.Series) -> bytes | None:
+    """The bytes ``_update_array`` writes for this column's object projection.
+
+    ``to_numpy(dtype=object)`` boxes a plain numpy column into one Python object
+    per value, and ``_update_scalar`` then frames each box.  For the three kinds
+    that dominate a survey frame the boxed value is always the same exact type,
+    so the whole framed stream is a pure function of the column's bytes and can
+    be built at C speed.  Byte-for-byte identical output is the contract; any
+    column this cannot reproduce exactly returns None and takes the loop.
+    """
+
+    dtype = series.dtype
+    if not isinstance(dtype, np.dtype) or dtype.kind not in "fiub":
+        return None
+    values = np.ascontiguousarray(series.to_numpy(copy=False))
+    if values.ndim != 1 or values.dtype != dtype:
+        return None
+    rows = values.shape[0]
+    if rows == 0:
+        return b""
+    if dtype.kind == "b":
+        # bool -> b"b1" / b"b0"; np.bool_ boxes to bool, checked before int.
+        payloads = np.empty((rows, 2), dtype=np.uint8)
+        payloads[:, 0] = ord("b")
+        payloads[:, 1] = np.where(values, ord("1"), ord("0"))
+        return _framed(payloads)
+    if dtype.kind == "f":
+        # Every float width boxes to an exact Python float, which packs to the
+        # same eight native-order IEEE-754 bytes as the float64 cast, NaN
+        # payload, signed zero and infinities included. The cast is to
+        # `np.float64`, whose byte order is the host's, because the loop this
+        # replaces packs `np.asarray([value], dtype=np.float64).tobytes()`: an
+        # explicit `<f8` would agree with it on every little-endian host and
+        # disagree on any other. The length prefixes stay explicitly
+        # little-endian, because the loop writes them with
+        # `int.to_bytes(8, "little")`.
+        wide = np.ascontiguousarray(values, dtype=np.float64)
+        payloads = np.empty((rows, 9), dtype=np.uint8)
+        payloads[:, 0] = ord("f")
+        payloads[:, 1:] = np.frombuffer(wide.tobytes(), dtype=np.uint8).reshape(rows, 8)
+        return _framed(payloads)
+    # Signed and unsigned integers box to Python ints, whose decimal rendering
+    # numpy reproduces exactly; payload width therefore varies per value.
+    text = values.astype("S")
+    width = text.dtype.itemsize
+    raw = np.frombuffer(text.tobytes(), dtype=np.uint8).reshape(rows, width)
+    filled = raw != 0  # 'S' pads the short renderings on the right with NUL
+    # The cast is the last step, not the first: a ufunc result carries the
+    # host's byte order whatever its operands carried, so casting the sum and
+    # then adding the type byte's length would emit big-endian prefixes on a
+    # big-endian host, while the loop this reproduces writes them with
+    # `int.to_bytes(8, "little")`. Casting the finished length pins the order.
+    lengths = (filled.sum(axis=1) + 1).astype("<u8")
+    block = np.empty((rows, 9 + width), dtype=np.uint8)
+    block[:, :8] = np.frombuffer(lengths.tobytes(), dtype=np.uint8).reshape(rows, 8)
+    block[:, 8] = ord("i")
+    block[:, 9:] = raw
+    keep = np.ones((rows, 9 + width), dtype=bool)
+    keep[:, 9:] = filled
+    return block[keep].tobytes()
+
+
 def _update_series(digest: hashlib._Hash, series: pd.Series) -> None:
     digest.update(str(series.dtype).encode("utf-8"))
     digest.update(b"\0")
@@ -452,7 +525,16 @@ def _update_series(digest: hashlib._Hash, series: pd.Series) -> None:
         _update_array(digest, data)
         _update_array(digest, mask)
     else:
-        _update_array(digest, series.to_numpy(dtype=object, copy=False))
+        stream = _object_stream(series)
+        if stream is None:
+            _update_array(digest, series.to_numpy(dtype=object, copy=False))
+        else:
+            # The header the object projection would have written: its dtype is
+            # "object" and its shape is the column's own one-dimensional shape.
+            digest.update(b"object\0")
+            digest.update(f"({len(series)},)".encode("ascii"))
+            digest.update(b"\0")
+            digest.update(stream)
     _update_array(digest, series.index.to_numpy(copy=False))
 
 
@@ -1979,6 +2061,8 @@ def _source_paths_and_keys(
     compiled: CompiledGraph,
     sources: Mapping[str, Path],
     store: ContentStore,
+    *,
+    identities: _SourceIdentities | None = None,
 ) -> tuple[dict[str, Path], dict[str, str]]:
     declared = {source.name: source for source in compiled.graph.sources}
     used = {name for node in compiled.graph.nodes for name in node.sources}
@@ -1989,7 +2073,7 @@ def _source_paths_and_keys(
     if unknown:
         raise ValueError(f"Source paths supplied for undeclared names {unknown!r}.")
     resolved: dict[str, Path] = {}
-    identities: dict[str, str] = {}
+    derived: dict[str, str] = {}
     for name in sorted(used):
         path = Path(sources[name]).resolve(strict=True)
         # Codec availability is verified before any kernel can execute.  The
@@ -2007,8 +2091,115 @@ def _source_paths_and_keys(
         else:  # defended by ContentStore.__init__
             raise StoreUnavailable("ContentStore has an invalid codec registry.")
         resolved[name] = path
-        identities[name] = source_content_key(name, path)
-    return resolved, identities
+        derived[name] = (
+            source_content_key(name, path)
+            if identities is None
+            else identities.key(name, path)
+        )
+    return resolved, derived
+
+
+def _stat_identity(info: object) -> tuple[int, int, int, int, int]:
+    """The five fields every source owner in this repository binds a file by."""
+
+    return (
+        info.st_dev,
+        info.st_ino,
+        info.st_size,
+        info.st_mtime_ns,
+        info.st_ctime_ns,
+    )
+
+
+def _followed_identity(path: Path) -> tuple[object, ...]:
+    """The identity a member link resolves to, or why it resolves to nothing."""
+
+    try:
+        info = path.stat()
+    except OSError as error:
+        return ("absent", error.errno)
+    return (stat.S_IFMT(info.st_mode), _stat_identity(info))
+
+
+def _source_stat_signature(path: Path) -> tuple[object, ...]:
+    """A read-free signature of everything ``source_content_key`` would read.
+
+    A regular file contributes its own stat identity. A directory contributes
+    its own, plus the relative name, file type and stat identity of every entry
+    ``_directory_identity`` would walk -- the roster as well as the members,
+    because the directory identity hashes relative names and a file added or
+    removed moves it with no member's stat changing. Anything else contributes
+    its stat identity alone, so an unexpected node type is never cached past a
+    change.
+
+    A member that is a symlink contributes its target's identity as well.
+    ``_directory_identity`` selects members with ``is_file()`` and reads them
+    with ``read_bytes()``, and both follow the link, so the target's bytes are
+    inside the content key while the link's own five stat fields never move
+    when those bytes change. Following it here keeps the same member selection
+    on both sides, so such a change is a miss at the next node that declares
+    the source rather than a refusal deferred to run end. A link that resolves
+    to nothing contributes that fact and raises nothing, exactly as
+    ``_directory_identity`` skips it.
+    """
+
+    info = path.lstat()
+    if stat.S_ISREG(info.st_mode):
+        return ("file", _stat_identity(info))
+    if not stat.S_ISDIR(info.st_mode):
+        return ("other", stat.S_IFMT(info.st_mode), _stat_identity(info))
+    entries = []
+    for candidate in sorted(path.rglob("*")):
+        entry = candidate.lstat()
+        member = (
+            candidate.relative_to(path).as_posix(),
+            stat.S_IFMT(entry.st_mode),
+            _stat_identity(entry),
+        )
+        if stat.S_ISLNK(entry.st_mode):
+            member = (*member, _followed_identity(candidate))
+        entries.append(member)
+    return ("dir", _stat_identity(info), tuple(entries))
+
+
+class _SourceIdentities:
+    """One run's source content keys, re-derived only when a source may have moved.
+
+    ``source_content_key`` itself stays pure and uncached: it is the declared
+    path-independent identity of a source's *content*, and three key and codec
+    tests call it twice on one path with the bytes changed in between. This
+    cache is the caller's, lives in one ``run_graph`` call, and never outlives
+    it.
+
+    A cached key is reused only when the path's stat signature is identical to
+    the signature taken both immediately before and immediately after the read
+    that produced the key. A derivation whose two signatures disagree -- the
+    source moved while it was being read -- is not cached at all, so every
+    consumer re-derives it exactly as it does today.
+    """
+
+    __slots__ = ("_entries",)
+
+    def __init__(self) -> None:
+        self._entries: dict[str, tuple[tuple[object, ...], str]] = {}
+
+    def key(self, name: str, path: Path) -> str:
+        signature = _source_stat_signature(path)
+        entry = self._entries.get(name)
+        if entry is not None and entry[0] == signature:
+            return entry[1]
+        identity = source_content_key(name, path)
+        if _source_stat_signature(path) == signature:
+            self._entries[name] = (signature, identity)
+        else:
+            self._entries.pop(name, None)
+        return identity
+
+    def derive(self, name: str, path: Path) -> str:
+        """Re-derive one key with the cache bypassed, and refresh the cache."""
+
+        self._entries.pop(name, None)
+        return self.key(name, path)
 
 
 def _all_node_keys(
@@ -2239,6 +2430,105 @@ def _preflight_expand_declarations(compiled: CompiledGraph) -> None:
             ) from error
 
 
+def _evict_run_writes(store: ContentStore, written: Iterable[str]) -> list[str]:
+    """Take back every object a refusing run published, and report what stayed.
+
+    Idempotent, because :meth:`ContentStore.evict` is: a key already gone is
+    not an error and a second pass removes nothing. It never raises. The
+    refusal it runs under is the outcome of the run, so an object that will
+    not go is named in that refusal rather than replacing it with a second
+    failure, and the caller says so in the message it raises.
+    """
+
+    stranded: list[str] = []
+    for key in sorted(written):
+        try:
+            store.evict(key)
+        except Exception as error:  # every failure is reported, none replaces
+            stranded.append(f"{key} ({type(error).__name__}: {error})")
+    return stranded
+
+
+@dataclass
+class _RunSources:
+    """What a failing run needs in order to decide whether its writes stand.
+
+    ``_execute_graph`` binds the run-start paths, keys and identity cache here
+    as soon as they exist, and marks the record ``settled`` once its own
+    run-end pass has decided -- either way. ``run_graph`` reads it only when
+    the run is leaving by exception.
+    """
+
+    paths: Mapping[str, Path] | None = None
+    keys: Mapping[str, str] | None = None
+    identities: _SourceIdentities | None = None
+    settled: bool = False
+
+
+def _settle_failed_run(
+    store: ContentStore,
+    written: Iterable[str],
+    run_sources: _RunSources,
+    error: BaseException,
+) -> None:
+    """Decide, for a run leaving by exception, whether what it wrote may stay.
+
+    The run-end pass is the only check that sees a rewrite leaving every stat
+    field identical, and a run that raises never reaches it. Without this, a
+    node that persisted under the pre-rewrite identity would survive a later
+    node's failure, a store error or an interrupt, and the next
+    ``resume="auto"`` run would hit its record.
+
+    So the same cache-bypassing re-derivation runs here. Sources that still
+    match leave the store alone, which is what keeps an interrupted or failed
+    run resumable. A source that moved, a re-derivation that cannot finish --
+    a member removed mid-walk, a second interrupt -- or a failure before the
+    run had any source key to compare against evicts everything the run
+    wrote. The original error is never replaced: what happened here is added
+    to it as a note.
+    """
+
+    written = set(written)
+    if run_sources.settled or not written:
+        return
+    moved: list[str] = []
+    unverified = ""
+    if (
+        run_sources.paths is None
+        or run_sources.keys is None
+        or run_sources.identities is None
+    ):
+        unverified = "the run failed before its source identities were derived"
+    else:
+        try:
+            for name in sorted(run_sources.paths):
+                identity = run_sources.identities.derive(name, run_sources.paths[name])
+                if identity != run_sources.keys[name]:
+                    moved.append(name)
+        except BaseException as derivation_error:
+            unverified = (
+                "its sources could not be re-derived "
+                f"({type(derivation_error).__name__}: {derivation_error})"
+            )
+    if not moved and not unverified:
+        return
+    stranded = _evict_run_writes(store, written)
+    reason = (
+        "source " + ", ".join(repr(name) for name in moved) + " changed while it ran"
+        if moved
+        else unverified
+    )
+    note = f"Everything this run wrote was evicted, because {reason}."
+    if stranded:
+        note = (
+            f"Everything this run wrote was evicted, because {reason}, except "
+            + ", ".join(stranded)
+            + ", which could not be evicted and may hold work derived from a"
+            " changed source."
+        )
+    error.add_note(note)
+
+
 def run_graph(
     compiled: CompiledGraph,
     *,
@@ -2248,6 +2538,7 @@ def run_graph(
     resume: ResumePolicy = "auto",
     decisions: tuple[Decision, ...] = (),
     _population_observer: Callable[[str, Population], None] | None = None,
+    _verification_epoch: Mapping[str, object] | None = None,
 ) -> RunManifest:
     """Execute a compiled graph with content-addressed reuse and receipts.
 
@@ -2258,6 +2549,66 @@ def run_graph(
     persistence, and an exception it raises refuses the run. It is never a
     kernel capability, enters no key or receipt, and an unreached node has no
     population to observe.
+
+    The private verification-epoch record is a caller's own counts mapping --
+    a country runtime that scopes source verification around the whole run
+    hands in the record that scope yields. It is attached to the manifest
+    unchanged, as a live view rather than a copy, so counts the caller
+    finalises when its scope closes are present by the time the caller holds
+    the manifest. It enters no key, no receipt and no cache record.
+
+    The run executes inside a store write ledger. A run that refuses at the
+    run-end source re-derivation has already persisted whatever the per-node
+    check could not discriminate, and the ledger is what lets it take those
+    objects back out before the refusal reaches this caller. A run that leaves
+    by any other exception never reaches that pass, so the same re-derivation
+    runs on its way out (:func:`_settle_failed_run`): unchanged sources keep
+    the run's work for a later resume, anything else evicts it.
+
+    What no in-process check can cover is a process that stops without
+    unwinding -- SIGKILL, a power loss -- after a rewrite that left every stat
+    field identical and before either pass ran. The objects such a run
+    persisted stay in the store under the pre-rewrite identity.
+    """
+
+    run_sources = _RunSources()
+    with store.recording_writes() as written:
+        try:
+            return _execute_graph(
+                compiled,
+                sources=sources,
+                store=store,
+                kernels=kernels,
+                resume=resume,
+                decisions=decisions,
+                written=written,
+                run_sources=run_sources,
+                _population_observer=_population_observer,
+                _verification_epoch=_verification_epoch,
+            )
+        except BaseException as error:
+            _settle_failed_run(store, written, run_sources, error)
+            raise
+
+
+def _execute_graph(
+    compiled: CompiledGraph,
+    *,
+    sources: Mapping[str, Path],
+    store: ContentStore,
+    kernels: KernelRegistry,
+    resume: ResumePolicy,
+    decisions: tuple[Decision, ...],
+    written: set[str],
+    run_sources: _RunSources,
+    _population_observer: Callable[[str, Population], None] | None,
+    _verification_epoch: Mapping[str, object] | None,
+) -> RunManifest:
+    """One run, with ``written`` collecting every key it publishes.
+
+    ``run_graph`` owns that ledger's lifetime; this is where it is spent.
+    ``run_sources`` is how a run that raises tells ``run_graph`` what it
+    started from, and whether its own run-end pass already decided.
     """
 
     if resume not in ("auto", "require", "forbid"):
@@ -2274,7 +2625,16 @@ def run_graph(
 
     _preflight_expand_declarations(compiled)
     started_at = _now()
-    source_paths, source_keys = _source_paths_and_keys(compiled, sources, store)
+    # One run's source identities. The run-start pass populates the cache as it
+    # reads, so the per-node check below is a stat walk rather than a second
+    # full read; the run-end pass bypasses the cache entirely.
+    source_identities = _SourceIdentities()
+    source_paths, source_keys = _source_paths_and_keys(
+        compiled, sources, store, identities=source_identities
+    )
+    run_sources.paths = source_paths
+    run_sources.keys = source_keys
+    run_sources.identities = source_identities
     keys, implementations = _all_node_keys(compiled, kernels, source_keys)
     contracts = {
         node_id: typed_contracts(compiled, compiled.graph.node(node_id), keys, kernels)
@@ -2443,7 +2803,19 @@ def run_graph(
             if before != after:
                 raise NodeRejected(f"Node {node.id!r} mutated its input context.")
             for name in node.sources:
-                current = source_content_key(name, source_paths[name])
+                # A stat signature that still matches the one taken across this
+                # source's own read reuses that read's key; a rewrite that moves
+                # any field of that signature is re-derived in full here and
+                # refused before this node's _write_node, so nothing derived
+                # from it reaches the store. What this check cannot see is a
+                # rewrite that leaves every field identical: that is a cache
+                # hit, this node persists under the pre-rewrite identity, and
+                # the run-end re-derivation below is what catches it -- and
+                # evicts everything the run persisted. A run that raises
+                # before reaching that pass gets the same re-derivation from
+                # run_graph on its way out. Only a process that stops without
+                # unwinding leaves such a node's objects behind.
+                current = source_identities.key(name, source_paths[name])
                 if current != source_keys[name]:
                     raise NodeRejected(
                         f"Node {node.id!r} changed source {name!r} while running."
@@ -2613,9 +2985,69 @@ def run_graph(
         )
         receipt_payloads[node_id] = receipts[node_id].receipt
 
+    # Every source is re-derived in full, cache bypassed, before any caller
+    # receives a manifest. This closes the two cases a stat signature cannot
+    # decide -- a rewrite that leaves every field of that signature identical,
+    # and a change made during a node that declares no source at all, which the
+    # per-node check has never seen.
+    #
+    # The guarantee the two checks make together: a rewrite that moves any stat
+    # field refuses before the node that would have persisted work derived from
+    # it, so the store never sees that work; a rewrite that moves none of them
+    # is refused only here, after nodes have persisted artifacts and cache
+    # records under the pre-rewrite source identity, so everything this run
+    # published is evicted before the refusal propagates and no later
+    # resume="auto" run can reuse it. The eviction is idempotent and never
+    # replaces the refusal: an object that will not go is named in it.
+    #
+    # This pass covers a run that gets this far. One that raises earlier is
+    # settled by run_graph with the same re-derivation, and one whose process
+    # stops without unwinding is covered by neither: that residual is stated
+    # on run_graph.
+    #
+    # It is written inline rather than through _source_paths_and_keys because
+    # that helper answers from the per-run cache, which this pass exists to
+    # bypass, and re-resolves every path and re-checks codec availability,
+    # which this pass has no reason to repeat.
+    #
+    # Keeping it out of that helper also keeps the helper's call count where
+    # the US build shard on the branch stacked on this one expects it: a
+    # sys.setprofile hook there tracks _source_paths_and_keys by its exact
+    # code object id and times one source-key pass per call. Grep the stacked
+    # branch for "source_key" in
+    # packages/microcosm-build/tests/test_us_graph_survey_population.py.
+    # Nothing at this head references the helper outside this module.
+    final_identities = {}
+    moved = []
+    for name in sorted(source_paths):
+        identity = source_identities.derive(name, source_paths[name])
+        final_identities[name] = identity
+        if identity != source_keys[name]:
+            moved.append(name)
+    # Decided here, either way: run_graph must not re-derive a second time for
+    # the refusal below, nor for anything raised while the manifest is built.
+    run_sources.settled = True
+    if moved:
+        stranded = _evict_run_writes(store, written)
+        refusal = (
+            "Run changed source "
+            + ", ".join(repr(name) for name in moved)
+            + " while executing."
+        )
+        if stranded:
+            refusal += (
+                " Everything this run wrote was evicted except "
+                + ", ".join(stranded)
+                + ", which could not be evicted and may hold work derived from"
+                " the changed source."
+            )
+        raise NodeRejected(refusal)
+
     return RunManifest(
         country=compiled.graph.country,
         nodes=MappingProxyType(receipts),
+        source_identities=MappingProxyType(final_identities),
+        verification_epoch=({} if _verification_epoch is None else _verification_epoch),
         decisions=decisions,
         started_at=started_at,
         finished_at=_now(),
diff --git a/packages/microcosm-graph/src/microcosm/graph/manifest.py b/packages/microcosm-graph/src/microcosm/graph/manifest.py
index 1adb8ded82..48ec44d233 100644
--- a/packages/microcosm-graph/src/microcosm/graph/manifest.py
+++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py
@@ -443,6 +443,23 @@ class RunManifest:
     timestamps, and attached ``Frame`` instances are also outside the
     content-addressed body. Computational reuse continues to use node keys,
     not this run-manifest identity.
+
+    :attr:`source_identities` is attached in the same way: the executor's
+    run-end re-derivation of every declared source's content key, recorded so a
+    reader can see which bytes the run actually finished against. Like
+    ``populations`` and ``mass_ledgers`` it is outside :attr:`key`, outside
+    ``to_json``, and outside every node receipt and cache record.
+
+    :attr:`verification_epoch` is attached the same way again: counts a caller
+    that scoped its own source verification around this run hands in, so a
+    reader can see how many validations the run actually performed rather than
+    inferring it. It is the caller's own mapping and is kept as a live view of
+    it, not a copy, because such a record is finalised when that scope closes --
+    after ``run_graph`` returns and before the caller receives this manifest.
+    That holds for any mapping, not only a ``dict``: the caller's object is
+    wrapped in a read-only proxy over itself, and a mapping type that cannot be
+    wrapped is refused rather than copied into a snapshot that would look live
+    and not be. It holds counts and labels only: no key, digest or path.
     """
 
     country: str
@@ -457,6 +474,12 @@ class RunManifest:
     mass_ledgers: Mapping[str, tuple[MassRecord, ...]] = field(
         default_factory=dict, repr=False, compare=False
     )
+    source_identities: Mapping[str, str] = field(
+        default_factory=dict, repr=False, compare=False
+    )
+    verification_epoch: Mapping[str, object] = field(
+        default_factory=dict, repr=False, compare=False
+    )
 
     def __post_init__(self) -> None:
         if not isinstance(self.country, str):
@@ -507,6 +530,50 @@ def __post_init__(self) -> None:
                 )
             mass_ledgers[version_id] = frozen_records
         object.__setattr__(self, "mass_ledgers", MappingProxyType(mass_ledgers))
+        source_identities: dict[str, str] = {}
+        for name, identity in self.source_identities.items():
+            if not isinstance(name, str):
+                raise TypeError("RunManifest.source_identities keys must be strings")
+            if not isinstance(identity, str):
+                raise TypeError("RunManifest.source_identities values must be strings")
+            source_identities[name] = identity
+        object.__setattr__(
+            self, "source_identities", MappingProxyType(source_identities)
+        )
+        verification_epoch = self.verification_epoch
+        if not isinstance(verification_epoch, Mapping):
+            raise TypeError("RunManifest.verification_epoch must be a mapping")
+        for name, value in verification_epoch.items():
+            if not isinstance(name, str):
+                raise TypeError("RunManifest.verification_epoch keys must be strings")
+            if not isinstance(value, (int, str)):
+                raise TypeError(
+                    "RunManifest.verification_epoch values must be integers or strings"
+                )
+        # A live view of the caller's record, deliberately not a copy: see the
+        # class docstring. Values are checked once, here.
+        #
+        # Every mapping is wrapped, not only a dict. `MappingProxyType` takes
+        # any object with the mapping protocol, so wrapping a caller's own
+        # `Mapping` subclass keeps the reference the contract promises, where
+        # `dict(...)` would have taken a snapshot and dropped exactly the
+        # counts the scope finalises after `run_graph` returns. An already
+        # read-only proxy is kept as it is rather than wrapped again.
+        if isinstance(verification_epoch, MappingProxyType):
+            record: Mapping[str, object] = verification_epoch
+        else:
+            try:
+                record = MappingProxyType(verification_epoch)
+            except TypeError as error:
+                # A type registered as a Mapping without implementing the
+                # protocol cannot be held live, and a copy would be a silent
+                # snapshot, so it is refused rather than accepted as one.
+                raise TypeError(
+                    "RunManifest.verification_epoch must be a mapping that "
+                    "supports MappingProxyType, so the manifest can hold it "
+                    "as a live view rather than a copy"
+                ) from error
+        object.__setattr__(self, "verification_epoch", record)
 
     @property
     def content_addressed(self) -> Mapping[str, object]:
diff --git a/packages/microcosm-graph/src/microcosm/graph/store.py b/packages/microcosm-graph/src/microcosm/graph/store.py
index 6321f46f66..5af840fd81 100644
--- a/packages/microcosm-graph/src/microcosm/graph/store.py
+++ b/packages/microcosm-graph/src/microcosm/graph/store.py
@@ -23,7 +23,8 @@
 import shutil
 import struct
 import uuid
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Iterator, Mapping
+from contextlib import contextmanager
 from pathlib import Path, PurePosixPath
 from typing import Any, Literal
 
@@ -695,6 +696,7 @@ def __init__(self, root: Path, *, codecs: object | None = None) -> None:
         ):
             raise TypeError("codecs must be a mapping, codec registry, or None.")
         self.codecs = codecs
+        self._write_ledgers: list[set[str]] = []
         self.objects.mkdir(parents=True, exist_ok=True)
         self.tmp.mkdir(parents=True, exist_ok=True)
 
@@ -715,6 +717,72 @@ def has(self, key: str) -> bool:
 
     contains = has
 
+    @contextmanager
+    def recording_writes(self) -> Iterator[set[str]]:
+        """Collect the key of every object published while this block is open.
+
+        The set is one caller's record of what it, and nothing before it, put
+        in this store, so it is the set that caller may take back out. A key
+        an object already satisfied is absent from it: that object is not this
+        caller's to remove. A key published over an existing object -- the
+        write-only replacement path -- is present, because the object standing
+        there now is this caller's.
+
+        Blocks nest, and every open block sees every publication under it. The
+        ledger is the store instance's, not a thread's: two callers sharing one
+        instance share its publications, so a block held open across concurrent
+        work collects that work too.
+        """
+
+        written: set[str] = set()
+        self._write_ledgers.append(written)
+        try:
+            yield written
+        finally:
+            # By identity: two ledgers holding the same keys compare equal, and
+            # `list.remove` would take the wrong one.
+            for index in reversed(range(len(self._write_ledgers))):
+                if self._write_ledgers[index] is written:
+                    del self._write_ledgers[index]
+                    break
+
+    def _note_write(self, key: str) -> None:
+        for ledger in self._write_ledgers:
+            ledger.add(key)
+
+    def evict(self, key: str) -> bool:
+        """Remove one object, so this key is a miss again.
+
+        Returns whether an object was there to remove, and removing one that
+        is already gone is not an error: calling this twice on a key leaves
+        the same store as calling it once. The object is renamed out of the
+        tree before anything is deleted, so a reader sees the whole object or
+        no object, never a partial one. The store has no reader lock, so a
+        loader that already opened this object reads on against the renamed
+        directory and one that has not sees the miss.
+
+        This is for a writer taking its own writes back. Every object is
+        content-addressed and derivable, so an eviction costs recomputation
+        and never a wrong answer, but it is still removal: a key another run
+        also published is a miss for that run too.
+        """
+
+        key = _require_key(key)
+        destination = self.object_path(key)
+        if not destination.exists():
+            return False
+        self.tmp.mkdir(parents=True, exist_ok=True)
+        displaced = self.tmp / f"{uuid.uuid4().hex}-evicted"
+        try:
+            os.replace(destination, displaced)
+        except FileNotFoundError:
+            return False
+        _fsync_directory(destination.parent)
+        # The object is already invisible; a tree that will not delete is
+        # tmp-directory litter, not a failed eviction.
+        shutil.rmtree(displaced, ignore_errors=True)
+        return True
+
     def metadata(self, key: str, *, kind: str | None = None) -> Mapping[str, Any]:
         """Return validated object metadata."""
 
@@ -736,6 +804,7 @@ def _put(
             if validate_existing is not None:
                 validate_existing(existing)
             return destination
+        published = True
         staging = self.tmp / uuid.uuid4().hex
         staging.mkdir(parents=False, exist_ok=False)
         try:
@@ -765,8 +834,12 @@ def _put(
                     existing = _verified_meta(destination, expected_kind=kind)
                     if validate_existing is not None:
                         validate_existing(existing)
+                    # The object standing here is the incumbent's, not ours.
+                    published = False
                 else:
                     self._replace_write_only_collision(staging, destination)
+            if published:
+                self._note_write(key)
             _fsync_directory(destination.parent)
             return destination
         finally:
diff --git a/packages/microcosm-graph/tests/test_graph_executor_series_stream.py b/packages/microcosm-graph/tests/test_graph_executor_series_stream.py
new file mode 100644
index 0000000000..abc79d06de
--- /dev/null
+++ b/packages/microcosm-graph/tests/test_graph_executor_series_stream.py
@@ -0,0 +1,362 @@
+"""Byte-exact parity for the executor's vectorised column digest stream.
+
+Invented columns only: no data file, no store, no engine.  ``_update_series``
+now builds the framed object-projection bytes for plain float, integer and
+boolean columns with numpy instead of boxing every value into a Python object
+and framing it one at a time.  The emitted bytes -- the ``object`` dtype
+header, the shape header, every per-value length prefix and every payload --
+must be identical to what the boxing loop wrote.  Every test compares the live
+helper against ``_reference_update_series`` below, a verbatim copy of the
+pre-change body, at the byte level rather than only at the digest.
+"""
+
+from __future__ import annotations
+
+import hashlib
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from microcosm.graph.executor import _object_stream, _update_series
+
+
+def _reference_update_scalar(digest, value):
+    """The pre-change scalar body, kept verbatim as the parity reference."""
+
+    if value is pd.NA:
+        payload = b"pd.NA"
+    elif value is pd.NaT:
+        payload = b"pd.NaT"
+    elif value is None:
+        payload = b"None"
+    elif isinstance(value, (float, np.floating)):
+        payload = b"f" + np.asarray([value], dtype=np.float64).tobytes()
+    elif isinstance(value, (bool, np.bool_)):
+        payload = b"b1" if bool(value) else b"b0"
+    elif isinstance(value, (int, np.integer)):
+        payload = b"i" + str(int(value)).encode("ascii")
+    elif isinstance(value, str):
+        payload = b"s" + value.encode("utf-8")
+    elif isinstance(value, (bytes, np.bytes_)):
+        payload = b"y" + bytes(value)
+    else:
+        payload = b"r" + repr(value).encode("utf-8")
+    digest.update(len(payload).to_bytes(8, "little"))
+    digest.update(payload)
+
+
+def _reference_update_array(digest, values):
+    """The pre-change array body, kept verbatim as the parity reference."""
+
+    array = np.asarray(values)
+    digest.update(str(array.dtype).encode("utf-8"))
+    digest.update(b"\0")
+    digest.update(str(array.shape).encode("ascii"))
+    digest.update(b"\0")
+    if array.dtype.hasobject:
+        for value in array.ravel(order="C"):
+            _reference_update_scalar(digest, value)
+    else:
+        digest.update(np.ascontiguousarray(array).tobytes())
+
+
+def _reference_update_series(digest, series):
+    """The pre-change series body, kept verbatim as the parity reference."""
+
+    digest.update(str(series.dtype).encode("utf-8"))
+    digest.update(b"\0")
+    extension = series.array
+    data = getattr(extension, "_data", None)
+    mask = getattr(extension, "_mask", None)
+    if isinstance(data, np.ndarray) and isinstance(mask, np.ndarray):
+        _reference_update_array(digest, data)
+        _reference_update_array(digest, mask)
+    else:
+        _reference_update_array(digest, series.to_numpy(dtype=object, copy=False))
+    _reference_update_array(digest, series.index.to_numpy(copy=False))
+
+
+class _Recorder:
+    """A digest-shaped sink that keeps every byte the helper writes."""
+
+    def __init__(self):
+        self.chunks = []
+
+    def update(self, payload):
+        self.chunks.append(bytes(payload))
+
+    def bytes(self):
+        return b"".join(self.chunks)
+
+
+def _written(helper, series):
+    recorder = _Recorder()
+    helper(recorder, series)
+    return recorder.bytes()
+
+
+def _digest(helper, series):
+    digest = hashlib.sha256()
+    helper(digest, series)
+    return digest.digest()
+
+
+def _assert_identical(series):
+    assert _written(_update_series, series) == _written(
+        _reference_update_series, series
+    )
+    assert _digest(_update_series, series) == _digest(_reference_update_series, series)
+
+
+def _invented_columns():
+    """One column of every kind a projected kernel context can carry."""
+
+    generator = np.random.default_rng(20260915)
+    columns = {
+        "float64": pd.Series(generator.normal(size=97)),
+        "float32": pd.Series(generator.normal(size=97).astype("float32")),
+        "float16": pd.Series(generator.normal(size=97).astype("float16")),
+        "longdouble": pd.Series(np.array([1.5, 1 / 3, np.nan], dtype=np.longdouble)),
+        "int64": pd.Series(generator.integers(-(10**15), 10**15, size=97)),
+        "int32": pd.Series(
+            generator.integers(-(2**30), 2**30, size=97).astype("int32")
+        ),
+        "int8": pd.Series(generator.integers(-128, 127, size=97).astype("int8")),
+        "uint64": pd.Series(np.array([0, 1, np.iinfo(np.uint64).max], dtype="uint64")),
+        "uint8": pd.Series(np.arange(256, dtype="uint8")),
+        "bool": pd.Series(generator.integers(0, 2, size=97).astype(bool)),
+        "complex": pd.Series(np.array([1 + 2j, 3 - 4j])),
+        "Int64": pd.Series(pd.array([1, None, -3], dtype="Int64")),
+        "Float64": pd.Series(pd.array([1.0, None, -0.0], dtype="Float64")),
+        "boolean": pd.Series(pd.array([True, None, False], dtype="boolean")),
+        "string": pd.Series(["a", "bb", None, "é\U0001f600"], dtype="str"),
+        "object": pd.Series(
+            ["x", 1, None, 2.5, True, b"z", pd.NA, pd.NaT, (1, 2)], dtype="object"
+        ),
+        "category": pd.Series(pd.Categorical(generator.choice(list("abcd"), size=97))),
+        "datetime": pd.Series(pd.to_datetime([0, 1_600_000_000], unit="s")),
+        "timedelta": pd.Series(pd.to_timedelta([0, 90_061], unit="s")),
+    }
+    for name in ("float64", "int64", "bool", "object", "str"):
+        columns[f"empty-{name}"] = pd.Series([], dtype=name)
+    columns["sliced-float"] = pd.Series(generator.normal(size=500))[13:400:7]
+    columns["strided-int"] = pd.Series(generator.integers(0, 10**6, size=500))[::11]
+    columns["labelled-index"] = pd.Series(
+        generator.normal(size=17), index=[f"k{i}" for i in range(17)]
+    )
+    return columns
+
+
+@pytest.mark.parametrize("name", sorted(_invented_columns()))
+def test_every_column_kind_streams_identical_bytes(name):
+    _assert_identical(_invented_columns()[name])
+
+
+def test_float_specials_stream_identical_bytes():
+    """Signed zero, both infinities, NaN and the subnormal extremes."""
+
+    _assert_identical(
+        pd.Series(
+            [
+                0.0,
+                -0.0,
+                np.inf,
+                -np.inf,
+                np.nan,
+                1.7976931348623157e308,
+                -1.7976931348623157e308,
+                5e-324,
+                -5e-324,
+                -1.5,
+            ]
+        )
+    )
+
+
+def test_nan_payload_bits_survive_the_vectorised_path():
+    """A non-canonical NaN keeps its exact payload, sign bit included."""
+
+    raw = np.array(
+        [0x7FF8000000ABCDEF, 0xFFF8000000ABCDEF, 0x7FF0000000000001], dtype="uint64"
+    )
+    series = pd.Series(raw.view("float64"))
+    assert series.isna().all()
+    _assert_identical(series)
+
+
+def test_integer_extremes_stream_identical_bytes():
+    """Variable-width decimal payloads, including both int64 endpoints."""
+
+    _assert_identical(
+        pd.Series(
+            np.array(
+                [0, -1, 1, 9, 10, -10, np.iinfo(np.int64).min, np.iinfo(np.int64).max],
+                dtype="int64",
+            )
+        )
+    )
+
+
+def test_categorical_is_not_mistaken_for_its_integer_codes():
+    """The fast path must refuse any column whose values are a coded view."""
+
+    series = pd.Series(pd.Categorical(list("abca")))
+    assert series.array.dtype.kind not in "fiub"
+    assert _object_stream(series) is None
+    _assert_identical(series)
+
+
+def test_datetime_and_string_columns_take_the_reference_loop():
+    for series in (
+        pd.Series(pd.to_datetime([0, 1], unit="s")),
+        pd.Series(pd.to_timedelta([0, 1], unit="s")),
+        pd.Series(["a", None], dtype="str"),
+        pd.Series([object()], dtype="object"),
+    ):
+        assert _object_stream(series) is None
+        _assert_identical(series)
+
+
+def test_masked_columns_keep_their_existing_data_and_mask_path():
+    """Nullable columns already streamed their raw buffers; nothing moves."""
+
+    for dtype in ("Int64", "Float64", "boolean"):
+        series = pd.Series(pd.array([1, None, 0], dtype=dtype))
+        assert isinstance(series.array._data, np.ndarray)
+        _assert_identical(series)
+
+
+def test_random_frames_of_every_kind_stream_identical_bytes():
+    """The property sweep: random widths and contents, every column kind."""
+
+    generator = np.random.default_rng(20260916)
+    kinds = (
+        "float64",
+        "float32",
+        "int64",
+        "int32",
+        "uint32",
+        "bool",
+        "Int64",
+        "str",
+        "category",
+        "object",
+    )
+    for trial in range(200):
+        rows = int(generator.integers(0, 60))
+        kind = kinds[trial % len(kinds)]
+        if kind.startswith("float"):
+            values = generator.normal(size=rows).astype(kind)
+            series = pd.Series(values)
+            if rows:
+                series.iloc[int(generator.integers(0, rows))] = np.nan
+        elif kind in ("int64", "int32", "uint32"):
+            series = pd.Series(generator.integers(0, 10**9, size=rows).astype(kind))
+        elif kind == "bool":
+            series = pd.Series(generator.integers(0, 2, size=rows).astype(bool))
+        elif kind == "Int64":
+            series = pd.Series(
+                pd.array(
+                    [
+                        None if generator.random() < 0.3 else int(value)
+                        for value in generator.integers(-99, 99, size=rows)
+                    ],
+                    dtype="Int64",
+                )
+            )
+        elif kind == "str":
+            series = pd.Series(
+                [
+                    None
+                    if generator.random() < 0.2
+                    else "s" * int(generator.integers(0, 5))
+                    for _ in range(rows)
+                ],
+                dtype="str",
+            )
+        elif kind == "category":
+            series = pd.Series(pd.Categorical(generator.choice(list("xyz"), size=rows)))
+        else:
+            series = pd.Series(
+                [
+                    [1, "a", None, 2.5, True][int(generator.integers(0, 5))]
+                    for _ in range(rows)
+                ],
+                dtype="object",
+            )
+        _assert_identical(series)
+
+
+# --------------------------------------------------------------------------
+# The length prefix, built without numpy on either side.
+# --------------------------------------------------------------------------
+
+
+def _expected_stream(payloads):
+    """The framed stream, assembled in pure Python.
+
+    Every parity assertion above compares one numpy construction against
+    another, so a byte order both sides take from the host cancels out. This
+    reference takes its byte order from `int.to_bytes(8, "little")` and its
+    payloads from Python `bytes`, which is what `_update_scalar` writes and
+    what neither branch of `_object_stream` may drift from.
+    """
+
+    return b"".join(
+        len(payload).to_bytes(8, "little") + payload for payload in payloads
+    )
+
+
+@pytest.mark.parametrize(
+    "dtype", ["int64", "int32", "int16", "int8", "uint32", "uint8"]
+)
+def test_the_integer_prefix_is_little_endian_whatever_the_host_is(dtype):
+    """The vectorised prefix is cast after the arithmetic, not before it.
+
+    `filled.sum(axis=1)` is a ufunc result and a ufunc's output carries the
+    host's byte order, so casting it to `<u8` and then adding one for the type
+    byte hands `tobytes()` a native-order array again. Casting the finished
+    length is what makes these eight bytes little-endian on every host. The
+    two agree on a little-endian host, which is every host this suite runs on
+    today, so what this case pins is the construction rather than a difference
+    visible here.
+    """
+
+    info = np.iinfo(dtype)
+    values = [0, 1, 9, 10, info.min, info.max]
+    series = pd.Series(np.array(values, dtype=dtype))
+    expected = _expected_stream(
+        [b"i" + str(int(value)).encode("ascii") for value in values]
+    )
+
+    assert _object_stream(series) == expected
+    # The first prefix is the shortest payload's, so a byte-swapped eight-byte
+    # word would be caught by this whole-stream comparison and not only by the
+    # widest one.
+    assert expected[:8] == (2).to_bytes(8, "little")
+
+
+def test_the_boolean_prefix_is_little_endian_too():
+    """`_framed`'s `<u8` prefix, against the same pure-Python reference."""
+
+    series = pd.Series(np.array([True, False, True], dtype=bool))
+    assert _object_stream(series) == _expected_stream([b"b1", b"b0", b"b1"])
+
+
+def test_the_float_prefix_is_little_endian_while_its_payload_is_native():
+    """Nine bytes of payload, prefixed little-endian, packed in host order.
+
+    The payload deliberately follows the host: the loop packs
+    `np.asarray([value], dtype=np.float64).tobytes()`. The prefix deliberately
+    does not.
+    """
+
+    values = [1.5, -0.0, np.inf]
+    series = pd.Series(np.array(values, dtype="float64"))
+    expected = _expected_stream(
+        [b"f" + np.asarray([value], dtype=np.float64).tobytes() for value in values]
+    )
+
+    assert _object_stream(series) == expected
+    assert expected[:8] == (9).to_bytes(8, "little")
diff --git a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py
new file mode 100644
index 0000000000..8e62059031
--- /dev/null
+++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py
@@ -0,0 +1,864 @@
+"""The executor's source-identity contract, and the per-run cache under it.
+
+Until this file the per-node source re-derivation at the heart of
+``run_graph`` had no regression test at all: nothing asserted that a source
+mutated while a node ran makes the run refuse, and nothing asserted that the
+refusal lands before the node's artifacts reach the store. Both are asserted
+here, first, so the cache added underneath them is measured against a pinned
+contract rather than against silence.
+
+The cache reuses a source's content key only while the path's stat signature
+is unchanged, so every mutation that moves any stat field -- content, size,
+inode, timestamps, or a directory's roster -- still refuses at the same node
+it refuses at today. The two cases a stat signature cannot decide are closed
+by an unconditional full re-derivation before the manifest is built, which
+also gives the executor something it never had: a source changed during a node
+that declares no source is now caught.
+
+Synthetic toy graph only: no data file, no country package, no engine.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import pytest
+
+import microcosm.graph.executor as graph_executor
+from microcosm.frame import EntitySchema, Frame, WeightKind, Weights
+from microcosm.graph.decl import (
+    Graph,
+    Node,
+    Owned,
+    Slice,
+    SourceRef,
+    StructuralDelta,
+    compile_graph,
+)
+from microcosm.graph.executor import NodeRejected, run_graph
+from microcosm.graph.kernel import (
+    Capabilities,
+    Determinism,
+    KernelContext,
+    KernelRegistry,
+    KernelResult,
+)
+from microcosm.graph.store import ContentStore
+
+if "_toy" not in sys.modules:
+    _SPEC = importlib.util.spec_from_file_location(
+        "_toy", Path(__file__).with_name("_toy.py")
+    )
+    assert _SPEC is not None and _SPEC.loader is not None
+    _TOY = importlib.util.module_from_spec(_SPEC)
+    sys.modules["_toy"] = _TOY
+    _SPEC.loader.exec_module(_TOY)
+
+SOURCE = SourceRef("survey", "csv-tables")
+
+
+class _Kernel:
+    def __init__(self, ref, capabilities, run):
+        self._ref = ref
+        self._capabilities = capabilities
+        self._run = run
+
+    @property
+    def ref(self):
+        return self._ref
+
+    @property
+    def capabilities(self):
+        return self._capabilities
+
+    def implementation_hash(self):
+        return f"impl-{self._ref}"
+
+    def run(self, context):
+        return self._run(context)
+
+
+def _source_path(root: Path, value: int = 0) -> Path:
+    root.mkdir()
+    (root / "value.txt").write_text(str(value), encoding="utf-8")
+    return root
+
+
+def _source_frame(path: Path) -> Frame:
+    offset = int((path / "value.txt").read_text(encoding="utf-8"))
+    person = pd.DataFrame(
+        {
+            "person_id": np.asarray([1, 2, 3], dtype=np.int64),
+            "person_household_id": np.asarray([10, 10, 20], dtype=np.int64),
+            "age": np.asarray([10, 20, 30], dtype=np.int64) + offset,
+        }
+    )
+    household = pd.DataFrame(
+        {
+            "household_id": np.asarray([10, 20], dtype=np.int64),
+            "size": np.asarray([2, 1], dtype=np.int64),
+        }
+    )
+    return Frame(
+        {"person": person, "household": household},
+        EntitySchema(group_entities=("household",)),
+        {"household": Weights(np.asarray([1.0, 2.0]), WeightKind.DESIGN)},
+        pd.Series(["a", "a", "b"], name="stratum"),
+    )
+
+
+CREATE = Node(
+    "survey",
+    "source@1",
+    structural=StructuralDelta.CREATE,
+    sources=("survey",),
+    outputs=(Owned("person", "age", "int64"), Owned("household", "size", "int64")),
+)
+FOLLOWER = Node(
+    "follower",
+    "follower@1",
+    inputs=(Slice("person", ("age",)),),
+    outputs=(Owned("person", "doubled", "float64"),),
+    population="survey",
+)
+
+
+def _graph() -> Graph:
+    return Graph("toy", (SOURCE,), (CREATE, FOLLOWER))
+
+
+def _registry(
+    *,
+    before_source: Callable[[Path], None] | None = None,
+    on_source: Callable[[Path], None] | None = None,
+    on_follower: Callable[[], None] | None = None,
+) -> KernelRegistry:
+    def source(context: KernelContext) -> KernelResult:
+        path = context.sources["survey"]
+        # `before_source` runs ahead of the read, so the kernel's result is
+        # computed from the rewritten bytes; `on_source` runs after it, so the
+        # kernel saw the bytes the run started with.
+        if before_source is not None:
+            before_source(path)
+        frame = _source_frame(path)
+        if on_source is not None:
+            on_source(path)
+        return KernelResult(frame=frame, receipt={"rows": frame.n("person")})
+
+    def follower(context: KernelContext) -> KernelResult:
+        if on_follower is not None:
+            on_follower()
+        table = context.tables["person"]
+        return KernelResult(
+            columns={
+                ("person", "doubled"): pd.Series(
+                    table["age"].to_numpy(dtype=np.float64) * 2.0,
+                    index=pd.Index(table["person_id"], name="person_id"),
+                    dtype="float64",
+                )
+            },
+            receipt={"rows": len(table)},
+        )
+
+    registry = KernelRegistry()
+    registry.register(
+        _Kernel(
+            "source@1",
+            Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE),
+            source,
+        )
+    )
+    registry.register(
+        _Kernel("follower@1", Capabilities(Determinism.DETERMINISTIC), follower)
+    )
+    return registry
+
+
+def _run(
+    source: Path,
+    store: ContentStore,
+    registry: KernelRegistry,
+    *,
+    resume: str = "auto",
+):
+    return run_graph(
+        compile_graph(_graph()),
+        sources={"survey": source},
+        store=store,
+        kernels=registry,
+        resume=resume,
+    )
+
+
+def _stored_objects(store: ContentStore) -> set[str]:
+    return {
+        path.relative_to(store.objects).as_posix()
+        for path in store.objects.rglob("*")
+        if path.is_file()
+    }
+
+
+# --------------------------------------------------------------------------
+# The contract: a source that moves while the run executes makes it refuse.
+# --------------------------------------------------------------------------
+
+
+def test_a_source_rewritten_by_its_own_node_refuses_that_node(tmp_path: Path) -> None:
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+
+    def rewrite(path: Path) -> None:
+        (path / "value.txt").write_text("7", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="changed source"):
+        _run(source, store, _registry(on_source=rewrite))
+    # The refusal precedes the node's own persistence: nothing new was written.
+    assert _stored_objects(store) == before
+
+
+def test_a_file_added_to_a_directory_source_refuses(tmp_path: Path) -> None:
+    """The roster is part of the signature, so an addition is never cached over."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+
+    def add(path: Path) -> None:
+        (path / "extra.txt").write_text("invented", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="changed source"):
+        _run(source, store, _registry(on_source=add))
+
+
+def test_a_file_removed_from_a_directory_source_refuses(tmp_path: Path) -> None:
+    source = _source_path(tmp_path / "source")
+    (source / "extra.txt").write_text("invented", encoding="utf-8")
+    store = ContentStore(tmp_path / "store")
+
+    def remove(path: Path) -> None:
+        (path / "extra.txt").unlink()
+
+    with pytest.raises(NodeRejected, match="changed source"):
+        _run(source, store, _registry(on_source=remove))
+
+
+def test_a_source_changed_during_a_source_free_node_refuses_at_run_end(
+    tmp_path: Path,
+) -> None:
+    """The guarantee the per-node check never had: no node declares it here."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+
+    def rewrite() -> None:
+        (source / "value.txt").write_text("9", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="Run changed source 'survey'"):
+        _run(source, store, _registry(on_follower=rewrite))
+
+
+def test_a_rewritten_source_still_refuses_when_only_its_bytes_moved(
+    tmp_path: Path,
+) -> None:
+    """Same length and modification time: st_ctime_ns is what still moves.
+
+    This is the strongest rewrite an unprivileged process can perform here, and
+    it is *not* stat-preserving: st_ctime_ns moves, so the signature moves and
+    the node refuses. The genuinely stat-preserving case -- all five fields
+    identical -- is the residual the run-end re-derivation exists for, and
+    `test_a_source_changed_during_a_source_free_node_refuses_at_run_end`
+    reaches that re-derivation by the other route.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    target = source / "value.txt"
+    target.write_text("1", encoding="utf-8")
+    original = target.stat()
+
+    def rewrite(path: Path) -> None:
+        (path / "value.txt").write_text("2", encoding="utf-8")
+        os.utime(path / "value.txt", ns=(original.st_atime_ns, original.st_mtime_ns))
+
+    with pytest.raises(NodeRejected, match="source"):
+        _run(source, store, _registry(on_source=rewrite))
+
+
+# --------------------------------------------------------------------------
+# The residual: a rewrite the signature cannot see, and what the run took back.
+# --------------------------------------------------------------------------
+
+
+def _four_field_identity(info: object) -> tuple[int, int, int, int]:
+    """A file identity whose ``st_ctime_ns`` does not discriminate a rewrite.
+
+    No unprivileged process can hold ``st_ctime_ns`` still on the filesystem
+    these tests run on, which is why
+    ``test_a_rewritten_source_still_refuses_when_only_its_bytes_moved``
+    refuses at the node rather than at run end. Dropping that one field models
+    a host where it does not move. The other four do here exactly what they do
+    anywhere: an in-place same-length rewrite leaves device, inode and size
+    alone, and ``os.utime`` puts the modification time back.
+    """
+
+    return (info.st_dev, info.st_ino, info.st_size, info.st_mtime_ns)
+
+
+def _same_length_rewrite(target: Path) -> Callable[[Path], None]:
+    stamp = target.stat()
+
+    def rewrite(_path: Path) -> None:
+        target.write_text("7", encoding="utf-8")  # one byte, as "0" is
+        os.utime(target, ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
+
+    return rewrite
+
+
+def test_a_stat_identical_rewrite_refuses_at_run_end_and_takes_the_run_back(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The one rewrite the per-node check answers out of the cache.
+
+    Every stat field this host discriminates is identical across it, so the
+    node's check is a signature hit and the node persists: its columns, its
+    frame, its weights and its cache record all reach the store, keyed on the
+    source identity the run started with, while the kernel read the rewritten
+    bytes. The refusal therefore lands after persistence, not before it. What
+    the run persisted is evicted before that refusal propagates, so the store
+    ends exactly where it started.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+    monkeypatch.setattr(graph_executor, "_stat_identity", _four_field_identity)
+    rewrite = _same_length_rewrite(source / "value.txt")
+
+    with pytest.raises(NodeRejected, match="Run changed source 'survey'"):
+        _run(source, store, _registry(before_source=rewrite))
+    assert _stored_objects(store) == before
+
+
+def test_the_record_a_stat_identical_rewrite_left_is_not_reused(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Why the eviction is the fix and the refusal alone is not.
+
+    The refused run's cache record is keyed on the pre-rewrite source identity
+    and holds artifacts computed from the rewritten bytes. Put the original
+    bytes back and the next run derives that same identity, and that same node
+    key: a record left behind is a hit, and the run returns ages the source on
+    disk does not imply. It misses and reruns instead.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    target = source / "value.txt"
+    monkeypatch.setattr(graph_executor, "_stat_identity", _four_field_identity)
+
+    with pytest.raises(NodeRejected, match="Run changed source"):
+        _run(source, store, _registry(before_source=_same_length_rewrite(target)))
+
+    target.write_text("0", encoding="utf-8")
+    manifest = _run(source, store, _registry(), resume="auto")
+    assert manifest.nodes["survey"].hit is False
+    ages = manifest.populations["survey"].table("person")["age"]
+    assert list(ages) == [10, 20, 30]  # the bytes on disk, not the refused run's
+
+
+def test_a_source_free_node_refusal_also_takes_back_what_the_run_wrote(
+    tmp_path: Path,
+) -> None:
+    """The other route to the run-end refusal evicts the same way.
+
+    Here the node that ran while the source moved declares no source, so both
+    nodes persisted before the run-end pass looked. Nothing is left behind for
+    a later run to find, whichever route reached the refusal.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+
+    def rewrite() -> None:
+        (source / "value.txt").write_text("9", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="Run changed source 'survey'"):
+        _run(source, store, _registry(on_follower=rewrite))
+    assert _stored_objects(store) == before
+
+
+def test_a_refusal_that_cannot_evict_still_refuses_and_names_what_stayed(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Fail-closed: a store that will not give an object up does not soften it."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+
+    def refuse(self: ContentStore, key: str) -> bool:
+        raise OSError(13, "Permission denied")
+
+    monkeypatch.setattr(ContentStore, "evict", refuse)
+
+    def rewrite() -> None:
+        (source / "value.txt").write_text("9", encoding="utf-8")
+
+    with pytest.raises(NodeRejected) as refusal:
+        _run(source, store, _registry(on_follower=rewrite))
+    assert "Run changed source 'survey' while executing." in str(refusal.value)
+    assert "could not be evicted" in str(refusal.value)
+    assert _stored_objects(store)  # the refusal did not pretend they were gone
+
+
+class _KernelFailedError(RuntimeError):
+    """A follower kernel's own failure, distinct from any executor refusal."""
+
+
+def _failing_follower(error: BaseException) -> Callable[[], None]:
+    def fail() -> None:
+        raise error
+
+    return fail
+
+
+def test_a_later_failure_after_a_stat_identical_rewrite_takes_the_run_back(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The run-end pass is not the only way out of a poisoned run.
+
+    The source node reads rewritten bytes and persists under the pre-rewrite
+    identity, exactly as above, and then the follower's kernel raises. That
+    exception leaves ``run_graph`` without ever reaching the run-end pass, so
+    the same re-derivation runs on the way out and evicts what the run wrote.
+    The kernel failure still reaches the caller, with the eviction noted on it.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+    monkeypatch.setattr(graph_executor, "_stat_identity", _four_field_identity)
+    rewrite = _same_length_rewrite(source / "value.txt")
+
+    with pytest.raises(NodeRejected) as failure:
+        _run(
+            source,
+            store,
+            _registry(
+                before_source=rewrite,
+                on_follower=_failing_follower(_KernelFailedError("follower broke")),
+            ),
+        )
+    assert "Run changed source" not in str(failure.value)  # not the run-end route
+    assert any(
+        "evicted, because source 'survey' changed while it ran" in note
+        for note in getattr(failure.value, "__notes__", [])
+    )
+    assert _stored_objects(store) == before
+
+
+def test_the_record_a_failed_poisoned_run_left_is_not_reused(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The same reuse the run-end eviction prevents, reached by the other exit."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    target = source / "value.txt"
+    monkeypatch.setattr(graph_executor, "_stat_identity", _four_field_identity)
+
+    with pytest.raises(NodeRejected):
+        _run(
+            source,
+            store,
+            _registry(
+                before_source=_same_length_rewrite(target),
+                on_follower=_failing_follower(_KernelFailedError("follower broke")),
+            ),
+        )
+
+    target.write_text("0", encoding="utf-8")
+    manifest = _run(source, store, _registry(), resume="auto")
+    assert manifest.nodes["survey"].hit is False
+    ages = manifest.populations["survey"].table("person")["age"]
+    assert list(ages) == [10, 20, 30]
+
+
+def test_an_interrupt_after_a_stat_identical_rewrite_takes_the_run_back(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """An interrupt is not an ``Exception``, and it is settled the same way."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+    monkeypatch.setattr(graph_executor, "_stat_identity", _four_field_identity)
+    rewrite = _same_length_rewrite(source / "value.txt")
+
+    with pytest.raises(KeyboardInterrupt) as interrupt:
+        _run(
+            source,
+            store,
+            _registry(
+                before_source=rewrite,
+                on_follower=_failing_follower(KeyboardInterrupt()),
+            ),
+        )
+    assert any("evicted" in note for note in getattr(interrupt.value, "__notes__", []))
+    assert _stored_objects(store) == before
+
+
+def test_a_failed_run_whose_sources_held_still_keeps_its_work_for_resume(
+    tmp_path: Path,
+) -> None:
+    """The re-derivation is what lets a failure stay cheap.
+
+    Nothing moved, so the source node's objects are correct and stay: the next
+    run resumes from them instead of recomputing. Evicting on every failure
+    would be safe and would throw this away.
+    """
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+
+    with pytest.raises(NodeRejected) as failure:
+        _run(
+            source,
+            store,
+            _registry(
+                on_follower=_failing_follower(_KernelFailedError("follower broke"))
+            ),
+        )
+    assert not getattr(failure.value, "__notes__", [])
+    kept = _stored_objects(store)
+    assert kept
+
+    manifest = _run(source, store, _registry(), resume="auto")
+    assert manifest.nodes["survey"].hit is True
+    assert kept <= _stored_objects(store)
+
+
+def test_a_failed_run_whose_sources_cannot_be_re_derived_is_taken_back(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Unverifiable is treated as moved: the store is not left to chance."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+
+    def remove_then_fail() -> None:
+        (source / "value.txt").unlink()
+        source.rmdir()
+        raise _KernelFailedError("follower broke")
+
+    with pytest.raises(NodeRejected) as failure:
+        _run(source, store, _registry(on_follower=remove_then_fail))
+    assert any(
+        "its sources could not be re-derived" in note
+        for note in getattr(failure.value, "__notes__", [])
+    )
+    assert _stored_objects(store) == before
+
+
+def test_a_run_end_refusal_is_not_re_derived_a_second_time(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The run-end pass already decided; the failure path must not repeat it."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    derived: list[str] = []
+    original = graph_executor._SourceIdentities.derive
+
+    def counted(self, name, path):
+        derived.append(name)
+        return original(self, name, path)
+
+    monkeypatch.setattr(graph_executor._SourceIdentities, "derive", counted)
+
+    def rewrite() -> None:
+        (source / "value.txt").write_text("9", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="Run changed source 'survey'") as refusal:
+        _run(source, store, _registry(on_follower=rewrite))
+    assert not getattr(refusal.value, "__notes__", [])
+    run_end_only = derived.count("survey")
+
+    derived.clear()
+    clean = tmp_path / "clean"
+    _run(_source_path(clean), ContentStore(tmp_path / "clean-store"), _registry())
+    assert run_end_only == derived.count("survey")
+
+
+# --------------------------------------------------------------------------
+# The cache: fewer reads, identical keys.
+# --------------------------------------------------------------------------
+
+
+def _counting(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
+    seen: list[tuple[str, str]] = []
+    original = graph_executor.source_content_key
+
+    def counted(name, path):
+        seen.append((name, str(path)))
+        return original(name, path)
+
+    monkeypatch.setattr(graph_executor, "source_content_key", counted)
+    return seen
+
+
+def test_each_source_is_read_twice_per_run_not_once_per_node(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    seen = _counting(monkeypatch)
+    manifest = _run(source, store, _registry())
+    # Once when the run's identities are established, once when they are
+    # re-proved at the end. The node in between reuses the first read.
+    assert [name for name, _ in seen] == ["survey", "survey"]
+    assert manifest.nodes["survey"].hit is False
+
+
+def test_an_unchanged_source_is_never_re_read_by_a_node(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A wider graph adds nodes, not reads."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    wide = Graph(
+        "toy",
+        (SOURCE,),
+        (
+            CREATE,
+            FOLLOWER,
+            Node(
+                "second",
+                "source@1",
+                structural=StructuralDelta.CREATE,
+                sources=("survey",),
+                outputs=(
+                    Owned("person", "age", "int64"),
+                    Owned("household", "size", "int64"),
+                ),
+            ),
+        ),
+    )
+    seen = _counting(monkeypatch)
+    run_graph(
+        compile_graph(wide),
+        sources={"survey": source},
+        store=store,
+        kernels=_registry(),
+    )
+    assert len(seen) == 2
+
+
+def test_a_touched_but_unchanged_source_re_reads_and_still_matches(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A signature miss is a re-read, not a refusal."""
+
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    seen = _counting(monkeypatch)
+
+    def touch(path: Path) -> None:
+        os.utime(path / "value.txt", None)
+
+    manifest = _run(source, store, _registry(on_source=touch))
+    assert len(seen) == 3  # run start, the node's miss, run end
+    assert manifest.nodes["survey"].hit is False
+
+
+# --------------------------------------------------------------------------
+# The record, and what it must not move.
+# --------------------------------------------------------------------------
+
+
+def test_the_manifest_records_the_run_end_identities_without_moving_anything(
+    tmp_path: Path,
+) -> None:
+    source = _source_path(tmp_path / "source")
+    store = ContentStore(tmp_path / "store")
+    manifest = _run(source, store, _registry())
+    identity = graph_executor.source_content_key("survey", source)
+    assert dict(manifest.source_identities) == {"survey": identity}
+
+    bare = type(manifest)(
+        country=manifest.country,
+        nodes=manifest.nodes,
+        decisions=manifest.decisions,
+        started_at=manifest.started_at,
+        finished_at=manifest.finished_at,
+        host=manifest.host,
+    )
+    assert bare.key == manifest.key
+    assert bare.to_json() == manifest.to_json()
+    assert "source_identities" not in manifest.to_json()
+    assert dict(manifest.content_addressed) == dict(bare.content_addressed)
+
+
+def test_the_recorded_identities_are_rejected_when_malformed() -> None:
+    from microcosm.graph.manifest import RunManifest
+
+    with pytest.raises(TypeError, match="source_identities keys"):
+        RunManifest(country="toy", nodes={}, source_identities={1: "a"})
+    with pytest.raises(TypeError, match="source_identities values"):
+        RunManifest(country="toy", nodes={}, source_identities={"a": 1})
+
+
+# --------------------------------------------------------------------------
+# The cache object itself.
+# --------------------------------------------------------------------------
+
+
+def test_the_cache_reuses_only_an_unchanged_signature(tmp_path: Path) -> None:
+    source = _source_path(tmp_path / "source")
+    identities = graph_executor._SourceIdentities()
+    first = identities.key("survey", source)
+    assert identities.key("survey", source) == first
+    (source / "value.txt").write_text("5", encoding="utf-8")
+    second = identities.key("survey", source)
+    assert second != first
+    assert identities.key("survey", source) == second
+
+
+def test_the_cache_is_bypassed_by_derive(tmp_path: Path) -> None:
+    source = _source_path(tmp_path / "source")
+    identities = graph_executor._SourceIdentities()
+    identities.key("survey", source)
+    calls: list[str] = []
+    original = graph_executor.source_content_key
+
+    def counted(name, path):
+        calls.append(name)
+        return original(name, path)
+
+    graph_executor.source_content_key = counted
+    try:
+        identities.key("survey", source)
+        assert calls == []
+        identities.derive("survey", source)
+        assert calls == ["survey"]
+    finally:
+        graph_executor.source_content_key = original
+
+
+def test_a_regular_file_source_signature_moves_with_its_bytes(tmp_path: Path) -> None:
+    path = tmp_path / "table.csv"
+    path.write_bytes(b"id\n1\n")
+    before = graph_executor._source_stat_signature(path)
+    assert before[0] == "file"
+    path.write_bytes(b"id\n2\n")
+    assert graph_executor._source_stat_signature(path) != before
+
+
+def test_a_directory_signature_covers_names_as_well_as_contents(
+    tmp_path: Path,
+) -> None:
+    root = _source_path(tmp_path / "source")
+    before = graph_executor._source_stat_signature(root)
+    assert before[0] == "dir"
+    (root / "value.txt").rename(root / "renamed.txt")
+    renamed = graph_executor._source_stat_signature(root)
+    assert renamed != before
+    nested = root / "inner"
+    nested.mkdir()
+    (nested / "deep.txt").write_text("x", encoding="utf-8")
+    assert graph_executor._source_stat_signature(root) != renamed
+
+
+def test_a_source_that_moves_during_its_own_read_is_not_cached(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A torn read leaves no cache entry, so the next consumer re-derives."""
+
+    path = tmp_path / "table.csv"
+    path.write_bytes(b"id\n1\n")
+    identities = graph_executor._SourceIdentities()
+    original = graph_executor.source_content_key
+
+    def tearing(name, source_path):
+        result = original(name, source_path)
+        Path(source_path).write_bytes(b"id\n22\n")
+        return result
+
+    monkeypatch.setattr(graph_executor, "source_content_key", tearing)
+    identities.key("table", path)
+    monkeypatch.setattr(graph_executor, "source_content_key", original)
+    calls: list[str] = []
+
+    def counted(name, source_path):
+        calls.append(name)
+        return original(name, source_path)
+
+    monkeypatch.setattr(graph_executor, "source_content_key", counted)
+    identities.key("table", path)
+    assert calls == ["table"]
+
+
+# --------------------------------------------------------------------------
+# A member symlink: the bytes are the target's, so the signature follows it.
+# --------------------------------------------------------------------------
+
+
+def test_a_symlinked_member_refuses_at_the_node_that_changed_its_target(
+    tmp_path: Path,
+) -> None:
+    """The link's own five stat fields do not move when its target's bytes do.
+
+    ``keys._directory_identity`` selects members with ``is_file()`` and reads
+    them with ``read_bytes()``, both of which follow the link, so a member
+    link's target is inside the source's content key. If the signature stopped
+    at the link, a target rewritten mid-run would be answered from cache at the
+    per-node check and refused only by the run-end re-derivation -- after
+    intervening nodes had written store records. It refuses at the node.
+    """
+
+    source = _source_path(tmp_path / "source")
+    target = tmp_path / "outside" / "linked.txt"
+    target.parent.mkdir()
+    target.write_text("one", encoding="utf-8")
+    (source / "linked.txt").symlink_to(target)
+    store = ContentStore(tmp_path / "store")
+    before = _stored_objects(store)
+
+    def rewrite(_path: Path) -> None:
+        target.write_text("another", encoding="utf-8")
+
+    with pytest.raises(NodeRejected, match="Node 'survey' changed source"):
+        _run(source, store, _registry(on_source=rewrite))
+    # The refusal still precedes the node's own persistence.
+    assert _stored_objects(store) == before
+
+
+def test_a_directory_signature_follows_a_member_symlink(tmp_path: Path) -> None:
+    root = _source_path(tmp_path / "source")
+    target = tmp_path / "outside.txt"
+    target.write_text("one", encoding="utf-8")
+    (root / "linked.txt").symlink_to(target)
+    before = graph_executor._source_stat_signature(root)
+    target.write_text("another", encoding="utf-8")
+    assert graph_executor._source_stat_signature(root) != before
+
+
+def test_a_broken_member_symlink_signs_as_absent_without_raising(
+    tmp_path: Path,
+) -> None:
+    """``_directory_identity`` skips it; the signature records that it is gone."""
+
+    root = _source_path(tmp_path / "source")
+    (root / "dangling.txt").symlink_to(tmp_path / "never-written.txt")
+    signature = graph_executor._source_stat_signature(root)
+    assert any(member[-1][0] == "absent" for member in signature[2] if len(member) == 4)
+    (tmp_path / "never-written.txt").write_text("now here", encoding="utf-8")
+    assert graph_executor._source_stat_signature(root) != signature
diff --git a/packages/microcosm-graph/tests/test_graph_store.py b/packages/microcosm-graph/tests/test_graph_store.py
index 1761ded91e..5fb6f37014 100644
--- a/packages/microcosm-graph/tests/test_graph_store.py
+++ b/packages/microcosm-graph/tests/test_graph_store.py
@@ -294,3 +294,113 @@ def test_json_receipt_and_opaque_bytes_are_content_validated(tmp_path: Path) ->
         "ok": True,
     }
     assert store.load_bytes(_key("e")) == b"model bytes"
+
+
+# --------------------------------------------------------------------------
+# The write ledger, and taking one caller's own writes back out.
+# --------------------------------------------------------------------------
+
+
+def test_the_ledger_holds_what_was_published_not_what_was_already_there(
+    tmp_path: Path,
+) -> None:
+    store = ContentStore(tmp_path / "store")
+    existing = _key("1")
+    store.put_bytes(existing, b"from an earlier run")
+
+    with store.recording_writes() as written:
+        store.put_bytes(existing, b"from an earlier run")  # satisfied, not published
+        store.put_bytes(_key("2"), b"this run's")
+        store.put_json(_key("3"), {"this": "run"})
+
+    assert written == {_key("2"), _key("3")}
+
+
+def test_a_write_only_replacement_is_this_caller_s_to_take_back(
+    tmp_path: Path,
+) -> None:
+    """``verify_existing=False`` puts this caller's object where the old one was."""
+
+    store = ContentStore(tmp_path / "store")
+    key = _key("4")
+    store.put_bytes(key, b"incumbent")
+
+    with store.recording_writes() as written:
+        store.put_bytes(key, b"replacement", verify_existing=False)
+
+    assert written == {key}
+
+
+def test_ledgers_nest_and_each_closes_without_disturbing_the_other(
+    tmp_path: Path,
+) -> None:
+    store = ContentStore(tmp_path / "store")
+
+    with store.recording_writes() as outer:
+        store.put_bytes(_key("5"), b"outer")
+        with store.recording_writes() as inner:
+            store.put_bytes(_key("6"), b"inner")
+        assert inner == {_key("6")}
+        store.put_bytes(_key("7"), b"outer again")
+
+    assert outer == {_key("5"), _key("6"), _key("7")}
+    assert store._write_ledgers == []
+
+
+def test_two_empty_ledgers_close_without_stranding_each_other(tmp_path: Path) -> None:
+    """Equal sets are not the same ledger; closing takes the one that opened."""
+
+    store = ContentStore(tmp_path / "store")
+    with store.recording_writes():
+        with store.recording_writes():
+            pass
+        store.put_bytes(_key("8"), b"after the inner close")
+    assert store._write_ledgers == []
+
+
+def test_a_ledger_closes_when_the_block_under_it_raises(tmp_path: Path) -> None:
+    store = ContentStore(tmp_path / "store")
+    with pytest.raises(RuntimeError, match="refused"):
+        with store.recording_writes():
+            store.put_bytes(_key("9"), b"written before the refusal")
+            raise RuntimeError("refused")
+    assert store._write_ledgers == []
+
+
+def test_eviction_makes_a_key_a_miss_again_and_repeats_harmlessly(
+    tmp_path: Path,
+) -> None:
+    store = ContentStore(tmp_path / "store")
+    key = _key("a")
+    store.put_bytes(key, b"evict me")
+    assert store.has(key)
+
+    assert store.evict(key) is True
+    assert store.has(key) is False
+    assert store.evict(key) is False  # idempotent
+    with pytest.raises(StoreMiss):
+        store.load_bytes(key)
+
+    store.put_bytes(key, b"written again")
+    assert store.load_bytes(key) == b"written again"
+
+
+def test_eviction_leaves_no_partial_object_and_no_tmp_litter(tmp_path: Path) -> None:
+    store = ContentStore(tmp_path / "store")
+    key = _key("b")
+    store.put_frame(key, _frame())
+
+    assert store.evict(key) is True
+    assert not store.object_path(key).exists()
+    assert list(store.tmp.iterdir()) == []
+
+
+def test_eviction_of_a_key_never_written_is_not_an_error(tmp_path: Path) -> None:
+    store = ContentStore(tmp_path / "store")
+    assert store.evict(_key("c")) is False
+
+
+def test_eviction_rejects_a_malformed_key(tmp_path: Path) -> None:
+    store = ContentStore(tmp_path / "store")
+    with pytest.raises(ValueError, match="key"):
+        store.evict("not-a-key")
diff --git a/packages/microcosm-graph/tests/test_graph_verification_epoch.py b/packages/microcosm-graph/tests/test_graph_verification_epoch.py
new file mode 100644
index 0000000000..2a2c89557f
--- /dev/null
+++ b/packages/microcosm-graph/tests/test_graph_verification_epoch.py
@@ -0,0 +1,410 @@
+"""A run's verification-epoch record: carried, and outside everything computed.
+
+``RunManifest.verification_epoch`` and ``run_graph(_verification_epoch=...)``
+are how a country runtime that scopes source verification around a whole run
+tells a reader how many validations that run actually performed. Both shipped
+with no test in this shard: nothing asserted that attaching a record leaves the
+node keys, the manifest's JSON and its content-addressed body exactly where
+they were; nothing asserted that the manifest keeps the caller's own mapping
+rather than a copy -- the property the whole design leans on, because the
+record is finalised when the caller's scope closes, after ``run_graph`` has
+returned; and nothing asserted that a malformed record is refused rather than
+carried.
+
+``source_identities`` next door has
+``test_the_manifest_records_the_run_end_identities_without_moving_anything``;
+this file is the same contract for the field beside it.
+
+Synthetic toy graph only: no data file, no country package, no engine.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping
+from pathlib import Path
+from types import MappingProxyType
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from microcosm.frame import EntitySchema, Frame, WeightKind, Weights
+from microcosm.graph.decl import (
+    Graph,
+    Node,
+    Owned,
+    Slice,
+    SourceRef,
+    StructuralDelta,
+    compile_graph,
+)
+from microcosm.graph.executor import run_graph
+from microcosm.graph.kernel import (
+    Capabilities,
+    Determinism,
+    KernelContext,
+    KernelRegistry,
+    KernelResult,
+)
+from microcosm.graph.manifest import NodeReceipt, RunManifest
+from microcosm.graph.store import ContentStore
+
+SOURCE = SourceRef("survey", "csv-tables")
+
+
+class _Kernel:
+    def __init__(self, ref, capabilities, run):
+        self._ref = ref
+        self._capabilities = capabilities
+        self._run = run
+
+    @property
+    def ref(self):
+        return self._ref
+
+    @property
+    def capabilities(self):
+        return self._capabilities
+
+    def implementation_hash(self):
+        return f"impl-{self._ref}"
+
+    def run(self, context):
+        return self._run(context)
+
+
+def _source_path(root: Path) -> Path:
+    root.mkdir()
+    (root / "value.txt").write_text("0", encoding="utf-8")
+    return root
+
+
+def _source_frame() -> Frame:
+    person = pd.DataFrame(
+        {
+            "person_id": np.asarray([1, 2, 3], dtype=np.int64),
+            "person_household_id": np.asarray([10, 10, 20], dtype=np.int64),
+            "age": np.asarray([10, 20, 30], dtype=np.int64),
+        }
+    )
+    household = pd.DataFrame(
+        {
+            "household_id": np.asarray([10, 20], dtype=np.int64),
+            "size": np.asarray([2, 1], dtype=np.int64),
+        }
+    )
+    return Frame(
+        {"person": person, "household": household},
+        EntitySchema(group_entities=("household",)),
+        {"household": Weights(np.asarray([1.0, 2.0]), WeightKind.DESIGN)},
+        pd.Series(["a", "a", "b"], name="stratum"),
+    )
+
+
+CREATE = Node(
+    "survey",
+    "source@1",
+    structural=StructuralDelta.CREATE,
+    sources=("survey",),
+    outputs=(Owned("person", "age", "int64"), Owned("household", "size", "int64")),
+)
+FOLLOWER = Node(
+    "follower",
+    "follower@1",
+    inputs=(Slice("person", ("age",)),),
+    outputs=(Owned("person", "doubled", "float64"),),
+    population="survey",
+)
+
+
+def _registry() -> KernelRegistry:
+    def source(context: KernelContext) -> KernelResult:
+        frame = _source_frame()
+        return KernelResult(frame=frame, receipt={"rows": frame.n("person")})
+
+    def follower(context: KernelContext) -> KernelResult:
+        table = context.tables["person"]
+        return KernelResult(
+            columns={
+                ("person", "doubled"): pd.Series(
+                    table["age"].to_numpy(dtype=np.float64) * 2.0,
+                    index=pd.Index(table["person_id"], name="person_id"),
+                    dtype="float64",
+                )
+            },
+            receipt={"rows": len(table)},
+        )
+
+    registry = KernelRegistry()
+    registry.register(
+        _Kernel(
+            "source@1",
+            Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE),
+            source,
+        )
+    )
+    registry.register(
+        _Kernel("follower@1", Capabilities(Determinism.DETERMINISTIC), follower)
+    )
+    return registry
+
+
+def _run(root: Path, record: Mapping[str, object] | None = None) -> RunManifest:
+    """One cold run under ``root``, with its own source tree and its own store."""
+
+    root.mkdir()
+    return run_graph(
+        compile_graph(Graph("toy", (SOURCE,), (CREATE, FOLLOWER))),
+        sources={"survey": _source_path(root / "source")},
+        store=ContentStore(root / "store"),
+        kernels=_registry(),
+        **({} if record is None else {"_verification_epoch": record}),
+    )
+
+
+def _keys(manifest: RunManifest) -> dict[str, str]:
+    return {node_id: manifest.nodes[node_id].key for node_id in sorted(manifest.nodes)}
+
+
+def _json_without_run_local_fields(manifest: RunManifest) -> dict[str, object]:
+    """This manifest's JSON document, less exactly the fields a run may move.
+
+    ``to_json`` is not identical between two runs of one computation and is not
+    meant to be: it carries three run-level manifest fields -- ``started_at``,
+    ``finished_at`` and ``host`` -- and, inside every node receipt, the two the
+    receipt itself names as run-level, ``NodeReceipt.RUN_LEVEL_FIELDS``. None of
+    the five reaches the manifest key, by two different routes: ``key`` hashes
+    ``content_addressed``, which is ``nodes`` and ``tier`` alone, so the two
+    timestamps and the host are outside it entirely, and each receipt enters it
+    through ``NodeReceipt._content_payload``, which drops ``RUN_LEVEL_FIELDS``.
+    Removing exactly those, by their own names and by that class attribute
+    rather than by a list written here, is what lets the rest of the document
+    be compared between two runs; a ``del`` of a field that stopped being
+    serialized would raise rather than quietly widen what this hides. Nothing
+    is removed from ``decisions``, ``content_addressed`` or any other field.
+    """
+
+    payload = json.loads(manifest.to_json())
+    for name in ("started_at", "finished_at", "host"):
+        del payload[name]
+    for receipt in payload["nodes"].values():
+        for name in sorted(NodeReceipt.RUN_LEVEL_FIELDS):
+            del receipt[name]
+    return payload
+
+
+# --------------------------------------------------------------------------
+# What the record must not move.
+# --------------------------------------------------------------------------
+
+
+def test_a_record_is_outside_the_key_the_json_and_the_content_address(
+    tmp_path: Path,
+) -> None:
+    """The same manifest without the field is byte-identical everywhere it counts."""
+
+    manifest = _run(
+        tmp_path / "run", {"protocol": "toy/verification-epoch/1", "hits": 4}
+    )
+    assert dict(manifest.verification_epoch) == {
+        "protocol": "toy/verification-epoch/1",
+        "hits": 4,
+    }
+
+    bare = type(manifest)(
+        country=manifest.country,
+        nodes=manifest.nodes,
+        decisions=manifest.decisions,
+        started_at=manifest.started_at,
+        finished_at=manifest.finished_at,
+        host=manifest.host,
+    )
+    assert bare.key == manifest.key
+    assert bare.to_json() == manifest.to_json()
+    assert "verification_epoch" not in manifest.to_json()
+    assert dict(manifest.content_addressed) == dict(bare.content_addressed)
+
+
+def test_two_runs_agree_on_every_node_key_with_and_without_a_record(
+    tmp_path: Path,
+) -> None:
+    """Executed separately, into separate stores: the record buys no reuse and no key."""
+
+    with_record = _run(
+        tmp_path / "with", {"protocol": "toy/verification-epoch/1", "hits": 9}
+    )
+    without = _run(tmp_path / "without")
+    assert _keys(with_record) == _keys(without)
+    assert with_record.key == without.key
+    assert dict(with_record.content_addressed) == dict(without.content_addressed)
+    # The serialized document too, and not only the key and the body it
+    # hashes: every field of it except the five a run is allowed to move.
+    # ``verification_epoch`` is not among them -- it is not serialized at all,
+    # which the bare-manifest test above pins directly.
+    assert _json_without_run_local_fields(
+        with_record
+    ) == _json_without_run_local_fields(without)
+    # Both ran cold, so the equality above is between two computations rather
+    # than between a run and a cache hit replaying the first one's receipts.
+    assert [node.hit for node in with_record.nodes.values()] == [False, False]
+    assert [node.hit for node in without.nodes.values()] == [False, False]
+
+
+def test_no_record_is_an_empty_mapping_not_a_missing_attribute(tmp_path: Path) -> None:
+    manifest = _run(tmp_path / "run")
+    assert dict(manifest.verification_epoch) == {}
+    assert "verification_epoch" not in manifest.to_json()
+
+
+# --------------------------------------------------------------------------
+# The live view, which is the whole reason the field exists.
+# --------------------------------------------------------------------------
+
+
+def test_the_manifest_holds_the_caller_s_own_record_not_a_copy(tmp_path: Path) -> None:
+    """A caller's scope closes after ``run_graph`` returns, and the counts follow.
+
+    The runner that opens a verification epoch hands ``run_graph`` the record
+    that epoch yields, and that epoch's final counts -- how many capsules were
+    re-validated as it closed -- are written into the record only when the
+    scope leaves, which is after this call has returned its manifest. A copy
+    taken at construction would freeze the record mid-run and report a final
+    validation count of zero for every run.
+    """
+
+    record = {"protocol": "toy/verification-epoch/1", "hits": 0, "misses": 0}
+    manifest = _run(tmp_path / "run", record)
+    assert manifest.verification_epoch["hits"] == 0
+
+    record["hits"] = 18
+    record["final_validations"] = 2
+    assert manifest.verification_epoch["hits"] == 18
+    assert manifest.verification_epoch["final_validations"] == 2
+    assert dict(manifest.verification_epoch) == record
+
+    # A view, so a reader of the manifest cannot write the caller's record.
+    assert isinstance(manifest.verification_epoch, MappingProxyType)
+    with pytest.raises(TypeError):
+        manifest.verification_epoch["hits"] = 0  # type: ignore[index]
+
+
+def test_a_mapping_proxy_is_kept_as_it_is_given(tmp_path: Path) -> None:
+    """A caller that already wrapped its record is not wrapped twice."""
+
+    record = {"protocol": "toy/verification-epoch/1", "capsules": 1}
+    view = MappingProxyType(record)
+    manifest = _run(tmp_path / "run", view)
+    assert manifest.verification_epoch is view
+    record["capsules"] = 3
+    assert manifest.verification_epoch["capsules"] == 3
+
+
+class _EpochRecord(Mapping):
+    """A caller's own record type: a ``Mapping``, and not a ``dict``.
+
+    A runtime that scopes verification around a run need not hand back a plain
+    dict -- the natural shape is an object the scope keeps writing into and
+    that reads as a mapping. The live-view contract has to hold for it too.
+    """
+
+    def __init__(self, backing: dict[str, object]) -> None:
+        self._backing = backing
+
+    def __getitem__(self, name: str) -> object:
+        return self._backing[name]
+
+    def __iter__(self):
+        return iter(self._backing)
+
+    def __len__(self) -> int:
+        return len(self._backing)
+
+
+def test_a_custom_mapping_is_held_live_exactly_as_a_dict_is(tmp_path: Path) -> None:
+    """The contract is the caller's mapping, not the caller's dict.
+
+    Wrapped rather than copied: ``dict(...)`` here would have frozen the
+    record at construction and reported a final validation count of zero,
+    which is the failure the live view exists to prevent -- silently, and only
+    for callers whose record is not a ``dict``.
+    """
+
+    backing: dict[str, object] = {
+        "protocol": "toy/verification-epoch/1",
+        "hits": 0,
+        "final_validations": 0,
+    }
+    manifest = _run(tmp_path / "run", _EpochRecord(backing))
+    assert manifest.verification_epoch["hits"] == 0
+
+    # The caller's scope closes here, after `run_graph` returned.
+    backing["hits"] = 18
+    backing["final_validations"] = 2
+    backing["capsules"] = 2
+
+    assert manifest.verification_epoch["hits"] == 18
+    assert manifest.verification_epoch["final_validations"] == 2
+    assert dict(manifest.verification_epoch) == backing
+    assert isinstance(manifest.verification_epoch, MappingProxyType)
+    with pytest.raises(TypeError):
+        manifest.verification_epoch["hits"] = 0  # type: ignore[index]
+
+
+def test_a_mapping_that_cannot_be_held_live_is_refused_not_copied() -> None:
+    """Fail-closed: a snapshot that looks live is worse than a refusal.
+
+    A type registered as a ``Mapping`` without implementing the subscript
+    protocol passes ``isinstance`` and cannot be wrapped. Copying it would
+    produce a manifest whose field reads like every other one and silently
+    stops tracking, so it is refused instead.
+    """
+
+    @Mapping.register
+    class _NotSubscriptable:
+        def items(self):
+            return iter((("hits", 1),))
+
+    with pytest.raises(TypeError, match="live view rather than a copy"):
+        RunManifest(country="toy", nodes={}, verification_epoch=_NotSubscriptable())
+
+
+# --------------------------------------------------------------------------
+# A malformed record is refused, not carried.
+# --------------------------------------------------------------------------
+
+
+def test_a_record_that_is_not_a_mapping_is_refused() -> None:
+    with pytest.raises(TypeError, match="verification_epoch must be a mapping"):
+        RunManifest(country="toy", nodes={}, verification_epoch=("hits", 1))
+
+
+def test_a_record_key_that_is_not_a_string_is_refused() -> None:
+    with pytest.raises(TypeError, match="verification_epoch keys must be strings"):
+        RunManifest(country="toy", nodes={}, verification_epoch={1: 2})
+
+
+def test_a_record_value_that_is_neither_an_integer_nor_a_string_is_refused() -> None:
+    with pytest.raises(
+        TypeError, match="verification_epoch values must be integers or strings"
+    ):
+        RunManifest(country="toy", nodes={}, verification_epoch={"wall_time": 1.5})
+
+
+def test_counts_and_labels_are_what_a_record_may_hold() -> None:
+    """Integers and strings, which is what an epoch record is: no key, digest or path."""
+
+    manifest = RunManifest(
+        country="toy",
+        nodes={},
+        verification_epoch={
+            "protocol": "toy/verification-epoch/1",
+            "hits": 18,
+            "misses": 1,
+            "final_validations": 2,
+            "capsules": 2,
+        },
+    )
+    assert manifest.verification_epoch["protocol"] == "toy/verification-epoch/1"
+    assert manifest.verification_epoch["final_validations"] == 2
+    assert "verification_epoch" not in manifest.to_json()