fix(plugin-system): unload/update race, failed-load cleanup, limits validation, schema lookup, install rollback (#653)

* fix(plugin-system): unload/update race, failed-load module cleanup, limits validation, schema lookup, install rollback, op-queue dedupe

- unload_plugin takes the per-plugin lock (5s bounded) before cleanup(),
  and an update() that finishes after its plugin was unloaded no longer
  sets the state back to ENABLED.
- A load that fails after import drops plugin_<id> and its submodules
  and forgets its manager fonts, so a fixed plugin reloads new code.
- Resource limits are validated as non-negative numbers: 400 at
  POST /plugins/limits, bad cached records ignored with one warning.
  Route docstrings note health/metrics reset and limits only change the
  web process's view.
- SchemaManager.get_schema_path resolves each search dir via
  resolve_plugin_dir (manifest id, ledmatrix-<id>) before the literal
  paths; plugins/ still before plugin-repos/. Misses cached 30s and
  logged once at DEBUG.
- install_from_url sets an existing copy aside and restores it if the
  move fails, under the per-plugin reinstall lock.
- Operation queue refuses a second pending op for a plugin and trims
  _operations with history.
- get_vegas_render_width reads display_manager.width first.
- get_logger in store/schema/health/resource/saved_repositories;
  UTF-8 reads in store_manager and state_manager.
- Docs: update_interval precedence (manifest over config) stated where
  users are told to set it in config.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix(web): build the limits 400 message from the field name, not an exception

CodeQL flagged str(e) flowing into the response. invalid_limit_field()
returns the offending field without raising, and limits_from_dict uses it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

test/test_async_plugin_updates.py

@@ -254,6 +254,51 @@ class TestFailurePaths:
         assert target_id not in pm._pending_updates
         assert pm.state_manager.get_state(target_id) == PluginState.UNLOADED
 
+    def test_unload_waits_for_in_flight_update(self, pm):
+        """unload_plugin() must not run cleanup() while update() is still
+        executing on the same instance -- it waits on the plugin lock."""
+        observed = {}
+
+        class CleanupPlugin(SlowPlugin):
+            def cleanup(self):
+                observed['in_update_at_cleanup'] = self.in_update
+
+        plugin = CleanupPlugin(update_seconds=0.5)
+        plugin_id = _install(pm, plugin)
+        pm._enqueue_update(plugin_id, time.time())
+        deadline = time.monotonic() + 2
+        while not plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert plugin.in_update
+
+        assert pm.unload_plugin(plugin_id) is True
+        assert observed == {'in_update_at_cleanup': False}
+
+    def test_update_finishing_after_unload_does_not_resurrect(self, pm):
+        """When unload gives up waiting for a hung update(), that update's
+        eventual completion must not flip the cleared state back to ENABLED."""
+        pm.UNLOAD_LOCK_TIMEOUT = 0.05
+        plugin = SlowPlugin(update_seconds=0.5)
+        plugin_id = _install(pm, plugin)
+        pm._enqueue_update(plugin_id, time.time())
+        deadline = time.monotonic() + 2
+        while not plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.01)
+        assert plugin.in_update
+
+        assert pm.unload_plugin(plugin_id) is True
+        deadline = time.monotonic() + 3
+        while plugin.update_calls and plugin.in_update and time.monotonic() < deadline:
+            time.sleep(0.02)
+        time.sleep(0.2)  # let _finish() run
+
+        assert pm.state_manager.get_state(plugin_id) == PluginState.UNLOADED
+        assert plugin_id not in pm.plugin_last_update
+        assert plugin_id not in pm._pending_updates
+        lock = pm.get_plugin_lock(plugin_id)
+        assert lock.acquire(blocking=False) is True
+        lock.release()
+
 
 class TestKillSwitch:
     def test_synchronous_mode_blocks_like_before(self, pm):