Derive graph source identities once per run and stream column digests from buffers

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.

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

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]:

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: