修复 recursive 子 Agent 调度与子树停止

cyber_agent/core/agent_mode.py

@@ -22,6 +22,9 @@ AGENT_MODE_REVISION_CONTEXT_KEY = "agent_mode_revision"
 # Runner 注入给本地 agent 工具的内部上下文。它们不来自 LLM 参数，也不持久化
 # 到 Trace；Recursive 子级只能继承调用方在本次运行中实际拥有的能力与调度策略。
 RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY = "_recursive_capability_tool_names"
+RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY = "_recursive_child_execution_mode"
+RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY = "_recursive_max_parallel_children"
+MAX_PARALLEL_RECURSIVE_CHILDREN = 2
 
 
 class AgentMode(str, Enum):
@@ -114,3 +117,24 @@ def apply_policy_to_context(
     merged = dict(context or {})
     merged.update(policy.to_context())
     return merged
+
+
+def validate_recursive_child_execution(
+    mode: Any,
+    max_parallel_children: Any,
+) -> tuple[str, int]:
+    """校验单批 Recursive 子 Agent 的执行策略。"""
+    if mode not in {"sequential", "parallel"}:
+        raise ValueError(
+            "child_execution_mode must be 'sequential' or 'parallel'"
+        )
+    if (
+        isinstance(max_parallel_children, bool)
+        or not isinstance(max_parallel_children, int)
+        or not 1 <= max_parallel_children <= MAX_PARALLEL_RECURSIVE_CHILDREN
+    ):
+        raise ValueError(
+            "max_parallel_children must be an integer between 1 and "
+            f"{MAX_PARALLEL_RECURSIVE_CHILDREN}"
+        )
+    return mode, max_parallel_children

cyber_agent/core/runner.py

@@ -43,10 +43,13 @@ from cyber_agent.core.agent_mode import (
     AGENT_MODE_CONTEXT_KEY,
     AgentMode,
     RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY,
+    RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
+    RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
     apply_policy_to_context,
     assert_removed_config_absent,
     policy_from_context,
     policy_from_environment,
+    validate_recursive_child_execution,
 )
 from cyber_agent.core.task_protocol import (
     ensure_task_protocol,
@@ -138,6 +141,8 @@ class RunConfig:
     name: Optional[str] = None                 # 显示名称（空则由 utility_llm 自动生成）
     enable_prompt_caching: bool = True         # 启用 Anthropic Prompt Caching（仅 Claude 模型有效）
     parallel_tool_execution: bool = False      # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突）
+    child_execution_mode: Literal["sequential", "parallel"] = "sequential"
+    max_parallel_children: int = 2
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None             # None = 新建
@@ -226,6 +231,9 @@ class AgentRunner:
         self.log = logging.getLogger(logger_name) if logger_name else logger
         self.stdin_check: Optional[Callable] = None  # 由外部设置，用于子 agent 执行期间检查 stdin
         self._cancel_events: Dict[str, asyncio.Event] = {}  # trace_id → cancel event
+        self._recursive_active_traces: Dict[str, asyncio.Event] = {}
+        self._active_children: Dict[str, set[str]] = {}
+        self._active_parents: Dict[str, str] = {}
 
         # 知识保存跟踪（每个 trace 独立）
         self._saved_knowledge_ids: Dict[str, List[str]] = {}  # trace_id → [knowledge_ids]
@@ -304,6 +312,7 @@ class AgentRunner:
 
         config = config or RunConfig()
         trace = None
+        run_cancel_event: Optional[asyncio.Event] = None
 
         # Memory 模式开关（dream 工具会读取此字段）
         self._current_memory_config = config.memory
@@ -311,8 +320,19 @@ class AgentRunner:
         try:
             # Phase 1: PREPARE TRACE
             trace, goal_tree, sequence = await self._prepare_trace(messages, config)
-            # 注册取消事件
-            self._cancel_events[trace.trace_id] = asyncio.Event()
+            # 子 Trace 可能已在排队阶段收到停止信号，不能覆盖既有 Event。
+            run_cancel_event = self._cancel_events.setdefault(
+                trace.trace_id,
+                asyncio.Event(),
+            )
+            if policy_from_context(trace.context).mode is AgentMode.RECURSIVE:
+                self._recursive_active_traces[trace.trace_id] = run_cancel_event
+                if trace.parent_trace_id:
+                    self._active_children.setdefault(
+                        trace.parent_trace_id,
+                        set(),
+                    ).add(trace.trace_id)
+                    self._active_parents[trace.trace_id] = trace.parent_trace_id
             yield trace
 
             # 检查是否有未完成的侧分支（用于用户追加消息场景）
@@ -367,9 +387,8 @@ class AgentRunner:
                     yield trace_obj
             raise
         finally:
-            # 清理取消事件
-            if trace:
-                self._cancel_events.pop(trace.trace_id, None)
+            if trace and run_cancel_event is not None:
+                self.release_recursive_trace(trace.trace_id, run_cancel_event)
 
     async def run_result(
         self,
@@ -413,7 +432,14 @@ class AgentRunner:
         error = final_trace.error_message if final_trace else None
         summary = last_assistant_text
 
-        if not summary:
+        stopped_recursively = bool(
+            final_trace
+            and status == "stopped"
+            and policy_from_context(final_trace.context).mode is AgentMode.RECURSIVE
+        )
+        if not summary and stopped_recursively:
+            summary = "Agent execution stopped."
+        elif not summary:
             status = "failed"
             error = error or "Agent 没有产生 assistant 文本结果"
 
@@ -444,12 +470,102 @@ class AgentRunner:
         Returns:
             True 如果成功发送停止信号，False 如果该 trace 不在运行中
         """
-        cancel_event = self._cancel_events.get(trace_id)
-        if cancel_event is None:
+        return self.request_stop(trace_id)
+
+    def request_stop(self, trace_id: str) -> bool:
+        """同步设置停止信号，供 API 与 stdin 回调共用。"""
+        if trace_id not in self._cancel_events:
             return False
-        cancel_event.set()
+
+        if trace_id not in self._recursive_active_traces:
+            self._cancel_events[trace_id].set()
+            return True
+
+        pending = [trace_id]
+        visited = set()
+        while pending:
+            current = pending.pop()
+            if current in visited:
+                continue
+            visited.add(current)
+            event = self._cancel_events.get(current)
+            if event:
+                event.set()
+            pending.extend(tuple(self._active_children.get(current, ())))
         return True
 
+    def register_recursive_child(
+        self,
+        parent_trace_id: str,
+        child_trace_id: str,
+    ) -> asyncio.Event:
+        """登记已创建但可能仍在排队的 Recursive 直属孩子。"""
+        event = self._cancel_events.setdefault(child_trace_id, asyncio.Event())
+        self._recursive_active_traces[child_trace_id] = event
+        self._active_children.setdefault(parent_trace_id, set()).add(child_trace_id)
+        self._active_parents[child_trace_id] = parent_trace_id
+        parent_event = self._cancel_events.get(parent_trace_id)
+        if parent_event and parent_event.is_set():
+            event.set()
+        return event
+
+    def is_cancel_requested(self, trace_id: str) -> bool:
+        event = self._cancel_events.get(trace_id)
+        return bool(event and event.is_set())
+
+    def unregister_recursive_trace(
+        self,
+        trace_id: str,
+        event: Optional[asyncio.Event] = None,
+    ) -> None:
+        """幂等清理运行登记；event 防止旧运行误删续跑状态。"""
+        current_event = self._recursive_active_traces.get(trace_id)
+        if event is not None and current_event is not event:
+            return
+        self._recursive_active_traces.pop(trace_id, None)
+        parent_id = self._active_parents.pop(trace_id, None)
+        if parent_id:
+            children = self._active_children.get(parent_id)
+            if children:
+                children.discard(trace_id)
+                if not children:
+                    self._active_children.pop(parent_id, None)
+        if not self._active_children.get(trace_id):
+            self._active_children.pop(trace_id, None)
+
+    def release_recursive_trace(
+        self,
+        trace_id: str,
+        event: Optional[asyncio.Event] = None,
+    ) -> None:
+        """完成一次运行并按对象身份释放取消 Event 与父子登记。"""
+        current_event = self._cancel_events.get(trace_id)
+        if event is not None and current_event is not event:
+            return
+        self.unregister_recursive_trace(trace_id, event)
+        if self._cancel_events.get(trace_id) is current_event:
+            self._cancel_events.pop(trace_id, None)
+
+    async def _mark_trace_stopped(
+        self,
+        trace_id: str,
+        head_sequence: Optional[int],
+    ) -> Optional[Trace]:
+        if not self.trace_store:
+            return None
+        await self.trace_store.update_trace(
+            trace_id,
+            status="stopped",
+            head_sequence=head_sequence,
+            completed_at=datetime.now(),
+        )
+        try:
+            from cyber_agent.trace.websocket import broadcast_trace_status_changed
+            await broadcast_trace_status_changed(trace_id, "stopped")
+        except Exception:
+            pass
+        return await self.trace_store.get_trace(trace_id)
+
     # ===== 单次调用（保留）=====
 
     async def call(
@@ -526,6 +642,11 @@ class AgentRunner:
         """创建新 Trace"""
         # 在任何标题生成/LLM 调用前完成模式校验。
         policy = policy_from_environment(recursive_revision=2)
+        if policy.mode is AgentMode.RECURSIVE:
+            validate_recursive_child_execution(
+                config.child_execution_mode,
+                config.max_parallel_children,
+            )
         trace_id = str(uuid.uuid4())
 
         # 生成任务名称
@@ -586,7 +707,12 @@ class AgentRunner:
                 context=trace_obj.context,
             )
         else:
-            policy_from_context(trace_obj.context)  # validate persisted metadata
+            policy = policy_from_context(trace_obj.context)
+            if policy.mode is AgentMode.RECURSIVE:
+                validate_recursive_child_execution(
+                    config.child_execution_mode,
+                    config.max_parallel_children,
+                )
 
         goal_tree = await self.trace_store.get_goal_tree(config.trace_id)
         if goal_tree is None:
@@ -1150,22 +1276,9 @@ class AgentRunner:
             cancel_event = self._cancel_events.get(trace_id)
             if cancel_event and cancel_event.is_set():
                 self.log.info(f"Trace {trace_id} stopped by user")
-                if self.trace_store:
-                    await self.trace_store.update_trace(
-                        trace_id,
-                        status="stopped",
-                        head_sequence=head_seq,
-                        completed_at=datetime.now(),
-                    )
-                    # 广播状态变化给前端
-                    try:
-                        from cyber_agent.trace.websocket import broadcast_trace_status_changed
-                        await broadcast_trace_status_changed(trace_id, "stopped")
-                    except Exception:
-                        pass
-                    trace_obj = await self.trace_store.get_trace(trace_id)
-                    if trace_obj:
-                        yield trace_obj
+                trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
+                if trace_obj:
+                    yield trace_obj
                 return
 
             # 检查Goal完成触发的知识评估
@@ -1331,6 +1444,15 @@ class AgentRunner:
                 **config.extra_llm_params,
             )
 
+            if (
+                runtime_policy.mode is AgentMode.RECURSIVE
+                and self.is_cancel_requested(trace_id)
+            ):
+                trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
+                if trace_obj:
+                    yield trace_obj
+                return
+
             response_content = result.get("content", "")
             reasoning_content = result.get("reasoning_content", "")
             tool_calls = result.get("tool_calls")
@@ -1732,6 +1854,14 @@ class AgentRunner:
                 continue
 
             if tool_calls and config.auto_execute_tools:
+                if (
+                    runtime_policy.mode is AgentMode.RECURSIVE
+                    and self.is_cancel_requested(trace_id)
+                ):
+                    trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
+                    if trace_obj:
+                        yield trace_obj
+                    return
                 history.append({
                     "role": "assistant",
                     "content": response_content,
@@ -2167,6 +2297,15 @@ class AgentRunner:
 
             break
 
+        if (
+            policy_from_context(trace.context).mode is AgentMode.RECURSIVE
+            and self.is_cancel_requested(trace_id)
+        ):
+            trace_obj = await self._mark_trace_stopped(trace_id, head_seq)
+            if trace_obj:
+                yield trace_obj
+            return
+
         # max_iterations 等非正常退出也必须经过同一完成门禁。
         if self.trace_store:
             fresh_trace = await self.trace_store.get_trace(trace_id)
@@ -3361,7 +3500,7 @@ class AgentRunner:
         trigger_event: Optional[str],
     ) -> Dict[str, Any]:
         """构建工具上下文；Recursive 内部字段最后覆盖，不能由调用方伪造。"""
-        context = {
+        framework_context = {
             "store": self.trace_store,
             "trace_id": trace_id,
             "goal_id": goal_id,
@@ -3376,9 +3515,9 @@ class AgentRunner:
                 "max_turns": side_branch_ctx.max_turns,
                 "trigger_event": trigger_event,
             } if side_branch_ctx else None,
-            **(config.context or {}),
         }
         if policy_from_context(trace.context).mode is AgentMode.RECURSIVE:
+            context = {**(config.context or {}), **framework_context}
             context.update({
                 RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: sorted(
                     self._get_configured_tool_names(
@@ -3387,8 +3526,11 @@ class AgentRunner:
                         config.exclude_tools,
                     )
                 ),
+                RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY: config.child_execution_mode,
+                RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY: config.max_parallel_children,
             })
-        return context
+            return context
+        return {**framework_context, **(config.context or {})}
 
     # 默认 system prompt 前缀（当 config.system_prompt 和前端都未提供 system message 时使用）
     # 注意：此常量已迁移到 cyber_agent.core.prompts，这里保留引用以保持向后兼容

cyber_agent/core/task_protocol.py

@@ -158,6 +158,20 @@ def protocol_error_report(child_trace_id: str, reason: str) -> TaskReport:
     )
 
 
+def stopped_task_report(child_trace_id: str, reason: str) -> TaskReport:
+    return TaskReport(
+        child_trace_id=child_trace_id,
+        summary="Child Agent execution was stopped before completion.",
+        outcome="failed",
+        validation=Validation(hard_passed=False, open_issues=[reason]),
+        next_step_suggestion=NextStepSuggestion(
+            direction="NONE",
+            reason="Parent review must revise the stopped child or fail the task.",
+        ),
+        remaining_issues=[reason],
+    )
+
+
 def pending_review_entry(
     *,
     goal_id: str | None,

cyber_agent/tools/builtin/subagent.py

@@ -20,9 +20,12 @@ from cyber_agent.core.agent_mode import (
     AgentMode,
     AgentPolicy,
     RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY,
+    RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
+    RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
     apply_policy_to_context,
     assert_removed_config_absent,
     policy_from_context,
+    validate_recursive_child_execution,
 )
 from cyber_agent.core.task_protocol import (
     TaskBrief,
@@ -31,6 +34,7 @@ from cyber_agent.core.task_protocol import (
     format_task_brief,
     pending_review_entry,
     protocol_error_report,
+    stopped_task_report,
 )
 from cyber_agent.trace.models import Trace, Messages
 from cyber_agent.trace.trace_id import generate_sub_trace_id
@@ -223,6 +227,10 @@ async def _load_or_create_task_report(
         return protocol_error_report(child_trace_id, fallback_reason)
     state = ensure_task_protocol(child.context)
     report_data = state.get("task_report")
+    execution_stopped = (
+        child_result_status == "stopped"
+        or child.status == "stopped"
+    )
     execution_failed = (
         child_result_status != "completed"
         or child.status != "completed"
@@ -231,22 +239,34 @@ async def _load_or_create_task_report(
         try:
             report = TaskReport.model_validate(report_data)
             if report.child_trace_id != child_trace_id:
-                state["report_history"].append(report.model_dump())
+                if report.model_dump() not in state["report_history"]:
+                    state["report_history"].append(report.model_dump())
                 fallback_reason = (
                     "Child persisted a TaskReport whose child_trace_id does not "
                     "match its Trace"
                 )
+            elif execution_stopped:
+                if report.outcome in {"failed", "protocol_error"}:
+                    return report
+                if report.model_dump() not in state["report_history"]:
+                    state["report_history"].append(report.model_dump())
+                fallback_reason = "Child Agent execution was stopped"
             elif not execution_failed or report.outcome in {"failed", "protocol_error"}:
                 return report
             else:
-                state["report_history"].append(report.model_dump())
+                if report.model_dump() not in state["report_history"]:
+                    state["report_history"].append(report.model_dump())
                 fallback_reason = (
                     f"Child execution ended as {child_result_status} after submitting "
                     f"a {report.outcome} TaskReport: {fallback_reason}"
                 )
         except ValidationError:
             fallback_reason = "Child persisted an invalid TaskReport"
-    report = protocol_error_report(child_trace_id, fallback_reason)
+    report = (
+        stopped_task_report(child_trace_id, fallback_reason)
+        if execution_stopped
+        else protocol_error_report(child_trace_id, fallback_reason)
+    )
     state["task_report"] = report.model_dump()
     state["task_report_submitted_at_sequence"] = generated_at_sequence
     await store.update_trace(child_trace_id, context=child.context)
@@ -315,6 +335,69 @@ def _aggregate_stats(results: List[Dict[str, Any]]) -> Dict[str, Any]:
     }
 
 
+async def _stopped_child_result(store, runner, trace_id: str) -> Dict[str, Any]:
+    """停止尚未启动的预创建孩子，不产生任何模型调用或消息统计。"""
+    await store.update_trace(
+        trace_id,
+        status="stopped",
+        completed_at=datetime.now(),
+    )
+    event = getattr(runner, "_cancel_events", {}).get(trace_id)
+    release_trace = getattr(runner, "release_recursive_trace", None)
+    if release_trace:
+        release_trace(trace_id, event)
+    return {
+        "status": "stopped",
+        "summary": "Agent execution stopped.",
+        "error": "Child Agent execution was stopped",
+        "saved_knowledge_ids": [],
+        "stats": {"total_messages": 0, "total_tokens": 0, "total_cost": 0.0},
+    }
+
+
+async def _execute_child_spec(
+    spec: Dict[str, Any],
+    runner,
+    store,
+    semaphore: Optional[asyncio.Semaphore] = None,
+) -> Dict[str, Any]:
+    """延迟启动一个孩子；排队期间收到停止信号时不进入 Runner。"""
+    trace_id = spec["trace_id"]
+
+    async def execute() -> Dict[str, Any]:
+        is_cancelled = getattr(runner, "is_cancel_requested", lambda _tid: False)
+        if spec["recursive"] and is_cancelled(trace_id):
+            return await _stopped_child_result(store, runner, trace_id)
+        return await runner.run_result(
+            messages=spec["messages"],
+            config=spec["config"],
+            on_event=spec["on_event"],
+        )
+
+    if semaphore is None:
+        return await execute()
+
+    is_cancelled = getattr(runner, "is_cancel_requested", lambda _tid: False)
+    if spec["recursive"] and is_cancelled(trace_id):
+        return await _stopped_child_result(store, runner, trace_id)
+    async with semaphore:
+        return await execute()
+
+
+def _get_recursive_parent_capabilities(context: dict) -> Optional[set[str]]:
+    runner = context.get("runner")
+    capabilities = context.get(RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY)
+    if (
+        not runner
+        or not hasattr(runner, "tools")
+        or not hasattr(runner.tools, "get_tool_names")
+        or not isinstance(capabilities, list)
+        or any(not isinstance(name, str) for name in capabilities)
+    ):
+        return None
+    return set(runner.tools.get_tool_names()) & set(capabilities)
+
+
 def _get_allowed_tools(
     context: dict,
     agent_depth: int,
@@ -325,14 +408,9 @@ def _get_allowed_tools(
     if runner and hasattr(runner, "tools") and hasattr(runner.tools, "get_tool_names"):
         registered_tools = set(runner.tools.get_tool_names())
         if policy.mode is AgentMode.RECURSIVE:
-            parent_capabilities = context.get(
-                RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY
-            )
-            if not isinstance(parent_capabilities, list) or any(
-                not isinstance(name, str) for name in parent_capabilities
-            ):
+            allowed_tools = _get_recursive_parent_capabilities(context)
+            if allowed_tools is None:
                 return None
-            allowed_tools = registered_tools & set(parent_capabilities)
         else:
             # Legacy 保留原行为：从全局 Registry 取工具全集。
             allowed_tools = registered_tools
@@ -572,11 +650,14 @@ def _make_interactive_handler(runner, sub_trace_id: str, parent_trace_id: str, d
             return
         cmd = check_fn()
         if cmd in ('pause', 'quit'):
-            # cancel_event.set() 是同步操作，可以在同步回调中直接调用
-            for tid in (sub_trace_id, parent_trace_id):
-                ev = runner._cancel_events.get(tid)
-                if ev:
-                    ev.set()
+            request_stop = getattr(runner, "request_stop", None)
+            if request_stop:
+                request_stop(parent_trace_id)
+            else:
+                for tid in (sub_trace_id, parent_trace_id):
+                    ev = runner._cancel_events.get(tid)
+                    if ev:
+                        ev.set()
 
     return on_event
 
@@ -612,13 +693,32 @@ async def _run_agents(
 
     if (
         policy.mode is AgentMode.RECURSIVE
-        and _get_allowed_tools(context, 0, policy) is None
+        and _get_recursive_parent_capabilities(context) is None
     ):
         return {
             "status": "failed",
             "error": "Recursive agent capability context is missing or invalid",
         }
 
+    child_execution_mode = "parallel"
+    max_parallel_children = 2
+    if policy.mode is AgentMode.RECURSIVE:
+        try:
+            child_execution_mode, max_parallel_children = (
+                validate_recursive_child_execution(
+                    context.get(
+                        RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
+                        "sequential",
+                    ),
+                    context.get(
+                        RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
+                        2,
+                    ),
+                )
+            )
+        except ValueError as exc:
+            return {"status": "failed", "error": str(exc)}
+
     protocol_state = None
     approved_action = None
     if policy.requires_task_protocol:
@@ -851,9 +951,9 @@ async def _run_agents(
         protocol_state["protocol_correction_attempts"] = 0
         await store.update_trace(trace_id, context=parent_trace.context)
 
-    # 创建执行协程。Trace 在上面已经按批次预创建，
-    # 因此孩子配额不会被并发调用绕过。
-    coros = []
+    # 创建延迟执行规格。Trace 已按批次预创建，但不会提前创建 coroutine，
+    # 因而排队孩子被停止时不会产生未 await coroutine 或模型调用。
+    execution_specs = []
     for i, (task_item, msgs, child_record) in enumerate(
         zip(tasks, per_agent_msgs, child_records)
     ):
@@ -888,24 +988,30 @@ async def _run_agents(
             runner, cur_stid, trace_id, debug_printer=debug_printer
         )
 
-        coro = runner.run_result(
-            messages=agent_msgs,
-            config=_make_run_config(
-                trace_id=cur_stid,
-                agent_type=agent_type or ("delegate" if single else "explore"),
-                max_iterations=50 if single else 50,
-                model=parent_trace.model if parent_trace else "gpt-4o",
-                uid=parent_trace.uid if parent_trace else None,
-                tools=allowed_tools,
-                tool_groups=[],  # tools 是精确白名单，不再合并默认 core 组
-                name=task_label[:50],
-                skills=skills,
-                knowledge=context.get("knowledge_config"),
-                context=child_record["context"],
-            ),
-            on_event=on_event,
+        child_config = _make_run_config(
+            trace_id=cur_stid,
+            agent_type=agent_type or ("delegate" if single else "explore"),
+            max_iterations=50,
+            model=parent_trace.model if parent_trace else "gpt-4o",
+            uid=parent_trace.uid if parent_trace else None,
+            tools=allowed_tools,
+            tool_groups=[],  # tools 是精确白名单，不再合并默认 core 组
+            name=task_label[:50],
+            skills=skills,
+            knowledge=context.get("knowledge_config"),
+            context=child_record["context"],
+            child_execution_mode=child_execution_mode,
+            max_parallel_children=max_parallel_children,
         )
-        coros.append((i, cur_stid, collab_name, coro))
+        execution_specs.append({
+            "index": i,
+            "trace_id": cur_stid,
+            "collaborator": collab_name,
+            "messages": agent_msgs,
+            "config": child_config,
+            "on_event": on_event,
+            "recursive": policy.mode is AgentMode.RECURSIVE,
+        })
 
     # continue_from 不进入新建临界区，但仍需要恢复 Goal 的运行状态。
     if not goal_started:
@@ -919,12 +1025,20 @@ async def _run_agents(
         if policy.requires_task_protocol:
             _project_goal_status(context, goal_id, "waiting_children")
 
+    if policy.mode is AgentMode.RECURSIVE:
+        register_child = getattr(runner, "register_recursive_child", None)
+        if register_child:
+            for spec in execution_specs:
+                register_child(trace_id, spec["trace_id"])
+
     # 执行
     if single:
         # 单任务直接执行（带异常处理）
-        _, stid, collab_name, coro = coros[0]
+        spec = execution_specs[0]
+        stid = spec["trace_id"]
+        collab_name = spec["collaborator"]
         try:
-            result = await coro
+            result = await _execute_child_spec(spec, runner, store)
 
             await broadcast_sub_trace_completed(
                 trace_id, stid,
@@ -992,28 +1106,35 @@ async def _run_agents(
                 )
             return failed_result
     else:
-        # 检查父 Agent 是否开启了并发配置
-        is_parallel = getattr(runner.config, "parallel_tool_execution", True) if runner and hasattr(runner, "config") else True
-        
-        if is_parallel:
-            # 多任务并行执行
-            raw_results = await asyncio.gather(
-                *(coro for _, _, _, coro in coros),
-                return_exceptions=True,
-            )
+        async def execute_captured(
+            spec: Dict[str, Any],
+            semaphore: Optional[asyncio.Semaphore] = None,
+        ):
+            try:
+                return await _execute_child_spec(spec, runner, store, semaphore)
+            except Exception as exc:
+                return exc
+
+        if policy.mode is AgentMode.LEGACY:
+            # Legacy 冻结旧行为：多任务整批并行，不读取新配置。
+            raw_results = await asyncio.gather(*(
+                execute_captured(spec) for spec in execution_specs
+            ))
+        elif child_execution_mode == "parallel":
+            semaphore = asyncio.Semaphore(max_parallel_children)
+            raw_results = await asyncio.gather(*(
+                execute_captured(spec, semaphore) for spec in execution_specs
+            ))
         else:
-            # 多任务串行执行（为了省显存/内存）
             raw_results = []
-            for _, _, _, coro in coros:
-                try:
-                    res = await coro
-                    raw_results.append(res)
-                except Exception as e:
-                    raw_results.append(e)
+            for spec in execution_specs:
+                raw_results.append(await execute_captured(spec))
 
         processed_results = []
         for idx, raw in enumerate(raw_results):
-            _, stid, collab_name, _ = coros[idx]
+            spec = execution_specs[idx]
+            stid = spec["trace_id"]
+            collab_name = spec["collaborator"]
             if isinstance(raw, Exception):
                 error_result = {
                     "status": "failed",

cyber_agent/trace/run_api.py

@@ -160,18 +160,40 @@ class CommitResponse(BaseModel):
 # ===== 后台执行 =====
 
 _running_tasks: Dict[str, asyncio.Task] = {}
+_running_runners: Dict[str, Any] = {}
+
+
+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 _run_in_background(trace_id: str, messages: List[Dict], config, runner_instance=None):
     """后台执行 agent，消费 run() 的所有 yield"""
     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:
-        _running_tasks.pop(trace_id, None)
+        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(
@@ -182,10 +204,14 @@ async def _run_with_trace_signal(
 
     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():
@@ -193,7 +219,10 @@ async def _run_with_trace_signal(
         logger.error(f"Background run failed: {e}")
     finally:
         if trace_id:
-            _running_tasks.pop(trace_id, None)
+            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)
 
 
 # ===== 路由 =====
@@ -233,6 +262,8 @@ async def create_and_run(req: CreateRequest):
                     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,
                     context={"project_name": req.project_name}
                 )
         except ImportError as e:
@@ -264,8 +295,6 @@ async def create_and_run(req: CreateRequest):
     )
 
     trace_id = await trace_id_future
-    _running_tasks[trace_id] = task
-
     return RunResponse(
         trace_id=trace_id,
         status="started",
@@ -389,7 +418,7 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
     from cyber_agent.core.runner import RunConfig
     import importlib
 
-    runner = _get_runner()
+    runner = _running_runners.get(trace_id) or _get_runner()
 
     # 将 message_id 转换为内部使用的 sequence 整数
     after_sequence: Optional[int] = None
@@ -426,8 +455,7 @@ async def run_trace(trace_id: str, req: TraceRunRequest):
                 f"run_trace: task for {trace_id} not done yet but store status={store_trace.status!r}, "
                 "forcing cleanup to allow resume"
             )
-            _running_tasks[trace_id].cancel()
-            _running_tasks.pop(trace_id, None)
+            await _cancel_and_wait_for_running_task(trace_id)
         else:
             raise HTTPException(status_code=409, detail="Trace is already running")
 
@@ -483,7 +511,7 @@ async def stop_trace(trace_id: str):
     设置取消信号，agent loop 在下一个 LLM 调用前检查并退出。
     Trace 状态置为 "stopped"。
     """
-    runner = _get_runner()
+    runner = _running_runners.get(trace_id) or _get_runner()
 
     # 通过 runner 的 stop 方法设置取消信号
     stopped = await runner.stop(trace_id)

tests/test_recursive_child_scheduling.py

@@ -0,0 +1,154 @@
+import asyncio
+import tempfile
+import unittest
+
+from cyber_agent.core.agent_mode import (
+    RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY,
+    RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY,
+    RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY,
+)
+from cyber_agent.core.runner import AgentRunner
+from cyber_agent.tools.builtin.subagent import agent
+from cyber_agent.trace.goal_models import GoalTree
+from cyber_agent.trace.models import Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+class SchedulingToolRegistry:
+    def get_tool_names(self, current_url=None, groups=None):
+        return ["agent", "read_file"]
+
+
+class CountingRunner(AgentRunner):
+    def __init__(self, store, delays=None):
+        super().__init__(
+            trace_store=store,
+            tool_registry=SchedulingToolRegistry(),
+            llm_call=lambda **kwargs: None,
+        )
+        self.delays = delays or {}
+        self.active = 0
+        self.max_active = 0
+        self.start_order = []
+        self.finish_order = []
+        self.configs = []
+
+    async def run_result(self, messages, config, on_event=None):
+        self.configs.append(config)
+        self.start_order.append(config.name)
+        self.active += 1
+        self.max_active = max(self.max_active, self.active)
+        try:
+            await asyncio.sleep(self.delays.get(config.name, 0.01))
+            self.finish_order.append(config.name)
+            await self.trace_store.update_trace(config.trace_id, status="completed")
+            return {
+                "status": "completed",
+                "summary": f"result:{config.name}",
+                "saved_knowledge_ids": [],
+                "stats": {"total_messages": 1, "total_tokens": 1, "total_cost": 0.0},
+            }
+        finally:
+            self.active -= 1
+            event = self._cancel_events.get(config.trace_id)
+            self.release_recursive_trace(config.trace_id, event)
+
+
+class RecursiveChildSchedulingTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    async def create_root(self, trace_id, mode="recursive"):
+        await self.store.create_trace(Trace(
+            trace_id=trace_id,
+            mode="agent",
+            task=trace_id,
+            uid="user-1",
+            model="fake",
+            context={
+                "agent_mode": mode,
+                "agent_mode_revision": 1,
+                "agent_depth": 0,
+                "root_trace_id": trace_id,
+            },
+        ))
+        await self.store.update_goal_tree(trace_id, GoalTree(mission=trace_id))
+
+    def context(self, runner, trace_id, mode=None, maximum=None):
+        context = {
+            "store": self.store,
+            "trace_id": trace_id,
+            "runner": runner,
+            RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: ["agent", "read_file"],
+        }
+        if mode is not None:
+            context[RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY] = mode
+        if maximum is not None:
+            context[RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY] = maximum
+        return context
+
+    async def test_recursive_defaults_to_strict_sequential(self):
+        await self.create_root("sequential-root")
+        runner = CountingRunner(self.store)
+        result = await agent(
+            task=["task-1", "task-2", "task-3"],
+            context=self.context(runner, "sequential-root"),
+        )
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(1, runner.max_active)
+        self.assertEqual(["task-1", "task-2", "task-3"], runner.start_order)
+
+    async def test_parallel_mode_never_exceeds_two_and_keeps_input_order(self):
+        await self.create_root("parallel-root")
+        delays = {f"task-{index}": (7 - index) * 0.005 for index in range(1, 7)}
+        runner = CountingRunner(self.store, delays)
+        tasks = [f"task-{index}" for index in range(1, 7)]
+        result = await agent(
+            task=tasks,
+            context=self.context(runner, "parallel-root", "parallel", 2),
+        )
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(2, runner.max_active)
+        self.assertNotEqual(tasks, runner.finish_order)
+        positions = [result["summary"].index(f"result:{task}") for task in tasks]
+        self.assertEqual(sorted(positions), positions)
+        self.assertTrue(all(config.child_execution_mode == "parallel" for config in runner.configs))
+        self.assertTrue(all(config.max_parallel_children == 2 for config in runner.configs))
+
+    async def test_invalid_recursive_settings_fail_before_trace_creation(self):
+        invalid_values = [
+            ("invalid", 2),
+            ("parallel", 0),
+            ("parallel", -1),
+            ("parallel", True),
+            ("parallel", 3),
+        ]
+        for index, (mode, maximum) in enumerate(invalid_values):
+            root_id = f"invalid-{index}"
+            await self.create_root(root_id)
+            runner = CountingRunner(self.store)
+            result = await agent(
+                task=["one", "two"],
+                context=self.context(runner, root_id, mode, maximum),
+            )
+            self.assertEqual("failed", result["status"])
+            children = await self.store.list_traces(parent_trace_id=root_id, limit=10)
+            self.assertEqual([], children)
+
+    async def test_legacy_ignores_new_settings_and_keeps_batch_parallel(self):
+        await self.create_root("legacy-root", mode="legacy")
+        runner = CountingRunner(self.store)
+        result = await agent(
+            task=["one", "two", "three"],
+            context=self.context(runner, "legacy-root", "invalid", 99),
+        )
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(3, runner.max_active)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_recursive_subtree_cancellation.py

@@ -0,0 +1,327 @@
+import asyncio
+import json
+import os
+import tempfile
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from cyber_agent.core.runner import AgentRunner, RunConfig
+from cyber_agent.core.task_protocol import ensure_task_protocol
+from cyber_agent.tools import get_tool_registry
+from cyber_agent.tools.builtin.subagent import _load_or_create_task_report
+from cyber_agent.trace import run_api
+from cyber_agent.trace.models import Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+def brief(index):
+    return {
+        "objective": f"child-{index}",
+        "completion_criteria": [f"finish child-{index}"],
+    }
+
+
+class RecursiveSubtreeCancellationTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+
+    async def asyncTearDown(self):
+        self.temp_dir.cleanup()
+
+    async def test_stop_child_only_cancels_its_descendants(self):
+        runner = AgentRunner(
+            trace_store=self.store,
+            llm_call=AsyncMock(),
+        )
+        root_event = asyncio.Event()
+        runner._cancel_events["root"] = root_event
+        runner._recursive_active_traces["root"] = root_event
+        child_event = runner.register_recursive_child("root", "child")
+        grandchild_event = runner.register_recursive_child("child", "grandchild")
+        sibling_event = runner.register_recursive_child("root", "sibling")
+
+        self.assertTrue(await runner.stop("child"))
+        self.assertTrue(child_event.is_set())
+        self.assertTrue(grandchild_event.is_set())
+        self.assertFalse(root_event.is_set())
+        self.assertFalse(sibling_event.is_set())
+
+    async def test_legacy_stop_keeps_single_trace_behavior(self):
+        runner = AgentRunner(trace_store=self.store, llm_call=AsyncMock())
+        root_event = asyncio.Event()
+        child_event = asyncio.Event()
+        runner._cancel_events.update({"legacy-root": root_event, "legacy-child": child_event})
+        self.assertTrue(await runner.stop("legacy-root"))
+        self.assertTrue(root_event.is_set())
+        self.assertFalse(child_event.is_set())
+
+    async def test_legacy_stop_during_llm_keeps_preexisting_completion_behavior(self):
+        llm_started = asyncio.Event()
+        release_llm = asyncio.Event()
+        late_tool_calls = 0
+
+        async def legacy_late_tool():
+            nonlocal late_tool_calls
+            late_tool_calls += 1
+            return "called"
+
+        registry = get_tool_registry()
+        registry.register(legacy_late_tool, groups=["legacy-cancel-test"])
+
+        async def fake_llm(**kwargs):
+            llm_started.set()
+            await release_llm.wait()
+            return {
+                "content": "legacy completed",
+                "tool_calls": [{
+                    "id": "legacy-late-call",
+                    "type": "function",
+                    "function": {"name": "legacy_late_tool", "arguments": "{}"},
+                }],
+                "finish_reason": "tool_calls",
+            }
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=fake_llm,
+        )
+        with patch.dict(os.environ, {"AGENT_MODE": "legacy"}, clear=False):
+            task = asyncio.create_task(runner.run_result(
+                messages=[{"role": "user", "content": "legacy"}],
+                config=RunConfig(enable_research_flow=False),
+            ))
+            await asyncio.wait_for(llm_started.wait(), timeout=1)
+            root = (await self.store.list_traces(limit=10))[0]
+            self.assertTrue(await runner.stop(root.trace_id))
+            release_llm.set()
+            result = await asyncio.wait_for(task, timeout=2)
+        self.assertEqual("stopped", result["status"])
+        self.assertEqual("legacy completed", result["summary"])
+        self.assertEqual(1, late_tool_calls)
+        self.assertEqual("stopped", (await self.store.get_trace(root.trace_id)).status)
+
+    async def test_stop_during_llm_discards_returned_tool_call(self):
+        dangerous_calls = 0
+        llm_started = asyncio.Event()
+        release_llm = asyncio.Event()
+        registry = get_tool_registry()
+
+        async def fake_llm(**kwargs):
+            llm_started.set()
+            await release_llm.wait()
+            return {
+                "content": "late response",
+                "tool_calls": [{
+                    "id": "late-tool",
+                    "type": "function",
+                    "function": {"name": "dangerous_cancel_test", "arguments": "{}"},
+                }],
+                "finish_reason": "tool_calls",
+            }
+
+        async def dangerous_cancel_test():
+            nonlocal dangerous_calls
+            dangerous_calls += 1
+            return "called"
+
+        registry.register(dangerous_cancel_test, groups=["cancel-test"])
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=fake_llm,
+        )
+
+        with patch.dict(os.environ, {"AGENT_MODE": "recursive"}, clear=False):
+            task = asyncio.create_task(runner.run_result(
+                messages=[{"role": "user", "content": "cancel while llm runs"}],
+                config=RunConfig(
+                    tools=["dangerous_cancel_test"],
+                    tool_groups=[],
+                    enable_research_flow=False,
+                ),
+            ))
+            await asyncio.wait_for(llm_started.wait(), timeout=1)
+            traces = await self.store.list_traces(limit=10)
+            root_id = traces[0].trace_id
+            self.assertTrue(await runner.stop(root_id))
+            release_llm.set()
+            result = await asyncio.wait_for(task, timeout=2)
+
+        self.assertEqual("stopped", result["status"])
+        self.assertEqual("Agent execution stopped.", result["summary"])
+        self.assertEqual(0, dangerous_calls)
+        self.assertEqual("stopped", (await self.store.get_trace(root_id)).status)
+
+    async def test_root_stop_cancels_running_and_queued_children_with_reports(self):
+        child_llm_started = 0
+        two_children_started = asyncio.Event()
+        release_children = asyncio.Event()
+
+        async def fake_llm(**kwargs):
+            nonlocal child_llm_started
+            messages = kwargs["messages"]
+            user_text = "\n".join(
+                str(message.get("content", ""))
+                for message in messages
+                if message.get("role") == "user"
+            )
+            if "root-cancel-batch" in user_text:
+                return {
+                    "content": "",
+                    "tool_calls": [{
+                        "id": "delegate-batch",
+                        "type": "function",
+                        "function": {
+                            "name": "agent",
+                            "arguments": json.dumps({
+                                "task_brief": [brief(index) for index in range(4)],
+                            }),
+                        },
+                    }],
+                    "finish_reason": "tool_calls",
+                }
+
+            child_llm_started += 1
+            if child_llm_started == 2:
+                two_children_started.set()
+            await release_children.wait()
+            return {"content": "late child result", "finish_reason": "stop"}
+
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=get_tool_registry(),
+            llm_call=fake_llm,
+        )
+        with patch.dict(os.environ, {"AGENT_MODE": "recursive"}, clear=False):
+            task = asyncio.create_task(runner.run_result(
+                messages=[{"role": "user", "content": "root-cancel-batch"}],
+                config=RunConfig(
+                    tools=["agent"],
+                    tool_groups=[],
+                    child_execution_mode="parallel",
+                    max_parallel_children=2,
+                    enable_research_flow=False,
+                ),
+            ))
+            await asyncio.wait_for(two_children_started.wait(), timeout=2)
+            traces = await self.store.list_traces(limit=10)
+            root = next(trace for trace in traces if trace.parent_trace_id is None)
+            self.assertTrue(await runner.stop(root.trace_id))
+            release_children.set()
+            result = await asyncio.wait_for(task, timeout=3)
+
+        traces = await self.store.list_traces(limit=10)
+        children = [trace for trace in traces if trace.parent_trace_id == root.trace_id]
+        self.assertEqual(4, len(children))
+        self.assertEqual(2, child_llm_started)
+        self.assertTrue(all(child.status == "stopped" for child in children))
+        self.assertEqual(2, sum(child.total_messages == 0 for child in children))
+        self.assertEqual("stopped", result["status"])
+        persisted_root = await self.store.get_trace(root.trace_id)
+        self.assertEqual("stopped", persisted_root.status)
+        pending = ensure_task_protocol(persisted_root.context)["pending_reviews"]
+        self.assertEqual(sorted(child.trace_id for child in children), list(pending))
+        self.assertTrue(all(
+            entry["task_report"]["outcome"] == "failed"
+            for entry in pending.values()
+        ))
+        self.assertEqual({}, runner._cancel_events)
+        self.assertEqual({}, runner._recursive_active_traces)
+        self.assertEqual({}, runner._active_children)
+        self.assertEqual({}, runner._active_parents)
+
+    async def test_stopped_report_replaces_satisfied_report_only_once(self):
+        child_id = "stale-report-child"
+        context = {
+            "agent_mode": "recursive",
+            "agent_mode_revision": 2,
+            "agent_depth": 1,
+            "root_trace_id": "root",
+        }
+        state = ensure_task_protocol(context)
+        state["task_report"] = {
+            "child_trace_id": child_id,
+            "summary": "previously satisfied",
+            "outcome": "satisfied",
+            "validation": {"hard_passed": True, "open_issues": []},
+            "next_step_suggestion": {
+                "direction": "NONE",
+                "reason": "nothing else",
+            },
+            "outputs": [],
+            "evidence": [],
+            "remaining_issues": [],
+        }
+        await self.store.create_trace(Trace(
+            trace_id=child_id,
+            mode="agent",
+            task="child",
+            parent_trace_id="root",
+            status="stopped",
+            context=context,
+        ))
+
+        for _ in range(2):
+            report = await _load_or_create_task_report(
+                self.store,
+                child_id,
+                "stopped by parent",
+                "stopped",
+                10,
+            )
+            self.assertEqual("failed", report.outcome)
+        persisted = await self.store.get_trace(child_id)
+        self.assertEqual(
+            1,
+            len(ensure_task_protocol(persisted.context)["report_history"]),
+        )
+
+    async def test_stop_api_uses_the_runner_that_started_the_root(self):
+        trace_id = "example-root"
+        actual_runner = AsyncMock()
+        actual_runner.stop.return_value = True
+        run_api._running_runners[trace_id] = actual_runner
+        try:
+            response = await run_api.stop_trace(trace_id)
+        finally:
+            run_api._running_runners.pop(trace_id, None)
+            run_api._running_tasks.pop(trace_id, None)
+        actual_runner.stop.assert_awaited_once_with(trace_id)
+        self.assertEqual("stopping", response.status)
+
+    async def test_immediate_resume_waits_for_old_event_cleanup(self):
+        trace_id = "resume-root"
+        runner = AgentRunner(trace_store=self.store, llm_call=AsyncMock())
+        old_event = asyncio.Event()
+        old_event.set()
+        runner._cancel_events[trace_id] = old_event
+        runner._recursive_active_traces[trace_id] = old_event
+        started = asyncio.Event()
+
+        async def old_run():
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            finally:
+                runner.release_recursive_trace(trace_id, old_event)
+                if run_api._running_runners.get(trace_id) is runner:
+                    run_api._running_runners.pop(trace_id, None)
+
+        old_task = asyncio.create_task(old_run())
+        run_api._running_tasks[trace_id] = old_task
+        run_api._running_runners[trace_id] = runner
+        await started.wait()
+        await run_api._cancel_and_wait_for_running_task(trace_id)
+
+        new_event = runner._cancel_events.setdefault(trace_id, asyncio.Event())
+        self.assertIsNot(old_event, new_event)
+        self.assertFalse(new_event.is_set())
+        self.assertNotIn(trace_id, run_api._running_tasks)
+        self.assertNotIn(trace_id, run_api._running_runners)
+        runner._cancel_events.pop(trace_id, None)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_recursive_tool_capabilities.py

