fix(display): order snapshot writes, retry a failed one, stop the writer on cleanup; measure refresh from the low end

From review (CodeRabbit):

- A static frame's inline save now waits for a queued write in flight and
  drops a stale queued frame, under one write lock taken by both paths, so
  the last scrolling frame can no longer land on top of the first static one.
- A background write that fails clears the recorded digest, so an unchanged
  frame is written again instead of mtime-touching a stale file into looking
  healthy.
- cleanup() stops and joins the writer thread, dropping a frame it hasn't
  started, instead of leaving it (and its DisplayManager) alive.
- Vegas measures the panel's refresh from the 10th percentile of swap gaps,
  not the median: a late swap only lengthens a gap, so with most frames late
  in the window the median tracked the loop and locked in a rate too low
  (then the scroll ran fast until restart). Samples are dropped whenever
  the pacing is re-solved, so gaps from an old frame hold aren't divided by
  a new one.

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

test/test_display_dirty_tracking.py

@@ -399,6 +399,74 @@ class TestSnapshotOffRenderThread:
         assert threads[0] != threading.current_thread().name
         assert os.path.exists(dm._snapshot_path)
 
+    def test_a_failed_background_write_is_retried_not_touched(
+            self, dm, tmp_path, monkeypatch):
+        # Queuing records the frame as written. If the writer then fails, an
+        # unchanged frame must be written again, not mtime-touched: touching
+        # would make a stale preview look healthy.
+        import threading
+        import time
+        failed = threading.Event()
+
+        def failing(image):
+            failed.set()
+            raise OSError("disk full")
+
+        monkeypatch.setattr(dm, "_save_snapshot", failing)
+        self._due(dm, tmp_path, (0, 255, 0))
+        dm.set_scrolling_state(True)
+        try:
+            dm.update_display()
+            assert failed.wait(5)
+            deadline = time.time() + 5
+            while dm._last_snapshot_digest is not None and time.time() < deadline:
+                time.sleep(0.01)
+        finally:
+            dm.set_scrolling_state(False)
+        assert dm._last_snapshot_digest is None
+
+    def test_a_static_frame_lands_after_a_queued_one_still_being_written(
+            self, dm, tmp_path, monkeypatch):
+        # The last frame of a scroll can still be encoding when the first
+        # static frame is due; the older one must not land on top.
+        import threading
+        written, started, release = [], threading.Event(), threading.Event()
+        real = dm._save_snapshot
+
+        def slow_then_record(image):
+            if threading.current_thread().name == "snapshot-writer":
+                started.set()
+                release.wait(5)
+            written.append((threading.current_thread().name, image.getpixel((0, 0))))
+            real(image)
+
+        monkeypatch.setattr(dm, "_save_snapshot", slow_then_record)
+        self._due(dm, tmp_path, (0, 0, 255))
+        dm.set_scrolling_state(True)
+        dm.update_display()                       # queued: the writer blocks mid-write
+        assert started.wait(5)
+        dm.set_scrolling_state(False)
+        self._due(dm, tmp_path, (255, 0, 0))
+        static = threading.Thread(target=dm.update_display)
+        static.start()
+        static.join(0.2)
+        assert static.is_alive(), "the static save must wait for the write in flight"
+        release.set()
+        static.join(5)
+        assert [colour for _, colour in written] == [(0, 0, 255), (255, 0, 0)]
+
+    def test_cleanup_stops_the_writer(self, dm, tmp_path, monkeypatch):
+        threads, done = self._record_saves(dm, monkeypatch)
+        self._due(dm, tmp_path, (0, 255, 255))
+        dm.set_scrolling_state(True)
+        dm.update_display()
+        assert done.wait(5)
+        writer = dm._snapshot_thread
+        dm.set_scrolling_state(False)
+        dm._stop_snapshot_writer()
+        writer.join(2)
+        assert not writer.is_alive()
+
     def test_static_frames_are_still_written_inline(
             self, dm, tmp_path, monkeypatch):
         import threading

test/test_vegas_crisp_pacing.py

@@ -9,6 +9,7 @@ import sys
 from pathlib import Path
 from unittest.mock import patch
 
+import pytest
 from PIL import Image
 
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
@@ -122,6 +123,31 @@ def test_a_panel_that_keeps_up_with_its_cap_is_left_alone():
     assert p._crisp == crisp
 
 
+def test_a_window_of_mostly_late_frames_still_measures_the_panel():
+    # A late swap only lengthens its gap, by whole refreshes. A window where
+    # most frames missed a vsync (startup, a prefetch) must not read as a
+    # slower panel: with 7 frames in 10 a refresh late, the median would say
+    # 76Hz here and the speed would be solved for a panel that isn't there.
+    p = _pipeline(FakeDM(refresh_hz=120.0), scroll_speed=90)
+    period, hold = 1 / 95.0, p._frame_hold
+    gaps = [(hold + 1) * period if i % 10 < 7 else hold * period for i in range(400)]
+    clock = [1000.0]
+    with patch.object(rp_module.time, 'monotonic', lambda: clock[0]):
+        for gap in gaps:
+            p.render_frame()
+            clock[0] += gap
+            if p._measured_hz is not None:
+                break
+    assert p._measured_hz == pytest.approx(95.0)
+
+
+def test_re_solving_the_pacing_drops_samples_timed_under_the_old_hold():
+    p = _pipeline(FakeDM(refresh_hz=120.0), scroll_speed=90)
+    p._swap_times.extend([1.0, 1.01, 1.02])
+    p._configure_scroll_helper()
+    assert len(p._swap_times) == 0
+
+
 def test_no_measurement_without_hardware():
     # Nothing blocks in the emulator, so swap gaps say nothing about a panel.
     p = _pipeline(FakeDM(refresh_hz=120.0, hardware=False), scroll_speed=90)