refactor(agent): split manager runtime owners

app/agent/__init__.py

@@ -1,19 +1,30 @@
-"""Agent 公共入口，按需加载完整编排实现。"""
+"""Agent 公共入口，仅按白名单延迟加载稳定契约。"""
 
 from importlib import import_module
 from typing import Any
 
+_PUBLIC_SYMBOLS = {
+    "AgentChain": ("app.chain.agent", "AgentChain"),
+    "AgentManager": ("app.agent.manager", "AgentManager"),
+    "HEARTBEAT_SESSION_PREFIX": ("app.agent.orchestrator", "HEARTBEAT_SESSION_PREFIX"),
+    "MoviePilotAgent": ("app.agent.orchestrator", "MoviePilotAgent"),
+    "ReplyMode": ("app.agent.contracts", "ReplyMode"),
+    "UNSUPPORTED_IMAGE_INPUT_MESSAGE": ("app.agent.orchestrator", "UNSUPPORTED_IMAGE_INPUT_MESSAGE"),
+    "agent_manager": ("app.agent.manager", "agent_manager"),
+}
+
 
 def __getattr__(name: str) -> Any:
-    """按需从 Agent 编排模块解析历史包级公开对象。"""
-    orchestrator = import_module("app.agent.orchestrator")
-    try:
-        return getattr(orchestrator, name)
-    except AttributeError as err:
-        raise AttributeError(f"module 'app.agent' has no attribute {name!r}") from err
+    """按白名单延迟解析稳定 Agent 契约。"""
+    target = _PUBLIC_SYMBOLS.get(name)
+    if target is None:
+        raise AttributeError(f"module 'app.agent' has no attribute {name!r}")
+    module_name, symbol_name = target
+    value = getattr(import_module(module_name), symbol_name)
+    globals()[name] = value
+    return value
 
 
 def __dir__() -> list[str]:
-    """返回 Agent 包和编排模块共同提供的可发现名称。"""
-    orchestrator = import_module("app.agent.orchestrator")
-    return sorted(set(globals()) | set(dir(orchestrator)))
+    """返回包内名称与白名单公开契约。"""
+    return sorted(set(globals()) | set(_PUBLIC_SYMBOLS))

app/agent/lifecycle.py

@@ -0,0 +1,166 @@
+"""Agent 管理器启动、关闭与空闲回收生命周期。"""
+
+import asyncio
+
+from app.agent.session import AgentManagerUnavailableError, AgentSessionOwner
+from app.application.agent import AgentDataContext
+from app.runtime.log import logger
+
+
+class AgentLifecycleOwner(AgentSessionOwner):
+    """管理 Agent 服务代际、接收门禁与有界关闭。"""
+
+    def configure_data_context(self, data: AgentDataContext) -> None:
+        """在服务启动前绑定唯一数据上下文并配置记忆服务。"""
+        if self._data is data:
+            return
+        if self._accepting_tasks or self.active_agents:
+            raise RuntimeError("AgentManager 已运行，不能替换数据上下文")
+        self._data = data
+        self._memory.configure(data.chat, data.chat_persistence)
+
+    async def initialize(self) -> None:
+        """
+        初始化管理器
+        """
+        async with self._lifecycle_lock:
+            if self._accepting_tasks:
+                return
+            if self._close_finalizer_task and not self._close_finalizer_task.done():
+                raise AgentManagerUnavailableError("AgentManager 仍在完成上一代关闭")
+            self._memory.initialize()
+            if not self._idle_cleanup_task or self._idle_cleanup_task.done():
+                self._idle_cleanup_task = asyncio.create_task(
+                    self._cleanup_idle_sessions()
+                )
+            self._accepting_tasks = True
+            self._closed = False
+
+    async def close(self) -> bool:
+        """
+        关闭管理器，并诚实返回全部会话 owner 是否已经收敛。
+        """
+        async with self._lifecycle_lock:
+            if self._closed:
+                return True
+            if self._close_finalizer_task and not self._close_finalizer_task.done():
+                return False
+            # 门禁必须先关闭；锁内完成清理可阻止等待中的请求在收口期间重新入队。
+            self._accepting_tasks = False
+            # 子代理提交门禁同样必须在第一次 await 前关闭，否则 detached task
+            # 可趁 idle-cleanup 收尾窗口继续创建不属于新生命周期的任务。
+            for agent in self.active_agents.values():
+                begin_shutdown = getattr(agent, "begin_shutdown", None)
+                if callable(begin_shutdown):
+                    begin_shutdown()
+            if self._idle_cleanup_task:
+                self._idle_cleanup_task.cancel()
+                try:
+                    await self._idle_cleanup_task
+                except asyncio.CancelledError:
+                    pass
+                self._idle_cleanup_task = None
+            # 先取消所有 worker，再以有限等待收口，避免关闭阶段无限挂起。
+            workers = list(self._session_workers.items())
+            for session_id, task in workers:
+                self._session_cancel_requested.add(session_id)
+                task.cancel()
+            timed_out_workers = []
+            for session_id, task in workers:
+                stopped = await self._wait_for_worker_shutdown(
+                    session_id,
+                    task,
+                    reason="manager_close",
+                )
+                if not stopped:
+                    timed_out_workers.append((session_id, task))
+            for queue in list(self._session_queues.values()):
+                self._discard_queued_messages(
+                    queue,
+                    error=AgentManagerUnavailableError("AgentManager 已关闭"),
+                )
+            self._session_queues.clear()
+            self._session_last_used.clear()
+            self._session_queue_rejections.clear()
+            self._session_last_queue_wait_ms.clear()
+
+            if timed_out_workers:
+                timed_out_session_ids = {
+                    session_id for session_id, _ in timed_out_workers
+                }
+                for session_id, task in workers:
+                    if session_id in timed_out_session_ids:
+                        continue
+                    if self._session_workers.get(session_id) is task:
+                        self._session_workers.pop(session_id, None)
+                for session_id, agent in list(self.active_agents.items()):
+                    if session_id not in timed_out_session_ids:
+                        if await agent.cleanup() is not False:
+                            self.active_agents.pop(session_id, None)
+                logger.error(
+                    "AgentManager 关闭时仍有 worker 未收敛，"
+                    f"保留 {len(timed_out_workers)} 个会话资源直到 worker 结束"
+                )
+                self._close_finalizer_task = asyncio.create_task(
+                    self._finish_deferred_close(timed_out_workers)
+                )
+                return False
+
+            self._session_workers.clear()
+            for session_id, agent in list(self.active_agents.items()):
+                if await agent.cleanup() is not False:
+                    self.active_agents.pop(session_id, None)
+            if self.active_agents:
+                logger.error(
+                    "AgentManager 仍有 %d 个 detached 子代理 owner 未收敛",
+                    len(self.active_agents),
+                )
+                return False
+            await self._memory.close()
+            self._closed = True
+            return True
+
+    async def _cleanup_idle_sessions(self) -> None:
+        """
+        周期性清理长时间没有新消息的 Agent 会话，避免长期运行后实例持续累积。
+        """
+        while True:
+            try:
+                await asyncio.sleep(self._idle_cleanup_interval)
+                for session_id, user_id in self._expired_idle_sessions():
+                    await self.clear_session(session_id=session_id, user_id=user_id)
+                    logger.info(f"已清理空闲Agent会话: session_id={session_id}")
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"清理空闲Agent会话失败: {e}")
+
+    async def _finish_deferred_close(
+            self,
+            workers: list[tuple[str, asyncio.Task[None]]],
+    ) -> None:
+        """关闭超时后等待遗留 worker，再释放共享 Agent 资源。"""
+        try:
+            await asyncio.gather(
+                *(worker for _, worker in workers),
+                return_exceptions=True,
+            )
+            async with self._lifecycle_lock:
+                for session_id, worker in workers:
+                    if self._session_workers.get(session_id) is worker:
+                        self._session_workers.pop(session_id, None)
+                for session_id, agent in list(self.active_agents.items()):
+                    if await agent.cleanup() is not False:
+                        self.active_agents.pop(session_id, None)
+                self._session_shutdown_pending.clear()
+                self._session_cancel_requested.clear()
+                if self.active_agents:
+                    logger.error(
+                        "AgentManager 延迟关闭后仍有 %d 个子代理 owner 未收敛",
+                        len(self.active_agents),
+                    )
+                    return
+                await self._memory.close()
+                self._closed = True
+        finally:
+            self._close_finalizer_task = None

app/agent/manager.py

@@ -0,0 +1,27 @@
+"""AgentManager 稳定门面。"""
+
+from typing import Optional
+
+from app.agent.memory import MemoryManager
+from app.agent.tasks import AgentTaskOwner
+from app.application.agent import AgentDataContext
+
+
+class AgentManager(AgentTaskOwner):
+    """
+    AI 智能体管理门面。
+
+    会话、生命周期和后台任务分别由继承链中的单一 owner 实现，门面只保留
+    稳定构造身份与公开方法集合。
+    """
+
+    def __init__(
+        self,
+        data: Optional[AgentDataContext] = None,
+        memory: Optional[MemoryManager] = None,
+    ) -> None:
+        """创建稳定门面并初始化会话 owner。"""
+        super().__init__(data=data, memory=memory)
+
+
+agent_manager = AgentManager()

app/agent/memory/__init__.py

@@ -26,14 +26,14 @@ class MemoryManager:
         self,
         chat: AgentChatService | None = None,
         persistence: AgentChatPersistenceService | None = None,
-    ):
+    ) -> None:
         """创建记忆缓存，并保存组合根显式注入的会话能力。"""
         self._chat = chat
         self._persistence = persistence
         # 内存中的会话记忆缓存
         self.memory_cache: Dict[str, ConversationMemory] = {}
         # 内存缓存清理任务
-        self.cleanup_task: Optional[asyncio.Task] = None
+        self.cleanup_task: Optional[asyncio.Task[None]] = None
 
     def configure(
         self,
@@ -52,7 +52,7 @@ class MemoryManager:
         """返回显式注入写服务，兼容测试未装配时使用既有应用服务。"""
         return self._persistence or get_configured_agent_chat_persistence()
 
-    def initialize(self):
+    def initialize(self) -> None:
         """
         初始化记忆管理器
         """
@@ -68,7 +68,7 @@ class MemoryManager:
         except Exception as e:
             logger.warning(f"Redis连接失败，将使用内存存储: {e}")
 
-    async def close(self):
+    async def close(self) -> None:
         """
         关闭记忆管理器
         """
@@ -83,7 +83,7 @@ class MemoryManager:
         logger.info("对话记忆管理器已关闭")
 
     @staticmethod
-    def _get_memory_key(session_id: str, user_id: str):
+    def _get_memory_key(session_id: str, user_id: Optional[str]) -> str:
         """
         计算内存Key
         """
@@ -173,7 +173,7 @@ class MemoryManager:
 
     def save_agent_messages(
             self, session_id: str, user_id: str, messages: List[BaseMessage]
-    ):
+    ) -> None:
         """
         保存Agent消息到内存缓存与持久化会话表。
         """
@@ -216,7 +216,7 @@ class MemoryManager:
         except Exception as e:
             logger.debug(f"持久化Agent消息失败: {e}")
 
-    def save_memory(self, memory: ConversationMemory):
+    def save_memory(self, memory: ConversationMemory) -> None:
         """
         保存记忆到内存缓存
 
@@ -225,7 +225,7 @@ class MemoryManager:
         cache_key = self._get_memory_key(memory.session_id, memory.user_id)
         self.memory_cache[cache_key] = memory
 
-    def clear_memory(self, session_id: str, user_id: str):
+    def clear_memory(self, session_id: str, user_id: str) -> None:
         """
         清空会话记忆
         """
@@ -235,7 +235,7 @@ class MemoryManager:
 
         logger.info(f"会话记忆已清空: session_id={session_id}, user_id={user_id}")
 
-    async def _cleanup_expired_memories(self):
+    async def _cleanup_expired_memories(self) -> None:
         """
         清理内存中过期记忆的后台任务
 

app/agent/session.py

