""" Agent 会话队列、worker 与资源状态的唯一 owner。 """ import asyncio from dataclasses import dataclass, field 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.agent.steering import SteeringInbox, SteeringMessage, SteeringStatusCallback from app.agent.terminal.ownership import TerminalScope, close_terminal_scope 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 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: """ 待处理的消息任务 :param session_id: 待处理消息所属的 Agent 会话标识 :param user_id: 由可信入口解析的用户标识 :param message: 用户提交的消息正文 :param images: 附带图片引用；没有图片时为 None :param files: 附带文件元数据；没有文件时为 None :param has_audio_input: 本次输入是否包含音频，用于选择回复能力 :param channel: 消息来源渠道；没有渠道上下文时为 None :param source: 渠道内的消息来源标识；无来源时为 None :param username: 用于回复和身份展示的用户名；未提供时为 None :param is_channel_admin: 可信入口确认的渠道管理员身份；尚未给定时为 None :param original_message_id: 原始渠道消息标识，用于回复关联 :param original_chat_id: 原始渠道会话标识，用于群组或线程回复 :param processing_status: 已建立的渠道处理中状态，供任务结束时收口 :param reply_mode: 回复投递方式，默认通过渠道派发 :param allow_message_tools: 是否允许本次 Agent 使用消息发送工具 :param output_callback: 接收增量文本的同步回调；未提供时为 None :param tool_event_callback: 接收工具调用事件的同步回调；未提供时为 None :param protected_output_callback: 接收敏感输出的受保护回调，返回值表示是否接收；未提供时为 None :param message_callback: 接收完整消息的回调，可返回等待对象；未提供时为 None :param agent_factory: 创建本次 Agent 的宿主工厂；未提供时使用默认工厂 :param agent_setup: Agent 创建后的宿主初始化回调；未提供时不额外设置 :param completion_future: 向排队调用方交付最终文本或异常的 Future；无需等待时为 None :param enqueued_at: 入队时的单调时钟秒数；尚未入队时为 None :param scheduled_run_id: 定时任务持久化运行标识；普通消息任务为 None :param execution_user_id: 宿主已验证的计划任务执行账号；普通会话为 None :param execution_user_identity: 执行账号的不可变身份；普通会话为 None :param steering_status_callback: 报告运行中补充消息处理状态的回调 :ivar steering_inbox: 宿主执行时绑定的补充消息队列，初始为 None，不接受构造输入 :ivar terminal_scope: 宿主装配的终端所有权作用域；定时任务在初始化时绑定 :ivar agent: 实际执行本任务的 Agent 实例；开始执行前为 None :ivar terminal_released: 终端作用域是否已释放，初始 False，供收尾流程避免重复清理 """ 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 tool_event_callback: Optional[Callable[[dict[str, Any]], 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 scheduled_run_id: Optional[str] = None execution_user_id: Optional[int] = None execution_user_identity: Optional[str] = None steering_status_callback: Optional[SteeringStatusCallback] = None steering_inbox: Optional[SteeringInbox] = field(default=None, init=False, repr=False) terminal_scope: Optional[TerminalScope] = field(default=None, init=False) agent: Optional[MoviePilotAgent] = field(default=None, init=False, repr=False) terminal_released: bool = field(default=False, init=False) def __post_init__(self) -> None: """ 由队列宿主使用持久化运行 ID 装配身份，不向模型暴露可写归属字段。 """ if self.scheduled_run_id is not None: self.terminal_scope = TerminalScope( user_id=self.user_id, task_id=self.scheduled_run_id, kind="scheduled" ) class AgentManagerUnavailableError(RuntimeError): """ AgentManager 未运行或已开始关闭，不能再接收新任务。 """ code = "agent_manager_unavailable" def __init__(self, *args: object, task_accepted: bool = True) -> None: """ 标明异常发生时是否可能已接收任务，供交互入口决定能否安全重试。 :param args: 原有异常说明，保持 RuntimeError 调用兼容 :param task_accepted: 默认按可能已执行处理；明确准入拒绝时为 False """ super().__init__(*args) self.task_accepted = task_accepted class AgentManagerQueueFullError(RuntimeError): """ Agent 会话的待处理消息达到容量上限。 """ code = "agent_manager_queue_full" # 排队已满发生在入队前，可安全重试原交互。 task_accepted = False 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_active_tasks: Dict[str, _MessageTask] = {} self._session_steering_inboxes: Dict[str, SteeringInbox] = {} 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"] = self.is_session_busy(session_id) inbox = self._session_steering_inboxes.get(session_id) status["steering_pending"] = inbox.pending_count if inbox else 0 status["steering_active"] = bool(inbox and inbox.running) 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: """ 查询实际执行或排队状态，空闲等待消息的 worker 不计入。 """ queue = self._session_queues.get(session_id) return session_id in self._session_active_tasks or bool(queue and not queue.empty()) 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, tool_event_callback: Optional[Callable[[dict[str, Any]], None]] = None, protected_output_callback: Optional[Callable[[str], Optional[bool]]] = None, message_callback: Optional[Callable[[Any], Awaitable[None] | None]] = None, steering_status_callback: Optional[SteeringStatusCallback] = None, agent_factory: Optional[Callable[..., MoviePilotAgent]] = None, agent_setup: Optional[Callable[[MoviePilotAgent], None]] = None, wait_for_completion: bool = False, scheduled_run_id: Optional[str] = None, execution_user_id: Optional[int] = None, execution_user_identity: Optional[str] = None, ) -> str: """ 处理用户消息：同渠道的运行中补充输入进入当前上下文，其余消息按会话排队。 等待完整结果的调用保留独立任务，不同会话之间互不影响。 :param execution_user_id: 宿主已验证的计划任务执行账号，不由模型或渠道输入指定 :param execution_user_identity: 对应账号的不可变身份，避免删除后的用户主键复用 """ 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, tool_event_callback=tool_event_callback, protected_output_callback=protected_output_callback, message_callback=message_callback, steering_status_callback=steering_status_callback, agent_factory=agent_factory, agent_setup=agent_setup, completion_future=completion_future, scheduled_run_id=scheduled_run_id, execution_user_id=execution_user_id, execution_user_identity=execution_user_identity, ) async with self._lifecycle_lock: if not self._accepting_tasks: if completion_future and not completion_future.done(): completion_future.cancel() raise AgentManagerUnavailableError("AgentManager 未运行或已关闭", task_accepted=False) 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} 仍在停止，暂时不能接收新任务", task_accepted=False, ) self._record_session_activity(session_id, user_id) if await self._try_steer_channel_message(task): return "" # 获取或创建会话队列 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: try: return await completion_future except asyncio.CancelledError: # 取消等待不能留下仍可执行命令的定时任务；对话 worker 可能仍在收尾。 await self._close_scheduled_task_scope(task) raise return "" async def _try_steer_channel_message(self, task: _MessageTask) -> bool: """ 在生命周期锁内将同一渠道会话的追加输入交给活动运行，保留其他调用的排队语义。 """ if ( not task.channel or task.channel in {NotificationChannel.Web.value, NotificationChannel.WebAgent.value} or task.reply_mode != ReplyMode.DISPATCH or task.completion_future is not None or task.scheduled_run_id is not None ): return False active = self._session_active_tasks.get(task.session_id) inbox = self._session_steering_inboxes.get(task.session_id) if active is None or inbox is None or active.scheduled_run_id is not None: return False identity_fields = ( "user_id", "channel", "source", "original_chat_id", "is_channel_admin", "reply_mode", "allow_message_tools", ) if any(getattr(active, name) != getattr(task, name) for name in identity_fields): return False message = await inbox.enqueue( user_id=task.user_id, text=task.message, images=task.images, files=task.files, original_message_id=task.original_message_id, ) return message is not None @staticmethod def _build_steering_status_callback(task: _MessageTask) -> SteeringStatusCallback: """ 在真实注入边界切分渠道回复，同时保留 WebAgent 等调用方的状态回调。 """ def callback(message: SteeringMessage, status: str) -> None: """ 先冻结旧回复的展示范围，再交付调用方的应用事件。 """ if ( status == "applied" and task.agent is not None and task.channel and task.channel not in {NotificationChannel.Web.value, NotificationChannel.WebAgent.value} ): task.agent.start_steering_reply(message.original_message_id) if task.steering_status_callback is not None: task.steering_status_callback(message, status) return callback async def submit_steering_message( self, *, session_id: str, user_id: str, message: str, images: Optional[List[str]] = None, files: Optional[List[dict[str, Any]]] = None, ) -> Optional[SteeringMessage]: """ 把运行中补充消息提交到当前 Agent；没有活动运行时返回 None 让入口走普通轮次。 """ async with self._lifecycle_lock: task = self._session_active_tasks.get(session_id) inbox = self._session_steering_inboxes.get(session_id) if task is None or task.user_id != str(user_id) or inbox is None: return None return await inbox.enqueue( user_id=str(user_id), text=message, images=images, files=files, ) @staticmethod async def _close_scheduled_task_scope(task: _MessageTask) -> bool: """ 取消等待与 worker 收尾共用任务归属，失败时保留已装配 Agent 的重试记录。 """ if task.terminal_scope is None or task.terminal_released: return True task.terminal_scope.seal() if task.agent is not None: released = await task.agent.release_terminal_scope(task.terminal_scope) else: released = await close_terminal_scope(task.terminal_scope) if released: task.terminal_released = True return released 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.terminal_scope is not None and task.terminal_scope.closed: if task.completion_future and not task.completion_future.done(): task.completion_future.cancel() continue 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, ) self._session_active_tasks[session_id] = task 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 not await self._close_scheduled_task_scope(task): raise AgentManagerUnavailableError("Agent 定时任务仍有终端在停止") 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, ) try: await self._close_scheduled_task_scope(task) finally: if self._session_active_tasks.get(session_id) is task: self._session_active_tasks.pop(session_id, None) 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.terminal_scope is not None: task.terminal_scope.seal() 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 ( existing_agent.user_id != task.user_id or ( 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) self._session_steering_inboxes.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: # Web 与兼容协议工厂同样属于当前宿主，必须继承其数据和记忆依赖。 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, "data": self._data, "memory": self._memory, } if task.tool_event_callback is not None: agent_kwargs["tool_event_callback"] = task.tool_event_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] # 每条队列任务都携带完整消息上下文，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 if hasattr(agent, "set_tool_event_callback"): agent.set_tool_event_callback(task.tool_event_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) # 每轮都覆盖执行主体，防止计划任务与交互消息复用会话时串用权限。 agent.execution_user_id = task.execution_user_id agent.execution_user_identity = task.execution_user_identity task.agent = agent inbox = self._session_steering_inboxes.get(session_id) if inbox is None: inbox = SteeringInbox(session_id, str(task.user_id)) self._session_steering_inboxes[session_id] = inbox task.steering_inbox = inbox configure_inbox = getattr(agent, "configure_steering_inbox", None) if callable(configure_inbox): configure_inbox(inbox) else: setattr(agent, "steering_inbox", inbox) 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 if task.terminal_scope is not None: process_kwargs["terminal_scope"] = task.terminal_scope steering_status_callback = self._build_steering_status_callback(task) await inbox.begin_run(steering_status_callback) try: result = await agent.process(task.message, **process_kwargs) # 模型最终回合返回后仍可能有一条已入 inbox 的消息；在同一 worker # 内继续处理，避免 WebAgent 已经 ACK 的补充消息变成无展示的后台轮次。 while True: pending = await inbox.consume() if not pending: break for message in pending: result = await agent.process( message.text, images=list(message.images) or None, files=list(message.files) or None, has_audio_input=any( str(file.get("mime_type") or "").startswith("audio/") for file in message.files ), terminal_scope=task.terminal_scope, ) # consume 与 finish_run 之间仍有一个极窄的入队窗口；已接受的消息 # 也在当前 worker 内执行，且先发布 applied 以完成前端展示收口。 pending = await inbox.finish_run() if pending: callback = steering_status_callback for message in pending: if callback is not None: try: callback(message, "applied") except Exception: pass result = await agent.process( message.text, images=list(message.images) or None, files=list(message.files) or None, has_audio_input=any( str(file.get("mime_type") or "").startswith("audio/") for file in message.files ), terminal_scope=task.terminal_scope, ) return result except BaseException: await inbox.finish_run() raise 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._session_steering_inboxes.pop(session_id, None) 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._session_steering_inboxes.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