Agent/cyber_agent/tools/builtin/subagent.py

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

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

import asyncio
import json
import os
from copy import deepcopy
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,
    require_mutable_trace_policy,
    validate_recursive_child_execution,
)
from cyber_agent.core.task_protocol import (
    TaskBrief,
    TaskReport,
    ensure_task_protocol,
    initialize_task_progress,
    format_task_brief,
    new_task_protocol,
    pending_review_entry,
    protocol_error_report,
    replace_task_brief,
    stopped_task_report,
)
from cyber_agent.core.task_protocol_service import TaskProtocolService
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.core.memory import compute_memory_identity
from cyber_agent.core.run_snapshot import (
    RunConfigSnapshotV1,
    RunConfigSnapshotV2,
    persist_run_config_snapshot,
)
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
from cyber_agent.tools.builtin.knowledge import KnowledgeConfig


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

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,
    *,
    task_protocol_service: TaskProtocolService | None = None,
) -> TaskReport:
    """读取孩子已提交的报告，或为异常、停止生成框架报告。

    子任务执行结束后由报告汇合阶段调用，将异常和停止统一转成可审核的 ``TaskReport``。
    """
    service = task_protocol_service or TaskProtocolService(store)
    async with service.transaction(child_trace_id):
        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_progress_revision"] = state.get(
            "task_progress_head_revision"
        )
        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 状态。
    """
    service = getattr(runner, "task_protocol_service", None) or TaskProtocolService(store)
    reports: List[TaskReport] = []
    pending_entries: dict[str, dict[str, Any]] = {}
    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,
            task_protocol_service=service,
        )
        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
        candidate_validations = []
        for candidate_ref in report.candidate_refs:
            candidate_run = await runner.validate_recursive_trace(
                child_trace_id,
                task_brief=task_brief,
                task_report=report_payload,
                candidate_ref=candidate_ref,
            )
            candidate_validations.append({
                "candidate_ref": candidate_ref.model_dump(mode="json"),
                "validation_result": candidate_run.result.model_dump(mode="json"),
            })
        pending_entry = pending_review_entry(
            goal_id=goal_id,
            report=report,
            validation_result=validation_result.model_dump(),
            received_at_sequence=received_at_sequence,
        )
        pending_entry["candidate_validations"] = candidate_validations
        pending_entries[child_trace_id] = pending_entry
        reports.append(report)
    async with service.transaction(parent_trace.trace_id):
        fresh_parent = await store.get_trace(parent_trace.trace_id)
        if fresh_parent is None:
            raise ValueError(f"Parent Trace not found: {parent_trace.trace_id}")
        state = ensure_task_protocol(fresh_parent.context)
        state["pending_reviews"].update(pending_entries)
        await store.update_trace(
            parent_trace.trace_id,
            context=fresh_parent.context,
        )
        parent_trace.context = fresh_parent.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

        return _filter_allowed_tools(allowed_tools, agent_depth, policy)
    return None


def _filter_allowed_tools(
    tool_names: set[str],
    agent_depth: int,
    policy: AgentPolicy,
) -> list[str]:
    """Apply framework protocol/depth restrictions to one capability set."""
    blocked_tools = {"evaluate", "bash_command"}
    if not policy.requires_task_protocol:
        blocked_tools.update({
            "submit_task_report",
            "review_task_result",
            "update_task_progress",
            "read_context_ref",
        })
    elif not policy.requires_task_progress:
        blocked_tools.add("update_task_progress")
    if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
        blocked_tools.add("agent")
    return sorted(tool_names - blocked_tools)


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 = require_mutable_trace_policy(parent_trace.context)
    except ValueError as exc:
        return {"status": "failed", "error": str(exc)}
    protocol_service = (
        getattr(runner, "task_protocol_service", None)
        or TaskProtocolService(store)
    )
    application_binding = getattr(runner, "application_binding", None)
    application_ref = parent_trace.context.get("application_ref")
    target_role_binding = None
    parent_role_binding = None
    target_role_id = agent_type
    if application_ref is not None:
        if application_binding is None:
            return {
                "status": "failed",
                "error": "Application child creation requires the bound ApplicationRuntime",
            }
        if (
            application_ref
            != application_binding.application_ref.model_dump(mode="json")
        ):
            return {
                "status": "failed",
                "error": "Parent ApplicationRef does not match Runner binding",
            }
        if skills is not None:
            return {
                "status": "failed",
                "error": "Application roles do not allow per-call skill overrides",
            }
        parent_role_id = parent_trace.context.get("application_role_id")
        try:
            parent_role_binding = application_binding.role(parent_role_id)
        except ValueError as exc:
            return {"status": "failed", "error": str(exc)}
        allowed_child_roles = parent_role_binding.role.allowed_child_roles
        if target_role_id is None and len(allowed_child_roles) == 1:
            target_role_id = allowed_child_roles[0]
        if target_role_id not in allowed_child_roles:
            return {
                "status": "failed",
                "error": (
                    f"Application role {parent_role_id} cannot create child role: "
                    f"{target_role_id}"
                ),
            }
        try:
            target_role_binding = application_binding.role(target_role_id)
        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)}
        effective_limits = parent_trace.context.get("effective_run_limits") or {}
        if effective_limits:
            max_parallel_children = min(
                max_parallel_children,
                int(effective_limits["max_parallel_children"]),
            )

    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": "Writable Recursive runs require 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",
            }
        if target_role_binding is not None and (
            existing.context.get("application_ref") != application_ref
            or existing.context.get("application_role_id")
            != target_role_binding.role.role_id
            or existing.context.get("application_role_hash")
            != target_role_binding.role_hash
        ):
            return {
                "status": "failed",
                "error": "continue_from application role binding does not match",
            }

        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",
            }
        effective_max_depth = int(
            (parent_trace.context.get("effective_run_limits") or {}).get(
                "max_depth",
                policy.max_depth,
            )
        )
        if child_depth > min(policy.max_depth, effective_max_depth):
            return {
                "status": "failed",
                "error": (
                    "continue_from trace exceeds the persisted Agent mode depth: "
                    f"depth={child_depth}, max={min(policy.max_depth, effective_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
            prepared_context = deepcopy(child_context)
            child_state = ensure_task_protocol(prepared_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,
            )
            brief_changed = child_state.get("task_brief") != normalized_dump
            if brief_changed:
                replace_task_brief(
                    child_state,
                    task_briefs[0],
                    effective_at_sequence=(existing.last_sequence or 0) + 1,
                )
            prepared_context[CONTEXT_ACCESS_KEY] = new_context_access
            persist_root_task_anchor(prepared_context, root_task_anchor)
            if (
                target_role_binding is not None
                and getattr(runner, "context_provider", None) is not None
                and brief_changed
            ):
                from cyber_agent.application.context import load_application_context
                from cyber_agent.application.ports import ContextRequest

                await load_application_context(
                    application_binding,
                    prepared_context,
                    ContextRequest(
                        application_ref=application_binding.application_ref,
                        root_trace_id=root_trace_id,
                        trace_id=existing.trace_id,
                        parent_trace_id=trace_id,
                        uid=parent_trace.uid,
                        role_id=target_role_binding.role.role_id,
                        task_brief=task_briefs[0],
                        task_brief_revision=child_state["task_brief_version"],
                        authorized_context_refs=tuple(
                            task_briefs[0].context_refs
                        ),
                    ),
                    root_task_anchor=root_task_anchor,
                    task_brief=task_briefs[0],
                    granted_at_sequence=(existing.last_sequence or 0) + 1,
                )
            async with protocol_service.transaction(existing.trace_id):
                fresh_child = await store.get_trace(existing.trace_id)
                if fresh_child is None:
                    return {
                        "status": "failed",
                        "error": f"continue_from Trace not found: {existing.trace_id}",
                    }
                fresh_context = apply_policy_to_context({
                    **fresh_child.context,
                    "agent_depth": child_depth,
                    "root_trace_id": root_trace_id,
                }, policy)
                fresh_state = ensure_task_protocol(fresh_context)
                if fresh_state.get("task_brief") != normalized_dump:
                    replace_task_brief(
                        fresh_state,
                        task_briefs[0],
                        effective_at_sequence=(fresh_child.last_sequence or 0) + 1,
                    )
                fresh_context[CONTEXT_ACCESS_KEY] = deepcopy(
                    prepared_context[CONTEXT_ACCESS_KEY]
                )
                persist_root_task_anchor(fresh_context, root_task_anchor)
                await store.update_trace(
                    fresh_child.trace_id,
                    context=fresh_context,
                )
                fresh_child.context = fresh_context
                existing = fresh_child
                child_context = fresh_context
                child_records[0]["context"] = fresh_context
    else:
        parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
        effective_max_depth = int(
            (parent_trace.context.get("effective_run_limits") or {}).get(
                "max_depth",
                policy.max_depth,
            )
        )
        if parent_depth >= min(policy.max_depth, effective_max_depth):
            return {
                "status": "failed",
                "error": (
                    f"Local Sub-Agent depth limit reached: "
                    f"depth={parent_depth}, max={min(policy.max_depth, effective_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 = target_role_id 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)
                    if policy.requires_task_progress:
                        initialize_task_progress(child_context["task_protocol"])
                if target_role_binding is not None:
                    parent_limits = parent_trace.context["effective_run_limits"]
                    role_limits = target_role_binding.effective_limits.model_dump(
                        mode="json"
                    )
                    child_limits = {
                        name: min(parent_limits[name], value)
                        for name, value in role_limits.items()
                    }
                    child_context.update({
                        "application_ref": application_ref,
                        "application_role_id": target_role_binding.role.role_id,
                        "application_role_hash": target_role_binding.role_hash,
                        "effective_run_limits": child_limits,
                    })
                    if getattr(runner, "context_provider", None) is not None:
                        from cyber_agent.application.context import (
                            load_application_context,
                        )
                        from cyber_agent.application.ports import ContextRequest

                        await load_application_context(
                            application_binding,
                            child_context,
                            ContextRequest(
                                application_ref=application_binding.application_ref,
                                root_trace_id=root_trace_id,
                                trace_id=stid,
                                parent_trace_id=trace_id,
                                uid=parent_trace.uid,
                                role_id=target_role_binding.role.role_id,
                                task_brief=task_briefs[i],
                                task_brief_revision=1,
                                authorized_context_refs=tuple(
                                    task_briefs[i].context_refs
                                ),
                            ),
                            root_task_anchor=root_task_anchor,
                            task_brief=task_briefs[i],
                            granted_at_sequence=0,
                        )
                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=(
                        target_role_binding.role.model
                        if target_role_binding is not None
                        else 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
        effective_child_limit = (
            parent_trace.context.get("effective_run_limits") or {}
        ).get("max_children_per_parent")
        if effective_child_limit is not None:
            child_limit = min(
                child_limit if child_limit is not None else effective_child_limit,
                int(effective_child_limit),
            )
        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:
        try:
            async with protocol_service.transaction(trace_id):
                fresh_parent = await store.get_trace(trace_id)
                if fresh_parent is None:
                    raise ValueError(f"Trace not found: {trace_id}")
                fresh_state = ensure_task_protocol(fresh_parent.context)
                if not fresh_state["next_actions"]:
                    raise ValueError("Approved next action was already consumed")
                if fresh_state["next_actions"][0] != approved_action:
                    raise ValueError(
                        "Approved next action changed before execution"
                    )
                fresh_state["next_actions"].pop(0)
                fresh_state["protocol_correction_attempts"] = 0
                await store.update_trace(trace_id, context=fresh_parent.context)
                parent_trace.context = fresh_parent.context
        except Exception as exc:
            await fail_created_children(exc)
            raise

    # 创建延迟执行规格。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)
        if target_role_binding is not None:
            assert parent_role_binding is not None
            allowed_tools = _filter_allowed_tools(
                set(parent_role_binding.delegated_tool_names)
                & set(target_role_binding.tool_names),
                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") or KnowledgeConfig(),
            context=child_record["context"],
            child_execution_mode=child_execution_mode,
            max_parallel_children=max_parallel_children,
        )
        if target_role_binding is not None:
            application_binding.configure_run_config(
                child_config,
                target_role_binding.role.role_id,
            )
            child_config.tools = allowed_tools
            child_config.context = child_record["context"]
            child_config.child_execution_mode = child_execution_mode
            child_config.effective_run_limits = dict(
                child_record["context"]["effective_run_limits"]
            )
            child_config.max_iterations = child_config.effective_run_limits[
                "max_iterations"
            ]
            child_config.max_parallel_children = child_config.effective_run_limits[
                "max_parallel_children"
            ]
        # Recursive children are persisted before their coroutine is scheduled so
        # the parent can account for queued work.  Bind the immutable run snapshot
        # before execution; otherwise the first local resume would look exactly
        # like an unsafe pre-snapshot historical Recursive trace.
        if child_record["is_new"]:
            persisted_child = await store.get_trace(cur_stid)
            if not persisted_child:
                raise RuntimeError(f"pre-created child Trace disappeared: {cur_stid}")
            child_memory_identity = (
                compute_memory_identity(child_config.memory)
                if child_config.memory
                else None
            )
            child_snapshot = (
                RunConfigSnapshotV2.from_run_config(
                    child_config,
                    memory_identity=child_memory_identity,
                )
                if target_role_binding is not None
                else RunConfigSnapshotV1.from_run_config(
                    child_config,
                    memory_identity=child_memory_identity,
                )
            )
            persist_run_config_snapshot(persisted_child.context, child_snapshot)
            await store.update_trace(
                cur_stid,
                context=persisted_child.context,
            )
        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 执行任务；可写 Recursive 协议仅允许本地结构化委派", 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 的公开工具。

    路由规则：
    - Legacy/Recursive revision 1 中 agent_type 以 "remote_" 开头：HTTP
      调用 KnowHub 服务器的 /api/agent（仅单任务，无本地文件访问）
    - Structured Recursive 不支持 remote Agent，也不会发出 HTTP 请求
    - 否则本地执行：Legacy/Recursive revision 1 使用 ``task``；Structured Recursive 使用 ``task_brief``

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

    # 任何路由决策都必须来自已持久化的父 Trace 策略。先看
    # ``remote_*`` 会让 Structured Recursive 绕过本地协议、预算与审核门禁。
    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 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 policy.requires_task_protocol:
            return {
                "status": "failed",
                "error": (
                    "Writable Recursive protocol does not support remote agents; "
                    "use a local task_brief delegation"
                ),
            }
        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,
        )

    if policy.requires_task_protocol and messages is not None:
        return {
            "status": "failed",
            "error": "Writable Recursive protocol 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": "Writable Recursive runs require task_brief and do 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 a writable Recursive run"}
        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,
        }