diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9395aafc..cfea8ecb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -569,6 +569,46 @@ policies are unchanged.
   scroller or Vegas) were counted as 0.5-1 s freezes and logged as a
   `Render stall ... mid-scroll`. The controller now ends the scroll state
   before drawing either.
+- A plugin that keeps helpers in a package (elections' `providers/`,
+  flights' `enrichment/`, olympics' `data/` and `renderers/`) now runs its
+  updated helpers after a reload. Unloading dropped the package itself but
+  left its modules (`providers.feed`) in `sys.modules`, so the reload after a
+  store update imported the new `manager.py` and got the old helpers back from
+  the cache until the display restarted. `PluginLoader` now drops a plugin's
+  package modules when it unloads, and when a load fails part-way.
+- Uninstalling a dev plugin that `scripts/dev/dev_plugin_setup.sh` linked
+  into the plugins directory now removes the link and leaves the checkout
+  alone. The store's removal passed the link to `shutil.rmtree`, which
+  refuses a symlink; its fallback then walked through the link and chmodded
+  every directory and file of the linked checkout to 0700, and the sudo stage
+  refused a path outside the plugins directory, so the uninstall failed with
+  the link still in place. The same removal discards the set-aside copy after
+  an install or update. A symlink, dangling or not, is now unlinked.
+- A dev plugin linked in under a name its checkout does not share now loads.
+  `dev_plugin_setup.sh link-github foo <url>` clones `ledmatrix-foo` (the
+  repository naming convention) and links it as `plugins/foo`. The loader's
+  containment check for dependency installs resolved the link and looked for
+  `ledmatrix-foo` among the plugins directory's entries, found none, and
+  refused the plugin, so the load failed with "Dependency installation
+  failed" even when it had no `requirements.txt`. The check now looks for the
+  entry the path itself names in the plugins directory, the link, and still
+  only ever answers with an entry it found there.
+- A plugin whose `update()` raises `asyncio.CancelledError` or `SystemExit`
+  no longer goes dark until a restart. Both derive from `BaseException`, not
+  `Exception`, and the update worker's bookkeeping caught only `Exception`:
+  the plugin kept its lock and stayed RUNNING, so it was never updated again
+  and every `display()` was skipped as busy. It is now recorded as that
+  update's failure, the same as any other raise. The plugin executor
+  reported such a call as a timeout; it now reports it as a failure.
+- Saving a config change no longer freezes the panel while a plugin is busy.
+  `ConfigService` told its subscribers about a change while holding its lock,
+  and the display's per-plugin subscriber waits up to 5 s for a plugin in the
+  middle of an update. A save that enables or disables a plugin also queues a
+  reconcile, which the render thread runs, and its `get_config()` and
+  `unsubscribe()` waited behind every one of those callbacks. Subscribers now
+  run after the lock is released. One reload's notifications still finish
+  before the next one's start, and a callback `unsubscribe()` removed is not
+  running, and will not run, once it returns.
 - A plugin whose `display()` raises now opens its circuit breaker. The first
   frame of each screen goes through the plugin executor, which caught the
   exception and returned False. The display read that as "no content" and
diff --git a/src/config_service.py b/src/config_service.py
index 31822776..050ccc7b 100644
--- a/src/config_service.py
+++ b/src/config_service.py
@@ -14,7 +14,7 @@ import json
 import time
 import threading
 from pathlib import Path
-from typing import Dict, Any, Optional, List, Callable
+from typing import Dict, Any, Optional, List, Callable, Tuple
 from collections import defaultdict
 import logging
 import hashlib
@@ -52,7 +52,18 @@ class ConfigService:
         
         # Thread safety
         self._lock: threading.RLock = threading.RLock()
-        
+        # Held across a whole reload -- read, swap, notify -- so one reload's
+        # notifications finish before the next one's start. Subscribers run
+        # under this lock and never under _lock: the display's per-plugin
+        # subscriber can wait seconds for a busy plugin, and get_config(),
+        # subscribe() and unsubscribe() -- called from the render thread --
+        # must not wait behind it.
+        self._notify_lock: threading.RLock = threading.RLock()
+        # (key, callback, thread id) of the callback a notification is running,
+        # so unsubscribe() can wait for that one call; signalled on its return.
+        self._running_callback: Optional[Tuple[str, Callable[..., None], int]] = None
+        self._callback_done = threading.Condition(self._lock)
+
         # Current configuration
         self._current_config: Dict[str, Any] = {}
         self._current_checksum: Optional[str] = None
