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

# Conflicts:
#	src/display_manager.py

scripts/dev_server.py

@@ -200,16 +200,62 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
                  skip_update):
     """Render one plugin at one size. Returns the /api/render response dict.
 
-    Thin wrapper over the shared render service (also used by the web UI's
-    config-page preview); a fresh plugin instance per call, mirroring the
-    safety harness, so sizes never share state.
+    A fresh plugin instance per call, mirroring the safety harness, so sizes
+    never share state.
     """
-    from src.plugin_system.testing.render_service import render_plugin_once
+    from src.plugin_system.testing import VisualTestDisplayManager, MockCacheManager, MockPluginManager
+    from src.plugin_system.plugin_loader import PluginLoader
 
-    return render_plugin_once(
-        plugin_id, plugin_dir, manifest=manifest, config=config,
-        mock_data=mock_data, width=width, height=height,
-        skip_update=skip_update)
+    display_manager = VisualTestDisplayManager(width=width, height=height)
+    cache_manager = MockCacheManager()
+    plugin_manager = MockPluginManager()
+
+    # Pre-populate cache with mock data
+    for key, value in mock_data.items():
+        cache_manager.set(key, value)
+
+    loader = PluginLoader()
+    errors = []
+    warnings = []
+
+    plugin_instance, _module = loader.load_plugin(
+        plugin_id=plugin_id,
+        manifest=manifest,
+        plugin_dir=plugin_dir,
+        config=config,
+        display_manager=display_manager,
+        cache_manager=cache_manager,
+        plugin_manager=plugin_manager,
+        install_deps=False,
+    )
+
+    start_time = time.time()
+
+    # Run update()
+    if not skip_update:
+        try:
+            plugin_instance.update()
+        except Exception as 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:
+        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)
+
+    return {
+        'image': f'data:image/png;base64,{display_manager.get_image_base64()}',
+        'width': width,
+        'height': height,
+        'render_time_ms': render_time_ms,
+        'errors': errors,
+        'warnings': warnings,
+    }
 
 
 def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:

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.
@@ -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")
@@ -621,18 +628,28 @@ class DisplayController:
 
         current_day = current_time.strftime('%A').lower()  # e.g. 'monday'
         current_time_only = current_time.time()
-        
+
         # 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:
-            # Check if days dict is not empty and contains current day
-            if days_config and current_day in days_config:
+        if mode_normalized == 'global':
+            use_per_day = False
+        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:
-                # Days dict exists but doesn't have current day - fall back to global
+            else:
                 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)
-                        # 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:
@@ -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
-            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/element_style.py

@@ -1,552 +0,0 @@
-"""Universal per-element style resolution for plugin customization.
-
-Plugins expose per-element user customization under ``config['customization']``:
-
-    "customization": {
-        "score_text": {"font": "PressStart2P-Regular.ttf", "font_size": 10,
-                       "text_color": [255, 255, 255]},
-        "layout": {"score": {"x_offset": 2, "y_offset": 0}}
-    }
-
-Before this module, every plugin re-implemented the same three pieces —
-a font loader, an x/y-offset reader, and (for adaptive layout mode) a
-"did the user actually override this?" check. The loaders diverged four
-ways across the sports plugins and music, the offset reader was copied
-twice, and the override check is subtle enough that it shipped broken
-twice: the web UI's save flow (schema_manager.merge_with_defaults) writes
-the FULL schema default object into config.json on every save, and the
-plugin manager merges defaults into ``config`` again before instantiation,
-so a key being *present* never means the user set it. The only correct
-test is "present AND different from the schema default", which requires
-knowing the schema defaults — previously a hand-maintained dict per plugin.
-
-This module is that logic, once:
-
-    resolver = ElementStyleResolver(config, schema_defaults)
-    style = resolver.style('score_text', classic_font='PressStart2P-Regular.ttf',
-                           classic_size=10)
-    style.font          # loaded PIL font, ready for draw.text
-    style.user_forced   # True only for a genuine user override
-    dx, dy = resolver.offset('score')
-
-``BasePlugin.element_style()`` wires this up automatically (schema defaults
-come from the plugin's own config_schema.json via the schema manager).
-Standalone helper classes (e.g. a plugin's GameRenderer) should receive a
-resolver from their owning plugin rather than build one themselves.
-
-Deliberately pure PIL + stdlib: no imports from the plugin system or web
-layer, so it is usable from any renderer and trivially testable.
-"""
-
-import json
-import logging
-import os
-from dataclasses import dataclass
-from typing import Any, Dict, Optional, Tuple, Union
-
-from PIL import ImageFont
-
-logger = logging.getLogger(__name__)
-
-
-# Font-family aliases accepted in customization configs. Filenames pass
-# through unchanged. (Supersedes the per-plugin copies in the baseball
-# plugin; keep names in sync with the web UI's /fonts/catalog so the
-# font-selector widget and this loader agree.)
-FONT_ALIASES: Dict[str, str] = {
-    "press_start": "PressStart2P-Regular.ttf",
-    "four_by_six": "4x6-font.ttf",
-    "five_by_seven": "5x7.bdf",
-}
-
-DEFAULT_FONTS_DIR = os.path.join("assets", "fonts")
-DEFAULT_FALLBACK_FONT = "PressStart2P-Regular.ttf"
-
-PILFont = Union[ImageFont.FreeTypeFont, ImageFont.ImageFont]
-
-
-def resolve_font_name(font_name: str) -> str:
-    """Resolve a font family alias to its filename, leaving filenames as-is."""
-    return FONT_ALIASES.get(font_name, font_name)
-
-
-def extract_schema_defaults(schema: Dict[str, Any]) -> Dict[str, Any]:
-    """Nested defaults dict from a JSON Schema (mirrors
-    SchemaManager.extract_defaults_from_schema, kept here so this module
-    stays importable without the plugin system — the parity test in
-    test_element_style.py guards against drift).
-
-    An object property carrying its own ``default`` short-circuits recursion,
-    and array-typed properties without one default to ``[]`` (or a
-    single-item list when the items schema declares a default), matching
-    the schema manager's behavior exactly.
-    """
-    defaults: Dict[str, Any] = {}
-    for key, prop in (schema.get("properties") or {}).items():
-        if not isinstance(prop, dict):
-            continue
-        if "default" in prop:
-            defaults[key] = prop["default"]
-        elif prop.get("type") == "object" and "properties" in prop:
-            nested = extract_schema_defaults(prop)
-            if nested:
-                defaults[key] = nested
-        elif prop.get("type") == "array" and "items" in prop:
-            items = prop.get("items")
-            if isinstance(items, dict) and "default" in items and \
-                    not (items.get("type") == "object" and "properties" in items):
-                defaults[key] = [items["default"]]
-            else:
-                defaults[key] = []
-    return defaults
-
-
-def defaults_from_schema_file(schema_path: str) -> Dict[str, Any]:
-    """Schema defaults straight from a plugin's own config_schema.json.
-
-    Plugins that hand a resolver to standalone helper classes should build
-    it with this, pointed at their own schema file — it works identically
-    in production, the test harness, and the dev server, unlike the plugin
-    manager's schema manager (absent under mocks). x-style-elements
-    declarations are expanded first, so declared elements' defaults are
-    included exactly as the web UI's schema manager sees them. Returns {}
-    on any error.
-    """
-    try:
-        with open(schema_path, "r", encoding="utf-8") as f:
-            schema = json.load(f)
-        if isinstance(schema, dict):
-            return extract_schema_defaults(expand_style_elements(schema))
-    except Exception as e:
-        logger.debug("Could not load schema defaults from %s: %s",
-                     schema_path, e)
-    return {}
-
-
-# ---------------------------------------------------------------------------
-# x-style-elements schema expansion
-# ---------------------------------------------------------------------------
-#
-# A plugin declares its styleable display elements ONCE, compactly, on its
-# customization object instead of hand-copying ~50-line property blocks:
-#
-#     "customization": {
-#         "type": "object",
-#         "x-style-elements": {
-#             "score_text": {
-#                 "title": "Game Score",
-#                 "font": {"default": "PressStart2P-Regular.ttf"},
-#                 "size": {"default": 10, "min": 4, "max": 16},
-#                 "color": true,                  # or {"default": [r,g,b]}
-#                 "offsets": true
-#             }
-#         }
-#     }
-#
-# expand_style_elements() turns each declaration into full font/font_size/
-# text_color/layout-offset property blocks (marked "x-style-managed": true)
-# using widgets the web config form already renders. The declaration stays
-# in the schema — it doubles as the element registry for tooling. Expansion
-# is idempotent, and a hand-written property block for the same element
-# always wins over the generated one.
-#
-# SchemaManager.load_schema() applies this at serve time (so the web form,
-# save path, validation, and defaults generation all see the expanded
-# shape), and defaults_from_schema_file() applies it when plugins read
-# their own schema — one implementation, no drift.
-
-def get_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
-    """The x-style-elements declaration from a schema ({} if none)."""
-    try:
-        decl = schema.get("properties", {}).get("customization", {}).get("x-style-elements")
-        return decl if isinstance(decl, dict) else {}
-    except AttributeError:
-        return {}
-
-
-def expand_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
-    """Expand x-style-elements into full customization property blocks.
-
-    Returns the schema unchanged (same object) when there is nothing to
-    expand; otherwise returns an expanded DEEP COPY, leaving the input
-    untouched. Never raises — on any error the original schema is returned
-    so a malformed declaration can't take a plugin down.
-    """
-    import copy
-
-    try:
-        declarations = get_style_elements(schema)
-        if not declarations:
-            return schema
-
-        schema = copy.deepcopy(schema)
-        customization = schema["properties"]["customization"]
-        properties = customization.setdefault("properties", {})
-        order = customization.get("x-propertyOrder")
-
-        offset_elements = []
-        for element_key, declaration in declarations.items():
-            if not isinstance(declaration, dict):
-                continue
-            if declaration.get("offsets") is True:
-                offset_elements.append((element_key, declaration))
-            if element_key in properties:
-                # Hand-written (or previously expanded) block wins.
-                continue
-            properties[element_key] = _style_element_block(element_key, declaration)
-            if isinstance(order, list) and element_key not in order:
-                # Keep generated elements ahead of the layout section.
-                insert_at = order.index("layout") if "layout" in order else len(order)
-                order.insert(insert_at, element_key)
-
-        if offset_elements:
-            _expand_offset_blocks(properties, order, offset_elements)
-
-        return schema
-    except Exception as e:
-        logger.error("x-style-elements expansion failed: %s", e)
-        return schema
-
-
-def _style_element_block(element_key: str, declaration: Dict[str, Any]) -> Dict[str, Any]:
-    """One generated customization.<element> property block."""
-    title = declaration.get("title") or element_key.replace("_", " ").title()
-    font_decl = declaration.get("font") if isinstance(declaration.get("font"), dict) else {}
-    size_decl = declaration.get("size") if isinstance(declaration.get("size"), dict) else {}
-
-    block_properties: Dict[str, Any] = {
-        "font": {
-            "type": "string",
-            "title": "Font Family",
-            "description": "Select the font to use",
-            "x-widget": "font-selector",
-            "default": font_decl.get("default", DEFAULT_FALLBACK_FONT),
-        },
-        "font_size": {
-            "type": "integer",
-            "title": "Font Size",
-            "description": ("Font size in pixels (BDF fonts are fixed-size "
-                            "and ignore this)"),
-            "minimum": size_decl.get("min", 4),
-            "maximum": size_decl.get("max", 32),
-            "default": size_decl.get("default", 8),
-        },
-    }
-    block_order = ["font", "font_size"]
-
-    color_decl = declaration.get("color")
-    if color_decl:
-        default_color = [255, 255, 255]
-        if isinstance(color_decl, dict) and isinstance(color_decl.get("default"), list):
-            default_color = color_decl["default"]
-        # The default doubles as the "untouched" sentinel: the resolver only
-        # honors a color that DIFFERS from it, so untouched saves (the web
-        # form always posts the RGB inputs) can't clobber a plugin's
-        # semantic/state-dependent colors.
-        block_properties["text_color"] = {
-            "type": "array",
-            "title": "Text Color",
-            "description": "RGB color as [red, green, blue] (0-255 each)",
-            "items": {"type": "integer", "minimum": 0, "maximum": 255},
-            "minItems": 3,
-            "maxItems": 3,
-            "x-widget": "color-picker",
-            "default": default_color,
-        }
-        block_order.append("text_color")
-
-    return {
-        "type": "object",
-        "title": title,
-        "description": f"Style settings for {title}",
-        "x-style-managed": True,
-        "properties": block_properties,
-        "x-propertyOrder": block_order,
-        "additionalProperties": False,
-    }
-
-
-def _expand_offset_blocks(properties: Dict[str, Any], order,
-                          offset_elements) -> None:
-    """Generate customization.layout.<element> x/y offset blocks."""
-    layout = properties.get("layout")
-    if not isinstance(layout, dict):
-        layout = {
-            "type": "object",
-            "title": "Layout Positioning",
-            "description": ("Adjust X,Y coordinate offsets for elements. "
-                            "Values are relative to default positions; "
-                            "negative moves left/up, positive right/down."),
-            "x-style-managed": True,
-            "properties": {},
-            "additionalProperties": False,
-        }
-        properties["layout"] = layout
-        if isinstance(order, list) and "layout" not in order:
-            order.append("layout")
-
-    layout_properties = layout.setdefault("properties", {})
-    layout_order = layout.get("x-propertyOrder")
-    for element_key, declaration in offset_elements:
-        if element_key in layout_properties:
-            continue  # hand-written layout entry wins
-        title = declaration.get("title") or element_key.replace("_", " ").title()
-        layout_properties[element_key] = {
-            "type": "object",
-            "title": title,
-            "x-style-managed": True,
-            "properties": {
-                "x_offset": {
-                    "type": "integer",
-                    "title": "X Offset",
-                    "description": "Horizontal offset in pixels (default: 0)",
-                    "default": 0,
-                },
-                "y_offset": {
-                    "type": "integer",
-                    "title": "Y Offset",
-                    "description": "Vertical offset in pixels (default: 0)",
-                    "default": 0,
-                },
-            },
-            "additionalProperties": False,
-        }
-        if isinstance(layout_order, list) and element_key not in layout_order:
-            layout_order.append(element_key)
-
-
-def load_font(font_name: str, size: int, *,
-              fonts_dir: str = DEFAULT_FONTS_DIR,
-              fallback_font: str = DEFAULT_FALLBACK_FONT) -> PILFont:
-    """Load a font by name at a pixel size, never raising.
-
-    Resolution order:
-      1. alias -> filename (``FONT_ALIASES``)
-      2. ``ImageFont.truetype`` — handles .ttf/.otf, and .bdf too (FreeType
-         loads BDF strikes at their native size; a non-native size raises
-         "invalid pixel size" and falls through)
-      3. for .bdf: a pre-converted ``.pil`` sidecar via ``ImageFont.load``
-      4. ``fallback_font`` at the requested size
-      5. ``ImageFont.load_default()``
-    """
-    font_name = resolve_font_name(font_name or "")
-    font_path = os.path.join(fonts_dir, font_name)
-    lower = font_name.lower()
-
-    if os.path.exists(font_path):
-        try:
-            return ImageFont.truetype(font_path, size)
-        except Exception as e:
-            logger.debug("truetype failed for %s@%s: %s", font_name, size, e)
-        if lower.endswith(".bdf"):
-            pil_path = font_path.rsplit(".", 1)[0] + ".pil"
-            if os.path.exists(pil_path):
-                try:
-                    return ImageFont.load(pil_path)
-                except Exception as e:
-                    logger.debug("PIL sidecar failed for %s: %s", pil_path, e)
-            logger.warning(
-                "BDF font %s could not be loaded at size %s (BDF fonts are "
-                "fixed-size; font_size must match the native size). Falling "
-                "back to %s.", font_name, size, fallback_font)
-    else:
-        logger.warning("Font file not found: %s, falling back to %s",
-                       font_path, fallback_font)
-
-    fallback_path = os.path.join(fonts_dir, resolve_font_name(fallback_font))
-    try:
-        return ImageFont.truetype(fallback_path, size)
-    except Exception as e:
-        logger.warning("Fallback font %s failed (%s); using PIL default",
-                       fallback_font, e)
-    return ImageFont.load_default()
-
-
-@dataclass(frozen=True)
-class ElementStyle:
-    """Resolved style for one display element."""
-    font: PILFont
-    font_name: str
-    font_size: int
-    #: The user's color when they genuinely changed it, else the plugin's
-    #: classic color (which may be state-dependent — e.g. a score that turns
-    #: gold on a touchdown — so an untouched schema default must never
-    #: clobber it; the web form always posts the color inputs).
-    color: Optional[Tuple[int, int, int]]
-    #: Additive (dx, dy) translation from customization.layout offsets.
-    offset: Tuple[int, int]
-    #: True when the configured value genuinely differs from the schema
-    #: default (NOT merely present — saved configs always contain defaults).
-    user_forced_font: bool
-    user_forced_size: bool
-    user_forced_color: bool = False
-
-    @property
-    def user_forced(self) -> bool:
-        """True when the user pinned this element's font or size; adaptive
-        layouts must use the font as-is instead of ladder-fitting. (Color is
-        deliberately excluded — it never affects sizing.)"""
-        return self.user_forced_font or self.user_forced_size
-
-
-def _as_int(value: Any, default: int) -> int:
-    """Int coercion tolerant of floats and numeric strings from configs."""
-    if value is None:
-        return default
-    if isinstance(value, bool):
-        return default
-    if isinstance(value, (int, float)):
-        return int(value)
-    try:
-        return int(float(value))
-    except (TypeError, ValueError):
-        return default
-
-
-def _as_color(value: Any) -> Optional[Tuple[int, int, int]]:
-    """[r, g, b] list/tuple -> tuple; anything else -> None."""
-    if isinstance(value, (list, tuple)) and len(value) == 3:
-        try:
-            return tuple(max(0, min(255, int(c))) for c in value)
-        except (TypeError, ValueError):
-            return None
-    return None
-
-
-class ElementStyleResolver:
-    """Resolves per-element fonts, colors and offsets from a plugin config.
-
-    ``schema_defaults`` is the nested defaults dict extracted from the
-    plugin's config_schema.json (SchemaManager.extract_defaults_from_schema).
-    It is the reference for the user-override check: a configured value
-    equal to its schema default is treated as untouched, because the save
-    flow persists all defaults. When ``schema_defaults`` is empty (older
-    cores, unit tests), the check degrades to comparing against the
-    ``classic_*`` values the caller supplies.
-    """
-
-    def __init__(self, config: Optional[Dict[str, Any]],
-                 schema_defaults: Optional[Dict[str, Any]] = None, *,
-                 fonts_dir: str = DEFAULT_FONTS_DIR,
-                 fallback_font: str = DEFAULT_FALLBACK_FONT):
-        self._config = config if isinstance(config, dict) else {}
-        self._defaults = schema_defaults if isinstance(schema_defaults, dict) else {}
-        self._fonts_dir = fonts_dir
-        self._fallback_font = fallback_font
-        self._cache: Dict[Any, ElementStyle] = {}
-
-    # -- internals ----------------------------------------------------
-
-    def _element_config(self, element_key: str) -> Dict[str, Any]:
-        cust = self._config.get("customization")
-        if not isinstance(cust, dict):
-            return {}
-        element = cust.get(element_key)
-        return element if isinstance(element, dict) else {}
-
-    def _element_defaults(self, element_key: str) -> Dict[str, Any]:
-        cust = self._defaults.get("customization")
-        if not isinstance(cust, dict):
-            return {}
-        element = cust.get(element_key)
-        return element if isinstance(element, dict) else {}
-
-    # -- public API ---------------------------------------------------
-
-    def style(self, element_key: str, *, classic_font: str, classic_size: int,
-              classic_color: Optional[Tuple[int, int, int]] = None) -> ElementStyle:
-        """Resolve the style for one element.
-
-        ``classic_font``/``classic_size``/``classic_color`` are the plugin's
-        hardcoded defaults for this element — used when the config has no
-        value, and as the override reference when schema defaults are
-        unavailable.
-        """
-        cache_key = (element_key, classic_font, classic_size, classic_color)
-        cached = self._cache.get(cache_key)
-        if cached is not None:
-            return cached
-
-        element_cfg = self._element_config(element_key)
-        element_defaults = self._element_defaults(element_key)
-
-        configured_font = element_cfg.get("font")
-        configured_size = element_cfg.get("font_size")
-
-        # Reference for "did the user change it": schema default when known,
-        # else the plugin's classic default.
-        reference_font = element_defaults.get("font", classic_font)
-        reference_size = _as_int(element_defaults.get("font_size"), classic_size)
-
-        user_forced_font = (configured_font is not None
-                            and configured_font != reference_font)
-        user_forced_size = (configured_size is not None
-                            and _as_int(configured_size, reference_size) != reference_size)
-
-        font_name = configured_font if configured_font is not None else classic_font
-        font_size = _as_int(configured_size, classic_size)
-        font = load_font(font_name, font_size, fonts_dir=self._fonts_dir,
-                         fallback_font=self._fallback_font)
-
-        # Color follows the same provenance rule as fonts: the web form
-        # always posts the RGB inputs, so a saved config carries the schema
-        # default whether or not the user touched it — only a value that
-        # DIFFERS from the schema default is a real override. Otherwise keep
-        # classic_color, which may be state-dependent (semantic colors like
-        # a gold touchdown score) and must not be clobbered by a default.
-        configured_color = _as_color(element_cfg.get("text_color"))
-        default_color = _as_color(element_defaults.get("text_color"))
-        if configured_color is None:
-            user_forced_color = False
-        elif default_color is None:
-            # no schema default to compare against — presence is intent
-            user_forced_color = True
-        else:
-            user_forced_color = configured_color != default_color
-        color = configured_color if user_forced_color else classic_color
-
-        resolved = ElementStyle(
-            font=font, font_name=font_name, font_size=font_size,
-            color=color, offset=self.offset(element_key),
-            user_forced_font=user_forced_font, user_forced_size=user_forced_size,
-            user_forced_color=user_forced_color,
-        )
-        self._cache[cache_key] = resolved
-        return resolved
-
-    def offset_value(self, element_key: str, axis: str, default: int = 0) -> int:
-        """One offset axis for an element (e.g. 'x_offset', 'away_x_offset').
-
-        Reads ``customization.layout.<element>`` first (the deployed sports
-        convention), falling back to ``customization.<element>`` for plugins
-        that keep offsets on the element itself.
-        """
-        cust = self._config.get("customization")
-        if not isinstance(cust, dict):
-            return default
-        layout = cust.get("layout")
-        if isinstance(layout, dict):
-            element = layout.get(element_key)
-            if isinstance(element, dict) and axis in element:
-                return _as_int(element.get(axis), default)
-        element = cust.get(element_key)
-        if isinstance(element, dict) and axis in element:
-            return _as_int(element.get(axis), default)
-        return default
-
-    def offset(self, element_key: str) -> Tuple[int, int]:
-        """(dx, dy) additive translation for an element; (0, 0) when unset."""
-        return (self.offset_value(element_key, "x_offset"),
-                self.offset_value(element_key, "y_offset"))
-
-    def is_for(self, config: Optional[Dict[str, Any]]) -> bool:
-        """True when this resolver was built over exactly this config object.
-
-        Callers that cache a resolver (plugins whose config dict gets
-        REPLACED on config change) use this to decide when to rebuild —
-        preferred over reaching into the private ``_config``.
-        """
-        return self._config is config
-
-    def clear_cache(self) -> None:
-        self._cache.clear()

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/base_plugin.py

