chore: satisfy Codacy's subprocess checks on the new test runner

Codacy runs Bandit and Opengrep (its Semgrep fork). The new
subprocess.run in scripts/run_plugin_tests.py trips three patterns, on
two different lines:

  Bandit   B404 on the import, B603 on the call
  Opengrep dangerous-subprocess-use-audit          on the run( line
           dangerous-subprocess-use-tainted-env-args on the argv line

A nosemgrep applies only to its own line, so the call line and the argv
line each need one; a single comment on the call covered neither rule
fully. Suppression is the right answer here rather than a rewrite: the
interpreter is sys.executable, the arguments are a list, and no shell is
involved, so there is nothing to word-split or expand.

Matches the pair the rest of the repo already uses for this shape --
permission_utils.py, plugin_loader.py, install_dependencies_apt.py.

Codacy: 0 new issues, up to standards.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9

plugin-repos/starlark-apps/manager.py

@@ -714,9 +714,11 @@ class StarlarkAppsPlugin(BasePlugin):
                     self.logger.error(f"Failed to render app: {self.current_app.app_id}")
                     return False
 
-            # Display current frame
-            self._display_frame()
-            return True
+            # Display current frame. The result is propagated: a failed frame
+            # update is not a displayed frame, and returning True regardless
+            # told the controller the mode had rendered, so it held the dead
+            # frame for the whole display_duration instead of rotating on.
+            return self._display_frame()
 
         except Exception as e:
             self.logger.error(f"Error displaying Starlark app: {e}")
@@ -843,10 +845,13 @@ class StarlarkAppsPlugin(BasePlugin):
             self.logger.error(f"Error loading frames for {app.app_id}: {e}")
             return False
 
-    def _display_frame(self) -> None:
-        """Display the current frame of the current app."""
+    def _display_frame(self) -> bool:
+        """Display the current frame of the current app.
+
+        :returns: whether a frame actually reached the display manager.
+        """
         if not self.current_app or not self.current_app.frames:
-            return
+            return False
 
         try:
             current_time = time.time()
@@ -864,8 +869,11 @@ class StarlarkAppsPlugin(BasePlugin):
                 )
                 self.current_app.last_frame_time = current_time
 
+            return True
+
         except Exception as e:
             self.logger.error(f"Error displaying frame: {e}")
+            return False
 
     def install_app(self, app_id: str, star_file_path: str, metadata: Optional[Dict[str, Any]] = None, assets_dir: Optional[str] = None) -> bool:
         """

scripts/run_plugin_tests.py

@@ -8,7 +8,7 @@ Supports both unittest and pytest.
 
 import os
 import re
-import subprocess
+import subprocess  # nosec B404 - list-form argv only, no shell  # nosemgrep
 import sys
 import argparse
 from pathlib import Path
@@ -97,16 +97,30 @@ def run_script_tests(test_files: list, verbose: bool = False) -> int:
     a script that needs a tty or an LED matrix is not a regression.
     """
     env = dict(os.environ)
-    env.setdefault("PYTHONPATH", str(PROJECT_ROOT))
-    env.setdefault("LEDMATRIX_CORE", str(PROJECT_ROOT))
+    # Prepend rather than setdefault. An inherited PYTHONPATH -- a developer's
+    # shell, a tox run, another checkout -- otherwise wins outright, and the
+    # subprocess imports a different copy of the core than the one under test.
+    # That is exactly the failure ledmatrix-plugins#467 describes, and it is
+    # invisible: the tests pass or fail against a tree nobody meant to test.
+    inherited = env.get("PYTHONPATH")
+    env["PYTHONPATH"] = (f"{PROJECT_ROOT}{os.pathsep}{inherited}"
+                         if inherited else str(PROJECT_ROOT))
+    env["LEDMATRIX_CORE"] = str(PROJECT_ROOT)
 
     passed = skipped = failed = 0
     failures = []
     for path in test_files:
         try:
