feat(protocol): 启用 Recursive revision 3 并将 revision 2 限定为只读

新 Recursive 根 Trace 固化 revision 3，恢复与 Runner 写入前统一执行协议可变性门禁。

历史 revision 2 仍可读取，但续跑、停止、审批、回溯和启动对账均不再改写其状态。

cyber_agent/core/agent_mode.py

@@ -23,6 +23,8 @@ RECURSIVE_CAPABILITY_TOOLS_CONTEXT_KEY = "_recursive_capability_tool_names"
 RECURSIVE_CHILD_EXECUTION_MODE_CONTEXT_KEY = "_recursive_child_execution_mode"
 RECURSIVE_MAX_PARALLEL_CHILDREN_CONTEXT_KEY = "_recursive_max_parallel_children"
 MAX_PARALLEL_RECURSIVE_CHILDREN = 2
+RECURSIVE_REVISION_READ_ONLY_ERROR = "recursive_revision_2_read_only"
+CURRENT_RECURSIVE_REVISION = 3
 
 
 class AgentMode(str, Enum):
@@ -51,9 +53,19 @@ class AgentPolicy:
 
     @property
     def requires_task_protocol(self) -> bool:
-        """Recursive revision 2 是否必须走结构化任务报告和审核流程。"""
+        """Recursive revision 2+ 是否使用结构化任务协议。"""
         return self.mode is AgentMode.RECURSIVE and self.revision >= 2
 
