Server.cpp

Go to the documentation of this file.

//
// Created by atezbas on 10/21/24.
//
 
#include "Server.h"
#include "../Global/SharedConfig.h"
#include "VisualizationServer.h"
#include "../Scheduler/Scheduler.h"
#include "../Global/DETCheckpoint.h"
#include "../Global/PEOutput.h"
#include "../Global/OutputMerge.h"
#include <sstream>
#include <iomanip>
#include <map>
#include <set>
#include <unordered_set>
#include <csignal>
#include <unistd.h>
#include "../Global/SimulationContext.h"
using namespace std;
 
// SIGUSR1 checkpoint request (for HPC wall-time limits)
static volatile sig_atomic_t mpi_checkpoint_requested = 0;
static void mpi_sigusr1_handler(int) { mpi_checkpoint_requested = 1; }
 
// SIGINT (Ctrl+C) interactive interrupt
static volatile sig_atomic_t mpi_sigint_received = 0;
static void mpi_sigint_handler(int) { mpi_sigint_received = 1; }
 
// Show interactive interrupt menu on rank 0, return user choice (1/2/3)
static int show_mpi_interrupt_menu(int seq_completed, int queue_depth, int busy_clients) {
    std::cout << "\n\n=== Simulation Interrupted (Ctrl+C) ===" << std::endl;
    std::cout << "  Sequences completed: " << seq_completed << std::endl;
    std::cout << "  Queue depth: " << queue_depth << std::endl;
    std::cout << "  Busy clients: " << busy_clients << std::endl;
    std::cout << std::endl;
    std::cout << "  1. Continue simulation" << std::endl;
    std::cout << "  2. Save checkpoint and exit" << std::endl;
    std::cout << "  3. Exit without saving" << std::endl;
    std::cout << std::endl;
    std::cout << "Choice [1]: " << std::flush;
 
    std::string line;
    if (!std::getline(std::cin, line) || line.empty()) return 1;
    if (line[0] == '2') return 2;
    if (line[0] == '3') return 3;
    return 1;
}
 
Server::Server(const int rank, int viz_port) {
    this->rank = rank;
    viz_port_ = viz_port;
    clients_busy = std::make_unique<int[]>(PG::size);
    for (int i = 0; i < PG::size; i++) clients_busy[i] = 0;
    num_free_clients = PG::size - 1;
    next_candidate = 1;
 
    // Start visualization server if requested
    if (viz_port_ > 0) {
        viz_state_ = std::make_shared<TreeState>();
        viz_state_->total_clients = PG::size - 1;
        viz_state_->start_time = MPI_Wtime();
        viz_server_ = std::make_unique<VisualizationServer>(viz_state_, viz_port_);
        viz_server_->start();
        PU::print_message(rank, "Visualization server at http://localhost:" + std::to_string(viz_port_), PG::log);
        // Also print to original stdout (rank 0 doesn't redirect)
        std::cout << "DET visualization: http://localhost:" << viz_port_ << std::endl;
    }
 
    g_sim.SScheduler.initiationForSimulation();
    PU::print_message(rank, "Server initiated..", PG::log);
    PU::print_message(rank, "Size of the MPI world: " + std::to_string(PG::size), PG::log);
}
 
Server::~Server() = default;
 
