feat(runner): 持久化运行契约并引入工具审批状态机

新增带版本和内容哈希的 RunConfigSnapshotV1，固化模型、循环、工具、Agent、Memory、Knowledge、并发、侧分支和项目配置；续跑、回溯、反思、压缩与审批恢复只使用持久快照。

历史 Recursive Trace 缺少快照时拒绝不可靠续跑；Legacy 从已有 Trace 字段和安全默认值推导一次。项目 Runner 在状态判断前由 snapshot.project_name 恢复，旧 Legacy 项目也先生成快照再选择 Runner。

tools 显式列表改为精确能力集合，Schema 与执行 allowlist 共用同一结果；ToolRegistry 在 dispatch 时复核工具名、页面 URL 和参数哈希绑定的确认授权。

实现 waiting_confirmation 持久状态机：混合批次整体暂停、一次性批准或拒绝、限定 editable_params、重复审批冲突、停止取消，以及 pending/decided/executing 三类重启恢复策略。

SideBranch 按 branch_id 和 branch_type 精确恢复，以持久 assistant 消息重建 turns_used，并在恢复时取持久值与实际值的最大值，已耗尽分支不再获得额外模型轮次。

同步 Trace API 文档和 TaskProtocol 兼容用例，明确本地开发边界、审批接口、execution_unknown 语义及运行快照规则。

cyber_agent/core/run_snapshot.py

@@ -0,0 +1,153 @@
+"""Versioned, hash-bound snapshots for reproducible Agent runs."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, is_dataclass
+from hashlib import sha256
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+RUN_CONFIG_SNAPSHOT_CONTEXT_KEY = "run_config_snapshot"
+RUN_CONFIG_SNAPSHOT_HASH_CONTEXT_KEY = "run_config_snapshot_hash"
+
+
+class RunConfigSnapshotError(ValueError):
+    """A persisted run snapshot is absent, malformed, or was modified."""
+
+
+class RunConfigSnapshotV1(BaseModel):
+    """Persisted behavior-affecting fields from ``RunConfig``.
+
+    Runtime routing fields (trace_id, parent IDs, rewind position and forced side
+    branches) are intentionally excluded because they describe one invocation,
+    not the immutable behavior of the run.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal[1] = 1
+    model: str
+    temperature: float
+    max_iterations: int = Field(gt=0)
+    extra_llm_params: dict[str, Any]
+
+    tools: list[str] | None
+    tool_groups: list[str] | None
+    exclude_tools: list[str]
+    auto_execute_tools: bool
+
+    agent_type: str
+    uid: str | None
+    skills: list[str] | None
+    system_prompt_hash: str | None = Field(
+        default=None,
+        pattern=r"^[0-9a-f]{64}$",
+    )
+
+    enable_memory: bool
+    memory: dict[str, Any] | None
+    memory_identity: str | None = Field(
+        default=None,
+        pattern=r"^memory-v1:[0-9a-f]{64}$",
+    )
+    knowledge: dict[str, Any]
+
+    parallel_tool_execution: bool
+    child_execution_mode: Literal["sequential", "parallel"]
+    max_parallel_children: int = Field(gt=0)
+    side_branch_max_turns: int = Field(gt=0)
+    goal_compression: Literal["none", "on_complete", "on_overflow"]
+
+    enable_prompt_caching: bool
+    enable_research_flow: bool
+    project_name: str | None = None
+    custom_context: dict[str, Any]
+    legacy_inferred: bool = False
+
+    @classmethod
+    def from_run_config(
+        cls,
+        config: Any,
+        *,
+        memory_identity: str | None,
+        system_prompt_hash: str | None = None,
+        legacy_inferred: bool = False,
+    ) -> "RunConfigSnapshotV1":
+        def dump(value: Any) -> Any:
+            if value is None:
+                return None
+            if is_dataclass(value):
+                return asdict(value)
+            if isinstance(value, BaseModel):
+                return value.model_dump(mode="json")
+            if isinstance(value, dict):
+                return dict(value)
+            raise TypeError(f"run snapshot field is not serializable: {type(value)!r}")
+
+        context = config.context or {}
+        return cls(
+            model=config.model,
+            temperature=config.temperature,
+            max_iterations=config.max_iterations,
+            extra_llm_params=dict(config.extra_llm_params),
+            tools=list(config.tools) if config.tools is not None else None,
+            tool_groups=(
+                list(config.tool_groups) if config.tool_groups is not None else None
+            ),
+            exclude_tools=list(config.exclude_tools),
+            auto_execute_tools=config.auto_execute_tools,
+            agent_type=config.agent_type,
+            uid=config.uid,
+            skills=list(config.skills) if config.skills is not None else None,
+            system_prompt_hash=system_prompt_hash,
+            enable_memory=config.enable_memory,
+            memory=dump(config.memory),
+            memory_identity=memory_identity,
+            knowledge=dump(config.knowledge),
+            parallel_tool_execution=config.parallel_tool_execution,
+            child_execution_mode=config.child_execution_mode,
+            max_parallel_children=config.max_parallel_children,
+            side_branch_max_turns=config.side_branch_max_turns,
+            goal_compression=config.goal_compression,
+            enable_prompt_caching=config.enable_prompt_caching,
+            enable_research_flow=config.enable_research_flow,
+            project_name=context.get("project_name"),
+            custom_context=dict(context),
+            legacy_inferred=legacy_inferred,
+        )
+
+    @property
+    def snapshot_hash(self) -> str:
+        payload = json.dumps(
+            self.model_dump(mode="json"),
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+            allow_nan=False,
+        )
+        return sha256(payload.encode("utf-8")).hexdigest()
+
+
+def persist_run_config_snapshot(
+    context: dict[str, Any],
+    snapshot: RunConfigSnapshotV1,
+) -> None:
+    context[RUN_CONFIG_SNAPSHOT_CONTEXT_KEY] = snapshot.model_dump(mode="json")
+    context[RUN_CONFIG_SNAPSHOT_HASH_CONTEXT_KEY] = snapshot.snapshot_hash
+
+
+def load_run_config_snapshot(context: dict[str, Any]) -> RunConfigSnapshotV1:
+    raw = context.get(RUN_CONFIG_SNAPSHOT_CONTEXT_KEY)
+    digest = context.get(RUN_CONFIG_SNAPSHOT_HASH_CONTEXT_KEY)
+    if raw is None or digest is None:
+        raise RunConfigSnapshotError("run config snapshot is missing")
+    try:
+        snapshot = RunConfigSnapshotV1.model_validate(raw)
+    except Exception as exc:
+        raise RunConfigSnapshotError(f"invalid run config snapshot: {exc}") from exc
+    if snapshot.snapshot_hash != digest:
+        raise RunConfigSnapshotError("run config snapshot hash mismatch")
+    return snapshot

