Agent/cyber_agent/tools/builtin/subagent.py

"""Sub-Agent 的创建、调度、汇合与旧评估工具。

``agent`` 由父 Agent 调用：Legacy 只创建一层，Recursive 还会执行深度、孩子数、
权限、预算和审核门禁；本地任务进程内运行，``remote_*`` 仍通过 KnowHub HTTP 路由。
``evaluate`` 保留给 Legacy 和旧的 Recursive revision 1；revision 2 使用框架管理的独立 Validator。
"""

import asyncio
import json
import os
from datetime import datetime
from typing import Any, Dict, List, Optional, Union

from pydantic import ValidationError

from cyber_agent.tools import tool
from cyber_agent.core.agent_mode import (
    AgentMode,
    AgentPolicy,
    RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY,
    RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
    RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
    apply_policy_to_context,
    assert_removed_config_absent,
    policy_from_context,
    validate_recursive_child_execution,
)
from cyber_agent.core.task_protocol import (
    TaskBrief,
    TaskReport,
    ensure_task_protocol,
    format_task_brief,
    new_task_protocol,
    pending_review_entry,
    protocol_error_report,
    replace_task_brief,
    stopped_task_report,
)
from cyber_agent.core.context_policy import (
    CONTEXT_ACCESS_KEY,
    ContextPolicyError,
    build_child_context_access,
    context_ref_descriptors,
    normalize_task_brief,
    persist_root_task_anchor,
    require_matching_root_task_anchor,
    task_briefs_match,
)
from cyber_agent.core.resource_budget import (
    ResourceBudgetExceeded,
    ResourceBudgetStateError,
)
from cyber_agent.trace.models import Trace, Messages
from cyber_agent.trace.trace_id import generate_sub_trace_id
from cyber_agent.trace.goal_models import GoalTree
from cyber_agent.trace.websocket import broadcast_sub_trace_started, broadcast_sub_trace_completed


# ===== 远端路由常量 =====

REMOTE_PREFIX = "remote_"
# POC 阶段使用进程内锁保护“检查配额 -> 创建 Trace”。
# 多 worker 部署时需要换成跨进程的原子配额。
_LOCAL_AGENT_SPAWN_LOCK = asyncio.Lock()


def _knowhub_api() -> str:
    """运行时读取 KNOWHUB_API，避免 module-load 时 .env 尚未加载的情况"""
    return os.getenv("KNOWHUB_API", "http://localhost:9999").rstrip("/")


def _remote_agent_timeout() -> float:
    return float(os.getenv("REMOTE_AGENT_TIMEOUT", "600"))


# 兼容旧代码对 module-level 常量的引用（运行时值 = 首次 import 时的快照）
KNOWHUB_API = _knowhub_api()
REMOTE_AGENT_TIMEOUT = _remote_agent_timeout()


# ===== prompts =====

# ===== 评估任务 =====

EVALUATE_PROMPT_TEMPLATE = """# 评估任务

请评估以下任务的执行结果是否满足要求。

## 目标描述

{goal_description}

## 执行结果

{result_text}

## 输出格式

## 评估结论
[通过/不通过]

## 评估理由
[详细说明通过或不通过原因]

## 修改建议（如果不通过）
1. [建议1]
2. [建议2]
"""

# ===== 结果格式化 =====

DELEGATE_RESULT_HEADER = "## 委托任务完成\n"

DELEGATE_SAVED_KNOWLEDGE_HEADER = "**保存的知识** ({count} 条):"

DELEGATE_STATS_HEADER = "**执行统计**:"

EXPLORE_RESULT_HEADER = "## 探索结果\n"

EXPLORE_BRANCH_TEMPLATE = "### 方案 {branch_name}: {task}"

EXPLORE_STATUS_SUCCESS = "**状态**: ✓ 完成"

EXPLORE_STATUS_FAILED = "**状态**: ✗ 失败"

EXPLORE_STATUS_ERROR = "**状态**: ✗ 异常"

EXPLORE_SUMMARY_HEADER = "## 总结"

def build_evaluate_prompt(goal_description: str, result_text: str) -> str:
    return EVALUATE_PROMPT_TEMPLATE.format(
        goal_description=goal_description,
        result_text=result_text or "（无执行结果）",
    )


def _make_run_config(**kwargs):
    """延迟导入 RunConfig 以避免循环导入"""
    from cyber_agent.core.runner import RunConfig
    return RunConfig(**kwargs)


# ===== 辅助函数 =====

async def _update_collaborator(
    store, trace_id: str,
    name: str, sub_trace_id: str,
    status: str, summary: str = "",
) -> None:
    """
    更新 trace.context["collaborators"] 中的协作者信息。

    如果同名协作者已存在则更新，否则追加。
    """
    trace = await store.get_trace(trace_id)
    if not trace:
        return

    collaborators = trace.context.get("collaborators", [])

    # 查找已有记录
    existing = None
    for c in collaborators:
        if c.get("trace_id") == sub_trace_id:
            existing = c
            break

    if existing:
        existing["status"] = status
        if summary:
            existing["summary"] = summary
    else:
        collaborators.append({
            "name": name,
            "type": "agent",
            "trace_id": sub_trace_id,
            "status": status,
            "summary": summary,
        })

    trace.context["collaborators"] = collaborators
    await store.update_trace(trace_id, context=trace.context)


async def _update_goal_start(
    store, trace_id: str, goal_id: str, mode: str,
    sub_trace_ids: List[Dict[str, str]],
    *,
    accumulate_sub_trace_ids: bool,
    status: str = "in_progress",
) -> None:
    """标记 Goal 开始执行"""
    if not goal_id:
        return

    next_entries = sub_trace_ids
    if accumulate_sub_trace_ids:
        tree = await store.get_goal_tree(trace_id)
        goal = tree.find(goal_id) if tree else None
        existing_entries = goal.sub_trace_ids if goal and goal.sub_trace_ids else []
        merged_entries: Dict[str, Dict[str, str]] = {}
        for entry in [*existing_entries, *sub_trace_ids]:
            if isinstance(entry, str):
                merged_entries[entry] = {"trace_id": entry, "mission": entry}
            elif isinstance(entry, dict) and entry.get("trace_id"):
                merged_entries[entry["trace_id"]] = entry
        next_entries = list(merged_entries.values())

    await store.update_goal(
        trace_id, goal_id,
        type="agent_call",
        agent_call_mode=mode,
        status=status,
        sub_trace_ids=next_entries,
    )


async def _update_goal_complete(
    store, trace_id: str, goal_id: str,
    status: str, summary: str,
) -> None:
    """标记 Goal 完成"""
    if not goal_id:
        return
    await store.update_goal(
        trace_id, goal_id,
        status=status,
        summary=summary,
    )


