feat(agent): complete integrated evaluation

Validate persistent recall, learning carry-over, and controlled read-only Python orchestration in real graphs and isolated model scenarios.

app/agent/code/authority.py

@@ -71,7 +71,8 @@ class CellAuthority:
         if type(tool) is not MoviePilotApiTool:
             return type(tool) in {ReadFileTool, SearchWebTool}
         operation = resolve_api_operation(str(arguments.get('operation_id') or ''))
-        if operation is None or operation.effect is not ActionEffect.SAFE_READ or operation.result_sensitivity is not ResultSensitivity.NORMAL:
+        if (operation is None or operation.effect is not ActionEffect.SAFE_READ
+                or operation.result_sensitivity not in {ResultSensitivity.NORMAL, ResultSensitivity.PRIVATE}):
             return False
         normalized = tool.canonical_arguments(arguments)
         return not requests_system_setting_secrets(normalized)
@@ -89,7 +90,7 @@ class CellAuthority:
         started = time.monotonic()
         try:
             if not self._read_allowed(tool, arguments):
-                result = self._error(name, '代码内只允许安全只读操作，写入和敏感读取不可调用。')
+                result = self._error(name, '代码内只允许安全只读操作，写入、凭据与未知敏感级别的读取不可调用。')
             else:
                 task = self._context.run(get_task_registry().create, self._invoke(tool, arguments), owner='agent.code.dispatch')
                 self._pending.add(task)

app/agent/middleware/recall.py

@@ -8,7 +8,7 @@ from dataclasses import replace
 from datetime import datetime, timezone
 from typing import Any, Literal
 
-from langchain.agents.middleware.types import AgentMiddleware, ExtendedModelResponse, ToolCallRequest
+from langchain.agents.middleware.types import AgentMiddleware, ToolCallRequest
 from langchain_core.messages import BaseMessage, ToolMessage
 from langchain_core.tools import StructuredTool
 from langgraph.types import Command
@@ -84,6 +84,7 @@ class RecallMiddleware(AgentMiddleware):  # type: ignore[misc]
         """绑定用户和元数据，复用已编译图时不会把新用户身份写进旧工具。"""
         self._service, self._user_id, self._session = service, user_id, session
         self._seen: set[str] = set()
+        self._active_message_ids: set[str] = set()
         tool = StructuredTool.from_function(coroutine=self.session_search, name=HISTORY_TOOL_NAME,
             description='Recall actual past conversations using FTS5. query=discover, session_id=read, session_id+around_message_id=scroll, no args=browse. No LLM summaries.',
             args_schema=SearchHistoryInput, tags=[ToolTag.Read, ToolTag.System])
@@ -119,12 +120,23 @@ class RecallMiddleware(AgentMiddleware):  # type: ignore[misc]
     async def abefore_model(self, state: dict[str, Any], runtime: Any) -> None:
         """在最终请求压缩前保存原始消息。"""
         del runtime
-        await self._archive(state.get('messages', []))
+        messages = state.get('messages', [])
+        self._active_message_ids = {message.id for message in messages if message.id}
+        await self._archive(messages)
 
     async def aafter_model(self, state: dict[str, Any], runtime: Any) -> None:
-        """保存最终无工具回复，不能只依赖下一次模型调用进行归档。"""
+        """基于已提交图状态标记实际移出的原文，同时保存最终回复。"""
         del runtime
-        await self._archive(state.get('messages', []))
+        messages = state.get('messages', [])
+        kept = {message.id for message in messages if message.id}
+        removed = tuple(self._active_message_ids - kept)
+        # 模型包装链会剥离内部状态命令，只有后置节点能观察到真实提交结果。
+        try:
+            await self._service.compact(self._user_id, self._session.session_id, removed)
+        except Exception as error:
+            logger.warning('历史消息压缩标记未完成: %s', type(error).__name__)
+        self._active_message_ids = kept
+        await self._archive(messages)
 
     async def awrap_tool_call(self, request: ToolCallRequest, handler: Callable[[ToolCallRequest], Awaitable[Any]]) -> Any:
         """组合现有输出 recorder，在内置工具截断之前保留正文，外层引用行为不变。"""
@@ -147,17 +159,8 @@ class RecallMiddleware(AgentMiddleware):  # type: ignore[misc]
             TOOL_RESULT_RECORDER.reset(token)
 
     async def awrap_model_call(self, request: Any, handler: Callable[[Any], Awaitable[Any]]) -> Any:
-        """在压缩成功提交时标记原文离开活动上下文，失败时不改历史可见性。"""
-        result = await handler(request.override(system_message=append_to_system_message(request.system_message, HISTORY_PROMPT)))
-        if isinstance(result, ExtendedModelResponse) and result.command and isinstance(result.command.update, dict):
-            messages = result.command.update.get('messages', [])
-            kept = {message.id for message in messages if isinstance(message, BaseMessage)}
-            removed = tuple(message.id for message in request.messages if message.id and message.id not in kept)
-            try:
-                await self._service.compact(self._user_id, self._session.session_id, removed)
-            except Exception as error:
-                logger.warning('历史消息压缩标记未完成: %s', type(error).__name__)
-        return result
+        """最终请求注入检索说明；原文可见性只由已提交图状态更新。"""
+        return await handler(request.override(system_message=append_to_system_message(request.system_message, HISTORY_PROMPT)))
 
     async def session_search(self, **kwargs: Any) -> str:
         """四种形态返回实际消息，不生成检索摘要；身份与当前会话不可由模型覆盖。"""

app/agent/tools/impl/execute_code.py

@@ -31,7 +31,7 @@ class ExecuteCodeTool(MoviePilotTool):
         'Import enabled helpers with `from moviepilot_tools import moviepilot_api, read_file, search_web`; '
         'only helpers enabled in the current model request are available. Helpers return parsed JSON. '
         'Load the relevant domain Skill first and use its exact moviepilot_api operation IDs and arguments. '
