ADSIDAC.cpp

Go to the documentation of this file.

#include "ADSIDAC.h"
#include "Solver/RelapSolver.h"
#include "Solver/RestApiSolver.h"
#include "Solver/AdsidacCallback.h"
#include "Notification/Notification.h"
 
#include "Global/DETCheckpoint.h"
#include "Global/SharedConfig.h"
#include "Global/CheckedIO.h"
 
#include <algorithm>
#include <fstream>
#include <sstream>
#include <memory>
#include <cassert>
#include <chrono>
#include <csignal>
#include "Global/PEOutput.h"
#include "Global/OutputMerge.h"
#include <iomanip>
#include <thread>
 
#ifndef _CONSOLE
#define _CONSOLE
#endif
 
using namespace std;
 
// ---- Solver and callback instances ----
// solver is created dynamically based on --solver flag
static std::unique_ptr<ThermalHydraulicSolver> solver_ptr;
static ThermalHydraulicSolver* solver_raw = nullptr;  // raw pointer for extern "C" trampolines
AdsidacCallback engine_callback;
 
// SIGUSR1 checkpoint request (for HPC wall-time limits)
static volatile sig_atomic_t checkpoint_requested = 0;
static void sigusr1_handler(int) { checkpoint_requested = 1; }
 
// SIGINT (Ctrl+C) interactive interrupt
static volatile sig_atomic_t sigint_received = 0;
static void sigint_handler(int) { sigint_received = 1; }
 
// Show interactive interrupt menu, return user choice (1=continue, 2=checkpoint+exit, 3=exit)
static int show_interrupt_menu(int seq_completed) {
    cout << "\n\n=== Simulation Interrupted (Ctrl+C) ===" << endl;
    cout << "  Sequences completed: " << seq_completed << endl;
    cout << endl;
    cout << "  1. Continue simulation" << endl;
    cout << "  2. Save checkpoint and exit" << endl;
    cout << "  3. Exit without saving" << endl;
    cout << endl;
    cout << "Choice [1]: " << flush;
 
    string line;
    if (!getline(cin, line) || line.empty()) return 1;
    if (line[0] == '2') return 2;
    if (line[0] == '3') return 3;
    return 1;
}
 
// Build a DETCheckpoint from current serial state
static DETCheckpoint build_serial_checkpoint(int seq_completed) {
    DETCheckpoint ckpt;
    ckpt.sequences_completed = seq_completed;
    ckpt.sequence_counter = g_sim.Branch_center.getIDOfRecentGeneratedES() + 1;
 
    // Serial execution has no transfer queue: restart_buffers always hold raw
    // membuf bytes (a resumed compressed-queue checkpoint is stripped at
    // intake), so serial checkpoints are always raw-marked, whatever
    // --compress-queue says. The flag records the encoding of the blobs, and
    // a resume decodes them according to it.
    ckpt.compress_queue = false;
    ckpt.config_fingerprint = g_sim.config_fingerprint;
 
    // Serialize BranchRecord into blob (avoids copy of unique_ptr members)
    ckpt.capture_branch_center(g_sim.Branch_center);
 
    // Collect restart states from restart_info deque (shared_ptr, no copy needed).
    // Each source-sequence blob is stored ONCE (cid_source maps sibling CIDs
    // to the shared source) and the live RestartNodeInfos are never mutated.
    ckpt.max_branch_id = g_sim.Branch_center.getIDOfRecentGeneratedBranch();
    for (auto& ri : g_sim.restart_info) {
        if (!ri) continue;
        int cid = ri->getCorrespondingBranchClusterID();
        if (ckpt.restart_states.find(cid) == ckpt.restart_states.end()) {
            ckpt.restart_states[cid] = ri;
            auto cs_it = g_sim.cid_to_source_sequence.find(cid);
            if (cs_it != g_sim.cid_to_source_sequence.end()) {
                int seq = cs_it->second;
                ckpt.cid_source[cid] = seq;
                if (ckpt.source_blobs.find(seq) == ckpt.source_blobs.end()) {
                    auto buf_it = g_sim.restart_buffers.find(seq);
                    if (buf_it != g_sim.restart_buffers.end()) {
                        ckpt.source_blobs[seq] = buf_it->second; // copy: the live run continues
                    } else {
                        // File-based restart: read from disk. A short/failed
                        // read would put a corrupt blob into the checkpoint
                        // that fails only at resume, so it is fatal here.
                        std::string path = restart_path("restart_seq_" + std::to_string(seq) + ".r");
                        if (!read_file_verified(path, ckpt.source_blobs[seq])) {
                            std::cerr << "FATAL: checkpoint would embed a corrupt restart blob for CID "
                                      << cid << " — aborting." << std::endl;
                            std::exit(1);
                        }
                    }
                }
            } else if (!ri->relap_restart_buf.empty()) {
                // Defensive: a rinfo carrying its blob inline (not the serial
                // layout) still checkpoints correctly.
                ckpt.cid_source[cid] = cid;
                ckpt.source_blobs[cid] = ri->relap_restart_buf;
            }
        }
    }
 
    // Scan branch clusters for unexplored branches
    for (auto& bc : g_sim.Branch_center.branch_cluster) {
        int cid = bc->getIDOfThisBranchCluster();
        int curr = bc->getCurrentBranchNum();
        int total = static_cast<int>(bc->all_branch.size());
        for (int bi = curr + 1; bi < total; ++bi) {
            PendingBranch pb;
            pb.cid = cid;
            pb.branch_index = bi;
            ckpt.pending_tasks.push_back(pb);
        }
    }
 
    // Populate per-task restart states for MPI resume.
    // branch_record_blobs and task_branch_blobs are left empty: MPI resume
    // always uses branch_center_blob (the merged blob), so per-task copies
    // would only duplicate it O(N) times in the checkpoint file.
    for (auto& pt : ckpt.pending_tasks) {
        auto ri_it = ckpt.restart_states.find(pt.cid);
        ckpt.task_restart_states.push_back(ri_it != ckpt.restart_states.end() ? ri_it->second : nullptr);
    }
 
    return ckpt;
}
 
