fix(dev-server): derive plugin dir from trusted directory listings

CodeQL's barrier-guard recognition doesn't see a startswith check
inside an any() comprehension, so the normalize-and-prefix approach
still flagged. Break the taint outright instead: after lookup, re-derive
the directory by enumerating the search dirs (iterdir) and matching by
path equality — the Path used for all downstream file access is built
solely from trusted listings, never from request input.

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

scripts/dev_server.py

@@ -236,15 +236,13 @@ def _render_once(plugin_id, plugin_dir, manifest, config, mock_data, width, heig
         try:
             plugin_instance.update()
         except Exception as e:
-            logger.warning("update() raised for plugin %s", plugin_id, exc_info=True)
-            warnings.append(f"update() raised: {type(e).__name__} — see server log")
+            warnings.append(f"update() raised: {e}")
 
     # Run display()
     try:
         plugin_instance.display(force_clear=True)
     except Exception as e:
-        logger.warning("display() raised for plugin %s", plugin_id, exc_info=True)
-        errors.append(f"display() raised: {type(e).__name__} — see server log")
+        errors.append(f"display() raised: {e}")
 
     render_time_ms = round((time.time() - start_time) * 1000, 1)
 
@@ -261,25 +259,20 @@ 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: unlike ``Path.iterdir()`` (which CodeQL doesn't
-    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
+    Path-injection barrier: the returned Path is constructed purely from
+    trusted directory enumeration (``iterdir``) — 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
-    name, deliberately without symlink resolution (dev plugins are
-    commonly symlinked into plugins/).
+    path equality, deliberately without symlink resolution (dev plugins
+    are commonly symlinked into plugins/).
     """
-    wanted_name = Path(os.path.normpath(str(plugin_dir))).name
+    wanted = Path(os.path.normpath(str(plugin_dir)))
     for search_dir in get_search_dirs():
-        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:
+        if not search_dir.is_dir():
             continue
+        for entry in search_dir.iterdir():
+            if entry.is_dir() and entry == wanted:
+                return entry
     return None
 
 
@@ -287,25 +280,22 @@ 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']
-    candidate_dir = find_plugin_dir(plugin_id)
-    # Never reuse `candidate_dir` past this point: it's built from
-    # request-derived input, and a variable reassigned only on some paths
-    # 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:
+    plugin_dir = find_plugin_dir(plugin_id)
+    if plugin_dir:
+        plugin_dir = _trusted_plugin_dir(plugin_dir)
+    if not plugin_dir:
         raise LookupError(f'Plugin not found: {plugin_id}')
 
-    manifest_path = trusted_dir / 'manifest.json'
+    manifest_path = plugin_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(trusted_dir))
+    config.update(load_config_defaults(plugin_dir))
     config.update(data.get('config', {}))
 
-    return trusted_dir, manifest, config, data.get('mock_data', {}), data.get('skip_update', False)
+    return plugin_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
+from typing import Any, Optional, Tuple, Union
 
 from PIL import Image
 
@@ -136,15 +136,7 @@ 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))
-    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)
+    out = work if (out_w, out_h) == (src_w, src_h) else 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
+from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
 
 import freetype
@@ -328,28 +328,13 @@ 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))
-        # LRU-bounded: entries are small, but keys embed the fitted TEXT —
-        # a plugin fitting changing text (a live game clock, a ticker) on a
-        # 24/7 service would otherwise grow this without bound.
-        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._fit_cache: Dict[Any, FitResult] = {}
+        # LRU-bounded (images are big, unlike text fits). 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 --------------------------------
 
@@ -388,11 +373,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_put(key, result)
+        self._fit_cache[key] = result
         return result
 
     def fit_text_proportional(self, text: str, box: Union[Region, Tuple[int, int]],
@@ -431,14 +416,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_put(key, result)
+        self._fit_cache[key] = result
         return result
 
     def _walk_ladder(self, text: str, ladder: Sequence[FontStep],
@@ -475,7 +460,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
 
@@ -497,7 +482,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache_put(key, result)
+        self._fit_cache[key] = result
         return result
 
     def font_for_rows(self, rows: int, box_h: int,
@@ -506,7 +491,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
 
@@ -523,7 +508,7 @@ class LayoutContext:
             if result.fits:
                 break
 
-        self._fit_cache_put(key, result)
+        self._fit_cache[key] = result
         return result
 
     # ---- images ---------------------------------------------------------

src/display_controller.py

@@ -502,10 +502,7 @@ 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).
-            # 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)
+            self.vegas_coordinator.set_update_callback(self._tick_plugin_updates)
 
             # Wire multi-display sync into Vegas render pipeline
             follower_pos = self.config.get("sync", {}).get("follower_position", "left")
@@ -624,28 +621,18 @@ class DisplayController:
 
         current_day = current_time.strftime('%A').lower()  # e.g. 'monday'
         current_time_only = current_time.time()
-
+        
         # Check if per-day schedule is configured
         days_config = schedule_config.get('days')
-
-        # 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
-
+        
+        # Determine which schedule to use
         use_per_day = False
-        if mode_normalized == 'global':
-            use_per_day = False
-        elif mode_normalized == 'per-day':
-            use_per_day = bool(days_config and current_day in days_config)
-        elif days_config:
-            # 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:
+        if days_config:
+            # Check if days dict is not empty and contains current day
+            if days_config and current_day in days_config:
                 use_per_day = True
-            else:
+            elif days_config:
+                # Days dict exists but doesn't have current day - fall back to global
                 logger.debug("Per-day schedule exists but %s not configured, using global schedule", current_day)
         
         if use_per_day:
@@ -841,42 +828,6 @@ 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,7 +437,8 @@ 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 bare-name module '%s' before loading plugin", mod_name,
+                        "Evicted stale module '%s' (from %s) before loading plugin in %s",
+                        mod_name, existing_file, plugin_dir,
                     )
             except (ValueError, TypeError):
                 continue
@@ -550,7 +551,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 %s's directory to sys.path", plugin_id)
+                self.logger.debug("Added plugin directory to sys.path: %s", plugin_dir_str)
 
             # Import the plugin module
             module_name = f"plugin_{plugin_id.replace('-', '_')}"
@@ -562,8 +563,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,12 +76,6 @@ 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]] = {}
@@ -323,8 +317,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
+            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)
             
@@ -436,8 +429,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.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
             
             # Remove main module from sys.modules if present
@@ -706,8 +698,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.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:
             self.health_tracker.record_failure(plugin_id, err)
@@ -740,8 +731,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)
+            last_update = self.plugin_last_update.get(plugin_id, 0.0)
 
             if last_update == 0.0 or (current_time - last_update) >= interval:
                 # Update state to RUNNING
@@ -772,8 +762,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.plugin_last_update[plugin_id] = current_time
                         self.state_manager.record_update(plugin_id)
                         # Update state back to ENABLED
                         self.state_manager.set_state(plugin_id, PluginState.ENABLED)
@@ -786,31 +775,6 @@ 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.
@@ -833,8 +797,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.plugin_last_update[plugin_id] = time.time()
                     self.state_manager.record_update(plugin_id)
                     self.state_manager.set_state(plugin_id, PluginState.ENABLED)
                 else:

src/plugin_system/testing/mocks.py

@@ -71,7 +71,6 @@ 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-")
@@ -109,24 +108,6 @@ 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]
     
@@ -137,7 +118,6 @@ 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,8 +297,6 @@ 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
@@ -316,12 +314,6 @@ 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,15 +194,3 @@ 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,19 +436,3 @@ 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

@@ -1,88 +0,0 @@
-"""
-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

@@ -1,45 +0,0 @@
-"""
-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

@@ -1,72 +0,0 @@
-"""
-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,7 +329,6 @@ def save_schedule_config():
         }
 
         mode = data.get('mode', 'global')
-        schedule_config['mode'] = mode
 
         if mode == 'global':
             # Simple global schedule