cyber_agent/core/runner.py

@@ -22,6 +22,7 @@ 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
@@ -38,8 +39,26 @@ from cyber_agent.trace.compaction import (
 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 MemoryConfig
+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,
+    load_run_config_snapshot,
+    persist_run_config_snapshot,
+)
 from cyber_agent.core.agent_mode import (
     AGENT_MODE_CONTEXT_KEY,
     AgentMode,
@@ -138,8 +157,9 @@ class SideBranchContext:
     start_head_seq: int          # 侧分支起点的 head_seq
     start_sequence: int          # 侧分支第一条消息的 sequence
     start_history_length: int    # 侧分支起点的 history 长度
-    start_iteration: int         # 侧分支开始时的 iteration
+    start_iteration: int         # 旧 Trace 兼容字段；不再用于轮次预算
     max_turns: int = 5           # 最大轮次
+    turns_used: int = 0          # 已持久化的 assistant 轮次
 
     def to_dict(self) -> Dict[str, Any]:
         """转换为字典（用于持久化和传递给工具）"""
@@ -150,6 +170,7 @@ class SideBranchContext:
             "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(),
         }
@@ -214,6 +235,36 @@ class RunConfig:
     # None = 默认 Agent（无长期记忆）；赋值 MemoryConfig 使该 Agent 成为 memory-bearing Agent
     memory: Optional["MemoryConfig"] = None
 
+    # --- 一次性恢复动作（不进入持久化 RunConfigSnapshot） ---
+    approval_batch_id: Optional[str] = None
+
+    def apply_snapshot(self, snapshot: RunConfigSnapshotV1) -> 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)
+
 
     # BUILTIN_TOOLS 硬编码列表已移除（2026-04）。
     # 工具可用性现在由 @tool(groups=[...]) 声明 + RunConfig.tool_groups 过滤控制。
@@ -306,9 +357,6 @@ class AgentRunner:
         # key: 图片内容的 hash, value: {"downscaled": ..., "description": ...}
         self._image_opt_cache: Dict[str, Dict[str, Any]] = {}
 
-        # 当前 run 的 MemoryConfig（由 run() 根据 RunConfig.memory 设置）
-        # dream 工具从 context.runner 读取此字段，判断是否 memory-bearing
-        self._current_memory_config: Optional[MemoryConfig] = None
         self.resource_budget = (
             ResourceBudgetController(trace_store) if trace_store else None
         )
@@ -742,7 +790,9 @@ class AgentRunner:
     async def dream(
         self,
         memory_config: MemoryConfig,
-        trace_filter: Optional[Callable[["Trace"], bool]] = None,
+        *,
+        uid: Optional[str],
+        agent_type: str,
         reflect_model: str = "gpt-4o-mini",
         dream_model: str = "gpt-4o",
     ) -> "DreamReport":
@@ -752,18 +802,22 @@ class AgentRunner:
 
         Args:
             memory_config: 记忆配置