-            proc = subprocess.run([sys.executable, str(path)], cwd=str(Path(path).parent),
-                                  capture_output=True, text=True, env=env,
-                                  stdin=subprocess.DEVNULL, timeout=300)
+            # Fixed interpreter (sys.executable) plus a test path this script
+            # discovered by globbing the repo; argument list, no shell, so
+            # nothing is word-split or expanded. Same suppression pair the
+            # rest of the repo uses for this shape (see permission_utils.py).
+            proc = subprocess.run(  # nosec B603 - no shell invoked (list-form argv)  # nosemgrep
+                [sys.executable, str(path)],  # nosemgrep
+                cwd=str(Path(path).parent),
+                capture_output=True, text=True, env=env,
+                stdin=subprocess.DEVNULL, timeout=300,
+            )
             rc = proc.returncode
             tail = " | ".join((proc.stdout or proc.stderr or "").strip().splitlines()[-2:])[:200]
         except subprocess.TimeoutExpired:

src/display_controller.py

@@ -198,6 +198,10 @@ class DisplayController:
         # the main run loop reconciles (loads/unloads) on its own thread so
         # mutating available_modes never races with rendering.
         self._pending_plugin_reconcile = False
+        # Monotonic stamp of the last mailbox disk read; see
+        # _poll_on_demand_requests. None means "never polled", so the first
+        # call always goes through.
+        self._last_on_demand_poll: Optional[float] = None
         self.on_demand_active = False
         self.on_demand_mode: Optional[str] = None
         self.on_demand_modes: List[str] = []  # All modes for the on-demand plugin
@@ -1267,8 +1271,22 @@ class DisplayController:
         self.on_demand_schedule_override = False
         self._publish_on_demand_state()
 
+    #: Shortest gap between mailbox disk reads. This is called after every
+    #: frame -- about 125 times a second on a scrolling mode -- and the read
+    #: below is deliberately uncached, so without a floor it was 125 disk reads
+    #: per second to find nothing. An on-demand request comes from a person
+    #: clicking in the web UI, so a quarter second of latency is not
+    #: perceptible, and it cuts the read rate by 30x.
+    ON_DEMAND_POLL_INTERVAL = 0.25
+
     def _poll_on_demand_requests(self) -> None:
         """Poll cache for new on-demand requests from external controllers."""
+        now = time.monotonic()
+        if (self._last_on_demand_poll is not None
+                and now - self._last_on_demand_poll < self.ON_DEMAND_POLL_INTERVAL):
+            return
+        self._last_on_demand_poll = now
+
         try:
             # Use a long max_age (1 hour) to ensure requests aren't expired before processing
             # The request_id check prevents duplicate processing.
@@ -1332,7 +1350,26 @@ class DisplayController:
         # panel silently showed the earlier plugin. processed_id still guards
         # against double-processing if this delete fails.
         try:
-            self.cache_manager.delete('display_on_demand_request')
+            # Compare before deleting. The web process can post a newer request
+            # between the read above and this delete; an unconditional delete
+            # threw that one away and it was never processed -- the user's
+            # second click did nothing. Re-reading uncached and only deleting
+            # our own request_id means a newer request is left in the mailbox
+            # for the next poll instead.
+            #
+            # This narrows the window rather than closing it: a request landing
+            # between this re-read and the delete is still lost. Closing it
+            # properly needs an atomic claim (a rename, or a compare-and-delete
+            # primitive) that the cache layer does not currently offer, so the
+            # honest fix is a smaller window plus this note, not a bigger lock.
+            current = self.cache_manager.get('display_on_demand_request',
+                                             max_age=3600, memory_ttl=0)
+            if not current or current.get('request_id') == request_id:
+                self.cache_manager.delete('display_on_demand_request')
+            else:
+                logger.debug("Newer on-demand request %s arrived while processing "
+                             "%s; leaving it in the mailbox",
+                             current.get('request_id'), request_id)
         except (OSError, AttributeError, KeyError) as err:
             # Best-effort: processed_id still guards against reprocessing if the
             # mailbox cannot be cleared.

