refactor(output): isolate ROOT gathers on run-owned MPI lanes (ADC-683)

docs/design/SPECIFICATION_TECHNIQUE_FINALE_POPS_ARCHITECTURE.md

@@ -1491,11 +1491,13 @@ scientifiques choisissent obligatoirement un `ParallelMode` typé :
 d'un unique writer rang 0, `COLLECTIVE` pour les hyperslabs HDF5 MPIO exacts, ou `PER_RANK` pour des
 artefacts locaux qualifiés par rang et un reçu agrégé. Le mode, le format, la sélection, la cible et
 l'identité de chaque pièce native (`global_box_index`, `owner_rank`, `replicated`) sont authentifiés
-entre rangs avant toute écriture. La route `COLLECTIVE` appelle le backend C++ HDF5 parallèle avec
-la lane MPI dupliquée possédée par la session observateur ; le writer ne redécouvre ni n'emprunte
-`MPI_COMM_WORLD`. `h5py` reste uniquement un lecteur/écrivain série optionnel et n'est jamais un
-transport MPI. Une dépendance HDF5 parallèle native absente, un mode incompatible ou un backend
-Kokkos GPU/device handle non supporté est refusé avant le
+entre rangs avant toute écriture. La capture native `ROOT` reçoit uniquement une lane consommateur
+dupliquée pour le run et la libère collectivement à sa fermeture ; les façades
+`System`/`AmrSystem` n'acceptent plus le singleton monde pour cette route. La route `COLLECTIVE`
+appelle le backend C++ HDF5 parallèle avec la lane MPI dupliquée possédée par la session observateur ;
+le writer ne redécouvre ni n'emprunte `MPI_COMM_WORLD`. `h5py` reste uniquement un
+lecteur/écrivain série optionnel et n'est jamais un transport MPI. Une dépendance HDF5 parallèle
+native absente, un mode incompatible ou un backend Kokkos GPU/device handle non supporté est refusé avant le
 constructeur de `System`/`AmrSystem`; aucune route série implicite ne remplace une demande MPI.
 
 Les maillages non structurés, mobiles/déformables ou changeant de topologie, de nouvelles familles de

docs/design/exact-output-consumers.md

@@ -82,8 +82,10 @@ count, target suffix, or writer availability:
 
 - `SERIAL` requires the proved serial `ExecutionContext` (rank 0, size 1) and one complete snapshot.
 - `ROOT` requires a distributed context. Every rank participates in the authenticated native
-  gather, but only rank 0 prepares, verifies and atomically publishes the single-file writer.
-  Preparation failures and the final receipt are broadcast to every participant.
+  gather over a run-scoped duplicated consumer lane, but only rank 0 prepares, verifies and
+  atomically publishes the single-file writer. The native `System`/`AmrSystem` output bridge
+  accepts only that owned lane, never the process-world singleton. Preparation failures and the
+  final receipt are broadcast to every participant.
 - `COLLECTIVE` requires a distributed context, an authenticated collective resource plan and the
   native C++ parallel-HDF5 provider. The observer runtime owns a duplicated MPI lane for the complete
   writer session; neither the Python writer nor the native HDF5 adapter borrows or rediscovers the
