Agent/cyber_agent/trace/run_api.py

"""
Trace 控制 API：新建、续跑、回溯、停止与反思。

路由将请求转为 RunConfig 并在后台驱动 AgentRunner，客户端通过 WebSocket 读取事件。
Recursive 的根完成标准和子树停止也在这一 API 边界传递给实际 Runner。

端点：
  POST /api/traces              — 新建 Trace 并执行
  POST /api/traces/{id}/run     — 运行（统一续跑 + 回溯）
  POST /api/traces/{id}/stop    — 停止运行中的 Trace
  POST /api/traces/{id}/reflect — 反思，在 trace 末尾追加反思 prompt 运行，结果追加到 experiences 文件
  GET  /api/traces/running      — 列出正在运行的 Trace
  GET  /api/experiences         — 读取经验文件内容
"""

import asyncio
import logging
import re
import uuid
import os
from datetime import datetime
from typing import Any, Dict, List, Optional

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field, model_validator

from cyber_agent.core.task_protocol import RootTaskAnchor

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/traces", tags=["run"])

# 经验 API 使用独立 prefix
experiences_router = APIRouter(prefix="/api", tags=["experiences"])


# ===== 全局 Runner（由 api_server.py 注入）=====

_runner = None
_application_runtime = None


def set_runner(runner):
    """注入 AgentRunner 实例"""
    global _runner
    _runner = runner


def set_application_runtime(runtime):
    """Inject the optional experimental application assembly runtime."""
    global _application_runtime
    _application_runtime = runtime


def _get_runner():
    if _runner is None:
        raise HTTPException(
            status_code=503,
            detail="AgentRunner not configured. Server is in read-only mode.",
        )
    return _runner


def _get_application_runtime():
    if _application_runtime is None:
        raise HTTPException(
            status_code=503,
            detail="ApplicationRuntime is not configured",
        )
    return _application_runtime


# ===== Request / Response 模型 =====


class CreateRequest(BaseModel):
    """新建执行"""
    messages: List[Dict[str, Any]] = Field(
        ...,
        description="OpenAI SDK 格式的输入消息。可包含 system + user 消息；若无 system 消息则从 skills 自动构建",
    )
    model: Optional[str] = Field(None, description="模型名称；省略时采用项目/框架默认值")
    temperature: Optional[float] = Field(None)
    max_iterations: Optional[int] = Field(None, gt=0)
    tools: Optional[List[str]] = Field(None, description="工具白名单（None = 全部）")
    name: Optional[str] = Field(None, description="任务名称（None = 自动生成）")
    uid: Optional[str] = Field(None)
    project_name: Optional[str] = Field(None, description="示例项目名称，若提供则动态加载其执行环境")
    application_id: Optional[str] = Field(None, min_length=1)
    application_version: Optional[str] = Field(None, min_length=1)
    root_task_anchor: Optional[RootTaskAnchor] = Field(
        None,
        description="Recursive 根任务必须显式提供的不可变目标、完成标准和硬约束；Legacy 忽略",
    )

    @model_validator(mode="before")
    @classmethod
    def reject_removed_root_criteria(cls, value):
        if isinstance(value, dict) and "root_completion_criteria" in value:
            raise ValueError(
                "root_completion_criteria has been removed; use root_task_anchor"
            )
        return value

    @model_validator(mode="after")
    def validate_application_selection(self):
        has_id = self.application_id is not None
        has_version = self.application_version is not None
        if has_id != has_version:
            raise ValueError(
                "application_id and application_version must be provided together"
            )
        if has_id and self.project_name is not None:
            raise ValueError("application and project_name are mutually exclusive")
        if has_id and any(
            value is not None
            for value in (self.model, self.temperature, self.tools)
        ):
            raise ValueError(
                "application runs cannot override model, temperature, or tools"
            )
        if has_id and any(message.get("role") == "system" for message in self.messages):
            raise ValueError(
                "application runs cannot provide caller-controlled system messages"
            )
        return self


class TraceRunRequest(BaseModel):
    """运行（统一续跑 + 回溯）"""
    messages: List[Dict[str, Any]] = Field(
        default_factory=list,
        description="追加的新消息（可为空，用于重新生成场景）",
    )
    after_message_id: Optional[str] = Field(
        None,
        description="从哪条消息后续跑。None = 从末尾续跑，message_id = 从该消息后运行（自动判断续跑/回溯）",
    )


class ReflectRequest(BaseModel):
    """反思请求"""
    focus: Optional[str] = Field(None, description="反思重点（可选）")


class RunResponse(BaseModel):
    """操作响应（立即返回，后台执行）"""
    trace_id: str
    status: str = "started"
    message: str = ""


class StopResponse(BaseModel):
    """停止响应"""
    trace_id: str
    status: str  # "stopping" | "not_running"


class ReflectResponse(BaseModel):
    """反思响应"""
    trace_id: str
    reflection: str


class CompactResponse(BaseModel):
    """压缩响应"""
    trace_id: str
    previous_count: int
    new_count: int
    message: str = ""


class ToolApprovalDecision(BaseModel):
    tool_call_id: str = Field(min_length=1)
    decision: str = Field(description="approve / reject")
    edited_arguments: Optional[Dict[str, Any]] = None


class DecideToolApprovalRequest(BaseModel):
    decisions: List[ToolApprovalDecision] = Field(min_length=1)


# ===== 提取审核（见 cyber_agent/docs/memory.md 第三节） =====

