feat(vegas): render plugin content off the render thread, and keep it off the GIL when the panel needs it (#630)

DisplayManager.offscreen() gives a thread its own canvas, so Vegas renders every plugin's ticker content on its prefetch thread instead of pausing the scroll for canvas-bound plugins on the render thread. A render gate (src/common/render_gate.py, vegas_scroll.prefetch_gate, on by default with the GIL-releasing binding) lets the prefetch thread run Python only while the render thread waits in SwapOnVSync: on hdpi, frames 2+ refreshes late fell eightfold and late frames overall from 0.90% to 0.60%. See docs/OFFSCREEN_RENDERING.md.

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

test/test_offscreen_rendering.py

@@ -0,0 +1,360 @@
+"""Per-thread off-screen rendering (DisplayManager.offscreen) and its Vegas use.
+
+The display canvas used to be one shared object, so any plugin that drew on it
+could only be rendered on the render thread, stalling the scroll for 40-600ms
+each. offscreen() gives the calling thread a canvas of its own. These tests pin
+the property that makes that safe: another thread's drawing never reaches what
+the render loop sees, presents or paces by.
+
+Runs against RGBMatrixEmulator, exercising the real DisplayManager.
+"""
+
+import os
+import sys
+import threading
+from types import SimpleNamespace
+
+os.environ["EMULATOR"] = "true"
+
+import pytest
+from PIL import Image
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.vegas_mode.config import VegasModeConfig  # noqa: E402
+from src.vegas_mode.plugin_adapter import PluginAdapter  # noqa: E402
+from src.vegas_mode.stream_manager import StreamManager  # noqa: E402
+
+WIDTH, HEIGHT = 128, 32
+YELLOW = (255, 255, 0)
+
+
+@pytest.fixture(scope="module")
+def dm(tmp_path_factory):
+    from src.display_manager import DisplayManager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+    manager = DisplayManager({
+        "display": {
+            "hardware": {"rows": 32, "cols": 64, "chain_length": 2,
+                         "parallel": 1, "brightness": 90},
+            "runtime": {"gpio_slowdown": 0},
+        },
+    }, suppress_test_pattern=True)
+    manager._snapshot_path = str(
+        tmp_path_factory.mktemp("offscreen") / "led_matrix_preview.png")
+    if manager.matrix is None:
+        pytest.fail("DisplayManager fell back to matrix=None; see the "
+                    "'Failed to initialize RGB Matrix' log line above.")
+    yield manager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+
+
+@pytest.fixture
+def fresh(dm):
+    """A known shared canvas, and scroll state reset around each test."""
+    dm.image = Image.new('RGB', (WIDTH, HEIGHT))
+    from PIL import ImageDraw
+    dm.draw = ImageDraw.Draw(dm.image)
+    dm.set_scrolling_state(False)
+    yield dm
+    dm.set_scrolling_state(False)
+
+
+class _Spy:
+    """Counts calls to one method of one object, and still calls it."""
+
+    def __init__(self, obj, name):
+        self.obj, self.name, self.count = obj, name, 0
+        self._orig = getattr(obj, name)
+
+    def __enter__(self):
+        def counting(*args, **kwargs):
+            self.count += 1
+            return self._orig(*args, **kwargs)
+        setattr(self.obj, self.name, counting)
+        return self
+
+    def __exit__(self, *exc):
+        setattr(self.obj, self.name, self._orig)
+
+
+def _in_thread(fn):
+    """Run fn on another thread and return its result, re-raising its error."""
+    box = {}
+
+    def run():
+        try:
+            box["value"] = fn()
+        except BaseException as exc:  # surfaced below
+            box["error"] = exc
+
+    thread = threading.Thread(target=run)
+    thread.start()
+    thread.join(10)
+    assert not thread.is_alive(), "worker hung"
+    if "error" in box:
+        raise box["error"]
+    return box.get("value")
+
+
+class TestIsolation:
+    def test_another_threads_drawing_is_invisible_here(self, fresh):
+        dm = fresh
+        shared = dm.image
+        inside, release = threading.Event(), threading.Event()
+        seen = {}
+
+        def worker():
+            with dm.offscreen(40) as surface:
+                dm.draw.rectangle([0, 0, 39, 31], fill=YELLOW)
+                seen.update(width=dm.width, matrix_width=dm.matrix.width,
+                            own_image=dm.image is surface.image)
+                inside.set()
+                release.wait(5)
+            seen["restored"] = dm.image is shared
+
+        thread = threading.Thread(target=worker)
+        thread.start()
+        try:
+            assert inside.wait(5)
+            # While the worker is mid-draw, this thread sees the real canvas.
+            assert dm.image is shared
+            assert (dm.width, dm.matrix.width) == (WIDTH, WIDTH)
+            assert shared.getpixel((0, 0)) == (0, 0, 0)
+        finally:
+            release.set()
+            thread.join(5)
+        assert seen == {"width": 40, "matrix_width": 40, "own_image": True,
+                        "restored": True}
+
+    def test_the_render_loop_keeps_presenting_while_another_thread_draws(self, fresh):
+        dm = fresh
+        inside, release = threading.Event(), threading.Event()
+
+        def worker():
+            with dm.offscreen():
+                dm.draw.rectangle([0, 0, 10, 10], fill=YELLOW)
+                dm.update_display()          # must not reach the panel
+                inside.set()
+                release.wait(5)
+
+        with _Spy(dm.matrix, "SwapOnVSync") as swaps:
+            thread = threading.Thread(target=worker)
+            thread.start()
+            try:
+                assert inside.wait(5)
+                dm.draw.point((5, 5), fill=(0, 0, 255))
+                dm.update_display()          # the render thread's own frame
+            finally:
+                release.set()
+                thread.join(5)
+        assert swaps.count == 1
+
+    def test_assignments_go_to_the_callers_canvas(self, fresh):
+        dm = fresh
+        shared = dm.image
+        replacement = Image.new('RGB', (WIDTH, HEIGHT), (1, 2, 3))
+
+        def worker():
+            with dm.offscreen():
+                dm.image = replacement       # e.g. clear() inside display()
+                return dm.image is replacement
+
+        assert _in_thread(worker) is True
+        assert dm.image is shared
+
+    def test_render_size_narrows_only_the_calling_thread(self, fresh):
+        dm = fresh
+        inside, release = threading.Event(), threading.Event()
+        seen = {}
+
+        def worker():
+            with dm.render_size(32):
+                seen["width"] = dm.width
+                inside.set()
+                release.wait(5)
+
+        thread = threading.Thread(target=worker)
+        thread.start()
+        try:
+            assert inside.wait(5)
+            assert dm.width == WIDTH
+        finally:
+            release.set()
+            thread.join(5)
+        assert seen["width"] == 32
+
+
+class TestNothingReachesThePanel:
+    def test_update_display_clear_and_pacing_are_inert_inside(self, fresh):
+        dm = fresh
+        real_matrix = dm.matrix
+        hold_before = dm._frame_hold
+        with _Spy(real_matrix, "SwapOnVSync") as swaps, \
+                _Spy(dm.offscreen_canvas, "Clear") as clears:
+            with dm.offscreen() as surface:
+                dm.draw.rectangle([0, 0, 5, 5], fill=YELLOW)
+                dm.update_display()
+                drawn = dm.image
+                dm.clear()
+                # clear() replaced the surface's image, not the shared one.
+                assert dm.image is surface.image and dm.image is not drawn
+                assert dm.image.getpixel((0, 0)) == (0, 0, 0)
+                dm.set_scrolling_state(True, 3)
+                dm.set_frame_hold(4)
+                dm.matrix.SwapOnVSync(object())       # inert through the proxy too
+        assert swaps.count == 0
+        assert clears.count == 0
+        assert dm._frame_hold == hold_before
+        assert dm._scrolling_state['is_scrolling'] is False
+
+    def test_capture_mode_inside_offscreen_does_not_end_suppression(self, fresh):
+        dm = fresh
+        real_matrix = dm.matrix
+        with _Spy(real_matrix, "SwapOnVSync") as swaps:
+            with dm.offscreen():
+                with dm.capture_mode():
+                    pass
+                dm.draw.point((1, 1), fill=YELLOW)
+                dm.update_display()
+        assert swaps.count == 0
+
+    def test_nesting_and_exceptions_restore_state(self, fresh):
+        dm = fresh
+        shared = dm.image
+        with pytest.raises(RuntimeError):
+            with dm.offscreen(100):
+                with dm.offscreen(50):
+                    assert dm.width == 50
+                    raise RuntimeError("plugin failed mid-draw")
+        assert dm.image is shared
+        assert dm.width == WIDTH
+        assert not dm._writes_suppressed()
+
+        with dm.offscreen(100) as outer:
+            with dm.offscreen(50):
+                pass
+            assert dm.image is outer.image
+            assert dm.width == 100
+
+
+# --- Vegas: the adapter draws plugins off the render thread ------------------
+
+class CapturePlugin:
+    """A plugin with neither get_vegas_content nor a scroll helper: captured."""
+
+    def __init__(self, display_manager):
+        self.display_manager = display_manager
+        self.calls = 0
+
+    def display(self, force_clear=False):
+        self.calls += 1
+        self.display_manager.draw.rectangle([0, 0, 30, 20], fill=YELLOW)
+        self.display_manager.update_display()
+
+
+class ScrollingPlugin:
+    """A ticker whose scroll image only exists once display() has run."""
+
+    def __init__(self, display_manager):
+        from src.common.scroll_helper import ScrollHelper
+        self.display_manager = display_manager
+        self.scroll_helper = ScrollHelper(WIDTH, HEIGHT)
+
+    def display(self, force_clear=False):
+        item = Image.new('RGB', (300, HEIGHT), YELLOW)
+        self.scroll_helper.create_scrolling_image([item], item_gap=0, element_gap=0)
+        self.display_manager.set_scrolling_state(True, 5)
+        self.display_manager.update_display()
+
+
+def _adapter(dm, **config):
+    return PluginAdapter(dm, VegasModeConfig(**config))
+
+
+def _has_yellow(images):
+    return any(YELLOW in {img.getpixel((x, y)) for x in range(min(img.width, 40))
+                          for y in range(img.height)} for img in images)
+
+
+class TestAdapterOffTheRenderThread:
+    def test_display_capture_runs_on_a_background_thread(self, fresh):
+        dm = fresh
+        shared = dm.image
+        before = shared.tobytes()
+        plugin = CapturePlugin(dm)
+        adapter = _adapter(dm)
+
+        with _Spy(dm.matrix, "SwapOnVSync") as swaps:
+            images = _in_thread(
+                lambda: adapter.get_content(plugin, "capture-bg", offscreen_only=True))
+
+        assert images and _has_yellow(images)
+        assert dm.image is shared and shared.tobytes() == before
+        assert swaps.count == 0
+
+    def test_scroll_content_is_generated_on_a_background_thread(self, fresh):
+        dm = fresh
+        plugin = ScrollingPlugin(dm)
+        adapter = _adapter(dm)
+        hold_before = dm._frame_hold
+
+        images = _in_thread(
+            lambda: adapter.get_content(plugin, "scroll-bg", offscreen_only=True))
+
+        assert images and _has_yellow(images)
+        # The plugin's own set_scrolling_state(True, 5) did not re-pace Vegas.
+        assert dm._frame_hold == hold_before
+        assert dm._scrolling_state['is_scrolling'] is False
+
+    def test_with_the_switch_off_background_capture_is_left_for_the_render_thread(self, fresh):
+        dm = fresh
+        plugin = CapturePlugin(dm)
+        adapter = _adapter(dm, offscreen_prefetch=False)
+
+        images = _in_thread(
+            lambda: adapter.get_content(plugin, "capture-legacy", offscreen_only=True))
+
+        assert images is None
+        assert plugin.calls == 0
+
+
+class TestPluginLock:
+    def test_a_background_fetch_waits_for_update_then_skips(self, fresh):
+        dm = fresh
+        lock = threading.Lock()
+        plugin = CapturePlugin(dm)
+        adapter = PluginAdapter(dm, VegasModeConfig(),
+                                plugin_manager=SimpleNamespace(
+                                    get_plugin_lock=lambda plugin_id: lock))
+        adapter.PLUGIN_LOCK_TIMEOUT = 0.05
+
+        with lock:   # update() in progress
+            skipped = _in_thread(
+                lambda: adapter.get_content(plugin, "locked", offscreen_only=True))
+        assert skipped is None
+        assert plugin.calls == 0
+
+        served = _in_thread(
+            lambda: adapter.get_content(plugin, "locked", offscreen_only=True))
+        assert served and plugin.calls == 1
+        assert not lock.locked()
+
+
+class TestStreamDoesNotDeferEmptyResults:
+    def _group(self, offscreen_prefetch):
+        config = VegasModeConfig(offscreen_prefetch=offscreen_prefetch)
+        adapter = SimpleNamespace(get_content=lambda *a, **k: None)
+        stream = StreamManager(config, SimpleNamespace(plugins={"p": object()}),
+                               adapter)
+        stream.refresh = lambda: None
+        stream._ordered_plugins = ["p"]
+        return stream.take_next_group(count=1, offscreen_only=True)
+
+    def test_nothing_to_show_is_not_sent_to_the_render_thread(self):
+        assert self._group(offscreen_prefetch=True) == [("p", [])]
+
+    def test_the_old_contract_still_defers_with_the_switch_off(self):
+        assert self._group(offscreen_prefetch=False) == [("p", None)]

test/test_render_gate.py

@@ -0,0 +1,432 @@
+"""The render gate (src/common/render_gate.py) and its wiring into Vegas."""
+
+import os
+import sys
+import threading
+import time
+from pathlib import Path
+
+os.environ.setdefault("EMULATOR", "true")
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.common import render_gate  # noqa: E402
+from src.common.render_gate import RenderGate  # noqa: E402
+
+PERIOD = 0.010
+
+
+class Clock:
+    def __init__(self):
+        self.now = 100.0
+
+    def __call__(self):
+        return self.now
+
+
+def _running(gate, clock, swaps=render_gate.MIN_SAMPLES + 1, hold=1):
+    """Drive ``swaps`` on-time swaps through the gate, a refresh apart."""
+    for _ in range(swaps):
+        gate.before_swap(hold)
+        clock.now += hold * PERIOD
+        gate.after_swap(hold)
+
+
+class TestWindow:
+    def test_no_window_until_the_period_is_known(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock, swaps=3)
+        assert gate.refresh_period() is None
+        gate.before_swap(1)
+        assert gate._open_until == 0.0
+        # ...and nothing parks meanwhile.
+        assert not gate._should_park(sys._getframe(), clock.now + 0.009)
+
+    def test_the_period_is_the_low_end_of_the_swap_gaps(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock, swaps=20)
+        gate.before_swap(1)
+        clock.now += 0.030            # one late frame lengthens its gap only
+        gate.after_swap(1)
+        assert gate.refresh_period() == pytest.approx(PERIOD)
+
+    def test_opens_until_just_before_the_refresh_the_swap_returns_on(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        last = clock.now
+        clock.now += 0.003            # the render thread's own work
+        gate.before_swap(1)
+        assert gate._open_until == pytest.approx(
+            last + PERIOD - render_gate.MARGIN_SECONDS)
+
+    def test_a_held_frame_opens_for_the_whole_hold(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock, hold=2)
+        last = clock.now
+        clock.now += 0.003
+        gate.before_swap(2)
+        assert gate._open_until == pytest.approx(
+            last + 2 * PERIOD - render_gate.MARGIN_SECONDS)
+
+    def test_a_late_frame_opens_until_the_next_boundary(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        last = clock.now
+        clock.now += 0.013            # missed its refresh: returns on the next
+        gate.before_swap(1)
+        assert gate._open_until == pytest.approx(
+            last + 2 * PERIOD - render_gate.MARGIN_SECONDS)
+
+    def test_closed_the_moment_the_swap_returns(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        gate.before_swap(1)
+        clock.now += PERIOD
+        gate.after_swap(1)
+        assert gate._open_until == 0.0
+
+
+class TestWhenToPark:
+    @pytest.fixture
+    def gate(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        gate.before_swap(1)
+        return gate
+
+    def test_runs_inside_the_window(self, gate):
+        assert not gate._should_park(sys._getframe(), gate._open_until - 0.001)
+
+    def test_parks_once_it_has_closed(self, gate):
+        assert gate._should_park(sys._getframe(), gate._open_until + 0.0001)
+
+    def test_never_with_no_render_loop_to_protect(self, gate):
+        stale = gate._last_return + render_gate.STALE_SECONDS + 0.001
+        assert not gate._should_park(sys._getframe(), stale)
+
+    def test_never_holding_a_guarded_lock(self, gate):
+        rlock, lock = threading.RLock(), threading.Lock()
+        gate.guard(rlock, lock, None)
+        closed = gate._open_until + 0.0001
+        with rlock:
+            assert not gate._should_park(sys._getframe(), closed)
+        with lock:
+            assert not gate._should_park(sys._getframe(), closed)
+        assert gate._should_park(sys._getframe(), closed)
+
+    def test_an_rlock_held_by_another_thread_is_not_this_ones(self, gate):
+        rlock = threading.RLock()
+        gate.guard(rlock)
+        taken, done = threading.Event(), threading.Event()
+
+        def hold():
+            with rlock:
+                taken.set()
+                done.wait(5)
+        other = threading.Thread(target=hold)
+        other.start()
+        try:
+            taken.wait(5)
+            assert gate._should_park(sys._getframe(), gate._open_until + 0.0001)
+        finally:
+            done.set()
+            other.join()
+
+    @staticmethod
+    def _frame_in(module):
+        namespace = {"__name__": module}
+        exec("import sys\ndef here():\n    return sys._getframe()\n", namespace)
+        return namespace["here"]()
+
+    @pytest.mark.parametrize("module", [
+        "logging", "logging.handlers", "threading", "importlib",
+        "_frozen_importlib", "_frozen_importlib_external",
+        "src.cache_manager", "src.cache.disk_cache",
+    ])
+    def test_never_inside_code_that_takes_shared_locks(self, gate, module):
+        frame = self._frame_in(module)
+        assert render_gate._unsafe(frame, None)
+        assert not gate._should_park(frame, gate._open_until + 0.0001)
+
+    @pytest.mark.parametrize("module", [
+        "src.cache_helpers", "plugin_logging_ticker", "src.vegas_mode.render_pipeline",
+    ])
+    def test_modules_that_only_sound_alike_are_fine(self, module):
+        assert not render_gate._unsafe(self._frame_in(module), None)
+
+    def test_where_python_is_installed_does_not_matter(self):
+        # GitHub's runners keep Python under /opt/hostedtoolcache; matching
+        # paths for "cache" made every stdlib frame unsafe there.
+        namespace = {"__name__": "json.decoder"}
+        exec(compile("import sys\ndef here():\n    return sys._getframe()\n",
+                     "/opt/hostedtoolcache/Python/3.11/lib/json/decoder.py", "exec"),
+             namespace)
+        assert not render_gate._unsafe(namespace["here"](), None)
+
+    def test_what_lies_below_the_yielding_block_does_not_count(self):
+        # A thread's stack always starts in threading.py; only frames above
+        # the one that entered yielding() matter.
+        namespace = {"__name__": "threading"}
+        exec("def bootstrap(fn):\n    return fn()\n", namespace)
+
+        def entered():
+            base = sys._getframe()
+
+            def work():
+                top = sys._getframe()
+                return render_gate._unsafe(top, None), render_gate._unsafe(top, base)
+            return work()
+        assert namespace["bootstrap"](entered) == (True, False)
+
+
+class TestYielding:
+    """A real background thread, parked and released by the gate."""
+
+    def _worker(self, gate, stop):
+        count = [0]
+
+        def step():
+            count[0] += 1
+
+        def work():
+            with gate.yielding():
+                while not stop.is_set():
+                    step()
+        thread = threading.Thread(target=work, daemon=True)
+        return thread, count
+
+    def test_parks_while_closed_and_runs_while_open(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        gate.before_swap(1)
+        stop = threading.Event()
+        thread, count = self._worker(gate, stop)
+        try:
+            clock.now = gate._open_until + 0.001      # window closed
+            thread.start()
+            time.sleep(0.2)
+            parked_steps = count[0]
+            # At most one step per MAX_WAIT timeout while parked.
+            assert parked_steps < 0.2 / render_gate.MAX_WAIT_SECONDS + 5
+            assert gate.parks >= 1
+
+            with gate._cond:                          # the next swap opens it
+                gate._open_until = clock.now + 3600.0
+                gate._generation += 1
+                gate._cond.notify_all()
+            time.sleep(0.1)
+            assert count[0] > parked_steps + 1000
+        finally:
+            stop.set()
+            with gate._cond:
+                gate._open_until = float("inf")
+                gate._generation += 1
+                gate._cond.notify_all()
+            thread.join(2)
+        assert not thread.is_alive()
+
+    def test_runs_freely_once_the_render_loop_stops(self):
+        clock = Clock()
+        gate = RenderGate(clock)
+        _running(gate, clock)
+        stop = threading.Event()
+        thread, count = self._worker(gate, stop)
+        clock.now += render_gate.STALE_SECONDS + 0.01
+        thread.start()
+        time.sleep(0.1)
+        stop.set()
+        thread.join(2)
+        assert count[0] > 1000
+        assert gate.parks == 0
+
+    def test_the_hook_comes_off_when_the_block_ends(self):
+        gate = RenderGate()
+        before = sys.getprofile()
+        with gate.yielding():
+            assert sys.getprofile() == gate._hook
+        assert sys.getprofile() is before
+
+
+class TestWhoIsGated:
+    def _in_thread(self, fn):
+        out = []
+        thread = threading.Thread(target=lambda: out.append(fn()))
+        thread.start()
+        thread.join(5)
+        return out[0]
+
+    def test_the_render_thread_never_gives_way(self):
+        gate = RenderGate()
+        gate.before_swap(1)
+        gate.after_swap(1)              # this thread swaps: it is the render thread
+        with gate.yielding():
+            assert sys.getprofile() is not gate._hook
+
+        def background():
+            with gate.yielding():
+                return sys.getprofile() == gate._hook
+        assert self._in_thread(background)
+
+    def test_a_live_refresh_from_another_thread_does_not_take_its_place(self):
+        gate = RenderGate()
+        gate.after_swap(1)              # the render loop
+        self._in_thread(lambda: gate.after_swap(1))    # a plugin pushing a frame
+        with gate.yielding():
+            assert sys.getprofile() is not gate._hook
+
+    def test_nested_blocks_keep_the_outer_boundary(self):
+        gate = RenderGate()
+
+        def nested():
+            with gate.yielding():
+                outer = gate._local.base
+                with gate.yielding():
+                    inner = gate._local.base
+                after = gate._local.base
+                still_hooked = sys.getprofile() == gate._hook
+            return outer is inner is after and still_hooked, sys.getprofile()
+        kept, final = self._in_thread(nested)
+        assert kept and final is None
+
+
+class TestDisplayManager:
+    @pytest.fixture
+    def dm(self):
+        from src.display_manager import DisplayManager
+        DisplayManager._instance = None
+        DisplayManager._initialized = False
+        manager = DisplayManager({"display": {
+            "hardware": {"rows": 32, "cols": 64, "chain_length": 1, "parallel": 1},
+            "runtime": {"gpio_slowdown": 0}}}, suppress_test_pattern=True)
+        yield manager
+        manager.render_gate = None
+        manager.set_scrolling_state(False)
+        DisplayManager._instance = None
+        DisplayManager._initialized = False
+
+    def test_opens_around_each_swap(self, dm):
+        calls = []
+
+        class Spy:
+            def before_swap(self, hold):
+                calls.append(("before", hold))
+
+            def after_swap(self, hold):
+                calls.append(("after", hold))
+
+        real_swap = dm.matrix.SwapOnVSync
+
+        def swap(canvas, *args, **kwargs):
+            calls.append(("swap",))
+            return real_swap(canvas, *args, **kwargs)
+        dm.matrix.SwapOnVSync = swap
+        dm.render_gate = Spy()
+        dm.set_scrolling_state(True, 2)
+        dm.draw.rectangle([0, 0, 3, 3], fill=(255, 0, 0))
+        dm.update_display()
+        assert calls == [("before", 2), ("swap",), ("after", 2)]
+
+    def test_off_screen_drawing_never_touches_it(self, dm):
+        class Boom:
+            def before_swap(self, hold):
+                raise AssertionError("an off-screen frame reached the gate")
+            after_swap = before_swap
+
+        dm.render_gate = Boom()
+        with dm.offscreen():
+            dm.draw.rectangle([0, 0, 3, 3], fill=(255, 0, 0))
+            dm.update_display()
+
+
+class TestVegasWiring:
+    def _coordinator(self, **config):
+        from src.vegas_mode.config import VegasModeConfig
+        from src.vegas_mode.coordinator import VegasModeCoordinator
+
+        class Holder:
+            def __init__(self):
+                self._buffer_lock = threading.RLock()
+                self._prefetch_lock = threading.Lock()
+                self._cache_lock = threading.Lock()
+
+        c = VegasModeCoordinator.__new__(VegasModeCoordinator)
+        c.vegas_config = VegasModeConfig(**config)
+        c._state_lock = threading.Lock()
+        c.stream_manager = Holder()
+        c.render_pipeline = Holder()
+        c.plugin_adapter = Holder()
+        c.display_manager = type("DM", (), {"render_gate": None})()
+        return c
+
+    def test_on_by_default(self, monkeypatch):
+        monkeypatch.setattr(render_gate, "swap_releases_gil", lambda: True)
+        c = self._coordinator()
+        c._install_render_gate()
+        assert isinstance(c.display_manager.render_gate, RenderGate)
+
+    def test_can_be_turned_off(self, monkeypatch):
+        monkeypatch.setattr(render_gate, "swap_releases_gil", lambda: True)
+        c = self._coordinator(prefetch_gate=False)
+        c._install_render_gate()
+        assert c.display_manager.render_gate is None
+
+    def test_installed_for_the_run_and_removed_after(self, monkeypatch):
+        monkeypatch.setattr(render_gate, "swap_releases_gil", lambda: True)
+        c = self._coordinator(prefetch_gate=True)
+        c._install_render_gate()
+        gate = c.display_manager.render_gate
+        assert isinstance(gate, RenderGate)
+        assert c._state_lock in gate._guarded
+        assert c.stream_manager._buffer_lock in gate._guarded
+        assert c.plugin_adapter._cache_lock in gate._guarded
+        c._remove_render_gate()
+        assert c.display_manager.render_gate is None
+
+    @pytest.mark.parametrize("releases", [False, None])
+    def test_ignored_without_a_binding_that_releases_the_gil(self, monkeypatch, releases):
+        monkeypatch.setattr(render_gate, "swap_releases_gil", lambda: releases)
+        c = self._coordinator(prefetch_gate=True)
+        c._install_render_gate()
+        assert c.display_manager.render_gate is None
+
+    def test_read_from_config(self):
+        from src.vegas_mode.config import VegasModeConfig
+        off = VegasModeConfig.from_config(
+            {"display": {"vegas_scroll": {"prefetch_gate": False}}})
+        assert off.prefetch_gate is False
+        assert off.to_dict()["prefetch_gate"] is False
+        assert VegasModeConfig.from_config(
+            {"display": {"vegas_scroll": {}}}).prefetch_gate is True
+
+    def test_the_prefetch_runs_inside_the_gate(self):
+        from src.vegas_mode.render_pipeline import RenderPipeline
+
+        gate = RenderGate()
+        seen = []
+
+        class Stream:
+            def take_next_group(self, offscreen_only=False):
+                seen.append((sys.getprofile() == gate._hook, offscreen_only))
+                return ["segment"]
+
+        pipeline = RenderPipeline.__new__(RenderPipeline)
+        pipeline.config = type("C", (), {"continuous_scroll": True})()
+        pipeline._prefetch_lock = threading.Lock()
+        pipeline._prefetch_thread = None
+        pipeline._prepared_group = None
+        pipeline.stream_manager = Stream()
+        pipeline.display_manager = type("DM", (), {"render_gate": gate})()
+        pipeline.start_prefetch()
+        pipeline._prefetch_thread.join(5)
+        assert seen == [(True, True)]
+        assert pipeline._prepared_group == ["segment"]

test/test_vegas_switch_interval.py

@@ -0,0 +1,41 @@
+"""vegas_scroll.switch_interval_ms: applied for a Vegas run, restored after."""
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from src.vegas_mode.config import VegasModeConfig  # noqa: E402
+from src.vegas_mode.coordinator import VegasModeCoordinator  # noqa: E402
+
+
+def _coordinator(ms):
+    c = VegasModeCoordinator.__new__(VegasModeCoordinator)
+    c.vegas_config = VegasModeConfig(switch_interval_ms=ms)
+    return c
+
+
+def test_applied_for_the_run_and_restored_after():
+    before = sys.getswitchinterval()
+    c = _coordinator(1.0)
+    try:
+        c._apply_switch_interval()
+        assert abs(sys.getswitchinterval() - 0.001) < 1e-9
+        c._apply_switch_interval()      # a second start must not lose the original
+    finally:
+        c._restore_switch_interval()
+    assert sys.getswitchinterval() == before
+
+
+def test_zero_leaves_the_interpreter_alone():
+    before = sys.getswitchinterval()
+    c = _coordinator(0.0)
+    c._apply_switch_interval()
+    c._restore_switch_interval()
+    assert sys.getswitchinterval() == before
+
+
+def test_read_from_config():
+    config = VegasModeConfig.from_config(
+        {"display": {"vegas_scroll": {"switch_interval_ms": 1}}})
+    assert config.switch_interval_ms == 1.0
+    assert VegasModeConfig.from_config({"display": {"vegas_scroll": {}}}).switch_interval_ms == 0.0