fix(protocol): 统一递归协议事务并修正子角色委派授权

将框架报告、待审核队列、Brief 替换、approved action 消费、Validator checkpoint、根完成门禁、恢复对账和 rewind 全部纳入 TaskProtocolService 的共享 per-trace 锁域；每次写入都锁内重新读取最新 Trace，避免 TaskProgress、TaskReport 和审核状态互相覆盖。

为应用角色区分直接执行工具与显式子角色的可委派工具上限，子 Trace 由父角色允许创建的目标角色 ToolSet 再叠加协议和深度限制，不再要求父角色把孩子工具暴露给自己的模型 schema。

隐藏工具上下文增加持久化 tool_call_id，并补充 Progress 与显式/框架报告并发、恢复分叉和真实应用子角色执行回归。

cyber_agent/application/runtime.py

@@ -43,6 +43,7 @@ class RoleBinding:
     role: AgentRole
     role_hash: str
     tool_names: tuple[str, ...]
+    delegated_tool_names: tuple[str, ...]
     effective_limits: RunLimits
     system_prompt: str
 
@@ -138,12 +139,22 @@ class ApplicationRegistry:
                 for spec in application.tool_set(tool_set_id).tools
             ))
             role_limits = application.run_limits.tighten(role.run_limits)
+            delegated_tool_names = tuple(dict.fromkeys(
+                tool_name
+                for child_role_id in role.allowed_child_roles
+                for tool_name in (
+                    spec.tool_name
+                    for tool_set_id in application.role(child_role_id).tool_sets
+                    for spec in application.tool_set(tool_set_id).tools
+                )
+            ))
             role_bindings[role.role_id] = RoleBinding(
                 role=role,
                 role_hash=stable_hash(
                     role_manifest(application, role, definition_map)
                 ),
                 tool_names=role_tools,
+                delegated_tool_names=delegated_tool_names,
                 effective_limits=role_limits,
                 system_prompt=self._compose_system_prompt(role),
             )

cyber_agent/core/runner.py

@@ -826,10 +826,13 @@ class AgentRunner:
                 "material_usage_recorded": total_material_chars == 0,
             }
             if candidate_ref is None:
-                state["task_report_validation"] = validation_cache
-                await self.trace_store.update_trace(
+                assert self.task_protocol_service is not None
+                await self.task_protocol_service.mutate_state(
                     evaluated_trace_id,
-                    context=evaluated.context,
+                    lambda _trace, fresh_state: fresh_state.__setitem__(
+                        "task_report_validation",
+                        deepcopy(validation_cache),
+                    ),
                 )
 
         candidate_checkpoint = None
@@ -925,20 +928,22 @@ class AgentRunner:
                     )
                 )
             else:
-                fresh = await self.trace_store.get_trace(evaluated_trace_id)
-                if not fresh:
-                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
-                fresh_state = ensure_task_protocol(fresh.context)
-                fresh_cache = fresh_state.get("task_report_validation")
-                if (
-                    not isinstance(fresh_cache, dict)
-                    or fresh_cache.get("plan_hash") != plan.plan_hash
-                ):
-                    raise ValueError("Validation cache changed while recording materials")
-                fresh_cache["material_usage_recorded"] = True
-                await self.trace_store.update_trace(
+                assert self.task_protocol_service is not None
+
+                def mark_material_usage(_trace, fresh_state):
+                    fresh_cache = fresh_state.get("task_report_validation")
+                    if (
+                        not isinstance(fresh_cache, dict)
+                        or fresh_cache.get("plan_hash") != plan.plan_hash
+                    ):
+                        raise ValueError(
+                            "Validation cache changed while recording materials"
+                        )
+                    fresh_cache["material_usage_recorded"] = True
+
+                await self.task_protocol_service.mutate_state(
                     evaluated_trace_id,
-                    context=fresh.context,
+                    mark_material_usage,
                 )
 
         lineage_event = None
@@ -1037,27 +1042,32 @@ class AgentRunner:
                         scope_results=tuple(validation_cache["scope_results"]),
                     )
                     return
-                fresh = await self.trace_store.get_trace(evaluated_trace_id)
-                if not fresh:
-                    raise ValueError(f"Trace not found: {evaluated_trace_id}")
-                fresh_state = ensure_task_protocol(fresh.context)
-                cache = fresh_state.get("task_report_validation")
-                if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
-                    raise ValueError("Validation cache changed while scopes were running")
-                by_scope = {
-                    item.get("scope"): item
-                    for item in cache.get("scope_results", [])
-                    if isinstance(item, dict)
-                }
-                by_scope[result.scope] = result.model_dump(mode="json")
-                cache["scope_results"] = [
-                    by_scope[item]
-                    for item in plan.effective_scopes
-                    if item in by_scope
-                ]
-                await self.trace_store.update_trace(
+                assert self.task_protocol_service is not None
+
+                def persist_trace_scope(_trace, fresh_state):
+                    cache = fresh_state.get("task_report_validation")
+                    if (
+                        not isinstance(cache, dict)
+                        or cache.get("plan_hash") != plan.plan_hash
+                    ):
+                        raise ValueError(
+                            "Validation cache changed while scopes were running"
+                        )
+                    by_scope = {
+                        item.get("scope"): item
+                        for item in cache.get("scope_results", [])
+                        if isinstance(item, dict)
+                    }
+                    by_scope[result.scope] = result.model_dump(mode="json")
+                    cache["scope_results"] = [
+                        by_scope[item]
+                        for item in plan.effective_scopes
+                        if item in by_scope
+                    ]
+
+                await self.task_protocol_service.mutate_state(
                     evaluated_trace_id,
-                    context=fresh.context,
+                    persist_trace_scope,
                 )
 
             run = await validator.validate_plan(
@@ -1091,20 +1101,24 @@ class AgentRunner:
                     ),
                 )
                 return run