src/plugin_system/testing/harness.py

@@ -256,11 +256,14 @@ def _settle_loop(inst, mode, dm, result, freezer) -> None:
             time.sleep(EMPTY_RECHECK_STEP)
         try:
             result.display_returned = _render_mode_again(inst, mode)
-        except Exception:  # noqa: BLE001
+        except Exception as e:  # noqa: BLE001
             # Deliberately broad: this calls a plugin's display(), which can
-            # raise anything. The first frame already rendered, so whatever
-            # happens on a re-draw must not turn a good result into an error --
-            # keep the frame we have and stop probing.
+            # raise anything. Recorded rather than swallowed -- a mode that
+            # renders one good frame and then crashes on the next is broken,
+            # and returning silently here reported it as passing. The frame
+            # already captured stays on the result so the failure is still
+            # inspectable.
+            result.error = repr(e)
             return
         image = dm.get_image()
         if _has_content(image):

src/plugin_system/testing/visual_display_manager.py

@@ -506,16 +506,17 @@ class VisualTestDisplayManager:
     # Scrolling state (no-op interface compat)
     # ------------------------------------------------------------------
 
-    def set_scrolling_state(self, is_scrolling: bool, frame_hold: int = 1):
+    def set_scrolling_state(self, is_scrolling: bool):
         """Set the current scrolling state (no-op for testing).
 
-        ``frame_hold`` mirrors DisplayManager.set_scrolling_state so a plugin
-        that paces its scroll can be rendered here. Without it every such
-        plugin raised TypeError at render time and failed every size, which is
-        invisible until a plugin happens to pass the argument.
+        Deliberately mirrors DisplayManager.set_scrolling_state exactly. A
+        double that accepts arguments production does not lets a call pass
+        every harness run and then raise TypeError on the panel, which is the
+        one failure a safety harness exists to prevent. ``frame_hold`` arrives
+        here in the same change that adds it to DisplayManager (#523), not
+        before.
         """
         self._scrolling_state['is_scrolling'] = is_scrolling
-        self._scrolling_state['frame_hold'] = frame_hold
         if is_scrolling:
             self._scrolling_state['last_scroll_activity'] = time.time()
 

test/test_harness_empty_claimed.py

@@ -101,3 +101,61 @@ class TestSkippedResults:
         r = _result(None, returned=None)
         check_empty_claimed([r], strict=True)
         assert r.empty_claimed is None
+
+
+class TestSettleRecordsLaterFailures:
+    """A mode that renders one good frame and then crashes is broken.
+
+    _settle_loop re-renders a mode that came back blank, to give a scroll or an
+    animation time to put something on the panel. Swallowing an exception from
+    those later frames meant the harness reported a passing result for a mode
+    that crashes as soon as it is asked for a second frame -- exactly the kind
+    of defect the harness exists to catch.
+    """
+
+    class Boom:
+        """Renders once, then raises."""
+
+        def __init__(self):
+            self.calls = 0
+
+        def display(self, force_clear=False):
+            self.calls += 1
+            raise RuntimeError("second frame exploded")
+
+    def _settle(self, inst, dm, result):
+        from src.plugin_system.testing import harness
+        harness._settle_loop(inst, "mode", dm, result, None)
+
+    def test_the_exception_is_recorded_on_the_result(self, monkeypatch):
+        from src.plugin_system.testing import harness
+        # Keep the probe short; this test is about the error, not the pacing.
+        monkeypatch.setattr(harness, "EMPTY_RECHECK_FRAMES", 1)
+        monkeypatch.setattr(harness, "EMPTY_RECHECK_STEP", 0)
+
+        result = _result(_blank())
+        assert result.error is None
+        self._settle(self.Boom(), _FakeDM(), result)
+
+        assert result.error is not None, "a crash on a later frame was swallowed"
+        assert "second frame exploded" in result.error
+
+    def test_the_already_captured_frame_is_kept(self, monkeypatch):
+        from src.plugin_system.testing import harness
+        monkeypatch.setattr(harness, "EMPTY_RECHECK_FRAMES", 1)
+        monkeypatch.setattr(harness, "EMPTY_RECHECK_STEP", 0)
+
+        image = _blank()
+        result = _result(image)
+        self._settle(self.Boom(), _FakeDM(), result)
+        assert result.image is image, "the good frame was discarded along with the error"
+
+
+class _FakeDM:
+    """Minimal display-manager double for _settle_loop."""
+
+    def get_image(self):
+        return _blank()
+
+    def check_overflow(self):
+        return None