class PendingExtractionModel(BaseModel):
    extraction_id: str
    sequence: Optional[int] = None
    goal_id: Optional[str] = None
    branch_id: Optional[str] = None
    payload: Dict[str, Any]
    reviewed: bool = False
    decision: Optional[str] = None
    committed: bool = False


class ListExtractionsResponse(BaseModel):
    trace_id: str
    count: int
    items: List[PendingExtractionModel]


class ReviewRequest(BaseModel):
    decision: str = Field(..., description="approve / edit / discard")
    edited_payload: Optional[Dict[str, Any]] = Field(
        None, description="decision=edit 时必填；只对本次 review 生效"
    )


class ReviewResponse(BaseModel):
    trace_id: str
    extraction_id: str
    decision: str


class CommitResponse(BaseModel):
    trace_id: str
    committed_count: int
    failed_count: int
    skipped_count: int
    committed: List[str]
    knowledge_ids: List[str]
    failed: List[Dict[str, str]]
    skipped: List[str]


# ===== 后台执行 =====

_running_tasks: Dict[str, asyncio.Task] = {}
# 记住每个运行中 Trace 真正使用的 Runner，避免 Example 专属 Runner 的续跑/停止误落到全局 Runner。
# 映射只覆盖当前进程活跃任务，由两个后台执行入口按对象身份清理。
_running_runners: Dict[str, Any] = {}
_approval_locks: Dict[str, asyncio.Lock] = {}


def _require_mutable_trace(trace) -> None:
    """将核心协议的只读门禁统一映射为 HTTP 409。"""
    from cyber_agent.core.agent_mode import (
        RECURSIVE_REVISION_READ_ONLY_ERROR,
        require_mutable_trace_policy,
    )

    try:
        require_mutable_trace_policy(trace.context)
    except ValueError as exc:
        if str(exc) == RECURSIVE_REVISION_READ_ONLY_ERROR:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        raise


async def _cancel_and_wait_for_running_task(trace_id: str) -> None:
    """等待旧后台运行完成 finally 清理，再允许同一 Trace 续跑。"""
    old_task = _running_tasks.get(trace_id)
    if not old_task:
        return
    old_task.cancel()
    try:
        await old_task
    except asyncio.CancelledError:
        pass
    if _running_tasks.get(trace_id) is old_task:
        _running_tasks.pop(trace_id, None)


async def _restore_runner_and_config(
    *,
    trace,
    base_runner,
    after_sequence: Optional[int] = None,
    force_side_branch: Optional[List[str]] = None,
    approval_batch_id: Optional[str] = None,
):
    """Restore the persisted run contract and its trusted project environment."""
    import importlib
    from cyber_agent.core.agent_mode import (
        AgentMode,
        RECURSIVE_REVISION_READ_ONLY_ERROR,
        policy_from_context,
        require_mutable_trace_policy,
    )
    from cyber_agent.core.run_snapshot import (
        RUN_CONFIG_SNAPSHOT_CONTEXT_KEY,
        RunConfigSnapshotV1,
        RunConfigSnapshotV2,
        RunConfigSnapshotError,
        load_run_config_snapshot,
        persist_run_config_snapshot,
    )
    from cyber_agent.core.runner import RunConfig

    try:
        require_mutable_trace_policy(trace.context)
    except ValueError as exc:
        if str(exc) == RECURSIVE_REVISION_READ_ONLY_ERROR:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        raise

    runner = base_runner
    config = RunConfig(
        trace_id=trace.trace_id,
        after_sequence=after_sequence,
        force_side_branch=force_side_branch,
        approval_batch_id=approval_batch_id,
    )
    project_name = None
    if RUN_CONFIG_SNAPSHOT_CONTEXT_KEY in (trace.context or {}):
        try:
            snapshot = load_run_config_snapshot(trace.context)
        except RunConfigSnapshotError as exc:
            raise HTTPException(status_code=409, detail=str(exc)) from exc
        if isinstance(snapshot, RunConfigSnapshotV2):
            try:
                runner, config = await _get_application_runtime().restore(
                    trace.trace_id
                )
            except HTTPException:
                raise
            except Exception as exc:
                raise HTTPException(
                    status_code=409,
                    detail=f"Failed to restore application binding: {exc}",
                ) from exc
        else:
            config.apply_snapshot(snapshot)
        # apply_snapshot intentionally leaves invocation-only controls untouched.
        config.trace_id = trace.trace_id
        config.after_sequence = after_sequence
        config.force_side_branch = force_side_branch
        config.approval_batch_id = approval_batch_id
        project_name = snapshot.project_name
    elif policy_from_context(trace.context).mode is AgentMode.RECURSIVE:
        raise HTTPException(
            status_code=409,
            detail="This Recursive trace predates RunConfig snapshots; create a new trace",
        )
    else:
        # Infer and persist the Legacy contract before selecting its Runner.
        # Otherwise an old project Trace would resume once on the global
        # Runner and only learn its project_name too late inside AgentRunner.
        inferred = RunConfig(
            model=trace.model or "gpt-4o",
            temperature=float((trace.llm_params or {}).get("temperature", 0.3)),
            tools=[
                item.get("function", {}).get("name")
                for item in (trace.tools or [])
                if item.get("function", {}).get("name")
            ],
            tool_groups=None,
            agent_type=trace.agent_type or "default",
            uid=trace.uid,
            extra_llm_params={
                key: value
                for key, value in (trace.llm_params or {}).items()
                if key != "temperature"
            },
            context=(
                {"project_name": trace.context.get("project_name")}
                if trace.context.get("project_name")
                else {}
            ),
        )
        snapshot = RunConfigSnapshotV1.from_run_config(
            inferred,
            memory_identity=None,
            legacy_inferred=True,
        )
        persist_run_config_snapshot(trace.context, snapshot)
        if not base_runner.trace_store:
            raise HTTPException(status_code=503, detail="TraceStore not configured")
        await base_runner.trace_store.update_trace(
            trace.trace_id,
            context=trace.context,
        )
        config.apply_snapshot(snapshot)
        config.trace_id = trace.trace_id
        config.after_sequence = after_sequence
        config.force_side_branch = force_side_branch
        config.approval_batch_id = approval_batch_id
        project_name = snapshot.project_name

    if project_name:
        if not re.fullmatch(r"[A-Za-z0-9_]+", project_name):
            raise HTTPException(status_code=409, detail="Invalid persisted project_name")
        module_name = f"examples.{project_name}.run"
        try:
            example_module = importlib.import_module(module_name)
            if hasattr(example_module, "init_project_env"):
                project_runner, _project_msgs, _default_config = (
                    await example_module.init_project_env()
                )
                runner = project_runner
        except ImportError as exc:
            if getattr(exc, "name", None) == module_name:
                logger.warning(
                    "Project %s has no custom run.py; using the default runner",
                    project_name,
                )
            else:
                raise HTTPException(
                    status_code=409,
                    detail=f"Failed to restore project environment: {project_name}",
                ) from exc
        except Exception as exc:
            logger.exception("Failed to restore project %s", project_name)
            raise HTTPException(
                status_code=409,
                detail=f"Failed to restore project environment: {project_name}",
            ) from exc
    if runner.trace_store and hasattr(
        runner.trace_store,
        "get_tool_approval_batch",
    ):
        batch = await runner.trace_store.get_tool_approval_batch(trace.trace_id)
        if (
            batch is not None
            and batch.status == "executing"
            and await runner._recover_candidate_approval_if_durable(
                trace.trace_id,
                batch,
            )
        ):
            batch = await runner.trace_store.get_tool_approval_batch(trace.trace_id)
        if batch is not None and batch.status == "recoverable_candidate_command":
            if after_sequence is not None or force_side_branch:
                raise HTTPException(
                    status_code=409,
                    detail="recoverable_candidate_command_must_resume_first",
                )
            if approval_batch_id is None:
                config.approval_batch_id = batch.batch_id
    return runner, config


