Adopt the UK local-geography and firms contracts and compile the local target surface (#708, #759)

changelog.d/708-legacy-json-strict-decoder.fixed.md

@@ -0,0 +1 @@
+Resources declared as `legacy_json` now parse with the strict JSON decoder instead of the pure-Python YAML 1.2 scanner, restoring `load_country_spec("uk")` from ~38s to ~4s after this PR's ~40MB of committed local reference, membership, fixture, and receipt rows joined the bundle. The JSON path preserves every `load_yaml12` refusal — single document, string keys, no duplicate mapping keys, no non-finite numbers — and spec digests are unchanged because legacy resources contribute byte receipts, not normative projections.

changelog.d/uk-local-ledger-targets.added.md

@@ -0,0 +1 @@
+Added UK local geography and firms Ledger target contracts, local area crosswalk validation, and local-area target reference authoring/compilation support.

packages/microcosm-build/src/microcosm/build/country_spec.py

@@ -819,6 +819,97 @@ def _validate_target_references(
     return tuple(references)
 
 
+def _validate_local_target_references(
+    raw: Mapping[str, Any],
+    *,
+    country: str,
+    crosswalk: Mapping[str, Any] | None,
+) -> tuple[LedgerTargetReference, ...]:
+    """Validate a ``local_target_references.json`` resource."""
+
+    context = "local_target_references.json"
+    if crosswalk is None:
+        raise ValueError(
+            f"{context}: local target references require local_area_crosswalk.json."
+        )
+    rosters = _local_area_rosters(crosswalk, context=context)
+    references = _validate_target_references(raw, country=country)
+    names = [reference.name for reference in references]
+    duplicates = sorted({name for name in names if names.count(name) > 1})
+    if duplicates:
+        raise ValueError(f"{context}: duplicate reference name(s) {duplicates}.")
+    for reference in references:
+        if "@" not in reference.name:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} must use "
+                "target_id@geography_id naming."
+            )
+        contract_target_id, geography_id = reference.name.rsplit("@", 1)
+        if not contract_target_id or not geography_id:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} must use "
+                "target_id@geography_id naming."
+            )
+        selector = reference.ledger_selector
+        selector_level = selector.get("geography_level")
+        selector_id = selector.get("geography_id")
+        if not selector_level or not selector_id:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} must pin "
+                "ledger_selector.geography_level and geography_id."
+            )
+        if str(selector_id) != geography_id:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} geography id does not "
+                f"match selector geography_id {selector_id!r}."
+            )
+        roster = rosters.get(str(selector_level))
+        if roster is None:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} uses unknown "
+                f"geography level {selector_level!r}."
+            )
+        if geography_id not in roster["area_ids"]:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} geography id "
+                f"{geography_id!r} is outside the {selector_level!r} roster "
+                f"for expected vintage {roster['expected_vintage']!r}."
+            )
+        if reference.metadata.get("contract_target_id") not in {"", contract_target_id}:
+            raise ValueError(
+                f"{context}: reference {reference.name!r} metadata "
+                "contract_target_id must match the name prefix."
+            )
+    return references
+
+
+def _local_area_rosters(
+    crosswalk: Mapping[str, Any],
+    *,
+    context: str,
+) -> dict[str, dict[str, Any]]:
+    levels = crosswalk.get("levels")
+    if not isinstance(levels, Mapping):
+        raise ValueError(f"{context}: local_area_crosswalk.json must expose levels.")
+    rosters: dict[str, dict[str, Any]] = {}
+    for level, payload in levels.items():
+        if not isinstance(payload, Mapping):
+            raise ValueError(
+                f"{context}: local_area_crosswalk level {level!r} must be an object."
+            )
+        area_ids = payload.get("area_ids")
+        if not isinstance(area_ids, list) or not area_ids:
+            raise ValueError(
+                f"{context}: local_area_crosswalk level {level!r} must expose "
+                "a non-empty area_ids list."
+            )
+        rosters[str(level)] = {
+            "area_ids": frozenset(str(area_id) for area_id in area_ids),
+            "expected_vintage": payload.get("expected_vintage", ""),
+        }
+    return rosters
+
+
 # ---------------------------------------------------------------------------
 # The country spec
 # ---------------------------------------------------------------------------
