fix(display): snapshot the plugin mappings under their locks

Addresses the review finding on the cross-thread reads.

_enabled_plugin_not_running runs on the config-watcher thread and read two
mappings the render thread mutates. Catching RuntimeError was not a fix: it
turned a torn read into a coin flip between an unnecessary discovery scan and
a missed retry, which is the bug this PR exists to remove.

Both reads are now snapshots taken under the lock that guards their writes:

- plugin_manifests via a new PluginManager.discovered_plugin_ids(), which
  copies the ids while holding the existing _discovery_lock. Discovery
  rebuilds that mapping entry by entry, so an unsynchronised reader can see
  it half-populated.
- plugin_display_modes under a new controller lock, taken at the only two
  sites that mutate it (_register_loaded_plugin / _unregister_plugin).

The locks are never nested -- each snapshot is taken and released before the
next -- so this cannot deadlock against discovery, which holds _discovery_lock
while it rebuilds.

No cost on the per-frame path. Both mutation sites run during reconcile, which
is rare, and every hot-path read of plugin_display_modes is on the render
thread itself, same thread as the writes, so those stay lock-free.

Tests: the accessor returns a snapshot rather than a live view, and actually
takes the discovery lock (proved from a second thread, since an RLock is
reentrant on the owning one) so a later refactor cannot quietly drop it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01STMbQE4YctTacQXfbYqKuW

src/display_controller.py

@@ -181,6 +181,12 @@ class DisplayController:
         self.plugin_modes = {}  # mode -> plugin_instance mapping for plugin-first dispatch
         self.mode_to_plugin_id: Dict[str, str] = {}
         self.plugin_display_modes: Dict[str, List[str]] = {}
+        # plugin_display_modes is mutated only by _register_loaded_plugin /
+        # _unregister_plugin on the render thread, but the config-watcher
+        # thread reads it in _enabled_plugin_not_running. Both mutation sites
+        # run during reconcile (rare), so this lock never touches the per-frame
+        # path -- the hot-path reads are same-thread as the writes.
+        self._plugin_modes_lock = threading.Lock()
         # Per-plugin config-change callbacks, kept so we can unsubscribe a
         # plugin when it is disabled live.
         self._plugin_config_callbacks: Dict[str, Callable] = {}
@@ -2814,7 +2820,8 @@ class DisplayController:
             logger.debug("Using manifest display_modes for %s: %s", plugin_id, display_modes)
         if not (isinstance(display_modes, list) and display_modes):
             display_modes = [plugin_id]
-        self.plugin_display_modes[plugin_id] = list(display_modes)
+        with self._plugin_modes_lock:
+            self.plugin_display_modes[plugin_id] = list(display_modes)
 
         # Subscribe to config changes for per-plugin hot-reload. Bind plugin_id
         # and instance as defaults so each plugin's callback targets its own
@@ -2848,7 +2855,8 @@ class DisplayController:
     def _unregister_plugin(self, plugin_id: str) -> None:
         """Remove a plugin's modes, config subscription and instance, then
         unload it. Used by live disable hot-reload."""
-        modes = self.plugin_display_modes.pop(plugin_id, [])
+        with self._plugin_modes_lock:
+            modes = self.plugin_display_modes.pop(plugin_id, [])
         for mode in modes:
             if mode in self.available_modes:
                 self.available_modes.remove(mode)
@@ -2912,16 +2920,22 @@ class DisplayController:
         this is False and costs nothing. That matters because reconcile runs
         ``discover_plugins()`` on the render thread, where a needless
         filesystem scan per config save would show up as a frame hitch.
+
+        Runs on the config-watcher thread, so both mappings it reads are
+        snapshotted under the lock that guards their writes.
         """
         if self.plugin_manager is None:
             return False
+        # Two snapshots, each taken under its own lock and never nested, so a
+        # half-written mapping is never observed and this can't deadlock
+        # against discovery (which holds the discovery lock while rebuilding).
         try:
-            known = set(self.plugin_manager.plugin_manifests)
+            known = self.plugin_manager.discovered_plugin_ids()
+        except AttributeError:
+            # Older manager without the accessor: fall back to a plain read.
+            known = set(getattr(self.plugin_manager, 'plugin_manifests', ()) or ())
+        with self._plugin_modes_lock:
             running = set(self.plugin_display_modes)
-        except (RuntimeError, AttributeError):
-            # Mid-mutation on the render thread, or a manager without the
-            # attribute. Let reconcile decide -- it no-ops when nothing differs.
-            return True
         for key, value in new_config.items():
             if (key in known and isinstance(value, dict)
                     and value.get('enabled', False) and key not in running):

src/plugin_system/plugin_manager.py

@@ -631,6 +631,17 @@ class PluginManager:
         
         return self.load_plugin(plugin_id)
     
+    def discovered_plugin_ids(self) -> set:
+        """Snapshot of the discovered plugin ids, taken under the discovery lock.
+
+        Callers on other threads (the config watcher) must not iterate
+        ``plugin_manifests`` directly: discovery rebuilds it entry by entry, so
+        an unsynchronised reader can see a half-populated mapping or raise
+        "dictionary changed size during iteration".
+        """
+        with self._discovery_lock:
+            return set(self.plugin_manifests)
+
     def get_plugin(self, plugin_id: str) -> Optional[Any]:
         """
         Get a loaded plugin instance by ID.

