接入 recursive 子 Agent 上下文继承与授权

创建子 Trace 时从根 Trace加载权威 RootTaskAnchor，并规范化 TaskBrief、累积硬约束和生成直接父级任务快照。

只向孩子下发显式授权的引用和冻结工具能力，depth 5 保留读取能力但继续禁止创建更深子 Agent。

cyber_agent/tools/builtin/subagent.py

@@ -30,13 +30,20 @@ from cyber_agent.core.task_protocol import (
     TaskReport,
     ensure_task_protocol,
     format_task_brief,
+    new_task_protocol,
     pending_review_entry,
     protocol_error_report,
+    replace_task_brief,
     stopped_task_report,
 )
 from cyber_agent.core.context_policy import (
+    CONTEXT_ACCESS_KEY,
     ContextPolicyError,
+    build_child_context_access,
+    context_ref_descriptors,
     normalize_task_brief,
+    persist_root_task_anchor,
+    require_matching_root_task_anchor,
     task_briefs_match,
 )
 from cyber_agent.core.resource_budget import (
@@ -489,7 +496,11 @@ def _get_allowed_tools(
 
         blocked_tools = {"evaluate", "bash_command"}
         if not policy.requires_task_protocol:
-            blocked_tools.update({"submit_task_report", "review_task_result"})
+            blocked_tools.update({
+                "submit_task_report",
+                "review_task_result",
+                "read_context_ref",
+            })
         if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
             blocked_tools.add("agent")
         return sorted(allowed_tools - blocked_tools)
@@ -526,6 +537,16 @@ async def _resolve_trace_lineage(store, trace: Trace) -> tuple[int, str]:
     return depth, root_trace_id
 
 
+async def _load_root_task_anchor(store, trace: Trace):
+    """从权威根 Trace读取 Anchor，并校验当前节点持有相同副本。"""
+    root_trace_id = trace.context.get("root_trace_id") or trace.trace_id
+    root = trace if root_trace_id == trace.trace_id else await store.get_trace(root_trace_id)
+    if not root:
+        raise ContextPolicyError(f"Recursive root Trace not found: {root_trace_id}")
+    anchor = require_matching_root_task_anchor(root.context, trace.context)
+    return root, anchor
+
+
 def _format_single_result(result: Dict[str, Any], sub_trace_id: str, continued: bool) -> Dict[str, Any]:
     """格式化单任务（delegate）结果"""
     lines = [DELEGATE_RESULT_HEADER]
@@ -759,6 +780,15 @@ async def _run_agents(
         policy = policy_from_context(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
+    root_task_anchor = None
+    if policy.requires_task_protocol:
+        try:
+            _, root_task_anchor = await _load_root_task_anchor(
+                store,
+                parent_trace,
+            )
+        except ContextPolicyError as exc:
+            return {"status": "failed", "error": str(exc)}
 
     if (
         policy.mode is AgentMode.RECURSIVE
@@ -823,6 +853,7 @@ async def _run_agents(
                         task_briefs[0],
                         expected_brief,
                         parent_task_brief=parent_task_brief,
+                        root_task_anchor=root_task_anchor,
                     )
                 except ContextPolicyError as exc:
                     return {
@@ -845,6 +876,7 @@ async def _run_agents(
                         task_briefs[0],
                         expected_brief,
                         parent_task_brief=parent_task_brief,
+                        root_task_anchor=root_task_anchor,
                     )
                 except ContextPolicyError as exc:
                     return {
@@ -948,20 +980,40 @@ async def _run_agents(
                 ),
             }
         all_sub_trace_ids = [{"trace_id": sub_trace_id, "mission": mission}]
+        child_context = apply_policy_to_context({
+            **existing.context,
+            "agent_depth": child_depth,
+            "root_trace_id": root_trace_id,
+        }, policy)
         child_records = [{
             "trace_id": sub_trace_id,
             "depth": child_depth,
-            "context": apply_policy_to_context({
-                **existing.context,
-                "agent_depth": child_depth,
-                "root_trace_id": root_trace_id,
-            }, policy),
+            "context": child_context,
             "is_new": False,
         }]
         if task_briefs:
-            child_state = ensure_task_protocol(child_records[0]["context"])
-            child_state["task_brief"] = task_briefs[0].model_dump()
-            existing.context = child_records[0]["context"]
+            assert root_task_anchor is not None
+            child_state = ensure_task_protocol(child_context)
+            normalized_dump = task_briefs[0].model_dump(mode="json")
+            new_context_access = build_child_context_access(
+                parent_context=parent_trace.context,
+                parent_trace_id=parent_trace.trace_id,
+                root_trace_id=root_trace_id,
+                uid=parent_trace.uid,
+                parent_task_state=protocol_state,
+                child_task_brief=task_briefs[0],
+                root_task_anchor=root_task_anchor,
+                granted_at_sequence=(existing.last_sequence or 0) + 1,
+            )
+            if child_state.get("task_brief") != normalized_dump:
+                replace_task_brief(
+                    child_state,
+                    task_briefs[0],
+                    effective_at_sequence=(existing.last_sequence or 0) + 1,
+                )
+            child_context[CONTEXT_ACCESS_KEY] = new_context_access
+            persist_root_task_anchor(child_context, root_task_anchor)
+            existing.context = child_context
             await store.update_trace(existing.trace_id, context=existing.context)
     else:
         parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
@@ -975,6 +1027,28 @@ async def _run_agents(
                 ),
             }
 
+        prepared_context_accesses: list[dict[str, Any]] = []
+        if policy.requires_task_protocol:
+            assert root_task_anchor is not None and task_briefs is not None
+            try:
+                prepared_context_accesses = [
+                    build_child_context_access(
+                        parent_context=parent_trace.context,
+                        parent_trace_id=parent_trace.trace_id,
+                        root_trace_id=root_trace_id,
+                        uid=parent_trace.uid,
+                        parent_task_state=protocol_state,
+                        child_task_brief=brief,
+                        root_task_anchor=root_task_anchor,
+                    )
+                    for brief in task_briefs
+                ]
+            except ContextPolicyError as exc:
+                return {
+                    "status": "failed",
+                    "error": f"Invalid TaskBrief context references: {exc}",
+                }
+
         async def create_child_traces() -> None:
             child_depth = parent_depth + 1
             for i, task_item in enumerate(tasks):
@@ -992,11 +1066,12 @@ async def _run_agents(
                     "root_trace_id": root_trace_id,
                 }, policy)
                 if policy.requires_task_protocol:
-                    child_context["task_protocol"] = ensure_task_protocol({
-                        "task_protocol": {
-                            "task_brief": task_briefs[i].model_dump(),
-                        }
-                    })
+                    assert root_task_anchor is not None
+                    child_context["task_protocol"] = new_task_protocol(
+                        task_briefs[i]
+                    )
+                    child_context[CONTEXT_ACCESS_KEY] = prepared_context_accesses[i]
+                    persist_root_task_anchor(child_context, root_task_anchor)
                 sub_trace = Trace(
                     trace_id=stid,
                     mode="agent",
@@ -1134,6 +1209,14 @@ async def _run_agents(
         )
 
         # 构建消息
+        if policy.requires_task_protocol:
+            assert task_briefs is not None and root_task_anchor is not None
+            task_item = format_task_brief(
+                task_briefs[i],
+                available_context_refs=context_ref_descriptors(
+                    child_record["context"]
+                ),
+            )
         agent_msgs = list(msgs) + [{"role": "user", "content": task_item}]
         allowed_tools = _get_allowed_tools(context, child_depth, policy)
 
@@ -1540,18 +1623,23 @@ async def agent(
                 return {"status": "failed", "error": f"Invalid TaskBrief JSON: {exc}"}
         raw_briefs = task_brief if isinstance(task_brief, list) else [task_brief]
         try:
+            _, root_task_anchor = await _load_root_task_anchor(
+                store,
+                parent_trace,
+            )
             parent_state = ensure_task_protocol(parent_trace.context)
             parent_task_brief = parent_state.get("task_brief")
             parsed_task_briefs = [
                 normalize_task_brief(
                     item,
                     parent_task_brief=parent_task_brief,
+                    root_task_anchor=root_task_anchor,
                 )
                 for item in raw_briefs
             ]
         except (ContextPolicyError, ValidationError) as exc:
             return {"status": "failed", "error": f"Invalid TaskBrief: {exc}"}
-        tasks = [format_task_brief(brief) for brief in parsed_task_briefs]
+        tasks = [brief.objective for brief in parsed_task_briefs]
         single = len(tasks) == 1
     else:
         if task_brief is not None: