Require provenance_class (measurement basis) on every record set and fact

README.md

@@ -30,7 +30,9 @@ This repository provides:
   geography, period, source table, and lineage. Publisher projections (CBO
   baselines, BFP outlooks, SSA trustees tables, TPC/JCT scores) are facts
   typed `assertion: source_projection`; measured outcomes are the default
-  `assertion: observation`.
+  `assertion: observation`. Every fact also carries a required, closed
+  `provenance_class` measurement basis; survey aggregates additionally name
+  their `survey_instrument`.
 - **Normalization**: Low-assumption representation changes such as unit/scale
   conversion and source-published total/share arithmetic.
 - **Target profiles**: Source-backed target contracts and model-measurement

docs/agent-source-package-harness.md

@@ -481,6 +481,28 @@ units, aggregation methods, filters, and first-class constraints. The harness
 compiles those rows and measures into atomic source records, validates selectors
 against parsed cells, then emits target facts and the relational Ledger DB.
 
+Every record set must declare a `provenance_class`; there is no default. The
+closed vocabulary describes the publisher's measurement basis:
+
+- `administrative` for program, tax, collection, caseload, or payment records;
+- `census` for full-enumeration or census-controlled counts;
+- `survey_aggregate` for published sample-survey tabulations; and
+- `model_output` for model-based estimates, outlooks, baselines, and other
+  evaluation/oracle outputs.
+
+A `survey_aggregate` record set must also name its source survey in a non-empty
+`survey_instrument` string. `survey_instrument` is forbidden for every other
+class. Missing, unknown, wrongly typed, and misplaced values fail package load
+and build validation.
+
+```yaml
+record_sets:
+  - record_set_id: census_acs.acs1_{year}.s0101.national_age
+    provenance_class: survey_aggregate
+    survey_instrument: ACS 1-year
+    record_set_spec_id: census_acs.s0101.national_age.v1
+```
+
 Agents may add new package directories and YAML specs. They should not modify
 `ledger.core`, `ledger.database`, or `ledger.suite` unless the package cannot be
 expressed in the current contract and the failure is documented in the build

docs/schemas/consumer_fact.v1.schema.json

@@ -17,6 +17,7 @@
     "value",
     "value_type",
     "assertion",
+    "provenance_class",
     "period",
     "geography",
     "entity",
@@ -27,6 +28,32 @@
     "source",
     "lineage"
   ],
+  "allOf": [
+    {
+      "if": {
+        "properties": {
+          "provenance_class": {
+            "const": "survey_aggregate"
+          }
+        },
+        "required": [
+          "provenance_class"
+        ]
+      },
+      "then": {
+        "required": [
+          "survey_instrument"
+        ]
+      },
+      "else": {
+        "not": {
+          "required": [
+            "survey_instrument"
+          ]
+        }
+      }
+    }
+  ],
   "properties": {
     "schema_version": {
       "const": "ledger.consumer_fact.v1"
@@ -339,6 +366,21 @@
       ],
       "description": "Who asserted the value: a publisher-measured outcome (observation) or the publisher's own forward-looking estimate (source_projection). PolicyEngine-computed values are never consumer facts."
     },