-            trace_filter: 可选 trace 过滤（按 agent_type/owner 等）
+            uid/agent_type: 与 MemoryConfig 一起形成强制 Dream 身份边界
             reflect_model: per-trace 反思模型
             dream_model: 跨 trace 整合模型
         """
-        from cyber_agent.core.dream import run_dream
+        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,
-            trace_filter=trace_filter,
+            dream_scope=DreamScope(
+                uid=uid,
+                agent_type=agent_type,
+                memory_identity=compute_memory_identity(memory_config),
+            ),
             reflect_model=reflect_model,
             dream_model=dream_model,
         )
@@ -797,9 +851,6 @@ class AgentRunner:
         trace = None
         run_cancel_event: Optional[asyncio.Event] = None
 
-        # Memory 模式开关（dream 工具会读取此字段）
-        self._current_memory_config = config.memory
-
         try:
             # Phase 1: PREPARE TRACE
             trace, goal_tree, sequence = await self._prepare_trace(messages, config)
@@ -832,6 +883,7 @@ class AgentRunner:
                     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
@@ -851,19 +903,34 @@ class AgentRunner:
             ):
                 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": str(e),
+                    "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):
@@ -933,6 +1000,8 @@ class AgentRunner:
         )
         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 文本结果"
@@ -1179,6 +1248,16 @@ class AgentRunner:
         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
+        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(
@@ -1247,6 +1326,67 @@ class AgentRunner:
                 "Validator traces cannot be continued or rewound"
             )
 
+        # 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")
+        config.apply_snapshot(snapshot)
+        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.
@@ -1381,14 +1521,31 @@ class AgentRunner:
                     trace_id, trace_obj.head_sequence
                 )
 
-                # 修复 orphaned tool_calls（中断导致的 tool_call 无 tool_result）
-                main_path, sequence = await self._heal_orphaned_tool_calls(
-                    main_path, trace_id, goal_tree, sequence,
-                )
+                # 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)
@@ -1497,6 +1654,32 @@ class AgentRunner:
         # 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
 
@@ -1837,6 +2020,193 @@ class AgentRunner:
 
         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,
@@ -1872,13 +2242,29 @@ class AgentRunner:
             branch_id = side_branch_data["branch_id"]
             start_sequence = side_branch_data["start_sequence"]
 
-            # 从数据库查询侧分支消息（按 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
+                    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(
@@ -1889,6 +2275,7 @@ class AgentRunner:
                     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(
@@ -1903,6 +2290,49 @@ class AgentRunner:
                 # 重新计算 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):
@@ -1979,6 +2409,7 @@ class AgentRunner:
                     start_history_length=len(history),
                     start_iteration=iteration,
                     max_turns=config.side_branch_max_turns,
+                    turns_used=0,
                 )
 
                 # 持久化侧分支状态
@@ -1993,6 +2424,7 @@ class AgentRunner:
                         "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(),
                     }
 
@@ -2161,10 +2593,7 @@ class AgentRunner:
                     not has_context_call
                     and not lifecycle_call_count
                     and not protocol_lifecycle_required
-                    and (
-                        runtime_policy.mode is not AgentMode.RECURSIVE
-                        or "get_current_context" in runtime_tool_names
-                    )
+                    and "get_current_context" in runtime_tool_names
                 ):
                     # 手动添加 get_current_context 工具调用
                     context_call_id = f"call_context_{uuid.uuid4().hex[:8]}"
@@ -2183,10 +2612,7 @@ class AgentRunner:
                 and iteration == 0
                 and not lifecycle_call_count
                 and not protocol_lifecycle_required
-                and (
-                    runtime_policy.mode is not AgentMode.RECURSIVE
-                    or "skill" in runtime_tool_names
-                )
+                and "skill" in runtime_tool_names
             ):
                 skills_to_inject = self._check_skills_need_injection(
                     trace, inject_skills, history, skill_recency_threshold
@@ -2304,7 +2730,18 @@ class AgentRunner:
                         )
                     self.log.info(f"[Knowledge Eval] 已写入 {len(eval_results.get('evaluations', []))} 条评估结果")
 
-            # 如果在侧分支，记录到 assistant_msg（已持久化，不需要额外维护）
+            # 一轮以成功持久化的侧分支 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
@@ -2325,8 +2762,7 @@ class AgentRunner:
 
             # 检查侧分支是否应该退出
             if side_branch_ctx:
-                # 计算侧分支已执行的轮次
-                turns_in_branch = iteration - side_branch_ctx.start_iteration
+                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:
@@ -2353,7 +2789,12 @@ class AgentRunner:
                         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
+                            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):
@@ -2522,6 +2963,87 @@ class AgentRunner:
                     })
                 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
@@ -2649,8 +3171,12 @@ class AgentRunner:
                                 import base64 as b64mod
                                 for img in tool_images:
                                     if img.get("data"):
-                                        png_path = self.trace_store._get_messages_dir(trace_id) / f"{tool_msg.message_id}.png"
-                                        png_path.write_bytes(b64mod.b64decode(img["data"]))
+                                        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
@@ -2828,8 +3354,12 @@ class AgentRunner:
                                 import base64 as b64mod
                                 for img in tool_images:
                                     if img.get("data"):
-                                        png_path = self.trace_store._get_messages_dir(trace_id) / f"{tool_msg.message_id}.png"
-                                        png_path.write_bytes(b64mod.b64decode(img["data"]))
+                                        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  # 只存第一张
     
@@ -4235,12 +4765,14 @@ class AgentRunner:
         exclude_tools: Optional[List[str]] = None,
     ) -> set[str]:
         """解析 RunConfig 的基础工具能力集合。"""
-        if tool_groups is not None:
+        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 tools is not None:
-            tool_names |= set(tools)
         if exclude_tools:
             tool_names -= set(exclude_tools)
         return tool_names
@@ -4310,8 +4842,8 @@ class AgentRunner:
     ) -> List[Dict]:
         """按 Trace 持久化模式和当前协议状态生成工具 Schema。
 