@@ -6,6 +6,7 @@ from unittest.mock import patch
 
 from cyber_agent.core.agent_mode import RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY
 from cyber_agent.core.runner import AgentRunner, RunConfig
+from cyber_agent.core.task_protocol import ensure_task_protocol
 from cyber_agent.tools.builtin.subagent import agent
 from cyber_agent.tools.registry import ToolRegistry
 from cyber_agent.trace.goal_models import GoalTree
@@ -112,6 +113,85 @@ class RecursiveCapabilityInheritanceTest(unittest.IsolatedAsyncioTestCase):
 
 
 class RecursiveRuntimeDispatchTest(unittest.IsolatedAsyncioTestCase):
+    async def test_recursive_framework_context_fields_cannot_be_overridden(self):
+        with tempfile.TemporaryDirectory() as directory:
+            store = FileSystemTraceStore(directory)
+            runner = AgentRunner(trace_store=store, llm_call=lambda **kwargs: None)
+            trace = Trace(
+                trace_id="real-trace",
+                mode="agent",
+                task="test",
+                context={"agent_mode": "recursive", "agent_mode_revision": 2},
+            )
+            goal_tree = GoalTree(mission="test")
+            config = RunConfig(context={
+                "store": "spoof-store",
+                "trace_id": "spoof-trace",
+                "goal_id": "spoof-goal",
+                "runner": "spoof-runner",
+                "goal_tree": "spoof-tree",
+                "sequence": -1,
+                RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY: ["spoof-tool"],
+            })
+            context = runner._build_tool_context(
+                config=config,
+                trace=trace,
+                trace_id=trace.trace_id,
+                goal_id="real-goal",
+                goal_tree=goal_tree,
+                sequence=7,
+                side_branch_ctx=None,
+                trigger_event=None,
+            )
+            self.assertIs(store, context["store"])
+            self.assertEqual("real-trace", context["trace_id"])
+            self.assertEqual("real-goal", context["goal_id"])
+            self.assertIs(runner, context["runner"])
+            self.assertIs(goal_tree, context["goal_tree"])
+            self.assertEqual(7, context["sequence"])
+            self.assertNotEqual(
+                ["spoof-tool"],
+                context[RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY],
+            )
+
+    async def test_pending_review_runtime_allowlist_rejects_forged_agent(self):
+        registry = ToolRegistry()
+        agent_calls = 0
+
+        async def agent():
+            nonlocal agent_calls
+            agent_calls += 1
+            return "called"
+
+        async def review_task_result():
+            return "reviewed"
+
+        registry.register(agent)
+        registry.register(review_task_result)
+        runner = AgentRunner(tool_registry=registry, llm_call=lambda **kwargs: None)
+        trace = Trace(
+            trace_id="pending-root",
+            mode="agent",
+            task="pending",
+            context={"agent_mode": "recursive", "agent_mode_revision": 2},
+        )
+        state = ensure_task_protocol(trace.context)
+        state["pending_reviews"]["child"] = {"task_report": {}}
+        config = RunConfig(
+            tools=["agent", "review_task_result"],
+            tool_groups=[],
+        )
+        allowed = runner._get_runtime_tool_names(config, trace)
+        self.assertEqual({"review_task_result"}, allowed)
+        result = json.loads(await registry.execute(
+            "agent",
+            {},
+            allowed_tool_names=allowed,
+        ))
+        self.assertIn("not allowed", result["error"])
+        self.assertEqual(0, agent_calls)
+        self.assertEqual(0, registry.get_stats("agent")["agent"]["call_count"])
+
     async def test_registry_rejects_before_counting_or_calling(self):
         registry = ToolRegistry()
         calls = 0