rugnux: offline, the first passes decide their near ties on all the frames from the start

A stored sweep is all there when the run starts, so a first pass now decides a near tie on its
validation frames and the later frames together at once, instead of deciding on the validation frames,
re-checking the pass after the run and making the run again where the re-check ends elsewhere. Live
(following a collection from the broker) keeps that early path unchanged: ProcessConfig::live.

The pre-pass decisions - the rotation-scale walk's refits, the demoted metric-symmetry arm, the
integration radius - stay on the pre-pass frames offline too, with their near-tie re-checks after the
run; only the run made again decides them on all the frames (decide_all_late_, split from decide_late_).

--simulate-live <ms> reads a stored sweep as the collection it was while being written - image i
readable (i+1)*ms after the run starts, 0 = all at once - and takes the live path, so that path can be
tested and run in the battery without a broker.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01EizBKhTqrAYago9KJG3dA3

rugnux/Rugnux.cpp

@@ -388,11 +388,16 @@ ProcessResult Rugnux::Run(RugnuxObserver *observer) {
     // first pass being invisible in it. Time the whole thing here, where every pass is inside.
     const auto run_start = std::chrono::steady_clock::now();
     DiffractionExperiment header_experiment = experiment_;
+    // A stored sweep is all there from the start, so its first passes decide their near ties on all of
+    // its frames at once. Live, only the frames in so far are: the first passes decide on those, and
+    // their near ties are re-checked once the collection is over (decide_late_).
+    decide_late_ = !config_.live;
     // The run as it starts, to be made again where a near tie decided early ends elsewhere on the rest
-    // of the sweep (decide_late_).
+    // of the sweep.
     const auto again = std::make_shared<Rugnux>(*this);
     again->near_tie_rechecks_ = std::make_shared<NearTieRechecks>();
     again->decide_late_ = true;
+    again->decide_all_late_ = true;
     Rugnux *run = this;
     ProcessResult result;
     bool lost = false;

rugnux/Rugnux.h

@@ -105,6 +105,11 @@ struct ProcessConfig {
     // keep the fixed budget they are configured with.
     bool measure_spot_budget = false;
 
+    // The images arrive while the run goes on, from a collection still being written (rugnux following
+    // a broker, or --simulate-live), rather than all being there from the start. Decides how near ties
+    // are decided (Rugnux::decide_late_).
+    bool live = false;
+
     // Rotation indexing (FullAnalysis)
     bool rotation_indexing = false;
     bool two_pass_rotation = true;
@@ -991,18 +996,24 @@ class Rugnux {
     // way integrates the same reflections and can only tie.
     std::string stop_at_first_pass_outcome_;
 
-    // --- Near ties: decided early, re-checked late (see FirstPassRun::DecideLate) ---
+    // --- Near ties: offline decided late, live decided early and re-checked late (see FirstPassRun::DecideLate) ---
     //
-    // A first pass acts on its validation frames at once. Where it decided a near tie that the rest
-    // of the sweep could re-check, the copy of the run taken as its pass began is kept, and once the
-    // run is over - every frame is in by then, live too - that pass is run again on it as a probe,
-    // deciding late. Only where a re-check's first pass ends elsewhere than the pass's did is the
-    // run made again, every first pass deciding late. Offline the answer is then the same as deciding
-    // late throughout; live, no near tie holds the run up while the collection goes on.
+    // Offline every frame is there from the start, and a first pass decides a near tie on the later
+    // frames as well as its validation frames. Live (ProcessConfig::live), a first pass acts on its
+    // validation frames at once. Where it decided a near tie that the rest of the sweep could re-check,
+    // the copy of the run taken as its pass began is kept, and once the run is over - every frame is in
+    // by then - that pass is run again on it as a probe, deciding late. Only where a re-check's first
+    // pass ends elsewhere than the pass's did is the run made again, deciding everything late; no near
+    // tie holds the run up while the collection goes on.
     //
-    // Whether the first passes decide their near ties on all the frames: on a re-check, and on the run
-    // made again.
+    // Whether the first passes decide their near ties on all the frames: offline, on a re-check, and on
+    // the run made again.
     bool decide_late_ = false;
+    // Whether what the run otherwise decides on the pre-pass frames - the rotation-scale walk's refits,
+    // the demoted metric-symmetry arm, the integration radius (RefinedPasses) - is decided on all the
+    // frames: only on the run made again. Offline too these are decided on the pre-pass frames, and a near
+    // tie among them is re-checked after the run.
+    bool decide_all_late_ = false;
     // Whether this copy of the run is a re-check (for the log).
     bool near_tie_recheck_ = false;
     // Set by the first pass now running where it decided a near tie the later frames can re-check.

rugnux/RugnuxFirstPass.h

@@ -172,15 +172,15 @@ struct Rugnux::FirstPassRun {
     bool NearTiePooled(const std::string &decision, const std::string &a, const PooledEvidence &ea,
                        const std::string &b, const PooledEvidence &eb) const;
 
-    // A near tie is decided early and checked late. The pass acts on the validation frames at once;
-    // the comparison made again, unchanged, on the validation frames and late_validation together -
-    // a gate set at a share of the validation frames at the same share of all of them - is the
-    // re-check, which Run() makes once the rest of the sweep is in (Rugnux::decide_late_). A
+    // A near tie is decided on the validation frames and late_validation together - the comparison made
+    // again, unchanged, a gate set at a share of the validation frames at the same share of all of
+    // them. Offline that is the decision; live the pass acts on the validation frames at once, and this
+    // is the re-check, which Run() makes once the rest of the sweep is in (Rugnux::decide_late_). A
     // decision that is not a near tie never reads later frames; nor does one on a sweep the first
     // part already covers, where there is nothing later to read (late_validation, below).
     //
-    // Whether this near tie is decided on all the frames now: on the re-check, and on the run Run()
-    // makes again where a re-check ended elsewhere. Otherwise it notes that the pass carried one.
+    // Whether this near tie is decided on all the frames now: offline, on the re-check, and on the run
+    // Run() makes again where a re-check ended elsewhere. Otherwise it notes that the pass carried one.
     bool DecideLate() const;
     // The validation frames followed by late_validation.
     std::vector<int> AllValidationFrames() const;

rugnux/RugnuxImagePass.cpp

@@ -147,9 +147,9 @@ void Rugnux::ImagePass(PipelineLocals &p) {
     // The crowding of the background rings at the widened radii, counted on frames spread over the
     // pass (widened_crowding_): on the geometry pre-pass, which integrates at the fixed radius while the
     // widened one is held back (bragg_adaptive_), and on a canonical pass a near tie left at the fixed
-    // radius (radius_recheck_). A run deciding late decides the radius on the canonical pass itself.
+    // radius (radius_recheck_). The run made again decides the radius on the canonical pass itself.
     const std::optional<BraggIntegrationSettings> crowding_at =
-            geometry_prepass ? (decide_late_ ? std::nullopt : bragg_adaptive_)
+            geometry_prepass ? (decide_all_late_ ? std::nullopt : bragg_adaptive_)
             : prepass_frames_only_ ? std::nullopt : radius_recheck_;
     const int crowding_stride = std::max(1, (geometry_prepass ? prepass_images : images_to_process)
                                             / CROWDING_FRAMES);

rugnux/RugnuxPasses.cpp

@@ -668,7 +668,7 @@ RotationScaleWalk Rugnux::WalkRotationScaleBeside(const ProcessResult &pass1, Ro
             // A refit reads only the scale its post-refinement fits on top of k - a step for the walk
             // to take, which the validation frames of the whole sweep then judge - so it is fitted on
             // the frames pass 1 fitted the first one on.
-            r->prepass_frames_only_ = !index_only && !decide_late_;
+            r->prepass_frames_only_ = !index_only && !decide_all_late_;
             // An indexing probe runs beside the refined passes, on the same card, and gives its memory
             // back in seconds: what of theirs does not fit beside it waits for it to end. A refit
             // runs only once they have ended, so it is nobody's work beside.
@@ -984,7 +984,7 @@ std::optional<ProcessResult> Rugnux::RefinedPasses(RugnuxObserver *observer, Pro
             demoted_run.lattice_angles_demoted_ = angle_arm;
             demoted_run.stop_at_first_pass_outcome_ = promoted_probe.first_pass_outcome_;
             // The arm is asked on the frames the pre-pass read, as the pre-pass asked its own fit, and
-            // the canonical pass compares its fit on the same frames. A run deciding late asks it on
+            // the canonical pass compares its fit on the same frames. The run made again asks it on
             // all of them; a near tie keeps this copy to do that once the run is over.
             demoted_recheck = std::make_shared<Rugnux>(demoted_run);
             demoted_recheck->decide_late_ = true;
@@ -992,7 +992,7 @@ std::optional<ProcessResult> Rugnux::RefinedPasses(RugnuxObserver *observer, Pro
             demoted_recheck->first_pass_memo_.clear();
             demoted_recheck->speculative_probe_.reset();
             demoted_recheck->rotation_scale_check_.reset();
-            demoted_run.prepass_frames_only_ = !decide_late_;
+            demoted_run.prepass_frames_only_ = !decide_all_late_;
             demoted_arm = demoted_run.RunPipeline(observer, /*write_output=*/false, /*geometry_prepass=*/false);
             ++arm_passes;
             for (const auto &m : demoted_run.first_pass_memo_)

rugnux/rugnux_cli.cpp

@@ -106,6 +106,12 @@ void print_usage() {
                  "collection delivered. Rotation data, --mode mx only. The collection's token is read from the "
                  "environment variable JUNGFRAUJOCH_HTTP_TOKEN and sent only to the broker. With <input.h5> on disk, "
                  "the URL is ignored" << std::endl;
+    std::cout << "   --simulate-live <ms>             Read the stored <input.h5> as the collection it was while being written, "
+                 "the way a broker URL follows one: image i becomes readable (i+1)*<ms> ms after the run starts (0: all "
+                 "at once). The run then takes the live path - the first passes decide their near ties on the images in "
+                 "so far, re-checked once the last one is in, and the run is made again where one ends elsewhere - "
+                 "where on a stored sweep they decide on all of its images at once. Same result; for testing the "
+                 "live workflow without a broker. --mode mx only" << std::endl;
     std::cout << "Options:" << std::endl;
     std::cout << "   -o, --output-prefix <txt>        Output file prefix (default: output). mx and scale runs always write "
                  "<prefix>_report.txt, the results report; every run writes <prefix>_log.txt, the full timestamped log "
@@ -310,6 +316,7 @@ enum {
     OPT_NO_SCALE_FULLS,
     OPT_WRITE_PROCESS_H5,
     OPT_FINALIST_LEDGER,
+    OPT_SIMULATE_LIVE,
     OPT_DEVELOPER,
     OPT_FORCE_STILL,
     OPT_AZIM_MIN_Q,
@@ -366,6 +373,7 @@ static option long_options[] = {
     {"no-scale-fulls", no_argument, nullptr, OPT_NO_SCALE_FULLS},
     {"write-process-h5", no_argument, nullptr, OPT_WRITE_PROCESS_H5},
     {"finalist-ledger", no_argument, nullptr, OPT_FINALIST_LEDGER},
+    {"simulate-live", required_argument, nullptr, OPT_SIMULATE_LIVE},
     {"developer", no_argument, nullptr, OPT_DEVELOPER},
     {"smooth-g", optional_argument, nullptr, OPT_SMOOTH_G},
     {"relative-b", optional_argument, nullptr, OPT_RELATIVE_B},
@@ -652,6 +660,20 @@ void configure_offline_output(DiffractionExperiment &experiment, const std::stri
 namespace {
     std::atomic<Rugnux *> g_active_process{nullptr};
 
+    // --simulate-live: a stored sweep read as the collection it was while being written - image i is
+    // read no earlier than (i + 1) frame times after the run began.
+    std::chrono::microseconds simulated_frame_time{0};
+    std::chrono::steady_clock::time_point simulated_start;
+
+    template <class Reader>
+    class SimulatedLive : public Reader {
+    public:
+        bool ReadRawImage(int64_t image_number, JFJochReaderRawImage &image) override {
+            std::this_thread::sleep_until(simulated_start + simulated_frame_time * (image_number + 1));
+            return Reader::ReadRawImage(image_number, image);
+        }
+    };
+
     // The terminal (see RugnuxConsole). Made first thing in RunRugnux; main uses it too, to close the
     // run on an error that escapes it.
     std::shared_ptr<RugnuxConsole> console;
@@ -907,6 +929,7 @@ static int RunRugnux(int argc, char **argv) {
     bool run_scaling = true;      // merge is on by default; --no-merge turns it off
     std::optional<bool> scale_fulls_arg;   // --scale-fulls / --no-scale-fulls; default on for rot3d
     bool write_process_h5_flag = false;    // --write-process-h5; also write _process.h5 when merging
+    bool simulate_live = false;            // --simulate-live
     bool finalist_ledger_flag = false;     // --finalist-ledger; report-only symmetry evidence table
     std::optional<bool> detect_ice_rings;  // --detect-ice-rings[=on|off]; unset => use the dataset (file) value
     bool index_ice_rings = false;          // --index-ice-rings[=on|off]; index on the ice-band spots too
@@ -1330,6 +1353,11 @@ static int RunRugnux(int argc, char **argv) {
         case OPT_FINALIST_LEDGER:
             finalist_ledger_flag = true;
             break;
+        case OPT_SIMULATE_LIVE:
+            simulate_live = true;
+            simulated_frame_time = std::chrono::microseconds(
+                    std::lround(1000.0 * parse_double_arg(optarg, "--simulate-live", logger)));
+            break;
         case OPT_DEVELOPER:
             provenance.developer = true;
             break;
@@ -1632,10 +1660,10 @@ static int RunRugnux(int argc, char **argv) {
     // any frame of a sweep reads the whole sweep, which is how these are archived and how a user
     // thinks of them.
     console->OnStep("Reading input");
-    JFJochHDF5Reader hdf5_reader;
-    JFJochCBFReader cbf_reader;
-    JFJochMarCCDReader marccd_reader;
-    JFJochSMVReader smv_reader;
+    SimulatedLive<JFJochHDF5Reader> hdf5_reader;
+    SimulatedLive<JFJochCBFReader> cbf_reader;
+    SimulatedLive<JFJochMarCCDReader> marccd_reader;
+    SimulatedLive<JFJochSMVReader> smv_reader;
     JFJochStartMessageReader live_reader;
     std::unique_ptr<BrokerFeed> broker_feed;
     JFJochReader *reader_ptr = nullptr;
@@ -3103,6 +3131,11 @@ static int RunRugnux(int argc, char **argv) {
     // _process.h5 unless explicitly requested. Without merging, the _process.h5 is the only output.
     config.write_process_h5 = run_scaling ? write_process_h5_flag : true;
     config.finalist_ledger = finalist_ledger_flag;
+    config.live = broker_feed || simulate_live;
+    if (simulate_live)
+        logger.Info("Simulated live collection: image i readable (i+1) x {:.1f} ms after the run starts",
+                    simulated_frame_time.count() / 1000.0);
+    simulated_start = std::chrono::steady_clock::now();
 
     Rugnux process(reader, experiment, *dataset->pixel_mask, config);