Fix acquisition error-handling and rugnux CLI issues from branch review

Broker acquisition failure paths:
- Report a truncated writer output / missing packets as errors ahead of the
  queue-full warning, so an incomplete dataset is no longer masked as a
  "reduce frame rate" warning.
- Route PCIe/FPGA hardware faults to the Error state: make PCIeDeviceException a
  JFJochCriticalException, and stop JFJochServices::Stop() from slicing the
  caught exception (capture/rethrow via exception_ptr) so the critical type
  survives to the state machine.
- Stop the (possibly partially-armed) detector, not just the receiver, when a
  detector arm fails, so the next start does not see it BUSY.
- Return structured Error_message JSON for bind_json operation failures (e.g. a
  failed synchronous /start), matching the no-arg endpoints.

rugnux CLI:
- Default -N to all hardware threads, resolved once after parsing so full
  analysis, --azint-only and --scale behave the same.
- Parse numeric option arguments strictly (reject non-numeric input and trailing
  garbage) instead of atof/stod, which silently yielded 0 or aborted the
  process; require --wavelength > 0.
- Emit -R attached (-R100) in the reproduced command line so it re-parses.
- Compare the reference MTZ against the cell that actually drives the merge in
  the --scale path.
- Correct the --scaling-high-resolution help text and de-duplicate the
  offline-output experiment setup.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

broker/JFJochBrokerHttp.cpp

