perf(wifi): one status fetch per monitor tick instead of three (#411)

* perf(wifi): one status fetch per monitor tick instead of three

The wifi monitor daemon fetched WiFi status + ethernet state before its
AP-mode check, the check internally fetched the same state again (with
retry), and the daemon fetched a third time afterwards — each fetch is
several nmcli subprocess forks, every 30s, forever, even on a perfectly
healthy link.

check_and_manage_ap_mode's decision logic is extracted to
_manage_ap_mode(status, ethernet, ap_active); the new
check_and_manage_ap_mode_with_state() runs the single (retrying) fetch
battery and returns (changed, status, ethernet, ap_active_after) — the
post-state is derivable because state only ever flips via one enable or
one disable. The original bool-returning method delegates, so existing
callers are untouched. The daemon's dead pre-fetch is removed and its
post-check reads use the returned state; the AP-enable retry semantics
(_get_wifi_status_with_retry) are preserved exactly. ~10-15 forks/30s
drops to ~4-6.

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

* fix(wifi): add missing type hints on AP-mode state helpers, fix unused-var lint in tests

- check_and_manage_ap_mode_with_state now declares its
  Tuple[bool, WiFiStatus, bool, bool] return type instead of being untyped.
- _manage_ap_mode's status/ethernet_connected/ap_active parameters are now
  typed (WiFiStatus, bool, bool), matching its existing -> bool return hint.
- test_wifi_check_state.py: rename unused unpacked variables (status/
  ethernet/ap_after) to underscore-prefixed names at the three call sites
  that don't use them, leaving the used ones untouched.

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

---------

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

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
-    find_plugin_dir, load_config_defaults, load_harness_spec, load_manifest,
+    build_full_config, find_plugin_dir, load_harness_spec, load_manifest,
 )
 from src.plugin_system.testing.harness import (  # noqa: E402
     RenderResult, render_plugin_matrix, compare_to_goldens, write_goldens,
@@ -97,12 +97,11 @@ 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), then harness.json config,
+    # config_schema defaults (real-install behavior, with enabled forced True
-    # then CLI --config — most specific wins.
+    # so a plugin's own enabled:false default can't accidentally disable
-    full_config = {"enabled": True}
+    # testing), then harness.json config, then CLI --config — most specific
-    full_config.update(load_config_defaults(plugin_dir))
+    # wins.
-    full_config.update(spec.get("config", {}))
+    full_config = build_full_config(plugin_dir, spec, 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
-    find_plugin_dir, load_manifest, load_config_defaults,
+    build_full_config, find_plugin_dir, load_manifest,
 )
 logger = get_logger("[Render Plugin]")
 
@@ -83,16 +83,13 @@ 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 = {'enabled': True}
+    config = build_full_config(Path(plugin_dir), cli_config=user_config)
-    config.update(config_defaults)
-    config.update(user_config)
 
     # Load mock data if provided
     mock_data = {}

scripts/utils/wifi_monitor_daemon.py

@@ -78,21 +78,17 @@ class WiFiMonitorDaemon:
         
         while self.running:
             try:
-                # Get current status before checking
+                # One combined check that also returns the state it observed —
-                status = self.wifi_manager.get_wifi_status()
+                # the previous flow fetched status before AND after the check
-                ethernet_connected = self.wifi_manager._is_ethernet_connected()
+                # on top of the check's own internal fetch, each one several
-                
+                # nmcli subprocess forks, every 30s, forever.
-                # Check WiFi status and manage AP mode
+                (state_changed, updated_status, updated_ethernet,
-                state_changed = self.wifi_manager.check_and_manage_ap_mode()
+                 ap_active) = self.wifi_manager.check_and_manage_ap_mode_with_state()
-                
+
-                # 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': updated_status.ap_mode_active,
+                    'ap_active': ap_active,
                     'ssid': updated_status.ssid
                 }
                 
@@ -109,7 +105,7 @@ class WiFiMonitorDaemon:
                     else:
                         logger.debug("Ethernet not connected")
                     
-                    if updated_status.ap_mode_active:
+                    if ap_active:
                         logger.info(f"AP mode ACTIVE - SSID: {ap_ssid} (IP: 192.168.4.1)")
                     else:
                         logger.debug("AP mode inactive")
