diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 3d57fcb5..c980a1a8 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -5,8 +5,6 @@ on: branches: - master pull_request: - branches: - - master permissions: contents: read diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 6b9cf94b..9d308c2d 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -5,8 +5,6 @@ on: branches: - master pull_request: - branches: - - master jobs: pytest-shard: @@ -54,7 +52,7 @@ jobs: - name: Require typed graph capabilities run: python -c "from populace_dynamics.graph._compat import require_graph; require_graph()" - name: Run graph and existing mortality regressions - run: pytest -q tests/test_graph_mortality.py tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py + run: pytest -q tests/test_graph_mortality.py tests/test_graph_mortality_trajectory.py tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py # Fan-in jobs keeping the branch-protection context names # ("pytest (3.11)" / "pytest (3.13)") stable across the shard split. diff --git a/docs/population-graph.md b/docs/population-graph.md index 92bc30dd..4d873d8a 100644 --- a/docs/population-graph.md +++ b/docs/population-graph.md @@ -3,7 +3,8 @@ The optional `populace_dynamics.graph` package fits the existing M6 mortality model on a historical synthetic panel, applies the fitted artifact to a separate starting population, and adds surviving observations for the next -year. It produces an engineering report and a content-addressed execution +year, or repeats those transitions through an explicit end year. It produces +an engineering report and a content-addressed execution manifest. It does not change the existing projection loop, candidate registries, scientific gates, or committed evidence. @@ -116,13 +117,95 @@ leaving fitting, application, draws, and accounting unchanged. Household accounting is explicitly unsupported and refused by the Python entry point. Household weight sharing, marriage, births, immigration, -alignment replay, repeated years, and the full M6 loop remain later work. +alignment replay, and the full M6 loop remain later work. No certified data release or scientific candidate is produced by this graph. +## Annual trajectories + +The optional `run_mortality_trajectory` Python entry point builds one graph +with a single mortality fit and separate application, expansion, age-ownership, +snapshot, and evaluation nodes for each year. It uses the same exact +graph/Frame pin as the one-year example. The fit cutoff stays fixed while the +application year +advances. This extends execution of the existing age/sex law; it does not add +a calendar-year mortality improvement model or establish long-horizon validity. + +Each application reads only the preceding period's observations. A typed +transition artifact binds each probability and survival decision to its +person and observation identities. EXPAND appends survivor observations with +lineage to that preceding period. Earlier ages, memberships, and trajectory +weights stay unchanged. The mass receipt covers every historical period, +not just the newest pair. After extinction, later years contain no at-risk +people and add no orphan period groups. + +Declare one aggregate synthetic holdout for each application year. For example: + +```python +import json +from pathlib import Path + +from populace_dynamics.graph import run_mortality_trajectory +from populace_dynamics.graph.synthetic import write_synthetic_inputs + +root = Path("mortality-trajectory") +sources = write_synthetic_inputs(root / "inputs") +sources.pop("holdout") +holdouts = {} +for year in range(2015, 2018): + path = root / "inputs" / f"aggregate-{year}.json" + path.write_text(json.dumps({ + "scope": "synthetic_engineering", + "year": year, + "expected_death_rate": 0.2, + "fixture_max_abs_death_rate_gap": 0.25, + })) + holdouts[year] = path + +result = run_mortality_trajectory( + **sources, holdouts=holdouts, end_year=2017, output_dir=root, +) +print(result.report) +``` + +These small aggregate fixtures are deliberately artificial, with input +tolerances used only for engineering tests. They contain no empirical +acceptance targets. Each evaluation reads a typed snapshot of the actual +materialized population on a separate population version. This keeps the +evaluation outside the next expansion's dependencies under the pinned core. +An annual evaluation depends on its own holdout; changing +or failing that evaluation does not alter later simulation. Extending the +horizon reuses the existing fit and annual nodes in the same verified store. +Changing experiment, replicate, or seed changes application identities while +reusing the fit. All sources remain declared and content-hashed by the executor, +including evaluation sources whose kernels are subsequently guarded. + +The output directory contains `trajectory.csv`, `model.json`, `report.json`, +and `manifest.json`. The trajectory includes the initial period and every +completed survivor period, with person identity, age, year, and weight. +Annual reports keep expected and generated deaths, survivor counts, and +period mass separate from fixture and engineering verdicts. + +Application ages outside the fitted bands must fail explicitly. In particular, +a survivor aged 120 can be advanced to 121, but cannot enter another mortality +draw under a law with support ending at 120. The graph does not silently assign +such people a zero death probability. A typed failure outcome guards later +applications and expansions, preserving the latest valid population and the +original diagnostic. Blocked application status is reported separately from +the core's execution/cache receipts: this pinned executor still runs guarded +nodes and does not provide native `unreached` receipts. A failed evaluation +does not propagate this application block. + +Snapshots include the full materialized history, so their storage grows with +both population size and horizon. This synthetic integration has not been +benchmarked for national-scale projection. Root creation, fitting, store +corruption, and unexpected structural failures can still abort execution; +the retained diagnostic path covers application and evaluation failures. + ## Tests ```sh python -m pytest -q tests/test_graph_mortality.py \ + tests/test_graph_mortality_trajectory.py \ tests/test_m6_engine_refit.py tests/test_m6_engine_steps.py ``` @@ -137,3 +220,8 @@ the optional runtime cases when the required core capabilities are unavailable; the JSON and dependency boundary tests still run. Importing `populace_dynamics.graph` remains safe under Python 3.10–3.12. + +The annual tests independently repeat the existing fit, mortality, ageing, +and keyed-draw operations; compare every retained person-period and weighted +diagnostic; and exercise horizon reuse, stream changes, holdout isolation, +extinction, and retained support-failure evidence. diff --git a/scripts/first_estimates_birth_evidence.py b/scripts/first_estimates_birth_evidence.py index 911ae21f..81d5bd41 100644 --- a/scripts/first_estimates_birth_evidence.py +++ b/scripts/first_estimates_birth_evidence.py @@ -165,6 +165,7 @@ Path("src/populace_dynamics/graph/model.py"), Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), + Path("src/populace_dynamics/graph/trajectory.py"), ) POST_REVIEW_SHARED_SOURCE_BLOBS = { Path( diff --git a/src/populace_dynamics/graph/__init__.py b/src/populace_dynamics/graph/__init__.py index 03c8c7ab..05eca38c 100644 --- a/src/populace_dynamics/graph/__init__.py +++ b/src/populace_dynamics/graph/__init__.py @@ -15,4 +15,14 @@ def run_mortality_graph(**kwargs): return run(**kwargs) -__all__ = ["run_mortality_graph"] +def run_mortality_trajectory(**kwargs): + """Run the existing mortality/ageing steps across annual graph periods.""" + from ._compat import require_graph + + require_graph() + from .trajectory import run_mortality_trajectory as run + + return run(**kwargs) + + +__all__ = ["run_mortality_graph", "run_mortality_trajectory"] diff --git a/src/populace_dynamics/graph/trajectory.py b/src/populace_dynamics/graph/trajectory.py new file mode 100644 index 00000000..b457b08b --- /dev/null +++ b/src/populace_dynamics/graph/trajectory.py @@ -0,0 +1,898 @@ +"""Synthetic annual mortality transport with immutable period history. + +The fitted law and ageing step are the existing Dynamics implementations. +This optional graph supplies typed transition edges, stable person draws, +and explicit annual engineering diagnostics. It certifies no population. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from types import SimpleNamespace + +import numpy as np +import pandas as pd +from microcosm.frame import Weights +from microcosm.graph.decl import ( + ArtifactInput, + ArtifactOutput, + ArtifactType, + Graph, + Node, + Owned, + Slice, + SourceRef, + StructuralDelta, + compile_graph, +) +from microcosm.graph.executor import run_graph +from microcosm.graph.kernel import KernelResult, source_hash +from microcosm.graph.store import ContentStore + +from . import runtime as rt +from .model import MortalityArtifact, json_bytes, parse_json, read_json + +TRANSITION_TYPE = ArtifactType("populace-dynamics.mortality-transition", 1) +SNAPSHOT_TYPE = ArtifactType("populace-dynamics.mortality-population", 1) +_TRANSITION_FIELDS = { + "person_id", + "observation_id", + "death_probability", + "survives", +} + + +def _risk_rows(context): + observations = context.tables[rt.OBS] + periods = rt._periods(context, observations) + return observations.loc[periods == context.params["year"] - 1].sort_values( + rt.PID + ) + + +def _model(context): + artifact = MortalityArtifact.from_bytes(context.artifacts["model"].payload) + if artifact.boundary_year != context.params["boundary_year"]: + raise ValueError("mortality fit and trajectory boundary years differ") + return artifact.model + + +def _check_age_support(frame, model): + age = frame.age.to_numpy(dtype=np.int64) + if ((age < model.bands[0][0]) | (age > model.bands[-1][1])).any(): + raise ValueError( + "mortality application age exceeds the fitted age support " + f"[{model.bands[0][0]}, {model.bands[-1][1]}]" + ) + + +def _outcome( + year, status, *, records=(), completed_year=None, diagnostic=None +): + return { + "format": TRANSITION_TYPE.name, + "schema_version": 1, + "from_year": year - 1, + "year": year, + "status": status, + "completed_year": year if completed_year is None else completed_year, + "records": list(records), + "diagnostic": diagnostic, + } + + +def _apply_complete(context): + model = _model(context) + risk = _risk_rows(context) + initial = rt._slice(context, risk) + _check_age_support(initial, model) + year = context.params["year"] + survived = rt.apply_mortality( + initial, + rt._GraphPeriodContext( + {"boundary_year": year - 1, "stream": context.params["stream"]} + ), + context.rng, + model=model, + ) + probability = model.probabilities(initial) + survivors = set(survived.person_id) + records = [ + { + "person_id": int(pid), + "observation_id": int(oid), + "death_probability": float(p), + "survives": int(pid) in survivors, + } + for pid, oid, p in zip( + risk[rt.PID], risk[rt.OID], probability, strict=True + ) + ] + return _outcome(year, "complete", records=records) + + +def _apply(context): + year = context.params["year"] + completed_year = context.params["boundary_year"] + try: + previous = context.artifacts.get("previous_transition") + if previous is not None: + previous = _decode_transition(previous.payload, year - 1) + if previous["completed_year"] < completed_year: + raise ValueError( + "prior transition predates the model boundary" + ) + completed_year = previous["completed_year"] + if previous is not None and previous["status"] != "complete": + outcome = _outcome( + year, + "blocked", + completed_year=previous["completed_year"], + diagnostic={ + "blocked_by": f"apply_{year - 1}", + "message": "previous mortality transition did not complete", + }, + ) + else: + outcome = _apply_complete(context) + except Exception as error: + # The exact pinned core requires every declared artifact even on a + # failed gate. Publish an explicit typed failure, never survivor data. + outcome = _outcome( + year, + "failed", + completed_year=completed_year, + diagnostic={ + "exception_type": type(error).__name__, + "message": str(error), + }, + ) + complete = outcome["status"] == "complete" + return KernelResult( + artifacts={"transition": json_bytes(outcome)}, + receipt={ + "outcome": ( + ("pass" if outcome["records"] else "not_applicable") + if complete + else "fail" + ), + "application_status": outcome["status"], + "completed_year": outcome["completed_year"], + "evidence": outcome["diagnostic"] + or {"year": year, "risk_records": len(outcome["records"])}, + }, + ) + + +def _decode_transition(payload, year): + raw = parse_json(payload) + if ( + not isinstance(raw, dict) + or set(raw) + != { + "format", + "schema_version", + "from_year", + "year", + "records", + "status", + "completed_year", + "diagnostic", + } + or raw["format"] != TRANSITION_TYPE.name + or type(raw["schema_version"]) is not int + or raw["schema_version"] != 1 + or type(raw["year"]) is not int + or type(raw["from_year"]) is not int + or raw["year"] != year + or raw["from_year"] != year - 1 + or not isinstance(raw["records"], list) + or raw["status"] not in ("complete", "failed", "blocked") + or type(raw["completed_year"]) is not int + ): + raise ValueError("invalid annual mortality transition contract") + if raw["status"] == "complete": + if raw["completed_year"] != year or raw["diagnostic"] is not None: + raise ValueError("invalid completed mortality transition") + elif ( + raw["records"] + or raw["completed_year"] >= year + or not isinstance(raw["diagnostic"], dict) + or not isinstance(raw["diagnostic"].get("message"), str) + ): + raise ValueError("invalid failed or blocked mortality transition") + return raw + + +def _transition(context): + """Validate both the byte contract and the recipient observation binding.""" + raw = _decode_transition( + context.artifacts["transition"].payload, context.params["year"] + ) + risk = _risk_rows(context) + if raw["status"] != "complete": + return risk, raw + expected = list(zip(risk[rt.PID], risk[rt.OID], strict=True)) + records = raw["records"] + if len(records) != len(expected): + raise ValueError("mortality transition differs from the risk set") + for row, (person_id, observation_id) in zip( + records, expected, strict=True + ): + if ( + not isinstance(row, dict) + or set(row) != _TRANSITION_FIELDS + or type(row["person_id"]) is not int + or type(row["observation_id"]) is not int + or row["person_id"] != person_id + or row["observation_id"] != observation_id + or type(row["survives"]) is not bool + ): + raise ValueError( + "invalid mortality transition observation binding" + ) + p = row["death_probability"] + if ( + isinstance(p, bool) + or not isinstance(p, (int, float)) + or not np.isfinite(p) + or not 0 <= p <= 1 + ): + raise ValueError("invalid mortality transition probability") + return risk, raw + + +def _mass(weights, strata, periods): + frame = pd.DataFrame( + {"weight": weights, "stratum": strata, "period": periods} + ) + totals = { + str(key): float(value) + for key, value in frame.groupby("stratum", observed=True) + .weight.sum() + .items() + } + partition = { + str(year): { + str(key): float(value) + for key, value in part.groupby("stratum", observed=True) + .weight.sum() + .items() + } + for year, part in frame.groupby("period", observed=True) + } + return totals, partition + + +def _advance(context): + observations = context.tables[rt.OBS] + risk, outcome = _transition(context) + complete = outcome["status"] == "complete" + records = outcome["records"] + mask = np.asarray([row["survives"] for row in records], dtype=bool) + surviving = risk.loc[mask] if complete else risk.iloc[:0] + year = context.params["year"] + aged = ( + rt.advance_age( + rt._slice(context, surviving), + SimpleNamespace(year=year, metadata={}), + context.rng, + ) + if complete + else rt._slice(context, surviving) + ) + old_ids = observations[rt.OID].tolist() + new_ids = list(range(max(old_ids) + 1, max(old_ids) + 1 + len(surviving))) + target_ids = old_ids + new_ids + period_ids = context.tables["period"].period_id.tolist() + period_values = context.tables["period"].period.tolist() + next_period = [year] if len(new_ids) else [] + weights = context.weights[rt.OBS].values + position = pd.Series(np.arange(len(observations)), index=old_ids) + source_positions = position.loc[surviving[rt.OID]].to_numpy(dtype=int) + survivor_weights = weights[source_positions] + expanded_weights = np.concatenate([weights, survivor_weights]) + strata = context.strata.to_numpy() + periods = rt._periods(context, observations).to_numpy() + before, partitions_before = _mass(weights, strata, periods) + after, partitions_after = _mass( + expanded_weights, + np.concatenate([strata, strata[source_positions]]), + np.concatenate([periods, np.full(len(new_ids), year, dtype=int)]), + ) + return KernelResult( + expand={ + rt.OBS: rt._series(surviving[rt.OID].tolist(), new_ids), + "person": rt._series([], [], "person"), + "period": rt._series( + [pd.NA] * len(next_period), next_period, "period", "Int64" + ), + }, + columns={ + (rt.OBS, rt.PERIOD_ID): rt._series( + observations[rt.PERIOD_ID].tolist() + [year] * len(new_ids), + target_ids, + ), + (rt.OBS, "age"): rt._series( + observations.age.tolist() + aged.age.tolist(), target_ids + ), + ("period", "period"): rt._series( + period_values + next_period, period_ids + next_period, "period" + ), + }, + weights=Weights(expanded_weights, context.weights[rt.OBS].kind), + receipt={ + "application_status": "complete" if complete else "blocked", + "completed_year": outcome["completed_year"], + "mass": { + "policy": "declared", + "before": float(weights.sum()), + "after": float(expanded_weights.sum()), + "stratum_before": before, + "stratum_after": after, + "partition": { + "entity": "period", + "column": "period", + "stratum_before": partitions_before, + "stratum_after": partitions_after, + }, + }, + }, + ) + + +def _holdout(context): + year = context.params["year"] + raw = read_json(context.sources[f"holdout_{year}"]) + if ( + not isinstance(raw, dict) + or set(raw) + != { + "scope", + "year", + "expected_death_rate", + "fixture_max_abs_death_rate_gap", + } + or raw["scope"] != "synthetic_engineering" + or type(raw["year"]) is not int + or raw["year"] != year + ): + raise ValueError( + "holdout requires the matching year and synthetic engineering " + "aggregate contract" + ) + for key in ("expected_death_rate", "fixture_max_abs_death_rate_gap"): + value = raw[key] + if ( + isinstance(value, bool) + or not isinstance(value, (int, float)) + or not np.isfinite(value) + or not 0 <= value <= 1 + ): + raise ValueError(f"holdout {key} must be finite and in [0, 1]") + return raw + + +def _snapshot(context): + """Transport the actual executor population to an isolated evaluation.""" + return KernelResult( + artifacts={ + "snapshot": json_bytes( + { + "format": SNAPSHOT_TYPE.name, + "schema_version": 1, + "year": context.params["year"], + "observations": context.tables[rt.OBS].to_dict( + orient="records" + ), + "periods": context.tables["period"].to_dict( + orient="records" + ), + "weights": context.weights[rt.OBS].values.tolist(), + } + ) + } + ) + + +def _evaluation_context(context): + raw = parse_json(context.artifacts["snapshot"].payload) + if ( + not isinstance(raw, dict) + or set(raw) + != { + "format", + "schema_version", + "year", + "observations", + "periods", + "weights", + } + or raw["format"] != SNAPSHOT_TYPE.name + or type(raw["schema_version"]) is not int + or raw["schema_version"] != 1 + or type(raw["year"]) is not int + or raw["year"] != context.params["year"] + or not isinstance(raw["observations"], list) + or not isinstance(raw["periods"], list) + or not isinstance(raw["weights"], list) + ): + raise ValueError("invalid annual mortality population snapshot") + observations = pd.DataFrame(raw["observations"]) + periods = pd.DataFrame(raw["periods"]) + if ( + set(observations.columns) + != {rt.OID, rt.PID, rt.PERIOD_ID, "age", "sex"} + or set(periods.columns) != {"period_id", "period"} + or len(observations) != len(raw["weights"]) + or not observations[rt.OID].is_unique + or not periods.period_id.is_unique + or not observations[rt.PERIOD_ID].isin(periods.period_id).all() + ): + raise ValueError("invalid annual mortality snapshot row binding") + for column in (rt.OID, rt.PID, rt.PERIOD_ID, "age"): + rt._integer_column(observations, column) + for column in ("period_id", "period"): + rt._integer_column(periods, column) + if not observations.sex.isin(["female", "male"]).all(): + raise ValueError("invalid annual mortality snapshot sex") + return SimpleNamespace( + tables={rt.OBS: observations, "period": periods}, + weights={rt.OBS: rt._weights(raw["weights"])}, + params=context.params, + artifacts=context.artifacts, + sources=context.sources, + ) + + +def _evaluate(context): + context = _evaluation_context(context) + risk, outcome = _transition(context) + if outcome["status"] != "complete": + report = { + "scope": "synthetic_engineering", + "from_year": context.params["year"] - 1, + "year": context.params["year"], + "application_status": outcome["status"], + "completed_year": outcome["completed_year"], + "engineering_verdict": "not_evaluated", + "fixture_verdict": "not_evaluated", + "application_gate": { + "node_id": f"apply_{context.params['year']}", + "kernel_ref": "dynamics.trajectory.mortality.apply@1", + "outcome": "fail", + "evidence": outcome["diagnostic"], + }, + } + return KernelResult( + artifacts={"report": json_bytes(report)}, + receipt={"outcome": "evidence_absent", "evidence": report}, + ) + holdout = _holdout(context) + model = _model(context) + records = outcome["records"] + observations = context.tables[rt.OBS] + year = context.params["year"] + future = observations.loc[ + rt._periods(context, observations) == year + ].sort_values(rt.PID) + initial = rt._slice(context, risk) + _check_age_support(initial, model) + probability = model.probabilities(initial) + stream = tuple(context.params["stream"]) + uniforms = rt.mortality_uniforms( + initial.person_id.tolist(), + experiment_id=stream[1], + replicate=stream[2], + base_seed=stream[3], + period=year, + ) + survives = uniforms >= probability + expected_ids = initial.loc[survives, "person_id"].tolist() + weights = pd.Series( + context.weights[rt.OBS].values, + index=observations[rt.OID], + ) + start_weights = weights.loc[risk[rt.OID]].to_numpy() + next_weights = weights.loc[future[rt.OID]].to_numpy() + start_mass = float(start_weights.sum()) + expected_deaths = float(np.dot(start_weights, probability)) + generated_deaths = float( + start_weights[~risk[rt.PID].isin(future[rt.PID]).to_numpy()].sum() + ) + expected_ages = initial.loc[survives, "age"].to_numpy() + 1 + transition_parity = [ + row["survives"] for row in records + ] == survives.tolist() and np.array_equal( + [row["death_probability"] for row in records], probability + ) + engineering_pass = ( + transition_parity + and expected_ids == future[rt.PID].tolist() + and np.array_equal(expected_ages, future.age.to_numpy()) + and np.array_equal(start_weights[survives], next_weights) + and observations[rt.OID].is_unique + and not observations.duplicated([rt.PID, rt.PERIOD_ID]).any() + ) + rate_gap = ( + abs(expected_deaths / start_mass - holdout["expected_death_rate"]) + if start_mass + else None + ) + fixture_pass = ( + rate_gap <= holdout["fixture_max_abs_death_rate_gap"] + if rate_gap is not None + else True + ) + report = { + "scope": "synthetic_engineering", + "application_status": "complete", + "completed_year": year, + "from_year": year - 1, + "year": year, + "initial_records": len(risk), + "survivor_records": len(future), + "expected_deaths": expected_deaths, + "generated_deaths": generated_deaths, + "start_mass": start_mass, + "next_period_mass": float(next_weights.sum()), + "absolute_death_rate_gap": rate_gap, + "fixture_expected_death_rate": holdout["expected_death_rate"], + "fixture_max_abs_death_rate_gap": holdout[ + "fixture_max_abs_death_rate_gap" + ], + "engineering_verdict": ( + ("pass" if len(risk) else "not_applicable") + if engineering_pass + else "fail" + ), + "fixture_verdict": ( + ("pass" if fixture_pass else "fail") + if len(risk) + else "not_applicable" + ), + } + return KernelResult( + artifacts={"report": json_bytes(report)}, + receipt={ + "outcome": ( + ("pass" if len(risk) else "not_applicable") + if engineering_pass and fixture_pass + else "fail" + ), + "evidence": report, + }, + ) + + +class _TrajectoryKernel(rt._Kernel): + def implementation_hash(self): + # source_hash includes complete defining modules, including helpers. + # Keep these new wrappers separate from the unchanged fit/root hashes. + return source_hash( + self.function, + rt, + rt.model_module, + rt.fit_mortality_model, + rt.prepare_mortality_refit_inputs, + rt.apply_mortality, + rt.advance_age, + rt.keyed_uniform, + rt.canonical_json, + dependencies=self.capabilities.dependencies, + ) + + +def build_trajectory_graph( + *, + end_year, + boundary_year=2014, + external_vintage_year=2014, + experiment_id="mortality", + replicate=0, + base_seed=0, +): + """Declare one fit and independent annual transition/evaluation nodes.""" + if type(end_year) is not int or end_year <= boundary_year: + raise ValueError("end_year must be an integer after boundary_year") + original, registry = rt.build_graph( + boundary_year=boundary_year, + external_vintage_year=external_vintage_year, + experiment_id=experiment_id, + replicate=replicate, + base_seed=base_seed, + ) + nodes = list(original.nodes[:3]) + sources = list(original.sources[:3]) + for kernel in ( + _TrajectoryKernel( + "dynamics.trajectory.mortality.apply@1", + _apply, + seeded=True, + gate=True, + ), + _TrajectoryKernel( + "dynamics.trajectory.advance@1", + _advance, + structural=StructuralDelta.EXPAND, + ), + _TrajectoryKernel( + "dynamics.trajectory.snapshot@1", + _snapshot, + ), + _TrajectoryKernel( + "dynamics.trajectory.mortality.evaluate@1", + _evaluate, + seeded=True, + gate=True, + ), + ): + registry.register(kernel) + model_binding = ArtifactInput("model", "fit", "model", rt.MODEL_TYPE) + stream = ("sha256-u53-v1", experiment_id, replicate, base_seed) + base = "initial" + previous_transition = () + for year in range(boundary_year + 1, end_year + 1): + source = f"holdout_{year}" + sources.append(SourceRef(source, rt.CODEC)) + params = { + "year": year, + "boundary_year": boundary_year, + "stream": stream, + } + apply_id, advance_id = f"apply_{year}", f"advance_{year}" + transition_binding = ArtifactInput( + "transition", apply_id, "transition", TRANSITION_TYPE + ) + slices = (Slice(rt.OBS, ("age", "sex")), Slice("period", ("period",))) + nodes.extend( + ( + Node( + apply_id, + "dynamics.trajectory.mortality.apply@1", + population=base, + inputs=slices, + params=params, + artifact_inputs=(model_binding, *previous_transition), + artifact_outputs=( + ArtifactOutput("transition", TRANSITION_TYPE), + ), + ), + Node( + advance_id, + "dynamics.trajectory.advance@1", + base=base, + structural=StructuralDelta.EXPAND, + entrants=True, + mass="declared", + inputs=slices, + artifact_inputs=(transition_binding,), + params={ + "year": year, + "expand_cells": ( + (rt.OBS, rt.PERIOD_ID, "int64"), + (rt.OBS, "age", "int64"), + ("period", "period", "int64"), + ), + "expand_weight_entity": rt.OBS, + "expand_weight_kind": "design", + }, + ), + Node( + f"age_{year}", + "dynamics.age-claim@1", + population=advance_id, + inputs=(Slice(rt.OBS, ("age",)),), + outputs=(Owned(rt.OBS, "age", "int64", rewrite=True),), + ), + Node( + f"snapshot_{year}", + "dynamics.trajectory.snapshot@1", + population=advance_id, + inputs=slices, + params={"year": year}, + artifact_outputs=( + ArtifactOutput("snapshot", SNAPSHOT_TYPE), + ), + ), + Node( + f"evaluate_{year}", + "dynamics.trajectory.mortality.evaluate@1", + # The exact pinned compiler makes EXPAND depend on all + # ordinary members of its base. Evaluate an actual + # population snapshot on a separate existing version so + # its holdout never enters the next transition's key. + population="training", + sources=(source,), + artifact_inputs=( + model_binding, + transition_binding, + ArtifactInput( + "snapshot", + f"snapshot_{year}", + "snapshot", + SNAPSHOT_TYPE, + ), + ), + params=params, + ), + ) + ) + base = advance_id + previous_transition = ( + ArtifactInput( + "previous_transition", apply_id, "transition", TRANSITION_TYPE + ), + ) + return ( + Graph( + "dynamics-mortality-trajectory", + tuple(sources), + tuple(nodes), + mass_partition=original.mass_partition, + ), + registry, + ) + + +@dataclass(frozen=True) +class MortalityTrajectoryRun: + manifest: object + report: dict + model_payload: bytes + trajectory: pd.DataFrame + + +def _gate_diagnostic(node_id, node): + return { + "node_id": node_id, + "kernel_ref": node.kernel_ref, + "outcome": node.receipt.get("outcome"), + "evidence": dict(node.receipt.get("evidence", {})), + } + + +def _rollup(periods, field): + verdicts = {period[field] for period in periods.values()} + if "fail" in verdicts: + return "fail" + if "not_evaluated" in verdicts: + return "not_evaluated" + return "pass" if "pass" in verdicts else "not_applicable" + + +def run_mortality_trajectory( + *, + training, + rates, + initial, + holdouts, + end_year, + output_dir, + boundary_year=2014, + external_vintage_year=2014, + experiment_id="mortality", + replicate=0, + base_seed=0, + household_accounting=False, +): + """Run the optional annual DAG, retaining explicit engineering evidence.""" + if household_accounting: + raise ValueError("household accounting is unsupported by this graph") + graph, registry = build_trajectory_graph( + end_year=end_year, + boundary_year=boundary_year, + external_vintage_year=external_vintage_year, + experiment_id=experiment_id, + replicate=replicate, + base_seed=base_seed, + ) + years = range(boundary_year + 1, end_year + 1) + if ( + not isinstance(holdouts, dict) + or any(type(year) is not int for year in holdouts) + or set(holdouts) != set(years) + ): + raise ValueError("holdouts must supply exactly one source per year") + sources = { + "training": Path(training).resolve(), + "rates": Path(rates).resolve(), + "initial": Path(initial).resolve(), + **{ + f"holdout_{year}": Path(holdouts[year]).resolve() for year in years + }, + } + output = Path(output_dir).resolve() + output.mkdir(parents=True, exist_ok=True) + store = ContentStore(output / "store") + manifest = run_graph( + compile_graph(graph), sources=sources, store=store, kernels=registry + ) + (output / "manifest.json").write_text(manifest.to_json(), encoding="utf-8") + model_key = manifest.nodes["fit"].opaque_artifacts["model"] + model_payload = store.load_bytes(model_key) + periods = {} + last_population = "initial" + for year in years: + evaluation = manifest.nodes[f"evaluate_{year}"] + key = evaluation.opaque_artifacts.get("report") + if key is not None: + periods[str(year)] = parse_json(store.load_bytes(key)) + else: + period = { + "scope": "synthetic_engineering", + "from_year": year - 1, + "year": year, + "engineering_verdict": "not_evaluated", + "fixture_verdict": "not_evaluated", + } + application = manifest.nodes[f"apply_{year}"] + if application.receipt.get("outcome") == "fail": + period["application_gate"] = _gate_diagnostic( + f"apply_{year}", application + ) + elif evaluation.receipt.get("outcome") == "fail": + period["evaluation_gate"] = _gate_diagnostic( + f"evaluate_{year}", evaluation + ) + periods[str(year)] = period + advance = manifest.nodes[f"advance_{year}"] + if advance.frame_key is not None: + last_population = f"advance_{year}" + population = manifest.population(last_population) + observations = population.table(rt.OBS) + period_values = population.table("period").set_index("period_id").period + trajectory = ( + pd.DataFrame( + { + "person_id": observations[rt.PID].to_numpy(dtype=np.int64), + "age": observations.age.to_numpy(dtype=np.int64), + "year": observations[rt.PERIOD_ID] + .map(period_values) + .to_numpy(dtype=np.int64), + "weight": population.weights_for(rt.OBS).values, + } + ) + .sort_values(["year", "person_id"]) + .reset_index(drop=True) + ) + report = { + "scope": "synthetic_engineering", + "boundary_year": boundary_year, + "end_year": end_year, + "completed_year": max( + boundary_year, + *( + node.receipt.get("completed_year", boundary_year) + for name, node in manifest.nodes.items() + if name.startswith("apply_") + ), + ), + "periods": periods, + "engineering_verdict": _rollup(periods, "engineering_verdict"), + "fixture_verdict": _rollup(periods, "fixture_verdict"), + "execution_status": ( + "failed" + if any( + period["engineering_verdict"] == "not_evaluated" + for period in periods.values() + ) + else "complete" + ), + "model_artifact_key": model_key, + "node_keys": {name: node.key for name, node in manifest.nodes.items()}, + "cache_hits": { + name: node.hit for name, node in manifest.nodes.items() + }, + "limitations": [ + "Synthetic engineering fixture; no scientific or national-population certification.", + "The mortality fit's external-rate factor cancels in the fitted-window level.", + "Household accounting, births, immigration, and the full M6 loop are outside this graph.", + "The pinned core executes guarded descendants after a typed failure; application-level blocked does not mean native executor unreached.", + ], + } + (output / "report.json").write_bytes(json_bytes(report)) + (output / "model.json").write_bytes(model_payload) + trajectory.to_csv(output / "trajectory.csv", index=False) + return MortalityTrajectoryRun(manifest, report, model_payload, trajectory) diff --git a/tests/README-tiers.md b/tests/README-tiers.md index 19b4bcf2..d19e45fe 100644 --- a/tests/README-tiers.md +++ b/tests/README-tiers.md @@ -38,9 +38,9 @@ pytest --collect-only -q -m oracle_policyengine | tail -1 | Tier | Tests at HEAD | |---|---:| -| `unit` | 1,596 | +| `unit` | 1,624 | | `artifact` | 2,668 | | `integration_psid` | 848 | | `reproduction_legacy` | 520 | | `oracle_policyengine` | 159 | -| **Total** | **5,791** | +| **Total** | **5,819** | diff --git a/tests/estimates/test_birth_evidence_artifact.py b/tests/estimates/test_birth_evidence_artifact.py index 2f4c176c..3641f43f 100644 --- a/tests/estimates/test_birth_evidence_artifact.py +++ b/tests/estimates/test_birth_evidence_artifact.py @@ -94,6 +94,7 @@ def test_post_review_sources_are_outside_historical_reducer_identity(): Path("src/populace_dynamics/graph/model.py"), Path("src/populace_dynamics/graph/runtime.py"), Path("src/populace_dynamics/graph/synthetic.py"), + Path("src/populace_dynamics/graph/trajectory.py"), ) assert reducer.POST_REVIEW_SHARED_SOURCE_BLOBS == { Path( diff --git a/tests/test_graph_mortality_trajectory.py b/tests/test_graph_mortality_trajectory.py new file mode 100644 index 00000000..70494b14 --- /dev/null +++ b/tests/test_graph_mortality_trajectory.py @@ -0,0 +1,664 @@ +"""Independent synthetic engineering checks for the annual mortality DAG. + +All sources are generated in the test directory. Aggregate fixtures below +are deliberately hand specified and are neither native population evidence +nor scientific acceptance thresholds. +""" + +import json +from types import SimpleNamespace + +import numpy as np +import pandas as pd +import pytest + +from populace_dynamics.graph.model import fit_mortality +from populace_dynamics.graph.synthetic import write_synthetic_inputs + +TRAJECTORY_COLUMNS = ["person_id", "age", "year", "weight"] + + +@pytest.fixture +def runtime(): + from populace_dynamics.graph._compat import require_graph + + try: + require_graph() + except ImportError as error: + pytest.skip(str(error)) + from populace_dynamics.graph import run_mortality_trajectory + + return run_mortality_trajectory + + +def _read(path): + return json.loads(path.read_text()) + + +def _write(path, value): + path.write_text(json.dumps(value)) + + +@pytest.fixture +def inputs(tmp_path): + sources = write_synthetic_inputs(tmp_path / "inputs") + sources.pop("holdout") + holdouts = {} + for year in range(2015, 2019): + path = tmp_path / "inputs" / f"aggregate-{year}.json" + _write( + path, + { + "scope": "synthetic_engineering", + "year": year, + "expected_death_rate": 0.2, + "fixture_max_abs_death_rate_gap": 0.25, + }, + ) + holdouts[year] = path + return {**sources, "holdouts": holdouts} + + +def _run(runtime, inputs, tmp_path, *, end_year=2017, **kwargs): + sources = { + **inputs, + "holdouts": { + year: path + for year, path in inputs["holdouts"].items() + if year <= end_year + }, + } + return runtime( + **sources, + end_year=end_year, + output_dir=tmp_path / "output", + **kwargs, + ) + + +def _ordered(frame): + return ( + frame[TRAJECTORY_COLUMNS] + .sort_values(["year", "person_id"]) + .reset_index(drop=True) + ) + + +def _direct_projection(inputs, *, end_year=2017, **coordinates): + """Use the original fit/steps and an independently assembled RNG key.""" + from microcosm.graph.randomness import keyed_uniform + + from populace_dynamics.engine.steps import advance_age, apply_mortality + + artifact = fit_mortality( + pd.DataFrame(_read(inputs["training"])), + pd.DataFrame(_read(inputs["rates"])), + boundary_year=2014, + external_vintage_year=2014, + ) + stream = ( + "sha256-u53-v1", + coordinates.get("experiment_id", "mortality"), + coordinates.get("replicate", 0), + coordinates.get("base_seed", 0), + ) + current = pd.DataFrame(_read(inputs["initial"])).sort_values("person_id") + current["year"] = 2014 + history = [current[TRAJECTORY_COLUMNS].copy()] + diagnostics = {} + for year in range(2015, end_year + 1): + current = current.sort_values("person_id").reset_index(drop=True) + uniforms = keyed_uniform( + stream=stream, + keys=[ + (int(pid), "mortality", year, 0) for pid in current.person_id + ], + ) + + class FixedUniforms: + def __init__(self, values): + self.values = values + + def random(self, n): + assert n == len(self.values) + return self.values.copy() + + context = SimpleNamespace(rng_registry=None, year=year, metadata={}) + survived = apply_mortality( + current, context, FixedUniforms(uniforms), model=artifact.model + ) + future = advance_age(survived, context, np.random.default_rng(0)) + probability = artifact.model.probabilities(current) + diagnostics[str(year)] = { + "from_year": year - 1, + "year": year, + "initial_records": len(current), + "survivor_records": len(future), + "expected_deaths": float(np.dot(current.weight, probability)), + "generated_deaths": float( + current.loc[ + ~current.person_id.isin(future.person_id), "weight" + ].sum() + ), + "start_mass": float(current.weight.sum()), + "next_period_mass": float(future.weight.sum()), + } + history.append(future[TRAJECTORY_COLUMNS].copy()) + current = future + return artifact.to_bytes(), _ordered(pd.concat(history)), diagnostics + + +def _set_death_regime(inputs, *, all_die): + training = _read(inputs["training"]) + for row in training: + row["death"] = 1.0 if all_die else 0.0 + row["exposure"] = 1e-9 if all_die else 1.0 + _write(inputs["training"], training) + for path in inputs["holdouts"].values(): + holdout = _read(path) + holdout["expected_death_rate"] = 1.0 if all_die else 0.0 + _write(path, holdout) + + +@pytest.mark.parametrize( + "coordinates", + [ + {}, + {"experiment_id": "trajectory-alternative"}, + {"replicate": 7}, + {"base_seed": 831}, + ], +) +def test_annual_projection_matches_independent_steps_and_weighted_diagnostics( + runtime, inputs, tmp_path, coordinates +): + result = _run(runtime, inputs, tmp_path, **coordinates) + payload, expected, diagnostics = _direct_projection(inputs, **coordinates) + assert result.model_payload == payload + pd.testing.assert_frame_equal(_ordered(result.trajectory), expected) + assert result.report["scope"] == "synthetic_engineering" + assert result.report["engineering_verdict"] == "pass" + assert result.report["fixture_verdict"] == "pass" + assert set(result.report["periods"]) == {"2015", "2016", "2017"} + for year, expected_period in diagnostics.items(): + actual = result.report["periods"][year] + for field, value in expected_period.items(): + assert actual[field] == pytest.approx(value), (year, field) + assert actual["engineering_verdict"] == "pass" + assert actual["fixture_verdict"] == "pass" + assert actual["start_mass"] == pytest.approx( + actual["generated_deaths"] + actual["next_period_mass"] + ) + + +def test_expansion_preserves_every_historical_row_and_person_link( + runtime, inputs, tmp_path +): + result = _run(runtime, inputs, tmp_path) + previous = result.manifest.population("initial").table("person_period") + for year in range(2015, 2018): + population = result.manifest.population(f"advance_{year}") + observations = population.table("person_period") + assert observations.person_period_id.is_unique + assert not observations.duplicated( + ["person_period_person_id", "person_period_period_id"] + ).any() + retained = observations.loc[ + observations.person_period_id.isin(previous.person_period_id), + previous.columns, + ] + pd.testing.assert_frame_equal( + retained.sort_values("person_period_id").reset_index(drop=True), + previous.sort_values("person_period_id").reset_index(drop=True), + ) + entrants = observations.loc[ + observations.person_period_period_id == year + ] + at_risk = previous.loc[ + previous.person_period_period_id == year - 1 + ].set_index("person_period_person_id") + assert set(entrants.person_period_person_id) <= set(at_risk.index) + for row in entrants.itertuples(index=False): + parent = at_risk.loc[row.person_period_person_id] + assert row.age == parent.age + 1 + assert row.sex == parent.sex + previous = observations + assert set(previous.person_period_period_id) == {2014, 2015, 2016, 2017} + + +def test_cold_warm_cache_and_exported_artifacts(runtime, inputs, tmp_path): + cold = _run(runtime, inputs, tmp_path) + warm = _run(runtime, inputs, tmp_path) + assert not any(node.hit for node in cold.manifest.nodes.values()) + assert all(node.hit for node in warm.manifest.nodes.values()) + assert type(warm.manifest).__module__.startswith("microcosm.graph") + assert cold.model_payload == warm.model_payload + assert cold.report["periods"] == warm.report["periods"] + pd.testing.assert_frame_equal(cold.trajectory, warm.trajectory) + output = tmp_path / "output" + assert _read(output / "report.json") == warm.report + assert _read(output / "manifest.json") == json.loads( + warm.manifest.to_json() + ) + assert (output / "model.json").read_bytes() == warm.model_payload + pd.testing.assert_frame_equal( + _ordered(pd.read_csv(output / "trajectory.csv")), + _ordered(warm.trajectory), + ) + + +def test_horizon_extension_reuses_fit_and_existing_annual_nodes( + runtime, inputs, tmp_path +): + short = _run(runtime, inputs, tmp_path, end_year=2015) + extended = _run(runtime, inputs, tmp_path, end_year=2018) + for name, node in short.manifest.nodes.items(): + assert extended.manifest.nodes[name].hit, name + assert extended.manifest.nodes[name].key == node.key, name + for year in range(2016, 2019): + for prefix in ("apply", "advance", "age", "evaluate"): + assert not extended.manifest.nodes[f"{prefix}_{year}"].hit + assert short.model_payload == extended.model_payload + pd.testing.assert_frame_equal( + _ordered(short.trajectory), + _ordered(extended.trajectory.query("year <= 2015")), + ) + + +@pytest.mark.parametrize( + "coordinates", + [ + {"experiment_id": "trajectory-alternative"}, + {"replicate": 7}, + {"base_seed": 831}, + ], +) +def test_stream_change_reuses_fit_but_invalidates_each_application( + runtime, inputs, tmp_path, coordinates +): + original = _run(runtime, inputs, tmp_path) + changed = _run(runtime, inputs, tmp_path, **coordinates) + for name in ("training", "fit", "initial"): + assert changed.manifest.nodes[name].hit + assert ( + changed.manifest.nodes[name].key + == original.manifest.nodes[name].key + ) + for year in range(2015, 2018): + name = f"apply_{year}" + assert not changed.manifest.nodes[name].hit + assert ( + changed.manifest.nodes[name].key + != original.manifest.nodes[name].key + ) + assert changed.model_payload == original.model_payload + _, expected, _ = _direct_projection(inputs, **coordinates) + pd.testing.assert_frame_equal(_ordered(changed.trajectory), expected) + + +def test_holdout_change_invalidates_only_its_own_evaluation( + runtime, inputs, tmp_path +): + original = _run(runtime, inputs, tmp_path) + holdout = _read(inputs["holdouts"][2016]) + holdout["expected_death_rate"] = 1.0 + holdout["fixture_max_abs_death_rate_gap"] = 0.0 + _write(inputs["holdouts"][2016], holdout) + changed = _run(runtime, inputs, tmp_path) + for name, node in changed.manifest.nodes.items(): + if name == "evaluate_2016": + assert not node.hit + assert node.key != original.manifest.nodes[name].key + else: + assert node.hit, name + assert node.key == original.manifest.nodes[name].key, name + assert changed.report["engineering_verdict"] == "pass" + assert changed.report["fixture_verdict"] == "fail" + assert changed.report["periods"]["2016"]["fixture_verdict"] == "fail" + for year in ("2015", "2017"): + assert ( + changed.report["periods"][year] == original.report["periods"][year] + ) + assert changed.model_payload == original.model_payload + pd.testing.assert_frame_equal(changed.trajectory, original.trajectory) + + +def test_recipient_change_reuses_the_fitted_model(runtime, inputs, tmp_path): + original = _run(runtime, inputs, tmp_path) + initial = _read(inputs["initial"]) + initial[0]["age"] += 1 + initial[0]["weight"] *= 2 + _write(inputs["initial"], initial) + changed = _run(runtime, inputs, tmp_path) + assert changed.manifest.nodes["fit"].hit + assert changed.model_payload == original.model_payload + assert not changed.manifest.nodes["initial"].hit + assert not changed.manifest.nodes["apply_2015"].hit + _, expected, _ = _direct_projection(inputs) + pd.testing.assert_frame_equal(_ordered(changed.trajectory), expected) + + +def test_training_change_refits_and_reapplies(runtime, inputs, tmp_path): + original = _run(runtime, inputs, tmp_path) + training = _read(inputs["training"]) + training[0]["start_weight"] = 8.0 + _write(inputs["training"], training) + changed = _run(runtime, inputs, tmp_path) + assert not changed.manifest.nodes["fit"].hit + assert not changed.manifest.nodes["apply_2015"].hit + assert changed.model_payload != original.model_payload + payload, expected, _ = _direct_projection(inputs) + assert changed.model_payload == payload + pd.testing.assert_frame_equal(_ordered(changed.trajectory), expected) + + +def test_complete_extinction_leaves_no_future_period_groups( + runtime, inputs, tmp_path +): + _set_death_regime(inputs, all_die=True) + result = _run(runtime, inputs, tmp_path) + warm = _run(runtime, inputs, tmp_path) + assert all(node.hit for node in warm.manifest.nodes.values()) + assert set(result.trajectory.year) == {2014} + first = result.report["periods"]["2015"] + assert first["generated_deaths"] == first["start_mass"] + assert first["survivor_records"] == 0 + assert first["next_period_mass"] == 0 + for year in (2016, 2017): + period = result.report["periods"][str(year)] + for field in ( + "initial_records", + "survivor_records", + "expected_deaths", + "generated_deaths", + "start_mass", + "next_period_mass", + ): + assert period[field] == 0, (year, field) + assert period["engineering_verdict"] == "not_applicable" + assert period["fixture_verdict"] == "not_applicable" + for year in range(2015, 2018): + population = result.manifest.population(f"advance_{year}") + assert population.table("period").period.tolist() == [2014] + assert set( + population.table("person_period").person_period_period_id + ) == {2014} + + +def test_zero_mortality_keeps_all_people_and_each_periods_mass( + runtime, inputs, tmp_path +): + _set_death_regime(inputs, all_die=False) + result = _run(runtime, inputs, tmp_path) + initial = pd.DataFrame(_read(inputs["initial"])) + for year in range(2014, 2018): + period = result.trajectory.loc[result.trajectory.year == year] + assert set(period.person_id) == set(initial.person_id) + assert period.weight.sum() == initial.weight.sum() + _, expected, _ = _direct_projection(inputs) + pd.testing.assert_frame_equal(_ordered(result.trajectory), expected) + for period in result.report["periods"].values(): + assert period["expected_deaths"] == 0 + assert period["generated_deaths"] == 0 + assert period["fixture_verdict"] == "pass" + + +def test_missing_annual_holdout_fails_closed(runtime, inputs, tmp_path): + inputs["holdouts"].pop(2016) + with pytest.raises(ValueError, match="holdout"): + _run(runtime, inputs, tmp_path) + + +@pytest.mark.parametrize("mutation", ["year", "scope", "rate", "json"]) +def test_bad_holdout_retains_gate_diagnostics_without_changing_simulation( + runtime, inputs, tmp_path, mutation +): + original = _run(runtime, inputs, tmp_path) + path = inputs["holdouts"][2016] + holdout = _read(path) + if mutation == "year": + holdout["year"] = 2015 + elif mutation == "scope": + holdout["scope"] = "scientific_acceptance" + elif mutation == "rate": + holdout["expected_death_rate"] = float("nan") + _write(path, holdout) + if mutation == "json": + path.write_text("{broken JSON") + for cached_failure in (False, True): + result = _run(runtime, inputs, tmp_path) + gate = result.manifest.nodes["evaluate_2016"] + assert gate.hit is cached_failure + assert gate.receipt["outcome"] == "fail" + diagnostic = result.report["periods"]["2016"]["evaluation_gate"] + assert diagnostic["node_id"] == "evaluate_2016" + assert diagnostic["outcome"] == "fail" + assert diagnostic["evidence"]["exception_type"] == "ValueError" + assert diagnostic["evidence"]["message"] + assert result.report["fixture_verdict"] == "not_evaluated" + for name, node in result.manifest.nodes.items(): + if name != "evaluate_2016": + assert node.hit, name + assert node.key == original.manifest.nodes[name].key + pd.testing.assert_frame_equal(result.trajectory, original.trajectory) + assert _read(tmp_path / "output" / "report.json") == result.report + assert _read(tmp_path / "output" / "manifest.json") == json.loads( + result.manifest.to_json() + ) + + +def test_unsupported_age_stops_future_application_and_preserves_evidence( + runtime, inputs, tmp_path +): + _set_death_regime(inputs, all_die=False) + initial = _read(inputs["initial"]) + initial[0]["age"] = 120 + _write(inputs["initial"], initial) + short = _run(runtime, inputs, tmp_path, end_year=2015) + assert short.trajectory.query("year == 2015").age.max() == 121 + for cached_failure in (False, True): + result = _run(runtime, inputs, tmp_path) + gate = result.manifest.nodes["apply_2016"] + assert gate.hit is cached_failure + assert gate.receipt["outcome"] == "fail" + assert set(gate.opaque_artifacts) == {"transition"} + assert gate.receipt["application_status"] == "failed" + diagnostic = result.report["periods"]["2016"]["application_gate"] + assert diagnostic["node_id"] == "apply_2016" + assert diagnostic["outcome"] == "fail" + assert diagnostic["evidence"]["exception_type"] == "ValueError" + assert "age" in diagnostic["evidence"]["message"].lower() + # The exact pinned core executes guarded descendants. Their native + # receipts stay honest; application-level blocked is not unreached. + for name in ("advance_2016", "apply_2017", "advance_2017"): + assert ( + result.manifest.nodes[name].receipt["application_status"] + == "blocked" + ) + for year in (2016, 2017): + assert ( + result.manifest.nodes[f"evaluate_{year}"].receipt["outcome"] + == "evidence_absent" + ) + period = result.report["periods"][str(year)] + assert period["completed_year"] == 2015 + assert period["engineering_verdict"] == "not_evaluated" + population = result.manifest.population(f"advance_{year}") + pd.testing.assert_frame_equal( + population.table("person_period"), + short.manifest.population("advance_2015").table( + "person_period" + ), + ) + assert result.report["periods"]["2017"]["application_status"] == ( + "blocked" + ) + assert result.report["completed_year"] == 2015 + assert result.report["engineering_verdict"] == "not_evaluated" + assert result.model_payload == short.model_payload + pd.testing.assert_frame_equal(result.trajectory, short.trajectory) + assert _read(tmp_path / "output" / "report.json") == result.report + assert _read(tmp_path / "output" / "manifest.json") == json.loads( + result.manifest.to_json() + ) + + +def test_household_accounting_remains_explicitly_unsupported( + runtime, inputs, tmp_path +): + with pytest.raises(ValueError, match="household"): + _run(runtime, inputs, tmp_path, household_accounting=True) + + +def test_known_fixture_failure_survives_later_missing_evaluation( + runtime, inputs, tmp_path +): + failed_fixture = _read(inputs["holdouts"][2015]) + failed_fixture["expected_death_rate"] = 1.0 + failed_fixture["fixture_max_abs_death_rate_gap"] = 0.0 + _write(inputs["holdouts"][2015], failed_fixture) + inputs["holdouts"][2016].write_text("{broken JSON") + result = _run(runtime, inputs, tmp_path) + assert result.report["periods"]["2015"]["fixture_verdict"] == "fail" + assert result.report["periods"]["2016"]["fixture_verdict"] == ( + "not_evaluated" + ) + assert result.report["periods"]["2017"]["engineering_verdict"] == "pass" + assert result.report["fixture_verdict"] == "fail" + assert result.report["execution_status"] == "failed" + assert result.report["completed_year"] == 2017 + + +def test_zero_risk_set_cannot_certify_an_unexpected_future_observation( + runtime, inputs, tmp_path, monkeypatch +): + from dataclasses import replace + + from populace_dynamics.graph import trajectory + + _set_death_regime(inputs, all_die=True) + original_snapshot = trajectory._snapshot + + def snapshot_with_unexpected_future_row(context): + result = original_snapshot(context) + if context.params["year"] != 2016: + return result + payload = json.loads(result.artifacts["snapshot"]) + record = dict(payload["observations"][0]) + record["person_period_id"] = 999 + record["person_period_period_id"] = 2016 + record["age"] += 2 + payload["observations"].append(record) + payload["periods"].append({"period_id": 2016, "period": 2016}) + payload["weights"].append(1.0) + return replace( + result, artifacts={"snapshot": json.dumps(payload).encode()} + ) + + monkeypatch.setattr( + trajectory, "_snapshot", snapshot_with_unexpected_future_row + ) + result = _run(runtime, inputs, tmp_path) + period = result.report["periods"]["2016"] + assert period["initial_records"] == 0 + assert period["survivor_records"] == 1 + assert period["engineering_verdict"] == "fail" + assert period["fixture_verdict"] == "not_applicable" + assert result.manifest.nodes["evaluate_2016"].receipt["outcome"] == "fail" + assert result.report["engineering_verdict"] == "fail" + + +@pytest.mark.parametrize( + "payload_kind", ["malformed_json", "unknown_status", "before_boundary"] +) +def test_malformed_prerequisite_does_not_invent_completed_year( + runtime, monkeypatch, payload_kind +): + from populace_dynamics.graph import trajectory + + def unexpected_application(context): + raise AssertionError("invalid prerequisite reached model application") + + monkeypatch.setattr(trajectory, "_apply_complete", unexpected_application) + payload = b"{broken JSON" + if payload_kind != "malformed_json": + payload = json.dumps( + { + "format": "populace-dynamics.mortality-transition", + "schema_version": 1, + "from_year": 2015, + "year": 2016, + "status": ( + "unknown" if payload_kind == "unknown_status" else "failed" + ), + "completed_year": ( + 2013 if payload_kind == "before_boundary" else 2015 + ), + "records": [], + "diagnostic": {"message": "prior failure"}, + } + ).encode() + context = SimpleNamespace( + params={"year": 2017, "boundary_year": 2014}, + artifacts={"previous_transition": SimpleNamespace(payload=payload)}, + ) + result = trajectory._apply(context) + outcome = json.loads(result.artifacts["transition"]) + assert outcome["status"] == "failed" + assert outcome["completed_year"] == 2014 + assert outcome["records"] == [] + assert outcome["diagnostic"]["exception_type"] == "ValueError" + assert result.receipt["outcome"] == "fail" + + +def test_blocked_years_do_not_apply_laws_or_parse_holdouts( + runtime, inputs, tmp_path, monkeypatch +): + from populace_dynamics.graph import trajectory + + _set_death_regime(inputs, all_die=False) + initial = _read(inputs["initial"]) + initial[0]["age"] = 120 + _write(inputs["initial"], initial) + for year in (2016, 2017): + # The executor still hashes these bytes, but guarded evaluations + # must not parse them after application fails in 2016. + inputs["holdouts"][year].write_text("{unparseable held-out fixture") + + application_years, ageing_years, holdout_years = [], [], [] + apply_complete = trajectory._apply_complete + advance_age = trajectory.rt.advance_age + holdout = trajectory._holdout + + def tracked_application(context): + application_years.append(context.params["year"]) + assert context.params["year"] <= 2016 + return apply_complete(context) + + def tracked_ageing(frame, context, rng): + ageing_years.append(context.year) + assert context.year == 2015 + return advance_age(frame, context, rng) + + def tracked_holdout(context): + holdout_years.append(context.params["year"]) + assert context.params["year"] == 2015 + return holdout(context) + + monkeypatch.setattr(trajectory, "_apply_complete", tracked_application) + monkeypatch.setattr(trajectory.rt, "advance_age", tracked_ageing) + monkeypatch.setattr(trajectory, "_holdout", tracked_holdout) + result = _run(runtime, inputs, tmp_path) + assert application_years == [2015, 2016] + assert ageing_years == [2015] + assert holdout_years == [2015] + assert result.report["completed_year"] == 2015 + assert ( + result.report["periods"]["2016"]["application_gate"]["evidence"][ + "exception_type" + ] + == "ValueError" + ) + assert result.report["periods"]["2017"]["application_status"] == ( + "blocked" + ) diff --git a/tests/tier_counts.json b/tests/tier_counts.json index e7b10bcb..c1ba14b2 100644 --- a/tests/tier_counts.json +++ b/tests/tier_counts.json @@ -1,7 +1,7 @@ { "schema_version": 1, "counts": { - "unit": 1596, + "unit": 1624, "artifact": 2668, "integration_psid": 848, "reproduction_legacy": 520,