@@ -952,6 +1043,7 @@ class ResolvedCountrySpec:
     support_spine: SupportSpineManifest | None
     geography_spine: GeographySpineManifest | None
     target_references: tuple[LedgerTargetReference, ...]
+    local_target_references: tuple[LedgerTargetReference, ...]
     gates: GatesManifest | None
     release_contract: ReleaseContractManifest | None
     take_up_contract: Mapping[str, Any] | None
@@ -1570,6 +1662,15 @@ def load_country_spec(country: str | Path) -> ResolvedCountrySpec:
         if "target_references.json" in payloads
         else ()
     )
+    local_target_references = (
+        _validate_local_target_references(
+            payloads["local_target_references.json"],
+            country=declared_country,
+            crosswalk=payloads.get("local_area_crosswalk.json"),
+        )
+        if "local_target_references.json" in payloads
+        else ()
+    )
     gates = (
         GatesManifest.from_mapping(payloads["gates.json"], country=declared_country)
         if "gates.json" in payloads
@@ -1596,6 +1697,7 @@ def load_country_spec(country: str | Path) -> ResolvedCountrySpec:
         support_spine=support_spine,
         geography_spine=geography_spine,
         target_references=target_references,
+        local_target_references=local_target_references,
         gates=gates,
         release_contract=release_contract,
         take_up_contract=take_up_contract,

packages/microcosm-build/src/microcosm/build/ledger_targets.py

@@ -22,10 +22,10 @@
 SUPPORTED_LEDGER_AGGREGATIONS = frozenset(("sum",))
 ALLOWED_ASSERTION_POLICIES = frozenset(("observed_only", "allow_source_projection"))
 ALLOWED_VALUE_OPERATIONS = frozenset(
-    ("identity", "sum", "calendar_year_average", "latest_plateau")
+    ("identity", "sum", "calendar_year_average", "latest_plateau", "count_x_mean")
 )
 MULTI_FACT_VALUE_OPERATIONS = frozenset(
-    ("sum", "calendar_year_average", "latest_plateau")
+    ("sum", "calendar_year_average", "latest_plateau", "count_x_mean")
 )
 DEFAULT_HIERARCHY_MATCH_SPEC_FIELDS = ("entity", "period", "family", "filter")
 
@@ -531,6 +531,8 @@ def target_spec_from_ledger_reference(
         numeric_value = sum(numeric_values) / len(numeric_values)
     elif reference.value_operation == "latest_plateau":
         numeric_value = numeric_values[-1]
+    elif reference.value_operation == "count_x_mean":
+        numeric_value = numeric_values[0] * numeric_values[1]
     elif reference.value_operation == "sum":
         numeric_value = sum(numeric_values)
     else:
@@ -615,7 +617,16 @@ def _validate_fact_aggregation(
         reference.metadata.get("fact_aggregation") == "time_mean"
         and aggregation == "mean"
     )
-    if aggregation not in SUPPORTED_LEDGER_AGGREGATIONS and not accepts_time_mean:
+    accepts_count_x_mean = (
+        reference.value_operation == "count_x_mean"
+        and aggregation == "mean"
+        and _count_mean_fact_role(fact) == "mean"
+    )
+    if (
+        aggregation not in SUPPORTED_LEDGER_AGGREGATIONS
+        and not accepts_time_mean
+        and not accepts_count_x_mean
+    ):
         raise ValueError(
             f"Ledger fact for {reference.name!r} has unsupported aggregation "
             f"{aggregation!r}; Microcosm targets must be compiled from sum "
@@ -888,6 +899,8 @@ def _resolve_reference_fact(
             )
         if reference.value_operation == "latest_plateau" and eligible_matches:
             return _resolve_latest_plateau_reference_facts(reference, eligible_matches)
+        if reference.value_operation == "count_x_mean" and eligible_matches:
+            return _resolve_count_x_mean_reference_facts(reference, eligible_matches)
         if len(eligible_matches) == 1:
             return eligible_matches[0]
         latest_match = _latest_period_selector_match(reference, eligible_matches)
@@ -994,6 +1007,49 @@ def _resolve_latest_plateau_reference_facts(
     return tuple(reversed(plateau))
 
 
+def _resolve_count_x_mean_reference_facts(
+    reference: LedgerTargetReference,
+    eligible_matches: list[object],
+) -> tuple[object, ...]:
+    partitions: dict[
+        tuple[tuple[str, ...], tuple[int, int, str]], dict[str, list[object]]
+    ] = {}
+    for fact in eligible_matches:
+        role = _count_mean_fact_role(fact)
+        if not role:
+            continue
+        key = (_selector_count_mean_partition_key(fact), _period_key(fact))
+        partitions.setdefault(key, {}).setdefault(role, []).append(fact)
+
+    valid = {
+        key: roles
+        for key, roles in partitions.items()
+        if len(roles.get("count", ())) == 1 and len(roles.get("mean", ())) == 1
+    }
+    if not valid:
+        observed = {
+            key: {role: len(facts) for role, facts in roles.items()}
+            for key, roles in partitions.items()
+        }
+        raise ValueError(
+            f"Ledger target reference {reference.name!r}: value_operation="
+            "'count_x_mean' requires exactly one count fact and one mean fact "
+            f"in a shared selector partition; observed {observed!r}."
+        )
+    latest_period = max(period_key for _, period_key in valid)
+    latest = [
+        roles for (_, period_key), roles in valid.items() if period_key == latest_period
+    ]
+    if len(latest) != 1:
+        raise ValueError(
+            f"Ledger target reference {reference.name!r}: value_operation="
+            "'count_x_mean' matched multiple count/mean selector partitions "
+            "at the latest eligible period."
+        )
+    roles = latest[0]
+    return (roles["count"][0], roles["mean"][0])
+
+
 def _monthly_operation_matches(
     reference: LedgerTargetReference,
     eligible_matches: list[object],
@@ -1080,6 +1136,34 @@ def _selector_sum_partition_key(fact: object) -> tuple[str, ...]:
     return invariant[:8] + invariant[11:]
 
 
+def _selector_count_mean_partition_key(fact: object) -> tuple[str, ...]:
+    return (
+        _source_name(fact),
+        _str_at(fact, "geography", "level"),
+        _str_at(fact, "geography", "id"),
+        _str_at(fact, "entity", "name"),
+        _normalized_record_set_id(_str_at(fact, "layout", "record_set_id")),
+        _str_at(fact, "layout", "record_set_spec_id"),
+        _str_at(fact, "layout", "groupby_dimension"),
+        _normalized_period_bearing_id(_str_at(fact, "layout", "groupby_value_id")),
+        json.dumps(_dimensions(fact), sort_keys=True, separators=(",", ":")),
+        json.dumps(_constraint_rows(fact), sort_keys=True, separators=(",", ":")),
+        _domain(fact),
+    )
+
+
+def _count_mean_fact_role(fact: object) -> str:
+    measure_id = (
+        _str_at(fact, "observed_measure", "source_measure_id")
+        or _str_at(fact, "layout", "measure_id")
+    ).lower()
+    if measure_id.endswith("_count") or measure_id == "count":
+        return "count"
+    if measure_id.endswith("_mean") or measure_id == "mean":
+        return "mean"
+    return ""
+
+
 def _eligible_selector_matches(
     reference: LedgerTargetReference,
     matches: list[object],
@@ -1425,6 +1509,8 @@ def _selector_candidates(fact: object, key: str) -> tuple[str, ...]:
         return (_str_at(fact, "entity", "name"),)
     if key in {"record_set_id", "layout_record_set_id"}:
         return (_str_at(fact, "layout", "record_set_id"),)
+    if key in {"record_set_spec_id", "layout_record_set_spec_id"}:
+        return (_str_at(fact, "layout", "record_set_spec_id"),)
     if key in {"groupby_dimension", "layout_groupby_dimension"}:
         return (_str_at(fact, "layout", "groupby_dimension"),)
     if key == "layout_groupby_value_id":
@@ -1640,6 +1726,9 @@ def _ledger_metadata(fact: object, *, fact_key: str) -> dict[str, str]:
         "ledger_entity_role": _str_at(fact, "entity", "role"),
         "ledger_domain": _domain(fact),
         "ledger_layout_record_set_id": _str_at(fact, "layout", "record_set_id"),
+        "ledger_layout_record_set_spec_id": _str_at(
+            fact, "layout", "record_set_spec_id"
+        ),
         "ledger_layout_groupby_dimension": _str_at(fact, "layout", "groupby_dimension"),
         "ledger_layout_groupby_value_id": _str_at(fact, "layout", "groupby_value_id"),
         "ledger_layout_measure_id": _str_at(fact, "layout", "measure_id"),

packages/microcosm-build/src/microcosm/build/spec_engine/loader.py

@@ -41,7 +41,7 @@
     assert_schema_id_allowed,
     load_schema_registry,
 )
-from .yaml12 import load_yaml12
+from .yaml12 import load_json_strict, load_yaml12
 
 SUPPORTED_SCHEMA_VERSION = 1
 BUNDLE_LOCK_FILENAME = "bundle.lock.json"
@@ -317,7 +317,13 @@ def load_bundle(
             )
         raw = _read_bytes(resource, label=path_text)
         text = _decode_utf8(raw, label=path_text)
-        parsed = load_yaml12(text, source=path_text)
+        if descriptor.kind is ResourceKind.LEGACY_JSON:
+            # Declared-JSON compatibility data can be megabytes of generated
+            # rows; the strict JSON decoder keeps the load_yaml12 value model
+            # without the pure-Python YAML scanner's cost.
+            parsed = load_json_strict(text, source=path_text)
+        else:
+            parsed = load_yaml12(text, source=path_text)
         if descriptor.kind is ResourceKind.SCHEMA:
             # Schema resources are grammar inputs and validated when the
             # registry is built; a country bundle does not normally repeat

packages/microcosm-build/src/microcosm/build/spec_engine/yaml12.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import json
 import math
 import re
 from collections.abc import Iterator
@@ -332,4 +333,48 @@ def load_yaml12_file(path: str | PathLike[str]) -> JSONValue:
     return load_yaml12(resource.read_text(encoding="utf-8"), source=str(resource))
 
 
+def load_json_strict(text: str, *, source: str = "<string>") -> JSONValue:
+    """Load one strict-JSON document into the same value model as load_yaml12.
+
+    JSON is a subset of the YAML 1.2 core schema, so a resource declared as
+    JSON parses with the C decoder instead of the pure-Python YAML scanner.
+    The JSON grammar already guarantees a single document with string mapping
+    keys and no tags or aliases; the two refusals it does not carry —
+    duplicate mapping keys and the non-finite number constants — are enforced
+    here so this path is never more permissive than :func:`load_yaml12`.
+    """
+
+    if not isinstance(text, str):
+        raise TypeError("JSON input must be text")
+
+    def _refuse_duplicate_keys(
+        pairs: list[tuple[str, JSONValue]],
+    ) -> dict[str, JSONValue]:
+        mapping: dict[str, JSONValue] = {}
+        for key, value in pairs:
+            if key in mapping:
+                raise _error(f"duplicate mapping key {key!r}", source=source)
+            mapping[key] = value
+        return mapping
+
+    def _refuse_constant(constant: str) -> JSONValue:
+        raise _error("non-finite numbers are not allowed", source=source)
+
+    try:
+        return json.loads(
+            text,
+            object_pairs_hook=_refuse_duplicate_keys,
+            parse_constant=_refuse_constant,
+        )
+    except SpecParseError:
+        raise
+    except json.JSONDecodeError as exc:
+        raise SpecParseError(
+            f"invalid JSON: {exc.msg}",
+            source=source,
+            line=exc.lineno,
+            column=exc.colno,
+        ) from exc
+
+
 __all__ = ["JSONScalar", "JSONValue", "load_yaml12", "load_yaml12_file"]