Merge origin/main into claude/plugin-hang-containment

Conflict with #678 in the plugin config-change callback: keep #678's
override (a plugin loaded for on-demand sees enabled: True) and apply the
result through the locked apply_config_change. The #678 test now asserts
on that call, which delivers the override to on_config_change.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -19,6 +19,56 @@ accepts both, but the store flags the old spelling as deprecated
 
 ## Unreleased
 
+### Fixes
+
+- On-demand no longer restarts a running display. `POST
+  /display/on-demand/start` treated `start_service` (on by default, and what
+  "Preview on display", the on-demand dialog and the MQTT bridge all send) as
+  "restart": it stopped the service, waited 1.5s and started it again, so
+  every request reloaded every plugin and left the panel blank for seconds.
+  The running display already reads the request within a quarter of a second,
+  mid-screen and mid-Vegas included, so the route now only starts the service
+  when it is not running. `POST /display/on-demand/stop` reads
+  `stop_service` as a boolean, so `"false"` no longer stops the service.
+- On-demand works for a disabled plugin. The display only loads enabled
+  plugins, so "Preview on display" on a disabled plugin's config page (which
+  says the plugin will be enabled for the preview) failed with
+  `invalid-mode`. The display now loads the plugin live for the session,
+  without writing `enabled` to `config.json`, and unloads it when on-demand
+  is stopped, expires or moves to another plugin. A plugin that fails to
+  load reports on-demand status `error` with `load-failed`. A session
+  restored after a restart unloads its disabled plugin the same way; it used
+  to stay loaded until the next restart.
+- A stop request now clears an on-demand error. After a failed request,
+  `/display/on-demand/status` kept reporting `status: error` for up to two
+  minutes even after a stop.
+- One hung plugin no longer stops every plugin from updating. The single
+  update worker waited on each plugin's lock with no time limit, and the
+  render thread holds that lock while it runs the plugin's display(); a
+  display() that never returned (or a first frame still running after the
+  executor's 30s timeout) parked the worker for good, so scores, weather and
+  clocks all froze while the panel kept scrolling. The worker now waits at
+  most 5s (the bound `unload_plugin()` already uses) and skips that update;
+  the other plugins keep updating. The skip is logged (at most once a minute
+  per plugin) and counted in plugin health as a busy skip (`busy_skip_count`,
+  `last_busy_skip`), but it is not a failure and never opens the circuit
+  breaker: Vegas mode holds a plugin's lock for its whole content render,
+  which on a slow Pi can outlast 5s, and a healthy plugin must not be pulled
+  from rotation for that.
+- display() calls are timed on every frame. One taking 2s or more is logged
+  (at most once a minute per plugin) and counted in plugin health
+  (`slow_call_count`, `last_slow_call`); one that runs past the executor's
+  timeout counts as a hang (`hang_count`, `last_hang`) and as a failure to
+  the circuit breaker. A first frame that times out is no longer recorded as
+  a success, and an update() still running after its timeout is recorded as
+  a hang instead of leaving the plugin silently stuck. Only these real hangs
+  count toward the breaker.
+- A plugin's `on_config_change()` no longer runs while its update() is
+  running on the worker thread. It now runs under the plugin's lock; if the
+  lock stays busy past the same 5s bound the change is handed to the update
+  worker, which applies the latest one as soon as the lock frees, and before
+  the plugin's next update() at the latest. The plugin API is unchanged.
+
 ## 3.7.0
 
 Sports consolidation stage 3 (#672). No behaviour change: nothing in core

docs/ARCHITECTURE.md

@@ -52,8 +52,10 @@ each other. They share three things:
 | Preview viewer marker | `/tmp/led_matrix_preview_viewer` | web, while a preview is open | display: writes full-rate snapshots only while it is fresh |
 | Hardware init status | `/tmp/led_matrix_hw_status.json` | display | web: `/api/v3/hardware/status` |
 
-The on-demand start route also restarts `ledmatrix.service` by default so the
+The on-demand start route starts `ledmatrix.service` when it is not running
-request takes effect straight away.
+(`start_service`, on by default) but never restarts a running one: the display
+reads the mailbox every `ON_DEMAND_POLL_INTERVAL` (0.25s), from its dwell
+sleep, its render loops and Vegas's interrupt check as well as the main loop.
 
 ## Display loop
 
@@ -82,7 +84,11 @@ then normal rotation.
 - **On-demand.** A request from the web interface pins one plugin (or mode)
   for a duration. `_activate_on_demand()` / `_clear_on_demand()`; the
   session is saved under `display_on_demand_config` so it survives a
-  restart. It also keeps the display on during scheduled off hours.
+  restart. It also keeps the display on during scheduled off hours. A
+  request for a plugin that is disabled in config loads it live
+  (`_load_plugin_for_on_demand()`, `load_plugin(force_enabled=True)`)
+  without writing `config.json`; the main loop unloads it once on-demand
+  moves off it (`_release_on_demand_plugins()`).
 - **Live priority.** `_check_live_priority()` looks for a plugin whose
   `has_live_priority()` and `has_live_content()` are both true and switches
   to it, rotating between several live games.
@@ -99,7 +105,8 @@ then normal rotation.
   changes. The controller refreshes its cached settings; enabling or
   disabling a plugin queues `_reconcile_enabled_plugins()`, which loads or
   unloads it on the display thread; each plugin gets `on_config_change()`
-  for its own section. Set `LEDMATRIX_HOT_RELOAD=false` to turn this off.
+  for its own section, under its plugin lock
+  (`PluginManager.apply_config_change()`). Set `LEDMATRIX_HOT_RELOAD=false` to turn this off.
   Matrix hardware settings are only read at start-up.
 - **Vegas mode.** [`src/vegas_mode/`](../src/vegas_mode/): the display loop
   calls `VegasModeCoordinator.run_iteration()`

docs/PLUGIN_API_REFERENCE.md

@@ -151,6 +151,12 @@ Clean up resources when plugin is unloaded. Override to close connections, stop
 
 Called after plugin configuration is updated via web API.
 
+In the display service it runs on the config watcher thread while holding
+the plugin's lock, so it never overlaps your `update()` or `display()`. If
+the plugin stays busy for more than 5 seconds, the change is applied later
+from the update thread: as soon as the plugin is free, and before its next
+`update()` at the latest.
+
 #### `on_enable() -> None`
 
 Called when plugin is enabled.

docs/REST_API_REFERENCE.md

@@ -390,7 +390,7 @@ Request a specific plugin to display on-demand.
 - `mode` (string, optional): Display mode name (plugin_id inferred if not provided)
 - `duration` (number, optional): Duration in seconds (0 = until stopped)
 - `pinned` (boolean, optional): Pin display (pause rotation)
-- `start_service` (boolean, optional): (Re)start the display service so it picks the request up (default: true)
+- `start_service` (boolean, optional): Start the display service if it is not running (default: true). A running service is never restarted: it picks the request up within about a quarter of a second. When false and the service is stopped, the route returns 400.
 
 **Response**:
 ```json

src/display_controller.py

@@ -29,7 +29,7 @@ import threading
 import types
 from collections import deque
 from contextlib import contextmanager
-from typing import Dict, Any, List, Optional, Callable, Tuple
+from typing import Dict, Any, List, Optional, Callable, Set, Tuple
 from datetime import datetime
 from concurrent.futures import ThreadPoolExecutor, as_completed  # pylint: disable=no-name-in-module
 import pytz
@@ -266,6 +266,10 @@ class DisplayController:
         self.on_demand_last_error: Optional[str] = None
         self.on_demand_last_event: Optional[str] = None
         self.on_demand_schedule_override = False
+        # Plugins that are disabled in config and loaded only because an
+        # on-demand request named them. The main loop unloads each one once
+        # on-demand has moved off it (_release_on_demand_plugins).
+        self._on_demand_loaded_plugins: Set[str] = set()
         self.rotation_resume_index: Optional[int] = None
         # Saved rotation position when a live-priority plugin preempts the
         # rotation, so it resumes where it left off (not after the live plugin)
@@ -369,7 +373,11 @@ class DisplayController:
                 """Load a single plugin and return result."""
                 plugin_load_start = time.time()
                 try:
-                    if self.plugin_manager.load_plugin(plugin_id):
+                    if plugin_id in self._on_demand_loaded_plugins:
+                        loaded = self.plugin_manager.load_plugin(plugin_id, force_enabled=True)
+                    else:
+                        loaded = self.plugin_manager.load_plugin(plugin_id)
+                    if loaded:
                         plugin_load_time = time.time() - plugin_load_start
                         return {
                             'success': True,
@@ -1021,17 +1029,28 @@ class DisplayController:
             accepts_display_mode: Whether display() takes ``display_mode``.
             force_clear: Passed through to display().
 
+        Each call is timed (two monotonic reads) and handed to
+        PluginManager.note_display_duration, which logs and records slow
+        calls and counts one that ran past the executor's timeout as a hang.
+
         Returns:
             display()'s result, or True when the frame was skipped because
             the plugin's update() holds its lock (the panel keeps the last
             frame; that is not a failure).
         """
-        with self._display_lock_or_skip(getattr(plugin, 'plugin_id', None)) as can_display:
+        plugin_id = getattr(plugin, 'plugin_id', None)
+        with self._display_lock_or_skip(plugin_id) as can_display:
             if not can_display:
                 return True
-            if accepts_display_mode:
+            started = time.monotonic()
-                return plugin.display(display_mode=mode, force_clear=force_clear)
+            try:
-            return plugin.display(force_clear=force_clear)
+                if accepts_display_mode:
+                    return plugin.display(display_mode=mode, force_clear=force_clear)
+                return plugin.display(force_clear=force_clear)
+            finally:
+                note = getattr(self.plugin_manager, 'note_display_duration', None)
+                if note is not None and plugin_id:
+                    note(plugin_id, time.monotonic() - started)
 
     def _health_tracker(self):
         """The plugin circuit breaker, or None when it is not enabled."""
@@ -1475,8 +1494,13 @@ class DisplayController:
         On-demand still resumes on its saved mode; this only widens what gets
         loaded, so normal rotation has somewhere to return to when it ends.
         A plugin that is disabled in config but named by the on-demand request
-        is still enabled and added, since otherwise the mode being resumed
+        is still loaded, since otherwise the mode being resumed would have
-        would have nothing behind it.
+        nothing behind it. It is tracked as loaded for on-demand only, the
+        same as one loaded live by _activate_on_demand, so it is unloaded
+        when the session ends instead of staying loaded until the next
+        restart. Its config section is not touched: setting ``enabled`` in
+        self.config wrote into the dict config_manager caches and returns to
+        every later load_config() in this process.
         """
         enabled_plugins = [p for p in discovered_plugins
                            if self.config.get(p, {}).get('enabled', False)]
@@ -1491,11 +1515,10 @@ class DisplayController:
             logger.warning("Falling back to normal mode (all enabled plugins)")
             return enabled_plugins
 
-        if not self.config.get(on_demand_plugin_id, {}).get('enabled', False):
+        if on_demand_plugin_id not in enabled_plugins:
-            logger.info("Temporarily enabling plugin '%s' for on-demand mode", on_demand_plugin_id)
+            logger.info("Loading disabled plugin '%s' for on-demand mode only", on_demand_plugin_id)
-            self.config.setdefault(on_demand_plugin_id, {})['enabled'] = True
+            self._on_demand_loaded_plugins.add(on_demand_plugin_id)
-            if on_demand_plugin_id not in enabled_plugins:
+            enabled_plugins.append(on_demand_plugin_id)
-                enabled_plugins.append(on_demand_plugin_id)
 
         # Restore on-demand state from the cached request so it resumes.
         self.on_demand_active = True
@@ -1591,6 +1614,11 @@ class DisplayController:
                 logger.debug("Stop request %s received but on-demand is not active", request_id)
                 # Still update request_id to acknowledge the request
                 self.on_demand_request_id = request_id
+                if self.on_demand_status == 'error':
+                    # A failed request left status 'error' published, and
+                    # without this the status route kept reporting it until
+                    # the state aged out (120s) or another request came in.
+                    self._clear_on_demand(reason='requested-stop')
             # Stop requests are deliberately exempt from the request_id/
             # processed_id guards above, so that a second click stops a mode
             # that a race left running. Consuming the mailbox is therefore the
@@ -1757,10 +1785,136 @@ class DisplayController:
                    plugin_id, ordered_modes, self.on_demand_mode_index, 
                    ordered_modes[self.on_demand_mode_index] if ordered_modes else 'N/A')
 
+    def _load_plugin_for_on_demand(self, plugin_id: str) -> bool:
+        """Load an installed plugin that isn't running so on-demand can show it.
+
+        This process only loads the plugins enabled in config, so a request
+        for a disabled one -- the config page's "Preview on display" button
+        offers it on every plugin -- failed with "invalid-mode" while the UI
+        said the plugin would be enabled for the session. Nothing did that
+        short of a restart, and restarts no longer happen on a request.
+
+        Loads through the same path as a live enable (load_plugin, then
+        _register_loaded_plugin), with force_enabled so the instance runs
+        enabled while config.json keeps saying disabled. The plugin is
+        recorded in _on_demand_loaded_plugins, and the main loop unloads it
+        once on-demand moves off it (_release_on_demand_plugins).
+
+        Returns False after publishing an error when the load fails. A
+        plugin that isn't installed returns True without loading anything:
+        the mode checks that follow report it as they always have.
+        """
+        if self.plugin_manager is None:
+            return True
+        try:
+            known = self.plugin_manager.discovered_plugin_ids()
+        except AttributeError:
+            known = set(getattr(self.plugin_manager, 'plugin_manifests', ()) or ())
+        if plugin_id not in known:
+            # Installed after this process scanned: the web process checked
+            # its own, fresher list before posting the request.
+            try:
+                known = set(self.plugin_manager.discover_plugins())
+            except Exception:  # pylint: disable=broad-except
+                logger.exception("On-demand: plugin discovery failed")
+                known = set()
+            if plugin_id not in known:
+                return True
+
+        logger.info("On-demand: loading disabled plugin '%s' for this session only", plugin_id)
+        self._on_demand_loaded_plugins.add(plugin_id)
+        try:
+            loaded = self.plugin_manager.load_plugin(plugin_id, force_enabled=True)
+            if loaded:
+                modes = self._register_loaded_plugin(plugin_id)
+                logger.info("On-demand: loaded plugin '%s' (modes: %s)", plugin_id, modes)
+        except Exception:  # pylint: disable=broad-except
+            logger.exception("On-demand: error loading plugin '%s'", plugin_id)
+            loaded = False
+        if not loaded:
+            # Stays in _on_demand_loaded_plugins so the main loop removes
+            # whatever part of it did get registered.
+            logger.error("On-demand: could not load plugin '%s'", plugin_id)
+            self._set_on_demand_error("load-failed")
+            return False
+        return True
+
+    def _release_on_demand_plugins(self) -> None:
+        """Unload plugins loaded only for on-demand that it has moved off.
+
+        Runs from the main loop, right after its own on-demand poll, not
+        where on-demand ends: a stop, an expiry or the next request is often
+        read from inside a render loop or a dwell sleep, where the plugin
+        being released may still be on the stack mid-display(). Unloading
+        goes through _unregister_plugin, as a live disable does, and nothing
+        is written to config.json.
+
+        A plugin the user enabled in the meantime stays loaded and takes its
+        place in the rotation, which is what the reconcile that the enable
+        queued would have done.
+        """
+        if self.plugin_manager is None:  # plugin system failed after startup restore
+            self._on_demand_loaded_plugins.clear()
+            return
+        keep = self.on_demand_plugin_id if self.on_demand_active else None
+        releasable = [p for p in self._on_demand_loaded_plugins if p != keep]
+        if not releasable:
+            return
+        try:
+            config = self.config_service.get_config()
+        except Exception as e:  # pylint: disable=broad-except
+            logger.warning("On-demand release: falling back to cached config: %s", e)
+            config = self.config
+        previous_mode = self.current_display_mode
+        for plugin_id in releasable:
+            self._on_demand_loaded_plugins.discard(plugin_id)
+            section = config.get(plugin_id)
+            if isinstance(section, dict) and section.get('enabled', False):
+                logger.info("On-demand: keeping plugin '%s' loaded; it was enabled "
+                            "while on-demand showed it", plugin_id)
+                continue
+            if (plugin_id in self.plugin_display_modes
+                    or self.plugin_manager.get_plugin(plugin_id) is not None):
+                logger.info("On-demand: unloading plugin '%s'; it is disabled in config",
+                            plugin_id)
+                self._unregister_plugin(plugin_id)
+        if not self.on_demand_active:
+            # Only outside a session: rotation_resume_index points into
+            # available_modes until the session ends.
+            self._apply_plugin_rotation_order()
+        self._resync_mode_index_after_change(previous_mode)
+        if self.current_display_mode != previous_mode:
+            self.force_change = True
+
+    def _rotation_index_outside_on_demand(self, start: int) -> Optional[int]:
+        """First index from `start` (wrapping) whose mode is not owned by a
+        plugin loaded only for on-demand, or None if every mode is.
+
+        Ending a session must not resume the rotation onto the plugin that
+        is about to be unloaded. A live load appends that plugin's modes
+        after the saved resume index, but a session restored after a
+        restart has no saved index and its plugin was ordered in with the
+        rest -- the rotation resumed onto it, and a stop read during its own
+        screen changed nothing on the panel until that screen ended.
+        """
+        if not self._on_demand_loaded_plugins:
+            return start
+        on_demand_only = {mode for plugin_id in self._on_demand_loaded_plugins
+                          for mode in self.plugin_display_modes.get(plugin_id, [])}
+        count = len(self.available_modes)
+        for step in range(count):
+            index = (start + step) % count
+            if self.available_modes[index] not in on_demand_only:
+                return index
+        return None
+
     def _activate_on_demand(self, request: Dict[str, Any]) -> None:
         """Activate on-demand mode for a specific plugin display."""
         plugin_id = request.get('plugin_id')
         mode = request.get('mode')
+        if (plugin_id and plugin_id not in self.plugin_display_modes
+                and not self._load_plugin_for_on_demand(plugin_id)):
+            return
         resolved_mode = self._resolve_mode_for_plugin(plugin_id, mode)
 
         if not resolved_mode:
@@ -1866,6 +2020,15 @@ class DisplayController:
                 self.on_demand_last_event = 'stop-request-ignored'  # Already idle
                 self._publish_on_demand_state()
             return
+        if not self.on_demand_active and self.on_demand_status == 'error':
+            # _set_on_demand_error already ended any session and dropped
+            # rotation_resume_index; the full clear below would only move
+            # the rotation and force a redraw. Just drop the error.
+            self.on_demand_status = 'idle'
+            self.on_demand_last_error = None
+            self.on_demand_last_event = reason or 'cleared'
+            self._publish_on_demand_state()
+            return
 
         self._reset_on_demand_fields()
         self.on_demand_status = 'idle'
@@ -1875,17 +2038,27 @@ class DisplayController:
         # Clear on-demand configuration from cache
         self.cache_manager.clear_cache('display_on_demand_config')
 
-        if self.rotation_resume_index is not None and self.available_modes:
+        if self.available_modes:
-            self.current_mode_index = self.rotation_resume_index % len(self.available_modes)
+            saved = self.rotation_resume_index
-            self.current_display_mode = self.available_modes[self.current_mode_index]
+            # Default to the current index if no resume index
-            logger.info("Resuming rotation from saved index %d: mode '%s'", 
+            start = saved if saved is not None else self.current_mode_index
-                       self.rotation_resume_index, self.current_display_mode)
+            index = self._rotation_index_outside_on_demand(start % len(self.available_modes))
-        elif self.available_modes:
+            if index is None:
-            # Default to first mode if no resume index
+                # Every mode belongs to a plugin loaded only for on-demand,
-            self.current_mode_index = self.current_mode_index % len(self.available_modes)
+                # which the main loop is about to unload; it then idles.
-            self.current_display_mode = self.available_modes[self.current_mode_index]
+                self.current_mode_index = 0
-            logger.info("Resuming rotation to mode '%s' (index %d)", 
+                self.current_display_mode = None
-                       self.current_display_mode, self.current_mode_index)
+                logger.info("No enabled mode to resume rotation to")
+            elif saved is not None:
+                self.current_mode_index = index
+                self.current_display_mode = self.available_modes[index]
+                logger.info("Resuming rotation from saved index %d: mode '%s'",
+                            saved, self.current_display_mode)
+            else:
+                self.current_mode_index = index
+                self.current_display_mode = self.available_modes[index]
+                logger.info("Resuming rotation to mode '%s' (index %d)",
+                            self.current_display_mode, self.current_mode_index)
         else:
             logger.warning("No available modes to resume rotation to")
 
@@ -2082,6 +2255,14 @@ class DisplayController:
                 # Handle on-demand commands before rendering
                 self._poll_on_demand_requests()
                 self._check_on_demand_expiration()
+                # Unload plugins loaded only to show them on-demand once it
+                # has moved off them. Here, where no display() is on the
+                # stack; one ended from inside a screen is caught here on
+                # the next pass.
+                if self._on_demand_loaded_plugins:
+                    self._release_on_demand_plugins()
+                    if not self.available_modes:
+                        continue  # it was all there was; idle as above
                 self._tick_plugin_updates()
                 
                 # Clean up expired WiFi status messages
@@ -2328,6 +2509,7 @@ class DisplayController:
                         pm = self.plugin_manager
                         display_lock = pm.get_plugin_lock(plugin_id) if pm else None
                         can_display = display_lock is None or display_lock.acquire(blocking=False)
+                        display_hung = False
 
                         if display_lock is None:
                             # Only when plugin loading failed part-way.
@@ -2347,7 +2529,7 @@ class DisplayController:
                             # thread actually finishes it, rather than
                             # here when this dispatch merely returns.
                             release_guard = threading.Lock()
-                            released = {'done': False}
+                            released = {'done': False, 'started': False}
 
                             def _release_display_lock():
                                 with release_guard:
@@ -2358,6 +2540,7 @@ class DisplayController:
 
                             if _accepts_display_mode:
                                 def _display_target(display_mode=None, force_clear=False):
+                                    released['started'] = True
                                     try:
                                         return manager_to_display.display(
                                             display_mode=display_mode, force_clear=force_clear)
@@ -2365,11 +2548,13 @@ class DisplayController:
                                         _release_display_lock()
                             else:
                                 def _display_target(force_clear=False):
+                                    released['started'] = True
                                     try:
                                         return manager_to_display.display(force_clear=force_clear)
                                     finally:
                                         _release_display_lock()
 
+                            dispatch_start = time.monotonic()
                             try:
                                 result = pm.plugin_executor.execute_display(
                                     types.SimpleNamespace(display=_display_target),
@@ -2392,6 +2577,18 @@ class DisplayController:
                                 _release_display_lock()
                                 raise
 
+                            dispatch_seconds = time.monotonic() - dispatch_start
+                            if released['started'] and not released['done']:
+                                # The executor gave up waiting and display()
+                                # is still running on its thread, holding
+                                # the lock. A hang, not a success: recorded
+                                # so repeats open the circuit breaker, and
+                                # the update worker's bounded wait skips it.
+                                display_hung = True
+                                pm.record_display_hang(plugin_id, dispatch_seconds)
+                            else:
+                                pm.note_display_duration(plugin_id, dispatch_seconds)
+
                         logger.debug(f"display() returned: {result} (type: {type(result)})")
                         if isinstance(result, bool):
                             display_result = result
@@ -2405,7 +2602,7 @@ class DisplayController:
                         # be lost when display() finally does run.
                         if can_display:
                             health_tracker = self._health_tracker()
-                            if health_tracker is not None:
+                            if health_tracker is not None and not display_hung:
                                 health_tracker.record_success(plugin_id)
                             self.force_change = False
                     except Exception as exc:  # pylint: disable=broad-except
@@ -3099,8 +3296,23 @@ class DisplayController:
                     prepared = prepare(_pid, new_config) if callable(prepare) else None
                     if isinstance(prepared, dict):
                         new_config = prepared
-                    _plugin.on_config_change(new_config)
+                    if _pid in self._on_demand_loaded_plugins:
-                    logger.debug("Plugin %s notified of config change", _pid)
+                        # Saved while on-demand shows it: config.json still
+                        # says disabled, and on_config_change would switch
+                        # the instance off mid-session.
+                        new_config = {**new_config, 'enabled': True}
+                    # Runs on ConfigService's watcher thread. Under the
+                    # plugin's lock, so it cannot interleave with update()
+                    # on the worker or display() on the render thread; a
+                    # lock held past the bound defers it to the worker.
+                    apply = getattr(self.plugin_manager, 'apply_config_change', None)
+                    if callable(apply):
+                        applied = apply(_pid, new_config, plugin_instance=_plugin)
+                    else:
+                        _plugin.on_config_change(new_config)
+                        applied = True
+                    logger.debug("Plugin %s notified of config change%s", _pid,
+                                 "" if applied else " (deferred: plugin busy)")
                 except Exception as e:
                     logger.error("Error in plugin %s config change handler: %s", _pid, e, exc_info=True)
 

src/plugin_system/plugin_executor.py

@@ -19,8 +19,25 @@ class PluginTimeoutError(Exception):
     """Raised when a plugin operation times out."""
 
 
+class PluginBusyError(PluginTimeoutError):
+    """A plugin's lock stayed held past its bound.
+
+    Not raised; recorded. The lock is held by the plugin's own display(),
+    update(), on_config_change() or a Vegas content render -- slow, or hung
+    -- so the caller skipped the plugin rather than wait on it. Report-only:
+    it is kept as the plugin's state error info and counted as a busy skip in
+    health, never as a failure, so it cannot open the circuit breaker.
+    """
+
+
 class PluginExecutor:
     """Handles plugin execution with timeout and error isolation."""
+
+    #: A display() call at least this long is logged and counted as slow.
+    #: A frame is milliseconds; two seconds is a plugin doing I/O in display().
+    SLOW_DISPLAY_SECONDS = 2.0
+    #: An update() call at least this long is logged as slow.
+    SLOW_UPDATE_SECONDS = 5.0
     
     def __init__(
         self,
@@ -117,15 +134,15 @@ class PluginExecutor:
             True if update succeeded, False otherwise
         """
         try:
-            start_time = time.time()
+            start_time = time.monotonic()
             self.execute_with_timeout(
                 lambda: plugin.update(),
                 timeout=timeout,
                 plugin_id=plugin_id
             )
-            duration = time.time() - start_time
+            duration = time.monotonic() - start_time
             
-            if duration > 5.0:  # Warn if update takes more than 5 seconds
+            if duration > self.SLOW_UPDATE_SECONDS:
                 self.logger.warning(
                     "Plugin %s update() took %.2fs (consider optimizing)",
                     plugin_id,
@@ -175,7 +192,7 @@ class PluginExecutor:
             True if display succeeded, False otherwise
         """
         try:
-            start_time = time.time()
+            start_time = time.monotonic()
             
             # Does display() take a display_mode keyword? The caller usually
             # knows and caches the answer, so prefer what it passed.
@@ -206,9 +223,9 @@ class PluginExecutor:
                     plugin_id=plugin_id
                 )
             
-            duration = time.time() - start_time
+            duration = time.monotonic() - start_time
             
-            if duration > 2.0:  # Warn if display takes more than 2 seconds
+            if duration > self.SLOW_DISPLAY_SECONDS:
                 self.logger.warning(
                     "Plugin %s display() took %.2fs (consider optimizing)",
                     plugin_id,

src/plugin_system/plugin_health.py

@@ -254,6 +254,104 @@ class PluginHealthTracker:
         
         self._save_health_state(plugin_id, state)
     
+    def record_hang(self, plugin_id: str, operation: str, seconds: float,
+                    error: Optional[Exception] = None) -> None:
+        """Record a display() or update() call that ran past its limit.
+
+        Counts as a failure, so the ordinary circuit breaker handles a plugin
+        that keeps hanging: after ``failure_threshold`` in a row it is skipped
+        by both the update scheduler and the display rotation until the
+        cooldown ends. The hang itself is kept alongside (``hang_count``,
+        ``last_hang``) so the health API can tell "hung" from "raised".
+
+        Not for an update skipped because the plugin's lock stayed held: the
+        holder may be a healthy but long render (Vegas prefetch). That is
+        :meth:`record_busy_skip`, which never touches the breaker.
+
+        Args:
+            plugin_id: Plugin identifier
+            operation: What hung: ``"display"`` or ``"update"``.
+            seconds: How long it had been running when this was recorded.
+            error: The error to store as ``last_error``; one is built from
+                the other arguments when omitted.
+        """
+        state = self.get_health_state(plugin_id)
+        count = state.get('hang_count')
+        state['hang_count'] = (count if isinstance(count, int) and not isinstance(count, bool)
+                               else 0) + 1
+        state['last_hang'] = {
+            'operation': operation,
+            'seconds': round(float(seconds), 3),
+            'time': time.time(),
+        }
+        if error is None:
+            error = TimeoutError(f"{operation} still running after {seconds:.1f}s")
+        # record_failure saves the record, hang fields included.
+        self.record_failure(plugin_id, error)
+
+    #: Minimum seconds between persisting a plugin's slow-call or busy-skip
+    #: counters. The in-memory record is updated every time; a plugin that is
+    #: slow on every frame must not become an SD-card write per frame.
+    SLOW_CALL_PERSIST_INTERVAL = 60.0
+
+    def record_slow_call(self, plugin_id: str, operation: str, seconds: float) -> None:
+        """Note a call that finished, but slowly. Reporting only.
+
+        Unlike :meth:`record_hang` this never touches the circuit breaker: a
+        slow display() still drew its frame.
+        """
+        state = self.get_health_state(plugin_id)
+        count = state.get('slow_call_count')
+        state['slow_call_count'] = (count if isinstance(count, int) and not isinstance(count, bool)
+                                    else 0) + 1
+        now = time.time()
+        state['last_slow_call'] = {
+            'operation': operation,
+            'seconds': round(float(seconds), 3),
+            'time': now,
+        }
+        self._save_reporting_throttled('slow', plugin_id, state, now)
+
+    def record_busy_skip(self, plugin_id: str, operation: str, seconds: float) -> None:
+        """Note a call skipped because the plugin's lock stayed held. Reporting only.
+
+        The update worker gives up on a plugin's lock after
+        ``PluginManager.PLUGIN_LOCK_TIMEOUT``. Whatever held it may be healthy
+        -- Vegas prefetch holds the lock for a plugin's whole content render,
+        which on a slow Pi can take longer than that -- so like
+        :meth:`record_slow_call` this never touches the circuit breaker, the
+        failure streak or ``last_error``. A real hang is recorded by
+        :meth:`record_hang` where it is measured.
+
+        Args:
+            plugin_id: Plugin identifier
+            operation: What was skipped, e.g. ``"update lock wait"``.
+            seconds: How long the lock was waited on.
+        """
+        state = self.get_health_state(plugin_id)
+        count = state.get('busy_skip_count')
+        state['busy_skip_count'] = (count if isinstance(count, int) and not isinstance(count, bool)
+                                    else 0) + 1
+        now = time.time()
+        state['last_busy_skip'] = {
+            'operation': operation,
+            'seconds': round(float(seconds), 3),
+            'time': now,
+        }
+        self._save_reporting_throttled('busy', plugin_id, state, now)
+
+    def _save_reporting_throttled(self, kind: str, plugin_id: str,
+                                  state: Dict[str, Any], now: float) -> None:
+        """Persist a reporting-only change at most once per
+        SLOW_CALL_PERSIST_INTERVAL per plugin and ``kind``. The first one is
+        saved at once, so the web process (which reads the persisted record)
+        sees it; repeats in between stay in memory until the next save."""
+        saved_at = self.__dict__.setdefault('_reporting_saved_at', {})
+        last = saved_at.get((kind, plugin_id))
+        if last is None or now - last >= self.SLOW_CALL_PERSIST_INTERVAL:
+            saved_at[(kind, plugin_id)] = now
+            self._save_health_state(plugin_id, state)
+
     def set_degraded(self, plugin_id: str, reason: Optional[str]) -> None:
         """Flag (or clear) a plugin as degraded without touching the circuit breaker.
 
@@ -345,7 +443,13 @@ class PluginHealthTracker:
             'degraded': state.get('degraded', False),
             'degraded_reason': state.get('degraded_reason'),
             'circuit_opened_time': state.get('circuit_opened_time'),
-            'half_open_start_time': state.get('half_open_start_time')
+            'half_open_start_time': state.get('half_open_start_time'),
+            'hang_count': state.get('hang_count', 0),
+            'last_hang': state.get('last_hang'),
+            'slow_call_count': state.get('slow_call_count', 0),
+            'last_slow_call': state.get('last_slow_call'),
+            'busy_skip_count': state.get('busy_skip_count', 0),
+            'last_busy_skip': state.get('last_busy_skip'),
         }
     
     def get_all_health_summaries(self) -> Dict[str, Dict[str, Any]]:

src/plugin_system/plugin_manager.py

@@ -16,12 +16,14 @@ import time
 import threading
 import types
 from pathlib import Path
-from typing import Dict, List, Optional, Any, Tuple
+from typing import Dict, List, NamedTuple, Optional, Any, Tuple, Union
 import logging
 from src.exceptions import PluginError, ConfigError
 from src.logging_config import get_logger
 from src.plugin_system.plugin_loader import PluginLoader
-from src.plugin_system.plugin_executor import PluginExecutor
+from src.plugin_system.plugin_executor import (
+    PluginBusyError, PluginExecutor, PluginTimeoutError,
+)
 from src.plugin_system.plugin_state import PluginStateManager, PluginState
 from src.plugin_system.schema_manager import (
     CORE_VEGAS_TUNING_KEYS, SchemaManager, normalize_legacy_booleans,
@@ -36,6 +38,16 @@ from src.common.permission_utils import (
 )
 
 
+class _DeferredConfigChange(NamedTuple):
+    """Update-queue item: apply the config change parked for ``plugin_id``.
+
+    Queued by apply_config_change() when the plugin's lock was busy; the
+    change itself waits in ``PluginManager._deferred_config_changes`` so only
+    the latest one is ever applied.
+    """
+    plugin_id: str
+
+
 class PluginManager:
     """
     Manages plugin discovery, loading, and lifecycle.
@@ -56,6 +68,19 @@ class PluginManager:
     # How long unload_plugin() waits for an in-flight update() to finish
     # before tearing the instance down anyway.
     UNLOAD_LOCK_TIMEOUT = 5.0
+
+    # How long the update worker and apply_config_change() wait for a
+    # plugin's lock -- the same bound unload already uses for the same lock.
+    # A display() frame holds it for milliseconds, so this only runs out when
+    # the holder is hung or pathologically slow. The worker then skips that
+    # plugin (recorded as a hang, so repeats open its circuit breaker)
+    # instead of stalling every other plugin's update behind it.
+    PLUGIN_LOCK_TIMEOUT = UNLOAD_LOCK_TIMEOUT
+
+    # Minimum seconds between repeats of the same hang/slow-call warning for
+    # one plugin. A hung plugin is re-detected every interval; a slow
+    # display() can be re-detected every frame.
+    HANG_LOG_INTERVAL = 60.0
     
     def __init__(self, plugins_dir: str = "plugins", 
                  config_manager: Optional[Any] = None, 
@@ -122,7 +147,33 @@ class PluginManager:
         # 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()
+        #
+        # Which thread runs each plugin hook, and what it holds:
+        #   __init__, on_enable   the loading thread (main thread at startup,
+        #                         the render thread on a live enable).
+        #   update()              plugin-update-worker, under the plugin lock,
+        #                         via PluginExecutor (whose daemon thread runs
+        #                         the call; if it outlives the executor's
+        #                         timeout it keeps the lock until it returns).
+        #                         Exceptions: the startup pass
+        #                         (DisplayController._run_initial_updates, main
+        #                         thread, before the display loop starts) and
+        #                         the synchronous_updates kill switch (render
+        #                         thread) run it without the lock.
+        #   display()             the render thread, under a try-lock: a busy
+        #                         lock skips the frame. The first frame of a
+        #                         screen goes through PluginExecutor. Vegas
+        #                         mode's adapter and coordinator take the lock
+        #                         with a bounded wait.
+        #   on_config_change()    ConfigService's watcher thread, under the
+        #                         plugin lock via apply_config_change(); if the
+        #                         lock stays busy it is deferred to the update
+        #                         worker, which applies it under the lock.
+        #   cleanup(), on_disable()  whoever calls unload_plugin(), under the
+        #                         lock with UNLOAD_LOCK_TIMEOUT.
+        # No wait on a plugin lock is unbounded, so one hung plugin can only
+        # cost the worker PLUGIN_LOCK_TIMEOUT per attempt.
+        self._update_queue: "queue.Queue[Union[None, Tuple[str, float], _DeferredConfigChange]]" = queue.Queue()
         self._pending_updates: set = set()
         self._pending_lock = threading.Lock()
         # Serializes the "is this plugin eligible?" -> "claim it (RUNNING)"
@@ -145,6 +196,12 @@ class PluginManager:
         # run_scheduled_updates_with_changes().
         self._completed_updates: set = set()
         self._completed_updates_lock = threading.Lock()
+        # Config changes that found the plugin's lock busy, latest per plugin,
+        # with the instance they were meant for. See apply_config_change().
+        self._deferred_config_changes: Dict[str, Tuple[Any, Dict[str, Any]]] = {}
+        self._deferred_config_lock = threading.Lock()
+        # key -> (monotonic time last logged, repeats suppressed since)
+        self._rate_limited_warnings: Dict[str, Tuple[float, int]] = {}
         self._synchronous_updates = False
         if self.config_manager is not None:
             try:
@@ -296,7 +353,7 @@ class PluginManager:
 
         return plugin_ids
 
-    def load_plugin(self, plugin_id: str) -> bool:
+    def load_plugin(self, plugin_id: str, force_enabled: bool = False) -> bool:
         """
         Load a plugin by ID.
         
@@ -310,6 +367,10 @@ class PluginManager:
         
         Args:
             plugin_id: Plugin identifier
+            force_enabled: Run the plugin enabled even though config.json has
+                it disabled. On-demand uses this to show a disabled plugin
+                (DisplayController._load_plugin_for_on_demand). Only the
+                instance's config says enabled; config.json is not written.
             
         Returns:
             True if loaded successfully, False otherwise
@@ -376,6 +437,12 @@ class PluginManager:
             # (prepare_plugin_config). In memory only: config.json is written
             # by saves, never by loading a plugin.
             config = self.prepare_plugin_config(plugin_id, config, schema=schema)
+            if force_enabled:
+                # A copy: prepare_plugin_config can hand back the section from
+                # config_manager's cached config, and setting the flag there
+                # would read as enabled to everything else in this process.
+                config = dict(config)
+                config['enabled'] = True
             
             # Use PluginLoader to load plugin
             plugin_instance, _module = self.plugin_loader.load_plugin(
@@ -676,6 +743,8 @@ class PluginManager:
             
             # Remove from active plugins
             del self.plugins[plugin_id]
+            with self._deferred_config_lock:
+                self._deferred_config_changes.pop(plugin_id, None)
             with self._plugin_last_update_lock:
                 self.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
@@ -1012,6 +1081,8 @@ class PluginManager:
         self,
         plugin_id: str,
         exc: Optional[Exception] = None,
+        log: bool = True,
+        count_failure: bool = True,
     ) -> None:
         """Apply the standard failure-recovery path for a plugin update.
 
@@ -1025,6 +1096,11 @@ class PluginManager:
             exc: The exception that caused the failure, if any.  When None a
                  synthetic ExecutionFailure exception is constructed from the
                  timeout/executor-error path.
+            log: Log the generic failure line. Callers that already logged
+                 something more specific (rate-limited) pass False.
+            count_failure: Record the failure in plugin health, where it
+                 counts toward the circuit breaker. A busy skip passes False:
+                 it records itself as a busy skip, reporting only.
         """
         failure_time = time.time()
         if exc is not None:
@@ -1040,13 +1116,91 @@ class PluginManager:
             'timestamp': failure_time,
             'recoverable': True,
         }
-        self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
+        if log:
+            self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
         with self._plugin_last_update_lock:
             self.plugin_last_update[plugin_id] = failure_time
         self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info)
-        if self.health_tracker:
+        if count_failure and self.health_tracker:
             self.health_tracker.record_failure(plugin_id, err)
 
+    def _warn_rate_limited(self, key: str, message: str, *args: Any) -> None:
+        """Log a warning at most once per HANG_LOG_INTERVAL for ``key``.
+
+        Repeats in between are counted and the count is appended to the next
+        one that is logged, so the journal shows the problem continuing
+        without a line per frame or per scheduler tick.
+        """
+        # setdefault: tests build bare managers with PluginManager.__new__.
+        seen = self.__dict__.setdefault('_rate_limited_warnings', {})
+        now = time.monotonic()
+        last, suppressed = seen.get(key, (None, 0))
+        if last is not None and now - last < self.HANG_LOG_INTERVAL:
+            seen[key] = (last, suppressed + 1)
+            return
+        seen[key] = (now, 0)
+        if suppressed:
+            message += " (%d more since the last warning)"
+            args = args + (suppressed,)
+        self.logger.warning(message, *args)
+
+    def _record_hang(self, plugin_id: str, operation: str, seconds: float,
+                     err: Exception) -> None:
+        """Record a hang in plugin health: a failure to the circuit breaker.
+
+        PluginHealthTracker.record_hang also counts the hang separately. Never
+        raises: this runs on the update worker and the render thread.
+        """
+        tracker = self.health_tracker
+        if tracker is None:
+            return
+        try:
+            tracker.record_hang(plugin_id, operation, seconds, err)
+        except Exception as e:  # pylint: disable=broad-except
+            self.logger.debug("Could not record hang for %s: %s", plugin_id, e)
+
+    def note_display_duration(self, plugin_id: str, seconds: float) -> None:
+        """Account for one display() call that took ``seconds``.
+
+        Called by the render loop for every frame, so the common case is one
+        comparison. At or above PluginExecutor.SLOW_DISPLAY_SECONDS the call
+        is logged (rate-limited) and counted as slow in plugin health; at or
+        above the executor's timeout -- the limit the first frame of a screen
+        is already held to -- it is recorded as a hang, which the circuit
+        breaker counts as a failure.
+        """
+        if seconds < PluginExecutor.SLOW_DISPLAY_SECONDS:
+            return
+        if seconds >= self.plugin_executor.default_timeout:
+            self.record_display_hang(plugin_id, seconds)
+            return
+        self._warn_rate_limited(
+            "slow-display:" + plugin_id,
+            "Plugin %s display() took %.2fs; a frame should take milliseconds "
+            "(is it fetching or loading files in display()?)", plugin_id, seconds)
+        tracker = self.health_tracker
+        record_slow = getattr(tracker, 'record_slow_call', None) if tracker is not None else None
+        if callable(record_slow):
+            try:
+                record_slow(plugin_id, 'display', seconds)
+            except Exception as e:  # pylint: disable=broad-except
+                self.logger.debug("Could not record slow display for %s: %s", plugin_id, e)
+
+    def record_display_hang(self, plugin_id: str, seconds: float) -> None:
+        """Record a display() call that ran ``seconds``, past its limit.
+
+        Either it has since returned (note_display_duration) or it is still
+        running on the executor's lingering thread, holding the plugin's lock
+        (the render loop's first-frame dispatch).
+        """
+        self._warn_rate_limited(
+            "hung-display:" + plugin_id,
+            "Plugin %s display() ran for at least %.1fs (limit %.0fs); recorded "
+            "as a hang -- repeated hangs open its circuit breaker",
+            plugin_id, seconds, self.plugin_executor.default_timeout)
+        self._record_hang(plugin_id, 'display', seconds, PluginTimeoutError(
+            f"Plugin {plugin_id} display() ran for at least {seconds:.1f}s"))
+
     def run_scheduled_updates(self, current_time: Optional[float] = None) -> None:
         """
         Trigger plugin updates based on their defined update intervals.
@@ -1132,11 +1286,13 @@ class PluginManager:
         self.state_manager.set_state(plugin_id, PluginState.ENABLED)
 
     def get_plugin_lock(self, plugin_id: str) -> threading.Lock:
-        """Per-plugin lock keeping update() and display() mutually exclusive.
+        """Per-plugin lock keeping update(), display() and on_config_change()
+        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.
+        display() call when the plugin is mid-update. Every other waiter uses
+        a bounded acquire (see the thread notes in __init__).
         """
         with self._plugin_locks_guard:
             lock = self._plugin_locks.get(plugin_id)
@@ -1202,14 +1358,28 @@ class PluginManager:
         real update() call genuinely finishes (see _execute_update_now),
         which can be after this dispatch returns if PluginExecutor's own
         timeout elapses first.
+
+        The lock wait is bounded by PLUGIN_LOCK_TIMEOUT. Whatever holds it
+        past that -- a hung display() on the render thread, a lingering
+        executor thread, or a long but healthy Vegas content render -- costs
+        this worker that long once per attempt, and the plugin's update is
+        skipped and reported as a busy skip (_skip_busy_update), which never
+        counts toward the circuit breaker; the other plugins' queued updates
+        carry on.
         """
         while True:
             item = self._update_queue.get()
             if item is None:  # shutdown sentinel
                 return
+            if isinstance(item, _DeferredConfigChange):
+                self._apply_deferred_config_change(item.plugin_id)
+                continue
             plugin_id, scheduled_time = item
             lock = self.get_plugin_lock(plugin_id)
-            lock.acquire()
+            wait_start = time.monotonic()
+            if not lock.acquire(timeout=self.PLUGIN_LOCK_TIMEOUT):
+                self._skip_busy_update(plugin_id, time.monotonic() - wait_start)
+                continue
             plugin_instance = self.plugins.get(plugin_id)
             if plugin_instance is None:  # unloaded while queued; its
                 # lifecycle state was already cleared by unload_plugin —
@@ -1218,6 +1388,9 @@ class PluginManager:
                 with self._pending_lock:
                     self._pending_updates.discard(plugin_id)
                 continue
+            # A config change that found the lock busy goes in first, so
+            # this update() runs against the settings the user saved.
+            self._apply_deferred_config_locked(plugin_id, plugin_instance)
             try:
                 self._execute_update_now(plugin_id, plugin_instance,
                                          scheduled_time, lock=lock)
@@ -1228,6 +1401,142 @@ class PluginManager:
                 self.logger.exception("update worker: unexpected error for %s",
                                       plugin_id)
 
+    def _skip_busy_update(self, plugin_id: str, waited: float) -> None:
+        """Give up on a queued update whose plugin lock stayed held.
+
+        Same bookkeeping as a failed update() -- pending slot dropped before
+        the state returns to ENABLED with PluginBusyError error info,
+        last-update stamped so the retry waits a full interval -- but
+        report-only in health: counted as a busy skip (``busy_skip_count`` /
+        ``last_busy_skip``), never as a failure or a hang. The lock holder
+        may be perfectly healthy: Vegas prefetch holds a plugin's lock for its
+        whole content render, which on a slow Pi can outlast
+        PLUGIN_LOCK_TIMEOUT, and counting that would pull a healthy plugin
+        from rotation. Real hangs -- display() or update() past the executor
+        timeout -- are recorded where they are measured and still open the
+        breaker.
+        """
+        with self._pending_lock:
+            self._pending_updates.discard(plugin_id)
+        if plugin_id not in self.plugins:
+            # Unloaded while we waited: its lifecycle state is already
+            # cleared; recording anything would resurrect it as ENABLED.
+            return
+        self._warn_rate_limited(
+            "busy-update:" + plugin_id,
+            "Plugin %s update skipped: its lock was still held after %.1fs "
+            "(a display(), Vegas render or update() of it is still running); "
+            "retrying next interval, not counted as a failure", plugin_id, waited)
+        self._record_update_failure(
+            plugin_id,
+            exc=PluginBusyError(
+                f"Plugin {plugin_id} busy: its lock was held for over {waited:.1f}s "
+                "by a slow or hung display()/update(); update skipped"),
+            log=False,
+            count_failure=False)
+        tracker = self.health_tracker
+        record_busy = getattr(tracker, 'record_busy_skip', None) if tracker is not None else None
+        if callable(record_busy):
+            try:
+                record_busy(plugin_id, 'update lock wait', waited)
+            except Exception as e:  # pylint: disable=broad-except
+                self.logger.debug("Could not record busy skip for %s: %s", plugin_id, e)
+
+    def apply_config_change(self, plugin_id: str, new_config: Dict[str, Any],
+                            plugin_instance: Optional[Any] = None) -> bool:
+        """Call ``on_config_change(new_config)`` without racing update()/display().
+
+        Runs on the calling thread -- ConfigService's watcher, for the display
+        service -- holding the plugin's lock, waited on for at most
+        PLUGIN_LOCK_TIMEOUT. If the lock is still busy (an update() mid-fetch
+        can outlast that) the change is parked and handed to the update
+        worker, which applies it under the same lock once it is free, and at
+        the latest just before the plugin's next update(). A later change for
+        the same plugin replaces a parked one.
+
+        Exceptions from on_config_change propagate on the immediate path,
+        as they did when the caller invoked it directly.
+
+        Args:
+            plugin_id: Plugin identifier.
+            new_config: The prepared config to hand the plugin.
+            plugin_instance: The instance to notify; defaults to the loaded one.
+
+        Returns:
+            True if on_config_change ran now, False if it was deferred or there
+            is no loaded plugin to notify.
+        """
+        if plugin_instance is None:
+            plugin_instance = self.plugins.get(plugin_id)
+        if plugin_instance is None or not hasattr(plugin_instance, 'on_config_change'):
+            return False
+        lock = self.get_plugin_lock(plugin_id)
+        if lock.acquire(timeout=self.PLUGIN_LOCK_TIMEOUT):
+            try:
+                with self._deferred_config_lock:
+                    # This change supersedes any older one still parked.
+                    self._deferred_config_changes.pop(plugin_id, None)
+                plugin_instance.on_config_change(new_config)
+            finally:
+                lock.release()
+            return True
+
+        with self._deferred_config_lock:
+            self._deferred_config_changes[plugin_id] = (plugin_instance, new_config)
+        self._warn_rate_limited(
+            "busy-config:" + plugin_id,
+            "Plugin %s is busy (lock held for over %.1fs); its config change "
+            "will be applied by the update worker once it is free",
+            plugin_id, self.PLUGIN_LOCK_TIMEOUT)
+        try:
+            self._ensure_update_worker()
+            self._update_queue.put(_DeferredConfigChange(plugin_id))
+        except Exception as exc:  # pylint: disable=broad-except
+            # No worker (thread start refused): still parked, so the next
+            # update() of this plugin applies it.
+            self.logger.error(
+                "Could not queue the config change for plugin %s (%s: %s); it "
+                "will be applied before its next update()",
+                plugin_id, type(exc).__name__, exc)
+        return False
+
+    def _apply_deferred_config_change(self, plugin_id: str) -> None:
+        """Worker side of a parked config change: take the lock, apply it."""
+        with self._deferred_config_lock:
+            if plugin_id not in self._deferred_config_changes:
+                return  # applied or superseded meanwhile
+        lock = self.get_plugin_lock(plugin_id)
+        wait_start = time.monotonic()
+        if not lock.acquire(timeout=self.PLUGIN_LOCK_TIMEOUT):
+            self._warn_rate_limited(
+                "busy-config:" + plugin_id,
+                "Plugin %s still busy after %.1fs; its config change stays "
+                "parked until its next update()",
+                plugin_id, time.monotonic() - wait_start)
+            return
+        try:
+            self._apply_deferred_config_locked(plugin_id, self.plugins.get(plugin_id))
+        finally:
+            lock.release()
+
+    def _apply_deferred_config_locked(self, plugin_id: str,
+                                      current_instance: Optional[Any]) -> None:
+        """Apply the parked config change for plugin_id; caller holds its lock."""
+        with self._deferred_config_lock:
+            entry = self._deferred_config_changes.pop(plugin_id, None)
+        if entry is None:
+            return
+        instance, new_config = entry
+        if current_instance is None or instance is not current_instance:
+            # Unloaded, or reloaded as a new instance built from the current
+            # config: nothing left to tell.
+            return
+        try:
+            instance.on_config_change(new_config)
+            self.logger.info("Applied deferred config change for plugin %s", plugin_id)
+        except Exception:  # pylint: disable=broad-except
+            self.logger.exception("Error in plugin %s config change handler", 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():
@@ -1338,14 +1647,29 @@ class PluginManager:
             else:
                 _finish(True)
 
+        started = time.monotonic()
         try:
-            self.plugin_executor.execute_update(
+            success = 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)
+            return
+        if not success and not finished['done']:
+            # The executor stopped waiting but update() is still running: it
+            # keeps the lock and the RUNNING state until it returns (then
+            # _finish records the outcome). Say so now, rather than leave the
+            # plugin silently stuck; record_success on a late return clears it.
+            elapsed = time.monotonic() - started
+            self._warn_rate_limited(
+                "hung-update:" + plugin_id,
+                "Plugin %s update() still running after %.1fs; it keeps its "
+                "lock until it returns, and is not rescheduled until then",
+                plugin_id, elapsed)
+            self._record_hang(plugin_id, 'update', elapsed, PluginTimeoutError(
+                f"Plugin {plugin_id} update() still running after {elapsed:.1f}s"))
 
     def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
         """

test/conftest.py

@@ -16,7 +16,54 @@ if str(project_root) not in sys.path:
     sys.path.insert(0, str(project_root))
 
 
+class _DisarmStartupReconciliation:
+    """Import hook: every ``web_interface.app`` this process builds starts disarmed.
+
+    app.py wires itself to the checkout's real config/config.json and
+    plugin-repos/ at import, and its before_request hook launches startup
+    reconciliation on the first request any test sends. Reconciliation
+    reinstalls every configured plugin missing on disk from the live store,
+    so a full run downloaded basketball-scoreboard, calendar,
+    football-scoreboard, leaderboard and ledmatrix-stocks into the real
+    plugin-repos/ (not gitignored), minutes in, from a daemon thread no test
+    waits on. Setting ``_reconciliation_started`` is the app's own run-once
+    latch; doing it as the module finishes executing covers fixtures that
+    import the app lazily and send a request at once, and ``importlib.reload``.
+    StateReconciliation itself stays fully testable.
+    """
+
+    _MODULE = "web_interface.app"
+
+    def find_spec(self, fullname, path, target=None):
+        if fullname != self._MODULE:
+            return None
+        import importlib.machinery
+        spec = importlib.machinery.PathFinder.find_spec(fullname, path, target)
+        if spec is None or spec.loader is None:
+            return spec
+        exec_module = spec.loader.exec_module
+
+        def exec_disarmed(module):
+            exec_module(module)
+            module._reconciliation_started = True
+
+        spec.loader.exec_module = exec_disarmed
+        return spec
+
+
+_DISARM_HOOK = _DisarmStartupReconciliation()
+
+
 def pytest_configure(config):
+    sys.meta_path.insert(0, _DISARM_HOOK)
+    app_module = sys.modules.get(_DisarmStartupReconciliation._MODULE)
+    if app_module is not None:
+        app_module._reconciliation_started = True
+
+    _point_emulator_at_raw_adapter(config)
+
+
+def _point_emulator_at_raw_adapter(config):
     """Point the emulator at a per-process config that binds no socket.
 
     Six test modules set EMULATOR=true and build a real DisplayManager. The
@@ -63,7 +110,9 @@ def pytest_configure(config):
 
 
 def pytest_unconfigure(config):
-    """Remove the throwaway emulator config written by pytest_configure."""
+    """Undo pytest_configure: the import hook and the throwaway emulator config."""
+    if _DISARM_HOOK in sys.meta_path:
+        sys.meta_path.remove(_DISARM_HOOK)
     tmp_dir = getattr(config, "_ledmatrix_emulator_tmp", None)
     if tmp_dir is not None:
         import shutil

test/test_api_v3_on_demand_restart.py

@@ -1,25 +1,27 @@
-"""Regression test: POST /display/on-demand/start restarting a running
+"""POST /display/on-demand/start and /stop must not restart a running display.
-service must not import a name that does not exist.
 
-display.py has `import web_interface.blueprints.api_v3 as _pkg` and reads
+The start route used to treat ``start_service`` (default True, and what both
-mutable, test-patched attributes back through it (`_pkg.time.time()`,
+the web UI and the MQTT bridge send) as "restart": with the service running it
-`_pkg._get_starlark_plugin()`, ...) rather than binding them by value, per
+ran ``systemctl stop``, slept 1.5s and started it again. Every on-demand or
-the package's own docstring. One spot went further and wrote a genuine
+"Preview on display" click therefore cold-restarted the display process --
-`import` *statement* against that alias --
+every plugin reloaded, the panel blank for seconds -- to deliver a request the
+running process polls for every ON_DEMAND_POLL_INTERVAL anyway (see
+test_on_demand_mailbox.py and test_display_pending_changes.py for the display
+side: the mailbox is read mid-dwell, mid-screen and mid-Vegas-iteration).
 
-    import _pkg.time as time_module
+The restart did not buy anything either: a freshly started display restores
+only the on-demand session it saved itself (``display_on_demand_config``), so
+the new request reached it through the same mailbox, one cold start later.
 
--- but `_pkg` is a local name bound by `import ... as _pkg` in this module,
+This file previously pinned that restart path (it guarded a broken
-not a real top-level package, so `import _pkg.time` is not something Python
+``import _pkg.time`` inside it). The path is gone; these tests pin its
-can resolve; it raises ModuleNotFoundError. That line only runs when the
+replacement: a running service is left alone, a stopped one is started (only
-display service is already running and the caller also asked to (re)start
+when start_service is set), and the request lands in the mailbox either way.
-it, so this endpoint failed on exactly the restart path -- the one where a
-cache write recording the new on-demand request had already happened.
 
-The route wraps its body in `except Exception`, so the failure reached the
+The service helpers are patched where they run. display.py binds
-caller as a handled 500 with a generic message, not an unhandled crash --
+_get_display_service_status by value, while _ensure_display_service_running
-but a 500 all the same on a request that should have restarted the service
+(in the package __init__) looks it up in its own module, so both are patched;
-and reported success.
+_run_systemctl_command is the one place a systemctl command is issued.
 """
 
 import sys
@@ -32,60 +34,137 @@ sys.path.insert(0, str(Path(__file__).parent.parent))
 
 from test._api_v3_test_helpers import api_v3_client, api_v3_module  # noqa: F401,E402
 
-URL = "/api/v3/display/on-demand/start"
+START_URL = "/api/v3/display/on-demand/start"
+STOP_URL = "/api/v3/display/on-demand/stop"
+MAILBOX = "display_on_demand_request"
 
 
 @pytest.fixture
-def restart_path(api_v3_module):
+def service(api_v3_module):
-    """Force the `service_was_running and start_service` branch.
+    """A display service whose state the test sets; records systemctl calls.
 
-    plugin_manager and config_manager are set to None so the route takes
+    plugin_manager and config_manager are None so the route skips plugin
-    the simplest path to that branch rather than tripping over unrelated
+    resolution (not what is under test here). The cache is the blueprint's
-    MagicMock plumbing. The cache is the blueprint's cache_manager, which
+    MagicMock cache_manager, so mailbox writes are visible as set() calls.
-    api_v3_module already set to a MagicMock. _get_display_service_status,
-    _stop_display_service and _ensure_display_service_running are bound by
-    value in display.py (see its own docstring), so they are patched on
-    that submodule rather than on the package.
     """
     api_v3_module.api_v3.plugin_manager = None
     api_v3_module.api_v3.config_manager = None
+    state = {"active": True}
 
-    with patch("web_interface.blueprints.api_v3.display._get_display_service_status") as get_status, \
+    def status():
-         patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop_service, \
+        return {"active": state["active"]}
-         patch("web_interface.blueprints.api_v3.display._ensure_display_service_running") as ensure_running:
+
-        # Active before the request: service_was_running becomes True.
+    def systemctl(args):
-        get_status.return_value = {"active": True}
+        if args[-2:] == ["start", "ledmatrix.service"]:
-        ensure_running.return_value = {"active": True}
+            state["active"] = True
+        elif args[-2:] == ["stop", "ledmatrix.service"]:
+            state["active"] = False
+        return {"returncode": 0, "stdout": "", "stderr": ""}
+
+    with patch("web_interface.blueprints.api_v3._get_display_service_status",
+               side_effect=status), \
+         patch("web_interface.blueprints.api_v3.display._get_display_service_status",
+               side_effect=status), \
+         patch("web_interface.blueprints.api_v3._run_systemctl_command",
+               side_effect=systemctl) as run_systemctl, \
+         patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop_service:
         yield {
-            "get_status": get_status,
+            "state": state,
+            "systemctl": run_systemctl,
             "stop_service": stop_service,
-            "ensure_running": ensure_running,
+            "cache": api_v3_module.api_v3.cache_manager,
         }
 
 
-class TestRestartingARunningService:
+def _mailbox_writes(cache):
-    def test_it_does_not_500(self, api_v3_client, restart_path):
+    return [c.args[1] for c in cache.set.call_args_list if c.args and c.args[0] == MAILBOX]
-        response = api_v3_client.post(
-            URL, json={"plugin_id": "weather", "start_service": True})
-        body = response.get_json()
-        assert response.status_code == 200, body
-        assert body["status"] == "success", body
 
-    def test_the_service_is_actually_stopped_and_restarted(
-            self, api_v3_client, restart_path):
-        api_v3_client.post(
-            URL, json={"plugin_id": "weather", "start_service": True})
-        restart_path["stop_service"].assert_called_once()
-        restart_path["ensure_running"].assert_called_once()
 
-    def test_a_service_that_was_not_running_is_not_stopped_first(
+def _systemctl_verbs(run_systemctl):
-            self, api_v3_client, restart_path):
+    return [c.args[0][-2] for c in run_systemctl.call_args_list]
-        # The buggy import sits inside `if service_was_running and
+
-        # start_service`, so it only ever fired on the restart path --
+
-        # this is the other side of that branch, unaffected either way,
+class TestStartWhileTheServiceIsRunning:
-        # kept here so the branch condition itself stays covered.
+    @pytest.mark.parametrize("body", [
-        restart_path["get_status"].return_value = {"active": False}
+        {"plugin_id": "weather"},                           # "Preview on display", MQTT
-        response = api_v3_client.post(
+        {"plugin_id": "weather", "start_service": True},    # on-demand modal, box ticked
-            URL, json={"plugin_id": "weather", "start_service": True})
+        {"plugin_id": "weather", "start_service": "true"},
+    ])
+    def test_the_service_is_not_stopped_or_restarted(self, api_v3_client, service, body):
+        response = api_v3_client.post(START_URL, json=body)
         assert response.status_code == 200, response.get_json()
-        restart_path["stop_service"].assert_not_called()
+        assert response.get_json()["status"] == "success"
+        service["stop_service"].assert_not_called()
+        assert _systemctl_verbs(service["systemctl"]) == [], (
+            "a running display service was sent a systemctl command")
+
+    def test_the_request_is_posted_for_the_running_display(self, api_v3_client, service):
+        response = api_v3_client.post(
+            START_URL, json={"plugin_id": "weather", "mode": "weather_current",
+                             "duration": 60, "pinned": True})
+        data = response.get_json()["data"]
+        writes = _mailbox_writes(service["cache"])
+        assert len(writes) == 1
+        assert writes[0]["action"] == "start"
+        assert writes[0]["request_id"] == data["request_id"]
+        assert writes[0]["plugin_id"] == "weather"
+        assert writes[0]["mode"] == "weather_current"
+        assert writes[0]["duration"] == 60
+        assert writes[0]["pinned"] is True
+
+    def test_the_response_reports_the_service_was_not_started(self, api_v3_client, service):
+        data = api_v3_client.post(START_URL, json={"plugin_id": "weather"}).get_json()["data"]
+        assert data["service"]["active"] is True
+        assert data["service"]["started"] is False
+
+    def test_it_answers_without_the_old_restart_pause(self, api_v3_client, service):
+        # The restart slept 1.5s; nothing here should sleep at all.
+        with patch("time.sleep") as sleep:
+            api_v3_client.post(START_URL, json={"plugin_id": "weather"})
+        sleep.assert_not_called()
+
+
+class TestStartWhileTheServiceIsStopped:
+    def test_start_service_starts_it_once_and_never_stops_it(self, api_v3_client, service):
+        service["state"]["active"] = False
+        response = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
+        assert response.status_code == 200, response.get_json()
+        assert _systemctl_verbs(service["systemctl"]) == ["start"]
+        service["stop_service"].assert_not_called()
+        # Written before the start, so the new process finds it on its first poll.
+        assert len(_mailbox_writes(service["cache"])) == 1
+
+    def test_without_start_service_it_is_left_stopped(self, api_v3_client, service):
+        service["state"]["active"] = False
+        response = api_v3_client.post(
+            START_URL, json={"plugin_id": "weather", "start_service": "false"})
+        assert response.status_code == 400
+        assert _systemctl_verbs(service["systemctl"]) == []
+
+    def test_a_start_that_fails_is_reported(self, api_v3_client, service):
+        service["state"]["active"] = False
+        service["systemctl"].side_effect = lambda args: {
+            "returncode": 1, "stdout": "", "stderr": "denied"}
+        response = api_v3_client.post(START_URL, json={"plugin_id": "weather"})
+        assert response.status_code == 500
+        assert response.get_json()["status"] == "error"
+
+
+class TestStop:
+    def test_stop_posts_a_stop_request_and_leaves_the_service_running(
+            self, api_v3_client, service):
+        response = api_v3_client.post(STOP_URL, json={})
+        assert response.status_code == 200, response.get_json()
+        writes = _mailbox_writes(service["cache"])
+        assert [w["action"] for w in writes] == ["stop"]
+        service["stop_service"].assert_not_called()
+        assert _systemctl_verbs(service["systemctl"]) == []
+
+    def test_a_string_false_stop_service_does_not_stop_it(self, api_v3_client, service):
+        # bool("false") is True: the flag was read raw and stopped the service.
+        api_v3_client.post(STOP_URL, json={"stop_service": "false"})
+        service["stop_service"].assert_not_called()
+
+    def test_stop_service_true_still_stops_it(self, api_v3_client, service):
+        api_v3_client.post(STOP_URL, json={"stop_service": True})
+        service["stop_service"].assert_called_once()

test/test_on_demand_disabled_plugin.py

@@ -0,0 +1,426 @@
+"""On-demand for a plugin that is installed but disabled in config.
+
+The display process only loads enabled plugins, so a request for a disabled
+one -- "Preview on display" offers it on every plugin's config page, with a
+note that the plugin will be enabled for the preview -- failed with
+"invalid-mode". Nothing loaded it short of a restart, and the on-demand
+route no longer restarts the service.
+
+The display now loads such a plugin live for the session (force_enabled, so
+config.json keeps saying disabled) and the main loop unloads it once
+on-demand moves off it: a stop, an expiry, or a request for another plugin.
+
+Also here: a stop sent after a failed request clears the error instead of
+leaving status 'error' published until the state ages out.
+"""
+
+import time
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.plugin_manager import PluginManager
+from src.plugin_system.plugin_state import PluginState
+
+
+def _make_plugin(modes):
+    plugin = MagicMock()
+    plugin.modes = list(modes)
+    return plugin
+
+
+@pytest.fixture
+def controller(test_display_controller):
+    """An idle controller running 'clock', with 'preview-me' installed but disabled."""
+    c = test_display_controller
+    clock = _make_plugin(['clock'])
+    preview = _make_plugin(['preview_a', 'preview_b'])
+    instances = {'clock': clock}
+    catalogue = {'clock': clock, 'preview-me': preview}
+
+    def load_plugin(plugin_id, force_enabled=False):
+        instances[plugin_id] = catalogue[plugin_id]
+        return True
+
+    def unload_plugin(plugin_id):
+        return instances.pop(plugin_id, None) is not None
+
+    pm = c.plugin_manager
+    pm.discovered_plugin_ids.return_value = set(catalogue)
+    pm.discover_plugins.return_value = list(catalogue)
+    pm.plugin_manifests = {}
+    pm.load_plugin = MagicMock(side_effect=load_plugin)
+    pm.unload_plugin = MagicMock(side_effect=unload_plugin)
+    pm.get_plugin.side_effect = instances.get
+
+    config = {'clock': {'enabled': True}, 'preview-me': {'enabled': False}}
+    c.config_service.get_config = lambda: config
+    c.config_manager.save_config = MagicMock()
+    c.cache_manager.set = MagicMock()
+    c.cache_manager.clear_cache = MagicMock()
+
+    c._register_loaded_plugin('clock')
+    c.current_mode_index = 0
+    c.current_display_mode = 'clock'
+    c.test_config = config
+    c.test_instances = instances
+    return c
+
+
+def _start(c, plugin_id='preview-me', mode=None, **extra):
+    request = {'request_id': 'r-' + plugin_id, 'action': 'start',
+               'plugin_id': plugin_id, 'mode': mode or plugin_id}
+    request.update(extra)
+    c._activate_on_demand(request)
+
+
+class TestLoadingForOnDemand:
+    def test_a_disabled_plugin_is_loaded_and_shown(self, controller):
+        _start(controller)
+
+        controller.plugin_manager.load_plugin.assert_called_once_with(
+            'preview-me', force_enabled=True)
+        assert controller.on_demand_active is True
+        assert controller.on_demand_status == 'active'
+        assert controller.on_demand_plugin_id == 'preview-me'
+        assert controller.current_display_mode == 'preview_a'
+        assert controller.plugin_display_modes['preview-me'] == ['preview_a', 'preview_b']
+
+    def test_config_json_is_not_written(self, controller):
+        _start(controller)
+
+        controller.config_manager.save_config.assert_not_called()
+        assert controller.test_config['preview-me'] == {'enabled': False}
+
+    def test_a_requested_mode_is_honoured(self, controller):
+        _start(controller, mode='preview_b')
+        assert controller.current_display_mode == 'preview_b'
+
+    def test_an_enabled_plugin_is_not_reloaded(self, controller):
+        _start(controller, plugin_id='clock')
+
+        controller.plugin_manager.load_plugin.assert_not_called()
+        assert controller.on_demand_active is True
+        assert controller._on_demand_loaded_plugins == set()
+
+    def test_a_plugin_that_is_not_installed_is_not_loaded(self, controller):
+        _start(controller, plugin_id='uninstalled')
+
+        controller.plugin_manager.load_plugin.assert_not_called()
+        assert controller.on_demand_status == 'error'
+        assert controller.on_demand_last_error == 'invalid-mode'
+
+    def test_a_plugin_installed_after_startup_is_found_by_rescanning(self, controller):
+        controller.plugin_manager.discovered_plugin_ids.return_value = {'clock'}
+
+        _start(controller)
+
+        controller.plugin_manager.discover_plugins.assert_called()
+        assert controller.on_demand_active is True
+
+
+class TestLoadFailures:
+    def test_a_failed_load_reports_load_failed(self, controller):
+        controller.plugin_manager.load_plugin = MagicMock(return_value=False)
+
+        _start(controller)
+
+        assert controller.on_demand_active is False
+        assert controller.on_demand_status == 'error'
+        assert controller.on_demand_last_error == 'load-failed'
+        assert 'preview_a' not in controller.available_modes
+        published = controller.cache_manager.set.call_args_list[-1]
+        assert published.args[0] == 'display_on_demand_state'
+        assert published.args[1]['status'] == 'error'
+        assert published.args[1]['error'] == 'load-failed'
+
+    def test_a_load_that_raises_reports_load_failed(self, controller):
+        controller.plugin_manager.load_plugin = MagicMock(side_effect=ImportError('no module'))
+
+        _start(controller)
+
+        assert controller.on_demand_status == 'error'
+        assert controller.on_demand_last_error == 'load-failed'
+
+    def test_a_failed_load_leaves_the_rotation_alone(self, controller):
+        controller.plugin_manager.load_plugin = MagicMock(return_value=False)
+
+        _start(controller)
+        controller._release_on_demand_plugins()
+
+        assert controller.available_modes == ['clock']
+        assert controller.current_display_mode == 'clock'
+        assert controller._on_demand_loaded_plugins == set()
+        controller.plugin_manager.unload_plugin.assert_not_called()
+
+    def test_a_plugin_that_loads_but_has_no_modes_is_unloaded_again(self, controller):
+        """Registered, then the activation fails: the release removes it."""
+        controller._on_demand_modes_for_plugin = MagicMock(return_value=[])
+
+        _start(controller)
+        assert controller.on_demand_last_error == 'no-modes'
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+        assert controller.available_modes == ['clock']
+
+
+class TestReleasingThePlugin:
+    def test_it_stays_loaded_while_on_demand_shows_it(self, controller):
+        _start(controller)
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_not_called()
+        assert 'preview_a' in controller.plugin_modes
+
+    def test_a_stop_unloads_it_and_resumes_the_rotation(self, controller):
+        _start(controller)
+        controller._clear_on_demand(reason='requested-stop')
+        # Deferred to the main loop: the stop may be read mid-display().
+        controller.plugin_manager.unload_plugin.assert_not_called()
+
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+        assert controller.available_modes == ['clock']
+        assert 'preview-me' not in controller.plugin_display_modes
+        assert 'preview_a' not in controller.plugin_modes
+        assert controller.current_display_mode == 'clock'
+        assert controller._on_demand_loaded_plugins == set()
+        assert controller.test_config['preview-me'] == {'enabled': False}
+
+    def test_expiry_unloads_it(self, controller):
+        _start(controller, duration=30)
+        controller.on_demand_expires_at = time.time() - 1
+        controller._check_on_demand_expiration()
+        controller._release_on_demand_plugins()
+
+        assert controller.on_demand_last_event == 'expired'
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+
+    def test_a_request_for_another_plugin_unloads_it(self, controller):
+        _start(controller)
+        _start(controller, plugin_id='clock')
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+        assert controller.on_demand_active is True
+        assert controller.on_demand_plugin_id == 'clock'
+        assert controller.current_display_mode == 'clock'
+
+    def test_a_failed_request_that_ends_the_session_unloads_it(self, controller):
+        _start(controller)
+        _start(controller, plugin_id='uninstalled')
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+        assert controller.current_display_mode == 'clock'
+        assert controller.force_change is True
+
+    def test_a_plugin_enabled_during_the_session_stays_loaded(self, controller):
+        _start(controller)
+        controller.test_config['preview-me'] = {'enabled': True}
+        controller._clear_on_demand(reason='requested-stop')
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_not_called()
+        assert 'preview_a' in controller.available_modes
+        assert controller._on_demand_loaded_plugins == set()
+
+    def test_the_main_loop_releases_right_after_its_own_poll(self, controller):
+        """A stop read by the main loop unloads before the next screen, not
+        one screen later. That poll runs with no display() on the stack."""
+        import inspect
+        source = inspect.getsource(type(controller).run)
+        poll = source.index('self._check_on_demand_expiration()')
+        release = source.index('self._release_on_demand_plugins()')
+        render = source.index('self._tick_plugin_updates()')
+        assert poll < release < render
+
+    def test_a_reconcile_that_runs_first_unloads_it_the_same_way(self, controller):
+        """A reconcile queued during the session runs at the top of the loop,
+        before the release: it removes the plugin itself (not in the enabled
+        set) and the release is then a no-op."""
+        _start(controller)
+        controller._clear_on_demand(reason='requested-stop')
+
+        controller._reconcile_enabled_plugins()
+        controller._release_on_demand_plugins()
+
+        controller.plugin_manager.unload_plugin.assert_called_once_with('preview-me')
+        assert controller.available_modes == ['clock']
+        assert controller._on_demand_loaded_plugins == set()
+
+    def test_a_config_save_mid_session_keeps_the_instance_enabled(self, controller):
+        """on_config_change would otherwise read enabled: false and switch it off."""
+        controller.config_service.subscribe = MagicMock()
+        _start(controller)
+        callback = controller._plugin_config_callbacks['preview-me']
+        controller.plugin_manager.prepare_plugin_config = None
+
+        callback({}, {'enabled': False, 'color': 'red'})
+
+        # The change goes through the manager's locked apply_config_change
+        # (which calls on_config_change under the plugin's lock).
+        plugin = controller.plugin_modes['preview_a']
+        controller.plugin_manager.apply_config_change.assert_called_once_with(
+            'preview-me', {'enabled': True, 'color': 'red'}, plugin_instance=plugin)
+
+
+class TestRestoredSession:
+    """A restart during a session for a disabled plugin restores it the same way."""
+
+    def test_the_plugin_is_tracked_and_config_is_left_alone(self, test_display_controller):
+        c = test_display_controller
+        c.config.update({'clock': {'enabled': True}, 'disabled-one': {'enabled': False}})
+
+        selected = c._select_startup_plugins(
+            ['clock', 'disabled-one'], {'plugin_id': 'disabled-one', 'mode': 'x'})
+
+        assert 'disabled-one' in selected
+        assert c._on_demand_loaded_plugins == {'disabled-one'}
+        assert c.config['disabled-one']['enabled'] is False
+
+
+class TestResumingAfterTheSession:
+    """Ending a session never resumes the rotation onto the plugin that is
+    about to be unloaded."""
+
+    def _restored_session(self, c, other_modes=('clock',)):
+        """As after a restart: no saved resume index, and the plugin's modes
+        ordered in ahead of the rest (load order is not deterministic)."""
+        c._on_demand_loaded_plugins.add('preview-me')
+        c.plugin_manager.load_plugin('preview-me', force_enabled=True)
+        c._register_loaded_plugin('preview-me')
+        c.available_modes = ['preview_a', 'preview_b'] + list(other_modes)
+        c.on_demand_active = True
+        c.on_demand_status = 'active'
+        c.on_demand_plugin_id = 'preview-me'
+        c.on_demand_modes = ['preview_a', 'preview_b']
+        c.rotation_resume_index = None
+        c.current_mode_index = 0
+        c.current_display_mode = 'preview_a'
+
+    def test_a_restored_session_resumes_on_an_enabled_mode(self, controller):
+        self._restored_session(controller)
+
+        controller._clear_on_demand(reason='requested-stop')
+
+        assert controller.current_display_mode == 'clock'
+        controller._release_on_demand_plugins()
+        assert controller.available_modes == ['clock']
+        assert controller.current_display_mode == 'clock'
+
+    def test_with_nothing_else_enabled_the_display_goes_idle(self, controller):
+        controller._unregister_plugin('clock')
+        self._restored_session(controller, other_modes=())
+
+        controller._clear_on_demand(reason='requested-stop')
+        assert controller.current_display_mode is None
+
+        controller._release_on_demand_plugins()
+        assert controller.available_modes == []
+        assert controller.current_display_mode is None
+
+    def test_a_saved_resume_index_is_still_used(self, controller):
+        c = controller
+        c.available_modes = ['clock', 'other']
+        c.plugin_modes['other'] = MagicMock()
+        c.current_mode_index = 1
+        c.current_display_mode = 'other'
+
+        _start(c)
+        c._clear_on_demand(reason='requested-stop')
+
+        assert c.current_display_mode == 'other'
+
+
+class TestStopClearsAnError:
+    def _post_stop(self, c):
+        stop = {'request_id': 'S1', 'action': 'stop'}
+        c._last_on_demand_poll = None
+        c.cache_manager.get = MagicMock(
+            side_effect=lambda key, *a, **kw:
+                stop if key == 'display_on_demand_request' else None)
+        c.cache_manager.delete = MagicMock()
+        c._poll_on_demand_requests()
+
+    def test_a_stop_after_a_failed_request_clears_the_error(self, controller):
+        _start(controller, plugin_id='uninstalled')
+        assert controller.on_demand_status == 'error'
+
+        self._post_stop(controller)
+
+        assert controller.on_demand_status == 'idle'
+        assert controller.on_demand_last_error is None
+        state = controller.cache_manager.set.call_args_list[-1].args[1]
+        assert state['status'] == 'idle'
+        assert state['error'] is None
+
+    def test_clearing_the_error_leaves_the_rotation_alone(self, controller):
+        _start(controller, plugin_id='uninstalled')
+        controller.force_change = False
+
+        self._post_stop(controller)
+
+        assert controller.current_display_mode == 'clock'
+        assert controller.force_change is False
+
+    def test_a_stop_while_idle_is_still_just_acknowledged(self, controller):
+        controller._clear_on_demand = MagicMock()
+
+        self._post_stop(controller)
+
+        assert controller.on_demand_status == 'idle'
+        assert controller.on_demand_request_id == 'S1'
+        controller._clear_on_demand.assert_not_called()
+
+
+class TestForceEnabledLoad:
+    """PluginManager.load_plugin(force_enabled=True) runs the plugin enabled
+    without touching the config it read."""
+
+    class _Plugin:
+        def __init__(self, config):
+            self.config = config
+            self.enabled_calls = 0
+
+        def on_enable(self):
+            self.enabled_calls += 1
+
+    @pytest.fixture
+    def pm(self, tmp_path):
+        plugins_dir = tmp_path / 'plugins'
+        (plugins_dir / 'demo').mkdir(parents=True)
+        manager = PluginManager(plugins_dir=str(plugins_dir))
+        manager.plugin_manifests['demo'] = {'id': 'demo', 'name': 'Demo'}
+        manager.schema_manager = MagicMock()
+        manager.schema_manager.get_schema_path.return_value = None
+        # Hand the section back as-is, as the fallback path can: the copy in
+        # load_plugin is what keeps the cached config clean.
+        manager.schema_manager.prepare_plugin_config.side_effect = (
+            lambda pid, cfg, schema=None, changed_paths=None: cfg)
+        manager.plugin_loader = MagicMock()
+        manager.plugin_loader.find_plugin_directory.return_value = plugins_dir / 'demo'
+        manager.plugin_loader.load_plugin.side_effect = (
+            lambda **kw: (self._Plugin(kw['config']), None))
+        manager.config_manager = MagicMock()
+        manager.cached_config = {'demo': {'enabled': False, 'color': 'red'}}
+        manager.config_manager.load_config.return_value = manager.cached_config
+        return manager
+
+    def test_a_disabled_plugin_loads_disabled_by_default(self, pm):
+        assert pm.load_plugin('demo') is True
+        assert pm.plugins['demo'].enabled_calls == 0
+        assert pm.state_manager.get_state('demo') == PluginState.DISABLED
+
+    def test_force_enabled_runs_it_enabled(self, pm):
+        assert pm.load_plugin('demo', force_enabled=True) is True
+        plugin = pm.plugins['demo']
+        assert plugin.config == {'enabled': True, 'color': 'red'}
+        assert plugin.enabled_calls == 1
+        assert pm.state_manager.get_state('demo') == PluginState.ENABLED
+
+    def test_force_enabled_does_not_touch_the_cached_config(self, pm):
+        pm.load_plugin('demo', force_enabled=True)
+        assert pm.cached_config['demo'] == {'enabled': False, 'color': 'red'}

test/test_on_demand_pinning_and_restart.py

@@ -164,12 +164,14 @@ class TestRestartDoesNotStarveTheOtherPlugins:
         assert controller.on_demand_mode == 'app_a'
         assert controller.on_demand_pinned is True
 
-    def test_a_disabled_on_demand_plugin_is_enabled_and_loaded(self, controller):
+    def test_a_disabled_on_demand_plugin_is_still_loaded(self, controller):
-        """Otherwise the mode being resumed has nothing behind it."""
+        """Otherwise the mode being resumed has nothing behind it. It loads
+        for on-demand only; its config section is left disabled."""
         selected = controller._select_startup_plugins(
             self.DISCOVERED, {'plugin_id': 'disabled-one', 'mode': 'x'})
         assert 'disabled-one' in selected
-        assert controller.config['disabled-one']['enabled'] is True
+        assert controller._on_demand_loaded_plugins == {'disabled-one'}
+        assert controller.config['disabled-one']['enabled'] is False
 
     def test_an_unknown_on_demand_plugin_falls_back_to_normal(self, controller):
         selected = controller._select_startup_plugins(

test/test_plugin_config_preparation.py

@@ -84,6 +84,19 @@ def plugin_manager(plugin_dir, tmp_path):
     return manager
 
 
+@pytest.fixture
+def delivering_manager(tmp_path):
+    """A real PluginManager whose apply_config_change() the mocked one uses.
+
+    The hot-reload callback hands on_config_change to the plugin manager, so
+    it runs under the plugin's lock; delegate to the real method so these
+    tests still see the plugin called.
+    """
+    manager = PluginManager(plugins_dir=str(tmp_path / "plugin-repos"))
+    yield manager
+    manager.stop_update_worker()
+
+
 class TestPluginManagerPreparation:
     def test_legacy_boolean_and_defaults(self, plugin_manager):
         prepared = plugin_manager.prepare_plugin_config(
@@ -107,11 +120,13 @@ class TestPluginManagerPreparation:
 
 
 class TestHotReload:
-    def test_on_config_change_gets_the_prepared_config(self, test_display_controller, plugin_manager):
+    def test_on_config_change_gets_the_prepared_config(self, test_display_controller, plugin_manager,
+                                                       delivering_manager):
         controller = test_display_controller
         plugin = MagicMock()
         plugin.modes = ["demo"]
         pm = controller.plugin_manager
+        pm.apply_config_change.side_effect = delivering_manager.apply_config_change
         pm.discover_plugins.return_value = ["demo"]
         pm.load_plugin.return_value = True
         pm.plugin_manifests = {}
@@ -128,11 +143,13 @@ class TestHotReload:
             "enabled": True, "max_duration_seconds": 300}
         assert new_config["nhl"]["show_records"] is True
 
-    def test_raw_section_still_delivered_without_a_preparer(self, test_display_controller):
+    def test_raw_section_still_delivered_without_a_preparer(self, test_display_controller,
+                                                            delivering_manager):
         controller = test_display_controller
         plugin = MagicMock()
         plugin.modes = ["demo"]
         pm = controller.plugin_manager
+        pm.apply_config_change.side_effect = delivering_manager.apply_config_change
         pm.discover_plugins.return_value = ["demo"]
         pm.load_plugin.return_value = True
         pm.plugin_manifests = {}

test/test_plugin_hang_containment.py

@@ -0,0 +1,524 @@
+"""One hung plugin must not stop every other plugin from updating.
+
+The update worker is a single thread. It took each plugin's lock with a
+blocking ``acquire()``, and the render thread holds that same lock while it
+runs the plugin's display(). A display() that never returned -- or a first
+frame that outlived PluginExecutor's timeout and kept running on its lingering
+thread -- parked the worker on that acquire for good, and from then on no
+plugin updated at all: scores, weather and clocks all froze while the panel
+kept scrolling stale data.
+
+Now:
+- the worker waits PLUGIN_LOCK_TIMEOUT at most and skips the busy plugin. The
+  skip is report-only (a busy skip in health): the holder may be a healthy
+  Vegas prefetch render, so it never counts toward the circuit breaker;
+- display() calls are timed, slow ones recorded, overlong ones counted as
+  hangs; update() past the executor timeout is a hang too. Only hangs open
+  the breaker;
+- on_config_change runs under the plugin's lock, deferred to the worker if the
+  lock stays busy, so it never interleaves with update().
+
+Timeouts here are fractions of a second so the suite stays fast.
+"""
+
+import copy
+import threading
+import time
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.plugin_health import CircuitState, PluginHealthTracker
+from src.plugin_system.plugin_manager import PluginManager
+from src.plugin_system.plugin_state import PluginState
+
+
+class _Cache:
+    """Serialising stand-in for CacheManager; counts writes."""
+
+    def __init__(self):
+        self.store = {}
+        self.writes = 0
+
+    def set(self, key, data, ttl=None, **kwargs):
+        self.writes += 1
+        self.store[key] = copy.deepcopy(data)
+
+    def get(self, key, max_age=None, memory_ttl=None, **kwargs):
+        return copy.deepcopy(self.store.get(key))
+
+
+class _Plugin:
+    """Counts update() calls; display() can be made to block on an Event."""
+
+    def __init__(self, plugin_id, update_seconds=0.0):
+        self.plugin_id = plugin_id
+        self.enabled = True
+        self.update_seconds = update_seconds
+        self.update_calls = 0
+        self.in_update = False
+        self.display_gate = None  # threading.Event: display() blocks on it
+        self.config_changes = []
+        self.overlap = False
+        self.events = []
+
+    def update(self):
+        self.in_update = True
+        self.update_calls += 1
+        self.events.append('update')
+        time.sleep(self.update_seconds)
+        self.in_update = False
+
+    def display(self, force_clear=False):
+        if self.display_gate is not None:
+            self.display_gate.wait(timeout=10)
+        return True
+
+    def on_config_change(self, new_config):
+        if self.in_update:
+            self.overlap = True
+        self.events.append('config')
+        self.config_changes.append(new_config)
+
+
+@pytest.fixture
+def pm(tmp_path):
+    manager = PluginManager(plugins_dir=str(tmp_path), config_manager=None,
+                            display_manager=None, cache_manager=None)
+    manager.PLUGIN_LOCK_TIMEOUT = 0.15
+    yield manager
+    manager.stop_update_worker()
+
+
+@pytest.fixture
+def tracker(pm):
+    t = PluginHealthTracker(cache_manager=_Cache())
+    pm.health_tracker = t
+    return t
+
+
+def _install(pm, plugin, interval=0.01):
+    pm.plugins[plugin.plugin_id] = plugin
+    pm._update_interval_cache[plugin.plugin_id] = interval
+    pm.state_manager.set_state(plugin.plugin_id, PluginState.ENABLED)
+    return plugin.plugin_id
+
+
+def _hang_display(pm, plugin):
+    """Run the plugin's display() the way the render loop does, on its own
+    thread, holding the plugin's lock while display() blocks on its gate."""
+    plugin.display_gate = threading.Event()
+    entered = threading.Event()
+
+    def render():
+        lock = pm.get_plugin_lock(plugin.plugin_id)
+        with lock:
+            entered.set()
+            plugin.display()
+
+    thread = threading.Thread(target=render, daemon=True)
+    thread.start()
+    assert entered.wait(timeout=2)
+    return plugin.display_gate, thread
+
+
+def _wait_for(predicate, timeout=3.0):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if predicate():
+            return True
+        time.sleep(0.01)
+    return predicate()
+
+
+class TestHungDisplayDoesNotStallTheWorker:
+    def test_other_plugins_keep_updating_on_schedule(self, pm):
+        hung = _Plugin('hung')
+        healthy = _Plugin('healthy')
+        _install(pm, hung)
+        _install(pm, healthy)
+        gate, render_thread = _hang_display(pm, hung)
+        try:
+            # 'hung' is dict-ordered first, so its item is dequeued ahead of
+            # 'healthy' every round. With a blocking acquire the worker parks
+            # on it forever and 'healthy' never updates.
+            for _ in range(8):
+                pm.run_scheduled_updates()
+                time.sleep(0.05)
+            assert _wait_for(lambda: healthy.update_calls >= 2), (
+                f"healthy plugin updated {healthy.update_calls} time(s) while "
+                "another plugin's display() was hung")
+            assert hung.update_calls == 0
+            assert pm._update_worker.is_alive()
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+
+    def test_hung_plugin_is_retried_once_its_display_returns(self, pm):
+        plugin = _Plugin('hung')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        pm.run_scheduled_updates()
+        # Skipped, and handed back rather than left RUNNING.
+        assert _wait_for(lambda: pm.state_manager.can_execute('hung'))
+        assert 'hung' not in pm._pending_updates
+        gate.set()
+        render_thread.join(timeout=2)
+        pm.plugin_last_update.pop('hung', None)  # due again now
+        pm.run_scheduled_updates()
+        assert _wait_for(lambda: plugin.update_calls == 1)
+
+
+class TestLockTimeoutIsReportOnly:
+    def test_skip_lands_in_health_and_state_as_a_busy_skip(self, pm, tracker):
+        plugin = _Plugin('hung')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        try:
+            pm.run_scheduled_updates()
+            assert _wait_for(lambda: tracker.get_health_summary('hung')['busy_skip_count'] == 1)
+            summary = tracker.get_health_summary('hung')
+            assert summary['last_busy_skip']['operation'] == 'update lock wait'
+            assert summary['last_busy_skip']['seconds'] >= pm.PLUGIN_LOCK_TIMEOUT - 0.01
+            # Reporting only: no failure, no hang, no error, breaker closed.
+            assert summary['consecutive_failures'] == 0
+            assert summary['total_failures'] == 0
+            assert summary['hang_count'] == 0
+            assert summary['last_error'] is None
+            assert summary['circuit_state'] == CircuitState.CLOSED.value
+            # Still visible in the plugin's state for the web UI.
+            error_info = pm.state_manager.get_error_info('hung')
+            assert error_info['error_type'] == 'PluginBusyError'
+            # Stamped like any failed update, so the retry waits an interval.
+            assert pm.plugin_last_update.get('hung', 0) > 0
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+
+    def test_repeated_busy_skips_never_open_the_circuit_breaker(self, pm, tracker):
+        """A Vegas prefetch render can hold a healthy plugin's lock past the
+        bound on every update; that must not pull it from rotation."""
+        plugin = _Plugin('busy')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        skips = tracker.failure_threshold * 3
+        try:
+            for attempt in range(1, skips + 1):
+                pm.run_scheduled_updates()
+                assert _wait_for(
+                    lambda: pm.state_manager.can_execute('busy')
+                    and tracker.get_health_summary('busy')['busy_skip_count'] == attempt)
+                time.sleep(0.02)  # past the 0.01s interval
+            summary = tracker.get_health_summary('busy')
+            assert summary['busy_skip_count'] == skips
+            assert summary['consecutive_failures'] == 0
+            assert summary['hang_count'] == 0
+            assert summary['circuit_state'] == CircuitState.CLOSED.value
+            assert tracker.should_skip_plugin('busy') is False
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+        # Once the lock frees the plugin updates normally.
+        pm.plugin_last_update.pop('busy', None)
+        pm.run_scheduled_updates()
+        assert _wait_for(lambda: plugin.update_calls == 1)
+
+    def test_busy_skips_do_not_add_to_a_real_hang_streak(self, pm, tracker):
+        plugin = _Plugin('p')
+        _install(pm, plugin)
+        tracker.record_hang('p', 'display', 31.0)
+        gate, render_thread = _hang_display(pm, plugin)
+        try:
+            for attempt in range(1, tracker.failure_threshold + 2):
+                pm.run_scheduled_updates()
+                assert _wait_for(
+                    lambda: pm.state_manager.can_execute('p')
+                    and tracker.get_health_summary('p')['busy_skip_count'] == attempt)
+                time.sleep(0.02)
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+        summary = tracker.get_health_summary('p')
+        assert summary['consecutive_failures'] == 1
+        assert summary['hang_count'] == 1
+        assert summary['circuit_state'] == CircuitState.CLOSED.value
+
+    def test_repeated_real_hangs_still_open_the_circuit_breaker(self, pm, tracker):
+        """display() past the executor timeout is a real hang: the breaker
+        opens after failure_threshold of them, busy skips in between or not."""
+        plugin = _Plugin('hung')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        too_long = pm.plugin_executor.default_timeout + 1
+        try:
+            for attempt in range(1, tracker.failure_threshold + 1):
+                pm.run_scheduled_updates()
+                assert _wait_for(
+                    lambda: pm.state_manager.can_execute('hung')
+                    and tracker.get_health_summary('hung')['busy_skip_count'] == attempt)
+                assert tracker.get_health_summary('hung')['circuit_state'] ==                     CircuitState.CLOSED.value
+                pm.note_display_duration('hung', too_long)
+                time.sleep(0.02)  # past the 0.01s interval
+            summary = tracker.get_health_summary('hung')
+            assert summary['hang_count'] == tracker.failure_threshold
+            assert summary['busy_skip_count'] == tracker.failure_threshold
+            assert summary['circuit_state'] == CircuitState.OPEN.value
+            assert tracker.should_skip_plugin('hung') is True
+            # Circuit open: the scheduler no longer queues it at all.
+            pm.run_scheduled_updates()
+            assert 'hung' not in pm._pending_updates
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+
+    def test_skip_is_logged_once_per_interval(self, pm):
+        pm.logger = MagicMock()
+        plugin = _Plugin('hung')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        try:
+            for _ in range(3):
+                pm.run_scheduled_updates()
+                assert _wait_for(lambda: pm.state_manager.can_execute('hung'))
+                time.sleep(0.02)
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+        busy_warnings = [c for c in pm.logger.warning.call_args_list
+                         if 'update skipped' in c.args[0]]
+        assert len(busy_warnings) == 1
+
+    def test_unloaded_while_waiting_is_not_resurrected(self, pm, tracker):
+        plugin = _Plugin('hung')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        try:
+            pm.PLUGIN_LOCK_TIMEOUT = 0.3
+            pm.UNLOAD_LOCK_TIMEOUT = 0.05
+            pm.run_scheduled_updates()
+            time.sleep(0.05)  # worker is now waiting on the lock
+            assert pm.unload_plugin('hung') is True
+            assert _wait_for(lambda: 'hung' not in pm._pending_updates)
+            time.sleep(0.35)
+            assert pm.state_manager.get_state('hung') == PluginState.UNLOADED
+            summary = tracker.get_health_summary('hung')
+            assert summary['hang_count'] == 0
+            assert summary['busy_skip_count'] == 0
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+
+
+class TestUpdateOutlivingTheExecutorTimeout:
+    def test_recorded_as_a_hang_then_cleared_when_it_returns(self, pm, tracker):
+        plugin = _Plugin('slow', update_seconds=0.4)
+        _install(pm, plugin)
+        pm.plugin_executor.default_timeout = 0.05
+        pm.run_scheduled_updates()
+        assert _wait_for(lambda: tracker.get_health_summary('slow')['hang_count'] == 1)
+        summary = tracker.get_health_summary('slow')
+        assert summary['last_hang']['operation'] == 'update'
+        assert summary['consecutive_failures'] == 1
+        # The real update() then returns: success resets the streak.
+        assert _wait_for(lambda: pm.state_manager.can_execute('slow'))
+        assert _wait_for(lambda: tracker.get_health_summary('slow')['consecutive_failures'] == 0)
+
+
+class TestDisplayTiming:
+    def test_fast_frames_record_nothing(self, pm, tracker):
+        pm.logger = MagicMock()
+        for _ in range(100):
+            pm.note_display_duration('p', 0.004)
+        assert tracker.get_health_summary('p')['slow_call_count'] == 0
+        pm.logger.warning.assert_not_called()
+
+    def test_slow_display_is_recorded_and_warned_once(self, pm, tracker):
+        pm.logger = MagicMock()
+        for _ in range(5):
+            pm.note_display_duration('p', 2.5)
+        summary = tracker.get_health_summary('p')
+        assert summary['slow_call_count'] == 5
+        assert summary['last_slow_call']['operation'] == 'display'
+        assert summary['last_slow_call']['seconds'] == 2.5
+        # Slow is not failing: the circuit breaker is untouched.
+        assert summary['consecutive_failures'] == 0
+        assert summary['circuit_state'] == CircuitState.CLOSED.value
+        assert pm.logger.warning.call_count == 1
+
+    def test_suppressed_repeats_are_counted_in_the_next_warning(self, pm):
+        pm.logger = MagicMock()
+        for _ in range(4):
+            pm.note_display_duration('p', 2.5)
+        pm.HANG_LOG_INTERVAL = 0.0
+        pm.note_display_duration('p', 2.5)
+        assert pm.logger.warning.call_count == 2
+        last = pm.logger.warning.call_args
+        assert '3 more since the last warning' in (last.args[0] % last.args[1:])
+
+    def test_display_past_the_executor_timeout_is_a_hang(self, pm, tracker):
+        pm.plugin_executor.default_timeout = 3.0
+        for _ in range(tracker.failure_threshold):
+            pm.note_display_duration('p', 3.5)
+        summary = tracker.get_health_summary('p')
+        assert summary['hang_count'] == tracker.failure_threshold
+        assert summary['last_hang']['operation'] == 'display'
+        assert summary['circuit_state'] == CircuitState.OPEN.value
+
+    def test_render_loop_times_each_frame(self, monkeypatch, emulator_mode):
+        """DisplayController._display_once hands every frame's duration on."""
+        from src import display_controller as dc_module
+        ticks = iter([100.0, 103.25])
+        monkeypatch.setattr(dc_module, 'time', types.SimpleNamespace(
+            monotonic=lambda: next(ticks)))
+        controller = dc_module.DisplayController.__new__(dc_module.DisplayController)
+        controller.plugin_manager = MagicMock()
+        lock = threading.Lock()
+        controller.plugin_manager.get_plugin_lock.return_value = lock
+        plugin = _Plugin('ticker')
+
+        assert controller._display_once(plugin, 'ticker', False) is True
+
+        controller.plugin_manager.note_display_duration.assert_called_once_with('ticker', 3.25)
+        assert not lock.locked()
+
+    def test_skipped_frame_is_not_timed(self, emulator_mode):
+        from src import display_controller as dc_module
+        controller = dc_module.DisplayController.__new__(dc_module.DisplayController)
+        controller.plugin_manager = MagicMock()
+        lock = threading.Lock()
+        lock.acquire()  # update() in flight
+        controller.plugin_manager.get_plugin_lock.return_value = lock
+        try:
+            assert controller._display_once(_Plugin('ticker'), 'ticker', False) is True
+        finally:
+            lock.release()
+        controller.plugin_manager.note_display_duration.assert_not_called()
+
+
+class TestConfigChangeIsSerialised:
+    def test_does_not_run_concurrently_with_update(self, pm):
+        pm.PLUGIN_LOCK_TIMEOUT = 2.0
+        plugin = _Plugin('p', update_seconds=0.3)
+        _install(pm, plugin)
+        pm.run_scheduled_updates()
+        assert _wait_for(lambda: plugin.in_update)
+
+        applied = pm.apply_config_change('p', {'enabled': True, 'n': 1})
+
+        assert applied is True
+        assert plugin.overlap is False
+        assert plugin.events == ['update', 'config']
+        assert plugin.config_changes == [{'enabled': True, 'n': 1}]
+
+    def test_runs_holding_the_plugin_lock(self, pm):
+        plugin = _Plugin('p')
+        _install(pm, plugin)
+        held = {}
+        plugin.on_config_change = lambda cfg: held.setdefault(
+            'locked', pm.get_plugin_lock('p').locked())
+        assert pm.apply_config_change('p', {}) is True
+        assert held == {'locked': True}
+        assert not pm.get_plugin_lock('p').locked()
+
+    def test_busy_lock_defers_the_latest_change_to_before_the_next_update(self, pm):
+        plugin = _Plugin('p')
+        _install(pm, plugin)
+        gate, render_thread = _hang_display(pm, plugin)
+        try:
+            start = time.monotonic()
+            assert pm.apply_config_change('p', {'n': 1}) is False
+            assert time.monotonic() - start < 1.0, "the wait must be bounded"
+            assert pm.apply_config_change('p', {'n': 2}) is False
+            time.sleep(0.4)  # the worker's own bounded retry also finds it busy
+            assert plugin.config_changes == []
+        finally:
+            gate.set()
+            render_thread.join(timeout=2)
+
+        pm.run_scheduled_updates()
+        assert _wait_for(lambda: plugin.update_calls == 1)
+        # Only the latest parked change, applied before the update ran.
+        assert plugin.config_changes == [{'n': 2}]
+        assert plugin.events == ['config', 'update']
+        assert plugin.overlap is False
+
+    def test_deferred_change_applied_by_worker_once_lock_frees(self, pm):
+        """No update due to piggyback on: the worker's own queued retry
+        applies it. Timeline: the watcher gives up at 0.2s, the worker waits
+        from 0.2s to 0.4s, the holder lets go at 0.3s."""
+        pm.PLUGIN_LOCK_TIMEOUT = 0.2
+        plugin = _Plugin('p')
+        _install(pm, plugin, interval=3600)
+        pm.plugin_last_update['p'] = time.time()  # not due
+        lock = pm.get_plugin_lock('p')
+        lock.acquire()
+        releaser = threading.Timer(0.3, lock.release)
+        releaser.start()
+        try:
+            assert pm.apply_config_change('p', {'n': 1}) is False
+            assert _wait_for(lambda: plugin.config_changes == [{'n': 1}])
+        finally:
+            releaser.join(timeout=2)
+        assert plugin.update_calls == 0
+        assert 'p' not in pm._deferred_config_changes
+
+    def test_stale_deferred_change_is_dropped_after_reload(self, pm):
+        plugin = _Plugin('p')
+        _install(pm, plugin)
+        pm._deferred_config_changes['p'] = (plugin, {'n': 1})
+        replacement = _Plugin('p')
+        pm.plugins['p'] = replacement
+        pm._apply_deferred_config_change('p')
+        assert plugin.config_changes == []
+        assert replacement.config_changes == []
+        assert 'p' not in pm._deferred_config_changes
+
+    def test_exceptions_still_reach_the_caller(self, pm):
+        plugin = _Plugin('p')
+        _install(pm, plugin)
+
+        def boom(cfg):
+            raise ValueError("bad config")
+
+        plugin.on_config_change = boom
+        with pytest.raises(ValueError):
+            pm.apply_config_change('p', {})
+        assert not pm.get_plugin_lock('p').locked()
+
+
+class TestHealthRecords:
+    def test_slow_call_persistence_is_rate_limited(self):
+        cache = _Cache()
+        t = PluginHealthTracker(cache_manager=cache)
+        t.record_slow_call('p', 'display', 2.5)
+        writes = cache.writes
+        for _ in range(50):
+            t.record_slow_call('p', 'display', 2.5)
+        assert cache.writes == writes
+        assert t.get_health_summary('p')['slow_call_count'] == 51
+
+    def test_busy_skip_persistence_is_rate_limited_and_breaker_free(self):
+        cache = _Cache()
+        t = PluginHealthTracker(cache_manager=cache)
+        t.record_busy_skip('p', 'update lock wait', 5.0)
+        writes = cache.writes
+        # The first one is saved at once, so the web process sees it.
+        assert PluginHealthTracker(cache_manager=cache).get_health_summary(
+            'p')['busy_skip_count'] == 1
+        for _ in range(50):
+            t.record_busy_skip('p', 'update lock wait', 5.0)
+        assert cache.writes == writes
+        summary = t.get_health_summary('p')
+        assert summary['busy_skip_count'] == 51
+        assert summary['consecutive_failures'] == 0
+        assert summary['circuit_state'] == CircuitState.CLOSED.value
+        assert t.should_skip_plugin('p') is False
+
+    def test_hang_survives_a_restart(self):
+        cache = _Cache()
+        PluginHealthTracker(cache_manager=cache).record_hang('p', 'display', 31.0)
+        summary = PluginHealthTracker(cache_manager=cache).get_health_summary('p')
+        assert summary['hang_count'] == 1
+        assert summary['last_hang']['operation'] == 'display'
+        assert summary['consecutive_failures'] == 1

test/web_interface/test_startup_reconciliation_disarmed.py

@@ -0,0 +1,23 @@
+"""
+No test may start the real app's startup reconciliation.
+
+web_interface/app.py reads the checkout's real config/config.json and
+plugin-repos/ at import, and its first request launches a reconciliation
+thread that reinstalls every configured-but-missing plugin from the live
+store. A full suite run on a dev checkout used to leave whole plugins
+untracked in plugin-repos/ that way. test/conftest.py disarms the run-once
+latch on every import of the module; this pins that.
+"""
+
+from unittest.mock import MagicMock, patch
+
+
+def test_a_request_to_the_imported_app_launches_no_reconciliation():
+    import web_interface.app as web_app
+
+    assert web_app._reconciliation_started is True
+
+    with patch.object(web_app, "threading", MagicMock()) as threading_mock:
+        web_app.app.test_client().get("/favicon.ico")
+
+    threading_mock.Thread.assert_not_called()

web_interface/blueprints/api_v3/display.py

@@ -180,9 +180,9 @@ def start_on_demand_display():
             if not resolved_plugin:
                 return jsonify({'status': 'error', 'message': f'Mode {resolved_mode} not found'}), 404
 
-    # Note: On-demand can work with disabled plugins - the display controller
+    # On-demand works with disabled plugins: the running display loads one
-    # will temporarily enable them during initialization if needed
+    # for the session and unloads it afterwards, leaving config.json alone
-    # We don't block the request here, but log it for debugging
+    # (DisplayController._load_plugin_for_on_demand). Logged for debugging.
     if api_v3.config_manager and resolved_plugin:
         config = api_v3.config_manager.load_config()
         plugin_config = config.get(resolved_plugin, {})
@@ -192,8 +192,9 @@ def start_on_demand_display():
                 resolved_plugin,
             )
 
-    # Set the on-demand request in cache FIRST (before starting service)
+    # Post the request to the mailbox the display process polls
-    # This ensures the request is available when the service starts/restarts
+    # (DisplayController._poll_on_demand_requests). Written before any
+    # service start, so a freshly started display finds it on its first poll.
     cache = _cache_manager()
     request_id = data.get('request_id') or str(uuid.uuid4())
     request_payload = {
@@ -207,18 +208,7 @@ def start_on_demand_display():
     }
     cache.set('display_on_demand_request', request_payload)
 
-    # Check if display service is running (or will be started)
     service_status = _get_display_service_status()
-    service_was_running = service_status.get('active', False)
-        
-    # Stop the display service first to ensure clean state when we will restart it
-    if service_was_running and start_service:
-        import time as time_module
-        logger.debug("Stopping display service before starting on-demand mode")
-        _stop_display_service()
-        # Wait a brief moment for the service to fully stop
-        time_module.sleep(1.5)
-        logger.debug("Display service stopped, now starting with on-demand request")
 
     if not service_status.get('active') and not start_service:
         return jsonify({
@@ -227,6 +217,18 @@ def start_on_demand_display():
             'service_status': service_status
         }), 400
 
+    # start_service means "start it if it is not running", as the UI's
+    # checkbox says; _ensure_display_service_running leaves a running service
+    # alone. This used to stop a running service, sleep 1.5s and start it
+    # again, so every on-demand or "Preview on display" click -- and every
+    # MQTT on-demand command, which posts here with the default -- cold-
+    # restarted the display process: every plugin reloaded and the panel was
+    # blank for seconds. The restart bought nothing. The running process
+    # reads this mailbox every ON_DEMAND_POLL_INTERVAL (0.25s), from its
+    # dwell sleep, its render loops and Vegas's interrupt check as well as
+    # the main loop, and a restarted one got the request the same way: the
+    # startup path only restores a session the display itself saved
+    # (display_on_demand_config), so it loaded nothing it would not have had.
     service_result = None
     if start_service:
         service_result = _ensure_display_service_running()
@@ -237,9 +239,6 @@ def start_on_demand_display():
                 'message': 'Failed to start display service. Please check service logs or start it manually.',
                 'service_result': service_result
             }), 500