-        主循环把结果交给 LLM；Recursive revision 2 本地委托使用 ``task_brief``，
-        仍保留 ``remote_*`` 的 ``task``，Legacy 和 Recursive revision 1 继续使用 ``task``。
+        主循环把结果交给 LLM；Recursive revision 2 只允许受治理的本地
+        ``task_brief`` 委托，Legacy 和 Recursive revision 1 继续使用 ``task``。
         """
         tool_names = (
             runtime_tool_names
@@ -4334,16 +4866,16 @@ class AgentRunner:
             required = parameters.setdefault("required", [])
             if policy.requires_task_protocol:
                 properties.pop("messages", None)
-                if "task" in properties:
-                    properties["task"]["description"] = (
-                        "Only for agent_type=remote_*; local Recursive calls use task_brief."
-                    )
+                properties.pop("task", None)
                 if "task_brief" in properties:
                     properties["task_brief"]["description"] = (
-                        "Required for local Recursive delegation; not used by remote_* calls."
+                        "Required structured contract for Recursive local delegation. "
+                        "remote_* agents are not supported in Recursive revision 2."
                     )
                 if "task" in required:
                     required.remove("task")
+                if "task_brief" in properties and "task_brief" not in required:
+                    required.append("task_brief")
             else:
                 properties.pop("task_brief", None)
                 if "task" in properties and "task" not in required:
@@ -4373,6 +4905,16 @@ class AgentRunner:
             "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,
@@ -4396,7 +4938,10 @@ class AgentRunner:
                 RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY: config.max_parallel_children,
             })
             return context
-        return {**framework_context, **(config.context or {})}
+        # 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，这里保留引用以保持向后兼容

cyber_agent/docs/trace-api.md

@@ -2,6 +2,37 @@
 
 > 执行轨迹记录和存储的后端实现
 
+## 运行边界
+
+Trace API 当前只面向单用户本地开发，不提供账号、租户或 Token 鉴权。
+
+- API 默认仅监听 `127.0.0.1:8000`，前端仅监听 `127.0.0.1:3000`。
+- HTTP 只接受 `localhost` 和 `127.0.0.1` Host；CORS 只允许这两个本地前端 Origin。
+- Trace 和日志 WebSocket 同样要求本地前端 Origin。
+- 默认禁用 API 热重载；确需时显式设置 `TRACE_API_RELOAD=true`。
+
+不应将当前 API 通过公网 IP、容器端口映射或反向代理对外暴露。
+
+## 工具执行确认
+
+当一批 tool calls 中任意一项需要人工确认时，Trace 进入
+`waiting_confirmation`，整批调用都不会提前执行。确认状态持久化在该
+Trace 目录的 `tool_approvals.json`。
+
+- `GET /api/traces/{trace_id}/tool-approvals/pending`：读取待决策批次。
+- `POST /api/traces/{trace_id}/tool-approvals/{batch_id}`：一次提交全部待决策
+  调用的 `approve` / `reject`；只能修改工具声明的 `editable_params`。
+- 待确认时普通 `/run` 返回 409；`/stop` 取消整批并停止 Trace。
+- 重启时，未决策批次恢复为等待确认；已决策但尚未开始执行的
+  批次从持久化 assistant tool-call 安全恢复。
+- 重复决策返回 409。如进程在已开始的副作用期间中断，批次进入
+  `execution_unknown`，Trace 失败关闭且不会自动重试。
+
+新 Trace 在 `context.run_config_snapshot` 保存带哈希的完整运行配置。
+续跑、回溯、反思和压缩只从该快照恢复行为配置，不接受本次覆盖。
+无快照的历史 Recursive Trace 拒绝续跑；Legacy Trace 使用安全默认值推导
+一次并标记 `legacy_inferred=true`。
+
 ---
 
 ## 架构概览

cyber_agent/tools/registry.py

@@ -16,7 +16,8 @@ import logging
 import time
 from typing import Any, Callable, Dict, List, Optional
 
-from cyber_agent.tools.url_matcher import filter_by_url
+from cyber_agent.tools.approval import tool_argument_hash
+from cyber_agent.tools.url_matcher import filter_by_url, match_url_with_patterns
 
 logger = logging.getLogger(__name__)
 
@@ -221,6 +222,26 @@ class ToolRegistry:
 		tool_names = self.get_tool_names(current_url)
 		return self.get_schemas(tool_names)
 
+	def get_runtime_policy(self, tool_name: str) -> Dict[str, Any]:
+		"""Return an immutable copy of execution-relevant tool policy metadata."""
+		if tool_name not in self._tools:
+			return {
+				"requires_confirmation": False,
+				"editable_params": [],
+				"url_patterns": None,
+			}
+		tool = self._tools[tool_name]
+		ui = tool.get("ui_metadata", {})
+		return {
+			"requires_confirmation": bool(ui.get("requires_confirmation", False)),
+			"editable_params": list(ui.get("editable_params", [])),
+			"url_patterns": (
+				list(tool["url_patterns"])
+				if tool.get("url_patterns") is not None
+				else None
+			),
+		}
+
 	async def execute(
 		self,
 		name: str,
@@ -230,6 +251,8 @@ class ToolRegistry:
 		sensitive_data: Optional[Dict[str, Any]] = None,
 		inject_values: Optional[Dict[str, Any]] = None,
 		allowed_tool_names: Optional[set[str]] = None,
+		tool_call_id: Optional[str] = None,
+		approval_grant: Optional[Dict[str, Any]] = None,
 	) -> str:
 		"""
 		执行一次工具调用，是 AgentRunner 的统一 dispatch 入口。
