Agent/agent/core/runner.py

"""
Agent Runner - Agent 执行引擎

核心职责：
1. 执行 Agent 任务（循环调用 LLM + 工具）
2. 记录执行轨迹（Trace + Messages + GoalTree）
3. 检索和注入记忆（Experience + Skill）
4. 管理执行计划（GoalTree）
5. 支持续跑（continue）和回溯重跑（rewind）

参数分层：
- Infrastructure: AgentRunner 构造时设置（trace_store, llm_call 等）
- RunConfig: 每次 run 时指定（model, trace_id, after_sequence 等）
- Messages: OpenAI SDK 格式的任务消息
"""

import asyncio
import json
import logging
import os
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from typing import AsyncIterator, Optional, Dict, Any, List, Callable, Literal, Tuple, Union

from agent.trace.models import Trace, Message
from agent.trace.protocols import TraceStore
from agent.trace.goal_models import GoalTree
from agent.tools.builtin.experience import _get_structured_experiences, _batch_update_experiences
from agent.trace.compaction import (
    CompressionConfig,
    filter_by_goal_status,
    estimate_tokens,
    needs_level2_compression,
    build_compression_prompt,
    build_reflect_prompt,
)
from agent.memory.models import Skill
from agent.memory.protocols import MemoryStore, StateStore
from agent.memory.skill_loader import load_skills_from_dir
from agent.tools import ToolRegistry, get_tool_registry

logger = logging.getLogger(__name__)


# ===== 运行配置 =====

@dataclass
class RunConfig:
    """
    运行参数 — 控制 Agent 如何执行

    分为模型层参数（由上游 agent 或用户决定）和框架层参数（由系统注入）。
    """
    # --- 模型层参数 ---
    model: str = "gpt-4o"
    temperature: float = 0.3
    max_iterations: int = 200
    tools: Optional[List[str]] = None          # None = 全部已注册工具

    # --- 框架层参数 ---
    agent_type: str = "default"
    uid: Optional[str] = None
    system_prompt: Optional[str] = None        # None = 从 skills 自动构建
    skills: Optional[List[str]] = None         # 注入 system prompt 的 skill 名称列表；None = 按 preset 决定
    enable_memory: bool = True
    auto_execute_tools: bool = True
    name: Optional[str] = None                 # 显示名称（空则由 utility_llm 自动生成）
    enable_prompt_caching: bool = True         # 启用 Anthropic Prompt Caching（仅 Claude 模型有效）

    # --- Trace 控制 ---
    trace_id: Optional[str] = None             # None = 新建
    parent_trace_id: Optional[str] = None      # 子 Agent 专用
    parent_goal_id: Optional[str] = None

    # --- 续跑控制 ---
    after_sequence: Optional[int] = None       # 从哪条消息后续跑（message sequence）

    # --- 额外 LLM 参数（传给 llm_call 的 **kwargs）---
    extra_llm_params: Dict[str, Any] = field(default_factory=dict)

    # --- 研究流程控制 ---
    enable_research_flow: bool = True  # 是否启用自动研究流程（知识检索→经验检索→调研→计划）


    # 内置工具列表（始终自动加载）
BUILTIN_TOOLS = [
    # 文件操作工具
    "read_file",
    "edit_file",
    "write_file",
    "glob_files",
    "grep_content",

    # 系统工具
    "bash_command",

    # 技能和目标管理
    "skill",
    "list_skills",
    "goal",
    "agent",
    "evaluate",

    # 搜索工具
    "search_posts",
    "get_experience",
    "get_search_suggestions",

    # 知识管理工具
    "search_knowledge",
    "save_knowledge",
    "update_knowledge",
    "list_knowledge",
    

    # 沙箱工具
    # "sandbox_create_environment",
    # "sandbox_run_shell",
    # "sandbox_rebuild_with_ports",
    # "sandbox_destroy_environment",

    # 浏览器工具
    "browser_navigate_to_url",
    "browser_search_web",
    "browser_go_back",
    "browser_wait",
    "browser_click_element",
    "browser_input_text",
    "browser_send_keys",
    "browser_upload_file",
    "browser_scroll_page",
    "browser_find_text",
    "browser_screenshot",
    "browser_switch_tab",
    "browser_close_tab",
    "browser_get_dropdown_options",
    "browser_select_dropdown_option",
    "browser_extract_content",
    "browser_read_long_content",
    "browser_download_direct_url",
    "browser_get_page_html",
    "browser_get_visual_selector_map",
    "browser_evaluate",
    "browser_ensure_login_with_cookies",
    "browser_wait_for_user_action",
    "browser_done",
    "browser_export_cookies",
    "browser_load_cookies"
]


# ===== 向后兼容 =====

@dataclass
class AgentConfig:
    """[向后兼容] Agent 配置，新代码请使用 RunConfig"""
    agent_type: str = "default"
    max_iterations: int = 200
    enable_memory: bool = True
    auto_execute_tools: bool = True


@dataclass
class CallResult:
    """单次调用结果"""
    reply: str
    tool_calls: Optional[List[Dict]] = None
    trace_id: Optional[str] = None
    step_id: Optional[str] = None
    tokens: Optional[Dict[str, int]] = None
    cost: float = 0.0


# ===== 执行引擎 =====

CONTEXT_INJECTION_INTERVAL = 10  # 每 N 轮注入一次 GoalTree + Collaborators


