Merge remote-tracking branch 'origin/main' into perf/dirty-tracking

# Conflicts:
#	src/display_manager.py

scripts/dev_server.py

@@ -236,13 +236,15 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
         try:
             plugin_instance.update()
         except Exception as e:
-            warnings.append(f"update() raised: {e}")
+            logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
+            warnings.append(f"update() raised: {type(e).__name__} — see server log")
 
     # Run display()
     try:
         plugin_instance.display(force_clear=True)
     except Exception as e:
-        errors.append(f"display() raised: {e}")
+        logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
+        errors.append(f"display() raised: {type(e).__name__} — see server log")
 
     render_time_ms = round((time.time() - start_time) * 1000, 1)
 
@@ -259,20 +261,25 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
 def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
     """Re-derive a plugin directory from the search dirs' own listings.
 
-    Path-injection barrier: the returned Path is constructed purely from
+    Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
-    trusted directory enumeration (``iterdir``) — request-derived strings
+    recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
+    returned Path is built from a trusted root plus a name the filesystem
+    itself produced under that root via scandir — request-derived strings
     never enter its construction — so a crafted plugin id can never make
     downstream file access leave the plugin search dirs. Comparison is by
-    path equality, deliberately without symlink resolution (dev plugins
+    name, deliberately without symlink resolution (dev plugins are
-    are commonly symlinked into plugins/).
+    commonly symlinked into plugins/).
     """
-    wanted = Path(os.path.normpath(str(plugin_dir)))
+    wanted_name = Path(os.path.normpath(str(plugin_dir))).name
     for search_dir in get_search_dirs():
-        if not search_dir.is_dir():
+        search_dir_str = str(search_dir)
+        try:
+            with os.scandir(search_dir_str) as entries:
+                for entry in entries:
+                    if entry.name == wanted_name and entry.is_dir():
+                        return Path(search_dir_str) / entry.name
+        except OSError:
             continue
-        for entry in search_dir.iterdir():
-            if entry.is_dir() and entry == wanted:
-                return entry
     return None
 
 
@@ -280,22 +287,25 @@ def _parse_render_request(data):
     """Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
     mock_data, skip_update) or raises ValueError with a client message."""
     plugin_id = data['plugin_id']
-    plugin_dir = find_plugin_dir(plugin_id)
+    candidate_dir = find_plugin_dir(plugin_id)
-    if plugin_dir:
+    # Never reuse `candidate_dir` past this point: it's built from
-        plugin_dir = _trusted_plugin_dir(plugin_dir)
+    # request-derived input, and a variable reassigned only on some paths
-    if not plugin_dir:
+    # isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
+    # sole name used below, always the scandir-sourced result.
+    trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
+    if not trusted_dir:
         raise LookupError(f'Plugin not found: {plugin_id}')
 
-    manifest_path = plugin_dir / 'manifest.json'
+    manifest_path = trusted_dir / 'manifest.json'
     with open(manifest_path, 'r') as f:
         manifest = json.load(f)
 
     # Build config: schema defaults + user overrides
     config = {'enabled': True}
-    config.update(load_config_defaults(plugin_dir))
+    config.update(load_config_defaults(trusted_dir))
     config.update(data.get('config', {}))
 
-    return plugin_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
+    return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
 
 
 @app.route('/api/render', methods=['POST'])

src/adaptive_images.py

@@ -25,7 +25,7 @@ uncached primitives.
 """
 
 from dataclasses import dataclass
-from typing import Any, Optional, Tuple, Union
+from typing import Any, Optional, Tuple
 
 from PIL import Image
 
@@ -136,7 +136,15 @@ def fit_image(img: Image.Image, box: Any, *, mode: str = "contain",
         scale = min(scale, 1.0)
     out_w = max(1, round(src_w * scale))
     out_h = max(1, round(src_h * scale))
-    out = work if (out_w, out_h) == (src_w, src_h) else work.resize((out_w, out_h), resample)
+    if (out_w, out_h) == (src_w, src_h):
+        # No resize needed — but `work` may still BE the caller's original
+        # image (RGBA source, no ink crop). The result must always be an
+        # independent copy: LayoutContext caches ImageFitResults, and an
+        # aliased image would let later mutations of the source corrupt
+        # cached fits (or vice versa).
+        out = work.copy() if work is img else work
+    else:
+        out = work.resize((out_w, out_h), resample)
     return ImageFitResult(out, out_w, out_h, scale, mode, (src_w, src_h))
 
 

src/adaptive_layout.py

@@ -29,7 +29,7 @@ freetype.Face, so it drops straight into DisplayManager.draw_text().
 
 import logging
 from collections import OrderedDict
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 import freetype
@@ -328,13 +328,28 @@ class LayoutContext:
         # fonts, which step between crisp ladder rungs instead.
         self.scale = min(self.width / max(1, design_w),
                          self.height / max(1, design_h))
-        self._fit_cache: Dict[Any, FitResult] = {}
+        # LRU-bounded: entries are small, but keys embed the fitted TEXT —
-        # LRU-bounded (images are big, unlike text fits). Entries hold a
+        # a plugin fitting changing text (a live game clock, a ticker) on a
-        # strong reference to the source image when keyed by id() so the id
+        # 24/7 service would otherwise grow this without bound.
-        # can't be recycled out from under the cache.
+        self._fit_cache: "OrderedDict[Any, FitResult]" = OrderedDict()
+        # LRU-bounded (images are big). Entries hold a strong reference to
+        # the source image when keyed by id() so the id can't be recycled
+        # out from under the cache.
         self._image_cache: "OrderedDict[Any, Tuple[Any, Any]]" = OrderedDict()
 
     _IMAGE_CACHE_MAX = 64
+    _FIT_CACHE_MAX = 512
+
+    def _fit_cache_get(self, key: Any) -> Optional["FitResult"]:
+        cached = self._fit_cache.get(key)
+        if cached is not None:
+            self._fit_cache.move_to_end(key)
+        return cached
+
+    def _fit_cache_put(self, key: Any, result: "FitResult") -> None:
+        self._fit_cache[key] = result
+        while len(self._fit_cache) > self._FIT_CACHE_MAX:
+            self._fit_cache.popitem(last=False)
 
     # ---- the three adaptation patterns --------------------------------
 
@@ -373,11 +388,11 @@ class LayoutContext:
         acceptable rendering exists."""
         box_w, box_h = _box_dims(box)
         key = ("text", text, box_w, box_h, ladder, ellipsis)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
         result = self._walk_ladder(text, ladder, box_w, box_h, ellipsis)
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
@@ -416,14 +431,14 @@ class LayoutContext:
         box_w, box_h = _box_dims(box)
         effective_scale = self.scale if scale is None else scale
         key = ("text_prop", text, box_w, box_h, ladder, base_size_px, ellipsis, effective_scale)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
         target = base_size_px * effective_scale
         eligible = [step for step in ladder if step.size_px <= target]
         candidates = eligible if eligible else (min(ladder, key=lambda s: s.size_px),)
         result = self._walk_ladder(text, candidates, box_w, box_h, ellipsis)
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
@@ -460,7 +475,7 @@ class LayoutContext:
         one wouldn't (baseball's multiline pattern). Text is the widest line."""
         box_w, box_h = _box_dims(box)
         key = ("lines", tuple(lines), box_w, box_h, ladder, spacing)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
 
@@ -482,7 +497,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def font_for_rows(self, rows: int, box_h: int,
@@ -491,7 +506,7 @@ class LayoutContext:
         (baseball's traditional-scoreboard pattern). Measures a digit/cap
         sample rather than specific strings."""
         key = ("rows", rows, box_h, ladder)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
 
@@ -508,7 +523,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     # ---- images ---------------------------------------------------------

src/cache/disk_cache.py

@@ -10,6 +10,7 @@ import time
 import tempfile
 import logging
 import threading
+import zlib
 from typing import Dict, Any, Optional, Protocol
 from datetime import datetime
 
@@ -53,6 +54,11 @@ class DiskCache:
         self.cache_dir = cache_dir
         self.logger = logger or logging.getLogger(__name__)
         self._lock = threading.Lock()