void Server::Shutdown()    {
    PU::print_message(0, "Sending shutdown signal to clients..", PG::log);
    for (int who = 1; who < PG::size; who++)
        MPI_Send (nullptr, 0, MPI_INT, who, PG::client_shutdown, MPI_COMM_WORLD);
    PU::print_message(0, "Shutdown signal sent to all clients..", PG::log);
    if (viz_server_) viz_server_->stop();
}
 
 
void Server::Simulation() // PG::rank 0 executes this
{
    PU::print_message(rank, "Started the simulation", PG::log);
 
    // The server deserializes crew snapshots (checkpoint gather, resume load)
    // but never runs RELAP5, so onFirstInit never registers the procedures
    // master here. Register it explicitly (idempotent).
    g_sim.Plant.initiationForSimulation();
    g_sim.Translator.initiationForSimulation();
    g_sim.Indicator.readInfoFromFile();  // procedures parser needs the panel maps
    // Procedures before the crew, same reason as Client.cpp: the reason
    // machine resolves procedure-step pointers during crew init, and rank 0
    // deserializes crew snapshots, so the master must already be registered.
    g_sim.All_procedure.initiationForSimulation();
    adsidac::set_shared_procedures(&g_sim.All_procedure);
    adsidac::set_shared_control_panel(&g_sim.Indicator);
    g_sim.Decision_Maker.initiationForSimulation();
    g_sim.Action_Taker.initiationForSimulation();
    adsidac::set_shared_kb(sim::Operator::ODM, &g_sim.Decision_Maker.knowledge_base);
    adsidac::set_shared_kb(sim::Operator::OAT, &g_sim.Action_Taker.knowledge_base);
    adsidac::set_shared_diagnosis_matrix(sim::Operator::ODM, &g_sim.Decision_Maker.diagnosis_matrix);
    adsidac::set_shared_diagnosis_matrix(sim::Operator::OAT, &g_sim.Action_Taker.diagnosis_matrix);
 
    // Register signal handlers
    signal(SIGUSR1, mpi_sigusr1_handler);
    signal(SIGINT, mpi_sigint_handler);
 
    // Occupancy trace: open once, header first. Off unless --occupancy-log given.
    if (!g_sim.config.occupancy_log.empty()) {
        occupancy_log_.open(g_sim.config.occupancy_log, std::ios::out | std::ios::trunc);
        if (occupancy_log_) {
            occupancy_t0_ = MPI_Wtime();
            occupancy_log_ << "elapsed_s,queue_depth,busy_workers,total_workers,sequences_completed\n";
        } else {
            PU::print_message(rank, "WARNING: cannot open occupancy log "
                + g_sim.config.occupancy_log + " -- continuing without it", PG::log);
        }
    }
 
    // Helper: push viz stats under lock (no-op if viz disabled), and record the
    // occupancy step function (no-op if --occupancy-log not given).
    auto update_viz_stats = [this]() {
        const int queue = static_cast<int>(tasks_queue.size());
        const int workers = PG::size - 1;
        const int busy = workers - num_free_clients;
 
        if (occupancy_log_ && (queue != last_queue_ || busy != last_busy_)) {
            occupancy_log_ << (MPI_Wtime() - occupancy_t0_) << ',' << queue << ','
                           << busy << ',' << workers << ',' << sequences_completed_ << '\n';
            last_queue_ = queue;
            last_busy_ = busy;
        }
 
        if (!viz_state_) return;
        std::lock_guard<std::mutex> lock(viz_state_->mu);
        viz_state_->sequences_completed = sequences_completed_;
        viz_state_->queue_depth = queue;
        viz_state_->num_free_clients = num_free_clients;
        viz_state_->memory_mb = get_rss_mb();
    };
 
    // Resume from checkpoint or start fresh
    if (!g_sim.config.resume_file.empty()) {
        dispatch_resume_tasks();
    } else {
        // Sequence 0 -> first free client
        int free_rank = find_next_free_client_round_robin();
        rank_current_sequence_[free_rank] = 0;
        PU::print_message(rank, "Number of free clients = " + std::to_string(num_free_clients), PG::log);
        PU::print_message(rank, "Sending the tag for initial simulation to Client " + std::to_string(free_rank), PG::log);
        MPI_Send (nullptr, 0, MPI_INT, free_rank, PG::client_simulate_this, MPI_COMM_WORLD);
        clients_busy[free_rank] = 1;
    }
    update_viz_stats();
 
    bool user_exit = false;
    while (true) {
        // Check for user interrupt (Ctrl+C)
        if (mpi_sigint_received) {
            mpi_sigint_received = 0;
            int busy = PG::size - 1 - num_free_clients;
            int choice = show_mpi_interrupt_menu(sequences_completed_,
                static_cast<int>(tasks_queue.size()), busy);
            if (choice == 2) {
                if (g_sim.config.checkpoint_dir.empty()) {
                    g_sim.config.checkpoint_dir = g_sim.SScheduler.getProjectDirectory() + "/checkpoint";
                    fs::create_directories(g_sim.config.checkpoint_dir);
                }
                checkpoint_pending_ = true;
                g_sim.config.checkpoint_stop = true;
                PU::print_message(rank, "Checkpoint+exit requested, waiting for "
                    + std::to_string(busy) + " clients to finish", PG::log);
            } else if (choice == 3) {
                PU::print_message(rank, "Exit requested by user", PG::log);
                // If an interval checkpoint was written this run, the outputs
                // are an intermediate state: keep natural PE names so that
                // checkpoint stays resumable.
                if (last_checkpoint_seq_ >= 0) stopped_at_checkpoint_ = true;
                user_exit = true;
                break;
            }
            // choice 1: re-register handler and continue
            signal(SIGINT, mpi_sigint_handler);
        }
 
        // Non-blocking probe — allows interrupt checks between messages
        int has_msg = 0;
        MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &has_msg, &status_server);
        if (!has_msg) {
            // Check termination conditions even during idle polling
            if (checkpoint_pending_ && num_free_clients == PG::size - 1) {
                write_checkpoint();
                checkpoint_pending_ = false;
                if (g_sim.config.checkpoint_stop) { stopped_at_checkpoint_ = true; break; }
            }
            if (tasks_queue.empty() && resume_pending_.empty() &&
                num_free_clients == PG::size - 1 && !checkpoint_pending_) {
                PU::print_message(0, "Finished the parallel simulation", PG::log);
                update_viz_stats();
                break;
            }
            usleep(1000);  // 1ms poll interval — negligible overhead
            continue;
        }
 
        if (status_server.MPI_TAG == PG::server_output_data) {
            // Receive output data (PE events, end states) — does NOT mark client free
            SequenceOutputData data;
            PU::receive_object(data, status_server.MPI_SOURCE,
                               PG::server_output_data, MPI_COMM_WORLD);
            if (viz_state_) {
                std::lock_guard<std::mutex> lock(viz_state_->mu);
                viz_state_->output_data[data.sequence_id] = std::move(data);
            }
 
        } else if (status_server.MPI_TAG == PG::server_done_with_report) {
            // The client sends a SequenceReport payload (sent as MPI_BYTE)
            SequenceReport report{};
            MPI_Recv(&report, sizeof(SequenceReport), MPI_BYTE,
                     status_server.MPI_SOURCE, PG::server_done_with_report,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            ++num_free_clients;
            clients_busy[status_server.MPI_SOURCE] = 0;
            ++sequences_completed_;
            PU::print_message(rank, "Sequences completed: " + std::to_string(sequences_completed_)
                + " (S" + std::to_string(report.sequence_id) + " from C["
                + std::to_string(status_server.MPI_SOURCE) + "])"
                + ", free_clients=" + std::to_string(num_free_clients)
                + ", queue=" + std::to_string(tasks_queue.size()), PG::log);
            if (viz_state_) {
                std::lock_guard<std::mutex> lock(viz_state_->mu);
                viz_state_->completed.push_back(report);
            }
 
        } else if (status_server.MPI_TAG == PG::server_done_simulating) {
            // Empty completion signal without a SequenceReport (no tree data).
            // Client.cpp reports completion with server_done_with_report instead.
            MPI_Recv(nullptr, 0, MPI_INT, status_server.MPI_SOURCE, PG::server_done_simulating,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            ++num_free_clients;
            clients_busy[status_server.MPI_SOURCE] = 0;
            ++sequences_completed_;
            PU::print_message(rank, "Sequences completed: " + std::to_string(sequences_completed_)
                + " (legacy, no report, from C[" + std::to_string(status_server.MPI_SOURCE) + "])"
                + ", free_clients=" + std::to_string(num_free_clients)
                + ", queue=" + std::to_string(tasks_queue.size()), PG::log);
 
        } else if (status_server.MPI_TAG == PG::server_send_complete) {
            MPI_Recv(nullptr, 0, MPI_INT, status_server.MPI_SOURCE, PG::server_send_complete,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            ++num_free_clients;
            clients_busy[status_server.MPI_SOURCE] = 0;
            PU::print_message(rank, "[PEER] Send complete from C[" + std::to_string(status_server.MPI_SOURCE)
                + "], free=" + std::to_string(num_free_clients), PG::log);
 
        } else if (status_server.MPI_TAG == PG::server_branch_meta) {
            int source = status_server.MPI_SOURCE;
            int local_id, num_branches;
            MPI_Recv(&local_id, 1, MPI_INT, source, PG::server_branch_meta,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            MPI_Recv(&num_branches, 1, MPI_INT, source, PG::server_branch_meta,
                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 
            // Parent for all children is whatever sequence the source is currently running
            int parent_seq = rank_current_sequence_.count(source) ? rank_current_sequence_[source] : -1;
 
            int needed = num_branches - 1;
            // A pending checkpoint must not hand work back out. This handler
            // assigns branches to free clients directly, outside the
            // `if (!checkpoint_pending_)` dispatch gate further down, so while
            // a checkpoint is pending it places nothing (direct = 0) and
            // queues every sibling branch. Otherwise retiring sequences would
            // re-arm the workers they free and `num_free_clients == size-1`
            // would never be reached. The donor still runs branch 0 itself,
            // so the drain converges once the running sequences finish.
            int direct = checkpoint_pending_ ? 0
                                             : std::min(needed, num_free_clients);
            int queued = needed - direct;
 
            PU::print_message(rank, "[PEER] C[" + std::to_string(source)
                + "] lid=" + std::to_string(local_id) + " branches=" + std::to_string(num_branches)
                + " parent_seq=" + std::to_string(parent_seq)
                + " -> direct=" + std::to_string(direct)
                + " queued=" + std::to_string(queued), PG::log);
 
            MPI_Send(&direct, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
 
            for (int i = 0; i < direct; i++) {
                int free_rank = find_next_free_client_round_robin();
                int branch_index = i + 1;
                int seq_id = sequence_counter_++;
 
                MPI_Send(&free_rank, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
                MPI_Send(&branch_index, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
                MPI_Send(&seq_id, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
                MPI_Send(&parent_seq, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
 
                rank_current_sequence_[free_rank] = seq_id;
 
                PU::print_message(rank, "[PEER] Direct: C[" + std::to_string(source)
                    + "] -> C[" + std::to_string(free_rank) + "] bi="
                    + std::to_string(branch_index) + " seq=" + std::to_string(seq_id), PG::log);
            }
 
            MPI_Send(&queued, 1, MPI_INT, source, PG::client_peer_assign, MPI_COMM_WORLD);
 
            for (int bi = direct + 1; bi < num_branches; ++bi) {
                RestartTask task{};
                task.source_rank  = source;
                task.local_id     = local_id;
                task.sequence_id  = sequence_counter_++;
                task.branch_index = bi;
                task.parent_sequence_id = parent_seq;
                tasks_queue.push(task);
            }
 
            PU::print_message(rank, "[PEER] Done: free=" + std::to_string(num_free_clients)
                + " queue=" + std::to_string(tasks_queue.size()), PG::log);
        }
 
        // Check checkpoint conditions. last_checkpoint_seq_ advances at every
        // write, so the interval condition arms once per interval and not
        // once per server message at the same sequence count.
        if (!checkpoint_pending_ && !g_sim.config.checkpoint_dir.empty()) {
            // Threshold, not modulo. This condition is evaluated inside a
            // message handler, so a `% interval == 0` test fires only if the
            // counter lands exactly on a multiple at the instant that handler
            // runs, which many workers reporting concurrently do not
            // guarantee. A >= comparison against the last checkpoint cannot
            // be skipped.
            if ((g_sim.config.checkpoint_interval > 0 && sequences_completed_ > 0 &&
                 sequences_completed_ - last_checkpoint_seq_ >= g_sim.config.checkpoint_interval) ||
                mpi_checkpoint_requested) {
                checkpoint_pending_ = true;
                mpi_checkpoint_requested = 0;
                PU::print_message(rank, "[CHECKPOINT] Pending — waiting for "
                    + std::to_string(PG::size - 1 - num_free_clients) + " clients to drain", PG::log);
            }
        }
 
        if (checkpoint_pending_ && num_free_clients == PG::size - 1) {
            write_checkpoint();
            last_checkpoint_seq_ = sequences_completed_;
            checkpoint_pending_ = false;
            if (g_sim.config.checkpoint_stop) {
                PU::print_message(rank, "[CHECKPOINT] --checkpoint-stop: exiting after "
                    + std::to_string(sequences_completed_) + " sequences", PG::log);
                stopped_at_checkpoint_ = true;
                break;
            }
        }
 
        if (!checkpoint_pending_) {
            // Resume tasks take priority (server holds data from checkpoint)
            while (!resume_pending_.empty() && num_free_clients > 0)
                dispatch_one_resume_task();
            // Then normal peer-based dispatch
            dispatch_available_tasks();
        }
        update_viz_stats();
 
        if (tasks_queue.empty() && resume_pending_.empty() && num_free_clients == PG::size - 1 && !checkpoint_pending_)
        {
            PU::print_message(0, "Finished the parallel simulation", PG::log);
            update_viz_stats();
            break;
        }
    }
}
 
int Server::find_next_free_client_round_robin()
{
    for (int i = 0; i < PG::size - 1; ++i) {
        int candidate = 1 + (next_candidate + i - 1) % (PG::size - 1);
        if (clients_busy[candidate] == 0) {
            clients_busy[candidate] = 1;
            --num_free_clients;
            next_candidate = candidate + 1;
            if (next_candidate >= PG::size) next_candidate = 1;
            return candidate;
        }
    }
    return -1;
}
 
int Server::find_free_client_except(int exclude_rank)
{
    for (int i = 0; i < PG::size - 1; ++i) {
        int candidate = 1 + (next_candidate + i - 1) % (PG::size - 1);
        if (clients_busy[candidate] == 0 && candidate != exclude_rank) {
            clients_busy[candidate] = 1;
            --num_free_clients;
            next_candidate = candidate + 1;
            if (next_candidate >= PG::size) next_candidate = 1;
            return candidate;
        }
    }
    return -1;
}
 
void Server::dispatch_available_tasks()
{
    // Scan queue for tasks whose source client is free.
    // Dispatch via peer send (source sends to another free client)
    // or self-run (source runs the task directly if no other client is free).
    size_t initial_size = tasks_queue.size();
    for (size_t i = 0; i < initial_size && num_free_clients > 0; i++) {
        RestartTask task = tasks_queue.front();
        tasks_queue.pop();
 
        if (clients_busy[task.source_rank] != 0) {
            tasks_queue.push(task);
            continue;
        }
 
        int receiver = find_free_client_except(task.source_rank);
        if (receiver != -1) {
            clients_busy[task.source_rank] = 1;
            --num_free_clients;
 
            MPI_Send(&task.local_id, 1, MPI_INT, task.source_rank, PG::client_send_to_peer, MPI_COMM_WORLD);
            MPI_Send(&receiver, 1, MPI_INT, task.source_rank, PG::client_send_to_peer, MPI_COMM_WORLD);
            MPI_Send(&task.branch_index, 1, MPI_INT, task.source_rank, PG::client_send_to_peer, MPI_COMM_WORLD);
            MPI_Send(&task.sequence_id, 1, MPI_INT, task.source_rank, PG::client_send_to_peer, MPI_COMM_WORLD);
            MPI_Send(&task.parent_sequence_id, 1, MPI_INT, task.source_rank, PG::client_send_to_peer, MPI_COMM_WORLD);
 
            rank_current_sequence_[receiver] = task.sequence_id;
 
            PU::print_message(rank, "[PEER] Deferred: C[" + std::to_string(task.source_rank)
                + "] -> C[" + std::to_string(receiver) + "] lid=" + std::to_string(task.local_id)
                + " bi=" + std::to_string(task.branch_index)
                + " seq=" + std::to_string(task.sequence_id), PG::log);
        } else {
            clients_busy[task.source_rank] = 1;
            --num_free_clients;
 
            MPI_Send(&task.local_id, 1, MPI_INT, task.source_rank, PG::client_run_held, MPI_COMM_WORLD);
            MPI_Send(&task.branch_index, 1, MPI_INT, task.source_rank, PG::client_run_held, MPI_COMM_WORLD);
            MPI_Send(&task.sequence_id, 1, MPI_INT, task.source_rank, PG::client_run_held, MPI_COMM_WORLD);
            MPI_Send(&task.parent_sequence_id, 1, MPI_INT, task.source_rank, PG::client_run_held, MPI_COMM_WORLD);
 
            rank_current_sequence_[task.source_rank] = task.sequence_id;
 
            PU::print_message(rank, "[PEER] Self-run: C[" + std::to_string(task.source_rank)
                + "] lid=" + std::to_string(task.local_id)
                + " bi=" + std::to_string(task.branch_index)
                + " seq=" + std::to_string(task.sequence_id), PG::log);
        }
    }
    PU::print_message(rank, "[MEM] RSS=" + std::to_string(get_rss_mb())
        + "MB, queue=" + std::to_string(tasks_queue.size())
        + ", free=" + std::to_string(num_free_clients), PG::log);
}
 
void Server::merge_outputs()
{
    const double t_merge = MPI_Wtime();
    // --merge=serial selects the rank-0 copy loop and the in-RAM PE rewrite
    // (escape hatch and equivalence oracle for the rename-based path).
    const bool legacy = (g_sim.config.merge_mode == 1);
    std::string dest_es  = g_sim.SScheduler.getESDirectory();
    std::string dest_pe  = g_sim.SScheduler.getPEDirectory();
    std::string dest_kps = g_sim.SScheduler.getKPSDirectory();
 
    // Create destination directories
    fs::create_directories(dest_es);
    fs::create_directories(dest_pe);
    fs::create_directories(dest_kps);
 
    // Under --merge=now the clients move their own ES/KPS files and stage the
    // append sources as parts inside their shutdown handler; wait for them.
    // Every rank parses the same argv, so the barrier count is consistent.
    // (merge_mode 2 = defer reaches this function only when a checkpoint-stop
    // forced an inline merge; the clients then staged nothing and entered no
    // barrier, so this must not either.)
    if (g_sim.config.merge_mode == 0) MPI_Barrier(MPI_COMM_WORLD);
 
    // Merge per-rank outputs into shared directory
    for (int r = 1; r < PG::size; ++r) {
        std::string rank_out = "../rank_" + std::to_string(r) + "/Output/";
        if (!fs::exists(rank_out)) continue;
 
        // Copy individual files — unique-by-ID files get copied, text logs get concatenated
        // The PE aggregate is append-only and self-delimiting, so
        // concatenating each rank's copy is exactly the right merge — without
        // this it would land on one shared path and ranks would clobber it.
        static const std::vector<std::string> append_files = {
            "Scenario info.txt", "Procedure data.txt", "Scan data.txt",
            adsidac::pe::aggregate_basename()
        };
        std::unordered_set<std::string> made_dirs;
        for (auto& entry : fs::recursive_directory_iterator(rank_out)) {
            if (!entry.is_regular_file()) continue;
            fs::path rel = fs::relative(entry.path(), rank_out);
            fs::path dest = fs::path(g_sim.SScheduler.getOutputDirectory()) / rel;
            if (made_dirs.insert(dest.parent_path().string()).second)
                fs::create_directories(dest.parent_path());
 
            std::string fname = entry.path().filename().string();
            bool should_append = false;
            for (auto& af : append_files)
                if (fname == af) { should_append = true; break; }
 
            if (should_append) {
                std::ifstream in(entry.path());
                std::ofstream out(dest, std::ios::app);
                out << in.rdbuf();
            } else if (legacy) {
                fs::copy(entry.path(), dest, fs::copy_options::overwrite_existing);
            } else {
                // Same filesystem by construction (rank_N and Output/ both live
                // under the project dir): a rename moves metadata, not bytes.
                // POSIX rename-over-existing gives colliding PE filenames the
                // same highest-rank-wins result as the --merge=serial copy loop.
                std::error_code mec;
                fs::rename(entry.path(), dest, mec);
                if (mec) {
                    // EXDEV or exotic filesystems: fall back to copy + remove.
                    mec.clear();
                    fs::copy(entry.path(), dest, fs::copy_options::overwrite_existing, mec);
                    if (mec) {
                        PU::print_message(rank, "[MERGE] failed to move " + entry.path().string()
                            + ": " + mec.message(), PG::log);
                    } else {
                        fs::remove(entry.path(), mec);
                    }
                }
            }
        }
    }
 
    // Ordered concatenation of the parts the clients staged (--merge=now).
    // The ES logs and err messages are truncated first — the same net
    // semantics as the per-run reset — while the PE aggregate destination is
    // APPEND-ONLY: it accumulates across checkpoint phases and collect()
    // depends on the accumulation.
    if (g_sim.config.merge_mode == 0) {
        const std::vector<std::pair<fs::path, fs::path>> part_sets = {
            {fs::path(dest_es) / "parts", fs::path(dest_es)},
            {fs::path(dest_pe) / "parts", fs::path(dest_pe)}};
        for (auto& [parts_dir, out_dir] : part_sets) {
            std::error_code pec;
            if (!fs::exists(parts_dir, pec)) continue;
            std::set<std::string> bases;
            for (auto& e : fs::directory_iterator(parts_dir, pec)) {
                const std::string f = e.path().filename().string();
                const auto pos = f.rfind(".rank_");
                if (pos != std::string::npos) bases.insert(f.substr(0, pos));
            }
            for (const auto& base : bases) {
                const bool is_agg = (base == adsidac::pe::aggregate_basename());
                std::ofstream out(out_dir / base,
                                  std::ios::binary | (is_agg ? std::ios::app : std::ios::trunc));
                for (int r = 1; r < PG::size; ++r) {
                    std::ifstream in(parts_dir / (base + ".rank_" + std::to_string(r)),
                                     std::ios::binary);
                    if (in) out << in.rdbuf();
                }
            }
            fs::remove_all(parts_dir, pec);
        }
        // The server's own err messages, if any (rank_0 cwd).
        std::error_code sec;
        if (fs::exists("err messages.txt", sec)) {
            std::ifstream in("err messages.txt", std::ios::binary);
            std::ofstream out(fs::path(dest_es) / "err messages.txt",
                              std::ios::binary | std::ios::app);
            out << in.rdbuf();
        }
    }
 
    // Build ES_summary.txt from all ZESNode files in the destination tree
    // (shared with --merge-only via Global/OutputMerge.h so the two paths
    // cannot diverge on parse or format).
    const size_t merged_seq_count = adsidac::merge::build_es_summary(dest_es);
 
    // Renumber PE files sequentially (clients use rank-offset IDs, need sequential for output).
    // ONLY at a true end of run. A --checkpoint-stop merge is an intermediate
    // state: files must keep their natural branch-ID names so a resumed run's
    // re-saves land on the same paths (overwrite = refresh, not
    // duplicate-or-clobber).
    size_t pe_count = 0;
    if (!stopped_at_checkpoint_) {
        // Collect from whichever layout the clients wrote, then rewrite with
        // sequential IDs in the layout this run is configured for. Both halves
        // live in Global/PEOutput.h so the serial merge in ADSIDAC.cpp cannot
        // drift from this one.
        pe_count = legacy
            ? adsidac::pe::rewrite_sequential(dest_pe, adsidac::pe::collect(dest_pe),
                                              g_sim.config.aggregate_pe_output)
            : adsidac::pe::finalize_sequential(dest_pe, g_sim.config.aggregate_pe_output);
    } else {
        PU::print_message(rank, "[CHECKPOINT] Intermediate stop — PE files keep natural IDs for resume", PG::log);
    }
 
    {
        std::ostringstream oss;
        oss << "Merged " << merged_seq_count << " sequences, " << pe_count
            << " PE nodes into " << g_sim.SScheduler.getOutputDirectory()
            << " in " << std::fixed << std::setprecision(1)
            << (MPI_Wtime() - t_merge) << " s ("
            << (legacy ? "serial copy" : "rename") << ")";
        PU::print_message(rank, oss.str(), PG::log);
    }
}
 
void Server::write_checkpoint()
{
    PU::print_message(rank, "[CHECKPOINT] Gathering held_data from " + std::to_string(PG::size - 1) + " clients", PG::log);
 
    // Gather held_data from all clients (serialize BranchRecord to blob immediately)
    struct GatheredEntry {
        int source_rank;
        int local_id;
        int cid;
        std::shared_ptr<RestartNodeInfo> rinfo;
        std::string branch_record_blob;
    };
    std::vector<GatheredEntry> all_held;
 
    for (int c = 1; c < PG::size; c++) {
        MPI_Send(nullptr, 0, MPI_INT, c, PG::client_checkpoint, MPI_COMM_WORLD);
        int count;
        MPI_Recv(&count, 1, MPI_INT, c, PG::client_checkpoint, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
        for (int i = 0; i < count; i++) {
            GatheredEntry entry{};
            entry.source_rank = c;
            MPI_Recv(&entry.local_id, 1, MPI_INT, c, PG::client_checkpoint, MPI_COMM_WORLD, MPI_STATUS_IGNORE);
            entry.rinfo = std::make_shared<RestartNodeInfo>();
            PU::receive_object(*entry.rinfo, c, PG::client_checkpoint, MPI_COMM_WORLD);
            BranchRecord br;
            PU::receive_object(br, c, PG::client_checkpoint, MPI_COMM_WORLD);
            // Serialize BranchRecord to blob immediately (BranchRecord is not movable)
            std::ostringstream oss(std::ios::binary);
            boost::archive::binary_oarchive oa(oss);
            oa << br;
            entry.branch_record_blob = oss.str();
            entry.cid = entry.rinfo->getCorrespondingBranchClusterID();
            all_held.push_back(std::move(entry));
        }
        PU::print_message(rank, "[CHECKPOINT] Client " + std::to_string(c) + ": " + std::to_string(count) + " entries", PG::log);
    }
 
    // Fix CID collisions: two workers may independently generate the same CID.
    // Renumber colliding CIDs so each held entry has a globally unique CID.
    {
        std::set<int> seen_cids;
        int next_unique_cid = 0;
        // Find max CID across all entries to start renumbering above it
        for (auto& entry : all_held)
            if (entry.cid > next_unique_cid) next_unique_cid = entry.cid;
        next_unique_cid += 1000;  // start well above any existing CID
 
        for (auto& entry : all_held) {
            if (seen_cids.count(entry.cid)) {
                int old_cid = entry.cid;
                int new_cid = next_unique_cid++;
                PU::print_message(rank, "[CHECKPOINT] CID collision: renumbering CID="
                    + std::to_string(old_cid) + " -> " + std::to_string(new_cid), PG::log);
 
                // Update the RestartNodeInfo's CID reference
                entry.rinfo->setCorrespondingBranchClusterID(new_cid);
 
                // Update the BranchCluster CID inside the serialized BranchRecord blob
                BranchRecord br;
                { std::istringstream iss(entry.branch_record_blob, std::ios::binary);
                  boost::archive::binary_iarchive ia(iss); ia >> br; }
                for (auto& bc : br.branch_cluster) {
                    if (bc->getIDOfThisBranchCluster() == old_cid)
                        bc->setIDOfThisBranchCluster(new_cid);
                }
                { std::ostringstream oss(std::ios::binary);
                  boost::archive::binary_oarchive oa(oss); oa << br;
                  entry.branch_record_blob = oss.str(); }
 
                entry.cid = new_cid;
            }
            seen_cids.insert(entry.cid);
        }
    }
 
    // Build DETCheckpoint
    DETCheckpoint ckpt;
    ckpt.sequences_completed = sequences_completed_;
    ckpt.sequence_counter = sequence_counter_;
 
    // Store restart states and branch record blobs per CID (collision-free after the renumbering above)
    for (auto& entry : all_held) {
        if (ckpt.restart_states.find(entry.cid) == ckpt.restart_states.end()) {
            ckpt.restart_states[entry.cid] = entry.rinfo;
            ckpt.branch_record_blobs[entry.cid] = entry.branch_record_blob;
        }
    }
 
    // Move each restart blob out of its RestartNodeInfo into source_blobs so
    // the archive stores it exactly once (MPI held entries are 1:1 with their
    // sources).
    for (auto& [cid, ri] : ckpt.restart_states) {
        if (ri && !ri->relap_restart_buf.empty()) {
            ckpt.source_blobs[cid] = std::move(ri->relap_restart_buf);
            ri->relap_restart_buf.clear();
            ri->relap_restart_buf.shrink_to_fit();
            ckpt.cid_source[cid] = cid;
        }
    }
 
    // Convert tasks_queue to pending_tasks with per-task data
    std::queue<RestartTask> temp = tasks_queue;
    while (!temp.empty()) {
        auto& task = temp.front();
        PendingBranch pb;
        pb.cid = 0;
        pb.branch_index = task.branch_index;
        pb.sequence_id = task.sequence_id;
        pb.parent_sequence_id = task.parent_sequence_id;
 
        std::shared_ptr<RestartNodeInfo> task_rinfo;
        for (auto& entry : all_held) {
            if (entry.source_rank == task.source_rank && entry.local_id == task.local_id) {
                pb.cid = entry.cid;
                task_rinfo = entry.rinfo;
                break;
            }
        }
        ckpt.pending_tasks.push_back(pb);
        ckpt.task_restart_states.push_back(task_rinfo);
        // task_branch_blobs stays empty: resume always uses the merged
        // branch_center_blob and nothing reads per-task copies.
        temp.pop();
    }
 
    // Chained resume: tasks inherited from a prior checkpoint that were never
    // dispatched this run must carry over, or their entire subtree is lost
    // from the DET. Their restart states/blobs still live in resume_ckpt_
    // (freed only at dispatch). Carried source blobs get negative keys so
    // they can never collide with this run's CID-keyed sources.
    if (resume_ckpt_ && !resume_pending_.empty()) {
        PU::print_message(rank, "[CHECKPOINT] Carrying over " + std::to_string(resume_pending_.size())
            + " undispatched inherited tasks", PG::log);
        for (size_t i = 0; i < resume_pending_.size(); ++i) {
            const auto& pb = resume_pending_[i];
            ckpt.pending_tasks.push_back(pb);
            std::shared_ptr<RestartNodeInfo> ri =
                (i < resume_ckpt_->task_restart_states.size()) ? resume_ckpt_->task_restart_states[i] : nullptr;
            ckpt.task_restart_states.push_back(ri);
            // Serial resume reads restart_states (CID-keyed), not the
            // per-task vector — register the carried rinfo there too.
            if (ri && !ckpt.restart_states.count(pb.cid))
                ckpt.restart_states[pb.cid] = ri;
            auto cs = resume_ckpt_->cid_source.find(pb.cid);
            if (cs != resume_ckpt_->cid_source.end()) {
                int carried_key = -(cs->second) - 1;
                ckpt.cid_source[pb.cid] = carried_key;
                auto sb = resume_ckpt_->source_blobs.find(cs->second);
                if (sb != resume_ckpt_->source_blobs.end() && !ckpt.source_blobs.count(carried_key))
                    ckpt.source_blobs[carried_key] = sb->second; // copy: this run continues
            }
        }
    }
 
    // Merge all client BranchRecords into one combined branch_center_blob.
    //
    // CIDs are per-lineage counters: every daughter lineage continues its
    // donor's counter, so two DIFFERENT physical clusters routinely share a
    // CID across gathered records. The merge therefore keys clusters by their
    // true identity: the branch-0 ID, minted exactly once at creation and
    // immutable. Same-CID clusters with different identities both survive;
    // pending-task targets keep their CID (task/rinfo alignment), displaced
    // non-targets are renumbered, and mother links are repaired via the
    // globally unique branch IDs. A CID alone does not identify a cluster: a
    // fully explored ancestor sharing a pending target's CID would displace
    // the target, and the resumed task would re-run an explored branch while
    // its real branch never ran.
    {
        std::map<int, int> pending_min_bi;
        for (auto& pt : ckpt.pending_tasks) {
            auto it = pending_min_bi.find(pt.cid);
            if (it == pending_min_bi.end() || pt.branch_index < it->second)
                pending_min_bi[pt.cid] = pt.branch_index;
        }
 
        BranchRecord merged;
        std::map<int, std::unique_ptr<BranchCluster>> best;  // identity -> best version
        std::map<int, int> reserved_cid;                     // identity -> pending target CID
 
        auto identity_of = [](const BranchCluster& bc) {
            return bc.all_branch.empty() ? -1 : bc.all_branch.front()->getIDOfThisBranch();
        };
        // Versions of one physical cluster differ in how many sibling branch
        // IDs they know (the explorer's copy has the reassigned ID, the
        // donor's still has the -999 placeholder). Keep the most complete.
        auto known_ids = [](const BranchCluster& bc) {
            int n = 0;
            for (auto& bp : bc.all_branch)
                if (bp && bp->getIDOfThisBranch() > 0) ++n;
            return n;
        };
        // Versions of one physical cluster carry complementary knowledge —
        // each explorer's copy knows its own branch's reassigned ID and child
        // links. Union them so the kept version is the most complete.
        auto merge_versions = [](BranchCluster& w, BranchCluster& l) {
            size_t n = std::min(w.all_branch.size(), l.all_branch.size());
            for (size_t i = 0; i < n; ++i) {
                auto* wb = w.all_branch[i].get();
                auto* lb = l.all_branch[i].get();
                if (!wb || !lb) continue;
                if (wb->getIDOfThisBranch() <= 0 && lb->getIDOfThisBranch() > 0)
                    wb->setIDOfThisBranch(lb->getIDOfThisBranch());
                for (int c : lb->child_IDs) {
                    bool have = false;
                    for (int wc : wb->child_IDs)
                        if (wc == c) { have = true; break; }
                    if (!have) wb->child_IDs.push_back(c);
                }
            }
        };
        auto absorb = [&](BranchRecord& br, int own_target_cid) {
            for (auto& bc : br.branch_cluster) {
                if (!bc) continue;
                int ident = identity_of(*bc);
                if (own_target_cid != -1 && bc->getIDOfThisBranchCluster() == own_target_cid
                    && !reserved_cid.count(ident))
                    reserved_cid[ident] = own_target_cid;
                auto it = best.find(ident);
                if (it == best.end()) {
                    best[ident] = std::move(bc);
                } else if (known_ids(*bc) > known_ids(*it->second)) {
                    merge_versions(*bc, *it->second);
                    it->second = std::move(bc);
                } else {
                    merge_versions(*it->second, *bc);
                }
            }
            if (br.getIDOfRecentGeneratedES() > merged.getIDOfRecentGeneratedES())
                merged.setIDOfRecentGeneratedES(br.getIDOfRecentGeneratedES());
        };
 
        for (auto& [cid, blob] : ckpt.branch_record_blobs) {
            BranchRecord br;
            std::istringstream iss(blob, std::ios::binary);
            boost::archive::binary_iarchive ia(iss);
            ia >> br;
            // Only the blob that OWNS target CID `cid` may reserve it — a
            // foreign record's same-CID cluster is a different physical node.
            absorb(br, cid);
        }
        // Chained resume: fold the prior checkpoint's merged record in too.
        // Carried-over tasks' clusters (and closed ancestors) may not appear
        // in any held record gathered this run. Carried tasks' targets live
        // in the prior merged record, so they reserve their CIDs here.
        if (resume_ckpt_ && !resume_ckpt_->branch_center_blob.empty()) {
            std::set<int> carried_targets;
            for (auto& pt : ckpt.pending_tasks)
                if (!ckpt.branch_record_blobs.count(pt.cid))
                    carried_targets.insert(pt.cid);
            BranchRecord br;
            std::istringstream iss(resume_ckpt_->branch_center_blob, std::ios::binary);
            boost::archive::binary_iarchive ia(iss);
            ia >> br;
            for (auto& bc : br.branch_cluster) {
                if (!bc) continue;
                int ident = identity_of(*bc);
                int c = bc->getIDOfThisBranchCluster();
                if (carried_targets.count(c) && !reserved_cid.count(ident))
                    reserved_cid[ident] = c;
                auto it = best.find(ident);
                if (it == best.end()) {
                    best[ident] = std::move(bc);
                } else if (known_ids(*bc) > known_ids(*it->second)) {
                    merge_versions(*bc, *it->second);
                    it->second = std::move(bc);
                } else {
                    merge_versions(*it->second, *bc);
                }
            }
            if (br.getIDOfRecentGeneratedES() > merged.getIDOfRecentGeneratedES())
                merged.setIDOfRecentGeneratedES(br.getIDOfRecentGeneratedES());
        }
 
        // Assign final CIDs: reserved identities take their pending target CID
        // (the collision renumbering of held entries above may have changed it
        // inside the owning blob while foreign copies of the same cluster kept
        // the unrenumbered CID); others keep their CID unless a
        // reserved/earlier cluster holds it.
        std::set<int> used_cids;
        int next_cid = 0;
        for (auto& [ident, bc] : best)
            next_cid = std::max(next_cid, bc->getIDOfThisBranchCluster());
        for (auto& [ident, cid] : reserved_cid) {
            used_cids.insert(cid);
            next_cid = std::max(next_cid, cid);
        }
        next_cid += 1000;
 
        std::map<int, std::pair<int, BranchCluster*>> branch_to_cluster; // branch id -> (final cid, cluster)
        for (auto& [ident, bc] : best) {
            int final_cid;
            auto rit = reserved_cid.find(ident);
            if (rit != reserved_cid.end()) {
                final_cid = rit->second;
            } else {
                final_cid = bc->getIDOfThisBranchCluster();
                if (used_cids.count(final_cid)) {
                    int old = final_cid;
                    final_cid = next_cid++;
                    PU::print_message(rank, "[CHECKPOINT] CID " + std::to_string(old)
                        + " names two physical clusters — renumbering non-target copy to "
                        + std::to_string(final_cid), PG::log);
                }
                used_cids.insert(final_cid);
            }
            bc->setIDOfThisBranchCluster(final_cid);
            for (auto& bp : bc->all_branch)
                if (bp && bp->getIDOfThisBranch() > 0)
                    branch_to_cluster[bp->getIDOfThisBranch()] = {final_cid, bc.get()};
        }
 
        // Repair mother links via the globally unique mother BRANCH id —
        // initiationForRestart appends new child branch IDs through this link,
        // so a stale CID would silently drop tree linkage in resumed output.
        for (auto& [ident, bc] : best) {
            int mb = bc->getIDOfTheMotherBranch();
            auto mit = branch_to_cluster.find(mb);
            if (mb > 0 && mit != branch_to_cluster.end()) {
                bc->setIDOfMotherCluster(mit->second.first);
                for (auto& bp : mit->second.second->all_branch)
                    if (bp && bp->getIDOfThisBranch() == mb) {
                        bp->setIDOfTheChildBranchCluster(bc->getIDOfThisBranchCluster());
                        break;
                    }
            }
        }
 
        for (auto& [ident, bc] : best) {
            auto pit = pending_min_bi.find(bc->getIDOfThisBranchCluster());
            if (pit != pending_min_bi.end()) {
                bc->setCurrentBranchNum(pit->second - 1);
            } else {
                bc->setCurrentBranchNum(static_cast<int>(bc->all_branch.size()) - 1);
            }
            merged.branch_cluster.push_back(std::move(bc));
        }
        // Set max CID so resumed workers assign new cluster CIDs above existing ones.
        // Without this, new clusters start at CID=0 which triggers the "first cluster"
        // check in checkAndAddBranchCluster(), resetting probability to 1.0.
        {
            int max_cid = -1;
            for (auto& bc : merged.branch_cluster)
                max_cid = std::max(max_cid, bc->getIDOfThisBranchCluster());
            merged.setIDOfRecentGeneratedBranchCluster(max_cid);
        }
        // Branch IDs are IMMUTABLE for the life of a node: the checkpoint
        // stores them exactly as issued. PE files are keyed by these IDs, so
        // a resumed client re-saves an ancestor node under the same name
        // phase 1 used. Both resume paths are ID-agnostic: serial continues
        // above the recorded max (set below), MPI windows seed in a fresh
        // epoch above max_branch_id.
        {
            int orig_max_branch_id = -1;
            for (auto& bc : merged.branch_cluster)
                for (auto& bp : bc->all_branch)
                    if (bp->getIDOfThisBranch() > orig_max_branch_id)
                        orig_max_branch_id = bp->getIDOfThisBranch();
            merged.setIDOfRecentGeneratedBranch(orig_max_branch_id);
            ckpt.max_branch_id = orig_max_branch_id;
        }
 
        std::ostringstream oss(std::ios::binary);
        boost::archive::binary_oarchive oa(oss);
        oa << merged;
        ckpt.branch_center_blob = oss.str();
    }
 
    // Store completed reports from viz state
    if (viz_state_) {
        std::lock_guard<std::mutex> lock(viz_state_->mu);
        ckpt.completed_reports = viz_state_->completed;
    }
 
    ckpt.compress_queue = g_sim.config.compress_queue;
    ckpt.config_fingerprint = g_sim.config_fingerprint;
    std::string ckpt_path = g_sim.config.checkpoint_dir + "/checkpoint_seq_" + std::to_string(sequences_completed_) + ".dat";
    ckpt.save(ckpt_path);
 
    for (auto& pt : ckpt.pending_tasks) {
        PU::print_message(rank, "[CHECKPOINT]   pending: CID=" + std::to_string(pt.cid)
            + " bi=" + std::to_string(pt.branch_index)
            + " seq=" + std::to_string(pt.sequence_id), PG::log);
    }
    PU::print_message(rank, "[CHECKPOINT] Saved: " + ckpt_path
        + " (" + std::to_string(sequences_completed_) + " sequences, "
        + std::to_string(ckpt.pending_tasks.size()) + " pending, "
        + std::to_string(ckpt.restart_states.size()) + " restart states)", PG::log);
    std::cout << "Resume with: --resume=" << ckpt_path << std::endl;
}
 
void Server::dispatch_resume_tasks()
{
    auto ckpt = DETCheckpoint::load(g_sim.config.resume_file);
    // Refuse to resume against changed configuration inputs.
    if (!ckpt.config_matches(g_sim.config_fingerprint)) {
        std::ostringstream msg;
        msg << "FATAL: checkpoint configuration mismatch (checkpoint fingerprint 0x"
            << std::hex << ckpt.config_fingerprint << ", current inputs 0x"
            << g_sim.config_fingerprint << std::dec
            << "). The input deck, Input/ tree, or ZCtrlPar.txt changed since this "
            << "checkpoint was written. Restore the original inputs or start a fresh run.";
        PU::print_message(rank, msg.str(), PG::log);
        std::cerr << msg.str() << std::endl;
        MPI_Abort(MPI_COMM_WORLD, 1);
    }
    if (ckpt.version >= 3 && ckpt.compress_queue != g_sim.config.compress_queue) {
        PU::print_message(rank, "[RESUME] Overriding --compress-queue to match the checkpoint ("
            + std::string(ckpt.compress_queue ? "on" : "off") + ")", PG::log);
        g_sim.config.compress_queue = ckpt.compress_queue;
    }
    sequences_completed_ = ckpt.sequences_completed;
    sequence_counter_ = ckpt.sequence_counter;
 
    // Restore viz state
    if (viz_state_) {
        std::lock_guard<std::mutex> lock(viz_state_->mu);
        viz_state_->completed = ckpt.completed_reports;
    }
 
    PU::print_message(rank, "[RESUME] Loaded: " + std::to_string(ckpt.sequences_completed)
        + " sequences done, " + std::to_string(ckpt.pending_tasks.size()) + " pending tasks"
        + " seq_counter=" + std::to_string(ckpt.sequence_counter), PG::log);
 
    // Derive this resumed run's branch-ID epoch. Phase-1 identities
    // (rank-offset IDs preserved in merged PE outputs and recorded as
    // max_branch_id) must never be re-issued; every phase-2 rank window
    // starts in a fresh epoch above them. The epoch base is the first
    // multiple of the stride (128 * 100000 IDs, i.e. 128 rank windows of
    // 100000 IDs) above the highest ID issued so far; a rank's window then
    // starts at epoch_base + rank * 100000.
    // The record only spans pending lineages, so also scan the merged PE
    // filenames, the ground truth of every ID any prior phase issued.
    {
        int max_issued_bid = ckpt.max_branch_id;
        int max_file_bid = -1;
        bool has_zero = false, has_negative = false;
        long nonneg_count = 0;
        std::string pe_dir = g_sim.SScheduler.getPEDirectory();
        if (fs::is_directory(pe_dir))
            for (auto& e : fs::recursive_directory_iterator(pe_dir)) {
                std::string fname = e.path().filename().string();
                if (e.is_regular_file() && fname.size() > 7 && fname.substr(0, 7) == "ZPENode"
                    && fname.back() == 't') {
                    // ZPENodes_agg.txt matches this filter; stoi("agg") must not throw.
                    int fid = 0;
                    try { fid = std::stoi(fname.substr(fname.find('_') + 1, fname.find('.') - fname.find('_') - 1)); }
                    catch (...) { continue; }
                    if (fid > max_issued_bid) max_issued_bid = fid;
                    if (fid > max_file_bid) max_file_bid = fid;
                    if (fid == 0) has_zero = true;
                    if (fid < 0) has_negative = true;
                    if (fid >= 0) ++nonneg_count;
                }
            }
        // A finalized Output (end-of-run sequential rename) is dense 0..N-1
        // with the -999 donor sentinels stripped; a checkpoint-stop state is
        // sparse (rank-window IDs) and carries the sentinel while work is
        // pending. Resuming onto a finalized tree would corrupt it
        // (natural-ID re-saves land next to renamed files), so refuse.
        if (has_zero && !has_negative
            && nonneg_count == static_cast<long>(max_file_bid) + 1) {
            std::string msg = "FATAL: the project Output/ already contains a finalized run "
                "(sequentially renamed PE files). This checkpoint's run has completed; "
                "resuming would corrupt the merged outputs. Move the existing Output/ "
                "aside to re-run from this checkpoint.";
            PU::print_message(rank, msg, PG::log);
            std::cerr << msg << std::endl;
            MPI_Abort(MPI_COMM_WORLD, 1);
        }
        constexpr int kEpochStride = 128 * 100000;
        int epoch_base = (std::max(max_issued_bid, 0) / kEpochStride + 1) * kEpochStride;
        for (auto& ri : ckpt.task_restart_states)
            if (ri) ri->id_epoch_base = epoch_base;
        PU::print_message(rank, "[RESUME] Branch-ID epoch base = " + std::to_string(epoch_base)
            + " (max issued so far: " + std::to_string(max_issued_bid) + ")", PG::log);
    }
 
    // Count tasks per source so blobs can be freed at last dispatch.
    resume_source_refs_.clear();
    for (auto& pt : ckpt.pending_tasks) {
        auto cs = ckpt.cid_source.find(pt.cid);
        if (cs != ckpt.cid_source.end())
            ++resume_source_refs_[cs->second];
    }
 
    // Store checkpoint for dispatch_one_resume_task (uses per-task indexing)
    resume_pending_ = ckpt.pending_tasks;
    resume_ckpt_ = std::make_unique<DETCheckpoint>(std::move(ckpt));
 
    // Dispatch as many as we can now (one per free client)
    while (!resume_pending_.empty() && num_free_clients > 0)
        dispatch_one_resume_task();
}
 
bool Server::dispatch_one_resume_task()
{
    if (resume_pending_.empty() || !resume_ckpt_) return false;
    int free_rank = find_next_free_client_round_robin();
    if (free_rank < 0) return false;
 
    // Pop from back — index into per-task vectors using current size
    int ti = static_cast<int>(resume_pending_.size()) - 1;
    auto task = resume_pending_.back();
    resume_pending_.pop_back();
 
    // Per-task data is indexed the same as pending_tasks in the checkpoint.
    // resume_pending_ starts as a copy of ckpt.pending_tasks and only ever
    // shrinks from the back, so ti (its size before the pop, minus 1) is also
    // the task's index into the checkpoint's per-task vectors.
 
    if (ti < 0 || ti >= static_cast<int>(resume_ckpt_->task_restart_states.size()) ||
        !resume_ckpt_->task_restart_states[ti]) {
        PU::print_message(rank, "[RESUME] ERROR: no data for task index=" + std::to_string(ti), PG::log);
        clients_busy[free_rank] = 0;
        ++num_free_clients;
        return false;
    }
 
    auto& rinfo = resume_ckpt_->task_restart_states[ti];
 
    int seq_id = task.sequence_id > 0 ? task.sequence_id : sequence_counter_++;
 
    // The restart blob lives once in source_blobs: attach it to the
    // RestartNodeInfo transiently for the wire and detach afterwards so the
    // server never accumulates per-task copies.
    {
        auto cs = resume_ckpt_->cid_source.find(task.cid);
        if (cs != resume_ckpt_->cid_source.end()) {
            auto sb = resume_ckpt_->source_blobs.find(cs->second);
            if (sb != resume_ckpt_->source_blobs.end())
                rinfo->relap_restart_buf = sb->second;
        }
    }
    PU::send_object(*rinfo, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
    rinfo->relap_restart_buf.clear();
    rinfo->relap_restart_buf.shrink_to_fit();
    // Free the source blob once its last referencing task is dispatched, so
    // the server's hold on checkpoint blobs shrinks as the resume proceeds.
    {
        auto cs = resume_ckpt_->cid_source.find(task.cid);
        if (cs != resume_ckpt_->cid_source.end()) {
            auto rc = resume_source_refs_.find(cs->second);
            if (rc != resume_source_refs_.end() && --rc->second <= 0) {
                resume_ckpt_->source_blobs.erase(cs->second);
                resume_source_refs_.erase(rc);
            }
        }
        resume_ckpt_->task_restart_states[ti].reset();
    }
    // Every resumed task is dispatched with the MERGED BranchRecord
    // (branch_center_blob: one version of each cluster, branch IDs exactly as
    // issued). write_checkpoint stores no per-task records.
    BranchRecord br;
    {
        std::istringstream iss(resume_ckpt_->branch_center_blob, std::ios::binary);
        boost::archive::binary_iarchive ia(iss);
        ia >> br;
    }
    PU::send_object(br, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
    int path_len = 0;
    MPI_Send(&path_len, 1, MPI_INT, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
    char empty = '\0';
    MPI_Send(&empty, 0, MPI_CHAR, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
    MPI_Send(&seq_id, 1, MPI_INT, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
    MPI_Send(&task.branch_index, 1, MPI_INT, free_rank, PG::client_restart_run, MPI_COMM_WORLD);
 
    rank_current_sequence_[free_rank] = seq_id;
 
    PU::print_message(rank, "[RESUME] Dispatched CID=" + std::to_string(task.cid)
        + " bi=" + std::to_string(task.branch_index) + " seq=" + std::to_string(seq_id)
        + " -> C[" + std::to_string(free_rank) + "]", PG::log);
    return true;
}