async def _load_or_create_task_report(
    store,
    child_trace_id: str,
    fallback_reason: str,
    child_result_status: str,
    generated_at_sequence: int,
) -> TaskReport:
    """读取孩子已提交的报告，或为异常、停止生成框架报告。

    子任务执行结束后由报告汇合阶段调用，将异常和停止统一转成可审核的 ``TaskReport``。
    """
    child = await store.get_trace(child_trace_id)
    if not child:
        return protocol_error_report(child_trace_id, fallback_reason)
    state = ensure_task_protocol(child.context)
    report_data = state.get("task_report")
    execution_stopped = (
        child_result_status == "stopped"
        or child.status == "stopped"
    )
    execution_failed = (
        child_result_status != "completed"
        or child.status != "completed"
    )
    if report_data:
        try:
            report = TaskReport.model_validate(report_data)
            if report.child_trace_id != child_trace_id:
                if report.model_dump() not in state["report_history"]:
                    state["report_history"].append(report.model_dump())
                fallback_reason = (
                    "Child persisted a TaskReport whose child_trace_id does not "
                    "match its Trace"
                )
            elif execution_stopped:
                if report.outcome in {"failed", "protocol_error"}:
                    return report
                if report.model_dump() not in state["report_history"]:
                    state["report_history"].append(report.model_dump())
                fallback_reason = "Child Agent execution was stopped"
            elif not execution_failed or report.outcome in {"failed", "protocol_error"}:
                return report
            else:
                if report.model_dump() not in state["report_history"]:
                    state["report_history"].append(report.model_dump())
                fallback_reason = (
                    f"Child execution ended as {child_result_status} after submitting "
                    f"a {report.outcome} TaskReport: {fallback_reason}"
                )
        except ValidationError:
            fallback_reason = "Child persisted an invalid TaskReport"
    report = (
        stopped_task_report(child_trace_id, fallback_reason)
        if execution_stopped
        else protocol_error_report(child_trace_id, fallback_reason)
    )
    state["task_report"] = report.model_dump()
    state["task_report_submitted_at_sequence"] = generated_at_sequence
    state["task_report_validation"] = None
    await store.update_trace(child_trace_id, context=child.context)
    return report


async def _record_pending_task_reports(
    store,
    runner,
    parent_trace: Trace,
    goal_id: str | None,
    child_results: List[tuple[str, Dict[str, Any]]],
    received_at_sequence: int,
) -> List[TaskReport]:
    """验收一批子 Agent 结果并写入父 Trace 的待审核队列。

    ``_run_agents`` 汇合孩子后按输入顺序调用，必要时触发 Validator 并更新 Goal 状态。
    """
    state = ensure_task_protocol(parent_trace.context)
    reports: List[TaskReport] = []
    for child_trace_id, result in child_results:
        reason = result.get("error") or result.get("summary") or "Child ended without TaskReport"
        report = await _load_or_create_task_report(
            store,
            child_trace_id,
            reason,
            result.get("status", "unknown"),
            received_at_sequence,
        )
        child_trace = await store.get_trace(child_trace_id)
        if not child_trace:
            raise ValueError(f"Child Trace not found for validation: {child_trace_id}")
        child_state = ensure_task_protocol(child_trace.context)
        task_brief = child_state.get("task_brief")
        report_payload = report.model_dump()
        if report.outcome in {"satisfied", "partial"}:
            validation_run = await runner.validate_recursive_trace(
                child_trace_id,
                scope="task",
                task_brief=task_brief,
                task_report=report_payload,
            )
        else:
            is_protocol_error = report.outcome == "protocol_error"
            validation_run = await runner.validate_recursive_trace(
                child_trace_id,
                scope="task",
                task_brief=task_brief,
                task_report=report_payload,
                deterministic_failure={
                    "outcome": "error" if is_protocol_error else "failed",
                    "reason": report.summary,
                    "issues": report.remaining_issues or [report.summary],
                    "retry_from": None if is_protocol_error else "task_definition",
                },
            )
        validation_result = validation_run.result
        state["pending_reviews"][child_trace_id] = pending_review_entry(
            goal_id=goal_id,
            report=report,
            validation_result=validation_result.model_dump(),
            received_at_sequence=received_at_sequence,
        )
        reports.append(report)
    await store.update_trace(parent_trace.trace_id, context=parent_trace.context)
    if goal_id:
        await store.update_goal(
            parent_trace.trace_id,
            goal_id,
            cascade_completion=False,
            status="pending_review",
        )
    return reports


def _project_goal_status(context: dict, goal_id: str | None, status: str) -> None:
    tree = context.get("goal_tree")
    goal = tree.find(goal_id) if tree and goal_id else None
    if goal:
        goal.status = status


def _aggregate_stats(results: List[Dict[str, Any]]) -> Dict[str, Any]:
    """聚合多个结果的统计信息"""
    total_messages = 0
    total_tokens = 0
    total_cost = 0.0

    for result in results:
        if isinstance(result, dict) and "stats" in result:
            stats = result["stats"]
            total_messages += stats.get("total_messages", 0)
            total_tokens += stats.get("total_tokens", 0)
            total_cost += stats.get("total_cost", 0.0)

    return {
        "total_messages": total_messages,
        "total_tokens": total_tokens,
        "total_cost": total_cost
    }


async def _stopped_child_result(store, runner, trace_id: str) -> Dict[str, Any]:
    """停止尚未启动的预创建孩子，不产生任何模型调用或消息统计。"""
    await store.update_trace(
        trace_id,
        status="stopped",
        completed_at=datetime.now(),
    )
    event = getattr(runner, "_cancel_events", {}).get(trace_id)
    release_trace = getattr(runner, "release_recursive_trace", None)
    if release_trace:
        release_trace(trace_id, event)
    return {
        "status": "stopped",
        "summary": "Agent execution stopped.",
        "error": "Child Agent execution was stopped",
        "saved_knowledge_ids": [],
        "stats": {"total_messages": 0, "total_tokens": 0, "total_cost": 0.0},
    }


async def _execute_child_spec(
    spec: Dict[str, Any],
    runner,
    store,
    semaphore: Optional[asyncio.Semaphore] = None,
) -> Dict[str, Any]:
    """按调度规格延迟启动一个子 Agent。

    ``_run_agents`` 串行执行或在并行信号量内调用；排队期间停止则不进入 Runner。
    """
    trace_id = spec["trace_id"]

    async def execute() -> Dict[str, Any]:
        is_cancelled = getattr(runner, "is_cancel_requested", lambda _tid: False)
        if spec["recursive"] and is_cancelled(trace_id):
            return await _stopped_child_result(store, runner, trace_id)
        return await runner.run_result(
            messages=spec["messages"],
            config=spec["config"],
            on_event=spec["on_event"],
        )

    if semaphore is None:
        return await execute()

    is_cancelled = getattr(runner, "is_cancel_requested", lambda _tid: False)
    if spec["recursive"] and is_cancelled(trace_id):
        return await _stopped_child_result(store, runner, trace_id)
    async with semaphore:
        return await execute()


def _get_recursive_parent_capabilities(context: dict) -> Optional[set[str]]:
    """从隐藏 Tool Context 读取并校验父 Agent 的冻结权限快照。

    ``_run_agents`` 和子级权限计算共用；快照缺失或格式不合法时 Recursive 拒绝创建孩子。
    """
    runner = context.get("runner")
    capabilities = context.get(RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY)
    if (
        not runner
        or not hasattr(runner, "tools")
        or not hasattr(runner.tools, "get_tool_names")
        or not isinstance(capabilities, list)
        or any(not isinstance(name, str) for name in capabilities)
    ):
        return None
    return set(runner.tools.get_tool_names()) & set(capabilities)


def _get_allowed_tools(
    context: dict,
    agent_depth: int,
    policy: AgentPolicy,
) -> Optional[List[str]]:
    """按父级能力、子级固定规则和当前深度计算有效工具集合。

    创建子 Runner 前调用；Recursive 权限只能逐层收紧，最深层会移除 ``agent``。
    """
    runner = context.get("runner")
    if runner and hasattr(runner, "tools") and hasattr(runner.tools, "get_tool_names"):
        registered_tools = set(runner.tools.get_tool_names())
        if policy.mode is AgentMode.RECURSIVE:
            allowed_tools = _get_recursive_parent_capabilities(context)
            if allowed_tools is None:
                return None
        else:
            # Legacy 保留原行为：从全局 Registry 取工具全集。
            allowed_tools = registered_tools

        blocked_tools = {"evaluate", "bash_command"}
        if not policy.requires_task_protocol:
            blocked_tools.update({
                "submit_task_report",
                "review_task_result",
                "read_context_ref",
            })
        if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
            blocked_tools.add("agent")
        return sorted(allowed_tools - blocked_tools)
    return None


