Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion astrbot/core/star/star_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -1013,7 +1013,7 @@ async def reload(self, specified_plugin_name=None):
logger.warning(
f"插件 {smd.name} 未被正常终止: {e!s}, 可能会导致该插件运行不正常。",
)
if smd.name:
if smd.name and smd.activated:
await self._unbind_plugin(smd.name, specified_module_path)
Comment on lines +1016 to 1017

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

critical

在单插件重载的情况下,如果该插件之前已被停用,跳过 _unbind_plugin 会导致现有的 StarMetadata 对象被保留并复用。

然而,在 load() 方法中,只有当插件在 inactivated_plugins 中时才会将其 metadata.activated 设为 False,但从未在启用时将其设为 True。这意味着,当用户重新启用该插件(turn_on_plugin)并触发重载时,虽然插件类会被实例化,但其 activated 状态仍将保持为 False。这会导致该插件的所有事件处理器在 get_handlers_by_event_type 中被过滤掉,从而使插件完全无法工作。

建议在跳过 _unbind_plugin 的同时,检查该插件是否已被移出 inactivated_plugins,如果是,则将其 activated 状态恢复为 True

Suggested change
if smd.name and smd.activated:
await self._unbind_plugin(smd.name, specified_module_path)
if smd.name and smd.activated:
await self._unbind_plugin(smd.name, specified_module_path)
elif smd.name:
inactivated_plugins = await sp.global_get("inactivated_plugins", [])
if specified_module_path not in inactivated_plugins:
smd.activated = True

@irmia2026 irmia2026 Jun 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch on the surface, but the underlying mechanism already handles this correctly. Let me explain:

load() always creates a fresh StarMetadata and overwrites star_map.

At L1294: star_map[path] = metadata — this unconditionally overwrites whatever was in star_map with the newly created metadata object.

At L1299: if metadata.module_path in inactivated_plugins: metadata.activated = False — it only flips to False for inactivated plugins.

The key insight: StarMetadata.activated defaults to True (defined at star.py:51: activated: bool = True). So when load() creates a fresh metadata and the plugin is NOT in inactivated_plugins (i.e., after turn_on_plugin), activated stays True automatically.

Scenario trace:

  1. Plugin is inactivated → reload() with our fix skips _unbind_pluginload() creates fresh metadata → activated = False
  2. User calls turn_on_plugin → plugin removed from inactivated_pluginsreload() skips _unbind_pluginload() creates fresh new metadata → activated = True ✅ (not in inactivated_plugins)

The old metadata's activated=False state is irrelevant because star_map[path] = metadata overwrites it every time.


result = await self.load(specified_module_path)
Expand Down
199 changes: 199 additions & 0 deletions tests/test_plugin_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -1976,3 +1976,202 @@ async def mock_cleanup(

assert len(cleanup_calls) == 1
assert cleanup_calls[0]["plugin_id"] is None


# --- reload + deactivated plugin regression tests ---


@pytest.mark.asyncio
async def test_reload_deactivated_plugin_preserves_tools(
plugin_manager_pm: PluginManager, monkeypatch
):
"""Specified reload of a deactivated plugin keeps its tools in func_list."""
_clear_star_runtime_state()
plugin_name = "demo_plugin"
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,
activated=False,
)
star_manager_module.star_map[module_path] = metadata
star_manager_module.star_registry.append(metadata)

plugin_tool = star_manager_module.FunctionTool(
name="plugin_search",
description="plugin search",
parameters={"type": "object", "properties": {}},
handler_module_path=f"data.plugins.{plugin_name}.main.tools.search",
)
llm_tools = cast(Any, star_manager_module.llm_tools)
original_func_list = llm_tools.func_list
llm_tools.func_list = [plugin_tool]

async def mock_terminate(smd):
pass # deactivated → no-op

async def mock_load(specified_module_path=None, **kwargs):
return True, None

monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)

try:
await plugin_manager_pm.reload(plugin_name)
assert plugin_tool in llm_tools.func_list
finally:
llm_tools.func_list = original_func_list
_clear_star_runtime_state()