@@ -258,6 +281,42 @@ class ToolRegistry:
 			logger.error(f"[ToolRegistry] {error_msg}")
 			return json.dumps({"error": error_msg}, ensure_ascii=False)
 
+		tool_policy = self.get_runtime_policy(name)
+		url_patterns = tool_policy.get("url_patterns")
+		if url_patterns:
+			current_url = context.get("page_url") if context else None
+			if not current_url or not match_url_with_patterns(current_url, url_patterns):
+				error_msg = f"Tool URL policy denied: {name}"
+				logger.warning("[ToolRegistry] %s", error_msg)
+				return json.dumps({"error": error_msg}, ensure_ascii=False)
+
+		if tool_policy.get("requires_confirmation"):
+			grant = approval_grant or {}
+			context_trace_id = context.get("trace_id") if context else None
+			expected_hash = (
+				tool_argument_hash(
+					tool_call_id=tool_call_id,
+					tool_name=name,
+					arguments=arguments,
+				)
+				if tool_call_id
+				else None
+			)
+			valid_grant = (
+				tool_call_id is not None
+				and bool(grant.get("batch_id"))
+				and context_trace_id is not None
+				and grant.get("trace_id") == context_trace_id
+				and grant.get("tool_call_id") == tool_call_id
+				and grant.get("tool_name") == name
+				and grant.get("argument_hash") == expected_hash
+				and grant.get("decision") in {"approved", "auto_approved"}
+			)
+			if not valid_grant:
+				error_msg = f"Tool confirmation required: {name}"
+				logger.warning("[ToolRegistry] %s", error_msg)
+				return json.dumps({"error": error_msg}, ensure_ascii=False)
+
 		start_time = time.time()
 		stats = self._stats[name]
 		stats.call_count += 1

cyber_agent/trace/run_api.py

@@ -63,9 +63,9 @@ class CreateRequest(BaseModel):
         ...,
         description="OpenAI SDK 格式的输入消息。可包含 system + user 消息；若无 system 消息则从 skills 自动构建",
     )
-    model: str = Field("gpt-4o", description="模型名称")
-    temperature: float = Field(0.3)
-    max_iterations: int = Field(200)
+    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)
@@ -129,6 +129,16 @@ class CompactResponse(BaseModel):
     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):
@@ -178,6 +188,7 @@ _running_tasks: Dict[str, asyncio.Task] = {}
 # 记住每个运行中 Trace 真正使用的 Runner，避免 Example 专属 Runner 的续跑/停止误落到全局 Runner。
 # 映射只覆盖当前进程活跃任务，由两个后台执行入口按对象身份清理。
 _running_runners: Dict[str, Any] = {}
+_approval_locks: Dict[str, asyncio.Lock] = {}
 
 
 async def _cancel_and_wait_for_running_task(trace_id: str) -> None:
