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

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

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

scripts/dev_server.py

@@ -236,13 +236,15 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
         try:
             plugin_instance.update()
         except Exception as e:
-            warnings.append(f"update() raised: {e}")
+            logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
+            warnings.append(f"update() raised: {type(e).__name__} — see server log")
 
     # Run display()
     try:
         plugin_instance.display(force_clear=True)
     except Exception as e:
-        errors.append(f"display() raised: {e}")
+        logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
+        errors.append(f"display() raised: {type(e).__name__} — see server log")
 
     render_time_ms = round((time.time() - start_time) * 1000, 1)
 
@@ -259,20 +261,25 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
 def _trusted_plugin_dir(plugin_dir: Path) -> Optional[Path]:
     """Re-derive a plugin directory from the search dirs' own listings.
 
-    Path-injection barrier: the returned Path is constructed purely from
+    Path-injection barrier: unlike ``Path.iterdir()`` (which CodeQL doesn't
-    trusted directory enumeration (``iterdir``) — request-derived strings
+    recognize as a taint-clearing enumeration), ``os.scandir()`` is. The
+    returned Path is built from a trusted root plus a name the filesystem
+    itself produced under that root via scandir — request-derived strings
     never enter its construction — so a crafted plugin id can never make
     downstream file access leave the plugin search dirs. Comparison is by
-    path equality, deliberately without symlink resolution (dev plugins
+    name, deliberately without symlink resolution (dev plugins are
-    are commonly symlinked into plugins/).
+    commonly symlinked into plugins/).
     """
-    wanted = Path(os.path.normpath(str(plugin_dir)))
+    wanted_name = Path(os.path.normpath(str(plugin_dir))).name
     for search_dir in get_search_dirs():
-        if not search_dir.is_dir():
+        search_dir_str = str(search_dir)
+        try:
+            with os.scandir(search_dir_str) as entries:
+                for entry in entries:
+                    if entry.name == wanted_name and entry.is_dir():
+                        return Path(search_dir_str) / entry.name
+        except OSError:
             continue
-        for entry in search_dir.iterdir():
-            if entry.is_dir() and entry == wanted:
-                return entry
     return None
 
 
@@ -280,22 +287,25 @@ def _parse_render_request(data):
     """Shared /api/render* request prep. Returns (plugin_dir, manifest, config,
     mock_data, skip_update) or raises ValueError with a client message."""
     plugin_id = data['plugin_id']
-    plugin_dir = find_plugin_dir(plugin_id)
+    candidate_dir = find_plugin_dir(plugin_id)
-    if plugin_dir:
+    # Never reuse `candidate_dir` past this point: it's built from
-        plugin_dir = _trusted_plugin_dir(plugin_dir)
+    # request-derived input, and a variable reassigned only on some paths
-    if not plugin_dir:
+    # isn't a barrier CodeQL's flow analysis honors. `trusted_dir` is the
+    # sole name used below, always the scandir-sourced result.
+    trusted_dir = _trusted_plugin_dir(candidate_dir) if candidate_dir else None
+    if not trusted_dir:
         raise LookupError(f'Plugin not found: {plugin_id}')
 
-    manifest_path = plugin_dir / 'manifest.json'
+    manifest_path = trusted_dir / 'manifest.json'
     with open(manifest_path, 'r') as f:
         manifest = json.load(f)
 
     # Build config: schema defaults + user overrides
     config = {'enabled': True}
-    config.update(load_config_defaults(plugin_dir))
+    config.update(load_config_defaults(trusted_dir))
     config.update(data.get('config', {}))
 
-    return plugin_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
+    return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
 
 
 @app.route('/api/render', methods=['POST'])

src/adaptive_images.py

