fix(agent): use async subscription data access (#6385)

app/agent/tools/impl/search_subscribe.py

@@ -58,7 +58,7 @@ class SearchSubscribeTool(MoviePilotTool):
         try:
             # 先验证订阅是否存在
             subscribe_oper = SubscribeOper()
-            subscribe = subscribe_oper.get(subscribe_id)
+            subscribe = await subscribe_oper.async_get(subscribe_id)
 
             if not subscribe:
                 return json.dumps({
@@ -93,7 +93,10 @@ class SearchSubscribeTool(MoviePilotTool):
 
             # 如果提供了 filter_groups 参数，先更新订阅的规则组
             if filter_groups is not None:
-                subscribe_oper.update(subscribe_id, {"filter_groups": filter_groups})
+                await subscribe_oper.async_update(
+                    subscribe_id,
+                    {"filter_groups": filter_groups},
+                )
                 logger.info(f"更新订阅 #{subscribe_id} 的规则组为: {filter_groups}")
 
             # 订阅搜索会触发大量同步站点访问，统一走 subscribe 线程池。
@@ -106,7 +109,7 @@ class SearchSubscribeTool(MoviePilotTool):
             )
 
             # 重新获取订阅信息以获取更新后的状态
-            updated_subscribe = subscribe_oper.get(subscribe_id)
+            updated_subscribe = await subscribe_oper.async_get(subscribe_id)
             if updated_subscribe:
                 subscribe_info.update({
                     "state": updated_subscribe.state,

tests/test_agent_search_subscribe.py

@@ -0,0 +1,94 @@
+"""订阅搜索 Agent 工具的数据访问边界。"""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, call
+
+from app.agent.tools.impl import search_subscribe as search_subscribe_module
+from app.agent.tools.impl.search_subscribe import SearchSubscribeTool
+
+
+def _subscribe(*, state: str = "R") -> SimpleNamespace:
+    """构造搜索工具需要的最小订阅记录。"""
+    return SimpleNamespace(
+        id=7,
+        name="Example",
+        year="2026",
+        type="TV",
+        season=1,
+        state=state,
+        total_episode=12,
+        lack_episode=2,
+        media_source="tmdb",
+        media_id="123",
+        music_type=None,
+        total_tracks=None,
+        description=None,
+        last_update="2026-08-22 00:00:00",
+        filter_groups=[],
+    )
+
+
+def test_search_subscribe_uses_async_data_port(monkeypatch) -> None:
+    """订阅搜索不能在 async 工具中调用同步 DB 端口。"""
+    record = _subscribe()
+    updated = _subscribe()
+
+    class _AsyncSubscribePort:
+        def __init__(self) -> None:
+            self.async_get = AsyncMock(side_effect=[record, updated])
+            self.async_update = AsyncMock()
+
+        def get(self, _subscribe_id):
+            raise AssertionError("async 工具不应调用同步订阅查询")
+
+        def update(self, _subscribe_id, _payload):
+            raise AssertionError("async 工具不应调用同步订阅更新")
+
+    port = _AsyncSubscribePort()
+    monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port)
+
+    async def _run_blocking(*_args, **_kwargs):
+        await asyncio.sleep(0)
+
+    monkeypatch.setattr(SearchSubscribeTool, "run_blocking", _run_blocking)
+
+    result = asyncio.run(
+        SearchSubscribeTool(session_id="test", user_id="1").run(
+            subscribe_id=record.id,
+            filter_groups=["default"],
+        )
+    )
+
+    payload = json.loads(result)
+    assert payload["success"] is True
+    assert port.async_get.await_args_list == [call(record.id), call(record.id)]
+    port.async_update.assert_awaited_once_with(
+        record.id,
+        {"filter_groups": ["default"]},
+    )
+
+
+def test_search_subscribe_rejects_paused_subscription_without_search(monkeypatch) -> None:
+    """暂停订阅仍应在异步读取后立即返回，不提交搜索任务。"""
+    record = _subscribe(state="S")
+
+    class _AsyncSubscribePort:
+        async_get = AsyncMock(return_value=record)
+        async_update = AsyncMock()
+
+    port = _AsyncSubscribePort()
+    monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port)
+    run_blocking = AsyncMock()
+    monkeypatch.setattr(SearchSubscribeTool, "run_blocking", run_blocking)
+
+    result = asyncio.run(
+        SearchSubscribeTool(session_id="test", user_id="1").run(
+            subscribe_id=record.id,
+        )
+    )
+
+    payload = json.loads(result)
+    assert payload["success"] is False
+    run_blocking.assert_not_awaited()