// Forward declarations for SEQ-FP diagnostic
static std::string compute_ancestry_path();
static void log_seq_fingerprint(double term_time);
 
// --merge-only: consolidate the per-rank outputs of a deferred (or
// interrupted) merge and exit. Runs serially, never through parallel_init
// (Setup_Directories would wipe the rank dirs this mode exists to read) and
// never constructing a Server (viz/occupancy/crew side effects).
static int run_merge_only(int argc, char* argv[]) {
    g_sim.SScheduler.parseOptions(argc, argv);
    const std::string project = g_sim.SScheduler.getProjectDirectory();
    const std::string pe_dir = (fs::path(project) / "Output" / "PE").string() + "/";
    const auto ranks = adsidac::merge::discover_ranks(project);
    const bool has_manifest = adsidac::merge::manifest_exists(project);
    if (ranks.empty()) {
        if (adsidac::merge::looks_finalized(pe_dir)) {
            std::cout << "[MERGE-ONLY] Output/ is already finalized and no rank "
                         "directories remain; nothing to do" << std::endl;
            return 0;
        }
        std::cerr << "ERROR: no rank_N directories under " << project
                  << "; nothing to merge" << std::endl;
        return 1;
    }
    if (adsidac::merge::looks_finalized(pe_dir)) {
        if (!adsidac::merge::ranks_hold_outputs(project, ranks)) {
            std::cout << "[MERGE-ONLY] Output/ is already finalized and the rank "
                         "directories hold no outputs; nothing to do" << std::endl;
            return 0;
        }
        std::cerr << "ERROR: Output/PE already carries a finalized run while rank "
                     "directories still hold outputs. Merging would corrupt the finalized "
                     "tree; move Output/ aside first" << std::endl;
        return 1;
    }
    if (!has_manifest)
        std::cout << "[MERGE-ONLY] no " << adsidac::merge::manifest_basename()
                  << " manifest: salvage mode (merging whatever the rank directories hold)"
                  << std::endl;
    const bool aggregate = adsidac::merge::manifest_aggregate(
        project, g_sim.config.aggregate_pe_output);
    int threads = g_sim.config.merge_threads;
    if (threads <= 0)
        threads = std::min(8u, std::max(1u, std::thread::hardware_concurrency()));
    const auto t0 = std::chrono::steady_clock::now();
    const auto [seqs, pe_count] = adsidac::merge::merge_project(project, aggregate, threads);
    const double secs = std::chrono::duration<double>(
        std::chrono::steady_clock::now() - t0).count();
    std::cout << "Merged " << seqs << " sequences, " << pe_count << " PE nodes into "
              << (fs::path(project) / "Output").string() << " in " << std::fixed
              << std::setprecision(1) << secs << " s (merge-only, " << ranks.size()
              << " rank dirs, " << threads << " threads)" << std::endl;
    return 0;
}
 