@@ -25,7 +25,7 @@ uncached primitives.
 """
 
 from dataclasses import dataclass
-from typing import Any, Optional, Tuple, Union
+from typing import Any, Optional, Tuple
 
 from PIL import Image
 
@@ -136,7 +136,15 @@ def fit_image(img: Image.Image, box: Any, *, mode: str = "contain",
         scale = min(scale, 1.0)
     out_w = max(1, round(src_w * scale))
     out_h = max(1, round(src_h * scale))
-    out = work if (out_w, out_h) == (src_w, src_h) else work.resize((out_w, out_h), resample)
+    if (out_w, out_h) == (src_w, src_h):
+        # No resize needed — but `work` may still BE the caller's original
+        # image (RGBA source, no ink crop). The result must always be an
+        # independent copy: LayoutContext caches ImageFitResults, and an
+        # aliased image would let later mutations of the source corrupt
+        # cached fits (or vice versa).
+        out = work.copy() if work is img else work
+    else:
+        out = work.resize((out_w, out_h), resample)
     return ImageFitResult(out, out_w, out_h, scale, mode, (src_w, src_h))
 
 

src/adaptive_layout.py

@@ -29,7 +29,7 @@ freetype.Face, so it drops straight into DisplayManager.draw_text().
 
 import logging
 from collections import OrderedDict
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 import freetype
@@ -328,13 +328,28 @@ class LayoutContext:
         # fonts, which step between crisp ladder rungs instead.
         self.scale = min(self.width / max(1, design_w),
                          self.height / max(1, design_h))
-        self._fit_cache: Dict[Any, FitResult] = {}
+        # LRU-bounded: entries are small, but keys embed the fitted TEXT —
-        # LRU-bounded (images are big, unlike text fits). Entries hold a
+        # a plugin fitting changing text (a live game clock, a ticker) on a
-        # strong reference to the source image when keyed by id() so the id
+        # 24/7 service would otherwise grow this without bound.
-        # can't be recycled out from under the cache.
+        self._fit_cache: "OrderedDict[Any, FitResult]" = OrderedDict()
+        # LRU-bounded (images are big). Entries hold a strong reference to
+        # the source image when keyed by id() so the id can't be recycled
+        # out from under the cache.
         self._image_cache: "OrderedDict[Any, Tuple[Any, Any]]" = OrderedDict()
 
     _IMAGE_CACHE_MAX = 64
+    _FIT_CACHE_MAX = 512
+
+    def _fit_cache_get(self, key: Any) -> Optional["FitResult"]:
+        cached = self._fit_cache.get(key)
+        if cached is not None:
+            self._fit_cache.move_to_end(key)
+        return cached
+
+    def _fit_cache_put(self, key: Any, result: "FitResult") -> None:
+        self._fit_cache[key] = result
+        while len(self._fit_cache) > self._FIT_CACHE_MAX:
+            self._fit_cache.popitem(last=False)
 
     # ---- the three adaptation patterns --------------------------------
 
@@ -373,11 +388,11 @@ class LayoutContext:
         acceptable rendering exists."""
         box_w, box_h = _box_dims(box)
         key = ("text", text, box_w, box_h, ladder, ellipsis)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
         result = self._walk_ladder(text, ladder, box_w, box_h, ellipsis)
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
@@ -416,14 +431,14 @@ class LayoutContext:
         box_w, box_h = _box_dims(box)
         effective_scale = self.scale if scale is None else scale
         key = ("text_prop", text, box_w, box_h, ladder, base_size_px, ellipsis, effective_scale)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
         target = base_size_px * effective_scale
         eligible = [step for step in ladder if step.size_px <= target]
         candidates = eligible if eligible else (min(ladder, key=lambda s: s.size_px),)
         result = self._walk_ladder(text, candidates, box_w, box_h, ellipsis)
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
@@ -460,7 +475,7 @@ class LayoutContext:
         one wouldn't (baseball's multiline pattern). Text is the widest line."""
         box_w, box_h = _box_dims(box)
         key = ("lines", tuple(lines), box_w, box_h, ladder, spacing)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
 
@@ -482,7 +497,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     def font_for_rows(self, rows: int, box_h: int,
@@ -491,7 +506,7 @@ class LayoutContext:
         (baseball's traditional-scoreboard pattern). Measures a digit/cap
         sample rather than specific strings."""
         key = ("rows", rows, box_h, ladder)
-        cached = self._fit_cache.get(key)
+        cached = self._fit_cache_get(key)
         if cached is not None:
             return cached
 
@@ -508,7 +523,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache[key] = result
+        self._fit_cache_put(key, result)
         return result
 
     # ---- images ---------------------------------------------------------

src/display_controller.py

@@ -502,7 +502,10 @@ class DisplayController:
 
             # Run plugin updates inside the Vegas loop so the inter-iteration
             # gap is <1 ms (nothing left for _tick_plugin_updates() to do).
-            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
+            # Use the Vegas-aware variant so plugins that got fresh data are
+            # hot-swapped into the scroll promptly instead of waiting for the
+            # next full cycle.
+            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates_for_vegas)
 
             # Wire multi-display sync into Vegas render pipeline
             follower_pos = self.config.get("sync", {}).get("follower_position", "left")