@@ -0,0 +1,837 @@
+"""Agent 会话队列、worker 与资源状态的唯一 owner。"""
+
+import asyncio
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Any, Awaitable, Callable, Dict, List, Optional, cast
+
+from app.agent.contracts import ReplyMode
+from app.agent.memory import MemoryManager, memory_manager
+from app.agent.orchestrator import MoviePilotAgent, _SessionUsageSnapshot
+from app.application.agent import AgentDataContext
+from app.chain.agent import AgentChain
+from app.runtime.execution import run_in_threadpool
+from app.runtime.log import logger
+from app.runtime.observability import record_metric
+from app.runtime.settings import get_runtime_setting
+from app.schemas.types import NotificationChannel
+
+
+def _agent_task_metric_type(source: object, channel: object = None) -> str:
+    """把 Agent 来源归一为交互、调度、后台三类稳定标签。"""
+    normalized = str(source or "").strip().lower()
+    if normalized in {"scheduler", "scheduled", "heartbeat", "agent_task"}:
+        return "scheduled"
+    if channel:
+        return "interactive"
+    return "background"
+
+
+def _finish_processing_status(
+    status: Optional[dict[str, Any]],
+    user_id: Optional[str] = None,
+) -> None:
+    """结束入站消息的渠道处理状态。"""
+    if not status:
+        return
+    AgentChain().finish_message_processing_status(
+        status=status,
+        userid=user_id,
+    )
+
+
+async def _async_start_processing_status(
+    task: "_MessageTask",
+) -> Optional[dict[str, Any]]:
+    """
+    在 Agent worker 中启动渠道处理状态。
+    渠道启动可能触发外部 API，同步实现需切到线程池避免阻塞事件循环。
+    """
+    if not task.channel:
+        return None
+
+    def _start() -> Optional[dict[str, Any]]:
+        """在线程池中通过统一 Chain 接口启动处理状态。"""
+        try:
+            return cast(
+                Optional[dict[str, Any]],
+                AgentChain().start_message_processing_status(
+                channel=NotificationChannel(task.channel),
+                source=task.source,
+                userid=task.user_id,
+                message_id=task.original_message_id,
+                chat_id=task.original_chat_id,
+                text=task.message,
+                ),
+            )
+        except Exception as err:
+            logger.debug(f"启动Agent消息处理状态失败: {err}")
+            return None
+
+    return cast(Optional[dict[str, Any]], await run_in_threadpool(_start))
+
+
+async def _async_finish_processing_status(
+        status: Optional[dict[str, Any]], user_id: Optional[str] = None
+) -> None:
+    """
+    在 Agent worker 中结束渠道处理状态。
+    渠道收口可能触发外部 API，同步实现需切到线程池避免阻塞事件循环。
+    """
+    if not status:
+        return
+    await run_in_threadpool(_finish_processing_status, status, user_id)
+
+
+
+@dataclass
+class _MessageTask:
+    """
+    待处理的消息任务
+    """
+
+    session_id: str
+    user_id: str
+    message: str
+    images: Optional[List[str]] = None
+    files: Optional[List[dict[str, Any]]] = None
+    has_audio_input: bool = False
+    channel: Optional[str] = None
+    source: Optional[str] = None
+    username: Optional[str] = None
+    is_channel_admin: Optional[bool] = None
+    original_message_id: Optional[str] = None
+    original_chat_id: Optional[str] = None
+    processing_status: Optional[dict[str, Any]] = None
+    reply_mode: ReplyMode = ReplyMode.DISPATCH
+    allow_message_tools: bool = True
+    output_callback: Optional[Callable[[str], None]] = None
+    protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None
+    message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None
+    agent_factory: Optional[Callable[..., MoviePilotAgent]] = None
+    agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None
+    completion_future: Optional[asyncio.Future[str]] = None
+    enqueued_at: Optional[float] = None
+
+
+class AgentManagerUnavailableError(RuntimeError):
+    """AgentManager 未运行或已开始关闭，不能再接收新任务。"""
+
+    code = "agent_manager_unavailable"
+
+
+class AgentManagerQueueFullError(RuntimeError):
+    """Agent 会话的待处理消息达到容量上限。"""
+
+    code = "agent_manager_queue_full"
+
+    def __init__(self, session_id: str, limit: int) -> None:
+        self.session_id = session_id
+        self.limit = limit
+        super().__init__(
+            f"Agent 会话当前排队消息已达上限（{limit} 条），请稍后重试"
+        )
+
+
+AGENT_SESSION_QUEUE_MAX_SIZE = 8
+AGENT_MANAGER_SHUTDOWN_TIMEOUT = 10.0
+
+
+class AgentSessionOwner:
+    """管理 Agent 会话队列、worker、活动实例与清理状态。"""
+
+    def __init__(
+        self,
+        data: Optional[AgentDataContext] = None,
+        memory: Optional[MemoryManager] = None,
+    ) -> None:
+        """创建会话 owner，并保存组合根注入的 Agent 数据与记忆能力。"""
+        self._data = data
+        self._memory = memory or memory_manager
+        if data is not None:
+            self._memory.configure(data.chat, data.chat_persistence)
+        self.active_agents: Dict[str, MoviePilotAgent] = {}
+        # 每个会话的消息队列
+        self._session_queues: Dict[str, asyncio.Queue[_MessageTask]] = {}
+        # 每个会话的worker任务
+        self._session_workers: Dict[str, asyncio.Task[None]] = {}
+        # 每个会话最后活动时间，用于回收空闲 Agent 实例
+        self._session_last_used: Dict[str, tuple[str, datetime]] = {}
+        self._idle_cleanup_task: Optional[asyncio.Task[None]] = None
+        self._idle_session_ttl = timedelta(hours=24)
+        self._idle_cleanup_interval = 60 * 60
+        self._session_queue_rejections: Dict[str, int] = {}
+        self._session_last_queue_wait_ms: Dict[str, float] = {}
+        self._session_shutdown_pending: Dict[str, asyncio.Task[None]] = {}
+        self._session_cleanup_pending: set[str] = set()
+        self._session_deferred_cleanup_tasks: Dict[str, asyncio.Task[None]] = {}
+        self._session_cancel_requested: set[str] = set()
+        self._close_finalizer_task: Optional[asyncio.Task[None]] = None
+        self._closed = False
+        self._shutdown_timeout = AGENT_MANAGER_SHUTDOWN_TIMEOUT
+        # 接收门禁与队列写入共用一把锁，确保关闭开始后不会再创建 worker。
+        self._lifecycle_lock = asyncio.Lock()
+        self._accepting_tasks = False
+
+    def get_session_status(self, session_id: str) -> dict[str, Any]:
+        """获取会话当前模型与 token 使用状态。"""
+        agent = self.active_agents.get(session_id)
+        if agent:
+            status = agent.get_session_status()
+        else:
+            status = _SessionUsageSnapshot(
+                model=get_runtime_setting('LLM_MODEL'),
+                context_window_tokens=(
+                    get_runtime_setting('LLM_MAX_CONTEXT_TOKENS') * 1000
+                    if get_runtime_setting('LLM_MAX_CONTEXT_TOKENS')
+                    else None
+                ),
+            ).to_dict(session_id)
+
+        queue = self._session_queues.get(session_id)
+        status["pending_messages"] = queue.qsize() if queue else 0
+        status["queue_capacity"] = AGENT_SESSION_QUEUE_MAX_SIZE
+        status["queue_saturated"] = bool(queue and queue.full())
+        status["queue_rejections"] = self._session_queue_rejections.get(
+            session_id,
+            0,
+        )
+        status["last_queue_wait_ms"] = self._session_last_queue_wait_ms.get(
+            session_id,
+            0.0,
+        )
+        pending_shutdown = self._session_shutdown_pending.get(session_id)
+        status["shutdown_pending"] = bool(
+            pending_shutdown and not pending_shutdown.done()
+        )
+        status["is_processing"] = (
+                session_id in self._session_workers
+                and not self._session_workers[session_id].done()
+        )
+        return status
+
+    def matches_secret_confirmation(
+            self,
+            session_id: str,
+            user_id: str,
+            channel: Optional[str] = None,
+            source: Optional[str] = None,
+    ) -> bool:
+        """判断指定用户是否可继续当前会话的敏感设置确认。"""
+        agent = self.active_agents.get(session_id)
+        pending = agent._pending_secret_confirmation if agent else None
+        return bool(
+            agent
+            and pending
+            and str(agent.user_id) == str(user_id)
+            and (channel is None or pending.channel == str(channel))
+            and (source is None or pending.source == str(source))
+        )
+
+    def _record_session_activity(self, session_id: str, user_id: str) -> None:
+        """
+        记录会话最近活动时间，供空闲会话清理任务判断是否可释放资源。
+        """
+        self._session_last_used[session_id] = (user_id, datetime.now())
+
+    def _is_session_busy(self, session_id: str) -> bool:
+        """
+        判断会话是否仍有正在执行的 worker 或待处理消息，避免误清理活跃会话。
+        """
+        worker = self._session_workers.get(session_id)
+        if worker and not worker.done():
+            return True
+        queue = self._session_queues.get(session_id)
+        return bool(queue and not queue.empty())
+
+    def is_session_busy(self, session_id: str) -> bool:
+        """
+        查询会话是否仍有正在执行或排队的任务。
+        """
+        return self._is_session_busy(session_id)
+
+    def _expired_idle_sessions(self) -> list[tuple[str, str]]:
+        """
+        收集已经超过空闲时间且当前不忙的会话。
+        """
+        expire_before = datetime.now() - self._idle_session_ttl
+        expired = []
+        for session_id, (user_id, last_used) in list(self._session_last_used.items()):
+            if last_used < expire_before and not self._is_session_busy(session_id):
+                expired.append((session_id, user_id))
+        return expired
+
+    async def process_message(
+            self,
+            session_id: str,
+            user_id: str,
+            message: str,
+            images: Optional[List[str]] = None,
+            files: Optional[List[dict[str, Any]]] = None,
+            has_audio_input: bool = False,
+            channel: Optional[str] = None,
+            source: Optional[str] = None,
+            username: Optional[str] = None,
+            is_channel_admin: Optional[bool] = None,
+            original_message_id: Optional[str] = None,
+            original_chat_id: Optional[str] = None,
+            reply_mode: ReplyMode = ReplyMode.DISPATCH,
+            allow_message_tools: bool = True,
+            output_callback: Optional[Callable[[str], None]] = None,
+            protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None,
+            message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None,
+            agent_factory: Optional[Callable[..., MoviePilotAgent]] = None,
+            agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None,
+            wait_for_completion: bool = False,
+    ) -> str:
+        """
+        处理用户消息：将消息放入会话队列，按顺序依次处理。
+        同一会话的消息排队等待，不同会话之间互不影响。
+        """
+        completion_future: Optional[asyncio.Future[str]] = (
+            asyncio.get_running_loop().create_future() if wait_for_completion else None
+        )
+        task = _MessageTask(
+            session_id=session_id,
+            user_id=user_id,
+            message=message,
+            images=images,
+            files=files,
+            has_audio_input=has_audio_input,
+            channel=channel,
+            source=source,
+            username=username,
+            is_channel_admin=is_channel_admin,
+            original_message_id=original_message_id,
+            original_chat_id=original_chat_id,
+            reply_mode=reply_mode,
+            allow_message_tools=allow_message_tools,
+            output_callback=output_callback,
+            protected_output_callback=protected_output_callback,
+            message_callback=message_callback,
+            agent_factory=agent_factory,
+            agent_setup=agent_setup,
+            completion_future=completion_future,
+        )
+        async with self._lifecycle_lock:
+            if not self._accepting_tasks:
+                if completion_future and not completion_future.done():
+                    completion_future.cancel()
+                raise AgentManagerUnavailableError("AgentManager 未运行或已关闭")
+            pending_shutdown = self._session_shutdown_pending.get(session_id)
+            if pending_shutdown:
+                if pending_shutdown.done():
+                    self._session_shutdown_pending.pop(session_id, None)
+                else:
+                    if completion_future and not completion_future.done():
+                        completion_future.cancel()
+                    raise AgentManagerUnavailableError(
+                        f"Agent 会话 {session_id} 仍在停止，暂时不能接收新任务"
+                    )
+            self._record_session_activity(session_id, user_id)
+
+            # 获取或创建会话队列
+            if session_id not in self._session_queues:
+                self._session_queues[session_id] = asyncio.Queue(
+                    maxsize=AGENT_SESSION_QUEUE_MAX_SIZE
+                )
+
+            queue = self._session_queues[session_id]
+            queue_size = queue.qsize()
+
+            if queue.full():
+                self._session_queue_rejections[session_id] = (
+                    self._session_queue_rejections.get(session_id, 0) + 1
+                )
+                logger.warning(
+                    f"会话 {session_id} 的 Agent 排队已满，拒绝新消息 "
+                    f"(上限: {AGENT_SESSION_QUEUE_MAX_SIZE})"
+                )
+                if completion_future and not completion_future.done():
+                    completion_future.cancel()
+                raise AgentManagerQueueFullError(
+                    session_id=session_id,
+                    limit=AGENT_SESSION_QUEUE_MAX_SIZE,
+                )
+
+            # 如果队列中已有等待的消息，通知用户消息已排队
+            if queue_size > 0 or (
+                    session_id in self._session_workers
+                    and not self._session_workers[session_id].done()
+            ):
+                logger.info(
+                    f"会话 {session_id} 有任务正在处理，消息已排队等待 "
+                    f"(队列中待处理: {queue_size} 条)"
+                )
+
+            # 非阻塞入队与 worker 创建在同一生命周期锁内完成，关闭期间不会留下悬挂入队。
+            task.enqueued_at = asyncio.get_running_loop().time()
+            queue.put_nowait(task)
+            if (
+                    session_id not in self._session_workers
+                    or self._session_workers[session_id].done()
+            ):
+                self._session_workers[session_id] = asyncio.create_task(
+                    self._session_worker(session_id)
+                )
+
+        if completion_future:
+            return await completion_future
+        return ""
+
+    async def _session_worker(self, session_id: str) -> None:
+        """
+        会话消息处理worker：从队列中逐条取出消息并处理。
+        处理完当前消息后才会处理下一条，确保同一会话的消息顺序执行。
+        """
+        queue = self._session_queues.get(session_id)
+        if not queue:
+            return
+
+        try:
+            while True:
+                try:
+                    # 等待消息，超时后自动退出worker
+                    task = await asyncio.wait_for(queue.get(), timeout=60.0)
+                except asyncio.TimeoutError:
+                    # 超时回调与入队可能在同一轮事件循环就绪；已有消息时继续消费，
+                    # 避免旧 worker 退出后留下没有消费者的非空队列。
+                    if not queue.empty():
+                        continue
+                    # 队列空闲超时，退出worker
+                    logger.debug(f"会话 {session_id} 的消息队列空闲，worker退出")
+                    break
+
+                task_type = _agent_task_metric_type(task.source, task.channel)
+                active_metric_recorded = False
+                try:
+                    if task.enqueued_at is not None:
+                        queue_wait_ms = max(
+                            0.0,
+                            (
+                                asyncio.get_running_loop().time()
+                                - task.enqueued_at
+                            )
+                            * 1000,
+                        )
+                        self._session_last_queue_wait_ms[session_id] = round(
+                            queue_wait_ms,
+                            3,
+                        )
+                    await self._start_task_processing_status(task)
+                    record_metric(
+                        "agent.active_tasks",
+                        1,
+                        task_type=task_type,
+                    )
+                    active_metric_recorded = True
+                    result = await self._process_message_internal(task)
+                    if task.completion_future and not task.completion_future.done():
+                        if (
+                                not self._accepting_tasks
+                                or session_id in self._session_cancel_requested
+                        ):
+                            task.completion_future.cancel()
+                        else:
+                            task.completion_future.set_result(result)
+                except asyncio.CancelledError:
+                    if task.completion_future and not task.completion_future.done():
+                        if self._accepting_tasks:
+                            task.completion_future.cancel()
+                        else:
+                            task.completion_future.set_exception(
+                                AgentManagerUnavailableError("AgentManager 已关闭")
+                            )
+                    raise
+                except Exception as e:
+                    logger.error(f"处理会话 {session_id} 的消息失败: {e}")
+                    if task.completion_future and not task.completion_future.done():
+                        task.completion_future.set_exception(e)
+                finally:
+                    if active_metric_recorded:
+                        record_metric(
+                            "agent.active_tasks",
+                            -1,
+                            task_type=task_type,
+                        )
+                    await self._finish_task_processing_status(task)
+                    queue.task_done()
+                if session_id in self._session_cancel_requested:
+                    break
+
+        except asyncio.CancelledError:
+            logger.info(f"会话 {session_id} 的worker被取消")
+        finally:
+            # 清理已完成的worker记录
+            current_worker = asyncio.current_task()
+            if self._session_workers.get(session_id) is current_worker:
+                self._session_workers.pop(session_id, None)  # noqa
+            self._session_cancel_requested.discard(session_id)
+            # 如果队列为空，清理队列
+            if (
+                    self._session_queues.get(session_id) is queue
+                    and queue.empty()
+            ):
+                self._session_queues.pop(session_id, None)
+
+    @staticmethod
+    def _discard_queued_messages(
+            queue: asyncio.Queue[_MessageTask],
+            error: Optional[Exception] = None,
+    ) -> None:
+        """丢弃会话队列时同步结束等待任务完成的调用方。"""
+        while not queue.empty():
+            try:
+                task = queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+            if task.completion_future and not task.completion_future.done():
+                if error is None:
+                    task.completion_future.cancel()
+                else:
+                    task.completion_future.set_exception(error)
+            queue.task_done()
+
+    @staticmethod
+    async def _start_task_processing_status(task: _MessageTask) -> None:
+        """
+        在 Agent worker 真正开始处理消息时启动渠道处理状态。
+        """
+        if task.processing_status:
+            return
+        task.processing_status = await _async_start_processing_status(task)
+
+    @staticmethod
+    async def _finish_task_processing_status(task: _MessageTask) -> None:
+        """
+        在 Agent worker 完成或异常后结束本条消息的渠道处理状态。
+        """
+        await _async_finish_processing_status(task.processing_status, task.user_id)
+        task.processing_status = None
+
+    async def _process_message_internal(self, task: _MessageTask) -> str:
+        """
+        实际处理单条消息
+        """
+        session_id = task.session_id
+        existing_agent = self.active_agents.get(session_id)
+        if (
+                existing_agent
+                and task.agent_factory
+                and isinstance(task.agent_factory, type)
+                and not isinstance(existing_agent, task.agent_factory)
+        ):
+            if await existing_agent.cleanup() is False:
+                raise AgentManagerUnavailableError(
+                    f"Agent 会话 {session_id} 仍有子代理任务在停止"
+                )
+            self.active_agents.pop(session_id, None)
+
+        if session_id not in self.active_agents:
+            logger.info(
+                f"创建新的AI智能体实例，session_id: {session_id}, user_id: {task.user_id}"
+            )
+            if task.agent_factory is None:
+                agent = MoviePilotAgent(
+                    session_id=session_id,
+                    user_id=task.user_id,
+                    channel=task.channel,
+                    source=task.source,
+                    username=task.username,
+                    is_channel_admin=task.is_channel_admin,
+                    original_message_id=task.original_message_id,
+                    original_chat_id=task.original_chat_id,
+                    replay_mode=task.reply_mode,
+                    allow_message_tools=task.allow_message_tools,
+                    output_callback=task.output_callback,
+                    protected_output_callback=task.protected_output_callback,
+                    data=self._data,
+                    memory=self._memory,
+                )
+            else:
+                agent_kwargs: dict[str, Any] = {
+                    "session_id": session_id,
+                    "user_id": task.user_id,
+                    "channel": task.channel,
+                    "source": task.source,
+                    "username": task.username,
+                    "is_channel_admin": task.is_channel_admin,
+                    "original_message_id": task.original_message_id,
+                    "original_chat_id": task.original_chat_id,
+                    "replay_mode": task.reply_mode,
+                    "allow_message_tools": task.allow_message_tools,
+                    "output_callback": task.output_callback,
+                    "protected_output_callback": task.protected_output_callback,
+                }
+                if task.message_callback is not None:
+                    agent_kwargs["message_callback"] = task.message_callback
+                agent = task.agent_factory(**agent_kwargs)
+            self.active_agents[session_id] = agent
+        else:
+            agent = self.active_agents[session_id]
+            agent.user_id = task.user_id
+            # 每条队列任务都携带完整消息上下文，None 也必须覆盖，避免后台任务
+            # 复用会话 Agent 时继续沿用上一条入站消息的渠道。
+            agent.channel = task.channel
+            agent.source = task.source
+            agent.username = task.username
+            agent.is_channel_admin = task.is_channel_admin
+            agent.original_message_id = task.original_message_id
+            agent.original_chat_id = task.original_chat_id
+            agent.reply_mode = task.reply_mode
+            agent.allow_message_tools = task.allow_message_tools
+            if hasattr(agent, "set_output_callback"):
+                agent.set_output_callback(task.output_callback)
+            else:
+                agent.output_callback = task.output_callback
+            agent.set_protected_output_callback(task.protected_output_callback)
+            if task.message_callback is not None and hasattr(agent, "set_message_callback"):
+                agent.set_message_callback(task.message_callback)
+
+        if task.agent_setup is not None:
+            task.agent_setup(agent)
+
+        process_kwargs: dict[str, Any] = {
+            "images": task.images,
+            "files": task.files,
+        }
+        if task.has_audio_input:
+            process_kwargs["has_audio_input"] = True
+        return await agent.process(task.message, **process_kwargs)
+
+    async def stop_current_task(self, session_id: str) -> bool:
+        """
+        应急停止当前正在执行的Agent推理任务，但保留会话和记忆。
+        与 clear_session 不同，此方法不会销毁Agent实例或清除记忆，
+        用户可以在停止后继续对话。
+        """
+        async with self._lifecycle_lock:
+            return await self._stop_current_task_locked(session_id)
+
+    async def _stop_current_task_locked(self, session_id: str) -> bool:
+        """在 lifecycle 互斥域内停止会话 worker。"""
+        stopped = False
+        active_agent = self.active_agents.get(session_id)
+        task_type = (
+            _agent_task_metric_type(
+                getattr(active_agent, "source", None),
+                getattr(active_agent, "channel", None),
+            )
+            if active_agent
+            else "unknown"
+        )
+
+        worker = self._session_workers.get(session_id)
+        queue = self._session_queues.get(session_id)
+        if queue and self._session_queues.get(session_id) is queue:
+            self._session_queues.pop(session_id, None)
+
+        # 先摘下旧队列再等待 worker 退出；lifecycle 锁保证清理期间不会并发建立新队列。
+        if worker:
+            self._session_cancel_requested.add(session_id)
+            worker.cancel()
+        if queue:
+            self._discard_queued_messages(queue)
+        if worker:
+            stopped_cleanly = await self._wait_for_worker_shutdown(
+                session_id,
+                worker,
+                reason="stop_current_task",
+            )
+            if stopped_cleanly and self._session_workers.get(session_id) is worker:
+                self._session_workers.pop(session_id, None)  # noqa
+            stopped = True
+        if queue:
+            stopped = True
+
+        new_queue = self._session_queues.get(session_id)
+        current_worker = self._session_workers.get(session_id)
+        if (
+                new_queue
+                and not new_queue.empty()
+                and (not current_worker or current_worker.done())
+        ):
+            if session_id not in self._session_shutdown_pending:
+                self._session_workers[session_id] = asyncio.create_task(
+                    self._session_worker(session_id)
+                )
+
+        if stopped:
+            logger.info(f"会话 {session_id} 的Agent推理已应急停止")
+        else:
+            logger.debug(f"会话 {session_id} 没有正在执行的Agent任务")
+
+        record_metric(
+            "agent.cancel",
+            task_type=task_type,
+            outcome="stopped" if stopped else "not_found",
+        )
+
+        return stopped
+
+    async def clear_session(self, session_id: str, user_id: str) -> None:
+        """
+        清空会话
+        """
+        async with self._lifecycle_lock:
+            await self._clear_session_locked(session_id=session_id, user_id=user_id)
+
+    async def _clear_session_locked(self, session_id: str, user_id: str) -> None:
+        """在 lifecycle 互斥域内释放会话、Agent 与记忆。"""
+        if session_id in self._session_cleanup_pending:
+            return
+        self._session_last_used.pop(session_id, None)
+        # 取消该会话的worker
+        if session_id in self._session_workers:
+            worker = self._session_workers[session_id]
+            self._session_cleanup_pending.add(session_id)
+            self._session_cancel_requested.add(session_id)
+            worker.cancel()
+            try:
+                stopped_cleanly = await self._wait_for_worker_shutdown(
+                    session_id,
+                    worker,
+                    reason="clear_session",
+                )
+            except asyncio.CancelledError:
+                self._defer_session_cleanup(session_id, user_id, worker)
+                raise
+            if not stopped_cleanly:
+                self._defer_session_cleanup(session_id, user_id, worker)
+                return
+            if self._session_workers.get(session_id) is worker:
+                self._session_workers.pop(session_id, None)  # noqa
+            self._session_cleanup_pending.discard(session_id)
+
+        # 清理队列时同步结束未执行请求，避免 wait_for_completion 调用方永久等待。
+        queue = self._session_queues.pop(session_id, None)
+        if queue:
+            self._discard_queued_messages(queue)
+        self._session_queue_rejections.pop(session_id, None)
+        self._session_last_queue_wait_ms.pop(session_id, None)
+
+        # 清理agent
+        if session_id in self.active_agents:
+            agent = self.active_agents[session_id]
+            if await agent.cleanup() is False:
+                logger.error(
+                    f"会话 {session_id} 仍有子代理 owner 未收敛，保留会话与记忆"
+                )
+                return
+            del self.active_agents[session_id]
+            self._memory.clear_memory(session_id, user_id)
+            logger.info(f"会话 {session_id} 的记忆已清空")
+
+    def _defer_session_cleanup(
+            self,
+            session_id: str,
+            user_id: str,
+            worker: asyncio.Task[None],
+    ) -> None:
+        """把中断或超时的清理转交给 worker 终态回调。"""
+        queue = self._session_queues.pop(session_id, None)
+        if queue:
+            self._discard_queued_messages(queue)
+        self._session_shutdown_pending[session_id] = worker
+        worker.add_done_callback(
+            lambda done: self._schedule_deferred_session_cleanup(
+                session_id,
+                user_id,
+                done,
+            )
+        )
+
+    def _schedule_deferred_session_cleanup(
+            self,
+            session_id: str,
+            user_id: str,
+            worker: asyncio.Task[None],
+    ) -> None:
+        """worker 取得终态后释放会话资源，避免与仍在运行的 Agent 竞态。"""
+        if self._session_shutdown_pending.get(session_id) is worker:
+            existing = self._session_deferred_cleanup_tasks.get(session_id)
+            if existing is not None and not existing.done():
+                return
+            cleanup_task = asyncio.create_task(
+                self._finish_deferred_session_cleanup(
+                    session_id=session_id,
+                    user_id=user_id,
+                    worker=worker,
+                )
+            )
+            self._session_deferred_cleanup_tasks[session_id] = cleanup_task
+
+    async def _finish_deferred_session_cleanup(
+            self,
+            session_id: str,
+            user_id: str,
+            worker: asyncio.Task[None],
+    ) -> None:
+        """等待超时 worker 真正结束后，再完成 clear_session 的资源释放。"""
+        try:
+            await worker
+        except BaseException:
+            pass
+        async with self._lifecycle_lock:
+            if self._session_workers.get(session_id) is worker:
+                self._session_workers.pop(session_id, None)
+            self._session_shutdown_pending.pop(session_id, None)
+            self._session_cleanup_pending.discard(session_id)
+            self._session_deferred_cleanup_tasks.pop(session_id, None)
+            self._session_queue_rejections.pop(session_id, None)
+            self._session_last_queue_wait_ms.pop(session_id, None)
+            agent = self.active_agents.get(session_id)
+            if agent:
+                if await agent.cleanup() is False:
+                    logger.error(
+                        f"会话 {session_id} 的延迟清理仍有子代理 owner 未收敛"
+                    )
+                    return
+                self.active_agents.pop(session_id, None)
+                self._memory.clear_memory(session_id, user_id)
+                logger.info(f"会话 {session_id} 的记忆已清空")
+
+    async def _wait_for_worker_shutdown(
+            self,
+            session_id: str,
+            worker: asyncio.Task[None],
+            *,
+            reason: str,
+    ) -> bool:
+        """有限等待 worker 结束，超时会话保持停止态直到旧 worker 收敛。"""
+        try:
+            await asyncio.wait_for(
+                asyncio.shield(worker),
+                timeout=self._shutdown_timeout,
+            )
+            return True
+        except asyncio.CancelledError:
+            current = asyncio.current_task()
+            if worker.cancelled() and (current is None or not current.cancelling()):
+                return True
+            raise
+        except asyncio.TimeoutError:
+            self._session_shutdown_pending[session_id] = worker
+
+            def _clear_pending(done: asyncio.Task[None]) -> None:
+                if (
+                        self._session_shutdown_pending.get(session_id) is done
+                        and session_id not in self._session_cleanup_pending
+                ):
+                    self._session_shutdown_pending.pop(session_id, None)
+                    logger.info(
+                        f"会话 {session_id} 的 Agent worker 已在超时后收敛"
+                    )
+
+            worker.add_done_callback(_clear_pending)
+            logger.error(
+                f"会话 {session_id} 的 Agent worker 关闭超时，"
+                f"已阻止新任务进入，reason={reason}, timeout={self._shutdown_timeout:g}s"
+            )
+            return False
+        except Exception as error:
+            logger.error(
+                f"等待会话 {session_id} 的 Agent worker 关闭失败，"
+                f"reason={reason}: {error}"
+            )
+            return True