@@ -270,64 +270,6 @@ class BasePlugin(ABC):
                 return (int(width), int(height))
         return DEFAULT_DESIGN_SIZE
 
-    # -------------------------------------------------------------------------
-    # Element style resolution (per-element user customization)
-    # -------------------------------------------------------------------------
-    @property
-    def style_resolver(self) -> Any:
-        """
-        ElementStyleResolver for this plugin's customization config.
-
-        Lazily built with the schema defaults from this plugin's own
-        config_schema.json (via the plugin manager's schema manager), so
-        "did the user override this font?" is answered correctly even though
-        saved configs always contain the schema defaults. Rebuilt when the
-        config changes (see on_config_change). Pass it into standalone helper
-        classes (game renderers etc.) instead of letting them build their own.
-
-        See src/element_style.py.
-        """
-        from src.element_style import ElementStyleResolver
-
-        cached = getattr(self, "_style_resolver", None)
-        if cached is not None:
-            return cached
-
-        schema_defaults: Dict[str, Any] = {}
-        schema_manager = getattr(self.plugin_manager, "schema_manager", None)
-        if schema_manager is not None:
-            try:
-                schema = schema_manager.load_schema(self.plugin_id)
-                if schema:
-                    schema_defaults = schema_manager.extract_defaults_from_schema(schema)
-            except Exception as e:
-                self.logger.debug("Schema defaults unavailable for %s: %s",
-                                  self.plugin_id, e)
-
-        resolver = ElementStyleResolver(self.config, schema_defaults)
-        self._style_resolver = resolver
-        return resolver
-
-    def element_style(self, element_key: str, *, classic_font: str,
-                      classic_size: int,
-                      classic_color: Optional[tuple] = None) -> Any:
-        """
-        Resolved font/color/offset for one display element, honoring the
-        user's customization.<element> config. classic_* are this plugin's
-        hardcoded defaults for the element.
-
-        Example:
-            style = self.element_style('title_text',
-                                       classic_font='PressStart2P-Regular.ttf',
-                                       classic_size=8)
-            draw.text((x + style.offset[0], y + style.offset[1]),
-                      title, font=style.font,
-                      fill=style.color or (255, 255, 255))
-        """
-        return self.style_resolver.style(element_key, classic_font=classic_font,
-                                         classic_size=classic_size,
-                                         classic_color=classic_color)
-
     def get_display_duration(self) -> float:
         """
         Get the display duration for this plugin instance.
@@ -790,9 +732,6 @@ class BasePlugin(ABC):
         # Update simple flags
         self.enabled = self.config.get("enabled", self.enabled)
 
-        # Invalidate the cached style resolver — it captured the old config
-        self._style_resolver = None
-
     def get_info(self) -> Dict[str, Any]:
         """
         Return plugin info for display in web UI.

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/schema_manager.py

@@ -110,17 +110,12 @@ class SchemaManager:
         try:
             with open(schema_path, 'r', encoding='utf-8') as f:
                 schema = json.load(f)
-
+            
             # Validate schema structure (basic check)
             if not isinstance(schema, dict):
                 self.logger.error(f"Invalid schema format for {plugin_id}: not a dictionary")
                 return None
-
-            # Expand x-style-elements declarations BEFORE caching, so every
-            # consumer (config form GET, save path, validation, defaults
-            # generation) sees the identical expanded shape.
-            schema = self._expand_style_elements(schema)
-
+            
             # Cache the schema
             self._schema_cache[plugin_id] = schema
             
@@ -137,41 +132,10 @@ class SchemaManager:
             self.logger.error(f"Error loading schema for {plugin_id}: {e}")
             return None
     
-    # ------------------------------------------------------------------
-    # x-style-elements expansion
-    # ------------------------------------------------------------------
-    # Plugins declare styleable display elements compactly via an
-    # "x-style-elements" object on their customization schema; load_schema
-    # expands each declaration into full font/font_size/text_color/offset
-    # property blocks before caching, so the config form, save path,
-    # validation, and defaults generation all see the same expanded shape.
-    # The single implementation lives in src.element_style (pure, also used
-    # by plugins reading their own schema file) — see that module for the
-    # declaration format.
-
-    @staticmethod
-    def get_style_elements(schema: Dict[str, Any]) -> Dict[str, Any]:
-        """The x-style-elements declaration from a schema ({} if none)."""
-        from src.element_style import get_style_elements
-        return get_style_elements(schema)
-
-    def _expand_style_elements(self, schema: Dict[str, Any]) -> Dict[str, Any]:
-        """Expand x-style-elements declarations (no-op without any).
-
-        Never raises; on any failure the original schema is returned so a
-        malformed declaration can't take a plugin down.
-        """
-        try:
-            from src.element_style import expand_style_elements
-            return expand_style_elements(schema)
-        except Exception as e:
-            self.logger.error(f"x-style-elements expansion failed: {e}")
-            return schema
-
     def invalidate_cache(self, plugin_id: Optional[str] = None) -> None:
         """
         Invalidate schema cache for a plugin or all plugins.
-
+        
         Args:
             plugin_id: Plugin identifier to invalidate, or None to clear all
         """

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/plugin_system/testing/render_service.py

