ADC-686: prove all five nonzero Balance terms over MPI

include/pops/numerics/time/amr/levels/amr_patch_range.hpp

@@ -468,6 +468,23 @@ struct FluxRegister {
     device_fence();
     all_reduce_sum_inplace(buf.data(), buf.size(), communicator);
   }
+  /// Sum the already-gathered sparse correction by conservative component.
+  ///
+  /// RefluxStorage is pinned host storage shared with device kernels. The fence makes the gathered
+  /// register host-readable; every communicator rank then traverses the same compact global order,
+  /// so this adds no second collective and produces the exact state increment applied below.
+  [[nodiscard]] std::vector<Real> component_sums(Real cell_measure) const {
+    if (!std::isfinite(static_cast<double>(cell_measure)) || cell_measure <= Real(0))
+      throw std::invalid_argument(
+          "FluxRegister component sum requires a finite positive cell measure");
+    device_fence();
+    std::vector<Real> result(static_cast<std::size_t>(nc), Real(0));
+    const std::size_t components = static_cast<std::size_t>(nc);
+    for (std::size_t offset = 0; offset < buf.size(); offset += components)
+      for (std::size_t component = 0; component < components; ++component)
+        result[component] += cell_measure * buf[offset + component];
+    return result;
+  }
   [[nodiscard]] std::size_t lookup_capacity() const noexcept { return cell_lookup.capacity(); }
   [[nodiscard]] std::size_t covered_cell_count() const noexcept { return cell_lookup.size(); }
 

include/pops/numerics/time/amr/levels/amr_subcycling.hpp

