diff --git a/.impl-b7-link b/.impl-b7-link deleted file mode 120000 index f49c784f9..000000000 --- a/.impl-b7-link +++ /dev/null @@ -1 +0,0 @@ -/private/tmp/microcosm-impl-b7 \ No newline at end of file diff --git a/packages/microcosm-fit/src/microcosm/fit/kernels.py b/packages/microcosm-fit/src/microcosm/fit/kernels.py index 1df027289..a48bc08e3 100644 --- a/packages/microcosm-fit/src/microcosm/fit/kernels.py +++ b/packages/microcosm-fit/src/microcosm/fit/kernels.py @@ -49,12 +49,12 @@ ) """Distributions whose versions form part of ``fit.qrf@1``'s identity.""" -#: How far ``fit.qrf@1`` numbers may move between machines. The forest stack -#: promises no cross-platform bit stability (charter H1 records the claim as -#: ``tolerance_bound``); this bound is provisional until measured on the H1 -#: fixture across arm64 and x86_64 (amendment 13 follow-up), and parity in -#: the locked environment is still asserted byte for byte. -FIT_QRF_TOLERANCE = Tolerance(rtol=1e-6) +#: Provisional one-ULP acceptance budget, not an established cross-platform +#: upper bound. On 2026-09-02, one 12-output positive-only H1 fixture was +#: bit-identical between native arm64 and x86_64 under Rosetta. Broader +#: native-x86, seed, regime, near-tie, and non-binary-exact coverage is required +#: before treating this as a kernel-wide bound. +FIT_QRF_TOLERANCE = Tolerance(ulps=1) QRF_EXECUTOR_SEED_HIGH = 2**31 - 1 diff --git a/packages/microcosm-fit/tests/test_kernels.py b/packages/microcosm-fit/tests/test_kernels.py index 930694005..a5ae21336 100644 --- a/packages/microcosm-fit/tests/test_kernels.py +++ b/packages/microcosm-fit/tests/test_kernels.py @@ -234,7 +234,7 @@ def test_capabilities_protocol_and_wrapped_source_hash() -> None: numeric=Numeric.TOLERANCE_BOUND, seed_source=SeedSource.PARAM, dependencies=FIT_QRF_DEPENDENCIES, - tolerance=Tolerance(rtol=1e-6), + tolerance=Tolerance(ulps=1), ) assert QRF_EXECUTOR_KERNEL.capabilities.seed_source is SeedSource.EXECUTOR assert QRF_PARAM_KERNEL.implementation_hash() == source_hash( diff --git a/packages/microcosm-graph/src/microcosm/graph/__init__.py b/packages/microcosm-graph/src/microcosm/graph/__init__.py index a090e96cb..81e919c80 100644 --- a/packages/microcosm-graph/src/microcosm/graph/__init__.py +++ b/packages/microcosm-graph/src/microcosm/graph/__init__.py @@ -83,6 +83,7 @@ "Ownership", "Param", "Population", + "PopulationView", "PopulationError", "ResumePolicy", "RunManifest", @@ -137,7 +138,7 @@ def _check_frame_version() -> None: ) from .executor import NodeRejected, run_graph # noqa: E402 from .explain import explain_html # noqa: E402 -from .manifest import Decision, NodeReceipt, RunManifest # noqa: E402 +from .manifest import Decision, NodeReceipt, PopulationView, RunManifest # noqa: E402 from .population import MassRecord, Population, PopulationError # noqa: E402 from .serialize import graph_from_json, graph_to_json # noqa: E402 from .store import ( # noqa: E402 diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 6bb035966..a14475d90 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -34,8 +34,10 @@ KernelRegistry, KernelResult, KernelRole, + Tolerance, ) from .keys import ( + _capabilities_projection, artifact_key, frame_key, node_key, @@ -46,7 +48,9 @@ from .manifest import Decision, NodeReceipt, RunManifest from .population import ( Population, + entrant_strata_receipt, expand_lineage_receipt, + mass_record_receipt, patch, restore_cached_expand, weight_cap_receipt, @@ -80,18 +84,6 @@ def _opaque_artifact_key(key: str, name: str) -> str: return sha256_domain("node-artifact", canonical_json((key, name))) -def _capabilities_payload(capabilities: Capabilities) -> dict[str, object]: - return { - "determinism": capabilities.determinism.value, - "numeric": capabilities.numeric.value, - "seed_source": capabilities.seed_source.value, - "structural": capabilities.structural.value, - "role": capabilities.role.value, - "consumes_se": capabilities.consumes_se, - "dependencies": list(capabilities.dependencies), - } - - def _normal_json_mapping(value: Mapping[str, object], label: str) -> dict[str, object]: """Validate and detach a descriptive mapping through canonical JSON.""" @@ -400,6 +392,7 @@ def _project_context( *, key: str, sources: Mapping[str, Path], + tolerances: Mapping[tuple[str, str], Tolerance | None], ) -> KernelContext: if population is None: return KernelContext( @@ -410,6 +403,7 @@ def _project_context( params=node.params, rng=np.random.default_rng(seed(key)), sources=MappingProxyType({name: sources[name] for name in node.sources}), + tolerances=tolerances, ) frame = population.frame @@ -519,9 +513,48 @@ def _project_context( params=node.params, rng=np.random.default_rng(seed(key)), sources=MappingProxyType({name: sources[name] for name in node.sources}), + tolerances=tolerances, ) +def _input_tolerances( + compiled: CompiledGraph, + node_id: str, + kernels: KernelRegistry, +) -> Mapping[tuple[str, str], Tolerance | None]: + """Resolve explicit inputs and rewrite incumbents as compilation does.""" + + node = compiled.graph.node(node_id) + if node.structural is StructuralDelta.CREATE: + return MappingProxyType({}) + input_version = ( + compiled.versions[node_id] + if node.structural is StructuralDelta.NONE + else node.base + ) + assert input_version is not None + rewritten = { + (owned.entity, owned.column) for owned in node.outputs if owned.rewrite + } + coordinates = rewritten | { + (slice_.entity, column) for slice_ in node.inputs for column in slice_.columns + } + resolved: dict[tuple[str, str], Tolerance | None] = {} + for coordinate in sorted(coordinates): + entity, column = coordinate + owner_id = ( + input_version + if coordinate in rewritten + else compiled.owners.get( + (input_version, entity, column), + input_version, + ) + ) + owner = compiled.graph.node(owner_id) + resolved[coordinate] = kernels.get(owner.kernel).capabilities.tolerance + return MappingProxyType(resolved) + + def _validate_series( node: Node, owned: Owned, @@ -645,6 +678,15 @@ def _validate_result( raise NodeRejected(f"Node {node.id!r} result.artifacts is not a mapping.") if not isinstance(result.receipt, Mapping): raise NodeRejected(f"Node {node.id!r} result.receipt is not a mapping.") + if result.strata is not None and not isinstance(result.strata, pd.Series): + raise NodeRejected(f"Node {node.id!r} result.strata is not a Series.") + if result.strata is not None and ( + cache_hit or node.structural is not StructuralDelta.EXPAND or not node.entrants + ): + raise NodeRejected( + f"Node {node.id!r} returned entrant strata outside a fresh " + "entrants=True EXPAND." + ) if kernel_capabilities.structural is not node.structural: raise NodeRejected( f"Node {node.id!r} declares structural={node.structural.value!r}, but " @@ -763,10 +805,18 @@ def _validate_result( assert result.expand is not None try: receipt["expand"] = expand_lineage_receipt(result.expand) + assert population is not None + strata_receipt = entrant_strata_receipt( + population.frame, node, result.expand, result.strata + ) except (TypeError, ValueError) as error: raise NodeRejected( - f"EXPAND node {node.id!r} returned malformed lineage: {error}" + f"EXPAND node {node.id!r} returned malformed lineage or " + f"entrant strata: {error}" ) from error + receipt.pop("entrant_strata", None) + if strata_receipt is not None: + receipt["entrant_strata"] = strata_receipt if kernel_capabilities.role is KernelRole.GATE: outcome = receipt.get("outcome") if outcome not in GATE_OUTCOMES: @@ -777,6 +827,86 @@ def _validate_result( return receipt, artifacts +def _validate_entrant_materialization_contract( + compiled: CompiledGraph, + node: Node, + population: Population | None, + receipt: Mapping[str, object], +) -> None: + """Require every entrant's carried data cells to have downstream claims.""" + + if not node.entrants or population is None: + return + raw_expand = receipt.get("expand") + if not isinstance(raw_expand, Mapping): + return # the ordinary EXPAND validation reports the malformed receipt + entrant_entities: set[str] = set() + for entity, entries in raw_expand.items(): + if not isinstance(entity, str) or not isinstance(entries, list): + continue + if any( + isinstance(entry, list) and len(entry) == 2 and entry[1] is None + for entry in entries + ): + entrant_entities.add(entity) + + frame = population.frame + for entity in sorted(entrant_entities): + if entity not in frame.entities: + continue # lineage validation supplies the node-naming rejection + structural = set(_structural_columns(frame, entity)) + if ( + compiled.graph.mass_partition is not None + and compiled.graph.mass_partition[0] == entity + ): + structural.add(compiled.graph.mass_partition[1]) + for column in frame.table(entity).columns: + column = str(column) + if column in structural: + continue + coordinate = (entity, column) + claimant_id = compiled.owners.get((node.id, entity, column)) + if claimant_id is None: + raise NodeRejected( + f"EXPAND node {node.id!r} entrant cell {entity}.{column} " + "has no materialized_expand_outputs ownership claim." + ) + claimant = compiled.graph.node(claimant_id) + claimed = claimant.params.get("materialized_expand_outputs", ()) + spelling = f"{entity}.{column}" + output = next( + ( + owned + for owned in claimant.outputs + if (owned.entity, owned.column) == coordinate + ), + None, + ) + if ( + not isinstance(claimed, tuple) + or spelling not in claimed + or output is None + or output.rewrite + ): + raise NodeRejected( + f"EXPAND node {node.id!r} entrant cell {spelling} is not " + f"declared through node {claimant_id!r}'s " + "materialized_expand_outputs." + ) + if output.rows != ROWS_ALL: + raise NodeRejected( + f"EXPAND node {node.id!r} entrant cell {spelling} is claimed " + f"through masked rows {output.rows!r}; materialization bridge " + "claims must use rows='all'." + ) + carried_dtype = _dtype_token(frame.table(entity)[column]) + if output.dtype != carried_dtype: + raise NodeRejected( + f"EXPAND node {node.id!r} entrant cell {spelling} is claimed " + f"as {output.dtype!r}; its carried dtype is {carried_dtype!r}." + ) + + def _create_population(node: Node, frame: Frame) -> Population: # Entity ids and membership columns are structural Frame columns rather # than declaration-owned data cells, but Population ownership is total @@ -816,7 +946,20 @@ def _apply_result( population: Population | None, *, cache_hit: bool = False, + mass_partition: tuple[str, str] | None = None, ) -> Population: + if ( + mass_partition is not None + and node.structural is StructuralDelta.NONE + and any( + (owned.entity, owned.column) == mass_partition for owned in node.outputs + ) + ): + entity, column = mass_partition + raise NodeRejected( + f"Node {node.id!r} cannot own mass partition {entity}.{column}; " + "partition values are fixed by the structural population." + ) if node.structural is StructuralDelta.CREATE: assert result.frame is not None return _create_population(node, result.frame) @@ -827,7 +970,9 @@ def _apply_result( and result.frame is not None ): try: - return restore_cached_expand(population, node, result) + return restore_cached_expand( + population, node, result, mass_partition=mass_partition + ) except (TypeError, ValueError) as error: raise NodeRejected( f"Node {node.id!r} cached EXPAND rejected: {error}" @@ -860,7 +1005,7 @@ def _apply_result( receipt=result.receipt, ) try: - return patch(population, node, result) + return patch(population, node, result, mass_partition=mass_partition) except NodeRejected: raise except (TypeError, ValueError) as error: @@ -979,7 +1124,7 @@ def _write_node( "node_key": key, "kernel_ref": node.kernel, "kernel_impl_hash": kernel_impl_hash, - "capabilities": _capabilities_payload(capabilities), + "capabilities": _capabilities_projection(capabilities), "receipt": dict(receipt), "columns": column_entries, "frame_key": stored_frame_key, @@ -996,7 +1141,12 @@ def _write_node( def _require_record_shape( - raw: object, node: Node, *, key: str, kernel_impl_hash: str + raw: object, + node: Node, + *, + key: str, + kernel_impl_hash: str, + capabilities: Capabilities, ) -> dict[str, object]: if not isinstance(raw, dict): raise StoreCorrupt(f"Cached receipt for node {node.id!r} is not an object.") @@ -1035,6 +1185,12 @@ def _require_record_shape( f"Cached receipt identity for node {node.id!r} is {actual!r}, " f"not {expected!r}." ) + expected_capabilities = _capabilities_projection(capabilities) + if raw["capabilities"] != expected_capabilities: + raise StoreMiss( + f"Cached receipt capabilities for node {node.id!r} disagree with " + "the registered kernel contract." + ) return raw @@ -1053,9 +1209,16 @@ def _load_record( *, key: str, kernel_impl_hash: str, + capabilities: Capabilities, ) -> dict[str, object]: raw = store.load_json(_cache_record_key(key)) - return _require_record_shape(raw, node, key=key, kernel_impl_hash=kernel_impl_hash) + return _require_record_shape( + raw, + node, + key=key, + kernel_impl_hash=kernel_impl_hash, + capabilities=capabilities, + ) def _preflight_record(store: ContentStore, record: Mapping[str, object]) -> None: @@ -1252,6 +1415,7 @@ def _all_node_keys( keys, implementation, source_keys, + kernel_capabilities=kernel.capabilities, ) return keys, implementations @@ -1261,6 +1425,7 @@ def _preflight_require( store: ContentStore, keys: Mapping[str, str], implementations: Mapping[str, str], + kernels: KernelRegistry, ) -> None: missing: list[str] = [] for node_id in compiled.order: @@ -1271,6 +1436,7 @@ def _preflight_require( node, key=keys[node_id], kernel_impl_hash=implementations[node_id], + capabilities=kernels.get(node.kernel).capabilities, ) _preflight_record(store, record) except StoreMiss: @@ -1309,7 +1475,7 @@ def run_graph( source_paths, source_keys = _source_paths_and_keys(compiled, sources, store) keys, implementations = _all_node_keys(compiled, kernels, source_keys) if resume == "require": - _preflight_require(compiled, store, keys, implementations) + _preflight_require(compiled, store, keys, implementations, kernels) populations: dict[str, Population] = {} receipts: dict[str, NodeReceipt] = {} @@ -1345,6 +1511,7 @@ def run_graph( node, key=key, kernel_impl_hash=implementation, + capabilities=kernel.capabilities, ) result, manifest_artifacts = _load_cached_result( store, node, incumbent, record @@ -1355,7 +1522,13 @@ def run_graph( raise if result is None: - context = _project_context(node, incumbent, key=key, sources=source_paths) + context = _project_context( + node, + incumbent, + key=key, + sources=source_paths, + tolerances=_input_tolerances(compiled, node_id, kernels), + ) before = _context_digest(context) try: result = kernel.run(context) @@ -1385,6 +1558,9 @@ def run_graph( incumbent, cache_hit=hit, ) + _validate_entrant_materialization_contract( + compiled, node, incumbent, normalized_receipt + ) if kernel.capabilities.role is KernelRole.RELEASE: derived_tier, gate_ids = _release_tier(compiled, node_id, receipts) _validate_release_tier(node, result, derived_tier) @@ -1393,8 +1569,37 @@ def run_graph( "pass" if derived_tier == "certified" else "fail" ) normalized_receipt["gate_ancestry"] = list(gate_ids) - normalized_receipt["capabilities"] = _capabilities_payload(kernel.capabilities) - updated = _apply_result(node, result, incumbent, cache_hit=hit) + normalized_receipt["capabilities"] = _capabilities_projection( + kernel.capabilities + ) + updated = _apply_result( + node, + result, + incumbent, + cache_hit=hit, + mass_partition=compiled.graph.mass_partition, + ) + author_mass = compiled.graph.mass_partition is not None or ( + node.structural is StructuralDelta.EXPAND + and node.entrants + and "entrant_strata" in normalized_receipt + ) + if author_mass and node.structural not in { + StructuralDelta.NONE, + StructuralDelta.CREATE, + }: + existing_mass = normalized_receipt.get("mass", {}) + if not isinstance(existing_mass, Mapping): # defended by mass validation + raise NodeRejected( + f"Node {node.id!r} receipt['mass'] is not a mapping." + ) + try: + authored_mass = mass_record_receipt(updated.mass_ledger[-1]) + except (TypeError, ValueError) as error: + raise NodeRejected( + f"Node {node.id!r} mass receipt rejected: {error}" + ) from error + normalized_receipt["mass"] = {**existing_mass, **authored_mass} normalized_receipt.update(weight_cap_receipt(updated, node)) cache_receipt = normalized_receipt run_receipt = dict(cache_receipt) diff --git a/packages/microcosm-graph/src/microcosm/graph/explain.py b/packages/microcosm-graph/src/microcosm/graph/explain.py index 2bc4f8996..3eb8292bb 100644 --- a/packages/microcosm-graph/src/microcosm/graph/explain.py +++ b/packages/microcosm-graph/src/microcosm/graph/explain.py @@ -460,6 +460,7 @@ def _render_graph( def _capabilities(receipt: NodeReceipt) -> dict[str, object]: capabilities = receipt.capabilities + tolerance = capabilities.tolerance return { "determinism": _value(capabilities.determinism), "numeric": _value(capabilities.numeric), @@ -468,6 +469,15 @@ def _capabilities(receipt: NodeReceipt) -> dict[str, object]: "role": _value(capabilities.role), "consumes_se": capabilities.consumes_se, "dependencies": capabilities.dependencies, + "tolerance": ( + None + if tolerance is None + else { + "rtol": tolerance.rtol, + "atol": tolerance.atol, + "ulps": tolerance.ulps, + } + ), } @@ -995,13 +1005,16 @@ def _mass_payload( ) -> dict[str, object] | None: raw = receipt.receipt.get("mass") if isinstance(raw, Mapping): - return { + payload = { "before": raw.get("before"), "after": raw.get("after"), "stratum_before": raw.get("stratum_before", {}), "stratum_after": raw.get("stratum_after", {}), "policy": raw.get("policy", node.mass), } + if isinstance(raw.get("partition"), Mapping): + payload["partition"] = raw["partition"] + return payload for record in reversed(manifest.mass_ledgers.get(node.id, ())): if record.node_id == node.id: return { diff --git a/packages/microcosm-graph/src/microcosm/graph/keys.py b/packages/microcosm-graph/src/microcosm/graph/keys.py index daf3f5bc3..8f59ddb0d 100644 --- a/packages/microcosm-graph/src/microcosm/graph/keys.py +++ b/packages/microcosm-graph/src/microcosm/graph/keys.py @@ -8,6 +8,7 @@ from .canonical import canonical_json, normative, sha256_domain from .decl import CompiledGraph, StructuralDelta +from .kernel import Capabilities __all__ = [ "artifact_key", @@ -94,12 +95,43 @@ def _required_key(keys: Mapping[str, str], node_id: str, consumer: str) -> str: ) from error +def _canonical_tolerance_float(value: int | float) -> float: + number = float(value) + return 0.0 if number == 0.0 else number + + +def _capabilities_projection(capabilities: Capabilities) -> dict[str, object]: + """Return the complete canonical payload for a kernel contract.""" + + tolerance = capabilities.tolerance + return { + "determinism": capabilities.determinism.value, + "numeric": capabilities.numeric.value, + "seed_source": capabilities.seed_source.value, + "structural": capabilities.structural.value, + "role": capabilities.role.value, + "consumes_se": capabilities.consumes_se, + "dependencies": list(capabilities.dependencies), + "tolerance": ( + None + if tolerance is None + else { + "rtol": _canonical_tolerance_float(tolerance.rtol), + "atol": _canonical_tolerance_float(tolerance.atol), + "ulps": tolerance.ulps, + } + ), + } + + def node_key( compiled: CompiledGraph, node_id: str, input_keys: Mapping[str, str], kernel_impl_hash: str, source_keys: Mapping[str, str], + *, + kernel_capabilities: Capabilities, ) -> str: """Derive a node key from its declaration and resolved input identities. @@ -119,12 +151,19 @@ def node_key( input_version = node.base resolved: dict[tuple[str, str], str] = {} + rewritten = { + (owned.entity, owned.column) for owned in node.outputs if owned.rewrite + } if input_version is not None: for slice_ in node.inputs: for column in slice_.columns: coordinate = (slice_.entity, column) - producer = compiled.owners.get( - (input_version, slice_.entity, column), input_version + producer = ( + input_version + if coordinate in rewritten + else compiled.owners.get( + (input_version, slice_.entity, column), input_version + ) ) producer_key = _required_key(input_keys, producer, node_id) resolved[coordinate] = artifact_key(producer_key, slice_.entity, column) @@ -174,6 +213,10 @@ def node_key( graph_facts = ( {} if node.structural is StructuralDelta.NONE else compiled.graph.normative() ) + # Capabilities are executable contract, independent of implementation + # bytes. Bind the complete declaration so a cache entry produced under one + # contract cannot satisfy another kernel with the same ref and code hash. + capabilities = _capabilities_projection(kernel_capabilities) return _hash_parts( "node", normative(node), @@ -182,6 +225,7 @@ def node_key( kernel_impl_hash, resolved_sources, graph_facts, + capabilities, ) diff --git a/packages/microcosm-graph/src/microcosm/graph/manifest.py b/packages/microcosm-graph/src/microcosm/graph/manifest.py index 8f0a9857b..b6065f8f7 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -11,23 +11,53 @@ from types import MappingProxyType from typing import TYPE_CHECKING, Self +from microcosm.frame import Frame + from .canonical import canonical_json, sha256_domain from .decl import GATE_OUTCOMES, StructuralDelta from .errors import NodeRejectedError, StoreCorruptError -from .kernel import Capabilities, Determinism, KernelRole, Numeric, SeedSource +from .kernel import ( + Capabilities, + Determinism, + KernelRole, + Numeric, + SeedSource, + Tolerance, +) from .population import MassRecord if TYPE_CHECKING: - from microcosm.frame import Frame - from .store import ContentStore -__all__ = ["Decision", "NodeReceipt", "RunManifest"] +__all__ = ["Decision", "NodeReceipt", "PopulationView", "RunManifest"] _SCHEMA_VERSION = 1 _CERTIFYING_GATE_OUTCOMES = frozenset({"pass", "not_applicable"}) +class PopulationView(Frame): + """Zero-copy manifest view with entity-name table access. + + All attached populations use this type. Existing :class:`Frame` accessors + remain available, and a group entity can also be read by name (for example, + `view.household` is equivalent to `view.table("household")`). The source + Frame keeps its original type. + """ + + __slots__ = () + + def __init__(self, frame: Frame) -> None: + if not isinstance(frame, Frame): + raise TypeError("PopulationView requires a Frame") + for slot in Frame.__slots__: + object.__setattr__(self, slot, getattr(frame, slot)) + + def __getattr__(self, name: str) -> object: + if name in self.entities: + return self.table(name) + raise AttributeError(f"{type(self).__name__!s} has no attribute {name!r}") + + def _freeze_json(value: object) -> object: """Copy JSON-like receipt data into immutable containers.""" @@ -55,6 +85,16 @@ def _enum_value(value: object) -> object: return value.value if isinstance(value, Enum) else value +def _tolerance_payload(tolerance: Tolerance | None) -> dict[str, object] | None: + if tolerance is None: + return None + return { + "rtol": float(tolerance.rtol), + "atol": float(tolerance.atol), + "ulps": tolerance.ulps, + } + + @dataclass(frozen=True) class Decision(Mapping[str, str]): """A signed human decision carried as provenance, never as a node input.""" @@ -241,6 +281,7 @@ def _payload(self) -> dict[str, object]: "role": _enum_value(capabilities.role), "consumes_se": capabilities.consumes_se, "dependencies": capabilities.dependencies, + "tolerance": _tolerance_payload(capabilities.tolerance), }, "receipt": self.receipt, "artifacts": tuple( @@ -301,8 +342,17 @@ def __post_init__(self) -> None: for name in ("started_at", "finished_at", "host"): if not isinstance(getattr(self, name), str): raise TypeError(f"RunManifest.{name} must be a string") + populations: dict[str, PopulationView] = {} + for version_id, frame in self.populations.items(): + if not isinstance(version_id, str): + raise TypeError("RunManifest.populations keys must be strings") + if not isinstance(frame, Frame): + raise TypeError("RunManifest.populations values must be Frame") + populations[version_id] = PopulationView(frame) object.__setattr__( - self, "populations", MappingProxyType(dict(self.populations)) + self, + "populations", + MappingProxyType(populations), ) mass_ledgers: dict[str, tuple[MassRecord, ...]] = {} for version_id, records in self.mass_ledgers.items(): @@ -416,7 +466,7 @@ def node(self, node_id: str) -> NodeReceipt: def receipt(self, node_id: str) -> NodeReceipt: return self.node(node_id) - def population(self, version_id: str) -> Frame: + def population(self, version_id: str) -> PopulationView: """Return an attached final population version. Population frames are deliberately not serialized in manifest JSON; @@ -425,11 +475,14 @@ def population(self, version_id: str) -> Frame: """ try: - return self.populations[version_id] + population = self.populations[version_id] except KeyError as error: raise KeyError( f"Population {version_id!r} is not attached to this manifest." ) from error + if not isinstance(population, PopulationView): # __post_init__ invariant + raise RuntimeError("attached population was not normalized") + return population def mass_ledger(self, version_id: str) -> tuple[MassRecord, ...]: """Return the transient mass audit trail for one attached version.""" @@ -687,6 +740,35 @@ def _capabilities_from_payload(value: object) -> Capabilities: raise ValueError("capabilities.dependencies must be an array") if not all(isinstance(item, str) for item in dependencies): raise ValueError("capabilities.dependencies must contain strings") + raw_tolerance = value.get("tolerance") + if raw_tolerance is None: + tolerance = None + else: + if not isinstance(raw_tolerance, Mapping) or set(raw_tolerance) != { + "rtol", + "atol", + "ulps", + }: + raise ValueError( + "capabilities.tolerance must be null or an object containing " + "rtol, atol, and ulps" + ) + rtol = raw_tolerance["rtol"] + atol = raw_tolerance["atol"] + ulps = raw_tolerance["ulps"] + if ( + isinstance(rtol, bool) + or not isinstance(rtol, int | float) + or isinstance(atol, bool) + or not isinstance(atol, int | float) + or isinstance(ulps, bool) + or not isinstance(ulps, int) + ): + raise ValueError( + "capabilities.tolerance rtol/atol must be numeric and ulps " + "must be an integer" + ) + tolerance = Tolerance(rtol=float(rtol), atol=float(atol), ulps=ulps) return Capabilities( determinism=Determinism(_string_field(value, "determinism")), numeric=Numeric(_string_field(value, "numeric")), @@ -695,6 +777,7 @@ def _capabilities_from_payload(value: object) -> Capabilities: role=KernelRole(str(value.get("role", KernelRole.COMPUTE.value))), consumes_se=consumes_se, dependencies=tuple(dependencies), + tolerance=tolerance, ) diff --git a/packages/microcosm-graph/src/microcosm/graph/population.py b/packages/microcosm-graph/src/microcosm/graph/population.py index 98a551303..e16606bde 100644 --- a/packages/microcosm-graph/src/microcosm/graph/population.py +++ b/packages/microcosm-graph/src/microcosm/graph/population.py @@ -16,6 +16,7 @@ from microcosm.frame import Frame, MassChangeRecord, WeightKind, Weights +from .canonical import canonical_json from .decl import ( MASS_POLICIES, ROWS_ALL, @@ -34,6 +35,8 @@ "dtype_for_token", "dtype_matches", "expand_lineage_receipt", + "entrant_strata_receipt", + "mass_record_receipt", "owned_ids", "patch", "population_from_frame", @@ -119,6 +122,14 @@ class MassRecord: before_by_stratum: tuple[tuple[object, float], ...] after_by_stratum: tuple[tuple[object, float], ...] entity: str | None = None + partition_entity: str | None = None + partition_column: str | None = None + before_by_partition_stratum: tuple[ + tuple[object, tuple[tuple[object, float], ...]], ... + ] = () + after_by_partition_stratum: tuple[ + tuple[object, tuple[tuple[object, float], ...]], ... + ] = () @property def before_strata(self) -> Mapping[object, float]: @@ -128,6 +139,24 @@ def before_strata(self) -> Mapping[object, float]: def after_strata(self) -> Mapping[object, float]: return MappingProxyType(dict(self.after_by_stratum)) + @property + def before_partitions(self) -> Mapping[object, Mapping[object, float]]: + return MappingProxyType( + { + partition: MappingProxyType(dict(strata)) + for partition, strata in self.before_by_partition_stratum + } + ) + + @property + def after_partitions(self) -> Mapping[object, Mapping[object, float]]: + return MappingProxyType( + { + partition: MappingProxyType(dict(strata)) + for partition, strata in self.after_by_partition_stratum + } + ) + @property def old_total(self) -> float: """Compatibility spelling used by :mod:`microcosm.frame`.""" @@ -141,6 +170,66 @@ def new_total(self) -> float: return self.after_total +def _receipt_key(value: object) -> str: + """Return the JSON-object-key spelling of a partition or stratum value.""" + + if isinstance(value, np.generic): + value = value.item() + return str(value) + + +def _partition_receipt_mapping( + values: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], +) -> dict[str, dict[str, float]]: + result: dict[str, dict[str, float]] = {} + for partition, strata in values: + partition_key = _receipt_key(partition) + if partition_key in result: + raise PopulationError( + f"Partition values collide as JSON key {partition_key!r}." + ) + converted: dict[str, float] = {} + for stratum, mass in strata: + stratum_key = _receipt_key(stratum) + if stratum_key in converted: + raise PopulationError( + f"Strata collide as JSON key {stratum_key!r} inside partition " + f"{partition_key!r}." + ) + converted[stratum_key] = float(mass) + result[partition_key] = converted + return result + + +def mass_record_receipt(record: MassRecord) -> dict[str, object]: + """Return executor-authored public mass accounting for one ledger record.""" + + payload: dict[str, object] = { + "policy": record.policy, + "before": record.before_total, + "after": record.after_total, + "stratum_before": _receipt_mass_mapping( + dict(record.before_by_stratum), label=f"Node {record.node_id!r} mass" + ), + "stratum_after": _receipt_mass_mapping( + dict(record.after_by_stratum), label=f"Node {record.node_id!r} mass" + ), + } + if record.partition_entity is not None: + assert record.partition_column is not None + payload["partition"] = { + "entity": record.partition_entity, + "column": record.partition_column, + "stratum_before": _partition_receipt_mapping( + record.before_by_partition_stratum + ), + "stratum_after": _partition_receipt_mapping( + record.after_by_partition_stratum + ), + } + return payload + + @dataclass(frozen=True) class Population: """One immutable graph view over a validated :class:`Frame`.""" @@ -272,9 +361,15 @@ def population_from_frame( ) -def _lineage_json_scalar(value: object) -> str | int | float | bool: +def _lineage_json_scalar( + value: object, *, allow_null: bool = False +) -> str | int | float | bool | None: """Detach one entity id into the scalar vocabulary accepted by receipts.""" + if pd.isna(value): + if allow_null: + return None + raise PopulationError(f"EXPAND lineage id {value!r} is not a JSON scalar.") if isinstance(value, np.generic): value = value.item() if not isinstance(value, str | int | float | bool): @@ -284,6 +379,46 @@ def _lineage_json_scalar(value: object) -> str | int | float | bool: return value +def _stratum_receipt_scalar(value: object) -> object: + """Encode one cache-safe entrant stratum label for a JSON receipt.""" + + missing = pd.isna(value) + if isinstance(missing, bool | np.bool_) and bool(missing): + raise PopulationError("EXPAND entrant stratum labels cannot be null.") + if isinstance(value, np.generic): + value = value.item() + if isinstance(value, bytes): + return {"bytes_hex": value.hex()} + if not isinstance(value, str | int | float | bool): + raise PopulationError( + f"EXPAND entrant stratum label {value!r} is not a cache-safe scalar." + ) + if isinstance(value, float) and not np.isfinite(value): + raise PopulationError(f"EXPAND entrant stratum label {value!r} is not finite.") + return value + + +def _stratum_from_receipt_scalar(value: object) -> object: + """Decode one executor-authored entrant stratum label from a receipt.""" + + if isinstance(value, Mapping): + if set(value) != {"bytes_hex"} or not isinstance(value["bytes_hex"], str): + raise PopulationError("EXPAND entrant stratum receipt label is malformed.") + encoded = value["bytes_hex"] + try: + decoded = bytes.fromhex(encoded) + except ValueError as error: + raise PopulationError( + "EXPAND entrant stratum receipt bytes are malformed." + ) from error + if encoded != decoded.hex(): + raise PopulationError( + "EXPAND entrant stratum receipt bytes are not canonical." + ) + return decoded + return _stratum_receipt_scalar(value) + + def expand_lineage_receipt( expand: Mapping[str, pd.Series], ) -> dict[str, list[list[object]]]: @@ -301,7 +436,10 @@ def expand_lineage_receipt( f"EXPAND lineage for {entity!r} is not a pandas Series." ) payload[entity] = [ - [_lineage_json_scalar(target), _lineage_json_scalar(source)] + [ + _lineage_json_scalar(target), + _lineage_json_scalar(source, allow_null=True), + ] for target, source in zip( lineage.index.tolist(), lineage.tolist(), strict=True ) @@ -309,6 +447,195 @@ def expand_lineage_receipt( return payload +def _entrant_person_ids(frame: Frame, lineage: Mapping[str, pd.Series]) -> pd.Index: + """Return null-lineage person targets in declared lineage order.""" + + person = frame.schema.person_entity + id_column = frame.schema.entity_id_column(person) + person_lineage = lineage[person] + entrant_positions = np.flatnonzero( + person_lineage.isna().to_numpy(dtype=np.bool_, copy=False) + ) + return pd.Index(person_lineage.index.take(entrant_positions), name=id_column) + + +def _validated_entrant_strata( + frame: Frame, + node: Node, + lineage: Mapping[str, pd.Series], + raw: object, +) -> pd.Series | None: + """Validate and align the iff contract for entrant-person strata.""" + + person = frame.schema.person_entity + id_column = frame.schema.entity_id_column(person) + id_dtype = frame.table(person)[id_column].dtype + entrant_ids = _entrant_person_ids(frame, lineage) + if not len(entrant_ids): + if raw is not None: + raise PopulationError( + f"EXPAND node {node.id!r} returned strata without entrant persons." + ) + return None + if raw is None: + raise PopulationError( + f"EXPAND node {node.id!r} omitted strata for entrant persons " + f"{entrant_ids[:5].tolist()}." + ) + if not isinstance(raw, pd.Series): + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata is not a Series." + ) + labels_index = pd.Index(raw.index, name=id_column) + if labels_index.nlevels != 1 or labels_index.dtype != id_dtype: + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata index must use " + f"{id_dtype!s} person ids." + ) + if not labels_index.is_unique: + raise PopulationError( + f"EXPAND node {node.id!r} repeats entrant strata person ids." + ) + if labels_index.isna().any(): + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata contains null person ids." + ) + missing = entrant_ids[~entrant_ids.isin(labels_index)] + extra = labels_index[~labels_index.isin(entrant_ids)] + if len(missing) or len(extra): + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata must name exactly the entrant " + f"persons; missing={missing[:5].tolist()}, extra={extra[:5].tolist()}." + ) + if not ( + pd.api.types.is_object_dtype(raw.dtype) or isinstance(raw.dtype, pd.StringDtype) + ): + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata must use object or string " + f"labels, got {raw.dtype!s}." + ) + if raw.isna().any(): + raise PopulationError( + f"EXPAND node {node.id!r} entrant strata contains missing labels." + ) + aligned = raw.reindex(entrant_ids).copy() + for value in aligned.array: + _stratum_receipt_scalar(value) + return aligned + + +def entrant_strata_receipt( + frame: Frame, + node: Node, + expand: Mapping[str, pd.Series], + strata: pd.Series | None, +) -> list[list[object]] | None: + """Return executor-authored entrant-person strata in lineage order.""" + + lineage = _validate_expand_lineage(frame, node, expand) + aligned = _validated_entrant_strata(frame, node, lineage, strata) + if aligned is None: + return None + return [ + [_lineage_json_scalar(target), _stratum_receipt_scalar(label)] + for target, label in zip(aligned.index, aligned.array, strict=True) + ] + + +def _cached_entrant_strata( + frame: Frame, + node: Node, + lineage: Mapping[str, pd.Series], + receipt: Mapping[str, object], +) -> pd.Series | None: + """Parse the executor-authored entrant-strata cache attestation.""" + + person = frame.schema.person_entity + id_column = frame.schema.entity_id_column(person) + id_dtype = frame.table(person)[id_column].dtype + entrant_ids = _entrant_person_ids(frame, lineage) + if not len(entrant_ids): + if "entrant_strata" in receipt: + raise PopulationError( + f"Cached EXPAND node {node.id!r} has entrant strata without " + "entrant persons." + ) + return None + raw = receipt.get("entrant_strata") + if not isinstance(raw, list): + raise PopulationError( + f"Cached EXPAND node {node.id!r} has no entrant-strata receipt." + ) + targets: list[object] = [] + labels: list[object] = [] + for entry in raw: + if not isinstance(entry, list) or len(entry) != 2: + raise PopulationError( + f"Cached EXPAND node {node.id!r} has malformed entrant strata." + ) + targets.append(entry[0]) + labels.append(_stratum_from_receipt_scalar(entry[1])) + try: + target_index = pd.Index( + pd.Series(targets, dtype=id_dtype).array, name=id_column + ) + except (TypeError, ValueError) as error: + raise PopulationError( + f"Cached EXPAND node {node.id!r} entrant strata contain invalid " + f"{id_dtype!s} person ids." + ) from error + if not target_index.equals(entrant_ids): + raise PopulationError( + f"Cached EXPAND node {node.id!r} entrant strata do not name its " + "entrant persons in lineage order." + ) + return pd.Series(labels, index=entrant_ids, dtype=object) + + +def _assert_cached_expand_strata( + before: Frame, + after: Frame, + node: Node, + lineage: Mapping[str, pd.Series], + receipt: Mapping[str, object], +) -> None: + """Verify cached incumbent, copied, and entrant person strata by lineage.""" + + person = before.schema.person_entity + id_column = before.schema.entity_id_column(person) + before_ids = pd.Index(before.table(person)[id_column], name=id_column) + person_lineage = lineage[person] + expected_ids = before_ids.append(pd.Index(person_lineage.index, name=id_column)) + after_ids = pd.Index(after.table(person)[id_column], name=id_column) + if not after_ids.equals(expected_ids): + raise PopulationError(f"Cached EXPAND node {node.id!r} reordered person ids.") + entrant_strata = _cached_entrant_strata(before, node, lineage, receipt) + expected = before.strata.astype(object).tolist() + entrant_positions: list[int] = [] + for target, source in zip(person_lineage.index, person_lineage.array, strict=True): + if pd.isna(source): + assert entrant_strata is not None + entrant_positions.append(len(expected)) + expected.append(entrant_strata.loc[target]) + continue + source_position = before_ids.get_loc(source) + expected.append(before.strata.iloc[source_position]) + actual = after.strata.astype(object).reset_index(drop=True) + if not actual.equals(pd.Series(expected, dtype=object)): + raise PopulationError( + f"Cached EXPAND node {node.id!r} strata disagree with its lineage " + "and entrant-strata receipt." + ) + for position in entrant_positions: + actual_label = _stratum_receipt_scalar(actual.iloc[position]) + expected_label = _stratum_receipt_scalar(expected[position]) + if canonical_json(actual_label) != canonical_json(expected_label): + raise PopulationError( + f"Cached EXPAND node {node.id!r} entrant stratum label " + "disagrees with its receipt." + ) + + def _expand_lineage_from_receipt( frame: Frame, node: Node, @@ -340,11 +667,21 @@ def _expand_lineage_from_receipt( ) id_column = frame.schema.entity_id_column(entity) dtype = frame.table(entity)[id_column].dtype + source_dtype: object = dtype + has_null_source = any(pd.isna(value) for value in sources) + if has_null_source and pd.api.types.is_bool_dtype(dtype): + source_dtype = pd.BooleanDtype() + elif has_null_source and pd.api.types.is_integer_dtype(dtype): + numpy_dtype = np.dtype(getattr(dtype, "numpy_dtype", dtype)) + prefix = "UInt" if np.issubdtype(numpy_dtype, np.unsignedinteger) else "Int" + source_dtype = pd.api.types.pandas_dtype( + f"{prefix}{numpy_dtype.itemsize * 8}" + ) lineage[entity] = pd.Series( sources, index=pd.Index(pd.Series(targets, dtype=dtype).array, name=id_column), name=id_column, - dtype=dtype, + dtype=source_dtype, ) return lineage @@ -378,7 +715,17 @@ def _validate_expand_lineage( id_column = frame.schema.entity_id_column(entity) source_ids = pd.Index(frame.table(entity)[id_column], name=id_column) targets = pd.Index(lineage.index, name=id_column) - if targets.dtype != source_ids.dtype or lineage.dtype != source_ids.dtype: + source_is_null = lineage.isna().to_numpy(dtype=np.bool_, copy=False) + nullable_sources = bool(source_is_null.any()) + nullable_source_dtype = getattr(lineage.dtype, "numpy_dtype", None) + if targets.dtype != source_ids.dtype or ( + lineage.dtype != source_ids.dtype + and not ( + nullable_sources + and nullable_source_dtype is not None + and np.dtype(nullable_source_dtype) == np.dtype(source_ids.dtype) + ) + ): raise PopulationError( f"EXPAND node {node.id!r} lineage for {entity!r} must use " f"{source_ids.dtype!s} ids for both targets and sources." @@ -387,9 +734,15 @@ def _validate_expand_lineage( raise PopulationError( f"EXPAND node {node.id!r} repeats new target {entity!r} ids." ) - if targets.isna().any() or lineage.isna().any(): + if targets.isna().any(): raise PopulationError( - f"EXPAND node {node.id!r} lineage for {entity!r} contains null ids." + f"EXPAND node {node.id!r} lineage for {entity!r} contains null " + "target ids." + ) + if nullable_sources and not node.entrants: + raise PopulationError( + f"EXPAND node {node.id!r} lineage for {entity!r} contains null " + "source ids without entrants=True." ) collisions = targets.intersection(source_ids) if len(collisions): @@ -397,31 +750,41 @@ def _validate_expand_lineage( f"EXPAND node {node.id!r} lineage target {entity!r} ids collide " f"with incumbents {collisions[:5].tolist()}." ) - source_positions = source_ids.get_indexer(lineage.to_numpy(copy=False)) + source_positions = np.full(len(lineage), -1, dtype=np.int64) + copied = ~source_is_null + source_positions[copied] = source_ids.get_indexer( + lineage.iloc[np.flatnonzero(copied)].to_numpy(copy=False) + ) if (source_positions < 0).any(): - bad = lineage.iloc[np.flatnonzero(source_positions < 0)[:5]].tolist() - raise PopulationError( - f"EXPAND node {node.id!r} lineage names unknown {entity!r} " - f"source ids {bad}." - ) + unknown = copied & (source_positions < 0) + if unknown.any(): + bad = lineage.iloc[np.flatnonzero(unknown)[:5]].tolist() + raise PopulationError( + f"EXPAND node {node.id!r} lineage names unknown {entity!r} " + f"source ids {bad}." + ) if after is not None: after_ids = pd.Index(after.table(entity)[id_column], name=id_column) if not source_ids.isin(after_ids).all(): raise PopulationError( f"Cached EXPAND node {node.id!r} dropped incumbent {entity!r} ids." ) - additions = after_ids[~after_ids.isin(source_ids)] - if not additions.equals(targets): + expected_ids = source_ids.append(targets) + if not after_ids.equals(expected_ids): raise PopulationError( - f"Cached EXPAND node {node.id!r} frame additions for " - f"{entity!r} disagree with its lineage receipt." + f"Cached EXPAND node {node.id!r} final {entity!r} ids " + "disagree with its lineage receipt." ) validated[entity] = lineage return validated def restore_cached_expand( - population: Population, node: Node, result: KernelResult + population: Population, + node: Node, + result: KernelResult, + *, + mass_partition: tuple[str, str] | None = None, ) -> Population: """Restore a previously validated EXPAND frame against its keyed base. @@ -434,6 +797,11 @@ def restore_cached_expand( if node.structural is not StructuralDelta.EXPAND or result.frame is None: raise PopulationError("restore_cached_expand requires an EXPAND Frame.") + if result.strata is not None: + raise PopulationError( + f"Cached EXPAND node {node.id!r} returned kernel strata instead of " + "its executor frame artifact." + ) frame = result.frame if frame.schema != population.frame.schema: raise PopulationError(f"Cached EXPAND node {node.id!r} changed schema.") @@ -443,6 +811,7 @@ def restore_cached_expand( lineage = _validate_expand_lineage( population.frame, node, receipt_lineage, after=frame ) + _assert_cached_expand_strata(population.frame, frame, node, lineage, result.receipt) _assert_expand_weights(population, frame, node, result) design_weights: dict[str, np.ndarray] = {} @@ -457,18 +826,46 @@ def restore_cached_expand( values[retained] = old_anchor[positions[retained]] if not retained.all(): sources = lineage[entity].reindex(after_ids[~retained]) - source_positions = before_ids.get_indexer(sources.to_numpy(copy=False)) + source_is_null = sources.isna().to_numpy(dtype=np.bool_, copy=False) + introduced_positions = np.flatnonzero(~retained) + copied_positions = introduced_positions[~source_is_null] + source_positions = before_ids.get_indexer( + sources.iloc[np.flatnonzero(~source_is_null)].to_numpy(copy=False) + ) if (source_positions < 0).any(): # defended by lineage validation raise PopulationError( f"Cached EXPAND node {node.id!r} has unknown design lineage " f"for new {entity!r} ids." ) - values[~retained] = old_anchor[source_positions] + values[copied_positions] = old_anchor[source_positions] + entrant_positions = introduced_positions[source_is_null] + if len(entrant_positions): + try: + current = frame.weights_for(entity) + except ValueError as error: + raise PopulationError( + f"Cached EXPAND node {node.id!r} has no design anchor for " + f"entrant {entity!r} ids." + ) from error + if current.kind is not WeightKind.DESIGN: + raise PopulationError( + f"Cached EXPAND node {node.id!r} cannot anchor entrant " + f"{entity!r} ids from {current.kind.value!r} weights; " + "explicit design weights are required." + ) + values[entrant_positions] = current.values[entrant_positions] design_weights[entity] = values ledger = ( *population.mass_ledger, - _mass_record(population.frame, frame, node, result, _mass_policy(node)), + _mass_record( + population.frame, + frame, + node, + result, + _mass_policy(node), + mass_partition=mass_partition, + ), ) owners = { (entity, str(column)): node.id @@ -531,7 +928,13 @@ def storage_equal( return _storage_parts(left, selected) == _storage_parts(right, selected) -def patch(population: Population, node: Node, result: KernelResult) -> Population: +def patch( + population: Population, + node: Node, + result: KernelResult, + *, + mass_partition: tuple[str, str] | None = None, +) -> Population: """Validate and apply one node result without mutating ``population``. ``EXPAND`` kernels return only the new-id to source-id mapping through @@ -554,6 +957,10 @@ def patch(population: Population, node: Node, result: KernelResult) -> Populatio _assert_no_ordinary_structural_outputs(population, node) expected_columns = {(owned.entity, owned.column) for owned in node.outputs} lineage_expand = node.structural is StructuralDelta.EXPAND and result.frame is None + if result.strata is not None and not lineage_expand: + raise PopulationError( + f"Node {node.id!r} returned entrant strata outside a lineage EXPAND." + ) if not lineage_expand and set(result.columns) != expected_columns: raise PopulationError( f"Node {node.id!r} returned columns {sorted(result.columns)}; " @@ -602,7 +1009,14 @@ def patch(population: Population, node: Node, result: KernelResult) -> Populatio ledger = population.mass_ledger if node.structural is not StructuralDelta.NONE or node.weights is not None: policy = _mass_policy(node) - record = _mass_record(before, frame, node, result, policy) + record = _mass_record( + before, + frame, + node, + result, + policy, + mass_partition=mass_partition, + ) ledger = (*ledger, record) return Population.from_frame( @@ -658,6 +1072,8 @@ def _targets_by_source(lineage: pd.Series) -> dict[object, list[object]]: grouped: dict[object, list[object]] = {} for target, source in zip(lineage.index, lineage.array, strict=True): + if pd.isna(source): + continue grouped.setdefault(source, []).append(target) return grouped @@ -680,8 +1096,10 @@ def _remap_expand_memberships( source_person = before.table(person) person_id = before.schema.entity_id_column(person) source_person_ids = pd.Index(source_person[person_id]) + entrant_mask = person_lineage.isna().to_numpy(dtype=np.bool_, copy=False) + copied_indices = np.flatnonzero(~entrant_mask) source_positions = source_person_ids.get_indexer( - person_lineage.to_numpy(copy=False) + person_lineage.iloc[copied_indices].to_numpy(copy=False) ) if (source_positions < 0).any(): # defended by lineage validation raise PopulationError( @@ -708,16 +1126,24 @@ def _remap_expand_memberships( ) seen: dict[object, int] = {} - remapped: list[object] = [] - for source_position, source_person_id in zip( - source_positions, person_lineage.array, strict=True + remapped = ( + tables[person][membership] + .iloc[len(source_person) :] + .reset_index(drop=True) + .copy() + ) + for addition_position, source_position, source_person_id in zip( + copied_indices, + source_positions, + person_lineage.iloc[copied_indices].array, + strict=True, ): # Select the membership Series directly. Selecting a mixed-type # DataFrame row can coerce a large integer group id through float. source_group = source_person[membership].iloc[source_position] candidates = group_targets.get(source_group, []) if not candidates: - remapped.append(source_group) + remapped.iloc[addition_position] = source_group continue ordinal = seen.get(source_person_id, 0) if ordinal >= len(candidates): @@ -725,13 +1151,12 @@ def _remap_expand_memberships( f"EXPAND node {node.id!r} cannot align {membership!r} for " f"copied person {source_person_id!r}." ) - remapped.append(candidates[ordinal]) + remapped.iloc[addition_position] = candidates[ordinal] seen[source_person_id] = ordinal + 1 carried = source_person[membership].reset_index(drop=True) - additions = pd.Series(remapped, dtype=source_person[membership].dtype) tables[person][membership] = pd.concat( - [carried, additions], ignore_index=True + [carried, remapped], ignore_index=True ).array @@ -746,11 +1171,17 @@ def _patch_expand( f"EXPAND node {node.id!r} cannot yet carry association link tables." ) cells = _expand_cells(node) - for entity, _, _ in cells: + for entity, column, _ in cells: if entity not in before.entities: raise PopulationError( f"EXPAND node {node.id!r} names unknown entity {entity!r}." ) + id_column = before.schema.entity_id_column(entity) + if column == id_column: + raise PopulationError( + f"EXPAND node {node.id!r} cannot overlay entity id column " + f"{entity}.{column}; lineage supplies final ids." + ) cell_coordinates = {(entity, column) for entity, column, _ in cells} if set(result.columns) != cell_coordinates: @@ -761,7 +1192,6 @@ def _patch_expand( lineage = _validate_expand_lineage(before, node, result.expand) - tables: dict[str, pd.DataFrame] = {} lineage_positions: dict[str, np.ndarray] = {} target_ids: dict[str, pd.Index] = {} for entity in before.entities: @@ -772,29 +1202,40 @@ def _patch_expand( source_table[id_column].to_numpy(copy=True), name=id_column ) new_targets = pd.Index(entity_lineage.index, name=id_column) - targets = source_ids.append(new_targets) - source_positions = source_ids.get_indexer(entity_lineage.to_numpy(copy=False)) - positions = np.concatenate( - [np.arange(len(source_ids), dtype=np.int64), source_positions] + target_ids[entity] = source_ids.append(new_targets) + entrants = entity_lineage.isna().to_numpy(dtype=np.bool_, copy=False) + source_positions = np.full(len(entity_lineage), -1, dtype=np.int64) + copied = ~entrants + source_positions[copied] = source_ids.get_indexer( + entity_lineage.iloc[np.flatnonzero(copied)].to_numpy(copy=False) ) - carried = source_table.iloc[positions].reset_index(drop=True) - replacement_ids = pd.Series( - targets.to_numpy(copy=True), dtype=source_table[id_column].dtype + lineage_positions[entity] = np.concatenate( + [np.arange(len(source_ids), dtype=np.int64), source_positions] ) - if len(replacement_ids) != len(carried): - raise PopulationError( - f"EXPAND node {node.id!r} lineage index/value lengths disagree " - f"for {entity!r}." - ) - carried[id_column] = replacement_ids.array - tables[entity] = carried - lineage_positions[entity] = positions - target_ids[entity] = targets - _remap_expand_memberships(before, tables, lineage, node) + if entrants.any(): + carried = { + (entity, str(column)) + for column in source_table.columns + if column != id_column + } + missing = sorted(carried - cell_coordinates) + if missing: + names = [ + f"{carried_entity}.{column}" for carried_entity, column in missing + ] + raise PopulationError( + f"EXPAND node {node.id!r} entrant rows do not materialize " + f"carried columns {names}." + ) + person = before.schema.person_entity + entrant_strata = _validated_entrant_strata(before, node, lineage, result.strata) + + aligned_cells: dict[tuple[str, str], pd.Series] = {} for entity, column, dtype in cells: - incoming = result.columns[(entity, column)] + coordinate = (entity, column) + incoming = result.columns[coordinate] if not isinstance(incoming, pd.Series): raise PopulationError( f"EXPAND node {node.id!r} cell {entity}.{column} is not a Series." @@ -815,8 +1256,63 @@ def _patch_expand( dtype, label=f"EXPAND node {node.id!r} cell {entity}.{column}", ) + source_table = before.table(entity) + if column in source_table: + carried_dtype = token_for_dtype(source_table[column].dtype) + if dtype != carried_dtype: + raise PopulationError( + f"EXPAND node {node.id!r} carried cell {entity}.{column} " + f"declares {dtype!r}; its incumbent dtype is {carried_dtype!r}." + ) + incumbent = aligned.iloc[: len(source_table)].reset_index(drop=True) + if not storage_equal(source_table[column], incumbent): + raise PopulationError( + f"EXPAND node {node.id!r} changed carried storage in " + f"{entity}.{column} for incumbent rows." + ) + aligned_cells[coordinate] = aligned + + tables: dict[str, pd.DataFrame] = {} + for entity in before.entities: + id_column = before.schema.entity_id_column(entity) + source_table = before.table(entity) + positions = lineage_positions[entity] + addition_positions = positions[len(source_table) :] + if len(source_table): + additions = source_table.iloc[ + np.maximum(addition_positions, 0) + ].reset_index(drop=True) + else: + additions = source_table.reindex(range(len(addition_positions))).copy() + carried = pd.concat( + [source_table.reset_index(drop=True), additions], ignore_index=True + ) + replacement_ids = pd.Series( + target_ids[entity].to_numpy(copy=True), + dtype=source_table[id_column].dtype, + ) + if len(replacement_ids) != len(carried): + raise PopulationError( + f"EXPAND node {node.id!r} lineage index/value lengths disagree " + f"for {entity!r}." + ) + carried[id_column] = replacement_ids.array + tables[entity] = carried + + _remap_expand_memberships(before, tables, lineage, node) + + for (entity, column), aligned in aligned_cells.items(): tables[entity][column] = aligned.array + for entity, expected_ids in target_ids.items(): + id_column = before.schema.entity_id_column(entity) + final_ids = pd.Index(tables[entity][id_column], name=id_column) + if not final_ids.equals(expected_ids): + raise PopulationError( + f"EXPAND node {node.id!r} final {entity!r} ids disagree with " + "its lineage targets after cell overlays." + ) + weight_entity = _expand_weight_entity(node) assert weight_entity is not None if weight_entity not in before.weighted_entities: @@ -832,16 +1328,39 @@ def _patch_expand( weights[entity] = result.weights continue old = before.weights_for(entity) - weights[entity] = Weights(old.values[lineage_positions[entity]], kind=old.kind) + positions = lineage_positions[entity] + if (positions < 0).any(): + raise PopulationError( + f"EXPAND node {node.id!r} cannot admit entrants on weighted " + f"entity {entity!r}; only {weight_entity!r} has materialized weights." + ) + weights[entity] = Weights(old.values[positions], kind=old.kind) - person = before.schema.person_entity person_positions = lineage_positions[person] - strata = pd.Series( - before.strata.iloc[person_positions].array.copy(), - index=tables[person].index, - name=before.strata.name, - dtype=before.strata.dtype, - ) + if entrant_strata is None: + strata = pd.Series( + before.strata.iloc[person_positions].array.copy(), + index=tables[person].index, + name=before.strata.name, + dtype=before.strata.dtype, + ) + else: + additions: list[object] = [] + entrant_values = iter(entrant_strata.array) + for source_position in person_positions[len(before.table(person)) :]: + additions.append( + next(entrant_values) + if source_position < 0 + else before.strata.iloc[source_position] + ) + strata = pd.concat( + [ + before.strata.astype(object).reset_index(drop=True), + pd.Series(additions, dtype=object), + ], + ignore_index=True, + ) + strata.name = before.strata.name frame = Frame( tables, before.schema, @@ -1210,13 +1729,11 @@ def _carry_design_weights( introduced = before_positions < 0 if introduced.any(): sources = lineage.reindex(after_ids[introduced]) - if sources.isna().any(): - raise PopulationError( - f"EXPAND node {node.id!r} has incomplete design lineage " - f"for {entity!r}." - ) - before_positions[introduced] = before_ids.get_indexer( - sources.to_numpy(copy=False) + source_is_null = sources.isna().to_numpy(dtype=np.bool_, copy=False) + introduced_positions = np.flatnonzero(introduced) + copied_positions = introduced_positions[~source_is_null] + before_positions[copied_positions] = before_ids.get_indexer( + sources.iloc[np.flatnonzero(~source_is_null)].to_numpy(copy=False) ) values = np.empty(len(after_ids), dtype=np.float64) retained = before_positions >= 0 @@ -1455,6 +1972,8 @@ def _mass_record( node: Node, result: KernelResult, policy: str, + *, + mass_partition: tuple[str, str] | None = None, ) -> MassRecord: if policy not in MASS_POLICIES: raise PopulationError(f"Node {node.id!r} has unknown mass policy {policy!r}.") @@ -1464,7 +1983,18 @@ def _mass_record( after_pairs = tuple((key, float(value)) for key, value in after_mass.items()) before_total = float(before_mass.sum()) after_total = float(after_mass.sum()) + before_partition: tuple[tuple[object, tuple[tuple[object, float], ...]], ...] = () + after_partition: tuple[tuple[object, tuple[tuple[object, float], ...]], ...] = () + if mass_partition is not None: + before_partition = _mass_by_partition(before, mass_partition, node.id) + after_partition = _mass_by_partition(after, mass_partition, node.id) if policy == "conserve": + if mass_partition is not None: + _assert_partition_mass_mapping( + before_partition, + after_partition, + label=f"Node {node.id!r} mass='conserve'", + ) _assert_mass_mapping( dict(before_pairs), dict(after_pairs), @@ -1481,6 +2011,9 @@ def _mass_record( before=dict(before_pairs), after=dict(after_pairs), node_id=node.id, + mass_partition=mass_partition, + before_partition=before_partition, + after_partition=after_partition, ) elif policy == "declared": raise PopulationError( @@ -1507,6 +2040,89 @@ def _mass_record( and "expand_weight_entity" in node.params else None ), + partition_entity=(None if mass_partition is None else mass_partition[0]), + partition_column=(None if mass_partition is None else mass_partition[1]), + before_by_partition_stratum=before_partition, + after_by_partition_stratum=after_partition, + ) + + +def _partition_values_on_person( + frame: Frame, + mass_partition: tuple[str, str], + node_id: str, +) -> pd.Series: + entity, column = mass_partition + if entity not in frame.entities: + raise PopulationError( + f"Node {node_id!r} mass partition names unknown entity {entity!r}." + ) + table = frame.table(entity) + if column not in table: + raise PopulationError( + f"Node {node_id!r} mass partition column {entity}.{column} is " + "missing at run time." + ) + person = frame.schema.person_entity + if entity == person: + return table[column].reset_index(drop=True) + if entity not in frame.schema.group_entities: + raise PopulationError( + f"Node {node_id!r} cannot broadcast mass partition entity {entity!r} " + "to persons." + ) + id_column = frame.schema.entity_id_column(entity) + membership = frame.schema.membership_column(entity) + positions = pd.Index(table[id_column]).get_indexer( + frame.table(person)[membership].to_numpy(copy=False) + ) + if (positions < 0).any(): # defended by Frame linkage validation + raise PopulationError( + f"Node {node_id!r} cannot align mass partition {entity}.{column} " + "to person memberships." + ) + return table[column].iloc[positions].reset_index(drop=True) + + +def _mass_by_partition( + frame: Frame, + mass_partition: tuple[str, str], + node_id: str, +) -> tuple[tuple[object, tuple[tuple[object, float], ...]], ...]: + partition = _partition_values_on_person(frame, mass_partition, node_id) + person = frame.schema.person_entity + weights = frame.resolve_weights(person).values + strata = frame.strata.reset_index(drop=True) + valid = partition.notna().to_numpy(dtype=np.bool_, copy=False) + if not valid.any(): + return () + grouped = ( + pd.DataFrame( + { + "_partition": partition.loc[valid].reset_index(drop=True), + "_stratum": strata.loc[valid].reset_index(drop=True), + "_mass": weights[valid], + } + ) + .groupby(["_partition", "_stratum"], observed=True, sort=False)["_mass"] + .sum() + ) + nested: dict[object, dict[object, float]] = {} + for (partition_value, stratum), mass in grouped.items(): + nested.setdefault(partition_value, {})[stratum] = float(mass) + return tuple( + ( + partition_value, + tuple( + sorted( + strata_mass.items(), + key=lambda item: _receipt_key(item[0]), + ) + ), + ) + for partition_value, strata_mass in sorted( + nested.items(), key=lambda item: _receipt_key(item[0]) + ) ) @@ -1519,6 +2135,9 @@ def _validate_mass_receipt( before: Mapping[object, float], after: Mapping[object, float], node_id: str, + mass_partition: tuple[str, str] | None, + before_partition: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + after_partition: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], ) -> None: if not isinstance(raw, Mapping): raise PopulationError(f"Node {node_id!r} receipt['mass'] must be a mapping.") @@ -1535,6 +2154,92 @@ def _validate_mass_receipt( _assert_receipt_mapping( raw.get("stratum_after"), after, f"Node {node_id!r} mass.stratum_after" ) + if mass_partition is None and "partition" in raw: + raise PopulationError( + f"Node {node_id!r} mass.partition is present but the graph " + "declares no mass partition." + ) + if mass_partition is not None and "partition" in raw: + _validate_partition_mass_receipt( + raw["partition"], + mass_partition=mass_partition, + before=before_partition, + after=after_partition, + node_id=node_id, + ) + + +def _validate_partition_mass_receipt( + raw: object, + *, + mass_partition: tuple[str, str], + before: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + after: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + node_id: str, +) -> None: + if not isinstance(raw, Mapping) or set(raw) != { + "entity", + "column", + "stratum_before", + "stratum_after", + }: + raise PopulationError( + f"Node {node_id!r} mass.partition must contain entity, column, " + "stratum_before, and stratum_after." + ) + entity, column = mass_partition + if raw.get("entity") != entity or raw.get("column") != column: + raise PopulationError( + f"Node {node_id!r} mass.partition names " + f"{raw.get('entity')}.{raw.get('column')}; expected {entity}.{column}." + ) + _assert_partition_receipt_mapping( + raw.get("stratum_before"), + before, + label=f"Node {node_id!r} mass.partition.stratum_before", + ) + _assert_partition_receipt_mapping( + raw.get("stratum_after"), + after, + label=f"Node {node_id!r} mass.partition.stratum_after", + ) + + +def _assert_partition_receipt_mapping( + observed: object, + expected: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + *, + label: str, +) -> None: + if not isinstance(observed, Mapping): + raise PopulationError(f"{label} must be a mapping.") + converted = _partition_receipt_mapping(expected) + if set(observed) != set(converted): + raise PopulationError( + f"{label} changed partitions: expected {list(converted)}, " + f"got {list(observed)}." + ) + for partition, strata in converted.items(): + _assert_receipt_mapping( + observed[partition], strata, f"{label} partition {partition!r}" + ) + + +def _assert_partition_mass_mapping( + expected: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + observed: tuple[tuple[object, tuple[tuple[object, float], ...]], ...], + *, + label: str, +) -> None: + before = {partition: dict(strata) for partition, strata in expected} + after = {partition: dict(strata) for partition, strata in observed} + partitions = sorted(set(before) | set(after), key=_receipt_key) + for partition in partitions: + _assert_mass_mapping( + before.get(partition, {}), + after.get(partition, {}), + label=f"{label} partition {_receipt_key(partition)!r}", + ) def _assert_close(observed: object, expected: float, label: str) -> None: @@ -1551,8 +2256,23 @@ def _assert_receipt_mapping( ) -> None: if not isinstance(observed, Mapping): raise PopulationError(f"{label} must be a mapping.") - converted = {key: float(value) for key, value in observed.items()} - _assert_mass_mapping(expected, converted, label=label) + expected_json = _receipt_mass_mapping(expected, label=label) + observed_json = _receipt_mass_mapping(observed, label=label) + _assert_mass_mapping(expected_json, observed_json, label=label) + + +def _receipt_mass_mapping( + values: Mapping[object, object], *, label: str +) -> dict[str, float]: + """Normalize a mass mapping to its stable JSON-object-key representation.""" + + result: dict[str, float] = {} + for raw_key, value in values.items(): + key = _receipt_key(raw_key) + if key in result: + raise PopulationError(f"{label} has colliding JSON key {key!r}.") + result[key] = float(value) + return result def _assert_mass_mapping( diff --git a/packages/microcosm-graph/src/microcosm/graph/view.py b/packages/microcosm-graph/src/microcosm/graph/view.py index 98c3f0a7b..adb8d802b 100644 --- a/packages/microcosm-graph/src/microcosm/graph/view.py +++ b/packages/microcosm-graph/src/microcosm/graph/view.py @@ -91,6 +91,16 @@ def describe( 'Seed: int.from_bytes(sha256(b"seed\\0" + node_key)[:8], "little")' ) else: + tolerance = run_receipt.capabilities.tolerance + tolerance_text = canonical_json( + None + if tolerance is None + else { + "rtol": tolerance.rtol, + "atol": tolerance.atol, + "ulps": tolerance.ulps, + } + ).decode("utf-8") lines.extend( [ 'Seed: int.from_bytes(sha256(b"seed\\0" + node_key)[:8], ' @@ -102,7 +112,8 @@ def describe( f"numeric={_value(run_receipt.capabilities.numeric)}, " f"seed={_value(run_receipt.capabilities.seed_source)}, " f"structural={_value(run_receipt.capabilities.structural)}, " - f"consumes_se={run_receipt.capabilities.consumes_se}", + f"consumes_se={run_receipt.capabilities.consumes_se}, " + f"tolerance={tolerance_text}", "Receipt: " + canonical_json(run_receipt.receipt).decode("utf-8"), ] ) diff --git a/packages/microcosm-graph/tests/fixtures/parity/kernels/fit.qrf/pins.json b/packages/microcosm-graph/tests/fixtures/parity/kernels/fit.qrf/pins.json index 3b8d76dad..491d7ed00 100644 --- a/packages/microcosm-graph/tests/fixtures/parity/kernels/fit.qrf/pins.json +++ b/packages/microcosm-graph/tests/fixtures/parity/kernels/fit.qrf/pins.json @@ -1 +1 @@ -{"dependencies":{"numpy":"2.4.6","pandas":"3.0.3","quantile-forest":"1.4.2","scikit-learn":"1.8.0"},"implementation_hash":"3a0fabca2f9bedf98c33846e7b6d59825f17007933b78ec5aeb31d3f0a2706e7","kernel":"fit.qrf@1","node":"fit_qrf","seed":947} +{"dependencies":{"numpy":"2.4.6","pandas":"3.0.3","quantile-forest":"1.4.2","scikit-learn":"1.8.0"},"implementation_hash":"9b412a44b3e9d2cfc44b6ed2635e46edefc09c942af10235c56b0839cac906fd","kernel":"fit.qrf@1","node":"fit_qrf","seed":947} diff --git a/packages/microcosm-graph/tests/test_acceptance_b_ownership.py b/packages/microcosm-graph/tests/test_acceptance_b_ownership.py index 2d0d6415a..cf331b7b8 100644 --- a/packages/microcosm-graph/tests/test_acceptance_b_ownership.py +++ b/packages/microcosm-graph/tests/test_acceptance_b_ownership.py @@ -244,7 +244,6 @@ def test_b5_null_means_absence(tmp_path: Path) -> None: toy.run_toy(liar, tmp_path / "liar") -@pytest.mark.xfail(strict=True, reason="charter B6: entrant execution pending") def test_b6_entrants_are_declared(tmp_path: Path) -> None: """Null lineage is an explicit, complete, and receipted entrant contract. @@ -311,7 +310,6 @@ def test_b6_entrants_are_declared(tmp_path: Path) -> None: assert "household_size" in str(error.value) -@pytest.mark.xfail(strict=True, reason="charter B7: entrant person strata pending") def test_b7_entrant_persons_carry_their_stratum(tmp_path: Path) -> None: """An entrant person's stratum arrives through ``KernelResult.strata``. diff --git a/packages/microcosm-graph/tests/test_acceptance_c_seeds.py b/packages/microcosm-graph/tests/test_acceptance_c_seeds.py index 75f3eb62c..d02def8b6 100644 --- a/packages/microcosm-graph/tests/test_acceptance_c_seeds.py +++ b/packages/microcosm-graph/tests/test_acceptance_c_seeds.py @@ -182,7 +182,6 @@ def test_c4_seed_from_identity(tmp_path: Path) -> None: assert len(set(elsewhere.seeds().values())) == len(elsewhere.seeds()) -@pytest.mark.xfail(strict=True, reason="charter C5: tolerance propagation pending") def test_c5_tolerance_is_declared(tmp_path: Path) -> None: """Receipts and readers carry an owner's exact declared tolerance. diff --git a/packages/microcosm-graph/tests/test_acceptance_d_weights.py b/packages/microcosm-graph/tests/test_acceptance_d_weights.py index cc4817e62..17bbc9284 100644 --- a/packages/microcosm-graph/tests/test_acceptance_d_weights.py +++ b/packages/microcosm-graph/tests/test_acceptance_d_weights.py @@ -221,7 +221,6 @@ def test_d5_uncertainty_travels(tmp_path: Path) -> None: assert toy.calibrated_node(kernel="calibrate.blind@1").params["target_se"] == 2500.0 -@pytest.mark.xfail(strict=True, reason="charter D6: partitioned mass pending") def test_d6_mass_is_partitioned(tmp_path: Path) -> None: """Mass is conserved and receipted inside every partition value. diff --git a/packages/microcosm-graph/tests/test_graph_acceptance_burndown.py b/packages/microcosm-graph/tests/test_graph_acceptance_burndown.py index 172acb70e..d86764883 100644 --- a/packages/microcosm-graph/tests/test_graph_acceptance_burndown.py +++ b/packages/microcosm-graph/tests/test_graph_acceptance_burndown.py @@ -271,18 +271,11 @@ def test_the_real_suite_is_all_strict_and_all_accounted_for() -> None: """The tool's own checks, run against the suite it exists to score.""" data = burndown.report(burndown.counts(burndown.suite_files())) root = burndown.ROOT - assert data["total"] == 4 + assert data["total"] == 0 assert not [entry for entry in data["properties"] if entry["state"] == "missing"] states = {entry["id"]: entry["state"] for entry in data["properties"]} - pending = {"B6", "B7", "C5", "D6"} # amendments 11-14, flipped by their lanes - assert { - identifier for identifier, state in states.items() if state == "red" - } == pending - assert all( - state == "green" - for identifier, state in states.items() - if identifier not in pending - ) + assert not {identifier for identifier, state in states.items() if state == "red"} + assert all(state == "green" for state in states.values()) for entry in data["files"]: source = (root / entry["file"]).read_text() for marker in markers_in(source, entry["file"]): diff --git a/packages/microcosm-graph/tests/test_graph_executor.py b/packages/microcosm-graph/tests/test_graph_executor.py index 8da32936b..8c7f55b58 100644 --- a/packages/microcosm-graph/tests/test_graph_executor.py +++ b/packages/microcosm-graph/tests/test_graph_executor.py @@ -11,6 +11,7 @@ import pandas as pd import pytest +import microcosm.graph.executor as graph_executor from microcosm.frame import EntitySchema, Frame, WeightKind, Weights from microcosm.graph.decl import ( Graph, @@ -31,6 +32,8 @@ KernelRegistry, KernelResult, KernelRole, + Numeric, + Tolerance, ) from microcosm.graph.manifest import Decision from microcosm.graph.store import ( @@ -667,10 +670,13 @@ def cross_entity(context: KernelContext) -> KernelResult: ) +@pytest.mark.parametrize("explicit_rewrite_input", [False, True]) def test_rewrite_incumbent_is_projected_from_its_owned_declaration( tmp_path: Path, + explicit_rewrite_input: bool, ) -> None: source = _source_path(tmp_path / "source") + boundary_tolerance = Tolerance(atol=1e-6) def keep_all(context: KernelContext) -> KernelResult: person = context.tables["person"] @@ -686,6 +692,11 @@ def rewrite(context: KernelContext) -> KernelResult: "age", "income", } + assert context.tolerances == { + ("person", "age"): boundary_tolerance, + ("person", "income"): boundary_tolerance, + } + return KernelResult( columns={ ("person", "income"): pd.Series( @@ -706,7 +717,9 @@ def rewrite(context: KernelContext) -> KernelResult: rewriter = Node( "rewrite_income", "rewrite.income@1", - inputs=(Slice("person", ("age",)),), + inputs=( + Slice("person", ("age", "income") if explicit_rewrite_input else ("age",)), + ), outputs=(Owned("person", "income", "float64", rewrite=True),), population=boundary.id, ) @@ -714,7 +727,12 @@ def rewrite(context: KernelContext) -> KernelResult: registry.register( _Kernel( boundary.kernel, - Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.FILTER), + Capabilities( + Determinism.DETERMINISTIC, + numeric=Numeric.TOLERANCE_BOUND, + structural=StructuralDelta.FILTER, + tolerance=boundary_tolerance, + ), keep_all, ) ) @@ -729,7 +747,7 @@ def rewrite(context: KernelContext) -> KernelResult: manifest = _run( Graph("toy", (SOURCE,), (CREATE, boundary, rewriter)), source, - ContentStore(tmp_path / "store"), + ContentStore(tmp_path / f"store-{explicit_rewrite_input}"), registry, ) @@ -889,6 +907,238 @@ def claim(context: KernelContext) -> KernelResult: assert claim_kernel.calls == 1 +def test_expand_id_overlay_is_rejected_without_committing_cache(tmp_path: Path) -> None: + source = _source_path(tmp_path / "source") + + def replace_lineage_id(context: KernelContext) -> KernelResult: + return KernelResult( + expand={ + "person": pd.Series( + [1], index=pd.Index([4], name="person_id"), dtype="int64" + ), + "household": pd.Series( + [10], index=pd.Index([30], name="household_id"), dtype="int64" + ), + }, + columns={ + ("person", "person_household_id"): pd.Series( + [10, 10, 20, 40], + index=pd.Index([1, 2, 3, 4], name="person_id"), + dtype="int64", + ), + ("household", "household_id"): pd.Series( + [10, 20, 40], + index=pd.Index([10, 20, 30], name="household_id"), + dtype="int64", + ), + }, + weights=Weights( + np.array([1.0, 2.0, 1.0], dtype=np.float64), WeightKind.DESIGN + ), + ) + + expand = Node( + "replace_lineage_id", + "bad.expand@1", + structural=StructuralDelta.EXPAND, + base="survey", + params={ + "expand_cells": ( + ("person", "person_household_id", "int64"), + ("household", "household_id", "int64"), + ), + "expand_weight_entity": "household", + "expand_weight_kind": "design", + }, + mass="free", + ) + kernel = _Kernel( + expand.kernel, + Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.EXPAND), + replace_lineage_id, + ) + registry = _registry(extra=kernel) + graph = Graph("toy", (SOURCE,), (CREATE, expand)) + store = ContentStore(tmp_path / "store") + + for _ in range(2): + with pytest.raises(NodeRejected, match="cannot overlay entity id column"): + _run(graph, source, store, registry) + if kernel.calls == 1: + first_store_bytes = _object_bytes(store) + else: + assert _object_bytes(store) == first_store_bytes + assert kernel.calls == 2 + + +def test_partitioned_graph_accepts_structural_entrant_partition_values( + tmp_path: Path, +) -> None: + source = _source_path(tmp_path / "source") + + def create_partitioned(context: KernelContext) -> KernelResult: + original = _source_frame(context.sources["survey"]) + tables = {entity: original.table(entity).copy() for entity in original.entities} + tables["household"]["period"] = np.array([2024, 2025], dtype=np.int64) + return KernelResult( + frame=Frame( + tables, + original.schema, + {"household": original.weights_for("household")}, + original.strata, + ) + ) + + def admit_household(context: KernelContext) -> KernelResult: + return KernelResult( + expand={ + "person": pd.Series( + [1], index=pd.Index([4], name="person_id"), dtype="int64" + ), + "household": pd.Series( + [pd.NA], + index=pd.Index([30], name="household_id"), + dtype="Int64", + ), + }, + columns={ + ("person", "person_household_id"): pd.Series( + [10, 10, 20, 30], + index=pd.Index([1, 2, 3, 4], name="person_id"), + dtype="int64", + ), + ("household", "size"): pd.Series( + [2, 1, 1], + index=pd.Index([10, 20, 30], name="household_id"), + dtype="int64", + ), + ("household", "period"): pd.Series( + [2024, 2025, 2026], + index=pd.Index([10, 20, 30], name="household_id"), + dtype="int64", + ), + }, + weights=Weights( + np.array([1.0, 2.0, 1.0], dtype=np.float64), WeightKind.DESIGN + ), + ) + + def pass_through(column: str) -> Callable[[KernelContext], KernelResult]: + def run(context: KernelContext) -> KernelResult: + household = context.tables["household"] + return KernelResult( + columns={ + ("household", column): pd.Series( + household[column].array.copy(), + index=pd.Index(household["household_id"], name="household_id"), + dtype="int64", + ) + } + ) + + return run + + create = replace( + CREATE, + kernel="partition.source@1", + outputs=(*CREATE.outputs, Owned("household", "period", "int64")), + ) + expand = Node( + "admit_household", + "partition.expand@1", + structural=StructuralDelta.EXPAND, + base="survey", + params={ + "expand_cells": ( + ("person", "person_household_id", "int64"), + ("household", "size", "int64"), + ("household", "period", "int64"), + ), + "expand_weight_entity": "household", + "expand_weight_kind": "design", + }, + mass="free", + entrants=True, + ) + claim_size = Node( + "claim_size", + "claim.size@1", + outputs=(Owned("household", "size", "int64"),), + params={"materialized_expand_outputs": ("household.size",)}, + population=expand.id, + ) + kernels = ( + _Kernel( + create.kernel, + Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.CREATE), + create_partitioned, + ), + _Kernel( + expand.kernel, + Capabilities(Determinism.DETERMINISTIC, structural=StructuralDelta.EXPAND), + admit_household, + ), + _Kernel( + claim_size.kernel, + Capabilities(Determinism.DETERMINISTIC), + pass_through("size"), + ), + ) + + def registry(*extra: _Kernel) -> KernelRegistry: + result = _registry() + for kernel in (*kernels, *extra): + result.register(kernel) + return result + + graph = Graph( + "toy", + (SOURCE,), + (create, expand, claim_size), + mass_partition=("household", "period"), + ) + store = ContentStore(tmp_path / "store") + first_registry = registry() + cold = _run(graph, source, store, first_registry) + warm = _run(graph, source, store, first_registry) + + for manifest in (cold, warm): + assert manifest.population(expand.id).table("household")["period"].tolist() == [ + 2024, + 2025, + 2026, + ] + partition = manifest.nodes[expand.id].receipt["mass"]["partition"] # type: ignore[index] + assert (partition["entity"], partition["column"]) == ("household", "period") + assert warm.nodes[expand.id].hit + assert warm.nodes[claim_size.id].hit + + claim_period = Node( + "claim_period", + "claim.period@1", + outputs=(Owned("household", "period", "int64"),), + params={"materialized_expand_outputs": ("household.period",)}, + population=expand.id, + ) + period_kernel = _Kernel( + claim_period.kernel, + Capabilities(Determinism.DETERMINISTIC), + pass_through("period"), + ) + with pytest.raises(NodeRejected, match="cannot own mass partition"): + _run( + Graph( + "toy", + (SOURCE,), + (create, expand, claim_size, claim_period), + mass_partition=("household", "period"), + ), + source, + ContentStore(tmp_path / "ordinary-owner"), + registry(period_kernel), + ) + + def test_create_rejects_undeclared_frame_columns(tmp_path: Path) -> None: source = _source_path(tmp_path / "source") @@ -1287,3 +1537,210 @@ def explode(context: KernelContext) -> KernelResult: ContentStore(tmp_path / "compute"), compute_registry, ) + + +def test_cache_load_misses_when_stored_capabilities_disagree( + tmp_path: Path, +) -> None: + source = _source_path(tmp_path / "source") + store = ContentStore(tmp_path / "store") + graph = _graph() + registry = _registry() + manifest = _run(graph, source, store, registry) + + node = graph.node("a") + key = manifest.nodes["a"].key + record_key = graph_executor._cache_record_key(key) + record = store.load_json(record_key) + stored_capabilities = record["capabilities"] + assert isinstance(stored_capabilities, dict) + record["capabilities"] = { + **stored_capabilities, + "seed_source": "param", + } + store.put_json(record_key, record, node_key=key, verify_existing=False) + + kernel = registry.get(node.kernel) + with pytest.raises(StoreMiss, match="capabilities"): + graph_executor._load_record( + store, + node, + key=key, + kernel_impl_hash=kernel.implementation_hash(), + capabilities=kernel.capabilities, + ) + + +def test_fit_qrf_seed_source_change_misses_a_shared_store( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from microcosm.fit.kernels import QRF_EXECUTOR_KERNEL, QRF_PARAM_KERNEL + from microcosm.graph import graph_from_json + from tools.graph_parity_fixtures import FIXTURES, ParityCsvSource + + monkeypatch.setenv("POPULACE_FIT_N_JOBS", "1") + monkeypatch.setenv("POPULACE_FIT_PREDICT_WORKERS", "1") + case = FIXTURES / "fit.qrf" + compiled = compile_graph(graph_from_json((case / "graph.json").read_text())) + store = ContentStore(tmp_path / "store") + + param_registry = KernelRegistry() + param_registry.register(ParityCsvSource()) + param_registry.register(QRF_PARAM_KERNEL) + cold = run_graph( + compiled, + sources={"fixture": case}, + store=store, + kernels=param_registry, + ) + assert not cold.nodes["fit_qrf"].hit + + assert QRF_PARAM_KERNEL.ref == QRF_EXECUTOR_KERNEL.ref + assert ( + QRF_PARAM_KERNEL.implementation_hash() + == QRF_EXECUTOR_KERNEL.implementation_hash() + ) + assert ( + QRF_PARAM_KERNEL.capabilities.seed_source + is not QRF_EXECUTOR_KERNEL.capabilities.seed_source + ) + + executor_registry = KernelRegistry() + executor_registry.register(ParityCsvSource()) + executor_registry.register(QRF_EXECUTOR_KERNEL) + with pytest.raises(NodeRejected, match="EXECUTOR-seeded.*must omit"): + run_graph( + compiled, + sources={"fixture": case}, + store=store, + kernels=executor_registry, + ) + + +def test_fit_qrf_tolerance_source_hash_pin_is_current() -> None: + import json + + from microcosm.fit.kernels import QRF_PARAM_KERNEL, QRFKernel + from tools.graph_parity_fixtures import FIXTURES + + pins = json.loads((FIXTURES / "fit.qrf" / "pins.json").read_text()) + assert QRF_PARAM_KERNEL.capabilities.tolerance == Tolerance(ulps=1) + assert pins["implementation_hash"] == QRF_PARAM_KERNEL.implementation_hash() + changed_tolerance = QRFKernel(QRF_PARAM_KERNEL.capabilities.seed_source) + changed_tolerance.capabilities = replace( + changed_tolerance.capabilities, tolerance=Tolerance(ulps=2) + ) + assert changed_tolerance.implementation_hash() == pins["implementation_hash"] + + +def test_entrant_materialization_rejects_a_masked_claimant( + tmp_path: Path, +) -> None: + source = _source_path(tmp_path / "source") + + def admit_household(context: KernelContext) -> KernelResult: + return KernelResult( + expand={ + "person": pd.Series( + [1], + index=pd.Index([4], name="person_id"), + dtype="int64", + ), + "household": pd.Series( + [pd.NA], + index=pd.Index([30], name="household_id"), + dtype="Int64", + ), + }, + columns={ + ("person", "person_household_id"): pd.Series( + [10, 10, 20, 30], + index=pd.Index([1, 2, 3, 4], name="person_id"), + dtype="int64", + ), + ("household", "size"): pd.Series( + [2, 1, 1], + index=pd.Index([10, 20, 30], name="household_id"), + dtype="int64", + ), + ("household", "claim_mask"): pd.Series( + [True, True, False], + index=pd.Index([10, 20, 30], name="household_id"), + dtype="boolean", + ), + }, + weights=Weights( + np.array([1.0, 2.0, 1.0], dtype=np.float64), + WeightKind.DESIGN, + ), + ) + + expand = Node( + "admit_household", + "masked.expand@1", + structural=StructuralDelta.EXPAND, + base="survey", + params={ + "expand_cells": ( + ("person", "person_household_id", "int64"), + ("household", "size", "int64"), + ("household", "claim_mask", "boolean"), + ), + "expand_weight_entity": "household", + "expand_weight_kind": "design", + }, + mass="free", + entrants=True, + ) + claim_mask = Node( + "claim_mask", + "claim.mask@1", + outputs=(Owned("household", "claim_mask", "boolean"),), + params={"materialized_expand_outputs": ("household.claim_mask",)}, + population=expand.id, + ) + claim_size = Node( + "claim_size", + "claim.masked-size@1", + inputs=(Slice("household", ("claim_mask",)),), + outputs=(Owned("household", "size", "int64", rows="claim_mask"),), + params={"materialized_expand_outputs": ("household.size",)}, + population=expand.id, + ) + + def must_not_run(context: KernelContext) -> KernelResult: + raise AssertionError(f"claimant {context.node.id} should not run") + + registry = _registry() + registry.register( + _Kernel( + expand.kernel, + Capabilities( + Determinism.DETERMINISTIC, + structural=StructuralDelta.EXPAND, + ), + admit_household, + ) + ) + mask_kernel = _Kernel( + claim_mask.kernel, + Capabilities(Determinism.DETERMINISTIC), + must_not_run, + ) + size_kernel = _Kernel( + claim_size.kernel, + Capabilities(Determinism.DETERMINISTIC), + must_not_run, + ) + registry.register(mask_kernel) + registry.register(size_kernel) + + with pytest.raises(NodeRejected, match="household.size.*rows='all'"): + _run( + Graph("toy", (SOURCE,), (CREATE, expand, claim_mask, claim_size)), + source, + ContentStore(tmp_path / "store"), + registry, + ) + assert mask_kernel.calls == size_kernel.calls == 0 diff --git a/packages/microcosm-graph/tests/test_graph_explain.py b/packages/microcosm-graph/tests/test_graph_explain.py index 4e2b8ddba..30ab061dc 100644 --- a/packages/microcosm-graph/tests/test_graph_explain.py +++ b/packages/microcosm-graph/tests/test_graph_explain.py @@ -158,6 +158,36 @@ def test_explain_html_is_the_public_export() -> None: assert graph_api.explain_html is explain_html +def test_entrant_person_strata_survive_cache_and_are_explained(tmp_path: Path) -> None: + expand, claim = toy.entrant_person_node() + graph = toy.small_graph(nodes=(toy.CREATE, expand, claim)) + cold = toy.run_toy(graph, tmp_path / "cold") + warm = toy.run_toy( + graph, + tmp_path / "warm", + sources=cold.sources, + registry=cold.registry, + store=cold.store, + ) + entrant_id = int(cold.manifest.population("survey").person["person_id"].max()) + 1 + + assert warm.manifest.nodes[expand.id].hit + assert warm.manifest.population(expand.id).strata.equals( + cold.manifest.population(expand.id).strata + ) + assert cold.manifest.nodes[expand.id].receipt["entrant_strata"] == ( + (entrant_id, "urban"), + ) + assert ( + warm.manifest.nodes[expand.id].receipt["entrant_strata"] + == cold.manifest.nodes[expand.id].receipt["entrant_strata"] + ) + detail = describe(cold.compiled, expand.id, cold.manifest) + rendered = explain_html(cold.compiled, cold.manifest) + assert "entrant_strata" in detail and "urban" in detail + assert "entrant_strata" in rendered and "urban" in rendered + + def test_page_contains_every_node_and_its_click_detail(explanation) -> None: run, _charter, rendered = explanation for node_id in run.compiled.order: @@ -178,8 +208,8 @@ def test_page_contains_every_charter_property(explanation) -> None: for identifier in identifiers: assert f"{identifier}" in rendered assert "35 green" not in rendered # V1-V4 are also represented. - assert "41 green" in rendered - assert "4 red" in rendered + assert "45 green" in rendered + assert "0 red" in rendered assert "Flip PR" in rendered assert "Not recorded" in rendered diff --git a/packages/microcosm-graph/tests/test_graph_keys.py b/packages/microcosm-graph/tests/test_graph_keys.py index 4b1acdd25..b5342394d 100644 --- a/packages/microcosm-graph/tests/test_graph_keys.py +++ b/packages/microcosm-graph/tests/test_graph_keys.py @@ -17,6 +17,14 @@ StructuralDelta, compile_graph, ) +from microcosm.graph.kernel import ( + Capabilities, + Determinism, + KernelRole, + Numeric, + SeedSource, + Tolerance, +) from microcosm.graph.keys import ( artifact_key, frame_key, @@ -39,6 +47,12 @@ ) +def _capabilities( + structural: StructuralDelta = StructuralDelta.NONE, +) -> Capabilities: + return Capabilities(Determinism.DETERMINISTIC, structural=structural) + + def _ordinary( node_id: str, inputs: tuple[str, ...], @@ -87,6 +101,7 @@ def _all_keys( keys, implementation_hashes[node.kernel], {"survey": source_key}, + kernel_capabilities=_capabilities(node.structural), ) return compiled, keys @@ -199,13 +214,21 @@ def test_carried_columns_resolve_to_the_structural_version() -> None: graph = Graph("toy", (SOURCE,), (CREATE, subset, model)) compiled = compile_graph(graph) keys = {"survey": "a" * 64, "adults": "b" * 64} - baseline = node_key(compiled, "model", keys, "c" * 64, {}) + baseline = node_key( + compiled, + "model", + keys, + "c" * 64, + {}, + kernel_capabilities=_capabilities(), + ) changed_unreachable_base = node_key( compiled, "model", {"survey": "d" * 64, "adults": "b" * 64}, "c" * 64, {}, + kernel_capabilities=_capabilities(), ) assert baseline == changed_unreachable_base @@ -227,6 +250,7 @@ def test_structural_key_binds_every_patch_in_its_base_version() -> None: {"survey": "a" * 64, "patched": "b" * 64}, "c" * 64, {}, + kernel_capabilities=_capabilities(StructuralDelta.FILTER), ) changed_patch = node_key( compiled, @@ -234,6 +258,7 @@ def test_structural_key_binds_every_patch_in_its_base_version() -> None: {"survey": "a" * 64, "patched": "d" * 64}, "c" * 64, {}, + kernel_capabilities=_capabilities(StructuralDelta.FILTER), ) assert baseline != changed_patch @@ -248,6 +273,7 @@ def test_non_create_source_consumers_bind_their_declared_source_bytes() -> None: {"survey": "a" * 64}, "b" * 64, {"survey": "c" * 64}, + kernel_capabilities=_capabilities(), ) changed = node_key( compiled, @@ -255,5 +281,46 @@ def test_non_create_source_consumers_bind_their_declared_source_bytes() -> None: {"survey": "a" * 64}, "b" * 64, {"survey": "d" * 64}, + kernel_capabilities=_capabilities(), ) assert baseline != changed + + +def test_every_capability_field_changes_the_node_key() -> None: + compiled = compile_graph(_graph()) + base = Capabilities( + determinism=Determinism.SEEDED, + numeric=Numeric.TOLERANCE_BOUND, + seed_source=SeedSource.EXECUTOR, + role=KernelRole.COMPUTE, + consumes_se=False, + dependencies=("numpy",), + tolerance=Tolerance(rtol=1e-6, atol=2e-6, ulps=1), + ) + + def key(capabilities: Capabilities) -> str: + return node_key( + compiled, + "a", + {"survey": "a" * 64}, + "b" * 64, + {}, + kernel_capabilities=capabilities, + ) + + baseline = key(base) + variants = ( + replace(base, determinism=Determinism.DETERMINISTIC), + replace(base, numeric=Numeric.BITWISE, tolerance=None), + replace(base, seed_source=SeedSource.PARAM), + replace(base, structural=StructuralDelta.FILTER), + replace(base, role=KernelRole.GATE), + replace(base, consumes_se=True), + replace(base, dependencies=("numpy", "pandas")), + replace(base, tolerance=Tolerance(rtol=3e-6, atol=2e-6, ulps=1)), + ) + assert all(key(capabilities) != baseline for capabilities in variants) + + positive_zero = replace(base, tolerance=Tolerance(rtol=0.0, atol=2e-6, ulps=1)) + negative_zero = replace(base, tolerance=Tolerance(rtol=-0.0, atol=2e-6, ulps=1)) + assert key(positive_zero) == key(negative_zero) diff --git a/packages/microcosm-graph/tests/test_graph_manifest.py b/packages/microcosm-graph/tests/test_graph_manifest.py index 8d6a50621..e90570d14 100644 --- a/packages/microcosm-graph/tests/test_graph_manifest.py +++ b/packages/microcosm-graph/tests/test_graph_manifest.py @@ -6,13 +6,15 @@ from dataclasses import FrozenInstanceError, replace from pathlib import Path +import numpy as np import pandas as pd import pytest import microcosm.graph as graph_api +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights from microcosm.graph.decl import StructuralDelta from microcosm.graph.kernel import Capabilities, Determinism, KernelRole, SeedSource -from microcosm.graph.manifest import Decision, NodeReceipt, RunManifest +from microcosm.graph.manifest import Decision, NodeReceipt, PopulationView, RunManifest from microcosm.graph.population import MassRecord @@ -26,6 +28,32 @@ def _capabilities(role: KernelRole = KernelRole.COMPUTE) -> Capabilities: ) +def _frame() -> Frame: + person = pd.DataFrame( + { + "person_id": np.asarray([1, 2], dtype=np.int64), + "person_household_id": np.asarray([10, 20], dtype=np.int64), + } + ) + household = pd.DataFrame( + { + "household_id": np.asarray([10, 20], dtype=np.int64), + "size": np.asarray([1, 1], dtype=np.int64), + } + ) + return Frame( + {"person": person, "household": household}, + EntitySchema(group_entities=("household",)), + { + "household": Weights( + np.asarray([1.0, 2.0], dtype=np.float64), + WeightKind.DESIGN, + ) + }, + pd.Series(["a", "b"], name="stratum"), + ) + + def _receipt(key: str, *, hit: bool = False, wall_time: float = 0.2) -> NodeReceipt: return NodeReceipt( key=key, @@ -88,8 +116,8 @@ def _persisted_manifest( return RunManifest("toy", {"release": release, "gate": gate}) -def test_manifest_json_round_trip_and_convenient_lookup() -> None: - population = object() +def test_manifest_json_round_trip_and_population_view() -> None: + raw = _frame() manifest = RunManifest( country="toy", nodes={"b": _receipt("b" * 64), "a": _receipt("a" * 64)}, @@ -97,7 +125,7 @@ def test_manifest_json_round_trip_and_convenient_lookup() -> None: started_at="2026-09-01T12:00:00Z", finished_at="2026-09-01T12:00:01Z", host="runner-1", - populations={"survey": population}, # type: ignore[dict-item] + populations={"survey": raw, "filtered": raw}, ) restored = RunManifest.from_json(manifest.to_json()) assert restored == manifest @@ -105,9 +133,30 @@ def test_manifest_json_round_trip_and_convenient_lookup() -> None: assert manifest.nodes["a"] is manifest.node("a") assert manifest.receipts["a"] is manifest.receipt("a") assert manifest["a"].artifacts[("person", "x")] == "d" * 64 - assert manifest.population("survey") is population + + survey = manifest.population("survey") + filtered = manifest.population("filtered") + assert type(survey) is type(filtered) is PopulationView + assert isinstance(survey, Frame) + assert manifest.population("survey") is survey + assert type(raw) is Frame + assert not hasattr(raw, "household") + assert survey.person is raw.person + assert survey.household is raw.table("household") + assert survey.table("household") is raw.table("household") + assert survey.weights_for("household") is raw.weights_for("household") + assert survey.strata is raw.strata + with pytest.raises(AttributeError, match="PopulationView.*missing"): + _ = survey.missing + with pytest.raises(KeyError, match="not attached"): restored.population("survey") + with pytest.raises(TypeError, match="values must be Frame"): + RunManifest( + "toy", + {"a": _receipt("a" * 64)}, + populations={"survey": object()}, # type: ignore[dict-item] + ) def test_manifest_key_excludes_every_operational_field() -> None: @@ -397,6 +446,7 @@ def test_load_requires_every_manifest_artifact( def test_package_exports_runtime_implementations_and_failures() -> None: assert graph_api.ContentStore.__module__.endswith(".store") assert graph_api.RunManifest is RunManifest + assert graph_api.PopulationView is PopulationView assert graph_api.NodeReceipt is NodeReceipt assert graph_api.Decision is Decision assert graph_api.run_graph.__module__.endswith(".executor") diff --git a/packages/microcosm-graph/tests/test_graph_population.py b/packages/microcosm-graph/tests/test_graph_population.py index 13bb682c9..6c3c7fca1 100644 --- a/packages/microcosm-graph/tests/test_graph_population.py +++ b/packages/microcosm-graph/tests/test_graph_population.py @@ -21,6 +21,7 @@ PopulationError, dtype_for_token, dtype_matches, + entrant_strata_receipt, expand_lineage_receipt, owned_ids, patch, @@ -393,6 +394,35 @@ def test_declared_mass_validates_the_kernel_receipt() -> None: patch(population, node, result) +def test_mass_receipt_rejects_partition_when_graph_has_none() -> None: + population = _population() + node = Node( + "importance", + "test@1", + structural=StructuralDelta.REWEIGHT, + base="source", + weights=WeightTransition("household", "importance", mass="declared"), + mass="declared", + ) + receipt = _mass_receipt( + policy="declared", + before=7.0, + after=14.0, + stratum_before={"a": 2.0, "b": 5.0}, + stratum_after={"a": 4.0, "b": 10.0}, + ) + mass = receipt["mass"] + assert isinstance(mass, dict) + mass["partition"] = {} + result = KernelResult( + weights=Weights(np.array([2.0, 4.0, 6.0]), WeightKind.IMPORTANCE), + receipt=receipt, + ) + + with pytest.raises(PopulationError, match="declares no mass partition"): + patch(population, node, result) + + def test_filter_requires_subset_ids_and_records_free_mass() -> None: population = _population() filtered = population.frame.select( @@ -475,6 +505,84 @@ def _lineage_expand_result(*, bad_source: bool = False) -> KernelResult: ) +def _entrant_person_expand_node(*, membership_dtype: str = "int64") -> Node: + return Node( + "entrant_person", + "test@1", + structural=StructuralDelta.EXPAND, + base="source", + params={ + "expand_cells": ( + ("person", "person_household_id", membership_dtype), + ("person", "keep", "bool"), + ("person", "owned", "boolean"), + ("person", "nullable", "boolean"), + ("person", "amount", "float64"), + ), + "expand_weight_entity": "household", + "expand_weight_kind": "design", + }, + mass="free", + entrants=True, + ) + + +def _entrant_person_expand_result( + strata: object, *, frame: Frame | None = None +) -> KernelResult: + frame = _frame() if frame is None else frame + person = frame.table("person") + person_id_dtype = person["person_id"].dtype + household_id_dtype = frame.table("household")["household_id"].dtype + entrant_id = int(person["person_id"].max()) + 1 + ids = pd.Index( + pd.Series([*person["person_id"], entrant_id], dtype=person_id_dtype).array, + name="person_id", + ) + additions = { + "person_household_id": 10, + "keep": True, + "owned": False, + "nullable": pd.NA, + "amount": 3.0, + } + tokens = { + "person_household_id": token_for_dtype(person["person_household_id"].dtype), + "keep": "bool", + "owned": "boolean", + "nullable": "boolean", + "amount": "float64", + } + columns = { + ("person", column): pd.Series( + pd.array([*person[column], value], dtype=tokens[column]), index=ids + ) + for column, value in additions.items() + } + return KernelResult( + expand={ + "person": pd.Series( + pd.array( + [pd.NA], + dtype=f"Int{np.dtype(person_id_dtype).itemsize * 8}", + ), + index=pd.Index( + pd.Series([entrant_id], dtype=person_id_dtype).array, + name="person_id", + ), + ), + "household": pd.Series( + [], + index=pd.Index([], dtype=household_id_dtype, name="household_id"), + dtype=household_id_dtype, + ), + }, + columns=columns, + weights=frame.weights_for("household"), + strata=strata, # type: ignore[arg-type] + ) + + def test_expand_lineage_carries_rows_remaps_memberships_and_restores_cache() -> None: population = _population() node = _lineage_expand_node() @@ -507,6 +615,265 @@ def test_expand_lineage_carries_rows_remaps_memberships_and_restores_cache() -> assert cached.mass_ledger == expanded.mass_ledger +def test_cached_expand_requires_exact_lineage_id_sequence() -> None: + population = _population() + node = Node( + "cached_midpoint", + "test@1", + structural=StructuralDelta.EXPAND, + base="source", + params={ + "expand_cells": (), + "expand_weight_entity": "household", + "expand_weight_kind": "design", + }, + mass="free", + ) + before = population.frame + person = before.table("person") + household = before.table("household") + added_person = person.iloc[[0]].copy() + added_person["person_id"] = np.array([5], dtype=np.int64) + added_person["person_household_id"] = np.array([15], dtype=np.int64) + final_person = pd.concat([person, added_person], ignore_index=True) + added_household = household.iloc[[0]].copy() + added_household["household_id"] = np.array([15], dtype=np.int64) + final_household = ( + pd.concat([household, added_household], ignore_index=True) + .sort_values("household_id") + .reset_index(drop=True) + ) + final_weights = Weights( + np.array([1.0, 1.0, 2.0, 3.0], dtype=np.float64), WeightKind.DESIGN + ) + cached_frame = Frame( + {"person": final_person, "household": final_household}, + before.schema, + {"household": final_weights}, + pd.concat([before.strata, before.strata.iloc[[0]]], ignore_index=True), + ) + lineage = { + "person": pd.Series([1], index=pd.Index([5], name="person_id"), dtype="int64"), + "household": pd.Series( + [10], index=pd.Index([15], name="household_id"), dtype="int64" + ), + } + + with pytest.raises(PopulationError, match="final 'household' ids"): + restore_cached_expand( + population, + node, + KernelResult( + frame=cached_frame, + weights=final_weights, + receipt={"expand": expand_lineage_receipt(lineage)}, + ), + ) + + +def test_entrant_person_strata_materialize_and_attest_cached_replay() -> None: + population = _population() + node = _entrant_person_expand_node() + result = _entrant_person_expand_result( + pd.Series( + ["new"], + index=pd.Index([5], dtype="int64", name="ignored"), + dtype=object, + name="ignored", + ) + ) + + expanded = patch(population, node, result) + + assert expanded.frame.table("person")["person_household_id"].tolist()[-1] == 10 + assert expanded.frame.strata.tolist() == ["a", "a", "b", "b", "new"] + assert expanded.mass_ledger[-1].before_total == 7.0 + assert expanded.mass_ledger[-1].after_total == 8.0 + assert result.expand is not None + entrant_receipt = entrant_strata_receipt( + population.frame, node, result.expand, result.strata + ) + receipt = { + "expand": expand_lineage_receipt(result.expand), + "entrant_strata": entrant_receipt, + } + cached = restore_cached_expand( + population, + node, + KernelResult( + frame=expanded.frame, + weights=result.weights, + receipt=receipt, + ), + ) + pd.testing.assert_series_equal(cached.frame.strata, expanded.frame.strata) + assert cached.mass_ledger == expanded.mass_ledger + + with pytest.raises(PopulationError, match="entrant-strata receipt"): + restore_cached_expand( + population, + node, + KernelResult( + frame=expanded.frame, + weights=result.weights, + receipt={"expand": expand_lineage_receipt(result.expand)}, + ), + ) + + +def test_cached_entrant_strata_rehydrate_the_base_id_dtype() -> None: + source = _frame() + person = source.table("person").copy() + household = source.table("household").copy() + for column in ("person_id", "person_household_id"): + person[column] = person[column].astype("int32") + household["household_id"] = household["household_id"].astype("int32") + frame = Frame( + {"person": person, "household": household}, + source.schema, + {"household": source.weights_for("household")}, + source.strata.copy(), + ) + population = Population.from_frame(frame, "source") + node = _entrant_person_expand_node(membership_dtype="int32") + result = _entrant_person_expand_result( + pd.Series(["new"], index=pd.Index([5], dtype="int32"), dtype=object), + frame=frame, + ) + + expanded = patch(population, node, result) + assert result.expand is not None + receipt = { + "expand": expand_lineage_receipt(result.expand), + "entrant_strata": entrant_strata_receipt( + frame, node, result.expand, result.strata + ), + } + cached = restore_cached_expand( + population, + node, + KernelResult( + frame=expanded.frame, + weights=result.weights, + receipt=receipt, + ), + ) + + assert cached.frame.table("person")["person_id"].dtype == np.dtype("int32") + pd.testing.assert_series_equal(cached.frame.strata, expanded.frame.strata) + + +@pytest.mark.parametrize( + ("receipt_label", "changed_label"), + [(1, True), (1, 1.0), (-0.0, 0.0)], + ids=["bool", "float", "signed-zero"], +) +def test_cached_entrant_strata_preserve_label_scalar( + receipt_label: object, changed_label: object +) -> None: + population = _population() + node = _entrant_person_expand_node() + result = _entrant_person_expand_result( + pd.Series([receipt_label], index=pd.Index([5], dtype="int64"), dtype=object) + ) + expanded = patch(population, node, result) + changed_strata = expanded.frame.strata.copy() + changed_strata.iloc[-1] = changed_label + changed_frame = _replace_person_table( + expanded.frame, + expanded.frame.table("person").copy(), + changed_strata, + ) + assert result.expand is not None + receipt = { + "expand": expand_lineage_receipt(result.expand), + "entrant_strata": entrant_strata_receipt( + population.frame, node, result.expand, result.strata + ), + } + + with pytest.raises(PopulationError, match="label"): + restore_cached_expand( + population, + node, + KernelResult( + frame=changed_frame, + weights=result.weights, + receipt=receipt, + ), + ) + + +def test_cached_entrant_strata_encode_bytes_labels() -> None: + population = _population() + node = _entrant_person_expand_node() + result = _entrant_person_expand_result( + pd.Series([b"new\x00stratum"], index=pd.Index([5], dtype="int64"), dtype=object) + ) + expanded = patch(population, node, result) + assert result.expand is not None + receipt = { + "expand": expand_lineage_receipt(result.expand), + "entrant_strata": entrant_strata_receipt( + population.frame, node, result.expand, result.strata + ), + } + + assert receipt["entrant_strata"] == [[5, {"bytes_hex": "6e6577007374726174756d"}]] + cached = restore_cached_expand( + population, + node, + KernelResult( + frame=expanded.frame, + weights=result.weights, + receipt=receipt, + ), + ) + pd.testing.assert_series_equal(cached.frame.strata, expanded.frame.strata) + + +@pytest.mark.parametrize( + "strata", + [ + None, + pd.Series(["new"], index=pd.Index([6], dtype="int64"), dtype=object), + pd.Series(["old", "new"], index=pd.Index([1, 5], dtype="int64"), dtype=object), + pd.Series(["new"], index=pd.Index([5], dtype="int32"), dtype=object), + pd.Series([pd.NA], index=pd.Index([5], dtype="int64"), dtype=object), + pd.Series([1], index=pd.Index([5], dtype="int64"), dtype="int64"), + ], + ids=[ + "missing", + "unknown-id", + "incumbent-id", + "wrong-id-dtype", + "missing-label", + "wrong-label-dtype", + ], +) +def test_entrant_person_strata_reject_malformed_exact_set(strata: object) -> None: + with pytest.raises(PopulationError, match="strata"): + patch( + _population(), + _entrant_person_expand_node(), + _entrant_person_expand_result(strata), + ) + + +def test_strata_are_rejected_without_entrant_persons() -> None: + result = _lineage_expand_result() + with pytest.raises(PopulationError, match="without entrant persons"): + patch( + _population(), + _lineage_expand_node(), + KernelResult( + expand=result.expand, + weights=result.weights, + strata=pd.Series([], dtype=object), + ), + ) + + def test_expand_lineage_rejects_an_unknown_source_id() -> None: with pytest.raises(PopulationError, match="unknown 'person' source ids"): patch(