工具：提供任务规划、派发、决策和结构化提交入口

注册 task_plan、dispatch_tasks、task_decide、validate_attempt 供 Planner 使用，并提供 terminal 类型的 submit_attempt、submit_validation。工具只通过受保护上下文访问 Coordinator；同时将 Feishu 和 IM 内建工具改为可选导入，确保框架核心在未安装服务器扩展时仍可加载。

agent/agent/tools/builtin/__init__.py

@@ -32,11 +32,26 @@ from agent.tools.builtin.content import (
     extract_video_clip, import_content,
 )
 from agent.trace.goal_tool import goal
+from agent.tools.builtin.orchestration import (
+    task_plan,
+    dispatch_tasks,
+    task_decide,
+    validate_attempt,
+    submit_attempt,
+    submit_validation,
+)
 # 导入浏览器工具以触发注册 (因 P1 流水线不需要，且加载缓慢，暂时全局屏蔽)
 # import agent.tools.builtin.browser  # noqa: F401
 
-import agent.tools.builtin.feishu
-import agent.tools.builtin.im
+try:
+    import agent.tools.builtin.feishu
+except ImportError:
+    pass  # optional feishu extra
+
+try:
+    import agent.tools.builtin.im
+except ImportError:
+    pass  # optional IM/server dependencies
 
 __all__ = [
     # 文件操作
@@ -80,6 +95,13 @@ __all__ = [
     "import_content",
     # Goal 管理
     "goal",
+    # Explicit task orchestration (role policy controls visibility)
+    "task_plan",
+    "dispatch_tasks",
+    "task_decide",
+    "validate_attempt",
+    "submit_attempt",
+    "submit_validation",
     # Memory & Knowledge 提取审核
     "knowledge_save_pending",  # 反思侧分支暂存（core 组默认可见）
     "dream",                    # memory-bearing Agent 整理长期记忆（memory 组）

agent/agent/tools/builtin/orchestration.py

@@ -0,0 +1,175 @@
+"""Agent tools for explicit task orchestration."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Dict, List, Optional
+
+from agent.orchestration.models import (
+    ArtifactRef,
+    AttemptSubmission,
+    CriterionResult,
+    DecisionAction,
+    ValidationVerdict,
+)
+from agent.tools import tool
+from agent.tools.models import ToolResult
+
+
+def _coordinator(context: Dict[str, Any]) -> Any:
+    coordinator = context.get("coordinator") if context else None
+    if coordinator is None:
+        raise RuntimeError("Orchestration is not wired")
+    return coordinator
+
+
+@tool(groups=["orchestration_planner"], hidden_params=["context"])
+async def task_plan(
+    operation: str = "create",
+    tasks: Optional[List[Dict[str, Any]]] = None,
+    parent_task_id: Optional[str] = None,
+    after_task_id: Optional[str] = None,
+    focus_task_id: Optional[str] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Create generic Tasks, add child/sibling Tasks, or change planner focus."""
+    context = context or {}
+    coordinator = _coordinator(context)
+    root_trace_id = context["root_trace_id"]
+    if operation == "focus":
+        if not focus_task_id:
+            raise ValueError("focus_task_id is required")
+        result = await coordinator.focus_task(root_trace_id, focus_task_id)
+    elif operation == "create":
+        placement = {"after_task_id": after_task_id, "focus": bool(focus_task_id)}
+        result = await coordinator.create_tasks(
+            root_trace_id,
+            tasks or [],
+            parent_task_id=parent_task_id,
+            placement=placement,
+            idempotency_key=context.get("tool_call_id"),
+        )
+        if focus_task_id:
+            await coordinator.focus_task(root_trace_id, focus_task_id)
+    else:
+        raise ValueError("operation must be create or focus")
+    return json.dumps(result, ensure_ascii=False, indent=2)
+
+
+@tool(groups=["orchestration_planner"], hidden_params=["context"])
+async def dispatch_tasks(
+    task_ids: List[str],
+    worker_presets: Optional[List[str]] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Run independent Worker -> Validator cycles for one or more Tasks."""
+    context = context or {}
+    results = await _coordinator(context).dispatch_tasks(
+        context["root_trace_id"],
+        task_ids,
+        worker_presets=worker_presets,
+        idempotency_key=context.get("tool_call_id"),
+    )
+    return json.dumps([x.to_dict() for x in results], ensure_ascii=False, indent=2)
+
+
+@tool(groups=["orchestration_planner"], hidden_params=["context"])
+async def task_decide(
+    task_id: str,
+    action: str,
+    reason: str,
+    validation_id: Optional[str] = None,
+    payload: Optional[Dict[str, Any]] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Record the Planner's accept/replan/block/cancel/supersede decision."""
+    context = context or {}
+    decision_payload = dict(payload or {})
+    decision_payload["reason"] = reason
+    result = await _coordinator(context).decide_task(
+        context["root_trace_id"],
+        task_id,
+        validation_id,
+        DecisionAction(action),
+        decision_payload,
+        context.get("tool_call_id"),
+    )
+    return json.dumps(result, ensure_ascii=False, indent=2)
+
+
+@tool(groups=["orchestration_planner"], hidden_params=["context"])
+async def validate_attempt(
+    task_id: str,
+    attempt_id: str,
+    context: Optional[Dict[str, Any]] = None,
+) -> str:
+    """Re-run a fresh validator after validation error/stopped/expired."""
+    context = context or {}
+    result = await _coordinator(context).revalidate_attempt(
+        context["root_trace_id"], task_id, attempt_id, context.get("tool_call_id")
+    )
+    return json.dumps(result.to_dict(), ensure_ascii=False, indent=2)
+
+
+@tool(groups=["orchestration_worker"], hidden_params=["context"])
+async def submit_attempt(
+    summary: str,
+    artifact_refs: Optional[List[Dict[str, Any]]] = None,
+    evidence_refs: Optional[List[Dict[str, Any]]] = None,
+    context: Optional[Dict[str, Any]] = None,
+) -> ToolResult:
+    """Submit one immutable Worker attempt and terminate the Worker trace."""
+    context = context or {}
+    submission = AttemptSubmission(
+        summary=summary,
+        artifact_refs=[ArtifactRef.from_dict(x) for x in artifact_refs or []],
+        evidence_refs=[ArtifactRef.from_dict(x) for x in evidence_refs or []],
+    )
+    result = await _coordinator(context).submit_attempt(context, submission)
+    result_json = json.dumps(result, ensure_ascii=False, sort_keys=True)
+    return ToolResult(
+        title="Attempt submitted",
+        output=result_json,
+        long_term_memory=f"Attempt {result['attempt_id']} submitted as snapshot {result['snapshot_id']}",
+        terminate_run=True,
+        result_summary=result_json,
+    )
+
+
+@tool(groups=["orchestration_validator"], hidden_params=["context"])
+async def submit_validation(
+    verdict: str,
+    criterion_results: List[Dict[str, Any]],
+    summary: str,
+    evidence_refs: Optional[List[Dict[str, Any]]] = None,
+    unverified_claims: Optional[List[str]] = None,
+    risks: Optional[List[str]] = None,
+    recommendation: str = "",
+    context: Optional[Dict[str, Any]] = None,
+) -> ToolResult:
+    """Submit one structured validation report and terminate the Validator trace."""
+    context = context or {}
+    result = await _coordinator(context).submit_validation(
+        context,
+        ValidationVerdict(verdict),
+        [CriterionResult.from_dict(x) for x in criterion_results],
+        summary,
+        [ArtifactRef.from_dict(x) for x in evidence_refs or []],
+        unverified_claims or [],
+        risks or [],
+        recommendation,
+    )
+    result_json = json.dumps(result, ensure_ascii=False, sort_keys=True)
+    return ToolResult(
+        title="Validation submitted",
+        output=result_json,
+        long_term_memory=f"Validation {result['validation_id']} submitted: {result['verdict']}",
+        terminate_run=True,
+        result_summary=result_json,
+    )
+
+
+__all__ = [
+    "task_plan", "dispatch_tasks", "task_decide", "validate_attempt",
+    "submit_attempt", "submit_validation",
+]