@@ -123,16 +119,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={updated_status.ap_mode_active}")
+                                   f"Ethernet={updated_ethernet}, AP={ap_active}")
                     else:
-                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={updated_status.ap_mode_active}")
+                        logger.debug(f"Status check: WiFi=disconnected, Ethernet={updated_ethernet}, AP={ap_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 updated_status.ap_mode_active:
+                if updated_status.connected and not ap_active:
                     if not self.wifi_manager.check_internet_connectivity():
                         self._consecutive_internet_failures += 1
                         logger.warning(

src/config_manager.py

@@ -56,6 +56,13 @@ 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
@@ -122,6 +129,14 @@ 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
@@ -179,13 +194,36 @@ 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."""
+        """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.
+        """
         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:
@@ -205,7 +243,10 @@ 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:
@@ -264,7 +305,8 @@ 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.config = new_config_data
+            self._loaded_sig = self._files_signature()
             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,6 +23,9 @@ 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
@@ -892,6 +895,30 @@ 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:
@@ -1879,6 +1906,7 @@ 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
@@ -1889,14 +1917,64 @@ class DisplayController:
                                 )
                             _accepts_display_mode = self._plugin_accepts_display_mode[_cache_key]
 
-                            # Use PluginExecutor for safe execution with timeout
+                            pm = self.plugin_manager
-                            if self.plugin_manager and hasattr(self.plugin_manager, 'plugin_executor'):
+                            display_lock = None
-                                result = self.plugin_manager.plugin_executor.execute_display(
+                            can_display = True
-                                    manager_to_display,
+                            if pm and hasattr(pm, 'get_plugin_lock'):
-                                    plugin_id,
+                                display_lock = pm.get_plugin_lock(plugin_id)
-                                    force_clear=self.force_change,
+                                can_display = display_lock.acquire(blocking=False)
-                                    display_mode=active_mode if _accepts_display_mode else None
+
-                                )
+                            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
                                 # execute_display returns bool, convert to expected format
                                 if result:
                                     result = True  # Success
@@ -1904,10 +1982,14 @@ class DisplayController:
                                     result = False  # Failed
                             else:
                                 # Fallback to direct call if executor not available
-                                if _accepts_display_mode:
+                                try:
-                                    result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
+                                    if _accepts_display_mode:
-                                else:
+                                        result = manager_to_display.display(display_mode=active_mode, force_clear=self.force_change)
-                                    result = manager_to_display.display(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()
                             
                             logger.debug(f"display() returned: {result} (type: {type(result)})")
                             # Check if display() returned a boolean (new behavior)
@@ -1916,11 +1998,15 @@ class DisplayController:
                                 if not display_result:
                                     logger.info("Plugin %s display() returned False for mode %s", plugin_id, active_mode)
                         
-                        # Record success if display completed without exception
+                        # Record success only when display() actually ran this
-                        if self.plugin_manager and hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
+                        # frame -- a skipped frame (lock busy) held the last
-                            self.plugin_manager.health_tracker.record_success(plugin_id)
+                        # frame, not a real success, and must not clear
-                        
+                        # force_change or the pending mode-switch clear will
-                        self.force_change = False
+                        # 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
                     except Exception as exc:  # pylint: disable=broad-except
                         logger.exception("Error displaying %s", self.current_display_mode)
                         # Record failure
@@ -2205,11 +2291,16 @@ class DisplayController:
 
                             while True:
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
-                                    if _accepts_display_mode:
+                                        if can_display:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            # Pass display_mode to maintain sticky manager state
-                                    else:
+                                            if _accepts_display_mode:
-                                        result = manager_to_display.display(force_clear=False)
+                                                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
                                     if isinstance(result, bool) and not result:
                                         logger.debug("Display returned False, breaking early")
                                         break
@@ -2269,11 +2360,16 @@ class DisplayController:
                                     break
 
                                 try:
-                                    # Pass display_mode to maintain sticky manager state
+                                    with self._display_lock_or_skip(plugin_id) as can_display:
-                                    if _accepts_display_mode:
+                                        if can_display:
-                                        result = manager_to_display.display(display_mode=active_mode, force_clear=False)
+                                            # Pass display_mode to maintain sticky manager state
-                                    else:
+                                            if _accepts_display_mode:
-                                        result = manager_to_display.display(force_clear=False)
+                                                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
                                     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
@@ -2791,6 +2887,14 @@ 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/plugin_system/plugin_manager.py

@@ -8,12 +8,13 @@ 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
+from typing import Dict, List, Optional, Any, Tuple
 import logging
 from src.exceptions import PluginError, ConfigError
 from src.logging_config import get_logger
@@ -97,6 +98,50 @@ 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:
@@ -744,47 +789,189 @@ class PluginManager:
                 last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
-                # Update state to RUNNING
+                if self._synchronous_updates:
-                self.state_manager.set_state(plugin_id, PluginState.RUNNING)
+                    # Kill-switch path: the original inline execution
-                
+                    # (blocks the caller until update() completes/times out)
-                try:
+                    self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                    # Use PluginExecutor for safe execution
+                    self._execute_update_now(plugin_id, plugin_instance, current_time)
-                    success = False
+                else:
-                    if self.resource_monitor:
+                    self._enqueue_update(plugin_id, current_time)
-                        # If resource monitor exists, wrap the call
+
-                        def monitored_update():
+    def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
-                            self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
+        """Per-plugin lock keeping update() and display() mutually exclusive.
-                        # SimpleNamespace stores `update` as an *instance*
+
-                        # attribute, so attribute lookup returns the plain
+        The update worker holds it for the duration of a plugin's update();
-                        # function object as-is. A dynamically-built class
+        the display side acquires it non-blocking and skips that frame's
-                        # (`type(..., {'update': monitored_update})`) instead
+        display() call when the plugin is mid-update.
-                        # stores it as a *class* attribute, which the
+        """
-                        # descriptor protocol turns into a bound method on
+        with self._plugin_locks_guard:
-                        # access -- silently prepending the instance as an
+            lock = self._plugin_locks.get(plugin_id)
-                        # implicit first argument to a function that takes
+            if lock is None:
-                        # none, raising "monitored_update() takes 0
+                lock = threading.Lock()
-                        # positional arguments but 1 was given" on every call.
+                self._plugin_locks[plugin_id] = lock
-                        success = self.plugin_executor.execute_update(
+            return lock
-                            types.SimpleNamespace(update=monitored_update),
+
-                            plugin_id
+    def _enqueue_update(self, plugin_id: str, scheduled_time: float) -> None:
-                        )
+        """Queue a due update for the background worker (dedup while pending)."""
-                    else:
+        with self._pending_lock:
-                        success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
+            if plugin_id in self._pending_updates:
-                    
+                return
-                    if success:
+            self._pending_updates.add(plugin_id)
-                        with self._plugin_last_update_lock:
+        # RUNNING is set at enqueue time so can_execute() blocks re-entry and
-                            self.plugin_last_update[plugin_id] = current_time
+        # the web UI shows the truthful state while the item waits its turn.
-                        self.state_manager.record_update(plugin_id)
+        self.state_manager.set_state(plugin_id, PluginState.RUNNING)
-                        # Update state back to ENABLED
+        self._ensure_update_worker()
-                        self.state_manager.set_state(plugin_id, PluginState.ENABLED)
+        self._update_queue.put((plugin_id, scheduled_time))
-                        # Record success
+
-                        if self.health_tracker:
+    def _ensure_update_worker(self) -> None:
-                            self.health_tracker.record_success(plugin_id)
+        if self._update_worker is not None and self._update_worker.is_alive():
-                    else:
+            return
-                        self._record_update_failure(plugin_id)
+        self._update_worker = threading.Thread(
-                except Exception as exc:  # pylint: disable=broad-except
+            target=self._update_worker_loop, name='plugin-update-worker',
-                    self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
+            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:
                     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/testing/loading.py

@@ -88,3 +88,27 @@ 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/plugin_system/testing/mocks.py

@@ -71,6 +71,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
         # Real temp dir for plugins that write/read files under cache_dir.
         # Registered for cleanup so each mock instance doesn't leak a tmp dir.
         self.cache_dir = tempfile.mkdtemp(prefix="ledmatrix-mock-cache-")
@@ -108,6 +109,24 @@ class MockCacheManager:
         self.delete_calls.append(key)
         if key in self._cache:
             del self._cache[key]
+
+    def get_cached_data_with_strategy(self, key: str, data_type: str = 'default') -> Optional[Any]:
+        """Mock of CacheManager.get_cached_data_with_strategy (src/cache_manager.py).
+
+        The real method picks a max_age/memory_ttl strategy per data_type
+        (and extends it during market-closed hours for market data) before
+        delegating to get_cached_data(). None of that timing nuance matters
+        for a mock -- plugins under test just need the method to exist and
+        return whatever was cached, so this delegates straight to get().
+        """
+        self.get_cached_data_with_strategy_calls.append({'key': key, 'data_type': data_type})
+        return self.get(key)
+
+    def save_cache(self, key: str, data: Any) -> None:
+        """Mock of CacheManager.save_cache (src/cache_manager.py) -- the
+        write-side counterpart to get_cached_data_with_strategy, used by the
+        same real-CacheManager-oriented plugins. Delegates to set()."""
+        self.set(key, data)
         if key in self._cache_timestamps:
             del self._cache_timestamps[key]
     
@@ -118,6 +137,7 @@ class MockCacheManager:
         self.get_calls = []
         self.set_calls = []
         self.delete_calls = []
+        self.get_cached_data_with_strategy_calls = []
 
 
 class MockConfigManager:

src/wifi_manager.py

@@ -2564,11 +2564,35 @@ 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/plugins/test_harness.py

@@ -253,3 +253,61 @@ 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 load_config_defaults, load_harness_spec
+from src.plugin_system.testing.loading import build_full_config, load_harness_spec
 from src.plugin_system.testing.sizes import resolve_test_sizes
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
@@ -81,9 +81,7 @@ def test_plugin_renders_across_sizes_and_screens(plugin_id: str) -> None:
     plugin_dir = _PLUGINS[plugin_id]
     spec = load_harness_spec(plugin_dir)
 
-    config = {"enabled": True}
+    config = build_full_config(plugin_dir, spec)
-    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

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

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

@@ -123,6 +123,14 @@ 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_testing_mocks.py

@@ -0,0 +1,45 @@
+"""
+Unit tests for src/plugin_system/testing/mocks.py.
+
+MockCacheManager/MockPluginManager stand in for the real production
+managers under the plugin safety harness -- a missing method here isn't a
+harness bug in the abstract, it's a plugin silently failing to render
+under test (confirmed on ledmatrix-leaderboard, which calls
+get_cached_data_with_strategy() and previously hit an AttributeError that
+its own broad except swallowed, producing an empty-but-green render).
+"""
+
+from src.plugin_system.testing.mocks import MockCacheManager
+
+
+class TestMockCacheManagerStrategyMethod:
+    def test_get_cached_data_with_strategy_returns_cached_value(self):
+        cm = MockCacheManager()
+        cm.set("standings_nfl", {"teams": ["KC", "BUF"]})
+        result = cm.get_cached_data_with_strategy("standings_nfl", "sports_live")
+        assert result == {"teams": ["KC", "BUF"]}
+
+    def test_get_cached_data_with_strategy_returns_none_when_missing(self):
+        cm = MockCacheManager()
+        assert cm.get_cached_data_with_strategy("missing_key") is None
+
+    def test_get_cached_data_with_strategy_defaults_data_type(self):
+        cm = MockCacheManager()
+        cm.set("k", "v")
+        assert cm.get_cached_data_with_strategy("k") == "v"
+
+    def test_calls_are_tracked(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        assert cm.get_cached_data_with_strategy_calls == [{"key": "k", "data_type": "sports_live"}]
+
+    def test_save_cache_is_readable_via_strategy_lookup(self):
+        cm = MockCacheManager()
+        cm.save_cache("standings_nfl", {"teams": ["KC", "BUF"]})
+        assert cm.get_cached_data_with_strategy("standings_nfl") == {"teams": ["KC", "BUF"]}
+
+    def test_reset_clears_strategy_call_tracking(self):
+        cm = MockCacheManager()
+        cm.get_cached_data_with_strategy("k", "sports_live")
+        cm.reset()
+        assert cm.get_cached_data_with_strategy_calls == []

test/test_wifi_check_state.py

@@ -0,0 +1,92 @@
+"""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"]))