test/test_on_demand_mailbox.py

@@ -0,0 +1,106 @@
+"""The on-demand request mailbox: how often it is read, and how it is consumed.
+
+The mailbox is a cache key the web process writes and the display process
+reads. Two properties matter and neither is obvious from the call site:
+
+  * it is polled after every rendered frame, so an uncached read here is a
+    disk read at frame rate;
+  * consuming it must not throw away a request that arrived while the previous
+    one was being processed.
+"""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+
+class TestPollingIsBounded:
+    """_poll_on_demand_requests runs ~125x/second on a scrolling mode.
+
+    The read is deliberately uncached (memory_ttl=0) because a cached one
+    pinned the first request for an hour. That makes the call a real disk read,
+    so it needs a floor -- without one it was ~125 reads per second to find
+    nothing at all.
+    """
+
+    def test_first_call_always_reads(self, test_display_controller):
+        c = test_display_controller
+        c.cache_manager.get = MagicMock(return_value=None)
+        c._poll_on_demand_requests()
+        assert c.cache_manager.get.call_count == 1
+
+    def test_immediate_second_call_does_not_read(self, test_display_controller):
+        c = test_display_controller
+        c.cache_manager.get = MagicMock(return_value=None)
+        c._poll_on_demand_requests()
+        for _ in range(50):
+            c._poll_on_demand_requests()
+        assert c.cache_manager.get.call_count == 1, "polling was not bounded"
+
+    def test_reads_again_once_the_interval_has_passed(self, test_display_controller, monkeypatch):
+        c = test_display_controller
+        c.cache_manager.get = MagicMock(return_value=None)
+        clock = {"t": 1000.0}
+        monkeypatch.setattr("src.display_controller.time.monotonic", lambda: clock["t"])
+
+        c._poll_on_demand_requests()
+        clock["t"] += c.ON_DEMAND_POLL_INTERVAL / 2
+        c._poll_on_demand_requests()
+        assert c.cache_manager.get.call_count == 1, "read before the interval elapsed"
+
+        clock["t"] += c.ON_DEMAND_POLL_INTERVAL
+        c._poll_on_demand_requests()
+        assert c.cache_manager.get.call_count == 2
+
+    def test_the_interval_is_short_enough_to_feel_instant(self, test_display_controller):
+        # A person clicking in the web UI must not notice the floor.
+        assert test_display_controller.ON_DEMAND_POLL_INTERVAL <= 0.5
+
+
+class TestMailboxIsConsumedByIdentity:
+    """Deleting whatever is in the mailbox loses a request that raced in."""
+
+    def _arrange(self, controller, first, later):
+        """Mailbox returns `first`, then `later` on the pre-delete re-read."""
+        controller.on_demand_active = False
+        controller.on_demand_request_id = None
+        controller._last_on_demand_poll = None
+        reads = iter([first, later])
+
+        def fake_get(key, *a, **kw):
+            if key == 'display_on_demand_request':
+                return next(reads, later)
+            return None  # processed-id lookup
+
+        controller.cache_manager.get = MagicMock(side_effect=fake_get)
+        controller.cache_manager.set = MagicMock()
+        controller.cache_manager.delete = MagicMock()
+        controller._activate_on_demand = MagicMock()
+
+    REQ_A = {'request_id': 'A', 'action': 'start', 'plugin_id': 'p', 'mode': 'm'}
+    REQ_B = {'request_id': 'B', 'action': 'start', 'plugin_id': 'p', 'mode': 'm'}
+
+    def test_own_request_is_deleted(self, test_display_controller):
+        c = test_display_controller
+        self._arrange(c, self.REQ_A, self.REQ_A)
+        c._poll_on_demand_requests()
+        c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
+
+    def test_a_newer_request_is_left_for_the_next_poll(self, test_display_controller):
+        c = test_display_controller
+        self._arrange(c, self.REQ_A, self.REQ_B)
+        c._poll_on_demand_requests()
+        assert c.cache_manager.delete.call_count == 0, \
+            "request B was deleted without ever being processed"
+
+    def test_an_already_empty_mailbox_is_still_cleared(self, test_display_controller):
+        c = test_display_controller
+        self._arrange(c, self.REQ_A, None)
+        c._poll_on_demand_requests()
+        c.cache_manager.delete.assert_called_once_with('display_on_demand_request')
+
+    def test_the_request_is_still_processed(self, test_display_controller):
+        c = test_display_controller
+        self._arrange(c, self.REQ_A, self.REQ_B)
+        c._poll_on_demand_requests()
+        c._activate_on_demand.assert_called_once()

