fix(plugin_manager): improve plugin state cleanup and add tests for unbinding and loading plugins

fixes: #8439

astrbot/core/star/star_manager.py

@@ -654,12 +654,23 @@ class PluginManager:
 
         """
         prefix = "astrbot.builtin_stars." if is_reserved else "data.plugins."
+        module_prefix = f"{prefix}{plugin_root_dir}"
         return [
             key
             for key in list(sys.modules.keys())
-            if key.startswith(f"{prefix}{plugin_root_dir}")
+            if PluginManager._is_plugin_module_path(key, module_prefix)
         ]
 
+    @staticmethod
+    def _is_plugin_module_path(module_path: str | None, module_prefix: str) -> bool:
+        return bool(
+            module_path
+            and (
+                module_path == module_prefix
+                or module_path.startswith(f"{module_prefix}.")
+            )
+        )
+
     def _purge_modules(
         self,
         module_patterns: list[str] | None = None,
@@ -694,33 +705,49 @@ class PluginManager:
                 except KeyError:
                     logger.warning(f"模块 {module_name} 未载入")
 
-    def _cleanup_plugin_state(self, dir_name: str) -> None:
-        plugin_root_name = "data.plugins."
+    def _cleanup_plugin_state(self, dir_name: str, is_reserved: bool = False) -> None:
+        plugin_root_name = "astrbot.builtin_stars." if is_reserved else "data.plugins."
+        module_prefix = f"{plugin_root_name}{dir_name}"
 
         # 清理 sys.modules
         for key in list(sys.modules.keys()):
-            if key.startswith(f"{plugin_root_name}{dir_name}"):
+            if self._is_plugin_module_path(key, module_prefix):
                 logger.info(f"清除了插件{dir_name}中的{key}模块")
                 del sys.modules[key]
 
-        possible_paths = [
-            f"{plugin_root_name}{dir_name}.main",
-            f"{plugin_root_name}{dir_name}.{dir_name}",
-        ]
+        # Clean plugin metadata registered before a failed load completes.
+        for module_path, metadata in list(star_map.items()):
+            if self._is_plugin_module_path(module_path, module_prefix) or (
+                metadata.root_dir_name == dir_name and metadata.reserved == is_reserved
+            ):
+                star_map.pop(module_path, None)
+                if metadata in star_registry:
+                    star_registry.remove(metadata)
+                logger.info(f"清理插件元数据: {module_path}")
+
+        for metadata in list(star_registry):
+            if self._is_plugin_module_path(metadata.module_path, module_prefix) or (
+                metadata.root_dir_name == dir_name and metadata.reserved == is_reserved
+            ):
+                star_registry.remove(metadata)
+                logger.info(f"清理插件注册项: {metadata.name or dir_name}")
 
         # 清理 handlers
-        for path in possible_paths:
-            handlers = star_handlers_registry.get_handlers_by_module_name(path)
-            for handler in handlers:
+        for handler in list(star_handlers_registry):
+            if self._is_plugin_module_path(handler.handler_module_path, module_prefix):
                 star_handlers_registry.remove(handler)
                 logger.info(f"清理处理器: {handler.handler_name}")
 
         # 清理工具
         for tool in list(llm_tools.func_list):
-            if tool.handler_module_path in possible_paths:
+            handler_module_path = getattr(tool, "handler_module_path", None)
+            if self._is_plugin_module_path(handler_module_path, module_prefix):
                 llm_tools.func_list.remove(tool)
                 logger.info(f"清理工具: {tool.name}")
 
+        for adapter_name in unregister_platform_adapters_by_module(module_prefix):
+            logger.info(f"清理平台适配器: {adapter_name}")
+
     def _build_failed_plugin_record(
         self,
         *,
@@ -836,7 +863,7 @@ class PluginManager:
             # 终止插件
             if not specified_module_path:
                 # 重载所有插件
-                for smd in star_registry:
+                for smd in list(star_registry):
                     try:
                         await self._terminate_plugin(smd)
                     except Exception as e:
@@ -948,11 +975,7 @@ class PluginManager:
                             error_trace=error_trace,
                         )
                     )
-                    if path in star_map:
-                        logger.info("失败插件依旧在插件列表中，正在清理...")
-                        metadata = star_map.pop(path)
-                        if metadata in star_registry:
-                            star_registry.remove(metadata)
+                    self._cleanup_plugin_state(root_dir_name, reserved)
                     continue
 
                 # 检查 _conf_schema.json
@@ -1097,21 +1120,29 @@ class PluginManager:
                         f"插件 {path} 未通过装饰器注册。尝试通过旧版本方式载入。",
                     )
                     classes = self._get_classes(module)
+                    if not classes:
+                        raise Exception(
+                            f"插件 {root_dir_name} 未通过 Star 注册，也没有找到旧版插件类。"
+                            "请确认插件主类继承 astrbot.api.star.Star，或类名以 Plugin 结尾 / 命名为 Main。",
+                        )
+
+                    plugin_cls = getattr(module, classes[0])
+                    obj = None
 
                     if path not in inactivated_plugins:
                         # 只有没有禁用插件时才实例化插件类
                         if plugin_config:
                             try:
-                                obj = getattr(module, classes[0])(
+                                obj = plugin_cls(
                                     context=self.context,
                                     config=plugin_config,
                                 )  # 实例化插件类
                             except TypeError as _:
-                                obj = getattr(module, classes[0])(
+                                obj = plugin_cls(
                                     context=self.context,
                                 )  # 实例化插件类
                         else:
-                            obj = getattr(module, classes[0])(
+                            obj = plugin_cls(
                                 context=self.context,
                             )  # 实例化插件类
 
@@ -1139,7 +1170,7 @@ class PluginManager:
                     metadata.module = module
                     metadata.root_dir_name = root_dir_name
                     metadata.reserved = reserved
-                    metadata.star_cls_type = obj.__class__
+                    metadata.star_cls_type = plugin_cls
                     metadata.module_path = path
                     star_map[path] = metadata
                     star_registry.append(metadata)
@@ -1226,12 +1257,7 @@ class PluginManager:
                         error_trace=errors,
                     )
                 )
-                # 记录注册失败的插件名称，以便后续重载插件
-                if path in star_map:
-                    logger.info("失败插件依旧在插件列表中，正在清理...")
-                    metadata = star_map.pop(path)
-                    if metadata in star_registry:
-                        star_registry.remove(metadata)
+                self._cleanup_plugin_state(root_dir_name, reserved)
 
         # 清除 pip.main 导致的多余的 logging handlers
         for handler in logging.root.handlers[:]:

tests/test_plugin_manager.py

@@ -28,9 +28,7 @@ class MockStar:
         self.info = {"repo": TEST_PLUGIN_REPO, "readme": ""}
 
 
-def _write_local_test_plugin(
-    plugin_path: Path, repo_url: str, version: str = "1.0.0"
-):
+def _write_local_test_plugin(plugin_path: Path, repo_url: str, version: str = "1.0.0"):
     """Creates a minimal valid plugin structure."""
     plugin_path.mkdir(parents=True, exist_ok=True)
     metadata = {
@@ -148,11 +146,22 @@ def _clear_module_cache():
     """Clear test-specific modules from sys.modules to allow reloading."""
     import sys
 
-    to_del = [m for m in sys.modules if m.startswith("data.plugins.helloworld")]
+    to_del = [
+        m
+        for m in sys.modules
+        if m.startswith("data.plugins.helloworld")
+        or m.startswith("data.plugins.broken_plugin")
+    ]
     for m in to_del:
         del sys.modules[m]
 
 
+def _clear_star_runtime_state():
+    star_manager_module.star_map.clear()
+    star_manager_module.star_registry.clear()
+    star_manager_module.star_handlers_registry.clear()
+
+
 def _build_load_mock(events):
     async def mock_load(specified_dir_name=None, ignore_version_check=False):
         del ignore_version_check
@@ -467,6 +476,107 @@ async def test_reload_failed_plugin_dependency_install_flow(
         assert events[1] == ("load", TEST_PLUGIN_DIR)
 
 
+@pytest.mark.asyncio
+async def test_reload_all_unbinds_every_registered_plugin(
+    plugin_manager_pm: PluginManager, monkeypatch
+):
+    _clear_star_runtime_state()
+    plugin_names = ["plugin_one", "plugin_two", "plugin_three"]
+    for plugin_name in plugin_names:
+        module_path = f"data.plugins.{plugin_name}.main"
+        metadata = star_manager_module.StarMetadata(
+            name=plugin_name,
+            root_dir_name=plugin_name,
+            module_path=module_path,
+        )
+        star_manager_module.star_map[module_path] = metadata
+        star_manager_module.star_registry.append(metadata)
+
+    terminated = []
+    unbound = []
+
+    async def mock_terminate(plugin):
+        terminated.append(plugin.name)
+
+    async def mock_unbind(plugin_name, plugin_module_path):
+        unbound.append(plugin_name)
+        star_manager_module.star_map.pop(plugin_module_path, None)
+        for index, metadata in enumerate(star_manager_module.star_registry):
+            if metadata.name == plugin_name:
+                del star_manager_module.star_registry[index]
+                break
+
+    async def mock_load(
+        specified_module_path=None,
+        specified_dir_name=None,
+        ignore_version_check=False,
+    ):
+        del specified_module_path, specified_dir_name, ignore_version_check
+        return True, None
+
+    monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
+    monkeypatch.setattr(plugin_manager_pm, "_unbind_plugin", mock_unbind)
+    monkeypatch.setattr(plugin_manager_pm, "load", mock_load)
+
+    try:
+        await plugin_manager_pm.reload()
+    finally:
+        _clear_star_runtime_state()
+
+    assert terminated == plugin_names
+    assert unbound == plugin_names
+
+
+@pytest.mark.asyncio
+async def test_load_reports_unregistered_plugin_without_index_error(
+    plugin_manager_pm: PluginManager, monkeypatch
+):
+    _clear_star_runtime_state()
+    plugin_root = Path(plugin_manager_pm.plugin_store_path).parents[1]
+    plugin_name = "broken_plugin"
+    plugin_path = Path(plugin_manager_pm.plugin_store_path) / plugin_name
+    plugin_path.mkdir(parents=True)
+    (plugin_path / "metadata.yaml").write_text(
+        yaml.dump(
+            {
+                "name": plugin_name,
+                "author": "AstrBot Team",
+                "desc": "Broken test plugin",
+                "version": "1.0.0",
+            }
+        ),
+        encoding="utf-8",
+    )
+    (plugin_path / "main.py").write_text("VALUE = 1\n", encoding="utf-8")
+
+    async def mock_global_get(key, default=None):
+        del key
+        return default
+
+    async def mock_sync_command_configs():
+        return None
+
+    monkeypatch.syspath_prepend(str(plugin_root))
+    monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
+    monkeypatch.setattr(
+        star_manager_module,
+        "sync_command_configs",
+        mock_sync_command_configs,
+    )
+
+    try:
+        success, error = await plugin_manager_pm.load(specified_dir_name=plugin_name)
+    finally:
+        _clear_star_runtime_state()
+        _clear_module_cache()
+
+    assert success is False
+    assert error is not None
+    assert "未通过 Star 注册" in error
+    assert "list index out of range" not in error
+    assert plugin_name in plugin_manager_pm.failed_plugin_dict
+
+
 @pytest.mark.asyncio
 async def test_ensure_plugin_requirements_reraises_cancelled_error(
     plugin_manager_pm: PluginManager, local_updator: Path, monkeypatch