权限：按角色能力隔离 Planner Worker 和 Validator

为内置读写、编排、提交和子 Agent 工具补充 capability；explicit_validation 模式改为未分类工具默认拒绝，并在 Schema 暴露与真实执行两层使用同一授权结果。补充任意名称写工具、派生 Agent 工具及 legacy 兼容测试。

agent/agent/docs/orchestration-v1.md

@@ -72,6 +72,7 @@ Worker 只能使用 `submit_attempt` 正式交付。Validator 只能使用 `subm
 - Planner 是唯一可以改变任务树和作出完成决策的角色。
 - Worker 看不到且不能执行 `agent`、`evaluate`、`goal` 或 Planner/Validator 工具。
 - Validator 使用独立 Trace，只能访问 preset 明确允许的只读工具。
+- explicit 模式的工具必须声明 capability；未分类工具默认拒绝。Validator 仅允许 `read + validation_submit`，Worker 不允许 `agent_spawn` 和 `task_control`。
 - Worker 未调用 `submit_attempt`，Attempt 失败，Task 进入 `needs_replan`。
 - Validator 未调用 `submit_validation`，Validation 状态为 `error`，不是业务 `failed`。
 - `passed` 只让 Task 进入 `awaiting_decision`；只有 Planner 对当前版本、当前 Attempt、当前 Snapshot 执行 `accept` 才完成。

agent/agent/docs/presets.md

@@ -26,6 +26,10 @@
 }
 ```
 
+Host 注册给 explicit-validation 角色的工具还必须通过 `@tool(capabilities=[...])` 或 `ToolRegistry.register(..., capabilities=[...])` 声明安全副作用。未分类工具在 explicit 模式中 fail-closed，但在 `legacy_auto` 中保持原行为。
+
+核心 capability：`read`、`write`、`agent_spawn`、`external_send`、`task_control`、`attempt_submit`、`validation_submit`。一个工具可以同时声明多个 capability，只要其中一个超过当前角色上限，Schema 和执行层都会拒绝。
+
 有效工具集合按以下顺序计算：RunConfig 候选工具，与 preset 白名单取交集，再减去 preset、RunConfig 和 role 的禁用项。Schema 生成和真实执行前都会执行相同授权检查，因此模型伪造未授权 Tool Call 也不会进入工具函数。
 
 `legacy_auto` 继续使用原来的 tools/tool_groups 并集算法，但 V1 orchestration 工具不会出现在 legacy Schema 中。

agent/agent/orchestration/policy.py

@@ -6,25 +6,29 @@ from dataclasses import dataclass
 from typing import Any, FrozenSet, List, Optional
 
 from .models import AgentRole, CompletionPolicy
+from agent.tools.models import ToolCapability
 
 
 PLANNER_TOOLS = frozenset({"task_plan", "dispatch_tasks", "task_decide", "validate_attempt"})
 WORKER_TOOLS = frozenset({"submit_attempt"})
 VALIDATOR_TOOLS = frozenset({"submit_validation"})
 