@@ -625,14 +628,24 @@ class DisplayController:
         # Check if per-day schedule is configured
         days_config = schedule_config.get('days')
 
-        # Determine which schedule to use
+        # Determine which schedule to use. Respect an explicit 'mode' field
+        # (like the dim schedule does) so a stray/legacy 'days' dict left over
+        # from config migration or a prior per-day setup can't silently
+        # override a user's Global schedule selection.
+        mode = schedule_config.get('mode')
+        mode_normalized = mode.replace('_', '-') if mode else None
+
         use_per_day = False
-        if days_config:
+        if mode_normalized == 'global':
-            # Check if days dict is not empty and contains current day
+            use_per_day = False
-            if days_config and current_day in days_config:
+        elif mode_normalized == 'per-day':
-                use_per_day = True
+            use_per_day = bool(days_config and current_day in days_config)
         elif days_config:
-                # Days dict exists but doesn't have current day - fall back to global
+            # No explicit mode recorded (legacy config) - fall back to
+            # inferring from presence of a 'days' dict for the current day.
+            if current_day in days_config:
+                use_per_day = True
+            else:
                 logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
         
         if use_per_day:
@@ -828,6 +841,42 @@ class DisplayController:
                     if hasattr(self.plugin_manager, 'health_tracker') and self.plugin_manager.health_tracker:
                         self.plugin_manager.health_tracker.record_failure(plugin_id, exc)
 
+    def _tick_plugin_updates_for_vegas(self) -> None:
+        """Run scheduled plugin updates and tell Vegas mode which plugins
+        actually got fresh data, so it can hot-swap them into the scroll
+        without waiting for a full cycle to complete.
+
+        Used as the Vegas coordinator's update callback instead of the plain
+        _tick_plugin_updates() so that a live score change is reflected in
+        the ticker within a few seconds rather than at the next cycle
+        boundary (which, depending on min/max_cycle_duration, can be
+        minutes away). Restores wiring that PR #299 added and PR #330's
+        sync-mode refactor inadvertently dropped: coordinator.mark_plugin_updated()
+        has been unreachable dead code since.
+
+        Delegates the before/after plugin_last_update snapshot to
+        PluginManager.run_scheduled_updates_with_changes() so the snapshot,
+        update pass, and diff are lock-protected against this callback's own
+        background update-tick thread racing the main render loop.
+        """
+        if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
+            self._tick_plugin_updates()
+            return
+
+        updated = self.plugin_manager.run_scheduled_updates_with_changes()
+
+        vc = getattr(self, "vegas_coordinator", None)
+        if vc is None:
+            return
+
+        if updated:
+            logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
+            for plugin_id in updated:
+                try:
+                    vc.mark_plugin_updated(plugin_id)
+                except Exception:  # pylint: disable=broad-except
+                    logger.exception("Error marking plugin %s updated for Vegas", plugin_id)
+
     def _tick_plugin_updates(self):
         """Run scheduled plugin updates if the plugin manager supports them."""
         if not self.plugin_manager:

src/plugin_system/plugin_loader.py

@@ -437,8 +437,7 @@ class PluginLoader:
                 if not Path(existing_file).resolve().is_relative_to(resolved_dir):
                     evicted[mod_name] = sys.modules.pop(mod_name)
                     self.logger.debug(
-                        "Evicted stale module '%s' (from %s) before loading plugin in %s",
+                        "Evicted stale bare-name module '%s' before loading plugin", mod_name,
-                        mod_name, existing_file, plugin_dir,
                     )
             except (ValueError, TypeError):
                 continue
@@ -551,7 +550,7 @@ class PluginLoader:
             plugin_dir_str = str(plugin_dir)
             if plugin_dir_str not in sys.path:
                 sys.path.insert(0, plugin_dir_str)
-                self.logger.debug("Added plugin directory to sys.path: %s", plugin_dir_str)
+                self.logger.debug("Added plugin %s's directory to sys.path", plugin_id)
 
             # Import the plugin module
             module_name = f"plugin_{plugin_id.replace('-', '_')}"
@@ -563,8 +562,8 @@ class PluginLoader:
 
             spec = importlib.util.spec_from_file_location(module_name, entry_file)
             if spec is None or spec.loader is None:
+                self.logger.error("Could not create module spec for plugin %s", plugin_id)
                 error_msg = f"Could not create module spec for {entry_file}"
