indexing: select predicted reflections by partiality, build indexers where it pays

When more reflections are predicted for a frame than the output can hold, the
surplus was dropped by keeping those closest to the Ewald sphere. On the rotation
path that quantity is identically zero by construction - the rocking coordinate is
chosen so the scattering vector lands exactly on the sphere - so the comparison
fell through to h, k and l and the survivors were whichever came first in
lexicographic order. Measured on a large cell: every value within one float ulp of
zero, and the kept set had a MEAN PARTIALITY BELOW that of the full set, i.e. worse
than choosing at random. Rank by partiality instead, which the predictor already
computes and which is what the header always claimed was being kept. On the one
regression crystal large enough to cross the cap this lifts completeness from 84.8%
to 90.2% on the same observations; multiplicity and R_meas move the way they must
when the same measurements cover more of reciprocal space.

The online path asked for a cap of ten thousand but the truncation was hardcoded to
the offline limit, so the broker predicted and integrated up to six times what it
could transport and discarded the rest after paying for it. Honour the caller's
limit, which also makes the post-integration re-truncation dead code.

Indexer pool construction becomes a policy. The online service needs every indexer
resident before data arrives, because a cuFFT plan built on the first frame is
planning time inside the measurement; spending memory to be ready is the intended
trade there and stays the default. Offline there is no such deadline, and a stills
run with a known cell was holding a fully allocated FFT indexer per worker that the
algorithm resolution can never dispatch - 2.8 GB where 0.4 GB is needed. rugnux and
the viewer opt into building on first use; the broker, the receiver and the tests
are untouched. This also removes a dangling reference that was latent: the worker
held the settings by reference although the pool is routinely constructed from a
temporary, which only survived because eager construction finished inside the
constructor call.

Finally, refuse a first-pass lattice that indexes fewer than a sixth of the
validation frames. It fires on nothing in the regression set - the weakest real
crystal sits at 22 of 60, more than twice the floor - so it is a backstop, but the
failure it prevents is one the set does contain: a dataset with no crystal at all
adopts a lattice from its powder rings, integrates every image against it, and dies
much later inside the merge complaining about resolution. It now stops in the first
pass and says what to try.

Regression set: 36 of 37 crystals byte-identical, the exception being the
completeness gain above; 34 of 37 space groups, no failures. Full unit suite passes.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

image_analysis/IndexAndRefine.cpp

@@ -496,6 +496,12 @@ void IndexAndRefine::QuickPredictAndIntegrate(DataMessage &msg,
     };
     ApplyPredictionRange(settings_prediction, experiment, latt);
 
+    // Online is bounded by what the image-buffer slot can carry, offline by the prediction limit; both
+    // come from BraggPrediction so the cap, the prediction and the transport headroom cannot drift.
+    // The predictor applies it, so what is over the cap is never integrated.
+    prediction.output_limit = real_time ? BraggPrediction::kOnlineMaxReflections
+                                        : BraggPrediction::kPredictionOutput;
+
     // Predict, then integrate with the selected integrator (box-sum or profile-fit).
     auto pred_start_time = std::chrono::steady_clock::now();
     auto nrefl = prediction.Calc(outcome.experiment, latt, settings_prediction);
@@ -511,25 +517,6 @@ void IndexAndRefine::QuickPredictAndIntegrate(DataMessage &msg,
     auto integration_end_time = std::chrono::steady_clock::now();
     msg.integration_time_s = std::chrono::duration<float>(integration_end_time - integration_start_time).count();
 
-    // Online is bounded by what the image-buffer slot can carry; offline by the prediction limit. Both
-    // come from BraggPrediction so the cap, the prediction and the transport headroom cannot drift.
-    const size_t kMaxReflections = real_time ? BraggPrediction::kOnlineMaxReflections
-                                            : BraggPrediction::kPredictionOutput;
-    if (i_outcome.reflections.size() > kMaxReflections) {
-        // Keep only smallest d (highest resolution)
-        std::nth_element(i_outcome.reflections.begin(),
-                         i_outcome.reflections.begin() + static_cast<long>(kMaxReflections),
-                         i_outcome.reflections.end(),
-                         [](const Reflection& a, const Reflection& b) {
-                             return a.d < b.d;
-                         });
-        i_outcome.reflections.resize(kMaxReflections);
-
-        // Optional: make output ordered by d (nice for downstream / debugging)
-        std::sort(i_outcome.reflections.begin(), i_outcome.reflections.end(),
-                  [](const Reflection& a, const Reflection& b) { return a.d < b.d; });
-    }
-
     CalcISigma(msg, i_outcome.reflections);
     CalcWilsonBFactor(msg, i_outcome.reflections);