fix(protocol): 用持久化命令重放封闭候选审核崩溃窗口

为父级 TaskReview 固化原 tool_call_id 与参数哈希，使 adoption 已提交但父协议尚未提交时可在重启后以同一命令幂等补完，并拒绝载荷篡改。

将 REVISE_CHILD 的孩子报告重置移动到父 Review 提交后的幂等投影，避免先清空孩子验收缓存导致恢复死锁；事件和 Goal 投影同样可重复补齐。

孤立审核调用只有在无需确认时才允许自动重放；缺失审批批次会恢复为 waiting_confirmation，executing 或 execution_unknown 批次继续失败关闭。

增加真实 Runner 进程中断恢复、外部采用恰好一次、REVISE_CHILD 投影崩溃、确认绕过和命令篡改测试。

cyber_agent/core/runner.py

@@ -153,6 +153,10 @@ from cyber_agent.core.prompts import (
 logger = logging.getLogger(__name__)
 
 
+class PendingToolApprovalRestored(RuntimeError):
+    """Internal control flow: an orphaned call was restored to user approval."""
+
+
 @dataclass
 class ContextUsage:
     """Context 使用情况"""
@@ -1350,6 +1354,14 @@ class AgentRunner:
             history, sequence, created_messages, head_seq = await self._build_history(
                 trace.trace_id, messages, goal_tree, config, sequence, side_branch_ctx_for_build
             )
+            # History recovery may replay an idempotent protocol command and
+            # commit a newer context before the model loop starts.  Keep the
+            # yielded Trace object aligned with that authoritative Store state.
+            if self.trace_store:
+                recovered_trace = await self.trace_store.get_trace(trace.trace_id)
+                if recovered_trace is not None:
+                    trace.context = recovered_trace.context
+                    trace.last_sequence = recovered_trace.last_sequence
             # Update trace's head_sequence in memory
             trace.head_sequence = head_seq
             for msg in created_messages:
@@ -1363,6 +1375,12 @@ class AgentRunner:
             ):
                 yield event
 