async def _run_in_background(trace_id: str, messages: List[Dict], config, runner_instance=None):
    """后台执行已知 Trace ID 的 Agent，消费 run() 的所有 yield。

    run_trace() 续跑/回溯时调用，并登记实际 Runner 供 stop_trace() 命中 Recursive 子树。"""
    runner = runner_instance or _get_runner()
    current_task = asyncio.current_task()
    if current_task:
        _running_tasks[trace_id] = current_task
    _running_runners[trace_id] = runner
    try:
        async for _item in runner.run(messages=messages, config=config):
            pass  # WebSocket 广播由 runner 内部的 store 事件驱动
    except Exception as e:
        logger.error(f"Background run failed for {trace_id}: {e}")
    finally:
        if _running_tasks.get(trace_id) is current_task:
            _running_tasks.pop(trace_id, None)
        if _running_runners.get(trace_id) is runner:
            _running_runners.pop(trace_id, None)


async def _run_with_trace_signal(
    messages: List[Dict], config, trace_id_future: asyncio.Future, runner_instance=None
):
    """后台新建 Agent，通过 Future 将首个 Trace 对象的 ID 传回 API。

    create_and_run() 调用它，同时登记实际 Runner，使后续停止不会丢失 Example 专属运行环境。"""
    from cyber_agent.trace.models import Trace

    runner = runner_instance or _get_runner()
    trace_id: Optional[str] = None
    current_task = asyncio.current_task()
    try:
        async for item in runner.run(messages=messages, config=config):
            if isinstance(item, Trace) and not trace_id_future.done():
                trace_id = item.trace_id
                if current_task:
                    _running_tasks[trace_id] = current_task
                _running_runners[trace_id] = runner
                trace_id_future.set_result(trace_id)
    except Exception as e:
        if not trace_id_future.done():
            trace_id_future.set_exception(e)
        logger.error(f"Background run failed: {e}")
    finally:
        if trace_id:
            if _running_tasks.get(trace_id) is current_task:
                _running_tasks.pop(trace_id, None)
            if _running_runners.get(trace_id) is runner:
                _running_runners.pop(trace_id, None)


# ===== 路由 =====