@pytest.mark.asyncio
async def test_reload_activated_plugin_still_unbinds(
plugin_manager_pm: PluginManager, monkeypatch
):
"""Specified reload of an activated plugin still calls _unbind_plugin."""
_clear_star_runtime_state()
plugin_name = "demo_plugin"
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,
activated=True,
)
star_manager_module.star_map[module_path] = metadata
star_manager_module.star_registry.append(metadata)

unbound = []

async def mock_terminate(smd):
pass

async def mock_unbind(name, path):
unbound.append(name)

async def mock_load(specified_module_path=None, **kwargs):
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(plugin_name)
assert unbound == [plugin_name]
finally:
_clear_star_runtime_state()


@pytest.mark.asyncio
async def test_full_reload_deactivated_plugin_stays_registered(
plugin_manager_pm: PluginManager, monkeypatch
):
"""Full reload keeps deactivated plugin in star_map with activated=False."""
_clear_star_runtime_state()
plugin_name = "demo_plugin"
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,
activated=False,
)
star_manager_module.star_map[module_path] = metadata
star_manager_module.star_registry.append(metadata)

async def mock_terminate(smd):
pass

async def mock_unbind_full(name, path):
pass

async def mock_load(specified_module_path=None, **kwargs):
# In full reload, load() re-registers all plugins.
# Deactivated plugins get registered with activated=False.
re_registered = star_manager_module.StarMetadata(
name=plugin_name,
root_dir_name=plugin_name,
module_path=module_path,
activated=False,
)
star_manager_module.star_map[module_path] = re_registered
star_manager_module.star_registry.append(re_registered)
return True, None

monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
monkeypatch.setattr(plugin_manager_pm, "_unbind_plugin", mock_unbind_full)
monkeypatch.setattr(plugin_manager_pm, "load", mock_load)

try:
await plugin_manager_pm.reload()
assert module_path in star_manager_module.star_map
assert star_manager_module.star_map[module_path].activated is False
finally:
_clear_star_runtime_state()


@pytest.mark.asyncio
async def test_turn_on_plugin_after_deactivated_reload_reactivates_tools(
plugin_manager_pm: PluginManager, monkeypatch
):
"""turn_on_plugin reactivates tools after a deactivated plugin is reloaded."""
_clear_star_runtime_state()
plugin_name = "demo_plugin"
module_path = f"data.plugins.{plugin_name}.main"
plugin = star_manager_module.StarMetadata(
name=plugin_name,
root_dir_name=plugin_name,
module_path=module_path,
activated=False,
)
cast(Any, plugin_manager_pm.context).stars.append(plugin)
star_manager_module.star_map[module_path] = plugin
star_manager_module.star_registry.append(plugin)

plugin_tool = star_manager_module.FunctionTool(
name="plugin_search",
description="plugin search",
parameters={"type": "object", "properties": {}},
handler_module_path=f"data.plugins.{plugin_name}.main.tools.search",
)
plugin_tool.active = False # simulate deactivated state
llm_tools = cast(Any, star_manager_module.llm_tools)
original_func_list = llm_tools.func_list
llm_tools.func_list = [plugin_tool]
preferences = {
"inactivated_plugins": [module_path],
"inactivated_llm_tools": [],
}

async def mock_global_get(key, default=None):
return preferences.get(key, default)

async def mock_global_put(key, value):
preferences[key] = value

async def mock_terminate(smd):
pass

async def mock_reload(plugin_name_arg):
assert plugin_name_arg == plugin_name
# Simulate what load() does: re-register with activated=True
# since it's no longer in inactivated_plugins
plugin.activated = True
return True, None

monkeypatch.setattr(star_manager_module.sp, "global_get", mock_global_get)
monkeypatch.setattr(star_manager_module.sp, "global_put", mock_global_put)
monkeypatch.setattr(plugin_manager_pm, "_terminate_plugin", mock_terminate)
monkeypatch.setattr(plugin_manager_pm, "reload", mock_reload)

try:
await plugin_manager_pm.turn_on_plugin(plugin_name)
assert plugin_tool.active is True
assert module_path not in preferences["inactivated_plugins"]
assert plugin.activated is True
finally:
llm_tools.func_list = original_func_list
cast(Any, plugin_manager_pm.context).stars.remove(plugin)
_clear_star_runtime_state()
Loading