fix(testing): clear get_cached_data_with_strategy_calls in MockCacheManager.reset()

reset() cleared get_calls/set_calls/delete_calls but not the newer
get_cached_data_with_strategy_calls tracker, so a reused mock (e.g. across
test cases sharing a fixture) retained stale strategy-call records after
reset().

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

scripts/check_plugin.py

@@ -37,7 +37,7 @@ os.environ['EMULATOR'] = 'true'
 
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
+    find_plugin_dir, load_config_defaults, load_harness_spec, load_manifest,
 )
 from src.plugin_system.testing.harness import (  # noqa: E402
     RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
@@ -97,11 +97,12 @@ def check_one(plugin_id: str, search_dirs: List[str], sizes, mock_data: Dict,
     # matrix path does; explicit CLI flags still override the file.
     spec = load_harness_spec(plugin_dir)
 
-    # config_schema defaults (real-install behavior, with enabled forced True
-    # so a plugin's own enabled:false default can't accidentally disable
-    # testing), then harness.json config, then CLI --config — most specific
-    # wins.
-    full_config = build_full_config(plugin_dir, spec, config)
+    # config_schema defaults (real-install behavior), then harness.json config,
+    # then CLI --config — most specific wins.
+    full_config = {"enabled": True}
+    full_config.update(load_config_defaults(plugin_dir))
+    full_config.update(spec.get("config", {}))
+    full_config.update(config)
 
     # Precedence: CLI flag > LEDMATRIX_TEST_SIZES env > harness.json > default.
     effective_sizes = sizes if sizes else resolve_test_sizes(spec.get("sizes"))

scripts/render_plugin.py

@@ -28,7 +28,7 @@ os.environ['EMULATOR'] = 'true'
 # Import logger after path setup so src.logging_config is importable
 from src.logging_config import get_logger  # noqa: E402
 from src.plugin_system.testing.loading import (  # noqa: E402
-    build_full_config, find_plugin_dir, load_manifest,
+    find_plugin_dir, load_manifest, load_config_defaults,
 )
 logger = get_logger("[Render Plugin]")
 
@@ -83,13 +83,16 @@ def main() -> int:
     manifest = load_manifest(Path(plugin_dir))
 
     # Parse config: start with schema defaults, then apply overrides
+    config_defaults = load_config_defaults(Path(plugin_dir))
     try:
         user_config = json.loads(args.config)
     except json.JSONDecodeError as e:
         logger.error("Invalid JSON config: %s", e)
         return 1
 
-    config = build_full_config(Path(plugin_dir), cli_config=user_config)
+    config = {'enabled': True}
+    config.update(config_defaults)
+    config.update(user_config)
 
     # Load mock data if provided
     mock_data = {}

scripts/utils/wifi_monitor_daemon.py

@@ -78,17 +78,21 @@ class WiFiMonitorDaemon:
         
         while self.running:
             try:
-                # One combined check that also returns the state it observed —
-                # the previous flow fetched status before AND after the check
-                # on top of the check's own internal fetch, each one several
-                # nmcli subprocess forks, every 30s, forever.
-                (state_changed, updated_status, updated_ethernet,
-                 ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
-
+                # Get current status before checking
+                status = self.wifi_manager.get_wifi_status()
+                ethernet_connected = self.wifi_manager._is_ethernet_connected()
+                
+                # Check WiFi status and manage AP mode
+                state_changed = self.wifi_manager.check_and_manage_ap_mode()
+                
+                # Get updated status after check
+                updated_status = self.wifi_manager.get_wifi_status()
+                updated_ethernet = self.wifi_manager._is_ethernet_connected()
+                
                 current_state = {
                     'connected': updated_status.connected,
                     'ethernet_connected': updated_ethernet,
-                    'ap_active': ap_active,
+                    'ap_active': updated_status.ap_mode_active,
                     'ssid': updated_status.ssid
                 }
                 
@@ -105,7 +109,7 @@ class WiFiMonitorDaemon:
                     else:
                         logger.debug("Ethernet not connected")
                     
-                    if ap_active:
+                    if updated_status.ap_mode_active:
                         logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
                     else:
                         logger.debug("AP mode inactive")
@@ -119,16 +123,16 @@ class WiFiMonitorDaemon:
                     # Log periodic status (less verbose)
                     if updated_status.connected:
                         logger.debug(f"Status check: WiFi={updated_status.ssid} ({updated_status.signal}%), "
-                                   f"Ethernet={updated_ethernet}, AP={ap_active}")
+                                   f"Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
                     else:
-                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_active}")
+                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
                 
                 # Escalating recovery: if nmcli reports connected but actual internet
                 # is unreachable for several consecutive checks, restart NetworkManager.
                 # This is done HERE (not inside check_and_manage_ap_mode) to keep the
                 # AP-enable trigger clean and avoid false-positive AP enables from
                 # transient packet loss on otherwise working WiFi.
-                if updated_status.connected and not ap_active:
+                if updated_status.connected and not updated_status.ap_mode_active:
                     if not self.wifi_manager.check_internet_connectivity():
                         self._consecutive_internet_failures += 1
                         logger.warning(

src/cache/disk_cache.py

@@ -10,7 +10,6 @@ import time
 import tempfile
 import logging
 import threading
-import zlib
 from typing import Dict, Any, Optional, Protocol
 from datetime import datetime
 
@@ -54,11 +53,6 @@ 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]:
         """
@@ -161,35 +155,10 @@ 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
@@ -212,17 +181,13 @@ class DiskCache:
                         fd = None
                     
                     if tmp_path and fd is not None:
-                        # 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.
+                        # Use atomic write with temp file
                         try:
                             with os.fdopen(fd, 'w', encoding='utf-8') as tmp_file:
-                                tmp_file.write(payload)
+                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
+                                tmp_file.flush()
+                                os.fsync(tmp_file.fileno())
                             os.replace(tmp_path, cache_path)
-                            self._write_digests[key] = digest
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -238,8 +203,9 @@ class DiskCache:
                         # Fallback: direct write (not atomic, but better than failing)
                         try:
                             with open(cache_path, 'w', encoding='utf-8') as cache_file:
-                                cache_file.write(payload)
-                            self._write_digests[key] = digest
+                                json.dump(data, cache_file, indent=4, cls=DateTimeEncoder)
+                                cache_file.flush()
+                                os.fsync(cache_file.fileno())
                             # Set proper permissions: 660 (rw-rw----) for group-readable cache files
                             try:
                                 os.chmod(cache_path, 0o660)  # nosec B103 - intentional; web UI and service share a group
@@ -263,12 +229,9 @@ 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:
-                                tmp_file.write(payload)
+                                json.dump(data, tmp_file, indent=4, cls=DateTimeEncoder)
                             # 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
@@ -309,7 +272,6 @@ 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:
@@ -318,7 +280,6 @@ 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

@@ -1,68 +0,0 @@
-"""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/config_manager.py

@@ -56,13 +56,6 @@ class ConfigManager:
         self.secrets_path: str = secrets_path or "config/config_secrets.json"
         self.template_path: str = "config/config.template.json"
         self.config: Dict[str, Any] = {}
-        # (mtime_ns, size) signature of (config, secrets, template) at the
-        # last successful load. load_config() skips the full re-read (3 file
-        # parses + recursive template migration) when nothing changed —
-        # ~30 web request handlers call it, some 2-3x per request. Cross-
-        # process freshness is preserved: another process's save bumps the
-        # mtime, so the next load here re-reads.
-        self._loaded_sig: Optional[tuple] = None
         self.logger: logging.Logger = get_logger(__name__)
         
         # Initialize atomic config manager
@@ -129,14 +122,6 @@ class ConfigManager:
         # Update in-memory config if save was successful
         if result.status == SaveResultStatus.SUCCESS:
             self.config = new_config_data
-            # In-memory config now matches what was just written; refresh
-            # the load signature so the fast path stays valid. NOTE: the
-            # in-memory copy includes merged secrets; the on-disk file has
-            # them stripped — the fast path returning self.config preserves
-            # exactly the pre-cache behavior (load-after-save also returned
-            # the secret-merged self.config only after re-reading secrets;
-            # here secrets file is unchanged, so contents are equivalent).
-            self._loaded_sig = self._files_signature()
             self.logger.info(f"Configuration successfully saved atomically to {os.path.abspath(self.config_path)}")
         elif result.status == SaveResultStatus.ROLLED_BACK:
             # Reload config from file after rollback
@@ -194,36 +179,13 @@ class ConfigManager:
         atomic_mgr = self._get_atomic_manager()
         return atomic_mgr.validate_config_file(config_path)
 
-    def _files_signature(self) -> tuple:
-        """(mtime_ns, size) of config/secrets/template, None for missing —
-        cheap staleness probe (3 stats) for the load_config fast path."""
-        sig = []
-        for path in (self.config_path, self.secrets_path, self.template_path):
-            try:
-                st = os.stat(path)
-                sig.append((st.st_mtime_ns, st.st_size))
-            except OSError:
-                sig.append(None)
-        return tuple(sig)
-
     def load_config(self) -> Dict[str, Any]:
-        """Load configuration from JSON files.
-
-        Fast path: when config.json, config_secrets.json and the template
-        are all unchanged since the last successful load (mtime_ns + size),
-        the already-parsed self.config is returned without touching the
-        files — same aliasing semantics as the full path, which also
-        returns self.config.
-        """
+        """Load configuration from JSON files."""
         try:
-            current_sig = self._files_signature()
-            if self.config and self._loaded_sig == current_sig:
-                return self.config
-
             # Check if config file exists, if not create from template
             if not os.path.exists(self.config_path):
                 self._create_config_from_template()
-
+            
             # Load main config
             self.logger.info(f"Attempting to load config from: {os.path.abspath(self.config_path)}")
             with open(self.config_path, 'r') as f:
@@ -243,10 +205,7 @@ class ConfigManager:
                     self.logger.warning(f"Secrets file not readable ({self.secrets_path}): {e}. Continuing without secrets.")
                 except (json.JSONDecodeError, OSError) as e:
                     self.logger.warning(f"Error reading secrets file ({self.secrets_path}): {e}. Continuing without secrets.")
-
-            # Signature taken AFTER load + migration (migration may write the
-            # config back), so it reflects exactly what was read/written.
-            self._loaded_sig = self._files_signature()
+            
             return self.config
             
         except FileNotFoundError as e:
@@ -305,8 +264,7 @@ class ConfigManager:
                 json.dump(config_to_write, f, indent=4)
             
             # Update the in-memory config to the new state (which includes secrets for runtime)
-            self.config = new_config_data
-            self._loaded_sig = self._files_signature()
+            self.config = new_config_data 
             self.logger.info(f"Configuration successfully saved to {os.path.abspath(self.config_path)}")
             if secrets_content:
                  self.logger.info("Secret values were preserved in memory and not written to the main config file.")

src/display_controller.py

@@ -23,9 +23,6 @@ Entry point: :func:`main` — instantiates :class:`DisplayController` and calls
 import time
 import os
 import json
-import threading
-import types
-from contextlib import contextmanager
 from pathlib import Path
 from typing import Dict, Any, List, Optional, Callable
 from datetime import datetime
@@ -202,10 +199,6 @@ 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.
@@ -895,30 +888,6 @@ class DisplayController:
             except Exception:  # pylint: disable=broad-except
                 logger.exception("Error running scheduled plugin updates")
 
-    @contextmanager
-    def _display_lock_or_skip(self, plugin_id):
-        """Try-lock guard keeping a plugin's display() off its in-flight update().
-
-        Yields True when display may run (lock held, released on exit) or
-        when no lock support exists (older plugin manager). Yields False when
-        the plugin's update() is currently executing on the background
-        worker — the caller should treat the frame as displayed (the panel
-        holds the last pushed frame) rather than as a plugin failure, so a
-        mid-update skip never advances the rotation.
-        """
-        pm = self.plugin_manager
-        if not pm or not hasattr(pm, 'get_plugin_lock') or not plugin_id:
-            yield True
-            return
-        lock = pm.get_plugin_lock(plugin_id)
-        if not lock.acquire(blocking=False):
-            yield False
-            return
-        try:
-            yield True
-        finally:
-            lock.release()
-
     _FOLLOWER_SEND_INTERVAL = 1.0 / 90  # raw bytes are cheap; 90fps > follower render rate
 
     def _follower_rebuild_scroll_image(self) -> None:
@@ -1715,7 +1684,7 @@ class DisplayController:
                     self._sleep_with_plugin_updates(60)
                     continue
                 
-                logger.debug("Display active, processing mode: %s", self.current_display_mode)
+                logger.info(f"Display active, processing mode: {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
@@ -1883,7 +1852,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("Skipping plugin %s due to circuit breaker (mode: %s)", plugin_id, active_mode)
+                                logger.info(f"Skipping plugin {plugin_id} due to circuit breaker (mode: {active_mode})")
                                 display_result = False
                                 # Skip to next mode - let existing logic handle it
                                 manager_to_display = None
@@ -1906,7 +1875,6 @@ class DisplayController:
                     plugin_id = getattr(manager_to_display, 'plugin_id', active_mode)
                     try:
                         logger.debug(f"Calling display() for {active_mode} with force_clear={self.force_change}")
-                        can_display = False
                         if hasattr(manager_to_display, 'display'):
                             # Opt #1: look up (or compute once) whether display() accepts display_mode
                             _cache_key = plugin_id
@@ -1917,64 +1885,14 @@ class DisplayController:
                                 )
                             _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key]
 
-                            pm = self.plugin_manager
-                            display_lock = None
-                            can_display = True
-                            if pm and hasattr(pm, 'get_plugin_lock'):
-                                display_lock = pm.get_plugin_lock(plugin_id)
-                                can_display = display_lock.acquire(blocking=False)
-
-                            if not can_display:
-                                # update() in flight on the worker — hold
-                                # the last frame; not a plugin failure
-                                result = True
-                            elif pm and hasattr(pm, 'plugin_executor'):
-                                # PluginExecutor's own thread.join(timeout) can
-                                # return before the real display() call
-                                # finishes (a lingering daemon thread keeps
-                                # running it) -- so the lock is released from
-                                # inside the wrapped call itself, whichever
-                                # thread actually finishes it, rather than
-                                # here when this dispatch merely returns.
-                                release_guard = threading.Lock()
-                                released = {'done': False}
-
-                                def _release_display_lock():
-                                    with release_guard:
-                                        if released['done']:
-                                            return
-                                        released['done'] = True
-                                    if display_lock is not None:
-                                        display_lock.release()
-
-                                if _accepts_display_mode:
-                                    def _display_target(display_mode=None, force_clear=False):
-                                        try:
-                                            return manager_to_display.display(
-                                                display_mode=display_mode, force_clear=force_clear)
-                                        finally:
-                                            _release_display_lock()
-                                else:
-                                    def _display_target(force_clear=False):
-                                        try:
-                                            return manager_to_display.display(force_clear=force_clear)
-                                        finally:
-                                            _release_display_lock()
-
-                                try:
-                                    result = self.plugin_manager.plugin_executor.execute_display(
-                                        types.SimpleNamespace(display=_display_target),
-                                        plugin_id,
-                                        force_clear=self.force_change,
-                                        display_mode=active_mode if _accepts_display_mode else None
-                                    )
-                                except Exception:  # pragma: no cover - defensive;
-                                    # execute_display catches everything
-                                    # internally, but guarantee the lock is
-                                    # never leaked if something unexpected
-                                    # slips through.
-                                    _release_display_lock()
-                                    raise
+                            # Use PluginExecutor for safe execution with timeout
+                            if self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'):
+                                result = self.plugin_manager.plugin_executor.execute_display(
+                                    manager_to_display,
+                                    plugin_id,
+                                    force_clear=self.force_change,
+                                    display_mode=active_mode if _accepts_display_mode else None
+                                )
                                 # execute_display returns bool, convert to expected format
                                 if result:
                                     result = True  # Success
@@ -1982,31 +1900,23 @@ class DisplayController:
                                     result = False  # Failed
                             else:
                                 # Fallback to direct call if executor not available
-                                try:
-                                    if _accepts_display_mode:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
-                                    else:
-                                        result = manager_to_display.display(force_clear=self.force_change)
-                                finally:
-                                    if display_lock is not None:
-                                        display_lock.release()
+                                if _accepts_display_mode:
+                                    result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
+                                else:
+                                    result = manager_to_display.display(force_clear=self.force_change)
                             
                             logger.debug(f"display() returned: {result} (type: {type(result)})")
                             # Check if display() returned a boolean (new behavior)
                             if isinstance(result, bool):
                                 display_result = result
                                 if not display_result:
-                                    logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
+                                    logger.info(f"Plugin {plugin_id} display() returned False for mode {active_mode}")
                         
-                        # Record success only when display() actually ran this
-                        # frame -- a skipped frame (lock busy) held the last
-                        # frame, not a real success, and must not clear
-                        # force_change or the pending mode-switch clear will
-                        # be lost when display() finally does run.
-                        if can_display:
-                            if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
-                                self.plugin_manager.health_tracker.record_success(plugin_id)
-                            self.force_change = False
+                        # Record success if display completed without exception
+                        if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
+                            self.plugin_manager.health_tracker.record_success(plugin_id)
+                        
+                        self.force_change = False
                     except Exception as exc:  # pylint: disable=broad-except
                         logger.exception("Error displaying %s", self.current_display_mode)
                         # Record failure
@@ -2211,23 +2121,10 @@ class DisplayController:
 
                     # For plugins, call display multiple times to allow game rotation
                     if manager_to_display and hasattr(manager_to_display, 'display'):
-                        # 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.
+                        # Check if plugin needs high FPS (like stock ticker)
+                        # Always enable high-FPS for static-image plugin (for GIF animation support)
                         plugin_id = getattr(manager_to_display, 'plugin_id', None)
-                        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':
+                        if plugin_id == 'static-image':
                             needs_high_fps = True
                             logger.debug("FPS check - static-image plugin: forcing high-FPS mode for GIF support")
                         else:
@@ -2291,16 +2188,11 @@ class DisplayController:
 
                             while True:
                                 try:
-                                    with self._display_lock_or_skip(plugin_id) as can_display:
-                                        if can_display:
-                                            # Pass display_mode to maintain sticky manager state
-                                            if _accepts_display_mode:
-                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                            else:
-                                                result = manager_to_display.display(force_clear=False)
-                                        else:
-                                            # update() in flight — hold the last frame
-                                            result = True
+                                    # Pass display_mode to maintain sticky manager state
+                                    if _accepts_display_mode:
+                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                    else:
+                                        result = manager_to_display.display(force_clear=False)
                                     if isinstance(result, bool) and not result:
                                         logger.debug("Display returned False, breaking early")
                                         break
@@ -2360,16 +2252,11 @@ class DisplayController:
                                     break
 
                                 try:
-                                    with self._display_lock_or_skip(plugin_id) as can_display:
-                                        if can_display:
-                                            # Pass display_mode to maintain sticky manager state
-                                            if _accepts_display_mode:
-                                                result = manager_to_display.display(display_mode=active_mode, force_clear=False)
-                                            else:
-                                                result = manager_to_display.display(force_clear=False)
-                                        else:
-                                            # update() in flight — hold the last frame
-                                            result = True
+                                    # Pass display_mode to maintain sticky manager state
+                                    if _accepts_display_mode:
+                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                    else:
+                                        result = manager_to_display.display(force_clear=False)
                                     if isinstance(result, bool) and not result:
                                         # For dynamic duration plugins, don't exit on False - keep looping
                                         # until cycle is complete or max duration is reached
@@ -2516,16 +2403,6 @@ 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
@@ -2576,14 +2453,13 @@ class DisplayController:
                     pass
                 return None
             
-            # Message is valid and not expired — cache for the throttle window
-            self._wifi_status_last_result = {
+            # Message is valid and not expired
+            return {
                 '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
@@ -2887,14 +2763,6 @@ class DisplayController:
 
     def cleanup(self):
         """Clean up resources."""
-        # Stop the async update worker first so no in-flight update() call
-        # is still touching display/cache-backed resources while they're
-        # torn down below.
-        if self.plugin_manager and hasattr(self.plugin_manager, 'stop_update_worker'):
-            try:
-                self.plugin_manager.stop_update_worker()
-            except Exception as e:
-                logger.warning("Error stopping plugin update worker: %s", e)
         # Shutdown config service if it exists
         if hasattr(self, 'config_service'):
             try:

src/display_manager.py

@@ -31,25 +31,13 @@ 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 collections import OrderedDict
-from typing import Dict, Any, List, Optional, Tuple
+from typing import Dict, Any, List, Optional
 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
@@ -192,45 +180,14 @@ 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)) -> (width, font_ref)
+        # Text-width measurement cache: (text, id(font)) -> pixel_width
         # 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: "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._text_width_cache: Dict[tuple, int] = {}
+        # Snapshot settings for web preview integration (service writes, web reads)
         self._snapshot_path = "/tmp/led_matrix_preview.png"  # nosec B108 - fixed path intentional; web UI reads same path
-        self._viewer_marker_path = "/tmp/led_matrix_preview_viewer"  # nosec B108 - touched by web SSE broadcaster
+        self._snapshot_min_interval_sec = 0.2  # max ~5 fps
         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 = {
@@ -461,10 +418,6 @@ 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:
@@ -556,70 +509,33 @@ class DisplayManager:
         return phys
 
     def update_display(self):
-        """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.
-        """
+        """Update the display using double buffering with proper sync."""
         try:
-            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)
+            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
+
+            # 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}")
 
@@ -653,9 +569,6 @@ 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:
@@ -786,15 +699,12 @@ 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. 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.
+        measurement per unique (text, font) pair.
         """
         cache_key = (text, id(font))
         cached = self._text_width_cache.get(cache_key)
         if cached is not None:
-            self._text_width_cache.move_to_end(cache_key)
-            return cached[0]
+            return cached
 
         try:
             if isinstance(font, freetype.Face):
@@ -809,9 +719,7 @@ class DisplayManager:
             logger.error("Error getting text width: %s", e)
             return 0
 
-        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)
+        self._text_width_cache[cache_key] = width
         return width
 
     def get_font_height(self, font):
@@ -1220,56 +1128,27 @@ 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:
-        """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."""
+        """Write the current image to a PNG snapshot file at a limited frequency."""
         try:
             now = time.time()
-            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:
+            if (now - self._last_snapshot_ts) < self._snapshot_min_interval_sec:
                 return
-            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
+            # 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
+            )
             snapshot_path_obj = Path(self._snapshot_path)
-            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
+            # 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())
             # Write atomically: temp then replace
             tmp_path = f"{self._snapshot_path}.tmp"
             self.image.save(tmp_path, format='PNG')
@@ -1284,18 +1163,6 @@ 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 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}")
+            # Snapshot failures should never break display; log at debug to avoid noise
+            logger.debug(f"Snapshot write skipped: {e}")

src/font_manager.py

@@ -35,7 +35,6 @@ 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
@@ -59,13 +58,7 @@ 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
-        # (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
+        self.metrics_cache: Dict[str, Tuple[int, int, int]] = {}  # (text, font_id) -> (width, height, baseline)
 
         # Plugin font management
         self.plugin_fonts: Dict[str, Dict[str, Any]] = {}  # plugin_id -> font_manifest
@@ -562,14 +555,10 @@ class FontManager:
         Returns:
             Tuple of (width, height, baseline_offset)
         """
-        # 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))
+        cache_key = f"{hash(text)}_{id(font)}"
 
-        cached = self.metrics_cache.get(cache_key)
-        if cached is not None:
-            self.metrics_cache.move_to_end(cache_key)
-            return cached[0]
+        if cache_key in self.metrics_cache:
+            return self.metrics_cache[cache_key]
 
         try:
             if isinstance(font, freetype.Face):
@@ -606,9 +595,7 @@ class FontManager:
             baseline = 10
 
         result = (width, height, baseline)
-        self.metrics_cache[cache_key] = (result, font)
-        while len(self.metrics_cache) > self._METRICS_CACHE_MAX:
-            self.metrics_cache.popitem(last=False)
+        self.metrics_cache[cache_key] = result
         return result
 
     def get_font_height(self, font: Union[ImageFont.FreeTypeFont, freetype.Face]) -> int:

src/plugin_system/plugin_manager.py

@@ -8,13 +8,12 @@ API Version: 1.0.0
 """
 
 import json
-import queue
 import sys
 import time
 import threading
 import types
 from pathlib import Path
-from typing import Dict, List, Optional, Any, Tuple
+from typing import Dict, List, Optional, Any
 import logging
 from src.exceptions import PluginError, ConfigError
 from src.logging_config import get_logger
@@ -98,50 +97,6 @@ class PluginManager:
         # Health tracking (optional, set by display_controller if available)
         self.health_tracker = None
         self.resource_monitor = None
-
-        # --- Asynchronous plugin updates -------------------------------
-        # update() used to run inline in the render loop (execute_update's
-        # internal thread.join(timeout=30) blocked it), so one slow plugin
-        # HTTP fetch froze scrolling for the whole fetch. Scheduling still
-        # happens on the render thread (run_scheduled_updates), but
-        # execution moves to this single background worker. Per-plugin
-        # locks keep a plugin's update() and display() mutually exclusive —
-        # today's implicit guarantee, now explicit (and, unlike today,
-        # also held across the post-timeout window).
-        # Kill switch: plugin_system.synchronous_updates: true restores the
-        # inline path.
-        self._update_queue: "queue.Queue[Optional[Tuple[str, float]]]" = queue.Queue()
-        self._pending_updates: set = set()
-        self._pending_lock = threading.Lock()
-        self._plugin_locks: Dict[str, threading.Lock] = {}
-        self._plugin_locks_guard = threading.Lock()
-        self._update_worker: Optional[threading.Thread] = None
-        self._synchronous_updates = False
-        if self.config_manager is not None:
-            try:
-                cfg = self.config_manager.get_config() or {}
-            except (OSError, ValueError) as exc:
-                self.logger.warning(
-                    "Could not load config to check plugin_system.synchronous_updates "
-                    "(%s: %s); defaulting to synchronous updates", type(exc).__name__, exc)
-                self._synchronous_updates = True
-            else:
-                plugin_system_cfg = cfg.get('plugin_system', {})
-                if not isinstance(plugin_system_cfg, dict):
-                    self.logger.warning(
-                        "config plugin_system must be a mapping, got %s; "
-                        "defaulting to synchronous updates",
-                        type(plugin_system_cfg).__name__)
-                    self._synchronous_updates = True
-                else:
-                    sync_value = plugin_system_cfg.get('synchronous_updates', False)
-                    if not isinstance(sync_value, bool):
-                        self.logger.warning(
-                            "config plugin_system.synchronous_updates must be a boolean, "
-                            "got %r; defaulting to synchronous updates", sync_value)
-                        self._synchronous_updates = True
-                    else:
-                        self._synchronous_updates = sync_value
         
         # Ensure plugins directory exists with proper permissions
         try:
@@ -789,189 +744,47 @@ class PluginManager:
                 last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
-                if self._synchronous_updates:
-                    # Kill-switch path: the original inline execution
-                    # (blocks the caller until update() completes/times out)
-                    self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                    self._execute_update_now(plugin_id, plugin_instance, current_time)
-                else:
-                    self._enqueue_update(plugin_id, current_time)
-
-    def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
-        """Per-plugin lock keeping update() and display() mutually exclusive.
-
-        The update worker holds it for the duration of a plugin's update();
-        the display side acquires it non-blocking and skips that frame's
-        display() call when the plugin is mid-update.
-        """
-        with self._plugin_locks_guard:
-            lock = self._plugin_locks.get(plugin_id)
-            if lock is None:
-                lock = threading.Lock()
-                self._plugin_locks[plugin_id] = lock
-            return lock
-
-    def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None:
-        """Queue a due update for the background worker (dedup while pending)."""
-        with self._pending_lock:
-            if plugin_id in self._pending_updates:
-                return
-            self._pending_updates.add(plugin_id)
-        # RUNNING is set at enqueue time so can_execute() blocks re-entry and
-        # the web UI shows the truthful state while the item waits its turn.
-        self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-        self._ensure_update_worker()
-        self._update_queue.put((plugin_id, scheduled_time))
-
-    def _ensure_update_worker(self) -> None:
-        if self._update_worker is not None and self._update_worker.is_alive():
-            return
-        self._update_worker = threading.Thread(
-            target=self._update_worker_loop, name='plugin-update-worker',
-            daemon=True)
-        self._update_worker.start()
-
-    def _update_worker_loop(self) -> None:
-        """Single worker: dispatches queued updates off the render thread
-        (matching the old inline behavior — no thundering herd of
-        concurrent fetches).
-
-        The plugin's lock is acquired here, before its instance is looked
-        up, and the instance is re-fetched under the lock — a concurrent
-        unload_plugin() can't leave this loop about to run update() on an
-        instance that's already been torn down. The lock — and RUNNING/
-        pending lifecycle state — is released by the update itself once the
-        real update() call genuinely finishes (see _execute_update_now),
-        which can be after this dispatch returns if PluginExecutor's own
-        timeout elapses first.
-        """
-        while True:
-            item = self._update_queue.get()
-            if item is None:  # shutdown sentinel
-                return
-            plugin_id, scheduled_time = item
-            lock = self.get_plugin_lock(plugin_id)
-            lock.acquire()
-            plugin_instance = self.plugins.get(plugin_id)
-            if plugin_instance is None:  # unloaded while queued; its
-                # lifecycle state was already cleared by unload_plugin —
-                # leave it alone rather than resurrecting it to ENABLED
-                lock.release()
-                with self._pending_lock:
-                    self._pending_updates.discard(plugin_id)
-                continue
-            try:
-                self._execute_update_now(plugin_id, plugin_instance,
-                                         scheduled_time, lock=lock)
-            except Exception:  # pylint: disable=broad-except
-                # _execute_update_now guarantees the lock/pending bookkeeping
-                # is released via its own _finish() before returning or
-                # raising; this is a last-resort log only.
-                self.logger.exception("update worker: unexpected error for %s",
-                                      plugin_id)
-
-    def stop_update_worker(self, timeout: float = 5.0) -> None:
-        """Signal the worker to exit (used by cleanup; thread is a daemon)."""
-        if self._update_worker is not None and self._update_worker.is_alive():
-            self._update_queue.put(None)
-            self._update_worker.join(timeout=timeout)
-            if self._update_worker.is_alive():
-                self.logger.warning(
-                    "Update worker did not stop within %.1fs; it is a daemon "
-                    "thread and will be abandoned on shutdown", timeout)
-
-    def _execute_update_now(self, plugin_id: str, plugin_instance: Any,
-                            scheduled_time: float,
-                            lock: Optional[threading.Lock] = None) -> None:
-        """Execute a plugin's update() via PluginExecutor, then bookkeep.
-
-        Caller is responsible for having set RUNNING state.
-
-        On the synchronous path (``lock=None``) this is the original,
-        unchanged inline behavior. On the async worker path, PluginExecutor's
-        internal thread.join(timeout) blocks only the calling thread -- on
-        timeout the lingering daemon update-thread keeps running the real
-        plugin.update() call unkillable in the background. So that the
-        plugin's lock (and its RUNNING/pending lifecycle state) stays held
-        for that real duration rather than just this bounded wait, ownership
-        of both is carried by the wrapped update callable itself, released
-        from whichever thread actually finishes it -- see _finish() below.
-        """
-        finish_guard = threading.Lock()
-        finished = {'done': False}
-
-        def _finish(success: bool, exc: Optional[Exception] = None) -> None:
-            with finish_guard:
-                if finished['done']:
-                    return
-                finished['done'] = True
-            try:
-                if success:
-                    with self._plugin_last_update_lock:
-                        self.plugin_last_update[plugin_id] = scheduled_time
-                    self.state_manager.record_update(plugin_id)
-                    self.state_manager.set_state(plugin_id, PluginState.ENABLED)
-                    if self.health_tracker:
-                        self.health_tracker.record_success(plugin_id)
-                else:
+                # Update state to RUNNING
+                self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+                
+                try:
+                    # Use PluginExecutor for safe execution
+                    success = False
+                    if self.resource_monitor:
+                        # If resource monitor exists, wrap the call
+                        def monitored_update():
+                            self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+                        # SimpleNamespace stores `update` as an *instance*
+                        # attribute, so attribute lookup returns the plain
+                        # function object as-is. A dynamically-built class
+                        # (`type(..., {'update': monitored_update})`) instead
+                        # stores it as a *class* attribute, which the
+                        # descriptor protocol turns into a bound method on
+                        # access -- silently prepending the instance as an
+                        # implicit first argument to a function that takes
+                        # none, raising "monitored_update() takes 0
+                        # positional arguments but 1 was given" on every call.
+                        success = self.plugin_executor.execute_update(
+                            types.SimpleNamespace(update=monitored_update),
+                            plugin_id
+                        )
+                    else:
+                        success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
+                    
+                    if success:
+                        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)
+                        # Record success
+                        if self.health_tracker:
+                            self.health_tracker.record_success(plugin_id)
+                    else:
+                        self._record_update_failure(plugin_id)
+                except Exception as exc:  # pylint: disable=broad-except
+                    self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
                     self._record_update_failure(plugin_id, exc=exc)
-            finally:
-                if lock is not None:
-                    lock.release()
-                    with self._pending_lock:
-                        self._pending_updates.discard(plugin_id)
-
-        if lock is None:
-            # Synchronous / no-lock path: unchanged behavior.
-            try:
-                if self.resource_monitor:
-                    def monitored_update():
-                        self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
-                    # SimpleNamespace stores `update` as an *instance*
-                    # attribute, so attribute lookup returns the plain
-                    # function object as-is. A dynamically-built class
-                    # (`type(..., {'update': monitored_update})`) instead
-                    # stores it as a *class* attribute, which the
-                    # descriptor protocol turns into a bound method on
-                    # access -- silently prepending the instance as an
-                    # implicit first argument to a function that takes
-                    # none, raising "monitored_update() takes 0
-                    # positional arguments but 1 was given" on every call.
-                    success = self.plugin_executor.execute_update(
-                        types.SimpleNamespace(update=monitored_update),
-                        plugin_id
-                    )
-                else:
-                    success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
-                _finish(success)
-            except Exception as exc:  # pylint: disable=broad-except
-                self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
-                _finish(False, exc=exc)
-            return
-
-        # Async worker path: the real update() call -- through the resource
-        # monitor, if configured -- owns finishing the lock/lifecycle
-        # bookkeeping, from whichever thread actually runs it to completion.
-        def _target_update() -> None:
-            try:
-                if self.resource_monitor:
-                    self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
-                else:
-                    plugin_instance.update()
-            except Exception as exc:
-                _finish(False, exc=exc)
-                raise
-            else:
-                _finish(True)
-
-        try:
-            self.plugin_executor.execute_update(
-                types.SimpleNamespace(update=_target_update), plugin_id)
-        except Exception as exc:  # pragma: no cover - defensive; execute_update
-            # catches everything internally, but guarantee _finish still
-            # runs (releasing the lock) if something unexpected slips through.
-            self.logger.exception("Unexpected error dispatching update for %s: %s", plugin_id, exc)
-            _finish(False, exc=exc)
 
     def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
         """

src/plugin_system/store_manager.py

@@ -142,28 +142,9 @@ 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
@@ -2282,74 +2263,6 @@ 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.
@@ -2412,7 +2325,10 @@ class PluginStoreManager:
                             f"Plugin {resolved_id} git remote ({local_remote}) differs from registry ({registry_repo}). "
                             f"Reinstalling from registry to migrate to new source."
                         )