async def _resolve_trace_lineage(store, trace: Trace) -> tuple[int, str]:
    """返回 Trace 的 (agent_depth, root_trace_id)。

    新 Trace 直接读 context；旧 Trace 缺少字段时沿 parent_trace_id
    回溯，不依赖 Trace ID 字符串格式。
    """
    depth = trace.context.get("agent_depth")
    root_trace_id = trace.context.get("root_trace_id")
    if isinstance(depth, int) and depth >= 0 and isinstance(root_trace_id, str):
        return depth, root_trace_id

    depth = 0
    current = trace
    root_trace_id = trace.trace_id
    visited = {trace.trace_id}
    while current.parent_trace_id:
        parent_id = current.parent_trace_id
        if parent_id in visited:
            break
        visited.add(parent_id)
        parent = await store.get_trace(parent_id)
        if not parent:
            break
        depth += 1
        root_trace_id = parent.trace_id
        current = parent

    return depth, root_trace_id


async def _load_root_task_anchor(store, trace: Trace):
    """从权威根 Trace读取 Anchor，并校验当前节点持有相同副本。"""
    root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
    root = trace if root_trace_id == trace.trace_id else await store.get_trace(root_trace_id)
    if not root:
        raise ContextPolicyError(f"Recursive root Trace not found: {root_trace_id}")
    anchor = require_matching_root_task_anchor(root.context, trace.context)
    return root, anchor


def _format_single_result(result: Dict[str, Any], sub_trace_id: str, continued: bool) -> Dict[str, Any]:
    """格式化单任务（delegate）结果"""
    lines = [DELEGATE_RESULT_HEADER]
    summary = result.get("summary", "")
    if summary:
        lines.append(summary)
        lines.append("")

    # 添加保存的知识 ID
    saved_knowledge_ids = result.get("saved_knowledge_ids", [])
    if saved_knowledge_ids:
        lines.append("---\n")
        lines.append(DELEGATE_SAVED_KNOWLEDGE_HEADER.format(count=len(saved_knowledge_ids)))
        for kid in saved_knowledge_ids:
            lines.append(f"- {kid}")
        lines.append("")

    lines.append("---\n")
    lines.append(DELEGATE_STATS_HEADER)
    stats = result.get("stats", {})
    if stats:
        lines.append(f"- 消息数: {stats.get('total_messages', 0)}")
        lines.append(f"- Tokens: {stats.get('total_tokens', 0)}")
        lines.append(f"- 成本: ${stats.get('total_cost', 0.0):.4f}")
    formatted_summary = "\n".join(lines)

    return {
        "mode": "delegate",
        "sub_trace_id": sub_trace_id,
        "continue_from": continued,
        "saved_knowledge_ids": saved_knowledge_ids,  # 传递给父 agent
        **result,
        "summary": formatted_summary,
    }


def _format_multi_result(
    tasks: List[str], results: List[Dict[str, Any]], sub_trace_ids: List[Dict]
) -> Dict[str, Any]:
    """格式化多任务（explore）聚合结果"""
    lines = [EXPLORE_RESULT_HEADER]
    successful = 0
    failed = 0
    total_tokens = 0
    total_cost = 0.0

    for i, (task_item, result) in enumerate(zip(tasks, results)):
        branch_name = chr(ord('A') + i)
        lines.append(EXPLORE_BRANCH_TEMPLATE.format(branch_name=branch_name, task=task_item))

        if isinstance(result, dict):
            status = result.get("status", "unknown")
            if status == "completed":
                lines.append(EXPLORE_STATUS_SUCCESS)
                successful += 1
            else:
                lines.append(EXPLORE_STATUS_FAILED)
                failed += 1

            summary = result.get("summary", "")
            if summary:
                lines.append(f"**摘要**: {summary[:200]}...")

            stats = result.get("stats", {})
            if stats:
                messages = stats.get("total_messages", 0)
                tokens = stats.get("total_tokens", 0)
                cost = stats.get("total_cost", 0.0)
                lines.append(f"**统计**: {messages} messages, {tokens} tokens, ${cost:.4f}")
                total_tokens += tokens
                total_cost += cost
        else:
            lines.append(EXPLORE_STATUS_ERROR)
            failed += 1

        lines.append("")

    lines.append("---\n")
    lines.append(EXPLORE_SUMMARY_HEADER)
    lines.append(f"- 总分支数: {len(tasks)}")
    lines.append(f"- 成功: {successful}")
    lines.append(f"- 失败: {failed}")
    lines.append(f"- 总 tokens: {total_tokens}")
    lines.append(f"- 总成本: ${total_cost:.4f}")

    aggregated_summary = "\n".join(lines)
    overall_status = "completed" if successful > 0 else "failed"

    return {
        "mode": "explore",
        "status": overall_status,
        "summary": aggregated_summary,
        "sub_trace_ids": sub_trace_ids,
        "tasks": tasks,
        "stats": _aggregate_stats(results),
    }


async def _get_goal_description(store, trace_id: str, goal_id: str) -> str:
    """从 GoalTree 获取目标描述"""
    if not goal_id:
        return ""
    goal_tree = await store.get_goal_tree(trace_id)
    if goal_tree:
        target_goal = goal_tree.find(goal_id)
        if target_goal:
            return target_goal.description
    return f"Goal {goal_id}"


def _build_evaluate_prompt(goal_description: str, messages: Optional[Messages]) -> str:
    """
    构建评估 prompt。

    Args:
        goal_description: 代码从 GoalTree 注入的目标描述
        messages: 模型提供的消息（执行结果+上下文）
    """
    # 从 messages 提取文本内容
    result_text = ""
    if messages:
        parts = []
        for msg in messages:
            content = msg.get("content", "")
            if isinstance(content, str):
                parts.append(content)
            elif isinstance(content, list):
                # 多模态内容，提取文本部分
                for item in content:
                    if isinstance(item, dict) and item.get("type") == "text":
                        parts.append(item.get("text", ""))
        result_text = "\n".join(parts)

    return build_evaluate_prompt(goal_description, result_text)


def _make_event_printer(label: str):
    """
    创建子 Agent 执行过程打印函数。

    当父 runner.debug=True 时，传给 run_result(on_event=...)，
    实时输出子 Agent 的工具调用和助手消息。
    """
    prefix = f"  [{label}]"

    def on_event(item):
        from cyber_agent.trace.models import Trace, Message
        if isinstance(item, Message):
            if item.role == "assistant":
                content = item.content
                if isinstance(content, dict):
                    text = content.get("text", "")
                    tool_calls = content.get("tool_calls")
                    if text:
                        preview = text[:120] + "..." if len(text) > 120 else text
                        print(f"{prefix} {preview}")
                    if tool_calls:
                        for tc in tool_calls:
                            name = tc.get("function", {}).get("name", "unknown")
                            print(f"{prefix} 🛠️  {name}")
            elif item.role == "tool":
                content = item.content
                if isinstance(content, dict):
                    name = content.get("tool_name", "unknown")
                    desc = item.description or ""
                    desc_short = (desc[:60] + "...") if len(desc) > 60 else desc
                    suffix = f": {desc_short}" if desc_short else ""
                    print(f"{prefix} ✅ {name}{suffix}")
        elif isinstance(item, Trace):
            if item.status == "completed":
                print(f"{prefix} ✓ 完成")
            elif item.status == "failed":
                err = (item.error_message or "")[:80]
                print(f"{prefix} ✗ 失败: {err}")

    return on_event