-            
-        # Service was restarted (or started fresh) with on-demand request in cache
-        # The display controller will read the request during initialization or when it polls
 
     response_data = {
         'request_id': request_id,
@@ -254,10 +253,12 @@ def start_on_demand_display():
 def stop_on_demand_display():
     """Request the display controller to stop on-demand mode."""
     data = request.get_json(silent=True) or {}
-    stop_service = data.get('stop_service', False)
+    # _coerce_to_bool: bool("false") is True, which stopped the service.
+    stop_service = _coerce_to_bool(data.get('stop_service', False))
 
-    # Set the stop request in cache FIRST
+    # The running display reads the stop from the mailbox within
-    # The display controller will poll this and restart without the on-demand filter
+    # ON_DEMAND_POLL_INTERVAL and resumes normal rotation in place
+    # (_clear_on_demand); nothing is restarted.
     cache = _cache_manager()
     request_id = data.get('request_id') or str(uuid.uuid4())
     request_payload = {
@@ -266,10 +267,7 @@ def stop_on_demand_display():
         'timestamp': _pkg.time.time()
     }
     cache.set('display_on_demand_request', request_payload)
-        
+
-    # Note: The display controller's _clear_on_demand() will handle the restart
-    # to restore normal operation with all plugins
-        
     service_result = None
     if stop_service:
         service_result = _stop_display_service()