From a904fa41db6f6eac964ff6a623b555b4fdab8695 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 16 Sep 2026 14:13:31 -0400 Subject: [PATCH 01/16] Build each column's context-digest stream with numpy, byte for byte `_context_digest` digests every column of every table twice per node, and `_update_series` took a fast path only for masked extension arrays exposing `_data` and `_mask` ndarrays. On pandas 3.0.3 a plain float, integer or boolean column is a `NumpyExtensionArray` with neither, so it fell through to `to_numpy(dtype=object)` and a per-value Python loop with two `hashlib` updates per value. `_object_stream` builds exactly the bytes that loop wrote -- the same `object` dtype header, the same shape header, the same per-value length prefix and payload -- with numpy, for the three kinds whose boxed value is always one exact Python type: `float` packed to eight IEEE-754 bytes, `bool` rendered as `b"b1"`/`b"b0"`, and `int` rendered as a variable-width decimal. It returns None for everything else, including a `Categorical`, whose values array is an integer-coded view that would otherwise be mistaken for an integer column, so those columns still take the loop. Both `_context_digest` calls per node remain: the second one is the guard that the kernel did not mutate its input context, and reusing its result would delete the guard. No digest value moves. `test_graph_executor_series_stream.py` compares the shipped helper against a verbatim copy of the pre-change body at the byte level, not only at the digest: every column kind the projection can carry, the float specials (signed zero, both infinities, the subnormal extremes), non-canonical NaN payloads with their sign bit, both `int64` endpoints, empty, sliced and strided columns, labelled indexes, and a 200-frame random sweep. Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 74 ++++- .../test_graph_executor_series_stream.py | 288 ++++++++++++++++++ 2 files changed, 361 insertions(+), 1 deletion(-) create mode 100644 packages/microcosm-graph/tests/test_graph_executor_series_stream.py diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index b61a678d4c..27356d883a 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -442,6 +442,69 @@ 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=" 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. + wide = np.ascontiguousarray(values, dtype=" None: digest.update(str(series.dtype).encode("utf-8")) digest.update(b"\0") @@ -452,7 +515,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)) 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..7b6d54e1c6 --- /dev/null +++ b/packages/microcosm-graph/tests/test_graph_executor_series_stream.py @@ -0,0 +1,288 @@ +"""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) From 873b0e9642173409d80785ec65460843b1960c35 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 16 Sep 2026 14:13:44 -0400 Subject: [PATCH 02/16] Derive each source content key once per run, and prove them at run end After every cold-executed node the executor re-derived `source_content_key(name, source_paths[name])` for every source that node declares, and `keys._directory_identity` `read_bytes()`es every file of a directory source and hashes the whole tree. Cost is O(executed source-declaring nodes x source bytes). `run_graph` gains `_SourceIdentities`, a cache in its own locals -- never inside `source_content_key`, which stays pure because existing assertions call it twice on one path with the bytes changed in between and require different answers. A cached key is reused only while the path's stat signature is identical to the signature taken both immediately before and immediately after the read that produced it: for a regular file its `(st_dev, st_ino, st_size, st_mtime_ns, st_ctime_ns)`, for a directory its own identity plus the relative name, file type and identity of every entry `_directory_identity` would walk, so a file added, removed or renamed inside the tree moves the signature even when no member's own stat changes. A derivation whose two signatures disagree is not cached at all. The refusal is not weakened. Every mutation that moves any stat field is a cache miss, re-derives in full and raises the same `NodeRejected` at the same node, before that node's `_write_node`. Immediately before the manifest is built, every source is then re-derived in full with the cache bypassed, which closes the two cases a stat signature cannot decide -- a rewrite that leaves all five stat fields identical, and a change made during a node that declares no source at all, which the per-node check has never seen. That pass is written inline rather than through `_source_paths_and_keys` because a build test counts calls to that exact code object. `RunManifest` gains `source_identities`, attached exactly like `populations` and `mass_ledgers`: `repr=False`, `compare=False`, outside `content_addressed`, outside `key`, outside `to_json`, and outside every node receipt and cache record. It carries the run-end re-derived key of every source. The per-node refusal had no test anywhere in this repository before this commit. `test_graph_executor_source_identity.py` pins it first -- a source rewritten by its own node refuses and the store gains no object, a file added to or removed from a directory source refuses, a rewrite that moves only bytes refuses, a source changed during a source-free node refuses at run end -- and only then measures the cache against that pinned contract. Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 131 ++++- .../src/microcosm/graph/manifest.py | 19 + .../test_graph_executor_source_identity.py | 477 ++++++++++++++++++ 3 files changed, 622 insertions(+), 5 deletions(-) create mode 100644 packages/microcosm-graph/tests/test_graph_executor_source_identity.py diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 27356d883a..2be65e210f 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -6,6 +6,7 @@ import json import pickle import socket +import stat import time from collections.abc import Callable, Mapping from copy import deepcopy @@ -2051,6 +2052,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} @@ -2061,7 +2064,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 @@ -2079,8 +2082,94 @@ 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 _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. + """ + + 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() + entries.append( + ( + candidate.relative_to(path).as_posix(), + stat.S_IFMT(entry.st_mode), + _stat_identity(entry), + ) + ) + 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( @@ -2346,7 +2435,13 @@ 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 + ) keys, implementations = _all_node_keys(compiled, kernels, source_keys) contracts = { node_id: typed_contracts(compiled, compiled.graph.node(node_id), keys, kernels) @@ -2515,7 +2610,11 @@ 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; anything that moved + # is re-derived in full here, so the refusal still precedes + # every _write_node this run performs. + 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." @@ -2685,9 +2784,31 @@ 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 all five stat fields identical, and a + # change made during a node that declares no source at all, which the + # per-node check has never seen -- and it is written inline rather than + # through _source_paths_and_keys because a build test counts calls to that + # exact code object. + 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) + if moved: + raise NodeRejected( + "Run changed source " + + ", ".join(repr(name) for name in moved) + + " while executing." + ) + return RunManifest( country=compiled.graph.country, nodes=MappingProxyType(receipts), + source_identities=MappingProxyType(final_identities), 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..6b1a526f3b 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -443,6 +443,12 @@ 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. """ country: str @@ -457,6 +463,9 @@ 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 + ) def __post_init__(self) -> None: if not isinstance(self.country, str): @@ -507,6 +516,16 @@ 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) + ) @property def content_addressed(self) -> Mapping[str, object]: 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..1d51ab0ed8 --- /dev/null +++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py @@ -0,0 +1,477 @@ +"""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( + *, + on_source: Callable[[Path], None] | None = None, + on_follower: Callable[[], None] | None = None, +) -> KernelRegistry: + def source(context: KernelContext) -> KernelResult: + frame = _source_frame(context.sources["survey"]) + if on_source is not None: + on_source(context.sources["survey"]) + 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): + return run_graph( + compile_graph(_graph()), + sources={"survey": source}, + store=store, + kernels=registry, + ) + + +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, restored mtime: the cache must not answer from memory.""" + + 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 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"] From 3f38fc3159ffa195da83a20d86dfa8e65c4c8d96 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 16 Sep 2026 14:13:48 -0400 Subject: [PATCH 03/16] Say where the executor actually checks that a source moved `load_raw_bytes` described the check as "the executor's post-run content check". It is not post-run: it runs after every cold node that declares the source, and now again over every source in full before the run manifest is built. Docstring only. Co-Authored-By: Claude Opus --- packages/microcosm-graph/src/microcosm/graph/codecs.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 From 8ea48447cf3245a8ac4acc0242cc9178d50acb13 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 16 Sep 2026 14:13:51 -0400 Subject: [PATCH 04/16] Add the changelog fragment for the graph verify-once change Co-Authored-By: Claude Opus --- changelog.d/graph-verify-once.changed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/graph-verify-once.changed.md diff --git a/changelog.d/graph-verify-once.changed.md b/changelog.d/graph-verify-once.changed.md new file mode 100644 index 0000000000..36fd9d5149 --- /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. No digest, key, seal or receipt value changes. From 6206b4a7522ae9dbbbb4032f980f5933f772634e Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 02:45:01 -0400 Subject: [PATCH 05/16] Carry a run's verification epoch onto the manifest it returns A country runtime that scopes source verification around a whole run yields a record of what that scope did -- memo hits, signature misses, final validations, capsule count. Nothing carried it out of the scope, so it was dropped the moment the run ended and nothing anywhere said how many full validations a real run performed. `run_graph` takes a private `_verification_epoch` mapping and attaches it to the manifest as `RunManifest.verification_epoch`. It is a live view of the caller's own mapping, kept rather than copied, because such a record is finalised when the scope closes -- which happens after `run_graph` returns and before the caller receives the manifest. `__post_init__` checks it is a mapping of string keys to integer or string values, so it carries counts and labels only: no key, digest or path. Like `source_identities` it is `repr=False`, `compare=False`, and outside the addressed surface: `content_addressed` projects `nodes` and `tier` alone, and `to_json` names its fields explicitly, so neither reaches it. No existing receipt field moves and no digest, key, seal or receipt value changes. The callers that bind such a record are the US native runners on #935's stack and are not on this branch; this is the graph-package half, which is the half that reaches CI here. Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 9 +++++ .../src/microcosm/graph/manifest.py | 34 +++++++++++++++++++ 2 files changed, 43 insertions(+) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 2be65e210f..a217ce53e7 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -2409,6 +2409,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. @@ -2419,6 +2420,13 @@ 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. """ if resume not in ("auto", "require", "forbid"): @@ -2809,6 +2817,7 @@ def run_graph( 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 6b1a526f3b..4620a6634c 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -449,6 +449,14 @@ class RunManifest: 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. + It holds counts and labels only: no key, digest or path. """ country: str @@ -466,6 +474,9 @@ class RunManifest: 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): @@ -526,6 +537,29 @@ def __post_init__(self) -> None: 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. + object.__setattr__( + self, + "verification_epoch", + verification_epoch + if isinstance(verification_epoch, MappingProxyType) + else MappingProxyType( + verification_epoch + if isinstance(verification_epoch, dict) + else dict(verification_epoch) + ), + ) @property def content_addressed(self) -> Mapping[str, object]: From d87196ce90bbfb4abb4e6bae3e0f25d89c896b09 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 02:45:48 -0400 Subject: [PATCH 06/16] Follow a member symlink in the source signature, as the content key does `_source_stat_signature` lstat()ed every entry of a directory source, while `keys._directory_identity` selects members with `is_file()` and reads them with `read_bytes()` -- both of which follow symlinks. A member link's target was therefore inside the source's content key and outside the signature over it, so rewriting that target left all five of the link's stat fields untouched, the per-node check answered from cache, and the refusal was deferred to the run-end re-derivation, after intervening nodes had written store records. The signature now contributes the resolved identity of every member that is a symlink, so both sides walk the same members. A link that resolves to nothing contributes that fact and raises nothing, matching `_directory_identity` skipping it. `test_a_symlinked_member_refuses_at_the_node_that_changed_its_target` runs a graph whose kernel rewrites the target and asserts the node refuses and the store gained no object; two unit tests pin the signature itself, including the broken-link case. This narrows what the run-end re-derivation is left holding -- it does not weaken it. The genuinely stat-preserving rewrite and the change made during a source-free node are still its residual, and both keep their tests. $ uv run --no-sync pytest \ packages/microcosm-graph/tests/test_graph_executor_source_identity.py \ -p no:cacheprovider 18 passed in 0.87s Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 35 ++++++++--- .../test_graph_executor_source_identity.py | 58 +++++++++++++++++++ 2 files changed, 86 insertions(+), 7 deletions(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index a217ce53e7..1a748e2bb9 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -2102,6 +2102,16 @@ def _stat_identity(info: object) -> tuple[int, int, int, int, int]: ) +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. @@ -2112,6 +2122,16 @@ def _source_stat_signature(path: Path) -> tuple[object, ...]: 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() @@ -2122,13 +2142,14 @@ def _source_stat_signature(path: Path) -> tuple[object, ...]: entries = [] for candidate in sorted(path.rglob("*")): entry = candidate.lstat() - entries.append( - ( - candidate.relative_to(path).as_posix(), - stat.S_IFMT(entry.st_mode), - _stat_identity(entry), - ) - ) + 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)) diff --git a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py index 1d51ab0ed8..a3ec1ab068 100644 --- a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py +++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py @@ -475,3 +475,61 @@ def counted(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 From 284252e56741ed96a6154cfd1bfed6749abee54f Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 02:46:07 -0400 Subject: [PATCH 07/16] Pack the float stream in native order, as the loop it replaces does `_object_stream`'s float branch cast to an explicit ` --- .../microcosm-graph/src/microcosm/graph/executor.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 1a748e2bb9..58c43add7b 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -482,8 +482,14 @@ def _object_stream(series: pd.Series) -> bytes | None: 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. - wide = np.ascontiguousarray(values, dtype=" Date: Thu, 17 Sep 2026 02:46:29 -0400 Subject: [PATCH 08/16] Say what the stat-preserving rewrite test actually exercises `test_a_rewritten_source_still_refuses_when_only_its_bytes_moved` read as though it exercised a stat-preserving rewrite. It does not: it restores st_mtime_ns and holds the length, but st_ctime_ns still moves, and st_ctime_ns is one of the five fields `_stat_identity` binds -- so the signature moves and the node refuses on the ordinary path. That is the strongest rewrite an unprivileged process can perform here. The genuinely stat-preserving case, all five fields identical, is the residual the run-end re-derivation exists for. The docstring now says which of the two it is and points at `test_a_source_changed_during_a_source_free_node_refuses_at_run_end`, which reaches that re-derivation by the other route. Docstring only: no assertion, no behaviour and no emitted byte changes. The rest of #935's docstring pass corrects the design note, the report and the epoch finalizer's nested-refusal chaining, none of which are on this branch. Co-Authored-By: Claude Opus --- .../tests/test_graph_executor_source_identity.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py index a3ec1ab068..55d6d1ed29 100644 --- a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py +++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py @@ -252,7 +252,15 @@ def rewrite() -> None: def test_a_rewritten_source_still_refuses_when_only_its_bytes_moved( tmp_path: Path, ) -> None: - """Same length, restored mtime: the cache must not answer from memory.""" + """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") From 33f3150bbe580d2742f8408eddc1ee93c1e247a1 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 03:42:56 -0400 Subject: [PATCH 09/16] Test the verification-epoch record this PR adds `RunManifest.verification_epoch` and `run_graph(_verification_epoch=...)` landed here with no test in this shard at all: 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 as a live view rather than a copy -- the property the 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. `test_graph_verification_epoch.py` asserts all three on a synthetic toy graph, including each of the three `__post_init__` refusals and the mapping-proxy passthrough, and pins that no record is an empty mapping rather than a missing attribute. Mirrored verbatim from the stacked lane branch `native-verify-once`, where the country runtime that fills the record lives. Co-Authored-By: Claude Opus --- .../tests/test_graph_verification_epoch.py | 306 ++++++++++++++++++ 1 file changed, 306 insertions(+) create mode 100644 packages/microcosm-graph/tests/test_graph_verification_epoch.py 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..5c54b02199 --- /dev/null +++ b/packages/microcosm-graph/tests/test_graph_verification_epoch.py @@ -0,0 +1,306 @@ +"""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 + +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 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)} + + +# -------------------------------------------------------------------------- +# 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) + # 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 + + +# -------------------------------------------------------------------------- +# 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() From ec2b7b414b9697924eba1a7a95ba35e3330d8abf Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 07:58:45 -0400 Subject: [PATCH 10/16] Compare the serialized manifest between the two cold runs too The two-run test asserted the node keys, the manifest key and the content-addressed body, but not the JSON document, so a record that leaked into `to_json` was caught only by the bare-manifest test beside it. It could not assert the document outright: two runs differ in the three run-level manifest fields and in the two `NodeReceipt` already names as run-level, which is why the manifest key hashes receipts without them. It now compares the whole document with exactly those five removed, taken by name and from `RUN_LEVEL_FIELDS` rather than from a list written in the test, so a field that stopped being serialized raises instead of quietly widening what the comparison hides. With the record written into `to_json`, this test now fails on that assertion alone. Co-Authored-By: Claude Opus --- .../tests/test_graph_verification_epoch.py | 36 ++++++++++++++++++- 1 file changed, 35 insertions(+), 1 deletion(-) diff --git a/packages/microcosm-graph/tests/test_graph_verification_epoch.py b/packages/microcosm-graph/tests/test_graph_verification_epoch.py index 5c54b02199..264f9a6c26 100644 --- a/packages/microcosm-graph/tests/test_graph_verification_epoch.py +++ b/packages/microcosm-graph/tests/test_graph_verification_epoch.py @@ -20,6 +20,7 @@ from __future__ import annotations +import json from collections.abc import Mapping from pathlib import Path from types import MappingProxyType @@ -46,7 +47,7 @@ KernelRegistry, KernelResult, ) -from microcosm.graph.manifest import RunManifest +from microcosm.graph.manifest import NodeReceipt, RunManifest from microcosm.graph.store import ContentStore SOURCE = SourceRef("survey", "csv-tables") @@ -166,6 +167,32 @@ 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``. Those + five are the fields the manifest's own docstring calls the ones a run may + change without changing what was computed, which is why the manifest key + already hashes each receipt without them. 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. # -------------------------------------------------------------------------- @@ -210,6 +237,13 @@ def test_two_runs_agree_on_every_node_key_with_and_without_a_record( 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] From 522dc44a1d4973ef763577ab306d3bc0aca54e94 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 12:03:21 -0400 Subject: [PATCH 11/16] Say where the five removed fields actually sit, not where they nearly sit The helper's note said all five removed fields are what the manifest's own docstring calls the ones a run may change without changing what was computed, and that this is why the key hashes each receipt without them. That is the argument for `hit` and `wall_time` only. `started_at`, `finished_at` and `host` are not receipt fields at all: they are absent from the content-addressed body the key hashes, which is `nodes` and `tier` alone. The note now gives the two routes separately -- outside `content_addressed` for the three, dropped by `_content_payload` for the two -- so a reader checking it against manifest.py finds each claim where the note says it is. No assertion changes. Co-Authored-By: Claude Opus --- .../tests/test_graph_verification_epoch.py | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/packages/microcosm-graph/tests/test_graph_verification_epoch.py b/packages/microcosm-graph/tests/test_graph_verification_epoch.py index 264f9a6c26..2922f1b492 100644 --- a/packages/microcosm-graph/tests/test_graph_verification_epoch.py +++ b/packages/microcosm-graph/tests/test_graph_verification_epoch.py @@ -173,15 +173,16 @@ def _json_without_run_local_fields(manifest: RunManifest) -> dict[str, object]: ``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``. Those - five are the fields the manifest's own docstring calls the ones a run may - change without changing what was computed, which is why the manifest key - already hashes each receipt without them. 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. + 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()) From 7e0178207a3b3bc79d3e311328ea2ad05c1c27ed Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:19:06 -0400 Subject: [PATCH 12/16] Take back what a refused run wrote, and say which check catches what The per-node source check answers from a stat signature, so the one rewrite that signature cannot see -- same length, in place, every field identical -- is a cache hit. The node proceeds, `_write_node` persists its columns, frame, weights and cache record under the source identity the run started with, and only the run-end re-derivation refuses. The ContentStore had no way to remove an object, so that record survived the refusal and the next `resume="auto"` run over the same store hit it: the run returned values the source on disk does not imply. The inline comments said the refusal precedes every `_write_node`, which is true only of the rewrites the signature does see. `ContentStore.recording_writes()` opens a ledger that collects the key of every object published while it is open -- publications only, so an object an earlier run already satisfied is not this caller's to remove, while a `verify_existing=False` replacement is. Blocks nest and close by identity. `ContentStore.evict(key)` renames the object out of the tree before deleting it, so a reader sees the whole object or none, returns whether one was there, and evicting twice leaves the same store as evicting once. `run_graph` holds one ledger around the run and spends it on the run-end refusal: everything the run published is evicted before the refusal propagates. The eviction never replaces the refusal -- an object that will not go is named in the refusal text, and the run still refuses. Both comments now state the guarantee the two checks actually make together: a rewrite that moves any stat field refuses before the node that would have persisted work derived from it; a rewrite that moves none of them is refused at run end, after persistence, and everything the run persisted is taken back. Four executor tests and nine store tests. The stat-identical rewrite is modelled by dropping `st_ctime_ns` from `_stat_identity`, because no unprivileged process can hold that field still here -- which is exactly why `test_a_rewritten_source_still_refuses_when_only_its_bytes_moved` refuses at the node instead. The other four fields do in the test what they do anywhere: an in-place same-length rewrite leaves device, inode and size alone and `os.utime(ns=)` puts the modification time back. Against the pre-fix executor and store, the two tests that matter fail on the defect itself rather than on a missing attribute: packages/.../test_a_stat_identical_rewrite_refuses_at_run_end_... E AssertionError: assert {'11/116c8180...ues.npy', ...} == set() packages/.../test_the_record_a_stat_identical_rewrite_left_is_not_reused E AssertionError: assert True is False E + where True = NodeReceipt(key='92ade67d...', hit=True, ...).hit No digest, key, seal or receipt value moves: eviction removes content-addressed objects that are all re-derivable, and the refusal path gains no `except` that could swallow it. $ uv run --no-sync pytest packages/microcosm-graph/tests 610 passed, 2 skipped in 19.16s Co-Authored-By: Claude Opus --- changelog.d/graph-verify-once.changed.md | 2 +- .../src/microcosm/graph/executor.py | 105 +++++++++++-- .../src/microcosm/graph/store.py | 75 ++++++++- .../test_graph_executor_source_identity.py | 147 +++++++++++++++++- .../microcosm-graph/tests/test_graph_store.py | 110 +++++++++++++ 5 files changed, 424 insertions(+), 15 deletions(-) diff --git a/changelog.d/graph-verify-once.changed.md b/changelog.d/graph-verify-once.changed.md index 36fd9d5149..cd1067d2c4 100644 --- a/changelog.d/graph-verify-once.changed.md +++ b/changelog.d/graph-verify-once.changed.md @@ -1 +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. No digest, key, seal or receipt value changes. +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. No digest, key, seal or receipt value changes. diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 58c43add7b..54f8ce357f 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -8,7 +8,7 @@ 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 datetime import UTC, datetime @@ -2427,6 +2427,25 @@ 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 + + def run_graph( compiled: CompiledGraph, *, @@ -2454,6 +2473,42 @@ def run_graph( 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. + """ + + with store.recording_writes() as written: + return _execute_graph( + compiled, + sources=sources, + store=store, + kernels=kernels, + resume=resume, + decisions=decisions, + written=written, + _population_observer=_population_observer, + _verification_epoch=_verification_epoch, + ) + + +def _execute_graph( + compiled: CompiledGraph, + *, + sources: Mapping[str, Path], + store: ContentStore, + kernels: KernelRegistry, + resume: ResumePolicy, + decisions: tuple[Decision, ...], + written: set[str], + _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. """ if resume not in ("auto", "require", "forbid"): @@ -2646,9 +2701,14 @@ def run_graph( raise NodeRejected(f"Node {node.id!r} mutated its input context.") for name in node.sources: # A stat signature that still matches the one taken across this - # source's own read reuses that read's key; anything that moved - # is re-derived in full here, so the refusal still precedes - # every _write_node this run performs. + # 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. current = source_identities.key(name, source_paths[name]) if current != source_keys[name]: raise NodeRejected( @@ -2821,11 +2881,27 @@ def run_graph( # 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 all five stat fields identical, and a - # change made during a node that declares no source at all, which the - # per-node check has never seen -- and it is written inline rather than - # through _source_paths_and_keys because a build test counts calls to that - # exact code object. + # 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. + # + # 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 paths and re-checks codec availability, which + # this pass has no reason to repeat. The build shard on the branch stacked + # on this one also profiles that helper's exact code object, in + # test_us_graph_survey_population.py, test_authenticated_cold_and_ + # materialized_warm_clones_both_sources, timing one source-key pass per + # call; nothing at this head references it. final_identities = {} moved = [] for name in sorted(source_paths): @@ -2834,11 +2910,20 @@ def run_graph( if identity != source_keys[name]: moved.append(name) if moved: - raise NodeRejected( + 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, 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_source_identity.py b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py index 55d6d1ed29..24c647d1f0 100644 --- a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py +++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py @@ -135,13 +135,20 @@ def _graph() -> Graph: 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: - frame = _source_frame(context.sources["survey"]) + 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(context.sources["survey"]) + on_source(path) return KernelResult(frame=frame, receipt={"rows": frame.n("person")}) def follower(context: KernelContext) -> KernelResult: @@ -173,12 +180,19 @@ def follower(context: KernelContext) -> KernelResult: return registry -def _run(source: Path, store: ContentStore, registry: KernelRegistry): +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, ) @@ -276,6 +290,133 @@ def rewrite(path: Path) -> None: _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 + + # -------------------------------------------------------------------------- # The cache: fewer reads, identical keys. # -------------------------------------------------------------------------- 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") From d1f405865dc1b063ce833dfe09613da0f4df5b3f Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:21:43 -0400 Subject: [PATCH 13/16] Cast the integer length after the arithmetic, not before it `_object_stream`'s integer branch built its eight-byte length prefixes as `filled.sum(axis=1).astype("u8` fails the new integer cases along with the existing parity ones, so they are wired to the prefix bytes; on a big-endian host the pre-fix code would fail them for the reason the fix addresses. No emitted byte changes on a little-endian host, so no digest, key, seal or receipt value moves here. $ uv run --no-sync pytest packages/microcosm-graph/tests 618 passed, 2 skipped in 18.58s Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 11 ++- .../test_graph_executor_series_stream.py | 74 +++++++++++++++++++ 2 files changed, 81 insertions(+), 4 deletions(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 54f8ce357f..dce83dbb04 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -500,11 +500,14 @@ def _object_stream(series: pd.Series) -> bytes | None: 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 - lengths = filled.sum(axis=1).astype(" Date: Thu, 17 Sep 2026 15:23:59 -0400 Subject: [PATCH 14/16] Hold every caller's epoch record live, not only a dict `RunManifest.verification_epoch` is documented as a live view of the caller's own mapping, because the record is finalised when the caller's verification scope closes -- after `run_graph` has returned and before the caller reads the manifest. `__post_init__` honoured that for a `dict` and for a `MappingProxyType` and copied everything else through `dict(...)`. A caller whose record is its own `Mapping` type -- the natural shape for an object a scope keeps writing into -- therefore got a snapshot taken mid-run, reporting a final validation count of zero, with nothing to distinguish it from a field that was working: the same type, the same reads, silently stale. `MappingProxyType` takes any object with the mapping protocol, so every mapping is now wrapped rather than copied, and an already read-only proxy is still passed through untouched. A type registered as a `Mapping` without implementing that protocol cannot be held live; it is refused with a message that says why, rather than accepted as the copy the contract forbids. The class docstring says both. Two tests. A `Mapping` subclass backed by a dict is mutated after `run_graph` returns and the manifest follows it, exactly as the dict case beside it does, and stays read-only through the proxy. The unwrappable type is refused. Against the pre-fix constructor the first fails on the snapshot itself -- E assert 0 == 18 -- and the second on the opaque error a copy attempt raises instead: E TypeError: '_NotSubscriptable' object is not iterable Values are still checked once at construction, which is what a live view can check; no key, digest, JSON field or receipt is touched, and the field remains outside `content_addressed`, `key` and `to_json`. $ uv run --no-sync pytest packages/microcosm-graph/tests 620 passed, 2 skipped in 17.97s Co-Authored-By: Claude Opus --- .../src/microcosm/graph/manifest.py | 38 ++++++---- .../tests/test_graph_verification_epoch.py | 69 +++++++++++++++++++ 2 files changed, 95 insertions(+), 12 deletions(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/manifest.py b/packages/microcosm-graph/src/microcosm/graph/manifest.py index 4620a6634c..48ec44d233 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -456,7 +456,10 @@ class RunManifest: 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. - It holds counts and labels only: no key, digest or path. + 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 @@ -549,17 +552,28 @@ def __post_init__(self) -> None: ) # A live view of the caller's record, deliberately not a copy: see the # class docstring. Values are checked once, here. - object.__setattr__( - self, - "verification_epoch", - verification_epoch - if isinstance(verification_epoch, MappingProxyType) - else MappingProxyType( - verification_epoch - if isinstance(verification_epoch, dict) - else dict(verification_epoch) - ), - ) + # + # 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/tests/test_graph_verification_epoch.py b/packages/microcosm-graph/tests/test_graph_verification_epoch.py index 2922f1b492..2a2c89557f 100644 --- a/packages/microcosm-graph/tests/test_graph_verification_epoch.py +++ b/packages/microcosm-graph/tests/test_graph_verification_epoch.py @@ -300,6 +300,75 @@ def test_a_mapping_proxy_is_kept_as_it_is_given(tmp_path: Path) -> None: 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. # -------------------------------------------------------------------------- From 46dee2bc990a066f748a1006932c79018229213b Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:25:11 -0400 Subject: [PATCH 15/16] Name the test that actually profiles the source-key helper The run-end block justified being inlined partly by "a build test counts calls to that exact code object", naming no test. Nothing at this head references `_source_paths_and_keys` outside executor.py, so a reader checking that claim here finds nothing and has no way to tell whether the constraint is real or left over. The test is real and lives on the stacked branch `native-verify-once`: `test_authenticated_cold_and_materialized_warm_clones_both_sources` in `packages/microcosm-build/tests/test_us_graph_survey_population.py` registers a `sys.setprofile` hook over a table of code-object ids that includes `_source_paths_and_keys.__code__` under the label `source_key`, and times one source-key pass per call/return pair. It profiles rather than asserts a count, which the old wording also got wrong. The comment now gives the file, what the hook does, and a grep that finds it, and says plainly that nothing at this head references the helper outside this module. It also leads with the reason that stands on its own here -- the helper answers from the per-run cache this pass exists to bypass, and re-resolves paths and re-checks codec availability that this pass has no reason to repeat -- with the stacked shard's expectation second. Comment only: no behaviour, no assertion and no emitted byte changes. $ uv run --no-sync pytest packages/microcosm-graph/tests 620 passed, 2 skipped in 19.09s Co-Authored-By: Claude Opus --- .../src/microcosm/graph/executor.py | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index dce83dbb04..030e21b00a 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -2899,12 +2899,16 @@ def _execute_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 paths and re-checks codec availability, which - # this pass has no reason to repeat. The build shard on the branch stacked - # on this one also profiles that helper's exact code object, in - # test_us_graph_survey_population.py, test_authenticated_cold_and_ - # materialized_warm_clones_both_sources, timing one source-key pass per - # call; nothing at this head references it. + # 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): From efe9cb6b5dead3d069eadfb2ea8f6fef52ba792d Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Thu, 17 Sep 2026 15:45:32 -0400 Subject: [PATCH 16/16] Settle a run that leaves by exception, not only one that reaches run end The run-end re-derivation is the only check that sees a rewrite leaving every stat field identical, and a run that raises never reaches it: a later kernel failure, a per-node refusal, a store error or an interrupt unwound through the write ledger with the poisoned columns, frame, weights and cache record still in the store under the pre-rewrite identity, where the next resume="auto" run would hit them. run_graph now runs the same cache-bypassing re-derivation on the way out of any exception. Sources that still match leave the store alone, so a failed or interrupted run stays resumable. A source that moved, a re-derivation that cannot finish, or a failure before the run had source keys evicts everything the run wrote. The original error is never replaced; what happened is added to it as a note. The run-end pass marks the run settled either way, so its own refusal is not re-derived twice. What no in-process check covers is a process that stops without unwinding after such a rewrite. That residual is now stated on run_graph, at the per-node check, at the run-end pass and in the changelog fragment. Co-Authored-By: Claude Fable 5.1 --- changelog.d/graph-verify-once.changed.md | 2 +- .../src/microcosm/graph/executor.py | 139 ++++++++++++-- .../test_graph_executor_source_identity.py | 180 ++++++++++++++++++ 3 files changed, 306 insertions(+), 15 deletions(-) diff --git a/changelog.d/graph-verify-once.changed.md b/changelog.d/graph-verify-once.changed.md index cd1067d2c4..d78631b6b9 100644 --- a/changelog.d/graph-verify-once.changed.md +++ b/changelog.d/graph-verify-once.changed.md @@ -1 +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. No digest, key, seal or receipt value changes. +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/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 030e21b00a..ff6f0e9fd5 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -10,7 +10,7 @@ import time 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 @@ -2449,6 +2449,86 @@ def _evict_run_writes(store: ContentStore, written: Iterable[str]) -> list[str]: 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, *, @@ -2480,21 +2560,35 @@ def run_graph( 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. + 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: - return _execute_graph( - compiled, - sources=sources, - store=store, - kernels=kernels, - resume=resume, - decisions=decisions, - written=written, - _population_observer=_population_observer, - _verification_epoch=_verification_epoch, - ) + 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( @@ -2506,12 +2600,15 @@ def _execute_graph( 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"): @@ -2535,6 +2632,9 @@ def _execute_graph( 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) @@ -2711,7 +2811,10 @@ def _execute_graph( # 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. + # 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( @@ -2897,6 +3000,11 @@ def _execute_graph( # 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, @@ -2916,6 +3024,9 @@ def _execute_graph( 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 = ( diff --git a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py index 24c647d1f0..8e62059031 100644 --- a/packages/microcosm-graph/tests/test_graph_executor_source_identity.py +++ b/packages/microcosm-graph/tests/test_graph_executor_source_identity.py @@ -417,6 +417,186 @@ def rewrite() -> None: 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. # --------------------------------------------------------------------------