def _make_interactive_handler(runner, sub_trace_id: str, parent_trace_id: str, debug_printer=None):
    """
    创建支持 stdin 交互检查的 on_event 回调。

    在每个子 Agent 事件触发时检查 stdin，检测到暂停/退出信号后
    通过 cancel_event.set() 停止子 agent 和父 agent 的执行。
    """
    def on_event(item):
        # 先执行 debug 打印
        if debug_printer:
            debug_printer(item)

        # 检查 stdin
        check_fn = getattr(runner, 'stdin_check', None)
        if not check_fn:
            return
        cmd = check_fn()
        if cmd in ('pause', 'quit'):
            request_stop = getattr(runner, "request_stop", None)
            if request_stop:
                request_stop(parent_trace_id)
            else:
                for tid in (sub_trace_id, parent_trace_id):
                    ev = runner._cancel_events.get(tid)
                    if ev:
                        ev.set()

    return on_event


# ===== 统一内部执行函数 =====

async def _run_agents(
    tasks: List[str],
    per_agent_msgs: List[Messages],
    continue_from: Optional[str],
    store, trace_id: str, goal_id: str, runner, context: dict,
    agent_type: Optional[str] = None,
    skills: Optional[List[str]] = None,
    task_briefs: Optional[List[TaskBrief]] = None,
) -> Dict[str, Any]:
    """
    本地 Sub-Agent 的统一创建、调度和结果汇合入口。

    ``agent`` 完成参数归一化后调用；此处执行 Spawn Guard、权限交集、串并行调度和
    报告汇合，并始终使用父 Trace 已持久化的 ``AgentPolicy``。
    """
    single = len(tasks) == 1
    parent_trace = await store.get_trace(trace_id)
    if not parent_trace:
        return {"status": "failed", "error": f"Parent trace not found: {trace_id}"}
    try:
        policy = policy_from_context(parent_trace.context)
    except ValueError as exc:
        return {"status": "failed", "error": str(exc)}
    root_task_anchor = None
    if policy.requires_task_protocol:
        try:
            _, root_task_anchor = await _load_root_task_anchor(
                store,
                parent_trace,
            )
        except ContextPolicyError as exc:
            return {"status": "failed", "error": str(exc)}

    if (
        policy.mode is AgentMode.RECURSIVE
        and _get_recursive_parent_capabilities(context) is None
    ):
        return {
            "status": "failed",
            "error": "Recursive agent capability context is missing or invalid",
        }

    child_execution_mode = "parallel"
    max_parallel_children = 2
    if policy.mode is AgentMode.RECURSIVE:
        try:
            child_execution_mode, max_parallel_children = (
                validate_recursive_child_execution(
                    context.get(
                        RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
                        "sequential",
                    ),
                    context.get(
                        RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
                        2,
                    ),
                )
            )
        except ValueError as exc:
            return {"status": "failed", "error": str(exc)}

    protocol_state = None
    approved_action = None
    if policy.requires_task_protocol:
        protocol_state = ensure_task_protocol(parent_trace.context)
        parent_task_brief = protocol_state.get("task_brief")
        if goal_id:
            parent_tree = await store.get_goal_tree(trace_id)
            parent_goal = parent_tree.find(goal_id) if parent_tree else None
            if parent_goal and parent_goal.status in {"completed", "failed", "abandoned"}:
                return {
                    "status": "failed",
                    "error": "Cannot create a child from a terminal Goal",
                }
        if protocol_state["pending_reviews"]:
            return {
                "status": "failed",
                "error": "Review pending child TaskReports before creating another child",
            }
        if not task_briefs or len(task_briefs) != len(tasks):
            return {"status": "failed", "error": "Recursive revision 2 requires task_brief"}
        if protocol_state["next_actions"]:
            approved_action = protocol_state["next_actions"][0]
            decision = approved_action["decision"]
            expected_brief = approved_action.get("task_brief")
            if decision == "REVISE_CHILD":
                if continue_from != approved_action["child_trace_id"] or len(tasks) != 1:
                    return {
                        "status": "failed",
                        "error": "REVISE_CHILD requires continue_from for the reviewed child",
                    }
                try:
                    matches_approved = not expected_brief or task_briefs_match(
                        task_briefs[0],
                        expected_brief,
                        parent_task_brief=parent_task_brief,
                        root_task_anchor=root_task_anchor,
                    )
                except ContextPolicyError as exc:
                    return {
                        "status": "failed",
                        "error": f"Approved Recursive TaskBrief is invalid; recreate the trace: {exc}",
                    }
                if not matches_approved:
                    return {
                        "status": "failed",
                        "error": "task_brief does not match the approved revision",
                    }
            else:
                if continue_from or len(tasks) != 1:
                    return {
                        "status": "failed",
                        "error": f"{decision} requires exactly one new approved child",
                    }
                try:
                    matches_approved = bool(expected_brief) and task_briefs_match(
                        task_briefs[0],
                        expected_brief,
                        parent_task_brief=parent_task_brief,
                        root_task_anchor=root_task_anchor,
                    )
                except ContextPolicyError as exc:
                    return {
                        "status": "failed",
                        "error": f"Approved Recursive TaskBrief is invalid; recreate the trace: {exc}",
                    }
                if not matches_approved:
                    return {
                        "status": "failed",
                        "error": "task_brief does not match the approved next task",
                    }
        elif continue_from:
            return {
                "status": "failed",
                "error": "continue_from requires a pending REVISE_CHILD action",
            }

    # continue_from: 复用已有 trace（仅 single）
    sub_trace_id = None
    continued = False
    goal_started = False
    child_records: List[Dict[str, Any]] = []
    all_sub_trace_ids: List[Dict[str, str]] = []
    created_trace_ids: list[str] = []

    async def fail_created_children(error: Exception) -> None:
        """将已创建但未进入执行阶段的 Recursive 孩子收敛为失败。"""
        for created_trace_id in created_trace_ids:
            created = await store.get_trace(created_trace_id)
            if created and created.status == "running":
                await store.update_trace(
                    created_trace_id,
                    status="failed",
                    error_message=f"Child initialization failed: {error}",
                    completed_at=datetime.now(),
                )

    async def run_initialization(operation):
        """执行一次运行前持久化，失败时统一清理已创建孩子。"""
        try:
            return await operation
        except Exception as exc:
            if policy.requires_task_protocol:
                await fail_created_children(exc)
            raise

    if single and continue_from:
        existing = await store.get_trace(continue_from)
        if not existing:
            return {"status": "failed", "error": f"Continue-from trace not found: {continue_from}"}
        if existing.parent_trace_id != trace_id:
            return {
                "status": "failed",
                "error": "continue_from must reference a direct child of the current trace",
            }
        if existing.uid != parent_trace.uid:
            return {
                "status": "failed",
                "error": "continue_from trace owner does not match the current trace owner",
            }
        if existing.context.get("created_by_tool") != "agent":
            return {
                "status": "failed",
                "error": "continue_from must reference a child created by the agent tool",
            }
        try:
            child_policy = policy_from_context(existing.context)
        except ValueError as exc:
            return {"status": "failed", "error": str(exc)}
        if (
            child_policy.mode is not policy.mode
            or child_policy.revision != policy.revision
        ):
            return {
                "status": "failed",
                "error": "continue_from trace Agent mode does not match the current trace",
            }

        sub_trace_id = continue_from
        continued = True
        goal_tree = await store.get_goal_tree(continue_from)
        mission = goal_tree.mission if goal_tree else tasks[0]
        parent_depth, expected_root_trace_id = await _resolve_trace_lineage(
            store, parent_trace
        )
        child_depth, root_trace_id = await _resolve_trace_lineage(store, existing)
        if (
            child_depth != parent_depth + 1
            or root_trace_id != expected_root_trace_id
        ):
            return {
                "status": "failed",
                "error": "continue_from trace lineage does not match the current trace",
            }
        if child_depth > policy.max_depth:
            return {
                "status": "failed",
                "error": (
                    "continue_from trace exceeds the persisted Agent mode depth: "
                    f"depth={child_depth}, max={policy.max_depth}"
                ),
            }
        all_sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
        child_context = apply_policy_to_context({
            **existing.context,
            "agent_depth": child_depth,
            "root_trace_id": root_trace_id,
        }, policy)
        child_records = [{
            "trace_id": sub_trace_id,
            "depth": child_depth,
            "context": child_context,
            "is_new": False,
        }]
        if task_briefs:
            assert root_task_anchor is not None
            child_state = ensure_task_protocol(child_context)
            normalized_dump = task_briefs[0].model_dump(mode="json")
            new_context_access = build_child_context_access(
                parent_context=parent_trace.context,
                parent_trace_id=parent_trace.trace_id,
                root_trace_id=root_trace_id,
                uid=parent_trace.uid,
                parent_task_state=protocol_state,
                child_task_brief=task_briefs[0],
                root_task_anchor=root_task_anchor,
                granted_at_sequence=(existing.last_sequence or 0) + 1,
            )
            if child_state.get("task_brief") != normalized_dump:
                replace_task_brief(
                    child_state,
                    task_briefs[0],
                    effective_at_sequence=(existing.last_sequence or 0) + 1,
                )
            child_context[CONTEXT_ACCESS_KEY] = new_context_access
            persist_root_task_anchor(child_context, root_task_anchor)
            existing.context = child_context
            await store.update_trace(existing.trace_id, context=existing.context)
    else:
        parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
        if parent_depth >= policy.max_depth:
            return {
                "status": "failed",
                "error": (
                    f"Local Sub-Agent depth limit reached: "
                    f"depth={parent_depth}, max={policy.max_depth}, "
                    f"mode={policy.mode.value}"
                ),
            }

        prepared_context_accesses: list[dict[str, Any]] = []
        if policy.requires_task_protocol:
            assert root_task_anchor is not None and task_briefs is not None
            try:
                prepared_context_accesses = [
                    build_child_context_access(
                        parent_context=parent_trace.context,
                        parent_trace_id=parent_trace.trace_id,
                        root_trace_id=root_trace_id,
                        uid=parent_trace.uid,
                        parent_task_state=protocol_state,
                        child_task_brief=brief,
                        root_task_anchor=root_task_anchor,
                    )
                    for brief in task_briefs
                ]
            except ContextPolicyError as exc:
                return {
                    "status": "failed",
                    "error": f"Invalid TaskBrief context references: {exc}",
                }

        async def create_child_traces() -> None:
            child_depth = parent_depth + 1
            for i, task_item in enumerate(tasks):
                task_label = (
                    task_briefs[i].objective
                    if task_briefs
                    else task_item
                )
                resolved_agent_type = agent_type or ("delegate" if single else "explore")
                suffix = "delegate" if single else f"explore-{i+1:03d}"
                stid = generate_sub_trace_id(trace_id, suffix)
                child_context = apply_policy_to_context({
                    "created_by_tool": "agent",
                    "agent_depth": child_depth,
                    "root_trace_id": root_trace_id,
                }, policy)
                if policy.requires_task_protocol:
                    assert root_task_anchor is not None
                    child_context["task_protocol"] = new_task_protocol(
                        task_briefs[i]
                    )
                    child_context[CONTEXT_ACCESS_KEY] = prepared_context_accesses[i]
                    persist_root_task_anchor(child_context, root_task_anchor)
                sub_trace = Trace(
                    trace_id=stid,
                    mode="agent",
                    task=task_label,
                    parent_trace_id=trace_id,
                    parent_goal_id=goal_id,
                    agent_type=resolved_agent_type,
                    uid=parent_trace.uid,
                    model=parent_trace.model,
                    status="running",
                    context=child_context,
                    created_at=datetime.now(),
                )
                await store.create_trace(sub_trace)
                # A successfully persisted Trace has consumed its Agent budget,
                # even if a later GoalTree write fails.
                created_trace_ids.append(stid)
                await store.update_goal_tree(stid, GoalTree(mission=task_label))
                all_sub_trace_ids.append({"trace_id": stid, "mission": task_label})
                child_records.append({
                    "trace_id": stid,
                    "depth": child_depth,
                    "context": child_context,
                    "is_new": True,
                    "agent_type": resolved_agent_type,
                })

            if single:
                nonlocal sub_trace_id
                sub_trace_id = child_records[0]["trace_id"]

        child_limit = policy.max_children_per_parent
        if child_limit is None:
            await create_child_traces()
        else:
            # Recursive 模式在单进程临界区内完成“计数 -> 创建”，
            # 保证并发批次不会共同突破六个直接孩子。
            async with _LOCAL_AGENT_SPAWN_LOCK:
                existing_children = await store.list_traces(
                    parent_trace_id=trace_id,
                    created_by_tool="agent",
                    limit=child_limit + 1,
                )
                requested_children = len(tasks)
                if len(existing_children) + requested_children > child_limit:
                    return {
                        "status": "failed",
                        "error": (
                            "Local child Agent limit exceeded: "
                            f"existing={len(existing_children)}, "
                            f"requested={requested_children}, max={child_limit}"
                        ),
                    }
                resolved_budget = None
                if policy.requires_task_protocol:
                    resolved_budget = await runner._resource_budget_for_trace(trace_id)
                reserved_agents = 0
                if resolved_budget is not None:
                    root_trace_id, budget = resolved_budget
                    if not runner.resource_budget:
                        return {
                            "status": "failed",
                            "error": "ResourceBudgetController is unavailable",
                        }
                    try:
                        await runner.resource_budget.reserve_agents(
                            root_trace_id,
                            budget,
                            requested_children,
                        )
                        reserved_agents = requested_children
                    except (ResourceBudgetExceeded, ResourceBudgetStateError) as exc:
                        return {"status": "failed", "error": str(exc)}
                try:
                    await create_child_traces()
                except Exception as exc:
                    created_count = len(created_trace_ids)
                    uncreated = max(0, reserved_agents - created_count)
                    if uncreated:
                        await runner.resource_budget.release_agents(
                            root_trace_id,
                            budget,
                            uncreated,
                        )
                    await fail_created_children(exc)
                    raise

        await run_initialization(
            _update_goal_start(
                store, trace_id, goal_id,
                "delegate" if single else "explore",
                all_sub_trace_ids,
                accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
                status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
            )
        )
        if policy.requires_task_protocol:
            _project_goal_status(context, goal_id, "waiting_children")
        goal_started = True

    if approved_action and protocol_state is not None:
        protocol_state["next_actions"].pop(0)
        protocol_state["protocol_correction_attempts"] = 0
        await run_initialization(
            store.update_trace(trace_id, context=parent_trace.context)
        )

    # 创建延迟执行规格。Trace 已按批次预创建，但不会提前创建 coroutine，
    # 因而排队孩子被停止时不会产生未 await coroutine 或模型调用。
    execution_specs = []
    for i, (task_item, msgs, child_record) in enumerate(
        zip(tasks, per_agent_msgs, child_records)
    ):
        cur_stid = child_record["trace_id"]
        child_depth = child_record["depth"]
        task_label = task_briefs[i].objective if task_briefs else task_item
        if child_record["is_new"]:
            await run_initialization(
                broadcast_sub_trace_started(
                    trace_id, cur_stid, goal_id or "",
                    child_record["agent_type"], task_label,
                )
            )

        # 注册为活跃协作者
        collab_name = task_label[:30] if single and not continued else (
            f"delegate-{cur_stid[:8]}" if single else f"explore-{i+1}"
        )
        await run_initialization(
            _update_collaborator(
                store, trace_id,
                name=collab_name, sub_trace_id=cur_stid,
                status="running", summary=task_label[:80],
            )
        )

        # 构建消息
        if policy.requires_task_protocol:
            assert task_briefs is not None and root_task_anchor is not None
            task_item = format_task_brief(
                task_briefs[i],
                available_context_refs=context_ref_descriptors(
                    child_record["context"]
                ),
            )
        agent_msgs = list(msgs) + [{"role": "user", "content": task_item}]
        allowed_tools = _get_allowed_tools(context, child_depth, policy)

        debug = getattr(runner, 'debug', False)
        agent_label = (agent_type or ("delegate" if single else f"explore-{i+1}"))
        debug_printer = _make_event_printer(agent_label) if debug else None
        # allowed_tools 已是当前深度的精确白名单。
        on_event = _make_interactive_handler(
            runner, cur_stid, trace_id, debug_printer=debug_printer
        )

        child_config = _make_run_config(
            trace_id=cur_stid,
            agent_type=agent_type or ("delegate" if single else "explore"),
            max_iterations=50,
            model=parent_trace.model if parent_trace else "gpt-4o",
            uid=parent_trace.uid if parent_trace else None,
            tools=allowed_tools,
            tool_groups=[],  # tools 是精确白名单，不再合并默认 core 组
            name=task_label[:50],
            skills=skills,
            knowledge=context.get("knowledge_config"),
            context=child_record["context"],
            child_execution_mode=child_execution_mode,
            max_parallel_children=max_parallel_children,
        )
        execution_specs.append({
            "index": i,
            "trace_id": cur_stid,
            "collaborator": collab_name,
            "messages": agent_msgs,
            "config": child_config,
            "on_event": on_event,
            "recursive": policy.mode is AgentMode.RECURSIVE,
        })

    # continue_from 不进入新建临界区，但仍需要恢复 Goal 的运行状态。
    if not goal_started:
        await run_initialization(
            _update_goal_start(
                store, trace_id, goal_id,
                "delegate" if single else "explore",
                all_sub_trace_ids,
                accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
                status=("waiting_children" if policy.requires_task_protocol else "in_progress"),
            )
        )
        if policy.requires_task_protocol:
            _project_goal_status(context, goal_id, "waiting_children")

    if policy.mode is AgentMode.RECURSIVE:
        register_child = getattr(runner, "register_recursive_child", None)
        if register_child:
            for spec in execution_specs:
                register_child(trace_id, spec["trace_id"])

    # 执行
    if single:
        # 单任务直接执行（带异常处理）
        spec = execution_specs[0]
        stid = spec["trace_id"]
        collab_name = spec["collaborator"]
        try:
            result = await _execute_child_spec(spec, runner, store)

            await broadcast_sub_trace_completed(
                trace_id, stid,
                result.get("status", "completed"),
                result.get("summary", ""),
                result.get("stats", {}),
            )
            await _update_collaborator(
                store, trace_id,
                name=collab_name, sub_trace_id=stid,
                status=result.get("status", "completed"),
                summary=result.get("summary", "")[:80],
            )

            formatted = _format_single_result(result, stid, continued)

            if policy.requires_task_protocol:
                reports = await _record_pending_task_reports(
                    store,
                    runner,
                    parent_trace,
                    goal_id,
                    [(stid, result)],
                    context.get("sequence", 0),
                )
                formatted["task_report"] = reports[0].model_dump()
                formatted["validation_result"] = ensure_task_protocol(
                    parent_trace.context
                )["pending_reviews"][stid]["validation_result"]
                _project_goal_status(context, goal_id, "pending_review")
            else:
                await _update_goal_complete(
                    store, trace_id, goal_id,
                    result.get("status", "completed"),
                    formatted["summary"],
                )
            return formatted

        except Exception as e:
            error_msg = str(e)
            await broadcast_sub_trace_completed(
                trace_id, stid, "failed", error_msg, {},
            )
            await _update_collaborator(
                store, trace_id,
                name=collab_name, sub_trace_id=stid,
                status="failed", summary=error_msg[:80],
            )
            failed_result = {
                "mode": "delegate",
                "status": "failed",
                "error": error_msg,
                "sub_trace_id": stid,
            }
            if policy.requires_task_protocol:
                reports = await _record_pending_task_reports(
                    store,
                    runner,
                    parent_trace,
                    goal_id,
                    [(stid, failed_result)],
                    context.get("sequence", 0),
                )
                failed_result["task_report"] = reports[0].model_dump()
                failed_result["validation_result"] = ensure_task_protocol(
                    parent_trace.context
                )["pending_reviews"][stid]["validation_result"]
                _project_goal_status(context, goal_id, "pending_review")
            else:
                await _update_goal_complete(
                    store, trace_id, goal_id,
                    "failed", f"委托任务失败: {error_msg}",
                )
            return failed_result
    else:
        async def execute_captured(
            spec: Dict[str, Any],
            semaphore: Optional[asyncio.Semaphore] = None,
        ):
            try:
                return await _execute_child_spec(spec, runner, store, semaphore)
            except Exception as exc:
                return exc

        if policy.mode is AgentMode.LEGACY:
            # Legacy 冻结旧行为：多任务整批并行，不读取新配置。
            raw_results = await asyncio.gather(*(
                execute_captured(spec) for spec in execution_specs
            ))
        elif child_execution_mode == "parallel":
            semaphore = asyncio.Semaphore(max_parallel_children)
            raw_results = await asyncio.gather(*(
                execute_captured(spec, semaphore) for spec in execution_specs
            ))
        else:
            raw_results = []
            for spec in execution_specs:
                raw_results.append(await execute_captured(spec))

        processed_results = []
        for idx, raw in enumerate(raw_results):
            spec = execution_specs[idx]
            stid = spec["trace_id"]
            collab_name = spec["collaborator"]
            if isinstance(raw, Exception):
                error_result = {
                    "status": "failed",
                    "summary": f"执行出错: {str(raw)}",
                    "stats": {"total_messages": 0, "total_tokens": 0, "total_cost": 0.0},
                }
                processed_results.append(error_result)
                await broadcast_sub_trace_completed(
                    trace_id, stid, "failed", str(raw), {},
                )
                await _update_collaborator(
                    store, trace_id,
                    name=collab_name, sub_trace_id=stid,
                    status="failed", summary=str(raw)[:80],
                )
            else:
                processed_results.append(raw)
                await broadcast_sub_trace_completed(
                    trace_id, stid,
                    raw.get("status", "completed"),
                    raw.get("summary", ""),
                    raw.get("stats", {}),
                )
                await _update_collaborator(
                    store, trace_id,
                    name=collab_name, sub_trace_id=stid,
                    status=raw.get("status", "completed"),
                    summary=raw.get("summary", "")[:80],
                )

        formatted = _format_multi_result(tasks, processed_results, all_sub_trace_ids)

        if policy.requires_task_protocol:
            reports = await _record_pending_task_reports(
                store,
                runner,
                parent_trace,
                goal_id,
                [
                    (entry["trace_id"], processed_results[index])
                    for index, entry in enumerate(all_sub_trace_ids)
                ],
                context.get("sequence", 0),
            )
            formatted["task_reports"] = [report.model_dump() for report in reports]
            pending = ensure_task_protocol(parent_trace.context)["pending_reviews"]
            formatted["validation_results"] = [
                pending[entry["trace_id"]]["validation_result"]
                for entry in all_sub_trace_ids
            ]
            _project_goal_status(context, goal_id, "pending_review")
        else:
            await _update_goal_complete(
                store, trace_id, goal_id,
                formatted["status"],
                formatted["summary"],
            )
        return formatted


