diff --git a/changelog.d/amend-gate-identity-change-not-a-verdict.fixed.md b/changelog.d/amend-gate-identity-change-not-a-verdict.fixed.md new file mode 100644 index 0000000000..9f994390b2 --- /dev/null +++ b/changelog.d/amend-gate-identity-change-not-a-verdict.fixed.md @@ -0,0 +1 @@ +A kernel that reports a changed identity no longer has that report filed as a gate verdict (graph amendment 29). `KernelIdentityChangedError` is new. `gates.battery@1` raises it when its binding registry, or a source file in a binding's closure, has changed since the kernel was built, or can no longer be read or parsed. It checks before it evaluates, before it returns, and when evaluation raises. The executor re-raises that error instead of filing a `fail` under the node key. The run refuses, nothing for that node is stored, and the nodes it had already completed keep their stored results. So a build edited mid-run no longer leaves a stale `fail` that later runs of the unedited registry are served. The change moves the implementation hashes, and therefore the node keys, of `gates.battery@1`, `export.prepare@1` and `transport.package@1` and of their descendants, so stores recompute those nodes once. diff --git a/docs/graph-acceptance.md b/docs/graph-acceptance.md index cea6ea1d08..fceb515d0f 100644 --- a/docs/graph-acceptance.md +++ b/docs/graph-acceptance.md @@ -685,6 +685,67 @@ lock unchanged: this validation changes no graph computation key. `decl.py` is re-locked. Adopted during the Orrery export review in #888. +29. **A kernel's report of a changed identity is never a verdict.** + Amendment 7 makes a raising gate a `fail` verdict, and the executor files + that verdict under the node key, so every later run that derives the key + is served it. Keys are derived from each kernel's `implementation_hash` + before any node runs. A kernel whose identity can move during a run would + therefore have its failure filed under the key of an identity it no + longer has. `gates.battery@1` is that kernel. It binds its binding + registry and the registry's source closure, and raises once either moved + after the kernel was built. An edit to a binding while a build ran, or a + closure file rewritten during a long build, stored `fail` under the + unedited registry's key, and a later run with the correct registry on the + same store was served that `fail` (#1125 review, finding 1). + + `errors.py` gains `KernelIdentityChangedError`, a `NodeRejectedError`. A + kernel raises it when state its `implementation_hash` describes has moved + since that hash was fixed. When any kernel's `run` raises it, whatever the + kernel's role, the executor re-raises it. For a gate, that replaces + amendment 7's verdict and F4's "becomes `fail`" in this one case: the run + refuses, and nothing for the node is applied or stored. The nodes the run + completed before the refusal keep their stored results, as after any + refused run whose sources are unchanged (`_settle_failed_run`). Their keys + still name what computed them, so the next run reuses them and computes + the refused node afresh. Every other gate exception is still a `fail` + verdict, filed and reused. + + The executor does not re-derive a kernel's hash after `run` to detect a + move itself. Most gate identities in this codebase are source hashes that + re-read module files on every call. A file rewritten under a running + process does not change the code that process already imported, so a + re-derived hash moves without the computation changing. Comparing against + the keyed hash would then refuse a correct build at the first gate whose + hashed files a save, checkout or pull touched after the keys were + derived, however long before that gate ran. Only the + kernel knows whether its identity is state that can move in memory. A + kernel whose identity can move in process must therefore check it in `run` + and raise `KernelIdentityChangedError`. `gates.battery@1` does so in three + places: + + - before it evaluates; + - before it returns; + - when evaluation raises, chained to that exception. + + A closure file that can no longer be read or parsed counts as a change. A + move that is undone before the kernel's next check is not seen. + + Runtime-only. `decl.py` and `kernel.py` are untouched and the interface + lock is unchanged. Three implementation hashes move: + + - `gates.battery@1`, whose own source changed. A binding whose source + closure reaches `microcosm.graph` also binds the edited executor and + errors modules. + - `export.prepare@1` and `transport.package@1`, whose hashes include the + battery's source. + + Their node keys move, and so do their descendants'. No other kernel's + hash reads a file this change edits; all 41 `implementation_hash` + definitions in `packages/*/src` were checked. A `fail` filed by an earlier + version under a stale identity therefore sits under a key that no run of + this version derives, and is never served. Raised by the independent + review of #1125 (finding 1, Subfleet job 20261007-081708); adopted with the + fix on 2026-10-07. Adding a normative field with a default changes the canonical projection of every node that carries it, so node keys moved with amendments 11 and diff --git a/docs/transport-graph-kernels.md b/docs/transport-graph-kernels.md index 56f6e36f71..68d0ec054c 100644 --- a/docs/transport-graph-kernels.md +++ b/docs/transport-graph-kernels.md @@ -141,9 +141,35 @@ source and resolved case-exactly with the import system's path finder, without executing anything. Closures, partials, bound methods, lambdas, nested functions, decorated wrappers, builtins, numpy scalars and other objects are refused when the kernel is constructed. The identity is fixed at -construction; if a binding or a module in its source closure changes -afterwards (a `dict` field edited in place, a source file rewritten), the -kernel refuses to run. +construction. If a binding or a module in its source closure changes +afterwards (a `dict` field edited in place, a source file rewritten, deleted +or left unparseable), the kernel raises `KernelIdentityChangedError` from +`implementation_hash()` and from `run()`. `run()` checks before it evaluates, +before it returns, and when evaluation raises. The run then refuses with that +error: + +- A change made before `run_graph` derives its node keys refuses the run + before anything is written. +- A change made after the keys are derived is refused at the gate node. That + covers an edit to a binding while the build runs, and a closure file + rewritten during a long build. The executor never files a kernel's + `KernelIdentityChangedError` as a `fail` verdict (graph amendment 29), and + nothing for the gate node is stored. Nodes the run completed earlier keep + their stored results, so a later run with the unedited registry on the same + store reuses them and computes the gate afresh. Before amendment 29 that + later run was served a `fail` stored under the unedited binding's key. + +The executor does not re-derive the hash itself, because it cannot tell this +kernel's in-memory bindings from the module files other gate kernels hash, and +a file rewritten under a running process does not change code that process has +already imported. The check therefore belongs to the kernel. A change made +during evaluation and undone before the next check is not seen. + +The change that introduced amendment 29 moved every `gates.battery@1` node +key: it edited this kernel's own source, and `export.prepare@1` and +`transport.package@1` hash that source as well. A `fail` that an earlier +version filed under a stale identity is therefore under a key that no later +run derives, so it is never served. The closure is honest rather than small: `microcosm.build.gates` imports `us_runtime` lazily inside one function, so a binding over a gate in that diff --git a/packages/microcosm-build/src/microcosm/build/transport/gate_kernels.py b/packages/microcosm-build/src/microcosm/build/transport/gate_kernels.py index 3cc0498479..2f56372f88 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/gate_kernels.py +++ b/packages/microcosm-build/src/microcosm/build/transport/gate_kernels.py @@ -37,9 +37,15 @@ and undecorated top-level functions without closures), and the hash covers its fields, its functions' defaults and the source of every first-party module its code can import, read from the import statements in the source. A binding -outside that vocabulary is refused when the kernel is built; a binding, or a -module in its source closure, that changes afterwards makes the kernel refuse -to run. Data a binding's modules read from disk is not bound, so a binding +outside that vocabulary is refused when the kernel is built. A binding, or a +module in its source closure, that changes afterwards (or can no longer be +read or parsed) makes the kernel raise +:class:`~microcosm.graph.KernelIdentityChangedError`: from +``implementation_hash``, and from ``run``, which checks before it evaluates, +before it returns, and when evaluation raises. The executor never files that +error as a ``fail`` verdict (graph amendment 29), so a change made while a +build runs leaves no record for the refused gate node under the unchanged +binding's key. Data a binding's modules read from disk is not bound, so a binding takes its values through the manifest's parameters or evidence artifacts. A declared gate with no binding still resolves to a named ``evidence_absent`` gap, exactly as the battery does. @@ -110,6 +116,7 @@ Determinism, KernelBase, KernelContext, + KernelIdentityChangedError, KernelRegistry, KernelResult, KernelRole, @@ -705,12 +712,34 @@ def _binding_identity(self) -> tuple[dict[str, object], dict[str, str]]: sources.update(digests) return descriptions, dict(sorted(sources.items())) + def _require_identity(self, cause: Exception | None = None) -> None: + """Refuse once a binding or its source closure moved since construction. + + Raised during key derivation, this refuses the run before anything is + written; raised from ``run``, the executor refuses the node instead of + filing an outcome under the key of the unchanged registry (graph + amendment 29). A closure file that can no longer be read or parsed is + a change too. ``cause`` is the exception evaluation raised, if any. + """ + + try: + moved = self._binding_identity() != self._identity + except Exception as error: + raise KernelIdentityChangedError( + f"{_REF} binding registry, or a module in its source closure, " + "changed after the kernel was built and can no longer be " + f"described ({type(error).__name__}: {error}); its node keys " + "would no longer describe it." + ) from error + if moved: + raise KernelIdentityChangedError( + f"{_REF} binding registry, or a module in its source closure, " + "changed after the kernel was built; its node keys would no " + "longer describe it." + ) from cause + def implementation_hash(self) -> str: - if self._binding_identity() != self._identity: - raise ValueError( - f"{_REF} binding registry changed after the kernel was built; its " - "node keys would no longer describe it." - ) + self._require_identity() descriptions, sources = self._identity code = source_hash( sys.modules[__name__], @@ -801,11 +830,19 @@ def _upstream( def run(self, context: KernelContext) -> KernelResult: require_params(context, _REF, required=self._required, optional=self._optional) require_outputs(context, _REF) - if self._binding_identity() != self._identity: - raise ValueError( - f"{_REF} binding registry changed after the kernel was built; its " - "node keys would no longer describe it." - ) + # The identity is checked on both sides of evaluation, and again when + # evaluation raises: an outcome or failure produced while it moved is + # not one the node key names, so it must never be filed as a verdict. + self._require_identity() + try: + result = self._evaluate(context) + except Exception as error: + self._require_identity(cause=error) + raise + self._require_identity() + return result + + def _evaluate(self, context: KernelContext) -> KernelResult: country = string_param(context, _REF, "country") phase = string_param(context, _REF, "phase") gates_sha256 = sha256_param(context, _REF, "gates_sha256") diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_kernels.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_kernels.py index 02c4550a9b..f8a71f67b4 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_kernels.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_gate_kernels.py @@ -71,6 +71,7 @@ ContentStore, Kernel, KernelContext, + KernelIdentityChangedError, KernelRegistry, KernelRole, Node, @@ -386,7 +387,9 @@ def test_a_binding_mutated_after_construction_refuses_to_run() -> None: kernel = GateBatteryKernel({**TOY_BINDINGS, "per_family_fit": binding}) kernel.run(_context(_node())) arguments["diagnostics"] = "other_alias" - with pytest.raises(ValueError, match="changed after the kernel was built"): + with pytest.raises( + KernelIdentityChangedError, match="changed after the kernel was built" + ): kernel.run(_context(_node())) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_graph.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_graph.py index 5c6a0e3c3f..bec6e5e3d4 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_graph.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_graph.py @@ -16,6 +16,11 @@ - A6 input identity: renaming source paths changes nothing; changing one source's bytes re-keys exactly that source's consumers and their descendants. +- Gate identity: a binding edited, or a source file in a binding's closure + rewritten, after the run derived its keys leaves nothing in the store for + the gate (graph amendment 29), so a later run with the unedited registry on + that store computes the gate instead of being served a stale ``fail`` -- + the three-run probe of the #1125 review, finding 1. - Differential: ``calibrate.ordered_adam@1`` over the compiled problem and ``calibrate.adam@1`` over the same targets as params install byte-identical weights. This holds because both paths build their @@ -35,8 +40,10 @@ import ast import functools +import importlib import re import shutil +import sys from dataclasses import replace from pathlib import Path @@ -45,6 +52,7 @@ from hypothesis import HealthCheck, given, settings from hypothesis import strategies as st +from microcosm.build.transport.gate_kernels import GateBatteryKernel from microcosm.build.transport.target_kernels import decode_target_surface from microcosm.calibrate.kernels import CALIBRATE_ADAM from microcosm.frame import WeightKind, Weights @@ -55,6 +63,7 @@ Graph, KernelBase, KernelContext, + KernelIdentityChangedError, KernelResult, Node, NodeRejectedError, @@ -72,6 +81,7 @@ descendants, run_through, run_toy, + through, toy_facts, toy_frame, toy_graph, @@ -86,6 +96,9 @@ paths_for("microcosm-build").package / "src" / "microcosm" / "build" / "transport" ) CALIBRATE = "toy.calibrate" +_IDENTITY_MOVED = "binding registry, or a module in its source closure, changed" +#: The battery's own refusal, unwrapped: no other layer rephrases it. +_BATTERY_REFUSED = r"^gates\.battery@1 " + re.escape(_IDENTITY_MOVED) def _bytes_of(run) -> dict[str, dict[str, bytes]]: @@ -152,7 +165,7 @@ def test_binding_edit_after_first_run_raises_without_changing_store(tmp_path) -> before = _store_files(store) bindings["weight_ratio"].artifact_arguments["solution"] = "missing" - with pytest.raises(ValueError, match="binding registry changed"): + with pytest.raises(KernelIdentityChangedError, match=_IDENTITY_MOVED): run_through( tmp_path, sources, @@ -175,7 +188,7 @@ def test_binding_edit_before_first_run_raises_then_correct_registry_runs_fresh( before = _store_files(store) bindings["weight_ratio"].artifact_arguments["solution"] = "missing" - with pytest.raises(ValueError, match="binding registry changed"): + with pytest.raises(KernelIdentityChangedError, match=_IDENTITY_MOVED): run_through( tmp_path, sources, @@ -198,6 +211,222 @@ def test_binding_edit_before_first_run_raises_then_correct_registry_runs_fresh( assert all(not receipt.hit for receipt in correct.nodes.values()) +GATE = "toy.gates.terminal" + + +def _run_to_gate(sources, store, registry, observer=None): + """Run the gate's ancestor closure, with an optional population observer.""" + + graph = through(toy_graph(DEFAULT_CONFIG, through_prepare=True), GATE) + used = {name for node in graph.nodes for name in node.sources} + return run_graph( + compile_graph(graph), + sources={k: v for k, v in sources.mapping().items() if k in used}, + store=store, + kernels=registry, + _population_observer=observer, + ) + + +def _once_after(node_id, action): + """An observer that runs ``action`` once ``node_id`` has been admitted.""" + + def observe(seen, population) -> None: + if seen == node_id: + action() + + return observe + + +def _assert_gate_computed_fresh_then_matches_cold(tmp_path, sources, store, bindings): + """Runs 2 and 3 of the review's probe: same store, then a cold store.""" + + resumed = _run_to_gate(sources, store, toy_registry(bindings())) + gate = resumed.nodes[GATE] + assert gate.receipt["outcome"] == "pass" + assert not gate.hit + assert all(r.hit for node, r in resumed.nodes.items() if node != GATE) + cold = _run_to_gate( + sources, ContentStore(tmp_path / "cold"), toy_registry(bindings()) + ) + assert cold.nodes[GATE].receipt["outcome"] == "pass" + assert cold.nodes[GATE].key == gate.key + assert _keys(cold) == _keys(resumed) + + +def test_binding_edit_mid_run_stores_no_gate_verdict_under_the_unedited_key( + tmp_path, +) -> None: + """The #1125 review's probe: run 1 refuses, runs 2 and 3 both pass. + + Before graph amendment 29, run 1 filed ``fail`` under the unedited + binding's key and run 2, with a fresh correct registry on the same store, + was served that ``fail`` as a hit; only the cold store in run 3 passed. + """ + sources = write_toy_sources(tmp_path / "src") + store = ContentStore(tmp_path / "store") + bindings = _editable_gate_bindings() + + def edit() -> None: + bindings["weight_ratio"].artifact_arguments["solution"] = "missing" + + with pytest.raises(KernelIdentityChangedError, match=_BATTERY_REFUSED): + _run_to_gate( + sources, store, toy_registry(bindings), _once_after(CALIBRATE, edit) + ) + + _assert_gate_computed_fresh_then_matches_cold( + tmp_path, sources, store, _editable_gate_bindings + ) + + +_PROBE_SOURCE = '''"""A gate binding whose source file a test changes while a build runs.""" + +#: Called inside the gate, so a test can change this file mid-evaluation. +HOOKS = [] +#: Non-empty: report details that are not plain data, so the kernel raises +#: after evaluation. +OBJECT_DETAILS = [] + + +def probe_weight_ratio(*, solution, problem, max_ratio): + from microcosm.build.gates import GateResult + from test_support.microcosm_build.transport_graph import weight_ratio_gate + + for hook in HOOKS: + hook() + if OBJECT_DETAILS: + return GateResult("weight_ratio", True, details={"object": object()}) + return weight_ratio_gate(solution=solution, problem=problem, max_ratio=max_ratio) +''' +_UNPARSEABLE = b"def probe_weight_ratio(:\n" + + +@pytest.fixture +def probe_module(tmp_path, monkeypatch): + """A binding module on disk outside the repository, imported fresh.""" + + name = f"_gate_identity_probe_{tmp_path.name.replace('-', '_')}" + root = tmp_path / "probe" + root.mkdir() + path = root / f"{name}.py" + path.write_text(_PROBE_SOURCE, encoding="utf-8") + monkeypatch.syspath_prepend(str(root)) + module = importlib.import_module(name) + original = path.read_bytes() + try: + yield module, path + finally: + module.HOOKS.clear() + module.OBJECT_DETAILS.clear() + path.write_bytes(original) + sys.modules.pop(name, None) + + +def _probe_bindings(module): + return { + **TOY_BINDINGS, + "weight_ratio": replace( + TOY_BINDINGS["weight_ratio"], gate=module.probe_weight_ratio + ), + } + + +def _change(path, change: str, original: bytes) -> None: + if change == "rewrite": + path.write_bytes(original + b"# rewritten while the build ran\n") + elif change == "unparseable": + path.write_bytes(_UNPARSEABLE) + else: + path.unlink() + + +@pytest.mark.parametrize( + ("when", "change", "cause"), + [ + ("before_the_gate", "rewrite", None), + ("before_the_gate", "unparseable", SyntaxError), + ("before_the_gate", "delete", FileNotFoundError), + ("before_the_gate_then_undone", "rewrite", None), + ("inside_the_gate", "rewrite", None), + ("inside_the_gate_then_raises", "rewrite", ValueError), + ], +) +def test_closure_source_changed_mid_run_stores_no_gate_verdict( + tmp_path, probe_module, when, change, cause +) -> None: + """The review's likeliest trigger: a closure file changed during a build. + + ``before_the_gate`` changes the binding's module once calibration is + admitted, so the kernel's check before it evaluates refuses; a file left + unparseable or deleted is a change too. ``before_the_gate_then_undone`` + restores the file while the gate evaluates, so only that first check can + see the move: the later checks read the original bytes again. + ``inside_the_gate`` rewrites it + while the gate evaluates, after that check passed: the gate would return + ``pass``, and its check before returning refuses, so not even a stale + ``pass`` is filed. ``inside_the_gate_then_raises`` makes evaluation raise + after the rewrite, and the check on that path refuses, chained to the + exception. Each time the file is then restored and a fresh registry derives + the original key, which must not be served anything run 1 computed. + """ + module, path = probe_module + original = path.read_bytes() + + def change_file() -> None: + _change(path, change, original) + + sources = write_toy_sources(tmp_path / "src") + store = ContentStore(tmp_path / "store") + registry = toy_registry(_probe_bindings(module)) + observer = None + if when.startswith("before_the_gate"): + observer = _once_after(CALIBRATE, change_file) + if when == "before_the_gate_then_undone": + module.HOOKS.append(lambda: path.write_bytes(original)) + else: + module.HOOKS.append(change_file) + if when == "inside_the_gate_then_raises": + module.OBJECT_DETAILS.append(True) + try: + with pytest.raises( + KernelIdentityChangedError, match=_BATTERY_REFUSED + ) as refused: + _run_to_gate(sources, store, registry, observer) + finally: + module.HOOKS.clear() + module.OBJECT_DETAILS.clear() + path.write_bytes(original) + if cause is None: + assert refused.value.__cause__ is None + else: + assert isinstance(refused.value.__cause__, cause) + if cause is ValueError: + assert "not plain data" in str(refused.value.__cause__) + + _assert_gate_computed_fresh_then_matches_cold( + tmp_path, sources, store, lambda: _probe_bindings(module) + ) + + +@pytest.mark.parametrize( + ("change", "cause"), [("unparseable", SyntaxError), ("delete", FileNotFoundError)] +) +def test_a_closure_file_that_can_no_longer_be_described_is_a_changed_identity( + probe_module, change, cause +) -> None: + """At key derivation too: the kernel's hash refuses with the same type.""" + module, path = probe_module + kernel = GateBatteryKernel(_probe_bindings(module)) + kernel.implementation_hash() + _change(path, change, path.read_bytes()) + + with pytest.raises(KernelIdentityChangedError, match=_BATTERY_REFUSED) as refused: + kernel.implementation_hash() + assert isinstance(refused.value.__cause__, cause) + assert "can no longer be described" in str(refused.value) + + def test_a4_description_and_citation_move_no_key(tmp_path) -> None: sources = write_toy_sources(tmp_path / "src") plain, _ = run_through( diff --git a/packages/microcosm-graph/src/microcosm/graph/__init__.py b/packages/microcosm-graph/src/microcosm/graph/__init__.py index f266a4d2b5..31ad91e9a2 100644 --- a/packages/microcosm-graph/src/microcosm/graph/__init__.py +++ b/packages/microcosm-graph/src/microcosm/graph/__init__.py @@ -37,6 +37,7 @@ ) from .errors import ( GraphRuntimeError, + KernelIdentityChangedError, NodeRejectedError, StoreCorruptError, StoreMissError, @@ -87,6 +88,7 @@ "Kernel", "KernelBase", "KernelContext", + "KernelIdentityChangedError", "KernelRegistry", "KernelResult", "KernelRole", diff --git a/packages/microcosm-graph/src/microcosm/graph/errors.py b/packages/microcosm-graph/src/microcosm/graph/errors.py index 297f1b5070..1ef89a61ff 100644 --- a/packages/microcosm-graph/src/microcosm/graph/errors.py +++ b/packages/microcosm-graph/src/microcosm/graph/errors.py @@ -5,13 +5,16 @@ ``NodeRejectedError``; an artifact whose bytes do not match its key is ``StoreCorruptError``; a codec or dependency missing at load time is ``StoreUnavailableError`` (never a rebuild); a node with no store hit under -``resume="require"`` is ``StoreMissError``. +``resume="require"`` is ``StoreMissError``; a kernel that is no longer the +computation its node key names is ``KernelIdentityChangedError``, a +``NodeRejectedError`` the executor never files as a gate verdict. """ from __future__ import annotations __all__ = [ "GraphRuntimeError", + "KernelIdentityChangedError", "NodeRejectedError", "StoreCorruptError", "StoreMissError", @@ -27,6 +30,21 @@ class NodeRejectedError(GraphRuntimeError): """A kernel's result violated its node's declaration; nothing was applied.""" +class KernelIdentityChangedError(NodeRejectedError): + """A kernel is no longer the computation its implementation hash named. + + A kernel raises it when state its ``implementation_hash`` describes moved + after that hash was fixed, so the node keys derived from it misname the + kernel. The executor never files it as a gate verdict, unlike any other + gate exception (amendment 7): the run refuses and nothing for the node is + stored (amendment 29), because a verdict filed under the key would be + served to every later run of the unchanged kernel. The executor cannot + detect the move itself, since a hash re-read from disk also moves when a + file changes under code already imported; a kernel whose identity can move + in process must check it in ``run`` and raise this. + """ + + class StoreCorruptError(GraphRuntimeError): """A stored artifact or manifest does not match the key it is filed under.""" diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 4a94fc1ef1..aad7ee4ab1 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -41,7 +41,7 @@ Ownership, StructuralDelta, ) -from .errors import NodeRejectedError +from .errors import KernelIdentityChangedError, NodeRejectedError from .expansion import materialized_expand_coordinates from .kernel import ( ArtifactValue, @@ -2900,6 +2900,13 @@ def _execute_graph( before = _context_digest(context) try: result = kernel.run(context) + except KernelIdentityChangedError: + # The kernel reports that it is no longer the computation its + # key names. A verdict filed under that key would be served to + # every later run of the unchanged kernel, so it is never one: + # the run refuses and nothing for this node is stored + # (amendment 29), whatever the kernel's role. + raise except Exception as error: if kernel.capabilities.role is KernelRole.GATE: result = _failed_gate_result(node, incumbent, error) diff --git a/packages/microcosm-graph/tests/engine_free/shared/test_graph_executor_gate_identity.py b/packages/microcosm-graph/tests/engine_free/shared/test_graph_executor_gate_identity.py new file mode 100644 index 0000000000..1557fb99b8 --- /dev/null +++ b/packages/microcosm-graph/tests/engine_free/shared/test_graph_executor_gate_identity.py @@ -0,0 +1,567 @@ +"""A kernel's report of a changed identity is never filed as a gate verdict. + +Amendment 7 turns a raising gate into a ``fail`` verdict, and the executor +files that verdict under the node key, so every later run that derives the +key is served it. Keys are derived from each kernel's ``implementation_hash`` +before any node runs. The review of #1125 (finding 1) found a kernel whose +identity can move during a run: ``gates.battery@1`` raises once its binding +registry, or a source file in a binding's closure, changed after the kernel +was built. So an edit made while a build ran stored ``fail`` under the +unedited registry's key, and a later run with the correct registry on that +store was served the ``fail``. + +Amendment 29: the kernel raises ``KernelIdentityChangedError``, and the +executor never files it as a verdict. The run refuses, nothing for the node is +stored, and the nodes it already completed keep their stored results, since +their keys still name what computed them. The executor does not re-derive a +hash itself: most gate hashes re-read module files, which move when a file is +rewritten under a process whose imported code did not change, so only the +kernel can tell a real move from that. + +Invariants, each executed below: + +- **No stale outcome.** After any first run, a run with the unchanged kernel + on the same store gets the same gate outcome as a run on a cold store (a + differential between the shared store and a fresh one), for a kernel whose + identity is in-memory state it checks and for one whose hash only re-reads + files, under ``resume="auto"`` and ``"forbid"``. +- **Refusal is exactly a reported move.** The first run refuses if and only + if the kernel reports that its identity moved. A hash that moves without a + report refuses nothing. Otherwise the outcome is the one the unchanged + kernel produces (amendment 7 untouched), and it is filed and reused. +- **A refusal stores nothing for the gate** and keeps every earlier node's + record, so the next run reuses that work, computes the gate afresh, and + then computes what follows it. + +Synthetic toy graph only: no data file, no country package, no engine. +""" + +from __future__ import annotations + +import hashlib +import tempfile +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest +from hypothesis import HealthCheck, example, given, settings +from hypothesis import strategies as st + +import microcosm.graph.executor as graph_executor +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from microcosm.graph import KernelIdentityChangedError, NodeRejectedError +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, + KernelRole, +) +from microcosm.graph.manifest import RunManifest +from microcosm.graph.population import Population +from microcosm.graph.store import ContentStore, StoreMiss + +SOURCE = SourceRef("survey", "csv-tables") +CREATE = Node( + "survey", + "source@1", + structural=StructuralDelta.CREATE, + sources=("survey",), + outputs=(Owned("person", "age", "int64"), Owned("household", "size", "int64")), +) +GATE = Node( + "gate", + "gate@1", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("household", "gate_verdict", "string"),), + population="survey", +) +AFTER = Node( + "after", + "after@1", + inputs=(Slice("household", ("gate_verdict",)),), + outputs=(Owned("household", "gate_copy", "string"),), + population="survey", +) +GRAPH = Graph("toy", (SOURCE,), (CREATE, GATE, AFTER)) +COMPILED = compile_graph(GRAPH) + +#: What a gate does: return a verdict, or raise an ordinary exception, which +#: amendment 7 makes a ``fail`` verdict. +BEHAVIOURS = ("pass", "fail", "raise") +#: When the gate's identity moves: never, from the population observer before +#: the gate runs (the review's mid-run binding edit), or inside the gate's own +#: ``run`` after it started evaluating. +MOVES = ("none", "observer", "during") +#: ``memory``: the identity is state the kernel runs with and checks, as +#: ``gates.battery@1``'s bindings are. ``disk``: only the hash moves, as a +#: source hash does when a file is rewritten under code already imported. +KINDS = ("memory", "disk") +BASE = "base" +GATE_MESSAGE = "gate evidence unavailable" +MOVED_MESSAGE = "gate@1 is no longer the kernel it was built as" + + +class _Kernel: + """A plain kernel with a fixed identity that counts its own executions.""" + + def __init__( + self, + ref: str, + capabilities: Capabilities, + compute: Callable[[KernelContext], KernelResult], + ) -> None: + self.ref = ref + self.capabilities = capabilities + self.compute = compute + self.calls = 0 + + def implementation_hash(self) -> str: + return hashlib.sha256(f"{self.ref}/{BASE}".encode()).hexdigest() + + def run(self, context: KernelContext) -> KernelResult: + self.calls += 1 + return self.compute(context) + + +class _Gate: + """A gate whose identity is a mutable attribute. + + A ``memory`` gate behaves as its current identity (``behaviours``, else + ``edited``) and keeps the battery's contract: its hash and its ``run`` + raise ``KernelIdentityChangedError`` once the identity left the one it + was built as, ``run`` checking before it evaluates, before it returns and + when evaluation raises. A ``disk`` gate always runs the behaviour it was + built with and never reports; only its hash follows the identity. + """ + + ref = "gate@1" + capabilities = Capabilities(Determinism.DETERMINISTIC, role=KernelRole.GATE) + + def __init__(self, base: str, *, kind: str = "memory", edited: str = "pass"): + self.kind = kind + self.identity = BASE + self.behaviours = {BASE: base} + self.edited = edited + self.during_run: Callable[[_Gate], None] | None = None + self.calls = 0 + + def _require_identity(self, cause: Exception | None = None) -> None: + if self.kind == "memory" and self.identity != BASE: + raise KernelIdentityChangedError(MOVED_MESSAGE) from cause + + def implementation_hash(self) -> str: + self._require_identity() + return hashlib.sha256(f"{self.ref}/{self.identity}".encode()).hexdigest() + + def run(self, context: KernelContext) -> KernelResult: + self.calls += 1 + self._require_identity() + try: + result = self._evaluate(context) + except Exception as error: + self._require_identity(cause=error) + raise + self._require_identity() + return result + + def _evaluate(self, context: KernelContext) -> KernelResult: + current = self.identity if self.kind == "memory" else BASE + behaviour = self.behaviours.get(current, self.edited) + if self.during_run is not None: + self.during_run(self) + if behaviour == "raise": + raise LookupError(GATE_MESSAGE) + ids = context.tables["household"]["household_id"] + return KernelResult( + columns={ + ("household", "gate_verdict"): pd.Series( + behaviour, index=ids, dtype="string" + ) + }, + receipt={"outcome": behaviour, "evidence": {"identity": current}}, + ) + + +class _Reports(_Gate): + """A gate that reports a changed identity its hash cannot see.""" + + def run(self, context: KernelContext) -> KernelResult: + self.calls += 1 + raise KernelIdentityChangedError(MOVED_MESSAGE) + + +def _source(context: KernelContext) -> KernelResult: + offset = int((context.sources["survey"] / "value.txt").read_text("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 KernelResult( + frame=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"), + ) + ) + + +def _after(context: KernelContext) -> KernelResult: + table = context.tables["household"] + return KernelResult( + columns={ + ("household", "gate_copy"): pd.Series( + table["gate_verdict"].array.copy(), + index=table["household_id"], + dtype="string", + ) + } + ) + + +@dataclass +class _Toy: + gate: _Gate + registry: KernelRegistry + + def calls(self) -> dict[str, int]: + return {ref: k.calls for ref, k in self.registry.as_mapping().items()} + + +def _toy(gate: _Gate, after: _Kernel | None = None) -> _Toy: + registry = KernelRegistry() + registry.register( + _Kernel( + "source@1", + Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE), + _source, + ) + ) + registry.register(gate) + registry.register( + after or _Kernel("after@1", Capabilities(Determinism.DETERMINISTIC), _after) + ) + return _Toy(gate, registry) + + +def _source_dir(root: Path) -> Path: + root.mkdir(parents=True) + (root / "value.txt").write_text("0", encoding="utf-8") + return root + + +def _run( + toy: _Toy, + source: Path, + store: ContentStore, + observer: Callable[[str, Population], None] | None = None, + *, + resume: str = "auto", +) -> RunManifest: + return run_graph( + COMPILED, + sources={"survey": source}, + store=store, + kernels=toy.registry, + resume=resume, # type: ignore[arg-type] + _population_observer=observer, + ) + + +def _move(gate: _Gate, identity: str) -> None: + gate.identity = identity + + +def _move_before_gate(gate: _Gate, identity: str) -> Callable[[str, Population], None]: + """An observer that edits the gate's identity once ``survey`` is admitted.""" + + def observe(node_id: str, population: Population) -> None: + if node_id == CREATE.id: + _move(gate, identity) + + return observe + + +def _record_stored(store: ContentStore, key: str) -> bool: + return store.has(graph_executor._cache_record_key(key)) + + +def _clean_keys(tmp_path: Path, source: Path) -> dict[str, str]: + manifest = _run(_toy(_Gate("pass")), source, ContentStore(tmp_path / "keys")) + return {node_id: receipt.key for node_id, receipt in manifest.nodes.items()} + + +def test_the_reviewers_probe_a_mid_run_edit_stores_no_verdict(tmp_path: Path) -> None: + """Review of #1125, finding 1, on the toy graph: run 1, run 2, run 3. + + Before amendment 29, run 1 filed ``fail`` under the unedited key and run 2, + with a correct kernel on the same store, was served it as a hit; only the + cold store in run 3 passed. + """ + source = _source_dir(tmp_path / "src") + store = ContentStore(tmp_path / "store") + edited = _toy(_Gate("pass")) + + with pytest.raises(KernelIdentityChangedError, match=MOVED_MESSAGE) as refused: + _run(edited, source, store, _move_before_gate(edited.gate, "edited")) + + assert isinstance(refused.value, NodeRejectedError) + assert edited.calls() == {"source@1": 1, "gate@1": 1, "after@1": 0} + + correct = _toy(_Gate("pass")) + resumed = _run(correct, source, store) + gate = resumed.nodes[GATE.id] + assert gate.receipt["outcome"] == "pass" + assert not gate.hit + assert resumed.nodes[CREATE.id].hit + assert correct.calls() == {"source@1": 0, "gate@1": 1, "after@1": 1} + + cold = _run(_toy(_Gate("pass")), source, ContentStore(tmp_path / "cold")) + assert cold.nodes[GATE.id].receipt["outcome"] == "pass" + assert cold.nodes[GATE.id].key == gate.key + assert cold.key == resumed.key + + +def test_a_refused_gate_leaves_no_record_and_keeps_the_runs_earlier_work( + tmp_path: Path, +) -> None: + source = _source_dir(tmp_path / "src") + keys = _clean_keys(tmp_path, source) + for resume in ("auto", "forbid"): + store = ContentStore(tmp_path / f"store-{resume}") + edited = _toy(_Gate("pass")) + with pytest.raises(KernelIdentityChangedError): + _run( + edited, + source, + store, + _move_before_gate(edited.gate, "edited"), + resume=resume, + ) + + assert _record_stored(store, keys[CREATE.id]) + assert not _record_stored(store, keys[GATE.id]) + assert not _record_stored(store, keys[AFTER.id]) + with pytest.raises(StoreMiss): + _run(_toy(_Gate("pass")), source, store, resume="require") + + +def test_an_identity_moved_inside_the_gates_own_run_is_refused( + tmp_path: Path, +) -> None: + """A move after the first check is caught on the way out, pass or raise.""" + source = _source_dir(tmp_path / "src") + for behaviour in BEHAVIOURS: + store = ContentStore(tmp_path / f"store-{behaviour}") + edited = _toy(_Gate(behaviour)) + edited.gate.during_run = lambda gate: _move(gate, "edited") + with pytest.raises(KernelIdentityChangedError) as refused: + _run(edited, source, store) + if behaviour == "raise": + assert isinstance(refused.value.__cause__, LookupError) + else: + assert refused.value.__cause__ is None + + again = _toy(_Gate(behaviour)) + resumed = _run(again, source, store) + expected = "fail" if behaviour == "raise" else behaviour + assert resumed.nodes[GATE.id].receipt["outcome"] == expected + assert not resumed.nodes[GATE.id].hit + + +def test_a_kernels_report_is_never_a_verdict_whatever_its_role( + tmp_path: Path, +) -> None: + """The report is final even when the hash holds, and keeps its type.""" + source = _source_dir(tmp_path / "src") + keys = _clean_keys(tmp_path, source) + + store = ContentStore(tmp_path / "gate") + reporting = _toy(_Reports("pass")) + with pytest.raises(KernelIdentityChangedError, match=MOVED_MESSAGE): + _run(reporting, source, store) + assert reporting.calls() == {"source@1": 1, "gate@1": 1, "after@1": 0} + assert not _record_stored(store, keys[GATE.id]) + + def report(context: KernelContext) -> KernelResult: + raise KernelIdentityChangedError("after@1 is not the kernel it was built as") + + compute = _toy( + _Gate("pass"), + _Kernel("after@1", Capabilities(Determinism.DETERMINISTIC), report), + ) + store = ContentStore(tmp_path / "compute") + with pytest.raises(KernelIdentityChangedError, match="after@1 is not") as refused: + _run(compute, source, store) + # Re-raised as the kernel raised it, not wrapped in a generic rejection. + assert refused.value.__cause__ is None + assert _record_stored(store, keys[GATE.id]) + assert not _record_stored(store, keys[AFTER.id]) + + +def test_a_gate_whose_identity_holds_still_files_its_failure(tmp_path: Path) -> None: + """Amendment 7 is unchanged: a genuine failure is a verdict, filed and reused.""" + source = _source_dir(tmp_path / "src") + store = ContentStore(tmp_path / "store") + first = _run(_toy(_Gate("raise")), source, store) + gate = first.nodes[GATE.id] + assert gate.receipt["outcome"] == "fail" + assert gate.receipt["evidence"] == { + "exception_type": "LookupError", + "message": GATE_MESSAGE, + } + assert _record_stored(store, gate.key) + + again = _toy(_Gate("raise")) + replay = _run(again, source, store) + assert all(receipt.hit for receipt in replay.nodes.values()) + assert sum(again.calls().values()) == 0 + + +def test_a_hash_that_moves_without_a_report_is_not_second_guessed( + tmp_path: Path, +) -> None: + """The executor does not re-derive hashes, so a disk-hashed gate is kept. + + Most gate hashes re-read their module files. A file rewritten under a + running process moves that hash without changing the code that runs, so + the outcome is still the keyed computation's, and refusing it would abort + a correct build. The gate runs as built, files its verdict under the key + derived at run start, and a later run of the same kernel reuses it. + """ + source = _source_dir(tmp_path / "src") + store = ContentStore(tmp_path / "store") + drifted = _toy(_Gate("pass", kind="disk")) + + first = _run(drifted, source, store, _move_before_gate(drifted.gate, "rewritten")) + assert drifted.gate.implementation_hash() != first.nodes[GATE.id].kernel_impl_hash + assert first.nodes[GATE.id].receipt["outcome"] == "pass" + assert drifted.calls() == {"source@1": 1, "gate@1": 1, "after@1": 1} + + again = _toy(_Gate("pass", kind="disk")) + replay = _run(again, source, store) + assert replay.nodes[GATE.id].hit + assert replay.nodes[GATE.id].key == first.nodes[GATE.id].key + + +def _clean_outcome(behaviour: str) -> str: + """The outcome an unperturbed gate with this behaviour files.""" + return "fail" if behaviour == "raise" else behaviour + + +def _outcome(toy: _Toy, source: Path, store: ContentStore) -> RunManifest | None: + try: + return _run(toy, source, store) + except KernelIdentityChangedError: + return None + + +@settings( + max_examples=60, + deadline=None, + suppress_health_check=[HealthCheck.function_scoped_fixture], +) +@given( + kind=st.sampled_from(KINDS), + base=st.sampled_from(BEHAVIOURS), + edited=st.sampled_from(BEHAVIOURS), + move=st.sampled_from(MOVES), + identity=st.one_of(st.just(BASE), st.text(min_size=1, max_size=6)), + resume=st.sampled_from(("auto", "forbid")), +) +# The review's probe, under both resume policies. +@example( + kind="memory", base="pass", edited="raise", move="observer", identity="x", + resume="auto", +) # fmt: skip +@example( + kind="memory", base="pass", edited="raise", move="observer", identity="x", + resume="forbid", +) # fmt: skip +# Its stale-pass twin, a move inside the run, and a disk-only hash move. +@example( + kind="memory", base="fail", edited="pass", move="observer", identity="x", + resume="auto", +) # fmt: skip +@example( + kind="memory", base="raise", edited="pass", move="during", identity="x", + resume="auto", +) # fmt: skip +@example( + kind="disk", base="raise", edited="pass", move="observer", identity="x", + resume="auto", +) # fmt: skip +# An edit that leaves the identity where it was is not a move. +@example( + kind="memory", base="raise", edited="pass", move="observer", identity=BASE, + resume="auto", +) # fmt: skip +def test_the_store_never_serves_a_gate_outcome_its_key_does_not_name( + tmp_path: Path, + kind: str, + base: str, + edited: str, + move: str, + identity: str, + resume: str, +) -> None: + root = Path(tempfile.mkdtemp(dir=tmp_path)) + source = _source_dir(root / "src") + store = ContentStore(root / "store") + + first = _toy(_Gate(base, kind=kind, edited=edited)) + observer = None + if move == "observer": + observer = _move_before_gate(first.gate, identity) + elif move == "during": + first.gate.during_run = lambda gate: _move(gate, identity) + reported = kind == "memory" and move != "none" and identity != BASE + + try: + manifest: RunManifest | None = _run( + first, source, store, observer, resume=resume + ) + except KernelIdentityChangedError: + manifest = None + # Refusal is exactly a reported move. + assert (manifest is None) == reported + if manifest is not None: + assert manifest.nodes[GATE.id].receipt["outcome"] == _clean_outcome(base) + + # No stale outcome: the shared store answers as a cold store does. + shared = _toy(_Gate(base, kind=kind, edited=edited)) + served = _outcome(shared, source, store) + cold = _outcome(_toy(_Gate(base, kind=kind)), source, ContentStore(root / "c")) + assert served is not None and cold is not None + assert served.nodes[GATE.id].receipt["outcome"] == _clean_outcome(base) + assert served.nodes[GATE.id].receipt == cold.nodes[GATE.id].receipt + assert served.key == cold.key + # A refusal stored nothing for the gate and kept the earlier work; a filed + # outcome is reused. + assert served.nodes[CREATE.id].hit + assert served.nodes[GATE.id].hit == (manifest is not None) + assert shared.calls()["gate@1"] == (0 if manifest is not None else 1)