removed the b-scan collection_id floor

python_app/gui/app_window.py

@@ -285,8 +285,18 @@ class AppWindow(
         self._bscan_cpp_replay_window = bscan_cpp_replay_window
         self._bscan_history_by_combo = {}
         self._bscan_depth_axis_by_combo = {}
-        self._bscan_history_floor_collection_id = 0
         self._bscan_render_signature = None
+        # Writer into the processor's input ring, used to re-feed retained sweeps so the
+        # whole visible B-scan is recomputed. Opened lazily, only while acquisition is
+        # stopped (see AppWindowBscanReplayMixin).
+        self._replay_ring_writer = None
+        self._bscan_replay_active = False
+        # Results from the last completed replay. While non-empty they, not
+        # `_result_history`, are what the B-scan renders (see AppWindowBscanReplayMixin).
+        self._bscan_replay_results = []
+        self._bscan_reprocess_timer = QTimer(self)
+        self._bscan_reprocess_timer.setSingleShot(True)
+        self._bscan_reprocess_timer.timeout.connect(self._reprocess_history_through_processor)
         self._gpr_lookup_table = None
         self._gpr_image_item = None
         self._gpr_tx_item = None
@@ -745,6 +755,10 @@ class AppWindow(
             # 0) Stop the GPIO button watcher so a late press cannot start work.
             self._stop_control_button_watcher()
             self._resume_pipeline_after_capture = False
+            # 0) Drop the B-scan replay writer so a pending debounce cannot push into a
+            # ring we are about to tear down.
+            self._bscan_reprocess_timer.stop()
+            self._close_replay_ring_writer()
             # 1) Abort active capture first (releases exclusive hardware resources).
             self._abort_capture_sequence(resume_pipeline=False)
             # 2) Stop all managed processes/readers.

python_app/gui/controllers/app_window_config/live_processing_mixin.py

@@ -358,6 +358,10 @@ class AppWindowLiveProcessingMixin:
                 self._drain_results_until_quiet(timeout_s=0.25, poll_s=0.01)
                 self._sync_bscan_history_from_results()
                 self._draw_bscan_heatmap_from_history()
+                # The processor only refreshed its own newest sweeps; queue a full
+                # recompute of everything on screen. Debounced, so dragging a spin box
+                # sends one burst instead of one per step.
+                self._schedule_bscan_history_reprocess()
             elif self._is_gpr_processing_mode(current_mode):
                 latest = self._drain_results_until_quiet(timeout_s=0.8, poll_s=0.02)
                 collection = latest
@@ -507,7 +511,6 @@ class AppWindowLiveProcessingMixin:
 
     def _clear_history_mode_caches(self) -> None:
         """Drop cached render state for pass-through, B-scan, and GPR views."""
-        self._bscan_history_floor_collection_id = 0
         self._clear_bscan_plot_history()
         if hasattr(self, "_bscan_plot"):
             self._bscan_plot.clear()

python_app/gui/controllers/app_window_pipeline_mixin.py

@@ -150,6 +150,13 @@ class AppWindowPipelineMixin:
             # completed and the GUI hung.
             single_capture_start_ns = time.monotonic_ns() if single_capture else None
 
+            # `data_preprocessor` is about to become the owner of the preprocessed ring
+            # again, so the GUI must stop holding a writer into it. A queued replay would
+            # otherwise interleave with the real producer.
+            self._bscan_reprocess_timer.stop()
+            self._close_replay_ring_writer()
+            self._discard_bscan_replay_results()
+
             self._supervisor.start(config_path, allow_clean_orchestrator_exit=single_capture)
             self._close_readers()
             self._raw_reader = ShmRingReader(config.rings.raw_tap.name)
@@ -682,7 +689,7 @@ class AppWindowPipelineMixin:
     def _reset_runtime_history(self) -> None:
         """Reset runtime history and B-scan caches."""
         self._replace_runtime_history(retained_raw=[], retained_pre=[], retained_result=[])
-        self._bscan_history_floor_collection_id = 0
+        self._discard_bscan_replay_results()
         self._clear_history_mode_caches()
         self._update_history_indicator()
 

python_app/gui/controllers/app_window_plot/__init__.py

@@ -1,11 +1,13 @@
 """Plot-rendering mixins split by rendering mode."""
 
 from python_app.gui.controllers.app_window_plot.bscan_plot_mixin import AppWindowBscanPlotMixin
+from python_app.gui.controllers.app_window_plot.bscan_replay_mixin import AppWindowBscanReplayMixin
 from python_app.gui.controllers.app_window_plot.gpr_plot_mixin import AppWindowGprPlotMixin
 from python_app.gui.controllers.app_window_plot.trace_plot_mixin import AppWindowTracePlotMixin
 
 __all__ = [
     "AppWindowBscanPlotMixin",
+    "AppWindowBscanReplayMixin",
     "AppWindowGprPlotMixin",
     "AppWindowTracePlotMixin",
 ]

python_app/gui/controllers/app_window_plot/bscan_plot_mixin.py

@@ -16,17 +16,19 @@ def _result_tail(
     *,
     result_history: list[ResultCollection],
     history_limit: int,
-    floor_collection_id: int,
 ) -> list[ResultCollection]:
-    """Return filtered and de-duplicated result-history tail for B-scan usage."""
-    filtered = [
-        collection
-        for collection in result_history[-history_limit:]
-        if int(collection.collection_id) > int(floor_collection_id)
-    ]
+    """Return the de-duplicated newest `history_limit` entries for B-scan usage.
+
+    Selection is purely positional. An earlier version also dropped entries below a
+    `collection_id` floor, which cannot work here: ids are neither dense (the results
+    ring overwrites unread slots) nor monotonic across a run boundary (the C++ side
+    numbers from 1 again). Given the floor was derived from the first entry of this
+    very slice, the comparison provably removed nothing when ids ascend, and removed
+    exactly the newest frames when they do not.
+    """
     unique_reversed_tail: list[ResultCollection] = []
     seen_keys: set[tuple[int, int]] = set()
-    for collection in reversed(filtered):
+    for collection in reversed(result_history[-history_limit:]):
         key = (int(collection.collection_id), int(collection.monotonic_ns))
         if key in seen_keys:
             continue
@@ -43,13 +45,11 @@ def build_bscan_signature(
     subtract_mean_ascan_enabled: bool,
     result_history: list[ResultCollection],
     history_limit: int,
-    floor_collection_id: int,
 ) -> tuple[object, ...]:
     """Build deterministic signature used to detect B-scan cache invalidation."""
     result_tail = _result_tail(
         result_history=result_history,
         history_limit=history_limit,
-        floor_collection_id=floor_collection_id,
     )
     return (
         int(history_limit),
@@ -61,7 +61,6 @@ def build_bscan_signature(
         float(live_config.bscan_start_freq_mhz),
         float(live_config.bscan_stop_freq_mhz),
         bool(subtract_mean_ascan_enabled),
-        int(floor_collection_id),
         tuple((int(collection.collection_id), int(collection.monotonic_ns), len(collection.blocks)) for collection in result_tail),
     )
 
@@ -82,19 +81,27 @@ def apply_mean_ascan_subtraction(
 def rebuild_bscan_history_from_results(
     result_history: list[ResultCollection],
     history_limit: int,
-    floor_collection_id: int,
+    stats: dict[str, object] | None = None,
 ) -> tuple[dict[tuple[int, int], deque[np.ndarray]], dict[tuple[int, int], np.ndarray]]:
-    """Rebuild B-scan history and depth axes from processed result payloads."""
+    """Rebuild B-scan history and depth axes from processed result payloads.
+
+    Pass `stats` to receive a breakdown of why the rebuilt image may hold fewer
+    columns than `history_limit`. The three causes are independent and only
+    distinguishable here: too little history, frames carrying no `bscan` payload, and
+    depth-axis changes that reset the accumulated deque (see below).
+    """
     history_by_combo: dict[tuple[int, int], deque[np.ndarray]] = {}
     depth_axis_by_combo: dict[tuple[int, int], np.ndarray] = {}
 
     result_tail = _result_tail(
         result_history=result_history,
         history_limit=history_limit,
-        floor_collection_id=floor_collection_id,
     )
+    axis_resets = 0
+    without_bscan = 0
 
     for collection in result_tail:
+        carried_bscan = False
         for block in collection.blocks:
             key = (block.combo.input, block.combo.output)
             for payload in block.payloads:
@@ -110,6 +117,7 @@ def rebuild_bscan_history_from_results(
                 if depth_axis.size == 0 or amplitudes.size == 0:
                     continue
 
+                carried_bscan = True
                 history = history_by_combo.get(key)
                 stored_axis = depth_axis_by_combo.get(key)
                 if (
@@ -118,11 +126,25 @@ def rebuild_bscan_history_from_results(
                     or stored_axis.shape != depth_axis.shape
                     or not np.allclose(stored_axis, depth_axis, rtol=1e-4, atol=1e-6)
                 ):
+                    # A changed depth axis makes previously accumulated columns
+                    # un-stackable, so the deque restarts and everything gathered so far
+                    # for this combo is dropped. Frames computed with different
+                    # bscan_max_depth_m / frequency bounds land here — which is exactly
+                    # what a partially replayed history looks like.
+                    if history is not None:
+                        axis_resets += 1
                     history = deque(maxlen=history_limit)
                     history_by_combo[key] = history
                     depth_axis_by_combo[key] = depth_axis.copy()
 
                 history.append(amplitudes.copy())
+        if not carried_bscan:
+            without_bscan += 1
+
+    if stats is not None:
+        stats["tail"] = len(result_tail)
+        stats["without_bscan"] = without_bscan
+        stats["axis_resets"] = axis_resets
 
     return history_by_combo, depth_axis_by_combo
 
@@ -214,7 +236,7 @@ class AppWindowBscanPlotMixin:
         depth_max = float(np.max(depth_axis))
         depth_span = max(depth_max - depth_min, 1e-6)
         sweep_count = sweeps.shape[0]
-        self._warn_if_bscan_exceeds_replay_window(sweep_count)
+        self._warn_if_bscan_window_underfilled()
         sweep_width = float(max(sweep_count, 1))
         x_min = 0.5
         x_max = x_min + sweep_width
@@ -238,33 +260,30 @@ class AppWindowBscanPlotMixin:
         )
         return True
 
-    def _warn_if_bscan_exceeds_replay_window(self, sweep_count: int) -> None:
-        """Warn once when the image reaches past what the C++ processor can replay.
+    def _warn_if_bscan_window_underfilled(self) -> None:
+        """Warn once when there is less retained raw material than the operator asked for.
 
-        Beyond that window a frame keeps the payload it was first computed with, so
-        editing Gain / Cut / Max depth / Start-Stop MHz silently leaves the older
-        columns on their previous settings — the image mixes two parameter sets.
+        Settings edits are recomputed across the whole image by re-feeding
+        `_pre_history` to the processor (see :class:`AppWindowBscanReplayMixin`), so a
+        wide window is no longer a coherency problem. What it can still be is an empty
+        promise: asking for more sweeps than were ever captured simply shows fewer.
         """
-        replay_window = int(self._bscan_cpp_replay_window)
-        if sweep_count <= replay_window:
+        window = self._bscan_display_window_scans()
+        retained = len(self._pre_history)
+        if retained >= window:
             return
 
-        stale_count = sweep_count - replay_window
         self._log_warning(
-            f"B-scan shows {sweep_count} sweeps but the processor replays only the newest "
-            f"{replay_window}; the older {stale_count} keep the settings they were captured with.",
+            f"B-scan is set to show {window} sweeps but only {retained} are retained; "
+            "the image shows what history there is.",
             details=(
-                "Changing Gain / Cut m / Max depth m / Start MHz / Stop MHz re-processes "
-                f"only the newest {replay_window} sweeps.\n"
-                f"Reduce 'Scans to show (stopped)' to {replay_window} for an image that is "
-                "coherent across every column."
-            ),
-            # Keyed on the operator-controlled window, not the live column count, so
-            # that repeated "Remove Last" in stopped mode does not re-warn every click.
-            once_key=(
-                f"bscan_window_exceeds_replay_{replay_window}_"
-                f"{self._bscan_display_window_scans()}"
+                "The GUI keeps a bounded history of preprocessed sweeps, so a window "
+                "wider than the run itself cannot be filled.\n"
+                f"Capture more sweeps, or set 'Scans to show (stopped)' to {retained} or less."
             ),
+            # Keyed on the operator-controlled window rather than the live retained count,
+            # so that repeated "Remove Last" does not re-warn on every click.
+            once_key=f"bscan_window_underfilled_{window}",
         )
 
     def _bscan_display_window_scans(self) -> int:
@@ -283,7 +302,6 @@ class AppWindowBscanPlotMixin:
 
     def _sync_bscan_history_from_results(self) -> None:
         """Rebuild B-scan history cache when live params or inputs changed."""
-        self._advance_bscan_floor_to_display_window()
         signature = self._bscan_signature()
         if signature == self._bscan_render_signature:
             return
@@ -293,25 +311,84 @@ class AppWindowBscanPlotMixin:
     def _bscan_signature(self) -> tuple[object, ...]:
         """Build state signature for B-scan history cache invalidation."""
         live_config = self._live_processing_config()
-        result_history = list(self._result_history)
+        # Must read the same source the rebuild will, or the cache decides nothing
+        # changed while the image would in fact be built from different collections.
+        result_history, _from_replay = self._bscan_source_collections()
         return build_bscan_signature(
             live_config=live_config,
             subtract_mean_ascan_enabled=bool(self._bscan_subtract_mean_ascan.isChecked()),
             result_history=result_history,
             history_limit=self._bscan_display_window_scans(),
-            floor_collection_id=self._bscan_history_floor_collection_id,
         )
 
+    def _bscan_source_collections(self) -> tuple[list[ResultCollection], bool]:
+        """Return the collections the image is built from, and whether they are replayed.
+
+        A completed replay is the better source: it is exactly `window` long and every
+        entry went through the processor with the same settings. The runtime history is
+        not usable right after one, because results whose ids it never held are appended
+        out of order, leaving the deque unsorted for the rest of the session.
+        """
+        replayed = getattr(self, "_bscan_replay_results", None)
+        if replayed:
+            return list(replayed), True
+        return list(self._result_history), False
+
     def _rebuild_bscan_history_from_results(self) -> None:
         """Recompute B-scan history cache from results history buffer."""
-        result_history = list(self._result_history)
+        result_history, from_replay = self._bscan_source_collections()
+        window = self._bscan_display_window_scans()
+        stats: dict[str, object] = {}
         history_by_combo, depth_axis_by_combo = rebuild_bscan_history_from_results(
             result_history=result_history,
-            history_limit=self._bscan_display_window_scans(),
-            floor_collection_id=self._bscan_history_floor_collection_id,
+            history_limit=window,
+            stats=stats,
         )
         self._bscan_history_by_combo = history_by_combo
         self._bscan_depth_axis_by_combo = depth_axis_by_combo
+        self._log_bscan_window_shortfall(
+            window=window,
+            result_history_len=len(result_history),
+            from_replay=from_replay,
+            history_by_combo=history_by_combo,
+            stats=stats,
+        )
+
+    def _log_bscan_window_shortfall(
+        self,
+        *,
+        window: int,
+        result_history_len: int,
+        from_replay: bool,
+        history_by_combo: dict[tuple[int, int], deque[np.ndarray]],
+        stats: dict[str, object],
+    ) -> None:
+        """Explain at DEBUG why the image holds fewer columns than were requested.
+
+        Three independent causes produce the same symptom, so each is reported as its
+        own number rather than a single verdict:
+          * `tail` — the run simply produced fewer sweeps than were asked for;
+          * `without_bscan` — frames processed in another mode carry no bscan payload;
+          * `axis_resets` — a changed depth axis restarted the deque, dropping every
+            column gathered before it (the usual cause after a partial replay);
+          * per-combo counts — the image renders one combo at a time.
+        """
+        rendered = max((len(history) for history in history_by_combo.values()), default=0)
+        if rendered >= window:
+            return
+
+        per_combo = ", ".join(
+            f"in{input_pos}/out{output_pos}={len(history)}"
+            for (input_pos, output_pos), history in sorted(history_by_combo.items())
+        )
+        self._log_debug(
+            f"B-scan window not filled: rendered={rendered} of requested={window}. "
+            f"source={'replay' if from_replay else 'result history'} len={result_history_len}, "
+            f"newest-{window} tail={stats.get('tail')}, "
+            f"of those without a bscan payload={stats.get('without_bscan')}, "
+            f"depth-axis resets={stats.get('axis_resets')}. "
+            f"Per combo: {per_combo or 'none'}."
+        )
 
     def _pick_bscan_display_key(self) -> tuple[int, int] | None:
         """Choose combo history key to render."""
@@ -396,36 +473,6 @@ class AppWindowBscanPlotMixin:
         self._bscan_depth_axis_by_combo.clear()
         self._bscan_render_signature = None
 
-    def _advance_bscan_floor_to_display_window(self) -> None:
-        """Clamp B-scan source history to the active display window.
-
-        The window counts RETAINED ENTRIES, so the floor is read off the n-th
-        newest entry rather than computed as `latest_id - window`. Collection ids
-        are not dense: the results ring overwrites unread slots when the producer
-        outruns the GUI poll loop, so the GUI keeps ids like 1..50, 81..130, ...
-        Subtracting the window from the newest id would then span far fewer than
-        `window` entries — asking for 150 sweeps yielded 87.
-
-        Recomputed unconditionally rather than ratcheted upwards: widening the
-        window in stopped mode must be able to LOWER the floor and bring older
-        frames back into view. A stale floor cannot survive this way either, so
-        the previous special case for collection ids restarting on a new C++ run
-        is no longer needed.
-        """
-        history = self._result_history
-        if not history:
-            return
-
-        window = self._bscan_display_window_scans()
-        if len(history) <= window:
-            self._bscan_history_floor_collection_id = 0
-            return
-
-        # `_result_tail` keeps entries with `collection_id > floor`, so sit the
-        # floor one below the oldest entry that still fits in the window.
-        oldest_visible = history[len(history) - window]
-        self._bscan_history_floor_collection_id = max(0, int(oldest_visible.collection_id) - 1)
-
     def _ensure_phase_view_box(self) -> pg.ViewBox:
         """Create or return secondary right-axis ViewBox for phase curves."""
         plot_item = self._bscan_plot.getPlotItem()

python_app/gui/controllers/app_window_plot/bscan_replay_mixin.py

@@ -0,0 +1,253 @@
+"""Re-feed retained sweeps to the processor so it recomputes the whole B-scan.
+
+The processor keeps only ~50 preprocessed sweeps of its own (`kBscanReplayWindow` in
+`data_processor.cpp`), so changing a live B-scan setting used to refresh just the newest
+50 columns while everything older kept the parameters it was captured with — one image
+stitched from two parameter sets.
+
+The GUI already holds up to 1000 preprocessed sweeps in `_pre_history`, byte-identical
+to what the processor consumes: `data_preprocessor` pushes the very same serialized
+buffer into both the working ring and the tap the GUI reads. So instead of making the
+processor hoard sweeps, we hand its own raw material back to it and let the ordinary
+`pop -> process -> publish` path recompute every column with the current settings.
+"""
+
+from __future__ import annotations
+
+import time
+
+from python_app.gui.runtime.history import record_result_history
+from python_app.orchestration.shm import ShmRingWriter
+from python_app.storage.npz.serialize import PREPROC_MAGIC, serialize_trace_collection
+
+# The results ring overwrites unread slots, so a burst must never exceed what the GUI
+# drains between chunks. Comfortably below the (typically 50-slot) ring capacity.
+_REPLAY_CHUNK_SIZE = 12
+
+# Per-chunk budget. Generous: the processor may be busy, and overshooting only costs
+# a redraw that reflects fewer columns.
+_REPLAY_CHUNK_TIMEOUT_S = 2.0
+
+# Restarting the timer on every edit collapses a spin-box drag into one replay.
+_BSCAN_REPROCESS_DEBOUNCE_MS = 300
+
+
+class AppWindowBscanReplayMixin:
+    """Recomputes the full visible B-scan by re-feeding sweeps to the processor."""
+
+    def _reprocess_history_through_processor(self) -> bool:
+        """Re-feed the visible tail of `_pre_history` so every column is recomputed.
+
+        Returns True when a replay actually ran. Refuses (returning False) whenever the
+        preconditions for safely writing into the processor's input ring do not hold.
+        """
+        # `_pump_events_during_drain` runs the event loop, so a settings edit made while
+        # a long replay is in flight can re-arm the debounce and fire this method inside
+        # itself — two nested bursts sharing one ring and one result count.
+        if getattr(self, "_bscan_replay_active", False):
+            return False
+        if not self._can_reprocess_history():
+            return False
+
+        window = self._bscan_display_window_scans()
+        tail = list(self._pre_history)[-window:]
+        # `_pre_history` and `_result_history` are fed by two different rings, each
+        # dropping independently when the producer outruns the GUI. A low overlap means
+        # the replayed results arrive under ids the result history never held, so they
+        # are appended rather than replacing the columns already on screen.
+        result_keys = {
+            (int(c.collection_id), int(c.monotonic_ns)) for c in self._result_history
+        }
+        overlap = sum(
+            1 for c in tail if (int(c.collection_id), int(c.monotonic_ns)) in result_keys
+        )
+        self._log_debug(
+            f"B-scan replay starting: window={window}, preprocessed history="
+            f"{len(self._pre_history)}, result history={len(self._result_history)}, "
+            f"to re-send={len(tail)}, of those already in result history={overlap}."
+        )
+        if not tail:
+            return False
+
+        writer = self._ensure_replay_ring_writer()
+        if writer is None:
+            return False
+
+        # The processor replays its own retained sweeps on every live-config revision
+        # bump. We are about to send the same collections (and more), so suppress it
+        # rather than let it publish the newest 50 twice.
+        self._write_live_processing_config(reprocess_current_result=False)
+
+        # Our own event pumping would otherwise let the poll timer fire and consume the
+        # replayed results through `_read_all_results`, which records them to disk and
+        # feeds the pipeline metrics. Restart in `finally`: losing the ring poll on an
+        # exception would leave the GUI permanently blind.
+        self._timer.stop()
+        self._bscan_replay_active = True
+        sent = 0
+        received = 0
+        replayed: list = []
+        try:
+            for start in range(0, len(tail), _REPLAY_CHUNK_SIZE):
+                chunk = tail[start : start + _REPLAY_CHUNK_SIZE]
+                pushed = 0
+                for collection in chunk:
+                    payload = serialize_trace_collection(collection, PREPROC_MAGIC)
+                    if not writer.push(payload):
+                        # Only fails when the payload exceeds the slot size, which is a
+                        # config problem rather than a transient one: stop the burst.
+                        self._log_warning(
+                            "B-scan replay stopped: a preprocessed sweep does not fit the ring slot.",
+                            details=(
+                                f"payload={len(payload)} bytes, "
+                                f"slot={writer.slot_size_bytes} bytes"
+                            ),
+                            once_key="bscan_replay_payload_too_large",
+                        )
+                        break
+                    pushed += 1
+                sent += pushed
+                got = self._collect_replayed_results(
+                    expected=pushed, timeout_s=_REPLAY_CHUNK_TIMEOUT_S, into=replayed
+                )
+                received += got
+                if got < pushed:
+                    # A short chunk means the processor did not answer in time; keep
+                    # going, but say which one so a systematic stall is visible.
+                    self._log_debug(
+                        f"B-scan replay chunk at offset {start}: pushed={pushed}, recovered={got}."
+                    )
+                if pushed != len(chunk):
+                    break
+        finally:
+            self._bscan_replay_active = False
+            self._timer.start()
+
+        # Render straight from what came back rather than from `_result_history`.
+        #
+        # The two histories are fed by different rings that drop independently, so the
+        # sweeps we re-sent only partly overlap the results already on record. The
+        # non-overlapping ones get appended to the deque even though their ids are old,
+        # so its newest `window` entries are a mix of freshly and stale-processed
+        # frames. What came back is by construction the right count and uniformly
+        # processed, so use it directly.
+        self._bscan_replay_results = replayed
+
+        # Replayed collections keep their original ids and timestamps, so the render
+        # signature is unchanged even though the payload values are not. Drop it or the
+        # cache would decide nothing needs rebuilding.
+        self._bscan_render_signature = None
+        self._sync_bscan_history_from_results()
+        self._draw_bscan_heatmap_from_history()
+
+        if received < sent:
+            self._log_warning(
+                f"B-scan replay recovered {received} of {sent} re-sent sweeps; "
+                "some columns may still show their captured settings.",
+                once_key=f"bscan_replay_incomplete_{sent}_{received}",
+            )
+        self._log_debug(f"B-scan replay finished: sent={sent}, recovered={received}.")
+        return True
+
+    def _can_reprocess_history(self) -> bool:
+        """Return whether re-feeding the processor's input ring is safe right now."""
+        if self._processing_mode.currentText() != "bscan":
+            return False
+        # `data_preprocessor` owns the preprocessed ring while acquisition runs; writing
+        # into it concurrently would corrupt the sequence counters.
+        if self._supervisor.is_running():
+            return False
+        if not self._supervisor.is_processor_running():
+            return False
+        return self._result_reader is not None
+
+    def _collect_replayed_results(
+        self, *, expected: int, timeout_s: float, into: list | None = None
+    ) -> int:
+        """Pop `expected` replayed results, recording them into runtime history.
+
+        Deliberately bypasses `_read_all_results`: that path also feeds the pipeline
+        metrics and the disk recorder, and a replay is neither new acquisition nor
+        something that should be written to disk a second time.
+
+        `into` also receives them in arrival order, which is what the B-scan renders —
+        see `_reprocess_history_through_processor` for why the runtime history alone is
+        not a usable source afterwards.
+        """
+        if expected <= 0 or self._result_reader is None:
+            return 0
+
+        deadline = time.monotonic() + timeout_s
+        received = 0
+        while received < expected and time.monotonic() < deadline:
+            collection = self._result_reader.pop_result_collection()
+            if collection is None:
+                self._pump_events_during_drain(0.005)
+                continue
+            record_result_history(self._result_history, collection)
+            if into is not None:
+                into.append(collection)
+            received += 1
+        return received
+
+    def _ensure_replay_ring_writer(self) -> ShmRingWriter | None:
+        """Return a writer attached to the processor's input ring, creating it lazily.
+
+        Geometry comes from `_active_run_config` — the config the C++ side actually
+        started with — never from the editable `_defaults_config`: `ShmRingWriter` owns
+        the rings it opens and *recreates* a segment whose geometry disagrees, which
+        would destroy the ring under a live processor.
+        """
+        existing = getattr(self, "_replay_ring_writer", None)
+        if existing is not None:
+            return existing
+
+        config = getattr(self, "_active_run_config", None)
+        if config is None:
+            # Processor outlived the GUI that started it: its ring geometry is unknown,
+            # and guessing risks recreating the segment underneath it.
+            self._log_warning(
+                "Cannot recompute the full B-scan: this GUI session did not start the "
+                "pipeline, so the processor's ring geometry is unknown.",
+                once_key="bscan_replay_no_active_run_config",
+            )
+            return None
+
+        ring = config.rings.preprocessed
+        try:
+            self._replay_ring_writer = ShmRingWriter(
+                ring.name, int(ring.capacity), int(ring.slot_size_bytes)
+            )
+        except Exception as exc:  # noqa: BLE001
+            self._log_exception("Failed to open the B-scan replay ring writer", exc, level="WARN")
+            self._replay_ring_writer = None
+        return self._replay_ring_writer
+
+    def _discard_bscan_replay_results(self) -> None:
+        """Fall back to the runtime history as the render source.
+
+        Called whenever the replayed set stops describing what should be on screen:
+        acquisition resuming (live frames must win) or the history being edited.
+        """
+        if not getattr(self, "_bscan_replay_results", None):
+            return
+        self._bscan_replay_results = []
+        self._bscan_render_signature = None
+
+    def _close_replay_ring_writer(self) -> None:
+        """Detach from the processor's input ring (safe to call repeatedly)."""
+        writer = getattr(self, "_replay_ring_writer", None)
+        if writer is None:
+            return
+        try:
+            writer.close()
+        except Exception as exc:  # noqa: BLE001
+            self._log_exception("Failed to close the B-scan replay ring writer", exc, level="WARN")
+        finally:
+            self._replay_ring_writer = None
+
+    def _schedule_bscan_history_reprocess(self) -> None:
+        """Debounce a full recompute so dragging a spin box does not send hundreds of sweeps."""
+        if not self._can_reprocess_history():
+            return
+        self._bscan_reprocess_timer.start(_BSCAN_REPROCESS_DEBOUNCE_MS)

python_app/gui/controllers/app_window_plot_mixin.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 from python_app.gui.controllers.app_window_plot import (
     AppWindowBscanPlotMixin,
+    AppWindowBscanReplayMixin,
     AppWindowGprPlotMixin,
     AppWindowTracePlotMixin,
 )
@@ -13,6 +14,7 @@ from python_app.models.dataset_model import ResultCollection
 class AppWindowPlotMixin(
     AppWindowTracePlotMixin,
     AppWindowBscanPlotMixin,
+    AppWindowBscanReplayMixin,
     AppWindowGprPlotMixin,
 ):
     """Routes plotting to trace, B-scan, or GPR-specific mixins."""

python_app/gui/controllers/app_window_snapshot_mixin.py

@@ -241,7 +241,7 @@ class AppWindowSnapshotMixin:
                 retained_pre=retained_pre,
                 retained_result=retained_result,
             )
-            self._bscan_history_floor_collection_id = 0
+            self._discard_bscan_replay_results()
             self._clear_history_mode_caches()
 
             self._write_live_processing_config(history_command=history_command, bump_history_seq=True)

python_app/tests/test_bscan_history_window.py

@@ -2,12 +2,13 @@
 
 The B-scan used to be pinned to the C++ replay window (~50 sweeps) by two separate
 mechanisms: the render-side history limit and a monotonically rising
-`_bscan_history_floor_collection_id`. Widening only the first would have changed
-nothing, because the floor kept filtering older collections out for good.
+`collection_id` floor. Widening only the first would have changed nothing, because
+the floor kept filtering older collections out for good.
 
-These tests pin the two properties that make the stopped-mode review work: the
-window follows acquisition state, and the floor is recomputed (not ratcheted) so a
-widened window can bring already-discarded frames back into view.
+The floor is gone: selection is positional, which is the only criterion that holds
+when ids are sparse (the results ring drops) or restart from 1 (a new C++ run).
+These tests pin the observable consequences — how many columns end up on screen —
+rather than any internal counter.
 """
 
 from __future__ import annotations
@@ -22,6 +23,7 @@ os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
 from PyQt6.QtWidgets import QApplication  # noqa: E402
 
 from python_app.gui.app_window import AppWindow  # noqa: E402
+from python_app.gui.runtime.history import record_result_history  # noqa: E402
 from python_app.models.dataset_model import (  # noqa: E402
     ComboKey,
     ResultBlock,
@@ -60,12 +62,12 @@ class BscanDisplayWindowTest(unittest.TestCase):
         self.w = _window
         self._original_is_running = self.w._supervisor.is_running
         self.w._result_history.clear()
-        self.w._bscan_history_floor_collection_id = 0
+        self.w._bscan_render_signature = None
 
     def tearDown(self) -> None:
         self.w._supervisor.is_running = self._original_is_running
         self.w._result_history.clear()
-        self.w._bscan_history_floor_collection_id = 0
+        self.w._bscan_render_signature = None
 
     def _set_running(self, running: bool) -> None:
         self.w._supervisor.is_running = lambda: running
@@ -86,44 +88,73 @@ class BscanDisplayWindowTest(unittest.TestCase):
         self.w._bscan_history_window.setValue(300)
         self.assertEqual(self.w._bscan_display_window_scans(), 300)
 
-    def test_widening_the_window_lowers_the_floor(self) -> None:
-        # The regression this whole change exists for: a run leaves the floor high,
-        # and widening the window afterwards must pull it back down.
+    def test_widening_the_window_brings_older_frames_back(self) -> None:
+        # The regression this whole change exists for: a live run renders 50 columns,
+        # and widening the window after Stop must reach back over the retained history
+        # rather than stay pinned to whatever the live path last drew.
         self._fill_history(300)
         self._set_running(True)
-        self.w._advance_bscan_floor_to_display_window()
-        raised_floor = self.w._bscan_history_floor_collection_id
-        self.assertEqual(raised_floor, 300 - self.w._bscan_cpp_replay_window)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(
+            len(self.w._bscan_history_by_combo[(0, 0)]), self.w._bscan_cpp_replay_window
+        )
 
         self._set_running(False)
         self.w._bscan_history_window.setValue(300)
-        self.w._advance_bscan_floor_to_display_window()
-        self.assertEqual(self.w._bscan_history_floor_collection_id, 0)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 300)
 
-    def test_floor_survives_collection_ids_restarting(self) -> None:
-        # A new C++ run restarts ids from 1. The unconditional recompute must not
-        # leave a stale high floor that hides the whole fresh run.
+    def test_restarted_collection_ids_do_not_hide_the_fresh_run(self) -> None:
+        # Regression: Start after a stopped review made the image melt to 0 columns and
+        # then snap back to 50. A new C++ run numbers from 1, so entries that are newest
+        # by position carry the smallest ids; any id-based cut-off derived from the old
+        # run rejected exactly them.
         self._fill_history(300)
-        self._set_running(True)
-        self.w._advance_bscan_floor_to_display_window()
-        self.assertGreater(self.w._bscan_history_floor_collection_id, 0)
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self.w._sync_bscan_history_from_results()
 
-        self.w._result_history.clear()
-        self._fill_history(5)
-        self.w._advance_bscan_floor_to_display_window()
-        self.assertEqual(self.w._bscan_history_floor_collection_id, 0)
+        self._set_running(True)
+        window = self.w._bscan_cpp_replay_window
+        for fresh in range(1, window + 1):
+            self.w._result_history.append(_bscan_result(fresh))
+            self.w._sync_bscan_history_from_results()
+            self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), window)
 
     def test_window_counts_entries_not_collection_ids(self) -> None:
         # The results ring overwrites unread slots when the producer outruns the GUI
-        # poll loop, so retained collection ids are sparse. A floor of
-        # `latest_id - window` then spans far fewer than `window` entries: this is
-        # exactly the case where asking for 150 sweeps rendered only 87.
+        # poll loop, so retained collection ids are sparse. Counting in ids rather than
+        # in entries is exactly the case where asking for 150 sweeps rendered only 87.
         self._fill_history(300, id_step=3)
         self._set_running(False)
         self.w._bscan_history_window.setValue(150)
         self.w._sync_bscan_history_from_results()
         self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 150)
 
+    def test_reprocessed_results_replace_rather_than_duplicate_columns(self) -> None:
+        # The processor's recompute is triggered by feeding sweeps back through it, and
+        # it republishes them under their ORIGINAL ids. Two independent mechanisms keep
+        # that from doubling every column, and neither may be confused with the removed
+        # `collection_id` floor: a threshold cannot tell a duplicate from its original,
+        # since they share the id. Deduplication is by key equality.
+        self._set_running(False)
+        self.w._bscan_history_window.setValue(300)
+        self._fill_history(200)
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200)
+
+        # Intake: a recomputed collection replaces the entry holding the same key.
+        for collection in list(self.w._result_history):
+            record_result_history(self.w._result_history, _bscan_result(collection.collection_id))
+        self.assertEqual(len(self.w._result_history), 200)
+
+        # Render: a duplicate that reached the deque by another path is still collapsed,
+        # keeping the newer of the two.
+        self.w._result_history.append(_bscan_result(200))
+        self.w._bscan_render_signature = None
+        self.w._sync_bscan_history_from_results()
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), 200)
+
     def test_rebuild_renders_the_full_widened_window(self) -> None:
         self._fill_history(300)
         self._set_running(False)

python_app/tests/test_bscan_replay.py

@@ -0,0 +1,282 @@
+"""Re-feeding retained sweeps to the processor to recompute the whole B-scan.
+
+The processor keeps only ~50 preprocessed sweeps of its own, so a live settings edit
+used to refresh just the newest 50 columns. The GUI holds up to 1000 of them and hands
+them back through the processor's input ring, which only works if the bytes the GUI
+writes are exactly the ones the C++ side expects — that wire-format contract is what
+the first test pins, without needing the pipeline running.
+
+The guard tests pin the other half: the GUI must never write into that ring while
+`data_preprocessor` owns it.
+"""
+
+from __future__ import annotations
+
+from contextlib import suppress
+import os
+from pathlib import Path
+import unittest
+
+import numpy as np
+
+os.environ.setdefault("QT_QPA_PLATFORM", "offscreen")
+from PyQt6.QtCore import QSignalBlocker  # noqa: E402
+from PyQt6.QtWidgets import QApplication  # noqa: E402
+
+from python_app.gui.app_window import AppWindow  # noqa: E402
+from python_app.models.dataset_model import (  # noqa: E402
+    ComboKey,
+    ResultBlock,
+    ResultCollection,
+    ResultPayload,
+    SweepCollection,
+    TraceData,
+)
+from python_app.orchestration.shm import ShmRingWriter  # noqa: E402
+from python_app.orchestration.shm.ring_reader import ShmRingReader  # noqa: E402
+from python_app.storage.npz.serialize import PREPROC_MAGIC, serialize_trace_collection  # noqa: E402
+
+# Mirrors kPreprocessedCollectionMagic in common_cpp/ipc/src/shared_types.cpp. If this
+# ever drifts, the processor silently rejects everything the GUI re-feeds.
+_CPP_PREPROCESSED_MAGIC = 0x32525050
+
+_app: QApplication | None = None
+_window: AppWindow | None = None
+
+
+def setUpModule() -> None:
+    global _app, _window
+    _app = QApplication.instance() or QApplication([])
+    _window = AppWindow(Path("."))
+
+
+def tearDownModule() -> None:
+    if _window is not None:
+        _window.close()
+
+
+def _sweep(collection_id: int, *, combos: int = 3, points: int = 64) -> SweepCollection:
+    traces = [
+        TraceData(
+            combo=ComboKey(input=index, output=index + 1),
+            frequency_hz=np.linspace(1e9, 8e9, points, dtype=np.float32),
+            s11=(np.arange(points) + index).astype(np.complex64) + 0.5j,
+            s21=(np.arange(points) * 2 + index).astype(np.complex64) - 0.25j,
+        )
+        for index in range(combos)
+    ]
+    return SweepCollection(
+        collection_id=collection_id,
+        monotonic_ns=collection_id * 1_000_000 + 7,
+        traces=traces,
+        capture_start_ns=collection_id * 10,
+        capture_end_ns=collection_id * 10 + 5,
+    )
+
+
+def _bscan_result(collection_id: int, *, points: int = 32) -> ResultCollection:
+    payload = ResultPayload(
+        processing_name="bscan",
+        kind=1,
+        frequency_hz=np.linspace(0.0, 1.0, points, dtype=np.float32),
+        trace=(np.arange(points) + collection_id).astype(np.complex64),
+    )
+    return ResultCollection(
+        collection_id=collection_id,
+        monotonic_ns=collection_id,
+        blocks=[ResultBlock(combo=ComboKey(input=0, output=0), payloads=[payload])],
+    )
+
+
+class PreprocessedWireFormatTest(unittest.TestCase):
+    """The bytes the GUI re-feeds must be the ones the C++ processor decodes."""
+
+    def setUp(self) -> None:
+        self.ring_name = f"/radar_test_{self._testMethodName}"
+        with suppress(OSError):
+            (Path("/dev/shm") / self.ring_name[1:]).unlink()
+        self.writer = ShmRingWriter(self.ring_name, 8, 1 << 20)
+        self.reader = ShmRingReader(self.ring_name)
+        self.addCleanup(self._cleanup)
+
+    def _cleanup(self) -> None:
+        with suppress(Exception):
+            self.reader.close()
+        with suppress(Exception):
+            self.writer.close()
+        with suppress(OSError):
+            (Path("/dev/shm") / self.ring_name[1:]).unlink()
+
+    def _assert_same(self, original: SweepCollection, decoded: SweepCollection) -> None:
+        self.assertEqual(decoded.collection_id, original.collection_id)
+        self.assertEqual(decoded.monotonic_ns, original.monotonic_ns)
+        self.assertEqual(decoded.capture_start_ns, original.capture_start_ns)
+        self.assertEqual(decoded.capture_end_ns, original.capture_end_ns)
+        self.assertEqual(len(decoded.traces), len(original.traces))
+        for left, right in zip(original.traces, decoded.traces, strict=True):
+            self.assertEqual((left.combo.input, left.combo.output), (right.combo.input, right.combo.output))
+            np.testing.assert_array_equal(left.frequency_hz, right.frequency_hz)
+            np.testing.assert_array_equal(left.s11, right.s11)
+            np.testing.assert_array_equal(left.s21, right.s21)
+
+    def test_magic_matches_the_cpp_decoder(self) -> None:
+        payload = serialize_trace_collection(_sweep(1), PREPROC_MAGIC)
+        self.assertEqual(int.from_bytes(payload[:4], "little"), _CPP_PREPROCESSED_MAGIC)
+
+    def test_round_trip_preserves_every_field(self) -> None:
+        original = _sweep(42)
+        self.assertTrue(self.writer.push(serialize_trace_collection(original, PREPROC_MAGIC)))
+
+        decoded = self.reader.pop_preprocessed_collection()
+        self.assertIsNotNone(decoded)
+        assert decoded is not None
+        self._assert_same(original, decoded)
+
+    def test_batch_keeps_order(self) -> None:
+        originals = [_sweep(cid) for cid in range(100, 105)]
+        for collection in originals:
+            self.assertTrue(self.writer.push(serialize_trace_collection(collection, PREPROC_MAGIC)))
+
+        decoded = []
+        while (collection := self.reader.pop_preprocessed_collection()) is not None:
+            decoded.append(collection)
+
+        self.assertEqual([c.collection_id for c in decoded], [c.collection_id for c in originals])
+        for left, right in zip(originals, decoded, strict=True):
+            self._assert_same(left, right)
+
+    def test_overflow_drops_oldest(self) -> None:
+        # Why the replay pushes in chunks instead of one burst: an undrained ring
+        # silently overwrites, which would punch holes into the very image we are
+        # trying to make coherent.
+        capacity = 8
+        for cid in range(200, 200 + capacity + 4):
+            self.writer.push(serialize_trace_collection(_sweep(cid), PREPROC_MAGIC))
+
+        survived = []
+        while (collection := self.reader.pop_preprocessed_collection()) is not None:
+            survived.append(collection.collection_id)
+
+        self.assertEqual(len(survived), capacity)
+        self.assertEqual(survived[-1], 200 + capacity + 3)
+
+
+class _IdleResultReader:
+    """Stands in for a connected results reader that simply has nothing to hand out."""
+
+    def pop_result_collection(self):
+        return None
+
+    def close(self) -> None:
+        return None
+
+
+class ReplayGuardTest(unittest.TestCase):
+    """The GUI must refuse to write into a ring `data_preprocessor` still owns."""
+
+    def setUp(self) -> None:
+        self.w = _window
+        # The 50 ms ring poll and the debounce would both run against this half-faked
+        # window while the test drives it by hand; park them for the duration.
+        self.w._timer.stop()
+        self.w._bscan_reprocess_timer.stop()
+        self.addCleanup(self.w._timer.start)
+        self.addCleanup(self.w._bscan_reprocess_timer.stop)
+
+        self._original_is_running = self.w._supervisor.is_running
+        self._original_is_processor_running = self.w._supervisor.is_processor_running
+        self._original_mode = self.w._processing_mode.currentText()
+        # Changing the mode fires the live-settings handler, which would kick off the
+        # very replay these tests are inspecting.
+        with QSignalBlocker(self.w._processing_mode):
+            self.w._processing_mode.setCurrentText("bscan")
+        self.w._supervisor.is_running = lambda: False
+        self.w._supervisor.is_processor_running = lambda: True
+        self.w._result_reader = _IdleResultReader()
+
+    def tearDown(self) -> None:
+        self.w._supervisor.is_running = self._original_is_running
+        self.w._supervisor.is_processor_running = self._original_is_processor_running
+        with QSignalBlocker(self.w._processing_mode):
+            self.w._processing_mode.setCurrentText(self._original_mode)
+        self.w._result_reader = None
+
+    def test_refuses_while_acquisition_runs(self) -> None:
+        self.w._supervisor.is_running = lambda: True
+        self.assertFalse(self.w._can_reprocess_history())
+
+    def test_refuses_when_processor_is_down(self) -> None:
+        self.w._supervisor.is_processor_running = lambda: False
+        self.assertFalse(self.w._can_reprocess_history())
+
+    def test_refuses_outside_bscan_mode(self) -> None:
+        self.w._processing_mode.setCurrentText("pass_through")
+        self.assertFalse(self.w._can_reprocess_history())
+
+    def test_refuses_without_a_results_reader(self) -> None:
+        self.w._result_reader = None
+        self.assertFalse(self.w._can_reprocess_history())
+
+    def test_allows_when_stopped_with_a_live_processor(self) -> None:
+        self.assertTrue(self.w._can_reprocess_history())
+
+    def test_replayed_results_are_rendered_instead_of_runtime_history(self) -> None:
+        """Regression: 300 re-sent, 300 recovered, only 145 drawn.
+
+        `_pre_history` and `_result_history` come from two rings that drop
+        independently, so the re-sent sweeps only partly overlap the recorded results.
+        The non-overlapping ones get appended to the deque under old ids, leaving its
+        newest `window` entries a mix of freshly and stale-processed frames. Rendering
+        from the replayed set instead sidesteps the whole problem.
+        """
+        window = 30
+        replayed = [_bscan_result(cid) for cid in range(9000, 9000 + window)]
+
+        self.w._result_history.clear()
+        # Stand-in for a runtime history whose ids barely overlap the replayed ones.
+        self.w._result_history.extend(_bscan_result(cid) for cid in range(1, 200))
+        self.addCleanup(self.w._result_history.clear)
+
+        self.w._bscan_replay_results = replayed
+        self.addCleanup(self.w._discard_bscan_replay_results)
+        self.w._bscan_history_window.setValue(window)
+        self.w._bscan_render_signature = None
+        self.w._sync_bscan_history_from_results()
+
+        self.assertEqual(len(self.w._bscan_history_by_combo[(0, 0)]), window)
+
+    def test_discarding_replay_falls_back_to_runtime_history(self) -> None:
+        self.w._bscan_replay_results = [_bscan_result(1)]
+        self.w._discard_bscan_replay_results()
+        self.assertEqual(self.w._bscan_replay_results, [])
+        self.assertIsNone(self.w._bscan_render_signature)
+
+    def test_replay_refuses_to_re_enter_itself(self) -> None:
+        # Pumping the event loop mid-replay can fire the debounce again; a nested burst
+        # would share the ring and corrupt the result accounting.
+        self.w._pre_history.clear()
+        self.w._pre_history.extend(_sweep(cid) for cid in range(1, 6))
+        self.addCleanup(self.w._pre_history.clear)
+
+        self.w._bscan_replay_active = True
+        self.addCleanup(setattr, self.w, "_bscan_replay_active", False)
+
+        self.assertFalse(self.w._reprocess_history_through_processor())
+
+    def test_replay_refuses_without_an_active_run_config(self) -> None:
+        # A GUI restarted against a still-running processor does not know the ring
+        # geometry, and ShmRingWriter would recreate the segment underneath it.
+        self.w._pre_history.clear()
+        self.w._pre_history.extend(_sweep(cid) for cid in range(1, 6))
+        self.addCleanup(self.w._pre_history.clear)
+
+        previous = getattr(self.w, "_active_run_config", None)
+        self.w._active_run_config = None
+        self.addCleanup(setattr, self.w, "_active_run_config", previous)
+
+        self.assertIsNone(self.w._ensure_replay_ring_writer())
+        self.assertFalse(self.w._reprocess_history_through_processor())
+
+
+if __name__ == "__main__":
+    unittest.main()

python_app/tests/test_processing.py

@@ -219,7 +219,7 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(1, (0, 0), [1.0, 2.0], [10.0, 20.0]),
             self._bscan_collection(2, (0, 0), [1.0, 2.0], [11.0, 21.0]),
         ]
-        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(len(by_combo[(0, 0)]), 2)
         self.assertTrue(np.array_equal(axes[(0, 0)], np.array([1.0, 2.0], dtype=np.float32)))
 
@@ -229,7 +229,7 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(2, (0, 0), [1.0, 2.0], [1.0, 2.0], kind=2),         # wrong kind
             self._bscan_collection(3, (0, 0), [1.0, 2.0, 3.0], [1.0, 2.0]),            # size mismatch
         ]
-        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(by_combo, {})
 
     def test_depth_axis_change_resets_history(self) -> None:
@@ -237,17 +237,21 @@ class RebuildBscanHistoryTest(unittest.TestCase):
             self._bscan_collection(1, (0, 0), [1.0, 2.0], [10.0, 20.0]),
             self._bscan_collection(2, (0, 0), [1.0, 2.0, 3.0], [11.0, 21.0, 31.0]),  # new depth axis
         ]
-        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=0)
+        by_combo, axes = rebuild_bscan_history_from_results(history, history_limit=10)
         self.assertEqual(len(by_combo[(0, 0)]), 1)  # reset on axis change; only the latest sweep remains
         self.assertEqual(axes[(0, 0)].shape, (3,))
 
-    def test_floor_collection_id_excludes_older(self) -> None:
+    def test_selection_is_positional_not_by_collection_id(self) -> None:
+        # Ids are neither dense nor monotonic across a run boundary, so the tail is
+        # taken by position only. Here the newest two entries carry the SMALLEST ids;
+        # an id-based cut-off would have dropped exactly them.
         history = [
-            self._bscan_collection(1, (0, 0), [1.0], [10.0]),
-            self._bscan_collection(2, (0, 0), [1.0], [20.0]),
+            self._bscan_collection(cid, (0, 0), [1.0], [float(cid)])
+            for cid in (98, 99, 100, 1, 2)
         ]
-        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=10, floor_collection_id=1)
-        self.assertEqual(len(by_combo[(0, 0)]), 1)  # only collection_id > 1
+        by_combo, _ = rebuild_bscan_history_from_results(history, history_limit=3)
+        self.assertEqual(len(by_combo[(0, 0)]), 3)
+        self.assertEqual([sweep[0] for sweep in by_combo[(0, 0)]], [100.0, 1.0, 2.0])
 
 
 # --------------------------------------------------------------------------- #