+        # key -> adler32 of the last payload successfully written to the
+        # primary cache path; lets set() skip rewriting identical data
+        # (per-process only — worst case another process rewrites, never
+        # a missed write). Guarded by _lock.
+        self._write_digests: Dict[str, int] = {}
     
     def get_cache_path(self, key: str) -> Optional[str]:
         """
@@ -156,9 +162,34 @@ class DiskCache:
         if not cache_path:
             return
 
+        # Serialize once, compact (no indent): the payload is reused by every
+        # write path below, and cache files are machine-read only — indenting
+        # them just multiplied the bytes written to the SD card.
+        try:
+            payload = json.dumps(data, cls=DateTimeEncoder)
+        except (TypeError, ValueError) as e:
+            self.logger.warning("Cache data for key '%s' not serializable: %s", key, e)
+            return
+
+        digest = zlib.adler32(payload.encode('utf-8'))
+
         try:
             # Atomic write to avoid partial/corrupt files
             with self._lock:
+                # Skip the disk entirely when this exact payload was already
+                # written for this key (plugins re-save unchanged API data
+                # every update cycle — each write is real SD-card wear).
+                # Refresh the file mtime so records that rely on it for TTL
+                # (no embedded 'timestamp') don't expire early; a metadata
+                # touch is journal-cheap compared to rewriting the data.
+                if self._write_digests.get(key) == digest:
+                    try:
+                        os.utime(cache_path, None)
+                        return
+                    except OSError:
+                        # File vanished or perms changed — fall through and write
+                        self._write_digests.pop(key, None)
+
                 tmp_dir = os.path.dirname(cache_path)
                 # Try to create temp file in cache directory first
                 # If that fails due to permissions, fall back to direct write
@@ -181,13 +212,17 @@ class DiskCache:
                         fd = None
                     
                     if tmp_path and fd is not None:
-                        # Use atomic write with temp file
+                        # Atomic write with temp file. No fsync: os.replace
+                        # already guarantees readers never see a torn file,
+                        # and cache data is re-fetchable — forcing a disk
+                        # flush per write was the single biggest SD-card
+                        # wear source (dozens of fsyncs/min on API-heavy
+                        # installs) for data that can be re-downloaded.
                         try:
                             with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
-                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
+                                tmp_file.write(payload)
-                                tmp_file.flush()
-                                os.fsync(tmp_file.fileno())
                             os.replace(tmp_path, cache_path)
+                            self._write_digests[key] = digest
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -203,9 +238,8 @@ class DiskCache:
                         # Fallback: direct write (not atomic, but better than failing)
                         try:
                             with open(cache_path, 'w', encoding='utf-8') as cache_file:
-                                json.dump(data, cache_file, indent=4, cls=DateTimeEncoder)
+                                cache_file.write(payload)
-                                cache_file.flush()
+                            self._write_digests[key] = digest
-                                os.fsync(cache_file.fileno())
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -229,9 +263,12 @@ class DiskCache:
                             pass
                         
                         if os.path.isdir(fallback_dir) and os.access(fallback_dir, os.W_OK):
+                            # NOTE: no digest record here — the fallback file
+                            # is a different path, so future sets must keep
+                            # retrying the primary location.
                             fallback_path = os.path.join(fallback_dir, os.path.basename(cache_path))
                             with open(fallback_path, 'w', encoding='utf-8') as tmp_file:
-                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
+                                tmp_file.write(payload)
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(fallback_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -272,6 +309,7 @@ class DiskCache:
         
         with self._lock:
             if key:
+                self._write_digests.pop(key, None)
                 cache_path = self.get_cache_path(key)
                 if cache_path and os.path.exists(cache_path):
                     try:
@@ -280,6 +318,7 @@ class DiskCache:
                         self.logger.warning("Could not remove cache file %s: %s", cache_path, e)
             else:
                 # Clear all cache files
+                self._write_digests.clear()
                 if os.path.exists(self.cache_dir):
                     for filename in os.listdir(self.cache_dir):
                         if filename.endswith('.json'):

src/common/snapshot_policy.py

@@ -0,0 +1,68 @@
+"""Snapshot write policy for the display preview mirror.
+
+The display service mirrors frames to /tmp/led_matrix_preview.png, which
+serves two consumers with different needs:
+
+- The web UI's live preview (SSE reader in web_interface/app.py) wants
+  fresh frames — but only while a browser is actually watching.
+- The health check (web_interface/blueprints/api_v3.py, hardware status)
+  uses the file's AGE as a liveness proxy: age >= 60s reads as degraded.
+
+PNG-encoding every frame at 5 fps forever — identical frames, no viewers —
+was one of the biggest fixed CPU costs on the Pi. This module is the pure
+decision logic (extracted so it's unit-testable off-Pi; display_manager
+imports rgbmatrix unconditionally and can't be):
+
+    WRITE  — encode + atomically replace the snapshot file
+    TOUCH  — os.utime only: keeps the health-check mtime fresh and lets
+             the SSE reader (mtime-gated) resend at a low rate, without
+             paying for a PNG encode of an unchanged frame
+    SKIP   — do nothing
+
+Policy:
+- With a fresh viewer marker: changed frames write at up to 1/VIEWER_INTERVAL.
+- Without viewers: changed frames still write at 1/IDLE_INTERVAL so the
+  preview page shows something recent on open.
+- Unchanged frames are never re-encoded; the mtime is touched every
+  TOUCH_INTERVAL so the health check (60s threshold) never degrades.
+
+If any constant here changes, re-check the health threshold in
+api_v3.py (get_hardware_status) — TOUCH_INTERVAL must stay well under it.
+"""
+
+from enum import Enum
+
+# Snapshot cadence with a browser preview open (seconds).
+VIEWER_INTERVAL = 0.2
+# Snapshot cadence with no viewers — cheap freshness for page-open (seconds).
+IDLE_INTERVAL = 30.0
+# Max age of the last write/touch before bumping mtime for the health
+# check. MUST stay well under api_v3's 60s degraded threshold.
+TOUCH_INTERVAL = 20.0
+# A viewer marker older than this no longer counts as a live viewer.
+VIEWER_MARKER_FRESH_SEC = 5.0
+
+
+class SnapshotAction(Enum):
+    WRITE = "write"
+    TOUCH = "touch"
+    SKIP = "skip"
+
+
+def decide(now: float, last_write_ts: float, last_touch_ts: float,
+           viewer_fresh: bool, frame_changed: bool) -> SnapshotAction:
+    """Decide what to do with the current frame.
+
+    Args:
+        now: current monotonic-ish timestamp (same clock as the ts args)
+        last_write_ts: when a frame was last actually encoded+written
+        last_touch_ts: when the file mtime was last bumped (write or touch)
+        viewer_fresh: a browser preview is currently watching
+        frame_changed: the frame differs from the last WRITTEN frame
+    """
+    interval = VIEWER_INTERVAL if viewer_fresh else IDLE_INTERVAL
+    if frame_changed and (now - last_write_ts) >= interval:
+        return SnapshotAction.WRITE
+    if (now - max(last_write_ts, last_touch_ts)) >= TOUCH_INTERVAL:
+        return SnapshotAction.TOUCH
+    return SnapshotAction.SKIP

src/display_controller.py

@@ -199,6 +199,10 @@ class DisplayController:
         self.wifi_status_file = WIFI_STATUS_FILE
         self.wifi_status_active = False
         self.wifi_status_expires_at: Optional[float] = None
+        # _check_wifi_status_message throttle state (checked at frame rate,
+        # stat'd at most once per second)
+        self._wifi_status_check_ts = 0.0
+        self._wifi_status_last_result: Optional[Dict[str, Any]] = None
 
         # Plugin display() signature cache — must be initialised before the plugin
         # loading loop below so the .pop() invalidation at load time is always safe.
@@ -502,7 +506,10 @@ class DisplayController:
 
             # Run plugin updates inside the Vegas loop so the inter-iteration
             # gap is <1 ms (nothing left for _tick_plugin_updates() to do).
-            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
+            # Use the Vegas-aware variant so plugins that got fresh data are
+            # hot-swapped into the scroll promptly instead of waiting for the
+            # next full cycle.
+            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates_for_vegas)
 
             # Wire multi-display sync into Vegas render pipeline
             follower_pos = self.config.get("sync", {}).get("follower_position", "left")
@@ -625,14 +632,24 @@ class DisplayController:
         # Check if per-day schedule is configured
         days_config = schedule_config.get('days')
 
-        # Determine which schedule to use
+        # Determine which schedule to use. Respect an explicit 'mode' field
+        # (like the dim schedule does) so a stray/legacy 'days' dict left over
+        # from config migration or a prior per-day setup can't silently
+        # override a user's Global schedule selection.
+        mode = schedule_config.get('mode')
+        mode_normalized = mode.replace('_', '-') if mode else None
+
         use_per_day = False
-        if days_config:
+        if mode_normalized == 'global':
-            # Check if days dict is not empty and contains current day
+            use_per_day = False
-            if days_config and current_day in days_config:
+        elif mode_normalized == 'per-day':
+            use_per_day = bool(days_config and current_day in days_config)
+        elif days_config:
+            # No explicit mode recorded (legacy config) - fall back to
+            # inferring from presence of a 'days' dict for the current day.
+            if current_day in days_config:
                 use_per_day = True
-            elif days_config:
+            else:
-                # Days dict exists but doesn't have current day - fall back to global
                 logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
         
         if use_per_day:
@@ -828,6 +845,42 @@ class DisplayController:
                     if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                         self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
 
+    def _tick_plugin_updates_for_vegas(self) -> None:
+        """Run scheduled plugin updates and tell Vegas mode which plugins
+        actually got fresh data, so it can hot-swap them into the scroll
+        without waiting for a full cycle to complete.
+
+        Used as the Vegas coordinator's update callback instead of the plain
+        _tick_plugin_updates() so that a live score change is reflected in
+        the ticker within a few seconds rather than at the next cycle
+        boundary (which, depending on min/max_cycle_duration, can be
+        minutes away). Restores wiring that PR #299 added and PR #330's
+        sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
+        has been unreachable dead code since.
+
+        Delegates the before/after plugin_last_update snapshot to
+        PluginManager.run_scheduled_updates_with_changes() so the snapshot,
+        update pass, and diff are lock-protected against this callback's own
+        background update-tick thread racing the main render loop.
+        """
+        if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
+            self._tick_plugin_updates()
+            return
+
+        updated = self.plugin_manager.run_scheduled_updates_with_changes()
+
+        vc = getattr(self, "vegas_coordinator", None)
+        if vc is None:
+            return
+
+        if updated:
+            logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
+            for plugin_id in updated:
+                try:
+                    vc.mark_plugin_updated(plugin_id)
+                except Exception:  # pylint: disable=broad-except
+                    logger.exception("Error marking plugin %s updated for Vegas", plugin_id)
+
     def _tick_plugin_updates(self):
         """Run scheduled plugin updates if the plugin manager supports them."""
         if not self.plugin_manager:
@@ -1635,7 +1688,7 @@ class DisplayController:
                     self._sleep_with_plugin_updates(60)
                     continue
                 
-                logger.info(f"Display active, processing mode: {self.current_display_mode}")
+                logger.debug("Display active, processing mode: %s", self.current_display_mode)
                 
                 # Plugins update on their own schedules - no forced sync updates needed
                 # Each plugin has its own update_interval and background services
@@ -1803,7 +1856,7 @@ class DisplayController:
                         if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                             should_skip = self.plugin_manager.health_tracker.should_skip_plugin(plugin_id)
                             if should_skip:
-                                logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
+                                logger.info("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
                                 display_result = False
                                 # Skip to next mode - let existing logic handle it
                                 manager_to_display = None
@@ -1861,7 +1914,7 @@ class DisplayController:
                             if isinstance(result, bool):
                                 display_result = result
                                 if not display_result:
-                                    logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
+                                    logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
                         
                         # Record success if display completed without exception
                         if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
@@ -2072,10 +2125,23 @@ class DisplayController:
 
                     # For plugins, call display multiple times to allow game rotation
                     if manager_to_display and hasattr(manager_to_display, 'display'):
-                        # Check if plugin needs high FPS (like stock ticker)
+                        # High-FPS decision, in precedence order:
-                        # Always enable high-FPS for static-image plugin (for GIF animation support)
+                        # 1. A plugin that declares needs_high_fps knows best
+                        #    (e.g. static-image sets it False for still PNGs,
+                        #    True for animated GIFs).
+                        # 2. Back-compat: older static-image versions without
+                        #    the attribute keep the historical forced high-FPS
+                        #    (GIF support).
+                        # 3. Otherwise scrolling plugins get high FPS.
                         plugin_id = getattr(manager_to_display, 'plugin_id', None)
-                        if plugin_id == 'static-image':
+                        declared = getattr(manager_to_display, 'needs_high_fps', None)
+                        if declared is not None:
+                            needs_high_fps = bool(declared)
+                            logger.debug(
+                                "[DisplayController] FPS check for %s (plugin=%s) - "
+                                "plugin declares needs_high_fps=%s",
+                                active_mode, plugin_id, needs_high_fps)
+                        elif plugin_id == 'static-image':
                             needs_high_fps = True
                             logger.debug("FPS check - static-image plugin: forcing high-FPS mode for GIF support")
                         else:
@@ -2354,6 +2420,16 @@ class DisplayController:
             Returns None on any error or if message is expired/invalid.
         """
         try:
+            # Throttle the existence stat to ~1 Hz: this runs on every render
+            # iteration (60+ fps), and the file usually doesn't exist — the
+            # status message's lifetime is measured in seconds anyway.
+            # Both attributes are initialised in __init__.
+            now = time.time()
+            if (now - self._wifi_status_check_ts) < 1.0:
+                return self._wifi_status_last_result
+            self._wifi_status_check_ts = now
+            self._wifi_status_last_result = None
+
             # Check if file exists
             if not self.wifi_status_file or not self.wifi_status_file.exists():
                 return None
@@ -2404,13 +2480,14 @@ class DisplayController:
                     pass
                 return None
             
-            # Message is valid and not expired
+            # Message is valid and not expired — cache for the throttle window
-            return {
+            self._wifi_status_last_result = {
                 'message': message,
                 'timestamp': timestamp,
                 'duration': duration,
                 'expires_at': expires_at
             }
+            return self._wifi_status_last_result
             
         except Exception as e:
             # Catch-all for any unexpected errors - log but don't break the display

src/display_manager.py

@@ -31,13 +31,25 @@ if os.getenv("EMULATOR", "false") == "true":
 else:
     from rgbmatrix import RGBMatrix, RGBMatrixOptions
 from contextlib import contextmanager
+from pathlib import Path
 from PIL import Image, ImageDraw, ImageFont
+import threading
 import time
-from typing import Dict, Any, List, Optional
+from collections import OrderedDict
+from typing import Dict, Any, List, Optional, Tuple
 import logging
 import math
+import zlib
 import freetype
 
+from src.common import snapshot_policy
+from src.common.permission_utils import (
+    ensure_directory_permissions,
+    ensure_file_permissions,
+    get_assets_dir_mode,
+    get_assets_file_mode,
+)
+
 # Get logger without configuring
 logger = logging.getLogger(__name__)
 logger.setLevel(logging.INFO)  # Set to INFO level
@@ -180,14 +192,45 @@ class DisplayManager:
         # the logical image is blitted to the matrix unchanged.
         self._double_sided = None  # dict {copies, axis, logical_width, logical_height} or None
         self._physical_image = None  # full-chain buffer reused each frame when tiling
-        # Text-width measurement cache: (text, id(font)) -> pixel_width
+        # Text-width measurement cache: (text, id(font)) -> (width, font_ref)
         # Avoids re-measuring the same string+font on every display() call.
+        # LRU-bounded: keys embed the TEXT, so changing strings (a clock, a
+        # live score) would otherwise grow it forever on a 24/7 service.
+        # Entries hold a strong reference to the font so its id() can't be
+        # recycled by a different font object — an id-keyed cache without
+        # the reference can return the WRONG width after garbage collection.
         # Cleared on _load_fonts() so stale entries don't survive a font reload.
-        self._text_width_cache: Dict[tuple, int] = {}
+        self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
-        # Snapshot settings for web preview integration (service writes, web reads)
+        self._TEXT_WIDTH_CACHE_MAX = 1024
+        # Snapshot mirror for web preview + health check (service writes, web
+        # reads). Cadence/skip decisions live in src/common/snapshot_policy.py:
+        # full rate only while the web SSE broadcaster keeps the viewer marker
+        # fresh; unchanged frames are never re-encoded, only mtime-touched.
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
-        self._snapshot_min_interval_sec = 0.2  # max ~5 fps
+        self._viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - touched by web SSE broadcaster
         self._last_snapshot_ts = 0.0
+        self._last_snapshot_touch_ts = 0.0
+        self._last_snapshot_digest: Optional[int] = None
+        self._snapshot_dir_prepared = False
+        self._viewer_check_ts = 0.0
+        self._viewer_fresh = False
+        self._viewer_was_fresh = False
+        # Snapshot failures are logged as warnings, rate-limited so a
+        # persistent failure (e.g. an unwritable file) can't spam the log —
+        # but is never silent: the snapshot's mtime doubles as the web UI's
+        # hardware-liveness signal, so a quiet failure makes health checks lie.
+        self._snapshot_fail_log_ts = 0.0
+        # Dirty tracking: (image digest, brightness) of the last frame pushed
+        # to the panel; update_display() skips identical pushes. Kill switch:
+        # display.dirty_tracking: false.
+        self._dirty_tracking_enabled = bool(
+            self.config.get('display', {}).get('dirty_tracking', True))
+        self._last_pushed_digest = None
+        # Serializes update_display(): plugins can call it directly from
+        # background threads (see docstring on update_display), not just the
+        # render loop. RLock in case a caller within the critical section
+        # ever re-enters (e.g. via a nested draw callback).
+        self._update_lock = threading.RLock()
         
         # Scrolling state tracking for graceful updates
         self._scrolling_state = {
@@ -418,6 +461,10 @@ class DisplayManager:
         try:
             # RGBMatrix accepts brightness as a property
             self.matrix.brightness = brightness
+            # Brightness applies on the next swap — force a re-push even if
+            # the image itself is unchanged (belt-and-braces: brightness is
+            # also part of the dirty-tracking digest when readable).
+            self._last_pushed_digest = None
             logger.info(f"[BRIGHTNESS] Display brightness set to {brightness}%")
             return True
         except AttributeError as e:
@@ -509,33 +556,70 @@ class DisplayManager:
         return phys
 
     def update_display(self):
-        """Update the display using double buffering with proper sync."""
+        """Update the display using double buffering with proper sync.
+
+        Skips the panel push entirely when the frame is byte-identical to
+        the last pushed one (same image digest AND same brightness) — static
+        content re-rendered every second, and 125 fps loops between actual
+        scroll steps, otherwise re-walk the full framebuffer for nothing.
+        The panel keeps refreshing the current frame from its own thread,
+        so skipping a swap never blanks or freezes the hardware.
+
+        Correctness hinges on invalidation: clear() resets the digest (it
+        writes to the matrix directly), and brightness is PART of the digest
+        so a dim-schedule change is never skipped. Disable via config
+        ``display.dirty_tracking: false`` if a redraw issue is ever suspected.
+
+        Serialized via ``_update_lock``: plugins can call this directly from
+        background threads (e.g. sports base classes push an immediate
+        "live" refresh from inside update()), so without a lock two callers
+        could both pass the digest check before either writes it back,
+        double-pushing a frame, or interleave the offscreen/current canvas
+        swap below. The lock is scoped to this method, so callers never
+        need to know about it.
+        """
         try:
-            if self.matrix is None:
+            with self._update_lock:
-                # Fallback mode - no actual hardware to update
+                if self.matrix is None:
-                logger.debug("Update display called in fallback mode (no hardware)")
+                    # Fallback mode - no actual hardware to update
-                # Still write a snapshot so the web UI can preview
+                    logger.debug("Update display called in fallback mode (no hardware)")
+                    # Still write a snapshot so the web UI can preview
+                    self._write_snapshot_if_due()
+                    return
+
+                if self._capture_mode_active:
+                    return  # Skip hardware write — content is being captured off-screen
+
+                digest = None
+                if self._dirty_tracking_enabled:
+                    try:
+                        brightness = getattr(self.matrix, 'brightness', None)
+                    except AttributeError:
+                        brightness = None
+                    digest = (zlib.adler32(self.image.tobytes()), brightness)
+                    if digest == self._last_pushed_digest:
+                        # Nothing changed since the last push — the panel is
+                        # already showing exactly this frame.
+                        self._write_snapshot_if_due()
+                        return
+
+                # Copy the current image to the offscreen canvas. In double-sided
+                # mode the logical screen is first tiled across the full chain.
+                if self._double_sided is not None:
+                    self.offscreen_canvas.SetImage(self._composite_double_sided())
+                else:
+                    self.offscreen_canvas.SetImage(self.image)
+
+                # Swap buffers immediately
+                self.matrix.SwapOnVSync(self.offscreen_canvas)
+
+                # Swap our canvas references
+                self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
+
+                self._last_pushed_digest = digest
+
+                # Write a snapshot for the web preview (throttled)
                 self._write_snapshot_if_due()
-                return
-
-            if self._capture_mode_active:
-                return  # Skip hardware write — content is being captured off-screen
-
-            # Copy the current image to the offscreen canvas. In double-sided
-            # mode the logical screen is first tiled across the full chain.
-            if self._double_sided is not None:
-                self.offscreen_canvas.SetImage(self._composite_double_sided())
-            else:
-                self.offscreen_canvas.SetImage(self.image)
-
-            # Swap buffers immediately
-            self.matrix.SwapOnVSync(self.offscreen_canvas)
-            
-            # Swap our canvas references
-            self.offscreen_canvas, self.current_canvas = self.current_canvas, self.offscreen_canvas
-
-            # Write a snapshot for the web preview (throttled)
-            self._write_snapshot_if_due()
         except Exception as e:
             logger.error(f"Error updating display: {e}")
 
@@ -569,6 +653,9 @@ class DisplayManager:
                 # Clear both canvases and the underlying matrix to ensure no artifacts.
                 # Failures are non-fatal — the image buffer is already black above, so
                 # the next update_display() call will push clean content regardless.
+                # The matrix content no longer matches the last pushed digest,
+                # so dirty tracking must not skip the next push.
+                self._last_pushed_digest = None
                 try:
                     self.offscreen_canvas.Clear()
                 except (RuntimeError, OSError) as e:
@@ -699,12 +786,15 @@ class DisplayManager:
 
         Results are cached by (text, font identity) so plugins that measure
         the same string every frame (e.g. to centre a score) pay only one
-        measurement per unique (text, font) pair.
+        measurement per unique (text, font) pair. The entry keeps the font
+        alive so its id() can't be recycled, and the cache is LRU-bounded so
+        ever-changing text (clocks, tickers) can't grow it without limit.
         """
         cache_key = (text, id(font))
         cached = self._text_width_cache.get(cache_key)
         if cached is not None:
-            return cached
+            self._text_width_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -719,7 +809,9 @@ class DisplayManager:
             logger.error("Error getting text width: %s", e)
             return 0
 
-        self._text_width_cache[cache_key] = width
+        self._text_width_cache[cache_key] = (width, font)
+        while len(self._text_width_cache) > self._TEXT_WIDTH_CACHE_MAX:
+            self._text_width_cache.popitem(last=False)
         return width
 
     def get_font_height(self, font):
@@ -1128,27 +1220,56 @@ class DisplayManager:
             'deferred_update_ttl': self._scrolling_state['deferred_update_ttl']
         }
 
+    def _viewer_is_fresh(self, now: float) -> bool:
+        """True when a browser preview is watching (marker file touched by
+        the web SSE broadcaster). The marker is stat'd at most once per
+        second — at 125 fps loops a per-call stat would be pure overhead."""
+        if (now - self._viewer_check_ts) >= 1.0:
+            self._viewer_check_ts = now
+            try:
+                marker_age = now - os.stat(self._viewer_marker_path).st_mtime
+                self._viewer_fresh = marker_age < snapshot_policy.VIEWER_MARKER_FRESH_SEC
+            except OSError:
+                self._viewer_fresh = False
+        return self._viewer_fresh
+
     def _write_snapshot_if_due(self) -> None:
-        """Write the current image to a PNG snapshot file at a limited frequency."""
+        """Mirror the current frame to the preview snapshot when the policy
+        says it's worth it — see src/common/snapshot_policy.py. Unchanged
+        frames are never re-encoded; without viewers the cadence drops to
+        the idle keepalive."""
         try:
             now = time.time()
-            if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
+            viewer_fresh = self._viewer_is_fresh(now)
+            if viewer_fresh and not self._viewer_was_fresh:
+                # A preview just opened: let the next changed frame through
+                # immediately instead of waiting out the idle interval.
+                self._last_snapshot_ts = 0.0
+            self._viewer_was_fresh = viewer_fresh
+
+            digest = zlib.adler32(self.image.tobytes())
+            action = snapshot_policy.decide(
+                now, self._last_snapshot_ts, self._last_snapshot_touch_ts,
+                viewer_fresh, digest != self._last_snapshot_digest)
+            if action is snapshot_policy.SnapshotAction.SKIP:
                 return
-            # Ensure directory exists with proper permissions
+            if action is snapshot_policy.SnapshotAction.TOUCH:
-            from pathlib import Path
+                # mtime bump only: keeps the health check (snapshot age)
-            from src.common.permission_utils import (
+                # green without paying for a PNG encode of an unchanged frame
-                ensure_directory_permissions,
+                os.utime(self._snapshot_path, None)
-                ensure_file_permissions,
+                self._last_snapshot_touch_ts = now
-                get_assets_dir_mode,
+                return
-                get_assets_file_mode
+
-            )
+            # WRITE: ensure directory permissions once, not per frame
             snapshot_path_obj = Path(self._snapshot_path)
-            # Only ensure permissions on non-system directories
+            if not self._snapshot_dir_prepared:
-            # Never modify /tmp permissions - it has special system permissions (1777)
+                # Never modify /tmp permissions - it has special system
-            # that must not be changed or it breaks apt and other system tools
+                # permissions (1777) that must not be changed or it breaks
-            parent_dir = snapshot_path_obj.parent
+                # apt and other system tools
-            if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
+                parent_dir = snapshot_path_obj.parent
-                ensure_directory_permissions(parent_dir, get_assets_dir_mode())
+                if parent_dir and str(parent_dir) != '/tmp':  # nosec B108 - guard to skip /tmp for permission ops
+                    ensure_directory_permissions(parent_dir, get_assets_dir_mode())
+                self._snapshot_dir_prepared = True
             # Write atomically: temp then replace
             tmp_path = f"{self._snapshot_path}.tmp"
             self.image.save(tmp_path, format='PNG')
@@ -1163,6 +1284,18 @@ class DisplayManager:
             except Exception:
                 pass
             self._last_snapshot_ts = now
+            self._last_snapshot_touch_ts = now
+            self._last_snapshot_digest = digest
         except Exception as e:
-            # Snapshot failures should never break display; log at debug to avoid noise
+            # Snapshot failures must never break display — but they must not
-            logger.debug(f"Snapshot write skipped: {e}")
+            # be silent either: the snapshot's mtime is the web UI's display
+            # mirror AND its hardware-liveness proxy, so a quietly failing
+            # write freezes the mirror and makes health checks lie (seen in
+            # the field: a stale root-owned /tmp file froze it for a day).
+            # Warn at most once per 5 minutes to avoid log spam.
+            if (now - self._snapshot_fail_log_ts) > 300:
+                self._snapshot_fail_log_ts = now
+                logger.warning("Snapshot write failing (web preview/health "
+                               "mirror is stale): %s", e)
+            else:
+                logger.debug(f"Snapshot write skipped: {e}")

src/font_manager.py

@@ -35,6 +35,7 @@ import urllib.request
 import zipfile
 import tempfile
 import time
+from collections import OrderedDict
 from pathlib import Path
 from PIL import ImageFont
 from typing import Dict, Tuple, Optional, Union, Any, List
@@ -58,7 +59,13 @@ class FontManager:
         # Font discovery and catalog
         self.font_catalog: Dict[str, str] = {}  # family_name -> file_path
         self.font_cache: Dict[str, Union[ImageFont.FreeTypeFont, freetype.Face]] = {}  # (family, size) -> font
-        self.metrics_cache: Dict[str, Tuple[int, int, int]] = {}  # (text, font_id) -> (width, height, baseline)
+        # (text, id(font)) -> ((width, height, baseline), font_ref).
+        # LRU-bounded — keys embed the measured TEXT, so changing strings
+        # (clocks, live scores) would otherwise grow it forever. Entries
+        # keep the font alive so its id() can't be recycled by a different
+        # font object (which would silently return wrong metrics).
+        self.metrics_cache: "OrderedDict[Any, Tuple[Tuple[int, int, int], Any]]" = OrderedDict()
+        self._METRICS_CACHE_MAX = 1024
 
         # Plugin font management
         self.plugin_fonts: Dict[str, Dict[str, Any]] = {}  # plugin_id -> font_manifest
@@ -555,10 +562,14 @@ class FontManager:
         Returns:
             Tuple of (width, height, baseline_offset)
         """
-        cache_key = f"{hash(text)}_{id(font)}"
+        # Key on the text itself (hash(text) could collide) + font identity;
+        # the entry below keeps the font referenced so the id stays valid.
+        cache_key = (text, id(font))
 
-        if cache_key in self.metrics_cache:
+        cached = self.metrics_cache.get(cache_key)
-            return self.metrics_cache[cache_key]
+        if cached is not None:
+            self.metrics_cache.move_to_end(cache_key)
+            return cached[0]
 
         try:
             if isinstance(font, freetype.Face):
@@ -595,7 +606,9 @@ class FontManager:
             baseline = 10
 
         result = (width, height, baseline)
-        self.metrics_cache[cache_key] = result
+        self.metrics_cache[cache_key] = (result, font)
+        while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
+            self.metrics_cache.popitem(last=False)
         return result
 
     def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:

src/plugin_system/plugin_loader.py

@@ -437,8 +437,7 @@ class PluginLoader:
                 if not Path(existing_file).resolve().is_relative_to(resolved_dir):
                     evicted[mod_name] = sys.modules.pop(mod_name)
                     self.logger.debug(
-                        "Evicted stale module '%s' (from %s) before loading plugin in %s",
+                        "Evicted stale bare-name module '%s' before loading plugin", mod_name,
-                        mod_name, existing_file, plugin_dir,
                     )
             except (ValueError, TypeError):
                 continue
@@ -551,7 +550,7 @@ class PluginLoader:
             plugin_dir_str = str(plugin_dir)
             if plugin_dir_str not in sys.path:
                 sys.path.insert(0, plugin_dir_str)
-                self.logger.debug("Added plugin directory to sys.path: %s", plugin_dir_str)
+                self.logger.debug("Added plugin %s's directory to sys.path", plugin_id)
 
             # Import the plugin module
             module_name = f"plugin_{plugin_id.replace('-', '_')}"
@@ -563,8 +562,8 @@ class PluginLoader:
 
             spec = importlib.util.spec_from_file_location(module_name, entry_file)
             if spec is None or spec.loader is None:
+                self.logger.error("Could not create module spec for plugin %s", plugin_id)
                 error_msg = f"Could not create module spec for {entry_file}"
-                self.logger.error(error_msg)
                 raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
 
             module = importlib.util.module_from_spec(spec)

src/plugin_system/plugin_manager.py

@@ -76,6 +76,12 @@ class PluginManager:
         # concurrent mutation (background reconciliation) and reads (requests).
         self._discovery_lock = threading.RLock()
 
+        # Lock protecting plugin_last_update from concurrent mutation/iteration.
+        # It's written from run_scheduled_updates()/update_all_plugins() (main
+        # loop) and read/diffed by run_scheduled_updates_with_changes(), which
+        # Vegas mode calls from its own background update-tick thread.
+        self._plugin_last_update_lock = threading.RLock()
+
         # Active plugins
         self.plugins: Dict[str, Any] = {}
         self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
@@ -317,7 +323,8 @@ class PluginManager:
 
             # Store plugin instance
             self.plugins[plugin_id] = plugin_instance
-            self.plugin_last_update[plugin_id] = 0.0
+            with self._plugin_last_update_lock:
+                self.plugin_last_update[plugin_id] = 0.0
             # Invalidate cached interval so next tick re-derives it for this plugin
             self._update_interval_cache.pop(plugin_id, None)
             
@@ -429,7 +436,8 @@ class PluginManager:
             
             # Remove from active plugins
             del self.plugins[plugin_id]
-            self.plugin_last_update.pop(plugin_id, None)
+            with self._plugin_last_update_lock:
+                self.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
             
             # Remove main module from sys.modules if present
@@ -698,7 +706,8 @@ class PluginManager:
             'recoverable': True,
         }
         self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
-        self.plugin_last_update[plugin_id] = failure_time
+        with self._plugin_last_update_lock:
+            self.plugin_last_update[plugin_id] = failure_time
         self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
         if self.health_tracker:
             self.health_tracker.record_failure(plugin_id, err)
@@ -731,7 +740,8 @@ class PluginManager:
             if interval is None:
                 continue
 
-            last_update = self.plugin_last_update.get(plugin_id, 0.0)
+            with self._plugin_last_update_lock:
+                last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
                 # Update state to RUNNING
@@ -762,7 +772,8 @@ class PluginManager:
                         success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                     
                     if success:
-                        self.plugin_last_update[plugin_id] = current_time
+                        with self._plugin_last_update_lock:
+                            self.plugin_last_update[plugin_id] = current_time
                         self.state_manager.record_update(plugin_id)
                         # Update state back to ENABLED
                         self.state_manager.set_state(plugin_id, PluginState.ENABLED)
@@ -775,6 +786,31 @@ class PluginManager:
                     self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
                     self._record_update_failure(plugin_id, exc=exc)
 
+    def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
+        """
+        Like run_scheduled_updates(), but also returns the plugin_ids whose
+        plugin_last_update timestamp actually advanced during this call.
+
+        The before/after snapshots and the update pass itself are each
+        individually lock-protected against concurrent plugin_last_update
+        mutation (Vegas mode calls this from its own background
+        update-tick thread, racing the main render loop's plugin updates),
+        so callers get an atomic "who got fresh data" answer without
+        reaching into plugin_last_update themselves. The lock is not held
+        across the update pass so slow/blocking plugin update() calls don't
+        serialize against other plugin_last_update readers.
+        """
+        with self._plugin_last_update_lock:
+            old_times = dict(self.plugin_last_update)
+
+        self.run_scheduled_updates(current_time)
+
+        with self._plugin_last_update_lock:
+            return [
+                plugin_id for plugin_id, new_time in self.plugin_last_update.items()
+                if new_time > old_times.get(plugin_id, 0.0)
+            ]
+
     def update_all_plugins(self) -> None:
         """
         Update all enabled plugins.
@@ -797,7 +833,8 @@ class PluginManager:
             try:
                 success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                 if success:
-                    self.plugin_last_update[plugin_id] = time.time()
+                    with self._plugin_last_update_lock:
+                        self.plugin_last_update[plugin_id] = time.time()
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                 else:

src/plugin_system/store_manager.py

@@ -142,9 +142,28 @@ class PluginStoreManager:
         # then get the result from the warm cache (double-checked locking).
         self._registry_fetch_lock = threading.Lock()
 
+        # Per-plugin locks for _reinstall_with_rollback: the web UI runs
+        # Flask with threaded=True, so two overlapping requests for the
+        # same plugin_id (double-click, two browser tabs) would otherwise
+        # both rename the same directory aside — one succeeds, and the
+        # loser can end up renaming the winner's in-progress install aside
+        # mid-download, stealing its own rollback safety net. Keyed by
+        # plugin_id so unrelated plugins still update concurrently.
+        self._reinstall_locks: Dict[str, threading.Lock] = {}
+        self._reinstall_locks_guard = threading.Lock()
+
         # Ensure plugins directory exists
         self.plugins_dir.mkdir(exist_ok=True)
 
+    def _get_reinstall_lock(self, plugin_id: str) -> threading.Lock:
+        """Lazily create (or fetch) the per-plugin reinstall lock."""
+        with self._reinstall_locks_guard:
+            lock = self._reinstall_locks.get(plugin_id)
+            if lock is None:
+                lock = threading.Lock()
+                self._reinstall_locks[plugin_id] = lock
+            return lock
+
     def _record_cache_backoff(self, cache_dict: Dict, cache_key: str,
                               cache_timeout: int, payload: Any) -> None:
         """Bump a cache entry's timestamp so subsequent lookups hit the
@@ -2263,6 +2282,74 @@ class PluginStoreManager:
             self.logger.error(f"Error uninstalling plugin {plugin_id}: {e}")
             return False
     
+    def _reinstall_with_rollback(self, plugin_id: str, plugin_path: Path) -> bool:
+        """Replace an installed plugin with a fresh install, atomically.
+
+        The old install is renamed aside (not deleted) until the new install
+        succeeds, then removed; on ANY install failure the old directory is
+        restored. This is the difference between a failed update and a
+        destroyed plugin: the previous delete-then-install flow permanently
+        removed plugins whenever the download failed mid-update (seen in the
+        field during the monorepo migration on a Pi with broken DNS — every
+        old-remote plugin was deleted and none could be re-downloaded).
+
+        The aside name embeds '.standalone-backup-' so plugin discovery
+        (plugin_manager._scan_directory_for_plugins) ignores it even though
+        it still contains a manifest.json.
+
+        Held for the whole operation under a per-plugin_id lock: two
+        overlapping requests for the same plugin (double-click, two
+        browser tabs — the web UI runs Flask with threaded=True) must not
+        interleave their renames, or the second could steal the first's
+        rollback safety net mid-install. Other plugin_ids are unaffected.
+        """
+        with self._get_reinstall_lock(plugin_id):
+            backup_path = plugin_path.with_name(
+                f"{plugin_path.name}.standalone-backup-migrating")
+            # A stale aside from a previous crash would block the rename
+            if backup_path.exists():
+                if not self._safe_remove_directory(backup_path):
+                    self.logger.error(
+                        f"Could not clear stale backup for {plugin_id} at "
+                        f"{backup_path}; leaving old install in place")
+                    return False
+            try:
+                plugin_path.rename(backup_path)
+            except OSError as e:
+                self.logger.error(
+                    f"Could not set aside old plugin directory for {plugin_id}: {e}")
+                return False
+
+            try:
+                installed = self.install_plugin(plugin_id)
+            except Exception as e:
+                self.logger.error(f"Reinstall of {plugin_id} raised: {e}")
+                installed = False
+
+            if installed:
+                if not self._safe_remove_directory(backup_path):
+                    self.logger.warning(
+                        f"Update of {plugin_id} succeeded but the old backup "
+                        f"at {backup_path} could not be removed; it will be "
+                        f"cleared on the next update")
+                return True
+
+            # Install failed (bad network, registry error...) — put the old
+            # version back so the user still has a working plugin.
+            self.logger.error(
+                f"Reinstall of {plugin_id} failed; restoring previous version")
+            try:
+                if plugin_path.exists():
+                    # partial download debris from the failed install
+                    self._safe_remove_directory(plugin_path)
+                backup_path.rename(plugin_path)
+                self.logger.info(f"Restored previous install of {plugin_id}")
+            except OSError as e:
+                self.logger.error(
+                    f"CRITICAL: could not restore {plugin_id} from {backup_path}: {e}. "
+                    f"The previous install is preserved there — rename it back manually.")
+            return False
+
     def update_plugin(self, plugin_id: str) -> bool:
         """
         Update a plugin to the latest commit on its upstream branch.
@@ -2325,10 +2412,7 @@ class PluginStoreManager:
                             f"Plugin {resolved_id} git remote ({local_remote}) differs from registry ({registry_repo}). "
                             f"Reinstalling from registry to migrate to new source."
                         )
-                        if not self._safe_remove_directory(plugin_path):
+                        return self._reinstall_with_rollback(resolved_id, plugin_path)
-                            self.logger.error(f"Failed to remove old plugin directory for {resolved_id}")
-                            return False
-                        return self.install_plugin(resolved_id)
 
                     # Check if already up to date
                     if remote_sha and local_sha and remote_sha.startswith(local_sha):
@@ -2632,11 +2716,11 @@ class PluginStoreManager:
             # Plugin is not a git repo but is in registry and has a newer version - reinstall
             self.logger.info(f"Plugin {plugin_id} not installed via git; re-installing latest archive (registry id: {registry_id})")
 
-            # Remove directory and reinstall fresh
+            # Reinstall with the old version kept aside until the new
-            if not self._safe_remove_directory(plugin_path):
+            # download succeeds — this is the path every routine store
-                self.logger.error(f"Failed to remove old plugin directory for {plugin_id}")
+            # update takes, and a mid-update network failure must not
-                return False
+            # destroy the user's plugin.
-            return self.install_plugin(registry_id)
+            return self._reinstall_with_rollback(registry_id, plugin_path)
 
         except Exception as e:
             import traceback

src/vegas_mode/render_pipeline.py

@@ -66,6 +66,10 @@ class RenderPipeline:
             else display_manager.height
         )
 
+        # Reusable blank frame for cycle-end pushes (allocated lazily,
+        # re-blacked before each reuse)
+        self._blank_frame = None
+
         # ScrollHelper for optimized scrolling
         self.scroll_helper = ScrollHelper(
             self.display_width,
@@ -234,11 +238,19 @@ class RenderPipeline:
                     )
                     # Push blank immediately so the hardware never shows any
                     # post-wrap content while the coordinator recomposes the
-                    # next cycle (~100 ms).
+                    # next cycle (~100 ms). The blank is allocated once and
+                    # reused across cycle wraps (fresh paste each time in case
+                    # a consumer drew on the previous one).
                     try:
-                        from PIL import Image as _Image
+                        if self._blank_frame is None or self._blank_frame.size != (
-                        blank = _Image.new('RGB', (self.display_width, self.display_height))
+                                self.display_width, self.display_height):
-                        self.display_manager.image = blank
+                            self._blank_frame = Image.new(
+                                'RGB', (self.display_width, self.display_height))
+                        else:
+                            self._blank_frame.paste(
+                                (0, 0, 0),
+                                (0, 0, self.display_width, self.display_height))
+                        self.display_manager.image = self._blank_frame
                         self.display_manager.update_display()
                     except Exception:
                         logger.exception("Failed to write blank frame to display at cycle end")
@@ -297,6 +309,8 @@ class RenderPipeline:
         Returns True when:
         - Cycle is complete and we should start fresh
         - Staging buffer has new content
+        - A plugin currently visible in the scroll has pending updated data
+          (e.g. a live score changed) — standalone (non-sync) mode only
         """
         if self._cycle_complete:
             return True
@@ -314,6 +328,12 @@ class RenderPipeline:
         if buffer_status['staging_count'] > 0:
             return True
 
+        # Trigger recompose when pending updates affect visible segments, so
+        # live score/status changes reach the display within a few seconds
+        # instead of waiting for the next full cycle.
+        if self.stream_manager.has_pending_updates_for_visible_segments():
+            return True
+
         return False
 
     def hot_swap_content(self) -> bool:

test/conftest.py

@@ -38,7 +38,11 @@ def mock_cache_manager():
     mock._memory_cache_timestamps = {}
     mock.cache_dir = "/tmp/test_cache"
     
-    def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
+    def mock_get(key: str, max_age: Optional[int] = 300,
+                 memory_ttl: Optional[int] = None) -> Optional[Dict]:
+        # Signature mirrors CacheManager.get — keep in sync or callers
+        # passing keyword args (health tracker, resource monitor) break
+        # only in tests, hiding real-API compatibility.
         return mock._memory_cache.get(key)
     
     def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:

test/test_adaptive_images.py

@@ -194,3 +194,15 @@ class TestBasePluginDrawImage:
         assert ifit.height == 32
         # pasted onto the mock's canvas
         assert plugin.display_manager.image.getpixel((16, 16)) != (0, 0, 0)
+
+
+class TestResultIndependence:
+    def test_same_size_fit_never_aliases_the_source(self):
+        """LayoutContext caches ImageFitResults — an aliased image would let
+        later mutations of the source corrupt cached fits (or vice versa)."""
+        from PIL import ImageDraw
+        src = Image.new("RGBA", (20, 20), (255, 0, 0, 255))
+        fit = fit_image(src, (20, 20))
+        assert fit.image is not src
+        ImageDraw.Draw(src).rectangle([0, 0, 19, 19], fill=(0, 255, 0, 255))
+        assert fit.image.getpixel((5, 5)) == (255, 0, 0, 255)

test/test_adaptive_layout.py

@@ -436,3 +436,19 @@ class TestBasePluginIntegration:
                          MockCacheManager(), pm)
         assert plugin.layout.design_size == (64, 32)
         assert plugin.layout.scale == 2.0
+
+
+class TestFitCacheBound:
+    def test_fit_cache_is_lru_bounded(self, ctx):
+        """A plugin fitting changing text (live game clock, ticker) on a
+        24/7 service must not grow the fit cache without bound."""
+        for i in range(ctx._FIT_CACHE_MAX + 100):
+            ctx.fit_text(f"tick {i}", Region(0, 0, 100, 20))
+        assert len(ctx._fit_cache) <= ctx._FIT_CACHE_MAX
+
+    def test_lru_keeps_recent_entries_hot(self, ctx):
+        hot = ctx.fit_text("stay hot", Region(0, 0, 100, 20))
+        for i in range(ctx._FIT_CACHE_MAX - 1):
+            ctx.fit_text(f"cold {i}", Region(0, 0, 100, 20))
+            ctx.fit_text("stay hot", Region(0, 0, 100, 20))  # keep touching it
+        assert ctx.fit_text("stay hot", Region(0, 0, 100, 20)) is hot

test/test_cache_manager.py

@@ -400,3 +400,61 @@ class TestDiskCache:
         assert stats['fetch_count'] == 3
         assert stats['total_fetch_time'] == 1.8
         assert stats['average_fetch_time'] == pytest.approx(0.6, abs=0.01)
+
+
+class TestDiskCacheWriteEconomy:
+    """SD-card wear guards: identical payloads skip the disk, files are
+    compact, and TTL semantics survive the skip (see PR: fix/diskcache-sd-wear)."""
+
+    def test_identical_set_skips_rewrite(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        path = cache.get_cache_path("k")
+        first = os.stat(path)
+        os.utime(path, (first.st_atime - 100, first.st_mtime - 100))  # age it
+        aged_mtime = os.stat(path).st_mtime
+        ino_before = os.stat(path).st_ino
+        cache.set("k", {"data": "v"})  # identical payload
+        after = os.stat(path)
+        # mtime refreshed (TTL for mtime-based records preserved)...
+        assert after.st_mtime > aged_mtime
+        # ...but the file was NOT rewritten (same inode: no replace happened)
+        assert after.st_ino == ino_before
+
+    def test_changed_data_rewrites(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v1"})
+        cache.set("k", {"data": "v2"})
+        assert cache.get("k") == {"data": "v2"}
+
+    def test_clear_resets_digest(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        cache.clear("k")
+        assert cache.get("k") is None
+        cache.set("k", {"data": "v"})  # same payload after clear must WRITE
+        assert cache.get("k") == {"data": "v"}
+
+    def test_skip_self_heals_when_file_deleted_externally(self, tmp_path):
+        import os
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"data": "v"})
+        os.remove(cache.get_cache_path("k"))  # e.g. expiry cleanup
+        cache.set("k", {"data": "v"})  # digest matches but file is gone
+        assert cache.get("k") == {"data": "v"}
+
+    def test_files_are_compact_json(self, tmp_path):
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"a": 1, "b": [1, 2, 3]})
+        raw = open(cache.get_cache_path("k")).read()
+        assert "\n" not in raw.strip()  # no indent
+        assert cache.get("k") == {"a": 1, "b": [1, 2, 3]}
+
+    def test_datetime_round_trip_still_works(self, tmp_path):
+        from datetime import datetime
+        cache = DiskCache(cache_dir=str(tmp_path))
+        cache.set("k", {"when": datetime(2026, 7, 12, 10, 30)})
+        assert cache.get("k") == {"when": "2026-07-12T10:30:00"}