int main(int argc, char* argv[]){
 
  // Handle --help/--version/--info before MPI_Init so all ranks exit cleanly
  for (int i = 1; i < argc; ++i) {
      std::string arg(argv[i]);
      if (arg == "--help" || arg == "-h" || arg == "--version" || arg == "-v" || arg == "--info" || arg == "-i") {
          g_sim.SScheduler.parseOptions(argc, argv);  // prints and calls exit(0)
      }
  }
 
  // --merge-only is caught before the MPI branch: under mpirun it would run
  // Setup_Directories and wipe the very rank directories it must read.
  for (int i = 1; i < argc; ++i) {
      if (std::string(argv[i]) == "--merge-only") {
          if (std::getenv("OMPI_COMM_WORLD_SIZE") || std::getenv("PMI_SIZE")) {
              std::cerr << "ERROR: --merge-only runs serially; launch it without mpirun\n";
              return 1;
          }
          return run_merge_only(argc, argv);
      }
  }
 
#ifdef ENABLE_MPI
    parallelexec parallel;
    g_sim.config.is_mpi_environment= (std::getenv("OMPI_COMM_WORLD_SIZE") != NULL) ||
                        (std::getenv("PMI_SIZE") != NULL);
 
    if (g_sim.config.is_mpi_environment) {
        MPI_Init(&argc, &argv);
        parallel.parallel_init(&argc, &argv);
        MPI_Finalize();
        ParallelUtilities::print_message(PG::rank, "Parallel Execution finished...", true);
        _exit(0);  // Skip global destructors — client ranks have inconsistent state after moves
    }
#endif
 
  // set system mode
    r5pars_.tsimmode = static_cast<int>(ConsoleMode::Console);
  int i;
 
  BranchEndState *es_pt;
 
  string str, time_begin, time_end;
  time_t t_current;
  time(&t_current);
  time_begin = "";
  time_begin += ctime(&t_current);
  // seed random number generator using system time
  srand((unsigned) time(0));
  //
  g_sim.SScheduler.parseOptions(argc, argv);
  //  SScheduler.readInfoFromFile();      // Legacy input parsing option
  g_sim.SScheduler.initiationForSimulation();
  g_sim.Plant.initiationForSimulation();
  g_sim.Translator.initiationForSimulation();
  g_sim.Branch_center.initiationForSimulation();
  // Register the shared procedures master before any snapshot is
  // deserialized (the --resume path unpacks crew snapshots before the first
  // RELAP transient loads the book via onFirstInit). Idempotent.
  g_sim.Indicator.readInfoFromFile();  // procedures parser needs the panel maps
  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);
  g_sim.All_procedure.initiationForSimulation();
  adsidac::set_shared_procedures(&g_sim.All_procedure);
 
  // Create solver based on --solver flag (solver_url built by Scheduler::parseOptions)
  if (!g_sim.config.solver_url.empty()) {
      solver_ptr = std::make_unique<RestApiSolver>(g_sim.config.solver_url);
      cout << "Using mock solver: " << g_sim.config.solver_url << std::endl;
  } else {
      solver_ptr = std::make_unique<RelapSolver>();
  }
  solver_raw = solver_ptr.get();
  auto& solver = *solver_raw;
 
   cout << "Serial execution" << std::endl;
   cout << "Project: " << g_sim.SScheduler.getProjectDirectory() << std::endl;
   g_sim.fresh_restart = false;
   int sequences_completed = 0;
 
   // Register signal handlers
   signal(SIGUSR1, sigusr1_handler);
   signal(SIGINT, sigint_handler);
 
   // Resume from checkpoint if --resume specified
   if (!g_sim.config.resume_file.empty()) {
       // The checkpoint record only spans pending lineages, so its max branch
       // ID can sit BELOW IDs phase 1 already issued (closed subtrees). The PE
       // filenames are the ground truth of issued IDs: scan their max so the
       // epoch jump below clears every ID any phase ever issued.
       int max_issued_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 == 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, with work pending, carries the
       // sentinel. Resuming onto a finalized tree would re-save nodes under
       // natural IDs next to the renamed files and corrupt it, so refuse.
       if (has_zero && !has_negative
           && nonneg_count == static_cast<long>(max_issued_bid) + 1) {
           std::cerr << "FATAL: the project Output/ already contains a finalized run\n"
                     << "  (sequentially renamed PE files). This checkpoint's run has\n"
                     << "  completed; resuming would corrupt the merged outputs.\n"
                     << "  Move the existing Output/ aside to re-run from this checkpoint."
                     << std::endl;
           return 1;
       }
       auto ckpt = DETCheckpoint::load(g_sim.config.resume_file);
       // Refuse to resume against changed configuration inputs: the
       // checkpoint's crew/plant state is meaningful only under the files it
       // was written with.
       if (!ckpt.config_matches(g_sim.config_fingerprint)) {
           std::cerr << "FATAL: checkpoint configuration mismatch.\n"
                     << "  checkpoint fingerprint: 0x" << std::hex << ckpt.config_fingerprint << "\n"
                     << "  current inputs:         0x" << g_sim.config_fingerprint << std::dec << "\n"
                     << "  The input deck, Input/ tree, or ZCtrlPar.txt changed since this\n"
                     << "  checkpoint was written. Restore the original inputs or start a fresh run."
                     << std::endl;
           return 1;
       }
       if (ckpt.version >= 3 && ckpt.compress_queue != g_sim.config.compress_queue) {
           std::cout << "[RESUME] Overriding --compress-queue to match the checkpoint ("
                     << (ckpt.compress_queue ? "on" : "off") << ")" << std::endl;
           g_sim.config.compress_queue = ckpt.compress_queue;
       }
 
       // Restore branch tree from serialized merged blob
       g_sim.Branch_center.clearBranchRecordAll();
       ckpt.restore_branch_center(g_sim.Branch_center);
       g_sim.config.resume_has_end_states = !g_sim.Branch_center.end_state.empty();
       g_sim.Branch_center.initiationForSimulation();
 
       // Set IDs so resumed sequences don't collide with phase 1 outputs.
       g_sim.Branch_center.setIDOfRecentGeneratedES(ckpt.sequence_counter - 1);
       int max_cid = -1, max_bid = -1;
       for (auto& bc : g_sim.Branch_center.branch_cluster) {
           int cid = bc->getIDOfThisBranchCluster();
           if (cid > max_cid) max_cid = cid;
           for (auto& bp : bc->all_branch) {
               int bid = bp->getIDOfThisBranch();
               if (bid > max_bid) max_bid = bid;
           }
       }
       // Seed new serial branch IDs in a fresh epoch above every ID any phase
       // issued (record max AND output-file max): reissuing an ID would
       // overwrite that node's PE file. Same epoch scheme as MPI resume
       // (Server::dispatch_resume_tasks), so chained serial/MPI resumes stay
       // collision-free.
       {
           constexpr int kEpochStride = 128 * 100000;
           int true_max = std::max({max_bid, max_issued_bid, ckpt.max_branch_id, 0});
           int epoch_base = (true_max / kEpochStride + 1) * kEpochStride;
           std::cout << "[RESUME] Branch-ID epoch base = " << epoch_base
                     << " (max issued so far: " << true_max << ")" << std::endl;
           g_sim.Branch_center.setIDOfRecentGeneratedBranch(epoch_base);
       }
       g_sim.Branch_center.setIDOfRecentGeneratedBranchCluster(max_cid);
 
       // Restore restart_info deque and restart_buffers.
       // Blobs are stored once per source: MOVE each into the runtime map the
       // serial FIFO shares across sibling CIDs (no per-CID duplication, and
       // no second copy held by the checkpoint object).
       g_sim.restart_info.clear();
       g_sim.restart_buffers.clear();
       g_sim.cid_to_source_sequence.clear();
       for (auto& [cid, ri] : ckpt.restart_states) {
           g_sim.restart_info.push_back(ri);
       }
       for (auto& [cid, seq] : ckpt.cid_source)
           g_sim.cid_to_source_sequence[cid] = seq;
       for (auto& [seq, blob] : ckpt.source_blobs)
           g_sim.restart_buffers[seq] = std::move(blob);
       ckpt.source_blobs.clear();
       // Serial execution feeds restart buffers straight into the RELAP5
       // membuf: no transfer queue strips the queue-layer LZ4 of a compressed
       // MPI checkpoint, so strip it here. This run is raw from here on (its
       // own re-checkpoints are stamped raw accordingly).
       if (g_sim.config.compress_queue) {
           for (auto& [seq, blob] : g_sim.restart_buffers) {
               if (blob.empty()) continue;
               auto raw = lz4_decompress(blob.data(), static_cast<int>(blob.size()));
               blob = std::move(raw);
           }
           g_sim.config.compress_queue = false;
           std::cout << "[RESUME] Stripped queue-layer compression from "
                     << g_sim.restart_buffers.size() << " restart buffers for serial execution" << std::endl;
       }
 
       sequences_completed = ckpt.sequences_completed;
       std::cout << "[RESUME] Loaded: " << ckpt.sequences_completed << " sequences done, "
                 << ckpt.pending_tasks.size() << " pending tasks, "
                 << g_sim.restart_buffers.size() << " restart buffers" << std::endl;
 
       engine_callback.restartForNewSequence(solver);
       r5pars_.cnewrst = false;
       solver.setNewRestart(false);
       solver.setInputDeck("indeck.i");
   }
 
   auto wall_start = std::chrono::steady_clock::now();
   bool stopped_at_checkpoint = false; 
   do { 
    i = g_sim.Branch_center.getIDOfRecentGeneratedES() + 1;
    cout << "*** Sequence " << i << " ****" << endl;
    g_sim.expected_abort = false;
    solver.setThermoCacheMode(g_sim.config.thermo_cache_mode);
    // Always use membuf for RELAP5 I/O — adaptive decision only affects
    // whether the extracted buffer is kept in RAM or dumped to file.
    solver.setMembufMode(1);
    solver.runSequence(engine_callback);
    // SEQ-FP: log fingerprint for this completed sequence
    log_seq_fingerprint(solver.currentTime());
    // Save restart data for FIFO if branches were generated this sequence
    if (g_sim.branches_generated_this_sequence) {
        // Decide storage: keep in memory or dump to file
        bool store_in_mem = should_use_memory_restart();
        if (g_sim.config.memory_limit_mb > 0)
            cout << "[ADAPTIVE] seq=" << i << " RSS=" << get_rss_mb() << "MB limit=" << g_sim.config.memory_limit_mb << "MB -> " << (store_in_mem ? "memory" : "file") << endl;
        auto buf = solver.checkpoint();
        if (!buf.empty()) {
            if (store_in_mem) {
                g_sim.restart_buffers[i] = std::move(buf);
            } else {
                std::string backup = restart_path("restart_seq_" + std::to_string(i) + ".r");
                write_file_or_die(backup, buf.data(), buf.size());
            }
        }
        g_sim.branches_generated_this_sequence = false;
    }
    i = remove("out.o"); 
    if (!g_sim.expected_abort) { //sequesce termination was not detected
        es_pt = new BranchEndState();
        str = "Sequence is terminated for unknown reason.";
        cout << "Sequence is terminated for unknown reason." << endl;
        es_pt->setTypeOfConsequence(Consequence::NoKB);
        es_pt->setConsequenceDescription(str);
        g_sim.Branch_center.addAnEndState(es_pt);
    }
    if (r5pars_.cnewrst) {
        solver.setInputDeck("indeck.i");
        solver.setNewRestart(false);
    }
    // Save per-sequence outputs before restartforanewsequence_ clears them
    g_sim.KPS.outputHistoryToFile();
    g_sim.Aux_info.saveInfoToFile();
    ++sequences_completed;
 
    // Interactive interrupt menu (Ctrl+C)
    if (sigint_received) {
        sigint_received = 0;
        int choice = show_interrupt_menu(sequences_completed);
        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);
            }
            auto ckpt = build_serial_checkpoint(sequences_completed);
            std::string path = g_sim.config.checkpoint_dir + "/checkpoint_seq_" + std::to_string(sequences_completed) + ".dat";
            ckpt.save(path);
            cout << "Resume with: --resume=" << path << endl;
            stopped_at_checkpoint = true;
            break;
        } else if (choice == 3) {
            cout << "Exiting without saving." << endl;
            break;
        }
        // choice 1: re-register handler (signal resets to SIG_DFL on some systems)
        signal(SIGINT, sigint_handler);
    }
 
    // Checkpoint save (interval-based or SIGUSR1-triggered)
    if (!g_sim.config.checkpoint_dir.empty() &&
        ((g_sim.config.checkpoint_interval > 0 && sequences_completed % g_sim.config.checkpoint_interval == 0) ||
         checkpoint_requested)) {
        checkpoint_requested = 0;
        auto ckpt = build_serial_checkpoint(sequences_completed);
        std::string ckpt_path = g_sim.config.checkpoint_dir + "/checkpoint_seq_" + std::to_string(sequences_completed) + ".dat";
        ckpt.save(ckpt_path);
        cout << "Resume with: --resume=" << ckpt_path << endl;
        if (g_sim.config.checkpoint_stop) {
            cout << "[CHECKPOINT] --checkpoint-stop: exiting after " << sequences_completed << " sequences" << endl;
            stopped_at_checkpoint = true;
            break;
        }
    }
 
    engine_callback.restartForNewSequence(solver);
    cout << "terminateall: " << solver.isTerminateAll() << endl;
   } while (!solver.isTerminateAll());
   //
  // Final cleanup of any remaining backup files
  for (auto& [c, s] : g_sim.cid_to_source_sequence) {
      std::string bkp = restart_path("restart_seq_" + std::to_string(s) + ".r");
      if (fs::exists(bkp)) fs::remove(bkp);
  }
  if (g_sim.config.memory_restart) {
      std::string shm_r = restart_path("restart.r");
      if (fs::exists(shm_r)) fs::remove(shm_r);
  }
  g_sim.cid_to_source_sequence.clear();
 
  g_sim.Branch_center.endSimulation();
 
  // After resume, PENode files may have rank-offset IDs from the MPI checkpoint.
  // Renumber them sequentially (same logic as Server::merge_outputs).
  // Only at a true end of run: if this run itself stopped at a checkpoint,
  // the natural-ID names must survive for the next resume.
  if (!g_sim.config.resume_file.empty() && !stopped_at_checkpoint
      && fs::is_directory(g_sim.SScheduler.getPEDirectory())) {
      // Same helper the MPI merge uses (Global/PEOutput.h) so the two
      // renumbering paths cannot diverge on layout or on sentinel handling.
      // finalize_sequential renames in the per-file layout (filename-only
      // renumber) and falls back to collect+rewrite for aggregate content.
      const std::string pe_dir = g_sim.SScheduler.getPEDirectory();
      adsidac::pe::finalize_sequential(pe_dir, g_sim.config.aggregate_pe_output);
  }
 
  g_sim.Communication_place.clearCommunicationPlaceAllPointers();
  g_sim.Command_handler.clearCommandControlCenteFromSimulation();
  g_sim.Action_Taker.clearOperatorATFinal();
  g_sim.Decision_Maker.clearOperatorDMFinal();
  for (auto &ri : g_sim.restart_info) {
    ri = nullptr;
  }
  g_sim.restart_info.clear();
  time(&t_current);
  time_end = "";
  time_end += ctime(&t_current);
  g_sim.Aux_info.addRecordWithTime(-999.0, "Start_time:");
  g_sim.Aux_info.addRecordWithTime(-999.0, time_begin);
  g_sim.Aux_info.addRecordWithTime(-999.0, "End_time:");
  g_sim.Aux_info.addRecordWithTime(-999.0, time_end);
  g_sim.Aux_info.saveInfoToFile();
  cleanup_fort_temps();
  double wall_seconds = std::chrono::duration<double>(
      std::chrono::steady_clock::now() - wall_start).count();
  // Send completion notification (no-op if neither --notify-ntfy nor --notify-discord set)
  {
      RunSummary summary;
      summary.project = g_sim.SScheduler.getProjectDirectory();
      summary.sequences = sequences_completed;
      summary.wall_seconds = wall_seconds;
      summary.ranks = 1;
      summary.solver = g_sim.config.solver_url.empty() ? "relap" : "mock";
      summary.truncation_time = g_sim.SScheduler.getSequenceTruncationTime();
      send_completion_notification(summary);
  }
  cout << "Simulation Finished!\nStart at: " << time_begin << "\nEnd at: "
       << time_end << endl;
  return 0;
}
 