@router.post("", response_model=RunResponse)
async def create_and_run(req: CreateRequest):
    """
    新建 Trace 并开始执行，将 HTTP 参数和 Example 默认值合并为 RunConfig。

    Recursive 的 root_task_anchor 从此传入 Runner 完成门禁；获取 trace_id 后立即返回，执行继续留在后台。
    """
    import importlib
    from dataclasses import replace
    from cyber_agent.core.runner import RunConfig

    runner = None
    config = None
    messages = req.messages

    if req.application_id is not None:
        try:
            runner, config = _get_application_runtime().new_run(
                req.application_id,
                req.application_version,
                uid=req.uid,
                name=req.name,
                root_task_anchor=req.root_task_anchor,
                max_iterations=req.max_iterations,
            )
        except HTTPException:
            raise
        except ValueError as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc

    if req.project_name:
        try:
            # 动态加载对应 example 的 run.py
            module_name = f"examples.{req.project_name}.run"
            example_module = importlib.import_module(module_name)
            if hasattr(example_module, "init_project_env"):
                # 获取该 example 专属的 runner, 带上下文 messages, 以及默认 config
                runner, example_messages, default_config = await example_module.init_project_env(req.messages)
                messages = example_messages
                
                # Preserve every project default, then apply only fields explicitly
                # supplied by the request before the Runner snapshots the result.
                config = replace(default_config)
                if req.model is not None:
                    config.model = req.model
                if req.temperature is not None:
                    config.temperature = req.temperature
                if req.max_iterations is not None:
                    config.max_iterations = req.max_iterations
                if req.tools is not None:
                    config.tools = list(req.tools)
                if req.name is not None:
                    config.name = req.name
                if req.uid is not None:
                    config.uid = req.uid
                config.root_task_anchor = req.root_task_anchor
                config.context = {
                    **(default_config.context or {}),
                    "project_name": req.project_name,
                }
        except ImportError as e:
            if getattr(e, "name", None) == module_name:
                logger.warning(f"Project '{req.project_name}' has no custom run.py, falling back to default.")
            else:
                import traceback
                logger.error(f"Error INSIDE {module_name}:\n{traceback.format_exc()}")
        except Exception as e:
            import traceback
            logger.error(f"Unexpected error loading project environment for {req.project_name}:\n{traceback.format_exc()}")
            
    if not runner:
        _get_runner()  # 验证全局默认 Runner 已配置
        config = RunConfig(
            model=req.model or "gpt-4o",
            temperature=req.temperature if req.temperature is not None else 0.3,
            max_iterations=req.max_iterations or 200,
            tools=req.tools,
            name=req.name,
            uid=req.uid,
            root_task_anchor=req.root_task_anchor,
            context={"project_name": req.project_name} if req.project_name else {}
        )

    # 启动后台执行，通过 Future 等待 trace_id（Phase 1 完成后即返回）
    trace_id_future: asyncio.Future[str] = asyncio.get_running_loop().create_future()
    task = asyncio.create_task(
        _run_with_trace_signal(messages, config, trace_id_future, runner_instance=runner)
    )

    trace_id = await trace_id_future
    return RunResponse(
        trace_id=trace_id,
        status="started",
        message=f"Execution started. Watch via WebSocket: /api/traces/{trace_id}/watch",
    )


async def _cleanup_incomplete_tool_calls(store, trace_id: str, after_sequence: int) -> int:
    """
    找到安全的插入点，保证不会把新消息插在一个不完整的工具调用序列中间。

    场景：
    1. after_sequence 刚好是一条带 tool_calls 的 assistant 消息，
       但其部分/全部 tool response 还没生成 → 回退到该 assistant 之前。
    2. after_sequence 是某条 tool response，但同一批 tool_calls 中
       还有其他 response 未生成 → 回退到该 assistant 之前。

    核心逻辑：从 after_sequence 往前找，定位到包含它的那条 assistant 消息，
    检查该 assistant 的所有 tool_calls 是否都有对应的 tool response。
    如果不完整，就把截断点回退到该 assistant 消息之前（即其 parent_sequence）。

    Args:
        store: TraceStore
        trace_id: Trace ID
        after_sequence: 用户指定的插入位置

    Returns:
        调整后的安全截断点（<= after_sequence）
    """
    all_messages = await store.get_trace_messages(trace_id)
    if not all_messages:
        return after_sequence

    by_seq = {msg.sequence: msg for msg in all_messages}

    target = by_seq.get(after_sequence)
    if target is None:
        return after_sequence

    # 找到"所属的 assistant 消息"：
    # - 如果 target 本身是 assistant → 就是它
    # - 如果 target 是 tool → 沿 parent_sequence 往上找 assistant
    assistant_msg = None
    if target.role == "assistant":
        assistant_msg = target
    elif target.role == "tool":
        cur = target
        while cur and cur.role == "tool":
            parent_seq = cur.parent_sequence
            cur = by_seq.get(parent_seq) if parent_seq is not None else None
        if cur and cur.role == "assistant":
            assistant_msg = cur

    if assistant_msg is None:
        return after_sequence

    # 该 assistant 是否带 tool_calls？
    content = assistant_msg.content
    if not isinstance(content, dict) or not content.get("tool_calls"):
        return after_sequence

    # 收集所有 tool_call_ids
    expected_ids = set()
    for tc in content["tool_calls"]:
        if isinstance(tc, dict) and tc.get("id"):
            expected_ids.add(tc["id"])

    if not expected_ids:
        return after_sequence

    # 查找已有的 tool responses
    found_ids = set()
    for msg in all_messages:
        if msg.role == "tool" and msg.tool_call_id in expected_ids:
            found_ids.add(msg.tool_call_id)

    missing = expected_ids - found_ids
    if not missing:
        # 全部 tool response 都在，这是一个完整的序列
        return after_sequence

    # 不完整 → 回退到 assistant 之前
    safe = assistant_msg.parent_sequence
    if safe is None:
        # assistant 已经是第一条消息，没有更早的位置
        safe = assistant_msg.sequence - 1

    logger.info(
        "检测到不完整的工具调用 (assistant seq=%d, 缺少 %d/%d tool responses)，"
        "自动回退插入点：%d -> %d",
        assistant_msg.sequence, len(missing), len(expected_ids),
        after_sequence, safe,
    )
    return safe


def _parse_sequence_from_message_id(message_id: str) -> int:
    """从 message_id 末尾解析 sequence 整数（格式：{trace_id}-{sequence:04d}）"""
    try:
        return int(message_id.rsplit("-", 1)[-1])
    except (ValueError, IndexError):
        raise HTTPException(
            status_code=422,
            detail=f"Invalid after_message_id format: {message_id!r}",
        )


