Agent/cyber_agent/core/runner.py

"""
Agent Runner - Agent 执行引擎

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

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

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

from cyber_agent.trace.models import Trace, Message
from cyber_agent.trace.protocols import TraceStore
from cyber_agent.trace.trace_id import generate_sub_trace_id
from cyber_agent.trace.goal_models import GoalTree
from cyber_agent.trace.compaction import (
    CompressionConfig,
    compress_completed_goals,
    estimate_tokens,
    needs_level2_compression,
    build_compression_prompt,
)
from cyber_agent.skill.models import Skill
from cyber_agent.skill.skill_loader import load_skills_from_dir
from cyber_agent.tools import ToolRegistry, get_tool_registry
from cyber_agent.tools.approval import (
    ToolApprovalBatchV1,
    ToolApprovalCallV1,
    approval_grant,
    tool_argument_hash,
)
from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
from cyber_agent.core.memory import (
    MEMORY_IDENTITY_CONTEXT_KEY,
    MemoryConfig,
    compute_memory_identity,
)
from cyber_agent.core.dream import DreamScope
from cyber_agent.core.run_snapshot import (
    RUN_CONFIG_SNAPSHOT_CONTEXT_KEY,
    RunConfigSnapshotError,
    RunConfigSnapshotV1,
    RunConfigSnapshotV2,
    load_run_config_snapshot,
    persist_run_config_snapshot,
)
from cyber_agent.core.agent_mode import (
    AGENT_MODE_CONTEXT_KEY,
    AgentMode,
    RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY,
    RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
    RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
    apply_policy_to_context,
    assert_removed_config_absent,
    policy_from_context,
    CURRENT_RECURSIVE_REVISION,
    policy_from_environment,
    require_mutable_trace_policy,
    validate_recursive_child_execution,
)
from cyber_agent.core.task_protocol import (
    RootTaskAnchor,
    ensure_task_protocol,
    initialize_task_progress,
    new_task_protocol,
    task_progress_artifact_refs,
    task_progress_at_revision,
    task_progress_readiness_error,
    rewind_task_progress,
    protocol_error_report,
    rebuild_pending_replans,
)
from cyber_agent.core.task_protocol_service import TaskProtocolService
from cyber_agent.core.context_policy import (
    canonical_json,
    ContextPolicyError,
    normalize_root_task_anchor,
    persist_root_task_anchor,
    prune_context_access,
    render_recursive_context,
    get_authorized_context_snapshot,
    require_matching_root_task_anchor,
    require_root_task_anchor,
    replace_context_access,
)
from cyber_agent.core.resource_budget import (
    RESOURCE_BUDGET_CONTEXT_KEY,
    ResourceBudget,
    ResourceBudgetController,
    ResourceBudgetExceeded,
    ResourceBudgetStateError,
)
from cyber_agent.core.artifacts import (
    ArtifactRef,
    ArtifactResolver,
    MaterialIssue,
    ValidationMaterial,
    extract_artifact_refs,
    material_chars,
    resolve_artifact_refs,
)
from cyber_agent.core.validation import (
    ValidationCheck,
    ValidationCheckSpec,
    LLMValidator,
    ScopeValidationResult,
    ValidationPolicy,
    ValidationResult,
    ValidationRun,
    ValidationScope,
    ValidatorSettings,
    persist_validation_policy,
    require_validation_policy,
)
from cyber_agent.core.validator_web import (
    PageFetcher,
    SerperWebSearchProvider,
    ValidatorToolLimits,
    ValidatorToolSession,
    ValidatorWebSearchProvider,
)
from cyber_agent.core.prompts import (
    DEFAULT_SYSTEM_PREFIX,
    TRUNCATION_HINT,
    TOOL_INTERRUPTED_MESSAGE,
    AGENT_INTERRUPTED_SUMMARY,
    AGENT_CONTINUE_HINT_TEMPLATE,
    TASK_NAME_GENERATION_SYSTEM_PROMPT,
    TASK_NAME_FALLBACK,
    SUMMARY_HEADER_TEMPLATE,
    build_summary_header,
    build_tool_interrupted_message,
    build_agent_continue_hint,
)

logger = logging.getLogger(__name__)


@dataclass
class ContextUsage:
    """Context 使用情况"""
    trace_id: str
    message_count: int
    token_count: int
    max_tokens: int
    usage_percent: float
    image_count: int = 0


@dataclass
class SideBranchContext:
    """侧分支上下文（压缩/反思/知识评估）"""
    type: Literal["compression", "reflection", "knowledge_eval"]
    branch_id: str
    start_head_seq: int          # 侧分支起点的 head_seq
    start_sequence: int          # 侧分支第一条消息的 sequence
    start_history_length: int    # 侧分支起点的 history 长度
    start_iteration: int         # 旧 Trace 兼容字段；不再用于轮次预算
    max_turns: int = 5           # 最大轮次
    turns_used: int = 0          # 已持久化的 assistant 轮次

    def to_dict(self) -> Dict[str, Any]:
        """转换为字典（用于持久化和传递给工具）"""
        return {
            "type": self.type,
            "branch_id": self.branch_id,
            "start_head_seq": self.start_head_seq,
            "start_sequence": self.start_sequence,
            "start_iteration": self.start_iteration,
            "max_turns": self.max_turns,
            "turns_used": self.turns_used,
            "is_side_branch": True,
            "started_at": datetime.now().isoformat(),
        }


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

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

    分为模型层参数（由上游 agent 或用户决定）和框架层参数（由系统注入）。
    """
    # --- 模型层参数 ---
    model: str = "gpt-4o"
    temperature: float = 0.3
    max_iterations: int = 200
    tools: Optional[List[str]] = None          # None = 按 tool_groups 过滤；显式列表 = 精确指定
    tool_groups: Optional[List[str]] = field(default_factory=lambda: ["core"])  # 工具分组白名单；默认仅 core，项目按需追加
    exclude_tools: List[str] = field(default_factory=list)  # 从 tools / tool_groups 结果中再排除的工具名（如远程 agent 禁用 agent/evaluate）
    side_branch_max_turns: int = 5             # 侧分支最大轮次（压缩/反思）
    goal_compression: Literal["none", "on_complete", "on_overflow"] = "on_overflow"  # Goal 压缩模式

    # --- 强制侧分支（用于 API 手动触发或自动压缩流程）---
    # 使用列表作为侧分支队列，每次完成一个侧分支后 pop(0) 取下一个
    force_side_branch: Optional[List[Literal["compression", "reflection"]]] = 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 模型有效）
    parallel_tool_execution: bool = False      # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突）
    child_execution_mode: Literal["sequential", "parallel"] = "sequential"
    max_parallel_children: int = 2
    root_task_anchor: Optional[RootTaskAnchor] = None

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

    # --- 自定义元数据上下文 ---
    context: Dict[str, Any] = field(default_factory=dict)

    # --- 研究流程控制 ---
    enable_research_flow: bool = True  # 是否启用自动研究流程（知识检索→经验检索→调研→计划）
    # --- 知识管理配置 ---
    knowledge: KnowledgeConfig = field(default_factory=KnowledgeConfig)
    # --- Memory 配置（见 cyber_agent/docs/framework/runtime/memory.md） ---
    # None = 默认 Agent（无长期记忆）；赋值 MemoryConfig 使该 Agent 成为 memory-bearing Agent
    memory: Optional["MemoryConfig"] = None

    # --- 一次性恢复动作（不进入持久化 RunConfigSnapshot） ---
    approval_batch_id: Optional[str] = None

    # --- ApplicationRuntime 固化身份（只由框架装配） ---
    application_ref: Any = None
    role_id: Optional[str] = None
    role_hash: Optional[str] = None
    effective_run_limits: Dict[str, Any] = field(default_factory=dict)

    def apply_snapshot(
        self,
        snapshot: RunConfigSnapshotV1 | RunConfigSnapshotV2,
    ) -> None:
        """Restore all persisted behavior fields while retaining invocation routing."""
        self.model = snapshot.model
        self.temperature = snapshot.temperature
        self.max_iterations = snapshot.max_iterations
        self.extra_llm_params = dict(snapshot.extra_llm_params)
        self.tools = list(snapshot.tools) if snapshot.tools is not None else None
        self.tool_groups = (
            list(snapshot.tool_groups) if snapshot.tool_groups is not None else None
        )
        self.exclude_tools = list(snapshot.exclude_tools)
        self.auto_execute_tools = snapshot.auto_execute_tools
        self.agent_type = snapshot.agent_type
        self.uid = snapshot.uid
        self.skills = list(snapshot.skills) if snapshot.skills is not None else None
        self.enable_memory = snapshot.enable_memory
        self.memory = MemoryConfig(**snapshot.memory) if snapshot.memory else None
        self.knowledge = KnowledgeConfig(**snapshot.knowledge)
        self.parallel_tool_execution = snapshot.parallel_tool_execution
        self.child_execution_mode = snapshot.child_execution_mode
        self.max_parallel_children = snapshot.max_parallel_children
        self.side_branch_max_turns = snapshot.side_branch_max_turns
        self.goal_compression = snapshot.goal_compression
        self.enable_prompt_caching = snapshot.enable_prompt_caching
        self.enable_research_flow = snapshot.enable_research_flow
        self.context = dict(snapshot.custom_context)
        if isinstance(snapshot, RunConfigSnapshotV2):
            self.application_ref = dict(snapshot.application_ref)
            self.role_id = snapshot.role_id
            self.role_hash = snapshot.role_hash
            self.effective_run_limits = dict(snapshot.effective_run_limits)


    # BUILTIN_TOOLS 硬编码列表已移除（2026-04）。
    # 工具可用性现在由 @tool(groups=[...]) 声明 + RunConfig.tool_groups 过滤控制。


