Prove extension calibration nodes' measurement producers in compose

changelog.d/transport-extension-measurement-proofs.fixed.md

@@ -0,0 +1 @@
+Composition now proves the measurement producer for each calibration node that a transport extension adds, such as a calibration-tier scenario's `calibrate.*` node. The declared proof (G8-r2c ruling R2C-1) reads only the declared graph's rows, so it never names those nodes, and the extension pass of `compose_transport_graph` never ran the compiled check. Now the extension pass runs it after the ancestry guard, as the skeleton pass does. For an extension's calibration node, exactly one compiled ancestor must declare the measurement, that ancestor must output every proof column, and no other ancestor may write one by output or EXPAND cell (R2C-15). The refusal names the node as an extension calibration node with no proved producer. Each skeleton calibration node keeps the declared comparison, which the extension pass now also runs on the extended graph.

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

@@ -50,7 +50,8 @@
 Composition refuses an extension whose nodes change any skeleton node's
 compiled predecessors (for example a new member of a calibration base
 version). Explicit ``package_inputs`` may add extension artifacts to the final
-receipt; every other skeleton predecessor set remains fixed.
+receipt, and explicit ``gate_inputs`` may add extension evidence to a skeleton
+gate battery; every other skeleton predecessor set remains fixed.
 
 Composition reads no donor, no facts and no engine, and writes nothing.
 Activation checks selected resource paths, non-null selected roots and encoding