@@ -194,6 +205,127 @@ async def _cancel_and_wait_for_running_task(trace_id: str) -> None:
         _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, policy_from_context
+    from cyber_agent.core.run_snapshot import (
+        RUN_CONFIG_SNAPSHOT_CONTEXT_KEY,
+        RunConfigSnapshotV1,
+        RunConfigSnapshotError,
+        load_run_config_snapshot,
+        persist_run_config_snapshot,
+    )
+    from cyber_agent.core.runner import RunConfig
+
+    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
+        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
+
+    runner = base_runner
+    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
+    return runner, config
+
+
 async def _run_in_background(trace_id: str, messages: List[Dict], config, runner_instance=None):
     """后台执行已知 Trace ID 的 Agent，消费 run() 的所有 yield。
 
@@ -257,6 +389,7 @@ async def create_and_run(req: CreateRequest):
     Recursive 的 root_task_anchor 从此传入 Runner 完成门禁；获取 trace_id 后立即返回，执行继续留在后台。
     """
     import importlib
+    from dataclasses import replace
     from cyber_agent.core.runner import RunConfig
 
     runner = None
@@ -273,20 +406,26 @@ async def create_and_run(req: CreateRequest):
                 runner, example_messages, default_config = await example_module.init_project_env(req.messages)
                 messages = example_messages
                 