test/test_display_controller_vegas_tick.py

@@ -0,0 +1,88 @@
+"""
+Regression tests for DisplayController._tick_plugin_updates_for_vegas().
+
+PR #299 added logic to detect which plugins actually got fresh data on a
+scheduled-update tick and notify Vegas mode via
+vegas_coordinator.mark_plugin_updated(), so a live score change reaches the
+scroll within seconds instead of waiting for a full cycle. PR #330's
+multi-display sync refactor deleted this method (folding the callback back
+to the plain _tick_plugin_updates(), which reports nothing), silently
+orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
+callers since.
+"""
+
+from typing import Dict, List, Optional
+from unittest.mock import MagicMock
+
+from src.display_controller import DisplayController
+
+
+def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
+    dc = object.__new__(DisplayController)
+    dc.plugin_manager = MagicMock()
+    dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
+    dc.vegas_coordinator = vegas_coordinator
+    return dc
+
+
+class TestTickPluginUpdatesForVegas:
+    def test_marks_only_plugins_whose_timestamp_advanced(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_called_once_with("stock-news")
+
+    def test_no_advance_marks_nothing(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=[], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_not_called()
+
+    def test_no_vegas_coordinator_does_not_raise(self):
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+    def test_mark_plugin_updated_exception_does_not_propagate(self):
+        """One plugin's mark_plugin_updated failing must not stop the tick
+        or crash the update loop it runs in."""
+        vc = MagicMock()
+        vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
+        dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+        assert vc.mark_plugin_updated.call_count == 2
+
+
+class TestVegasCoordinatorCallbackWiring:
+    def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
+        """The Vegas coordinator must be given the Vegas-aware
+        _tick_plugin_updates_for_vegas as its update callback, not the plain
+        _tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
+        dc = object.__new__(DisplayController)
+        dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
+        dc.display_manager = MagicMock()
+        dc.plugin_manager = MagicMock()
+        dc.sync_manager = MagicMock()
+        dc._check_live_priority = MagicMock()
+        dc._check_vegas_interrupt = MagicMock(return_value=False)
+
+        fake_coordinator = MagicMock()
+
+        import src.display_controller as dc_module
+        original_imported = dc_module._vegas_mode_imported
+        original_class = dc_module.VegasModeCoordinator
+        try:
+            dc_module._vegas_mode_imported = True
+            dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
+            dc._initialize_vegas_mode()
+        finally:
+            dc_module._vegas_mode_imported = original_imported
+            dc_module.VegasModeCoordinator = original_class
+
+        fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)

test/test_display_dirty_tracking.py

@@ -0,0 +1,162 @@
+"""Tests for update_display dirty tracking (src/display_manager.py).
+
+Runs against RGBMatrixEmulator (EMULATOR=true), exercising the REAL
+DisplayManager — not a mock — so the skip logic, its invalidation hooks,
+and the kill switch are verified off-Pi.
+
+The invariants:
+- identical frames are pushed exactly once (SwapOnVSync not re-called)
+- ANY pixel change pushes
+- clear() and set_brightness() invalidate (the two paths that alter panel
+  state outside the digest's view)
+- the kill switch (display.dirty_tracking: false) restores always-push
+"""
+
+import os
+import sys
+import time
+
+os.environ["EMULATOR"] = "true"
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+
+@pytest.fixture(scope="module")
+def dm():
+    """One real DisplayManager on the emulator (it's a process singleton)."""
+    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)
+    yield manager
+    DisplayManager._instance = None
+    DisplayManager._initialized = False
+
+
+class _SwapSpy:
+    """Counts SwapOnVSync calls through the real matrix object."""
+
+    def __init__(self, matrix):
+        self.matrix = matrix
+        self.count = 0
+        self._orig = matrix.SwapOnVSync
+
+    def __enter__(self):
+        def counting(canvas):
+            self.count += 1
+            return self._orig(canvas)
+        self.matrix.SwapOnVSync = counting
+        return self
+
+    def __exit__(self, *exc):
+        self.matrix.SwapOnVSync = self._orig
+
+
+class TestDirtyTracking:
+    def test_identical_frames_push_once(self, dm):
+        dm.draw.rectangle([0, 0, 10, 10], fill=(255, 0, 0))
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()
+            dm.update_display()
+            dm.update_display()
+        assert spy.count == 1
+
+    def test_pixel_change_pushes(self, dm):
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.draw.point((5, 5), fill=(0, 255, 0))
+            dm.update_display()
+            dm.update_display()  # unchanged again
+        assert spy.count == 1
+
+    def test_clear_invalidates(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(0, 0, 255))
+        dm.update_display()
+        dm.clear()  # writes to the matrix directly; digest must reset
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # black frame after clear must still push
+        assert spy.count == 1
+
+    def test_brightness_change_forces_push(self, dm):
+        dm.draw.rectangle([0, 0, 20, 20], fill=(200, 200, 200))
+        dm.update_display()
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()          # identical -> skipped
+            assert spy.count == 0
+            dm.set_brightness(40)        # dim schedule scenario
+            dm.update_display()          # same image, new brightness -> push
+        assert spy.count == 1
+        dm.set_brightness(90)
+
+    def test_snapshot_still_written_on_skip(self, dm, tmp_path):
+        """The web preview mirror must keep working through skipped panel
+        pushes: _write_snapshot_if_due() still runs on the dirty-tracking
+        skip path and applies its own write/touch policy rather than being
+        bypassed entirely (see src/common/snapshot_policy.py — an unchanged
+        frame is touched, not re-encoded, once TOUCH_INTERVAL elapses)."""
+        dm._snapshot_path = str(tmp_path / "snap.png")
+        dm._last_snapshot_ts = 0.0
+        dm._last_snapshot_touch_ts = 0.0
+        dm._last_snapshot_digest = None
+        dm.draw.rectangle([0, 0, 30, 8], fill=(255, 255, 0))
+        dm.update_display()   # push + snapshot write (first frame)
+        assert os.path.exists(dm._snapshot_path)
+        first_mtime = os.path.getmtime(dm._snapshot_path)
+
+        # Age the write/touch bookkeeping past TOUCH_INTERVAL so the next
+        # identical frame is due for a touch, then push it again: dirty
+        # tracking must skip the panel write, but the snapshot mirror must
+        # still get its mtime bumped so the health check doesn't go stale.
+        from src.common import snapshot_policy
+        stale_ts = time.time() - snapshot_policy.TOUCH_INTERVAL - 1.0
+        dm._last_snapshot_ts = stale_ts
+        dm._last_snapshot_touch_ts = stale_ts
+        with _SwapSpy(dm.matrix) as spy:
+            dm.update_display()  # identical frame -> panel push skipped
+        assert spy.count == 0
+        assert os.path.getmtime(dm._snapshot_path) > first_mtime
+
+
+class TestKillSwitch:
+    def test_dirty_tracking_can_be_disabled(self, dm):
+        dm._dirty_tracking_enabled = False
+        try:
+            dm.draw.rectangle([0, 0, 10, 10], fill=(1, 2, 3))
+            with _SwapSpy(dm.matrix) as spy:
+                dm.update_display()
+                dm.update_display()
+                dm.update_display()
+            assert spy.count == 3  # always-push, exactly the old behavior
+        finally:
+            dm._dirty_tracking_enabled = True
+            dm._last_pushed_digest = None
+
+    def test_config_flag_wires_through(self):
+        from src.display_manager import DisplayManager
+        DisplayManager._instance = None
+        DisplayManager._initialized = False
+        try:
+            manager = DisplayManager({
+                "display": {
+                    "hardware": {"rows": 32, "cols": 64, "chain_length": 1,
+                                 "parallel": 1},
+                    "runtime": {"gpio_slowdown": 0},
+                    "dirty_tracking": False,
+                },
+            }, suppress_test_pattern=True)
+            assert manager._dirty_tracking_enabled is False
+        finally:
+            DisplayManager._instance = None
+            DisplayManager._initialized = False
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_snapshot_policy.py

@@ -0,0 +1,93 @@
+"""Tests for the snapshot write policy (src/common/snapshot_policy.py).
+
+The invariants that matter:
+- unchanged frames are NEVER re-encoded (the old code PNG-encoded identical
+  frames at 5 fps, 24/7)
+- the file mtime never goes stale enough to trip the health check's 60s
+  degraded threshold (api_v3 get_hardware_status)
+- a viewer gets full cadence; no viewer drops to the idle keepalive
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.common.snapshot_policy import (  # noqa: E402
+    IDLE_INTERVAL,
+    TOUCH_INTERVAL,
+    VIEWER_INTERVAL,
+    SnapshotAction,
+    decide,
+)
+
+
+class TestViewerCadence:
+    def test_changed_frame_with_viewer_writes_at_full_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+    def test_changed_frame_with_viewer_respects_min_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - VIEWER_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=True,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_unchanged_frame_with_viewer_never_writes(self):
+        """A static screen with a viewer must not burn PNG encodes."""
+        assert decide(now=100.0, last_write_ts=90.0, last_touch_ts=90.0,
+                      viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+
+class TestIdleCadence:
+    def test_changed_frame_without_viewer_waits_for_idle_interval(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL / 2,
+                      last_touch_ts=100.0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.SKIP
+
+    def test_changed_frame_without_viewer_writes_at_idle_rate(self):
+        assert decide(now=100.0, last_write_ts=100.0 - IDLE_INTERVAL,
+                      last_touch_ts=0, viewer_fresh=False,
+                      frame_changed=True) is SnapshotAction.WRITE
+
+
+class TestHealthKeepalive:
+    def test_stale_mtime_gets_touched(self):
+        """Whatever else happens, mtime must be bumped within TOUCH_INTERVAL
+        so the health check (60s threshold) never reads the display as dead."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_touch_applies_with_viewer_too(self):
+        """Viewer watching a static screen: no writes, but health stays green."""
+        assert decide(now=100.0, last_write_ts=100.0 - TOUCH_INTERVAL - 1,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL - 1, viewer_fresh=True,
+                      frame_changed=False) is SnapshotAction.TOUCH
+
+    def test_recent_touch_suppresses_another(self):
+        assert decide(now=100.0, last_write_ts=0.0,
+                      last_touch_ts=100.0 - TOUCH_INTERVAL / 2, viewer_fresh=False,
+                      frame_changed=False) is SnapshotAction.SKIP
+
+    def test_touch_interval_stays_under_health_threshold(self):
+        """api_v3's hardware status treats snapshot age >= 60s as degraded.
+        Keep a 2x margin so scheduling jitter can't trip it."""
+        assert TOUCH_INTERVAL <= 30
+
+    def test_worst_case_mtime_age_is_bounded(self):
+        """Simulate any interleaving: from any state, within one policy call
+        after TOUCH_INTERVAL elapses, mtime gets refreshed (WRITE or TOUCH)."""
+        for viewer in (True, False):
+            for changed in (True, False):
+                action = decide(now=1000.0, last_write_ts=900.0,
+                                last_touch_ts=900.0, viewer_fresh=viewer,
+                                frame_changed=changed)
+                assert action in (SnapshotAction.WRITE, SnapshotAction.TOUCH)
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_store_update_rollback.py

@@ -0,0 +1,171 @@
+"""Tests for atomic plugin updates (store_manager._reinstall_with_rollback).
+
+Regression for a field data-loss incident: update_plugin's reinstall paths
+(monorepo migration AND routine archive updates) deleted the installed
+plugin BEFORE downloading its replacement — a mid-update network failure
+permanently destroyed the plugin. Seen live: a Pi with broken DNS lost 12
+plugins from one update pass.
+"""
+
+import json
+import os
+import sys
+import threading
+import time
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from src.plugin_system.store_manager import PluginStoreManager  # noqa: E402
+
+PLUGIN_ID = "rollback-test-plugin"
+
+
+@pytest.fixture
+def store(tmp_path):
+    mgr = PluginStoreManager(plugins_dir=str(tmp_path))
+    plugin_dir = tmp_path / PLUGIN_ID
+    plugin_dir.mkdir()
+    (plugin_dir / "manifest.json").write_text(json.dumps(
+        {"id": PLUGIN_ID, "name": "Rollback Test", "version": "1.0.0"}))
+    (plugin_dir / "manager.py").write_text("# old version marker\n")
+    return mgr, plugin_dir
+
+
+class TestReinstallWithRollback:
+    def test_failed_install_restores_old_version(self, store):
+        """The whole point: a failed download must leave the old install."""
+        mgr, plugin_dir = store
+        with patch.object(mgr, "install_plugin", return_value=False):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+        # no aside debris left behind
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_install_exception_restores_old_version(self, store):
+        mgr, plugin_dir = store
+        with patch.object(mgr, "install_plugin",
+                          side_effect=RuntimeError("network down")):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+
+    def test_successful_install_removes_aside(self, store):
+        mgr, plugin_dir = store
+
+        def fake_install(plugin_id):
+            new_dir = plugin_dir  # same path, new content
+            new_dir.mkdir(exist_ok=True)
+            (new_dir / "manager.py").write_text("# new version\n")
+            (new_dir / "manifest.json").write_text(json.dumps(
+                {"id": PLUGIN_ID, "name": "Rollback Test", "version": "2.0.0"}))
+            return True
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_install):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is True
+        assert "new version" in (plugin_dir / "manager.py").read_text()
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_partial_download_debris_is_replaced_by_old_version(self, store):
+        """A failed install that left a partial directory must still roll back."""
+        mgr, plugin_dir = store
+
+        def fake_partial_install(plugin_id):
+            plugin_dir.mkdir(exist_ok=True)
+            (plugin_dir / "half-downloaded.tmp").write_text("junk")
+            return False
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_partial_install):
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        assert ok is False
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+        assert not (plugin_dir / "half-downloaded.tmp").exists()
+
+    def test_stale_aside_from_previous_crash_is_cleared(self, store):
+        mgr, plugin_dir = store
+        stale = plugin_dir.parent / f"{PLUGIN_ID}.standalone-backup-migrating"
+        stale.mkdir()
+        (stale / "old.txt").write_text("stale")
+        with patch.object(mgr, "install_plugin", return_value=False) as mock_install:
+            ok = mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir)
+        # The reinstall itself still fails (mocked) and the old install is
+        # restored, but the stale aside must not have survived — otherwise
+        # it would have blocked this run's own rename (or a future one).
+        assert not stale.exists()
+        mock_install.assert_called_once_with(PLUGIN_ID)
+        assert ok is False
+        assert plugin_dir.exists()
+        assert "old version marker" in (plugin_dir / "manager.py").read_text()
+
+    def test_concurrent_updates_for_same_plugin_are_serialized(self, store):
+        """Two overlapping requests for the same plugin_id (double-click,
+        two browser tabs — the web UI runs Flask with threaded=True) must
+        not interleave: the loser must wait for the winner to finish
+        rather than renaming the winner's in-progress install aside and
+        stealing its rollback safety net."""
+        mgr, plugin_dir = store
+
+        active = 0
+        max_active = 0
+        guard = threading.Lock()
+
+        def fake_install(plugin_id):
+            nonlocal active, max_active
+            with guard:
+                active += 1
+                max_active = max(max_active, active)
+            time.sleep(0.05)
+            plugin_dir.mkdir(exist_ok=True)
+            (plugin_dir / "manager.py").write_text("# new version\n")
+            (plugin_dir / "manifest.json").write_text(json.dumps(
+                {"id": PLUGIN_ID, "name": "Rollback Test", "version": "2.0.0"}))
+            with guard:
+                active -= 1
+            return True
+
+        results = []
+
+        def worker():
+            results.append(mgr._reinstall_with_rollback(PLUGIN_ID, plugin_dir))
+
+        with patch.object(mgr, "install_plugin", side_effect=fake_install):
+            threads = [threading.Thread(target=worker) for _ in range(2)]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join(timeout=5)
+
+        assert max_active == 1, "install_plugin ran concurrently for the same plugin_id"
+        assert results == [True, True]
+        assert plugin_dir.exists()
+        assert "new version" in (plugin_dir / "manager.py").read_text()
+        leftovers = [p for p in plugin_dir.parent.iterdir()
+                     if "standalone-backup" in p.name]
+        assert leftovers == []
+
+    def test_aside_name_is_invisible_to_discovery(self, store, tmp_path):
+        """The aside still contains a manifest.json — discovery must skip it
+        (relies on the existing '.standalone-backup-' exclusion)."""
+        mgr, plugin_dir = store
+        from src.plugin_system.plugin_manager import PluginManager
+        aside = plugin_dir.parent / f"{PLUGIN_ID}.standalone-backup-migrating"
+        plugin_dir.rename(aside)
+        pm = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                           display_manager=None, cache_manager=None)
+        found = pm._scan_directory_for_plugins(Path(tmp_path))
+        assert PLUGIN_ID not in found
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-v"]))