app/agent/tasks.py

@@ -0,0 +1,208 @@
+"""Agent 后台提示、持久化定时任务与心跳执行 owner。"""
+
+import asyncio
+import uuid
+from typing import Callable, Optional
+
+from app.agent.contracts import ReplyMode
+from app.agent.lifecycle import AgentLifecycleOwner
+from app.agent.middleware.jobs import filter_active_jobs, load_jobs_metadata
+from app.agent.orchestrator import AGENT_EXECUTION_ERROR_PREFIX, HEARTBEAT_SESSION_PREFIX
+from app.agent.prompt import prompt_manager
+from app.agent.runtime import agent_runtime_manager
+from app.application.agenttask import get_agent_task_execution_service
+from app.chain.agent import AgentChain
+from app.foundation.identity import SYSTEM_INTERNAL_USER_ID
+from app.runtime.log import logger
+from app.runtime.settings import get_runtime_setting
+from app.schemas.message import Message
+from app.schemas.types import MessageType
+
+
+class AgentTaskOwner(AgentLifecycleOwner):
+    """执行不绑定交互请求的 Agent 后台与调度任务。"""
+
+    async def run_background_prompt(
+            self,
+            message: str,
+            session_prefix: str = "__agent_background",
+            output_callback: Optional[Callable[[str], None]] = None,
+            reply_mode: ReplyMode = ReplyMode.CAPTURE_ONLY,
+            allow_message_tools: Optional[bool] = None,
+    ) -> None:
+        """
+        以独立后台会话执行一段 prompt。
+        """
+        session_id = f"{session_prefix}_{uuid.uuid4().hex[:8]}__"
+        user_id = SYSTEM_INTERNAL_USER_ID
+
+        if reply_mode == ReplyMode.CAPTURE_ONLY:
+            allow_message_tools = False
+        elif allow_message_tools is None:
+            allow_message_tools = True
+
+        try:
+            await self.process_message(
+                session_id=session_id,
+                user_id=user_id,
+                message=message,
+                channel=None,
+                source=None,
+                username=get_runtime_setting('SUPERUSER'),
+                reply_mode=reply_mode,
+                output_callback=output_callback,
+                allow_message_tools=allow_message_tools,
+                wait_for_completion=True,
+            )
+        finally:
+            await self.clear_session(session_id=session_id, user_id=user_id)
+
+    async def execute_scheduled_task(
+            self,
+            task_id: int,
+            trigger_source: str = "scheduled",
+            scheduler_generation: int | None = None,
+            remove_schedule: Callable[[int, int, str], bool] | None = None,
+    ) -> tuple[bool, str]:
+        """
+        按持久化上下文唤醒 Agent 执行自主定时任务并向用户回传结果。
+
+        :param task_id: Agent 定时任务 ID
+        :param trigger_source: 触发入口，scheduled-自动调度，manual-显式立即执行
+        :return: 执行是否成功及结果摘要
+        """
+        if not get_runtime_setting('AI_AGENT_ENABLE'):
+            return False, "AI Agent 未启用"
+        accepting_before_claim = self._accepting_tasks
+        task_service = get_agent_task_execution_service()
+        claim = await task_service.claim(
+            task_id=task_id,
+            trigger_source=trigger_source,
+            scheduler_generation=scheduler_generation,
+            remove_schedule=remove_schedule,
+        )
+        run = claim.run
+        if run is None:
+            return False, claim.rejection or "Agent 定时任务当前不可执行"
+
+        trigger_description = (
+            "已手动触发" if run.trigger_source == "manual" else "已按计划触发"
+        )
+        task_message = (
+            f"定时任务{trigger_description}。请立即完成下面的任务，不要只确认收到，"
+            f"也不要重复创建同一个定时任务。\n\n"
+            f"任务名称：{run.name}\n"
+            f"任务内容：{run.content}\n\n"
+            "完成后请直接向用户发送消息报告本次执行结果；如果无法完成，也需发送消息说明原因。"
+        )
+        success = True
+        result = ""
+        notification_username = run.username or get_runtime_setting('SUPERUSER')
+        try:
+            result = await self.process_message(
+                session_id=run.session_id,
+                user_id=run.user_id,
+                message=task_message,
+                channel=None,
+                source=None,
+                username=notification_username,
+                original_chat_id=None,
+                reply_mode=ReplyMode.DISPATCH,
+                allow_message_tools=True,
+                wait_for_completion=True,
+            )
+            result_text = str(result or "").strip()
+            success = not result_text.startswith(
+                (AGENT_EXECUTION_ERROR_PREFIX, "处理消息时发生错误")
+            )
+        except asyncio.CancelledError:
+            success = False
+            result = "Agent 定时任务已取消"
+            raise
+        except Exception as err:
+            success = False
+            error_message = str(err)
+            if (
+                accepting_before_claim
+                and not self._accepting_tasks
+                and error_message == "AgentManager 未运行或已关闭"
+            ):
+                error_message = "AgentManager 已关闭"
+            result = f"Agent 定时任务执行失败：{error_message}"
+            logger.error(f"Agent 定时任务 {task_id} 执行失败: {str(err)}")
+            await AgentChain().async_post_message(
+                Message(
+                    mtype=MessageType.Agent,
+                    username=notification_username,
+                    title=f"定时任务执行失败：{run.name}",
+                    text=result,
+                    save_history=False,
+                )
+            )
+        finally:
+            await task_service.finalize(
+                run,
+                success=success,
+                result=str(result or ""),
+                scheduler_generation=scheduler_generation,
+                remove_schedule=remove_schedule,
+            )
+
+        return success, str(result or "任务执行完成")
+
+    @staticmethod
+    def _build_heartbeat_prompt() -> str:
+        """使用程序内置 System Tasks 定义构建心跳任务提示词。"""
+        return prompt_manager.render_system_task_message("heartbeat")
+
+    async def heartbeat_check_jobs(self) -> None:
+        """
+        心跳唤醒：检查并执行待处理的定时任务（Jobs）。
+        由定时调度器周期性调用，每次使用独立的会话避免上下文干扰。
+        """
+        try:
+            active_jobs = filter_active_jobs(
+                await load_jobs_metadata([str(agent_runtime_manager.jobs_dir)])
+            )
+            # 先在本地判断是否存在活跃任务。没有任务时直接短路，避免一次完整
+            # 的后台 Agent/LLM 空调用。
+            if not active_jobs:
+                logger.info("智能体心跳唤醒：没有活跃任务，跳过模型调用")
+                return
+
+            # 每次使用唯一的 session_id，避免共享上下文
+            session_id = f"{HEARTBEAT_SESSION_PREFIX}{uuid.uuid4().hex[:12]}__"
+            user_id = SYSTEM_INTERNAL_USER_ID
+
+            logger.info("智能体心跳唤醒：开始检查待处理任务...")
+            heartbeat_message = self._build_heartbeat_prompt()
+
+            await self.process_message(
+                session_id=session_id,
+                user_id=user_id,
+                message=heartbeat_message,
+                channel=None,
+                source=None,
+                username=get_runtime_setting('SUPERUSER'),
+                reply_mode=ReplyMode.CAPTURE_ONLY,
+                allow_message_tools=True,
+            )
+
+            # 等待消息队列处理完成
+            if session_id in self._session_queues:
+                await self._session_queues[session_id].join()
+
+            # 等待worker结束
+            if session_id in self._session_workers:
+                try:
+                    await self._session_workers[session_id]
+                except asyncio.CancelledError:
+                    pass
+
+            logger.info("智能体心跳唤醒：任务检查完成")
+
+            # 心跳会话用完即弃，清理资源
+            await self.clear_session(session_id, user_id)
+
+        except Exception as e:
+            logger.error(f"智能体心跳唤醒失败: {e}")