@@ -824,7 +824,8 @@ class PreparedAmrProgramRefluxTransition {
   template <class CoarseStripRange, class FineStripRange>
   void synchronize_integrated(MultiFab& parent_state, Real dx, Real dy,
                               const CoarseStripRange& coarse_role, const FineStripRange& fine_role,
-                              const CommunicatorView& communicator) {
+                              const CommunicatorView& communicator,
+                              std::vector<Real>* integrated_state_correction = nullptr) {
     validate_communicator_(communicator);
     using CoarseStrip = typename CoarseStripRange::value_type;
     using FineStrip = typename FineStripRange::value_type;
@@ -882,6 +883,8 @@ class PreparedAmrProgramRefluxTransition {
                                                               ncomp_);
       }
       correction_.gather(communicator);
+      if (integrated_state_correction != nullptr)
+        *integrated_state_correction = correction_.component_sums(dx * dy);
       for (int local_parent = 0; local_parent < parent_state.local_size(); ++local_parent)
         for_each_cell(parent_state.box(local_parent),
                       detail::ApplyRefluxRegisterKernel{parent_state.fab(local_parent).array(),

include/pops/runtime/amr/amr_program_reflux.hpp

@@ -525,14 +525,19 @@ inline void sample_fine_role_strip(const MultiFab& state, const MultiFab& Fx, co
 /// per (cell,direction) (ADC-636 ownership: each C/F face is owned by the rank holding the covering fine
 /// patch), so the gather is associativity-free -> distributed == replicated bit-for-bit.
 inline void route_reflux_program(AmrRuntime& eng, std::size_t b, int k, const EdgeFlux& coarse_role,
-                                 const EdgeFlux& fine_role) {
+                                 const EdgeFlux& fine_role,
+                                 std::vector<Real>* integrated_state_correction = nullptr) {
   MultiFab& Uc = eng.level_state(b, k - 1);  // the PARENT (coarse) live state we correct
   const BoxArray child_ba = eng.level_state(b, k).box_array();  // GLOBAL level-k patches
-  if (child_ba.size() == 0)
+  if (child_ba.size() == 0) {
+    if (integrated_state_correction != nullptr)
+      integrated_state_correction->assign(static_cast<std::size_t>(Uc.ncomp()), Real(0));
     return;
+  }
   const Geometry gc = eng.level_geom(k - 1);
   eng.prepared_reflux_transition(b, k).synchronize_integrated(
-      Uc, gc.dx(), gc.dy(), coarse_role.coarse, fine_role.fine, world_communicator_view());
+      Uc, gc.dx(), gc.dy(), coarse_role.coarse, fine_role.fine, world_communicator_view(),
+      integrated_state_correction);
 }
 
 }  // namespace detail

include/pops/runtime/program/amr_program_context.hpp

@@ -1115,16 +1115,35 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
     amr::ClockStamp sync_clock = accepted;
     sync_clock.level = parent;
     for (int b = 0; b < n_blocks(); ++b) {
-      const std::size_t sb = static_cast<std::size_t>(sys_block(b));
+      const int runtime_block = sys_block(b);
+      const std::size_t sb = static_cast<std::size_t>(runtime_block);
       if (capturing()) {
         sync_report_.push_back({parent, child, b, SyncPhase::Reflux, sync_clock});
         const EdgeFlux coarse_role = reflux_flux_from_ledger_(b, parent, ledger_begin, ledger_end);
         const EdgeFlux fine_role = reflux_flux_from_ledger_(b, child, ledger_begin, ledger_end);
         if (coarse_role.empty() != fine_role.empty())
           throw std::runtime_error(
               "AMR conservative ledger contains only one side of a parent/child flux pair");
-        if (!coarse_role.empty())
-          pops::detail::route_reflux_program(*eng_, sb, child, coarse_role, fine_role);
+        const bool capture_balance =
+            facade_->program_runtime_state_().automatic_balance_capture_due();
+        std::vector<Real> integrated_reflux;
+        if (!coarse_role.empty()) {
+          pops::detail::route_reflux_program(*eng_, sb, child, coarse_role, fine_role,
+                                             capture_balance ? &integrated_reflux : nullptr);
+        } else if (capture_balance) {
+          integrated_reflux.assign(static_cast<std::size_t>(eng_->level_state(sb, parent).ncomp()),
+                                   Real(0));
+        }
+        if (capture_balance) {
+          const int components = eng_->level_state(sb, parent).ncomp();
+          if (integrated_reflux.size() != static_cast<std::size_t>(components))
+            throw std::runtime_error(
+                "AMR automatic reflux balance contribution changed component width");
+          for (int component = 0; component < components; ++component)
+            facade_->program_runtime_state_().record_automatic_balance_term(
+                runtime_block, parent, component, "reflux",
+                integrated_reflux[static_cast<std::size_t>(component)], "AmrProgramContext");
+        }
       }
       sync_report_.push_back({parent, child, b, SyncPhase::AverageDown, sync_clock});
       eng_->average_down_level(sb, child);

include/pops/runtime/program/program_runtime_state.hpp

@@ -143,6 +143,29 @@ struct HistoryManager {
   }
 };
 
+/// Attempt-local native balance evidence emitted by one exact runtime operator.
+///
+/// The coordinate deliberately remains independent of a user-facing BalanceLedger route: native
+/// operators know their qualified runtime block, hierarchy level and conservative component, while
+/// the route-to-quantity selector is a separate planning authority. Keeping both identities
+/// separate prevents a reflux correction from being silently relabelled as a complete balance.
+struct AutomaticBalanceKey {
+  int runtime_block = -1;
+  int level = -1;
+  int component = -1;
+  std::string term;
+
+  friend bool operator<(const AutomaticBalanceKey& left, const AutomaticBalanceKey& right) {
+    if (left.runtime_block != right.runtime_block)
+      return left.runtime_block < right.runtime_block;
+    if (left.level != right.level)
+      return left.level < right.level;
+    if (left.component != right.component)
+      return left.component < right.component;
+    return left.term < right.term;
+  }
+};
+
 /// The compiled time-Program runtime state, extracted from the System / AmrSystem god-object (ADC-594).
 ///
 /// A plain aggregate: the owning Impl embeds ONE instance and routes every Program seam through it. The
@@ -272,6 +295,11 @@ struct ProgramRuntimeState {
   /// consumers read it while the facade's outer transaction still retains U^n, so a missing term
   /// cannot silently reuse the preceding step.
   std::map<std::string, Real> step_balance_terms_;
+  /// Native operator contributions captured only for a due Balance attempt. These values are keyed
+  /// by their physical runtime coordinate instead of a user ledger route and are therefore not read
+  /// by accepted_balance_terms(). The owning facade snapshots this map with the rest of the attempt,
+  /// so rejection cannot leak automatic evidence into a retry.
+  std::map<AutomaticBalanceKey, Real> automatic_balance_terms_;
   /// Attempt-local outer accepted-step target used by ConsumerGraph-fused balance guards. Program
   /// substeps temporarily publish their window-start macro step through the facade, so generated
   /// balance code must not infer the public target from `macro_step()+1`.
@@ -791,6 +819,14 @@ struct ProgramRuntimeState {
       throw std::invalid_argument(runtime + " requires one canonical five-term balance name");
   }
 
+  static void require_automatic_balance_term(const std::string& term, const std::string& runtime) {
+    static constexpr std::array<std::string_view, 4> kTerms{"outward_boundary_flux", "sources",
+                                                            "reflux", "projection"};
+    if (std::find(kTerms.begin(), kTerms.end(), std::string_view(term)) == kTerms.end())
+      throw std::invalid_argument(runtime +
+                                  " requires one native operator balance contribution name");
+  }
+
   /// Record a compiled-Program scalar. Ordinary P.record_scalar names remain inspectable after the
   /// step with last-write-wins semantics. The balance namespace has a separate typed sink.
   void record_diagnostic(const std::string& name, Real value) {
@@ -816,6 +852,40 @@ struct ProgramRuntimeState {
       entry->second += value;
   }
 
+  /// Whether a compiled Program has actually emitted a due Balance route in this attempt.
+  ///
+  /// Generated balance records are cadence-guarded before their reductions. Reflux executes after
+  /// the Program body, so observing a non-empty authored mailbox here avoids every extra native
+  /// reduction on an off-cadence or replay step without introducing a second scheduler.
+  [[nodiscard]] bool automatic_balance_capture_due() const noexcept {
+    return !balance_replay_active_ && !step_balance_terms_.empty();
+  }
+
+  /// Accumulate one signed, metric-integrated native operator contribution.
+  ///
+  /// This is intentionally not accepted_balance_terms(): automatic evidence remains qualified by
+  /// block/level/component until a resolved quantity selector proves which BalanceLedger route owns
+  /// it. The separation is fail-closed and lets boundary/source/projection producers join the same
+  /// mailbox later without fabricating missing terms.
+  void record_automatic_balance_term(int runtime_block, int level, int component,
+                                     const std::string& term, Real value,
+                                     const std::string& runtime) {
+    if (!automatic_balance_capture_due())
+      throw std::logic_error(runtime +
+                             "::record_automatic_balance_term requires a due authored balance");
+    if (runtime_block < 0 || level < 0 || component < 0)
+      throw std::invalid_argument(
+          runtime + "::record_automatic_balance_term requires non-negative coordinates");
+    require_automatic_balance_term(term, runtime + "::record_automatic_balance_term");
+    if (!std::isfinite(static_cast<double>(value)))
+      throw std::invalid_argument(runtime +
+                                  "::record_automatic_balance_term requires a finite value");
+    auto [entry, inserted] = automatic_balance_terms_.try_emplace(
+        AutomaticBalanceKey{runtime_block, level, component, term}, value);
+    if (!inserted)
+      entry->second += value;
+  }
+
   /// Read the named diagnostic, FAIL-LOUD if the Program never recorded it. @p runtime names the
   /// Program subsystem setter in the message (not a generic getter). @throws std::out_of_range.
   Real diagnostic(const std::string& name, const std::string& runtime) const {
@@ -833,6 +903,7 @@ struct ProgramRuntimeState {
   void begin_step_projection_report() {
     step_projections_.clear();
     step_balance_terms_.clear();
+    automatic_balance_terms_.clear();
     balance_due_window_active_ = false;
     balance_due_target_step_ = 0;
     balance_step_completed_ = false;

src/runtime/amr/amr_system.cpp

@@ -462,6 +462,7 @@ struct AmrSystem::Impl {
     int cadence_clock_restore_macro_step = 0;
     std::map<std::string, Real> program_diagnostics;
     std::map<std::string, Real> step_balance_terms;
+    std::map<pops::runtime::program::AutomaticBalanceKey, Real> automatic_balance_terms;
     bool balance_step_completed = false;
     bool balance_program_was_due = false;
     pops::runtime::program::CacheManager cache;
@@ -509,6 +510,7 @@ struct AmrSystem::Impl {
       cadence_clock_restore_macro_step = impl.program_.cadence_clock_restore_macro_step_;
       copy_value_map_into(program_diagnostics, impl.program_.diagnostics_);
       copy_value_map_into(step_balance_terms, impl.program_.step_balance_terms_);
+      copy_value_map_into(automatic_balance_terms, impl.program_.automatic_balance_terms_);
       balance_step_completed = impl.program_.balance_step_completed_;
       balance_program_was_due = impl.program_.balance_program_was_due_;
       // AMR currently owns its native cache/history rings inside AmrRuntime.  These two shared
@@ -542,6 +544,7 @@ struct AmrSystem::Impl {
       impl.program_.cadence_clock_restore_macro_step_ = cadence_clock_restore_macro_step;
       copy_value_map_into(impl.program_.diagnostics_, program_diagnostics);
       copy_value_map_into(impl.program_.step_balance_terms_, step_balance_terms);
+      copy_value_map_into(impl.program_.automatic_balance_terms_, automatic_balance_terms);
       impl.program_.balance_step_completed_ = balance_step_completed;
       impl.program_.balance_program_was_due_ = balance_program_was_due;
       impl.program_.cache_ = cache;

src/runtime/system/system_impl.hpp

@@ -617,6 +617,7 @@ struct System::Impl {
     int cadence_clock_restore_macro_step;
     std::map<std::string, Real> program_diagnostics;
     std::map<std::string, Real> step_balance_terms;
+    std::map<pops::runtime::program::AutomaticBalanceKey, Real> automatic_balance_terms;
     bool balance_step_completed;
     bool balance_program_was_due;
     pops::runtime::program::CacheManager cache;
@@ -643,6 +644,7 @@ struct System::Impl {
           cadence_clock_restore_macro_step(impl.program_.cadence_clock_restore_macro_step_),
           program_diagnostics(impl.program_.diagnostics_),
           step_balance_terms(impl.program_.step_balance_terms_),
+          automatic_balance_terms(impl.program_.automatic_balance_terms_),
           balance_step_completed(impl.program_.balance_step_completed_),
           balance_program_was_due(impl.program_.balance_program_was_due_),
           cache(impl.program_.cache_),
@@ -677,6 +679,7 @@ struct System::Impl {
       impl.program_.cadence_clock_restore_macro_step_ = cadence_clock_restore_macro_step;
       impl.program_.diagnostics_ = program_diagnostics;
       impl.program_.step_balance_terms_ = step_balance_terms;
+      impl.program_.automatic_balance_terms_ = automatic_balance_terms;
       impl.program_.balance_step_completed_ = balance_step_completed;
       impl.program_.balance_program_was_due_ = balance_program_was_due;
       impl.program_.cache_ = cache;

tests/python/architecture/test_automatic_reflux_balance_fence.py

@@ -0,0 +1,99 @@
+"""ADC-686: automatic reflux evidence stays exact, sparse, and fail-closed."""
+
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[3]
+PROGRAM_STATE = (
+    ROOT / "include" / "pops" / "runtime" / "program" / "program_runtime_state.hpp"
+)
+AMR_CONTEXT = (
+    ROOT / "include" / "pops" / "runtime" / "program" / "amr_program_context.hpp"
+)
+AMR_REFLUX = ROOT / "include" / "pops" / "runtime" / "amr" / "amr_program_reflux.hpp"
+AMR_SUBCYCLING = (
+    ROOT / "include" / "pops" / "numerics" / "time" / "amr" / "levels"
+    / "amr_subcycling.hpp"
+)
+AMR_PATCH_RANGE = (
+    ROOT / "include" / "pops" / "numerics" / "time" / "amr" / "levels"
+    / "amr_patch_range.hpp"
+)
+UNIFORM_IMPL = ROOT / "src" / "runtime" / "system" / "system_impl.hpp"
+AMR_IMPL = ROOT / "src" / "runtime" / "amr" / "amr_system.cpp"
+
+
+def _between(text: str, begin: str, end: str) -> str:
+    return text.split(begin, 1)[1].split(end, 1)[0]
+
+
+def test_automatic_balance_mailbox_is_attempt_local_and_not_a_route_fallback() -> None:
+    state = PROGRAM_STATE.read_text()
+    assert "struct AutomaticBalanceKey" in state
+    assert "std::map<AutomaticBalanceKey, Real> automatic_balance_terms_;" in state
+    assert "automatic_balance_terms_.clear();" in state
+    assert "record_automatic_balance_term(" in state
+    assert "automatic_balance_capture_due()" in state
+
+    accepted = _between(
+        state,
+        "std::map<std::string, Real> accepted_balance_terms(",
+        "void begin_balance_due_window(",
+    )
+    assert "step_balance_terms_" in accepted
+    assert "automatic_balance_terms_" not in accepted
+
+    uniform = UNIFORM_IMPL.read_text()
+    adaptive = AMR_IMPL.read_text()
+    for source in (uniform, adaptive):
+        assert "automatic_balance_terms" in source
+        assert "impl.program_.automatic_balance_terms_" in source
+
+
+def test_reflux_integral_comes_from_the_gathered_sparse_correction() -> None:
+    register = AMR_PATCH_RANGE.read_text()
+    component_sums = _between(
+        register,
+        "[[nodiscard]] std::vector<Real> component_sums(",
+        "[[nodiscard]] std::size_t lookup_capacity()",
+    )
+    assert "device_fence();" in component_sums
+    assert "cell_measure * buf[offset + component]" in component_sums
+    assert "all_reduce" not in component_sums
+
+    transition = AMR_SUBCYCLING.read_text()
+    synchronize = _between(
+        transition,
+        "void synchronize_integrated(",
+        "\n private:",
+    )
+    assert synchronize.index("correction_.gather(communicator);") < synchronize.index(
+        "correction_.component_sums(dx * dy)"
+    )
+    assert synchronize.index("correction_.component_sums(dx * dy)") < synchronize.index(
+        "ApplyRefluxRegisterKernel"
+    )
+
+    route = AMR_REFLUX.read_text()
+    routing = _between(route, "inline void route_reflux_program(", "\n}\n\n}  // namespace detail")
+    assert "std::vector<Real>* integrated_state_correction = nullptr" in routing
+    assert "integrated_state_correction);" in routing
+
+
+def test_amr_records_reflux_before_average_down_only_when_balance_is_due() -> None:
+    context = AMR_CONTEXT.read_text()
+    synchronize = _between(
+        context,
+        "void synchronize_level_pair_(",
+        "void finalize_history_rotation_()",
+    )
+    assert "automatic_balance_capture_due()" in synchronize
+    assert "record_automatic_balance_term(" in synchronize
+    assert '"reflux"' in synchronize
+    assert "reduce_sum(" not in synchronize
+    assert synchronize.index("route_reflux_program(") < synchronize.index(
+        "record_automatic_balance_term("
+    )
+    assert synchronize.index("record_automatic_balance_term(") < synchronize.index(
+        "SyncPhase::AverageDown"
+    )