Merge branch 'refactor/4.0-p3-ingest' into refactor/4.0-p5-events-api

backend/src/module/core/plugin_tasks.py

@@ -47,9 +47,17 @@ class PluginTasks:
         plugin_id = entry.plugin_id
         try:
             spec: ScheduledTask = entry.factory()
-            spec_run, interval = spec.run, spec.interval
-            initial_delay = spec.initial_delay
+            spec_run, interval_src = spec.run, spec.interval
+            initial_delay = float(spec.initial_delay)
             enabled = spec.enabled or (lambda: True)
+
+            def read_interval() -> float:
+                src = interval_src
+                return float(src() if callable(src) else src)
+
+            # 间隔由插件代码给出，只在受保护的 run() 里读取；调度器拿到的是
+            # 已校验的值，插件出错不会打断调度循环
+            interval = read_interval()
         except Exception as e:
             reason = f"{points.SCHEDULED_TASK} {task_id}: {type(e).__name__}: {e}"
             logger.warning("[Plugin:%s] 定时任务创建失败：%s", plugin_id, reason)
@@ -57,22 +65,25 @@ class PluginTasks:
             return None
 
         async def run() -> None:
-            # enabled 每轮重新读取（ScheduledTask 契约）；调度器只在启动时读
-            # enabled，故任务总是启动，停用时空转跳过
+            # enabled 与 interval 每轮重新读取（ScheduledTask 契约）；调度器只在
+            # 启动时读 enabled，故任务总是启动，停用时空转跳过
+            nonlocal interval
             try:
-                if not enabled():
-                    return
-                await spec_run()
+                ran = enabled()
+                if ran:
+                    await spec_run()
+                interval = read_interval()
             except Exception as e:
                 reason = f"定时任务 {task_id}: {type(e).__name__}: {e}"
                 logger.warning("[Plugin:%s] %s", plugin_id, reason)
                 self._breaker.record_failure(plugin_id, reason)
                 return
-            self._breaker.record_success(plugin_id)
+            if ran:
+                self._breaker.record_success(plugin_id)
 
         return PeriodicTask(
             name=f"plugin:{plugin_id}:{task_id}",
             run=run,
-            interval=interval if callable(interval) else (lambda: float(interval)),
+            interval=lambda: interval,
             initial_delay=initial_delay,
         )

backend/src/module/parser/analyser/providers/registry.py

@@ -86,7 +86,13 @@ class ProviderRegistry:
         from module.plugin.host import provider_impls
 
         result: dict[str, type[LLMProviderAdapter]] = {}