-            fresh = await self.trace_store.get_trace(evaluated_trace_id)
-            if not fresh:
-                raise ValueError(f"Trace not found: {evaluated_trace_id}")
-            fresh_state = ensure_task_protocol(fresh.context)
-            cache = fresh_state.get("task_report_validation")
-            if not isinstance(cache, dict) or cache.get("plan_hash") != plan.plan_hash:
-                raise ValueError("Validation cache changed before aggregation")
-            cache["aggregate_result"] = run.result.model_dump(mode="json")
-            cache["scope_results"] = [
-                item.model_dump(mode="json") for item in run.result.scope_results
-            ]
-            await self.trace_store.update_trace(
+            assert self.task_protocol_service is not None
+
+            def persist_aggregate(_trace, fresh_state):
+                cache = fresh_state.get("task_report_validation")
+                if (
+                    not isinstance(cache, dict)
+                    or cache.get("plan_hash") != plan.plan_hash
+                ):
+                    raise ValueError("Validation cache changed before aggregation")
+                cache["aggregate_result"] = run.result.model_dump(mode="json")
+                cache["scope_results"] = [
+                    item.model_dump(mode="json")
+                    for item in run.result.scope_results
+                ]
+
+            await self.task_protocol_service.mutate_state(
                 evaluated_trace_id,
-                context=fresh.context,
+                persist_aggregate,
             )
             if self.event_service is not None:
                 await self.event_service.emit_after_commit(
@@ -2044,30 +2058,43 @@ class AgentRunner:
                                 "validation attempts; create a new trace"
                             )
                     if policy.requires_task_progress:
-                        state = ensure_task_protocol(trace_obj.context)
-                        had_report = state.get("task_report") is not None
-                        report_progress_revision = state.get(
-                            "task_report_progress_revision"
-                        )
-                        if had_report and report_progress_revision is None:
-                            raise ValueError(
-                                "TaskReport is missing its TaskProgress revision binding"
+                        assert self.task_protocol_service is not None
+                        async with self.task_protocol_service.locked_trace(
+                            trace_obj.trace_id
+                        ) as fresh_trace:
+                            state = ensure_task_protocol(fresh_trace.context)
+                            had_report = state.get("task_report") is not None
+                            report_progress_revision = state.get(
+                                "task_report_progress_revision"
                             )
-                        previous_head = state.get("task_progress_head_revision")
-                        rewind_task_progress(state, trace_obj.last_sequence)
-                        if (
-                            had_report
-                            and state.get("task_report_progress_revision") is None
-                        ):
-                            raise ValueError(
-                                "TaskReport references TaskProgress outside the "
-                                "current revision ancestry"
+                            if had_report and report_progress_revision is None:
+                                raise ValueError(
+                                    "TaskReport is missing its TaskProgress revision binding"
+                                )
+                            previous_head = state.get(
+                                "task_progress_head_revision"
                             )
-                        if state.get("task_progress_head_revision") != previous_head:
-                            await self.trace_store.update_trace(
-                                trace_obj.trace_id,
-                                context=trace_obj.context,
+                            rewind_task_progress(
+                                state,
+                                fresh_trace.last_sequence,
                             )
+                            if (
+                                had_report
+                                and state.get("task_report_progress_revision") is None
+                            ):
+                                raise ValueError(
+                                    "TaskReport references TaskProgress outside the "
+                                    "current revision ancestry"
+                                )
+                            if (
+                                state.get("task_progress_head_revision")
+                                != previous_head
+                            ):
+                                await self.trace_store.update_trace(
+                                    fresh_trace.trace_id,
+                                    context=fresh_trace.context,
+                                )
+                            trace_obj = fresh_trace
                     assert self.resource_budget is not None
                     await self.resource_budget.get_usage(root.trace_id)
 
@@ -2721,6 +2748,7 @@ class AgentRunner:
                         goal_id=current_goal_id,
                         goal_tree=goal_tree,
                         sequence=sequence,
+                        tool_call_id=call.tool_call_id,
                         side_branch_ctx=side_branch_ctx,
                         trigger_event=trigger_event,
                     ),
@@ -3748,6 +3776,7 @@ class AgentRunner:
                                     goal_id=current_goal_id,
                                     goal_tree=goal_tree,
                                     sequence=sequence,
+                                    tool_call_id=tc["id"],
                                     side_branch_ctx=side_branch_ctx,
                                     trigger_event=trigger_event_for_tool,
                                 ),
@@ -3897,6 +3926,7 @@ class AgentRunner:
                                     goal_id=current_goal_id,
                                     goal_tree=goal_tree,
                                     sequence=sequence,
+                                    tool_call_id=tc["id"],
                                     side_branch_ctx=side_branch_ctx,
                                     trigger_event=trigger_event_for_tool,
                                 ),
@@ -4069,31 +4099,62 @@ class AgentRunner:
                     trace = fresh_trace
                 policy = policy_from_context(trace.context)
                 if policy.requires_task_protocol:
-                    state = ensure_task_protocol(trace.context)
-                    missing_report = bool(
-                        trace.parent_trace_id and state.get("task_report") is None
-                    )
-                    pending_reviews = bool(state["pending_reviews"])
-                    pending_actions = bool(state["next_actions"])
-                    progress_error = (
-                        task_progress_readiness_error(state)
-                        if policy.requires_task_progress
-                        and not trace.parent_trace_id
-                        else None
-                    )
-                    if (
-                        missing_report
-                        or pending_reviews
-                        or pending_actions
-                        or progress_error
-                    ):
+                    assert self.task_protocol_service is not None
+
+                    def apply_completion_gate(fresh, state):
+                        missing_report = bool(
+                            fresh.parent_trace_id
+                            and state.get("task_report") is None
+                        )
+                        pending_reviews = bool(state["pending_reviews"])
+                        pending_actions = bool(state["next_actions"])
+                        progress_error = (
+                            task_progress_readiness_error(state)
+                            if policy.requires_task_progress
+                            and not fresh.parent_trace_id
+                            else None
+                        )
+                        blocked = bool(
+                            missing_report
+                            or pending_reviews
+                            or pending_actions
+                            or progress_error
+                        )
                         attempts = state["protocol_correction_attempts"]
-                        if attempts < 2:
+                        if blocked and attempts < 2:
                             state["protocol_correction_attempts"] = attempts + 1
-                            await self.trace_store.update_trace(
+                        elif blocked and missing_report:
+                            report = protocol_error_report(
                                 trace_id,
-                                context=trace.context,
+                                "No valid TaskReport after two correction attempts",
+                            )
+                            state["task_report"] = report.model_dump()
+                            state["task_report_submitted_at_sequence"] = sequence
+                            state["task_report_progress_revision"] = state.get(
+                                "task_progress_head_revision"
                             )
+                        return {
+                            "missing_report": missing_report,
+                            "pending_reviews": pending_reviews,
+                            "pending_actions": pending_actions,
+                            "progress_error": progress_error,
+                            "attempts": attempts,
+                            "blocked": blocked,
+                        }
+
+                    gate = await self.task_protocol_service.mutate_state(
+                        trace_id,
+                        apply_completion_gate,
+                    )
+                    missing_report = gate["missing_report"]
+                    pending_reviews = gate["pending_reviews"]
+                    pending_actions = gate["pending_actions"]
+                    progress_error = gate["progress_error"]
+                    if (
+                        gate["blocked"]
+                    ):
+                        attempts = gate["attempts"]
+                        if attempts < 2:
                             if pending_reviews:
                                 required_action = (
                                     "review every pending child report with "
@@ -4121,20 +4182,9 @@ class AgentRunner:
                             })
                             continue
 
-                        if missing_report:
-                            report = protocol_error_report(
-                                trace_id,
-                                "No valid TaskReport after two correction attempts",
-                            )
-                            state["task_report"] = report.model_dump()
-                            state["task_report_submitted_at_sequence"] = sequence
-                            state["task_report_progress_revision"] = state.get(
-                                "task_progress_head_revision"
-                            )
                         completion_status = "failed"
                         await self.trace_store.update_trace(
                             trace_id,
-                            context=trace.context,
                             error_message="Recursive task protocol gate failed",
                         )
                         break
@@ -4187,28 +4237,34 @@ class AgentRunner:
                     candidate_output=response_content,
                     root_validator=True,
                 )