-                self.logger.error(error_msg)
                 raise PluginError(error_msg, plugin_id=plugin_id, context={'entry_file': str(entry_file)})
 
             module = importlib.util.module_from_spec(spec)

src/plugin_system/plugin_manager.py

@@ -76,6 +76,12 @@ class PluginManager:
         # concurrent mutation (background reconciliation) and reads (requests).
         self._discovery_lock = threading.RLock()
 
+        # Lock protecting plugin_last_update from concurrent mutation/iteration.
+        # It's written from run_scheduled_updates()/update_all_plugins() (main
+        # loop) and read/diffed by run_scheduled_updates_with_changes(), which
+        # Vegas mode calls from its own background update-tick thread.
+        self._plugin_last_update_lock = threading.RLock()
+
         # Active plugins
         self.plugins: Dict[str, Any] = {}
         self.plugin_manifests: Dict[str, Dict[str, Any]] = {}
@@ -317,6 +323,7 @@ class PluginManager:
 
             # Store plugin instance
             self.plugins[plugin_id] = plugin_instance
+            with self._plugin_last_update_lock:
                 self.plugin_last_update[plugin_id] = 0.0
             # Invalidate cached interval so next tick re-derives it for this plugin
             self._update_interval_cache.pop(plugin_id, None)
@@ -429,6 +436,7 @@ class PluginManager:
             
             # Remove from active plugins
             del self.plugins[plugin_id]
+            with self._plugin_last_update_lock:
                 self.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
             
@@ -698,6 +706,7 @@ class PluginManager:
             'recoverable': True,
         }
         self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
+        with self._plugin_last_update_lock:
             self.plugin_last_update[plugin_id] = failure_time
         self.state_manager.set_state_with_error(plugin_id, PluginState.ENABLED, error_info, error=err)
         if self.health_tracker:
@@ -731,6 +740,7 @@ class PluginManager:
             if interval is None:
                 continue
 
+            with self._plugin_last_update_lock:
                 last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
@@ -762,6 +772,7 @@ class PluginManager:
                         success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                     
                     if success:
+                        with self._plugin_last_update_lock:
                             self.plugin_last_update[plugin_id] = current_time
                         self.state_manager.record_update(plugin_id)
                         # Update state back to ENABLED
@@ -775,6 +786,31 @@ class PluginManager:
                     self.logger.exception("Error updating plugin %s: %s", plugin_id, exc)
                     self._record_update_failure(plugin_id, exc=exc)
 
+    def run_scheduled_updates_with_changes(self, current_time: Optional[float] = None) -> List[str]:
+        """
+        Like run_scheduled_updates(), but also returns the plugin_ids whose
+        plugin_last_update timestamp actually advanced during this call.
+
+        The before/after snapshots and the update pass itself are each
+        individually lock-protected against concurrent plugin_last_update
+        mutation (Vegas mode calls this from its own background
+        update-tick thread, racing the main render loop's plugin updates),
+        so callers get an atomic "who got fresh data" answer without
+        reaching into plugin_last_update themselves. The lock is not held
+        across the update pass so slow/blocking plugin update() calls don't
+        serialize against other plugin_last_update readers.
+        """
+        with self._plugin_last_update_lock:
+            old_times = dict(self.plugin_last_update)
+
+        self.run_scheduled_updates(current_time)
+
+        with self._plugin_last_update_lock:
+            return [
+                plugin_id for plugin_id, new_time in self.plugin_last_update.items()
+                if new_time > old_times.get(plugin_id, 0.0)
+            ]
+
     def update_all_plugins(self) -> None:
         """
         Update all enabled plugins.
@@ -797,6 +833,7 @@ class PluginManager:
             try:
                 success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                 if success:
+                    with self._plugin_last_update_lock:
                         self.plugin_last_update[plugin_id] = time.time()
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)

src/plugin_system/testing/mocks.py

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

src/vegas_mode/render_pipeline.py

@@ -297,6 +297,8 @@ class RenderPipeline:
         Returns True when:
         - Cycle is complete and we should start fresh
         - Staging buffer has new content
+        - A plugin currently visible in the scroll has pending updated data
+          (e.g. a live score changed) — standalone (non-sync) mode only
         """
         if self._cycle_complete:
             return True
@@ -314,6 +316,12 @@ class RenderPipeline:
         if buffer_status['staging_count'] > 0:
             return True
 
+        # Trigger recompose when pending updates affect visible segments, so
+        # live score/status changes reach the display within a few seconds
+        # instead of waiting for the next full cycle.
+        if self.stream_manager.has_pending_updates_for_visible_segments():
+            return True
+
         return False
 
     def hot_swap_content(self) -> bool:

