feat(balance): capture metric projection evidence when due (ADC-686)

docs/design/exact-output-consumers.md

@@ -498,12 +498,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 route is explicit evidence, not automatic numerical instrumentation: a Program that cannot
-produce its actual reflux or projection increment cannot declare `Balance`. In particular, the
-generic automatic extraction of AMR reflux/projection contributions from the internal native
-operator ledgers remains separate work. On an adaptive layout the recorded values must already be
-composite and coverage-corrected; an ordinary sum of every per-level state would double-count
-covered coarse cells. Neither `Balance` nor `BalanceTerms` silently claims otherwise.
+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.
 
 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/program/amr_program_context.hpp

@@ -36,6 +36,7 @@
 #include <pops/numerics/time/amr/reflux/amr_flux_ledger.hpp>
 #include <pops/numerics/time/amr/reflux/amr_interface_flux_ledger.hpp>
 #include <pops/runtime/amr/amr_runtime.hpp>  // AmrRuntime (the engine the driver wraps)
+#include <pops/runtime/amr/composite_reduction.hpp>
 #include <pops/runtime/amr/hierarchy_tensor_solver_provider.hpp>
 #include <pops/runtime/context/grid_context.hpp>  // GridContext (per-level Schur assembly seam, ADC-633)
 #include <pops/runtime/amr_system.hpp>  // AmrSystem (the facade: params / block map / engine)
@@ -3164,6 +3165,45 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
   void program_execution_apply_projection_(int runtime_block, MultiFab& state) const {
     eng_->project_level_state(static_cast<std::size_t>(runtime_block), level_, state);
   }
+  std::optional<std::vector<Real>> program_execution_projection_balance_integrals_(
+      int program_block, const MultiFab& state) const {
+    const std::size_t runtime_block = static_cast<std::size_t>(sys_block(program_block));
+    if (level_ < 0 || level_ >= nlev())
+      throw std::out_of_range("AMR Program projection balance active level is out of range");
+    const MultiFab& live = eng_->level_state(runtime_block, level_);
+    if (state.box_array().boxes() != live.box_array().boxes() ||
+        state.dmap().ranks() != live.dmap().ranks() || state.ncomp() != live.ncomp() ||
+        state.n_grow() != live.n_grow() || state.local_size() != live.local_size())
+      throw std::invalid_argument(
+          "AMR Program projection balance candidate changed its exact level layout");
+
+    std::vector<pops::runtime::amr::composite_detail::CompositeLevelView> views;
+    views.reserve(static_cast<std::size_t>(nlev()));
+    for (int level = 0; level < nlev(); ++level) {
+      const Geometry geometry = eng_->level_geom(level);
+      const MultiFab* values = level == level_ ? &state : &eng_->level_state(runtime_block, level);
+      views.push_back({values, geometry.dx(), geometry.dy()});
+    }
+    const int next = level_ + 1 < nlev() ? level_ + 1 : -1;
+    MultiFab mask = pops::runtime::amr::composite_detail::active_mask(views, level_, next);
+    std::vector<double> result(static_cast<std::size_t>(state.ncomp()), 0.0);
+    for (int component = 0; component < state.ncomp(); ++component)
+      result[static_cast<std::size_t>(component)] =
+          static_cast<double>(pops::runtime::amr::composite_detail::local_sum(
+              state, mask, component, pops::runtime::amr::composite_detail::CompositeSumKind::Sum));
+    if (!eng_->level_is_replicated(level_))
+      all_reduce_sum_inplace(result.data(), result.size());
+    const Geometry geometry = eng_->level_geom(level_);
+    const double cell_measure =
+        static_cast<double>(geometry.dx()) * static_cast<double>(geometry.dy());
+    if (!std::isfinite(cell_measure) || cell_measure <= 0.0)
+      throw std::runtime_error(
+          "AMR Program projection balance requires a positive finite cell measure");
+    std::vector<Real> integrated(result.size(), Real(0));
+    for (std::size_t component = 0; component < result.size(); ++component)
+      integrated[component] = static_cast<Real>(cell_measure * result[component]);
+    return integrated;
+  }
   Real program_execution_hmin_() const { return eng_->level_hmin(level_); }
   Real program_execution_max_wave_speed_(int runtime_block, const MultiFab& state) const {
     return eng_->level_max_speed(static_cast<std::size_t>(runtime_block), level_, state);
@@ -3273,8 +3313,8 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
       const HistoryRegistration& registration) const {
     return pops::detail::AmrHistoryOps::initialized(*eng_, registration.name);
   }