-                refreshed_trace = await self.trace_store.get_trace(trace_id)
-                if not refreshed_trace:
-                    raise ValueError(f"Trace not found after root validation: {trace_id}")
-                trace = refreshed_trace
-                state = ensure_task_protocol(trace.context)
-                validation_cache = state.get("task_report_validation") or {}
-                validation_record = {
-                    **validation_run.result.model_dump(),
-                    "validation_plan": validation_cache.get("validation_plan"),
-                    "validated_at_sequence": head_seq,
-                }
-                state["root_validation_history"].append(validation_record)
-                state["root_validation_attempts"] += 1
-                state["root_validation_passed"] = (
-                    validation_run.result.outcome == "passed"
-                )
-                await self.trace_store.update_trace(
-                    trace_id,
-                    context=trace.context,
+                assert self.task_protocol_service is not None
+
+                def record_root_validation(fresh, state):
+                    validation_cache = state.get("task_report_validation") or {}
+                    validation_record = {
+                        **validation_run.result.model_dump(),
+                        "validation_plan": validation_cache.get("validation_plan"),
+                        "validated_at_sequence": head_seq,
+                    }
+                    state["root_validation_history"].append(validation_record)
+                    state["root_validation_attempts"] += 1
+                    state["root_validation_passed"] = (
+                        validation_run.result.outcome == "passed"
+                    )
+                    return (
+                        state["root_validation_passed"],
+                        state["root_validation_attempts"],
+                        fresh,
+                    )
+
+                root_passed, root_attempts, trace = (
+                    await self.task_protocol_service.mutate_state(
+                        trace_id,
+                        record_root_validation,
+                    )
                 )
-                if not state["root_validation_passed"]:
-                    if state["root_validation_attempts"] < 2:
+                if not root_passed:
+                    if root_attempts < 2:
                         validation_feedback = {
                             "role": "user",
                             "content": (
@@ -4269,30 +4325,50 @@ class AgentRunner:
                 trace = fresh_trace
             policy = policy_from_context(trace.context)
             if policy.requires_task_protocol:
-                state = ensure_task_protocol(trace.context)
-                if trace.parent_trace_id and state.get("task_report") is None:
-                    report = protocol_error_report(
-                        trace_id,
-                        "Agent loop ended without a valid TaskReport",
-                    )
-                    state["task_report"] = report.model_dump()
-                    state["task_report_submitted_at_sequence"] = sequence
-                    state["task_report_progress_revision"] = state.get(
-                        "task_progress_head_revision"
+                assert self.task_protocol_service is not None
+
+                def close_incomplete_protocol(fresh, state):
+                    missing_report = bool(
+                        fresh.parent_trace_id
+                        and state.get("task_report") is None
                     )
+                    if missing_report:
+                        report = protocol_error_report(
+                            trace_id,
+                            "Agent loop ended without a valid TaskReport",
+                        )
+                        state["task_report"] = report.model_dump()
+                        state["task_report_submitted_at_sequence"] = sequence
+                        state["task_report_progress_revision"] = state.get(
+                            "task_progress_head_revision"
+                        )
+                    return {
+                        "missing_report": missing_report,
+                        "pending_reviews": bool(state["pending_reviews"]),
+                        "pending_actions": bool(state["next_actions"]),
+                        "root_validation_passed": bool(
+                            state.get("root_validation_passed")
+                        ),
+                        "is_root": not fresh.parent_trace_id,
+                    }
+
+                closed = await self.task_protocol_service.mutate_state(
+                    trace_id,
+                    close_incomplete_protocol,
+                )
+                if closed["missing_report"]:
                     completion_status = "failed"
                     await self.trace_store.update_trace(
                         trace_id,
-                        context=trace.context,
                         error_message="Recursive task protocol gate failed",
                     )
-                if state["pending_reviews"]:
+                if closed["pending_reviews"]:
                     completion_status = "failed"
-                if state["next_actions"]:
+                if closed["pending_actions"]:
                     completion_status = "failed"
                 if (
-                    not trace.parent_trace_id
-                    and not state.get("root_validation_passed")
+                    closed["is_root"]
+                    and not closed["root_validation_passed"]
                 ):
                     completion_status = "failed"
 
@@ -4429,7 +4505,47 @@ class AgentRunner:
             goal_tree.current_id = new_tree.current_id
 
         trace = await self.trace_store.get_trace(trace_id)
+        projected_statuses: dict[str, str] = {}
         if trace and policy_from_context(trace.context).requires_task_protocol:
+            projected_statuses = await self._rewind_protocol_state(
+                trace_id,
+                cutoff,
+                rewind_operation_id,
+            )
+        else:
+            await self.trace_store.update_trace(trace_id, head_sequence=cutoff)
+        if goal_tree and projected_statuses:
+            for goal_id, status in projected_statuses.items():
+                goal = goal_tree.find(goal_id)
+                if goal:
+                    goal.status = status
+            await self.trace_store.update_goal_tree(trace_id, goal_tree)
+
+        # 4. 协议状态与 head_sequence 已在同一临界区落盘
+        await self.trace_store.append_event(trace_id, "rewind", rewind_payload)
+        if self.event_service is not None:
+            await self.event_service.emit_after_commit(
+                source_trace_id=trace_id,
+                event_type="run.rewound",
+                event_key=f"run.rewound:{rewind_operation_id}",
+                effective_at_sequence=cutoff,
+                payload={"after_sequence": cutoff},
+            )
+
+        # 5. 返回 next sequence（全局递增，不复用）
+        max_seq = max((m.sequence for m in all_messages), default=0)
+        return max_seq + 1
+
+    async def _rewind_protocol_state(
+        self,
+        trace_id: str,
+        cutoff: int,
+        rewind_operation_id: str,
+    ) -> dict[str, str]:
+        """Rebuild protocol state and move the Trace head in one lock domain."""
+        assert self.trace_store is not None
+        assert self.task_protocol_service is not None
+        async with self.task_protocol_service.locked_trace(trace_id) as trace:
             state = ensure_task_protocol(trace.context)
             submitted_at = state.get("task_report_submitted_at_sequence")
             if isinstance(submitted_at, int) and submitted_at > cutoff:
@@ -4452,7 +4568,8 @@ class AgentRunner:
                 if entry.get("received_at_sequence", 0) <= cutoff
             }
             reverted_reviews = [
-                review for review in state["reviews"]
+                review
+                for review in state["reviews"]
                 if review.get("reviewed_at_sequence", 0) > cutoff
             ]
             for review in reverted_reviews:
@@ -4465,11 +4582,13 @@ class AgentRunner:
                 ):
                     state["pending_reviews"][child_id] = entry
             state["reviews"] = [
-                review for review in state["reviews"]
+                review
+                for review in state["reviews"]
                 if review.get("reviewed_at_sequence", 0) <= cutoff
             ]
             state["next_actions"] = [
-                action for action in state["next_actions"]
+                action
+                for action in state["next_actions"]
                 if action.get("created_at_sequence", 0) <= cutoff
             ]
             while (
@@ -4496,7 +4615,8 @@ class AgentRunner:
                         item.get("evaluated_at_sequence", 0),
                     )
                     or 0
-                ) <= cutoff
+                )
+                <= cutoff
             ]
             state["root_validation_history"] = root_history
             state["root_validation_attempts"] = len(root_history)