@@ -1,121 +0,0 @@
-"""Headless single-render service for plugins.
-
-Renders one plugin instance at one panel size to an in-memory PIL image —
-no hardware, no singletons, no pip (install_deps is always False). Shared
-by the dev server's /api/render endpoints and the production web UI's
-config-page live preview.
-
-A fresh plugin instance is created per call (mirroring the safety
-harness), so repeated renders never share instance state. The plugin's
-module does stay imported in the process — module-level globals persist
-across calls, which is fine for previewing but worth knowing.
-"""
-
-import json
-import logging
-import time
-from pathlib import Path
-from typing import Any, Dict, Optional
-
-logger = logging.getLogger(__name__)
-
-
-def render_plugin_once(plugin_id: str, plugin_dir: Path,
-                       manifest: Optional[Dict[str, Any]] = None,
-                       config: Optional[Dict[str, Any]] = None,
-                       mock_data: Optional[Dict[str, Any]] = None,
-                       width: int = 128, height: int = 32,
-                       skip_update: bool = True) -> Dict[str, Any]:
-    """Render one plugin at one size. Returns a response-shaped dict:
-
-        {'image': 'data:image/png;base64,...', 'width', 'height',
-         'render_time_ms', 'errors': [...], 'warnings': [...]}
-
-    ``skip_update`` defaults to True: update() may block on live network
-    (sports APIs, Spotify) — callers that want real data should prime
-    ``mock_data`` (e.g. from the plugin's test/harness.json fixture, see
-    ``load_harness_spec``) or explicitly pass skip_update=False.
-
-    Raises on plugin load failure; update()/display() exceptions are
-    captured into warnings/errors instead so a broken render still shows
-    whatever was drawn.
-    """
-    from src.plugin_system.plugin_loader import PluginLoader
-    from src.plugin_system.testing import (
-        MockCacheManager, MockPluginManager, VisualTestDisplayManager)
-
-    plugin_dir = Path(plugin_dir)
-    if manifest is None:
-        with open(plugin_dir / 'manifest.json', 'r', encoding='utf-8') as f:
-            manifest = json.load(f)
-    config = config or {'enabled': True}
-    mock_data = mock_data or {}
-
-    display_manager = VisualTestDisplayManager(width=width, height=height)
-    cache_manager = MockCacheManager()
-    plugin_manager = MockPluginManager()
-
-    # Pre-populate cache with mock data
-    for key, value in mock_data.items():
-        cache_manager.set(key, value)
-
-    loader = PluginLoader()
-    errors = []
-    warnings = []
-
-    plugin_instance, _module = loader.load_plugin(
-        plugin_id=plugin_id,
-        manifest=manifest,
-        plugin_dir=plugin_dir,
-        config=config,
-        display_manager=display_manager,
-        cache_manager=cache_manager,
-        plugin_manager=plugin_manager,
-        install_deps=False,
-    )
-
-    start_time = time.time()
-
-    # Exception detail policy (matches the dev server's convention): full
-    # tracebacks go to the server log; clients get only the exception class
-    # name — CodeQL flags raw exception text in responses as stack-trace
-    # exposure, and the log is where developers look anyway.
-    try:
-        if not skip_update:
-            try:
-                plugin_instance.update()
-            except Exception as e:
-                logger.warning("update() raised for plugin %s", plugin_id,
-                               exc_info=True)
-                warnings.append(f"update() raised: {type(e).__name__} — see server log")
-
-        try:
-            plugin_instance.display(force_clear=True)
-        except Exception as 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)
-        # Capture BEFORE cleanup — a plugin's cleanup may clear the canvas
-        image_b64 = display_manager.get_image_base64()
-    finally:
-        # The instance is throwaway, but its __init__ may have opened
-        # sessions or started threads (music's clients, sports API
-        # sessions). In a long-running web process, previews without
-        # cleanup would leak those per request.
-        try:
-            plugin_instance.cleanup()
-        except Exception as e:
-            logger.warning("cleanup() raised for plugin %s", plugin_id,
-                           exc_info=True)
-            warnings.append(f"cleanup() raised: {type(e).__name__} — see server log")
-
-    return {
-        'image': f'data:image/png;base64,{image_b64}',
-        'width': width,
-        'height': height,
-        'render_time_ms': render_time_ms,
-        'errors': errors,
-        'warnings': warnings,
-    }

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")
@@ -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_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_element_style.py

@@ -1,519 +0,0 @@
-"""Tests for src/element_style.py — universal per-element style resolution.
-
-The load-bearing behavior is the user-override check: saved configs ALWAYS
-contain the schema defaults (merge_with_defaults runs at save time and again
-before plugin instantiation), so "key present" must never be read as "user
-set it". Only "present and different from the schema default" counts.
-"""
-
-import json
-import os
-import sys
-
-import pytest
-from PIL import ImageFont
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.element_style import (  # noqa: E402
-    ElementStyle,
-    ElementStyleResolver,
-    FONT_ALIASES,
-    defaults_from_schema_file,
-    extract_schema_defaults,
-    load_font,
-    resolve_font_name,
-)
-
-PRESS_START = "PressStart2P-Regular.ttf"
-FOUR_BY_SIX = "4x6-font.ttf"
-FIVE_BY_SEVEN_BDF = "5x7.bdf"
-
-
-# ---------------------------------------------------------------------------
-# load_font
-# ---------------------------------------------------------------------------
-
-class TestLoadFont:
-    def test_ttf(self):
-        font = load_font(PRESS_START, 8)
-        assert isinstance(font, ImageFont.FreeTypeFont)
-        assert font.size == 8
-
-    def test_bdf_at_native_size(self):
-        """FreeType loads BDF strikes directly at their native size."""
-        font = load_font(FIVE_BY_SEVEN_BDF, 7)
-        assert isinstance(font, ImageFont.FreeTypeFont)
-
-    def test_bdf_at_wrong_size_falls_back(self):
-        """BDF fonts are fixed-size; a non-native size falls back to the
-        fallback font at the requested size rather than raising."""
-        font = load_font(FIVE_BY_SEVEN_BDF, 14)
-        assert isinstance(font, ImageFont.FreeTypeFont)
-        assert font.size == 14  # fallback font honored the requested size
-
-    def test_alias_resolves(self):
-        assert resolve_font_name("press_start") == PRESS_START
-        font = load_font("press_start", 16)
-        assert isinstance(font, ImageFont.FreeTypeFont)
-        assert font.size == 16
-
-    def test_filename_passes_through_alias(self):
-        assert resolve_font_name(FOUR_BY_SIX) == FOUR_BY_SIX
-
-    def test_missing_file_falls_back(self):
-        font = load_font("no-such-font.ttf", 10)
-        assert isinstance(font, ImageFont.FreeTypeFont)
-        assert font.size == 10
-
-    @pytest.mark.parametrize("garbage", ["", None, "../../etc/passwd", "x" * 300])
-    def test_garbage_never_raises(self, garbage):
-        font = load_font(garbage, 8)
-        assert font is not None
-
-    def test_everything_missing_uses_pil_default(self):
-        font = load_font("nope.ttf", 8, fonts_dir="/nonexistent",
-                         fallback_font="also-nope.ttf")
-        assert font is not None  # ImageFont.load_default()
-
-    def test_aliases_cover_the_baseball_set(self):
-        """The centralized map must be a superset of the per-plugin copies
-        it replaces (baseball game_renderer.py + sports.py)."""
-        assert FONT_ALIASES["press_start"] == PRESS_START
-        assert FONT_ALIASES["four_by_six"] == FOUR_BY_SIX
-        assert FONT_ALIASES["five_by_seven"] == FIVE_BY_SEVEN_BDF
-
-
-# ---------------------------------------------------------------------------
-# user_forced provenance
-# ---------------------------------------------------------------------------
-
-SCHEMA_DEFAULTS = {
-    "customization": {
-        "score_text": {"font": PRESS_START, "font_size": 10},
-    }
-}
-
-
-def _style(config, defaults=SCHEMA_DEFAULTS):
-    return ElementStyleResolver(config, defaults).style(
-        "score_text", classic_font=PRESS_START, classic_size=10)
-
-
-class TestUserForced:
-    def test_absent_is_not_forced(self):
-        style = _style({})
-        assert not style.user_forced
-        assert style.font_name == PRESS_START
-        assert style.font_size == 10
-
-    def test_schema_default_present_is_not_forced(self):
-        """THE bug this module exists to fix: the save flow writes schema
-        defaults into every saved config, so their presence means nothing."""
-        config = {"customization": {"score_text": {
-            "font": PRESS_START, "font_size": 10}}}
-        style = _style(config)
-        assert not style.user_forced
-
-    def test_different_font_is_forced(self):
-        config = {"customization": {"score_text": {
-            "font": FOUR_BY_SIX, "font_size": 10}}}
-        style = _style(config)
-        assert style.user_forced_font
-        assert not style.user_forced_size
-        assert style.user_forced
-
-    def test_different_size_is_forced(self):
-        config = {"customization": {"score_text": {
-            "font": PRESS_START, "font_size": 14}}}
-        style = _style(config)
-        assert style.user_forced_size
-        assert not style.user_forced_font
-        assert style.font_size == 14
-
-    def test_string_size_equal_to_default_is_not_forced(self):
-        config = {"customization": {"score_text": {"font_size": "10"}}}
-        assert not _style(config).user_forced
-
-    def test_without_schema_defaults_compares_against_classic(self):
-        """Degraded mode (old cores, tests): classic_* is the reference."""
-        config = {"customization": {"score_text": {
-            "font": PRESS_START, "font_size": 10}}}
-        style = _style(config, defaults={})
-        assert not style.user_forced
-        forced = _style({"customization": {"score_text": {"font_size": 12}}},
-                        defaults={})
-        assert forced.user_forced
-
-    def test_schema_default_differing_from_classic_wins_as_reference(self):
-        """When the schema declares a different default than the classic_*
-        args, the schema is the reference — a config equal to the schema
-        default is untouched."""
-        defaults = {"customization": {"score_text": {
-            "font": FOUR_BY_SIX, "font_size": 6}}}
-        config = {"customization": {"score_text": {
-            "font": FOUR_BY_SIX, "font_size": 6}}}
-        style = ElementStyleResolver(config, defaults).style(
-            "score_text", classic_font=PRESS_START, classic_size=10)
-        assert not style.user_forced
-
-    def test_unknown_element_uses_classic(self):
-        style = ElementStyleResolver({}, SCHEMA_DEFAULTS).style(
-            "no_such_element", classic_font=FOUR_BY_SIX, classic_size=6)
-        assert not style.user_forced
-        assert style.font_name == FOUR_BY_SIX
-        assert style.font_size == 6
-
-    def test_malformed_customization_is_tolerated(self):
-        for bad in [{"customization": "oops"},
-                    {"customization": {"score_text": "oops"}},
-                    {"customization": {"score_text": {"font_size": "huge"}}},
-                    None]:
-            style = _style(bad)
-            assert not style.user_forced
-            assert style.font_size == 10
-
-
-# ---------------------------------------------------------------------------
-# color
-# ---------------------------------------------------------------------------
-
-class TestColor:
-    """Color provenance mirrors fonts: the web form ALWAYS posts the RGB
-    inputs, so a saved config carries the schema-default color whether or
-    not the user touched it. Only a value differing from the schema default
-    is an override; otherwise the plugin's classic color survives — critical
-    for state-dependent colors (a score that turns gold on a touchdown)."""
-
-    COLOR_DEFAULTS = {"customization": {"score_text": {
-        "font": PRESS_START, "font_size": 10, "text_color": [255, 255, 255]}}}
-
-    def _color_style(self, config, defaults=None, classic_color=(255, 215, 0)):
-        return ElementStyleResolver(config, defaults or self.COLOR_DEFAULTS).style(
-            "score_text", classic_font=PRESS_START, classic_size=10,
-            classic_color=classic_color)
-
-    def test_absent_returns_classic_color(self):
-        style = self._color_style({})
-        assert style.color == (255, 215, 0)
-        assert not style.user_forced_color
-
-    def test_absent_with_no_classic_is_none(self):
-        assert _style({}).color is None
-
-    def test_schema_default_present_keeps_classic_color(self):
-        """A saved config always contains the default — it must not clobber
-        the plugin's (possibly semantic) classic color."""
-        config = {"customization": {"score_text": {"text_color": [255, 255, 255]}}}
-        style = self._color_style(config)
-        assert style.color == (255, 215, 0)
-        assert not style.user_forced_color
-
-    def test_changed_color_is_an_override(self):
-        config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
-        style = self._color_style(config)
-        assert style.color == (0, 128, 255)
-        assert style.user_forced_color
-
-    def test_present_without_schema_default_is_an_override(self):
-        """Hand-written schemas without a text_color default: presence is
-        intent (there is nothing to compare against)."""
-        config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
-        style = self._color_style(config, defaults=SCHEMA_DEFAULTS)
-        assert style.color == (0, 128, 255)
-        assert style.user_forced_color
-
-    def test_values_clamped(self):
-        config = {"customization": {"score_text": {"text_color": [300, -5, 128]}}}
-        assert self._color_style(config).color == (255, 0, 128)
-
-    def test_color_never_affects_user_forced_sizing(self):
-        config = {"customization": {"score_text": {"text_color": [0, 128, 255]}}}
-        style = self._color_style(config)
-        assert style.user_forced_color
-        assert not style.user_forced
-
-    @pytest.mark.parametrize("bad", [[1, 2], [1, 2, 3, 4], "red",
-                                     ["a", "b", "c"], 255, None])
-    def test_malformed_falls_back_to_classic(self, bad):
-        config = {"customization": {"score_text": {"text_color": bad}}}
-        style = self._color_style(config, classic_color=(1, 2, 3))
-        assert style.color == (1, 2, 3)
-        assert not style.user_forced_color
-
-
-# ---------------------------------------------------------------------------
-# offsets
-# ---------------------------------------------------------------------------
-
-class TestOffsets:
-    def test_unset_is_zero(self):
-        assert ElementStyleResolver({}).offset("score") == (0, 0)
-
-    def test_layout_section(self):
-        """The deployed sports convention: customization.layout.<element>."""
-        config = {"customization": {"layout": {"score": {
-            "x_offset": 3, "y_offset": -2}}}}
-        assert ElementStyleResolver(config).offset("score") == (3, -2)
-
-    def test_element_section_fallback(self):
-        config = {"customization": {"score": {"x_offset": 5}}}
-        assert ElementStyleResolver(config).offset("score") == (5, 0)
-
-    def test_layout_section_wins_over_element_section(self):
-        config = {"customization": {
-            "layout": {"score": {"x_offset": 1}},
-            "score": {"x_offset": 9, "y_offset": 9},
-        }}
-        resolver = ElementStyleResolver(config)
-        assert resolver.offset_value("score", "x_offset") == 1
-        # y_offset absent from layout section -> element section supplies it
-        assert resolver.offset_value("score", "y_offset") == 9
-
-    @pytest.mark.parametrize("raw,expected", [
-        (2, 2), (2.7, 2), ("3", 3), ("2.0", 2), ("-4", -4),
-        (None, 0), ("junk", 0), ([], 0), (True, 0),
-    ])
-    def test_coercion_matches_sports_helper(self, raw, expected):
-        """Same tolerance as the sports.py/_get_layout_offset copies this
-        replaces: int/float/numeric-string pass, anything else -> default."""
-        config = {"customization": {"layout": {"e": {"x_offset": raw}}}}
-        assert ElementStyleResolver(config).offset_value("e", "x_offset") == expected
-
-    def test_custom_axis_names(self):
-        """Football's records use away_x_offset/home_x_offset."""
-        config = {"customization": {"layout": {"records": {
-            "away_x_offset": 4, "home_x_offset": -4}}}}
-        resolver = ElementStyleResolver(config)
-        assert resolver.offset_value("records", "away_x_offset") == 4
-        assert resolver.offset_value("records", "home_x_offset") == -4
-
-    def test_style_carries_offset(self):
-        config = {"customization": {"layout": {"score_text": {
-            "x_offset": 2, "y_offset": 1}}}}
-        assert _style(config).offset == (2, 1)
-
-
-# ---------------------------------------------------------------------------
-# caching
-# ---------------------------------------------------------------------------
-
-class TestCaching:
-    def test_same_call_is_cached(self):
-        resolver = ElementStyleResolver({}, SCHEMA_DEFAULTS)
-        a = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
-        b = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
-        assert a is b
-
-    def test_clear_cache(self):
-        resolver = ElementStyleResolver({}, SCHEMA_DEFAULTS)
-        a = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
-        resolver.clear_cache()
-        b = resolver.style("score_text", classic_font=PRESS_START, classic_size=10)
-        assert a is not b
-        # PIL fonts compare by identity; compare the value fields
-        assert (a.font_name, a.font_size, a.color, a.offset, a.user_forced) == \
-               (b.font_name, b.font_size, b.color, b.offset, b.user_forced)
-
-
-# ---------------------------------------------------------------------------
-# schema default extraction
-# ---------------------------------------------------------------------------
-
-class TestSchemaDefaults:
-    def test_matches_schema_manager_extraction(self):
-        """The pure helper must agree with SchemaManager.extract_defaults_from_schema
-        on a real plugin-style schema — it exists so plugins get the same
-        answer in harness contexts where the schema manager is absent."""
-        from src.plugin_system.schema_manager import SchemaManager
-        schema = {
-            "type": "object",
-            "properties": {
-                "enabled": {"type": "boolean", "default": True},
-                "customization": {
-                    "type": "object",
-                    "properties": {
-                        "score_text": {
-                            "type": "object",
-                            "properties": {
-                                "font": {"type": "string", "default": PRESS_START},
-                                "font_size": {"type": "integer", "default": 10},
-                                "y_percent": {"type": "number"},
-                            },
-                        },
-                        "layout": {
-                            "type": "object",
-                            "properties": {
-                                "score": {
-                                    "type": "object",
-                                    "properties": {
-                                        "x_offset": {"type": "integer", "default": 0},
-                                    },
-                                },
-                            },
-                        },
-                    },
-                },
-                "opaque_with_default": {"type": "object", "default": {},
-                                        "properties": {"x": {"default": 1}}},
-                # array shapes the schema manager special-cases — parity
-                # must hold for these too (found divergent in review)
-                "plain_array": {"type": "array", "items": {"type": "string"}},
-                "obj_array": {"type": "array", "items": {
-                    "type": "object", "properties": {"x": {"default": 1}}}},
-                "item_default_array": {"type": "array",
-                                       "items": {"type": "string", "default": "a"}},
-            },
-        }
-        pure = extract_schema_defaults(schema)
-        managed = SchemaManager().extract_defaults_from_schema(schema)
-        assert pure == managed
-        assert pure["customization"]["score_text"]["font"] == PRESS_START
-        # object-level default short-circuits recursion (both must agree)
-        assert pure["opaque_with_default"] == {}
-        assert pure["plain_array"] == []
-        assert pure["obj_array"] == []
-        assert pure["item_default_array"] == ["a"]
-
-    def test_defaults_from_schema_file(self, tmp_path):
-        schema_path = tmp_path / "config_schema.json"
-        schema_path.write_text(json.dumps({
-            "type": "object",
-            "properties": {
-                "customization": {
-                    "type": "object",
-                    "properties": {
-                        "title_text": {
-                            "type": "object",
-                            "properties": {
-                                "font": {"type": "string", "default": PRESS_START},
-                            },
-                        },
-                    },
-                },
-            },
-        }))
-        defaults = defaults_from_schema_file(str(schema_path))
-        assert defaults["customization"]["title_text"]["font"] == PRESS_START
-
-    def test_defaults_from_missing_or_bad_file(self, tmp_path):
-        assert defaults_from_schema_file("/nonexistent/schema.json") == {}
-        bad = tmp_path / "bad.json"
-        bad.write_text("{not json")
-        assert defaults_from_schema_file(str(bad)) == {}
-
-
-# ---------------------------------------------------------------------------
-# BasePlugin integration
-# ---------------------------------------------------------------------------
-
-class _StubSchemaManager:
-    """Schema manager double exposing the two methods the resolver path uses."""
-
-    def __init__(self, schema):
-        self._schema = schema
-
-    def load_schema(self, plugin_id, use_cache=True):
-        return self._schema
-
-    def extract_defaults_from_schema(self, schema, prefix=""):
-        # Mirror the real nested-dict extraction for this simple shape
-        def walk(props):
-            out = {}
-            for key, spec in props.get("properties", {}).items():
-                if "default" in spec:
-                    out[key] = spec["default"]
-                elif spec.get("type") == "object" and "properties" in spec:
-                    nested = walk(spec)
-                    if nested:
-                        out[key] = nested
-            return out
-        return walk(schema)
-
-
-def _make_plugin(config, schema=None):
-    from src.plugin_system.base_plugin import BasePlugin
-    from src.plugin_system.testing.mocks import (
-        MockCacheManager, MockPluginManager)
-    from src.plugin_system.testing.visual_display_manager import (
-        VisualTestDisplayManager)
-
-    class _Plugin(BasePlugin):
-        def update(self):
-            return True
-
-        def display(self, force_clear=False):
-            return None
-
-    plugin_manager = MockPluginManager()
-    if schema is not None:
-        plugin_manager.schema_manager = _StubSchemaManager(schema)
-    return _Plugin("test-plugin", config,
-                   VisualTestDisplayManager(64, 32),
-                   MockCacheManager(), plugin_manager)
-
-
-TEST_SCHEMA = {
-    "type": "object",
-    "properties": {
-        "customization": {
-            "type": "object",
-            "properties": {
-                "score_text": {
-                    "type": "object",
-                    "properties": {
-                        "font": {"type": "string", "default": PRESS_START},
-                        "font_size": {"type": "integer", "default": 10},
-                    },
-                },
-            },
-        },
-    },
-}
-
-
-class TestBasePluginIntegration:
-    def test_element_style_with_schema_defaults(self):
-        """The full path: saved config carries schema defaults, plugin's
-        element_style still reports not-forced."""
-        config = {"enabled": True, "customization": {"score_text": {
-            "font": PRESS_START, "font_size": 10}}}
-        plugin = _make_plugin(config, schema=TEST_SCHEMA)
-        style = plugin.element_style("score_text", classic_font=PRESS_START,
-                                     classic_size=10)
-        assert not style.user_forced
-
-    def test_element_style_detects_real_override(self):
-        config = {"enabled": True, "customization": {"score_text": {
-            "font": PRESS_START, "font_size": 14}}}
-        plugin = _make_plugin(config, schema=TEST_SCHEMA)
-        style = plugin.element_style("score_text", classic_font=PRESS_START,
-                                     classic_size=10)
-        assert style.user_forced_size
-        assert style.font_size == 14
-
-    def test_works_without_schema_manager(self):
-        """MockPluginManager has no schema_manager attribute by default —
-        the resolver degrades to classic-default comparison, no crash."""
-        config = {"enabled": True, "customization": {"score_text": {
-            "font_size": 12}}}
-        plugin = _make_plugin(config, schema=None)
-        style = plugin.element_style("score_text", classic_font=PRESS_START,
-                                     classic_size=10)
-        assert style.user_forced_size  # 12 != classic 10
-
-    def test_resolver_is_cached_and_invalidated_on_config_change(self):
-        plugin = _make_plugin({"enabled": True}, schema=TEST_SCHEMA)
-        first = plugin.style_resolver
-        assert plugin.style_resolver is first
-        plugin.on_config_change({"enabled": True, "customization": {
-            "score_text": {"font_size": 14}}})
-        second = plugin.style_resolver
-        assert second is not first
-        style = plugin.element_style("score_text", classic_font=PRESS_START,
-                                     classic_size=10)
-        assert style.user_forced_size
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-v"]))