-                        return self._reinstall_with_rollback(resolved_id, plugin_path)
+                        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)
 
                     # Check if already up to date
                     if remote_sha and local_sha and remote_sha.startswith(local_sha):
@@ -2716,11 +2632,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})")
 
-            # 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)
+            # 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)
 
         except Exception as e:
             import traceback

src/plugin_system/testing/loading.py

@@ -88,27 +88,3 @@ def load_harness_spec(plugin_dir: Union[str, Path]) -> Dict[str, Any]:
         with open(mock_path, 'r') as mf:
             spec['mock_data_contents'] = json.load(mf)
     return spec
-
-
-def build_full_config(
-    plugin_dir: Union[str, Path],
-    spec: Optional[Dict[str, Any]] = None,
-    cli_config: Optional[Dict[str, Any]] = None,
-) -> Dict[str, Any]:
-    """Build the config a plugin sees under test.
-
-    Merge order: config_schema.json defaults, then a forced ``enabled: True``,
-    then harness.json's config overlay, then the caller's explicit config --
-    most specific wins. `enabled` is re-asserted *after* the schema defaults
-    so a plugin that reasonably ships `enabled: false` (e.g. a seasonal or
-    opt-in plugin) can't silently make every harness run test "disabled, do
-    nothing" by accident -- callers that genuinely want to test the disabled
-    path can still do so via `cli_config={"enabled": False}`.
-    """
-    spec = spec or {}
-    config: Dict[str, Any] = {}
-    config.update(load_config_defaults(plugin_dir))
-    config["enabled"] = True
-    config.update(spec.get("config", {}))
-    config.update(cli_config or {})
-    return config