-                # 合并请求配置和 example 默认配置
-                config = RunConfig(
-                    model=req.model or default_config.model,
-                    temperature=req.temperature if req.temperature is not None else default_config.temperature,
-                    max_iterations=req.max_iterations or default_config.max_iterations,
-                    tools=req.tools or default_config.tools,
-                    name=req.name or default_config.name,
-                    uid=req.uid or default_config.uid,
-                    enable_research_flow=default_config.enable_research_flow,
-                    child_execution_mode=default_config.child_execution_mode,
-                    max_parallel_children=default_config.max_parallel_children,
-                    root_task_anchor=req.root_task_anchor,
-                    context={"project_name": req.project_name}
-                )
+                # 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.")
@@ -300,9 +439,9 @@ async def create_and_run(req: CreateRequest):
     if not runner:
         _get_runner()  # 验证全局默认 Runner 已配置
         config = RunConfig(
-            model=req.model,
-            temperature=req.temperature,
-            max_iterations=req.max_iterations,
+            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,
@@ -439,9 +578,6 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
     如果人工插入 message 的位置打断了一个工具调用过程（assistant 消息有 tool_calls
     但缺少对应的 tool responses），框架会自动检测并调整插入位置，确保不会产生不一致的状态。
     """
-    from cyber_agent.core.runner import RunConfig
-    import importlib
-
     runner = _running_runners.get(trace_id) or _get_runner()
 
     # 将 message_id 转换为内部使用的 sequence 整数
@@ -449,11 +585,28 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
     if req.after_message_id is not None:
         after_sequence = _parse_sequence_from_message_id(req.after_message_id)
 
-    # 验证 trace 存在
+    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,
+    )
+
+    # 验证 trace 状态
     if runner.trace_store:
-        trace = await runner.trace_store.get_trace(trace_id)
-        if not trace:
-            raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
+        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:
@@ -483,30 +636,6 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
         else:
             raise HTTPException(status_code=409, detail="Trace is already running")
 
-        # 检测 trace 中是否包含 project_name 环境定义
-        trace_context = trace.context or {}
-        project_name = trace_context.get("project_name")
-        
-        if project_name:
-            try:
-                module_name = f"examples.{project_name}.run"
-                example_module = importlib.import_module(module_name)
-                if hasattr(example_module, "init_project_env"):
-                    logger.info(f"Trace {trace_id} 绑定了项目 {project_name}，动态加载执行环境...")
-                    project_runner, project_msgs, default_config = await example_module.init_project_env()
-                    runner = project_runner  # 发生替换
-            except ImportError as e:
-                if getattr(e, "name", None) == module_name:
-                    logger.warning(f"Project '{project_name}' has no custom run.py, keeping default runner.")
-                else:
-                    import traceback
-                    logger.error(f"Error INSIDE {module_name} during resume:\n{traceback.format_exc()}")
-            except Exception as e:
-                import traceback
-                logger.error(f"Unexpected error loading run.py environment for project {project_name} in trace {trace_id}:\n{traceback.format_exc()}")
-
-    config = RunConfig(trace_id=trace_id, after_sequence=after_sequence)
-
     # 恢复运行时，将状态从 stopped 改回 running，并广播状态变化
     if runner.trace_store and trace_id:
         current_trace = await runner.trace_store.get_trace(trace_id)
@@ -527,6 +656,134 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
     )
 
 
+@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):
     """
@@ -536,6 +793,26 @@ async def stop_trace(trace_id: str):
     """
     runner = _running_runners.get(trace_id) or _get_runner()
 
+    if runner.trace_store:
+        trace = await runner.trace_store.get_trace(trace_id)
+        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)
 
@@ -561,8 +838,6 @@ async def reflect_trace(trace_id: str, req: ReflectRequest):
     LLM 可以调用工具（如 knowledge_search, knowledge_save）进行多轮推理。
     反思消息标记为侧分支（branch_type="reflection"），不在主路径上。
     """
-    from cyber_agent.core.runner import RunConfig
-
     runner = _get_runner()
 
     if not runner.trace_store:
@@ -573,19 +848,16 @@ async def reflect_trace(trace_id: str, req: ReflectRequest):
     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.")
 
-    # 使用 force_side_branch 触发反思侧分支
-    config = RunConfig(
-        trace_id=trace_id,
-        model=trace.model or "gpt-4o",
-        force_side_branch=["reflection"],  # 使用列表格式
-        max_iterations=20,  # 给侧分支足够的轮次
-        enable_prompt_caching=True,
-    )
-
     # 如果有 focus，可以通过追加消息传递（可选）
     messages = []
     if req.focus:
@@ -694,8 +966,6 @@ async def compact_trace(trace_id: str):
     LLM 可以调用工具（如 goal）进行多轮推理。
     压缩消息标记为侧分支（branch_type="compression"），不在主路径上。
     """
-    from cyber_agent.core.runner import RunConfig
-
     runner = _get_runner()
     if not runner.trace_store:
         raise HTTPException(status_code=503, detail="TraceStore not configured")
@@ -705,19 +975,16 @@ async def compact_trace(trace_id: str):
     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.")
 
-    # 使用 force_side_branch 触发压缩侧分支
-    config = RunConfig(
-        trace_id=trace_id,
-        model=trace.model or "gpt-4o",
-        force_side_branch=["compression"],  # 使用列表格式
-        max_iterations=20,  # 给侧分支足够的轮次
-        enable_prompt_caching=True,
-    )
-
     # 启动压缩任务（后台执行）
     task = asyncio.create_task(_run_trace_background(runner, [], config))
     _running_tasks[trace_id] = task
@@ -765,10 +1032,10 @@ async def list_running():
 
 async def reconcile_traces():
     """
-    状态对齐：启动时清理残留的 running 状态。
-    
-    当服务异常停止或重启后，磁盘上的 trace 状态可能仍显示为 running，
-    但对应的内存任务已不存在。本函数将其强制标记为 stopped。
+    状态对齐：启动时恢复审批门禁，再清理其他残留的 running 状态。
+
+    审批文件是这个局部状态机的持久化真相：pending 恢复等待；
+    decided 且尚未 executing 时安全恢复执行；executing 则失败关闭。
     """
     runner = _get_runner()
     if not runner or not runner.trace_store:
@@ -776,16 +1043,84 @@ async def reconcile_traces():
         return
 
     try:
-        # 获取所有 running 状态的 trace
-        running_traces = await runner.trace_store.list_traces(status="running", limit=1000)
-        if not running_traces:
+        traces = await runner.trace_store.list_traces(limit=10_000)
+        if not traces:
             return
 
         count = 0
-        for trace in running_traces:
+        for trace in traces:
             tid = trace.trace_id
-            # 如果不在活跃任务字典中（服务初次启动时此字典为空），则视为异常残留
-            if tid not in _running_tasks:
+            if tid in _running_tasks and not _running_tasks[tid].done():
+                continue
+            batch = (
+                await runner.trace_store.get_tool_approval_batch(tid)
+                if hasattr(runner.trace_store, "get_tool_approval_batch")
+                else None
+            )
+            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 runner.trace_store.update_trace(
+                        tid,
+                        status="waiting_confirmation",
+                        error_message=None,
+                        completed_at=None,
+                    )
+                    count += 1
+                continue
+            if batch and batch.status == "decided" and trace.status in {
+                "running", "stopped", "waiting_confirmation",
+            }:
+                logger.info(
+                    "[Reconciliation] Resuming safely-decided approval batch for %s",
+                    tid,
+                )
+                restored_runner, config = await _restore_runner_and_config(
+                    trace=trace,
+                    base_runner=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 batch and batch.status == "executing":
+                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 runner.trace_store.replace_tool_approval_batch(tid, batch)
+                await 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 trace.status == "running":
                 logger.info(f"[Reconciliation] Fixing trace {tid}: running -> stopped")
                 await runner.trace_store.update_trace(
                     tid,
@@ -797,7 +1132,8 @@ async def reconcile_traces():
         if count > 0:
             logger.info(f"[Reconciliation] Successfully reconciled {count} traces.")
     except Exception as e:
-        logger.error(f"[Reconciliation] Failed to reconcile traces: {e}")
+        logger.exception("[Reconciliation] Failed to reconcile traces: %s", e)
+        raise
 
 
 # ===== 经验 API =====

tests/test_task_protocol.py

@@ -12,6 +12,10 @@ from cyber_agent.core.resource_budget import (
     ResourceBudgetController,
 )
 from cyber_agent.core.runner import AgentRunner, RunConfig
+from cyber_agent.core.run_snapshot import (
+    RunConfigSnapshotV1,
+    persist_run_config_snapshot,
+)
 from cyber_agent.core.context_policy import (
     add_context_snapshot,
     create_context_snapshot,
@@ -850,21 +854,23 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual("failed", rejected["status"])
         self.assertIn("does not match", rejected["error"])
 
-    async def test_remote_agent_keeps_legacy_task_interface(self):
-        remote_result = {
+    async def test_remote_agent_is_rejected_by_recursive_revision_2(self):
+        remote_call = AsyncMock(return_value={
             "status": "completed",
             "summary": "remote result",
-        }
+        })
         with patch(
             "cyber_agent.tools.builtin.subagent._run_remote_agent",
-            new=AsyncMock(return_value=remote_result),
-        ) as remote_call:
+            new=remote_call,
+        ):
             result = await agent(
                 task="remote free text",
                 agent_type="remote_librarian",
+                context=self.context(),
             )
-        self.assertEqual(remote_result, result)
-        self.assertEqual("remote free text", remote_call.await_args.kwargs["task"])
+        self.assertEqual("failed", result["status"])
+        self.assertIn("does not support remote agents", result["error"])
+        remote_call.assert_not_awaited()
 
     async def test_revise_child_reuses_trace_and_preserves_report_history(self):
         first = await agent(task_brief=BRIEF, context=self.context())
@@ -1232,6 +1238,22 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
     async def asyncTearDown(self):
         self.temp_dir.cleanup()
 
+    async def bind_snapshot(self, trace_id, config):
+        """Make manually-created Recursive fixtures equivalent to new traces."""
+        trace = await self.store.get_trace(trace_id)
+        config.model = trace.model or config.model
+        config.uid = trace.uid
+        config.agent_type = trace.agent_type or "default"
+        persist_run_config_snapshot(
+            trace.context,
+            RunConfigSnapshotV1.from_run_config(
+                config,
+                memory_identity=None,
+            ),
+        )
+        await self.store.update_trace(trace_id, context=trace.context)
+        return config
+
     async def test_missing_report_becomes_protocol_error_after_two_corrections(self):
         calls = 0
 
@@ -1245,9 +1267,9 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             }
 
         runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
-        result = await runner.run_result(
-            messages=[],
-            config=RunConfig(
+        config = await self.bind_snapshot(
+            self.trace_id,
+            RunConfig(
                 trace_id=self.trace_id,
                 max_iterations=8,
                 knowledge=KnowledgeConfig(
@@ -1257,6 +1279,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
                 ),
             ),
         )
+        result = await runner.run_result(messages=[], config=config)
         self.assertEqual("failed", result["status"])
         trace = await self.store.get_trace(self.trace_id)
         report = ensure_task_protocol(trace.context)["task_report"]
@@ -1294,9 +1317,9 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             }
 
         runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
-        result = await runner.run_result(
-            messages=[],
-            config=RunConfig(
+        config = await self.bind_snapshot(
+            self.trace_id,
+            RunConfig(
                 trace_id=self.trace_id,
                 max_iterations=4,
                 knowledge=KnowledgeConfig(
@@ -1306,6 +1329,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
                 ),
             ),
         )
+        result = await runner.run_result(messages=[], config=config)
         self.assertEqual("completed", result["status"])
         trace = await self.store.get_trace(self.trace_id)
         self.assertEqual(
@@ -1357,9 +1381,9 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             }
 
         runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
-        result = await runner.run_result(
-            messages=[],
-            config=RunConfig(
+        config = await self.bind_snapshot(
+            root_id,
+            RunConfig(
                 trace_id=root_id,
                 max_iterations=3,
                 knowledge=KnowledgeConfig(
@@ -1369,6 +1393,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
                 ),
             ),
         )
+        result = await runner.run_result(messages=[], config=config)
         self.assertEqual("completed", result["status"])
         children = await self.store.list_traces(
             parent_trace_id=root_id,
@@ -1573,9 +1598,9 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
             }
 
         runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
-        result = await runner.run_result(
-            messages=[],
-            config=RunConfig(
+        config = await self.bind_snapshot(
+            root_id,
+            RunConfig(
                 trace_id=root_id,
                 max_iterations=5,
                 knowledge=KnowledgeConfig(
@@ -1585,6 +1610,7 @@ class RunnerCompletionGateTest(unittest.IsolatedAsyncioTestCase):
                 ),
             ),
         )
+        result = await runner.run_result(messages=[], config=config)
         self.assertEqual("failed", result["status"])
         self.assertEqual(3, calls)
         root = await self.store.get_trace(root_id)