[ADC-756] Persist cell temporal-partition restart authority

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

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<std::vector<std::string>> program_accepted_state_manifest() const;
   POPS_EXPORT std::vector<std::vector<std::string>> 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<std::vector<std::string>> program_temporal_partition_manifest() const;
   POPS_EXPORT std::vector<std::vector<std::string>> program_flux_ledger_manifest() const;
   POPS_EXPORT std::vector<std::vector<std::string>> program_interface_flux_ledger_manifest() const;
   POPS_EXPORT std::vector<std::vector<std::string>> program_sync_manifest() const;

include/pops/runtime/program/amr_program_checkpoint.hpp

@@ -14,6 +14,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_program_reflux.hpp>
+#include <pops/runtime/program/cell_temporal_partition.hpp>
 
 namespace pops::runtime::program {
 
@@ -52,6 +53,7 @@ struct AmrProgramSyncEvent {
 struct AmrProgramAcceptedState {
   std::vector<amr::ClockStamp> level_clocks;
   std::map<std::string, std::int64_t> logical_clock_ticks;
+  CellTemporalPartitionAcceptedState temporal_partition;
   /// Rank-independent canonical image of the runtime-owned AMR tagging hysteresis.
   std::vector<std::uint8_t> tagging_hysteresis_state;
   std::map<std::string, int> history_owners;
@@ -448,12 +450,25 @@ Map read_map(Reader& in, ReadValue&& read_value) {
 inline std::vector<std::uint8_t> 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<std::uint64_t>(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<std::uint8_t>& 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<decltype(state.logical_clock_ticks)>(in, [](Reader& r) { return r.i64(); });
+  if (carries_temporal_partition) {
+    const std::uint64_t kind = in.u64();
+    if (kind > static_cast<std::uint64_t>(TemporalPartitionKind::CellLocal))
+      throw std::runtime_error(
+          "invalid AMR Program accepted-state payload: unsupported temporal partition kind");
+    state.temporal_partition.kind = static_cast<TemporalPartitionKind>(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<std::map<std::string, int>>(in, [](Reader& r) { return r.i32(); });

include/pops/runtime/program/amr_program_context.hpp

@@ -1278,6 +1278,10 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
     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<AmrProgramContext> {
   }
 
   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<std::size_t>(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<AmrProgramContext> {
     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<AmrProgramContext> {
     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<AmrProgramContext> {
     std::uint64_t engine_topology_generation = 0;
     std::vector<std::uint8_t> program_accepted_state;
     std::uint64_t program_accepted_state_revision = 0;
+    CellTemporalPartitionAcceptedState temporal_partition;
     std::set<FluxKey> active_flux;
     std::map<FluxKey, EdgeFlux> flux;
     std::map<FluxKey, std::vector<FluxContribution>> flux_contributions;
@@ -1928,6 +1945,7 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
       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<AmrProgramContext> {
     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 <class Body>
@@ -3275,8 +3295,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,
@@ -3534,6 +3554,7 @@ class AmrProgramContext : public ProgramExecutionServices<AmrProgramContext> {
   mutable bool restart_regrid_prepared_ = false;
   mutable int automatic_regrid_macro_step_ = -1;
   mutable std::vector<amr::ClockStamp> 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_ =