src/vegas_mode/render_pipeline.py

@@ -66,10 +66,6 @@ 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,
@@ -238,19 +234,11 @@ class RenderPipeline:
                     )
                     # Push blank immediately so the hardware never shows any
                     # post-wrap content while the coordinator recomposes the
-                    # 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).
+                    # next cycle (~100 ms).
                     try:
-                        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
+                        from PIL import Image as _Image
+                        blank = _Image.new('RGB', (self.display_width, self.display_height))
+                        self.display_manager.image = blank
                         self.display_manager.update_display()
                     except Exception:
                         logger.exception("Failed to write blank frame to display at cycle end")

src/wifi_manager.py

@@ -2564,35 +2564,11 @@ address=/detectportal.firefox.com/192.168.4.1
         Returns:
             True if AP mode state changed, False otherwise
         """
-        changed, _status, _ethernet, _ap = self.check_and_manage_ap_mode_with_state()
-        return changed
-
-    def check_and_manage_ap_mode_with_state(self) -> Tuple[bool, WiFiStatus, bool, bool]:
-        """Like check_and_manage_ap_mode, but also returns the state it
-        observed, so callers (the wifi monitor daemon) don't have to re-run
-        the same nmcli subprocess battery before AND after the check —
-        each status fetch is several process forks.
-
-        Returns:
-            (state_changed, WiFiStatus, ethernet_connected, ap_active_after)
-        """
         try:
             # Get status with retry for more reliable detection
             status = self._get_wifi_status_with_retry()
             ethernet_connected = self._is_ethernet_connected()
             ap_active = self._is_ap_mode_active()
-            changed = self._manage_ap_mode(status, ethernet_connected, ap_active)
-            # State only ever changes via one enable or one disable, so the
-            # post-state is the inverse of the pre-state when changed.
-            ap_after = (not ap_active) if changed else ap_active
-            return changed, status, ethernet_connected, ap_after
-        except Exception as e:
-            logger.error(f"Error checking AP mode: {e}", exc_info=True)
-            return False, WiFiStatus(connected=False), False, False
-
-    def _manage_ap_mode(self, status: WiFiStatus, ethernet_connected: bool, ap_active: bool) -> bool:
-        """AP-mode decision logic against an already-fetched state snapshot."""
-        try:
             auto_enable = self.config.get("auto_enable_ap_mode", True)  # Default: True (safe due to grace period)
             
             # Log current state for debugging

test/conftest.py

@@ -38,11 +38,7 @@ def mock_cache_manager():
     mock._memory_cache_timestamps = {}
     mock.cache_dir = "/tmp/test_cache"
     
-    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.
+    def mock_get(key: str, max_age: int = 300) -> Optional[Dict]:
         return mock._memory_cache.get(key)
     
     def mock_set(key: str, data: Dict, ttl: Optional[int] = None) -> None:

test/plugins/test_harness.py

@@ -253,61 +253,3 @@ class TestCheckPluginHonorsHarnessJson:
         )
         assert captured["freeze_time"] == "2030-01-01 00:00:00"
         assert captured["config"]["timezone"] == "America/New_York"
-
-
-class TestBuildFullConfigForcesEnabled:
-    """Regression: a plugin's own config_schema.json may reasonably default
-    enabled to False (e.g. a seasonal or opt-in plugin) -- march-madness and
-    14 other real plugins do. The harness must still test it as enabled
-    unless a caller explicitly asks otherwise, or every render silently
-    becomes a same-shaped "disabled, do nothing" no-op."""
-
-    def _make_plugin_with_disabled_default(self, tmp_path):
-        pdir = tmp_path / "plugins" / "demo-seasonal"
-        pdir.mkdir(parents=True)
-        (pdir / "manifest.json").write_text(json.dumps({
-            "id": "demo-seasonal", "name": "Demo Seasonal", "version": "1.0.0",
-            "author": "test", "entry_point": "manager.py",
-            "class_name": "DemoSeasonal", "display_modes": ["demo-seasonal"],
-            "compatible_versions": ["*"],
-        }))
-        (pdir / "config_schema.json").write_text(json.dumps({
-            "type": "object",
-            "properties": {"enabled": {"type": "boolean", "default": False}},
-        }))
-        return pdir
-
-    def test_schema_disabled_default_does_not_win(self, tmp_path):
-        from src.plugin_system.testing.loading import build_full_config
-        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
-        config = build_full_config(plugin_dir)
-        assert config["enabled"] is True
-
-    def test_harness_json_config_can_still_disable(self, tmp_path):
-        from src.plugin_system.testing.loading import build_full_config
-        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
-        config = build_full_config(plugin_dir, spec={"config": {"enabled": False}})
-        assert config["enabled"] is False
-
-    def test_explicit_cli_config_can_still_disable(self, tmp_path):
-        from src.plugin_system.testing.loading import build_full_config
-        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
-        config = build_full_config(plugin_dir, cli_config={"enabled": False})
-        assert config["enabled"] is False
-
-    def test_check_one_renders_a_schema_disabled_plugin_as_enabled(self, tmp_path, monkeypatch):
-        """End-to-end: check_plugin.py's check_one() must not blank-render a
-        plugin just because its own schema defaults enabled to False."""
-        mod = _load_check_plugin_cli()
-        plugin_dir = self._make_plugin_with_disabled_default(tmp_path)
-        captured = {}
-        monkeypatch.setattr(mod, "render_plugin_matrix",
-                            lambda **kw: captured.update(kw) or [])
-        monkeypatch.setattr(mod, "compare_to_goldens", lambda *a, **k: [])
-        mod.check_one(
-            plugin_id="demo-seasonal", search_dirs=[str(tmp_path / "plugins")],
-            sizes=None, mock_data={}, config={}, run_update=True,
-            out_dir=None, update_golden=False, golden_dir_override=None,
-            freeze_time=None,
-        )
-        assert captured["config"]["enabled"] is True

test/plugins/test_plugin_matrix.py

@@ -22,7 +22,7 @@ import pytest
 from src.plugin_system.testing.harness import (
     render_plugin_matrix, compare_to_goldens,
 )
-from src.plugin_system.testing.loading import build_full_config, load_harness_spec
+from src.plugin_system.testing.loading import load_config_defaults, load_harness_spec
 from src.plugin_system.testing.sizes import resolve_test_sizes
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -81,7 +81,9 @@ def test_plugin_renders_across_sizes_and_screens(plugin_id: str) -> None:
     plugin_dir = _PLUGINS[plugin_id]
     spec = load_harness_spec(plugin_dir)
 
-    config = build_full_config(plugin_dir, spec)
+    config = {"enabled": True}
+    config.update(load_config_defaults(plugin_dir))
+    config.update(spec.get("config", {}))
 
     # Sizes: LEDMATRIX_TEST_SIZES env (test on real hardware) wins, then the
     # plugin's own harness.json "sizes", else the default representative sample.

test/test_async_plugin_updates.py

@@ -1,270 +0,0 @@
-"""Tests for asynchronous plugin updates (plugin_manager background worker).
-
-The invariants that keep this change safe:
-1. run_scheduled_updates returns immediately — a slow update() can never
-   again freeze the render loop (the original defect: 30s scroll freezes).
-2. A plugin's update() and display() are NEVER concurrent — the per-plugin
-   lock makes the old implicit no-overlap guarantee explicit, and it stays
-   held through PluginExecutor's own timeout: a lingering, still-running
-   update() keeps the lock even after PluginExecutor gives up waiting on it.
-3. Failure/timeout bookkeeping is unchanged (same executor, same
-   _record_update_failure path, same last-update stamping).
-4. The kill switch (plugin_system.synchronous_updates) restores the
-   inline path exactly.
-"""
-
-import os
-import sys
-import threading
-import time
-
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.plugin_system.plugin_manager import PluginManager  # noqa: E402
-from src.plugin_system.plugin_state import PluginState  # noqa: E402
-
-
-class SlowPlugin:
-    """Fake plugin whose update() sleeps and records overlap violations."""
-
-    def __init__(self, update_seconds=0.5):
-        self.enabled = True
-        self.update_seconds = update_seconds
-        self.update_calls = 0
-        self.display_calls = 0
-        self.in_update = False
-        self.overlap_detected = False
-
-    def update(self):
-        self.in_update = True
-        self.update_calls += 1
-        time.sleep(self.update_seconds)
-        self.in_update = False
-        return True
-
-    def display(self, force_clear=False):
-        if self.in_update:
-            self.overlap_detected = True
-        self.display_calls += 1
-        return True
-
-
-@pytest.fixture
-def pm(tmp_path):
-    manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
-                            display_manager=None, cache_manager=None)
-    yield manager
-    manager.stop_update_worker()
-
-
-def _install(pm, plugin, plugin_id="slow-plugin"):
-    pm.plugins[plugin_id] = plugin
-    pm._update_interval_cache[plugin_id] = 0.01  # always due
-    # load_plugin normally registers state; can_execute() gates on it
-    pm.state_manager.set_state(plugin_id, PluginState.ENABLED)
-    return plugin_id
-
-
-class TestSchedulerNonBlocking:
-    def test_run_scheduled_updates_returns_immediately(self, pm):
-        plugin_id = _install(pm, SlowPlugin(update_seconds=2.0))
-        start = time.monotonic()
-        pm.run_scheduled_updates()
-        elapsed = time.monotonic() - start
-        assert elapsed < 0.1, f"scheduler blocked for {elapsed:.2f}s"
-        # the update actually runs in the background
-        deadline = time.monotonic() + 5
-        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
-            time.sleep(0.05)
-        assert pm.plugins[plugin_id].update_calls == 1
-
-    def test_no_double_enqueue_while_pending(self, pm):
-        plugin_id = _install(pm, SlowPlugin(update_seconds=0.8))
-        for _ in range(20):
-            pm.run_scheduled_updates()
-            time.sleep(0.01)
-        time.sleep(1.5)  # let the single queued update finish
-        assert pm.plugins[plugin_id].update_calls == 1
-
-
-class TestUpdateDisplayExclusion:
-    def test_display_lock_held_during_update(self, pm):
-        plugin_id = _install(pm, SlowPlugin(update_seconds=0.6))
-        pm.run_scheduled_updates()
-        # give the worker a moment to take the lock and enter update()
-        deadline = time.monotonic() + 2
-        while not pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
-            time.sleep(0.01)
-        lock = pm.get_plugin_lock(plugin_id)
-        assert lock.acquire(blocking=False) is False, \
-            "lock must be held while update() runs"
-        # and released afterwards
-        deadline = time.monotonic() + 3
-        while pm.plugins[plugin_id].in_update and time.monotonic() < deadline:
-            time.sleep(0.05)
-        time.sleep(0.1)
-        assert lock.acquire(blocking=False) is True
-        lock.release()
-
-    def test_no_overlap_under_hammering_display_loop(self, pm):
-        """Simulate the render loop's try-lock display pattern at high rate
-        while updates fire — the plugin itself asserts no overlap."""
-        plugin = SlowPlugin(update_seconds=0.15)
-        plugin_id = _install(pm, plugin)
-        stop = threading.Event()
-
-        def render_loop():
-            while not stop.is_set():
-                lock = pm.get_plugin_lock(plugin_id)
-                if lock.acquire(blocking=False):
-                    try:
-                        plugin.display()
-                    finally:
-                        lock.release()
-                time.sleep(0.002)
-
-        renderer = threading.Thread(target=render_loop, daemon=True)
-        renderer.start()
-        try:
-            for _ in range(6):
-                pm.plugin_last_update.pop(plugin_id, None)  # force due
-                pm.run_scheduled_updates()
-                time.sleep(0.3)
-        finally:
-            stop.set()
-            renderer.join(timeout=2)
-        assert plugin.update_calls >= 3
-        assert plugin.display_calls > 10
-        assert plugin.overlap_detected is False
-
-    def test_lock_held_through_timeout_until_real_update_finishes(self, pm):
-        """PluginExecutor's join(timeout) can return before the real
-        update() call does -- the lingering daemon thread keeps running it.
-        The plugin's lock must stay held for that whole real duration, not
-        just PluginExecutor's bounded wait, or display() could run
-        concurrently with a still-executing update()."""
-        plugin = SlowPlugin(update_seconds=0.3)
-        plugin_id = _install(pm, plugin)
-        pm.plugin_executor.default_timeout = 0.05  # times out well before
-                                                    # update_seconds elapses
-
-        pm.run_scheduled_updates()
-
-        deadline = time.monotonic() + 2
-        while not plugin.in_update and time.monotonic() < deadline:
-            time.sleep(0.01)
-        assert plugin.in_update is True
-
-        # PluginExecutor's own timeout has now elapsed, but the real
-        # update() (0.3s) is still running in its lingering daemon thread.
-        time.sleep(0.15)
-        assert plugin.in_update is True, "test setup: update should still be running"
-        lock = pm.get_plugin_lock(plugin_id)
-        assert lock.acquire(blocking=False) is False, \
-            "lock must stay held through PluginExecutor's timeout while the real update() runs"
-
-        # Once the real update() genuinely finishes, the lock is released.
-        deadline = time.monotonic() + 2
-        while plugin.in_update and time.monotonic() < deadline:
-            time.sleep(0.02)
-        time.sleep(0.1)
-        assert lock.acquire(blocking=False) is True
-        lock.release()
-
-    def test_state_returns_to_enabled_after_update(self, pm):
-        """RUNNING is set at enqueue (blocks re-entry via can_execute) and
-        must return to an executable state once the update finishes."""
-        plugin_id = _install(pm, SlowPlugin(update_seconds=0.1))
-        pm.run_scheduled_updates()
-        # while queued/running, re-entry is blocked
-        assert pm.state_manager.can_execute(plugin_id) is False
-        deadline = time.monotonic() + 3
-        while time.monotonic() < deadline:
-            if (pm.plugins[plugin_id].update_calls
-                    and pm.state_manager.can_execute(plugin_id)):
-                break
-            time.sleep(0.05)
-        assert pm.plugins[plugin_id].update_calls == 1
-        assert pm.state_manager.can_execute(plugin_id) is True
-
-
-class TestFailurePaths:
-    def test_update_failure_routes_through_failure_bookkeeping(self, pm):
-        class FailingPlugin(SlowPlugin):
-            def update(self):
-                self.update_calls += 1
-                raise RuntimeError("boom")
-
-        plugin_id = _install(pm, FailingPlugin())
-        pm.run_scheduled_updates()
-        deadline = time.monotonic() + 3
-        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
-            time.sleep(0.05)
-        time.sleep(0.2)
-        # failure stamped so the interval gate holds (no hot retry loop)
-        assert pm.plugin_last_update.get(plugin_id, 0) > 0
-        # lock released after failure
-        assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True
-        pm.get_plugin_lock(plugin_id).release()
-
-    def test_unloaded_while_queued_is_harmless(self, pm):
-        """Exercise the public unload_plugin() lifecycle rather than
-        deleting pm.plugins directly: queue the target's update behind a
-        deterministic blocker (occupying the single worker), unload the
-        target while its item still sits queued, then release the blocker
-        and confirm the target's update never ran and its state stayed
-        unloaded rather than being resurrected to ENABLED."""
-        blocker_event = threading.Event()
-
-        class BlockerPlugin(SlowPlugin):
-            def update(self):
-                self.update_calls += 1
-                blocker_event.wait(timeout=5)
-                return True
-
-        blocker_id = _install(pm, BlockerPlugin(), plugin_id="blocker-plugin")
-        target = SlowPlugin(update_seconds=0.05)
-        target_id = _install(pm, target, plugin_id="slow-plugin")
-
-        # Dispatch the blocker first so it occupies the single worker
-        # thread, then enqueue the target behind it -- deterministically
-        # queued, not yet started.
-        pm._enqueue_update(blocker_id, time.time())
-        deadline = time.monotonic() + 2
-        while pm.plugins[blocker_id].update_calls == 0 and time.monotonic() < deadline:
-            time.sleep(0.01)
-        assert pm.plugins[blocker_id].update_calls == 1
-
-        pm._enqueue_update(target_id, time.time())
-        assert target_id in pm._pending_updates
-
-        assert pm.unload_plugin(target_id) is True
-        assert target_id not in pm.plugins
-
-        blocker_event.set()  # let the blocker finish; worker moves on to
-                              # the target's queued item
-
-        deadline = time.monotonic() + 3
-        while target_id in pm._pending_updates and time.monotonic() < deadline:
-            time.sleep(0.02)
-
-        assert target.update_calls == 0, "update() must not run for an unloaded plugin"
-        assert target_id not in pm._pending_updates
-        assert pm.state_manager.get_state(target_id) == PluginState.UNLOADED
-
-
-class TestKillSwitch:
-    def test_synchronous_mode_blocks_like_before(self, pm):
-        pm._synchronous_updates = True
-        plugin_id = _install(pm, SlowPlugin(update_seconds=0.4))
-        start = time.monotonic()
-        pm.run_scheduled_updates()
-        elapsed = time.monotonic() - start
-        assert elapsed >= 0.4, "synchronous mode must run inline"
-        assert pm.plugins[plugin_id].update_calls == 1
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-v"]))

test/test_cache_manager.py

@@ -400,61 +400,3 @@ 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_config_load_cache.py

@@ -1,129 +0,0 @@
-"""Tests for load_config's mtime fast path (src/config_manager.py).
-
-load_config used to re-read + re-parse config.json, the template (with a
-recursive migration diff) and secrets on EVERY call — ~30 web request
-handlers call it, some 2-3x per request. The fast path skips all of it
-when the three files' (mtime_ns, size) signatures are unchanged.
-
-The invariant that matters most: cross-process freshness — a save from
-the web process must be picked up by the display process's next load.
-That's guaranteed because the signature is re-stat'd on every call.
-"""
-
-import json
-import os
-import sys
-
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.config_manager import ConfigManager  # noqa: E402
-
-
-@pytest.fixture
-def mgr(tmp_path):
-    config = tmp_path / "config.json"
-    secrets = tmp_path / "secrets.json"
-    template = tmp_path / "template.json"
-    config.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
-    secrets.write_text(json.dumps({"weather": {"api_key": "sek"}}))
-    template.write_text(json.dumps({"display": {"brightness": 90}, "timezone": "UTC"}))
-    m = ConfigManager(config_path=str(config), secrets_path=str(secrets))
-    m.template_path = str(template)
-    return m, config, secrets, template
-
-
-def _count_opens(monkeypatch, mgr_paths):
-    """Count open() calls hitting the config files."""
-    counts = {"n": 0}
-    real_open = open
-
-    def counting_open(file, *args, **kwargs):
-        if str(file) in mgr_paths:
-            counts["n"] += 1
-        return real_open(file, *args, **kwargs)
-
-    import builtins
-    monkeypatch.setattr(builtins, "open", counting_open)
-    return counts
-
-
-class TestFastPath:
-    def test_unchanged_files_are_not_reread(self, mgr, monkeypatch):
-        m, config, secrets, template = mgr
-        first = m.load_config()
-        assert first["weather"]["api_key"] == "sek"  # secrets merged
-        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
-        for _ in range(10):
-            again = m.load_config()
-        assert counts["n"] == 0, "fast path must not re-open any config file"
-        assert again is first  # same aliasing semantics as the full path
-
-    def test_config_change_triggers_reload(self, mgr):
-        m, config, secrets, template = mgr
-        m.load_config()
-        data = json.loads(config.read_text())
-        data["display"]["brightness"] = 55
-        config.write_text(json.dumps(data))
-        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
-        assert m.load_config()["display"]["brightness"] == 55
-
-    def test_secrets_change_triggers_reload(self, mgr):
-        m, config, secrets, template = mgr
-        m.load_config()
-        secrets.write_text(json.dumps({"weather": {"api_key": "NEW"}}))
-        os.utime(secrets, (os.stat(secrets).st_atime, os.stat(secrets).st_mtime + 2))
-        assert m.load_config()["weather"]["api_key"] == "NEW"
-
-    def test_template_change_triggers_reload_and_migration(self, mgr):
-        m, config, secrets, template = mgr
-        m.load_config()
-        template.write_text(json.dumps({
-            "display": {"brightness": 90}, "timezone": "UTC",
-            "brand_new_key": {"added": True}}))
-        os.utime(template, (os.stat(template).st_atime, os.stat(template).st_mtime + 2))
-        reloaded = m.load_config()
-        assert reloaded.get("brand_new_key") == {"added": True}
-
-    def test_same_second_edit_detected_via_mtime_ns_or_size(self, mgr):
-        """Coarse-mtime same-second edits: size difference still busts it."""
-        m, config, secrets, template = mgr
-        m.load_config()
-        st = os.stat(config)
-        data = json.loads(config.read_text())
-        data["timezone"] = "America/New_York"  # different byte length
-        config.write_text(json.dumps(data))
-        os.utime(config, (st.st_atime, st.st_mtime))  # force same mtime
-        assert m.load_config()["timezone"] == "America/New_York"
-
-
-class TestSaveCoherence:
-    def test_save_config_then_load_returns_saved_data(self, mgr, monkeypatch):
-        m, config, secrets, template = mgr
-        m.load_config()
-        new = {"display": {"brightness": 42}, "timezone": "UTC",
-               "weather": {"api_key": "sek"}}
-        m.save_config(new)
-        counts = _count_opens(monkeypatch, {str(config), str(secrets), str(template)})
-        loaded = m.load_config()
-        assert loaded["display"]["brightness"] == 42
-        assert loaded["weather"]["api_key"] == "sek"  # secrets survive in memory
-        assert counts["n"] == 0  # signature refreshed by save; no re-read
-
-    def test_cross_process_save_is_picked_up(self, mgr):
-        """Another process writing config.json (different mtime) must bust
-        this process's fast path — the core cross-process guarantee."""
-        m, config, secrets, template = mgr
-        m.load_config()
-        other = ConfigManager(config_path=str(config), secrets_path=str(secrets))
-        other.template_path = str(template)
-        other.load_config()
-        other.save_config({"display": {"brightness": 11}, "timezone": "UTC",
-                           "weather": {"api_key": "sek"}})
-        os.utime(config, (os.stat(config).st_atime, os.stat(config).st_mtime + 2))
-        assert m.load_config()["display"]["brightness"] == 11
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-v"]))

