diff --git a/app/agent/__init__.py b/app/agent/__init__.py index f48dd30ee..6db4e19c5 100644 --- a/app/agent/__init__.py +++ b/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)) diff --git a/app/agent/lifecycle.py b/app/agent/lifecycle.py new file mode 100644 index 000000000..6bc7cad28 --- /dev/null +++ b/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 diff --git a/app/agent/manager.py b/app/agent/manager.py new file mode 100644 index 000000000..1a43efc96 --- /dev/null +++ b/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() diff --git a/app/agent/memory/__init__.py b/app/agent/memory/__init__.py index 027ff1808..d0d52c056 100644 --- a/app/agent/memory/__init__.py +++ b/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: """ 清理内存中过期记忆的后台任务 diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py index 0a45d1ca5..2973a4aa6 100644 --- a/app/agent/orchestrator.py +++ b/app/agent/orchestrator.py @@ -9,7 +9,7 @@ import uuid import warnings from dataclasses import dataclass from datetime import datetime, timedelta -from typing import Any, Awaitable, Callable, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional from langchain.agents import create_agent from langchain_core.messages import ( # noqa: F401 @@ -31,8 +31,6 @@ from app.agent.middleware.activity_log import ( ) from app.agent.middleware.jobs import ( JobsMiddleware, - filter_active_jobs, - load_jobs_metadata, ) from app.agent.middleware.memory import MemoryMiddleware from app.agent.middleware.patch_tool_calls import PatchToolCallsMiddleware @@ -68,7 +66,6 @@ from app.agent.tools.impl.mcp import ( ) from app.agent.tools.impl.query_system_settings import QuerySystemSettingsTool from app.application.agent import AgentDataContext -from app.application.agenttask import get_agent_task_execution_service from app.application.messaging.chat import ( get_configured_agent_chat_persistence, get_configured_agent_chat_service, @@ -76,7 +73,6 @@ from app.application.messaging.chat import ( ) from app.application.plugin.runtime import get_plugin_manager from app.chain.agent import AgentChain -from app.foundation.identity import SYSTEM_INTERNAL_USER_ID from app.runtime.events import eventmanager from app.runtime.execution import run_in_threadpool from app.runtime.log import logger @@ -116,64 +112,6 @@ def _agent_provider_metric_type(provider: object) -> str: return "custom" -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], 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]: - """ - 在 Agent worker 中启动渠道处理状态。 - 渠道启动可能触发外部 API,同步实现需切到线程池避免阻塞事件循环。 - """ - if not task.channel: - return None - - def _start() -> Optional[dict]: - """在线程池中通过统一 Chain 接口启动处理状态。""" - try: - return 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 await run_in_threadpool(_start) - - -async def _async_finish_processing_status( - status: Optional[dict], 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 _SessionUsageSnapshot: model: Optional[str] = None @@ -399,10 +337,10 @@ class MoviePilotAgent: def __init__( self, session_id: str, - user_id: str = None, - channel: str = None, - source: str = None, - username: str = None, + user_id: Optional[str] = None, + 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, @@ -2202,8 +2140,8 @@ class MoviePilotAgent: async def process( self, message: str, - images: List[str] = None, - files: Optional[List[dict]] = None, + images: Optional[List[str]] = None, + files: Optional[List[dict[str, Any]]] = None, has_audio_input: bool = False, ) -> str: """ @@ -2618,1103 +2556,3 @@ class MoviePilotAgent: self.protected_output_callback = None logger.info(f"MoviePilot智能体已清理: session_id={self.session_id}") return True - - -@dataclass -class _MessageTask: - """ - 待处理的消息任务 - """ - - session_id: str - user_id: str - message: str - images: Optional[List[str]] = None - files: Optional[List[dict]] = 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] = 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] = 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): - 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 AgentManager: - """ - AI智能体管理器 - 同一会话的消息按顺序排队处理,不同会话之间互不影响。 - """ - - def __init__( - self, - data: Optional[AgentDataContext] = None, - memory: Optional[MemoryManager] = None, - ) -> None: - """创建管理器,并保存组合根注入的 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] = {} - # 每个会话的worker任务 - self._session_workers: Dict[str, asyncio.Task] = {} - # 每个会话最后活动时间,用于回收空闲 Agent 实例 - self._session_last_used: Dict[str, tuple[str, datetime]] = {} - self._idle_cleanup_task: Optional[asyncio.Task] = 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] = {} - self._session_cleanup_pending: set[str] = set() - self._session_deferred_cleanup_tasks: Dict[str, asyncio.Task] = {} - self._session_cancel_requested: set[str] = set() - self._close_finalizer_task: Optional[asyncio.Task] = None - self._closed = False - self._shutdown_timeout = AGENT_MANAGER_SHUTDOWN_TIMEOUT - # 接收门禁与队列写入共用一把锁,确保关闭开始后不会再创建 worker。 - self._lifecycle_lock = asyncio.Lock() - self._accepting_tasks = False - - 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) - - 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)) - ) - - async def initialize(self): - """ - 初始化管理器 - """ - 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 - - 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 _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 process_message( - self, - session_id: str, - user_id: str, - message: str, - images: List[str] = None, - files: Optional[List[dict]] = None, - has_audio_input: bool = False, - channel: str = None, - source: str = None, - username: 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 = ( - 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): - """ - 会话消息处理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, - 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): - """ - 实际处理单条消息 - """ - 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 = { - "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 = { - "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): - """ - 应急停止当前正在执行的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): - """在 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): - """ - 清空会话 - """ - 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: - """把中断或超时的清理转交给 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: - """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: - """等待超时 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 _finish_deferred_close( - self, - workers: list[tuple[str, asyncio.Task]], - ) -> 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 - - async def _wait_for_worker_shutdown( - self, - session_id: str, - worker: asyncio.Task, - *, - 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: - 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 - - 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): - """ - 心跳唤醒:检查并执行待处理的定时任务(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}") - - -# 全局智能体管理器实例 -agent_manager = AgentManager() diff --git a/app/agent/session.py b/app/agent/session.py new file mode 100644 index 000000000..44ebcbe9f --- /dev/null +++ b/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 diff --git a/app/agent/tasks.py b/app/agent/tasks.py new file mode 100644 index 000000000..cdf35c7c7 --- /dev/null +++ b/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}") diff --git a/app/runtime/compat/manifest.py b/app/runtime/compat/manifest.py index 8fb0de570..ecc2172ab 100644 --- a/app/runtime/compat/manifest.py +++ b/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 diff --git a/app/startup/initializers/agent.py b/app/startup/initializers/agent.py index 31370ac68..e44f2673b 100644 --- a/app/startup/initializers/agent.py +++ b/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 diff --git a/docs/architecture-optimization-checklist.md b/docs/architecture-optimization-checklist.md index 2a6d5efb8..df42a8bc0 100644 --- a/docs/architecture-optimization-checklist.md +++ b/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 高扇出 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index c3392b9ff..cecc58844 100644 --- a/docs/architecture-overview.md +++ b/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
生命周期显式注册/重置 Provider"] Reg --> Facade Facade -.能力启用或首次使用时物化.-> RT["app/agent/runtime_loader.py
能力发现与服务物化"] - RT --> ORC["app/agent/orchestrator.py
会话编排"] + RT --> Manager["app/agent/manager.py
稳定 AgentManager 门面"] + Manager --> Session["session.py / lifecycle.py
会话队列与有界生命周期"] + Manager --> Tasks["tasks.py
后台、调度与心跳任务"] + Session --> ORC["orchestrator.py
单 MoviePilotAgent 执行"] + Tasks --> Session ORC --> Tools["app/agent/tools
系统工具(经 application 门面)"] ORC --> LLM["app/agent/llm
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) | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index edbf9af5b..f349b8a41 100644 --- a/docs/rules/05-architecture.md +++ b/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 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 64013d270..3d5e99151 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/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", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 0486a030b..461746e43 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/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 }, diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index 611085f32..5c8d30a09 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/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 }, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 9f8df57c0..986f4ca71 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/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": { diff --git a/tests/test_agent_background_output.py b/tests/test_agent_background_output.py index af6700fa0..87f4f1347 100644 --- a/tests/test_agent_background_output.py +++ b/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 diff --git a/tests/test_agent_chat_history.py b/tests/test_agent_chat_history.py index 25d1207fc..ccce5f586 100644 --- a/tests/test_agent_chat_history.py +++ b/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, ) diff --git a/tests/test_agent_manager_facade.py b/tests/test_agent_manager_facade.py new file mode 100644 index 000000000..539a6fa59 --- /dev/null +++ b/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) diff --git a/tests/test_agent_scheduled_tasks.py b/tests/test_agent_scheduled_tasks.py index 8d1cfce52..ece72fa78 100644 --- a/tests/test_agent_scheduled_tasks.py +++ b/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 diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py index a03d257d2..a470f8796 100644 --- a/tests/test_agent_summarization_streaming.py +++ b/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] == ["数据库中的旧事实"] diff --git a/tests/test_architecture_dependencies.py b/tests/test_architecture_dependencies.py index e0ce7f88a..73c3fd0d6 100644 --- a/tests/test_architecture_dependencies.py +++ b/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" diff --git a/tests/test_message_processing_status.py b/tests/test_message_processing_status.py index 93c562d07..7d2a7c826 100644 --- a/tests/test_message_processing_status.py +++ b/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( diff --git a/tests/test_telegram_typing_lifecycle.py b/tests/test_telegram_typing_lifecycle.py index 01d838621..1f40ed36e 100644 --- a/tests/test_telegram_typing_lifecycle.py +++ b/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(