# ===== 远端 Agent 路由 =====

async def _run_remote_agent(
    agent_type: str,
    task: str,
    messages: Optional[Messages],
    continue_from: Optional[str],
    skills: Optional[List[str]] = None,
) -> Dict[str, Any]:
    """
    通过 HTTP 调用 KnowHub 服务器上的远端 Agent。

    远端 Agent 的 tools / model / prompt 由服务器端 preset 决定。
    skills 由 caller 指定并原样发给服务器，最终权限由远端实现决定。
    """
    import httpx

    payload = {
        "agent_type": agent_type,
        "task": task,
        "messages": messages,
        "continue_from": continue_from,
        "skills": skills,
    }

    api_base = _knowhub_api()
    timeout = _remote_agent_timeout()
    try:
        async with httpx.AsyncClient(timeout=timeout) as client:
            response = await client.post(f"{api_base}/api/agent", json=payload)
            response.raise_for_status()
            result = response.json()

        return {
            "mode": "remote",
            "agent_type": agent_type,
            "sub_trace_id": result.get("sub_trace_id"),
            "status": result.get("status", "completed"),
            "summary": result.get("summary", ""),
            "stats": result.get("stats", {}),
            "error": result.get("error"),
        }

    except httpx.HTTPStatusError as e:
        return {
            "mode": "remote",
            "agent_type": agent_type,
            "status": "failed",
            "error": f"HTTP {e.response.status_code}: {e.response.text[:200]}",
        }
    except Exception as e:
        return {
            "mode": "remote",
            "agent_type": agent_type,
            "status": "failed",
            "error": f"远端调用失败: {type(e).__name__}: {e}",
        }


