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

@@ -21,23 +21,48 @@ accepts both, but the store flags the old spelling as deprecated
 
 ### 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), skips the busy plugin
-  and records the skip as a hang, so a plugin that keeps hanging opens its
-  circuit breaker and drops out of updates and rotation until the cooldown.
-  The other plugins keep updating.
+  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.
+  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

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
-request takes effect straight away.
+The on-demand start route starts `ledmatrix.service` when it is not running
+(`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.

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,
@@ -1486,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
-        would have nothing behind it.
+        is still loaded, since otherwise the mode being resumed would have
+        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)]
@@ -1502,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):
-            logger.info("Temporarily enabling plugin '%s' for on-demand mode", on_demand_plugin_id)
-            self.config.setdefault(on_demand_plugin_id, {})['enabled'] = True
-            if on_demand_plugin_id not in enabled_plugins:
-                enabled_plugins.append(on_demand_plugin_id)
+        if on_demand_plugin_id not in enabled_plugins:
+            logger.info("Loading disabled plugin '%s' for on-demand mode only", on_demand_plugin_id)
+            self._on_demand_loaded_plugins.add(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
@@ -1602,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
@@ -1768,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:
@@ -1877,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'
@@ -1886,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:
-            self.current_mode_index = self.rotation_resume_index % len(self.available_modes)
-            self.current_display_mode = self.available_modes[self.current_mode_index]
-            logger.info("Resuming rotation from saved index %d: mode '%s'", 
-                       self.rotation_resume_index, self.current_display_mode)
-        elif self.available_modes:
-            # Default to first mode if no resume index
-            self.current_mode_index = self.current_mode_index % len(self.available_modes)
-            self.current_display_mode = self.available_modes[self.current_mode_index]
-            logger.info("Resuming rotation to mode '%s' (index %d)", 
-                       self.current_display_mode, self.current_mode_index)
+        if self.available_modes:
+            saved = self.rotation_resume_index
+            # Default to the current index if no resume index
+            start = saved if saved is not None else self.current_mode_index
+            index = self._rotation_index_outside_on_demand(start % len(self.available_modes))
+            if index is None:
+                # Every mode belongs to a plugin loaded only for on-demand,
+                # which the main loop is about to unload; it then idles.
+                self.current_mode_index = 0
+                self.current_display_mode = None
+                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")
 
@@ -2093,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
@@ -3126,6 +3296,11 @@ class DisplayController:
                     prepared = prepare(_pid, new_config) if callable(prepare) else None
                     if isinstance(prepared, dict):
                         new_config = prepared
+                    if _pid in self._on_demand_loaded_plugins:
+                        # 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

src/plugin_system/plugin_executor.py

@@ -23,8 +23,10 @@ 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() or on_config_change() -- one that is hung or far slower than it
-    should be -- so the caller skipped the plugin rather than wait on it.
+    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.
     """
 
 

src/plugin_system/plugin_health.py

@@ -256,7 +256,7 @@ class PluginHealthTracker:
     
     def record_hang(self, plugin_id: str, operation: str, seconds: float,
                     error: Optional[Exception] = None) -> None:
-        """Record a call that ran past its limit, or held the plugin's lock past it.
+        """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
@@ -264,12 +264,14 @@ class PluginHealthTracker:
         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"``, ``"update"``, or the lock
-                wait that found one of them still running.
-            seconds: How long it had been running, or how long the lock
-                was waited on, when this was recorded.
+            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.
         """
@@ -287,9 +289,9 @@ class PluginHealthTracker:
         # record_failure saves the record, hang fields included.
         self.record_failure(plugin_id, error)
 
-    #: Minimum seconds between persisting a plugin's slow-call counters. The
-    #: in-memory record is updated on every slow call; a plugin that is slow
-    #: on every frame must not become an SD-card write per frame.
+    #: 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:
@@ -308,10 +310,46 @@ class PluginHealthTracker:
             'seconds': round(float(seconds), 3),
             'time': now,
         }
-        saved_at = self.__dict__.setdefault('_slow_call_saved_at', {})
-        last = saved_at.get(plugin_id)
+        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[plugin_id] = now
+            saved_at[(kind, plugin_id)] = now
             self._save_health_state(plugin_id, state)
 
     def set_degraded(self, plugin_id: str, reason: Optional[str]) -> None:
@@ -410,6 +448,8 @@ class PluginHealthTracker:
             '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

@@ -353,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.
         
@@ -367,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
@@ -433,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(
@@ -1071,8 +1081,8 @@ class PluginManager:
         self,
         plugin_id: str,
         exc: Optional[Exception] = None,
-        hang: Optional[Tuple[str, float]] = None,
         log: bool = True,
+        count_failure: bool = True,
     ) -> None:
         """Apply the standard failure-recovery path for a plugin update.
 
@@ -1086,10 +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.
-            hang: ``(operation, seconds)`` when the failure is a hang rather
-                 than an error, so health records it as one.
             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:
@@ -1110,9 +1121,7 @@ class PluginManager:
         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 hang is not None:
-            self._record_hang(plugin_id, hang[0], hang[1], err)
-        elif 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:
@@ -1139,19 +1148,14 @@ class PluginManager:
                      err: Exception) -> None:
         """Record a hang in plugin health: a failure to the circuit breaker.
 
-        Goes through PluginHealthTracker.record_hang when the tracker has it
-        (it also counts the hang separately), else plain record_failure. Never
+        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:
-            record_hang = getattr(tracker, 'record_hang', None)
-            if callable(record_hang):
-                record_hang(plugin_id, operation, seconds, err)
-            else:
-                tracker.record_failure(plugin_id, err)
+            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)
 
@@ -1356,10 +1360,12 @@ class PluginManager:
         timeout elapses first.
 
         The lock wait is bounded by PLUGIN_LOCK_TIMEOUT. Whatever holds it
-        past that -- a hung display() on the render thread or a lingering
-        executor thread -- costs this worker that long once per attempt, and
-        the plugin is skipped and recorded as hung (_skip_busy_update); the
-        other plugins' queued updates carry on.
+        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()
@@ -1399,29 +1405,42 @@ class PluginManager:
         """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, last-update stamped so the retry waits
-        a full interval -- recorded as a hang, so a plugin stuck this way
-        opens its circuit breaker after the usual number of attempts and
-        stops being scheduled or displayed until the cooldown.
+        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 a failure would resurrect it as ENABLED.
+            # 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() or update() of it is hung or very slow); other "
-            "plugins keep updating", plugin_id, waited)
+            "(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 hung or slow display()/update(); update skipped"),
-            hang=('update lock wait', waited),
-            log=False)
+                "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:

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
-service must not import a name that does not exist.
+"""POST /display/on-demand/start and /stop must not restart a running display.
 
-display.py has `import web_interface.blueprints.api_v3 as _pkg` and reads
-mutable, test-patched attributes back through it (`_pkg.time.time()`,
-`_pkg._get_starlark_plugin()`, ...) rather than binding them by value, per
-the package's own docstring. One spot went further and wrote a genuine
-`import` *statement* against that alias --
+The start route used to treat ``start_service`` (default True, and what both
+the web UI and the MQTT bridge send) as "restart": with the service running it
+ran ``systemctl stop``, slept 1.5s and started it again. Every on-demand or
+"Preview on display" click therefore cold-restarted the display process --
+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,
-not a real top-level package, so `import _pkg.time` is not something Python
-can resolve; it raises ModuleNotFoundError. That line only runs when the
-display service is already running and the caller also asked to (re)start
-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.
+This file previously pinned that restart path (it guarded a broken
+``import _pkg.time`` inside it). The path is gone; these tests pin its
+replacement: a running service is left alone, a stopped one is started (only
+when start_service is set), and the request lands in the mailbox either way.
 
-The route wraps its body in `except Exception`, so the failure reached the
-caller as a handled 500 with a generic message, not an unhandled crash --
-but a 500 all the same on a request that should have restarted the service
-and reported success.
+The service helpers are patched where they run. display.py binds
+_get_display_service_status by value, while _ensure_display_service_running
+(in the package __init__) looks it up in its own module, so both are patched;
+_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):
-    """Force the `service_was_running and start_service` branch.
+def service(api_v3_module):
+    """A display service whose state the test sets; records systemctl calls.
 
-    plugin_manager and config_manager are set to None so the route takes
-    the simplest path to that branch rather than tripping over unrelated
-    MagicMock plumbing. The cache is the blueprint's cache_manager, which
-    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.
+    plugin_manager and config_manager are None so the route skips plugin
+    resolution (not what is under test here). The cache is the blueprint's
+    MagicMock cache_manager, so mailbox writes are visible as set() calls.
     """
     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, \
-         patch("web_interface.blueprints.api_v3.display._stop_display_service") as stop_service, \
-         patch("web_interface.blueprints.api_v3.display._ensure_display_service_running") as ensure_running:
-        # Active before the request: service_was_running becomes True.
-        get_status.return_value = {"active": True}
-        ensure_running.return_value = {"active": True}
+    def status():
+        return {"active": state["active"]}
+
+    def systemctl(args):
+        if args[-2:] == ["start", "ledmatrix.service"]:
+            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 test_it_does_not_500(self, api_v3_client, restart_path):
-        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 _mailbox_writes(cache):
+    return [c.args[1] for c in cache.set.call_args_list if c.args and c.args[0] == MAILBOX]
 
-    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(
-            self, api_v3_client, restart_path):
-        # 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,
-        # kept here so the branch condition itself stays covered.
-        restart_path["get_status"].return_value = {"active": False}
-        response = api_v3_client.post(
-            URL, json={"plugin_id": "weather", "start_service": True})
+def _systemctl_verbs(run_systemctl):
+    return [c.args[0][-2] for c in run_systemctl.call_args_list]
+
+
+class TestStartWhileTheServiceIsRunning:
+    @pytest.mark.parametrize("body", [
+        {"plugin_id": "weather"},                           # "Preview on display", MQTT
+        {"plugin_id": "weather", "start_service": True},    # on-demand modal, box ticked
+        {"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):
-        """Otherwise the mode being resumed has nothing behind it."""
+    def test_a_disabled_on_demand_plugin_is_still_loaded(self, controller):
+        """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_hang_containment.py

@@ -9,9 +9,12 @@ 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, skips the busy plugin and
-  records the skip as a hang, so repeats open that plugin's circuit breaker;
-- display() calls are timed, slow ones recorded, overlong ones counted as hangs;
+- 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().
 
@@ -167,18 +170,24 @@ class TestHungDisplayDoesNotStallTheWorker:
         assert _wait_for(lambda: plugin.update_calls == 1)
 
 
-class TestLockTimeoutIsRecordedAsAHang:
-    def test_skip_lands_in_health_and_state(self, pm, tracker):
+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')['hang_count'] == 1)
+            assert _wait_for(lambda: tracker.get_health_summary('hung')['busy_skip_count'] == 1)
             summary = tracker.get_health_summary('hung')
-            assert summary['consecutive_failures'] == 1
-            assert summary['last_hang']['operation'] == 'update lock wait'
-            assert 'busy' in summary['last_error']
+            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.
@@ -187,24 +196,78 @@ class TestLockTimeoutIsRecordedAsAHang:
             gate.set()
             render_thread.join(timeout=2)
 
-    def test_repeated_hangs_open_the_circuit_breaker(self, pm, tracker):
+    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')['hang_count'] == attempt)
+                    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
-            assert pm.state_manager.can_execute('hung')
         finally:
             gate.set()
             render_thread.join(timeout=2)
@@ -239,7 +302,9 @@ class TestLockTimeoutIsRecordedAsAHang:
             assert _wait_for(lambda: 'hung' not in pm._pending_updates)
             time.sleep(0.35)
             assert pm.state_manager.get_state('hung') == PluginState.UNLOADED
-            assert tracker.get_health_summary('hung')['hang_count'] == 0
+            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)
@@ -433,6 +498,23 @@ class TestHealthRecords:
         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)

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
-    # will temporarily enable them during initialization if needed
-    # We don't block the request here, but log it for debugging
+    # On-demand works with disabled plugins: the running display loads one
+    # for the session and unloads it afterwards, leaving config.json alone
+    # (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)
-    # This ensures the request is available when the service starts/restarts
+    # Post the request to the mailbox the display process polls
+    # (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 display controller will poll this and restart without the on-demand filter
+    # The running display reads the stop from the mailbox within
+    # 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()