diff --git a/chronicle/artifacts.py b/chronicle/artifacts.py index d962076..78bac89 100644 --- a/chronicle/artifacts.py +++ b/chronicle/artifacts.py @@ -18,6 +18,8 @@ import httpx import yaml +from chronicle.epoch import EMIT_EPOCH, Epoch, canonicalize_key, hash_domain + DEFAULT_R2_RAW_BUCKET = "ledger-raw" DEFAULT_R2_DERIVED_BUCKET = "ledger-derived" @@ -529,6 +531,10 @@ def publish_derived_artifacts( errors=("missing_build_id",), ) + # Validate the resolved identity before deriving object keys, invoking the + # uploader, or opening the optional registry output. + canonicalize_key("build", resolved_build_id) + resolved_r2_prefix = resolve_r2_prefix( prefix=r2_prefix, default_prefix=DEFAULT_R2_DERIVED_PREFIX, @@ -956,18 +962,20 @@ def build_artifact_key( build_id: str, artifact_name: str, sha256: str, + epoch: Epoch = EMIT_EPOCH, ) -> str: """Build a stable key for a derived build artifact registry row.""" payload = json.dumps( { "artifact_name": artifact_name, - "build_id": build_id, + "build_id": canonicalize_key("build", build_id), "sha256": sha256, }, sort_keys=True, separators=(",", ":"), ).encode("utf-8") - return f"ledger.build_artifact.v1:{hashlib.sha256(payload).hexdigest()[:32]}" + domain = hash_domain("build_artifact", epoch) + return f"{domain}:{hashlib.sha256(payload).hexdigest()[:32]}" def infer_build_id(input_dir: str | Path) -> str | None: diff --git a/chronicle/bundle.py b/chronicle/bundle.py index 514b8b1..ce139b6 100644 --- a/chronicle/bundle.py +++ b/chronicle/bundle.py @@ -9,16 +9,20 @@ from pathlib import Path from typing import Any, Callable, Sequence +from chronicle.epoch import canonicalize_key, schema_id from chronicle.source_package import ( SOURCE_PACKAGE_ALIASES, assert_alias_map_covers_packages, validate_source_package, ) from chronicle.suite import BuildSuiteReport, build_source_suite +from policyengine_chronicle.schema import ( + validate_consumer_fact_row_epochs, +) -BUNDLE_SCHEMA_VERSION = "ledger.bundle.v1" -BUNDLE_COVERAGE_SCHEMA_VERSION = "ledger.bundle_coverage.v1" -BUNDLE_SOURCES_SCHEMA_VERSION = "ledger.bundle_sources.v1" +BUNDLE_SCHEMA_VERSION = schema_id("bundle") +BUNDLE_COVERAGE_SCHEMA_VERSION = schema_id("bundle_coverage") +BUNDLE_SOURCES_SCHEMA_VERSION = schema_id("bundle_sources") DEFAULT_BUNDLE_SOURCES = tuple(sorted(SOURCE_PACKAGE_ALIASES)) UK_BUNDLE_SOURCE_PREFIXES = ( "dfe", @@ -123,6 +127,17 @@ "welshgov-ctrs-annual-report-2025-26", ) +_KEY_DOMAINS = { + "aggregate_fact_key": "aggregate_fact", + "semantic_fact_key": "semantic_fact", + "legacy_fact_key": "fact", + "source_release_key": "source_release", + "source_series_key": "source_series", + "observed_measure_key": "observed_measure", + "dimension_set_key": "dimension_set", + "universe_constraint_set_key": "universe_constraint_set", +} + def uk_bundle_sources_from_aliases() -> tuple[str, ...]: """Return UK-package aliases implied by the source-package directory prefixes.""" @@ -490,9 +505,9 @@ def _duplicate_key_reports( rows: list[dict[str, Any]], key: str, ) -> list[dict[str, Any]]: - grouped: dict[str, list[dict[str, Any]]] = {} + grouped: dict[Any, list[dict[str, Any]]] = {} for row in rows: - grouped.setdefault(row[key], []).append(row) + grouped.setdefault(_canonical_key(row[key], key), []).append(row) return [ { "key": key_value, @@ -500,13 +515,17 @@ def _duplicate_key_reports( "sources": sorted({_source_table_key(row) for row in key_rows}), "legacy_fact_keys": sorted( { - legacy_key + _canonical_key(legacy_key, "legacy_fact_key") for row in key_rows if (legacy_key := row.get("legacy_fact_key")) - } + }, + key=_identity_sort_key, ), } - for key_value, key_rows in sorted(grouped.items()) + for key_value, key_rows in sorted( + grouped.items(), + key=lambda item: _identity_sort_key(item[0]), + ) if len(key_rows) > 1 ] @@ -525,7 +544,24 @@ def _counts_by( def _unique_count(rows: list[dict[str, Any]], key: str) -> int: - return len({row[key] for row in rows if key in row}) + return len({_canonical_key(row[key], key) for row in rows if key in row}) + + +def _canonical_key(value: Any, field_name: str) -> Any: + """Return a stable identity for either accepted naming epoch.""" + + domain_name = _KEY_DOMAINS.get(field_name) + if domain_name is None or not isinstance(value, str): + return value + return canonicalize_key(domain_name, value) + + +def _identity_sort_key(value: Any) -> tuple[int, str]: + """Sort identity scalars deterministically without comparing their types.""" + + if isinstance(value, str): + return (0, value) + return (1, f"{type(value).__name__}:{value!r}") def _source_name(row: dict[str, Any]) -> str | None: @@ -600,11 +636,19 @@ def _prepare_output_dir(output_path: Path, *, replace: bool) -> None: def _load_jsonl(path: Path) -> list[dict[str, Any]]: - return [ - json.loads(line) - for line in path.read_text(encoding="utf-8").splitlines() - if line - ] + rows: list[dict[str, Any]] = [] + for line_number, line in enumerate( + path.read_text(encoding="utf-8").splitlines(), start=1 + ): + if not line: + continue + row = json.loads(line) + # Bundle assembly historically consumes suite output without applying + # the stricter consumer-artifact schema. Keep that boundary intact, + # while still rejecting identifiers outside the two accepted epochs. + validate_consumer_fact_row_epochs(row, line_number, path) + rows.append(row) + return rows def _write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None: diff --git a/chronicle/consumer_contract.py b/chronicle/consumer_contract.py index 08276b8..350923f 100644 --- a/chronicle/consumer_contract.py +++ b/chronicle/consumer_contract.py @@ -23,9 +23,10 @@ build_aggregate_constraints, build_fact_key, ) +from chronicle.epoch import EMIT_EPOCH, HASH_DOMAINS, Epoch, hash_domain, schema_id from chronicle.store import fact_to_mapping -CONSUMER_FACT_SCHEMA_VERSION = "ledger.consumer_fact.v1" +CONSUMER_FACT_SCHEMA_VERSION = schema_id("consumer_fact") @dataclass(frozen=True) @@ -79,10 +80,14 @@ def to_dict(self) -> dict[str, Any]: } -def build_source_release_key(fact: AggregateFact) -> str: - """Build the current best v2 source-release key from fact provenance.""" +def build_source_release_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: + """Build a source-release key from fact provenance.""" return _hash_key( - "ledger.source_release.v2", + hash_domain("source_release", epoch), { "source_name": fact.source.source_name, "source_table": fact.source.source_table, @@ -96,10 +101,14 @@ def build_source_release_key(fact: AggregateFact) -> str: ) -def build_source_series_key(fact: AggregateFact) -> str: +def build_source_series_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a source-series key for a logical publisher table or series.""" return _hash_key( - "ledger.source_series.v2", + hash_domain("source_series", epoch), { "source_name": fact.source.source_name, "source_table": fact.source.source_table, @@ -110,47 +119,79 @@ def build_source_series_key(fact: AggregateFact) -> str: ) -def build_observed_measure_key(fact: AggregateFact) -> str: +def build_observed_measure_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a source-observed measure key, separate from concept alignment.""" - return _hash_key("ledger.observed_measure.v2", _observed_measure_payload(fact)) + return _hash_key( + hash_domain("observed_measure", epoch), + _observed_measure_payload(fact), + ) -def build_dimension_set_key(fact: AggregateFact) -> str: +def build_dimension_set_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a canonical key for fact dimensions represented as filters.""" - return _hash_key("ledger.dimension_set.v2", _dimension_set_payload(fact)) + return _hash_key( + hash_domain("dimension_set", epoch), + _dimension_set_payload(fact), + ) -def build_universe_constraint_set_key(fact: AggregateFact) -> str: +def build_universe_constraint_set_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a canonical key for semantic universe constraints.""" return _hash_key( - "ledger.universe_constraint_set.v2", + hash_domain("universe_constraint_set", epoch), _universe_constraint_set_payload(fact), ) -def build_aggregate_fact_key(fact: AggregateFact) -> str: - """Build v2 source-specific aggregate fact identity.""" +def build_aggregate_fact_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: + """Build source-specific aggregate fact identity.""" return _hash_key( - "ledger.aggregate_fact.v2", + hash_domain("aggregate_fact", epoch), { - "source_release_key": build_source_release_key(fact), - "source_series_key": build_source_series_key(fact), - "observed_measure_key": build_observed_measure_key(fact), + # Nested identities stay in their frozen Ledger form so changing + # only the outer domain cannot change the canonical payload. + "source_release_key": build_source_release_key(fact, epoch=Epoch.LEDGER), + "source_series_key": build_source_series_key(fact, epoch=Epoch.LEDGER), + "observed_measure_key": build_observed_measure_key( + fact, epoch=Epoch.LEDGER + ), "aggregation": _aggregation_payload(fact), "period": asdict(fact.period), "geography": _geography_payload(fact), "entity": asdict(fact.entity), - "dimension_set_key": build_dimension_set_key(fact), - "universe_constraint_set_key": build_universe_constraint_set_key(fact), + "dimension_set_key": build_dimension_set_key(fact, epoch=Epoch.LEDGER), + "universe_constraint_set_key": build_universe_constraint_set_key( + fact, epoch=Epoch.LEDGER + ), "assertion": _assertion_key_value(fact), }, ) -def build_semantic_fact_key(fact: AggregateFact) -> str: - """Build v2 source-agnostic fact identity for downstream reconciliation.""" +def build_semantic_fact_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: + """Build source-agnostic fact identity for downstream reconciliation.""" return _hash_key( - "ledger.semantic_fact.v2", + hash_domain("semantic_fact", epoch), { "canonical_measure": { "concept": fact.measure.concept, @@ -160,7 +201,9 @@ def build_semantic_fact_key(fact: AggregateFact) -> str: "period": asdict(fact.period), "geography": _geography_payload(fact), "entity": asdict(fact.entity), - "universe_constraint_set_key": build_universe_constraint_set_key(fact), + "universe_constraint_set_key": build_universe_constraint_set_key( + fact, epoch=Epoch.LEDGER + ), "assertion": _assertion_key_value(fact), }, ) @@ -178,7 +221,11 @@ def _assertion_key_value(fact: AggregateFact) -> str | None: return fact.assertion -def build_concept_alignment_key(fact: AggregateFact) -> str | None: +def build_concept_alignment_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str | None: """Build a concept-alignment key when source alignment metadata exists.""" if not ( fact.measure.source_concept @@ -188,9 +235,11 @@ def build_concept_alignment_key(fact: AggregateFact) -> str | None: ): return None return _hash_key( - "ledger.concept_alignment.v2", + hash_domain("concept_alignment", epoch), { - "observed_measure_key": build_observed_measure_key(fact), + "observed_measure_key": build_observed_measure_key( + fact, epoch=Epoch.LEDGER + ), "canonical_concept": fact.measure.concept, "relation": fact.measure.concept_relation, "authority": fact.measure.concept_authority, @@ -199,12 +248,16 @@ def build_concept_alignment_key(fact: AggregateFact) -> str | None: ) -def consumer_fact_rows(facts: list[AggregateFact]) -> list[dict[str, Any]]: +def consumer_fact_rows( + facts: list[AggregateFact], + *, + emit_epoch: Epoch = EMIT_EPOCH, +) -> list[dict[str, Any]]: """Build JSON-compatible consumer-contract rows for facts.""" contract_report = validate_consumer_fact_contract(facts) if not contract_report.valid: raise ValueError("Cannot export invalid Chronicle consumer-contract facts.") - return [_consumer_fact_row(fact) for fact in facts] + return [_consumer_fact_row(fact, emit_epoch=emit_epoch) for fact in facts] def validate_consumer_fact_contract( @@ -214,6 +267,31 @@ def validate_consumer_fact_contract( errors: list[ConsumerFactContractIssue] = [] for index, fact in enumerate(facts): fact_key = build_fact_key(fact) + for lineage_field, domain_name in ( + ("source_cell_keys", "source_cell"), + ("source_row_keys", "source_row"), + ): + pair = HASH_DOMAINS[domain_name] + for key in getattr(fact, lineage_field): + try: + pair.infer_key_epoch(key) + except ValueError as error: + errors.append( + ConsumerFactContractIssue( + code="malformed_lineage_key", + message=str(error), + fact_index=index, + fact_key=fact_key, + field=f"lineage/{lineage_field}", + ) + ) + if any( + issue.code == "malformed_lineage_key" and issue.fact_index == index + for issue in errors + ): + # The remaining checks build the row, which canonicalizes lineage + # keys and would raise on the same malformed key. + continue filter_constraints = _filter_derived_constraints(fact) source_filter_variables = _source_filter_variables(fact, filter_constraints) if filter_constraints and not fact.constraints: @@ -450,7 +528,10 @@ def _fact_provenance_class_issue( "provenance_class", ) if fact.provenance_class == "survey_aggregate": - if type(fact.survey_instrument) is not str or not fact.survey_instrument.strip(): + if ( + type(fact.survey_instrument) is not str + or not fact.survey_instrument.strip() + ): return ( "missing_survey_instrument", "Survey aggregates need a non-empty survey instrument.", @@ -554,27 +635,37 @@ def _constraint_compare_payload(constraint: AggregateConstraint) -> str: ) -def consumer_fact_row(fact: AggregateFact) -> dict[str, Any]: +def consumer_fact_row( + fact: AggregateFact, + *, + emit_epoch: Epoch = EMIT_EPOCH, +) -> dict[str, Any]: """Build one consumer-contract row from an aggregate fact.""" - return consumer_fact_rows([fact])[0] + return consumer_fact_rows([fact], emit_epoch=emit_epoch)[0] -def _consumer_fact_row(fact: AggregateFact) -> dict[str, Any]: +def _consumer_fact_row( + fact: AggregateFact, + *, + emit_epoch: Epoch = EMIT_EPOCH, +) -> dict[str, Any]: """Build one consumer-contract row without recursive validation.""" - aggregate_fact_key = build_aggregate_fact_key(fact) - semantic_fact_key = build_semantic_fact_key(fact) - concept_alignment_key = build_concept_alignment_key(fact) + aggregate_fact_key = build_aggregate_fact_key(fact, epoch=emit_epoch) + semantic_fact_key = build_semantic_fact_key(fact, epoch=emit_epoch) + concept_alignment_key = build_concept_alignment_key(fact, epoch=emit_epoch) row: dict[str, Any] = { - "schema_version": CONSUMER_FACT_SCHEMA_VERSION, + "schema_version": schema_id("consumer_fact", emit_epoch), "aggregate_fact_key": aggregate_fact_key, "semantic_fact_key": semantic_fact_key, - "legacy_fact_key": build_fact_key(fact), - "source_release_key": build_source_release_key(fact), - "source_series_key": build_source_series_key(fact), - "observed_measure_key": build_observed_measure_key(fact), - "dimension_set_key": build_dimension_set_key(fact), - "universe_constraint_set_key": build_universe_constraint_set_key(fact), + "legacy_fact_key": build_fact_key(fact, epoch=emit_epoch), + "source_release_key": build_source_release_key(fact, epoch=emit_epoch), + "source_series_key": build_source_series_key(fact, epoch=emit_epoch), + "observed_measure_key": build_observed_measure_key(fact, epoch=emit_epoch), + "dimension_set_key": build_dimension_set_key(fact, epoch=emit_epoch), + "universe_constraint_set_key": build_universe_constraint_set_key( + fact, epoch=emit_epoch + ), "value": _json_value(fact.value), "value_type": _value_type(fact.value), "assertion": fact.assertion, @@ -590,8 +681,12 @@ def _consumer_fact_row(fact: AggregateFact) -> dict[str, Any]: "source": _clean(fact_to_mapping(fact)["source"]), "lineage": { "source_record_id": fact.source_record_id, - "source_cell_keys": list(fact.source_cell_keys), - "source_row_keys": list(fact.source_row_keys), + "source_cell_keys": _lineage_keys_for_epoch( + "source_cell", fact.source_cell_keys, emit_epoch + ), + "source_row_keys": _lineage_keys_for_epoch( + "source_row", fact.source_row_keys, emit_epoch + ), }, "layout": _clean(asdict(fact.layout)) if fact.layout else {}, "label": fact.label, @@ -615,9 +710,11 @@ def _consumer_fact_row(fact: AggregateFact) -> dict[str, Any]: def write_consumer_facts_jsonl( facts: list[AggregateFact], path: str | Path, + *, + emit_epoch: Epoch = EMIT_EPOCH, ) -> ConsumerFactExportReport: """Write consumer-contract fact rows to JSON Lines.""" - rows = consumer_fact_rows(facts) + rows = consumer_fact_rows(facts, emit_epoch=emit_epoch) output_path = Path(path) output_path.parent.mkdir(parents=True, exist_ok=True) with output_path.open("w") as file: @@ -625,12 +722,23 @@ def write_consumer_facts_jsonl( file.write(json.dumps(row, sort_keys=True)) file.write("\n") return ConsumerFactExportReport( - schema_version=CONSUMER_FACT_SCHEMA_VERSION, + schema_version=schema_id("consumer_fact", emit_epoch), fact_count=len(rows), output=str(output_path), ) +def _lineage_keys_for_epoch( + domain_name: str, + keys: tuple[str, ...], + epoch: Epoch, +) -> list[str]: + """Return accepted lineage identities under the requested emit epoch.""" + + pair = HASH_DOMAINS[domain_name] + return list(dict.fromkeys(pair.key_for_epoch(key, epoch) for key in keys)) + + def _hash_key(namespace: str, payload: dict[str, Any]) -> str: raw = json.dumps( _clean(payload), diff --git a/chronicle/core.py b/chronicle/core.py index 411388e..43d0125 100644 --- a/chronicle/core.py +++ b/chronicle/core.py @@ -14,6 +14,8 @@ from decimal import Decimal from typing import Any +from chronicle.epoch import EMIT_EPOCH, HASH_DOMAINS, Epoch, hash_domain + Scalar = str | int | float | bool | None # Year-typed periods store one integer. Split-label years store the opening @@ -100,7 +102,7 @@ "projection_horizon", } ALLOWED_ACCOUNTING_BASES = {"cash", "accrual"} -FACT_KEY_PREFIX = "ledger.fact.v1" +FACT_KEY_PREFIX = hash_domain("fact") @dataclass(frozen=True) @@ -313,12 +315,16 @@ def to_dict(self) -> dict[str, Any]: } -def build_fact_key(fact: AggregateFact) -> str: +def build_fact_key( + fact: AggregateFact, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a stable key from fact schema fields, not human labels.""" payload = _canonical_key_payload(fact) raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] - return f"{FACT_KEY_PREFIX}:{digest}" + return f"{hash_domain('fact', epoch)}:{digest}" def build_label(fact: AggregateFact) -> str: @@ -506,6 +512,18 @@ def validate_fact(fact: AggregateFact) -> tuple[ValidationIssue, ...]: _validate_filters(errors, fact.filters) _validate_constraints(errors, fact.constraints) _validate_provenance(errors, fact.source) + _validate_lineage_keys( + errors, + fact.source_cell_keys, + field="source_cell_keys", + domain="source_cell", + ) + _validate_lineage_keys( + errors, + fact.source_row_keys, + field="source_row_keys", + domain="source_row", + ) if fact.period_coverage is not None: _validate_period_coverage(errors, fact.period_coverage) @@ -631,6 +649,51 @@ def _validate_value(errors: list[ValidationIssue], value: Any) -> None: errors.append(_issue("missing_value", "Fact value is required", "value")) +def _validate_lineage_keys( + errors: list[ValidationIssue], + keys: tuple[str, ...], + *, + field: str, + domain: str, +) -> None: + pair = HASH_DOMAINS[domain] + seen: set[str] = set() + for key in keys: + if not isinstance(key, str): + errors.append( + _issue( + "malformed_lineage_key", + f"Unsupported lineage key {key!r}; accepted prefixes are " + f"{pair.ledger!r} and {pair.chronicle!r}", + field, + ) + ) + continue + try: + canonical_key = pair.key_for_epoch(key, Epoch.LEDGER) + except ValueError as error: + errors.append( + _issue( + "malformed_lineage_key", + str(error), + field, + ) + ) + continue + if canonical_key in seen: + errors.append( + _issue( + "duplicate_lineage_key", + f"Duplicate canonical lineage key {canonical_key!r}; each " + "lineage identity may appear only once across accepted " + "Ledger and Chronicle aliases", + field, + ) + ) + continue + seen.add(canonical_key) + + def _validate_provenance_class( errors: list[ValidationIssue], fact: AggregateFact, diff --git a/chronicle/database.py b/chronicle/database.py index 9c9aaed..95b0d9f 100644 --- a/chronicle/database.py +++ b/chronicle/database.py @@ -17,6 +17,13 @@ from typing import Any from chronicle.core import AggregateFact, build_aggregate_constraints, build_fact_key +from chronicle.epoch import ( + EMIT_EPOCH, + HASH_DOMAINS, + Epoch, + hash_domain, + schema_id, +) from chronicle.sources.cells import ( SourceCell, build_source_cell_key, @@ -33,7 +40,7 @@ source_row_to_mapping, ) -LEDGER_DB_SCHEMA_VERSION = "ledger.relational.v1" +LEDGER_DB_SCHEMA_VERSION = schema_id("relational", Epoch.LEDGER) @dataclass(frozen=True) @@ -63,6 +70,7 @@ def build_chronicle_db( source_rows: list[SourceRow] | None = None, build_id: str | None = None, replace: bool = False, + emit_epoch: Epoch = EMIT_EPOCH, ) -> ChronicleDbBuildReport: """Build a deterministic SQLite Chronicle database artifact.""" path = Path(db_path) @@ -76,7 +84,11 @@ def build_chronicle_db( rows = source_rows or [] columns = source_columns_from_source_rows(rows) source_row_values_count = sum(len(row.values) for row in rows) - resolved_build_id = build_id or _build_id(facts, cells, rows) + resolved_build_id = ( + HASH_DOMAINS["build"].key_for_epoch(build_id, emit_epoch) + if build_id is not None + else _build_id(facts, cells, rows, epoch=emit_epoch) + ) fact_constraints = [(fact, build_aggregate_constraints(fact)) for fact in facts] source_record_ids = { fact.source_record_id for fact in facts if fact.source_record_id is not None @@ -102,13 +114,24 @@ def build_chronicle_db( source_row_values_count=source_row_values_count, source_cells_count=len(cells), source_artifacts_count=len(artifact_sha256s), + emit_epoch=emit_epoch, ) - _insert_source_rows(connection, rows) - _insert_source_columns(connection, columns) - _insert_source_row_values(connection, rows, columns) - _insert_source_cells(connection, cells) + _insert_source_rows(connection, rows, emit_epoch=emit_epoch) + _insert_source_columns(connection, columns, emit_epoch=emit_epoch) + _insert_source_row_values( + connection, + rows, + columns, + emit_epoch=emit_epoch, + ) + _insert_source_cells(connection, cells, emit_epoch=emit_epoch) _insert_concept_alignments(connection, facts, resolved_build_id) - _insert_facts(connection, fact_constraints, resolved_build_id) + _insert_facts( + connection, + fact_constraints, + resolved_build_id, + emit_epoch=emit_epoch, + ) _create_indexes(connection) connection.commit() @@ -393,6 +416,7 @@ def _insert_build( source_row_values_count: int, source_cells_count: int, source_artifacts_count: int, + emit_epoch: Epoch, ) -> None: connection.execute( """ @@ -413,7 +437,7 @@ def _insert_build( """, ( build_id, - LEDGER_DB_SCHEMA_VERSION, + schema_id("relational", emit_epoch), datetime.now(timezone.utc).isoformat(), facts_count, constraints_count, @@ -430,6 +454,8 @@ def _insert_build( def _insert_source_cells( connection: sqlite3.Connection, cells: list[SourceCell], + *, + emit_epoch: Epoch, ) -> None: artifacts = {cell.artifact.sha256: cell.artifact for cell in cells} for artifact in artifacts.values(): @@ -458,9 +484,16 @@ def _insert_source_cells( VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( - build_source_cell_key(cell), + build_source_cell_key(cell, epoch=emit_epoch), cell.artifact.sha256, - cell.source_row_key, + ( + HASH_DOMAINS["source_row"].key_for_epoch( + cell.source_row_key, + emit_epoch, + ) + if cell.source_row_key is not None + else None + ), cell.sheet_name, cell.row_number, cell.column_number, @@ -479,6 +512,8 @@ def _insert_source_cells( def _insert_source_rows( connection: sqlite3.Connection, rows: list[SourceRow], + *, + emit_epoch: Epoch, ) -> None: artifacts = {row.artifact.sha256: row.artifact for row in rows} for artifact in artifacts.values(): @@ -497,7 +532,7 @@ def _insert_source_rows( VALUES (?, ?, ?, ?, ?) """, ( - build_source_row_key(row), + build_source_row_key(row, epoch=emit_epoch), row.artifact.sha256, row.sheet_name, row.row_number, @@ -509,6 +544,8 @@ def _insert_source_rows( def _insert_source_columns( connection: sqlite3.Connection, columns: list[SourceColumn], + *, + emit_epoch: Epoch, ) -> None: for column in columns: connection.execute( @@ -524,7 +561,7 @@ def _insert_source_columns( VALUES (?, ?, ?, ?, ?, ?) """, ( - build_source_column_key(column), + build_source_column_key(column, epoch=emit_epoch), column.artifact.sha256, column.sheet_name, column.column_number, @@ -538,10 +575,12 @@ def _insert_source_row_values( connection: sqlite3.Connection, rows: list[SourceRow], columns: list[SourceColumn], + *, + emit_epoch: Epoch, ) -> None: column_keys = { (column.artifact.sha256, column.sheet_name, column.column_number): ( - build_source_column_key(column), + build_source_column_key(column, epoch=emit_epoch), column.normalized_name, ) for column in columns @@ -565,7 +604,7 @@ def _insert_source_row_values( """ batch = [] for row in rows: - source_row_key = build_source_row_key(row) + source_row_key = build_source_row_key(row, epoch=emit_epoch) for column_number, (raw_name, value) in enumerate( row.values.items(), start=1, @@ -584,7 +623,7 @@ def _insert_source_row_values( ) batch.append( ( - build_source_row_value_key(row_value), + build_source_row_value_key(row_value, epoch=emit_epoch), source_row_key, source_column_key, row.artifact.sha256, @@ -648,9 +687,21 @@ def _insert_facts( connection: sqlite3.Connection, fact_constraints: list[tuple[AggregateFact, tuple[Any, ...]]], build_id: str, + *, + emit_epoch: Epoch, ) -> None: for fact, constraints in fact_constraints: - fact_key = build_fact_key(fact) + fact_key = build_fact_key(fact, epoch=emit_epoch) + source_cell_keys = _lineage_keys_for_epoch( + "source_cell", + fact.source_cell_keys, + emit_epoch, + ) + source_row_keys = _lineage_keys_for_epoch( + "source_row", + fact.source_row_keys, + emit_epoch, + ) if fact.source_record_id is not None: connection.execute( """ @@ -673,8 +724,8 @@ def _insert_facts( fact.source.source_table, fact.source.source_file, fact.source.vintage, - len(fact.source_row_keys), - len(fact.source_cell_keys), + len(source_row_keys), + len(source_cell_keys), ), ) _insert_aggregate_fact(connection, fact, fact_key, build_id) @@ -708,7 +759,7 @@ def _insert_facts( constraint.label, ), ) - for ordinal, source_cell_key in enumerate(fact.source_cell_keys): + for ordinal, source_cell_key in enumerate(source_cell_keys): connection.execute( """ INSERT INTO fact_source_cells ( @@ -718,9 +769,13 @@ def _insert_facts( ) VALUES (?, ?, ?) """, - (fact_key, source_cell_key, ordinal), + ( + fact_key, + source_cell_key, + ordinal, + ), ) - for ordinal, source_row_key in enumerate(fact.source_row_keys): + for ordinal, source_row_key in enumerate(source_row_keys): connection.execute( """ INSERT INTO fact_source_rows ( @@ -730,7 +785,11 @@ def _insert_facts( ) VALUES (?, ?, ?) """, - (fact_key, source_row_key, ordinal), + ( + fact_key, + source_row_key, + ordinal, + ), ) @@ -926,41 +985,96 @@ def _build_id( facts: list[AggregateFact], cells: list[SourceCell], rows: list[SourceRow], + *, + epoch: Epoch = EMIT_EPOCH, ) -> str: digest = hashlib.sha256() _update_build_hash(digest, "schema", {"version": LEDGER_DB_SCHEMA_VERSION}) - for fact in sorted(facts, key=build_fact_key): + for fact in sorted( + facts, + key=lambda item: build_fact_key(item, epoch=Epoch.LEDGER), + ): _update_build_hash( digest, "fact", { - "fact_key": build_fact_key(fact), - "fact": asdict(fact), + "fact_key": build_fact_key(fact, epoch=Epoch.LEDGER), + "fact": _canonical_fact_mapping(fact), "constraints": [ asdict(constraint) for constraint in build_aggregate_constraints(fact) ], }, ) - for cell in sorted(cells, key=build_source_cell_key): + for cell in sorted( + cells, + key=lambda item: build_source_cell_key(item, epoch=Epoch.LEDGER), + ): _update_build_hash( digest, "source_cell", { - "source_cell_key": build_source_cell_key(cell), - "source_cell": source_cell_to_mapping(cell), + "source_cell_key": build_source_cell_key( + cell, + epoch=Epoch.LEDGER, + ), + "source_cell": _canonical_source_cell_mapping(cell), }, ) - for row in sorted(rows, key=build_source_row_key): + for row in sorted( + rows, + key=lambda item: build_source_row_key(item, epoch=Epoch.LEDGER), + ): _update_build_hash( digest, "source_row", { - "source_row_key": build_source_row_key(row), + "source_row_key": build_source_row_key( + row, + epoch=Epoch.LEDGER, + ), "source_row": source_row_to_mapping(row), }, ) - return f"ledger.build.v1:{digest.hexdigest()[:24]}" + return f"{hash_domain('build', epoch)}:{digest.hexdigest()[:24]}" + + +def _canonical_fact_mapping(fact: AggregateFact) -> dict[str, Any]: + """Return a build-hash payload independent of accepted key epochs.""" + mapping = asdict(fact) + mapping["source_cell_keys"] = _lineage_keys_for_epoch( + "source_cell", + fact.source_cell_keys, + Epoch.LEDGER, + ) + mapping["source_row_keys"] = _lineage_keys_for_epoch( + "source_row", + fact.source_row_keys, + Epoch.LEDGER, + ) + return mapping + + +def _lineage_keys_for_epoch( + domain_name: str, + keys: tuple[str, ...], + epoch: Epoch, +) -> tuple[str, ...]: + """Return stable, first-seen lineage identities for database emission.""" + + pair = HASH_DOMAINS[domain_name] + return tuple(dict.fromkeys(pair.key_for_epoch(key, epoch) for key in keys)) + + +def _canonical_source_cell_mapping(cell: SourceCell) -> dict[str, Any]: + """Return a source-cell build payload independent of row-key epoch.""" + mapping = source_cell_to_mapping(cell) + if cell.source_row_key is not None: + mapping["source_row_key"] = HASH_DOMAINS["source_row"].key_for_epoch( + cell.source_row_key, + Epoch.LEDGER, + ) + return mapping def _update_build_hash( diff --git a/chronicle/epoch.py b/chronicle/epoch.py new file mode 100644 index 0000000..281d450 --- /dev/null +++ b/chronicle/epoch.py @@ -0,0 +1,174 @@ +"""Epoch registry for frozen Ledger and successor Chronicle identifiers. + +The migration is additive: readers accept both identifiers in each pair, while +emitters use :data:`EMIT_EPOCH` and therefore remain Ledger-named until a later +cutover changes that single default. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import StrEnum +from types import MappingProxyType +from typing import Mapping + + +class Epoch(StrEnum): + """Fact-identity naming epochs.""" + + LEDGER = "ledger" + CHRONICLE = "chronicle" + + +# Changing this default is a separate, consumer-gated migration step. +EMIT_EPOCH = Epoch.LEDGER + + +@dataclass(frozen=True) +class EpochPair: + """A frozen Ledger identifier and its Chronicle-era successor.""" + + ledger: str + chronicle: str + + @property + def accepted(self) -> tuple[str, str]: + return (self.ledger, self.chronicle) + + def for_epoch(self, epoch: Epoch = EMIT_EPOCH) -> str: + if epoch == Epoch.LEDGER: + return self.ledger + if epoch == Epoch.CHRONICLE: + return self.chronicle + raise ValueError(f"unknown emit epoch {epoch!r}; expected ledger or chronicle") + + def infer_identifier_epoch(self, identifier: str) -> Epoch: + if identifier == self.ledger: + return Epoch.LEDGER + if identifier == self.chronicle: + return Epoch.CHRONICLE + raise ValueError( + f"unsupported identifier {identifier!r}; accepted forms are " + f"{self.ledger!r} and {self.chronicle!r}" + ) + + def infer_key_epoch(self, key: str) -> Epoch: + prefix, separator, _digest = key.partition(":") + if not separator: + raise ValueError( + f"unsupported key {key!r}; accepted prefixes are " + f"{self.ledger!r} and {self.chronicle!r}" + ) + try: + return self.infer_identifier_epoch(prefix) + except ValueError as error: + raise ValueError( + f"unsupported key {key!r}; accepted prefixes are " + f"{self.ledger!r} and {self.chronicle!r}" + ) from error + + def key_for_epoch(self, key: str, epoch: Epoch) -> str: + """Return *key* under *epoch* without changing its payload digest.""" + + self.infer_key_epoch(key) + _prefix, _separator, digest = key.partition(":") + return f"{self.for_epoch(epoch)}:{digest}" + + +HASH_DOMAINS: Mapping[str, EpochPair] = MappingProxyType( + { + "source_release": EpochPair( + "ledger.source_release.v2", "chronicle.source_release.v3" + ), + "source_series": EpochPair( + "ledger.source_series.v2", "chronicle.source_series.v3" + ), + "observed_measure": EpochPair( + "ledger.observed_measure.v2", "chronicle.observed_measure.v3" + ), + "dimension_set": EpochPair( + "ledger.dimension_set.v2", "chronicle.dimension_set.v3" + ), + "universe_constraint_set": EpochPair( + "ledger.universe_constraint_set.v2", + "chronicle.universe_constraint_set.v3", + ), + "aggregate_fact": EpochPair( + "ledger.aggregate_fact.v2", "chronicle.aggregate_fact.v3" + ), + "semantic_fact": EpochPair( + "ledger.semantic_fact.v2", "chronicle.semantic_fact.v3" + ), + "concept_alignment": EpochPair( + "ledger.concept_alignment.v2", "chronicle.concept_alignment.v3" + ), + "fact": EpochPair("ledger.fact.v1", "chronicle.fact.v2"), + "source_cell": EpochPair("ledger.source_cell.v1", "chronicle.source_cell.v2"), + "source_row": EpochPair("ledger.source_row.v1", "chronicle.source_row.v2"), + "source_column": EpochPair( + "ledger.source_column.v1", "chronicle.source_column.v2" + ), + "source_row_value": EpochPair( + "ledger.source_row_value.v1", "chronicle.source_row_value.v2" + ), + "build": EpochPair("ledger.build.v1", "chronicle.build.v2"), + "build_artifact": EpochPair( + "ledger.build_artifact.v1", "chronicle.build_artifact.v2" + ), + } +) + + +SCHEMA_IDS: Mapping[str, EpochPair] = MappingProxyType( + { + "bundle": EpochPair("ledger.bundle.v1", "chronicle.bundle.v2"), + "bundle_coverage": EpochPair( + "ledger.bundle_coverage.v1", "chronicle.bundle_coverage.v2" + ), + "bundle_sources": EpochPair( + "ledger.bundle_sources.v1", "chronicle.bundle_sources.v2" + ), + "consumer_fact": EpochPair( + "ledger.consumer_fact.v1", "chronicle.consumer_fact.v2" + ), + "relational": EpochPair("ledger.relational.v1", "chronicle.relational.v2"), + "source_package": EpochPair( + "ledger.source_package.v1", "chronicle.source_package.v2" + ), + "offline_fetch_manifest": EpochPair( + "ledger.offline_fetch_manifest.v1", + "chronicle.offline_fetch_manifest.v2", + ), + "fetch_manifest": EpochPair( + "ledger.fetch_manifest.v1", "chronicle.fetch_manifest.v2" + ), + # Facts-only v2 is the live artifact contract. The retired target-profile + # and resolved-target v1 contracts intentionally have no successors. + "consumer_artifact": EpochPair( + "policyengine_ledger.consumer_artifact.v2", + "policyengine_chronicle.consumer_artifact.v3", + ), + "approved_agents": EpochPair( + "policyengine_ledger.approved_agents.v1", + "policyengine_chronicle.approved_agents.v2", + ), + } +) + + +def hash_domain(name: str, epoch: Epoch = EMIT_EPOCH) -> str: + """Return the hash domain registered for *name* and *epoch*.""" + + return HASH_DOMAINS[name].for_epoch(epoch) + + +def schema_id(name: str, epoch: Epoch = EMIT_EPOCH) -> str: + """Return the schema id registered for *name* and *epoch*.""" + + return SCHEMA_IDS[name].for_epoch(epoch) + + +def canonicalize_key(name: str, key: str) -> str: + """Normalize either accepted key form to its immutable Ledger form.""" + + return HASH_DOMAINS[name].key_for_epoch(key, Epoch.LEDGER) diff --git a/chronicle/harness.py b/chronicle/harness.py index 1d96f09..298a138 100644 --- a/chronicle/harness.py +++ b/chronicle/harness.py @@ -1011,7 +1011,11 @@ def main(argv: list[str] | None = None) -> int: ) derived_publish_parser.add_argument( "--build-id", - help="Build ID. Defaults to the ID inferred from reports or ledger.db.", + help=( + "Build ID under an accepted epoch prefix, ledger.build.v1: or " + "chronicle.build.v2:; any other form is refused. Defaults to " + "the ID inferred from reports or ledger.db." + ), ) derived_publish_parser.add_argument( "--r2-bucket", diff --git a/chronicle/source_package.py b/chronicle/source_package.py index 7234621..20e1c0d 100644 --- a/chronicle/source_package.py +++ b/chronicle/source_package.py @@ -32,6 +32,7 @@ AggregateFact, build_label, ) +from chronicle.epoch import SCHEMA_IDS, schema_id from chronicle.sources.cells import ( SourceArtifactMetadata, SourceCell, @@ -73,6 +74,7 @@ ) SOURCE_PACKAGE_RESOURCE_PACKAGE = "packages" +SOURCE_PACKAGE_SCHEMA_VERSION = schema_id("source_package") SOURCE_PACKAGE_ALIASES = { "bea-nipa-personal-income-components": Path("bea/nipa_personal_income_components"), "bea-nipa-personal-income-disposition": Path( @@ -1135,8 +1137,13 @@ def load_source_package(source: str | Path) -> SourcePackage: with path.open("r", encoding="utf-8") as file: payload = yaml.safe_load(file) schema_version = _required(payload, "schema_version", str(path)) - if schema_version != "ledger.source_package.v1": - raise ValueError(f"Unsupported source package schema: {schema_version}") + source_package_schema = SCHEMA_IDS["source_package"] + if schema_version not in source_package_schema.accepted: + raise ValueError( + f"Unsupported source package schema: {schema_version!r}; accepted " + f"forms are {source_package_schema.ledger!r} and " + f"{source_package_schema.chronicle!r}" + ) package_dir = path.parent return SourcePackage( package_id=_required(payload, "package_id", str(path)), @@ -2326,7 +2333,7 @@ def _scaffold_template( resource_directory: str, manifest: str, ) -> str: - return f"""schema_version: ledger.source_package.v1 + return f"""schema_version: {SOURCE_PACKAGE_SCHEMA_VERSION} package_id: {package_id} label: TODO package label artifact: diff --git a/chronicle/sources/admin_packages.py b/chronicle/sources/admin_packages.py index d7cd79a..37958d3 100644 --- a/chronicle/sources/admin_packages.py +++ b/chronicle/sources/admin_packages.py @@ -22,6 +22,7 @@ from openpyxl import load_workbook import yaml +from chronicle.epoch import schema_id from chronicle.sources.cells import decode_delimited_text @@ -68,9 +69,7 @@ def build_irs_soi_county_package(root: Path = REPO_ROOT) -> dict[str, Any]: seen_fips: set[str] = set() eligible = [] for source_row in source_rows: - county_fips = _county_fips( - source_row["STATEFIPS"], source_row["COUNTYFIPS"] - ) + county_fips = _county_fips(source_row["STATEFIPS"], source_row["COUNTYFIPS"]) if county_fips.endswith("000"): continue if county_fips.endswith("999"): @@ -142,7 +141,7 @@ def build_irs_soi_county_package(root: Path = REPO_ROOT) -> dict[str, Any]: ) record_set_id = "irs_soi.ty2022.county_totals" return { - "schema_version": "ledger.source_package.v1", + "schema_version": schema_id("source_package"), "package_id": IRS_PACKAGE_ID, "label": f"{fixture_prefix}IRS SOI 2022 county return and AGI totals", "fixture": fixture, @@ -323,7 +322,7 @@ def build_census_pep_county_package(root: Path = REPO_ROOT) -> dict[str, Any]: ) record_set_id = "census_pep.vintage2024.county_population" return { - "schema_version": "ledger.source_package.v1", + "schema_version": schema_id("source_package"), "package_id": PEP_PACKAGE_ID, "label": f"{fixture_prefix}Census PEP Vintage 2024 county population", "fixture": fixture, @@ -365,9 +364,7 @@ def build_census_pep_county_package(root: Path = REPO_ROOT) -> dict[str, Any]: "measure_id": "resident_population", "label": "Resident population", "ordinal": 0, - "column": _excel_column( - headers.index("POPESTIMATE2024") + 1 - ), + "column": _excel_column(headers.index("POPESTIMATE2024") + 1), "source_column_id": "POPESTIMATE2024", "expected_column_header_row": 1, "expected_column_header": "POPESTIMATE2024", @@ -462,7 +459,10 @@ def build_usda_snap_monthly_package(root: Path = REPO_ROOT) -> dict[str, Any]: period = row_period source_month_label = str(month_label) geography_name = template_row["geography_name"] - if worksheet.cell(row=heading_row, column=1).value != geography_name: + if ( + worksheet.cell(row=heading_row, column=1).value + != geography_name + ): raise ValueError( f"Unexpected geography heading for {geography_name}" ) @@ -551,7 +551,7 @@ def build_usda_snap_monthly_package(root: Path = REPO_ROOT) -> dict[str, Any]: latest_month = datetime.strptime(latest_period, "%Y-%m").strftime("%B %Y") return { - "schema_version": "ledger.source_package.v1", + "schema_version": schema_id("source_package"), "package_id": SNAP_PACKAGE_ID, "label": f"USDA FNS SNAP FY2025 monthly state caseloads through {latest_month}", "fixture": False, diff --git a/chronicle/sources/cells.py b/chronicle/sources/cells.py index 25aa225..5b75c7c 100644 --- a/chronicle/sources/cells.py +++ b/chronicle/sources/cells.py @@ -20,8 +20,10 @@ import openpyxl import xlrd +from chronicle.epoch import EMIT_EPOCH, HASH_DOMAINS, Epoch, hash_domain + Scalar = str | int | float | bool | None -SOURCE_CELL_KEY_PREFIX = "ledger.source_cell.v1" +SOURCE_CELL_KEY_PREFIX = hash_domain("source_cell") @dataclass(frozen=True) @@ -99,7 +101,11 @@ def to_dict(self) -> dict[str, Any]: } -def build_source_cell_key(cell: SourceCell) -> str: +def build_source_cell_key( + cell: SourceCell, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a stable key from artifact hash and sheet coordinates.""" payload = { "artifact_sha256": cell.artifact.sha256, @@ -109,7 +115,7 @@ def build_source_cell_key(cell: SourceCell) -> str: } raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] - return f"{SOURCE_CELL_KEY_PREFIX}:{digest}" + return f"{hash_domain('source_cell', epoch)}:{digest}" def source_cells_from_xls( @@ -437,6 +443,30 @@ def validate_source_cells(cells: list[SourceCell]) -> SourceCellReport: cell_index=index, ) ) + if cell.source_row_key is not None: + source_row_pair = HASH_DOMAINS["source_row"] + if not isinstance(cell.source_row_key, str): + message = ( + f"Unsupported source-row key {cell.source_row_key!r}; accepted " + f"prefixes are {source_row_pair.ledger!r} and " + f"{source_row_pair.chronicle!r}" + ) + else: + try: + source_row_pair.infer_key_epoch(cell.source_row_key) + except ValueError as error: + message = str(error) + else: + message = None + if message is not None: + errors.append( + SourceCellIssue( + code="malformed_source_row_key", + message=message, + source_cell_key=key, + cell_index=index, + ) + ) for key, indices in key_indices.items(): if len(indices) > 1: diff --git a/chronicle/sources/offline_fetch.py b/chronicle/sources/offline_fetch.py index 6686203..f6154e0 100644 --- a/chronicle/sources/offline_fetch.py +++ b/chronicle/sources/offline_fetch.py @@ -11,8 +11,9 @@ from typing import Any, NoReturn from urllib.parse import urlsplit +from chronicle.epoch import SCHEMA_IDS, schema_id -OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION = "ledger.offline_fetch_manifest.v1" +OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION = schema_id("offline_fetch_manifest") _LOWERCASE_SHA256 = re.compile(r"[0-9a-f]{64}") _MISSING = object() @@ -123,18 +124,20 @@ def validate_offline_fetch_manifest( source: str = "offline fetch manifest", require_discovery_notes: bool = False, ) -> OfflineFetchManifest: - """Validate a parsed ``ledger.offline_fetch_manifest.v1`` mapping.""" + """Validate a parsed offline-fetch manifest from either accepted epoch.""" if not isinstance(payload, Mapping): _fail(source, "$", "must be a JSON object") _reject_unknown_fields(payload, _MANIFEST_FIELDS, source=source, location="$") schema_version = _required_string(payload, "schema_version", source=source) - if schema_version != OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION: + offline_fetch_schema = SCHEMA_IDS["offline_fetch_manifest"] + if schema_version not in offline_fetch_schema.accepted: _fail( source, "schema_version", - f"must be {OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION!r}", + f"must be {offline_fetch_schema.ledger!r} or " + f"{offline_fetch_schema.chronicle!r}", ) generated_for = _required_string(payload, "generated_for", source=source) diff --git a/chronicle/sources/rows.py b/chronicle/sources/rows.py index ba2fe30..449d703 100644 --- a/chronicle/sources/rows.py +++ b/chronicle/sources/rows.py @@ -17,6 +17,7 @@ import openpyxl +from chronicle.epoch import EMIT_EPOCH, Epoch, canonicalize_key, hash_domain from chronicle.sources.cells import ( SourceArtifactMetadata, SourceCell, @@ -24,9 +25,9 @@ ) Scalar = str | int | float | bool | None -SOURCE_ROW_KEY_PREFIX = "ledger.source_row.v1" -SOURCE_COLUMN_KEY_PREFIX = "ledger.source_column.v1" -SOURCE_ROW_VALUE_KEY_PREFIX = "ledger.source_row_value.v1" +SOURCE_ROW_KEY_PREFIX = hash_domain("source_row") +SOURCE_COLUMN_KEY_PREFIX = hash_domain("source_column") +SOURCE_ROW_VALUE_KEY_PREFIX = hash_domain("source_row_value") @dataclass(frozen=True) @@ -103,7 +104,11 @@ def to_dict(self) -> dict[str, Any]: } -def build_source_row_key(row: SourceRow) -> str: +def build_source_row_key( + row: SourceRow, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a stable key from artifact hash and row coordinate.""" payload = { "artifact_sha256": row.artifact.sha256, @@ -112,10 +117,14 @@ def build_source_row_key(row: SourceRow) -> str: } raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] - return f"{SOURCE_ROW_KEY_PREFIX}:{digest}" + return f"{hash_domain('source_row', epoch)}:{digest}" -def build_source_column_key(column: SourceColumn) -> str: +def build_source_column_key( + column: SourceColumn, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a stable key from artifact hash and column coordinate.""" payload = { "artifact_sha256": column.artifact.sha256, @@ -124,18 +133,24 @@ def build_source_column_key(column: SourceColumn) -> str: } raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] - return f"{SOURCE_COLUMN_KEY_PREFIX}:{digest}" + return f"{hash_domain('source_column', epoch)}:{digest}" -def build_source_row_value_key(row_value: SourceRowValue) -> str: +def build_source_row_value_key( + row_value: SourceRowValue, + *, + epoch: Epoch = EMIT_EPOCH, +) -> str: """Build a stable key from source row and column keys.""" payload = { - "source_row_key": row_value.source_row_key, - "source_column_key": row_value.source_column_key, + "source_row_key": canonicalize_key("source_row", row_value.source_row_key), + "source_column_key": canonicalize_key( + "source_column", row_value.source_column_key + ), } raw = json.dumps(payload, sort_keys=True, separators=(",", ":")) digest = hashlib.sha256(raw.encode("utf-8")).hexdigest()[:24] - return f"{SOURCE_ROW_VALUE_KEY_PREFIX}:{digest}" + return f"{hash_domain('source_row_value', epoch)}:{digest}" def source_rows_from_delimited_text( @@ -248,7 +263,9 @@ def source_rows_from_xlsx_table( ] continue values = { - column: _xlsx_row_scalar(raw_row[index]) if index < len(raw_row) else None + column: _xlsx_row_scalar(raw_row[index]) + if index < len(raw_row) + else None for index, column in enumerate(header or ()) } rows.append( diff --git a/chronicle/suite.py b/chronicle/suite.py index 07b9551..157b742 100644 --- a/chronicle/suite.py +++ b/chronicle/suite.py @@ -23,6 +23,7 @@ validate_facts, ) from chronicle.database import ChronicleDbBuildReport, build_chronicle_db +from chronicle.epoch import canonicalize_key from chronicle.sources.cells import ( SourceCell, SourceCellReport, @@ -554,6 +555,19 @@ def validate_source_regions( ) +def _canonicalize_lineage_key(domain_name: str, key: Any) -> str: + """Canonicalize an accepted lineage key without hiding invalid prefixes.""" + + if not isinstance(key, str): + return f"" + try: + return canonicalize_key(domain_name, key) + except ValueError: + # ``validate_facts`` reports the dual-prefix error. Keeping an unknown + # value here lets agent acceptance also report it as unresolved. + return key + + def build_agent_acceptance_report( facts: list[AggregateFact], rows: list[SourceRow], @@ -576,9 +590,14 @@ def build_agent_acceptance_report( **{row.artifact.sha256: row.artifact for row in rows}, **{cell.artifact.sha256: cell.artifact for cell in cells}, } - source_rows_by_key = {build_source_row_key(row): row for row in rows} + source_rows_by_key = { + canonicalize_key("source_row", build_source_row_key(row)): row for row in rows + } source_row_keys = set(source_rows_by_key) - source_cells_by_key = {build_source_cell_key(cell): cell for cell in cells} + source_cells_by_key = { + canonicalize_key("source_cell", build_source_cell_key(cell)): cell + for cell in cells + } source_column_dimensions_by_record_id = source_column_dimensions_by_record_id or {} raw_r2_link_count = 0 @@ -654,8 +673,12 @@ def build_agent_acceptance_report( ) ) continue + canonical_row_keys = [ + _canonicalize_lineage_key("source_row", key) + for key in fact.source_row_keys + ] unresolved_keys = [ - key for key in fact.source_row_keys if key not in source_row_keys + key for key in canonical_row_keys if key not in source_row_keys ] if unresolved_keys: missing_row_resolution_count += 1 @@ -673,11 +696,17 @@ def build_agent_acceptance_report( continue for issue in _row_semantic_evidence_issues( fact, - [source_rows_by_key[key] for key in fact.source_row_keys], + [source_rows_by_key[key] for key in canonical_row_keys], [ - source_cells_by_key[key] + source_cells_by_key[canonical_key] for key in fact.source_cell_keys - if key in source_cells_by_key + if ( + canonical_key := _canonicalize_lineage_key( + "source_cell", + key, + ) + ) + in source_cells_by_key ], source_column_dimensions=( source_column_dimensions_by_record_id.get( diff --git a/packages/statbel/fiscal_income_distribution_2023/build_package.py b/packages/statbel/fiscal_income_distribution_2023/build_package.py index 4dc4843..46bf7ed 100644 --- a/packages/statbel/fiscal_income_distribution_2023/build_package.py +++ b/packages/statbel/fiscal_income_distribution_2023/build_package.py @@ -17,6 +17,7 @@ import json from pathlib import Path import posixpath +import sys from typing import Any import xml.etree.ElementTree as ET from zipfile import ZipFile @@ -26,7 +27,15 @@ import yaml +PACKAGE_DIR = Path(__file__).resolve().parent +REPO_ROOT = PACKAGE_DIR.parents[2] +sys.path.insert(0, str(REPO_ROOT)) + +from chronicle.epoch import schema_id # noqa: E402 + + PACKAGE_ID = "statbel-fiscal-income-distribution-2023" +SOURCE_PACKAGE_SCHEMA_VERSION = schema_id("source_package") LANDING_PAGE = "https://statbel.fgov.be/en/themes/households/taxable-income" RAW_BASE_URL = ( "https://statbel.fgov.be/sites/default/files/files/documents/" @@ -37,8 +46,6 @@ ASSESSMENT_YEAR = 2024 CENT = Decimal("0.01") -PACKAGE_DIR = Path(__file__).resolve().parent -REPO_ROOT = PACKAGE_DIR.parents[2] DATA_DIR = REPO_ROOT / "db" / "data" / "statbel" / "fiscal_income_distribution_2023" CSV_FILENAME = "statbel_fiscal_income_distribution_2023.csv" CSV_PATH = DATA_DIR / CSV_FILENAME @@ -1654,7 +1661,7 @@ def _build_source_package( ) payload = { - "schema_version": "ledger.source_package.v1", + "schema_version": SOURCE_PACKAGE_SCHEMA_VERSION, "package_id": PACKAGE_ID, "label": ( "Statbel fiscal income distribution 2023 by income class, decile, " diff --git a/policyengine_chronicle/consumer.py b/policyengine_chronicle/consumer.py index f1d808e..b0d432c 100644 --- a/policyengine_chronicle/consumer.py +++ b/policyengine_chronicle/consumer.py @@ -22,12 +22,21 @@ ALLOWED_PROVENANCE_CLASSES, DEFAULT_ASSERTION, ) +from chronicle.epoch import ( + EMIT_EPOCH, + HASH_DOMAINS, + SCHEMA_IDS, + Epoch, + canonicalize_key, + hash_domain, + schema_id, +) from policyengine_chronicle.schema import ( CONSUMER_FACT_SCHEMA_SHA256, validate_consumer_fact_row, ) -CONSUMER_ARTIFACT_SCHEMA_VERSION = "policyengine_ledger.consumer_artifact.v2" +CONSUMER_ARTIFACT_SCHEMA_VERSION = schema_id("consumer_artifact") @dataclass(frozen=True) @@ -57,6 +66,7 @@ def build_consumer_artifact( *, facts_path: str | Path, replace: bool = False, + emit_epoch: Epoch = EMIT_EPOCH, ) -> ConsumerArtifactBuildReport: """Build a reproducible facts-only artifact from consumer fact rows. @@ -81,8 +91,9 @@ def build_consumer_artifact( file.write(json.dumps(row, sort_keys=True)) file.write("\n") + artifact_schema_version = schema_id("consumer_artifact", emit_epoch) manifest = { - "schema_version": CONSUMER_ARTIFACT_SCHEMA_VERSION, + "schema_version": artifact_schema_version, "consumer_fact_schema_versions": sorted( {row.get("schema_version") for row in rows} ), @@ -93,7 +104,7 @@ def build_consumer_artifact( _write_json(output_path / "manifest.json", manifest) return ConsumerArtifactBuildReport( - schema_version=CONSUMER_ARTIFACT_SCHEMA_VERSION, + schema_version=artifact_schema_version, output_dir=str(output_path), fact_row_count=len(rows), ) @@ -103,16 +114,37 @@ def load_consumer_artifact(path: str | Path) -> ConsumerArtifact: """Load a facts-only consumer artifact and verify its manifest hashes.""" artifact_path = Path(path) manifest = json.loads((artifact_path / "manifest.json").read_text()) - if manifest.get("schema_version") != CONSUMER_ARTIFACT_SCHEMA_VERSION: + manifest_schema_version = manifest.get("schema_version") + try: + SCHEMA_IDS["consumer_artifact"].infer_identifier_epoch(manifest_schema_version) + except ValueError as error: raise ValueError( "Unsupported consumer artifact schema_version: " - f"{manifest.get('schema_version')!r}." - ) + f"{manifest_schema_version!r}; accepted forms are " + f"{SCHEMA_IDS['consumer_artifact'].ledger!r} and " + f"{SCHEMA_IDS['consumer_artifact'].chronicle!r}." + ) from error if "profiles" in manifest: raise ValueError( "Consumer artifact manifests must not contain profiles; target profiles " "are consumer-owned contracts and must be loaded by Microcosm." ) + declared_fact_schema_versions = manifest.get("consumer_fact_schema_versions") + if declared_fact_schema_versions is not None: + if not isinstance(declared_fact_schema_versions, list): + raise ValueError( + "Consumer artifact consumer_fact_schema_versions must be a list." + ) + for row_schema_version in declared_fact_schema_versions: + try: + SCHEMA_IDS["consumer_fact"].infer_identifier_epoch(row_schema_version) + except ValueError as error: + raise ValueError( + "Unsupported consumer fact schema_version in artifact " + f"manifest: {row_schema_version!r}; accepted forms are " + f"{SCHEMA_IDS['consumer_fact'].ledger!r} and " + f"{SCHEMA_IDS['consumer_fact'].chronicle!r}." + ) from error manifest_schema_sha256 = manifest.get("consumer_fact_schema_sha256") if ( manifest_schema_sha256 is not None @@ -132,6 +164,16 @@ def load_consumer_artifact(path: str | Path) -> ConsumerArtifact: f"{actual_sha256} != {manifest['facts_sha256']}." ) rows = _load_consumer_rows(facts_file, validate_schema=True) + actual_fact_schema_versions = sorted({row.get("schema_version") for row in rows}) + if ( + declared_fact_schema_versions is not None + and declared_fact_schema_versions != actual_fact_schema_versions + ): + raise ValueError( + "Consumer artifact manifest declares consumer_fact_schema_versions " + f"{declared_fact_schema_versions!r} but its rows use " + f"{actual_fact_schema_versions!r}." + ) declared_row_count = manifest.get("fact_row_count") if declared_row_count is not None and declared_row_count != len(rows): raise ValueError( @@ -179,20 +221,32 @@ def _assert_finite_numbers(value: Any, *, line_number: int, path: Path) -> None: def _recompute_aggregate_fact_key(row: dict[str, Any]) -> str: """Recompute the aggregate fact key from the row's content.""" + declared_key = row.get("aggregate_fact_key") + epoch = HASH_DOMAINS["aggregate_fact"].infer_key_epoch(declared_key) assertion = row.get("assertion") payload = { - "source_release_key": row.get("source_release_key"), - "source_series_key": row.get("source_series_key"), - "observed_measure_key": row.get("observed_measure_key"), + "source_release_key": canonicalize_key( + "source_release", row.get("source_release_key") + ), + "source_series_key": canonicalize_key( + "source_series", row.get("source_series_key") + ), + "observed_measure_key": canonicalize_key( + "observed_measure", row.get("observed_measure_key") + ), "aggregation": row.get("aggregation"), "period": row.get("period"), "geography": row.get("geography"), "entity": row.get("entity"), - "dimension_set_key": row.get("dimension_set_key"), - "universe_constraint_set_key": row.get("universe_constraint_set_key"), + "dimension_set_key": canonicalize_key( + "dimension_set", row.get("dimension_set_key") + ), + "universe_constraint_set_key": canonicalize_key( + "universe_constraint_set", row.get("universe_constraint_set_key") + ), "assertion": None if assertion == DEFAULT_ASSERTION else assertion, } - return _hash_key("ledger.aggregate_fact.v2", payload) + return _hash_key(hash_domain("aggregate_fact", epoch), payload) def _load_consumer_rows( @@ -226,16 +280,23 @@ def _load_consumer_rows( f"{key!r} but its content hashes to {recomputed!r}; the " "identity key does not match the row." ) - if key in seen_keys: + canonical_key = canonicalize_key("aggregate_fact", key) + if canonical_key in seen_keys: raise ValueError( f"Row {line_number} of {path} repeats aggregate_fact_key " f"{key!r}; consumer artifact fact rows must be unique." ) - seen_keys.add(key) + seen_keys.add(canonical_key) rows.append(row) return rows +def load_consumer_rows(path: str | Path) -> tuple[dict[str, Any], ...]: + """Load and verify consumer-fact JSONL rows from either accepted epoch.""" + + return tuple(_load_consumer_rows(Path(path), validate_schema=True)) + + def _validate_consumer_row_provenance( row: Mapping[str, Any], *, @@ -287,4 +348,5 @@ def _sha256_file(path: Path) -> str: "ConsumerArtifactBuildReport", "build_consumer_artifact", "load_consumer_artifact", + "load_consumer_rows", ] diff --git a/policyengine_chronicle/schema.py b/policyengine_chronicle/schema.py index 381dbd1..4f29bfc 100644 --- a/policyengine_chronicle/schema.py +++ b/policyengine_chronicle/schema.py @@ -9,17 +9,30 @@ from __future__ import annotations +from copy import deepcopy import hashlib import json from functools import lru_cache from importlib.resources import files as _resource_files from typing import Any +from chronicle.epoch import SCHEMA_IDS, canonicalize_key from jsonschema import Draft202012Validator _SCHEMA_PACKAGE = "policyengine_chronicle.schemas" _SCHEMA_RESOURCE = "consumer_fact.v1.schema.json" +_TOP_LEVEL_KEY_DOMAINS = { + "aggregate_fact_key": "aggregate_fact", + "semantic_fact_key": "semantic_fact", + "legacy_fact_key": "fact", + "source_release_key": "source_release", + "source_series_key": "source_series", + "observed_measure_key": "observed_measure", + "dimension_set_key": "dimension_set", + "universe_constraint_set_key": "universe_constraint_set", +} + def _packaged_schema_bytes() -> bytes: return _resource_files(_SCHEMA_PACKAGE).joinpath(_SCHEMA_RESOURCE).read_bytes() @@ -39,6 +52,183 @@ def _validator() -> Draft202012Validator: return Draft202012Validator(consumer_fact_schema()) +def _epoch_validation_error( + *, + line_number: int, + path: Any, + location: str, + error: ValueError, +) -> ValueError: + return ValueError( + f"Consumer fact row {line_number} of {path} failed epoch validation " + f"at {location!r}: {error}" + ) + + +def _normalize_key( + key: Any, + *, + domain_name: str, + line_number: int, + path: Any, + location: str, +) -> Any: + # Leave type errors to the frozen JSON schema so its established messages + # stay stable. String identifiers receive the additive epoch check first. + if not isinstance(key, str): + return key + try: + return canonicalize_key(domain_name, key) + except ValueError as error: + raise _epoch_validation_error( + line_number=line_number, + path=path, + location=location, + error=error, + ) from error + + +def normalize_consumer_fact_row_epochs( + row: Any, + line_number: int, + path: Any, +) -> Any: + """Return a Ledger-canonical copy of a dual-epoch consumer-fact row. + + The frozen v1 JSON schema remains the byte-identical validation contract. + This adapter accepts either registered naming epoch on each identifier, + independently, then normalizes only the copy passed to that schema. The + caller's row is never mutated, and mixed-epoch rows remain valid. + """ + + normalized = deepcopy(row) + if not isinstance(normalized, dict): + return normalized + + schema_version = normalized.get("schema_version") + if isinstance(schema_version, str): + pair = SCHEMA_IDS["consumer_fact"] + try: + pair.infer_identifier_epoch(schema_version) + except ValueError as error: + raise _epoch_validation_error( + line_number=line_number, + path=path, + location="schema_version", + error=error, + ) from error + normalized["schema_version"] = pair.ledger + + for field_name, domain_name in _TOP_LEVEL_KEY_DOMAINS.items(): + if field_name in normalized: + normalized[field_name] = _normalize_key( + normalized[field_name], + domain_name=domain_name, + line_number=line_number, + path=path, + location=field_name, + ) + + concept_alignment = normalized.get("concept_alignment") + if ( + isinstance(concept_alignment, dict) + and "concept_alignment_key" in concept_alignment + ): + concept_alignment["concept_alignment_key"] = _normalize_key( + concept_alignment["concept_alignment_key"], + domain_name="concept_alignment", + line_number=line_number, + path=path, + location="concept_alignment/concept_alignment_key", + ) + + lineage = normalized.get("lineage") + if isinstance(lineage, dict): + for field_name, domain_name in ( + ("source_cell_keys", "source_cell"), + ("source_row_keys", "source_row"), + ): + keys = lineage.get(field_name) + if not isinstance(keys, list): + continue + for index, key in enumerate(keys): + keys[index] = _normalize_key( + key, + domain_name=domain_name, + line_number=line_number, + path=path, + location=f"lineage/{field_name}/{index}", + ) + + return normalized + + +def validate_consumer_fact_row_epochs( + row: Any, + line_number: int, + path: Any, +) -> None: + """Check every epoch-bearing identifier of a row without copying it. + + This is :func:`normalize_consumer_fact_row_epochs` minus the deep copy and + the rewrite: it raises the same error for the same identifier and leaves + the caller's row untouched, so a bundle build can validate ~150k rows + without materializing a canonical copy of each. + """ + + if not isinstance(row, dict): + return + schema_version = row.get("schema_version") + if isinstance(schema_version, str): + try: + SCHEMA_IDS["consumer_fact"].infer_identifier_epoch(schema_version) + except ValueError as error: + raise _epoch_validation_error( + line_number=line_number, + path=path, + location="schema_version", + error=error, + ) from error + for field_name, domain_name in _TOP_LEVEL_KEY_DOMAINS.items(): + if field_name in row: + _normalize_key( + row[field_name], + domain_name=domain_name, + line_number=line_number, + path=path, + location=field_name, + ) + concept_alignment = row.get("concept_alignment") + if ( + isinstance(concept_alignment, dict) + and "concept_alignment_key" in concept_alignment + ): + _normalize_key( + concept_alignment["concept_alignment_key"], + domain_name="concept_alignment", + line_number=line_number, + path=path, + location="concept_alignment/concept_alignment_key", + ) + lineage = row.get("lineage") + if isinstance(lineage, dict): + for field_name, domain_name in ( + ("source_cell_keys", "source_cell"), + ("source_row_keys", "source_row"), + ): + keys = lineage.get(field_name) + if not isinstance(keys, list): + continue + for index, key in enumerate(keys): + _normalize_key( + key, + domain_name=domain_name, + line_number=line_number, + path=path, + location=f"lineage/{field_name}/{index}", + ) + + def validate_consumer_fact_row( row: Any, line_number: int, @@ -50,8 +240,9 @@ def validate_consumer_fact_row( ``line_number``, the failing JSON location, and the schema reason. The first error by schema location is reported so the message is stable. """ + normalized = normalize_consumer_fact_row_epochs(row, line_number, path) errors = sorted( - _validator().iter_errors(row), + _validator().iter_errors(normalized), key=lambda error: ( [str(part) for part in error.absolute_path], error.message, @@ -70,5 +261,7 @@ def validate_consumer_fact_row( __all__ = [ "CONSUMER_FACT_SCHEMA_SHA256", "consumer_fact_schema", + "normalize_consumer_fact_row_epochs", "validate_consumer_fact_row", + "validate_consumer_fact_row_epochs", ] diff --git a/tests/test_belgium_targets.py b/tests/test_belgium_targets.py index afec1b2..b1241f9 100644 --- a/tests/test_belgium_targets.py +++ b/tests/test_belgium_targets.py @@ -8,7 +8,10 @@ from functools import lru_cache import hashlib import json +import os from pathlib import Path +import subprocess +import sys import pytest import yaml @@ -120,6 +123,61 @@ def test_statbel_fiscal_distribution_alias_resolves(): ) +def test_statbel_fiscal_distribution_generator_keeps_ledger_schema_bytes(): + payload = yaml.safe_load( + ( + REPO_ROOT + / "packages" + / "statbel" + / "fiscal_income_distribution_2023" + / "source_package.yaml" + ).read_text() + ) + + assert payload["schema_version"] == "ledger.source_package.v1" + + +def test_statbel_fiscal_distribution_generator_runs_as_direct_file(tmp_path): + relative_script = Path( + "packages/statbel/fiscal_income_distribution_2023/build_package.py" + ) + checkout = tmp_path / "checkout" + script = checkout / relative_script + script.parent.mkdir(parents=True) + script.write_bytes((REPO_ROOT / relative_script).read_bytes()) + + chronicle_dir = checkout / "chronicle" + chronicle_dir.mkdir() + for filename in ("__init__.py", "epoch.py"): + (chronicle_dir / filename).write_bytes( + (REPO_ROOT / "chronicle" / filename).read_bytes() + ) + + openpyxl_dir = script.parent / "openpyxl" + openpyxl_dir.mkdir() + (openpyxl_dir / "__init__.py").write_text("") + (openpyxl_dir / "utils.py").write_text( + "def get_column_letter(index):\n return str(index)\n" + ) + (script.parent / "yaml.py").write_text("") + + environment = os.environ.copy() + environment["PYTHONPATH"] = "" + result = subprocess.run( + [sys.executable, "-S", str(script)], + cwd=tmp_path, + env=environment, + capture_output=True, + text=True, + timeout=30, + check=False, + ) + + assert result.returncode == 1 + assert "FileNotFoundError: Missing staged Statbel workbooks" in result.stderr + assert "ModuleNotFoundError" not in result.stderr + + def test_statbel_fiscal_distribution_artifact_pins_and_r2_keys(): data_dir = REPO_ROOT / "db" / "data" / "statbel" / "fiscal_income_distribution_2023" manifest = yaml.safe_load((data_dir / "manifest.yaml").read_text()) @@ -516,9 +574,7 @@ def test_belgium_euromod_comparator_pairs_c2_rows_with_model_outputs(): missing_ratios = { row["value_id"] for row in c2_rows - if row["value_id"] - .replace("_external_", "_ratio_") - .replace("_silc_", "_ratio_") + if row["value_id"].replace("_external_", "_ratio_").replace("_silc_", "_ratio_") not in ids } assert not missing_ratios diff --git a/tests/test_chronicle_artifacts.py b/tests/test_chronicle_artifacts.py index e08beab..480b66c 100644 --- a/tests/test_chronicle_artifacts.py +++ b/tests/test_chronicle_artifacts.py @@ -10,6 +10,7 @@ from chronicle.cli import main as cli_main from chronicle.artifacts import ( + build_artifact_key, build_artifact_rows, build_derived_r2_key, bootstrap_r2_buckets, @@ -21,6 +22,7 @@ publish_derived_artifacts, publish_source_artifacts, ) +from chronicle.epoch import Epoch from chronicle.harness import main as harness_main @@ -51,6 +53,39 @@ def test_build_derived_r2_key_is_build_scoped(): ) +def test_build_artifact_key_hashes_one_payload_across_epochs(): + ledger = build_artifact_key( + build_id="ledger.build.v1:abc123", + artifact_name="reports/build_summary.json", + sha256="def456", + ) + chronicle = build_artifact_key( + build_id="chronicle.build.v2:abc123", + artifact_name="reports/build_summary.json", + sha256="def456", + epoch=Epoch.CHRONICLE, + ) + + assert ledger.startswith("ledger.build_artifact.v1:") + assert chronicle.startswith("chronicle.build_artifact.v2:") + assert ledger.split(":", maxsplit=1)[1] == chronicle.split(":", maxsplit=1)[1] + + +def test_build_artifact_key_rejects_unknown_build_epoch(): + with pytest.raises( + ValueError, + match=( + "ledger[.]build[.]v1.*chronicle[.]build[.]v2|" + "chronicle[.]build[.]v2.*ledger[.]build[.]v1" + ), + ): + build_artifact_key( + build_id="future.build.v9:abc123", + artifact_name="facts.jsonl", + sha256="def456", + ) + + @pytest.mark.parametrize( ("source_id", "package_path", "expected_country"), [ @@ -489,6 +524,47 @@ def test_publish_derived_artifacts_uploads_build_directory(tmp_path): assert "reports/build_summary.json" in command_log +@pytest.mark.parametrize("write_build_artifacts", [False, True]) +def test_publish_derived_artifacts_rejects_unknown_build_before_side_effects( + tmp_path, + write_build_artifacts, +): + suite = tmp_path / "suite" + reports = suite / "reports" + reports.mkdir(parents=True) + (reports / "database.json").write_text( + json.dumps({"build_id": "future.build.v9:invalid"}) + ) + (suite / "facts.jsonl").write_text("{}\n") + upload_log = tmp_path / "wrangler.log" + wrangler = tmp_path / "wrangler" + wrangler.write_text(f"#!/bin/sh\nprintf '%s\\n' \"$*\" >> {upload_log}\necho ok\n") + wrangler.chmod(0o755) + build_artifacts_path = tmp_path / "build_artifacts.jsonl" + build_artifacts_path.write_text("sentinel\n") + + with pytest.raises( + ValueError, + match=( + "ledger[.]build[.]v1.*chronicle[.]build[.]v2|" + "chronicle[.]build[.]v2.*ledger[.]build[.]v1" + ), + ): + publish_derived_artifacts( + suite, + source_id="irs_soi", + package_id="soi-table-1-1", + year=2023, + build_artifacts_output=( + build_artifacts_path if write_build_artifacts else None + ), + wrangler_command=str(wrangler), + ) + + assert not upload_log.exists() + assert build_artifacts_path.read_text() == "sentinel\n" + + def test_build_artifact_rows_skips_failed_uploads(tmp_path): suite = tmp_path / "suite" reports = suite / "reports" diff --git a/tests/test_chronicle_bundle.py b/tests/test_chronicle_bundle.py index 3a9e14f..da217e3 100644 --- a/tests/test_chronicle_bundle.py +++ b/tests/test_chronicle_bundle.py @@ -3,8 +3,20 @@ from __future__ import annotations import json +from pathlib import Path -from chronicle.bundle import UK_BUNDLE_SOURCES, build_bundle, build_bundle_coverage +import pytest + +from chronicle.bundle import ( + BUNDLE_COVERAGE_SCHEMA_VERSION, + BUNDLE_SCHEMA_VERSION, + BUNDLE_SOURCES_SCHEMA_VERSION, + UK_BUNDLE_SOURCES, + _load_jsonl as load_bundle_jsonl, + build_bundle, + build_bundle_coverage, +) +from chronicle.epoch import HASH_DOMAINS, SCHEMA_IDS, Epoch from chronicle.harness import build_bundle_dir from chronicle.harness import main as harness_main @@ -13,6 +25,44 @@ def _load_jsonl(path): return [json.loads(line) for line in path.read_text().splitlines() if line] +def _fixture_consumer_rows(): + path = Path(__file__).parents[1] / "chronicle" / "fixtures" / "consumer_facts.jsonl" + return _load_jsonl(path) + + +def _row_for_epoch(row, epoch): + transformed = json.loads(json.dumps(row)) + transformed["schema_version"] = SCHEMA_IDS["consumer_fact"].for_epoch(epoch) + for field_name, domain_name in ( + ("aggregate_fact_key", "aggregate_fact"), + ("semantic_fact_key", "semantic_fact"), + ("legacy_fact_key", "fact"), + ("source_release_key", "source_release"), + ("source_series_key", "source_series"), + ("observed_measure_key", "observed_measure"), + ("dimension_set_key", "dimension_set"), + ("universe_constraint_set_key", "universe_constraint_set"), + ): + transformed[field_name] = HASH_DOMAINS[domain_name].key_for_epoch( + transformed[field_name], epoch + ) + alignment = transformed.get("concept_alignment") + if alignment: + alignment["concept_alignment_key"] = HASH_DOMAINS[ + "concept_alignment" + ].key_for_epoch(alignment["concept_alignment_key"], epoch) + lineage = transformed["lineage"] + for field_name, domain_name in ( + ("source_cell_keys", "source_cell"), + ("source_row_keys", "source_row"), + ): + lineage[field_name] = [ + HASH_DOMAINS[domain_name].key_for_epoch(key, epoch) + for key in lineage[field_name] + ] + return transformed + + def test_build_bundle_dir_uk_suite_uses_curated_sources(tmp_path, monkeypatch): captured = {} @@ -1154,3 +1204,122 @@ def test_build_bundle_coverage_reports_duplicate_keys(): assert coverage["counts"]["by_source"] == {"irs_soi": 2} assert coverage["duplicates"]["aggregate_fact_keys"][0]["count"] == 2 assert coverage["duplicates"]["semantic_fact_keys"][0]["count"] == 2 + + +def test_bundle_jsonl_ingestion_accepts_chronicle_only_rows(tmp_path): + row = _row_for_epoch(_fixture_consumer_rows()[0], Epoch.CHRONICLE) + path = tmp_path / "consumer_facts.jsonl" + path.write_text(json.dumps(row, sort_keys=True) + "\n") + + loaded = load_bundle_jsonl(path) + + assert loaded == [row] + assert loaded[0]["schema_version"] == "chronicle.consumer_fact.v2" + assert loaded[0]["aggregate_fact_key"].startswith("chronicle.aggregate_fact.v3:") + + +def test_bundle_jsonl_ingestion_accepts_mixed_epoch_rows(tmp_path): + ledger_row, chronicle_source = _fixture_consumer_rows()[:2] + chronicle_row = _row_for_epoch(chronicle_source, Epoch.CHRONICLE) + path = tmp_path / "consumer_facts.jsonl" + path.write_text( + "".join( + json.dumps(row, sort_keys=True) + "\n" + for row in (ledger_row, chronicle_row) + ) + ) + + loaded = load_bundle_jsonl(path) + + assert loaded == [ledger_row, chronicle_row] + assert {row["schema_version"] for row in loaded} == { + "ledger.consumer_fact.v1", + "chronicle.consumer_fact.v2", + } + + +def test_bundle_jsonl_ingestion_rejects_unknown_key_domain(tmp_path): + row = _row_for_epoch(_fixture_consumer_rows()[0], Epoch.CHRONICLE) + digest = row["aggregate_fact_key"].partition(":")[2] + row["aggregate_fact_key"] = f"future.aggregate_fact.v4:{digest}" + path = tmp_path / "consumer_facts.jsonl" + path.write_text(json.dumps(row, sort_keys=True) + "\n") + + with pytest.raises(ValueError) as error: + load_bundle_jsonl(path) + + message = str(error.value) + assert "ledger.aggregate_fact.v2" in message + assert "chronicle.aggregate_fact.v3" in message + + +def test_bundle_coverage_canonicalizes_cross_epoch_identities(): + ledger_row = _fixture_consumer_rows()[0] + chronicle_row = _row_for_epoch(ledger_row, Epoch.CHRONICLE) + + coverage = build_bundle_coverage([ledger_row, chronicle_row]) + + assert BUNDLE_SCHEMA_VERSION == "ledger.bundle.v1" + assert BUNDLE_COVERAGE_SCHEMA_VERSION == "ledger.bundle_coverage.v1" + assert BUNDLE_SOURCES_SCHEMA_VERSION == "ledger.bundle_sources.v1" + assert coverage["unique_counts"] == { + "aggregate_fact_key": 1, + "semantic_fact_key": 1, + "source_release_key": 1, + "source_series_key": 1, + "observed_measure_key": 1, + "dimension_set_key": 1, + "universe_constraint_set_key": 1, + } + assert coverage["duplicates"]["aggregate_fact_keys"] == [ + { + "key": ledger_row["aggregate_fact_key"], + "count": 2, + "sources": ["irs_soi:Publication 1304 Table 1.1"], + "legacy_fact_keys": [ledger_row["legacy_fact_key"]], + } + ] + + +def test_bundle_coverage_preserves_non_string_identity_scalars(tmp_path): + identity_fields = ( + "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", + ) + rows = [_fixture_consumer_rows()[0]] + for value in (None, 7): + row = json.loads(json.dumps(rows[0])) + for field_name in identity_fields: + row[field_name] = value + rows.append(row) + path = tmp_path / "consumer_facts.jsonl" + path.write_text("".join(json.dumps(row, sort_keys=True) + "\n" for row in rows)) + + loaded = load_bundle_jsonl(path) + coverage = build_bundle_coverage(loaded) + + for field_name in identity_fields: + assert [row[field_name] for row in loaded] == [ + rows[0][field_name], + None, + 7, + ] + assert coverage["unique_counts"] == { + "aggregate_fact_key": 3, + "semantic_fact_key": 3, + "source_release_key": 3, + "source_series_key": 3, + "observed_measure_key": 3, + "dimension_set_key": 3, + "universe_constraint_set_key": 3, + } + assert coverage["duplicates"] == { + "aggregate_fact_keys": [], + "semantic_fact_keys": [], + } diff --git a/tests/test_chronicle_consumer.py b/tests/test_chronicle_consumer.py index a733471..e6477b6 100644 --- a/tests/test_chronicle_consumer.py +++ b/tests/test_chronicle_consumer.py @@ -19,6 +19,7 @@ SourceRecordLayout, ) from chronicle.harness import main +from chronicle.epoch import Epoch from policyengine_chronicle.consumer import ( build_consumer_artifact, load_consumer_artifact, @@ -81,6 +82,12 @@ def _write_facts(tmp_path): return facts_path +def _write_rows(path, rows): + with path.open("w") as file: + for row in rows: + file.write(json.dumps(row, sort_keys=True) + "\n") + + def _rewrite_manifest_hash(out_dir): facts_file = out_dir / "consumer_facts.jsonl" manifest_path = out_dir / "manifest.json" @@ -131,6 +138,50 @@ def test_artifact_is_reproducible(tmp_path): assert (first / name).read_bytes() == (second / name).read_bytes() +def test_chronicle_epoch_artifact_loads_end_to_end(tmp_path): + facts_path = tmp_path / "chronicle-consumer-facts.jsonl" + rows = consumer_fact_rows( + [_fact(value=100, period_value=2021)], emit_epoch=Epoch.CHRONICLE + ) + _write_rows(facts_path, rows) + out_dir = tmp_path / "chronicle-artifact" + + report = build_consumer_artifact( + out_dir, + facts_path=facts_path, + emit_epoch=Epoch.CHRONICLE, + ) + artifact = load_consumer_artifact(out_dir) + + assert report.schema_version == "policyengine_chronicle.consumer_artifact.v3" + assert artifact.manifest["schema_version"] == ( + "policyengine_chronicle.consumer_artifact.v3" + ) + assert artifact.rows[0]["schema_version"] == "chronicle.consumer_fact.v2" + assert artifact.rows[0]["aggregate_fact_key"].startswith( + "chronicle.aggregate_fact.v3:" + ) + + +def test_mixed_epoch_rows_load_in_one_artifact(tmp_path): + ledger_row = consumer_fact_rows([_fact(value=100, period_value=2021)])[0] + chronicle_row = consumer_fact_rows( + [_fact(value=110, period_value=2022)], emit_epoch=Epoch.CHRONICLE + )[0] + facts_path = tmp_path / "mixed-consumer-facts.jsonl" + _write_rows(facts_path, [ledger_row, chronicle_row]) + out_dir = tmp_path / "mixed-artifact" + + build_consumer_artifact(out_dir, facts_path=facts_path) + artifact = load_consumer_artifact(out_dir) + + assert artifact.manifest["consumer_fact_schema_versions"] == [ + "chronicle.consumer_fact.v2", + "ledger.consumer_fact.v1", + ] + assert len(artifact.rows) == 2 + + def test_artifact_load_rejects_tampered_facts(tmp_path): facts_path = _write_facts(tmp_path) out_dir = tmp_path / "artifact" @@ -176,6 +227,53 @@ def test_artifact_load_rejects_legacy_v1_schema(tmp_path): load_consumer_artifact(out_dir) +def test_artifact_load_unknown_schema_names_both_accepted_forms(tmp_path): + facts_path = _write_facts(tmp_path) + out_dir = tmp_path / "artifact" + build_consumer_artifact(out_dir, facts_path=facts_path) + manifest_path = out_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["schema_version"] = "future.consumer_artifact.v9" + manifest_path.write_text(json.dumps(manifest, sort_keys=True, indent=2) + "\n") + + with pytest.raises(ValueError) as error: + load_consumer_artifact(out_dir) + + message = str(error.value) + assert "policyengine_ledger.consumer_artifact.v2" in message + assert "policyengine_chronicle.consumer_artifact.v3" in message + + +def test_artifact_load_unknown_row_schema_names_both_accepted_forms(tmp_path): + facts_path = _write_facts(tmp_path) + out_dir = tmp_path / "artifact" + build_consumer_artifact(out_dir, facts_path=facts_path) + manifest_path = out_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["consumer_fact_schema_versions"] = ["future.consumer_fact.v9"] + manifest_path.write_text(json.dumps(manifest, sort_keys=True, indent=2) + "\n") + + with pytest.raises(ValueError) as error: + load_consumer_artifact(out_dir) + + message = str(error.value) + assert "ledger.consumer_fact.v1" in message + assert "chronicle.consumer_fact.v2" in message + + +def test_artifact_load_rejects_declared_row_schema_drift(tmp_path): + facts_path = _write_facts(tmp_path) + out_dir = tmp_path / "artifact" + build_consumer_artifact(out_dir, facts_path=facts_path) + manifest_path = out_dir / "manifest.json" + manifest = json.loads(manifest_path.read_text()) + manifest["consumer_fact_schema_versions"] = ["chronicle.consumer_fact.v2"] + manifest_path.write_text(json.dumps(manifest, sort_keys=True, indent=2) + "\n") + + with pytest.raises(ValueError, match="but its rows use"): + load_consumer_artifact(out_dir) + + @pytest.mark.parametrize( ("case", "message"), [ diff --git a/tests/test_chronicle_consumer_contract.py b/tests/test_chronicle_consumer_contract.py index 8dd13e4..23e62ff 100644 --- a/tests/test_chronicle_consumer_contract.py +++ b/tests/test_chronicle_consumer_contract.py @@ -30,6 +30,7 @@ build_aggregate_constraints, validate_facts, ) +from chronicle.epoch import HASH_DOMAINS, SCHEMA_IDS, Epoch from chronicle.harness import main from chronicle.jurisdictions.us.soi import build_soi_table_1_1_facts from chronicle.store import save_facts_jsonl @@ -170,6 +171,134 @@ def test_consumer_fact_row_exposes_chronicle_and_lineage_keys(): assert row["lineage"]["source_cell_keys"] +@pytest.mark.parametrize( + ("domain_name", "builder_name"), + [ + ("source_release", "build_source_release_key"), + ("source_series", "build_source_series_key"), + ("observed_measure", "build_observed_measure_key"), + ("dimension_set", "build_dimension_set_key"), + ("universe_constraint_set", "build_universe_constraint_set_key"), + ("aggregate_fact", "build_aggregate_fact_key"), + ("semantic_fact", "build_semantic_fact_key"), + ("concept_alignment", "build_concept_alignment_key"), + ], +) +def test_consumer_key_epochs_change_only_the_hash_domain( + domain_name, + builder_name, +): + """The epoch is domain separation, not a canonical-payload migration.""" + builder = getattr(consumer_contract, builder_name) + + for fact in (_soi_agi_fact(), _soi_agi_bracket_fact()): + ledger_key = builder(fact, epoch=Epoch.LEDGER) + chronicle_key = builder(fact, epoch=Epoch.CHRONICLE) + pair = HASH_DOMAINS[domain_name] + + assert ledger_key.startswith(f"{pair.ledger}:") + assert chronicle_key.startswith(f"{pair.chronicle}:") + assert ledger_key.partition(":")[2] == chronicle_key.partition(":")[2] + + +def test_chronicle_epoch_consumer_row_uses_successor_ids_consistently(): + fact = replace( + _soi_agi_fact(), + source_row_keys=("ledger.source_row.v1:source-row",), + ) + + ledger_row = consumer_fact_row(fact) + chronicle_row = consumer_fact_row(fact, emit_epoch=Epoch.CHRONICLE) + + assert ledger_row["schema_version"] == SCHEMA_IDS["consumer_fact"].ledger + assert chronicle_row["schema_version"] == SCHEMA_IDS["consumer_fact"].chronicle + key_fields = { + "aggregate_fact_key": "aggregate_fact", + "semantic_fact_key": "semantic_fact", + "legacy_fact_key": "fact", + "source_release_key": "source_release", + "source_series_key": "source_series", + "observed_measure_key": "observed_measure", + "dimension_set_key": "dimension_set", + "universe_constraint_set_key": "universe_constraint_set", + } + for field, domain_name in key_fields.items(): + pair = HASH_DOMAINS[domain_name] + assert ledger_row[field].startswith(f"{pair.ledger}:") + assert chronicle_row[field].startswith(f"{pair.chronicle}:") + assert ( + ledger_row[field].partition(":")[2] + == (chronicle_row[field].partition(":")[2]) + ) + + ledger_alignment = ledger_row["concept_alignment"]["concept_alignment_key"] + chronicle_alignment = chronicle_row["concept_alignment"]["concept_alignment_key"] + assert ledger_alignment.startswith(f"{HASH_DOMAINS['concept_alignment'].ledger}:") + assert chronicle_alignment.startswith( + f"{HASH_DOMAINS['concept_alignment'].chronicle}:" + ) + assert ledger_alignment.partition(":")[2] == chronicle_alignment.partition(":")[2] + + assert chronicle_row["lineage"]["source_cell_keys"] == [ + HASH_DOMAINS["source_cell"].key_for_epoch( + key, + Epoch.CHRONICLE, + ) + for key in fact.source_cell_keys + ] + assert chronicle_row["lineage"]["source_row_keys"] == [ + "chronicle.source_row.v2:source-row" + ] + + +@pytest.mark.parametrize("emit_epoch", [Epoch.LEDGER, Epoch.CHRONICLE]) +def test_consumer_row_defensively_deduplicates_lineage_aliases(emit_epoch): + fact = _soi_agi_fact() + source_cell_key = fact.source_cell_keys[0] + source_row_key = "ledger.source_row.v1:source-row" + fact = replace( + fact, + source_cell_keys=( + source_cell_key, + HASH_DOMAINS["source_cell"].key_for_epoch( + source_cell_key, + Epoch.CHRONICLE, + ), + ), + source_row_keys=( + source_row_key, + HASH_DOMAINS["source_row"].key_for_epoch( + source_row_key, + Epoch.CHRONICLE, + ), + ), + ) + + row = consumer_fact_row(fact, emit_epoch=emit_epoch) + + assert row["lineage"]["source_cell_keys"] == [ + HASH_DOMAINS["source_cell"].key_for_epoch(source_cell_key, emit_epoch) + ] + assert row["lineage"]["source_row_keys"] == [ + HASH_DOMAINS["source_row"].key_for_epoch(source_row_key, emit_epoch) + ] + + +def test_chronicle_epoch_writer_reports_successor_schema(tmp_path): + output = tmp_path / "consumer_facts.jsonl" + + report = write_consumer_facts_jsonl( + [_soi_agi_fact()], + output, + emit_epoch=Epoch.CHRONICLE, + ) + row = json.loads(output.read_text()) + + assert report.schema_version == SCHEMA_IDS["consumer_fact"].chronicle + assert row["schema_version"] == SCHEMA_IDS["consumer_fact"].chronicle + assert row["aggregate_fact_key"].startswith("chronicle.aggregate_fact.v3:") + + def test_aggregate_fact_key_ignores_lineage_labels_and_evidence_notes(): fact = _soi_agi_fact() changed = replace( @@ -766,3 +895,14 @@ def test_export_consumer_facts_cli_preserves_decimal_values(tmp_path, capsys): assert json.loads(capsys.readouterr().out)["valid"] assert row["value"] == "1.25" assert row["value_type"] == "decimal" + + +def test_contract_reports_malformed_lineage_keys_instead_of_raising(tmp_path): + fact = replace(_soi_agi_fact(), source_cell_keys=("bogus.domain.v9:" + "a" * 24,)) + report = validate_consumer_fact_contract([fact]) + codes = {issue.code for issue in report.errors} + assert "malformed_lineage_key" in codes + with pytest.raises( + ValueError, match="Cannot export invalid Chronicle consumer-contract facts" + ): + write_consumer_facts_jsonl([fact], tmp_path / "facts.jsonl") diff --git a/tests/test_chronicle_core.py b/tests/test_chronicle_core.py index 8828ea4..37cd0f4 100644 --- a/tests/test_chronicle_core.py +++ b/tests/test_chronicle_core.py @@ -2,6 +2,10 @@ from __future__ import annotations +import hashlib +import json +from pathlib import Path + from chronicle.core import ( Aggregation, EntityDimension, @@ -16,6 +20,22 @@ validate_fact, validate_facts, ) +from chronicle.epoch import EMIT_EPOCH, HASH_DOMAINS, SCHEMA_IDS, Epoch +from chronicle.harness import main as harness_main +from chronicle.sources.cells import ( + SourceArtifactMetadata, + SourceCell, + build_source_cell_key, +) +from chronicle.sources.rows import ( + SourceColumn, + SourceRow, + SourceRowValue, + build_source_column_key, + build_source_row_key, + build_source_row_value_key, +) +from chronicle.store import save_facts_jsonl def _fact(**overrides): @@ -48,6 +68,20 @@ def _fact(**overrides): return AggregateFact(**{**fact.__dict__, **overrides}) +def _source_artifact() -> SourceArtifactMetadata: + return SourceArtifactMetadata( + source_name="test_publisher", + source_table="table", + source_file="table.csv", + url="https://example.test/table.csv", + vintage="2026", + sha256="a" * 64, + size_bytes=10, + extracted_at="2026-09-02", + extraction_method="test", + ) + + def test_valid_fact_passes_validation(): assert validate_fact(_fact()) == () assert validate_facts([_fact()]).valid @@ -73,7 +107,6 @@ def test_quantile_aggregation_passes_validation(): assert validate_fact(fact) == () - def test_stable_key_ignores_human_label(): fact = _fact() relabeled = _fact(label="A different display label") @@ -156,3 +189,275 @@ def test_label_generation_uses_metadata_not_key_path(): "for tax unit (filing status=all) " "[irs_soi Publication 1304 Table 1.1 23in11si.xls tax_year_2023]" ) + + +def test_epoch_registry_covers_frozen_domains_and_schema_ids(): + expected_hash_domains = { + "source_release": ("ledger.source_release.v2", "chronicle.source_release.v3"), + "source_series": ("ledger.source_series.v2", "chronicle.source_series.v3"), + "observed_measure": ( + "ledger.observed_measure.v2", + "chronicle.observed_measure.v3", + ), + "dimension_set": ("ledger.dimension_set.v2", "chronicle.dimension_set.v3"), + "universe_constraint_set": ( + "ledger.universe_constraint_set.v2", + "chronicle.universe_constraint_set.v3", + ), + "aggregate_fact": ("ledger.aggregate_fact.v2", "chronicle.aggregate_fact.v3"), + "semantic_fact": ("ledger.semantic_fact.v2", "chronicle.semantic_fact.v3"), + "concept_alignment": ( + "ledger.concept_alignment.v2", + "chronicle.concept_alignment.v3", + ), + "fact": ("ledger.fact.v1", "chronicle.fact.v2"), + "source_cell": ("ledger.source_cell.v1", "chronicle.source_cell.v2"), + "source_row": ("ledger.source_row.v1", "chronicle.source_row.v2"), + "source_column": ("ledger.source_column.v1", "chronicle.source_column.v2"), + "source_row_value": ( + "ledger.source_row_value.v1", + "chronicle.source_row_value.v2", + ), + "build": ("ledger.build.v1", "chronicle.build.v2"), + "build_artifact": ( + "ledger.build_artifact.v1", + "chronicle.build_artifact.v2", + ), + } + expected_schema_ids = { + "bundle": ("ledger.bundle.v1", "chronicle.bundle.v2"), + "bundle_coverage": ( + "ledger.bundle_coverage.v1", + "chronicle.bundle_coverage.v2", + ), + "bundle_sources": ("ledger.bundle_sources.v1", "chronicle.bundle_sources.v2"), + "consumer_fact": ("ledger.consumer_fact.v1", "chronicle.consumer_fact.v2"), + "relational": ("ledger.relational.v1", "chronicle.relational.v2"), + "source_package": ("ledger.source_package.v1", "chronicle.source_package.v2"), + "offline_fetch_manifest": ( + "ledger.offline_fetch_manifest.v1", + "chronicle.offline_fetch_manifest.v2", + ), + "fetch_manifest": ("ledger.fetch_manifest.v1", "chronicle.fetch_manifest.v2"), + "consumer_artifact": ( + "policyengine_ledger.consumer_artifact.v2", + "policyengine_chronicle.consumer_artifact.v3", + ), + "approved_agents": ( + "policyengine_ledger.approved_agents.v1", + "policyengine_chronicle.approved_agents.v2", + ), + } + + assert EMIT_EPOCH == Epoch.LEDGER + assert {name: pair.accepted for name, pair in HASH_DOMAINS.items()} == ( + expected_hash_domains + ) + assert {name: pair.accepted for name, pair in SCHEMA_IDS.items()} == ( + expected_schema_ids + ) + + +def test_epoch_registry_unknown_key_names_both_accepted_forms(): + pair = HASH_DOMAINS["fact"] + + try: + pair.infer_key_epoch("future.fact.v9:abc") + except ValueError as error: + message = str(error) + else: + raise AssertionError("unknown key domain was accepted") + + assert pair.ledger in message + assert pair.chronicle in message + + +def test_fact_key_epochs_hash_the_same_canonical_payload(): + ledger_key = build_fact_key(_fact(), epoch=Epoch.LEDGER) + chronicle_key = build_fact_key(_fact(), epoch=Epoch.CHRONICLE) + + assert build_fact_key(_fact()) == ledger_key + assert ledger_key.partition(":")[0] == "ledger.fact.v1" + assert chronicle_key.partition(":")[0] == "chronicle.fact.v2" + assert ledger_key.partition(":")[2] == chronicle_key.partition(":")[2] + + +def test_source_key_epochs_hash_the_same_canonical_payload(): + artifact = _source_artifact() + cell = SourceCell( + artifact=artifact, + sheet_name="Sheet1", + row_number=2, + column_number=3, + address="C2", + cell_type="number", + raw_value=42, + display_value="42", + ) + row = SourceRow( + artifact=artifact, + sheet_name="Sheet1", + row_number=2, + values={"amount": 42}, + ) + column = SourceColumn( + artifact=artifact, + sheet_name="Sheet1", + column_number=1, + raw_name="amount", + normalized_name="amount", + ) + cases = ( + (build_source_cell_key, cell, "source_cell"), + (build_source_row_key, row, "source_row"), + (build_source_column_key, column, "source_column"), + ) + + for builder, record, domain in cases: + ledger_key = builder(record, epoch=Epoch.LEDGER) + chronicle_key = builder(record, epoch=Epoch.CHRONICLE) + assert builder(record) == ledger_key + assert ledger_key.partition(":")[0] == HASH_DOMAINS[domain].ledger + assert chronicle_key.partition(":")[0] == HASH_DOMAINS[domain].chronicle + assert ledger_key.partition(":")[2] == chronicle_key.partition(":")[2] + + +def test_source_row_value_hash_canonicalizes_nested_key_epochs(): + artifact = _source_artifact() + row = SourceRow( + artifact=artifact, + sheet_name="Sheet1", + row_number=2, + values={"amount": 42}, + ) + column = SourceColumn( + artifact=artifact, + sheet_name="Sheet1", + column_number=1, + raw_name="amount", + normalized_name="amount", + ) + ledger_value = SourceRowValue( + source_row_key=build_source_row_key(row, epoch=Epoch.LEDGER), + source_column_key=build_source_column_key(column, epoch=Epoch.LEDGER), + row_number=2, + column_number=1, + raw_column_name="amount", + normalized_column_name="amount", + value=42, + ) + chronicle_value = SourceRowValue( + source_row_key=build_source_row_key(row, epoch=Epoch.CHRONICLE), + source_column_key=build_source_column_key(column, epoch=Epoch.CHRONICLE), + row_number=2, + column_number=1, + raw_column_name="amount", + normalized_column_name="amount", + value=42, + ) + + ledger_key = build_source_row_value_key(ledger_value, epoch=Epoch.LEDGER) + chronicle_key = build_source_row_value_key( + chronicle_value, + epoch=Epoch.CHRONICLE, + ) + + assert build_source_row_value_key(ledger_value) == ledger_key + assert ledger_key.partition(":")[2] == chronicle_key.partition(":")[2] + assert ledger_key.startswith("ledger.source_row_value.v1:") + assert chronicle_key.startswith("chronicle.source_row_value.v2:") + + +def test_fact_validation_accepts_mixed_lineage_epochs_with_distinct_digests(): + fact = _fact( + source_cell_keys=( + "ledger.source_cell.v1:ledger-cell", + "chronicle.source_cell.v2:chronicle-cell", + ), + source_row_keys=( + "ledger.source_row.v1:ledger-row", + "chronicle.source_row.v2:chronicle-row", + ), + ) + + assert validate_fact(fact) == () + + +def test_fact_validation_rejects_canonical_duplicate_lineage_keys(): + fact = _fact( + source_cell_keys=( + "ledger.source_cell.v1:same-cell", + "chronicle.source_cell.v2:same-cell", + ), + source_row_keys=( + "ledger.source_row.v1:same-row", + "chronicle.source_row.v2:same-row", + ), + ) + + errors = [(issue.code, issue.field, issue.message) for issue in validate_fact(fact)] + + assert [(code, field) for code, field, _message in errors] == [ + ("duplicate_lineage_key", "source_cell_keys"), + ("duplicate_lineage_key", "source_row_keys"), + ] + assert all("Ledger and Chronicle aliases" in message for _, _, message in errors) + + +def test_fact_validation_rejects_unknown_lineage_prefix_with_both_forms(): + errors = validate_fact(_fact(source_cell_keys=("future.source_cell.v9:1234",))) + + error = next(issue for issue in errors if issue.code == "malformed_lineage_key") + assert error.field == "source_cell_keys" + assert "ledger.source_cell.v1" in error.message + assert "chronicle.source_cell.v2" in error.message + + +def test_validate_facts_cli_accepts_chronicle_lineage_end_to_end(tmp_path, capsys): + fact = _fact( + source_cell_keys=("chronicle.source_cell.v2:accepted",), + source_row_keys=("chronicle.source_row.v2:accepted",), + ) + path = tmp_path / "chronicle-facts.jsonl" + save_facts_jsonl([fact], path) + + exit_code = harness_main(["validate-facts", "--input", str(path)]) + payload = json.loads(capsys.readouterr().out) + + assert exit_code == 0 + assert payload["valid"] + + +def test_validate_facts_cli_unknown_lineage_names_both_forms(tmp_path, capsys): + path = tmp_path / "unknown-facts.jsonl" + save_facts_jsonl( + [_fact(source_cell_keys=("future.source_cell.v9:rejected",))], + path, + ) + + exit_code = harness_main(["validate-facts", "--input", str(path)]) + payload = json.loads(capsys.readouterr().out) + message = json.dumps(payload) + + assert exit_code == 1 + assert "ledger.source_cell.v1" in message + assert "chronicle.source_cell.v2" in message + + +def test_frozen_fixture_bytes_are_unchanged(): + fixture_root = Path(__file__).parents[1] / "chronicle" / "fixtures" + expected_sha256 = { + fixture_root / "facts.jsonl": ( + "b0dd06765db7932c16a678b1ab321a7d908af26e2f2014d7da99c0eb5127e401" + ), + fixture_root / "consumer_facts.jsonl": ( + "6123f1cca28ccc72c053b105b8d50b5c25a72a5f5b92e73e7219f32de152a96a" + ), + fixture_root / "source_cells" / "soi_table_1_1_2023_cells.jsonl": ( + "615639f21ee63e54595c677e24c3eddff484c00a795a2f91b45a8575f021c7e2" + ), + } + + assert { + path: hashlib.sha256(path.read_bytes()).hexdigest() for path in expected_sha256 + } == expected_sha256 diff --git a/tests/test_chronicle_database.py b/tests/test_chronicle_database.py index b5a1715..b21993d 100644 --- a/tests/test_chronicle_database.py +++ b/tests/test_chronicle_database.py @@ -8,13 +8,16 @@ import pytest from chronicle.core import build_aggregate_constraints -from chronicle.jurisdictions.us.soi import AXIOM_IRC_AGI_CONCEPT from chronicle.database import build_chronicle_db +from chronicle.epoch import HASH_DOMAINS, Epoch from chronicle.harness import build_chronicle_db_file from chronicle.jurisdictions.us.soi import ( + AXIOM_IRC_AGI_CONCEPT, build_soi_table_1_1_source_cells, build_soi_table_1_1_facts, ) +from chronicle.sources.cells import build_source_cell_key +from chronicle.sources.rows import SourceRow, build_source_row_key def test_build_aggregate_constraints_lifts_agi_filters(): @@ -48,6 +51,9 @@ def test_build_chronicle_db_writes_aggregate_fact_constraints_and_lineage(tmp_pa with sqlite3.connect(db_path) as connection: connection.row_factory = sqlite3.Row + build = connection.execute( + "SELECT build_id, schema_version FROM ledger_builds" + ).fetchone() artifact = connection.execute( """ SELECT raw_r2_bucket, raw_r2_key, raw_r2_uri @@ -75,6 +81,8 @@ def test_build_chronicle_db_writes_aggregate_fact_constraints_and_lineage(tmp_pa assert all_returns["domain"] == "all_individual_income_tax_returns" assert all_returns["provenance_class"] == "administrative" assert all_returns["survey_instrument"] is None + assert build["build_id"].startswith("ledger.build.v1:") + assert build["schema_version"] == "ledger.relational.v1" assert artifact["raw_r2_bucket"] == "ledger-raw" assert artifact["raw_r2_key"].startswith("raw/irs_soi/soi-table-1-1/2023/") assert artifact["raw_r2_uri"].startswith("r2://ledger-raw/") @@ -176,6 +184,184 @@ def test_build_chronicle_db_build_id_changes_when_fact_payload_changes(tmp_path) assert original.build_id != changed.build_id +def test_build_chronicle_db_emits_chronicle_epoch_with_valid_lineage(tmp_path): + fact = next( + fact + for fact in build_soi_table_1_1_facts(2023) + if fact.source_record_id == "irs_soi.ty2023.table_1_1.all.return_count" + ) + cell = next( + cell for cell in build_soi_table_1_1_source_cells(2023) if cell.address == "B10" + ) + row = SourceRow( + artifact=cell.artifact, + sheet_name=cell.sheet_name, + row_number=cell.row_number, + values={"Returns": cell.raw_value}, + ) + ledger_row_key = build_source_row_key(row) + cell = replace(cell, source_row_key=ledger_row_key) + fact = replace(fact, source_row_keys=(ledger_row_key,)) + + ledger = build_chronicle_db( + [fact], + tmp_path / "ledger.db", + source_cells=[cell], + source_rows=[row], + ) + chronicle = build_chronicle_db( + [fact], + tmp_path / "chronicle.db", + source_cells=[cell], + source_rows=[row], + emit_epoch=Epoch.CHRONICLE, + ) + + assert ( + ledger.build_id.split(":", maxsplit=1)[1] + == chronicle.build_id.split(":", maxsplit=1)[1] + ) + assert chronicle.build_id.startswith("chronicle.build.v2:") + with sqlite3.connect(tmp_path / "chronicle.db") as connection: + build = connection.execute( + "SELECT build_id, schema_version FROM ledger_builds" + ).fetchone() + keys = connection.execute( + """ + SELECT + aggregate_facts.fact_key, + fact_source_cells.source_cell_key, + fact_source_rows.source_row_key, + source_cells.source_row_key + FROM aggregate_facts + JOIN fact_source_cells USING (fact_key) + JOIN fact_source_rows USING (fact_key) + JOIN source_cells USING (source_cell_key) + """ + ).fetchone() + row_value_keys = connection.execute( + """ + SELECT + source_row_values.source_row_value_key, + source_rows.source_row_key, + source_columns.source_column_key + FROM source_row_values + JOIN source_rows USING (source_row_key) + JOIN source_columns USING (source_column_key) + """ + ).fetchone() + foreign_key_errors = connection.execute("PRAGMA foreign_key_check").fetchall() + + assert build == (chronicle.build_id, "chronicle.relational.v2") + assert keys[0].startswith("chronicle.fact.v2:") + assert keys[1].startswith("chronicle.source_cell.v2:") + assert keys[2].startswith("chronicle.source_row.v2:") + assert keys[3] == keys[2] + assert row_value_keys[0].startswith("chronicle.source_row_value.v2:") + assert row_value_keys[1].startswith("chronicle.source_row.v2:") + assert row_value_keys[2].startswith("chronicle.source_column.v2:") + assert foreign_key_errors == [] + + +@pytest.mark.parametrize("emit_epoch", [Epoch.LEDGER, Epoch.CHRONICLE]) +def test_build_chronicle_db_defensively_deduplicates_lineage_aliases( + tmp_path, + emit_epoch, +): + fact = next( + fact + for fact in build_soi_table_1_1_facts(2023) + if fact.source_record_id == "irs_soi.ty2023.table_1_1.all.return_count" + ) + cell = next( + cell for cell in build_soi_table_1_1_source_cells(2023) if cell.address == "B10" + ) + row = SourceRow( + artifact=cell.artifact, + sheet_name=cell.sheet_name, + row_number=cell.row_number, + values={"Returns": cell.raw_value}, + ) + source_cell_key = build_source_cell_key(cell) + source_row_key = build_source_row_key(row) + cell = replace(cell, source_row_key=source_row_key) + fact = replace( + fact, + source_cell_keys=( + source_cell_key, + HASH_DOMAINS["source_cell"].key_for_epoch( + source_cell_key, + Epoch.CHRONICLE, + ), + ), + source_row_keys=( + source_row_key, + HASH_DOMAINS["source_row"].key_for_epoch( + source_row_key, + Epoch.CHRONICLE, + ), + ), + ) + db_path = tmp_path / f"{emit_epoch}.db" + + build_chronicle_db( + [fact], + db_path, + source_cells=[cell], + source_rows=[row], + emit_epoch=emit_epoch, + ) + + with sqlite3.connect(db_path) as connection: + source_cell_links = connection.execute( + "SELECT source_cell_key, ordinal FROM fact_source_cells" + ).fetchall() + source_row_links = connection.execute( + "SELECT source_row_key, ordinal FROM fact_source_rows" + ).fetchall() + lineage_counts = connection.execute( + """ + SELECT source_cell_count, source_row_count + FROM source_records + """ + ).fetchone() + + assert source_cell_links == [ + ( + HASH_DOMAINS["source_cell"].key_for_epoch( + source_cell_key, + emit_epoch, + ), + 0, + ) + ] + assert source_row_links == [ + ( + HASH_DOMAINS["source_row"].key_for_epoch( + source_row_key, + emit_epoch, + ), + 0, + ) + ] + assert lineage_counts == (1, 1) + + +def test_build_chronicle_db_rejects_unknown_explicit_build_epoch(tmp_path): + with pytest.raises( + ValueError, + match=( + "ledger[.]build[.]v1.*chronicle[.]build[.]v2|" + "chronicle[.]build[.]v2.*ledger[.]build[.]v1" + ), + ): + build_chronicle_db( + [], + tmp_path / "unknown-build.db", + build_id="future.build.v9:unknown", + ) + + def test_build_chronicle_db_rejects_unresolved_source_cell_lineage(tmp_path): fact = replace( build_soi_table_1_1_facts(2023)[0], diff --git a/tests/test_chronicle_governance.py b/tests/test_chronicle_governance.py index f914237..9767da0 100644 --- a/tests/test_chronicle_governance.py +++ b/tests/test_chronicle_governance.py @@ -6,6 +6,8 @@ import yaml +from chronicle.epoch import SCHEMA_IDS, schema_id + ROOT = Path(__file__).resolve().parents[1] @@ -18,7 +20,8 @@ def test_chronicle_governance_files_define_required_review_surface(): codeowners = codeowners_path.read_text() pr_template = template_path.read_text() - assert agents["schema_version"] == "policyengine_ledger.approved_agents.v1" + assert agents["schema_version"] in SCHEMA_IDS["approved_agents"].accepted + assert agents["schema_version"] == schema_id("approved_agents") assert agents["owners"]["github_team"] == "PolicyEngine/core-developers" assert "@PolicyEngine/core-developers" in codeowners assert "/packages/** @PolicyEngine/core-developers" in codeowners diff --git a/tests/test_chronicle_offline_fetch.py b/tests/test_chronicle_offline_fetch.py index 6e39991..c0fb913 100644 --- a/tests/test_chronicle_offline_fetch.py +++ b/tests/test_chronicle_offline_fetch.py @@ -8,6 +8,7 @@ import pytest +from chronicle.epoch import SCHEMA_IDS from chronicle.sources.offline_fetch import ( OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION, OfflineFetchManifestError, @@ -80,6 +81,30 @@ def test_existing_v1_fetch_manifest_remains_valid(): ) +def test_accepts_chronicle_epoch_and_keeps_ledger_emit_default(): + schema_pair = SCHEMA_IDS["offline_fetch_manifest"] + payload = _manifest() + payload["schema_version"] = schema_pair.chronicle + + manifest = validate_offline_fetch_manifest(payload) + + assert manifest.schema_version == schema_pair.chronicle + assert OFFLINE_FETCH_MANIFEST_SCHEMA_VERSION == schema_pair.ledger + + +def test_rejects_unknown_schema_and_names_both_epochs(): + schema_pair = SCHEMA_IDS["offline_fetch_manifest"] + payload = _manifest() + payload["schema_version"] = "future.offline_fetch_manifest.v9" + + with pytest.raises(OfflineFetchManifestError) as error: + validate_offline_fetch_manifest(payload) + + message = str(error.value) + assert schema_pair.ledger in message + assert schema_pair.chronicle in message + + def test_discovery_notes_are_optional_by_default_but_can_be_required(): payload = _manifest() del payload["artifacts"][0]["discovery_note"] diff --git a/tests/test_chronicle_source_cells.py b/tests/test_chronicle_source_cells.py index 9693d02..cae91cb 100644 --- a/tests/test_chronicle_source_cells.py +++ b/tests/test_chronicle_source_cells.py @@ -8,6 +8,7 @@ import openpyxl import pytest +from chronicle.epoch import HASH_DOMAINS from chronicle.harness import ( build_fixture_source_cell_file, validate_fixture_source_cells, @@ -65,6 +66,25 @@ def test_fixture_source_cells_validate(): assert report.counts["by_sheet"] == {"TBL11": 1932} +def test_source_cell_validation_accepts_both_row_epochs_and_rejects_unknown(): + cell = build_soi_table_1_1_source_cells(2023)[0] + pair = HASH_DOMAINS["source_row"] + + for prefix in pair.accepted: + accepted = replace(cell, source_row_key=f"{prefix}:same-payload") + assert validate_source_cells([accepted]).valid + + report = validate_source_cells( + [replace(cell, source_row_key="future.source_row.v9:same-payload")] + ) + + assert not report.valid + error = report.errors[0] + assert error.code == "malformed_source_row_key" + assert pair.ledger in error.message + assert pair.chronicle in error.message + + def test_source_record_selector_guard_fails_on_changed_row_header(): cells = build_soi_table_1_1_source_cells(2023) spec = build_soi_table_1_1_source_record_specs(2023)[0] diff --git a/tests/test_chronicle_source_package.py b/tests/test_chronicle_source_package.py index 7e095e5..0e3cda7 100644 --- a/tests/test_chronicle_source_package.py +++ b/tests/test_chronicle_source_package.py @@ -17,15 +17,18 @@ validate_consumer_fact_contract, ) from chronicle.core import validate_facts +from chronicle.epoch import SCHEMA_IDS from chronicle.source_package import ( SOURCE_ARTIFACT_CACHE_ENV, SOURCE_ARTIFACT_FETCH_ENV, + SOURCE_PACKAGE_SCHEMA_VERSION, DeclarativeRecordSet, SourceArtifactSpec, _read_source_artifact_content, _render_string, _source_artifact_cache_path, load_source_package, + scaffold_source_package, validate_source_package, ) from chronicle.sources.cells import build_source_cell_key, validate_source_cells @@ -168,6 +171,51 @@ def test_record_set_provenance_fields_propagate_to_specs_and_facts(): assert survey_spec.survey_instrument == "ACS 1-year" +def test_source_package_accepts_chronicle_epoch_and_keeps_ledger_emit_default( + tmp_path, +): + schema_pair = SCHEMA_IDS["source_package"] + source_path = REPO_ROOT / "packages" / "irs_soi" / "table_1_1" + payload = yaml.safe_load((source_path / "source_package.yaml").read_text()) + payload["schema_version"] = schema_pair.chronicle + package_dir = tmp_path / "chronicle-source-package" + package_dir.mkdir() + (package_dir / "source_package.yaml").write_text(yaml.safe_dump(payload)) + + package = load_source_package(package_dir) + scaffold_dir = tmp_path / "ledger-default-scaffold" + report = scaffold_source_package( + scaffold_dir, + source_id="test", + package_id="test-package", + ) + + assert package.package_id == payload["package_id"] + assert SOURCE_PACKAGE_SCHEMA_VERSION == schema_pair.ledger + assert ( + Path(report.source_package_path) + .read_text() + .startswith(f"schema_version: {schema_pair.ledger}\n") + ) + + +def test_source_package_rejects_unknown_schema_and_names_both_epochs(tmp_path): + schema_pair = SCHEMA_IDS["source_package"] + source_path = REPO_ROOT / "packages" / "irs_soi" / "table_1_1" + payload = yaml.safe_load((source_path / "source_package.yaml").read_text()) + payload["schema_version"] = "future.source_package.v9" + package_dir = tmp_path / "unknown-source-package-schema" + package_dir.mkdir() + (package_dir / "source_package.yaml").write_text(yaml.safe_dump(payload)) + + with pytest.raises(ValueError) as error: + load_source_package(package_dir) + + message = str(error.value) + assert schema_pair.ledger in message + assert schema_pair.chronicle in message + + def test_hmrc_packages_preserve_provenance_and_definition_year_metadata(): """HMRC facts must retain the axes needed to distinguish source claims.""" hmrc_packages = sorted( diff --git a/tests/test_chronicle_suite.py b/tests/test_chronicle_suite.py index 3f41813..6911d6e 100644 --- a/tests/test_chronicle_suite.py +++ b/tests/test_chronicle_suite.py @@ -2,6 +2,7 @@ from __future__ import annotations +from dataclasses import replace import json import sqlite3 import textwrap @@ -19,6 +20,7 @@ SourceRecordLayout, validate_facts, ) +from chronicle.epoch import HASH_DOMAINS, Epoch from chronicle.harness import main as harness_main from chronicle.sources.cells import ( SourceArtifactMetadata, @@ -530,6 +532,67 @@ def test_agent_acceptance_accepts_source_row_bound_constraints(): assert report.valid assert report.checks["row_lineage_semantics_evidenced"] + chronicle_row_key = HASH_DOMAINS["source_row"].key_for_epoch( + row_key, + Epoch.CHRONICLE, + ) + chronicle_cell_key = HASH_DOMAINS["source_cell"].key_for_epoch( + build_source_cell_key(cell), + Epoch.CHRONICLE, + ) + chronicle_cell = replace(cell, source_row_key=chronicle_row_key) + chronicle_fact = replace( + fact, + source_cell_keys=(chronicle_cell_key,), + source_row_keys=(chronicle_row_key,), + ) + chronicle_report = build_agent_acceptance_report( + [chronicle_fact], + [row], + [chronicle_cell], + source_rows=validate_source_rows([row]), + source_cells=validate_source_cells([chronicle_cell]), + source_regions=SourceRegionSuiteReport( + region_count=0, + covered_cell_count=0, + errors=(), + ), + source_records=SourceRecordSuiteReport( + spec_count=1, + resolved_count=1, + lineaged_count=1, + errors=(), + ), + fact_report=validate_facts([chronicle_fact]), + concept_alignments=ConceptAlignmentReport( + alignment_count=0, + checked_count=0, + alignments=(), + errors=(), + ), + ) + + assert chronicle_report.valid + assert chronicle_report.checks["row_lineage_semantics_evidenced"] + + invalid_fact = replace(fact, source_row_keys=(123,)) + invalid_report = build_agent_acceptance_report( + [invalid_fact], + [row], + [cell], + source_rows=validate_source_rows([row]), + source_cells=validate_source_cells([cell]), + source_regions=SourceRegionSuiteReport(0, 0, ()), + source_records=SourceRecordSuiteReport(1, 1, 1, ()), + fact_report=validate_facts([invalid_fact]), + concept_alignments=ConceptAlignmentReport(0, 0, (), ()), + ) + + assert not invalid_report.valid + assert "fact_source_row_lineage_unresolved" in { + error.code for error in invalid_report.errors + } + def test_agent_acceptance_accepts_age_constraints_from_source_cell_header(): artifact = SourceArtifactMetadata( diff --git a/tests/test_policyengine_chronicle_schema.py b/tests/test_policyengine_chronicle_schema.py index 12937c9..263d0cf 100644 --- a/tests/test_policyengine_chronicle_schema.py +++ b/tests/test_policyengine_chronicle_schema.py @@ -13,21 +13,62 @@ import pytest +from chronicle.epoch import Epoch, HASH_DOMAINS, SCHEMA_IDS from policyengine_chronicle.schema import ( CONSUMER_FACT_SCHEMA_SHA256, consumer_fact_schema, + normalize_consumer_fact_row_epochs, + validate_consumer_fact_row_epochs, validate_consumer_fact_row, ) _REPO_ROOT = Path(__file__).parents[1] _DOCS_SCHEMA_PATH = _REPO_ROOT / "docs" / "schemas" / "consumer_fact.v1.schema.json" _PACKAGED_SCHEMA_PATH = ( - _REPO_ROOT - / "policyengine_chronicle" - / "schemas" - / "consumer_fact.v1.schema.json" + _REPO_ROOT / "policyengine_chronicle" / "schemas" / "consumer_fact.v1.schema.json" ) _SAMPLE_PATH = _REPO_ROOT / "chronicle" / "fixtures" / "consumer_facts.jsonl" +_FROZEN_SCHEMA_SHA256 = ( + "76ac268e626c86146cee51193e0cbecbb197ddbf3bf410156fe7da7c0edae3ad" +) + +_TOP_LEVEL_KEY_DOMAINS = { + "aggregate_fact_key": "aggregate_fact", + "semantic_fact_key": "semantic_fact", + "legacy_fact_key": "fact", + "source_release_key": "source_release", + "source_series_key": "source_series", + "observed_measure_key": "observed_measure", + "dimension_set_key": "dimension_set", + "universe_constraint_set_key": "universe_constraint_set", +} + + +def _fixture_row(index=0): + return json.loads(_SAMPLE_PATH.read_text().splitlines()[index]) + + +def _chronicle_epoch_row(row): + row["schema_version"] = SCHEMA_IDS["consumer_fact"].chronicle + for field_name, domain_name in _TOP_LEVEL_KEY_DOMAINS.items(): + row[field_name] = HASH_DOMAINS[domain_name].key_for_epoch( + row[field_name], Epoch.CHRONICLE + ) + alignment = row.get("concept_alignment") + if alignment is not None: + alignment["concept_alignment_key"] = HASH_DOMAINS[ + "concept_alignment" + ].key_for_epoch(alignment["concept_alignment_key"], Epoch.CHRONICLE) + lineage = row["lineage"] + for field_name, domain_name in ( + ("source_cell_keys", "source_cell"), + ("source_row_keys", "source_row"), + ): + lineage[field_name] = [ + HASH_DOMAINS[domain_name].key_for_epoch(key, Epoch.CHRONICLE) + for key in lineage.get(field_name, []) + ] + return row def test_packaged_schema_is_byte_identical_to_docs_schema(): @@ -35,7 +76,9 @@ def test_packaged_schema_is_byte_identical_to_docs_schema(): packaged_bytes = _PACKAGED_SCHEMA_PATH.read_bytes() assert packaged_bytes == docs_bytes + assert hashlib.sha256(docs_bytes).hexdigest() == _FROZEN_SCHEMA_SHA256 assert hashlib.sha256(packaged_bytes).hexdigest() == CONSUMER_FACT_SCHEMA_SHA256 + assert CONSUMER_FACT_SCHEMA_SHA256 == _FROZEN_SCHEMA_SHA256 def test_consumer_fact_schema_is_the_v1_contract_row(): @@ -58,6 +101,106 @@ def test_valid_fixture_rows_pass_validation(): validate_consumer_fact_row(row, line_number, _SAMPLE_PATH) +def test_all_chronicle_epoch_identifiers_pass_without_mutating_row(): + row = _chronicle_epoch_row(_fixture_row(1)) + row["lineage"]["source_row_keys"] = [ + HASH_DOMAINS["source_row"].chronicle + ":" + "a" * 24 + ] + original = json.loads(json.dumps(row)) + + validate_consumer_fact_row(row, 2, _SAMPLE_PATH) + + assert row == original + normalized = normalize_consumer_fact_row_epochs(row, 2, _SAMPLE_PATH) + assert normalized["schema_version"] == SCHEMA_IDS["consumer_fact"].ledger + for field_name, domain_name in _TOP_LEVEL_KEY_DOMAINS.items(): + assert normalized[field_name].startswith(HASH_DOMAINS[domain_name].ledger + ":") + assert normalized["concept_alignment"]["concept_alignment_key"].startswith( + HASH_DOMAINS["concept_alignment"].ledger + ":" + ) + assert normalized["lineage"]["source_cell_keys"][0].startswith( + HASH_DOMAINS["source_cell"].ledger + ":" + ) + assert normalized["lineage"]["source_row_keys"][0].startswith( + HASH_DOMAINS["source_row"].ledger + ":" + ) + + +def test_mixed_epoch_identifiers_pass_validation(): + row = _fixture_row(1) + row["schema_version"] = SCHEMA_IDS["consumer_fact"].chronicle + row["aggregate_fact_key"] = HASH_DOMAINS["aggregate_fact"].key_for_epoch( + row["aggregate_fact_key"], Epoch.CHRONICLE + ) + row["concept_alignment"]["concept_alignment_key"] = HASH_DOMAINS[ + "concept_alignment" + ].key_for_epoch(row["concept_alignment"]["concept_alignment_key"], Epoch.CHRONICLE) + row["lineage"]["source_cell_keys"][0] = HASH_DOMAINS["source_cell"].key_for_epoch( + row["lineage"]["source_cell_keys"][0], Epoch.CHRONICLE + ) + row["lineage"]["source_row_keys"] = [ + HASH_DOMAINS["source_row"].ledger + ":" + "b" * 24 + ] + + validate_consumer_fact_row(row, 2, _SAMPLE_PATH) + + +@pytest.mark.parametrize( + ("mutate", "ledger_form", "chronicle_form"), + [ + ( + lambda row: row.__setitem__("schema_version", "future.consumer_fact.v9"), + SCHEMA_IDS["consumer_fact"].ledger, + SCHEMA_IDS["consumer_fact"].chronicle, + ), + ( + lambda row: row.__setitem__( + "aggregate_fact_key", "future.aggregate_fact.v9:" + "0" * 24 + ), + HASH_DOMAINS["aggregate_fact"].ledger, + HASH_DOMAINS["aggregate_fact"].chronicle, + ), + ( + lambda row: row["concept_alignment"].__setitem__( + "concept_alignment_key", + "future.concept_alignment.v9:" + "0" * 24, + ), + HASH_DOMAINS["concept_alignment"].ledger, + HASH_DOMAINS["concept_alignment"].chronicle, + ), + ( + lambda row: row["lineage"]["source_cell_keys"].__setitem__( + 0, "future.source_cell.v9:" + "0" * 24 + ), + HASH_DOMAINS["source_cell"].ledger, + HASH_DOMAINS["source_cell"].chronicle, + ), + ( + lambda row: row["lineage"].__setitem__( + "source_row_keys", ["future.source_row.v9:" + "0" * 24] + ), + HASH_DOMAINS["source_row"].ledger, + HASH_DOMAINS["source_row"].chronicle, + ), + ], +) +def test_unknown_epoch_identifier_names_both_accepted_forms( + mutate, + ledger_form, + chronicle_form, +): + row = _fixture_row(1) + mutate(row) + + with pytest.raises(ValueError) as excinfo: + validate_consumer_fact_row(row, 7, "mixed.jsonl") + + message = str(excinfo.value) + assert "row 7 of mixed.jsonl" in message + assert ledger_form in message + assert chronicle_form in message + + def test_missing_nested_required_field_names_field_and_location(): row = json.loads(_SAMPLE_PATH.read_text().splitlines()[0]) del row["observed_measure"]["unit"] @@ -77,3 +220,18 @@ def test_unknown_extra_field_is_rejected(): with pytest.raises(ValueError, match="surprise_field"): validate_consumer_fact_row(row, 1, "sample.jsonl") + + +def test_validate_only_epoch_check_matches_normalizer_and_never_copies(): + row = _chronicle_epoch_row(_fixture_row(1)) + original = json.loads(json.dumps(row)) + assert validate_consumer_fact_row_epochs(row, 2, _SAMPLE_PATH) is None + assert row == original + bad = _fixture_row(1) + bad["lineage"]["source_cell_keys"] = ["bogus.domain.v9:" + "a" * 24] + with pytest.raises(ValueError) as normalizer_error: + normalize_consumer_fact_row_epochs(bad, 3, _SAMPLE_PATH) + with pytest.raises(ValueError) as validator_error: + validate_consumer_fact_row_epochs(bad, 3, _SAMPLE_PATH) + assert str(validator_error.value) == str(normalizer_error.value) + assert bad["lineage"]["source_cell_keys"] == ["bogus.domain.v9:" + "a" * 24]