diff --git a/packages/microcosm-build/src/microcosm/build/reference_profiles.py b/packages/microcosm-build/src/microcosm/build/reference_profiles.py index 5354a795bc..e42ef39769 100644 --- a/packages/microcosm-build/src/microcosm/build/reference_profiles.py +++ b/packages/microcosm-build/src/microcosm/build/reference_profiles.py @@ -69,9 +69,44 @@ 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 +178,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 +201,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 +214,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 + # 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 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", [] + ): + 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,6 +393,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] 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): @@ -224,6 +431,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 +635,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 +672,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 +691,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 +702,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 +763,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 +825,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 +894,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 +904,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 +1070,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 +1152,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 0fef19eef3..e85c658fde 100644 --- a/packages/microcosm-build/src/microcosm/build/transport/compose.py +++ b/packages/microcosm-build/src/microcosm/build/transport/compose.py @@ -460,6 +460,319 @@ 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 + + +MEASUREMENT_PARAM = "measurement_id" + + +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"]``. + """ + 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(): + 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]) + 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 _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. An EXPAND writes the cells it declares. + """ + 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 _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]]: + """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 + ``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. 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 + 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 {}, {} + if _has_resource(spec, GRAPH_RESOURCE): + document = _raw_transport_resource(spec, GRAPH_RESOURCE) + 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 _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()): + 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 _measurement_declaration(row) == binding + ) + producers, reason = {}, None + 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" + 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 = _measurement_refusal(binding, root, found, declared) + 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 + reason = _second_writer_refusal( + binding, + root, + producer, + { + (entity, column): sorted( + name + for name in closure + if _declared_writer(rows[name], entity, column) + ) + for entity, column in required + }, + ) + if reason is not None: + 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 _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. That producer must also be the only compiled + ancestor whose outputs or EXPAND cells hold each of the proof's columns + (R2C-15). + """ + 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 + } + 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): + 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}." + ) + 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 (entity, column) in expand_cells[name] + or 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( spec: Mapping[str, object], selection: Mapping | None = None ): @@ -488,6 +801,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) @@ -644,8 +958,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 @@ -665,7 +981,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]: @@ -1112,6 +1445,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) @@ -1211,6 +1545,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(): @@ -1221,6 +1556,7 @@ def validate_transport_activation(spec: Mapping[str, object]) -> None: resource, reference_selections.get(resource), selection_closure.get(resource), + measurement_refusals, ) ) receipt = ( @@ -3021,21 +3357,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 @@ -4041,6 +4363,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 new file mode 100644 index 0000000000..9302b0bb85 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_reference_summed_bands.py @@ -0,0 +1,353 @@ +"""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 ( + reference_document_sha256, + 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, + with_target_labels, +) + +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]] + 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 + + +@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"), + ("mapping_member", "repeat or are not scalars"), + ("list_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) + + +@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: + 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 + for row in rows: + row["metadata"]["contract_target_id"] = row["name"] + with_target_labels(base) + 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_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/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..de485f7674 --- /dev/null +++ b/packages/microcosm-build/tests/engine_free/shared/test_transport_measurement_proof.py @@ -0,0 +1,845 @@ +"""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 +``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. +""" + +from __future__ import annotations + +import copy +import json + +import pytest +from hypothesis import HealthCheck, example, 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, +) +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): + """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", + "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["params"] = {"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": producer_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"]["params"] + + +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): + # 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( + 2, + { + "id": "toy.universe", + "kernel": "measure.combine@1", + "population": "toy.create", + "outputs": [ + {"entity": "person", "column": UNIVERSE_COLUMN, "dtype": "str"} + ], + }, + ) + + +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")) + + +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, + "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"), + ("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"), + ], +) +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 + + +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. With + ``expand_cells``, preparation and calibration live in an EXPAND version + that declares those cells. + """ + 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") + 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),), + ), + ] + 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", + kernel="calibration.prepare@1", + 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=population, + 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, 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": list(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_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_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"] = { + "measurement_id": {"resource": "factor_context", "path": [], "encoding": "json"} + } + 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"} + + +@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. + + 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], +) +# 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]] + 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, 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 + 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", + }, + ) + 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]} + ) + 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 + + +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 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": { diff --git a/test_support/microcosm_build/reference_intervals.py b/test_support/microcosm_build/reference_intervals.py index ad7d8bf489..3c12f891ca 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,190 @@ 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}, + "dimension_labels": {"band": "Band"}, + "dimension_value_labels": {"band": {label: f"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 = 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 { + SUMMED_MEASUREMENT: { + "measurement_id": SUMMED_MEASUREMENT, + "producers": {calibration: producer}, + "columns": [f"person.{INCOME_COLUMN}", f"person.{UNIVERSE_COLUMN}"], + } + }