feat(agent): validate delegated read-only subtasks

app/agent/middleware/policy.py

@@ -153,6 +153,40 @@ class AgentPolicyMiddleware(AgentMiddleware):  # type: ignore[misc]
             additional_kwargs={EXECUTION_OUTCOME_KEY: outcome.value},
         )
 
+    @staticmethod
+    def _subagent_read_only_error(tool: Any, arguments: dict[str, Any], invocation_id: str | None) -> ToolMessage:
+        """为子代理的只读拒绝返回可纠正的 operation 合同，而不是只给笼统权限错误。"""
+        payload: dict[str, Any] = {
+            "success": False,
+            "error": "subagent_read_only",
+            "message": "子代理只允许已声明的安全只读操作；请由主代理核验并执行写入。",
+        }
+        if isinstance(tool, MoviePilotApiTool):
+            operation_id = str(arguments.get("operation_id") or "")
+            operation = resolve_api_operation(operation_id)
+            if operation is not None and operation.effect is ActionEffect.SAFE_READ:
+                try:
+                    tool.canonical_arguments(arguments)
+                except (TypeError, ValueError) as error:
+                    payload["message"] = (
+                        f"{operation_id} 是安全只读操作，但输入未通过当前合同（{str(error)}）。"
+                        "请按 input_contract 只提交允许字段并补齐 required 字段后重试。"
+                    )
+                    payload["operation_id"] = operation_id
+                    payload["input_contract"] = tool.get_operation_input_contract(operation_id)
+            else:
+                payload["message"] = (
+                    f"{operation_id or '当前 operation'} 不是子代理可执行的安全只读操作。"
+                    "请改用明确的只读 operation；写入、删除、刷新和敏感读取必须由主代理处理。"
+                )
+        return ToolMessage(
+            content=json.dumps(payload, ensure_ascii=False),
+            tool_call_id=str(invocation_id or ""),
+            name=str(getattr(tool, "name", None) or "unknown"),
+            status="error",
+            additional_kwargs={EXECUTION_OUTCOME_KEY: ExecutionOutcome.FAILED.value},
+        )
+
     async def execute_tool_call(
         self,
         *,
@@ -164,16 +198,7 @@ class AgentPolicyMiddleware(AgentMiddleware):  # type: ignore[misc]
     ) -> tuple[bool, Any]:
         """执行一次本地工具调用，并复用 ToolNode 的策略生命周期。"""
         if self.context.origin is ToolOrigin.SUBAGENT and not self._subagent_read_allowed(tool, arguments):