app/runtime/compat/manifest.py

@@ -826,16 +826,61 @@ SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = {
         for name in ("TransferTask", "TransferQueue")
     },
     "app.agent.orchestrator": {
+        "AGENT_SESSION_QUEUE_MAX_SIZE": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="AGENT_SESSION_QUEUE_MAX_SIZE",
+            replacement="app.agent.session.AGENT_SESSION_QUEUE_MAX_SIZE",
+        ),
         "AgentChain": SymbolAlias(
             target_module="app.chain.agent",
             target_name="AgentChain",
             replacement="app.chain.agent.AgentChain",
         ),
+        "AgentManager": SymbolAlias(
+            target_module="app.agent.manager",
+            target_name="AgentManager",
+            replacement="app.agent.manager.AgentManager",
+        ),
+        "AgentManagerQueueFullError": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="AgentManagerQueueFullError",
+            replacement="app.agent.session.AgentManagerQueueFullError",
+        ),
+        "AgentManagerUnavailableError": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="AgentManagerUnavailableError",
+            replacement="app.agent.session.AgentManagerUnavailableError",
+        ),
         "ReplyMode": SymbolAlias(
             target_module="app.schemas.types",
             target_name="ReplyMode",
             replacement="app.schemas.types.ReplyMode",
         ),
+        "_MessageTask": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="_MessageTask",
+            replacement="app.agent.session._MessageTask",
+        ),
+        "_async_finish_processing_status": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="_async_finish_processing_status",
+            replacement="app.agent.session._async_finish_processing_status",
+        ),
+        "_async_start_processing_status": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="_async_start_processing_status",
+            replacement="app.agent.session._async_start_processing_status",
+        ),
+        "_finish_processing_status": SymbolAlias(
+            target_module="app.agent.session",
+            target_name="_finish_processing_status",
+            replacement="app.agent.session._finish_processing_status",
+        ),
+        "agent_manager": SymbolAlias(
+            target_module="app.agent.manager",
+            target_name="agent_manager",
+            replacement="app.agent.manager.agent_manager",
+        ),
     },
     # 刮削能力从 MediaChain 拆出为独立 ScrapingChain 后，
     # 原 app.chain.media 模块级公开的刮削选项与策略配置随迁到 app.chain.scraping

app/startup/initializers/agent.py

@@ -75,7 +75,7 @@ def _get_injected_agent_manager() -> Any:
     if _agent_data_context is None:
         raise RuntimeError("Agent 数据上下文尚未由启动组合根装配")
     if _injected_agent_manager is None:
