Tail: the molecular replacement of a prediction or --mr starts beside the analyses; a pass's integrated observations are handed back on a thread of their own

The search (or the carry of the early search) reads of the merge only its French-Wilson amplitudes, which the
analyses make last from a tensor fitted to the merge alone, so - as the first validation of an experimental model
already does - it starts before the analyses on a copy given the same amplitudes, where no twin hypothesis' merge
is searched beside it and no reference supplies the free set. The pass's integrated observations
(ReflectionArena mappings, gigabytes) are freed on a thread of their own as the pass returns, joined when the run
goes, instead of before the outputs are written and before the broker publishes the result. Output unchanged.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015CeEGgpKdfkCTJb3YNMJcu

rugnux/Rugnux.cpp

@@ -1132,6 +1132,16 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
     std::atomic<uint64_t> total_uncompressed_bytes = 0;
     EndMessage end_msg;
     std::future<MtzTable> unmerged_mtz;
+    // The pass's integrated observations - gigabytes, in mappings of their own (ReflectionArena) - are handed
+    // back on a thread of their own as the pass returns: nothing after it reads them, and unmapping them was most
+    // of the time the pass took to end. The unmerged table made from them beside the refinement goes first.
+    const auto retire_outcomes = [&] {
+        retired_outcomes_ = std::async(std::launch::async,
+                [i = std::move(indexer), u = std::move(unmerged_mtz)]() mutable {
+            u = {};
+            i.reset();
+        }).share();
+    };
     PipelineLocals p{
             .logger = logger, .result = result, .observer = observer,
             .geometry_prepass = geometry_prepass, .full = full, .calibration = calibration,
@@ -1218,6 +1228,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
                 Note(note);
         }
         if (finished) {
+            retire_outcomes();
             teardown_mark.emplace(logger, "pass teardown");
             return result;
         }
@@ -1363,6 +1374,7 @@ ProcessResult Rugnux::RunPipeline(RugnuxObserver *observer, bool write_output, b
         TimingMark mark(logger, "leftover-lattice census wait");
         result.leftover_lattices = census_future.get();
     }
+    retire_outcomes();
     teardown_mark.emplace(logger, "pass teardown");
     return result;
 }

rugnux/Rugnux.h

@@ -1623,4 +1623,8 @@ public:
     // pass cut short gives way to the last complete one.
     void Cancel() { cancelled_ = true; }
     bool IsCancelled() const { return cancelled_; }
+
+private:
+    // The last pass's integrated observations being handed back (RunPipeline); last, so it is waited for first.
+    std::shared_future<void> retired_outcomes_;
 };

rugnux/RugnuxScaleMerge.cpp

@@ -3840,6 +3840,47 @@ bool Rugnux::ScaleMergeAndSymmetry(PipelineLocals &p) {
                                         report_shell_d_min, first_schedule);
         });
     }
