fix(vegas): lock plugin_last_update snapshot/diff against concurrent mutation

src/display_controller.py

@@ -831,7 +831,7 @@ def _update_modules(self):
                     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):
+    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.
@@ -843,22 +843,22 @@ def _tick_plugin_updates_for_vegas(self):
         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, "plugin_last_update"):
+        if not self.plugin_manager or not hasattr(self.plugin_manager, "run_scheduled_updates_with_changes"):
             self._tick_plugin_updates()
             return
 
-        old_times = dict(self.plugin_manager.plugin_last_update)
-        self._tick_plugin_updates()
+        updated = self.plugin_manager.run_scheduled_updates_with_changes()
 
         vc = getattr(self, "vegas_coordinator", None)
         if vc is None:
             return
 
-        updated = [
-            plugin_id for plugin_id, new_time in self.plugin_manager.plugin_last_update.items()
-            if new_time > old_times.get(plugin_id, 0.0)
-        ]
         if updated:
             logger.info("Vegas update tick: %d plugin(s) updated: %s", len(updated), updated)
             for plugin_id in updated:

src/plugin_system/plugin_manager.py

@@ -76,6 +76,12 @@ def __init__(self, plugins_dir: str = "plugins",
         # 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,7 +323,8 @@ def load_plugin(self, plugin_id: str) -> bool:
 
             # Store plugin instance
             self.plugins[plugin_id] = plugin_instance
-            self.plugin_last_update[plugin_id] = 0.0
+            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,7 +436,8 @@ def unload_plugin(self, plugin_id: str) -> bool:
 
             # Remove from active plugins
             del self.plugins[plugin_id]
-            self.plugin_last_update.pop(plugin_id, None)
+            with self._plugin_last_update_lock:
+                self.plugin_last_update.pop(plugin_id, None)
             self._update_interval_cache.pop(plugin_id, None)
 
             # Remove main module from sys.modules if present
@@ -698,7 +706,8 @@ def _record_update_failure(
             'recoverable': True,
         }
         self.logger.warning("Plugin %s update() failed; will retry after interval", plugin_id)
-        self.plugin_last_update[plugin_id] = failure_time
+        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:
             self.health_tracker.record_failure(plugin_id, err)
@@ -731,7 +740,8 @@ def run_scheduled_updates(self, current_time: Optional[float] = None) -> None:
             if interval is None:
                 continue
 
-            last_update = self.plugin_last_update.get(plugin_id, 0.0)
+            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:
                 # Update state to RUNNING
@@ -762,7 +772,8 @@ def monitored_update():
                         success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
 
                     if success:
-                        self.plugin_last_update[plugin_id] = current_time
+                        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
                         self.state_manager.set_state(plugin_id, PluginState.ENABLED)
@@ -775,6 +786,31 @@ def monitored_update():
                     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,7 +833,8 @@ def update_all_plugins(self) -> None:
             try:
                 success = self.plugin_executor.execute_update(plugin_instance, plugin_id)
                 if success:
-                    self.plugin_last_update[plugin_id] = time.time()
+                    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)
                 else:

test/test_display_controller_vegas_tick.py

@@ -11,66 +11,78 @@
 callers since.
 """
 
+from typing import Dict, List, Optional
 from unittest.mock import MagicMock
 
 from src.display_controller import DisplayController
 
 
-def _make_controller(plugin_last_update, vegas_coordinator=None):
+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.plugin_last_update = dict(plugin_last_update)
+    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):
-        dc = _make_controller({"stock-news": 100.0, "odds-ticker": 100.0})
         vc = MagicMock()
-        dc.vegas_coordinator = vc
-
-        # Simulate run_scheduled_updates() advancing only stock-news.
-        def fake_tick():
-            dc.plugin_manager.plugin_last_update["stock-news"] = 200.0
-        dc._tick_plugin_updates = fake_tick
+        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):
-        dc = _make_controller({"stock-news": 100.0})
         vc = MagicMock()
-        dc.vegas_coordinator = vc
-        dc._tick_plugin_updates = lambda: None
+        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({"stock-news": 100.0}, vegas_coordinator=None)
-
-        def fake_tick():
-            dc.plugin_manager.plugin_last_update["stock-news"] = 200.0
-        dc._tick_plugin_updates = fake_tick
+        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."""
-        dc = _make_controller({"a": 1.0, "b": 1.0})
         vc = MagicMock()
         vc.mark_plugin_updated.side_effect = [RuntimeError("boom"), None]
-        dc.vegas_coordinator = vc
-
-        def fake_tick():
-            dc.plugin_manager.plugin_last_update["a"] = 2.0
-            dc.plugin_manager.plugin_last_update["b"] = 2.0
-        dc._tick_plugin_updates = fake_tick
+        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)