Validate transport activation before preparation; drop the wall-clock assertion (G6 review nits)

changelog.d/transport-compose-cli.added.md

@@ -6,9 +6,18 @@ router, and later packages append branches through a typed extension point
 that cannot re-key the skeleton. The driver runs ancestor-closed checkpoints
 on `<out>/.graph-store` and writes every output under `--out`; it refuses an
 existing store tree that contains a link, which could redirect store writes.
-Activation checks every selected resource and reference set and every null
-scenario knob before donor preparation. Calibration ancestry refuses hold-out
-selections and sources, benefit-unit engine references, and graph digests.
+Activation checks run before registry preparation, CREATE or any graph source
+read. Declaration selections must name present resources; value and JSON
+selections must resolve existing paths to non-null selected values. Parameter
+resolution shares these checks: JSON selections require objects, while value
+selections reject objects. Optional null fields within selected JSON remain
+allowed. Preflight also checks the dependent-child age limit, mandatory and
+selected reference activation rows, required engine commit and wheel pin
+presence, and every null scenario knob. The receipt kernel's shared validator
+checks contract fields, required text, program and exclusion structure, and
+declared rates; checks needing person data, weights or target surfaces run during
+receipt assignment. Calibration ancestry refuses hold-out selections and sources,
+benefit-unit engine references, and graph digests.
 Selected resource lists remain literal data. The driver writes deterministic
 HDF5 container bytes while readback continues to hash the complete file.
 CREATE inventories are cached by implementation, parameters and source bytes;

docs/agent-guide.md

@@ -358,9 +358,18 @@ writes, including its `.graph-store`, lies under `--out`, and it has no
 upload or staging path. Before it opens the store it refuses any link in an
 existing store tree (the root, `objects`, `tmp` or a shard), because the
 store creates those directories by following links; it does not guard
-against links planted while a build is running. It refuses before reading any graph source while the
-country spec still has unresolved evidence: every selected resource/reference
-set and every null scenario knob is checked before donor preparation.
+against links planted while a build is running. Activation checks run before
+registry preparation, CREATE or any graph source read. Every declaration
+selection must name a present resource; value and JSON selections must have an
+existing path and a non-null selected value. JSON selections require objects,
+and value selections reject objects, using the same checks as parameter
+resolution. Selected JSON may retain optional null fields. The preflight also
+checks the dependent-child age limit, mandatory and selected reference activation
+rows, required engine commit and wheel pin presence, and every null scenario
+knob. The receipt kernel's shared validator checks the executable receipt
+contract's fields, required text, program and exclusion structure, and declared
+rates before preparation; checks needing person data, weights or target surfaces
+run during receipt assignment.
 Tests for it run on the engine-free toy package in
 `test_support/microcosm_build/transport_composed.py`. The rules registry checks
 module hashes and binds the RuleSpec tree digest. Engine commit and wheel

packages/microcosm-build/src/microcosm/build/transport/cli.py

