接入 recursive 审核结果引用与回溯清理

子报告审核完成后生成不可变 reviewed_task_result 快照，并把可转授引用返回给直接父级。

REVISE_CHILD 时替换孩子授权，审核和回溯按序列清理失效快照；Validator 输入同时固定包含根任务锚点。

cyber_agent/core/validation.py

@@ -262,6 +262,7 @@ def build_validation_packet(
     *,
     validation_scope: ValidationScope,
     trajectory: Sequence[Message | Mapping[str, Any]],
+    root_task_anchor: Mapping[str, Any] | BaseModel | None = None,
     task_brief: Mapping[str, Any] | BaseModel | None = None,
     task_report: Mapping[str, Any] | BaseModel | None = None,
     completion_criteria: Sequence[str] | None = None,
@@ -274,6 +275,7 @@ def build_validation_packet(
     `LLMValidator.validate` 在发起单次验收前调用，包含 TaskBrief、TaskReport、标准和真实主路径消息。
     """
 
+    anchor = _jsonable(root_task_anchor)
     brief = _jsonable(task_brief)
     report = _jsonable(task_report)
     if completion_criteria is None and isinstance(brief, dict):
@@ -283,6 +285,7 @@ def build_validation_packet(
 
     packet: dict[str, Any] = {
         "validation_scope": validation_scope,
+        "root_task_anchor": anchor,
         "task_brief": brief,
         "completion_criteria": _jsonable(completion_criteria or []),
         "expected_outputs": _jsonable(expected_outputs or []),
@@ -337,6 +340,7 @@ class LLMValidator:
         evaluated_trace: Trace,
         trajectory: Sequence[Message | Mapping[str, Any]],
         scope: ValidationScope,
+        root_task_anchor: Mapping[str, Any] | BaseModel | None = None,
         task_brief: Mapping[str, Any] | BaseModel | None = None,
         task_report: Mapping[str, Any] | BaseModel | None = None,
         completion_criteria: Sequence[str] | None = None,
@@ -375,6 +379,7 @@ class LLMValidator:
             packet = build_validation_packet(
                 validation_scope=scope,
                 trajectory=trajectory,
+                root_task_anchor=root_task_anchor,
                 task_brief=task_brief,
                 task_report=task_report,
                 completion_criteria=completion_criteria,

cyber_agent/tools/builtin/task_protocol.py

@@ -13,7 +13,16 @@ from typing import Any
 from pydantic import ValidationError
 
 from cyber_agent.core.agent_mode import policy_from_context
-from cyber_agent.core.context_policy import ContextPolicyError, normalize_task_brief
+from cyber_agent.core.context_policy import (
+    ContextPolicyError,
+    add_context_snapshot,
+    build_child_context_access,
+    create_context_snapshot,
+    normalize_task_brief,
+    replace_context_access,
+    require_matching_root_task_anchor,
+    require_root_task_anchor,
+)
 from cyber_agent.core.task_protocol import (
     TaskReport,
     TaskReportSubmission,
@@ -73,6 +82,17 @@ async def submit_task_report(
     if not trace.parent_trace_id:
         return _error("Root Agent does not submit a TaskReport")
 
+    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:
+        return _error(f"Recursive root Trace not found: {root_trace_id}")
+    try:
+        require_matching_root_task_anchor(root.context, trace.context)
+    except ContextPolicyError as exc:
+        return _error(str(exc))
+
     state = ensure_task_protocol(trace.context)
     if state["pending_reviews"]:
         return _error("Review all child TaskReports before submitting this TaskReport")
@@ -168,6 +188,17 @@ async def review_task_result(
     revision_task_brief = None
     child_state = None
     try:
+        root_trace_id = parent.context.get("root_trace_id") or parent.trace_id
+        root = parent if root_trace_id == parent.trace_id else await store.get_trace(
+            root_trace_id
+        )
+        if not root:
+            return _error(f"Recursive root Trace not found: {root_trace_id}")
+        root_task_anchor = require_matching_root_task_anchor(
+            root.context,
+            parent.context,
+        )
+        require_matching_root_task_anchor(root.context, child.context)
         parent_task_brief = state.get("task_brief")
         if isinstance(approved_next_task, str):
             approved_next_task = TaskBrief.model_validate_json(approved_next_task)
@@ -175,6 +206,7 @@ async def review_task_result(
             approved_next_task = normalize_task_brief(
                 approved_next_task,
                 parent_task_brief=parent_task_brief,
+                root_task_anchor=root_task_anchor,
             )
         validation_result = ValidationResult.model_validate(
             entry.get("validation_result")
@@ -213,6 +245,7 @@ async def review_task_result(
             revision_task_brief = normalize_task_brief(
                 review.approved_next_task or child_state.get("task_brief"),
                 parent_task_brief=parent_task_brief,
+                root_task_anchor=root_task_anchor,
             )
     except (ContextPolicyError, ValidationError) as exc:
         return _error(f"Invalid TaskReview: {exc}")
@@ -252,12 +285,84 @@ async def review_task_result(
                     f"reserved={reserved_children}, max={child_limit})"
                 )
 
+    action_task_brief = revision_task_brief or review.approved_next_task
+    if action_task_brief is not None:
+        try:
+            build_child_context_access(
+                parent_context=parent.context,
+                parent_trace_id=parent_trace_id,
+                root_trace_id=root_trace_id,
+                uid=parent.uid,
+                parent_task_state=state,
+                child_task_brief=action_task_brief,
+                root_task_anchor=root_task_anchor,
+            )
+        except ContextPolicyError as exc:
+            return _error(f"Approved TaskBrief context is invalid: {exc}")
+
+    reviewed_at_sequence = int(context.get("sequence", 0) or 0)
+    context_ref = None
+    context_ref_warning = None
+    try:
+        snapshot = create_context_snapshot(
+            kind="reviewed_task_result",
+            summary=review.reason,
+            source_trace_id=child_trace_id,
+            root_trace_id=root_trace_id,
+            uid=parent.uid,
+            content={
+                "task_report": deepcopy(entry["task_report"]),
+                "validation_result": deepcopy(entry["validation_result"]),
+                "task_review": review.model_dump(mode="json"),
+            },
+            granted_at_sequence=reviewed_at_sequence,
+        )
+        context_ref = add_context_snapshot(
+            parent.context,
+            snapshot,
+            root_task_anchor=root_task_anchor,
+            task_brief=state.get("task_brief"),
+        )
+    except ContextPolicyError as exc:
+        context_ref_warning = str(exc)
+
+    if context_ref is not None and action_task_brief is not None:
+        try:
+            action_task_brief = normalize_task_brief(
+                action_task_brief.model_copy(update={
+                    "context_refs": [
+                        *action_task_brief.context_refs,
+                        context_ref,
+                    ],
+                }),
+                parent_task_brief=parent_task_brief,
+                root_task_anchor=root_task_anchor,
+            )
+            build_child_context_access(
+                parent_context=parent.context,
+                parent_trace_id=parent_trace_id,
+                root_trace_id=root_trace_id,
+                uid=parent.uid,
+                parent_task_state=state,
+                child_task_brief=action_task_brief,
+                root_task_anchor=root_task_anchor,
+            )
+        except ContextPolicyError as exc:
+            action_task_brief = revision_task_brief or review.approved_next_task
+            context_ref_warning = (
+                f"{context_ref_warning}; {exc}"
+                if context_ref_warning
+                else str(exc)
+            )
+
     pending.pop(child_trace_id)
     review_record = {
         **review.model_dump(),
         "pending_review": deepcopy(entry),
         "reviewed_at": datetime.now().isoformat(),
-        "reviewed_at_sequence": context.get("sequence", 0),
+        "reviewed_at_sequence": reviewed_at_sequence,
+        "context_ref": context_ref.model_dump() if context_ref else None,
+        "context_ref_warning": context_ref_warning,
     }
     state["reviews"].append(review_record)
     state["protocol_correction_attempts"] = 0
@@ -271,17 +376,21 @@ async def review_task_result(
         child_state["task_report_submitted_at_sequence"] = None
         child_state["task_report_validation"] = None
         child_state["protocol_correction_attempts"] = 0
-        child_state["task_brief"] = revision_task_brief.model_dump()
+        replace_context_access(
+            child.context,
+            [],
+            root_task_anchor=require_root_task_anchor(child.context),
+            task_brief=child_state.get("task_brief"),
+        )
         await store.update_trace(child_trace_id, context=child.context)
 
     if review.decision in {"ACCEPT_REFINE", "DESCEND_AGAIN", "REVISE_CHILD"}:
-        action_task_brief = revision_task_brief or review.approved_next_task
         state["next_actions"].append({
             "decision": review.decision,
             "child_trace_id": child_trace_id,
             "goal_id": origin_goal_id,
             "task_brief": action_task_brief.model_dump() if action_task_brief else None,
-            "created_at_sequence": context.get("sequence", 0),
+            "created_at_sequence": reviewed_at_sequence,
         })
 
     resolved_replan = None
@@ -343,6 +452,13 @@ async def review_task_result(
     return {
         "status": "completed",
         "task_review": review.model_dump(),
+        "context_ref": context_ref.model_dump() if context_ref else None,
+        "context_ref_warning": context_ref_warning,
+        "next_action_task_brief": (
+            action_task_brief.model_dump(mode="json")
+            if action_task_brief is not None
+            else None
+        ),
         "pending_review_count": len(pending),
         "next_action_count": len(state["next_actions"]),
     }