@router.post("/{trace_id}/run", response_model=RunResponse)
async def run_trace(trace_id: str, req: TraceRunRequest):
    """
    运行已有 Trace，统一处理续跑与回溯。

    它优先取活跃 Trace 实际 Runner，再由 AgentRunner 恢复持久化的 Legacy/Recursive 模式和协议状态。

    - after_message_id 为 null（或省略）：从末尾续跑
    - after_message_id 为 message_id 字符串：从该消息后运行（Runner 自动判断续跑/回溯）
    - messages 为空 + after_message_id 有值：重新生成（从该位置重跑，不插入新消息）

    **自动清理不完整工具调用**：
    如果人工插入 message 的位置打断了一个工具调用过程（assistant 消息有 tool_calls
    但缺少对应的 tool responses），框架会自动检测并调整插入位置，确保不会产生不一致的状态。
    """
    runner = _running_runners.get(trace_id) or _get_runner()

    # 将 message_id 转换为内部使用的 sequence 整数
    after_sequence: Optional[int] = None
    if req.after_message_id is not None:
        after_sequence = _parse_sequence_from_message_id(req.after_message_id)

    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")

    # Restore the immutable run contract and project environment before any
    # status/rewind decision. Status checks must use the same Runner that will
    # actually resume the Trace.
    runner, config = await _restore_runner_and_config(
        trace=trace,
        base_runner=runner,
        after_sequence=after_sequence,
    )
    approval_batch = (
        await runner.trace_store.get_tool_approval_batch(trace_id)
        if runner.trace_store
        and hasattr(runner.trace_store, "get_tool_approval_batch")
        else None
    )
    if (
        approval_batch is not None
        and approval_batch.status == "recoverable_candidate_command"
        and req.messages
    ):
        raise HTTPException(
            status_code=409,
            detail="recoverable_candidate_command_must_resume_first",
        )

    # 验证 trace 状态
    if runner.trace_store:
        if trace.status == "waiting_confirmation":
            raise HTTPException(
                status_code=409,
                detail="Trace is waiting for tool approval; use the tool-approvals endpoint",
            )

        # 自动检查并清理不完整的工具调用
        if after_sequence is not None and req.messages:
            adjusted_seq = await _cleanup_incomplete_tool_calls(
                runner.trace_store, trace_id, after_sequence
            )
            if adjusted_seq != after_sequence:
                logger.info(
                    f"已自动调整插入位置：{after_sequence} -> {adjusted_seq}"
                )
                after_sequence = adjusted_seq

    # 检查是否已在运行
    if trace_id in _running_tasks and not _running_tasks[trace_id].done():
        # 竞态窗口修复：task 还没退出，但 store 里 trace 已经是 stopped/failed
        # 这发生在 cancel_event 触发后 → store 更新 → task finally 块还未执行之间
        # 此时可以安全地强制清除旧 task，允许续跑
        store_trace = None
        if runner.trace_store:
            store_trace = await runner.trace_store.get_trace(trace_id)
        if store_trace and store_trace.status in ("stopped", "failed", "completed"):
            logger.info(
                f"run_trace: task for {trace_id} not done yet but store status={store_trace.status!r}, "
                "forcing cleanup to allow resume"
            )
            await _cancel_and_wait_for_running_task(trace_id)
        else:
            raise HTTPException(status_code=409, detail="Trace is already running")

    # 恢复运行时，将状态从 stopped 改回 running，并广播状态变化
    if runner.trace_store and trace_id:
        current_trace = await runner.trace_store.get_trace(trace_id)
        if current_trace and current_trace.status == "stopped":
            await runner.trace_store.update_trace(trace_id, status="running")
            # 广播状态变化给前端
            from cyber_agent.trace.websocket import broadcast_trace_status_changed
            await broadcast_trace_status_changed(trace_id, "running")

    task = asyncio.create_task(_run_in_background(trace_id, req.messages, config, runner_instance=runner))
    _running_tasks[trace_id] = task

    mode = "rewind" if after_sequence is not None else "continue"
    return RunResponse(
        trace_id=trace_id,
        status="started",
        message=f"Run ({mode}) started. Watch via WebSocket: /api/traces/{trace_id}/watch",
    )


@router.get("/{trace_id}/tool-approvals/pending")
async def get_pending_tool_approval(trace_id: str):
    runner = _running_runners.get(trace_id) or _get_runner()
    if not runner.trace_store or not hasattr(
        runner.trace_store,
        "get_tool_approval_batch",
    ):
        raise HTTPException(status_code=503, detail="Tool approval store unavailable")
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
    batch = await runner.trace_store.get_tool_approval_batch(trace_id)
    if batch is None or batch.status != "pending":
        return {"trace_id": trace_id, "batch": None}
    return {"trace_id": trace_id, "batch": batch.model_dump(mode="json")}