test/test_schema_style_expansion.py

@@ -1,273 +0,0 @@
-"""Tests for x-style-elements schema expansion.
-
-A plugin declares styleable elements once, compactly; expansion generates
-the full customization property blocks at schema-load time. The invariants
-that matter:
-
-- idempotent (expand(expand(s)) == expand(s)) and the input is never mutated
-- both load paths (cached GET, uncached save) see the identical shape
-- generated defaults flow into generate_default_config, and saving twice is
-  round-trip stable (merge_with_defaults produces no churn)
-- hand-written property blocks for the same element always win
-- defaults_from_schema_file (what plugins use to build resolvers from their
-  RAW schema file) agrees exactly with the schema manager's expanded view
-"""
-
-import copy
-import json
-import os
-import sys
-
-import jsonschema
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.element_style import (  # noqa: E402
-    defaults_from_schema_file,
-    expand_style_elements,
-    extract_schema_defaults,
-    get_style_elements,
-)
-from src.plugin_system.schema_manager import SchemaManager  # noqa: E402
-
-PRESS_START = "PressStart2P-Regular.ttf"
-
-
-def _declared_schema():
-    return {
-        "$schema": "http://json-schema.org/draft-07/schema#",
-        "type": "object",
-        "properties": {
-            "enabled": {"type": "boolean", "default": True},
-            "customization": {
-                "type": "object",
-                "title": "Display Customization",
-                "x-style-elements": {
-                    "score_text": {
-                        "title": "Game Score",
-                        "font": {"default": PRESS_START},
-                        "size": {"default": 10, "min": 4, "max": 16},
-                        "color": True,
-                        "offsets": True,
-                    },
-                    "detail_text": {
-                        "font": {"default": "4x6-font.ttf"},
-                        "size": {"default": 6},
-                    },
-                },
-                "properties": {},
-                "additionalProperties": False,
-            },
-        },
-    }
-
-
-class TestExpansionShape:
-    def test_generates_element_blocks(self):
-        expanded = expand_style_elements(_declared_schema())
-        cust = expanded["properties"]["customization"]["properties"]
-        score = cust["score_text"]
-        assert score["x-style-managed"] is True
-        assert score["title"] == "Game Score"
-        assert score["properties"]["font"]["default"] == PRESS_START
-        assert score["properties"]["font"]["x-widget"] == "font-selector"
-        assert score["properties"]["font_size"]["default"] == 10
-        assert score["properties"]["font_size"]["minimum"] == 4
-        assert score["properties"]["font_size"]["maximum"] == 16
-        assert score["properties"]["text_color"]["x-widget"] == "color-picker"
-        assert score["properties"]["text_color"]["default"] == [255, 255, 255]
-        assert score["additionalProperties"] is False
-
-    def test_color_and_offsets_are_optional(self):
-        expanded = expand_style_elements(_declared_schema())
-        cust = expanded["properties"]["customization"]["properties"]
-        detail = cust["detail_text"]
-        assert "text_color" not in detail["properties"]
-        assert detail["title"] == "Detail Text"  # prettified from the key
-        layout = cust["layout"]["properties"]
-        assert "score_text" in layout
-        assert "detail_text" not in layout
-
-    def test_offsets_block_shape(self):
-        expanded = expand_style_elements(_declared_schema())
-        layout = expanded["properties"]["customization"]["properties"]["layout"]
-        assert layout["x-style-managed"] is True
-        entry = layout["properties"]["score_text"]
-        assert entry["properties"]["x_offset"]["default"] == 0
-        assert entry["properties"]["y_offset"]["default"] == 0
-
-    def test_declared_color_default(self):
-        schema = _declared_schema()
-        decl = schema["properties"]["customization"]["x-style-elements"]
-        decl["score_text"]["color"] = {"default": [255, 215, 0]}
-        expanded = expand_style_elements(schema)
-        color = (expanded["properties"]["customization"]["properties"]
-                 ["score_text"]["properties"]["text_color"])
-        assert color["default"] == [255, 215, 0]
-
-    def test_declaration_survives_expansion(self):
-        """The declaration is the element registry for tooling — it must
-        remain readable from the expanded schema."""
-        expanded = expand_style_elements(_declared_schema())
-        assert set(get_style_elements(expanded)) == {"score_text", "detail_text"}
-        assert set(SchemaManager.get_style_elements(expanded)) == {
-            "score_text", "detail_text"}
-
-    def test_no_declaration_returns_same_object(self):
-        schema = {"type": "object", "properties": {"enabled": {"default": True}}}
-        assert expand_style_elements(schema) is schema
-
-    def test_valid_draft7(self):
-        jsonschema.Draft7Validator.check_schema(
-            expand_style_elements(_declared_schema()))
-
-    def test_property_order_updated_when_present(self):
-        schema = _declared_schema()
-        schema["properties"]["customization"]["x-propertyOrder"] = []
-        expanded = expand_style_elements(schema)
-        order = expanded["properties"]["customization"]["x-propertyOrder"]
-        # generated elements before layout (the template only renders keys
-        # in x-propertyOrder when one exists)
-        assert set(order) == {"score_text", "detail_text", "layout"}
-        assert order.index("score_text") < order.index("layout")
-
-    def test_malformed_declaration_is_harmless(self):
-        schema = _declared_schema()
-        schema["properties"]["customization"]["x-style-elements"] = {
-            "bad": "not a dict", "score_text": {"size": {"default": 10}}}
-        expanded = expand_style_elements(schema)
-        cust = expanded["properties"]["customization"]["properties"]
-        assert "bad" not in cust
-        assert "score_text" in cust
-
-
-class TestExpansionInvariants:
-    def test_idempotent(self):
-        once = expand_style_elements(_declared_schema())
-        twice = expand_style_elements(once)
-        assert once == twice
-
-    def test_input_never_mutated(self):
-        schema = _declared_schema()
-        snapshot = copy.deepcopy(schema)
-        expand_style_elements(schema)
-        assert schema == snapshot
-
-    def test_hand_written_block_wins(self):
-        schema = _declared_schema()
-        hand_written = {
-            "type": "object",
-            "properties": {"font": {"type": "string", "default": "custom.ttf"}},
-        }
-        schema["properties"]["customization"]["properties"]["score_text"] = \
-            copy.deepcopy(hand_written)
-        expanded = expand_style_elements(schema)
-        assert (expanded["properties"]["customization"]["properties"]["score_text"]
-                == hand_written)
-
-    def test_hand_written_layout_entry_wins(self):
-        schema = _declared_schema()
-        schema["properties"]["customization"]["properties"]["layout"] = {
-            "type": "object",
-            "properties": {"score_text": {"type": "object", "properties": {
-                "x_offset": {"type": "integer", "default": 5}}}},
-        }
-        expanded = expand_style_elements(schema)
-        layout = expanded["properties"]["customization"]["properties"]["layout"]
-        assert layout["properties"]["score_text"]["properties"]["x_offset"]["default"] == 5
-
-
-class TestSchemaManagerIntegration:
-    def _manager_with_schema(self, tmp_path, schema):
-        plugin_dir = tmp_path / "test-plugin"
-        plugin_dir.mkdir()
-        (plugin_dir / "config_schema.json").write_text(json.dumps(schema))
-        return SchemaManager(plugins_dir=tmp_path)
-
-    def test_load_schema_expands(self, tmp_path):
-        mgr = self._manager_with_schema(tmp_path, _declared_schema())
-        loaded = mgr.load_schema("test-plugin")
-        assert "score_text" in loaded["properties"]["customization"]["properties"]
-
-    def test_cached_and_uncached_loads_agree(self, tmp_path):
-        """The save path uses use_cache=False while the form GET uses the
-        cache — they must see the identical expanded shape."""
-        mgr = self._manager_with_schema(tmp_path, _declared_schema())
-        cached = mgr.load_schema("test-plugin", use_cache=True)
-        again = mgr.load_schema("test-plugin", use_cache=True)
-        uncached = mgr.load_schema("test-plugin", use_cache=False)
-        assert cached == uncached == again
-
-    def test_disk_file_untouched(self, tmp_path):
-        schema = _declared_schema()
-        mgr = self._manager_with_schema(tmp_path, schema)
-        mgr.load_schema("test-plugin")
-        on_disk = json.loads(
-            (tmp_path / "test-plugin" / "config_schema.json").read_text())
-        assert on_disk == schema
-        assert "score_text" not in on_disk["properties"]["customization"]["properties"]
-
-    def test_defaults_include_generated_elements(self, tmp_path):
-        mgr = self._manager_with_schema(tmp_path, _declared_schema())
-        defaults = mgr.generate_default_config("test-plugin")
-        assert defaults["customization"]["score_text"]["font"] == PRESS_START
-        assert defaults["customization"]["score_text"]["font_size"] == 10
-        assert defaults["customization"]["score_text"]["text_color"] == [255, 255, 255]
-        assert defaults["customization"]["layout"]["score_text"]["x_offset"] == 0
-
-    def test_save_twice_is_round_trip_stable(self, tmp_path):
-        """merge_with_defaults(merged, defaults) must be a fixed point —
-        saving a config twice can't keep growing/altering it."""
-        mgr = self._manager_with_schema(tmp_path, _declared_schema())
-        defaults = mgr.generate_default_config("test-plugin")
-        user_config = {"enabled": True, "customization": {
-            "score_text": {"font_size": 14}}}
-        merged_once = mgr.merge_with_defaults(user_config, defaults)
-        merged_twice = mgr.merge_with_defaults(merged_once, defaults)
-        assert merged_once == merged_twice
-        assert merged_once["customization"]["score_text"]["font_size"] == 14
-
-
-class TestResolverParity:
-    def test_defaults_from_schema_file_matches_manager_view(self, tmp_path):
-        """Plugins build resolvers from their RAW schema file; the web UI
-        merges defaults from the EXPANDED schema. Both must produce the
-        same defaults or override detection diverges between contexts."""
-        schema = _declared_schema()
-        plugin_dir = tmp_path / "test-plugin"
-        plugin_dir.mkdir()
-        schema_path = plugin_dir / "config_schema.json"
-        schema_path.write_text(json.dumps(schema))
-
-        mgr = SchemaManager(plugins_dir=tmp_path)
-        manager_defaults = mgr.extract_defaults_from_schema(
-            mgr.load_schema("test-plugin"))
-        raw_file_defaults = defaults_from_schema_file(str(schema_path))
-        assert raw_file_defaults == manager_defaults
-
-    def test_resolver_treats_generated_defaults_as_untouched(self, tmp_path):
-        """End to end: a config saved through the web UI (all generated
-        defaults baked in) must not read as a user override, and the
-        schema-default color must not clobber a classic color."""
-        from src.element_style import ElementStyleResolver
-        schema_path = tmp_path / "config_schema.json"
-        schema_path.write_text(json.dumps(_declared_schema()))
-        defaults = defaults_from_schema_file(str(schema_path))
-
-        saved_config = {"enabled": True, "customization": {
-            "score_text": {"font": PRESS_START, "font_size": 10,
-                           "text_color": [255, 255, 255]},
-            "layout": {"score_text": {"x_offset": 0, "y_offset": 0}},
-        }}
-        resolver = ElementStyleResolver(saved_config, defaults)
-        style = resolver.style("score_text", classic_font=PRESS_START,
-                               classic_size=10, classic_color=(255, 215, 0))
-        assert not style.user_forced
-        assert not style.user_forced_color
-        assert style.color == (255, 215, 0)  # semantic classic color survives
-        assert style.offset == (0, 0)
-
-
-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

