接入 recursive 根任务锚点与运行恢复

RunConfig 与创建接口统一改用 root_task_anchor，并在根 Trace 创建时固化锚点及哈希。

Runner 在新建、续跑、回溯和根验收阶段校验权威锚点；Legacy 保持原有接口语义，新旧 Recursive 实验 Trace 不做隐式升级。

cyber_agent/__init__.py

@@ -12,6 +12,7 @@ Reson Agent - 模块化、可扩展的 Agent 框架
 
 # 核心引擎
 from cyber_agent.core.runner import AgentRunner, CallResult, RunConfig
+from cyber_agent.core.task_protocol import RootTaskAnchor
 from cyber_agent.core.presets import AgentPreset, AGENT_PRESETS, get_preset
 
 # 执行追踪
@@ -37,6 +38,7 @@ __all__ = [
     "AgentRunner",
     "CallResult",
     "RunConfig",
+    "RootTaskAnchor",
     "AgentPreset",
     "AGENT_PRESETS",
     "get_preset",

cyber_agent/core/__init__.py

@@ -8,6 +8,7 @@ Agent Core - 核心引擎模块
 """
 
 from cyber_agent.core.runner import AgentRunner, CallResult, RunConfig
+from cyber_agent.core.task_protocol import RootTaskAnchor
 from cyber_agent.core.presets import (
     AgentPreset,
     AGENT_PRESETS,
@@ -21,6 +22,7 @@ __all__ = [
     "AgentRunner",
     "CallResult",
     "RunConfig",
+    "RootTaskAnchor",
     "AgentPreset",
     "AGENT_PRESETS",
     "get_preset",

cyber_agent/core/runner.py

@@ -53,10 +53,22 @@ from cyber_agent.core.agent_mode import (
     validate_recursive_child_execution,
 )
 from cyber_agent.core.task_protocol import (
+    RootTaskAnchor,
     ensure_task_protocol,
     protocol_error_report,
     rebuild_pending_replans,
 )
+from cyber_agent.core.context_policy import (
+    canonical_json,
+    ContextPolicyError,
+    normalize_root_task_anchor,
+    persist_root_task_anchor,
+    prune_context_access,
+    render_recursive_context,
+    require_matching_root_task_anchor,
+    require_root_task_anchor,
+    replace_context_access,
+)
 from cyber_agent.core.resource_budget import (
     RESOURCE_BUDGET_CONTEXT_KEY,
     ResourceBudget,
@@ -157,7 +169,7 @@ class RunConfig:
     parallel_tool_execution: bool = False      # 是否启用并发 Tool Call 执行（慎用，需确保无资源冲突）
     child_execution_mode: Literal["sequential", "parallel"] = "sequential"
     max_parallel_children: int = 2
-    root_completion_criteria: List[str] = field(default_factory=list)
+    root_task_anchor: Optional[RootTaskAnchor] = None
 
     # --- Trace 控制 ---
     trace_id: Optional[str] = None             # None = 新建
@@ -442,10 +454,25 @@ class AgentRunner:
                 evaluated_trace_id,
                 evaluated.head_sequence or evaluated.last_sequence,
             )
+            root_trace_id = evaluated.context.get("root_trace_id") or evaluated.trace_id
+            root = (
+                evaluated
+                if root_trace_id == evaluated.trace_id
+                else await self.trace_store.get_trace(root_trace_id)
+            )
+            if not root:
+                raise ContextPolicyError(
+                    f"Recursive root Trace not found: {root_trace_id}"
+                )
+            root_anchor = require_matching_root_task_anchor(
+                root.context,
+                evaluated.context,
+            )
             return await validator.validate(
                 evaluated_trace=evaluated,
                 trajectory=trajectory,
                 scope=scope,
+                root_task_anchor=root_anchor,
                 task_brief=task_brief,
                 task_report=task_report,
                 completion_criteria=completion_criteria,
@@ -862,7 +889,7 @@ class AgentRunner:
     ) -> Tuple[Trace, Optional[GoalTree], int]:
         """创建并持久化一个新根 Trace。
 
-        ``run`` 首次执行时调用；Recursive 会在此固化模式、根验收标准和树级预算。
+        ``run`` 首次执行时调用；Recursive 会在此固化根任务锚点和树级预算。
         """
         # 在任何标题生成/LLM 调用前完成模式校验。
         policy = policy_from_environment(recursive_revision=2)
@@ -871,19 +898,16 @@ class AgentRunner:
                 config.child_execution_mode,
                 config.max_parallel_children,
             )
-            criteria = [
-                item.strip()
-                for item in config.root_completion_criteria
-                if isinstance(item, str) and item.strip()
-            ]
-            if not criteria:
+            if config.root_task_anchor is None:
                 raise ValueError(
-                    "New Recursive root traces require root_completion_criteria"
+                    "New Recursive root traces require root_task_anchor"
                 )
-            if len(criteria) != len(config.root_completion_criteria):
-                raise ValueError(
-                    "root_completion_criteria must contain non-empty strings"
+            try:
+                root_task_anchor = normalize_root_task_anchor(
+                    config.root_task_anchor
                 )
+            except ContextPolicyError as exc:
+                raise ValueError(str(exc)) from exc
             budget = ResourceBudget.from_environment()
         trace_id = str(uuid.uuid4())
 
@@ -901,9 +925,15 @@ class AgentRunner:
         trace_context.setdefault("agent_depth", 0)
         trace_context.setdefault("root_trace_id", trace_id)
         if policy.requires_task_protocol:
-            trace_context["root_completion_criteria"] = criteria
+            persist_root_task_anchor(trace_context, root_task_anchor)
             trace_context[RESOURCE_BUDGET_CONTEXT_KEY] = budget.to_dict()
-            ensure_task_protocol(trace_context)
+            state = ensure_task_protocol(trace_context)
+            replace_context_access(
+                trace_context,
+                [],
+                root_task_anchor=root_task_anchor,
+                task_brief=state.get("task_brief"),
+            )
 
         trace_obj = Trace(
             trace_id=trace_id,
@@ -976,19 +1006,23 @@ class AgentRunner:
                 )
                 if policy.requires_task_protocol:
                     root_trace_id = trace_obj.context.get("root_trace_id")
-                    if root_trace_id == trace_obj.trace_id and not trace_obj.context.get(
-                        "root_completion_criteria"
-                    ):
-                        raise ValueError(
-                            "This experimental Recursive trace predates required root "
-                            "completion criteria; create a new trace"
-                        )
                     root = (
                         trace_obj
                         if root_trace_id == trace_obj.trace_id
                         else await self.trace_store.get_trace(root_trace_id)
                     )
-                    if not root or RESOURCE_BUDGET_CONTEXT_KEY not in root.context:
+                    if not root:
+                        raise ValueError(
+                            f"Recursive root Trace not found: {root_trace_id}"
+                        )
+                    try:
+                        require_matching_root_task_anchor(
+                            root.context,
+                            trace_obj.context,
+                        )
+                    except ContextPolicyError as exc:
+                        raise ValueError(str(exc)) from exc
+                    if RESOURCE_BUDGET_CONTEXT_KEY not in root.context:
                         raise ValueError(
                             "This experimental Recursive trace predates tree resource "
                             "budgets; create a new trace"
@@ -1093,6 +1127,48 @@ class AgentRunner:
                 if main_path:
                     head_seq = main_path[-1].sequence
 
+        current_trace = (
+            await self.trace_store.get_trace(trace_id)
+            if self.trace_store
+            else None
+        )
+        if (
+            current_trace
+            and current_trace.head_sequence == 0
+            and policy_from_context(current_trace.context).requires_task_protocol
+        ):
+            anchor = require_root_task_anchor(current_trace.context)
+            anchor_text = (
+                "# Root Task Anchor\n\n"
+                + canonical_json(anchor.model_dump(mode="json"))
+            )
+            anchored_messages = []
+            injected = False
+            for message in new_messages:
+                if not injected and message.get("role") == "user":
+                    content = message.get("content") or ""
+                    if isinstance(content, str):
+                        anchored_content = f"{anchor_text}\n\n{content}"
+                    elif isinstance(content, list):
+                        anchored_content = [
+                            {"type": "text", "text": anchor_text},
+                            *content,
+                        ]
+                    else:
+                        raise ValueError(
+                            "Recursive root anchor requires text or multimodal user content"
+                        )
+                    anchored_messages.append({
+                        **message,
+                        "content": anchored_content,
+                    })
+                    injected = True
+                else:
+                    anchored_messages.append(message)
+            if not injected:
+                anchored_messages.append({"role": "user", "content": anchor_text})
+            new_messages = anchored_messages
+
         # 2. 构建/注入 skills 到 system prompt
         has_system = any(m.get("role") == "system" for m in history)
         has_system_in_new = any(m.get("role") == "system" for m in new_messages)
@@ -2657,7 +2733,9 @@ class AgentRunner:
                 validation_run = await self.validate_recursive_trace(
                     trace_id,
                     scope="root",
-                    completion_criteria=trace.context["root_completion_criteria"],
+                    completion_criteria=require_root_task_anchor(
+                        trace.context
+                    ).completion_criteria,
                     candidate_output=response_content,
                     root_validator=True,
                 )
@@ -2917,8 +2995,21 @@ class AgentRunner:
                 action for action in state["next_actions"]
                 if action.get("created_at_sequence", 0) <= cutoff
             ]
+            while (
+                state.get("task_brief") is not None
+                and state.get("task_brief_effective_at_sequence", 0) > cutoff
+                and state.get("task_brief_history")
+            ):
+                previous = state["task_brief_history"].pop()
+                state["task_brief"] = previous["task_brief"]
+                state["task_brief_version"] = previous["version"]
+                state["task_brief_effective_at_sequence"] = previous.get(
+                    "effective_at_sequence",
+                    0,
+                )
             state["pending_replans"] = rebuild_pending_replans(state)
             state["protocol_correction_attempts"] = 0
+            prune_context_access(trace.context, cutoff)
             await self.trace_store.update_trace(trace_id, context=trace.context)
 
             projected_statuses: dict[str, str] = {}
@@ -3194,6 +3285,9 @@ class AgentRunner:
         from datetime import datetime
         parts = [f"## Current Time\n\n{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"]
 
+        if trace and policy_from_context(trace.context).requires_task_protocol:
+            parts.append(render_recursive_context(trace.context))
+
         # GoalTree
         if goal_tree and goal_tree.goals:
             parts.append(f"## Current Plan\n\n{goal_tree.to_prompt()}")
@@ -3887,7 +3981,11 @@ class AgentRunner:
             config.exclude_tools,
         )
         policy = policy_from_context(trace.context)
-        protocol_tools = {"submit_task_report", "review_task_result"}
+        protocol_tools = {
+            "submit_task_report",
+            "review_task_result",
+            "read_context_ref",
+        }
 
         if not policy.requires_task_protocol or in_side_branch:
             available = tool_names - protocol_tools
@@ -3898,9 +3996,9 @@ class AgentRunner:
         state = ensure_task_protocol(trace.context)
         tool_names.discard("evaluate")
         if state["pending_reviews"]:
-            return tool_names & {"review_task_result"}
+            return tool_names & {"review_task_result", "read_context_ref"}
         if state["next_actions"]:
-            return tool_names & {"agent"}
+            return tool_names & {"agent", "read_context_ref"}
 
         tool_names.discard("review_task_result")
         if not trace.parent_trace_id or state.get("task_report") is not None:

cyber_agent/trace/run_api.py

@@ -22,7 +22,9 @@ from datetime import datetime
 from typing import Any, Dict, List, Optional
 
 from fastapi import APIRouter, HTTPException
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, model_validator
+
+from cyber_agent.core.task_protocol import RootTaskAnchor
 
 logger = logging.getLogger(__name__)
 
@@ -68,11 +70,20 @@ class CreateRequest(BaseModel):
     name: Optional[str] = Field(None, description="任务名称（None = 自动生成）")
     uid: Optional[str] = Field(None)
     project_name: Optional[str] = Field(None, description="示例项目名称，若提供则动态加载其执行环境")
-    root_completion_criteria: List[str] = Field(
-        default_factory=list,
-        description="Recursive 根任务必须显式提供的完成标准；Legacy 忽略",
+    root_task_anchor: Optional[RootTaskAnchor] = Field(
+        None,
+        description="Recursive 根任务必须显式提供的不可变目标、完成标准和硬约束；Legacy 忽略",
     )
 
+    @model_validator(mode="before")
+    @classmethod
+    def reject_removed_root_criteria(cls, value):
+        if isinstance(value, dict) and "root_completion_criteria" in value:
+            raise ValueError(
+                "root_completion_criteria has been removed; use root_task_anchor"
+            )
+        return value
+
 
 class TraceRunRequest(BaseModel):
     """运行（统一续跑 + 回溯）"""
@@ -243,7 +254,7 @@ async def create_and_run(req: CreateRequest):
     """
     新建 Trace 并开始执行，将 HTTP 参数和 Example 默认值合并为 RunConfig。
 
-    Recursive 的 root_completion_criteria 从此传入 Runner 完成门禁；获取 trace_id 后立即返回，执行继续留在后台。
+    Recursive 的 root_task_anchor 从此传入 Runner 完成门禁；获取 trace_id 后立即返回，执行继续留在后台。
     """
     import importlib
     from cyber_agent.core.runner import RunConfig
@@ -273,7 +284,7 @@ async def create_and_run(req: CreateRequest):
                     enable_research_flow=default_config.enable_research_flow,
                     child_execution_mode=default_config.child_execution_mode,
                     max_parallel_children=default_config.max_parallel_children,
-                    root_completion_criteria=req.root_completion_criteria,
+                    root_task_anchor=req.root_task_anchor,
                     context={"project_name": req.project_name}
                 )
         except ImportError as e:
@@ -295,7 +306,7 @@ async def create_and_run(req: CreateRequest):
             tools=req.tools,
             name=req.name,
             uid=req.uid,
-            root_completion_criteria=req.root_completion_criteria,
+            root_task_anchor=req.root_task_anchor,
             context={"project_name": req.project_name} if req.project_name else {}
         )