+    // The molecular replacement (below): the search started on the geometry pre-pass's merge, where there is one,
+    // taken over where it holds on this one - not where the twin hypothesis' merge is searched too, as that search
+    // is this merge's own - and this merge searched itself where it does not.
+    const auto search_model = [&](const std::vector<MRMerge> &merges, Logger &log) {
+        if (early_mr_ && merges.size() == 1) {
+            TimingMark early_mark(log, "early molecular replacement wait");
+            try {
+                MolecularReplacementResult early = early_mr_->run.get();
+                early_mark.Stop();
+                early_mr_->log.ReplayInto(log);
+                if (auto carried = CarryMolecularReplacement(
+                        early, {"geometry pre-pass's merge", &early_mr_->merged, early_mr_->cell, early_mr_->space_group},
+                        merges[0], log, static_cast<size_t>(config_.nthreads), merge_on_card))
+                    return std::move(*carried);
+            } catch (const std::exception &e) {
+                log.Warning("Molecular replacement on the geometry pre-pass's merge failed ({}) - searched again",
+                            e.what());
+            }
+        }
+        return RunMolecularReplacement(*config_.model, merges, log, static_cast<size_t>(config_.nthreads),
+                                       merge_on_card);
+    };
+    // A prediction, or --mr, has no validation to start here, but the search reads of the merge only its amplitudes
+    // too. Where there is no twin hypothesis' merge to search beside it and no reference to take the free set from,
+    // nothing up to the search changes the merge or the group, so it starts here the same way, beside the analyses,
+    // on a copy given the same amplitudes. Its log is held, and printed where it is taken.
+    Logger model_search_log = Logger::Buffered();
+    std::future<MolecularReplacementResult> model_search_run;
+    if (model_run && (model_predicted || config_.force_mr) && !twin_merge && !model_twin_merge
+        && config_.reference_data.empty()) {
+        model_merged = sm.merged;
+        model_search_run = std::async(std::launch::async,
+                [&, device = get_gpu(), sg = experiment_.GetSpaceGroupOrP1()] {
+            set_gpu(device);
+            FrenchWilsonOptions fw_opts;
+            fw_opts.num_threads = config_.nthreads;
+            fw_opts.anisotropy_hkl = AnisotropyPriorHKL(model_merged, gemmi::UnitCell(*result.consensus_cell), twin_sg);
+            ApplyFrenchWilson(model_merged, sg, fw_opts);
+            return search_model({{"MERGE", &model_merged, *result.consensus_cell, sg}}, model_search_log);
+        });
+    }
     // Not on the geometry pre-pass, nor on a superseded one: the analysis goes into that pass's
     // statistics text and its written reflections, and neither survives the run. The promotion flag
     // below is a different thing - it is what the SEARCH did, the second pass reads it, and it is
@@ -4692,31 +4733,19 @@ bool Rugnux::ScaleMergeAndSymmetry(PipelineLocals &p) {
         if (!ledger_ahead.valid())
             rsm.reset();
         const size_t merge_on_card = rsm && rsm->OnGPU() ? rsm->DeviceBytes() : 0;
-        std::vector<MRMerge> merges{{"MERGE", &sm.merged, *result.consensus_cell, experiment_.GetSpaceGroupOrP1()}};
-        if (model_twin)
-            merges.push_back({"TWIN_SUBGROUP_MERGE", &model_twin->merged, model_twin->cell, model_twin->space_group,
-                              model_twin->law});
-        // The search started on the geometry pre-pass's merge, where there is one, taken over where it holds on
-        // this one. Not where the twin hypothesis' merge is searched too: that search is this merge's own.
-        std::optional<MolecularReplacementResult> carried;
-        if (early_mr_ && !model_twin) {
-            TimingMark early_mark(logger, "early molecular replacement wait");
-            try {
-                MolecularReplacementResult early = early_mr_->run.get();
-                early_mark.Stop();
-                early_mr_->log.ReplayInto(logger);
-                carried = CarryMolecularReplacement(
-                        early, {"geometry pre-pass's merge", &early_mr_->merged, early_mr_->cell, early_mr_->space_group},
-                        merges[0], logger, static_cast<size_t>(config_.nthreads), merge_on_card);
-            } catch (const std::exception &e) {
-                logger.Warning("Molecular replacement on the geometry pre-pass's merge failed ({}) - searched again",
-                               e.what());
-            }
+        MolecularReplacementResult mr;
+        if (model_search_run.valid()) {
+            TimingMark wait_mark(logger, "molecular replacement (beside the analyses) wait");
+            mr = model_search_run.get();
+            wait_mark.Stop();
+            model_search_log.ReplayInto(logger);
+        } else {
+            std::vector<MRMerge> merges{{"MERGE", &sm.merged, *result.consensus_cell, experiment_.GetSpaceGroupOrP1()}};
+            if (model_twin)
+                merges.push_back({"TWIN_SUBGROUP_MERGE", &model_twin->merged, model_twin->cell,
+                                  model_twin->space_group, model_twin->law});
+            mr = search_model(merges, logger);
         }
-        MolecularReplacementResult mr = carried ? std::move(*carried)
-                                                : RunMolecularReplacement(*config_.model, merges, logger,
-                                                                          static_cast<size_t>(config_.nthreads),
-                                                                          merge_on_card);
         if (mr.chosen >= 0) {
             const MRCandidate &c = mr.candidates[mr.chosen];
             if (mr.decision == "ENANTIOMORPH")