test/web_interface/test_plugin_preview.py

@@ -1,252 +0,0 @@
-"""Tests for POST /api/v3/plugins/preview — the config-page live preview.
-
-The endpoint renders a plugin headlessly (pure PIL, no hardware, no pip)
-with a CANDIDATE config: either the current form state (parsed by the same
-parse_plugin_config_form used by save, so preview and save can never
-disagree) or a JSON config body.
-"""
-
-import base64
-import io
-import json
-import sys
-from pathlib import Path
-from unittest.mock import MagicMock
-
-import pytest
-from flask import Flask
-from PIL import Image
-
-project_root = Path(__file__).parent.parent.parent
-sys.path.insert(0, str(project_root))
-
-from web_interface.blueprints import api_v3 as api_v3_module  # noqa: E402
-from web_interface.blueprints.api_v3 import api_v3  # noqa: E402
-
-PLUGIN_ID = "preview-test-plugin"
-
-MANAGER_PY = '''
-from PIL import ImageFont
-from src.plugin_system.base_plugin import BasePlugin
-
-
-class PreviewTestPlugin(BasePlugin):
-    def update(self):
-        return True
-
-    def display(self, force_clear=False):
-        if force_clear:
-            self.display_manager.clear()
-        text = self.config.get("message", "hello")
-        self.display_manager.draw.text((1, 1), text, fill=(255, 255, 255))
-        self.display_manager.update_display()
-'''
-
-MANIFEST = {
-    "id": PLUGIN_ID,
-    "name": "Preview Test Plugin",
-    "version": "1.0.0",
-    "class_name": "PreviewTestPlugin",
-    "entry_point": "manager.py",
-    "display_modes": ["preview_test"],
-}
-
-SCHEMA = {
-    "type": "object",
-    "properties": {
-        "enabled": {"type": "boolean", "default": True},
-        "message": {"type": "string", "default": "hello"},
-    },
-}
-
-
-@pytest.fixture
-def plugin_dir(tmp_path):
-    plugin = tmp_path / PLUGIN_ID
-    plugin.mkdir()
-    (plugin / "manager.py").write_text(MANAGER_PY)
-    (plugin / "manifest.json").write_text(json.dumps(MANIFEST))
-    (plugin / "config_schema.json").write_text(json.dumps(SCHEMA))
-    return plugin
-
-
-@pytest.fixture
-def client(plugin_dir, tmp_path):
-    from src.plugin_system.schema_manager import SchemaManager
-
-    test_app = Flask(__name__)
-    test_app.register_blueprint(api_v3, url_prefix="/api/v3")
-
-    config_manager = MagicMock()
-    config_manager.load_config.return_value = {
-        "display": {"hardware": {"cols": 64, "chain_length": 2,
-                                 "rows": 32, "parallel": 1}},
-        PLUGIN_ID: {"enabled": False, "message": "saved"},
-    }
-
-    plugin_manager = MagicMock()
-    plugin_manager.plugins_dir = str(tmp_path)
-
-    old = (getattr(api_v3_module.api_v3, "config_manager", None),
-           getattr(api_v3_module.api_v3, "plugin_manager", None),
-           getattr(api_v3_module.api_v3, "schema_manager", None))
-    api_v3_module.api_v3.config_manager = config_manager
-    api_v3_module.api_v3.plugin_manager = plugin_manager
-    api_v3_module.api_v3.schema_manager = SchemaManager(plugins_dir=tmp_path)
-
-    with test_app.test_client() as c:
-        yield c
-
-    (api_v3_module.api_v3.config_manager,
-     api_v3_module.api_v3.plugin_manager,
-     api_v3_module.api_v3.schema_manager) = old
-
-
-def _decode_image(data_url):
-    assert data_url.startswith("data:image/png;base64,")
-    raw = base64.b64decode(data_url.split(",", 1)[1])
-    return Image.open(io.BytesIO(raw))
-
-
-class TestPreviewEndpoint:
-    def test_json_body_renders_at_default_panel_size(self, client):
-        """No width/height -> the user's real panel (64*2 x 32*1)."""
-        resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
-                           json={"config": {"message": "hi"}})
-        assert resp.status_code == 200
-        data = resp.get_json()["data"]
-        img = _decode_image(data["image"])
-        assert img.size == (128, 32)
-        assert data["errors"] == []
-
-    def test_explicit_size(self, client):
-        resp = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=64&height=64",
-            json={"config": {}})
-        assert resp.status_code == 200
-        img = _decode_image(resp.get_json()["data"]["image"])
-        assert img.size == (64, 64)
-
-    def test_form_encoding_matches_json(self, client):
-        """The form path (what HTMX posts) and the JSON path must render
-        the same candidate config identically."""
-        via_json = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            json={"config": {"message": "same"}})
-        via_form = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            data={"message": "same"})
-        a = _decode_image(via_json.get_json()["data"]["image"])
-        b = _decode_image(via_form.get_json()["data"]["image"])
-        assert list(a.getdata()) == list(b.getdata())
-
-    def test_candidate_config_wins_over_saved(self, client):
-        """The preview must show the UNSAVED form state, not the saved
-        config ('saved' vs 'candidate' render differently)."""
-        saved = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            json={"config": {}})
-        candidate = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            json={"config": {"message": "candidate"}})
-        a = _decode_image(saved.get_json()["data"]["image"])
-        b = _decode_image(candidate.get_json()["data"]["image"])
-        assert list(a.getdata()) != list(b.getdata())
-
-    def test_disabled_plugin_still_previews(self, client):
-        """Saved config has enabled: False — preview forces enabled."""
-        resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
-                           json={"config": {}})
-        assert resp.status_code == 200
-        assert resp.get_json()["data"]["errors"] == []
-
-    def test_preview_size_form_field(self, client):
-        """The UI size selector posts __preview_size=WxH via hx-vals (htmx
-        caches hx-post's path, so it can't ride the query string). It must
-        set the render size and must NOT leak into the candidate config."""
-        resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
-                           data={"message": "hi", "__preview_size": "64x64"})
-        assert resp.status_code == 200
-        data = resp.get_json()["data"]
-        img = _decode_image(data["image"])
-        assert img.size == (64, 64)
-        assert data["errors"] == []
-
-    def test_query_args_beat_preview_size_field(self, client):
-        resp = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            data={"message": "hi", "__preview_size": "64x64"})
-        img = _decode_image(resp.get_json()["data"]["image"])
-        assert img.size == (128, 32)
-
-    def test_malformed_preview_size_falls_back_to_panel(self, client):
-        resp = client.post(f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}",
-                           data={"message": "hi", "__preview_size": "bogus x"})
-        img = _decode_image(resp.get_json()["data"]["image"])
-        assert img.size == (128, 32)  # cols*chain x rows*parallel
-
-    def test_json_candidate_deep_merges_onto_saved_config(self, client):
-        """A partial JSON candidate must not wipe saved sibling values in
-        the same nested section (form path and save both deep-merge)."""
-        # Saved config has message "saved"; posting an unrelated nested key
-        # must not discard it — render must still differ from a candidate
-        # that explicitly changes message.
-        keep_saved = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            json={"config": {}})
-        explicit = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=128&height=32",
-            json={"config": {"message": "saved"}})
-        a = _decode_image(keep_saved.get_json()["data"]["image"])
-        b = _decode_image(explicit.get_json()["data"]["image"])
-        assert list(a.getdata()) == list(b.getdata())
-
-    def test_hanging_plugin_times_out(self, client, plugin_dir, monkeypatch):
-        """A plugin whose display() hangs must not pin the web worker.
-
-        Uses its own plugin id: the loader caches the module per id, so
-        reusing PLUGIN_ID would run the already-imported (non-hanging) code
-        when this test follows others in the suite.
-        """
-        from web_interface.blueprints import api_v3 as api_v3_module
-        monkeypatch.setattr(api_v3_module, "PREVIEW_RENDER_TIMEOUT_SEC", 1)
-        hang_id = "preview-hang-plugin"
-        hang_dir = plugin_dir.parent / hang_id
-        hang_dir.mkdir()
-        (hang_dir / "manager.py").write_text(MANAGER_PY.replace(
-            "self.display_manager.update_display()",
-            "import time; time.sleep(10); self.display_manager.update_display()"))
-        manifest = dict(MANIFEST, id=hang_id, name="Hang Plugin")
-        (hang_dir / "manifest.json").write_text(json.dumps(manifest))
-        (hang_dir / "config_schema.json").write_text(json.dumps(SCHEMA))
-        resp = client.post(f"/api/v3/plugins/preview?plugin_id={hang_id}",
-                           json={"config": {}})
-        assert resp.status_code == 504
-
-    def test_htmx_gets_html_fragment(self, client):
-        resp = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=64&height=32",
-            data={"message": "hi"}, headers={"HX-Request": "true"})
-        assert resp.status_code == 200
-        assert resp.mimetype == "text/html"
-        body = resp.get_data(as_text=True)
-        assert "<img" in body and "data:image/png;base64," in body
-
-    def test_unknown_plugin_404(self, client):
-        resp = client.post("/api/v3/plugins/preview?plugin_id=nope",
-                           json={"config": {}})
-        assert resp.status_code == 404
-
-    def test_missing_plugin_id_400(self, client):
-        resp = client.post("/api/v3/plugins/preview", json={"config": {}})
-        assert resp.status_code == 400
-
-    def test_absurd_size_rejected(self, client):
-        resp = client.post(
-            f"/api/v3/plugins/preview?plugin_id={PLUGIN_ID}&width=99999&height=32",
-            json={"config": {}})
-        assert resp.status_code == 400
-
-
-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)

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
@@ -4282,373 +4283,6 @@ def _filter_config_by_schema(config, schema, prefix=''):
     return filtered
 
 