@@ -155,13 +155,23 @@ void JFJochBrokerHttp::register_routes(httplib::Server &server) {
     auto bind_json = [this](auto method, auto model_tag) {
         return [this, method](const httplib::Request &req, httplib::Response &res) {
             using Model = decltype(model_tag);
+            Model v;
             try {
-                Model v;
                 nlohmann::json::parse(req.body).get_to(v);
                 v.validate();
+            } catch (const std::exception &e) {
+                // Malformed or invalid request body.
+                auto [c, s] = handleParsingException(e);
+                send_plain(res, c, s);
+                return;
+            }
+            try {
                 (this->*method)(v, res);
             } catch (const std::exception &e) {
-                auto [c, s] = handleParsingException(e);
+                // Operation failure (e.g. WrongDAQState, a failed synchronous start): return the
+                // structured Error_message JSON with a reason field, like the no-arg endpoints, so
+                // clients can distinguish the cause instead of receiving opaque plain text.
+                auto [c, s] = handleOperationException(e);
                 send_plain(res, c, s);
             }
         };

broker/JFJochServices.cpp

@@ -36,14 +36,21 @@ void JFJochServices::Start(const DiffractionExperiment& experiment,
             detector->Start(experiment);
         }
     } catch (const std::exception &e) {
-        logger.Error("   ... detector failed to start ({}) - stopping receiver", e.what());
-        // Best-effort receiver cleanup - it must not replace the original detector exception (which
-        // may be critical), so swallow anything it throws and let the throw below re-raise e.
+        logger.Error("   ... detector failed to start ({}) - stopping detector and receiver", e.what());
+        // Best-effort cleanup - it must not replace the original detector exception (which may be
+        // critical), so swallow anything it throws and let the throw below re-raise e. The detector
+        // may have partially armed before failing, so stop it too; otherwise it lingers in a BUSY
+        // state and the next start fails with "detector busy" until a manual re-initialisation.
         try {
+            {
+                std::shared_lock ul(detector_mutex);
+                if (detector)
+                    detector->Stop();
+            }
             receiver->Cancel(false);
             receiver->Stop();
         } catch (const std::exception &stop_error) {
-            logger.Warning("Receiver stop after failed start reported: {}", stop_error.what());
+            logger.Warning("Stop after failed start reported: {}", stop_error.what());
         }
         throw;
     }
@@ -96,7 +103,10 @@ void JFJochServices::On(DiffractionExperiment &x) {
 JFJochServicesOutput JFJochServices::Stop() {
     JFJochServicesOutput ret;
 
-    std::unique_ptr<JFJochException> exception;
+    // Captured as an exception_ptr (not a copied JFJochException) so the dynamic type survives: a
+    // critical hardware fault (e.g. PCIeDeviceException) must stay critical when re-thrown, otherwise
+    // slicing it to a plain JFJochException would send the state machine to Idle instead of Error.
+    std::exception_ptr exception;
 
     bool detector_error = false;
 
@@ -149,7 +159,7 @@ JFJochServicesOutput JFJochServices::Stop() {
             logger.Info("    ... finished with success");
         } catch (const JFJochException &e) {
             logger.Error("   ... finished with error {}", e.what());
-            exception = std::make_unique<JFJochException>(e);
+            exception = std::current_exception();
         }
     } else {
         logger.Info("No receiver - sleeping for 30 seconds");
@@ -165,7 +175,7 @@ JFJochServicesOutput JFJochServices::Stop() {
         throw JFJochCriticalException("Error in detector operation");
 
     if (exception)
-        throw JFJochException(*exception);
+        std::rethrow_exception(exception);
 
     return ret;
 }

broker/JFJochStateMachine.cpp

@@ -445,11 +445,11 @@ void JFJochStateMachine::MeasurementThread() {
                         image_mean_time.integration * 1e6,
                         image_mean_time.processing * 1e6);
 
-            if (tmp_output.receiver_output.writer_queue_full_warning)
-                SetState(JFJochState::Idle,
-                         "Stream receiver (writer or downstream analysis) cannot cope with data; reduce frame rate",
-                         BrokerStatus::MessageSeverity::Warning);
-            else if (tmp_output.receiver_output.status.cancelled)
+            // Priority order matters. A cancel is checked first (it legitimately leaves efficiency < 1),
+            // then the hard errors (missing packets, truncated writer output). The queue-full warning is
+            // only the primary status when the run otherwise succeeded - it must not mask a real error
+            // by downgrading an incomplete/truncated dataset to a "reduce frame rate" warning.
+            if (tmp_output.receiver_output.status.cancelled)
                 SetState(JFJochState::Idle,
                          "Data collection cancelled",
                          BrokerStatus::MessageSeverity::Info);
@@ -461,6 +461,10 @@ void JFJochStateMachine::MeasurementThread() {
                 SetState(JFJochState::Idle,
                          tmp_output.receiver_output.writer_err,
                          BrokerStatus::MessageSeverity::Error);
+            else if (tmp_output.receiver_output.writer_queue_full_warning)
+                SetState(JFJochState::Idle,
+                         "Stream receiver (writer or downstream analysis) cannot cope with data; reduce frame rate",
+                         BrokerStatus::MessageSeverity::Warning);
             else
                 SetState(JFJochState::Idle,
                          "Data collection without problems",

common/JFJochException.h

@@ -165,10 +165,13 @@ public:
             : JFJochException(in_category, description) {}
 };
 
-class PCIeDeviceException : public JFJochException {
+// A PCIe/FPGA hardware fault leaves the acquisition device in an undefined state that a retry cannot
+// recover from, so it is critical: the state machine forces re-initialisation (Error state) rather
+// than returning to Idle as it does for ordinary acquisition failures.
+class PCIeDeviceException : public JFJochCriticalException {
 public:
     explicit PCIeDeviceException(const std::string &description)
-    : JFJochException(JFJochExceptionCategory::PCIeError, description) {
+    : JFJochCriticalException(description, JFJochExceptionCategory::PCIeError) {
         msg += " (" + std::string(strerror(errno)) + ")";
     }
 };

rugnux/RugnuxCommandLine.cpp

@@ -108,7 +108,9 @@ std::string RugnuxCommandLine(const ProcessConfig &config,
 
         if (config.rotation_indexing) {
             if (config.two_pass_rotation)
-                add("-R", std::to_string(config.rotation_indexing_image_count));
+                // -R takes an optional argument, which getopt only accepts attached (-R100), never as a
+                // separate token - so emit it joined or the copied command line will not re-parse.
+                args.push_back("-R" + std::to_string(config.rotation_indexing_image_count));
             else
                 args.emplace_back("--single-pass-rotation");
             if (!config.reuse_rotation_spots)

tests/RugnuxTest.cpp

@@ -158,7 +158,10 @@ TEST_CASE("RugnuxCommandLine_Full", "[process]") {
     CHECK(cmd.find("-o run1") != std::string::npos);
     CHECK(cmd.find("-X fft") != std::string::npos);
     CHECK(cmd.find("-S 96") != std::string::npos);
-    CHECK(cmd.find("-R 30") != std::string::npos);
+    // -R takes an optional argument, so its value must be attached (-R30); a separate "-R 30" token
+    // would not re-parse (getopt would leave 30 as a positional and drop the count).
+    CHECK(cmd.find("-R30") != std::string::npos);
+    CHECK(cmd.find("-R 30") == std::string::npos);
     CHECK(cmd.find("/data/lyso_master.h5") != std::string::npos);
 }
 

tools/rugnux_cli.cpp

@@ -9,6 +9,7 @@
 #include <iostream>
 #include <sstream>
 #include <string>
+#include <thread>
 #include <vector>
 
 #include "../reader/JFJochHDF5Reader.h"
@@ -37,7 +38,7 @@ void print_usage() {
     std::cout << "Usage rugnux {<options>} <input.h5>" << std::endl;
     std::cout << "Options:" << std::endl;
     std::cout << "   -o, --output-prefix <txt>        Output file prefix (default: output)" << std::endl;
-    std::cout << "   -N, --threads <num>              Number of threads (default: 1)" << std::endl;
+    std::cout << "   -N, --threads <num>              Number of threads (default: all hardware threads)" << std::endl;
     std::cout << "   -s, --start-image <num>          Start image number (default: 0)" << std::endl;
     std::cout << "   -e, --end-image <num>            End image number (default: all)" << std::endl;
     std::cout << "   -t, --stride <num>               Image stride (default: 1)" << std::endl;
@@ -78,7 +79,7 @@ void print_usage() {
     std::cout << "   --smooth-g[=deg]                rot3d: smooth per-frame scale G over a deg-degree rotation range (XDS DELPHI-like) before the combine (default: 5 for rot3d; 0 = off)" << std::endl;
     std::cout << "   -A, --anomalous                 Anomalous mode (don't merge Friedel pairs)" << std::endl;
     std::cout << "   -B, --refine-bfactor            Refine per image B-factor" << std::endl;
-    std::cout << "   --scaling-high-resolution <num> High resolution limit for spot finding (default: no limit)" << std::endl;
+    std::cout << "   --scaling-high-resolution <num> High resolution limit for scaling/merging (default: no limit)" << std::endl;
     std::cout << "   --min-partiality <num>          Minimum partiality to accept reflection (default: 0.02)" << std::endl;
     std::cout << "   --capture-uncertainty <num>     rot3d: systematic sigma ~num*(1-captured_fraction)*I on under-captured fulls (default: 1.0 for rot3d, 0 otherwise)" << std::endl;
     std::cout << "   --mosaicity <num>               Diagnostic: fix the scaling mosaicity (deg) instead of the per-image seed" << std::endl;
@@ -240,6 +241,30 @@ bool parse_float_strict(const std::string &t, float &out) {
     }
 };
 
+// Parse a required numeric option argument, or print a clear error and exit. getopt hands option
+// arguments over as raw C strings; atof() silently returns 0 on non-numeric input (so a typo like
+// "--beam-x 21OO" would move the beam to pixel 0) and std::stod() throws std::invalid_argument,
+// which terminates the program. This rejects both, plus trailing garbage like "1.5foo".
+double parse_double_arg(const char *arg, const char *option_name, Logger &logger) {
+    std::string s = arg ? arg : "";
+    trim_in_place(s);
+    if (!s.empty()) {
+        try {
+            size_t idx = 0;
+            double out = std::stod(s, &idx);
+            if (idx == s.size())
+                return out;
+        } catch (...) {}
+    }
+    logger.Error("Invalid numeric value for {}: '{}'", option_name, arg ? arg : "<null>");
+    print_usage();
+    exit(EXIT_FAILURE);
+}
+
+float parse_float_arg(const char *arg, const char *option_name, Logger &logger) {
+    return static_cast<float>(parse_double_arg(arg, option_name, logger));
+}
+
 bool parse_on_off(const char *arg, bool &out) {
     std::string s = arg ? arg : "";
     std::transform(s.begin(), s.end(), s.begin(),
@@ -335,6 +360,19 @@ std::optional<CrystalLattice> parse_lattice_arg(const char *arg) {
     return CrystalLattice(vals);
 }
 
+// Shared offline-output settings for the _process.h5 / reflection writer, used by both the --scale
+// path and the full-analysis path (each then sets its own space group and images-per-trigger).
+void configure_offline_output(DiffractionExperiment &experiment, const std::string &output_prefix) {
+    experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4);
+    experiment.FilePrefix(output_prefix);
+    experiment.Mode(DetectorMode::Standard);   // full image analysis
+    experiment.PixelSigned(true);
+    experiment.OverwriteExistingFiles(true);
+    experiment.PolarizationFactor(0.99);
+    experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy);
+    experiment.NumTriggers(1);
+}
+
 namespace {
     std::atomic<Rugnux *> g_active_process{nullptr};
     void handle_sigint(int) {
@@ -358,7 +396,7 @@ int main(int argc, char **argv) {
 
     std::string input_file;
     std::string output_prefix = "output";
-    int nthreads = 1;
+    int nthreads = 0;   // 0 = auto: resolved to all hardware threads after parsing (see below)
     int start_image = 0;
     int end_image = -1; // -1 indicates process until end
     int image_stride = 1;
@@ -613,16 +651,16 @@ int main(int argc, char **argv) {
             write_process_h5_flag = true;
             break;
         case OPT_SMOOTH_G:
-            smooth_g_deg_arg = optarg ? std::stod(optarg) : SMOOTH_G_DEFAULT_DEG;
+            smooth_g_deg_arg = optarg ? parse_double_arg(optarg, "--smooth-g", logger) : SMOOTH_G_DEFAULT_DEG;
             break;
         case OPT_MIN_PARTIALITY:
-            min_partiality = std::stod(optarg);
+            min_partiality = parse_double_arg(optarg, "--min-partiality", logger);
             break;
         case OPT_CAPTURE_UNCERTAINTY:
-            capture_uncertainty_arg = std::stod(optarg);
+            capture_uncertainty_arg = parse_double_arg(optarg, "--capture-uncertainty", logger);
             break;
         case OPT_MOSAICITY:
-            forced_mosaicity_arg = std::stod(optarg);
+            forced_mosaicity_arg = parse_double_arg(optarg, "--mosaicity", logger);
             break;
         case OPT_INTEGRATION_RADIUS:
             integration_radius_arg = optarg;
@@ -634,13 +672,13 @@ int main(int argc, char **argv) {
             else { logger.Error("--integrator expects boxsum|gaussian|empirical"); return 1; }
             break;
         case OPT_REJECT_OUTLIERS:
-            outlier_reject_nsigma = std::stod(optarg);
+            outlier_reject_nsigma = parse_double_arg(optarg, "--reject-outliers", logger);
             break;
         case OPT_REJECT_DELTA_CCHALF:
-            delta_cchalf_nsigma = std::stod(optarg);
+            delta_cchalf_nsigma = parse_double_arg(optarg, "--reject-delta-cchalf", logger);
             break;
         case OPT_MIN_IMAGE_CC:
-            min_image_cc = std::stod(optarg);
+            min_image_cc = parse_double_arg(optarg, "--min-image-cc", logger);
             break;
         case OPT_SCALING_HIGH_RESOLUTION:
             d_min_scale_merge = atof(optarg);
@@ -690,13 +728,23 @@ int main(int argc, char **argv) {
             solid_angle_correction = value;
             break;
         }
-        case OPT_BEAM_X: beam_x = atof(optarg); break;
-        case OPT_BEAM_Y: beam_y = atof(optarg); break;
-        case OPT_DETECTOR_DISTANCE: detector_distance_mm = atof(optarg); break;
-        case OPT_WAVELENGTH: wavelength_A = atof(optarg); break;
-        case OPT_ROT1: rot1_rad = atof(optarg); break;
-        case OPT_ROT2: rot2_rad = atof(optarg); break;
-        case OPT_POLARIZATION: polarization_factor = atof(optarg); break;
+        case OPT_BEAM_X: beam_x = parse_float_arg(optarg, "--beam-x", logger); break;
+        case OPT_BEAM_Y: beam_y = parse_float_arg(optarg, "--beam-y", logger); break;
+        case OPT_DETECTOR_DISTANCE: detector_distance_mm = parse_float_arg(optarg, "--detector-distance", logger); break;
+        case OPT_WAVELENGTH: {
+            // Guard > 0: wavelength is used as a divisor (WVL_1A_IN_KEV / wavelength) below, and a 0
+            // would produce a non-finite incident energy that throws unguarded and aborts the process.
+            float w = parse_float_arg(optarg, "--wavelength", logger);
+            if (!(w > 0.0f)) {
+                logger.Error("Invalid wavelength (must be > 0 A): {}", optarg);
+                exit(EXIT_FAILURE);
+            }
+            wavelength_A = w;
+            break;
+        }
+        case OPT_ROT1: rot1_rad = parse_float_arg(optarg, "--rot1", logger); break;
+        case OPT_ROT2: rot2_rad = parse_float_arg(optarg, "--rot2", logger); break;
+        case OPT_POLARIZATION: polarization_factor = parse_float_arg(optarg, "--polarization", logger); break;
         case OPT_SCALING_ITERATIONS:
             scaling_iter = atoi(optarg);
             if (scaling_iter <= 0) {
@@ -727,6 +775,15 @@ int main(int argc, char **argv) {
     input_file = argv[optind];
     logger.Verbose(verbose);
 
+    // -N defaults to 0 = "use all hardware threads"; resolve it to a concrete count here so every mode
+    // behaves the same. The scale/merge engines expand 0 on their own, but the per-image processing
+    // loop (Rugnux) spawns exactly nthreads workers, so passing 0 there would spawn none and process
+    // nothing - hence resolving it centrally rather than relying on each consumer.
+    if (nthreads <= 0) {
+        unsigned int hw = std::thread::hardware_concurrency();
+        nthreads = hw > 0 ? static_cast<int>(hw) : 1;
+    }
+
     if (azint_only && scale_only) {
         logger.Error("--azint-only and --scale are mutually exclusive");
         exit(EXIT_FAILURE);
@@ -786,8 +843,14 @@ int main(int argc, char **argv) {
                 logger.Info("Reference space group: {} (number {})",
                             reference.space_group_name, reference.space_group_number.value_or(0));
 
+            // Check the reference against the cell that will actually drive the merge. --scale merges
+            // in the cell stored in the input file (as the former jfjoch_scale did); the -C override
+            // only takes effect on the full-analysis path, which otherwise determines its cell later by
+            // indexing (unknown here, so nothing can be checked yet).
+            const std::optional<UnitCell> data_cell =
+                scale_only ? dataset->experiment.GetUnitCell() : fixed_reference_unit_cell;
             const auto warning = ReferenceConsistencyWarning(
-                reference, fixed_reference_unit_cell,
+                reference, data_cell,
                 space_group_number.has_value() ? std::optional<int>(static_cast<int>(*space_group_number))
                                                : std::nullopt);
             if (!warning.empty())
@@ -806,17 +869,10 @@ int main(int argc, char **argv) {
         auto reflections = reader.ReadReflections(start_image, last_image);
 
         DiffractionExperiment experiment(dataset->experiment);
-        experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4);
-        experiment.FilePrefix(output_prefix);
-        experiment.Mode(DetectorMode::Standard);
-        experiment.PixelSigned(true);
-        experiment.OverwriteExistingFiles(true);
-        experiment.PolarizationFactor(0.99);
-        experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy);
+        configure_offline_output(experiment, output_prefix);
         // Keep the space group stored in the input file (written by the full pipeline) unless -S overrides.
         if (space_group_number.has_value())
             experiment.SpaceGroupNumber(space_group_number);
-        experiment.NumTriggers(1);
 
         // A rotation (goniometer) dataset uses RotationScaleMerge unless --force-still asks for stills scaling.
         IndexingSettings indexing_settings;
@@ -1000,16 +1056,9 @@ int main(int argc, char **argv) {
         return 0;
     }
 
-    experiment.BitDepthImage(32).Compression(CompressionAlgorithm::BSHUF_LZ4);
-    experiment.FilePrefix(output_prefix);
-    experiment.Mode(DetectorMode::Standard); // Ensure full image analysis
-    experiment.PixelSigned(true);
-    experiment.OverwriteExistingFiles(true);
-    experiment.PolarizationFactor(0.99);
-    experiment.SetFileWriterFormat(FileWriterFormat::NXmxLegacy);
+    configure_offline_output(experiment, output_prefix);
     experiment.SpaceGroupNumber(space_group_number);
     experiment.ImagesPerTrigger(images_to_process);
-    experiment.NumTriggers(1);
 
     // Re-determine the unit cell from scratch: discard any cell stored in the input file so
     // indexing is not biased by it. A stale or wrong stored cell otherwise resolves the indexing