Make estimate-vs-projection resolution explicit: assertion_policy on target profiles

chronicle/core.py

@@ -79,6 +79,21 @@
 }
 ALLOWED_ASSERTIONS = {"observation", "source_projection"}
 DEFAULT_ASSERTION = "observation"
+# How profile resolution treats the assertion axis. observed_only is the safe
+# default: projections are invisible and a projection-only family fails loudly.
+# prefer_observed applies per series (one geography/entity/dimension tuple):
+# a series with any observed fact resolves only from observations, and a
+# series with none may fall back to projections — no single series ever
+# mixes bases across periods, and a projection-only series is never starved
+# by a neighbouring series' observation. allow_source_projection treats both
+# equally (for forecast families such as the OBR EFO lines), with the
+# observation winning an exact-period tie within a series.
+ASSERTION_POLICIES = {
+    "observed_only",
+    "prefer_observed",
+    "allow_source_projection",
+}
+DEFAULT_ASSERTION_POLICY = "observed_only"
 ALLOWED_PROVENANCE_CLASSES = {
     "administrative",
     "census",

docs/agent-source-package-harness.md

@@ -495,6 +495,38 @@ A `survey_aggregate` record set must also name its source survey in a non-empty
 class. Missing, unknown, wrongly typed, and misplaced values fail package load
 and build validation.
 
+Orthogonal to `provenance_class`, every fact carries an `assertion`:
+`observation` (a realized value, the default) or `source_projection` (the
+publisher's forward statement — OBR forecast years, NPP projection years,
+budget-allocation years). `provenance_class` says how the publisher measured;
+`assertion` says whether the period had happened. The two cross freely: a
+national-balance-sheet estimate is `model_output` + `observation`, an NPP
+projection year is `model_output` + `source_projection`.
+
+Consumer profiles resolve this axis explicitly rather than by convention.
+Each profile (or individual target) declares an `assertion_policy`:
+`observed_only` (the default — projections are invisible and a
+projection-only family fails loudly with `only_projection_facts`),
+`prefer_observed` (per series — one geography/entity/dimension tuple: a
+series with any observed fact resolves only from observations, and a series
+with none may fall back to projections, so no single series ever mixes
+bases across periods and a projection-only series is never starved by a
+neighbouring series' observation), or
+`allow_source_projection` (both compete under the period policy; an
+observation and a projection colliding within one series at the chosen
+period resolve to the observation with an `ambiguous_assertion_at_period`
+warning naming the series rather than double-counting it). A target whose `chronicle_selector` names `assertion`
+explicitly bypasses the policy — the selector is already maximal intent —
+and declaring both on one target is rejected at profile load as a
+contradiction. Whenever a
+projection is resolved, the report carries a `resolved_from_projection`
+warning and the resolved row exposes its `assertion`, so downstream builds
+never discover the estimate/projection boundary by accident. A fact that
+predates the axis loads as `observation`: both the file loader and the
+resolver default a missing `assertion` and reject unknown values, so every
+pre-existing package keeps its meaning and a typo'd assertion fails loudly
+instead of vanishing from the candidate set.
+
 ```yaml
 record_sets:
   - record_set_id: census_acs.acs1_{year}.s0101.national_age

policyengine_chronicle/consumer.py

@@ -263,6 +263,7 @@ def resolve_profile_targets(
     blocking coverage issues instead of returning an invalid report.
     """
     requested = _normalize_period(requested_period)
+    rows = _normalize_assertion_rows(rows)
     alignment_map = _normalize_alignments(alignments)
     if profile.base_period_policy not in SUPPORTED_BASE_PERIOD_POLICIES:
         raise ValueError(
@@ -296,6 +297,48 @@ def resolve_profile_targets(
             )
             continue
 
+        assertion_policy = (
+            target.assertion_policy or profile.default_assertion_policy
+        )
+        if "assertion" in target.chronicle_selector:
+            # An explicit assertion selector is maximal author intent; the
+            # policy governs only targets that do not select on assertion.
+            assertion_policy = "allow_source_projection"
+        observed = [
+            row for row in candidates if row["assertion"] == "observation"
+        ]
+        if assertion_policy == "observed_only":
+            if not observed:
+                issues.append(
+                    ResolutionIssue(
+                        code="only_projection_facts",
+                        message=(
+                            f"Target {target.target_id!r} matched only "
+                            "source_projection facts and its assertion_policy "
+                            "is 'observed_only'; declare 'prefer_observed' or "
+                            "'allow_source_projection' to resolve projections "
+                            "deliberately."
+                        ),
+                        profile_id=profile.profile_id,
+                        target_id=target.target_id,
+                        severity="error",
+                    )
+                )
+                continue
+            candidates = observed
+        elif assertion_policy == "prefer_observed" and observed:
+            # Per series, not per family: a series with any observed fact
+            # resolves only from observations, while a projection-only
+            # series keeps its projections instead of being starved by a
+            # neighbouring series' observation.
+            observed_series = {_series_key(row) for row in observed}
+            candidates = [
+                row
+                for row in candidates
+                if row["assertion"] == "observation"
+                or _series_key(row) not in observed_series
+            ]
+
         chosen_period, period_issue = _choose_period(
             profile.profile_id,
             target,
@@ -341,9 +384,63 @@ def resolve_profile_targets(
         else:
             basis = "declared_alignment"
 
-        for row in candidates:
-            if dict(row["period"]) != chosen_period:
+        rows_at_period = [
+            row for row in candidates if dict(row["period"]) == chosen_period
+        ]
+        series_at_period: dict[str, list[Mapping[str, Any]]] = {}
+        for row in rows_at_period:
+            series_at_period.setdefault(_series_key(row), []).append(row)
+        drop_keys: set[str] = set()
+        for series_rows in series_at_period.values():
+            if len({row["assertion"] for row in series_rows}) < 2:
                 continue
+            # An observation and a publisher projection collide within one
+            # series at the chosen period; emitting both would double-count
+            # it. The realized value wins the tie, loudly. Series that were
+            # never in a tie — a geography whose only fact is a projection —
+            # are untouched.
+            drop_keys.update(
+                row["aggregate_fact_key"]
+                for row in series_rows
+                if row["assertion"] != "observation"
+            )
+            sample = series_rows[0]
+            geography = sample.get("geography", {})
+            dimensions = sample.get("dimensions") or {}
+            where = f"geography {geography.get('level')}:{geography.get('id')}"
+            if dimensions:
+                where += f", dimensions {json.dumps(dimensions, sort_keys=True)}"
+            issues.append(
+                ResolutionIssue(
+                    code="ambiguous_assertion_at_period",
+                    message=(
+                        f"Target {target.target_id!r} matched both an "
+                        f"observation and a source_projection for one series "
+                        f"({where}) at "
+                        f"{chosen_period['type']}:{chosen_period['value']}; "
+                        "resolved the observation. Select on assertion or "
+                        "tighten dimensions/record_set_id to address the "
+                        "overlap explicitly."
+                    ),
+                    profile_id=profile.profile_id,
+                    target_id=target.target_id,
+                    severity="warning",
+                )
+            )
+        if drop_keys:
+            rows_at_period = [
+                row
+                for row in rows_at_period
+                if row["aggregate_fact_key"] not in drop_keys
+            ]
+
+        # The flag is set inside the row loop deliberately: one
+        # resolved_from_projection warning per target, however many of its
+        # rows resolve from projections.
+        projection_resolved = False
+        for row in rows_at_period:
+            if row["assertion"] == "source_projection":
+                projection_resolved = True
             resolved.append(
                 _resolved_target(
                     profile.profile_id,
@@ -354,6 +451,21 @@ def resolve_profile_targets(
                     alignment=alignment,
                 )
             )
+        if projection_resolved:
+            issues.append(
+                ResolutionIssue(
+                    code="resolved_from_projection",
+                    message=(
+                        f"Target {target.target_id!r} resolved from a "
+                        "source_projection fact at "
+                        f"{chosen_period['type']}:{chosen_period['value']} "
+                        f"under assertion_policy {assertion_policy!r}."
+                    ),
+                    profile_id=profile.profile_id,
+                    target_id=target.target_id,
+                    severity="warning",
+                )
+            )
 
     report = ResolutionReport(
         profile_id=profile.profile_id,
@@ -539,6 +651,55 @@ def _latest(values) -> Any:
     return max(values, key=str)
 
 
+def _series_key(row: Mapping[str, Any]) -> str:
+    """Identity of a co-resolving series, blind to source, period, assertion.
+
+    Groups the rows a target resolves together so the per-series rules (the
+    prefer_observed filter, the assertion tie-break) never let one series'
+    observation starve a different series that only has a projection.
+    Source identity is deliberately excluded so one publisher's estimate and
+    another table's projection of the same series still collide. The
+    canonical concept is not carried on the row, so two different concepts
+    sharing every axis below within one selector match are a selector-hygiene
+    problem this key does not adjudicate.
+    """
+    observed_measure = row.get("observed_measure", {})
+    return json.dumps(
+        {
+            "geography": row.get("geography"),
+            "entity": row.get("entity"),
+            "aggregation": row.get("aggregation"),
+            "dimension_set_key": row.get("dimension_set_key"),
+            "universe_constraint_set_key": row.get("universe_constraint_set_key"),
+            "unit": observed_measure.get("unit"),
+        },
+        sort_keys=True,
+    )
+
+
+def _normalize_assertion_rows(
+    rows: Sequence[Mapping[str, Any]],
+) -> list[dict[str, Any]]:
+    """Police the assertion axis for rows that bypassed the file loader.
+
+    ``_load_consumer_rows`` already defaults and validates ``assertion``;
+    rows handed to :func:`resolve_profile_targets` directly get the same
+    treatment here, so a typo such as ``assertion: projection`` fails loudly
+    instead of silently vanishing from every policy's candidate set.
+    """
+    normalized: list[dict[str, Any]] = []
+    for index, row in enumerate(rows):
+        mutable = dict(row)
+        assertion = mutable.setdefault("assertion", DEFAULT_ASSERTION)
+        if assertion not in ALLOWED_ASSERTIONS:
+            raise ValueError(
+                f"Consumer fact row {index} has unsupported assertion "
+                f"{assertion!r}; allowed: {sorted(ALLOWED_ASSERTIONS)}."
+            )
+        normalized.append(mutable)
+    return normalized
+
+
 def _normalize_period(period: Mapping[str, Any]) -> dict[str, Any]:
     if not isinstance(period, Mapping):
         raise ValueError("requested_period must be a mapping with type and value.")

policyengine_chronicle/target_profiles/model.py

@@ -14,6 +14,8 @@
 from importlib.resources import files
 from typing import Any
 
+from chronicle.core import ASSERTION_POLICIES, DEFAULT_ASSERTION_POLICY
+
 TARGET_PROFILE_SCHEMA_VERSION = "policyengine_ledger.target_profile.v1"
 FORBIDDEN_VALUE_KEYS = {"aggregation", "operation", "registry", "target_value", "value"}
 FORBIDDEN_RUNTIME_KEYS = {
@@ -52,6 +54,7 @@ class TargetProfileTarget:
     measurement: Mapping[str, Any]
     bindings: Mapping[str, TargetProfileBinding]
     tolerance: float | None = None
+    assertion_policy: str | None = None
 
     def binding(self, backend: str) -> TargetProfileBinding:
         """Return the binding for ``backend`` or raise a useful error."""
@@ -72,6 +75,7 @@ class TargetProfile:
     label: str
     base_period_policy: str
     default_operation: str
+    default_assertion_policy: str
     targets: tuple[TargetProfileTarget, ...]
 
     def targets_for_geography(
@@ -118,6 +122,15 @@ def target_profile_from_mapping(raw: Mapping[str, Any]) -> TargetProfile:
             f"target profile {profile_id!r} must use operation 'sum', "
             f"got {default_operation!r}."
         )
+    default_assertion_policy = defaults.get(
+        "assertion_policy", DEFAULT_ASSERTION_POLICY
+    )
+    if default_assertion_policy not in ASSERTION_POLICIES:
+        raise ValueError(
+            f"target profile {profile_id!r} defaults.assertion_policy must be "
+            f"one of {sorted(ASSERTION_POLICIES)}, got "
+            f"{default_assertion_policy!r}."
+        )
     targets = tuple(
         _target_from_mapping(target)
         for target in _required_mapping_sequence(raw, "targets")
@@ -140,6 +153,7 @@ def target_profile_from_mapping(raw: Mapping[str, Any]) -> TargetProfile:
         label=label,
         base_period_policy=base_period_policy,
         default_operation=default_operation,
+        default_assertion_policy=default_assertion_policy,
         targets=targets,
     )
 
@@ -177,6 +191,18 @@ def _target_from_mapping(raw: Mapping[str, Any]) -> TargetProfileTarget:
         if not isinstance(tolerance, int | float) or isinstance(tolerance, bool):
             raise ValueError(f"target profile row {target_id!r}: invalid tolerance.")
         tolerance = float(tolerance)
+    assertion_policy = raw.get("assertion_policy")
+    if assertion_policy is not None and assertion_policy not in ASSERTION_POLICIES:
+        raise ValueError(
+            f"target profile row {target_id!r} assertion_policy must be one of "
+            f"{sorted(ASSERTION_POLICIES)}, got {assertion_policy!r}."
+        )
+    if assertion_policy is not None and "assertion" in chronicle_selector:
+        raise ValueError(
+            f"target profile row {target_id!r} declares both assertion_policy "
+            f"{assertion_policy!r} and a chronicle_selector on 'assertion'; "
+            "the selector already pins the axis — declare one or the other."
+        )
     return TargetProfileTarget(
         target_id=target_id,
         family=family,
@@ -185,6 +211,7 @@ def _target_from_mapping(raw: Mapping[str, Any]) -> TargetProfileTarget:
         measurement=measurement,
         bindings=bindings,
         tolerance=tolerance,
+        assertion_policy=assertion_policy,
     )