class AgentRunner:
    """
    Agent 执行引擎

    支持三种运行模式（通过 RunConfig 区分）：
    1. 新建：trace_id=None
    2. 续跑：trace_id=已有ID, after_sequence=None 或 == head
    3. 回溯：trace_id=已有ID, after_sequence=N（N < head_sequence）
    """

    def __init__(
        self,
        trace_store: Optional[TraceStore] = None,
        memory_store: Optional[MemoryStore] = None,
        state_store: Optional[StateStore] = None,
        tool_registry: Optional[ToolRegistry] = None,
        llm_call: Optional[Callable] = None,
        utility_llm_call: Optional[Callable] = None,
        embedding_call: Optional[Callable] = None,
        config: Optional[AgentConfig] = None,
        skills_dir: Optional[str] = None,
        experiences_path: Optional[str] = "./.cache/experiences.md",
        goal_tree: Optional[GoalTree] = None,
        debug: bool = False,
    ):
        """
        初始化 AgentRunner

        Args:
            trace_store: Trace 存储
            memory_store: Memory 存储（可选）
            state_store: State 存储（可选）
            tool_registry: 工具注册表（默认使用全局注册表）
            llm_call: 主 LLM 调用函数
            embedding_call: 语义嵌入向量LLM
            utility_llm_call: 轻量 LLM（用于生成任务标题等），可选
            config: [向后兼容] AgentConfig
            skills_dir: Skills 目录路径
            experiences_path: 经验文件路径（默认 ./.cache/experiences.md）
            goal_tree: 初始 GoalTree（可选）
            debug: 保留参数（已废弃）
        """
        self.trace_store = trace_store
        self.memory_store = memory_store
        self.state_store = state_store
        self.tools = tool_registry or get_tool_registry()
        self.llm_call = llm_call
        self.embedding_call = embedding_call
        self.utility_llm_call = utility_llm_call
        self.config = config or AgentConfig()
        self.skills_dir = skills_dir
        # 确保 experiences_path 不为 None，且文件存在
        self.experiences_path = experiences_path or "./.cache/experiences.md"
        if not os.path.exists(self.experiences_path):
            os.makedirs(os.path.dirname(self.experiences_path), exist_ok=True)
            with open(self.experiences_path, "w", encoding="utf-8") as f:
                f.write("")
            logger.info(f"自动创建经验文件: {self.experiences_path}")
        self.goal_tree = goal_tree
        self.debug = debug
        self._cancel_events: Dict[str, asyncio.Event] = {}  # trace_id → cancel event
        self.used_ex_ids: List[str] = []  # 当前运行中使用过的经验 ID

        # 研究流程状态管理（每个 trace 独立）
        self._research_states: Dict[str, Dict[str, Any]] = {}  # trace_id → research_state

    # ===== 核心公开方法 =====

    async def run(
        self,
        messages: List[Dict],
        config: Optional[RunConfig] = None,
    ) -> AsyncIterator[Union[Trace, Message]]:
        """
        Agent 模式执行（核心方法）

        Args:
            messages: OpenAI SDK 格式的输入消息
                新建: 初始任务消息 [{"role": "user", "content": "..."}]
                续跑: 追加的新消息
                回溯: 在插入点之后追加的消息
            config: 运行配置

        Yields:
            Union[Trace, Message]: Trace 对象（状态变化）或 Message 对象（执行过程）
        """
        if not self.llm_call:
            raise ValueError("llm_call function not provided")

        config = config or RunConfig()
        trace = None

        try:
            # Phase 1: PREPARE TRACE
            trace, goal_tree, sequence = await self._prepare_trace(messages, config)
            # 注册取消事件
            self._cancel_events[trace.trace_id] = asyncio.Event()
            yield trace

            # Phase 2: BUILD HISTORY
            history, sequence, created_messages, head_seq = await self._build_history(
                trace.trace_id, messages, goal_tree, config, sequence
            )
            # Update trace's head_sequence in memory
            trace.head_sequence = head_seq
            for msg in created_messages:
                yield msg

            # Phase 3: AGENT LOOP
            async for event in self._agent_loop(trace, history, goal_tree, config, sequence):
                yield event

        except Exception as e:
            logger.error(f"Agent run failed: {e}")
            tid = config.trace_id or (trace.trace_id if trace else None)
            if self.trace_store and tid:
                # 读取当前 last_sequence 作为 head_sequence，确保续跑时能加载完整历史
                current = await self.trace_store.get_trace(tid)
                head_seq = current.last_sequence if current else None
                await self.trace_store.update_trace(
                    tid,
                    status="failed",
                    head_sequence=head_seq,
                    error_message=str(e),
                    completed_at=datetime.now()
                )
                trace_obj = await self.trace_store.get_trace(tid)
                if trace_obj:
                    yield trace_obj
            raise
        finally:
            # 清理取消事件
            if trace:
                self._cancel_events.pop(trace.trace_id, None)

    async def run_result(
        self,
        messages: List[Dict],
        config: Optional[RunConfig] = None,
        on_event: Optional[Callable] = None,
    ) -> Dict[str, Any]:
        """
        结果模式 — 消费 run()，返回结构化结果。

        主要用于 agent/evaluate 工具内部。

        Args:
            on_event: 可选回调，每个 Trace/Message 事件触发一次，用于实时输出子 Agent 执行过程。
        """
        last_assistant_text = ""
        final_trace: Optional[Trace] = None

        async for item in self.run(messages=messages, config=config):
            if on_event:
                on_event(item)
            if isinstance(item, Message) and item.role == "assistant":
                content = item.content
                text = ""
                if isinstance(content, dict):
                    text = content.get("text", "") or ""
                elif isinstance(content, str):
                    text = content
                if text and text.strip():
                    last_assistant_text = text
            elif isinstance(item, Trace):
                final_trace = item

        config = config or RunConfig()
        if not final_trace and config.trace_id and self.trace_store:
            final_trace = await self.trace_store.get_trace(config.trace_id)

        status = final_trace.status if final_trace else "unknown"
        error = final_trace.error_message if final_trace else None
        summary = last_assistant_text

        if not summary:
            status = "failed"
            error = error or "Agent 没有产生 assistant 文本结果"

        return {
            "status": status,
            "summary": summary,
            "trace_id": final_trace.trace_id if final_trace else config.trace_id,
            "error": error,
            "stats": {
                "total_messages": final_trace.total_messages if final_trace else 0,
                "total_tokens": final_trace.total_tokens if final_trace else 0,
                "total_cost": final_trace.total_cost if final_trace else 0.0,
            },
        }

    async def stop(self, trace_id: str) -> bool:
        """
        停止运行中的 Trace

        设置取消信号，agent loop 在下一个 LLM 调用前检查并退出。
        Trace 状态置为 "stopped"。

        Returns:
            True 如果成功发送停止信号，False 如果该 trace 不在运行中
        """
        cancel_event = self._cancel_events.get(trace_id)
        if cancel_event is None:
            return False
        cancel_event.set()
        return True

    # ===== 单次调用（保留）=====

    async def call(
        self,
        messages: List[Dict],
        model: str = "gpt-4o",
        tools: Optional[List[str]] = None,
        uid: Optional[str] = None,
        trace: bool = True,
        **kwargs
    ) -> CallResult:
        """
        单次 LLM 调用（无 Agent Loop）
        """
        if not self.llm_call:
            raise ValueError("llm_call function not provided")

        trace_id = None
        message_id = None

        tool_schemas = self._get_tool_schemas(tools)

        if trace and self.trace_store:
            trace_obj = Trace.create(mode="call", uid=uid, model=model, tools=tool_schemas, llm_params=kwargs)
            trace_id = await self.trace_store.create_trace(trace_obj)

        result = await self.llm_call(messages=messages, model=model, tools=tool_schemas, **kwargs)

        if trace and self.trace_store and trace_id:
            msg = Message.create(
                trace_id=trace_id, role="assistant", sequence=1, goal_id=None,
                content={"text": result.get("content", ""), "tool_calls": result.get("tool_calls")},
                prompt_tokens=result.get("prompt_tokens", 0),
                completion_tokens=result.get("completion_tokens", 0),
                finish_reason=result.get("finish_reason"),
                cost=result.get("cost", 0),
            )
            message_id = await self.trace_store.add_message(msg)
            await self.trace_store.update_trace(trace_id, status="completed", completed_at=datetime.now())

        return CallResult(
            reply=result.get("content", ""),
            tool_calls=result.get("tool_calls"),
            trace_id=trace_id,
            step_id=message_id,
            tokens={"prompt": result.get("prompt_tokens", 0), "completion": result.get("completion_tokens", 0)},
            cost=result.get("cost", 0)
        )

    # ===== Phase 1: PREPARE TRACE =====

    async def _prepare_trace(
        self,
        messages: List[Dict],
        config: RunConfig,
    ) -> Tuple[Trace, Optional[GoalTree], int]:
        """
        准备 Trace：创建新的或加载已有的

        Returns:
            (trace, goal_tree, next_sequence)
        """
        if config.trace_id:
            return await self._prepare_existing_trace(config)
        else:
            return await self._prepare_new_trace(messages, config)

    async def _prepare_new_trace(
        self,
        messages: List[Dict],
        config: RunConfig,
    ) -> Tuple[Trace, Optional[GoalTree], int]:
        """创建新 Trace"""
        trace_id = str(uuid.uuid4())

        # 生成任务名称
        task_name = config.name or await self._generate_task_name(messages)

        # 准备工具 Schema
        tool_schemas = self._get_tool_schemas(config.tools)

        trace_obj = Trace(
            trace_id=trace_id,
            mode="agent",
            task=task_name,
            agent_type=config.agent_type,
            parent_trace_id=config.parent_trace_id,
            parent_goal_id=config.parent_goal_id,
            uid=config.uid,
            model=config.model,
            tools=tool_schemas,
            llm_params={"temperature": config.temperature, **config.extra_llm_params},
            status="running",
        )

        goal_tree = self.goal_tree or GoalTree(mission=task_name)

        if self.trace_store:
            await self.trace_store.create_trace(trace_obj)
            await self.trace_store.update_goal_tree(trace_id, goal_tree)

        return trace_obj, goal_tree, 1

    async def _prepare_existing_trace(
        self,
        config: RunConfig,
    ) -> Tuple[Trace, Optional[GoalTree], int]:
        """加载已有 Trace（续跑或回溯）"""
        if not self.trace_store:
            raise ValueError("trace_store required for continue/rewind")

        trace_obj = await self.trace_store.get_trace(config.trace_id)
        if not trace_obj:
            raise ValueError(f"Trace not found: {config.trace_id}")

        goal_tree = await self.trace_store.get_goal_tree(config.trace_id)
        if goal_tree is None:
            # 防御性兜底：trace 存在但 goal.json 丢失时，创建空树
            goal_tree = GoalTree(mission=trace_obj.task or "Agent task")
            await self.trace_store.update_goal_tree(config.trace_id, goal_tree)

        # 自动判断行为：after_sequence 为 None 或 == head → 续跑；< head → 回溯
        after_seq = config.after_sequence

        # 如果 after_seq > head_sequence，说明 generator 被强制关闭时 store 的
        # head_sequence 未来得及更新（仍停在 Phase 2 写入的初始值）。
        # 用 last_sequence 修正 head_sequence，确保续跑时能看到完整历史。
        if after_seq is not None and after_seq > trace_obj.head_sequence:
            trace_obj.head_sequence = trace_obj.last_sequence
            await self.trace_store.update_trace(
                config.trace_id, head_sequence=trace_obj.head_sequence
            )

        if after_seq is not None and after_seq < trace_obj.head_sequence:
            # 回溯模式
            sequence = await self._rewind(config.trace_id, after_seq, goal_tree)
        else:
            # 续跑模式：从 last_sequence + 1 开始
            sequence = trace_obj.last_sequence + 1

        # 状态置为 running
        await self.trace_store.update_trace(
            config.trace_id,
            status="running",
            completed_at=None,
        )
        trace_obj.status = "running"

        return trace_obj, goal_tree, sequence

    # ===== Phase 2: BUILD HISTORY =====
    async def _get_embedding(self, text: str) -> List[float]:
        """
        获取文本的嵌入向量（Embedding）
        
        Args:
            text: 需要向量化的文本
            
        Returns:
            List[float]: 嵌入向量
        """
        if not text or not text.strip():
            return []

        # 优先使用注入的 embedding_call
        if self.embedding_call:
            try:
                return await self.embedding_call(text)
            except Exception as e:
                logger.error(f"Error in embedding_call: {e}")
                raise

        # 兜底方案：如果没有注入 embedding_call，但有 llm_call，
        # 某些 SDK 封装可能支持通过 llm_call 的客户端直接获取
        # 这里建议强制要求基础设施层提供该函数以保证分层清晰
        raise ValueError("embedding_call function not provided to AgentRunner")
    
    async def _build_history(
        self,
        trace_id: str,
        new_messages: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
    ) -> Tuple[List[Dict], int, List[Message]]:
        """
        构建完整的 LLM 消息历史

        1. 从 head_sequence 沿 parent chain 加载主路径消息（续跑/回溯场景）
        2. 构建 system prompt（新建时注入 skills）
        3. 新建时：在第一条 user message 末尾注入当前经验
        4. 追加 input messages（设置 parent_sequence 链接到当前 head）

        Returns:
            (history, next_sequence, created_messages, head_sequence)
            created_messages: 本次新创建并持久化的 Message 列表，供 run() yield 给调用方
            head_sequence: 当前主路径头节点的 sequence
        """
        history: List[Dict] = []
        created_messages: List[Message] = []
        head_seq: Optional[int] = None  # 当前主路径的头节点 sequence

        # 1. 加载已有 messages（通过主路径遍历）
        if config.trace_id and self.trace_store:
            trace_obj = await self.trace_store.get_trace(trace_id)
            if trace_obj and trace_obj.head_sequence > 0:
                main_path = await self.trace_store.get_main_path_messages(
                    trace_id, trace_obj.head_sequence
                )

                # 修复 orphaned tool_calls（中断导致的 tool_call 无 tool_result）
                main_path, sequence = await self._heal_orphaned_tool_calls(
                    main_path, trace_id, goal_tree, sequence,
                )

                history = [msg.to_llm_dict() for msg in main_path]
                if main_path:
                    head_seq = main_path[-1].sequence

        # 2. 构建/注入 skills 到 system prompt
        has_system = any(m.get("role") == "system" for m in history)
        has_system_in_new = any(m.get("role") == "system" for m in new_messages)

        if not has_system:
            if has_system_in_new:
                # 入参消息已含 system，将 skills 注入其中（在 step 4 持久化之前）
                augmented = []
                for msg in new_messages:
                    if msg.get("role") == "system":
                        base = msg.get("content") or ""
                        enriched = await self._build_system_prompt(config, base_prompt=base)
                        augmented.append({**msg, "content": enriched or base})
                    else:
                        augmented.append(msg)
                new_messages = augmented
            else:
                # 没有 system，自动构建并插入历史
                system_prompt = await self._build_system_prompt(config)
                if system_prompt:
                    history = [{"role": "system", "content": system_prompt}] + history

                    if self.trace_store:
                        system_msg = Message.create(
                            trace_id=trace_id, role="system", sequence=sequence,
                            goal_id=None, content=system_prompt,
                            parent_sequence=None,  # system message 是 root
                        )
                        await self.trace_store.add_message(system_msg)
                        created_messages.append(system_msg)
                        head_seq = sequence
                        sequence += 1

        # 3. 追加新 messages（设置 parent_sequence 链接到当前 head）
        for msg_dict in new_messages:
            history.append(msg_dict)

            if self.trace_store:
                stored_msg = Message.from_llm_dict(
                    msg_dict, trace_id=trace_id, sequence=sequence,
                    goal_id=None, parent_sequence=head_seq,
                )
                await self.trace_store.add_message(stored_msg)
                created_messages.append(stored_msg)
                head_seq = sequence
                sequence += 1

        # 5. 更新 trace 的 head_sequence
        if self.trace_store and head_seq is not None:
            await self.trace_store.update_trace(trace_id, head_sequence=head_seq)

        # 6. 初始化研究流程（仅在新建 trace 且启用研究流程时）
        if config.enable_research_flow and not config.trace_id:
            await self._init_research_flow(trace_id, new_messages, goal_tree, config)

        return history, sequence, created_messages, head_seq or 0

    async def _init_research_flow(
        self,
        trace_id: str,
        messages: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
    ):
        """
        初始化研究流程状态

        研究流程阶段：
        1. knowledge_search: 知识库检索
        2. experience_search: 经验库检索
        3. research_decision: 决定是否需要调研
        4. research: 执行调研（如果需要）
        5. planning: 制定计划
        6. execution: 正常执行
        """
        # 提取任务描述
        task_desc = self._extract_task_description(messages)

        # 初始化研究状态
        self._research_states[trace_id] = {
            "stage": "knowledge_search",  # 当前阶段
            "task_desc": task_desc,
            "knowledge_found": False,
            "experience_found": False,
            "need_research": False,
            "research_completed": False,
            "planning_completed": False,
            "knowledge_results": [],
            "experience_results": [],
        }

        logger.info(f"[Research Flow] 初始化研究流程: {task_desc[:50]}...")

    def _extract_task_description(self, messages: List[Dict]) -> str:
        """从消息中提取任务描述"""
        for msg in messages:
            if msg.get("role") == "user":
                content = msg.get("content", "")
                if isinstance(content, str):
                    return content
                elif isinstance(content, list):
                    for part in content:
                        if isinstance(part, dict) and part.get("type") == "text":
                            return part.get("text", "")
        return "未知任务"

    def _get_research_state(self, trace_id: str) -> Optional[Dict[str, Any]]:
        """获取研究流程状态"""
        return self._research_states.get(trace_id)

    def _update_research_stage(self, trace_id: str, stage: str, **kwargs):
        """更新研究流程阶段"""
        if trace_id in self._research_states:
            self._research_states[trace_id]["stage"] = stage
            self._research_states[trace_id].update(kwargs)
            logger.info(f"[Research Flow] 阶段切换: {stage}")

    def _build_research_guide(self, research_state: Dict[str, Any]) -> str:
        """根据研究流程状态构建引导消息"""
        stage = research_state["stage"]
        task_desc = research_state["task_desc"]

        if stage == "knowledge_search":
            return f"""
## 🔍 研究流程 - 阶段 1: 知识库检索

当前任务: {task_desc}

**请立即执行以下操作**：
1. 使用 `search_knowledge(query="{task_desc[:100]}")` 检索知识库
2. 检查返回的知识是否足够完成任务
3. 如果找到相关知识，记录下来并继续下一阶段

注意：这是强制步骤，必须先检索知识库再继续。
"""

        elif stage == "experience_search":
            knowledge_results = research_state.get("knowledge_results", [])
            knowledge_summary = f"找到 {len(knowledge_results)} 条相关知识" if knowledge_results else "未找到相关知识"

            return f"""
## 🔍 研究流程 - 阶段 2: 经验库检索

知识库检索结果: {knowledge_summary}

**请立即执行以下操作**：
1. 使用 `get_experience(query="{task_desc[:100]}", k=3)` 检索经验库
2. 结合知识库和经验库的结果，评估是否需要进行调研
3. 如果知识+经验足够，可以跳过调研直接进入计划阶段

注意：这是强制步骤，必须检索经验库。
"""

        elif stage == "research_decision":
            knowledge_results = research_state.get("knowledge_results", [])
            experience_results = research_state.get("experience_results", [])

            # 构建知识库摘要
            knowledge_summary = "### 知识库检索结果\n\n"
            if knowledge_results:
                knowledge_summary += f"找到 {len(knowledge_results)} 条相关知识：\n\n"
                for i, item in enumerate(knowledge_results[:5], 1):  # 最多显示 5 条
                    tags = item.get("tags", [])
                    summary = item.get("summary", "")
                    knowledge_summary += f"{i}. [{', '.join(tags)}] {summary[:150]}...\n"
                if len(knowledge_results) > 5:
                    knowledge_summary += f"\n（还有 {len(knowledge_results) - 5} 条知识未显示）\n"
            else:
                knowledge_summary += "❌ 未找到相关知识\n"

            # 构建经验库摘要
            experience_summary = "\n### 经验库检索结果\n\n"
            if experience_results:
                experience_summary += f"找到 {len(experience_results)} 条相关经验：\n\n"
                for i, item in enumerate(experience_results[:5], 1):  # 最多显示 5 条
                    exp_id = item.get("id", "unknown")
                    content = item.get("content", "")
                    # 提取第一行作为摘要
                    first_line = content.split('\n')[0] if content else ""
                    experience_summary += f"{i}. [{exp_id}] {first_line[:150]}...\n"
                if len(experience_results) > 5:
                    experience_summary += f"\n（还有 {len(experience_results) - 5} 条经验未显示）\n"
            else:
                experience_summary += "❌ 未找到相关经验\n"

            return f"""
## 🤔 研究流程 - 阶段 3: 决定是否调研

当前任务: {task_desc}

{knowledge_summary}
{experience_summary}

---

**现在请你评估并明确回答以下问题**：

1. **现有信息是否足够完成任务？**
   - 如果知识库和经验库的信息已经足够，请回答："无需调研，现有信息充足"
   - 如果信息不足或缺少关键知识，请回答："需要调研"

2. **如果需要调研，请列出具体需要调研的问题**：
   - 缺少哪些工具或方法的信息？
   - 缺少哪些案例或最佳实践？
   - 需要了解哪些定义或原理？

**重要**：
- 请在回复中明确包含"无需调研"或"需要调研"这几个字，以便系统自动判断下一步操作
- **本阶段只需要回复判断结果，不要调用任何工具（包括 goal 工具）**
- 系统会根据你的判断自动进入下一阶段（调研或计划）

根据你的判断，系统会自动进入下一阶段（调研或计划）。
"""

        elif stage == "research":
            # 读取 research.md 的内容
            research_skill_content = ""
            research_skill_path = os.path.join(
                os.path.dirname(__file__),
                "..", "memory", "skills", "research.md"
            )
            try:
                with open(research_skill_path, "r", encoding="utf-8") as f:
                    research_skill_content = f.read()
            except Exception as e:
                logger.warning(f"无法读取 research.md: {e}")
                research_skill_content = "（无法加载 research.md 内容）"

            return f"""
## 📚 研究流程 - 阶段 4: 执行调研

现有信息不足，需要进行调研。

---

{research_skill_content}

---

**请按照上述 research.md 中的方法执行调研**：

1. **拆解搜索维度**：使用 `goal(add="...")` 创建调研子目标
2. **多维度搜索**：使用 `search_posts` 搜索相关信息
3. **结构化记录**：使用 `save_knowledge` 保存每条有价值的知识
4. **完成标记**：调研完成后，使用 `goal(done="...")` 标记完成

**注意**：
- 每发现一条有价值的信息，立即使用 `save_knowledge` 保存
- 调研完成后会自动进入计划阶段
"""

        elif stage == "planning":
            return f"""
## 📋 研究流程 - 阶段 5: 制定计划

调研已完成，现在需要制定执行计划。

**请使用 goal 工具制定计划**：
1. 参考 `agent/memory/skills/planning.md` 中的计划方法
2. 使用 `goal(add="...")` 创建目标树
3. 将任务拆解为可执行的子目标
4. 使用 `goal(focus="...")` 聚焦到第一个目标

计划完成后会自动打印 goal tree 并进入执行阶段。
"""

        return ""

    async def _handle_research_flow_transition(
        self,
        trace_id: str,
        tool_name: str,
        tool_result: Any,
        goal_tree: Optional[GoalTree],
    ):
        """处理研究流程的状态转换"""
        research_state = self._get_research_state(trace_id)
        if not research_state:
            return

        stage = research_state["stage"]

        # 阶段 1: 知识库检索完成
        if stage == "knowledge_search" and tool_name == "search_knowledge":
            # 提取知识检索结果
            # tool_result 现在总是 dict，包含 {"text": ..., "metadata": ...}
            items = []
            if isinstance(tool_result, dict):
                metadata = tool_result.get("metadata", {})
                items = metadata.get("items", [])

            self._update_research_stage(
                trace_id,
                "experience_search",
                knowledge_found=len(items) > 0,
                knowledge_results=items
            )
            logger.info(f"[Research Flow] 知识检索完成，找到 {len(items)} 条知识")

        # 阶段 2: 经验库检索完成
        elif stage == "experience_search" and tool_name == "get_experience":
            # 提取经验检索结果
            experience_items = []
            if isinstance(tool_result, dict):
                experience_items = tool_result.get("items", [])
            elif isinstance(tool_result, str):
                # 解析字符串结果
                try:
                    import json
                    parsed = json.loads(tool_result)
                    experience_items = parsed.get("items", [])
                except:
                    pass

            self._update_research_stage(
                trace_id,
                "research_decision",
                experience_found=len(experience_items) > 0,
                experience_results=experience_items
            )
            logger.info(f"[Research Flow] 经验检索完成，找到 {len(experience_items)} 条经验")

        # 阶段 3: 调研决策（通过 assistant 的文本回复判断）
        # 这个阶段的转换在 assistant 回复后处理，不在这里

        # 阶段 4: 调研完成（检测到 save_knowledge 调用）
        elif stage == "research" and tool_name == "save_knowledge":
            # 标记调研完成，进入计划阶段
            self._update_research_stage(
                trace_id,
                "planning",
                research_completed=True
            )
            logger.info(f"[Research Flow] 调研完成，进入计划阶段")

        # 阶段 5: 计划完成（检测到 goal 工具调用）
        elif stage == "planning" and tool_name == "goal":
            # 检查是否创建了 goal tree
            if goal_tree and goal_tree.goals:
                self._update_research_stage(
                    trace_id,
                    "execution",
                    planning_completed=True
                )
                logger.info(f"[Research Flow] 计划完成，进入执行阶段")

                # 打印 goal tree
                print("\n" + "="*60)
                print("📋 Goal Tree 已创建:")
                print("="*60)
                print(goal_tree.to_prompt())
                print("="*60 + "\n")

    # ===== Phase 3: AGENT LOOP =====

    async def _agent_loop(
        self,
        trace: Trace,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
    ) -> AsyncIterator[Union[Trace, Message]]:
        """ReAct 循环"""
        trace_id = trace.trace_id
        tool_schemas = self._get_tool_schemas(config.tools)

        # 当前主路径头节点的 sequence（用于设置 parent_sequence）
        head_seq = trace.head_sequence

        # 经验检索缓存：只在 goal 切换时重新检索
        _last_goal_id = None
        _cached_exp_text = ""

        for iteration in range(config.max_iterations):
            # 检查取消信号
            cancel_event = self._cancel_events.get(trace_id)
            if cancel_event and cancel_event.is_set():
                logger.info(f"Trace {trace_id} stopped by user")
                if self.trace_store:
                    await self.trace_store.update_trace(
                        trace_id,
                        status="stopped",
                        head_sequence=head_seq,
                        completed_at=datetime.now(),
                    )
                    trace_obj = await self.trace_store.get_trace(trace_id)
                    if trace_obj:
                        yield trace_obj
                return

            # Level 1 压缩：GoalTree 过滤（当消息超过阈值时触发）
            compression_config = CompressionConfig()
            token_count = estimate_tokens(history)
            max_tokens = compression_config.get_max_tokens(config.model)

            # 压缩评估日志
            progress_pct = (token_count / max_tokens * 100) if max_tokens > 0 else 0
            msg_count = len(history)
            img_count = sum(
                1 for msg in history
                if isinstance(msg.get("content"), list)
                for part in msg["content"]
                if isinstance(part, dict) and part.get("type") in ("image", "image_url")
            )

            # 检查是否需要压缩（token 或消息数量超限）
            needs_compression_by_tokens = token_count > max_tokens
            needs_compression_by_count = (
                compression_config.max_messages > 0 and
                msg_count > compression_config.max_messages
            )
            needs_compression = needs_compression_by_tokens or needs_compression_by_count

            print(f"\n[压缩评估] 消息数: {msg_count} / {compression_config.max_messages} | 图片数: {img_count} | Token: {token_count:,} / {max_tokens:,} ({progress_pct:.1f}%)")

            if needs_compression:
                if needs_compression_by_count:
                    print(f"[压缩评估] ⚠️  消息数超过阈值 ({msg_count} > {compression_config.max_messages})，触发压缩流程")
                if needs_compression_by_tokens:
                    print(f"[压缩评估] ⚠️  Token 数超过阈值，触发压缩流程")
            else:
                print(f"[压缩评估] ✅ 未超阈值，无需压缩")

            if needs_compression and self.trace_store and goal_tree:
                # 使用本地 head_seq（store 中的 head_sequence 在 loop 期间未更新，是过时的）
                if head_seq > 0:
                    main_path_msgs = await self.trace_store.get_main_path_messages(
                        trace_id, head_seq
                    )
                    filtered_msgs = filter_by_goal_status(main_path_msgs, goal_tree)
                    if len(filtered_msgs) < len(main_path_msgs):
                        filtered_tokens = estimate_tokens([msg.to_llm_dict() for msg in filtered_msgs])
                        print(
                            f"[Level 1 压缩] 消息: {len(main_path_msgs)} → {len(filtered_msgs)} 条 | "
                            f"Token: {token_count:,} → ~{filtered_tokens:,}"
                        )
                        logger.info(
                            "Level 1 压缩: %d -> %d 条消息 (tokens ~%d, 阈值 %d)",
                            len(main_path_msgs), len(filtered_msgs), token_count, max_tokens,
                        )
                        history = [msg.to_llm_dict() for msg in filtered_msgs]
                    else:
                        print(
                            f"[Level 1 压缩] 无可过滤消息 ({len(main_path_msgs)} 条全部保留, "
                            f"completed/abandoned goals={sum(1 for g in goal_tree.goals if g.status in ('completed', 'abandoned'))})"
                        )
                        logger.info(
                            "Level 1 压缩: 无可过滤消息 (%d 条全部保留, completed/abandoned goals=%d)",
                            len(main_path_msgs),
                            sum(1 for g in goal_tree.goals
                                if g.status in ("completed", "abandoned")),
                        )
            elif needs_compression:
                print("[压缩评估] ⚠️  无法执行 Level 1 压缩（缺少 store 或 goal_tree）")
                logger.warning(
                    "消息数 (%d) 或 token 数 (%d) 超过阈值 (max_messages=%d, max_tokens=%d)，但无法执行 Level 1 压缩（缺少 store 或 goal_tree）",
                    msg_count, token_count, compression_config.max_messages, max_tokens,
                )

            # Level 2 压缩：LLM 总结（Level 1 后仍超阈值时触发）
            token_count_after = estimate_tokens(history)
            msg_count_after = len(history)
            needs_level2_by_tokens = token_count_after > max_tokens
            needs_level2_by_count = (
                compression_config.max_messages > 0 and
                msg_count_after > compression_config.max_messages
            )
            needs_level2 = needs_level2_by_tokens or needs_level2_by_count

            if needs_level2:
                progress_pct_after = (token_count_after / max_tokens * 100) if max_tokens > 0 else 0
                reason = []
                if needs_level2_by_count:
                    reason.append(f"消息数 {msg_count_after} > {compression_config.max_messages}")
                if needs_level2_by_tokens:
                    reason.append(f"Token {token_count_after:,} > {max_tokens:,} ({progress_pct_after:.1f}%)")
                print(
                    f"[Level 2 压缩] Level 1 后仍超阈值: {' | '.join(reason)} "
                    f"→ 触发 LLM 总结"
                )
                logger.info(
                    "Level 1 后仍超阈值 (消息数=%d/%d, token=%d/%d)，触发 Level 2 压缩",
                    msg_count_after, compression_config.max_messages, token_count_after, max_tokens,
                )
                history, head_seq, sequence = await self._compress_history(
                    trace_id, history, goal_tree, config, sequence, head_seq,
                )
                final_tokens = estimate_tokens(history)
                print(f"[Level 2 压缩] 完成: Token {token_count_after:,} → {final_tokens:,}")
            elif needs_compression:
                # Level 1 压缩成功，未触发 Level 2
                print(f"[压缩评估] ✅ Level 1 压缩后达标: 消息数 {msg_count_after} | Token {token_count_after:,} / {max_tokens:,}")
            print()  # 空行分隔

            # 构建 LLM messages（注入上下文）
            llm_messages = list(history)

            # 收集需要持久化的 system 消息
            system_messages_to_persist = []

            # 研究流程引导（仅在启用且处于研究阶段时）
            research_state = self._get_research_state(trace_id)
            if research_state and research_state["stage"] != "execution":
                research_guide = self._build_research_guide(research_state)
                if research_guide:
                    system_msg = {"role": "system", "content": research_guide}
                    llm_messages.append(system_msg)
                    system_messages_to_persist.append(("研究流程引导", system_msg))

            # 周期性注入 GoalTree + Collaborators
            if iteration % CONTEXT_INJECTION_INTERVAL == 0:
                context_injection = self._build_context_injection(trace, goal_tree)
                if context_injection:
                    system_msg = {"role": "system", "content": context_injection}
                    llm_messages.append(system_msg)
                    system_messages_to_persist.append(("上下文注入", system_msg))

            # 经验检索：goal 切换时重新检索，注入为 system message
            current_goal_id = goal_tree.current_id if goal_tree else None
            if current_goal_id and current_goal_id != _last_goal_id:
                _last_goal_id = current_goal_id
                current_goal = goal_tree.find(current_goal_id)
                if current_goal:
                    try:
                        relevant_exps = await _get_structured_experiences(
                            query_text=current_goal.description,
                            top_k=3,
                            context={"runner": self}
                        )
                        if relevant_exps:
                            self.used_ex_ids = [exp['id'] for exp in relevant_exps]
                            parts = [f"[{exp['id']}] {exp['content']}" for exp in relevant_exps]
                            _cached_exp_text = "## 参考历史经验\n" + "\n\n".join(parts)
                            logger.info(
                                "经验检索: goal='%s', 命中 %d 条 %s",
                                current_goal.description[:40],
                                len(relevant_exps),
                                self.used_ex_ids,
                            )
                        else:
                            _cached_exp_text = ""
                    except Exception as e:
                        logger.warning("经验检索失败: %s", e)
                        _cached_exp_text = ""

            if _cached_exp_text:
                system_msg = {"role": "system", "content": _cached_exp_text}
                llm_messages.append(system_msg)
                system_messages_to_persist.append(("经验检索", system_msg))

            # 持久化 system 消息到 trace 和 history
            for label, sys_msg in system_messages_to_persist:
                # 添加到 history（这样会被包含在后续的对话中）
                history.append(sys_msg)

                # 保存到 trace store
                if self.trace_store:
                    system_message = Message.create(
                        trace_id=trace_id,
                        role="system",
                        sequence=sequence,
                        goal_id=current_goal_id,
                        parent_sequence=head_seq if head_seq > 0 else None,
                        content=sys_msg["content"],
                        description=f"[{label}]"  # 添加标签说明这是什么类型的 system 消息
                    )
                    await self.trace_store.add_message(system_message)
                    yield system_message
                    head_seq = sequence
                    sequence += 1

            # 应用 Prompt Caching（不修改原始 history，只在发送给 LLM 时添加缓存标记）
            llm_messages = self._add_cache_control(
                llm_messages,
                config.model,
                config.enable_prompt_caching
            )

            # 调用 LLM
            result = await self.llm_call(
                messages=llm_messages,
                model=config.model,
                tools=tool_schemas,
                temperature=config.temperature,
                **config.extra_llm_params,
            )

            response_content = result.get("content", "")
            tool_calls = result.get("tool_calls")
            finish_reason = result.get("finish_reason")
            prompt_tokens = result.get("prompt_tokens", 0)
            completion_tokens = result.get("completion_tokens", 0)
            step_cost = result.get("cost", 0)
            cache_creation_tokens = result.get("cache_creation_tokens")
            cache_read_tokens = result.get("cache_read_tokens")

            # 研究流程：处理 research_decision 阶段的转换
            research_state = self._get_research_state(trace_id)
            research_decision_handled = False
            if research_state and research_state["stage"] == "research_decision":
                # 根据 assistant 的回复判断是否需要调研
                response_lower = response_content.lower()
                if "需要调研" in response_content or "need research" in response_lower:
                    self._update_research_stage(trace_id, "research", need_research=True)
                    logger.info(f"[Research Flow] 决定需要调研，进入调研阶段")
                    research_decision_handled = True
                    # 如果 Agent 错误地调用了工具，忽略它们
                    if tool_calls:
                        logger.warning(f"[Research Flow] 在 research_decision 阶段不应调用工具，已忽略 {len(tool_calls)} 个工具调用")
                        tool_calls = None
                elif "无需调研" in response_content or "不需要调研" in response_content or "信息充足" in response_content:
                    self._update_research_stage(trace_id, "planning", need_research=False)
                    logger.info(f"[Research Flow] 决定无需调研，直接进入计划阶段")
                    research_decision_handled = True
                    # 如果 Agent 错误地调用了工具，忽略它们
                    if tool_calls:
                        logger.warning(f"[Research Flow] 在 research_decision 阶段不应调用工具，已忽略 {len(tool_calls)} 个工具调用")
                        tool_calls = None

            # 按需自动创建 root goal
            if goal_tree and not goal_tree.goals and tool_calls:
                has_goal_call = any(
                    tc.get("function", {}).get("name") == "goal"
                    for tc in tool_calls
                )
                if not has_goal_call:
                    mission = goal_tree.mission
                    root_desc = mission[:200] if len(mission) > 200 else mission
                    goal_tree.add_goals(
                        descriptions=[root_desc],
                        reasons=["系统自动创建：Agent 未显式创建目标"],
                        parent_id=None
                    )
                    goal_tree.focus(goal_tree.goals[0].id)
                    if self.trace_store:
                        await self.trace_store.add_goal(trace_id, goal_tree.goals[0])
                        await self.trace_store.update_goal_tree(trace_id, goal_tree)
                    logger.info(f"自动创建 root goal: {goal_tree.goals[0].id}")

            # 获取当前 goal_id
            current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None

            # 记录 assistant Message（parent_sequence 指向当前 head）
            assistant_msg = Message.create(
                trace_id=trace_id,
                role="assistant",
                sequence=sequence,
                goal_id=current_goal_id,
                parent_sequence=head_seq if head_seq > 0 else None,
                content={"text": response_content, "tool_calls": tool_calls},
                prompt_tokens=prompt_tokens,
                completion_tokens=completion_tokens,
                cache_creation_tokens=cache_creation_tokens,
                cache_read_tokens=cache_read_tokens,
                finish_reason=finish_reason,
                cost=step_cost,
            )

            if self.trace_store:
                await self.trace_store.add_message(assistant_msg)

            yield assistant_msg
            head_seq = sequence
            sequence += 1

            # 处理工具调用
            # 截断兜底：finish_reason == "length" 说明响应被 max_tokens 截断，
            # tool call 参数很可能不完整，不应执行，改为提示模型分批操作
            if tool_calls and finish_reason == "length":
                logger.warning(
                    "[Runner] 响应被 max_tokens 截断，跳过 %d 个不完整的 tool calls",
                    len(tool_calls),
                )
                truncation_hint = (
                    "你的响应因为 max_tokens 限制被截断，tool call 参数不完整，未执行。"
                    "请将大内容拆分为多次小的工具调用（例如用 write_file 的 append 模式分批写入）。"
                )
                history.append({
                    "role": "assistant",
                    "content": response_content,
                    "tool_calls": tool_calls,
                })
                # 为每个被截断的 tool call 返回错误结果
                for tc in tool_calls:
                    history.append({
                        "role": "tool",
                        "tool_call_id": tc["id"],
                        "content": truncation_hint,
                    })
                continue

            if tool_calls and config.auto_execute_tools:
                history.append({
                    "role": "assistant",
                    "content": response_content,
                    "tool_calls": tool_calls,
                })

                for tc in tool_calls:
                    current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None

                    tool_name = tc["function"]["name"]
                    tool_args = tc["function"]["arguments"]

                    if isinstance(tool_args, str):
                        tool_args = json.loads(tool_args) if tool_args.strip() else {}
                    elif tool_args is None:
                        tool_args = {}

                    tool_result = await self.tools.execute(
                        tool_name,
                        tool_args,
                        uid=config.uid or "",
                        context={
                            "store": self.trace_store,
                            "trace_id": trace_id,
                            "goal_id": current_goal_id,
                            "runner": self,
                            "goal_tree": goal_tree,
                        }
                    )

                    # --- 支持多模态工具反馈 ---
                    # execute() 返回 dict{"text", "images", "metadata"} 或 str
                    if isinstance(tool_result, dict):
                        tool_result_text = tool_result.get("text", str(tool_result))

                        # 处理图片
                        if tool_result.get("images"):
                            # 构建多模态消息格式
                            tool_content_for_llm = [{"type": "text", "text": tool_result_text}]
                            for img in tool_result["images"]:
                                if img.get("type") == "base64" and img.get("data"):
                                    media_type = img.get("media_type", "image/png")
                                    tool_content_for_llm.append({
                                        "type": "image_url",
                                        "image_url": {
                                            "url": f"data:{media_type};base64,{img['data']}"
                                        }
                                    })
                            img_count = len(tool_content_for_llm) - 1  # 减去 text 块
                            print(f"[Runner] 多模态工具反馈: tool={tool_name}, images={img_count}, text_len={len(tool_result_text)}")
                        else:
                            # 无图片，只有文本
                            tool_content_for_llm = tool_result_text
                    else:
                        tool_result_text = str(tool_result)
                        tool_content_for_llm = tool_result_text

                    tool_msg = Message.create(
                        trace_id=trace_id,
                        role="tool",
                        sequence=sequence,
                        goal_id=current_goal_id,
                        parent_sequence=head_seq,
                        tool_call_id=tc["id"],
                        # 存储完整内容：有图片时保留 list（含 image_url），纯文本时存字符串
                        content={"tool_name": tool_name, "result": tool_content_for_llm},
                    )

                    if self.trace_store:
                        await self.trace_store.add_message(tool_msg)
                        # 截图单独存为同名 PNG 文件
                        if isinstance(tool_result, dict) and tool_result.get("images"):
                            import base64 as b64mod
                            for img in tool_result["images"]:
                                if img.get("data"):
                                    png_path = self.trace_store._get_messages_dir(trace_id) / f"{tool_msg.message_id}.png"
                                    png_path.write_bytes(b64mod.b64decode(img["data"]))
                                    print(f"[Runner] 截图已保存: {png_path.name}")
                                    break  # 只存第一张

                    yield tool_msg
                    head_seq = sequence
                    sequence += 1

                    history.append({
                        "role": "tool",
                        "tool_call_id": tc["id"],
                        "name": tool_name,
                        "content": tool_content_for_llm, # 这里传入 list 即可触发模型的视觉能力
                    })
                    # ------------------------------------------

                    # 研究流程状态转换
                    await self._handle_research_flow_transition(
                        trace_id, tool_name, tool_result, goal_tree
                    )

                continue  # 继续循环

            # 研究流程决策阶段：即使没有工具调用也要继续
            if research_decision_handled:
                logger.info(f"[Research Flow] 研究决策完成，继续下一轮循环")
                continue

            # 无工具调用，任务完成
            break

        # 更新 head_sequence 并完成 Trace
        if self.trace_store:
            await self.trace_store.update_trace(
                trace_id,
                status="completed",
                head_sequence=head_seq,
                completed_at=datetime.now(),
            )
            trace_obj = await self.trace_store.get_trace(trace_id)
            if trace_obj:
                yield trace_obj

    # ===== Level 2: LLM 压缩 =====

    async def _compress_history(
        self,
        trace_id: str,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
        head_seq: int,
    ) -> Tuple[List[Dict], int, int]:
        """
        Level 2 压缩：LLM 总结

        Step 1: 经验提取（reflect）— 纯内存 LLM 调用 + 文件追加，不影响 trace
        Step 2: 压缩总结 — LLM 生成 summary
        Step 3: 存储 summary 为新消息，parent_sequence 跳到 system msg
        Step 4: 重建 history

        Returns:
            (new_history, new_head_seq, next_sequence)
        """
        logger.info("Level 2 压缩开始: trace=%s, 当前 history 长度=%d", trace_id, len(history))

        # 找到 system message 的 sequence（主路径第一条消息）
        system_msg_seq = None
        system_msg_dict = None
        if self.trace_store:
            trace_obj = await self.trace_store.get_trace(trace_id)
            if trace_obj and trace_obj.head_sequence > 0:
                main_path = await self.trace_store.get_main_path_messages(
                    trace_id, trace_obj.head_sequence
                )
                for msg in main_path:
                    if msg.role == "system":
                        system_msg_seq = msg.sequence
                        system_msg_dict = msg.to_llm_dict()
                        break

        # Fallback: 从 history 中找 system message
        if system_msg_dict is None:
            for msg_dict in history:
                if msg_dict.get("role") == "system":
                    system_msg_dict = msg_dict
                    break

        if system_msg_dict is None:
            logger.warning("Level 2 压缩跳过：未找到 system message")
            return history, head_seq, sequence

        # --- Step 1: 经验提取（reflect）---
        try:
            # 1. 构造 Reflect Prompt（确保包含格式要求）
            # 建议在 build_reflect_prompt() 里加入：
            # "请使用格式：- [intent: 意图, state: 状态描述] 具体的经验内容"
            reflect_prompt = build_reflect_prompt()
            reflect_messages = list(history) + [{"role": "user", "content": reflect_prompt}]

            # 应用 Prompt Caching
            reflect_messages = self._add_cache_control(
                reflect_messages,
                config.model,
                config.enable_prompt_caching
            )

            reflect_result = await self.llm_call(
                messages=reflect_messages,
                model=config.model,
                tools=[],
                temperature=0.2, # 略微保持一点发散性
                **config.extra_llm_params,
            )

            reflection_text = reflect_result.get("content", "").strip()
            
            if reflection_text:
                import re as _re2
                import uuid as _uuid2

                pattern = r"-\s*\[(?P<tags>.*?)\]\s*(?P<content>.*)"
                matches = list(_re2.finditer(pattern, reflection_text))

                structured_entries = []
                for match in matches:
                    tags_str = match.group("tags")
                    content = match.group("content")

                    intent_match = _re2.search(r"intent:\s*(.*?)(?:,|$)", tags_str, _re2.IGNORECASE)
                    state_match = _re2.search(r"state:\s*(.*?)(?:,|$)", tags_str, _re2.IGNORECASE)

                    intents = [i.strip() for i in intent_match.group(1).split(",")] if intent_match and intent_match.group(1) else []
                    states = [s.strip() for s in state_match.group(1).split(",")] if state_match and state_match.group(1) else []

                    ex_id = f"ex_{datetime.now().strftime('%m%d%H%M')}_{_uuid2.uuid4().hex[:4]}"
                    entry = f"""---
id: {ex_id}
trace_id: {trace_id}
tags: {{intent: {intents}, state: {states}}}
metrics: {{helpful: 1, harmful: 0}}
created_at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
---
- {content}
- 经验ID: [{ex_id}]"""
                    structured_entries.append(entry)

                if structured_entries:
                    os.makedirs(os.path.dirname(self.experiences_path), exist_ok=True)
                    with open(self.experiences_path, "a", encoding="utf-8") as f:
                        f.write("\n\n" + "\n\n".join(structured_entries))
                    logger.info(f"已提取并保存 {len(structured_entries)} 条结构化经验")
                else:
                    logger.warning("未能解析出符合格式的经验条目，请检查 REFLECT_PROMPT。")
                    logger.debug(f"LLM Raw Output:\n{reflection_text}")
            else:
                logger.warning("LLM 未生成反思内容")

        except Exception as e:
            logger.error(f"Level 2 经验提取失败: {e}")

        # --- Step 2: 压缩总结 + 经验评估 ---
        compress_prompt = build_compression_prompt(goal_tree, used_ex_ids=self.used_ex_ids)
        compress_messages = list(history) + [{"role": "user", "content": compress_prompt}]

        # 应用 Prompt Caching
        compress_messages = self._add_cache_control(
            compress_messages,
            config.model,
            config.enable_prompt_caching
        )

        compress_result = await self.llm_call(
            messages=compress_messages,
            model=config.model,
            tools=[],
            temperature=config.temperature,
            **config.extra_llm_params,
        )

        raw_output = compress_result.get("content", "").strip()
        if not raw_output:
            logger.warning("Level 2 压缩跳过：LLM 未返回内容")
            return history, head_seq, sequence

        # 解析 [[EVALUATION]] 块并更新经验
        if self.used_ex_ids:
            try:
                eval_block = ""
                if "[[EVALUATION]]" in raw_output:
                    eval_start = raw_output.index("[[EVALUATION]]") + len("[[EVALUATION]]")
                    eval_end = raw_output.index("[[SUMMARY]]") if "[[SUMMARY]]" in raw_output else len(raw_output)
                    eval_block = raw_output[eval_start:eval_end].strip()

                if eval_block:
                    import re as _re
                    update_map = {}
                    for line in eval_block.splitlines():
                        m = _re.search(r"ID:\s*(ex_\S+)\s*\|\s*Result:\s*(\w+)", line)
                        if m:
                            ex_id, result = m.group(1), m.group(2).lower()
                            if result in ("helpful", "harmful"):
                                update_map[ex_id] = {"action": result, "feedback": ""}
                            elif result == "mixed":
                                update_map[ex_id] = {"action": "helpful", "feedback": ""}
                    if update_map:
                        count = await _batch_update_experiences(update_map, context={"runner": self})
                        logger.info("经验评估完成，更新了 %s 条经验", count)
            except Exception as e:
                logger.warning("经验评估解析失败（不影响压缩）: %s", e)

        # 提取 [[SUMMARY]] 块
        summary_text = raw_output
        if "[[SUMMARY]]" in raw_output:
            summary_text = raw_output[raw_output.index("[[SUMMARY]]") + len("[[SUMMARY]]"):].strip()

        # 压缩完成后清空 used_ex_ids
        self.used_ex_ids = []
        if not summary_text:
            logger.warning("Level 2 压缩跳过：LLM 未返回 summary")
            return history, head_seq, sequence

        # --- Step 3: 存储 summary 消息 ---
        summary_with_header = (
            f"## 对话历史摘要（自动压缩）\n\n{summary_text}\n\n"
            "---\n请基于以上摘要和当前 GoalTree 继续执行任务。"
        )

        summary_msg = Message.create(
            trace_id=trace_id,
            role="user",
            sequence=sequence,
            goal_id=None,
            parent_sequence=system_msg_seq,  # 跳到 system msg，跳过所有中间消息
            content=summary_with_header,
        )

        if self.trace_store:
            await self.trace_store.add_message(summary_msg)

        new_head_seq = sequence
        sequence += 1

        # --- Step 4: 重建 history ---
        new_history = [system_msg_dict, summary_msg.to_llm_dict()]

        # 更新 trace head_sequence
        if self.trace_store:
            await self.trace_store.update_trace(
                trace_id,
                head_sequence=new_head_seq,
            )

        logger.info(
            "Level 2 压缩完成: 旧 history %d 条 → 新 history %d 条, summary 长度=%d",
            len(history), len(new_history), len(summary_text),
        )

        return new_history, new_head_seq, sequence

    # ===== 回溯（Rewind）=====

    async def _rewind(
        self,
        trace_id: str,
        after_sequence: int,
        goal_tree: Optional[GoalTree],
    ) -> int:
        """
        执行回溯：快照 GoalTree，重建干净树，设置 head_sequence

        新消息的 parent_sequence 将指向 rewind 点，旧消息通过树结构自然脱离主路径。

        Returns:
            下一个可用的 sequence 号
        """
        if not self.trace_store:
            raise ValueError("trace_store required for rewind")

        # 1. 加载所有 messages（用于 safe cutoff 和 max sequence）
        all_messages = await self.trace_store.get_trace_messages(trace_id)

        if not all_messages:
            return 1

        # 2. 找到安全截断点（确保不截断在 tool_call 和 tool response 之间）
        cutoff = self._find_safe_cutoff(all_messages, after_sequence)

        # 3. 快照并重建 GoalTree
        if goal_tree:
            # 获取截断点消息的 created_at 作为时间界限
            cutoff_msg = None
            for msg in all_messages:
                if msg.sequence == cutoff:
                    cutoff_msg = msg
                    break

            cutoff_time = cutoff_msg.created_at if cutoff_msg else datetime.now()

            # 快照到 events（含 head_sequence 供前端感知分支切换）
            await self.trace_store.append_event(trace_id, "rewind", {
                "after_sequence": cutoff,
                "head_sequence": cutoff,
                "goal_tree_snapshot": goal_tree.to_dict(),
            })

            # 按时间重建干净的 GoalTree
            new_tree = goal_tree.rebuild_for_rewind(cutoff_time)
            await self.trace_store.update_goal_tree(trace_id, new_tree)

            # 更新内存中的引用
            goal_tree.goals = new_tree.goals
            goal_tree.current_id = new_tree.current_id

        # 4. 更新 head_sequence 到 rewind 点
        await self.trace_store.update_trace(trace_id, head_sequence=cutoff)

        # 5. 返回 next sequence（全局递增，不复用）
        max_seq = max((m.sequence for m in all_messages), default=0)
        return max_seq + 1

    def _find_safe_cutoff(self, messages: List[Message], after_sequence: int) -> int:
        """
        找到安全的截断点。

        如果 after_sequence 指向一条带 tool_calls 的 assistant message，
        则自动扩展到其所有对应的 tool response 之后。
        """
        cutoff = after_sequence

        # 找到 after_sequence 对应的 message
        target_msg = None
        for msg in messages:
            if msg.sequence == after_sequence:
                target_msg = msg
                break

        if not target_msg:
            return cutoff

        # 如果是 assistant 且有 tool_calls，找到所有对应的 tool responses
        if target_msg.role == "assistant":
            content = target_msg.content
            if isinstance(content, dict) and content.get("tool_calls"):
                tool_call_ids = set()
                for tc in content["tool_calls"]:
                    if isinstance(tc, dict) and tc.get("id"):
                        tool_call_ids.add(tc["id"])

                # 找到这些 tool_call 对应的 tool messages
                for msg in messages:
                    if (msg.role == "tool" and msg.tool_call_id
                            and msg.tool_call_id in tool_call_ids):
                        cutoff = max(cutoff, msg.sequence)

        return cutoff

    async def _heal_orphaned_tool_calls(
        self,
        messages: List[Message],
        trace_id: str,
        goal_tree: Optional[GoalTree],
        sequence: int,
    ) -> tuple:
        """
        检测并修复消息历史中的 orphaned tool_calls。

        当 agent 被 stop/crash 中断时，可能有 assistant 的 tool_calls 没有对应的
        tool results（包括多 tool_call 部分完成的情况）。直接发给 LLM 会导致 400。

        修复策略：为每个缺失的 tool_result 插入合成的"中断通知"消息，而非裁剪。
        - 普通工具：简短中断提示
        - agent/evaluate：包含 sub_trace_id、执行统计、continue_from 指引

        合成消息持久化到 store，确保幂等（下次续跑不再触发）。

        Returns:
            (healed_messages, next_sequence)
        """
        if not messages:
            return messages, sequence

        # 收集所有 tool_call IDs → (assistant_msg, tool_call_dict)
        tc_map: Dict[str, tuple] = {}
        result_ids: set = set()

        for msg in messages:
            if msg.role == "assistant":
                content = msg.content
                if isinstance(content, dict) and content.get("tool_calls"):
                    for tc in content["tool_calls"]:
                        tc_id = tc.get("id")
                        if tc_id:
                            tc_map[tc_id] = (msg, tc)
            elif msg.role == "tool" and msg.tool_call_id:
                result_ids.add(msg.tool_call_id)

        orphaned_ids = [tc_id for tc_id in tc_map if tc_id not in result_ids]
        if not orphaned_ids:
            return messages, sequence

        logger.info(
            "检测到 %d 个 orphaned tool_calls，生成合成中断通知",
            len(orphaned_ids),
        )

        healed = list(messages)
        head_seq = messages[-1].sequence

        for tc_id in orphaned_ids:
            assistant_msg, tc = tc_map[tc_id]
            tool_name = tc.get("function", {}).get("name", "unknown")

            if tool_name in ("agent", "evaluate"):
                result_text = self._build_agent_interrupted_result(
                    tc, goal_tree, assistant_msg,
                )
            else:
                result_text = (
                    f"⚠️ 工具 {tool_name} 执行被中断（进程异常退出），"
                    "未获得执行结果。请根据需要重新调用。"
                )

            synthetic_msg = Message.create(
                trace_id=trace_id,
                role="tool",
                sequence=sequence,
                goal_id=assistant_msg.goal_id,
                parent_sequence=head_seq,
                tool_call_id=tc_id,
                content={"tool_name": tool_name, "result": result_text},
            )

            if self.trace_store:
                await self.trace_store.add_message(synthetic_msg)

            healed.append(synthetic_msg)
            head_seq = sequence
            sequence += 1

        # 更新 trace head/last sequence
        if self.trace_store:
            await self.trace_store.update_trace(
                trace_id,
                head_sequence=head_seq,
                last_sequence=max(head_seq, sequence - 1),
            )

        return healed, sequence

    def _build_agent_interrupted_result(
        self,
        tc: Dict,
        goal_tree: Optional[GoalTree],
        assistant_msg: Message,
    ) -> str:
        """为中断的 agent/evaluate 工具调用构建合成结果（对齐正常返回值格式）"""
        args_str = tc.get("function", {}).get("arguments", "{}")
        try:
            args = json.loads(args_str) if isinstance(args_str, str) else args_str
        except json.JSONDecodeError:
            args = {}

        task = args.get("task", "未知任务")
        if isinstance(task, list):
            task = "; ".join(task)

        tool_name = tc.get("function", {}).get("name", "agent")
        mode = "evaluate" if tool_name == "evaluate" else "delegate"

        # 从 goal_tree 查找 sub_trace 信息
        sub_trace_id = None
        stats = None
        if goal_tree and assistant_msg.goal_id:
            goal = goal_tree.find(assistant_msg.goal_id)
            if goal and goal.sub_trace_ids:
                first = goal.sub_trace_ids[0]
                if isinstance(first, dict):
                    sub_trace_id = first.get("trace_id")
                elif isinstance(first, str):
                    sub_trace_id = first
                if goal.cumulative_stats:
                    s = goal.cumulative_stats
                    if s.message_count > 0:
                        stats = {
                            "message_count": s.message_count,
                            "total_tokens": s.total_tokens,
                            "total_cost": round(s.total_cost, 4),
                        }

        result: Dict[str, Any] = {
            "mode": mode,
            "status": "interrupted",
            "summary": "⚠️ 子Agent执行被中断（进程异常退出）",
            "task": task,
        }
        if sub_trace_id:
            result["sub_trace_id"] = sub_trace_id
            result["hint"] = (
                f'使用 continue_from="{sub_trace_id}" 可继续执行，保留已有进度'
            )
        if stats:
            result["stats"] = stats

        return json.dumps(result, ensure_ascii=False, indent=2)

    # ===== 上下文注入 =====

    def _build_context_injection(
        self,
        trace: Trace,
        goal_tree: Optional[GoalTree],
    ) -> str:
        """构建周期性注入的上下文（GoalTree + Active Collaborators + Focus 提醒）"""
        parts = []

        # GoalTree
        if goal_tree and goal_tree.goals:
            parts.append(f"## Current Plan\n\n{goal_tree.to_prompt()}")

            # 检测 focus 在有子节点的父目标上：提醒模型 focus 到具体子目标
            if goal_tree.current_id:
                children = goal_tree.get_children(goal_tree.current_id)
                pending_children = [c for c in children if c.status in ("pending", "in_progress")]
                if pending_children:
                    child_ids = ", ".join(
                        goal_tree._generate_display_id(c) for c in pending_children[:3]
                    )
                    parts.append(
                        f"**提醒**：当前焦点在父目标上，建议用 `goal(focus=\"...\")` "
                        f"切换到具体子目标（如 {child_ids}）再执行。"
                    )

        # Active Collaborators
        collaborators = trace.context.get("collaborators", [])
        if collaborators:
            lines = ["## Active Collaborators"]
            for c in collaborators:
                status_str = c.get("status", "unknown")
                ctype = c.get("type", "agent")
                summary = c.get("summary", "")
                name = c.get("name", "unnamed")
                lines.append(f"- {name} [{ctype}, {status_str}]: {summary}")
            parts.append("\n".join(lines))

        return "\n\n".join(parts)

    # ===== 辅助方法 =====

    def _add_cache_control(
        self,
        messages: List[Dict],
        model: str,
        enable: bool
    ) -> List[Dict]:
        """
        为支持的模型添加 Prompt Caching 标记

        策略：
        1. system message 添加缓存（如果存在且足够长）
        2. 倒数第 3-5 条 user/assistant 消息添加缓存点

        Args:
            messages: 原始消息列表
            model: 模型名称
            enable: 是否启用缓存

        Returns:
            添加了 cache_control 的消息列表（深拷贝）
        """
        if not enable:
            return messages

        # 只对 Claude 模型启用
        if "claude" not in model.lower():
            return messages

        # 深拷贝避免修改原始数据
        import copy
        messages = copy.deepcopy(messages)

        # 策略 1: 为 system message 添加缓存
        for msg in messages:
            if msg.get("role") == "system":
                content = msg.get("content", "")
                # 只有足够长的 system prompt 才值得缓存（>1024 tokens 约 4000 字符）
                if isinstance(content, str) and len(content) > 1000:
                    # Anthropic API 格式：在 content 的最后一个 block 添加 cache_control
                    # 如果 content 是 string，需要转换为 list 格式
                    msg["content"] = [
                        {
                            "type": "text",
                            "text": content,
                            "cache_control": {"type": "ephemeral"}
                        }
                    ]
                    logger.debug(f"[Cache] 为 system message 添加缓存标记 (len={len(content)})")
                break

        # 策略 2: 为倒数第 3-5 条消息添加缓存点
        # 这样可以缓存大部分历史对话，只有最新的几条消息是新的
        cache_positions = []
        user_assistant_msgs = [
            (i, msg) for i, msg in enumerate(messages)
            if msg.get("role") in ("user", "assistant")
        ]

        if len(user_assistant_msgs) >= 5:
            # 在倒数第 5 条添加缓存点
            cache_positions.append(user_assistant_msgs[-5][0])
        elif len(user_assistant_msgs) >= 3:
            # 在倒数第 3 条添加缓存点
            cache_positions.append(user_assistant_msgs[-3][0])

        for idx in cache_positions:
            msg = messages[idx]
            content = msg.get("content", "")

            # 处理 string content
            if isinstance(content, str):
                msg["content"] = [
                    {
                        "type": "text",
                        "text": content,
                        "cache_control": {"type": "ephemeral"}
                    }
                ]
                logger.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记")

            # 处理 list content（多模态消息）
            elif isinstance(content, list) and len(content) > 0:
                # 在最后一个 text block 添加 cache_control
                for i in range(len(content) - 1, -1, -1):
                    if isinstance(content[i], dict) and content[i].get("type") == "text":
                        content[i]["cache_control"] = {"type": "ephemeral"}
                        logger.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 的 content[{i}] 添加缓存标记")
                        break

        return messages

    def _get_tool_schemas(self, tools: Optional[List[str]]) -> List[Dict]:
        """
        获取工具 Schema

        - tools=None: 使用 registry 中全部已注册工具（含内置 + 外部注册的）
        - tools=["a", "b"]: 在 BUILTIN_TOOLS 基础上追加指定工具
        """
        if tools is None:
            # 全部已注册工具
            tool_names = self.tools.get_tool_names()
        else:
            # BUILTIN_TOOLS + 显式指定的额外工具
            tool_names = BUILTIN_TOOLS.copy()
            for t in tools:
                if t not in tool_names:
                    tool_names.append(t)
        return self.tools.get_schemas(tool_names)

    # 默认 system prompt 前缀（当 config.system_prompt 和前端都未提供 system message 时使用）
    DEFAULT_SYSTEM_PREFIX = "你是最顶尖的AI助手，可以拆分并调用工具逐步解决复杂问题。"

    async def _build_system_prompt(self, config: RunConfig, base_prompt: Optional[str] = None) -> Optional[str]:
        """构建 system prompt（注入 skills）

        优先级：
        1. config.skills 显式指定 → 按名称过滤
        2. config.skills 为 None → 查 preset 的默认 skills 列表
        3. preset 也无 skills（None）→ 加载全部（向后兼容）

        Args:
            base_prompt: 已有 system 内容（来自消息或 config.system_prompt），
                         None 时使用 config.system_prompt
        """
        from agent.core.presets import AGENT_PRESETS

        system_prompt = base_prompt if base_prompt is not None else config.system_prompt

        # 确定要加载哪些 skills
        skills_filter: Optional[List[str]] = config.skills
        if skills_filter is None:
            preset = AGENT_PRESETS.get(config.agent_type)
            if preset is not None:
                skills_filter = preset.skills  # 可能仍为 None（加载全部）

        # 加载并过滤
        all_skills = load_skills_from_dir(self.skills_dir)
        if skills_filter is not None:
            skills = [s for s in all_skills if s.name in skills_filter]
        else:
            skills = all_skills

        skills_text = self._format_skills(skills) if skills else ""

        if system_prompt:
            if skills_text:
                system_prompt += f"\n\n## Skills\n{skills_text}"
        else:
            system_prompt = self.DEFAULT_SYSTEM_PREFIX
            if skills_text:
                system_prompt += f"\n\n## Skills\n{skills_text}"

        return system_prompt

    async def _generate_task_name(self, messages: List[Dict]) -> str:
        """生成任务名称：优先使用 utility_llm，fallback 到文本截取"""
        # 提取 messages 中的文本内容
        text_parts = []
        for msg in messages:
            content = msg.get("content", "")
            if isinstance(content, str):
                text_parts.append(content)
            elif isinstance(content, list):
                for part in content:
                    if isinstance(part, dict) and part.get("type") == "text":
                        text_parts.append(part.get("text", ""))
        raw_text = " ".join(text_parts).strip()

        if not raw_text:
            return "未命名任务"

        # 尝试使用 utility_llm 生成标题
        if self.utility_llm_call:
            try:
                result = await self.utility_llm_call(
                    messages=[
                        {"role": "system", "content": "用中文为以下任务生成一个简短标题（10-30字），只输出标题本身："},
                        {"role": "user", "content": raw_text[:2000]},
                    ],
                    model="gpt-4o-mini",  # 使用便宜模型
                )
                title = result.get("content", "").strip()
                if title and len(title) < 100:
                    return title
            except Exception:
                pass

        # Fallback: 截取前 50 字符
        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")

    def _format_skills(self, skills: List[Skill]) -> str:
        if not skills:
            return ""
        return "\n\n".join(s.to_prompt_text() for s in skills)

    def _load_experiences(self) -> str:
        """从文件加载经验（./.cache/experiences.md）"""
        if not self.experiences_path:
            return ""
        try:
            if os.path.exists(self.experiences_path):
                with open(self.experiences_path, "r", encoding="utf-8") as f:
                    return f.read().strip()
        except Exception as e:
            logger.warning(f"Failed to load experiences from {self.experiences_path}: {e}")
        return ""