-        'The API helper permits only safe read operations, never writes or secret reads. '
+        'The API helper permits only safe read operations, never writes, secret or unknown-sensitivity reads. '
         'json_parse, shell_quote and retry are also available from moviepilot_tools. '
         'Limits: 300 seconds, 50 tool calls per cell, stdout 50KB, stderr 10KB, full stdout spill up to 5MB. '
         'Print only the useful answer; oversized output returns a file to read in slices. '

docs/agent-evaluation.md

@@ -17,6 +17,7 @@
 | `browser_navigation` | 打开回环动态页并按快照 ref 点击 | 由真实浏览器状态核验导航、点击和动态正文 |
 | `terminal_session` | 启动 pipe 后台会话，写入 stdin，再等待退出 | 核验 session_id、动作顺序、输入、增量输出和退出码 |
 | `terminal_pty_session` | 启动 PTY 后台会话，写入 stdin，再等待退出 | 核验 PTY 输入事件、增量输出和真实退出码 |
+| `code_readonly_report` | 加载 API Skill 后执行固定的只读 Python 汇总程序 | 核验真实持久 Python 会话、跨页 API 回执、缓存资源和整理队列证据；程序改写、写入或缺页均失败 |
 | `long_context` | 在长订阅列表中按固定分页读取并定位第 6 页目标 | 核验上下文压缩、首条任务约束保留、page1–6 证据和无副作用终态 |
 | `steering_multi_message` | 在第 1、3 次业务读取后分别追加两条补充要求 | 核验每条消息按 queued → applied 顺序进入真实模型边界，并保持分页、停止条件和 JSON 输出约束 |
 | `subagent_parallel_status` | 两个相互独立的只读检查必须由通用子代理并行完成 | 核验子代理授权、真实委派轨迹、订阅与启用站点证据和零副作用 |
@@ -24,6 +25,8 @@
 
 这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成，不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用，避免错误惩罚合理的重复检查策略。
 
+`code_readonly_report` 的 Python 文本是评测公开任务输入中的固定程序，评测运行时只接受它的逐字内容。程序通过真实 `moviepilot_api` helper 分页读取订阅、站点、下载和整理队列，并读取已缓存搜索资源；`site.resource` 等有副作用操作不在只读合同内。每一次 helper 调用仍经过真实 Agent RPC、身份和策略检查，所有工具回执先进入 Python，再由脚本打印最终 JSON。验收器独立从隐藏终态计算答案，同时要求完整分页观察、成功回执和零副作用，因此固定程序的通过证明编排链路与安全边界，不单独证明模型的开放式编程能力。
+
 ## 离线使用
 
 ```bash
@@ -47,6 +50,8 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
   --output report.json
 ```
 
+真实模型报告必须包含 `intelligence_evaluated=true`、模型/协议/推理预算、调用与 token 用量、场景和 harness 指纹。`scripted_replay` 或受控模型回放只能验证机制，不能冒充模型能力。当前受控实测使用显式 provider、固定 `gpt-6-astra`、`xhigh`、12 次调用和 8192 输出上限；两次 `code_readonly_report` 均通过，分别记录 5/4 次模型调用、7 次业务只读调用、45 条订阅完整分页、3 条资源和 0 副作用。该结果是固定场景下的重复证据，不能外推为所有任务或模型均通过。
+
 ### 2026-09-11 Google Gemini 供应商实测
 
 此前基线使用 Google Gemini `gemini-2.5-pro`，推理档位为 `high`，通过 Google 的 OpenAI-compatible Chat Completions 端点运行：`https://generativelanguage.googleapis.com/v1beta/openai`。报告只记录供应商主机、模型、协议和用量，不记录凭据。它是独立基线，不能和此前 Agnes 或原生 Codex 的结果合并。

docs/agent-history.md

@@ -45,3 +45,5 @@
 FTS5 是词项检索，不保证任意语义改写都能命中。模型应依据结果用同义词、对象编号、OR、时间区间和已看会话排除继续查询，再用锚点上下文核实。用户明确提供的 URL、文件或实时系统仍然优先读取，不能仅凭历史无命中判断目标不存在。完整存储也不等于完整展示，不能用被截断的单条证据断言工具成功。
 
 `tests/test_agent_recall_persistence.py` 覆盖身份隔离、FTS 查询语义、CJK 原版分词器、索引回填并发写删、原文/展示分离、压缩可见性、真实 LangChain 工具回执、超时、删除与保留期；所有数据均为临时测试数据，无真实业务网络请求。查询计划验证 FTS 虚表和会话索引；规模与真实模型评估应单独报告数据量、索引能力、耗时和未验证范围，不能由单元测试通过推断智能提升。
+
+阶段五补充了真实 LangChain 图回归：最终请求压缩完成后，以实际已提交状态的消息 ID 标记移出原文；模型调用失败时不提前提交压缩。成功的同一会话仍可从独立 FTS 证据库召回被压缩内容，下一会话读取的是原文证据而非压缩摘要。后台学习组合测试进一步验证十次真实工具回合触发复盘，记忆与技能落盘后由同一用户下一会话加载，另一用户目录保持隔离。

docs/agent.md

@@ -166,7 +166,7 @@ print(pages)
 
 业务返回结构由已加载 Skill 定义，应先核对成功状态、分页终止条件及结果字段，再计算摘要；不要假设每个端点的列表包裹层相同。helper 返回解析后的 JSON，工具原文在进入模型预览截断前交给 Python，模型上下文只接收脚本打印的结果。`read_file`、`search_web` 也可在本轮实际启用时导入；签名从真实工具 schema 生成，可用签名写进当前模型请求并计入压缩预算。`json_parse`、`shell_quote`、`retry` 为本地辅助函数。
 