test/test_display_dirty_tracking.py

@@ -1,162 +0,0 @@
-"""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_plugin_system.py

@@ -123,14 +123,6 @@ class TestPluginManager:
 
             pm.run_scheduled_updates(current_time=time.time())
 
-            # Updates now execute on the background worker (the scheduler
-            # returns immediately) — wait for completion before asserting.
-            deadline = time.time() + 5
-            while (plugin_instance.update.call_count == 0
-                   and time.time() < deadline):
-                time.sleep(0.02)
-            pm.stop_update_worker()
-
             plugin_instance.update.assert_called_once()
             assert "test_plugin" in pm.plugin_last_update
             assert pm.state_manager.get_state("test_plugin") == PluginState.ENABLED

test/test_snapshot_policy.py

@@ -1,93 +0,0 @@
-"""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

@@ -1,171 +0,0 @@
-"""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_wifi_check_state.py

@@ -1,92 +0,0 @@
-"""Tests for check_and_manage_ap_mode_with_state (src/wifi_manager.py).
-
-The wifi monitor daemon used to fetch WiFi status before AND after each
-check on top of the check's own internal fetch — every fetch is several
-nmcli subprocess forks, every 30s, forever. The new API returns the state
-the check observed, so the daemon runs exactly one fetch battery per tick.
-"""
-
-import os
-import sys
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
-
-from src.wifi_manager import WiFiManager, WiFiStatus  # noqa: E402
-
-
-@pytest.fixture
-def wm(tmp_path):
-    with patch.object(WiFiManager, "_load_config", return_value={}, create=True):
-        manager = WiFiManager.__new__(WiFiManager)
-        # minimal attribute setup without running the real __init__
-        manager.config = {"auto_enable_ap_mode": True}
-        manager._disconnected_checks = 0
-        manager._disconnected_checks_required = 3
-        manager._ap_enabled_at = None
-        return manager
-
-
-def _wire(wm, connected, ethernet, ap_active):
-    wm._get_wifi_status_with_retry = MagicMock(
-        return_value=WiFiStatus(connected=connected, ssid="net" if connected else None))
-    wm._is_ethernet_connected = MagicMock(return_value=ethernet)
-    wm._is_ap_mode_active = MagicMock(return_value=ap_active)
-    wm.enable_ap_mode = MagicMock(return_value=(True, "ok"))
-    wm.disable_ap_mode = MagicMock(return_value=(True, "ok"))
-    wm.scan_networks = MagicMock(return_value=([], False))
-    wm._save_cached_scan = MagicMock()
-    wm._FORCE_AP_FLAG_PATH = MagicMock()
-    wm._FORCE_AP_FLAG_PATH.exists.return_value = False
-
-
-class TestWithState:
-    def test_single_fetch_per_call(self, wm):
-        _wire(wm, connected=True, ethernet=False, ap_active=False)
-        wm.check_and_manage_ap_mode_with_state()
-        assert wm._get_wifi_status_with_retry.call_count == 1
-        assert wm._is_ethernet_connected.call_count == 1
-        assert wm._is_ap_mode_active.call_count == 1
-
-    def test_returns_observed_state(self, wm):
-        _wire(wm, connected=True, ethernet=True, ap_active=False)
-        changed, status, ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
-        assert changed is False
-        assert status.connected is True
-        assert ethernet is True
-        assert ap_after is False
-
-    def test_ap_after_inverts_on_disable(self, wm):
-        """WiFi reconnects while AP is up -> auto-disable -> ap_after False."""
-        _wire(wm, connected=True, ethernet=False, ap_active=True)
-        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
-        assert changed is True
-        assert ap_after is False
-        wm.disable_ap_mode.assert_called_once()
-
-    def test_ap_after_inverts_on_enable(self, wm):
-        """Grace period exhausted with nothing connected -> enable -> True."""
-        _wire(wm, connected=False, ethernet=False, ap_active=False)
-        wm._disconnected_checks = 2  # this call is the 3rd
-        changed, _status, _ethernet, ap_after = wm.check_and_manage_ap_mode_with_state()
-        assert changed is True
-        assert ap_after is True
-        wm.enable_ap_mode.assert_called_once()
-
-    def test_bool_wrapper_is_back_compatible(self, wm):
-        _wire(wm, connected=True, ethernet=False, ap_active=False)
-        assert wm.check_and_manage_ap_mode() is False
-        _wire(wm, connected=True, ethernet=False, ap_active=True)
-        assert wm.check_and_manage_ap_mode() is True
-
-    def test_exception_path_never_raises(self, wm):
-        wm._get_wifi_status_with_retry = MagicMock(side_effect=RuntimeError("nmcli gone"))
-        changed, status, _ethernet, _ap_after = wm.check_and_manage_ap_mode_with_state()
-        assert changed is False
-        assert status.connected is False
-
-
-if __name__ == "__main__":
-    sys.exit(pytest.main([__file__, "-v"]))

web_interface/app.py

@@ -608,23 +608,9 @@ 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:
@@ -641,7 +627,6 @@ 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)