fix(chat): resume unsubmitted native handoffs with one bounded format review

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1827,3 +1827,23 @@ submits none of its actions. Recovery itself does not execute even valid action
 metadata. Such work remains pending after the recovered prose is delivered;
 answer restoration does not certify the requested business or handoff complete.
 Cancellation failure retains the primary provider error and unconfirmed cleanup.
+
+After the recovered answer is independently delivered, a malformed batch whose
+only actions are string-brief handoffs can reserve one **format review** through
+the original Session queue. Fresh original native input/answer hashes, current
+binding, source readback and an idle queue are required. The review sends no
+business request again: that invocation disables native tools and user hooks,
+requires exact normal text-only completion, and withholds model prose until
+validation. Each normalized brief must retain its original context text and the
+same ordered recipients. Other actions, changed targets, unknown completion,
+new input, steering, stop or revoked authority admit no handoff.
+
+The existing ingress preserves the original source id, so ordinary publication
+uses the same recipient-specific request identity and fresh grants. The normal
+handoff owner supplies actual receipts; receiver conclusions remain separate.
+The original reply worker returns those receipts in a distinct presentation
+phase alongside the immutable failed Turn, failure notice and recovered answer.
+Reservation/enqueue crashes adopt the same queued Turn. A failed, partial or
+unknown review stays pending and cannot become another review or business retry.
+Other malformed actions and completed Turns without a recovered native result
+remain explicit holds; this path is not a generic action or execution replayer.

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -1198,3 +1198,17 @@ steering、stop/new 与 attached-host 路由。provider 读回必须完整覆盖
 不能盲目重发。操作元数据无效时保留正文、明确显示 gate，不提交其中任何动作；
 结果恢复本身也不执行有效的操作元数据。即使正文已补回，这些工作仍保持待办；
 正文恢复不证明业务或委派完成。取消失败同时保留原 provider 错误和清理未确认状态。
+
+恢复正文并独立验明已送达后，若原动作批次只有字符串 brief 的委派，原 Session
+队列可以预约一次**格式复核**。准入重新核验原生输入/答复 hash、当前 binding、
+原消息读回和空闲队列。这次调用不重做业务：禁用原生工具和用户 hooks，要求准确
+的正常纯文本结束，并在校验前隐藏模型正文。规范化后的 brief 必须保留原 context
+文本、接收者及顺序；附加动作、目标变化、未知结束、新输入、steering、停止或
+授权撤销均不能提交委派。
+
+既有 ingress 保持原 source id，普通投递因此沿用每个接收者的稳定请求身份和
+当前授权。正常 handoff owner 提供实际回执，接收者结论仍须单独回传。原回复
+worker 使用独立展示阶段返回这些回执，保留原失败 Turn、失败通知和恢复正文。
+预约或入队后的崩溃只采用同一个队列 Turn；失败、部分提交或结果未知的复核继续
+记为欠办，不能变成又一次复核或业务重试。其他错误动作、没有已恢复原生结果的
+completed Turn 继续明确保留，本路径不提供通用动作或执行重放。

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -21,6 +21,7 @@
 from datetime import datetime
 
 from ...chat import (
+    CHAT_REVIEW_OPEN_TAG, CHAT_REVIEW_CLOSE_TAG,
     VisibleResponseStreamFilter,
     parse_agent_response,
 )