// ---- extern "C" trampolines ----
// These are called by Fortran (dtstep.f). They route to the active SolverCallback.
 
void interfc_() {
    auto* cb = getActiveSolverCallback();
    auto* s = getActiveSolver();
    if (cb && s) cb->onEvaluationPoint(*s);
}
 
void actte_() {
    auto* cb = getActiveSolverCallback();
    auto* s = getActiveSolver();
    if (cb && s) cb->onTimestepAction(*s);
}
 
void secini_() {
    auto* cb = getActiveSolverCallback();
    auto* s = getActiveSolver();
    if (cb && s) cb->onScenarioStart(*s);
}
 
void actonce_() {
    auto* cb = getActiveSolverCallback();
    auto* s = getActiveSolver();
    if (cb && s) cb->onFirstInit(*s);
}
 
void initiate_branching_() {
    auto* cb = getActiveSolverCallback();
    auto* s = getActiveSolver();
    if (cb && s) cb->onBranchingPoint(*s);
}
 
// These are only called from C++ (not Fortran), but kept as extern "C"
// for symbol compatibility. They delegate to the engine_callback.
void restartforanewsequence_() {
    engine_callback.restartForNewSequence(*solver_raw);
}
 
void initiationforrestart_() {
    engine_callback.initiationForRestart(*solver_raw);
}
 