@router.post("/{trace_id}/tool-approvals/{batch_id}", response_model=RunResponse)
async def decide_tool_approval(
    trace_id: str,
    batch_id: str,
    req: DecideToolApprovalRequest,
):
    from cyber_agent.tools.approval import tool_argument_hash

    runner = _running_runners.get(trace_id) or _get_runner()
    if not runner.trace_store or not hasattr(
        runner.trace_store,
        "get_tool_approval_batch",
    ):
        raise HTTPException(status_code=503, detail="Tool approval store unavailable")
    lock = _approval_locks.setdefault(trace_id, asyncio.Lock())
    async with lock:
        trace = await runner.trace_store.get_trace(trace_id)
        if not trace:
            raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
        runner, config = await _restore_runner_and_config(
            trace=trace,
            base_runner=runner,
            approval_batch_id=batch_id,
        )
        if not runner.trace_store:
            raise HTTPException(status_code=503, detail="TraceStore not configured")
        trace = await runner.trace_store.get_trace(trace_id)
        if not trace:
            raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
        batch = await runner.trace_store.get_tool_approval_batch(trace_id)
        if batch is None or batch.batch_id != batch_id:
            raise HTTPException(status_code=404, detail="Tool approval batch not found")
        if batch.status != "pending" or trace.status != "waiting_confirmation":
            raise HTTPException(status_code=409, detail="Tool approval was already decided")
        if trace_id in _running_tasks and not _running_tasks[trace_id].done():
            raise HTTPException(status_code=409, detail="Trace is already running")

        decisions = {item.tool_call_id: item for item in req.decisions}
        if len(decisions) != len(req.decisions):
            raise HTTPException(status_code=422, detail="Duplicate tool_call_id decision")
        expected_ids = {call.tool_call_id for call in batch.pending_calls}
        if set(decisions) != expected_ids:
            raise HTTPException(
                status_code=422,
                detail=(
                    "Decisions must cover every pending tool call exactly once; "
                    f"expected={sorted(expected_ids)}"
                ),
            )

        for call in batch.pending_calls:
            decision = decisions[call.tool_call_id]
            if decision.decision not in {"approve", "reject"}:
                raise HTTPException(
                    status_code=422,
                    detail="decision must be approve or reject",
                )
            if decision.decision == "reject":
                if decision.edited_arguments is not None:
                    raise HTTPException(
                        status_code=422,
                        detail="rejected calls cannot edit arguments",
                    )
                call.decision = "rejected"
            else:
                edited = (
                    dict(decision.edited_arguments)
                    if decision.edited_arguments is not None
                    else dict(call.original_arguments)
                )
                changed = {
                    key
                    for key in set(call.original_arguments) | set(edited)
                    if call.original_arguments.get(key) != edited.get(key)
                    or (key in call.original_arguments) != (key in edited)
                }
                forbidden = changed - set(call.editable_params)
                if forbidden:
                    raise HTTPException(
                        status_code=422,
                        detail=f"Arguments are not editable: {sorted(forbidden)}",
                    )
                call.effective_arguments = edited
                call.argument_hash = tool_argument_hash(
                    tool_call_id=call.tool_call_id,
                    tool_name=call.tool_name,
                    arguments=edited,
                )
                call.decision = "approved"
            call.decided_at = datetime.now().isoformat()

        batch.status = "decided"
        batch.updated_at = datetime.now().isoformat()
        await runner.trace_store.replace_tool_approval_batch(trace_id, batch)
        await runner.trace_store.update_trace(trace_id, status="running")
        task = asyncio.create_task(
            _run_in_background(
                trace_id,
                [],
                config,
                runner_instance=runner,
            )
        )
        _running_tasks[trace_id] = task
        return RunResponse(
            trace_id=trace_id,
            status="started",
            message="Approved tool batch resume started",
        )


@router.post("/{trace_id}/stop", response_model=StopResponse)
async def stop_trace(trace_id: str):
    """
    向运行中 Trace 的实际 Runner 发送协作式停止信号。

    Legacy 只停当前 Trace；Recursive 由 AgentRunner.stop() 向当前进程的活跃子孙传播。
    """
    runner = _running_runners.get(trace_id) or _get_runner()

    if runner.trace_store:
        trace = await runner.trace_store.get_trace(trace_id)
        if trace:
            runner, _config = await _restore_runner_and_config(
                trace=trace,
                base_runner=runner,
            )
            batch = (
                await runner.trace_store.get_tool_approval_batch(trace_id)
                if hasattr(runner.trace_store, "get_tool_approval_batch")
                else None
            )
            if (
                batch is not None
                and batch.status == "recoverable_candidate_command"
            ):
                raise HTTPException(
                    status_code=409,
                    detail="recoverable_candidate_command_must_resume_first",
                )
        if trace and trace.status == "waiting_confirmation":
            batch = (
                await runner.trace_store.get_tool_approval_batch(trace_id)
                if hasattr(runner.trace_store, "get_tool_approval_batch")
                else None
            )
            if batch and batch.status == "pending":
                batch.status = "cancelled"
                batch.updated_at = datetime.now().isoformat()
                await runner.trace_store.replace_tool_approval_batch(trace_id, batch)
            await runner.trace_store.update_trace(trace_id, status="stopped")
            try:
                from cyber_agent.trace.websocket import broadcast_trace_status_changed
                await broadcast_trace_status_changed(trace_id, "stopped")
            except Exception:
                pass
            return StopResponse(trace_id=trace_id, status="stopping")

    # 通过 runner 的 stop 方法设置取消信号
    stopped = await runner.stop(trace_id)

    if not stopped:
        # 检查是否在 _running_tasks 但 runner 不知道（可能已完成）
        if trace_id in _running_tasks:
            task = _running_tasks[trace_id]
            if not task.done():
                task.cancel()
                _running_tasks.pop(trace_id, None)
                return StopResponse(trace_id=trace_id, status="stopping")
        return StopResponse(trace_id=trace_id, status="not_running")

    return StopResponse(trace_id=trace_id, status="stopping")