-def parse_plugin_config_form(form, schema, existing_config):
-    """Convert an HTMX config-form submission into a nested plugin config.
-
-    form is the werkzeug form MultiDict; existing_config is the saved
-    config to merge updates onto (mutated and returned). Handles dotted
-    field names, bracket/indexed array fields (color pickers), schema-
-    driven type coercion, and unchecked-checkbox fixup.
-
-    Shared by save_plugin_config and the plugin preview endpoint so the
-    two interpretations of the form can never drift apart."""
-    plugin_config = existing_config
-    # Convert form data to config dict
-    # Form fields can use dot notation for nested values (e.g., "transition.type")
-    form_data = form.to_dict()
-
-    # First pass: handle bracket notation array fields (e.g., "field_name[]" from checkbox-group)
-    # These fields use getlist() to preserve all values, then replace in form_data
-    # Sentinel empty value ("") allows clearing array to [] when all checkboxes unchecked
-    bracket_array_fields = {}  # Maps base field path to list of values
-    for key in form.keys():
-        # Check if key ends with "[]" (bracket notation for array fields)
-        if key.endswith('[]'):
-            base_path = key[:-2]  # Remove "[]" suffix
-            values = form.getlist(key)
-            # Filter out sentinel empty string - if only sentinel present, array should be []
-            # If sentinel + values present, use the actual values
-            filtered_values = [v for v in values if v and v.strip()]
-            # If no non-empty values but key exists, it means all checkboxes unchecked (empty array)
-            bracket_array_fields[base_path] = filtered_values
-            # Remove the bracket notation key from form_data if present
-            if key in form_data:
-                del form_data[key]
-    
-    # Process bracket notation fields and set directly in plugin_config
-    # Use JSON encoding instead of comma-join to handle values containing commas
-    import json
-    for base_path, values in bracket_array_fields.items():
-        # Get schema property to verify it's an array
-        base_prop = _get_schema_property(schema, base_path)
-        if base_prop and base_prop.get('type') == 'array':
-            # Filter out empty values and sentinel empty strings
-            filtered_values = [v for v in values if v and v.strip()]
-            # Set directly in plugin_config (values are already strings, no need to parse)
-            # Empty array (all unchecked) is represented as []
-            _set_nested_value(plugin_config, base_path, filtered_values)
-            logger.debug(f"Processed bracket notation array field {base_path}: {values} -> {filtered_values}")
-            # Remove from form_data to avoid double processing
-            if base_path in form_data:
-                del form_data[base_path]
-
-    # Second pass: detect and combine array index fields (e.g., "text_color.0", "text_color.1" -> "text_color" as array)
-    # This handles cases where forms send array fields as indexed inputs
-    array_fields = {}  # Maps base field path to list of (index, value) tuples
-    processed_keys = set()
-    indexed_base_paths = set()  # Track which base paths have indexed fields
-
-    for key, value in form_data.items():
-        # Check if this looks like an array index field (ends with .0, .1, .2, etc.)
-        if '.' in key:
-            parts = key.rsplit('.', 1)  # Split on last dot
-            if len(parts) == 2:
-                base_path, last_part = parts
-                # Check if last part is a numeric string (array index)
-                if last_part.isdigit():
-                    # Get schema property for the base path to verify it's an array
-                    base_prop = _get_schema_property(schema, base_path)
-                    if base_prop and base_prop.get('type') == 'array':
-                        # This is an array index field
-                        index = int(last_part)
-                        if base_path not in array_fields:
-                            array_fields[base_path] = []
-                        array_fields[base_path].append((index, value))
-                        processed_keys.add(key)
-                        indexed_base_paths.add(base_path)
-                        continue
-
-    # Process combined array fields
-    for base_path, index_values in array_fields.items():
-        # Sort by index and extract values
-        index_values.sort(key=lambda x: x[0])
-        values = [v for _, v in index_values]
-        # Combine values into comma-separated string for parsing
-        combined_value = ', '.join(str(v) for v in values)
-        # Parse as array using schema
-        parsed_value = _parse_form_value_with_schema(combined_value, base_path, schema)
-        # Debug logging
-        logger.debug(f"Combined indexed array field {base_path}: {values} -> {combined_value} -> {parsed_value}")
-        # Only set if not skipped
-        if parsed_value is not _SKIP_FIELD:
-            _set_nested_value(plugin_config, base_path, parsed_value)
-    
-    # Process remaining (non-indexed) fields
-    # Skip any base paths that were processed as indexed arrays
-    for key, value in form_data.items():
-        if key not in processed_keys:
-            # Skip if this key is a base path that was processed as indexed array
-            # (to avoid overwriting the combined array with a single value)
-            if key not in indexed_base_paths:
-                # Parse value using schema to determine correct type
-                parsed_value = _parse_form_value_with_schema(value, key, schema)
-                # Debug logging for array fields
-                if schema:
-                    prop = _get_schema_property(schema, key)
-                    if prop and prop.get('type') == 'array':
-                        logger.debug(f"Array field {key}: form value='{value}' -> parsed={parsed_value}")
-                # Use helper to set nested values correctly (skips if _SKIP_FIELD)
-                if parsed_value is not _SKIP_FIELD:
-                    _set_nested_value(plugin_config, key, parsed_value)
-    
-    # Post-process: Fix array fields that might have been incorrectly structured
-    # This handles cases where array fields are stored as dicts (e.g., from indexed form fields)
-    def fix_array_structures(config_dict, schema_props, prefix=''):
-        """Recursively fix array structures (convert dicts with numeric keys to arrays, fix length issues)"""
-        for prop_key, prop_schema in schema_props.items():
-            prop_type = prop_schema.get('type')
-
-            if prop_type == 'array':
-                # Navigate to the field location
-                if prefix:
-                    parent_parts = prefix.split('.')
-                    parent = config_dict
-                    for part in parent_parts:
-                        if isinstance(parent, dict) and part in parent:
-                            parent = parent[part]
-                        else:
-                            parent = None
-                            break
-
-                    if parent is not None and isinstance(parent, dict) and prop_key in parent:
-                        current_value = parent[prop_key]
-                        # If it's a dict with numeric string keys, convert to array
-                        if isinstance(current_value, dict) and not isinstance(current_value, list):
-                            try:
-                                # Check if all keys are numeric strings (array indices)
-                                keys = [k for k in current_value.keys()]
-                                if all(k.isdigit() for k in keys):
-                                    # Convert to sorted array by index
-                                    sorted_keys = sorted(keys, key=int)
-                                    array_value = [current_value[k] for k in sorted_keys]
-                                    # Convert array elements to correct types based on schema
-                                    items_schema = prop_schema.get('items', {})
-                                    item_type = items_schema.get('type')
-                                    if item_type in ('number', 'integer'):
-                                        converted_array = []
-                                        for v in array_value:
-                                            if isinstance(v, str):
-                                                try:
-                                                    if item_type == 'integer':
-                                                        converted_array.append(int(v))
-                                                    else:
-                                                        converted_array.append(float(v))
-                                                except (ValueError, TypeError):
-                                                    converted_array.append(v)
-                                            else:
-                                                converted_array.append(v)
-                                        array_value = converted_array
-                                    parent[prop_key] = array_value
-                                    current_value = array_value  # Update for length check below
-                            except (ValueError, KeyError, TypeError):
-                                # Conversion failed, check if we should use default
-                                pass
-
-                        # If it's an array, ensure correct types and check minItems
-                        if isinstance(current_value, list):
-                            # First, ensure array elements are correct types
-                            items_schema = prop_schema.get('items', {})
-                            item_type = items_schema.get('type')
-                            if item_type in ('number', 'integer'):
-                                converted_array = []
-                                for v in current_value:
-                                    if isinstance(v, str):
-                                        try:
-                                            if item_type == 'integer':
-                                                converted_array.append(int(v))
-                                            else:
-                                                converted_array.append(float(v))
-                                        except (ValueError, TypeError):
-                                            converted_array.append(v)
-                                    else:
-                                        converted_array.append(v)
-                                parent[prop_key] = converted_array
-                                current_value = converted_array
-
-                            # Then check minItems
-                            min_items = prop_schema.get('minItems')
-                            if min_items is not None and len(current_value) < min_items:
-                                # Use default if available, otherwise keep as-is (validation will catch it)
-                                default = prop_schema.get('default')
-                                if default and isinstance(default, list) and len(default) >= min_items:
-                                    parent[prop_key] = default
-                else:
-                    # Top-level field
-                    if prop_key in config_dict:
-                        current_value = config_dict[prop_key]
-                        # If it's a dict with numeric string keys, convert to array
-                        if isinstance(current_value, dict) and not isinstance(current_value, list):
-                            try:
-                                keys = list(current_value.keys())
-                                if keys and all(str(k).isdigit() for k in keys):
-                                    sorted_keys = sorted(keys, key=lambda x: int(str(x)))
-                                    array_value = [current_value[k] for k in sorted_keys]
-                                    # Convert array elements to correct types based on schema
-                                    items_schema = prop_schema.get('items', {})
-                                    item_type = items_schema.get('type')
-                                    if item_type in ('number', 'integer'):
-                                        converted_array = []
-                                        for v in array_value:
-                                            if isinstance(v, str):
-                                                try:
-                                                    if item_type == 'integer':
-                                                        converted_array.append(int(v))
-                                                    else:
-                                                        converted_array.append(float(v))
-                                                except (ValueError, TypeError):
-                                                    converted_array.append(v)
-                                            else:
-                                                converted_array.append(v)
-                                        array_value = converted_array
-                                    config_dict[prop_key] = array_value
-                                    current_value = array_value  # Update for length check below
-                            except (ValueError, KeyError, TypeError) as e:
-                                logger.debug(f"Failed to convert {prop_key} to array: {e}")
-
-                        # If it's an array, ensure correct types and check minItems
-                        if isinstance(current_value, list):
-                            # First, ensure array elements are correct types
-                            items_schema = prop_schema.get('items', {})
-                            item_type = items_schema.get('type')
-                            if item_type in ('number', 'integer'):
-                                converted_array = []
-                                for v in current_value:
-                                    if isinstance(v, str):
-                                        try:
-                                            if item_type == 'integer':
-                                                converted_array.append(int(v))
-                                            else:
-                                                converted_array.append(float(v))
-                                        except (ValueError, TypeError):
-                                            converted_array.append(v)
-                                    else:
-                                        converted_array.append(v)
-                                config_dict[prop_key] = converted_array
-                                current_value = converted_array
-
-                            # Then check minItems
-                            min_items = prop_schema.get('minItems')
-                            if min_items is not None and len(current_value) < min_items:
-                                default = prop_schema.get('default')
-                                if default and isinstance(default, list) and len(default) >= min_items:
-                                    config_dict[prop_key] = default
-
-            # Recurse into nested objects
-            elif prop_type == 'object' and 'properties' in prop_schema:
-                nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
-                if prefix:
-                    parent_parts = prefix.split('.')
-                    parent = config_dict
-                    for part in parent_parts:
-                        if isinstance(parent, dict) and part in parent:
-                            parent = parent[part]
-                        else:
-                            parent = None
-                            break
-                    nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
-                else:
-                    nested_dict = config_dict.get(prop_key)
-
-                if isinstance(nested_dict, dict):
-                    # Pass no prefix: config_dict is already the navigated sub-dict,
-                    # so path segments from the parent would mis-navigate it.
-                    fix_array_structures(nested_dict, prop_schema['properties'])
-
-    # Also ensure array fields that are None get converted to empty arrays
-    def ensure_array_defaults(config_dict, schema_props, prefix=''):
-        """Recursively ensure array fields have defaults if None"""
-        for prop_key, prop_schema in schema_props.items():
-            prop_type = prop_schema.get('type')
-
-            if prop_type == 'array':
-                if prefix:
-                    parent_parts = prefix.split('.')
-                    parent = config_dict
-                    for part in parent_parts:
-                        if isinstance(parent, dict) and part in parent:
-                            parent = parent[part]
-                        else:
-                            parent = None
-                            break
-
-                    if parent is not None and isinstance(parent, dict):
-                        if prop_key not in parent or parent[prop_key] is None:
-                            default = prop_schema.get('default', [])
-                            parent[prop_key] = default if default else []
-                else:
-                    if prop_key not in config_dict or config_dict[prop_key] is None:
-                        default = prop_schema.get('default', [])
-                        config_dict[prop_key] = default if default else []
-
-            elif prop_type == 'object' and 'properties' in prop_schema:
-                nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
-                if prefix:
-                    parent_parts = prefix.split('.')
-                    parent = config_dict
-                    for part in parent_parts:
-                        if isinstance(parent, dict) and part in parent:
-                            parent = parent[part]
-                        else:
-                            parent = None
-                            break
-                    nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
-                else:
-                    nested_dict = config_dict.get(prop_key)
-
-                if nested_dict is None:
-                    if prefix:
-                        parent_parts = prefix.split('.')
-                        parent = config_dict
-                        for part in parent_parts:
-                            if part not in parent:
-                                parent[part] = {}
-                            parent = parent[part]
-                        if prop_key not in parent:
-                            parent[prop_key] = {}
-                        nested_dict = parent[prop_key]
-                    else:
-                        if prop_key not in config_dict:
-                            config_dict[prop_key] = {}
-                        nested_dict = config_dict[prop_key]
-
-                if isinstance(nested_dict, dict):
-                    # Pass no prefix: config_dict is already navigated.
-                    ensure_array_defaults(nested_dict, prop_schema['properties'])
-
-    if schema and 'properties' in schema:
-        # First, fix any dict structures that should be arrays
-        # This must be called BEFORE validation to convert dicts with numeric keys to arrays
-        fix_array_structures(plugin_config, schema['properties'])
-        # Then, ensure None arrays get defaults
-        ensure_array_defaults(plugin_config, schema['properties'])
-        
-        # Debug: Log the structure after fixing
-        if 'feeds' in plugin_config and 'custom_feeds' in plugin_config.get('feeds', {}):
-            custom_feeds = plugin_config['feeds']['custom_feeds']
-            logger.debug(f"After fix_array_structures: custom_feeds type={type(custom_feeds)}, value={custom_feeds}")
-        
-        # Force fix for feeds.custom_feeds if it's still a dict (fallback)
-        if 'feeds' in plugin_config:
-            feeds_config = plugin_config.get('feeds') or {}
-            if feeds_config and 'custom_feeds' in feeds_config and isinstance(feeds_config['custom_feeds'], dict):
-                custom_feeds_dict = feeds_config['custom_feeds']
-                # Check if all keys are numeric
-                keys = list(custom_feeds_dict.keys())
-                if keys and all(str(k).isdigit() for k in keys):
-                    # Convert to array
-                    sorted_keys = sorted(keys, key=lambda x: int(str(x)))
-                    feeds_config['custom_feeds'] = [custom_feeds_dict[k] for k in sorted_keys]
-                    logger.info(f"Force-converted feeds.custom_feeds from dict to array: {len(feeds_config['custom_feeds'])} items")
-
-    # Fix unchecked boolean checkboxes: HTML checkboxes don't submit values
-    # when unchecked, so the existing config value (potentially True) persists.
-    # Walk the schema and set any boolean fields missing from form data to False.
-    if schema and 'properties' in schema:
-        form_keys = set(form.keys())
-        _set_missing_booleans_to_false(plugin_config, schema['properties'], form_keys)
-    return plugin_config
-
-
 @api_v3.route('/plugins/config', methods=['POST'])
 def save_plugin_config():
     """Save plugin configuration, separating secrets from regular config"""
