perf(display): dirty tracking in update_display + plugin FPS declaration (#406)

* perf(display): dirty tracking in update_display + plugin FPS declaration

update_display now skips SetImage+SwapOnVSync when the frame is
byte-identical to the last pushed one (adler32 digest) AND brightness
is unchanged — brightness is part of the digest, and set_brightness
additionally resets it, so a dim-schedule change can never be skipped.
clear() resets the digest (it writes to the matrix directly). Skipping
a swap is hardware-safe: the panel refreshes the current frame from the
driver's own thread; swaps only change content.

Kill switch: display.dirty_tracking: false restores always-push.

display_controller's high-FPS decision gains a precedence step: a
plugin exposing needs_high_fps is honored first (so static-image can
declare False for still PNGs and stop burning a 125fps loop on them);
static-image without the attribute keeps its historical forced
high-FPS (GIF back-compat); scrolling logic is otherwise unchanged.

Verified with 7 tests against the real DisplayManager on
RGBMatrixEmulator (identical-frame skip, pixel-change push, clear and
brightness invalidation, snapshot-through-skip, kill switch) plus the
202-test display/controller/vegas suites, and a clean devpi deploy.
Audit: every SetImage/SwapOnVSync/Clear/brightness call site is inside
display_manager — no external writer can bypass the digest.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FqzC1nzTWL4kaqgMaQZFam

* fix(display): serialize update_display, narrow brightness exception, log fixes

CodeRabbit review on #406, verified against current code:

- update_display() can genuinely be called from background threads (some
  sports base classes call it directly from inside update() for an
  immediate "live" refresh), not just the render loop — confirmed via the
  existing follower-mode gating wrapper in display_controller.py, which
  exists specifically because "background plugin threads" can reach it.
  Without a lock, two callers could both pass the digest check before
  either writes _last_pushed_digest back, causing a redundant push, or
  interleave the offscreen/current canvas swap. Added self._update_lock
  (RLock, in case of re-entrant callers) around the full method body so
  every call site is automatically covered — no caller changes needed.
  (No prior lock existed to reuse on DisplayManager; this adds one.)
- Narrowed the brightness-read exception handler to AttributeError,
  matching the established pattern in get_brightness()/set_brightness()
  — a getattr() with a default already swallows AttributeError, so the
  only case this guards is the property getter itself raising, and the
  established pattern treats that as an expected, specific failure mode
  rather than something to blanket-catch.
- FPS-check debug log now includes the plugin_id already in scope
  (previously only active_mode) and a "[DisplayController]" prefix for
  grep-ability, matching the sibling log two lines below it.
- test_display_dirty_tracking.py: dm fixture and test_config_flag_wires_through
  now reset the DisplayManager singleton on teardown, matching the pattern
  test_display_manager.py already uses elsewhere in the same file family.
- test_snapshot_still_written_on_skip previously only exercised the
  non-skip (push) path despite its name; now performs a second update that
  meets the skip conditions (identical frame) and asserts the snapshot is
  still written even though the panel push itself is skipped.

All 7 dirty-tracking tests pass, plus the full display_manager/
display_controller/vegas suite (140 passed). Full repo suite has only the
5 known pre-existing failures (double-sided config x2, state_reconciliation
x2, and test_circuit_breaker's conftest.py mock signature drift — the
latter fixed in #400, which this branch's base predates).

---------

Co-authored-by: Chuck <chuck@example.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

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]:
         """
@@ -155,10 +161,35 @@ class DiskCache:
         cache_path = self.get_cache_path(key)
         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.flush()
-                                os.fsync(tmp_file.fileno())
+                                tmp_file.write(payload)
                             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.flush()
-                                os.fsync(cache_file.fileno())
+                                cache_file.write(payload)
+                            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
@@ -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.
@@ -1684,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
@@ -1852,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
@@ -1910,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:
@@ -2121,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)
-                        # Always enable high-FPS for static-image plugin (for GIF animation support)
+                        # High-FPS decision, in precedence order:
+                        # 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:
@@ -2403,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
@@ -2453,13 +2480,14 @@ class DisplayController:
                     pass
                 return None
             
-            # Message is valid and not expired
-            return {
+            # Message is valid and not expired — cache for the throttle window
+            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] = {}
-        # Snapshot settings for web preview integration (service writes, web reads)
+        self._text_width_cache: "OrderedDict[tuple, Tuple[int, Any]]" = OrderedDict()
+        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:
-                # Fallback mode - no actual hardware to update
-                logger.debug("Update display called in fallback mode (no hardware)")
-                # Still write a snapshot so the web UI can preview
+            with self._update_lock:
+                if self.matrix is None:
+                    # Fallback mode - no actual hardware to update
+                    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
-            from pathlib import Path
-            from src.common.permission_utils import (
-                ensure_directory_permissions,
-                ensure_file_permissions,
-                get_assets_dir_mode,
-                get_assets_file_mode
-            )
+            if action is snapshot_policy.SnapshotAction.TOUCH:
+                # mtime bump only: keeps the health check (snapshot age)
+                # green without paying for a PNG encode of an unchanged frame
+                os.utime(self._snapshot_path, None)
+                self._last_snapshot_touch_ts = now
+                return
+
+            # WRITE: ensure directory permissions once, not per frame
             snapshot_path_obj = Path(self._snapshot_path)
-            # Only ensure permissions on non-system directories
-            # Never modify /tmp permissions - it has special system permissions (1777)
-            # that must not be changed or it breaks apt and other system tools
-            parent_dir = snapshot_path_obj.parent
-            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())
+            if not self._snapshot_dir_prepared:
+                # Never modify /tmp permissions - it has special system
+                # permissions (1777) that must not be changed or it breaks
+                # apt and other system tools
+                parent_dir = snapshot_path_obj.parent
+                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
-            logger.debug(f"Snapshot write skipped: {e}")
+            # Snapshot failures must never break display — but they must not
+            # 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:
-            return self.metrics_cache[cache_key]
+        cached = self.metrics_cache.get(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/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):
-                            self.logger.error(f"Failed to remove old plugin directory for {resolved_id}")
-                            return False
-                        return self.install_plugin(resolved_id)
+                        return self._reinstall_with_rollback(resolved_id, plugin_path)
 
                     # 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
-            if not self._safe_remove_directory(plugin_path):
-                self.logger.error(f"Failed to remove old plugin directory for {plugin_id}")
-                return False
-            return self.install_plugin(registry_id)
+            # Reinstall with the old version kept aside until the new
+            # download succeeds — this is the path every routine store
+            # update takes, and a mid-update network failure must not
+            # destroy the user's plugin.
+            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
-                        blank = _Image.new('RGB', (self.display_width, self.display_height))
-                        self.display_manager.image = blank
+                        if self._blank_frame is None or self._blank_frame.size != (
+                                self.display_width, self.display_height):
+                            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")

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_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_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"]))

web_interface/app.py

@@ -608,9 +608,23 @@ def display_preview_generator():
     import base64
     from PIL import Image
     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:
@@ -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)