-        from app.agent.orchestrator import AgentManager
+        from app.agent.manager import AgentManager
 
         _injected_agent_manager = AgentManager(data=_agent_data_context)
     return _injected_agent_manager

docs/architecture-optimization-checklist.md

@@ -70,7 +70,7 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 
 | 指标 | 当前值 | 解释 |
 |---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 955 / 8,106 | `dependency-baseline.json` 当前快照 |
+| 宿主 Python 模块 / 内部依赖边 | 959 / 8,149 | `dependency-baseline.json` 当前快照 |
 | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
 | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
 | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
@@ -78,8 +78,8 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement |
 | Python 源码量 | 约 298,700 行 | 60 个文件超过 1,000 行，10 个超过 2,000 行 |
 | 长方法 | 299 个超过 80 行 | 70 个超过 150 行，23 个超过 250 行；大量是私有方法 |
-| 全量 mypy 历史债务 | 10,480 / 594 文件 | strict frontier 当前覆盖 41 个文件，Plugin 与 LLM Provider owner 拆分及受益调用方低水位已固化 |
-| Ruff 历史诊断 | 654 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
+| 全量 mypy 历史债务 | 10,415 / 593 文件 | strict frontier 当前覆盖 41 个文件，Plugin、LLM Provider 与 Agent Manager owner 拆分及受益调用方低水位已固化 |
+| Ruff 历史诊断 | 652 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
 | 覆盖率低水位 | Application 80.92%，Domain 79.87% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
@@ -87,8 +87,15 @@ MoviePilot V3 已经形成较清晰的模块化单体：`foundation`、`domain`
 | 文件 | 行数 | 主要职责混合 |
 |---|---:|---|
 | `app/chain/subscribe/`（已治理） | Facade 91 行 | 搜索、匹配、刷新、完成、规则引用与通知已拆至单一 owner；原 3,895 行单文件已删除 |
-| `app/agent/orchestrator.py` | 3,678 | 会话、运行、流式输出、工具、任务与 Manager |
-| `app/agent/llm/provider.py` | 3,528 | provider 目录、认证、模型发现、配置和运行时构建 |
+| `app/agent/orchestrator.py` | 2,558 | 单 Agent 运行、流式输出、工具与中间件编排 |
+| `app/agent/session.py` | 837 | 会话队列、worker、状态、取消与延迟清理 |
+| `app/agent/tasks.py` | 208 | 后台 prompt、持久化定时任务与心跳 |
+| `app/agent/lifecycle.py` | 166 | 接收门禁、启动、空闲回收与有界关闭 |
+| `app/agent/manager.py` | 27 | 稳定 `AgentManager` 构造门面 |
+| `app/agent/llm/provider.py` | 420 | 稳定 `LLMProviderManager` Facade 与兼容方法 |
+| `app/agent/llm/catalog.py` | 1,571 | provider spec、预设和模型元数据 |
+| `app/agent/llm/discovery.py` | 985 | 远端目录发现与 SDK 客户端 I/O |
+| `app/agent/llm/auth.py` | 582 | 持久鉴权和外部授权协议 |
 | `app/adapters/external/market.py`（已治理） | Facade 226 行 | `PluginHelper` 只保留精确旧 ABI、安装 Gateway 及 owner 委托 |
 | `app/adapters/external/plugin/client.py` | 1,600 | 插件市场传输、索引、Release 与本地仓候选查询 |
 | `app/adapters/system/plugin/package.py` | 2,324 | 插件安装、checkpoint、备份、恢复和物理删除事务 |
@@ -644,10 +651,12 @@ Pylint 10/10、Ruff、mypy/复杂度 ratchet、宿主与最新官方插件基线
 
 ### ARCH-202 拆分 Agent 与 LLM provider
 
-- [ ] 将内置 provider spec 数据从 `LLMProviderManager` 分离为只读 catalog。
-- [ ] 模型发现、认证会话、运行时构建和用户配置分成可替换服务。
+- [x] 将内置 provider spec 数据从 `LLMProviderManager` 分离为只读 catalog。
+- [x] 模型发现、认证会话和运行时构建已拆至单词 owner。
 - [ ] 将 WebAgent SSE、文件/音频与会话桥接移出 API endpoint。
-- [ ] 保留 `AgentManager`/`LLMProviderManager` 公开方法为稳定 Facade，并逐批纳入 strict mypy。
+- [x] `AgentManager` 会话、生命周期和后台任务已拆到单词 owner，公开方法由
+  `app.agent.AgentManager` 稳定 Facade 精确保持，包根不再无边界转发。
+- [x] `LLMProviderManager` 公开方法由稳定 Facade 精确保持；owner 已纳入 mypy 低水位。
 
 ### ARCH-203 拆分 Domain 投影与 Startup 高扇出
 

docs/architecture-overview.md

@@ -213,7 +213,7 @@ SDK/Compat。领域查询与写入继续使用 Application 所属的冻结 DTO/P
 | `app/schemas/` | Pydantic 传输模型、枚举（`ModuleType`、`EventType`、`SystemConfigKey` 等） | `types.py`、`context.py` |
 | `app/api/` | FastAPI 主端点、鉴权依赖、统一 `Response` 响应封装；动态插件端点不走此统一包装 | `apiv1.py`、`endpoints/`、`response.py` |
 | `app/adapters/web/plugin/` | FastAPI 动态插件路由的技术适配：注册/移除、认证依赖、OpenAPI 重建；保留插件原生响应结构 | `routes.py` |
-| `app/agent/` | AI Agent：编排器、运行时、工具、中间件、LLM、记忆、技能、策略 | `orchestrator.py`、`runtime_loader.py`、`tools/` |
+| `app/agent/` | AI Agent：稳定 Manager 门面、会话/生命周期/后台任务 owner、单 Agent 编排器、运行时、工具、中间件、LLM、记忆、技能、策略 | `manager.py`、`session.py`、`lifecycle.py`、`tasks.py`、`orchestrator.py` |
 | `app/startup/` | 唯一组合根：跨层装配、领域初始化、声明式生命周期排序与重启策略 | `composition/`、`initializers/`、`lifecycle/` |
 | `app/sdk/` | 面向新插件的稳定导入面（网络、缓存、日志、浏览器等）；`_legacy/` 只承载旧插件行为适配薄门面 | `network.py`、`browser.py`、`cache.py`、`_legacy/` |
 | `app/monitor/` | 源目录监控 → 触发整理 | `watcher.py`、`dispatcher.py` |
@@ -578,7 +578,11 @@ flowchart TB
     Reg["app/startup/initializers/agent.py<br/>生命周期显式注册/重置 Provider"]
     Reg --> Facade
     Facade -.能力启用或首次使用时物化.-> RT["app/agent/runtime_loader.py<br/>能力发现与服务物化"]
-    RT --> ORC["app/agent/orchestrator.py<br/>会话编排"]
+    RT --> Manager["app/agent/manager.py<br/>稳定 AgentManager 门面"]
+    Manager --> Session["session.py / lifecycle.py<br/>会话队列与有界生命周期"]
+    Manager --> Tasks["tasks.py<br/>后台、调度与心跳任务"]
+    Session --> ORC["orchestrator.py<br/>单 MoviePilotAgent 执行"]
+    Tasks --> Session
     ORC --> Tools["app/agent/tools<br/>系统工具（经 application 门面）"]
     ORC --> LLM["app/agent/llm<br/>LLM 提供商管理"]
     ORC --> MW["middleware / policy / memory / skills"]
@@ -590,6 +594,9 @@ flowchart TB
 
 - Chain 访问 Agent 运行时只能经 `app/application/agent.py`；
   `app/chain/agent.py` 的 `AgentChain` 是链层入口，Agent 实现保持在 `app/agent/`。
+- `app.agent.AgentManager` 是插件稳定路径，精确解析到
+  `app/agent/manager.py`；宿主直接导入 owner，包根不得通配转发
+  `orchestrator.py` 的任意内部名称。
 - Agent 工具不直接 import API / 调度器 / 命令：插件动态路由与文件夹操作使用
   `application/plugin/routes.py`、`application/plugin/folders.py`，调度和命令分别使用
   `application/scheduling.py`、`application/commands.py`。FastAPI 具体实现位于
@@ -728,8 +735,8 @@ flowchart LR
 
 | 指标 | 当前值 |
 |---|---:|
-| Python 模块 | 955 |
-| 内部导入边 | 8,106 |
+| Python 模块 | 959 |
+| 内部导入边 | 8,149 |
 | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
 | Application / Chain 具体 Adapter 直连 | 0 / 0 |
 | Direct egress | 53（债务已清零，53 条精确 containment） |

docs/rules/05-architecture.md

@@ -77,6 +77,23 @@ to make the directory tree look symmetrical.
 | `app/application/messaging/` | Message rendering/routing, interactions and the Agent-to-message bridge: `ingress.py` owns the single channel-to-host loopback boundary; `interaction.py` shared interaction contracts and view helpers; `router.py` unified interaction priority and callback dispatch; `site.py`/`subscribe.py`/`skill.py` per-command sessions, input parsing and views; `media.py` media interaction state while the business workflow stays in `MediaInteractionChain`; `plugin.py` plugin input capture and plugin button callbacks; `agent.py` owns Agent choice state, callback protocol, bounded WebAgent event publication, display projection, session identity/persistence coordination, temporary attachment registry and audio preparation/transcription; `message.py` notification rendering, templates and queue. Not a public SDK recommended for direct plugin use |
 | `app/application/security/` | Authentication, authorization, frozen user/auth projections, atomic user aggregate commands, per-user configuration publication, cookies, passkeys, OTP/two-factor, path/URL safety, SSRF and signing policy |
 
+### Agent runtime boundaries
+
+| Path | Ownership |
+|---|---|
+| `app/agent/manager.py` | Stable `AgentManager` facade and process singleton identity; no session, lifecycle or scheduled-task implementation |
+| `app/agent/session.py` | Per-session queues, workers, active Agent instances, status projection, cancellation and deferred cleanup |
+| `app/agent/lifecycle.py` | Runtime generation admission, initialization, idle-session collection and bounded manager shutdown |
+| `app/agent/tasks.py` | Background prompts, durable scheduled-task execution and heartbeat wakeups |
+| `app/agent/orchestrator.py` | One `MoviePilotAgent` execution instance: prompt/tool/middleware assembly, model invocation, streaming and per-agent state |
+
+`app.agent.AgentManager` remains the stable plugin-facing import and resolves lazily to
+`app.agent.manager.AgentManager`. Historical symbols formerly imported from
+`app.agent.orchestrator` are preserved only by exact Compat aliases; the physical
+orchestrator must not re-export manager/session implementations. The package root
+uses a reviewed symbol whitelist and must not restore an unbounded `__getattr__`
+forwarder.
+
 Application services may use domain rules and runtime contracts. They own the
 persistence Protocol needed by a use case, but must not import `app.db`,
 SQLAlchemy, Session, Oper classes or concrete adapters. `app/db/adapters/`
@@ -837,6 +854,11 @@ driven workflow registration.
 | `app/application/agent.py` | Agent orchestration facade plus typed `AgentDataContext`; lightweight service providers register through `app/startup/initializers/agent.py`, with no static `application -> agent` edge or global persistence locator |
 | `app/application/network.py` | Server-owned network-test catalog, public/private DTOs, URL and redirect admission, content validation and the injected HTTPS transport Port |
 | `app/db/adapters/agent.py` | Agent task and plugin-data persistence implementations; ORM values are projected to Application snapshots before Session close |
+| `app/agent/manager.py` | Stable `AgentManager` construction facade and singleton identity |
+| `app/agent/session.py` | Session queue/worker/status/cancellation and deferred-cleanup owner |
+| `app/agent/lifecycle.py` | Agent manager admission, startup, idle collection and bounded shutdown owner |
+| `app/agent/tasks.py` | Background prompt, scheduled task and heartbeat execution owner |
+| `app/agent/orchestrator.py` | Per-session `MoviePilotAgent` execution and LLM/tool/middleware orchestration only |
 | `app/agent/runtime_loader.py` | Agent-specific capability discovery and canonical entrypoint/service materialization; reuses the generic Capability Runtime while keeping Agent ownership under `app/agent/` |
 | `app/application/subscription/contract.py` | Deeply frozen Subscription/History DTOs, media identity, write patch and typed query/write/staging Repository contracts |
 | `app/application/subscription/write.py` | Subscription media translation and sync/async write-command orchestration |

tests/fixtures/architecture/dependency-baseline.json

@@ -1089,8 +1089,8 @@
       "runtime_only": true
     }
   },