test/test_adaptive_images.py

@@ -194,3 +194,15 @@ class TestBasePluginDrawImage:
         assert ifit.height == 32
         # pasted onto the mock's canvas
         assert plugin.display_manager.image.getpixel((16, 16)) != (0, 0, 0)
+
+
+class TestResultIndependence:
+    def test_same_size_fit_never_aliases_the_source(self):
+        """LayoutContext caches ImageFitResults — an aliased image would let
+        later mutations of the source corrupt cached fits (or vice versa)."""
+        from PIL import ImageDraw
+        src = Image.new("RGBA", (20, 20), (255, 0, 0, 255))
+        fit = fit_image(src, (20, 20))
+        assert fit.image is not src
+        ImageDraw.Draw(src).rectangle([0, 0, 19, 19], fill=(0, 255, 0, 255))
+        assert fit.image.getpixel((5, 5)) == (255, 0, 0, 255)

test/test_adaptive_layout.py

@@ -436,3 +436,19 @@ class TestBasePluginIntegration:
                          MockCacheManager(), pm)
         assert plugin.layout.design_size == (64, 32)
         assert plugin.layout.scale == 2.0
+
+
+class TestFitCacheBound:
+    def test_fit_cache_is_lru_bounded(self, ctx):
+        """A plugin fitting changing text (live game clock, ticker) on a
+        24/7 service must not grow the fit cache without bound."""
+        for i in range(ctx._FIT_CACHE_MAX + 100):
+            ctx.fit_text(f"tick {i}", Region(0, 0, 100, 20))
+        assert len(ctx._fit_cache) <= ctx._FIT_CACHE_MAX
+
+    def test_lru_keeps_recent_entries_hot(self, ctx):
+        hot = ctx.fit_text("stay hot", Region(0, 0, 100, 20))
+        for i in range(ctx._FIT_CACHE_MAX - 1):
+            ctx.fit_text(f"cold {i}", Region(0, 0, 100, 20))
+            ctx.fit_text("stay hot", Region(0, 0, 100, 20))  # keep touching it
+        assert ctx.fit_text("stay hot", Region(0, 0, 100, 20)) is hot

test/test_display_controller_vegas_tick.py