-        for provider_id, adapter_cls in provider_impls(points.LLM_PROVIDER).items():
+
+        def valid(impl: object) -> bool:
+            return isinstance(impl, type) and issubclass(impl, LLMProviderAdapter)
+
+        for provider_id, adapter_cls in provider_impls(
+            points.LLM_PROVIDER, valid
+        ).items():
             if adapter_cls.info.id != provider_id:
                 logger.warning(
                     "LLM provider %s declares info.id=%s; skipped",

backend/src/module/plugin/host.py

@@ -9,6 +9,7 @@ AB 自己的下载器、通知渠道与搜索站点和第三方插件走同一
 """
 
 import logging
+from collections.abc import Callable
 from typing import Any
 
 from ab_sdk import Event, points
@@ -96,16 +97,23 @@ def hook_runner(point: str) -> HookRunner | None:
     return runner
 
 
-def provider_impls(point: str) -> dict[str, Any]:
-    """调用该扩展点全部 Provider 工厂；工厂出错的跳过并记录，不连累其它 Provider。"""
+def provider_impls(
+    point: str, valid: Callable[[Any], bool] = lambda impl: True
+) -> dict[str, Any]:
+    """调用该扩展点全部 Provider 工厂；工厂出错或产物不合契约（``valid``
+    返回 False）的跳过并记录，不连累其它 Provider。"""
     impls: dict[str, Any] = {}
     for provider_id, entry in get_registry().providers(point).items():
         try:
-            impls[provider_id] = entry.factory()
+            impl = entry.factory()
+            if not valid(impl):
+                raise TypeError(f"不合契约的返回值 {impl!r}")
         except Exception as e:
             logger.warning(
                 "[Plugin:%s] %s %s 创建失败：%s", entry.plugin_id, point, provider_id, e
             )
+            continue
+        impls[provider_id] = impl
     return impls
 
 

backend/src/module/searcher/provider.py

@@ -7,6 +7,14 @@ from module.models import RSSItem
 from module.plugin.host import provider_impls
 
 
+def _valid_site(site: object) -> bool:
+    return (
+        isinstance(site, SearchSite)
+        and isinstance(site.url, str)
+        and isinstance(site.parser, str)
+    )
+
+
 def available_sites() -> dict[str, ProviderConfig]:
     """可用搜索站点：插件提供的站点 + 用户配置的站点（同名时用户配置优先）。
 
@@ -14,7 +22,7 @@ def available_sites() -> dict[str, ProviderConfig]:
     """
     sites: dict[str, ProviderConfig] = {}
     site: SearchSite
-    for site_id, site in provider_impls(points.SEARCH_SITE).items():
+    for site_id, site in provider_impls(points.SEARCH_SITE, _valid_site).items():
         sites[site_id] = {"url": site.url, "parser": site.parser}
     sites.update(get_provider())
     return sites

backend/src/test/test_plugin_extensions.py

@@ -207,14 +207,15 @@ class TestLLMProvider:
         with pytest.raises(ValueError):
             ProviderRegistry().resolve("claimed")
 
-    def test_provider_registry_broken_factory_skipped(self, registry):
+    @pytest.mark.parametrize("bad_factory", [broken_factory, object])
+    def test_provider_registry_broken_factory_skipped(self, registry, bad_factory):
         from module.parser.analyser.providers.registry import ProviderRegistry
 
         add_plugin_provider(
             registry, points.LLM_PROVIDER, "ext-llm", make_adapter("ext-llm")
         )
         registry.add_provider(
-            points.LLM_PROVIDER, ProviderEntry("bad", "bad-llm", broken_factory)
+            points.LLM_PROVIDER, ProviderEntry("bad", "bad-llm", bad_factory)
         )
         llm_registry = ProviderRegistry()
         assert "ext-llm" in {i.id for i in llm_registry.list_infos()}
@@ -253,14 +254,17 @@ class TestSearchSite:
         assert item.url == "https://acg/?q=Frieren+S2"
         assert item.parser == "tmdb"
 
-    def test_available_sites_broken_factory_skipped(self, registry):
+    @pytest.mark.parametrize(
+        "bad_factory", [broken_factory, object, lambda: SearchSite(None)]  # type: ignore[arg-type]
+    )
+    def test_available_sites_broken_factory_skipped(self, registry, bad_factory):
         from module.searcher import available_sites
 
         add_plugin_provider(
             registry, points.SEARCH_SITE, "acg", SearchSite("https://acg/?q=%s")
         )
         registry.add_provider(
-            points.SEARCH_SITE, ProviderEntry("bad", "bad-site", broken_factory)
+            points.SEARCH_SITE, ProviderEntry("bad", "bad-site", bad_factory)
         )
         with patch("module.searcher.provider.get_provider", return_value={}):
             sites = available_sites()
@@ -344,7 +348,16 @@ class TestPluginTasks:
         await bridge.sync()
         assert scheduler.tasks == []
 
-    @pytest.mark.parametrize("bad_spec", [None, object()])
+    @pytest.mark.parametrize(
+        "bad_spec",
+        [
+            None,
+            object(),
+            ScheduledTask(AsyncMock(), "abc"),  # type: ignore[arg-type]
+            ScheduledTask(AsyncMock(), lambda: 1 / 0),
+            ScheduledTask(AsyncMock(), 60, initial_delay=None),  # type: ignore[arg-type]
+        ],
+    )
     async def test_sync_malformed_spec_skipped_and_recorded(self, registry, bad_spec):
         scheduler, bridge, tripped = self.make(registry, threshold=1)
         registry.add_provider(