补齐 recursive 上下文生命周期与模式隔离回归

验证旧 Recursive Trace拒绝隐式迁移、REVISE_CHILD和回溯正确清理授权、取消路径保持一致。

同时冻结 Legacy 行为及 read_context_ref 的深度、协议状态和父级工具权限边界，避免上下文能力泄漏到其他模式。

tests/test_agent_mode.py

@@ -44,7 +44,10 @@ class AgentModePolicyTest(unittest.IsolatedAsyncioTestCase):
             [{"role": "user", "content": "test"}],
             RunConfig(
                 name="mode test",
-                root_completion_criteria=["return a checked answer"],
+                root_task_anchor={
+                    "objective": "return a checked answer",
+                    "completion_criteria": ["return a checked answer"],
+                },
             ),
         )
 

tests/test_recursive_replan_context.py

@@ -6,6 +6,7 @@ from pydantic import ValidationError
 from cyber_agent.core.context_policy import (
     ContextPolicyError,
     normalize_task_brief,
+    persist_root_task_anchor,
     task_briefs_match,
 )
 from cyber_agent.core.task_protocol import (
@@ -35,6 +36,16 @@ BRIEF = {
     "context": {"claim": "example"},
     "constraints": ["Do not guess"],
 }
+ROOT_ANCHOR = {
+    "objective": "Check the parent claim",
+    "completion_criteria": ["Reach one checked conclusion"],
+    "constraints": ["Do not guess"],
+}
+
+
+def with_root_anchor(context):
+    persist_root_task_anchor(context, ROOT_ANCHOR)
+    return context
 
 
 def task_report(child_trace_id: str) -> TaskReport:
@@ -128,13 +139,13 @@ class ReplanCurrentTest(unittest.IsolatedAsyncioTestCase):
             task="parent",
             uid="user-1",
             model="fake",
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 1,
-                "root_trace_id": "root",
+                "root_trace_id": self.parent_id,
                 "task_protocol": new_task_protocol(BRIEF),
-            },
+            }),
         ))
         self.tree = GoalTree(
             mission="parent",
@@ -161,13 +172,13 @@ class ReplanCurrentTest(unittest.IsolatedAsyncioTestCase):
             parent_goal_id="1",
             uid="user-1",
             model="fake",
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 2,
-                "root_trace_id": "root",
+                "root_trace_id": self.parent_id,
                 "task_protocol": new_task_protocol(BRIEF),
-            },
+            }),
         ))
         parent = await self.store.get_trace(self.parent_id)
         state = ensure_task_protocol(parent.context)

tests/test_recursive_runtime_extensions.py

@@ -13,6 +13,7 @@ from cyber_agent.core.resource_budget import (
 )
 from cyber_agent.core.runner import AgentRunner, RunConfig
 from cyber_agent.core.task_protocol import ensure_task_protocol, new_task_protocol
+from cyber_agent.core.context_policy import persist_root_task_anchor
 from cyber_agent.tools import get_tool_registry
 from cyber_agent.tools.builtin.knowledge import KnowledgeConfig
 from cyber_agent.tools.builtin.subagent import agent
@@ -22,6 +23,11 @@ from cyber_agent.trace.store import FileSystemTraceStore
 
 
 ROOT_CRITERIA = ["return a checked final answer"]
+ROOT_ANCHOR = {
+    "objective": "return a checked final answer",
+    "completion_criteria": ROOT_CRITERIA,
+    "constraints": ["do not invent evidence"],
+}
 CHILD_BRIEF = {
     "objective": "check one concrete fact",
     "reason": "the root needs independently checked evidence",
@@ -39,6 +45,11 @@ def knowledge_disabled():
     )
 
 
+def with_root_anchor(context):
+    persist_root_task_anchor(context, ROOT_ANCHOR)
+    return context
+
+
 def recursive_env(**overrides):
     values = {
         "AGENT_MODE": "recursive",
@@ -102,7 +113,7 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
             "tools": [],
             "tool_groups": [],
             "enable_research_flow": False,
-            "root_completion_criteria": ROOT_CRITERIA,
+            "root_task_anchor": ROOT_ANCHOR,
             "knowledge": knowledge_disabled(),
         }
         values.update(overrides)
@@ -308,16 +319,16 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
             for item in trajectory
         ))
 
-    async def test_new_recursive_root_requires_explicit_completion_criteria(self):
+    async def test_new_recursive_root_requires_explicit_task_anchor(self):
         async def llm_call(**_kwargs):
             raise AssertionError("LLM must not be called")
 
         runner = AgentRunner(trace_store=self.store, llm_call=llm_call)
         with recursive_env():
-            with self.assertRaisesRegex(ValueError, "root_completion_criteria"):
+            with self.assertRaisesRegex(ValueError, "root_task_anchor"):
                 await runner.run_result(
                     messages=[{"role": "user", "content": "missing criteria"}],
-                    config=self.config(root_completion_criteria=[]),
+                    config=self.config(root_task_anchor=None),
                 )
         self.assertEqual([], await self.store.list_traces(limit=10))
 
@@ -346,12 +357,12 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
             agent_type="validator",
             tools=[],
             status="completed",
-            context={
+            context=with_root_anchor({
                 "created_by_tool": "validator",
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "root_trace_id": "root",
-            },
+            }),
         ))
         calls = 0
 
@@ -413,7 +424,7 @@ class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
             tools=["dangerous"],
             tool_groups=[],
             enable_research_flow=False,
-            root_completion_criteria=ROOT_CRITERIA,
+            root_task_anchor=ROOT_ANCHOR,
             knowledge=knowledge_disabled(),
         )
         with recursive_env(AGENT_MAX_TOTAL_TOKENS="10"):