-  "edge_count": 8106,
-  "edge_sha256": "b722e4939f1c879c80505778e0194255bd62d4bd17955b49fc642f865fa30213",
+  "edge_count": 8149,
+  "edge_sha256": "ba993b931e67bbd7901b6dbfc344634f508483533000d5b8f2372b7b130ea420",
   "edges": [
     "app -> app.foundation",
     "app -> app.foundation.environment",
@@ -1322,6 +1322,12 @@
     "app.agent.capabilities.adapter -> app.runtime.settings",
     "app.agent.contracts -> app.schemas",
     "app.agent.contracts -> app.schemas.types",
+    "app.agent.lifecycle -> app.agent",
+    "app.agent.lifecycle -> app.agent.session",
+    "app.agent.lifecycle -> app.application",
+    "app.agent.lifecycle -> app.application.agent",
+    "app.agent.lifecycle -> app.runtime",
+    "app.agent.lifecycle -> app.runtime.log",
     "app.agent.llm.auth -> app.agent",
     "app.agent.llm.auth -> app.agent.llm",
     "app.agent.llm.auth -> app.agent.llm.runtime",
@@ -1390,6 +1396,11 @@
     "app.agent.llm.session -> app.agent",
     "app.agent.llm.session -> app.agent.llm",
     "app.agent.llm.session -> app.agent.llm.runtime",
+    "app.agent.manager -> app.agent",
+    "app.agent.manager -> app.agent.memory",
+    "app.agent.manager -> app.agent.tasks",
+    "app.agent.manager -> app.application",
+    "app.agent.manager -> app.application.agent",
     "app.agent.mcp -> app.adapters",
     "app.agent.mcp -> app.adapters.network",
     "app.agent.mcp -> app.adapters.network.http",
@@ -1514,15 +1525,12 @@
     "app.agent.orchestrator -> app.agent.tools.impl.query_system_settings",
     "app.agent.orchestrator -> app.application",
     "app.agent.orchestrator -> app.application.agent",
-    "app.agent.orchestrator -> app.application.agenttask",
     "app.agent.orchestrator -> app.application.messaging",
     "app.agent.orchestrator -> app.application.messaging.chat",
     "app.agent.orchestrator -> app.application.plugin",
     "app.agent.orchestrator -> app.application.plugin.runtime",
     "app.agent.orchestrator -> app.chain",
     "app.agent.orchestrator -> app.chain.agent",
-    "app.agent.orchestrator -> app.foundation",
-    "app.agent.orchestrator -> app.foundation.identity",
     "app.agent.orchestrator -> app.runtime",
     "app.agent.orchestrator -> app.runtime.events",
     "app.agent.orchestrator -> app.runtime.execution",
@@ -1571,6 +1579,21 @@
     "app.agent.runtime_loader -> app.runtime.capabilities",
     "app.agent.runtime_loader -> app.runtime.capabilities.model",
     "app.agent.runtime_loader -> app.runtime.capabilities.runtime",
+    "app.agent.session -> app.agent",
+    "app.agent.session -> app.agent.contracts",
+    "app.agent.session -> app.agent.memory",
+    "app.agent.session -> app.agent.orchestrator",
+    "app.agent.session -> app.application",
+    "app.agent.session -> app.application.agent",
+    "app.agent.session -> app.chain",
+    "app.agent.session -> app.chain.agent",
+    "app.agent.session -> app.runtime",
+    "app.agent.session -> app.runtime.execution",
+    "app.agent.session -> app.runtime.log",
+    "app.agent.session -> app.runtime.observability",
+    "app.agent.session -> app.runtime.settings",
+    "app.agent.session -> app.schemas",
+    "app.agent.session -> app.schemas.types",
     "app.agent.skills.registry -> app.adapters",
     "app.agent.skills.registry -> app.adapters.network",
     "app.agent.skills.registry -> app.adapters.network.http",
@@ -1587,6 +1610,26 @@
     "app.agent.skills.registry -> app.runtime.config",
     "app.agent.skills.registry -> app.runtime.log",
     "app.agent.skills.registry -> app.runtime.settings",
+    "app.agent.tasks -> app.agent",
+    "app.agent.tasks -> app.agent.contracts",
+    "app.agent.tasks -> app.agent.lifecycle",
+    "app.agent.tasks -> app.agent.middleware",
+    "app.agent.tasks -> app.agent.middleware.jobs",
+    "app.agent.tasks -> app.agent.orchestrator",
+    "app.agent.tasks -> app.agent.prompt",
+    "app.agent.tasks -> app.agent.runtime",
+    "app.agent.tasks -> app.application",
+    "app.agent.tasks -> app.application.agenttask",
+    "app.agent.tasks -> app.chain",
+    "app.agent.tasks -> app.chain.agent",
+    "app.agent.tasks -> app.foundation",
+    "app.agent.tasks -> app.foundation.identity",
+    "app.agent.tasks -> app.runtime",
+    "app.agent.tasks -> app.runtime.log",
+    "app.agent.tasks -> app.runtime.settings",
+    "app.agent.tasks -> app.schemas",
+    "app.agent.tasks -> app.schemas.message",
+    "app.agent.tasks -> app.schemas.types",
     "app.agent.tools.base -> app.agent",
     "app.agent.tools.base -> app.agent.callback",
     "app.agent.tools.base -> app.agent.policy",
@@ -8620,7 +8663,7 @@
     "app.startup.initializers.agent -> app.agent.llm.gateway",
     "app.startup.initializers.agent -> app.agent.llm.helper",
     "app.startup.initializers.agent -> app.agent.llm.provider",
-    "app.startup.initializers.agent -> app.agent.orchestrator",
+    "app.startup.initializers.agent -> app.agent.manager",
     "app.startup.initializers.agent -> app.agent.prompt",
     "app.startup.initializers.agent -> app.agent.prompt.transfer_redo",
     "app.startup.initializers.agent -> app.agent.runtime_loader",
@@ -9199,7 +9242,7 @@
     "app.workflow.actions.transfer_file -> app.workflow",
     "app.workflow.actions.transfer_file -> app.workflow.actions"
   ],
-  "module_count": 955,
+  "module_count": 959,
   "modules": [
     "app",
     "app.adapters",
@@ -9256,6 +9299,7 @@
     "app.agent.capabilities",
     "app.agent.capabilities.adapter",
     "app.agent.contracts",
+    "app.agent.lifecycle",
     "app.agent.llm",
     "app.agent.llm.auth",
     "app.agent.llm.capability",
@@ -9267,6 +9311,7 @@
     "app.agent.llm.runtime",
     "app.agent.llm.server_tools",
     "app.agent.llm.session",
+    "app.agent.manager",
     "app.agent.mcp",
     "app.agent.memory",
     "app.agent.middleware",
@@ -9293,9 +9338,11 @@
     "app.agent.prompt.transfer_redo",
     "app.agent.runtime",
     "app.agent.runtime_loader",
+    "app.agent.session",
     "app.agent.skills",
     "app.agent.skills.metadata",
     "app.agent.skills.registry",
+    "app.agent.tasks",
     "app.agent.tools",
     "app.agent.tools.base",
     "app.agent.tools.catalog",

tests/fixtures/architecture/mypy-baseline.json

@@ -163,12 +163,6 @@
     "type-arg": 7,
     "type-var": 1
   },
-  "app/agent/memory/__init__.py": {
-    "arg-type": 1,
-    "no-untyped-call": 1,
-    "no-untyped-def": 7,
-    "type-arg": 1
-  },
   "app/agent/middleware/activity_log.py": {
     "misc": 3,
     "type-arg": 5,
@@ -221,14 +215,13 @@
     "no-any-return": 2
   },
   "app/agent/orchestrator.py": {
-    "arg-type": 13,
-    "assignment": 14,
+    "arg-type": 6,
     "attr-defined": 7,
     "dict-item": 1,
-    "no-any-return": 9,
-    "no-untyped-call": 6,
-    "no-untyped-def": 20,
-    "type-arg": 35,
+    "no-any-return": 7,
+    "no-untyped-call": 3,
+    "no-untyped-def": 13,
+    "type-arg": 13,
     "union-attr": 3,
     "var-annotated": 1
   },

tests/fixtures/architecture/ruff-baseline.json

@@ -857,9 +857,6 @@
   "tests/test_agent_activity_log.py": {
     "I001": 1
   },
-  "tests/test_agent_background_output.py": {
-    "I001": 1
-  },
   "tests/test_agent_lazy_runtime_boundary.py": {
     "F401": 1
   },
@@ -881,9 +878,6 @@
   "tests/test_agent_subagents.py": {
     "I001": 1
   },
-  "tests/test_agent_summarization_streaming.py": {
-    "I001": 1
-  },
   "tests/test_agent_tokens_events.py": {
     "I001": 1
   },

tests/fixtures/architecture/runtime-contract-baseline.json

@@ -879,15 +879,60 @@
     },
     "symbol_aliases": {
       "app.agent.orchestrator": {
+        "AGENT_SESSION_QUEUE_MAX_SIZE": {
+          "replacement": "app.agent.session.AGENT_SESSION_QUEUE_MAX_SIZE",
+          "target_module": "app.agent.session",
+          "target_name": "AGENT_SESSION_QUEUE_MAX_SIZE"
+        },
         "AgentChain": {
           "replacement": "app.chain.agent.AgentChain",
           "target_module": "app.chain.agent",
           "target_name": "AgentChain"
         },
+        "AgentManager": {
+          "replacement": "app.agent.manager.AgentManager",
+          "target_module": "app.agent.manager",
+          "target_name": "AgentManager"
+        },
+        "AgentManagerQueueFullError": {
+          "replacement": "app.agent.session.AgentManagerQueueFullError",
+          "target_module": "app.agent.session",
+          "target_name": "AgentManagerQueueFullError"
+        },
+        "AgentManagerUnavailableError": {
+          "replacement": "app.agent.session.AgentManagerUnavailableError",
+          "target_module": "app.agent.session",
+          "target_name": "AgentManagerUnavailableError"
+        },
         "ReplyMode": {
           "replacement": "app.schemas.types.ReplyMode",
           "target_module": "app.schemas.types",
           "target_name": "ReplyMode"
+        },
+        "_MessageTask": {
+          "replacement": "app.agent.session._MessageTask",
+          "target_module": "app.agent.session",
+          "target_name": "_MessageTask"
+        },
+        "_async_finish_processing_status": {
+          "replacement": "app.agent.session._async_finish_processing_status",
+          "target_module": "app.agent.session",
+          "target_name": "_async_finish_processing_status"
+        },
+        "_async_start_processing_status": {
+          "replacement": "app.agent.session._async_start_processing_status",
+          "target_module": "app.agent.session",
+          "target_name": "_async_start_processing_status"
+        },
+        "_finish_processing_status": {
+          "replacement": "app.agent.session._finish_processing_status",
+          "target_module": "app.agent.session",
+          "target_name": "_finish_processing_status"
+        },
+        "agent_manager": {
+          "replacement": "app.agent.manager.agent_manager",
+          "target_module": "app.agent.manager",
+          "target_name": "agent_manager"
         }
       },
       "app.application.transfer": {

tests/test_agent_background_output.py

@@ -1,17 +1,11 @@
-import unittest
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
+import pytest
 from langchain_core.messages import AIMessage, HumanMessage
 
-from app.agent import (
-    HEARTBEAT_SESSION_PREFIX,
-    MoviePilotAgent,
-    AgentManager,
-    ReplyMode,
-    UNSUPPORTED_IMAGE_INPUT_MESSAGE,
-    _MessageTask,
-)
+from app.agent.contracts import ReplyMode
+from app.agent.manager import AgentManager
 from app.agent.memory import memory_manager
 from app.agent.middleware.activity_log import QUERY_ACTIVITY_LOG_TOOL_NAME
 from app.agent.middleware.skills import SKILL_TOOL_NAME
@@ -19,9 +13,17 @@ from app.agent.middleware.subagents import (
     SUBAGENT_CONTROL_TOOL_NAME,
     SUBAGENT_TASK_TOOL_NAME,
 )
+from app.agent.orchestrator import (
+    HEARTBEAT_SESSION_PREFIX,
+    UNSUPPORTED_IMAGE_INPUT_MESSAGE,
+    MoviePilotAgent,
+)
+from app.agent.session import _MessageTask
 from app.agent.tools.factory import MoviePilotToolFactory
-from app.runtime.config import settings
 from app.foundation.identity import SYSTEM_INTERNAL_USER_ID
+from app.runtime.config import settings
+
+pytestmark = pytest.mark.anyio
 
 
 class _FakeGraphState:
@@ -53,16 +55,15 @@ class _FakeFailingAgent:
 
 class _FakeStreamingFailingAgent(_FakeFailingAgent):
     async def astream(self, _messages, **_kwargs):
+        if _kwargs.get("__keep_async_generator__"):
+            yield None
         raise self._error
-        # 保持 async generator 形态，避免测试替身变成普通 coroutine。
-        yield None
 
 
 class _FakeStreamingAgent(_FakeAgent):
     async def astream(self, _messages, **_kwargs):
-        return
-        # 保持 async generator 形态，当前用例不需要实际 token。
-        yield None
+        if _kwargs.get("__keep_async_generator__"):
+            yield None
 
 
 class StreamChunkTimeoutError(RuntimeError):
@@ -79,7 +80,7 @@ def _fake_activity_log_middleware(tool=None):
     return SimpleNamespace(name="activity", tools=[] if tool is None else [tool])
 
 
-class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
+class TestAgentBackgroundOutput:
     async def test_background_non_streaming_does_not_send_by_default(self):
         agent = MoviePilotAgent(session_id="bg-test", user_id="system")
         agent.channel = None
@@ -101,7 +102,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
 
         agent.send_agent_message.assert_not_awaited()
         save_messages.assert_not_called()
-        self.assertEqual("后台结果", agent._streamed_output)
+        assert agent._streamed_output == "后台结果"
 
     async def test_non_streaming_image_unsupported_error_sends_friendly_notice(self):
         agent = MoviePilotAgent(session_id="image-test", user_id="user-1")
@@ -131,11 +132,11 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             ]
         )
 
