feat(vegas): render every plugin's ticker content off the render thread

The plugin-facing canvas (DisplayManager.image, draw, matrix) was one
shared object, so any plugin whose Vegas content needed it -- display
capture, scroll-content generation, narrowed rendering -- was deferred to
the render thread and fetched there one at a time. On hdpi that is most
plugins, and each fetch stalled the scroll: news ~320ms, hockey ~660ms,
in bursts whenever the strip extended.

DisplayManager.offscreen() gives the calling thread a canvas of its own.
image, draw and matrix are now properties that resolve to the thread's
surface while it is inside the block and to the shared canvas otherwise,
so the ~100 existing uses become thread-correct unchanged. Inside,
update_display(), the hardware half of clear(), and set_scrolling_state()/
set_frame_hold() are inert, so a plugin drawn for Vegas can neither reach
the panel nor re-pace the live scroll. render_size() is rebuilt on it.
capture_mode() now restores the previous state instead of clearing it,
so it cannot end suppression inside an offscreen block.

The adapter draws every path on its own canvas (_isolated_canvas) and
drops the copy-and-restore of the shared image, which from a background
thread would have written a stale frame back over the render loop's.
Background fetches take the plugin's update/display lock, waiting up to
2s for a running update() and skipping the plugin that round otherwise;
Vegas never took that lock, so render-thread captures already raced
update(). A background fetch that comes back empty is no longer queued
for the render thread.

vegas_scroll.offscreen_prefetch (default true) restores the old deferred
path when false. 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)]