@@ -472,7 +483,7 @@ class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
             tools=["agent"],
             tool_groups=[],
             enable_research_flow=False,
-            root_completion_criteria=ROOT_CRITERIA,
+            root_task_anchor=ROOT_ANCHOR,
             knowledge=knowledge_disabled(),
         )
         with recursive_env(AGENT_MAX_TOTAL_AGENTS="1"):
@@ -502,15 +513,14 @@ class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
             mode="agent",
             uid="user-1",
             model="fake",
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 0,
                 "root_trace_id": root_id,
-                "root_completion_criteria": ROOT_CRITERIA,
                 RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
                 "task_protocol": new_task_protocol(),
-            },
+            }),
         ))
         await ResourceBudgetController(self.store).initialize(root_id, budget)
         original_update_goal_tree = self.store.update_goal_tree
@@ -557,15 +567,14 @@ class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
             mode="agent",
             uid="user-1",
             model="fake",
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 0,
                 "root_trace_id": root_id,
-                "root_completion_criteria": ROOT_CRITERIA,
                 RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
                 "task_protocol": new_task_protocol(),
-            },
+            }),
         ))
         await ResourceBudgetController(self.store).initialize(root_id, budget)
         runner = AgentRunner(
@@ -605,15 +614,14 @@ class RecursiveBudgetIntegrationTest(unittest.IsolatedAsyncioTestCase):
     async def test_approved_action_write_failure_closes_precreated_child(self):
         root_id = "approved-action-write-root"
         budget = ResourceBudget()
-        context = {
+        context = with_root_anchor({
             "agent_mode": "recursive",
             "agent_mode_revision": 2,
             "agent_depth": 0,
             "root_trace_id": root_id,
-            "root_completion_criteria": ROOT_CRITERIA,
             RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
             "task_protocol": new_task_protocol(),
-        }
+        })
         state = ensure_task_protocol(context)
         state["next_actions"] = [{
             "decision": "ACCEPT_REFINE",
@@ -783,7 +791,7 @@ class RecursiveLifecycleIntegrationTest(unittest.IsolatedAsyncioTestCase):
             tools=["agent", "submit_task_report", "review_task_result"],
             tool_groups=[],
             enable_research_flow=False,
-            root_completion_criteria=ROOT_CRITERIA,
+            root_task_anchor=ROOT_ANCHOR,
             knowledge=knowledge_disabled(),
         )
         with recursive_env():
@@ -840,29 +848,28 @@ class RecursiveLifecycleIntegrationTest(unittest.IsolatedAsyncioTestCase):
             trace_id=root_id,
             mode="agent",
             model="fake-model",
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 0,
                 "root_trace_id": root_id,
-                "root_completion_criteria": ROOT_CRITERIA,
                 RESOURCE_BUDGET_CONTEXT_KEY: budget.to_dict(),
                 "task_protocol": new_task_protocol(),
-            },
+            }),
         ))
         await self.store.create_trace(Trace(
             trace_id=child_id,
             mode="agent",
             model="fake-model",
             parent_trace_id=root_id,
-            context={
+            context=with_root_anchor({
                 "agent_mode": "recursive",
                 "agent_mode_revision": 2,
                 "agent_depth": 1,
                 "root_trace_id": root_id,
                 "created_by_tool": "agent",
                 "task_protocol": new_task_protocol(CHILD_BRIEF),
-            },
+            }),
         ))
         await ResourceBudgetController(self.store).initialize(
             root_id,

tests/test_recursive_subtree_cancellation.py

@@ -14,6 +14,12 @@ from cyber_agent.trace.models import Trace
 from cyber_agent.trace.store import FileSystemTraceStore
 
 
+ROOT_ANCHOR = {
+    "objective": "exercise recursive subtree cancellation",
+    "completion_criteria": ["all stopped descendants are accounted for"],
+}
+
+
 def brief(index):
     return {
         "objective": f"child-{index}",
@@ -142,7 +148,7 @@ class RecursiveSubtreeCancellationTest(unittest.IsolatedAsyncioTestCase):
                     tools=["dangerous_cancel_test"],
                     tool_groups=[],
                     enable_research_flow=False,
-                    root_completion_criteria=["stopping prevents late tool execution"],
+                    root_task_anchor=ROOT_ANCHOR,
                 ),
             ))
             await asyncio.wait_for(llm_started.wait(), timeout=1)
@@ -206,7 +212,7 @@ class RecursiveSubtreeCancellationTest(unittest.IsolatedAsyncioTestCase):
                     child_execution_mode="parallel",
                     max_parallel_children=2,
                     enable_research_flow=False,
-                    root_completion_criteria=["stopping propagates through the child tree"],
+                    root_task_anchor=ROOT_ANCHOR,
                 ),
             ))
             await asyncio.wait_for(two_children_started.wait(), timeout=2)

tests/test_recursive_tool_capabilities.py

@@ -276,7 +276,12 @@ class RecursiveRuntimeDispatchTest(unittest.IsolatedAsyncioTestCase):
                             tool_groups=[],
                             parallel_tool_execution=parallel,
                             enable_research_flow=False,
-                            root_completion_criteria=["forged tools are never executed"],
+                            root_task_anchor={
+                                "objective": "reject forged tool calls",
+                                "completion_criteria": [
+                                    "forged tools are never executed"
+                                ],
+                            },
                             knowledge=KnowledgeConfig(
                                 enable_extraction=False,
                                 enable_completion_extraction=False,