@@ -4702,9 +4336,359 @@ def save_plugin_config():
             # Start with existing config and apply form updates
             plugin_config = existing_config
 
-            # Convert form data to config dict (shared with the preview
-            # endpoint — see parse_plugin_config_form)
-            plugin_config = parse_plugin_config_form(request.form, schema, plugin_config)
+            # Convert form data to config dict
+            # Form fields can use dot notation for nested values (e.g., "transition.type")
+            form_data = request.form.to_dict()
+
+            # First pass: handle bracket notation array fields (e.g., "field_name[]" from checkbox-group)
+            # These fields use getlist() to preserve all values, then replace in form_data
+            # Sentinel empty value ("") allows clearing array to [] when all checkboxes unchecked
+            bracket_array_fields = {}  # Maps base field path to list of values
+            for key in request.form.keys():
+                # Check if key ends with "[]" (bracket notation for array fields)
+                if key.endswith('[]'):
+                    base_path = key[:-2]  # Remove "[]" suffix
+                    values = request.form.getlist(key)
+                    # Filter out sentinel empty string - if only sentinel present, array should be []
+                    # If sentinel + values present, use the actual values
+                    filtered_values = [v for v in values if v and v.strip()]
+                    # If no non-empty values but key exists, it means all checkboxes unchecked (empty array)
+                    bracket_array_fields[base_path] = filtered_values
+                    # Remove the bracket notation key from form_data if present
+                    if key in form_data:
+                        del form_data[key]
+            
+            # Process bracket notation fields and set directly in plugin_config
+            # Use JSON encoding instead of comma-join to handle values containing commas
+            import json
+            for base_path, values in bracket_array_fields.items():
+                # Get schema property to verify it's an array
+                base_prop = _get_schema_property(schema, base_path)
+                if base_prop and base_prop.get('type') == 'array':
+                    # Filter out empty values and sentinel empty strings
+                    filtered_values = [v for v in values if v and v.strip()]
+                    # Set directly in plugin_config (values are already strings, no need to parse)
+                    # Empty array (all unchecked) is represented as []
+                    _set_nested_value(plugin_config, base_path, filtered_values)
+                    logger.debug(f"Processed bracket notation array field {base_path}: {values} -> {filtered_values}")
+                    # Remove from form_data to avoid double processing
+                    if base_path in form_data:
+                        del form_data[base_path]
+
+            # Second pass: detect and combine array index fields (e.g., "text_color.0", "text_color.1" -> "text_color" as array)
+            # This handles cases where forms send array fields as indexed inputs
+            array_fields = {}  # Maps base field path to list of (index, value) tuples
+            processed_keys = set()
+            indexed_base_paths = set()  # Track which base paths have indexed fields
+
+            for key, value in form_data.items():
+                # Check if this looks like an array index field (ends with .0, .1, .2, etc.)
+                if '.' in key:
+                    parts = key.rsplit('.', 1)  # Split on last dot
+                    if len(parts) == 2:
+                        base_path, last_part = parts
+                        # Check if last part is a numeric string (array index)
+                        if last_part.isdigit():
+                            # Get schema property for the base path to verify it's an array
+                            base_prop = _get_schema_property(schema, base_path)
+                            if base_prop and base_prop.get('type') == 'array':
+                                # This is an array index field
+                                index = int(last_part)
+                                if base_path not in array_fields:
+                                    array_fields[base_path] = []
+                                array_fields[base_path].append((index, value))
+                                processed_keys.add(key)
+                                indexed_base_paths.add(base_path)
+                                continue
+
+            # Process combined array fields
+            for base_path, index_values in array_fields.items():
+                # Sort by index and extract values
+                index_values.sort(key=lambda x: x[0])
+                values = [v for _, v in index_values]
+                # Combine values into comma-separated string for parsing
+                combined_value = ', '.join(str(v) for v in values)
+                # Parse as array using schema
+                parsed_value = _parse_form_value_with_schema(combined_value, base_path, schema)
+                # Debug logging
+                logger.debug(f"Combined indexed array field {base_path}: {values} -> {combined_value} -> {parsed_value}")
+                # Only set if not skipped
+                if parsed_value is not _SKIP_FIELD:
+                    _set_nested_value(plugin_config, base_path, parsed_value)
+            
+            # Process remaining (non-indexed) fields
+            # Skip any base paths that were processed as indexed arrays
+            for key, value in form_data.items():
+                if key not in processed_keys:
+                    # Skip if this key is a base path that was processed as indexed array
+                    # (to avoid overwriting the combined array with a single value)
+                    if key not in indexed_base_paths:
+                        # Parse value using schema to determine correct type
+                        parsed_value = _parse_form_value_with_schema(value, key, schema)
+                        # Debug logging for array fields
+                        if schema:
+                            prop = _get_schema_property(schema, key)
+                            if prop and prop.get('type') == 'array':
+                                logger.debug(f"Array field {key}: form value='{value}' -> parsed={parsed_value}")
+                        # Use helper to set nested values correctly (skips if _SKIP_FIELD)
+                        if parsed_value is not _SKIP_FIELD:
+                            _set_nested_value(plugin_config, key, parsed_value)
+            
+            # Post-process: Fix array fields that might have been incorrectly structured
+            # This handles cases where array fields are stored as dicts (e.g., from indexed form fields)
+            def fix_array_structures(config_dict, schema_props, prefix=''):
+                """Recursively fix array structures (convert dicts with numeric keys to arrays, fix length issues)"""
+                for prop_key, prop_schema in schema_props.items():
+                    prop_type = prop_schema.get('type')
+
+                    if prop_type == 'array':
+                        # Navigate to the field location
+                        if prefix:
+                            parent_parts = prefix.split('.')
+                            parent = config_dict
+                            for part in parent_parts:
+                                if isinstance(parent, dict) and part in parent:
+                                    parent = parent[part]
+                                else:
+                                    parent = None
+                                    break
+
+                            if parent is not None and isinstance(parent, dict) and prop_key in parent:
+                                current_value = parent[prop_key]
+                                # If it's a dict with numeric string keys, convert to array
+                                if isinstance(current_value, dict) and not isinstance(current_value, list):
+                                    try:
+                                        # Check if all keys are numeric strings (array indices)
+                                        keys = [k for k in current_value.keys()]
+                                        if all(k.isdigit() for k in keys):
+                                            # Convert to sorted array by index
+                                            sorted_keys = sorted(keys, key=int)
+                                            array_value = [current_value[k] for k in sorted_keys]
+                                            # Convert array elements to correct types based on schema
+                                            items_schema = prop_schema.get('items', {})
+                                            item_type = items_schema.get('type')
+                                            if item_type in ('number', 'integer'):
+                                                converted_array = []
+                                                for v in array_value:
+                                                    if isinstance(v, str):
+                                                        try:
+                                                            if item_type == 'integer':
+                                                                converted_array.append(int(v))
+                                                            else:
+                                                                converted_array.append(float(v))
+                                                        except (ValueError, TypeError):
+                                                            converted_array.append(v)
+                                                    else:
+                                                        converted_array.append(v)
+                                                array_value = converted_array
+                                            parent[prop_key] = array_value
+                                            current_value = array_value  # Update for length check below
+                                    except (ValueError, KeyError, TypeError):
+                                        # Conversion failed, check if we should use default
+                                        pass
+
+                                # If it's an array, ensure correct types and check minItems
+                                if isinstance(current_value, list):
+                                    # First, ensure array elements are correct types
+                                    items_schema = prop_schema.get('items', {})
+                                    item_type = items_schema.get('type')
+                                    if item_type in ('number', 'integer'):
+                                        converted_array = []
+                                        for v in current_value:
+                                            if isinstance(v, str):
+                                                try:
+                                                    if item_type == 'integer':
+                                                        converted_array.append(int(v))
+                                                    else:
+                                                        converted_array.append(float(v))
+                                                except (ValueError, TypeError):
+                                                    converted_array.append(v)
+                                            else:
+                                                converted_array.append(v)
+                                        parent[prop_key] = converted_array
+                                        current_value = converted_array
+
+                                    # Then check minItems
+                                    min_items = prop_schema.get('minItems')
+                                    if min_items is not None and len(current_value) < min_items:
+                                        # Use default if available, otherwise keep as-is (validation will catch it)
+                                        default = prop_schema.get('default')
+                                        if default and isinstance(default, list) and len(default) >= min_items:
+                                            parent[prop_key] = default
+                        else:
+                            # Top-level field
+                            if prop_key in config_dict:
+                                current_value = config_dict[prop_key]
+                                # If it's a dict with numeric string keys, convert to array
+                                if isinstance(current_value, dict) and not isinstance(current_value, list):
+                                    try:
+                                        keys = list(current_value.keys())
+                                        if keys and all(str(k).isdigit() for k in keys):
+                                            sorted_keys = sorted(keys, key=lambda x: int(str(x)))
+                                            array_value = [current_value[k] for k in sorted_keys]
+                                            # Convert array elements to correct types based on schema
+                                            items_schema = prop_schema.get('items', {})
+                                            item_type = items_schema.get('type')
+                                            if item_type in ('number', 'integer'):
+                                                converted_array = []
+                                                for v in array_value:
+                                                    if isinstance(v, str):
+                                                        try:
+                                                            if item_type == 'integer':
+                                                                converted_array.append(int(v))
+                                                            else:
+                                                                converted_array.append(float(v))
+                                                        except (ValueError, TypeError):
+                                                            converted_array.append(v)
+                                                    else:
+                                                        converted_array.append(v)
+                                                array_value = converted_array
+                                            config_dict[prop_key] = array_value
+                                            current_value = array_value  # Update for length check below
+                                    except (ValueError, KeyError, TypeError) as e:
+                                        logger.debug(f"Failed to convert {prop_key} to array: {e}")
+
+                                # If it's an array, ensure correct types and check minItems
+                                if isinstance(current_value, list):
+                                    # First, ensure array elements are correct types
+                                    items_schema = prop_schema.get('items', {})
+                                    item_type = items_schema.get('type')
+                                    if item_type in ('number', 'integer'):
+                                        converted_array = []
+                                        for v in current_value:
+                                            if isinstance(v, str):
+                                                try:
+                                                    if item_type == 'integer':
+                                                        converted_array.append(int(v))
+                                                    else:
+                                                        converted_array.append(float(v))
+                                                except (ValueError, TypeError):
+                                                    converted_array.append(v)
+                                            else:
+                                                converted_array.append(v)
+                                        config_dict[prop_key] = converted_array
+                                        current_value = converted_array
+
+                                    # Then check minItems
+                                    min_items = prop_schema.get('minItems')
+                                    if min_items is not None and len(current_value) < min_items:
+                                        default = prop_schema.get('default')
+                                        if default and isinstance(default, list) and len(default) >= min_items:
+                                            config_dict[prop_key] = default
+
+                    # Recurse into nested objects
+                    elif prop_type == 'object' and 'properties' in prop_schema:
+                        nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
+                        if prefix:
+                            parent_parts = prefix.split('.')
+                            parent = config_dict
+                            for part in parent_parts:
+                                if isinstance(parent, dict) and part in parent:
+                                    parent = parent[part]
+                                else:
+                                    parent = None
+                                    break
+                            nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
+                        else:
+                            nested_dict = config_dict.get(prop_key)
+
+                        if isinstance(nested_dict, dict):
+                            # Pass no prefix: config_dict is already the navigated sub-dict,
+                            # so path segments from the parent would mis-navigate it.
+                            fix_array_structures(nested_dict, prop_schema['properties'])
+
+            # Also ensure array fields that are None get converted to empty arrays
+            def ensure_array_defaults(config_dict, schema_props, prefix=''):
+                """Recursively ensure array fields have defaults if None"""
+                for prop_key, prop_schema in schema_props.items():
+                    prop_type = prop_schema.get('type')
+
+                    if prop_type == 'array':
+                        if prefix:
+                            parent_parts = prefix.split('.')
+                            parent = config_dict
+                            for part in parent_parts:
+                                if isinstance(parent, dict) and part in parent:
+                                    parent = parent[part]
+                                else:
+                                    parent = None
+                                    break
+
+                            if parent is not None and isinstance(parent, dict):
+                                if prop_key not in parent or parent[prop_key] is None:
+                                    default = prop_schema.get('default', [])
+                                    parent[prop_key] = default if default else []
+                        else:
+                            if prop_key not in config_dict or config_dict[prop_key] is None:
+                                default = prop_schema.get('default', [])
+                                config_dict[prop_key] = default if default else []
+
+                    elif prop_type == 'object' and 'properties' in prop_schema:
+                        nested_prefix = f"{prefix}.{prop_key}" if prefix else prop_key
+                        if prefix:
+                            parent_parts = prefix.split('.')
+                            parent = config_dict
+                            for part in parent_parts:
+                                if isinstance(parent, dict) and part in parent:
+                                    parent = parent[part]
+                                else:
+                                    parent = None
+                                    break
+                            nested_dict = parent.get(prop_key) if parent is not None and isinstance(parent, dict) else None
+                        else:
+                            nested_dict = config_dict.get(prop_key)
+
+                        if nested_dict is None:
+                            if prefix:
+                                parent_parts = prefix.split('.')
+                                parent = config_dict
+                                for part in parent_parts:
+                                    if part not in parent:
+                                        parent[part] = {}
+                                    parent = parent[part]
+                                if prop_key not in parent:
+                                    parent[prop_key] = {}
+                                nested_dict = parent[prop_key]
+                            else:
+                                if prop_key not in config_dict:
+                                    config_dict[prop_key] = {}
+                                nested_dict = config_dict[prop_key]
+
+                        if isinstance(nested_dict, dict):
+                            # Pass no prefix: config_dict is already navigated.
+                            ensure_array_defaults(nested_dict, prop_schema['properties'])
+
+            if schema and 'properties' in schema:
+                # First, fix any dict structures that should be arrays
+                # This must be called BEFORE validation to convert dicts with numeric keys to arrays
+                fix_array_structures(plugin_config, schema['properties'])
+                # Then, ensure None arrays get defaults
+                ensure_array_defaults(plugin_config, schema['properties'])
+                
+                # Debug: Log the structure after fixing
+                if 'feeds' in plugin_config and 'custom_feeds' in plugin_config.get('feeds', {}):
+                    custom_feeds = plugin_config['feeds']['custom_feeds']
+                    logger.debug(f"After fix_array_structures: custom_feeds type={type(custom_feeds)}, value={custom_feeds}")
+                
+                # Force fix for feeds.custom_feeds if it's still a dict (fallback)
+                if 'feeds' in plugin_config:
+                    feeds_config = plugin_config.get('feeds') or {}
+                    if feeds_config and 'custom_feeds' in feeds_config and isinstance(feeds_config['custom_feeds'], dict):
+                        custom_feeds_dict = feeds_config['custom_feeds']
+                        # Check if all keys are numeric
+                        keys = list(custom_feeds_dict.keys())
+                        if keys and all(str(k).isdigit() for k in keys):
+                            # Convert to array
+                            sorted_keys = sorted(keys, key=lambda x: int(str(x)))
+                            feeds_config['custom_feeds'] = [custom_feeds_dict[k] for k in sorted_keys]
+                            logger.info(f"Force-converted feeds.custom_feeds from dict to array: {len(feeds_config['custom_feeds'])} items")
+
+            # Fix unchecked boolean checkboxes: HTML checkboxes don't submit values
+            # when unchecked, so the existing config value (potentially True) persists.
+            # Walk the schema and set any boolean fields missing from form data to False.
+            if schema and 'properties' in schema:
+                form_keys = set(request.form.keys())
+                _set_missing_booleans_to_false(plugin_config, schema['properties'], form_keys)
 
         # Get schema manager instance (for JSON requests)
         schema_mgr = api_v3.schema_manager