test/test_display_controller_plugin_toggle.py

@@ -274,6 +274,7 @@ class TestEnabledPluginNotRunning:
     def test_queues_reconcile_when_enabled_plugin_is_absent(self, test_display_controller):
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {}  # failed to load
         cfg = {"hockey-scoreboard": {"enabled": True, "nhl": {"enabled": True}}}
         assert controller._enabled_plugin_not_running(cfg) is True
@@ -281,6 +282,7 @@ class TestEnabledPluginNotRunning:
     def test_quiet_when_every_enabled_plugin_is_running(self, test_display_controller):
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
         cfg = {"hockey-scoreboard": {"enabled": True}}
         assert controller._enabled_plugin_not_running(cfg) is False
@@ -288,6 +290,7 @@ class TestEnabledPluginNotRunning:
     def test_disabled_plugin_does_not_queue(self, test_display_controller):
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {}
         cfg = {"hockey-scoreboard": {"enabled": False}}
         assert controller._enabled_plugin_not_running(cfg) is False
@@ -298,6 +301,7 @@ class TestEnabledPluginNotRunning:
         a reconcile, and therefore a filesystem scan, on every config save."""
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
         cfg = {
             "hockey-scoreboard": {"enabled": True},
@@ -309,6 +313,7 @@ class TestEnabledPluginNotRunning:
     def test_non_dict_section_is_ignored(self, test_display_controller):
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {}
         assert controller._enabled_plugin_not_running({"hockey-scoreboard": "nonsense"}) is False
 
@@ -347,6 +352,7 @@ class TestReconcileQueuedThroughSubscriber:
     def test_nested_edit_queues_reconcile_for_absent_plugin(self, test_display_controller):
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {}          # validate_config() said False
         controller._pending_plugin_reconcile = False
 
@@ -367,6 +373,7 @@ class TestReconcileQueuedThroughSubscriber:
         render thread."""
         controller = test_display_controller
         controller.plugin_manager.plugin_manifests = {"hockey-scoreboard": {}}
+        controller.plugin_manager.discovered_plugin_ids.return_value = {"hockey-scoreboard"}
         controller.plugin_display_modes = {"hockey-scoreboard": ["nhl"]}
         controller._pending_plugin_reconcile = False
 

test/test_plugin_manager_discovered_ids.py

@@ -0,0 +1,63 @@
+"""Tests for PluginManager.discovered_plugin_ids().
+
+The config-watcher thread needs the set of discovered plugin ids while the
+render thread may be rebuilding plugin_manifests. Iterating that dict directly
+can observe a half-populated mapping or raise "dictionary changed size during
+iteration", so the accessor snapshots it under the discovery lock.
+"""
+
+import tempfile
+import threading
+from pathlib import Path
+
+import pytest
+
+from src.plugin_system.plugin_manager import PluginManager
+
+
+@pytest.fixture
+def pm():
+    with tempfile.TemporaryDirectory() as tmp:
+        yield PluginManager(plugins_dir=str(Path(tmp) / "plugins"))
+
+
+def test_returns_the_discovered_ids(pm):
+    pm.plugin_manifests = {"clock-simple": {}, "hockey-scoreboard": {}}
+    assert pm.discovered_plugin_ids() == {"clock-simple", "hockey-scoreboard"}
+
+
+def test_empty_when_nothing_discovered(pm):
+    pm.plugin_manifests = {}
+    assert pm.discovered_plugin_ids() == set()
+
+
+def test_is_a_snapshot_not_a_live_view(pm):
+    """The caller iterates the result on another thread; it must not alias
+    the mapping discovery is still writing to."""
+    pm.plugin_manifests = {"clock-simple": {}}
+    snapshot = pm.discovered_plugin_ids()
+    pm.plugin_manifests["hockey-scoreboard"] = {}
+    assert snapshot == {"clock-simple"}
+
+
+def test_takes_the_discovery_lock(pm):
+    """Guards against the lock being dropped in a later refactor: with the
+    lock held by another thread the call must block rather than read."""
+    pm.plugin_manifests = {"clock-simple": {}}
+    finished = threading.Event()
+
+    def call():
+        pm.discovered_plugin_ids()
+        finished.set()
+
+    pm._discovery_lock.acquire()
+    try:
+        # RLock is reentrant per-thread, so use a *different* thread to prove
+        # the accessor actually waits on it.
+        t = threading.Thread(target=call, daemon=True)
+        t.start()
+        assert not finished.wait(timeout=0.3), "accessor did not take the discovery lock"
+    finally:
+        pm._discovery_lock.release()
+    t.join(timeout=2)
+    assert finished.is_set()