@router.post("/{trace_id}/reflect", response_model=ReflectResponse)
async def reflect_trace(trace_id: str, req: ReflectRequest):
    """
    触发反思

    通过 force_side_branch="reflection" 触发侧分支多轮 agent 模式，
    LLM 可以调用工具（如 knowledge_search, knowledge_save）进行多轮推理。
    反思消息标记为侧分支（branch_type="reflection"），不在主路径上。
    """
    runner = _get_runner()

    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")

    # 验证 trace 存在
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")

    runner, config = await _restore_runner_and_config(
        trace=trace,
        base_runner=runner,
        force_side_branch=["reflection"],
    )

    # 检查是否仍在运行
    if trace_id in _running_tasks and not _running_tasks[trace_id].done():
        raise HTTPException(status_code=409, detail="Cannot reflect on a running trace. Stop it first.")

    # 如果有 focus，可以通过追加消息传递（可选）
    messages = []
    if req.focus:
        messages = [{"role": "user", "content": f"反思重点：{req.focus}"}]

    # 启动反思任务（后台执行）
    task = asyncio.create_task(_run_trace_background(runner, messages, config))
    _running_tasks[trace_id] = task

    return ReflectResponse(
        trace_id=trace_id,
        reflection="反思任务已启动，通过 WebSocket 监听实时更新",
    )


@router.get("/{trace_id}/extractions", response_model=ListExtractionsResponse)
async def list_extractions(trace_id: str, include_reviewed: bool = False):
    """列出 trace 的待审核提取条目。"""
    runner = _get_runner()
    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")

    from cyber_agent.trace.extraction_review import list_pending
    pendings = await list_pending(
        runner.trace_store, trace_id, include_reviewed=include_reviewed
    )
    return ListExtractionsResponse(
        trace_id=trace_id,
        count=len(pendings),
        items=[
            PendingExtractionModel(
                extraction_id=p.extraction_id,
                sequence=p.sequence,
                goal_id=p.goal_id,
                branch_id=p.branch_id,
                payload=p.payload,
                reviewed=p.reviewed,
                decision=p.decision,
                committed=p.committed,
            )
            for p in pendings
        ],
    )


@router.post(
    "/{trace_id}/extractions/{extraction_id}/review",
    response_model=ReviewResponse,
)
async def review_extraction(trace_id: str, extraction_id: str, req: ReviewRequest):
    """对单条 pending 提交 review 决策（approve/edit/discard）。"""
    runner = _get_runner()
    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
    _require_mutable_trace(trace)

    if req.decision not in ("approve", "edit", "discard"):
        raise HTTPException(
            status_code=400,
            detail=f"decision must be approve/edit/discard, got {req.decision}",
        )
    if req.decision == "edit" and not req.edited_payload:
        raise HTTPException(
            status_code=400, detail="decision=edit 必须提供 edited_payload"
        )

    from cyber_agent.trace.extraction_review import review_one
    await review_one(
        runner.trace_store,
        trace_id,
        extraction_id,
        req.decision,  # type: ignore[arg-type]
        edited_payload=req.edited_payload,
    )
    return ReviewResponse(
        trace_id=trace_id, extraction_id=extraction_id, decision=req.decision
    )


@router.post("/{trace_id}/extractions/commit", response_model=CommitResponse)
async def commit_extractions(trace_id: str):
    """批量把已 approved/edited 的条目上传到 KnowHub。"""
    runner = _get_runner()
    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
    _require_mutable_trace(trace)

    from cyber_agent.trace.extraction_review import commit_approved
    report = await commit_approved(runner.trace_store, trace_id)
    return CommitResponse(
        trace_id=trace_id,
        committed_count=len(report.committed),
        failed_count=len(report.failed),
        skipped_count=len(report.skipped),
        committed=report.committed,
        knowledge_ids=report.knowledge_ids,
        failed=report.failed,
        skipped=report.skipped,
    )


@router.post("/{trace_id}/compact", response_model=CompactResponse)
async def compact_trace(trace_id: str):
    """
    压缩 Trace 的上下文 (Compact)

    通过 force_side_branch="compression" 触发侧分支多轮 agent 模式，
    LLM 可以调用工具（如 goal）进行多轮推理。
    压缩消息标记为侧分支（branch_type="compression"），不在主路径上。
    """
    runner = _get_runner()
    if not runner.trace_store:
        raise HTTPException(status_code=503, detail="TraceStore not configured")

    # 验证 trace 存在
    trace = await runner.trace_store.get_trace(trace_id)
    if not trace:
        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")

    runner, config = await _restore_runner_and_config(
        trace=trace,
        base_runner=runner,
        force_side_branch=["compression"],
    )

    # 检查是否仍在运行
    if trace_id in _running_tasks and not _running_tasks[trace_id].done():
        raise HTTPException(status_code=409, detail="Cannot compact a running trace. Stop it first.")

    # 启动压缩任务（后台执行）
    task = asyncio.create_task(_run_trace_background(runner, [], config))
    _running_tasks[trace_id] = task

    return CompactResponse(
        trace_id=trace_id,
        previous_count=0,  # 无法立即获取，需通过 WebSocket 监听
        new_count=0,
        message="压缩任务已启动，通过 WebSocket 监听实时更新",
    )


@router.get("/running", tags=["run"])
async def list_running():
    """列出正在运行的 Trace（包含活跃状态判断）"""
    from datetime import datetime, timedelta

    runner = _get_runner()
    running = []

    for tid, task in list(_running_tasks.items()):
        if task.done():
            _running_tasks.pop(tid, None)
        else:
            # 获取trace详情，检查最后活动时间
            trace_info = {"trace_id": tid, "is_active": True}

            if runner.trace_store:
                try:
                    trace = await runner.trace_store.get_trace(tid)
                    if trace:
                        # 判断是否真正活跃：最后活动时间在30秒内
                        if hasattr(trace, 'last_activity_at') and trace.last_activity_at:
                            time_since_activity = (datetime.now() - trace.last_activity_at).total_seconds()
                            trace_info["is_active"] = time_since_activity < 30
                            trace_info["seconds_since_activity"] = int(time_since_activity)
                        trace_info["status"] = trace.status
                except Exception:
                    pass

            running.append(trace_info)

    return {"running": running}