@@ -87,32 +98,33 @@ class ConfigService:
             True if config changed, False otherwise
         """
         try:
-            new_config = self.config_manager.load_config()
-            new_checksum = self._calculate_checksum(new_config)
-            
-            with self._lock:
-                # Check if config actually changed
-                if new_checksum == self._current_checksum:
-                    self.logger.debug("Configuration unchanged, skipping reload")
-                    return False
-                
-                # Store old config for change detection
-                old_config = self._current_config.copy()
-                
-                # Update current config
-                self._current_config = new_config
-                self._current_checksum = new_checksum
-                
-                # Notify subscribers
+            with self._notify_lock:
+                new_config = self.config_manager.load_config()
+                new_checksum = self._calculate_checksum(new_config)
+
+                with self._lock:
+                    # Check if config actually changed
+                    if new_checksum == self._current_checksum:
+                        self.logger.debug("Configuration unchanged, skipping reload")
+                        return False
+
+                    # Store old config for change detection
+                    old_config = self._current_config.copy()
+
+                    # Update current config
+                    self._current_config = new_config
+                    self._current_checksum = new_checksum
+
+                # Notify subscribers, outside _lock (see _notify_lock)
                 self._notify_subscribers(old_config, new_config)
-                
+
                 self.logger.info(
                     "Configuration reloaded (checksum: %s)",
                     new_checksum[:8]
                 )
-                
+
                 return True
-                
+
         except ConfigError as e:
             self.logger.error("Error loading configuration: %s", e, exc_info=True)
             return False
@@ -127,35 +139,64 @@ class ConfigService:
         Args:
             old_config: Previous configuration
             new_config: New configuration
+
+        Called without _lock held. The subscriber lists are copied under it,
+        and each callback is checked against them again just before it runs.
         """
+        with self._lock:
+            subscribers = {key: list(callbacks) for key, callbacks in self._subscribers.items()}
+
         # Notify global subscribers (key: '*')
-        for callback in self._subscribers.get('*', []):
-            try:
-                callback(old_config, new_config)
-            except Exception as e:
-                self.logger.error("Error in global config change callback: %s", e, exc_info=True)
-        
+        for callback in subscribers.get('*', []):
+            self._call_subscriber('*', callback, old_config, new_config)
+
         # Notify plugin-specific subscribers
-        for plugin_id in self._subscribers.keys():
+        for plugin_id, callbacks in subscribers.items():
             if plugin_id == '*':
                 continue
-            
+
             old_plugin_config = old_config.get(plugin_id, {})
             new_plugin_config = new_config.get(plugin_id, {})
-            
+
             # Only notify if plugin config actually changed
             if old_plugin_config != new_plugin_config:
-                for callback in self._subscribers[plugin_id]:
-                    try:
-                        callback(old_plugin_config, new_plugin_config)
-                    except Exception as e:
-                        self.logger.error(
-                            "Error in config change callback for %s: %s",
-                            plugin_id,
-                            e,
-                            exc_info=True
-                        )
-    
+                for callback in callbacks:
+                    self._call_subscriber(plugin_id, callback,
+                                          old_plugin_config, new_plugin_config)
+
+    def _call_subscriber(
+        self,
+        key: str,
+        callback: Callable[[Dict[str, Any], Dict[str, Any]], None],
+        old_config: Dict[str, Any],
+        new_config: Dict[str, Any],
+    ) -> None:
+        """Run one callback, unless it was unsubscribed since the snapshot.
+
+        unsubscribe() promises that once it returns the callback is neither
+        running nor will run: the display unloads the plugin straight after.
+        """
+        with self._lock:
+            if callback not in self._subscribers.get(key, ()):
+                return
+            self._running_callback = (key, callback, threading.get_ident())
+        try:
+            callback(old_config, new_config)
+        except Exception as e:
+            if key == '*':
+                self.logger.error("Error in global config change callback: %s", e, exc_info=True)
+            else:
+                self.logger.error(
+                    "Error in config change callback for %s: %s",
+                    key,
+                    e,
+                    exc_info=True
+                )
+        finally:
+            with self._lock:
+                self._running_callback = None
+                self._callback_done.notify_all()
+
     def _check_file_changes(self) -> bool:
         """
         Check if configuration files have been modified.
@@ -276,6 +317,11 @@ class ConfigService:
         """
         Unsubscribe from configuration changes.
         
+        Once this returns the callback is not running and will not be called
+        again. A notification that is running this very callback is waited
+        for (unless the callback is the caller); one running any other
+        callback is not.
+
         Args:
             callback: Callback function to remove
             plugin_id: Optional plugin ID (must match subscription)
@@ -285,6 +331,10 @@ class ConfigService:
             if callback in self._subscribers[key]:
                 self._subscribers[key].remove(callback)
                 self.logger.debug("Unsubscribed from config changes for %s", key)
+            while (self._running_callback is not None
+                   and self._running_callback[:2] == (key, callback)
+                   and self._running_callback[2] != threading.get_ident()):
+                self._callback_done.wait()
     
     def shutdown(self) -> None:
         """Shutdown the configuration service."""
diff --git a/src/plugin_system/plugin_executor.py b/src/plugin_system/plugin_executor.py
index b8cfed15..2d1ca432 100644
--- a/src/plugin_system/plugin_executor.py
+++ b/src/plugin_system/plugin_executor.py
@@ -92,7 +92,10 @@ class PluginExecutor:
                 with plugin_scope(plugin_id):
                     result_container['value'] = operation()
                 result_container['completed'] = True
-            except Exception as e:
+            except BaseException as e:  # pylint: disable=broad-except
+                # asyncio.CancelledError and SystemExit too: uncaught, one
+                # ended this thread with 'completed' unset, and an operation
+                # that failed at once was reported as timing out.
                 result_container['exception'] = e
                 result_container['completed'] = True
         
