Live stills merge: prototype of a full scaling pass after the last frame (owner decision)

The incremental merge prepares each crystal after the first checkpoint once, against the checkpoint's
merge, and never scales the run as a whole; replayed on a stored run the loss scales with the share of
crystals prepared that way (R_split 0.244 cold, 0.288 with a checkpoint at half the frames, 0.443 at a
tenth; CC1/2 0.854 / 0.773 / 0.736 at one cut). StillsLiveMerge::Final can now scale and post-refine
every crystal over the whole run instead - the cold merge, at the cold merge's cost after the last
frame - and the broker's merge does so (FINAL_FULL_SCALING), a prototype against the live budget for
the owner to keep or revert. --live-final-full replays it in --mode scale --live-checkpoints, with the
time of the merge after the last frame in the log.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HVeDQsU5a8AJUdJPgWevAK

docs/RUGNUX_CLI.md

@@ -296,6 +296,7 @@ serial stills; `fft` (GPU) / `fftw` (CPU) index **de novo** and suit strong rota
 | `--relative-b[=deg]` | rot3d: fit a per-batch relative-*B* beyond the single decay slope over deg-degree batches, cross-validated (default 10° when bare; off otherwise) |
 | `--simple-stills` | Stills: treat every reflection as a full (*p* = 1, single-pass scale/merge) — disables the default-on physical partiality post-refinement |
 | `--live-checkpoints <f1>[,<f2>...]` | `--mode scale`, stills (for developers): replay the stored frames in image order as the broker's live merge of a serial collection merges them - a merge at each fraction of the images (e.g. `0.5,0.75`), each crystal after the first checkpoint scaled, post-refined and placed in its indexing against the latest checkpoint's merge as it comes, and after the last frame one merge of all of them without a scaling or post-refinement over the whole run. The log gives each checkpoint's statistics and time, and the time of the merge after the last frame; the merge written is that last one. Without it `--mode scale` merges cold, as the broker does when no checkpoint is reached Each fraction 0.01-0.99. |
