权限：增加 Planner、Worker、Validator 角色策略

为 AgentPreset 增加不可由调用上下文伪造的 role，并提供 planner、worker、validator 内置预设。实现显式模式下候选工具、预设白名单、排除项和角色硬拒绝的统一计算，同时新增最终追加的角色契约 Prompt，保证 Host Prompt 不能覆盖框架不变量。

agent/agent/core/presets.py

@@ -5,15 +5,20 @@ Agent Presets - Agent 类型预设配置
 用户可通过 .agent/presets.json 覆盖或添加预设。
 """
 
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Optional, List
 from pathlib import Path
 
+from agent.orchestration.models import AgentRole
+
 
 @dataclass
 class AgentPreset:
     """Agent 预设配置"""
 
+    # 安全角色；自定义 explicit-validation preset 必须显式继承一个角色
+    role: AgentRole = AgentRole.LEGACY
+
     # 工具权限
     allowed_tools: Optional[List[str]] = None  # None 表示允许全部
     denied_tools: Optional[List[str]] = None   # 黑名单
@@ -36,18 +41,21 @@ class AgentPreset:
 # 默认不预置 skills，项目按需在 presets.json 或 RunConfig.skills 中指定
 AGENT_PRESETS = {
     "default": AgentPreset(
+        role=AgentRole.LEGACY,
         allowed_tools=None,
         max_iterations=30,
         skills=[],
         description="默认 Agent，拥有全部工具权限",
     ),
     "delegate": AgentPreset(
+        role=AgentRole.LEGACY,
         allowed_tools=None,
         max_iterations=30,
         skills=[],
         description="委托子 Agent，拥有全部工具权限（由 agent 工具创建）",
     ),
     "explore": AgentPreset(
+        role=AgentRole.LEGACY,
         allowed_tools=["read", "glob", "grep", "list_files"],
         denied_tools=["write", "edit", "bash", "task"],
         max_iterations=15,
@@ -55,11 +63,37 @@ AGENT_PRESETS = {
         description="探索型 Agent，只读权限，用于代码分析",
     ),
     "evaluate": AgentPreset(
+        role=AgentRole.LEGACY,
         allowed_tools=["read_file", "grep_content", "glob_files", "goal"],
         max_iterations=10,
         skills=[],
         description="评估型 Agent，只读权限，用于结果评估",
     ),
+    "planner": AgentPreset(
+        role=AgentRole.PLANNER,
+        allowed_tools=["task_plan", "dispatch_tasks", "task_decide", "validate_attempt", "read_file", "glob_files", "grep_content", "get_current_context"],
+        denied_tools=["agent", "evaluate", "goal", "submit_attempt", "submit_validation"],
+        max_iterations=100,
+        skills=[],
+        description="唯一规划者：创建、派发任务并根据独立验证结果决策",
+    ),
+    "worker": AgentPreset(
+        role=AgentRole.WORKER,
+        allowed_tools=["read_file", "read_images", "glob_files", "grep_content", "write_file", "edit_file", "bash_command", "submit_attempt"],
+        denied_tools=["agent", "evaluate", "goal", "task_plan", "dispatch_tasks", "task_decide", "validate_attempt", "submit_validation"],
+        max_iterations=50,
+        skills=[],
+        description="任务执行者：不能规划或创建子 Agent，必须以 submit_attempt 结束",
+    ),
+    "validator": AgentPreset(
+        role=AgentRole.VALIDATOR,
+        allowed_tools=["read_file", "read_images", "glob_files", "grep_content", "submit_validation"],
+        denied_tools=["write_file", "edit_file", "bash_command", "agent", "evaluate", "goal", "task_plan", "dispatch_tasks", "task_decide", "validate_attempt", "submit_attempt"],
+        max_iterations=30,
+        temperature=0.0,
+        skills=[],
+        description="独立验证者：只读核验固定快照，必须以 submit_validation 结束",
+    ),
 }
 
 
@@ -145,5 +179,7 @@ def load_presets_from_json(json_path: str) -> None:
 
             cfg["system_prompt"] = system_prompt
 
+        if "role" in cfg:
+            cfg["role"] = AgentRole(cfg["role"])
         preset = AgentPreset(**cfg)
         register_preset(name, preset)

agent/agent/core/prompts/__init__.py

@@ -35,6 +35,13 @@ from agent.core.prompts.compression import (
     build_summary_header,
 )
 
+from agent.core.prompts.orchestration import (
+    PLANNER_ROLE_CONTRACT,
+    WORKER_ROLE_CONTRACT,
+    VALIDATOR_ROLE_CONTRACT,
+    ROLE_CONTRACTS,
+)
+
 __all__ = [
     # runner
     "DEFAULT_SYSTEM_PREFIX",
@@ -55,5 +62,11 @@ __all__ = [
     "COMPRESSION_EVAL_PROMPT_TEMPLATE",
     "SUMMARY_HEADER_TEMPLATE",
     "build_compression_eval_prompt",
+    "build_single_turn_prompt",
     "build_summary_header",
+    # orchestration
+    "PLANNER_ROLE_CONTRACT",
+    "WORKER_ROLE_CONTRACT",
+    "VALIDATOR_ROLE_CONTRACT",
+    "ROLE_CONTRACTS",
 ]

agent/agent/core/prompts/orchestration.py

@@ -0,0 +1,41 @@
+"""Non-overridable role contracts appended by AgentRunner."""
+
+PLANNER_ROLE_CONTRACT = """
+## Framework role contract: Planner
+
+You are the only planner. Create and dispatch tasks with task_plan and
+dispatch_tasks. Worker completion is only an attempt submission. A task is
+complete only after an independent passed validation and your task_decide
+action=accept. For failed, inconclusive, or validation errors, explicitly
+repair, retry, revise, split, revalidate, block, cancel, or supersede.
+""".strip()
+
+WORKER_ROLE_CONTRACT = """
+## Framework role contract: Worker
+
+Execute only the assigned immutable TaskSpec. Do not plan, modify the task
+tree, create another agent, or validate your own work. You must finish by
+calling submit_attempt exactly once with immutable artifact/evidence
+references. Free text without submit_attempt is not a successful delivery.
+""".strip()
+
+VALIDATOR_ROLE_CONTRACT = """
+## Framework role contract: Validator
+
+Independently evaluate the supplied fixed ArtifactSnapshot against every
+acceptance criterion. Use only authorized read-only evidence tools. Do not
+modify artifacts, goals, tasks, or external systems. You must finish by
+calling submit_validation exactly once. Use inconclusive when evidence is
+insufficient; never treat a runtime/reporting error as a failed verdict.
+""".strip()
+
+ROLE_CONTRACTS = {
+    "planner": PLANNER_ROLE_CONTRACT,
+    "worker": WORKER_ROLE_CONTRACT,
+    "validator": VALIDATOR_ROLE_CONTRACT,
+}
+
+__all__ = [
+    "PLANNER_ROLE_CONTRACT", "WORKER_ROLE_CONTRACT",
+    "VALIDATOR_ROLE_CONTRACT", "ROLE_CONTRACTS",
+]

agent/agent/orchestration/policy.py

@@ -0,0 +1,95 @@
+"""Role-aware tool policy for explicit-validation runs."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, FrozenSet, List, Optional
+
+from .models import AgentRole, CompletionPolicy
+
+
+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(),
+}
+
+
+@dataclass(frozen=True)
+class ResolvedAgentPolicy:
+    role: AgentRole
+    effective_tools: FrozenSet[str]
+    max_iterations: int
+    temperature: float
+    skills: Optional[List[str]]
+    system_prompt: Optional[str]
+    completion_policy: CompletionPolicy
+
+
+@dataclass(frozen=True)
+class AuthorizationResult:
+    allowed: bool
+    reason: Optional[str] = None
+
+
+class DefaultToolPolicy:
+    """Resolve schemas and enforce the same authorization at execution time."""
+
+    def resolve(self, config: Any, preset: Any, registry: Any) -> ResolvedAgentPolicy:
+        policy = CompletionPolicy(getattr(config, "completion_policy", CompletionPolicy.LEGACY_AUTO.value))
+        role = AgentRole(getattr(preset, "role", AgentRole.LEGACY.value))
+
+        if getattr(config, "tools", None) is not None:
+            candidates = set(config.tools)
+        elif getattr(config, "tool_groups", None) is not None:
+            candidates = set(registry.get_tool_names(groups=config.tool_groups))
+        else:
+            candidates = set(registry.get_tool_names())
+
+        if preset.allowed_tools is not None:
+            candidates &= set(preset.allowed_tools)
+        if preset.denied_tools:
+            candidates -= set(preset.denied_tools)
+        candidates -= set(getattr(config, "exclude_tools", []) or [])
+        candidates -= set(ROLE_HARD_DENIES[role])
+
+        return ResolvedAgentPolicy(
+            role=role,
+            effective_tools=frozenset(candidates),
+            max_iterations=min(int(config.max_iterations), int(preset.max_iterations)),
+            temperature=preset.temperature if preset.temperature is not None else float(config.temperature),
+            skills=config.skills if config.skills is not None else preset.skills,
+            system_prompt=config.system_prompt or preset.system_prompt,
+            completion_policy=policy,
+        )
+
+    def authorize(
+        self,
+        role: AgentRole,
+        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(True)
+
+
+__all__ = [
+    "AuthorizationResult", "ResolvedAgentPolicy", "DefaultToolPolicy",
+    "PLANNER_TOOLS", "WORKER_TOOLS", "VALIDATOR_TOOLS", "ROLE_HARD_DENIES",
+]