diff --git a/src/plugin_system/plugin_loader.py b/src/plugin_system/plugin_loader.py
index c98b3256..0fea41b4 100644
--- a/src/plugin_system/plugin_loader.py
+++ b/src/plugin_system/plugin_loader.py
@@ -199,9 +199,22 @@ def contained_plugin_dir(plugin_dir: Path, plugins_dir: Path) -> Optional[str]:
     name that came out of ``os.scandir()`` on the trusted root carries no
     taint, which is a real containment guarantee (and one CodeQL's
     path-injection query can follow), not a string sanitiser.
+
+    The entry looked for is the one ``plugin_dir`` itself names when it sits
+    directly in ``plugins_dir``: for a dev plugin symlinked in under its id,
+    the link's name. Resolving the link first and looking for the target's
+    folder name refused ``plugins/foo -> ~/.ledmatrix-dev-plugins/ledmatrix-foo``
+    (what ``dev_plugin_setup.sh link-github foo <url>`` makes), so the plugin
+    never loaded. Any other path is resolved and matched by its final name,
+    as before.
     """
-    plugin_dir_real = os.path.realpath(str(plugin_dir))
     plugins_dir_real = os.path.realpath(str(plugins_dir))
+    plugin_dir_abs = os.path.abspath(str(plugin_dir))
+    if os.path.realpath(os.path.dirname(plugin_dir_abs)) == plugins_dir_real:
+        matched_name = find_trusted_subdir(plugins_dir_real, os.path.basename(plugin_dir_abs))
+        if matched_name is not None:
+            return os.path.join(plugins_dir_real, matched_name)
+    plugin_dir_real = os.path.realpath(str(plugin_dir))
     matched_name = find_trusted_subdir(plugins_dir_real, os.path.basename(plugin_dir_real))
     if matched_name is None:
         return None
@@ -243,6 +256,10 @@ class PluginLoader:
         self.logger = logger or get_logger(__name__)
         self._loaded_modules: Dict[str, Any] = {}
         self._plugin_module_registry: Dict[str, set] = {}  # Maps plugin_id to set of module names
+        # plugin_id -> {dotted name: module} for the modules of the plugin's
+        # own packages (``providers.feed``). They keep their names while the
+        # plugin runs and are dropped with it; see _iter_plugin_submodules.
+        self._plugin_submodules: Dict[str, Dict[str, Any]] = {}
         # Lock to serialize module loading when plugins share module names
         # (e.g., scroll_display.py, game_renderer.py across sport plugins).
         # During exec_module, bare-name sub-modules temporarily appear in
@@ -449,6 +466,45 @@ class PluginLoader:
                 continue
         return result
 
+    @staticmethod
+    def _iter_plugin_submodules(
+        plugin_dir: Path, before_keys: set
+    ) -> list:
+        """Return dotted-name modules from plugin_dir added after before_keys.
+
+        The modules of a package the plugin ships (``providers.feed`` from
+        ``providers/feed.py``). _iter_plugin_bare_modules skips them, so the
+        bare ``providers`` was namespaced and dropped on unload while
+        ``providers.feed`` stayed in sys.modules: a reload after a store update
+        imported a fresh ``providers`` and then got the old ``feed`` back from
+        the cache, running the new manager.py against the old helpers until the
+        display restarted.
+
+        A module counts when its ``__file__`` -- or, for a namespace package,
+        which has none, every ``__path__`` entry -- is inside plugin_dir, so a
+        library the plugin imports (``requests.adapters``) never does.
+
+        Returns a list of (mod_name, module) tuples.
+        """
+        resolved_dir = plugin_dir.resolve()
+        result = []
+        for key in set(sys.modules.keys()) - before_keys:
+            if "." not in key:
+                continue
+            mod = sys.modules.get(key)
+            if mod is None:
+                continue
+            mod_file = getattr(mod, "__file__", None)
+            locations = [mod_file] if mod_file else list(getattr(mod, "__path__", None) or [])
+            if not locations:
+                continue
+            try:
+                if all(Path(loc).resolve().is_relative_to(resolved_dir) for loc in locations):
+                    result.append((key, mod))
+            except (ValueError, TypeError, OSError):
+                continue
+        return result
+
     def _evict_stale_bare_modules(self, plugin_dir: Path) -> dict:
         """Temporarily remove bare-name sys.modules entries from other plugins.
 
@@ -527,6 +583,13 @@ class PluginLoader:
         # Track for cleanup during unload
         self._plugin_module_registry[plugin_id] = namespaced_names
 