@@ -5306,233 +5290,6 @@ def get_plugin_schema():
         logger.error('Error in get_plugin_schema', exc_info=True)
         return jsonify({'status': 'error', 'message': 'An error occurred; see logs for details'}), 500
 
-
-# plugin_id arrives in request input and is used to build filesystem paths —
-# allowlist it (same pattern pages_v3 uses)
-_SAFE_PREVIEW_PLUGIN_ID_RE = re.compile(r'^[a-zA-Z0-9_-]{1,64}$')
-
-
-def _find_plugin_dir_for_preview(plugin_id: str) -> 'Path | None':
-    """Locate an installed plugin's directory (store dir, then dev dirs) —
-    same search order the schema manager uses. Rejects any id that could
-    name a path outside the plugin directories."""
-    if not isinstance(plugin_id, str) or not _SAFE_PREVIEW_PLUGIN_ID_RE.match(plugin_id):
-        return None
-    candidates = []
-    active_pm = getattr(api_v3, 'plugin_manager', None)
-    if active_pm and getattr(active_pm, 'plugins_dir', None):
-        candidates.append(Path(active_pm.plugins_dir))
-    else:
-        _cm = getattr(api_v3, 'config_manager', None)
-        _cfg = _cm.load_config() if _cm else {}
-        _dir_name = _cfg.get('plugin_system', {}).get('plugins_directory', 'plugin-repos')
-        candidates.append(Path(_dir_name) if os.path.isabs(_dir_name)
-                          else PROJECT_ROOT / _dir_name)
-    candidates.append(PROJECT_ROOT / 'plugins')
-    candidates.append(PROJECT_ROOT / 'plugin-repos')
-    for base in candidates:
-        plugin_dir = base / plugin_id
-        if (plugin_dir / 'manifest.json').exists():
-            return plugin_dir
-    return None
-
-
-PREVIEW_MIN_SIZE, PREVIEW_MAX_W, PREVIEW_MAX_H = 8, 1024, 512
-PREVIEW_RENDER_TIMEOUT_SEC = 15
-
-
-@api_v3.route('/plugins/preview', methods=['POST'])
-def preview_plugin_render():
-    """Render a plugin headlessly with a CANDIDATE (unsaved) config.
-
-    Powers the config page's live preview: the browser posts the current
-    form state (same encoding as save — parsed by the same
-    parse_plugin_config_form, so preview and save can never disagree) or a
-    JSON body {"config": {...}}, and gets back a base64 PNG of what the
-    panel would show.
-
-    Entirely hardware-free: renders through VisualTestDisplayManager (pure
-    PIL) with install_deps=False. update() is skipped by default so the
-    request never blocks on live APIs — plugins with a test/harness.json
-    get their mock-data fixture primed into the cache instead, and
-    ?skip_update=0 opts into a real update() for plugins that need it.
-
-    Query params: plugin_id (required); width/height (defaults: the real
-    panel size from display.hardware); skip_update (default 1).
-    """
-    try:
-        plugin_id = request.args.get('plugin_id')
-        if not plugin_id:
-            return error_response(ErrorCode.INVALID_INPUT,
-                                  'plugin_id required in query string',
-                                  status_code=400)
-
-        plugin_dir = _find_plugin_dir_for_preview(plugin_id)
-        if not plugin_dir:
-            return error_response(ErrorCode.PLUGIN_NOT_FOUND,
-                                  f'Plugin not found: {plugin_id}',
-                                  status_code=404)
-
-        schema_mgr = api_v3.schema_manager
-        if not schema_mgr:
-            return error_response(ErrorCode.SYSTEM_ERROR,
-                                  'Schema manager not initialized',
-                                  status_code=500)
-
-        # ---- panel size: explicit query args, else the real panel ----
-        main_config = {}
-        if api_v3.config_manager:
-            try:
-                main_config = api_v3.config_manager.load_config() or {}
-            except Exception:
-                main_config = {}
-        hardware = main_config.get('display', {}).get('hardware', {})
-        default_width = int(hardware.get('cols', 64)) * int(hardware.get('chain_length', 2))
-        default_height = int(hardware.get('rows', 32)) * int(hardware.get('parallel', 1))
-        # The UI's size selector posts "__preview_size=WxH" via the button's
-        # hx-vals (evaluated at request time — htmx caches hx-post's path at
-        # process time, so a dynamically updated query string doesn't work).
-        # Explicit query args still take precedence for API callers.
-        preview_size = request.values.get('__preview_size', '')
-        if preview_size and 'x' in preview_size and 'width' not in request.args:
-            size_w, _, size_h = preview_size.partition('x')
-            try:
-                default_width, default_height = int(size_w), int(size_h)
-            except (TypeError, ValueError):
-                pass  # malformed selector value — fall back to panel size
-        try:
-            width = int(request.args.get('width', default_width))
-            height = int(request.args.get('height', default_height))
-        except (TypeError, ValueError):
-            return error_response(ErrorCode.INVALID_INPUT,
-                                  'width and height must be integers',
-                                  status_code=400)
-        if not (PREVIEW_MIN_SIZE <= width <= PREVIEW_MAX_W
-                and PREVIEW_MIN_SIZE <= height <= PREVIEW_MAX_H):
-            return error_response(
-                ErrorCode.INVALID_INPUT,
-                f'size must be within {PREVIEW_MIN_SIZE}x{PREVIEW_MIN_SIZE} '
-                f'and {PREVIEW_MAX_W}x{PREVIEW_MAX_H}',
-                status_code=400)
-
-        # ---- candidate config: saved config + submitted changes + defaults ----
-        schema = schema_mgr.load_schema(plugin_id, use_cache=False)
-        existing_config = (main_config.get(plugin_id) or {}).copy()
-
-        content_type = request.content_type or ''
-        if 'application/json' in content_type:
-            import copy
-            data = request.get_json(silent=True) or {}
-
-            # Deep-merge the candidate onto the saved config, matching how
-            # the form path (and save) treat partial updates — a shallow
-            # update() would silently drop the user's saved values in any
-            # nested section the candidate touches (e.g. posting one
-            # element's color would discard the saved font of its sibling).
-            def _deep_merge(base, overlay):
-                for key, value in overlay.items():
-                    if (isinstance(value, dict)
-                            and isinstance(base.get(key), dict)):
-                        _deep_merge(base[key], value)
-                    else:
-                        base[key] = value
-
-            plugin_config = copy.deepcopy(existing_config)
-            _deep_merge(plugin_config, data.get('config', {}))
-        else:
-            # Strip the preview-only control field so it never lands in the
-            # candidate config the plugin sees
-            form = request.form.copy()
-            form.poplist('__preview_size')
-            plugin_config = parse_plugin_config_form(form, schema,
-                                                     existing_config)
-
-        if schema:
-            defaults = schema_mgr.generate_default_config(plugin_id, use_cache=True)
-            plugin_config = schema_mgr.merge_with_defaults(plugin_config, defaults)
-        # Preview regardless of the enabled toggle
-        plugin_config['enabled'] = True
-
-        # ---- deterministic data: the plugin's own harness fixture ----
-        mock_data = {}
-        try:
-            from src.plugin_system.testing.loading import load_harness_spec
-            spec = load_harness_spec(plugin_dir)
-            mock_data = spec.get('mock_data_contents', {}) or {}
-            harness_config = spec.get('config') or {}
-            if harness_config:
-                # harness settings under the candidate config: user's
-                # in-form values always win
-                merged = dict(harness_config)
-                merged.update(plugin_config)
-                plugin_config = merged
-        except Exception as e:
-            logger.debug(f'No usable harness spec for {plugin_id}: {e}')
-
-        skip_update = request.args.get('skip_update', '1') not in ('0', 'false')
-
-        # Bounded render: a plugin whose update()/display() hangs must not
-        # pin a web worker forever. The runaway thread can't be killed, but
-        # the request returns and the thread is daemonized so it can't block
-        # shutdown either.
-        from src.plugin_system.testing.render_service import render_plugin_once
-        import threading
-        render_out: dict = {}
-
-        def _do_render():
-            try:
-                render_out['result'] = render_plugin_once(
-                    plugin_id, plugin_dir, config=plugin_config,
-                    mock_data=mock_data, width=width, height=height,
-                    skip_update=skip_update)
-            except Exception as e:  # surfaced below
-                render_out['error'] = e
-
-        render_thread = threading.Thread(target=_do_render, daemon=True,
-                                         name=f'preview-{plugin_id}')
-        render_thread.start()
-        render_thread.join(timeout=PREVIEW_RENDER_TIMEOUT_SEC)
-        if render_thread.is_alive():
-            logger.warning('preview render timed out for %s after %ss',
-                           plugin_id, PREVIEW_RENDER_TIMEOUT_SEC)
-            if request.headers.get('HX-Request'):
-                return Response('<p class="text-xs text-red-600">Preview timed '
-                                'out &mdash; the plugin took too long to render.</p>',
-                                mimetype='text/html')
-            return error_response(ErrorCode.SYSTEM_ERROR,
-                                  'Preview render timed out', status_code=504)
-        if 'error' in render_out:
-            raise render_out['error']
-        result = render_out['result']
-
-        # HTMX callers get a ready-to-swap fragment; API callers get JSON
-        if request.headers.get('HX-Request'):
-            import html as _html
-            meta = f"{result['width']}&times;{result['height']} &middot; {result['render_time_ms']} ms"
-            errors_html = ''
-            if result['errors'] or result['warnings']:
-                notes = _html.escape('; '.join(result['errors'] + result['warnings']))
-                errors_html = (f'<p class="text-xs text-red-600 mt-1">'
-                               f'{notes}</p>')
-            return Response(
-                f'<img src="{result["image"]}" alt="Plugin preview" '
-                f'class="preview-pixelated" '
-                f'style="image-rendering: pixelated; width: 100%; max-width: '
-                f'{result["width"] * 4}px; border: 1px solid #333; '
-                f'border-radius: 4px; background: #000;">'
-                f'<p class="text-xs text-gray-500 mt-1">{meta}</p>'
-                f'{errors_html}',
-                mimetype='text/html')
-        return jsonify({'status': 'success', 'data': result})
-    except Exception:
-        logger.error('Error in preview_plugin_render', exc_info=True)
-        if request.headers.get('HX-Request'):
-            return Response('<p class="text-xs text-red-600">Preview failed — '
-                            'see logs for details.</p>', mimetype='text/html')
-        return error_response(ErrorCode.SYSTEM_ERROR,
-                              'Preview failed; see logs for details',
-                              status_code=500)
-
 @api_v3.route('/plugins/config/reset', methods=['POST'])
 def reset_plugin_config():
     """Reset plugin configuration to schema defaults"""

web_interface/blueprints/pages_v3.py

@@ -706,12 +706,6 @@ def _load_plugin_config_partial(plugin_id):
             try:
                 with open(schema_path, 'r', encoding='utf-8') as f:
                     schema = json.load(f)
-                # Expand x-style-elements declarations into full property
-                # blocks — the same expansion SchemaManager.load_schema
-                # applies on the API paths. The form must render the exact
-                # shape the save path parses and validates against.
-                from src.element_style import expand_style_elements
-                schema = expand_style_elements(schema)
             except Exception as e:
                 logger.warning("Could not load schema for plugin: %s", e)
 

web_interface/templates/v3/partials/plugin_config.html

@@ -994,51 +994,8 @@
                     <p class="text-xs text-amber-600">Plugin is disabled, but on-demand will temporarily enable it.</p>
                     {% endif %}
                 </div>
-
-                {# Live Preview — renders the plugin headlessly with the CURRENT
-                   (unsaved) form values, at the real panel size or a chosen one #}
-                <div class="mt-4 pt-4 border-t border-gray-200 space-y-3">
-                    <div class="flex items-center gap-2">
-                        <i class="fas fa-eye text-blue-500"></i>
-                        <span class="text-sm font-semibold text-gray-900">Live Preview</span>
-                    </div>
-                    <div class="flex flex-wrap items-center gap-2">
-                        {# No name attribute: must not be submitted with the config
-                           save. The size travels via the button's hx-vals (read at
-                           request time — htmx caches hx-post's path at process
-                           time, so mutating the attribute onchange doesn't work). #}
-                        <select id="preview-size-{{ plugin.id }}"
-                                aria-label="Preview panel size"
-                                class="text-sm rounded-md border-gray-300 py-1.5">
-                            <option value="">My panel size</option>
-                            <option value="64x32">64 &times; 32</option>
-                            <option value="128x32">128 &times; 32</option>
-                            <option value="128x64">128 &times; 64</option>
-                            <option value="192x48">192 &times; 48</option>
-                            <option value="256x128">256 &times; 128</option>
-                        </select>
-                        <button type="button"
-                                id="preview-btn-{{ plugin.id }}"
-                                hx-post="/api/v3/plugins/preview?plugin_id={{ plugin.id }}"
-                                hx-include="closest form"
-                                hx-vals='js:{"__preview_size": document.getElementById("preview-size-{{ plugin.id }}").value}'
-                                hx-target="#plugin-preview-{{ plugin.id }}"
-                                hx-swap="innerHTML"
-                                hx-indicator="#preview-indicator-{{ plugin.id }}"
-                                class="px-3 py-2 text-sm bg-blue-600 hover:bg-blue-700 text-white rounded-md flex items-center gap-2 transition-colors">
-                            <i class="fas fa-tv"></i>
-                            <span>Preview</span>
-                        </button>
-                        <span id="preview-indicator-{{ plugin.id }}" class="htmx-indicator text-xs text-gray-500">
-                            <i class="fas fa-spinner fa-spin"></i> rendering&hellip;
-                        </span>
-                    </div>
-                    <div id="plugin-preview-{{ plugin.id }}">
-                        <p class="text-xs text-gray-400">Renders this plugin with the current (unsaved) settings &mdash; try it before you save.</p>
-                    </div>
-                </div>
             </div>
-
+            
             {# Configuration Form Panel #}
             <div class="bg-gray-50 rounded-lg p-4">
                 <h3 class="text-md font-medium text-gray-900 mb-3">Configuration</h3>