void mutiproactonce_() {
    //empty routine for console version.
}
 
void setrnode_() {
    // empty routine for console version
}
 
// ---- Utility ----
 
void outputErrMessageToFile(string str) {
    ofstream fout("err messages.txt", ios::app);
    fout << r5pars_.timesno << "  " << str << '\n';
    fout.close();
}
 
// ---- SEQ-FP diagnostics ----
 
static std::string compute_ancestry_path() {
    if (g_sim.seq_restart_cid == 0) return "root";
    std::vector<std::string> path_parts;
    int cur_cid = g_sim.seq_restart_cid;
    int cur_branch_pos = g_sim.seq_restart_branch_num;
    while (cur_cid != 0) {
        BranchCluster *bc = g_sim.Branch_center.findBrnachCluster(cur_cid);
        if (!bc) break;
        char buf[64];
        snprintf(buf, sizeof(buf), "%.4f:%d",
                 bc->getTimeAtWhichThisEventHappen(), cur_branch_pos);
        path_parts.emplace_back(buf);
        int mother_cid = bc->getIDOfMotherCluster();
        if (mother_cid == 0) break;
        BranchCluster *mother_bc = g_sim.Branch_center.findBrnachCluster(mother_cid);
        if (!mother_bc) break;
        int mother_branch_pos = -1;
        for (int i = 0; i < static_cast<int>(mother_bc->all_branch.size()); ++i) {
            if (mother_bc->all_branch[i]->getIDOfTheChildBranchCluster() == cur_cid) {
                mother_branch_pos = i;
                break;
            }
        }
        cur_branch_pos = (mother_branch_pos >= 0) ? mother_branch_pos : 0;
        cur_cid = mother_cid;
    }
    std::string result;
    for (int i = static_cast<int>(path_parts.size()) - 1; i >= 0; --i) {
        if (!result.empty()) result += ">";
        result += path_parts[i];
    }
    return result;
}
 
static void log_seq_fingerprint(double term_time) {
    if (!getenv("ADSIDAC_BRANCH_DIAG")) return;
    std::string ancestry = compute_ancestry_path();
    std::string events;
    for (auto& [t, n] : g_sim.current_seq_events) {
        if (!events.empty()) events += ",";
        char buf[64];
        snprintf(buf, sizeof(buf), "%.4f:%d", t, n);
        events += buf;
    }
    if (events.empty()) events = "none";
    fprintf(stderr, "[SEQ-FP] ancestry=%s events=%s term=%.4f\n",
            ancestry.c_str(), events.c_str(), term_time);
}