@@ -193,10 +195,11 @@ therefore write NPZ, HDF5 or the complete VTU/PVTU/PVD/state ParaView bundle. `q
 retained detached snapshots; a full queue deliberately applies backpressure.
 
 The selected format owns the topology. `SERIAL` uses the sole rank. `ROOT` performs the complete
-snapshot gather on the main execution path, then writes from the rank-zero worker without worker
-MPI. `PER_RANK` and `COLLECTIVE` run one worker per rank over a run-scoped communicator duplicated
-collectively before any worker starts. That private lane has a distinct MPI context from
-`MPI_COMM_WORLD`, so numerical and output collective orderings cannot alias. PoPS requires
+snapshot gather on the main execution path over one run-scoped duplicated consumer lane, then
+writes from the rank-zero worker without MPI. `PER_RANK` and `COLLECTIVE` run one worker per rank
+over a run-scoped communicator duplicated collectively before any worker starts. Those private
+lanes have distinct MPI contexts from `MPI_COMM_WORLD`, so numerical and output collective
+orderings cannot alias. PoPS requires
 `MPI_THREAD_MULTIPLE`, authenticates the lane on every worker call and fixes distributed
 `max_attempts` to one: retrying after entry into an MPI publication would not be safe. Supported mode
 combinations remain those of the format itself; in particular, ParaView has no `COLLECTIVE` mode and
@@ -385,9 +388,10 @@ re-emission; a rank-local `KeyboardInterrupt`/`SystemExit` cannot split collecti
 - HDF5 uses native datasets and `read_hdf5()` verification. Serial/root fields must be complete.
   Collective mode requires the compiled C++ parallel-HDF5 route before preparation; every rank
   writes its declared non-overlapping hyperslabs through the exact authenticated communicator and
-  the manifest authenticates all pieces. A synchronous consumer uses the execution communicator;
-  an asynchronous consumer uses its private duplicated worker lane. Python never emulates this
-  mode with a gather-to-root writer: the compiled provider owns the MPIO dataset transfers.
+  the manifest authenticates all pieces. The HDF5 session uses its private duplicated observer lane;
+  neither synchronous nor asynchronous publication borrows the process world. Python never
+  emulates this mode with a gather-to-root writer: the compiled provider owns the MPIO dataset
+  transfers.
   Partition validation scales with piece count rather than global cell count, and shared geometry
   is written once by rank zero. Unlike the default relayed PVTU topology, the single collective HDF5
   target is opened by every rank through parallel HDF5/MPI-IO and must therefore be genuinely

include/pops/parallel/comm.hpp

@@ -96,6 +96,17 @@ inline void require_mpi_success(int code, std::string_view operation) {
     throw_mpi_error(code, operation);
 }
 
+inline int chunk_capacity(int ranks) {
+  const int divisor = std::max(1, ranks);
+  return std::max(1, std::numeric_limits<int>::max() / divisor);
+}
+
+inline const char* chunk_pointer(const std::string& payload, unsigned long long offset, int count) {
+  if (count == 0)
+    return nullptr;
+  return payload.data() + static_cast<std::size_t>(offset);
+}
+
 inline bool comm_active_unlocked() noexcept {
   int initialized = 0;
   int finalized = 0;

include/pops/parallel/execution_lane.hpp

@@ -560,37 +560,115 @@ class ObserverMpiLane {
       throw std::out_of_range("observer collective root is outside the lane");
     const int me = lane.rank();
 
-    std::optional<std::vector<std::string>> result;
+    long length_overflow = 0;
+    if constexpr (sizeof(std::size_t) > sizeof(unsigned long long)) {
+      if (payload.size() > static_cast<std::size_t>(std::numeric_limits<unsigned long long>::max()))
+        length_overflow = 1;
+    }
+    if (all_reduce_max(length_overflow, lane) != 0)
+      throw std::overflow_error("consumer gather payload exceeds the MPI length domain");
+    const unsigned long long local_length = static_cast<unsigned long long>(payload.size());
+
+    std::vector<unsigned long long> lengths;
     long allocation_failed = 0;
     if (me == root) {
       try {
-        result.emplace(static_cast<std::size_t>(ranks));
+        lengths.resize(static_cast<std::size_t>(ranks), 0ULL);
       } catch (const std::bad_alloc&) {
         allocation_failed = 1;
       } catch (const std::length_error&) {
         allocation_failed = 1;
       }
     }
     if (all_reduce_max(allocation_failed, lane) != 0)
-      throw std::runtime_error("observer root could not allocate gathered results");
+      throw std::runtime_error("consumer root could not allocate gathered lengths");
+    detail::require_mpi_success(
+        MPI_Gather(&local_length, 1, MPI_UNSIGNED_LONG_LONG, me == root ? lengths.data() : nullptr,
+                   1, MPI_UNSIGNED_LONG_LONG, root, lane.native_handle()),
+        "MPI_Gather(consumer payload lengths)");
 
-    for (int source = 0; source < ranks; ++source) {
-      std::string source_payload;
-      long copy_failed = 0;
-      if (me == source) {
+    unsigned long long maximum_length = local_length;
+    detail::require_mpi_success(
+        MPI_Allreduce(MPI_IN_PLACE, &maximum_length, 1, MPI_UNSIGNED_LONG_LONG, MPI_MAX,
+                      lane.native_handle()),
+        "MPI_Allreduce(maximum consumer gather length)");
+
+    std::optional<std::vector<std::string>> result;
+    std::vector<int> counts;
+    std::vector<int> displacements;
+    allocation_failed = 0;
+    if (me == root) {
+      try {
+        result.emplace(static_cast<std::size_t>(ranks));
+        counts.resize(static_cast<std::size_t>(ranks), 0);
+        displacements.resize(static_cast<std::size_t>(ranks), 0);
+        for (int rank = 0; rank < ranks; ++rank) {
+          const unsigned long long length = lengths[static_cast<std::size_t>(rank)];
+          if (length > static_cast<unsigned long long>(std::numeric_limits<std::size_t>::max())) {
+            allocation_failed = 1;
+            break;
+          }
+          (*result)[static_cast<std::size_t>(rank)].resize(static_cast<std::size_t>(length));
+        }
+      } catch (const std::bad_alloc&) {
+        allocation_failed = 1;
+      } catch (const std::length_error&) {
+        allocation_failed = 1;
+      }
+    }
+    if (all_reduce_max(allocation_failed, lane) != 0)
+      throw std::runtime_error("consumer root could not allocate gathered payloads");
+
+    const int capacity = detail::chunk_capacity(ranks);
+    for (unsigned long long offset = 0; offset < maximum_length;
+         offset += static_cast<unsigned long long>(capacity)) {
+      int total = 0;
+      if (me == root) {
+        for (int rank = 0; rank < ranks; ++rank) {
+          const unsigned long long length = lengths[static_cast<std::size_t>(rank)];
+          const int count = offset < length
+                                ? static_cast<int>(std::min<unsigned long long>(
+                                      length - offset, static_cast<unsigned long long>(capacity)))
+                                : 0;
+          counts[static_cast<std::size_t>(rank)] = count;
+          displacements[static_cast<std::size_t>(rank)] = total;
+          total += count;
+        }
+      }
+      std::vector<char> round;
+      long round_allocation_failed = 0;
+      if (me == root) {
         try {
-          source_payload = payload;
+          round.resize(static_cast<std::size_t>(total));
         } catch (const std::bad_alloc&) {
-          copy_failed = 1;
+          round_allocation_failed = 1;
         } catch (const std::length_error&) {
-          copy_failed = 1;
+          round_allocation_failed = 1;
         }
       }
-      if (all_reduce_max(copy_failed, lane) != 0)
-        throw std::runtime_error("an observer rank could not stage its gather payload");
-      std::string received = broadcast_bytes(std::move(source_payload), source);
-      if (me == root)
-        (*result)[static_cast<std::size_t>(source)] = std::move(received);
+      if (all_reduce_max(round_allocation_failed, lane) != 0)
+        throw std::runtime_error("consumer root could not allocate a gathered chunk");
+      const int send_count =
+          offset < local_length
+              ? static_cast<int>(std::min<unsigned long long>(
+                    local_length - offset, static_cast<unsigned long long>(capacity)))
+              : 0;
+      detail::require_mpi_success(
+          MPI_Gatherv(detail::chunk_pointer(payload, offset, send_count), send_count, MPI_BYTE,
+                      me == root ? round.data() : nullptr, me == root ? counts.data() : nullptr,
+                      me == root ? displacements.data() : nullptr, MPI_BYTE, root,
+                      lane.native_handle()),
+          "MPI_Gatherv(consumer payload chunk)");
+      if (me != root)
+        continue;
+      for (int rank = 0; rank < ranks; ++rank) {
+        const int count = counts[static_cast<std::size_t>(rank)];
+        if (count == 0)
+          continue;
+        std::copy_n(
+            round.data() + displacements[static_cast<std::size_t>(rank)], count,
+            (*result)[static_cast<std::size_t>(rank)].data() + static_cast<std::size_t>(offset));
+      }
     }
     return result;
 #else

include/pops/parallel/world_communicator.hpp

@@ -60,17 +60,6 @@ inline int validated_collective_root(int root) {
   return root;
 }
 
-inline int chunk_capacity(int ranks) {
-  const int divisor = std::max(1, ranks);
-  return std::max(1, std::numeric_limits<int>::max() / divisor);
-}
-
-inline const char* chunk_pointer(const std::string& payload, unsigned long long offset, int count) {
-  if (count == 0)
-    return nullptr;
-  return payload.data() + static_cast<std::size_t>(offset);
-}
-
 #endif
 
 }  // namespace detail

include/pops/runtime/amr_system.hpp

@@ -58,7 +58,7 @@
 
 namespace pops {
 
-class WorldCommunicator;
+class ObserverMpiLane;
 namespace runtime::program {
 class AmrProgramContext;
 }
@@ -668,7 +668,7 @@ class AmrSystem {
   /// Exact rank-local valid-cell pieces for one qualified field provider.  The returned metadata
   /// explicitly marks replicated level-zero ownership so output modes never infer it from box counts.
   std::vector<OutputPiece> output_field_local_pieces(const std::string& provider_slot, int level);
-  std::vector<OutputPiece> output_field_root_pieces(const WorldCommunicator& world,
+  std::vector<OutputPiece> output_field_root_pieces(const ObserverMpiLane& lane,
                                                     const std::string& provider_slot, int level);
   /// Transaction bracket used by the accepted-state reader after complete payload preflight. Every
   /// hierarchy,
@@ -1039,7 +1039,7 @@ class AmrSystem {
   /// without allocating a global level buffer.
   std::vector<OutputPiece> output_state_local_pieces(const std::string& name, int k);
   std::vector<PatchBox> output_geometry_boxes();
-  std::vector<OutputPiece> output_state_root_pieces(const WorldCommunicator& world,
+  std::vector<OutputPiece> output_state_root_pieces(const ObserverMpiLane& lane,
                                                     const std::string& name, int k);
   /// Owner rank per box of level @p k (the shared layout's DistributionMapping), aligned with the
   /// level-@p k rows of patch_boxes(). The v3 checkpoint (ADC-542) serializes it so a restart

include/pops/runtime/output_piece_collective.hpp

@@ -5,10 +5,10 @@
 ///
 /// Local providers are evaluated on every rank under an all-rank error consensus.  Metadata and
 /// IEEE-754 values are framed in a versioned, endian-stable native wire payload and transferred by
-/// WorldCommunicator's chunked MPI_Gatherv transport.  Only rank zero materializes the global piece
-/// vector; Python never gathers NumPy arrays or executes an MPI collective.
+/// an explicitly owned consumer lane. Only rank zero materializes the global piece vector; Python
+/// never gathers NumPy arrays or executes an MPI collective.
 
-#include <pops/parallel/world_communicator.hpp>
+#include <pops/parallel/execution_lane.hpp>
 #include <pops/runtime/output_piece.hpp>
 
 #include <algorithm>
@@ -185,13 +185,21 @@ inline std::string current_exception_text() {
 
 /// Evaluate a local OutputPiece provider and gather its exact result onto MPI rank zero.
 template <typename Provider>
-std::vector<OutputPiece> output_pieces_to_root(const WorldCommunicator& world,
+std::vector<OutputPiece> output_pieces_to_root(const ObserverMpiLane& lane,
                                                std::string operation_identity,
                                                Provider&& provider) {
-  world.require_active_mpi_world();
-  const int rank = world.rank();
+#ifndef POPS_HAS_MPI
+  (void)lane;
+  (void)operation_identity;
+  (void)provider;
+  throw std::runtime_error("native output-piece ROOT gather requires an MPI-enabled build");
+#endif
+  if (!lane.active())
+    throw std::runtime_error(
+        "native output-piece root gather requires an active consumer MPI lane");
+  const int rank = lane.rank();
 
-  const std::vector<std::string> operations = world.allgather_bytes(operation_identity);
+  const std::vector<std::string> operations = lane.allgather_bytes(operation_identity);
   if (!std::all_of(operations.begin(), operations.end(),
                    [&](const std::string& value) { return value == operation_identity; }))
     throw std::invalid_argument("output-piece root gather arguments differ across MPI ranks");
@@ -215,19 +223,19 @@ std::vector<OutputPiece> output_pieces_to_root(const WorldCommunicator& world,
     local_error = detail::current_exception_text();
   }
 
-  const std::vector<std::string> errors = world.allgather_bytes(local_error);
+  const std::vector<std::string> errors = lane.allgather_bytes(local_error);
   for (std::size_t source = 0; source < errors.size(); ++source) {
     if (!errors[source].empty())
       throw std::runtime_error("native output-piece provider failed on rank " +
                                std::to_string(source) + ": " + errors[source]);
   }
 
-  const std::optional<std::vector<std::string>> gathered = world.gather_bytes(packed, 0);
+  const std::optional<std::vector<std::string>> gathered = lane.gather_bytes(packed, 0);
   std::vector<OutputPiece> result;
   std::string root_error;
   if (rank == 0) {
     try {
-      if (!gathered || gathered->size() != static_cast<std::size_t>(world.size()))
+      if (!gathered || gathered->size() != static_cast<std::size_t>(lane.size()))
         throw std::runtime_error("native output-piece root gather has invalid rank cardinality");
       for (std::size_t source = 0; source < gathered->size(); ++source) {
         std::vector<OutputPiece> decoded =
@@ -248,7 +256,7 @@ std::vector<OutputPiece> output_pieces_to_root(const WorldCommunicator& world,
       root_error = detail::current_exception_text();
     }
   }
-  root_error = world.broadcast_bytes(std::move(root_error), 0);
+  root_error = lane.broadcast_bytes(std::move(root_error), 0);
   if (!root_error.empty())
     throw std::runtime_error("native output-piece reconstruction failed: " + root_error);
   return result;

include/pops/runtime/system.hpp

@@ -47,7 +47,7 @@
 
 namespace pops {
 
-class WorldCommunicator;
+class ObserverMpiLane;
 class PreparedSystemLayoutTransfer;
 
 namespace component {
@@ -1333,9 +1333,9 @@ class System {
   std::vector<OutputPiece> output_field_local_pieces(const std::string& provider_slot, int level);
   /// Collective ROOT views.  Local provider errors are agreed before native MPI_Gatherv; only rank
   /// zero receives complete pieces and every non-root rank receives an empty vector.
-  std::vector<OutputPiece> output_state_root_pieces(const WorldCommunicator& world,
+  std::vector<OutputPiece> output_state_root_pieces(const ObserverMpiLane& lane,
                                                     const std::string& name, int level) const;
-  std::vector<OutputPiece> output_field_root_pieces(const WorldCommunicator& world,
+  std::vector<OutputPiece> output_field_root_pieces(const ObserverMpiLane& lane,
                                                     const std::string& provider_slot, int level);
   /// @}