@@ -0,0 +1,88 @@
+"""
+Regression tests for DisplayController._tick_plugin_updates_for_vegas().
+
+PR #299 added logic to detect which plugins actually got fresh data on a
+scheduled-update tick and notify Vegas mode via
+vegas_coordinator.mark_plugin_updated(), so a live score change reaches the
+scroll within seconds instead of waiting for a full cycle. PR #330's
+multi-display sync refactor deleted this method (folding the callback back
+to the plain _tick_plugin_updates(), which reports nothing), silently
+orphaning VegasModeCoordinator.mark_plugin_updated() -- it has had zero
+callers since.
+"""
+
+from typing import Dict, List, Optional
+from unittest.mock import MagicMock
+
+from src.display_controller import DisplayController
+
+
+def _make_controller(updated: Optional[List[str]] = None, vegas_coordinator: Optional[MagicMock] = None) -> DisplayController:
+    dc = object.__new__(DisplayController)
+    dc.plugin_manager = MagicMock()
+    dc.plugin_manager.run_scheduled_updates_with_changes.return_value = list(updated or [])
+    dc.vegas_coordinator = vegas_coordinator
+    return dc
+
+
+class TestTickPluginUpdatesForVegas:
+    def test_marks_only_plugins_whose_timestamp_advanced(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_called_once_with("stock-news")
+
+    def test_no_advance_marks_nothing(self):
+        vc = MagicMock()
+        dc = _make_controller(updated=[], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()
+
+        vc.mark_plugin_updated.assert_not_called()
+
+    def test_no_vegas_coordinator_does_not_raise(self):
+        dc = _make_controller(updated=["stock-news"], vegas_coordinator=None)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+    def test_mark_plugin_updated_exception_does_not_propagate(self):
+        """One plugin's mark_plugin_updated failing must not stop the tick
+        or crash the update loop it runs in."""
+        vc = MagicMock()
+        vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
+        dc = _make_controller(updated=["a", "b"], vegas_coordinator=vc)
+
+        dc._tick_plugin_updates_for_vegas()  # must not raise
+
+        assert vc.mark_plugin_updated.call_count == 2
+
+
+class TestVegasCoordinatorCallbackWiring:
+    def test_initialize_wires_vegas_aware_tick_as_update_callback(self):
+        """The Vegas coordinator must be given the Vegas-aware
+        _tick_plugin_updates_for_vegas as its update callback, not the plain
+        _tick_plugin_updates() -- that's the exact wiring PR #330 dropped."""
+        dc = object.__new__(DisplayController)
+        dc.config = {"display": {"vegas_scroll": {"enabled": True}}, "sync": {}}
+        dc.display_manager = MagicMock()
+        dc.plugin_manager = MagicMock()
+        dc.sync_manager = MagicMock()
+        dc._check_live_priority = MagicMock()
+        dc._check_vegas_interrupt = MagicMock(return_value=False)
+
+        fake_coordinator = MagicMock()
+
+        import src.display_controller as dc_module
+        original_imported = dc_module._vegas_mode_imported
+        original_class = dc_module.VegasModeCoordinator
+        try:
+            dc_module._vegas_mode_imported = True
+            dc_module.VegasModeCoordinator = MagicMock(return_value=fake_coordinator)
+            dc._initialize_vegas_mode()
+        finally:
+            dc_module._vegas_mode_imported = original_imported
+            dc_module.VegasModeCoordinator = original_class
+
+        fake_coordinator.set_update_callback.assert_called_once_with(dc._tick_plugin_updates_for_vegas)

test/test_testing_mocks.py

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

test/test_vegas_render_pipeline_recompose.py

@@ -0,0 +1,72 @@
+"""
+Regression tests for RenderPipeline.should_recompose()'s pending-updates check.
+
+PR #299 added a check so a plugin's live score/status change (a "pending
+update" in StreamManager) triggers a hot-swap within a few seconds instead
+of waiting for a full scroll cycle to complete. PR #330 (multi-display sync)
+refactored should_recompose() and dropped that check entirely -- not just
+gated behind the new sync-mode deferral it added, but removed outright, so
+even standalone (non-sync) installations silently lost live-refresh and fell
+back to waiting for full cycle boundaries (which, depending on
+min/max_cycle_duration, can be minutes).
+"""
+
+from unittest.mock import MagicMock
+
+from src.vegas_mode.config import VegasModeConfig
+from src.vegas_mode.render_pipeline import RenderPipeline
+
+
+class FakeDisplayManager:
+    width = 64
+    height = 32
+
+
+def _make_pipeline(sync_manager=None):
+    stream_manager = MagicMock()
+    stream_manager.get_buffer_status.return_value = {'staging_count': 0}
+    pipeline = RenderPipeline(VegasModeConfig(), FakeDisplayManager(), stream_manager)
+    pipeline.sync_manager = sync_manager
+    return pipeline, stream_manager
+
+
+class TestShouldRecompose:
+    def test_cycle_complete_always_recomposes(self):
+        pipeline, stream_manager = _make_pipeline()
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_no_pending_updates_no_staging_does_not_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is False
+
+    def test_pending_updates_on_visible_segment_triggers_recompose(self):
+        """The actual regression: a live-updated plugin currently in view
+        must trigger a recompose instead of waiting for cycle end."""
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True
+
+    def test_staging_buffer_content_triggers_recompose(self):
+        pipeline, stream_manager = _make_pipeline()
+        stream_manager.get_buffer_status.return_value = {'staging_count': 1}
+        stream_manager.has_pending_updates_for_visible_segments.return_value = False
+        assert pipeline.should_recompose() is True
+
+    def test_sync_active_defers_pending_updates_to_cycle_boundary(self):
+        """Sync-mode deferral (PR #330's actual intent) must still hold:
+        pending updates alone must NOT trigger a mid-cycle hot-swap when a
+        follower display is attached, since that causes a visible
+        freeze+jump on the follower. This must keep working after
+        restoring the non-sync pending-updates check above."""
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is False
+
+    def test_sync_active_still_recomposes_on_cycle_complete(self):
+        pipeline, stream_manager = _make_pipeline(sync_manager=MagicMock())
+        pipeline._cycle_complete = True
+        stream_manager.has_pending_updates_for_visible_segments.return_value = True
+        assert pipeline.should_recompose() is True

web_interface/blueprints/api_v3.py

@@ -329,6 +329,7 @@ def save_schedule_config():
         }
 
         mode = data.get('mode', 'global')
+        schedule_config['mode'] = mode
 
         if mode == 'global':
             # Simple global schedule