Merge remote-tracking branch 'origin/main' into claude/fix-cache-write-dedup

# Conflicts:
#	CHANGELOG.md

CHANGELOG.md

@@ -19,6 +19,39 @@ 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.
+- Re-saving unchanged data through `CacheManager.set` no longer rewrites its
+  cache file. The disk cache already skipped a payload identical to the last
+  one written, but `set()` stamps every record with the current time, so the
+  skip never fired and every plugin rewrote its unchanged API data to the SD
+  card on every update cycle. Records are now compared without that
+  timestamp, and a skipped write moves the file's mtime to it instead; reads
+  treat a record as fresh from the later of the two, so it expires exactly
+  when the rewrite would have. Changed data or a changed `ttl` still writes,
+  and so does a file another process has replaced since.
+
 ## 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
-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/cache/disk_cache.py

@@ -14,7 +14,7 @@ import tempfile
 import logging
 import threading
 import zlib
-from typing import Dict, Any, Optional, Protocol
+from typing import Dict, Any, Optional, Protocol, Tuple
 from datetime import datetime
 
 from src.common.path_safety import safe_path_component
@@ -111,18 +111,66 @@ _HEAD_RE = re.compile(
 )
 
 
-def _stale_from_head(head: bytes, max_age: Optional[int], now: float) -> bool:
+def _head_timestamp(head: bytes) -> Optional[Tuple[float, int]]:
+    """A header-first record's timestamp and the offset just past it.
+
+    None when the record does not start with a finite numeric timestamp.
+    """
+    match = _HEAD_RE.match(head)
+    if not match:
+        return None
+    try:
+        timestamp = float(match.group(1))
+    except ValueError:
+        return None
+    if not math.isfinite(timestamp):
+        return None
+    return timestamp, match.end(1)
+
+
+# FRESHNESS OF A SKIPPED WRITE
+# ----------------------------
+# CacheManager.set stamps every record with time.time(), so re-saving
+# unchanged data produced a different payload every time and DiskCache.set's
+# identical-payload skip never fired: every plugin rewrote its unchanged API
+# data to the SD card every update cycle. set() now compares header-first
+# records without their timestamp, and on a skip moves the file's mtime to
+# the timestamp the skipped record carried instead of rewriting it. So the
+# file's mtime is when its content was last saved, and a header-first
+# record is as fresh as the later of its embedded timestamp and its mtime.
+#
+# A real write sets the mtime to the embedded timestamp too, so mtime is
+# never later than the timestamp for a record written with an old one on
+# purpose -- only a skip can move it forward.
+
+
+def _refreshed_at(timestamp: float, mtime: float) -> float:
+    """When a header-first record was last saved, embedded time or mtime.
+
+    An mtime within a second of the timestamp is the write that carried it
+    (float rounding, or an older file whose mtime was not set to match),
+    not a skipped rewrite, and leaves the record as written.
+    """
+    return mtime if mtime > timestamp + 1.0 else timestamp
+
+
+def _stale_from_head(head: bytes, max_age: Optional[int], now: float,
+                     refreshed: Optional[float] = None) -> bool:
     """True when a record's header alone shows it has expired.
 
     Mirrors the expiry rule in DiskCache.get: a per-entry ttl wins over the
     caller's max_age, and no limit at all means never stale. False whenever the
     header cannot be read, so the full parse decides as it always did.
+    ``refreshed`` is the file's mtime: a skipped rewrite advances it rather
+    than the embedded timestamp (see "FRESHNESS OF A SKIPPED WRITE").
     """
     match = _HEAD_RE.match(head)
     if not match:
         return False
     try:
         timestamp = float(match.group(1))
+        if refreshed is not None:
+            timestamp = max(timestamp, refreshed)
         limit = max_age
         if match.group(2) is not None:
             ttl = float(match.group(2))
@@ -248,11 +296,13 @@ class DiskCache:
         self.cache_dir = cache_dir
         self.logger = logger or logging.getLogger(__name__)
         self._lock = threading.Lock()