+| `--live-final-full` | With `--live-checkpoints`: the merge after the last frame scales and post-refines every crystal over the whole run, as a cold merge does, instead of merging the crystals as they were prepared against the checkpoints - what the broker's live merge would do with a full scaling pass at the end; its time is in the log beside the incremental one's. |
 | `--no-expected-variance-merge` | Stills: disable the default expected-variance merge weighting (which rebuilds each weak observation's signal variance at the reflection mean to de-bias the inverse-variance merge); restores observed-sigma weighting |
 | `--capture-uncertainty <num>` | Systematic sigma num·(1−captured fraction)·⟨I⟩: on rot3d fulls the share of the rocking curve not captured, on stills the share of each partial its partiality does not see (default: 1.0 for rotation, 0 otherwise) |
 | `--min-captured-fraction <num>` | rot3d: drop a combined full whose rocking curve was captured below this fraction — edge-of-sweep truncated fulls (default: 0.7 for rotation, 0 otherwise; 0 = off) |

receiver/StillsGPUAnalysis.cpp

@@ -426,7 +426,7 @@ void StillsGPUAnalysis::MergeThread() {
         for (int64_t n = merge_queue.GetBlocking(); n >= 0; n = merge_queue.GetBlocking())
             add(n, indexer->TakeIntegrationOutcome(n));
         if (have_cell()) {
-            merged = merge->Final(static_cast<int>(AvailableCpus()));
+            merged = merge->Final(static_cast<int>(AvailableCpus()), FINAL_FULL_SCALING);
             logger.Info("Stills merge after the last image: {:.2f} s", merged->seconds);
         }
         // The frames' reflections are not read again: what the model step and the end message need is the merge

receiver/StillsGPUAnalysis.h

@@ -53,6 +53,10 @@ public:
     constexpr static int64_t PREPASS_MAX_IMAGES = 4 * PREPASS_IMAGES;
     constexpr static std::array<double, 2> CHECKPOINTS = {0.5, 0.75};
     constexpr static int MERGE_THREADS = 8;
+    // The merge after the last frame scales and post-refines every crystal over the whole run (StillsLiveMerge::Final):
+    // the cold merge's result at the cold merge's cost after the last frame, instead of the fixed-cost merge of the
+    // crystals as they were prepared against the checkpoints. Prototype: an owner decision against the live budget.
+    constexpr static bool FINAL_FULL_SCALING = true;
 
     // `start`: the collection's start message, as it would be sent without the pre-pass.
     StillsGPUAnalysis(const DiffractionExperiment &experiment, const PixelMask &mask, const StartMessage &start,

rugnux/StillsLiveMerge.cpp

@@ -199,8 +199,8 @@ void StillsLiveMerge::Prepare(int nthreads) {
     pending.clear();
 }
 
-StillsMergeOutput StillsLiveMerge::Final(int nthreads) {
-    if (!checkpointed)
+StillsMergeOutput StillsLiveMerge::Final(int nthreads, bool full_scaling) {
+    if (!checkpointed || full_scaling)
         return MergeStills(experiment, outcomes, given_cell, run_width, {}, flag_ice, ice_width, nthreads, logger);
     const auto t0 = std::chrono::steady_clock::now();
     Prepare(nthreads);
@@ -240,7 +240,8 @@ StillsMergeOutput ReplayLiveStillsMerge(const DiffractionExperiment &experiment,
                                         std::vector<IntegrationOutcome> &outcomes,
                                         const std::optional<UnitCell> &given_cell,
                                         const std::optional<StillsRunWidth> &run_width, bool flag_ice, float ice_width,
-                                        const std::vector<double> &checkpoints, int nthreads, Logger &logger) {
+                                        const std::vector<double> &checkpoints, bool final_full, int nthreads,
+                                        Logger &logger) {
     constexpr int64_t PREPARE_EVERY = 500;   // frames between two Prepare calls after a checkpoint
     // CPU time of the whole process, all threads: what each part costs the machine, whatever else runs on it
     const auto cpu_s = [] { return static_cast<double>(std::clock()) / CLOCKS_PER_SEC; };
@@ -268,7 +269,7 @@ StillsMergeOutput ReplayLiveStillsMerge(const DiffractionExperiment &experiment,
     logger.Info("Live: crystals prepared between the checkpoints and the end in {:.2f} s ({:.1f} CPU s)", prepare_s,
                 prepare_cpu_s);
     const double c0 = cpu_s();
-    auto out = live.Final(nthreads);
+    auto out = live.Final(nthreads, final_full);
     logger.Info("Live: the merge after the last frame took {:.2f} s ({:.1f} CPU s, {} frames)", out.seconds,
                 cpu_s() - c0, out.images);
     outcomes = live.TakeOutcomes();

rugnux/StillsLiveMerge.h

@@ -75,8 +75,10 @@ public:
     StillsMergeOutput Checkpoint(int nthreads);
     // The crystals added since the last checkpoint, each against its reference. No-op before the first one.
     void Prepare(int nthreads);
-    // The merge of the whole collection: the cold MergeStills where no checkpoint was made.
-    StillsMergeOutput Final(int nthreads);
+    // The merge of the whole collection: the cold MergeStills where no checkpoint was made, and with
+    // full_scaling after one too - every crystal scaled and post-refined over the whole run, as a cold merge
+    // of the same frames would, at the cold merge's cost after the last frame.
+    StillsMergeOutput Final(int nthreads, bool full_scaling = false);
     // The frames as merged, moved out.
     std::vector<IntegrationOutcome> TakeOutcomes();
 
@@ -107,12 +109,14 @@ private:
 // The frames of `outcomes` given to a StillsLiveMerge one by one in image order, as a live collection gives
 // them, with a checkpoint at each of the fractions of the images in `checkpoints` (ascending) and the
 // crystals after one prepared every few hundred frames; the timings logged. For --mode scale
-// --live-checkpoints, to measure what a live run would do on a collection already integrated.
+// --live-checkpoints, to measure what a live run would do on a collection already integrated; final_full
+// is Final's full_scaling (--live-final-full).
 StillsMergeOutput ReplayLiveStillsMerge(const DiffractionExperiment &experiment,
                                         std::vector<IntegrationOutcome> &outcomes,
                                         const std::optional<UnitCell> &given_cell,
                                         const std::optional<StillsRunWidth> &run_width, bool flag_ice, float ice_width,
-                                        const std::vector<double> &checkpoints, int nthreads, Logger &logger);
+                                        const std::vector<double> &checkpoints, bool final_full, int nthreads,
+                                        Logger &logger);
 
 // The result the report is read from (BuildReportDocument), of a merge made here: the cell and group merged in
 // (`experiment`), the merge's statistics and error model, the frames analysed and indexed, the indexing

rugnux/rugnux_cli.cpp

@@ -264,6 +264,7 @@ enum {
     OPT_INTEGRATOR,
     OPT_SIMPLE_STILLS,
     OPT_LIVE_CHECKPOINTS,
+    OPT_LIVE_FINAL_FULL,
     OPT_SCALE_FULLS,
     OPT_CAPTURE_UNCERTAINTY,
     OPT_MIN_CAPTURED_FRACTION,
@@ -448,6 +449,7 @@ static option long_options[] = {
     {"integrator", required_argument, nullptr, OPT_INTEGRATOR},
     {"simple-stills", no_argument, nullptr, OPT_SIMPLE_STILLS},
     {"live-checkpoints", required_argument, nullptr, OPT_LIVE_CHECKPOINTS},
+    {"live-final-full", no_argument, nullptr, OPT_LIVE_FINAL_FULL},
     {"detect-ice-rings", optional_argument, nullptr, OPT_DETECT_ICE_RINGS},
     {"index-ice-rings", optional_argument, nullptr, OPT_INDEX_ICE_RINGS},
     {"ice-min-score", required_argument, nullptr, OPT_ICE_MIN_SCORE},
@@ -998,6 +1000,7 @@ struct RugnuxCli {
     std::optional<IntegratorMode> integrator_mode;     // --integrator boxsum|gaussian|empirical
     bool simple_stills_flag = false;                   // --simple-stills: disable the default stills partiality post-refinement
     std::vector<double> live_checkpoints;              // --live-checkpoints: --mode scale stills replayed as a live run
+    bool live_final_full = false;                      // --live-final-full: that replay's merge after the last frame scales over the whole run
     std::optional<double> outlier_reject_nsigma;       // merge per-observation outlier rejection
 
     std::vector<std::string> input_files;
@@ -1563,6 +1566,9 @@ int RugnuxCli::Parse(int argc, char **argv) {
             std::sort(live_checkpoints.begin(), live_checkpoints.end());
             break;
         }
+        case OPT_LIVE_FINAL_FULL:
+            live_final_full = true;
+            break;
         case OPT_REJECT_OUTLIERS:
             outlier_reject_nsigma = parse_double_arg(optarg, "--reject-outliers", logger);
             break;
@@ -2649,7 +2655,7 @@ int RugnuxInputRun::PrepareRugnuxInput() {
             else
                 so = ReplayLiveStillsMerge(experiment, reflections, dataset->stills_given_cell,
                                            dataset->stills_run_width, flag_ice, ice_width, live_checkpoints,
-                                           nthreads, logger);
+                                           live_final_full, nthreads, logger);
             experiment.ImagesPerTrigger(static_cast<int64_t>(so.images));
             auto &sm = so.merge;
             merged_reflections = std::move(sm.merged);