diff --git a/app/agent/middleware/patching.py b/app/agent/middleware/patching.py
index 4e4870d3d..093efa585 100644
--- a/app/agent/middleware/patching.py
+++ b/app/agent/middleware/patching.py
@@ -11,17 +11,20 @@ class PatchToolCallsMiddleware(AgentMiddleware):
@staticmethod
def _build_cancelled_tool_message(tool_call: dict[str, Any]) -> ToolMessage:
- """构造取消状态的工具响应消息。"""
+ """缺少回执只表示结果未知，不能据此重放可能已生效的写操作。"""
tool_name = tool_call.get("name") or "unknown_tool"
tool_call_id = tool_call.get("id") or ""
tool_msg = (
f"Tool call {tool_name} with id {tool_call_id} was "
- "cancelled - another message came in before it could be completed."
+ "interrupted before its result was recorded; its outcome is unknown. "
+ "It may already have changed external state. Before retrying any mutation, "
+ "verify the actual state with read-only tools and avoid duplicate side effects."
)
return ToolMessage(
content=tool_msg,
name=tool_name,
tool_call_id=tool_call_id,
+ status="error",
)
@classmethod
diff --git a/app/agent/middleware/plan.py b/app/agent/middleware/plan.py
new file mode 100644
index 000000000..367a8de54
--- /dev/null
+++ b/app/agent/middleware/plan.py
@@ -0,0 +1,229 @@
+"""在消息压缩和会话图重建之间保留当前任务的结构化计划。"""
+
+import json
+from collections.abc import Awaitable, Callable
+from typing import Annotated, Any, Literal, NotRequired, Optional, TypedDict, cast
+
+from langchain.agents.middleware.types import (
+ AgentMiddleware,
+ ModelRequest,
+ ModelResponse,
+ OmitFromInput,
+)
+from langchain.tools import ToolRuntime
+from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
+from langchain_core.tools import StructuredTool
+from langgraph.types import Command
+from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator
+
+from app.agent.middleware.utils import append_to_system_message
+from app.agent.policy.sanitizer import sanitize_for_host
+from app.agent.tools.tags import ToolTag
+
+PLAN_TOOL_NAME = "update_plan"
+PLAN_SNAPSHOT_KEY = "moviepilot_task_plan"
+MAX_PLAN_STEPS = 12
+
+PLAN_SYSTEM_PROMPT = """
+Use `update_plan` for complex work with several dependent steps or parallel investigations.
+Skip planning for simple questions and short actions. Track the user's objective, the
+remaining steps, and concise evidence or blockers. Update the complete plan when a
+meaningful stage finishes or new evidence changes the approach. Never call update_plan
+more than once in one model response.
+Preserve unfinished work when the user asks a status question or corrects the current
+task. Replace the objective and steps when the user clearly starts a different task.
+Use completed only when the step's outcome has been checked, and include the evidence.
+Use blocked with the concrete reason when progress requires missing input or an external
+change; keep failures, uncertain outcomes, and partial results visible.
+The plan records model-reported progress. It grants no permission, executes no action,
+and is not independent proof of success. Verify current state before resuming writes.
+Continue doing the work after updating the plan, and deliver the requested result in
+your final answer. A plan update alone does not complete the user's request.
+"""
+
+PLAN_TOOL_DESCRIPTION = (
+ "Record or replace the current task objective and complete step list. This only "
+ "updates private planning state; it does not perform actions or grant permission. "
+ "Use pending, in_progress, completed, or blocked. Completed and blocked steps need "
+ "concise evidence or a concrete blocker. For a new task replace the old objective "
+ "and steps. Do not call more than once in the same model response."
+)
+
+
+# follow_imports=skip 不分析第三方基类与装饰器；仅在这些 SDK 边界忽略 misc。
+class PlanStep(BaseModel): # type: ignore[misc]
+ """保存一个执行步骤及模型声明的检查结果或阻塞原因。"""
+
+ model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
+
+ step: str = Field(min_length=1, max_length=300, description="Specific action or outcome to check.")
+ status: Literal["pending", "in_progress", "completed", "blocked"]
+ evidence: str = Field(default="", max_length=600, description="Observed result or concrete blocker; never credentials.")
+
+ @model_validator(mode="after") # type: ignore[misc]
+ def require_result_context(self) -> "PlanStep":
+ """避免把没有检查依据的完成声明或无原因的阻塞写入计划。"""
+ if self.status in {"completed", "blocked"} and not self.evidence:
+ raise ValueError("completed 和 blocked 步骤必须提供 evidence，说明检查结果或具体阻塞原因")
+ return self
+
+
+class TaskPlan(BaseModel): # type: ignore[misc]
+ """限定当前任务计划的大小，避免它本身耗尽不可压缩的系统上下文。"""
+
+ # ToolNode 在 schema 校验前注入 runtime；公开 schema 不包含该内部参数。
+ model_config = ConfigDict(str_strip_whitespace=True)
+
+ objective: str = Field(min_length=1, max_length=800, description="Current user objective, preserving important constraints.")
+ steps: list[PlanStep] = Field(min_length=1, max_length=MAX_PLAN_STEPS)
+ explanation: str = Field(default="", max_length=800, description="Why the plan changed or important continuation context.")
+
+ @model_validator(mode="after") # type: ignore[misc]
+ def require_distinct_steps(self) -> "TaskPlan":
+ """同名步骤无法稳定区分进度，因此要求单份计划内步骤名称唯一。"""
+ names = [item.step.casefold() for item in self.steps]
+ if len(names) != len(set(names)):
+ raise ValueError("计划步骤名称不能重复")
+ return self
+
+
+class PlanState(TypedDict):
+ """计划独立于可压缩消息保存，只允许工具或内部历史恢复写入。"""
+
+ # create_agent 将中间件 schema 与标准 AgentState 合并，保留框架消息 reducer。
+ messages: list[BaseMessage]
+ task_plan: Annotated[NotRequired[dict[str, Any]], OmitFromInput]
+
+
+def _validated_plan(value: Any) -> Optional[dict[str, Any]]:
+ """校验并脱敏工具状态和历史快照，非法历史不会进入系统提示词。"""
+ try:
+ plan = TaskPlan.model_validate(value)
+ return cast(dict[str, Any], TaskPlan.model_validate(sanitize_for_host(plan.model_dump())).model_dump())
+ except (ValidationError, TypeError, ValueError):
+ return None
+
+
+def attach_plan_snapshot(
+ messages: list[BaseMessage], plan: Optional[dict[str, Any]],
+) -> list[BaseMessage]:
+ """把脱敏计划附在末条宿主消息上，让已有消息持久化同时保存恢复点。
+
+ 返回新列表和末条消息副本，不修改 checkpoint 中原有对象；无有效计划时
+ 原样返回。异常恢复调用方应先追加中断提示，再调用本函数。
+ """
+ snapshot = _validated_plan(plan)
+ if not messages or snapshot is None or not isinstance(messages[-1], (AIMessage, ToolMessage)):
+ return list(messages)
+ last_message = messages[-1].model_copy(
+ update={"additional_kwargs": {**messages[-1].additional_kwargs, PLAN_SNAPSHOT_KEY: snapshot}}
+ )
+ return [*messages[:-1], last_message]
+
+
+def _update_plan(
+ runtime: ToolRuntime, objective: str, steps: list[PlanStep], explanation: str = "",
+) -> Command:
+ """一次原子替换计划及工具回执，供后续模型调用和中断恢复使用。"""
+ plan = TaskPlan(objective=objective, steps=steps, explanation=explanation)
+ snapshot = _validated_plan(plan.model_dump())
+ receipt = ToolMessage(
+ content="计划已更新。步骤状态是模型声明，请继续执行并核实结果。",
+ tool_call_id=runtime.tool_call_id,
+ name=PLAN_TOOL_NAME,
+ additional_kwargs={PLAN_SNAPSHOT_KEY: snapshot},
+ )
+ return Command(update={"task_plan": snapshot, "messages": [receipt]})
+
+
+async def _aupdate_plan(
+ runtime: ToolRuntime, objective: str, steps: list[PlanStep], explanation: str = "",
+) -> Command:
+ """异步工具入口沿用相同的无 I/O 状态更新逻辑。"""
+ return _update_plan(runtime, objective, steps, explanation)
+
+
+class PlanMiddleware(AgentMiddleware): # type: ignore[misc]
+ """注入任务计划工具，并在压缩、后续用户消息和图重建后恢复工作上下文。"""
+
+ state_schema = PlanState
+
+ def __init__(self) -> None:
+ """注册只更新当前会话内部状态的计划工具。"""
+ super().__init__()
+ self.tools = [StructuredTool.from_function(
+ name=PLAN_TOOL_NAME,
+ description=PLAN_TOOL_DESCRIPTION,
+ func=_update_plan,
+ coroutine=_aupdate_plan,
+ args_schema=TaskPlan,
+ infer_schema=False,
+ tags=[ToolTag.AgentTool.value, ToolTag.AgentTask.value],
+ )]
+
+ def before_agent(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """仅在新图没有计划时从历史恢复，旧快照不能覆盖本轮更新。"""
+ if state.get("task_plan") is not None:
+ return None
+ for message in reversed(state.get("messages", [])):
+ if isinstance(message, (AIMessage, ToolMessage)) and PLAN_SNAPSHOT_KEY in message.additional_kwargs:
+ plan = _validated_plan(message.additional_kwargs[PLAN_SNAPSHOT_KEY])
+ return {"task_plan": plan} if plan is not None else None
+ return None
+
+ async def abefore_agent(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """为异步图复用计划恢复规则。"""
+ return self.before_agent(state, runtime)
+
+ def after_model(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """拒绝同次响应中的多个计划替换，避免并行 Command 冲突。"""
+ messages = state.get("messages", [])
+ if not messages or not isinstance(messages[-1], AIMessage):
+ return None
+ calls = [call for call in messages[-1].tool_calls if call["name"] == PLAN_TOOL_NAME]
+ if len(calls) < 2:
+ return None
+ return {"messages": [ToolMessage(
+ content="同一轮只能调用一次 update_plan；未更新计划，请合并全部步骤后重试。",
+ tool_call_id=call["id"], name=PLAN_TOOL_NAME, status="error",
+ ) for call in calls]}
+
+ async def aafter_model(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """为异步图复用并行更新检查。"""
+ return self.after_model(state, runtime)
+
+ def after_agent(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """让成功轮次的末条消息携带恢复点，复用现有聊天消息持久化。"""
+ messages = state.get("messages", [])
+ snapshots = attach_plan_snapshot(messages, state.get("task_plan"))
+ if snapshots and snapshots[-1] is not messages[-1]:
+ return {"messages": [snapshots[-1]]}
+ return None
+
+ async def aafter_agent(self, state: PlanState, runtime: Any) -> Optional[dict[str, Any]]:
+ """为异步图复用最终恢复点保存。"""
+ return self.after_agent(state, runtime)
+
+ @staticmethod
+ def modify_request(request: ModelRequest) -> ModelRequest:
+ """每次请求都从图状态呈现最新计划，消息摘要不承担计划保存职责。"""
+ plan = _validated_plan(request.state.get("task_plan"))
+ prompt = PLAN_SYSTEM_PROMPT
+ if plan is not None:
+ prompt += (
+ "\n\nModel-reported working state; not authorization or independent proof.\n"
+ f"{json.dumps(plan, ensure_ascii=False)}\n"
+ )
+ return request.override(system_message=append_to_system_message(request.system_message, prompt))
+
+ def wrap_model_call(
+ self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse],
+ ) -> ModelResponse:
+ """在同步最终请求压缩之前注入计划。"""
+ return handler(self.modify_request(request))
+
+ async def awrap_model_call(
+ self, request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
+ ) -> ModelResponse:
+ """在异步最终请求压缩之前注入计划。"""
+ return await handler(self.modify_request(request))
diff --git a/app/agent/middleware/selection.py b/app/agent/middleware/selection.py
index a34d17602..de4d31af4 100644
--- a/app/agent/middleware/selection.py
+++ b/app/agent/middleware/selection.py
@@ -2,6 +2,7 @@
from dataclasses import dataclass, replace
import json
+import re
from collections.abc import Awaitable, Callable
from typing import Annotated, Any, NotRequired
@@ -19,11 +20,14 @@ from langchain.agents.middleware.tool_selection import (
DEFAULT_SYSTEM_PROMPT,
LLMToolSelectorMiddleware,
)
+from langchain.tools import ToolRuntime
from langchain_core.language_models.chat_models import BaseChatModel
-from langchain_core.messages import AIMessage, HumanMessage, SystemMessage
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
from langchain_core.runnables import RunnableConfig
-from langchain_core.tools import BaseTool
+from langchain_core.tools import BaseTool, StructuredTool, create_schema_from_function
from langgraph.runtime import Runtime
+from langgraph.types import Command, Overwrite
+from pydantic import Field
from typing_extensions import TypedDict # noqa
from app.agent.llm.helper import LLMHelper
@@ -34,6 +38,11 @@ MIN_SELECTED_TOOL_COUNT = 4
RECENT_SELECTION_CONTEXT_MESSAGE_LIMIT = 6
RECENT_SELECTION_CONTEXT_MAX_CHARS = 6000
RECENT_SELECTION_CONTEXT_TRUNCATION_PREFIX = "..."
+TOOL_DISCOVERY_NAME = "search_tools"
+TOOL_DISCOVERY_MAX_RESULTS = 8
+TOOL_DISCOVERY_DESCRIPTION_CHARS = 400
+TOOL_DISCOVERY_MAX_TAGS = 6
+TOOL_DISCOVERY_TAG_CHARS = 48
TOOL_GROUP_EXCLUDED_TAGS = frozenset(
{
ToolTag.AgentTool.value,
@@ -55,17 +64,25 @@ MoviePilot tool-chain hints:
"""
+def _merge_discovered_tool_names(current: list[str], added: list[str]) -> list[str]:
+ """合并并行工具发现的增量，避免各工具调用互相覆盖启用结果。"""
+ return list(dict.fromkeys([*current, *added]))
+
+
class ToolSelectionState(AgentState):
"""工具筛选中间件私有状态。"""
selected_tool_names: NotRequired[Annotated[list[str] | None, PrivateStateAttr]]
"""当前这条用户请求首轮筛选得到的工具名列表。"""
+ discovered_tool_names: Annotated[list[str], PrivateStateAttr, _merge_discovered_tool_names]
+ """图归并器初始化的工具增量；新用户请求开始时清空。"""
class ToolSelectionStateUpdate(TypedDict):
"""工具筛选中间件状态更新项。"""
selected_tool_names: list[str] | None
+ discovered_tool_names: NotRequired[Overwrite]
@dataclass(frozen=True)
@@ -92,7 +109,7 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
的每次“模型回合”前都重新筛选一次工具。对于会多轮调用工具的复杂任务，
这会重复消耗一次额外的 LLM 调用。这里改成：
- `abefore_agent()`：在本轮 Agent 执行开始时筛选一次；
- - `awrap_model_call()`：从 `request.state` 读取首轮筛选结果并复用。
+ - `awrap_model_call()`：复用首轮结果，合入本轮按需发现的额外工具。
"""
state_schema = ToolSelectionState
@@ -104,14 +121,116 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
selection_tools: list[Any] | None = None,
max_tools: int | None = None,
always_include: list[str] | None = None,
+ enable_discovery: bool = False,
) -> None:
+ """配置首轮筛选及可选的本轮工具发现，后者每次最多额外启用八个工具。"""
super().__init__(
model=model,
system_prompt=self._append_tool_selection_hint(system_prompt),
max_tools=max_tools,
always_include=always_include,
)
- self.selection_tools = selection_tools or []
+ self.selection_tools = list(selection_tools or [])
+ self.always_include: list[str] = list(always_include or [])
+ self.tools = [self._build_discovery_tool()] if enable_discovery else []
+ if self.tools:
+ self.selection_tools.extend(self.tools)
+ self.always_include = [*self.always_include, TOOL_DISCOVERY_NAME]
+
+ def _build_discovery_tool(self) -> StructuredTool:
+ """注册仅供当前 Agent 使用的发现工具，保持目录身份与模型参数可序列化。"""
+ tool = StructuredTool.from_function(
+ coroutine=self._asearch_tools,
+ name=TOOL_DISCOVERY_NAME,
+ description=(
+ "Find and enable tools missing from the current tool list. Search by exact tool name "
+ "or concise keywords from descriptions or capability tags. Matching tools become "
+ "available with their full input schemas on the next model call in this user turn. "
+ "Use when the task changes or a skill references an unavailable tool. "
+ "This searches only the current authorized catalog; no match enables nothing."
+ ),
+ args_schema=create_schema_from_function(
+ TOOL_DISCOVERY_NAME, self._asearch_tools, filter_args=["runtime"],
+ ),
+ tags=[ToolTag.AgentTool.value, ToolTag.Read.value],
+ )
+ object.__setattr__(tool, "_agent_tool_source", "middleware:selection")
+ return tool
+
+ @classmethod
+ def _discovery_score(cls, tool: BaseTool, search: str, keywords: list[str]) -> int:
+ """按工具名、能力标签和说明排序，优先保留精确名称命中的能力。"""
+ name = tool.name.casefold()
+ description = str(tool.description or "").casefold()
+ tags = " ".join(cls._normalize_tool_tags(tool)).casefold()
+ if search == name:
+ return 1000
+ return sum(
+ 10 * (keyword in name) + 5 * (keyword in tags) + (keyword in description)
+ for keyword in keywords
+ )
+
+ def _find_discovery_tools(self, search: str, limit: int) -> list[BaseTool]:
+ """只在当前授权目录内离线匹配关键词，空白或无关查询不会展开全量工具。"""
+ query = search.strip().casefold()
+ keywords = list(dict.fromkeys(re.findall(r"[^\W_]+", query)))
+ if not keywords:
+ return []
+ ranked_tools = [
+ (self._discovery_score(tool, query, keywords), tool)
+ for tool in self.selection_tools
+ if not isinstance(tool, dict) and tool.name != TOOL_DISCOVERY_NAME
+ ]
+ ranked_tools.sort(key=lambda item: (-item[0], item[1].name))
+ return [tool for score, tool in ranked_tools if score > 0][:limit]
+
+ async def _asearch_tools(
+ self,
+ search: Annotated[str, Field(
+ min_length=1, max_length=160,
+ description="Tool name or concise keywords matching descriptions or capability tags.",
+ )],
+ runtime: ToolRuntime,
+ limit: Annotated[int, Field(
+ ge=1, le=TOOL_DISCOVERY_MAX_RESULTS,
+ description="Maximum number of matching tools to enable for subsequent model calls.",
+ )] = 5,
+ ) -> Command:
+ """返回有界工具目录并以图状态增量启用结果，不修改共享中间件实例。"""
+ matched_tools = self._find_discovery_tools(search, limit)
+ catalog = [
+ {
+ "name": tool.name,
+ "description": str(tool.description or "")[:TOOL_DISCOVERY_DESCRIPTION_CHARS],
+ "tags": [
+ tag[:TOOL_DISCOVERY_TAG_CHARS]
+ for tag in self._normalize_tool_tags(tool)[:TOOL_DISCOVERY_MAX_TAGS]
+ ],
+ }
+ for tool in matched_tools
+ ]
+ payload = {
+ "tools": catalog,
+ "message": (
+ "Matching tools are enabled for the next model call in this user turn."
+ if matched_tools else "No matching tools. Try a precise tool name or different keywords."
+ ),
+ }
+ return Command(update={
+ "discovered_tool_names": [tool.name for tool in matched_tools],
+ "messages": [ToolMessage(
+ content=json.dumps(payload, ensure_ascii=False),
+ tool_call_id=runtime.tool_call_id,
+ name=TOOL_DISCOVERY_NAME,
+ )],
+ })
+
+ def _selection_state_update(self, selected_tool_names: list[str] | None) -> ToolSelectionStateUpdate:
+ """写入首轮筛选结果并清空前一用户请求的工具发现状态。"""
+ update = ToolSelectionStateUpdate(selected_tool_names=selected_tool_names)
+ if self.tools:
+ update["discovered_tool_names"] = Overwrite([])
+ return update
@classmethod
def _render_recent_conversation_context(
@@ -519,7 +638,7 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
"""
将已筛选出的工具集应用到当前模型请求。
- 这里只复用首次筛选出的客户端工具名；provider-specific 的 dict 工具仍然
+ 这里复用首轮筛选及本轮发现的客户端工具名；provider-specific 的 dict 工具仍然
原样保留，避免破坏 LangChain/provider 自身的工具绑定约定。
"""
current_tools_by_name = {
@@ -608,7 +727,7 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
detail=detail,
)
)
- return ToolSelectionStateUpdate(selected_tool_names=None)
+ return self._selection_state_update(selected_tool_names=None)
selection_request = ModelRequest(
model=self.model,
@@ -620,7 +739,7 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
attempt = await self._aselect_request_once_with_status(selection_request)
self._log_selection_attempt(attempt)
selected_tool_names = attempt.selected_tool_names
- return ToolSelectionStateUpdate(selected_tool_names=selected_tool_names)
+ return self._selection_state_update(selected_tool_names=selected_tool_names)
async def awrap_model_call(
self,
@@ -630,7 +749,7 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
],
) -> ModelResponse[ResponseT]:
"""
- 从 state 中读取首次筛选结果，并应用到每次模型回合。
+ 复用首轮筛选并合入本轮按需发现结果，无需追加工具筛选模型调用。
"""
selected_tool_names = request.state.get("selected_tool_names") # noqa
@@ -648,6 +767,11 @@ class ToolSelectorMiddleware(LLMToolSelectorMiddleware):
request.state["selected_tool_names"] = selected_tool_names # noqa
if selected_tool_names is not None:
+ selected_tool_names = list(dict.fromkeys([
+ *selected_tool_names,
+ *request.state.get("discovered_tool_names", []),
+ *self.always_include,
+ ]))
request = self._apply_selected_tools(request, selected_tool_names)
return await handler(request)
diff --git a/app/agent/middleware/subagents.py b/app/agent/middleware/subagents.py
index 1c65e94a7..a238e7320 100644
--- a/app/agent/middleware/subagents.py
+++ b/app/agent/middleware/subagents.py
@@ -25,6 +25,10 @@ from pydantic import BaseModel, Field
from app.agent.llm.helper import LLMHelper
from app.agent.middleware.policy import AgentPolicyMiddleware
+from app.agent.middleware.summarization import (
+ ContextPreservingSummarizationMiddleware,
+ FinalRequestCompactionMiddleware,
+)
from app.agent.middleware.utils import append_to_system_message
from app.agent.policy.contracts import (
AuthSource,
@@ -469,7 +473,13 @@ class _SubAgentAgentProvider:
AgentPolicyMiddleware(
context=self._policy_context,
catalog=subagent_catalog,
- )
+ ),
+ FinalRequestCompactionMiddleware(
+ summarizer=ContextPreservingSummarizationMiddleware(
+ model=self._model,
+ keep=("messages", 20),
+ ),
+ ),
],
)
self._agents[profile.name] = agent
@@ -946,6 +956,10 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
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 = [
@@ -960,6 +974,8 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
cancel_requested=self._close_cancel_requested,
)
if pending_records:
+ # 未收敛的子任务仍由该 owner 持有，不能让下一轮叠加新任务。
+ self.seal()
self._tasks = {
record.task_id: record
for record in pending_records
@@ -1232,8 +1248,8 @@ class SubAgentTaskControlMiddleware(AgentMiddleware):
return self._json_response(response)
async def aafter_agent(self, state: Any, runtime: Any) -> None:
- """Agent 结束时取消未完成的子代理任务，避免后台泄漏。"""
- await self.close()
+ """本轮结束时回收子任务；永久关闭只由会话生命周期触发。"""
+ await self._drain_tasks()
async def awrap_tool_call(
self,
diff --git a/app/agent/orchestrator.py b/app/agent/orchestrator.py
index 6016ad003..3b9abea64 100644
--- a/app/agent/orchestrator.py
+++ b/app/agent/orchestrator.py
@@ -13,9 +13,11 @@ from typing import Any, Callable, Dict, List, Optional
from langchain.agents import create_agent
from langchain_core.messages import ( # noqa: F401
+ AIMessage,
BaseMessage,
HumanMessage,
SystemMessage,
+ ToolMessage,
)
from langgraph.checkpoint.memory import InMemorySaver
@@ -25,22 +27,18 @@ from app.agent.llm.helper import LLMHelper
from app.agent.llm.tools import ServerToolRegistry
from app.agent.mcp import agent_mcp_manager
from app.agent.memory import MemoryManager, memory_manager
-from app.agent.middleware.activity import (
- QUERY_ACTIVITY_LOG_TOOL_NAME,
- ActivityLogMiddleware,
-)
+from app.agent.middleware.activity import ActivityLogMiddleware
from app.agent.middleware.config import RuntimeConfigMiddleware
from app.agent.middleware.jobs import (
JobsMiddleware,
)
from app.agent.middleware.memory import MemoryMiddleware
from app.agent.middleware.patching import PatchToolCallsMiddleware
+from app.agent.middleware.plan import PLAN_SNAPSHOT_KEY, PlanMiddleware, attach_plan_snapshot
from app.agent.middleware.policy import AgentPolicyMiddleware
from app.agent.middleware.selection import ToolSelectorMiddleware
-from app.agent.middleware.skills import SKILL_TOOL_NAME, SkillsMiddleware
+from app.agent.middleware.skills import SkillsMiddleware
from app.agent.middleware.subagents import (
- SUBAGENT_CONTROL_TOOL_NAME,
- SUBAGENT_TASK_TOOL_NAME,
create_subagent_middlewares,
is_subagent_stream_metadata,
)
@@ -58,6 +56,7 @@ from app.agent.policy.contracts import (
ToolPolicyContext,
)
from app.agent.policy.registry import requests_system_setting_secrets
+from app.agent.policy.sanitizer import sanitize_for_host
from app.agent.prompt import prompt_manager
from app.agent.runtime import agent_runtime_manager
from app.agent.tools.catalog import ToolCatalogSnapshot
@@ -89,6 +88,8 @@ def _get_plugin_tools_revision() -> int:
warnings.filterwarnings("ignore", message=".*allowed_objects.*")
+AGENT_RECOVERY_SNAPSHOT_TIMEOUT = 2.0
+
_KNOWN_AGENT_PROVIDER_TYPES = (
"anthropic",
@@ -1777,6 +1778,26 @@ class MoviePilotAgent:
specs=specs,
)
+ @staticmethod
+ def _initialize_tool_selector(
+ tools: list[Any], internal_tools: list[Any], model: Any,
+ ) -> Optional[ToolSelectorMiddleware]:
+ """首轮限制业务工具数量，同时保留计划、技能和按需发现入口。"""
+ max_tools = get_runtime_setting("LLM_MAX_TOOLS")
+ if max_tools <= 0:
+ return None
+ from app.agent.loader import get_tool_factory
+
+ always_include = get_tool_factory().get_tool_selector_always_include_names(tools)
+ always_include.extend(tool.name for tool in internal_tools)
+ return ToolSelectorMiddleware(
+ model=model,
+ selection_tools=[*tools, *internal_tools],
+ max_tools=max_tools,
+ always_include=always_include,
+ enable_discovery=True,
+ )
+
async def _create_agent(self, streaming: bool = False):
"""
创建 LangGraph Agent（使用 create_agent + SummarizationMiddleware）
@@ -1871,14 +1892,18 @@ class MoviePilotAgent:
catalog=subagent_catalog,
)
temporary_subagent_middlewares = tuple(subagent_middlewares)
+ plan_middleware = PlanMiddleware()
+ internal_tools = [
+ *skill_tools, *activity_log_tools, *subagent_task_tools, *plan_middleware.tools,
+ ]
+ tool_selector = self._initialize_tool_selector(tools, internal_tools, non_streaming_model)
# 严格目录必须覆盖 LangGraph ToolNode 可执行的全部 client-side 工具。
tool_catalog = ToolCatalogSnapshot.from_tools(
[
*local_tools,
*mcp_tools,
- *skill_tools,
- *activity_log_tools,
- *subagent_task_tools,
+ *internal_tools,
+ *(tool_selector.tools if tool_selector else []),
],
plugin_revision=base_tool_catalog.plugin_revision,
factory_revision=base_tool_catalog.factory_revision,
@@ -1901,27 +1926,6 @@ class MoviePilotAgent:
temporary_subagent_middlewares = ()
logger.debug(f"复用会话内 Agent 图: session_id={self.session_id}")
return cached_agent
- max_tools = get_runtime_setting("LLM_MAX_TOOLS")
- from app.agent.loader import get_tool_factory
-
- always_include_tools = get_tool_factory().get_tool_selector_always_include_names(tools)
- if subagent_task_tools:
- always_include_tools.extend(
- tool.name
- for tool in subagent_task_tools
- if getattr(tool, "name", None) in {SUBAGENT_TASK_TOOL_NAME, SUBAGENT_CONTROL_TOOL_NAME}
- )
- if skill_tools:
- always_include_tools.extend(
- tool.name for tool in skill_tools if getattr(tool, "name", None) == SKILL_TOOL_NAME
- )
- if activity_log_tools:
- always_include_tools.extend(
- tool.name
- for tool in activity_log_tools
- if getattr(tool, "name", None) == QUERY_ACTIVITY_LOG_TOOL_NAME
- )
-
summarization_middleware = SummarizationMiddleware(
model=non_streaming_model,
trigger=("fraction", 0.85),
@@ -1944,6 +1948,8 @@ class MoviePilotAgent:
),
# 运行时人格与核心规则
RuntimeConfigMiddleware(),
+ # 计划独立保存，最终请求压缩仍计入其系统上下文预算。
+ plan_middleware,
# 记忆管理
MemoryMiddleware(memory_dir=str(agent_runtime_manager.memory_dir)),
# 活动日志依赖记忆上下文，并应在最终请求压缩前完成读取与记录。
@@ -1955,20 +1961,8 @@ class MoviePilotAgent:
]
# 工具选择
- if max_tools > 0:
- middlewares.append(
- ToolSelectorMiddleware(
- model=non_streaming_model,
- selection_tools=[
- *tools,
- *skill_tools,
- *activity_log_tools,
- *subagent_task_tools,
- ],
- max_tools=max_tools,
- always_include=always_include_tools,
- )
- )
+ if tool_selector is not None:
+ middlewares.append(tool_selector)
# 所有压缩都在最终请求边界完成，避免主模型失败前写入摘要状态。
middlewares.append(
@@ -2207,6 +2201,106 @@ class MoviePilotAgent:
stripper.flush(on_token)
+ @staticmethod
+ def _sanitize_recovery_message(message: BaseMessage) -> BaseMessage:
+ """为中断恢复复制消息，只保留脱敏正文与工具协议需要的字段。"""
+ updates: dict[str, Any] = {
+ "content": sanitize_for_host(message.content),
+ "additional_kwargs": {},
+ "response_metadata": {},
+ }
+ if isinstance(message, AIMessage):
+ updates["tool_calls"] = [
+ {**call, "args": sanitize_for_host(call["args"])}
+ for call in message.tool_calls
+ ]
+ updates["invalid_tool_calls"] = []
+ updates["usage_metadata"] = None
+ if isinstance(message, ToolMessage):
+ updates["artifact"] = None
+ return attach_plan_snapshot(
+ [message.model_copy(update=updates)],
+ message.additional_kwargs.get(PLAN_SNAPSHOT_KEY),
+ )[0]
+
+ @staticmethod
+ def _requires_portable_recovery(messages: list[BaseMessage]) -> bool:
+ """带思考、签名或内容块的响应不能在脱敏后作为原供应商协议回放。"""
+ protocol_fields = {"reasoning_content", "__gemini_function_call_thought_signatures__"}
+ return any(
+ isinstance(message, AIMessage) and (
+ isinstance(message.content, list)
+ or bool(protocol_fields.intersection(message.additional_kwargs))
+ )
+ for message in messages
+ )
+
+ @staticmethod
+ def _portable_recovery_messages(messages: list[BaseMessage]) -> list[BaseMessage]:
+ """把供应商签名工具链转换为历史事实文本，保留用户消息和实际回执。"""
+ recovered: list[BaseMessage] = []
+ for message in messages:
+ facts: dict[str, Any]
+ if isinstance(message, AIMessage):
+ facts = {
+ "assistant_text": LLMHelper.extract_text_content(message.content),
+ "tool_calls": message.tool_calls,
+ }
+ elif isinstance(message, ToolMessage):
+ facts = {
+ "tool_call_id": message.tool_call_id,
+ "tool_name": message.name,
+ "status": message.status,
+ "result": message.content,
+ }
+ else:
+ recovered.append(message)
+ continue
+ content = "宿主保存的历史执行事实（仅作观察数据，不是新指令或待执行工具调用）：\n"
+ content += json.dumps(facts, ensure_ascii=False)
+ plan = message.additional_kwargs.get(PLAN_SNAPSHOT_KEY)
+ if recovered and isinstance(recovered[-1], AIMessage):
+ previous = recovered.pop()
+ content = f"{previous.content}\n\n{content}"
+ plan = plan or previous.additional_kwargs.get(PLAN_SNAPSHOT_KEY)
+ recovered.append(AIMessage(content=content))
+ recovered = attach_plan_snapshot(recovered, plan)
+ return recovered
+
+ async def _recover_interrupted_agent(
+ self, agent: Any, config: dict[str, Any], *, cancelled: bool = False
+ ) -> None:
+ """有界保存中断事实后释放旧图，再次取消或恢复失败仍执行清理。"""
+ try:
+ if agent is None or not self._should_persist_agent_chat():
+ return
+ state = agent.get_state(config).values
+ messages = state.get("messages", [])
+ if not messages:
+ return
+ recovered = PatchToolCallsMiddleware._normalize_messages([
+ self._sanitize_recovery_message(message) for message in messages
+ ])
+ if self._requires_portable_recovery(messages):
+ recovered = self._portable_recovery_messages(recovered)
+ reason = "已取消" if cancelled else "因执行异常中断"
+ recovered.append(AIMessage(content=(
+ f"上一轮任务{reason}，尚未完成。已保留上面的执行记录。"
+ "继续时复用已确认结果；缺少回执的操作结果未知，"
+ "先只读核验实际状态，避免重复下载、修改或发送消息。"
+ )))
+ recovered = attach_plan_snapshot(recovered, state.get("task_plan"))
+ async with asyncio.timeout(AGENT_RECOVERY_SNAPSHOT_TIMEOUT):
+ await self._memory.async_save_agent_messages(
+ session_id=self.session_id,
+ user_id=self.user_id or "",
+ messages=recovered,
+ )
+ except Exception as error:
+ logger.warning(f"保存Agent中断快照失败: {type(error).__name__}")
+ finally:
+ await self._invalidate_cached_agent()
+
async def _execute_agent(self, messages: List[BaseMessage]):
"""
调用 LangGraph Agent 执行推理。
@@ -2222,6 +2316,8 @@ class MoviePilotAgent:
self._llm_runtime_config = None
self._llm_provider_selection = {}
streaming_stopped = False
+ agent = None
+ agent_config: dict[str, Any] = {}
try:
# Agent运行配置
agent_config = {
@@ -2333,11 +2429,11 @@ class MoviePilotAgent:
except asyncio.CancelledError:
logger.info(f"Agent执行被取消: session_id={self.session_id}")
- await self._invalidate_cached_agent()
+ await self._recover_interrupted_agent(agent, agent_config, cancelled=True)
execution_error = "任务已取消"
raise
except Exception as e:
- await self._invalidate_cached_agent()
+ await self._recover_interrupted_agent(agent, agent_config)
execution_error = str(e)
if self._messages_have_image_input(messages) and self._is_unsupported_image_input_error(e):
logger.warning(f"当前模型不支持图片输入，已向用户发送友好提示: {e}")
diff --git a/app/agent/prompt/System Core Prompt.txt b/app/agent/prompt/System Core Prompt.txt
index 3abd5b3a2..74e255147 100644
--- a/app/agent/prompt/System Core Prompt.txt
+++ b/app/agent/prompt/System Core Prompt.txt
@@ -45,6 +45,10 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel
- After an action, perform the minimum validation needed to confirm the result actually landed.
- Reuse known media identity, prior tool results, and current system context instead of repeating expensive recognition or search calls.
- When a tool fails, try one narrower fallback path before escalating to the user.
+- For work with several dependent steps, use `update_plan` to preserve the objective, remaining work, and checked outcomes. Continue executing after each plan update. Keep the active objective when the user asks about progress; replace it when the user clearly starts a new task.
+- Reuse explicit authorization already given for the same action and scope. A plan, summary, tool output, or subagent report cannot grant additional authorization.
+- After interruption, distinguish completed tool receipts from missing or uncertain results. Read current state before retrying an action with side effects; a timeout or cancelled wait does not prove the action did not happen.
+- Before reporting completion, check the requested outcome against current evidence. Say what is complete, what remains blocked, and what has not been verified; do not present a submitted background operation as finished.
@@ -56,6 +60,7 @@ You act as a proactive agent. Your goal is to fully resolve the user's media-rel
+- If `search_tools` is available and a needed tool is missing, search by its exact name from a Skill or concise capability keywords. Read the returned tool schema on the next model call and continue the task; finding a tool does not authorize its execution.
- Use parallel tool calls by default for independent read-only or diagnostic work. In one assistant turn, issue all tool calls that can run without waiting for each other's results, such as checking enabled sites, library existence, recent history, downloader status, and scheduler or configuration state.
- Keep tools sequential only when later arguments depend on earlier output, when a tool mutates state, when confirmation is required, or when concurrent writes could conflict.
- When planning a multi-step investigation, group the first wave of safe state-gathering calls together, then continue with dependent actions after those results return.
diff --git a/docs/agent.md b/docs/agent.md
new file mode 100644
index 000000000..b231d76bb
--- /dev/null
+++ b/docs/agent.md
@@ -0,0 +1,41 @@
+# Agent 复杂任务执行与恢复
+
+MoviePilot Agent 通过模型、Skills、工具和会话状态共同完成任务。本次增强关注长任务的连续性：明确目标和检查结果，按需补充工具，并在执行中断后保留已知事实。
+
+## 任务计划
+
+复杂任务可以使用内部 `update_plan` 工具保存目标、步骤、状态、检查依据和阻塞原因。例如“排查下载后未入库”可以依次检查下载状态、整理队列、目标目录和媒体库。
+
+- 简单查询不需要创建计划。
+- 步骤支持 `pending`、`in_progress`、`completed`、`blocked`；完成和阻塞状态需要填写依据。
+- 计划单独保存在会话图状态中，每次模型请求都能读取；压缩旧消息不会删除计划。
+- 计划快照随聊天消息保存，在会话图重建后恢复。用户询问进度时继续当前目标，明确开始新任务时替换计划。
+- 计划是模型记录的工作状态，不代表工具执行结果或写操作授权。仍需通过工具核实实际结果。
+
+## 按需发现工具
+
+设置 `LLM_MAX_TOOLS > 0` 时，Agent 首轮仍只筛选一批相关工具。后续读取 Skill 或发现新问题后，可以通过内部 `search_tools(search, limit)` 按名称、说明或标签搜索当前会话工具目录，并在下一次模型调用获得匹配工具的完整参数定义。
+
+每次最多返回并启用 8 个匹配工具；没有匹配结果时不会展开全部工具。发现状态仅作用于当前用户请求，新的用户请求重新筛选。`LLM_MAX_TOOLS` 约束首轮筛选数量，显式按需发现可以增加工具。发现不会安装工具、连接新的 MCP 服务或扩大执行权限。
+
+`update_plan` 和 `search_tools` 是 Agent 内部会话能力，不通过外部 MCP 或 `moviepilot tool` 发布。
+
+## 异常和取消后的继续
+
+对已启用聊天持久化的会话，执行异常或用户取消时，Agent 在释放会话图之前尝试保存脱敏恢复快照。已收到的工具回执保留；未收到结果的调用明确标记为“结果未知”，不能当成未执行或已成功。
+
+用户随后说“继续”时，Agent 可以读取已有结果和计划，从剩余工作继续。下载、修改配置、删除文件等操作如缺少回执，应先只读核验实际状态，再决定是否重试。取消仍会终止当前轮次，不会自动重放工具。
+
+快照保存采用有界等待。进程强制退出、持久化失败，或工具在中断后才完成的外部副作用，不能保证被该快照记录。它不是外部操作的事务日志，也不保证恰好执行一次。
+
+## 子代理连续工作
+
+一轮结束会回收未完成的子任务，正常收敛后同一会话下一轮仍可委派。会话关闭时永久停止接收子任务；无法在等待上限内收敛的任务仍由原 owner 持有并封闭新提交。
+
+子代理沿用主 Agent 的最终请求预算和上下文压缩机制。长调查可以压缩旧消息；最新单条工具事务或固定提示本身超过模型窗口时仍会明确失败，不会静默删除关键内容。
+
+## 验证范围
+
+自动化测试使用脚本模型和真实 LangGraph 执行图，覆盖工具发现后的实际调用、并行发现合并、计划更新和消息序列化恢复、执行中断以及连续两轮子代理委派。测试不调用付费模型或外部服务。
+
+这些机制提高执行可靠性；真实任务的理解、规划质量和成功率仍取决于所选模型、上下文窗口、工具权限及外部服务状态。模型能力需要用真实业务场景单独评估。
diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md
index aaefd3229..395dc82a8 100644
--- a/docs/architecture-overview.md
+++ b/docs/architecture-overview.md
@@ -583,6 +583,8 @@ flowchart LR
## 七、AI Agent 子系统
+复杂任务的计划、工具发现和中断恢复使用说明见 [Agent 复杂任务执行与恢复](agent.md)。
+
Agent 采用**门面 + 惰性物化**设计，避免 `application → agent` 形成静态依赖边：
```mermaid
@@ -754,8 +756,8 @@ flowchart LR
| 指标 | 当前值 |
|---|---:|
-| Python 模块 | 986 |
-| 内部导入边 | 8,385 |
+| Python 模块 | 987 |
+| 内部导入边 | 8,395 |
| 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） |
| Application / Chain 具体 Adapter 直连 | 0 / 0 |
| Direct egress | 53（债务已清零，53 条精确 containment） |
diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md
index 0152c75bb..bdfb262b4 100644
--- a/docs/architecture/optimization-checklist.md
+++ b/docs/architecture/optimization-checklist.md
@@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
| 指标 | 当前值 | 解释 |
|---|---:|---|
-| 宿主 Python 模块 / 内部依赖边 | 986 / 8,385 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索与整理任务恢复模块的受控依赖 |
+| 宿主 Python 模块 / 内部依赖边 | 987 / 8,395 | `dependency-baseline.json` 当前快照；分类、下载资源归类、订阅搜索、整理恢复与 Agent 计划模块的受控依赖 |
| 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 |
| 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 |
| Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 |
diff --git a/docs/mcp-api.md b/docs/mcp-api.md
index d912e5970..da341e061 100644
--- a/docs/mcp-api.md
+++ b/docs/mcp-api.md
@@ -35,6 +35,8 @@ MCP 使用系统配置中的 `API_TOKEN` 作为认证密钥，文档中的 API K
### 动态插件工具
+内置 Agent 的 `update_plan`、`search_tools` 为会话中间件工具，不通过外部 `tools/list` 发布；它们只维护计划或发现当前目录内的工具，不能授予业务操作权限。使用与恢复语义见 [Agent 复杂任务执行与恢复](agent.md)。
+
`tools/list` 会同时返回 MoviePilot 内置工具和已启用插件通过 `get_agent_tools()` 声明的工具。插件启动、停止、重载或配置生效后，MCP 工具管理器会在下一次列出或调用工具时按注册表版本惰性刷新，避免继续暴露已移除的工具或遗漏新工具。
MCP 当前不会主动发送工具列表变更通知（`listChanged=false`）。如果客户端缓存了工具列表，插件状态变化后需要让客户端重新请求 `tools/list`；无法手动刷新的客户端应重新连接 MCP 服务或新建会话。
diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md
index 08536f381..5fa6f242f 100644
--- a/docs/rules/05-architecture.md
+++ b/docs/rules/05-architecture.md
@@ -86,6 +86,8 @@ to make the directory tree look symmetrical.
| `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/middleware/plan.py` | Current task objective, step status and evidence in graph state; sanitized snapshots travel through existing message persistence and never authorize tool effects |
+| `app/agent/middleware/selection.py` | First-turn tool selection and bounded, on-demand discovery within the same authorized catalog; discovered tool names remain local to the current user request |
| `app/agent/shell.py` | Agent command-shell selection and subprocess text-encoding policy; Windows prefers Git Bash, then PowerShell 7, while POSIX keeps native shell/PTY behavior |
| `app/agent/policy/api.py` | Fixed `moviepilot_api` operation registry, HTTP route templates and per-operation authorization/effect policy; no arbitrary URL or method input |
| `app/agent/policy/mcp.py` | Generated external MCP input-contract builder for the fixed API registry; owns exact English oneOf parameter projection, not runtime authorization |
diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json
index 65b7d2c83..238d9f31a 100644
--- a/tests/fixtures/architecture/dependency-baseline.json
+++ b/tests/fixtures/architecture/dependency-baseline.json
@@ -1074,8 +1074,8 @@
"runtime_only": true
}
},
- "edge_count": 8385,
- "edge_sha256": "ce9171f07d0649b1e54a725d5a1314852f076b73512d94c3661ea834373b51d9",
+ "edge_count": 8395,
+ "edge_sha256": "0af2b2a8f5616e192c9e8db19f70b942513a34cf988170c1a43e13ce7c59b066",
"edges": [
"app -> app.foundation",
"app -> app.foundation.environment",
@@ -1457,6 +1457,13 @@
"app.agent.middleware.memory -> app.agent.middleware.utils",
"app.agent.middleware.memory -> app.runtime",
"app.agent.middleware.memory -> app.runtime.log",
+ "app.agent.middleware.plan -> app.agent",
+ "app.agent.middleware.plan -> app.agent.middleware",
+ "app.agent.middleware.plan -> app.agent.middleware.utils",
+ "app.agent.middleware.plan -> app.agent.policy",
+ "app.agent.middleware.plan -> app.agent.policy.sanitizer",
+ "app.agent.middleware.plan -> app.agent.tools",
+ "app.agent.middleware.plan -> app.agent.tools.tags",
"app.agent.middleware.policy -> app.agent",
"app.agent.middleware.policy -> app.agent.policy",
"app.agent.middleware.policy -> app.agent.policy.contracts",
@@ -1489,6 +1496,7 @@
"app.agent.middleware.subagents -> app.agent.llm.helper",
"app.agent.middleware.subagents -> app.agent.middleware",
"app.agent.middleware.subagents -> app.agent.middleware.policy",
+ "app.agent.middleware.subagents -> app.agent.middleware.summarization",
"app.agent.middleware.subagents -> app.agent.middleware.utils",
"app.agent.middleware.subagents -> app.agent.policy",
"app.agent.middleware.subagents -> app.agent.policy.contracts",
@@ -1521,6 +1529,7 @@
"app.agent.orchestrator -> app.agent.middleware.jobs",
"app.agent.orchestrator -> app.agent.middleware.memory",
"app.agent.orchestrator -> app.agent.middleware.patching",
+ "app.agent.orchestrator -> app.agent.middleware.plan",
"app.agent.orchestrator -> app.agent.middleware.policy",
"app.agent.orchestrator -> app.agent.middleware.selection",
"app.agent.orchestrator -> app.agent.middleware.skills",
@@ -1530,6 +1539,7 @@
"app.agent.orchestrator -> app.agent.policy",
"app.agent.orchestrator -> app.agent.policy.contracts",
"app.agent.orchestrator -> app.agent.policy.registry",
+ "app.agent.orchestrator -> app.agent.policy.sanitizer",
"app.agent.orchestrator -> app.agent.prompt",
"app.agent.orchestrator -> app.agent.runtime",
"app.agent.orchestrator -> app.agent.tools",
@@ -9463,7 +9473,7 @@
"app.workflow.actions.transfer_file -> app.workflow",
"app.workflow.actions.transfer_file -> app.workflow.actions"
],
- "module_count": 986,
+ "module_count": 987,
"modules": [
"app",
"app.adapters",
@@ -9544,6 +9554,7 @@
"app.agent.middleware.jobs",
"app.agent.middleware.memory",
"app.agent.middleware.patching",
+ "app.agent.middleware.plan",
"app.agent.middleware.policy",
"app.agent.middleware.selection",
"app.agent.middleware.skills",
diff --git a/tests/test_agent_background_output.py b/tests/test_agent_background_output.py
index fa8b7c093..70fa5bb95 100644
--- a/tests/test_agent_background_output.py
+++ b/tests/test_agent_background_output.py
@@ -8,6 +8,8 @@ from app.agent.contracts import ReplyMode
from app.agent.manager import AgentManager
from app.agent.memory import memory_manager
from app.agent.middleware.activity import QUERY_ACTIVITY_LOG_TOOL_NAME
+from app.agent.middleware.plan import PLAN_TOOL_NAME, PlanMiddleware
+from app.agent.middleware.selection import TOOL_DISCOVERY_NAME, ToolSelectorMiddleware
from app.agent.middleware.skills import SKILL_TOOL_NAME
from app.agent.middleware.subagents import (
SUBAGENT_CONTROL_TOOL_NAME,
@@ -80,6 +82,38 @@ def _fake_activity_log_middleware(tool=None):
return SimpleNamespace(name="activity", tools=[] if tool is None else [tool])
+def _capture_tool_selector(captured, **kwargs):
+ """保留真实发现工具与筛选合同，只替换无需执行的测试模型实例。"""
+ selector_options = {**kwargs, "model": None}
+ selector = ToolSelectorMiddleware(**selector_options)
+ selector.model = kwargs["model"]
+ captured.update(
+ selector=selector,
+ selection_tools=selector.selection_tools,
+ always_include=selector.always_include,
+ enable_discovery=kwargs["enable_discovery"],
+ )
+ return selector
+
+
+def _assert_internal_tool_registration(created, captured):
+ """核实计划与发现工具同时注册到严格目录和强制保留集合。"""
+ middlewares = created["middleware"]
+ plan = next(item for item in middlewares if isinstance(item, PlanMiddleware))
+ selector = captured["selector"]
+ policy = middlewares[0]
+ assert policy.name == "AgentPolicyMiddleware"
+ assert captured["enable_discovery"] is True
+ assert {PLAN_TOOL_NAME, TOOL_DISCOVERY_NAME} <= set(selector.always_include)
+ for tool in [*plan.tools, *selector.tools]:
+ assert policy.catalog.resolve_unique(tool.name).tool is tool
+ assert tool in selector.selection_tools
+ assert middlewares.index(plan) < middlewares.index(selector)
+ assert middlewares.index(selector) == len(middlewares) - 3
+ assert middlewares[-2].name == "FinalRequestCompactionMiddleware"
+ assert middlewares[-1] == "usage"
+
+
class TestAgentBackgroundOutput:
"""验证后台任务的回复策略、流式结束行为和工具装配。"""
@@ -381,6 +415,7 @@ class TestAgentBackgroundOutput:
)
async def test_create_agent_excludes_activity_log_for_heartbeat_session(self):
+ """心跳任务保留计划上下文，但不注入渠道活动日志。"""
agent = MoviePilotAgent(
session_id=f"{HEARTBEAT_SESSION_PREFIX}test__",
user_id="system",
@@ -421,6 +456,7 @@ class TestAgentBackgroundOutput:
"skills",
"jobs",
"runtime",
+ "PlanMiddleware",
"memory",
"patch",
"FinalRequestCompactionMiddleware",
@@ -436,9 +472,8 @@ class TestAgentBackgroundOutput:
agent._initialize_subagent_tools = lambda: []
def _tool_selector(**kwargs):
- captured["selection_tools"] = kwargs["selection_tools"]
- captured["always_include"] = kwargs["always_include"]
- return "selector"
+ """记录技能能力进入真实筛选目录后的参数。"""
+ return _capture_tool_selector(captured, **kwargs)
with (
patch.object(settings, "LLM_MAX_TOOLS", 5),
@@ -481,6 +516,7 @@ class TestAgentBackgroundOutput:
assert skill_tool in created["tools"]
assert skill_tool in captured["selection_tools"]
assert SKILL_TOOL_NAME in captured["always_include"]
+ _assert_internal_tool_registration(created, captured)
async def test_create_agent_excludes_activity_log_without_message_context(self):
"""无渠道信息的后台捕获任务不应注入活动日志。"""
@@ -534,6 +570,7 @@ class TestAgentBackgroundOutput:
"skills",
"jobs",
"runtime",
+ "PlanMiddleware",
"memory",
"patch",
"FinalRequestCompactionMiddleware",
@@ -574,9 +611,8 @@ class TestAgentBackgroundOutput:
agent._initialize_subagent_tools = lambda: []
def _tool_selector(**kwargs):
- captured["selection_tools"] = kwargs["selection_tools"]
- captured["always_include"] = kwargs["always_include"]
- return "selector"
+ """记录活动日志能力进入真实筛选目录后的参数。"""
+ return _capture_tool_selector(captured, **kwargs)
with (
patch.object(settings, "LLM_MAX_TOOLS", 5),
@@ -621,6 +657,7 @@ class TestAgentBackgroundOutput:
assert activity_tool in created["tools"]
assert activity_tool in captured["selection_tools"]
assert QUERY_ACTIVITY_LOG_TOOL_NAME in captured["always_include"]
+ _assert_internal_tool_registration(created, captured)
async def test_create_agent_always_includes_subagent_tools(self):
"""工具筛选开启时应保留同步和异步子代理入口。"""
@@ -630,8 +667,8 @@ class TestAgentBackgroundOutput:
agent._initialize_subagent_tools = lambda: []
def _tool_selector(**kwargs):
- captured["always_include"] = kwargs["always_include"]
- return "selector"
+ """记录子代理能力进入真实筛选目录后的参数。"""
+ return _capture_tool_selector(captured, **kwargs)
with (
patch.object(settings, "LLM_MAX_TOOLS", 5),
@@ -678,12 +715,14 @@ class TestAgentBackgroundOutput:
patch("app.agent.orchestrator.InMemorySaver", return_value="checkpointer"),
patch("app.agent.orchestrator.create_agent", side_effect=lambda **kwargs: kwargs),
):
- await agent._create_agent(streaming=False)
+ created = await agent._create_agent(streaming=False)
assert SUBAGENT_TASK_TOOL_NAME in captured["always_include"]
assert SUBAGENT_CONTROL_TOOL_NAME in captured["always_include"]
+ _assert_internal_tool_registration(created, captured)
async def test_create_agent_keeps_activity_log_for_normal_session(self):
+ """普通渠道会话在计划和记忆上下文后保留活动日志。"""
agent = MoviePilotAgent(
session_id="normal-session",
user_id="system",
@@ -735,6 +774,7 @@ class TestAgentBackgroundOutput:
"skills",
"jobs",
"runtime",
+ "PlanMiddleware",
"memory",
"activity",
"patch",
diff --git a/tests/test_agent_patch_tool_calls.py b/tests/test_agent_patch_tool_calls.py
index 6da6bc562..e66ad15c7 100644
--- a/tests/test_agent_patch_tool_calls.py
+++ b/tests/test_agent_patch_tool_calls.py
@@ -1,6 +1,6 @@
-import asyncio
-import unittest
+"""工具调用历史修复保留真实回执，并明确缺失结果的未知状态。"""
+import pytest
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from app.agent.middleware.patching import PatchToolCallsMiddleware
@@ -16,70 +16,74 @@ def _build_tool_call(tool_call_id: str = "call_1", name: str = "search") -> dict
}
-class TestPatchToolCallsMiddleware(unittest.TestCase):
- """测试工具调用历史修复中间件。"""
+def test_adds_missing_tool_messages_immediately_after_ai_message():
+ """缺失工具响应时应立即补齐 ToolMessage。"""
+ middleware = PatchToolCallsMiddleware()
+ messages = [
+ HumanMessage(content="查天气"),
+ AIMessage(content="", tool_calls=[_build_tool_call()]),
+ HumanMessage(content="不用查了"),
+ ]
- def test_adds_missing_tool_messages_immediately_after_ai_message(self):
- """缺失工具响应时应立即补齐 ToolMessage。"""
- middleware = PatchToolCallsMiddleware()
- messages = [
- HumanMessage(content="查天气"),
- AIMessage(content="", tool_calls=[_build_tool_call()]),
- HumanMessage(content="不用查了"),
- ]
+ result = middleware.before_agent({"messages": messages}, runtime=None)
- result = middleware.before_agent({"messages": messages}, runtime=None)
+ patched_messages = result["messages"].value
+ assert patched_messages[1] is messages[1]
+ assert isinstance(patched_messages[2], ToolMessage)
+ assert patched_messages[2].tool_call_id == "call_1"
+ assert patched_messages[2].status == "error"
+ assert "outcome is unknown" in patched_messages[2].content
+ assert "read-only" in patched_messages[2].content
+ assert patched_messages[3] is messages[2]
- patched_messages = result["messages"].value
- self.assertIs(patched_messages[1], messages[1])
- self.assertIsInstance(patched_messages[2], ToolMessage)
- self.assertEqual(patched_messages[2].tool_call_id, "call_1")
- self.assertIs(patched_messages[3], messages[2])
- def test_moves_late_tool_messages_next_to_matching_ai_message(self):
- """乱序工具响应应移动到对应 assistant 消息之后。"""
- middleware = PatchToolCallsMiddleware()
- tool_message = ToolMessage(content="晴天", tool_call_id="call_1")
- messages = [
- HumanMessage(content="查天气"),
- AIMessage(content="", tool_calls=[_build_tool_call()]),
- HumanMessage(content="再问一句"),
- tool_message,
- ]
+def test_moves_late_tool_messages_next_to_matching_ai_message():
+ """乱序工具响应应移动到对应 assistant 消息之后。"""
+ middleware = PatchToolCallsMiddleware()
+ tool_message = ToolMessage(content="晴天", tool_call_id="call_1")
+ messages = [
+ HumanMessage(content="查天气"),
+ AIMessage(content="", tool_calls=[_build_tool_call()]),
+ HumanMessage(content="再问一句"),
+ tool_message,
+ ]
- result = middleware.before_agent({"messages": messages}, runtime=None)
+ result = middleware.before_agent({"messages": messages}, runtime=None)
- patched_messages = result["messages"].value
- self.assertIs(patched_messages[1], messages[1])
- self.assertIs(patched_messages[2], tool_message)
- self.assertIs(patched_messages[3], messages[2])
- self.assertNotIn(tool_message, patched_messages[4:])
+ patched_messages = result["messages"].value
+ assert patched_messages[1] is messages[1]
+ assert patched_messages[2] is tool_message
+ assert patched_messages[3] is messages[2]
+ assert tool_message not in patched_messages[4:]
- def test_drops_orphan_tool_messages(self):
- """孤立工具响应不应继续进入模型请求历史。"""
- middleware = PatchToolCallsMiddleware()
- orphan_tool_message = ToolMessage(content="晴天", tool_call_id="call_orphan")
- messages = [
- HumanMessage(content="查天气"),
- orphan_tool_message,
- HumanMessage(content="继续"),
- ]
- result = middleware.before_agent({"messages": messages}, runtime=None)
+def test_drops_orphan_tool_messages():
+ """孤立工具响应不应继续进入模型请求历史。"""
+ middleware = PatchToolCallsMiddleware()
+ orphan_tool_message = ToolMessage(content="晴天", tool_call_id="call_orphan")
+ messages = [
+ HumanMessage(content="查天气"),
+ orphan_tool_message,
+ HumanMessage(content="继续"),
+ ]
- patched_messages = result["messages"].value
- self.assertEqual([msg.type for msg in patched_messages], ["human", "human"])
- self.assertNotIn(orphan_tool_message, patched_messages)
+ result = middleware.before_agent({"messages": messages}, runtime=None)
- def test_async_hook_normalizes_messages(self):
- """异步 Agent 执行入口也应修复工具调用历史。"""
- middleware = PatchToolCallsMiddleware()
- messages = [
- HumanMessage(content="查天气"),
- AIMessage(content="", tool_calls=[_build_tool_call()]),
- ]
+ patched_messages = result["messages"].value
+ assert [msg.type for msg in patched_messages] == ["human", "human"]
+ assert orphan_tool_message not in patched_messages
- result = asyncio.run(middleware.abefore_agent({"messages": messages}, runtime=None))
- patched_messages = result["messages"].value
- self.assertEqual([msg.type for msg in patched_messages], ["human", "ai", "tool"])
+@pytest.mark.asyncio
+async def test_async_hook_normalizes_messages():
+ """异步 Agent 执行入口也应修复工具调用历史。"""
+ middleware = PatchToolCallsMiddleware()
+ messages = [
+ HumanMessage(content="查天气"),
+ AIMessage(content="", tool_calls=[_build_tool_call()]),
+ ]
+
+ result = await middleware.abefore_agent({"messages": messages}, runtime=None)
+
+ patched_messages = result["messages"].value
+ assert [msg.type for msg in patched_messages] == ["human", "ai", "tool"]
diff --git a/tests/test_agent_plan.py b/tests/test_agent_plan.py
new file mode 100644
index 000000000..aa8baddfa
--- /dev/null
+++ b/tests/test_agent_plan.py
@@ -0,0 +1,309 @@
+"""验证计划经过真实工具执行、上下文压缩和历史恢复后的行为。"""
+
+import asyncio
+import json
+from unittest.mock import Mock
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import (
+ AIMessage,
+ BaseMessage,
+ HumanMessage,
+ ToolMessage,
+ messages_from_dict,
+ messages_to_dict,
+)
+from langchain_core.tools import StructuredTool
+from langgraph.checkpoint.memory import InMemorySaver
+from pydantic import Field
+
+from app.agent.middleware.plan import PLAN_SNAPSHOT_KEY, PlanMiddleware, attach_plan_snapshot
+from app.agent.middleware.policy import AgentPolicyMiddleware
+from app.agent.middleware.selection import TOOL_DISCOVERY_NAME, ToolSelectorMiddleware
+from app.agent.middleware.summarization import (
+ ContextPreservingSummarizationMiddleware,
+ FinalRequestCompactionMiddleware,
+)
+from app.agent.policy.contracts import AuthSource, PrincipalType, ToolOrigin, ToolPolicyContext
+from app.agent.policy.orchestrator import AgentToolPolicyOrchestrator
+from app.agent.tools.catalog import ToolCatalogSnapshot
+
+
+class _RecordingModel(FakeMessagesListChatModel):
+ """在离线图运行中捕获工具 schema 和送往模型的真实消息。"""
+
+ seen: list[list[BaseMessage]] = Field(default_factory=list)
+ tool_names: list[str] = Field(default_factory=list)
+ tool_history: list[list[str]] = Field(default_factory=list)
+
+ def bind_tools(self, tools, **_kwargs):
+ """记录工具绑定并复用确定性响应模型。"""
+ self.tool_names = [tool.name for tool in tools]
+ self.tool_history.append(list(self.tool_names))
+ return self
+
+ def _generate(self, messages, *args, **kwargs):
+ """保留每次模型调用收到的请求内容。"""
+ self.seen.append(list(messages))
+ return super()._generate(messages, *args, **kwargs)
+
+
+def _plan(objective="修复整理失败并验证结果"):
+ """构造包含未完成工作和具体上下文的合法计划。"""
+ return {
+ "objective": objective,
+ "steps": [
+ {"step": "检查失败记录", "status": "completed", "evidence": "记录 42 缺少目标目录"},
+ {"step": "修复目标配置", "status": "in_progress", "evidence": ""},
+ {"step": "检查整理结果", "status": "pending", "evidence": ""},
+ ],
+ "explanation": "用户允许修复该目标目录，整理成功后核对记录",
+ }
+
+
+def _update_message(plan=None, call_id="plan-1"):
+ """生成符合真实工具调用协议的计划更新请求。"""
+ return AIMessage(content="", tool_calls=[{
+ "name": "update_plan", "args": plan or _plan(), "id": call_id,
+ }])
+
+
+def _graph(model, *middlewares):
+ """构造带独立会话 checkpoint 的最小计划运行图。"""
+ return create_agent(
+ model=model, middleware=[PlanMiddleware(), *middlewares], checkpointer=InMemorySaver(),
+ )
+
+
+def _config(thread="plan-session"):
+ """为每个独立会话提供固定的图配置。"""
+ return {"configurable": {"thread_id": thread}}
+
+
+def test_plan_tool_updates_model_context_and_final_snapshot():
+ """工具更新须真实写入图状态，并在下一模型调用中呈现未完成步骤。"""
+ model = _RecordingModel(responses=[_update_message(), AIMessage(content="正在处理目录配置。")])
+ graph = _graph(model)
+
+ result = asyncio.run(graph.ainvoke({"messages": [HumanMessage(content="修复整理失败")]}, _config()))
+
+ assert result.get("task_plan") == _plan(), [message.content for message in result["messages"]]
+ assert "update_plan" in model.tool_names
+ assert "记录 42 缺少目标目录" in model.seen[1][0].text
+ assert "修复目标配置" in model.seen[1][0].text
+ assert result["messages"][-1].additional_kwargs[PLAN_SNAPSHOT_KEY] == _plan()
+ schema = PlanMiddleware().tools[0].args_schema.model_json_schema()
+ assert "runtime" not in schema["properties"]
+
+
+def test_plan_survives_real_compaction_and_message_history_rebuild():
+ """真实压缩删除旧工具回执后，计划仍进模型请求并能从序列化历史恢复。"""
+ model = _RecordingModel(
+ responses=[_update_message(), AIMessage(content="已定位。"), AIMessage(content="继续修复。")],
+ profile={"max_input_tokens": 8192},
+ )
+ summary_model = FakeMessagesListChatModel(
+ responses=[AIMessage(content="旧对话已压缩。")], profile={"max_input_tokens": 8192},
+ )
+ compactor = FinalRequestCompactionMiddleware(summarizer=ContextPreservingSummarizationMiddleware(
+ model=summary_model, trigger=("fraction", 0.85), keep=("messages", 4),
+ ))
+ graph = _graph(model, compactor)
+
+ async def _run():
+ """先建立计划，再在后续轮次制造真实上下文压缩。"""
+ await graph.ainvoke({"messages": [HumanMessage(content="修复整理失败")]}, _config())
+ history = [
+ message for _ in range(12)
+ for message in (HumanMessage(content="旧排查信息 " * 400), AIMessage(content="旧观测数据 " * 400))
+ ]
+ return await graph.ainvoke({"messages": [*history, HumanMessage(content="继续修复，不改变目标")]}, _config())
+
+ result = asyncio.run(_run())
+ assert any(message.additional_kwargs.get("lc_source") == "summarization" for message in result["messages"])
+ assert not any(isinstance(message, ToolMessage) and message.name == "update_plan" for message in result["messages"])
+ assert _plan()["objective"] in model.seen[-1][0].text
+
+ history = messages_from_dict(messages_to_dict(result["messages"]))
+ restored_model = _RecordingModel(responses=[AIMessage(content="继续原任务。")])
+ restored = _graph(restored_model)
+ restored_result = asyncio.run(restored.ainvoke({"messages": [*history, HumanMessage(content="继续")]}, _config("rebuilt")))
+ assert restored_result["task_plan"] == _plan()
+ assert "记录 42 缺少目标目录" in restored_model.seen[0][0].text
+
+
+def test_plan_status_question_preserves_objective_and_new_task_replaces_it():
+ """同一会话的状态追问保留计划，明确新任务能通过工具替换旧计划。"""
+ new_plan = _plan("排查另一个下载器连接问题")
+ model = _RecordingModel(responses=[
+ _update_message(), AIMessage(content="正在处理。"), AIMessage(content="尚在修复配置。"),
+ _update_message(new_plan, "plan-2"), AIMessage(content="开始排查新下载器。"),
+ ])
+ graph = _graph(model)
+
+ async def _run():
+ """在同一缓存图执行建计划、追问和替换三个轮次。"""
+ await graph.ainvoke({"messages": [HumanMessage(content="修复整理")]}, _config())
+ progress = await graph.ainvoke({"messages": [HumanMessage(content="现在进展如何")]}, _config())
+ changed = await graph.ainvoke({"messages": [HumanMessage(content="现在排查另一个下载器")]}, _config())
+ return progress, changed
+
+ progress, changed = asyncio.run(_run())
+ assert progress["task_plan"] == _plan()
+ assert changed["task_plan"] == new_plan
+
+
+@pytest.mark.parametrize("invalid", [
+ {"objective": "", "steps": [{"step": "检查", "status": "pending"}]},
+ {"objective": "检查", "steps": [{"step": "检查", "status": "completed"}]},
+ {"objective": "检查", "steps": [{"step": "检查", "status": "blocked"}]},
+ {"objective": "检查", "steps": [{"step": "检查", "status": "pending"}] * 2},
+ {"objective": "检查", "steps": [{"step": str(index), "status": "pending"} for index in range(13)]},
+])
+def test_invalid_plan_update_keeps_existing_plan(invalid):
+ """非法输入必须成为可恢复工具错误，不覆盖已保存的真实任务上下文。"""
+ model = _RecordingModel(responses=[_update_message(invalid), AIMessage(content="保留原计划。")])
+ graph = _graph(model)
+ history = attach_plan_snapshot([AIMessage(content="旧进度")], _plan())
+
+ result = asyncio.run(graph.ainvoke({"messages": [*history, HumanMessage(content="继续")]}, _config()))
+
+ assert result["task_plan"] == _plan()
+ assert any(isinstance(message, ToolMessage) and message.status == "error" for message in result["messages"])
+
+
+def test_parallel_plan_updates_are_rejected_without_state_conflict():
+ """同次响应的两个计划更新均拒绝，真实 ToolNode 不得出现并发写冲突。"""
+ double_update = AIMessage(content="", tool_calls=[
+ {"name": "update_plan", "args": _plan("替换一"), "id": "one"},
+ {"name": "update_plan", "args": _plan("替换二"), "id": "two"},
+ ])
+ model = _RecordingModel(responses=[double_update, AIMessage(content="已保留原计划。")])
+ graph = _graph(model)
+ history = attach_plan_snapshot([AIMessage(content="旧进度")], _plan())
+
+ result = asyncio.run(graph.ainvoke({"messages": [*history, HumanMessage(content="继续")]}, _config()))
+
+ assert result["task_plan"] == _plan()
+ errors = [message for message in result["messages"] if isinstance(message, ToolMessage) and message.status == "error"]
+ assert {message.tool_call_id for message in errors} == {"one", "two"}
+
+
+def test_plan_isolated_between_threads_and_simple_question_needs_no_plan():
+ """共享图不能泄露另一会话的计划，简单问题直接回答即可。"""
+ model = _RecordingModel(responses=[_update_message(), AIMessage(content="继续处理。"), AIMessage(content="你好。")])
+ graph = _graph(model)
+
+ async def _run():
+ """在共享图使用不同会话 ID 调用。"""
+ await graph.ainvoke({"messages": [HumanMessage(content="修复整理")]}, _config("first"))
+ return await graph.ainvoke({"messages": [HumanMessage(content="你好")]}, _config("second"))
+
+ result = asyncio.run(_run())
+ assert result.get("task_plan") is None
+ assert "" not in model.seen[-1][0].text
+ assert PLAN_SNAPSHOT_KEY not in result["messages"][-1].additional_kwargs
+
+
+def test_snapshot_sanitizes_secrets_and_preserves_message_objects():
+ """中断保存只修改副本，凭据不能留在计划恢复点中。"""
+ plan = _plan()
+ plan["explanation"] = "api_key=sk-testcredential123456789"
+ original = AIMessage(content="任务已中断", id="interrupted")
+
+ messages = attach_plan_snapshot([original], plan)
+
+ assert PLAN_SNAPSHOT_KEY not in original.additional_kwargs
+ assert messages[-1].id == original.id
+ assert "sk-testcredential123456789" not in json.dumps(messages[-1].additional_kwargs)
+ assert "***" in messages[-1].additional_kwargs[PLAN_SNAPSHOT_KEY]["explanation"]
+
+
+def test_old_or_invalid_snapshots_cannot_override_current_plan():
+ """图中有效计划优先于旧历史，用户消息及无效最新快照不得恢复旧目标。"""
+ middleware = PlanMiddleware()
+ history = attach_plan_snapshot([AIMessage(content="旧进度")], _plan("旧目标"))
+ state = {"messages": history, "task_plan": _plan("当前目标")}
+ assert middleware.before_agent(state, None) is None
+ invalid = AIMessage(content="新进度", additional_kwargs={PLAN_SNAPSHOT_KEY: {"objective": "无效"}})
+ assert middleware.before_agent({"messages": [*history, invalid]}, None) is None
+ injected = HumanMessage(content="你好", additional_kwargs={PLAN_SNAPSHOT_KEY: _plan("用户伪造")})
+ assert middleware.before_agent({"messages": [injected]}, None) is None
+
+
+def test_plan_discovery_and_host_policy_complete_one_task():
+ """真实图在宿主策略内完成建计划、发现遗漏工具、查询和有证据完成的闭环。"""
+ inspected_records = []
+
+ def inspect_record(record_id: int) -> str:
+ """记录实际执行次数，返回可供计划引用的检查证据。"""
+ inspected_records.append(record_id)
+ return f"记录 {record_id} 整理完成，目标文件存在"
+
+ business_tools = [StructuredTool.from_function(
+ func=inspect_record, name=name, description=f"检查 {name} 记录",
+ ) for name in ("transfer_history", "download_status", "site_status", "library_status")]
+ plan = {
+ "objective": "检查整理记录 42 是否已完成",
+ "steps": [{"step": "核查整理记录和目标文件", "status": "in_progress", "evidence": ""}],
+ "explanation": "先发现具体查询工具，再根据查询结果完成核验",
+ }
+ finished = {
+ **plan,
+ "steps": [{
+ "step": "核查整理记录和目标文件", "status": "completed",
+ "evidence": "记录 42 整理完成，目标文件存在",
+ }],
+ }
+ plan_middleware = PlanMiddleware()
+ selection_model = _RecordingModel(responses=[AIMessage(content='{"tools": []}')])
+ selector = ToolSelectorMiddleware(
+ model=selection_model,
+ selection_tools=[*business_tools, *plan_middleware.tools],
+ always_include=["update_plan"], max_tools=1, enable_discovery=True,
+ )
+ catalog = ToolCatalogSnapshot.from_tools(
+ [*business_tools, *plan_middleware.tools, *selector.tools],
+ plugin_revision=0, factory_revision="integration-test",
+ ).require_unique()
+ policy_orchestrator = AgentToolPolicyOrchestrator()
+ policy_orchestrator.start = Mock(wraps=policy_orchestrator.start)
+ policy = AgentPolicyMiddleware(
+ context=ToolPolicyContext(
+ session_id="integrated-plan", user_id="test-user", origin=ToolOrigin.AGENT_INTERACTIVE,
+ principal_type=PrincipalType.HUMAN, auth_source=AuthSource.INTERNAL, agent_context={},
+ ),
+ orchestrator=policy_orchestrator, catalog=catalog,
+ )
+ model = _RecordingModel(responses=[
+ _update_message(plan),
+ AIMessage(content="", tool_calls=[{
+ "name": TOOL_DISCOVERY_NAME, "args": {"search": "transfer_history", "limit": 1}, "id": "discover",
+ }]),
+ AIMessage(content="", tool_calls=[{
+ "name": "transfer_history", "args": {"record_id": 42}, "id": "inspect",
+ }]),
+ _update_message(finished, "plan-finished"),
+ AIMessage(content="记录 42 已整理完成，目标文件存在。"),
+ ])
+ graph = create_agent(
+ model=model, tools=business_tools, middleware=[policy, plan_middleware, selector],
+ checkpointer=InMemorySaver(),
+ )
+
+ result = asyncio.run(graph.ainvoke(
+ {"messages": [HumanMessage(content="检查整理记录 42")]}, _config("integrated-plan"),
+ ))
+
+ assert inspected_records == [42]
+ assert result["task_plan"] == finished
+ assert result["messages"][-1].additional_kwargs[PLAN_SNAPSHOT_KEY] == finished
+ assert "transfer_history" not in model.tool_history[0]
+ assert "transfer_history" in model.tool_history[2]
+ assert all("update_plan" in names and TOOL_DISCOVERY_NAME in names for names in model.tool_history)
+ observed_tools = [call.kwargs["tool"] for call in policy_orchestrator.start.call_args_list]
+ assert [tool.name for tool in observed_tools] == ["update_plan", TOOL_DISCOVERY_NAME, "transfer_history", "update_plan"]
+ assert all(catalog.resolve_unique(tool.name).tool is tool for tool in observed_tools)
+ assert "记录 42 整理完成，目标文件存在" in model.seen[-1][0].text
diff --git a/tests/test_agent_recovery.py b/tests/test_agent_recovery.py
new file mode 100644
index 000000000..406121392
--- /dev/null
+++ b/tests/test_agent_recovery.py
@@ -0,0 +1,460 @@
+"""Agent 中断后恢复已确认事实，避免重复执行有副作用的工具。"""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage, messages_to_dict
+from langchain_core.tools import tool
+from langgraph.checkpoint.memory import InMemorySaver
+
+from app.agent.memory import MemoryManager
+from app.agent.middleware.patching import PatchToolCallsMiddleware
+from app.agent.middleware.plan import PLAN_SNAPSHOT_KEY
+from app.agent.orchestrator import MoviePilotAgent
+
+
+class _InterruptedGraph:
+ """在图已保存工具状态后失败或等待取消的执行替身。"""
+
+ def __init__(self, messages, *, blocked=False, state_error=False):
+ """保存可恢复图状态和由用例控制的故障模式。"""
+ self.messages = messages
+ self.blocked = blocked
+ self.state_error = state_error
+ self.started = asyncio.Event()
+
+ async def ainvoke(self, _payload, config=None):
+ """通知工具执行已结束，再模拟下一次模型调用中断。"""
+ self.started.set()
+ if self.blocked:
+ await asyncio.Event().wait()
+ raise RuntimeError("model request failed")
+
+ async def astream(self, payload, **_kwargs):
+ """保持与真实流式图相同的异常传播边界。"""
+ await self.ainvoke(payload)
+ yield {}
+
+ def get_state(self, _config):
+ """提供已完成工具回执，或模拟 checkpoint 不可读。"""
+ if self.state_error:
+ raise RuntimeError("checkpoint unavailable")
+ return SimpleNamespace(values={"messages": self.messages})
+
+
+@pytest.fixture
+def recovery_agent(monkeypatch):
+ """隔离模型、持久化和通知边界，仅保留真实会话记忆缓存。"""
+ persistence = SimpleNamespace(async_save_agent_messages=AsyncMock())
+ memory = MemoryManager(persistence=persistence)
+ agent = MoviePilotAgent(
+ session_id="recovery-session",
+ user_id="10001",
+ channel="WebAgent",
+ source="web",
+ memory=memory,
+ )
+ monkeypatch.setattr(agent, "_should_stream", lambda: False)
+ monkeypatch.setattr(agent, "_dispatch_execution_notice", AsyncMock())
+ monkeypatch.setattr(agent, "_send_agent_tokens_usage_event", lambda **_kwargs: None)
+ monkeypatch.setattr(agent, "_save_assistant_display_message_once", AsyncMock())
+ monkeypatch.setattr(agent, "send_agent_message", AsyncMock())
+ agent.stream_handler = SimpleNamespace(
+ stop_streaming=AsyncMock(return_value=(False, "")),
+ set_dispatch_policy=lambda **_kwargs: None,
+ start_streaming=AsyncMock(),
+ )
+ return agent, persistence
+
+
+def _completed_tool_messages():
+ """构造已创建下载、另一个写操作尚无回执的会话。"""
+ return [
+ HumanMessage(content="下载电影并更新订阅"),
+ AIMessage(
+ content="开始处理",
+ tool_calls=[
+ {"id": "download-call", "name": "download", "args": {"media_id": "42"}},
+ {"id": "subscription-call", "name": "subscribe", "args": {"media_id": "42"}},
+ ],
+ ),
+ ToolMessage(content='{"download_id": "already-created"}', tool_call_id="download-call"),
+ ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_failure_preserves_completed_receipts_and_marks_unknown_outcomes(recovery_agent, streaming):
+ """失败后的新一轮输入应包含已完成回执，悬空写调用保持结果未知。"""
+ agent, persistence = recovery_agent
+ agent._should_stream = lambda: streaming
+ original = _completed_tool_messages()
+ graph = _InterruptedGraph(original)
+ agent._create_agent = AsyncMock(return_value=graph)
+ agent._compiled_agent_bundle = object()
+
+ result, _ = await agent._execute_agent(original[:1])
+
+ assert result == "智能助手执行失败，请稍后重试"
+ assert agent._compiled_agent_bundle is None
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ receipts = [message for message in recovered if isinstance(message, ToolMessage)]
+ assert receipts[0].content == original[2].content
+ assert receipts[0].status == "success"
+ assert receipts[1].tool_call_id == "subscription-call"
+ assert receipts[1].status == "error"
+ assert "unknown" in receipts[1].content.lower()
+ assert "read-only" in receipts[1].content.lower()
+ assert not recovered[-1].tool_calls
+ assert len(original) == 3
+ persistence.async_save_agent_messages.assert_awaited_once()
+
+ agent._execute_agent = AsyncMock(return_value=("继续核验状态", {}))
+ agent.prepare_chat_title = AsyncMock()
+ agent._save_display_history_messages = AsyncMock()
+ await agent.process("继续")
+ next_messages = agent._execute_agent.call_args.args[0]
+ assert next_messages[:-1] == recovered
+ assert json.loads(next_messages[-1].content[0]["text"])["message"] == "继续"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("streaming", [False, True])
+async def test_cancellation_preserves_snapshot_and_still_propagates(recovery_agent, streaming):
+ """用户取消必须保存完成事实，同时保持调用方收到 CancelledError。"""
+ agent, persistence = recovery_agent
+ agent._should_stream = lambda: streaming
+ graph = _InterruptedGraph(_completed_tool_messages(), blocked=True)
+ agent._create_agent = AsyncMock(return_value=graph)
+ agent._compiled_agent_bundle = object()
+ execution = asyncio.create_task(agent._execute_agent(graph.messages[:1]))
+ await graph.started.wait()
+ execution.cancel()
+
+ with pytest.raises(asyncio.CancelledError):
+ await execution
+
+ assert agent._compiled_agent_bundle is None
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ assert "already-created" in str(messages_to_dict(recovered))
+ assert "取消" in recovered[-1].content
+ persistence.async_save_agent_messages.assert_awaited_once()
+ agent.stream_handler.stop_streaming.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_recovery_sanitizes_checkpoint_without_mutating_live_messages(recovery_agent):
+ """中断快照清理凭据和供应商载荷，不把工具 artifact 或异常体落盘。"""
+ agent, persistence = recovery_agent
+ original = _completed_tool_messages()
+ original[0].content = "api_key=raw-user-secret"
+ original[1].tool_calls[0]["args"]["password"] = "raw-tool-secret"
+ original[1].additional_kwargs = {"raw_provider_payload": "private-provider-body"}
+ original[2].content = '{"download_id": "already-created", "token": "raw-result-secret"}'
+ original[2].artifact = {"payload": "private-artifact"}
+ original[2].response_metadata = {"payload": "private-metadata"}
+ agent._create_agent = AsyncMock(return_value=_InterruptedGraph(original))
+
+ await agent._execute_agent(original[:1])
+
+ saved = json.dumps(persistence.async_save_agent_messages.call_args.kwargs["messages"])
+ assert "already-created" in saved
+ for secret in (
+ "raw-user-secret", "raw-tool-secret", "raw-result-secret",
+ "private-provider-body", "private-artifact", "private-metadata",
+ ):
+ assert secret not in saved
+ assert original[1].tool_calls[0]["args"]["password"] == "raw-tool-secret"
+ assert original[2].artifact == {"payload": "private-artifact"}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", ["checkpoint", "persistence", "creation", "configuration"])
+async def test_snapshot_failure_does_not_replace_original_error(recovery_agent, failure):
+ """读取或保存恢复快照失败时，仍按原来的执行错误收尾。"""
+ agent, _persistence = recovery_agent
+ graph = _InterruptedGraph(_completed_tool_messages(), state_error=failure == "checkpoint")
+ agent._create_agent = AsyncMock(return_value=graph)
+ if failure == "creation":
+ agent._create_agent.side_effect = RuntimeError("creation failed")
+ if failure == "configuration":
+ agent._get_recursion_limit = lambda: int("invalid configuration")
+ if failure == "persistence":
+ agent._memory.async_save_agent_messages = AsyncMock(side_effect=RuntimeError("save failed"))
+ agent._compiled_agent_bundle = object()
+
+ result, _ = await agent._execute_agent(graph.messages[:1])
+
+ assert result == "智能助手执行失败，请稍后重试"
+ assert agent._compiled_agent_bundle is None
+ agent.stream_handler.stop_streaming.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_recovery_does_not_create_chat_history_for_background_tasks(recovery_agent):
+ """无渠道后台任务沿用不保存普通会话的边界。"""
+ agent, persistence = recovery_agent
+ agent.channel = None
+ agent.source = None
+ agent._create_agent = AsyncMock(return_value=_InterruptedGraph(_completed_tool_messages()))
+
+ await agent._execute_agent([])
+
+ persistence.async_save_agent_messages.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_snapshot_timeout_keeps_memory_and_does_not_delay_cancellation(recovery_agent, monkeypatch):
+ """持久化等待有界，超时后内存事实仍可复用且取消照常收敛。"""
+ agent, persistence = recovery_agent
+ graph = _InterruptedGraph(_completed_tool_messages(), blocked=True)
+ agent._create_agent = AsyncMock(return_value=graph)
+ monkeypatch.setattr("app.agent.orchestrator.AGENT_RECOVERY_SNAPSHOT_TIMEOUT", 0.01)
+
+ async def block_save(**_kwargs):
+ """模拟数据库写边界迟迟未返回，不访问真实数据库。"""
+ await asyncio.Event().wait()
+
+ persistence.async_save_agent_messages.side_effect = block_save
+ execution = asyncio.create_task(agent._execute_agent(graph.messages[:1]))
+ await graph.started.wait()
+ execution.cancel()
+ with pytest.raises(asyncio.CancelledError):
+ await asyncio.wait_for(execution, timeout=1)
+
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ assert "already-created" in str(messages_to_dict(recovered))
+
+
+@pytest.mark.asyncio
+async def test_second_cancellation_during_snapshot_still_invalidates_graph(recovery_agent):
+ """保存中断快照时再次取消，也不能遗留旧图让后续请求重用。"""
+ agent, persistence = recovery_agent
+ graph = _InterruptedGraph(_completed_tool_messages(), blocked=True)
+ agent._create_agent = AsyncMock(return_value=graph)
+ agent._compiled_agent_bundle = object()
+ saving = asyncio.Event()
+
+ async def block_save(**_kwargs):
+ """将第二次取消定位到持久化等待阶段。"""
+ saving.set()
+ await asyncio.Event().wait()
+
+ persistence.async_save_agent_messages.side_effect = block_save
+ execution = asyncio.create_task(agent._execute_agent(graph.messages[:1]))
+ await graph.started.wait()
+ execution.cancel()
+ await asyncio.wait_for(saving.wait(), timeout=1)
+ execution.cancel()
+
+ with pytest.raises(asyncio.CancelledError):
+ await execution
+
+ assert agent._compiled_agent_bundle is None
+ agent.stream_handler.stop_streaming.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_interrupted_snapshot_preserves_sanitized_plan(recovery_agent):
+ """图中的未完成计划随中断提示写入快照，凭据不会进入计划恢复数据。"""
+ agent, persistence = recovery_agent
+ graph = _InterruptedGraph(_completed_tool_messages())
+ plan = {
+ "objective": "处理下载",
+ "steps": [{"step": "核验任务", "status": "in_progress", "evidence": ""}],
+ "explanation": "password=private-plan-secret",
+ }
+ graph.get_state = lambda _config: SimpleNamespace(values={"messages": graph.messages, "task_plan": plan})
+ agent._create_agent = AsyncMock(return_value=graph)
+
+ await agent._execute_agent(graph.messages[:1])
+
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ snapshot = recovered[-1].additional_kwargs[PLAN_SNAPSHOT_KEY]
+ assert snapshot["steps"][0]["status"] == "in_progress"
+ assert "private-plan-secret" not in str(snapshot)
+ assert "private-plan-secret" not in str(persistence.async_save_agent_messages.call_args)
+ assert plan["explanation"] == "password=private-plan-secret"
+
+
+class _InterruptedModel(FakeMessagesListChatModel):
+ """允许首次工具执行后模型失败，再以同样的历史继续回应。"""
+
+ fail_after_tool: bool = True
+
+ def bind_tools(self, _tools, **_kwargs):
+ """测试模型固定返回工具协议消息，不需要连接真实供应商。"""
+ return self
+
+ def _generate(self, messages, stop=None, run_manager=None, **kwargs):
+ """第二次模型请求只失败一次，保留已完成工具所在的 checkpoint。"""
+ if self.i == 1 and self.fail_after_tool:
+ self.fail_after_tool = False
+ raise RuntimeError("model disconnected after download")
+ return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+@pytest.mark.asyncio
+async def test_real_graph_rebuild_uses_receipts_without_replaying_completed_tool(recovery_agent):
+ """真实 LangGraph 中工具已成功、后续模型失败时，重建图不重放旧工具。"""
+ agent, _persistence = recovery_agent
+ writes = []
+
+ @tool
+ def add_download() -> str:
+ """记录一次不可重复创建的测试下载。"""
+ writes.append("download-created")
+ return '{"download_id": "created-once"}'
+
+ model = _InterruptedModel(responses=[
+ AIMessage(content="", tool_calls=[{"id": "add-once", "name": "add_download", "args": {}}]),
+ AIMessage(content="已核验之前创建的下载"),
+ ])
+ graphs = [
+ create_agent(
+ model=model,
+ tools=[add_download],
+ middleware=[PatchToolCallsMiddleware()],
+ checkpointer=InMemorySaver(),
+ )
+ for _ in range(2)
+ ]
+ agent._create_agent = AsyncMock(side_effect=graphs)
+ first_message = HumanMessage(content="创建一个下载")
+
+ await agent._execute_agent([first_message])
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ assert any(isinstance(message, ToolMessage) and "created-once" in message.content for message in recovered)
+ await agent._execute_agent([*recovered, HumanMessage(content="继续核验")])
+
+ assert writes == ["download-created"]
+ assert agent._streamed_output == "已核验之前创建的下载"
+
+
+@pytest.mark.asyncio
+async def test_parallel_tool_failure_keeps_successful_pending_write(recovery_agent):
+ """同批并行工具部分失败时，checkpoint 的成功 pending write 仍进入恢复历史。"""
+ agent, _persistence = recovery_agent
+ completed = asyncio.Event()
+ writes = []
+
+ @tool
+ async def add_download() -> str:
+ """模拟已生效的下载写操作，并通知并行失败工具。"""
+ writes.append("download-created")
+ completed.set()
+ return '{"download_id": "parallel-created-once"}'
+
+ @tool
+ async def update_subscription() -> str:
+ """等待下载生效后模拟另一路写操作失败。"""
+ await completed.wait()
+ raise RuntimeError("subscription connection lost")
+
+ model = _InterruptedModel(fail_after_tool=False, responses=[
+ AIMessage(content="", tool_calls=[
+ {"id": "parallel-download", "name": "add_download", "args": {}},
+ {"id": "parallel-subscription", "name": "update_subscription", "args": {}},
+ ]),
+ AIMessage(content="下载已有成功回执，继续核验订阅状态"),
+ ])
+ graphs = [
+ create_agent(
+ model=model,
+ tools=[add_download, update_subscription],
+ middleware=[PatchToolCallsMiddleware()],
+ checkpointer=InMemorySaver(),
+ )
+ for _ in range(2)
+ ]
+ agent._create_agent = AsyncMock(side_effect=graphs)
+
+ await agent._execute_agent([HumanMessage(content="下载并更新订阅")])
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ receipts = {message.tool_call_id: message for message in recovered if isinstance(message, ToolMessage)}
+ assert receipts["parallel-download"].status == "success"
+ assert "parallel-created-once" in receipts["parallel-download"].content
+ assert receipts["parallel-subscription"].status == "error"
+ assert "outcome is unknown" in receipts["parallel-subscription"].content
+
+ await agent._execute_agent([*recovered, HumanMessage(content="继续")])
+
+ assert writes == ["download-created"]
+ assert agent._streamed_output == "下载已有成功回执，继续核验订阅状态"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("provider", ["interleaved", "gemini", "anthropic", "responses"])
+async def test_signed_history_recovers_as_portable_facts_without_broken_protocol(recovery_agent, provider):
+ """签名或思考工具链转为普通历史事实，供应商 SDK 可重新组装下一轮请求。"""
+ from langchain_anthropic.chat_models import _format_messages
+ from langchain_google_genai.chat_models import _parse_chat_history
+ from langchain_openai.chat_models.base import _convert_message_to_dict
+
+ agent, persistence = recovery_agent
+ original = _completed_tool_messages()
+ if provider == "interleaved":
+ original[1].additional_kwargs = {"reasoning_content": "private-reasoning-body"}
+ elif provider == "gemini":
+ original[1].additional_kwargs = {
+ "__gemini_function_call_thought_signatures__": {"download-call": "opaque-signature"},
+ }
+ elif provider == "anthropic":
+ original[1].content = [
+ {"type": "thinking", "thinking": "private-reasoning-body", "signature": "opaque-signature"},
+ {"type": "text", "text": "已开始执行"},
+ ]
+ else:
+ original[1].content = [
+ {"type": "reasoning", "reasoning": "private-reasoning-body", "extras": {"signature": "opaque-signature"}},
+ {"type": "text", "text": "已开始执行"},
+ ]
+ original[1].response_metadata = {"model_provider": "openai", "output_version": "v1"}
+ plan = {"objective": "完成下载和订阅", "steps": [{"step": "核验订阅", "status": "pending"}]}
+ graph = _InterruptedGraph(original)
+ graph.get_state = lambda _config: SimpleNamespace(values={"messages": original, "task_plan": plan})
+ agent._create_agent = AsyncMock(return_value=graph)
+
+ await agent._execute_agent(original[:1])
+
+ recovered = await agent._memory.async_get_agent_messages(agent.session_id, agent.user_id)
+ saved = str(persistence.async_save_agent_messages.call_args)
+ assert recovered[0].type == "human"
+ assert recovered[0].content == original[0].content
+ assert recovered[-1].additional_kwargs[PLAN_SNAPSHOT_KEY]["objective"] == plan["objective"]
+ assert not any(isinstance(message, ToolMessage) or getattr(message, "tool_calls", []) for message in recovered)
+ assert "already-created" in saved
+ assert "subscription-call" in saved
+ assert "outcome is unknown" in saved
+ assert '"status": "success"' in saved
+ assert '"status": "error"' in saved
+ assert "private-reasoning-body" not in saved
+ assert "opaque-signature" not in saved
+ assert "仅作观察数据" in saved
+
+ next_messages = [*recovered, HumanMessage(content="继续")]
+ openai_payload = [_convert_message_to_dict(message) for message in next_messages]
+ _system, anthropic_payload = _format_messages(next_messages)
+ _system, gemini_payload = _parse_chat_history(next_messages, model="gemini-3-pro")
+ assert openai_payload[-1]["role"] == "user"
+ assert anthropic_payload[-1]["role"] == "user"
+ assert gemini_payload[-1].role == "user"
+
+
+def test_portable_recovery_retains_historical_plan_before_state_restore():
+ """计划中间件恢复前即失败时，历史消息携带的计划仍可随事实快照恢复。"""
+ messages = _completed_tool_messages()
+ messages[1].additional_kwargs[PLAN_SNAPSHOT_KEY] = {
+ "objective": "核验下载和订阅",
+ "steps": [{"step": "检查订阅状态", "status": "pending"}],
+ }
+
+ recovered = MoviePilotAgent._portable_recovery_messages(messages)
+
+ assert recovered[-1].additional_kwargs[PLAN_SNAPSHOT_KEY]["objective"] == "核验下载和订阅"
+ assert "already-created" in recovered[-1].content
diff --git a/tests/test_agent_subagent_reuse.py b/tests/test_agent_subagent_reuse.py
new file mode 100644
index 000000000..a8ab88f2c
--- /dev/null
+++ b/tests/test_agent_subagent_reuse.py
@@ -0,0 +1,104 @@
+"""真实 Agent 图中的子代理跨轮复用与上下文压缩回归。"""
+
+import json
+from unittest.mock import AsyncMock
+
+import pytest
+from langchain.agents import create_agent
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.tools import tool
+
+from app.agent.middleware.subagents import (
+ SubAgentTaskControlMiddleware,
+ _builtin_subagent_profiles,
+)
+from app.agent.middleware.summarization import ContextPreservingSummarizationMiddleware
+
+
+@pytest.fixture
+def anyio_backend():
+ """固定使用运行时支持的 asyncio 后端。"""
+ return "asyncio"
+
+
+class _TaskModel(FakeMessagesListChatModel):
+ """提供确定的工具调用响应，避免测试访问真实模型。"""
+
+ def bind_tools(self, tools, **kwargs):
+ """允许 LangChain 绑定工具并复用脚本响应。"""
+ return self
+
+
+def _task_call(call_id: str) -> AIMessage:
+ """生成需要子代理返回结果的委派调用。"""
+ return AIMessage(content="", tool_calls=[{
+ "name": "subagent_task",
+ "args": {"action": "run", "description": "检查下载器状态", "timeout_ms": 1000},
+ "id": call_id,
+ "type": "tool_call",
+ }])
+
+
+@pytest.mark.anyio
+async def test_cached_graph_can_delegate_on_consecutive_turns(monkeypatch):
+ """正常结束仅回收本轮任务，缓存图下一轮仍能实际调用子代理。"""
+ model = _TaskModel(responses=[_task_call("one"), AIMessage(content="检查完成"),
+ _task_call("two"), AIMessage(content="再次完成")])
+ middleware = SubAgentTaskControlMiddleware(
+ model=model, profiles=_builtin_subagent_profiles(), tools=[],
+ )
+ run_task = AsyncMock(return_value="下载器在线")
+ monkeypatch.setattr(middleware._provider, "run_task", run_task)
+ graph = create_agent(model=model, middleware=[middleware])
+
+ for question in ("检查下载器", "再检查一次"):
+ result = await graph.ainvoke({"messages": [HumanMessage(content=question)]})
+ output = next(message for message in result["messages"] if message.type == "tool")
+ assert json.loads(output.content)["tasks"][0]["result"] == "下载器在线"
+ assert middleware._tasks == {}
+
+ assert run_task.await_count == 2
+
+
+@pytest.mark.anyio
+async def test_turn_cleanup_never_reopens_sealed_controller():
+ """永久关闭的 owner 不会因回合结束被重新开放。"""
+ middleware = SubAgentTaskControlMiddleware(
+ model=_TaskModel(responses=[AIMessage(content="完成")]),
+ profiles=_builtin_subagent_profiles(), tools=[],
+ )
+ middleware.seal()
+ await middleware.aafter_agent({}, None)
+ response = json.loads(await middleware._control_task(action="start", description="检查"))
+ assert response["success"] is False
+ assert "关闭" in response["error"]
+ assert await middleware.close() is True
+
+
+@pytest.mark.anyio
+async def test_subagent_compacts_large_input_before_model_call(monkeypatch):
+ """独立子图也执行最终请求压缩，大量工具输出不会绕过预算边界。"""
+ @tool
+ def diagnostic() -> str:
+ """读取诊断日志，不访问外部服务。"""
+ return "诊断日志中的连接信息。" * 450
+
+ diagnostic.tags = ["read", "system"]
+ model = _TaskModel(
+ responses=[*[AIMessage(content="", tool_calls=[{
+ "name": "diagnostic", "args": {}, "id": f"log-{index}", "type": "tool_call",
+ }]) for index in range(3)], AIMessage(content="诊断完成")],
+ profile={"max_input_tokens": 4096, "max_output_tokens": 512},
+ )
+ middleware = SubAgentTaskControlMiddleware(
+ model=model, profiles=_builtin_subagent_profiles(), tools=[diagnostic],
+ )
+ summarize = AsyncMock(return_value="用户要求检查下载器；已收集大量日志，继续分析。")
+ monkeypatch.setattr(ContextPreservingSummarizationMiddleware, "acreate_summary", summarize)
+ result = await middleware._provider.run_task(
+ description="检查下载器。",
+ subagent_type="general-purpose",
+ )
+ assert result == "诊断完成"
+ assert summarize.await_count >= 1
diff --git a/tests/test_agent_summarization_streaming.py b/tests/test_agent_summarization_streaming.py
index 80dc3e116..d8ad4a8e4 100644
--- a/tests/test_agent_summarization_streaming.py
+++ b/tests/test_agent_summarization_streaming.py
@@ -26,6 +26,8 @@ from pydantic import Field
import app.agent.orchestrator as agent_module
from app.agent.memory import MemoryManager
from app.agent.middleware.config import RuntimeConfigMiddleware
+from app.agent.middleware.plan import PLAN_TOOL_NAME, PlanMiddleware
+from app.agent.middleware.selection import TOOL_DISCOVERY_NAME, ToolSelectorMiddleware
from app.agent.middleware.summarization import (
ContextPreservingSummarizationMiddleware,
ContextSummarizationError,
@@ -236,21 +238,14 @@ def test_streaming_agent_uses_non_streaming_llm_for_model_middlewares():
non_streaming_llm = _FakeLLM("non-streaming")
captured: dict = {}
- class _FakeToolSelectorMiddleware:
- """记录工具选择中间件初始化参数。"""
+ class _FakeToolSelectorMiddleware(ToolSelectorMiddleware):
+ """保留工具选择和发现目录，仅替换无需调用的测试模型。"""
- def __init__(
- self,
- model,
- max_tools,
- always_include=None,
- selection_tools=None,
- ):
- """保存测试断言需要的参数。"""
+ def __init__(self, **kwargs):
+ """通过真实构造函数校验启用发现后的工具与状态合同。"""
+ model = kwargs.pop("model")
+ super().__init__(model=None, **kwargs)
self.model = model
- self.max_tools = max_tools
- self.always_include = always_include or []
- self.selection_tools = selection_tools or []
def _fake_create_agent(**kwargs):
"""捕获 create_agent 参数。"""
@@ -302,11 +297,33 @@ def test_streaming_agent_uses_non_streaming_llm_for_model_middlewares():
"execute_command",
"agent_task",
"read_skill",
+ PLAN_TOOL_NAME,
+ TOOL_DISCOVERY_NAME,
]
assert tool_selector_middleware.selection_tools[: len(fake_tools)] == fake_tools
assert [getattr(tool, "name", None) for tool in tool_selector_middleware.selection_tools[len(fake_tools) :]] == [
- "read_skill"
+ "read_skill", PLAN_TOOL_NAME, TOOL_DISCOVERY_NAME,
]
+ middlewares = captured["middleware"]
+ plan_middleware = next(item for item in middlewares if isinstance(item, PlanMiddleware))
+ compaction_middleware = next(item for item in middlewares if isinstance(item, FinalRequestCompactionMiddleware))
+ assert compaction_middleware.summarizer.model is non_streaming_llm
+ assert [item.name for item in middlewares] == [
+ "AgentPolicyMiddleware",
+ "SkillsMiddleware",
+ "JobsMiddleware",
+ "RuntimeConfigMiddleware",
+ "PlanMiddleware",
+ "MemoryMiddleware",
+ "PatchToolCallsMiddleware",
+ "_FakeToolSelectorMiddleware",
+ "FinalRequestCompactionMiddleware",
+ "UsageMiddleware",
+ ]
+ policy_middleware = middlewares[0]
+ for internal_tool in [*plan_middleware.tools, *tool_selector_middleware.tools]:
+ assert policy_middleware.catalog.resolve_unique(internal_tool.name).tool is internal_tool
+ assert internal_tool in tool_selector_middleware.selection_tools
def test_non_streaming_agent_reuses_main_llm_for_summary():
diff --git a/tests/test_agent_tool_discovery.py b/tests/test_agent_tool_discovery.py
new file mode 100644
index 000000000..8761cbe77
--- /dev/null
+++ b/tests/test_agent_tool_discovery.py
@@ -0,0 +1,197 @@
+"""工具按需发现的真实图执行与目录边界回归测试。"""
+
+import asyncio
+import json
+from types import SimpleNamespace
+from typing import Any, Optional
+
+import pytest
+from langchain.agents import create_agent
+from langchain.agents.middleware.types import ModelRequest
+from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+from langchain_core.messages import AIMessage, HumanMessage
+from langchain_core.tools import StructuredTool
+from langgraph.checkpoint.memory import InMemorySaver
+from pydantic import Field, ValidationError
+
+from app.agent.middleware.selection import (
+ TOOL_DISCOVERY_DESCRIPTION_CHARS,
+ TOOL_DISCOVERY_MAX_RESULTS,
+ TOOL_DISCOVERY_NAME,
+ ToolSelectorMiddleware,
+)
+from app.agent.tools.catalog import ToolCatalogSnapshot
+
+
+class _RecordingModel(FakeMessagesListChatModel):
+ """记录实际绑定工具的离线模型，驱动真实模型和工具循环。"""
+
+ tool_history: list[list[Any]] = Field(default_factory=list)
+ invocation_count: int = 0
+
+ def bind_tools(self, tools: list[Any], **kwargs: Any) -> "_RecordingModel":
+ """记录每轮精确工具实例，便于核实发现后的完整参数契约。"""
+ self.tool_history.append(list(tools))
+ return self
+
+ def _generate(self, messages: list[Any], stop: Optional[list[str]] = None, run_manager: Any = None, **kwargs: Any) -> Any:
+ """统计真实模型调用次数，确认发现工具不会增加额外筛选模型请求。"""
+ self.invocation_count += 1
+ return super()._generate(messages, stop=stop, run_manager=run_manager, **kwargs)
+
+
+def _make_tool(name: str, description: str, tags: Optional[list[str]] = None) -> StructuredTool:
+ """创建完全离线的测试能力，携带真实结构化输入参数。"""
+ def invoke(record_id: int) -> str:
+ """用传入标识生成观察结果，验证发现后可以实际调用工具。"""
+ return f"{name}:{record_id}"
+
+ return StructuredTool.from_function(
+ func=invoke,
+ name=name,
+ description=description,
+ tags=tags or [],
+ )
+
+
+def _tool_call(name: str, arguments: dict[str, Any], call_id: str) -> dict[str, Any]:
+ """构造模型发出的标准工具调用。"""
+ return {"name": name, "args": arguments, "id": call_id, "type": "tool_call"}
+
+
+def test_discovery_enables_exact_schemas_merges_parallel_calls_and_resets_turns():
+ """真实图并行发现遗漏能力后可调用，且同会话新请求与其他会话均重新筛选。"""
+ status = _make_tool("status", "运行状态")
+ transfer = _make_tool("transfer_history", "查询整理历史", ["transfer"])
+ subscribe = _make_tool("subscribe_media", "管理媒体订阅", ["subscription"])
+ hidden = _make_tool("delete_library", "删除媒体库")
+ selection_model = _RecordingModel(responses=[AIMessage(content='{"tools": []}')])
+ model = _RecordingModel(responses=[
+ AIMessage(content="", tool_calls=[
+ _tool_call(TOOL_DISCOVERY_NAME, {"search": "transfer_history", "limit": 1}, "find-transfer"),
+ _tool_call(TOOL_DISCOVERY_NAME, {"search": "subscribe_media", "limit": 1}, "find-subscribe"),
+ ]),
+ AIMessage(content="", tool_calls=[
+ _tool_call("transfer_history", {"record_id": 42}, "read-history"),
+ ]),
+ AIMessage(content="已查询整理历史"),
+ AIMessage(content="新请求"),
+ AIMessage(content="其他会话"),
+ ])
+ original_tools = [status, transfer, subscribe, hidden]
+ middleware = ToolSelectorMiddleware(
+ model=selection_model,
+ selection_tools=original_tools,
+ always_include=["status"],
+ max_tools=1,
+ enable_discovery=True,
+ )
+ graph = create_agent(
+ model=model,
+ tools=original_tools,
+ middleware=[middleware],
+ checkpointer=InMemorySaver(),
+ )
+
+ async def execute() -> None:
+ """串行执行多个用户请求并读取持久化图状态。"""
+ first_config = {"configurable": {"thread_id": "first-session"}}
+ result = await graph.ainvoke({"messages": [HumanMessage(content="检查系统")]}, first_config)
+ assert any(message.content == "transfer_history:42" for message in result["messages"])
+ state = await graph.aget_state(first_config)
+ assert set(state.values["discovered_tool_names"]) == {"transfer_history", "subscribe_media"}
+ await graph.ainvoke({"messages": [HumanMessage(content="新的请求")]}, first_config)
+ state = await graph.aget_state(first_config)
+ assert state.values["discovered_tool_names"] == []
+ await graph.ainvoke(
+ {"messages": [HumanMessage(content="其他会话")]},
+ {"configurable": {"thread_id": "second-session"}},
+ )
+
+ asyncio.run(execute())
+
+ bound_names = [{tool.name for tool in tools} for tools in model.tool_history]
+ assert bound_names[0] == {"status", TOOL_DISCOVERY_NAME}
+ assert bound_names[1] == {"status", TOOL_DISCOVERY_NAME, "transfer_history", "subscribe_media"}
+ assert bound_names[2] == bound_names[1]
+ assert bound_names[3] == bound_names[4] == bound_names[0]
+ assert transfer in model.tool_history[1]
+ assert transfer.tool_call_schema.model_json_schema()["properties"]["record_id"]["type"] == "integer"
+ assert "delete_library" not in set.union(*bound_names)
+ assert selection_model.invocation_count == 3
+ assert original_tools == [status, transfer, subscribe, hidden]
+
+
+def test_discovery_catalog_is_bounded_and_matches_name_description_and_tags():
+ """发现仅返回已授权目录的有界说明，并支持名称、中文说明和能力标签检索。"""
+ exact = _make_tool("transfer_history", "查询失败整理历史", ["transfer"])
+ subscriptions = [
+ _make_tool(f"subscription_{index}", "媒体订阅" * 300, ["subscription"])
+ for index in range(12)
+ ]
+ middleware = ToolSelectorMiddleware(selection_tools=[*subscriptions, exact], enable_discovery=True)
+ discovery = middleware.tools[0]
+
+ async def search(query: str, limit: int = TOOL_DISCOVERY_MAX_RESULTS) -> Any:
+ """直接调用已注册工具以覆盖参数校验和状态增量结果。"""
+ return await discovery.ainvoke({
+ "search": query,
+ "limit": limit,
+ "runtime": SimpleNamespace(tool_call_id="discovery"),
+ })
+
+ result = asyncio.run(search("subscription"))
+ payload = json.loads(result.update["messages"][0].content)
+ assert len(payload["tools"]) == TOOL_DISCOVERY_MAX_RESULTS
+ assert len(result.update["discovered_tool_names"]) == TOOL_DISCOVERY_MAX_RESULTS
+ assert all(len(tool["description"]) <= TOOL_DISCOVERY_DESCRIPTION_CHARS for tool in payload["tools"])
+ assert asyncio.run(search("TRANSFER_HISTORY", 1)).update["discovered_tool_names"] == ["transfer_history"]
+ assert asyncio.run(search("失败整理", 1)).update["discovered_tool_names"] == ["transfer_history"]
+ assert asyncio.run(search("transfer", 1)).update["discovered_tool_names"] == ["transfer_history"]
+ assert asyncio.run(search("unavailable_secret_tool")).update["discovered_tool_names"] == []
+ assert asyncio.run(search(" !!! ")).update["discovered_tool_names"] == []
+
+
+@pytest.mark.parametrize("arguments", [{"search": ""}, {"search": "x" * 161}, {"search": "media", "limit": 9}])
+def test_discovery_rejects_unbounded_queries(arguments: dict[str, Any]):
+ """超界请求被明确拒绝，避免模型通过参数扩大目录输出。"""
+ middleware = ToolSelectorMiddleware(enable_discovery=True)
+ with pytest.raises(ValidationError):
+ middleware.tools[0].args_schema.model_validate(arguments)
+
+
+def test_discovery_preserves_provider_and_mandatory_tools_without_fabricating_unknown_tools():
+ """按需启用与首轮工具上限独立，保留强制工具和 provider 字典且拒绝不存在的名称。"""
+ selected = _make_tool("status", "运行状态")
+ mandatory = _make_tool("read_skill", "读取技能")
+ discovered = _make_tool("transfer_history", "整理历史")
+ provider = {"type": "web_search_preview"}
+ middleware = ToolSelectorMiddleware(
+ selection_tools=[selected, mandatory, discovered],
+ always_include=["read_skill"],
+ max_tools=1,
+ enable_discovery=True,
+ )
+ request = ModelRequest(
+ model=_RecordingModel(responses=[AIMessage(content="完成")]),
+ tools=[selected, mandatory, discovered, *middleware.tools, provider],
+ messages=[HumanMessage(content="查询记录")],
+ state={"selected_tool_names": ["status"], "discovered_tool_names": ["transfer_history", "fabricated"]},
+ runtime=None,
+ )
+
+ async def handler(updated: ModelRequest) -> ModelRequest:
+ """捕获最终模型绑定请求。"""
+ return updated
+
+ result = asyncio.run(middleware.awrap_model_call(request, handler))
+ assert result.tools == [selected, discovered, mandatory, middleware.tools[0], provider]
+
+
+def test_discovery_has_serializable_private_catalog_identity_and_injected_runtime():
+ """严格目录可以稳定签名发现工具，模型参数中不暴露宿主运行时状态。"""
+ middleware = ToolSelectorMiddleware(selection_tools=[], enable_discovery=True)
+ catalog = ToolCatalogSnapshot.from_tools(middleware.tools, plugin_revision=1, factory_revision="test")
+ assert catalog.require_unique() is catalog
+ assert catalog.entries[0].source == "middleware:selection"
+ assert set(middleware.tools[0].tool_call_schema.model_json_schema()["properties"]) == {"search", "limit"}