-        # key -> adler32 of the last payload successfully written to the
-        # primary cache path; lets set() skip rewriting identical data
-        # (per-process only — worst case another process rewrites, never
-        # a missed write). Guarded by _lock.
-        self._write_digests: Dict[str, int] = {}
+        # key -> ((length, adler32) of the last content written to the
+        # primary cache path, (st_ino, st_size) of the file it left); lets
+        # set() skip rewriting identical data. The file identity catches
+        # another process -- the web interface writes and clears keys too --
+        # having replaced the file since, which would otherwise make the skip
+        # a missed write. Per-process only. Guarded by _lock.
+        self._write_digests: Dict[str, Tuple[Tuple[int, int], Tuple[int, int]]] = {}
     
     def get_cache_path(self, key: str) -> Optional[str]:
         """
@@ -306,7 +356,12 @@ class DiskCache:
                     # records (a season schedule is re-fetched when its cache
                     # expires), and parsing 53MB to throw it away held the GIL
                     # for ~1.8s -- a visible freeze on the panel.
-                    if _stale_from_head(f.read(_HEAD_BYTES), max_age, time.time()):
+                    head = f.read(_HEAD_BYTES)
+                    stamp = _head_timestamp(head)
+                    fresh_at = None
+                    if stamp is not None:
+                        fresh_at = _refreshed_at(stamp[0], os.fstat(f.fileno()).st_mtime)
+                    if _stale_from_head(head, max_age, time.time(), fresh_at):
                         return None
                     f.seek(0)
                     record = _loads(f.read())
@@ -315,6 +370,11 @@ class DiskCache:
             record_ts = None
             if isinstance(record, dict):
                 record_ts = record.get('timestamp')
+                if fresh_at is not None and fresh_at > stamp[0]:
+                    # A skipped rewrite refreshed this record (see "FRESHNESS
+                    # OF A SKIPPED WRITE"); hand callers the time it was last
+                    # saved, as the rewrite would have.
+                    record['timestamp'] = record_ts = fresh_at
             if record_ts is None:
                 try:
                     record_ts = os.path.getmtime(cache_path)
@@ -403,24 +463,37 @@ class DiskCache:
             self.logger.warning("Cache data for key '%s' not serializable: %s", key, e)
             return
 
-        digest = zlib.adler32(payload)
+        # A header-first record is compared without its timestamp, which
+        # CacheManager.set changes on every call (see "FRESHNESS OF A SKIPPED
+        # WRITE"). The length rides along with adler32, which is weak on its
+        # own for short payloads, and a collision here is a missed write.
+        stamp = _head_timestamp(payload[:_HEAD_BYTES])
+        stamped_at = stamp[0] if stamp is not None else None
+        content = memoryview(payload)[stamp[1]:] if stamp is not None else payload
+        digest = (len(content), zlib.adler32(content))
 
         try:
             # Atomic write to avoid partial/corrupt files
             with self._lock:
-                # Skip the disk entirely when this exact payload was already
+                # Skip the disk entirely when this content was already
                 # written for this key (plugins re-save unchanged API data
                 # every update cycle — each write is real SD-card wear).
-                # Refresh the file mtime so records that rely on it for TTL
-                # (no embedded 'timestamp') don't expire early; a metadata
-                # touch is journal-cheap compared to rewriting the data.
-                if self._write_digests.get(key) == digest:
+                # Move the file mtime instead, so the record stays as fresh as
+                # the rewrite would have left it; a metadata touch is
+                # journal-cheap compared to rewriting the data.
+                known = self._write_digests.get(key)
+                if known is not None and known[0] == digest:
                     try:
-                        os.utime(cache_path, None)
-                        return
+                        st = os.stat(cache_path)
+                        if (st.st_ino, st.st_size) == known[1]:
+                            os.utime(cache_path, None if stamped_at is None
+                                     else (stamped_at, stamped_at))
+                            return
                     except OSError:
-                        # File vanished or perms changed — fall through and write
-                        self._write_digests.pop(key, None)
+                        pass
+                    # File vanished, was replaced by another process, or its
+                    # times cannot be set — fall through and write
+                    self._write_digests.pop(key, None)
 
                 tmp_dir = os.path.dirname(cache_path)
                 # Try to create temp file in cache directory first
@@ -458,7 +531,7 @@ class DiskCache:
                                 # opened it in between was refused.
                                 _share_open_file(tmp_file.fileno(), _shared_group(tmp_dir))
                             os.replace(tmp_path, cache_path)
-                            self._write_digests[key] = digest
+                            self._remember_write(key, cache_path, digest, stamped_at)
                         finally:
                             if os.path.exists(tmp_path):
                                 try:
@@ -471,7 +544,7 @@ class DiskCache:
                             with open(cache_path, 'wb') as cache_file:
                                 cache_file.write(payload)
                                 _share_open_file(cache_file.fileno(), _shared_group(tmp_dir))
-                            self._write_digests[key] = digest
+                            self._remember_write(key, cache_path, digest, stamped_at)
                             self.logger.debug("Wrote cache for %s directly (non-atomic)", key)
                         except (IOError, OSError, PermissionError) as write_error:
                             # If direct write also fails, try fallback location
@@ -520,6 +593,26 @@ class DiskCache:
             )
             return  # Exit gracefully without raising exception
     
+    def _remember_write(self, key: str, cache_path: str,
+                        digest: Tuple[int, int], stamped_at: Optional[float]) -> None:
+        """Record a completed write so an identical set() can skip the disk.
+
+        Caller holds _lock. A header-first record's mtime is set to its
+        timestamp, so only a skipped rewrite ever moves it later (see
+        "FRESHNESS OF A SKIPPED WRITE").
+        """
+        try:
+            if stamped_at is not None:
+                os.utime(cache_path, (stamped_at, stamped_at))
+            st = os.stat(cache_path)
+        except OSError:
+            # Written but not stamped (another user's file, on the direct
+            # write path): mtime is the write time, which _refreshed_at reads
+            # as the write itself. Remember nothing; the next set() writes.
+            self._write_digests.pop(key, None)
+            return
+        self._write_digests[key] = (digest, (st.st_ino, st.st_size))
+
     def clear(self, key: Optional[str] = None) -> None:
         """
         Clear cache entry or all entries.

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,
@@ -1475,8 +1483,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)]
@@ -1491,11 +1504,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
@@ -1591,6 +1603,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 +1774,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 +2009,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 +2027,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")
 