test/test_vegas_render_pipeline_recompose.py

@@ -0,0 +1,72 @@
+"""
+Regression tests for RenderPipeline.should_recompose()'s pending-updates check.
+
+PR #299 added a check so a plugin's live score/status change (a "pending
+update" in StreamManager) triggers a hot-swap within a few seconds instead
+of waiting for a full scroll cycle to complete. PR #330 (multi-display sync)
+refactored should_recompose() and dropped that check entirely -- not just
+gated behind the new sync-mode deferral it added, but removed outright, so
+even standalone (non-sync) installations silently lost live-refresh and fell
+back to waiting for full cycle boundaries (which, depending on
+min/max_cycle_duration, can be minutes).
+"""
+
+from unittest.mock import MagicMock
+
+from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.render_pipeline import RenderPipeline
+
+
+class FakeDisplayManager:
+    width = 64
+    height = 32
+
+
+def _make_pipeline(sync_manager=None):
+    stream_manager = MagicMock()
+    stream_manager.get_buffer_status.return_value = {'staging_count': 0}
+    pipeline = RenderPipeline(VegasModeConfig(), FakeDisplayManager(), stream_manager)
+    pipeline.sync_manager = sync_manager
+    return pipeline, stream_manager
+
+
+class TestShouldRecompose:
+    def test_cycle_complete_always_recomposes(self):
+        pipeline, stream_manager = _make_pipeline()
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_no_pending_updates_no_staging_does_not_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is False
+
+    def test_pending_updates_on_visible_segment_triggers_recompose(self):
+        """The actual regression: a live-updated plugin currently in view
+        must trigger a recompose instead of waiting for cycle end."""
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True
+
+    def test_staging_buffer_content_triggers_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.get_buffer_status.return_value = {'staging_count': 1}
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_sync_active_defers_pending_updates_to_cycle_boundary(self):
+        """Sync-mode deferral (PR #330's actual intent) must still hold:
+        pending updates alone must NOT trigger a mid-cycle hot-swap when a
+        follower display is attached, since that causes a visible
+        freeze+jump on the follower. This must keep working after
+        restoring the non-sync pending-updates check above."""
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is False
+
+    def test_sync_active_still_recomposes_on_cycle_complete(self):
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True