@@ -306,7 +307,8 @@ class AntigravityAdapter:
     @staticmethod
     def completed_result_observation(*, cascade_id: str, request_ref: str, message: str,
                                      started_at: str, completed_at: str,
-                                     work_dir: Path, input_anchor: dict[str, Any] | None = None) -> dict[str, Any] | None:
+                                     work_dir: Path, input_anchor: dict[str, Any] | None = None,
+                                     include_raw_response: bool = False) -> dict[str, Any] | None:
         """Read an existing result; never start, send to or cancel a cascade.
 
         Legacy owner requests require the exact Core-generated source context,
@@ -380,7 +382,8 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
             return {"thread_id": cascade_id, "request_ref": request_ref, "execution_id": execution,
                 "input_created_at": metadata["createdAt"], "start_step_index": start, "total_steps": total,
                 "input_sha256": prompt_hash, "answer_sha256": hashlib.sha256(answer.encode()).hexdigest(),
-                "latest_input_verified": True, "normal_terminal_verified": True, "response": response}
+                "latest_input_verified": True, "normal_terminal_verified": True, "response": response,
+                **({"raw_response": answer} if include_raw_response else {})}
         except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
             return None
 
@@ -412,6 +415,7 @@ def __init__(
         self.model_placeholder = model_placeholder
         self.reasoning_effort = reasoning_effort
         self.runtime_profile = runtime_profile
+        self._read_only_review = False
         self.team_plan_context = team_plan_context
         self._pending_history: list[dict[str, str]] = (
             list(initial_history) if initial_history and not resumed else []
@@ -559,15 +563,15 @@ def _build_turn_prompt(self, message: str) -> str:
         parts.append(
             _turn_prompt(
                 message,
-                runtime_profile=self.runtime_profile,
+                runtime_profile="restricted" if self._read_only_review else self.runtime_profile,
             )
         )
         return "\n\n".join(parts)
 
     def _build_cascade_config(self) -> dict[str, Any]:
         # Never inherit an old cascade's tool permissions or userAgentSpec.
         # Restricted Chat is text-only; trusted_owner retains native host approval.
-        restricted = self.runtime_profile == "restricted"
+        restricted = self.runtime_profile == "restricted" or self._read_only_review
         return {"plannerConfig": {
             "planModel": self.model_placeholder,
             "requestedModel": {"model": self.model_placeholder},
@@ -586,7 +590,9 @@ def capabilities(self) -> dict[str, Any]:
                 "resume": True, "interrupt": True, "steering": False}
 
     def start_turn(self, message: str,
-                   event_sink: Callable[[str, dict[str, Any]], None] | None = None) -> dict[str, Any]:
+                   event_sink: Callable[[str, dict[str, Any]], None] | None = None,
+                   *, read_only_review: bool = False) -> dict[str, Any]:
+        self._read_only_review = read_only_review
         self._interrupted.clear()
         self._cancel_acknowledged = False
         try:
@@ -604,6 +610,7 @@ def start_turn(self, message: str,
             raise
         finally:
             self._active_turn_id = None
+            self._read_only_review = False
 
     def _run_turn(
         self,
@@ -712,6 +719,11 @@ def _run_turn(
                     for step_idx, step in current_steps:
                         step_type = str(step.get("type") or "")
                         step_status = str(step.get("status") or "")
+                        if self._read_only_review and step_type not in {
+                                "CORTEX_STEP_TYPE_USER_INPUT", "CORTEX_STEP_TYPE_PLANNER_RESPONSE",
+                                "CORTEX_STEP_TYPE_CHECKPOINT"}:
+                            raise CodexChatAgentError("The format review produced a native tool step; no actions were admitted.",
+                                                      gate=None, error_code="unsupported_configuration")
                         if step_status == "CORTEX_STEP_STATUS_WAITING":
                             raise CodexChatAgentError(
                                 "Antigravity requested native permission; this invocation was cancelled. Review its permissions, then retry.",
@@ -735,7 +747,7 @@ def _run_turn(
                                         },
                                     )
                             tool_calls = pr.get("toolCalls") or []
-                            if tool_calls and self.runtime_profile == "restricted":
+                            if tool_calls and (self.runtime_profile == "restricted" or self._read_only_review):
                                 raise CodexChatAgentError("Antigravity did not honor text-only Chat.",
                                                           gate=None, error_code="unsupported_configuration")
                             if isinstance(tool_calls, list) and tool_calls and (step_idx, "tool") not in announced_phases:
@@ -797,6 +809,15 @@ def _run_turn(
             if saw_running_or_new_step and status == "CASCADE_RUN_STATUS_IDLE" and observed_total == total_steps:
                 final = current_steps[-1][1] if current_steps else {}
                 planner = final.get("plannerResponse") or {}
+                if self._read_only_review:
+                    inputs = [step for _, step in current_steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+                    execution = (inputs[0].get("metadata") or {}).get("executionId") if len(inputs) == 1 else None
+                    if (not execution or (inputs[0].get("userInput") or {}).get("userResponse") != full_prompt
+                            or planner.get("stopReason") != "STOP_REASON_STOP_PATTERN"
+                            or any(step.get("status") != "CORTEX_STEP_STATUS_DONE"
+                                or (step.get("metadata") or {}).get("executionId") != execution for _, step in current_steps)):
+                        raise CodexChatAgentError("The format review lacks exact normal text-only completion; no actions were admitted.",
+                                                  gate=None, error_code="upstream_transport_failure")
                 if (final.get("type") != "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
                         or final.get("status") != "CORTEX_STEP_STATUS_DONE" or planner.get("toolCalls")
                         or not str(planner.get("modifiedResponse") or planner.get("response") or "").strip()
@@ -818,6 +839,23 @@ def _run_turn(
                 error_code="empty_agent_response",
             )
 
+        if self._read_only_review:
+            try:
+                start = latest_response_text.rfind(CHAT_REVIEW_OPEN_TAG)
+                end = latest_response_text.rfind(CHAT_REVIEW_CLOSE_TAG)
+                if start < 0 or end <= start:
+                    raise ValueError("missing format review envelope")
+                payload = json.loads(latest_response_text[start + len(CHAT_REVIEW_OPEN_TAG):end])
+                if (not isinstance(payload, dict)
+                        or set(payload) - {"schema_version", "message", "context_handoff", "proposals", "gate",
+                                               "goal_draft", "protected_action", "unfinished_work"}
+                        or payload.get("proposals") not in (None, [])
+                        or any(payload.get(key) is not None for key in
+                            ("gate", "goal_draft", "protected_action", "unfinished_work"))):
+                    raise ValueError("format review contains other or unresolved actions")
+            except (TypeError, ValueError) as exc:
+                raise CodexChatAgentError("The format review is incomplete or contains other actions; no actions were admitted.",
+                                          gate=None, error_code="host_gate") from exc
         parsed = parse_agent_response(
             latest_response_text,
             protected_paths=[self.work_dir],