@@ -2082,6 +2244,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
@@ -3099,6 +3269,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}
                     _plugin.on_config_change(new_config)
                     logger.debug("Plugin %s notified of config change", _pid)
                 except Exception as e:

src/plugin_system/plugin_manager.py

@@ -296,7 +296,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 +310,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 +380,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(

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_cache_write_dedup.py

@@ -0,0 +1,158 @@
+"""Re-saving unchanged data through CacheManager.set does not rewrite the file.
+
+Regression under test: DiskCache.set skipped the disk when a payload matched
+the last one written for the key, but CacheManager.set stamps every record
+with time.time(), so no two payloads ever matched and every plugin rewrote its
+unchanged API data to the SD card on every update cycle. The skip now ignores
+the timestamp, and a skipped write moves the file's mtime instead -- so the
+entry must stay exactly as fresh as the rewrite would have left it.
+"""
+
+import os
+import time
+from unittest.mock import patch
+
+import pytest
+
+from src.cache import disk_cache as disk_cache_module
+from src.cache.disk_cache import DiskCache
+from src.cache_manager import CacheManager
+
+class Clock:
+    def __init__(self):
+        self.now = time.time()
+
+    def __call__(self):
+        return self.now
+
+
+@pytest.fixture
+def clock(monkeypatch):
+    fake = Clock()
+    monkeypatch.setattr(time, "time", fake)
+    return fake
+
+
+@pytest.fixture
+def writes(monkeypatch):
+    """Paths DiskCache.set actually replaced (its atomic write path)."""
+    replaced = []
+    real = os.replace
+
+    def counting(src, dst, *args, **kwargs):
+        replaced.append(os.path.basename(dst))
+        return real(src, dst, *args, **kwargs)
+
+    monkeypatch.setattr(disk_cache_module.os, "replace", counting)
+    return replaced
+
+
+def _manager(cache_dir):
+    with patch('src.cache_manager.CacheManager._get_writable_cache_dir',
+               return_value=str(cache_dir)):
+        manager = CacheManager()
+    manager.stop_cleanup_thread()
+    return manager
+
+
+@pytest.fixture
+def cm(tmp_path):
+    return _manager(tmp_path)
+
+
+DATA = {"events": [{"id": n, "name": "x" * 20} for n in range(50)]}
+
+
+def test_resaving_unchanged_data_does_not_rewrite_the_file(cm, clock, writes):
+    cm.set("scores", DATA)
+    path = cm._get_cache_path("scores")
+    first = os.stat(path)
+    with open(path, "rb") as f:
+        first_bytes = f.read()
+
+    for _ in range(5):
+        clock.now += 60
+        cm.set("scores", DATA)
+
+    assert writes == ["scores.json"]
+    after = os.stat(path)
+    assert after.st_ino == first.st_ino
+    with open(path, "rb") as f:
+        assert f.read() == first_bytes
+    # The file records when its content was last saved.
+    assert after.st_mtime == pytest.approx(clock.now, abs=1e-3)
+
+
+def test_skipped_writes_keep_the_entry_fresh(cm, clock, tmp_path):
+    cm.set("scores", DATA)
+    for _ in range(4):          # re-saved unchanged every 200s
+        clock.now += 200
+        cm.set("scores", DATA)
+    # 800s past the only real write, well beyond max_age=300 of it.
+    assert cm.get("scores", max_age=300) == DATA
+
+    # A reader with no memory tier -- the web interface, or this service
+    # after a restart -- sees the same freshness from disk.
+    other = _manager(tmp_path)
+    record = other.get_cached_data("scores", max_age=300)
+    assert record is not None and record["data"] == DATA
+    assert record["timestamp"] == pytest.approx(clock.now, abs=1e-3)
+    assert DiskCache(str(tmp_path)).get("scores", max_age=300) is not None
+
+    # Freshness is the last save, not forever.
+    clock.now += 301
+    assert DiskCache(str(tmp_path)).get("scores", max_age=300) is None
+    assert _manager(tmp_path).get("scores", max_age=300) is None
+
+
+def test_skipped_writes_keep_a_ttl_entry_fresh(cm, clock, tmp_path):
+    cm.set("odds", DATA, ttl=120)
+    for _ in range(3):
+        clock.now += 100
+        cm.set("odds", DATA, ttl=120)
+    assert _manager(tmp_path).get("odds", max_age=10) == DATA
+
+
+def test_changed_data_writes(cm, clock, writes, tmp_path):
+    cm.set("scores", DATA)
+    clock.now += 60
+    cm.set("scores", {"events": []})
+    assert writes == ["scores.json", "scores.json"]
+    assert _manager(tmp_path).get("scores", max_age=300) == {"events": []}
+
+
+def test_changed_ttl_writes(cm, clock, writes, tmp_path):
+    cm.set("scores", DATA, ttl=60)
+    clock.now += 10
+    cm.set("scores", DATA, ttl=600)
+    clock.now += 10
+    cm.set("scores", DATA)
+    assert writes == ["scores.json"] * 3
+    record = _manager(tmp_path).get_cached_data("scores", max_age=300)
+    assert "ttl" not in record
+
+
+def test_a_record_written_old_stays_old(tmp_path, clock):
+    # Only a skip may move freshness forward: a record saved with an old
+    # timestamp on purpose is not made fresh by the write's own mtime.
+    disk = DiskCache(str(tmp_path))
+    disk.set("k", {"timestamp": clock.now - 600, "data": DATA})
+    assert disk.get("k", max_age=300) is None
+
+
+def test_a_file_replaced_by_another_process_is_rewritten(tmp_path, clock):
+    ours, theirs = DiskCache(str(tmp_path)), DiskCache(str(tmp_path))
+    ours.set("k", {"timestamp": clock.now, "data": "ours"})
+    clock.now += 10
+    theirs.set("k", {"timestamp": clock.now, "data": "theirs"})
+    clock.now += 10
+    ours.set("k", {"timestamp": clock.now, "data": "ours"})
+    assert DiskCache(str(tmp_path)).get("k", max_age=300)["data"] == "ours"
+
+
+def test_a_cleared_file_is_rewritten(cm, clock, tmp_path):
+    cm.set("scores", DATA)
+    os.remove(cm._get_cache_path("scores"))   # e.g. the web UI's delete
+    clock.now += 10
+    cm.set("scores", DATA)
+    assert _manager(tmp_path).get("scores", max_age=300) == DATA

test/test_on_demand_disabled_plugin.py

@@ -0,0 +1,423 @@
+"""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'})
+
+        plugin = controller.plugin_modes['preview_a']
+        plugin.on_config_change.assert_called_once_with({'enabled': True, 'color': 'red'})
+
+
+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/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()