test/test_starlark_display_contract.py

@@ -0,0 +1,82 @@
+"""starlark-apps: display() must report what actually reached the panel.
+
+The display controller skips a mode only on a boolean False. Returning True
+after the frame update failed told it the mode had rendered, so it held a dead
+frame for the whole display_duration instead of rotating on -- the same class
+of defect as a display() that returns None.
+"""
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+PLUGIN_DIR = Path(__file__).resolve().parent.parent / "plugin-repos" / "starlark-apps"
+
+
+@pytest.fixture(scope="module")
+def manager_module():
+    if not PLUGIN_DIR.exists():
+        pytest.skip("starlark-apps plugin is not checked out")
+    sys.path.insert(0, str(PLUGIN_DIR))
+    try:
+        import importlib
+        spec = importlib.util.spec_from_file_location(
+            "starlark_manager_under_test", PLUGIN_DIR / "manager.py")
+        module = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(module)
+        return module
+    except Exception as e:  # noqa: BLE001 - optional deps (pixlet, fcntl) may be absent
+        pytest.skip(f"starlark-apps manager is not importable here: {e}")
+    finally:
+        sys.path.remove(str(PLUGIN_DIR))
+
+
+def _plugin(manager_module):
+    """A manager with __init__ bypassed -- only display paths are under test."""
+    cls = manager_module.StarlarkAppsPlugin
+    inst = cls.__new__(cls)
+    inst.logger = MagicMock()
+    inst.display_manager = MagicMock()
+    inst.current_app = None
+    return inst
+
+
+class _App:
+    def __init__(self, frames):
+        self.frames = frames
+        self.current_frame_index = 0
+        self.last_frame_time = 0.0
+        self.app_id = "app"
+
+
+class TestDisplayFramePropagates:
+    def test_a_failed_update_returns_false(self, manager_module):
+        p = _plugin(manager_module)
+        p.current_app = _App([("frame", 100)])
+        p.display_manager.update_display.side_effect = RuntimeError("panel gone")
+
+        assert p._display_frame() is False
+
+    def test_a_good_update_returns_true(self, manager_module):
+        p = _plugin(manager_module)
+        p.current_app = _App([("frame", 100)])
+
+        assert p._display_frame() is True
+
+    def test_no_frames_returns_false(self, manager_module):
+        p = _plugin(manager_module)
+        p.current_app = _App([])
+
+        assert p._display_frame() is False
+
+    def test_display_reports_the_frame_failure(self, manager_module):
+        p = _plugin(manager_module)
+        p.current_app = _App([("frame", 100)])
+        p.display_manager.update_display.side_effect = RuntimeError("panel gone")
+
+        result = p.display()
+
+        assert result is False, "display() claimed success over a failed frame update"
+        assert isinstance(result, bool), "the controller only skips on a real bool"