# ===== 工具定义 =====

@tool(description="创建子 Agent 执行任务（本地执行或路由到远端服务器，由 agent_type 决定）", hidden_params=["context"], groups=["core"])
async def agent(
    task: Optional[Union[str, List[str]]] = None,
    task_brief: Optional[Union[TaskBrief, List[TaskBrief]]] = None,
    messages: Optional[Union[Messages, List[Messages]]] = None,
    continue_from: Optional[str] = None,
    agent_type: Optional[str] = None,
    skills: Optional[List[str]] = None,
    context: Optional[dict] = None,
) -> Dict[str, Any]:
    """
    父 Agent 用来创建或续跑直属子 Agent 的公开工具。

    路由规则：
    - agent_type 以 "remote_" 开头：HTTP 调用 KnowHub 服务器的 /api/agent（仅单任务，无本地文件访问）
    - 否则本地执行：Legacy/Recursive revision 1 使用 ``task``；revision 2 使用 ``task_brief``

    Args:
        task: Legacy/Recursive revision 1 任务描述。
        task_brief: Recursive revision 2 的结构化任务说明。
        messages: 预置消息。1D 列表=所有 agent 共享；2D 列表=per-agent
        continue_from: 继续已有 trace（仅单任务）
        agent_type: 子 Agent 类型。带 "remote_" 前缀走远端；否则本地 preset
        skills: 指定本次调用使用的 skill 列表
                - 本地：附加到 system prompt
                - 远端：原样发送，由远端实现决定最终权限
        context: 框架自动注入的上下文
    """
    try:
        assert_removed_config_absent()
    except ValueError as exc:
        return {"status": "failed", "error": str(exc)}

    if isinstance(messages, str):
        try:
            messages = json.loads(messages)
        except json.JSONDecodeError as exc:
            return {"status": "failed", "error": f"Invalid messages JSON: {exc}"}
    if messages is not None:
        is_1d = isinstance(messages, list) and all(
            isinstance(message, dict) for message in messages
        )
        is_2d = isinstance(messages, list) and bool(messages) and all(
            isinstance(message_list, list)
            and all(isinstance(message, dict) for message in message_list)
            for message_list in messages
        )
        if not (is_1d or is_2d):
            return {
                "status": "failed",
                "error": "messages must be a 1D or 2D message list without mixed dimensions",
            }

    # 远端路由：agent_type 以 remote_ 开头
    if agent_type and agent_type.startswith(REMOTE_PREFIX):
        if not isinstance(task, str):
            return {"status": "failed", "error": "remote agent 只支持单任务 (task: str)"}
        # 归一化 messages：远端只接受 1D Messages 或 None
        remote_msgs: Optional[Messages] = None
        if messages is not None:
            if messages and isinstance(messages[0], list):
                return {"status": "failed", "error": "remote agent 不支持 2D messages (per-agent)"}
            remote_msgs = messages
        return await _run_remote_agent(
            agent_type=agent_type,
            task=task,
            messages=remote_msgs,
            continue_from=continue_from,
            skills=skills,
        )

    # 本地路径：需要 context
    if not context:
        return {"status": "failed", "error": "context is required"}

    store = context.get("store")
    trace_id = context.get("trace_id")
    goal_id = context.get("goal_id")
    runner = context.get("runner")

    missing = []
    if not store:
        missing.append("store")
    if not trace_id:
        missing.append("trace_id")
    if not runner:
        missing.append("runner")
    if missing:
        return {"status": "failed", "error": f"Missing required context: {', '.join(missing)}"}

    parent_trace = await store.get_trace(trace_id)
    if not parent_trace:
        return {"status": "failed", "error": f"Parent trace not found: {trace_id}"}
    try:
        policy = policy_from_context(parent_trace.context)
    except ValueError as exc:
        return {"status": "failed", "error": str(exc)}
    if policy.requires_task_protocol and messages is not None:
        return {
            "status": "failed",
            "error": "Recursive revision 2 does not accept messages; pass bounded context in task_brief",
        }

    parsed_task_briefs: Optional[List[TaskBrief]] = None
    if policy.requires_task_protocol:
        if task is not None or task_brief is None:
            return {
                "status": "failed",
                "error": "Recursive revision 2 requires task_brief and does not accept task",
            }
        if isinstance(task_brief, str):
            try:
                task_brief = json.loads(task_brief)
            except json.JSONDecodeError as exc:
                return {"status": "failed", "error": f"Invalid TaskBrief JSON: {exc}"}
        raw_briefs = task_brief if isinstance(task_brief, list) else [task_brief]
        try:
            _, root_task_anchor = await _load_root_task_anchor(
                store,
                parent_trace,
            )
            parent_state = ensure_task_protocol(parent_trace.context)
            parent_task_brief = parent_state.get("task_brief")
            parsed_task_briefs = [
                normalize_task_brief(
                    item,
                    parent_task_brief=parent_task_brief,
                    root_task_anchor=root_task_anchor,
                )
                for item in raw_briefs
            ]
        except (ContextPolicyError, ValidationError) as exc:
            return {"status": "failed", "error": f"Invalid TaskBrief: {exc}"}
        tasks = [brief.objective for brief in parsed_task_briefs]
        single = len(tasks) == 1
    else:
        if task_brief is not None:
            return {"status": "failed", "error": "task_brief requires Recursive revision 2"}
        if task is None:
            return {"status": "failed", "error": "task is required"}

        # 归一化 task → list（保留 Legacy 字符串 JSON 列表兼容）
        if isinstance(task, str):
            task_str = task.strip()
            if task_str.startswith("[") and task_str.endswith("]"):
                try:
                    parsed_task = json.loads(task_str)
                    if isinstance(parsed_task, list):
                        task = parsed_task
                except (json.JSONDecodeError, TypeError):
                    pass
        single = isinstance(task, str)
        tasks = [task] if single else task

    if not tasks:
        return {"status": "failed", "error": "task is required"}

    # 归一化 messages → List[Messages]（per-agent）
    if messages is None:
        per_agent_msgs: List[Messages] = [[] for _ in tasks]
    elif messages and isinstance(messages[0], list):
        if len(messages) != len(tasks):
            return {
                "status": "failed",
                "error": (
                    "2D messages must contain exactly one message list per task: "
                    f"tasks={len(tasks)}, messages={len(messages)}"
                ),
            }
        per_agent_msgs = messages  # 2D: per-agent
    else:
        per_agent_msgs = [messages] * len(tasks)  # 1D: 共享

    if continue_from and not single:
        return {"status": "failed", "error": "continue_from requires single task"}

    return await _run_agents(
        tasks, per_agent_msgs, continue_from,
        store, trace_id, goal_id, runner, context,
        agent_type=agent_type,
        skills=skills,
        task_briefs=parsed_task_briefs,
    )


