From fb02634689d99f0365674f880cfa6d8866bbcc36 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 03:59:59 +0200 Subject: [PATCH 1/4] test(mpi): prove exact nonzero balance terms --- .../mpi/test_async_balance_cadence_mpi.py | 98 ++++++++++++++----- 1 file changed, 74 insertions(+), 24 deletions(-) diff --git a/tests/python/integration/mpi/test_async_balance_cadence_mpi.py b/tests/python/integration/mpi/test_async_balance_cadence_mpi.py index 40ce1a255..0ba380290 100644 --- a/tests/python/integration/mpi/test_async_balance_cadence_mpi.py +++ b/tests/python/integration/mpi/test_async_balance_cadence_mpi.py @@ -5,8 +5,11 @@ ``Case -> Program.cadence -> compile -> mpi_world -> bind -> run`` route. The Program closes one stride-3 window every third accepted macro-step, while async Balance consumers fire every two and three accepted steps. Held windows must therefore publish exact zero ledgers and due windows must -publish native nonzero ledgers. A separate every-step async field series proves that each worker -receives the accepted field image captured on its own tick, never the latest native state. +publish an exact signed five-term ledger built from five real collective Program reductions. The +fixture explicitly authors that accounting split; it proves transport, signs, residual closure and +rank agreement, not automatic extraction of AMR reflux or projection terms. A separate every-step +async field series proves that each worker receives the accepted field image captured on its own +tick, never the latest native state. """ from __future__ import annotations @@ -163,22 +166,35 @@ def _authored_case(*, adaptive: bool) -> tuple[pops.Case, Any]: case.numerics(numerics, block=block) program = pops.Program("async-balance-%s-program" % label) temporal = program.state(evolved) - total = program.sum(temporal.n) - zero = total * 0.0 - ledger = BalanceLedger("accepted-mass") - program.record_balance( - ledger, - storage_change=total, - outward_boundary_flux=zero, - sources=zero, - reflux=zero, - projection=zero, - ) accepted = program.value( "accepted_growth", temporal.n + program.dt * Fraction(1, 2) * temporal.n, at=temporal.next.point, ) + increment = program.value( + "accepted_increment", + accepted - temporal.n, + at=temporal.next.point, + ) + # This is an explicitly authored accounting fixture, not an automatic AMR-term extractor. + # Every term owns a real native Program.sum so the installed mpiexec route enters five + # collectives. The signed split closes the actual accepted storage increment exactly: + # storage + outward - sources - reflux - projection + # = q - q - q - q - (-2q) = 0. + storage_change = program.sum(increment) + outward_boundary_flux = -program.sum(increment) + sources = program.sum(increment) + reflux = program.sum(increment) + projection = -2.0 * program.sum(increment) + ledger = BalanceLedger("accepted-mass") + program.record_balance( + ledger, + storage_change=storage_change, + outward_boundary_flux=outward_boundary_flux, + sources=sources, + reflux=reflux, + projection=projection, + ) program.commit(temporal.next, accepted) program.cadence(stride=3) program.step_strategy(FixedDt(DT)) @@ -310,6 +326,34 @@ def _balance(reopened: Any) -> tuple[float, dict[str, float]]: ) +def _require_exact_signed_balance( + label: str, + step: int, + value: float, + terms: dict[str, float], +) -> None: + q = terms["storage_change"] + expected = { + "storage_change": q, + "outward_boundary_flux": -q, + "sources": q, + "reflux": q, + "projection": -2.0 * q, + } + residual = ( + terms["storage_change"] + + terms["outward_boundary_flux"] + - terms["sources"] + - terms["reflux"] + - terms["projection"] + ) + if q <= 0.0 or terms != expected or residual != 0.0 or value != residual: + raise AssertionError( + "%s due step %d did not preserve its exact signed five-term Balance: " + "value=%r terms=%r" % (label, step, value, terms) + ) + + def _verify(root: Path, *, adaptive: bool) -> None: if RANK != 0: return @@ -354,17 +398,13 @@ def _verify(root: Path, *, adaptive: bool) -> None: for series, steps in ((every_two, (6,)), (every_three, (3, 6))): for step in steps: value, terms = _balance(series[step]) - if set(terms) != expected_terms \ - or value <= 0.0 \ - or terms["storage_change"] <= 0.0 \ - or any( - terms[name] != 0.0 - for name in expected_terms - {"storage_change"} - ): - raise AssertionError( - "%s due step %d did not publish its native nonzero Balance ledger" - % (label, step) - ) + if set(terms) != expected_terms: + raise AssertionError("%s due step %d omitted a Balance term" % (label, step)) + _require_exact_signed_balance(label, step, value, terms) + if _balance(every_two[6]) != _balance(every_three[6]): + raise AssertionError( + "%s independent due consumers disagreed on the accepted step-6 Balance" % label + ) def _run_case(root: Path, *, adaptive: bool) -> None: @@ -405,6 +445,16 @@ def _run_case(root: Path, *, adaptive: bool) -> None: ) if any(row != reports[0] for row in reports[1:]) or reports[0][0] != NSTEPS: raise AssertionError("%s run report differs across ranks: %r" % (label, reports)) + accepted_balance = tuple( + row + for row in runtime.inspect().to_dict()["instance"]["accepted_diagnostics"] + if row["key"]["reduction"] == "discrete_balance" + ) + accepted_by_rank = allgather_value(COMM, accepted_balance) + if not accepted_balance or any(row != accepted_by_rank[0] for row in accepted_by_rank[1:]): + raise AssertionError( + "%s accepted Balance registry differs across ranks: %r" % (label, accepted_by_rank) + ) barrier(COMM) _collective_local(label + " output verification", lambda: _verify(root, adaptive=adaptive)) From 0d8c0158fd7ed9bf1fc6aeff63025dfa873cc189 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:05:16 +0200 Subject: [PATCH 2/4] feat(runtime): retain automatic balance evidence per attempt --- .../runtime/program/program_runtime_state.hpp | 71 +++++++++++++++++++ src/runtime/amr/amr_system.cpp | 3 + src/runtime/system/system_impl.hpp | 3 + 3 files changed, 77 insertions(+) diff --git a/include/pops/runtime/program/program_runtime_state.hpp b/include/pops/runtime/program/program_runtime_state.hpp index a2ac587c9..e27984c07 100644 --- a/include/pops/runtime/program/program_runtime_state.hpp +++ b/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 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 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 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(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; diff --git a/src/runtime/amr/amr_system.cpp b/src/runtime/amr/amr_system.cpp index d84124499..cf5d1d318 100644 --- a/src/runtime/amr/amr_system.cpp +++ b/src/runtime/amr/amr_system.cpp @@ -462,6 +462,7 @@ struct AmrSystem::Impl { int cadence_clock_restore_macro_step = 0; std::map program_diagnostics; std::map step_balance_terms; + std::map 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; diff --git a/src/runtime/system/system_impl.hpp b/src/runtime/system/system_impl.hpp index 13abe58bd..40fddf113 100644 --- a/src/runtime/system/system_impl.hpp +++ b/src/runtime/system/system_impl.hpp @@ -617,6 +617,7 @@ struct System::Impl { int cadence_clock_restore_macro_step; std::map program_diagnostics; std::map step_balance_terms; + std::map 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; From f2135ace5ee757b2ab61290761f65597a621d1dc Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:05:22 +0200 Subject: [PATCH 3/4] feat(amr): extract signed reflux balance corrections --- .../time/amr/levels/amr_patch_range.hpp | 17 +++++++++++++ .../time/amr/levels/amr_subcycling.hpp | 5 +++- .../pops/runtime/amr/amr_program_reflux.hpp | 11 +++++--- .../runtime/program/amr_program_context.hpp | 25 ++++++++++++++++--- 4 files changed, 51 insertions(+), 7 deletions(-) diff --git a/include/pops/numerics/time/amr/levels/amr_patch_range.hpp b/include/pops/numerics/time/amr/levels/amr_patch_range.hpp index 02a52a46b..82d8d1bf0 100644 --- a/include/pops/numerics/time/amr/levels/amr_patch_range.hpp +++ b/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 component_sums(Real cell_measure) const { + if (!std::isfinite(static_cast(cell_measure)) || cell_measure <= Real(0)) + throw std::invalid_argument( + "FluxRegister component sum requires a finite positive cell measure"); + device_fence(); + std::vector result(static_cast(nc), Real(0)); + const std::size_t components = static_cast(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(); } diff --git a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp index 12292338a..0b4bde30e 100644 --- a/include/pops/numerics/time/amr/levels/amr_subcycling.hpp +++ b/include/pops/numerics/time/amr/levels/amr_subcycling.hpp @@ -824,7 +824,8 @@ class PreparedAmrProgramRefluxTransition { template 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* 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(), diff --git a/include/pops/runtime/amr/amr_program_reflux.hpp b/include/pops/runtime/amr/amr_program_reflux.hpp index 96ab1bd65..3e79b5af4 100644 --- a/include/pops/runtime/amr/amr_program_reflux.hpp +++ b/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* 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(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 diff --git a/include/pops/runtime/program/amr_program_context.hpp b/include/pops/runtime/program/amr_program_context.hpp index c22140dc3..05127b51c 100644 --- a/include/pops/runtime/program/amr_program_context.hpp +++ b/include/pops/runtime/program/amr_program_context.hpp @@ -1115,7 +1115,8 @@ class AmrProgramContext : public ProgramExecutionServices { amr::ClockStamp sync_clock = accepted; sync_clock.level = parent; for (int b = 0; b < n_blocks(); ++b) { - const std::size_t sb = static_cast(sys_block(b)); + const int runtime_block = sys_block(b); + const std::size_t sb = static_cast(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); @@ -1123,8 +1124,26 @@ class AmrProgramContext : public ProgramExecutionServices { 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 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(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(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(component)], "AmrProgramContext"); + } } sync_report_.push_back({parent, child, b, SyncPhase::AverageDown, sync_clock}); eng_->average_down_level(sb, child); From 31d21432d373be50ff0416bf66332f7469bbc84b Mon Sep 17 00:00:00 2001 From: desp0042 Date: Thu, 30 Jul 2026 06:05:26 +0200 Subject: [PATCH 4/4] test(architecture): fence automatic reflux balance evidence --- .../test_automatic_reflux_balance_fence.py | 99 +++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 tests/python/architecture/test_automatic_reflux_balance_fence.py diff --git a/tests/python/architecture/test_automatic_reflux_balance_fence.py b/tests/python/architecture/test_automatic_reflux_balance_fence.py new file mode 100644 index 000000000..7b2c866a6 --- /dev/null +++ b/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 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 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 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* 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" + )