@@ -225,7 +225,11 @@ def prepare_transport_build(
 ) -> PreparedTransportBuild:
     """Check activation, build the registry and CREATE inventory, then compose.
 
-    Activation is checked before any engine is built or donor read.
+    Before registry preparation, CREATE or graph source reads, activation
+    checks declaration resource paths, non-null selected roots and encoding
+    shapes, the receipt kernel's data-independent contract constraints,
+    reference activation rows, required age/pin presence and null scenario
+    knobs. Optional null JSON fields remain allowed.
     ``sources`` maps every declared source name except the exported dataset
     (which the run writes) to a local path. ``engines_by_binding`` is the
     registry's seam for pure-Python adapters; without it, Axiom adapters are

packages/microcosm-build/src/microcosm/build/transport/column_kernels.py

@@ -97,6 +97,7 @@
     "UnitAttributesKernel",
     "assign_receipts",
     "register_column_kernels",
+    "validate_receipt_contract",
 ]
 
 
@@ -494,18 +495,18 @@ def _finite_nonnegative(value: object, name: str) -> float:
     return value
 
 
-def assign_receipts(
-    person: pd.DataFrame,
+def validate_receipt_contract(
     contract: Mapping[str, object],
-    *,
-    weights: np.ndarray | None = None,
-    targets: Mapping[str, float] | None = None,
-) -> tuple[pd.DataFrame, dict[str, object]]:
-    """Apply explicit receipt priorities to determined eligible program draws.
-
-    This shared pure operation contains receipt orchestration only. G4's
-    judgment decoder and the content layer's stable program draws supply its
-    semantics. Existing exclusions are read as booleans and never rewritten.
+) -> tuple[str, list[dict[str, object]], list | tuple]:
+    """Validate all receipt contract checks independent of person/target data.
+
+    Return the seed column, parsed programs and exclusion groups without reading
+    person data, weights or a target surface. Check the exact contract fields,
+    non-empty seed and required program text, an ordered non-empty program list,
+    program fields and rate-or-target declarations, distinct program/output
+    identities, exclusion groups naming distinct declared outputs, optional
+    column/reference text, distinct exclusion columns, and finite rates in
+    [0, 1]. A null optional payment column remains allowed.
     """
     if set(contract) != {"seed_column", "programs", "exclusion_groups"}:
         raise ValueError(
@@ -536,7 +537,30 @@ def assign_receipts(
             raise ValueError("A receipt program declares exactly one rate or target.")
         if "target_measure" in row and "target" not in row:
             raise ValueError("A target_measure requires a target reference.")
-        parsed.append({**row, **{key: _text(row[key], key) for key in required}})
+        parsed_row = {**row, **{key: _text(row[key], key) for key in required}}
+        payment_column = row.get("payment_column")
+        if payment_column is not None:
+            parsed_row["payment_column"] = _text(payment_column, "payment_column")
+        excludes = row.get("exclude_columns", [])
+        if not isinstance(excludes, list | tuple) or len(set(excludes)) != len(
+            excludes
+        ):
+            raise ValueError("Receipt exclude_columns must be distinct columns.")
+        parsed_row["exclude_columns"] = [
+            _text(column, "exclude column") for column in excludes
+        ]
+        if "target" in row:
+            parsed_row["target"] = _text(row["target"], "target")
+            if "target_measure" in row:
+                parsed_row["target_measure"] = _text(
+                    row["target_measure"], "target_measure"
+                )
+        else:
+            rate = _finite_nonnegative(row["rate"], "rate")
+            if rate > 1:
+                raise ValueError("Receipt rate must lie on [0, 1].")
+            parsed_row["rate"] = rate
+        parsed.append(parsed_row)
     outputs = [row["output"] for row in parsed]
     keys = [row["program"] for row in parsed]
     if len(set(outputs)) != len(outputs) or len(set(keys)) != len(keys):
@@ -549,6 +573,24 @@ def assign_receipts(
             or not set(group) <= set(outputs)
         ):
             raise ValueError("Each receipt exclusion group names distinct outputs.")
+    return seed_column, parsed, groups
+
+
+def assign_receipts(
+    person: pd.DataFrame,
+    contract: Mapping[str, object],
+    *,
+    weights: np.ndarray | None = None,
+    targets: Mapping[str, float] | None = None,
+) -> tuple[pd.DataFrame, dict[str, object]]:
+    """Apply explicit receipt priorities to determined eligible program draws.
+
+    This shared pure operation contains receipt orchestration only. G4's
+    judgment decoder and the content layer's stable program draws supply its
+    semantics. Existing exclusions are read as booleans and never rewritten.
+    The data-independent contract checks use ``validate_receipt_contract``.
+    """
+    seed_column, parsed, groups = validate_receipt_contract(contract)
     out = person.loc[:, ["person_id"]].copy()
     audit = {}
     if weights is not None:
@@ -568,19 +610,12 @@ def assign_receipts(
         )
         payment_column = row.get("payment_column")
         if payment_column is not None:
-            payment = person[_text(payment_column, "payment_column")].to_numpy(
-                dtype=np.float64
-            )
+            payment = person[payment_column].to_numpy(dtype=np.float64)
             if not np.isfinite(payment).all():
                 raise ValueError("Receipt payment values must be finite.")
             eligible &= payment > 0
-        excludes = row.get("exclude_columns", [])
-        if not isinstance(excludes, list | tuple) or len(set(excludes)) != len(
-            excludes
-        ):
-            raise ValueError("Receipt exclude_columns must be distinct columns.")
-        for column in excludes:
-            values = person[_text(column, "exclude column")]
+        for column in row["exclude_columns"]:
+            values = person[column]
             if str(values.dtype) not in {"bool", "boolean"} or values.isna().any():
                 raise ValueError(
                     "Existing receipt exclusions must be non-null booleans."
@@ -594,7 +629,7 @@ def assign_receipts(
         target = None
         mass = None if weights is None else math.fsum(weights[eligible].tolist())
         if "target" in row:
-            name = _text(row["target"], "target")
+            name = row["target"]
             if targets is None or name not in targets or weights is None:
                 raise ValueError(
                     f"Receipt target {name!r} needs a surface and weights."
@@ -607,9 +642,7 @@ def assign_receipts(
             # target <= mass, so a zero eligible mass has a zero target.
             rate = target / mass if mass else target
         else:
-            rate = _finite_nonnegative(row["rate"], "rate")
-            if rate > 1:
-                raise ValueError("Receipt rate must lie on [0, 1].")
+            rate = row["rate"]
         draws = derive_take_up_draws(person[seed_column].to_numpy(), row["program"])
         assigned = eligible & (draws < rate)
         out[output] = assigned
@@ -662,6 +695,7 @@ def run(self, context: KernelContext) -> KernelResult:
             context, self.ref, ("person",), artifact_aliases=frozenset({"surface"})
         )
         contract = _document(context, self.ref, "receipt_contract")
+        _, programs, _ = validate_receipt_contract(contract)
         targets = None
         if context.artifacts:
             if (
@@ -671,14 +705,11 @@ def run(self, context: KernelContext) -> KernelResult:
                 raise ValueError(f"{self.ref} reads only the typed surface artifact.")
             surface = decode_target_surface(context.artifacts["surface"].payload)
             targets = {}
-            programs = contract.get("programs", ())
             for row in programs:
-                if not isinstance(row, Mapping) or "target" not in row:
+                if "target" not in row:
                     continue
-                name = _text(row["target"], "target")
-                expected_measure = _text(
-                    row.get("target_measure", row.get("output")), "target_measure"
-                )
+                name = row["target"]
+                expected_measure = row.get("target_measure", row["output"])
                 matches = [
                     target for target in surface.registry.specs if target.name == name
                 ]

packages/microcosm-build/src/microcosm/build/transport/compose.py

@@ -42,8 +42,11 @@
 version), so an extension can never re-key the skeleton.
 
 Composition reads no donor, no facts and no engine, and writes nothing.
-Unresolved country evidence is refused up front with every missing item
-named (:func:`validate_transport_activation`); no proposed value is chosen.
+Activation checks selected resource paths, non-null selected roots and encoding
+shapes, the receipt kernel's data-independent contract constraints, reference
+activation rows, required age/pin presence and null scenario knobs. Refusals
+name every enumerated gap (:func:`validate_transport_activation`). Optional
+null JSON fields remain allowed; no proposed value is chosen.
 """
 
 from __future__ import annotations
@@ -80,6 +83,7 @@
 from microcosm.graph.canonical import canonical_json
 
 from .artifact_types import KERNEL_OUTPUTS
+from .column_kernels import validate_receipt_contract
 
 __all__ = [
     "GRAPH_RESOURCE",
@@ -375,8 +379,9 @@ def _selections(value: object) -> Iterator[Mapping]:
             yield from _selections(item)
 
 
-def _selected_resources(document: Mapping) -> set[str]:
-    names = set()
+def _declared_resource_selections(document: Mapping) -> Iterator[Mapping]:
+    """Resource selections in declaration parameters, including nested lists."""
+
     for row in document.get("nodes", ()):
         if not isinstance(row, Mapping):
             continue
@@ -385,9 +390,10 @@ def _selected_resources(document: Mapping) -> set[str]:
             continue
         for value in params.values():
             for selection in _selections(value):
-                if isinstance(selection.get("resource"), str):
-                    names.add(_stem(selection["resource"]))
-    return names
+                # Any mapping naming a resource is a selection, whatever the
+                # name's type: a malformed name is refused here, not later.
+                if "resource" in selection:
+                    yield selection
 
 
 def _null_paths(value: object, path: str = "") -> Iterator[str]:
@@ -402,20 +408,35 @@ def _null_paths(value: object, path: str = "") -> Iterator[str]:
 
 
 def validate_transport_activation(spec: Mapping[str, object]) -> None:
-    """Refuse unresolved declaration resources and country evidence before I/O.
-
-    Check every selected reference resource, the mandatory pre-calibration,
-    target and hold-out reference sets, and every null scenario knob. Missing
-    resources are the declaration's explicit selections, including selections
-    inside parameter lists; selected resource data remains literal data.
+    """Refuse enumerated activation gaps before registry/CREATE/source reads.
+
+    Resolve declaration resource selections with the same path, non-null root
+    and encoding checks as parameter resolution. Optional JSON nulls remain
+    allowed. Reuse the receipt kernel's data-independent contract validator.
+    Also check dependent-child age and required engine pin presence, mandatory
+    and selected reference activation rows, and every null scenario knob.
+    Selections inside parameter lists count; selected resource data is literal.
     """
 
     missing = []
-    selected = (
-        _selected_resources(transport_resource(spec, GRAPH_RESOURCE))
+    selections = (
+        tuple(_declared_resource_selections(transport_resource(spec, GRAPH_RESOURCE)))
         if _has_resource(spec, GRAPH_RESOURCE)
-        else set()
+        else ()
     )
+    selected = {
+        selection["resource"].removesuffix(".json")
+        for selection in selections
+        if isinstance(selection["resource"], str) and selection["resource"]
+    }
+    for selection in selections:
+        try:
+            _resource_param(spec, selection)
+        except (ValueError, TypeError) as error:
+            missing.append(
+                f"Resource {selection['resource']!r} path {selection.get('path', [])!r}: "
+                f"{error}"
+            )
     for name in ("benefit_unit_rule", "axiom_rules_bindings", "scenarios"):
         if not _has_resource(spec, name):
             missing.append(f"{name}.json")
@@ -458,15 +479,15 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None:
     contracts = None if receipt is None else receipt.get("receipts")
     if not isinstance(contracts, Mapping):
         missing.append(
-            "receipt_contract.json (the input closure's receipt_requirements "
+            "receipt_contract.json path ['receipts'] "
+            "(the input closure's receipt_requirements "
             "are prose, not an executable contract)"
         )
-    elif any(
-        not isinstance(row.get("programs"), list) or not row["programs"]
-        for row in ((contracts,) if "programs" in contracts else contracts.values())
-        if isinstance(row, Mapping)
-    ):
-        missing.append("receipt_contract.json receipts.programs")
+    else:
+        try:
+            validate_receipt_contract(contracts)
+        except (ValueError, TypeError) as error:
+            missing.append(f"receipt_contract.json path ['receipts']: {error}")
     if _has_resource(spec, "axiom_rules_bindings"):
         bindings = transport_resource(spec, "axiom_rules_bindings")
         pins = _mapping(bindings.get("engine", {}), "axiom_rules_bindings.engine")
@@ -482,12 +503,6 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None:
                 missing.append(f"scenario {scenario.get('id', '?')} {path}")
     if not _has_resource(spec, GRAPH_RESOURCE):
         missing.append(f"{GRAPH_RESOURCE}.json (the skeleton declaration)")
-    else:
-        missing.extend(
-            f"{name}.json (selected by {GRAPH_RESOURCE}.json)"
-            for name in sorted(selected)
-            if not _has_resource(spec, name)
-        )
     if missing:
         raise ValueError("Transport spec is not activated: " + "; ".join(missing) + ".")
 
@@ -818,6 +833,16 @@ def _param(
         return canonical_json(
             {"engine_refs": dict(sorted(engine_refs.items()))}
         ).decode("utf-8")
+    return _resource_param(spec, value)
+
+
+def _resource_param(spec: Mapping[str, object], value: Mapping) -> object:
+    """Resolve and validate one literal resource selection for preflight/params.
+
+    Selected roots cannot be null; JSON roots must be objects, and value roots
+    must not be objects. Optional JSON fields remain unchanged.
+    """
+
     if set(value) - {"resource", "path", "encoding"} or "resource" not in value:
         raise ValueError(
             "An object parameter must be a resource or prepared selection."
@@ -844,11 +869,14 @@ def _param(
         raise ValueError(f"Resource {name!r} path {list(path)!r} is not activated.")
     if encoding == "json":
         if not isinstance(document, Mapping):
-            raise ValueError(f"Resource {name!r} json selection must be an object.")
+            raise ValueError(
+                f"Resource {name!r} path {list(path)!r} json selection must be an object."
+            )
         return canonical_json(document).decode("utf-8")
     if isinstance(document, Mapping):
         raise ValueError(
-            f"Resource {name!r} value selection is an object; use json encoding."
+            f"Resource {name!r} path {list(path)!r} value selection is an object; "
+            "use json encoding."
         )
     return _literal_data(document)