@tool(description="评估目标执行结果是否满足要求", hidden_params=["context"], groups=["core"])
async def evaluate(
    messages: Optional[Messages] = None,
    target_goal_id: Optional[str] = None,
    continue_from: Optional[str] = None,
    context: Optional[dict] = None,
) -> Dict[str, Any]:
    """
    评估目标执行结果是否满足要求。

    代码自动从 GoalTree 注入目标描述。模型把执行结果和上下文放在 messages 中。

    Args:
        messages: 执行结果和上下文消息（OpenAI 格式）
        target_goal_id: 要评估的目标 ID（默认当前 goal_id）
        continue_from: 继续已有评估 trace
        context: 框架自动注入的上下文
    """
    if not context:
        return {"status": "failed", "error": "context is required"}

    store = context.get("store")
    trace_id = context.get("trace_id")
    current_goal_id = context.get("goal_id")
    runner = context.get("runner")

    missing = []
    if not store:
        missing.append("store")
    if not trace_id:
        missing.append("trace_id")
    if not runner:
        missing.append("runner")
    if missing:
        return {"status": "failed", "error": f"Missing required context: {', '.join(missing)}"}

    # target_goal_id 默认 context["goal_id"]
    goal_id = target_goal_id or current_goal_id

    # 从 GoalTree 获取目标描述
    goal_desc = await _get_goal_description(store, trace_id, goal_id)

    # 构建 evaluator prompt
    eval_prompt = _build_evaluate_prompt(goal_desc, messages)

    # 获取父 Trace 信息
    parent_trace = await store.get_trace(trace_id)
    if not parent_trace:
        return {"status": "failed", "error": f"Parent trace not found: {trace_id}"}
    try:
        policy = policy_from_context(parent_trace.context)
    except ValueError as exc:
        return {"status": "failed", "error": str(exc)}
    if policy.requires_task_protocol:
        return {
            "status": "failed",
            "error": "evaluate is unavailable in Recursive mode; validation is framework-managed",
        }

    # 处理 continue_from 或创建新 Sub-Trace
    if continue_from:
        existing_trace = await store.get_trace(continue_from)
        if not existing_trace:
            return {"status": "failed", "error": f"Continue-from trace not found: {continue_from}"}
        if existing_trace.parent_trace_id != trace_id:
            return {
                "status": "failed",
                "error": "continue_from must reference a direct child of the current trace",
            }
        if existing_trace.uid != parent_trace.uid:
            return {
                "status": "failed",
                "error": "continue_from trace owner does not match the current trace owner",
            }
        if existing_trace.context.get("created_by_tool") != "evaluate":
            return {
                "status": "failed",
                "error": "continue_from must reference a child created by evaluate",
            }
        try:
            existing_policy = policy_from_context(existing_trace.context)
        except ValueError as exc:
            return {"status": "failed", "error": str(exc)}
        if (
            existing_policy.mode is not policy.mode
            or existing_policy.revision != policy.revision
        ):
            return {
                "status": "failed",
                "error": "continue_from trace Agent mode does not match the current trace",
            }
        parent_depth, expected_root_trace_id = await _resolve_trace_lineage(
            store, parent_trace
        )
        child_depth, root_trace_id = await _resolve_trace_lineage(
            store, existing_trace
        )
        if (
            child_depth != parent_depth + 1
            or root_trace_id != expected_root_trace_id
        ):
            return {
                "status": "failed",
                "error": "continue_from trace lineage does not match the current trace",
            }
        sub_trace_id = continue_from
        goal_tree = await store.get_goal_tree(continue_from)
        mission = goal_tree.mission if goal_tree else eval_prompt
        sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
    else:
        sub_trace_id = generate_sub_trace_id(trace_id, "evaluate")
        parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
        evaluator_context = apply_policy_to_context({
            "created_by_tool": "evaluate",
            "agent_depth": parent_depth + 1,
            "root_trace_id": root_trace_id,
        }, policy)
        sub_trace = Trace(
            trace_id=sub_trace_id,
            mode="agent",
            task=eval_prompt,
            parent_trace_id=trace_id,
            parent_goal_id=current_goal_id,
            agent_type="evaluate",
            uid=parent_trace.uid if parent_trace else None,
            model=parent_trace.model if parent_trace else None,
            status="running",
            context=evaluator_context,
            created_at=datetime.now(),
        )
        await store.create_trace(sub_trace)
        await store.update_goal_tree(sub_trace_id, GoalTree(mission=eval_prompt))
        sub_trace_ids = [{"trace_id": sub_trace_id, "mission": eval_prompt}]

        # 广播 sub_trace_started
        await broadcast_sub_trace_started(
            trace_id, sub_trace_id, current_goal_id or "",
            "evaluate", eval_prompt,
        )

    # 更新主 Goal 为 in_progress
    await _update_goal_start(
        store,
        trace_id,
        current_goal_id,
        "evaluate",
        sub_trace_ids,
        accumulate_sub_trace_ids=policy.accumulate_sub_trace_ids,
    )

    # 注册为活跃协作者
    eval_name = f"评估: {(goal_id or 'unknown')[:20]}"
    await _update_collaborator(
        store, trace_id,
        name=eval_name, sub_trace_id=sub_trace_id,
        status="running", summary=f"评估 Goal {goal_id}",
    )

    # 执行评估
    try:
        # evaluate 使用只读工具 + goal
        allowed_tools = ["read_file", "grep_content", "glob_files", "goal"]
        result = await runner.run_result(
            messages=[{"role": "user", "content": eval_prompt}],
            config=_make_run_config(
                trace_id=sub_trace_id,
                agent_type="evaluate",
                model=parent_trace.model if parent_trace else "gpt-4o",
                uid=parent_trace.uid if parent_trace else None,
                tools=allowed_tools,
                tool_groups=[],
                exclude_tools=["agent", "evaluate", "bash_command"],
                name=f"评估: {goal_id}",
            ),
            on_event=_make_interactive_handler(
                runner, sub_trace_id, trace_id,
                debug_printer=_make_event_printer("evaluate") if getattr(runner, 'debug', False) else None,
            ),
        )

        await broadcast_sub_trace_completed(
            trace_id, sub_trace_id,
            result.get("status", "completed"),
            result.get("summary", ""),
            result.get("stats", {}),
        )
        await _update_collaborator(
            store, trace_id,
            name=eval_name, sub_trace_id=sub_trace_id,
            status=result.get("status", "completed"),
            summary=result.get("summary", "")[:80],
        )

        formatted_summary = result.get("summary", "")

        await _update_goal_complete(
            store, trace_id, current_goal_id,
            result.get("status", "completed"),
            formatted_summary,
        )

        return {
            "mode": "evaluate",
            "sub_trace_id": sub_trace_id,
            "continue_from": bool(continue_from),
            **result,
            "summary": formatted_summary,
        }

    except Exception as e:
        error_msg = str(e)
        await broadcast_sub_trace_completed(
            trace_id, sub_trace_id, "failed", error_msg, {},
        )
        await _update_collaborator(
            store, trace_id,
            name=eval_name, sub_trace_id=sub_trace_id,
            status="failed", summary=error_msg[:80],
        )
        await _update_goal_complete(
            store, trace_id, current_goal_id,
            "failed", f"评估任务失败: {error_msg}",
        )
        return {
            "mode": "evaluate",
            "status": "failed",
            "error": error_msg,
            "sub_trace_id": sub_trace_id,
        }