Epoch prep: accept chronicle-era fact domains and schema ids; emit unchanged (#143, mechanism 1)

README.md

@@ -506,6 +506,30 @@ Target inputs use a three-table schema:
 These are source-backed inputs. Microcosm owns the contracts that select them,
 the active support-aware subset, and calibrated solver execution.
 
+## Identifier Epochs
+
+Fact identity migrates by epoch, never in place. `chronicle/epoch.py` is the
+single registry of frozen Ledger-era hash domains and schema ids
+(`ledger.aggregate_fact.v2`, `ledger.consumer_fact.v1`, ...) and their
+Chronicle-era successors (`chronicle.aggregate_fact.v3`,
+`chronicle.consumer_fact.v2`, ...). A successor key hashes the same canonical
+payload as its Ledger key; only the prefix differs.
+
+- **Readers accept both epochs.** Every validator, key verifier, bundle loader,
+  and relational reader accepts either form on each identifier independently,
+  so mixed-epoch inputs load. Anything outside both forms is rejected with an
+  error naming both.
+- **Emitters stay Ledger-named.** `EMIT_EPOCH` is the one default a later,
+  consumer-gated cutover flips. Package scaffolds, relational builds, and
+  consumer artifacts emit Ledger identifiers today.
+- **Artifacts canonicalize on emit.** The consumer artifact pins the sha256 of
+  the frozen v1 consumer-fact schema, whose identifiers are Ledger-named, so
+  `build_consumer_artifact` rewrites every row it read to the emit epoch before
+  writing it; an artifact built from mixed-epoch rows is byte-identical to one
+  built from the same rows written Ledger-named. Asking an artifact boundary to
+  emit Chronicle names is refused until a successor schema is packaged and
+  pinned, and the refusal happens before any existing output is touched.
+
 ## Chronicle Facts And Microcosm Targets
 
 Source facts should be structurally normalized before Microcosm considers them

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,25 @@ 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; a malformed id is an
+    # input failure like the ones above, reported rather than raised.
+    try:
+        canonicalize_key("build", resolved_build_id)
+    except ValueError:
+        return DerivedArtifactPublishReport(
+            input_dir=str(input_path),
+            source_id=source_id,
+            package_id=package_id,
+            year=year,
+            build_id=resolved_build_id,
+            entries=(),
+            build_artifacts_path=str(build_artifacts_output)
+            if build_artifacts_output
+            else None,
+            errors=("malformed_build_id",),
+        )
+
     resolved_r2_prefix = resolve_r2_prefix(
         prefix=r2_prefix,
         default_prefix=DEFAULT_R2_DERIVED_PREFIX,
@@ -956,18 +977,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:

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,23 +505,27 @@ 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,
             "count": len(key_rows),
             "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: