""" MoviePilot 子代理中间件适配。 """ import asyncio import json import uuid from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from datetime import datetime from functools import lru_cache from typing import Any, Literal, Optional, cast from langchain.agents import create_agent from langchain.agents.middleware.types import ( AgentMiddleware, ContextT, ModelRequest, ModelResponse, ResponseT, ToolCallRequest, ) from langchain_core.language_models.chat_models import BaseChatModel from langchain_core.messages import AIMessage, HumanMessage from langchain_core.tools import BaseTool, StructuredTool from pydantic import BaseModel, Field from app.agent.llm.helper import LLMHelper from app.agent.middleware.guardrails import ToolGuardrailsMiddleware from app.agent.middleware.policy import AgentPolicyMiddleware from app.agent.middleware.summarization import ( ContextPreservingSummarizationMiddleware, FinalRequestCompactionMiddleware, ) from app.agent.middleware.terminal import SubAgentTerminalGrant, subagent_terminal_scope from app.agent.middleware.utils import append_to_system_message from app.agent.middleware.vision import VisionMiddleware from app.agent.policy.contracts import ( AuthSource, PrincipalType, ToolOrigin, ToolPolicyContext, ) from app.agent.policy.sanitizer import ( sanitize_for_host, summarize_error, ) from app.agent.runtime import SubAgentDefinition, agent_runtime_manager from app.agent.tools.catalog import ToolCatalogSnapshot from app.agent.tools.tags import ToolTag from app.runtime.log import logger SUBAGENT_TASK_TOOL_NAME = "task" SUBAGENT_CONTROL_TOOL_NAME = "subagent_task" SUBAGENT_STREAM_MARKER_KEY = "ls_agent_type" SUBAGENT_STREAM_MARKER_VALUE = "subagent" SUBAGENT_DEFAULT_WAIT_TIMEOUT_MS = 60000 SUBAGENT_MAX_WAIT_TIMEOUT_MS = 300000 SUBAGENT_CANCEL_GRACE_SECONDS = 5.0 SUBAGENT_MAX_ACTIVE_TASKS = 8 SUBAGENT_MAX_CONCURRENT_TASKS = 4 SUBAGENT_RESULT_MAX_CHARS = 12000 SUBAGENT_DESCRIPTION_MAX_CHARS = 500 SUBAGENT_PIPELINE_CONTEXT_MAX_CHARS = 12000 SUBAGENT_PARENT_PROMPT = """ You may use subagent tools to delegate independent research, retrieval, diagnosis, or planning work to built-in subagents. Delegation modes: - Use `task` for one blocking subtask when you need the result immediately. - Use `subagent_task` for two or more independent subtasks. Start them first with `action=start` and a `tasks` array, then use `action=status`, `action=wait`, `action=update`, or `action=cancel` with the returned task IDs. - Use `subagent_task` with `action=run` when you want to launch a bounded batch and wait for the batch in one tool call. - Use `subagent_task` with `action=pipeline` when later subtasks must use previous subagent results. Pipeline steps run sequentially, and each step's result is passed as private context to the next step. Rules: - Delegate when a task benefits from focused investigation, such as media identity checks, site/resource search, subscription analysis, download/transfer diagnosis, MoviePilot code/config exploration, or read-only system inspection. - Subagent output is private context for your decision-making. Do not expose a subagent's process or final report verbatim to the user. - Subagents must not send messages to the user, ask for interaction, or reveal their internal tool activity. - Synthesize child evidence into your own progress updates and final answer, following the main agent's communication rules. Distinguish verified findings from child hypotheses; delegation does not justify silent work or premature completion. - If a task requires configuration changes, deletion, adding downloads, adding subscriptions, or any high-impact action, the main agent must handle it directly under the confirmation policy. - Child tools enforce read-only operations. Perform command launches, browser navigation/interactions, and external MCP calls in the main agent; pass the resulting evidence to a child for analysis when useful. - To let a child inspect a parent terminal, declare `terminal_sessions=[{session_id, actions:["read","wait"]}]` on that task. Mentioning a handle in its description does not grant access. Share separately for each batch or pipeline task; process control remains with the parent. """ SUBAGENT_TASK_DESCRIPTION = ( "将独立的 MoviePilot 调查或规划任务交给内置子智能体。" "子智能体结果仅作为主智能体的私有上下文，不得原样转发给用户。" ) SUBAGENT_CONTROL_DESCRIPTION = ( "异步启动和管理多个 MoviePilot 子任务。" "使用 action=start 和 tasks=[{description}] 启动批次并立即取得任务标识。" "action=status 查看状态；" "action=wait 等待全部或任一结果；" "action=update 携带 task_id 和新 description，在有界取消后替换任务；" "action=cancel 停止任务；" "action=run 在一次调用中启动并等待有界批次。" "action=pipeline 顺序执行任务，并把每步结果作为下一步的私有上下文。" ) SUBAGENT_BASE_PROMPT = """You are a silent subagent working for the MoviePilot main agent. Requirements: - Handle only the delegated subtask from the main agent. Do not converse with the user. - Do not send messages, request user interaction, or output progress updates. - Use tool results only for analysis, and return the final result only to the main agent. - Limit yourself to the read-only operations permitted by the host. Command launches, browser navigation/interactions, external MCP calls, and writes must be handled by the main agent; request the resulting evidence instead of bypassing a denial. - If user confirmation or a high-impact change is needed, explain why the main agent must confirm it instead of executing it yourself. - Return a concise structured Chinese result with key evidence, judgment, and recommended next step. """ def _default_subagent_policy_context(tools: list[BaseTool]) -> ToolPolicyContext: """ 从已注入工具继承会话归属，确保独立构造的子图也经过宿主策略。 """ for tool in tools: session_id = getattr(tool, "_session_id", None) user_id = getattr(tool, "_user_id", None) if not session_id and not user_id: continue agent_context = getattr(tool, "_agent_context", None) if not isinstance(agent_context, dict): agent_context = {} return ToolPolicyContext( session_id=str(session_id or "subagent"), user_id=str(user_id or "subagent"), origin=ToolOrigin.SUBAGENT, principal_type=PrincipalType.SUBAGENT, auth_source=AuthSource.INTERNAL, agent_context=agent_context, channel=getattr(tool, "_channel", None), source=getattr(tool, "_source", None), ) return ToolPolicyContext( session_id="subagent", user_id="subagent", origin=ToolOrigin.SUBAGENT, principal_type=PrincipalType.SUBAGENT, auth_source=AuthSource.INTERNAL, agent_context={}, ) @dataclass(frozen=True) class _SubAgentProfile: """ 子代理运行时定义。 :param name: 内置子智能体的可调用名称 :param description: 子智能体职责和适用任务摘要 :param prompt: 创建子智能体时注入的基础提示词 :param include_tags: 可进入子智能体工具目录的标签集合 :param exclude_tags: 必须排除的工具标签集合 """ name: str description: str prompt: str include_tags: frozenset[str] exclude_tags: frozenset[str] class _TaskToolInput(BaseModel): """ 子代理任务工具输入。 """ description: str = Field(..., description="需要子智能体处理的完整任务描述") terminal_sessions: list[SubAgentTerminalGrant] = Field( default_factory=list, description="明确共享给此子智能体的父任务终端，仅用于只读检查", ) class _SubAgentTaskSpec(BaseModel): """ 异步子代理任务定义。 """ description: str = Field(..., description="需要子智能体处理的完整任务描述") terminal_sessions: list[SubAgentTerminalGrant] = Field( default_factory=list, description="仅对当前任务生效的父终端授权；同批其他任务不继承", ) class _SubAgentControlInput(BaseModel): """ 异步子代理管控工具输入。 """ action: Literal["start", "status", "wait", "cancel", "update", "run", "pipeline"] = Field( default="start", description="任务动作：start、status、wait、cancel、update、run 或 pipeline，默认 start", ) description: Optional[str] = Field( default=None, description="action=start 或 action=run 时的单任务描述", ) terminal_sessions: list[SubAgentTerminalGrant] = Field( default_factory=list, description="单任务 description 使用的终端授权；使用 tasks 时须在各任务内分别指定", ) tasks: Optional[list[_SubAgentTaskSpec]] = Field( default=None, description="action=start 或 action=run 时的批量任务定义，也用于 pipeline 的顺序步骤", ) task_ids: Optional[list[str]] = Field( default=None, description="action=start 返回的任务标识；空列表表示所有已知任务", ) task_id: Optional[str] = Field( default=None, description="status、wait、cancel 或 update 使用的单个任务标识", ) wait_mode: Literal["all", "any"] = Field( default="all", description="action=wait 或 action=run 的等待方式：all 等待全部，any 等待任一任务", ) timeout_ms: Optional[int] = Field( default=SUBAGENT_DEFAULT_WAIT_TIMEOUT_MS, description=( "action=wait、action=run 或 pipeline 每一步的最长等待毫秒数，默认 60000" ), ) @dataclass class _SubAgentRuntimeTask: """ 运行中的异步子代理任务记录。 :param task_id: 宿主分配的子任务唯一标识 :param description: 派发给子智能体的完整任务描述 :param subagent_type: 执行该任务的内置子智能体类型 :param task: 负责执行子任务的异步任务对象 :param created_at: 任务登记创建的时间 :param started_at: 实际开始执行的时间；排队期间为 None :param finished_at: 任务完成或取消后的结束时间；尚未结束时为 None :param terminal_sessions: 仅对本子任务生效的父终端只读授权列表，其他子任务不继承 """ task_id: str description: str subagent_type: str task: asyncio.Task[Any] created_at: datetime started_at: Optional[datetime] = None finished_at: Optional[datetime] = None terminal_sessions: list[SubAgentTerminalGrant] = field(default_factory=list) def is_subagent_stream_metadata(metadata: Any) -> bool: """ 判断流式 token 元数据是否来自子代理。 """ if not isinstance(metadata, dict): return False if metadata.get(SUBAGENT_STREAM_MARKER_KEY) == SUBAGENT_STREAM_MARKER_VALUE: return True nested_metadata = metadata.get("metadata") if isinstance(nested_metadata, dict) and nested_metadata.get( SUBAGENT_STREAM_MARKER_KEY ) == SUBAGENT_STREAM_MARKER_VALUE: return True configurable = metadata.get("configurable") if isinstance(configurable, dict) and configurable.get( SUBAGENT_STREAM_MARKER_KEY ) == SUBAGENT_STREAM_MARKER_VALUE: return True return bool( metadata.get("lc_agent_name") in builtin_subagent_names(agent_runtime_manager.current_signature()) ) def builtin_subagent_names( runtime_signature: Optional[tuple[tuple[str, int, int], ...]] = None, ) -> frozenset[str]: """ 返回内置子代理名称集合。 """ runtime_signature = runtime_signature or agent_runtime_manager.current_signature() return _cached_builtin_subagent_names(runtime_signature) @lru_cache(maxsize=8) def _cached_builtin_subagent_names( runtime_signature: tuple[tuple[str, int, int], ...], ) -> frozenset[str]: """ 按运行时签名缓存内置子代理名称集合。 """ return frozenset( profile.name for profile in _builtin_subagent_profiles(runtime_signature) ) def _builtin_subagent_profiles( runtime_signature: Optional[tuple[tuple[str, int, int], ...]] = None, ) -> tuple[_SubAgentProfile, ...]: """ 从运行时配置目录加载 MoviePilot 子代理定义。 """ runtime_signature = runtime_signature or agent_runtime_manager.current_signature() return _cached_builtin_subagent_profiles(runtime_signature) def _default_subagent_profiles( runtime_signature: Optional[tuple[tuple[str, int, int], ...]] = None, ) -> tuple[_SubAgentProfile, ...]: """ 返回生产主 Agent 默认可见的通用子代理目录。 """ profiles = tuple( profile for profile in _builtin_subagent_profiles(runtime_signature) if profile.name == "general-purpose" ) if not profiles: raise RuntimeError("运行时未定义 general-purpose 子代理") return profiles @lru_cache(maxsize=8) def _cached_builtin_subagent_profiles( runtime_signature: tuple[tuple[str, int, int], ...], ) -> tuple[_SubAgentProfile, ...]: """ 按运行时签名缓存 MoviePilot 子代理定义。 """ definitions = agent_runtime_manager.list_subagents() profiles = tuple( _profile_from_runtime_definition(definition) for definition in definitions ) if profiles: return profiles raise RuntimeError("运行时未加载任何子代理定义") cast(Any, builtin_subagent_names).cache_clear = cast(Any, _cached_builtin_subagent_names).cache_clear cast(Any, _builtin_subagent_profiles).cache_clear = cast(Any, _cached_builtin_subagent_profiles).cache_clear def _profile_from_runtime_definition( definition: SubAgentDefinition, ) -> _SubAgentProfile: """ 把运行时子代理定义转换为中间件可用的 profile。 """ prompt_parts = [SUBAGENT_BASE_PROMPT] if definition.text.strip(): prompt_parts.append(definition.text.strip()) return _SubAgentProfile( name=definition.subagent_id, description=definition.description, prompt="\n".join(prompt_parts), include_tags=frozenset(definition.include_tags), exclude_tags=frozenset(definition.exclude_tags), ) def _tool_tag_values(tool: BaseTool) -> set[str]: """ 读取工具实例上的标签集合。 """ tags = getattr(tool, "tags", None) or [] if isinstance(tags, str): return {tags} return {str(tag) for tag in tags if tag} def _select_tools(tools: list[BaseTool], profile: _SubAgentProfile) -> list[BaseTool]: """ 根据工具标签筛选子代理可用工具。 """ selected_tools = [] for tool in tools: tags = _tool_tag_values(tool) if ToolTag.Read.value not in tags: continue if profile.exclude_tags & tags: continue if profile.include_tags & tags: selected_tools.append(tool) return selected_tools def _format_subagent_catalog(profiles: tuple[_SubAgentProfile, ...]) -> str: """ 渲染子代理目录供任务工具描述使用。 """ return "\n".join( f"- {profile.name}: {profile.description}" for profile in profiles ) def _extract_final_text(result: Any) -> str: """ 从子代理执行结果中提取最后一条 AI 文本。 """ if isinstance(result, dict): messages = result.get("messages") or [] else: messages = getattr(result, "messages", []) or [] for message in reversed(messages): if isinstance(message, AIMessage) and message.content: text = LLMHelper.extract_text_content(message.content).strip() if text: return text return LLMHelper.extract_text_content(result, fallback_to_string=True).strip() def _clip_text(text: Any, max_chars: int) -> tuple[str, bool]: """ 裁剪过长文本，返回文本和是否被裁剪。 """ normalized = "" if text is None else str(text) if len(normalized) <= max_chars: return normalized, False return normalized[:max_chars], True def _format_datetime(value: Optional[datetime]) -> Optional[str]: """ 格式化任务时间。 """ if not value: return None return value.strftime("%Y-%m-%d %H:%M:%S") def _extract_tool_call_args(request: ToolCallRequest) -> dict[str, Any]: """ 提取工具调用参数，并规整为字典。 """ tool_call = request.tool_call or {} tool_args = tool_call.get("args") or {} if not isinstance(tool_args, dict): return {} return tool_args def _record_subagent_tool_call( *, stream_handler: Any, tool_name: str, tool_args: dict[str, Any], ) -> str: """ 按当前详细模式展示或汇总子代理工具调用。 """ if not stream_handler or not getattr(stream_handler, "is_streaming", False): return "" safe_args = sanitize_for_host(tool_args) if not isinstance(safe_args, dict): safe_args = {} if tool_name == SUBAGENT_TASK_TOOL_NAME: tool_message = "调用子代理：general-purpose" else: tool_message = f"管理子代理任务：action={safe_args.get('action') or 'start'}" return str(stream_handler.report_tool_call( tool_name=tool_name, tool_message=tool_message, tool_kwargs=tool_args, ) or "") class _SubAgentAgentProvider: """ 子代理图懒加载与执行器。 """ def __init__( self, *, model: BaseChatModel, profiles: tuple[_SubAgentProfile, ...], tools: list[BaseTool], server_tools: Optional[list[dict[str, Any]]] = None, policy_context: Optional[ToolPolicyContext] = None, catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """ 初始化子代理执行器。 """ self._model = model self._profiles = {profile.name: profile for profile in profiles} self._tools = tools self._server_tools = server_tools or [] self._policy_context = policy_context or _default_subagent_policy_context(tools) self._catalog = catalog self._agents: dict[str, Any] = {} self._guardrails: dict[str, ToolGuardrailsMiddleware] = {} self._default_agent_name = "general-purpose" def _resolve_profile(self, _agent_name: Optional[str] = None) -> _SubAgentProfile: """ 解析唯一的通用子代理。 """ return self._profiles[self._default_agent_name] def get_agent(self, agent_name: Optional[str] = None) -> tuple[str, Any]: """ 懒加载唯一通用子代理图。 """ profile = self._resolve_profile(agent_name) cached_agent = self._agents.get(profile.name) if cached_agent: return profile.name, cached_agent subagent_tools = _select_tools(self._tools, profile) subagent_catalog = ( self._catalog.select(subagent_tools) if self._catalog is not None else None ) logger.info( f"创建子代理图: subagent_type={profile.name}, tools={len(subagent_tools)}" ) guardrails = ToolGuardrailsMiddleware(self._policy_context) middlewares: list[AgentMiddleware[Any, Any, Any]] = [ AgentPolicyMiddleware( context=self._policy_context, catalog=subagent_catalog, guardrails=guardrails, ), guardrails, FinalRequestCompactionMiddleware( summarizer=ContextPreservingSummarizationMiddleware( model=self._model, keep=("messages", 20), ), ), VisionMiddleware(), ] agent = create_agent( model=self._model, tools=[*subagent_tools, *self._server_tools], system_prompt=profile.prompt, name=profile.name, middleware=middlewares, ) self._agents[profile.name] = agent self._guardrails[profile.name] = guardrails return profile.name, agent async def run_task( self, *, description: str, subagent_type: Optional[str] = None, task_id: Optional[str] = None, terminal_sessions: Optional[list[SubAgentTerminalGrant]] = None, ) -> str: """ 调用通用子代理并只返回供主代理读取的结果。 """ del subagent_type agent_name, agent = self.get_agent() thread_suffix = task_id or uuid.uuid4().hex log_task_id = task_id or "-" logger.info( f"开始调用子代理: subagent_type={agent_name}, task_id={log_task_id}" ) try: async with subagent_terminal_scope( task_id=thread_suffix, user_id=self._policy_context.user_id, terminal_sessions=terminal_sessions, ) as terminal_context: result = await agent.ainvoke( {"messages": [HumanMessage(content=description + terminal_context)]}, config={ "configurable": { "thread_id": f"subagent-{agent_name}-{thread_suffix}", SUBAGENT_STREAM_MARKER_KEY: SUBAGENT_STREAM_MARKER_VALUE, }, "metadata": { "lc_agent_name": agent_name, SUBAGENT_STREAM_MARKER_KEY: SUBAGENT_STREAM_MARKER_VALUE, }, }, ) except Exception as err: logger.error( f"子代理调用失败: subagent_type={agent_name}, " f"task_id={log_task_id}, error={summarize_error(err)}" ) raise finally: guardrails = self._guardrails.get(agent_name) if guardrails: guardrails.release(f"subagent-{agent_name}-{thread_suffix}") final_text = _extract_final_text(result) logger.info( f"子代理调用完成: subagent_type={agent_name}, " f"task_id={log_task_id}, result_chars={len(final_text)}" ) return final_text or "The subagent did not return a usable result." class MoviePilotSubAgentMiddleware(AgentMiddleware): """ MoviePilot 本地子代理中间件兜底实现。 """ def __init__( self, *, model: BaseChatModel, profiles: tuple[_SubAgentProfile, ...], tools: list[BaseTool], server_tools: Optional[list[dict[str, Any]]] = None, system_prompt: str = SUBAGENT_PARENT_PROMPT, task_description: str = SUBAGENT_TASK_DESCRIPTION, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """ 初始化同步子代理中间件。 """ self.system_prompt = system_prompt self.stream_handler = stream_handler self._provider = _SubAgentAgentProvider( model=model, profiles=profiles, tools=tools, server_tools=server_tools, policy_context=policy_context, catalog=catalog, ) self.tools = [ StructuredTool.from_function( coroutine=self._run_task, name=SUBAGENT_TASK_TOOL_NAME, description=( f"{task_description}\n\n可用子智能体：\n" f"{_format_subagent_catalog(profiles)}" ), args_schema=_TaskToolInput, ) ] def _get_agent(self, agent_name: str) -> Any: """ 懒加载指定名称的子代理图。 """ return self._provider.get_agent(agent_name)[1] async def _run_task( self, description: str, terminal_sessions: Optional[list[SubAgentTerminalGrant]] = None, ) -> str: """ 调用通用子代理并只返回供主代理读取的结果。 """ return await self._provider.run_task( description=description, terminal_sessions=terminal_sessions, ) async def awrap_model_call( self, request: ModelRequest[ContextT], handler: Callable[ [ModelRequest[ContextT]], Awaitable[ModelResponse[ResponseT]] ], ) -> ModelResponse[ResponseT]: """ 在主代理模型调用前注入子代理使用说明。 """ new_system_message = append_to_system_message( request.system_message, self.system_prompt, ) return await handler(request.override(system_message=new_system_message)) async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]], ) -> Any: """ 在 task 子代理工具执行时输出当前模式对应的执行信息。 """ tool = request.tool tool_name = getattr(tool, "name", None) if tool_name != SUBAGENT_TASK_TOOL_NAME: return await handler(request) tool_args = _extract_tool_call_args(request) logger.info( f"开始执行子代理工具: tool_name={tool_name}, " "subagent_type=general-purpose" ) tool_call_id = _record_subagent_tool_call( stream_handler=self.stream_handler, tool_name=SUBAGENT_TASK_TOOL_NAME, tool_args=tool_args, ) try: result = await handler(request) except Exception as err: if tool_call_id: finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) if callable(finish_tool_call): finish_tool_call(tool_call_id, "error") logger.error( f"子代理工具执行失败: tool_name={tool_name}, " f"error={summarize_error(err)}" ) raise if tool_call_id: finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) if callable(finish_tool_call): finish_tool_call(tool_call_id, "done") logger.info(f"子代理工具执行完成: tool_name={tool_name}") return result class SubAgentTaskControlMiddleware(AgentMiddleware): """ 提供异步子代理任务调度工具的中间件。 """ def __init__( self, *, model: BaseChatModel, profiles: tuple[_SubAgentProfile, ...], tools: list[BaseTool], server_tools: Optional[list[dict[str, Any]]] = None, task_description: str = SUBAGENT_CONTROL_DESCRIPTION, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, catalog: Optional[ToolCatalogSnapshot] = None, ) -> None: """ 初始化异步子代理调度中间件。 """ self.stream_handler = stream_handler self._provider = _SubAgentAgentProvider( model=model, profiles=profiles, tools=tools, server_tools=server_tools, policy_context=policy_context, catalog=catalog, ) self._semaphore = asyncio.Semaphore(SUBAGENT_MAX_CONCURRENT_TASKS) self._tasks: dict[str, _SubAgentRuntimeTask] = {} self._accepting_tasks = True self._close_cancel_requested: set[asyncio.Task[Any]] = set() self.tools = [ StructuredTool.from_function( coroutine=self._control_task, name=SUBAGENT_CONTROL_TOOL_NAME, description=( f"{task_description}\n\n可用子智能体：\n" f"{_format_subagent_catalog(profiles)}" ), args_schema=_SubAgentControlInput, ) ] @staticmethod def _json_response(payload: dict[str, Any]) -> str: """ 将工具响应序列化为稳定 JSON，并为失败提供模型恢复路径。 """ normalized = dict(payload) if normalized.get("success") is False: normalized.setdefault("execution_outcome", "failed") normalized.setdefault( "recovery", "根据 error 或 tasks 中的状态修正任务描述或参数；不要重复启动已在途的子代理。", ) return json.dumps(normalized, ensure_ascii=False, indent=2) @staticmethod def _normalize_timeout_ms(timeout_ms: Optional[int]) -> int: """ 规范化等待超时时间。 """ if timeout_ms is None: return SUBAGENT_DEFAULT_WAIT_TIMEOUT_MS return max(0, min(int(timeout_ms), SUBAGENT_MAX_WAIT_TIMEOUT_MS)) @staticmethod def _task_status(record: _SubAgentRuntimeTask) -> str: """ 读取任务当前状态。 """ task = record.task if task.cancelled(): return "cancelled" if not task.done(): return "running" if record.started_at else "pending" if task.exception(): return "failed" return "completed" @staticmethod def _task_output(record: _SubAgentRuntimeTask) -> dict[str, Any]: """ 格式化单个任务状态和结果。 """ description, description_truncated = _clip_text( record.description, SUBAGENT_DESCRIPTION_MAX_CHARS, ) payload: dict[str, Any] = { "task_id": record.task_id, "subagent_type": record.subagent_type, "status": SubAgentTaskControlMiddleware._task_status(record), "description": description, "description_truncated": description_truncated, "created_at": _format_datetime(record.created_at), "started_at": _format_datetime(record.started_at), "finished_at": _format_datetime(record.finished_at), } if not record.task.done(): payload["execution_outcome"] = "pending" payload["recovery"] = "使用 action=wait 或 action=status 继续观察，不要重复启动同一任务。" return payload if record.task.cancelled(): payload["execution_outcome"] = "failed" payload["recovery"] = "任务已取消；检查其他任务结果，只有在需要时用新的 task_id 重新派发。" return payload error = record.task.exception() if error: payload["error"] = summarize_error(error) payload["execution_outcome"] = "failed" payload["recovery"] = "根据子代理错误修正任务描述或参数后重试；不要重复启动仍在途的任务。" return payload result, result_truncated = _clip_text( record.task.result(), SUBAGENT_RESULT_MAX_CHARS, ) payload["result"] = result payload["result_truncated"] = result_truncated return payload def _selected_records( self, *, task_ids: Optional[list[str]] = None, task_id: Optional[str] = None, active_only: bool = False, ) -> tuple[list[_SubAgentRuntimeTask], list[str]]: """ 根据任务 ID 选择记录。 """ selected_ids = [] if task_id: selected_ids.append(task_id) selected_ids.extend(task_ids or []) if not selected_ids: records = list(self._tasks.values()) if active_only: records = [record for record in records if not record.task.done()] return records, [] records = [] missing_ids = [] seen_ids = set() for selected_id in selected_ids: if selected_id in seen_ids: continue seen_ids.add(selected_id) record = self._tasks.get(selected_id) if record: records.append(record) else: missing_ids.append(selected_id) return records, missing_ids def _normalize_specs( self, *, description: Optional[str], tasks: Optional[list[_SubAgentTaskSpec]], terminal_sessions: Optional[list[SubAgentTerminalGrant]] = None, ) -> tuple[list[_SubAgentTaskSpec], Optional[str]]: """ 规范化单任务和批量任务输入。 """ if tasks and terminal_sessions: return [], "批量或管道任务请在每个 tasks 条目中单独声明 terminal_sessions。" specs = [] for task in tasks or []: if isinstance(task, dict): task = _SubAgentTaskSpec(**task) if task.description.strip(): specs.append(task) if not specs and description and description.strip(): specs.append( _SubAgentTaskSpec( description=description, terminal_sessions=terminal_sessions or [], ) ) if not specs: return [], "缺少可执行的子代理任务描述。" if len(specs) > SUBAGENT_MAX_ACTIVE_TASKS: return [], f"单次最多可提交 {SUBAGENT_MAX_ACTIVE_TASKS} 个子代理任务。" active_count = sum( 1 for record in self._tasks.values() if not record.task.done() ) if active_count + len(specs) > SUBAGENT_MAX_ACTIVE_TASKS: return [], ( f"当前仍有 {active_count} 个子代理任务未完成，" f"总并发上限为 {SUBAGENT_MAX_ACTIVE_TASKS}。" ) return specs, None async def _execute_managed_task(self, record: _SubAgentRuntimeTask) -> str: """ 执行受调度器管理的子代理任务。 """ async with self._semaphore: record.started_at = datetime.now() logger.info( f"异步子代理任务开始执行: task_id={record.task_id}, " f"subagent_type={record.subagent_type}" ) try: result = await self._provider.run_task( description=record.description, subagent_type=record.subagent_type, task_id=record.task_id, terminal_sessions=record.terminal_sessions, ) logger.info( f"异步子代理任务执行完成: task_id={record.task_id}, " f"subagent_type={record.subagent_type}, result_chars={len(result)}" ) return result except asyncio.CancelledError: logger.info( f"异步子代理任务已取消: task_id={record.task_id}, " f"subagent_type={record.subagent_type}" ) raise except Exception as err: logger.error( f"子代理任务执行失败: task_id={record.task_id}, " f"error={summarize_error(err)}" ) raise def _mark_task_finished(self, task_id: str, task: asyncio.Task[Any]) -> None: """ 记录任务完成时间并取出异常避免未读取告警。 """ cancel_requested = getattr(self, "_close_cancel_requested", None) if cancel_requested is not None: cancel_requested.discard(task) record = self._tasks.get(task_id) if record and record.task is task: record.finished_at = datetime.now() if task.cancelled(): return try: task.exception() except Exception: return def _start_tasks(self, specs: list[_SubAgentTaskSpec]) -> list[_SubAgentRuntimeTask]: """ 启动一批异步子代理任务。 """ records = [] for spec in specs: task_id = f"subagent-{uuid.uuid4().hex[:12]}" record = _SubAgentRuntimeTask( task_id=task_id, description=spec.description.strip(), subagent_type="general-purpose", task=cast(asyncio.Task[Any], None), created_at=datetime.now(), terminal_sessions=spec.terminal_sessions, ) task = asyncio.create_task( self._execute_managed_task(record), name=task_id, ) record.task = task task.add_done_callback(self._task_finished_callback(task_id)) self._tasks[task_id] = record records.append(record) logger.info( f"已启动子代理任务: task_id={task_id}, " f"subagent_type={record.subagent_type}" ) return records async def _wait_records( self, *, records: list[_SubAgentRuntimeTask], wait_mode: str, timeout_ms: Optional[int], ) -> None: """ 按等待模式等待一组任务完成。 """ pending_tasks = [record.task for record in records if not record.task.done()] if not pending_tasks: if records: logger.info(f"子代理任务无需等待: tasks={len(records)}") return normalized_timeout_ms = self._normalize_timeout_ms(timeout_ms) timeout = normalized_timeout_ms / 1000 if timeout <= 0: logger.info( f"子代理任务等待超时时间为 0，跳过等待: tasks={len(pending_tasks)}" ) return return_when = ( asyncio.FIRST_COMPLETED if wait_mode == "any" else asyncio.ALL_COMPLETED ) logger.info( f"开始等待子代理任务: tasks={len(pending_tasks)}, " f"wait_mode={wait_mode}, timeout_ms={normalized_timeout_ms}" ) await asyncio.wait( pending_tasks, timeout=timeout, return_when=return_when, ) finished_count = sum(1 for task in pending_tasks if task.done()) logger.info( f"子代理任务等待结束: finished={finished_count}, " f"pending={len(pending_tasks) - finished_count}" ) @staticmethod async def _cancel_records( records: list[_SubAgentRuntimeTask], *, cancel_requested: Optional[set[asyncio.Task[Any]]] = None, ) -> list[_SubAgentRuntimeTask]: """ 取消一组任务，并返回等待上限内仍未收敛的记录。 """ cancellable_tasks = [ record.task for record in records if not record.task.done() ] if cancellable_tasks: logger.info(f"开始取消子代理任务: tasks={len(cancellable_tasks)}") for task in cancellable_tasks: if cancel_requested is not None and task in cancel_requested: continue task.cancel() if cancel_requested is not None: cancel_requested.add(task) if not cancellable_tasks: return [] done, pending = await asyncio.wait( cancellable_tasks, timeout=SUBAGENT_CANCEL_GRACE_SECONDS, ) if done: await asyncio.gather(*done, return_exceptions=True) if pending: logger.warning( f"子代理任务取消等待超时: pending={len(pending)}, " f"timeout={SUBAGENT_CANCEL_GRACE_SECONDS}s" ) else: logger.info(f"子代理任务取消完成: tasks={len(cancellable_tasks)}") return [record for record in records if record.task in pending] async def _update_record( self, record: _SubAgentRuntimeTask, *, description: str, terminal_sessions: Optional[list[SubAgentTerminalGrant]], ) -> tuple[bool, Optional[str]]: """ 有界取消旧任务后复用 task_id 启动更新后的子代理任务。 """ if not self._accepting_tasks or self._tasks.get(record.task_id) is not record: return False, "子代理任务控制器正在关闭或任务已被回收，更新被拒绝。" if not record.task.done(): pending = await self._cancel_records([record]) if pending: return False, "旧子代理任务未能在取消窗口内收敛，更新被拒绝。" if not self._accepting_tasks or self._tasks.get(record.task_id) is not record: return False, "子代理任务控制器正在关闭或任务已被回收，更新被拒绝。" record.description = description.strip() record.terminal_sessions = terminal_sessions or [] record.started_at = None record.finished_at = None record.task = asyncio.create_task( self._execute_managed_task(record), name=record.task_id, ) record.task.add_done_callback(self._task_finished_callback(record.task_id)) return True, None def _task_finished_callback( self, task_id: str, ) -> Callable[[asyncio.Task[Any]], None]: """ 返回带稳定 task_id 的完成回调，供任务更新复用。 """ return lambda finished_task: self._mark_task_finished(task_id, finished_task) def seal(self) -> None: """ 封住新的 detached 子代理提交，既有任务继续由记录表持有。 """ self._accepting_tasks = False async def close(self) -> bool: """ 有限等待 detached 子代理；超时保留记录并返回 False。 """ self.seal() return await self._drain_tasks() async def _drain_tasks(self) -> bool: """ 清理本轮子任务，正常收敛后允许缓存图继续处理下一轮。 """ if not hasattr(self, "_close_cancel_requested"): self._close_cancel_requested = set() unfinished_records = [ record for record in self._tasks.values() if not record.task.done() ] if unfinished_records: logger.info( f"关闭子代理任务控制器，取消未完成任务: tasks={len(unfinished_records)}" ) pending_records = await self._cancel_records( unfinished_records, cancel_requested=self._close_cancel_requested, ) if pending_records: # 未收敛的子任务仍由该 owner 持有，不能让下一轮叠加新任务。 self.seal() self._tasks = { record.task_id: record for record in pending_records } return False self._tasks.clear() self._close_cancel_requested.clear() return True @staticmethod def _pipeline_description( *, description: str, previous_results: list[tuple[_SubAgentRuntimeTask, str]], ) -> str: """ 追加上游子代理结果，生成当前管道步骤的任务描述。 """ normalized_description = description.strip() if not previous_results: return normalized_description context_parts = [] for step_index, (record, result) in enumerate(previous_results, start=1): clipped_result, result_truncated = _clip_text( result, SUBAGENT_RESULT_MAX_CHARS, ) truncated_note = "\n[Result truncated]" if result_truncated else "" context_parts.append( f"Step {step_index} ({record.subagent_type}) result:\n" f"{clipped_result}{truncated_note}" ) context_text, context_truncated = _clip_text( "\n\n".join(context_parts), SUBAGENT_PIPELINE_CONTEXT_MAX_CHARS, ) truncated_note = "\n[Pipeline context truncated]" if context_truncated else "" return ( f"{normalized_description}\n\n" "\n" "Previous subagent results are private context for this delegated " "subtask. Use them to complete the current task, but do not expose " "the prior reports verbatim.\n\n" f"{context_text}{truncated_note}\n" "" ) async def _execute_pipeline_task( self, *, record: _SubAgentRuntimeTask, description: str, ) -> str: """ 执行单个管道步骤，保留原始步骤描述用于状态展示。 """ async with self._semaphore: record.started_at = datetime.now() logger.info( f"管道子代理任务开始执行: task_id={record.task_id}, " f"subagent_type={record.subagent_type}" ) try: result = await self._provider.run_task( description=description, subagent_type=record.subagent_type, task_id=record.task_id, terminal_sessions=record.terminal_sessions, ) logger.info( f"管道子代理任务执行完成: task_id={record.task_id}, " f"subagent_type={record.subagent_type}, result_chars={len(result)}" ) return result except asyncio.CancelledError: logger.info( f"管道子代理任务已取消: task_id={record.task_id}, " f"subagent_type={record.subagent_type}" ) raise except Exception as err: logger.error( f"管道子代理任务执行失败: task_id={record.task_id}, " f"error={summarize_error(err)}" ) raise @staticmethod def _create_pipeline_record( spec: _SubAgentTaskSpec, ) -> _SubAgentRuntimeTask: """ 创建一个管道步骤记录。 """ task_id = f"subagent-{uuid.uuid4().hex[:12]}" return _SubAgentRuntimeTask( task_id=task_id, description=spec.description.strip(), subagent_type="general-purpose", task=cast(asyncio.Task[Any], None), created_at=datetime.now(), terminal_sessions=spec.terminal_sessions, ) def _track_pipeline_task( self, record: _SubAgentRuntimeTask, task: asyncio.Task[Any], ) -> None: """ 登记管道步骤任务，复用统一的状态和异常收口逻辑。 """ record.task = task task.add_done_callback( lambda finished_task: self._mark_task_finished( record.task_id, finished_task, ) ) self._tasks[record.task_id] = record async def _run_pipeline( self, specs: list[_SubAgentTaskSpec], timeout_ms: Optional[int], ) -> tuple[list[_SubAgentRuntimeTask], Optional[str]]: """ 按顺序执行管道任务，并把每一步结果传给下一步。 """ normalized_timeout_ms = self._normalize_timeout_ms(timeout_ms) if normalized_timeout_ms <= 0: return [], "管道任务需要大于 0 的等待时间。" records: list[_SubAgentRuntimeTask] = [] previous_results: list[tuple[_SubAgentRuntimeTask, str]] = [] timeout = normalized_timeout_ms / 1000 for step_index, spec in enumerate(specs, start=1): if not self._accepting_tasks: return records, "子代理任务控制器正在关闭，不能再启动新任务。" record = self._create_pipeline_record(spec) records.append(record) pipeline_description = self._pipeline_description( description=record.description, previous_results=previous_results, ) task = asyncio.create_task( self._execute_pipeline_task( record=record, description=pipeline_description, ), name=record.task_id, ) self._track_pipeline_task(record, task) logger.info( f"已启动管道子代理任务: step={step_index}, task_id={record.task_id}, " f"subagent_type={record.subagent_type}" ) done, pending = await asyncio.wait({task}, timeout=timeout) if pending: task.cancel() error = f"第 {step_index} 个管道子代理任务等待超时。" logger.info( f"{error} task_id={record.task_id}, timeout_ms={normalized_timeout_ms}" ) return records, error try: result = next(iter(done)).result() except Exception as err: error = ( f"第 {step_index} 个管道子代理任务执行失败: " f"{summarize_error(err)}" ) logger.info( f"{error} task_id={record.task_id}" ) return records, error previous_results.append((record, result)) return records, None async def _control_task( self, action: str = "start", description: Optional[str] = None, tasks: Optional[list[_SubAgentTaskSpec]] = None, task_ids: Optional[list[str]] = None, task_id: Optional[str] = None, wait_mode: str = "all", timeout_ms: Optional[int] = SUBAGENT_DEFAULT_WAIT_TIMEOUT_MS, terminal_sessions: Optional[list[SubAgentTerminalGrant]] = None, ) -> str: """ 管理异步子代理任务。 """ logger.info(f"收到子代理管控操作: action={action}") if action == "update": if not task_id or not description or not description.strip(): return self._json_response({ "success": False, "error": "update 必须提供 task_id 和新的 description。", }) record = self._tasks.get(task_id) if record is None: return self._json_response({ "success": False, "error": f"未知子代理任务: {task_id}", }) updated, error = await self._update_record( record, description=description, terminal_sessions=terminal_sessions, ) return self._json_response({ "success": updated, "action": action, "task_id": task_id, "error": error, "tasks": [self._task_output(record)], }) if action in {"start", "run", "pipeline"}: if not self._accepting_tasks: error = "子代理任务控制器正在关闭，不能再启动新任务。" return self._json_response({"success": False, "error": error}) specs, error = self._normalize_specs( description=description, tasks=tasks, terminal_sessions=terminal_sessions, ) if error: logger.info( f"子代理管控操作未启动任务: action={action}, " f"error={sanitize_for_host(error)}" ) return self._json_response({"success": False, "error": error}) logger.info(f"准备启动子代理任务: action={action}, tasks={len(specs)}") if action == "pipeline": records, pipeline_error = await self._run_pipeline( specs=specs, timeout_ms=timeout_ms, ) return self._json_response( { "success": pipeline_error is None, "action": action, "error": pipeline_error, "tasks": [self._task_output(record) for record in records], } ) records = self._start_tasks(specs) if action == "run": await self._wait_records( records=records, wait_mode=wait_mode, timeout_ms=timeout_ms, ) return self._json_response( { "success": True, "action": action, "wait_mode": wait_mode if action == "run" else None, "tasks": [self._task_output(record) for record in records], } ) records, missing_ids = self._selected_records( task_ids=task_ids, task_id=task_id, active_only=action in {"wait", "cancel"} and not task_ids and not task_id, ) cancellation_pending: list[_SubAgentRuntimeTask] = [] if action == "wait": logger.info( f"准备等待子代理任务: selected={len(records)}, missing={len(missing_ids)}" ) await self._wait_records( records=records, wait_mode=wait_mode, timeout_ms=timeout_ms, ) elif action == "cancel": logger.info( f"准备取消子代理任务: selected={len(records)}, missing={len(missing_ids)}" ) cancellation_pending = await self._cancel_records(records) elif action == "status": logger.info( f"查询子代理任务状态: selected={len(records)}, missing={len(missing_ids)}" ) response = { "success": not cancellation_pending, "action": action, "wait_mode": wait_mode if action == "wait" else None, "missing_task_ids": missing_ids, "tasks": [self._task_output(record) for record in records], } if action == "cancel": response["cancel_pending_task_ids"] = [ record.task_id for record in cancellation_pending ] return self._json_response(response) async def aafter_agent(self, state: Any, runtime: Any) -> None: """ 本轮结束时回收子任务；永久关闭只由会话生命周期触发。 """ await self._drain_tasks() async def awrap_tool_call( self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]], ) -> Any: """ 在 subagent_task 子代理工具执行时输出当前模式对应的执行信息。 """ tool = request.tool tool_name = getattr(tool, "name", None) if tool_name != SUBAGENT_CONTROL_TOOL_NAME: return await handler(request) tool_args = _extract_tool_call_args(request) logged_args = sanitize_for_host(tool_args) if not isinstance(logged_args, dict): logged_args = {} logger.info( f"开始执行子代理工具: tool_name={tool_name}, " f"action={logged_args.get('action') or '-'}, " "subagent_type=general-purpose" ) tool_call_id = _record_subagent_tool_call( stream_handler=self.stream_handler, tool_name=SUBAGENT_CONTROL_TOOL_NAME, tool_args=tool_args, ) try: result = await handler(request) except Exception as err: if tool_call_id: finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) if callable(finish_tool_call): finish_tool_call(tool_call_id, "error") logger.error( f"子代理工具执行失败: tool_name={tool_name}, " f"error={summarize_error(err)}" ) raise if tool_call_id: finish_tool_call = getattr(self.stream_handler, "tool_call_finished", None) if callable(finish_tool_call): finish_tool_call(tool_call_id, "done") logger.info(f"子代理工具执行完成: tool_name={tool_name}") return result def create_subagent_middlewares( *, model: BaseChatModel, tools: list[BaseTool], server_tools: Optional[list[dict[str, Any]]] = None, stream_handler: Any = None, policy_context: Optional[ToolPolicyContext] = None, catalog: Optional[ToolCatalogSnapshot] = None, ) -> tuple[list[AgentMiddleware], list[BaseTool]]: """ 创建子代理中间件列表和任务工具列表。 """ runtime_signature = agent_runtime_manager.current_signature() profiles = _default_subagent_profiles(runtime_signature) subagent_middleware = MoviePilotSubAgentMiddleware( model=model, profiles=profiles, tools=tools, server_tools=server_tools or [], stream_handler=stream_handler, policy_context=policy_context, catalog=catalog, ) control_middleware = SubAgentTaskControlMiddleware( model=model, profiles=profiles, tools=tools, server_tools=server_tools or [], stream_handler=stream_handler, policy_context=policy_context, catalog=catalog, ) task_tools = [ *list(getattr(subagent_middleware, "tools", []) or []), *list(getattr(control_middleware, "tools", []) or []), ] return [ subagent_middleware, control_middleware, ], task_tools __all__ = [ "SUBAGENT_CONTROL_TOOL_NAME", "SUBAGENT_TASK_TOOL_NAME", "SubAgentTaskControlMiddleware", "create_subagent_middlewares", "is_subagent_stream_metadata", ]