+    @property
+    def requires_task_progress(self) -> bool:
+        """Recursive revision 3 是否强制使用 TaskProgress。"""
+        return self.mode is AgentMode.RECURSIVE and self.revision >= 3
+
+    @property
+    def is_read_only(self) -> bool:
+        """历史 Recursive revision 2 只保留读取和导出。"""
+        return self.mode is AgentMode.RECURSIVE and self.revision == 2
+
     def to_context(self) -> dict[str, Any]:
         return {
             AGENT_MODE_CONTEXT_KEY: self.mode.value,
@@ -142,6 +154,14 @@ def apply_policy_to_context(
     return merged
 
 
+def require_mutable_trace_policy(context: Mapping[str, Any] | None) -> AgentPolicy:
+    """返回可写 Trace 的持久化策略，并统一拒绝已退役的 revision 2。"""
+    policy = policy_from_context(context)
+    if policy.is_read_only:
+        raise ValueError(RECURSIVE_REVISION_READ_ONLY_ERROR)
+    return policy
+
+
 def validate_recursive_child_execution(
     mode: Any,
     max_parallel_children: Any,

cyber_agent/core/runner.py

@@ -68,7 +68,9 @@ from cyber_agent.core.agent_mode import (
     apply_policy_to_context,
     assert_removed_config_absent,
     policy_from_context,
+    CURRENT_RECURSIVE_REVISION,
     policy_from_environment,
+    require_mutable_trace_policy,
     validate_recursive_child_execution,
 )
 from cyber_agent.core.task_protocol import (
@@ -1215,7 +1217,9 @@ class AgentRunner:
         ``run`` 首次执行时调用；Recursive 会在此固化根任务锚点和树级预算。
         """
         # 在任何标题生成/LLM 调用前完成模式校验。
-        policy = policy_from_environment(recursive_revision=2)
+        policy = policy_from_environment(
+            recursive_revision=CURRENT_RECURSIVE_REVISION,
+        )
         if policy.mode is AgentMode.RECURSIVE:
             validate_recursive_child_execution(
                 config.child_execution_mode,
@@ -1325,6 +1329,7 @@ class AgentRunner:
             raise ValueError(
                 "Validator traces cannot be continued or rewound"
             )
+        require_mutable_trace_policy(trace_obj.context)
 
         # Resume/rewind behavior is always recovered from the immutable snapshot.
         # Old Recursive traces cannot be reconstructed safely; Legacy traces get

cyber_agent/trace/api.py

@@ -12,6 +12,10 @@ from typing import List, Optional, Dict, Any
 from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel
 from cyber_agent.llm.openrouter import openrouter_llm_call
+from cyber_agent.core.agent_mode import (
+    RECURSIVE_REVISION_READ_ONLY_ERROR,
+    require_mutable_trace_policy,
+)
 
 from .protocols import TraceStore
 
@@ -58,6 +62,15 @@ def get_trace_store() -> TraceStore:
     return _trace_store
 
 
+def _require_mutable_trace(trace) -> None:
+    try:
+        require_mutable_trace_policy(trace.context)
+    except ValueError as exc:
+        if str(exc) == RECURSIVE_REVISION_READ_ONLY_ERROR:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        raise
+
+
 # ===== 路由 =====
 
 
@@ -210,6 +223,7 @@ async def submit_knowledge_feedback(trace_id: str, req: KnowledgeFeedbackRequest
     trace = await store.get_trace(trace_id)
     if not trace:
         raise HTTPException(status_code=404, detail="Trace not found")
+    _require_mutable_trace(trace)
 
     knowhub_url = os.getenv("KNOWHUB_API") or os.getenv("KNOWHUB_URL", "http://localhost:9999")
     updated_count = 0

cyber_agent/trace/run_api.py

@@ -191,6 +191,21 @@ _running_runners: Dict[str, Any] = {}
 _approval_locks: Dict[str, asyncio.Lock] = {}
 
 
+def _require_mutable_trace(trace) -> None:
+    """将核心协议的只读门禁统一映射为 HTTP 409。"""
+    from cyber_agent.core.agent_mode import (
+        RECURSIVE_REVISION_READ_ONLY_ERROR,
+        require_mutable_trace_policy,
+    )
+
+    try:
+        require_mutable_trace_policy(trace.context)
+    except ValueError as exc:
+        if str(exc) == RECURSIVE_REVISION_READ_ONLY_ERROR:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        raise
+
+
 async def _cancel_and_wait_for_running_task(trace_id: str) -> None:
     """等待旧后台运行完成 finally 清理，再允许同一 Trace 续跑。"""
     old_task = _running_tasks.get(trace_id)
@@ -215,7 +230,12 @@ async def _restore_runner_and_config(
 ):
     """Restore the persisted run contract and its trusted project environment."""
     import importlib
-    from cyber_agent.core.agent_mode import AgentMode, policy_from_context
+    from cyber_agent.core.agent_mode import (
+        AgentMode,
+        RECURSIVE_REVISION_READ_ONLY_ERROR,
+        policy_from_context,
+        require_mutable_trace_policy,
+    )
     from cyber_agent.core.run_snapshot import (
         RUN_CONFIG_SNAPSHOT_CONTEXT_KEY,
         RunConfigSnapshotV1,
@@ -225,6 +245,13 @@ async def _restore_runner_and_config(
     )
     from cyber_agent.core.runner import RunConfig
 
+    try:
+        require_mutable_trace_policy(trace.context)
+    except ValueError as exc:
+        if str(exc) == RECURSIVE_REVISION_READ_ONLY_ERROR:
+            raise HTTPException(status_code=409, detail=str(exc)) from exc
+        raise
+
     config = RunConfig(
         trace_id=trace.trace_id,
         after_sequence=after_sequence,
@@ -795,6 +822,8 @@ async def stop_trace(trace_id: str):
 
     if runner.trace_store:
         trace = await runner.trace_store.get_trace(trace_id)
+        if trace:
+            _require_mutable_trace(trace)
         if trace and trace.status == "waiting_confirmation":
             batch = (
                 await runner.trace_store.get_tool_approval_batch(trace_id)
@@ -912,6 +941,10 @@ async def review_extraction(trace_id: str, extraction_id: str, req: ReviewReques
     runner = _get_runner()
     if not runner.trace_store:
         raise HTTPException(status_code=503, detail="TraceStore not configured")
+    trace = await runner.trace_store.get_trace(trace_id)
+    if not trace:
+        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
+    _require_mutable_trace(trace)
 
     if req.decision not in ("approve", "edit", "discard"):
         raise HTTPException(
@@ -942,6 +975,10 @@ async def commit_extractions(trace_id: str):
     runner = _get_runner()
     if not runner.trace_store:
         raise HTTPException(status_code=503, detail="TraceStore not configured")
+    trace = await runner.trace_store.get_trace(trace_id)
+    if not trace:
+        raise HTTPException(status_code=404, detail=f"Trace not found: {trace_id}")
+    _require_mutable_trace(trace)
 
     from cyber_agent.trace.extraction_review import commit_approved
     report = await commit_approved(runner.trace_store, trace_id)
@@ -1052,6 +1089,11 @@ async def reconcile_traces():
             tid = trace.trace_id
             if tid in _running_tasks and not _running_tasks[tid].done():
                 continue
+            # Recursive revision 2 是历史只读记录，启动对账不得修改
+            # 其审批文件或 Trace.status。
+            from cyber_agent.core.agent_mode import policy_from_context
+            if policy_from_context(trace.context).is_read_only:
+                continue
             batch = (
                 await runner.trace_store.get_tool_approval_batch(tid)
                 if hasattr(runner.trace_store, "get_tool_approval_batch")

tests/test_agent_mode.py

@@ -123,7 +123,8 @@ class AgentModePolicyTest(unittest.IsolatedAsyncioTestCase):
 
     async def test_existing_trace_keeps_persisted_mode_after_env_switch(self):
         trace, _, _ = await self._prepare_new("recursive")
-        self.assertEqual(2, trace.context["agent_mode_revision"])
+        self.assertEqual(3, trace.context["agent_mode_revision"])
+        self.assertTrue(policy_from_context(trace.context).requires_task_progress)
         self.assertIn("task_protocol", trace.context)
         os.environ["AGENT_MODE"] = "legacy"
 
@@ -135,6 +136,22 @@ class AgentModePolicyTest(unittest.IsolatedAsyncioTestCase):
         persisted = await self.store.get_trace(trace.trace_id)
         self.assertEqual("recursive", persisted.context["agent_mode"])
 
+    async def test_recursive_revision_two_is_read_only_but_still_parseable(self):
+        trace = Trace(
+            trace_id="recursive-v2-read-only",
+            mode="agent",
+            task="historical",
+            context={"agent_mode": "recursive", "agent_mode_revision": 2},
+        )
+        await self.store.create_trace(trace)
+        persisted = await self.store.get_trace(trace.trace_id)
+        self.assertTrue(policy_from_context(persisted.context).is_read_only)
+        with self.assertRaisesRegex(ValueError, "recursive_revision_2_read_only"):
+            await self.runner._prepare_trace(
+                [],
+                RunConfig(trace_id=trace.trace_id, name="continue"),
+            )
+
     async def test_historical_trace_is_persisted_as_legacy_on_continue(self):
         trace = Trace(trace_id="historical", mode="agent", task="old", context={})
         await self.store.create_trace(trace)

tests/test_recursive_runtime_extensions.py

@@ -376,7 +376,7 @@ class RootValidationIntegrationTest(unittest.IsolatedAsyncioTestCase):
             },
         ))
         runner = AgentRunner(trace_store=self.store, llm_call=lambda **_: None)
-        with self.assertRaisesRegex(ValueError, "create a new trace"):
+        with self.assertRaisesRegex(ValueError, "recursive_revision_2_read_only"):
             await runner.run_result([], RunConfig(trace_id=trace_id))
 
     async def test_validator_trace_cannot_be_continued_as_an_agent(self):

tests/test_runtime_hardening.py

@@ -178,7 +178,7 @@ class RunSnapshotHardeningTest(unittest.IsolatedAsyncioTestCase):
             await store.create_trace(legacy)
             runner = AgentRunner(trace_store=store, llm_call=lambda **_: None)
 
-            with self.assertRaisesRegex(ValueError, "predates RunConfig snapshots"):
+            with self.assertRaisesRegex(ValueError, "recursive_revision_2_read_only"):
                 await runner._prepare_existing_trace(
                     RunConfig(trace_id=recursive.trace_id)
                 )