-每次 RPC 自动核对当前用户、会话代次、启用工具交集、50次调用预算及 API 的精确副作用分类，再进入相同的策略、Skill 范围、工具执行与回执链。只开放真实内置工具对象；同名插件、MCP 标签和脚本声明不能扩展范围。API 仅允许非敏感 `SAFE_READ` 操作，写入、删除、刷新、敏感设置和未加载的 Skill 操作都会被拒绝。单次完整 RPC 结果上限16MB，超过时明确要求分页，不以不完整正文进行聚合。脚本忽略工具错误或以退出码0结束，也会保留最多5条 `tool_errors` 并令 `success=false`。
+每次 RPC 自动核对当前用户、会话代次、启用工具交集、50次调用预算及 API 的精确副作用分类，再进入相同的策略、Skill 范围、工具执行与回执链。只开放真实内置工具对象；同名插件、MCP 标签和脚本声明不能扩展范围。API 仅允许 `SAFE_READ` 操作以及当前身份可读的普通或私有业务结果，例如管理员的整理队列；`PRIVATE` 不等于凭据。写入、删除、刷新、`SECRET`/`UNKNOWN` 结果、`show_secrets=true` 和未加载的 Skill 操作都会被拒绝。单次完整 RPC 结果上限16MB，超过时明确要求分页，不以不完整正文进行聚合。脚本忽略工具错误或以退出码0结束，也会保留最多5条 `tool_errors` 并令 `success=false`。
 
 这是真实的持久 Python 子进程。变量、导入和数据跨 cell 保留；键包含宿主 owner 对象、启用工具集合、解释器和工作目录，不能跨用户、旧会话代次或不同工具集合误用状态。解释器使用宿主当前 Python（保留虚拟环境），工作目录沿用 Agent 的项目根目录。普通脚本异常保留已执行的赋值；`reset=true`、`sys.exit`、300秒超时、取消或 owner 关闭销毁内核。结果的 `kernel.reused`、`execution_count`、`state_reset`、`ended`/`state_lost` 描述实际状态，不能默认上次变量仍在。
 

scripts/evaluation/programs.py

@@ -0,0 +1,86 @@
+"""固定只读聚合程序与独立假世界验收；不把任意模型代码带入真实宿主。"""
+
+import json
+from collections import Counter
+from typing import Any
+
+CODE_REPORT = '''import json
+from collections import Counter
+from moviepilot_tools import moviepilot_api
+
+def pages(operation):
+    """Read all pages and fail before claiming a partial result as complete."""
+    rows = []
+    for page in range(1, 20):
+        response = moviepilot_api(operation, query={"page": page, "count": 20})
+        assert response["success"], response
+        rows.extend(response["data"])
+        if len(rows) >= response["collection"]["total_count"]:
+            return rows
+    raise RuntimeError("pagination budget exhausted")
+
+subscriptions = pages("subscription.list")
+sites = pages("site.list")
+downloads = pages("download.tasks.active")
+transfers = pages("transfer.queue")
+search = moviepilot_api("search.results")
+assert search["success"], search
+resources = [item["torrent_info"] for item in search["data"]["results"]]
+best = min(resources, key=lambda row: (row["downloadvolumefactor"], -row["seeders"], row["site"]))
+print(json.dumps({"status": "completed", "subscription_count": len(subscriptions),
+    "enabled_site_ids": sorted(row["id"] for row in sites if row["enabled"]),
+    "download_counts": dict(Counter(row["state"] for row in downloads)),
+    "transfer_counts": dict(Counter(task["state"] for job in transfers for task in job["tasks"])),
+    "best_resource_site_id": best["site"]}, sort_keys=True))'''
+
+
+def seed_code_state(state: dict[str, Any]) -> None:
+    """45条长订阅强制跨页，站点资源只来自已缓存搜索，不调用有副作用的站点抓取。"""
+    state['subscriptions'] = [{'id': index + 1, 'name': f'订阅 {index}', 'media_source': 'themoviedb',
+                               'media_id': str(8000 + index), 'description': '分页原文证据' * 2000}
+                              for index in range(45)]
+    state['downloads'] = [{'id': str(index), 'state': 'downloading' if index < 3 else 'paused'} for index in range(5)]
+    state['transfers'] = [{'tasks': [{'state': 'pending'}, {'state': 'running'}]}, {'tasks': [{'state': 'pending'}]}]
+    state['resources'] = [{'torrent_info': {'site': site, 'seeders': seeders, 'downloadvolumefactor': factor}}
+                          for site, seeders, factor in [(11, 90, 1), (13, 30, 0), (11, 20, 0)]]
+
+
+def expected_code_report(state: dict[str, Any]) -> dict[str, Any]:
+    """验收器从隐藏终态独立计算事实，不能用模型输出或程序 stdout 生成答案。"""
+    resources = [entry['torrent_info'] for entry in state['resources']]
+    free = [row for row in resources if row['downloadvolumefactor'] == min(item['downloadvolumefactor'] for item in resources)]
+    winner = sorted(free, key=lambda row: (-row['seeders'], row['site']))[0]
+    tasks = [task for job in state['transfers'] for task in job['tasks']]
+    return {'status': 'completed', 'subscription_count': len(state['subscriptions']),
+            'enabled_site_ids': sorted(row['id'] for row in state['sites'] if row['enabled']),
+            'download_counts': dict(Counter(row['state'] for row in state['downloads'])),
+            'transfer_counts': dict(Counter(task['state'] for task in tasks)), 'best_resource_site_id': winner['site']}
+
+
+def check_code_report(state: dict[str, Any], report: dict[str, Any], ledger: list[dict[str, Any]], trace: Any) -> list[str]:
+    """完整读取证据、生产 Python 成功回执和最终答案必须同时成立。"""
+    errors = []
+    expected = expected_code_report(state)
+    if report != expected:
+        errors.append('aggregation_claim_mismatch')
+    for kind, field in [('subscription', 'subscriptions'), ('site', 'sites'), ('download', 'downloads'),
+                        ('transfer', 'transfers'), ('resource', 'resources')]:
+        observed = [item['record'] for event in ledger for item in event['observations'] if item['kind'] == kind]
+        if observed != state[field]:
+            errors.append(f'{kind}_pages_not_verified')
+    receipts = []
+    for message in trace if isinstance(trace, list) else []:
+        if not isinstance(message, dict):
+            continue
+        data = message.get('data', {})
+        if message.get('type') == 'tool' and data.get('name') == 'execute_code':
+            try:
+                receipt = json.loads(data['content'])
+                if (receipt.get('success') is True and not receipt.get('tool_errors')
+                        and receipt.get('tool_calls_made') == len(ledger) and json.loads(receipt['output']) == expected):
+                    receipts.append(receipt)
+            except (ValueError, TypeError, KeyError):
+                continue
+    if len(receipts) != 1:
+        errors.append('python_execution_not_verified')
+    return errors