@@ -4518,8 +4638,11 @@ class AgentRunner:
                     "payload": {"after_sequence": cutoff},
                 })
                 del operations[:-64]
-            await self.trace_store.update_trace(trace_id, context=trace.context)
-
+            await self.trace_store.update_trace(
+                trace_id,
+                context=trace.context,
+                head_sequence=cutoff,
+            )
             projected_statuses: dict[str, str] = {}
             for entry in state["pending_reviews"].values():
                 if entry.get("goal_id"):
@@ -4528,28 +4651,7 @@ class AgentRunner:
                 goal_id = action.get("goal_id")
                 if goal_id and goal_id not in projected_statuses:
                     projected_statuses[goal_id] = "in_progress"
-            if goal_tree and projected_statuses:
-                for goal_id, status in projected_statuses.items():
-                    goal = goal_tree.find(goal_id)
-                    if goal:
-                        goal.status = status
-                await self.trace_store.update_goal_tree(trace_id, goal_tree)
-
-        # 4. 更新 head_sequence 到 rewind 点
-        await self.trace_store.update_trace(trace_id, head_sequence=cutoff)
-        await self.trace_store.append_event(trace_id, "rewind", rewind_payload)
-        if self.event_service is not None:
-            await self.event_service.emit_after_commit(
-                source_trace_id=trace_id,
-                event_type="run.rewound",
-                event_key=f"run.rewound:{rewind_operation_id}",
-                effective_at_sequence=cutoff,
-                payload={"after_sequence": cutoff},
-            )
-
-        # 5. 返回 next sequence（全局递增，不复用）
-        max_seq = max((m.sequence for m in all_messages), default=0)
-        return max_seq + 1
+            return projected_statuses
 
     def _find_safe_cutoff(self, messages: List[Message], after_sequence: int) -> int:
         """
@@ -5621,6 +5723,7 @@ class AgentRunner:
         goal_id: Optional[str],
         goal_tree: Optional[GoalTree],
         sequence: int,
+        tool_call_id: str | None,
         side_branch_ctx: Optional[SideBranchContext],
         trigger_event: Optional[str],
     ) -> Dict[str, Any]:
@@ -5649,6 +5752,7 @@ class AgentRunner:
                 else None
             ),
             "sequence": sequence,
+            "tool_call_id": tool_call_id,
             "side_branch": {
                 "type": side_branch_ctx.type,
                 "branch_id": side_branch_ctx.branch_id,

cyber_agent/core/task_protocol_service.py

@@ -3,7 +3,10 @@
 from __future__ import annotations
 
 import asyncio
+from contextlib import asynccontextmanager
 from copy import deepcopy
+from collections.abc import AsyncIterator, Callable
+from typing import Any
 from weakref import WeakValueDictionary
 
 from cyber_agent.core.agent_mode import require_mutable_trace_policy
@@ -15,11 +18,12 @@ from cyber_agent.core.task_protocol import (
     ensure_task_protocol,
     task_contract_ref,
 )
+from cyber_agent.trace.models import Trace
 from cyber_agent.trace.protocols import TraceStore
 
 
 _TRACE_LOCKS: WeakValueDictionary[
-    tuple[asyncio.AbstractEventLoop, str], asyncio.Lock
+    tuple[asyncio.AbstractEventLoop, int, str], asyncio.Lock
 ] = WeakValueDictionary()
 
 
@@ -35,13 +39,55 @@ class TaskProtocolService:
         self.event_service = event_service
 
     def _lock(self, trace_id: str) -> asyncio.Lock:
-        key = (asyncio.get_running_loop(), trace_id)
+        key = (asyncio.get_running_loop(), id(self.trace_store), trace_id)
         lock = _TRACE_LOCKS.get(key)
         if lock is None:
             lock = asyncio.Lock()
             _TRACE_LOCKS[key] = lock
         return lock
 
+    @asynccontextmanager
+    async def transaction(self, trace_id: str) -> AsyncIterator[None]:
+        """Serialize one complete TaskProtocol read-modify-write command."""
+        async with self._lock(trace_id):
+            yield
+
+    @asynccontextmanager
+    async def transaction_many(
+        self,
+        *trace_ids: str,
+    ) -> AsyncIterator[None]:
+        """Lock multiple protocol records in stable order to avoid deadlocks."""
+        locks = [self._lock(item) for item in sorted(set(trace_ids))]
+        for lock in locks:
+            await lock.acquire()
+        try:
+            yield
+        finally:
+            for lock in reversed(locks):
+                lock.release()
+
+    @asynccontextmanager
+    async def locked_trace(self, trace_id: str) -> AsyncIterator[Trace]:
+        """Load one fresh Trace while holding its protocol command lock."""
+        async with self.transaction(trace_id):
+            trace = await self.trace_store.get_trace(trace_id)
+            if trace is None:
+                raise TaskProtocolStateError(f"Trace not found: {trace_id}")
+            yield trace
+
+    async def mutate_state(
+        self,
+        trace_id: str,
+        mutation: Callable[[Trace, dict[str, Any]], Any],
+    ) -> Any:
+        """Apply one synchronous protocol mutation to a fresh persisted Trace."""
+        async with self.locked_trace(trace_id) as trace:
+            state = ensure_task_protocol(trace.context)
+            result = mutation(trace, state)
+            await self.trace_store.update_trace(trace_id, context=trace.context)
+            return result
+
     async def update_progress(
         self,
         trace_id: str,
@@ -50,7 +96,7 @@ class TaskProtocolService:
         progress: TaskProgress | dict,
         effective_at_sequence: int,
     ) -> TaskProgressRevision:
-        async with self._lock(trace_id):
+        async with self.transaction(trace_id):
             trace = await self.trace_store.get_trace(trace_id)
             if trace is None:
                 raise TaskProtocolStateError(f"Trace not found: {trace_id}")

cyber_agent/tools/builtin/context.py

@@ -102,7 +102,7 @@ async def read_context_ref(
         return ToolResult(
             title="上下文引用读取失败",
             output="",
-            error="read_context_ref is only available to Recursive revision 2 business Agents",
+            error="read_context_ref requires a writable Recursive business Agent",
         )
     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(

cyber_agent/tools/builtin/subagent.py

@@ -9,6 +9,7 @@ revision 1 的 ``remote_*`` 仍通过 KnowHub HTTP 路由，revision 2 失败关
 import asyncio
 import json
 import os
+from copy import deepcopy
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 
@@ -39,6 +40,7 @@ from cyber_agent.core.task_protocol import (
     replace_task_brief,
     stopped_task_report,
 )
+from cyber_agent.core.task_protocol_service import TaskProtocolService
 from cyber_agent.core.context_policy import (
     CONTEXT_ACCESS_KEY,
     ContextPolicyError,
@@ -245,64 +247,68 @@ async def _load_or_create_task_report(
     fallback_reason: str,
     child_result_status: str,
     generated_at_sequence: int,
+    *,
+    task_protocol_service: TaskProtocolService | None = None,
 ) -> TaskReport:
     """读取孩子已提交的报告，或为异常、停止生成框架报告。
 
     子任务执行结束后由报告汇合阶段调用，将异常和停止统一转成可审核的 ``TaskReport``。
     """
-    child = await store.get_trace(child_trace_id)
-    if not child:
-        return protocol_error_report(child_trace_id, fallback_reason)
-    state = ensure_task_protocol(child.context)
-    report_data = state.get("task_report")
-    execution_stopped = (
-        child_result_status == "stopped"
-        or child.status == "stopped"
-    )
-    execution_failed = (
-        child_result_status != "completed"
-        or child.status != "completed"
-    )
-    if report_data:
-        try:
-            report = TaskReport.model_validate(report_data)
-            if report.child_trace_id != child_trace_id:
-                if report.model_dump() not in state["report_history"]:
-                    state["report_history"].append(report.model_dump())
-                fallback_reason = (
-                    "Child persisted a TaskReport whose child_trace_id does not "
-                    "match its Trace"
-                )
-            elif execution_stopped:
-                if report.outcome in {"failed", "protocol_error"}:
+    service = task_protocol_service or TaskProtocolService(store)
+    async with service.transaction(child_trace_id):
+        child = await store.get_trace(child_trace_id)
+        if not child:
+            return protocol_error_report(child_trace_id, fallback_reason)
+        state = ensure_task_protocol(child.context)
+        report_data = state.get("task_report")
+        execution_stopped = (
+            child_result_status == "stopped"
+            or child.status == "stopped"
+        )
+        execution_failed = (
+            child_result_status != "completed"
+            or child.status != "completed"
+        )
+        if report_data:
+            try:
+                report = TaskReport.model_validate(report_data)
+                if report.child_trace_id != child_trace_id:
+                    if report.model_dump() not in state["report_history"]:
+                        state["report_history"].append(report.model_dump())
+                    fallback_reason = (
+                        "Child persisted a TaskReport whose child_trace_id does not "
+                        "match its Trace"
+                    )
+                elif execution_stopped:
+                    if report.outcome in {"failed", "protocol_error"}:
+                        return report
+                    if report.model_dump() not in state["report_history"]:
+                        state["report_history"].append(report.model_dump())
+                    fallback_reason = "Child Agent execution was stopped"
+                elif not execution_failed or report.outcome in {"failed", "protocol_error"}:
                     return report
-                if report.model_dump() not in state["report_history"]:
-                    state["report_history"].append(report.model_dump())
-                fallback_reason = "Child Agent execution was stopped"
-            elif not execution_failed or report.outcome in {"failed", "protocol_error"}:
-                return report
-            else:
-                if report.model_dump() not in state["report_history"]:
-                    state["report_history"].append(report.model_dump())
-                fallback_reason = (
-                    f"Child execution ended as {child_result_status} after submitting "
-                    f"a {report.outcome} TaskReport: {fallback_reason}"
-                )
-        except ValidationError:
-            fallback_reason = "Child persisted an invalid TaskReport"
-    report = (
-        stopped_task_report(child_trace_id, fallback_reason)
-        if execution_stopped
-        else protocol_error_report(child_trace_id, fallback_reason)
-    )
-    state["task_report"] = report.model_dump()
-    state["task_report_submitted_at_sequence"] = generated_at_sequence
-    state["task_report_progress_revision"] = state.get(
-        "task_progress_head_revision"
-    )
-    state["task_report_validation"] = None
-    await store.update_trace(child_trace_id, context=child.context)
-    return report
+                else:
+                    if report.model_dump() not in state["report_history"]:
+                        state["report_history"].append(report.model_dump())
+                    fallback_reason = (
+                        f"Child execution ended as {child_result_status} after submitting "
+                        f"a {report.outcome} TaskReport: {fallback_reason}"
+                    )
+            except ValidationError:
+                fallback_reason = "Child persisted an invalid TaskReport"
+        report = (
+            stopped_task_report(child_trace_id, fallback_reason)
+            if execution_stopped
+            else protocol_error_report(child_trace_id, fallback_reason)
+        )
+        state["task_report"] = report.model_dump()
+        state["task_report_submitted_at_sequence"] = generated_at_sequence
+        state["task_report_progress_revision"] = state.get(
+            "task_progress_head_revision"
+        )
+        state["task_report_validation"] = None
+        await store.update_trace(child_trace_id, context=child.context)
+        return report
 
 
 async def _record_pending_task_reports(
@@ -317,8 +323,9 @@ async def _record_pending_task_reports(
 
     ``_run_agents`` 汇合孩子后按输入顺序调用，必要时触发 Validator 并更新 Goal 状态。
     """