-        self.assertEqual(UNSUPPORTED_IMAGE_INPUT_MESSAGE, result)
+        assert result == UNSUPPORTED_IMAGE_INPUT_MESSAGE
         agent.send_agent_message.assert_awaited_once_with(
             UNSUPPORTED_IMAGE_INPUT_MESSAGE, title=""
         )
-        self.assertEqual(UNSUPPORTED_IMAGE_INPUT_MESSAGE, agent._streamed_output)
+        assert agent._streamed_output == UNSUPPORTED_IMAGE_INPUT_MESSAGE
 
     async def test_streaming_image_unsupported_error_sends_friendly_notice(self):
         agent = MoviePilotAgent(session_id="image-test", user_id="user-1")
@@ -168,7 +169,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             ]
         )
 
-        self.assertEqual(UNSUPPORTED_IMAGE_INPUT_MESSAGE, result)
+        assert result == UNSUPPORTED_IMAGE_INPUT_MESSAGE
         agent.send_agent_message.assert_awaited_once_with(
             UNSUPPORTED_IMAGE_INPUT_MESSAGE, title=""
         )
@@ -203,12 +204,12 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             "智能助手执行失败: No streaming chunk received for 120.0s "
             "(model=mimo-v2.5-pro, chunks_received=1)."
         )
-        self.assertEqual(expected, result)
+        assert result == expected
         agent.send_agent_message.assert_awaited_once_with(expected, title="")
         sent_message = agent.send_agent_message.await_args.args[0]
-        self.assertIn("No streaming chunk received for 120.0s", sent_message)
-        self.assertNotIn("Tune or disable", sent_message)
-        self.assertEqual(expected, agent._streamed_output)
+        assert "No streaming chunk received for 120.0s" in sent_message
+        assert "Tune or disable" not in sent_message
+        assert agent._streamed_output == expected
 
     async def test_streaming_success_stops_streaming_once(self):
         """流式正常完成时不应在 finally 中重复停止流式输出。"""
@@ -258,9 +259,9 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
 
         save_messages.assert_called_once()
         _, kwargs = save_messages.call_args
-        self.assertEqual("tool-reply", kwargs["session_id"])
-        self.assertEqual("user-1", kwargs["user_id"])
-        self.assertEqual("消息已发送", kwargs["messages"][0].content)
+        assert kwargs["session_id"] == "tool-reply"
+        assert kwargs["user_id"] == "user-1"
+        assert kwargs["messages"][0].content == "消息已发送"
 
     async def test_process_does_not_mutate_cached_agent_messages(self):
         """处理新消息时不应直接修改记忆缓存中的历史消息列表。"""
@@ -290,10 +291,10 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             result = await agent.process("继续")
 
-        self.assertEqual("消息已发送", result)
-        self.assertEqual(1, len(cached_messages))
-        self.assertIsNot(cached_messages, captured["messages"])
-        self.assertEqual(2, len(captured["messages"]))
+        assert result == "消息已发送"
+        assert len(cached_messages) == 1
+        assert cached_messages is not captured["messages"]
+        assert len(captured["messages"]) == 2
 
     async def test_background_non_streaming_sends_when_reply_mode_dispatch(self):
         agent = MoviePilotAgent(session_id="bg-test", user_id="system")
@@ -322,7 +323,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             "后台结果", title="MoviePilot助手"
         )
         save_messages.assert_not_called()
-        self.assertEqual("后台结果", agent._streamed_output)
+        assert agent._streamed_output == "后台结果"
 
     async def test_background_non_streaming_captures_without_sending_when_capture_only(self):
         agent = MoviePilotAgent(session_id="bg-test", user_id="system")
@@ -345,13 +346,13 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
 
         agent.send_agent_message.assert_not_awaited()
         save_messages.assert_not_called()
