diff --git a/packages/populace-build/src/populace/build/ledger_artifact.py b/packages/populace-build/src/populace/build/ledger_artifact.py index 1638b792..e9aeffd6 100644 --- a/packages/populace-build/src/populace/build/ledger_artifact.py +++ b/packages/populace-build/src/populace/build/ledger_artifact.py @@ -8,32 +8,66 @@ (PolicyEngine/populace#160, #271) and so tampered or mismatched feeds fail before they calibrate anything. +Strict loading (findings #7, #8): ``validate_rows`` (default on) checks every +row against the vendored ``ledger.consumer_fact.v1`` schema and enforces unique +``aggregate_fact_key``; ``require_pins`` fails closed for release builds --- +bare feeds are rejected, a manifest is required, and both content-hash pins +must be supplied and match. Verified feeds can be vendored byte-for-byte into a +release with :func:`vendor_ledger_consumer_artifact` so the exact facts that +resolved a release travel with it. + Like the rest of Populace's Ledger consumption, this module is duck-typed -against the published artifact contract (stdlib only); it does not import -the Ledger implementation package. +against the published artifact contract (stdlib only); it does not import the +Ledger implementation package. The row-schema pin lives in the sibling +``populace.build.ledger_schema`` module (also stdlib-only). """ from __future__ import annotations import hashlib import json -from dataclasses import dataclass +from collections.abc import Mapping +from dataclasses import dataclass, field from pathlib import Path -from typing import Any +from typing import Any, NoReturn + +from populace.build.ledger_schema import ( + CONSUMER_FACT_SCHEMA_SHA256, + validate_consumer_fact_row, +) __all__ = [ "ALLOWED_LEDGER_ASSERTIONS", "CONSUMER_ARTIFACT_SCHEMA_VERSION", "DEFAULT_LEDGER_ASSERTION", + "VENDORED_ARTIFACT_SCHEMA_VERSION", "LedgerConsumerArtifact", "load_ledger_consumer_artifact", + "vendor_ledger_consumer_artifact", + "verify_vendored_ledger_artifact", ] CONSUMER_ARTIFACT_SCHEMA_VERSION = "policyengine_ledger.consumer_artifact.v1" +VENDORED_ARTIFACT_SCHEMA_VERSION = "populace.vendored_ledger_artifact.v1" ALLOWED_LEDGER_ASSERTIONS = frozenset(("observation", "source_projection")) DEFAULT_LEDGER_ASSERTION = "observation" +def _reject_non_finite_json_constant(token: str) -> NoReturn: + """``json.loads`` ``parse_constant`` hook: reject NaN/Infinity/-Infinity. + + Stdlib ``json`` accepts the JavaScript literals ``NaN``, ``Infinity`` and + ``-Infinity`` by default and hands back the corresponding float, which then + satisfies a ``number`` schema check and silently poisons a calibration + target (finding #7). Consumer facts are a numeric contract, so a non-finite + value is a hard load error. + """ + raise ValueError( + f"Ledger facts JSONL contains the non-finite JSON constant {token!r}; " + "consumer fact numbers must be finite." + ) + + @dataclass(frozen=True) class LedgerConsumerArtifact: """A loaded, hash-verified Ledger consumer fact feed. @@ -41,7 +75,9 @@ class LedgerConsumerArtifact: ``manifest`` and ``manifest_sha256`` are ``None`` when the feed was a bare ``consumer_facts.jsonl`` file rather than an artifact directory; bare feeds are still content-addressed by ``facts_sha256`` so builds can - pin them, but they carry no Ledger-side provenance. Rows are exactly as + pin them, but they carry no Ledger-side provenance and are rejected under + ``require_pins``. When ``rows_validated`` is set every row passed the + vendored ``ledger.consumer_fact.v1`` schema; otherwise rows are exactly as published: an ``assertion`` field is validated when present and never fabricated when absent (missing means observation-by-default to readers). """ @@ -51,6 +87,10 @@ class LedgerConsumerArtifact: facts_sha256: str manifest: dict[str, Any] | None = None manifest_sha256: str | None = None + require_pins: bool = False + rows_validated: bool = False + schema_sha256: str | None = None + profile_hash_semantics: Mapping[str, str] = field(default_factory=dict) @property def fact_row_count(self) -> int: @@ -61,8 +101,12 @@ def provenance(self) -> dict[str, Any]: """Ledger-artifact identity block for build and release manifests.""" payload: dict[str, Any] = { "path_name": self.path.name, + "path": str(self.path), "fact_row_count": self.fact_row_count, "facts_sha256": self.facts_sha256, + "require_pins": self.require_pins, + "rows_validated": self.rows_validated, + "consumer_fact_schema_sha256": self.schema_sha256, } if self.manifest is not None: payload["schema_version"] = self.manifest.get("schema_version") @@ -73,6 +117,9 @@ def provenance(self) -> dict[str, Any]: str(profile_id): { "sha256": (profile_meta or {}).get("sha256"), "target_count": (profile_meta or {}).get("target_count"), + "hash_semantics": self.profile_hash_semantics.get( + str(profile_id) + ), } for profile_id, profile_meta in sorted(profiles.items()) } @@ -87,17 +134,41 @@ def load_ledger_consumer_artifact( *, expected_facts_sha256: str | None = None, expected_manifest_sha256: str | None = None, + require_pins: bool = False, + validate_rows: bool = True, ) -> LedgerConsumerArtifact: """Load a Ledger consumer artifact directory or bare consumer-facts file. Artifact directories are verified against their own manifest: the fact - file's SHA-256 must match ``manifest.facts_sha256``. The optional - ``expected_*`` pins let a build config assert the exact artifact it was - reviewed against; any mismatch raises before facts are used. + file's SHA-256 must match ``manifest.facts_sha256`` and every profile file + the manifest lists is re-hashed from bytes. The optional ``expected_*`` + pins let a build config assert the exact artifact it was reviewed against; + any mismatch raises before facts are used. + + ``validate_rows`` (default) validates every row against the vendored + ``ledger.consumer_fact.v1`` schema and enforces unique ``aggregate_fact_key``. + ``require_pins`` fails closed for release builds: a bare feed is rejected, a + manifest is required, and both ``expected_facts_sha256`` and + ``expected_manifest_sha256`` must be provided and match. """ artifact_path = Path(path) if not artifact_path.exists(): raise FileNotFoundError(f"Ledger consumer artifact not found: {artifact_path}") + artifact_path = artifact_path.resolve() + + if require_pins: + if not artifact_path.is_dir(): + raise ValueError( + "Ledger consumer artifact must be a hash-pinned artifact " + "directory under require_pins; bare consumer-facts files carry " + f"no manifest and are rejected: {artifact_path}." + ) + if expected_facts_sha256 is None or expected_manifest_sha256 is None: + raise ValueError( + "Ledger consumer artifact require_pins demands both " + "expected_facts_sha256 and expected_manifest_sha256; got " + f"facts={expected_facts_sha256!r}, manifest={expected_manifest_sha256!r}." + ) manifest: dict[str, Any] | None = None manifest_sha256: str | None = None @@ -159,24 +230,40 @@ def load_ledger_consumer_artifact( f"hash: {manifest_sha256} != {expected_manifest_sha256}." ) + profile_hash_semantics: dict[str, str] = {} + if manifest is not None: + profile_hash_semantics = _verify_manifest_profiles( + artifact_path, manifest, require_pins=require_pins + ) + return LedgerConsumerArtifact( path=artifact_path, - facts=_load_fact_rows(facts_path), + facts=_load_fact_rows(facts_path, validate_rows=validate_rows), facts_sha256=facts_sha256, manifest=manifest, manifest_sha256=manifest_sha256, + require_pins=require_pins, + rows_validated=validate_rows, + schema_sha256=CONSUMER_FACT_SCHEMA_SHA256 if validate_rows else None, + profile_hash_semantics=profile_hash_semantics, ) -def _load_fact_rows(path: Path) -> tuple[dict[str, Any], ...]: +def _load_fact_rows( + path: Path, *, validate_rows: bool +) -> tuple[dict[str, Any], ...]: rows: list[dict[str, Any]] = [] + seen_aggregate_keys: set[str] = set() + source = str(path) with path.open() as file: for line_number, line in enumerate(file, start=1): stripped = line.strip() if not stripped: continue try: - row = json.loads(stripped) + row = json.loads( + stripped, parse_constant=_reject_non_finite_json_constant + ) except json.JSONDecodeError as exc: raise ValueError( f"Invalid Ledger facts JSONL row {line_number}: {exc.msg}" @@ -186,23 +273,245 @@ def _load_fact_rows(path: Path) -> tuple[dict[str, Any], ...]: f"Invalid Ledger facts JSONL row {line_number}: expected " f"object, got {type(row).__name__}." ) - assertion = row.get("assertion", DEFAULT_LEDGER_ASSERTION) - if assertion not in ALLOWED_LEDGER_ASSERTIONS: - raise ValueError( - f"Ledger facts JSONL row {line_number} has unsupported " - f"assertion {assertion!r}; expected one of " - f"{sorted(ALLOWED_LEDGER_ASSERTIONS)}." + if validate_rows: + validate_consumer_fact_row( + row, line_number=line_number, source=source ) - # Do not stamp the default onto rows that omit the field: legacy - # feeds predate the assertion schema, and fabricating the key - # would make downstream assertion checks treat unlabeled - # publisher projections as mistyped observations. + aggregate_key = row["aggregate_fact_key"] + if aggregate_key in seen_aggregate_keys: + raise ValueError( + f"Ledger facts JSONL row {line_number} repeats " + f"aggregate_fact_key {aggregate_key!r}; consumer facts " + "must be unique." + ) + seen_aggregate_keys.add(aggregate_key) + else: + # Legacy passthrough: the pinned schema is not enforced, so + # only the assertion enum is checked and a missing assertion is + # left untouched (readers treat absence as observation-by- + # default; fabricating it would mistype publisher projections). + assertion = row.get("assertion", DEFAULT_LEDGER_ASSERTION) + if assertion not in ALLOWED_LEDGER_ASSERTIONS: + raise ValueError( + f"Ledger facts JSONL row {line_number} has unsupported " + f"assertion {assertion!r}; expected one of " + f"{sorted(ALLOWED_LEDGER_ASSERTIONS)}." + ) rows.append(row) if not rows: raise ValueError(f"Ledger facts feed is empty: {path}") return tuple(rows) +def _verify_manifest_profiles( + artifact_dir: Path, + manifest: Mapping[str, Any], + *, + require_pins: bool, +) -> dict[str, str]: + """Re-hash each manifest-listed profile file from bytes. + + Accepts an exact byte match, or --- only for a legacy manifest that predates + the schema pin --- a match against the bytes minus one trailing newline + (upstream Ledger's legacy pre-newline hash semantics), recording + ``legacy_pre_newline`` in provenance rather than accepting it silently. The + pre-newline exception is version-gated exactly as Ledger's loader does it: + it is available only when the manifest omits ``consumer_fact_schema_sha256`` + (finding #12). A manifest that declares the schema pin is a modern, fully + pinned feed and its profiles must match byte-for-byte. Under ``require_pins`` + a listed profile whose file is absent is a hard error. + """ + profiles = manifest.get("profiles") + semantics: dict[str, str] = {} + if not isinstance(profiles, dict): + return semantics + allow_legacy_pre_newline = manifest.get("consumer_fact_schema_sha256") is None + for profile_id, profile_meta in sorted(profiles.items()): + meta = profile_meta or {} + profile_file = _find_profile_file(artifact_dir, str(profile_id), meta) + if profile_file is None: + if require_pins: + raise ValueError( + "Ledger consumer artifact manifest lists profile " + f"{profile_id!r} but no profile file was found in " + f"{artifact_dir}." + ) + continue + declared = meta.get("sha256") + if not declared: + raise ValueError( + f"Ledger consumer artifact profile {profile_id!r} file " + f"{profile_file.name} has no manifest sha256 to verify against." + ) + actual, semantic = _profile_hash_match( + profile_file, + str(declared), + allow_legacy_pre_newline=allow_legacy_pre_newline, + ) + if semantic is None: + legacy_note = ( + "" + if allow_legacy_pre_newline + else " (a schema-pinned manifest requires an exact byte match; " + "the legacy pre-newline hash is not accepted)" + ) + raise ValueError( + f"Ledger consumer artifact profile {profile_id!r} file " + f"{profile_file.name} does not match its manifest hash: " + f"{actual} != {declared}{legacy_note}." + ) + semantics[str(profile_id)] = semantic + return semantics + + +def _find_profile_file( + artifact_dir: Path, profile_id: str, meta: Mapping[str, Any] +) -> Path | None: + for key in ("path", "file", "filename"): + declared_name = meta.get(key) + if declared_name: + candidate = artifact_dir / str(declared_name) + return candidate if candidate.is_file() else None + for candidate in ( + artifact_dir / "profiles" / f"{profile_id}.json", + artifact_dir / f"{profile_id}.json", + ): + if candidate.is_file(): + return candidate + return None + + +def _profile_hash_match( + path: Path, + declared: str, + *, + allow_legacy_pre_newline: bool, +) -> tuple[str, str | None]: + raw = path.read_bytes() + exact = hashlib.sha256(raw).hexdigest() + if exact == declared: + return exact, "exact" + if allow_legacy_pre_newline and raw.endswith(b"\n"): + pre_newline = hashlib.sha256(raw[:-1]).hexdigest() + if pre_newline == declared: + return pre_newline, "legacy_pre_newline" + return exact, None + + +def vendor_ledger_consumer_artifact( + artifact: LedgerConsumerArtifact, + dest_dir: str | Path, + *, + verified_facts_sha256: str | None = None, + verified_manifest_sha256: str | None = None, +) -> dict[str, Any]: + """Copy a verified artifact's exact bytes into a release directory. + + Writes ``manifest.json``, ``consumer_facts.jsonl``, any ``profiles/*.json`` + and ``coverage.json`` verbatim, plus a ``vendored_manifest.json`` recording + each file's sha256, byte count, and the verified pins. Each copy's hash is + recomputed from the written bytes and must match the source, so a copy that + diverges fails the release. Returns the vendored block for the release + manifest. + """ + source_dir = artifact.path + if not source_dir.is_dir(): + raise ValueError( + "Only directory artifacts can be vendored into a release; got a " + f"bare consumer-facts feed: {source_dir}." + ) + destination = Path(dest_dir) + destination.mkdir(parents=True, exist_ok=True) + + files_meta: list[dict[str, Any]] = [] + for relative in _artifact_files_to_vendor(source_dir): + source_file = source_dir / relative + source_bytes = source_file.read_bytes() + source_sha = hashlib.sha256(source_bytes).hexdigest() + copy_path = destination / relative + copy_path.parent.mkdir(parents=True, exist_ok=True) + copy_path.write_bytes(source_bytes) + copy_bytes = copy_path.read_bytes() + copy_sha = hashlib.sha256(copy_bytes).hexdigest() + if copy_sha != source_sha: + raise ValueError( + f"Vendored Ledger artifact copy {relative!r} diverged from its " + f"source: {copy_sha} != {source_sha}." + ) + files_meta.append( + { + "path": relative, + "sha256": copy_sha, + "size_bytes": len(copy_bytes), + } + ) + + vendored: dict[str, Any] = { + "schema_version": VENDORED_ARTIFACT_SCHEMA_VERSION, + "source_path": str(source_dir), + "facts_sha256": artifact.facts_sha256, + "manifest_sha256": artifact.manifest_sha256, + "verified_facts_sha256": verified_facts_sha256, + "verified_manifest_sha256": verified_manifest_sha256, + "consumer_fact_schema_sha256": artifact.schema_sha256, + "files": files_meta, + } + (destination / "vendored_manifest.json").write_text( + json.dumps(vendored, indent=2, sort_keys=True) + "\n" + ) + return vendored + + +def verify_vendored_ledger_artifact(dest_dir: str | Path) -> dict[str, Any]: + """Re-hash a vendored artifact against its ``vendored_manifest.json``. + + Raises ``ValueError`` if any vendored file is missing or its bytes no + longer match the recorded sha256 --- the tamper check for a release's + embedded facts. + """ + destination = Path(dest_dir) + manifest_path = destination / "vendored_manifest.json" + if not manifest_path.is_file(): + raise ValueError( + f"Vendored Ledger artifact has no vendored_manifest.json: {destination}." + ) + vendored = json.loads(manifest_path.read_bytes()) + for entry in vendored.get("files", ()): + relative = entry["path"] + copy_path = destination / relative + if not copy_path.is_file(): + raise ValueError( + f"Vendored Ledger artifact file {relative!r} is missing from " + f"{destination}." + ) + raw = copy_path.read_bytes() + actual = hashlib.sha256(raw).hexdigest() + if actual != entry["sha256"]: + raise ValueError( + f"Vendored Ledger artifact file {relative!r} was tampered: " + f"{actual} != {entry['sha256']}." + ) + if len(raw) != entry["size_bytes"]: + raise ValueError( + f"Vendored Ledger artifact file {relative!r} size changed: " + f"{len(raw)} != {entry['size_bytes']}." + ) + return vendored + + +def _artifact_files_to_vendor(source_dir: Path) -> list[str]: + relatives = ["manifest.json", "consumer_facts.jsonl"] + profiles_dir = source_dir / "profiles" + if profiles_dir.is_dir(): + relatives.extend( + f"profiles/{profile.name}" + for profile in sorted(profiles_dir.glob("*.json")) + ) + if (source_dir / "coverage.json").is_file(): + relatives.append("coverage.json") + return relatives + + def _sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as file: diff --git a/packages/populace-build/src/populace/build/ledger_schema.py b/packages/populace-build/src/populace/build/ledger_schema.py new file mode 100644 index 00000000..3fb38086 --- /dev/null +++ b/packages/populace-build/src/populace/build/ledger_schema.py @@ -0,0 +1,268 @@ +"""Deterministic, stdlib-only validation of Ledger consumer fact rows. + +Populace pins the PolicyEngine Ledger consumer-fact contract by vendoring the +published JSON Schema (``schemas/consumer_fact.v1.schema.json``) and validating +every consumed row against it before a fact can resolve a calibration target +(finding #8: nominal row validation). Structurally incomplete or mistyped rows +fail at load with a precise field path rather than silently compiling into a +wrong target. + +Like the rest of Populace's Ledger consumption this module is stdlib-only and +does not import the Ledger implementation package. It implements exactly the +JSON Schema (draft 2020-12) constructs the vendored schema uses --- ``type``, +``const``, ``enum``, ``pattern``, ``required``, ``properties``, +``additionalProperties`` (boolean or subschema), ``items``, and ``$ref`` into +``$defs`` --- and raises on any construct it does not implement, so a future +schema revision that introduces an unhandled keyword fails loudly at import +instead of silently passing rows. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import re +from importlib.resources import files +from typing import Any + +__all__ = [ + "CONSUMER_FACT_SCHEMA_SHA256", + "CONSUMER_FACT_SCHEMA_VERSION", + "validate_consumer_fact_row", +] + +CONSUMER_FACT_SCHEMA_VERSION = "ledger.consumer_fact.v1" + +_SCHEMA_BYTES = ( + files("populace.build.schemas") + .joinpath("consumer_fact.v1.schema.json") + .read_bytes() +) +CONSUMER_FACT_SCHEMA_SHA256 = hashlib.sha256(_SCHEMA_BYTES).hexdigest() +_SCHEMA: dict[str, Any] = json.loads(_SCHEMA_BYTES) +_DEFS: dict[str, Any] = _SCHEMA.get("$defs", {}) + +# JSON Schema keywords this validator implements. Any other keyword appearing in +# the vendored schema is a construct we do not enforce; we raise on it so the +# pin cannot silently weaken when the schema evolves. +_SUPPORTED_KEYWORDS = frozenset( + { + "type", + "const", + "enum", + "pattern", + "required", + "properties", + "additionalProperties", + "items", + "$ref", + } +) +# Annotation / structural keywords that carry no row-level validation semantics. +_ANNOTATION_KEYWORDS = frozenset( + {"$schema", "$id", "$defs", "title", "description"} +) + +_TYPE_CHECKS = { + "object": lambda value: isinstance(value, dict), + "array": lambda value: isinstance(value, list), + "string": lambda value: isinstance(value, str), + # JSON has one number type; booleans are a distinct JSON type and must not + # satisfy integer/number even though Python bool subclasses int. + "integer": lambda value: isinstance(value, int) and not isinstance(value, bool), + "number": lambda value: isinstance(value, (int, float)) + and not isinstance(value, bool), + "boolean": lambda value: isinstance(value, bool), + "null": lambda value: value is None, +} + + +class _UnsupportedSchemaError(RuntimeError): + """The vendored schema uses a construct this validator does not implement.""" + + +def _assert_supported(schema: Any) -> None: + """Walk the schema once at import so unhandled keywords fail loudly.""" + if not isinstance(schema, dict): + raise _UnsupportedSchemaError( + f"Consumer-fact schema node must be an object, got {type(schema).__name__}." + ) + for keyword in schema: + if keyword in _SUPPORTED_KEYWORDS or keyword in _ANNOTATION_KEYWORDS: + continue + raise _UnsupportedSchemaError( + f"Consumer-fact schema uses unimplemented keyword {keyword!r}; the " + "vendored stdlib validator must be extended to enforce it before " + "this schema revision can be pinned." + ) + type_value = schema.get("type") + if type_value is not None: + types = type_value if isinstance(type_value, list) else [type_value] + for name in types: + if name not in _TYPE_CHECKS: + raise _UnsupportedSchemaError( + f"Consumer-fact schema uses unimplemented type {name!r}." + ) + for subschema in schema.get("properties", {}).values(): + _assert_supported(subschema) + additional = schema.get("additionalProperties") + if isinstance(additional, dict): + _assert_supported(additional) + elif additional is not None and not isinstance(additional, bool): + raise _UnsupportedSchemaError( + "Consumer-fact schema additionalProperties must be a boolean or a " + f"subschema, got {type(additional).__name__}." + ) + items = schema.get("items") + if isinstance(items, dict): + _assert_supported(items) + elif items is not None: + raise _UnsupportedSchemaError( + "Consumer-fact schema items must be a single subschema, got " + f"{type(items).__name__}." + ) + ref = schema.get("$ref") + if ref is not None and not (isinstance(ref, str) and ref.startswith("#/$defs/")): + raise _UnsupportedSchemaError( + f"Consumer-fact schema $ref must target #/$defs/, got {ref!r}." + ) + + +for _definition in _DEFS.values(): + _assert_supported(_definition) +_assert_supported(_SCHEMA) + + +def _resolve_ref(ref: str) -> dict[str, Any]: + name = ref[len("#/$defs/") :] + target = _DEFS.get(name) + if not isinstance(target, dict): + raise _UnsupportedSchemaError( + f"Consumer-fact schema $ref {ref!r} does not resolve to a definition." + ) + return target + + +def _type_names(schema: dict[str, Any]) -> str: + type_value = schema["type"] + types = type_value if isinstance(type_value, list) else [type_value] + return " | ".join(str(name) for name in types) + + +def _matches_type(value: Any, schema: dict[str, Any]) -> bool: + type_value = schema["type"] + types = type_value if isinstance(type_value, list) else [type_value] + return any(_TYPE_CHECKS[name](value) for name in types) + + +def _validate(value: Any, schema: dict[str, Any], path: str, *, prefix: str) -> None: + ref = schema.get("$ref") + if ref is not None: + # The vendored schema only uses $ref as a sole-key subschema; a sibling + # validation keyword would be a combination we do not enforce. + siblings = set(schema) - {"$ref"} - _ANNOTATION_KEYWORDS + if siblings: + raise _UnsupportedSchemaError( + f"Consumer-fact schema $ref combined with unimplemented keywords " + f"{sorted(siblings)!r}." + ) + _validate(value, _resolve_ref(ref), path, prefix=prefix) + return + + where = f"{prefix} field {path!r}" if path else prefix + + # A non-finite float (NaN / Infinity / -Infinity) satisfies a bare + # ``number`` type check but is not a valid contract value (finding #7): + # ``json.loads`` accepts the JS constants by default, so reject them on + # every numeric regardless of where they appear in the row. + if isinstance(value, float) and not math.isfinite(value): + raise ValueError(f"{where}: number {value!r} is not finite.") + + if "type" in schema and not _matches_type(value, schema): + raise ValueError( + f"{where}: expected type {_type_names(schema)}, got " + f"{_json_type_name(value)}." + ) + if "const" in schema and value != schema["const"]: + raise ValueError( + f"{where}: expected constant {schema['const']!r}, got {value!r}." + ) + if "enum" in schema and value not in schema["enum"]: + raise ValueError( + f"{where}: {value!r} is not one of {schema['enum']!r}." + ) + if "pattern" in schema and isinstance(value, str): + if re.search(schema["pattern"], value) is None: + raise ValueError( + f"{where}: {value!r} does not match pattern {schema['pattern']!r}." + ) + + if isinstance(value, dict) and ( + "properties" in schema or "required" in schema or "additionalProperties" in schema + ): + _validate_object(value, schema, path, prefix=prefix) + if isinstance(value, list) and "items" in schema: + for index, item in enumerate(value): + item_path = f"{path}[{index}]" if path else f"[{index}]" + _validate(item, schema["items"], item_path, prefix=prefix) + + +def _validate_object( + value: dict[str, Any], schema: dict[str, Any], path: str, *, prefix: str +) -> None: + properties = schema.get("properties", {}) + for required in schema.get("required", ()): + if required not in value: + child = f"{path}.{required}" if path else required + raise ValueError( + f"{prefix} field {child!r} is required but missing." + ) + additional = schema.get("additionalProperties", True) + for key, child_value in value.items(): + child = f"{path}.{key}" if path else key + if key in properties: + _validate(child_value, properties[key], child, prefix=prefix) + elif additional is False: + raise ValueError( + f"{prefix} field {child!r} is not an allowed property." + ) + elif isinstance(additional, dict): + _validate(child_value, additional, child, prefix=prefix) + + +def _json_type_name(value: Any) -> str: + if isinstance(value, bool): + return "boolean" + if isinstance(value, int): + return "integer" + if isinstance(value, float): + return "number" + if isinstance(value, str): + return "string" + if isinstance(value, list): + return "array" + if isinstance(value, dict): + return "object" + if value is None: + return "null" + return type(value).__name__ + + +def validate_consumer_fact_row( + row: Any, + *, + line_number: int, + source: str, +) -> None: + """Validate one consumer fact row against the pinned contract schema. + + Raises ``ValueError`` with a precise field path when the row is not a valid + ``ledger.consumer_fact.v1`` row. + """ + prefix = f"{source} line {line_number}" + if not isinstance(row, dict): + raise ValueError( + f"{prefix}: expected a JSON object, got {_json_type_name(row)}." + ) + _validate(row, _SCHEMA, "", prefix=prefix) diff --git a/packages/populace-build/src/populace/build/ledger_targets.py b/packages/populace-build/src/populace/build/ledger_targets.py index d6c94158..e3165d64 100644 --- a/packages/populace-build/src/populace/build/ledger_targets.py +++ b/packages/populace-build/src/populace/build/ledger_targets.py @@ -42,6 +42,11 @@ class LedgerTargetMapping: entity_by_ledger_entity: Mapping[str, str] = field(default_factory=dict) family_by_source_name: Mapping[str, str] = field(default_factory=dict) family_by_concept: Mapping[str, str] = field(default_factory=dict) + # Model measure (or fact concept) -> the unit its facts must carry. A fact + # that compiles to a measure with no declared expected unit is a hard error, + # and a fact whose unit disagrees is a hard error: the model boundary gate + # against a thousands-vs-millions scale swap (finding #9). + expected_unit_by_measure: Mapping[str, str] = field(default_factory=dict) default_family: str = "ledger" @@ -66,6 +71,21 @@ class LedgerTargetReference: measure: str | None = None filter: str | None = None period: int | str | None = None + # Typed-period selection (finding #10): when set, only facts whose + # period.type matches are eligible; when unset and candidates span multiple + # period types, compilation fails closed rather than silently mixing series. + period_type: str | None = None + # Unit gate (finding #9): the exact unit the resolved fact must carry, and a + # scale applied to the observed value to reach the model's numeric scale. + # Required at compile time (a None expected_unit is a hard error). + expected_unit: str | None = None + value_scale: float = 1.0 + # A non-identity value_scale is a rescale of the observed fact and must be + # reviewed: production references only apply a scale they explicitly declare. + # An undeclared value_scale other than 1.0 is a hard error at compile time so + # an unreviewed thousands/millions rescale cannot ride in on a default + # (finding #9). + value_scale_declared: bool = False source: str | None = None family: str = "ledger" signed: bool = False @@ -100,6 +120,19 @@ def __post_init__(self) -> None: f"LedgerTargetReference {self.name!r}: ledger_selector must be a " f"mapping, got {type(self.ledger_selector).__name__}." ) + try: + value_scale = float(self.value_scale) + except (TypeError, ValueError) as exc: + raise ValueError( + f"LedgerTargetReference {self.name!r}: value_scale must be " + f"numeric, got {self.value_scale!r}." + ) from exc + if not math.isfinite(value_scale) or value_scale == 0: + raise ValueError( + f"LedgerTargetReference {self.name!r}: value_scale must be a " + f"finite non-zero number, got {self.value_scale!r}." + ) + object.__setattr__(self, "value_scale", value_scale) if not self.entity: raise ValueError( f"LedgerTargetReference {self.name!r}: entity must be non-empty." @@ -535,6 +568,48 @@ def target_spec_from_ledger_reference( "count-like facts must be represented as sums of prepared indicator " "columns." ) + + # The unit/scale gate runs after the structural checks so a fact that + # cannot compile at all (missing value, unsupported aggregation) fails on + # that cause rather than on a missing declaration (finding #9). + if reference.expected_unit is None: + raise ValueError( + f"Ledger target reference {reference.name!r}: expected_unit is " + "required; every reference must declare the unit its fact carries " + "so a thousands-vs-millions scale swap fails at the model boundary " + "(finding #9)." + ) + if reference.value_scale != 1.0 and not reference.value_scale_declared: + raise ValueError( + f"Ledger target reference {reference.name!r}: value_scale " + f"{reference.value_scale!r} rescales the observed fact but is not " + "declared; set value_scale_declared=True to authorize a reviewed " + "non-identity scale (an undeclared scale must not silently reach the " + "model, finding #9)." + ) + + fact_unit = _measure_unit(fact) + if fact_unit != reference.expected_unit: + raise ValueError( + f"Ledger target reference {reference.name!r}: resolved fact unit " + f"{fact_unit!r} does not match the declared expected_unit " + f"{reference.expected_unit!r} (the thousands-vs-millions gate)." + ) + if reference.period_type is not None: + fact_period_type = _str_at(fact, "period", "type") + if fact_period_type != reference.period_type: + raise ValueError( + f"Ledger target reference {reference.name!r}: resolved fact " + f"period type {fact_period_type!r} does not match the declared " + f"period_type {reference.period_type!r}." + ) + scaled_value = numeric_value * reference.value_scale + if not math.isfinite(scaled_value): + raise ValueError( + f"Ledger fact for {reference.name!r} is non-finite after applying " + f"value_scale {reference.value_scale!r}: {scaled_value!r}." + ) + period = ( reference.period if reference.period is not None @@ -547,7 +622,7 @@ def target_spec_from_ledger_reference( name=reference.name, entity=reference.entity, measure=reference.measure, - value=numeric_value, + value=scaled_value, filter=reference.filter, period=period, se=reference.se, @@ -695,6 +770,30 @@ def target_spec_from_ledger_fact( if numeric_value < 0 and not _allows_signed_target(fact, mapping): raise _unsupported("missing_signed_target_mapping", fact) + # Unit gate at the model boundary (finding #9): a fact that is about to + # compile into a target must have a declared expected unit for its measure + # (or concept), and the fact's own unit must match it exactly. Both a + # missing declaration and a mismatch are hard errors, not soft "unsupported" + # skips, so a thousands-vs-millions swap cannot pass silently. + expected_unit = _expected_unit_for_measure( + mapping, measure, _measure_concepts(fact) + ) + if not expected_unit: + raise ValueError( + f"Ledger fact {identifier!r} compiles to model measure {measure!r} " + "but LedgerTargetMapping declares no expected unit for it; add an " + "expected_unit_by_measure entry (the thousands-vs-millions gate)." + ) + fact_unit = _measure_unit(fact) + if fact_unit != expected_unit: + raise ValueError( + f"Ledger fact {identifier!r} unit {fact_unit!r} does not match the " + f"expected unit {expected_unit!r} declared for model measure " + f"{measure!r} (the thousands-vs-millions gate)." + ) + + metadata = _ledger_metadata(fact, fact_key=fact_key) + metadata["ledger_expected_unit"] = expected_unit try: return TargetSpec( name=identifier, @@ -706,7 +805,7 @@ def target_spec_from_ledger_fact( source=_source_citation(fact), family=family, signed=numeric_value < 0, - metadata=_ledger_metadata(fact, fact_key=fact_key), + metadata=metadata, ) except (TypeError, ValueError) as exc: raise _unsupported(f"invalid_target_spec:{type(exc).__name__}", fact) from exc @@ -800,6 +899,7 @@ def _resolve_reference_fact( for fact in fact_index.facts if _fact_matches_selector(fact, reference.ledger_selector) ] + matches = _selector_matches_for_period_type(reference, matches) eligible_matches = _eligible_selector_matches(reference, matches) if len(eligible_matches) == 1: return eligible_matches[0] @@ -833,6 +933,35 @@ def _resolve_reference_fact( ) +def _selector_matches_for_period_type( + reference: LedgerTargetReference, + matches: list[object], +) -> list[object]: + """Constrain selector matches to a single, unambiguous period type. + + When the reference declares a ``period_type`` only facts of that type are + eligible. When it does not and the matched facts span more than one period + type, compilation fails closed (finding #10) rather than silently letting + "latest wins" pick across tax-year / calendar-year / fiscal-year series. + """ + if reference.period_type is not None: + return [ + fact + for fact in matches + if _str_at(fact, "period", "type") == reference.period_type + ] + period_types = {_str_at(fact, "period", "type") for fact in matches} + if len(period_types) > 1: + raise ValueError( + f"Ledger target reference {reference.name!r} matched Ledger facts " + f"with multiple period types {sorted(period_types)!r} for selector " + f"{dict(reference.ledger_selector)!r} but declares no period_type; " + "set period_type to disambiguate (tax-year vs calendar-year vs " + "fiscal-year facts are distinct series and must not be merged)." + ) + return matches + + def _eligible_selector_matches( reference: LedgerTargetReference, matches: list[object], @@ -884,6 +1013,10 @@ def _selector_period_invariant_key(fact: object) -> tuple[str, ...]: _str_at(fact, "geography", "id"), _str_at(fact, "entity", "name"), _str_at(fact, "aggregation", "method"), + # Period TYPE is part of the series identity (finding #10): tax-year, + # calendar-year, and fiscal-year facts for the same measure are distinct + # series and must never collapse into one "latest" candidate set. + _str_at(fact, "period", "type"), _normalized_record_set_id(_str_at(fact, "layout", "record_set_id")), _str_at(fact, "layout", "groupby_dimension"), _str_at(fact, "layout", "groupby_value_id"), @@ -896,21 +1029,53 @@ def _selector_period_invariant_key(fact: object) -> tuple[str, ...]: def _normalized_record_set_id(record_set_id: str) -> str: if not record_set_id: return "" + # Replace period-like tokens with a period-TYPE marker rather than dropping + # them: erasing the year lets same-series facts across years share one + # invariant key, but keeping the ty/cy/fy/month marker stops tax-year, + # calendar-year, and fiscal-year record sets from collapsing into one + # "latest" series (finding #10). return ".".join( - part for part in record_set_id.split(".") if not _is_period_token(part) + _period_token_type(part) or part for part in record_set_id.split(".") ) -def _is_period_token(value: str) -> bool: +def _period_token_type(value: str) -> str | None: + """Return a period-type marker for a period-like token, else ``None``. + + ``ty2024`` -> ``ty``, ``cy2024`` -> ``cy``, ``fy2024`` -> ``fy``, + ``month2024_01`` / ``2024_01`` -> ``month``, a bare ``2024`` -> ``year``. A + non-period token returns ``None`` and is preserved verbatim by callers. + """ normalized = value.lower().replace("-", "_") if normalized.startswith("month"): - normalized = normalized[len("month") :] - parts = normalized.split("_", maxsplit=1) - if len(parts) == 2 and all(part.isdigit() for part in parts): - return len(parts[0]) == 4 and len(parts[1]) in {1, 2} - if normalized[:2] in {"ty", "cy", "fy"}: - normalized = normalized[2:] - return normalized.isdigit() and len(normalized) == 4 + rest = normalized[len("month") :] + if _is_year_month(rest) or _is_year(rest): + return "month" + return None + if normalized[:2] in {"ty", "cy", "fy"} and _is_year(normalized[2:]): + return normalized[:2] + if _is_year_month(normalized): + return "month" + if _is_year(normalized): + return "year" + return None + + +def _is_year(value: str) -> bool: + return value.isdigit() and len(value) == 4 + + +def _is_year_month(value: str) -> bool: + parts = value.split("_", maxsplit=1) + if len(parts) != 2: + return False + year, month = parts + return ( + year.isdigit() + and len(year) == 4 + and month.isdigit() + and len(month) in {1, 2} + ) def _period_key(fact: object) -> tuple[int, int, str]: @@ -960,6 +1125,13 @@ def _not_after_target_period( def _reference_metadata(reference: LedgerTargetReference) -> dict[str, str]: metadata = dict(reference.metadata) metadata["ledger_value_operation"] = reference.value_operation + # The unit gate and scale, recorded next to ledger_measure_unit for audit + # (finding #9); the reference's declared period type, when set (finding #10). + if reference.expected_unit: + metadata["ledger_expected_unit"] = reference.expected_unit + metadata["ledger_value_scale"] = _format_float(reference.value_scale) + if reference.period_type is not None: + metadata["ledger_reference_period_type"] = reference.period_type for key, value in sorted(reference.ledger_selector.items()): if isinstance(value, Mapping): continue @@ -974,6 +1146,23 @@ def _reference_metadata(reference: LedgerTargetReference) -> dict[str, str]: return metadata +def _measure_unit(fact: object) -> str: + return _str_at(fact, "measure", "unit") or _str_at(fact, "observed_measure", "unit") + + +def _expected_unit_for_measure( + mapping: LedgerTargetMapping, + measure: str, + concepts: tuple[str, ...], +) -> str | None: + if measure in mapping.expected_unit_by_measure: + return mapping.expected_unit_by_measure[measure] + for concept in concepts: + if concept in mapping.expected_unit_by_measure: + return mapping.expected_unit_by_measure[concept] + return None + + def _at(obj: object, *path: str) -> Any: current = obj for key in path: diff --git a/packages/populace-build/src/populace/build/schemas/consumer_fact.v1.schema.json b/packages/populace-build/src/populace/build/schemas/consumer_fact.v1.schema.json new file mode 100644 index 00000000..37a29e12 --- /dev/null +++ b/packages/populace-build/src/populace/build/schemas/consumer_fact.v1.schema.json @@ -0,0 +1,413 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://policyengine.org/ledger/schemas/consumer_fact.v1.schema.json", + "title": "Ledger consumer fact contract row", + "type": "object", + "additionalProperties": false, + "required": [ + "schema_version", + "aggregate_fact_key", + "semantic_fact_key", + "legacy_fact_key", + "source_release_key", + "source_series_key", + "observed_measure_key", + "dimension_set_key", + "universe_constraint_set_key", + "value", + "value_type", + "assertion", + "period", + "geography", + "entity", + "aggregation", + "observed_measure", + "dimensions", + "universe_constraints", + "source", + "lineage" + ], + "properties": { + "schema_version": { + "const": "ledger.consumer_fact.v1" + }, + "aggregate_fact_key": { + "type": "string", + "pattern": "^ledger\\.aggregate_fact\\.v2:[0-9a-f]{24}$" + }, + "semantic_fact_key": { + "type": "string", + "pattern": "^ledger\\.semantic_fact\\.v2:[0-9a-f]{24}$" + }, + "legacy_fact_key": { + "type": "string", + "pattern": "^ledger\\.fact\\.v1:[0-9a-f]{24}$" + }, + "source_release_key": { + "type": "string", + "pattern": "^ledger\\.source_release\\.v2:[0-9a-f]{24}$" + }, + "source_series_key": { + "type": "string", + "pattern": "^ledger\\.source_series\\.v2:[0-9a-f]{24}$" + }, + "observed_measure_key": { + "type": "string", + "pattern": "^ledger\\.observed_measure\\.v2:[0-9a-f]{24}$" + }, + "dimension_set_key": { + "type": "string", + "pattern": "^ledger\\.dimension_set\\.v2:[0-9a-f]{24}$" + }, + "universe_constraint_set_key": { + "type": "string", + "pattern": "^ledger\\.universe_constraint_set\\.v2:[0-9a-f]{24}$" + }, + "value": { + "type": [ + "integer", + "number", + "string", + "boolean" + ] + }, + "value_type": { + "enum": [ + "integer", + "number", + "decimal", + "string", + "boolean" + ] + }, + "period": { + "type": "object", + "additionalProperties": false, + "required": [ + "type", + "value" + ], + "properties": { + "type": { + "type": "string" + }, + "value": { + "type": [ + "integer", + "string" + ] + } + } + }, + "geography": { + "type": "object", + "additionalProperties": false, + "required": [ + "level", + "id" + ], + "properties": { + "level": { + "type": "string" + }, + "id": { + "type": "string" + }, + "vintage": { + "type": "string" + } + } + }, + "entity": { + "type": "object", + "additionalProperties": false, + "required": [ + "name" + ], + "properties": { + "name": { + "type": "string" + }, + "role": { + "type": "string" + } + } + }, + "aggregation": { + "type": "object", + "additionalProperties": false, + "required": [ + "method" + ], + "properties": { + "method": { + "type": "string" + }, + "denominator": { + "type": "string" + } + } + }, + "observed_measure": { + "type": "object", + "additionalProperties": false, + "required": [ + "source_name", + "source_table", + "source_measure_id", + "source_concept", + "unit" + ], + "properties": { + "source_name": { + "type": "string" + }, + "source_table": { + "type": "string" + }, + "source_measure_id": { + "type": "string" + }, + "source_concept": { + "type": "string" + }, + "unit": { + "type": "string" + } + } + }, + "dimensions": { + "type": "object", + "additionalProperties": { + "type": [ + "integer", + "number", + "string", + "boolean" + ] + } + }, + "universe_constraints": { + "type": "object", + "additionalProperties": false, + "required": [ + "domain" + ], + "properties": { + "domain": { + "type": "string" + }, + "constraints": { + "type": "array", + "items": { + "$ref": "#/$defs/constraint" + } + } + } + }, + "source": { + "type": "object", + "additionalProperties": true, + "required": [ + "source_name", + "source_table", + "source_file", + "vintage", + "extracted_at", + "extraction_method", + "source_sha256", + "source_size_bytes", + "raw_r2_uri" + ], + "properties": { + "source_name": { + "type": "string" + }, + "source_table": { + "type": "string" + }, + "source_file": { + "type": "string" + }, + "url": { + "type": "string" + }, + "vintage": { + "type": "string" + }, + "extracted_at": { + "type": "string" + }, + "extraction_method": { + "type": "string" + }, + "method_notes": { + "type": "string" + }, + "source_sha256": { + "type": "string" + }, + "source_size_bytes": { + "type": "integer" + }, + "raw_r2_bucket": { + "type": "string" + }, + "raw_r2_key": { + "type": "string" + }, + "raw_r2_uri": { + "type": "string" + } + } + }, + "lineage": { + "type": "object", + "additionalProperties": false, + "required": [ + "source_record_id", + "source_cell_keys" + ], + "properties": { + "source_record_id": { + "type": "string" + }, + "source_cell_keys": { + "type": "array", + "items": { + "type": "string" + } + }, + "source_row_keys": { + "type": "array", + "items": { + "type": "string" + } + } + } + }, + "layout": { + "type": "object", + "additionalProperties": true + }, + "label": { + "type": "string" + }, + "concept_alignment": { + "type": "object", + "additionalProperties": false, + "required": [ + "concept_alignment_key", + "source_concept", + "canonical_concept", + "relation", + "authority" + ], + "properties": { + "concept_alignment_key": { + "type": "string", + "pattern": "^ledger\\.concept_alignment\\.v2:[0-9a-f]{24}$" + }, + "source_concept": { + "type": "string" + }, + "canonical_concept": { + "type": "string" + }, + "relation": { + "type": "string" + }, + "authority": { + "type": "string" + }, + "evidence_url": { + "type": "string" + }, + "evidence_notes": { + "type": "string" + }, + "legal_vintage": { + "type": "string" + } + } + }, + "assertion": { + "type": "string", + "enum": [ + "observation", + "source_projection" + ], + "description": "Who asserted the value: a publisher-measured outcome (observation) or the publisher's own forward-looking estimate (source_projection). PolicyEngine-computed values are never consumer facts." + }, + "period_coverage": { + "type": "object", + "additionalProperties": false, + "description": "Non-identity provenance for the fact's reference period: coverage dates, basis, the publisher's period label, and accounting basis.", + "properties": { + "start_date": { + "type": "string" + }, + "end_date": { + "type": "string" + }, + "basis": { + "type": "string", + "enum": [ + "calendar", + "tax", + "fiscal", + "survey_reference", + "projection_horizon" + ] + }, + "source_period_label": { + "type": "string" + }, + "accounting_basis": { + "type": "string", + "enum": [ + "cash", + "accrual" + ] + }, + "notes": { + "type": "string" + } + } + } + }, + "$defs": { + "constraint": { + "type": "object", + "additionalProperties": false, + "required": [ + "variable", + "operator", + "value", + "role" + ], + "properties": { + "variable": { + "type": "string" + }, + "operator": { + "type": "string" + }, + "value": { + "type": [ + "integer", + "number", + "string", + "boolean" + ] + }, + "unit": { + "type": "string" + }, + "role": { + "type": "string" + } + } + } + } +} diff --git a/packages/populace-build/src/populace/build/us/fiscal_target_references.json b/packages/populace-build/src/populace/build/us/fiscal_target_references.json index 55402ddf..33034484 100644 --- a/packages/populace-build/src/populace/build/us/fiscal_target_references.json +++ b/packages/populace-build/src/populace/build/us/fiscal_target_references.json @@ -113,6 +113,7 @@ "target_references": [ { "entity": "household", + "expected_unit": "usd", "family": "jct", "ledger_source_record_id": "jct.tax_expenditures.cy2024.salt_deduction.revenue_loss", "measure": "jct.tax_expenditures.cy2024.salt_deduction.revenue_loss", @@ -129,6 +130,7 @@ }, { "entity": "household", + "expected_unit": "usd", "family": "jct", "ledger_source_record_id": "jct.tax_expenditures.cy2024.medical_expense_deduction.revenue_loss", "measure": "jct.tax_expenditures.cy2024.medical_expense_deduction.revenue_loss", @@ -145,6 +147,7 @@ }, { "entity": "household", + "expected_unit": "usd", "family": "jct", "ledger_source_record_id": "jct.tax_expenditures.cy2024.charitable_deduction.revenue_loss", "measure": "jct.tax_expenditures.cy2024.charitable_deduction.revenue_loss", @@ -161,6 +164,7 @@ }, { "entity": "household", + "expected_unit": "usd", "family": "jct", "ledger_source_record_id": "jct.tax_expenditures.cy2024.deductible_mortgage_interest.revenue_loss", "measure": "jct.tax_expenditures.cy2024.deductible_mortgage_interest.revenue_loss", @@ -177,6 +181,7 @@ }, { "entity": "household", + "expected_unit": "usd", "family": "jct", "ledger_source_record_id": "jct.tax_expenditures.cy2024.qualified_business_income_deduction.revenue_loss", "measure": "jct.tax_expenditures.cy2024.qualified_business_income_deduction.revenue_loss", diff --git a/packages/populace-build/src/populace/build/us_runtime/fiscal_targets.py b/packages/populace-build/src/populace/build/us_runtime/fiscal_targets.py index 253e5e14..7348d274 100644 --- a/packages/populace-build/src/populace/build/us_runtime/fiscal_targets.py +++ b/packages/populace-build/src/populace/build/us_runtime/fiscal_targets.py @@ -509,6 +509,66 @@ } +def _build_us_fiscal_expected_units() -> dict[tuple[str, str], str]: + """Committed expected input unit for every production fiscal measure. + + Keyed by ``(source_name, measure_id)`` and populated from the checked-in + measure maps, never from an incoming fact. The dynamic reference builders + read the expected unit from here instead of echoing the fact's own declared + unit, so a fact whose unit drifted (for example ``usd`` -> ``usd_thousands``) + fails the model-boundary unit gate rather than validating against itself + (finding #9). The values are the denomination the published fact stream + carries for each measure: dollar measures in ``usd`` and the CMS Medicaid/ + CHIP enrollment counts in ``people``. A production measure with no committed + unit here is a hard error, never a silent default. + """ + units: dict[tuple[str, str], str] = {} + for measure_id in (*SOI_AMOUNT_MEASURE_VARIABLES, *SOI_RETURN_MEASURE_VARIABLES): + units[("irs_soi", measure_id)] = "usd" + units[("census_acs", "population")] = "usd" + units[("census_pep", "population")] = "usd" + units[("census_stc", "collections")] = "usd" + for source_name, measure_id, _group in DIRECT_LEDGER_TARGETS: + units[(source_name, measure_id)] = "usd" + # Count-based indicators carry a count denomination in the fact stream, not + # dollars: CMS Medicaid/CHIP enrollment as ``people``, USDA SNAP average + # monthly caseload as ``count``. Every other indicator is a dollar measure. + for source_name, measure_id in INDICATOR_LEDGER_TARGETS: + if source_name == "cms_medicaid": + units[(source_name, measure_id)] = "people" + elif source_name == "usda_snap": + units[(source_name, measure_id)] = "count" + else: + units[(source_name, measure_id)] = "usd" + return units + + +US_FISCAL_MEASURE_EXPECTED_UNITS: dict[tuple[str, str], str] = ( + _build_us_fiscal_expected_units() +) + + +def _expected_unit_for_production_measure(fact: object) -> str: + """Look up the committed expected unit for a fact's production measure. + + The lookup key is the fact's ``(source_name, measure_id)`` identity, but the + returned unit is committed source-of-truth: it does not come from the fact's + own unit field, so a drifted-unit fact cannot authorize itself. An + undeclared production measure fails closed (finding #9). + """ + source_name = _source_name(fact) + measure_id = _measure_id(fact) + unit = US_FISCAL_MEASURE_EXPECTED_UNITS.get((source_name, measure_id)) + if not unit: + raise ValueError( + "No committed expected unit for production fiscal measure " + f"({source_name!r}, {measure_id!r}); add it to " + "US_FISCAL_MEASURE_EXPECTED_UNITS rather than echoing the fact's own " + "unit into its gate (finding #9)." + ) + return unit + + US_FISCAL_TARGET_SUPPORT_EXCLUSIONS: dict[str, str] = { "census_stc.fy2023.individual_income_tax_collections.tn.t40.collections": ( "Tennessee has no modeled 2024 state individual income tax support in " @@ -1901,8 +1961,46 @@ def _model_target_key(fact: object) -> tuple[str, ...] | None: def _normalized_record_set_id(record_set_id: str) -> str: if not record_set_id: return "" + # Preserve the period TYPE while erasing the year so tax-year, calendar-year, + # and fiscal-year record sets stay distinct in _dynamic_target_key's + # latest-fact selection (finding #10); mirrors ledger_targets._normalized_ + # record_set_id. return ".".join( - part for part in record_set_id.split(".") if not _is_period_token(part) + _period_token_type(part) or part for part in record_set_id.split(".") + ) + + +def _period_token_type(value: str) -> str | None: + """Return a period-type marker for a period-like token, else ``None``.""" + normalized = value.lower().replace("-", "_") + if normalized.startswith("month"): + rest = normalized[len("month") :] + if _is_year_month(rest) or _is_year(rest): + return "month" + return None + if normalized[:2] in {"ty", "cy", "fy"} and _is_year(normalized[2:]): + return normalized[:2] + if _is_year_month(normalized): + return "month" + if _is_year(normalized): + return "year" + return None + + +def _is_year(value: str) -> bool: + return value.isdigit() and len(value) == 4 + + +def _is_year_month(value: str) -> bool: + parts = value.split("_", maxsplit=1) + if len(parts) != 2: + return False + year, month = parts + return ( + year.isdigit() + and len(year) == 4 + and month.isdigit() + and len(month) in {1, 2} ) @@ -1938,18 +2036,6 @@ def _is_soi_cd_premium_tax_credit_amount_conflict( ) -def _is_period_token(value: str) -> bool: - normalized = value.lower().replace("-", "_") - if normalized.startswith("month"): - normalized = normalized[len("month") :] - parts = normalized.split("_", maxsplit=1) - if len(parts) == 2 and all(part.isdigit() for part in parts): - return len(parts[0]) == 4 and len(parts[1]) in {1, 2} - if normalized[:2] in {"ty", "cy", "fy"}: - normalized = normalized[2:] - return normalized.isdigit() and len(normalized) == 4 - - def _period_key(fact: object) -> tuple[int, int, str]: return _period_key_from_value(_period_value(fact)) @@ -2160,6 +2246,10 @@ def _soi_reference_from_fact( entity="household", measure=source_record_id, period=target_period, + # The expected unit is committed truth keyed by the measure, not the + # fact's own unit; a fact whose unit drifted then fails the gate rather + # than validating against itself (finding #9). + expected_unit=_expected_unit_for_production_measure(fact), family="irs_soi", signed=_numeric_value(fact) < 0, metadata=metadata, @@ -2256,6 +2346,7 @@ def _state_income_tax_reference_from_fact( entity="household", measure=source_record_id, period=target_period, + expected_unit=_expected_unit_for_production_measure(fact), family="state_income_tax", metadata={ "source_measure_id": "collections", @@ -2325,6 +2416,7 @@ def _population_age_reference_from_fact( entity="household", measure=source_record_id, period=target_period, + expected_unit=_expected_unit_for_production_measure(fact), family="census_population", metadata=metadata, ) @@ -2383,6 +2475,7 @@ def _direct_reference_from_fact( entity="household", measure=source_record_id, period=target_period, + expected_unit=_expected_unit_for_production_measure(fact), family=family, signed=_numeric_value(fact) < 0, metadata=metadata, diff --git a/packages/populace-build/tests/test_ledger_artifact.py b/packages/populace-build/tests/test_ledger_artifact.py index 8b36b29c..afe6946c 100644 --- a/packages/populace-build/tests/test_ledger_artifact.py +++ b/packages/populace-build/tests/test_ledger_artifact.py @@ -11,10 +11,70 @@ CONSUMER_ARTIFACT_SCHEMA_VERSION, LedgerConsumerArtifact, load_ledger_consumer_artifact, + vendor_ledger_consumer_artifact, + verify_vendored_ledger_artifact, ) +from populace.build.ledger_schema import CONSUMER_FACT_SCHEMA_SHA256 +# 24 lowercase-hex characters: the identity-key suffix the schema pins. +_HEX = "0123456789abcdef01234567" -def _fact_row(**overrides): + +def _key(prefix: str, tag: str) -> str: + suffix = (hashlib.sha256(tag.encode()).hexdigest())[:24] + return f"ledger.{prefix}.v2:{suffix}" + + +def _schema_complete_fact_row(tag: str = "agi", **overrides): + """A row satisfying every required field of consumer_fact.v1.""" + row = { + "schema_version": "ledger.consumer_fact.v1", + "aggregate_fact_key": _key("aggregate_fact", tag), + "semantic_fact_key": _key("semantic_fact", tag), + "legacy_fact_key": f"ledger.fact.v1:{_HEX}", + "source_release_key": f"ledger.source_release.v2:{_HEX}", + "source_series_key": f"ledger.source_series.v2:{_HEX}", + "observed_measure_key": f"ledger.observed_measure.v2:{_HEX}", + "dimension_set_key": f"ledger.dimension_set.v2:{_HEX}", + "universe_constraint_set_key": f"ledger.universe_constraint_set.v2:{_HEX}", + "value": 100, + "value_type": "integer", + "assertion": "observation", + "period": {"type": "tax_year", "value": 2023}, + "geography": {"level": "country", "id": "0100000US"}, + "entity": {"name": "tax_unit"}, + "aggregation": {"method": "sum"}, + "observed_measure": { + "source_name": "irs_soi", + "source_table": "t", + "source_measure_id": "agi", + "source_concept": "irs_soi.agi", + "unit": "usd", + }, + "dimensions": {"income_range": "all"}, + "universe_constraints": {"domain": "all_individual_income_tax_returns"}, + "source": { + "source_name": "irs_soi", + "source_table": "t", + "source_file": "f.xls", + "vintage": "2023", + "extracted_at": "2024-01-01T00:00:00Z", + "extraction_method": "manual", + "source_sha256": "ab" * 32, + "source_size_bytes": 1024, + "raw_r2_uri": "r2://ledger/f.xls", + }, + "lineage": { + "source_record_id": "irs_soi.ty2023.t.all.agi", + "source_cell_keys": ["ledger.source_cell.v1:cell"], + }, + } + row.update(overrides) + return row + + +def _legacy_fact_row(**overrides): + """A deliberately minimal, pre-schema row (no assertion key).""" row = { "aggregate_fact_key": "ledger.aggregate_fact.v2:abc123", "value": 100, @@ -37,16 +97,29 @@ def _write_facts(path, rows): return hashlib.sha256(path.read_bytes()).hexdigest() -def _write_artifact_dir(tmp_path, rows, *, manifest_overrides=None): +def _write_artifact_dir( + tmp_path, + rows, + *, + manifest_overrides=None, + profile_body=b'{"targets": []}', + profile_trailing_newline=True, + declared_profile_sha=None, +): artifact_dir = tmp_path / "artifact" artifact_dir.mkdir() facts_sha = _write_facts(artifact_dir / "consumer_facts.jsonl", rows) + profiles_dir = artifact_dir / "profiles" + profiles_dir.mkdir() + profile_bytes = profile_body + (b"\n" if profile_trailing_newline else b"") + (profiles_dir / "us_fiscal.json").write_bytes(profile_bytes) + profile_sha = declared_profile_sha or hashlib.sha256(profile_bytes).hexdigest() manifest = { "schema_version": CONSUMER_ARTIFACT_SCHEMA_VERSION, "fact_row_count": len(rows), "facts_sha256": facts_sha, "profiles": { - "us_fiscal": {"sha256": "ab" * 32, "target_count": 3}, + "us_fiscal": {"sha256": profile_sha, "target_count": 3}, }, } manifest.update(manifest_overrides or {}) @@ -58,26 +131,32 @@ def _write_artifact_dir(tmp_path, rows, *, manifest_overrides=None): def test_loads_bare_consumer_facts_file(tmp_path): facts_path = tmp_path / "consumer_facts.jsonl" - facts_sha = _write_facts(facts_path, [_fact_row()]) + facts_sha = _write_facts(facts_path, [_legacy_fact_row()]) - artifact = load_ledger_consumer_artifact(facts_path) + # Bare, pre-schema rows are only accepted with validation off: they predate + # the pinned contract and deliberately omit the assertion key. + artifact = load_ledger_consumer_artifact(facts_path, validate_rows=False) assert isinstance(artifact, LedgerConsumerArtifact) assert artifact.fact_row_count == 1 assert artifact.facts_sha256 == facts_sha assert artifact.manifest is None - # Legacy rows that omit the assertion field pass through untouched: - # readers treat a missing assertion as observation-by-default, but the - # loader must not fabricate the key (structural projection checks would - # then see unlabeled projections as mistyped observations). + assert artifact.rows_validated is False + # Legacy rows that omit the assertion field pass through untouched: readers + # treat a missing assertion as observation-by-default, but the loader must + # not fabricate the key. assert "assertion" not in artifact.facts[0] provenance = artifact.provenance() assert provenance["facts_sha256"] == facts_sha assert provenance["schema_version"] is None + assert provenance["consumer_fact_schema_sha256"] is None def test_loads_artifact_directory_and_records_provenance(tmp_path): - rows = [_fact_row(), _fact_row(assertion="source_projection")] + rows = [ + _schema_complete_fact_row(tag="agi"), + _schema_complete_fact_row(tag="tax", assertion="source_projection"), + ] artifact_dir = _write_artifact_dir(tmp_path, rows) artifact = load_ledger_consumer_artifact(artifact_dir) @@ -85,14 +164,38 @@ def test_loads_artifact_directory_and_records_provenance(tmp_path): assert artifact.fact_row_count == 2 assert artifact.manifest["schema_version"] == CONSUMER_ARTIFACT_SCHEMA_VERSION assert artifact.facts[1]["assertion"] == "source_projection" + assert artifact.rows_validated is True provenance = artifact.provenance() assert provenance["schema_version"] == CONSUMER_ARTIFACT_SCHEMA_VERSION assert provenance["manifest_sha256"] == artifact.manifest_sha256 + assert provenance["consumer_fact_schema_sha256"] == CONSUMER_FACT_SCHEMA_SHA256 assert provenance["profiles"]["us_fiscal"]["target_count"] == 3 + assert provenance["profiles"]["us_fiscal"]["hash_semantics"] == "exact" + assert provenance["path"].endswith("artifact") + + +def test_validate_rows_rejects_structurally_incomplete_row(tmp_path): + incomplete = _schema_complete_fact_row() + del incomplete["entity"] + artifact_dir = _write_artifact_dir(tmp_path, [incomplete]) + + with pytest.raises(ValueError, match="'entity' is required but missing"): + load_ledger_consumer_artifact(artifact_dir) + + +def test_rejects_duplicate_aggregate_fact_key(tmp_path): + rows = [ + _schema_complete_fact_row(tag="dup"), + _schema_complete_fact_row(tag="dup", value=200), + ] + artifact_dir = _write_artifact_dir(tmp_path, rows) + + with pytest.raises(ValueError, match="repeats aggregate_fact_key"): + load_ledger_consumer_artifact(artifact_dir) def test_rejects_tampered_fact_rows(tmp_path): - artifact_dir = _write_artifact_dir(tmp_path, [_fact_row()]) + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) facts_path = artifact_dir / "consumer_facts.jsonl" tampered = json.loads(facts_path.read_text()) tampered["value"] = 999 @@ -103,7 +206,7 @@ def test_rejects_tampered_fact_rows(tmp_path): def test_rejects_facts_pin_mismatch(tmp_path): - artifact_dir = _write_artifact_dir(tmp_path, [_fact_row()]) + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) with pytest.raises(ValueError, match="pinned hash"): load_ledger_consumer_artifact( @@ -113,7 +216,7 @@ def test_rejects_facts_pin_mismatch(tmp_path): def test_rejects_manifest_pin_mismatch_and_bare_manifest_pin(tmp_path): - artifact_dir = _write_artifact_dir(tmp_path, [_fact_row()]) + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) with pytest.raises(ValueError, match="manifest does not match"): load_ledger_consumer_artifact( artifact_dir, @@ -121,7 +224,7 @@ def test_rejects_manifest_pin_mismatch_and_bare_manifest_pin(tmp_path): ) facts_path = tmp_path / "bare.jsonl" - _write_facts(facts_path, [_fact_row()]) + _write_facts(facts_path, [_schema_complete_fact_row()]) with pytest.raises(ValueError, match="no manifest"): load_ledger_consumer_artifact( facts_path, @@ -130,7 +233,7 @@ def test_rejects_manifest_pin_mismatch_and_bare_manifest_pin(tmp_path): def test_matching_pins_pass(tmp_path): - artifact_dir = _write_artifact_dir(tmp_path, [_fact_row()]) + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) unpinned = load_ledger_consumer_artifact(artifact_dir) pinned = load_ledger_consumer_artifact( @@ -141,21 +244,187 @@ def test_matching_pins_pass(tmp_path): assert pinned.facts == unpinned.facts +def test_require_pins_rejects_bare_feed(tmp_path): + facts_path = tmp_path / "bare.jsonl" + _write_facts(facts_path, [_schema_complete_fact_row()]) + + with pytest.raises(ValueError, match="must be a hash-pinned artifact directory"): + load_ledger_consumer_artifact( + facts_path, + expected_facts_sha256="0" * 64, + expected_manifest_sha256="0" * 64, + require_pins=True, + ) + + +def test_require_pins_requires_both_pins(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact(artifact_dir) + + with pytest.raises(ValueError, match="demands both"): + load_ledger_consumer_artifact( + artifact_dir, + expected_facts_sha256=loaded.facts_sha256, + require_pins=True, + ) + with pytest.raises(ValueError, match="demands both"): + load_ledger_consumer_artifact( + artifact_dir, + expected_manifest_sha256=loaded.manifest_sha256, + require_pins=True, + ) + + +def test_require_pins_rejects_wrong_pin(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact(artifact_dir) + + with pytest.raises(ValueError, match="pinned hash"): + load_ledger_consumer_artifact( + artifact_dir, + expected_facts_sha256="0" * 64, + expected_manifest_sha256=loaded.manifest_sha256, + require_pins=True, + ) + + +def test_require_pins_passes_with_matching_pins(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact(artifact_dir) + + pinned = load_ledger_consumer_artifact( + artifact_dir, + expected_facts_sha256=loaded.facts_sha256, + expected_manifest_sha256=loaded.manifest_sha256, + require_pins=True, + ) + assert pinned.require_pins is True + assert pinned.provenance()["require_pins"] is True + assert pinned.provenance()["profiles"]["us_fiscal"]["hash_semantics"] == "exact" + + +def test_require_pins_rejects_missing_profile_file(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact(artifact_dir) + (artifact_dir / "profiles" / "us_fiscal.json").unlink() + + with pytest.raises(ValueError, match="no profile file was found"): + load_ledger_consumer_artifact( + artifact_dir, + expected_facts_sha256=loaded.facts_sha256, + expected_manifest_sha256=loaded.manifest_sha256, + require_pins=True, + ) + + +def test_profile_hash_legacy_pre_newline_semantics(tmp_path): + body = b'{"targets": [1, 2, 3]}' + pre_newline_sha = hashlib.sha256(body).hexdigest() + artifact_dir = _write_artifact_dir( + tmp_path, + [_schema_complete_fact_row()], + profile_body=body, + profile_trailing_newline=True, + declared_profile_sha=pre_newline_sha, + ) + + artifact = load_ledger_consumer_artifact(artifact_dir) + assert ( + artifact.provenance()["profiles"]["us_fiscal"]["hash_semantics"] + == "legacy_pre_newline" + ) + + +def test_profile_hash_mismatch_rejected(tmp_path): + artifact_dir = _write_artifact_dir( + tmp_path, + [_schema_complete_fact_row()], + declared_profile_sha="0" * 64, + ) + + with pytest.raises(ValueError, match="does not match its manifest hash"): + load_ledger_consumer_artifact(artifact_dir) + + +def test_vendor_and_verify_roundtrip(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact( + artifact_dir, + expected_facts_sha256=None, + ) + + dest = tmp_path / "release" / "ledger_artifact" + vendored = vendor_ledger_consumer_artifact( + loaded, + dest, + verified_facts_sha256=loaded.facts_sha256, + verified_manifest_sha256=loaded.manifest_sha256, + ) + + paths = {entry["path"] for entry in vendored["files"]} + assert "manifest.json" in paths + assert "consumer_facts.jsonl" in paths + assert "profiles/us_fiscal.json" in paths + assert vendored["verified_facts_sha256"] == loaded.facts_sha256 + assert vendored["consumer_fact_schema_sha256"] == CONSUMER_FACT_SCHEMA_SHA256 + # The vendored consumer facts reload byte-for-byte to the same content hash. + assert ( + hashlib.sha256((dest / "consumer_facts.jsonl").read_bytes()).hexdigest() + == loaded.facts_sha256 + ) + # A clean vendored copy verifies. + verify_vendored_ledger_artifact(dest) + + +def test_vendor_detects_tampered_copy(tmp_path): + artifact_dir = _write_artifact_dir(tmp_path, [_schema_complete_fact_row()]) + loaded = load_ledger_consumer_artifact(artifact_dir) + dest = tmp_path / "release" / "ledger_artifact" + vendor_ledger_consumer_artifact(loaded, dest) + + tampered = json.loads((dest / "consumer_facts.jsonl").read_text()) + tampered["value"] = 424242 + (dest / "consumer_facts.jsonl").write_text(json.dumps(tampered) + "\n") + + with pytest.raises(ValueError, match="was tampered"): + verify_vendored_ledger_artifact(dest) + + def test_rejects_unknown_assertion_and_schema_version(tmp_path): facts_path = tmp_path / "bad_assertion.jsonl" - _write_facts(facts_path, [_fact_row(assertion="policyengine_aged")]) - with pytest.raises(ValueError, match="unsupported assertion"): + _write_facts(facts_path, [_schema_complete_fact_row(assertion="policyengine_aged")]) + # Validation on (default): the pinned schema's assertion enum rejects it. + with pytest.raises(ValueError, match="is not one of"): load_ledger_consumer_artifact(facts_path) + # Legacy passthrough still guards the assertion enum without the schema. + with pytest.raises(ValueError, match="unsupported assertion"): + load_ledger_consumer_artifact(facts_path, validate_rows=False) artifact_dir = _write_artifact_dir( tmp_path, - [_fact_row()], + [_schema_complete_fact_row()], manifest_overrides={"schema_version": "policyengine_ledger.other.v9"}, ) with pytest.raises(ValueError, match="schema_version"): load_ledger_consumer_artifact(artifact_dir) +def test_sol_counterexample_non_finite_fact_number_is_rejected(tmp_path): + # json.dumps writes the bare tokens NaN / Infinity / -Infinity, which + # json.loads accepts by default. The feed loader must reject them at parse + # time rather than compile a non-finite value into a target (finding #7). + for bad in (float("nan"), float("inf"), float("-inf")): + facts_path = tmp_path / "nonfinite.jsonl" + _write_facts(facts_path, [_schema_complete_fact_row(value=bad)]) + raw = facts_path.read_text() + assert ("NaN" in raw) or ("Infinity" in raw) + with pytest.raises(ValueError, match="non-finite JSON constant|not finite"): + load_ledger_consumer_artifact(facts_path) + # Legacy passthrough (validation off) must also refuse the constant. + with pytest.raises(ValueError, match="non-finite JSON constant|not finite"): + load_ledger_consumer_artifact(facts_path, validate_rows=False) + + def test_rejects_missing_and_empty_feeds(tmp_path): with pytest.raises(FileNotFoundError): load_ledger_consumer_artifact(tmp_path / "missing.jsonl") diff --git a/packages/populace-build/tests/test_ledger_schema.py b/packages/populace-build/tests/test_ledger_schema.py new file mode 100644 index 00000000..a0e72c6a --- /dev/null +++ b/packages/populace-build/tests/test_ledger_schema.py @@ -0,0 +1,218 @@ +"""Tests for the vendored stdlib consumer-fact schema validator.""" + +from __future__ import annotations + +import hashlib + +import pytest + +from populace.build.ledger_schema import ( + CONSUMER_FACT_SCHEMA_SHA256, + validate_consumer_fact_row, +) + +# 24 lowercase-hex characters: the identity-key suffix the schema pins. +_HEX = "0123456789abcdef01234567" + +# Pinning the exact bytes of the vendored schema. A silent edit to +# schemas/consumer_fact.v1.schema.json (drift from the published Ledger +# contract) fails here instead of quietly changing what rows are accepted. +_EXPECTED_SCHEMA_SHA256 = ( + "a3e52d3bac4dea497a30722fb70446a79bf11614d1c5e88a34e21a4b3f27acf1" +) + + +def _schema_complete_fact_row(**overrides): + """A row satisfying every required field and identity-key pattern.""" + row = { + "schema_version": "ledger.consumer_fact.v1", + "aggregate_fact_key": f"ledger.aggregate_fact.v2:{_HEX}", + "semantic_fact_key": f"ledger.semantic_fact.v2:{_HEX}", + "legacy_fact_key": f"ledger.fact.v1:{_HEX}", + "source_release_key": f"ledger.source_release.v2:{_HEX}", + "source_series_key": f"ledger.source_series.v2:{_HEX}", + "observed_measure_key": f"ledger.observed_measure.v2:{_HEX}", + "dimension_set_key": f"ledger.dimension_set.v2:{_HEX}", + "universe_constraint_set_key": f"ledger.universe_constraint_set.v2:{_HEX}", + "value": 15_286_017_359_000, + "value_type": "integer", + "assertion": "observation", + "period": {"type": "tax_year", "value": 2023}, + "geography": {"level": "country", "id": "0100000US"}, + "entity": {"name": "tax_unit"}, + "aggregation": {"method": "sum"}, + "observed_measure": { + "source_name": "irs_soi", + "source_table": "Publication 1304 Table 1.1", + "source_measure_id": "adjusted_gross_income", + "source_concept": "irs_soi.adjusted_gross_income", + "unit": "usd", + }, + "dimensions": {"income_range": "all", "filing_status": "all"}, + "universe_constraints": {"domain": "all_individual_income_tax_returns"}, + "source": { + "source_name": "irs_soi", + "source_table": "Publication 1304 Table 1.1", + "source_file": "23in11si.xls", + "vintage": "tax_year_2023", + "extracted_at": "2024-01-01T00:00:00Z", + "extraction_method": "manual", + "source_sha256": "ab" * 32, + "source_size_bytes": 1024, + "raw_r2_uri": "r2://ledger/irs_soi/23in11si.xls", + }, + "lineage": { + "source_record_id": "irs_soi.ty2023.table_1_1.all.adjusted_gross_income", + "source_cell_keys": ["ledger.source_cell.v1:cell"], + }, + } + row.update(overrides) + return row + + +def test_vendored_schema_sha256_is_pinned(): + assert CONSUMER_FACT_SCHEMA_SHA256 == _EXPECTED_SCHEMA_SHA256 + + +def test_schema_sha256_matches_vendored_bytes(): + from importlib.resources import files + + raw = files("populace.build.schemas").joinpath( + "consumer_fact.v1.schema.json" + ).read_bytes() + assert hashlib.sha256(raw).hexdigest() == CONSUMER_FACT_SCHEMA_SHA256 + + +def test_schema_complete_row_validates(): + validate_consumer_fact_row(_schema_complete_fact_row(), line_number=1, source="feed") + + +def test_row_must_be_object(): + with pytest.raises(ValueError, match="expected a JSON object"): + validate_consumer_fact_row([1, 2, 3], line_number=1, source="feed") + + +def test_missing_top_level_required_field_reports_path(): + row = _schema_complete_fact_row() + del row["entity"] + with pytest.raises(ValueError, match="'entity' is required but missing"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_missing_nested_required_field_reports_path(): + row = _schema_complete_fact_row(period={"type": "tax_year"}) + with pytest.raises(ValueError, match="'period.value' is required but missing"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + row = _schema_complete_fact_row( + observed_measure={ + "source_name": "irs_soi", + "source_table": "t", + "source_measure_id": "agi", + "source_concept": "irs_soi.agi", + } + ) + with pytest.raises(ValueError, match="'observed_measure.unit' is required"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_wrong_const_schema_version_rejected(): + row = _schema_complete_fact_row(schema_version="ledger.consumer_fact.v2") + with pytest.raises(ValueError, match="expected constant 'ledger.consumer_fact.v1'"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_wrong_enum_rejected(): + row = _schema_complete_fact_row(assertion="policyengine_aged") + with pytest.raises(ValueError, match="is not one of"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + row = _schema_complete_fact_row(value_type="money") + with pytest.raises(ValueError, match="'value_type'"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_pattern_violation_reports_field_and_pattern(): + row = _schema_complete_fact_row(aggregate_fact_key="ledger.aggregate_fact.v2:XYZ") + with pytest.raises(ValueError, match="does not match pattern"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_extra_top_level_field_rejected(): + row = _schema_complete_fact_row(surprise_field=1) + with pytest.raises(ValueError, match="'surprise_field' is not an allowed property"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_extra_nested_field_rejected(): + row = _schema_complete_fact_row( + period={"type": "tax_year", "value": 2023, "unexpected": 1} + ) + with pytest.raises(ValueError, match="'period.unexpected' is not an allowed"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_wrong_scalar_type_rejected(): + row = _schema_complete_fact_row(value={"nested": "object"}) + with pytest.raises(ValueError, match="'value': expected type"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_integer_field_rejects_string_and_boolean(): + row = _schema_complete_fact_row() + row["source"]["source_size_bytes"] = "1024" + with pytest.raises(ValueError, match="source.source_size_bytes.*expected type integer"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + row = _schema_complete_fact_row() + row["source"]["source_size_bytes"] = True + with pytest.raises(ValueError, match="source.source_size_bytes.*got boolean"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_dimensions_additional_property_type_enforced(): + row = _schema_complete_fact_row(dimensions={"income_range": {"nested": "bad"}}) + with pytest.raises(ValueError, match="'dimensions.income_range': expected type"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_array_items_validated_via_ref(): + row = _schema_complete_fact_row( + universe_constraints={ + "domain": "all_returns", + "constraints": [ + {"variable": "age", "operator": ">=", "value": 0, "role": "filter"}, + {"variable": "age", "operator": "<", "value": 5}, + ], + } + ) + with pytest.raises( + ValueError, match=r"universe_constraints.constraints\[1\].role' is required" + ): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_lineage_array_item_type_enforced(): + row = _schema_complete_fact_row() + row["lineage"]["source_cell_keys"] = [1, 2] + with pytest.raises( + ValueError, match=r"lineage.source_cell_keys\[0\]': expected type string" + ): + validate_consumer_fact_row(row, line_number=3, source="feed") + + +def test_sol_counterexample_non_finite_number_value_is_rejected(): + # json.loads accepts the JS constants NaN/Infinity/-Infinity and hands back + # a float that satisfies the ``number`` type check, silently poisoning a + # calibration target (finding #7). The validator must reject a non-finite + # numeric wherever it appears in a row. + for bad in (float("nan"), float("inf"), float("-inf")): + row = _schema_complete_fact_row(value=bad) + with pytest.raises(ValueError, match="is not finite"): + validate_consumer_fact_row(row, line_number=3, source="feed") + + # Non-finite in a nested numeric field is caught too. + nested = _schema_complete_fact_row() + nested["source"]["source_size_bytes"] = float("inf") + with pytest.raises(ValueError, match="is not finite"): + validate_consumer_fact_row(nested, line_number=3, source="feed") diff --git a/packages/populace-build/tests/test_ledger_targets.py b/packages/populace-build/tests/test_ledger_targets.py index b3d72549..75d854c8 100644 --- a/packages/populace-build/tests/test_ledger_targets.py +++ b/packages/populace-build/tests/test_ledger_targets.py @@ -255,6 +255,7 @@ def test__given_supported_ledger_fact__then_populace_target_preserves_lineage() measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, entity_by_ledger_entity={"tax_unit": "tax_unit"}, family_by_source_name={"irs_soi": "irs_soi"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, @@ -287,6 +288,7 @@ def test__given_consumer_contract_row__then_populace_target_preserves_lineage() measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, entity_by_ledger_entity={"tax_unit": "tax_unit"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -317,6 +319,7 @@ def test__given_consumer_contract_jsonl__then_populace_selects_targets( measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, entity_by_ledger_entity={"tax_unit": "tax_unit"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -333,6 +336,7 @@ def test__given_ledger_target_reference__then_it_compiles_model_mapping() -> Non # Given reference = LedgerTargetReference( name="nation/irs/adjusted gross income/total", + expected_unit="usd", ledger_fact_key="ledger.aggregate_fact.v2:abc123", entity="tax_unit", measure="adjusted_gross_income", @@ -405,6 +409,7 @@ def test__given_count_ledger_target_reference__then_compilation_fails() -> None: # Given reference = LedgerTargetReference( name="census_pep.cy2024.national_resident_population_age.0_to_4.population", + expected_unit="count", ledger_fact_key="ledger.aggregate_fact.v2:abc123", entity="person", measure="person_count", @@ -473,6 +478,7 @@ def test__given_duplicate_semantic_facts__then_aggregate_reference_still_compile ) reference = LedgerTargetReference( name="nation/irs/total tax/total", + expected_unit="usd", ledger_fact_key="ledger.aggregate_fact.v2:def456", entity="tax_unit", measure="income_tax", @@ -546,6 +552,7 @@ def test__given_selector_matches_multiple_years__then_latest_source_period_is_us # Given reference = LedgerTargetReference( name="latest SOI AGI total", + expected_unit="usd", ledger_selector={ "source_name": "irs_soi", "source_measure_id": "adjusted_gross_income", @@ -585,6 +592,7 @@ def test__given_selector_matches_future_year__then_latest_eligible_period_is_use # Given reference = LedgerTargetReference( name="latest eligible SOI AGI total", + expected_unit="usd", ledger_selector={ "source_name": "irs_soi", "source_measure_id": "adjusted_gross_income", @@ -624,6 +632,7 @@ def test__given_selector_matches_future_month__then_latest_eligible_period_is_us # Given reference = LedgerTargetReference( name="latest eligible Medicaid enrollment", + expected_unit="people", ledger_selector={ "source_name": "cms_medicaid", "source_measure_id": "total_medicaid_chip_enrollment", @@ -663,6 +672,7 @@ def test__given_selector_matches_multiple_eligible_months__then_latest_month_is_ # Given reference = LedgerTargetReference( name="latest eligible Medicaid enrollment", + expected_unit="people", ledger_selector={ "source_name": "cms_medicaid", "source_measure_id": "total_medicaid_chip_enrollment", @@ -700,6 +710,7 @@ def test__given_selector_matches_only_future_year__then_compilation_fails() -> N # Given reference = LedgerTargetReference( name="latest eligible SOI AGI total", + expected_unit="usd", ledger_selector={ "source_name": "irs_soi", "source_measure_id": "adjusted_gross_income", @@ -751,6 +762,7 @@ def test__given_non_count_reference_without_measure__then_compilation_fails( # Given reference = LedgerTargetReference( name="missing measure target", + expected_unit="usd", ledger_fact_key="ledger.aggregate_fact.v2:abc123", entity="tax_unit", measure=measure, @@ -857,6 +869,7 @@ def test__given_domain_scoped_fact_with_domain_filter__then_target_uses_filter() measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -907,6 +920,7 @@ def test__given_scoped_fact_with_only_domain_filter__then_it_is_unsupported() -> measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) fact = _ledger_fact( @@ -933,6 +947,7 @@ def test__given_scoped_fact_with_filter_mapping__then_target_uses_filter() -> No measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_source_record_id={source_record_id: "agi_under_1"}, ) fact = _ledger_fact( @@ -971,6 +986,7 @@ def test__given_rate_fact__then_it_is_reported_as_unsupported() -> None: measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) fact = _ledger_fact(aggregation={"method": "rate"}) @@ -989,6 +1005,7 @@ def test__given_count_fact__then_it_is_reported_as_unsupported() -> None: measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -1204,6 +1221,7 @@ def test__given_malformed_value__then_it_is_reported_as_unsupported() -> None: measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -1221,6 +1239,7 @@ def test__given_overflow_value__then_it_is_reported_as_unsupported() -> None: measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -1238,6 +1257,7 @@ def test__given_negative_value_without_signed_mapping__then_it_is_unsupported() measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) @@ -1255,6 +1275,7 @@ def test__given_negative_value_with_signed_mapping__then_target_is_signed() -> N measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, signed_by_concept=frozenset({"us:statutes/26/62#adjusted_gross_income"}), ) @@ -1275,6 +1296,7 @@ def test__given_ledger_target_metadata__then_coverage_gate_uses_structured_field measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, ) selection = select_ledger_targets([_ledger_fact()], mapping) @@ -1305,6 +1327,7 @@ def test__given_current_soi_like_row__then_ledger_adapter_matches_current_target measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, entity_by_ledger_entity={"tax_unit": "tax_unit"}, family_by_source_name={"irs_soi": "irs_soi"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, @@ -1332,6 +1355,7 @@ def test_ledger_metadata_records_assertion_and_fact_period(): measure_by_concept={ "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" }, + expected_unit_by_measure={"adjusted_gross_income": "usd"}, filter_by_domain={"all_individual_income_tax_returns": "is_tax_filer"}, ) # Legacy rows that omit the assertion field are not stamped in metadata @@ -1375,6 +1399,7 @@ def test_ledger_reference_selector_matches_on_assertion(): ] reference = LedgerTargetReference( name="cbo_projected_agi", + expected_unit="usd", ledger_selector={ "source_measure_id": "adjusted_gross_income", "assertion": "source_projection", @@ -1388,3 +1413,298 @@ def test_ledger_reference_selector_matches_on_assertion(): (spec,) = registry.specs assert spec.value == 16_000_000_000_000 assert spec.metadata["ledger_assertion"] == "source_projection" + + +# --------------------------------------------------------------------------- +# Finding #9: expected-unit gate (the thousands-vs-millions failure class). +# --------------------------------------------------------------------------- + + +def _agi_row_with_unit(unit: str, **overrides): + return _consumer_fact_row( + observed_measure={ + "source_name": "irs_soi", + "source_table": "Publication 1304 Table 1.1", + "source_measure_id": "adjusted_gross_income", + "source_concept": "irs_soi.adjusted_gross_income", + "unit": unit, + }, + **overrides, + ) + + +def test__given_mapping_unit_mismatch__then_compilation_fails_naming_both_units() -> ( + None +): + # Given: the model measure is declared in usd_billions, the fact is thousands. + mapping = LedgerTargetMapping( + measure_by_concept={ + "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" + }, + filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, + expected_unit_by_measure={"adjusted_gross_income": "usd_billions"}, + ) + + # When / Then: hard error at compile, naming both units. + with pytest.raises( + ValueError, + match="unit 'usd_thousands' does not match the expected unit 'usd_billions'", + ): + select_ledger_targets([_agi_row_with_unit("usd_thousands")], mapping) + + +def test__given_mapping_without_expected_unit__then_compilation_fails() -> None: + # Given: a measure mapping with no declared expected unit. + mapping = LedgerTargetMapping( + measure_by_concept={ + "us:statutes/26/62#adjusted_gross_income": "adjusted_gross_income" + }, + filter_by_domain={"all_individual_income_tax_returns": "is_tax_return"}, + ) + + # When / Then + with pytest.raises(ValueError, match="declares no expected unit"): + select_ledger_targets([_agi_row_with_unit("usd")], mapping) + + +def test__given_reference_unit_mismatch__then_compilation_fails_naming_both_units() -> ( + None +): + # Given + reference = LedgerTargetReference( + name="soi agi total", + expected_unit="usd_billions", + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + family="irs_soi", + ) + + # When / Then + with pytest.raises( + ValueError, + match="fact unit 'usd_thousands' does not match the declared " + "expected_unit 'usd_billions'", + ): + compile_ledger_target_references( + [_agi_row_with_unit("usd_thousands")], [reference], country="us" + ) + + +def test__given_reference_without_expected_unit__then_compilation_fails() -> None: + # Given: a reference that resolves a fact but declares no expected_unit. + reference = LedgerTargetReference( + name="soi agi total no unit", + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + family="irs_soi", + ) + + # When / Then + with pytest.raises(ValueError, match="expected_unit is required"): + compile_ledger_target_references( + [_consumer_fact_row()], [reference], country="us" + ) + + +def test__given_reference_value_scale__then_value_is_scaled_and_recorded() -> None: + # Given: a fact published in billions, scaled to raw dollars for the model. + reference = LedgerTargetReference( + name="soi agi billions", + expected_unit="usd_billions", + value_scale=1e9, + value_scale_declared=True, + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + family="irs_soi", + ) + + # When + registry = compile_ledger_target_references( + [_agi_row_with_unit("usd_billions", value=15_286)], + [reference], + country="us", + ) + + # Then + spec = registry.specs[0] + assert spec.value == pytest.approx(15_286 * 1e9) + assert spec.metadata["ledger_expected_unit"] == "usd_billions" + assert spec.metadata["ledger_value_scale"] == "1000000000" + + +def test__given_zero_value_scale__then_reference_construction_fails() -> None: + with pytest.raises(ValueError, match="value_scale must be a finite non-zero"): + LedgerTargetReference( + name="bad scale", + expected_unit="usd", + value_scale=0.0, + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + ) + + +def test_sol_counterexample_any_finite_scale_is_treated_as_authorized() -> None: + # value_scale defaults to 1.0 and __post_init__ accepts any finite non-zero + # scale, so an arbitrary 1_000_000 rescale would silently reach the model as + # if it were authorized. A production reference must explicitly declare an + # intended non-identity scale; an undeclared scale is a hard error at compile + # (finding #9 / Sol finding 4). + undeclared = LedgerTargetReference( + name="soi agi undeclared scale", + expected_unit="usd", + value_scale=1_000_000, + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + family="irs_soi", + ) + with pytest.raises(ValueError, match="value_scale.*is not declared"): + compile_ledger_target_references( + [_agi_row_with_unit("usd", value=15_286)], + [undeclared], + country="us", + ) + + # The identical scale, explicitly declared, is authorized and applied. + declared = LedgerTargetReference( + name="soi agi declared scale", + expected_unit="usd", + value_scale=1_000_000, + value_scale_declared=True, + ledger_fact_key="ledger.aggregate_fact.v2:abc123", + entity="tax_unit", + measure="adjusted_gross_income", + family="irs_soi", + ) + registry = compile_ledger_target_references( + [_agi_row_with_unit("usd", value=15_286)], + [declared], + country="us", + ) + assert registry.specs[0].value == pytest.approx(15_286 * 1_000_000) + + +# --------------------------------------------------------------------------- +# Finding #10: typed periods must not collapse in latest-fact selection. +# --------------------------------------------------------------------------- + +_TYPED_AGI_SELECTOR = { + "source_name": "irs_soi", + "source_measure_id": "adjusted_gross_income", + "geography_level": "country", + "geography_id": "0100000US", + "entity_name": "tax_unit", + "layout_groupby_value_id": "all", +} + + +def _typed_agi_fact(period_type: str, record_token: str, *, value: float): + return _consumer_fact_row( + aggregate_fact_key=f"ledger.aggregate_fact.v2:agi-{record_token}", + legacy_fact_key=f"ledger.fact.v1:agi-{record_token}", + semantic_fact_key=f"ledger.semantic_fact.v2:agi-{record_token}", + lineage={ + "source_record_id": ( + f"irs_soi.{record_token}.table_1_1.all.adjusted_gross_income" + ), + "source_cell_keys": ["ledger.source_cell.v1:cell"], + "source_row_keys": [], + }, + value=value, + period={"type": period_type, "value": int(record_token[2:])}, + layout={ + "record_set_id": f"irs_soi.{record_token}.table_1_1", + "groupby_dimension": "us:statutes/26/62#adjusted_gross_income", + "groupby_value_id": "all", + "measure_id": "adjusted_gross_income", + }, + ) + + +def test__given_typed_periods__then_period_type_selects_the_right_series() -> None: + # Given: a tax-year 2023 fact and a (later) calendar-year 2024 fact that a + # bare selector would otherwise merge into one "latest" series. + tax_year = _typed_agi_fact("tax_year", "ty2023", value=15_000_000_000_000) + calendar_year = _typed_agi_fact("calendar_year", "cy2024", value=16_000_000_000_000) + facts = [tax_year, calendar_year] + + tax_reference = LedgerTargetReference( + name="agi tax-year total", + expected_unit="usd", + period_type="tax_year", + ledger_selector=_TYPED_AGI_SELECTOR, + entity="tax_unit", + measure="adjusted_gross_income", + period=2024, + family="irs_soi", + ) + calendar_reference = LedgerTargetReference( + name="agi calendar-year total", + expected_unit="usd", + period_type="calendar_year", + ledger_selector=_TYPED_AGI_SELECTOR, + entity="tax_unit", + measure="adjusted_gross_income", + period=2024, + family="irs_soi", + ) + + # When / Then: the tax-year reference resolves the tax-year fact even though + # the calendar-year fact has a later period; typed selection never crosses. + tax_registry = compile_ledger_target_references(facts, [tax_reference], country="us") + assert tax_registry.specs[0].value == 15_000_000_000_000 + assert tax_registry.specs[0].metadata["ledger_period_type"] == "tax_year" + assert ( + tax_registry.specs[0].metadata["ledger_reference_period_type"] == "tax_year" + ) + + calendar_registry = compile_ledger_target_references( + facts, [calendar_reference], country="us" + ) + assert calendar_registry.specs[0].value == 16_000_000_000_000 + assert calendar_registry.specs[0].metadata["ledger_period_type"] == "calendar_year" + + +def test__given_mixed_period_types_without_period_type__then_ambiguity_error() -> None: + # Given: both a tax-year and calendar-year fact match the selector. + facts = [ + _typed_agi_fact("tax_year", "ty2024", value=15_000_000_000_000), + _typed_agi_fact("calendar_year", "cy2024", value=16_000_000_000_000), + ] + reference = LedgerTargetReference( + name="agi ambiguous total", + expected_unit="usd", + ledger_selector=_TYPED_AGI_SELECTOR, + entity="tax_unit", + measure="adjusted_gross_income", + period=2024, + family="irs_soi", + ) + + # When / Then: fail closed rather than silently picking one series. + with pytest.raises(ValueError, match="multiple period types"): + compile_ledger_target_references(facts, [reference], country="us") + + +def test__given_same_period_type_across_years__then_latest_still_selected() -> None: + # Regression: the typed-period fix must not break same-type latest-selection. + facts = [ + _typed_agi_fact("tax_year", "ty2022", value=14_000_000_000_000), + _typed_agi_fact("tax_year", "ty2023", value=15_000_000_000_000), + ] + reference = LedgerTargetReference( + name="agi latest tax-year", + expected_unit="usd", + ledger_selector=_TYPED_AGI_SELECTOR, + entity="tax_unit", + measure="adjusted_gross_income", + period=2024, + family="irs_soi", + ) + + registry = compile_ledger_target_references(facts, [reference], country="us") + assert registry.specs[0].value == 15_000_000_000_000 diff --git a/packages/populace-build/tests/test_spec_only_country_packages.py b/packages/populace-build/tests/test_spec_only_country_packages.py index 11116c26..b05f89aa 100644 --- a/packages/populace-build/tests/test_spec_only_country_packages.py +++ b/packages/populace-build/tests/test_spec_only_country_packages.py @@ -90,6 +90,10 @@ def test__given_country_package_manifests__then_resources_are_local_specs() -> N assert offenders == [] +# Shared package-data directories under build/ that are not country packages. +NON_COUNTRY_DIRECTORIES = frozenset({"schemas"}) + + def test__given_country_like_directories__then_each_has_a_manifest() -> None: # Given roots = [ @@ -98,6 +102,7 @@ def test__given_country_like_directories__then_each_has_a_manifest() -> None: if path.is_dir() and not path.name.startswith("__") and not path.name.endswith("_runtime") + and path.name not in NON_COUNTRY_DIRECTORIES ] # When diff --git a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py index 3714bd14..b299bbc7 100644 --- a/packages/populace-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/populace-build/tests/test_us_fiscal_refresh_builder.py @@ -503,6 +503,10 @@ def test_export_target_audit_is_opt_in(monkeypatch) -> None: "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", ], @@ -517,6 +521,10 @@ def test_export_target_audit_is_opt_in(monkeypatch) -> None: "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", "--audit-export-targets", @@ -536,6 +544,10 @@ def test_cd_targets_require_vintage_crosswalk(monkeypatch) -> None: "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", "--include-congressional-district-targets", @@ -551,6 +563,10 @@ def test_cd_targets_require_vintage_crosswalk(monkeypatch) -> None: "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", "--include-congressional-district-targets", @@ -574,6 +590,10 @@ def test_maximum_microsim_batch_size_defaults_and_overrides(monkeypatch) -> None "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", ], @@ -590,6 +610,10 @@ def test_maximum_microsim_batch_size_defaults_and_overrides(monkeypatch) -> None "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", "--maximum-microsim-batch-size", @@ -611,6 +635,10 @@ def test_staging_repo_can_default_from_environment(monkeypatch) -> None: "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", ], @@ -1738,6 +1766,10 @@ def n(self, entity): str(base_h5), "--ledger-facts", str(facts), + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", str(out), "--release-id", @@ -4757,6 +4789,10 @@ def test_staging_telemetry_defaults_on_and_no_staging_disables(tmp_path, monkeyp "build_us_fiscal_refresh_release.py", "--ledger-facts", "facts.jsonl", + "--ledger-facts-sha256", + "a" * 64, + "--ledger-manifest-sha256", + "b" * 64, "--out", "release", ], diff --git a/packages/populace-build/tests/test_us_fiscal_targets.py b/packages/populace-build/tests/test_us_fiscal_targets.py index 603794ab..90a021f0 100644 --- a/packages/populace-build/tests/test_us_fiscal_targets.py +++ b/packages/populace-build/tests/test_us_fiscal_targets.py @@ -2,6 +2,8 @@ from hashlib import sha256 from importlib.resources import files +import pytest + from populace.build import nonnegative_columns_gate, target_profile_coverage_gate from populace.build.us_runtime import ( US_FISCAL_MACRO_REALISM_BANDS, @@ -789,6 +791,7 @@ def test_medicaid_chip_enrollment_reference_uses_medicaid_and_chip_support() -> "observed_measure": { "source_name": "cms_medicaid", "source_measure_id": "total_medicaid_chip_enrollment", + "unit": "people", }, }, ], @@ -1227,6 +1230,27 @@ def test_dynamic_us_fiscal_targets_do_not_prefer_future_observed_years() -> None assert spec.period == 2024 +def test_sol_counterexample_dynamic_unit_gate_echoes_drifted_fact_unit() -> None: + # The production dynamic builders must not copy the expected unit from the + # very fact they gate. The latest income-tax fact here silently drifted from + # usd to usd_thousands with a correspondingly 1e6-smaller number; echoing the + # fact's own unit would let it validate against itself. The committed unit + # gate must reject it instead (finding #9 / Sol finding 4). + drifted_latest = _soi_income_tax_fact(2023, value=2_100_000_000_000) + drifted_latest["observed_measure"]["unit"] = "usd_thousands" + drifted_latest["value"] = 2_100_000 + + with pytest.raises(ValueError, match="does not match the declared expected_unit"): + compile_us_fiscal_target_registry( + [ + *packaged_reference_facts(), + _soi_income_tax_fact(2022, value=2_000_000_000_000), + drifted_latest, + ], + allow_unaged_dollar_targets=True, + ) + + def test_dynamic_us_fiscal_targets_skip_future_only_source_periods() -> None: registry = compile_us_fiscal_target_registry( [ diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 6b062542..a21fb180 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -39,7 +39,12 @@ parity_gate, target_profile_coverage_gate, ) -from populace.build.ledger_artifact import load_ledger_consumer_artifact +from populace.build.ledger_artifact import ( + LedgerConsumerArtifact, + load_ledger_consumer_artifact, + vendor_ledger_consumer_artifact, + verify_vendored_ledger_artifact, +) from populace.build.source_runtime import SourceRuntimeConfig, run_source_stage from populace.build.staging import StagingTelemetry from populace.build.us_runtime import ( @@ -605,16 +610,20 @@ def _parse_args() -> argparse.Namespace: ) parser.add_argument( "--ledger-facts-sha256", + required=True, help=( - "Pin: expected SHA-256 of consumer_facts.jsonl. The build " - "refuses to start if the feed does not match." + "Required pin: expected SHA-256 of consumer_facts.jsonl. The build " + "loads the feed fail-closed (require_pins) and refuses to start if " + "the feed does not match." ), ) parser.add_argument( "--ledger-manifest-sha256", + required=True, help=( - "Pin: expected SHA-256 of the Ledger consumer artifact " - "manifest.json. Requires an artifact directory feed." + "Required pin: expected SHA-256 of the Ledger consumer artifact " + "manifest.json. A hash-pinned artifact directory feed is required; " + "bare consumer-facts files are rejected." ), ) parser.add_argument( @@ -5043,6 +5052,7 @@ def _build_manifests( medicaid_enrollment_substitutions: Sequence[Mapping[str, object]] = (), staging: Mapping[str, object] | None = None, ledger_artifact: Mapping[str, object] | None = None, + ledger_artifact_vendored: Mapping[str, object] | None = None, ) -> None: dataset_path = artifact_root / DATASET_FILENAME calibration_path = artifact_root / CALIBRATION_FILENAME @@ -5101,6 +5111,9 @@ def _build_manifests( "runtime": runtime, "timing": timing_payload, "ledger_artifact": dict(ledger_artifact) if ledger_artifact else None, + "ledger_artifact_vendored": ( + dict(ledger_artifact_vendored) if ledger_artifact_vendored else None + ), "dataset": { "filename": DATASET_FILENAME, "sha256": dataset_sha, @@ -5279,6 +5292,9 @@ def _build_manifests( }, "timing": timing_payload, "ledger_artifact": dict(ledger_artifact) if ledger_artifact else None, + "ledger_artifact_vendored": ( + dict(ledger_artifact_vendored) if ledger_artifact_vendored else None + ), "warm_start_calibration": warm_start_payload, "selection_source": selection_source_payload, "default_dataset": default_dataset_payload, @@ -5555,6 +5571,23 @@ def _staging_telemetry( ) +def _load_release_ledger_artifact( + args: argparse.Namespace, +) -> LedgerConsumerArtifact: + """Load the pinned Ledger consumer feed for a release, fail-closed. + + ``require_pins=True`` (finding #12) rejects a bare feed, demands both + content-hash pins, and makes a manifest-listed profile whose file is missing + a hard error instead of a silent skip that then vendors an incomplete feed. + """ + return load_ledger_consumer_artifact( + args.ledger_facts, + expected_facts_sha256=args.ledger_facts_sha256, + expected_manifest_sha256=args.ledger_manifest_sha256, + require_pins=True, + ) + + def main() -> None: args = _parse_args() if _git_dirty(): @@ -5645,11 +5678,7 @@ def main() -> None: # build, not merely a test. assert_take_up_contract_current() assert_take_up_treatments_consistent() - ledger_artifact = load_ledger_consumer_artifact( - args.ledger_facts, - expected_facts_sha256=args.ledger_facts_sha256, - expected_manifest_sha256=args.ledger_manifest_sha256, - ) + ledger_artifact = _load_release_ledger_artifact(args) extra_support_exclusions = _load_zero_support_exclusions( args.zero_support_exclusions ) @@ -6774,6 +6803,15 @@ def main() -> None: ) telemetry.stage("manifests", message="Writing release manifests.") timing["total_build_seconds"] = time.perf_counter() - build_started + # The exact verified feed bytes ship inside the release: a hash alone + # cannot reconstruct a feed whose source file later disappears. + vendored_ledger = vendor_ledger_consumer_artifact( + ledger_artifact, + release_dir / "ledger_artifact", + verified_facts_sha256=args.ledger_facts_sha256, + verified_manifest_sha256=args.ledger_manifest_sha256, + ) + verify_vendored_ledger_artifact(release_dir / "ledger_artifact") _build_manifests( release_id=release_id, release_dir=release_dir, @@ -6798,6 +6836,7 @@ def main() -> None: warm_start_calibration=warm_start_calibration, selection_source=selection_source_payload, ledger_artifact=ledger_artifact.provenance(), + ledger_artifact_vendored=vendored_ledger, default_dataset=default_dataset, medicaid_enrollment_substitutions=medicaid_enrollment_substitutions, staging=(