refactor: unify agent data port access

app/agent/memory/__init__.py

@@ -9,7 +9,7 @@ from langchain_core.messages import BaseMessage, messages_from_dict, messages_to
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
-from app.application.agentdata import AgentChatPort as AgentChatOper
+from app.application.agentdata import get_agent_chat_port
 from app.application.messaging.chat import (
     get_configured_agent_chat_persistence,
     get_configured_agent_chat_service,
@@ -86,9 +86,9 @@ class MemoryManager:
             return memory.messages
 
         try:
-            chat = AgentChatOper().get(session_id=session_id, user_id=user_id)
+            chat = get_agent_chat_port().get(session_id=session_id, user_id=user_id)
             if not chat:
-                chat = AgentChatOper().get(session_id=session_id)
+                chat = get_agent_chat_port().get(session_id=session_id)
         except Exception as e:
             logger.debug(f"读取持久化Agent会话失败: {e}")
             return []
@@ -161,7 +161,7 @@ class MemoryManager:
         # 更新内存缓存
         self.save_memory(memory)
         try:
-            AgentChatOper().save_agent_messages(
+            get_agent_chat_port().save_agent_messages(
                 session_id=session_id,
                 user_id=user_id,
                 messages=messages_to_dict(messages),

app/agent/orchestrator.py

@@ -79,8 +79,8 @@ from app.application.plugin.runtime import get_plugin_manager
 def _get_plugin_tools_revision() -> int:
     """读取插件工具目录修订号，避免 Agent 编排依赖具体管理器类型。"""
     return get_plugin_manager().get_plugin_agent_tools_revision()
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
-from app.application.agentdata import UserPort as UserOper
+from app.application.agentdata import get_agent_task_port
+from app.application.agentdata import get_agent_user_port
 from app.application.messaging.chat import (
     get_configured_agent_chat_service,
     get_configured_agent_chat_persistence,
@@ -977,7 +977,7 @@ class MoviePilotAgent:
         if not self.username:
             return False
         try:
-            user = await UserOper().async_get_by_name(self.username)
+            user = await get_agent_user_port().async_get_by_name(self.username)
         except Exception as e:
             logger.error(f"检查 Agent 用户管理员身份失败: {e}")
             return False
@@ -3514,7 +3514,7 @@ class AgentManager:
         """
         if not settings.AI_AGENT_ENABLE:
             return False, "AI Agent 未启用"
-        oper = AgentTaskOper()
+        oper = get_agent_task_port()
         task = oper.get(task_id)
         if not task or not task.enabled:
             return False, "Agent 定时任务不存在或已停用"

app/agent/tools/impl/_filter_rule_utils.py

@@ -5,7 +5,7 @@ import re
 from typing import Any, Dict, Iterable, Optional
 
 from app.runtime.events import eventmanager
-from app.application.agentdata import SubscribePort as SubscribeOper
+from app.application.agentdata import get_agent_subscribe_port
 from app.application.configuration import get_configured_system_config
 from app.application.rules import RuleHelper
 from app.application.rules import RuleParser
@@ -284,7 +284,7 @@ async def collect_rule_group_usages(
             continue
         ensure_usage(name)["used_in_global_best_version"] = True
 
-    subscribes = await SubscribeOper().async_list()
+    subscribes = await get_agent_subscribe_port().async_list()
     for subscribe in subscribes:
         filter_groups = subscribe.filter_groups or []
         for name in filter_groups:
@@ -481,7 +481,7 @@ async def rename_rule_group_references(old_name: str, new_name: str) -> dict:
             await save_system_config(config_key, updated)
             changed["global_settings"][config_key.value] = updated
 
-    subscribe_oper = SubscribeOper()
+    subscribe_oper = get_agent_subscribe_port()
     subscribes = await subscribe_oper.async_list()
     for subscribe in subscribes:
         original = subscribe.filter_groups or []
@@ -519,7 +519,7 @@ async def remove_rule_group_references(group_name: str) -> dict:
             await save_system_config(config_key, updated)
             changed["global_settings"][config_key.value] = updated
 
-    subscribe_oper = SubscribeOper()
+    subscribe_oper = get_agent_subscribe_port()
     subscribes = await subscribe_oper.async_list()
     for subscribe in subscribes:
         original = subscribe.filter_groups or []

app/agent/tools/impl/add_download_tasks.py

@@ -17,7 +17,7 @@ from app.runtime.settings import RuntimeSettingsCompat
 settings = RuntimeSettingsCompat()
 from app.domain.context import Context
 from app.domain.metainfo import MetaInfo
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.application.directory import DirectoryHelper, validate_download_save_path
 from app.runtime.log import logger
 from app.schemas.file import FileURI
@@ -277,7 +277,7 @@ class AddDownloadTasksTool(MoviePilotTool):
                         failed_messages.append(f"{torrent_input} 缺少站点名称")
                         continue
 
-                    siteinfo = await SiteOper().async_get_by_name(site_name)
+                    siteinfo = await get_agent_site_port().async_get_by_name(site_name)
                     if not siteinfo:
                         failed_messages.append(f"{torrent_input} 未找到站点信息 {site_name}")
                         continue

app/agent/tools/impl/add_subscribe.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.subscribe import SubscribeChain
-from app.application.agentdata import UserPort as UserOper
+from app.application.agentdata import get_agent_user_port
 from app.runtime.log import logger
 from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaSource, MediaType, NotificationChannel
 from app.domain.media import normalize_music_type
@@ -154,7 +154,7 @@ class AddSubscribeTool(MoviePilotTool):
 
         mapped_username = await self.run_blocking(
             "db",
-            UserOper().get_name,
+            get_agent_user_port().get_name,
             **{key: self._user_id for key in binding_keys},
         )
         return mapped_username or resolved_username

app/agent/tools/impl/create_agent_task.py

@@ -10,8 +10,8 @@ from app.agent.tools.tags import ToolTag
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
-from app.application.agentdata import AgentChatPort as AgentChatOper
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
+from app.application.agentdata import get_agent_chat_port
+from app.application.agentdata import get_agent_task_port
 from app.runtime.scheduling import TimerUtils
 
 
@@ -115,11 +115,11 @@ class CreateAgentTaskTool(MoviePilotTool):
             timezone_name=settings.TZ,
             require_future=True,
         )
-        chat = AgentChatOper().get(
+        chat = get_agent_chat_port().get(
             session_id=self._session_id,
             user_id=self._user_id,
         )
-        task = AgentTaskOper().add(
+        task = get_agent_task_port().add(
             name=payload.name.strip(),
             content=payload.content.strip(),
             trigger_type=payload.trigger_type,
@@ -133,7 +133,7 @@ class CreateAgentTaskTool(MoviePilotTool):
             original_chat_id=chat.original_chat_id if chat else None,
         )
         next_run_at = update_agent_task_job(task.id)
-        return AgentTaskOper.to_dict(
+        return get_agent_task_port().to_dict(
             task,
             next_run_at=next_run_at,
             timezone=settings.TZ,

app/agent/tools/impl/delete_agent_task.py

@@ -4,7 +4,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
+from app.application.agentdata import get_agent_task_port
 
 
 class DeleteAgentTaskInput(BaseModel):
@@ -33,7 +33,7 @@ class DeleteAgentTaskTool(MoviePilotTool):
         """删除当前用户的任务并移除运行时调度。"""
         from app.application.scheduling import remove_agent_task_job
 
-        deleted = AgentTaskOper().delete(
+        deleted = get_agent_task_port().delete(
             task_id=task_id,
             user_id=str(self._user_id),
         )

app/agent/tools/impl/delete_download_history.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import DownloadHistoryPort as DownloadHistoryOper
+from app.application.agentdata import get_agent_download_history_port
 from app.runtime.log import logger
 
 
@@ -37,7 +37,7 @@ class DeleteDownloadHistoryTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: history_id={history_id}")
 
         try:
-            await DownloadHistoryOper().async_delete_history(history_id)
+            await get_agent_download_history_port().async_delete_history(history_id)
             return f"下载历史记录 ID: {history_id} 已成功删除"
         except Exception as e:
             logger.error(f"删除下载历史记录失败: {e}", exc_info=True)

app/agent/tools/impl/delete_transfer_history.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.storage import StorageChain
-from app.application.agentdata import TransferHistoryPort as TransferHistoryOper
+from app.application.agentdata import get_agent_transfer_history_port
 from app.runtime.log import logger
 from app.schemas.workflow import FileItem
 
@@ -53,7 +53,7 @@ class DeleteTransferHistoryTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: history_id={history_id}")
 
         try:
-            transferhis = TransferHistoryOper()
+            transferhis = get_agent_transfer_history_port()
             history = await transferhis.async_get(history_id)
             if not history:
                 return f"错误：整理历史记录不存在，ID={history_id}"

app/agent/tools/impl/query_agent_tasks.py

@@ -8,7 +8,7 @@ from app.agent.tools.tags import ToolTag
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
+from app.application.agentdata import get_agent_task_port
 
 
 class QueryAgentTasksInput(BaseModel):
@@ -52,7 +52,7 @@ class QueryAgentTasksTool(MoviePilotTool):
         """读取当前用户的任务及运行时下一次触发时间。"""
         from app.application.scheduling import get_agent_task_next_run
 
-        oper = AgentTaskOper()
+        oper = get_agent_task_port()
         if task_id:
             task = oper.get(task_id=task_id, user_id=str(self._user_id))
             tasks = [task] if task else []

app/agent/tools/impl/query_download_tasks.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.download import DownloadChain
-from app.application.agentdata import DownloadHistoryPort as DownloadHistoryOper
+from app.application.agentdata import get_agent_download_history_port
 from app.runtime.log import logger
 from app.schemas.transfer import DownloaderTorrent
 from app.schemas.types import MUSIC_ENTITY_RECORDING, TorrentQueryStatus, media_type_to_agent
@@ -143,7 +143,7 @@ class QueryDownloadTasksTool(MoviePilotTool):
         hashes = [torrent.hash for torrent in torrents if getattr(torrent, "hash", None)]
         if not hashes:
             return {}
-        return DownloadHistoryOper().get_by_hashes(hashes)
+        return get_agent_download_history_port().get_by_hashes(hashes)
 
     @classmethod
     def _query_downloads_sync(

app/agent/tools/impl/query_plugin_data.py

@@ -12,7 +12,7 @@ from app.agent.tools.impl._plugin_tool_utils import (
     build_preview_payload,
     get_plugin_snapshot,
 )
-from app.application.agentdata import PluginDataPort as PluginDataOper
+from app.application.agentdata import get_agent_plugin_data_port
 from app.runtime.log import logger
 
 
@@ -73,7 +73,7 @@ class QueryPluginDataTool(MoviePilotTool):
                 ensure_ascii=False,
             )
 
-        plugin_data_oper = PluginDataOper()
+        plugin_data_oper = get_agent_plugin_data_port()
         if key:
             value = await plugin_data_oper.async_get_data(plugin_id, key)
             if value is None:

app/agent/tools/impl/query_site_userdata.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.runtime.log import logger
 
 SITE_USERDATA_DETAIL_PREVIEW_LIMIT = 10
@@ -62,7 +62,7 @@ class QuerySiteUserdataTool(MoviePilotTool):
         )
 
         try:
-            site_oper = SiteOper()
+            site_oper = get_agent_site_port()
             site = await site_oper.async_get(site_id)
             if not site:
                 return json.dumps(

app/agent/tools/impl/query_sites.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.runtime.log import logger
 
 
@@ -58,7 +58,7 @@ class QuerySitesTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: status={status}, name={name}")
         try:
             is_admin = await self.is_admin_user()
-            site_oper = SiteOper()
+            site_oper = get_agent_site_port()
             # 获取所有站点（按优先级排序）
             sites = await site_oper.async_list()
             filtered_sites = []

app/agent/tools/impl/query_subscribe_history.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import SubscribeHistoryPort as SubscribeHistoryOper
+from app.application.agentdata import get_agent_subscribe_history_port
 from app.runtime.log import logger
 from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaType, media_type_to_agent
 from app.domain.media import normalize_music_type
@@ -106,7 +106,7 @@ class QuerySubscribeHistoryTool(MoviePilotTool):
                 if MediaType.MUSIC.value not in requested_types:
                     return "错误：music_type 仅能与 media_type='music' 或 'all' 一起使用"
 
-            subscribe_history_oper = SubscribeHistoryOper()
+            subscribe_history_oper = get_agent_subscribe_history_port()
             if name:
                 # 有名称过滤时，获取足够多的记录在内存中过滤，不分页
                 fetch_count = 500

app/agent/tools/impl/query_subscribes.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import SubscribePort as SubscribeOper
+from app.application.agentdata import get_agent_subscribe_port
 from app.runtime.log import logger
 from app.schemas.subscribe import Subscribe as SubscribeSchema
 from app.schemas.types import (
@@ -157,7 +157,7 @@ class QuerySubscribesTool(MoviePilotTool):
                 if media_type not in ("all", "music"):
                     return "错误：music_type 仅能与 media_type='music' 或 'all' 一起使用"
 
-            subscribe_oper = SubscribeOper()
+            subscribe_oper = get_agent_subscribe_port()
             subscribes = await subscribe_oper.async_list()
             filtered_subscribes = []
             for sub in subscribes:

app/agent/tools/impl/query_transfer_history.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import TransferHistoryPort as TransferHistoryOper
+from app.application.agentdata import get_agent_transfer_history_port
 from app.runtime.log import logger
 from app.schemas.types import media_type_to_agent
 from app.foundation.text import cut as jieba_cut
@@ -75,7 +75,7 @@ class QueryTransferHistoryTool(MoviePilotTool):
             # 每页固定 30 条，与工具说明保持一致，避免整理路径等字段撑大上下文。
             count = 30
 
-            transferhis = TransferHistoryOper()
+            transferhis = get_agent_transfer_history_port()
             # 处理标题搜索
             if title:
                 # 使用统一分词封装处理标题，便于替换底层实现。

app/agent/tools/impl/query_workflows.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import WorkflowPort as WorkflowOper
+from app.application.agentdata import get_agent_workflow_port
 from app.runtime.log import logger
 
 
@@ -54,7 +54,7 @@ class QueryWorkflowsTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: state={state}, name={name}, trigger_type={trigger_type}")
 
         try:
-            workflow_oper = WorkflowOper()
+            workflow_oper = get_agent_workflow_port()
             workflows = await workflow_oper.async_list()
 
             # 过滤工作流

app/agent/tools/impl/run_agent_task.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel, Field
 
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
+from app.application.agentdata import get_agent_task_port
 
 
 class RunAgentTaskInput(BaseModel):
@@ -42,7 +42,7 @@ class RunAgentTaskTool(MoviePilotTool):
 
     def _get_task_state(self, task_id: int) -> tuple[str, Optional[str]]:
         """校验任务归属和状态，返回可执行性及任务名称。"""
-        task = AgentTaskOper().get(
+        task = get_agent_task_port().get(
             task_id=task_id,
             user_id=str(self._user_id),
         )

app/agent/tools/impl/search_subscribe.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.subscribe import SubscribeChain
-from app.application.agentdata import SubscribePort as SubscribeOper
+from app.application.agentdata import get_agent_subscribe_port
 from app.runtime.log import logger
 from app.schemas.types import media_type_to_agent
 
@@ -57,7 +57,7 @@ class SearchSubscribeTool(MoviePilotTool):
 
         try:
             # 先验证订阅是否存在
-            subscribe_oper = SubscribeOper()
+            subscribe_oper = get_agent_subscribe_port()
             subscribe = await subscribe_oper.async_get(subscribe_id)
 
             if not subscribe:

app/agent/tools/impl/test_site.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.site import SiteChain
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.runtime.log import logger
 
 
@@ -33,7 +33,7 @@ class TestSiteTool(MoviePilotTool):
     @staticmethod
     def _test_site_sync(site_identifier: int) -> tuple[Optional[str], Optional[str], bool, str]:
         """在同步线程里执行站点联通测试，避免网络请求卡住事件循环。"""
-        site = SiteOper().get(site_identifier)
+        site = get_agent_site_port().get(site_identifier)
         if not site:
             return None, None, False, f"未找到站点：{site_identifier}，请使用 query_sites 工具查询可用的站点"
 

app/agent/tools/impl/update_agent_task.py

@@ -10,7 +10,7 @@ from app.agent.tools.tags import ToolTag
 from app.runtime.settings import RuntimeSettingsCompat
 
 settings = RuntimeSettingsCompat()
-from app.application.agentdata import AgentTaskPort as AgentTaskOper
+from app.application.agentdata import get_agent_task_port
 from app.runtime.scheduling import TimerUtils
 
 
@@ -104,7 +104,7 @@ class UpdateAgentTaskTool(MoviePilotTool):
         """更新当前用户的任务并刷新运行时调度。"""
         from app.application.scheduling import update_agent_task_job
 
-        oper = AgentTaskOper()
+        oper = get_agent_task_port()
         task = oper.get(task_id=payload.task_id, user_id=str(self._user_id))
         if not task:
             return None

app/agent/tools/impl/update_site.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.runtime.events import eventmanager
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.runtime.log import logger
 from app.schemas.types import EventType
 from app.foundation import url as url_tools
@@ -124,7 +124,7 @@ class UpdateSiteTool(MoviePilotTool):
         logger.info(f"执行工具: {self.name}, 参数: site_id={site_id}")
 
         try:
-            site_oper = SiteOper()
+            site_oper = get_agent_site_port()
             site = await site_oper.async_get(site_id)
             if not site:
                 return json.dumps(

app/agent/tools/impl/update_site_cookie.py

@@ -7,7 +7,7 @@ from pydantic import BaseModel, Field
 from app.agent.tools.base import MoviePilotTool
 from app.agent.tools.tags import ToolTag
 from app.chain.site import SiteChain
-from app.application.agentdata import SitePort as SiteOper
+from app.application.agentdata import get_agent_site_port
 from app.runtime.log import logger
 
 
@@ -59,7 +59,7 @@ class UpdateSiteCookieTool(MoviePilotTool):
         """
         在同步线程里执行站点登录和 Cookie 更新，避免网络登录阻塞协程。
         """
-        site = SiteOper().get(site_identifier)
+        site = get_agent_site_port().get(site_identifier)
         if not site:
             return None, False, f"未找到站点：{site_identifier}，请使用 query_sites 工具查询可用的站点"
 

app/application/agentdata.py

@@ -125,3 +125,53 @@ def get_agent_data_ports() -> AgentDataPorts:
     if _ports is None:
         raise RuntimeError("Agent 数据端口尚未配置")
     return _ports
+
+
+def get_agent_chat_port() -> Any:
+    """创建 Agent 会话数据端口实例。"""
+    return get_agent_data_ports().agent_chat()
+
+
+def get_agent_task_port() -> Any:
+    """创建 Agent 定时任务数据端口实例。"""
+    return get_agent_data_ports().agent_task()
+
+
+def get_agent_user_port() -> Any:
+    """创建 Agent 用户数据端口实例。"""
+    return get_agent_data_ports().user()
+
+
+def get_agent_site_port() -> Any:
+    """创建 Agent 站点数据端口实例。"""
+    return get_agent_data_ports().site()
+
+
+def get_agent_subscribe_port() -> Any:
+    """创建 Agent 订阅数据端口实例。"""
+    return get_agent_data_ports().subscribe()
+
+
+def get_agent_subscribe_history_port() -> Any:
+    """创建 Agent 订阅历史数据端口实例。"""
+    return get_agent_data_ports().subscribe_history()
+
+
+def get_agent_transfer_history_port() -> Any:
+    """创建 Agent 整理历史数据端口实例。"""
+    return get_agent_data_ports().transfer_history()
+
+
+def get_agent_download_history_port() -> Any:
+    """创建 Agent 下载历史数据端口实例。"""
+    return get_agent_data_ports().download_history()
+
+
+def get_agent_workflow_port() -> Any:
+    """创建 Agent 工作流数据端口实例。"""
+    return get_agent_data_ports().workflow()
+
+
+def get_agent_plugin_data_port() -> Any:
+    """创建 Agent 插件数据端口实例。"""
+    return get_agent_data_ports().plugin_data()

docs/refactor/backend-architecture-next-stage.md

@@ -6,7 +6,7 @@
 > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本
 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文
 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md`
-> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口。
+> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配和 Outbox 外围扩展仍按风险切片推进。Model/Base 查询与写装饰器、legacy 隐式会话外壳均已清零，插件 SDK 也不再导出宿主 Model。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义；阶段 1b1 已收口整理 worker、pending 回放、失败通知、进程内 AI 重试、插件监控与事件投递的生命周期所有权；2026-08-24 的阶段 2 已将 212 个已观察宿主模块方法的 legacy aggregation 清零，并补齐可执行 fanout 与下载器文件 DTO 边界；阶段 3 已将消息交互和远程命令的订阅删除统一到 Application/UoW/outbox，宿主不再调用裸线程统计入口；阶段 4 已统一七种消息渠道的宿主回环与后台执行边界；阶段 5 已补齐事件窗口聚合任务的生命周期所有权；阶段 6 已统一插件文件操作的取消完成语义；阶段 7 已统一插件协程补偿的终态等待；阶段 8 已统一宿主同步函数的异步线程池入口；阶段 9 已统一工作流运行时的宿主获取路径；阶段 10 已统一模块、插件与调度运行时的显式 getter 调用；阶段 11 已清除系统配置 getter 的 Oper 形别名；阶段 12 已完成工作流域的显式 Chain 数据端口迁移；阶段 13 已收口用户、交互与消息链的数据端口；阶段 14 已收口音乐订阅数据端口；阶段 15 已收口站点数据端口；阶段 16 已收口媒体服务器数据端口；阶段 17 已收口下载数据端口；阶段 18 已收口主订阅数据端口；阶段 19 已收口整理数据端口；阶段 20 已收口 Agent 数据端口。
 
 ## 当前复核结论（2026-08-24）
 
@@ -212,6 +212,15 @@
 - 兼容边界不变：三个 DB Oper、Proxy 类、整理公开入口、队列和生命周期、历史字段、事件 payload 以及
   V2/V3 插件调用方式均未修改。
 
+### 长期整改阶段 20：Agent 数据端口收口（2026-08-24）
+
+- Agent 编排、会话记忆和 22 个工具实现原先把十种 `AgentDataPort` 代理重新别名为数据库 Oper，形成与
+  Chain 命名 getter 不同的第二套数据端口用法；现在统一调用 `get_agent_*_port()`。
+- Agent 测试接缝改为替换 getter 返回的端口实例，架构门禁遍历 `app/agent/**`，禁止生产模块重新导入
+  十个兼容 Port 代理。
+- 兼容边界不变：`AgentDataPorts` 注册表、十个旧 Port 代理类、DB Oper、Agent/工具公开参数和返回值、
+  权限判断以及 V2/V3 插件调用方式均未修改。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：

docs/rules/05-architecture.md

@@ -136,6 +136,10 @@ Chain consumers use the named `get_chain_*_port()` functions from
 `app/application/chain/data.py`; they must not alias migration-time `*PortProxy`
 classes back to database Oper names. Those proxy classes remain compatibility
 boundaries while the other established Chain domains migrate independently.
+Agent orchestration, memory and tool implementations follow the same rule via
+the named `get_agent_*_port()` functions from `app/application/agentdata.py`.
+The legacy Agent `*Port` proxy classes remain import-compatible boundaries and
+must not be reintroduced as Oper aliases in canonical Agent modules.
 
 ### Adapter boundaries
 

tests/test_agent_add_subscribe_tool.py

@@ -1,5 +1,6 @@
 import asyncio
 import unittest
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
 from app.agent.tools.impl.add_subscribe import AddSubscribeTool
@@ -28,8 +29,10 @@ class TestAgentAddSubscribeTool(unittest.TestCase):
             "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add",
             new=AsyncMock(return_value=(1, "")),
         ) as async_add, patch(
-            "app.agent.tools.impl.add_subscribe.UserOper.get_name",
-            return_value="moviepilot-user",
+            "app.agent.tools.impl.add_subscribe.get_agent_user_port",
+            return_value=SimpleNamespace(
+                get_name=lambda **_kwargs: "moviepilot-user"
+            ),
         ):
             result = asyncio.run(
                 tool.run(
@@ -55,8 +58,8 @@ class TestAgentAddSubscribeTool(unittest.TestCase):
             "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add",
             new=AsyncMock(return_value=(1, "")),
         ) as async_add, patch(
-            "app.agent.tools.impl.add_subscribe.UserOper.get_name",
-            return_value=None,
+            "app.agent.tools.impl.add_subscribe.get_agent_user_port",
+            return_value=SimpleNamespace(get_name=lambda **_kwargs: None),
         ):
             result = asyncio.run(
                 tool.run(
@@ -81,8 +84,8 @@ class TestAgentAddSubscribeTool(unittest.TestCase):
             "app.agent.tools.impl.add_subscribe.SubscribeChain.async_add",
             new=AsyncMock(return_value=(1, "")),
         ) as async_add, patch(
-            "app.agent.tools.impl.add_subscribe.UserOper.get_name",
-            return_value=None,
+            "app.agent.tools.impl.add_subscribe.get_agent_user_port",
+            return_value=SimpleNamespace(get_name=lambda **_kwargs: None),
         ):
             result = asyncio.run(
                 tool.run(

tests/test_agent_data_ports.py

@@ -0,0 +1,28 @@
+"""Agent 命名数据端口工厂测试。"""
+
+from app.application import agentdata
+
+
+def test_named_agent_data_getters_use_registered_factories(monkeypatch) -> None:
+    """每个命名 getter 都应创建组合根登记的对应端口实例。"""
+    names = {
+        "agent_chat": agentdata.get_agent_chat_port,
+        "agent_task": agentdata.get_agent_task_port,
+        "user": agentdata.get_agent_user_port,
+        "site": agentdata.get_agent_site_port,
+        "subscribe": agentdata.get_agent_subscribe_port,
+        "subscribe_history": agentdata.get_agent_subscribe_history_port,
+        "transfer_history": agentdata.get_agent_transfer_history_port,
+        "download_history": agentdata.get_agent_download_history_port,
+        "workflow": agentdata.get_agent_workflow_port,
+        "plugin_data": agentdata.get_agent_plugin_data_port,
+    }
+    factories = {
+        name: (lambda current=name: current)
+        for name in names
+    }
+    monkeypatch.setattr(agentdata, "_ports", agentdata.AgentDataPorts(**factories))
+
+    assert {name: getter() for name, getter in names.items()} == {
+        name: name for name in names
+    }

tests/test_agent_music_tools.py

@@ -428,8 +428,10 @@ def test_add_download_preserves_album_context_and_full_coverage_marker():
         "_async_resolve_cached_context",
         new=AsyncMock(return_value=cached_context),
     ), patch(
-        "app.agent.tools.impl.add_download_tasks.SiteOper.async_get_by_name",
-        new=AsyncMock(return_value=site),
+        "app.agent.tools.impl.add_download_tasks.get_agent_site_port",
+        return_value=SimpleNamespace(
+            async_get_by_name=AsyncMock(return_value=site)
+        ),
     ), patch.object(
         AddDownloadTasksTool,
         "_download_single_sync",
@@ -479,7 +481,7 @@ def test_query_subscribe_history_uses_database_media_values_and_music_fields(mon
             return [record] if mtype == MediaType.MUSIC.value else []
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.query_subscribe_history.SubscribeHistoryOper",
+        "app.agent.tools.impl.query_subscribe_history.get_agent_subscribe_history_port",
         FakeHistoryOper,
     )
     tool = QuerySubscribeHistoryTool(session_id="session-1", user_id="10001")
@@ -513,7 +515,7 @@ def test_music_history_filter_excludes_video_records(monkeypatch):
             return [movie_record] if mtype == MediaType.MOVIE.value else []
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.query_subscribe_history.SubscribeHistoryOper",
+        "app.agent.tools.impl.query_subscribe_history.get_agent_subscribe_history_port",
         FakeHistoryOper,
     )
     tool = QuerySubscribeHistoryTool(session_id="session-1", user_id="10001")

tests/test_agent_plugin_tools.py

@@ -422,7 +422,7 @@ def test_query_plugin_data_truncates_large_payload() -> None:
             return_value=_plugin_snapshot(),
         ),
         patch(
-            "app.agent.tools.impl.query_plugin_data.PluginDataOper",
+            "app.agent.tools.impl.query_plugin_data.get_agent_plugin_data_port",
             return_value=plugin_data_oper,
         ),
     ):

tests/test_agent_query_subscribes.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
-from unittest.mock import patch
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
 
 from app.agent.tools.impl.query_subscribes import QuerySubscribesTool
 from app.db.models.subscribe import Subscribe
@@ -22,8 +23,10 @@ def test_agent_query_subscribes_returns_manual_total_episode():
     )
 
     with patch(
-        "app.agent.tools.impl.query_subscribes.SubscribeOper.async_list",
-        return_value=[subscribe],
+        "app.agent.tools.impl.query_subscribes.get_agent_subscribe_port",
+        return_value=SimpleNamespace(
+            async_list=AsyncMock(return_value=[subscribe])
+        ),
     ):
         result = asyncio.run(
             QuerySubscribesTool(session_id="session-1", user_id="10001").run(

tests/test_agent_query_workflows_tool.py

@@ -28,7 +28,7 @@ class TestQueryWorkflowsTool(unittest.TestCase):
         workflow_oper.async_list = AsyncMock(return_value=[workflow])
 
         with patch(
-            "app.agent.tools.impl.query_workflows.WorkflowOper",
+            "app.agent.tools.impl.query_workflows.get_agent_workflow_port",
             return_value=workflow_oper,
         ):
             result = asyncio.run(tool.run())

tests/test_agent_resource_flow_permissions.py

@@ -115,7 +115,7 @@ def test_query_sites_hides_only_sensitive_fields_for_non_admin_user():
     )
 
     with patch(
-        "app.agent.tools.impl.query_sites.SiteOper"
+        "app.agent.tools.impl.query_sites.get_agent_site_port"
     ) as site_oper:
         site_oper.return_value.async_list = AsyncMock(return_value=[site])
         result = asyncio.run(tool.run())
@@ -177,7 +177,7 @@ def test_query_sites_keeps_full_fields_for_admin_context():
     )
 
     with patch(
-        "app.agent.tools.impl.query_sites.SiteOper"
+        "app.agent.tools.impl.query_sites.get_agent_site_port"
     ) as site_oper:
         site_oper.return_value.async_list = AsyncMock(return_value=[site])
         result = asyncio.run(tool.run())
@@ -369,7 +369,7 @@ def test_channel_agent_admin_user_id_does_not_bypass_user_lookup():
         username="normal-user",
     )
 
-    with patch("app.agent.orchestrator.UserOper") as user_oper:
+    with patch("app.agent.orchestrator.get_agent_user_port") as user_oper:
         user_oper.return_value.async_get_by_name.return_value = SimpleNamespace(
             is_superuser=False
         )
@@ -391,7 +391,7 @@ def test_channel_agent_rejects_local_admin_username_without_trusted_principal():
     )
     agent.is_channel_admin = False
 
-    with patch("app.agent.orchestrator.UserOper") as user_oper:
+    with patch("app.agent.orchestrator.get_agent_user_port") as user_oper:
         user_oper.return_value.async_get_by_name = AsyncMock(
             return_value=SimpleNamespace(is_superuser=True)
         )
@@ -414,7 +414,7 @@ def test_channel_agent_accepts_trusted_admin_principal_without_local_user():
     )
     agent.is_channel_admin = True
 
-    with patch("app.agent.orchestrator.UserOper") as user_oper:
+    with patch("app.agent.orchestrator.get_agent_user_port") as user_oper:
         context = asyncio.run(
             agent._build_tool_context(should_dispatch_reply=True)
         )

tests/test_agent_search_subscribe.py

@@ -47,7 +47,7 @@ def test_search_subscribe_uses_async_data_port(monkeypatch) -> None:
             raise AssertionError("async 工具不应调用同步订阅更新")
 
     port = _AsyncSubscribePort()
-    monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port)
+    monkeypatch.setattr(search_subscribe_module, "get_agent_subscribe_port", lambda: port)
 
     async def _run_blocking(*_args, **_kwargs):
         await asyncio.sleep(0)
@@ -79,7 +79,7 @@ def test_search_subscribe_rejects_paused_subscription_without_search(monkeypatch
         async_update = AsyncMock()
 
     port = _AsyncSubscribePort()
-    monkeypatch.setattr(search_subscribe_module, "SubscribeOper", lambda: port)
+    monkeypatch.setattr(search_subscribe_module, "get_agent_subscribe_port", lambda: port)
     run_blocking = AsyncMock()
     monkeypatch.setattr(SearchSubscribeTool, "run_blocking", run_blocking)
 

tests/test_architecture_dependencies.py

@@ -568,6 +568,37 @@ def test_transfer_chains_use_explicit_data_port_getters():
     assert violations == []
 
 
+def test_agent_consumers_use_explicit_data_port_getters():
+    """Agent 生产模块不得把兼容数据端口代理重新伪装成数据库 Oper。"""
+    forbidden = {
+        "AgentChatPort",
+        "AgentTaskPort",
+        "DownloadHistoryPort",
+        "PluginDataPort",
+        "SitePort",
+        "SubscribeHistoryPort",
+        "SubscribePort",
+        "TransferHistoryPort",
+        "UserPort",
+        "WorkflowPort",
+    }
+    violations: list[str] = []
+    for path in (APP_ROOT / "agent").rglob("*.py"):
+        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        for node in ast.walk(tree):
+            if not isinstance(node, ast.ImportFrom):
+                continue
+            if node.module != "app.application.agentdata":
+                continue
+            for alias in node.names:
+                if alias.name in forbidden:
+                    violations.append(
+                        f"{path.relative_to(PROJECT_ROOT).as_posix()}:{node.lineno}:{alias.name}"
+                    )
+
+    assert violations == []
+
+
 def test_plugin_components_do_not_reexport_legacy_abi_names():
     """新插件组件只提供 canonical 能力，不得复制旧 Helper、Manager 或 Oper 导出。"""
     violations: list[str] = []

tests/test_delete_transfer_history_tool.py

@@ -41,7 +41,7 @@ def test_delete_transfer_history_tool_removes_old_dest_file_before_history(monke
             return True
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -96,7 +96,7 @@ def test_delete_transfer_history_tool_keeps_history_when_old_dest_delete_fails(m
             return False
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -150,7 +150,7 @@ def test_delete_transfer_history_tool_deletes_history_when_old_dest_is_missing(m
             return False
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -205,7 +205,7 @@ def test_delete_transfer_history_tool_keeps_successful_move_dest_as_reorganize_s
             return True
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -258,7 +258,7 @@ def test_delete_transfer_history_tool_only_treats_exact_move_as_reorganize_sourc
             return True
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -313,7 +313,7 @@ def test_delete_transfer_history_storage_work_runs_outside_event_loop(monkeypatc
             return True
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(
@@ -371,7 +371,7 @@ def test_delete_transfer_history_cancellation_keeps_history_record(monkeypatch):
             return True
 
     monkeypatch.setattr(
-        "app.agent.tools.impl.delete_transfer_history.TransferHistoryOper",
+        "app.agent.tools.impl.delete_transfer_history.get_agent_transfer_history_port",
         FakeTransferHistoryOper,
     )
     monkeypatch.setattr(