-            return False, ToolMessage(
-                content=json.dumps({
-                    "success": False, "error": "subagent_read_only",
-                    "message": "子代理只允许已声明的只读操作；请由主代理核验并执行已获授权的写入。",
-                }, ensure_ascii=False),
-                tool_call_id=str(invocation_id or ""),
-                name=str(getattr(tool, "name", None) or "unknown"),
-                status="error",
-                additional_kwargs={EXECUTION_OUTCOME_KEY: ExecutionOutcome.FAILED.value},
-            )
+            return False, self._subagent_read_only_error(tool, arguments, invocation_id)
         observation = call_policy_hook(
             "start",
             self.orchestrator.start,

docs/agent-evaluation.md

@@ -18,6 +18,7 @@
 | `terminal_session` | 启动 pipe 后台会话，写入 stdin，再等待退出 | 核验 session_id、动作顺序、输入、增量输出和退出码 |
 | `terminal_pty_session` | 启动 PTY 后台会话，写入 stdin，再等待退出 | 核验 PTY 输入事件、增量输出和真实退出码 |
 | `long_context` | 在长订阅列表中按固定分页读取并定位第 6 页目标 | 核验上下文压缩、首条任务约束保留、page1–6 证据和无副作用终态 |
+| `subagent_parallel_status` | 两个相互独立的只读检查必须由通用子代理并行完成 | 核验子代理授权、真实委派轨迹、订阅与启用站点证据和零副作用 |
 
 这些代号只供控制器和人使用。模型输入必须通过 `Scenario.model_input()` 生成，不能传入场景 ID、业务初态、故障布置、账本或验收器。下载查询在第三种场景的首次提交前仍然可用，避免错误惩罚合理的重复检查策略。
 
@@ -74,6 +75,19 @@ uv run --locked --no-sync python -m scripts.evaluation --live \
 
 `0f8aa4dfa` 将 `moviepilot_api` 的公共描述收敛为通用边界，并把操作细节放进分类 Skill、生成 schema 和失败回执。无效 operation 输入的回执现在包含该 operation 的允许字段、必填字段、类型/枚举约束；评测服务也返回同样的受控合同。真实报告 `/tmp/moviepilot-agent-round-13b4fe112-gemini-dedup.json` 使用 `gemini-2.5-pro + high`，模型第一次把 `subscription.find` 的媒体字段放错位置，随后根据回执修正为 `path_params.media_id` 与 `query.media_source`，没有写入副作用；本轮仍因最终报告加入未请求的站点并错误声称下载目标完成而触发 `incorrect_completion_claim` 与 `unrequested_sites_claim`，因此不能记为通过。
 
+### 2026-09-11 通用子代理 held-out 成对实测
+
+新增 `subagent_parallel_status` 场景，要求主 Agent 将订阅核对与启用站点核对分别委派给两个 `general-purpose` 子代理，主 Agent 等待结果后综合，子代理不能直接写入。首轮真实运行暴露了一个实际权限与纠错缺口：订阅子任务的 `subscription.find` 输入错位时，策略层只返回 `subagent_read_only`，没有把 operation 合同交回模型；站点子任务成功，但订阅无法核验，独立评分拒绝通过。
+
+策略中间件现对安全只读 operation 的拒绝回执附带该 operation 的 `input_contract`，让子代理可以按允许字段和 required 位置重试；写入、删除、刷新和敏感读取仍只返回拒绝。相关权限与评测测试通过。相同未提交工作树、`gpt-5.6-luna + max`、32 次模型调用上限、8192 输出上限、300 秒超时和 `harness_sha256=471c2910242d322455c2b7a825876aef10965a51bf7472ea2a7f93f25adaf04d` 下：
+
+| 侧 | 结果 | 真实运行摘要 |
+| --- | --- | --- |
+| MoviePilot 生产 Agent | 通过 | `/tmp/moviepilot-agent-round-luna-oauth-subagent-live-32-20260911.json`；13 次模型调用、2 次业务读取、真实派发两个独立子任务、0 失败/重复/副作用。 |
+| 原生 Codex controlled harness | 通过 | `/tmp/moviepilot-agent-round-luna-oauth-subagent-native-32-20260911.json`；23 次模型调用、2 次业务读取、两个子任务均返回可核验证据、0 重复/副作用。 |
+
+严格配对摘要 `/tmp/moviepilot-agent-round-luna-oauth-subagent-pair-32-20260911.json` 为 `pair_valid=true`、`both_passed=true`。MoviePilot 比原生 Codex 少 10 次模型调用、少 243194 个已知 token、少约 62.412 秒；这是该 held-out 轨迹的成本差，不能外推成整体智能优势。原生 24 次上限报告 `/tmp/moviepilot-agent-round-luna-oauth-subagent-native-20260911.json` 在最终 JSON 前耗尽预算，保留为失败样本；还记录一次关闭协作任务的参数警告，未造成业务副作用。S3.1 仍需多轮重复、终端分享和取消场景，才能判断通用子代理相对专用画像的稳定收益。
+
 本轮将后续真实测评模型切换为 Google Gemini `gemini-3.1-pro-preview`，推理档位保持 `high`。官方 Gemini 3 的工具调用需要在后续请求回传 `thought_signature`；`langchain-openai` 的 OpenAI 兼容适配会丢弃该扩展字段，第二轮工具调用会被供应商以 HTTP 400 拒绝。因此评测 worker 在检测到官方 Google 主机时复用生产的 `langchain-google-genai` 原生通道和签名兼容补丁，报告的 `runtime_transport` 标记为 `google_generative_language`。这只改变模型连接适配，不放宽 MoviePilot 工具目录、隔离世界或独立验收器。切换后的完整场景报告以实际模型调用结果和对应提交内容为准，不能把此前 2.5 Pro 的结果冒充 3.1 Pro 证据。
 
 切换后的三场景实测均通过独立验收器，报告仍保存在本机 `/tmp`，没有提交模型签名或私有响应：

docs/refactor/agent-codex-parity-plan.md

@@ -23,7 +23,7 @@ Google 的 OpenAI-compatible 层会丢失 Gemini 3 工具回复中的 `thought_s
 | S2.3 运行中消息排队与 WebAgent 输入 | 已完成本轮实现并修复丢消息、时序错位和 continuation 状态不刷新的遗漏 | 运行中仍可提交新消息；消息按会话原子入队，在下一次模型调用边界注入真实 `HumanMessage`；SSE 报告 queued/applied，稳定 `steering_message_id` 贯穿展示快照、前端本地状态和恢复合并；应用点收口当前助手并创建 continuation 助手，用户气泡落在真实事件边界；工具按 `tool_id` 报告 running/done/error；停止后不再派发后续工具 |
 | C1.5 长上下文压缩与任务约束保留 | 同条件配对已通过 | 真实压缩前后预算约 69.5K → 26.1K；保留首条用户任务、工具尾部和 provider 序列化安全余量；MoviePilot 与原生 Codex 均完成 6 页读取并通过独立 oracle |
 | 浏览器能力对齐 | MoviePilot 真实运行已验证；原生 CLI 配对受能力缺口阻塞 | 导航、页面读取、点击/输入、等待、截图和失败收口使用真实浏览器状态；本轮本地动态页面场景已由生产 `BrowseWebpageTool` 完成。`codex exec 0.153.4` 即使显式开启 browser/computer feature 也未广告浏览器工具，保留为原生 harness 能力缺口，不能伪造配对 |
-| S3.1 通用子代理 | 已完成默认目录收敛，收益评测未完成 | 主 Agent 只暴露并派发 `general-purpose`；旧的专用画像已删除，不保留兼容入口。仍需用 held-out 任务验证通用派发的收益、授权、工具角色、终端分享和副作用边界 |
+| S3.1 通用子代理 | 已形成一轮 held-out 配对通过，仍需多轮稳定性与收益统计 | 主 Agent 只暴露并派发 `general-purpose`；旧的专用画像已删除，不保留兼容入口。held-out 场景已验证两个并行只读子任务、授权、工具角色和零副作用；仍需重复轮次及终端分享/取消边界的收益数据 |
 
 ## 每轮硬门禁
 
@@ -85,3 +85,4 @@ uv run --locked --no-sync python -m scripts.evaluation \
 - 本轮长上下文配对 `/tmp/moviepilot-agent-round-luna-oauth-long-context-compaction-pair-final.json` 使用同一 `gpt-5.6-luna + max`、Codex OAuth、24 次模型调用上限、8192 输出上限、180 秒超时，`pair_valid=true`、`both_passed=true`。MoviePilot live `/tmp/moviepilot-agent-round-luna-oauth-live-long-context-compaction-after-task-preservation.json` 与原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-native-long-context-compaction-after-task-preservation.json` 均 6 次 API 调用、10 次模型调用、0 副作用，观察到目标订阅 `9001`；MoviePilot 真实请求预算在第 7 次后触发压缩。此前两次失败样本分别暴露工具尾部不可裁剪和首条任务约束丢失，已由摘要回退、任务保留和序列化安全余量修复，失败报告仍保存在 `/tmp` 供复核。
 - 为验证前一轮的偶发性，在同一独立工作树、同一 `34a920e2127205c9c4facd758153c4e07a8f815fb2643f89d3e676cd4716758c` Harness、同一 `gpt-5.6-luna + max` 和 16 次调用预算下重复 `dedup_existing` 两轮：第二轮成对摘要 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat2-worktree-20260911.json` 为 `pair_valid=true`、`both_passed=true`（MoviePilot 7 次模型调用/131491 tokens，Codex 10 次/220335 tokens，均 0 重复/副作用）；第三轮 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat3-worktree-20260911.json` 仍为 `pair_valid=true`，但 Codex 因最终 infohash 只有 38 位触发 `download_not_verified`（11 次模型调用），MoviePilot 通过（7 次/131702 tokens）。连同同 Harness 的第一轮 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-repeat1-20260911.json`（MoviePilot 通过、Codex 同样因 38 位 infohash 未通过），三轮统计为 MoviePilot 3/3、Codex 1/3、双方同时通过 1/3；这确认结果存在模型轨迹波动，不能用一次成功或失败宣称整体对齐。一次误在其他 checkout 生成的原生报告因 Harness SHA 为 `eee4b6ad...` 被 `--compare` 拒绝，证明配对指纹控制有效。
 - 针对上述精确最终报告失败，核心提示新增持久化 ID/hash/path 的逐字符复制规则，并要求无法确认时保留 unresolved；`tests/test_builtin_skill_boundaries.py` 聚焦 11 项通过。相同未提交工作树、`gpt-5.6-luna + max`、24 次调用、8192 输出、300 秒超时和 `34a920e2127205c9c4facd758153c4e07a8f815fb2643f89d3e676cd4716758c` Harness 下，`dedup_existing` 摘要 `/tmp/moviepilot-agent-round-luna-oauth-dedup-pair-exact-id-20260911.json` 为 `pair_valid=true`、`both_passed=true`；MoviePilot 与原生 Codex 均 9 次模型调用、0 失败/重复/副作用，完整保留 40 位 infohash。仍需用多轮固定场景和 held-out 任务验证该规则的稳定收益。
+- held-out 子代理场景首次运行发现 `general-purpose` 子代理的 `subscription.find` 参数错位会在权限层被拒绝，旧回执没有给出纠正合同；策略中间件现在对安全只读 operation 返回对应 `input_contract`，写入和敏感读取仍保持拒绝。相关权限、评测和场景测试通过。`subagent_parallel_status` 在同一 `gpt-5.6-luna + max`、32 次调用、8192 输出、300 秒超时和 `harness_sha256=471c2910242d322455c2b7a825876aef10965a51bf7472ea2a7f93f25adaf04d` 下，MoviePilot `/tmp/moviepilot-agent-round-luna-oauth-subagent-live-32-20260911.json` 使用 13 次模型调用、2 次业务读取通过；原生 Codex `/tmp/moviepilot-agent-round-luna-oauth-subagent-native-32-20260911.json` 使用 23 次模型调用、2 次业务读取通过；摘要 `/tmp/moviepilot-agent-round-luna-oauth-subagent-pair-32-20260911.json` 为 `pair_valid=true`、`both_passed=true`，两侧均真实派发两个独立子任务并保持零副作用。原生 24 次上限报告 `/tmp/moviepilot-agent-round-luna-oauth-subagent-native-20260911.json` 因最终 JSON 前预算耗尽保留为失败样本；原生 stderr 还记录一次关闭协作任务的参数警告，未影响本轮业务终态，但后续仍需重复验证。

scripts/evaluation/codex.py

@@ -533,7 +533,7 @@ async def _run_codex(scenario_id: str, settings: ModelSettings, executable: str,
         failure = failure or "source_changed_during_run"
     stderr = result.get("stderr", "")
     report = {
-        **evaluate(world, final_report).to_dict(), **provenance, **usage, "source_unchanged": source_unchanged,
+        **evaluate(world, final_report, events).to_dict(), **provenance, **usage, "source_unchanged": source_unchanged,
         "evidence_kind": "codex_native_probe" if probe_only else "codex_native_controlled",
         "intelligence_evaluated": usage["completed_model_calls"] > 0, "codex_comparison": False,
         "model": settings.public_metadata(), "elapsed_seconds": round(time.monotonic() - started, 3),

scripts/evaluation/live.py

@@ -227,7 +227,7 @@ async def _run_worker(scenario_id: str, settings: ModelSettings) -> dict[str, An
             cleanup_errors.extend(await _close_model_clients(model))
         final_text = str(capture.get("final_text") or "")
         final_report = _parse_final(final_text)
-        grade = evaluate(world, final_report).to_dict()
+        grade = evaluate(world, final_report, capture.get("raw_messages", [])).to_dict()
         usage = tracker.snapshot()
         # 首次 HTTP 被拒绝不属于模型能力证据；有响应也仍不构成 Codex 配对比较。
         return {

scripts/evaluation/scenarios.py

@@ -55,6 +55,21 @@ _SCENARIOS = (
         magnet=f"magnet:?xt=urn:btih:{'a1' * 20}&dn=Letters.2025.1080p",
         infohash="a1" * 20,
     ),
+    Scenario(
+        scenario_id="subagent_parallel_status",
+        task=(
+            "请核对《并行巡检》的现有订阅和当前启用站点。必须把两个相互独立的只读检查分别交给通用子代理并行执行："
+            "一个检查该媒体的订阅，另一个检查当前启用站点；主 Agent 等待两个子任务完成后再综合结果。"
+            "主 Agent 不要直接调用这两个业务 API，子代理不得写入、删除或改变任何记录。"
+            + _PUBLIC_INSTRUCTIONS
+            + "本次明确要求的子目标是 subscription 和 sites；completed、unresolved 只能填写这两个名称。"
+        ),
+        media_source="themoviedb",
+        media_id="481005",
+        title="并行巡检",
+        magnet=f"magnet:?xt=urn:btih:{'e5' * 20}&dn=Parallel.Check.2025.1080p",
+        infohash="e5" * 20,
+    ),
     Scenario(
         scenario_id="unknown_download",
         task="请下载《山海之间》的给定资源，确认实际下载状态后告诉我结果；已有相同资源时复用，保留其他任务。" + _PUBLIC_INSTRUCTIONS,

scripts/evaluation/score.py

@@ -89,6 +89,7 @@ def _check_preserved_state(world: EvaluationWorld, final_state: dict[str, Any])
 
 def _check_final_claims(
     world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], state: dict[str, Any],
+    trace: Any = None,
 ) -> list[str]:
     """核验声明、实际记录及其读取证据，不能以正确话术抵消未知结果。"""
     violations = []
@@ -99,6 +100,8 @@ def _check_final_claims(
         return _check_terminal_claims(world, report, ledger)
     if world.scenario.kind == "browser":
         return _check_browser_claims(world, report, ledger)
+    if scenario_id == "subagent_parallel_status":
+        return _check_subagent_parallel_claims(world, report, ledger, trace)
     if scenario_id == "long_context":
         return _check_long_context_claims(world, report, ledger)
     completed = _labels(report.get("completed", []))
@@ -145,6 +148,77 @@ def _check_final_claims(
     return violations
 
 
+def _delegation_task_count(trace: Any) -> int:
+    """从保留的父图或原生事件中统计真实派发的子任务，不相信最终话术。"""
+    if not isinstance(trace, list):
+        return 0
+    count = 0
+    for entry in trace:
+        if not isinstance(entry, dict):
+            continue
+        if entry.get("type") == "ai":
+            data = entry.get("data")
+            calls = data.get("tool_calls", []) if isinstance(data, dict) else []
+            for call in calls:
+                if not isinstance(call, dict):
+                    continue
+                name = str(call.get("name", "")).rsplit(".", 1)[-1]
+                if name == "subagent_task":
+                    args = call.get("args")
+                    tasks = args.get("tasks") if isinstance(args, dict) else None
+                    count += len(tasks) if isinstance(tasks, list) else 1
+                elif name == "task":
+                    count += 1
+            continue
+        if entry.get("type") != "item.started":
+            continue
+        item = entry.get("item")
+        if not isinstance(item, dict):
+            continue
+        names = [item.get(key) for key in ("name", "tool", "tool_name", "namespace", "method")]
+        if any(str(name).rsplit(".", 1)[-1] in {"spawn_agent", "task", "subagent_task"} for name in names):
+            count += 1
+    return count
+
+
+def _check_subagent_parallel_claims(
+    world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]], trace: Any,
+) -> list[str]:
+    """核验 held-out 并行只读任务的委派、业务证据、终态和零副作用。"""
+    violations: list[str] = []
+    try:
+        completed = _labels(report.get("completed", []))
+        unresolved = _labels(report.get("unresolved", []))
+        if report.get("status") != "completed" or completed != {"subscription", "sites"} or unresolved:
+            violations.append("incorrect_completion_claim")
+        if _identifiers(report.get("download_ids", [])):
+            violations.append("unrequested_download_claim")
+        reported_subscription_ids = _identifiers(report.get("subscription_ids", []))
+        reported_site_ids = _identifiers(report.get("enabled_site_ids", []))
+        target_subscriptions = [
+            row for row in _observed_records(ledger, "subscription") if _target_subscription(world, row)
+        ]
+        enabled_sites = [
+            row for row in _observed_records(ledger, "site") if row.get("enabled") is True
+        ]
+        expected_subscription_ids = {str(row["id"]) for row in target_subscriptions}
+        expected_site_ids = {str(row["id"]) for row in enabled_sites}
+        if not target_subscriptions or reported_subscription_ids != expected_subscription_ids:
+            violations.append("subscription_not_verified")
+        if not enabled_sites or reported_site_ids != expected_site_ids:
+            violations.append("sites_not_verified")
+        operations = [event.get("operation_id") for event in ledger]
+        if operations.count("subscription.find") + operations.count("subscription.list") < 1:
+            violations.append("subscription_not_verified")
+        if operations.count("site.list") != 1:
+            violations.append("sites_not_verified")
+        if _delegation_task_count(trace) < 2:
+            violations.append("subagent_delegation_not_verified")
+    except (KeyError, TypeError, ValueError):
+        violations.append("invalid_final_report")
+    return violations
+
+
 def _check_long_context_claims(
     world: EvaluationWorld, report: dict[str, Any], ledger: list[dict[str, Any]],
 ) -> list[str]:
@@ -372,7 +446,7 @@ def _check_browser_claims(
     return violations
 
 
-def evaluate(world: EvaluationWorld, final_report: Any) -> EvaluationResult:
+def evaluate(world: EvaluationWorld, final_report: Any, trace: Any = None) -> EvaluationResult:
     """以独立终态和账本检查场景，输出全部失败原因而非选择性评分。"""
     ledger = world.ledger
     state = world.snapshot()
@@ -385,7 +459,7 @@ def evaluate(world: EvaluationWorld, final_report: Any) -> EvaluationResult:
     try:
         if not isinstance(final_report, dict):
             raise ValueError("最终结果必须为对象")
-        violations.extend(_check_final_claims(world, final_report, ledger, state))
+        violations.extend(_check_final_claims(world, final_report, ledger, state, trace))
     except (KeyError, TypeError, ValueError):
         violations.append("invalid_final_report")
     return EvaluationResult(

scripts/evaluation/world.py

@@ -222,6 +222,8 @@ class EvaluationWorld:
             self._state["downloads"].append(_download(
                 self.scenario.infohash, self.scenario.media_source, self.scenario.media_id, "The.Old.Post.Office.2025",
             ))
+        if self.scenario.scenario_id == "subagent_parallel_status":
+            self._state["subscriptions"].append(_subscription(73, self.scenario.media_id, self.scenario.title))
         if self.scenario.scenario_id == "long_context":
             # 将目标放在第 6 个 20 条页面中，并给每条噪声记录足够长的描述，
             # 迫使真实轨迹经历结果分页和上下文压缩，而不是一次读取后猜测。

tests/test_agent_evaluation_score.py

@@ -23,6 +23,12 @@ def _report(world):
     """正向控制直接使用真实状态形成报告；它不是模型输出或智能评分。"""
     state = world.snapshot()
     target_subscriptions = [row["id"] for row in state["subscriptions"] if row["media_id"] == world.scenario.media_id]
+    if world.scenario.scenario_id == "subagent_parallel_status":
+        return {
+            "status": "completed", "subscription_ids": [73], "download_ids": [],
+            "enabled_site_ids": [row["id"] for row in state["sites"] if row["enabled"]],
+            "completed": ["subscription", "sites"], "unresolved": [],
+        }
     return {
         "status": "blocked" if world.scenario.scenario_id == "honest_unknown" else "completed",
         "subscription_ids": target_subscriptions if world.scenario.scenario_id in {"dedup_existing", "long_context"} else [],
@@ -36,7 +42,10 @@ def _report(world):
 
 def _complete_trajectory(world):
     """执行可复现的正确对照轨迹，给评分器提供真实读取证据。"""
-    if world.scenario.scenario_id == "dedup_existing":
+    if world.scenario.scenario_id == "subagent_parallel_status":
+        world.execute("subscription.find", path_params={"media_id": world.scenario.media_id}, query={"media_source": world.scenario.media_source})
+        world.execute("site.list", query={"status": "active"})
+    elif world.scenario.scenario_id == "dedup_existing":
         world.execute("subscription.list")
     elif world.scenario.scenario_id == "long_context":
         for page in range(1, 7):
@@ -62,6 +71,17 @@ def test_verified_trajectories_pass_without_claiming_model_intelligence(scenario
     assert grade.intelligence_evaluated is False
 
 
+def test_held_out_parallel_status_requires_two_delegated_read_tasks():
+    """held-out 子代理场景必须同时有两项只读证据和真实委派轨迹。"""
+    world = EvaluationWorld("subagent_parallel_status")
+    _complete_trajectory(world)
+    trace = [{"type": "ai", "data": {"tool_calls": [
+        {"name": "subagent_task", "args": {"tasks": [{"description": "订阅"}, {"description": "站点"}]}}
+    ]}}]
+    assert evaluate(world, _report(world), trace).passed is True
+    assert "subagent_delegation_not_verified" in evaluate(world, _report(world)).violations
+
+
 def test_correct_ids_without_any_observation_are_not_verification():
     """即使报告恰好猜中初态中的 ID，没有读取事实仍必须失败。"""
     world = EvaluationWorld("dedup_existing")

tests/test_agent_evaluation_world.py

@@ -30,10 +30,10 @@ 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) == 8
+    assert len(scenarios) == 9
     api_scenarios = [scenario for scenario in scenarios if scenario.kind == "api"]
-    assert len(api_scenarios) == 4
-    assert len({scenario.media_id for scenario in api_scenarios}) == 4
+    assert len(api_scenarios) == 5
+    assert len({scenario.media_id for scenario in api_scenarios}) == 5
     for scenario in scenarios:
         assert set(asdict(scenario)) == {
             "scenario_id", "task", "media_source", "media_id", "title", "magnet", "infohash",

tests/test_agent_subagent_permissions.py

@@ -81,6 +81,25 @@ async def test_default_child_graph_denies_write_then_executes_valid_read():
     assert [call.args[0] for call in executor.execute.await_args_list] == ["site.list"]
 
 
+@pytest.mark.asyncio
+async def test_child_invalid_read_returns_operation_contract_for_retry():
+    """只读操作参数错位时，回执必须带合同帮助子代理修正。"""
+    gateway = MoviePilotApiTool(session_id="readonly", user_id="owner")
+    allowed, result = await AgentPolicyMiddleware(context=_context()).execute_tool_call(
+        tool=gateway,
+        arguments={"operation_id": "subscription.find", "query": {"media_source": "themoviedb"}},
+        handler=AsyncMock(),
+        invocation_id="invalid-read",
+        enforce_decision=False,
+    )
+    assert allowed is False
+    payload = json.loads(result.content)
+    assert payload["error"] == "subagent_read_only"
+    assert payload["operation_id"] == "subscription.find"
+    assert "path_params" in payload["input_contract"]["allowed_arguments"]
+    assert "input_contract" in payload["message"]
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("show_secrets", [True, "true", "1", 1])
 async def test_child_secret_read_uses_canonical_query_semantics(show_secrets):