From 97038888a51c85629290118972c8cbc6d122176c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 16:27:52 +0000 Subject: [PATCH 1/8] Sum published bands in money rows and prove measurement producers scale_money_and_intervals now accepts a summed money row: an identity row with no aggregate_fact_key, a selector that pins a dimension to a list of members, and an expected_member_count equal to the cells those lists span. The resolver emits one operand {dimension_values, expected_member_count, weight}; the ledger's multi-member linear_combination path sums it. The count companion keeps value operation sum over the same members, because the ledger's identity path resolves one fact. The resolver refuses a wrong member count, a non-identity money row, count and money member sets that differ, and members two rows share. The receipt records each factor's binary64 weight and its exact relative rounding (R2C-10). compose.measurement_proofs satisfies measurement: only when, for every calibration node, exactly one upstream node declares measurement_id: and writes the scaled table's income and universe columns on its entity (R2C-1). Upstream follows the graph compiler's predecessor rule over the declared rows. reference_profile_view passes the proofs to the resolver; a proven row becomes active, any other row stays parked, and its activation gap names the measurement and the reason. Invariants: a compiled summed amount equals the factor times the sum of its members, and the count equals the sum of its members; a row is active exactly when its measurement is proven. The tests in test_reference_summed_bands.py and test_transport_measurement_proof.py cover them, but this commit is pushed before the environment finished installing, so they have not run yet. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01N2tyHP1LrUqzw1Kaau8bvP --- .../src/microcosm/build/reference_profiles.py | 312 +++++++++++++++-- .../src/microcosm/build/transport/compose.py | 224 ++++++++++++- .../shared/test_reference_summed_bands.py | 317 ++++++++++++++++++ .../test_transport_measurement_proof.py | 248 ++++++++++++++ .../microcosm_build/reference_intervals.py | 179 +++++++++- 5 files changed, 1247 insertions(+), 33 deletions(-) create mode 100644 packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py create mode 100644 packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py diff --git a/packages/microcosm-build/src/microcosm/build/reference_profiles.py b/packages/microcosm-build/src/microcosm/build/reference_profiles.py index 5354a795bc..7f9e16d61a 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_profiles.py +++ b/packages/microcosm-build/src/microcosm/build/reference_profiles.py @@ -69,9 +69,42 @@ def _selection_identity(receipt: Mapping) -> dict: identity["transformed_names"] = receipt["transformed_names"] if "interval_tables" in receipt: identity["interval_tables_sha256"] = receipt["interval_tables_sha256"] + if "measurement_proofs" in receipt: + identity["measurement_proofs"] = receipt["measurement_proofs"] return identity +def _measurement_proofs(proofs: Mapping | None) -> dict: + """Validate composition's proof map. + + Each proof names its measurement and, for every calibration node, the + one producer node upstream of it. + """ + if proofs is None: + return {} + proofs = _mapping(proofs, "measurement proofs") + for binding, proof in proofs.items(): + proof = _mapping(proof, "measurement proof") + producers = proof.get("producers") + if ( + not isinstance(binding, str) + or not binding + or proof.get("measurement_id") != binding + or not isinstance(producers, Mapping) + or not producers + or any( + not isinstance(key, str) or not key or not isinstance(value, str) + or not value + for key, value in producers.items() + ) + ): + raise ValueError( + "A measurement proof names its measurement and, for each " + "calibration node, one producer node." + ) + return _copy(proofs) + + def validate_reference_factor_evidence(receipt: Mapping) -> dict: """Bind selected declarations and receipts without claiming feed authenticity.""" from microcosm.build.reference_factors import validate_factor_context @@ -143,6 +176,20 @@ def _explicit_names(transform: Mapping, field: str) -> list[str]: return names +def _member_order(value: object) -> bytes: + return canonical_json(value) + + +def _canonical_pins(pins: object) -> object: + """Order a summed row's member lists, so a pair compares as member sets.""" + if not isinstance(pins, Mapping): + return pins + return { + key: sorted(value, key=_member_order) if isinstance(value, list) else value + for key, value in pins.items() + } + + def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: selector = _mapping(row.get("ledger_selector"), "companion selector") excluded = {"aggregate_fact_key", "source_record_id"} @@ -152,7 +199,9 @@ def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: excluded.add("dimension_values") payload = { "selector": { - key: value for key, value in selector.items() if key not in excluded + key: _canonical_pins(value) if key == "dimension_values" else value + for key, value in selector.items() + if key not in excluded }, "entity": row.get("entity"), "period": row.get("period"), @@ -163,6 +212,161 @@ def _companion_key(row: Mapping, *, scope: bool = False) -> bytes: return canonical_json(payload) +def _summed_members(row: Mapping) -> frozenset | None: + """The member cells of a summed row, or ``None`` for a singleton row. + + A summed row names no ``aggregate_fact_key``. Its selector pins at least + one dimension to a list of members, and its ``expected_member_count`` + equals the number of cells those lists span (a scalar pin spans one). + The ledger sums exactly those cells (``ledger_targets.py`` + ``_resolve_linear_combination_reference_facts`` for money and + ``_resolve_sum_reference_facts`` for counts), so a declared count that + disagrees with the lists would sum a different member set than the + reviewer read. + """ + selector = _mapping(row.get("ledger_selector"), "summed selector") + pins = selector.get("dimension_values") + listed = isinstance(pins, Mapping) and any( + isinstance(value, list) for value in pins.values() + ) + if "aggregate_fact_key" in selector: + if listed or "expected_member_count" in row: + raise ValueError( + f"Reference {row.get('name')!r} names a singleton fact key and " + "a member set; a summed row names members only." + ) + return None + if not listed: + return None + cells = [()] + for key in sorted(pins): + value = pins[key] + members = value if isinstance(value, list) else [value] + ordered = sorted(members, key=_member_order) + if any( + _member_order(item) == _member_order(previous) + for item, previous in zip(ordered[1:], ordered, strict=False) + ) or any(isinstance(item, (Mapping, list)) for item in members): + raise ValueError( + f"Reference {row.get('name')!r} lists dimension {key!r} members " + "that repeat or are not scalars." + ) + cells = [ + (*cell, (key, _member_order(item).decode())) for cell in cells for item in members + ] + count = row.get("expected_member_count") + if type(count) is not int or count != len(cells): + raise ValueError( + f"Reference {row.get('name')!r} lists {len(cells)} members but " + f"declares expected_member_count {count!r}." + ) + return frozenset(cells) + + +def _row_cells(row: Mapping) -> frozenset: + """The dimension cells one row selects, singleton or summed.""" + members = _summed_members(row) + if members is not None: + return members + pins = _mapping(row.get("ledger_selector"), "selector").get("dimension_values") + if not isinstance(pins, Mapping): + return frozenset() + return frozenset( + { + tuple( + (key, _member_order(pins[key]).decode()) for key in sorted(pins) + ) + } + ) + + +def _validate_summed_rows(money: list, counts: list, by_name: Mapping) -> None: + """Pair summed money and count rows on one member set, without overlap. + + A summed money row is an ``identity`` row the resolver turns into one + weighted operand. Its count companion keeps ``sum`` semantics over the + same members: the ledger's ``identity`` path resolves a single fact, so + it cannot sum a member set. Within one selector scope no two rows of one + role may share a member, because the summed facts would then count twice. + """ + pairs = {_companion_key(by_name[name]): name for name in counts} + summed = set() + for name in money: + money_row = by_name[name] + members = _summed_members(money_row) + count_name = pairs.get(_companion_key(money_row)) + count_members = ( + None if count_name is None else _summed_members(by_name[count_name]) + ) + if members is None and count_members is None: + continue + summed.update({name, count_name}) + if members != count_members or count_name is None: + raise ValueError( + f"Summed references {name!r} and {count_name!r} select " + "different member sets." + ) + count_row = by_name[count_name] + if count_row.get("value_operation") != "sum" or count_row.get( + "value_operands", [] + ): + raise ValueError( + f"Summed count reference {count_name!r} must sum its members " + "with value_operation 'sum'." + ) + for role in (money, counts): + if not summed & set(role): + continue + seen: dict[tuple, str] = {} + for name in role: + scope = _companion_key(by_name[name], scope=True) + for cell in _row_cells(by_name[name]): + other = seen.setdefault((scope, cell), name) + if other != name and summed & {name, other}: + raise ValueError( + f"References {other!r} and {name!r} share a summed member." + ) + + +def _money_operand(row: Mapping, weight: float) -> dict: + """The one weighted operand that scales a money row by its factor.""" + selector = _mapping(row.get("ledger_selector"), "money selector") + members = _summed_members(row) + if members is not None: + return { + "dimension_values": _copy(selector["dimension_values"]), + "expected_member_count": row["expected_member_count"], + "weight": weight, + } + key = selector.get("aggregate_fact_key") + if not isinstance(key, str) or not key: + raise ValueError( + "Money scaling needs a singleton source fact key or a summed member set." + ) + return {"aggregate_fact_key": key, "expected_member_count": 1, "weight": weight} + + +def _binary64_weight(rational: Mapping) -> dict: + """State how far the binary64 operand weight sits from the exact factor. + + Interval cuts stay exact rationals, while the ledger multiplies each + member value by a float weight. The record gives the float, the exact + factor and the exact relative rounding as a reduced fraction. + """ + from fractions import Fraction + + exact = Fraction(rational["numerator"], rational["denominator"]) + weight = rational["numerator"] / rational["denominator"] + relative = (Fraction(weight) - exact) / exact + return { + "weight": weight, + "exact_numerator": rational["numerator"], + "exact_denominator": rational["denominator"], + "binary64_relative_rounding": f"{relative.numerator}/{relative.denominator}", + "rounding": "round_half_even_binary64", + } + + def _validate_companions(transform: Mapping, by_name: Mapping) -> tuple[list, list]: money = _explicit_names(transform, "reference_names") counts = _explicit_names(transform, "count_reference_names") @@ -187,7 +391,8 @@ def _validate_companions(transform: Mapping, by_name: Mapping) -> tuple[list, li } if complete != set(names): raise ValueError("Money and count rosters are incomplete companions.") - money_keys = [_companion_key(by_name[name]) for name in money] + _validate_summed_rows(money, counts, by_name) + money_keys =[_companion_key(by_name[name]) for name in money] count_keys = [_companion_key(by_name[name]) for name in counts] if len(set(money_keys)) != len(money) or set(money_keys) != set(count_keys): raise ValueError("Money and count rosters must match one-to-one companions.") @@ -224,6 +429,15 @@ def _factor_companions(chosen: Mapping, by_name: Mapping) -> dict: "value_operation", "identity" ), "value_operands": row.get("value_operands", []), + **( + { + "expected_member_count": row[ + "expected_member_count" + ] + } + if "expected_member_count" in row + else {} + ), "metadata": { "reference_value_kind": row["metadata"][ "reference_value_kind" @@ -419,7 +633,17 @@ def _validate_interval_resolution( raise ValueError("Effective interval table scaling identity mismatch.") -def _satisfy_factor_prerequisite(row: dict, dimension: str, binding: str) -> list: +def _satisfy_factor_prerequisite( + row: dict, dimension: str, binding: str, *, proven: bool = False +) -> list: + """Satisfy the profile half, and the measurement half only on a proof. + + ``proven`` is composition's result (``compose.measurement_proofs``): it + is true only when exactly one node upstream of every calibration node + declares the measurement and writes its interval columns. A proven row + becomes active; any other row stays parked with the measurement named in + ``reference_prerequisites_remaining``. + """ metadata = row.setdefault("metadata", {}) if metadata.get("activation_status") != REQUIRES_PREREQUISITES: raise ValueError( @@ -446,10 +670,15 @@ def _satisfy_factor_prerequisite(row: dict, dimension: str, binding: str) -> lis raise ValueError( "Reference prerequisites must name the profile and measurement." ) - remaining = [ - item for item in prerequisites if item != f"reference_profile:{dimension}" - ] + satisfied = {f"reference_profile:{dimension}"} + if proven: + satisfied.add(f"measurement:{binding}") + remaining = [item for item in prerequisites if item not in satisfied] metadata["reference_prerequisites_remaining"] = canonical_json(remaining).decode() + if not remaining: + metadata["activation_status"] = "active" + metadata.pop("activation_reason", None) + metadata.pop("parking_reason", None) return remaining @@ -460,6 +689,7 @@ def resolve_reference_profiles( interval_document: Mapping | None = None, *, factor_context: Mapping | None = None, + measurement_proofs: Mapping | None = None, ) -> tuple[dict, dict | None, dict]: """Return effective references, effective intervals and a bound receipt. @@ -470,7 +700,10 @@ def resolve_reference_profiles( scaled by the same selected factor as their money companions; without named tables the interval document passes through unchanged. Resource headers are bound into the base identity; effective references use - the compiler's closed document shape. + the compiler's closed document shape. ``measurement_proofs`` is + composition's proof result (``compose.measurement_proofs``), keyed by + measurement id: only a proven binding satisfies ``measurement:``, and + the proofs a selection uses enter its identity. """ base = _copy(_mapping(base_reference_document, "base references")) declarations = _copy(_mapping(profile_declarations, "profile declarations")) @@ -528,6 +761,17 @@ def resolve_reference_profiles( if not isinstance(factor_id, str) or not factor_id: raise ValueError("Money scaling needs a declared factor id.") factor_ids.add(factor_id) + proofs = _measurement_proofs(measurement_proofs) + bindings = { + transform.get("required_measurement_binding") + for choice in chosen.values() + for transform in choice["declaration"].get("transformations", []) + if isinstance(transform, Mapping) + and transform.get("operation") == SCALE_MONEY_AND_INTERVALS + } + used_proofs = { + binding: proofs[binding] for binding in sorted(bindings & set(proofs)) + } selected_factors = {} if factor_ids: from microcosm.build.reference_factors import validate_factor_context @@ -579,6 +823,12 @@ def resolve_reference_profiles( interval_tables=interval_records, interval_tables_sha256=reference_document_sha256(interval_records), ) + receipt["factor_weight_rounding"] = { + factor_id: _binary64_weight(entry["operational_factor"]) + for factor_id, entry in sorted(selected_factors.items()) + } + if used_proofs: + receipt["measurement_proofs"] = _copy(used_proofs) selection_hash = reference_document_sha256(_selection_identity(receipt)) interval_hash = None if intervals is None else reference_document_sha256(intervals) transformed, policies = {}, [] @@ -642,7 +892,9 @@ def resolve_reference_profiles( raise ValueError( "Factor and reference source/target periods are incompatible." ) - remaining = _satisfy_factor_prerequisite(row, dimension, binding) + remaining = _satisfy_factor_prerequisite( + row, dimension, binding, proven=binding in used_proofs + ) if name in money: if row.get( "value_operation", "identity" @@ -650,22 +902,8 @@ def resolve_reference_profiles( raise ValueError( "Money scaling requires an untransformed source reference." ) - selector = _mapping( - row.get("ledger_selector"), "money selector" - ) - key = selector.get("aggregate_fact_key") - if not isinstance(key, str) or not key: - raise ValueError( - "Money scaling needs a singleton source fact key." - ) row["value_operation"] = "linear_combination" - row["value_operands"] = [ - { - "aggregate_fact_key": key, - "expected_member_count": 1, - "weight": weight, - } - ] + row["value_operands"] = [_money_operand(row, weight)] row_metadata = row.setdefault("metadata", {}) row_metadata.update(metadata) row_metadata["reference_selection_sha256"] = selection_hash @@ -830,6 +1068,12 @@ def validate_reference_resolution( ) != resolution.get("effective_intervals_sha256"): raise ValueError("Effective interval document hash mismatch.") _validate_interval_resolution(resolution, factors, interval_document) + if factors and resolution.get("factor_weight_rounding") != { + factor_id: _binary64_weight(entry["operational_factor"]) + for factor_id, entry in sorted(factors.items()) + }: + raise ValueError("Factor weight rounding record mismatch.") + _measurement_proofs(resolution.get("measurement_proofs")) factor_bindings = _factor_reference_bindings(resolution) transformed_names = resolution.get("transformed_names") if not isinstance(transformed_names, list) or not all( @@ -906,15 +1150,23 @@ def validate_reference_resolution( or action.get("target_period") != declaration["target_period"] ): raise ValueError("Effective reference factor identity mismatch.") + proven = metadata.get("required_measurement_binding") in resolution.get( + "measurement_proofs", {} + ) + if (metadata.get("activation_status") == "active") is not proven: + raise ValueError( + "A factor companion is active exactly when composition " + "proved its measurement." + ) if role == "money": expected = [ - { - "aggregate_fact_key": source.get("ledger_selector", {}).get( - "aggregate_fact_key" - ), - "expected_member_count": 1, - "weight": rational["numerator"] / rational["denominator"], - } + _money_operand( + { + **source, + "name": name, + }, + rational["numerator"] / rational["denominator"], + ) ] if ( row.get("value_operation") != "linear_combination" diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 39b616895b..c9e8a825fa 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -252,6 +252,7 @@ "artifact_inputs", "artifact_outputs", "rules_binding", + "measurement_id", } ) _ENCODINGS = frozenset({"value", "json", "sha256"}) @@ -458,6 +459,204 @@ def _raw_transport_resource(spec: Mapping[str, object], name: str) -> Mapping: return _mapping(value, f"{stem}.json") +def _measurement_tables(spec: Mapping[str, object]) -> dict[str, list[Mapping]]: + """Each money-scaling measurement binding with the interval tables it scales. + + Every declared profile counts, selected or not, so a proof never depends + on which profile a scenario selects. + """ + declarations = _raw_transport_resource(spec, "reference_profiles") + interval = declarations.get("interval_resource") + tables = {} + if interval is not None and _has_resource(spec, interval): + tables = { + table.get("id"): table + for table in _raw_transport_resource(spec, interval).get("tables", ()) + if isinstance(table, Mapping) + } + found: dict[str, list[Mapping]] = {} + dimensions = declarations.get("dimensions", {}) + for dimension in dimensions.values() if isinstance(dimensions, Mapping) else (): + profiles = dimension.get("profiles", {}) if isinstance(dimension, Mapping) else {} + for profile in profiles.values() if isinstance(profiles, Mapping) else (): + for transform in ( + profile.get("transformations", ()) if isinstance(profile, Mapping) else () + ): + if ( + not isinstance(transform, Mapping) + or transform.get("operation") != "scale_money_and_intervals" + ): + continue + binding = transform.get("required_measurement_binding") + if not isinstance(binding, str) or not binding: + continue + listed = found.setdefault(binding, []) + for table_id in transform.get("interval_table_ids", ()): + table = tables.get(table_id) + if table is not None and table not in listed: + listed.append(table) + return found + + +def _declared_predecessors(rows: Mapping[str, Mapping]) -> dict[str, set[str]]: + """Declared-row predecessors, by the rule ``graph.decl.compile_graph`` uses. + + A node in version ``V`` follows ``V`` and the owner in ``V`` of each + column it reads (the version itself when no node in ``V`` owns it). A + structural node follows its base and every member of its base. Artifact + edges join the same sets. + """ + structural = { + name for name, row in rows.items() if row.get("structural", "none") != "none" + } + default = next(iter(structural)) if len(structural) == 1 else None + owners: dict[tuple, str] = {} + members: dict[object, set[str]] = {} + for name, row in rows.items(): + if row.get("structural", "none") == "none": + version = row.get("population") or default + members.setdefault(version, set()).add(name) + for entity, column in _owned_columns(row): + owners[version, entity, column] = name + predecessors: dict[str, set[str]] = {name: set() for name in rows} + for name, row in rows.items(): + if row.get("structural", "none") == "create": + continue + if row.get("structural", "none") != "none": + version = row.get("base") + predecessors[name].add(version) + predecessors[name].update(members.get(version, ())) + rewritten = set() + else: + version = row.get("population") or default + predecessors[name].add(version) + rewritten = _owned_columns(row, rewrite=True) + for item in row.get("inputs", ()): + if not isinstance(item, Mapping): + continue + for column in item.get("columns", ()): + if (item.get("entity"), column) not in rewritten: + predecessors[name].add( + owners.get((version, item.get("entity"), column), version) + ) + predecessors[name].update( + edge.get("producer") + for edge in row.get("artifact_inputs", ()) + if isinstance(edge, Mapping) + ) + predecessors[name].discard(name) + return predecessors + + +def measurement_proofs( + spec: Mapping[str, object], +) -> tuple[dict[str, dict], dict[str, str]]: + """Prove each money-scaling measurement's producer from the declaration. + + G8-r2c ruling R2C-1: composition satisfies ``measurement:`` only when, + for every calibration node, exactly one node upstream of it declares + ``measurement_id: `` and writes the income and universe columns of + every interval table the measurement's profiles scale, on that table's + entity. Upstream follows the graph compiler's own predecessor rule over + the declared ``transport_graph`` rows (:func:`_declared_predecessors`). + Returns the proofs by measurement id and, for each measurement without + one, the reason it stays parked. + """ + if not _has_resource(spec, "reference_profiles"): + return {}, {} + wanted = _measurement_tables(spec) + if not wanted: + return {}, {} + nodes = ( + _raw_transport_resource(spec, GRAPH_RESOURCE).get("nodes", ()) + if _has_resource(spec, GRAPH_RESOURCE) + else () + ) + rows = { + row["id"]: row + for row in nodes + if isinstance(row, Mapping) and isinstance(row.get("id"), str) + } + document = ( + _raw_transport_resource(spec, GRAPH_RESOURCE) + if _has_resource(spec, GRAPH_RESOURCE) + else {} + ) + roots = { + name + for name, row in rows.items() + if isinstance(row.get("kernel"), str) and row["kernel"].startswith("calibrate.") + } + endpoint = (document.get("endpoints") or {}).get("calibration") + if isinstance(endpoint, str) and endpoint in rows: + roots.add(endpoint) + predecessors = _declared_predecessors(rows) + + def ancestors(root: str) -> set[str]: + seen, pending = set(), [root] + while pending: + name = pending.pop() + if name in seen or name not in rows: + continue + seen.add(name) + pending.extend(predecessors[name]) + return seen + + upstream = {root: ancestors(root) for root in sorted(roots)} + proofs: dict[str, dict] = {} + refusals: dict[str, str] = {} + for binding, tables in sorted(wanted.items()): + required = sorted( + { + (table.get("entity"), table.get(field)) + for table in tables + for field in ("income_column", "universe_column") + }, + key=str, + ) + declared = sorted( + name for name, row in rows.items() if row.get("measurement_id") == binding + ) + producers, reason = {}, None + if not tables: + reason = "its profiles scale no declared interval table" + elif not upstream: + reason = "the graph declares no calibration node" + for root, closure in upstream.items(): + if reason is not None: + break + found = [name for name in declared if name in closure] + if len(found) != 1: + reason = ( + f"{len(found)} nodes declaring measurement_id {binding!r} " + f"sit upstream of calibration node {root!r} " + f"(declared by {declared}); exactly one must" + ) + break + (producer,) = found + missing = [ + f"{entity}.{column}" + for entity, column in required + if (entity, column) not in _owned_columns(rows[producer]) + ] + if missing: + reason = ( + f"producer {producer!r} does not write {missing} on the " + "interval table's entity" + ) + break + producers[root] = producer + if reason is None: + proofs[binding] = { + "measurement_id": binding, + "producers": producers, + "columns": [f"{entity}.{column}" for entity, column in required], + } + else: + refusals[binding] = reason + return proofs, refusals + + def reference_profile_view( spec: Mapping[str, object], selection: Mapping | None = None ): @@ -486,6 +685,7 @@ def reference_profile_view( factor_context=None if factor_context_resource is None else _raw_transport_resource(spec, factor_context_resource), + measurement_proofs=measurement_proofs(spec)[0], ) if _has_resource(spec, CHRONICLE_FEED_RESOURCE): feed_pin = _raw_transport_resource(spec, CHRONICLE_FEED_RESOURCE) @@ -642,8 +842,10 @@ def _reference_activation_gaps( resource: str, exempt: Mapping | None = None, selected: Mapping[str, Sequence[str]] | None = None, + refusals: Mapping[str, str] | None = None, ) -> Iterator[str]: # The absent-resource path preserves the legacy activation walk and messages. + # A row parked on an unproved measurement names it and the proof's reason. exemptions = ( set() if exempt is None @@ -663,7 +865,24 @@ def _reference_activation_gaps( reference.get("name") not in exemptions and metadata.get("activation_status", "active") != "active" ): - yield f"placeholder reference {resource}.{reference.get('name', '?')}" + gap = f"placeholder reference {resource}.{reference.get('name', '?')}" + try: + remaining = json.loads( + metadata.get("reference_prerequisites_remaining", "[]") + ) + except (TypeError, ValueError): + remaining = [] + measurements = [ + item.removeprefix("measurement:") + for item in remaining if isinstance(item, str) + and item.startswith("measurement:") + ] + if measurements: + gap += "; awaits " + ", ".join( + f"measurement:{item} ({(refusals or {}).get(item, 'no proof')})" + for item in measurements + ) + yield gap def _reference_selections(spec: Mapping[str, object]) -> dict[str, dict]: @@ -1110,6 +1329,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: missing = [] # Resolve first so activation never inspects an executable profile's raw rows. reference_profile_views(spec) + _, measurement_refusals = measurement_proofs(spec) reference_selections = _reference_selections(spec) selection_closure = ( _reference_selection_closure(spec) @@ -1209,6 +1429,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: resource, reference_selections.get(resource), selection_closure.get(resource), + measurement_refusals, ) ) for (resource, identity), document in selected_reference_views.items(): @@ -1219,6 +1440,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: resource, reference_selections.get(resource), selection_closure.get(resource), + measurement_refusals, ) ) receipt = ( diff --git a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py new file mode 100644 index 0000000000..bb8235c3ea --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py @@ -0,0 +1,317 @@ +"""Summed-band money rows in ``scale_money_and_intervals`` (G8-r2c hunk A). + +A coarse money row sums several published bands: the resolver turns its +``identity`` row into one weighted operand over the member set, and its count +companion keeps ``sum`` semantics over the same members. Compilation runs the +real ledger, so these tests check the arithmetic the matrix receives. +""" + +from __future__ import annotations + +import copy +import json +import math +from fractions import Fraction + +import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st + +from microcosm.build.country_spec import _validate_target_references +from microcosm.build.ledger_targets import compile_ledger_target_references +from microcosm.build.reference_profiles import ( + resolve_reference_profiles, + validate_reference_resolution, +) +from test_support.microcosm_build.reference_intervals import ( + SUMMED_MEASUREMENT, + band_fact, + proof_for, + summed_profile_fixture, + summed_row, +) + +TEN = [f"n{index:02d}" for index in range(10)] + + +def _facts(labels, *, count=lambda i: 100 + i, money=lambda i: 1_000 * (i + 1)): + facts = [] + for index, label in enumerate(labels): + facts.append(band_fact("count", label, count(index))) + facts.append(band_fact("money", label, money(index))) + return facts + + +def _resolve(inputs, proofs=None, selection=None): + base, profiles, context, intervals = inputs + return resolve_reference_profiles( + base, + profiles, + selection or {}, + intervals, + factor_context=context, + measurement_proofs=proofs, + ) + + +def _compile(effective, facts): + references = _validate_target_references(effective, country=effective["country"]) + registry = compile_ledger_target_references( + facts, references, country=effective["country"] + ) + return {spec.name: spec for spec in registry.specs} + + +def _rows(document): + return {row["name"]: row for row in document["target_references"]} + + +def test_ten_member_pair_resolves_to_one_operand_with_count_ten(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs, proof_for()) + rows = _rows(effective) + money, count = rows["toy_money_b0"], rows["toy_count_b0"] + assert money["value_operation"] == "linear_combination" + assert money["value_operands"] == [ + { + "dimension_values": {"band": TEN}, + "expected_member_count": 10, + "weight": 1.1, + } + ] + assert count["value_operation"] == "sum" + assert count["expected_member_count"] == 10 + assert "value_operands" not in count or count["value_operands"] == [] + for row in (money, count): + assert row["metadata"]["activation_status"] == "active" + assert json.loads(row["metadata"]["reference_prerequisites_remaining"]) == [] + assert "parking_reason" not in row["metadata"] + assert receipt["measurement_proofs"] == proof_for() + assert validate_reference_resolution(effective, receipt) == receipt + + +def test_summed_money_operand_resolves_amount_facts_only(): + """Counts and amounts share one record set and one band dimension.""" + inputs = summed_profile_fixture([TEN]) + effective, _, _ = _resolve(inputs, proof_for()) + facts = _facts(TEN) + specs = _compile(effective, facts) + money_keys = json.loads(specs["toy_money_b0"].metadata["ledger_member_fact_keys"]) + count_keys = json.loads(specs["toy_count_b0"].metadata["ledger_member_fact_keys"]) + assert sorted(money_keys) == sorted( + f"ledger.aggregate_fact.v2:toy-money-{label}" for label in TEN + ) + assert sorted(count_keys) == sorted( + f"ledger.aggregate_fact.v2:toy-count-{label}" for label in TEN + ) + assert specs["toy_count_b0"].value == sum(100 + i for i in range(10)) + assert math.isclose( + specs["toy_money_b0"].value, + 1.1 * sum(1_000 * (i + 1) for i in range(10)), + rel_tol=1e-15, + ) + + +def test_without_a_proof_rows_stay_parked_and_compilation_refuses(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs) + for row in effective["target_references"]: + assert row["metadata"]["activation_status"] == "requires_prerequisites" + assert json.loads(row["metadata"]["reference_prerequisites_remaining"]) == [ + f"measurement:{SUMMED_MEASUREMENT}" + ] + assert "measurement_proofs" not in receipt + assert validate_reference_resolution(effective, receipt) == receipt + with pytest.raises(ValueError, match="non-executable placeholder"): + _compile(effective, _facts(TEN)) + + +def test_a_proof_for_another_measurement_does_not_satisfy_this_one(): + inputs = summed_profile_fixture([TEN]) + other = { + "other": { + "measurement_id": "other", + "producers": {"toy.calibrate": "toy.other"}, + } + } + effective, _, receipt = _resolve(inputs, other) + assert { + row["metadata"]["activation_status"] for row in effective["target_references"] + } == {"requires_prerequisites"} + assert "measurement_proofs" not in receipt + + +def test_an_active_row_without_its_proof_in_the_receipt_refuses(): + inputs = summed_profile_fixture([TEN]) + effective, _, receipt = _resolve(inputs, proof_for()) + parked, _, parked_receipt = _resolve(inputs) + with pytest.raises(ValueError): + validate_reference_resolution(effective, parked_receipt) + forged = copy.deepcopy(parked) + for row in forged["target_references"]: + row["metadata"]["activation_status"] = "active" + with pytest.raises(ValueError): + validate_reference_resolution(forged, parked_receipt) + assert validate_reference_resolution(effective, receipt) == receipt + + +def _mutate(mutation): + groups = [TEN[:5], TEN[5:]] + base, profiles, context, intervals = summed_profile_fixture(groups) + rows = _rows(base) + if mutation == "money_member_count": + rows["toy_money_b0"]["expected_member_count"] = 4 + elif mutation == "count_member_count": + rows["toy_count_b0"]["expected_member_count"] = 6 + elif mutation == "non_identity_money": + rows["toy_money_b0"]["value_operation"] = "sum" + elif mutation == "mismatched_members": + rows["toy_count_b0"]["ledger_selector"]["dimension_values"]["band"] = [ + *TEN[:4], + "elsewhere", + ] + elif mutation == "overlapping_members": + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b1"]["ledger_selector"] + selector["dimension_values"]["band"] = [TEN[4], *TEN[6:]] + elif mutation == "identity_count": + rows["toy_count_b0"]["value_operation"] = "identity" + elif mutation == "fact_key_and_members": + rows["toy_money_b0"]["ledger_selector"]["aggregate_fact_key"] = "toy-money-x" + elif mutation == "repeated_member": + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b0"]["ledger_selector"] + selector["dimension_values"]["band"] = [*TEN[:4], TEN[0]] + else: # pragma: no cover - parametrization guard + raise AssertionError(mutation) + return base, profiles, context, intervals + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + ("money_member_count", "lists 5 members but declares expected_member_count 4"), + ("count_member_count", "lists 5 members but declares expected_member_count 6"), + ("non_identity_money", "untransformed source reference"), + ("mismatched_members", "different member sets"), + ("overlapping_members", "share a summed member"), + ("identity_count", "value_operation 'sum'"), + ("fact_key_and_members", "singleton fact key and a member set"), + ("repeated_member", "repeat or are not scalars"), + ], +) +def test_summed_rows_refuse_each_inconsistency(mutation, message): + with pytest.raises(ValueError, match=message): + _resolve(_mutate(mutation), proof_for()) + + +def test_receipt_states_the_binary64_rounding_of_the_weight(): + inputs = summed_profile_fixture([TEN]) + _, _, receipt = _resolve(inputs, proof_for()) + record = receipt["factor_weight_rounding"]["toy_central"] + exact = Fraction(record["exact_numerator"], record["exact_denominator"]) + assert record["weight"] == 1.1 + assert Fraction(record["binary64_relative_rounding"]) == ( + (Fraction(1.1) - exact) / exact + ) + assert abs(Fraction(record["binary64_relative_rounding"])) <= Fraction(1, 2**53) + + +def _partition(draw_sizes, labels): + groups, start = [], 0 + for size in draw_sizes: + groups.append(labels[start : start + size]) + start += size + return groups + + +@given( + sizes=st.lists(st.integers(min_value=1, max_value=6), min_size=1, max_size=5), + values=st.lists( + st.integers(min_value=0, max_value=10**9), min_size=30, max_size=30 + ), + counts=st.lists(st.integers(min_value=0, max_value=10**6), min_size=30, max_size=30), + numerator=st.integers(min_value=500_000, max_value=2_000_000), +) +@settings( + max_examples=25, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +def test_compiled_coarse_rows_equal_factor_times_member_sums( + sizes, values, counts, numerator +): + labels = [f"m{index:02d}" for index in range(sum(sizes))] + groups = _partition(sizes, labels) + anchor = f"{numerator // 10_000}.{numerator % 10_000:04d}" + inputs = summed_profile_fixture(groups, anchors={"central": anchor}) + effective, _, receipt = _resolve(inputs, proof_for()) + rational = receipt["selected_factors"]["toy_central"]["operational_factor"] + factor = Fraction(rational["numerator"], rational["denominator"]) + assert factor == Fraction(numerator, 10**6) + facts = _facts(labels, count=lambda i: counts[i], money=lambda i: values[i]) + specs = _compile(effective, facts) + for band, members in enumerate(groups): + indices = [labels.index(label) for label in members] + assert specs[f"toy_count_b{band}"].value == sum(counts[i] for i in indices) + exact = factor * sum(values[i] for i in indices) + assert math.isclose( + specs[f"toy_money_b{band}"].value, float(exact), rel_tol=1e-12, abs_tol=1e-6 + ) + + +def _singleton_inputs(labels, anchors=None): + """Today's path: one singleton count/money pair per native band.""" + base, profiles, context, intervals = summed_profile_fixture( + [labels], anchors=anchors + ) + rows = [] + for label in labels: + for kind in ("count", "money"): + row = summed_row(kind, f"single_{kind}_{label}", [label], indicator="i") + row.pop("expected_member_count") + row["filter"] = f"single_{label}" + selector = row["ledger_selector"] + selector["dimension_values"] = {"band": label} + selector["aggregate_fact_key"] = f"ledger.aggregate_fact.v2:toy-{kind}-{label}" + if kind == "count": + row["value_operation"] = "identity" + rows.append(row) + base["target_references"] = rows + transform = profiles["dimensions"]["index"]["profiles"]["central"][ + "transformations" + ][0] + for profile in profiles["dimensions"]["index"]["profiles"].values(): + (transform,) = profile["transformations"] + transform["reference_names"] = [ + row["name"] for row in rows if row["name"].startswith("single_money") + ] + transform["count_reference_names"] = [ + row["name"] for row in rows if row["name"].startswith("single_count") + ] + transform["interval_table_ids"] = [] + return base, profiles, context, None + + +@pytest.mark.parametrize("anchor", ["105", "110", "117.1109"]) +def test_summed_path_equals_the_sum_of_singleton_rows_aged_today(anchor): + facts = _facts(TEN, money=lambda i: 123_457 * (i + 3) + 11) + summed, _, _ = _resolve( + summed_profile_fixture([TEN], anchors={"central": anchor}), proof_for() + ) + singles, _, _ = _resolve(_singleton_inputs(TEN, {"central": anchor}), proof_for()) + summed_specs = _compile(summed, facts) + single_specs = _compile(singles, facts) + assert summed_specs["toy_count_b0"].value == sum( + single_specs[f"single_count_{label}"].value for label in TEN + ) + assert math.isclose( + summed_specs["toy_money_b0"].value, + sum(single_specs[f"single_money_{label}"].value for label in TEN), + rel_tol=1e-13, + ) + for label in TEN: + operand = _rows(singles)[f"single_money_{label}"]["value_operands"][0] + assert set(operand) == {"aggregate_fact_key", "expected_member_count", "weight"} + assert operand["expected_member_count"] == 1 diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py new file mode 100644 index 0000000000..bf3f7e4ce2 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -0,0 +1,248 @@ +"""Composition's proof of a measurement producer (G8-r2c ruling R2C-1). + +``compose.measurement_proofs`` satisfies ``measurement:`` only when, for +every calibration node, exactly one node upstream of it declares +``measurement_id: `` and writes the scaled interval table's income and +universe columns on that table's entity. Each mutant below breaks one clause +and must leave the rows parked, with the activation gap naming the +measurement. +""" + +from __future__ import annotations + +import copy +import json + +import pytest + +from microcosm.build.transport.compose import ( + _reference_activation_gaps, + measurement_proofs, + reference_profile_view, +) +from test_support.microcosm_build.reference_intervals import ( + INCOME_COLUMN, + SUMMED_MEASUREMENT, + UNIVERSE_COLUMN, + summed_profile_fixture, +) + +GROUPS = [[f"n{index}" for index in range(5)], [f"n{index}" for index in range(5, 9)]] + + +def _graph(entity="person", *, producer_entity=None, measured=True): + producer_entity = entity if producer_entity is None else producer_entity + measure = { + "id": "toy.measure", + "kernel": "measure.combine@1", + "population": "toy.create", + "inputs": [{"entity": producer_entity, "columns": ["toy_wages"]}], + "outputs": [ + {"entity": producer_entity, "column": INCOME_COLUMN, "dtype": "int64"}, + {"entity": producer_entity, "column": UNIVERSE_COLUMN, "dtype": "str"}, + ], + } + if measured: + measure["measurement_id"] = SUMMED_MEASUREMENT + return { + "schema_version": 1, + "country": "toy", + "nodes": [ + { + "id": "toy.create", + "kernel": "transport.create@1", + "structural": "create", + }, + measure, + { + "id": "toy.prepare", + "kernel": "calibration.prepare@1", + "population": "toy.create", + "inputs": [ + {"entity": entity, "columns": [INCOME_COLUMN, UNIVERSE_COLUMN]} + ], + "outputs": [ + {"entity": entity, "column": "toy_income_band_b0", "dtype": "bool"} + ], + }, + { + "id": "toy.problem", + "kernel": "targets.problem@1", + "population": "toy.create", + "artifact_inputs": [{"name": "prepared", "producer": "toy.prepare"}], + }, + { + "id": "toy.calibrate", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + "artifact_inputs": [{"name": "problem", "producer": "toy.problem"}], + }, + ], + "endpoints": {"calibration": "toy.calibrate"}, + } + + +def _spec(graph, *, entity="person"): + base, profiles, context, intervals = summed_profile_fixture(GROUPS, entity=entity) + return { + "country": "toy", + "resources": { + "reference_profiles": profiles, + "target_references": base, + "factor_context": context, + "profile_intervals": intervals, + "transport_graph": graph, + }, + } + + +def _nodes(graph): + return {row["id"]: row for row in graph["nodes"]} + + +def test_one_upstream_producer_proves_and_activates_the_rows(): + spec = _spec(_graph()) + proofs, refusals = measurement_proofs(spec) + assert refusals == {} + assert proofs == { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {"toy.calibrate": "toy.measure"}, + "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], + } + } + references, _, receipt = reference_profile_view(spec) + assert receipt["measurement_proofs"] == proofs + assert { + row["metadata"]["activation_status"] for row in references["target_references"] + } == {"active"} + assert list(_reference_activation_gaps(references, "target_references")) == [] + + +def test_a_third_entity_table_passes_the_proof(): + """A table on a non-household, non-person entity, as ``tax_unit`` is.""" + spec = _spec(_graph("tax_unit"), entity="tax_unit") + proofs, refusals = measurement_proofs(spec) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["columns"] == [ + f"tax_unit.{INCOME_COLUMN}", + f"tax_unit.{UNIVERSE_COLUMN}", + ] + + +def _no_producer(graph): + del _nodes(graph)["toy.measure"]["measurement_id"] + + +def _after_calibration(graph): + # The producer reads the calibration artifact, so nothing upstream of + # calibration can read what it writes; preparation reads a CREATE column. + nodes = _nodes(graph) + nodes["toy.measure"]["artifact_inputs"] = [ + {"name": "fit", "producer": "toy.calibrate"} + ] + nodes["toy.prepare"]["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + + +def _two_producers(graph): + second = copy.deepcopy(_nodes(graph)["toy.measure"]) + second["id"] = "toy.measure.again" + second["outputs"] = [ + {"entity": "person", "column": "toy_other_income", "dtype": "int64"} + ] + graph["nodes"].insert(2, second) + _nodes(graph)["toy.prepare"]["inputs"][0]["columns"].append("toy_other_income") + + +def _wrong_column(graph): + nodes = _nodes(graph) + nodes["toy.measure"]["outputs"] = nodes["toy.measure"]["outputs"][:1] + graph["nodes"].insert( + 2, + { + "id": "toy.universe", + "kernel": "measure.combine@1", + "population": "toy.create", + "outputs": [ + {"entity": "person", "column": UNIVERSE_COLUMN, "dtype": "str"} + ], + }, + ) + + +@pytest.mark.parametrize( + ("mutant", "graph", "entity", "reason"), + [ + ("no_producer", _no_producer, "person", "0 nodes declaring"), + ("after_calibration", _after_calibration, "person", "0 nodes declaring"), + ("two_producers", _two_producers, "person", "2 nodes declaring"), + ("wrong_column", _wrong_column, "person", "does not write"), + ("wrong_entity", None, "person", "does not write"), + ("wrong_entity_tax_unit", None, "tax_unit", "does not write"), + ], +) +def test_each_proof_mutant_leaves_rows_parked_and_names_the_measurement( + mutant, graph, entity, reason +): + if graph is None: + document = _graph(entity, producer_entity="household") + else: + document = _graph(entity) + graph(document) + spec = _spec(document, entity=entity) + proofs, refusals = measurement_proofs(spec) + assert proofs == {} + assert reason in refusals[SUMMED_MEASUREMENT] + references, _, receipt = reference_profile_view(spec) + assert "measurement_proofs" not in receipt + rows = references["target_references"] + assert {row["metadata"]["activation_status"] for row in rows} == { + "requires_prerequisites" + } + gaps = list( + _reference_activation_gaps( + references, "target_references", refusals=refusals + ) + ) + assert len(gaps) == len(rows) + assert all(f"measurement:{SUMMED_MEASUREMENT} (" in gap for gap in gaps) + assert all(reason in gap for gap in gaps) + + +def test_every_calibration_node_needs_its_own_upstream_producer(): + graph = _graph() + graph["nodes"].append( + { + "id": "toy.calibrate.variant", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + } + ) + proofs, refusals = measurement_proofs(_spec(graph)) + assert proofs == {} + assert "toy.calibrate.variant" in refusals[SUMMED_MEASUREMENT] + + +def test_a_spec_without_profiles_or_scaling_proves_nothing(): + assert measurement_proofs({"country": "toy", "resources": {}}) == ({}, {}) + spec = _spec(_graph()) + for profile in spec["resources"]["reference_profiles"]["dimensions"]["index"][ + "profiles" + ].values(): + profile["transformations"] = [] + assert measurement_proofs(spec) == ({}, {}) + + +def test_proof_result_is_bound_into_the_selection_identity(): + proved, _, proved_receipt = reference_profile_view(_spec(_graph())) + document = _graph() + _no_producer(document) + parked, _, parked_receipt = reference_profile_view(_spec(document)) + assert ( + proved_receipt["reference_selection_sha256"] + != parked_receipt["reference_selection_sha256"] + ) + assert json.loads( + proved["target_references"][0]["metadata"]["reference_prerequisites_remaining"] + ) == [] + assert parked != proved diff --git a/test_support/microcosm_build/reference_intervals.py b/test_support/microcosm_build/reference_intervals.py index ad7d8bf489..bffec51fd3 100644 --- a/test_support/microcosm_build/reference_intervals.py +++ b/test_support/microcosm_build/reference_intervals.py @@ -91,8 +91,10 @@ def interval_bindings(table, *, entity=None): } -def _factor_entry(profile): - declaration, facts, identity = factor_fixture(anchor_level=PROFILE_ANCHORS[profile]) +def _factor_entry(profile, anchor=None): + declaration, facts, identity = factor_fixture( + anchor_level=PROFILE_ANCHORS[profile] if anchor is None else anchor + ) declaration["id"] = f"toy_{profile}" factor, receipt = resolve_reference_factor(declaration, facts, identity) return identity, { @@ -188,3 +190,176 @@ def interval_profile_fixture(): base = reference_document(references) intervals = {"schema_version": 1, "tables": [table]} return base, profiles, context, intervals + + +# --------------------------------------------------------------------------- +# Summed-band money rows (G8-r2c hunk A) +# --------------------------------------------------------------------------- + +SUMMED_MEASUREMENT = "reviewed_measurement" + + +def band_fact(kind, label, value, *, period=2001, record_set="toy.bands"): + """One Ledger consumer fact for a native band's count or amount. + + Count and amount facts share one record set and one band dimension, as + a published table that reports both measures over the same classes does. + """ + key = f"toy-{kind}-{label}" + return { + "aggregate_fact_key": f"ledger.aggregate_fact.v2:{key}", + "semantic_fact_key": f"ledger.semantic_fact.v2:{key}", + "legacy_fact_key": f"ledger.fact.v1:{key}", + "label": f"Toy {kind} in band {label}", + "lineage": { + "source_record_id": f"toy.bands.{kind}.{label}", + "source_cell_keys": [f"ledger.source_cell.v1:{key}"], + "source_row_keys": [], + }, + "value": value, + "period": {"type": "year", "value": period}, + "geography": {"level": "country", "id": "XT", "name": "Toy", "vintage": "v1"}, + "entity": {"name": "person", "role": "person"}, + "observed_measure": { + "source_name": "toy_office", + "source_table": "Toy bands", + "source_measure_id": kind, + "source_concept": f"toy.{kind}", + "unit": "people" if kind == "count" else "xts", + }, + "aggregation": {"method": "sum"}, + "source": { + "source_name": "toy_office", + "source_table": "Toy bands", + "source_file": "toy.csv", + "url": "https://example.invalid/toy.csv", + "vintage": "v1", + }, + "dimensions": {"band": label}, + "universe_constraints": {"domain": "toy"}, + "layout": { + "record_set_id": record_set, + "groupby_dimension": "band", + "groupby_value_id": label, + "measure_id": kind, + }, + } + + +def summed_row(kind, name, members, *, indicator, entity="person"): + """A parked coarse row over ``members``: ``sum`` counts, ``identity`` money.""" + row = { + "name": name, + "entity": entity, + "measure": "people" if kind == "count" else "toy_income", + "filter": indicator, + "period": 2002, + "family": "toy", + "category_id": "toy.population", + "period_match_policy": "latest_not_after", + "expected_member_count": len(members), + "ledger_selector": { + "source_name": "toy_office", + "source_measure_id": kind, + "source_concept": f"toy.{kind}", + "entity_name": "person", + "period_type": "year", + "period_value": 2001, + "dimensions": ["band"], + "dimension_values": {"band": list(members)}, + }, + "metadata": { + "contract_target_id": name, + "reference_value_kind": kind, + "activation_status": "requires_prerequisites", + "required_reference_profile": "index", + "required_measurement_binding": SUMMED_MEASUREMENT, + "reference_prerequisites": json.dumps( + ["reference_profile:index", f"measurement:{SUMMED_MEASUREMENT}"] + ), + "parking_reason": "Reviewed measurement is required.", + }, + } + if kind == "count": + row["value_operation"] = "sum" + return row + + +def summed_profile_fixture( + groups, *, entity="person", table_id="toy_coarse", anchors=None +): + """Coarse references, profiles, factor context and a coarse interval table. + + ``groups`` lists the native member labels of each coarse band, in band + order; band ``b`` sums ``groups[i]``. The table has no nil band, so + every coarse band is positive and names a summed count/money pair. + """ + table = interval_table_fixture( + table_id=table_id, + entity=entity, + cuts=tuple(100_000 * (index + 1) for index in range(len(groups) - 1)), + nil=False, + companions=True, + ) + references = [] + for band, members in zip(table["bands"], groups, strict=True): + for kind in ("count", "money"): + references.append( + summed_row( + kind, + f"toy_{kind}_{band['id']}", + members, + indicator=band["indicator"], + entity=entity, + ) + ) + identity, factors = None, [] + for profile in PROFILE_ANCHORS: + identity, entry = _factor_entry(profile, (anchors or {}).get(profile)) + factors.append(entry) + context = {"schema_version": 1, "feed_identity": identity, "factors": factors} + transformation = { + "operation": "scale_money_and_intervals", + "reference_names": [ + row["name"] for row in references if row["name"].startswith("toy_money") + ], + "count_reference_names": [ + row["name"] for row in references if row["name"].startswith("toy_count") + ], + "interval_table_ids": [table["id"]], + "required_measurement_binding": SUMMED_MEASUREMENT, + } + profiles = { + "schema_version": 1, + "reference_resource": "target_references", + "factor_context_resource": "factor_context", + "interval_resource": "profile_intervals", + "dimensions": { + "index": { + "default": "central", + "profiles": { + profile: { + "transformations": [ + {**deepcopy(transformation), "factor_id": f"toy_{profile}"} + ] + } + for profile in PROFILE_ANCHORS + }, + } + }, + } + base = reference_document(references) + intervals = {"schema_version": 1, "tables": [table]} + return base, profiles, context, intervals + + +def proof_for(producer="toy.measure", calibration="toy.calibrate"): + """The proof composition returns for one producer upstream of calibration.""" + return { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {calibration: producer}, + "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], + } + } + From a9cc2a97e0a47b90be0dc158de89a0ec7f95a09a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 19:18:12 +0000 Subject: [PATCH 2/8] Read measurement_id from params and check the proof on the compiled graph The measurement proof (R2C-1) now reads the id a producer declares in its params, under measurement_id, as a literal string: measure.combine@1 takes it as context.params["measurement_id"]. The node-level measurement_id field is gone from the transport node schema, and a non-string value proves nothing. Upstream of calibration now means ancestry under the S5 guard's declared rule (population or base, artifact producers and the writers of each read), factored out of _validate_precalibration_ancestry and shared. Composition then rechecks every proof against the compiled skeleton's exact predecessors and refuses a proof the compiled graph contradicts. The summed-band fixture now carries the dimension and target labels the ledger requires for a multi-fact target, so the compile tests run the real ledger. Invariant: a measurement is proven exactly when each calibration node has one declaring ancestor that writes the table's columns on its entity. Checked by test_transport_measurement_proof.py (16 passed) and test_reference_summed_bands.py (18 passed). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwQq6HQomkwKHCkdAgY9cy --- .../src/microcosm/build/reference_profiles.py | 16 +- .../src/microcosm/build/transport/compose.py | 241 ++++++++++-------- .../shared/test_reference_summed_bands.py | 12 +- .../test_transport_measurement_proof.py | 150 ++++++++++- .../microcosm_build/reference_intervals.py | 18 +- 5 files changed, 305 insertions(+), 132 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/reference_profiles.py b/packages/microcosm-build/src/microcosm/build/reference_profiles.py index 7f9e16d61a..94c2501d93 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_profiles.py +++ b/packages/microcosm-build/src/microcosm/build/reference_profiles.py @@ -93,7 +93,9 @@ def _measurement_proofs(proofs: Mapping | None) -> dict: or not isinstance(producers, Mapping) or not producers or any( - not isinstance(key, str) or not key or not isinstance(value, str) + not isinstance(key, str) + or not key + or not isinstance(value, str) or not value for key, value in producers.items() ) @@ -252,7 +254,9 @@ def _summed_members(row: Mapping) -> frozenset | None: "that repeat or are not scalars." ) cells = [ - (*cell, (key, _member_order(item).decode())) for cell in cells for item in members + (*cell, (key, _member_order(item).decode())) + for cell in cells + for item in members ] count = row.get("expected_member_count") if type(count) is not int or count != len(cells): @@ -272,11 +276,7 @@ def _row_cells(row: Mapping) -> frozenset: if not isinstance(pins, Mapping): return frozenset() return frozenset( - { - tuple( - (key, _member_order(pins[key]).decode()) for key in sorted(pins) - ) - } + {tuple((key, _member_order(pins[key]).decode()) for key in sorted(pins))} ) @@ -392,7 +392,7 @@ def _validate_companions(transform: Mapping, by_name: Mapping) -> tuple[list, li if complete != set(names): raise ValueError("Money and count rosters are incomplete companions.") _validate_summed_rows(money, counts, by_name) - money_keys =[_companion_key(by_name[name]) for name in money] + money_keys = [_companion_key(by_name[name]) for name in money] count_keys = [_companion_key(by_name[name]) for name in counts] if len(set(money_keys)) != len(money) or set(money_keys) != set(count_keys): raise ValueError("Money and count rosters must match one-to-one companions.") diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 8e8ded6711..9673a3ab7f 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -254,7 +254,6 @@ "artifact_inputs", "artifact_outputs", "rules_binding", - "measurement_id", } ) _ENCODINGS = frozenset({"value", "json", "sha256"}) @@ -479,10 +478,14 @@ def _measurement_tables(spec: Mapping[str, object]) -> dict[str, list[Mapping]]: found: dict[str, list[Mapping]] = {} dimensions = declarations.get("dimensions", {}) for dimension in dimensions.values() if isinstance(dimensions, Mapping) else (): - profiles = dimension.get("profiles", {}) if isinstance(dimension, Mapping) else {} + profiles = ( + dimension.get("profiles", {}) if isinstance(dimension, Mapping) else {} + ) for profile in profiles.values() if isinstance(profiles, Mapping) else (): for transform in ( - profile.get("transformations", ()) if isinstance(profile, Mapping) else () + profile.get("transformations", ()) + if isinstance(profile, Mapping) + else () ): if ( not isinstance(transform, Mapping) @@ -500,54 +503,69 @@ def _measurement_tables(spec: Mapping[str, object]) -> dict[str, list[Mapping]]: return found -def _declared_predecessors(rows: Mapping[str, Mapping]) -> dict[str, set[str]]: - """Declared-row predecessors, by the rule ``graph.decl.compile_graph`` uses. +MEASUREMENT_PARAM = "measurement_id" + - A node in version ``V`` follows ``V`` and the owner in ``V`` of each - column it reads (the version itself when no node in ``V`` owns it). A - structural node follows its base and every member of its base. Artifact - edges join the same sets. +def _measurement_declaration(row: Mapping) -> object: + """The ``measurement_id`` a node declares in its params, or ``None``. + + A producer declares the id as a literal string param, as + ``measure.combine@1`` reads ``context.params["measurement_id"]``. """ - structural = { - name for name, row in rows.items() if row.get("structural", "none") != "none" - } - default = next(iter(structural)) if len(structural) == 1 else None - owners: dict[tuple, str] = {} - members: dict[object, set[str]] = {} - for name, row in rows.items(): - if row.get("structural", "none") == "none": - version = row.get("population") or default - members.setdefault(version, set()).add(name) - for entity, column in _owned_columns(row): - owners[version, entity, column] = name - predecessors: dict[str, set[str]] = {name: set() for name in rows} + params = row.get("params") + return params.get(MEASUREMENT_PARAM) if isinstance(params, Mapping) else None + + +def _declared_calibration_predecessors( + rows: Mapping[str, Mapping], +) -> tuple[dict[str, set[str]], dict[str, set[tuple[str, str, str]]]]: + """Declared predecessors and reads, by the S5 ancestry guard's rule. + + A node follows its population version (or, if structural, its base and + every member of the base), each artifact producer it names, and each + writer :func:`_precalibration_declared_reads` resolves for its reads. + """ + predecessors = {name: set() for name in rows} + reads = _precalibration_declared_reads(rows) for name, row in rows.items(): - if row.get("structural", "none") == "create": - continue - if row.get("structural", "none") != "none": - version = row.get("base") - predecessors[name].add(version) - predecessors[name].update(members.get(version, ())) - rewritten = set() - else: - version = row.get("population") or default - predecessors[name].add(version) - rewritten = _owned_columns(row, rewrite=True) - for item in row.get("inputs", ()): - if not isinstance(item, Mapping): - continue - for column in item.get("columns", ()): - if (item.get("entity"), column) not in rewritten: - predecessors[name].add( - owners.get((version, item.get("entity"), column), version) - ) + base = row.get("base") + if isinstance(base, str): + predecessors[name].add(base) + predecessors[name].update( + item["id"] for item in rows.values() if item.get("population") == base + ) + elif isinstance(row.get("population"), str): + predecessors[name].add(row["population"]) predecessors[name].update( - edge.get("producer") - for edge in row.get("artifact_inputs", ()) - if isinstance(edge, Mapping) + edge["producer"] for edge in row.get("artifact_inputs", ()) ) - predecessors[name].discard(name) - return predecessors + predecessors[name].update(producer for _, _, producer in reads[name]) + return predecessors, reads + + +def _calibration_roots(rows: Mapping[str, Mapping], document: Mapping) -> set[str]: + roots = { + name + for name, row in rows.items() + if isinstance(row.get("kernel"), str) and row["kernel"].startswith("calibrate.") + } + endpoint = (document.get("endpoints") or {}).get("calibration") + if isinstance(endpoint, str) and endpoint in rows: + roots.add(endpoint) + return roots + + +def _measurement_refusal( + binding: str, + root: str, + found: Sequence[str], + declared: Sequence[str], +) -> str: + return ( + f"{len(found)} nodes declaring measurement_id {binding!r} " + f"sit upstream of calibration node {root!r} " + f"(declared by {list(declared)}); exactly one must" + ) def measurement_proofs( @@ -557,54 +575,34 @@ def measurement_proofs( G8-r2c ruling R2C-1: composition satisfies ``measurement:`` only when, for every calibration node, exactly one node upstream of it declares - ``measurement_id: `` and writes the income and universe columns of - every interval table the measurement's profiles scale, on that table's - entity. Upstream follows the graph compiler's own predecessor rule over - the declared ``transport_graph`` rows (:func:`_declared_predecessors`). - Returns the proofs by measurement id and, for each measurement without - one, the reason it stays parked. + ``params.measurement_id: `` (a string) and writes the income and + universe columns of every interval table the measurement's profiles + scale, on that table's entity. "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 + proofs by measurement id and, for each measurement without one, the + reason it stays parked. """ if not _has_resource(spec, "reference_profiles"): return {}, {} wanted = _measurement_tables(spec) if not wanted: return {}, {} - nodes = ( - _raw_transport_resource(spec, GRAPH_RESOURCE).get("nodes", ()) - if _has_resource(spec, GRAPH_RESOURCE) - else () - ) - rows = { - row["id"]: row - for row in nodes - if isinstance(row, Mapping) and isinstance(row.get("id"), str) - } - document = ( - _raw_transport_resource(spec, GRAPH_RESOURCE) - if _has_resource(spec, GRAPH_RESOURCE) - else {} - ) - roots = { + if _has_resource(spec, GRAPH_RESOURCE): + document = transport_graph_document(spec) + rows = _precalibration_rows(document) + else: + document, rows = {}, {} + roots = _calibration_roots(rows, document) + predecessors, _ = _declared_calibration_predecessors(rows) + upstream = {root: _ancestor_closure(predecessors, {root}) for root in sorted(roots)} + malformed = sorted( name for name, row in rows.items() - if isinstance(row.get("kernel"), str) and row["kernel"].startswith("calibrate.") - } - endpoint = (document.get("endpoints") or {}).get("calibration") - if isinstance(endpoint, str) and endpoint in rows: - roots.add(endpoint) - predecessors = _declared_predecessors(rows) - - def ancestors(root: str) -> set[str]: - seen, pending = set(), [root] - while pending: - name = pending.pop() - if name in seen or name not in rows: - continue - seen.add(name) - pending.extend(predecessors[name]) - return seen - - upstream = {root: ancestors(root) for root in sorted(roots)} + if _measurement_declaration(row) is not None + and not _is_text(_measurement_declaration(row)) + ) proofs: dict[str, dict] = {} refusals: dict[str, str] = {} for binding, tables in sorted(wanted.items()): @@ -617,10 +615,14 @@ def ancestors(root: str) -> set[str]: key=str, ) declared = sorted( - name for name, row in rows.items() if row.get("measurement_id") == binding + name + for name, row in rows.items() + if _measurement_declaration(row) == binding ) producers, reason = {}, None - if not tables: + if malformed: + reason = f"nodes {malformed} declare a measurement_id that is not a string" + elif not tables: reason = "its profiles scale no declared interval table" elif not upstream: reason = "the graph declares no calibration node" @@ -629,11 +631,7 @@ def ancestors(root: str) -> set[str]: break found = [name for name in declared if name in closure] if len(found) != 1: - reason = ( - f"{len(found)} nodes declaring measurement_id {binding!r} " - f"sit upstream of calibration node {root!r} " - f"(declared by {declared}); exactly one must" - ) + reason = _measurement_refusal(binding, root, found, declared) break (producer,) = found missing = [ @@ -659,6 +657,42 @@ def ancestors(root: str) -> set[str]: return proofs, refusals +def _check_measurement_proofs( + spec: Mapping[str, object], graph: Graph, compiled: CompiledGraph +) -> None: + """Refuse a declared proof the compiled graph's ancestry contradicts. + + 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. + """ + proofs, _ = measurement_proofs(spec) + if not proofs: + return + roots = {node.id for node in graph.nodes if node.kernel.startswith("calibrate.")} + endpoint = transport_endpoints(spec).get("calibration") + if isinstance(endpoint, str): + roots.add(endpoint) + declared = { + node.id: node.params.get(MEASUREMENT_PARAM) + for node in graph.nodes + if isinstance(node.params, Mapping) and MEASUREMENT_PARAM in node.params + } + for binding, proof in sorted(proofs.items()): + named = sorted(name for name, value in declared.items() if value == binding) + 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( + 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}." + ) + + def reference_profile_view( spec: Mapping[str, object], selection: Mapping | None = None ): @@ -876,8 +910,8 @@ def _reference_activation_gaps( remaining = [] measurements = [ item.removeprefix("measurement:") - for item in remaining if isinstance(item, str) - and item.startswith("measurement:") + for item in remaining + if isinstance(item, str) and item.startswith("measurement:") ] if measurements: gap += "; awaits " + ", ".join( @@ -3243,21 +3277,7 @@ def _validate_precalibration_ancestry( "name a node." ) roots.add(endpoints["calibration"]) - predecessors = {name: set() for name in rows} - reads = _precalibration_declared_reads(rows) - for name, row in rows.items(): - base = row.get("base") - if isinstance(base, str): - predecessors[name].add(base) - predecessors[name].update( - item["id"] for item in rows.values() if item.get("population") == base - ) - elif isinstance(row.get("population"), str): - predecessors[name].add(row["population"]) - predecessors[name].update( - edge["producer"] for edge in row.get("artifact_inputs", ()) - ) - predecessors[name].update(producer for _, _, producer in reads[name]) + predecessors, reads = _declared_calibration_predecessors(rows) declarations = rows @@ -4263,6 +4283,7 @@ def compose_transport_graph( ) compiled = compile_graph(skeleton) validate_transport_calibration_ancestry(spec, graph=skeleton, config=config) + _check_measurement_proofs(spec, skeleton, compiled) # Money ownership is checked after each graph pass of the ancestry guard. _check_single_money_normalization(spec, skeleton, compiled) _check_money_bridge_owners(skeleton, compiled) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py index bb8235c3ea..4c9402a09f 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py @@ -29,6 +29,7 @@ proof_for, summed_profile_fixture, summed_row, + with_target_labels, ) TEN = [f"n{index:02d}" for index in range(10)] @@ -230,7 +231,9 @@ def _partition(draw_sizes, labels): values=st.lists( st.integers(min_value=0, max_value=10**9), min_size=30, max_size=30 ), - counts=st.lists(st.integers(min_value=0, max_value=10**6), min_size=30, max_size=30), + counts=st.lists( + st.integers(min_value=0, max_value=10**6), min_size=30, max_size=30 + ), numerator=st.integers(min_value=500_000, max_value=2_000_000), ) @settings( @@ -274,11 +277,16 @@ def _singleton_inputs(labels, anchors=None): row["filter"] = f"single_{label}" selector = row["ledger_selector"] selector["dimension_values"] = {"band": label} - selector["aggregate_fact_key"] = f"ledger.aggregate_fact.v2:toy-{kind}-{label}" + selector["aggregate_fact_key"] = ( + f"ledger.aggregate_fact.v2:toy-{kind}-{label}" + ) if kind == "count": row["value_operation"] = "identity" rows.append(row) base["target_references"] = rows + for row in rows: + row["metadata"]["contract_target_id"] = row["name"] + with_target_labels(base) transform = profiles["dimensions"]["index"]["profiles"]["central"][ "transformations" ][0] diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py index bf3f7e4ce2..6e21d25c06 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -2,7 +2,7 @@ ``compose.measurement_proofs`` satisfies ``measurement:`` only when, for every calibration node, exactly one node upstream of it declares -``measurement_id: `` and writes the scaled interval table's income and +``params.measurement_id: `` and writes the scaled interval table's income and universe columns on that table's entity. Each mutant below breaks one clause and must leave the rows parked, with the activation gap naming the measurement. @@ -31,6 +31,11 @@ def _graph(entity="person", *, producer_entity=None, measured=True): + """CREATE, the producer, preparation, the problem and calibration. + + Preparation reads the income and universe columns on the producer's + entity, so the producer always sits upstream of calibration. + """ producer_entity = entity if producer_entity is None else producer_entity measure = { "id": "toy.measure", @@ -43,7 +48,7 @@ def _graph(entity="person", *, producer_entity=None, measured=True): ], } if measured: - measure["measurement_id"] = SUMMED_MEASUREMENT + measure["params"] = {"measurement_id": SUMMED_MEASUREMENT} return { "schema_version": 1, "country": "toy", @@ -59,7 +64,10 @@ def _graph(entity="person", *, producer_entity=None, measured=True): "kernel": "calibration.prepare@1", "population": "toy.create", "inputs": [ - {"entity": entity, "columns": [INCOME_COLUMN, UNIVERSE_COLUMN]} + { + "entity": producer_entity, + "columns": [INCOME_COLUMN, UNIVERSE_COLUMN], + } ], "outputs": [ {"entity": entity, "column": "toy_income_band_b0", "dtype": "bool"} @@ -131,7 +139,7 @@ def test_a_third_entity_table_passes_the_proof(): def _no_producer(graph): - del _nodes(graph)["toy.measure"]["measurement_id"] + del _nodes(graph)["toy.measure"]["params"] def _after_calibration(graph): @@ -155,6 +163,13 @@ def _two_producers(graph): def _wrong_column(graph): + # The producer writes another income column; CREATE supplies the one + # the table names. + nodes = _nodes(graph) + nodes["toy.measure"]["outputs"][0]["column"] = "toy_other_cents" + + +def _income_without_universe(graph): nodes = _nodes(graph) nodes["toy.measure"]["outputs"] = nodes["toy.measure"]["outputs"][:1] graph["nodes"].insert( @@ -177,6 +192,12 @@ def _wrong_column(graph): ("after_calibration", _after_calibration, "person", "0 nodes declaring"), ("two_producers", _two_producers, "person", "2 nodes declaring"), ("wrong_column", _wrong_column, "person", "does not write"), + ( + "income_without_universe", + _income_without_universe, + "person", + "does not write", + ), ("wrong_entity", None, "person", "does not write"), ("wrong_entity_tax_unit", None, "tax_unit", "does not write"), ], @@ -200,9 +221,7 @@ def test_each_proof_mutant_leaves_rows_parked_and_names_the_measurement( "requires_prerequisites" } gaps = list( - _reference_activation_gaps( - references, "target_references", refusals=refusals - ) + _reference_activation_gaps(references, "target_references", refusals=refusals) ) assert len(gaps) == len(rows) assert all(f"measurement:{SUMMED_MEASUREMENT} (" in gap for gap in gaps) @@ -242,7 +261,118 @@ def test_proof_result_is_bound_into_the_selection_identity(): proved_receipt["reference_selection_sha256"] != parked_receipt["reference_selection_sha256"] ) - assert json.loads( - proved["target_references"][0]["metadata"]["reference_prerequisites_remaining"] - ) == [] + assert ( + json.loads( + proved["target_references"][0]["metadata"][ + "reference_prerequisites_remaining" + ] + ) + == [] + ) assert parked != proved + + +def _compiled_toy(*, second_producer=False): + """A compiled graph: CREATE, one or two producers, preparation, calibration.""" + from microcosm.graph import ( + Graph, + Node, + Owned, + Slice, + SourceRef, + StructuralDelta, + compile_graph, + ) + + def producer(name, column): + return Node( + id=name, + kernel="measure.combine@1", + population="toy.create", + params={"measurement_id": SUMMED_MEASUREMENT}, + outputs=(Owned("person", column, "int64"),), + ) + + nodes = [ + Node( + id="toy.create", + kernel="transport.create@1", + structural=StructuralDelta.CREATE, + sources=("toy_donor",), + outputs=(Owned("person", "person_id", "int64"),), + ), + producer("toy.measure", INCOME_COLUMN), + ] + read = [INCOME_COLUMN] + if second_producer: + nodes.append(producer("toy.measure.again", "toy_other_cents")) + read.append("toy_other_cents") + nodes += [ + Node( + id="toy.prepare", + kernel="calibration.prepare@1", + population="toy.create", + inputs=(Slice("person", tuple(read)),), + outputs=(Owned("person", "toy_income_band_b0", "bool"),), + ), + Node( + id="toy.calibrate", + kernel="calibrate.ordered_adam@1", + population="toy.create", + inputs=(Slice("person", ("toy_income_band_b0",)),), + ), + ] + graph = Graph( + country="toy", + sources=(SourceRef("toy_donor", "toy-codec-v1"),), + nodes=tuple(nodes), + ) + return graph, compile_graph(graph) + + +def _forced_proof(monkeypatch, producer): + from microcosm.build.transport import compose as compose_module + + proof = { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {"toy.calibrate": producer}, + "columns": [], + } + } + monkeypatch.setattr(compose_module, "measurement_proofs", lambda spec: (proof, {})) + monkeypatch.setattr( + compose_module, + "transport_endpoints", + lambda spec: {"calibration": "toy.calibrate"}, + ) + return compose_module._check_measurement_proofs + + +def test_the_compiled_graph_confirms_the_declared_producer(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy() + check({}, graph, compiled) + + +@pytest.mark.parametrize( + ("producer", "second", "found"), + [("toy.elsewhere", False, "1 nodes"), ("toy.measure", True, "2 nodes")], +) +def test_composition_refuses_a_proof_the_compiled_graph_contradicts( + monkeypatch, producer, second, found +): + check = _forced_proof(monkeypatch, producer) + graph, compiled = _compiled_toy(second_producer=second) + with pytest.raises(ValueError, match=f"contradicts the declared proof; {found}"): + check({}, graph, compiled) + + +def test_a_non_string_measurement_id_param_proves_nothing(): + document = _graph() + _nodes(document)["toy.measure"]["params"] = { + "measurement_id": {"resource": "factor_context", "path": [], "encoding": "json"} + } + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert "not a string" in refusals[SUMMED_MEASUREMENT] diff --git a/test_support/microcosm_build/reference_intervals.py b/test_support/microcosm_build/reference_intervals.py index bffec51fd3..3c12f891ca 100644 --- a/test_support/microcosm_build/reference_intervals.py +++ b/test_support/microcosm_build/reference_intervals.py @@ -236,6 +236,8 @@ def band_fact(kind, label, value, *, period=2001, record_set="toy.bands"): "vintage": "v1", }, "dimensions": {"band": label}, + "dimension_labels": {"band": "Band"}, + "dimension_value_labels": {"band": {label: f"Band {label}"}}, "universe_constraints": {"domain": "toy"}, "layout": { "record_set_id": record_set, @@ -348,11 +350,24 @@ def summed_profile_fixture( } }, } - base = reference_document(references) + base = with_target_labels(reference_document(references)) intervals = {"schema_version": 1, "tables": [table]} return base, profiles, context, intervals +def with_target_labels(document): + """Give every row the Microcosm label a summed or scaled target needs. + + The ledger refuses a multi-fact or transformed target without an explicit + target label (``ledger_targets.py`` ``_target_label``). + """ + document["hierarchy"]["target_labels"] = { + row["metadata"]["contract_target_id"]: f"Toy target {row['name']}" + for row in document["target_references"] + } + return document + + def proof_for(producer="toy.measure", calibration="toy.calibrate"): """The proof composition returns for one producer upstream of calibration.""" return { @@ -362,4 +377,3 @@ def proof_for(producer="toy.measure", calibration="toy.calibrate"): "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], } } - From 41bb6ad9943ce8544b35a235ac5b34bca5dc838f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 19:34:43 +0000 Subject: [PATCH 3/8] Supply measurement proofs where tests activated factor rows by hand Hunk A's resolution validator refuses a factor companion that is active without its measurement's proof. Two fixtures forged activation after resolution: factor_transport_fixture(activate=True) and the CLI replay test's measured view. Both now hand the resolver a synthetic proof, so the real resolver activates the rows and binds the proof into the selection identity. The proof reads the raw transport_graph resource, as the rest of the reference-profile view does, so a spec whose graph composition would refuse for another reason still resolves its references. Invariant: a factor row is active exactly when its receipt carries the measurement's proof. Checked by test_transport_target_kernels.py, test_nz_money_normalization.py, test_transport_run_entry_factors.py and the CLI replay test (200 passed). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwQq6HQomkwKHCkdAgY9cy --- .../src/microcosm/build/transport/compose.py | 2 +- .../shared/test_transport_composed_graph.py | 26 +++++++++-------- .../reference_factor_transport.py | 28 ++++++++++++------- 3 files changed, 33 insertions(+), 23 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 9673a3ab7f..06cd0d4674 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -590,7 +590,7 @@ def measurement_proofs( if not wanted: return {}, {} if _has_resource(spec, GRAPH_RESOURCE): - document = transport_graph_document(spec) + document = _raw_transport_resource(spec, GRAPH_RESOURCE) rows = _precalibration_rows(document) else: document, rows = {}, {} diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py index f597aa1335..addd69d4bf 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_composed_graph.py @@ -56,7 +56,6 @@ from hypothesis import HealthCheck, example, given, settings from hypothesis import strategies as st -from microcosm.build.reference_profiles import reference_document_sha256 from microcosm.build.transport import cli as cli_module from microcosm.build.transport import compose as compose_module from microcosm.build.transport import target_kernels as target_kernels_module @@ -114,6 +113,7 @@ from test_support.microcosm_build.chronicle_feed import write_pinned_artifact from test_support.microcosm_build.money_normalization import graph_declaration from test_support.microcosm_build.reference_factor_transport import ( + REVIEWED_MEASUREMENT_PROOF, factor_transport_fixture, ) from test_support.microcosm_build.reference_intervals import ( @@ -3207,19 +3207,21 @@ def test_cli_require_replays_a_warm_store_after_selected_factors_expire( ) profile_view = compose_module.reference_profile_view + proofs = compose_module.measurement_proofs + def measured_view(spec, selection=None): # As in factor_transport_fixture(activate=True), supply a synthetic - # reviewed measurement closure. The real resolver still selects and - # binds the factors; composition must attach its rehashed receipt. - # Copies, so the resolver's own return values are never edited. - references, intervals, receipt = profile_view(spec, selection) - references, receipt = copy.deepcopy(references), copy.deepcopy(receipt) - for row in references["target_references"]: - row["metadata"]["activation_status"] = "active" - receipt["effective_references_sha256"] = reference_document_sha256(references) - receipt.pop("receipt_sha256") - receipt["receipt_sha256"] = reference_document_sha256(receipt) - return references, intervals, receipt + # proof of the reviewed measurement, only while the view resolves. + # The real resolver still selects and binds the factors and activates + # the proved rows; composition must attach its receipt. + compose_module.measurement_proofs = lambda spec: ( + REVIEWED_MEASUREMENT_PROOF, + {}, + ) + try: + return profile_view(spec, selection) + finally: + compose_module.measurement_proofs = proofs monkeypatch.setattr(compose_module, "reference_profile_view", measured_view) declaration = spec["resources"]["transport_graph"] diff --git a/test_support/microcosm_build/reference_factor_transport.py b/test_support/microcosm_build/reference_factor_transport.py index 1fe2f52bb4..ada34c86d4 100644 --- a/test_support/microcosm_build/reference_factor_transport.py +++ b/test_support/microcosm_build/reference_factor_transport.py @@ -9,13 +9,21 @@ ) from microcosm.build.reference_factors import resolve_reference_factor from microcosm.build.reference_profiles import ( - reference_document_sha256, resolve_reference_profiles, ) from test_support.microcosm_build.chronicle_feed import write_pinned_artifact from test_support.microcosm_build.reference_factors import factor_fixture from test_support.microcosm_build.transport_graph import reference_document +#: The proof composition returns for one ``reviewed_measurement`` producer. +REVIEWED_MEASUREMENT_PROOF = { + "reviewed_measurement": { + "measurement_id": "reviewed_measurement", + "producers": {"toy.calibrate": "toy.measure"}, + "columns": [], + } +} + def factor_transport_fixture( tmp_path, *, forged=False, activate=False, manifest_commit=False @@ -146,17 +154,17 @@ def factor_transport_fixture( base["hierarchy"]["target_labels"] = { row["name"]: f"Aged toy {row['name']}" for row in references } + # A test-only measurement proof (what composition's measurement_proofs + # returns for one producer upstream of calibration) lets the compiler test + # its own replay and arithmetic independently of this lane's parked + # country declarations. document, _, resolution = resolve_reference_profiles( - base, profiles, {}, factor_context=context + base, + profiles, + {}, + factor_context=context, + measurement_proofs=REVIEWED_MEASUREMENT_PROOF if activate else None, ) - if activate: - # Test-only future measurement closure lets the compiler test its own replay - # and arithmetic independently of this lane's parked country declarations. - for row in document["target_references"]: - row["metadata"]["activation_status"] = "active" - resolution["effective_references_sha256"] = reference_document_sha256(document) - resolution.pop("receipt_sha256") - resolution["receipt_sha256"] = reference_document_sha256(resolution) spec = { "country": "xx", "resources": { From 4d2f3b2be410955846e04edeffdff82b9fbdefd4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 19:45:47 +0000 Subject: [PATCH 4/8] Drop the unreachable member-set comparison in summed rows A mutant that removed the comparison between a summed money row's members and its count companion's members left every test green. The comparison cannot fire: the companion key includes the canonical member lists, so a count row pairs with a money row only over the same member set, and a mismatch already refuses as a missing companion. The refusal now says so. Invariant: summed money and count rows pair only over one member set. Checked by test_summed_rows_refuse_each_inconsistency[mismatched_members] (18 passed in test_reference_summed_bands.py). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwQq6HQomkwKHCkdAgY9cy --- .../src/microcosm/build/reference_profiles.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/reference_profiles.py b/packages/microcosm-build/src/microcosm/build/reference_profiles.py index 94c2501d93..e42ef39769 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_profiles.py +++ b/packages/microcosm-build/src/microcosm/build/reference_profiles.py @@ -300,12 +300,14 @@ def _validate_summed_rows(money: list, counts: list, by_name: Mapping) -> None: ) if members is None and count_members is None: continue - summed.update({name, count_name}) - if members != count_members or count_name is None: + # The companion key holds the canonical member lists, so a count row + # pairs with this money row only over the same member set. + if count_name is None: raise ValueError( - f"Summed references {name!r} and {count_name!r} select " - "different member sets." + f"Summed money reference {name!r} and its count companion " + "select different member sets: no count row sums its members." ) + summed.update({name, count_name}) count_row = by_name[count_name] if count_row.get("value_operation") != "sum" or count_row.get( "value_operands", [] From 05bb18ca99a5f44ddd7612d0a457c94c8dacc6bf Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 23:11:41 +0000 Subject: [PATCH 5/8] Refuse a second writer of a measured column (A1) Ruling R2C-15: a measurement holds only when its declaring producer is the only ancestor of each calibration node that writes each of the interval table's income and universe columns on the table's entity. The review's item 2c built a graph where an undeclared later-version rewrite of the income column passed both checks while the receipt named the declarer. Before this change measurement_proofs proved it, the compiled check accepted it, and compose_transport_graph composed the composed toy with the rewrite grafted on. measurement_proofs now collects every node in the calibration node's declared closure whose outputs hold each required column, and parks the rows unless that list is exactly the producer. _check_measurement_proofs applies the same rule over the compiled ancestors and the composed nodes' outputs and raises. Both share _second_writer_refusal, which names the measurement, the calibration node, the column and every writer. The rule has no ordering logic: a writer before or after the producer refuses, with or without rewrite. Checked with new pins in test_transport_measurement_proof.py: item 2c, a writer before the producer, a universe-only rewrite, a rewrite-free second writer, the composed toy, a compiled-only case, two controls that still prove, and a Hypothesis property over declared chains. Dropping the declared check turns 7 tests red; dropping the compiled check turns test_the_compiled_check_alone_refuses_a_second_writer red. measurement_proofs and validate_transport_activation give identical outcomes on every packaged spec (am, be, nz, uk, us) before and after. The same commit adds test_a_calibration_endpoint_without_a_calibrate_kernel_is_a_root for the review's OA2. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwQq6JfFhwRFMYkokanqow --- .../src/microcosm/build/transport/compose.py | 67 +++- .../test_transport_measurement_proof.py | 366 +++++++++++++++++- 2 files changed, 425 insertions(+), 8 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 06cd0d4674..030283063a 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -568,6 +568,28 @@ def _measurement_refusal( ) +def _second_writer_refusal( + binding: str, + root: str, + producer: str, + writers: Mapping[tuple[str, str], Sequence[str]], +) -> str | None: + """Why a measured column has a writer other than its producer, or ``None``. + + R2C-15: the producer is the only ancestor of each calibration node that + writes each measured column, wherever another writer sits and whether + or not it rewrites. + """ + for (entity, column), names in sorted(writers.items(), key=str): + if list(names) != [producer]: + return ( + f"measurement {binding!r}: calibration node {root!r} has " + f"writers {list(names)} of {entity}.{column} upstream; " + f"only its producer {producer!r} may write it" + ) + return None + + def measurement_proofs( spec: Mapping[str, object], ) -> tuple[dict[str, dict], dict[str, str]]: @@ -577,7 +599,9 @@ def measurement_proofs( for every calibration node, exactly one node upstream of it declares ``params.measurement_id: `` (a string) and writes the income and universe columns of every interval table the measurement's profiles - scale, on that table's entity. "Upstream" is ancestry in the declared + scale, on that table's entity. R2C-15 adds that the producer is the only + node upstream of that calibration node which writes each of those + 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 @@ -645,6 +669,21 @@ def measurement_proofs( "interval table's entity" ) break + reason = _second_writer_refusal( + binding, + root, + producer, + { + (entity, column): sorted( + name + for name in closure + if (entity, column) in _owned_columns(rows[name]) + ) + for entity, column in required + }, + ) + if reason is not None: + break producers[root] = producer if reason is None: proofs[binding] = { @@ -665,7 +704,8 @@ def _check_measurement_proofs( 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. + producer for that node. That producer must also be the only compiled + ancestor whose outputs hold each of the proof's columns (R2C-15). """ proofs, _ = measurement_proofs(spec) if not proofs: @@ -679,6 +719,7 @@ def _check_measurement_proofs( for node in graph.nodes if isinstance(node.params, Mapping) and MEASUREMENT_PARAM in node.params } + by_id = {node.id: node for node in graph.nodes} for binding, proof in sorted(proofs.items()): named = sorted(name for name, value in declared.items() if value == binding) for root in sorted(roots): @@ -691,6 +732,28 @@ def _check_measurement_proofs( + _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"]] + refusal = _second_writer_refusal( + binding, + root, + proof["producers"][root], + { + (entity, column): sorted( + name + for name in closure + if any( + (output.entity, output.column) == (entity, column) + for output in by_id[name].outputs + ) + ) + for entity, column in columns + }, + ) + if refusal is not None: + raise ValueError( + f"Measurement {binding!r}: the compiled graph contradicts " + f"the declared proof; {refusal}." + ) def reference_profile_view( diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py index 6e21d25c06..12685de9c3 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -14,9 +14,12 @@ import json import pytest +from hypothesis import HealthCheck, given, settings +from hypothesis import strategies as st from microcosm.build.transport.compose import ( _reference_activation_gaps, + compose_transport_graph, measurement_proofs, reference_profile_view, ) @@ -185,9 +188,94 @@ def _income_without_universe(graph): ) +def _filtered(graph): + """Open a ``filter`` version on CREATE and move calibration into it.""" + nodes = _nodes(graph) + position = graph["nodes"].index(nodes["toy.prepare"]) + graph["nodes"].insert( + position, + { + "id": "toy.filter", + "kernel": "transport.boundary@1", + "structural": "filter", + "base": "toy.create", + }, + ) + for name in ("toy.prepare", "toy.problem", "toy.calibrate"): + nodes[name]["population"] = "toy.filter" + return position + 1 + + +def _rewriter(name, column, dtype, *, population="toy.filter", rewrite=True): + output = {"entity": "person", "column": column, "dtype": dtype} + if rewrite: + output["rewrite"] = True + return { + "id": name, + "kernel": "toy.rewrite@1", + "population": population, + "inputs": [{"entity": "person", "columns": [column]}], + "outputs": [output], + } + + +def _later_version_rewrite(graph): + """The review's item 2c: an undeclared later-version rewrite of the income.""" + position = _filtered(graph) + graph["nodes"].insert(position, _rewriter("toy.overwrite", INCOME_COLUMN, "int64")) + + +def _writer_before_the_producer(graph): + """An undeclared CREATE-version writer; the producer rewrites it later.""" + _filtered(graph) + nodes = _nodes(graph) + nodes["toy.measure"]["population"] = "toy.filter" + nodes["toy.measure"]["outputs"][0]["rewrite"] = True + graph["nodes"].remove(nodes["toy.measure"]) + graph["nodes"].insert( + graph["nodes"].index(nodes["toy.prepare"]), nodes["toy.measure"] + ) + graph["nodes"].insert( + 1, + _rewriter( + "toy.early", + INCOME_COLUMN, + "int64", + population="toy.create", + rewrite=False, + ), + ) + graph["nodes"][1]["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + + +def _universe_rewrite(graph): + position = _filtered(graph) + graph["nodes"].insert(position, _rewriter("toy.overwrite", UNIVERSE_COLUMN, "str")) + + @pytest.mark.parametrize( ("mutant", "graph", "entity", "reason"), [ + ( + "later_version_rewrite", + _later_version_rewrite, + "person", + "writers ['toy.measure', 'toy.overwrite'] of person." + f"{INCOME_COLUMN} upstream", + ), + ( + "writer_before_the_producer", + _writer_before_the_producer, + "person", + f"writers ['toy.early', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), + ( + "universe_rewrite", + _universe_rewrite, + "person", + "writers ['toy.measure', 'toy.overwrite'] of person." + f"{UNIVERSE_COLUMN} upstream", + ), ("no_producer", _no_producer, "person", "0 nodes declaring"), ("after_calibration", _after_calibration, "person", "0 nodes declaring"), ("two_producers", _two_producers, "person", "2 nodes declaring"), @@ -272,8 +360,12 @@ def test_proof_result_is_bound_into_the_selection_identity(): assert parked != proved -def _compiled_toy(*, second_producer=False): - """A compiled graph: CREATE, one or two producers, preparation, calibration.""" +def _compiled_toy(*, second_producer=False, second_writer=False): + """A compiled graph: CREATE, one or two producers, preparation, calibration. + + With ``second_writer``, a ``filter`` version holds an undeclared rewrite + of the income column, and preparation and calibration live in it. + """ from microcosm.graph import ( Graph, Node, @@ -307,18 +399,36 @@ def producer(name, column): if second_producer: nodes.append(producer("toy.measure.again", "toy_other_cents")) read.append("toy_other_cents") + population = "toy.create" + if second_writer: + population = "toy.filter" + nodes += [ + Node( + id="toy.filter", + kernel="transport.boundary@1", + structural=StructuralDelta.FILTER, + base="toy.create", + ), + Node( + id="toy.overwrite", + kernel="toy.rewrite@1", + population="toy.filter", + inputs=(Slice("person", (INCOME_COLUMN,)),), + outputs=(Owned("person", INCOME_COLUMN, "int64", rewrite=True),), + ), + ] nodes += [ Node( id="toy.prepare", kernel="calibration.prepare@1", - population="toy.create", + population=population, inputs=(Slice("person", tuple(read)),), outputs=(Owned("person", "toy_income_band_b0", "bool"),), ), Node( id="toy.calibrate", kernel="calibrate.ordered_adam@1", - population="toy.create", + population=population, inputs=(Slice("person", ("toy_income_band_b0",)),), ), ] @@ -330,14 +440,14 @@ def producer(name, column): return graph, compile_graph(graph) -def _forced_proof(monkeypatch, producer): +def _forced_proof(monkeypatch, producer, columns=(f"person.{INCOME_COLUMN}",)): from microcosm.build.transport import compose as compose_module proof = { SUMMED_MEASUREMENT: { "measurement_id": SUMMED_MEASUREMENT, "producers": {"toy.calibrate": producer}, - "columns": [], + "columns": list(columns), } } monkeypatch.setattr(compose_module, "measurement_proofs", lambda spec: (proof, {})) @@ -368,6 +478,21 @@ def test_composition_refuses_a_proof_the_compiled_graph_contradicts( check({}, graph, compiled) +def test_the_compiled_check_alone_refuses_a_second_writer(monkeypatch): + """The declared check never sees this graph: a forced proof drives it.""" + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(second_writer=True) + with pytest.raises(ValueError) as refused: + check({}, graph, compiled) + assert str(refused.value) == ( + f"Measurement {SUMMED_MEASUREMENT!r}: the compiled graph contradicts the " + f"declared proof; measurement {SUMMED_MEASUREMENT!r}: calibration node " + "'toy.calibrate' has writers ['toy.measure', 'toy.overwrite'] of " + f"person.{INCOME_COLUMN} upstream; only its producer 'toy.measure' may " + "write it." + ) + + def test_a_non_string_measurement_id_param_proves_nothing(): document = _graph() _nodes(document)["toy.measure"]["params"] = { @@ -376,3 +501,232 @@ def test_a_non_string_measurement_id_param_proves_nothing(): proofs, refusals = measurement_proofs(_spec(document)) assert proofs == {} assert "not a string" in refusals[SUMMED_MEASUREMENT] + + +# --------------------------------------------------------------------------- +# R2C-15: a measurement has one writer +# --------------------------------------------------------------------------- + + +def test_a_later_version_rewrite_names_the_measurement_node_column_and_writers(): + """The review's item 2c refuses in the declared check, naming every fact.""" + document = _graph() + _later_version_rewrite(document) + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert refusals == { + SUMMED_MEASUREMENT: ( + f"measurement {SUMMED_MEASUREMENT!r}: calibration node " + "'toy.calibrate' has writers ['toy.measure', 'toy.overwrite'] of " + f"person.{INCOME_COLUMN} upstream; only its producer 'toy.measure' " + "may write it" + ) + } + + +def test_a_second_writer_refuses_without_rewrite(): + """A later writer refuses whether or not its output says ``rewrite``.""" + document = _graph() + position = _filtered(document) + document["nodes"].insert( + position, _rewriter("toy.overwrite", INCOME_COLUMN, "int64", rewrite=False) + ) + proofs, refusals = measurement_proofs(_spec(document)) + assert proofs == {} + assert "['toy.measure', 'toy.overwrite']" in refusals[SUMMED_MEASUREMENT] + + +def test_a_same_named_column_on_another_entity_is_not_a_second_writer(): + document = _graph() + position = _filtered(document) + other = _rewriter("toy.household", INCOME_COLUMN, "int64", rewrite=False) + other["inputs"] = [{"entity": "person", "columns": ["toy_wages"]}] + other["outputs"][0]["entity"] = "household" + document["nodes"].insert(position, other) + _nodes(document)["toy.prepare"]["inputs"].append( + {"entity": "household", "columns": [INCOME_COLUMN]} + ) + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +def test_a_writer_outside_the_calibration_closure_is_not_a_second_writer(): + """A later version that reads the fit rewrites the column after calibration.""" + document = _graph() + document["nodes"] += [ + { + "id": "toy.after", + "kernel": "transport.boundary@1", + "structural": "filter", + "base": "toy.create", + "artifact_inputs": [{"name": "fit", "producer": "toy.calibrate"}], + }, + _rewriter("toy.overwrite", INCOME_COLUMN, "int64", population="toy.after"), + ] + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +def _chain(draws): + """CREATE, then nodes chained by artifacts, all upstream of calibration. + + Each draw is (declares the id, writes the income, writes the universe). + """ + nodes = [ + {"id": "toy.create", "kernel": "transport.create@1", "structural": "create"} + ] + previous = None + for index, (declares, income, universe) in enumerate(draws): + row = { + "id": f"toy.n{index}", + "kernel": "toy.chain@1", + "population": "toy.create", + "outputs": [ + {"entity": "person", "column": column, "dtype": dtype} + for column, dtype, wanted in ( + (INCOME_COLUMN, "int64", income), + (UNIVERSE_COLUMN, "str", universe), + (f"toy_step_{index}", "bool", True), + ) + if wanted + ], + } + if declares: + row["params"] = {"measurement_id": SUMMED_MEASUREMENT} + if previous is not None: + row["inputs"] = [{"entity": "person", "columns": [previous]}] + previous = f"toy_step_{index}" + nodes.append(row) + nodes.append( + { + "id": "toy.calibrate", + "kernel": "calibrate.ordered_adam@1", + "population": "toy.create", + "inputs": [{"entity": "person", "columns": [previous]}], + } + ) + return { + "schema_version": 1, + "country": "toy", + "nodes": nodes, + "endpoints": {"calibration": "toy.calibrate"}, + } + + +@given( + draws=st.lists( + st.tuples(st.booleans(), st.booleans(), st.booleans()), + min_size=1, + max_size=4, + ) +) +@settings( + max_examples=60, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +def test_a_chain_proves_exactly_when_one_declarer_is_the_only_writer(draws): + proofs, refusals = measurement_proofs(_spec(_chain(draws))) + declarers = [index for index, draw in enumerate(draws) if draw[0]] + income = [index for index, draw in enumerate(draws) if draw[1]] + universe = [index for index, draw in enumerate(draws) if draw[2]] + proved = len(declarers) == 1 and income == universe == declarers + assert (SUMMED_MEASUREMENT in proofs) is proved + assert (SUMMED_MEASUREMENT in refusals) is not proved + if proved: + assert proofs[SUMMED_MEASUREMENT]["producers"] == { + "toy.calibrate": f"toy.n{declarers[0]}" + } + + +def test_a_calibration_endpoint_without_a_calibrate_kernel_is_a_root(): + """Only the endpoint clause of the roots finds this calibration node.""" + document = _graph() + _nodes(document)["toy.calibrate"]["kernel"] = "toy.fit@1" + proofs, refusals = measurement_proofs(_spec(document)) + assert refusals == {} + assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} + + +@pytest.fixture(scope="module") +def composed_toy(tmp_path_factory): + from test_support.microcosm_build.transport_composed import ( + make_composed_fixture, + ) + + return make_composed_fixture(tmp_path_factory.mktemp("composed")) + + +def _composed_spec(composed_toy, *, overwrite): + """The composed toy with summed rows and a producer upstream of calibration. + + ``transport.currency@1`` stands in for the producer's kernel, because + composition checks the skeleton vocabulary and runs no kernel. + """ + spec = copy.deepcopy(composed_toy.spec) + base, profiles, context, intervals = summed_profile_fixture(GROUPS) + references = spec["resources"]["target_references"] + references["target_references"] += base["target_references"] + for key, value in base["hierarchy"].items(): + references["hierarchy"].setdefault(key, {}).update(value) + spec["resources"].update( + reference_profiles=profiles, + factor_context=context, + profile_intervals=intervals, + ) + graph = spec["resources"]["transport_graph"] + nodes = _nodes(graph) + calibration = graph["endpoints"]["calibration"] + (create,) = ( + row["id"] for row in graph["nodes"] if row.get("structural") == "create" + ) + opened = nodes[calibration]["base"] + graph["nodes"].insert( + 1, + { + "id": "toy.measure", + "kernel": "transport.currency@1", + "population": create, + "inputs": [{"entity": "person", "columns": ["age"]}], + "outputs": [ + {"entity": "person", "column": INCOME_COLUMN, "dtype": "Int64"}, + {"entity": "person", "column": UNIVERSE_COLUMN, "dtype": "string"}, + ], + "params": {"measurement_id": SUMMED_MEASUREMENT}, + }, + ) + if overwrite: + graph["nodes"].insert( + graph["nodes"].index(nodes[opened]) + 1, + { + **_rewriter("toy.overwrite", INCOME_COLUMN, "Int64", population=opened), + "kernel": "transport.currency@1", + }, + ) + nodes[calibration].setdefault("inputs", []).append( + {"entity": "person", "columns": [INCOME_COLUMN, UNIVERSE_COLUMN]} + ) + return spec, calibration + + +def test_composition_refuses_a_later_version_rewrite_of_a_measured_column( + composed_toy, +): + spec, calibration = _composed_spec(composed_toy, overwrite=False) + assert measurement_proofs(spec)[0][SUMMED_MEASUREMENT]["producers"] == { + calibration: "toy.measure" + } + compose_transport_graph(spec, composed_toy.config) + spec, calibration = _composed_spec(composed_toy, overwrite=True) + with pytest.raises(ValueError) as refused: + compose_transport_graph(spec, composed_toy.config) + message = str(refused.value) + assert message.startswith("Transport spec is not activated:") + assert f"awaits measurement:{SUMMED_MEASUREMENT}" in message + assert ( + f"calibration node {calibration!r} has writers ['toy.measure', " + f"'toy.overwrite'] of person.{INCOME_COLUMN} upstream" + ) in message From 30ae4599ab72a1721c64e9f7f483ec48c89e449e Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 2 Nov 2026 23:18:15 +0000 Subject: [PATCH 6/8] Pin the rounding record and scalar members (A3: OA1, OA3) The review's OA1 and OA3 mutants left every test green. Two new pins in test_reference_summed_bands.py now kill them. OA1: test_a_tampered_rounding_record_refuses changes one field of the receipt's factor_weight_rounding record (weight, exact numerator or relative rounding), recomputes the receipt hash so that only the rounding recomputation can see it, and expects validate_reference_resolution to refuse. Deleting that recomputation turns all three cases red. OA3: test_summed_rows_refuse_each_inconsistency gains a mapping member and a list member inside a summed row's member list. Dropping the scalar check in _summed_members turns both cases red. OA2's pin landed with A1 (05bb18ca): dropping the endpoint clause of _calibration_roots turns test_a_calibration_endpoint_without_a_calibrate_kernel_is_a_root red. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LwQq6JfFhwRFMYkokanqow --- .../shared/test_reference_summed_bands.py | 28 +++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py index 4c9402a09f..9302b0bb85 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py @@ -20,6 +20,7 @@ from microcosm.build.country_spec import _validate_target_references from microcosm.build.ledger_targets import compile_ledger_target_references from microcosm.build.reference_profiles import ( + reference_document_sha256, resolve_reference_profiles, validate_reference_resolution, ) @@ -183,6 +184,11 @@ def _mutate(mutation): for kind in ("count", "money"): selector = rows[f"toy_{kind}_b0"]["ledger_selector"] selector["dimension_values"]["band"] = [*TEN[:4], TEN[0]] + elif mutation in ("mapping_member", "list_member"): + member = {"band": TEN[4]} if mutation == "mapping_member" else [TEN[4]] + for kind in ("count", "money"): + selector = rows[f"toy_{kind}_b0"]["ledger_selector"] + selector["dimension_values"]["band"] = [*TEN[:4], member] else: # pragma: no cover - parametrization guard raise AssertionError(mutation) return base, profiles, context, intervals @@ -199,6 +205,8 @@ def _mutate(mutation): ("identity_count", "value_operation 'sum'"), ("fact_key_and_members", "singleton fact key and a member set"), ("repeated_member", "repeat or are not scalars"), + ("mapping_member", "repeat or are not scalars"), + ("list_member", "repeat or are not scalars"), ], ) def test_summed_rows_refuse_each_inconsistency(mutation, message): @@ -218,6 +226,26 @@ def test_receipt_states_the_binary64_rounding_of_the_weight(): assert abs(Fraction(record["binary64_relative_rounding"])) <= Fraction(1, 2**53) +@pytest.mark.parametrize( + "field", ["weight", "exact_numerator", "binary64_relative_rounding"] +) +def test_a_tampered_rounding_record_refuses(field): + """Only the rounding recomputation sees this: the receipt hash is redone.""" + effective, _, receipt = _resolve(summed_profile_fixture([TEN]), proof_for()) + tampered = copy.deepcopy(receipt) + record = tampered["factor_weight_rounding"]["toy_central"] + record[field] = { + "weight": 1.2, + "exact_numerator": record["exact_numerator"] + 1, + "binary64_relative_rounding": "0", + }[field] + del tampered["receipt_sha256"] + tampered["receipt_sha256"] = reference_document_sha256(tampered) + with pytest.raises(ValueError, match="Factor weight rounding record mismatch"): + validate_reference_resolution(effective, tampered) + assert validate_reference_resolution(effective, receipt) == receipt + + def _partition(draw_sizes, labels): groups, start = [], 0 for size in draw_sizes: From d916ea8d5850d274b4e07d4721dd18ebe96c28d7 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Mon, 2 Nov 2026 19:35:39 -0500 Subject: [PATCH 7/8] Count EXPAND cells as writers of a measured column (R2C-15) R2C-15 makes the producer the only ancestor of a calibration node that writes each measured column. Both checks collected writers from node outputs, so an EXPAND whose `params.expand_cells` names the income passed `measurement_proofs`, `_check_measurement_proofs` and `compose_transport_graph`, while the S5 reader in the same file counts those cells as writes. The declared check now asks `_declared_writer`. It also counts an EXPAND's literal cells, and it treats an EXPAND whose cells come from a resource as writing every column, as `_precalibration_declared_reads` does. The compiled check reads the cells through `_precalibration_expansions`. No packaged spec declares an EXPAND, so no packaged outcome changes. The invariant: a measurement proves only when no ancestor other than its producer may write a measured column, by output or by EXPAND cell. `test_transport_measurement_proof.py` gives 33 passed. Dropping the resource clause fails `[expand_resource_cells]` alone, and `test_packaged_spec_refuses_with_every_activation_gap_named` still passes. The patch comes from Belgium's hub, patch-id a48429b46f51315760a66b27bdf60daf2c342bba. Co-Authored-By: Claude Opus 5.5 --- .../src/microcosm/build/transport/compose.py | 25 +++- .../test_transport_measurement_proof.py | 114 +++++++++++++++++- 2 files changed, 132 insertions(+), 7 deletions(-) diff --git a/packages/microcosm-build/src/microcosm/build/transport/compose.py b/packages/microcosm-build/src/microcosm/build/transport/compose.py index 030283063a..e85c658fde 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -578,7 +578,7 @@ def _second_writer_refusal( R2C-15: the producer is the only ancestor of each calibration node that writes each measured column, wherever another writer sits and whether - or not it rewrites. + or not it rewrites. An EXPAND writes the cells it declares. """ for (entity, column), names in sorted(writers.items(), key=str): if list(names) != [producer]: @@ -590,6 +590,20 @@ def _second_writer_refusal( return None +def _declared_writer(row: Mapping, entity: object, column: object) -> bool: + """Whether a declared node may write a cell, by output or as an EXPAND. + + An EXPAND whose cells only composition can tell writes every column, as + :func:`_precalibration_declared_reads` reads it. + """ + cells = _precalibration_cells(row) + return ( + (entity, column) in _owned_columns(row) + or cells is None + or (entity, column) in cells + ) + + def measurement_proofs( spec: Mapping[str, object], ) -> tuple[dict[str, dict], dict[str, str]]: @@ -677,7 +691,7 @@ def measurement_proofs( (entity, column): sorted( name for name in closure - if (entity, column) in _owned_columns(rows[name]) + if _declared_writer(rows[name], entity, column) ) for entity, column in required }, @@ -705,7 +719,8 @@ def _check_measurement_proofs( 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 - ancestor whose outputs hold each of the proof's columns (R2C-15). + ancestor whose outputs or EXPAND cells hold each of the proof's columns + (R2C-15). """ proofs, _ = measurement_proofs(spec) if not proofs: @@ -720,6 +735,7 @@ def _check_measurement_proofs( if isinstance(node.params, Mapping) and MEASUREMENT_PARAM in node.params } 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) for root in sorted(roots): @@ -741,7 +757,8 @@ def _check_measurement_proofs( (entity, column): sorted( name for name in closure - if any( + if (entity, column) in expand_cells[name] + or any( (output.entity, output.column) == (entity, column) for output in by_id[name].outputs ) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py index 12685de9c3..0d1673aacf 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -253,9 +253,47 @@ def _universe_rewrite(graph): graph["nodes"].insert(position, _rewriter("toy.overwrite", UNIVERSE_COLUMN, "str")) +def _expand_writer(graph, cells=None): + """An EXPAND version that may write the income, with calibration in it.""" + nodes = _nodes(graph) + graph["nodes"].insert( + graph["nodes"].index(nodes["toy.prepare"]), + { + "id": "toy.expand", + "kernel": "transport.boundary@1", + "structural": "expand", + "base": "toy.create", + "params": { + "expand_cells": ( + [["person", INCOME_COLUMN, "int64"]] if cells is None else cells + ) + }, + }, + ) + for name in ("toy.prepare", "toy.problem", "toy.calibrate"): + nodes[name]["population"] = "toy.expand" + + +def _expand_with_resource_cells(graph): + """An EXPAND whose cells only composition can tell writes every column.""" + _expand_writer(graph, {"resource": "toy_cells"}) + + @pytest.mark.parametrize( ("mutant", "graph", "entity", "reason"), [ + ( + "expand_cells", + _expand_writer, + "person", + f"writers ['toy.expand', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), + ( + "expand_resource_cells", + _expand_with_resource_cells, + "person", + f"writers ['toy.expand', 'toy.measure'] of person.{INCOME_COLUMN} upstream", + ), ( "later_version_rewrite", _later_version_rewrite, @@ -360,11 +398,13 @@ def test_proof_result_is_bound_into_the_selection_identity(): assert parked != proved -def _compiled_toy(*, second_producer=False, second_writer=False): +def _compiled_toy(*, second_producer=False, second_writer=False, expand_cells=None): """A compiled graph: CREATE, one or two producers, preparation, calibration. With ``second_writer``, a ``filter`` version holds an undeclared rewrite - of the income column, and preparation and calibration live in it. + of the income column, and preparation and calibration live in it. With + ``expand_cells``, preparation and calibration live in an EXPAND version + that declares those cells. """ from microcosm.graph import ( Graph, @@ -417,6 +457,17 @@ def producer(name, column): outputs=(Owned("person", INCOME_COLUMN, "int64", rewrite=True),), ), ] + if expand_cells is not None: + population = "toy.expand" + nodes.append( + Node( + id="toy.expand", + kernel="transport.boundary@1", + structural=StructuralDelta.EXPAND, + base="toy.create", + params={"expand_cells": expand_cells}, + ) + ) nodes += [ Node( id="toy.prepare", @@ -493,6 +544,19 @@ def test_the_compiled_check_alone_refuses_a_second_writer(monkeypatch): ) +def test_the_compiled_check_alone_counts_an_expand_as_a_writer(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(expand_cells=(("person", INCOME_COLUMN, "int64"),)) + with pytest.raises(ValueError, match=r"writers \['toy.expand', 'toy.measure'\]"): + check({}, graph, compiled) + + +def test_the_compiled_check_passes_an_expand_of_another_column(monkeypatch): + check = _forced_proof(monkeypatch, "toy.measure") + graph, compiled = _compiled_toy(expand_cells=(("person", "toy_other", "int64"),)) + check({}, graph, compiled) + + def test_a_non_string_measurement_id_param_proves_nothing(): document = _graph() _nodes(document)["toy.measure"]["params"] = { @@ -569,6 +633,26 @@ def test_a_writer_outside_the_calibration_closure_is_not_a_second_writer(): assert proofs[SUMMED_MEASUREMENT]["producers"] == {"toy.calibrate": "toy.measure"} +@given( + cells=st.lists( + st.sampled_from([INCOME_COLUMN, UNIVERSE_COLUMN, "toy_other"]), unique=True + ) +) +@settings( + max_examples=30, + database=None, + deadline=None, + suppress_health_check=[HealthCheck.too_slow], +) +def test_an_expand_proves_exactly_when_its_cells_avoid_the_measured_columns(cells): + document = _graph() + _expand_writer(document, [["person", column, "int64"] for column in cells]) + proofs, refusals = measurement_proofs(_spec(document)) + proved = not {INCOME_COLUMN, UNIVERSE_COLUMN} & set(cells) + assert (SUMMED_MEASUREMENT in proofs) is proved + assert (SUMMED_MEASUREMENT in refusals) is not proved + + def _chain(draws): """CREATE, then nodes chained by artifacts, all upstream of calibration. @@ -660,7 +744,7 @@ def composed_toy(tmp_path_factory): return make_composed_fixture(tmp_path_factory.mktemp("composed")) -def _composed_spec(composed_toy, *, overwrite): +def _composed_spec(composed_toy, *, overwrite, expand=False): """The composed toy with summed rows and a producer upstream of calibration. ``transport.currency@1`` stands in for the producer's kernel, because @@ -706,6 +790,18 @@ def _composed_spec(composed_toy, *, overwrite): "kernel": "transport.currency@1", }, ) + if expand: + graph["nodes"].insert( + graph["nodes"].index(nodes[calibration]), + { + "id": "toy.expand", + "kernel": "transport.boundary@1", + "structural": "expand", + "base": opened, + "params": {"expand_cells": [["person", INCOME_COLUMN, "Int64"]]}, + }, + ) + nodes[calibration]["base"] = "toy.expand" nodes[calibration].setdefault("inputs", []).append( {"entity": "person", "columns": [INCOME_COLUMN, UNIVERSE_COLUMN]} ) @@ -730,3 +826,15 @@ def test_composition_refuses_a_later_version_rewrite_of_a_measured_column( f"calibration node {calibration!r} has writers ['toy.measure', " f"'toy.overwrite'] of person.{INCOME_COLUMN} upstream" ) in message + + +def test_composition_refuses_an_expand_of_a_measured_column(composed_toy): + spec, calibration = _composed_spec(composed_toy, overwrite=False, expand=True) + with pytest.raises(ValueError) as refused: + compose_transport_graph(spec, composed_toy.config) + message = str(refused.value) + assert message.startswith("Transport spec is not activated:") + assert ( + f"calibration node {calibration!r} has writers ['toy.expand', " + f"'toy.measure'] of person.{INCOME_COLUMN} upstream" + ) in message From bf418af14c70abc8cf310b8034ea824c05c0f2e1 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Mon, 2 Nov 2026 19:48:54 -0500 Subject: [PATCH 8/8] Pin the chain property's rare branches Review 200 measured the property reaching R2C-15's branch in 16 of 20 seeded runs, about three examples a run, and the proof about once a run. Three explicit examples now cover one declarer alone and a second writer after it and before it. With _second_writer_refusal never refusing, both second-writer examples fail. Co-Authored-By: Claude Opus 5.5 --- .../engine_free/shared/test_transport_measurement_proof.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py index 0d1673aacf..de485f7674 100644 --- a/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -14,7 +14,7 @@ import json import pytest -from hypothesis import HealthCheck, given, settings +from hypothesis import HealthCheck, example, given, settings from hypothesis import strategies as st from microcosm.build.transport.compose import ( @@ -712,6 +712,11 @@ def _chain(draws): deadline=None, suppress_health_check=[HealthCheck.too_slow], ) +# Random draws rarely reach these: one declarer alone, then with a second +# writer after it and before it. +@example(draws=[(True, True, True)]) +@example(draws=[(True, True, True), (False, True, False)]) +@example(draws=[(False, False, True), (True, True, True)]) def test_a_chain_proves_exactly_when_one_declarer_is_the_only_writer(draws): proofs, refusals = measurement_proofs(_spec(_chain(draws))) declarers = [index for index, draw in enumerate(draws) if draw[0]]