@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 = 5  # 每 N 轮注入一次 GoalTree + Collaborators + IM 通知


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,
        tool_registry: Optional[ToolRegistry] = None,
        llm_call: Optional[Callable] = None,
        utility_llm_call: Optional[Callable] = None,
        skills_dir: Optional[str] = None,
        goal_tree: Optional[GoalTree] = None,
        debug: bool = False,
        logger_name: Optional[str] = None,
        validation_policy: Optional[ValidationPolicy] = None,
        validator_search_provider: Optional[ValidatorWebSearchProvider] = None,
        validator_page_fetcher: Optional[PageFetcher] = None,
        artifact_resolver: Optional[ArtifactResolver] = None,
        application_binding: Any = None,
        context_provider: Any = None,
        candidate_service: Any = None,
        event_service: Any = None,
    ):
        """
        初始化 AgentRunner

        Args:
            trace_store: Trace 存储
            tool_registry: 工具注册表（默认使用全局注册表）
            llm_call: 主 LLM 调用函数
            utility_llm_call: 轻量 LLM（用于生成任务标题等），可选
            skills_dir: Skills 目录路径
            goal_tree: 初始 GoalTree（可选）
            debug: 保留参数（已废弃）
            logger_name: 自定义日志名称（如 "agents.knowledge_manager"），默认用模块名
            validation_policy: Recursive 根 Trace 固化的可信验收策略
            validator_search_provider: Validator 私有网页搜索适配器
            validator_page_fetcher: Validator 受控页面读取适配器
            artifact_resolver: Validator 只读产物解析器
        """
        self.trace_store = trace_store
        self.tools = tool_registry or get_tool_registry()
        self.llm_call = llm_call
        self.utility_llm_call = utility_llm_call
        self.skills_dir = skills_dir
        self.goal_tree = goal_tree
        self.debug = debug
        self.log = logging.getLogger(logger_name) if logger_name else logger
        self.validation_policy = validation_policy or ValidationPolicy()
        self.validator_search_provider = validator_search_provider
        self.validator_page_fetcher = validator_page_fetcher
        self.artifact_resolver = artifact_resolver
        self.application_binding = application_binding
        self.context_provider = context_provider
        self.candidate_service = candidate_service
        self.event_service = event_service
        self.stdin_check: Optional[Callable] = None  # 由外部设置，用于子 agent 执行期间检查 stdin
        self._cancel_events: Dict[str, asyncio.Event] = {}  # trace_id → cancel event
        self._recursive_active_traces: Dict[str, asyncio.Event] = {}
        self._active_children: Dict[str, set[str]] = {}
        self._active_parents: Dict[str, str] = {}

        # 知识保存跟踪（每个 trace 独立）
        self._saved_knowledge_ids: Dict[str, List[str]] = {}  # trace_id → [knowledge_ids]

        # Context 使用跟踪
        self._context_warned: Dict[str, set] = {}  # trace_id → {30, 50, 80} 已警告过的阈值
        self._context_usage: Dict[str, ContextUsage] = {}  # trace_id → 当前用量快照

        # 图片优化缓存（避免重复处理）
        # key: 图片内容的 hash, value: {"downscaled": ..., "description": ...}
        self._image_opt_cache: Dict[str, Dict[str, Any]] = {}

        self.resource_budget = (
            ResourceBudgetController(trace_store) if trace_store else None
        )
        self.task_protocol_service = (
            TaskProtocolService(trace_store, event_service) if trace_store else None
        )

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

    def get_context_usage(self, trace_id: str) -> Optional[ContextUsage]:
        """获取指定 trace 的 context 使用情况"""
        return self._context_usage.get(trace_id)

    async def _resource_budget_for_trace(
        self,
        trace_id: str,
    ) -> tuple[str, ResourceBudget] | None:
        """解析本地 Trace 所属 Recursive 根树的不可变预算快照。

        由模型调用、工具用量记账和子 Agent 创建入口调用；Legacy Trace 直接返回无预算。
        """
        if not self.trace_store:
            return None
        trace = await self.trace_store.get_trace(trace_id)
        if not trace or not policy_from_context(trace.context).requires_task_protocol:
            return None
        root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
        root = trace if root_trace_id == trace.trace_id else await self.trace_store.get_trace(root_trace_id)
        if not root:
            raise ResourceBudgetStateError(
                f"Recursive root Trace not found: {root_trace_id}"
            )
        snapshot = root.context.get(RESOURCE_BUDGET_CONTEXT_KEY)
        if snapshot is None:
            raise ResourceBudgetStateError(
                "Recursive tree has no persisted resource budget; create a new trace"
            )
        return root_trace_id, ResourceBudget.from_dict(snapshot)

    async def call_recursive_llm(
        self,
        trace_id: str,
        *,
        purpose: Literal["ordinary", "root_validator"] = "ordinary",
        call: Optional[Callable] = None,
        fail_on_post_response_exhaustion: bool = False,
        **kwargs: Any,
    ) -> Dict[str, Any]:
        """Recursive 树中统一的 LLM 调用和资源记账入口。

        Agent 主循环、上下文压缩、图片描述和 Validator 共用；请求前预留次数，响应后记账。
        """
        llm = call or self.llm_call
        if not llm:
            raise ValueError("llm_call function not provided")
        resolved = await self._resource_budget_for_trace(trace_id)
        if resolved is None:
            return await llm(**kwargs)
        root_trace_id, budget = resolved
        if not self.resource_budget:
            raise ResourceBudgetStateError("ResourceBudgetController is unavailable")
        await self.resource_budget.reserve_llm_call(
            root_trace_id,
            budget,
            purpose=purpose,
        )
        result = await llm(**kwargs)
        try:
            await self.resource_budget.record_llm_usage(
                root_trace_id,
                budget,
                prompt_tokens=int(result.get("prompt_tokens", 0) or 0),
                completion_tokens=int(result.get("completion_tokens", 0) or 0),
                cost_usd=float(result.get("cost", 0) or 0),
            )
        except ResourceBudgetExceeded as exc:
            if fail_on_post_response_exhaustion:
                raise
            result = dict(result)
            result["_resource_budget_exceeded"] = exc.dimension
        return result

    async def record_recursive_tool_usage(
        self,
        trace_id: str,
        tool_usage: Dict[str, Any],
    ) -> None:
        """登记工具内部自行发起的模型用量。

        Agent 主循环在 Tool Result 携带 ``tool_usage`` 时调用，并计入同一棵 Recursive 树。
        """
        resolved = await self._resource_budget_for_trace(trace_id)
        if resolved is None:
            return
        root_trace_id, budget = resolved
        if not self.resource_budget:
            raise ResourceBudgetStateError("ResourceBudgetController is unavailable")
        await self.resource_budget.record_external_llm_usage(
            root_trace_id,
            budget,
            prompt_tokens=int(tool_usage.get("prompt_tokens", 0) or 0),
            completion_tokens=int(tool_usage.get("completion_tokens", 0) or 0),
            cost_usd=float(tool_usage.get("cost", 0) or 0),
        )

    async def record_recursive_validation_usage(
        self,
        trace_id: str,
        *,
        tool_calls: int = 0,
        material_chars_count: int = 0,
        provider_cost_usd: float = 0.0,
    ) -> None:
        """把 Validator网页工具和真实材料字符计入同一棵树的预算。"""
        resolved = await self._resource_budget_for_trace(trace_id)
        if resolved is None:
            return
        root_trace_id, budget = resolved
        if not self.resource_budget:
            raise ResourceBudgetStateError("ResourceBudgetController is unavailable")
        await self.resource_budget.record_validation_usage(
            root_trace_id,
            budget,
            tool_calls=tool_calls,
            material_chars=material_chars_count,
            provider_cost_usd=provider_cost_usd,
        )

    async def validate_recursive_trace(
        self,
        evaluated_trace_id: str,
        *,
        scope: Optional[ValidationScope] = None,
        task_brief: Optional[Dict[str, Any]] = None,
        task_report: Optional[Dict[str, Any]] = None,
        completion_criteria: Optional[List[str]] = None,
        expected_outputs: Optional[List[str]] = None,
        candidate_output: Optional[str] = None,
        deterministic_failure: Optional[Dict[str, Any]] = None,
        root_validator: bool = False,
        candidate_ref: Any = None,
    ) -> ValidationRun:
        """编译Plan、解析材料、顺序运行Scope并持久化聚合缓存。"""
        if not self.trace_store or not self.llm_call:
            raise RuntimeError("Validator requires trace_store and llm_call")
        evaluated = await self.trace_store.get_trace(evaluated_trace_id)
        if not evaluated:
            raise ValueError(f"Trace not found: {evaluated_trace_id}")
        root_trace_id = evaluated.context.get("root_trace_id") or evaluated.trace_id
        root = (
            evaluated
            if root_trace_id == evaluated.trace_id
            else await self.trace_store.get_trace(root_trace_id)
        )
        if not root:
            raise ContextPolicyError(f"Recursive root Trace not found: {root_trace_id}")
        root_anchor = require_matching_root_task_anchor(
            root.context,
            evaluated.context,
        )
        policy, settings = require_validation_policy(root.context)
        state = ensure_task_protocol(evaluated.context)
        if candidate_ref is not None:
            from cyber_agent.application.candidate import CandidateRef

            candidate_ref = CandidateRef.model_validate(candidate_ref)
            if self.candidate_service is None:
                raise ValueError("Candidate validation requires CandidateService")
            if root_validator:
                raise ValueError("Candidate validation cannot be root validation")
        runtime_policy = policy_from_context(evaluated.context)
        authoritative_brief = state.get("task_brief")
        if authoritative_brief is not None:
            task_brief = authoritative_brief
        task_brief_version = int(state.get("task_brief_version", 0) or 0)
        progress_revision = (
            state.get("task_progress_head_revision")
            if root_validator or task_report is None
            else state.get("task_report_progress_revision")
        )
        task_progress = task_progress_at_revision(state, progress_revision)
        if runtime_policy.requires_task_progress and task_progress is None:
            raise ValueError("Recursive revision 3 validation requires TaskProgress")
        trajectory = await self.trace_store.get_main_path_messages(
            evaluated_trace_id,
            evaluated.head_sequence or evaluated.last_sequence,
        )

        refs: list[ArtifactRef] = []
        materials: list[ValidationMaterial] = []
        source_urls: list[str] = []
        material_issues: list[MaterialIssue] = []
        try:
            if isinstance(task_report, dict):
                refs.extend(extract_artifact_refs(task_report))
                source_urls = list(task_report.get("source_urls") or [])
            refs.extend(task_progress_artifact_refs(task_progress))
            if task_progress is not None:
                for item in (
                    *task_progress.questions,
                    *task_progress.blockers,
                    *task_progress.findings,
                    *task_progress.hypotheses,
                    *task_progress.work_items,
                ):
                    for ref in item.context_refs:
                        snapshot = get_authorized_context_snapshot(
                            evaluated.context,
                            ref_id=ref.ref_id,
                            version=ref.version,
                            root_trace_id=root_trace_id,
                            uid=evaluated.uid,
                        )
                        materials.append(ValidationMaterial(
                            artifact_id=ref.ref_id,
                            version=snapshot.version,
                            content_hash=snapshot.version,
                            kind=f"context.{snapshot.kind}",
                            mime_type="application/json",
                            root_trace_id=root_trace_id,
                            uid=evaluated.uid,
                            content=snapshot.content,
                        ))
            if root_validator:
                for message in trajectory:
                    if message.role == "tool" and isinstance(message.content, dict):
                        refs.extend(extract_artifact_refs(message.content))
        except Exception as exc:
            material_issues.append(MaterialIssue(
                artifact_id="artifact_metadata",
                outcome="error",
                reason=f"Invalid artifact metadata: {exc}",
            ))
        unique_refs = {
            (item.artifact_id, item.version, item.content_hash): item for item in refs
        }
        resolved_materials, resolved_issues = await resolve_artifact_refs(
            list(unique_refs.values()),
            resolver=self.artifact_resolver,
            root_trace_id=root_trace_id,
            uid=evaluated.uid,
        )
        materials.extend(resolved_materials)
        material_issues.extend(resolved_issues)
        candidate_material = None
        if candidate_ref is not None:
            try:
                candidate_material = await self.candidate_service.resolve_for_validation(
                    evaluated_trace_id,
                    candidate_ref,
                )
                materials = [
                    item for item in materials
                    if not (
                        item.artifact_id == candidate_material.artifact_id
                        and item.version == candidate_material.version
                    )
                ]
                materials.append(candidate_material)
                candidate_output = json.dumps(
                    candidate_material.content,
                    ensure_ascii=False,
                    sort_keys=True,
                    separators=(",", ":"),
                )
            except Exception as exc:
                material_issues.append(MaterialIssue(
                    artifact_id=candidate_ref.artifact_ref.artifact_id,
                    outcome="error",
                    reason=f"Candidate material cannot be resolved: {exc}",
                    scopes=["output"],
                ))
        total_material_chars = sum(material_chars(item) for item in materials)
        if deterministic_failure:
            material_issues.append(MaterialIssue(
                artifact_id="execution",
                outcome=deterministic_failure.get("outcome", "error"),
                reason=deterministic_failure.get("reason", "Task did not complete"),
            ))

        default_model = settings.validator_model or evaluated.model or ""
        root_model = settings.root_validator_model or default_model
        model_by_scope = {
            item: (root_model if item == "root" else default_model)
            for item in ("evidence", "hypothesis", "output", "task", "root")
        }
        if scope == "root" or root_validator:
            root_validator = True
        elif scope and task_brief is None:
            task_brief = {
                "completion_criteria": completion_criteria or [],
                "expected_outputs": expected_outputs or [],
                "validation_scopes": [] if scope == "task" else [scope],
            }
        validation_subject = None
        quality_specs: list[ValidationCheckSpec] = []
        quality_manifest: list[dict[str, Any]] = []
        fixed_checks: dict[str, ValidationCheck] = {}
        fixed_scope_errors: dict[ValidationScope, str] = {}
        quality_rules: tuple[Any, ...] = ()
        if candidate_ref is not None:
            from cyber_agent.application.quality import ValidationSubject

            validation_subject = ValidationSubject(
                subject_type="candidate",
                trace_id=evaluated_trace_id,
                candidate_ref=candidate_ref,
            )
            quality_rules = tuple(
                self.application_binding.application.quality_rules
            )
            quality_specs = [
                ValidationCheckSpec(
                    check_id=f"quality.{rule.rule_id}",
                    scope=rule.scope,
                    criterion=rule.criterion,
                    method="deterministic",
                )
                for rule in quality_rules
            ]
            quality_manifest = [
                item.model_dump(mode="json") for item in quality_rules
            ]
        plan = policy.compile_plan(
            task_brief=task_brief,
            task_brief_version=task_brief_version,
            root_task_anchor=root_anchor,
            task_report=task_report,
            candidate_output=candidate_output,
            evaluated_head_sequence=evaluated.head_sequence or evaluated.last_sequence,
            materials=materials,
            material_issues=material_issues,
            model_by_scope=model_by_scope,
            root=root_validator,
            task_progress=task_progress,
            subject=validation_subject,
            quality_checks=quality_specs,
            quality_manifest=quality_manifest,
        )

        cached = state.get("task_report_validation")
        if candidate_ref is not None:
            candidate_cached = await self.candidate_service.cached_validation(
                evaluated_trace_id,
                candidate_ref,
                plan.plan_hash,
            )
            if candidate_cached is not None:
                aggregate = ValidationResult.model_validate(
                    candidate_cached.validation_result
                )
                return ValidationRun(
                    result=aggregate,
                    trace_ids=[
                        item.validator_trace_id
                        for item in aggregate.scope_results
                    ],
                    cached=True,
                )
            cached = None
            if quality_rules and candidate_material is not None:
                fixed_checks, fixed_scope_errors = await self._run_quality_checks(
                    validation_subject,
                    candidate_material,
                    quality_rules,
                )
        validation_cache: dict[str, Any]
        resume_scope_results: list[ScopeValidationResult] = []
        if isinstance(cached, dict) and cached.get("plan_hash") == plan.plan_hash:
            validation_cache = cached
            try:
                aggregate_raw = cached.get("aggregate_result")
                if aggregate_raw:
                    aggregate = ValidationResult.model_validate(aggregate_raw)
                    if (
                        aggregate.evaluated_trace_id == evaluated_trace_id
                        and aggregate.plan_hash == plan.plan_hash
                    ):
                        return ValidationRun(
                            result=aggregate,
                            trace_ids=[
                                item.validator_trace_id
                                for item in aggregate.scope_results
                            ],
                            cached=True,
                        )
                resume_scope_results = [
                    ScopeValidationResult.model_validate(item)
                    for item in cached.get("scope_results", [])
                ]
            except Exception:
                resume_scope_results = []
        else:
            validation_cache = {
                "validation_plan": plan.model_dump(mode="json"),
                "plan_hash": plan.plan_hash,
                "scope_results": [],
                "aggregate_result": None,
                "validated_at_sequence": plan.evaluated_head_sequence,
                "material_usage_recorded": total_material_chars == 0,
            }
            if candidate_ref is None:
                state["task_report_validation"] = validation_cache
                await self.trace_store.update_trace(
                    evaluated_trace_id,
                    context=evaluated.context,
                )

        if (
            total_material_chars
            and not validation_cache.get("material_usage_recorded", False)
        ):
            await self.record_recursive_validation_usage(
                evaluated_trace_id,
                material_chars_count=total_material_chars,
            )
            if candidate_ref is not None:
                validation_cache["material_usage_recorded"] = True
            else:
                fresh = await self.trace_store.get_trace(evaluated_trace_id)
                if not fresh:
                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
                fresh_state = ensure_task_protocol(fresh.context)
                fresh_cache = fresh_state.get("task_report_validation")
                if (
                    not isinstance(fresh_cache, dict)
                    or fresh_cache.get("plan_hash") != plan.plan_hash
                ):
                    raise ValueError("Validation cache changed while recording materials")
                fresh_cache["material_usage_recorded"] = True
                await self.trace_store.update_trace(
                    evaluated_trace_id,
                    context=fresh.context,
                )

        lineage_event = None
        if (
            evaluated.parent_trace_id
            and evaluated_trace_id not in self._recursive_active_traces
        ):
            lineage_event = self.register_recursive_child(
                evaluated.parent_trace_id,
                evaluated_trace_id,
            )

        async def validator_llm_call(**kwargs: Any) -> Dict[str, Any]:
            result = await self.call_recursive_llm(
                evaluated_trace_id,
                purpose="root_validator" if root_validator else "ordinary",
                **kwargs,
            )
            dimension = result.get("_resource_budget_exceeded")
            if dimension:
                raise RuntimeError(
                    f"Validator exceeded tree resource budget: {dimension}"
                )
            return result

        provider: ValidatorWebSearchProvider | None
        if settings.search_provider == "disabled":
            provider = None
        else:
            provider = self.validator_search_provider or SerperWebSearchProvider()

        def tool_session_factory(
            validation_scope: ValidationScope,
            allowed_urls: set[str],
            validator_trace_id: str,
        ) -> ValidatorToolSession | None:
            limits_by_scope = {
                "evidence": ValidatorToolLimits(5, 10, 15),
                "hypothesis": ValidatorToolLimits(2, 4, 6),
                "root": ValidatorToolLimits(2, 5, 7),
            }
            limits = limits_by_scope.get(validation_scope)
            if limits is None:
                return None

            async def record_usage(
                tool_calls: int,
                chars: int,
                provider_cost_usd: float,
            ) -> None:
                await self.record_recursive_validation_usage(
                    evaluated_trace_id,
                    tool_calls=tool_calls,
                    material_chars_count=chars,
                    provider_cost_usd=provider_cost_usd,
                )

            session_kwargs: Dict[str, Any] = {}
            if self.validator_page_fetcher is not None:
                session_kwargs["page_fetcher"] = self.validator_page_fetcher
            return ValidatorToolSession(
                provider=provider,
                allowed_urls=allowed_urls,
                limits=limits,
                usage_recorder=record_usage,
                **session_kwargs,
            )

        validator = LLMValidator(
            llm_call=validator_llm_call,
            trace_store=self.trace_store,
            policy=policy,
            tool_session_factory=tool_session_factory,
            cancel_check=self.is_cancel_requested,
            trace_register=self.register_recursive_child,
            trace_release=self.release_recursive_trace,
        )
        try:
            async def persist_scope(result: ScopeValidationResult) -> None:
                if candidate_ref is not None:
                    by_scope = {
                        item.get("scope"): item
                        for item in validation_cache.get("scope_results", [])
                        if isinstance(item, dict)
                    }
                    by_scope[result.scope] = result.model_dump(mode="json")
                    validation_cache["scope_results"] = [
                        by_scope[item]
                        for item in plan.effective_scopes
                        if item in by_scope
                    ]
                    return
                fresh = await self.trace_store.get_trace(evaluated_trace_id)
                if not fresh:
                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
                fresh_state = ensure_task_protocol(fresh.context)
                cache = fresh_state.get("task_report_validation")
                if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
                    raise ValueError("Validation cache changed while scopes were running")
                by_scope = {
                    item.get("scope"): item
                    for item in cache.get("scope_results", [])
                    if isinstance(item, dict)
                }
                by_scope[result.scope] = result.model_dump(mode="json")
                cache["scope_results"] = [
                    by_scope[item]
                    for item in plan.effective_scopes
                    if item in by_scope
                ]
                await self.trace_store.update_trace(
                    evaluated_trace_id,
                    context=fresh.context,
                )

            run = await validator.validate_plan(
                evaluated_trace=evaluated,
                trajectory=trajectory,
                plan=plan,
                root_task_anchor=root_anchor,
                task_brief=task_brief,
                task_report=task_report,
                task_progress=task_progress,
                candidate_output=candidate_output,
                materials=materials,
                material_issues=material_issues,
                model_by_scope=model_by_scope,
                source_urls=source_urls,
                resume_scope_results=resume_scope_results,
                on_scope_result=persist_scope,
                fixed_checks=fixed_checks,
                fixed_scope_errors=fixed_scope_errors,
            )
            if candidate_ref is not None:
                from cyber_agent.application.quality import CandidateValidationRecord

                await self.candidate_service.record_validation(
                    evaluated_trace_id,
                    CandidateValidationRecord(
                        candidate_ref=candidate_ref,
                        plan_hash=plan.plan_hash,
                        validation_result=run.result.model_dump(mode="json"),
                        validated_at_sequence=plan.evaluated_head_sequence,
                    ),
                )
                return run
            fresh = await self.trace_store.get_trace(evaluated_trace_id)
            if not fresh:
                raise ValueError(f"Trace not found: {evaluated_trace_id}")
            fresh_state = ensure_task_protocol(fresh.context)
            cache = fresh_state.get("task_report_validation")
            if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
                raise ValueError("Validation cache changed before aggregation")
            cache["aggregate_result"] = run.result.model_dump(mode="json")
            cache["scope_results"] = [
                item.model_dump(mode="json") for item in run.result.scope_results
            ]
            await self.trace_store.update_trace(
                evaluated_trace_id,
                context=fresh.context,
            )
            if self.event_service is not None:
                await self.event_service.emit_after_commit(
                    source_trace_id=evaluated_trace_id,
                    event_type="validation.completed",
                    event_key=(
                        f"validation.completed:trace:{evaluated_trace_id}:"
                        f"{plan.plan_hash}"
                    ),
                    effective_at_sequence=plan.evaluated_head_sequence,
                    payload={
                        "subject_type": "trace",
                        "validation_result": run.result.model_dump(mode="json"),
                    },
                )
            return run
        finally:
            if lineage_event is not None:
                self.release_recursive_trace(evaluated_trace_id, lineage_event)

    async def _run_quality_checks(
        self,
        subject: Any,
        material: ValidationMaterial,
        rules: tuple[Any, ...],
    ) -> tuple[dict[str, ValidationCheck], dict[ValidationScope, str]]:
        """Execute one frozen quality batch and reject provider-shaped plans."""
        from cyber_agent.application.quality import (
            QualityCheckInput,
            QualityCheckOutcome,
        )

        provider = self.application_binding.services.quality_provider
        check_ids = {rule.rule_id: f"quality.{rule.rule_id}" for rule in rules}

        def provider_error(reason: str):
            return (
                {
                    check_id: ValidationCheck(
                        check_id=check_id,
                        status="unknown",
                        issue=reason,
                    )
                    for check_id in check_ids.values()
                },
                {"output": reason},
            )

        if provider is None:
            return provider_error("QualityCheckProvider is unavailable")
        request = QualityCheckInput(
            subject=subject,
            rules=rules,
            material=material,
        )
        try:
            raw = await asyncio.wait_for(
                provider.check(request),
                timeout=max(rule.timeout_seconds for rule in rules),
            )
            outcomes = [QualityCheckOutcome.model_validate(item) for item in raw]
        except Exception as exc:
            return provider_error(f"QualityCheckProvider failed: {exc}")
        returned = [(item.rule_id, item.rule_version) for item in outcomes]
        expected = [(item.rule_id, item.version) for item in rules]
        if len(returned) != len(set(returned)) or set(returned) != set(expected):
            extras = sorted(set(returned) - set(expected))
            missing = sorted(set(expected) - set(returned))
            return provider_error(
                "QualityCheckProvider results do not match the frozen rules: "
                f"extras={extras}, missing={missing}"
            )
        by_rule = {item.rule_id: item for item in outcomes}
        checks: dict[str, ValidationCheck] = {}
        errors: dict[ValidationScope, str] = {}
        for rule in rules:
            outcome = by_rule[rule.rule_id]
            check_id = check_ids[rule.rule_id]
            checks[check_id] = ValidationCheck(
                check_id=check_id,
                status=("unknown" if outcome.status == "error" else outcome.status),
                evidence_refs=list(outcome.evidence_refs),
                issue=outcome.issue,
            )
            if outcome.status == "error":
                errors[rule.scope] = outcome.issue or "Quality check failed"
        try:
            await self.record_recursive_validation_usage(
                subject.trace_id,
                tool_calls=len(rules),
            )
        except Exception as exc:
            return provider_error(f"Quality validation budget failed: {exc}")
        return checks, errors

    async def dream(
        self,
        memory_config: MemoryConfig,
        *,
        uid: Optional[str],
        agent_type: str,
        reflect_model: str = "gpt-4o-mini",
        dream_model: str = "gpt-4o",
    ) -> "DreamReport":
        """执行 dream（整理长期记忆）——外部调度入口。

        Agent 主动调用走 dream 工具；外部调度（定时器、CLI）走这个方法。

        Args:
            memory_config: 记忆配置
            uid/agent_type: 与 MemoryConfig 一起形成强制 Dream 身份边界
            reflect_model: per-trace 反思模型
            dream_model: 跨 trace 整合模型
        """
        from cyber_agent.core.dream import DreamScope, run_dream
        if not self.trace_store or not self.llm_call:
            raise RuntimeError("dream 需要 trace_store 和 llm_call 均已配置")
        return await run_dream(
            store=self.trace_store,
            llm_call=self.llm_call,
            memory_config=memory_config,
            dream_scope=DreamScope(
                uid=uid,
                agent_type=agent_type,
                memory_identity=compute_memory_identity(memory_config),
            ),
            reflect_model=reflect_model,
            dream_model=dream_model,
        )

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

        Args:
            messages: OpenAI SDK 格式的输入消息
                新建: 初始任务消息 [{"role": "user", "content": "..."}]
                续跑: 追加的新消息
                回溯: 在插入点之后追加的消息
            config: 运行配置
            inject_skills: 本次调用需要指定注入的 skill 列表（skill 名称）
            skill_recency_threshold: 最近 N 条消息内有该 skill 就不重复注入

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

        config = config or RunConfig()
        trace = None
        run_cancel_event: Optional[asyncio.Event] = None

        try:
            # Phase 1: PREPARE TRACE
            trace, goal_tree, sequence = await self._prepare_trace(messages, config)
            # 子 Trace 可能已在排队阶段收到停止信号，不能覆盖既有 Event。
            run_cancel_event = self._cancel_events.setdefault(
                trace.trace_id,
                asyncio.Event(),
            )
            if policy_from_context(trace.context).mode is AgentMode.RECURSIVE:
                self._recursive_active_traces[trace.trace_id] = run_cancel_event
                if trace.parent_trace_id:
                    self._active_children.setdefault(
                        trace.parent_trace_id,
                        set(),
                    ).add(trace.trace_id)
                    self._active_parents[trace.trace_id] = trace.parent_trace_id
            yield trace

            # 检查是否有未完成的侧分支（用于用户追加消息场景）
            side_branch_ctx_for_build: Optional[SideBranchContext] = None
            if trace.context.get("active_side_branch") and messages:
                side_branch_data = trace.context["active_side_branch"]

                # 创建侧分支上下文（用于标记用户追加的消息）
                side_branch_ctx_for_build = SideBranchContext(
                    type=side_branch_data["type"],
                    branch_id=side_branch_data["branch_id"],
                    start_head_seq=side_branch_data["start_head_seq"],
                    start_sequence=side_branch_data["start_sequence"],
                    start_history_length=0,
                    start_iteration=side_branch_data.get("start_iteration", 0),
                    max_turns=side_branch_data.get("max_turns", config.side_branch_max_turns),
                    turns_used=side_branch_data.get("turns_used", 0),
                )

            # Phase 2: BUILD HISTORY
            history, sequence, created_messages, head_seq = await self._build_history(
                trace.trace_id, messages, goal_tree, config, sequence, side_branch_ctx_for_build
            )
            # 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,
                inject_skills=inject_skills,
                skill_recency_threshold=skill_recency_threshold,
            ):
                yield event

        except asyncio.CancelledError:
            if trace and config.approval_batch_id:
                await asyncio.shield(
                    self._mark_approval_execution_unknown(trace.trace_id)
                )
            raise
        except Exception as e:
            self.log.error(f"Agent run failed: {e}")
            # Preparation rejections (for example, attempting to resume an
            # immutable Validator Trace) must not rewrite the existing record.
            tid = trace.trace_id if trace else None
            if self.trace_store and tid:
                approval_unknown = (
                    await self._mark_approval_execution_unknown(tid)
                    if config.approval_batch_id
                    else False
                )
                # 读取当前 last_sequence 作为 head_sequence，确保续跑时能加载完整历史
                current = await self.trace_store.get_trace(tid)
                head_seq = current.last_sequence if current else None
                updates: Dict[str, Any] = {
                    "status": "failed",
                    "head_sequence": head_seq,
                    "error_message": (
                        "Tool execution outcome is unknown; automatic retry was refused"
                        if approval_unknown
                        else str(e)
                    ),
                    "completed_at": datetime.now(),
                }
                if isinstance(e, ResourceBudgetExceeded):
                    current_context = dict(current.context if current else {})
                    current_context["termination_reason"] = (
                        f"budget_exhausted:{e.dimension}"
                    )
                    updates["context"] = current_context
                await self.trace_store.update_trace(
                    tid,
                    **updates,
                )
                trace_obj = await self.trace_store.get_trace(tid)
                if trace_obj:
                    yield trace_obj
            raise
        finally:
            if trace and run_cancel_event is not None:
                self.release_recursive_trace(trace.trace_id, run_cancel_event)

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

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

        Args:
            on_event: 可选回调，每个 Trace/Message 事件触发一次，用于实时输出子 Agent 执行过程。
            inject_skills: 本次调用需要指定注入的 skill 列表（透传给 run()）。
        """
        last_assistant_text = ""
        final_trace: Optional[Trace] = None

        async for item in self.run(messages=messages, config=config, inject_skills=inject_skills):
            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

        stopped_recursively = bool(
            final_trace
            and status == "stopped"
            and policy_from_context(final_trace.context).mode is AgentMode.RECURSIVE
        )
        if not summary and stopped_recursively:
            summary = "Agent execution stopped."
        elif not summary and status == "waiting_confirmation":
            summary = "Agent is waiting for local tool approval."
        elif not summary:
            status = "failed"
            error = error or "Agent 没有产生 assistant 文本结果"

        # 获取保存的知识 ID
        trace_id = final_trace.trace_id if final_trace else config.trace_id
        saved_knowledge_ids = self._saved_knowledge_ids.get(trace_id, [])

        return {
            "status": status,
            "summary": summary,
            "trace_id": trace_id,
            "error": error,
            "saved_knowledge_ids": saved_knowledge_ids,  # 新增：返回保存的知识 ID
            "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

        Trace API 定位实际 Runner 后调用本方法；Legacy 只停当前 Trace，
        Recursive 由 ``request_stop`` 把信号传给当前进程内已登记的子树。

        Returns:
            True 如果成功发送停止信号，False 如果该 trace 不在运行中
        """
        return self.request_stop(trace_id)

    def request_stop(self, trace_id: str) -> bool:
        """同步设置停止信号，供 API 与 stdin 回调共用。

        Recursive Trace 会沿进程内父子登记表向下遍历，不影响父级或兄弟分支。
        """
        if trace_id not in self._cancel_events:
            return False

        if trace_id not in self._recursive_active_traces:
            self._cancel_events[trace_id].set()
            return True

        pending = [trace_id]
        visited = set()
        while pending:
            current = pending.pop()
            if current in visited:
                continue
            visited.add(current)
            event = self._cancel_events.get(current)
            if event:
                event.set()
            pending.extend(tuple(self._active_children.get(current, ())))
        return True

    def register_recursive_child(
        self,
        parent_trace_id: str,
        child_trace_id: str,
    ) -> asyncio.Event:
        """登记已创建但可能仍在排队的 Recursive 直属孩子。

        ``agent`` 工具预创建孩子后、Runner 启动 Validator 前调用，使父级停止能传递到后代。
        """
        event = self._cancel_events.setdefault(child_trace_id, asyncio.Event())
        self._recursive_active_traces[child_trace_id] = event
        self._active_children.setdefault(parent_trace_id, set()).add(child_trace_id)
        self._active_parents[child_trace_id] = parent_trace_id
        parent_event = self._cancel_events.get(parent_trace_id)
        if parent_event and parent_event.is_set():
            event.set()
        return event

    def is_cancel_requested(self, trace_id: str) -> bool:
        event = self._cancel_events.get(trace_id)
        return bool(event and event.is_set())

    def unregister_recursive_trace(
        self,
        trace_id: str,
        event: Optional[asyncio.Event] = None,
    ) -> None:
        """幂等清理运行登记；event 防止旧运行误删续跑状态。"""
        current_event = self._recursive_active_traces.get(trace_id)
        if event is not None and current_event is not event:
            return
        self._recursive_active_traces.pop(trace_id, None)
        parent_id = self._active_parents.pop(trace_id, None)
        if parent_id:
            children = self._active_children.get(parent_id)
            if children:
                children.discard(trace_id)
                if not children:
                    self._active_children.pop(parent_id, None)
        if not self._active_children.get(trace_id):
            self._active_children.pop(trace_id, None)

    def release_recursive_trace(
        self,
        trace_id: str,
        event: Optional[asyncio.Event] = None,
    ) -> None:
        """完成一次运行并按对象身份释放取消 Event 与父子登记。"""
        current_event = self._cancel_events.get(trace_id)
        if event is not None and current_event is not event:
            return
        self.unregister_recursive_trace(trace_id, event)
        if self._cancel_events.get(trace_id) is current_event:
            self._cancel_events.pop(trace_id, None)

    async def _mark_trace_stopped(
        self,
        trace_id: str,
        head_sequence: Optional[int],
    ) -> Optional[Trace]:
        if not self.trace_store:
            return None
        await self.trace_store.update_trace(
            trace_id,
            status="stopped",
            head_sequence=head_sequence,
            completed_at=datetime.now(),
        )
        try:
            from cyber_agent.trace.websocket import broadcast_trace_status_changed
            await broadcast_trace_status_changed(trace_id, "stopped")
        except Exception:
            pass
        return await self.trace_store.get_trace(trace_id)

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

    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：为 ``run`` 选择新建、续跑或回溯路径。

        在 Agent 主循环前调用，并在任何 Trace 操作前校验已废弃的配置项。

        Returns:
            (trace, goal_tree, next_sequence)
        """
        assert_removed_config_absent()
        if self.application_binding is not None and any(
            message.get("role") == "system" for message in messages
        ):
            raise ValueError(
                "Application runs do not allow caller-provided system messages"
            )
        if config.trace_id:
            return await self._prepare_existing_trace(config)
        else:
            return await self._prepare_new_trace(messages, config)

    @staticmethod
    def _application_ref_dict(value: Any) -> dict[str, Any] | None:
        if value is None:
            return None
        if hasattr(value, "model_dump"):
            return value.model_dump(mode="json")
        return dict(value) if isinstance(value, dict) else None

    def _validate_application_config(self, config: RunConfig) -> None:
        """Reject a new application RunConfig that diverges from its Binding."""
        binding = self.application_binding
        if binding is None:
            raise ValueError("Application binding is unavailable")
        expected_ref = binding.application_ref.model_dump(mode="json")
        if self._application_ref_dict(config.application_ref) != expected_ref:
            raise ValueError("RunConfig ApplicationRef does not match Runner binding")
        if not config.role_id:
            raise ValueError("Application role_id is required")
        role = binding.role(config.role_id)
        if config.role_hash != role.role_hash:
            raise ValueError("RunConfig role hash does not match Runner binding")
        if (
            config.model != role.role.model
            or config.temperature != role.role.temperature
            or config.extra_llm_params != role.role.model_parameters
            or not set(config.tools or []).issubset(set(role.tool_names))
            or config.system_prompt != role.system_prompt
            or config.skills != []
        ):
            raise ValueError("RunConfig behavior does not match the bound application role")
        limits = dict(config.effective_run_limits)
        if not limits:
            raise ValueError("Application effective_run_limits are required")
        allowed_limits = role.effective_limits.model_dump(mode="json")
        for name, allowed in allowed_limits.items():
            value = limits.get(name)
            if value is None or value > allowed:
                raise ValueError(
                    f"Application run limit exceeds the bound role: {name}"
                )
        if (
            config.max_iterations != limits["max_iterations"]
            or config.max_parallel_children != limits["max_parallel_children"]
        ):
            raise ValueError("RunConfig limits do not match effective_run_limits")

    def _validate_application_snapshot(
        self,
        snapshot: RunConfigSnapshotV1 | RunConfigSnapshotV2,
        trace: Trace,
    ) -> None:
        """Perform the Runner-side binding gate before any resume mutation."""
        if isinstance(snapshot, RunConfigSnapshotV1) and not isinstance(
            snapshot,
            RunConfigSnapshotV2,
        ):
            if self.application_binding is not None:
                raise ValueError("Application-bound Runner cannot resume a V1 Trace")
            return
        if self.application_binding is None:
            raise ValueError(
                "Application Trace requires an ApplicationRuntime-bound Runner"
            )
        expected_ref = self.application_binding.application_ref.model_dump(mode="json")
        if snapshot.application_ref != expected_ref:
            raise ValueError("ApplicationRef does not match Runner binding")
        role = self.application_binding.role(snapshot.role_id)
        if snapshot.role_hash != role.role_hash:
            raise ValueError("Application role hash does not match Runner binding")
        if (
            trace.context.get("application_ref") != snapshot.application_ref
            or trace.context.get("application_role_id") != snapshot.role_id
            or trace.context.get("application_role_hash") != snapshot.role_hash
            or trace.context.get("effective_run_limits")
            != snapshot.effective_run_limits
        ):
            raise ValueError("Application snapshot does not match Trace context")
        restored = RunConfig()
        restored.apply_snapshot(snapshot)
        restored.system_prompt = role.system_prompt
        restored.skills = []
        self._validate_application_config(restored)

    async def _prepare_new_trace(
        self,
        messages: List[Dict],
        config: RunConfig,
    ) -> Tuple[Trace, Optional[GoalTree], int]:
        """创建并持久化一个新根 Trace。

        ``run`` 首次执行时调用；Recursive 会在此固化根任务锚点和树级预算。
        """
        if self.application_binding is not None:
            self._validate_application_config(config)
        elif config.application_ref is not None:
            raise ValueError(
                "Application runs require an ApplicationRuntime-bound Runner"
            )
        # 在任何标题生成/LLM 调用前完成模式校验。
        policy = policy_from_environment(
            recursive_revision=CURRENT_RECURSIVE_REVISION,
        )
        if policy.mode is AgentMode.RECURSIVE:
            validate_recursive_child_execution(
                config.child_execution_mode,
                config.max_parallel_children,
            )
            if config.root_task_anchor is None:
                raise ValueError(
                    "New Recursive root traces require root_task_anchor"
                )
            try:
                root_task_anchor = normalize_root_task_anchor(
                    config.root_task_anchor
                )
            except ContextPolicyError as exc:
                raise ValueError(str(exc)) from exc
            deployment_budget = ResourceBudget.from_environment()
            if self.application_binding is not None:
                limits = config.effective_run_limits
                budget = ResourceBudget(
                    enabled=deployment_budget.enabled,
                    max_total_agents=int(limits["max_total_agents"]),
                    max_llm_calls=int(limits["max_llm_calls"]),
                    max_total_tokens=int(limits["max_total_tokens"]),
                    max_total_cost_usd=float(limits["max_total_cost_usd"]),
                    max_duration_seconds=int(limits["max_duration_seconds"]),
                    reserved_final_calls=min(
                        deployment_budget.reserved_final_calls,
                        int(limits["max_llm_calls"]) - 1,
                    ),
                    max_validation_tool_calls=int(
                        limits["max_validation_tool_calls"]
                    ),
                    max_validation_material_chars=int(
                        limits["max_validation_material_chars"]
                    ),
                )
            else:
                budget = deployment_budget
            validator_settings = ValidatorSettings.from_environment()
        trace_id = str(uuid.uuid4())

        # 生成任务名称
        task_name = config.name or (
            self._fallback_task_name(messages)
            if policy.mode is AgentMode.RECURSIVE
            else await self._generate_task_name(messages)
        )

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

        trace_context = apply_policy_to_context(config.context, policy)
        trace_context.setdefault("agent_depth", 0)
        trace_context.setdefault("root_trace_id", trace_id)
        memory_identity = (
            compute_memory_identity(config.memory) if config.memory else None
        )
        if memory_identity is not None:
            trace_context[MEMORY_IDENTITY_CONTEXT_KEY] = memory_identity
        if self.application_binding is not None:
            run_snapshot = RunConfigSnapshotV2.from_run_config(
                config,
                memory_identity=memory_identity,
            )
            trace_context.update({
                "application_ref": run_snapshot.application_ref,
                "application_role_id": run_snapshot.role_id,
                "application_role_hash": run_snapshot.role_hash,
                "effective_run_limits": run_snapshot.effective_run_limits,
            })
        else:
            run_snapshot = RunConfigSnapshotV1.from_run_config(
                config,
                memory_identity=memory_identity,
            )
        persist_run_config_snapshot(trace_context, run_snapshot)
        if policy.requires_task_protocol:
            persist_root_task_anchor(trace_context, root_task_anchor)
            persist_validation_policy(
                trace_context,
                self.validation_policy,
                validator_settings,
            )
            trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict()
            trace_context["task_protocol"] = new_task_protocol()
            state = ensure_task_protocol(trace_context)
            if policy.requires_task_progress:
                initialize_task_progress(
                    state,
                    root_task_anchor_hash=trace_context.get(
                        "root_task_anchor_hash"
                    ),
                )
            replace_context_access(
                trace_context,
                [],
                root_task_anchor=root_task_anchor,
                task_brief=state.get("task_brief"),
            )
            if self.application_binding is not None and self.context_provider is not None:
                from cyber_agent.application.context import load_application_context
                from cyber_agent.application.ports import ContextRequest

                await load_application_context(
                    self.application_binding,
                    trace_context,
                    ContextRequest(
                        application_ref=self.application_binding.application_ref,
                        root_trace_id=trace_id,
                        trace_id=trace_id,
                        uid=config.uid,
                        role_id=config.role_id,
                        task_brief=None,
                        task_brief_revision=0,
                    ),
                    root_task_anchor=root_task_anchor,
                    task_brief=None,
                    granted_at_sequence=0,
                )

        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},
            context=trace_context,
            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)
            assert self.resource_budget is not None
            if policy.mode is AgentMode.RECURSIVE:
                await self.resource_budget.initialize(
                    trace_id,
                    budget,
                    initial_agents=1,
                )

        return trace_obj, goal_tree, 1

    async def _prepare_existing_trace(
        self,
        config: RunConfig,
    ) -> Tuple[Trace, Optional[GoalTree], int]:
        """加载已有 Trace，决定续跑或回溯。

        ``run`` 携带 ``trace_id`` 时调用；Recursive 只信任持久化模式、预算和协议状态。
        """
        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}")
        if (
            trace_obj.agent_type == "validator"
            or trace_obj.context.get("created_by_tool") == "validator"
        ):
            raise ValueError(
                "Validator traces cannot be continued or rewound"
            )
        require_mutable_trace_policy(trace_obj.context)

        # Resume/rewind behavior is always recovered from the immutable snapshot.
        # Old Recursive traces cannot be reconstructed safely; Legacy traces get
        # one explicitly-marked, conservative inferred snapshot for compatibility.
        if RUN_CONFIG_SNAPSHOT_CONTEXT_KEY in trace_obj.context:
            try:
                snapshot = load_run_config_snapshot(trace_obj.context)
            except RunConfigSnapshotError as exc:
                raise ValueError(str(exc)) from exc
        else:
            existing_policy = policy_from_context(trace_obj.context)
            if existing_policy.mode is AgentMode.RECURSIVE:
                raise ValueError(
                    "This Recursive trace predates RunConfig snapshots; create a new trace"
                )
            inferred = RunConfig(
                model=trace_obj.model or "gpt-4o",
                temperature=float((trace_obj.llm_params or {}).get("temperature", 0.3)),
                tools=[
                    item.get("function", {}).get("name")
                    for item in (trace_obj.tools or [])
                    if item.get("function", {}).get("name")
                ],
                tool_groups=None,
                agent_type=trace_obj.agent_type or "default",
                uid=trace_obj.uid,
                extra_llm_params={
                    key: value
                    for key, value in (trace_obj.llm_params or {}).items()
                    if key != "temperature"
                },
                context={
                    "project_name": trace_obj.context.get("project_name")
                } if trace_obj.context.get("project_name") else {},
            )
            snapshot = RunConfigSnapshotV1.from_run_config(
                inferred,
                memory_identity=None,
                legacy_inferred=True,
            )
            persist_run_config_snapshot(trace_obj.context, snapshot)
            await self.trace_store.update_trace(
                trace_obj.trace_id,
                context=trace_obj.context,
            )
        if snapshot.uid != trace_obj.uid or snapshot.agent_type != (trace_obj.agent_type or "default"):
            raise ValueError("run config snapshot identity does not match Trace metadata")
        self._validate_application_snapshot(snapshot, trace_obj)
        config.apply_snapshot(snapshot)
        if isinstance(snapshot, RunConfigSnapshotV2):
            role = self.application_binding.role(snapshot.role_id)
            config.system_prompt = role.system_prompt
            config.skills = []
        persisted_memory_identity = trace_obj.context.get(
            MEMORY_IDENTITY_CONTEXT_KEY
        )
        if persisted_memory_identity != snapshot.memory_identity:
            raise ValueError(
                "run config snapshot memory identity does not match Trace context"
            )
        if config.memory is not None and (
            compute_memory_identity(config.memory) != snapshot.memory_identity
        ):
            raise ValueError(
                "restored MemoryConfig does not match the persisted memory identity"
            )

        # Historical traces predate AGENT_MODE.  They resume in safe Legacy mode
        # and the choice is persisted immediately; environment changes never
        # mutate an existing tree.
        if AGENT_MODE_CONTEXT_KEY not in trace_obj.context:
            legacy_policy = policy_from_context(None)
            trace_obj.context = apply_policy_to_context(trace_obj.context, legacy_policy)
            await self.trace_store.update_trace(
                config.trace_id,
                context=trace_obj.context,
            )
        else:
            policy = policy_from_context(trace_obj.context)
            if policy.mode is AgentMode.RECURSIVE:
                validate_recursive_child_execution(
                    config.child_execution_mode,
                    config.max_parallel_children,
                )
                if policy.requires_task_protocol:
                    root_trace_id = trace_obj.context.get("root_trace_id")
                    root = (
                        trace_obj
                        if root_trace_id == trace_obj.trace_id
                        else await self.trace_store.get_trace(root_trace_id)
                    )
                    if not root:
                        raise ValueError(
                            f"Recursive root Trace not found: {root_trace_id}"
                        )
                    try:
                        require_matching_root_task_anchor(
                            root.context,
                            trace_obj.context,
                        )
                        require_validation_policy(root.context)
                    except ContextPolicyError as exc:
                        raise ValueError(str(exc)) from exc
                    except ValueError as exc:
                        raise ValueError(str(exc)) from exc
                    if RESOURCE_BUDGET_CONTEXT_KEY not in root.context:
                        raise ValueError(
                            "This experimental Recursive trace predates tree resource "
                            "budgets; create a new trace"
                        )
                    ResourceBudget.from_dict(
                        root.context[RESOURCE_BUDGET_CONTEXT_KEY]
                    )
                    if root_trace_id == trace_obj.trace_id:
                        state = ensure_task_protocol(trace_obj.context)
                        if state["root_validation_attempts"] >= 2:
                            raise ValueError(
                                "Root task already used its two independent "
                                "validation attempts; create a new trace"
                            )
                    if policy.requires_task_progress:
                        state = ensure_task_protocol(trace_obj.context)
                        had_report = state.get("task_report") is not None
                        report_progress_revision = state.get(
                            "task_report_progress_revision"
                        )
                        if had_report and report_progress_revision is None:
                            raise ValueError(
                                "TaskReport is missing its TaskProgress revision binding"
                            )
                        previous_head = state.get("task_progress_head_revision")
                        rewind_task_progress(state, trace_obj.last_sequence)
                        if (
                            had_report
                            and state.get("task_report_progress_revision") is None
                        ):
                            raise ValueError(
                                "TaskReport references TaskProgress outside the "
                                "current revision ancestry"
                            )
                        if state.get("task_progress_head_revision") != previous_head:
                            await self.trace_store.update_trace(
                                trace_obj.trace_id,
                                context=trace_obj.context,
                            )
                    assert self.resource_budget is not None
                    await self.resource_budget.get_usage(root.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"
        # 广播状态变化给前端
        try:
            from cyber_agent.trace.websocket import broadcast_trace_status_changed
            await broadcast_trace_status_changed(config.trace_id, "running")
        except Exception:
            pass

        return trace_obj, goal_tree, sequence

    # ===== Phase 2: BUILD HISTORY =====
   
    async def _build_history(
        self,
        trace_id: str,
        new_messages: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
        side_branch_ctx: Optional[SideBranchContext] = None,
    ) -> Tuple[List[Dict], int, List[Message], int]:
        """
        构建完整的 LLM 消息历史

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

        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
                )

                # An approved pending batch deliberately has orphaned calls;
                # they are executed before the next model turn.
                if not config.approval_batch_id:
                    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
                snapshot = load_run_config_snapshot(trace_obj.context)
                if snapshot.system_prompt_hash:
                    persisted_system = next(
                        (
                            message.get("content")
                            for message in history
                            if message.get("role") == "system"
                            and isinstance(message.get("content"), str)
                        ),
                        None,
                    )
                    if persisted_system is None or sha256(
                        persisted_system.encode("utf-8")
                    ).hexdigest() != snapshot.system_prompt_hash:
                        raise ValueError("persisted system prompt does not match RunConfig snapshot")

        current_trace = (
            await self.trace_store.get_trace(trace_id)
            if self.trace_store
            else None
        )
        if (
            current_trace
            and current_trace.head_sequence == 0
            and policy_from_context(current_trace.context).requires_task_protocol
        ):
            anchor = require_root_task_anchor(current_trace.context)
            anchor_text = (
                "# Root Task Anchor\n\n"
                + canonical_json(anchor.model_dump(mode="json"))
            )
            anchored_messages = []
            injected = False
            for message in new_messages:
                if not injected and message.get("role") == "user":
                    content = message.get("content") or ""
                    if isinstance(content, str):
                        anchored_content = f"{anchor_text}\n\n{content}"
                    elif isinstance(content, list):
                        anchored_content = [
                            {"type": "text", "text": anchor_text},
                            *content,
                        ]
                    else:
                        raise ValueError(
                            "Recursive root anchor requires text or multimodal user content"
                        )
                    anchored_messages.append({
                        **message,
                        "content": anchored_content,
                    })
                    injected = True
                else:
                    anchored_messages.append(message)
            if not injected:
                anchored_messages.append({"role": "user", "content": anchor_text})
            new_messages = anchored_messages

        # 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:
                # 如果在侧分支中，标记为侧分支消息
                if side_branch_ctx:
                    stored_msg = Message.create(
                        trace_id=trace_id,
                        role=msg_dict["role"],
                        sequence=sequence,
                        goal_id=goal_tree.current_id if goal_tree else None,
                        parent_sequence=head_seq,
                        branch_type=side_branch_ctx.type,
                        branch_id=side_branch_ctx.branch_id,
                        content=msg_dict.get("content"),
                    )
                    self.log.info(f"用户在侧分支 {side_branch_ctx.type} 中追加消息")
                else:
                    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)
            persisted = await self.trace_store.get_trace(trace_id)
            system_content = next(
                (
                    message.get("content")
                    for message in history
                    if message.get("role") == "system"
                    and isinstance(message.get("content"), str)
                ),
                None,
            )
            if persisted and system_content is not None:
                snapshot = load_run_config_snapshot(persisted.context)
                # Pre-created Recursive children enter through the resume path
                # on their very first execution. Bind their actual persisted
                # system prompt once, just like a newly-created root Trace.
                if snapshot.system_prompt_hash is None:
                    snapshot = snapshot.model_copy(update={
                        "system_prompt_hash": sha256(
                            system_content.encode("utf-8")
                        ).hexdigest()
                    })
                    persist_run_config_snapshot(persisted.context, snapshot)
                    await self.trace_store.update_trace(
                        trace_id,
                        context=persisted.context,
                    )

        return history, sequence, created_messages, head_seq or 0

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

    async def _manage_context_usage(
        self,
        trace_id: str,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
        head_seq: int,
    ) -> Tuple[List[Dict], int, int, bool]:
        """
        管理 context 用量：检查、预警、压缩

        Returns:
            (updated_history, new_head_seq, next_sequence, needs_enter_compression_branch)
        """
        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")
        )

        # 更新 context usage 快照
        self._context_usage[trace_id] = ContextUsage(
            trace_id=trace_id,
            message_count=msg_count,
            token_count=token_count,
            max_tokens=max_tokens,
            usage_percent=progress_pct,
            image_count=img_count,
        )

        # 阈值警告（30%, 50%, 80%）
        if trace_id not in self._context_warned:
            self._context_warned[trace_id] = set()

        for threshold in [30, 50, 80]:
            if progress_pct >= threshold and threshold not in self._context_warned[trace_id]:
                self._context_warned[trace_id].add(threshold)
                self.log.warning(
                    f"Context 使用率达到 {threshold}%: {token_count:,} / {max_tokens:,} tokens ({msg_count} 条消息)"
                )

        # 检查是否需要压缩（仅基于 token 数量）
        needs_compression = token_count > max_tokens

        if not needs_compression:
            return history, head_seq, sequence, False

        # 检查是否有待评估知识（压缩前必须先评估）
        if self.trace_store and not config.force_side_branch:
            pending = await self.trace_store.get_pending_knowledge_entries(trace_id)
            if pending:
                # 设置侧分支队列：反思 → 知识评估 → 压缩
                # 反思放在前面，确保反思期间完成的 goal 产生的新知识也能在压缩前被评估
                if config.knowledge.enable_extraction:
                    config.force_side_branch = ["reflection", "knowledge_eval", "compression"]
                else:
                    config.force_side_branch = ["knowledge_eval", "compression"]

                # 在 trace.context 中设置触发事件
                trace = await self.trace_store.get_trace(trace_id)
                if trace:
                    if not trace.context:
                        trace.context = {}
                    trace.context["knowledge_eval_trigger"] = "compression"
                    await self.trace_store.update_trace(trace_id, context=trace.context)

                self.log.info(f"[Knowledge Eval] 压缩前触发知识评估，待评估: {len(pending)} 条")
                return history, head_seq, sequence, True

        # 知识提取：在任何压缩发生前，用完整 history 做反思（进入反思侧分支）
        if config.knowledge.enable_extraction and not config.force_side_branch:
            # 设置侧分支队列：先反思，再压缩
            config.force_side_branch = ["reflection", "compression"]
            return history, head_seq, sequence, True

        # 以下为未启用反思、需要压缩的情况，直接进行level 1压缩，并检查是否需要进行level 2压缩（进入侧分支）
        # Level 1 压缩：Goal 完成压缩
        if config.goal_compression != "none" and self.trace_store and goal_tree:
            if head_seq > 0:
                main_path_msgs = await self.trace_store.get_main_path_messages(
                    trace_id, head_seq
                )
                compressed_msgs = compress_completed_goals(main_path_msgs, goal_tree)
                if len(compressed_msgs) < len(main_path_msgs):
                    self.log.info(
                        "Level 1 压缩: %d -> %d 条消息",
                        len(main_path_msgs), len(compressed_msgs),
                    )
                    history = [msg.to_llm_dict() for msg in compressed_msgs]
                else:
                    self.log.info(
                        "Level 1 压缩: 无可过滤消息 (%d 条全部保留)",
                        len(main_path_msgs),
                    )
        elif needs_compression:
            self.log.warning(
                "Token 数 (%d) 超过阈值，但无法执行 Level 1 压缩（缺少 store 或 goal_tree，或 goal_compression=none）",
                token_count,
            )

        # Level 2 压缩：检查 Level 1 后是否仍超阈值
        # 注意：Level 1 压缩后需要重新优化图片并计算 token
        optimized_history_after = await self._optimize_images(
            history,
            config.model,
            trace_id=trace_id,
        )
        token_count_after = estimate_tokens(optimized_history_after)
        needs_level2 = token_count_after > max_tokens

        if needs_level2:
            self.log.info(
                "Level 1 后仍超阈值 (token=%d/%d)，需要进入压缩侧分支",
                token_count_after, max_tokens,
            )
            # 如果还没有设置侧分支（说明没有启用知识提取），直接进入压缩
            if not config.force_side_branch:
                config.force_side_branch = ["compression"]
            # 返回标志，让主循环进入侧分支
            return history, head_seq, sequence, True

        # 压缩完成后，输出最终发给模型的消息列表
        self.log.info("Level 1 压缩完成，发送给模型的消息列表:")
        for idx, msg in enumerate(history):
            role = msg.get("role", "unknown")
            content = msg.get("content", "")
            if isinstance(content, str):
                preview = content[:100] + ("..." if len(content) > 100 else "")
            elif isinstance(content, list):
                preview = f"[{len(content)} blocks]"
            else:
                preview = str(content)[:100]
            self.log.info(f"  [{idx}] {role}: {preview}")

        return history, head_seq, sequence, False

    async def _build_knowledge_eval_prompt(
        self,
        trace_id: str,
        goal_tree: Optional[GoalTree]
    ) -> str:
        """构建知识评估 prompt"""
        if not self.trace_store:
            return ""

        pending = await self.trace_store.get_pending_knowledge_entries(trace_id)
        if not pending:
            return ""

        # 获取mission
        trace = await self.trace_store.get_trace(trace_id)
        mission = trace.task if trace else "未知任务"

        # 获取当前Goal
        current_goal = goal_tree.find(goal_tree.current_id) if goal_tree and goal_tree.current_id else None
        goal_desc = current_goal.description if current_goal else "无当前目标"

        # 构建知识列表
        knowledge_list = []
        for idx, entry in enumerate(pending, 1):
            knowledge_list.append(
                f"### 知识 {idx}: {entry['knowledge_id']}\n"
                f"- task: {entry['task']}\n"
                f"- content: {entry['content']}\n"
                f"- 注入于: sequence {entry['injected_at_sequence']}, goal {entry['goal_id']}"
            )

        prompt = f"""你是知识评估助手。请评估以下知识在本次任务执行中的实际效果。

## 当前任务（Mission）
{mission}

## 当前 Goal
{goal_desc}

## 待评估知识列表
{chr(10).join(knowledge_list)}

## 评估维度
1. **helpfulness**: 知识内容是否对完成任务有实质帮助？
2. **relevance**: 执行过程中是否体现了该知识的内容？

## 评估分类
- irrelevant: task与当前任务无关
- unused: 相关但未使用
- helpful: 有帮助
- harmful: 有负面作用
- neutral: 无明显作用

## 输出格式
请直接输出评估结果，使用JSON格式：

{{
  "evaluations": [
    {{
      "knowledge_id": "knowledge-xxx",
      "eval_status": "helpful",
      "reason": "1-2句评估理由"
    }}
  ]
}}
"""
        return prompt

    async def _single_turn_compress(
        self,
        trace_id: str,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
    ) -> str:
        """单次 LLM 调用生成压缩摘要，返回 summary 文本"""

        self.log.info("执行单次 LLM 压缩")

        # 构建压缩 prompt（使用 SINGLE_TURN_PROMPT）
        from cyber_agent.core.prompts import build_single_turn_prompt
        goal_prompt = goal_tree.to_prompt(include_summary=True) if goal_tree else ""
        compress_prompt = build_single_turn_prompt(goal_prompt)
        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
        )

        # 单次 LLM 调用（无工具）
        result = await self.call_recursive_llm(
            trace_id,
            purpose="ordinary",
            messages=compress_messages,
            model=config.model,
            tools=[],  # 不提供工具
            temperature=config.temperature,
            fail_on_post_response_exhaustion=True,
            **config.extra_llm_params,
        )

        summary_text = result.get("content", "").strip()

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

        return summary_text

    @staticmethod
    def _try_fix_json(s: str) -> Optional[dict]:
        """尝试修复常见的 JSON 截断/格式问题，返回 dict 或 None"""
        import re

        fixed = s.strip()

        # 1. 修复值中未转义的引号（如 "key": "he said "hello" to me"）
        # 策略：找到 key-value 模式中值字符串内部的裸引号并转义
        def _fix_inner_quotes(text: str) -> str:
            # 匹配 ": "..." 模式，修复值内部的未转义引号
            result = []
            i = 0
            while i < len(text):
                # 找到 ": " 后面的值字符串开头
                if text[i] == '"':
                    # 找到这个引号对应的字符串结束位置
                    j = i + 1
                    while j < len(text):
                        if text[j] == '\\':
                            j += 2  # 跳过转义字符
                            continue
                        if text[j] == '"':
                            break
                        j += 1
                    # 检查引号后面是否是合法的 JSON 分隔符
                    if j < len(text):
                        after = j + 1
                        # 跳过空白
                        while after < len(text) and text[after] in ' \t\n\r':
                            after += 1
                        if after < len(text) and text[after] not in ':,}]\n\r':
                            # 这个引号不是真正的结束引号，继续往后找
                            # 找到下一个后面跟合法分隔符的引号
                            k = j + 1
                            found_end = False
                            while k < len(text):
                                if text[k] == '"':
                                    peek = k + 1
                                    while peek < len(text) and text[peek] in ' \t\n\r':
                                        peek += 1
                                    if peek >= len(text) or text[peek] in ':,}]':
                                        # 这才是真正的结束引号，转义中间的引号
                                        inner = text[i+1:k].replace('"', '\\"')
                                        result.append('"' + inner + '"')
                                        i = k + 1
                                        found_end = True
                                        break
                                k += 1
                            if found_end:
                                continue
                result.append(text[i])
                i += 1
            return ''.join(result)

        fixed = _fix_inner_quotes(fixed)

        # 2. 去掉尾部多余逗号
        fixed = re.sub(r',\s*([}\]])', r'\1', fixed)

        # 3. 尝试补全截断的字符串和括号
        for suffix in ['', '"', '"}', '"]', '"}]', '"}}']:
            try:
                attempt = fixed + suffix
                open_braces = attempt.count('{') - attempt.count('}')
                open_brackets = attempt.count('[') - attempt.count(']')
                attempt += '}' * max(0, open_braces) + ']' * max(0, open_brackets)
                result = json.loads(attempt)
                if isinstance(result, dict):
                    self.log.info(f"[JSON Fix] 成功修复 JSON (suffix={repr(suffix)})")
                    return result
            except json.JSONDecodeError:
                continue

        return None

    async def _resume_approved_tool_batch(
        self,
        *,
        trace: Trace,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
        head_seq: int,
        side_branch_ctx: Optional[SideBranchContext],
        runtime_tool_names: set[str],
    ) -> Tuple[List[Dict], int, int]:
        """Execute one fully-decided persisted approval batch exactly once."""
        if not config.approval_batch_id:
            return history, sequence, head_seq
        if not self.trace_store or not hasattr(
            self.trace_store, "get_tool_approval_batch"
        ):
            raise RuntimeError("TraceStore does not support tool approvals")
        batch = await self.trace_store.get_tool_approval_batch(trace.trace_id)
        if batch is None or batch.batch_id != config.approval_batch_id:
            raise RuntimeError("approved tool batch was not found")
        if batch.status != "decided":
            raise RuntimeError(f"tool approval batch is not executable: {batch.status}")
        if any(call.decision == "pending" for call in batch.calls):
            raise RuntimeError("tool approval batch still has pending decisions")
        if any(call.execution_status == "executing" for call in batch.calls):
            batch.status = "execution_unknown"
            for call in batch.calls:
                if call.execution_status == "executing":
                    call.execution_status = "execution_unknown"
            batch.updated_at = datetime.now().isoformat()
            await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch)
            raise RuntimeError("tool execution outcome is unknown; refusing automatic retry")

        batch.status = "executing"
        batch.updated_at = datetime.now().isoformat()
        await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch)
        current_goal_id = goal_tree.current_id if goal_tree and goal_tree.current_id else None
        trigger_event = None
        if side_branch_ctx and side_branch_ctx.type == "knowledge_eval":
            trigger_event = (trace.context.get("active_side_branch") or {}).get(
                "trigger_event",
                "unknown",
            )

        for call in batch.calls:
            if call.execution_status in {"executed", "rejected"}:
                continue
            if call.decision == "rejected":
                tool_result: Any = json.dumps({
                    "status": "rejected",
                    "error": "Tool call rejected by the local user",
                }, ensure_ascii=False)
                call.execution_status = "rejected"
            elif call.decision in {"approved", "auto_approved"}:
                call.execution_status = "executing"
                batch.updated_at = datetime.now().isoformat()
                await self.trace_store.replace_tool_approval_batch(
                    trace.trace_id,
                    batch,
                )
                tool_result = await self.tools.execute(
                    call.tool_name,
                    call.effective_arguments,
                    uid=config.uid or "",
                    context=self._build_tool_context(
                        config=config,
                        trace=trace,
                        trace_id=trace.trace_id,
                        goal_id=current_goal_id,
                        goal_tree=goal_tree,
                        sequence=sequence,
                        side_branch_ctx=side_branch_ctx,
                        trigger_event=trigger_event,
                    ),
                    allowed_tool_names=runtime_tool_names,
                    tool_call_id=call.tool_call_id,
                    approval_grant=approval_grant(batch, call),
                )
            else:
                raise RuntimeError(f"unsupported tool approval decision: {call.decision}")

            if isinstance(tool_result, str):
                normalized = {"text": tool_result}
            elif isinstance(tool_result, dict):
                normalized = tool_result
            else:
                normalized = {"text": str(tool_result)}
            tool_text = normalized.get("text", str(normalized))
            tool_images = normalized.get("images", [])
            artifact_refs = normalized.get("artifact_refs", [])
            tool_content: Any = tool_text
            if tool_images:
                tool_content = [{"type": "text", "text": tool_text}]
                for image in tool_images:
                    if image.get("type") == "base64" and image.get("data"):
                        media_type = image.get("media_type", "image/png")
                        tool_content.append({
                            "type": "image_url",
                            "image_url": {
                                "url": f"data:{media_type};base64,{image['data']}"
                            },
                        })
                    elif image.get("type") == "url" and image.get("url"):
                        tool_content.append({
                            "type": "image_url",
                            "image_url": {"url": image["url"]},
                        })
            tool_msg = Message.create(
                trace_id=trace.trace_id,
                role="tool",
                sequence=sequence,
                goal_id=current_goal_id,
                parent_sequence=head_seq,
                tool_call_id=call.tool_call_id,
                branch_type=side_branch_ctx.type if side_branch_ctx else None,
                branch_id=side_branch_ctx.branch_id if side_branch_ctx else None,
                content={
                    "tool_name": call.tool_name,
                    "result": tool_content,
                    "artifact_refs": artifact_refs,
                },
            )
            await self.trace_store.add_message(tool_msg)
            if tool_images and hasattr(self.trace_store, "write_message_attachment"):
                import base64 as b64mod
                for image in tool_images:
                    if image.get("data"):
                        await self.trace_store.write_message_attachment(
                            trace.trace_id,
                            tool_msg.message_id,
                            suffix=".png",
                            data=b64mod.b64decode(image["data"]),
                        )
                        break
            call.result_message_id = tool_msg.message_id
            if call.execution_status != "rejected":
                call.execution_status = "executed"
            batch.updated_at = datetime.now().isoformat()
            await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch)
            history.append({
                "role": "tool",
                "tool_call_id": call.tool_call_id,
                "name": call.tool_name,
                "content": tool_content,
                "_message_id": tool_msg.message_id,
            })
            head_seq = sequence
            sequence += 1

        batch.status = "completed"
        batch.updated_at = datetime.now().isoformat()
        await self.trace_store.replace_tool_approval_batch(trace.trace_id, batch)
        await self.trace_store.update_trace(
            trace.trace_id,
            status="running",
            head_sequence=head_seq,
        )
        config.approval_batch_id = None
        return history, sequence, head_seq

    async def _mark_approval_execution_unknown(self, trace_id: str) -> bool:
        """Fail closed if an approved batch stopped after execution began."""
        if not self.trace_store or not hasattr(
            self.trace_store, "get_tool_approval_batch"
        ):
            return False
        batch = await self.trace_store.get_tool_approval_batch(trace_id)
        if batch is None or batch.status != "executing":
            return False
        batch.status = "execution_unknown"
        batch.updated_at = datetime.now().isoformat()
        for call in batch.calls:
            if call.execution_status == "executing":
                call.execution_status = "execution_unknown"
        await self.trace_store.replace_tool_approval_batch(trace_id, batch)
        await self.trace_store.update_trace(
            trace_id,
            status="failed",
            error_message=(
                "Tool execution outcome is unknown; automatic retry was refused"
            ),
            completed_at=datetime.now(),
        )
        return True

    async def _agent_loop(
        self,
        trace: Trace,
        history: List[Dict],
        goal_tree: Optional[GoalTree],
        config: RunConfig,
        sequence: int,
        inject_skills: Optional[List[str]] = None,
        skill_recency_threshold: int = 10,
    ) -> AsyncIterator[Union[Trace, Message]]:
        """执行 Agent 的 ReAct 主循环。

        ``run`` 在 Trace 准备后调用；Recursive 的预算、停止、工具门禁和根验收都在此串联。
        """
        trace_id = trace.trace_id
        runtime_tool_names = self._get_runtime_tool_names(config, trace)
        tool_schemas = self._get_runtime_tool_schemas(
            config,
            trace,
            runtime_tool_names=runtime_tool_names,
        )
        completion_status = "completed"

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

        # 侧分支状态（None = 主路径）
        side_branch_ctx: Optional[SideBranchContext] = None

        # 检查是否有未完成的侧分支需要恢复
        if trace.context.get("active_side_branch"):
            side_branch_data = trace.context["active_side_branch"]
            branch_id = side_branch_data["branch_id"]
            start_sequence = side_branch_data["start_sequence"]

            # 只恢复同一持久分支；sequence 范围会把相邻分支混进来。
            if self.trace_store:
                all_messages = await self.trace_store.get_trace_messages(trace_id)
                side_messages = [
                    m for m in all_messages
                    if (
                        m.sequence >= start_sequence
                        and m.branch_id == branch_id
                        and m.branch_type == side_branch_data["type"]
                        and m.status == "active"
                    )
                ]
                reconstructed_turns = sum(
                    1 for message in side_messages if message.role == "assistant"
                )
                persisted_turns = int(side_branch_data.get("turns_used", 0))
                turns_used = max(persisted_turns, reconstructed_turns)
                if turns_used != persisted_turns or "turns_used" not in side_branch_data:
                    side_branch_data["turns_used"] = turns_used
                    await self.trace_store.update_trace(
                        trace_id,
                        context=trace.context,
                    )

                # 恢复侧分支上下文
                side_branch_ctx = SideBranchContext(
                    type=side_branch_data["type"],
                    branch_id=branch_id,
                    start_head_seq=side_branch_data["start_head_seq"],
                    start_sequence=side_branch_data["start_sequence"],
                    start_history_length=0,  # 稍后重新计算
                    start_iteration=side_branch_data.get("start_iteration", 0),
                    max_turns=side_branch_data.get("max_turns", config.side_branch_max_turns),
                    turns_used=int(turns_used),
                )

                self.log.info(
                    f"恢复未完成的侧分支: {side_branch_ctx.type}, "
                    f"max_turns={side_branch_ctx.max_turns}"
                )

                # 将侧分支消息追加到 history
                for m in side_messages:
                    history.append(m.to_llm_dict())

                # 重新计算 start_history_length
                side_branch_ctx.start_history_length = len(history) - len(side_messages)

                # The assistant Message is the durable turn record.  A crash can
                # happen after that Message is stored but before Trace.context is
                # updated.  If reconstruction shows the budget was already
                # exhausted, fail the branch closed instead of buying one extra
                # model turn after restart.
                if side_branch_ctx.turns_used >= side_branch_ctx.max_turns:
                    self.log.warning(
                        "恢复的侧分支 %s 已用尽轮次 %d/%d，直接返回主路径",
                        side_branch_ctx.type,
                        side_branch_ctx.turns_used,
                        side_branch_ctx.max_turns,
                    )
                    main_path_messages = await self.trace_store.get_main_path_messages(
                        trace_id,
                        side_branch_ctx.start_head_seq,
                    )
                    history = [message.to_llm_dict() for message in main_path_messages]
                    head_seq = side_branch_ctx.start_head_seq
                    trace.context.pop("active_side_branch", None)
                    if config.force_side_branch:
                        if config.force_side_branch[0] == side_branch_ctx.type:
                            config.force_side_branch.pop(0)
                        if not config.force_side_branch:
                            config.force_side_branch = None
                    await self.trace_store.update_trace(
                        trace_id,
                        context=trace.context,
                        head_sequence=head_seq,
                    )
                    side_branch_ctx = None

        if config.approval_batch_id:
            history, sequence, head_seq = await self._resume_approved_tool_batch(
                trace=trace,
                history=history,
                goal_tree=goal_tree,
                config=config,
                sequence=sequence,
                head_seq=head_seq,
                side_branch_ctx=side_branch_ctx,
                runtime_tool_names=runtime_tool_names,
            )

        break_after_side_branch = False  # 侧分支退出后是否 break 主循环

        for iteration in range(config.max_iterations):
            # 更新活动时间（表明trace正在活跃运行）
            if self.trace_store:
                await self.trace_store.update_trace(
                    trace_id,
                    last_activity_at=datetime.now()
                )

            # 检查取消信号
            cancel_event = self._cancel_events.get(trace_id)
            if cancel_event and cancel_event.is_set():
                self.log.info(f"Trace {trace_id} stopped by user")
                trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
                if trace_obj:
                    yield trace_obj
                return

            # 检查Goal完成触发的知识评估
            if not side_branch_ctx and self.trace_store:
                trace = await self.trace_store.get_trace(trace_id)
                if trace and trace.context and trace.context.get("pending_knowledge_eval"):
                    # 清除标志
                    trace.context.pop("pending_knowledge_eval", None)
                    await self.trace_store.update_trace(trace_id, context=trace.context)
                    # 设置侧分支队列
                    config.force_side_branch = ["knowledge_eval"]
                    self.log.info("[Knowledge Eval] 检测到Goal完成触发，进入知识评估侧分支")

            # Context 管理（仅主路径）
            needs_enter_side_branch = False
            if not side_branch_ctx:
                # 侧分支退出后需要 break 主循环
                if break_after_side_branch and not config.force_side_branch:
                    break

                # 检查是否强制进入侧分支（API 手动触发或自动压缩流程）
                if config.force_side_branch:
                    needs_enter_side_branch = True
                    self.log.info(f"强制进入侧分支: {config.force_side_branch}")
                else:
                    # 正常的 context 管理逻辑
                    history, head_seq, sequence, needs_enter_side_branch = await self._manage_context_usage(
                        trace_id, history, goal_tree, config, sequence, head_seq
                    )

            # 进入侧分支
            if needs_enter_side_branch and not side_branch_ctx:
                # 刷新 trace，获取 _manage_context_usage 可能写入 DB 的 knowledge_eval_trigger
                if self.trace_store:
                    fresh = await self.trace_store.get_trace(trace_id)
                    if fresh:
                        trace = fresh
                # 从队列中取出第一个侧分支类型
                branch_type: Literal["compression", "reflection", "knowledge_eval"]
                if config.force_side_branch and isinstance(config.force_side_branch, list) and len(config.force_side_branch) > 0:
                    branch_type = config.force_side_branch.pop(0)  # type: ignore
                    self.log.info(f"从队列取出侧分支: {branch_type}, 剩余队列: {config.force_side_branch}")
                elif config.knowledge.enable_extraction:
                    # 兼容旧的单值模式（如果 force_side_branch 是字符串）
                    branch_type = "reflection"
                else:
                    # 自动触发：压缩
                    branch_type = "compression"

                branch_id = f"{branch_type}_{uuid.uuid4().hex[:8]}"

                side_branch_ctx = SideBranchContext(
                    type=branch_type,
                    branch_id=branch_id,
                    start_head_seq=head_seq,
                    start_sequence=sequence,
                    start_history_length=len(history),
                    start_iteration=iteration,
                    max_turns=config.side_branch_max_turns,
                    turns_used=0,
                )

                # 持久化侧分支状态
                if self.trace_store:
                    # 获取触发事件（如果是 knowledge_eval 分支）
                    trigger_event = trace.context.get("knowledge_eval_trigger", "unknown") if branch_type == "knowledge_eval" else None

                    trace.context["active_side_branch"] = {
                        "type": side_branch_ctx.type,
                        "branch_id": side_branch_ctx.branch_id,
                        "start_head_seq": side_branch_ctx.start_head_seq,
                        "start_sequence": side_branch_ctx.start_sequence,
                        "start_iteration": side_branch_ctx.start_iteration,
                        "max_turns": side_branch_ctx.max_turns,
                        "turns_used": side_branch_ctx.turns_used,
                        "started_at": datetime.now().isoformat(),
                    }

                    # 如果是 knowledge_eval 分支，添加 trigger_event
                    if trigger_event:
                        trace.context["active_side_branch"]["trigger_event"] = trigger_event
                        # 清除触发事件标记
                        trace.context.pop("knowledge_eval_trigger", None)

                    await self.trace_store.update_trace(
                        trace_id,
                        context=trace.context
                    )

                # 追加侧分支 prompt
                if branch_type == "reflection":
                    # 完成场景用全局复盘 prompt，压缩场景用阶段性反思 prompt
                    if break_after_side_branch:
                        prompt = config.knowledge.get_completion_reflect_prompt()
                    else:
                        prompt = config.knowledge.get_reflect_prompt()
                elif branch_type == "knowledge_eval":
                    prompt = await self._build_knowledge_eval_prompt(trace_id, goal_tree)
                else:  # compression
                    from cyber_agent.trace.compaction import build_compression_prompt
                    prompt = build_compression_prompt(goal_tree)

                branch_user_msg = Message.create(
                    trace_id=trace_id,
                    role="user",
                    sequence=sequence,
                    parent_sequence=head_seq,
                    goal_id=goal_tree.current_id if goal_tree else None,
                    branch_type=branch_type,
                    branch_id=branch_id,
                    content=prompt,
                )

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

                history.append(branch_user_msg.to_llm_dict())
                head_seq = sequence
                sequence += 1

                self.log.info(f"进入侧分支: {branch_type}, branch_id={branch_id}")
                continue  # 跳过本轮，下一轮开始侧分支

            if self.trace_store:
                fresh_trace = await self.trace_store.get_trace(trace_id)
                if fresh_trace:
                    trace = fresh_trace
            runtime_policy = policy_from_context(trace.context)
            runtime_tool_names = self._get_runtime_tool_names(
                config,
                trace,
                in_side_branch=side_branch_ctx is not None,
            )
            tool_schemas = self._get_runtime_tool_schemas(
                config,
                trace,
                in_side_branch=side_branch_ctx is not None,
                runtime_tool_names=runtime_tool_names,
            )
            dispatch_allowlist = (
                runtime_tool_names
                if runtime_policy.mode is AgentMode.RECURSIVE
                else None
            )

            # 构建 LLM messages（注入上下文，移除内部字段）
            llm_messages = [{k: v for k, v in msg.items() if not k.startswith("_")} for msg in history]

            # 优化已处理的图片（分级处理：保留/压缩/描述）
            llm_messages = await self._optimize_images(
                llm_messages,
                config.model,
                trace_id=trace_id,
            )

            # 对历史消息应用 Prompt Caching
            llm_messages = self._add_cache_control(
                llm_messages,
                config.model,
                config.enable_prompt_caching
            )

            # 调用 LLM（等待完成后再检查 cancel 信号，不中断正在进行的调用）
            result = await self.call_recursive_llm(
                trace_id,
                purpose="ordinary",
                messages=llm_messages,
                model=config.model,
                tools=tool_schemas,
                temperature=config.temperature,
                **config.extra_llm_params,
            )

            if (
                runtime_policy.mode is AgentMode.RECURSIVE
                and self.is_cancel_requested(trace_id)
            ):
                trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
                if trace_obj:
                    yield trace_obj
                return

            response_content = result.get("content", "")
            reasoning_content = result.get("reasoning_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")
            budget_exceeded_dimension = result.get("_resource_budget_exceeded")
            if budget_exceeded_dimension:
                tool_calls = None
                finish_reason = "budget_exhausted"

            lifecycle_tools = {
                "agent",
                "submit_task_report",
                "review_task_result",
                "update_task_progress",
                "manage_candidate",
            }
            lifecycle_call_count = sum(
                1 for tc in (tool_calls or [])
                if tc.get("function", {}).get("name") in lifecycle_tools
            )
            protocol_batch_error = None
            if (
                runtime_policy.requires_task_protocol
                and lifecycle_call_count
                and len(tool_calls or []) != 1
            ):
                protocol_batch_error = (
                    "Recursive lifecycle tools must be the only tool call in an LLM turn"
                )
            runtime_protocol_state = (
                ensure_task_protocol(trace.context)
                if runtime_policy.requires_task_protocol
                else None
            )
            protocol_lifecycle_required = bool(
                runtime_protocol_state
                and (
                    runtime_protocol_state["pending_reviews"]
                    or runtime_protocol_state["next_actions"]
                )
            )

            # 周期性自动注入上下文（仅主路径）
            if (
                not side_branch_ctx
                and not budget_exceeded_dimension
                and iteration % CONTEXT_INJECTION_INTERVAL == 0
            ):
                # 检查是否已经调用了 get_current_context
                if tool_calls:
                    has_context_call = any(
                        tc.get("function", {}).get("name") == "get_current_context"
                        for tc in tool_calls
                    )
                else:
                    has_context_call = False
                    tool_calls = []

                if (
                    not has_context_call
                    and not lifecycle_call_count
                    and not protocol_lifecycle_required
                    and "get_current_context" in runtime_tool_names
                ):
                    # 手动添加 get_current_context 工具调用
                    context_call_id = f"call_context_{uuid.uuid4().hex[:8]}"
                    tool_calls.append({
                        "id": context_call_id,
                        "type": "function",
                        "function": {"name": "get_current_context", "arguments": "{}"}
                    })
                    self.log.info(f"[周期性注入] 自动添加 get_current_context 工具调用 (iteration={iteration})")

            # Skill 指定注入（仅主路径，首轮 iteration==0 时执行）
            if (
                not side_branch_ctx
                and not budget_exceeded_dimension
                and inject_skills
                and iteration == 0
                and not lifecycle_call_count
                and not protocol_lifecycle_required
                and "skill" in runtime_tool_names
            ):
                skills_to_inject = self._check_skills_need_injection(
                    trace, inject_skills, history, skill_recency_threshold
                )
                if skills_to_inject:
                    if not tool_calls:
                        tool_calls = []
                    for skill_name in skills_to_inject:
                        skill_call_id = f"call_skill_{skill_name}_{uuid.uuid4().hex[:8]}"
                        tool_calls.append({
                            "id": skill_call_id,
                            "type": "function",
                            "function": {
                                "name": "skill",
                                "arguments": json.dumps({"skill_name": skill_name})
                            }
                        })
                        self.log.info(f"[Skill 指定注入] 自动添加 skill(\"{skill_name}\") 工具调用")

            # 按需自动创建 root goal（仅主路径）
            if not side_branch_ctx and 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
                )
                self.log.debug(f"[Auto Root Goal] Before tool execution: goal_tree.goals={len(goal_tree.goals)}, has_goal_call={has_goal_call}, tool_calls={[tc.get('function', {}).get('name') 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
                    )
                    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)
                    self.log.info(f"自动创建 root goal: {goal_tree.goals[0].id}（未自动 focus，等待模型决定）")
                else:
                    self.log.debug(f"[Auto Root Goal] 检测到 goal 工具调用，跳过自动创建")

            # 获取当前 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,
                branch_type=side_branch_ctx.type if side_branch_ctx else None,
                branch_id=side_branch_ctx.branch_id if side_branch_ctx else None,
                content={"text": response_content, "tool_calls": tool_calls, "reasoning_content": reasoning_content or None},
                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)
                # 记录模型使用
                await self.trace_store.record_model_usage(
                    trace_id=trace_id,
                    sequence=sequence,
                    role="assistant",
                    model=config.model,
                    prompt_tokens=prompt_tokens,
                    completion_tokens=completion_tokens,
                    cache_read_tokens=cache_read_tokens or 0,
                )

            # 知识评估侧分支：即时检测并写入评估结果
            if side_branch_ctx and side_branch_ctx.type == "knowledge_eval":
                text = response_content if isinstance(response_content, str) else ""
                eval_results = None

                try:
                    eval_results = json.loads(text.strip())
                    if "evaluations" not in eval_results:
                        eval_results = None
                except json.JSONDecodeError:
                    import re
                    json_match = re.search(r'```json\s*(\{.*?\})\s*```', text, re.DOTALL)
                    if json_match:
                        try:
                            eval_results = json.loads(json_match.group(1))
                        except json.JSONDecodeError:
                            pass

                    if not eval_results:
                        json_match = re.search(r'\{[^{]*"evaluations"[^}]*\[[^\]]*\][^}]*\}', text, re.DOTALL)
                        if json_match:
                            try:
                                eval_results = json.loads(json_match.group(0))
                            except json.JSONDecodeError:
                                pass

                if eval_results and self.trace_store:
                    current_trace = await self.trace_store.get_trace(trace_id)
                    trigger_event = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown")

                    for eval_item in eval_results.get("evaluations", []):
                        await self.trace_store.update_knowledge_evaluation(
                            trace_id=trace_id,
                            knowledge_id=eval_item["knowledge_id"],
                            eval_result={
                                "eval_status": eval_item["eval_status"],
                                "reason": eval_item.get("reason", "")
                            },
                            trigger_event=trigger_event
                        )
                    self.log.info(f"[Knowledge Eval] 已写入 {len(eval_results.get('evaluations', []))} 条评估结果")

            # 一轮以成功持久化的侧分支 assistant 回复为准；先落盘计数，
            # 再 yield，确保停止/重启不会重新获得已消费轮次。
            if side_branch_ctx:
                side_branch_ctx.turns_used += 1
                active_branch = trace.context.get("active_side_branch")
                if isinstance(active_branch, dict):
                    active_branch["turns_used"] = side_branch_ctx.turns_used
                    if self.trace_store:
                        await self.trace_store.update_trace(
                            trace_id,
                            context=trace.context,
                        )

            yield assistant_msg
            head_seq = sequence
            sequence += 1

            if budget_exceeded_dimension:
                completion_status = "failed"
                trace.context["termination_reason"] = (
                    f"budget_exhausted:{budget_exceeded_dimension}"
                )
                if self.trace_store:
                    await self.trace_store.update_trace(
                        trace_id,
                        context=trace.context,
                        error_message=trace.context["termination_reason"],
                    )
                break

            # 检查侧分支是否应该退出
            if side_branch_ctx:
                turns_in_branch = side_branch_ctx.turns_used
                should_exit = turns_in_branch >= side_branch_ctx.max_turns or not tool_calls

                if turns_in_branch >= side_branch_ctx.max_turns:
                    self.log.warning(
                        f"侧分支 {side_branch_ctx.type} 达到最大轮次 "
                        f"{side_branch_ctx.max_turns}，强制退出"
                    )

                if should_exit and side_branch_ctx.type == "compression":
                    # === 压缩侧分支退出（超时 + 正常完成统一处理）===
                    summary_text = ""

                    # 1. 从当前回复提取
                    if response_content:
                        if "[[SUMMARY]]" in response_content:
                            summary_text = response_content[
                                response_content.index("[[SUMMARY]]") + len("[[SUMMARY]]"):
                            ].strip()
                        elif response_content.strip():
                            summary_text = response_content.strip()

                    # 2. 从持久化存储按 sequence 范围查询
                    if not summary_text and self.trace_store:
                        all_messages = await self.trace_store.get_trace_messages(trace_id)
                        side_messages = [
                            m for m in all_messages
                            if (
                                m.sequence >= side_branch_ctx.start_sequence
                                and m.branch_id == side_branch_ctx.branch_id
                                and m.branch_type == side_branch_ctx.type
                                and m.status == "active"
                            )
                        ]
                        for msg in reversed(side_messages):
                            if msg.role == "assistant" and isinstance(msg.content, dict):
                                text = msg.content.get("text", "")
                                if "[[SUMMARY]]" in text:
                                    summary_text = text[text.index("[[SUMMARY]]") + len("[[SUMMARY]]"):].strip()
                                    break
                                elif text:
                                    summary_text = text
                                    break

                    # 3. 单次 LLM 调用
                    if not summary_text:
                        self.log.warning("侧分支未生成有效 summary，fallback 到单次 LLM 压缩")
                        pre_branch_history = history[:side_branch_ctx.start_history_length]
                        summary_text = await self._single_turn_compress(
                            trace_id, pre_branch_history, goal_tree, config,
                        )

                    # 创建主路径 summary 消息并重建 history
                    if summary_text:
                        # 清理侧分支指令，防止泄露到主分支
                        summary_text = summary_text.replace(
                            "**生成摘要后立即停止，不要继续执行原有任务。**", ""
                        ).strip()
                        from cyber_agent.core.prompts import build_summary_header
                        summary_content = build_summary_header(summary_text)

                        if goal_tree and goal_tree.goals:
                            goal_tree_detail = goal_tree.to_prompt(include_summary=True)
                            summary_content += f"\n\n## Current Plan\n\n{goal_tree_detail}"

                        # 找第一条 user message 的 sequence 作为 parent
                        # 续跑时 get_main_path_messages 沿 parent 链回溯，
                        # 指向 first_user 可以跳过所有被压缩的中间消息
                        first_user_seq = None
                        if self.trace_store:
                            all_msgs = await self.trace_store.get_trace_messages(trace_id)
                            for m in all_msgs:
                                if m.role == "user":
                                    first_user_seq = m.sequence
                                    break

                        summary_msg = Message.create(
                            trace_id=trace_id,
                            role="user",
                            sequence=sequence,
                            parent_sequence=first_user_seq,
                            branch_type=None,
                            content=summary_content,
                        )

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

                        history = self._rebuild_history_after_compression(
                            history, summary_msg.to_llm_dict(), label="压缩侧分支"
                        )
                        head_seq = sequence
                        sequence += 1
                    else:
                        self.log.error("所有压缩方案均未生成有效 summary，跳过压缩")
                        # 回退 history 到侧分支开始前，防止侧分支指令泄露到主分支
                        history = history[:side_branch_ctx.start_history_length]
                        head_seq = side_branch_ctx.start_head_seq

                    # 清理
                    trace.context.pop("active_side_branch", None)
                    config.force_side_branch = None
                    if self.trace_store:
                        await self.trace_store.update_trace(
                            trace_id, context=trace.context, head_sequence=head_seq,
                        )
                    side_branch_ctx = None
                    continue

                elif should_exit and side_branch_ctx.type == "reflection":
                    # === 反思侧分支退出（超时 + 正常完成统一处理）===
                    self.log.info("反思侧分支退出")

                    # auto-commit hook：默认 pending 要等人工 review，
                    # 但 reflect_auto_commit=True 时视作全部 approved，直接批量 upload。
                    if (
                        self.trace_store
                        and getattr(config.knowledge, "reflect_auto_commit", False)
                    ):
                        try:
                            from cyber_agent.trace.extraction_review import auto_commit_branch
                            report = await auto_commit_branch(
                                self.trace_store,
                                trace_id,
                                side_branch_ctx.branch_id,
                            )
                            if report.committed or report.failed:
                                self.log.info(
                                    f"[auto-commit] committed={len(report.committed)} "
                                    f"failed={len(report.failed)} skipped={len(report.skipped)}"
                                )
                        except Exception as e:
                            self.log.error(f"[auto-commit] 反思分支自动提交失败: {e}")

                    # 恢复主路径
                    if self.trace_store:
                        main_path_messages = await self.trace_store.get_main_path_messages(
                            trace_id, side_branch_ctx.start_head_seq
                        )
                        history = [m.to_llm_dict() for m in main_path_messages]
                        head_seq = side_branch_ctx.start_head_seq

                    # 清理
                    trace.context.pop("active_side_branch", None)
                    if not config.force_side_branch or len(config.force_side_branch) == 0:
                        config.force_side_branch = None
                        self.log.info("反思完成，队列为空")
                    if self.trace_store:
                        await self.trace_store.update_trace(
                            trace_id, context=trace.context, head_sequence=head_seq,
                        )
                    side_branch_ctx = None
                    continue

                elif should_exit and side_branch_ctx.type == "knowledge_eval":
                    # === 知识评估侧分支退出 ===
                    self.log.info("知识评估侧分支退出")

                    # 恢复主路径
                    if self.trace_store:
                        main_path_messages = await self.trace_store.get_main_path_messages(
                            trace_id, side_branch_ctx.start_head_seq
                        )
                        history = [m.to_llm_dict() for m in main_path_messages]
                        head_seq = side_branch_ctx.start_head_seq

                    # 清理
                    trace.context.pop("active_side_branch", None)
                    if not config.force_side_branch or len(config.force_side_branch) == 0:
                        config.force_side_branch = None
                        self.log.info("知识评估完成，队列为空")
                    if self.trace_store:
                        await self.trace_store.update_trace(
                            trace_id, context=trace.context, head_sequence=head_seq,
                        )
                    side_branch_ctx = None
                    continue

            # 处理工具调用
            # 截断兜底：finish_reason == "length" 说明响应被 max_tokens 截断，
            # tool call 参数很可能不完整，不应执行，改为提示模型分批操作
            if tool_calls and finish_reason == "length":
                self.log.warning(
                    "[Runner] 响应被 max_tokens 截断，跳过 %d 个不完整的 tool calls",
                    len(tool_calls),
                )
                truncation_hint = TRUNCATION_HINT
                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 not config.approval_batch_id:
                needs_confirmation = (
                    not config.auto_execute_tools
                    or self.tools.check_confirmation_required(tool_calls)
                )
                if needs_confirmation:
                    approval_calls: List[ToolApprovalCallV1] = []
                    for tool_call in tool_calls:
                        tool_name = tool_call.get("function", {}).get("name", "")
                        raw_arguments = tool_call.get("function", {}).get(
                            "arguments", {}
                        )
                        if isinstance(raw_arguments, str):
                            try:
                                parsed_arguments = (
                                    json.loads(raw_arguments)
                                    if raw_arguments.strip()
                                    else {}
                                )
                            except json.JSONDecodeError:
                                parsed_arguments = {"_raw": raw_arguments}
                        elif isinstance(raw_arguments, dict):
                            parsed_arguments = dict(raw_arguments)
                        else:
                            parsed_arguments = {}
                        policy = self.tools.get_runtime_policy(tool_name)
                        requires_confirmation = (
                            not config.auto_execute_tools
                            or bool(policy.get("requires_confirmation"))
                        )
                        decision = (
                            "pending" if requires_confirmation else "auto_approved"
                        )
                        call_id = str(tool_call.get("id") or "")
                        approval_calls.append(ToolApprovalCallV1(
                            tool_call_id=call_id,
                            tool_name=tool_name,
                            original_arguments=parsed_arguments,
                            effective_arguments=parsed_arguments,
                            argument_hash=tool_argument_hash(
                                tool_call_id=call_id,
                                tool_name=tool_name,
                                arguments=parsed_arguments,
                            ),
                            editable_params=list(policy.get("editable_params", [])),
                            requires_confirmation=requires_confirmation,
                            decision=decision,
                        ))
                    batch = ToolApprovalBatchV1.create(
                        trace_id=trace_id,
                        assistant_message_id=assistant_msg.message_id,
                        assistant_sequence=assistant_msg.sequence,
                        calls=approval_calls,
                    )
                    if not self.trace_store or not hasattr(
                        self.trace_store, "replace_tool_approval_batch"
                    ):
                        raise RuntimeError(
                            "persistent TraceStore tool approval support is required"
                        )
                    await self.trace_store.replace_tool_approval_batch(
                        trace_id,
                        batch,
                    )
                    await self.trace_store.update_trace(
                        trace_id,
                        status="waiting_confirmation",
                        head_sequence=head_seq,
                    )
                    trace.status = "waiting_confirmation"
                    try:
                        from cyber_agent.trace.websocket import broadcast_trace_status_changed
                        await broadcast_trace_status_changed(
                            trace_id,
                            "waiting_confirmation",
                        )
                    except Exception:
                        pass
                    yield trace
                    return

            if tool_calls and config.auto_execute_tools:
                if (
                    runtime_policy.mode is AgentMode.RECURSIVE
                    and self.is_cancel_requested(trace_id)
                ):
                    trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
                    if trace_obj:
                        yield trace_obj
                    return
                history.append({
                    "role": "assistant",
                    "content": response_content,
                    "tool_calls": tool_calls,
                })

                if config.parallel_tool_execution:
                    # === 并发执行 ===
                    current_goal_id = goal_tree.current_id if (goal_tree and goal_tree.current_id) else None
                    async def _execute_single_tool(tc: dict) -> tuple:
                        tool_name = tc["function"]["name"]
                        tool_args = tc["function"]["arguments"]
                        if protocol_batch_error:
                            return (
                                tc,
                                {},
                                json.dumps({
                                    "status": "failed",
                                    "error": protocol_batch_error,
                                }, ensure_ascii=False),
                            )
                        if isinstance(tool_args, str):
                            if not tool_args.strip():
                                tool_args = {}
                            else:
                                try:
                                    tool_args = json.loads(tool_args)
                                except json.JSONDecodeError:
                                    tool_args = self._try_fix_json(tool_args)
                                    if tool_args is None:
                                        self.log.warning(f"[Tool Call] JSON 解析失败: {tc['function']['arguments'][:200]}")
                                        tc["function"]["arguments"] = json.dumps({"_error": "JSON parse failed", "_raw": tc["function"]["arguments"][:200]}, ensure_ascii=False)
                                        return (tc, None, f"Error: 工具参数 JSON 格式错误，无法解析。原始参数: {tc['function']['arguments'][:200]}")
                        elif tool_args is None:
                            tool_args = {}
                        args_str = json.dumps(tool_args, ensure_ascii=False)
                        args_display = args_str[:100] + "..." if len(args_str) > 100 else args_str
                        self.log.info(f"[Tool Call] {tool_name}({args_display})")
                        trigger_event_for_tool = None
                        if side_branch_ctx and side_branch_ctx.type == "knowledge_eval" and self.trace_store:
                            current_trace = await self.trace_store.get_trace(trace_id)
                            if current_trace:
                                trigger_event_for_tool = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown")
                        if tool_name in ("toolhub_call", "toolhub_search", "toolhub_health"):
                            try:
                                from cyber_agent.tools.builtin.toolhub import set_trace_context
                                set_trace_context(trace_id)
                            except ImportError:
                                pass
                        try:
                            tool_result = await self.tools.execute(
                                tool_name,
                                tool_args,
                                uid=config.uid or "",
                                context=self._build_tool_context(
                                    config=config,
                                    trace=trace,
                                    trace_id=trace_id,
                                    goal_id=current_goal_id,
                                    goal_tree=goal_tree,
                                    sequence=sequence,
                                    side_branch_ctx=side_branch_ctx,
                                    trigger_event=trigger_event_for_tool,
                                ),
                                allowed_tool_names=dispatch_allowlist,
                            )
                            return (tc, tool_args, tool_result)
                        except Exception as e:
                            import traceback
                            return (tc, tool_args, f"Error executing tool {tool_name}: {str(e)}\n{traceback.format_exc()}")
                    tasks = [_execute_single_tool(tc) for tc in tool_calls]
                    results = await asyncio.gather(*tasks)
                    for res in results:
                        tc, tool_args, tool_result = res
                        tool_name = tc["function"]["name"]
                        if tool_args is None:
                            history.append({"role": "tool", "tool_call_id": tc["id"], "name": tool_name, "content": tool_result})
                            yield Message.create(trace_id=trace_id, role="tool", sequence=sequence, parent_sequence=head_seq, tool_call_id=tc["id"], content=tool_result)
                            head_seq = sequence
                            sequence += 1
                            continue
                        if tool_name == "goal" and goal_tree:
                            self.log.debug(f"[Goal Tool] After execution: goal_tree.goals={len(goal_tree.goals)}, current_id={goal_tree.current_id}")
                        if tool_name == "upload_knowledge" and isinstance(tool_result, dict):
                            self.log.info(f"[Knowledge Tracking] 知识已上传")
                        if isinstance(tool_result, str):
                            tool_result = {"text": tool_result}
                        elif not isinstance(tool_result, dict):
                            tool_result = {"text": str(tool_result)}
                        tool_text = tool_result.get("text", str(tool_result))
                        tool_images = tool_result.get("images", [])
                        tool_usage = tool_result.get("tool_usage")
                        artifact_refs = tool_result.get("artifact_refs", [])
                        if tool_images:
                            tool_result_text = tool_text
                            tool_content_for_llm = [{"type": "text", "text": tool_text}]
                            for img in tool_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']}"}})
                                elif img.get("type") == "url" and img.get("url"):
                                    tool_content_for_llm.append({"type": "image_url", "image_url": {"url": img["url"]}})
                        else:
                            tool_result_text = tool_text
                            tool_content_for_llm = tool_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"], branch_type=side_branch_ctx.type if side_branch_ctx else None, branch_id=side_branch_ctx.branch_id if side_branch_ctx else None, content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs})
                        if self.trace_store:
                            await self.trace_store.add_message(tool_msg)
                            if tool_usage:
                                await self.trace_store.record_model_usage(trace_id=trace_id, sequence=sequence, role="tool", tool_name=tool_name, model=tool_usage.get("model"), prompt_tokens=tool_usage.get("prompt_tokens", 0), completion_tokens=tool_usage.get("completion_tokens", 0), cache_read_tokens=tool_usage.get("cache_read_tokens", 0))
                                await self.record_recursive_tool_usage(
                                    trace_id,
                                    tool_usage,
                                )
                            if tool_images:
                                import base64 as b64mod
                                for img in tool_images:
                                    if img.get("data"):
                                        await self.trace_store.write_message_attachment(
                                            trace_id,
                                            tool_msg.message_id,
                                            suffix=".png",
                                            data=b64mod.b64decode(img["data"]),
                                        )
                                        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, "_message_id": tool_msg.message_id})
                        if tool_name == "skill" and tc["id"].startswith("call_skill_"):
                            try:
                                skill_args = json.loads(tc["function"]["arguments"]) if isinstance(tc["function"]["arguments"], str) else tc["function"]["arguments"]
                                injected_skill_name = skill_args.get("skill_name", "")
                                if injected_skill_name:
                                    await self._update_skill_injection_record(trace_id, trace, injected_skill_name, tool_msg.message_id, tool_msg.sequence)
                                    self.log.info(f"[Skill 指定注入] 已记录 {injected_skill_name} → msg={tool_msg.message_id}")
                            except Exception as e:
                                self.log.warning(f"[Skill 指定注入] 记录追踪失败: {e}")
                else:
                    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):
                            if not tool_args.strip():
                                tool_args = {}
                            else:
                                try:
                                    tool_args = json.loads(tool_args)
                                except json.JSONDecodeError:
                                    # 尝试修复常见的截断/格式问题
                                    tool_args = self._try_fix_json(tool_args)
                                    if tool_args is None:
                                        self.log.warning(f"[Tool Call] JSON 解析失败，跳过工具调用 {tool_name}: {tc['function']['arguments'][:200]}")
                                        # 修复 history 中 assistant message 里的残缺 JSON，
                                        # 避免 Qwen API 拒绝 "function.arguments must be in JSON format"
                                        tc["function"]["arguments"] = json.dumps(
                                            {"_error": "JSON parse failed", "_raw": tc["function"]["arguments"][:200]},
                                            ensure_ascii=False,
                                        )
                                        history.append({
                                            "role": "tool",
                                            "tool_call_id": tc["id"],
                                            "content": f"Error: 工具参数 JSON 格式错误，无法解析。请重新生成正确的 JSON 参数调用此工具。原始参数: {tc['function']['arguments'][:200]}",
                                        })
                                        # 注意：这里不 yield Message，因为缺少必需参数会导致错误
                                        # yield Message 应该由 trace_store 统一管理
                                        continue
                        elif tool_args is None:
                            tool_args = {}
    
                        # 记录工具调用（INFO 级别，显示参数）
                        args_str = json.dumps(tool_args, ensure_ascii=False)
                        args_display = args_str[:100] + "..." if len(args_str) > 100 else args_str
                        self.log.info(f"[Tool Call] {tool_name}({args_display})")
    
                        # 获取trigger_event（如果在knowledge_eval侧分支中）
                        trigger_event_for_tool = None
                        if side_branch_ctx and side_branch_ctx.type == "knowledge_eval" and self.trace_store:
                            current_trace = await self.trace_store.get_trace(trace_id)
                            if current_trace:
                                trigger_event_for_tool = current_trace.context.get("active_side_branch", {}).get("trigger_event", "unknown")
    
                        # 设置 trace_id 上下文供 toolhub 使用（图片保存到 outputs/{trace_id}/）
                        if tool_name in ("toolhub_call", "toolhub_search", "toolhub_health"):
                            try:
                                from cyber_agent.tools.builtin.toolhub import set_trace_context
                                set_trace_context(trace_id)
                            except ImportError:
                                pass
    
                        if protocol_batch_error:
                            tool_result = json.dumps({
                                "status": "failed",
                                "error": protocol_batch_error,
                            }, ensure_ascii=False)
                        else:
                            tool_result = await self.tools.execute(
                                tool_name,
                                tool_args,
                                uid=config.uid or "",
                                context=self._build_tool_context(
                                    config=config,
                                    trace=trace,
                                    trace_id=trace_id,
                                    goal_id=current_goal_id,
                                    goal_tree=goal_tree,
                                    sequence=sequence,
                                    side_branch_ctx=side_branch_ctx,
                                    trigger_event=trigger_event_for_tool,
                                ),
                                allowed_tool_names=dispatch_allowlist,
                            )
    
                        # 如果是 goal 工具，记录执行后的状态
                        if tool_name == "goal" and goal_tree:
                            self.log.debug(f"[Goal Tool] After execution: goal_tree.goals={len(goal_tree.goals)}, current_id={goal_tree.current_id}")
    
                        # 跟踪上传的知识（通过 upload_knowledge）
                        if tool_name == "upload_knowledge" and isinstance(tool_result, dict):
                            metadata = tool_result.get("metadata", {})
                            # upload_knowledge 返回的是统计信息，不是单个 knowledge_id
                            # 这里只记录上传动作，不跟踪具体 ID
                            self.log.info(f"[Knowledge Tracking] 知识已上传到 Knowledge Manager")
    
                        # --- 支持多模态工具反馈 ---
                        # execute() 返回 dict{"text","images","tool_usage"} 或 str
                        # 统一为dict格式
                        if isinstance(tool_result, str):
                            tool_result = {"text": tool_result}
    
                        tool_text = tool_result.get("text", str(tool_result))
                        tool_images = tool_result.get("images", [])
                        tool_usage = tool_result.get("tool_usage")  # 新增：提取tool_usage
                        artifact_refs = tool_result.get("artifact_refs", [])
    
                        # 处理多模态消息
                        if tool_images:
                            tool_result_text = tool_text
                            # 构建多模态消息格式
                            tool_content_for_llm = [{"type": "text", "text": tool_text}]
                            for img in tool_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']}"
                                        }
                                    })
                                elif img.get("type") == "url" and img.get("url"):
                                    tool_content_for_llm.append({
                                        "type": "image_url",
                                        "image_url": {
                                            "url": img["url"]
                                        }
                                    })
                            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_result_text = tool_text
                            tool_content_for_llm = tool_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"],
                            branch_type=side_branch_ctx.type if side_branch_ctx else None,
                            branch_id=side_branch_ctx.branch_id if side_branch_ctx else None,
                            # 存储完整内容：有图片时保留 list（含 image_url），纯文本时存字符串
                            content={"tool_name": tool_name, "result": tool_content_for_llm, "artifact_refs": artifact_refs},
                        )
    
                        if self.trace_store:
                            await self.trace_store.add_message(tool_msg)
                            # 记录工具的模型使用
                            if tool_usage:
                                await self.trace_store.record_model_usage(
                                    trace_id=trace_id,
                                    sequence=sequence,
                                    role="tool",
                                    tool_name=tool_name,
                                    model=tool_usage.get("model"),
                                    prompt_tokens=tool_usage.get("prompt_tokens", 0),
                                    completion_tokens=tool_usage.get("completion_tokens", 0),
                                    cache_read_tokens=tool_usage.get("cache_read_tokens", 0),
                                )
                                await self.record_recursive_tool_usage(
                                    trace_id,
                                    tool_usage,
                                )
                            # 截图单独存为同名 PNG 文件
                            if tool_images:
                                import base64 as b64mod
                                for img in tool_images:
                                    if img.get("data"):
                                        png_path = await self.trace_store.write_message_attachment(
                                            trace_id,
                                            tool_msg.message_id,
                                            suffix=".png",
                                            data=b64mod.b64decode(img["data"]),
                                        )
                                        print(f"[Runner] 截图已保存: {png_path.name}")
                                        break  # 只存第一张
    
                        # 如果在侧分支，tool_msg 已持久化（不需要额外维护）
    
                        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,
                            "_message_id": tool_msg.message_id,
                        })
    
                        # 更新 skill 注入追踪记录
                        if tool_name == "skill" and tc["id"].startswith("call_skill_"):
                            try:
                                skill_args = json.loads(tc["function"]["arguments"]) if isinstance(tc["function"]["arguments"], str) else tc["function"]["arguments"]
                                injected_skill_name = skill_args.get("skill_name", "")
                                if injected_skill_name:
                                    await self._update_skill_injection_record(
                                        trace_id, trace, injected_skill_name,
                                        tool_msg.message_id, tool_msg.sequence,
                                    )
                                    self.log.info(f"[Skill 指定注入] 已记录 {injected_skill_name} → msg={tool_msg.message_id}")
                            except Exception as e:
                                self.log.warning(f"[Skill 指定注入] 记录追踪失败: {e}")

                # on_complete 模式：goal(done=...) 后立即压缩该 goal 的消息
                if (
                    not side_branch_ctx
                    and config.goal_compression == "on_complete"
                    and self.trace_store
                    and goal_tree
                ):
                    has_goal_done = False
                    for tc in tool_calls:
                        if tc["function"]["name"] != "goal":
                            continue
                        try:
                            raw = tc["function"]["arguments"]
                            args = json.loads(raw) if isinstance(raw, str) and raw.strip() else {}
                        except (json.JSONDecodeError, TypeError):
                            args = {}
                        if args.get("done") is not None:
                            has_goal_done = True
                            break

                    if has_goal_done:
                        main_path_msgs = await self.trace_store.get_main_path_messages(
                            trace_id, head_seq
                        )
                        compressed_msgs = compress_completed_goals(main_path_msgs, goal_tree)
                        if len(compressed_msgs) < len(main_path_msgs):
                            self.log.info(
                                "on_complete 压缩: %d -> %d 条消息",
                                len(main_path_msgs), len(compressed_msgs),
                            )
                            history = [msg.to_llm_dict() for msg in compressed_msgs]

                continue  # 继续循环

            # 无工具调用
            # 如果在侧分支中，已经在上面处理过了（不会走到这里）
            # 主路径无工具调用 → 任务完成，检查是否需要完成后反思或知识评估

            if not side_branch_ctx and self.trace_store:
                fresh_trace = await self.trace_store.get_trace(trace_id)
                if fresh_trace:
                    trace = fresh_trace
                policy = policy_from_context(trace.context)
                if policy.requires_task_protocol:
                    state = ensure_task_protocol(trace.context)
                    missing_report = bool(
                        trace.parent_trace_id and state.get("task_report") is None
                    )
                    pending_reviews = bool(state["pending_reviews"])
                    pending_actions = bool(state["next_actions"])
                    progress_error = (
                        task_progress_readiness_error(state)
                        if policy.requires_task_progress
                        and not trace.parent_trace_id
                        else None
                    )
                    if (
                        missing_report
                        or pending_reviews
                        or pending_actions
                        or progress_error
                    ):
                        attempts = state["protocol_correction_attempts"]
                        if attempts < 2:
                            state["protocol_correction_attempts"] = attempts + 1
                            await self.trace_store.update_trace(
                                trace_id,
                                context=trace.context,
                            )
                            if pending_reviews:
                                required_action = (
                                    "review every pending child report with "
                                    "review_task_result"
                                )
                            elif pending_actions:
                                required_action = (
                                    "execute the approved next action with agent"
                                )
                            elif missing_report:
                                required_action = (
                                    "submit a valid TaskReport with submit_task_report"
                                )
                            else:
                                required_action = (
                                    "update TaskProgress to a ready_to_submit snapshot "
                                    f"({progress_error})"
                                )
                            history.append({
                                "role": "user",
                                "content": (
                                    "Protocol gate: you cannot finish yet. You must "
                                    f"{required_action}."
                                ),
                            })
                            continue

                        if missing_report:
                            report = protocol_error_report(
                                trace_id,
                                "No valid TaskReport after two correction attempts",
                            )
                            state["task_report"] = report.model_dump()
                            state["task_report_submitted_at_sequence"] = sequence
                            state["task_report_progress_revision"] = state.get(
                                "task_progress_head_revision"
                            )
                        completion_status = "failed"
                        await self.trace_store.update_trace(
                            trace_id,
                            context=trace.context,
                            error_message="Recursive task protocol gate failed",
                        )
                        break

            # 检查是否有待评估的知识
            if not side_branch_ctx and self.trace_store:
                pending = await self.trace_store.get_pending_knowledge_entries(trace_id)
                if pending:
                    self.log.info(f"任务即将结束，但仍有 {len(pending)} 条知识未评估，强制触发评估")
                    config.force_side_branch = ["knowledge_eval"]
                    trace = await self.trace_store.get_trace(trace_id)
                    if trace:
                        trace.context["knowledge_eval_trigger"] = "task_completion"
                        await self.trace_store.update_trace(trace_id, context=trace.context)
                    continue

            if not side_branch_ctx and config.knowledge.enable_completion_extraction and not break_after_side_branch:
                config.force_side_branch = ["reflection"]
                break_after_side_branch = True
                self.log.info("任务完成，进入完成后反思侧分支")
                continue

            if (
                not side_branch_ctx
                and self.trace_store
                and policy_from_context(trace.context).requires_task_protocol
                and not trace.parent_trace_id
            ):
                state = ensure_task_protocol(trace.context)
                if (
                    state["root_validation_attempts"] >= 2
                ):
                    raise ValueError(
                        "Root task already used its two independent validation "
                        "attempts; create a new trace"
                    )
                # The Validator must read the exact persisted main path that
                # produced this candidate, including real Tool Results.
                await self.trace_store.update_trace(
                    trace_id,
                    head_sequence=head_seq,
                )
                trace.head_sequence = head_seq
                validation_run = await self.validate_recursive_trace(
                    trace_id,
                    scope="root",
                    completion_criteria=require_root_task_anchor(
                        trace.context
                    ).completion_criteria,
                    candidate_output=response_content,
                    root_validator=True,
                )
                refreshed_trace = await self.trace_store.get_trace(trace_id)
                if not refreshed_trace:
                    raise ValueError(f"Trace not found after root validation: {trace_id}")
                trace = refreshed_trace
                state = ensure_task_protocol(trace.context)
                validation_cache = state.get("task_report_validation") or {}
                validation_record = {
                    **validation_run.result.model_dump(),
                    "validation_plan": validation_cache.get("validation_plan"),
                    "validated_at_sequence": head_seq,
                }
                state["root_validation_history"].append(validation_record)
                state["root_validation_attempts"] += 1
                state["root_validation_passed"] = (
                    validation_run.result.outcome == "passed"
                )
                await self.trace_store.update_trace(
                    trace_id,
                    context=trace.context,
                )
                if not state["root_validation_passed"]:
                    if state["root_validation_attempts"] < 2:
                        validation_feedback = {
                            "role": "user",
                            "content": (
                                "Root validation did not pass. Revise the current "
                                "answer using this framework-owned ValidationResult:\n"
                                + json.dumps(
                                    validation_run.result.model_dump(),
                                    ensure_ascii=False,
                                )
                            ),
                        }
                        history.append(validation_feedback)
                        feedback_msg = Message.from_llm_dict(
                            validation_feedback,
                            trace_id=trace_id,
                            sequence=sequence,
                            goal_id=(goal_tree.current_id if goal_tree else None),
                            parent_sequence=head_seq,
                        )
                        await self.trace_store.add_message(feedback_msg)
                        head_seq = sequence
                        sequence += 1
                        await self.trace_store.update_trace(
                            trace_id,
                            head_sequence=head_seq,
                        )
                        trace.head_sequence = head_seq
                        if goal_tree and goal_tree.current_id:
                            goal = goal_tree.find(goal_tree.current_id)
                            if goal:
                                goal.status = "in_progress"
                                await self.trace_store.update_goal_tree(
                                    trace_id,
                                    goal_tree,
                                )
                        continue
                    completion_status = "failed"
                    await self.trace_store.update_trace(
                        trace_id,
                        error_message="Root task did not pass independent validation",
                    )

            break

        if (
            policy_from_context(trace.context).mode is AgentMode.RECURSIVE
            and self.is_cancel_requested(trace_id)
        ):
            trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
            if trace_obj:
                yield trace_obj
            return

        # max_iterations 等非正常退出也必须经过同一完成门禁。
        if self.trace_store:
            fresh_trace = await self.trace_store.get_trace(trace_id)
            if fresh_trace:
                trace = fresh_trace
            policy = policy_from_context(trace.context)
            if policy.requires_task_protocol:
                state = ensure_task_protocol(trace.context)
                if trace.parent_trace_id and state.get("task_report") is None:
                    report = protocol_error_report(
                        trace_id,
                        "Agent loop ended without a valid TaskReport",
                    )
                    state["task_report"] = report.model_dump()
                    state["task_report_submitted_at_sequence"] = sequence
                    state["task_report_progress_revision"] = state.get(
                        "task_progress_head_revision"
                    )
                    completion_status = "failed"
                    await self.trace_store.update_trace(
                        trace_id,
                        context=trace.context,
                        error_message="Recursive task protocol gate failed",
                    )
                if state["pending_reviews"]:
                    completion_status = "failed"
                if state["next_actions"]:
                    completion_status = "failed"
                if (
                    not trace.parent_trace_id
                    and not state.get("root_validation_passed")
                ):
                    completion_status = "failed"

        # 清理 trace 相关的跟踪数据
        self._context_warned.pop(trace_id, None)
        self._context_usage.pop(trace_id, None)
        self._saved_knowledge_ids.pop(trace_id, None)

        # 更新 head_sequence 并完成 Trace
        if self.trace_store:
            await self.trace_store.update_trace(
                trace_id,
                status=completion_status,
                head_sequence=head_seq,
                completed_at=datetime.now(),
            )
            if self.event_service is not None:
                await self.event_service.try_pump(
                    trace.context.get("root_trace_id") or trace_id
                )
            trace_obj = await self.trace_store.get_trace(trace_id)
            if trace_obj:
                yield trace_obj

    # ===== 压缩辅助方法 =====

    def _rebuild_history_after_compression(
        self,
        history: List[Dict],
        summary_msg_dict: Dict,
        label: str = "压缩",
    ) -> List[Dict]:
        """
        压缩后重建 history：system prompt + 第一条 user message + summary

        Args:
            history: 压缩前的 history
            summary_msg_dict: summary 消息的 LLM dict
            label: 日志标签

        Returns:
            新的 history
        """
        system_msg = None
        first_user_msg = None
        for msg in history:
            if msg.get("role") == "system" and not system_msg:
                system_msg = msg
            elif msg.get("role") == "user" and not first_user_msg:
                first_user_msg = msg
            if system_msg and first_user_msg:
                break

        new_history = []
        if system_msg:
            new_history.append(system_msg)
        if first_user_msg:
            new_history.append(first_user_msg)
        new_history.append(summary_msg_dict)

        self.log.info(f"{label}完成: {len(history)} → {len(new_history)} 条消息")
        for idx, msg in enumerate(new_history):
            role = msg.get("role", "unknown")
            content = msg.get("content", "")
            if isinstance(content, str):
                preview = content
            elif isinstance(content, list):
                preview = f"[{len(content)} blocks]"
            else:
                preview = str(content)
            self.log.info(f"  {label}后[{idx}] {role}: {preview}")

        return new_history

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

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

        ``_prepare_existing_trace`` 判定为回溯时调用；Recursive 同时重建待审核和待重规划状态。

        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)
        if self.candidate_service is not None:
            await self.candidate_service.assert_rewind_allowed(trace_id, cutoff)

        rewind_operation_id = (
            f"rewind:{trace_id}:"
            f"{max((item.sequence for item in all_messages), default=0)}:{cutoff}"
        )
        rewind_payload: dict[str, Any] = {
            "operation_id": rewind_operation_id,
            "after_sequence": cutoff,
            "head_sequence": cutoff,
        }

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

            rewind_payload["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

        trace = await self.trace_store.get_trace(trace_id)
        if trace and policy_from_context(trace.context).requires_task_protocol:
            state = ensure_task_protocol(trace.context)
            submitted_at = state.get("task_report_submitted_at_sequence")
            if isinstance(submitted_at, int) and submitted_at > cutoff:
                if state.get("task_report"):
                    state["report_history"].append(state["task_report"])
                state["task_report"] = None
                state["task_report_submitted_at_sequence"] = None
                state["task_report_progress_revision"] = None
                state["task_report_validation"] = None
            validation_cache = state.get("task_report_validation")
            if (
                isinstance(validation_cache, dict)
                and int(validation_cache.get("validated_at_sequence", 0) or 0)
                > cutoff
            ):
                state["task_report_validation"] = None
            state["pending_reviews"] = {
                child_id: entry
                for child_id, entry in state["pending_reviews"].items()
                if entry.get("received_at_sequence", 0) <= cutoff
            }
            reverted_reviews = [
                review for review in state["reviews"]
                if review.get("reviewed_at_sequence", 0) > cutoff
            ]
            for review in reverted_reviews:
                entry = review.get("pending_review")
                child_id = review.get("child_trace_id")
                if (
                    child_id
                    and isinstance(entry, dict)
                    and entry.get("received_at_sequence", 0) <= cutoff
                ):
                    state["pending_reviews"][child_id] = entry
            state["reviews"] = [
                review for review in state["reviews"]
                if review.get("reviewed_at_sequence", 0) <= cutoff
            ]
            state["next_actions"] = [
                action for action in state["next_actions"]
                if action.get("created_at_sequence", 0) <= cutoff
            ]
            while (
                state.get("task_brief") is not None
                and state.get("task_brief_effective_at_sequence", 0) > cutoff
                and state.get("task_brief_history")
            ):
                previous = state["task_brief_history"].pop()
                state["task_brief"] = previous["task_brief"]
                state["task_brief_version"] = previous["version"]
                state["task_brief_effective_at_sequence"] = previous.get(
                    "effective_at_sequence",
                    0,
                )
            if policy_from_context(trace.context).requires_task_progress:
                rewind_task_progress(state, cutoff)
            state["pending_replans"] = rebuild_pending_replans(state)
            root_history = [
                item
                for item in state.get("root_validation_history", [])
                if int(
                    item.get(
                        "validated_at_sequence",
                        item.get("evaluated_at_sequence", 0),
                    )
                    or 0
                ) <= cutoff
            ]
            state["root_validation_history"] = root_history
            state["root_validation_attempts"] = len(root_history)
            state["root_validation_passed"] = bool(
                root_history and root_history[-1].get("outcome") == "passed"
            )
            state["protocol_correction_attempts"] = 0
            prune_context_access(trace.context, cutoff)
            operations = trace.context.setdefault("run_event_operations", [])
            if not any(
                item.get("operation_id") == rewind_operation_id
                for item in operations
                if isinstance(item, dict)
            ):
                operations.append({
                    "operation_id": rewind_operation_id,
                    "event_type": "run.rewound",
                    "effective_at_sequence": cutoff,
                    "payload": {"after_sequence": cutoff},
                })
                del operations[:-64]
            await self.trace_store.update_trace(trace_id, context=trace.context)

            projected_statuses: dict[str, str] = {}
            for entry in state["pending_reviews"].values():
                if entry.get("goal_id"):
                    projected_statuses[entry["goal_id"]] = "pending_review"
            for action in state["next_actions"]:
                goal_id = action.get("goal_id")
                if goal_id and goal_id not in projected_statuses:
                    projected_statuses[goal_id] = "in_progress"
            if goal_tree and projected_statuses:
                for goal_id, status in projected_statuses.items():
                    goal = goal_tree.find(goal_id)
                    if goal:
                        goal.status = status
                await self.trace_store.update_goal_tree(trace_id, goal_tree)

        # 4. 更新 head_sequence 到 rewind 点
        await self.trace_store.update_trace(trace_id, head_sequence=cutoff)
        await self.trace_store.append_event(trace_id, "rewind", rewind_payload)
        if self.event_service is not None:
            await self.event_service.emit_after_commit(
                source_trace_id=trace_id,
                event_type="run.rewound",
                event_key=f"run.rewound:{rewind_operation_id}",
                effective_at_sequence=cutoff,
                payload={"after_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

        self.log.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 = build_tool_interrupted_message(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_INTERRUPTED_SUMMARY,
            "task": task,
        }
        if sub_trace_id:
            result["sub_trace_id"] = sub_trace_id
            result["hint"] = build_agent_continue_hint(sub_trace_id)
        if stats:
            result["stats"] = stats

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

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

    # ===== Skill 指定注入 =====

    def _check_skills_need_injection(
        self,
        trace: Trace,
        inject_skills: List[str],
        history: List[Dict],
        recency_threshold: int,
    ) -> List[str]:
        """
        检查哪些 skill 需要注入。

        通过 trace.context["injected_skills"] 中记录的 message_id
        检查是否仍在当前 history 的最近 recency_threshold 条消息中。

        Returns:
            需要注入的 skill 名称列表
        """
        injected = (trace.context or {}).get("injected_skills", {})
        # 收集 history 中最近 recency_threshold 条消息的 message_id
        recent_msgs = history[-recency_threshold:] if recency_threshold > 0 else []
        recent_ids = set()
        for msg in recent_msgs:
            mid = msg.get("message_id") or msg.get("_message_id")
            if mid:
                recent_ids.add(mid)

        needs_inject = []
        for skill_name in inject_skills:
            record = injected.get(skill_name)
            if not record:
                needs_inject.append(skill_name)
                continue
            if record.get("message_id") not in recent_ids:
                needs_inject.append(skill_name)

        return needs_inject

    async def _update_skill_injection_record(
        self,
        trace_id: str,
        trace: Trace,
        skill_name: str,
        message_id: str,
        sequence: int,
    ):
        """更新 trace.context 中的 skill 注入记录"""
        if not trace.context:
            trace.context = {}
        if "injected_skills" not in trace.context:
            trace.context["injected_skills"] = {}
        trace.context["injected_skills"][skill_name] = {
            "message_id": message_id,
            "sequence": sequence,
        }
        if self.trace_store:
            await self.trace_store.update_trace(trace_id, context=trace.context)

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

    def _build_context_injection(
        self,
        trace: Trace,
        goal_tree: Optional[GoalTree],
    ) -> str:
        """构建周期性注入的上下文（GoalTree + Active Collaborators + Focus 提醒 + IM 消息通知）"""
        from datetime import datetime
        parts = [f"## Current Time\n\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]

        if trace and policy_from_context(trace.context).requires_task_protocol:
            parts.append(render_recursive_context(trace.context))

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

            if goal_tree.current_id:
                # 检测 focus 在有子节点的父目标上：提醒模型 focus 到具体子目标
                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}）再执行。"
                    )
            else:
                # 无焦点：提醒模型 focus
                parts.append(
                    "**提醒**：当前没有焦点目标。请用 `goal(focus=\"...\")` 选择一个目标开始执行。"
                )

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

        # IM 消息通知（Research Agent）
        im_config = trace.context.get("im_config")
        if im_config:
            contact_id = im_config.get("contact_id")
            chat_id = im_config.get("chat_id")
            if contact_id and chat_id:
                # 尝试导入 IM 模块并检查通知
                try:
                    from cyber_agent.tools.builtin.im import chat as im_chat
                    notification = im_chat._notifications.get((contact_id, chat_id))
                    if notification:
                        count = notification.get("count", 0)
                        senders = notification.get("from", [])
                        senders_str = ", ".join(senders)
                        parts.append(
                            f"## IM 消息通知\n\n"
                            f"你有 {count} 条新消息，来自: {senders_str}\n"
                            f"使用 `im_receive_messages(contact_id=\"{contact_id}\", chat_id=\"{chat_id}\")` 查看消息内容。"
                        )
                    else:
                        parts.append("## IM 消息通知\n\n暂无新消息")
                except (ImportError, AttributeError):
                    # IM 模块未加载或不可用
                    pass

        # Knowledge Manager 队列状态
        km_queue_size = trace.context.get("km_queue_size")
        if km_queue_size is not None:
            current_sender = trace.context.get("current_sender", "unknown")
            if km_queue_size > 0:
                parts.append(
                    f"## 消息队列状态\n\n"
                    f"当前处理: {current_sender} 的消息\n"
                    f"队列中还有 {km_queue_size} 条待处理消息"
                )
            else:
                parts.append(
                    f"## 消息队列状态\n\n"
                    f"当前处理: {current_sender} 的消息\n"
                    f"队列为空，处理完本条消息后将进入休眠"
                )

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

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

    async def _optimize_images(
        self,
        messages: List[Dict],
        model: str,
        *,
        trace_id: Optional[str] = None,
    ) -> List[Dict]:
        """
        分级优化已处理的图片，节省 token

        策略（基于图片距离最后一条 assistant 的"轮次"）：
        1. 最近 1-2 轮：保留原图
        2. 3-5 轮：降低分辨率和压缩（节省 token 但保留视觉信息）
        3. 5 轮以上：调用小模型生成文本描述 + 保留 URL

        处理结果会缓存，避免重复的 PIL 解码/编码和 LLM 调用。

        Args:
            messages: 原始消息列表
            model: 当前使用的模型（用于选择描述生成模型）

        Returns:
            优化后的消息列表（深拷贝）
        """
        if not messages:
            return messages

        # 找到最后一条 assistant message 的位置
        last_assistant_idx = -1
        for i in range(len(messages) - 1, -1, -1):
            if messages[i].get("role") == "assistant":
                last_assistant_idx = i
                break

        # 如果没有 assistant message，说明还没开始对话，不优化
        if last_assistant_idx == -1:
            return messages

        # 统计从每个位置到最后一条 assistant 之间的 assistant 数量（作为"轮次"）
        assistant_count_after = [0] * len(messages)
        count = 0
        for i in range(len(messages) - 1, -1, -1):
            assistant_count_after[i] = count
            if messages[i].get("role") == "assistant":
                count += 1

        # 深拷贝避免修改原始数据
        import copy
        import hashlib
        import asyncio
        import base64 as b64mod
        import httpx
        import mimetypes
        messages = copy.deepcopy(messages)

        # 预处理：将所有 HTTP(S) URL 图片下载并转为 base64 data URL
        # Qwen API 无法访问外部签名 URL（如 BFL、火山引擎 TOS），必须在本地转换
        url_download_jobs = []  # [(msg_idx, block_idx, url)]
        for i, msg in enumerate(messages):
            if msg.get("role") != "tool":
                continue
            content = msg.get("content")
            if not isinstance(content, list):
                continue
            for block_idx, block in enumerate(content):
                if isinstance(block, dict) and block.get("type") == "image_url":
                    url = block.get("image_url", {}).get("url", "")
                    if url.startswith(("http://", "https://")):
                        url_download_jobs.append((i, block_idx, url))

        if url_download_jobs:
            async def _download_image_to_data_url(url: str) -> str | None:
                try:
                    async with httpx.AsyncClient(timeout=60, trust_env=False) as client:
                        resp = await client.get(url)
                        resp.raise_for_status()
                        ct = resp.headers.get("content-type", "").split(";")[0].strip()
                        if not ct.startswith("image/"):
                            ct = mimetypes.guess_type(url.split("?")[0])[0] or "image/png"
                        b64 = b64mod.b64encode(resp.content).decode()
                        return f"data:{ct};base64,{b64}"
                except Exception:
                    return None

            results = await asyncio.gather(
                *[_download_image_to_data_url(url) for _, _, url in url_download_jobs],
                return_exceptions=True
            )
            converted = 0
            for (msg_idx, block_idx, original_url), result in zip(url_download_jobs, results):
                if isinstance(result, str) and result.startswith("data:"):
                    messages[msg_idx]["content"][block_idx]["image_url"]["url"] = result
                    converted += 1
            if converted:
                self.log.info(f"[Image Optimization] URL→base64 预转换: {converted}/{len(url_download_jobs)} 张")

        # 统计优化情况
        stats = {"kept": 0, "downscaled": 0, "described": 0, "cache_hit": 0}

        # 收集需要降分辨率或尺寸补齐的图片（用于并发处理）
        process_jobs = []  # [(msg_idx, block_idx, image_url, cache_key, max_size, cache_field)]

        # 第一遍：扫描并收集需要处理的图片
        for i in range(last_assistant_idx):
            msg = messages[i]
            if msg.get("role") != "tool":
                continue

            content = msg.get("content")
            if not isinstance(content, list):
                continue

            rounds_ago = assistant_count_after[i]

            for block_idx, block in enumerate(content):
                if isinstance(block, dict) and block.get("type") == "image_url":
                    image_url_obj = block.get("image_url", {})
                    image_url = image_url_obj.get("url", "")

                    if image_url.startswith("data:"):
                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
                    else:
                        cache_key = hashlib.md5(image_url.encode()).hexdigest()

                    # 1-5 轮都需要检查尺寸
                    if rounds_ago <= 5:
                        cached = self._image_opt_cache.get(cache_key, {})
                        cache_field = "pad_only" if rounds_ago <= 2 else "downscaled"
                        
                        if cache_field not in cached and image_url.startswith("data:"):
                            max_size = None if rounds_ago <= 2 else 512
                            process_jobs.append((i, block_idx, image_url, cache_key, max_size, cache_field))

        # 并发处理所有尺寸任务
        if process_jobs:
            process_results = await asyncio.gather(
                *[self._process_image_size(url, max_size=ms) for _, _, url, _, ms, _ in process_jobs],
                return_exceptions=True
            )
            for (_, _, _, cache_key, _, cache_field), result in zip(process_jobs, process_results):
                if not isinstance(result, Exception) and result is not None:
                    self._image_opt_cache.setdefault(cache_key, {})[cache_field] = result

        # 第二遍：应用处理结果
        for i in range(last_assistant_idx):
            msg = messages[i]
            if msg.get("role") != "tool":
                continue

            content = msg.get("content")
            if not isinstance(content, list):
                continue

            # 计算这条消息距离最后一条 assistant 的"轮次"
            rounds_ago = assistant_count_after[i]

            # 处理每个 content block
            new_content = []
            for block in content:
                if isinstance(block, dict) and block.get("type") == "image_url":
                    image_url_obj = block.get("image_url", {})
                    image_url = image_url_obj.get("url", "")

                    # 生成缓存 key（URL 图片用 URL 本身，base64 用前 64 字符 hash）
                    if image_url.startswith("data:"):
                        cache_key = hashlib.md5(image_url[:200].encode()).hexdigest()
                    else:
                        cache_key = hashlib.md5(image_url.encode()).hexdigest()

                    # 根据距离决定处理策略
                    if rounds_ago <= 2:
                        # 最近 1-2 轮：只补齐过小图片，保留原分辨率
                        cached = self._image_opt_cache.get(cache_key, {})
                        if "pad_only" in cached:
                            new_content.append({
                                "type": "image_url",
                                "image_url": {"url": cached["pad_only"]}
                            })
                            stats["kept"] += 1
                            stats["cache_hit"] += 1
                        elif image_url.startswith("data:"):
                            processed = await self._process_image_size(image_url, max_size=None)
                            if processed:
                                self._image_opt_cache.setdefault(cache_key, {})["pad_only"] = processed
                                new_content.append({
                                    "type": "image_url",
                                    "image_url": {"url": processed}
                                })
                            else:
                                new_content.append(block)
                            stats["kept"] += 1
                        else:
                            new_content.append(block)
                            stats["kept"] += 1

                    elif rounds_ago <= 5:
                        # 3-5 轮：降低分辨率（优先从缓存取）
                        cached = self._image_opt_cache.get(cache_key, {})
                        if "downscaled" in cached:
                            new_content.append({
                                "type": "image_url",
                                "image_url": {"url": cached["downscaled"]}
                            })
                            stats["downscaled"] += 1
                            stats["cache_hit"] += 1
                        elif image_url.startswith("data:"):
                            processed = await self._process_image_size(image_url, max_size=512)
                            if processed:
                                # 缓存结果
                                self._image_opt_cache.setdefault(cache_key, {})["downscaled"] = processed
                                new_content.append({
                                    "type": "image_url",
                                    "image_url": {"url": processed}
                                })
                                stats["downscaled"] += 1
                            else:
                                new_content.append(block)
                                stats["kept"] += 1
                        else:
                            # URL 图片：无法直接处理，保留原图
                            new_content.append(block)
                            stats["kept"] += 1

                    else:
                        # 5 轮以上：生成文本描述（优先从缓存取）
                        cached = self._image_opt_cache.get(cache_key, {})
                        if "description" in cached:
                            new_content.append(cached["description"])
                            stats["described"] += 1
                            stats["cache_hit"] += 1
                        else:
                            description = await self._generate_image_description(
                                image_url,
                                model,
                                trace_id=trace_id,
                            )
                            url_info = f" (URL: {image_url[:100]}...)" if not image_url.startswith("data:") else ""
                            desc_block = {
                                "type": "text",
                                "text": f"[Image description: {description}]{url_info}"
                            }
                            # 缓存结果
                            self._image_opt_cache.setdefault(cache_key, {})["description"] = desc_block
                            new_content.append(desc_block)
                            stats["described"] += 1
                else:
                    new_content.append(block)

            msg["content"] = new_content
        # print(f"[Image Opt Check] 扫描到 {stats['kept'] + stats['downscaled'] + stats['described']} 张图片上下文")
        if stats["downscaled"] > 0 or stats["described"] > 0:
            self.log.info(
                f"[Image Optimization] 保留 {stats['kept']} 张，"
                f"降分辨率 {stats['downscaled']} 张，"
                f"文本描述 {stats['described']} 张，"
                f"缓存命中 {stats['cache_hit']} 次"
            )

        return messages

    async def _process_image_size(self, base64_url: str, max_size: Optional[int] = 512, min_size: int = 11) -> Optional[str]:
        """
        处理 base64 图片的尺寸：
        - 若 max_size 不为 None 且大于该值，则等比例缩放
        - 若任意一边小于 min_size，则补充白边 (Padding)
        """
        try:
            from PIL import Image
            import io
            import base64

            # 解析 base64 数据
            if not base64_url.startswith("data:"):
                return None

            header, data = base64_url.split(",", 1)
            media_type = header.split(";")[0].split(":")[1]  # image/png

            # 解码图片
            img_data = base64.b64decode(data)
            img = Image.open(io.BytesIO(img_data))

            width, height = img.size

            needs_downscale = max_size is not None and (width > max_size or height > max_size)
            needs_pad = width < min_size or height < min_size

            # 尺寸正常，无需处理
            if not needs_downscale and not needs_pad:
                return base64_url

            new_width, new_height = width, height

            # 1. 降分辨率
            if needs_downscale:
                if width > height:
                    new_width = max_size
                    new_height = int(height * max_size / width)
                else:
                    new_height = max_size
                    new_width = int(width * max_size / height)
            
            if (new_width, new_height) != (width, height):
                img_resized = img.resize((new_width, new_height), Image.Resampling.BILINEAR)
            else:
                img_resized = img

            # 2. 补齐白边 (Padding)
            pad_width = max(new_width, min_size)
            pad_height = max(new_height, min_size)

            if pad_width > new_width or pad_height > new_height:
                # 创建白色背景
                padded_img = Image.new("RGBA" if img_resized.mode in ("RGBA", "P") else "RGB", (pad_width, pad_height), (255, 255, 255, 255))
                offset_x = (pad_width - new_width) // 2
                offset_y = (pad_height - new_height) // 2
                padded_img.paste(img_resized, (offset_x, offset_y))
                img_resized = padded_img

            # 转换为 RGB（JPEG不支持 RGBA, P 等具有透明度或索引的模式）
            if img_resized.mode != "RGB":
                if img_resized.mode == "RGBA" or img_resized.mode == "P":
                    # Create a white background for transparent images
                    background = Image.new("RGB", img_resized.size, (255, 255, 255))
                    if img_resized.mode == "P" and "transparency" in img_resized.info:
                        img_resized = img_resized.convert("RGBA")
                    if img_resized.mode == "RGBA":
                        background.paste(img_resized, mask=img_resized.split()[3])
                        img_resized = background
                img_resized = img_resized.convert("RGB")

            # 重新编码为 JPEG（如果只是补齐没有缩放，可以稍微保留高点质量）
            buffer = io.BytesIO()
            quality = 60 if needs_downscale else 85
            img_resized.save(buffer, format="JPEG", quality=quality, optimize=False)
            new_data = base64.b64encode(buffer.getvalue()).decode("utf-8")

            return f"data:image/jpeg;base64,{new_data}"

        except Exception as e:
            self.log.warning(f"[Image Process] 处理图片尺寸失败: {e}")
            return None

    async def _generate_image_description(
        self,
        image_url: str,
        current_model: str,
        *,
        trace_id: Optional[str] = None,
    ) -> str:
        """
        使用小模型生成图片的文本描述

        Args:
            image_url: 图片 URL（base64 或 http(s)）
            current_model: 当前使用的模型

        Returns:
            图片描述文本
        """
        try:
            # 使用 qwen-vl-max（通义千问视觉模型）生成描述
            # 注意：qwen-vl 系列专门支持视觉输入
            description_model = "qwen-vl-max"

            # 构建描述请求
            messages = [
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "image_url",
                            "image_url": {"url": image_url}
                        },
                        {
                            "type": "text",
                            "text": "请用 1-2 句话简洁描述这张图片的主要内容。"
                        }
                    ]
                }
            ]

            # 调用 LLM
            call_kwargs = {
                "messages": messages,
                "model": description_model,
                "tools": None,
                "temperature": 0.3,
            }
            result = (
                await self.call_recursive_llm(
                    trace_id,
                    purpose="ordinary",
                    fail_on_post_response_exhaustion=True,
                    **call_kwargs,
                )
                if trace_id
                else await self.llm_call(**call_kwargs)
            )

            description = result.get("content", "").strip()
            return description if description else "图片内容"

        except (ResourceBudgetExceeded, ResourceBudgetStateError):
            raise
        except Exception as e:
            self.log.warning(f"[Image Description] 生成描述失败: {e}")
            return "图片内容"

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

        策略：固定位置 + 延迟缓存
        1. 如果有未处理的图片（最后一条 assistant 之后的 tool messages 中有图片），跳过缓存
        2. system message 添加缓存（如果足够长）
        3. 固定位置缓存点（20, 40, 60, 80），确保每个缓存点间隔 >= 1024 tokens
        4. 最多使用 4 个缓存点（含 system）

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

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

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

        # 延迟缓存：检查是否有未处理的图片
        last_assistant_idx = -1
        for i in range(len(messages) - 1, -1, -1):
            if messages[i].get("role") == "assistant":
                last_assistant_idx = i
                break

        # 检查最后一条 assistant 之后是否有包含图片的 tool messages
        has_unprocessed_images = False
        if last_assistant_idx >= 0:
            for i in range(last_assistant_idx + 1, len(messages)):
                msg = messages[i]
                if msg.get("role") == "tool":
                    content = msg.get("content")
                    if isinstance(content, list):
                        has_unprocessed_images = any(
                            isinstance(block, dict) and block.get("type") == "image_url"
                            for block in content
                        )
                        if has_unprocessed_images:
                            break

        if has_unprocessed_images:
            self.log.debug("[Cache] 检测到未处理的图片，延迟缓存建立")
            return messages

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

        # 策略 1: 为 system message 添加缓存
        system_cached = False
        for msg in messages:
            if msg.get("role") == "system":
                content = msg.get("content", "")
                if isinstance(content, str) and len(content) > 1000:
                    msg["content"] = [{
                        "type": "text",
                        "text": content,
                        "cache_control": {"type": "ephemeral"}
                    }]
                    system_cached = True
                    self.log.debug(f"[Cache] 为 system message 添加缓存标记 (len={len(content)})")
                break

        # 策略 2: 固定位置缓存点
        CACHE_INTERVAL = 20
        MAX_POINTS = 3 if system_cached else 4
        MIN_TOKENS = 1024
        AVG_TOKENS_PER_MSG = 70

        total_msgs = len(messages)
        if total_msgs == 0:
            return messages

        cache_positions = []
        last_cache_pos = 0

        for i in range(1, MAX_POINTS + 1):
            target_pos = i * CACHE_INTERVAL - 1  # 19, 39, 59, 79

            if target_pos >= total_msgs:
                break

            # 从目标位置开始查找合适的 user/assistant 消息
            for j in range(target_pos, total_msgs):
                msg = messages[j]

                if msg.get("role") not in ("user", "assistant"):
                    continue

                content = msg.get("content", "")
                if not content:
                    continue

                # 检查 content 是否非空
                is_valid = False
                if isinstance(content, str):
                    is_valid = len(content) > 0
                elif isinstance(content, list):
                    is_valid = any(
                        isinstance(block, dict) and
                        block.get("type") == "text" and
                        len(block.get("text", "")) > 0
                        for block in content
                    )

                if not is_valid:
                    continue

                # 检查 token 距离
                msg_count = j - last_cache_pos
                estimated_tokens = msg_count * AVG_TOKENS_PER_MSG

                if estimated_tokens >= MIN_TOKENS:
                    cache_positions.append(j)
                    last_cache_pos = j
                    self.log.debug(f"[Cache] 在位置 {j} 添加缓存点 (估算 {estimated_tokens} tokens)")
                    break

        # 应用缓存标记
        for idx in cache_positions:
            msg = messages[idx]
            content = msg.get("content", "")

            if isinstance(content, str):
                msg["content"] = [{
                    "type": "text",
                    "text": content,
                    "cache_control": {"type": "ephemeral"}
                }]
                self.log.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记")
            elif isinstance(content, list):
                # 在最后一个 text block 添加 cache_control
                for block in reversed(content):
                    if isinstance(block, dict) and block.get("type") == "text":
                        block["cache_control"] = {"type": "ephemeral"}
                        self.log.debug(f"[Cache] 为 message[{idx}] ({msg.get('role')}) 添加缓存标记")
                        break

        self.log.debug(
            f"[Cache] 总消息: {total_msgs}, "
            f"缓存点: {len(cache_positions)} at {cache_positions}"
        )
        return messages

    def _get_configured_tool_names(
        self,
        tools: Optional[List[str]] = None,
        tool_groups: Optional[List[str]] = None,
        exclude_tools: Optional[List[str]] = None,
    ) -> set[str]:
        """解析 RunConfig 的基础工具能力集合。"""
        if tools is not None:
            # Explicit tools are an exact capability set, not an addition to
            # the default groups.
            tool_names = set(tools)
        elif tool_groups is not None:
            tool_names = set(self.tools.get_tool_names(groups=tool_groups))
        else:
            tool_names = set(self.tools.get_tool_names())
        if exclude_tools:
            tool_names -= set(exclude_tools)
        return tool_names

    def _get_tool_schemas(
        self,
        tools: Optional[List[str]] = None,
        tool_groups: Optional[List[str]] = None,
        exclude_tools: Optional[List[str]] = None,
    ) -> List[Dict]:
        """获取 RunConfig 基础能力对应的工具 Schema。"""
        tool_names = self._get_configured_tool_names(
            tools,
            tool_groups,
            exclude_tools,
        )
        return self.tools.get_schemas(list(tool_names))

    def _get_runtime_tool_names(
        self,
        config: RunConfig,
        trace: Trace,
        *,
        in_side_branch: bool = False,
    ) -> set[str]:
        """在 RunConfig 基础能力上应用 Recursive 协议状态门禁。

        主循环每轮调用，使待审核、待执行与报告阶段只暴露当前允许的工具。
        """
        tool_names = self._get_configured_tool_names(
            config.tools,
            config.tool_groups,
            config.exclude_tools,
        )
        policy = policy_from_context(trace.context)
        protocol_tools = {
            "submit_task_report",
            "review_task_result",
            "update_task_progress",
            "read_context_ref",
            "manage_candidate",
        }

        if not policy.requires_task_protocol or in_side_branch:
            available = tool_names - protocol_tools
            if policy.requires_task_protocol:
                available.discard("evaluate")
            return available

        if self.application_binding is not None and trace.context.get(
            "application_ref"
        ):
            role = self.application_binding.role(
                trace.context.get("application_role_id")
            )
            limits = trace.context.get("effective_run_limits") or {}
            depth = int(trace.context.get("agent_depth", 0) or 0)
            if (
                not role.role.allowed_child_roles
                or depth >= int(limits.get("max_depth", policy.max_depth))
            ):
                tool_names.discard("agent")

        state = ensure_task_protocol(trace.context)
        tool_names.discard("evaluate")
        if state["pending_reviews"]:
            return tool_names & {"review_task_result", "read_context_ref"}
        if state["next_actions"]:
            return tool_names & {"agent", "read_context_ref"}

        tool_names.discard("review_task_result")
        if self.candidate_service is None or state.get("task_report") is not None:
            tool_names.discard("manage_candidate")
        if not policy.requires_task_progress or state.get("task_report") is not None:
            tool_names.discard("update_task_progress")
        if not trace.parent_trace_id or state.get("task_report") is not None:
            tool_names.discard("submit_task_report")
        return tool_names

    def _get_runtime_tool_schemas(
        self,
        config: RunConfig,
        trace: Trace,
        *,
        in_side_branch: bool = False,
        runtime_tool_names: Optional[set[str]] = None,
    ) -> List[Dict]:
        """按 Trace 持久化模式和当前协议状态生成工具 Schema。

        主循环把结果交给 LLM；Recursive revision 2 只允许受治理的本地
        ``task_brief`` 委托，Legacy 和 Recursive revision 1 继续使用 ``task``。
        """
        tool_names = (
            runtime_tool_names
            if runtime_tool_names is not None
            else self._get_runtime_tool_names(
                config,
                trace,
                in_side_branch=in_side_branch,
            )
        )
        schemas = deepcopy(self.tools.get_schemas(list(tool_names)))
        policy = policy_from_context(trace.context)

        for schema in schemas:
            function = schema.get("function", {})
            if function.get("name") != "agent":
                continue
            parameters = function.get("parameters", {})
            properties = parameters.get("properties", {})
            required = parameters.setdefault("required", [])
            if policy.requires_task_protocol:
                properties.pop("messages", None)
                properties.pop("task", None)
                if "task_brief" in properties:
                    properties["task_brief"]["description"] = (
                        "Required structured contract for Recursive local delegation. "
                        "remote_* agents are not supported in Recursive revision 3."
                    )
                if "task" in required:
                    required.remove("task")
                if "task_brief" in properties and "task_brief" not in required:
                    required.append("task_brief")
                if self.application_binding is not None and trace.context.get(
                    "application_ref"
                ):
                    role = self.application_binding.role(
                        trace.context.get("application_role_id")
                    )
                    properties.pop("skills", None)
                    if "skills" in required:
                        required.remove("skills")
                    if "agent_type" in properties:
                        properties["agent_type"]["enum"] = list(
                            role.role.allowed_child_roles
                        )
                        properties["agent_type"]["description"] = (
                            "Required application role for the direct child."
                        )
                        if "agent_type" not in required:
                            required.append("agent_type")
            else:
                properties.pop("task_brief", None)
                if "task" in properties and "task" not in required:
                    required.append("task")
        return schemas

    def _build_tool_context(
        self,
        *,
        config: RunConfig,
        trace: Trace,
        trace_id: str,
        goal_id: Optional[str],
        goal_tree: Optional[GoalTree],
        sequence: int,
        side_branch_ctx: Optional[SideBranchContext],
        trigger_event: Optional[str],
    ) -> Dict[str, Any]:
        """构建 ToolRegistry 执行时注入的隐藏上下文。

        主循环在 dispatch 前调用；Recursive 权限快照和调度配置最后覆盖，不可伪造。
        """
        framework_context = {
            "store": self.trace_store,
            "task_protocol_service": self.task_protocol_service,
            "candidate_service": self.candidate_service,
            "event_service": self.event_service,
            "trace_id": trace_id,
            "goal_id": goal_id,
            "runner": self,
            "goal_tree": goal_tree,
            "knowledge_config": config.knowledge,
            "memory_config": config.memory,
            "dream_scope": (
                DreamScope(
                    uid=trace.uid,
                    agent_type=trace.agent_type or config.agent_type,
                    memory_identity=compute_memory_identity(config.memory),
                )
                if config.memory is not None
                else None
            ),
            "sequence": sequence,
            "side_branch": {
                "type": side_branch_ctx.type,
                "branch_id": side_branch_ctx.branch_id,
                "is_side_branch": True,
                "max_turns": side_branch_ctx.max_turns,
                "trigger_event": trigger_event,
            } if side_branch_ctx else None,
        }
        if policy_from_context(trace.context).mode is AgentMode.RECURSIVE:
            context = {**(config.context or {}), **framework_context}
            context.update({
                RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: sorted(
                    self._get_configured_tool_names(
                        config.tools,
                        config.tool_groups,
                        config.exclude_tools,
                    )
                ),
                RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY: config.child_execution_mode,
                RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY: config.max_parallel_children,
            })
            return context
        # Framework-owned hidden values are authoritative in every mode.  A
        # caller-provided custom context may add page/business metadata, but it
        # cannot replace the store, Trace identity or per-run MemoryConfig.
        return {**(config.context or {}), **framework_context}

    # 默认 system prompt 前缀（当 config.system_prompt 和前端都未提供 system message 时使用）
    # 注意：此常量已迁移到 cyber_agent.core.prompts，这里保留引用以保持向后兼容

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

        优先级：
        1. base_prompt（来自消息）
        2. config.system_prompt（显式指定）
        3. preset.system_prompt（预设的完整 system prompt）
        4. 默认模板 + skills

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

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

        # 确定 system_prompt 来源
        if base_prompt is not None:
            system_prompt = base_prompt
        elif config.system_prompt is not None:
            system_prompt = config.system_prompt
        else:
            # 尝试从 preset 获取 system_prompt
            preset = AGENT_PRESETS.get(config.agent_type)
            system_prompt = preset.system_prompt if preset and preset.system_prompt else None

        # 确定要加载哪些 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 = DEFAULT_SYSTEM_PREFIX
            if skills_text:
                system_prompt += f"\n\n## Skills\n{skills_text}"

        if config.max_iterations and config.max_iterations > 0:
            system_prompt += f"\n\n## Execution Constraint\n这是一项有严格步数限制的任务。你最多可以用 {config.max_iterations} 轮交互来解决问题。\n请务必【边查边写、随时存档】！每当你收集或得出一个有价值的独立结果（如收集到一个独立 Case），请立刻调用工具写入或追加到结果文件中，绝对不要等到所有任务都做完再最后一次性输出。这样即使触达步数上限被强制打断，你已经收集的成果也能安全保留！"
        # Memory 注入（memory-bearing Agent）——在 system prompt 末尾追加
        # 初版选择 system prompt 追加（见 cyber_agent/docs/framework/runtime/memory.md 待定问题 1）。
        # 好处：run 启动一次性注入、所有后续轮次都能看到、与 skills 注入方式一致。
        # 代价：若记忆文件很大会持续占 prompt tokens —— 待观察后决定是否切换方案。
        if config.memory:
            try:
                from cyber_agent.core.memory import load_memory_files, format_memory_injection
                files = load_memory_files(config.memory)
                memory_text = format_memory_injection(files)
                if memory_text:
                    system_prompt += f"\n\n{memory_text}"
            except Exception as e:
                self.log.warning(f"[Memory] 加载记忆失败，跳过注入: {e}")

        return system_prompt

    @staticmethod
    def _task_text(messages: List[Dict]) -> str:
        """Extract the same plain task text used by Legacy title generation."""
        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", ""))
        return " ".join(text_parts).strip()

    @classmethod
    def _fallback_task_name(cls, messages: List[Dict]) -> str:
        """Build a deterministic title without spending an untracked LLM call."""
        raw_text = cls._task_text(messages)
        if not raw_text:
            return TASK_NAME_FALLBACK
        return raw_text[:50] + ("..." if len(raw_text) > 50 else "")

    async def _generate_task_name(self, messages: List[Dict]) -> str:
        """生成任务名称：优先使用 utility_llm，fallback 到文本截取"""
        fallback = self._fallback_task_name(messages)
        raw_text = self._task_text(messages)

        # 尝试使用 utility_llm 生成标题
        if self.utility_llm_call:
            try:
                result = await self.utility_llm_call(
                    messages=[
                        {"role": "system", "content": TASK_NAME_GENERATION_SYSTEM_PROMPT},
                        {"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 fallback

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