fix(plugins): build get_state_info() as one locked snapshot

Every field was read under its own lock, so an unload running concurrently
could be observed half-done: 'state' read before clear_state() removed it
and 'state_history_count' read after, handing PluginManager.get_plugin_info()
a plugin that is ENABLED with zero transitions.

The whole payload is now built in one critical section. _lock is an RLock,
so the helpers called inside it can still take it.

The regression test runs a reader against a thread that repeatedly fills and
clears the same plugin, and fails on the first torn snapshot. Verified by
removing only the lock: fails on 3 of 3 runs, passes on 3 of 3 with it.

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

src/plugin_system/plugin_state.py

@@ -306,19 +306,26 @@ class PluginStateManager:
         Returns:
             Dictionary with state information
         """
-        state = self.get_state(plugin_id)
-        info = {
-            'state': state.value,
-            'is_loaded': self.is_loaded(plugin_id),
-            'is_enabled': self.is_enabled(plugin_id),
-            'is_running': self.is_running(plugin_id),
-            'is_error': self.is_error(plugin_id),
-            'can_execute': self.can_execute(plugin_id),
-            'last_update': self.get_last_update(plugin_id),
-            'last_display': self.get_last_display(plugin_id),
-            'error_info': self.get_error_info(plugin_id),
-            'state_history_count': self._state_transition_counts.get(plugin_id, 0)
-        }
+        # One snapshot, one critical section. Each field was read under its own
+        # lock, so an unload running concurrently could be observed half-done:
+        # 'state' read before clear_state() removed it and
+        # 'state_history_count' read after, giving a caller a plugin that is
+        # ENABLED with zero transitions. _lock is an RLock, so the helpers
+        # below can still take it.
+        with self._lock:
+            state = self.get_state(plugin_id)
+            info = {
+                'state': state.value,
+                'is_loaded': self.is_loaded(plugin_id),
+                'is_enabled': self.is_enabled(plugin_id),
+                'is_running': self.is_running(plugin_id),
+                'is_error': self.is_error(plugin_id),
+                'can_execute': self.can_execute(plugin_id),
+                'last_update': self.get_last_update(plugin_id),
+                'last_display': self.get_last_display(plugin_id),
+                'error_info': self.get_error_info(plugin_id),
+                'state_history_count': self._state_transition_counts.get(plugin_id, 0)
+            }
         return info
     
     def clear_state(self, plugin_id: str) -> None:

test/test_plugin_state_history_retention.py

@@ -164,3 +164,46 @@ def test_a_monotonic_clock_is_used_not_the_wall_clock(clock):
 
     assert len(m.get_state_history("clock")) == before + 1, (
         "a wall-clock jump must not trim anything")
+
+
+def test_get_state_info_is_a_consistent_snapshot():
+    """An unload running concurrently must not be observed half-done.
+
+    Each field used to be read under its own lock, so clear_state() could
+    interleave: 'state' read before the removal, 'state_history_count' after,
+    handing a caller a plugin that is ENABLED with zero transitions. The whole
+    payload is now built in one critical section.
+    """
+    import threading
+
+    m = PluginStateManager()
+    for _ in range(50):
+        m.set_state("clock", PluginState.RUNNING)
+        m.set_state("clock", PluginState.ENABLED)
+
+    inconsistent = []
+    stop = threading.Event()
+
+    def reader():
+        while not stop.is_set():
+            info = m.get_state_info("clock")
+            # Either fully present or fully cleared -- never a live state with
+            # a wiped count.
+            if info["state"] != PluginState.UNLOADED.value and \
+                    info["state_history_count"] == 0:
+                inconsistent.append(info)
+                return
+
+    def clearer():
+        for _ in range(200):
+            for _ in range(20):
+                m.set_state("clock", PluginState.ENABLED)
+            m.clear_state("clock")
+
+    t = threading.Thread(target=reader, daemon=True)
+    t.start()
+    clearer()
+    stop.set()
+    t.join(timeout=5)
+
+    assert not inconsistent, f"observed a torn snapshot: {inconsistent[:1]}"