-WRITE_TOOL_NAMES = frozenset({
-    "write", "write_file", "write_json", "edit", "edit_file", "bash", "bash_command",
-    "agent", "evaluate", "goal", "task_plan", "dispatch_tasks", "task_decide",
-    "validate_attempt", "submit_attempt", "publish", "send_message", "import_content",
-    "knowledge_save", "knowledge_update", "knowledge_batch_update", "knowledge_save_pending",
-})
-
-
-ROLE_HARD_DENIES = {
-    AgentRole.PLANNER: WORKER_TOOLS | VALIDATOR_TOOLS,
-    AgentRole.WORKER: frozenset({"agent", "evaluate", "goal"}) | PLANNER_TOOLS | VALIDATOR_TOOLS,
-    AgentRole.VALIDATOR: WRITE_TOOL_NAMES | WORKER_TOOLS,
-    AgentRole.LEGACY: frozenset(),
+ROLE_ALLOWED_CAPABILITIES = {
+    AgentRole.PLANNER: frozenset({
+        ToolCapability.READ,
+        ToolCapability.TASK_CONTROL,
+    }),
+    AgentRole.WORKER: frozenset({
+        ToolCapability.READ,
+        ToolCapability.WRITE,
+        ToolCapability.EXTERNAL_SEND,
+        ToolCapability.ATTEMPT_SUBMIT,
+    }),
+    AgentRole.VALIDATOR: frozenset({
+        ToolCapability.READ,
+        ToolCapability.VALIDATION_SUBMIT,
+    }),
+    AgentRole.LEGACY: frozenset(ToolCapability),
 }
 
 
@@ -64,7 +68,17 @@ class DefaultToolPolicy:
         if preset.denied_tools:
             candidates -= set(preset.denied_tools)
         candidates -= set(getattr(config, "exclude_tools", []) or [])
-        candidates -= set(ROLE_HARD_DENIES[role])
+
+        # Explicit mode is fail-closed: a Host tool must declare its security
+        # effects, and every effect must be allowed by the role.  Legacy mode
+        # intentionally ignores this metadata to preserve existing behavior.
+        if policy == CompletionPolicy.EXPLICIT_VALIDATION:
+            allowed_capabilities = ROLE_ALLOWED_CAPABILITIES[role]
+            candidates = {
+                name for name in candidates
+                if (declared := registry.get_capabilities(name))
+                and declared.issubset(allowed_capabilities)
+            }
 
         return ResolvedAgentPolicy(
             role=role,
@@ -82,14 +96,15 @@ class DefaultToolPolicy:
         tool_name: str,
         resolved_policy: ResolvedAgentPolicy,
     ) -> AuthorizationResult:
-        if tool_name in ROLE_HARD_DENIES[role]:
-            return AuthorizationResult(False, f"Tool '{tool_name}' is forbidden for role '{role.value}'")
         if tool_name not in resolved_policy.effective_tools:
-            return AuthorizationResult(False, f"Tool '{tool_name}' is not authorized for role '{role.value}'")
+            return AuthorizationResult(
+                False,
+                f"Tool '{tool_name}' is unclassified or its capabilities are forbidden for role '{role.value}'",
+            )
         return AuthorizationResult(True)
 
 
 __all__ = [
     "AuthorizationResult", "ResolvedAgentPolicy", "DefaultToolPolicy",
-    "PLANNER_TOOLS", "WORKER_TOOLS", "VALIDATOR_TOOLS", "ROLE_HARD_DENIES",
+    "PLANNER_TOOLS", "WORKER_TOOLS", "VALIDATOR_TOOLS", "ROLE_ALLOWED_CAPABILITIES",
 ]

agent/agent/tools/builtin/bash.py

@@ -158,7 +158,7 @@ def _kill_process_tree(pid: int) -> None:
         pass
 
 
-@tool(description="执行 bash 命令", hidden_params=["context"], groups=["system"])
+@tool(description="执行 bash 命令", hidden_params=["context"], groups=["system"], capabilities=["write"])
 async def bash_command(
     command: str,
     timeout: Optional[int] = None,

agent/agent/tools/builtin/context.py

@@ -16,6 +16,7 @@ from agent.tools import tool, ToolResult, ToolContext
     description="获取当前执行上下文，包括计划状态、焦点提醒、协作者信息等。当你感到困惑或需要回顾当前任务状态时调用。",
     hidden_params=["context"],
     groups=["core"],
+    capabilities=["read"],
 )
 async def get_current_context(
     context: ToolContext,

agent/agent/tools/builtin/file/edit.py

@@ -17,7 +17,7 @@ import re
 from agent.tools import tool, ToolResult, ToolContext
 
 
-@tool(description="编辑文件，使用精确字符串替换。支持多种智能匹配策略。", hidden_params=["context"], groups=["core"])
+@tool(description="编辑文件，使用精确字符串替换。支持多种智能匹配策略。", hidden_params=["context"], groups=["core"], capabilities=["write"])
 async def edit_file(
     file_path: str,
     old_string: str,

agent/agent/tools/builtin/file/glob.py

@@ -19,7 +19,7 @@ from agent.tools import tool, ToolResult, ToolContext
 LIMIT = 100  # 最大返回数量（参考 opencode glob.ts:35）
 
 
-@tool(description="使用 glob 模式匹配文件", hidden_params=["context"])
+@tool(description="使用 glob 模式匹配文件", hidden_params=["context"], capabilities=["read"])
 async def glob_files(
     pattern: str,
     path: Optional[str] = None,

agent/agent/tools/builtin/file/grep.py

@@ -21,7 +21,7 @@ LIMIT = 100  # 最大返回匹配数（参考 opencode grep.ts:107）
 MAX_LINE_LENGTH = 2000  # 最大行长度（参考 opencode grep.ts:10）
 
 
-@tool(description="在文件内容中搜索模式", hidden_params=["context"], groups=["core"])
+@tool(description="在文件内容中搜索模式", hidden_params=["context"], groups=["core"], capabilities=["read"])
 async def grep_content(
     pattern: str,
     path: Optional[str] = None,

agent/agent/tools/builtin/file/read.py

@@ -27,7 +27,7 @@ MAX_LINE_LENGTH = 2000
 MAX_BYTES = 50 * 1024  # 50KB
 
 
-@tool(description="读取单个文件内容，支持文本文件、图片、PDF 等多种格式，也支持 HTTP/HTTPS URL", hidden_params=["context"], groups=["core"])
+@tool(description="读取单个文件内容，支持文本文件、图片、PDF 等多种格式，也支持 HTTP/HTTPS URL", hidden_params=["context"], groups=["core"], capabilities=["read"])
 async def read_file(
     file_path: str,
     offset: int = 0,

agent/agent/tools/builtin/file/read_images.py

@@ -72,6 +72,7 @@ def _adaptive_layout(count: int) -> Tuple[int, int]:
         },
     },
     groups=["core"],
+    capabilities=["read"],
 )
 async def read_images(
     paths: List[str],

agent/agent/tools/builtin/file/write.py

@@ -16,7 +16,7 @@ import difflib
 from agent.tools import tool, ToolResult, ToolContext
 
 
-@tool(description="写入文件内容（创建新文件、覆盖现有文件或追加内容）", hidden_params=["context"], groups=["core"])
+@tool(description="写入文件内容（创建新文件、覆盖现有文件或追加内容）", hidden_params=["context"], groups=["core"], capabilities=["write"])
 async def write_file(
     file_path: str,
     content: str,

agent/agent/tools/builtin/file/write_json.py

@@ -9,7 +9,7 @@ from typing import Optional, Dict, Any
 from agent.tools import tool, ToolResult, ToolContext
 
 
-@tool(description="专门且唯一安全的 JSON 数据文件写入工具。传入 Python Dict/Object，自动为你生成格式化和转义无误的 JSON 文件。严禁使用普通的 write_file 写 JSON。参数 file_path 是文件绝对路径字符串，json_data 是要写入的原生 JSON 对象（直接传 dict，无需提前序列化）", hidden_params=["context"], groups=["core"])
+@tool(description="专门且唯一安全的 JSON 数据文件写入工具。传入 Python Dict/Object，自动为你生成格式化和转义无误的 JSON 文件。严禁使用普通的 write_file 写 JSON。参数 file_path 是文件绝对路径字符串，json_data 是要写入的原生 JSON 对象（直接传 dict，无需提前序列化）", hidden_params=["context"], groups=["core"], capabilities=["write"])
 async def write_json(
     file_path: str = "",
     json_data: dict = None,

agent/agent/tools/builtin/glob_tool.py

@@ -19,7 +19,7 @@ from agent.tools import tool, ToolResult, ToolContext
 LIMIT = 100  # 最大返回数量（参考 opencode glob.ts:35）
 
 
-@tool(description="使用 glob 模式匹配文件", hidden_params=["context"], groups=["core"])
+@tool(description="使用 glob 模式匹配文件", hidden_params=["context"], groups=["core"], capabilities=["read"])
 async def glob_files(
     pattern: str,
     path: Optional[str] = None,

agent/agent/tools/builtin/orchestration.py

@@ -23,7 +23,7 @@ def _coordinator(context: Dict[str, Any]) -> Any:
     return coordinator
 
 
-@tool(groups=["orchestration_planner"], hidden_params=["context"])
+@tool(groups=["orchestration_planner"], hidden_params=["context"], capabilities=["task_control"])
 async def task_plan(
     operation: str = "create",
     tasks: Optional[List[Dict[str, Any]]] = None,
@@ -56,7 +56,7 @@ async def task_plan(
     return json.dumps(result, ensure_ascii=False, indent=2)
 
 
-@tool(groups=["orchestration_planner"], hidden_params=["context"])
+@tool(groups=["orchestration_planner"], hidden_params=["context"], capabilities=["task_control"])
 async def dispatch_tasks(
     task_ids: List[str],
     worker_presets: Optional[List[str]] = None,
@@ -73,7 +73,7 @@ async def dispatch_tasks(
     return json.dumps([x.to_dict() for x in results], ensure_ascii=False, indent=2)
 
 
-@tool(groups=["orchestration_planner"], hidden_params=["context"])
+@tool(groups=["orchestration_planner"], hidden_params=["context"], capabilities=["task_control"])
 async def task_decide(
     task_id: str,
     action: str,
@@ -97,7 +97,7 @@ async def task_decide(
     return json.dumps(result, ensure_ascii=False, indent=2)
 
 
-@tool(groups=["orchestration_planner"], hidden_params=["context"])
+@tool(groups=["orchestration_planner"], hidden_params=["context"], capabilities=["task_control"])
 async def validate_attempt(
     task_id: str,
     attempt_id: str,
@@ -111,7 +111,7 @@ async def validate_attempt(
     return json.dumps(result.to_dict(), ensure_ascii=False, indent=2)
 
 
-@tool(groups=["orchestration_worker"], hidden_params=["context"])
+@tool(groups=["orchestration_worker"], hidden_params=["context"], capabilities=["attempt_submit"])
 async def submit_attempt(
     summary: str,
     artifact_refs: Optional[List[Dict[str, Any]]] = None,
@@ -136,7 +136,7 @@ async def submit_attempt(
     )
 
 
-@tool(groups=["orchestration_validator"], hidden_params=["context"])
+@tool(groups=["orchestration_validator"], hidden_params=["context"], capabilities=["validation_submit"])
 async def submit_validation(
     verdict: str,
     criterion_results: List[Dict[str, Any]],

agent/agent/tools/builtin/subagent.py

@@ -722,7 +722,7 @@ async def _run_remote_agent(
 
 # ===== 工具定义 =====
 
-@tool(description="创建子 Agent 执行任务（本地执行或路由到远端服务器，由 agent_type 决定）", hidden_params=["context"], groups=["core"])
+@tool(description="创建子 Agent 执行任务（本地执行或路由到远端服务器，由 agent_type 决定）", hidden_params=["context"], groups=["core"], capabilities=["agent_spawn"])
 async def agent(
     task: Union[str, List[str]],
     messages: Optional[Union[Messages, List[Messages]]] = None,
@@ -822,7 +822,7 @@ async def agent(
     )
 
 
-@tool(description="评估目标执行结果是否满足要求", hidden_params=["context"], groups=["core"])
+@tool(description="评估目标执行结果是否满足要求", hidden_params=["context"], groups=["core"], capabilities=["agent_spawn"])
 async def evaluate(
     messages: Optional[Messages] = None,
     target_goal_id: Optional[str] = None,

agent/agent/trace/goal_tool.py

@@ -108,7 +108,7 @@ async def inject_knowledge_for_goal(
 
 # ===== LLM 可调用的 goal 工具 =====
 
-@tool(description="管理执行计划，添加/完成/放弃目标，切换焦点", hidden_params=["context"], groups=["core"])
+@tool(description="管理执行计划，添加/完成/放弃目标，切换焦点", hidden_params=["context"], groups=["core"], capabilities=["task_control"])
 async def goal(
     add: Optional[str] = None,
     reason: Optional[str] = None,

agent/tests/test_runner_policy.py

@@ -3,7 +3,7 @@ import pytest
 from agent.core.presets import AgentPreset, register_preset
 from agent.core.runner import AgentRunner, RunConfig
 from agent.orchestration.models import AgentRole, CompletionPolicy
-from agent.tools.models import ToolResult
+from agent.tools.models import ToolCapability, ToolResult
 from agent.tools.registry import ToolRegistry
 from agent.trace.models import Trace
 
@@ -104,6 +104,138 @@ async def test_framework_context_overrides_host_spoofing():
     assert context["coordinator"] == "real"
 
 
+@pytest.mark.asyncio
+async def test_validator_capabilities_hide_and_reject_arbitrary_write_tool():
+    registry = ToolRegistry()
+    executed = False
+
+    async def db_modify():
+        nonlocal executed
+        executed = True
+        return "modified"
+
+    async def db_lookup():
+        return "row"
+
+    registry.register(
+        db_modify,
+        schema=_schema("db_modify"),
+        capabilities=[ToolCapability.WRITE],
+    )
+    registry.register(
+        db_lookup,
+        schema=_schema("db_lookup"),
+        capabilities=[ToolCapability.READ],
+    )
+    register_preset(
+        "test_validator_capabilities",
+        AgentPreset(
+            role=AgentRole.VALIDATOR,
+            allowed_tools=["db_modify", "db_lookup"],
+            max_iterations=5,
+            skills=[],
+        ),
+    )
+    runner = AgentRunner(tool_registry=registry, llm_call=lambda **_: None, task_coordinator=object())
+    config = RunConfig(
+        agent_type="test_validator_capabilities",
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        tools=["db_modify", "db_lookup"],
+    )
+    names = {
+        item["function"]["name"] for item in runner._get_run_tool_schemas(config)
+    }
+    assert names == {"db_lookup"}
+
+    trace = Trace(trace_id="validator", mode="agent", agent_role="validator", context={})
+    result = await runner._execute_authorized_tool(
+        "db_modify", {}, "forged", config, trace, None, 1
+    )
+    assert result["error"] == "unauthorized_tool"
+    assert executed is False
+
+
+@pytest.mark.asyncio
+async def test_worker_rejects_agent_spawn_capability_with_unrelated_name():
+    registry = ToolRegistry()
+    executed = False
+
+    async def spawn_helper():
+        nonlocal executed
+        executed = True
+        return "spawned"
+
+    registry.register(
+        spawn_helper,
+        schema=_schema("spawn_helper"),
+        capabilities=[ToolCapability.AGENT_SPAWN],
+    )
+    register_preset(
+        "test_worker_spawn_capability",
+        AgentPreset(
+            role=AgentRole.WORKER,
+            allowed_tools=["spawn_helper"],
+            max_iterations=5,
+            skills=[],
+        ),
+    )
+    runner = AgentRunner(tool_registry=registry, llm_call=lambda **_: None, task_coordinator=object())
+    config = RunConfig(
+        agent_type="test_worker_spawn_capability",
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        tools=["spawn_helper"],
+    )
+    assert runner._get_run_tool_schemas(config) == []
+    trace = Trace(trace_id="worker", mode="agent", agent_role="worker", context={})
+    result = await runner._execute_authorized_tool(
+        "spawn_helper", {}, "forged", config, trace, None, 1
+    )
+    assert result["error"] == "unauthorized_tool"
+    assert executed is False
+
+
+@pytest.mark.asyncio
+async def test_unclassified_tool_fails_closed_only_in_explicit_mode():
+    registry = ToolRegistry()
+    calls = 0
+
+    async def mystery_tool():
+        nonlocal calls
+        calls += 1
+        return "ok"
+
+    registry.register(mystery_tool, schema=_schema("mystery_tool"))
+    register_preset(
+        "test_unclassified_worker",
+        AgentPreset(
+            role=AgentRole.WORKER,
+            allowed_tools=["mystery_tool"],
+            max_iterations=5,
+            skills=[],
+        ),
+    )
+    runner = AgentRunner(tool_registry=registry, llm_call=lambda **_: None, task_coordinator=object())
+    explicit = RunConfig(
+        agent_type="test_unclassified_worker",
+        completion_policy=CompletionPolicy.EXPLICIT_VALIDATION,
+        tools=["mystery_tool"],
+    )
+    worker_trace = Trace(trace_id="worker", mode="agent", agent_role="worker", context={})
+    rejected = await runner._execute_authorized_tool(
+        "mystery_tool", {}, "explicit", explicit, worker_trace, None, 1
+    )
+    assert rejected["error"] == "unauthorized_tool"
+    assert calls == 0
+
+    legacy = RunConfig(tools=["mystery_tool"], tool_groups=[])
+    legacy_trace = Trace(trace_id="legacy", mode="agent", agent_role="legacy", context={})
+    assert {x["function"]["name"] for x in runner._get_run_tool_schemas(legacy)} == {"mystery_tool"}
+    assert await runner._execute_authorized_tool(
+        "mystery_tool", {}, "legacy", legacy, legacy_trace, None, 1
+    ) == "ok"
+    assert calls == 1
+
+
 def _disabled_knowledge():
     from agent.tools.builtin.knowledge import KnowledgeConfig
     return KnowledgeConfig(enable_extraction=False, enable_completion_extraction=False, enable_injection=False)