From 595d3746eb836100ea353daf3ed800c140215dbb Mon Sep 17 00:00:00 2001 From: Estrella Pan Date: Thu, 8 Oct 2026 10:51:33 +0200 Subject: [PATCH] =?UTF-8?q?feat(4.0-P4):=20=E6=8F=92=E4=BB=B6=E5=8F=AF?= =?UTF-8?q?=E9=80=9A=E7=9F=A5=E4=BA=8B=E4=BB=B6=E8=BF=9B=E5=85=A5=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E4=B8=AD=E5=BF=83=EF=BC=8C@subscribe=20=E6=94=AF?= =?UTF-8?q?=E6=8C=81=E8=B6=85=E6=97=B6=EF=BC=8C=E6=B8=85=E5=8D=95=E5=A2=9E?= =?UTF-8?q?=E5=8A=A0=20default=5Fenabled?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - PluginBus 收到插件发布的 SystemEvent 时经 NotificationManager.send_event 投递（通知中心 → 事件总线 → 外部渠道） - @subscribe(kind, timeout=...) 覆盖总线默认 30 秒的处理超时 - plugin.toml 的 default_enabled = false 让内置插件默认停用 --- backend/src/ab_sdk/hooks.py | 8 +-- backend/src/module/plugin/bus.py | 9 +++- backend/src/module/plugin/context.py | 19 ++++++- backend/src/module/plugin/manager.py | 12 +++-- backend/src/module/plugin/manifest.py | 3 ++ backend/src/test/test_plugin_runtime.py | 69 +++++++++++++++++++++++-- 6 files changed, 106 insertions(+), 14 deletions(-) diff --git a/backend/src/ab_sdk/hooks.py b/backend/src/ab_sdk/hooks.py index 221c62c4..3216fb49 100644 --- a/backend/src/ab_sdk/hooks.py +++ b/backend/src/ab_sdk/hooks.py @@ -32,6 +32,7 @@ class ProviderSpec: @dataclass(frozen=True, slots=True) class SubscribeSpec: kind: str + timeout: float | None = None @dataclass(frozen=True, slots=True) @@ -83,15 +84,16 @@ def provider(point: str, *, id: str): return mark -def subscribe(kind: str): +def subscribe(kind: str, *, timeout: float | None = None): """把方法声明为事件订阅者，参数为事件对象。 订阅者在独立队列中按发布顺序异步执行，失败或超时不影响发布方。 - ``kind`` 为 ``"*"`` 时接收全部事件。 + ``kind`` 为 ``"*"`` 时接收全部事件。``timeout`` 覆盖宿主默认的单个事件 + 处理超时（秒），适合复制大文件等耗时操作。 """ def mark(func: F) -> F: - setattr(func, SUBSCRIBE_ATTR, SubscribeSpec(kind)) + setattr(func, SUBSCRIBE_ATTR, SubscribeSpec(kind, timeout)) return func return mark diff --git a/backend/src/module/plugin/bus.py b/backend/src/module/plugin/bus.py index 52d8f034..ef956da4 100644 --- a/backend/src/module/plugin/bus.py +++ b/backend/src/module/plugin/bus.py @@ -111,13 +111,18 @@ class EventBus: self.on_error = on_error def subscribe( - self, kind: str, handler: Handler, *, owner: str | None = None + self, + kind: str, + handler: Handler, + *, + owner: str | None = None, + timeout: float | None = None, ) -> Callable[[], None]: sub = _Subscriber( kind, handler, owner, - self._timeout, + timeout if timeout is not None else self._timeout, self._queue_size, self._report_error, ) diff --git a/backend/src/module/plugin/context.py b/backend/src/module/plugin/context.py index 05505862..417e115e 100644 --- a/backend/src/module/plugin/context.py +++ b/backend/src/module/plugin/context.py @@ -1,5 +1,6 @@ """宿主侧的 PluginContext 实现。""" +import asyncio import logging from pathlib import Path from typing import Any @@ -7,11 +8,15 @@ from typing import Any from pydantic import BaseModel from ab_sdk import Event +from ab_sdk.events import SystemEvent from .bus import EventBus, Handler PLUGIN_DATA_ROOT = Path("config") / "plugin-data" +# 插件可通知事件的 fire-and-forget 发送任务，保留强引用防止被 GC +_notify_tasks: set[asyncio.Task] = set() + def _database(): # 延迟 import：module.plugin.host 被网络层、解析器等底层模块引用， @@ -43,7 +48,11 @@ class DatabaseKV: class PluginBus: """插件视角的总线：订阅归属到插件（卸载时统一清理、失败计入熔断）， - 发布的事件 kind 必须以 ``.`` 开头，防止冒充宿主事件。""" + 发布的事件 kind 必须以 ``.`` 开头，防止冒充宿主事件。 + + 插件发布的 :class:`SystemEvent` 与宿主事件走同一条通知路径 + （``NotificationManager.send_event``：通知中心 → 事件总线 → 外部渠道）。 + """ def __init__(self, bus: EventBus, plugin_id: str) -> None: self._bus = bus @@ -55,6 +64,14 @@ class PluginBus: f"插件 {self._plugin_id} 只能发布以 '{self._plugin_id}.' 开头的事件，" f"收到 {event.kind!r}" ) + if isinstance(event, SystemEvent): + # 延迟 import：notification 依赖 module.plugin.host + from module.notification import NotificationManager + + task = asyncio.create_task(NotificationManager().send_event(event)) + _notify_tasks.add(task) + task.add_done_callback(_notify_tasks.discard) + return self._bus.publish(event) def subscribe(self, kind: str, handler: Handler): diff --git a/backend/src/module/plugin/manager.py b/backend/src/module/plugin/manager.py index 8769c067..70a401ee 100644 --- a/backend/src/module/plugin/manager.py +++ b/backend/src/module/plugin/manager.py @@ -194,8 +194,9 @@ class PluginManager: return None def _enabled(self, candidate: PluginCandidate) -> bool: - """启用开关；未设置时内置插件默认启用，其它来源默认禁用。""" - default = candidate.source == "builtin" + """启用开关；未设置时内置插件默认启用（清单 ``default_enabled = false`` + 的除外），其它来源默认禁用。""" + default = candidate.source == "builtin" and candidate.manifest.default_enabled return self._settings.plugins.enabled.get(candidate.manifest.id, default) def _snapshot(self, plugin_id: str) -> str: @@ -279,7 +280,12 @@ class PluginManager: ProviderEntry(plugin_id, provider_spec.id, bound), ) if subscribe_spec is not None: - self.bus.subscribe(subscribe_spec.kind, bound, owner=plugin_id) + self.bus.subscribe( + subscribe_spec.kind, + bound, + owner=plugin_id, + timeout=subscribe_spec.timeout, + ) async def _deactivate(self, plugin_id: str) -> None: active = self._active.pop(plugin_id, None) diff --git a/backend/src/module/plugin/manifest.py b/backend/src/module/plugin/manifest.py index 95130a46..f2ccd336 100644 --- a/backend/src/module/plugin/manifest.py +++ b/backend/src/module/plugin/manifest.py @@ -30,6 +30,9 @@ class PluginManifest(BaseModel): description: str = "" authors: list[str] = Field(default_factory=list) permissions: list[str] = Field(default_factory=list) + default_enabled: bool = Field( + True, description="未设置启用开关时是否默认启用；只对内置插件生效" + ) @field_validator("id") @classmethod diff --git a/backend/src/test/test_plugin_runtime.py b/backend/src/test/test_plugin_runtime.py index 0b76ce88..75b8bbae 100644 --- a/backend/src/test/test_plugin_runtime.py +++ b/backend/src/test/test_plugin_runtime.py @@ -11,6 +11,7 @@ from typing import ClassVar, cast import pytest from ab_sdk import Event, Plugin, PluginDisabled, Verdict, hook, provider, subscribe +from ab_sdk.events import SystemEvent from module.models.config import Plugins from module.plugin import ExtensionPoint, ExtensionRegistry, RegistryError from module.plugin.bus import EventBus @@ -51,8 +52,11 @@ def write_plugin( return plugin_dir -def manifest(plugin_id: str = "demo", sdk: str = ">=0.1,<1") -> PluginManifest: - return parse_manifest(MANIFEST.format(id=plugin_id, sdk=sdk, entry="x:Y")) +def manifest( + plugin_id: str = "demo", sdk: str = ">=0.1,<1", default_enabled: bool = True +) -> PluginManifest: + text = MANIFEST.format(id=plugin_id, sdk=sdk, entry="x:Y") + return parse_manifest(f"{text}default_enabled = {str(default_enabled).lower()}\n") @dataclass(frozen=True, slots=True) @@ -328,6 +332,33 @@ class TestHostContext: ctx.bus.publish(PluginDisabled(plugin_id="x", reason="fake")) await bus.close() + async def test_plugin_system_event_goes_through_notification_center( + self, tmp_path, monkeypatch + ): + from module.notification.manager import NotificationManager + from module.plugin import context + + @dataclass(frozen=True, slots=True) + class Synced(SystemEvent): + kind: ClassVar[str] = "demo.synced" + + sent: list[Event] = [] + + async def send_event(self, event): + sent.append(event) + + monkeypatch.setattr(NotificationManager, "send_event", send_event) + bus = EventBus() + on_bus: list[Event] = [] + bus.subscribe("*", on_bus.append) + ctx = HostPluginContext("demo", None, bus, tmp_path) + ctx.bus.publish(Synced()) + await asyncio.gather(*context._notify_tasks) + await bus.drain() + # send_event 自己负责发布到总线，PluginBus 不再重复入队 + assert sent == [Synced()] and on_bus == [] + await bus.close() + async def test_data_dir_created_lazily(self, tmp_path): ctx = HostPluginContext("demo", None, EventBus(), tmp_path) assert not (tmp_path / "demo").exists() @@ -470,9 +501,12 @@ def candidate_for( plugin_id: str = "demo", source: str = "builtin", sdk: str = ">=0.1,<1", + default_enabled: bool = True, ) -> PluginCandidate: return PluginCandidate( - manifest=manifest(plugin_id, sdk), source=source, load=lambda: cls # type: ignore[arg-type] + manifest=manifest(plugin_id, sdk, default_enabled), + source=source, # type: ignore[arg-type] + load=lambda: cls, ) @@ -558,10 +592,15 @@ class TestManager: assert manager.statuses()[0].state == "active" await manager.stop() - async def test_non_builtin_disabled_by_default(self, tmp_path): + @pytest.mark.parametrize( + "source, default_enabled", [("local", True), ("builtin", False)] + ) + async def test_disabled_by_default(self, tmp_path, source, default_enabled): recorder = Recorder() manager, _ = make_manager( - candidate_for(build_plugin(recorder), source="local"), + candidate_for( + build_plugin(recorder), source=source, default_enabled=default_enabled + ), tmp_path=tmp_path, allow_unsigned=True, ) @@ -591,6 +630,26 @@ class TestManager: assert len(seen) == 1 await manager.stop() + async def test_subscribe_timeout_overrides_bus_default(self, tmp_path): + seen: list[int] = [] + + class Slow(Plugin): + @subscribe("demo.happened", timeout=5) + async def on_demo(self, event): + await asyncio.sleep(0.05) + seen.append(event.value) + + manager, _ = make_manager(candidate_for(Slow), tmp_path=tmp_path) + errors: list[str] = [] + manager.bus = EventBus( + handler_timeout=0.01, on_error=lambda owner, reason: errors.append(reason) + ) + await manager.start() + manager.bus.publish(DemoEvent(1)) + await manager.bus.drain() + assert (seen, errors) == ([1], []) + await manager.stop() + async def test_failed_setup_rolls_back_registrations(self, tmp_path): class Broken(Plugin): @hook("torrent.filter")