+    "provenance_class": {
+      "type": "string",
+      "enum": [
+        "administrative",
+        "census",
+        "survey_aggregate",
+        "model_output"
+      ],
+      "description": "Publisher measurement basis. Consumers can gate calibration inputs on this required, closed classification."
+    },
+    "survey_instrument": {
+      "type": "string",
+      "pattern": ".*\\S.*",
+      "description": "Survey name required exactly when provenance_class is survey_aggregate."
+    },
     "period_coverage": {
       "type": "object",
       "additionalProperties": false,

ledger/consumer_contract.py

@@ -16,6 +16,7 @@
 from typing import Any
 
 from ledger.core import (
+    ALLOWED_PROVENANCE_CLASSES,
     DEFAULT_ASSERTION,
     AggregateConstraint,
     AggregateFact,
@@ -332,6 +333,19 @@ def validate_consumer_fact_contract(
                 )
             )
 
+        provenance_issue = _fact_provenance_class_issue(fact)
+        if provenance_issue is not None:
+            code, message, field = provenance_issue
+            errors.append(
+                ConsumerFactContractIssue(
+                    code=code,
+                    message=message,
+                    fact_index=index,
+                    fact_key=fact_key,
+                    field=field,
+                )
+            )
+
         if not fact.source_record_id:
             errors.append(
                 ConsumerFactContractIssue(
@@ -419,6 +433,33 @@ def _derived_source_provenance_issue(fact: AggregateFact) -> str | None:
     return None
 
 
+def _fact_provenance_class_issue(
+    fact: AggregateFact,
+) -> tuple[str, str, str] | None:
+    if type(fact.provenance_class) is not str or (
+        fact.provenance_class not in ALLOWED_PROVENANCE_CLASSES
+    ):
+        return (
+            "malformed_provenance_class",
+            f"Unsupported provenance class: {fact.provenance_class!r}.",
+            "provenance_class",
+        )
+    if fact.provenance_class == "survey_aggregate":
+        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.",
+                "survey_instrument",
+            )
+    elif fact.survey_instrument is not None:
+        return (
+            "misplaced_survey_instrument",
+            "survey_instrument is only valid for survey_aggregate facts.",
+            "survey_instrument",
+        )
+    return None
+
+
 def _filter_derived_constraints(
     fact: AggregateFact,
 ) -> tuple[AggregateConstraint, ...]:
@@ -532,6 +573,8 @@ def _consumer_fact_row(fact: AggregateFact) -> dict[str, Any]:
         "value": _json_value(fact.value),
         "value_type": _value_type(fact.value),
         "assertion": fact.assertion,
+        "provenance_class": fact.provenance_class,
+        "survey_instrument": fact.survey_instrument,
         "period": asdict(fact.period),
         "geography": _geography_payload(fact),
         "entity": asdict(fact.entity),
@@ -689,6 +732,7 @@ def _json_value(value: Any) -> Any:
     "value",
     "value_type",
     "assertion",
+    "provenance_class",
     "period",
     "geography",
     "entity",

ledger/core.py

@@ -64,6 +64,12 @@
 }
 ALLOWED_ASSERTIONS = {"observation", "source_projection"}
 DEFAULT_ASSERTION = "observation"
+ALLOWED_PROVENANCE_CLASSES = {
+    "administrative",
+    "census",
+    "model_output",
+    "survey_aggregate",
+}
 ALLOWED_PERIOD_BASES = {
     "calendar",
     "tax",
@@ -217,6 +223,10 @@ class AggregateFact:
     forward-looking estimate (CBO baselines, BFP outlooks, SSA trustees
     tables, TPC/JCT scores). PolicyEngine-computed values are never facts;
     aged, uprated, or reconciled levels belong in downstream builds.
+
+    ``provenance_class`` records the publisher's measurement basis and has no
+    default. Survey aggregates additionally require ``survey_instrument``;
+    that field is forbidden for administrative, census, and model outputs.
     """
 
     value: int | float | str | Decimal
@@ -226,6 +236,8 @@ class AggregateFact:
     measure: Measure
     aggregation: Aggregation
     source: SourceProvenance
+    provenance_class: str
+    survey_instrument: str | None = None
     filters: dict[str, Scalar] = field(default_factory=dict)
     domain: str = "all"
     label: str | None = None
@@ -466,6 +478,8 @@ def validate_fact(fact: AggregateFact) -> tuple[ValidationIssue, ...]:
             )
         )
 
+    _validate_provenance_class(errors, fact)
+
     _validate_value(errors, fact.value)
     _validate_filters(errors, fact.filters)
     _validate_constraints(errors, fact.constraints)
@@ -539,6 +553,9 @@ def fact_counts(facts: list[AggregateFact]) -> dict[str, dict[str, int]]:
             f"{fact.period.type}:{fact.period.value}" for fact in facts
         ),
         "by_assertion": _counter_dict(fact.assertion for fact in facts),
+        "by_provenance_class": _counter_dict(
+            fact.provenance_class for fact in facts
+        ),
         "missing_labels": {"count": sum(1 for fact in facts if not fact.label)},
         "missing_provenance": {
             "count": sum(1 for fact in facts if _has_missing_provenance(fact))
@@ -594,6 +611,42 @@ def _validate_value(errors: list[ValidationIssue], value: Any) -> None:
         errors.append(_issue("missing_value", "Fact value is required", "value"))
 
 
+def _validate_provenance_class(
+    errors: list[ValidationIssue],
+    fact: AggregateFact,
+) -> None:
+    provenance_class = fact.provenance_class
+    if type(provenance_class) is not str or (
+        provenance_class not in ALLOWED_PROVENANCE_CLASSES
+    ):
+        errors.append(
+            _issue(
+                "malformed_provenance_class",
+                f"Unsupported provenance class: {provenance_class!r}",
+                "provenance_class",
+            )
+        )
+        return
+
+    if provenance_class == "survey_aggregate":
+        if type(fact.survey_instrument) is not str or not fact.survey_instrument.strip():
+            errors.append(
+                _issue(
+                    "missing_survey_instrument",
+                    "Survey-aggregate facts need a non-empty survey instrument.",
+                    "survey_instrument",
+                )
+            )
+    elif fact.survey_instrument is not None:
+        errors.append(
+            _issue(
+                "misplaced_survey_instrument",
+                "survey_instrument is only valid for survey_aggregate facts.",
+                "survey_instrument",
+            )
+        )
+
+
 def _validate_filters(
     errors: list[ValidationIssue],
     filters: dict[str, Scalar],

ledger/database.py

@@ -291,6 +291,8 @@ def _create_schema(connection: sqlite3.Connection) -> None:
             aggregation_method TEXT NOT NULL,
             aggregation_denominator TEXT,
             domain TEXT NOT NULL,
+            provenance_class TEXT NOT NULL,
+            survey_instrument TEXT,
             assertion TEXT NOT NULL DEFAULT 'observation',
             period_coverage_json TEXT,
             filters_json TEXT NOT NULL,
@@ -834,6 +836,8 @@ def _insert_aggregate_fact(
             aggregation_method,
             aggregation_denominator,
             domain,
+            provenance_class,
+            survey_instrument,
             assertion,
             period_coverage_json,
             filters_json,
@@ -847,7 +851,7 @@ def _insert_aggregate_fact(
             source_extraction_method,
             source_method_notes
         )
-        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
         """,
         (
             fact_key,
@@ -890,6 +894,8 @@ def _insert_aggregate_fact(
             fact.aggregation.method,
             fact.aggregation.denominator,
             fact.domain,
+            fact.provenance_class,
+            fact.survey_instrument,
             fact.assertion,
             (
                 _json_dumps(