-  double program_execution_history_slot_dt_storage_(
-      const HistoryRegistration& registration, int lag) const {
+  double program_execution_history_slot_dt_storage_(const HistoryRegistration& registration,
+                                                    int lag) const {
     return pops::detail::AmrHistoryOps::slot_dt(*eng_, registration.name, lag);
   }
   void program_execution_set_history_initialized_storage_(const HistoryRegistration& registration,

include/pops/runtime/program/program_context.hpp

@@ -12,6 +12,7 @@
 #include <limits>
 #include <memory>
 #include <map>
+#include <optional>
 #include <stdexcept>
 #include <string>
 #include <string_view>
@@ -660,6 +661,29 @@ class ProgramContext : public ProgramExecutionServices<ProgramContext> {
   void program_execution_apply_projection_(int runtime_block, MultiFab& state) const {
     sys_->block_project(runtime_block, state);
   }
+  std::optional<std::vector<Real>> program_execution_projection_balance_integrals_(
+      int program_block, const MultiFab& state) const {
+    // The public polar diagnostic path has no exact per-cell volume provider yet. Keep automatic
+    // evidence absent instead of relabelling Cartesian dx*dy as a polar measure; authored balance
+    // terms remain available and the future selector must fail closed on this missing producer.
+    if (sys_->program_is_polar())
+      return std::nullopt;
+    const GridContext context = program_execution_block_grid_context_(program_block);
+    const Real cell_measure = context.geom.dx() * context.geom.dy();
+    if (!std::isfinite(static_cast<double>(cell_measure)) || cell_measure <= Real(0))
+      throw std::runtime_error(
+          "Uniform Program projection balance requires a positive finite cell measure");
+    RelativeCellMeasure measure;
+    if (context.domain_mask != nullptr) {
+      measure.active_cells = context.domain_mask;
+      measure.inverse_volume_fraction = context.eb_inverse_volume_fraction;
+    }
+    std::vector<Real> result(static_cast<std::size_t>(state.ncomp()), Real(0));
+    for (int component = 0; component < state.ncomp(); ++component)
+      result[static_cast<std::size_t>(component)] =
+          cell_measure * pops::reduce_sum(state, component, measure);
+    return result;
+  }
   Real program_execution_hmin_() const { return sys_->cfl_min_dx(); }
   Real program_execution_max_wave_speed_(int runtime_block, const MultiFab& state) const {
     return sys_->block_max_speed(runtime_block, state);

include/pops/runtime/program/program_execution_services.hpp

@@ -502,9 +502,33 @@ class ProgramExecutionServices {
   /// Project one candidate state through the exact authored block closure.
   ///
   /// Program-to-runtime block qualification is topology-independent. The provider owns only the
-  /// Uniform or level-qualified native projection call.
+  /// Uniform or level-qualified native projection call. When a generated Balance route is due, the
+  /// provider also supplies exact metric-integrated component values before and after projection;
+  /// their signed delta stays qualified by runtime block/level/component in the attempt mailbox.
   void apply_projection(int block, MultiFab& state) const {
-    provider_().program_execution_apply_projection_(sys_block(block), state);
+    const int runtime_block = sys_block(block);
+    ProgramRuntimeState& runtime = program_runtime_state_();
+    if (!runtime.automatic_balance_capture_due()) {
+      provider_().program_execution_apply_projection_(runtime_block, state);
+      return;
+    }
+    const std::optional<std::vector<Real>> before =
+        provider_().program_execution_projection_balance_integrals_(block, state);
+    provider_().program_execution_apply_projection_(runtime_block, state);
+    if (!before)
+      return;
+    const std::optional<std::vector<Real>> after =
+        provider_().program_execution_projection_balance_integrals_(block, state);
+    if (!after || before->size() != after->size() ||
+        before->size() != static_cast<std::size_t>(state.ncomp()))
+      throw std::runtime_error(
+          "Program projection balance provider changed its conservative component width");
+    const int level = program_resource_field_level();
+    for (int component = 0; component < state.ncomp(); ++component)
+      runtime.record_automatic_balance_term(runtime_block, level, component, "projection",
+                                            (*after)[static_cast<std::size_t>(component)] -
+                                                (*before)[static_cast<std::size_t>(component)],
+                                            "ProgramExecutionServices");
   }
 
   /// Minimum physical cell size used by the native CFL authority.
@@ -1397,6 +1421,10 @@ class ProgramExecutionServices {
     provider_().program_execution_record_balance_term_(route, term, value);
   }
 
+  void note_automatic_balance_capture_due(bool due) const {
+    program_runtime_state_().note_automatic_balance_capture_due(due, "ProgramExecutionServices");
+  }
+
   void note_step_projection(const std::string& name) const {
     program_runtime_state_().note_step_projection(name);
   }

include/pops/runtime/program/program_runtime_state.hpp

@@ -300,6 +300,11 @@ struct ProgramRuntimeState {
   /// 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_;
+  /// Monotone attempt-local decision emitted by generated code before any Program operator runs.
+  /// It is the OR of the exact ConsumerGraph-derived route decisions for this public step. Keeping
+  /// this separate from step_balance_terms_ lets projection operators execute before their later
+  /// Program.record_balance sinks without losing due automatic evidence.
+  bool automatic_balance_due_ = false;
   /// 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`.
@@ -852,13 +857,30 @@ struct ProgramRuntimeState {
       entry->second += value;
   }
 
-  /// Whether a compiled Program has actually emitted a due Balance route in this attempt.
+  /// Whether generated code proved that at least one Balance route is due 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.
+  /// The exact ConsumerGraph-derived decision is emitted before any Program operator, so both an
+  /// in-body projection and post-body reflux observe the same cadence without a second scheduler.
   [[nodiscard]] bool automatic_balance_capture_due() const noexcept {
-    return !balance_replay_active_ && !step_balance_terms_.empty();
+    return !balance_replay_active_ && automatic_balance_due_;
+  }
+
+  /// Publish one generated ConsumerGraph due decision before Program operators execute.
+  ///
+  /// Several compiled Program invocations may share one outer accepted-step window. The marker is
+  /// therefore monotone inside an attempt and is reset only at attempt entry. Static-false routes
+  /// emit no call, so a run without Balance consumers retains no generated hot-path branch.
+  void note_automatic_balance_capture_due(bool due, const std::string& runtime) {
+    if (balance_replay_active_) {
+      if (due)
+        throw std::logic_error(runtime +
+                               "::note_automatic_balance_capture_due cannot enable replay capture");
+      return;
+    }
+    if (!balance_due_window_active_)
+      throw std::logic_error(
+          runtime + "::note_automatic_balance_capture_due requires an active public-step window");
+    automatic_balance_due_ = automatic_balance_due_ || due;
   }
 
   /// Accumulate one signed, metric-integrated native operator contribution.
@@ -904,6 +926,7 @@ struct ProgramRuntimeState {
     step_projections_.clear();
     step_balance_terms_.clear();
     automatic_balance_terms_.clear();
+    automatic_balance_due_ = false;
     balance_due_window_active_ = false;
     balance_due_target_step_ = 0;
     balance_step_completed_ = false;

python/pops/codegen/program_balance_due.py

@@ -212,6 +212,7 @@ def emit_balance_due_guards(
     if type(lowering) is not BalanceDueLowering:
         raise TypeError("balance due guard emission requires BalanceDueLowering")
     contract = json.dumps(lowering.contract.token)
+    automatic_tokens = []
     for index, (route, periods) in enumerate(sorted(lowering.route_periods.items())):
         if not periods:
             token = "false"
@@ -223,7 +224,13 @@ def emit_balance_due_guards(
             ]
             token = "balance_due_%d" % index
             lines.append("const bool %s = (%s);" % (token, " || ".join(calls)))
+            automatic_tokens.append(token)
         var[("balance_due_route", route)] = token
+    if automatic_tokens:
+        lines.append(
+            "ctx.note_automatic_balance_capture_due(%s);"
+            % (" || ".join(automatic_tokens))
+        )
     var[("balance_guarded_values",)] = lowering.guarded_values
     var[("balance_record_routes",)] = lowering.record_routes
 

src/runtime/amr/amr_system.cpp

@@ -463,6 +463,7 @@ struct AmrSystem::Impl {
     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 automatic_balance_due = false;
     bool balance_step_completed = false;
     bool balance_program_was_due = false;
     pops::runtime::program::CacheManager cache;
@@ -511,6 +512,7 @@ struct AmrSystem::Impl {
       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_);
+      automatic_balance_due = impl.program_.automatic_balance_due_;
       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
@@ -545,6 +547,7 @@ struct AmrSystem::Impl {
       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_.automatic_balance_due_ = automatic_balance_due;
       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

@@ -618,6 +618,7 @@ struct System::Impl {
     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 automatic_balance_due;
     bool balance_step_completed;
     bool balance_program_was_due;
     pops::runtime::program::CacheManager cache;
@@ -645,6 +646,7 @@ struct System::Impl {
           program_diagnostics(impl.program_.diagnostics_),
           step_balance_terms(impl.program_.step_balance_terms_),
           automatic_balance_terms(impl.program_.automatic_balance_terms_),
+          automatic_balance_due(impl.program_.automatic_balance_due_),
           balance_step_completed(impl.program_.balance_step_completed_),
           balance_program_was_due(impl.program_.balance_program_was_due_),
           cache(impl.program_.cache_),
@@ -680,6 +682,7 @@ struct System::Impl {
       impl.program_.diagnostics_ = program_diagnostics;
       impl.program_.step_balance_terms_ = step_balance_terms;
       impl.program_.automatic_balance_terms_ = automatic_balance_terms;
+      impl.program_.automatic_balance_due_ = automatic_balance_due;
       impl.program_.balance_step_completed_ = balance_step_completed;
       impl.program_.balance_program_was_due_ = balance_program_was_due;
       impl.program_.cache_ = cache;

tests/cpp/integration/runtime/test_program_runtime.cpp

@@ -139,6 +139,31 @@ TEST(ProgramRuntime, BalanceDueWindowUsesTheOuterAcceptedStepAndCleansUpOnFailur
   EXPECT_THROW((void)state.balance_consumer_is_due(contract, route, 4, "test"), std::logic_error);
 }
 
+TEST(ProgramRuntime, AutomaticBalanceDueMarkerIsAttemptLocalMonotoneAndReplaySafe) {
+  runtime::program::ProgramRuntimeState state;
+
+  EXPECT_FALSE(state.automatic_balance_capture_due());
+  EXPECT_THROW(state.note_automatic_balance_capture_due(true, "test"), std::logic_error);
+  state.run_balance_due_window(0, "test", [&] {
+    state.note_automatic_balance_capture_due(false, "test");
+    EXPECT_FALSE(state.automatic_balance_capture_due());
+    state.note_automatic_balance_capture_due(true, "test");
+    EXPECT_TRUE(state.automatic_balance_capture_due());
+    state.note_automatic_balance_capture_due(false, "test");
+    EXPECT_TRUE(state.automatic_balance_capture_due());
+  });
+  EXPECT_TRUE(state.automatic_balance_capture_due());
+
+  state.begin_step_projection_report();
+  EXPECT_FALSE(state.automatic_balance_capture_due());
+  state.run_balance_replay("test", [&] {
+    state.note_automatic_balance_capture_due(false, "test");
+    EXPECT_FALSE(state.automatic_balance_capture_due());
+    EXPECT_THROW(state.note_automatic_balance_capture_due(true, "test"), std::logic_error);
+  });
+  EXPECT_FALSE(state.automatic_balance_capture_due());
+}
+
 TEST(ProgramRuntime, SelectiveReplayCompilesBalanceOffAndRestoresTheGuard) {
   runtime::program::ProgramRuntimeState state;
   const std::string contract = "pops.balance-due-contract.v1:sha256:" + std::string(64, '3');