From ff2bc2be93c71448b82bca17e94f9921188499b8 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 12 Sep 2026 08:27:52 -0400 Subject: [PATCH 1/3] Consolidate graph source codecs and exceptional replay contracts --- .../893-explain-execution-state.fixed.md | 2 + ...bservation-replay-consolidation.changed.md | 1 + docs/graph-acceptance.md | 100 ++- .../src/microcosm/graph/__init__.py | 4 + .../src/microcosm/graph/availability.py | 92 +++ .../src/microcosm/graph/codecs.py | 254 ++++++- .../src/microcosm/graph/executor.py | 374 +++++++++- .../src/microcosm/graph/explain.py | 23 +- .../src/microcosm/graph/manifest.py | 90 ++- .../src/microcosm/graph/store.py | 129 +++- .../tests/test_frame_metadata_store.py | 198 +++++ .../tests/test_graph_codecs.py | 196 ++++- .../tests/test_graph_executor.py | 676 +++++++++++++++++- .../tests/test_graph_explain.py | 22 + 14 files changed, 2082 insertions(+), 79 deletions(-) create mode 100644 changelog.d/893-explain-execution-state.fixed.md create mode 100644 changelog.d/graph-observation-replay-consolidation.changed.md create mode 100644 packages/microcosm-graph/src/microcosm/graph/availability.py create mode 100644 packages/microcosm-graph/tests/test_frame_metadata_store.py diff --git a/changelog.d/893-explain-execution-state.fixed.md b/changelog.d/893-explain-execution-state.fixed.md new file mode 100644 index 000000000..87341666c --- /dev/null +++ b/changelog.d/893-explain-execution-state.fixed.md @@ -0,0 +1,2 @@ +Show unreached computations and gate exceptions independently from cache hits +in the offline graph explanation, including cached unavailable artifacts. diff --git a/changelog.d/graph-observation-replay-consolidation.changed.md b/changelog.d/graph-observation-replay-consolidation.changed.md new file mode 100644 index 000000000..a4aad2cef --- /dev/null +++ b/changelog.d/graph-observation-replay-consolidation.changed.md @@ -0,0 +1 @@ +Add explicit raw-byte source codecs, preserve complete Frame metadata in the content store, represent unavailable gate artifacts and unreached consumers independently of cache hits, and isolate population observers from execution and persistence. diff --git a/docs/graph-acceptance.md b/docs/graph-acceptance.md index 64b59bfde..c78073571 100644 --- a/docs/graph-acceptance.md +++ b/docs/graph-acceptance.md @@ -158,7 +158,9 @@ recorded in `docs/graph-interface.lock` at the start of parallel work. Changing either file requires the owner's sign-off on the pull request and re-recording the lock. Everything else moves freely. -Amendments so far (each re-locked): +Amendments to the contract are numbered below. Changes to the two frozen +interface files are re-locked; runtime-only amendments leave their existing +lock unchanged: 1. **Structural kernels return data; the executor does the structural work.** Only `CREATE` returns `KernelResult.frame`. `FILTER` returns the @@ -414,6 +416,102 @@ Amendments so far (each re-locked): silently select the first class. Raised by the US launch integration branch, which carried the code without an amendment; adopted 2026-09-11. + +21. **Raw sources have an explicit byte codec.** A lookup, crosswalk or fitted + input may be a file rather than a population. `SourceCodecRegistry` + therefore has separate Frame and byte registration/loading methods under + one codec-name namespace. Registering a name in both modes is refused; + loading through the wrong mode raises `TypeError`. Existing Frame + enumeration and mapping methods remain Frame-only, and byte codecs have + their own equivalents. `get` resolves availability in either mode, so a + missing codec or import dependency still raises fatal `StoreUnavailable` + before execution instead of authorizing recomputation (E2). + + The built-in `raw-bytes-v1` reads one regular file into immutable bytes, + bounded at 64 MiB. Its descriptor is opened without blocking and checked + for regular-file type; directories, pipes, devices and oversized files + are refused. The consuming kernel owns payload parsing and validation. + This codec does not turn a lookup into a Frame or a source receipt into + survey data. Existing executor source-content identity and post-run + mutation checks still apply. Source declarations and canonical projections + do not change. A future country import must ship its declared resource + files with its first consumer; this generic codec supplies no country + resources. Extracted from the US launch integration on 2026-09-12. + +22. **Frame metadata survives the content store.** Frame format + `microcosm-graph-frame-v2` persists the complete recursively frozen + metadata alongside entity/link tables, schema, strata, typed weights and + the Frame mass log. Tagged metadata preserves mapping order, tuples, + frozensets, scalar kinds and binary float values, including signed zero + and supported non-finite values. The metadata payload has a separately + recorded SHA-256, checked on load; writing the same frame key with + different metadata is corruption, including concurrent write collisions. + + A stored v1 Frame is `StoreUnavailable`, never a metadata-empty substitute + or an automatic cache miss. Malformed metadata is `StoreCorrupt`. + Non-finite JSON literals and overflowing JSON numbers are refused at the + JSON decode boundary; supported non-finite *metadata floats* use tagged + binary encodings and remain valid. This is a complete Frame persistence + contract, not a new source authority or a release verdict. Extracted from + the US launch integration on 2026-09-12. + +23. **Unavailable artifacts have executor-owned outcomes.** This amendment + supersedes amendment 19's interim refusal of gate kernels with declared + typed outputs. Such a gate may now produce verified bytes normally. If + the kernel raises, amendment 7 still applies: the executor records its + failed verdict and exception evidence. It additionally records + `microcosm.graph.execution.v1`, state `gate_exception`, with the exact + declared artifact names that were not produced. No replacement bytes or + successful computation are invented. An ordinary returned result that + omits a declared output is still rejected. + + A node requiring an unavailable artifact is `unreached`; this propagates + through causal predecessors, including version/base, cell and byte + dependencies. Its kernel does not run and it has no columns, Frame, + weights, opaque bytes or observable population. Its receipt names each + direct blocker by node id and key. A failed gate's existing verdict + columns remain available, so an independent branch or a reader of that + verdict can still run. An unreached release remains evidence-tier and + cannot certify a file. + + Execution status and cache status are independent. Exceptional node + records use schema 3 and manifests containing them use schema 4. A valid + cached exceptional record is a hit only with the same blocker provenance; + strict required replay validates that provenance before any kernel runs. + Missing cache records remain misses, unavailable codecs remain fatal, + and corrupt, fabricated or inconsistent execution evidence is refused. + Manifests validate unavailable typed inputs and blocker identities on + load. Ordinary typed records/manifests retain their previous schemas. + + Only the executor may author the exact tagged execution schema. Kernel + attempts to return it are rejected; older free-form `execution` + diagnostics without that schema remain uninterpreted. The graph view + displays exceptional execution outcomes separately from hit/miss state. + No fields in `Node`, `KernelContext`, `KernelResult` or their canonical + declarations change. Extracted from the US launch integration on + 2026-09-12. + +24. **Population observers cannot change computation.** The private executor + observer receives a detached snapshot of each admitted Population, on + cold execution and restored cache hits, before the current node is + persisted. Entity/link tables, object-cell leaves, pandas attributes and + axis/category buffers, strata, schema/link records, weights, design + anchors, owners, metadata and both mass ledgers are detached. The + observer may retain or mutate its snapshot during later callbacks or + after return without changing downstream inputs or stored outputs. + Its exception still refuses the run. An unreached node has no snapshot; + an absent observer allocates none. + + The snapshot is an observation seam for an integrating verifier, not a + kernel capability, authority-bearing receipt or input to node identity. + The implementation uses an in-memory round trip of its own pandas + objects and explicit record reconstruction; it accepts no external + pickle bytes and introduces no pickle cache format. One complete copy + and a temporary serialization buffer are needed per callback, and + retained snapshots retain memory. Larger data pilots must measure that + cost before scaling. Extracted with the independently reviewed observer + isolation repair on 2026-09-12. + 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 13's sibling field `entrants`; no released artifact pins a graph key yet. diff --git a/packages/microcosm-graph/src/microcosm/graph/__init__.py b/packages/microcosm-graph/src/microcosm/graph/__init__.py index ef69b3fcc..fdc4d6be5 100644 --- a/packages/microcosm-graph/src/microcosm/graph/__init__.py +++ b/packages/microcosm-graph/src/microcosm/graph/__init__.py @@ -104,6 +104,7 @@ "NumericScope", "Tolerance", "Slice", + "SourceBytesCodec", "SourceCodec", "SourceCodecRegistry", "SourceRef", @@ -123,6 +124,7 @@ "graph_to_json", "keyed_uniform", "load_source", + "load_source_bytes", "run_graph", "source_hash", ] @@ -146,9 +148,11 @@ def _check_frame_version() -> None: from .codecs import ( # noqa: E402 - check dependency series before runtime import SOURCE_CODECS, + SourceBytesCodec, SourceCodec, SourceCodecRegistry, load_source, + load_source_bytes, ) from .executor import NodeRejected, run_graph # noqa: E402 from .explain import explain_html # noqa: E402 diff --git a/packages/microcosm-graph/src/microcosm/graph/availability.py b/packages/microcosm-graph/src/microcosm/graph/availability.py new file mode 100644 index 000000000..e89b1d90f --- /dev/null +++ b/packages/microcosm-graph/src/microcosm/graph/availability.py @@ -0,0 +1,92 @@ +"""Executor-owned exceptional outcomes, with no fabricated data products.""" + +from collections.abc import Mapping + +from .decl import StructuralDelta +from .kernel import Capabilities, KernelRole + +EXECUTION_SCHEMA = "microcosm.graph.execution.v1" + + +def has_execution(receipt: Mapping[str, object]) -> bool: + """Leave pre-existing free-form execution diagnostics uninterpreted.""" + execution = receipt.get("execution") + return ( + isinstance(execution, Mapping) and execution.get("schema") == EXECUTION_SCHEMA + ) + + +def execution_state(receipt: Mapping[str, object]) -> str | None: + execution = receipt.get("execution") + state = execution.get("state") if has_execution(receipt) else None + return state if isinstance(state, str) else None + + +def unavailable_artifacts( + receipt: Mapping[str, object], outputs: Mapping[str, object] +) -> frozenset[str]: + return ( + frozenset(outputs) + if execution_state(receipt) in {"gate_exception", "unreached"} + else frozenset() + ) + + +def validate_execution( + receipt: Mapping[str, object], + capabilities: Capabilities, + outputs: Mapping[str, object], + opaque: Mapping[str, object], + *, + has_products: bool, +) -> None: + """Validate exceptional receipt shape at cache and manifest boundaries.""" + if not has_execution(receipt): + return + execution = receipt["execution"] + if not isinstance(execution, Mapping): + raise ValueError("Executor execution metadata must be a mapping.") + state = execution.get("state") + if state == "gate_exception": + evidence = receipt.get("evidence") + missing = execution.get("unavailable_artifacts") + if ( + set(execution) != {"schema", "state", "unavailable_artifacts"} + or capabilities.role is not KernelRole.GATE + or capabilities.structural is not StructuralDelta.NONE + or receipt.get("outcome") != "fail" + or not outputs + or not isinstance(missing, list | tuple) + or tuple(missing) != tuple(sorted(outputs)) + or opaque + or not isinstance(evidence, Mapping) + or not isinstance(evidence.get("exception_type"), str) + or not isinstance(evidence.get("message"), str) + ): + raise ValueError("Invalid gate exception or unavailable artifact evidence.") + elif state == "unreached": + blockers = execution.get("blocked_by") + if ( + set(execution) != {"schema", "state", "blocked_by"} + or receipt.get("outcome") != "unreached" + or not isinstance(blockers, Mapping) + or not blockers + or any( + not isinstance(name, str) + or not name + or not isinstance(key, str) + or len(key) != 64 + or any(char not in "0123456789abcdef" for char in key) + for name, key in blockers.items() + ) + or has_products + or opaque + ): + raise ValueError("Invalid unreached blocker evidence or invented products.") + if ( + capabilities.role is KernelRole.RELEASE + and receipt.get("tier") != "evidence" + ): + raise ValueError("An unreached release must remain evidence-tier.") + else: + raise ValueError(f"Unknown executor execution state {state!r}.") diff --git a/packages/microcosm-graph/src/microcosm/graph/codecs.py b/packages/microcosm-graph/src/microcosm/graph/codecs.py index 4cf4bd3e4..3ebc9d8e9 100644 --- a/packages/microcosm-graph/src/microcosm/graph/codecs.py +++ b/packages/microcosm-graph/src/microcosm/graph/codecs.py @@ -1,12 +1,32 @@ -"""Source codecs: the sole boundary from source bytes to :class:`Frame`. - -Two codecs ship with the graph runtime: - -``frame-store`` +"""Source codecs: the boundary from source bytes to a decoded input. + +A codec is registered in exactly one of two explicit modes, and a name +belongs to at most one mode: + +*Frame mode* (:data:`SourceCodec`, :meth:`SourceCodecRegistry.register`, +:meth:`SourceCodecRegistry.load`) decodes a source into a population +:class:`Frame`. A ``CREATE`` node's kernel calls it to build a population, +and so does any other node whose source really is one — a held-out reference +frame read for scoring, say. + +*Raw-byte mode* (:data:`SourceBytesCodec`, +:meth:`SourceCodecRegistry.register_bytes`, +:meth:`SourceCodecRegistry.load_bytes`) decodes a source into immutable +``bytes``. A lookup table (an NPZ of ratios, a CSV crosswalk) is not a +population, and an import kernel that turns one into a typed +:class:`~microcosm.graph.decl.ArtifactOutput` must be able to read its real +bytes without a Frame codec registered as a pretence. Neither mode can be +loaded through the other: the mismatch is a :class:`TypeError` naming the +mode the codec actually has, so no caller receives bytes where it declared a +Frame. + +Three codecs ship with the graph runtime: + +``frame-store`` (Frame) Loads a content-verified Frame from the path of a ``ContentStore`` frame object directory. -``csv-tables`` +``csv-tables`` (Frame) Loads one CSV per entity using ``schema.json``. The schema may give a ``tables`` mapping (otherwise ``.csv`` is used), global or per-entity dtype mappings, a ``strata_column``, and either (a) a weight @@ -14,11 +34,19 @@ A JSON weight entry is ``{"kind": "design", "values": [...]}`` or ``{"kind": "design", "column": "household_weight"}``; entries are keyed by entity, or may carry their own ``entity`` field. + +``raw-bytes-v1`` (raw bytes) + Reads one regular file, bounded at :data:`RAW_BYTES_MAX_BYTES`. It + interprets nothing: the consuming kernel owns the payload's format and + validates it. Identity, the pre-run content key, and the post-run + mutation check stay where they already are, in the executor. """ from __future__ import annotations import json +import os +import stat from collections.abc import Callable, Mapping from pathlib import Path from types import MappingProxyType @@ -32,39 +60,97 @@ from .store import ContentStore, StoreUnavailable __all__ = [ + "RAW_BYTES_MAX_BYTES", "SOURCE_CODECS", + "SourceBytesCodec", "SourceCodec", "SourceCodecRegistry", "load_csv_tables", "load_frame_store", + "load_raw_bytes", "load_source", + "load_source_bytes", ] type SourceCodec = Callable[..., Frame] +type SourceBytesCodec = Callable[..., bytes] + +RAW_BYTES_MAX_BYTES = 64 * 1024 * 1024 +"""The most ``raw-bytes-v1`` will read from one source file (64 MiB).""" class SourceCodecRegistry: - """Named source-to-Frame loaders.""" + """Named source loaders in two explicit modes: Frame and raw bytes. + + A name is registered in one mode only. :meth:`get` answers the + availability question the executor asks before any kernel runs, in + either mode; :meth:`load` and :meth:`load_bytes` are what hold a codec + to the mode it was registered in. + """ def __init__(self) -> None: self._loaders: dict[str, SourceCodec] = {} + self._byte_loaders: dict[str, SourceBytesCodec] = {} - def register(self, name: str, loader: SourceCodec) -> SourceCodec: - """Register and return ``loader`` under a non-empty codec name.""" - + @staticmethod + def _check_declaration(name: str, loader: object) -> None: if not isinstance(name, str) or not name: raise ValueError("Source codec names must be non-empty strings.") if not callable(loader): raise TypeError("Source codec loaders must be callable.") + + def register(self, name: str, loader: SourceCodec) -> SourceCodec: + """Register and return a Frame ``loader`` under a non-empty codec name.""" + + self._check_declaration(name, loader) + if name in self._byte_loaders: + raise ValueError( + f"Source codec {name!r} is already registered as a raw-bytes codec." + ) incumbent = self._loaders.get(name) if incumbent is not None and incumbent is not loader: raise ValueError(f"Source codec {name!r} is already registered.") self._loaders[name] = loader return loader - def get(self, name: str) -> SourceCodec: - """Resolve ``name`` or raise fatal :class:`StoreUnavailable`.""" + def register_bytes(self, name: str, loader: SourceBytesCodec) -> SourceBytesCodec: + """Register and return a raw-byte ``loader`` under a non-empty codec name. + + A raw-byte codec returns the source's bytes and claims nothing about + their meaning; it never stands in for a population. + """ + + self._check_declaration(name, loader) + if name in self._loaders: + raise ValueError( + f"Source codec {name!r} is already registered as a Frame codec." + ) + incumbent = self._byte_loaders.get(name) + if incumbent is not None and incumbent is not loader: + raise ValueError(f"Source codec {name!r} is already registered.") + self._byte_loaders[name] = loader + return loader + + def get(self, name: str) -> SourceCodec | SourceBytesCodec: + """Resolve ``name`` in either mode, or raise fatal :class:`StoreUnavailable`. + + This is availability, not decoding: a registered raw-byte codec is + installed, and resolving it here never turns it into a Frame codec. + """ + + for loaders in (self._loaders, self._byte_loaders): + try: + return loaders[name] + except KeyError: + continue + raise StoreUnavailable(f"Source codec {name!r} is not installed.") + def _frame_loader(self, name: str) -> SourceCodec: + if name in self._byte_loaders: + raise TypeError( + f"Source codec {name!r} is a raw-bytes codec; read it with " + "load_bytes. Bytes are never presented as a Frame." + ) try: return self._loaders[name] except KeyError as error: @@ -72,6 +158,19 @@ def get(self, name: str) -> SourceCodec: f"Source codec {name!r} is not installed." ) from error + def _bytes_loader(self, name: str) -> SourceBytesCodec: + if name in self._loaders: + raise TypeError( + f"Source codec {name!r} is a Frame codec; read it with load. " + "A Frame is never presented as raw bytes." + ) + try: + return self._byte_loaders[name] + except KeyError as error: + raise StoreUnavailable( + f"Source codec {name!r} is not installed." + ) from error + def load( self, name: str, @@ -79,9 +178,9 @@ def load( *, store: ContentStore | None = None, ) -> Frame: - """Decode ``path`` with ``name`` and require a Frame result.""" + """Decode ``path`` with the Frame codec ``name`` and require a Frame.""" - loader = self.get(name) + loader = self._frame_loader(name) try: frame = loader(Path(path), store=store) except StoreUnavailable: @@ -96,16 +195,66 @@ def load( ) return frame + def load_bytes( + self, + name: str, + path: Path, + *, + store: ContentStore | None = None, + ) -> bytes: + """Read ``path`` with the raw-byte codec ``name`` and require bytes. + + This decodes an external source path, unlike + :meth:`ContentStore.load_bytes`, which reads a stored object back by + its content key. + + Unavailability keeps the classification :meth:`load` gives it: an + uninstalled codec or a missing dependency is fatal + :class:`StoreUnavailable`, never a decoded value. + """ + + loader = self._bytes_loader(name) + try: + payload = loader(Path(path), store=store) + except StoreUnavailable: + raise + except ImportError as error: + raise StoreUnavailable( + f"Source codec {name!r} needs an unavailable dependency." + ) from error + if not isinstance(payload, bytes): + raise TypeError( + f"Source codec {name!r} returned {type(payload).__name__}, not bytes." + ) + return payload + def names(self) -> tuple[str, ...]: - """Registered names in canonical order.""" + """The registered Frame codec names, in canonical order. + + Each mode is enumerated by its own pair — ``names``/``as_mapping`` + here, :meth:`bytes_names`/:meth:`as_bytes_mapping` there — so a + snapshot taken through either pair stays resolvable through it. The + name space is still shared: a name in neither tuple is not therefore + free, because registering it in one mode reserves it in both. + """ return tuple(sorted(self._loaders)) + def bytes_names(self) -> tuple[str, ...]: + """The registered raw-byte codec names, in canonical order.""" + + return tuple(sorted(self._byte_loaders)) + def as_mapping(self) -> Mapping[str, SourceCodec]: - """A read-only snapshot of registered loaders.""" + """A read-only snapshot of the registered Frame loaders.""" return MappingProxyType(dict(self._loaders)) + def as_bytes_mapping(self) -> Mapping[str, SourceBytesCodec]: + """A read-only snapshot of the registered raw-byte loaders.""" + + return MappingProxyType(dict(self._byte_loaders)) + def load_frame_store(path: Path, *, store: ContentStore | None = None) -> Frame: """Load a verified frame object from its content-store directory.""" @@ -372,9 +521,68 @@ def load_csv_tables(path: Path, *, store: ContentStore | None = None) -> Frame: return Frame(tables, schema, weights, strata) +def _nonblocking_opener(path: str, flags: int) -> int: + """Open without blocking, so a FIFO in a source's place cannot hang a run.""" + + return os.open(path, flags | getattr(os, "O_NONBLOCK", 0)) + + +def load_raw_bytes(path: Path, *, store: ContentStore | None = None) -> bytes: + """Read one regular file's bytes, bounded at :data:`RAW_BYTES_MAX_BYTES`. + + The codec claims nothing about the payload: a lookup NPZ, a crosswalk + CSV, and a corrupt file are all just bytes here, and the kernel that + imports them validates their format. What this function does own is the + refusal to read something that is not one bounded regular file. + + Symlinks are followed, as the executor's ``resolve(strict=True)`` and + ``source_content_key`` already do. A directory, a FIFO, a device, or a + socket is refused: the mode is read from the descriptor this function + 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. + """ + + del store # raw bytes are self-describing; no content store is consulted + source = Path(path) + try: + handle = open(source, "rb", opener=_nonblocking_opener) + except IsADirectoryError as error: + raise ValueError( + f"Raw source at {source} is a directory; raw-bytes-v1 reads one " + "regular file." + ) from error + except OSError as error: + raise ValueError(f"Raw source at {source} is not readable: {error}") from error + with handle: + # A directory never reaches here: opening one raises IsADirectoryError + # above. Everything else that is not a regular file is refused from the + # descriptor's own mode, not from a second look at the path. + if not stat.S_ISREG(os.fstat(handle.fileno()).st_mode): + raise ValueError( + f"Raw source at {source} is not a regular file; raw-bytes-v1 " + "reads one regular file." + ) + try: + payload = handle.read(RAW_BYTES_MAX_BYTES + 1) + except OSError as error: + raise ValueError( + f"Raw source at {source} is not readable: {error}" + ) from error + if len(payload) > RAW_BYTES_MAX_BYTES: + raise ValueError( + f"Raw source at {source} is larger than the " + f"{RAW_BYTES_MAX_BYTES}-byte raw-bytes-v1 limit." + ) + return payload + + SOURCE_CODECS = SourceCodecRegistry() SOURCE_CODECS.register("frame-store", load_frame_store) SOURCE_CODECS.register("csv-tables", load_csv_tables) +SOURCE_CODECS.register_bytes("raw-bytes-v1", load_raw_bytes) def load_source( @@ -384,6 +592,18 @@ def load_source( store: ContentStore | None = None, registry: SourceCodecRegistry = SOURCE_CODECS, ) -> Frame: - """Decode one source through the selected registry.""" + """Decode one source into a Frame through the selected registry.""" return registry.load(codec, path, store=store) + + +def load_source_bytes( + codec: str, + path: Path, + *, + store: ContentStore | None = None, + registry: SourceCodecRegistry = SOURCE_CODECS, +) -> bytes: + """Read one source's bytes through the selected registry.""" + + return registry.load_bytes(codec, path, store=store) diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 10d3488cf..b61a678d4 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -4,9 +4,12 @@ import hashlib import json +import pickle import socket import time -from collections.abc import Mapping +from collections.abc import Callable, Mapping +from copy import deepcopy +from dataclasses import fields, replace from datetime import UTC, datetime from pathlib import Path from types import MappingProxyType @@ -18,6 +21,13 @@ from . import keys as graph_keys from .artifact_edges import scope_payload, typed_contracts, value_from_descriptor +from .availability import ( + EXECUTION_SCHEMA, + execution_state, + has_execution, + unavailable_artifacts, + validate_execution, +) from .canonical import canonical_json, sha256_domain from .codecs import SOURCE_CODECS, SourceCodecRegistry from .decl import ( @@ -136,6 +146,23 @@ def _failed_gate_result( columns=MappingProxyType(columns), receipt={ "outcome": "fail", + # A gate that declares typed outputs and raises produced none of + # them. The executor, not the kernel, records that state so the + # outputs' consumers are left unreached rather than the run + # aborted (amendment 7 stays true for every legal node shape). + **( + { + "execution": { + "schema": EXECUTION_SCHEMA, + "state": "gate_exception", + "unavailable_artifacts": sorted( + output.name for output in node.artifact_outputs + ), + } + } + if node.artifact_outputs + else {} + ), "evidence": { "exception_type": type(error).__name__, "message": str(error), @@ -324,6 +351,61 @@ def _freeze_frame(table: pd.DataFrame) -> pd.DataFrame: return frozen +def _observer_snapshot(population: Population) -> Population: + """Detach every observation from the executable population and its cache. + + Pandas deep copies retain object-cell referents and some immutable-by-API + axis/category buffers. An in-memory, in-band round trip copies those too; + only pandas objects from this admitted population are serialized here. + No external pickle bytes are accepted, retained, or persisted. Frame and + graph records are reconstructed explicitly to avoid reflective copy/pickle + writes to their dataclass namespaces (which source identities may seal). + + When enabled, this costs one full detached population and a temporary + serialized table buffer per callback. Observers may retain that snapshot; + changing it immediately or later cannot change a kernel input or store write. + """ + frame = population.frame + tables = {name: frame.table(name) for name in frame.entities} + tables.update({name: frame.link(name) for name in frame.links}) + tables, strata = pickle.loads(pickle.dumps((tables, frame.strata), protocol=5)) + + def copied_record(record): + return replace( + record, + **{ + field.name: deepcopy(getattr(record, field.name)) + for field in fields(record) + }, + ) + + snapshot = Frame( + tables, + replace( + frame.schema, + group_entities=deepcopy(frame.schema.group_entities), + links=tuple(copied_record(link) for link in frame.schema.links), + ), + { + entity: Weights( + frame.weights_for(entity).values, frame.weights_for(entity).kind + ) + for entity in frame.weighted_entities + }, + strata, + mass_log=tuple(copied_record(record) for record in frame.mass_log), + metadata=frame.metadata, + ) + return Population( + snapshot, + population.version, + dict(population.owners), + dict(population.weight_kind), + mass_ledger=tuple(copied_record(record) for record in population.mass_ledger), + design_weights=population.design_weights, + ) + + def _update_scalar(digest: hashlib._Hash, value: object) -> None: if value is pd.NA: payload = b"pd.NA" @@ -1146,8 +1228,9 @@ def _validate_result( if not isinstance(payload, bytes): raise NodeRejected(f"Node {node.id!r} artifact {name!r} is not bytes.") artifacts[name] = payload + unavailable = execution_state(result.receipt) == "gate_exception" for output in node.artifact_outputs: - if output.name not in artifacts: + if output.name not in artifacts and not unavailable: # A cached record that lacks a declared artifact is a miss, but # that decision belongs to `_require_record_shape`, which runs # inside the miss-to-recompute fallback. By the time a restored @@ -1158,6 +1241,20 @@ def _validate_result( f"Node {node.id!r} is missing declared artifact {output.name!r}." ) receipt = _normal_json_mapping(result.receipt, f"Node {node.id!r} receipt") + try: + validate_execution( + receipt, + kernel_capabilities, + {output.name: output for output in node.artifact_outputs}, + artifacts, + has_products=bool( + result.columns or result.frame is not None or result.weights is not None + ), + ) + except ValueError as error: + raise NodeRejected( + f"Node {node.id!r} execution evidence rejected: {error}" + ) from error if node.structural is StructuralDelta.EXPAND: if cache_hit: if not isinstance(receipt.get("expand"), dict): @@ -1496,8 +1593,13 @@ def _write_node( record: dict[str, object] = { # Schema 2 only when the node declares typed artifacts, so a graph # that predates amendment 19 keeps its schema-1 records and its - # store hits. - "schema_version": 2 if typed_artifacts else 1, + # store hits; schema 3 only for a record carrying an executor + # execution state (gate exception or unreached). + "schema_version": 3 + if execution_state(receipt) + else 2 + if typed_artifacts + else 1, **({"typed_artifacts": dict(typed_artifacts)} if typed_artifacts else {}), "node_id": node.id, "node_key": key, @@ -1550,11 +1652,34 @@ def _require_record_shape( f"Cached receipt for node {node.id!r} has fields {sorted(raw)}, " f"not {sorted(required)}." ) - if raw["schema_version"] != (2 if typed_artifacts else 1): + raw_receipt = raw["receipt"] + if not isinstance(raw_receipt, Mapping): + raise StoreCorrupt(f"Cached node {node.id!r} receipt is malformed.") + exceptional = execution_state(raw_receipt) + if raw["schema_version"] != (3 if exceptional else 2 if typed_artifacts else 1): raise StoreUnavailable( f"Cached receipt for node {node.id!r} uses unsupported schema " f"{raw['schema_version']!r}." ) + try: + validate_execution( + raw_receipt, + capabilities, + (typed_artifacts or {}).get("outputs", {}), + { + entry.get("name"): entry.get("key") + for entry in _record_entries(raw, "opaque") + }, + has_products=bool( + raw["columns"] + or raw["frame_key"] is not None + or raw["weight"] is not None + ), + ) + except ValueError as error: + raise StoreCorrupt( + f"Cached node {node.id!r} execution evidence rejected: {error}" + ) from error if typed_artifacts: if raw.get("typed_artifacts") != dict(typed_artifacts): raise StoreCorrupt( @@ -1567,6 +1692,10 @@ def _require_record_shape( raise StoreCorrupt(f"Cached node {node.id!r} repeats an opaque artifact.") actual_outputs = {entry.get("name"): entry.get("key") for entry in opaque} for output in node.artifact_outputs: + if exceptional: + # The record proves the output was never produced; its + # absence is that evidence, not a miss. + continue if output.name not in actual_outputs: raise StoreMiss( f"Cached node {node.id!r} is missing declared artifact " @@ -1603,10 +1732,7 @@ def _require_record_shape( f"Cached receipt capabilities for node {node.id!r} disagree with " "the registered kernel contract." ) - if node.structural is StructuralDelta.EXPAND: - raw_receipt = raw["receipt"] - if not isinstance(raw_receipt, Mapping): - raise StoreCorrupt(f"Cached node {node.id!r} receipt is malformed.") + if node.structural is StructuralDelta.EXPAND and exceptional != "unreached": if "expand_writes" not in raw_receipt: raise StoreMiss( f"Cached EXPAND node {node.id!r} predates expand_writes provenance." @@ -1912,6 +2038,134 @@ def _all_node_keys( return keys, implementations +def _blocked_by( + compiled: CompiledGraph, + node: Node, + keys: Mapping[str, str], + receipts: Mapping[str, Mapping[str, object]], +) -> dict[str, str]: + """The predecessors whose outputs this node cannot have, by node key. + + An unreached causal parent propagates (its version, base, cells or bytes + were never produced); a typed byte input whose producer recorded it as + unavailable blocks its consumer. Every other predecessor is available, + including a failed gate that produced its verdict column. + """ + + blocked = { + parent: keys[parent] + for parent in compiled.predecessors[node.id] + if execution_state(receipts.get(parent, {})) == "unreached" + } + for binding in node.artifact_inputs: + producer = compiled.graph.node(binding.producer) + if binding.artifact in unavailable_artifacts( + receipts.get(binding.producer, {}), + {output.name: output for output in producer.artifact_outputs}, + ): + blocked[binding.producer] = keys[binding.producer] + return dict(sorted(blocked.items())) + + +def _unreached_node( + compiled: CompiledGraph, + node: Node, + *, + blockers: Mapping[str, str], + receipts: Mapping[str, NodeReceipt], + store: ContentStore, + key: str, + implementation: str, + capabilities: Capabilities, + typed: Mapping[str, object], + resume: ResumePolicy, +) -> NodeReceipt: + """Record a proven lack of inputs without running or inventing products. + + The receipt names each blocker by node key, so a cached unreached record + is a hit only while the same inputs are unavailable for the same reason; + the record holds no columns, frame, weights or bytes. A release that is + unreached has a failed or unreached gate in its ancestry by construction + and stays evidence-tier. + """ + + receipt: dict[str, object] = { + "outcome": "unreached", + "execution": { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": dict(blockers), + }, + "evidence": {"reason": "Required graph inputs are unavailable."}, + "capabilities": _capabilities_projection(capabilities), + } + if capabilities.role is KernelRole.RELEASE: + tier, gate_ids = _release_tier(compiled, node.id, receipts) + if tier != "evidence": + raise NodeRejected( + f"Release node {node.id!r} is unreached but no ancestral gate " + "failed or was unreached." + ) + receipt.update( + tier=tier, + gate_ancestry=list(gate_ids), + requires_decisions=list(_required_decision_names(node)), + ) + hit = False + replace_stale_record = False + if resume != "forbid": + try: + record = _load_record( + store, + node, + key=key, + kernel_impl_hash=implementation, + capabilities=capabilities, + typed_artifacts=typed, + ) + if record["receipt"] != receipt: + raise StoreCorrupt( + f"Cached node {node.id!r} blocked provenance disagrees with " + "its inputs." + ) + hit = True + except StoreMiss: + replace_stale_record = store.has(_cache_record_key(key)) + if resume == "require": # defended by preflight; handles races + raise + if not hit: + record = { + "schema_version": 3, + **({"typed_artifacts": dict(typed)} if typed else {}), + "node_id": node.id, + "node_key": key, + "kernel_ref": node.kernel, + "kernel_impl_hash": implementation, + "capabilities": _capabilities_projection(capabilities), + "receipt": receipt, + "columns": [], + "frame_key": None, + "weight": None, + "opaque": [], + } + store.put_json( + _cache_record_key(key), + record, + node_key=key, + verify_existing=resume != "forbid" and not replace_stale_record, + ) + return NodeReceipt( + key=key, + hit=hit, + seed=seed(key), + kernel_ref=node.kernel, + kernel_impl_hash=implementation, + capabilities=capabilities, + receipt=receipt, + typed_artifacts=typed, + ) + + def _preflight_require( compiled: CompiledGraph, store: ContentStore, @@ -1920,6 +2174,7 @@ def _preflight_require( kernels: KernelRegistry, ) -> None: missing: list[str] = [] + receipts: dict[str, Mapping[str, object]] = {} for node_id in compiled.order: node = compiled.graph.node(node_id) try: @@ -1931,13 +2186,35 @@ def _preflight_require( capabilities=kernels.get(node.kernel).capabilities, typed_artifacts=typed_contracts(compiled, node, keys, kernels), ) - _require_tolerance_writer_receipt( - node, - record, - _input_writers(compiled, node_id), - exact=False, - ) + if any(parent not in receipts for parent in compiled.predecessors[node_id]): + # A missing parent already made this run a miss; without its + # receipt the blockers below could not be derived honestly. + missing.append(node_id) + continue + blockers = _blocked_by(compiled, node, keys, receipts) + if blockers: + if record["receipt"].get("execution") != { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": blockers, + }: + raise StoreCorrupt( + f"Cached node {node_id!r} blocked provenance disagrees " + "with its inputs." + ) + elif execution_state(record["receipt"]) == "unreached": + raise StoreCorrupt( + f"Cached node {node_id!r} has no unavailable input blocker." + ) + else: + _require_tolerance_writer_receipt( + node, + record, + _input_writers(compiled, node_id), + exact=False, + ) _preflight_record(store, record) + receipts[node_id] = record["receipt"] except StoreMiss: missing.append(node_id) if missing: @@ -1970,8 +2247,18 @@ def run_graph( kernels: KernelRegistry, resume: ResumePolicy = "auto", decisions: tuple[Decision, ...] = (), + _population_observer: Callable[[str, Population], None] | None = None, ) -> RunManifest: - """Execute a compiled graph with content-addressed reuse and receipts.""" + """Execute a compiled graph with content-addressed reuse and receipts. + + The private population observer exposes a detached snapshot of each node's + admitted population, design anchors included, to an integrating verifier. + It runs for cold execution and for restored cache hits alike, before the + node is persisted; changes to the snapshot cannot alter execution or + 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. + """ if resume not in ("auto", "require", "forbid"): raise ValueError("resume must be 'auto', 'require', or 'forbid'.") @@ -1993,26 +2280,12 @@ def run_graph( node_id: typed_contracts(compiled, compiled.graph.node(node_id), keys, kernels) for node_id in compiled.order } - for node_id in compiled.order: - node = compiled.graph.node(node_id) - # A gate whose kernel raises becomes a `fail` verdict and the run - # continues (amendment 7), so its synthesized result carries no - # artifacts. Amendment 19 has no regime for an output a node was - # unable to produce, so a gate that declares one is refused rather - # than allowed to turn a verdict into an aborted run. - if node.artifact_outputs and ( - kernels.get(node.kernel).capabilities.role is KernelRole.GATE - ): - raise NodeRejected( - f"Node {node_id!r}: a gate kernel may not declare a typed artifact " - "output, because a gate exception is a verdict and would leave the " - "output unproduced." - ) if resume == "require": _preflight_require(compiled, store, keys, implementations, kernels) populations: dict[str, Population] = {} receipts: dict[str, NodeReceipt] = {} + receipt_payloads: dict[str, Mapping[str, object]] = {} for node_id in compiled.order: node_started = time.perf_counter() node = compiled.graph.node(node_id) @@ -2025,6 +2298,23 @@ def run_graph( "capabilities." ) + blockers = _blocked_by(compiled, node, keys, receipt_payloads) + if blockers: + receipts[node_id] = _unreached_node( + compiled, + node, + blockers=blockers, + receipts=receipts, + store=store, + key=key, + implementation=implementation, + capabilities=kernel.capabilities, + typed=contracts[node_id], + resume=resume, + ) + receipt_payloads[node_id] = receipts[node_id].receipt + continue + if node.structural is StructuralDelta.CREATE: incumbent: Population | None = None elif node.structural is StructuralDelta.NONE: @@ -2084,6 +2374,10 @@ def run_graph( capabilities=kernel.capabilities, typed_artifacts=typed, ) + if execution_state(record["receipt"]) == "unreached": + raise StoreCorrupt( + f"Cached node {node_id!r} has no unavailable input blocker." + ) try: _require_tolerance_writer_receipt( node, record, input_writers, exact=True @@ -2131,6 +2425,20 @@ def run_graph( raise NodeRejected( f"Node {node.id!r} kernel {node.kernel!r} failed: {error}" ) from error + else: + # The execution state is executor evidence about what a + # kernel could not produce; a kernel that returns one is a + # contract rejection, not an exception raised while a gate + # computed its verdict. + if ( + isinstance(result, KernelResult) + and isinstance(result.receipt, Mapping) + and has_execution(result.receipt) + ): + raise NodeRejected( + f"Node {node.id!r} kernel receipt may not author executor " + "execution metadata." + ) after = _context_digest(context) if before != after: raise NodeRejected(f"Node {node.id!r} mutated its input context.") @@ -2250,6 +2558,9 @@ def run_graph( else: populations[node.id] = updated + if _population_observer is not None: + _population_observer(node_id, _observer_snapshot(updated)) + if not hit: manifest_artifacts, record = _write_node( store, @@ -2300,6 +2611,7 @@ def run_graph( weight_key=receipt_weight_key, opaque_artifacts=MappingProxyType(receipt_opaque), ) + receipt_payloads[node_id] = receipts[node_id].receipt return RunManifest( country=compiled.graph.country, diff --git a/packages/microcosm-graph/src/microcosm/graph/explain.py b/packages/microcosm-graph/src/microcosm/graph/explain.py index 26abfdbd4..d17d8be94 100644 --- a/packages/microcosm-graph/src/microcosm/graph/explain.py +++ b/packages/microcosm-graph/src/microcosm/graph/explain.py @@ -12,6 +12,7 @@ from enum import Enum from typing import TYPE_CHECKING +from .availability import execution_state from .decl import GATE_OUTCOMES, CompiledGraph, StructuralDelta from .manifest import NodeReceipt, RunManifest from .population import mass_record_receipt @@ -194,6 +195,10 @@ .graph-node.gate-unreached .node-box { stroke: var(--color-red-600); stroke-width: 3; } .graph-node.gate-pass .node-box, .graph-node.gate-not_applicable .node-box { stroke: var(--color-green-600); stroke-width: 3; } +.graph-node.execution-unreached .node-box { fill: var(--surface-muted); + stroke: var(--color-red-600); stroke-dasharray: 5 3; } +.graph-node.execution-gate_exception .node-box { fill: var(--color-red-100); + stroke: var(--color-red-600); } .graph-node:focus .node-box, .graph-node.selected .node-box { stroke-width: 4.5; } .node-title { fill: var(--text); font-size: 13px; font-weight: 760; } .node-line { fill: var(--muted); font-size: 10.5px; } @@ -308,11 +313,19 @@ def _role(receipt: NodeReceipt) -> str: def _node_status(receipt: NodeReceipt) -> tuple[str, str]: store = "hit" if receipt.hit else "miss" role = _role(receipt) - if role != "gate": - return f"status-{store}", store - outcome = str(receipt.receipt.get("outcome", "unrecorded")) - outcome_class = outcome if outcome in GATE_OUTCOMES else "unknown" - return f"status-{store} gate-{outcome_class}", f"{store} · gate {outcome}" + classes, label = f"status-{store}", store + if role == "gate": + outcome = str(receipt.receipt.get("outcome", "unrecorded")) + outcome_class = outcome if outcome in GATE_OUTCOMES else "unknown" + classes += f" gate-{outcome_class}" + label += f" · gate {outcome}" + # Cache reuse and execution are independent: a cached refusal did not run + # the downstream computation. Interpret only executor-owned state. + state = execution_state(receipt.receipt) + if state in {"unreached", "gate_exception"}: + classes += f" execution-{state}" + label += " · unreached" if state == "unreached" else " · exception" + return classes, label def _depths(compiled: CompiledGraph) -> dict[str, int]: diff --git a/packages/microcosm-graph/src/microcosm/graph/manifest.py b/packages/microcosm-graph/src/microcosm/graph/manifest.py index e61680bf2..1adb8ded8 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -14,6 +14,12 @@ from microcosm.frame import Frame from .artifact_edges import require_compatible_scope, value_from_descriptor +from .availability import ( + execution_state, + has_execution, + unavailable_artifacts, + validate_execution, +) from .canonical import canonical_json, sha256_domain from .decl import GATE_OUTCOMES, StructuralDelta from .errors import NodeRejectedError, StoreCorruptError @@ -36,6 +42,7 @@ _SCHEMA_VERSION = 2 _TYPED_SCHEMA_VERSION = 3 +_EXCEPTION_SCHEMA_VERSION = 4 _LEGACY_SCHEMA_VERSION = 1 _CERTIFYING_GATE_OUTCOMES = frozenset({"pass", "not_applicable"}) @@ -301,6 +308,21 @@ def __post_init__(self) -> None: opaque_artifacts[name] = key object.__setattr__(self, "opaque_artifacts", MappingProxyType(opaque_artifacts)) + if has_execution(self.receipt): + if self.legacy_capabilities: + raise ValueError("Legacy receipts cannot carry executor outcomes.") + validate_execution( + self.receipt, + self.capabilities, + self.typed_artifacts.get("outputs", {}), + self.opaque_artifacts, + has_products=bool( + self.artifacts + or self.frame_key is not None + or self.weight_key is not None + ), + ) + wall_time = float(self.wall_time) if not math.isfinite(wall_time) or wall_time < 0: raise ValueError("NodeReceipt.wall_time must be finite and non-negative") @@ -621,7 +643,9 @@ def to_json(self) -> str: """Serialize the complete portable provenance as canonical JSON.""" payload = { - "schema_version": _TYPED_SCHEMA_VERSION + "schema_version": _EXCEPTION_SCHEMA_VERSION + if any(execution_state(node.receipt) for node in self.nodes.values()) + else _TYPED_SCHEMA_VERSION if any(node.typed_artifacts for node in self.nodes.values()) else _SCHEMA_VERSION, "key": self.key, @@ -665,6 +689,7 @@ def from_json(cls, value: str | bytes | bytearray) -> Self: _LEGACY_SCHEMA_VERSION, _SCHEMA_VERSION, _TYPED_SCHEMA_VERSION, + _EXCEPTION_SCHEMA_VERSION, }: raise ValueError(f"unsupported manifest schema version {schema_version!r}") @@ -693,6 +718,10 @@ def from_json(cls, value: str | bytes | bytearray) -> Self: node.typed_artifacts for node in nodes.values() ): raise ValueError("Schema-v3 manifest must carry typed artifact provenance.") + if schema_version == _EXCEPTION_SCHEMA_VERSION and not any( + execution_state(node.receipt) for node in nodes.values() + ): + raise ValueError("Schema-v4 manifest must carry executor outcomes.") body = raw.get("content_addressed") if not isinstance(body, Mapping): raise ValueError("manifest content-addressed body must be an object") @@ -1127,8 +1156,17 @@ def _node_receipt_from_payload(value: object, *, schema_version: int) -> NodeRec weight_key = value.get("weight_key") opaque_artifacts = value.get("opaque_artifacts", {}) typed_artifacts = value.get("typed_artifacts", {}) - if "typed_artifacts" in value and schema_version != _TYPED_SCHEMA_VERSION: - raise ValueError("Typed artifact provenance requires manifest schema 3.") + if ( + isinstance(receipt, Mapping) + and has_execution(receipt) + and schema_version != _EXCEPTION_SCHEMA_VERSION + ): + raise ValueError("Executor exceptional outcomes require manifest schema 4.") + if "typed_artifacts" in value and schema_version not in { + _TYPED_SCHEMA_VERSION, + _EXCEPTION_SCHEMA_VERSION, + }: + raise ValueError("Typed artifact provenance requires manifest schema 3 or 4.") capabilities_payload = value.get("capabilities") if schema_version == _LEGACY_SCHEMA_VERSION: # Every schema-v1 receipt is legacy: v1 never recorded a tolerance, so @@ -1228,7 +1266,16 @@ def _validate_typed_ancestry(nodes: Mapping[str, NodeReceipt]) -> None: entry["producer"] != node_id or entry["artifact"] != name or value.producer_key != node.key - or node.opaque_artifacts.get(name) != value.key + or ( + # A gate exception recorded the output as unavailable, so + # no stored identity exists for it; every other output + # must resolve to its stored bytes. + name + not in unavailable_artifacts( + node.receipt, node.typed_artifacts["outputs"] + ) + and node.opaque_artifacts.get(name) != value.key + ) ): raise ValueError( f"Node {node_id!r} typed artifact output provenance mismatch." @@ -1268,8 +1315,43 @@ def _validate_typed_ancestry(nodes: Mapping[str, NodeReceipt]) -> None: f"Node {node_id!r} typed artifact carries a numeric scope its " f"consumer may not read: {error}" ) from error + if entry["artifact"] in unavailable_artifacts( + producer.receipt, producer.typed_artifacts.get("outputs", {}) + ) and ( + execution_state(node.receipt) != "unreached" + or producer_id not in node.receipt["execution"]["blocked_by"] + ): + raise ValueError( + f"Node {node_id!r} consumes an unavailable artifact but is not " + "unreached by its producer." + ) edges[node_id].add(producer_id) + for node_id, node in nodes.items(): + if execution_state(node.receipt) != "unreached": + continue + for parent_id, parent_key in node.receipt["execution"]["blocked_by"].items(): + parent = nodes.get(parent_id) + if parent is None or parent.key != parent_key: + raise ValueError( + f"Node {node_id!r} unreached blocker {parent_id!r} is missing or " + "has a different key." + ) + if execution_state(parent.receipt) not in {"gate_exception", "unreached"}: + raise ValueError( + f"Node {node_id!r} unreached blocker {parent_id!r} has no " + "unavailable outputs." + ) + if execution_state(parent.receipt) == "gate_exception" and not any( + entry["producer"] == parent_id + for entry in node.typed_artifacts.get("inputs", {}).values() + ): + raise ValueError( + f"Node {node_id!r} names a gate exception blocker that is not " + "one of its declared typed inputs." + ) + edges[node_id].add(parent_id) + memo: dict[str, frozenset[str]] = {} def visit(node_id: str, trail: frozenset[str]) -> frozenset[str]: diff --git a/packages/microcosm-graph/src/microcosm/graph/store.py b/packages/microcosm-graph/src/microcosm/graph/store.py index 2d943d4a4..d30050b00 100644 --- a/packages/microcosm-graph/src/microcosm/graph/store.py +++ b/packages/microcosm-graph/src/microcosm/graph/store.py @@ -62,7 +62,7 @@ type ResumePolicy = Literal["auto", "require", "forbid"] _STORE_FORMAT = "microcosm-graph-content-store-v1" -_FRAME_FORMAT = "microcosm-graph-frame-v1" +_FRAME_FORMAT = "microcosm-graph-frame-v2" _KEY = re.compile(r"[0-9a-f]{64}\Z") _ENCODING_NUMPY = "numpy-v1" @@ -194,10 +194,32 @@ def _require_key(key: str) -> str: return key +def _reject_non_finite_constant(token: str) -> float: + """Refuse the ``NaN``/``Infinity`` literals ``json`` accepts by default.""" + raise ValueError(f"Stored JSON carries the non-finite constant {token}.") + + +def _finite_json_number(token: str) -> float: + """Refuse numeric literals that overflow to an infinity (e.g. ``1e999``).""" + value = float(token) + if not math.isfinite(value): + raise ValueError(f"Stored JSON carries the non-finite number {token}.") + return value + + def _load_json_file(path: Path, *, label: str) -> Any: + # ``_canonical_json`` writes every store JSON with ``allow_nan=False``, so + # no valid payload can carry a non-finite value and the decode boundary is + # the right place to refuse one. Without these hooks a corrupt frame + # manifest survives the load and only fails later inside the canonical + # re-encode, which escapes as TypeError instead of StoreCorrupt. try: - return json.loads(path.read_text(encoding="utf-8")) - except (OSError, UnicodeDecodeError, json.JSONDecodeError) as error: + return json.loads( + path.read_text(encoding="utf-8"), + parse_constant=_reject_non_finite_constant, + parse_float=_finite_json_number, + ) + except (OSError, UnicodeDecodeError, ValueError) as error: raise StoreCorrupt(f"Stored {label} is not readable canonical JSON.") from error @@ -704,11 +726,14 @@ def _put( build: Callable[[Path], Mapping[str, object]], *, verify_existing: bool = True, + validate_existing: Callable[[Mapping[str, Any]], None] | None = None, ) -> Path: key = _require_key(key) destination = self.object_path(key) if verify_existing and destination.exists(): - _verified_meta(destination, expected_kind=kind) + existing = _verified_meta(destination, expected_kind=kind) + if validate_existing is not None: + validate_existing(existing) return destination staging = self.tmp / uuid.uuid4().hex staging.mkdir(parents=False, exist_ok=False) @@ -736,7 +761,9 @@ def _put( if verify_existing: if not destination.exists(): raise - _verified_meta(destination, expected_kind=kind) + existing = _verified_meta(destination, expected_kind=kind) + if validate_existing is not None: + validate_existing(existing) else: self._replace_write_only_collision(staging, destination) _fsync_directory(destination.parent) @@ -902,12 +929,35 @@ def put_frame( raise TypeError(f"frame must be a Frame, got {type(frame).__name__}.") frame.revalidate() bound_node_key = key if node_key is None else node_key + metadata_sha256 = hashlib.sha256( + _canonical_json(_encode_frame_metadata(frame.metadata)) + ).hexdigest() + + def validate_existing(metadata: Mapping[str, Any]) -> None: + if metadata.get("frame_format") != _FRAME_FORMAT: + raise StoreUnavailable( + "Stored frame predates complete metadata storage." + ) + if metadata.get("frame_metadata_sha256") != metadata_sha256: + raise StoreCorrupt( + "The same frame key cannot carry different metadata." + ) def build(root: Path) -> Mapping[str, object]: _write_frame(root, frame) - return {"frame_format": _FRAME_FORMAT, "node_key": bound_node_key} + return { + "frame_format": _FRAME_FORMAT, + "node_key": bound_node_key, + "frame_metadata_sha256": metadata_sha256, + } - return self._put(key, "frame", build, verify_existing=verify_existing) + return self._put( + key, + "frame", + build, + verify_existing=verify_existing, + validate_existing=validate_existing, + ) write_frame = put_frame @@ -1006,6 +1056,61 @@ def _schema_payload(schema: EntitySchema) -> dict[str, object]: } +def _encode_frame_metadata(value: object) -> object: + """Preserve Frame metadata kinds without pickle or lossy scalar coercion.""" + if isinstance(value, Mapping): + return [ + "mapping", + [[key, _encode_frame_metadata(item)] for key, item in value.items()], + ] + if isinstance(value, tuple): + return ["tuple", [_encode_frame_metadata(item) for item in value]] + if isinstance(value, frozenset): + return [ + "frozenset", + sorted( + (_encode_frame_metadata(item) for item in value), key=_canonical_json + ), + ] + if isinstance(value, float): + # Binary float bytes preserve signed zero and any allowed NaN payload. + return ["float64", struct.pack(">d", value).hex()] + if value is None or isinstance(value, (str, int, bool)): + return ["scalar", value] + raise TypeError(f"Unsupported Frame metadata value {type(value).__name__}.") + + +def _decode_frame_metadata(encoded: object) -> object: + try: + if not isinstance(encoded, list) or len(encoded) != 2: + raise ValueError + kind, value = encoded + if kind == "mapping" and isinstance(value, list): + result = {} + for pair in value: + if not isinstance(pair, list) or len(pair) != 2: + raise ValueError + key, item = pair + if not isinstance(key, str) or not key or key in result: + raise ValueError + result[key] = _decode_frame_metadata(item) + return result + if kind in ("tuple", "frozenset") and isinstance(value, list): + items = [_decode_frame_metadata(item) for item in value] + return tuple(items) if kind == "tuple" else frozenset(items) + if ( + kind == "float64" + and isinstance(value, str) + and re.fullmatch(r"[0-9a-f]{16}", value) + ): + return struct.unpack(">d", bytes.fromhex(value))[0] + if kind == "scalar" and (value is None or type(value) in (str, int, bool)): + return value + raise ValueError + except (ValueError, TypeError, KeyError, struct.error) as error: + raise StoreCorrupt("Stored Frame metadata is malformed.") from error + + def _write_frame(root: Path, frame: Frame) -> None: _write_json(root / "schema.json", _schema_payload(frame.schema)) table_specs: list[dict[str, object]] = [] @@ -1060,6 +1165,7 @@ def _write_frame(root: Path, frame: Frame) -> None: "weights": weight_specs, "strata": strata_spec, "mass_log": mass_log, + "metadata": _encode_frame_metadata(frame.metadata), }, ) @@ -1106,6 +1212,14 @@ def _read_frame(path: Path, metadata: Mapping[str, Any]) -> Frame: raw_weights = manifest.get("weights") strata_spec = manifest.get("strata") raw_mass_log = manifest.get("mass_log") + raw_metadata = manifest.get("metadata") + if hashlib.sha256(_canonical_json(raw_metadata)).hexdigest() != metadata.get( + "frame_metadata_sha256" + ): + raise StoreCorrupt("Stored Frame metadata identity is missing or incorrect.") + frame_metadata = _decode_frame_metadata(raw_metadata) + if not isinstance(frame_metadata, Mapping): + raise StoreCorrupt("Stored Frame metadata must be a mapping.") if not all( ( isinstance(raw_tables, list), @@ -1245,6 +1359,7 @@ def _read_frame(path: Path, metadata: Mapping[str, Any]) -> Frame: weights, strata, mass_log=tuple(mass_log), + metadata=frame_metadata, ) except ImportError as error: raise StoreUnavailable( diff --git a/packages/microcosm-graph/tests/test_frame_metadata_store.py b/packages/microcosm-graph/tests/test_frame_metadata_store.py new file mode 100644 index 000000000..917119d77 --- /dev/null +++ b/packages/microcosm-graph/tests/test_frame_metadata_store.py @@ -0,0 +1,198 @@ +"""Invented complete Frame metadata persistence and stale codec refusal.""" + +from __future__ import annotations + +import hashlib +import json +import math +import struct + +import numpy as np +import pandas as pd +import pytest + +import microcosm.graph.store as store_module +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from microcosm.graph.store import ContentStore, StoreCorrupt, StoreUnavailable + + +def _frame(metadata): + return Frame( + { + "person": pd.DataFrame( + {"person_id": [1, 2], "person_household_id": [1, 1]} + ), + "household": pd.DataFrame({"household_id": [1]}), + }, + EntitySchema(group_entities=("household",)), + {"household": Weights(np.array([100.0]), WeightKind.DESIGN)}, + metadata=metadata, + ) + + +def test_roundtrip_preserves_complete_nested_source_metadata_and_types(tmp_path): + original = _frame( + { + "us_spine_assembly_manifest": { + "source_arm": "invented", + "sources": ({"sha256": "a" * 64, "rows": 2, "design_total": 100.0},), + "flags": frozenset({"native", "unknown"}), + "nullable": None, + "flag": True, + }, + "signed_zero": -0.0, + "nan": struct.unpack(">d", bytes.fromhex("7ff8000000000011"))[0], + "infinity": float("inf"), + } + ) + store = ContentStore(tmp_path / "invented-store") + store.put_frame("a" * 64, original) + restored = store.load_frame("a" * 64) + assert store_module._encode_frame_metadata( + restored.metadata + ) == store_module._encode_frame_metadata(original.metadata) + assert isinstance(restored.metadata["us_spine_assembly_manifest"]["sources"], tuple) + assert isinstance( + restored.metadata["us_spine_assembly_manifest"]["flags"], frozenset + ) + assert math.copysign(1.0, restored.metadata["signed_zero"]) == -1.0 + assert struct.pack(">d", restored.metadata["nan"]).hex() == "7ff8000000000011" + assert ( + restored.weights_for("household").values.tobytes() + == original.weights_for("household").values.tobytes() + ) + with pytest.raises(TypeError): + restored.metadata["us_spine_assembly_manifest"]["source_arm"] = "changed" + + +def test_same_key_with_changed_metadata_refuses_without_replacing_original(tmp_path): + store = ContentStore(tmp_path / "invented-store") + original = _frame({"source_sha256": "a" * 64}) + store.put_frame("b" * 64, original) + store.put_frame("b" * 64, original) + with pytest.raises(StoreCorrupt, match="different metadata"): + store.put_frame("b" * 64, _frame({"source_sha256": "b" * 64})) + assert store.load_frame("b" * 64).metadata == original.metadata + + +def test_v1_frame_is_unavailable_instead_of_silently_losing_metadata( + tmp_path, monkeypatch +): + store = ContentStore(tmp_path / "invented-store") + with monkeypatch.context() as old: + old.setattr(store_module, "_FRAME_FORMAT", "microcosm-graph-frame-v1") + store.put_frame("c" * 64, _frame({"source": "invented"})) + with pytest.raises(StoreUnavailable, match="codec"): + store.load_frame("c" * 64) + with pytest.raises(StoreUnavailable, match="predates complete metadata"): + store.put_frame("c" * 64, _frame({"source": "invented"})) + + +@pytest.mark.parametrize( + "value", + [ + ["mapping", [["duplicate", ["scalar", 1]], ["duplicate", ["scalar", 2]]]], + ["float64", "bad"], + ["scalar", 1.2], + ["mapping", [["", ["scalar", None]]]], + ["unknown", []], + ], +) +def test_malformed_metadata_refuses(value): + with pytest.raises(StoreCorrupt, match="metadata"): + store_module._decode_frame_metadata(value) + + +def _restated_manifest(store, key, *, literal=None): + """Rewrite one stored frame manifest, keeping the store's own gate honest. + + The manifest is re-registered with its true size and SHA-256 so + :func:`_verified_meta` still runs and still passes; the refusal under test + therefore comes from the decode boundary, not from a skipped checksum. + Passing ``literal=None`` restates identical content, which is the control + proving valid bytes are unaffected. + """ + object_path = store.object_path(key) + manifest_path = object_path / "frame.json" + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + if literal is not None: + manifest["metadata"] = [ + "mapping", + [["us_spine_assembly_manifest", ["float64", "@NON_FINITE@"]]], + ] + text = json.dumps(manifest, separators=(",", ":"), sort_keys=True) + manifest_path.write_text( + text.replace('"@NON_FINITE@"', literal or ""), encoding="utf-8" + ) + meta_path = object_path / "meta.json" + meta = json.loads(meta_path.read_text(encoding="utf-8")) + record = meta["payloads"]["frame.json"] + record["sha256"] = hashlib.sha256(manifest_path.read_bytes()).hexdigest() + record["size"] = manifest_path.stat().st_size + meta_path.write_text( + json.dumps(meta, separators=(",", ":"), sort_keys=True), encoding="utf-8" + ) + return manifest_path + + +@pytest.mark.parametrize( + "literal", ["NaN", "Infinity", "-Infinity", "1e999", "-1e999", "1E9999"] +) +def test_non_finite_stored_manifest_is_store_corrupt_not_type_error(tmp_path, literal): + """A checksum-consistent but non-finite manifest keeps the store taxonomy.""" + store = ContentStore(tmp_path / "invented-store") + store.put_frame("d" * 64, _frame({"source": "invented"})) + manifest_path = _restated_manifest(store, "d" * 64, literal=literal) + assert literal in manifest_path.read_text(encoding="utf-8") + with pytest.raises(StoreCorrupt, match="frame manifest") as caught: + store.load_frame("d" * 64) + # The escape this pins is a bare TypeError from the canonical re-encode. + assert type(caught.value) is StoreCorrupt + + +def test_restating_the_same_manifest_still_loads_the_identical_frame(tmp_path): + """Control: the stricter decoder does not reject any valid stored bytes.""" + store = ContentStore(tmp_path / "invented-store") + original = _frame({"source": "invented", "share": 0.25, "count": 3}) + store.put_frame("e" * 64, original) + _restated_manifest(store, "e" * 64) + restored = store.load_frame("e" * 64) + assert store_module._encode_frame_metadata( + restored.metadata + ) == store_module._encode_frame_metadata(original.metadata) + + +@pytest.mark.parametrize("literal", ["NaN", "-Infinity", "1e999"]) +def test_non_finite_object_metadata_is_store_corrupt(tmp_path, literal): + """The same boundary covers meta.json, which no payload checksum guards.""" + store = ContentStore(tmp_path / "invented-store") + store.put_frame("f" * 64, _frame({"source": "invented"})) + meta_path = store.object_path("f" * 64) / "meta.json" + meta = json.loads(meta_path.read_text(encoding="utf-8")) + meta["invented_non_finite"] = "@NON_FINITE@" + meta_path.write_text( + json.dumps(meta, separators=(",", ":"), sort_keys=True).replace( + '"@NON_FINITE@"', literal + ), + encoding="utf-8", + ) + with pytest.raises(StoreCorrupt, match="canonical JSON"): + store.load_frame("f" * 64) + + +def test_non_finite_json_never_reaches_the_canonical_encoder(tmp_path, monkeypatch): + """The refusal happens at decode, before any canonical re-encode runs.""" + store = ContentStore(tmp_path / "invented-store") + store.put_frame("0" * 64, _frame({"source": "invented"})) + _restated_manifest(store, "0" * 64, literal="NaN") + seen = [] + canonical = store_module._canonical_json + + def recorded(value): + seen.append(value) + return canonical(value) + + monkeypatch.setattr(store_module, "_canonical_json", recorded) + with pytest.raises(StoreCorrupt): + store.load_frame("0" * 64) + assert seen == [], "malformed JSON reached the canonical encoder" diff --git a/packages/microcosm-graph/tests/test_graph_codecs.py b/packages/microcosm-graph/tests/test_graph_codecs.py index 64d7c522e..1053a2eb4 100644 --- a/packages/microcosm-graph/tests/test_graph_codecs.py +++ b/packages/microcosm-graph/tests/test_graph_codecs.py @@ -3,15 +3,39 @@ from __future__ import annotations import json +import os from pathlib import Path import numpy as np import pandas as pd import pytest +import microcosm.graph.codecs as codecs_module from microcosm.frame import EntitySchema, Frame, WeightKind, Weights -from microcosm.graph.codecs import SOURCE_CODECS, SourceCodecRegistry, load_source -from microcosm.graph.store import ContentStore, StoreUnavailable +from microcosm.graph import ( + Capabilities, + ContentStore, + Determinism, + Graph, + KernelContext, + KernelRegistry, + KernelResult, + Node, + Owned, + SourceRef, + StructuralDelta, + compile_graph, + run_graph, +) +from microcosm.graph.codecs import ( + SOURCE_CODECS, + SourceCodecRegistry, + load_raw_bytes, + load_source, + load_source_bytes, +) +from microcosm.graph.keys import source_content_key +from microcosm.graph.store import StoreUnavailable def _frame() -> Frame: @@ -156,3 +180,171 @@ def missing_dependency(_path: Path, *, store: ContentStore | None = None) -> Fra registry.register("engine", missing_dependency) with pytest.raises(StoreUnavailable, match="dependency"): registry.load("engine", tmp_path) + + +# --- raw-byte codecs --------------------------------------------------------- + + +def test_raw_bytes_codec_reads_one_regular_file_verbatim(tmp_path: Path) -> None: + """``raw-bytes-v1`` ships registered in raw-byte mode and interprets nothing.""" + + payload = b"\x00lookup\xff" * 3 + source = tmp_path / "table.npz" + source.write_bytes(payload) + assert SOURCE_CODECS.bytes_names() == ("raw-bytes-v1",) + assert "raw-bytes-v1" not in SOURCE_CODECS.names() + assert SOURCE_CODECS.get("raw-bytes-v1") is load_raw_bytes + assert load_source_bytes("raw-bytes-v1", source) == payload + assert SOURCE_CODECS.load_bytes("raw-bytes-v1", source) == payload + empty = tmp_path / "empty.bin" + empty.write_bytes(b"") + assert load_source_bytes("raw-bytes-v1", empty) == b"" + # A symlink is followed, as the executor's source resolution already does. + link = tmp_path / "link.bin" + link.symlink_to(source) + assert load_source_bytes("raw-bytes-v1", link) == payload + + +def test_raw_bytes_codec_refuses_anything_but_one_bounded_regular_file( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + with pytest.raises(ValueError, match="is a directory"): + load_raw_bytes(tmp_path) + with pytest.raises(ValueError, match="not readable"): + load_raw_bytes(tmp_path / "absent.bin") + fifo = tmp_path / "pipe" + os.mkfifo(fifo) + # Opened non-blocking, so a FIFO with no writer is refused from its mode + # instead of hanging the run. + with pytest.raises(ValueError, match="not a regular file"): + load_raw_bytes(fifo) + monkeypatch.setattr(codecs_module, "RAW_BYTES_MAX_BYTES", 8) + big = tmp_path / "big.bin" + big.write_bytes(b"x" * 9) + with pytest.raises(ValueError, match="larger than the 8-byte"): + load_raw_bytes(big) + bounded = tmp_path / "bounded.bin" + bounded.write_bytes(b"x" * 8) + assert load_raw_bytes(bounded) == b"x" * 8 + + +def test_registry_modes_are_exclusive_and_names_reserved_across_modes( + tmp_path: Path, +) -> None: + registry = SourceCodecRegistry() + registry.register("frame", lambda path, *, store=None: _frame()) + registry.register_bytes("bytes", lambda path, *, store=None: b"raw") + assert registry.names() == ("frame",) + assert registry.bytes_names() == ("bytes",) + assert set(registry.as_mapping()) == {"frame"} + assert set(registry.as_bytes_mapping()) == {"bytes"} + # Availability resolves in either mode; decoding holds each to its own. + registry.get("frame") + registry.get("bytes") + source = tmp_path / "source" + source.write_bytes(b"") + assert registry.load_bytes("bytes", source) == b"raw" + assert isinstance(registry.load("frame", source), Frame) + with pytest.raises(TypeError, match="is a raw-bytes codec"): + registry.load("bytes", source) + with pytest.raises(TypeError, match="is a Frame codec"): + registry.load_bytes("frame", source) + with pytest.raises(ValueError, match="already registered as a raw-bytes codec"): + registry.register("bytes", lambda path, *, store=None: _frame()) + with pytest.raises(ValueError, match="already registered as a Frame codec"): + registry.register_bytes("frame", lambda path, *, store=None: b"") + with pytest.raises(ValueError, match="already registered"): + registry.register_bytes("bytes", lambda path, *, store=None: b"other") + with pytest.raises(ValueError, match="non-empty strings"): + registry.register_bytes("", lambda path, *, store=None: b"") + with pytest.raises(TypeError, match="must be callable"): + registry.register_bytes("thing", object()) # type: ignore[arg-type] + with pytest.raises(StoreUnavailable, match="not installed"): + registry.load_bytes("absent", source) + with pytest.raises(StoreUnavailable, match="not installed"): + registry.get("absent") + + registry.register_bytes("text", lambda path, *, store=None: "text") # type: ignore[arg-type,return-value] + with pytest.raises(TypeError, match="returned str, not bytes"): + registry.load_bytes("text", source) + + def needs_dependency(path: Path, *, store: ContentStore | None = None) -> bytes: + raise ImportError("no dependency") + + registry.register_bytes("needs", needs_dependency) + with pytest.raises(StoreUnavailable, match="unavailable dependency"): + registry.load_bytes("needs", source) + + +class _BytesCreate: + """A CREATE kernel whose only source is a raw-byte lookup table.""" + + ref = "bytes.create@1" + capabilities = Capabilities( + Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE + ) + + def implementation_hash(self) -> str: + return "0" * 64 + + def run(self, context: KernelContext) -> KernelResult: + payload = load_source_bytes("raw-bytes-v1", context.sources["table"]) + frame = _frame() + household = frame.table("household").assign( + size=pd.Series([len(payload)] * 2, dtype="int64") + ) + return KernelResult( + frame=Frame( + {"person": frame.table("person"), "household": household}, + frame.schema, + {"household": frame.weights_for("household")}, + frame.strata, + ), + receipt={"payload_bytes": len(payload)}, + ) + + +def test_a_graph_source_may_be_raw_bytes_and_keys_on_its_content( + tmp_path: Path, +) -> None: + """The executor admits a raw-byte SourceRef and hands the kernel its path. + + Identity, the pre-run content key and the post-run mutation check stay in + the executor, so the codec adds no second identity scheme. + """ + + table = tmp_path / "table.bin" + table.write_bytes(b"0123456789") + graph = Graph( + "toy", + (SourceRef("table", "raw-bytes-v1"),), + ( + Node( + "survey", + _BytesCreate.ref, + sources=("table",), + structural=StructuralDelta.CREATE, + outputs=( + Owned("person", "flag", "boolean"), + Owned("household", "size", "int64"), + ), + ), + ), + ) + registry = KernelRegistry() + registry.register(_BytesCreate()) + store = ContentStore(tmp_path / "store") + manifest = run_graph( + compile_graph(graph), sources={"table": table}, store=store, kernels=registry + ) + assert manifest.nodes["survey"].receipt["payload_bytes"] == 10 + assert set(manifest.populations["survey"].household["size"]) == {10} + cold_key = manifest.nodes["survey"].key + cold_source = source_content_key("table", table) + table.write_bytes(b"0123456789!") + assert source_content_key("table", table) != cold_source + changed = run_graph( + compile_graph(graph), sources={"table": table}, store=store, kernels=registry + ) + assert changed.nodes["survey"].key != cold_key + assert changed.nodes["survey"].receipt["payload_bytes"] == 11 diff --git a/packages/microcosm-graph/tests/test_graph_executor.py b/packages/microcosm-graph/tests/test_graph_executor.py index e6c12a8f1..6dc5ca1b6 100644 --- a/packages/microcosm-graph/tests/test_graph_executor.py +++ b/packages/microcosm-graph/tests/test_graph_executor.py @@ -15,7 +15,15 @@ import pytest import microcosm.graph.executor as graph_executor -from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from microcosm.frame import ( + EntitySchema, + Frame, + LinkSpec, + MassChangeRecord, + WeightKind, + Weights, +) +from microcosm.graph.availability import EXECUTION_SCHEMA from microcosm.graph.decl import ( ArtifactInput, ArtifactOutput, @@ -33,6 +41,7 @@ ) from microcosm.graph.errors import NodeRejectedError from microcosm.graph.executor import NodeRejected, run_graph +from microcosm.graph.explain import explain_html from microcosm.graph.kernel import ( ArtifactValue, Capabilities, @@ -47,6 +56,7 @@ ) from microcosm.graph.keys import opaque_artifact_key, platform_fingerprint from microcosm.graph.manifest import Decision, RunManifest +from microcosm.graph.population import MassRecord, Population from microcosm.graph.store import ( ContentStore, StoreCorrupt, @@ -3620,13 +3630,15 @@ def test_an_artifact_payload_enters_the_input_context_digest(tmp_path: Path) -> ) -def test_a_gate_kernel_may_not_declare_a_typed_artifact_output(tmp_path: Path) -> None: - """Amendment 19 holds amendment 7: a gate exception stays a verdict. +EVIDENCE = ArtifactType("gate.evidence", 1) + - A gate whose kernel raises produces a synthesized ``fail`` result with no - artifacts, so a declared typed output would turn that verdict into an - aborted run. Amendment 19 carries no regime for an output a node was - unable to produce, so the declaration is refused outright. +def _gate_artifact_graph(*, release_behind: bool, answer: str) -> Graph: + """A gate declaring typed evidence, its byte consumer, and a cell consumer. + + ``release`` reads the cell consumer's column when ``release_behind`` is + true (so it sits behind the byte edge) and the gate's own verdict column + otherwise (so it sits beside it). """ gate = Node( "gate", @@ -3634,22 +3646,662 @@ def test_a_gate_kernel_may_not_declare_a_typed_artifact_output(tmp_path: Path) - inputs=(Slice("person", ("age",)),), outputs=(Owned("household", "gate_verdict", "string"),), population="survey", - artifact_outputs=( - ArtifactOutput("evidence", ArtifactType("gate.evidence", 1)), + artifact_outputs=(ArtifactOutput("evidence", EVIDENCE),), + ) + use = Node( + "use", + "use@1", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("person", "used", "float64"),), + population="survey", + artifact_inputs=(ArtifactInput("evidence", "gate", "evidence", EVIDENCE),), + ) + after = Node( + "after", + "after@1", + inputs=(Slice("person", ("used",)),), + outputs=(Owned("person", "after", "float64"),), + population="survey", + ) + release = Node( + "release", + "release@1", + inputs=( + (Slice("person", ("after",)),) + if release_behind + else (Slice("household", ("gate_verdict",)),) ), + outputs=(Owned("household", "tier", "string"),), + params={"answer": answer, "requires_decisions": ()}, + population="survey", ) - graph = Graph("toy", (SOURCE,), (CREATE, gate)) + return Graph("toy", (SOURCE,), (CREATE, gate, use, after, release)) + + +def _gate_artifact_registry(*, raising: bool) -> KernelRegistry: + def failing_gate(context: KernelContext) -> KernelResult: + raise RuntimeError("evidence unavailable") + + def passing_gate(context: KernelContext) -> KernelResult: + ids = context.tables["household"]["household_id"] + return KernelResult( + columns={ + ("household", "gate_verdict"): pd.Series( + "pass", index=ids, dtype="string" + ) + }, + artifacts={"evidence": b"evidence-bytes"}, + receipt={"outcome": "pass", "evidence": {"fixture": True}}, + ) + + def use(context: KernelContext) -> KernelResult: + table = context.tables["person"] + payload = context.artifacts["evidence"].payload + return KernelResult( + columns={ + ("person", "used"): pd.Series( + np.full(len(table), float(len(payload))), + index=pd.Index(table["person_id"], name="person_id"), + dtype="float64", + ) + } + ) + + def after(context: KernelContext) -> KernelResult: + table = context.tables["person"] + return KernelResult( + columns={ + ("person", "after"): pd.Series( + table["used"].to_numpy(dtype=np.float64) * 2.0, + index=pd.Index(table["person_id"], name="person_id"), + dtype="float64", + ) + } + ) + + def release(context: KernelContext) -> KernelResult: + ids = context.tables["household"]["household_id"] + answer = str(context.params["answer"]) + return KernelResult( + columns={ + ("household", "tier"): pd.Series(answer, index=ids, dtype="string") + }, + receipt={"outcome": "pass"}, + ) + + deterministic = Capabilities(Determinism.DETERMINISTIC) registry = _registry() registry.register( _Kernel( "gate@1", Capabilities(Determinism.DETERMINISTIC, role=KernelRole.GATE), - lambda context: KernelResult(receipt={"outcome": "pass"}), + failing_gate if raising else passing_gate, + ) + ) + registry.register(_Kernel("use@1", deterministic, use)) + registry.register(_Kernel("after@1", deterministic, after)) + registry.register( + _Kernel( + "release@1", + Capabilities(Determinism.DETERMINISTIC, role=KernelRole.RELEASE), + release, ) ) + return registry + + +def test_a_gate_that_declares_evidence_and_raises_leaves_its_consumers_unreached( + tmp_path: Path, +) -> None: + """A gate exception is still a verdict (amendment 7); its outputs are absent. + + The gate records ``fail`` with a ``gate_exception`` execution state naming + the outputs it could not produce; the byte consumer and everything causally + behind it are ``unreached`` with their blockers named by node key, no + kernel behind the edge runs, nothing is invented for them, the release + behind the edge stays evidence-tier, and the manifest serializes at schema + 4, round-trips, and replays as hits under every resume policy. + """ store = ContentStore(tmp_path / "store") - with pytest.raises(NodeRejected, match="gate kernel may not declare"): + source = _source_path(tmp_path / "src") + graph = _gate_artifact_graph(release_behind=True, answer="evidence") + registry = _gate_artifact_registry(raising=True) + manifest = _run(graph, source, store, registry) + + gate = manifest.nodes["gate"] + assert gate.receipt["outcome"] == "fail" + assert gate.receipt["execution"] == { + "schema": EXECUTION_SCHEMA, + "state": "gate_exception", + "unavailable_artifacts": ("evidence",), + } + assert gate.receipt["evidence"]["exception_type"] == "RuntimeError" + assert not gate.opaque_artifacts + assert set(gate.typed_artifacts["outputs"]) == {"evidence"} + verdict = manifest.populations["survey"].household["gate_verdict"] + assert set(verdict.to_numpy()) == {"fail"} + assert "used" not in manifest.populations["survey"].person.columns + + use = manifest.nodes["use"] + assert use.receipt["outcome"] == "unreached" + assert use.receipt["execution"] == { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": {"gate": gate.key}, + } + assert not use.artifacts and use.frame_key is None and not use.opaque_artifacts + after = manifest.nodes["after"] + assert after.receipt["execution"]["blocked_by"] == {"use": use.key} + release = manifest.nodes["release"] + assert release.receipt["execution"]["blocked_by"] == {"after": after.key} + assert release.receipt["tier"] == "evidence" + assert release.receipt["gate_ancestry"] == ("gate",) + assert manifest.tier == "evidence" + calls = _calls(registry) + assert calls["gate@1"] == 1 + assert calls["use@1"] == calls["after@1"] == calls["release@1"] == 0 + + text = manifest.to_json() + assert '"schema_version":4' in text + restored = RunManifest.from_json(text) + assert restored.key == manifest.key + assert restored.nodes["use"].receipt == use.receipt + assert restored.tier == "evidence" + + def assert_explained(outcome: RunManifest, cache: str) -> None: + rendered = explain_html(compile_graph(graph), outcome) + for node_id in ("use", "after", "release"): + role = "release" if node_id == "release" else "compute" + assert ( + f'aria-label="{node_id}; {node_id}@1; {role}; none; ' + f'{cache} · unreached"' + ) in rendered + assert rendered.count('execution-unreached" data-node-detail=') == 3 + assert f'status-{cache} gate-fail execution-gate_exception"' in rendered + assert f"{cache} · gate fail · exception" in rendered + + assert_explained(manifest, "miss") + + for resume in ("auto", "require"): + again = _gate_artifact_registry(raising=True) + replay = _run(graph, source, store, again, resume=resume) + assert all(receipt.hit for receipt in replay.nodes.values()) + assert replay.key == manifest.key + assert sum(_calls(again).values()) == 0 + assert_explained(replay, "hit") + + +def test_unreached_propagates_through_a_structural_node_and_its_version( + tmp_path: Path, +) -> None: + """A FILTER whose input is unreached is unreached, and so is its version. + + The version the filter would have opened has no population, so a node + placed on it is blocked by the filter itself (its version is one of its + compiled predecessors), while a node beside the edge still runs. + """ + gate = Node( + "gate", + "gate@1", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("household", "gate_verdict", "string"),), + population="survey", + artifact_outputs=(ArtifactOutput("evidence", EVIDENCE),), + ) + use = Node( + "use", + "use@1", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("person", "used", "float64"),), + population="survey", + artifact_inputs=(ArtifactInput("evidence", "gate", "evidence", EVIDENCE),), + ) + boundary = Node( + "boundary", + "keep@1", + inputs=(Slice("person", ("used",)),), + structural=StructuralDelta.FILTER, + base="survey", + ) + on_boundary = Node( + "on_boundary", + "after@1", + inputs=(Slice("person", ("used",)),), + outputs=(Owned("person", "after", "float64"),), + population="boundary", + ) + beside = Node( + "beside", + "a@1", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("person", "a", "float64"),), + params={"source": "age", "target": "a", "scale": 1.0}, + population="survey", + ) + graph = Graph("toy", (SOURCE,), (CREATE, gate, use, boundary, on_boundary, beside)) + + def keep(context: KernelContext) -> KernelResult: + person = context.tables["person"] + return KernelResult( + keep=pd.Series(True, index=person["person_id"], dtype="bool") + ) + + def registry_with_filter() -> KernelRegistry: + registry = _gate_artifact_registry(raising=True) + registry.register( + _Kernel( + "keep@1", + Capabilities( + Determinism.DETERMINISTIC, structural=StructuralDelta.FILTER + ), + keep, + ) + ) + return registry + + registry = registry_with_filter() + store = ContentStore(tmp_path / "store") + source = _source_path(tmp_path / "src") + manifest = _run(graph, source, store, registry) + nodes = manifest.nodes + assert nodes["boundary"].receipt["execution"]["blocked_by"] == { + "use": nodes["use"].key + } + assert nodes["on_boundary"].receipt["execution"]["blocked_by"] == { + "boundary": nodes["boundary"].key + } + assert nodes["boundary"].frame_key is None + assert "boundary" not in manifest.populations + assert "execution" not in nodes["beside"].receipt + assert set(manifest.populations["survey"].person["a"]) == {10.0, 20.0, 30.0} + calls = _calls(registry) + assert calls["keep@1"] == calls["after@1"] == 0 and calls["a@1"] == 1 + restored = RunManifest.from_json(manifest.to_json()) + assert restored.key == manifest.key + replay = _run(graph, source, store, registry_with_filter(), resume="require") + assert all(receipt.hit for receipt in replay.nodes.values()) + + +def test_a_gate_that_declares_evidence_and_passes_produces_it( + tmp_path: Path, +) -> None: + """The same declaration on a gate that succeeds is an ordinary byte edge.""" + store = ContentStore(tmp_path / "store") + source = _source_path(tmp_path / "src") + graph = _gate_artifact_graph(release_behind=True, answer="certified") + manifest = _run(graph, source, store, _gate_artifact_registry(raising=False)) + gate = manifest.nodes["gate"] + assert gate.receipt["outcome"] == "pass" + assert "execution" not in gate.receipt + assert gate.opaque_artifacts["evidence"] == opaque_artifact_key( + gate.key, "evidence" + ) + assert store.load_bytes(gate.opaque_artifacts["evidence"]) == b"evidence-bytes" + used = manifest.populations["survey"].person["used"] + assert set(used.to_numpy()) == {float(len(b"evidence-bytes"))} + assert manifest.nodes["release"].receipt["gate_ancestry"] == ("gate",) + assert manifest.tier == "certified" + assert '"schema_version":3' in manifest.to_json() + + +def test_a_kernel_may_not_author_the_executor_execution_state( + tmp_path: Path, +) -> None: + """The execution state is executor evidence; a kernel returning one is rejected.""" + + def authoring(context: KernelContext) -> KernelResult: + table = context.tables["person"] + return KernelResult( + columns={ + ("person", "a"): pd.Series( + np.zeros(len(table)), + index=pd.Index(table["person_id"], name="person_id"), + dtype="float64", + ) + }, + receipt={ + "execution": { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": {}, + } + }, + ) + + registry = KernelRegistry() + registry.register( + _Kernel( + "source@1", + Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE), + _source, + ) + ) + registry.register( + _Kernel("a@1", Capabilities(Determinism.DETERMINISTIC), authoring) + ) + graph = Graph("toy", (SOURCE,), (CREATE, _ordinary("a", "a@1", "age", "a"))) + store = ContentStore(tmp_path / "store") + with pytest.raises(NodeRejected, match="may not author executor execution"): _run(graph, _source_path(tmp_path / "src"), store, registry) + # A free-form "execution" diagnostic that does not claim the executor's + # schema is still just a receipt field. + assert ( + graph_executor.has_execution({"execution": {"literal_full_scans": 1}}) is False + ) + + +def test_a_cached_unreached_record_is_refused_once_its_inputs_exist( + tmp_path: Path, +) -> None: + """An unreached record is a hit only while the same inputs are unavailable.""" + store = ContentStore(tmp_path / "store") + source = _source_path(tmp_path / "src") + graph = _gate_artifact_graph(release_behind=True, answer="certified") + manifest = _run(graph, source, store, _gate_artifact_registry(raising=False)) + gate_key = manifest.nodes["gate"].key + use_key = manifest.nodes["use"].key + record_key = graph_executor._cache_record_key(use_key) + raw = store.load_json(record_key) + raw.update( + schema_version=3, + receipt={ + "outcome": "unreached", + "execution": { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": {"gate": gate_key}, + }, + "evidence": {"reason": "Required graph inputs are unavailable."}, + "capabilities": raw["capabilities"], + }, + columns=[], + frame_key=None, + weight=None, + opaque=[], + ) + store.put_json(record_key, raw, node_key=use_key, verify_existing=False) + for resume in ("auto", "require"): + with pytest.raises(StoreCorrupt, match="has no unavailable input blocker"): + _run( + graph, + source, + store, + _gate_artifact_registry(raising=False), + resume=resume, + ) + + +def test_a_manifest_authenticates_its_unreached_blockers(tmp_path: Path) -> None: + """Portable provenance names each blocker by key and the manifest checks it.""" + store = ContentStore(tmp_path / "store") + source = _source_path(tmp_path / "src") + graph = _gate_artifact_graph(release_behind=True, answer="evidence") + manifest = _run(graph, source, store, _gate_artifact_registry(raising=True)) + payload = json.loads(manifest.to_json()) + + forged = json.loads(json.dumps(payload)) + forged["nodes"]["use"]["receipt"]["execution"]["blocked_by"] = {"gate": "0" * 64} + with pytest.raises(ValueError, match="missing or has a different key"): + RunManifest.from_json(json.dumps(forged)) + + forged = json.loads(json.dumps(payload)) + forged["nodes"]["after"]["receipt"]["execution"]["blocked_by"] = { + "gate": payload["nodes"]["gate"]["key"] + } + with pytest.raises(ValueError, match="not one of its declared typed inputs"): + RunManifest.from_json(json.dumps(forged)) + + forged = json.loads(json.dumps(payload)) + forged["schema_version"] = 3 + with pytest.raises(ValueError, match="require manifest schema 4"): + RunManifest.from_json(json.dumps(forged)) + + +def test_the_private_population_observer_sees_every_admitted_population( + tmp_path: Path, +) -> None: + """``_population_observer`` runs per node, cold and on hits, before persistence. + + It is an integration seam for verifiers, not a kernel capability: nothing + it does enters a key or a receipt, and an exception it raises refuses the + run. + """ + store = ContentStore(tmp_path / "store") + source = _source_path(tmp_path / "src") + seen: list[tuple[str, int]] = [] + + def observe(node_id: str, population: Population) -> None: + seen.append((node_id, population.frame.n("person"))) + + cold = run_graph( + compile_graph(_graph()), + sources={"survey": source}, + store=store, + kernels=_registry(), + _population_observer=observe, + ) + assert [node_id for node_id, _ in seen] == list(cold.nodes) + assert {n for _, n in seen} == {3} + plain = _run(_graph(), source, store, _registry()) + assert plain.key == cold.key + + seen.clear() + warm = run_graph( + compile_graph(_graph()), + sources={"survey": source}, + store=store, + kernels=_registry(), + _population_observer=observe, + ) + assert all(receipt.hit for receipt in warm.nodes.values()) + assert [node_id for node_id, _ in seen] == list(warm.nodes) + + def refuse(node_id: str, population: Population) -> None: + raise RuntimeError(f"verifier refused {node_id}") + + with pytest.raises(RuntimeError, match="verifier refused survey"): + run_graph( + compile_graph(_graph()), + sources={"survey": source}, + store=ContentStore(tmp_path / "other"), + kernels=_registry(), + _population_observer=refuse, + ) + + +@pytest.mark.parametrize("warm", (False, True)) +def test_mutating_and_retained_observers_cannot_change_execution_or_cache( + tmp_path: Path, warm: bool +) -> None: + source = _source_path(tmp_path / "source") + compiled = compile_graph(_graph(leaf=False)) + plain = run_graph( + compiled, + sources={"survey": source}, + store=ContentStore(tmp_path / "plain"), + kernels=_registry(), + ) + store = ContentStore(tmp_path / "observed") + if warm: + run_graph( + compiled, sources={"survey": source}, store=store, kernels=_registry() + ) + retained = [] + + def mutate(population): + population.frame.person.loc[:, "age"] += 100 + weights = population.frame.weights_for("household").values + weights.setflags(write=True) + weights[:] = 999 + population.frame._metadata = {"observer": "changed"} + object.__setattr__(population, "owners", {}) + + def observe(node_id, population): + # Also mutate earlier snapshots during later callbacks, after a simple + # before/after check around their own callback would have completed. + retained.append(population) + for previous in retained: + mutate(previous) + + observed = run_graph( + compiled, + sources={"survey": source}, + store=store, + kernels=_registry(), + _population_observer=observe, + ) + for population in retained: + mutate(population) # retained references remain harmless after return + replay = run_graph( + compiled, sources={"survey": source}, store=store, kernels=_registry() + ) + assert all(receipt.hit for receipt in observed.nodes.values()) is warm + assert all(receipt.hit for receipt in replay.nodes.values()) + for actual in (observed, replay): + assert actual.key == plain.key + assert {name: item.key for name, item in actual.nodes.items()} == { + name: item.key for name, item in plain.nodes.items() + } + for entity in plain.populations["survey"].entities: + pd.testing.assert_frame_equal( + actual.populations["survey"].table(entity), + plain.populations["survey"].table(entity), + ) + np.testing.assert_array_equal( + actual.populations["survey"].weights_for("household").values, + plain.populations["survey"].weights_for("household").values, + ) + assert ( + actual.populations["survey"].metadata + == plain.populations["survey"].metadata + ) + assert observed.populations["survey"].person["b"].tolist() == [60.0, 120.0, 180.0] + + +def test_observer_snapshot_detaches_complete_population_storage(tmp_path: Path) -> None: + original = _source_frame(_source_path(tmp_path / "source")) + person = original.person.copy() + person.index = pd.MultiIndex.from_tuples( + [("a", 1), ("a", 2), ("b", 3)], names=["part", "row"] + ) + person["object_cell"] = pd.Series( + [{"nested": [1]}, {"nested": [2]}, {"nested": [3]}], + index=person.index, + dtype=object, + ) + person["category"] = pd.Categorical(["x", "y", "x"]) + person["selected"] = pd.array([True, pd.NA, False], dtype="boolean") + person["selected"].array._data[1] = True # preserve storage beneath the mask + person.attrs["nested"] = {"values": [1, 2]} + schema = EntitySchema( + group_entities=("household",), + links=(LinkSpec("relations", "person", "household"),), + ) + link = pd.DataFrame({"person_id": [1, 2, 3], "household_id": [10, 10, 20]}) + mass_log = (MassChangeRecord("household", 3.0, 3.0, 1.0, "unchanged"),) + frame = Frame( + {"person": person, "household": original.table("household"), "relations": link}, + schema, + {"household": original.weights_for("household")}, + pd.Series(["a", "a", "b"], index=person.index, name="stratum"), + metadata={"nested": [{"source": "fixture"}], "signed_zero": -0.0}, + mass_log=mass_log, + ) + ledger = ( + MassRecord( + "fixture", + "reweight", + "conserve", + 3.0, + 3.0, + (("a", 3.0),), + (("a", 3.0),), + entity="household", + ), + ) + population = Population.from_frame(frame, "fixture", mass_ledger=ledger) + snapshot = graph_executor._observer_snapshot(population) + assert snapshot.frame.schema == population.frame.schema + assert snapshot.frame.mass_log == population.frame.mass_log + assert snapshot.mass_ledger == population.mass_ledger + assert dict(snapshot.owners) == dict(population.owners) + assert dict(snapshot.weight_kind) == dict(population.weight_kind) + assert snapshot.frame.metadata == population.frame.metadata + assert np.signbit(snapshot.frame.metadata["signed_zero"]) + assert snapshot.frame.metadata is not frame.metadata + assert snapshot.frame.metadata["nested"][0] is not frame.metadata["nested"][0] + for name in frame.entities: + pd.testing.assert_frame_equal(snapshot.frame.table(name), frame.table(name)) + pd.testing.assert_frame_equal( + snapshot.frame.link("relations"), frame.link("relations") + ) + pd.testing.assert_series_equal(snapshot.frame.strata, frame.strata) + np.testing.assert_array_equal( + snapshot.design_weights["household"], population.design_weights["household"] + ) + assert not np.shares_memory( + snapshot.design_weights["household"], population.design_weights["household"] + ) + np.testing.assert_array_equal( + snapshot.frame.person["selected"].array._data, + frame.person["selected"].array._data, + ) + + snapshot.frame.person.at[("a", 1), "object_cell"]["nested"].append(99) + snapshot.frame.person.attrs["nested"]["values"].append(99) + snapshot.frame.person.index.set_names(["changed", "row"], inplace=True) + level = snapshot.frame.person.index.levels[0].to_numpy(copy=False) + level.setflags(write=True) + level[0] = "changed" + categories = snapshot.frame.person["category"].cat.categories.to_numpy(copy=False) + categories.setflags(write=True) + categories[0] = "changed" + snapshot.frame.link("relations").iloc[0, 0] = 999 + snapshot.frame.strata.iloc[0] = "changed" + snapshot.frame.person["selected"].array._data[1] = False + snapshot.frame.weights_for("household").values.setflags(write=True) + snapshot.frame.weights_for("household").values[:] = 999 + captured_metadata = snapshot.frame.metadata["nested"][0] + object.__setattr__(captured_metadata, "_items", (("source", "changed"),)) + assert frame.metadata["nested"][0]["source"] == "fixture" + snapshot.frame._metadata = {"changed": True} + object.__setattr__(snapshot.frame.schema.links[0], "name", "changed") + object.__setattr__(snapshot.frame.schema, "group_entities", ("changed",)) + object.__setattr__(snapshot.frame.mass_log[0], "reason", "changed") + object.__setattr__(snapshot.mass_ledger[0], "policy", "changed") + object.__setattr__(snapshot, "owners", {}) + assert frame.person.at[("a", 1), "object_cell"] == {"nested": [1]} + assert frame.person.attrs["nested"] == {"values": [1, 2]} + assert frame.person.index.names == ["part", "row"] + assert frame.person.index.levels[0].tolist() == ["a", "b"] + assert frame.person["category"].cat.categories.tolist() == ["x", "y"] + assert frame.link("relations").iloc[0, 0] == 1 + assert frame.strata.iloc[0] == "a" + assert bool(frame.person["selected"].array._data[1]) is True + assert frame.weights_for("household").values.tolist() == [1.0, 2.0] + assert frame.metadata["nested"][0]["source"] == "fixture" + assert frame.schema.links[0].name == "relations" + assert frame.schema.group_entities == ("household",) + assert frame.mass_log[0].reason == "unchanged" + assert population.mass_ledger[0].policy == "conserve" + assert population.owners + + # Record annotations do not freeze nested members. Even a caller-supplied + # container inside a record must not remain an alias across the seam. + nested_record = replace(ledger[0], before_by_stratum=((["mutable"], 3.0),)) + nested = replace(population, mass_ledger=(nested_record,)) + nested_snapshot = graph_executor._observer_snapshot(nested) + nested_snapshot.mass_ledger[0].before_by_stratum[0][0].append("changed") + assert nested.mass_ledger[0].before_by_stratum[0][0] == ["mutable"] + + +def test_absent_observer_allocates_no_snapshot(tmp_path: Path, monkeypatch) -> None: + def forbidden(population): + raise AssertionError("snapshot without observer") + + monkeypatch.setattr(graph_executor, "_observer_snapshot", forbidden) + source = _source_path(tmp_path / "source") + _run(_graph(), source, ContentStore(tmp_path / "store"), _registry()) def test_a_gate_reached_only_through_bytes_still_derives_the_tier( diff --git a/packages/microcosm-graph/tests/test_graph_explain.py b/packages/microcosm-graph/tests/test_graph_explain.py index da74a5385..b8bac925f 100644 --- a/packages/microcosm-graph/tests/test_graph_explain.py +++ b/packages/microcosm-graph/tests/test_graph_explain.py @@ -425,6 +425,28 @@ def test_manifest_only_page_omits_optional_sections(tmp_path: Path) -> None: assert "Incident replays" not in rendered +@pytest.mark.parametrize( + "diagnostic", + ({"state": "unreached"}, {"schema": "custom/v1", "state": "unreached"}), +) +def test_free_form_diagnostics_do_not_claim_executor_state( + tmp_path: Path, diagnostic: dict[str, str] +) -> None: + run = toy.run_toy(toy.full_graph(), tmp_path / "run") + original = run.manifest.nodes["calibrated"] + changed = replace( + original, receipt={**dict(original.receipt), "execution": diagnostic} + ) + manifest = replace( + run.manifest, nodes={**dict(run.manifest.nodes), "calibrated": changed} + ) + + rendered = explain_html(run.compiled, manifest) + + assert 'execution-unreached" data-node-detail=' not in rendered + assert 'execution-gate_exception" data-node-detail=' not in rendered + + def test_saved_run_cli_validates_store_and_reattaches_frames(tmp_path: Path) -> None: run = toy.run_toy(toy.full_graph(), tmp_path / "run") manifest_path = tmp_path / "run" / "manifest.json" From 48e2197cbcb1c67844f0dd24be66ed256ee641fd Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 12 Sep 2026 08:44:07 -0400 Subject: [PATCH 2/3] Verify unreached gates remain non-certifying on replay --- ...bservation-replay-consolidation.changed.md | 4 ++ .../tests/test_graph_executor.py | 70 +++++++++++++++++++ 2 files changed, 74 insertions(+) diff --git a/changelog.d/graph-observation-replay-consolidation.changed.md b/changelog.d/graph-observation-replay-consolidation.changed.md index a4aad2cef..91ebd6ecb 100644 --- a/changelog.d/graph-observation-replay-consolidation.changed.md +++ b/changelog.d/graph-observation-replay-consolidation.changed.md @@ -1 +1,5 @@ Add explicit raw-byte source codecs, preserve complete Frame metadata in the content store, represent unavailable gate artifacts and unreached consumers independently of cache hits, and isolate population observers from execution and persistence. + +Frame store v1 objects are unavailable under the new metadata contract. Rebuild +them with `resume="forbid"`; automatic and required replay refuse them rather +than silently dropping metadata. diff --git a/packages/microcosm-graph/tests/test_graph_executor.py b/packages/microcosm-graph/tests/test_graph_executor.py index 6dc5ca1b6..1ceddbc72 100644 --- a/packages/microcosm-graph/tests/test_graph_executor.py +++ b/packages/microcosm-graph/tests/test_graph_executor.py @@ -3831,6 +3831,76 @@ def assert_explained(outcome: RunManifest, cache: str) -> None: assert_explained(replay, "hit") +def test_unreached_gate_cannot_certify_a_downstream_release(tmp_path: Path) -> None: + base = _gate_artifact_graph(release_behind=True, answer="certified") + second_gate = Node( + "second_gate", + "second_gate@1", + inputs=(Slice("person", ("used",)),), + outputs=(Owned("household", "second_verdict", "string"),), + population="survey", + ) + release = replace( + base.node("release"), inputs=(Slice("household", ("second_verdict",)),) + ) + graph = Graph( + "toy", + (SOURCE,), + (CREATE, base.node("gate"), base.node("use"), second_gate, release), + ) + + def forbidden_gate(context: KernelContext) -> KernelResult: + raise AssertionError("An unreached gate must not run") + + def registry_with_second_gate() -> KernelRegistry: + registry = _gate_artifact_registry(raising=True) + registry.register( + _Kernel( + "second_gate@1", + Capabilities(Determinism.DETERMINISTIC, role=KernelRole.GATE), + forbidden_gate, + ) + ) + return registry + + source = _source_path(tmp_path / "src") + store = ContentStore(tmp_path / "store") + cold_key = None + for resume in ("auto", "require"): + registry = registry_with_second_gate() + manifest = _run(graph, source, store, registry, resume=resume) + gate = manifest.nodes["second_gate"] + assert gate.receipt["outcome"] == "unreached" + assert gate.receipt["execution"] == { + "schema": EXECUTION_SCHEMA, + "state": "unreached", + "blocked_by": {"use": manifest.nodes["use"].key}, + } + assert not gate.artifacts and gate.frame_key is None + assert not gate.opaque_artifacts + assert "second_verdict" not in manifest.population("survey").household + assert manifest.nodes["release"].receipt["execution"]["blocked_by"] == { + "second_gate": gate.key + } + assert manifest.nodes["release"].receipt["tier"] == "evidence" + assert manifest.tier == "evidence" + calls = _calls(registry) + assert calls["second_gate@1"] == calls["release@1"] == 0 + restored = RunManifest.from_json(manifest.to_json()) + assert restored.nodes["second_gate"].receipt == gate.receipt + assert restored.key == manifest.key and restored.tier == "evidence" + rendered = explain_html(compile_graph(graph), manifest) + cache = "hit" if resume == "require" else "miss" + assert f"{cache} · gate unreached · unreached" in rendered + if resume == "require": + assert manifest.key == cold_key + assert all(node.hit for node in manifest.nodes.values()) + assert sum(calls.values()) == 0 + else: + cold_key = manifest.key + assert calls["gate@1"] == 1 + + def test_unreached_propagates_through_a_structural_node_and_its_version( tmp_path: Path, ) -> None: From 4adac59aff72803593816f4f319a3182686ab112 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Sat, 12 Sep 2026 08:56:38 -0400 Subject: [PATCH 3/3] Round-trip frozen mapping members in Frame metadata sets --- .../src/microcosm/graph/store.py | 9 ++++++- .../tests/test_frame_metadata_store.py | 24 +++++++++++++++++++ 2 files changed, 32 insertions(+), 1 deletion(-) diff --git a/packages/microcosm-graph/src/microcosm/graph/store.py b/packages/microcosm-graph/src/microcosm/graph/store.py index d30050b00..6321f46f6 100644 --- a/packages/microcosm-graph/src/microcosm/graph/store.py +++ b/packages/microcosm-graph/src/microcosm/graph/store.py @@ -39,6 +39,7 @@ Weights, nullable_boolean_values_and_mask, ) +from microcosm.frame.bundle import _freeze_metadata_value from .errors import ( GraphRuntimeError, @@ -1097,7 +1098,13 @@ def _decode_frame_metadata(encoded: object) -> object: return result if kind in ("tuple", "frozenset") and isinstance(value, list): items = [_decode_frame_metadata(item) for item in value] - return tuple(items) if kind == "tuple" else frozenset(items) + if kind == "tuple": + return tuple(items) + # Frame admits hashable frozen mappings, including inside tuple + # members. Reapply its recursive freezing before building a set. + return frozenset( + _freeze_metadata_value(item, path="stored metadata[]") for item in items + ) if ( kind == "float64" and isinstance(value, str) diff --git a/packages/microcosm-graph/tests/test_frame_metadata_store.py b/packages/microcosm-graph/tests/test_frame_metadata_store.py index 917119d77..d916b998c 100644 --- a/packages/microcosm-graph/tests/test_frame_metadata_store.py +++ b/packages/microcosm-graph/tests/test_frame_metadata_store.py @@ -75,6 +75,30 @@ def test_same_key_with_changed_metadata_refuses_without_replacing_original(tmp_p assert store.load_frame("b" * 64).metadata == original.metadata +@pytest.mark.parametrize("wrap_in_tuple", [False, True]) +def test_roundtrip_preserves_frozen_sets_of_inherited_mappings(tmp_path, wrap_in_tuple): + parent = _frame({"sources": ({"id": "invented", "nested": {"rows": 2}},)}) + source = parent.metadata["sources"][0] + member = (source,) if wrap_in_tuple else source + original = _frame({"source_set": frozenset({member})}) + store = ContentStore(tmp_path / "invented-store") + key = "9" * 64 + store.put_frame(key, original) + restored = store.load_frame(key) + assert isinstance(restored.metadata["source_set"], frozenset) + assert store_module._encode_frame_metadata( + restored.metadata + ) == store_module._encode_frame_metadata(original.metadata) + restored_member = next(iter(restored.metadata["source_set"])) + restored_source = restored_member[0] if wrap_in_tuple else restored_member + assert hash(restored_member) == hash(member) + with pytest.raises(TypeError): + restored_source["nested"]["rows"] = 3 + # A normal same-key cache write must keep loading the admitted metadata. + store.put_frame(key, restored) + assert store.load_frame(key).metadata == original.metadata + + def test_v1_frame_is_unavailable_instead_of_silently_losing_metadata( tmp_path, monkeypatch ):