broker: rugnux live - rugnux runs inside the broker while a DECTRIS rotation collection arrives (demo prototype)

A DECTRIS rotation collection, with the mode switched on (PUT /config/rugnux), goes to
JFJochReceiverRugnux instead of JFJochReceiverLite: the images are forwarded as before, without the
per-image analysis, and one rugnux thread reads them back from the image buffer
(JFJochImageBufferReader) as they arrive. END waits for rugnux and carries its lattice, refined
geometry and per-image results. A small second run on the first 60 degrees reports early whether
there is a lattice. The live state, steps and result are at GET /rugnux/status and on a new
"Rugnux live" frontend page.

rugnux: ProcessConfig::fast_live takes every decision where it is first made and does nothing again
once the sweep is in (near-tie re-checks and the run made again, rotation-scale walk, metric-symmetry
arms, geometry walk, the supercell and quality guards, the doubled cell), so the run ends a fixed
time after the last image. A cut (/cancel, or deadline_s after the last image) hands over what is
established. Off by default: the CLI and the battery are unchanged (--fast-live is hidden, with
--simulate-live).

With the mode off the broker is unchanged. Not built yet.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

reader/JFJochImageBufferReader.cpp

@@ -0,0 +1,69 @@
+// SPDX-FileCopyrightText: 2026 Filip Leonarski, Paul Scherrer Institute <filip.leonarski@psi.ch>
+// SPDX-License-Identifier: GPL-3.0-only
+
+#include "JFJochImageBufferReader.h"
+
+#include <cstring>
+#include <thread>
+
+#include "JFJochStartMessageReader.h"
+#include "../common/JFJochException.h"
+#include "../frame_serialize/CBORStream2Deserializer.h"
+
+JFJochImageBufferReader::JFJochImageBufferReader(ImageBuffer &buffer) : buffer_(buffer) {}
+
+void JFJochImageBufferReader::Open(const StartMessage &start) {
+    number_of_images_ = start.number_of_images;
+    ended_ = false;
+    SetStartMessage(DatasetFromStartMessage(start, default_experiment));
+}
+
+void JFJochImageBufferReader::Ended() {
+    ended_ = true;
+}
+
+uint64_t JFJochImageBufferReader::GetNumberOfImages() const {
+    return number_of_images_;
+}
+
+void JFJochImageBufferReader::Close() {
+    SetStartMessage({});
+}
+
+bool JFJochImageBufferReader::WaitForMessage(int64_t image_number, std::vector<uint8_t> &cbor) {
+    if (image_number < 0 || static_cast<uint64_t>(image_number) >= number_of_images_)
+        throw JFJochException(JFJochExceptionCategory::ArrayOutOfBounds, "Image out of bounds");
+    // The buffer has no notification of its own; at 100 Hz a millisecond's poll costs nothing.
+    while (!buffer_.GetImage(cbor, image_number)) {
+        if (ended_)
+            return buffer_.GetImage(cbor, image_number);   // one that came in just before the end
+        std::this_thread::sleep_for(std::chrono::milliseconds(1));
+    }
+    return true;
+}
+
+bool JFJochImageBufferReader::ReadRawImage(int64_t image_number, JFJochReaderRawImage &ret) {
+    if (!WaitForMessage(image_number, ret.read_buffer))
+        return false;
+    const auto msg = CBORStream2Deserialize(ret.read_buffer);
+    if (msg->msg_type != CBORImageType::IMAGE || !msg->data_message)
+        return false;
+    // Copied out of the message: read_buffer is scratch a caller may hand to the next read.
+    const CompressedImage &image = msg->data_message->image;
+    ret.image_buffer.resize(image.GetCompressedSize());
+    memcpy(ret.image_buffer.data(), image.GetCompressed(), image.GetCompressedSize());
+    ret.image = CompressedImage(ret.image_buffer.data(), ret.image_buffer.size(), image.GetWidth(),
+                                image.GetHeight(), image.GetMode(), image.GetCompressionAlgorithm());
+    return true;
+}
+
+bool JFJochImageBufferReader::LoadImage_i(std::shared_ptr<JFJochReaderDataset> &dataset, DataMessage &message,
+                                          std::vector<uint8_t> &buffer, int64_t image_number, bool) {
+    if (!dataset || !WaitForMessage(image_number, buffer))
+        return false;
+    const auto msg = CBORStream2Deserialize(buffer);
+    if (msg->msg_type != CBORImageType::IMAGE || !msg->data_message)
+        return false;
+    message = *msg->data_message;
+    return true;
+}