+        except PendingToolApprovalRestored:
+            if trace and self.trace_store:
+                waiting_trace = await self.trace_store.get_trace(trace.trace_id)
+                if waiting_trace is not None:
+                    yield waiting_trace
+            return
         except asyncio.CancelledError:
             if trace and config.approval_batch_id:
                 await asyncio.shield(
@@ -2188,7 +2206,11 @@ class AgentRunner:
                 # they are executed before the next model turn.
                 if not config.approval_batch_id:
                     main_path, sequence = await self._heal_orphaned_tool_calls(
-                        main_path, trace_id, goal_tree, sequence,
+                        main_path,
+                        trace_obj,
+                        goal_tree,
+                        config,
+                        sequence,
                     )
 
                 history = [msg.to_llm_dict() for msg in main_path]
@@ -2683,6 +2705,135 @@ class AgentRunner:
 
         return None
 
+    def _create_tool_approval_batch(
+        self,
+        *,
+        trace_id: str,
+        assistant_msg: Message,
+        tool_calls: List[Dict[str, Any]],
+        auto_execute_tools: bool,
+    ) -> ToolApprovalBatchV1:
+        """Compile the one persisted approval representation used by all paths."""
+        approval_calls: List[ToolApprovalCallV1] = []
+        for tool_call in tool_calls:
+            tool_name = tool_call.get("function", {}).get("name", "")
+            raw_arguments = tool_call.get("function", {}).get("arguments", {})
+            if isinstance(raw_arguments, str):
+                try:
+                    parsed_arguments = (
+                        json.loads(raw_arguments)
+                        if raw_arguments.strip()
+                        else {}
+                    )
+                except json.JSONDecodeError:
+                    parsed_arguments = {"_raw": raw_arguments}
+            elif isinstance(raw_arguments, dict):
+                parsed_arguments = dict(raw_arguments)
+            else:
+                parsed_arguments = {}
+            policy = self.tools.get_runtime_policy(tool_name)
+            requires_confirmation = (
+                not auto_execute_tools
+                or bool(policy.get("requires_confirmation"))
+            )
+            decision = "pending" if requires_confirmation else "auto_approved"
+            call_id = str(tool_call.get("id") or "")
+            approval_calls.append(ToolApprovalCallV1(
+                tool_call_id=call_id,
+                tool_name=tool_name,
+                original_arguments=parsed_arguments,
+                effective_arguments=parsed_arguments,
+                argument_hash=tool_argument_hash(
+                    tool_call_id=call_id,
+                    tool_name=tool_name,
+                    arguments=parsed_arguments,
+                ),
+                editable_params=list(policy.get("editable_params", [])),
+                requires_confirmation=requires_confirmation,
+                decision=decision,
+            ))
+        return ToolApprovalBatchV1.create(
+            trace_id=trace_id,
+            assistant_message_id=assistant_msg.message_id,
+            assistant_sequence=assistant_msg.sequence,
+            calls=approval_calls,
+        )
+
+    async def _restore_orphaned_tool_approval(
+        self,
+        *,
+        trace: Trace,
+        assistant_msg: Message,
+        tool_calls: List[Dict[str, Any]],
+        config: RunConfig,
+    ) -> None:
+        """Restore the approval gate without executing an orphaned side effect."""
+        if not self.trace_store or not hasattr(
+            self.trace_store,
+            "get_tool_approval_batch",
+        ):
+            raise RuntimeError("TraceStore does not support tool approvals")
+        expected = self._create_tool_approval_batch(
+            trace_id=trace.trace_id,
+            assistant_msg=assistant_msg,
+            tool_calls=tool_calls,
+            auto_execute_tools=config.auto_execute_tools,
+        )
+        batch = await self.trace_store.get_tool_approval_batch(trace.trace_id)
+        if batch is None:
+            batch = expected
+            await self.trace_store.replace_tool_approval_batch(
+                trace.trace_id,
+                batch,
+            )
+        else:
+            same_source = (
+                batch.trace_id == expected.trace_id
+                and batch.assistant_message_id == expected.assistant_message_id
+                and batch.assistant_sequence == expected.assistant_sequence
+                and len(batch.calls) == len(expected.calls)
+                and all(
+                    current.tool_call_id == wanted.tool_call_id
+                    and current.tool_name == wanted.tool_name
+                    and current.original_arguments == wanted.original_arguments
+                    and current.argument_hash == wanted.argument_hash
+                    and current.editable_params == wanted.editable_params
+                    and current.requires_confirmation == wanted.requires_confirmation
+                    for current, wanted in zip(batch.calls, expected.calls)
+                )
+            )
+            if not same_source:
+                raise RuntimeError(
+                    "Persisted tool approval batch does not match the orphaned call"
+                )
+            if batch.status == "executing" or any(
+                call.execution_status == "executing" for call in batch.calls
+            ):
+                batch.status = "execution_unknown"
+                for call in batch.calls:
+                    if call.execution_status == "executing":
+                        call.execution_status = "execution_unknown"
+                batch.updated_at = datetime.now().isoformat()
+                await self.trace_store.replace_tool_approval_batch(
+                    trace.trace_id,
+                    batch,
+                )
+                raise RuntimeError(
+                    "Tool execution outcome is unknown; refusing automatic retry"
+                )
+            if batch.status in {"execution_unknown", "completed", "cancelled"}:
+                raise RuntimeError(
+                    f"Orphaned tool approval is not recoverable: {batch.status}"
+                )
+        await self.trace_store.update_trace(
+            trace.trace_id,
+            status="waiting_confirmation",
+            head_sequence=assistant_msg.sequence,
+        )
+        raise PendingToolApprovalRestored(
+            "orphaned tool call restored to persistent approval"
+        )
+
     async def _resume_approved_tool_batch(
         self,
         *,
@@ -3640,53 +3791,11 @@ class AgentRunner:
                     or self.tools.check_confirmation_required(tool_calls)
                 )
                 if needs_confirmation:
-                    approval_calls: List[ToolApprovalCallV1] = []
-                    for tool_call in tool_calls:
-                        tool_name = tool_call.get("function", {}).get("name", "")
-                        raw_arguments = tool_call.get("function", {}).get(
-                            "arguments", {}
-                        )
-                        if isinstance(raw_arguments, str):
-                            try:
-                                parsed_arguments = (
-                                    json.loads(raw_arguments)
-                                    if raw_arguments.strip()
-                                    else {}
-                                )
-                            except json.JSONDecodeError:
-                                parsed_arguments = {"_raw": raw_arguments}
-                        elif isinstance(raw_arguments, dict):
-                            parsed_arguments = dict(raw_arguments)
-                        else:
-                            parsed_arguments = {}
-                        policy = self.tools.get_runtime_policy(tool_name)
-                        requires_confirmation = (
-                            not config.auto_execute_tools
-                            or bool(policy.get("requires_confirmation"))
-                        )
-                        decision = (
-                            "pending" if requires_confirmation else "auto_approved"
-                        )
-                        call_id = str(tool_call.get("id") or "")
-                        approval_calls.append(ToolApprovalCallV1(
-                            tool_call_id=call_id,
-                            tool_name=tool_name,
-                            original_arguments=parsed_arguments,
-                            effective_arguments=parsed_arguments,
-                            argument_hash=tool_argument_hash(
-                                tool_call_id=call_id,
-                                tool_name=tool_name,
-                                arguments=parsed_arguments,
-                            ),
-                            editable_params=list(policy.get("editable_params", [])),
-                            requires_confirmation=requires_confirmation,
-                            decision=decision,
-                        ))
-                    batch = ToolApprovalBatchV1.create(
+                    batch = self._create_tool_approval_batch(
                         trace_id=trace_id,
-                        assistant_message_id=assistant_msg.message_id,
-                        assistant_sequence=assistant_msg.sequence,
-                        calls=approval_calls,
+                        assistant_msg=assistant_msg,
+                        tool_calls=tool_calls,
+                        auto_execute_tools=config.auto_execute_tools,
                     )
                     if not self.trace_store or not hasattr(
                         self.trace_store, "replace_tool_approval_batch"
@@ -4700,11 +4809,98 @@ class AgentRunner:
 
         return cutoff
 
+    async def _replay_orphaned_review_call(
+        self,
+        *,
+        trace: Trace,
+        goal_tree: Optional[GoalTree],
+        config: RunConfig,
+        assistant_msg: Message,
+        tool_call: Dict[str, Any],
+        sequence: int,
+        head_sequence: int,
+    ) -> Message:
+        """Replay one durable, idempotent review command before another LLM turn."""
+        function = tool_call.get("function") or {}
+        raw_arguments = function.get("arguments", {})
+        if isinstance(raw_arguments, str):
+            try:
+                arguments = json.loads(raw_arguments) if raw_arguments.strip() else {}
+            except json.JSONDecodeError as exc:
+                raise RuntimeError(
+                    "Orphaned review command has invalid persisted arguments"
+                ) from exc
+        elif isinstance(raw_arguments, dict):
+            arguments = dict(raw_arguments)
+        else:
+            raise RuntimeError(
+                "Orphaned review command has invalid persisted arguments"
+            )
+        tool_call_id = str(tool_call.get("id") or "")
+        if not tool_call_id:
+            raise RuntimeError("Orphaned review command has no tool_call_id")
+        fresh_trace = (
+            await self.trace_store.get_trace(trace.trace_id)
+            if self.trace_store
+            else None
+        )
+        if fresh_trace is None:
+            raise RuntimeError("Orphaned review Trace no longer exists")
+        result = await self.tools.execute(
+            "review_task_result",
+            arguments,
+            uid=config.uid or "",
+            context=self._build_tool_context(
+                config=config,
+                trace=fresh_trace,
+                trace_id=fresh_trace.trace_id,
+                goal_id=assistant_msg.goal_id,
+                goal_tree=goal_tree,
+                sequence=sequence,
+                head_sequence=head_sequence,
+                tool_call_id=tool_call_id,
+                side_branch_ctx=None,
+                trigger_event=None,
+            ),
+            allowed_tool_names={"review_task_result"},
+            tool_call_id=tool_call_id,
+        )
+        result_text = (
+            result
+            if isinstance(result, str)
+            else json.dumps(result, ensure_ascii=False)
+        )
+        try:
+            payload = json.loads(result_text)
+        except (json.JSONDecodeError, TypeError) as exc:
+            raise RuntimeError(
+                "Orphaned review replay returned a non-JSON result"
+            ) from exc
+        if not isinstance(payload, dict) or payload.get("status") != "completed":
+            reason = payload.get("error") if isinstance(payload, dict) else None
+            raise RuntimeError(
+                "Orphaned review replay failed closed: "
+                + str(reason or "unexpected tool result")
+            )
+        return Message.create(
+            trace_id=fresh_trace.trace_id,
+            role="tool",
+            sequence=sequence,
+            goal_id=assistant_msg.goal_id,
+            parent_sequence=head_sequence,
+            tool_call_id=tool_call_id,
+            content={
+                "tool_name": "review_task_result",
+                "result": result_text,
+            },
+        )
+
     async def _heal_orphaned_tool_calls(
         self,
         messages: List[Message],
-        trace_id: str,
+        trace: Trace,
         goal_tree: Optional[GoalTree],
+        config: RunConfig,
         sequence: int,
     ) -> tuple:
         """
@@ -4713,7 +4909,8 @@ class AgentRunner:
         当 agent 被 stop/crash 中断时，可能有 assistant 的 tool_calls 没有对应的
         tool results（包括多 tool_call 部分完成的情况）。直接发给 LLM 会导致 400。
 
-        修复策略：为每个缺失的 tool_result 插入合成的"中断通知"消息，而非裁剪。
+        修复策略：幂等的 ``review_task_result`` 使用原 tool_call_id 在 LLM 前
+        受控重放；其他缺失调用插入合成的"中断通知"消息，而非裁剪。
         - 普通工具：简短中断提示
         - agent/evaluate：包含 sub_trace_id、执行统计、continue_from 指引
 
@@ -4756,6 +4953,49 @@ class AgentRunner:
             assistant_msg, tc = tc_map[tc_id]
             tool_name = tc.get("function", {}).get("name", "unknown")
 
+            if tool_name == "review_task_result":
+                assistant_calls = (
+                    assistant_msg.content.get("tool_calls", [])
+                    if isinstance(assistant_msg.content, dict)
+                    else []
+                )
+                if len(assistant_calls) != 1:
+                    raise RuntimeError(
+                        "Orphaned lifecycle-exclusive review was persisted in a mixed batch"
+                    )
+                needs_confirmation = (
+                    not config.auto_execute_tools
+                    or self.tools.check_confirmation_required([tc])
+                )
+                persisted_approval = (
+                    await self.trace_store.get_tool_approval_batch(trace.trace_id)
+                    if self.trace_store
+                    and hasattr(self.trace_store, "get_tool_approval_batch")
+                    else None
+                )
+                if needs_confirmation or persisted_approval is not None:
+                    await self._restore_orphaned_tool_approval(
+                        trace=trace,
+                        assistant_msg=assistant_msg,
+                        tool_calls=[tc],
+                        config=config,
+                    )
+                replayed = await self._replay_orphaned_review_call(
+                    trace=trace,
+                    goal_tree=goal_tree,
+                    config=config,
+                    assistant_msg=assistant_msg,
+                    tool_call=tc,
+                    sequence=sequence,
+                    head_sequence=head_seq,
+                )
+                if self.trace_store:
+                    await self.trace_store.add_message(replayed)
+                healed.append(replayed)
+                head_seq = sequence
+                sequence += 1
+                continue
+
             if tool_name in ("agent", "evaluate"):
                 result_text = self._build_agent_interrupted_result(
                     tc, goal_tree, assistant_msg,
@@ -4764,7 +5004,7 @@ class AgentRunner:
                 result_text = build_tool_interrupted_message(tool_name)
 
             synthetic_msg = Message.create(
-                trace_id=trace_id,
+                trace_id=trace.trace_id,
                 role="tool",
                 sequence=sequence,
                 goal_id=assistant_msg.goal_id,
@@ -4783,7 +5023,7 @@ class AgentRunner:
         # 更新 trace head/last sequence
         if self.trace_store:
             await self.trace_store.update_trace(
-                trace_id,
+                trace.trace_id,
                 head_sequence=head_seq,
                 last_sequence=max(head_seq, sequence - 1),
             )

cyber_agent/tools/builtin/task_protocol.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 from copy import deepcopy
 from datetime import datetime
+from hashlib import sha256
 from typing import Any
 
 from pydantic import ValidationError
@@ -39,6 +40,7 @@ from cyber_agent.core.task_protocol_service import (
 )
 from cyber_agent.core.validation import ValidationResult
 from cyber_agent.application.candidate import CandidateReviewAction
+from cyber_agent.application.models import canonical_json
 from cyber_agent.tools import tool
 
 
@@ -61,6 +63,150 @@ def _error(message: str) -> dict[str, Any]:
     return {"status": "failed", "error": message}
 
 
+def _review_command_hash(
+    *,
+    child_trace_id: str,
+    decision: ReviewDecision,
+    reason: str,
+    approved_next_task: TaskBrief | str | None,
+    candidate_actions: list[CandidateReviewAction] | None,
+) -> str:
+    """Hash the persisted assistant command before protocol normalization."""
+
+    def json_value(value: Any) -> Any:
+        if hasattr(value, "model_dump"):
+            return value.model_dump(mode="json")
+        if isinstance(value, list):
+            return [json_value(item) for item in value]
+        if isinstance(value, tuple):
+            return [json_value(item) for item in value]
+        if isinstance(value, dict):
+            return {key: json_value(item) for key, item in value.items()}
+        return value
+
+    payload = {
+        "child_trace_id": child_trace_id,
+        "decision": decision,
+        "reason": reason,
+        "approved_next_task": json_value(approved_next_task),
+        "candidate_actions": json_value(candidate_actions or []),
+    }
+    return sha256(canonical_json(payload).encode("utf-8")).hexdigest()
+
+
+def _completed_review_response(
+    review_record: dict[str, Any],
+    state: dict[str, Any],
+) -> dict[str, Any]:
+    task_review = TaskReview.model_validate({
+        field: review_record.get(field)
+        for field in TaskReview.model_fields
+    })
+    return {
+        "status": "completed",
+        "task_review": task_review.model_dump(),
+        "context_ref": review_record.get("context_ref"),
+        "context_ref_warning": review_record.get("context_ref_warning"),
+        "next_action_task_brief": review_record.get(
+            "next_action_task_brief"
+        ),
+        "pending_review_count": len(state["pending_reviews"]),
+        "next_action_count": len(state["next_actions"]),
+    }
+
+
+async def _project_committed_review(
+    *,
+    store: Any,
+    context: dict[str, Any],
+    parent_trace_id: str,
+    child_trace_id: str,
+    state: dict[str, Any],
+    review_record: dict[str, Any],
+) -> None:
+    """Idempotently finish projections after the parent protocol commit."""
+    reviewed_at_sequence = int(review_record["reviewed_at_sequence"])
+    if review_record["decision"] == "REVISE_CHILD":
+        child = await store.get_trace(child_trace_id)
+        if child is None:
+            raise ValueError(
+                "Committed REVISE_CHILD review references a missing child Trace"
+            )
+        child_state = ensure_task_protocol(child.context)
+        old_report = child_state.get("task_report")
+        if old_report and old_report not in child_state["report_history"]:
+            child_state["report_history"].append(old_report)
+        child_state["task_report"] = None
+        child_state["task_report_submitted_at_sequence"] = None
+        child_state["task_report_progress_revision"] = None
+        child_state["task_report_validation"] = None
+        child_state["protocol_correction_attempts"] = 0
+        replace_context_access(
+            child.context,
+            [],
+            root_task_anchor=require_root_task_anchor(child.context),
+            task_brief=child_state.get("task_brief"),
+        )
+        await store.update_trace(child_trace_id, context=child.context)
+
+    event_service = context.get("event_service")
+    if event_service is not None:
+        await event_service.emit_after_commit(
+            source_trace_id=parent_trace_id,
+            event_type="task.review.recorded",
+            event_key=(
+                f"task.review.recorded:{parent_trace_id}:"
+                f"{child_trace_id}:{reviewed_at_sequence}"
+            ),
+            effective_at_sequence=reviewed_at_sequence,
+            payload={"task_review": review_record},
+        )
+
+    pending_entry = review_record.get("pending_review") or {}
+    goal_id = pending_entry.get("goal_id")
+    if not goal_id:
+        return
+    decision = review_record["decision"]
+    resolved_replan = review_record.get("resolved_replan")
+    if state["pending_reviews"]:
+        next_status = "pending_review"
+    elif resolved_replan:
+        next_status = "in_progress"
+        goal_id = resolved_replan["goal_id"]
+    elif decision == "ASCEND":
+        next_status = "completed"
+    elif decision == "FAIL":
+        next_status = "failed"
+    else:
+        next_status = "in_progress"
+    reason = review_record["reason"]
+    await store.update_goal(
+        parent_trace_id,
+        goal_id,
+        cascade_completion=False,
+        status=next_status,
+        summary=reason if next_status in {"completed", "failed"} else None,
+    )
+    tree = context.get("goal_tree")
+    goal = tree.find(goal_id) if tree else None
+    if goal:
+        goal.status = next_status
+        if next_status in {"completed", "failed"}:
+            goal.summary = reason
+            if tree.current_id == goal_id:
+                tree.current_id = goal.parent_id
+        elif resolved_replan:
+            tree.current_id = goal_id
+
+    if resolved_replan:
+        persisted_tree = await store.get_goal_tree(parent_trace_id)
+        persisted_goal = persisted_tree.find(goal_id) if persisted_tree else None
+        if persisted_goal:
+            persisted_goal.status = "in_progress"
+            persisted_tree.current_id = goal_id
+            await store.update_goal_tree(parent_trace_id, persisted_tree)
+
+
 @tool(
     description=(
         "Replace the current Recursive task's structured progress with one "
@@ -279,6 +425,15 @@ async def _review_task_result_locked(
     if not store or not parent_trace_id:
         return _error("store and trace_id are required")
 
+    command_id = str(context.get("tool_call_id") or "") or None
+    command_hash = _review_command_hash(
+        child_trace_id=child_trace_id,
+        decision=decision,
+        reason=reason,
+        approved_next_task=approved_next_task,
+        candidate_actions=candidate_actions,
+    )
+
     parent = await store.get_trace(parent_trace_id)
     child = await store.get_trace(child_trace_id)
     if not parent or not child:
@@ -293,6 +448,26 @@ async def _review_task_result_locked(
         return _error("TaskReport must belong to a direct child with the same owner")
 
     state = ensure_task_protocol(parent.context)
+    committed = [
+        item for item in state["reviews"]
+        if command_id is not None
+        and item.get("review_command_id") == command_id
+    ]
+    if len(committed) > 1:
+        return _error("Persisted TaskReview command identity is duplicated")
+    if committed:
+        review_record = committed[0]
+        if review_record.get("review_command_hash") != command_hash:
+            return _error("TaskReview command payload changed during recovery")
+        await _project_committed_review(
+            store=store,
+            context=context,
+            parent_trace_id=parent_trace_id,
+            child_trace_id=child_trace_id,
+            state=state,
+            review_record=review_record,
+        )
+        return _completed_review_response(review_record, state)
     pending = state["pending_reviews"]
     entry = pending.get(child_trace_id)
     if not entry:
@@ -509,33 +684,22 @@ async def _review_task_result_locked(
     pending.pop(child_trace_id)
     review_record = {
         **review.model_dump(),
+        "review_command_id": command_id,
+        "review_command_hash": command_hash,
         "pending_review": deepcopy(entry),
         "reviewed_at": datetime.now().isoformat(),
         "reviewed_at_sequence": reviewed_at_sequence,
         "context_ref": context_ref.model_dump() if context_ref else None,
         "context_ref_warning": context_ref_warning,
+        "next_action_task_brief": (
+            action_task_brief.model_dump(mode="json")
+            if action_task_brief is not None
+            else None
+        ),
     }
     state["reviews"].append(review_record)
     state["protocol_correction_attempts"] = 0
 
-    if review.decision == "REVISE_CHILD":
-        assert child_state is not None and revision_task_brief is not None
-        old_report = child_state.get("task_report")
-        if old_report:
-            child_state["report_history"].append(old_report)
-        child_state["task_report"] = None
-        child_state["task_report_submitted_at_sequence"] = None
-        child_state["task_report_progress_revision"] = None
-        child_state["task_report_validation"] = None
-        child_state["protocol_correction_attempts"] = 0
-        replace_context_access(
-            child.context,
-            [],
-            root_task_anchor=require_root_task_anchor(child.context),
-            task_brief=child_state.get("task_brief"),
-        )
-        await store.update_trace(child_trace_id, context=child.context)
-
     if review.decision in {"ACCEPT_REFINE", "DESCEND_AGAIN", "REVISE_CHILD"}:
         state["next_actions"].append({
             "decision": review.decision,
@@ -559,70 +723,15 @@ async def _review_task_result_locked(
     if not pending and state["pending_replans"]:
         resolved_replan = state["pending_replans"][0]
         state["pending_replans"].clear()
+    review_record["resolved_replan"] = deepcopy(resolved_replan)
 
     await store.update_trace(parent_trace_id, context=parent.context)
-    event_service = context.get("event_service")
-    if event_service is not None:
-        await event_service.emit_after_commit(
-            source_trace_id=parent_trace_id,
-            event_type="task.review.recorded",
-            event_key=(
-                f"task.review.recorded:{parent_trace_id}:"
-                f"{child_trace_id}:{reviewed_at_sequence}"
-            ),
-            effective_at_sequence=reviewed_at_sequence,
-            payload={"task_review": review_record},
-        )
-
-    goal_id = origin_goal_id
-    if goal_id:
-        if pending:
-            next_status = "pending_review"
-        elif resolved_replan:
-            next_status = "in_progress"
-            goal_id = resolved_replan["goal_id"]
-        elif review.decision == "ASCEND":
-            next_status = "completed"
-        elif review.decision == "FAIL":
-            next_status = "failed"
-        else:
-            next_status = "in_progress"
-        await store.update_goal(
-            parent_trace_id,
-            goal_id,
-            cascade_completion=False,
-            status=next_status,
-            summary=reason if next_status in {"completed", "failed"} else None,
-        )
-        tree = context.get("goal_tree")
-        goal = tree.find(goal_id) if tree else None
-        if goal:
-            goal.status = next_status
-            if next_status in {"completed", "failed"}:
-                goal.summary = reason
-                if tree.current_id == goal_id:
-                    tree.current_id = goal.parent_id
-            elif resolved_replan:
-                tree.current_id = goal_id
-
-        if resolved_replan:
-            persisted_tree = await store.get_goal_tree(parent_trace_id)
-            persisted_goal = persisted_tree.find(goal_id) if persisted_tree else None
-            if persisted_goal:
-                persisted_goal.status = "in_progress"
-                persisted_tree.current_id = goal_id
-                await store.update_goal_tree(parent_trace_id, persisted_tree)
-
-    return {
-        "status": "completed",
-        "task_review": review.model_dump(),
-        "context_ref": context_ref.model_dump() if context_ref else None,
-        "context_ref_warning": context_ref_warning,
-        "next_action_task_brief": (
-            action_task_brief.model_dump(mode="json")
-            if action_task_brief is not None
-            else None
-        ),
-        "pending_review_count": len(pending),
-        "next_action_count": len(state["next_actions"]),
-    }
+    await _project_committed_review(
+        store=store,
+        context=context,
+        parent_trace_id=parent_trace_id,
+        child_trace_id=child_trace_id,
+        state=state,
+        review_record=review_record,
+    )
+    return _completed_review_response(review_record, state)

tests/test_application_reference.py

@@ -1,4 +1,5 @@
 from copy import deepcopy
+import asyncio
 import json
 import tempfile
 import unittest
@@ -64,7 +65,7 @@ def _tool_names(schemas):
 
 
 class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
-    async def test_real_agent_candidate_report_review_and_adoption_chain(self):
+    async def test_real_agent_recovers_orphaned_adoption_review_after_restart(self):
         with tempfile.TemporaryDirectory() as temp_dir:
             store = FileSystemTraceStore(temp_dir)
             components = build_reference_components()
@@ -234,11 +235,83 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                 enable_completion_extraction=False,
                 enable_injection=False,
             )
-            with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
-                result = await runner.run_result(
-                    [{"role": "user", "content": "Create the output"}],
-                    config,
+            original_update_trace = store.update_trace
+            interrupted = False
+
+            async def interrupt_after_adoption(trace_id, **updates):
+                nonlocal interrupted
+                context = updates.get("context")
+                reviews = (
+                    context.get("task_protocol", {}).get("reviews", [])
+                    if isinstance(context, dict)
+                    else []
                 )
+                if (
+                    not interrupted
+                    and any(
+                        item.get("review_command_id") == "adopt-review"
+                        for item in reviews
+                    )
+                ):
+                    interrupted = True
+                    raise asyncio.CancelledError()
+                return await original_update_trace(trace_id, **updates)
+
+            with patch.object(
+                store,
+                "update_trace",
+                side_effect=interrupt_after_adoption,
+            ), patch.dict(
+                "os.environ",
+                {"AGENT_MODE": "recursive"},
+                clear=False,
+            ):
+                with self.assertRaises(asyncio.CancelledError):
+                    await runner.run_result(
+                        [{"role": "user", "content": "Create the output"}],
+                        config,
+                    )
+            self.assertTrue(interrupted)
+            roots = [
+                item for item in await store.list_traces(limit=20)
+                if item.parent_trace_id is None
+            ]
+            self.assertEqual(1, len(roots))
+            root = roots[0]
+            self.assertEqual(1, components.candidates.adoption_calls)
+            self.assertTrue(
+                root.context["task_protocol"]["pending_reviews"]
+            )
+            orphaned = [
+                item for item in await store.get_trace_messages(root.trace_id)
+                if item.role == "assistant"
+                and isinstance(item.content, dict)
+                and any(
+                    call.get("id") == "adopt-review"
+                    for call in item.content.get("tool_calls", [])
+                )
+            ]
+            self.assertEqual(1, len(orphaned))
+            self.assertFalse(any(
+                item.role == "tool" and item.tool_call_id == "adopt-review"
+                for item in await store.get_trace_messages(root.trace_id)
+            ))
+
+            restarted_registry = ApplicationRegistry()
+            restarted_registry.register(
+                components.application,
+                components.services,
+            )
+            restarted_runtime = ApplicationRuntime(
+                registry=restarted_registry,
+                trace_store=store,
+                llm_call=llm_call,
+            )
+            restored_runner, restored_config = await restarted_runtime.restore(
+                root.trace_id
+            )
+            with patch.dict("os.environ", {"AGENT_MODE": "recursive"}, clear=False):
+                result = await restored_runner.run_result([], restored_config)
             self.assertEqual("completed", result["status"])
             self.assertTrue(reviewed)
             self.assertEqual(1, len(components.candidates.versions))
@@ -256,6 +329,14 @@ class ApplicationReferenceEndToEndTest(unittest.IsolatedAsyncioTestCase):
                 "adopt-review",
                 {item.command_id for item in ledger.operations},
             )
+            self.assertEqual(1, len([
+                item for item in await store.get_trace_messages(root.trace_id)
+                if item.role == "tool" and item.tool_call_id == "adopt-review"
+            ]))
+            self.assertEqual(
+                1,
+                len(root.context["task_protocol"]["reviews"]),
+            )
 
     async def test_local_candidate_retry_preserves_fact_and_peer_validation(self):
         with tempfile.TemporaryDirectory() as temp_dir:

tests/test_runtime_hardening.py

@@ -638,6 +638,83 @@ class ToolApprovalStateMachineTest(unittest.IsolatedAsyncioTestCase):
         self.assertEqual("pending", persisted.status)
         self.assertEqual([], self.side_effects)
 
+    async def test_orphaned_review_restores_confirmation_and_never_replays_unknown(self):
+        registry = ToolRegistry()
+        review_side_effects = []
+
+        async def review_task_result(child_trace_id: str):
+            review_side_effects.append(child_trace_id)
+            return {"status": "completed"}
+
+        registry.register(review_task_result)
+        runner = AgentRunner(
+            trace_store=self.store,
+            tool_registry=registry,
+            llm_call=self.runner.llm_call,
+        )
+        trace = Trace(
+            trace_id="orphaned-review-approval",
+            mode="agent",
+            uid="user-1",
+            context={},
+        )
+        await self.store.create_trace(trace)
+        assistant = Message.create(
+            trace_id=trace.trace_id,
+            role="assistant",
+            sequence=1,
+            content={
+                "text": "",
+                "tool_calls": [{
+                    "id": "orphaned-review-call",
+                    "type": "function",
+                    "function": {
+                        "name": "review_task_result",
+                        "arguments": json.dumps({"child_trace_id": "child-1"}),
+                    },
+                }],
+            },
+        )
+        await self.store.add_message(assistant)
+        await self.store.update_trace(
+            trace.trace_id,
+            head_sequence=1,
+            last_sequence=1,
+        )
+        config = RunConfig(auto_execute_tools=False)
+
+        with self.assertRaisesRegex(RuntimeError, "restored to persistent approval"):
+            await runner._heal_orphaned_tool_calls(
+                [assistant],
+                trace,
+                None,
+                config,
+                2,
+            )
+        batch = await self.store.get_tool_approval_batch(trace.trace_id)
+        self.assertEqual("pending", batch.status)
+        self.assertEqual("waiting_confirmation", (
+            await self.store.get_trace(trace.trace_id)
+        ).status)
+        self.assertEqual([], review_side_effects)
+
+        batch.status = "executing"
+        batch.calls[0].decision = "approved"
+        batch.calls[0].execution_status = "executing"
+        await self.store.replace_tool_approval_batch(trace.trace_id, batch)
+        await self.store.update_trace(trace.trace_id, status="running")
+        with self.assertRaisesRegex(RuntimeError, "outcome is unknown"):
+            await runner._heal_orphaned_tool_calls(
+                [assistant],
+                await self.store.get_trace(trace.trace_id),
+                None,
+                config,
+                2,
+            )
+        unknown = await self.store.get_tool_approval_batch(trace.trace_id)
+        self.assertEqual("execution_unknown", unknown.status)
+        self.assertEqual([], review_side_effects)
+
     async def test_restart_resumes_decided_batch_before_any_side_effect(self):
         trace_id, batch = await self.start_waiting_batch()
         # Crash window: decisions are durable, but no call has yet been marked

tests/test_task_protocol.py

@@ -1,3 +1,4 @@
+import asyncio
 import json
 import tempfile
 import unittest
@@ -569,6 +570,106 @@ class RecursiveTaskProtocolTest(unittest.IsolatedAsyncioTestCase):
         )
         self.assertEqual("completed", reviewed["status"])
 
+    async def test_persisted_review_command_replays_idempotently_and_rejects_tampering(self):
+        result = await agent(task_brief=BRIEF, context=self.context())
+        review_context = {
+            **self.context(),
+            "tool_call_id": "durable-review-command",
+        }
+        first = await review_task_result(
+            child_trace_id=result["sub_trace_id"],
+            decision="ASCEND",
+            reason="The exact persisted report is accepted",
+            context=review_context,
+        )
+        self.assertEqual("completed", first["status"])
+
+        replayed = await review_task_result(
+            child_trace_id=result["sub_trace_id"],
+            decision="ASCEND",
+            reason="The exact persisted report is accepted",
+            context={**review_context, "sequence": 999},
+        )
+        self.assertEqual(first["task_review"], replayed["task_review"])
+        parent = await self.store.get_trace(self.root_id)
+        state = ensure_task_protocol(parent.context)
+        self.assertEqual(1, len(state["reviews"]))
+        self.assertEqual({}, state["pending_reviews"])
+
+        tampered = await review_task_result(
+            child_trace_id=result["sub_trace_id"],
+            decision="ASCEND",
+            reason="Changed after the durable command was committed",
+            context={**review_context, "sequence": 1_000},
+        )
+        self.assertEqual("failed", tampered["status"])
+        self.assertIn("payload changed", tampered["error"])
+
+    async def test_revise_child_projection_recovers_after_parent_commit_crash(self):
+        result = await agent(task_brief=BRIEF, context=self.context())
+        child_id = result["sub_trace_id"]
+        review_context = {
+            **self.context(),
+            "tool_call_id": "durable-revise-child",
+        }
+        original_update_trace = self.store.update_trace
+        interrupted = False
+
+        async def interrupt_child_projection(trace_id, **updates):
+            nonlocal interrupted
+            context = updates.get("context")
+            child_state = (
+                context.get("task_protocol", {})
+                if isinstance(context, dict)
+                else {}
+            )
+            if (
+                not interrupted
+                and trace_id == child_id
+                and child_state.get("task_report") is None
+            ):
+                interrupted = True
+                raise asyncio.CancelledError()
+            return await original_update_trace(trace_id, **updates)
+
+        with patch.object(
+            self.store,
+            "update_trace",
+            side_effect=interrupt_child_projection,
+        ):
+            with self.assertRaises(asyncio.CancelledError):
+                await review_task_result(
+                    child_trace_id=child_id,
+                    decision="REVISE_CHILD",
+                    reason="Revise only the child's current output",
+                    context=review_context,
+                )
+        self.assertTrue(interrupted)
+        parent = await self.store.get_trace(self.root_id)
+        parent_state = ensure_task_protocol(parent.context)
+        self.assertEqual(1, len(parent_state["reviews"]))
+        self.assertEqual(1, len(parent_state["next_actions"]))
+        self.assertEqual({}, parent_state["pending_reviews"])
+        child = await self.store.get_trace(child_id)
+        self.assertIsNotNone(ensure_task_protocol(child.context)["task_report"])
+
+        replayed = await review_task_result(
+            child_trace_id=child_id,
+            decision="REVISE_CHILD",
+            reason="Revise only the child's current output",
+            context={**review_context, "sequence": 999},
+        )
+        self.assertEqual("completed", replayed["status"])
+        parent = await self.store.get_trace(self.root_id)
+        parent_state = ensure_task_protocol(parent.context)
+        self.assertEqual(1, len(parent_state["reviews"]))
+        self.assertEqual(1, len(parent_state["next_actions"]))
+        child = await self.store.get_trace(child_id)
+        child_state = ensure_task_protocol(child.context)
+        self.assertIsNone(child_state["task_report"])
+        self.assertIsNone(child_state["task_report_validation"])
+        self.assertEqual(1, len(child_state["report_history"]))
+
     async def test_accept_refine_approves_exactly_one_next_child(self):
         first = await agent(task_brief=BRIEF, context=self.context())
         next_brief = {