From c8c1c029eee79884c39c682a271a920561f0fed8 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Sun, 2 Aug 2026 00:33:39 +0200 Subject: [PATCH 1/3] feat(time): persist cell temporal partition authority --- include/pops/runtime/amr_system.hpp | 3 + .../program/amr_program_checkpoint.hpp | 45 +++- .../runtime/program/amr_program_context.hpp | 25 +- .../program/cell_temporal_partition.hpp | 225 ++++++++++++++++++ python/bindings/core/init/init_amr.cpp | 1 + .../pops/runtime/_amr_checkpoint_contract.py | 31 ++- python/pops/runtime/_amr_checkpoint_v3.py | 24 +- python/pops/runtime/program_report.py | 159 +++++++++---- src/runtime/amr/amr_system.cpp | 16 ++ 9 files changed, 464 insertions(+), 65 deletions(-) create mode 100644 include/pops/runtime/program/cell_temporal_partition.hpp diff --git a/include/pops/runtime/amr_system.hpp b/include/pops/runtime/amr_system.hpp index 71c70314a..0b483efb0 100644 --- a/include/pops/runtime/amr_system.hpp +++ b/include/pops/runtime/amr_system.hpp @@ -852,6 +852,9 @@ class AmrSystem { /// Human/audit-readable qualification rows decoded from the same accepted image persisted as bytes. POPS_EXPORT std::vector> program_accepted_state_manifest() const; POPS_EXPORT std::vector> program_clock_manifest() const; + /// Accepted temporal-partition provider, synchronization tick and per-rung cell counts. The rows + /// are decoded from the same opaque image used by strict restart, never a capability ledger. + POPS_EXPORT std::vector> program_temporal_partition_manifest() const; POPS_EXPORT std::vector> program_flux_ledger_manifest() const; POPS_EXPORT std::vector> program_interface_flux_ledger_manifest() const; POPS_EXPORT std::vector> program_sync_manifest() const; diff --git a/include/pops/runtime/program/amr_program_checkpoint.hpp b/include/pops/runtime/program/amr_program_checkpoint.hpp index c0f5ebbc1..608400295 100644 --- a/include/pops/runtime/program/amr_program_checkpoint.hpp +++ b/include/pops/runtime/program/amr_program_checkpoint.hpp @@ -14,6 +14,7 @@ #include #include #include +#include namespace pops::runtime::program { @@ -52,6 +53,7 @@ struct AmrProgramSyncEvent { struct AmrProgramAcceptedState { std::vector level_clocks; std::map logical_clock_ticks; + CellTemporalPartitionAcceptedState temporal_partition; /// Rank-independent canonical image of the runtime-owned AMR tagging hysteresis. std::vector tagging_hysteresis_state; std::map history_owners; @@ -448,12 +450,25 @@ Map read_map(Reader& in, ReadValue&& read_value) { inline std::vector serialize_amr_program_accepted_state( const AmrProgramAcceptedState& state) { using namespace checkpoint_detail; + validate_cell_temporal_partition_state(state.temporal_partition); Writer out; - out.u64(0x3454534153504f50ULL); // "POPSAST4", little-endian bytes + out.u64(0x3554534153504f50ULL); // "POPSAST5", little-endian bytes out.size(state.level_clocks.size()); for (const auto& clock : state.level_clocks) write_clock(out, clock); write_map(out, state.logical_clock_ticks, [](Writer& w, std::int64_t value) { w.i64(value); }); + out.u64(static_cast(state.temporal_partition.kind)); + out.string(state.temporal_partition.provider_identity); + out.u64(state.temporal_partition.topology_epoch); + out.i64(state.temporal_partition.synchronization_tick); + out.i64(state.temporal_partition.tick_denominator); + out.size(state.temporal_partition.cells.size()); + for (const CellTemporalPartitionRecord& cell : state.temporal_partition.cells) { + out.i32(cell.level); + out.u64(cell.cell); + out.i32(cell.rung); + out.i64(cell.accepted_tick); + } out.bytes(state.tagging_hysteresis_state); write_map(out, state.history_owners, [](Writer& w, int v) { w.i32(v); }); write_map(out, state.history_states, [](Writer& w, const std::string& v) { w.string(v); }); @@ -520,7 +535,9 @@ inline AmrProgramAcceptedState deserialize_amr_program_accepted_state( const std::vector& bytes) { using namespace checkpoint_detail; Reader in(bytes); - if (in.u64() != 0x3454534153504f50ULL) + const std::uint64_t magic = in.u64(); + const bool carries_temporal_partition = magic == 0x3554534153504f50ULL; + if (!carries_temporal_partition && magic != 0x3454534153504f50ULL) throw std::runtime_error( "invalid AMR Program accepted-state payload: unsupported magic/version"); AmrProgramAcceptedState state; @@ -529,6 +546,30 @@ inline AmrProgramAcceptedState deserialize_amr_program_accepted_state( clock = read_clock(in); state.logical_clock_ticks = read_map(in, [](Reader& r) { return r.i64(); }); + if (carries_temporal_partition) { + const std::uint64_t kind = in.u64(); + if (kind > static_cast(TemporalPartitionKind::CellLocal)) + throw std::runtime_error( + "invalid AMR Program accepted-state payload: unsupported temporal partition kind"); + state.temporal_partition.kind = static_cast(kind); + state.temporal_partition.provider_identity = in.string(); + state.temporal_partition.topology_epoch = in.u64(); + state.temporal_partition.synchronization_tick = in.i64(); + state.temporal_partition.tick_denominator = in.i64(); + state.temporal_partition.cells.resize(in.size()); + for (CellTemporalPartitionRecord& cell : state.temporal_partition.cells) { + cell.level = in.i32(); + cell.cell = in.u64(); + cell.rung = in.i32(); + cell.accepted_tick = in.i64(); + } + try { + validate_cell_temporal_partition_state(state.temporal_partition); + } catch (const std::exception& error) { + throw std::runtime_error(std::string("invalid AMR Program accepted-state payload: ") + + error.what()); + } + } state.tagging_hysteresis_state = in.bytes(); state.history_owners = read_map>(in, [](Reader& r) { return r.i32(); }); diff --git a/include/pops/runtime/program/amr_program_context.hpp b/include/pops/runtime/program/amr_program_context.hpp index 4cf7c6b37..0c3bcd222 100644 --- a/include/pops/runtime/program/amr_program_context.hpp +++ b/include/pops/runtime/program/amr_program_context.hpp @@ -1278,6 +1278,10 @@ class AmrProgramContext : public ProgramExecutionServices { AttemptSnapshot& saved = attempt_snapshot_; capture_engine_attempt_snapshot_(saved, borrows_facade_snapshot); import_program_accepted_state_(); + // ADC-756 foundation: an authenticated cell-local checkpoint must never fall through to the + // existing hierarchy-global driver. The future prepared Kokkos partition executor will consume + // this same authority; until then, fail before the Program body or any published clock mutates. + temporal_partition_.require_global_execution_route(); capture_program_attempt_snapshot_(saved); conservative_ledger_.begin(); try { @@ -1382,6 +1386,16 @@ class AmrProgramContext : public ProgramExecutionServices { } void validate_program_accepted_state_(const AmrProgramAcceptedState& state) const { + validate_cell_temporal_partition_state(state.temporal_partition); + if (state.temporal_partition.kind == TemporalPartitionKind::CellLocal) { + if (state.temporal_partition.topology_epoch != eng_->topology_epoch()) + throw std::runtime_error( + "AMR Program cell-local temporal partition targets another topology epoch"); + for (const CellTemporalPartitionRecord& cell : state.temporal_partition.cells) + if (cell.level >= nlev()) + throw std::runtime_error( + "AMR Program cell-local temporal partition targets an inactive level"); + } if (state.level_clocks.size() != static_cast(nlev())) throw std::runtime_error( "AMR Program accepted state does not match the restored hierarchy level count"); @@ -1507,6 +1521,7 @@ class AmrProgramContext : public ProgramExecutionServices { const std::int64_t accepted_step = level_clocks_.empty() ? macro_step() : level_clocks_.front().macro_step; state.logical_clock_ticks = clock_schedule_.accepted_ticks(accepted_step); + state.temporal_partition = temporal_partition_.checkpoint(); state.tagging_hysteresis_state = eng_->checkpoint_tagging_state(); state.history_owners = history_owners_; state.history_states = history_state_ids_; @@ -1561,6 +1576,7 @@ class AmrProgramContext : public ProgramExecutionServices { const std::int64_t accepted_step = level_clocks_.empty() ? macro_step() : level_clocks_.front().macro_step; clock_schedule_.restore_accepted_ticks(state.logical_clock_ticks, accepted_step); + temporal_partition_.restore(std::move(state.temporal_partition)); history_owners_ = std::move(state.history_owners); history_state_ids_ = std::move(state.history_states); history_space_ids_ = std::move(state.history_spaces); @@ -1705,6 +1721,7 @@ class AmrProgramContext : public ProgramExecutionServices { std::uint64_t engine_topology_generation = 0; std::vector program_accepted_state; std::uint64_t program_accepted_state_revision = 0; + CellTemporalPartitionAcceptedState temporal_partition; std::set active_flux; std::map flux; std::map> flux_contributions; @@ -1928,6 +1945,7 @@ class AmrProgramContext : public ProgramExecutionServices { throw std::logic_error( "AMR Program accepted state changed while capturing an attempt snapshot"); snapshot.program_accepted_state_revision = accepted_revision; + snapshot.temporal_partition = temporal_partition_.checkpoint(); copy_set_in_place_(snapshot.active_flux, active_flux_ledger_); for (auto entry = snapshot.flux.begin(); entry != snapshot.flux.end();) { @@ -2047,6 +2065,8 @@ class AmrProgramContext : public ProgramExecutionServices { current_window_ = snapshot.window; current_sync_clock_ = snapshot.sync_clock; current_level_dt_ = snapshot.level_dt; + temporal_partition_.rollback(); + temporal_partition_.restore(snapshot.temporal_partition); } template @@ -3275,8 +3295,8 @@ class AmrProgramContext : public ProgramExecutionServices { 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, @@ -3534,6 +3554,7 @@ class AmrProgramContext : public ProgramExecutionServices { mutable bool restart_regrid_prepared_ = false; mutable int automatic_regrid_macro_step_ = -1; mutable std::vector level_clocks_; + mutable BatchedCellTemporalPartition temporal_partition_; mutable std::uint64_t accepted_state_revision_ = 0; mutable std::uint64_t operator_topology_revision_counter_ = 0; mutable std::uint64_t observed_operator_topology_epoch_ = diff --git a/include/pops/runtime/program/cell_temporal_partition.hpp b/include/pops/runtime/program/cell_temporal_partition.hpp new file mode 100644 index 000000000..003c67540 --- /dev/null +++ b/include/pops/runtime/program/cell_temporal_partition.hpp @@ -0,0 +1,225 @@ +#pragma once + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace pops::runtime::program { + +inline constexpr const char* kGlobalTemporalPartitionProvider = "pops.temporal-partition.global@1"; + +enum class TemporalPartitionKind : std::uint8_t { Global = 0, CellLocal = 1 }; + +/// Rank-independent identity and accepted logical clock of one cell. +/// +/// ``cell`` is a provider-owned canonical cell id within ``level``. It deliberately does not carry +/// an MPI rank or patch-local address, so ownership migration can rematerialize device storage from +/// the same accepted scientific image. +struct CellTemporalPartitionRecord { + int level = 0; + std::uint64_t cell = 0; + int rung = 0; + std::int64_t accepted_tick = 0; + + friend bool operator==(const CellTemporalPartitionRecord&, + const CellTemporalPartitionRecord&) = default; +}; + +/// Compact accepted-boundary image for a prepared temporal-partition provider. +/// +/// Physical time is represented by integer ticks over one provider-owned denominator. Accepted +/// checkpoints are synchronization barriers: every record must be at ``synchronization_tick``. +/// Attempt-local clocks live in ``BatchedCellTemporalPartition`` and can never leak into this image. +struct CellTemporalPartitionAcceptedState { + TemporalPartitionKind kind = TemporalPartitionKind::Global; + std::string provider_identity = kGlobalTemporalPartitionProvider; + std::uint64_t topology_epoch = 0; + std::int64_t synchronization_tick = 0; + std::int64_t tick_denominator = 1; + std::vector cells; + + friend bool operator==(const CellTemporalPartitionAcceptedState&, + const CellTemporalPartitionAcceptedState&) = default; +}; + +inline void validate_cell_temporal_partition_state( + const CellTemporalPartitionAcceptedState& state) { + if (state.provider_identity.empty()) + throw std::invalid_argument("temporal partition provider identity cannot be empty"); + if (state.synchronization_tick < 0 || state.tick_denominator <= 0) + throw std::invalid_argument( + "temporal partition accepted clock requires non-negative ticks and a positive denominator"); + if (state.kind == TemporalPartitionKind::Global) { + if (state.provider_identity != kGlobalTemporalPartitionProvider || state.topology_epoch != 0 || + !state.cells.empty()) + throw std::invalid_argument( + "global temporal partition state cannot carry topology or cell-local clocks"); + return; + } + if (state.kind != TemporalPartitionKind::CellLocal) + throw std::invalid_argument("temporal partition state has an unsupported kind"); + if (state.provider_identity == kGlobalTemporalPartitionProvider || state.cells.empty()) + throw std::invalid_argument( + "cell-local temporal partition state requires its prepared provider and cell clocks"); + + std::tuple previous{-1, 0}; + bool first = true; + for (const CellTemporalPartitionRecord& cell : state.cells) { + if (cell.level < 0 || cell.rung < 0 || cell.rung > 30 || cell.accepted_tick < 0) + throw std::invalid_argument("temporal partition cell record is outside its bounded domain"); + const std::tuple identity{cell.level, cell.cell}; + if (!first && !(previous < identity)) + throw std::invalid_argument( + "temporal partition cell records must be unique and canonically ordered"); + if (cell.accepted_tick != state.synchronization_tick) + throw std::invalid_argument( + "temporal partition accepted checkpoint is not at a synchronization barrier"); + const std::int64_t stride = std::int64_t{1} << cell.rung; + if (cell.accepted_tick % stride != 0) + throw std::invalid_argument( + "temporal partition accepted tick is not aligned to its cell rung"); + previous = identity; + first = false; + } +} + +/// Host authority for one bounded batch schedule and its transactional clocks. +/// +/// This class owns no numerical field and launches no per-cell task. A future Kokkos execution +/// provider consumes the canonical records in rung batches; this authority supplies exact attempt, +/// rollback, barrier, checkpoint and diagnostic semantics independently of the execution space. +class BatchedCellTemporalPartition { + public: + explicit BatchedCellTemporalPartition( + CellTemporalPartitionAcceptedState accepted = CellTemporalPartitionAcceptedState{}) + : accepted_(std::move(accepted)) { + validate_cell_temporal_partition_state(accepted_); + } + + const CellTemporalPartitionAcceptedState& accepted_state() const noexcept { return accepted_; } + bool attempt_active() const noexcept { return attempt_active_; } + + void begin_attempt(std::int64_t target_tick) { + if (attempt_active_) + throw std::logic_error("temporal partition attempt is already active"); + if (target_tick <= accepted_.synchronization_tick) + throw std::invalid_argument("temporal partition attempt target must advance accepted time"); + pending_ticks_.clear(); + pending_ticks_.reserve(accepted_.cells.size()); + for (const CellTemporalPartitionRecord& cell : accepted_.cells) { + const std::int64_t stride = std::int64_t{1} << cell.rung; + if ((target_tick - cell.accepted_tick) % stride != 0) + throw std::invalid_argument( + "temporal partition attempt target is unreachable for one prepared rung"); + pending_ticks_.push_back(cell.accepted_tick); + } + target_tick_ = target_tick; + attempt_active_ = true; + } + + /// Advance a canonically ordered batch of record indices belonging to exactly one rung. + void advance_batch(int rung, const std::vector& indices, std::int64_t target_tick) { + if (!attempt_active_) + throw std::logic_error("temporal partition batch requires an active attempt"); + if (indices.empty()) + throw std::invalid_argument("temporal partition batch cannot be empty"); + if (target_tick > target_tick_) + throw std::invalid_argument("temporal partition batch crosses its synchronization target"); + std::size_t previous = std::numeric_limits::max(); + for (std::size_t index : indices) { + if (index >= accepted_.cells.size() || + (previous != std::numeric_limits::max() && index <= previous)) + throw std::invalid_argument( + "temporal partition batch indices must be unique and canonically ordered"); + const CellTemporalPartitionRecord& cell = accepted_.cells[index]; + if (cell.rung != rung) + throw std::invalid_argument("temporal partition batch mixes prepared rungs"); + const std::int64_t stride = std::int64_t{1} << cell.rung; + if (target_tick <= pending_ticks_[index] || + (target_tick - pending_ticks_[index]) % stride != 0) + throw std::invalid_argument( + "temporal partition batch target is not a forward rung-aligned tick"); + previous = index; + } + for (std::size_t index : indices) + pending_ticks_[index] = target_tick; + } + + void require_barrier(const std::string& operation) const { + if (!attempt_active_) + return; + if (std::any_of(pending_ticks_.begin(), pending_ticks_.end(), + [this](std::int64_t tick) { return tick != target_tick_; })) + throw std::logic_error(operation + + " requires every cell-local clock at the synchronization barrier"); + } + + void commit() { + if (!attempt_active_) + throw std::logic_error("temporal partition commit requires an active attempt"); + require_barrier("temporal partition commit"); + for (std::size_t index = 0; index < accepted_.cells.size(); ++index) + accepted_.cells[index].accepted_tick = pending_ticks_[index]; + accepted_.synchronization_tick = target_tick_; + clear_attempt_(); + } + + void rollback() noexcept { clear_attempt_(); } + + CellTemporalPartitionAcceptedState checkpoint() const { + if (attempt_active_) + throw std::logic_error( + "temporal partition checkpoint requires an accepted synchronization barrier"); + return accepted_; + } + + void restore(CellTemporalPartitionAcceptedState accepted) { + if (attempt_active_) + throw std::logic_error("temporal partition restore cannot replace an active attempt"); + validate_cell_temporal_partition_state(accepted); + accepted_ = std::move(accepted); + } + + void require_global_execution_route() const { + if (accepted_.kind != TemporalPartitionKind::Global) + throw std::logic_error( + "cell-local temporal partition requires its prepared batched executor; the global AMR " + "step cannot silently replace it"); + } + + std::vector> manifest() const { + std::map rung_counts; + for (const CellTemporalPartitionRecord& cell : accepted_.cells) + ++rung_counts[cell.rung]; + std::vector> rows; + rows.push_back( + {"summary", accepted_.kind == TemporalPartitionKind::Global ? "global" : "cell_local", + accepted_.provider_identity, std::to_string(accepted_.topology_epoch), + std::to_string(accepted_.synchronization_tick), std::to_string(accepted_.tick_denominator), + std::to_string(accepted_.cells.size())}); + for (const auto& [rung, count] : rung_counts) + rows.push_back({"rung", std::to_string(rung), std::to_string(count)}); + return rows; + } + + private: + void clear_attempt_() noexcept { + pending_ticks_.clear(); + target_tick_ = 0; + attempt_active_ = false; + } + + CellTemporalPartitionAcceptedState accepted_; + std::vector pending_ticks_; + std::int64_t target_tick_ = 0; + bool attempt_active_ = false; +}; + +} // namespace pops::runtime::program diff --git a/python/bindings/core/init/init_amr.cpp b/python/bindings/core/init/init_amr.cpp index 324affaa5..25fab07ab 100644 --- a/python/bindings/core/init/init_amr.cpp +++ b/python/bindings/core/init/init_amr.cpp @@ -824,6 +824,7 @@ void bind_amr_program(py::class_& cls) { py::arg("payload"), py::arg("names"), py::arg("depths"), py::arg("ncomps")) .def("program_accepted_state_manifest", &AmrSystem::program_accepted_state_manifest) .def("program_clock_manifest", &AmrSystem::program_clock_manifest) + .def("program_temporal_partition_manifest", &AmrSystem::program_temporal_partition_manifest) .def("program_flux_ledger_manifest", &AmrSystem::program_flux_ledger_manifest) .def("program_interface_flux_ledger_manifest", &AmrSystem::program_interface_flux_ledger_manifest) diff --git a/python/pops/runtime/_amr_checkpoint_contract.py b/python/pops/runtime/_amr_checkpoint_contract.py index 152a0fa0a..a60cbbefc 100644 --- a/python/pops/runtime/_amr_checkpoint_contract.py +++ b/python/pops/runtime/_amr_checkpoint_contract.py @@ -8,7 +8,7 @@ from pops.identity import make_identity -_SCHEMA = 4 +_SCHEMA = 5 _GUARANTEE = "bit_identical_accepted_state" _CONTRACT_KEYS = { "schema_version", @@ -17,6 +17,7 @@ "ledger", "interface_ledger", "clocks", + "temporal_partition", "synchronization", "history_qualifications", "level_relations", @@ -39,9 +40,7 @@ def restart_topology_image(sim): """Return the compact identity of one accepted AMR hierarchy.""" levels = int(sim.n_levels()) boxes = [[int(value) for value in box] for box in sim.patch_boxes()] - owners = [ - [int(rank) for rank in sim.level_owner_ranks(level)] for level in range(levels) - ] + owners = [[int(rank) for rank in sim.level_owner_ranks(level)] for level in range(levels)] topology_identity = make_identity( "restart-topology", { @@ -94,6 +93,7 @@ def contract_for(sim): "entries": interface_flux_ledger, }, "clocks": _rows(sim.program_clock_manifest()), + "temporal_partition": _rows(sim.program_temporal_partition_manifest()), "synchronization": _rows(sim.program_sync_manifest()), "history_qualifications": _rows(sim.program_accepted_state_manifest()), "level_relations": relations, @@ -116,6 +116,25 @@ def _decode_contract(payload): return contract +def checkpoint_temporal_partition_kind(payload): + """Return the exact accepted temporal-partition kind before native restart mutation.""" + contract = _decode_contract(payload) + rows = contract["temporal_partition"] + if ( + not isinstance(rows, list) + or not rows + or not isinstance(rows[0], list) + or len(rows[0]) != 7 + or rows[0][0] != "summary" + or rows[0][1] not in {"global", "cell_local"} + ): + raise ValueError("restart: AMR temporal-partition contract has an invalid summary") + for row in rows[1:]: + if not isinstance(row, list) or len(row) != 3 or row[0] != "rung": + raise ValueError("restart: AMR temporal-partition contract has an invalid rung row") + return rows[0][1] + + def preflight_contract(sim, payload): """Authenticate shape and static provenance before the native restart transaction.""" import numpy as np @@ -159,9 +178,7 @@ def _validate_interface_ledger_against_live_hierarchy(sim, contract): blocks = int(sim.n_blocks()) for row in contract["interface_ledger"]["entries"]: if len(row) != 28: - raise ValueError( - "restart: restored AMR interface-flux audit has an invalid native row" - ) + raise ValueError("restart: restored AMR interface-flux audit has an invalid native row") coarse_level, fine_level = int(row[2]), int(row[3]) left_block, right_block = int(row[21]), int(row[22]) if ( diff --git a/python/pops/runtime/_amr_checkpoint_v3.py b/python/pops/runtime/_amr_checkpoint_v3.py index b83678cab..8969f0d8f 100644 --- a/python/pops/runtime/_amr_checkpoint_v3.py +++ b/python/pops/runtime/_amr_checkpoint_v3.py @@ -345,9 +345,7 @@ def _capture_v3(owner, sim, prepared): if prepared.local_program_state: rematerialize = getattr(sim, "rematerialize_program_accepted_state", None) if not callable(rematerialize): - raise TypeError( - "checkpoint AMR engine lacks accepted-state consensus validation" - ) + raise TypeError("checkpoint AMR engine lacks accepted-state consensus validation") # Re-materializing onto the unchanged ownership is a non-mutating validation pass. It # authenticates every rank-independent accepted field, including persistent tagging, # before any rank may seal or publish a checkpoint. @@ -447,7 +445,10 @@ def prepare_v3( """ import numpy as np from pops.output._checkpoint_collective import checkpoint_topology - from pops.runtime._amr_checkpoint_contract import preflight_contract + from pops.runtime._amr_checkpoint_contract import ( + checkpoint_temporal_partition_kind, + preflight_contract, + ) from pops.runtime._program_cadence_checkpoint import prepare_program_cadence from pops.runtime._temporal_restart import TemporalRestartState @@ -488,6 +489,17 @@ def prepare_v3( raise ValueError( "restart: RegridOnRestart requires an artifact-backed compiled AMR Program" ) + if checkpoint_temporal_partition_kind(d) == "cell_local": + if checkpoint_ranks != current_ranks: + raise ValueError( + "restart: cell-local temporal partitions require the recorded MPI cardinality " + "until ownership rematerialization is implemented" + ) + if hierarchy_mode != "restore_recorded_hierarchy": + raise ValueError( + "restart: cell-local temporal partitions require RestoreRecordedHierarchy " + "until regrid rematerialization is implemented" + ) checkpoint_levels, checkpoint_configured_levels = _checkpoint_amr_level_envelope(sim, d) from pops.runtime._amr_checkpoint_topology import recorded_rank_topology @@ -1060,9 +1072,7 @@ def apply_v3(owner, sim, prepared): "recorded accepted-contract identity", lambda: _restart_accepted_contract_identity(sim), ) - before_history_identity = _restart_history_identity( - owner, sim, phase="recorded hierarchy" - ) + before_history_identity = _restart_history_identity(owner, sim, phase="recorded hierarchy") before_integrals = _restart_composite_integrals(owner, sim, phase="recorded hierarchy") _restart_collective_phase( owner, diff --git a/python/pops/runtime/program_report.py b/python/pops/runtime/program_report.py index 6194a5727..28fa57b14 100644 --- a/python/pops/runtime/program_report.py +++ b/python/pops/runtime/program_report.py @@ -42,13 +42,28 @@ class ProgramRuntimeReport: sections); a bound program fills the sections from the C++ Program subsystem accessors. """ - schema_version = 3 + schema_version = 4 report_type = "program_runtime" - def __init__(self, *, installed: Any, program_hash: Any, step_transaction: Any, block_map: Any, - params: Any, diagnostics: Any, histories: Any, cache: Any, - profiler: Any, clocks: Any, level_relations: Any, - flux_ledger: Any, synchronization: Any, temporal: Any) -> None: + def __init__( + self, + *, + installed: Any, + program_hash: Any, + step_transaction: Any, + block_map: Any, + params: Any, + diagnostics: Any, + histories: Any, + cache: Any, + profiler: Any, + clocks: Any, + level_relations: Any, + flux_ledger: Any, + synchronization: Any, + temporal_partition: Any, + temporal: Any, + ) -> None: self.installed = bool(installed) self.program_hash = program_hash or "" self.step_transaction = dict(step_transaction) @@ -62,6 +77,7 @@ def __init__(self, *, installed: Any, program_hash: Any, step_transaction: Any, self.level_relations = [dict(row) for row in level_relations] self.flux_ledger = [dict(row) for row in flux_ledger] self.synchronization = [dict(row) for row in synchronization] + self.temporal_partition = dict(temporal_partition) self.temporal = dict(temporal) def to_dict(self) -> Any: @@ -81,6 +97,7 @@ def to_dict(self) -> Any: "level_relations": [dict(row) for row in self.level_relations], "flux_ledger": [dict(row) for row in self.flux_ledger], "synchronization": [dict(row) for row in self.synchronization], + "temporal_partition": dict(self.temporal_partition), "temporal": dict(self.temporal), } @@ -93,9 +110,12 @@ def to_json(self, path: Any = None, *, indent: int = 2) -> Any: return text def __repr__(self) -> Any: - return ("ProgramRuntimeReport(installed=%r, hash=%r, histories=%d, cache=%d)" - % (self.installed, self.program_hash or "(none)", len(self.histories), - len(self.cache))) + return "ProgramRuntimeReport(installed=%r, hash=%r, histories=%d, cache=%d)" % ( + self.installed, + self.program_hash or "(none)", + len(self.histories), + len(self.cache), + ) def __str__(self) -> Any: strategy = self.step_transaction.get("strategy", {}) @@ -112,6 +132,7 @@ def __str__(self) -> Any: lines.append(" clocks : %d cursor(s)" % len(self.clocks)) lines.append(" flux ledger : %d accepted contribution(s)" % len(self.flux_ledger)) lines.append(" sync : %d phase event(s)" % len(self.synchronization)) + lines.append(" partition : %s" % (self.temporal_partition.get("kind") or "(none)")) return "\n".join(lines) @@ -121,6 +142,7 @@ def _params(sim: Any) -> Any: count 0. The limit (ADC-610) surfaces the previously-hidden fixed-array capacity so a block's headroom is introspectable.""" from pops.physics.aux import max_runtime_params # lazy: keep the report import-light + limit = max_runtime_params() rows = [] block_map = list(_call(sim, "program_block_map", []) or []) @@ -135,36 +157,44 @@ def _params(sim: Any) -> Any: def _histories(sim: Any) -> Any: rows = [] for name in _call(sim, "history_names", []) or []: - rows.append({ - "name": name, - "depth": _call(sim, "history_depth", None, name), - "ncomp": _call(sim, "history_ncomp", None, name), - "initialized": _call(sim, "history_initialized", None, name), - }) + rows.append( + { + "name": name, + "depth": _call(sim, "history_depth", None, name), + "ncomp": _call(sim, "history_ncomp", None, name), + "initialized": _call(sim, "history_initialized", None, name), + } + ) return rows def _cache(sim: Any) -> Any: rows = [] for node_id in _call(sim, "program_cache_nodes", []) or []: - rows.append({ - "node_id": int(node_id), - "name": _call(sim, "program_cache_name", "", node_id), - "last_update_step": _call(sim, "program_cache_last_update_step", None, node_id), - "accumulated_dt": _call(sim, "program_cache_accumulated_dt", None, node_id), - }) + rows.append( + { + "node_id": int(node_id), + "name": _call(sim, "program_cache_name", "", node_id), + "last_update_step": _call(sim, "program_cache_last_update_step", None, node_id), + "accumulated_dt": _call(sim, "program_cache_accumulated_dt", None, node_id), + } + ) return rows -def _amr_temporal_report(sim: Any) -> tuple[Any, Any, Any, Any]: +def _amr_temporal_report(sim: Any) -> tuple[Any, Any, Any, Any, Any]: clocks = [] for row in _call(sim, "program_clock_manifest", []) or []: if row[0] == "level" and len(row) == 6: - clocks.append({ - "kind": "level", "level": int(row[1]), "macro_step": int(row[2]), - "phase": {"numerator": int(row[3]), "denominator": int(row[4])}, - "physical_time": float(row[5]), - }) + clocks.append( + { + "kind": "level", + "level": int(row[1]), + "macro_step": int(row[2]), + "phase": {"numerator": int(row[3]), "denominator": int(row[4])}, + "physical_time": float(row[5]), + } + ) elif row[0] == "logical" and len(row) == 3: clocks.append({"kind": "logical", "clock": row[1], "tick": int(row[2])}) else: @@ -173,33 +203,66 @@ def _amr_temporal_report(sim: Any) -> tuple[Any, Any, Any, Any]: for row in _call(sim, "checkpoint_temporal_relations", []) or []: if len(row) != 5: raise ValueError("native AMR temporal relation report has an invalid row") - relations.append({ - "parent_level": int(row[0]), "child_level": int(row[1]), - "temporal_ratio": {"numerator": int(row[2]), "denominator": int(row[3])}, - "remainder_policy": row[4], - }) + relations.append( + { + "parent_level": int(row[0]), + "child_level": int(row[1]), + "temporal_ratio": {"numerator": int(row[2]), "denominator": int(row[3])}, + "remainder_policy": row[4], + } + ) ledger = [] for row in _call(sim, "program_flux_ledger_manifest", []) or []: if len(row) != 13: raise ValueError("native AMR Program flux-ledger report has an invalid row") - ledger.append({ - "owner": row[0], "state": row[1], "rate": row[2], "flux": row[3], - "level": int(row[4]), "macro_step": int(row[5]), - "phase": {"numerator": int(row[6]), "denominator": int(row[7])}, - "stage_weight": {"numerator": int(row[8]), "denominator": int(row[9])}, - "orientation": row[10], "face_measure": float(row[11]), - "substep_duration": float(row[12]), - }) + ledger.append( + { + "owner": row[0], + "state": row[1], + "rate": row[2], + "flux": row[3], + "level": int(row[4]), + "macro_step": int(row[5]), + "phase": {"numerator": int(row[6]), "denominator": int(row[7])}, + "stage_weight": {"numerator": int(row[8]), "denominator": int(row[9])}, + "orientation": row[10], + "face_measure": float(row[11]), + "substep_duration": float(row[12]), + } + ) synchronization = [] for row in _call(sim, "program_sync_manifest", []) or []: if len(row) != 7: raise ValueError("native AMR Program synchronization report has an invalid row") - synchronization.append({ - "parent_level": int(row[0]), "child_level": int(row[1]), - "block": int(row[2]), "phase": row[3], "macro_step": int(row[4]), - "clock_phase": {"numerator": int(row[5]), "denominator": int(row[6])}, - }) - return clocks, relations, ledger, synchronization + synchronization.append( + { + "parent_level": int(row[0]), + "child_level": int(row[1]), + "block": int(row[2]), + "phase": row[3], + "macro_step": int(row[4]), + "clock_phase": {"numerator": int(row[5]), "denominator": int(row[6])}, + } + ) + temporal_partition = {} + for row in _call(sim, "program_temporal_partition_manifest", []) or []: + if row[0] == "summary" and len(row) == 7: + if temporal_partition: + raise ValueError("native temporal-partition report has duplicate summary rows") + temporal_partition = { + "kind": row[1], + "provider_identity": row[2], + "topology_epoch": int(row[3]), + "synchronization_tick": int(row[4]), + "tick_denominator": int(row[5]), + "cell_count": int(row[6]), + "rungs": [], + } + elif row[0] == "rung" and len(row) == 3 and temporal_partition: + temporal_partition["rungs"].append({"rung": int(row[1]), "cells": int(row[2])}) + else: + raise ValueError("native temporal-partition report has an invalid row") + return clocks, relations, ledger, synchronization, temporal_partition def build_program_report(sim: Any) -> Any: @@ -210,7 +273,7 @@ def build_program_report(sim: Any) -> Any: missing an accessor yields ``None`` for that field. """ program_hash = _call(sim, "installed_program_hash", "") or "" - clocks, relations, ledger, synchronization = _amr_temporal_report(sim) + clocks, relations, ledger, synchronization, temporal_partition = _amr_temporal_report(sim) temporal_state = getattr(sim, "_temporal_restart_state", None) temporal = temporal_state.to_data() if temporal_state is not None else {} return ProgramRuntimeReport( @@ -218,7 +281,8 @@ def build_program_report(sim: Any) -> Any: program_hash=program_hash, step_transaction=( sim._step_transaction_plan.to_data() - if getattr(sim, "_step_transaction_plan", None) is not None else {} + if getattr(sim, "_step_transaction_plan", None) is not None + else {} ), block_map=list(_call(sim, "program_block_map", []) or []), params=_params(sim), @@ -230,5 +294,6 @@ def build_program_report(sim: Any) -> Any: level_relations=relations, flux_ledger=ledger, synchronization=synchronization, + temporal_partition=temporal_partition, temporal=temporal, ) diff --git a/src/runtime/amr/amr_system.cpp b/src/runtime/amr/amr_system.cpp index 683a7b0d7..a79594bce 100644 --- a/src/runtime/amr/amr_system.cpp +++ b/src/runtime/amr/amr_system.cpp @@ -3328,6 +3328,15 @@ void AmrSystem::restore_checkpoint_accepted_state(const std::vectorruntime->topology_epoch()) + throw std::runtime_error( + "AMR checkpoint cell-local temporal partition targets another topology epoch"); + for (const auto& cell : accepted.temporal_partition.cells) + if (cell.level < 0 || cell.level >= p_->runtime->nlev()) + throw std::runtime_error( + "AMR checkpoint cell-local temporal partition targets an inactive level"); + } tagging_candidate = p_->runtime->prepare_checkpoint_tagging_state(accepted.tagging_hysteresis_state); } else { @@ -3435,6 +3444,13 @@ std::vector> AmrSystem::program_clock_manifest() const rows.push_back({"logical", identity, std::to_string(tick)}); return rows; } +std::vector> AmrSystem::program_temporal_partition_manifest() const { + if (p_->program_accepted_state_.empty()) + return {}; + const auto state = + runtime::program::deserialize_amr_program_accepted_state(p_->program_accepted_state_); + return runtime::program::BatchedCellTemporalPartition(state.temporal_partition).manifest(); +} std::vector> AmrSystem::program_flux_ledger_manifest() const { std::vector> rows; if (p_->program_accepted_state_.empty()) From ca4862e68640d914e01a8ded125cb1f8db2b5df8 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Sun, 2 Aug 2026 00:33:50 +0200 Subject: [PATCH 2/3] test(time): cover temporal partition restart and no-bypass --- tests/CMakeLists.txt | 3 + tests/cpp/build_durations.json | 6 +- .../amr/test_temporal_partition_restart.cpp | 155 ++++++++++++++++++ tests/cpp/test_durations.json | 6 +- tests/cpp/test_sources.cmake | 1 + .../runtime/test_amr_checkpoint_contract.py | 68 ++++++-- .../unit/runtime/test_program_report.py | 57 +++++-- tests/test_manifest.toml | 5 + 8 files changed, 272 insertions(+), 29 deletions(-) create mode 100644 tests/cpp/integration/amr/test_temporal_partition_restart.cpp diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index c28cb399d..b11d2f92c 100644 --- a/tests/CMakeLists.txt +++ b/tests/CMakeLists.txt @@ -692,6 +692,9 @@ pops_add_gtest_suite( pops_test_source(_src test_program_reflux_ledger) pops_add_gtest_suite(NAME test_program_reflux_ledger SOURCES "${_src}" EXTRA_LIBS ${CMAKE_DL_LIBS} pops_runtime_amr) +pops_test_source(_src test_temporal_partition_restart) +pops_add_gtest_suite(NAME test_temporal_partition_restart SOURCES "${_src}" EXTRA_LIBS ${CMAKE_DL_LIBS} pops_runtime_amr) + pops_test_source(_src test_amr_transfer_properties) pops_add_gtest_suite(NAME test_amr_transfer_properties SOURCES "${_src}" EXTRA_LIBS ${CMAKE_DL_LIBS} pops_runtime_amr) diff --git a/tests/cpp/build_durations.json b/tests/cpp/build_durations.json index 63f1b081b..7231d4a15 100644 --- a/tests/cpp/build_durations.json +++ b/tests/cpp/build_durations.json @@ -6,7 +6,8 @@ "test_amr_program_diffusion", "test_amr_program_positivity_floor", "test_flux_failure_loader_transaction", - "test_interface_flux_fragment_ledger" + "test_interface_flux_fragment_ledger", + "test_temporal_partition_restart" ], "estimate_policy": "new unmeasured targets use a conservative analogous-target estimate, falling back to the catalog median until the next cold-CI timing refresh", "measured_refresh": { @@ -16,7 +17,7 @@ "refresh_source_run": "30190778708", "source_job": "87152034744", "source_run": "29352485297", - "target_count": 187, + "target_count": 188, "unit_seconds": "modeled shard wall time: measured serial-pool TU or parallel-share floor" }, "test_adaptive_multirate": 2.0, @@ -191,6 +192,7 @@ "test_step_attempt_rejected_header_only": 2.0, "test_structured_solver_diagnostics": 2.0, "test_sync_residence": 2.0, + "test_temporal_partition_restart": 15.0, "test_system_abstraction": 2.0, "test_system_coupler": 2.0, "test_system_hardening": 2.0, diff --git a/tests/cpp/integration/amr/test_temporal_partition_restart.cpp b/tests/cpp/integration/amr/test_temporal_partition_restart.cpp new file mode 100644 index 000000000..841b834ee --- /dev/null +++ b/tests/cpp/integration/amr/test_temporal_partition_restart.cpp @@ -0,0 +1,155 @@ +#include + +#include "explicit_amr_program.hpp" + +#include +#include +#include +#include + +#include +#include +#include +#include + +#if defined(POPS_HAS_KOKKOS) +#include +#endif + +using namespace pops; +using namespace pops::runtime::program; + +namespace { + +CellTemporalPartitionAcceptedState cell_local_state(std::uint64_t topology_epoch = 7) { + CellTemporalPartitionAcceptedState state; + state.kind = TemporalPartitionKind::CellLocal; + state.provider_identity = "test.temporal-partition.batched-cells@1"; + state.topology_epoch = topology_epoch; + state.synchronization_tick = 8; + state.tick_denominator = 16; + state.cells = {{0, 10, 0, 8}, {0, 11, 1, 8}, {1, 20, 2, 8}}; + return state; +} + +ModelSpec exb_spec() { + ModelSpec spec; + spec.transport = "exb"; + spec.source = "none"; + spec.elliptic = "charge"; + return spec; +} + +} // namespace + +TEST(test_temporal_partition_restart, batched_attempt_commit_and_rollback_are_exact) { + const CellTemporalPartitionAcceptedState accepted = cell_local_state(); + BatchedCellTemporalPartition partition(accepted); + + partition.begin_attempt(16); + partition.advance_batch(0, {0}, 12); + partition.advance_batch(0, {0}, 16); + EXPECT_THROW(partition.require_barrier("field solve"), std::logic_error); + partition.rollback(); + EXPECT_EQ(partition.checkpoint(), accepted); + + partition.begin_attempt(16); + partition.advance_batch(0, {0}, 16); + partition.advance_batch(1, {1}, 16); + partition.advance_batch(2, {2}, 16); + EXPECT_NO_THROW(partition.require_barrier("output")); + partition.commit(); + + const CellTemporalPartitionAcceptedState committed = partition.checkpoint(); + EXPECT_EQ(committed.synchronization_tick, 16); + for (const CellTemporalPartitionRecord& cell : committed.cells) + EXPECT_EQ(cell.accepted_tick, 16); + const auto manifest = partition.manifest(); + ASSERT_EQ(manifest.size(), 4u); + EXPECT_EQ(manifest[0][1], "cell_local"); + EXPECT_EQ(manifest[0][6], "3"); + EXPECT_EQ(manifest[1], (std::vector{"rung", "0", "1"})); + EXPECT_EQ(manifest[2], (std::vector{"rung", "1", "1"})); + EXPECT_EQ(manifest[3], (std::vector{"rung", "2", "1"})); +} + +TEST(test_temporal_partition_restart, malformed_state_and_batches_fail_before_mutation) { + const CellTemporalPartitionAcceptedState accepted = cell_local_state(); + BatchedCellTemporalPartition partition(accepted); + + CellTemporalPartitionAcceptedState unsynchronized = accepted; + unsynchronized.cells[1].accepted_tick = 6; + EXPECT_THROW(partition.restore(unsynchronized), std::invalid_argument); + EXPECT_EQ(partition.checkpoint(), accepted); + + partition.begin_attempt(16); + EXPECT_THROW(partition.advance_batch(0, {0, 0}, 12), std::invalid_argument); + EXPECT_THROW(partition.advance_batch(0, {1}, 12), std::invalid_argument); + EXPECT_THROW(partition.advance_batch(2, {2}, 10), std::invalid_argument); + partition.rollback(); + EXPECT_EQ(partition.checkpoint(), accepted); + + EXPECT_THROW(partition.require_global_execution_route(), std::logic_error); + EXPECT_NO_THROW(BatchedCellTemporalPartition().require_global_execution_route()); +} + +TEST(test_temporal_partition_restart, accepted_image_round_trips_canonically) { + AmrProgramAcceptedState accepted; + accepted.temporal_partition = cell_local_state(); + + const std::vector encoded = serialize_amr_program_accepted_state(accepted); + const AmrProgramAcceptedState decoded = deserialize_amr_program_accepted_state(encoded); + EXPECT_EQ(decoded.temporal_partition, accepted.temporal_partition); + EXPECT_EQ(serialize_amr_program_accepted_state(decoded), encoded); + + CellTemporalPartitionAcceptedState duplicate = accepted.temporal_partition; + duplicate.cells[1].cell = duplicate.cells[0].cell; + accepted.temporal_partition = duplicate; + EXPECT_THROW(serialize_amr_program_accepted_state(accepted), std::invalid_argument); +} + +TEST(test_temporal_partition_restart, + strict_amr_restore_consumes_manifest_and_refuses_global_step_bypass) { +#if defined(POPS_HAS_KOKKOS) + Kokkos::ScopeGuard guard; +#endif + AmrSystemConfig config; + config.n = 4; + config.L = 1.0; + config.regrid_every = 0; + config.periodicity = {true, true}; + + AmrSystem system(config); + system.add_block("tracer", exb_spec(), "none", "rusanov", "conservative", "explicit", 1); + test::install_forward_euler_program(system); + system.step(0.01); + + AmrProgramAcceptedState accepted = + deserialize_amr_program_accepted_state(system.program_accepted_state()); + accepted.temporal_partition = cell_local_state(system.engine()->topology_epoch()); + const std::vector cell_local_image = serialize_amr_program_accepted_state(accepted); + system.restore_checkpoint_accepted_state(cell_local_image); + + const auto manifest = system.program_temporal_partition_manifest(); + ASSERT_EQ(manifest.size(), 4u); + EXPECT_EQ(manifest[0][1], "cell_local"); + EXPECT_EQ(manifest[0][2], "test.temporal-partition.batched-cells@1"); + + const double time_before = system.time(); + const int step_before = system.macro_step(); + const std::vector bytes_before = system.program_accepted_state(); + EXPECT_THROW(system.step(0.01), std::logic_error) + << "an authenticated cell-local schedule cannot degrade to the global AMR driver"; + EXPECT_DOUBLE_EQ(system.time(), time_before); + EXPECT_EQ(system.macro_step(), step_before); + EXPECT_EQ(system.program_accepted_state(), bytes_before); + EXPECT_EQ(system.program_temporal_partition_manifest(), manifest); + + AmrProgramAcceptedState wrong_topology = accepted; + ++wrong_topology.temporal_partition.topology_epoch; + EXPECT_THROW(system.restore_checkpoint_accepted_state( + serialize_amr_program_accepted_state(wrong_topology)), + std::runtime_error); + EXPECT_EQ(system.program_accepted_state(), bytes_before) + << "rejected restore must not replace the accepted image"; +} diff --git a/tests/cpp/test_durations.json b/tests/cpp/test_durations.json index 855ccd973..29032f556 100644 --- a/tests/cpp/test_durations.json +++ b/tests/cpp/test_durations.json @@ -6,7 +6,8 @@ "test_amr_program_diffusion", "test_amr_program_positivity_floor", "test_flux_failure_loader_transaction", - "test_interface_flux_fragment_ledger" + "test_interface_flux_fragment_ledger", + "test_temporal_partition_restart" ], "estimate_policy": "new unmeasured targets use a conservative analogous-target estimate, falling back to the catalog median until the next CTest timing refresh", "measured_refresh": { @@ -16,7 +17,7 @@ "refresh_source_run": "30190778708", "source_job": "87152034744", "source_run": "29352485297", - "target_count": 187, + "target_count": 188, "unit_seconds": "aggregate CTest wall time per build target" }, "test_adaptive_multirate": 0.02, @@ -191,6 +192,7 @@ "test_step_attempt_rejected_header_only": 0.01, "test_structured_solver_diagnostics": 0.01, "test_sync_residence": 0.01, + "test_temporal_partition_restart": 0.05, "test_system_abstraction": 0.01, "test_system_coupler": 0.01, "test_system_hardening": 0.01, diff --git a/tests/cpp/test_sources.cmake b/tests/cpp/test_sources.cmake index 8dee480e5..76106bb50 100644 --- a/tests/cpp/test_sources.cmake +++ b/tests/cpp/test_sources.cmake @@ -55,6 +55,7 @@ set(POPS_CPP_TEST_SOURCE_test_capability_report "tests/cpp/integration/runtime/t set(POPS_CPP_TEST_SOURCE_test_canonical_identity "tests/cpp/unit/core/test_canonical_identity.cpp") set(POPS_CPP_TEST_SOURCE_test_cf_interface "tests/cpp/integration/amr/test_cf_interface.cpp") set(POPS_CPP_TEST_SOURCE_test_program_reflux_ledger "tests/cpp/integration/amr/test_program_reflux_ledger.cpp") +set(POPS_CPP_TEST_SOURCE_test_temporal_partition_restart "tests/cpp/integration/amr/test_temporal_partition_restart.cpp") set(POPS_CPP_TEST_SOURCE_test_cfl_dt "tests/cpp/unit/numerics/test_cfl_dt.cpp") set(POPS_CPP_TEST_SOURCE_test_checkpoint_cache "tests/cpp/integration/runtime/test_checkpoint_cache.cpp") set(POPS_CPP_TEST_SOURCE_test_checkpoint_history "tests/cpp/integration/runtime/test_checkpoint_history.cpp") diff --git a/tests/python/unit/runtime/test_amr_checkpoint_contract.py b/tests/python/unit/runtime/test_amr_checkpoint_contract.py index d4c495fd4..f4ce7c36d 100644 --- a/tests/python/unit/runtime/test_amr_checkpoint_contract.py +++ b/tests/python/unit/runtime/test_amr_checkpoint_contract.py @@ -16,6 +16,7 @@ ) from pops.output._checkpoint_collective import restore_checkpoint_payload from pops.runtime._amr_checkpoint_contract import ( + checkpoint_temporal_partition_kind, contract_for, encode_contract, preflight_contract, @@ -102,6 +103,19 @@ def program_accepted_state_manifest(self): def program_clock_manifest(self): return [["level", "0", "4", "0", "1", "0.4"], ["logical", "clock.macro", "4"]] + def program_temporal_partition_manifest(self): + return [ + [ + "summary", + "global", + "pops.temporal-partition.global@1", + "0", + "0", + "1", + "0", + ] + ] + def program_flux_ledger_manifest(self): return [ [ @@ -176,7 +190,7 @@ def _payload(sim=None): def test_contract_names_guarantee_relations_qualified_histories_and_transfer_plans(): contract = contract_for(_Sim()) - assert contract["schema_version"] == 4 + assert contract["schema_version"] == 5 assert contract["guarantee"] == "bit_identical_accepted_state" assert contract["ledger"]["accepted_entries"] == 1 assert contract["ledger"]["transaction_depth"] == 0 @@ -210,6 +224,10 @@ def test_contract_names_guarantee_relations_qualified_histories_and_transfer_pla ] assert contract["transfer_routes"][0][2:5] == ["route.u", "provider.u", "kernel.linear"] assert contract["clocks"][1] == ["logical", "clock.macro", "4"] + assert contract["temporal_partition"][0][1:3] == [ + "global", + "pops.temporal-partition.global@1", + ] assert [row[3] for row in contract["synchronization"]] == ["reflux", "average_down"] @@ -219,6 +237,25 @@ def test_preflight_returns_exact_native_payload_and_counters(): assert (regrids, epoch) == (4, 7) +def test_checkpoint_temporal_partition_kind_is_strict_and_data_only(): + payload = _payload() + assert checkpoint_temporal_partition_kind(payload) == "global" + + data = json.loads(str(payload["amr_accepted_contract"])) + data["temporal_partition"] = [ + ["summary", "cell_local", "test.partition@1", "7", "8", "16", "2"], + ["rung", "0", "1"], + ["rung", "1", "1"], + ] + payload["amr_accepted_contract"] = np.array(json.dumps(data)) + assert checkpoint_temporal_partition_kind(payload) == "cell_local" + + data["temporal_partition"].append(["not-a-rung"]) + payload["amr_accepted_contract"] = np.array(json.dumps(data)) + with pytest.raises(ValueError, match="invalid rung row"): + checkpoint_temporal_partition_kind(payload) + + @pytest.mark.parametrize("mutation", ["ratio", "route", "guarantee"]) def test_preflight_refuses_any_static_provenance_mismatch(mutation): payload = _payload() @@ -241,6 +278,7 @@ def test_preflight_refuses_any_static_provenance_mismatch(mutation): "clocks", "ledger", "interface_ledger", + "temporal_partition", "synchronization", ], ) @@ -251,6 +289,8 @@ def test_dynamic_contract_is_checked_after_the_opaque_state_is_restored(section) data[section][0][1] = "program.block.1" elif section in {"ledger", "interface_ledger"}: data[section]["accepted_entries"] += 1 + elif section == "temporal_partition": + data[section][0][1] = "cell_local" else: data[section].append(["tampered"]) payload["amr_accepted_contract"] = np.array(json.dumps(data)) @@ -344,8 +384,7 @@ def program_sync_manifest(self): } assert ( - receipt["history_consensus_identity_before"] - != receipt["history_consensus_identity_after"] + receipt["history_consensus_identity_before"] != receipt["history_consensus_identity_after"] ) # Phase-local all-rank consensus is the contract: interpolation may legitimately change the # dense history image while conserved solution components are checked independently. @@ -437,14 +476,19 @@ def __getitem__(self, key): ], ) def test_uniform_and_amr_payload_versions_are_exact_current_integer_scalars( - runtime_kind, key, expected, + runtime_kind, + key, + expected, ): - assert require_exact_payload_version( - {key: np.array(expected, dtype=np.int64)}, - key=key, - expected=expected, - runtime_kind=runtime_kind, - ) == expected + assert ( + require_exact_payload_version( + {key: np.array(expected, dtype=np.int64)}, + key=key, + expected=expected, + runtime_kind=runtime_kind, + ) + == expected + ) for incompatible in ( np.array(True), @@ -487,7 +531,9 @@ def test_uniform_and_amr_payload_versions_are_exact_current_integer_scalars( ], ) def test_historical_version_refusal_happens_before_restart_transaction( - runtime_kind, key, expected, + runtime_kind, + key, + expected, ): calls = [] diff --git a/tests/python/unit/runtime/test_program_report.py b/tests/python/unit/runtime/test_program_report.py index 005d933eb..1d948b959 100644 --- a/tests/python/unit/runtime/test_program_report.py +++ b/tests/python/unit/runtime/test_program_report.py @@ -4,6 +4,7 @@ executor. These unit checks exercise the single report owner directly: no legacy ``System`` is constructed and no native state array is read. """ + from __future__ import annotations import json @@ -82,8 +83,14 @@ def checkpoint_temporal_relations(self): return [(0, 1, 1, 2, "exact")] def program_flux_ledger_manifest(self): - return [("fluid", "U", "rhs", "transport", 1, 4, 1, 2, 1, 2, - "outward", 0.25, 0.125)] + return [("fluid", "U", "rhs", "transport", 1, 4, 1, 2, 1, 2, "outward", 0.25, 0.125)] + + def program_temporal_partition_manifest(self): + return [ + ("summary", "cell_local", "test.partition@1", 7, 16, 32, 5), + ("rung", 0, 3), + ("rung", 1, 2), + ] def program_sync_manifest(self): return [(0, 1, 0, "reflux", 4, 1, 2)] @@ -108,6 +115,7 @@ def test_empty_authority_produces_an_honest_empty_report(): assert report.level_relations == [] assert report.flux_ledger == [] assert report.synchronization == [] + assert report.temporal_partition == {} assert report.temporal == {} assert report.profiler == {"enabled": None} @@ -123,23 +131,44 @@ def test_accepted_program_report_preserves_owned_metadata(): assert report.params[0]["count"] == 2 assert report.params[0]["limit"] > 0 assert report.diagnostics == {"mass": 3.5} - assert report.histories == [{ - "name": "u_prev", "depth": 2, "ncomp": 3, "initialized": True, - }] - assert report.cache == [{ - "node_id": 7, - "name": "stage_rhs", - "last_update_step": 4, - "accumulated_dt": 0.125, - }] + assert report.histories == [ + { + "name": "u_prev", + "depth": 2, + "ncomp": 3, + "initialized": True, + } + ] + assert report.cache == [ + { + "node_id": 7, + "name": "stage_rhs", + "last_update_step": 4, + "accumulated_dt": 0.125, + } + ] assert report.clocks == [ {"kind": "logical", "clock": "main", "tick": 4}, - {"kind": "level", "level": 1, "macro_step": 4, - "phase": {"numerator": 1, "denominator": 2}, "physical_time": 0.5}, + { + "kind": "level", + "level": 1, + "macro_step": 4, + "phase": {"numerator": 1, "denominator": 2}, + "physical_time": 0.5, + }, ] assert report.level_relations[0]["remainder_policy"] == "exact" assert report.flux_ledger[0]["flux"] == "transport" assert report.synchronization[0]["phase"] == "reflux" + assert report.temporal_partition == { + "kind": "cell_local", + "provider_identity": "test.partition@1", + "topology_epoch": 7, + "synchronization_tick": 16, + "tick_denominator": 32, + "cell_count": 5, + "rungs": [{"rung": 0, "cells": 3}, {"rung": 1, "cells": 2}], + } assert report.temporal == {"schema_version": 1, "accepted_step": 4} @@ -147,7 +176,7 @@ def test_report_serialization_is_array_free_and_detached(): report = build_program_report(_AcceptedProgramAuthority()) data = report.to_dict() - assert data["schema_version"] == 3 + assert data["schema_version"] == 4 assert data["report_type"] == "program_runtime" assert json.loads(report.to_json()) == data assert "accepted-program" in str(report) diff --git a/tests/test_manifest.toml b/tests/test_manifest.toml index 3b55de01e..eddf28dd6 100644 --- a/tests/test_manifest.toml +++ b/tests/test_manifest.toml @@ -1134,6 +1134,11 @@ name = "test_program_reflux_ledger" sources = ["tests/cpp/integration/amr/test_program_reflux_ledger.cpp"] labels = ["integration", "amr", "medium"] +[[cpp.suite]] +name = "test_temporal_partition_restart" +sources = ["tests/cpp/integration/amr/test_temporal_partition_restart.cpp"] +labels = ["integration", "runtime", "amr", "medium"] + [[cpp.suite]] name = "test_residual_operator" sources = ["tests/cpp/unit/runtime/test_residual_operator.cpp"] From 653f582442453b44f0adde09689398bd4821c372 Mon Sep 17 00:00:00 2001 From: desp0042 Date: Sun, 2 Aug 2026 00:33:54 +0200 Subject: [PATCH 3/3] docs(time): bound the ADC-756 restart foundation --- docs/design/temporal-execution-contract.md | 28 ++++++++++++++++++++++ 1 file changed, 28 insertions(+) diff --git a/docs/design/temporal-execution-contract.md b/docs/design/temporal-execution-contract.md index 6e25499c8..ac83e90c9 100644 --- a/docs/design/temporal-execution-contract.md +++ b/docs/design/temporal-execution-contract.md @@ -61,6 +61,34 @@ program schedule with the installed program before native state mutation and req checkpointed step strategy for the next attempt. Schema v1 and other historical payloads require an offline migration; runtime restart contains no compatibility branch. +## Cell-local temporal-partition restart foundation + +The AMR Program accepted image now has an explicit temporal-partition section. Its cell-local form +stores a prepared-provider identity, hierarchy topology epoch, integer synchronization tick and tick +denominator, plus canonically ordered `(level, cell, rung, accepted_tick)` records. Floating-point +cell clocks and rank-local addresses are not checkpoint authority. Every persisted cell must be at +the same rung-aligned synchronization tick; a provisional attempt cannot be serialized. + +`BatchedCellTemporalPartition` supplies the execution-provider-independent transaction semantics: +one attempt target, ordered same-rung batches, synchronization barriers, commit, rollback, strict +restore and an accepted-state manifest. The AMR restart path decodes and validates this state before +replacing accepted bytes. A malformed provider identity or topology/level mismatch leaves the +previous image untouched. +The public Program report consumes the same native image and exposes provider identity, accepted +tick, denominator, cell count and per-rung counts. + +For this bounded slice, a cell-local checkpoint restarts only with the recorded MPI cardinality and +`RestoreRecordedHierarchy`. Rank-change and `RegridOnRestart` are rejected during Python preflight, +before the native restart transaction, because rematerializing canonical cell ids onto a new owner +or topology is not implemented yet. + +This foundation deliberately does not pretend that the existing global AMR driver is cell-local +stepping. If a cell-local image reaches that driver, execution fails before the Program body instead +of silently falling back to a global `dt`. ADC-707/ADC-708 still own the prepared patch/task graph; +ADC-756 still requires Kokkos rung batches, actual local-stage boundary evaluation, time-integrated +same-level/MPI/coarse-fine flux ledgers, regrid/rank-change rematerialization, device determinism and +performance evidence. None of those execution or conservation claims is made by this restart slice. + Offline envelope inspection authenticates only the integrity of a canonical checkpoint; it is not a migration. The frozen release-v2 Uniform checkpoint predates the envelope and omits lifecycle identities, temporal state, consumer cursors, and field-provider state. The explicit