diff --git a/docs/design/exact-output-consumers.md b/docs/design/exact-output-consumers.md index 45e587f90..6f74b73d8 100644 --- a/docs/design/exact-output-consumers.md +++ b/docs/design/exact-output-consumers.md @@ -53,6 +53,28 @@ ScientificOutput( ) ``` +When the native runtime owns an AMR reflux correction and/or an authored projection, the ledger can +delegate those exact terms instead of requiring zero placeholders. `component` is the exact +conservative index shared by the explicit Program sums and native evidence (it defaults to zero for +a scalar state); the optional typed role is checked against that index at bind: + +```python +from pops.physics.roles import Density + +mass = BalanceLedger( + "mass", + role=Density(), + component=0, + automatic_terms=("projection", "reflux"), +) +program.record_balance( + mass, + storage_change=storage_increment, + outward_boundary_flux=boundary_flux_increment, + sources=source_increment, +) +``` + Le fournisseur possède l'extension. Une cible comme `solution/tracer.vtu` est refusée dès l'authoring, avant le bind ; elle empêcherait le changement de format et entrerait en collision au deuxième échantillon. Chaque pas accepté dû publie immédiatement un fichier distinct sous le chemin @@ -451,15 +473,20 @@ schedule and transaction. Its reductions are completed on the simulation thread the post-commit worker receives only immutable arrays and scalar payloads, never the native mailbox or communicator facade. -Each argument to `record_balance` is a signed, time-integrated native Program sum/dot reduction, -or scalar arithmetic composed only from such reductions and exact literals. +Each non-automatic argument to `record_balance` is a signed, time-integrated native Program sum/dot +reduction, or scalar arithmetic composed only from such reductions and exact literals. When any +term is delegated to a native producer, every explicit term must instead be composed from +component-qualified `sum` reductions for the ledger's exact `component`; an all-state dot product +cannot be reconciled with one component's reflux/projection evidence. The reported residual is `storage_change + outward_boundary_flux - sources - reflux - projection`. The native attempt mailbox accumulates repeated cadence/substep invocations, rejects missing or non-finite terms, and is cleared before the next attempt. The consumer reads it only while the outer accepted-step transaction still retains the pre-step image. Python therefore packages the five returned scalars and residual but never traverses arrays, invents a zero term, or reuses a -previous step. A rejected attempt or failed consumer publication restores the mailbox with the -rest of the native transaction. +previous step. Selected automatic terms are resolved by exact runtime block, active hierarchy level +and conservative component. A missing coordinate, a non-finite value, or simultaneous Program and +native authority for one term fails the accepted transaction. A rejected attempt or failed consumer +publication restores both mailboxes with the rest of the native transaction. The `pops.balance-term` namespace is reserved. Ordinary `Program.record_scalar(...)` authoring and the Python runtime diagnostic binding both reject it; generated `record_balance` code reaches a @@ -476,8 +503,9 @@ by an OR of their exact accepted-step periods. `Always` and `when(True)` are per The compiler traces the complete reduction/scalar chain rather than scheduling only the terminal records. If a value is also consumed by an ordinary Program diagnostic or another non-balance operation, that shared producer remains unconditional so cadence fusion cannot change unrelated -semantics. A `Balance` consumer with no matching five-term `Program.record_balance` producer fails -before native code generation. Program stride/substeps use one attempt-local outer accepted-step +semantics. A `Balance` consumer with no complete matching `Program.record_balance` producer for all +non-automatic terms fails before native code generation. Program stride/substeps use one +attempt-local outer accepted-step target, so every substep of one due public step sees the same decision and accumulates into the same attempt mailbox. The cadence is authored once as part of the Program identity, for example `program.cadence(substeps=2, stride=3)`, then authenticated and installed before runtime freeze on @@ -498,22 +526,22 @@ balance reductions are not yet skipped. This fallback can add work but cannot su evidence. A zero-step run has no accepted native occurrence: its coincident start/end moment cannot publish an accepted-step consumer, including `Balance`. -This public route still consumes explicit evidence: a Program that cannot produce every actual term -cannot declare `Balance`. Native operator instrumentation is deliberately kept in a separate, -qualified attempt-local mailbox until a resolved quantity selector can prove which -`BalanceLedger` route owns each block/level/component contribution. Generated code publishes the OR -of the exact due route decisions before the first Program operator; the marker is monotone for the -attempt, disabled during replay, and reset at attempt entry. Consequently off-cadence steps do not -pay for automatic operator reductions. - -That private mailbox currently captures the signed AMR reflux correction and the before/after -projection delta. Uniform Cartesian projection uses the authenticated cell measure and embedded -boundary mask; AMR projection excludes covered coarse cells and performs one component-vector -collective per participating level. Polar projection stays absent because no exact per-cell polar -volume provider exists on this path. Automatic physical-boundary flux and source evidence are also -not yet producers. None of these private values is read by `accepted_balance_terms()`, so this -instrumentation does not silently complete an authored five-term balance or widen the public -contract. +The selected public route now consumes signed AMR reflux corrections and before/after projection +deltas from the separate qualified attempt mailbox. Uniform Cartesian projection uses the +authenticated cell measure and embedded-boundary mask; AMR projection excludes covered coarse cells +and performs one component-vector collective per participating level. A reflux selection requires +an adaptive hierarchy and expects one contribution for every active parent/fine interface; +projection expects one for every selected active level. Generated code publishes the OR of the exact +due route decisions before the first Program operator; the marker is monotone for the attempt, +disabled during replay, and reset at attempt entry. Consequently off-cadence steps do not pay for +automatic operator reductions. + +The capability remains deliberately bounded. Polar projection is rejected because no exact +per-cell polar volume provider exists on this path. Automatic physical-boundary flux and source +evidence are not yet producers and therefore remain explicit `Program.record_balance` arguments. +The native selector never substitutes a missing automatic value with zero (except the exact reflux +identity for a hierarchy with no coarse/fine interface), and the legacy all-explicit ledger route +retains its original identity and behavior. Checkpoint remains a separate restart effect. These consumers do not define a checkpoint schema or reader and do not call the scientific-output manifest a restart identity. The checkpoint provider diff --git a/include/pops/runtime/amr_system.hpp b/include/pops/runtime/amr_system.hpp index 29bdfe8fe..06d7c10ab 100644 --- a/include/pops/runtime/amr_system.hpp +++ b/include/pops/runtime/amr_system.hpp @@ -903,6 +903,10 @@ class AmrSystem { /// Five current-attempt scalars for one typed balance route. RuntimeInstance calls this only /// inside its active outer accepted-step transaction; missing/stale/non-finite evidence fails. POPS_EXPORT std::map accepted_balance_terms(const std::string& route) const; + /// The same accepted route with selected attempt-local native reflux/projection producers. + POPS_EXPORT std::map selected_accepted_balance_terms( + const std::string& route, const std::string& block, int component, + const std::vector& levels, const std::vector& automatic_terms) const; POPS_EXPORT void begin_step_projection_report(); POPS_EXPORT void note_step_projection(const std::string& name); POPS_EXPORT std::vector consume_step_projections(); diff --git a/include/pops/runtime/program/program_runtime_state.hpp b/include/pops/runtime/program/program_runtime_state.hpp index 406faaeed..436a5fe3c 100644 --- a/include/pops/runtime/program/program_runtime_state.hpp +++ b/include/pops/runtime/program/program_runtime_state.hpp @@ -969,6 +969,94 @@ struct ProgramRuntimeState { return result; } + /// Resolve one public Balance route against exact native operator coordinates. + /// + /// Explicit Program records remain authoritative for every term not listed in @p automatic_terms. + /// Reflux and projection may instead be selected from the attempt-local native mailbox. The + /// selector is complete and owner-qualified: one runtime block, one conservative component and + /// the full active contiguous hierarchy. A selected producer must have published every expected + /// coordinate; missing evidence and duplicate Program/native authority fail instead of becoming + /// zero or reusing a stale value. + std::map selected_accepted_balance_terms( + const std::string& route, int runtime_block, int component, const std::vector& levels, + const std::vector& automatic_terms, const std::string& runtime) const { + static constexpr std::array kTerms{"storage_change", "outward_boundary_flux", + "sources", "reflux", "projection"}; + require_balance_route(route, runtime + "::_selected_accepted_balance_terms"); + if (runtime_block < 0 || component < 0) + throw std::invalid_argument( + runtime + "::_selected_accepted_balance_terms requires non-negative coordinates"); + if (levels.empty() || levels.front() < 0 || + std::adjacent_find(levels.begin(), levels.end(), + [](int left, int right) { return right != left + 1; }) != levels.end()) + throw std::invalid_argument( + runtime + "::_selected_accepted_balance_terms requires a non-empty contiguous hierarchy"); + if (!std::is_sorted(automatic_terms.begin(), automatic_terms.end()) || + std::adjacent_find(automatic_terms.begin(), automatic_terms.end()) != automatic_terms.end()) + throw std::invalid_argument( + runtime + "::_selected_accepted_balance_terms requires sorted unique automatic terms"); + for (const std::string& term : automatic_terms) + if (term != "reflux" && term != "projection") + throw std::invalid_argument( + runtime + "::_selected_accepted_balance_terms has no native producer for '" + term + + "'"); + + std::map result; + if (step_balance_terms_.empty() && balance_step_completed_ && !balance_program_was_due_) { + for (const char* term : kTerms) + result.emplace(term, Real(0)); + return result; + } + for (const char* term_value : kTerms) { + const std::string term = term_value; + const bool automatic = + std::binary_search(automatic_terms.begin(), automatic_terms.end(), term); + const std::string record = "pops.balance-term.v1:" + route + ":" + term; + const auto authored = step_balance_terms_.find(record); + if (!automatic) { + if (authored == step_balance_terms_.end()) + throw std::runtime_error( + runtime + + "::_selected_accepted_balance_terms: current native attempt omitted term '" + term + + "'; Program.record_balance must publish every non-automatic term"); + if (!std::isfinite(static_cast(authored->second))) + throw std::runtime_error( + runtime + + "::_selected_accepted_balance_terms: current native attempt produced " + "non-finite term '" + + term + "'"); + result.emplace(term, authored->second); + continue; + } + if (authored != step_balance_terms_.end()) + throw std::runtime_error(runtime + "::_selected_accepted_balance_terms: term '" + term + + "' has both Program and native producer authority"); + + Real value = Real(0); + const std::size_t expected = term == "reflux" ? levels.size() - 1 : levels.size(); + for (std::size_t index = 0; index < expected; ++index) { + const AutomaticBalanceKey key{runtime_block, levels[index], component, term}; + const auto found = automatic_balance_terms_.find(key); + if (found == automatic_balance_terms_.end()) + throw std::runtime_error( + runtime + "::_selected_accepted_balance_terms: native producer omitted term '" + + term + "' at level " + std::to_string(levels[index])); + if (!std::isfinite(static_cast(found->second))) + throw std::runtime_error( + runtime + + "::_selected_accepted_balance_terms: native producer returned non-finite " + "term '" + + term + "'"); + value += found->second; + } + if (!std::isfinite(static_cast(value))) + throw std::runtime_error( + runtime + "::_selected_accepted_balance_terms: native term accumulation overflowed"); + result.emplace(term, value); + } + return result; + } + void begin_balance_due_window(int accepted_macro_step, const std::string& runtime) { if (balance_due_window_active_) throw std::logic_error(runtime + " balance due window is already active"); diff --git a/include/pops/runtime/system.hpp b/include/pops/runtime/system.hpp index 579275ff9..63837952b 100644 --- a/include/pops/runtime/system.hpp +++ b/include/pops/runtime/system.hpp @@ -1240,6 +1240,10 @@ class System { /// Five current-attempt scalars for one typed balance route. RuntimeInstance calls this only /// inside its active outer accepted-step transaction; missing/stale/non-finite evidence fails. POPS_EXPORT std::map accepted_balance_terms(const std::string& route) const; + /// The same accepted route with selected attempt-local native reflux/projection producers. + POPS_EXPORT std::map selected_accepted_balance_terms( + const std::string& route, const std::string& block, int component, + const std::vector& levels, const std::vector& automatic_terms) const; POPS_EXPORT void begin_step_projection_report(); POPS_EXPORT void note_step_projection(const std::string& name); POPS_EXPORT std::vector consume_step_projections(); diff --git a/python/bindings/core/init/init_amr.cpp b/python/bindings/core/init/init_amr.cpp index cbde10c19..3336d071b 100644 --- a/python/bindings/core/init/init_amr.cpp +++ b/python/bindings/core/init/init_amr.cpp @@ -824,6 +824,9 @@ void bind_amr_program(py::class_& cls) { .def("program_diagnostic", &AmrSystem::program_diagnostic, py::arg("name")) .def("program_diagnostics", &AmrSystem::program_diagnostics) .def("_accepted_balance_terms", &AmrSystem::accepted_balance_terms, py::arg("route")) + .def("_selected_accepted_balance_terms", &AmrSystem::selected_accepted_balance_terms, + py::arg("route"), py::arg("block"), py::arg("component"), py::arg("levels"), + py::arg("automatic_terms")) .def("_consume_step_projections", &AmrSystem::consume_step_projections) .def("record_program_diagnostic", &AmrSystem::record_program_diagnostic, py::arg("name"), py::arg("value")) diff --git a/python/bindings/core/init/init_system.cpp b/python/bindings/core/init/init_system.cpp index cca9cc0ed..51c260046 100644 --- a/python/bindings/core/init/init_system.cpp +++ b/python/bindings/core/init/init_system.cpp @@ -355,6 +355,9 @@ void bind_system_program(py::class_& cls) { .def("program_diagnostic", &System::program_diagnostic, py::arg("name")) .def("program_diagnostics", &System::program_diagnostics) .def("_accepted_balance_terms", &System::accepted_balance_terms, py::arg("route")) + .def("_selected_accepted_balance_terms", &System::selected_accepted_balance_terms, + py::arg("route"), py::arg("block"), py::arg("component"), py::arg("levels"), + py::arg("automatic_terms")) .def("_consume_step_projections", &System::consume_step_projections) // ADC-542: the native collective reduction over a named block the diagnostics driver drives to // fire a declared typed measure (Norm / Integral / MinMax) each cadence tick, and the sink the diff --git a/python/pops/_balance_contract.py b/python/pops/_balance_contract.py index 2fcf86671..14499db5a 100644 --- a/python/pops/_balance_contract.py +++ b/python/pops/_balance_contract.py @@ -31,31 +31,79 @@ def _canonical_name(value: Any, *, where: str) -> str: class BalanceLedger: """Identity joining one Program-authored discrete balance to one consumer. - The ledger does not contain values. :meth:`Program.record_balance` writes the five reduced - scalars into the current native step-attempt mailbox, while + The ledger does not contain values. :meth:`Program.record_balance` writes the explicitly + authored reduced scalars into the current native step-attempt mailbox. A ledger may delegate + ``reflux`` and/or ``projection`` to exact native operators for one typed component role, while :class:`pops.diagnostics.Balance` selects the same identity after that attempt has advanced successfully. """ name: str + role: Any = None + component: int | None = None + automatic_terms: tuple[str, ...] = () identity: Identity = field(init=False) __pops_ir_immutable__ = True def __post_init__(self) -> None: name = _canonical_name(self.name, where="BalanceLedger.name") + role = None + if self.role is not None: + from pops.physics.roles import native_role_token + + try: + role = native_role_token(self.role) + except TypeError as exc: + raise TypeError( + "BalanceLedger.role must be a typed pops.physics.roles.ComponentRole" + ) from exc + if not isinstance(self.automatic_terms, tuple): + raise TypeError("BalanceLedger.automatic_terms must be a tuple") + automatic_terms = tuple(sorted(set(self.automatic_terms))) + if len(automatic_terms) != len(self.automatic_terms): + raise ValueError("BalanceLedger.automatic_terms must be unique") + unsupported = set(automatic_terms).difference({"reflux", "projection"}) + if unsupported: + raise ValueError( + "BalanceLedger automatic native producers currently support only " + "reflux and projection; got %s" % sorted(unsupported) + ) + component = self.component + if automatic_terms and component is None: + component = 0 + if component is not None and (type(component) is not int or component < 0): + raise TypeError("BalanceLedger.component must be a non-negative int or None") object.__setattr__(self, "name", name) + object.__setattr__(self, "component", component) + object.__setattr__(self, "automatic_terms", automatic_terms) + payload: dict[str, Any] = {"schema_version": 1, "name": name} + if role is not None: + payload["role"] = role + if component is not None: + payload["component"] = component + if automatic_terms: + payload["automatic_terms"] = list(automatic_terms) object.__setattr__( self, "identity", - make_identity("balance-ledger", {"schema_version": 1, "name": name}), + make_identity("balance-ledger", payload), ) def to_data(self) -> dict[str, Any]: - return { + data = { "schema_version": 1, "name": self.name, "identity": self.identity.to_data(), } + if self.role is not None: + from pops.physics.roles import native_role_token + + data["role"] = native_role_token(self.role) + if self.component is not None: + data["component"] = self.component + if self.automatic_terms: + data["automatic_terms"] = list(self.automatic_terms) + return data def route_identity(self, block: Any) -> Identity: from pops.problem.handles import BlockHandle diff --git a/python/pops/_balance_due_contract.py b/python/pops/_balance_due_contract.py index 3052caaf7..f2bf15186 100644 --- a/python/pops/_balance_due_contract.py +++ b/python/pops/_balance_due_contract.py @@ -51,6 +51,7 @@ class BalanceDueRoute: route: Identity consumers: tuple[BalanceDueConsumer, ...] + automatic_terms: tuple[str, ...] = () def __post_init__(self) -> None: object.__setattr__( @@ -75,12 +76,21 @@ def __post_init__(self) -> None: if len(identities) != len(set(identities)): raise ValueError("BalanceDueRoute contains a duplicate consumer") object.__setattr__(self, "consumers", consumers) + if not isinstance(self.automatic_terms, tuple): + raise TypeError("BalanceDueRoute.automatic_terms must be a tuple") + if self.automatic_terms != tuple(sorted(set(self.automatic_terms))): + raise ValueError("BalanceDueRoute.automatic_terms must be sorted and unique") + if set(self.automatic_terms).difference({"reflux", "projection"}): + raise ValueError("BalanceDueRoute names an unavailable automatic balance producer") def to_data(self) -> dict[str, Any]: - return { + data = { "route": self.route.to_data(), "consumers": [value.to_data() for value in self.consumers], } + if self.automatic_terms: + data["automatic_terms"] = list(self.automatic_terms) + return data def accepted_step_periods(self) -> tuple[int, ...]: """Return exact native periods, conservatively using period one when unprovable. @@ -161,7 +171,9 @@ def from_consumer_graph(cls, graph: Any) -> BalanceDueContract: raise TypeError( "BalanceDueContract requires an exact resolved ConsumerGraph or None" ) - by_route: dict[str, tuple[Identity, list[BalanceDueConsumer]]] = {} + by_route: dict[ + str, tuple[Identity, list[BalanceDueConsumer], tuple[str, ...]] + ] = {} for manifest in graph.nodes: for quantity in manifest.diagnostic_quantities: for operation in quantity.execution["operations"]: @@ -173,15 +185,22 @@ def from_consumer_graph(cls, graph: Any) -> BalanceDueContract: "balance-ledger-route", where="accepted balance operation route", ) - existing = by_route.setdefault(route.token, (route, [])) + automatic_terms = tuple(operation.get("automatic_terms", ())) + existing = by_route.setdefault( + route.token, (route, [], automatic_terms) + ) + if existing[2] != automatic_terms: + raise ValueError( + "one balance route cannot select different automatic producers" + ) existing[1].append( BalanceDueConsumer(manifest.identity, manifest.schedule) ) return cls( graph.identity, tuple( - BalanceDueRoute(route, tuple(consumers)) - for route, consumers in by_route.values() + BalanceDueRoute(route, tuple(consumers), automatic_terms) + for route, consumers, automatic_terms in by_route.values() ), ) diff --git a/python/pops/_pops.pyi b/python/pops/_pops.pyi index 4b8c5d49d..0c6ffdc46 100644 --- a/python/pops/_pops.pyi +++ b/python/pops/_pops.pyi @@ -277,6 +277,14 @@ class System: def solve_fields(self) -> _SolveReport: ... def _consume_step_projections(self) -> list[str]: ... def _accepted_balance_terms(self, route: str) -> dict[str, float]: ... + def _selected_accepted_balance_terms( + self, + route: str, + block: str, + component: int, + levels: list[int], + automatic_terms: list[str], + ) -> dict[str, float]: ... def output_state_local_pieces( self, block: str, level: int ) -> tuple[dict[str, object], ...]: ... @@ -297,6 +305,14 @@ class AmrSystem: def configured_n_levels(self) -> int: ... def _consume_step_projections(self) -> list[str]: ... def _accepted_balance_terms(self, route: str) -> dict[str, float]: ... + def _selected_accepted_balance_terms( + self, + route: str, + block: str, + component: int, + levels: list[int], + automatic_terms: list[str], + ) -> dict[str, float]: ... def materialize_program_restart_histories( self, payload: bytes, diff --git a/python/pops/codegen/program_balance_due.py b/python/pops/codegen/program_balance_due.py index 1a57dafe5..2d2c2336b 100644 --- a/python/pops/codegen/program_balance_due.py +++ b/python/pops/codegen/program_balance_due.py @@ -99,15 +99,6 @@ def _program_balance_records( ) by_term[term] = value record_routes[value.id] = route.token - expected = set(BALANCE_TERM_NAMES) - for route, by_term in terms.items(): - if set(by_term) != expected: - missing = sorted(expected.difference(by_term)) - extra = sorted(set(by_term).difference(expected)) - raise ValueError( - "Program balance route %s must record exactly five terms; missing=%s extra=%s" - % (route, missing, extra) - ) return operations, record_routes, terms @@ -118,13 +109,23 @@ def validate_balance_due_contract(program: Any, contract: Any) -> None: "balance due validation requires an exact BalanceDueContract" ) _operations, _records, terms = _program_balance_records(program) - missing = sorted( - row.route.token for row in contract.routes if row.route.token not in terms - ) - if missing: + failures = [] + for row in contract.routes: + expected = set(BALANCE_TERM_NAMES).difference(row.automatic_terms) + actual = set(terms.get(row.route.token, {})) + if actual != expected: + failures.append( + "%s missing=%s extra=%s" + % ( + row.route.token, + sorted(expected.difference(actual)), + sorted(actual.difference(expected)), + ) + ) + if failures: raise ValueError( "ConsumerGraph Balance routes have no Program.record_balance producer: %s" - % ", ".join(missing) + % "; ".join(failures) ) @@ -137,6 +138,7 @@ def prepare_balance_due_lowering( raise TypeError( "balance due lowering requires an exact BalanceDueContract" ) + validate_balance_due_contract(program, contract) operations, record_routes, terms = _program_balance_records(program) route_periods = { route: ( diff --git a/python/pops/diagnostics/measures.py b/python/pops/diagnostics/measures.py index bf4c36cf8..d595d5398 100644 --- a/python/pops/diagnostics/measures.py +++ b/python/pops/diagnostics/measures.py @@ -334,7 +334,7 @@ def __init__( ) if block is None: raise TypeError("Balance(block=...) requires an exact physics BlockHandle") - super().__init__(block=block, role=None, cadence=cadence) + super().__init__(block=block, role=ledger.role, cadence=cadence) self.ledger = ledger def options(self) -> dict: @@ -346,11 +346,19 @@ def diagnostic_execution(self) -> dict[str, Any]: route = self.ledger.route_identity(self.block) return { "schema_version": 1, - "role": None, + "role": _role_name(self.ledger.role), "operations": [ { **_operation("balance", "accepted_balance"), "balance_route": route.token, + **( + { + "automatic_terms": list(self.ledger.automatic_terms), + "balance_component": self.ledger.component, + } + if self.ledger.automatic_terms + else {} + ), }, ], "conservation": None, diff --git a/python/pops/output/_consumer_contracts.py b/python/pops/output/_consumer_contracts.py index 6d4f7517c..a0323c9b2 100644 --- a/python/pops/output/_consumer_contracts.py +++ b/python/pops/output/_consumer_contracts.py @@ -343,6 +343,9 @@ def _diagnostic_execution(value: Any) -> Mapping[str, Any]: expected = {"name", "reduction", "transform", "metric_weighted"} if reduction == "accepted_balance": expected.add("balance_route") + if "automatic_terms" in operation: + expected.add("automatic_terms") + expected.add("balance_component") if set(operation) != expected: raise TypeError("%s has an unknown schema" % where) name = _text(operation["name"], "%s.name" % where) @@ -373,6 +376,27 @@ def _diagnostic_execution(value: Any) -> Mapping[str, Any]: "accepted balance route must use the version-1 balance-ledger-route identity" ) row["balance_route"] = route.token + automatic_terms = operation.get("automatic_terms", ()) + if not isinstance(automatic_terms, (tuple, list)): + raise TypeError("%s.automatic_terms must be a sequence" % where) + automatic_terms = tuple(automatic_terms) + if automatic_terms != tuple(sorted(set(automatic_terms))): + raise ValueError( + "%s.automatic_terms must be sorted and unique" % where + ) + unsupported = set(automatic_terms).difference({"reflux", "projection"}) + if unsupported: + raise ValueError( + "%s.automatic_terms names an unavailable native producer" % where + ) + if automatic_terms: + row["automatic_terms"] = list(automatic_terms) + component = operation["balance_component"] + if type(component) is not int or component < 0: + raise TypeError( + "%s.balance_component must be a non-negative int" % where + ) + row["balance_component"] = component normalized.append(row) if len({row["name"] for row in normalized}) != len(normalized): raise ValueError("DiagnosticQuantity execution operation names must be unique") @@ -383,8 +407,6 @@ def _diagnostic_execution(value: Any) -> Mapping[str, Any]: raise ValueError( "accepted balance evidence must be the sole diagnostic execution operation" ) - if has_accepted_balance and role is not None: - raise ValueError("accepted balance evidence cannot select one component role") conservation = value["conservation"] normalized_conservation = None if conservation is not None: diff --git a/python/pops/runtime/_runtime_consumers.py b/python/pops/runtime/_runtime_consumers.py index a0d040ee1..872cb4466 100644 --- a/python/pops/runtime/_runtime_consumers.py +++ b/python/pops/runtime/_runtime_consumers.py @@ -2433,9 +2433,44 @@ def _validate_diagnostic_providers(self) -> None: if reductions == {"accepted_balance"}: if len(quantity.execution["operations"]) != 1: raise ValueError("accepted balance requires exactly one native evidence route") - if quantity.execution["role"] is not None: - raise ValueError("accepted balance route cannot carry a component role") - if not callable(getattr(engine, "_accepted_balance_terms", None)): + operation, = quantity.execution["operations"] + automatic_terms = tuple(operation.get("automatic_terms", ())) + if automatic_terms: + if not callable( + getattr(engine, "_selected_accepted_balance_terms", None) + ): + raise NotImplementedError( + "automatic balance terms require native " + "_selected_accepted_balance_terms(...)" + ) + component = operation["balance_component"] + if component >= len(names): + raise ValueError( + "automatic balance component %d is outside block %r width %d" + % (component, block, len(names)) + ) + if quantity.execution["role"] is not None: + role_component, _ = self._diagnostic_component( + names, roles, quantity.execution["role"] + ) + if role_component != component: + raise ValueError( + "automatic balance role selects component %d but ledger " + "declares component %d" % (role_component, component) + ) + if "reflux" in automatic_terms and not layout.adaptive: + raise NotImplementedError( + "automatic reflux balance requires an adaptive hierarchy" + ) + if ( + "projection" in automatic_terms + and layout.geometry.cell_measure != CARTESIAN_CELL_AREA + ): + raise NotImplementedError( + "automatic projection balance requires exact Cartesian cell " + "measure support" + ) + elif not callable(getattr(engine, "_accepted_balance_terms", None)): raise NotImplementedError( "balance diagnostic requires native _accepted_balance_terms(route)" ) @@ -2556,14 +2591,31 @@ def _native_diagnostic_reduction( return float(cast(Any, native)(block, kind, component)), False @staticmethod - def _native_balance_terms(engine: Any, route: str) -> Any: + def _native_balance_terms( + engine: Any, + route: str, + *, + block: str, + component: int, + levels: tuple[int, ...], + automatic_terms: tuple[str, ...], + ) -> Any: """Read one current-attempt balance tuple from the native transaction mailbox.""" from pops.output.diagnostics import BalanceTerms - native = getattr(engine, "_accepted_balance_terms", None) + native_name = ( + "_selected_accepted_balance_terms" + if automatic_terms + else "_accepted_balance_terms" + ) + native = getattr(engine, native_name, None) if not callable(native): raise RuntimeError("installed runtime has no accepted balance evidence provider") - raw = native(route) + raw = ( + native(route, block, component, list(levels), list(automatic_terms)) + if automatic_terms + else native(route) + ) required = { "storage_change", "outward_boundary_flux", @@ -2612,8 +2664,16 @@ def _diagnostic_values( if "accepted_balance" in skip_reductions: continue operation, = execution["operations"] + automatic_terms = tuple(operation.get("automatic_terms", ())) + component = operation.get("balance_component", 0) balance = self._native_balance_terms( - engine, operation["balance_route"]) + engine, + operation["balance_route"], + block=block, + component=component, + levels=levels, + automatic_terms=automatic_terms, + ) terms = { "storage_change": balance.storage_change, "outward_boundary_flux": balance.outward_boundary_flux, diff --git a/python/pops/time/_program/contract.py b/python/pops/time/_program/contract.py index 4e39deeaf..cc262ddf4 100644 --- a/python/pops/time/_program/contract.py +++ b/python/pops/time/_program/contract.py @@ -208,8 +208,8 @@ def record_balance( storage_change: Any, outward_boundary_flux: Any, sources: Any, - reflux: Any, - projection: Any, + reflux: Any = None, + projection: Any = None, ) -> tuple[Any, ...]: ... # --- solve / commit / board sugar (_ProgramSolve) --- diff --git a/python/pops/time/_program/diagnostics.py b/python/pops/time/_program/diagnostics.py index 75d3ab00f..16bf5a501 100644 --- a/python/pops/time/_program/diagnostics.py +++ b/python/pops/time/_program/diagnostics.py @@ -34,17 +34,19 @@ def record_balance( storage_change: Any, outward_boundary_flux: Any, sources: Any, - reflux: Any, - projection: Any, + reflux: Any = None, + projection: Any = None, ) -> tuple[ProgramValue, ...]: """Publish one exact five-term balance into the current native attempt. - Every term is a signed, time-integrated increment for this Program invocation and - must be an additive global Program reduction (sum/dot), or scalar arithmetic composed - exclusively from such reductions and exact literals. The native mailbox accumulates - these increments across cadence substeps in the same public macro-step. Raw Python values, - extrema/norm reductions, and rank-local runtime scalars are rejected. The five records are - attempt-local: a rejected step or consumer rollback cannot leave evidence for a later sample. + Every explicitly authored term is a signed, time-integrated increment for this Program + invocation and must be an additive global Program reduction (sum/dot), or scalar arithmetic + composed exclusively from such reductions and exact literals. A ledger that explicitly + delegates ``reflux`` or ``projection`` to its native producer requires the corresponding + argument to remain ``None``. The native mailbox accumulates all increments across cadence + substeps in the same public macro-step. Raw Python values, extrema/norm reductions, and + rank-local runtime scalars are rejected. The records are attempt-local: a rejected step or + consumer rollback cannot leave evidence for a later sample. """ from pops._balance_contract import ( BALANCE_TERM_NAMES, @@ -90,10 +92,47 @@ def require_reduced(value: Any, term: str, seen: set[int]) -> ProgramValue: "only from global reductions; got scalar op %r" % (term, value.op) ) + automatic = set(ledger.automatic_terms) + for name in automatic: + if supplied[name] is not None: + raise ValueError( + "record_balance %s is owned by the ledger's native automatic producer; " + "leave it as None" % name + ) terms = { name: require_reduced(supplied[name], name, set()) for name in BALANCE_TERM_NAMES + if name not in automatic } + if automatic: + expected_component = ledger.component + + def reduced_components( + value: ProgramValue, term: str, seen: set[int] + ) -> set[int]: + if value.id in seen: + return set() + seen.add(value.id) + if value.op == "reduce": + component = value.attrs.get("comp") + if value.attrs.get("kind") != "sum" or type(component) is not int: + raise ValueError( + "record_balance %s must use component-qualified sum reductions " + "when native terms are selected" % term + ) + return {component} + components: set[int] = set() + for item in value.inputs: + components.update(reduced_components(item, term, seen)) + return components + + for name, value in terms.items(): + components = reduced_components(value, name, set()) + if components != {expected_component}: + raise ValueError( + "record_balance %s selects components %s but the native ledger owns " + "component %d" % (name, sorted(components), expected_component) + ) blocks = {value.block for value in terms.values()} if None in blocks or len(blocks) != 1: raise ValueError( @@ -114,6 +153,7 @@ def require_reduced(value: Any, term: str, seen: set[int]) -> ProgramValue: terms[name].block, ) for name in BALANCE_TERM_NAMES + if name in terms ) @atomic_authoring diff --git a/src/runtime/amr/amr_system.cpp b/src/runtime/amr/amr_system.cpp index 3e0403973..22050acba 100644 --- a/src/runtime/amr/amr_system.cpp +++ b/src/runtime/amr/amr_system.cpp @@ -3579,6 +3579,28 @@ std::map AmrSystem::accepted_balance_terms(const std::strin "transaction"); return p_->program_.accepted_balance_terms(route, "AmrSystem"); } +std::map AmrSystem::selected_accepted_balance_terms( + const std::string& route, const std::string& block, int component, + const std::vector& levels, const std::vector& automatic_terms) const { + if (!p_->external_step_transaction_active_ || p_->external_step_transaction_committed_) + throw std::runtime_error( + "AmrSystem::_selected_accepted_balance_terms requires an active uncommitted external step " + "transaction"); + if (!p_->runtime) + throw std::runtime_error( + "AmrSystem::_selected_accepted_balance_terms requires an installed AMR runtime"); + const std::size_t runtime_block = p_->block_index_or_throw(block); + if (component < 0 || component >= p_->runtime->block_n_vars(runtime_block)) + throw std::out_of_range( + "AmrSystem::_selected_accepted_balance_terms component is out of range"); + if (levels.empty() || std::any_of(levels.begin(), levels.end(), [&](int level) { + return level < 0 || level >= p_->runtime->nlev(); + })) + throw std::out_of_range( + "AmrSystem::_selected_accepted_balance_terms level is out of active hierarchy range"); + return p_->program_.selected_accepted_balance_terms( + route, static_cast(runtime_block), component, levels, automatic_terms, "AmrSystem"); +} void AmrSystem::begin_step_projection_report() { p_->program_.begin_step_projection_report(); } diff --git a/src/runtime/system/system_program.cpp b/src/runtime/system/system_program.cpp index ee3c308f2..6f1ceb023 100644 --- a/src/runtime/system/system_program.cpp +++ b/src/runtime/system/system_program.cpp @@ -435,6 +435,23 @@ std::map System::accepted_balance_terms(const std::string& ro "System::_accepted_balance_terms requires an active uncommitted external step transaction"); return p_->program_.accepted_balance_terms(route, "System"); } +std::map System::selected_accepted_balance_terms( + const std::string& route, const std::string& block, int component, + const std::vector& levels, const std::vector& automatic_terms) const { + if (!p_->external_step_transaction_ || p_->external_step_transaction_committed_) + throw std::runtime_error( + "System::_selected_accepted_balance_terms requires an active uncommitted external step " + "transaction"); + const int runtime_block = p_->index(block); + const auto& state = p_->find(block); + if (component < 0 || component >= state.ncomp) + throw std::out_of_range("System::_selected_accepted_balance_terms component is out of range"); + if (levels != std::vector{0}) + throw std::invalid_argument( + "System::_selected_accepted_balance_terms requires exactly uniform level 0"); + return p_->program_.selected_accepted_balance_terms(route, runtime_block, component, levels, + automatic_terms, "System"); +} void System::begin_step_projection_report() { p_->program_.begin_step_projection_report(); } diff --git a/tests/cpp/integration/runtime/test_program_runtime.cpp b/tests/cpp/integration/runtime/test_program_runtime.cpp index 7dbdd6202..a70d1481f 100644 --- a/tests/cpp/integration/runtime/test_program_runtime.cpp +++ b/tests/cpp/integration/runtime/test_program_runtime.cpp @@ -164,6 +164,37 @@ TEST(ProgramRuntime, AutomaticBalanceDueMarkerIsAttemptLocalMonotoneAndReplaySaf EXPECT_FALSE(state.automatic_balance_capture_due()); } +TEST(ProgramRuntime, SelectedAutomaticBalanceTermsRequireCompleteQualifiedEvidence) { + runtime::program::ProgramRuntimeState state; + const std::string route = "pops.balance-ledger-route.v1:sha256:" + std::string(64, '5'); + state.begin_step_projection_report(); + state.run_balance_due_window(0, "test", [&] { + state.note_automatic_balance_capture_due(true, "test"); + state.record_balance_term(route, "storage_change", 1.0, "test"); + state.record_balance_term(route, "outward_boundary_flux", 2.0, "test"); + state.record_balance_term(route, "sources", 3.0, "test"); + state.record_automatic_balance_term(2, 0, 1, "projection", 0.25, "test"); + state.record_automatic_balance_term(2, 1, 1, "projection", 0.75, "test"); + state.record_automatic_balance_term(2, 0, 1, "reflux", 0.5, "test"); + }); + state.complete_balance_step(true); + + const auto selected = + state.selected_accepted_balance_terms(route, 2, 1, {0, 1}, {"projection", "reflux"}, "test"); + EXPECT_EQ(selected.at("storage_change"), 1.0); + EXPECT_EQ(selected.at("outward_boundary_flux"), 2.0); + EXPECT_EQ(selected.at("sources"), 3.0); + EXPECT_EQ(selected.at("projection"), 1.0); + EXPECT_EQ(selected.at("reflux"), 0.5); + + EXPECT_THROW((void)state.selected_accepted_balance_terms(route, 2, 1, {0, 1, 2}, + {"projection", "reflux"}, "test"), + std::runtime_error); + EXPECT_THROW((void)state.selected_accepted_balance_terms(route, 2, 1, {0, 2}, + {"projection", "reflux"}, "test"), + std::invalid_argument); +} + TEST(ProgramRuntime, SelectiveReplayCompilesBalanceOffAndRestoresTheGuard) { runtime::program::ProgramRuntimeState state; const std::string contract = "pops.balance-due-contract.v1:sha256:" + std::string(64, '3'); diff --git a/tests/python/architecture/test_automatic_projection_balance_fence.py b/tests/python/architecture/test_automatic_projection_balance_fence.py index 8ce73ec53..14064e070 100644 --- a/tests/python/architecture/test_automatic_projection_balance_fence.py +++ b/tests/python/architecture/test_automatic_projection_balance_fence.py @@ -79,7 +79,11 @@ def test_projection_delta_is_captured_only_when_due_and_stays_qualified() -> Non "std::map accepted_balance_terms(", "void begin_balance_due_window(", ) - assert "automatic_balance_terms_" not in accepted + explicit_only = accepted.split( + "std::map selected_accepted_balance_terms(", 1 + )[0] + assert "automatic_balance_terms_" not in explicit_only + assert "automatic_balance_terms_" in accepted def test_uniform_projection_evidence_uses_exact_available_measure() -> None: diff --git a/tests/python/architecture/test_automatic_reflux_balance_fence.py b/tests/python/architecture/test_automatic_reflux_balance_fence.py index 7b2c866a6..2b300d152 100644 --- a/tests/python/architecture/test_automatic_reflux_balance_fence.py +++ b/tests/python/architecture/test_automatic_reflux_balance_fence.py @@ -40,8 +40,12 @@ def test_automatic_balance_mailbox_is_attempt_local_and_not_a_route_fallback() - "std::map accepted_balance_terms(", "void begin_balance_due_window(", ) - assert "step_balance_terms_" in accepted - assert "automatic_balance_terms_" not in accepted + explicit_only = accepted.split( + "std::map selected_accepted_balance_terms(", 1 + )[0] + assert "step_balance_terms_" in explicit_only + assert "automatic_balance_terms_" not in explicit_only + assert "automatic_balance_terms_" in accepted uniform = UNIFORM_IMPL.read_text() adaptive = AMR_IMPL.read_text() diff --git a/tests/python/architecture/test_qualified_automatic_balance_route_fence.py b/tests/python/architecture/test_qualified_automatic_balance_route_fence.py new file mode 100644 index 000000000..71768e4c8 --- /dev/null +++ b/tests/python/architecture/test_qualified_automatic_balance_route_fence.py @@ -0,0 +1,72 @@ +"""ADC-686: public Balance routes select qualified native evidence fail-closed.""" + +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[3] +LEDGER = ROOT / "python" / "pops" / "_balance_contract.py" +MEASURES = ROOT / "python" / "pops" / "diagnostics" / "measures.py" +CONSUMERS = ROOT / "python" / "pops" / "runtime" / "_runtime_consumers.py" +PROGRAM_STATE = ( + ROOT / "include" / "pops" / "runtime" / "program" / "program_runtime_state.hpp" +) +SYSTEM = ROOT / "src" / "runtime" / "system" / "system_program.cpp" +AMR = ROOT / "src" / "runtime" / "amr" / "amr_system.cpp" +SYSTEM_BINDING = ROOT / "python" / "bindings" / "core" / "init" / "init_system.cpp" +AMR_BINDING = ROOT / "python" / "bindings" / "core" / "init" / "init_amr.cpp" + + +def _between(text: str, begin: str, end: str) -> str: + return text.split(begin, 1)[1].split(end, 1)[0] + + +def test_public_ledger_owns_role_and_exact_automatic_term_selection() -> None: + ledger = LEDGER.read_text() + assert "role: Any = None" in ledger + assert "component: int | None = None" in ledger + assert "automatic_terms: tuple[str, ...] = ()" in ledger + assert '{"reflux", "projection"}' in ledger + + measures = MEASURES.read_text() + balance = _between(measures, "class Balance(_Measure):", "class ConservationCheck") + assert "role=ledger.role" in balance + assert '"automatic_terms": list(self.ledger.automatic_terms)' in balance + assert '"balance_component": self.ledger.component' in balance + + +def test_runtime_uses_selected_native_entrypoint_only_for_delegated_terms() -> None: + consumers = CONSUMERS.read_text() + native = _between( + consumers, + "def _native_balance_terms(", + "def _diagnostic_values(", + ) + assert '"_selected_accepted_balance_terms"' in native + assert 'if automatic_terms' in native + assert "native(route, block, component, list(levels), list(automatic_terms))" in native + assert "else native(route)" in native + + for binding in (SYSTEM_BINDING, AMR_BINDING): + assert '"_selected_accepted_balance_terms"' in binding.read_text() + + +def test_native_selector_requires_complete_owner_level_component_evidence() -> None: + state = PROGRAM_STATE.read_text() + selector = _between( + state, + "std::map selected_accepted_balance_terms(", + "void begin_balance_due_window(", + ) + assert "AutomaticBalanceKey key{runtime_block, levels[index], component, term}" in selector + assert "native producer omitted term" in selector + assert "both Program and native producer authority" in selector + assert 'term == "reflux" ? levels.size() - 1 : levels.size()' in selector + + uniform = SYSTEM.read_text() + assert "const int runtime_block = p_->index(block);" in uniform + assert "levels != std::vector{0}" in uniform + + adaptive = AMR.read_text() + assert "const std::size_t runtime_block = p_->block_index_or_throw(block);" in adaptive + assert "p_->runtime->block_n_vars(runtime_block)" in adaptive + assert "p_->runtime->nlev()" in adaptive diff --git a/tests/python/unit/output/test_async_scientific_output_diagnostics.py b/tests/python/unit/output/test_async_scientific_output_diagnostics.py index 6a1298553..7fb6b856a 100644 --- a/tests/python/unit/output/test_async_scientific_output_diagnostics.py +++ b/tests/python/unit/output/test_async_scientific_output_diagnostics.py @@ -30,6 +30,7 @@ from pops.output._restart_provider import RestartAuthority from pops.output._writers.common import writer_session_authority from pops.problem.handles import BlockHandle +from pops.runtime._runtime_consumers import RuntimeConsumerPublisher from pops.runtime._runtime_instance import RuntimeInstance from pops.time import Clock, every from tests.python.support.layout_plan import cartesian_grid @@ -282,6 +283,43 @@ def _accepted_balance_terms(self, route): } +def test_selected_native_balance_forwards_exact_owner_coordinates(): + class _SelectedExecutor: + def __init__(self): + self.call = None + + def _selected_accepted_balance_terms( + self, route, block, component, levels, automatic_terms + ): + self.call = (route, block, component, levels, automatic_terms) + return { + "storage_change": 7.0, + "outward_boundary_flux": 2.0, + "sources": 3.0, + "reflux": 1.0, + "projection": 0.5, + } + + executor = _SelectedExecutor() + terms = RuntimeConsumerPublisher._native_balance_terms( + executor, + "route", + block="fluid", + component=2, + levels=(0, 1), + automatic_terms=("projection", "reflux"), + ) + + assert executor.call == ( + "route", + "fluid", + 2, + [0, 1], + ["projection", "reflux"], + ) + assert terms.residual == pytest.approx(4.5) + + def _async_balance_runtime(tmp_path: Path): base = _install() mode = _scientific_output_mode(base.artifact) diff --git a/tests/python/unit/runtime/test_consumer_authoring.py b/tests/python/unit/runtime/test_consumer_authoring.py index c5d2ed660..73569e27f 100644 --- a/tests/python/unit/runtime/test_consumer_authoring.py +++ b/tests/python/unit/runtime/test_consumer_authoring.py @@ -293,6 +293,45 @@ def test_balance_consumer_resolves_one_exact_native_ledger_route(): assert contract.identity.domain == "balance-due-contract" +def test_balance_consumer_retains_native_term_selector_in_due_contract(): + case, block, state = _case() + clock = Clock("macro", owner=case.owner_path) + schedule = every(4, clock=clock) + ledger = BalanceLedger( + "mass-native", automatic_terms=("projection", "reflux") + ) + graph = ConsumerGraph.from_consumers(( + ScientificOutput( + format=ParaView(), + schedule=schedule, + fields=(state,), + diagnostics=(Balance(ledger, block=block),), + target="state/native-balance", + ), + )) + case.consumers(graph) + pops.validate(case) + subjects = case.layout_subjects() + layout = normalize_layout_plan( + Uniform(cartesian_grid(n=8)), + owner=case.owner_path.canonical(), + states=subjects.states, + fields=subjects.fields, + blocks=subjects.blocks, + handle_resolver=case.resolve, + ) + + resolved = graph.resolve(case.resolve, layout, owner=case.owner_path.canonical()) + quantity, = resolved.nodes[0].diagnostic_quantities + operation, = quantity.execution["operations"] + route = ledger.route_identity(case.resolve(block)) + contract = BalanceDueContract.from_consumer_graph(resolved) + + assert operation["automatic_terms"] == ("projection", "reflux") + assert operation["balance_component"] == 0 + assert contract.route(route.token).automatic_terms == ("projection", "reflux") + + def test_balance_consumer_refuses_a_schedule_that_can_fire_at_start(): case, block, state = _case() clock = Clock("macro", owner=case.owner_path) diff --git a/tests/python/unit/runtime/test_diagnostics_typed.py b/tests/python/unit/runtime/test_diagnostics_typed.py index a2670174d..b93c3ea36 100644 --- a/tests/python/unit/runtime/test_diagnostics_typed.py +++ b/tests/python/unit/runtime/test_diagnostics_typed.py @@ -126,6 +126,39 @@ def test_balance_uses_one_typed_native_attempt_route(): ConservationCheck(balance).diagnostic_execution() +def test_balance_ledger_selects_exact_native_component_terms(): + ledger = BalanceLedger( + "mass-native", + role=Density(), + automatic_terms=("projection", "reflux"), + ) + balance = Balance(ledger, block=_NE_BLOCK) + execution = balance.diagnostic_execution() + operation, = execution["operations"] + + assert execution["role"] == "Density" + assert operation["automatic_terms"] == ["projection", "reflux"] + assert operation["balance_component"] == 0 + assert balance.options()["role"] == "Density" + assert ledger.to_data()["role"] == "Density" + assert ledger.to_data()["component"] == 0 + assert ledger.to_data()["automatic_terms"] == ["projection", "reflux"] + assert ledger.identity != BalanceLedger("mass-native").identity + + with pytest.raises(TypeError, match="ComponentRole"): + BalanceLedger("bad-role", role="Density") + reordered = BalanceLedger( + "canonical-order", automatic_terms=("reflux", "projection") + ) + assert reordered.automatic_terms == ("projection", "reflux") + with pytest.raises(ValueError, match="must be unique"): + BalanceLedger("duplicate", automatic_terms=("reflux", "reflux")) + with pytest.raises(ValueError, match="only reflux and projection"): + BalanceLedger("bad-producer", automatic_terms=("sources",)) + with pytest.raises(TypeError, match="non-negative int"): + BalanceLedger("bad-component", component=-1, automatic_terms=("projection",)) + + # --- Integral / MinMax ------------------------------------------------------------------ def test_integral_is_a_sum_reduction(): mass = Integral(role=Density()) diff --git a/tests/python/unit/time/test_time_ops_polish.py b/tests/python/unit/time/test_time_ops_polish.py index b996b6950..788566df8 100644 --- a/tests/python/unit/time/test_time_ops_polish.py +++ b/tests/python/unit/time/test_time_ops_polish.py @@ -59,7 +59,7 @@ def t(): return time -def _balance_due_contract(route, *schedules): +def _balance_due_contract(route, *schedules, automatic_terms=()): return BalanceDueContract( make_identity("consumer-graph", {"test": "balance-due"}), ( @@ -72,6 +72,7 @@ def _balance_due_contract(route, *schedules): ) for index, schedule in enumerate(schedules) ), + automatic_terms, ), ), ) @@ -378,6 +379,65 @@ def test_record_balance_emits_exact_five_term_native_attempt_mailbox(t): assert "ctx.note_automatic_balance_capture_due(" not in unreachable_source +def test_record_balance_delegates_selected_native_terms_without_placeholders(t): + from pops.diagnostics import BalanceLedger + + P = t.Program("native-balance-terms") + U = typed_state(P, "blk") + total = P.sum(U) + ledger = BalanceLedger( + "mass-native", automatic_terms=("projection", "reflux") + ) + records = P.record_balance( + ledger, + storage_change=total, + outward_boundary_flux=total * 2.0, + sources=total * 3.0, + ) + route = ledger.route_identity(U.block) + assert tuple(record.attrs["term"] for record in records) == ( + "storage_change", + "outward_boundary_flux", + "sources", + ) + endpoint = typed_state(P, "blk", state_name="U").next + P.commit(endpoint, P.value("balance_next", U, at=endpoint.point)) + contract = _balance_due_contract( + route, + every(2, clock=P.clock), + automatic_terms=("projection", "reflux"), + ) + source = emit_cpp_program(P, balance_due_contract=contract) + assert source.count("ctx.record_balance_term(") == 3 + assert source.count("ctx.note_automatic_balance_capture_due(") == 1 + + P_bad = t.Program("duplicate-native-balance-term") + U_bad = typed_state(P_bad, "blk") + total_bad = P_bad.sum(U_bad) + with pytest.raises(ValueError, match="owned by.*native automatic producer"): + P_bad.record_balance( + ledger, + storage_change=total_bad, + outward_boundary_flux=total_bad, + sources=total_bad, + projection=total_bad, + ) + + component_ledger = BalanceLedger( + "component-one", + component=1, + automatic_terms=("projection",), + ) + with pytest.raises(ValueError, match="selects components.*component 1"): + P_bad.record_balance( + component_ledger, + storage_change=total_bad, + outward_boundary_flux=total_bad, + sources=total_bad, + reflux=total_bad, + ) + + def test_balance_due_contract_unions_consumers_and_ignores_static_false(t): from pops.diagnostics import BalanceLedger