feat(balance): resolve qualified native balance evidence (ADC-686)

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

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<std::string, double> accepted_balance_terms(const std::string& route) const;
+  /// The same accepted route with selected attempt-local native reflux/projection producers.
+  POPS_EXPORT std::map<std::string, double> selected_accepted_balance_terms(
+      const std::string& route, const std::string& block, int component,
+      const std::vector<int>& levels, const std::vector<std::string>& automatic_terms) const;
   POPS_EXPORT void begin_step_projection_report();
   POPS_EXPORT void note_step_projection(const std::string& name);
   POPS_EXPORT std::vector<std::string> consume_step_projections();

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<std::string, Real> selected_accepted_balance_terms(
+      const std::string& route, int runtime_block, int component, const std::vector<int>& levels,
+      const std::vector<std::string>& automatic_terms, const std::string& runtime) const {
+    static constexpr std::array<const char*, 5> 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<std::string, Real> 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<double>(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<double>(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<double>(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");

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<std::string, Real> accepted_balance_terms(const std::string& route) const;
+  /// The same accepted route with selected attempt-local native reflux/projection producers.
+  POPS_EXPORT std::map<std::string, Real> selected_accepted_balance_terms(
+      const std::string& route, const std::string& block, int component,
+      const std::vector<int>& levels, const std::vector<std::string>& automatic_terms) const;
   POPS_EXPORT void begin_step_projection_report();
   POPS_EXPORT void note_step_projection(const std::string& name);
   POPS_EXPORT std::vector<std::string> consume_step_projections();

python/bindings/core/init/init_amr.cpp

@@ -824,6 +824,9 @@ void bind_amr_program(py::class_<AmrSystem>& 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"))

python/bindings/core/init/init_system.cpp

@@ -355,6 +355,9 @@ void bind_system_program(py::class_<System>& 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

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