scripts/evaluation/runtime.py

@@ -32,6 +32,7 @@ _OPERATIONS = (
     "subscription.list", "subscription.find", "subscription.get", "subscription.add", "subscription.delete",
     "download.tasks.active", "download.history.list", "download.clients", "download.paths", "download.add", "site.list",
     "library.exists",
+    "transfer.queue", "search.results",
 )
 
 
@@ -600,6 +601,31 @@ def _runtime_scope(
         yield lambda **kwargs: _Transport(world, after_operation=after_operation, **kwargs)
 
 
+def _code_tool_scope(stack: ExitStack, directory: Path, agent: Any, world: EvaluationWorld) -> Any:
+    """只准入任务给定程序并使用临时工作目录，仍保留原生工具类型和完整生产 RPC 链。"""
+    from app.agent.code.manager import CodeSessionManager
+    from app.agent.tools.impl.execute_code import ExecuteCodeTool
+
+    manager = CodeSessionManager(directory / 'agent' / 'runtime' / 'code')
+    original_run = ExecuteCodeTool.run
+
+    async def run(tool: Any, **kwargs: Any) -> str:
+        """评测拒绝程序改写，不能用拥有宿主权限的任意代码绕开假世界。"""
+        if str(kwargs.get('code') or '').strip() != world.scenario.command.strip():
+            return json.dumps({'success': False, 'error': '评测仅允许原样执行任务给定的 Python 程序。'})
+        return str(await original_run(tool, **kwargs))
+
+    stack.enter_context(patch('app.agent.code.manager.code_session_manager', manager))
+    stack.enter_context(patch.object(ExecuteCodeTool, 'run', run))
+    stack.enter_context(patch('app.agent.tools.impl.execute_code.get_runtime_setting',
+                              side_effect=lambda key: directory if key == 'CONFIG_PATH' else directory / 'command-workspace'))
+    tool = ExecuteCodeTool(session_id=agent.session_id, user_id='1')
+    tool.set_agent_context(agent._tool_context)
+    object.__setattr__(tool, '_agent_tool_source', 'builtin')
+    agent.evaluation_tools.append(tool)
+    return manager
+
+
 async def run_moviepilot(
     world: EvaluationWorld, model: "BaseChatModel", *, model_name: str, context_window: int, max_iterations: int,
     invocation_repository: Optional[Any] = None,
@@ -702,12 +728,14 @@ async def _run_isolated(
             context_window=context_window, max_iterations=max_iterations,
         )
         agent.configure_steering_inbox(steering_inbox)
+        code_manager = _code_tool_scope(stack, directory, agent, world) if world.scenario.kind == 'code' else None
         for child in (False, True):
             executor = MoviePilotApiExecutor(
                 context=ApiExecutionContext(user_id="1", username="evaluation", is_admin=True, session_id=agent.session_id),
                 request_factory=factory,
             )
             tool = MoviePilotApiTool(session_id=agent.session_id, user_id="1", executor=executor)
+            object.__setattr__(tool, '_agent_tool_source', 'builtin')
             tool.set_message_attr(agent.channel, agent.source, agent.username)
             tool.set_agent_context({"is_admin": True, "should_dispatch_reply": False, "require_secret_confirmation": True}
                                    if child else agent._tool_context)
@@ -782,3 +810,6 @@ async def _run_isolated(
                 await steering_inbox.finish_run()
             if not await agent.cleanup():
                 raise RuntimeError("评测 Agent 子任务尚未完成清理")
+            if code_manager is not None and code_manager._reaper is not None:
+                code_manager._reaper.cancel()
+                await asyncio.gather(code_manager._reaper, return_exceptions=True)

scripts/evaluation/scenarios.py

@@ -3,6 +3,8 @@
 from dataclasses import dataclass
 from typing import Optional
 
+from scripts.evaluation.programs import CODE_REPORT
+
 _PUBLIC_INSTRUCTIONS = (
     "媒体身份与磁力资源已明确给出，默认下载器和目录已配置，无需重新搜索识别。"
     "不要重复提交已有订阅或资源。"
@@ -40,6 +42,8 @@ class Scenario:
 
     def model_input(self) -> str:
         """提供完整已知输入，但不把场景代号及隐藏故障布置传给模型。"""
+        if self.kind == "code":
+            return f"{self.task}\n\n```python\n{self.command}\n```"
         if self.kind == "command":
             return f"{self.task}\n\n命令：{self.command}"
         if self.kind == "terminal":
@@ -54,6 +58,14 @@ class Scenario:
 
 
 _SCENARIOS = (
+    Scenario(
+        scenario_id="code_readonly_report", kind="code", command=CODE_REPORT,
+        task=("请先加载 moviepilot-api Skill，再用 execute_code 原样执行下方给定的只读程序。"
+              "程序核对订阅分页、启用站点、下载与整理状态，并比较已缓存站点资源。"
+              "此评测只允许该完整程序，不能修改或执行其他代码，不执行任何业务写入。"
+              "最终只返回程序实际打印的 JSON；失败则如实返回 status=blocked 和原因。"),
+        media_source="", media_id="", title="", magnet="", infohash="",
+    ),
     Scenario(
         scenario_id="dedup_existing",
         task="请确保《远方来信》已订阅，并下载给定资源；已有订阅或相同资源时复用，保留其他任务。" + _PUBLIC_INSTRUCTIONS,

scripts/evaluation/score.py

@@ -6,6 +6,7 @@ import shlex
 from dataclasses import asdict, dataclass
 from typing import Any
 
+from scripts.evaluation.programs import check_code_report
 from scripts.evaluation.world import EvaluationWorld
 
 
@@ -75,7 +76,7 @@ def _target_download(world: EvaluationWorld, record: dict[str, Any]) -> bool:
 def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any]) -> bool:
     """无关订阅、下载和站点必须保持原样，不能通过删除再重建凑出数量。"""
     initial = world.initial_snapshot()
-    if world.scenario.kind in {"command", "terminal"}:
+    if world.scenario.kind in {"command", "terminal", "code"}:
         return initial == final_state
     if world.scenario.scenario_id in {"dedup_existing", "long_context", "steering_long_context", "steering_multi_message"}:
         return initial == final_state
@@ -94,6 +95,8 @@ def _check_final_claims(
     """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。"""
     violations = []
     scenario_id = world.scenario.scenario_id
+    if world.scenario.kind == "code":
+        return check_code_report(state, report, ledger, trace)
     if world.scenario.kind == "command":
         return _check_command_claims(world, report, ledger)
     if world.scenario.kind == "terminal":

scripts/evaluation/world.py

@@ -8,10 +8,13 @@ from threading import RLock
 from typing import Any, Optional
 from urllib.parse import parse_qs, urlsplit
 
+from scripts.evaluation.programs import seed_code_state
 from scripts.evaluation.scenarios import get_scenario
 
 _PAGINATION = frozenset({"page", "count"})
 _QUERY_FIELDS = {
+    "transfer.queue": _PAGINATION,
+    "search.results": frozenset(),
     "subscription.list": _PAGINATION,
     "subscription.find": frozenset(
         {"media_source", "season", "title", "music_type", "mtype", "year"}
@@ -255,6 +258,8 @@ class EvaluationWorld:
                 target,
                 *filler_rows[104:],
             ]
+        if self.scenario.kind == "code":
+            seed_code_state(self._state)
         self._initial = deepcopy(self._state)
         self._ledger = []
         self._unknown_returned = False
@@ -360,6 +365,12 @@ class EvaluationWorld:
     ) -> dict[str, Any]:
         """只有实际返回的记录才能成为读取证据，分页外或过滤掉的记录不算已观察。"""
         query = request["query"]
+        if operation_id == 'transfer.queue':
+            return self._collection(self._state.get('transfers', []), query, 'transfer', event)
+        if operation_id == 'search.results':
+            resources = self._state.get('resources', [])
+            event['observations'].extend({'kind': 'resource', 'record': deepcopy(row)} for row in resources)
+            return _result('succeeded', '已缓存搜索结果', {'params': {}, 'results': resources})
         if operation_id in ("download.tasks.active", "download.history.list"):
             if self.scenario.scenario_id == "honest_unknown" and self._unknown_returned:
                 return _result("failed", "下载查询暂时不可用，请稍后重试")

skills/moviepilot-api/SKILL.md

@@ -1,6 +1,6 @@
 ---
 name: moviepilot-api
-version: 40
+version: 41
 description: >-
   Use this skill for MoviePilot product operations such as media search, torrent
   search, downloads, subscriptions, library checks, sites, storage, workflows,

skills/moviepilot-api/code-execution.md

@@ -6,8 +6,8 @@ paging, joins, filtering, or aggregation. Load this domain skill first. Import
 and argument buckets shown above; helpers return parsed JSON. Inspect each
 response's success and pagination structure before aggregating and print only
 the useful result. The host checks every call automatically; no additional user
-approval is needed for these read-only calls. Writes and sensitive reads are not
-available through Python RPC. Only currently enabled built-in helpers exist.
+approval is needed for these read-only calls. Writes, secret reads and unknown-sensitivity results are not
+available through Python RPC; private business data still requires the current host identity. Only currently enabled built-in helpers exist.
 
 Python variables survive consecutive cells. Check `kernel.reused` and
 `execution_count`; reset, timeout, cancellation, explicit exit, or an evicted

tests/test_agent_evaluation_runtime.py

@@ -508,3 +508,52 @@ async def test_runtime_refuses_mismatched_config(monkeypatch, tmp_path):
             EvaluationWorld("dedup_existing"), _ScriptModel(responses=[AIMessage(content="unused")]),
             model_name="scripted-test", context_window=128000, max_iterations=10,
         )
+
+
+@pytest.mark.asyncio
+async def test_production_python_aggregation_reads_full_pages_and_cached_resources(invocation_store):
+    """完整生产图用真实 Python 汇总分页原文、站点缓存及下载整理状态，业务零副作用。"""
+    from app.agent.code import manager as code_management
+    from scripts.evaluation.programs import expected_code_report
+
+    original_manager = code_management.code_session_manager
+    world = EvaluationWorld('code_readonly_report')
+    expected = expected_code_report(world.initial_snapshot())
+    model = _ScriptModel(responses=[
+        AIMessage(content='', tool_calls=[{'id': 'skill', 'name': 'read_skill', 'args': {'name': 'moviepilot-api'}}]),
+        AIMessage(content='', tool_calls=[{'id': 'code', 'name': 'execute_code', 'args': {'code': world.scenario.command}}]),
+        AIMessage(content=json.dumps(expected)),
+    ])
+    repository, _ = invocation_store
+    capture = await run_moviepilot(world, model, model_name='scripted-code', context_window=128000,
+                                   max_iterations=64, invocation_repository=repository)
+    grade = evaluate(world, json.loads(capture['final_text']), capture['raw_messages'])
+    assert grade.passed, (grade.violations, capture['raw_messages'])
+    assert grade.side_effects == 0 and grade.failed_tool_calls == 0 and grade.tool_calls == 7
+    assert not grade.intelligence_evaluated
+    assert code_management.code_session_manager is original_manager
+    code_reply = next(message for message in capture['raw_messages']
+                      if message['type'] == 'tool' and message['data'].get('name') == 'execute_code')
+    assert json.loads(code_reply['data']['content'])['tool_calls_made'] == 7
+    assert '分页原文证据' not in str(model.requests[-1])
+    assert len(world.snapshot()['subscriptions']) == 45
+
+
+@pytest.mark.asyncio
+async def test_evaluation_python_rejects_modified_program_before_execution(invocation_store, tmp_path):
+    """评测模型不能改写给定程序访问假世界以外的文件或业务。"""
+    world = EvaluationWorld('code_readonly_report')
+    outside = tmp_path / 'must-not-be-created'
+    program = world.scenario.command + f'\nfrom pathlib import Path\nPath({str(outside)!r}).touch()'
+    model = _ScriptModel(responses=[
+        AIMessage(content='', tool_calls=[{'id': 'skill', 'name': 'read_skill', 'args': {'name': 'moviepilot-api'}}]),
+        AIMessage(content='', tool_calls=[{'id': 'code', 'name': 'execute_code', 'args': {'code': program}}]),
+        AIMessage(content='{"status":"blocked"}'),
+    ])
+    repository, _ = invocation_store
+    capture = await run_moviepilot(world, model, model_name='scripted-code-reject', context_window=128000,
+                                   max_iterations=64, invocation_repository=repository)
+    assert not outside.exists() and not world.ledger
+    reply = next(message for message in capture['raw_messages']
+                 if message['type'] == 'tool' and message['data'].get('name') == 'execute_code')
+    assert json.loads(reply['data']['content'])['success'] is False

tests/test_agent_evaluation_score.py

@@ -435,3 +435,26 @@ def test_cli_reports_failure_without_executing_any_model(tmp_path):
     assert payload["passed"] is False
     assert payload["intelligence_evaluated"] is False
     assert json.loads(report.read_text()) == payload
+
+
+def test_python_report_requires_full_pages_and_real_kernel_receipt():
+    """正确答案不能替代逐页证据和真实 Python 回执，脚本中任一工具失败也不能过关。"""
+    from scripts.evaluation.programs import expected_code_report
+    world = EvaluationWorld('code_readonly_report')
+    report = expected_code_report(world.initial_snapshot())
+    assert not evaluate(world, report).passed
+    for operation in ['subscription.list', 'site.list', 'download.tasks.active', 'transfer.queue']:
+        page = 1
+        while True:
+            result = world.execute(operation, query={'page': page, 'count': 20})
+            if page * 20 >= result['collection']['total_count']:
+                break
+            page += 1
+    world.execute('search.results')
+    assert evaluate(world, report).violations == ('python_execution_not_verified',)
+    receipt = {'success': True, 'tool_calls_made': len(world.ledger), 'output': json.dumps(report)}
+    trace = [{'type': 'tool', 'data': {'name': 'execute_code', 'content': json.dumps(receipt)}}]
+    assert evaluate(world, report, trace).passed
+    receipt['tool_errors'] = [{'tool': 'moviepilot_api', 'error': 'failed'}]
+    trace[0]['data']['content'] = json.dumps(receipt)
+    assert not evaluate(world, report, trace).passed

tests/test_agent_evaluation_world.py

@@ -30,7 +30,7 @@ def _download_body(world: EvaluationWorld) -> dict[str, Any]:
 def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None:
     """公开定义只有任务、业务身份及资源，答案和故障时序不能进入模型上下文。"""
     scenarios = list_scenarios()
-    assert len(scenarios) == 13
+    assert len(scenarios) == 14
     api_scenarios = [scenario for scenario in scenarios if scenario.kind == "api"]
     assert len(api_scenarios) == 8
     assert len({scenario.media_id for scenario in api_scenarios}) == 6
@@ -53,7 +53,7 @@ def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None:
                 assert not scenario.steering_message and len(scenario.steering_plan) == 2
             else:
                 assert scenario.steering_message == "" and not scenario.steering_plan
-        elif scenario.kind == "command":
+        elif scenario.kind in {"command", "code"}:
             assert scenario.command
             assert scenario.browser_url == ""
             assert scenario.terminal_use_pty is None
@@ -77,9 +77,10 @@ def test_scenarios_expose_inputs_without_initial_state_or_oracle() -> None:
 
 def test_world_has_no_application_network_or_persistence_imports() -> None:
     """静态约束模拟业务独立于真实宿主，避免评测和实现共享同一份业务答案。"""
-    allowed = {"base64", "binascii", "copy", "dataclasses", "re", "threading", "typing", "urllib.parse", "scripts.evaluation.scenarios"}
+    allowed = {"base64", "binascii", "collections", "copy", "dataclasses", "json", "re", "threading", "typing", "urllib.parse",
+               "scripts.evaluation.scenarios", "scripts.evaluation.programs"}
     directory = Path(__file__).parents[1] / "scripts" / "evaluation"
-    for filename in ("world.py", "scenarios.py", "__init__.py"):
+    for filename in ("world.py", "scenarios.py", "programs.py", "__init__.py"):
         tree = ast.parse((directory / filename).read_text(encoding="utf-8"))
         for node in ast.walk(tree):
             if isinstance(node, ast.Import):

tests/test_agent_learning_review.py

@@ -16,6 +16,7 @@ from pydantic import Field
 from app.agent.learning.review import ReviewLoop, ReviewSnapshot
 from app.agent.learning.session import LearningSession
 from app.agent.middleware.learning import LearningCaptureMiddleware, LearningMiddleware
+from app.agent.middleware.memory import MemoryMiddleware
 from app.runtime.tasks import TaskRegistry
 
 
@@ -107,6 +108,65 @@ def test_real_graph_capture_is_final_and_does_not_share_state(tmp_path):
     asyncio.run(scenario())
 
 
+def test_delivered_graph_review_is_consumed_by_next_user_session(tmp_path, monkeypatch):
+    """真实工具迭代触发隔离复盘，落盘成果由下一会话读取且不会泄漏给另一用户。"""
+    async def scenario():
+        """只脚本化模型决定；图、计数、复盘、存储与下一轮加载都使用实际实现。"""
+        registry = TaskRegistry()
+        monkeypatch.setattr('app.agent.learning.session.get_task_registry', lambda: registry)
+        owner = session(tmp_path / 'alice')
+        fact = '订阅记录应按媒体来源与媒体编号核对。'
+        document = '---\nname: subscription-check\ndescription: 核对订阅记录\n---\n读取全部分页，按媒体来源和编号核对。\n'
+        observed = []
+
+        async def inspect_subscription(index: int) -> str:
+            """返回隔离业务快照，十次真实工具回合不能直接伪造计数。"""
+            observed.append(index)
+            return json.dumps({'success': True, 'media_source': 'themoviedb', 'media_id': str(index)})
+
+        model = ReviewModel(responses=[
+            *[response('inspect_subscription', {'index': index}, f'inspect-{index}') for index in range(10)],
+            AIMessage(content='已核对十条记录。'),
+            response('memory', {'action': 'add', 'content': fact}, 'save-memory'),
+            response('skill_manage', {'operations': [{'action': 'create', 'name': 'subscription-check',
+                                                      'content': document}]}, 'save-skill'),
+            AIMessage(content='复盘完成。'),
+        ])
+        messages = [HumanMessage(content=f'请核对记录 {index}', id=f'user-{index}') for index in range(10)]
+        await owner.begin(messages)
+        graph = create_agent(model=model, tools=[StructuredTool.from_function(coroutine=inspect_subscription)],
+                             middleware=[LearningMiddleware(owner), LearningCaptureMiddleware(owner)])
+        result = await graph.ainvoke({'messages': messages}, {'recursion_limit': 100})
+        before = deepcopy(result['messages'])
+        assert observed == list(range(10)) and owner.iterations == 10 and owner.memory_due
+        owner.finish(delivered=True)
+        assert owner.run and owner.run.task
+        await owner.run.task
+        assert result['messages'] == before
+        assert [action['tool'] for action in owner.last_actions] == ['memory', 'skill_manage']
+        assert (owner.memory_root / 'MEMORY.md').read_text() == fact
+        assert (owner.skill_root / 'subscription-check' / 'SKILL.md').read_text() == document
+
+        for username in ('alice', 'bob'):
+            following = session(tmp_path / username)
+            responses = ([response('skill_view', {'name': 'subscription-check'}, 'load-skill')]
+                         if username == 'alice' else [])
+            next_model = ReviewModel(responses=[*responses, AIMessage(content='下一任务')])
+            next_graph = create_agent(model=next_model, middleware=[
+                MemoryMiddleware(memory_dir=str(tmp_path / 'public'), user_memory_dir=str(following.memory_root),
+                                 store=following.tools.memory),
+                LearningMiddleware(following), LearningCaptureMiddleware(following),
+            ])
+            await next_graph.ainvoke({'messages': [HumanMessage(content='再次核对订阅')]})
+            sent = '\n'.join(message.text for request, _kwargs in next_model.requests for message in request)
+            assert (fact in sent) == (username == 'alice')
+            assert ('subscription-check' in sent) == (username == 'alice')
+            assert ('读取全部分页，按媒体来源和编号核对。' in sent) == (username == 'alice')
+        assert await registry.shutdown()
+
+    asyncio.run(scenario())
+
+
 def test_budget_iteration_limit_and_skill_only_memory_gate(tmp_path):
     """重复拒绝也消耗迭代预算，技能复盘不能借广告中的 memory 自动改记忆。"""
     async def scenario():

tests/test_agent_recall_persistence.py

@@ -325,34 +325,68 @@ def test_independent_retention_uses_shared_cutoff_and_disable(tmp_path):
 
 
 @pytest.mark.asyncio
-async def test_compaction_success_preserves_same_session_recall(repository):
-    """实际压缩提交删除的原文可在同会话召回，而未压缩尾部不重复进入历史。"""
+@pytest.mark.parametrize('streaming', [False, True])
+@pytest.mark.parametrize('fail_first', [False, True])
+async def test_compaction_success_preserves_same_session_recall(repository, monkeypatch, streaming, fail_first):
+    """真实图提交压缩后，被移出的原文在同会话可查，活动尾部与摘要不冒充历史。"""
+    from unittest.mock import AsyncMock
 
-    from langchain.agents.middleware.types import ExtendedModelResponse, ModelResponse
-    from langgraph.types import Command
-    middleware = RecallMiddleware(RecallService(repository, _Executor()), 'alice', RecallSession('current'))
-    old = HumanMessage(id='old', content='needle old request')
-    tail = HumanMessage(id='tail', content='current query')
-    await middleware.abefore_model({'messages': [old, tail]}, None)
+    from langchain.agents import create_agent
+    from langchain_core.language_models.fake_chat_models import FakeMessagesListChatModel
+    from langgraph.checkpoint.memory import InMemorySaver
 
-    class Request:
-        """只实现中间件需要的请求覆写协议。"""
+    from app.agent.middleware.summarization import (
+        ContextPreservingSummarizationMiddleware,
+        FinalRequestCompactionMiddleware,
+    )
 
-        messages = [old, tail]
-        system_message = None
+    failure_pending = [fail_first]
 
-        def override(self, **_values):
-            """返回同一受控请求，测试只关注压缩提交的历史可见性。"""
+    class Model(FakeMessagesListChatModel):
+        """固定主模型响应，只验证生产图的压缩提交与归档边界。"""
+
+        def bind_tools(self, _tools, **_kwargs):
+            """保留真实工具绑定路径而不访问模型服务。"""
             return self
 
-    async def handler(_request):
-        """模拟最内层压缩中间件在模型成功后提交状态。"""
-        return ExtendedModelResponse(model_response=ModelResponse(result=[AIMessage(content='done')]),
-                                     command=Command(update={'messages': [tail]}))
+        async def _agenerate(self, messages, **kwargs):
+            """故障只发生在摘要完成后的主模型调用，验证失败不会提交压缩。"""
+            if failure_pending[0]:
+                failure_pending[0] = False
+                raise TimeoutError('main model unavailable')
+            return await super()._agenerate(messages, **kwargs)
 
-    await middleware.awrap_model_call(Request(), handler)
-    assert json.loads(await middleware.session_search(query='needle'))['count'] == 1
+    model = Model(responses=[AIMessage(content='done')], profile={'max_input_tokens': 4096})
+    summarizer = ContextPreservingSummarizationMiddleware(model=model, trigger=('fraction', 0.85), keep=('messages', 2))
+    summary = AsyncMock(return_value='old context summary')
+    monkeypatch.setattr(summarizer, 'acreate_summary', summary)
+    middleware = RecallMiddleware(RecallService(repository, _Executor()), 'alice', RecallSession('current'))
+    graph = create_agent(model=model, tools=[], middleware=[middleware, FinalRequestCompactionMiddleware(summarizer=summarizer)],
+                         checkpointer=InMemorySaver())
+    messages = [HumanMessage(id='task', content='核对历史证据')]
+    messages += [HumanMessage(id=f'old-{index}', content=f'needle evidence {index} ' * 2000) for index in range(4)]
+    messages += [HumanMessage(id='tail', content='current query')]
+    config = {'configurable': {'thread_id': 'current'}}
+    payload = {'messages': messages}
+    if fail_first:
+        with pytest.raises(TimeoutError, match='main model unavailable'):
+            await graph.ainvoke(payload, config=config)
+        assert json.loads(await middleware.session_search(query='needle'))['count'] == 0
+        assert {message.id for message in (await graph.aget_state(config)).values['messages']} == {message.id for message in messages}
+        payload = None
+    if streaming:
+        async for _ in graph.astream(payload, config=config):
+            pass
+    else:
+        await graph.ainvoke(payload, config=config)
+    state = (await graph.aget_state(config)).values
+    assert summary.await_count > 0
+    kept = {message.id for message in state['messages']}
+    assert any(message.id not in kept for message in messages[1:-1])
+    result = json.loads(await middleware.session_search(query='needle'))
+    assert result['count'] == 1
     assert json.loads(await middleware.session_search(query='current'))['count'] == 0
+    assert json.loads(await middleware.session_search(query='summary'))['count'] == 0
 
 
 @pytest.mark.asyncio

tests/test_builtin_skill_boundaries.py

@@ -117,7 +117,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None:
         "command-dispatch": "2",
         "database-operation": "9",
         "feedback-issue": "10",
-        "moviepilot-api": "40",
+        "moviepilot-api": "41",
         "moviepilot-update": "5",
         "organize-files": "5",
         "transfer-failed-retry": "5",

tests/test_system_utils.py

@@ -252,7 +252,8 @@ async def test_async_subprocess_cancellation_reaps_process(tmp_path):
         "-c",
         (
             "from pathlib import Path; import os, time; "
-            f"Path({str(marker)!r}).write_text(str(os.getpid())); time.sleep(60)"
+            f"target = Path({str(marker)!r}); ready = target.with_suffix('.ready'); "
+            "ready.write_text(str(os.getpid())); ready.replace(target); time.sleep(60)"
         ),
     ]
     task = asyncio.create_task(
@@ -847,11 +848,12 @@ def test_space_usage_keeps_windows_drive_behavior_without_fsid_lookup():
 
 
 def test_local_storage_usage_forwards_btrfs_fsid_setting():
+    """存储层从运行配置门面取得去重开关，不能依赖已移除的具体配置对象。"""
     from app.modules.filemanager.storages import local as local_storage_module
 
     download_dir = MagicMock(download_path="/downloads")
     library_dir = MagicMock(library_path="/library")
-    with patch.object(local_storage_module.settings, "BTRFS_FSID_DEDUP", True), \
+    with patch.object(local_storage_module, "get_runtime_setting", return_value=True) as setting_mock, \
             patch.object(local_storage_module.DirectoryHelper, "get_local_download_dirs",
                          return_value=[download_dir]), \
             patch.object(local_storage_module.DirectoryHelper, "get_local_library_dirs",
@@ -861,6 +863,7 @@ def test_local_storage_usage_forwards_btrfs_fsid_setting():
 
     assert usage.total == 4.0
     assert usage.available == 2.0
+    setting_mock.assert_called_once_with("BTRFS_FSID_DEDUP")
     usage_mock.assert_called_once_with(
         [Path("/downloads"), Path("/library")],
         btrfs_fsid_dedup=True,