diff --git a/changelog.d/reusable-qrf-models.added.md b/changelog.d/reusable-qrf-models.added.md new file mode 100644 index 00000000..7bee3cf7 --- /dev/null +++ b/changelog.d/reusable-qrf-models.added.md @@ -0,0 +1,6 @@ +Add separate `fit.qrf.train@1` and `fit.qrf.apply@1` graph kernels with typed, +reusable model artifacts and stable entity-coordinate draws. The fitted QRF's +new `predict_from_uniforms` API supports stateless chained predictions while +preserving the existing `predict` RNG stream. Training excludes zero-weight +rows from its effective support and records both source and resolved weight +provenance. diff --git a/changelog.d/typed-model-artifacts.added.md b/changelog.d/typed-model-artifacts.added.md new file mode 100644 index 00000000..0d55d545 --- /dev/null +++ b/changelog.d/typed-model-artifacts.added.md @@ -0,0 +1 @@ +Add typed graph artifact dependencies across populations, verified cold/warm model inputs, preserved numeric scope, and stable coordinate-keyed random draws. Preserve legacy node identities and serialization when the new declarations are empty. diff --git a/changelog.d/typed-model-transfer-example.added.md b/changelog.d/typed-model-transfer-example.added.md new file mode 100644 index 00000000..25371eb3 --- /dev/null +++ b/changelog.d/typed-model-transfer-example.added.md @@ -0,0 +1 @@ +Add a runnable synthetic destination-transfer graph using one fitted QRF model across two independent recipient populations, explicit annual unit conversion, real household calibration, and held-out evaluation. The example records model/cache identities and separate calibration/evaluation verdicts, with tests for selective invalidation and holdout isolation. diff --git a/docs/evidence/spec-engine/us-f0-coverage.json b/docs/evidence/spec-engine/us-f0-coverage.json index 5365a086..797d697d 100644 --- a/docs/evidence/spec-engine/us-f0-coverage.json +++ b/docs/evidence/spec-engine/us-f0-coverage.json @@ -1656,13 +1656,13 @@ "compiler_ir.node_slices" ], "expected": { - "map_sha256": "6b4902d9a640dd459942e588ae2cc7fb937f1c42ad814365c7c1616c98ce1b68", - "protocol_sha256": "e63bbfa0f05302a672acd9914e3652c151cc7ee690182d8823dfb85329b1e911" + "map_sha256": "fdf77a621dfb74c58278a443f73e7addabcd42ebf9af7105ce36c9da3f9cd35c", + "protocol_sha256": "6a29390792a7111bc359e2d7a6c13a55e894c3404708cb501fa142cc8880f2c9" }, "failures": [], "observed": { - "map_sha256": "6b4902d9a640dd459942e588ae2cc7fb937f1c42ad814365c7c1616c98ce1b68", - "protocol_sha256": "e63bbfa0f05302a672acd9914e3652c151cc7ee690182d8823dfb85329b1e911" + "map_sha256": "fdf77a621dfb74c58278a443f73e7addabcd42ebf9af7105ce36c9da3f9cd35c", + "protocol_sha256": "6a29390792a7111bc359e2d7a6c13a55e894c3404708cb501fa142cc8880f2c9" }, "status": "covered" }, @@ -1677,7 +1677,7 @@ "compiler_ir.seed_stream_map" ], "expected": { - "implementation_sha256": "e63bbfa0f05302a672acd9914e3652c151cc7ee690182d8823dfb85329b1e911", + "implementation_sha256": "6a29390792a7111bc359e2d7a6c13a55e894c3404708cb501fa142cc8880f2c9", "protocol": "legacy-v1", "streams": [ "build_model", @@ -1698,7 +1698,7 @@ }, "failures": [], "observed": { - "implementation_sha256": "e63bbfa0f05302a672acd9914e3652c151cc7ee690182d8823dfb85329b1e911", + "implementation_sha256": "6a29390792a7111bc359e2d7a6c13a55e894c3404708cb501fa142cc8880f2c9", "protocol": "legacy-v1", "streams": [ "build_model", @@ -2599,7 +2599,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "a6bc79878eb6f64637b9f3eceeea6cc2b050c0e5b8f9aca446179258940c44f2" + "spec_sha256": "be31b72f7960a3da54dbaead103c6b7660f8f4cc709ced75a24a19588e477198" } }, "report_schema_version": 3, @@ -2609,7 +2609,7 @@ "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "a6bc79878eb6f64637b9f3eceeea6cc2b050c0e5b8f9aca446179258940c44f2" + "spec_sha256": "be31b72f7960a3da54dbaead103c6b7660f8f4cc709ced75a24a19588e477198" }, "status": "pass" } diff --git a/docs/graph-acceptance.md b/docs/graph-acceptance.md index 4f602d4a..e99d3635 100644 --- a/docs/graph-acceptance.md +++ b/docs/graph-acceptance.md @@ -292,6 +292,29 @@ Amendments so far (each re-locked): `hit` forced to false) and `load_certified` refuses it. Raised by the #847 gate review; adopted 2026-09-03. +19. **Typed artifacts and stable draw coordinates.** Nodes may declare + `artifact_outputs` (named nominal type/version contracts) and + `artifact_inputs` (consumer aliases referencing a producer output). Edges + may cross population versions and participate in compilation, identity, + cache validation and gate ancestry. The executor supplies immutable + `ArtifactValue` bytes and producer numeric scopes; consumers validate + decoded payloads. Legacy opaque diagnostics remain legal. Only the two + newly added empty artifact fields are omitted from old declarations and + keys; the base-96faa5d acceptance graph is pinned in + `tests/fixtures/legacy-graph-key-baseline.json`. Typed cache records use + schema 2 and typed manifests schema 3; legacy runs retain their formats. + Numeric scope weakening is refused: platform-bitwise requires a + platform-bitwise consumer, tolerance-bound requires a tolerance-bound + consumer with its own output tolerance, and mixed platform/tolerance + inputs are unsupported. `SeedSource.KEYED` opts into the versioned + `keyed_uniform` helper over normative stream params and stable coordinates; + the default executor RNG remains unchanged. Tests in + `test_artifact_edges.py` and `test_keyed_randomness.py` cover these additions. + The existing B2/context shape assertions include the appended immutable + artifact field. Implemented after the 2026-09-04 Fable plan gate; this + records the interface amendment for owner review, not a claim of code + approval or release certification. + Adding a normative field with a default changes the canonical projection of every node that carries it, so node keys moved with amendments 11 and 13's sibling field `entrants`; no released artifact pins a graph key yet. diff --git a/docs/graph-interface.lock b/docs/graph-interface.lock index cabec2d3..c3356b7c 100644 --- a/docs/graph-interface.lock +++ b/docs/graph-interface.lock @@ -1,2 +1,2 @@ -635fef92c599c298e7f19ca0badfa85aa040bf8e81eafed59f37c48db1fcff06 decl.py -eaf07da2eded1b1895aa0c59f603eb93744ed928df65aa9e65aa633762833949 kernel.py +8229270f3328f537c8f8e83d6c81af39aa75d4ca5aa6a7bd3d37ecfc25aee2fe decl.py +07691fb5cf45ae700a258713ebdc9aa845891a432ec9107e223b8c943b5f3cb6 kernel.py diff --git a/docs/model-artifact-transfer-example.md b/docs/model-artifact-transfer-example.md new file mode 100644 index 00000000..35d40ac8 --- /dev/null +++ b/docs/model-artifact-transfer-example.md @@ -0,0 +1,122 @@ +# One fitted model, two synthetic destinations + +Run the complete example from an installed Microcosm constellation with the +typed-artifact graph extension: + +```bash +python -m microcosm.build.transfer_example --output /tmp/microcosm-transfer-example +``` + +In a development checkout, prefix the command with `uv run`. No country engine, +download, credentials, or restricted microdata is needed. The command requires +an explicit output directory and writes the following there: + +- `report.json`: identities, calibration diagnostics, and held-out comparisons. +- `run_manifest.json`: execution receipts and verified store references. +- `graph.json`: the complete graph declaration. +- `store/`: source frames, fitted model, intermediate columns, and cache records. + +Running again with the same output directory reuses verified results. Pickled +QRF models follow the graph's trusted-local-store convention: a digest verifies +bytes, not the safety of executing an externally supplied pickle. This example +generates and fits its own model and accepts no external model files. + +## What actually runs + +The example uses the real split QRF training/application kernels and the +`calibrate.adam@1` solver. The donor and both destinations are independent graph +populations. The fitted model is a typed artifact edge crossing those population +boundaries; it is not refitted separately for each destination. + +```mermaid +flowchart LR + D[Synthetic donors] --> F[Fit QRF once] + F --> M[Typed model artifact] + A[Alpha predictors] --> PA[Apply] + B[Beta predictors] --> PB[Apply] + M --> PA + M --> PB + PA --> UA[Annualize minor units] + PB --> UB[Annualize minor units] + UA --> CA[Calibrate count and size margins] + UB --> CB[Calibrate count and size margins] + CA --> EA[Evaluate consumption] + CB --> EB[Evaluate consumption] + HA[Separate Alpha reference] --> EA + HB[Separate Beta reference] --> EB +``` + +All data are generated engineering fixtures. There are 64 donor households and +24 recipient households per fictional destination. Household size and a binary +dwelling category predict synthetic monthly consumption. Donors carry unequal +design weights; recipient sources contain no consumption outcomes. Each source +has one linked synthetic person per household to meet Frame's structural +contract. That person is a linkage placeholder: estimates use household weights, +and the number of person rows is not an estimate of population size. + +The conversion records the floating-point factor `12 / 100`: twelve monthly +periods and one hundred minor units per base unit. Its input convention is +explicitly **monthly synthetic minor units**. The destination `MonetaryBasis` +declares annual 2024 flows in synthetic `XXX` base currency, with a fictional +household-consumption perimeter. This is an explicit unit conversion, not a +currency exchange-rate assumption. The conversion test checks byte equality +with `raw * recorded_factor`; it does not claim exact rational arithmetic. + +Alpha starts at design mass 1,200 households and targets an average household +size of 2.9; Beta starts at 800 and targets 2.1. Calibration uses only household +count and summed household size. The solver, REWEIGHT node, and weight transition +all explicitly declare `mass="free"`. The report records actual initial and +final household mass, residuals, effective sample size, and maximum weight share. +Consumption does not enter the target matrix. The existing calibrator reports +that it does not consume target standard errors; none are supplied here. + +## Separate evidence, separate verdicts + +Each held-out reference is independently generated from the fixture's declared +process, with 96 households and disjoint identifiers. Monthly reference +consumption is `8000 + 5000 * household_size + 1000 * dwelling`; reference design +weights have a specified size tilt. Donors also carry residual variation. The +reference generator never reads fitted models, recipient predictions, or solver +results. Reference sources are declared only on evaluation nodes. + +The report separates `calibration_passed` from `heldout_passed`. Its 2% margin +tolerance and 15% consumption tolerance are engineering fixture expectations, +not reviewed scientific thresholds. Held-out checks compare the consumption +mean and consumption means by household size. The tests deliberately multiply +one reference's consumption by ten: its held-out verdict fails while marginal +calibration remains successful. The fitted model, application outputs, and +calibrated weights remain unchanged. + +Every report declares `scope="synthetic_engineering"`. Neither a successful run +nor a passing fixture check certifies a country population, an empirical +transfer method, a monetary target profile, or any tax-benefit result. There is +no rules-engine evaluation or national-release promotion in this example. + +## Exercising reuse from Python + +```python +from dataclasses import replace +from microcosm.build.transfer_example import ( + default_targets, + make_synthetic_inputs, + run_transfer_example, +) + +inputs = make_synthetic_inputs() +first = run_transfer_example("/tmp/microcosm-transfer-example", inputs=inputs) +targets = default_targets() +targets["alpha"] = replace(targets["alpha"], size_total=3600.0) +second = run_transfer_example( + "/tmp/microcosm-transfer-example", inputs=inputs, targets=targets +) +assert second.manifest.node("donor.fit").hit +assert second.manifest.node("alpha.apply").hit +assert not second.manifest.node("alpha.calibrate").hit +assert second.manifest.node("beta.calibrate").hit +``` + +`test_transfer_graph_example.py` also checks real fitting occurs once across +both destinations and warm runs; recipient edits affect their branch only; +donor values or weights invalidate both applications; mismatched destination +bases refuse; and held-out changes affect evaluation alone. The test lives in +the flat build test inventory and is explicitly assigned to shared/spec CI. diff --git a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py index 1c38f835..90d79ca9 100644 --- a/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py +++ b/packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py @@ -359,8 +359,8 @@ "late_schedule": "e59c019d3d454eac99ac0ac209b6c5b6faaf9bdfcaeee18c36a25be19bf7da2f", "ownership": "5f64f0aac49e2313177564f71876bffc8c81b3ded4df701e70930e60e9c98356", "primary_tuples": "987b501c695e31f45521c4a178528f75ab3df22c09bc407b182213b2de99ee57", - "seed_map": "6b4902d9a640dd459942e588ae2cc7fb937f1c42ad814365c7c1616c98ce1b68", - "seed_protocol": "e63bbfa0f05302a672acd9914e3652c151cc7ee690182d8823dfb85329b1e911", + "seed_map": "fdf77a621dfb74c58278a443f73e7addabcd42ebf9af7105ce36c9da3f9cd35c", + "seed_protocol": "6a29390792a7111bc359e2d7a6c13a55e894c3404708cb501fa142cc8880f2c9", "source_manifest": "cd5ba8924d64da5425ee14cca82a774e3f4b2bb5aabe06df291cc3cc457287a9", "take_up": "fa186daea0f8dd641cc470e41d1a2953f887d45282ec990201298f47bedf8d4d", "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c", diff --git a/packages/microcosm-build/src/microcosm/build/transfer_example.py b/packages/microcosm-build/src/microcosm/build/transfer_example.py new file mode 100644 index 00000000..15786f43 --- /dev/null +++ b/packages/microcosm-build/src/microcosm/build/transfer_example.py @@ -0,0 +1,641 @@ +"""A synthetic engineering example of one fitted model and two destinations. + +Run with ``python -m microcosm.build.transfer_example --output ``. +All inputs are generated, all populations are fictional, and consumption is a +synthetic quantity. No national population or tax-benefit result is certified. +The real QRF, content store, and Adam calibrator execute this example. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from collections.abc import Mapping +from dataclasses import asdict, dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +from microcosm.build.monetary_targets import ( + MonetaryBasis, + prepare_monetary_measure, +) +from microcosm.calibrate.kernels import CALIBRATE_ADAM +from microcosm.fit.graph_models import ( + QRF_MODEL_TYPE, + QRFApplyKernel, + QRFTrainKernel, +) +from microcosm.frame import EntitySchema, Frame, WeightKind, Weights +from microcosm.graph import ( + ArtifactInput, + ArtifactOutput, + Capabilities, + ContentStore, + Determinism, + Graph, + KernelBase, + KernelContext, + KernelRegistry, + KernelResult, + Node, + Numeric, + Owned, + RunManifest, + Slice, + SourceRef, + StructuralDelta, + WeightTransition, + compile_graph, + graph_to_json, + load_source, + run_graph, + source_hash, +) + +CONVERSION_FACTOR = 12 / 100 +"""Recorded floating-point conversion, not an economic exchange rate.""" + +ANNUAL_BASIS = MonetaryBasis( + currency="XXX", + unit="base_currency", + period="2024", + temporal_basis="annual_flow", + sector="synthetic_households", + perimeter="fictional household consumption", + valuation="synthetic nominal units", +) + +_PREDICTORS = ("household_size", "dwelling") +_MONTHLY = "monthly_consumption_minor" +_ANNUAL = "annual_consumption" +_DESTINATIONS = ("alpha", "beta") +_SCOPE = "synthetic_engineering" + + +@dataclass(frozen=True) +class DestinationTargets: + """Synthetic count margins; no consumption outcome enters calibration.""" + + household_total: float + size_total: float + + def __post_init__(self) -> None: + values = (self.household_total, self.size_total) + if any( + isinstance(value, bool) or not np.isfinite(value) or value <= 0 + for value in values + ): + raise ValueError("Synthetic target totals must be positive finite numbers.") + + def rows(self) -> tuple[tuple, ...]: + return ( + ("household_count", "households", None, self.household_total, None), + ("household_size_total", "household_size", None, self.size_total, None), + ) + + +@dataclass(frozen=True) +class TransferInputs: + """Separate donor, recipient, and held-out data boundaries.""" + + donor: Frame + recipients: Mapping[str, Frame] + references: Mapping[str, Frame] + + +@dataclass(frozen=True) +class TransferResult: + manifest: RunManifest + report: dict + + +def default_targets() -> dict[str, DestinationTargets]: + return { + "alpha": DestinationTargets(1200.0, 3480.0), + "beta": DestinationTargets(800.0, 1680.0), + } + + +def _frame(table: pd.DataFrame, weights: np.ndarray, *, stratum: str) -> Frame: + ids = table["household_id"].to_numpy(copy=True) + # A linked person is a structural placeholder, not a claim that household + # size is one. All estimates and calibration use household weights. + person = pd.DataFrame({"person_id": ids, "person_household_id": ids}) + return Frame( + {"person": person, "household": table}, + EntitySchema(group_entities=("household",)), + {"household": Weights(weights, WeightKind.DESIGN)}, + pd.Series(stratum, index=person.index, name="stratum"), + ) + + +def _predictors(count: int, start: int) -> pd.DataFrame: + index = np.arange(count, dtype=np.int64) + return pd.DataFrame( + { + "household_id": index + start, + "household_size": 1 + index % 4, + "dwelling": (index // 4) % 2, + "households": np.ones(count, dtype=np.float64), + } + ) + + +def make_synthetic_inputs() -> TransferInputs: + """Generate donors, predictor-only recipients, and independent references. + + The reference process is specified independently of model predictions and + solver results: monthly consumption is 8000 + 5000*size + 1000*dwelling. + Its known design weights tilt household size to each fictional population's + declared mean. Donors additionally carry +/-400 residuals and unequal + design weights. These are fixtures for integration behavior, not empirical + evidence about transferring populations between real countries. + """ + donor = _predictors(64, 1) + donor[_MONTHLY] = ( + 8000.0 + + 5000.0 * donor["household_size"] + + 1000.0 * donor["dwelling"] + + np.where((np.arange(len(donor)) // 8) % 2, -400.0, 400.0) + ) + donor_frame = _frame( + donor, + 1.0 + np.arange(len(donor)) % 3, + stratum="synthetic_donor", + ) + recipients: dict[str, Frame] = {} + references: dict[str, Frame] = {} + for index, (name, targets) in enumerate(default_targets().items(), start=1): + table = _predictors(24, index * 1000) + recipients[name] = _frame( + table, + np.full(len(table), targets.household_total / len(table)), + stratum=f"synthetic_recipient_{name}", + ) + reference = _predictors(96, index * 10000) + reference[_ANNUAL] = ( + 8000.0 + + 5000.0 * reference["household_size"] + + 1000.0 * reference["dwelling"] + ) * CONVERSION_FACTOR + size = reference["household_size"].to_numpy(dtype=np.float64) + target_mean = targets.size_total / targets.household_total + design = (targets.household_total / len(size)) * ( + 1 + (target_mean - size.mean()) * (size - size.mean()) / size.var() + ) + references[name] = _frame( + reference, design, stratum=f"synthetic_holdout_{name}" + ) + return TransferInputs(donor_frame, recipients, references) + + +def _digest(payload: object) -> str: + return hashlib.sha256( + json.dumps( + payload, sort_keys=True, separators=(",", ":"), allow_nan=False + ).encode() + ).hexdigest() + + +def _source_frame_path(store: ContentStore, frame: Frame) -> Path: + """Persist the fixture with an identity covering all tables and weights.""" + frame.revalidate() + tables = { + **{name: frame.table(name) for name in frame.entities}, + **{name: frame.link(name) for name in frame.links}, + } + payload = { + "scope": _SCOPE, + "schema": asdict(frame.schema), + "tables": { + name: table.to_dict(orient="tight") for name, table in tables.items() + }, + "dtypes": { + name: [str(dtype) for dtype in table.dtypes] + for name, table in tables.items() + }, + "weights": { + name: { + "kind": frame.weights_for(name).kind.value, + "values": frame.weights_for(name).values.tolist(), + } + for name in frame.weighted_entities + }, + "strata": frame.strata.to_frame().to_dict(orient="tight"), + "strata_dtype": str(frame.strata.dtype), + "mass_log": [asdict(record) for record in frame.mass_log], + } + return store.put_frame(_digest(payload), frame) + + +class _SourceKernel(KernelBase): + ref = "example.transfer.source@1" + capabilities = Capabilities( + Determinism.DETERMINISTIC, + structural=StructuralDelta.CREATE, + dependencies=("numpy", "pandas"), + ) + + def implementation_hash(self) -> str: + return source_hash( + type(self), load_source, dependencies=self.capabilities.dependencies + ) + + def run(self, context: KernelContext) -> KernelResult: + source = context.node.sources[0] + frame = load_source("frame-store", context.sources[source]) + return KernelResult(frame=frame, receipt={"scope": _SCOPE, "source": source}) + + +class _AnnualizeKernel(KernelBase): + ref = "example.transfer.annualize@1" + capabilities = Capabilities( + Determinism.DETERMINISTIC, + numeric=Numeric.PLATFORM_BITWISE, + dependencies=("numpy", "pandas"), + ) + + def implementation_hash(self) -> str: + return source_hash( + type(self), + prepare_monetary_measure, + dependencies=self.capabilities.dependencies, + ) + + def run(self, context: KernelContext) -> KernelResult: + basis = MonetaryBasis(**dict(context.params["basis"])) + if basis != ANNUAL_BASIS: + raise ValueError( + "Destination basis does not match this synthetic annual measure." + ) + table = context.tables["household"] + raw = table[_MONTHLY].to_numpy(dtype=np.float64) + source_convention = "monthly synthetic minor units" + factor = float(context.params["factor"]) + prepared = prepare_monetary_measure( + raw, + record_ids=table["household_id"].to_numpy(), + basis=basis, + factor=factor, + source_identity_sha256=_digest( + {"source_convention": source_convention, "values": raw.tolist()} + ), + bridge_description="Synthetic unit conversion: monthly minor units to annual base units; no economic exchange rate.", + bridge_source_sha256=_digest( + { + "source_convention": source_convention, + "basis": asdict(basis), + "factor": factor, + } + ), + ) + return KernelResult( + columns={ + ("household", _ANNUAL): pd.Series( + prepared.values, + index=pd.Index(table["household_id"], name="household_id"), + dtype="float64", + ) + }, + receipt={ + "scope": _SCOPE, + "source_convention": source_convention, + "factor": factor, + "prepared": prepared.receipt(), + }, + ) + + +class _EvaluateKernel(KernelBase): + ref = "example.transfer.evaluate@1" + capabilities = Capabilities( + Determinism.DETERMINISTIC, + numeric=Numeric.PLATFORM_BITWISE, + dependencies=("numpy", "pandas"), + ) + + def implementation_hash(self) -> str: + return source_hash( + type(self), load_source, dependencies=self.capabilities.dependencies + ) + + def run(self, context: KernelContext) -> KernelResult: + table = context.tables["household"] + weights = context.weights["household"].values + reference = load_source("frame-store", context.sources[context.node.sources[0]]) + heldout = reference.table("household") + reference_weights = reference.weights_for("household").values + actual_mean = float(np.average(table[_ANNUAL], weights=weights)) + reference_mean = float(np.average(heldout[_ANNUAL], weights=reference_weights)) + relative_error = abs(actual_mean / reference_mean - 1) + comparisons = [] + for size in sorted(set(heldout["household_size"])): + candidate_mask = table["household_size"].to_numpy() == size + reference_mask = heldout["household_size"].to_numpy() == size + if not candidate_mask.any(): + comparisons.append({"household_size": int(size), "supported": False}) + continue + candidate = float( + np.average( + table.loc[candidate_mask, _ANNUAL], weights=weights[candidate_mask] + ) + ) + expected = float( + np.average( + heldout.loc[reference_mask, _ANNUAL], + weights=reference_weights[reference_mask], + ) + ) + comparisons.append( + { + "household_size": int(size), + "supported": True, + "candidate_mean": candidate, + "reference_mean": expected, + "relative_error": abs(candidate / expected - 1), + } + ) + residuals = [] + for name, measure, _filter, value, _se in context.params["targets"]: + observed = float(np.dot(table[measure].to_numpy(dtype=np.float64), weights)) + residuals.append( + { + "name": name, + "target": value, + "observed": observed, + "relative_error": abs(observed / value - 1), + } + ) + calibration_tolerance = float(context.params["calibration_tolerance"]) + heldout_tolerance = float(context.params["heldout_tolerance"]) + calibration_passed = all( + row["relative_error"] <= calibration_tolerance for row in residuals + ) + heldout_passed = relative_error <= heldout_tolerance and all( + row["supported"] and row["relative_error"] <= heldout_tolerance + for row in comparisons + ) + receipt = { + "scope": _SCOPE, + "reference_source": context.node.sources[0], + "candidate_households": len(table), + "reference_households": len(heldout), + "weight_kind": context.weights["household"].kind.value, + "effective_sample_size": float( + weights.sum() ** 2 / np.square(weights).sum() + ), + "max_weight_share": float(weights.max() / weights.sum()), + "calibration_residuals": residuals, + "calibration_passed": calibration_passed, + "calibration_tolerance": calibration_tolerance, + "heldout_passed": heldout_passed, + "heldout_tolerance": heldout_tolerance, + "heldout_relative_error": relative_error, + "candidate_consumption_mean": actual_mean, + "reference_consumption_mean": reference_mean, + "consumption_by_household_size": comparisons, + "acceptance_scope": "fixture expectation only; not scientific certification", + } + # Evidence belongs in the evaluation receipt, without mutating the + # population whose independent predictions and weights it evaluates. + return KernelResult(receipt=receipt) + + +def _source_node(node_id: str, source: str, *, donor: bool = False) -> Node: + outputs = ( + Owned("household", "household_size", "int64"), + Owned("household", "dwelling", "int64"), + Owned("household", "households", "float64"), + ) + if donor: + outputs += (Owned("household", _MONTHLY, "float64"),) + return Node( + node_id, + _SourceKernel.ref, + outputs=outputs, + sources=(source,), + structural=StructuralDelta.CREATE, + ) + + +def transfer_graph( + targets: Mapping[str, DestinationTargets], *, basis: MonetaryBasis = ANNUAL_BASIS +) -> Graph: + """Declare model reuse and ensure held-out data enters evaluation only.""" + if basis != ANNUAL_BASIS: + raise ValueError( + "Destination basis must match the declared synthetic annual basis." + ) + if set(targets) != set(_DESTINATIONS): + raise ValueError( + "The synthetic example requires alpha and beta destination targets." + ) + sources = [SourceRef("donor", "frame-store")] + nodes = [ + _source_node("donor.source", "donor", donor=True), + Node( + "donor.fit", + QRFTrainKernel.ref, + inputs=(Slice("household", (*_PREDICTORS, _MONTHLY)),), + params={ + "predictors": _PREDICTORS, + "targets": (_MONTHLY,), + "n_estimators": 24, + "seed": 314159, + }, + population="donor.source", + artifact_outputs=(ArtifactOutput("model", QRF_MODEL_TYPE),), + ), + ] + for index, destination in enumerate(_DESTINATIONS): + source = f"{destination}.recipient" + reference = f"{destination}.reference" + population = f"{destination}.source" + calibrated = f"{destination}.calibrate" + rows = targets[destination].rows() + sources.extend( + (SourceRef(source, "frame-store"), SourceRef(reference, "frame-store")) + ) + nodes.extend( + ( + _source_node(population, source), + Node( + f"{destination}.apply", + QRFApplyKernel.ref, + inputs=(Slice("household", _PREDICTORS),), + outputs=(Owned("household", _MONTHLY, "float64"),), + params={ + "random_stream": ( + "sha256-u53-v1", + "synthetic-transfer", + index, + 271828, + ), + "period": 2024, + }, + population=population, + artifact_inputs=( + ArtifactInput("model", "donor.fit", "model", QRF_MODEL_TYPE), + ), + ), + Node( + f"{destination}.annualize", + _AnnualizeKernel.ref, + inputs=(Slice("household", (_MONTHLY,)),), + outputs=(Owned("household", _ANNUAL, "float64"),), + params={ + "basis": tuple(asdict(basis).items()), + "factor": CONVERSION_FACTOR, + }, + population=population, + ), + Node( + calibrated, + CALIBRATE_ADAM.ref, + inputs=(Slice("household", ("households", "household_size")),), + params={ + "targets": rows, + "epochs": 350, + "learning_rate": 0.03, + "max_weight_ratio": 5.0, + "weight_anchor": "design", + "mass": "free", + }, + structural=StructuralDelta.REWEIGHT, + base=population, + weights=WeightTransition("household", "calibrated", mass="free"), + mass="free", + ), + Node( + f"{destination}.evaluate", + _EvaluateKernel.ref, + inputs=( + Slice("household", ("households", "household_size", _ANNUAL)), + ), + params={ + "targets": rows, + "calibration_tolerance": 0.02, + "heldout_tolerance": 0.15, + }, + population=calibrated, + sources=(reference,), + ), + ) + ) + return Graph("synthetic-transfer", tuple(sources), tuple(nodes)) + + +def _plain(value): + if isinstance(value, Mapping): + return {key: _plain(child) for key, child in value.items()} + if isinstance(value, tuple | list): + return [_plain(child) for child in value] + return value + + +def run_transfer_example( + output: str | Path, + *, + inputs: TransferInputs | None = None, + targets: Mapping[str, DestinationTargets] | None = None, + basis: MonetaryBasis = ANNUAL_BASIS, +) -> TransferResult: + """Execute the real kernels and write only to the explicit output directory. + + Reusing this directory reuses verified content-store results. Changed + fixture inputs are separately addressed; existing source objects survive. + No network access, country engines, credentials, or publication are needed. + """ + graph = transfer_graph( + default_targets() if targets is None else targets, basis=basis + ) + inputs = make_synthetic_inputs() if inputs is None else inputs + if set(inputs.recipients) != set(_DESTINATIONS) or set(inputs.references) != set( + _DESTINATIONS + ): + raise ValueError( + "The synthetic example requires separate alpha/beta recipients and references." + ) + output = Path(output) + output.mkdir(parents=True, exist_ok=True) + store = ContentStore(output / "store") + sources = {"donor": _source_frame_path(store, inputs.donor)} + for destination in _DESTINATIONS: + sources[f"{destination}.recipient"] = _source_frame_path( + store, inputs.recipients[destination] + ) + sources[f"{destination}.reference"] = _source_frame_path( + store, inputs.references[destination] + ) + registry = KernelRegistry() + for kernel in ( + _SourceKernel(), + QRFTrainKernel(), + QRFApplyKernel(), + _AnnualizeKernel(), + CALIBRATE_ADAM, + _EvaluateKernel(), + ): + registry.register(kernel) + manifest = run_graph( + compile_graph(graph), sources=sources, store=store, kernels=registry + ) + fit = manifest.node("donor.fit") + artifact_key = fit.opaque_artifacts["model"] + report = { + "schema_version": 1, + "scope": _SCOPE, + "description": "Fictional household consumption transfer; no national or tax-benefit claims.", + "model": { + "node_key": fit.key, + "artifact_key": artifact_key, + "training_population_key": manifest.node("donor.source").key, + }, + "destinations": {}, + } + for destination in _DESTINATIONS: + evaluation = _plain(manifest.node(f"{destination}.evaluate").receipt) + initial = inputs.recipients[destination].weights_for("household").values + final = ( + manifest.population(f"{destination}.calibrate") + .weights_for("household") + .values + ) + report["destinations"][destination] = { + **evaluation, + "model_artifact_key": artifact_key, + "application_key": manifest.node(f"{destination}.apply").key, + "transformation_key": manifest.node(f"{destination}.annualize").key, + "calibration_key": manifest.node(f"{destination}.calibrate").key, + "evaluation_key": manifest.node(f"{destination}.evaluate").key, + "mass_policy": "free", + "mass_before": float(initial.sum()), + "mass_after": float(final.sum()), + } + manifest.save(output / "run_manifest.json") + (output / "graph.json").write_text(graph_to_json(graph), encoding="utf-8") + (output / "report.json").write_text( + json.dumps(report, indent=2, allow_nan=False) + "\n", encoding="utf-8" + ) + return TransferResult(manifest, report) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--output", + type=Path, + required=True, + help="Directory for synthetic inputs, cache, and reports.", + ) + args = parser.parse_args(argv) + run_transfer_example(args.output) + print(args.output / "report.json") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/packages/microcosm-build/tests/test_spec_engine_country_bundles.py b/packages/microcosm-build/tests/test_spec_engine_country_bundles.py index 3bb3d801..e5b2c9a2 100644 --- a/packages/microcosm-build/tests/test_spec_engine_country_bundles.py +++ b/packages/microcosm-build/tests/test_spec_engine_country_bundles.py @@ -25,7 +25,7 @@ "spine", "vintages", } -AM_SPEC_SHA256 = "b128d14f8e6351d745a16ef4537d5c1b8d71d11b9d96cb8a65f1a4fb953f13ed" +AM_SPEC_SHA256 = "f88170f34c600b19014aa7e091552fac029ba80775b4aea27c9b1ff2c732cc69" @pytest.mark.parametrize( @@ -45,7 +45,7 @@ ), ( "be", - "8c6018eaf518b239a5625f8df8bf783ce4723a03704d100f3dbf4c2bb7765772", + "98f0cf7bbe87a74fef3cce847be1cd1bb9923340328dfcc0e6e79f493e858abe", { "household.household_id", "person.person_id", @@ -55,7 +55,7 @@ ), ( "uk", - "5751a2b7b6ea771c6c885f62438927603e6acec3d1bdacbbbced2a1088b69175", + "37c651bf10e9fe1f1564f12e02f1434d8e79632566970b866ae7244526b7cf68", { "benunit.benunit_id", "household.household_id", diff --git a/packages/microcosm-build/tests/test_spec_engine_loader.py b/packages/microcosm-build/tests/test_spec_engine_loader.py index a7effd50..4ddde943 100644 --- a/packages/microcosm-build/tests/test_spec_engine_loader.py +++ b/packages/microcosm-build/tests/test_spec_engine_loader.py @@ -236,7 +236,7 @@ def test_semantic_hash_has_golden_vector_and_surface_separation(tmp_path) -> Non # Pin the domain separator, normalization rules, schema-set receipt, and # exact normative projection as one reviewable golden vector. assert first.spec_sha256 == ( - "7a0fbb0a2a16aff28a9ac6e205e6679b013c580171db87b39597a0b14a97e423" + "e5ebde4eb9b50cb33bdf6eba3b97e337d112c54cad88d76c2af4335847b8dd58" ) second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b") diff --git a/packages/microcosm-build/tests/test_transfer_graph_example.py b/packages/microcosm-build/tests/test_transfer_graph_example.py new file mode 100644 index 00000000..74804e6c --- /dev/null +++ b/packages/microcosm-build/tests/test_transfer_graph_example.py @@ -0,0 +1,187 @@ +"""Synthetic transfer exercises real model reuse without learning its holdout.""" + +from __future__ import annotations + +import json +from dataclasses import replace + +import numpy as np +import pandas as pd +import pytest + +from microcosm.build.transfer_example import ( + ANNUAL_BASIS, + CONVERSION_FACTOR, + default_targets, + make_synthetic_inputs, + run_transfer_example, +) +from microcosm.frame import Frame, Weights + + +def _changed_frame(frame, *, column=None, delta=0, change_weights=False): + tables = {entity: frame.table(entity).copy(deep=True) for entity in frame.entities} + weights = frame.weights_for("household") + if change_weights: + values = weights.values.copy() + values[0] *= 4 + weights = Weights(values, weights.kind) + if column is not None: + tables["household"].loc[0, column] += delta + return Frame(tables, frame.schema, {"household": weights}, frame.strata) + + +def _assert_branch_hits(result, destination): + for stage in ("source", "apply", "annualize", "calibrate", "evaluate"): + assert result.manifest.node(f"{destination}.{stage}").hit + + +def test_one_real_fit_is_shared_by_two_destinations_and_warm_runs( + tmp_path, monkeypatch +): + import microcosm.fit.graph_train as models + + calls = [] + original = models.fit_qrf + + def counted(*args, **kwargs): + calls.append(True) + return original(*args, **kwargs) + + monkeypatch.setattr(models, "fit_qrf", counted) + inputs = make_synthetic_inputs() + cold = run_transfer_example(tmp_path, inputs=inputs) + warm = run_transfer_example(tmp_path, inputs=inputs) + assert len(calls) == 1 + assert not cold.manifest.node("donor.fit").hit + assert warm.manifest.node("donor.fit").hit + assert cold.report["model"] == warm.report["model"] + for destination in ("alpha", "beta"): + _assert_branch_hits(warm, destination) + report = cold.report["destinations"][destination] + assert report["model_artifact_key"] == cold.report["model"]["artifact_key"] + assert report["calibration_passed"] + assert report["heldout_passed"] + assert report["mass_before"] > 0 and report["mass_after"] > 0 + assert report["effective_sample_size"] > 0 + assert report["weight_kind"] == "calibrated" + payload = json.loads((tmp_path / "report.json").read_text()) + assert payload["scope"] == "synthetic_engineering" + assert (tmp_path / "run_manifest.json").is_file() + assert (tmp_path / "graph.json").is_file() + + +def test_target_edit_only_recomputes_affected_calibration_and_evaluation(tmp_path): + inputs = make_synthetic_inputs() + cold = run_transfer_example(tmp_path, inputs=inputs) + targets = default_targets() + targets["alpha"] = replace( + targets["alpha"], size_total=targets["alpha"].size_total * 1.05 + ) + changed = run_transfer_example(tmp_path, inputs=inputs, targets=targets) + assert changed.manifest.node("donor.fit").hit + for stage in ("source", "apply", "annualize"): + assert changed.manifest.node(f"alpha.{stage}").hit + for stage in ("calibrate", "evaluate"): + assert not changed.manifest.node(f"alpha.{stage}").hit + _assert_branch_hits(changed, "beta") + assert changed.report["model"] == cold.report["model"] + + +def test_recipient_edit_preserves_model_and_other_destination(tmp_path): + inputs = make_synthetic_inputs() + cold = run_transfer_example(tmp_path, inputs=inputs) + recipients = dict(inputs.recipients) + recipients["alpha"] = _changed_frame( + recipients["alpha"], column="household_size", delta=1 + ) + changed = run_transfer_example( + tmp_path, inputs=replace(inputs, recipients=recipients) + ) + assert changed.manifest.node("donor.fit").hit + for stage in ("source", "apply", "annualize", "calibrate", "evaluate"): + assert not changed.manifest.node(f"alpha.{stage}").hit + _assert_branch_hits(changed, "beta") + assert changed.report["model"] == cold.report["model"] + + +@pytest.mark.parametrize("change_weights", [False, True]) +def test_donor_values_or_weights_invalidate_both_applications(tmp_path, change_weights): + inputs = make_synthetic_inputs() + cold = run_transfer_example(tmp_path, inputs=inputs) + donor = _changed_frame( + inputs.donor, + column=None if change_weights else "monthly_consumption_minor", + delta=3000, + change_weights=change_weights, + ) + changed = run_transfer_example(tmp_path, inputs=replace(inputs, donor=donor)) + assert not changed.manifest.node("donor.fit").hit + assert ( + changed.report["model"]["artifact_key"] != cold.report["model"]["artifact_key"] + ) + for destination in ("alpha", "beta"): + assert changed.manifest.node(f"{destination}.source").hit + assert not changed.manifest.node(f"{destination}.apply").hit + + +def test_conversion_is_exactly_the_recorded_float_operation(tmp_path): + result = run_transfer_example(tmp_path) + for destination in ("alpha", "beta"): + table = result.manifest.population(f"{destination}.calibrate").table( + "household" + ) + expected = table["monthly_consumption_minor"].to_numpy() * CONVERSION_FACTOR + actual = table["annual_consumption"].to_numpy() + assert expected.tobytes() == actual.tobytes() + receipt = result.manifest.node(f"{destination}.annualize").receipt + assert receipt["source_convention"] == "monthly synthetic minor units" + assert receipt["factor"] == 12 / 100 + assert receipt["prepared"]["basis"]["unit"] == "base_currency" + assert receipt["prepared"]["basis"]["currency"] == "XXX" + assert set(table) >= {"household_id", "household_size", "annual_consumption"} + assert result.report["destinations"][destination]["weight_kind"] == "calibrated" + + +def test_incompatible_destination_basis_is_rejected(tmp_path): + with pytest.raises(ValueError, match="basis"): + run_transfer_example(tmp_path, basis=replace(ANNUAL_BASIS, currency="USD")) + + +def test_holdout_only_edit_changes_evaluation_and_can_fail_despite_good_calibration( + tmp_path, +): + inputs = make_synthetic_inputs() + cold = run_transfer_example(tmp_path, inputs=inputs) + references = dict(inputs.references) + reference = references["alpha"] + tables = { + entity: reference.table(entity).copy(deep=True) for entity in reference.entities + } + tables["household"]["annual_consumption"] *= 10 + references["alpha"] = Frame( + tables, + reference.schema, + {"household": reference.weights_for("household")}, + reference.strata, + ) + changed = run_transfer_example( + tmp_path, inputs=replace(inputs, references=references) + ) + assert changed.manifest.node("donor.fit").hit + for stage in ("source", "apply", "annualize", "calibrate"): + assert changed.manifest.node(f"alpha.{stage}").hit + assert not changed.manifest.node("alpha.evaluate").hit + _assert_branch_hits(changed, "beta") + report = changed.report["destinations"]["alpha"] + assert report["calibration_passed"] + assert not report["heldout_passed"] + assert report["heldout_relative_error"] > 0.5 + pd.testing.assert_frame_equal( + cold.manifest.population("alpha.calibrate").table("household"), + changed.manifest.population("alpha.calibrate").table("household"), + ) + np.testing.assert_array_equal( + cold.manifest.population("alpha.calibrate").weights_for("household").values, + changed.manifest.population("alpha.calibrate").weights_for("household").values, + ) diff --git a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py index a9049e40..82c436d0 100644 --- a/packages/microcosm-build/tests/test_us_multispine_pool_tool.py +++ b/packages/microcosm-build/tests/test_us_multispine_pool_tool.py @@ -2499,7 +2499,7 @@ def capture_equality(expected: object, actual: object) -> None: "country": "us", "schema_id": "country_spec", "schema_version": 1, - "spec_sha256": "a6bc79878eb6f64637b9f3eceeea6cc2b050c0e5b8f9aca446179258940c44f2", + "spec_sha256": "be31b72f7960a3da54dbaead103c6b7660f8f4cc709ced75a24a19588e477198", }, } diff --git a/packages/microcosm-fit/README.md b/packages/microcosm-fit/README.md index dff310b3..9cfbf601 100644 --- a/packages/microcosm-fit/README.md +++ b/packages/microcosm-fit/README.md @@ -45,6 +45,45 @@ draws = fitted.predict(frame) # one column per target fitted_unweighted = fit(frame, predictors, targets, weights="none") ``` +## Reusable graph models + +`microcosm.fit.graph_models` separates donor training from recipient draws. +Register `QRFTrainKernel()` and `QRFApplyKernel()` in the graph kernel registry. +The training node reads one donor population Slice containing `predictors` +then `targets`, owns no columns, and declares +`ArtifactOutput("model", QRF_MODEL_TYPE)`. Its required parameters are tuples +`predictors`, `targets`, and an integer `seed`; `n_estimators`, `zero_atol` and +`max_samples_leaf` retain the public fitter's defaults. + +Each application node reads a recipient Slice containing exactly those +predictors and declares `ArtifactInput("model", train_node_id, "model", +QRF_MODEL_TYPE)`. It owns one all-row `float64` column per target in fitted +chain order. Parameters `random_stream=("sha256-u53-v1", experiment_id, +replicate, base_seed)` and integer `period` control its draws. Random +coordinates include the entity ID, target, period and draw kind. Give the same +person the same stream and period to couple counterfactual draws; change the +experiment or replicate for a distinct set of draws. Recipient edits preserve +the fitted model's cache identity; donor values and effective weights do not. + +The new training kernel excludes zero-weight rows before regime detection and +records the source typed weight kind separately from the public DataFrame +fitter's resolved `explicit` weights. Both kernels declare platform-bitwise +numerics; neither promises cross-platform prediction equality. Model artifacts +contain validated versioned metadata and a pickle from trusted local training. +Content verification establishes integrity, not safe loading of untrusted +pickle. The existing combined `fit.qrf@1` kernel remains available. + +Outside a graph, `FittedRegimeGatedQRF.predict_from_uniforms` accepts mappings +`quantiles={target: array}` and `sign_uniforms={target: array}`. Both must name +every fitted target and carry finite arrays in `[0, 1)` aligned to recipient +rows. The method preserves fitted forests and RNG state. Keep each row's +uniforms with its identity when reordering or batching. Ordinary `predict()` +retains its existing stateful RNG behavior. + +The synthetic two-destination integration is runnable with +`python -m microcosm.build.transfer_example --output `; see the +microcosm-build documentation for its separate calibration and held-out checks. + ## Dependencies The heavy dependencies (`scikit-learn`, `quantile-forest`) live here, never in diff --git a/packages/microcosm-fit/src/microcosm/fit/_graph_qrf.py b/packages/microcosm-fit/src/microcosm/fit/_graph_qrf.py new file mode 100644 index 00000000..9c84ed3e --- /dev/null +++ b/packages/microcosm-fit/src/microcosm/fit/_graph_qrf.py @@ -0,0 +1,106 @@ +"""Shared versioned QRF envelope and input validation for graph kernels. + +The payload includes a trusted local pickle. Integrity checks do not make an +untrusted pickle safe to execute; only consume trusted graph-produced bytes. +""" + +from __future__ import annotations + +import hashlib +import json +import pickle + +from microcosm.fit.qrf import FittedRegimeGatedQRF +from microcosm.graph import ROWS_ALL, ArtifactType, ArtifactValue + +QRF_MODEL_TYPE = ArtifactType("microcosm.fit.qrf", 1) +_MAGIC = b"microcosm.fit.qrf/1\n" + + +def _names(value, label): + if ( + not isinstance(value, tuple) + or not value + or any(not isinstance(v, str) or not v for v in value) + or len(set(value)) != len(value) + ): + raise ValueError(f"{label} must be a nonempty tuple of distinct column names.") + return value + + +def _table(context, ref): + node = context.node + if node.kernel != ref or len(node.inputs) != 1: + raise ValueError(f"{ref} requires exactly one declared input Slice.") + declared = node.inputs[0] + if declared.rows != ROWS_ALL: + raise ValueError(f"{ref} requires an all-row input Slice.") + table = context.tables[declared.entity] + id_column = f"{declared.entity}_id" + if id_column not in table or not set(declared.columns).issubset(table.columns): + raise ValueError(f"{ref} table is missing declared columns or entity IDs.") + if table[id_column].isna().any() or table[id_column].duplicated().any(): + raise ValueError(f"{ref} requires non-null unique entity IDs.") + return declared, table, id_column + + +def _encode_model(model, source_weight_kind): + payload = pickle.dumps(model, protocol=pickle.HIGHEST_PROTOCOL) + metadata = { + "schema_version": 1, + "family": "regime_gated_qrf", + "predictors": model.predictors, + "targets": model.targets, + "regimes": model.regimes(), + "fit_weight_kind": model.weight_kind, + "source_weight_kind": source_weight_kind, + "pickle_sha256": hashlib.sha256(payload).hexdigest(), + } + header = json.dumps(metadata, sort_keys=True, separators=(",", ":")).encode() + return _MAGIC + len(header).to_bytes(8, "big") + header + payload + + +def load_qrf_model(artifact: ArtifactValue) -> FittedRegimeGatedQRF: + """Validate and load a trusted graph-produced model (never untrusted pickle).""" + if not isinstance(artifact, ArtifactValue) or artifact.type != QRF_MODEL_TYPE: + raise ValueError("Expected a microcosm.fit.qrf version 1 model artifact.") + data = artifact.payload + offset = len(_MAGIC) + if not data.startswith(_MAGIC) or len(data) < offset + 8: + raise ValueError("Invalid QRF model artifact envelope.") + size = int.from_bytes(data[offset : offset + 8], "big") + offset += 8 + if not 0 < size < len(data) - offset: + raise ValueError("Invalid QRF model artifact header length.") + try: + metadata = json.loads(data[offset : offset + size]) + except (ValueError, UnicodeError) as error: + raise ValueError("Invalid QRF model artifact metadata.") from error + payload = data[offset + size :] + if ( + not isinstance(metadata, dict) + or type(metadata.get("schema_version")) is not int + or metadata["schema_version"] != 1 + or metadata.get("family") != "regime_gated_qrf" + or metadata.get("pickle_sha256") != hashlib.sha256(payload).hexdigest() + or metadata.get("source_weight_kind") + not in {"design", "importance", "calibrated"} + or metadata.get("fit_weight_kind") != "explicit" + ): + raise ValueError("Invalid QRF model artifact metadata or payload digest.") + # The executor verifies the content-store bytes before this trusted loader. + model = pickle.loads(payload) # noqa: S301 - trusted local graph artifact only + if ( + type(model) is not FittedRegimeGatedQRF + or model.entity is not None + or model.predictors != metadata.get("predictors") + or model.targets != metadata.get("targets") + or model.regimes() != metadata.get("regimes") + or model.weight_kind != metadata["fit_weight_kind"] + ): + raise ValueError("QRF model object does not match its artifact metadata.") + _names(tuple(model.predictors), "model predictors") + _names(tuple(model.targets), "model targets") + if set(model.predictors) & set(model.targets): + raise ValueError("QRF model predictors and targets overlap.") + return model diff --git a/packages/microcosm-fit/src/microcosm/fit/graph_apply.py b/packages/microcosm-fit/src/microcosm/fit/graph_apply.py new file mode 100644 index 00000000..be098b8c --- /dev/null +++ b/packages/microcosm-fit/src/microcosm/fit/graph_apply.py @@ -0,0 +1,114 @@ +"""Apply reusable QRF models independently of training adapter code.""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +import microcosm.fit._graph_qrf as shared_module +import microcosm.fit.qrf as qrf_module +import microcosm.graph.canonical as canonical_module +import microcosm.graph.randomness as randomness_module +from microcosm.fit._graph_qrf import QRF_MODEL_TYPE, _table, load_qrf_model +from microcosm.fit.kernels import FIT_QRF_DEPENDENCIES +from microcosm.graph import ( + ROWS_ALL, + Capabilities, + Determinism, + KernelBase, + KernelContext, + KernelResult, + Numeric, + SeedSource, + keyed_uniform, + source_hash, +) + +_APPLY_PARAMS = {"random_stream", "period"} + + +class QRFApplyKernel(KernelBase): + """Reuse one trained model and draw targets at stable recipient coordinates.""" + + ref = "fit.qrf.apply@1" + capabilities = Capabilities( + Determinism.SEEDED, + numeric=Numeric.PLATFORM_BITWISE, + seed_source=SeedSource.KEYED, + dependencies=FIT_QRF_DEPENDENCIES, + ) + + def implementation_hash(self): + return source_hash( + type(self), + shared_module, + qrf_module, + randomness_module, + canonical_module, + dependencies=self.capabilities.dependencies, + ) + + def run(self, context: KernelContext) -> KernelResult: + declared, table, id_column = _table(context, self.ref) + node = context.node + if ( + len(node.artifact_inputs) != 1 + or node.artifact_inputs[0].name != "model" + or node.artifact_inputs[0].type != QRF_MODEL_TYPE + or node.artifact_outputs + or set(context.artifacts) != {"model"} + ): + raise ValueError("QRF apply requires exactly one typed model input.") + if set(context.params) != _APPLY_PARAMS: + raise ValueError( + "QRF apply requires only random_stream and period parameters." + ) + period = context.params["period"] + if type(period) is not int: + raise ValueError("QRF apply period must be an integer.") + stream = context.params["random_stream"] + # Validate even an empty recipient table before loading the model. + keyed_uniform(stream=stream, keys=[]) + artifact = context.artifacts["model"] + model = load_qrf_model(artifact) + if declared.columns != tuple(model.predictors): + raise ValueError("QRF apply Slice does not match the model predictors.") + if tuple(o.column for o in node.outputs) != tuple(model.targets) or any( + o.entity != declared.entity or o.dtype != "float64" or o.rows != ROWS_ALL + for o in node.outputs + ): + raise ValueError( + "QRF apply must own every model target as all-row float64." + ) + arrays = { + kind: { + target: keyed_uniform( + stream=stream, + keys=[(i, "qrf", target, period, kind) for i in table[id_column]], + ) + for target in model.targets + } + for kind in ("quantiles", "sign_uniforms") + } + drawn = model.predict_from_uniforms(table, **arrays) + index = pd.Index(table[id_column].to_numpy(copy=True), name=id_column) + return KernelResult( + columns={ + (declared.entity, target): pd.Series( + drawn[target].to_numpy(dtype=np.float64, copy=True), + index=index, + name=target, + dtype="float64", + ) + for target in model.targets + }, + receipt={ + "entity": declared.entity, + "recipient_rows": len(table), + "model_key": artifact.key, + "model_producer_key": artifact.producer_key, + "random_stream": stream, + "period": period, + "seed_source": "keyed", + }, + ) diff --git a/packages/microcosm-fit/src/microcosm/fit/graph_models.py b/packages/microcosm-fit/src/microcosm/fit/graph_models.py new file mode 100644 index 00000000..d803a6ff --- /dev/null +++ b/packages/microcosm-fit/src/microcosm/fit/graph_models.py @@ -0,0 +1,13 @@ +"""Public split QRF graph API. + +Training and application have separate implementing modules so changing an +application adapter preserves donor fit identity. Shared QRF/model-envelope +changes intentionally invalidate both. Model payloads include trusted local +pickle bytes; content verification does not establish untrusted pickle safety. +""" + +from microcosm.fit._graph_qrf import QRF_MODEL_TYPE, load_qrf_model +from microcosm.fit.graph_apply import QRFApplyKernel +from microcosm.fit.graph_train import QRFTrainKernel + +__all__ = ["QRF_MODEL_TYPE", "QRFTrainKernel", "QRFApplyKernel", "load_qrf_model"] diff --git a/packages/microcosm-fit/src/microcosm/fit/graph_train.py b/packages/microcosm-fit/src/microcosm/fit/graph_train.py new file mode 100644 index 00000000..507434af --- /dev/null +++ b/packages/microcosm-fit/src/microcosm/fit/graph_train.py @@ -0,0 +1,116 @@ +"""Train reusable QRF models independently of application adapter code.""" + +from __future__ import annotations + +import microcosm.fit._graph_qrf as shared_module +import microcosm.fit.model as fit_model_module +import microcosm.fit.qrf as qrf_module +from microcosm.fit import fit as fit_qrf +from microcosm.fit._graph_qrf import QRF_MODEL_TYPE, _encode_model, _names, _table +from microcosm.fit.kernels import FIT_QRF_DEPENDENCIES +from microcosm.fit.qrf import DEFAULT_N_ESTIMATORS, DEFAULT_ZERO_ATOL +from microcosm.graph import ( + ArtifactOutput, + Capabilities, + Determinism, + KernelBase, + KernelContext, + KernelResult, + Numeric, + SeedSource, + source_hash, +) + +_TRAIN_PARAMS = { + "predictors", + "targets", + "seed", + "n_estimators", + "zero_atol", + "max_samples_leaf", +} + + +class QRFTrainKernel(KernelBase): + """Fit a weighted donor model; produce no recipient columns or draws.""" + + ref = "fit.qrf.train@1" + capabilities = Capabilities( + Determinism.SEEDED, + numeric=Numeric.PLATFORM_BITWISE, + seed_source=SeedSource.PARAM, + dependencies=FIT_QRF_DEPENDENCIES, + ) + + def implementation_hash(self): + return source_hash( + type(self), + shared_module, + fit_qrf, + fit_model_module, + qrf_module, + dependencies=self.capabilities.dependencies, + ) + + def run(self, context: KernelContext) -> KernelResult: + declared, table, _ = _table(context, self.ref) + node = context.node + if ( + node.outputs + or node.artifact_inputs + or node.artifact_outputs != (ArtifactOutput("model", QRF_MODEL_TYPE),) + ): + raise ValueError("QRF training owns exactly one typed model artifact.") + if set(context.params) - _TRAIN_PARAMS: + raise ValueError("Unknown QRF training parameters.") + predictors = _names(context.params.get("predictors"), "predictors") + targets = _names(context.params.get("targets"), "targets") + if set(predictors) & set(targets) or declared.columns != ( + *predictors, + *targets, + ): + raise ValueError( + "Training Slice must contain predictors then disjoint targets." + ) + seed = context.params.get("seed") + trees = context.params.get("n_estimators", DEFAULT_N_ESTIMATORS) + if type(seed) is not int or seed < 0 or type(trees) is not int or trees < 1: + raise ValueError( + "QRF training requires a nonnegative seed and positive tree count." + ) + weights = context.weights[declared.entity] + if len(weights) != len(table): + raise ValueError("QRF donor weights must align to the input rows.") + support = weights.values > 0 + if not support.any(): + raise ValueError("QRF training requires positive donor weight mass.") + # Zero-mass rows are outside this training population's support. Filter + # before regime detection to avoid bootstrapping an empty sign class. + donor = table.loc[support, [*predictors, *targets]].copy() + model = fit_qrf( + donor, + list(predictors), + list(targets), + weights=weights.values[support], + n_estimators=trees, + zero_atol=context.params.get("zero_atol", DEFAULT_ZERO_ATOL), + max_samples_leaf=context.params.get("max_samples_leaf"), + seed=seed, + ) + return KernelResult( + artifacts={"model": _encode_model(model, weights.kind.value)}, + receipt={ + "entity": declared.entity, + "predictors": predictors, + "targets": targets, + "donor_rows": len(donor), + "excluded_zero_weight_rows": int((~support).sum()), + "donor_weight_sum": float(weights.values[support].sum()), + "source_weight_kind": weights.kind.value, + "fit_weight_kind": model.weight_kind, + "regimes": model.regimes(), + "seed": seed, + "n_estimators": trees, + "seed_source": "param", + }, + ) diff --git a/packages/microcosm-fit/src/microcosm/fit/qrf.py b/packages/microcosm-fit/src/microcosm/fit/qrf.py index a39c57d2..e8bb2c95 100644 --- a/packages/microcosm-fit/src/microcosm/fit/qrf.py +++ b/packages/microcosm-fit/src/microcosm/fit/qrf.py @@ -1034,6 +1034,36 @@ def _draw_target_with_rng( return values +def _draw_target_from_uniforms( + features: pd.DataFrame, + model: _TargetModel, + quantiles: np.ndarray, + sign_uniforms: np.ndarray, +) -> np.ndarray: + """Evaluate a target without consuming or replacing any model RNG state.""" + if model is _RELEASED: + raise RuntimeError("This target's fitted forests were released; refit to draw.") + if model.regime == Regime.DEGENERATE_ZERO: + return np.zeros(len(features), dtype=np.float64) + if model.regime == Regime.POSITIVE_ONLY: + return model.positive.draw(features, quantiles) + if model.regime == Regime.NEGATIVE_ONLY: + return model.negative.draw(features, quantiles) + x = features.loc[:, list(model.columns)].to_numpy(dtype=np.float64) + cumulative = np.cumsum(model.gate.predict_proba(x), axis=1) + # Uniforms occupy [0, 1): strict comparison skips zero-probability classes + # even at u=0. Close the final CDF bin against floating-point roundoff. + cumulative[:, -1] = 1.0 + chosen = (cumulative > sign_uniforms[:, None]).argmax(axis=1) + signs = np.asarray(model.gate.classes_)[chosen] + values = np.zeros(len(features), dtype=np.float64) + for sign, forest in ((1, model.positive), (-1, model.negative)): + mask = signs == sign + if mask.any() and forest is not None: + values[mask] = forest.draw(features.loc[mask], quantiles[mask]) + return values + + class RegimeGatedQRF: """The canonical :class:`~microcosm.fit.model.ConditionalModel`. @@ -1564,6 +1594,60 @@ def predict( self._target_models[target] = _RELEASED return out + def predict_from_uniforms( + self, + frame_or_df: Frame | pd.DataFrame, + *, + quantiles: Mapping[str, np.ndarray], + sign_uniforms: Mapping[str, np.ndarray], + ) -> pd.DataFrame: + """Draw using caller-supplied per-row uniforms, without advancing RNG. + + Each mapping must contain exactly the fitted targets, with one finite + one-dimensional array in ``[0, 1)`` per target, aligned to input rows. + Supply both arrays even for single-sign or all-zero targets. Later + targets condition on earlier draws, just as in :meth:`predict`. + + Pairing uniforms with stable entity IDs makes results invariant to + recipient ordering and batching. Fitted forests remain reusable. The + legacy :meth:`predict` stream and its consumption order are unchanged. + """ + features = self._predictor_frame(frame_or_df) + arrays = {} + for name, supplied in ( + ("quantiles", quantiles), + ("sign_uniforms", sign_uniforms), + ): + if not isinstance(supplied, Mapping) or set(supplied) != set(self.targets): + raise ValueError(f"{name} must contain exactly the fitted targets.") + arrays[name] = {} + for target in self.targets: + values = np.asarray(supplied[target], dtype=np.float64) + if values.shape != (len(features),): + raise ValueError( + f"{name}[{target!r}] must have shape ({len(features)},)." + ) + if ( + not np.isfinite(values).all() + or ((values < 0) | (values >= 1)).any() + ): + raise ValueError(f"{name}[{target!r}] uniforms must be in [0, 1).") + arrays[name][target] = values + out = pd.DataFrame(index=features.index) + if features.empty: + return out.reindex(columns=self.targets).astype(np.float64) + augmented = features.copy() + for target in self.targets: + drawn = _draw_target_from_uniforms( + augmented, + self._target_models[target], + arrays["quantiles"][target], + arrays["sign_uniforms"][target], + ) + out[target] = drawn + augmented[target] = drawn + return out + def _predictor_frame(self, frame_or_df: Frame | pd.DataFrame) -> pd.DataFrame: """Extract the predictor columns from a Frame or DataFrame input.""" if isinstance(frame_or_df, Frame): diff --git a/packages/microcosm-fit/tests/test_graph_models.py b/packages/microcosm-fit/tests/test_graph_models.py new file mode 100644 index 00000000..3493e8c6 --- /dev/null +++ b/packages/microcosm-fit/tests/test_graph_models.py @@ -0,0 +1,206 @@ +"""Split model kernels fit once and apply by stable recipient identity.""" + +from dataclasses import replace +from importlib import import_module +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest + +from microcosm.fit import fit +from microcosm.fit.graph_models import ( + QRF_MODEL_TYPE, + QRFApplyKernel, + QRFTrainKernel, + load_qrf_model, +) +from microcosm.frame import WeightKind, Weights +from microcosm.graph import ( + ArtifactInput, + ArtifactOutput, + ArtifactValue, + KernelContext, + Node, + NumericScope, + Owned, + Slice, + keyed_uniform, +) + +STREAM = ("sha256-u53-v1", "test-qrf", 0, 19) + + +def context(node, table, artifact=None, weights=None): + return KernelContext( + node=node, + tables={"household": table}, + weights={ + "household": Weights( + np.ones(len(table)) if weights is None else weights, + WeightKind.DESIGN, + ) + }, + strata=pd.Series(dtype="string"), + params=node.params, + rng=np.random.default_rng(9), + artifacts={} if artifact is None else {"model": artifact}, + ) + + +@pytest.fixture(scope="module") +def trained(): + donor = pd.DataFrame( + { + "household_id": np.arange(60), + "x": np.arange(60) % 3, + "y": np.arange(60) + 1.0, + } + ) + node = Node( + "train", + QRFTrainKernel.ref, + inputs=(Slice("household", ("x", "y")),), + artifact_outputs=(ArtifactOutput("model", QRF_MODEL_TYPE),), + params={"predictors": ("x",), "targets": ("y",), "n_estimators": 4, "seed": 8}, + ) + result = QRFTrainKernel().run(context(node, donor)) + artifact = ArtifactValue( + result.artifacts["model"], QRF_MODEL_TYPE, "a" * 64, "b" * 64, NumericScope() + ) + return node, donor, result, artifact + + +def apply_context(table, artifact): + node = Node( + "apply", + QRFApplyKernel.ref, + inputs=(Slice("household", ("x",)),), + outputs=(Owned("household", "y", "float64"),), + artifact_inputs=(ArtifactInput("model", "train", "model", QRF_MODEL_TYPE),), + params={"random_stream": STREAM, "period": 2025}, + ) + return context(node, table, artifact) + + +def test_model_matches_public_fit_and_stateless_apply(trained): + node, donor, result, artifact = trained + model = load_qrf_model(artifact) + direct = fit( + donor[["x", "y"]], + ["x"], + ["y"], + weights=np.ones(len(donor)), + n_estimators=4, + seed=8, + ) + assert model.regimes() == direct.regimes() + assert result.receipt["source_weight_kind"] == "design" + assert result.receipt["fit_weight_kind"] == "explicit" + table = pd.DataFrame({"household_id": [21, 32, 87], "x": [0, 1, 2]}) + actual = ( + QRFApplyKernel().run(apply_context(table, artifact)).columns[("household", "y")] + ) + arrays = { + kind: { + "y": keyed_uniform( + stream=STREAM, + keys=[(i, "qrf", "y", 2025, kind) for i in table.household_id], + ) + } + for kind in ("quantiles", "sign_uniforms") + } + expected = direct.predict_from_uniforms(table, **arrays) + np.testing.assert_array_equal(actual.values, expected.y.values) + + +def test_reordering_and_unrelated_recipient_leave_draws_unchanged(trained): + artifact = trained[-1] + table = pd.DataFrame({"household_id": [21, 32, 87], "x": [0, 1, 2]}) + kernel = QRFApplyKernel() + first = kernel.run(apply_context(table, artifact)).columns[("household", "y")] + added = pd.concat([pd.DataFrame({"household_id": [1], "x": [1]}), table.iloc[::-1]]) + second = kernel.run(apply_context(added, artifact)).columns[("household", "y")] + pd.testing.assert_series_equal(first, second.loc[first.index]) + + +def test_zero_weight_support_excluded_before_regime_fit(trained): + node, donor, _, _ = trained + donor = donor.copy() + donor.loc[0, "y"] = -999.0 + weights = np.ones(len(donor)) + weights[0] = 0 + result = QRFTrainKernel().run(context(node, donor, weights=weights)) + assert result.receipt["regimes"] == {"y": "positive_only"} + assert result.receipt["excluded_zero_weight_rows"] == 1 + + +@pytest.mark.parametrize("mutation", ["type", "payload"]) +def test_invalid_model_artifact_rejected(trained, mutation): + artifact = trained[-1] + if mutation == "payload": + artifact = replace(artifact, payload=b"bad model") + else: + from microcosm.graph import ArtifactType + + artifact = replace(artifact, type=ArtifactType("other", 1)) + with pytest.raises(ValueError, match="model|artifact"): + load_qrf_model(artifact) + + +def test_training_honors_unequal_effective_weights(trained): + node, donor, _, _ = trained + donor = donor.copy() + donor["x"] = 1 + donor["y"] = np.where(np.arange(len(donor)) % 2, 100.0, 1.0) + weights = np.where(donor.y == 1, 1000.0, 1.0) + result = QRFTrainKernel().run(context(node, donor, weights=weights)) + artifact = ArtifactValue( + result.artifacts["model"], QRF_MODEL_TYPE, "a" * 64, "b" * 64, NumericScope() + ) + recipient = pd.DataFrame({"household_id": np.arange(200), "x": 1}) + weighted = QRFApplyKernel().run(apply_context(recipient, artifact)) + assert weighted.columns[("household", "y")].mean() < 3.0 + assert result.receipt["donor_weight_sum"] == weights.sum() + + +def test_distinct_streams_make_distinct_applications(trained): + recipient = pd.DataFrame({"household_id": np.arange(100), "x": 1}) + first = apply_context(recipient, trained[-1]) + node = replace( + first.node, + params={"random_stream": (*STREAM[:2], 1, STREAM[3]), "period": 2025}, + ) + second = replace(first, node=node, params=node.params) + first_draws = QRFApplyKernel().run(first).columns[("household", "y")] + second_draws = QRFApplyKernel().run(second).columns[("household", "y")] + assert not np.array_equal(first_draws, second_draws) + + +@pytest.mark.parametrize( + "module", + [ + QRFApplyKernel.__module__, + "microcosm.graph.randomness", + "microcosm.graph.canonical", + ], +) +def test_application_code_change_preserves_training_identity(monkeypatch, module): + train = QRFTrainKernel() + apply = QRFApplyKernel() + fit_before = train.implementation_hash() + apply_before = apply.implementation_hash() + application_source = Path(import_module(module).__file__).resolve() + original = Path.read_bytes + + def changed_source(path): + payload = original(path) + return ( + payload + b"\n# application-only change\n" + if path.resolve() == application_source + else payload + ) + + monkeypatch.setattr(Path, "read_bytes", changed_source) + assert train.implementation_hash() == fit_before + assert apply.implementation_hash() != apply_before diff --git a/packages/microcosm-fit/tests/test_qrf_stateless.py b/packages/microcosm-fit/tests/test_qrf_stateless.py new file mode 100644 index 00000000..e051aa03 --- /dev/null +++ b/packages/microcosm-fit/tests/test_qrf_stateless.py @@ -0,0 +1,118 @@ +"""Caller-owned uniforms make repeated QRF draws stable by identity.""" + +import copy + +import numpy as np +import pandas as pd +import pytest + +from microcosm.fit import fit + + +@pytest.fixture(scope="module") +def model(): + x = np.tile(np.arange(30, dtype=float), 6) + donor = pd.DataFrame( + { + "x": x, + "positive": x + 1, + "negative": -x - 1, + "zero": np.zeros(len(x)), + "mixed": np.tile([-2.0, 0.0, 3.0], len(x) // 3), + "inflated": np.tile([0.0, 4.0], len(x) // 2), + "negative_inflated": np.tile([0.0, -4.0], len(x) // 2), + "two_sign": np.tile([-3.0, 4.0], len(x) // 2), + } + ) + return fit( + donor, + ["x"], + list(donor.columns[1:]), + weights="none", + n_estimators=4, + seed=7, + ) + + +def uniforms(model, n): + return { + "quantiles": {t: np.linspace(0, 0.99, n) for t in model.targets}, + "sign_uniforms": {t: np.linspace(0.99, 0, n) for t in model.targets}, + } + + +def test_stateless_draws_preserve_rng_and_forests(model): + recipient = pd.DataFrame({"x": np.arange(20, dtype=float)}) + before = copy.deepcopy(model._rng.bit_generator.state) + draws = uniforms(model, len(recipient)) + first = model.predict_from_uniforms(recipient, **draws) + pd.testing.assert_frame_equal( + first, model.predict_from_uniforms(recipient, **draws) + ) + assert model._rng.bit_generator.state == before + assert (first.positive > 0).all() + assert (first.negative < 0).all() + assert (first.zero == 0).all() + assert set(first.mixed) == {-2.0, 0.0, 3.0} + assert set(first.inflated) == {0.0, 4.0} + assert set(first.negative_inflated) == {0.0, -4.0} + assert set(first.two_sign) == {-3.0, 4.0} + + +def test_permutation_and_chunking_preserve_chained_draws(model): + recipient = pd.DataFrame({"x": np.arange(20, dtype=float)}) + draws = uniforms(model, len(recipient)) + expected = model.predict_from_uniforms(recipient, **draws) + order = np.random.default_rng(4).permutation(len(recipient)) + reordered = model.predict_from_uniforms( + recipient.iloc[order], + **{k: {t: v[order] for t, v in d.items()} for k, d in draws.items()}, + ) + pd.testing.assert_frame_equal(expected, reordered.sort_index()) + chunks = [] + for rows in (slice(0, 7), slice(7, 20)): + chunks.append( + model.predict_from_uniforms( + recipient.iloc[rows], + **{k: {t: v[rows] for t, v in d.items()} for k, d in draws.items()}, + ) + ) + pd.testing.assert_frame_equal(expected, pd.concat(chunks)) + + +@pytest.mark.parametrize("bad", [[-0.1, 0.5], [0.1, 1.0], [np.nan, 0.1], [0.1]]) +@pytest.mark.parametrize("field", ["quantiles", "sign_uniforms"]) +def test_invalid_uniforms_rejected_before_drawing(model, bad, field): + recipient = pd.DataFrame({"x": [1.0, 2.0]}) + draws = uniforms(model, 2) + draws[field][model.targets[-1]] = np.array(bad) + with pytest.raises(ValueError, match="uniform|shape"): + model.predict_from_uniforms(recipient, **draws) + + +def test_uniform_target_names_must_match(model): + draws = uniforms(model, 2) + del draws["quantiles"][model.targets[-1]] + with pytest.raises(ValueError, match="targets"): + model.predict_from_uniforms(pd.DataFrame({"x": [1.0, 2.0]}), **draws) + + +def test_empty_recipient_batch(model): + actual = model.predict_from_uniforms( + pd.DataFrame({"x": pd.Series(dtype=float)}), **uniforms(model, 0) + ) + assert list(actual.columns) == model.targets + assert actual.empty + assert all(dtype == np.dtype("float64") for dtype in actual.dtypes) + + +def test_zero_uniform_skips_a_zero_probability_sign(model, monkeypatch): + # Exercise the inverse-CDF boundary that ordinary RNG draws almost never hit. + gate = model._target_models["mixed"].gate + monkeypatch.setattr( + gate, "predict_proba", lambda x: np.tile([0.0, 0.0, 1.0], (len(x), 1)) + ) + draws = uniforms(model, 2) + draws["sign_uniforms"]["mixed"] = np.zeros(2) + actual = model.predict_from_uniforms(pd.DataFrame({"x": [1.0, 2.0]}), **draws) + assert (actual.mixed == 3.0).all() diff --git a/packages/microcosm-graph/README.md b/packages/microcosm-graph/README.md index cb0d4614..aca5ad87 100644 --- a/packages/microcosm-graph/README.md +++ b/packages/microcosm-graph/README.md @@ -32,3 +32,53 @@ Module map: The shard depends on `microcosm-frame` only. Kernels that wrap fit, calibrate, or a rules engine live in those shards and register here. + +## Reusable typed artifacts + +A fitted model can be a dependency across population versions without owning a +population column. `ArtifactOutput("model", ArtifactType("example.model", 1))` +declares a required byte output. A consumer names it with +`ArtifactInput("fitted", "train", "model", ArtifactType("example.model", 1))` +and reads `context.artifacts["fitted"].payload`. The executor exposes only the +declared aliases as immutable `ArtifactValue` objects, including producer +identity and numeric scope. A nominal type/version is an interface contract; +the consuming kernel must validate its decoded payload. Existing opaque +model/diagnostic bytes remain supported, but cannot satisfy an edge unless the +producer declares their type. + +The compiler adds these edges to dependency ordering, cycle checks, and gate +ancestry. A change to recipient inputs can reuse the fitted producer. Typed +node cache records use schema 2; runs carrying typed edges or outputs use +manifest schema 3. Legacy nodes omit the new empty declarations from keys and +JSON, and legacy runs retain manifest schema 2. Cache reload validates the same +contracts as fresh execution. + +Numeric contracts are deliberately restrictive: bitwise artifacts permit any +consumer class; platform-bitwise artifacts require platform-bitwise consumers; +tolerance-bound artifacts require tolerance-bound consumers with their own +output tolerance. Mixed platform/tolerance artifact inputs are refused. The +executor does not infer error propagation through an arbitrary computation. + +## Stable random coordinates + +Opt-in kernels declare `SeedSource.KEYED`, put their stream tuple in normative +node parameters, and include `microcosm.graph.randomness` in their implementation +hash. The existing node-key RNG behavior remains available unchanged. + +```python +from microcosm.graph import keyed_uniform + +stream = ("sha256-u53-v1", "comparison", 0, 42) +values = keyed_uniform( + stream=stream, + keys=[(person_id, "mortality", 2027, 0) for person_id in person_ids], +) +``` + +The read-only float64 result is stable through reordering, chunk boundaries, and +unrelated inserted identities. Duplicate coordinates intentionally repeat a +draw. Integer and string identities differ. The versioned algorithm hashes a +canonical, type-tagged coordinate tuple with the stream and maps the leading +53 digest bits to `[0, 1)`. This is an explicit experiment stream independent +of cache identity; changing its normative specification still invalidates the +application node. diff --git a/packages/microcosm-graph/src/microcosm/graph/__init__.py b/packages/microcosm-graph/src/microcosm/graph/__init__.py index aaa78148..c66bac7f 100644 --- a/packages/microcosm-graph/src/microcosm/graph/__init__.py +++ b/packages/microcosm-graph/src/microcosm/graph/__init__.py @@ -16,6 +16,9 @@ PARTITION_DTYPES, ROWS_ALL, WEIGHT_KINDS, + ArtifactInput, + ArtifactOutput, + ArtifactType, CompiledGraph, Graph, GraphError, @@ -37,6 +40,7 @@ StoreUnavailableError, ) from .kernel import ( + ArtifactValue, Capabilities, Determinism, Kernel, @@ -52,8 +56,14 @@ source_hash, ) from .keys import platform_fingerprint +from .randomness import keyed_uniform __all__ = [ + "ArtifactInput", + "ArtifactOutput", + "ArtifactType", + "ArtifactValue", + "keyed_uniform", "platform_fingerprint", "DESCRIPTIVE_FIELDS", "DTYPES", diff --git a/packages/microcosm-graph/src/microcosm/graph/artifact_edges.py b/packages/microcosm-graph/src/microcosm/graph/artifact_edges.py new file mode 100644 index 00000000..0b84849e --- /dev/null +++ b/packages/microcosm-graph/src/microcosm/graph/artifact_edges.py @@ -0,0 +1,164 @@ +"""Typed byte edges: declarations, numeric refusal, and portable descriptors.""" + +from __future__ import annotations + +from collections.abc import Mapping + +from .decl import ArtifactType, CompiledGraph, Node +from .errors import NodeRejectedError +from .kernel import ( + ArtifactValue, + Capabilities, + KernelRegistry, + Numeric, + NumericScope, + Tolerance, +) +from .keys import opaque_artifact_key, platform_fingerprint + + +def numeric_scope(capabilities: Capabilities) -> NumericScope: + return NumericScope( + numeric=capabilities.numeric, + tolerance=capabilities.tolerance, + platform=platform_fingerprint() + if capabilities.numeric is Numeric.PLATFORM_BITWISE + else None, + ) + + +def scope_payload(scope: NumericScope) -> dict[str, object]: + return { + "numeric": scope.numeric.value, + "tolerance": None + if scope.tolerance is None + else { + "rtol": scope.tolerance.rtol, + "atol": scope.tolerance.atol, + "ulps": scope.tolerance.ulps, + }, + "platform": scope.platform, + } + + +def scope_from_payload(raw: object) -> NumericScope: + if not isinstance(raw, Mapping) or set(raw) != {"numeric", "tolerance", "platform"}: + raise ValueError("Malformed artifact numeric scope.") + tolerance = raw["tolerance"] + if tolerance is not None: + if not isinstance(tolerance, Mapping) or set(tolerance) != { + "rtol", + "atol", + "ulps", + }: + raise ValueError("Malformed artifact tolerance.") + tolerance = Tolerance(**tolerance) + return NumericScope( + Numeric(raw["numeric"]), tolerance=tolerance, platform=raw["platform"] + ) + + +def require_compatible_scope(scope: NumericScope, consumer: Capabilities) -> None: + """Refuse scope laundering; output tolerance is the consumer's own contract.""" + if scope.platform is not None and scope.numeric is Numeric.TOLERANCE_BOUND: + raise NodeRejectedError( + "Typed artifacts combining platform and tolerance scopes are unsupported." + ) + if ( + scope.numeric is Numeric.PLATFORM_BITWISE + and consumer.numeric is not Numeric.PLATFORM_BITWISE + ): + raise NodeRejectedError( + "A platform_bitwise artifact requires a platform_bitwise consumer." + ) + if ( + scope.numeric is Numeric.TOLERANCE_BOUND + and consumer.numeric is not Numeric.TOLERANCE_BOUND + ): + raise NodeRejectedError( + "A tolerance_bound artifact requires a tolerance_bound consumer." + ) + + +def descriptor( + *, + producer: str, + artifact: str, + type_: ArtifactType, + producer_key: str, + capabilities: Capabilities, +) -> dict[str, object]: + return { + "producer": producer, + "artifact": artifact, + "producer_key": producer_key, + "key": opaque_artifact_key(producer_key, artifact), + "type": {"name": type_.name, "schema_version": type_.schema_version}, + "numerics": scope_payload(numeric_scope(capabilities)), + } + + +def typed_contracts( + compiled: CompiledGraph, + node: Node, + keys: Mapping[str, str], + kernels: KernelRegistry, +) -> dict[str, object]: + if not node.artifact_inputs and not node.artifact_outputs: + return {} + consumer = kernels.get(node.kernel).capabilities + inputs = {} + for binding in node.artifact_inputs: + producer = compiled.graph.node(binding.producer) + capabilities = kernels.get(producer.kernel).capabilities + require_compatible_scope(numeric_scope(capabilities), consumer) + inputs[binding.name] = descriptor( + producer=producer.id, + artifact=binding.artifact, + type_=binding.type, + producer_key=keys[producer.id], + capabilities=capabilities, + ) + return { + "inputs": inputs, + "outputs": { + output.name: descriptor( + producer=node.id, + artifact=output.name, + type_=output.type, + producer_key=keys[node.id], + capabilities=consumer, + ) + for output in node.artifact_outputs + }, + } + + +def value_from_descriptor(payload: bytes, raw: object) -> ArtifactValue: + if not isinstance(raw, Mapping) or set(raw) != { + "producer", + "artifact", + "producer_key", + "key", + "type", + "numerics", + }: + raise ValueError("Malformed typed artifact descriptor.") + if any( + not isinstance(raw[name], str) or not raw[name] + for name in ("producer", "artifact") + ): + raise ValueError("Typed artifact producer/name must be nonempty strings.") + type_raw = raw["type"] + if not isinstance(type_raw, Mapping) or set(type_raw) != {"name", "schema_version"}: + raise ValueError("Malformed typed artifact type.") + value = ArtifactValue( + payload, + ArtifactType(type_raw["name"], type_raw["schema_version"]), + raw["key"], + raw["producer_key"], + scope_from_payload(raw["numerics"]), + ) + if value.key != opaque_artifact_key(value.producer_key, raw["artifact"]): + raise ValueError("Typed artifact identity does not match its producer.") + return value diff --git a/packages/microcosm-graph/src/microcosm/graph/decl.py b/packages/microcosm-graph/src/microcosm/graph/decl.py index b25e4f62..b6134c21 100644 --- a/packages/microcosm-graph/src/microcosm/graph/decl.py +++ b/packages/microcosm-graph/src/microcosm/graph/decl.py @@ -48,6 +48,9 @@ from types import MappingProxyType __all__ = [ + "ArtifactType", + "ArtifactInput", + "ArtifactOutput", "DESCRIPTIVE_FIELDS", "DTYPES", "GATE_OUTCOMES", @@ -175,6 +178,48 @@ def __post_init__(self) -> None: _nonempty("SourceRef.codec", self.codec) +@dataclass(frozen=True) +class ArtifactType: + """Nominal, versioned byte-payload contract; consumers validate the payload.""" + + name: str + schema_version: int + + def __post_init__(self) -> None: + _nonempty("ArtifactType.name", self.name) + if type(self.schema_version) is not int or self.schema_version < 1: + raise GraphError("ArtifactType.schema_version must be a positive integer.") + + +@dataclass(frozen=True) +class ArtifactOutput: + """A named, typed subset of a kernel's existing opaque byte outputs.""" + + name: str + type: ArtifactType + + def __post_init__(self) -> None: + _nonempty("ArtifactOutput.name", self.name) + if not isinstance(self.type, ArtifactType): + raise GraphError("ArtifactOutput.type must be an ArtifactType.") + + +@dataclass(frozen=True) +class ArtifactInput: + """A declared artifact edge, with a consumer-local alias and exact type.""" + + name: str + producer: str + artifact: str + type: ArtifactType + + def __post_init__(self) -> None: + for field_name in ("name", "producer", "artifact"): + _nonempty(f"ArtifactInput.{field_name}", getattr(self, field_name)) + if not isinstance(self.type, ArtifactType): + raise GraphError("ArtifactInput.type must be an ArtifactType.") + + @dataclass(frozen=True) class Slice: """What a node reads: columns of one entity, optionally under a row mask. @@ -279,6 +324,11 @@ class Node: Attributes: id: Unique within the graph. kernel: Kernel reference, e.g. ``"fit.qrf@1"``. + artifact_inputs: Typed byte dependencies, including other populations. + Each local alias names a declared producer output of exactly the + expected nominal type/version. + artifact_outputs: Required typed outputs within KernelResult.artifacts; + other untyped diagnostic bytes remain legal. inputs: Slices the kernel receives. Their owners are this node's predecessors. outputs: Cells this node owns. A ``CREATE`` node declares every @@ -318,10 +368,25 @@ class Node: description: str = "" citation: str = "" entrants: bool = False + artifact_inputs: tuple[ArtifactInput, ...] = () + artifact_outputs: tuple[ArtifactOutput, ...] = () def __post_init__(self) -> None: _nonempty("Node.id", self.id) _nonempty("Node.kernel", self.kernel) + for name, kind in ( + ("artifact_inputs", ArtifactInput), + ("artifact_outputs", ArtifactOutput), + ): + declarations = getattr(self, name) + if not isinstance(declarations, tuple) or any( + not isinstance(item, kind) for item in declarations + ): + raise GraphError( + f"Node {self.id!r}: {name} must be a tuple of {kind.__name__}." + ) + if len({item.name for item in declarations}) != len(declarations): + raise GraphError(f"Node {self.id!r}: duplicate names in {name}.") if not isinstance(self.structural, StructuralDelta): raise GraphError(f"Node {self.id!r}: structural must be a StructuralDelta.") if self.mass not in MASS_POLICIES: @@ -417,6 +482,10 @@ def normative(self) -> dict[str, object]: f.name: getattr(self, f.name) for f in fields(self) if f.name not in DESCRIPTIVE_FIELDS + and not ( + f.name in {"artifact_inputs", "artifact_outputs"} + and not getattr(self, f.name) + ) } @@ -678,6 +747,30 @@ def check_mask(node_id: str, version: str, entity: str, mask: str) -> None: f"but the incumbent is declared {base_dtype!r}." ) + # Artifact edges cross population versions, without changing cell ownership. + for node in graph.nodes: + for binding in node.artifact_inputs: + if binding.producer == node.id: + raise GraphError( + f"Node {node.id!r} depends on itself through an artifact." + ) + producer = by_id.get(binding.producer) + if producer is None: + raise GraphError( + f"Node {node.id!r}: unknown artifact producer {binding.producer!r}." + ) + outputs = {output.name: output for output in producer.artifact_outputs} + output = outputs.get(binding.artifact) + if output is None: + raise GraphError( + f"Node {node.id!r}: producer {producer.id!r} has no declared artifact {binding.artifact!r}." + ) + if output.type != binding.type: + raise GraphError( + f"Node {node.id!r}: artifact {binding.artifact!r} type does not match its producer." + ) + predecessors[node.id].add(producer.id) + depth: dict[str, int] = {} def depth_of(node_id: str, trail: tuple[str, ...]) -> int: diff --git a/packages/microcosm-graph/src/microcosm/graph/executor.py b/packages/microcosm-graph/src/microcosm/graph/executor.py index 689c1cf9..4f825ad8 100644 --- a/packages/microcosm-graph/src/microcosm/graph/executor.py +++ b/packages/microcosm-graph/src/microcosm/graph/executor.py @@ -17,6 +17,7 @@ from microcosm.frame import Frame, WeightKind, Weights from . import keys as graph_keys +from .artifact_edges import scope_payload, typed_contracts, value_from_descriptor from .canonical import canonical_json, sha256_domain from .codecs import SOURCE_CODECS, SourceCodecRegistry from .decl import ( @@ -30,6 +31,7 @@ ) from .errors import NodeRejectedError from .kernel import ( + ArtifactValue, Capabilities, KernelContext, KernelRegistry, @@ -386,6 +388,21 @@ def _context_digest(context: KernelContext) -> bytes: digest.update(weights.kind.value.encode("ascii") + b"\0") _update_array(digest, weights.values) _update_series(digest, context.strata) + for name, value in sorted(context.artifacts.items()): + digest.update( + canonical_json( + ( + name, + value.key, + value.producer_key, + value.type.name, + value.type.schema_version, + scope_payload(value.numerics), + ) + ) + ) + digest.update(len(value.payload).to_bytes(8, "little")) + digest.update(value.payload) return digest.digest() @@ -461,6 +478,7 @@ def _project_context( sources: Mapping[str, Path], tolerances: Mapping[tuple[str, str], Tolerance | None], numerics: Mapping[tuple[str, str], NumericScope], + artifacts: Mapping[str, ArtifactValue] | None = None, ) -> KernelContext: if population is None: return KernelContext( @@ -473,6 +491,7 @@ def _project_context( sources=MappingProxyType({name: sources[name] for name in node.sources}), tolerances=tolerances, numerics=numerics, + artifacts={} if artifacts is None else artifacts, ) frame = population.frame @@ -564,6 +583,7 @@ def _project_context( sources=MappingProxyType({name: sources[name] for name in node.sources}), tolerances=tolerances, numerics=numerics, + artifacts={} if artifacts is None else artifacts, ) @@ -1126,6 +1146,12 @@ def _validate_result( if not isinstance(payload, bytes): raise NodeRejected(f"Node {node.id!r} artifact {name!r} is not bytes.") artifacts[name] = payload + for output in node.artifact_outputs: + if output.name not in artifacts: + error = StoreMiss if cache_hit else NodeRejected + raise error( + f"Node {node.id!r} is missing declared artifact {output.name!r}." + ) receipt = _normal_json_mapping(result.receipt, f"Node {node.id!r} receipt") if node.structural is StructuralDelta.EXPAND: if cache_hit: @@ -1378,6 +1404,7 @@ def _write_node( receipt: Mapping[str, object], opaque_artifacts: Mapping[str, bytes], verify_existing: bool, + typed_artifacts: Mapping[str, object] | None = None, ) -> tuple[dict[tuple[str, str], str], dict[str, object]]: columns: dict[tuple[str, str], tuple[pd.Series, str]] = {} if node.structural is StructuralDelta.NONE: @@ -1462,7 +1489,8 @@ def _write_node( opaque_entries.append({"name": name, "key": output_key}) record: dict[str, object] = { - "schema_version": 1, + "schema_version": 2 if typed_artifacts else 1, + **({"typed_artifacts": dict(typed_artifacts)} if typed_artifacts else {}), "node_id": node.id, "node_key": key, "kernel_ref": node.kernel, @@ -1490,6 +1518,7 @@ def _require_record_shape( key: str, kernel_impl_hash: str, capabilities: Capabilities, + typed_artifacts: Mapping[str, object] | None = None, ) -> dict[str, object]: if not isinstance(raw, dict): raise StoreCorrupt(f"Cached receipt for node {node.id!r} is not an object.") @@ -1506,16 +1535,37 @@ def _require_record_shape( "weight", "opaque", } + if typed_artifacts: + required.add("typed_artifacts") if set(raw) != required: raise StoreCorrupt( f"Cached receipt for node {node.id!r} has fields {sorted(raw)}, " f"not {sorted(required)}." ) - if raw["schema_version"] != 1: + if raw["schema_version"] != (2 if typed_artifacts else 1): raise StoreUnavailable( f"Cached receipt for node {node.id!r} uses unsupported schema " f"{raw['schema_version']!r}." ) + if typed_artifacts and raw.get("typed_artifacts") != dict(typed_artifacts): + raise StoreCorrupt( + f"Cached node {node.id!r} typed artifact contracts disagree with the graph." + ) + if typed_artifacts: + opaque = _record_entries(raw, "opaque") + names = [entry.get("name") for entry in opaque] + if len(set(names)) != len(names): + raise StoreCorrupt(f"Cached node {node.id!r} repeats an opaque artifact.") + actual_outputs = {entry.get("name"): entry.get("key") for entry in opaque} + for output in node.artifact_outputs: + if output.name not in actual_outputs: + raise StoreMiss( + f"Cached node {node.id!r} is missing declared artifact {output.name!r}." + ) + if actual_outputs[output.name] != _opaque_artifact_key(key, output.name): + raise StoreCorrupt( + f"Cached node {node.id!r} artifact identity mismatch." + ) expected = (node.id, key, node.kernel, kernel_impl_hash) actual = ( raw["node_id"], @@ -1581,6 +1631,7 @@ def _load_record( key: str, kernel_impl_hash: str, capabilities: Capabilities, + typed_artifacts: Mapping[str, object] | None = None, ) -> dict[str, object]: raw = store.load_json(_cache_record_key(key)) return _require_record_shape( @@ -1589,6 +1640,7 @@ def _load_record( key=key, kernel_impl_hash=kernel_impl_hash, capabilities=capabilities, + typed_artifacts=typed_artifacts, ) @@ -1867,6 +1919,7 @@ def _preflight_require( key=keys[node_id], kernel_impl_hash=implementations[node_id], capabilities=kernels.get(node.kernel).capabilities, + typed_artifacts=typed_contracts(compiled, node, keys, kernels), ) _require_tolerance_writer_receipt( node, @@ -1926,6 +1979,10 @@ def run_graph( started_at = _now() source_paths, source_keys = _source_paths_and_keys(compiled, sources, store) keys, implementations = _all_node_keys(compiled, kernels, source_keys) + contracts = { + node_id: typed_contracts(compiled, compiled.graph.node(node_id), keys, kernels) + for node_id in compiled.order + } if resume == "require": _preflight_require(compiled, store, keys, implementations, kernels) @@ -1965,6 +2022,18 @@ def run_graph( ) tolerance_writers = _tolerance_writer_payload(input_writers) + typed = contracts[node_id] + artifact_values = {} + for binding in node.artifact_inputs: + entry = typed["inputs"][binding.name] + producer_receipt = receipts[binding.producer] + if producer_receipt.opaque_artifacts.get(binding.artifact) != entry["key"]: + raise StoreCorrupt( + f"Node {node.id!r} artifact producer receipt disagrees with its declaration." + ) + artifact_values[binding.name] = value_from_descriptor( + store.load_bytes(entry["key"]), entry + ) hit = False replace_stale_record = False result: KernelResult | None = None @@ -1978,6 +2047,7 @@ def run_graph( key=key, kernel_impl_hash=implementation, capabilities=kernel.capabilities, + typed_artifacts=typed, ) try: _require_tolerance_writer_receipt( @@ -2005,6 +2075,7 @@ def run_graph( sources=source_paths, tolerances=input_tolerances, numerics=input_numerics, + artifacts=artifact_values, ) before = _context_digest(context) try: @@ -2149,6 +2220,7 @@ def run_graph( receipt=cache_receipt, opaque_artifacts=opaque, verify_existing=(resume != "forbid" and not replace_stale_record), + typed_artifacts=typed, ) assert record is not None @@ -2172,6 +2244,7 @@ def run_graph( receipt_opaque[name] = artifact_identity receipts[node_id] = NodeReceipt( + typed_artifacts=typed, key=key, hit=hit, seed=seed(key), diff --git a/packages/microcosm-graph/src/microcosm/graph/explain.py b/packages/microcosm-graph/src/microcosm/graph/explain.py index 9d10d8be..26abfdbd 100644 --- a/packages/microcosm-graph/src/microcosm/graph/explain.py +++ b/packages/microcosm-graph/src/microcosm/graph/explain.py @@ -506,6 +506,11 @@ def _receipt_payload(receipt: NodeReceipt) -> dict[str, object]: "frame_key": receipt.frame_key, "weight_key": receipt.weight_key, "opaque_artifacts": receipt.opaque_artifacts, + **( + {"typed_artifacts": receipt.typed_artifacts} + if receipt.typed_artifacts + else {} + ), "wall_time": receipt.wall_time, } if receipt.legacy_capabilities: diff --git a/packages/microcosm-graph/src/microcosm/graph/kernel.py b/packages/microcosm-graph/src/microcosm/graph/kernel.py index 567a57ea..0846a3ea 100644 --- a/packages/microcosm-graph/src/microcosm/graph/kernel.py +++ b/packages/microcosm-graph/src/microcosm/graph/kernel.py @@ -59,9 +59,10 @@ from microcosm.frame import Frame, Weights -from .decl import Node, Param, StructuralDelta +from .decl import ArtifactType, Node, Param, StructuralDelta __all__ = [ + "ArtifactValue", "Capabilities", "Determinism", "Kernel", @@ -151,6 +152,7 @@ class SeedSource(StrEnum): EXECUTOR = "executor" # ``KernelContext.rng``, derived from the node key PARAM = "param" # a literal ``seed`` parameter (legacy parity kernels) + KEYED = "keyed" # normative stream params and stable draw coordinates NONE = "none" @@ -270,6 +272,33 @@ def __post_init__(self) -> None: raise ValueError("A bitwise scope holds on every platform.") +@dataclass(frozen=True) +class ArtifactValue: + """Verified immutable bytes and the executor's typed producer provenance.""" + + payload: bytes + type: ArtifactType + key: str + producer_key: str + numerics: NumericScope + + def __post_init__(self) -> None: + if not isinstance(self.payload, bytes): + raise TypeError("ArtifactValue.payload must be immutable bytes.") + if not isinstance(self.type, ArtifactType) or not isinstance( + self.numerics, NumericScope + ): + raise TypeError("ArtifactValue requires an ArtifactType and NumericScope.") + for name in ("key", "producer_key"): + value = getattr(self, name) + if ( + not isinstance(value, str) + or len(value) != 64 + or any(c not in "0123456789abcdef" for c in value) + ): + raise ValueError(f"ArtifactValue.{name} must be a SHA-256 identity.") + + @dataclass(frozen=True) class KernelContext: """Everything a kernel may read. Nothing here is writable. @@ -286,8 +315,12 @@ class KernelContext: in the node's inputs or outputs. strata: Read-only per-person strata of the population version. params: The node's parameters. - rng: A generator seeded from the node key. The only randomness a - kernel may use. + rng: The default generator seeded from the node key. KEYED kernels + instead use normative stream params and stable coordinates through + keyed_uniform; PARAM kernels use their declared literal seed. + artifacts: Immutable typed bytes for declared artifact aliases only. + Consumers validate versioned payloads before using them; nominal + types do not themselves verify arbitrary serialized data. sources: Source name to a content-verified path, for declared sources only. tolerances: ``(entity, column)`` of each declared input column to @@ -309,6 +342,20 @@ class KernelContext: sources: Mapping[str, Path] = field(default_factory=dict) tolerances: Mapping[tuple[str, str], Tolerance | None] = field(default_factory=dict) numerics: Mapping[tuple[str, str], NumericScope] = field(default_factory=dict) + artifacts: Mapping[str, ArtifactValue] = field(default_factory=dict) + + def __post_init__(self) -> None: + values = dict(self.artifacts) + if any( + not isinstance(name, str) + or not name + or not isinstance(value, ArtifactValue) + for name, value in values.items() + ): + raise TypeError( + "KernelContext.artifacts must map non-empty aliases to ArtifactValue." + ) + object.__setattr__(self, "artifacts", MappingProxyType(values)) @dataclass(frozen=True) diff --git a/packages/microcosm-graph/src/microcosm/graph/keys.py b/packages/microcosm-graph/src/microcosm/graph/keys.py index 43939dba..99bb9f5c 100644 --- a/packages/microcosm-graph/src/microcosm/graph/keys.py +++ b/packages/microcosm-graph/src/microcosm/graph/keys.py @@ -13,6 +13,7 @@ from .kernel import Capabilities, Numeric __all__ = [ + "opaque_artifact_key", "platform_fingerprint", "artifact_key", "frame_key", @@ -88,6 +89,11 @@ def artifact_key(node_key: str, entity: str, column: str) -> str: return _hash_parts("artifact", node_key, entity, column) +def opaque_artifact_key(node_key: str, name: str) -> str: + """Identity of an opaque or typed byte output (legacy domain preserved).""" + return _hash_parts("node-artifact", node_key, name) + + def frame_key(node_key: str) -> str: """Derive the structural frame artifact identity from its node.""" @@ -240,6 +246,27 @@ def node_key( if kernel_capabilities.numeric is Numeric.PLATFORM_BITWISE else () ) + typed_inputs = ( + ( + { + "typed_artifacts": tuple( + ( + item.name, + opaque_artifact_key( + _required_key(input_keys, item.producer, node_id), + item.artifact, + ), + normative(item.type), + ) + for item in sorted( + node.artifact_inputs, key=lambda value: value.name + ) + ) + }, + ) + if node.artifact_inputs + else () + ) return _hash_parts( "node", normative(node), @@ -250,6 +277,7 @@ def node_key( graph_facts, capabilities, *platform_scope, + *typed_inputs, ) diff --git a/packages/microcosm-graph/src/microcosm/graph/manifest.py b/packages/microcosm-graph/src/microcosm/graph/manifest.py index 90b4fa3b..34b121a7 100644 --- a/packages/microcosm-graph/src/microcosm/graph/manifest.py +++ b/packages/microcosm-graph/src/microcosm/graph/manifest.py @@ -13,6 +13,7 @@ from microcosm.frame import Frame +from .artifact_edges import require_compatible_scope, value_from_descriptor from .canonical import canonical_json, sha256_domain from .decl import GATE_OUTCOMES, StructuralDelta from .errors import NodeRejectedError, StoreCorruptError @@ -34,6 +35,7 @@ __all__ = ["Decision", "NodeReceipt", "PopulationView", "RunManifest"] _SCHEMA_VERSION = 2 +_TYPED_SCHEMA_VERSION = 3 _LEGACY_SCHEMA_VERSION = 1 _CERTIFYING_GATE_OUTCOMES = frozenset({"pass", "not_applicable"}) @@ -220,6 +222,7 @@ class NodeReceipt: weight_key: str | None = None opaque_artifacts: Mapping[str, str] = field(default_factory=dict) legacy_capabilities: bool = field(default=False, kw_only=True) + typed_artifacts: Mapping[str, object] = field(default_factory=dict, kw_only=True) def __post_init__(self) -> None: if not isinstance(self.key, str): @@ -244,6 +247,26 @@ def __post_init__(self) -> None: object.__setattr__(self, "hit", False) elif not isinstance(self.capabilities, Capabilities): raise TypeError("NodeReceipt.capabilities must be Capabilities") + typed = _freeze_json(self.typed_artifacts) + if not isinstance(typed, Mapping): + raise TypeError("NodeReceipt.typed_artifacts must be a mapping.") + if typed: + if set(typed) != {"inputs", "outputs"} or any( + not isinstance(typed[name], Mapping) for name in typed + ): + raise ValueError( + "Typed artifact provenance requires input/output mappings." + ) + for bindings in typed.values(): + for alias, descriptor in bindings.items(): + if not isinstance(alias, str) or not alias: + raise ValueError( + "Typed artifact aliases must be nonempty strings." + ) + value_from_descriptor(b"", descriptor) + if self.legacy_capabilities: + raise ValueError("Legacy capabilities cannot describe typed artifacts.") + object.__setattr__(self, "typed_artifacts", typed) frozen_receipt = _freeze_json(self.receipt) if not isinstance(frozen_receipt, Mapping): raise TypeError("NodeReceipt.receipt must be a mapping") @@ -347,6 +370,11 @@ def _payload(self) -> dict[str, object]: "frame_key": self.frame_key, "weight_key": self.weight_key, "opaque_artifacts": self.opaque_artifacts, + **( + {"typed_artifacts": self.typed_artifacts} + if self.typed_artifacts + else {} + ), "wall_time": self.wall_time, } @@ -419,6 +447,7 @@ def __post_init__(self) -> None: raise TypeError("RunManifest.nodes values must be NodeReceipt") nodes[node_id] = receipt object.__setattr__(self, "nodes", MappingProxyType(nodes)) + _validate_typed_ancestry(nodes) decisions = tuple( decision @@ -592,7 +621,9 @@ def to_json(self) -> str: """Serialize the complete portable provenance as canonical JSON.""" payload = { - "schema_version": _SCHEMA_VERSION, + "schema_version": _TYPED_SCHEMA_VERSION + if any(node.typed_artifacts for node in self.nodes.values()) + else _SCHEMA_VERSION, "key": self.key, "tier": self.tier, "known_failures": self.known_failures, @@ -633,6 +664,7 @@ def from_json(cls, value: str | bytes | bytearray) -> Self: if type(schema_version) is not int or schema_version not in { _LEGACY_SCHEMA_VERSION, _SCHEMA_VERSION, + _TYPED_SCHEMA_VERSION, }: raise ValueError(f"unsupported manifest schema version {schema_version!r}") @@ -657,6 +689,10 @@ def from_json(cls, value: str | bytes | bytearray) -> Self: finished_at=_string_field(raw, "finished_at"), host=_string_field(raw, "host"), ) + if schema_version == _TYPED_SCHEMA_VERSION and not any( + node.typed_artifacts for node in nodes.values() + ): + raise ValueError("Schema-v3 manifest must carry typed artifact provenance.") body = raw.get("content_addressed") if not isinstance(body, Mapping): raise ValueError("manifest content-addressed body must be an object") @@ -678,7 +714,7 @@ def from_json(cls, value: str | bytes | bytearray) -> Self: raise ValueError( "manifest content key mismatch: serialized provenance was altered" ) - if schema_version == _SCHEMA_VERSION and serialized_key != manifest.key: + if schema_version != _LEGACY_SCHEMA_VERSION and serialized_key != manifest.key: raise ValueError( "manifest content key mismatch: serialized key differs from " "reconstructed portable provenance" @@ -1090,6 +1126,9 @@ def _node_receipt_from_payload(value: object, *, schema_version: int) -> NodeRec frame_key = value.get("frame_key") weight_key = value.get("weight_key") opaque_artifacts = value.get("opaque_artifacts", {}) + typed_artifacts = value.get("typed_artifacts", {}) + if "typed_artifacts" in value and schema_version != _TYPED_SCHEMA_VERSION: + raise ValueError("Typed artifact provenance requires manifest schema 3.") capabilities_payload = value.get("capabilities") if schema_version == _LEGACY_SCHEMA_VERSION: # Every schema-v1 receipt is legacy: v1 never recorded a tolerance, so @@ -1163,4 +1202,71 @@ def _node_receipt_from_payload(value: object, *, schema_version: int) -> NodeRec weight_key=weight_key, opaque_artifacts=opaque_artifacts, legacy_capabilities=legacy_capabilities, + typed_artifacts=typed_artifacts, ) + + +def _validate_typed_ancestry(nodes: Mapping[str, NodeReceipt]) -> None: + edges: dict[str, set[str]] = {node_id: set() for node_id in nodes} + for node_id, node in nodes.items(): + if not node.typed_artifacts: + continue + for name, entry in node.typed_artifacts["outputs"].items(): + value = value_from_descriptor(b"", entry) + if ( + entry["producer"] != node_id + or entry["artifact"] != name + or value.producer_key != node.key + or node.opaque_artifacts.get(name) != value.key + ): + raise ValueError( + f"Node {node_id!r} typed artifact output provenance mismatch." + ) + if ( + value.numerics.numeric is not node.capabilities.numeric + or value.numerics.tolerance != node.capabilities.tolerance + ): + raise ValueError( + f"Node {node_id!r} typed artifact numeric contract mismatch." + ) + for entry in node.typed_artifacts["inputs"].values(): + value = value_from_descriptor(b"", entry) + producer_id = entry["producer"] + producer = nodes.get(producer_id) + if producer is None or producer.key != value.producer_key: + raise ValueError( + f"Node {node_id!r} typed artifact producer is missing or inconsistent." + ) + if ( + producer.typed_artifacts.get("outputs", {}).get(entry["artifact"]) + != entry + ): + raise ValueError( + f"Node {node_id!r} typed artifact does not match its producer output." + ) + require_compatible_scope(value.numerics, node.capabilities) + edges[node_id].add(producer_id) + + memo: dict[str, frozenset[str]] = {} + + def visit(node_id: str, trail: frozenset[str]) -> frozenset[str]: + if node_id in trail: + raise ValueError("Typed artifact ancestry contains a cycle.") + if node_id in memo: + return memo[node_id] + ancestors = set(edges[node_id]) + for parent in edges[node_id]: + ancestors.update(visit(parent, trail | {node_id})) + memo[node_id] = frozenset(ancestors) + return memo[node_id] + + for node_id, node in nodes.items(): + ancestors = visit(node_id, frozenset()) + if _capability_role(node) is KernelRole.RELEASE: + artifact_gates = { + parent + for parent in ancestors + if _capability_role(nodes[parent]) is KernelRole.GATE + } + if not artifact_gates.issubset(set(node.receipt.get("gate_ancestry", ()))): + raise ValueError("Release omitted a typed artifact gate ancestor.") diff --git a/packages/microcosm-graph/src/microcosm/graph/randomness.py b/packages/microcosm-graph/src/microcosm/graph/randomness.py new file mode 100644 index 00000000..5a5510bc --- /dev/null +++ b/packages/microcosm-graph/src/microcosm/graph/randomness.py @@ -0,0 +1,69 @@ +"""Stable random coordinates, independent of graph packing and cache identity. + +A stream is ("sha256-u53-v1", experiment_id, replicate, base_seed). Each +nonempty coordinate tuple identifies a draw, conventionally (person_id, +process, period, draw_index). The top 53 bits of the SHA-256 digest, interpreted +big-endian, divided by 2**53 define a uniform in [0, 1). Stream parameters must +be normative node params; kernels must hash this module as implementing source +and declare SeedSource.KEYED. Repeated coordinates intentionally repeat draws. +""" + +from __future__ import annotations + +import hashlib +import math +from collections.abc import Sequence + +import numpy as np + +from .canonical import canonical_json + +__all__ = ["keyed_uniform"] + + +def _coordinate(value: object) -> list[object]: + if isinstance(value, np.generic): + value = value.item() + if isinstance(value, bool): + return ["bool", value] + if isinstance(value, int): + return ["int", value] + if isinstance(value, str): + return ["str", value] + if isinstance(value, float) and math.isfinite(value): + return ["float", value] + raise TypeError("Random coordinates must be non-null finite scalar identities.") + + +def keyed_uniform(*, stream: tuple, keys: Sequence[tuple]) -> np.ndarray: + """Return bytes-backed, read-only float64 draws keyed by stable coordinates. + + Row order, chunk boundaries, and unrelated inserted identities cannot affect + a draw. Integers, strings, booleans and floats have distinct canonical tags. + The experiment name is nonempty; replicate and base seed are non-negative + Python integers (booleans refused). No numpy RNG state is read or mutated. + """ + if not isinstance(stream, tuple) or len(stream) != 4: + raise TypeError( + "stream must be (algorithm, experiment_id, replicate, base_seed)." + ) + algorithm, experiment, replicate, base_seed = stream + if algorithm != "sha256-u53-v1": + raise ValueError(f"Unsupported random stream algorithm {algorithm!r}.") + if not isinstance(experiment, str) or not experiment: + raise ValueError("Random stream experiment_id must be non-empty.") + if any(type(value) is not int or value < 0 for value in (replicate, base_seed)): + raise ValueError( + "Random stream replicate/base_seed must be non-negative integers." + ) + prefix = b"microcosm-graph/keyed-uniform/1\0" + canonical_json(stream) + b"\0" + values = [] + for key in keys: + if not isinstance(key, tuple) or not key: + raise TypeError("Each random key must be a nonempty coordinate tuple.") + encoded = canonical_json([_coordinate(value) for value in key]) + digest = hashlib.sha256(prefix + encoded).digest() + values.append((int.from_bytes(digest[:8], "big") >> 11) / 2**53) + return np.frombuffer( + np.asarray(values, dtype=np.float64).tobytes(), dtype=np.float64 + ) diff --git a/packages/microcosm-graph/src/microcosm/graph/serialize.py b/packages/microcosm-graph/src/microcosm/graph/serialize.py index 33b5437f..85c71b4d 100644 --- a/packages/microcosm-graph/src/microcosm/graph/serialize.py +++ b/packages/microcosm-graph/src/microcosm/graph/serialize.py @@ -7,6 +7,9 @@ from .canonical import canonical_json from .decl import ( + ArtifactInput, + ArtifactOutput, + ArtifactType, Graph, Node, Owned, @@ -90,6 +93,40 @@ def _partition_from_payload(value: object, label: str) -> tuple[str, str] | None def _node_payload(node: Node) -> dict[str, object]: return { + **( + { + "artifact_inputs": [ + { + "name": item.name, + "producer": item.producer, + "artifact": item.artifact, + "type": { + "name": item.type.name, + "schema_version": item.type.schema_version, + }, + } + for item in node.artifact_inputs + ] + } + if node.artifact_inputs + else {} + ), + **( + { + "artifact_outputs": [ + { + "name": item.name, + "type": { + "name": item.type.name, + "schema_version": item.type.schema_version, + }, + } + for item in node.artifact_outputs + ] + } + if node.artifact_outputs + else {} + ), "id": node.id, "kernel": node.kernel, "inputs": [ @@ -161,6 +198,9 @@ def _node_from_payload(value: object, index: int) -> Node: "description", "citation", } + fields.update( + name for name in ("artifact_inputs", "artifact_outputs") if name in payload + ) if "entrants" in payload: fields.add("entrants") _exact_fields(payload, fields, label) @@ -174,6 +214,18 @@ def _node_from_payload(value: object, index: int) -> Node: population = _optional_string(payload["population"], f"{label}.population") base = _optional_string(payload["base"], f"{label}.base") return Node( + artifact_inputs=tuple( + _artifact_from_payload(value, input_=True) + for value in _array( + payload.get("artifact_inputs", []), f"{label}.artifact_inputs" + ) + ), + artifact_outputs=tuple( + _artifact_from_payload(value, input_=False) + for value in _array( + payload.get("artifact_outputs", []), f"{label}.artifact_outputs" + ) + ), id=_string(payload["id"], f"{label}.id"), kernel=_string(payload["kernel"], f"{label}.kernel"), inputs=tuple( @@ -302,3 +354,25 @@ def _exact_fields( def _reject_json_constant(value: str) -> object: raise ValueError(f"graph JSON contains non-finite constant {value}") + + +def _artifact_from_payload( + value: object, *, input_: bool +) -> ArtifactInput | ArtifactOutput: + raw = _mapping(value, "artifact declaration") + fields = {"name", "type"} | ({"producer", "artifact"} if input_ else set()) + _exact_fields(raw, fields, "artifact declaration") + type_raw = _mapping(raw["type"], "artifact type") + _exact_fields(type_raw, {"name", "schema_version"}, "artifact type") + type_ = ArtifactType( + _string(type_raw["name"], "artifact type.name"), type_raw["schema_version"] + ) + name = _string(raw["name"], "artifact name") + if input_: + return ArtifactInput( + name, + _string(raw["producer"], "artifact producer"), + _string(raw["artifact"], "artifact output"), + type_, + ) + return ArtifactOutput(name, type_) diff --git a/packages/microcosm-graph/src/microcosm/graph/view.py b/packages/microcosm-graph/src/microcosm/graph/view.py index 63820048..446a7883 100644 --- a/packages/microcosm-graph/src/microcosm/graph/view.py +++ b/packages/microcosm-graph/src/microcosm/graph/view.py @@ -87,6 +87,22 @@ def describe( f"Implementation hash: {implementation}", ] ) + if node.artifact_inputs: + lines.append( + "Artifact inputs: " + + "; ".join( + f"{item.name} <- {item.producer}.{item.artifact} ({item.type.name}@{item.type.schema_version})" + for item in node.artifact_inputs + ) + ) + if node.artifact_outputs: + lines.append( + "Artifact outputs: " + + "; ".join( + f"{item.name} ({item.type.name}@{item.type.schema_version})" + for item in node.artifact_outputs + ) + ) if run_receipt is None: lines.append( 'Seed: int.from_bytes(sha256(b"seed\\0" + node_key)[:8], "little")' diff --git a/packages/microcosm-graph/tests/fixtures/legacy-graph-key-baseline.json b/packages/microcosm-graph/tests/fixtures/legacy-graph-key-baseline.json new file mode 100644 index 00000000..8f28af74 --- /dev/null +++ b/packages/microcosm-graph/tests/fixtures/legacy-graph-key-baseline.json @@ -0,0 +1,107 @@ +{ + "baseline_version": 1, + "base_commit": "96faa5d53748316c940d91b8210d25c6aacc245f", + "source_worktree": "/Users/maxghenis/PolicyEngine/_worktrees/microcosm-model-graph-20260904", + "source_fixture": "packages/microcosm-graph/tests/_toy.py::small_graph", + "acceptance_usage": "packages/microcosm-graph/tests/test_acceptance_a_identity.py::test_a1_determinism_across_processes_and_a_reloaded_store", + "source_fixture_sha256": "b8feedc576266929836e239cd3014707ab0980db45d4337136a0cc3b7b71beda", + "source_fixture_files_sha256": { + "household.csv": "a142233699a5477c29c371cef8db888cfc21b00e81ab8d9b6c0dd511874468c2", + "person.csv": "a1a9781284d8994fa00dd90440bfddf8f28ff39c5b5bfd2c534b87fb8b34b239", + "release.csv": "a3163198f1182cfbd46d9cb25341650292c2360c0e4bf67cf550af3c686e2053", + "schema.json": "4317d883e16984aa908bd0b4133a7c50fdc668815d3886d7240577d870ce8763", + "weights.csv": "af4f4b6ae364770ff4166eaa38005c141aa83cbf669e91caeb6758d1e2f30a09" + }, + "capture_command": "/Users/maxghenis/PolicyEngine/microcosm/.venv/bin/python /Users/maxghenis/architecture-reviews/microcosm-20260904/capture_legacy_graph_key_baseline.py", + "identity_isolation": "Recorded kernel hashes are fixed SHA256(\"legacy-graph-key-baseline-v1\\0\" + kernel_ref) sent directly to node_key; implementation_hash/source_hash is never called. Tests must use these recorded hash literals and source keys.", + "imported_key_module": "/Users/maxghenis/PolicyEngine/_worktrees/microcosm-model-graph-20260904/packages/microcosm-graph/src/microcosm/graph/keys.py", + "graph_json": "{\"country\":\"toy\",\"nodes\":[{\"base\":null,\"citation\":\"\",\"description\":\"load the toy country\",\"id\":\"survey\",\"inputs\":[],\"kernel\":\"source.csv@1\",\"mass\":\"conserve\",\"outputs\":[{\"column\":\"age\",\"dtype\":\"int64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"},{\"column\":\"income\",\"dtype\":\"float64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"},{\"column\":\"is_adult\",\"dtype\":\"boolean\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"},{\"column\":\"receives_x\",\"dtype\":\"boolean\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"},{\"column\":\"household_size\",\"dtype\":\"int64\",\"entity\":\"household\",\"ownership\":\"produced\",\"rows\":\"all\"}],\"params\":{},\"population\":null,\"sources\":[\"survey\"],\"structural\":\"create\",\"weights\":null},{\"base\":null,\"citation\":\"\",\"description\":\"\",\"id\":\"resources\",\"inputs\":[{\"columns\":[\"age\",\"income\"],\"entity\":\"person\",\"rows\":\"all\"}],\"kernel\":\"derive.add@1\",\"mass\":\"conserve\",\"outputs\":[{\"column\":\"resources\",\"dtype\":\"float64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"}],\"params\":{\"columns\":[\"age\",\"income\"],\"entity\":\"person\",\"scale\":1.5,\"target\":\"resources\"},\"population\":\"survey\",\"sources\":[],\"structural\":\"none\",\"weights\":null},{\"base\":null,\"citation\":\"\",\"description\":\"\",\"id\":\"draw_a\",\"inputs\":[{\"columns\":[\"age\"],\"entity\":\"person\",\"rows\":\"all\"}],\"kernel\":\"draw.uniform@1\",\"mass\":\"conserve\",\"outputs\":[{\"column\":\"noise_a\",\"dtype\":\"float64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"}],\"params\":{\"entity\":\"person\",\"target\":\"noise_a\"},\"population\":\"survey\",\"sources\":[],\"structural\":\"none\",\"weights\":null},{\"base\":null,\"citation\":\"\",\"description\":\"\",\"id\":\"target_a\",\"inputs\":[{\"columns\":[\"age\",\"income\"],\"entity\":\"person\",\"rows\":\"all\"}],\"kernel\":\"impute.chain@1\",\"mass\":\"conserve\",\"outputs\":[{\"column\":\"target_a\",\"dtype\":\"float64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"}],\"params\":{\"entity\":\"person\",\"noise\":0.5,\"predictors\":[\"age\",\"income\"],\"target\":\"target_a\"},\"population\":\"survey\",\"sources\":[],\"structural\":\"none\",\"weights\":null},{\"base\":null,\"citation\":\"\",\"description\":\"\",\"id\":\"target_b\",\"inputs\":[{\"columns\":[\"age\",\"target_a\"],\"entity\":\"person\",\"rows\":\"all\"}],\"kernel\":\"impute.chain@1\",\"mass\":\"conserve\",\"outputs\":[{\"column\":\"target_b\",\"dtype\":\"float64\",\"entity\":\"person\",\"ownership\":\"produced\",\"rows\":\"all\"}],\"params\":{\"entity\":\"person\",\"noise\":0.5,\"predictors\":[\"age\",\"target_a\"],\"target\":\"target_b\"},\"population\":\"survey\",\"sources\":[],\"structural\":\"none\",\"weights\":null}],\"sources\":[{\"codec\":\"csv-tables\",\"description\":\"the toy country's tables\",\"name\":\"survey\"}]}", + "kernel_implementation_hashes": { + "derive.add@1": "9c2df73423cfafb817f2986dd1af1866172a7817fb9731a5dd2143f777f3a8e3", + "draw.uniform@1": "6af02411c041ea1a021d6963c589e3b79af7556193b199402d8f0da27b88c396", + "impute.chain@1": "ce6ee14d3e3107486b0827f314a1c28b7c5aaaf43cf6b46b69e1be9d5149f58d", + "source.csv@1": "67df998f8afb82646051a44c96bcb9370b3119281d8e1c59e2722428c597218c" + }, + "kernel_capabilities": { + "derive.add@1": { + "determinism": "deterministic", + "numeric": "bitwise", + "seed_source": "none", + "structural": "none", + "role": "compute", + "consumes_se": false, + "dependencies": [], + "tolerance": null + }, + "draw.uniform@1": { + "determinism": "seeded", + "numeric": "bitwise", + "seed_source": "executor", + "structural": "none", + "role": "compute", + "consumes_se": false, + "dependencies": [], + "tolerance": null + }, + "impute.chain@1": { + "determinism": "seeded", + "numeric": "bitwise", + "seed_source": "executor", + "structural": "none", + "role": "compute", + "consumes_se": false, + "dependencies": [], + "tolerance": null + }, + "source.csv@1": { + "determinism": "deterministic", + "numeric": "bitwise", + "seed_source": "none", + "structural": "create", + "role": "compute", + "consumes_se": false, + "dependencies": [], + "tolerance": null + } + }, + "source_keys": { + "survey": "4fcb5d84ed4e4de1802dd503208e1199c4307948f0abae9d50936158b2ea377b" + }, + "compiled_order": [ + "survey", + "draw_a", + "resources", + "target_a", + "target_b" + ], + "predecessors": { + "survey": [], + "resources": [ + "survey" + ], + "draw_a": [ + "survey" + ], + "target_a": [ + "survey" + ], + "target_b": [ + "survey", + "target_a" + ] + }, + "node_keys": { + "survey": "f8e33d6cddc6c2c86ce9364d81fc6ebc5427fb1c869aad6b3a54525a876be1b9", + "draw_a": "14961b908e77336f5bd7a963cbabf71443e3634fcfbf7cc0bc7a65b4d6b3a3ae", + "resources": "756ef685a1329346b0bce608aa27945c5fa2a291d4a818238813db0738bf85a7", + "target_a": "259de014b0b972fe6de6d6ecb28aa66efa6a6de8c13319e0e6c8f8b1f986ad14", + "target_b": "5ff8a0b08a2f9b7f54ebc81c336465fb6109210d86764ed924ecdcc10c27453c" + }, + "node_seeds": { + "survey": 8555589241223682503, + "draw_a": 8325377662925372478, + "resources": 514626482650024594, + "target_a": 14508388850083053968, + "target_b": 17049535455440935095 + } +} 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 1bfbebe2..3b61d6fc 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":"02db8f5c849d876be20a95152b5302a5cacc0a7c77c58d8b436a3a00f57b4c92","kernel":"fit.qrf@1","node":"fit_qrf","node_key":"8878352db3439439f412f26c8762ff5871b94fe8e5fc8d3469dd1c45d7ef7da4","numeric":"platform_bitwise","platform":"arm64/darwin/py3.14","platforms":{"arm64/darwin/py3.14":{"direct":"direct.csv","node_key":"8878352db3439439f412f26c8762ff5871b94fe8e5fc8d3469dd1c45d7ef7da4"},"x86_64/linux/py3.13":{"direct":"platforms/x86_64-linux-py3_13/direct.csv","node_key":"f6984280e1ef0f156bd24345f7d3677d573c650ac706f9a3f949627c0d76d2d4"},"x86_64/linux/py3.14":{"direct":"platforms/x86_64-linux-py3_14/direct.csv","node_key":"9e80ee3ac5c30f99725c4dd932535983dabde7913c653e79819b1df3a289720c"}},"seed":947} +{"dependencies":{"numpy":"2.4.6","pandas":"3.0.3","quantile-forest":"1.4.2","scikit-learn":"1.8.0"},"implementation_hash":"d1f8b1929e6452aa507b0d9c64ba42a59851dc31c71021303c25f060e34c075b","kernel":"fit.qrf@1","node":"fit_qrf","node_key":"35af6a452bd32ca39d313d78255236df01877d684c2d847b1bd5a6016a68e237","numeric":"platform_bitwise","platform":"arm64/darwin/py3.14","platforms":{"arm64/darwin/py3.14":{"direct":"direct.csv","node_key":"35af6a452bd32ca39d313d78255236df01877d684c2d847b1bd5a6016a68e237"},"x86_64/linux/py3.13":{"direct":"platforms/x86_64-linux-py3_13/direct.csv","node_key":"6c43edcb3edf2bdef20d915b1be16503d37583c678aced78f1442aca1139e1ae"},"x86_64/linux/py3.14":{"direct":"platforms/x86_64-linux-py3_14/direct.csv","node_key":"ecb6b20c02b0bc442c4baf3fd7def6d6aec06e6b9567910294f945d66c43bbf8"}},"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 2ce41a7f..feefb7b1 100644 --- a/packages/microcosm-graph/tests/test_acceptance_b_ownership.py +++ b/packages/microcosm-graph/tests/test_acceptance_b_ownership.py @@ -118,6 +118,7 @@ def test_b2_executor_enforces_ownership(tmp_path: Path) -> None: "sources", "tolerances", # amendment 13: declared tolerances of the inputs' owners "numerics", # amendment 17: per-coordinate numeric class, bound, platform + "artifacts", # amendment 19: declared immutable typed bytes } graph = toy.small_graph( diff --git a/packages/microcosm-graph/tests/test_artifact_edges.py b/packages/microcosm-graph/tests/test_artifact_edges.py new file mode 100644 index 00000000..0626d6d5 --- /dev/null +++ b/packages/microcosm-graph/tests/test_artifact_edges.py @@ -0,0 +1,585 @@ +"""Typed model edges preserve ownership, reuse and numeric provenance.""" + +import importlib.util +import json +import sys +from dataclasses import replace +from pathlib import Path + +import pandas as pd +import pytest + +from microcosm.graph import ( + ArtifactInput, + ArtifactOutput, + ArtifactType, + ArtifactValue, + Capabilities, + ContentStore, + Determinism, + Graph, + GraphError, + KernelBase, + KernelResult, + Node, + NodeRejectedError, + Numeric, + Owned, + RunManifest, + Slice, + Tolerance, + compile_graph, + describe, + graph_from_json, + graph_to_json, + run_graph, +) +from microcosm.graph.decl import StructuralDelta +from microcosm.graph.kernel import KernelRole, SeedSource +from microcosm.graph.keys import node_key, seed + +spec = importlib.util.spec_from_file_location( + "_artifact_toy", Path(__file__).with_name("_toy.py") +) +toy = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = toy +spec.loader.exec_module(toy) +MODEL = ArtifactType("test.scalar-model", 1) + + +class Train(KernelBase): + ref = "artifact.train@1" + capabilities = Capabilities(Determinism.DETERMINISTIC) + calls = 0 + + def run(self, context): + self.calls += 1 + value = float(context.tables["person"]["income"].mean()) + return KernelResult(artifacts={"model": str(value).encode(), "debug": b"extra"}) + + +class Apply(KernelBase): + ref = "artifact.apply@1" + capabilities = Capabilities(Determinism.DETERMINISTIC) + calls = 0 + + def run(self, context): + self.calls += 1 + assert set(context.artifacts) == {"fitted"} + fitted = context.artifacts["fitted"] + assert isinstance(fitted, ArtifactValue) and fitted.type == MODEL + with pytest.raises(TypeError): + context.artifacts["unrelated"] = fitted + assert isinstance(fitted.payload, bytes) + table = context.tables["person"] + values = table.age.to_numpy() + float(fitted.payload) + return KernelResult( + columns={ + ("person", "predicted"): pd.Series( + values, + index=pd.Index(table.person_id, name="person_id"), + dtype="float64", + ) + } + ) + + +def graph(): + return Graph( + "toy", + (toy.SOURCE,), + ( + toy.CREATE, + replace(toy.CREATE, id="recipient"), + Node( + "train", + "artifact.train@1", + population="survey", + inputs=(Slice("person", ("income",)),), + artifact_outputs=(ArtifactOutput("model", MODEL),), + ), + Node( + "apply", + "artifact.apply@1", + population="recipient", + inputs=(Slice("person", ("age",)),), + outputs=(Owned("person", "predicted", "float64"),), + artifact_inputs=(ArtifactInput("fitted", "train", "model", MODEL),), + ), + ), + ) + + +def registry(*, producer=Numeric.BITWISE, consumer=Numeric.BITWISE): + reg = toy.toy_registry() + train = Train() + apply = Apply() + for obj, numeric in ((train, producer), (apply, consumer)): + obj.capabilities = Capabilities( + Determinism.DETERMINISTIC, + numeric=numeric, + tolerance=Tolerance(rtol=1e-5) + if numeric is Numeric.TOLERANCE_BOUND + else None, + ) + reg.register(obj) + return reg, train, apply + + +def test_cross_population_cold_warm_and_roundtrip(tmp_path): + original = graph() + compiled = compile_graph(original) + assert "train" in compiled.predecessors["apply"] + assert graph_from_json(graph_to_json(original)) == original + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + reg, train, apply = registry() + cold = run_graph(compiled, sources=sources, store=store, kernels=reg) + warm = run_graph(compiled, sources=sources, store=store, kernels=reg) + assert train.calls == apply.calls == 1 + assert cold.key == warm.key + assert all(node.hit for node in warm.nodes.values()) + assert "model" in describe(compiled, "apply", warm) + restored = RunManifest.from_json(warm.to_json()) + assert restored.key == warm.key + assert restored.to_json() == warm.to_json() + assert json.loads(warm.to_json())["schema_version"] == 3 + + +@pytest.mark.parametrize( + "change", + [ + {"producer": "missing"}, + {"artifact": "debug"}, + {"type": ArtifactType("other", 1)}, + {"type": ArtifactType("test.scalar-model", 2)}, + {"producer": "apply"}, + ], +) +def test_invalid_edges_rejected(change): + g = graph() + app = g.node("apply") + bad = replace(app, artifact_inputs=(replace(app.artifact_inputs[0], **change),)) + with pytest.raises(GraphError): + compile_graph(toy.replace_node(g, bad)) + + +def test_duplicate_alias_and_bad_type_rejected(): + app = graph().node("apply") + with pytest.raises(GraphError): + replace(app, artifact_inputs=app.artifact_inputs * 2) + with pytest.raises((GraphError, TypeError)): + ArtifactType("model", True) + with pytest.raises((GraphError, TypeError)): + ArtifactType("model", 0) + + +def test_artifact_cycle(): + g = graph() + train = g.node("train") + app = g.node("apply") + train = replace( + train, artifact_inputs=(ArtifactInput("back", "apply", "back", MODEL),) + ) + app = replace(app, artifact_outputs=(ArtifactOutput("back", MODEL),)) + with pytest.raises(GraphError, match="Cycle"): + compile_graph(toy.replace_node(g, train, app)) + + +@pytest.mark.parametrize( + "producer,consumer,allowed", + [ + (Numeric.BITWISE, Numeric.BITWISE, True), + (Numeric.BITWISE, Numeric.PLATFORM_BITWISE, True), + (Numeric.BITWISE, Numeric.TOLERANCE_BOUND, True), + (Numeric.PLATFORM_BITWISE, Numeric.BITWISE, False), + (Numeric.PLATFORM_BITWISE, Numeric.PLATFORM_BITWISE, True), + (Numeric.PLATFORM_BITWISE, Numeric.TOLERANCE_BOUND, False), + (Numeric.TOLERANCE_BOUND, Numeric.BITWISE, False), + (Numeric.TOLERANCE_BOUND, Numeric.PLATFORM_BITWISE, False), + (Numeric.TOLERANCE_BOUND, Numeric.TOLERANCE_BOUND, True), + ], +) +def test_numeric_contract_table(tmp_path, producer, consumer, allowed): + reg, train, apply = registry(producer=producer, consumer=consumer) + args = dict( + sources=toy.toy_sources(tmp_path), + store=ContentStore(tmp_path / "store"), + kernels=reg, + ) + if not allowed: + with pytest.raises( + NodeRejectedError, match="numeric|Numeric|platform|tolerance" + ): + run_graph(compile_graph(graph()), **args) + assert apply.calls == 0 + else: + cold = run_graph(compile_graph(graph()), **args) + warm = run_graph(compile_graph(graph()), **args) + assert cold.key == warm.key and apply.calls == 1 + + +def test_legacy_pinned_keys_and_json(): + baseline = json.loads( + (Path(__file__).parent / "fixtures/legacy-graph-key-baseline.json").read_text() + ) + g = graph_from_json(baseline["graph_json"]) + compiled = compile_graph(g) + keys = {} + assert graph_to_json(g) == baseline["graph_json"] + for node_id in compiled.order: + node = g.node(node_id) + raw = baseline["kernel_capabilities"][node.kernel] + cap = Capabilities( + Determinism(raw["determinism"]), + numeric=Numeric(raw["numeric"]), + seed_source=SeedSource(raw["seed_source"]), + structural=StructuralDelta(raw["structural"]), + role=KernelRole(raw["role"]), + consumes_se=raw["consumes_se"], + dependencies=tuple(raw["dependencies"]), + ) + keys[node_id] = node_key( + compiled, + node_id, + keys, + baseline["kernel_implementation_hashes"][node.kernel], + baseline["source_keys"], + kernel_capabilities=cap, + ) + assert keys == baseline["node_keys"] + assert {n: seed(k) for n, k in keys.items()} == baseline["node_seeds"] + + +@pytest.mark.parametrize("artifacts", [{}, {"model": bytearray(b"2")}, {"model": "2"}]) +def test_missing_or_mutable_declared_output(tmp_path, artifacts): + reg, train, apply = registry() + train.run = lambda context: KernelResult(artifacts=artifacts) + with pytest.raises(NodeRejectedError, match="artifact|bytes"): + run_graph( + compile_graph(graph()), + sources=toy.toy_sources(tmp_path), + store=ContentStore(tmp_path / "store"), + kernels=reg, + ) + assert apply.calls == 0 + + +def test_artifact_inputs_enter_keys_without_recipient_refit(tmp_path): + original = graph() + reg, _, _ = registry() + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + first = run_graph( + compile_graph(original), sources=sources, store=store, kernels=reg + ) + changed = toy.replace_node( + original, replace(original.node("apply"), params={"application": "changed"}) + ) + second = run_graph( + compile_graph(changed), sources=sources, store=store, kernels=reg + ) + assert second.nodes["train"].hit and not second.nodes["apply"].hit + assert first.nodes["train"].key == second.nodes["train"].key + changed = toy.replace_node( + original, replace(original.node("train"), params={"training": "changed"}) + ) + third = run_graph(compile_graph(changed), sources=sources, store=store, kernels=reg) + assert not third.nodes["train"].hit and not third.nodes["apply"].hit + assert third.nodes["recipient"].hit + + +def test_tampered_model_refused_even_when_consumer_cached(tmp_path): + from microcosm.graph import StoreCorruptError + + reg, train, apply = registry() + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + first = run_graph(compile_graph(graph()), sources=sources, store=store, kernels=reg) + key = first.nodes["train"].opaque_artifacts["model"] + (store.object_path(key) / "payload.bin").write_bytes(b"tampered") + with pytest.raises(StoreCorruptError): + run_graph(compile_graph(graph()), sources=sources, store=store, kernels=reg) + assert train.calls == apply.calls == 1 + + +def test_require_preflights_missing_model_and_missing_codec(tmp_path): + import shutil + + from microcosm.graph import StoreMissError, StoreUnavailableError + + reg, _, _ = registry() + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + first = run_graph(compile_graph(graph()), sources=sources, store=store, kernels=reg) + key = first.nodes["train"].opaque_artifacts["model"] + shutil.rmtree(store.object_path(key)) + fresh, train, apply = registry() + with pytest.raises(StoreMissError): + run_graph( + compile_graph(graph()), + sources=sources, + store=store, + kernels=fresh, + resume="require", + ) + assert train.calls == apply.calls == toy.total_calls(fresh) == 0 + with pytest.raises(StoreUnavailableError): + run_graph( + compile_graph(graph()), + sources=sources, + store=ContentStore(tmp_path / "store", codecs={}), + kernels=fresh, + resume="require", + ) + assert toy.total_calls(fresh) == 0 + + +def test_cached_contract_and_missing_declared_output_refused(tmp_path, monkeypatch): + from microcosm.graph import StoreCorruptError, StoreMissError + + reg, _, _ = registry() + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + first = run_graph(compile_graph(graph()), sources=sources, store=store, kernels=reg) + real = store.load_json + mode = ["type"] + + def altered(key): + record = real(key) + if record.get("node_id") == "train": + if mode[0] == "type": + record["typed_artifacts"]["outputs"]["model"]["type"][ + "schema_version" + ] = 2 + else: + record["opaque"] = [ + item for item in record["opaque"] if item["name"] != "model" + ] + return record + + monkeypatch.setattr(store, "load_json", altered) + with pytest.raises(StoreCorruptError, match="contracts"): + run_graph( + compile_graph(graph()), + sources=sources, + store=store, + kernels=reg, + resume="require", + ) + mode[0] = "missing" + with pytest.raises(StoreMissError): + run_graph( + compile_graph(graph()), + sources=sources, + store=store, + kernels=reg, + resume="require", + ) + assert first.nodes["train"].typed_artifacts + + +class ReportScope(KernelBase): + ref = "artifact.scope@1" + capabilities = Capabilities(Determinism.DETERMINISTIC, role=KernelRole.GATE) + + def run(self, context): + scope = context.numerics[("person", "predicted")] + return KernelResult( + receipt={ + "outcome": "pass", + "evidence": { + "numeric": scope.numeric.value, + "platform": scope.platform, + }, + } + ) + + +def test_model_scope_reaches_downstream_gate_on_cold_warm(tmp_path): + reg, _, _ = registry( + producer=Numeric.PLATFORM_BITWISE, consumer=Numeric.PLATFORM_BITWISE + ) + reg.register(ReportScope()) + g = graph() + gate = Node( + "scope", + "artifact.scope@1", + population="recipient", + inputs=(Slice("person", ("predicted",)),), + ) + g = replace(g, nodes=(*g.nodes, gate)) + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + first = run_graph(compile_graph(g), sources=sources, store=store, kernels=reg) + second = run_graph(compile_graph(g), sources=sources, store=store, kernels=reg) + for result in (first, second): + evidence = result.nodes["scope"].receipt["evidence"] + assert evidence["numeric"] == "platform_bitwise" and evidence["platform"] + assert second.nodes["scope"].hit + + +def test_mixed_numeric_inputs_and_combined_scope_refused(tmp_path): + from microcosm.graph import NumericScope + from microcosm.graph.artifact_edges import require_compatible_scope + + with pytest.raises(NodeRejectedError, match="platform and tolerance"): + require_compatible_scope( + NumericScope(Numeric.TOLERANCE_BOUND, Tolerance(rtol=1e-5), "arm64/test"), + Capabilities( + Determinism.DETERMINISTIC, + numeric=Numeric.TOLERANCE_BOUND, + tolerance=Tolerance(rtol=1e-5), + ), + ) + reg, train, _ = registry( + producer=Numeric.PLATFORM_BITWISE, consumer=Numeric.PLATFORM_BITWISE + ) + other = Train() + other.ref = "artifact.other@1" + other.capabilities = Capabilities( + Determinism.DETERMINISTIC, + numeric=Numeric.TOLERANCE_BOUND, + tolerance=Tolerance(rtol=1e-5), + ) + reg.register(other) + g = graph() + second = replace(g.node("train"), id="other", kernel=other.ref) + app = replace( + g.node("apply"), + artifact_inputs=( + *g.node("apply").artifact_inputs, + ArtifactInput("second", "other", "model", MODEL), + ), + ) + g = toy.replace_node(replace(g, nodes=(*g.nodes, second)), app) + with pytest.raises(NodeRejectedError, match="tolerance"): + run_graph( + compile_graph(g), + sources=toy.toy_sources(tmp_path), + store=ContentStore(tmp_path / "store"), + kernels=reg, + ) + assert train.calls == 0 + + +class EvidenceGate(KernelBase): + ref = "artifact.evidence@1" + capabilities = Capabilities(Determinism.DETERMINISTIC, role=KernelRole.GATE) + + def run(self, context): + return KernelResult( + artifacts={"evidence": b"fail"}, + receipt={"outcome": "fail", "evidence": {"reason": "synthetic failure"}}, + ) + + +class EvidenceRelease(KernelBase): + ref = "artifact.release@1" + capabilities = Capabilities(Determinism.DETERMINISTIC, role=KernelRole.RELEASE) + + def run(self, context): + table = context.tables["release"] + assert context.artifacts["gate"].payload == b"fail" + return KernelResult( + columns={ + ("release", "tier"): pd.Series( + ["evidence"], + index=pd.Index(table.release_id, name="release_id"), + dtype="string", + ) + }, + receipt={"outcome": "fail", "tier": "evidence"}, + ) + + +def test_artifact_only_gate_ancestry_and_manifest_tamper(tmp_path): + evidence_type = ArtifactType("test.evidence", 1) + g = Graph( + "toy", + (toy.SOURCE,), + ( + toy.CREATE, + Node( + "gate", + "artifact.evidence@1", + population="survey", + artifact_outputs=(ArtifactOutput("evidence", evidence_type),), + ), + Node( + "release", + "artifact.release@1", + population="survey", + outputs=(Owned("release", "tier", "string"),), + params={"requires_decisions": ()}, + artifact_inputs=( + ArtifactInput("gate", "gate", "evidence", evidence_type), + ), + ), + ), + ) + reg = toy.toy_registry() + reg.register(EvidenceGate()) + reg.register(EvidenceRelease()) + sources = toy.toy_sources(tmp_path) + store = ContentStore(tmp_path / "store") + manifest = run_graph(compile_graph(g), sources=sources, store=store, kernels=reg) + assert manifest.tier == "evidence" and manifest.nodes["release"].receipt[ + "gate_ancestry" + ] == ("gate",) + path = tmp_path / "manifest.json" + manifest.save(path) + assert RunManifest.load(path, store).key == manifest.key + raw = json.loads(manifest.to_json()) + raw["nodes"]["release"]["typed_artifacts"]["inputs"]["gate"]["type"][ + "schema_version" + ] = 3 + with pytest.raises(ValueError, match="producer|artifact|content"): + RunManifest.from_json(json.dumps(raw)) + raw = json.loads(manifest.to_json()) + raw["schema_version"] = 2 + with pytest.raises(ValueError, match="schema 3"): + RunManifest.from_json(json.dumps(raw)) + + +def test_shared_typed_ancestry_is_memoized(): + """A layered shared DAG must not enumerate exponentially many paths.""" + import hashlib + + from microcosm.graph import NodeReceipt + from microcosm.graph.artifact_edges import descriptor + + nodes = {} + prior = [] + caps = Capabilities(Determinism.DETERMINISTIC) + for layer in range(30): + current = [] + for column in range(2): + node_id = f"layer{layer}_{column}" + key = hashlib.sha256(node_id.encode()).hexdigest() + output = descriptor( + producer=node_id, + artifact="model", + type_=MODEL, + producer_key=key, + capabilities=caps, + ) + nodes[node_id] = NodeReceipt( + key=key, + hit=False, + seed=0, + kernel_ref="toy@1", + kernel_impl_hash="a" * 64, + capabilities=caps, + opaque_artifacts={"model": output["key"]}, + typed_artifacts={ + "inputs": { + parent: nodes[parent].typed_artifacts["outputs"]["model"] + for parent in prior + }, + "outputs": {"model": output}, + }, + ) + current.append(node_id) + prior = current + assert len(RunManifest(country="toy", nodes=nodes).nodes) == 60 diff --git a/packages/microcosm-graph/tests/test_graph_kernel_contract.py b/packages/microcosm-graph/tests/test_graph_kernel_contract.py index 057bec79..eaff7184 100644 --- a/packages/microcosm-graph/tests/test_graph_kernel_contract.py +++ b/packages/microcosm-graph/tests/test_graph_kernel_contract.py @@ -193,7 +193,7 @@ def test_numeric_scope_validates_class_tolerance_and_platform() -> None: def test_context_numerics_default_empty_and_carry_scopes() -> None: """Amendment 17: ``numerics`` defaults empty and rides at the end of the context.""" fields = [f.name for f in dataclasses.fields(KernelContext)] - assert fields[-2:] == ["tolerances", "numerics"] + assert fields[-3:] == ["tolerances", "numerics", "artifacts"] scope = NumericScope( numeric=Numeric.PLATFORM_BITWISE, platform="arm64/darwin/py3.13" ) diff --git a/packages/microcosm-graph/tests/test_keyed_randomness.py b/packages/microcosm-graph/tests/test_keyed_randomness.py new file mode 100644 index 00000000..b4fd7661 --- /dev/null +++ b/packages/microcosm-graph/tests/test_keyed_randomness.py @@ -0,0 +1,89 @@ +"""Random coordinates retain draws through ordering and chunking changes.""" + +import numpy as np +import pytest + +from microcosm.graph import SeedSource, keyed_uniform + +STREAM = ("sha256-u53-v1", "comparison", 0, 42) +KEYS = [ + (10, "mortality", 2027, 0), + (11, "mortality", 2027, 0), + (12, "mortality", 2027, 0), +] + + +def test_stability(): + whole = keyed_uniform(stream=STREAM, keys=KEYS) + assert whole.dtype == np.float64 and not whole.flags.writeable + assert ((whole >= 0) & (whole < 1)).all() + np.testing.assert_array_equal( + whole[::-1], keyed_uniform(stream=STREAM, keys=KEYS[::-1]) + ) + np.testing.assert_array_equal( + whole, + np.concatenate( + [ + keyed_uniform(stream=STREAM, keys=KEYS[:1]), + keyed_uniform(stream=STREAM, keys=KEYS[1:]), + ] + ), + ) + np.testing.assert_array_equal( + whole, keyed_uniform(stream=STREAM, keys=[("extra",), *KEYS])[1:] + ) + assert SeedSource.KEYED.value == "keyed" + assert keyed_uniform(stream=STREAM, keys=[]).shape == (0,) + with pytest.raises(ValueError): + whole.setflags(write=True) + + +def test_coordinate_boundaries_and_streams(): + keys = [(1,), ("1",), ("ab", "c"), ("a", "bc")] + values = keyed_uniform(stream=STREAM, keys=keys) + assert len(set(values)) == 4 + assert not np.array_equal( + values, keyed_uniform(stream=(STREAM[0], STREAM[1], 1, 42), keys=keys) + ) + np.testing.assert_array_equal( + keyed_uniform(stream=STREAM, keys=[(np.int64(1),)]), values[:1] + ) + + +@pytest.mark.parametrize( + "keys", + [ + [(None,)], + [(float("nan"),)], + [(float("inf"),)], + [(object(),)], + [()], + ["not-a-tuple"], + ], +) +def test_bad_coordinates(keys): + with pytest.raises((TypeError, ValueError)): + keyed_uniform(stream=STREAM, keys=keys) + + +@pytest.mark.parametrize( + "stream", + [ + ("unknown", "x", 0, 1), + ("sha256-u53-v1", "", 0, 1), + ("sha256-u53-v1", "x", True, 1), + ("sha256-u53-v1", "x", -1, 1), + ], +) +def test_bad_stream(stream): + with pytest.raises((TypeError, ValueError)): + keyed_uniform(stream=stream, keys=KEYS) + + +def test_version_one_fixed_vectors(): + # SHA-256 domain/coordinate encoding + high 53 bits, not an RNG-library pin. + assert [value.hex() for value in keyed_uniform(stream=STREAM, keys=KEYS)] == [ + "0x1.1128e9d291d62p-1", + "0x1.854814a32b494p-3", + "0x1.780f9dd36f1e0p-3", + ] diff --git a/tools/ci_test_groups.py b/tools/ci_test_groups.py index bc737de4..aad07d21 100644 --- a/tools/ci_test_groups.py +++ b/tools/ci_test_groups.py @@ -30,7 +30,10 @@ # Shared-engine behavioral contracts that deliberately do not carry a country # prefix or the spec-engine ``test_spec_*`` prefix. Listing them makes their # shared-lane placement reviewed rather than a silent classifier default. -EXPLICIT_SHARED_SPEC = ("packages/microcosm-build/tests/test_cross_grain.py",) +EXPLICIT_SHARED_SPEC = ( + "packages/microcosm-build/tests/test_cross_grain.py", + "packages/microcosm-build/tests/test_transfer_graph_example.py", +) PROCESSES = { "trade": ("main",),