-        self.assertEqual("后台结果", agent._streamed_output)
+        assert agent._streamed_output == "后台结果"
 
     async def test_heartbeat_check_jobs_captures_final_reply_and_keeps_message_tools(self):
         manager = AgentManager()
 
         with (
-            patch("app.agent.orchestrator.load_jobs_metadata", new=AsyncMock(return_value=[{
+            patch("app.agent.tasks.load_jobs_metadata", new=AsyncMock(return_value=[{
                 "id": "job-1",
                 "name": "测试任务",
                 "description": "desc",
@@ -367,14 +368,14 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
 
         process_message.assert_awaited_once()
         kwargs = process_message.await_args.kwargs
-        self.assertEqual(ReplyMode.CAPTURE_ONLY, kwargs["reply_mode"])
-        self.assertTrue(kwargs["allow_message_tools"])
+        assert kwargs["reply_mode"] == ReplyMode.CAPTURE_ONLY
+        assert kwargs["allow_message_tools"] is True
 
     async def test_heartbeat_check_jobs_skips_when_no_active_jobs(self):
         manager = AgentManager()
 
         with (
-            patch("app.agent.orchestrator.load_jobs_metadata", new=AsyncMock(return_value=[])),
+            patch("app.agent.tasks.load_jobs_metadata", new=AsyncMock(return_value=[])),
             patch.object(manager, "process_message", new=AsyncMock()) as process_message,
         ):
             await manager.heartbeat_check_jobs()
@@ -439,8 +440,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             created = await agent._create_agent(streaming=False)
 
-        self.assertEqual(
-            [
+        assert [getattr(item, "name", item) for item in created["middleware"]] == [
                 "AgentPolicyMiddleware",
                 "skills",
                 "jobs",
@@ -449,9 +449,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
                 "patch",
                 "FinalRequestCompactionMiddleware",
                 "usage",
-            ],
-            [getattr(item, "name", item) for item in created["middleware"]],
-        )
+        ]
 
     async def test_create_agent_registers_skill_tool_from_middleware(self):
         """SkillsMiddleware 暴露的 skill 工具应进入 Agent 工具和筛选候选。"""
@@ -504,9 +502,9 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             created = await agent._create_agent(streaming=False)
 
-        self.assertIn(skill_tool, created["tools"])
-        self.assertIn(skill_tool, captured["selection_tools"])
-        self.assertIn(SKILL_TOOL_NAME, captured["always_include"])
+        assert skill_tool in created["tools"]
+        assert skill_tool in captured["selection_tools"]
+        assert SKILL_TOOL_NAME in captured["always_include"]
 
     async def test_create_agent_excludes_activity_log_without_message_context(self):
         """无渠道信息的后台捕获任务不应注入活动日志。"""
@@ -555,8 +553,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             created = await agent._create_agent(streaming=False)
 
-        self.assertEqual(
-            [
+        assert [getattr(item, "name", item) for item in created["middleware"]] == [
                 "AgentPolicyMiddleware",
                 "skills",
                 "jobs",
@@ -565,9 +562,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
                 "patch",
                 "FinalRequestCompactionMiddleware",
                 "usage",
-            ],
-            [getattr(item, "name", item) for item in created["middleware"]],
-        )
+        ]
 
     def test_message_tool_is_not_always_included_by_tool_selector(self):
         """消息发送工具不应绕过工具筛选。"""
@@ -577,7 +572,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             [send_message_tool]
         )
 
-        self.assertNotIn("send_message", always_include)
+        assert "send_message" not in always_include
 
     def test_activity_log_tool_is_not_registered_by_tool_factory(self):
         """活动日志查询工具不应再由全局工具工厂保留。"""
@@ -587,7 +582,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
             [activity_log_tool]
         )
 
-        self.assertNotIn(QUERY_ACTIVITY_LOG_TOOL_NAME, always_include)
+        assert QUERY_ACTIVITY_LOG_TOOL_NAME not in always_include
 
     async def test_create_agent_registers_activity_log_tool_from_middleware(self):
         """ActivityLogMiddleware 暴露的工具应进入 Agent 工具和筛选候选。"""
@@ -647,9 +642,9 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             created = await agent._create_agent(streaming=False)
 
-        self.assertIn(activity_tool, created["tools"])
-        self.assertIn(activity_tool, captured["selection_tools"])
-        self.assertIn(QUERY_ACTIVITY_LOG_TOOL_NAME, captured["always_include"])
+        assert activity_tool in created["tools"]
+        assert activity_tool in captured["selection_tools"]
+        assert QUERY_ACTIVITY_LOG_TOOL_NAME in captured["always_include"]
 
     async def test_create_agent_always_includes_subagent_tools(self):
         """工具筛选开启时应保留同步和异步子代理入口。"""
@@ -709,8 +704,8 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             await agent._create_agent(streaming=False)
 
-        self.assertIn(SUBAGENT_TASK_TOOL_NAME, captured["always_include"])
-        self.assertIn(SUBAGENT_CONTROL_TOOL_NAME, captured["always_include"])
+        assert SUBAGENT_TASK_TOOL_NAME in captured["always_include"]
+        assert SUBAGENT_CONTROL_TOOL_NAME in captured["always_include"]
 
     async def test_create_agent_keeps_activity_log_for_normal_session(self):
         agent = MoviePilotAgent(
@@ -759,8 +754,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         ):
             created = await agent._create_agent(streaming=False)
 
-        self.assertEqual(
-            [
+        assert [getattr(item, "name", item) for item in created["middleware"]] == [
                 "AgentPolicyMiddleware",
                 "skills",
                 "jobs",
@@ -770,9 +764,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
                 "patch",
                 "FinalRequestCompactionMiddleware",
                 "usage",
-            ],
-            [getattr(item, "name", item) for item in created["middleware"]],
-        )
+        ]
 
     async def test_run_background_prompt_forces_disable_message_tools_when_capture_only(self):
         captured = {}
@@ -795,7 +787,7 @@ class AgentBackgroundOutputTest(unittest.IsolatedAsyncioTestCase):
         finally:
             await manager.close()
 
-        self.assertEqual("background task", captured["message"])
-        self.assertEqual(ReplyMode.CAPTURE_ONLY, captured["reply_mode"])
-        self.assertFalse(captured["allow_message_tools"])
-        self.assertEqual(SYSTEM_INTERNAL_USER_ID, captured["user_id"])
+        assert captured["message"] == "background task"
+        assert captured["reply_mode"] == ReplyMode.CAPTURE_ONLY
+        assert captured["allow_message_tools"] is False
+        assert captured["user_id"] == SYSTEM_INTERNAL_USER_ID

tests/test_agent_chat_history.py

@@ -5,8 +5,8 @@ from unittest.mock import AsyncMock
 
 from langchain_core.messages import AIMessage, HumanMessage
 
-from app.agent import HEARTBEAT_SESSION_PREFIX, MoviePilotAgent
-from app.agent.memory import memory_manager
+from app.agent.memory import MemoryManager, memory_manager
+from app.agent.orchestrator import HEARTBEAT_SESSION_PREFIX, MoviePilotAgent
 from app.db.oper.agentchat import AgentChatOper
 from app.foundation.identity import SYSTEM_INTERNAL_USER_ID
 
@@ -254,7 +254,7 @@ def test_memory_manager_restores_agent_messages_from_database():
     """内存缓存缺失时应从 Agent 会话历史表恢复原始 messages。"""
     session_id = "session-memory"
     user_id = "2"
-    memory_manager.clear_memory(session_id, user_id)
+    isolated_memory = MemoryManager()
     AgentChatOper().save_agent_messages(
         session_id=session_id,
         user_id=user_id,
@@ -274,7 +274,10 @@ def test_memory_manager_restores_agent_messages_from_database():
         ],
     )
 
-    messages = memory_manager.get_agent_messages(session_id=session_id, user_id=user_id)
+    messages = isolated_memory.get_agent_messages(
+        session_id=session_id,
+        user_id=user_id,
+    )
 
     assert len(messages) == 1
     assert isinstance(messages[0], HumanMessage)
@@ -285,7 +288,7 @@ def test_async_memory_manager_restores_through_native_async_service(monkeypatch)
     """异步记忆恢复只能通过会话应用服务的异步查询端口。"""
     session_id = "session-memory-async"
     user_id = "3"
-    memory_manager.clear_memory(session_id, user_id)
+    isolated_memory = MemoryManager()
     service = SimpleNamespace(
         get=AsyncMock(
             return_value=SimpleNamespace(
@@ -312,7 +315,7 @@ def test_async_memory_manager_restores_through_native_async_service(monkeypatch)
     )
 
     messages = asyncio.run(
-        memory_manager.async_get_agent_messages(
+        isolated_memory.async_get_agent_messages(
             session_id=session_id,
             user_id=user_id,
         )

tests/test_agent_manager_facade.py

@@ -0,0 +1,114 @@
+"""AgentManager 门面、职责 owner 与兼容路径合同。"""
+
+import json
+import subprocess
+import sys
+
+import pytest
+
+from app.agent.lifecycle import AgentLifecycleOwner
+from app.agent.manager import AgentManager
+from app.agent.session import AgentSessionOwner
+from app.agent.tasks import AgentTaskOwner
+
+
+def _run_isolated(script: str) -> dict:
+    """在干净解释器中执行包根延迟加载探针。"""
+    completed = subprocess.run(
+        [sys.executable, "-c", script],
+        check=True,
+        capture_output=True,
+        text=True,
+    )
+    return json.loads(completed.stdout)
+
+
+def test_agent_package_root_uses_precise_lazy_contracts() -> None:
+    """包根不得加载完整编排器，也不得转发白名单外名称。"""
+    result = _run_isolated(
+        """
+import json
+import sys
+
+import app.agent as facade
+
+initial = {
+    "manager": "app.agent.manager" in sys.modules,
+    "orchestrator": "app.agent.orchestrator" in sys.modules,
+}
+try:
+    facade.NotAnAgentContract
+except AttributeError:
+    rejected = True
+else:
+    rejected = False
+
+from app.agent import AgentManager
+from app.agent.manager import AgentManager as CanonicalManager
+
+print(json.dumps({
+    "initial": initial,
+    "identity": AgentManager is CanonicalManager,
+    "rejected": rejected,
+}))
+"""
+    )
+
+    assert result == {
+        "initial": {"manager": False, "orchestrator": False},
+        "identity": True,
+        "rejected": True,
+    }
+
+
+def test_agent_manager_methods_have_single_owners() -> None:
+    """稳定门面只继承 owner 方法，不复制会话、生命周期或任务实现。"""
+    assert AgentManager.__dict__.keys() >= {"__init__"}
+    assert "process_message" not in AgentManager.__dict__
+    assert "initialize" not in AgentManager.__dict__
+    assert "run_background_prompt" not in AgentManager.__dict__
+
+    assert AgentManager.process_message is AgentSessionOwner.process_message
+    assert AgentManager.clear_session is AgentSessionOwner.clear_session
+    assert AgentManager.get_session_status is AgentSessionOwner.get_session_status
+    assert AgentManager.initialize is AgentLifecycleOwner.initialize
+    assert AgentManager.close is AgentLifecycleOwner.close
+    assert AgentManager.run_background_prompt is AgentTaskOwner.run_background_prompt
+    assert AgentManager.execute_scheduled_task is AgentTaskOwner.execute_scheduled_task
+    assert AgentManager.heartbeat_check_jobs is AgentTaskOwner.heartbeat_check_jobs
+
+
+def test_legacy_orchestrator_manager_path_resolves_to_canonical_facade() -> None:
+    """历史编排器导入只经精确 Compat 指向 canonical 门面。"""
+    from importlib import import_module
+
+    LegacyAgentManager = getattr(import_module("app.agent.orchestrator"), "AgentManager")
+
+    assert LegacyAgentManager is AgentManager
+
+
+def test_agent_manager_facade_keeps_public_method_set() -> None:
+    """重构后继续提供宿主和插件已使用的稳定公开方法。"""
+    expected = {
+        "clear_session",
+        "close",
+        "configure_data_context",
+        "execute_scheduled_task",
+        "get_session_status",
+        "heartbeat_check_jobs",
+        "initialize",
+        "is_session_busy",
+        "matches_secret_confirmation",
+        "process_message",
+        "run_background_prompt",
+        "stop_current_task",
+    }
+
+    missing = {name for name in expected if not callable(getattr(AgentManager, name, None))}
+    assert missing == set()
+
+
+def test_agent_manager_constructor_rejects_unknown_arguments() -> None:
+    """稳定构造门面不得静默接收未定义依赖。"""
+    with pytest.raises(TypeError):
+        AgentManager(unknown=True)

tests/test_agent_scheduled_tasks.py

@@ -15,14 +15,11 @@ import pytz
 from apscheduler.schedulers.background import BackgroundScheduler
 from langchain_core.messages import AIMessage
 
-from app.agent import (
-    AgentChain,
-    AgentManager,
-    MoviePilotAgent,
-    ReplyMode,
-    _MessageTask,
-)
+from app.agent.contracts import ReplyMode
+from app.agent.manager import AgentManager
 from app.agent.middleware.tool_selection import ToolSelectorMiddleware
+from app.agent.orchestrator import MoviePilotAgent
+from app.agent.session import _MessageTask
 from app.agent.tools.factory import MoviePilotToolFactory
 from app.agent.tools.impl.create_agent_task import (
     CreateAgentTaskInput,
@@ -39,6 +36,7 @@ from app.agent.tools.impl.update_agent_task import (
     UpdateAgentTaskTool,
 )
 from app.agent.tools.tags import ToolTag
+from app.chain.agent import AgentChain
 from app.db import SessionFactory
 from app.db.adapters.agent import TransactionalAgentTaskRepository
 from app.db.models.agenttask import AgentTask

tests/test_agent_summarization_streaming.py

@@ -24,11 +24,11 @@ from langgraph.checkpoint.memory import InMemorySaver
 from pydantic import Field
 
 import app.agent.orchestrator as agent_module
-from app.agent.memory import memory_manager
+from app.agent.memory import MemoryManager
 from app.agent.middleware.runtime_config import RuntimeConfigMiddleware
 from app.agent.middleware.summarization import (
-    ContextSummarizationError,
     ContextPreservingSummarizationMiddleware,
+    ContextSummarizationError,
     FinalRequestCompactionMiddleware,
 )
 from app.agent.middleware.usage import UsageMiddleware
@@ -1204,14 +1204,19 @@ def test_summary_failure_preserves_database_history():
     """上下文压缩失败时不得覆盖数据库中的上一轮消息。"""
     session_id = f"summary-failure-{uuid.uuid4().hex}"
     user_id = "10001"
-    memory_manager.save_agent_messages(
+    isolated_memory = MemoryManager()
+    isolated_memory.save_agent_messages(
         session_id=session_id,
         user_id=user_id,
         messages=[HumanMessage(content="数据库中的旧事实")],
     )
-    memory_manager.clear_memory(session_id, user_id)
-    restored_messages = memory_manager.get_agent_messages(session_id, user_id)
-    agent = agent_module.MoviePilotAgent(session_id=session_id, user_id=user_id)
+    isolated_memory.clear_memory(session_id, user_id)
+    restored_messages = isolated_memory.get_agent_messages(session_id, user_id)
+    agent = agent_module.MoviePilotAgent(
+        session_id=session_id,
+        user_id=user_id,
+        memory=isolated_memory,
+    )
     agent._compiled_agent_bundle = object()
     agent._should_stream = lambda: False
     agent._create_agent = AsyncMock(return_value=_FailingGraph())
@@ -1229,8 +1234,8 @@ def test_summary_failure_preserves_database_history():
             )
         )
 
-    memory_manager.clear_memory(session_id, user_id)
-    recovered_messages = memory_manager.get_agent_messages(session_id, user_id)
+    isolated_memory.clear_memory(session_id, user_id)
+    recovered_messages = isolated_memory.get_agent_messages(session_id, user_id)
     assert result == "智能助手执行失败: 会话上下文压缩失败，原有上下文已保留，请稍后重试"
     assert agent._compiled_agent_bundle is None
     assert [message.content for message in recovered_messages] == ["数据库中的旧事实"]

tests/test_architecture_dependencies.py

@@ -283,6 +283,55 @@ def test_retired_canonical_filenames_do_not_return():
     assert leftovers == []
 
 
+def test_agent_manager_uses_focused_owner_modules_and_precise_facade():
+    """AgentManager 必须保持薄门面，编排器不得重新聚合 Manager 实现。"""
+    owner_paths = {
+        "manager": APP_ROOT / "agent" / "manager.py",
+        "session": APP_ROOT / "agent" / "session.py",
+        "lifecycle": APP_ROOT / "agent" / "lifecycle.py",
+        "tasks": APP_ROOT / "agent" / "tasks.py",
+        "orchestrator": APP_ROOT / "agent" / "orchestrator.py",
+    }
+    assert all(path.is_file() for path in owner_paths.values())
+
+    manager_tree = ast.parse(owner_paths["manager"].read_text(encoding="utf-8-sig"))
+    manager_class = next(
+        node
+        for node in manager_tree.body
+        if isinstance(node, ast.ClassDef) and node.name == "AgentManager"
+    )
+    manager_methods = {
+        node.name
+        for node in manager_class.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+    }
+    assert manager_methods == {"__init__"}
+    assert [ast.unparse(base) for base in manager_class.bases] == ["AgentTaskOwner"]
+
+    orchestrator_tree = ast.parse(
+        owner_paths["orchestrator"].read_text(encoding="utf-8-sig")
+    )
+    orchestrator_classes = {
+        node.name for node in orchestrator_tree.body if isinstance(node, ast.ClassDef)
+    }
+    orchestrator_assignments = {
+        target.id
+        for node in orchestrator_tree.body
+        if isinstance(node, ast.Assign)
+        for target in node.targets
+        if isinstance(target, ast.Name)
+    }
+    assert "AgentManager" not in orchestrator_classes
+    assert "_MessageTask" not in orchestrator_classes
+    assert "agent_manager" not in orchestrator_assignments
+
+    startup_source = (
+        APP_ROOT / "startup" / "initializers" / "agent.py"
+    ).read_text(encoding="utf-8-sig")
+    assert "from app.agent.manager import AgentManager" in startup_source
+    assert "from app.agent.orchestrator import AgentManager" not in startup_source
+
+
 def test_runtime_dependencies_use_same_named_single_word_package() -> None:
     """运行依赖能力必须使用同名包，且宿主直接依赖单一职责子模块。"""
     package = APP_ROOT / "runtime" / "dependencies"

tests/test_message_processing_status.py

@@ -2,7 +2,7 @@ import json
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
-from app.agent.orchestrator import _finish_processing_status
+from app.agent.session import _finish_processing_status
 from app.modules.discord import DiscordModule
 from app.modules.slack import SlackModule
 from app.schemas.notification import ChannelCapability, ChannelCapabilityManager
@@ -136,7 +136,7 @@ def test_agent_finish_processing_status_uses_module_interface() -> None:
         "metadata": {"kind": "typing"},
     }
 
-    with patch("app.agent.orchestrator.AgentChain") as chain_cls:
+    with patch("app.agent.session.AgentChain") as chain_cls:
         _finish_processing_status(status, user_id="fallback")
 
     chain_cls.return_value.finish_message_processing_status.assert_called_once_with(

tests/test_telegram_typing_lifecycle.py

@@ -7,11 +7,8 @@ from unittest.mock import AsyncMock, Mock, patch
 
 import pytest
 
-from app.agent import (  # pylint: disable=no-name-in-module
-    AgentManager,
-    _async_start_processing_status,
-    _MessageTask,
-)
+from app.agent.manager import AgentManager
+from app.agent.session import _async_start_processing_status, _MessageTask
 from app.chain.message import MessageChain
 from app.command import Command, _finish_command_processing_status
 from app.modules.telegram import TelegramModule  # pylint: disable=no-name-in-module
@@ -421,7 +418,7 @@ def test_agent_manager_starts_processing_status_when_task_runs():
         }
 
         with patch(
-                "app.agent.orchestrator._async_start_processing_status",
+                "app.agent.session._async_start_processing_status",
                 new_callable=AsyncMock,
                 return_value=status,
         ) as start_status:
@@ -462,7 +459,7 @@ def test_agent_start_processing_status_uses_chain_interface():
                 calls.append(kwargs)
                 return status
 
-        with patch("app.agent.orchestrator.AgentChain", FakeAgentChain):
+        with patch("app.agent.session.AgentChain", FakeAgentChain):
             result = await _async_start_processing_status(task)
 
         assert calls == [{
@@ -560,7 +557,7 @@ def test_agent_manager_finishes_processing_status_after_each_task():
         )
 
         with patch(
-                "app.agent.orchestrator._async_finish_processing_status",
+                "app.agent.session._async_finish_processing_status",
                 new_callable=AsyncMock,
         ) as finish_status:
             await manager._finish_task_processing_status(task)
@@ -607,7 +604,7 @@ def test_agent_worker_starts_and_finishes_each_queued_task():
         ))
 
         with patch(
-                "app.agent.orchestrator._async_start_processing_status",
+                "app.agent.session._async_start_processing_status",
                 new_callable=AsyncMock,
                 side_effect=[first_status, second_status],
         ) as start_status, patch.object(
@@ -615,7 +612,7 @@ def test_agent_worker_starts_and_finishes_each_queued_task():
                 "_process_message_internal",
                 new_callable=AsyncMock,
         ), patch(
-                "app.agent.orchestrator._async_finish_processing_status",
+                "app.agent.session._async_finish_processing_status",
                 new_callable=AsyncMock,
         ) as finish_status:
             manager._session_workers["session-1"] = asyncio.create_task(