实现 Sub-Agent 五层递归与可视化钻取

.env.template

@@ -5,6 +5,9 @@ BROWSER_USE_API_KEY=
 QWEN_BASE_URL=
 QWEN_API_KEY=
 
+# 本地 Sub-Agent 递归（默认关闭；开启后固定最多 5 层、每个父 Agent 最多 6 个直接孩子）
+AGENT_RECURSION_ENABLED=false
+
 # sonnet模型
 OPEN_ROUTER_API_KEY=
 # postgreSQL database配置

cyber_agent/docs/tools.md

@@ -941,8 +941,8 @@ async def agent(
 
 | task 类型 | 模式 | 并行执行 | 工具权限 |
 |-----------|------|---------|---------|
-| `str`（单任务） | delegate | ❌ | 完整（除 agent/evaluate 外） |
-| `List[str]`（多任务） | explore | ✅ | 只读（read_file, grep_content, glob_files, goal） |
+| `str`（单任务） | delegate | ❌ | 完整（递归关闭时除 agent/evaluate；递归开启时按深度开放 agent） |
+| `List[str]`（多任务） | explore | ✅ | 沿用现有 Sub-Agent 工具集，并按递归深度控制 `agent` |
 
 **messages 参数**：
 - `None`：无预置消息
@@ -952,7 +952,25 @@ async def agent(
 运行时判断：`messages[0]` 是 dict → 1D 共享；是 list → 2D per-agent。
 
 **单任务（delegate）**：适合委托专门任务，完整工具权限；支持 `continue_from` 续跑已有 Sub-Trace。
-**多任务（explore）**：适合对比多个方案，并行执行，只读权限，不支持 `continue_from`。
+**多任务（explore）**：适合对比多个方案，并行执行，不支持 `continue_from`。
+
+#### 本地 Sub-Agent 递归实验
+
+递归默认关闭。在运行环境设置以下变量后，本地 Sub-Agent
+可继续调用 `agent` 创建下一层本地 Sub-Agent：
+
+```bash
+AGENT_RECURSION_ENABLED=true
+```
+
+实验版限制为固定值，不由模型或 `RunConfig` 放大：
+
+- 主 Agent 是 `depth=0`；本地 Sub-Agent 最深到 `depth=5`。
+- `depth=5` 不再获得 `agent` 工具，运行时入口也会拒绝继续新建。
+- 主 Agent 和每层本地 Sub-Agent 都最多累计创建 6 个直接本地孩子。
+- 已创建后失败或停止的孩子仍占名额；`continue_from` 续跑已有孩子不占新名额。
+- `agent_type="remote_*"` 保持现有路由行为，不计入本地深度和孩子配额。
+- `evaluate` 和 `bash_command` 不向递归 Sub-Agent 开放。
 
 #### 远端模式
 

cyber_agent/tools/builtin/subagent.py

@@ -22,6 +22,12 @@ from cyber_agent.trace.websocket import broadcast_sub_trace_started, broadcast_s
 # ===== 远端路由常量 =====
 
 REMOTE_PREFIX = "remote_"
+MAX_AGENT_DEPTH = 5
+MAX_LOCAL_CHILDREN_PER_TRACE = 6
+
+# POC 阶段使用进程内锁保护“检查配额 -> 创建 Trace”。
+# 多 worker 部署时需要换成跨进程的原子配额。
+_LOCAL_AGENT_SPAWN_LOCK = asyncio.Lock()
 
 
 def _knowhub_api() -> str:
@@ -33,6 +39,13 @@ def _remote_agent_timeout() -> float:
     return float(os.getenv("REMOTE_AGENT_TIMEOUT", "600"))
 
 
+def _recursion_enabled() -> bool:
+    """是否允许本地 Sub-Agent 继续创建 Sub-Agent。"""
+    return os.getenv("AGENT_RECURSION_ENABLED", "false").strip().lower() in {
+        "1", "true", "yes", "on",
+    }
+
+
 # 兼容旧代码对 module-level 常量的引用（运行时值 = 首次 import 时的快照）
 KNOWHUB_API = _knowhub_api()
 REMOTE_AGENT_TIMEOUT = _remote_agent_timeout()
@@ -143,23 +156,35 @@ async def _update_collaborator(
 
 
 async def _update_goal_start(
-    store, trace_id: str, goal_id: str, mode: str, sub_trace_ids: List[str]
+    store, trace_id: str, goal_id: str, mode: str,
+    sub_trace_ids: List[Dict[str, str]],
 ) -> None:
     """标记 Goal 开始执行"""
     if not goal_id:
         return
+
+    tree = await store.get_goal_tree(trace_id)
+    goal = tree.find(goal_id) if tree else None
+    existing_entries = goal.sub_trace_ids if goal and goal.sub_trace_ids else []
+    merged_entries: Dict[str, Dict[str, str]] = {}
+    for entry in [*existing_entries, *sub_trace_ids]:
+        if isinstance(entry, str):
+            merged_entries[entry] = {"trace_id": entry, "mission": entry}
+        elif isinstance(entry, dict) and entry.get("trace_id"):
+            merged_entries[entry["trace_id"]] = entry
+
     await store.update_goal(
         trace_id, goal_id,
         type="agent_call",
         agent_call_mode=mode,
         status="in_progress",
-        sub_trace_ids=sub_trace_ids
+        sub_trace_ids=list(merged_entries.values()),
     )
 
 
 async def _update_goal_complete(
     store, trace_id: str, goal_id: str,
-    status: str, summary: str, sub_trace_ids: List[str]
+    status: str, summary: str,
 ) -> None:
     """标记 Goal 完成"""
     if not goal_id:
@@ -168,7 +193,6 @@ async def _update_goal_complete(
         trace_id, goal_id,
         status=status,
         summary=summary,
-        sub_trace_ids=sub_trace_ids
     )
 
 
@@ -192,15 +216,50 @@ def _aggregate_stats(results: List[Dict[str, Any]]) -> Dict[str, Any]:
     }
 
 
-def _get_allowed_tools(single: bool, context: dict) -> Optional[List[str]]:
-    """获取允许工具列表。获取所有工具，排除 agent 和 evaluate"""
+def _get_allowed_tools(context: dict, agent_depth: int) -> Optional[List[str]]:
+    """按子 Agent 深度生成唯一的工具权限列表。"""
     runner = context.get("runner")
-    if runner and hasattr(runner, "tools") and hasattr(runner.tools, "_tools"):
-        all_tools = list(runner.tools._tools.keys())
-        return [t for t in all_tools if t not in ("agent", "evaluate")]
+    if runner and hasattr(runner, "tools") and hasattr(runner.tools, "get_tool_names"):
+        blocked_tools = {"evaluate", "bash_command"}
+        if not _recursion_enabled() or agent_depth >= MAX_AGENT_DEPTH:
+            blocked_tools.add("agent")
+        return [
+            name for name in runner.tools.get_tool_names()
+            if name not in blocked_tools
+        ]
     return None
 
 
+async def _resolve_trace_lineage(store, trace: Trace) -> tuple[int, str]:
+    """返回 Trace 的 (agent_depth, root_trace_id)。
+
+    新 Trace 直接读 context；旧 Trace 缺少字段时沿 parent_trace_id
+    回溯，不依赖 Trace ID 字符串格式。
+    """
+    depth = trace.context.get("agent_depth")
+    root_trace_id = trace.context.get("root_trace_id")
+    if isinstance(depth, int) and depth >= 0 and isinstance(root_trace_id, str):
+        return depth, root_trace_id
+
+    depth = 0
+    current = trace
+    root_trace_id = trace.trace_id
+    visited = {trace.trace_id}
+    while current.parent_trace_id:
+        parent_id = current.parent_trace_id
+        if parent_id in visited:
+            break
+        visited.add(parent_id)
+        parent = await store.get_trace(parent_id)
+        if not parent:
+            break
+        depth += 1
+        root_trace_id = parent.trace_id
+        current = parent
+
+    return depth, root_trace_id
+
+
 def _format_single_result(result: Dict[str, Any], sub_trace_id: str, continued: bool) -> Dict[str, Any]:
     """格式化单任务（delegate）结果"""
     lines = [DELEGATE_RESULT_HEADER]
@@ -419,69 +478,149 @@ async def _run_agents(
     """
     统一 agent 执行逻辑。
 
-    single (len(tasks)==1): delegate 模式，全量工具（排除 agent/evaluate）
-    multi (len(tasks)>1): explore 模式，只读工具，并行执行
+    single (len(tasks)==1): delegate 模式，全量工具
+    multi (len(tasks)>1): explore 模式，并行执行
+
+    AGENT_RECURSION_ENABLED=true 时，depth < MAX_AGENT_DEPTH 的本地
+    Sub-Agent 可继续使用 agent 工具。
     """
     single = len(tasks) == 1
     parent_trace = await store.get_trace(trace_id)
+    if not parent_trace:
+        return {"status": "failed", "error": f"Parent trace not found: {trace_id}"}
 
     # continue_from: 复用已有 trace（仅 single）
     sub_trace_id = None
     continued = False
+    goal_started = False
+    child_records: List[Dict[str, Any]] = []
+    all_sub_trace_ids: List[Dict[str, str]] = []
+
     if single and continue_from:
         existing = await store.get_trace(continue_from)
         if not existing:
             return {"status": "failed", "error": f"Continue-from trace not found: {continue_from}"}
+        if existing.parent_trace_id != trace_id:
+            return {
+                "status": "failed",
+                "error": "continue_from must reference a direct child of the current trace",
+            }
+        if existing.uid != parent_trace.uid:
+            return {
+                "status": "failed",
+                "error": "continue_from trace owner does not match the current trace owner",
+            }
+
         sub_trace_id = continue_from
         continued = True
         goal_tree = await store.get_goal_tree(continue_from)
         mission = goal_tree.mission if goal_tree else tasks[0]
-        sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
+        child_depth, root_trace_id = await _resolve_trace_lineage(store, existing)
+        all_sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
+        child_records = [{
+            "trace_id": sub_trace_id,
+            "depth": child_depth,
+            "context": {
+                **existing.context,
+                "agent_depth": child_depth,
+                "root_trace_id": root_trace_id,
+            },
+            "is_new": False,
+        }]
     else:
-        sub_trace_ids = []
-
-    # 创建 sub-traces 和执行协程
-    coros = []
-    all_sub_trace_ids = list(sub_trace_ids)  # copy for continue_from case
+        parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
+        if parent_depth > 0 and not _recursion_enabled():
+            return {
+                "status": "failed",
+                "error": "Local Sub-Agent recursion is disabled",
+            }
+        if parent_depth >= MAX_AGENT_DEPTH:
+            return {
+                "status": "failed",
+                "error": (
+                    f"Local Sub-Agent depth limit reached: "
+                    f"depth={parent_depth}, max={MAX_AGENT_DEPTH}"
+                ),
+            }
 
-    for i, (task_item, msgs) in enumerate(zip(tasks, per_agent_msgs)):
-        if single and continued:
-            # continue_from 已经设置了 sub_trace_id
-            pass
-        else:
-            resolved_agent_type = agent_type or ("delegate" if single else "explore")
-            suffix = "delegate" if single else f"explore-{i+1:03d}"
-            stid = generate_sub_trace_id(trace_id, suffix)
-
-            sub_trace = Trace(
-                trace_id=stid,
-                mode="agent",
-                task=task_item,
+        # 将配额检查和 Trace 创建放在同一进程内临界区，
+        # 保证两个并发 agent() 调用不会共同突破 6 个直接孩子。
+        async with _LOCAL_AGENT_SPAWN_LOCK:
+            existing_children = await store.list_traces(
                 parent_trace_id=trace_id,
-                parent_goal_id=goal_id,
-                agent_type=resolved_agent_type,
-                uid=parent_trace.uid if parent_trace else None,
-                model=parent_trace.model if parent_trace else None,
-                status="running",
-                context={"created_by_tool": "agent"},
-                created_at=datetime.now(),
+                created_by_tool="agent",
+                limit=MAX_LOCAL_CHILDREN_PER_TRACE + 1,
             )
-            await store.create_trace(sub_trace)
-            await store.update_goal_tree(stid, GoalTree(mission=task_item))
+            requested_children = len(tasks)
+            if len(existing_children) + requested_children > MAX_LOCAL_CHILDREN_PER_TRACE:
+                return {
+                    "status": "failed",
+                    "error": (
+                        f"Local child Agent limit exceeded: existing={len(existing_children)}, "
+                        f"requested={requested_children}, "
+                        f"max={MAX_LOCAL_CHILDREN_PER_TRACE}"
+                    ),
+                }
 
-            all_sub_trace_ids.append({"trace_id": stid, "mission": task_item})
+            child_depth = parent_depth + 1
+            for i, task_item in enumerate(tasks):
+                resolved_agent_type = agent_type or ("delegate" if single else "explore")
+                suffix = "delegate" if single else f"explore-{i+1:03d}"
+                stid = generate_sub_trace_id(trace_id, suffix)
+                child_context = {
+                    "created_by_tool": "agent",
+                    "agent_depth": child_depth,
+                    "root_trace_id": root_trace_id,
+                }
+                sub_trace = Trace(
+                    trace_id=stid,
+                    mode="agent",
+                    task=task_item,
+                    parent_trace_id=trace_id,
+                    parent_goal_id=goal_id,
+                    agent_type=resolved_agent_type,
+                    uid=parent_trace.uid,
+                    model=parent_trace.model,
+                    status="running",
+                    context=child_context,
+                    created_at=datetime.now(),
+                )
+                await store.create_trace(sub_trace)
+                await store.update_goal_tree(stid, GoalTree(mission=task_item))
+                all_sub_trace_ids.append({"trace_id": stid, "mission": task_item})
+                child_records.append({
+                    "trace_id": stid,
+                    "depth": child_depth,
+                    "context": child_context,
+                    "is_new": True,
+                    "agent_type": resolved_agent_type,
+                })
 
-            # 广播 sub_trace_started
-            await broadcast_sub_trace_started(
-                trace_id, stid, goal_id or "",
-                resolved_agent_type, task_item,
+            if single:
+                sub_trace_id = child_records[0]["trace_id"]
+
+            await _update_goal_start(
+                store, trace_id, goal_id,
+                "delegate" if single else "explore",
+                all_sub_trace_ids,
             )
+            goal_started = True
 
-            if single:
-                sub_trace_id = stid
+    # 创建执行协程。Trace 在上面已经按批次预创建，
+    # 因此孩子配额不会被并发调用绕过。
+    coros = []
+    for i, (task_item, msgs, child_record) in enumerate(
+        zip(tasks, per_agent_msgs, child_records)
+    ):
+        cur_stid = child_record["trace_id"]
+        child_depth = child_record["depth"]
+        if child_record["is_new"]:
+            await broadcast_sub_trace_started(
+                trace_id, cur_stid, goal_id or "",
+                child_record["agent_type"], task_item,
+            )
 
         # 注册为活跃协作者
-        cur_stid = sub_trace_id if single else all_sub_trace_ids[-1]["trace_id"]
         collab_name = task_item[:30] if single and not continued else (
             f"delegate-{cur_stid[:8]}" if single else f"explore-{i+1}"
         )
@@ -493,12 +632,12 @@ async def _run_agents(
 
         # 构建消息
         agent_msgs = list(msgs) + [{"role": "user", "content": task_item}]
-        allowed_tools = _get_allowed_tools(single, context)
+        allowed_tools = _get_allowed_tools(context, child_depth)
 
         debug = getattr(runner, 'debug', False)
         agent_label = (agent_type or ("delegate" if single else f"explore-{i+1}"))
         debug_printer = _make_event_printer(agent_label) if debug else None
-        # 为了彻底封死逃逸风险，禁用由于缺省 tool_groups 注入进来的 agent 和 bash_command
+        # allowed_tools 已是当前深度的精确白名单。
         on_event = _make_interactive_handler(
             runner, cur_stid, trace_id, debug_printer=debug_printer
         )
@@ -512,22 +651,23 @@ async def _run_agents(
                 model=parent_trace.model if parent_trace else "gpt-4o",
                 uid=parent_trace.uid if parent_trace else None,
                 tools=allowed_tools,
-                tool_groups=[], # 清空默认的 "core" 组，防止把 agent 和 bash_command 偷渡进来
-                exclude_tools=["agent", "evaluate", "bash_command"], # 严格锁死危险元工具
+                tool_groups=[],  # tools 是精确白名单，不再合并默认 core 组
                 name=task_item[:50],
                 skills=skills,
                 knowledge=context.get("knowledge_config"),
+                context=child_record["context"],
             ),
             on_event=on_event,
         )
         coros.append((i, cur_stid, collab_name, coro))
 
-    # 更新主 Goal 为 in_progress
-    await _update_goal_start(
-        store, trace_id, goal_id,
-        "delegate" if single else "explore",
-        all_sub_trace_ids,
-    )
+    # continue_from 不进入新建临界区，但仍需要恢复 Goal 的运行状态。
+    if not goal_started:
+        await _update_goal_start(
+            store, trace_id, goal_id,
+            "delegate" if single else "explore",
+            all_sub_trace_ids,
+        )
 
     # 执行
     if single:
@@ -555,7 +695,6 @@ async def _run_agents(
                 store, trace_id, goal_id,
                 result.get("status", "completed"),
                 formatted["summary"],
-                all_sub_trace_ids,
             )
             return formatted
 
@@ -572,7 +711,6 @@ async def _run_agents(
             await _update_goal_complete(
                 store, trace_id, goal_id,
                 "failed", f"委托任务失败: {error_msg}",
-                all_sub_trace_ids,
             )
             return {
                 "mode": "delegate",
@@ -639,7 +777,6 @@ async def _run_agents(
             store, trace_id, goal_id,
             formatted["status"],
             formatted["summary"],
-            all_sub_trace_ids,
         )
         return formatted
 
@@ -790,6 +927,14 @@ async def agent(
     if messages is None:
         per_agent_msgs: List[Messages] = [[] for _ in tasks]
     elif messages and isinstance(messages[0], list):
+        if len(messages) != len(tasks):
+            return {
+                "status": "failed",
+                "error": (
+                    "2D messages must contain exactly one message list per task: "
+                    f"tasks={len(tasks)}, messages={len(messages)}"
+                ),
+            }
         per_agent_msgs = messages  # 2D: per-agent
     else:
         per_agent_msgs = [messages] * len(tasks)  # 1D: 共享
@@ -910,6 +1055,8 @@ async def evaluate(
                 model=parent_trace.model if parent_trace else "gpt-4o",
                 uid=parent_trace.uid if parent_trace else None,
                 tools=allowed_tools,
+                tool_groups=[],
+                exclude_tools=["agent", "evaluate", "bash_command"],
                 name=f"评估: {goal_id}",
             ),
             on_event=_make_interactive_handler(
@@ -937,7 +1084,6 @@ async def evaluate(
             store, trace_id, current_goal_id,
             result.get("status", "completed"),
             formatted_summary,
-            sub_trace_ids,
         )
 
         return {
@@ -961,7 +1107,6 @@ async def evaluate(
         await _update_goal_complete(
             store, trace_id, current_goal_id,
             "failed", f"评估任务失败: {error_msg}",
-            sub_trace_ids,
         )
         return {
             "mode": "evaluate",

cyber_agent/trace/protocols.py

@@ -48,7 +48,9 @@ class TraceStore(Protocol):
         agent_type: Optional[str] = None,
         uid: Optional[str] = None,
         status: Optional[str] = None,
-        limit: int = 50
+        limit: int = 50,
+        parent_trace_id: Optional[str] = None,
+        created_by_tool: Optional[str] = None,
     ) -> List[Trace]:
         """列出 Traces"""
         ...

cyber_agent/trace/store.py

@@ -126,7 +126,9 @@ class FileSystemTraceStore:
         agent_type: Optional[str] = None,
         uid: Optional[str] = None,
         status: Optional[str] = None,
-        limit: int = 50
+        limit: int = 50,
+        parent_trace_id: Optional[str] = None,
+        created_by_tool: Optional[str] = None,
     ) -> List[Trace]:
         """列出 Traces"""
         traces = []
@@ -154,6 +156,10 @@ class FileSystemTraceStore:
                     continue
                 if status and data.get("status") != status:
                     continue
+                if parent_trace_id and data.get("parent_trace_id") != parent_trace_id:
+                    continue
+                if created_by_tool and data.get("context", {}).get("created_by_tool") != created_by_tool:
+                    continue
 
                 # 解析 datetime
                 if data.get("created_at"):

cyber_agent/trace/trace_id.py

@@ -83,7 +83,7 @@ def parse_parent_trace_id(trace_id: str) -> Optional[str]:
         None
     """
     if '@' in trace_id:
-        return trace_id.split('@')[0]
+        return trace_id.rsplit('@', 1)[0]
     return None
 
 
@@ -131,7 +131,7 @@ def extract_mode(trace_id: str) -> Optional[str]:
         return None
 
     # 格式: parent@mode-timestamp-seq
-    parts = trace_id.split('@')[1]  # "mode-timestamp-seq"
+    parts = trace_id.rsplit('@', 1)[1]  # "mode-timestamp-seq"
     mode = parts.split('-')[0]
     return mode
 

frontend/react-template/src/components/FlowChart/FlowChart.tsx

@@ -23,6 +23,16 @@ export interface FlowChartRef {
 
 export type SubTraceEntry = { id: string; mission?: string };
 
+const normalizeSubTraceEntries = (goal: Goal): SubTraceEntry[] =>
+  (goal.sub_trace_ids ?? [])
+    .map((item) => {
+      if (typeof item === "string") return { id: item };
+      if (!item || typeof item.trace_id !== "string") return null;
+      return { id: item.trace_id, mission: item.mission };
+    })
+    .filter((entry): entry is SubTraceEntry => entry !== null && entry.id.length > 0)
+    .slice(0, 6);
+
 interface LayoutNode {
   id: string;
   x: number;
@@ -49,7 +59,7 @@ interface LayoutEdge {
 }
 
 const FlowChartComponent: ForwardRefRenderFunction<FlowChartRef, FlowChartProps> = (
-  { goals, msgGroups = {}, invalidBranches, onNodeClick },
+  { goals, msgGroups = {}, invalidBranches, onNodeClick, onSubTraceClick },
   ref,
 ) => {
   const displayGoals = useMemo(() => {
@@ -1091,6 +1101,7 @@ const FlowChartComponent: ForwardRefRenderFunction<FlowChartRef, FlowChartProps>
 
                 const isGoal = node.type === "goal";
                 const data = node.data as Goal;
+                const subTraceEntries = isGoal ? normalizeSubTraceEntries(data) : [];
                 let text = isGoal ? data.description : (node.data as Message).description || "";
 
                 let thumbnail: string | null = null;
@@ -1220,6 +1231,43 @@ const FlowChartComponent: ForwardRefRenderFunction<FlowChartRef, FlowChartProps>
                         </text>
                       </g>
                     )}
+                    {subTraceEntries.map((entry, index) => {
+                      const spacing = 22;
+                      const startX = -((subTraceEntries.length - 1) * spacing) / 2;
+                      return (
+                        <g
+                          key={entry.id}
+                          transform={`translate(${startX + index * spacing}, 36)`}
+                          style={{ cursor: "pointer" }}
+                          onClick={(event) => {
+                            event.stopPropagation();
+                            onSubTraceClick?.(data, entry);
+                          }}
+                        >
+                          <title>{entry.mission || entry.id}</title>
+                          <rect
+                            x={-10}
+                            y={-8}
+                            width={20}
+                            height={16}
+                            rx={5}
+                            fill="#fffbeb"
+                            stroke="#f59e0b"
+                            strokeWidth={1}
+                          />
+                          <text
+                            x={0}
+                            y={4}
+                            fontSize={9}
+                            fontWeight="bold"
+                            fill="#b45309"
+                            textAnchor="middle"
+                          >
+                            A{index + 1}
+                          </text>
+                        </g>
+                      );
+                    })}
                   </g>
                 );
               })}

frontend/react-template/src/components/FlowChart/hooks/useFlowChartData.ts

@@ -342,7 +342,7 @@ export const useFlowChartData = (traceId: string | null, refreshTrigger?: number
           {};
         if (!goalId) return;
         setGoals((prev: Goal[]) =>
-          prev.map((g: Goal) => {
+          buildSubGoals(prev.map((g: Goal) => {
             if (g.id !== goalId) return g;
             const next: Goal = { ...g };
             if ("status" in updates) {
@@ -357,8 +357,14 @@ export const useFlowChartData = (traceId: string | null, refreshTrigger?: number
                 next.summary = summary;
               }
             }
+            if (Array.isArray(updates.sub_trace_ids)) {
+              next.sub_trace_ids = updates.sub_trace_ids as Goal["sub_trace_ids"];
+            }
+            if (typeof updates.agent_call_mode === "string") {
+              next.agent_call_mode = updates.agent_call_mode;
+            }
             return next;
-          }),
+          })),
         );
         return;
       }

tests/test_subagent_recursion.py

@@ -0,0 +1,303 @@
+import asyncio
+import os
+import tempfile
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from cyber_agent.tools.builtin.subagent import agent, evaluate
+from cyber_agent.trace.goal_models import Goal, GoalTree
+from cyber_agent.trace.models import Trace
+from cyber_agent.trace.store import FileSystemTraceStore
+
+
+class FakeToolRegistry:
+    TOOL_NAMES = [
+        "agent",
+        "evaluate",
+        "bash_command",
+        "read_file",
+        "grep_content",
+        "glob_files",
+        "goal",
+    ]
+
+    def get_tool_names(self):
+        return list(self.TOOL_NAMES)
+
+
+class FakeRunner:
+    def __init__(self, store):
+        self.store = store
+        self.tools = FakeToolRegistry()
+        self.debug = False
+        self.stdin_check = None
+        self._cancel_events = {}
+        self.configs = []
+
+    async def run_result(self, messages, config, on_event=None):
+        self.configs.append(config)
+        await self.store.update_trace(config.trace_id, status="completed")
+        return {
+            "status": "completed",
+            "summary": f"completed: {config.trace_id}",
+            "saved_knowledge_ids": [],
+            "stats": {
+                "total_messages": 1,
+                "total_tokens": 1,
+                "total_cost": 0.0,
+            },
+        }
+
+
+class SubAgentRecursionTest(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self):
+        self.temp_dir = tempfile.TemporaryDirectory()
+        self.store = FileSystemTraceStore(self.temp_dir.name)
+        self.runner = FakeRunner(self.store)
+        self.previous_recursion_setting = os.environ.get("AGENT_RECURSION_ENABLED")
+        os.environ["AGENT_RECURSION_ENABLED"] = "true"
+        self.root_id = await self._create_root("root")
+
+    async def asyncTearDown(self):
+        if self.previous_recursion_setting is None:
+            os.environ.pop("AGENT_RECURSION_ENABLED", None)
+        else:
+            os.environ["AGENT_RECURSION_ENABLED"] = self.previous_recursion_setting
+        self.temp_dir.cleanup()
+
+    async def _create_root(self, trace_id, uid="user-1"):
+        trace = Trace(
+            trace_id=trace_id,
+            mode="agent",
+            task=f"task-{trace_id}",
+            uid=uid,
+            model="fake-model",
+            context={},
+        )
+        await self.store.create_trace(trace)
+        await self.store.update_goal_tree(trace_id, GoalTree(mission=trace.task))
+        return trace_id
+
+    def _context(self, trace_id, goal_id=None):
+        return {
+            "store": self.store,
+            "trace_id": trace_id,
+            "goal_id": goal_id,
+            "runner": self.runner,
+            "knowledge_config": None,
+        }
+
+    async def _spawn_one(self, parent_id, task="child"):
+        return await agent(task=task, context=self._context(parent_id))
+
+    async def _local_children(self, parent_id):
+        return await self.store.list_traces(
+            parent_trace_id=parent_id,
+            created_by_tool="agent",
+            limit=20,
+        )
+
+    async def test_recursion_disabled_preserves_single_level_behavior(self):
+        os.environ["AGENT_RECURSION_ENABLED"] = "false"
+        first = await self._spawn_one(self.root_id)
+        self.assertEqual("completed", first["status"])
+
+        child_id = first["sub_trace_id"]
+        child_config = self.runner.configs[-1]
+        self.assertNotIn("agent", child_config.tools)
+
+        blocked = await self._spawn_one(child_id, "grandchild")
+        self.assertEqual("failed", blocked["status"])
+        self.assertIn("recursion is disabled", blocked["error"])
+
+    async def test_five_subagent_levels_and_sixth_level_rejected(self):
+        parent_id = self.root_id
+        for expected_depth in range(1, 6):
+            result = await self._spawn_one(parent_id, f"depth-{expected_depth}")
+            self.assertEqual("completed", result["status"])
+            child_id = result["sub_trace_id"]
+            child = await self.store.get_trace(child_id)
+            self.assertEqual(parent_id, child.parent_trace_id)
+            self.assertEqual(expected_depth, child.context["agent_depth"])
+            self.assertEqual(self.root_id, child.context["root_trace_id"])
+
+            tools = self.runner.configs[-1].tools
+            if expected_depth < 5:
+                self.assertIn("agent", tools)
+            else:
+                self.assertNotIn("agent", tools)
+            self.assertNotIn("evaluate", tools)
+            self.assertNotIn("bash_command", tools)
+            parent_id = child_id
+
+        blocked = await self._spawn_one(parent_id, "depth-6")
+        self.assertEqual("failed", blocked["status"])
+        self.assertIn("depth limit reached", blocked["error"])
+        self.assertEqual([], await self._local_children(parent_id))
+
+    async def test_six_children_allowed_and_seventh_rejected(self):
+        tasks = [f"child-{index}" for index in range(6)]
+        result = await agent(task=tasks, context=self._context(self.root_id))
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(6, len(await self._local_children(self.root_id)))
+
+        blocked = await self._spawn_one(self.root_id, "child-7")
+        self.assertEqual("failed", blocked["status"])
+        self.assertIn("child Agent limit exceeded", blocked["error"])
+        self.assertEqual(6, len(await self._local_children(self.root_id)))
+
+    async def test_child_batch_is_rejected_without_partial_creation(self):
+        initial = [f"child-{index}" for index in range(5)]
+        await agent(task=initial, context=self._context(self.root_id))
+
+        blocked = await agent(
+            task=["overflow-1", "overflow-2"],
+            context=self._context(self.root_id),
+        )
+        self.assertEqual("failed", blocked["status"])
+        self.assertEqual(5, len(await self._local_children(self.root_id)))
+
+    async def test_mismatched_per_agent_messages_are_rejected_before_spawn(self):
+        result = await agent(
+            task=["first", "second"],
+            messages=[[{"role": "user", "content": "only one message list"}]],
+            context=self._context(self.root_id),
+        )
+        self.assertEqual("failed", result["status"])
+        self.assertIn("exactly one message list per task", result["error"])
+        self.assertEqual([], await self._local_children(self.root_id))
+
+    async def test_goal_keeps_children_created_across_multiple_calls(self):
+        tree = await self.store.get_goal_tree(self.root_id)
+        tree.goals = [Goal(id="1", description="delegate twice")]
+        await self.store.update_goal_tree(self.root_id, tree)
+        context = self._context(self.root_id, goal_id="1")
+
+        await agent(task="first", context=context)
+        await agent(task="second", context=context)
+
+        updated_tree = await self.store.get_goal_tree(self.root_id)
+        sub_trace_ids = updated_tree.find("1").sub_trace_ids
+        self.assertEqual(2, len(sub_trace_ids))
+        self.assertEqual(2, len({entry["trace_id"] for entry in sub_trace_ids}))
+
+    async def test_concurrent_batches_never_exceed_limit(self):
+        results = await asyncio.gather(
+            agent(
+                task=[f"left-{index}" for index in range(4)],
+                context=self._context(self.root_id),
+            ),
+            agent(
+                task=[f"right-{index}" for index in range(4)],
+                context=self._context(self.root_id),
+            ),
+        )
+        self.assertEqual({"completed", "failed"}, {item["status"] for item in results})
+        self.assertLessEqual(len(await self._local_children(self.root_id)), 6)
+
+    async def test_continue_from_does_not_consume_child_slot(self):
+        tasks = [f"child-{index}" for index in range(6)]
+        created = await agent(task=tasks, context=self._context(self.root_id))
+        child_id = created["sub_trace_ids"][0]["trace_id"]
+
+        continued = await agent(
+            task="continue existing child",
+            continue_from=child_id,
+            context=self._context(self.root_id),
+        )
+        self.assertEqual("completed", continued["status"])
+        self.assertTrue(continued["continue_from"])
+        self.assertEqual(6, len(await self._local_children(self.root_id)))
+
+    async def test_continue_from_requires_direct_child_and_matching_owner(self):
+        created = await self._spawn_one(self.root_id)
+        child_id = created["sub_trace_id"]
+        other_root = await self._create_root("other-root")
+
+        wrong_parent = await agent(
+            task="continue",
+            continue_from=child_id,
+            context=self._context(other_root),
+        )
+        self.assertEqual("failed", wrong_parent["status"])
+        self.assertIn("direct child", wrong_parent["error"])
+
+        await self.store.update_trace(child_id, uid="other-user")
+        wrong_owner = await agent(
+            task="continue",
+            continue_from=child_id,
+            context=self._context(self.root_id),
+        )
+        self.assertEqual("failed", wrong_owner["status"])
+        self.assertIn("owner", wrong_owner["error"])
+
+    async def test_evaluator_never_receives_recursive_or_shell_tools(self):
+        await evaluate(messages=[], context=self._context(self.root_id))
+        config = self.runner.configs[-1]
+        self.assertEqual([], config.tool_groups)
+        self.assertEqual(
+            ["agent", "evaluate", "bash_command"],
+            config.exclude_tools,
+        )
+
+    async def test_evaluator_traces_do_not_consume_local_child_slots(self):
+        for _ in range(2):
+            result = await evaluate(messages=[], context=self._context(self.root_id))
+            self.assertEqual("completed", result["status"])
+
+        result = await agent(
+            task=[f"child-{index}" for index in range(6)],
+            context=self._context(self.root_id),
+        )
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(6, len(await self._local_children(self.root_id)))
+
+    async def test_legacy_trace_depth_is_recovered_from_parent_links(self):
+        legacy_child_id = "legacy-child"
+        await self.store.create_trace(
+            Trace(
+                trace_id=legacy_child_id,
+                mode="agent",
+                task="legacy child",
+                parent_trace_id=self.root_id,
+                uid="user-1",
+                model="fake-model",
+                context={"created_by_tool": "agent"},
+            )
+        )
+        await self.store.update_goal_tree(
+            legacy_child_id,
+            GoalTree(mission="legacy child"),
+        )
+
+        result = await self._spawn_one(legacy_child_id, "grandchild")
+        self.assertEqual("completed", result["status"])
+        grandchild = await self.store.get_trace(result["sub_trace_id"])
+        self.assertEqual(2, grandchild.context["agent_depth"])
+        self.assertEqual(self.root_id, grandchild.context["root_trace_id"])
+
+    async def test_remote_agent_is_not_counted_as_local_child(self):
+        child = await self._spawn_one(self.root_id)
+        before = len(await self._local_children(child["sub_trace_id"]))
+        remote_result = {
+            "mode": "remote",
+            "agent_type": "remote_research",
+            "status": "completed",
+            "summary": "remote completed",
+            "stats": {},
+        }
+        with patch(
+            "cyber_agent.tools.builtin.subagent._run_remote_agent",
+            new=AsyncMock(return_value=remote_result),
+        ):
+            result = await agent(
+                task="remote task",
+                agent_type="remote_research",
+                context=self._context(child["sub_trace_id"]),
+            )
+        self.assertEqual("completed", result["status"])
+        self.assertEqual(before, len(await self._local_children(child["sub_trace_id"])))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_trace_id.py

@@ -0,0 +1,20 @@
+import unittest
+
+from cyber_agent.trace.trace_id import extract_mode, parse_parent_trace_id
+
+
+class NestedTraceIdTest(unittest.TestCase):
+    def test_nested_trace_returns_direct_parent_and_current_mode(self):
+        parent = "root@delegate-20260716120000-001"
+        child = f"{parent}@explore-20260716120100-001"
+
+        self.assertEqual(parent, parse_parent_trace_id(child))
+        self.assertEqual("explore", extract_mode(child))
+
+    def test_root_has_no_parent_or_mode(self):
+        self.assertIsNone(parse_parent_trace_id("root"))
+        self.assertIsNone(extract_mode("root"))
+
+
+if __name__ == "__main__":
+    unittest.main()