+        # The modules of the plugin's own packages keep their dotted names
+        # while it runs -- as they always have, so the package and its
+        # children stay a matching set in sys.modules -- and are dropped
+        # with the plugin by unregister_plugin_modules().
+        self._plugin_submodules[plugin_id] = dict(
+            self._iter_plugin_submodules(plugin_dir, before_keys))
+
         if namespaced_names:
             self.logger.info(
                 "Namespace-isolated %d module(s) for plugin %s",
@@ -537,10 +600,16 @@ class PluginLoader:
         """Remove namespaced sub-modules and cached module for a plugin from sys.modules.
 
         Called by PluginManager during unload to clean up all module entries
-        that were created when the plugin was loaded.
+        that were created when the plugin was loaded, including the dotted
+        modules of its packages. A dotted name is dropped only while it still
+        holds this plugin's module: the name is not namespaced, so another
+        plugin may have put its own there since.
         """
         for ns_name in self._plugin_module_registry.pop(plugin_id, set()):
             sys.modules.pop(ns_name, None)
+        for name, mod in self._plugin_submodules.pop(plugin_id, {}).items():
+            if sys.modules.get(name) is mod:
+                sys.modules.pop(name, None)
         self._loaded_modules.pop(plugin_id, None)
 
     def load_module(
@@ -646,11 +715,13 @@ class PluginLoader:
                     if evicted_name not in sys.modules:
                         sys.modules[evicted_name] = evicted_mod
                 # Clean up the partially-initialized main module and any
-                # bare-name sub-modules that were added during exec_module
-                # so they don't leak into subsequent plugin loads.
+                # bare-name or package sub-modules that were added during
+                # exec_module so they don't leak into subsequent plugin loads.
                 sys.modules.pop(module_name, None)
                 for key, _ in self._iter_plugin_bare_modules(plugin_dir, before_keys):
                     sys.modules.pop(key, None)
+                for key, _ in self._iter_plugin_submodules(plugin_dir, before_keys):
+                    sys.modules.pop(key, None)
                 raise
 
             self._loaded_modules[plugin_id] = module
diff --git a/src/plugin_system/plugin_manager.py b/src/plugin_system/plugin_manager.py
index 79d689b3..69c9548b 100644
--- a/src/plugin_system/plugin_manager.py
+++ b/src/plugin_system/plugin_manager.py
@@ -1163,7 +1163,7 @@ class PluginManager:
     def _record_update_failure(
         self,
         plugin_id: str,
-        exc: Optional[Exception] = None,
+        exc: Optional[BaseException] = None,
         log: bool = True,
         count_failure: bool = True,
     ) -> None:
@@ -1187,7 +1187,7 @@ class PluginManager:
         """
         failure_time = time.time()
         if exc is not None:
-            err: Exception = exc
+            err: BaseException = exc
             error_type = type(exc).__name__
         else:
             err = Exception(f"Plugin {plugin_id} execution failed (timeout or executor error)")
@@ -1653,7 +1653,7 @@ class PluginManager:
         finish_guard = threading.Lock()
         finished = {'done': False}
 
-        def _finish(success: bool, exc: Optional[Exception] = None) -> None:
+        def _finish(success: bool, exc: Optional[BaseException] = None) -> None:
             with finish_guard:
                 if finished['done']:
                     return
@@ -1727,7 +1727,13 @@ class PluginManager:
                     self.resource_monitor.monitor_call(plugin_id, plugin_instance.update)
                 else:
                     plugin_instance.update()
-            except Exception as exc:
+            except BaseException as exc:  # pylint: disable=broad-except
+                # BaseException, not just Exception: asyncio.CancelledError
+                # and SystemExit derive from it. Either one skipped _finish,
+                # so the plugin kept its lock and stayed RUNNING for good --
+                # never rescheduled, and every display() skipped as busy.
+                # Re-raised for the executor, which reports it as this
+                # update's failure.
                 _finish(False, exc=exc)
                 raise
             else:
diff --git a/src/plugin_system/store_manager.py b/src/plugin_system/store_manager.py
index d7b32e68..80f60f67 100644
--- a/src/plugin_system/store_manager.py
+++ b/src/plugin_system/store_manager.py
@@ -344,12 +344,26 @@ class PluginStoreManager(_RegistryMixin, _InstallMixin, _UpdateMixin):
         2. Fix permissions via os.chmod() then retry (works for same-owner files)
         3. Use sudo rm -rf as last resort (works for root-owned __pycache__, etc.)
 
+        A symlink -- a dev plugin linked in by scripts/dev/dev_plugin_setup.sh
+        -- is removed as a link, before any of that: rmtree refuses one, and
+        stage 2 would walk through it and chmod the developer's checkout.
+
         Args:
             path: Path to directory to remove
 
         Returns:
             True if directory was removed successfully, False otherwise
         """
+        if path.is_symlink():
+            # Checked before exists(), which follows the link: a dangling one
+            # would read as already removed and be left behind.
+            try:
+                path.unlink()
+                return True
+            except OSError as e:
+                self.logger.error(f"Could not remove the symlink {path}: {e}")
+                return False
+
         if not path.exists():
             return True  # Already removed
 
diff --git a/test/test_async_plugin_updates.py b/test/test_async_plugin_updates.py
index b908749f..9a23e5d3 100644
--- a/test/test_async_plugin_updates.py
+++ b/test/test_async_plugin_updates.py
@@ -13,6 +13,7 @@ The invariants that keep this change safe:
    inline path exactly.
 """
 
+import asyncio
 import os
 import sys
 import threading
@@ -209,6 +210,32 @@ class TestFailurePaths:
         assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True
         pm.get_plugin_lock(plugin_id).release()
 
+    @pytest.mark.parametrize("raised", [asyncio.CancelledError, SystemExit])
+    def test_update_raising_a_base_exception_still_releases_the_plugin(self, pm, raised):
+        """asyncio.CancelledError and SystemExit derive from BaseException,
+        not Exception. Raised from update() on the worker, one skipped the
+        bookkeeping entirely: the plugin kept its lock and stayed RUNNING for
+        the life of the process -- never updated again, and every display()
+        skipped as busy."""
+        class CancellingPlugin(SlowPlugin):
+            def update(self):
+                self.update_calls += 1
+                raise raised()
+
+        plugin_id = _install(pm, CancellingPlugin())
+        pm.run_scheduled_updates()
+        deadline = time.monotonic() + 3
+        while pm.plugins[plugin_id].update_calls == 0 and time.monotonic() < deadline:
+            time.sleep(0.05)
+        time.sleep(0.2)
+
+        assert pm.get_plugin_lock(plugin_id).acquire(blocking=False) is True
+        pm.get_plugin_lock(plugin_id).release()
+        assert pm.state_manager.can_execute(plugin_id) is True
+        assert pm.plugin_last_update.get(plugin_id, 0) > 0
+        error = pm.state_manager.get_error_info(plugin_id)
+        assert error is not None and error["error_type"] == raised.__name__
+
     def test_unloaded_while_queued_is_harmless(self, pm):
         """Exercise the public unload_plugin() lifecycle rather than
         deleting pm.plugins directly: queue the target's update behind a
diff --git a/test/test_config_service_notify.py b/test/test_config_service_notify.py
new file mode 100644
index 00000000..cf687520
--- /dev/null
+++ b/test/test_config_service_notify.py
@@ -0,0 +1,199 @@
+"""ConfigService notifies subscribers outside its lock, in order.
+
+Subscribers ran while _load_config held the service's lock. The display's
+per-plugin subscriber is PluginManager.apply_config_change, which waits up to
+PLUGIN_LOCK_TIMEOUT (5 s) for a busy plugin. The same save that toggles a
+plugin's ``enabled`` flags a reconcile, and the render thread runs it: its
+get_config() -- and the unsubscribe() of a plugin it disables -- waited behind
+every slow callback, freezing the panel for up to 5 s per busy plugin.
+
+What callers could rely on before still holds: one reload's notifications
+finish before the next reload's start, and a callback unsubscribe() removed is
+not running, and will not run, once unsubscribe() returns.
+"""
+
+import itertools
+import json
+import os
+import threading
+import time
+
+import pytest
+
+from src.config_manager import ConfigManager
+from src.config_service import ConfigService
+
+SLOW = 2.0  # how long a blocked callback waits before giving up
+
+
+@pytest.fixture
+def service(tmp_path):
+    config_path = tmp_path / "config.json"
+    config_path.write_text(json.dumps({"display": {"brightness": 50},
+                                       "weather": {"enabled": True}}),
+                           encoding="utf-8")
+    manager = ConfigManager(str(config_path), str(tmp_path / "config_secrets.json"))
+    manager.template_path = str(tmp_path / "no-template.json")
+    svc = ConfigService(manager, enable_hot_reload=False)
+    yield svc, config_path
+    svc.shutdown()
+
+
+_saves = itertools.count(1)
+
+
+def _save(config_path, **sections):
+    config = json.loads(config_path.read_text(encoding="utf-8"))
+    config.update(sections)
+    config_path.write_text(json.dumps(config), encoding="utf-8")
+    # ConfigManager re-reads only when (mtime, size) moves. Two quick saves of
+    # the same size can share an mtime tick (about 16 ms on Windows), so step
+    # it forward explicitly.
+    st = config_path.stat()
+    os.utime(config_path, ns=(st.st_atime_ns, st.st_mtime_ns + next(_saves) * 50_000_000))
+
+
+def _reload_in_background(svc):
+    thread = threading.Thread(target=svc._load_config, daemon=True)
+    thread.start()
+    return thread
+
+
+def test_get_config_does_not_wait_for_a_slow_subscriber(service):
+    svc, config_path = service
+    entered, release = threading.Event(), threading.Event()
+
+    def slow(_old, _new):
+        entered.set()
+        release.wait(SLOW)
+
+    svc.subscribe(slow, plugin_id="weather")
+    _save(config_path, weather={"enabled": False})
+    reload = _reload_in_background(svc)
+    assert entered.wait(SLOW)
+
+    start = time.monotonic()
+    config = svc.get_config()
+    waited = time.monotonic() - start
+    release.set()
+    reload.join(SLOW)
+
+    assert waited < 0.5
+    # Swapped before anyone was told: a subscriber that reads it sees the new one.
+    assert config["weather"]["enabled"] is False
+
+
+def test_unsubscribing_another_callback_does_not_wait(service):
+    svc, config_path = service
+    entered, release = threading.Event(), threading.Event()
+
+    def slow(_old, _new):
+        entered.set()
+        release.wait(SLOW)
+
+    def other(_old, _new):
+        pass
+
+    svc.subscribe(slow, plugin_id="weather")
+    svc.subscribe(other, plugin_id="clock")
+    _save(config_path, weather={"enabled": False})
+    reload = _reload_in_background(svc)
+    assert entered.wait(SLOW)
+
+    start = time.monotonic()
+    svc.unsubscribe(other, plugin_id="clock")
+    waited = time.monotonic() - start
+    release.set()
+    reload.join(SLOW)
+
+    assert waited < 0.5
+
+
+def test_a_callback_unsubscribed_mid_notification_is_not_called(service):
+    svc, config_path = service
+    entered, release = threading.Event(), threading.Event()
+    called = []
+
+    def slow_global(_old, _new):  # global subscribers are notified first
+        entered.set()
+        release.wait(SLOW)
+
+    def weather(_old, _new):
+        called.append("weather")
+
+    svc.subscribe(slow_global)
+    svc.subscribe(weather, plugin_id="weather")
+    _save(config_path, weather={"enabled": False})
+    reload = _reload_in_background(svc)
+    assert entered.wait(SLOW)
+
+    svc.unsubscribe(weather, plugin_id="weather")
+    release.set()
+    reload.join(SLOW)
+
+    assert called == []
+
+
+def test_unsubscribe_waits_for_its_own_callback_to_return(service):
+    svc, config_path = service
+    entered, release = threading.Event(), threading.Event()
+    returned = threading.Event()
+
+    def slow(_old, _new):
+        entered.set()
+        release.wait(SLOW)
+        returned.set()
+
+    svc.subscribe(slow, plugin_id="weather")
+    _save(config_path, weather={"enabled": False})
+    reload = _reload_in_background(svc)
+    assert entered.wait(SLOW)
+
+    threading.Timer(0.2, release.set).start()
+    svc.unsubscribe(slow, plugin_id="weather")
+
+    assert returned.is_set()
+    reload.join(SLOW)
+
+
+def test_a_callback_may_read_config_and_unsubscribe_itself(service):
+    svc, config_path = service
+    seen = []
+
+    def once(_old, _new):
+        seen.append(svc.get_config()["weather"]["enabled"])
+        svc.unsubscribe(once, plugin_id="weather")
+
+    svc.subscribe(once, plugin_id="weather")
+    _save(config_path, weather={"enabled": False})
+    reload = _reload_in_background(svc)
+    reload.join(SLOW)
+
+    assert not reload.is_alive()
+    assert seen == [False]
+
+
+def test_two_reloads_notify_in_order(service):
+    svc, config_path = service
+    entered, release = threading.Event(), threading.Event()
+    seen = []
+
+    def record(old, new):
+        seen.append((old["brightness"], new["brightness"]))
+        if len(seen) == 1:
+            entered.set()
+            release.wait(SLOW)
+
+    svc.subscribe(record, plugin_id="display")
+    _save(config_path, display={"brightness": 60})
+    first = _reload_in_background(svc)
+    assert entered.wait(SLOW)
+
+    _save(config_path, display={"brightness": 100})
+    second = _reload_in_background(svc)
+    time.sleep(0.2)
+    release.set()
+    first.join(SLOW)
+    second.join(SLOW)
+
+    assert seen == [(50, 60), (60, 100)]
diff --git a/test/test_plugin_loader_reload_isolation.py b/test/test_plugin_loader_reload_isolation.py
index 014153d1..3c0dbdfe 100644
--- a/test/test_plugin_loader_reload_isolation.py
+++ b/test/test_plugin_loader_reload_isolation.py
@@ -58,3 +58,83 @@ def test_a_reloaded_plugin_still_gets_its_own_bare_module(plugins):
     assert reloaded.WHO == "alpha"
     assert sys.path.index(str(plugins["alpha"])) < sys.path.index(str(plugins["beta"]))
     assert sys.path.count(str(plugins["alpha"])) == 1
+
+
+# -- sub-packages ------------------------------------------------------------
+#
+# A plugin that keeps helpers in a package (``providers/feed.py``, imported as
+# ``from providers.feed import ...``) leaves dotted entries in sys.modules.
+# Only the bare ``providers`` used to be tracked, so ``providers.feed`` outlived
+# the plugin: a reload after a store update re-ran the new manager.py against
+# the old feed.py, until the display restarted. Elections (providers/),
+# flights (enrichment/) and olympics (data/, renderers/) ship packages.
+
+
+@pytest.fixture
+def package_plugin(tmp_path):
+    before_path = list(sys.path)
+    before_modules = set(sys.modules)
+    plugin_dir = tmp_path / "pkgdemo"
+    (plugin_dir / "providers").mkdir(parents=True)
+    (plugin_dir / "providers" / "__init__.py").write_text("", encoding="utf-8")
+    (plugin_dir / "providers" / "feed.py").write_text("VERSION = 'v1'\n", encoding="utf-8")
+    (plugin_dir / "manager.py").write_text(
+        "from providers.feed import VERSION\n", encoding="utf-8")
+    yield plugin_dir
+    sys.path[:] = before_path
+    for key in set(sys.modules) - before_modules:
+        sys.modules.pop(key, None)
+
+
+def test_a_reloaded_plugin_runs_its_updated_subpackage_module(package_plugin):
+    loader = PluginLoader()
+    assert loader.load_module("pkgdemo", package_plugin, "manager.py").VERSION == "v1"
+
+    _unload(loader, "pkgdemo")
+    # The store update: a different size, so no cached bytecode can match.
+    (package_plugin / "providers" / "feed.py").write_text(
+        "VERSION = 'v2 from the update'\n", encoding="utf-8")
+    reloaded = loader.load_module("pkgdemo", package_plugin, "manager.py")
+
+    assert reloaded.VERSION == "v2 from the update"
+
+
+def test_unload_drops_the_plugins_subpackage_modules(package_plugin):
+    loader = PluginLoader()
+    loader.load_module("pkgdemo", package_plugin, "manager.py")
+    # Still importable while the plugin runs, as before.
+    assert "providers.feed" in sys.modules
+
+    _unload(loader, "pkgdemo")
+
+    assert not [k for k in sys.modules if k.startswith("providers")]
+
+
+def test_a_failed_load_leaves_no_subpackage_module_behind(package_plugin):
+    (package_plugin / "manager.py").write_text(
+        "from providers.feed import VERSION\nraise RuntimeError('broken')\n",
+        encoding="utf-8")
+    loader = PluginLoader()
+
+    with pytest.raises(RuntimeError):
+        loader.load_module("pkgdemo", package_plugin, "manager.py")
+
+    assert not [k for k in sys.modules if k.startswith("providers")]
+
+
+def test_unload_leaves_packages_from_outside_the_plugin_alone(package_plugin, tmp_path):
+    # A library the plugin imports is not the plugin's to drop.
+    lib_root = tmp_path / "site"
+    (lib_root / "extlib").mkdir(parents=True)
+    (lib_root / "extlib" / "__init__.py").write_text("", encoding="utf-8")
+    (lib_root / "extlib" / "sub.py").write_text("X = 1\n", encoding="utf-8")
+    sys.path.append(str(lib_root))
+    (package_plugin / "manager.py").write_text(
+        "import extlib.sub\nfrom providers.feed import VERSION\n", encoding="utf-8")
+    loader = PluginLoader()
+    loader.load_module("pkgdemo", package_plugin, "manager.py")
+
+    _unload(loader, "pkgdemo")
+
+    assert "extlib.sub" in sys.modules
+    assert "extlib" in sys.modules
diff --git a/test/test_plugin_loader_symlinked_dir.py b/test/test_plugin_loader_symlinked_dir.py
new file mode 100644
index 00000000..17a36244
--- /dev/null
+++ b/test/test_plugin_loader_symlinked_dir.py
@@ -0,0 +1,81 @@
+"""A dev plugin linked in under a name its checkout does not share still loads.
+
+``scripts/dev/dev_plugin_setup.sh`` links a checkout into the plugins
+directory under the plugin's id: ``link-github foo <url>`` clones
+``ledmatrix-foo`` (the repository naming convention) and links it as
+``plugins/foo``. ``contained_plugin_dir`` resolved the link and looked for the
+*target's* folder name, ``ledmatrix-foo``, among the plugins directory's
+entries. There is none, so ``install_dependencies`` refused the plugin as
+outside the plugins directory and the load failed with "Dependency
+installation failed" -- even with no requirements.txt at all.
+
+The containment it exists for still holds: the answer is always rebuilt from
+an entry enumerated under the plugins directory.
+
+Skipped where this process cannot create a symlink (Windows without the
+privilege).
+"""
+
+import os
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from src.plugin_system.plugin_loader import PluginLoader, contained_plugin_dir
+
+
+def _symlink_or_skip(target, link):
+    try:
+        os.symlink(target, link, target_is_directory=True)
+    except (OSError, NotImplementedError) as e:
+        pytest.skip(f"cannot create a symlink here: {e}")
+
+
+@pytest.fixture
+def linked(tmp_path):
+    checkout = tmp_path / "dev-plugins" / "ledmatrix-foo"
+    checkout.mkdir(parents=True)
+    plugins_dir = tmp_path / "plugins"
+    plugins_dir.mkdir()
+    link = plugins_dir / "foo"
+    _symlink_or_skip(checkout, link)
+    return plugins_dir, link, checkout
+
+
+def test_a_link_resolves_to_its_own_entry_in_the_plugins_dir(linked):
+    plugins_dir, link, _checkout = linked
+
+    assert contained_plugin_dir(link, plugins_dir) == os.path.join(
+        os.path.realpath(plugins_dir), "foo")
+
+
+def test_a_linked_plugin_without_requirements_needs_no_install(linked):
+    plugins_dir, link, _checkout = linked
+
+    with patch("subprocess.run") as pip:
+        assert PluginLoader().install_dependencies(link, "foo", plugins_dir=plugins_dir) is True
+    pip.assert_not_called()
+
+
+@patch("src.plugin_system.plugin_loader.requirements_are_satisfied", return_value=False)
+def test_a_linked_plugins_requirements_are_installed_through_the_link(_satisfied, linked):
+    plugins_dir, link, checkout = linked
+    (checkout / "requirements.txt").write_text("package1==1.0.0\n", encoding="utf-8")
+
+    with patch("subprocess.run", return_value=MagicMock(returncode=0, stderr="")) as pip:
+        assert PluginLoader().install_dependencies(link, "foo", plugins_dir=plugins_dir) is True
+
+    argv = pip.call_args[0][0]
+    assert argv[argv.index("-r") + 1] == os.path.join(
+        os.path.realpath(plugins_dir), "foo", "requirements.txt")
+
+
+def test_a_link_outside_the_plugins_dir_is_still_refused(linked, tmp_path):
+    plugins_dir, _link, checkout = linked
+    elsewhere = tmp_path / "elsewhere"
+    elsewhere.mkdir()
+    stray = elsewhere / "bar"
+    _symlink_or_skip(checkout, stray)
+
+    assert contained_plugin_dir(stray, plugins_dir) is None
+    assert contained_plugin_dir(plugins_dir / ".." / "elsewhere" / "bar", plugins_dir) is None
diff --git a/test/test_plugin_system.py b/test/test_plugin_system.py
index 555f5891..3e4f7e7b 100644
--- a/test/test_plugin_system.py
+++ b/test/test_plugin_system.py
@@ -189,6 +189,24 @@ class TestPluginExecutor:
         
         assert result is False
 
+    def test_a_base_exception_is_a_failure_not_a_timeout(self):
+        """asyncio.CancelledError derives from BaseException. Uncaught on
+        the executor's thread it ended the thread with the call never marked
+        complete, so a call that failed at once was reported, and recorded,
+        as timing out."""
+        import asyncio
+        import pytest
+        from src.exceptions import PluginError
+        from src.plugin_system.plugin_executor import PluginExecutor
+        executor = PluginExecutor(default_timeout=5.0)
+
+        def cancelled():
+            raise asyncio.CancelledError()
+
+        with pytest.raises(PluginError) as raised:
+            executor.execute_with_timeout(cancelled, plugin_id="test_plugin")
+        assert isinstance(raised.value.__cause__, asyncio.CancelledError)
+
 
 class TestPluginHealth:
     """Test plugin health monitoring."""
diff --git a/test/test_store_symlinked_plugin.py b/test/test_store_symlinked_plugin.py
new file mode 100644
index 00000000..6cbbb5d6
--- /dev/null
+++ b/test/test_store_symlinked_plugin.py
@@ -0,0 +1,90 @@
+"""Removing a dev plugin linked into the plugins directory removes the link.
+
+``scripts/dev/dev_plugin_setup.sh`` symlinks a checkout into the plugins
+directory. ``PluginStoreManager._safe_remove_directory`` -- behind uninstall,
+and behind discarding the set-aside copy after an install or update -- handed
+the link to ``shutil.rmtree``, which refuses a symlink. Its fallback then
+walked through the link and chmodded every directory and file of the linked
+checkout to 0700, and the sudo stage refused a path outside the plugins
+directory. So the uninstall failed, the link stayed, and the developer's
+checkout lost its group/other permissions and gained execute bits.
+
+Skipped where this process cannot create a symlink (Windows without the
+privilege).
+"""
+
+import json
+import os
+from unittest.mock import MagicMock
+
+import pytest
+
+from src.plugin_system.store_manager import PluginStoreManager
+
+PLUGIN_ID = "linked-demo"
+
+
+def _symlink_or_skip(target, link):
+    try:
+        os.symlink(target, link, target_is_directory=True)
+    except (OSError, NotImplementedError) as e:
+        pytest.skip(f"cannot create a symlink here: {e}")
+
+
+@pytest.fixture
+def linked(tmp_path):
+    checkout = tmp_path / "dev-plugins" / PLUGIN_ID
+    checkout.mkdir(parents=True)
+    (checkout / "manifest.json").write_text(
+        json.dumps({"id": PLUGIN_ID, "name": "Linked", "class_name": "P",
+                    "display_modes": ["linked"], "version": "1.0.0"}),
+        encoding="utf-8")
+    (checkout / "manager.py").write_text("X = 1\n", encoding="utf-8")
+    plugins_dir = tmp_path / "plugins"
+    plugins_dir.mkdir()
+    link = plugins_dir / PLUGIN_ID
+    _symlink_or_skip(checkout, link)
+    store = PluginStoreManager(plugins_dir=str(plugins_dir))
+    store.logger = MagicMock()
+    return store, link, checkout
+
+
+def test_removing_a_linked_plugin_removes_only_the_link(linked):
+    store, link, checkout = linked
+
+    assert store._safe_remove_directory(link) is True
+
+    assert not os.path.lexists(link)
+    assert (checkout / "manager.py").read_text(encoding="utf-8") == "X = 1\n"
+
+
+@pytest.mark.skipif(os.name != "posix", reason="POSIX permission bits")
+def test_removing_a_linked_plugin_leaves_the_checkouts_permissions(linked):
+    store, link, checkout = linked
+    os.chmod(checkout, 0o755)
+    os.chmod(checkout / "manager.py", 0o644)
+
+    store._safe_remove_directory(link)
+
+    assert checkout.stat().st_mode & 0o777 == 0o755
+    assert (checkout / "manager.py").stat().st_mode & 0o777 == 0o644
+
+
+def test_uninstalling_a_linked_plugin_removes_the_link(linked):
+    store, link, checkout = linked
+
+    assert store.uninstall_plugin(PLUGIN_ID) is True
+
+    assert not os.path.lexists(link)
+    assert (checkout / "manifest.json").exists()
+
+
+def test_a_dangling_link_is_removed_too(linked):
+    store, link, checkout = linked
+    for child in checkout.iterdir():
+        child.unlink()
+    checkout.rmdir()
+
+    assert store._safe_remove_directory(link) is True
+
+    assert not os.path.lexists(link)