Agent/cyber_agent/tools/builtin/subagent.py

"""
Sub-Agent 工具 - agent / evaluate

agent: 创建子 Agent 执行任务。
  - 本地：`agent_type` 无 `remote_` 前缀，进程内执行（单任务 delegate / 多任务并行 explore）
  - 远端：`agent_type` 以 `remote_` 开头，HTTP 路由到 KnowHub 服务器的 /api/agent
evaluate: 评估目标执行结果是否满足要求
"""

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

from cyber_agent.tools import tool
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_"
MAX_AGENT_DEPTH = 5
MAX_LOCAL_CHILDREN_PER_TRACE = 6

# 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"))


def _recursion_enabled() -> bool:
    """是否允许本地 Sub-Agent 继续创建 Sub-Agent。"""
    return os.getenv("AGENT_RECURSION_ENABLED", "false").strip().lower() in {
        "1", "true", "yes", "on",
    }


# 兼容旧代码对 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]],
) -> None:
    """标记 Goal 开始执行"""
    if not goal_id:
        return

    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

    await store.update_goal(
        trace_id, goal_id,
        type="agent_call",
        agent_call_mode=mode,
        status="in_progress",
        sub_trace_ids=list(merged_entries.values()),
    )


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,
    )


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
    }


def _get_allowed_tools(context: dict, agent_depth: int) -> Optional[List[str]]:
    """按子 Agent 深度生成唯一的工具权限列表。"""
    runner = context.get("runner")
    if runner and hasattr(runner, "tools") and hasattr(runner.tools, "get_tool_names"):
        blocked_tools = {"evaluate", "bash_command"}
        if not _recursion_enabled() or agent_depth >= MAX_AGENT_DEPTH:
            blocked_tools.add("agent")
        return [
            name for name in runner.tools.get_tool_names()
            if name not in 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


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'):
            # cancel_event.set() 是同步操作，可以在同步回调中直接调用
            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,
) -> Dict[str, Any]:
    """
    统一 agent 执行逻辑。

    single (len(tasks)==1): delegate 模式，全量工具
    multi (len(tasks)>1): explore 模式，并行执行

    AGENT_RECURSION_ENABLED=true 时，depth < MAX_AGENT_DEPTH 的本地
    Sub-Agent 可继续使用 agent 工具。
    """
    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}"}

    # 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]] = []

    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",
            }

        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]
        child_depth, root_trace_id = await _resolve_trace_lineage(store, existing)
        all_sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
        child_records = [{
            "trace_id": sub_trace_id,
            "depth": child_depth,
            "context": {
                **existing.context,
                "agent_depth": child_depth,
                "root_trace_id": root_trace_id,
            },
            "is_new": False,
        }]
    else:
        parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
        if parent_depth > 0 and not _recursion_enabled():
            return {
                "status": "failed",
                "error": "Local Sub-Agent recursion is disabled",
            }
        if parent_depth >= MAX_AGENT_DEPTH:
            return {
                "status": "failed",
                "error": (
                    f"Local Sub-Agent depth limit reached: "
                    f"depth={parent_depth}, max={MAX_AGENT_DEPTH}"
                ),
            }

        # 将配额检查和 Trace 创建放在同一进程内临界区，
        # 保证两个并发 agent() 调用不会共同突破 6 个直接孩子。
        async with _LOCAL_AGENT_SPAWN_LOCK:
            existing_children = await store.list_traces(
                parent_trace_id=trace_id,
                created_by_tool="agent",
                limit=MAX_LOCAL_CHILDREN_PER_TRACE + 1,
            )
            requested_children = len(tasks)
            if len(existing_children) + requested_children > MAX_LOCAL_CHILDREN_PER_TRACE:
                return {
                    "status": "failed",
                    "error": (
                        f"Local child Agent limit exceeded: existing={len(existing_children)}, "
                        f"requested={requested_children}, "
                        f"max={MAX_LOCAL_CHILDREN_PER_TRACE}"
                    ),
                }

            child_depth = parent_depth + 1
            for i, task_item in enumerate(tasks):
                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 = {
                    "created_by_tool": "agent",
                    "agent_depth": child_depth,
                    "root_trace_id": root_trace_id,
                }
                sub_trace = Trace(
                    trace_id=stid,
                    mode="agent",
                    task=task_item,
                    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)
                await store.update_goal_tree(stid, GoalTree(mission=task_item))
                all_sub_trace_ids.append({"trace_id": stid, "mission": task_item})
                child_records.append({
                    "trace_id": stid,
                    "depth": child_depth,
                    "context": child_context,
                    "is_new": True,
                    "agent_type": resolved_agent_type,
                })

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

            await _update_goal_start(
                store, trace_id, goal_id,
                "delegate" if single else "explore",
                all_sub_trace_ids,
            )
            goal_started = True

    # 创建执行协程。Trace 在上面已经按批次预创建，
    # 因此孩子配额不会被并发调用绕过。
    coros = []
    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"]
        if child_record["is_new"]:
            await broadcast_sub_trace_started(
                trace_id, cur_stid, goal_id or "",
                child_record["agent_type"], task_item,
            )

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

        # 构建消息
        agent_msgs = list(msgs) + [{"role": "user", "content": task_item}]
        allowed_tools = _get_allowed_tools(context, child_depth)

        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
        )

        coro = runner.run_result(
            messages=agent_msgs,
            config=_make_run_config(
                trace_id=cur_stid,
                agent_type=agent_type or ("delegate" if single else "explore"),
                max_iterations=50 if single else 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_item[:50],
                skills=skills,
                knowledge=context.get("knowledge_config"),
                context=child_record["context"],
            ),
            on_event=on_event,
        )
        coros.append((i, cur_stid, collab_name, coro))

    # continue_from 不进入新建临界区，但仍需要恢复 Goal 的运行状态。
    if not goal_started:
        await _update_goal_start(
            store, trace_id, goal_id,
            "delegate" if single else "explore",
            all_sub_trace_ids,
        )

    # 执行
    if single:
        # 单任务直接执行（带异常处理）
        _, stid, collab_name, coro = coros[0]
        try:
            result = await coro

            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)

            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],
            )
            await _update_goal_complete(
                store, trace_id, goal_id,
                "failed", f"委托任务失败: {error_msg}",
            )
            return {
                "mode": "delegate",
                "status": "failed",
                "error": error_msg,
                "sub_trace_id": stid,
            }
    else:
        # 检查父 Agent 是否开启了并发配置
        is_parallel = getattr(runner.config, "parallel_tool_execution", True) if runner and hasattr(runner, "config") else True
        
        if is_parallel:
            # 多任务并行执行
            raw_results = await asyncio.gather(
                *(coro for _, _, _, coro in coros),
                return_exceptions=True,
            )
        else:
            # 多任务串行执行（为了省显存/内存）
            raw_results = []
            for _, _, _, coro in coros:
                try:
                    res = await coro
                    raw_results.append(res)
                except Exception as e:
                    raw_results.append(e)

        processed_results = []
        for idx, raw in enumerate(raw_results):
            _, stid, collab_name, _ = coros[idx]
            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)

        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 指定，服务器按 agent_type 的白名单过滤。
    """
    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: Union[str, List[str]],
    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_type 以 "remote_" 开头：HTTP 调用 KnowHub 服务器的 /api/agent（仅单任务，无本地文件访问）
    - 否则本地执行：单任务 str = delegate（全量工具）；多任务 List[str] = explore（并行、只读）

    Args:
        task: 任务描述。字符串=单任务，列表=多任务并行（远端模式只支持单任务）
        messages: 预置消息。1D 列表=所有 agent 共享；2D 列表=per-agent
        continue_from: 继续已有 trace（仅单任务）
        agent_type: 子 Agent 类型。带 "remote_" 前缀走远端；否则本地 preset
        skills: 指定本次调用使用的 skill 列表
                - 本地：附加到 system prompt
                - 远端：由服务器按 agent_type 白名单过滤（如 remote_librarian 允许 ask_strategy / upload_strategy）
        context: 框架自动注入的上下文
    """
    # 远端路由：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)}"}

    # 归一化 task → list
    if isinstance(task, str):
        task_str = task.strip()
        if task_str.startswith("[") and task_str.endswith("]"):
            try:
                import json
                parsed_task = json.loads(task_str)
                if isinstance(parsed_task, list):
                    task = parsed_task
            except:
                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,
    )


@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)

    # 处理 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}"}
        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")
        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={"created_by_tool": "evaluate"},
            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)

    # 注册为活跃协作者
    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,
        }