async def reconcile_traces():
    """
    状态对齐：启动时恢复审批门禁，再清理其他残留的 running 状态。

    审批文件是这个局部状态机的持久化真相：pending 恢复等待；
    decided 且尚未 executing 时安全恢复执行；executing 则失败关闭。
    """
    runner = _get_runner()
    if not runner or not runner.trace_store:
        logger.warning("[Reconciliation] Runner or TraceStore not initialized, skipping.")
        return

    try:
        traces = await runner.trace_store.list_traces(limit=10_000)
        if not traces:
            return

        count = 0
        for trace in traces:
            tid = trace.trace_id
            if tid in _running_tasks and not _running_tasks[tid].done():
                continue
            # Recursive revision 2 是历史只读记录，启动对账不得修改
            # 其审批文件或 Trace.status。
            from cyber_agent.core.agent_mode import policy_from_context
            if policy_from_context(trace.context).is_read_only:
                continue
            trace_runner = runner
            batch = (
                await runner.trace_store.get_tool_approval_batch(tid)
                if hasattr(runner.trace_store, "get_tool_approval_batch")
                else None
            )
            raw_snapshot = (trace.context or {}).get("run_config_snapshot")
            is_application_trace = (
                isinstance(raw_snapshot, dict)
                and raw_snapshot.get("schema_version") == 2
            )
            needs_recovery_action = bool(
                trace.status == "running"
                or (
                    batch is not None
                    and batch.status in {
                        "pending",
                        "decided",
                        "executing",
                        "recoverable_candidate_command",
                    }
                )
            )
            if is_application_trace and needs_recovery_action:
                try:
                    trace_runner, _restored_config = (
                        await _restore_runner_and_config(
                            trace=trace,
                            base_runner=runner,
                        )
                    )
                except Exception as exc:
                    logger.error(
                        "[Reconciliation] Application binding recovery failed for %s: %s",
                        tid,
                        exc,
                    )
                    await runner.trace_store.update_trace(
                        tid,
                        status="failed",
                        error_message=f"Application binding recovery failed: {exc}",
                    )
                    count += 1
                    continue
            if batch and batch.status == "pending" and trace.status in {
                "running", "stopped", "waiting_confirmation",
            }:
                if trace.status != "waiting_confirmation":
                    logger.info(
                        "[Reconciliation] Restoring pending approval %s: %s -> waiting_confirmation",
                        tid,
                        trace.status,
                    )
                    await trace_runner.trace_store.update_trace(
                        tid,
                        status="waiting_confirmation",
                        error_message=None,
                        completed_at=None,
                    )
                    count += 1
                continue
            if batch and batch.status == "executing":
                if await trace_runner._recover_candidate_approval_if_durable(
                    tid,
                    batch,
                ):
                    batch = await trace_runner.trace_store.get_tool_approval_batch(
                        tid
                    )
                else:
                    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 trace_runner.trace_store.replace_tool_approval_batch(
                        tid,
                        batch,
                    )
                    await trace_runner.trace_store.update_trace(
                        tid,
                        status="failed",
                        error_message=(
                            "Tool execution outcome is unknown after process restart; "
                            "automatic retry was refused"
                        ),
                    )
                    count += 1
                    continue
            if batch and batch.status in {
                "decided",
                "recoverable_candidate_command",
            } and trace.status in {
                "running", "stopped", "waiting_confirmation", "failed",
            }:
                logger.info(
                    "[Reconciliation] Resuming safely-decided approval batch for %s",
                    tid,
                )
                restored_runner, config = await _restore_runner_and_config(
                    trace=trace,
                    base_runner=trace_runner,
                    approval_batch_id=batch.batch_id,
                )
                await restored_runner.trace_store.update_trace(
                    tid,
                    status="running",
                    error_message=None,
                    completed_at=None,
                )
                task = asyncio.create_task(
                    _run_in_background(
                        tid,
                        [],
                        config,
                        runner_instance=restored_runner,
                    )
                )
                _running_tasks[tid] = task
                count += 1
                continue
            if trace.status == "running":
                logger.info(f"[Reconciliation] Fixing trace {tid}: running -> stopped")
                await trace_runner.trace_store.update_trace(
                    tid,
                    status="stopped",
                    result_summary="[Reconciliation] 任务由于服务重启或异常中断已自动停止。"
                )
                count += 1

        if _application_runtime is not None:
            try:
                await _application_runtime.reconcile_all_events()
            except Exception:
                logger.exception(
                    "[Reconciliation] Application run-event reconciliation failed"
                )
        
        if count > 0:
            logger.info(f"[Reconciliation] Successfully reconciled {count} traces.")
    except Exception as e:
        logger.exception("[Reconciliation] Failed to reconcile traces: %s", e)
        raise


# ===== 经验 API =====


@experiences_router.get("/experiences")
async def list_experiences():
    """读取经验文件内容"""
    runner = _get_runner()
    experiences_path = getattr(runner, "experiences_path", "./.cache/experiences.md")

    if not experiences_path or not os.path.exists(experiences_path):
        return {"content": "", "path": experiences_path}

    with open(experiences_path, "r", encoding="utf-8") as f:
        content = f.read()

    return {"content": content, "path": experiences_path}