web_interface/app.py

@@ -610,8 +610,22 @@ def display_preview_generator():
     import io
 
     snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path matches display_manager; only read here
+    # Viewer marker: this generator only runs while the broadcaster has
+    # subscribers (it exits with no clients), so touching the marker each
+    # loop tells the DISPLAY service a browser is actually watching — it
+    # only pays for full-rate PNG snapshot encodes while this stays fresh
+    # (see src/common/snapshot_policy.py).
+    viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - fixed path matches display_manager
     last_modified = None
 
+    def _touch_viewer_marker():
+        try:
+            with open(viewer_marker_path, 'a'):
+                pass
+            os.utime(viewer_marker_path, None)
+        except OSError:
+            pass  # display side treats a missing marker as "no viewer"
+    
     # Get display dimensions from config
     try:
         main_config = config_manager.load_config()
@@ -627,6 +641,7 @@ def display_preview_generator():
     
     while True:
         try:
+            _touch_viewer_marker()
             # Check if snapshot file exists and has been modified
             if os.path.exists(snapshot_path):
                 current_modified = os.path.getmtime(snapshot_path)

web_interface/blueprints/api_v3.py

@@ -329,6 +329,7 @@ def save_schedule_config():
         }
 
         mode = data.get('mode', 'global')
+        schedule_config['mode'] = mode
 
         if mode == 'global':
             # Simple global schedule