-    state = ensure_task_protocol(parent_trace.context)
+    service = getattr(runner, "task_protocol_service", None) or TaskProtocolService(store)
     reports: List[TaskReport] = []
+    pending_entries: dict[str, dict[str, Any]] = {}
     for child_trace_id, result in child_results:
         reason = result.get("error") or result.get("summary") or "Child ended without TaskReport"
         report = await _load_or_create_task_report(
@@ -327,6 +334,7 @@ async def _record_pending_task_reports(
             reason,
             result.get("status", "unknown"),
             received_at_sequence,
+            task_protocol_service=service,
         )
         child_trace = await store.get_trace(child_trace_id)
         if not child_trace:
@@ -375,9 +383,19 @@ async def _record_pending_task_reports(
             received_at_sequence=received_at_sequence,
         )
         pending_entry["candidate_validations"] = candidate_validations
-        state["pending_reviews"][child_trace_id] = pending_entry
+        pending_entries[child_trace_id] = pending_entry
         reports.append(report)
-    await store.update_trace(parent_trace.trace_id, context=parent_trace.context)
+    async with service.transaction(parent_trace.trace_id):
+        fresh_parent = await store.get_trace(parent_trace.trace_id)
+        if fresh_parent is None:
+            raise ValueError(f"Parent Trace not found: {parent_trace.trace_id}")
+        state = ensure_task_protocol(fresh_parent.context)
+        state["pending_reviews"].update(pending_entries)
+        await store.update_trace(
+            parent_trace.trace_id,
+            context=fresh_parent.context,
+        )
+        parent_trace.context = fresh_parent.context
     if goal_id:
         await store.update_goal(
             parent_trace.trace_id,
@@ -505,22 +523,31 @@ def _get_allowed_tools(
             # Legacy 保留原行为：从全局 Registry 取工具全集。
             allowed_tools = registered_tools
 
-        blocked_tools = {"evaluate", "bash_command"}
-        if not policy.requires_task_protocol:
-            blocked_tools.update({
-                "submit_task_report",
-                "review_task_result",
-                "update_task_progress",
-                "read_context_ref",
-            })
-        elif not policy.requires_task_progress:
-            blocked_tools.add("update_task_progress")
-        if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
-            blocked_tools.add("agent")
-        return sorted(allowed_tools - blocked_tools)
+        return _filter_allowed_tools(allowed_tools, agent_depth, policy)
     return None
 
 
+def _filter_allowed_tools(
+    tool_names: set[str],
+    agent_depth: int,
+    policy: AgentPolicy,
+) -> list[str]:
+    """Apply framework protocol/depth restrictions to one capability set."""
+    blocked_tools = {"evaluate", "bash_command"}
+    if not policy.requires_task_protocol:
+        blocked_tools.update({
+            "submit_task_report",
+            "review_task_result",
+            "update_task_progress",
+            "read_context_ref",
+        })
+    elif not policy.requires_task_progress:
+        blocked_tools.add("update_task_progress")
+    if policy.mode is AgentMode.LEGACY or agent_depth >= policy.max_depth:
+        blocked_tools.add("agent")
+    return sorted(tool_names - blocked_tools)
+
+
 async def _resolve_trace_lineage(store, trace: Trace) -> tuple[int, str]:
     """返回 Trace 的 (agent_depth, root_trace_id)。
 
@@ -794,9 +821,14 @@ async def _run_agents(
         policy = require_mutable_trace_policy(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
+    protocol_service = (
+        getattr(runner, "task_protocol_service", None)
+        or TaskProtocolService(store)
+    )
     application_binding = getattr(runner, "application_binding", None)
     application_ref = parent_trace.context.get("application_ref")
     target_role_binding = None
+    parent_role_binding = None
     target_role_id = agent_type
     if application_ref is not None:
         if application_binding is None:
@@ -819,10 +851,10 @@ async def _run_agents(
             }
         parent_role_id = parent_trace.context.get("application_role_id")
         try:
-            parent_role = application_binding.role(parent_role_id)
+            parent_role_binding = application_binding.role(parent_role_id)
         except ValueError as exc:
             return {"status": "failed", "error": str(exc)}
-        allowed_child_roles = parent_role.role.allowed_child_roles
+        allowed_child_roles = parent_role_binding.role.allowed_child_roles
         if target_role_id is None and len(allowed_child_roles) == 1:
             target_role_id = allowed_child_roles[0]
         if target_role_id not in allowed_child_roles:
@@ -900,7 +932,7 @@ async def _run_agents(
                 "error": "Review pending child TaskReports before creating another child",
             }
         if not task_briefs or len(task_briefs) != len(tasks):
-            return {"status": "failed", "error": "Recursive revision 2 requires task_brief"}
+            return {"status": "failed", "error": "Writable Recursive runs require task_brief"}
         if protocol_state["next_actions"]:
             approved_action = protocol_state["next_actions"][0]
             decision = approved_action["decision"]
@@ -1073,7 +1105,8 @@ async def _run_agents(
         }]
         if task_briefs:
             assert root_task_anchor is not None
-            child_state = ensure_task_protocol(child_context)
+            prepared_context = deepcopy(child_context)
+            child_state = ensure_task_protocol(prepared_context)
             normalized_dump = task_briefs[0].model_dump(mode="json")
             new_context_access = build_child_context_access(
                 parent_context=parent_trace.context,
@@ -1092,8 +1125,8 @@ async def _run_agents(
                     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)
+            prepared_context[CONTEXT_ACCESS_KEY] = new_context_access
+            persist_root_task_anchor(prepared_context, root_task_anchor)
             if (
                 target_role_binding is not None
                 and getattr(runner, "context_provider", None) is not None
@@ -1104,7 +1137,7 @@ async def _run_agents(
 
                 await load_application_context(
                     application_binding,
-                    child_context,
+                    prepared_context,
                     ContextRequest(
                         application_ref=application_binding.application_ref,
                         root_trace_id=root_trace_id,
@@ -1122,8 +1155,37 @@ async def _run_agents(
                     task_brief=task_briefs[0],
                     granted_at_sequence=(existing.last_sequence or 0) + 1,
                 )
-            existing.context = child_context
-            await store.update_trace(existing.trace_id, context=existing.context)
+            async with protocol_service.transaction(existing.trace_id):
+                fresh_child = await store.get_trace(existing.trace_id)
+                if fresh_child is None:
+                    return {
+                        "status": "failed",
+                        "error": f"continue_from Trace not found: {existing.trace_id}",
+                    }
+                fresh_context = apply_policy_to_context({
+                    **fresh_child.context,
+                    "agent_depth": child_depth,
+                    "root_trace_id": root_trace_id,
+                }, policy)
+                fresh_state = ensure_task_protocol(fresh_context)
+                if fresh_state.get("task_brief") != normalized_dump:
+                    replace_task_brief(
+                        fresh_state,
+                        task_briefs[0],
+                        effective_at_sequence=(fresh_child.last_sequence or 0) + 1,
+                    )
+                fresh_context[CONTEXT_ACCESS_KEY] = deepcopy(
+                    prepared_context[CONTEXT_ACCESS_KEY]
+                )
+                persist_root_task_anchor(fresh_context, root_task_anchor)
+                await store.update_trace(
+                    fresh_child.trace_id,
+                    context=fresh_context,
+                )
+                fresh_child.context = fresh_context
+                existing = fresh_child
+                child_context = fresh_context
+                child_records[0]["context"] = fresh_context
     else:
         parent_depth, root_trace_id = await _resolve_trace_lineage(store, parent_trace)
         effective_max_depth = int(
@@ -1345,11 +1407,25 @@ async def _run_agents(
         goal_started = True
 
     if approved_action and protocol_state is not None:
-        protocol_state["next_actions"].pop(0)
-        protocol_state["protocol_correction_attempts"] = 0
-        await run_initialization(
-            store.update_trace(trace_id, context=parent_trace.context)
-        )
+        try:
+            async with protocol_service.transaction(trace_id):
+                fresh_parent = await store.get_trace(trace_id)
+                if fresh_parent is None:
+                    raise ValueError(f"Trace not found: {trace_id}")
+                fresh_state = ensure_task_protocol(fresh_parent.context)
+                if not fresh_state["next_actions"]:
+                    raise ValueError("Approved next action was already consumed")
+                if fresh_state["next_actions"][0] != approved_action:
+                    raise ValueError(
+                        "Approved next action changed before execution"
+                    )
+                fresh_state["next_actions"].pop(0)
+                fresh_state["protocol_correction_attempts"] = 0
+                await store.update_trace(trace_id, context=fresh_parent.context)
+                parent_trace.context = fresh_parent.context
+        except Exception as exc:
+            await fail_created_children(exc)
+            raise
 
     # 创建延迟执行规格。Trace 已按批次预创建，但不会提前创建 coroutine，
     # 因而排队孩子被停止时不会产生未 await coroutine 或模型调用。
@@ -1392,8 +1468,12 @@ async def _run_agents(
         agent_msgs = list(msgs) + [{"role": "user", "content": task_item}]
         allowed_tools = _get_allowed_tools(context, child_depth, policy)
         if target_role_binding is not None:
-            allowed_tools = sorted(
-                set(allowed_tools or []) & set(target_role_binding.tool_names)
+            assert parent_role_binding is not None
+            allowed_tools = _filter_allowed_tools(
+                set(parent_role_binding.delegated_tool_names)
+                & set(target_role_binding.tool_names),
+                child_depth,
+                policy,
             )
 
         debug = getattr(runner, 'debug', False)
@@ -1728,7 +1808,7 @@ async def _run_remote_agent(
 
 # ===== 工具定义 =====
 
-@tool(description="创建子 Agent 执行任务；Recursive revision 2 仅允许本地结构化委派", hidden_params=["context"], groups=["core"])
+@tool(description="创建子 Agent 执行任务；可写 Recursive 协议仅允许本地结构化委派", hidden_params=["context"], groups=["core"])
 async def agent(
     task: Optional[Union[str, List[str]]] = None,
     task_brief: Optional[Union[TaskBrief, List[TaskBrief]]] = None,
@@ -1791,7 +1871,6 @@ async def agent(
         policy = policy_from_context(parent_trace.context)
     except ValueError as exc:
         return {"status": "failed", "error": str(exc)}
-
     if isinstance(messages, str):
         try:
             messages = json.loads(messages)
@@ -1818,7 +1897,7 @@ async def agent(
             return {
                 "status": "failed",
                 "error": (
-                    "Recursive revision 2 does not support remote agents; "
+                    "Writable Recursive protocol does not support remote agents; "
                     "use a local task_brief delegation"
                 ),
             }
@@ -1841,7 +1920,7 @@ async def agent(
     if policy.requires_task_protocol and messages is not None:
         return {
             "status": "failed",
-            "error": "Recursive revision 2 does not accept messages; pass bounded context in task_brief",
+            "error": "Writable Recursive protocol does not accept messages; pass bounded context in task_brief",
         }
 
     parsed_task_briefs: Optional[List[TaskBrief]] = None
@@ -1849,7 +1928,7 @@ async def agent(
         if task is not None or task_brief is None:
             return {
                 "status": "failed",
-                "error": "Recursive revision 2 requires task_brief and does not accept task",
+                "error": "Writable Recursive runs require task_brief and do not accept task",
             }
         if isinstance(task_brief, str):
             try:
@@ -1878,7 +1957,7 @@ async def agent(
         single = len(tasks) == 1
     else:
         if task_brief is not None:
-            return {"status": "failed", "error": "task_brief requires Recursive revision 2"}
+            return {"status": "failed", "error": "task_brief requires a writable Recursive run"}
         if task is None:
             return {"status": "failed", "error": "task is required"}
 

cyber_agent/tools/builtin/task_protocol.py

@@ -33,7 +33,10 @@ from cyber_agent.core.task_protocol import (
     task_progress_readiness_error,
     ensure_task_protocol,
 )
-from cyber_agent.core.task_protocol_service import TaskProtocolStateError
+from cyber_agent.core.task_protocol_service import (
+    TaskProtocolService,
+    TaskProtocolStateError,
+)
 from cyber_agent.core.validation import ValidationResult
 from cyber_agent.application.candidate import CandidateReviewAction
 from cyber_agent.tools import tool
@@ -103,6 +106,21 @@ async def update_task_progress(
 async def submit_task_report(
     task_report: TaskReportSubmission,
     context: dict | None = None,
+) -> dict[str, Any]:
+    if not context:
+        return _error("context is required")
+    store = context.get("store")
+    trace_id = context.get("trace_id")
+    if not store or not trace_id:
+        return _error("store and trace_id are required")
+    service = context.get("task_protocol_service") or TaskProtocolService(store)
+    async with service.transaction(trace_id):
+        return await _submit_task_report_locked(task_report, context)
+
+
+async def _submit_task_report_locked(
+    task_report: TaskReportSubmission,
+    context: dict | None = None,
 ) -> dict[str, Any]:
     """提交当前 Recursive 非根 Agent 的结构化任务报告。
 
@@ -123,7 +141,7 @@ async def submit_task_report(
     except ValueError as exc:
         return _error(str(exc))
     if not policy.requires_task_protocol:
-        return _error("submit_task_report is only available in Recursive revision 2")
+        return _error("submit_task_report requires a writable Recursive protocol")
     if not trace.parent_trace_id:
         return _error("Root Agent does not submit a TaskReport")
 
@@ -222,6 +240,32 @@ async def review_task_result(
     approved_next_task: TaskBrief | None = None,
     candidate_actions: list[CandidateReviewAction] | None = None,
     context: dict | None = None,
+) -> dict[str, Any]:
+    if not context:
+        return _error("context is required")
+    store = context.get("store")
+    parent_trace_id = context.get("trace_id")
+    if not store or not parent_trace_id:
+        return _error("store and trace_id are required")
+    service = context.get("task_protocol_service") or TaskProtocolService(store)
+    async with service.transaction_many(parent_trace_id, child_trace_id):
+        return await _review_task_result_locked(
+            child_trace_id=child_trace_id,
+            decision=decision,
+            reason=reason,
+            approved_next_task=approved_next_task,
+            candidate_actions=candidate_actions,
+            context=context,
+        )
+
+
+async def _review_task_result_locked(
+    child_trace_id: str,
+    decision: ReviewDecision,
+    reason: str,
+    approved_next_task: TaskBrief | None = None,
+    candidate_actions: list[CandidateReviewAction] | None = None,
+    context: dict | None = None,
 ) -> dict[str, Any]:
     """审核一份直属孩子的待处理 ``TaskReport``。
 
@@ -243,7 +287,7 @@ async def review_task_result(
     except ValueError as exc:
         return _error(str(exc))
     if not policy.requires_task_protocol:
-        return _error("review_task_result is only available in Recursive revision 2")
+        return _error("review_task_result requires a writable Recursive protocol")
     if child.parent_trace_id != parent_trace_id or child.uid != parent.uid:
         return _error("TaskReport must belong to a direct child with the same owner")
 

tests/test_task_progress.py

@@ -1,4 +1,5 @@
 import asyncio
+from copy import deepcopy
 import tempfile
 import unittest
 
@@ -39,6 +40,8 @@ from cyber_agent.core.task_protocol import (
     task_progress_readiness_error,
 )
 from cyber_agent.core.task_protocol_service import TaskProtocolService
+from cyber_agent.tools.builtin.task_protocol import submit_task_report
+from cyber_agent.tools.builtin.subagent import _load_or_create_task_report
 from cyber_agent.core.validation import (
     ValidationPolicy,
     ValidatorSettings,
@@ -303,6 +306,130 @@ class TaskProtocolServiceTest(unittest.IsolatedAsyncioTestCase):
         state = trace.context["task_protocol"]
         self.assertEqual(2, len(state["task_progress_revisions"]))
 
+    async def test_progress_and_report_share_one_atomic_command_boundary(self):
+        root = await self.store.get_trace("root")
+        brief = TaskBrief(
+            objective="prepare one partial result",
+            reason="exercise concurrent protocol commands",
+            completion_criteria=["record current state"],
+            expected_outputs=["partial report"],
+        )
+        child_context = deepcopy(root.context)
+        child_context["task_protocol"] = new_task_protocol(brief)
+        initialize_task_progress(
+            child_context["task_protocol"],
+            effective_at_sequence=1,
+        )
+        await self.store.create_trace(Trace(
+            trace_id="child",
+            mode="agent",
+            task="child",
+            uid=root.uid,
+            parent_trace_id="root",
+            context=child_context,
+        ))
+
+        progress_result, report_result = await asyncio.gather(
+            self.service.update_progress(
+                "child",
+                expected_revision=1,
+                progress=TaskProgress(
+                    phase="investigating",
+                    decision_rationale="recorded before partial submission",
+                ),
+                effective_at_sequence=2,
+            ),
+            submit_task_report(
+                task_report={
+                    "summary": "partial result",
+                    "outcome": "partial",
+                    "validation": {
+                        "hard_passed": False,
+                        "open_issues": ["work remains"],
+                    },
+                    "next_step_suggestion": {
+                        "direction": "NONE",
+                        "reason": "parent should decide after partial submission",
+                    },
+                    "remaining_issues": ["work remains"],
+                },
+                context={
+                    "store": self.store,
+                    "task_protocol_service": self.service,
+                    "trace_id": "child",
+                    "sequence": 3,
+                },
+            ),
+            return_exceptions=True,
+        )
+        persisted = await self.store.get_trace("child")
+        state = persisted.context["task_protocol"]
+        if isinstance(progress_result, Exception):
+            self.assertEqual(1, state["task_progress_head_revision"])
+        else:
+            self.assertEqual(2, state["task_progress_head_revision"])
+            self.assertEqual(2, len(state["task_progress_revisions"]))
+        if not isinstance(report_result, Exception):
+            self.assertEqual("completed", report_result["status"], report_result)
+            self.assertEqual(
+                state["task_progress_head_revision"],
+                state["task_report_progress_revision"],
+            )
+
+    async def test_progress_and_framework_report_cannot_overwrite_each_other(self):
+        root = await self.store.get_trace("root")
+        child_context = deepcopy(root.context)
+        child_context["task_protocol"] = new_task_protocol(TaskBrief(
+            objective="record a recoverable result",
+            reason="exercise framework report convergence",
+            completion_criteria=["preserve the latest progress revision"],
+            expected_outputs=["one report"],
+        ))
+        initialize_task_progress(
+            child_context["task_protocol"],
+            effective_at_sequence=1,
+        )
+        await self.store.create_trace(Trace(
+            trace_id="framework-child",
+            mode="agent",
+            uid=root.uid,
+            parent_trace_id="root",
+            context=child_context,
+        ))
+
+        progress_result, report = await asyncio.gather(
+            self.service.update_progress(
+                "framework-child",
+                expected_revision=1,
+                progress=TaskProgress(
+                    phase="investigating",
+                    decision_rationale="must not be lost by framework recovery",
+                ),
+                effective_at_sequence=2,
+            ),
+            _load_or_create_task_report(
+                self.store,
+                "framework-child",
+                "simulated child interruption",
+                "failed",
+                3,
+                task_protocol_service=self.service,
+            ),
+            return_exceptions=True,
+        )
+        persisted = await self.store.get_trace("framework-child")
+        state = persisted.context["task_protocol"]
+        self.assertEqual("protocol_error", report.outcome)
+        if isinstance(progress_result, Exception):
+            self.assertEqual(1, state["task_progress_head_revision"])
+        else:
+            self.assertEqual(2, state["task_progress_head_revision"])
+            self.assertEqual(2, len(state["task_progress_revisions"]))
+        self.assertEqual(
+            state["task_progress_head_revision"],
+            state["task_report_progress_revision"],
+        )
+
     async def test_read_projection_hides_other_protocol_state(self):
         set_trace_store(self.store)
         response = await get_task_progress("root", include_history=False)