@@ -621,7 +622,9 @@ def measurement_proofs(
     columns. "Upstream" is ancestry in the declared
     graph under the S5 guard's rule (:func:`_declared_calibration_predecessors`);
     composition then checks the same producers against the compiled graph's
-    exact predecessors (:func:`_check_measurement_proofs`). Returns the
+    exact predecessors (:func:`_check_measurement_proofs`). A calibration node
+    that a transport extension adds has no row here, so that check proves its
+    producer from the compiled graph under the same two rules. Returns the
     proofs by measurement id and, for each measurement without one, the
     reason it stays parked.
     """
@@ -714,20 +717,32 @@ def measurement_proofs(
 
 
 def _check_measurement_proofs(
-    spec: Mapping[str, object], graph: Graph, compiled: CompiledGraph
+    spec: Mapping[str, object],
+    graph: Graph,
+    compiled: CompiledGraph,
+    *,
+    extension_nodes: Iterable[str] = (),
 ) -> None:
-    """Refuse a declared proof the compiled graph's ancestry contradicts.
+    """Refuse a composed graph whose calibration ancestry breaks a proof.
 
     For each proved measurement and each calibration node of the composed
     graph, the compiled ancestors of that node must hold exactly one node
-    whose params declare the measurement, and it must be the proof's
-    producer for that node. That producer must also be the only compiled
+    whose params declare the measurement. For a node of the declared graph,
+    it must be the proof's producer for that node. The declaration never
+    names a calibration node that a transport extension adds
+    (``extension_nodes``), so for one of those the compiled graph proves the
+    producer under R2C-1's rule: the one declarer must output each of the
+    proof's columns. Either way, the producer must also be the only compiled
     ancestor whose outputs or EXPAND cells hold each of the proof's columns
-    (R2C-15).
+    (R2C-15). It returns at once when the declaration proves no measurement.
+    Otherwise, as the declared check does, it also refuses a node whose
+    ``measurement_id`` is neither ``None`` nor a nonempty string. Composition
+    runs this check on the skeleton and again on the extended graph.
     """
     proofs, _ = measurement_proofs(spec)
     if not proofs:
         return
+    extension_nodes = frozenset(extension_nodes)
     roots = {node.id for node in graph.nodes if node.kernel.startswith("calibrate.")}
     endpoint = transport_endpoints(spec).get("calibration")
     if isinstance(endpoint, str):
@@ -737,25 +752,63 @@ def _check_measurement_proofs(
         for node in graph.nodes
         if isinstance(node.params, Mapping) and MEASUREMENT_PARAM in node.params
     }
+    malformed = sorted(
+        name
+        for name, value in declared.items()
+        if value is not None and not _is_text(value)
+    )
+    if malformed:
+        raise ValueError(
+            f"Composition refuses: nodes {malformed} declare a measurement_id "
+            "that is not a string."
+        )
     by_id = {node.id: node for node in graph.nodes}
     _, expand_cells = _precalibration_expansions(graph)
     for binding, proof in sorted(proofs.items()):
         named = sorted(name for name, value in declared.items() if value == binding)
+        columns = [tuple(name.split(".", 1)) for name in proof["columns"]]
         for root in sorted(roots):
             closure = _ancestor_closure(compiled.predecessors, {root})
             found = [name for name in named if name in closure]
-            if found != [proof["producers"].get(root)]:
-                raise ValueError(
+            if root in extension_nodes:
+                prefix = (
+                    f"Measurement {binding!r}: the composed graph proves no "
+                    f"producer for extension calibration node {root!r}; "
+                )
+                if len(found) != 1:
+                    raise ValueError(
+                        prefix + _measurement_refusal(binding, root, found, named) + "."
+                    )
+                (producer,) = found
+                outputs = {
+                    (output.entity, output.column) for output in by_id[producer].outputs
+                }
+                missing = [
+                    f"{entity}.{column}"
+                    for entity, column in columns
+                    if (entity, column) not in outputs
+                ]
+                if missing:
+                    raise ValueError(
+                        prefix + f"its one declarer {producer!r} does not write "
+                        f"{missing}."
+                    )
+            else:
+                prefix = (
                     f"Measurement {binding!r}: the compiled graph contradicts "
                     "the declared proof; "
-                    + _measurement_refusal(binding, root, found, named)
-                    + f", and the declaration proved {proof['producers'].get(root)!r}."
                 )
-            columns = [tuple(name.split(".", 1)) for name in proof["columns"]]
+                producer = proof["producers"].get(root)
+                if found != [producer]:
+                    raise ValueError(
+                        prefix
+                        + _measurement_refusal(binding, root, found, named)
+                        + f", and the declaration proved {producer!r}."
+                    )
             refusal = _second_writer_refusal(
                 binding,
                 root,
-                proof["producers"][root],
+                producer,
                 {
                     (entity, column): sorted(
                         name
@@ -770,10 +823,7 @@ def _check_measurement_proofs(
                 },
             )
             if refusal is not None:
-                raise ValueError(
-                    f"Measurement {binding!r}: the compiled graph contradicts "
-                    f"the declared proof; {refusal}."
-                )
+                raise ValueError(prefix + f"{refusal}.")
 
 
 def reference_profile_view(
@@ -4563,6 +4613,9 @@ def compose_transport_graph(
     extended = compile_graph(graph)
     _check_extension_checkpoints(spec, config, graph, extended)
     validate_transport_calibration_ancestry(spec, graph=graph, config=config)
+    _check_measurement_proofs(
+        spec, graph, extended, extension_nodes=(node.id for node in additional)
+    )
     # Money ownership is checked after each graph pass of the ancestry guard.
     _check_single_money_normalization(spec, graph, extended)
     _check_money_bridge_owners(graph, extended)

packages/microcosm-build/tests/engine_free/shared/test_nz_transport_toy_graph.py

@@ -19,6 +19,7 @@
 from microcosm.build.transport.cli import through
 from microcosm.build.transport.compose import (
     compose_transport_graph,
+    measurement_proofs,
     transport_resource,
     validate_transport_activation,
 )
@@ -1235,3 +1236,160 @@ def test_h2_sharing_binding_on_the_household_shapes_the_toy_lacks():
         10: False,  # 75, no household id
         11: False,  # 30, no household id either
     }
+
+
+# ---------------------------------------------------------------------------
+# G8-r2c A on the toy: compose proves each scenario calibration's producer
+# ---------------------------------------------------------------------------
+
+
+def _toy_measurement_proof(fixture):
+    proofs, refusals = measurement_proofs(fixture.spec)
+    assert refusals == {}
+    ((binding, proof),) = proofs.items()
+    return binding, proof
+
+
+def test_composition_proves_each_extension_calibration_producer(toy):
+    """The declaration names the skeleton's calibration node; compose proves V3's."""
+    fixture, _ = toy
+    _, proof = _toy_measurement_proof(fixture)
+    graph = compose_transport_graph(copy.deepcopy(fixture.spec), fixture.config)
+    calibration = {
+        node.id for node in graph.nodes if node.kernel.startswith("calibrate.")
+    }
+    assert "nz.v.V3.calibrate" in calibration - set(proof["producers"])
+
+
+def _v3_without_declaration(rows, nodes, binding, columns):
+    """M1: V3's stand-in keeps its outputs and drops the measurement id."""
+    del rows["nz.v.V3.income.standin"]["params"]["measurement_id"]
+
+
+def _v3_second_declarer(rows, nodes, binding, columns):
+    """M2: V3's earnings node declares the measurement too."""
+    rows["nz.v.V3.earnings.cents"]["params"]["measurement_id"] = binding
+
+
+def _v3_universe_elsewhere(rows, nodes, binding, columns):
+    """M4: V3's universe column moves from the stand-in to the earnings node."""
+    standin = rows["nz.v.V3.income.standin"]
+    (moved,) = [row for row in standin["outputs"] if row["column"] == columns[1]]
+    standin["outputs"].remove(moved)
+    rows["nz.v.V3.earnings.cents"]["outputs"].append(moved)
+
+
+def _v3_review_version(rows, nodes, columns):
+    """Open a filter version inside V3 whose one node rewrites both columns.
+
+    Preparation, the compiled targets and the problem move into it.
+    """
+    standin = rows["nz.v.V3.income.standin"]
+    dtypes = {row["column"]: row["dtype"] for row in standin["outputs"]}
+    moved = [
+        rows[name]
+        for name in (
+            "nz.v.V3.calibration.prepare",
+            "nz.v.V3.targets.compile",
+            "nz.v.V3.targets.problem",
+        )
+    ]
+    for row in moved:
+        nodes.remove(row)
+        row["population"] = "nz.v.V3.review"
+    position = nodes.index(standin) + 1
+    nodes[position:position] = [
+        {
+            "id": "nz.v.V3.review",
+            "kernel": "transport.boundary@1",
+            "structural": "filter",
+            "base": "nz.v.V3",
+        },
+        {
+            "id": "nz.v.V3.review.rewrite",
+            "kernel": "transport.currency@1",
+            "population": "nz.v.V3.review",
+            "inputs": [{"entity": "person", "columns": list(columns)}],
+            "outputs": [
+                {
+                    "entity": "person",
+                    "column": column,
+                    "dtype": dtypes[column],
+                    "rewrite": True,
+                }
+                for column in columns
+            ],
+        },
+        *moved,
+    ]
+
+
+def _v3_rewrite_in_the_calibration_base(rows, nodes, binding, columns):
+    """G7: V3's calibration opens from the review version, rewrite and all."""
+    _v3_review_version(rows, nodes, columns)
+    rows["nz.v.V3.calibrate"]["base"] = "nz.v.V3.review"
+
+
+def _v3_rewrite_behind_the_problem(rows, nodes, binding, columns):
+    """G7 in NZ's placement: V3's calibration keeps its base on V3.
+
+    The problem moves into the review version and reads the rewritten
+    columns, so only the calibration's artifact edge from the problem puts
+    the rewrite among its compiled ancestors.
+    """
+    _v3_review_version(rows, nodes, columns)
+
+
+_V3_SECOND_WRITER = (
+    "calibration node 'nz.v.V3.calibrate' has writers "
+    "['nz.v.V3.income.standin', 'nz.v.V3.review.rewrite'] of "
+    "person.{income} upstream; only its producer 'nz.v.V3.income.standin' "
+    "may write it"
+)
+
+
+@pytest.mark.parametrize(
+    ("mutant", "reason"),
+    [
+        (
+            _v3_without_declaration,
+            "0 nodes declaring measurement_id {binding!r} sit upstream of "
+            "calibration node 'nz.v.V3.calibrate'",
+        ),
+        (
+            _v3_second_declarer,
+            "2 nodes declaring measurement_id {binding!r} sit upstream of "
+            "calibration node 'nz.v.V3.calibrate'",
+        ),
+        (
+            _v3_universe_elsewhere,
+            "its one declarer 'nz.v.V3.income.standin' does not write "
+            "['person.{universe}']",
+        ),
+        (_v3_rewrite_in_the_calibration_base, _V3_SECOND_WRITER),
+        (_v3_rewrite_behind_the_problem, _V3_SECOND_WRITER),
+    ],
+)
+def test_composition_refuses_a_scenario_calibration_the_declaration_never_sees(
+    toy, mutant, reason
+):
+    fixture, _ = toy
+    binding, proof = _toy_measurement_proof(fixture)
+    income, universe = (name.split(".", 1)[1] for name in proof["columns"])
+    spec = copy.deepcopy(fixture.spec)
+    (scenario,) = (
+        row for row in spec["resources"]["scenarios"]["scenarios"] if row["id"] == "V3"
+    )
+    rows = {row["id"]: row for row in scenario["nodes"]}
+    mutant(rows, scenario["nodes"], binding, (income, universe))
+    # Scenario rows sit outside the declared graph, so the declaration still
+    # proves the measurement and only the extension pass can refuse.
+    assert binding in measurement_proofs(spec)[0]
+    with pytest.raises(ValueError) as refused:
+        compose_transport_graph(spec, fixture.config)
+    message = str(refused.value)
+    assert message.startswith(
+        f"Measurement {binding!r}: the composed graph proves no producer for "
+        "extension calibration node 'nz.v.V3.calibrate'; "
+    )
+    assert reason.format(binding=binding, income=income, universe=universe) in message