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..91ebd6ecb --- /dev/null +++ b/changelog.d/graph-observation-replay-consolidation.changed.md @@ -0,0 +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/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..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, @@ -62,7 +63,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 +195,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 +727,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 +762,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 +930,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 +1057,67 @@ 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] + 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) + 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 +1172,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 +1219,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 +1366,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..d916b998c --- /dev/null +++ b/packages/microcosm-graph/tests/test_frame_metadata_store.py @@ -0,0 +1,222 @@ +"""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 + + +@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 +): + 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..1ceddbc72 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,732 @@ 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") + 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_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") - with pytest.raises(NodeRejected, match="gate kernel may not declare"): + 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: + """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"