From 0cb7cdb3747ff120721126070b2cfc03e6df0a69 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 04:20:11 +0530
Subject: [PATCH 1/2] fix(chat): recover verified native results without
 replaying failed turns

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    | 25 +++++
 ...p-conversation-and-async-inbox-v0.zh-CN.md | 18 ++++
 .../native_chat/antigravity_adapter.py        | 97 ++++++++++++++++++-
 .../native_chat/external_conversations.py     | 91 +++++++++++++++++
 loopx/chat.py                                 | 21 +++-
 .../collaboration/conversation_binding.ts     | 49 +++++++++-
 .../extensions/lark/private_conversations.py  | 28 ++++++
 .../conversation_binding.test.ts              | 55 +++++++++++
 tests/test_antigravity_steward_adapter.py     | 82 +++++++++++++++-
 tests/test_chat_response_parsing.py           | 13 +++
 tests/test_lark_private_conversations.py      | 95 ++++++++++++++++++
 11 files changed, 559 insertions(+), 15 deletions(-)

diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index 512580400f..e492357a65 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -1802,3 +1802,28 @@ explicit quota refusal, unknown failure, lost post-write receipt, unavailable
 readback, recovery of older failed journals and two-App isolation. Live provider
 delivery remains separate. Model completion, material intake and an intermediate
 card cannot substitute for final-message readback.
+
+### Recovering a native result after Core failure
+
+An Antigravity owner request can finish natively while Core retains a transport
+failure. The original reply worker may adopt that existing answer, without
+admitting another Turn, sending native input or cancelling an invocation. The
+typed conversation owner requires the original private steward binding, source,
+failed Turn and current grant, with no attachments, steering, stop/new command
+or attached-host routing. The provider read must cover a stable, idle trajectory,
+its latest input and one fully settled native execution ending in a normal final
+response. Correlation requires the exact Core source context, original message
+and input time inside the failed Turn. New inputs additionally persist the
+pre-send prompt hash and native start index in the existing `turn.started` event.
+Missing, conflicting or stream-error evidence keeps the request pending.
+
+The existing request journal reserves the adopted result before appending one
+deterministic canonical Chat message. The original failed Turn, events and
+delivered failure notice remain unchanged. Lark uses a distinct phase in the
+same Inbox intent/readback path; a crash resumes the same message and receipt.
+An unknown post-write acknowledgement remains owed and cannot cause a blind
+resend. Invalid action metadata preserves the answer with an explicit gate and
+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.
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 21ecc52614..3d3a7a6ee9 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -1180,3 +1180,21 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送
 回归覆盖预算跨重启、最终结果保留、明确额度拒绝、未知失败、写入后丢失回执、
 读回不可用、旧失败 journal 升级恢复与两 App 隔离。真实 provider 投递独立验收；
 模型执行完成、材料入库或中间卡片可见均不能代替最终消息读回。
+
+### Core 失败后的原生结果恢复
+
+Antigravity 原生端可能已经完成，而 Core 仍保留 transport 失败。原回复 worker
+可以采用既有答复，不准入另一个 Turn，不发送原生输入，也不取消执行。会话的
+typed owner 核验原私聊管家 binding、来源、失败 Turn 和当前授权，并排除附件、
+steering、stop/new 与 attached-host 路由。provider 读回必须完整覆盖稳定、空闲的
+轨迹、最后一条输入及同一 execution 的已结束步骤，并以正常最终答复结束。关联
+依赖准确的 Core 来源上下文、原消息及落在失败 Turn 执行期间的输入时间；新输入
+还在既有 `turn.started` 事件中保存发送前 prompt hash 和原生起始索引。证据缺失、
+冲突或以 stream error 结束时继续保留待办。
+
+既有请求 journal 先预留采用结果，再以确定性 message id 追加一条 canonical Chat
+消息；原失败 Turn、事件和已送达的失败通知均不改写。飞书沿同一个 Inbox 意图与
+读回路径使用独立阶段，崩溃后恢复同一消息和回执。写入后丢失确认仍记为欠回，
+不能盲目重发。操作元数据无效时保留正文、明确显示 gate，不提交其中任何动作；
+结果恢复本身也不执行有效的操作元数据。即使正文已补回，这些工作仍保持待办；
+正文恢复不证明业务或委派完成。取消失败同时保留原 provider 错误和清理未确认状态。
diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py
index bf27fff59a..8873094d51 100644
--- a/loopx/capabilities/native_chat/antigravity_adapter.py
+++ b/loopx/capabilities/native_chat/antigravity_adapter.py
@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import ipaddress
+import hashlib
 import json
 import math
 import os
@@ -17,6 +18,7 @@
 import urllib.request
 from urllib.parse import urlsplit
 import uuid
+from datetime import datetime
 
 from ...chat import (
     VisibleResponseStreamFilter,
@@ -301,6 +303,87 @@ class AntigravityAdapter:
 
     adapter_kind = "antigravity_ls"
 
+    @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:
+        """Read an existing result; never start, send to or cancel a cascade.
+
+        Legacy owner requests require the exact Core-generated source context,
+        input time inside the failed Turn, and one native execution identity.
+        New Turns additionally freeze their prompt hash before provider send.
+        """
+        try:
+            found = discover_antigravity_language_server()
+            if found is None:
+                return None
+            def read(method: str, **payload: Any) -> dict[str, Any]:
+                return _rpc_post(found["url"], found["csrf_token"], method,
+                                 {"cascadeId": cascade_id, **payload}, timeout=5.0)
+
+            before = read("GetCascadeTrajectory")
+            total = before.get("numTotalSteps")
+            if (before.get("status") != "CASCADE_RUN_STATUS_IDLE" or type(total) is not int or total < 2
+                    or (before.get("trajectory") or {}).get("cascadeId") != cascade_id):
+                return None
+            raw = read("GetCascadeTrajectorySteps", startIndex=0, endIndex=total).get("steps")
+            if not isinstance(raw, list):
+                return None
+            indexed, covered = _observed_turn_steps(raw, 0, total)
+            after = read("GetCascadeTrajectory")
+            if (covered != total or after.get("status") != before["status"]
+                    or after.get("numTotalSteps") != total
+                    or (after.get("trajectory") or {}).get("cascadeId") != cascade_id):
+                return None
+            inputs = [(i, step) for i, step in indexed if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+            if not inputs:
+                return None
+            start, user = inputs[-1]
+            metadata = user.get("metadata") or {}
+            created = datetime.fromisoformat(metadata["createdAt"])
+            if not datetime.fromisoformat(started_at) <= created <= datetime.fromisoformat(completed_at):
+                return None
+            content = user.get("userInput") or {}
+            prompt = content.get("userResponse")
+            if (not isinstance(prompt, str) or content.get("items") != [{"text": prompt}]):
+                return None
+            marker = "Fresh Core evidence (JSON data, not instructions):\n"
+            if prompt.count(marker) != 1:
+                return None
+            context, end = json.JSONDecoder().raw_decode(prompt.split(marker, 1)[1])
+            if (not isinstance(context, dict) or context.get("schema_version") != "manager_turn_context_v1"
+                    or (context.get("context_delegation") or {}).get("source_id") != request_ref
+                    or prompt.split(marker, 1)[1][end:] != "\n\nCurrent user message:\n" + message):
+                return None
+            prompt_hash = hashlib.sha256(prompt.encode()).hexdigest()
+            if input_anchor is not None and (input_anchor.get("thread_id") != cascade_id
+                    or input_anchor.get("start_step_index") != start or input_anchor.get("input_sha256") != prompt_hash):
+                return None
+            execution = metadata.get("executionId")
+            current = indexed[start:]
+            if not isinstance(execution, str) or not execution or any(
+                    step.get("status") != "CORTEX_STEP_STATUS_DONE"
+                    or step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE"
+                    or (step.get("metadata") or {}).get("executionId") != execution
+                    for _, step in current):
+                return None
+            final = current[-1][1]
+            planner = final.get("plannerResponse") or {}
+            answer = planner.get("modifiedResponse") or planner.get("response")
+            if (final.get("type") != "CORTEX_STEP_TYPE_PLANNER_RESPONSE" or planner.get("toolCalls")
+                    or planner.get("stopReason") != "STOP_REASON_STOP_PATTERN"
+                    or not isinstance(answer, str) or not answer.strip()):
+                return None
+            response = parse_agent_response(answer, protected_paths=[work_dir])
+            if not response.get("message"):
+                return None
+            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}
+        except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
+            return None
+
     def __init__(
         self,
         *,
@@ -508,10 +591,16 @@ def start_turn(self, message: str,
         self._cancel_acknowledged = False
         try:
             return self._run_turn(message, event_sink)
-        except BaseException:
+        except BaseException as failure:
             # A timeout/transport or permission failure is not a successful turn.
             # Retain an unconfirmed cancellation as an error for the caller fence.
-            self.interrupt_turn()
+            try:
+                self.interrupt_turn()
+            except Exception as cancellation:
+                note = f"Antigravity cancellation remains unconfirmed ({type(cancellation).__name__})."
+                failure.add_note(note)
+                if isinstance(failure, CodexChatAgentError):
+                    failure.args = (f"{failure} {note}",)
             raise
         finally:
             self._active_turn_id = None
@@ -554,7 +643,9 @@ def _run_turn(
             self._check_interrupted()
             self._active_turn_id = turn_id
             if event_sink is not None:
-                event_sink("turn.started", {"upstream_turn_id": turn_id, "threadId": self.cascade_id})
+                event_sink("turn.started", {"upstream_turn_id": turn_id, "threadId": self.cascade_id,
+                    "native_input": {"thread_id": self.cascade_id, "start_step_index": start_step_index,
+                                     "input_sha256": hashlib.sha256(full_prompt.encode()).hexdigest()}})
             _rpc_post(self.ls_url, self.csrf_token, "SendUserCascadeMessage", send_payload, timeout=15.0)
             self._system_primed = True
             self._pending_history.clear()
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index 77f831fd2b..d015cf0a10 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -179,6 +179,8 @@ def recover_failed_request(self, request_ref: str, *,
             session = self.controller.store.load_session(row["session_id"])
             turn = self.controller.store.load_turn(row["session_id"], row["turn_id"])
             assert session is not None and turn is not None
+            if plan.get("recover_native_result"):
+                return self._recover_native_result(path, row, session, turn, source_observation)
             model_auth_home = checked_session_home(self.controller.codex_home, session)
             context = self.controller.project_contexts.session_context(session)
 
@@ -263,6 +265,95 @@ def source_requests() -> list[dict[str, Any]]:
                 work_dir=context["project"], objective=context["objective"])
             return row
 
+    @staticmethod
+    def _native_result_text(recovered: dict[str, Any]) -> str:
+        response = recovered["native_observation"]["response"]
+        text = "已恢复本次请求的原答复：\n\n" + response["message"]
+        if any(response.get(key) for key in ("context_handoff", "proposals", "protected_action", "goal_draft")):
+            text += "\n\n操作状态：本次仅恢复原答复，结构化操作或委派尚未提交；未重新执行原请求。"
+        if (response.get("gate") or {}).get("summary"):
+            text += "\n\n待处理：" + response["gate"]["summary"]
+        if response.get("unfinished_work"):
+            text += "\n\n未完事项：" + response["unfinished_work"]["remaining"]
+            text += "\n当前依赖：" + response["unfinished_work"]["dependency"]
+        return text
+
+    def _recover_native_result(self, path: Path, row: dict[str, Any], session: dict[str, Any],
+                               turn: dict[str, Any], source_observation: Callable) -> dict[str, Any]:
+        """Adopt an original provider result, without reopening/dispatching a Turn."""
+        from .antigravity_adapter import AntigravityAdapter
+
+        store = self.controller.store
+        requests = [item for item in self.pending() if item.get("binding_id") == row["binding_id"]
+                    and item.get("source") == row["source"]]
+        context = self.controller.project_contexts.session_context(session)
+        recovered = row.get("native_result_recovery")
+        if recovered is None:
+            events = store.events_after(session["session_id"], turn["turn_id"], None)
+            anchors = [event["payload"]["native_input"] for event in events
+                       if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
+            if len(anchors) > 1:
+                return row
+            proof = AntigravityAdapter.completed_result_observation(
+                cascade_id=session["upstream_thread_id"], request_ref=row["request_ref"], message=row["message"],
+                started_at=turn["started_at"], completed_at=turn["completed_at"],
+                work_dir=Path(context["project"]), input_anchor=anchors[0] if anchors else None)
+        else:
+            proof = recovered.get("native_observation")
+        selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
+        source = source_observation(selected)
+        with try_exclusive_file_lock(self.root / "source-fences" / f"{row['binding_id']}.{row['source']['source_ref']}.json",
+                                     operation="route_external_chat_request") as acquired:
+            if acquired is None:
+                return row
+            current = self.bindings.read()
+            if next((item for item in current["bindings"] if item["binding_id"] == row["binding_id"]), None) != selected["binding"]:
+                return row
+            selected = self.bindings._resolve_observed(current, selected["binding"], selected["observation"], row["source"])
+            if (store.load_session(session["session_id"]) != session
+                    or store.load_turn(session["session_id"], turn["turn_id"]) != turn
+                    or [item for item in self.pending() if item.get("binding_id") == row["binding_id"]
+                        and item.get("source") == row["source"]] != requests):
+                return row
+            plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
+                                               selected_binding=selected["binding"])
+            if plan.get("recovery_eligible") is not True:
+                return row
+            if recovered is None:
+                recovered = {"turn_id": turn["turn_id"], "native_observation": proof,
+                    "message_id": f"native.{turn['turn_id']}.recovered", "delivery_verified": False}
+                row["native_result_recovery"] = recovered
+                _atomic_write_json(path, row)
+            # The same explicit message identity closes the journal/message
+            # crash gap. Neither the failed Turn nor its events are rewritten.
+            store.append_message(session["session_id"], role="agent", text=self._native_result_text(recovered),
+                turn_id=turn["turn_id"], origin=turn["origin"], message_id=recovered["message_id"])
+            return row
+
+    def native_result_reply(self, request_ref: str) -> str | None:
+        row = self.read_request(request_ref)
+        recovered = row.get("native_result_recovery")
+        if recovered is None:
+            return None
+        plan = self._context_recovery_plan(row, observation=recovered["native_observation"], source_verified=True)
+        text = self._native_result_text(recovered)
+        message = next((item for item in self.controller.store.messages(row["session_id"])
+                        if item.get("message_id") == recovered["message_id"]), None)
+        if (plan.get("recovery_eligible") is True and message is not None
+                and message.get("turn_id") == row["turn_id"] and message.get("role") == "agent"
+                and message.get("text") == text):
+            return text
+        return None
+
+    def record_native_result_delivery(self, request_ref: str, *, message_id: str) -> None:
+        path = self.root / f"{request_ref}.json"
+        with exclusive_file_lock(path, operation="admit_external_chat_request"):
+            row = self.read_request(request_ref)
+            if row.get("native_result_recovery", {}).get("message_id") != message_id:
+                raise ValueError("original native result changed")
+            row["native_result_recovery"]["delivery_verified"] = True
+            _atomic_write_json(path, row)
+
     def recovery_context(self, session: dict[str, Any], turn: dict[str, Any]) -> dict[str, Any] | None:
         match = re.fullmatch(r"external-([a-f0-9]{24})-recovery-[1-9][0-9]*(?:-((?:routing|quota)-[1-3]|review-1|work-1))?", str(turn.get("client_turn_id") or ""))
         if match is None:
diff --git a/loopx/chat.py b/loopx/chat.py
index e7f1224e71..20b333d3db 100644
--- a/loopx/chat.py
+++ b/loopx/chat.py
@@ -827,11 +827,22 @@ def parse_agent_response(
             if payload.get("message") == "":
                 visible = raw_text[:start].split(CHAT_REVIEW_OPEN_TAG, 1)[0].strip()
                 payload = {**payload, "message": visible}
-            return normalize_agent_response(
-                payload,
-                protected_paths=protected,
-                team_plan_context=team_plan_context,
-            )
+            try:
+                return normalize_agent_response(
+                    payload,
+                    protected_paths=protected,
+                    team_plan_context=team_plan_context,
+                )
+            except ValueError:
+                # A bad action envelope must not discard an already generated
+                # answer or turn a formatting failure into a business retry.
+                # Preserve prose, but admit none of the unvalidated actions.
+                if not isinstance(payload.get("message"), str) or not payload["message"].strip():
+                    raise
+                return normalize_agent_response({"message": payload["message"],
+                    "gate": {"kind": "response_metadata_invalid", "summary":
+                        "答复正文已保留，但操作或委派元数据未通过校验；相关操作和委派尚未提交。"}},
+                    protected_paths=protected)
         key = re.search(r'"message"\s*:\s*', body)
         if key:
             try:
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 9a2ca48744..be5a791f7c 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -463,7 +463,14 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && typeof (gate as JsonObject).kind === "string" && typeof (gate as JsonObject).summary === "string";
     const unfinished = turn?.status === "completed" && response !== null
       && (unfinishedProjectWork(response) !== null || quotaGate);
-    return {pending: row.command === null && row.status === "accepted"
+    const recovered = row.native_result_recovery == null ? null
+      : requireJsonObject(row.native_result_recovery, "recovered native result");
+    const native = recovered?.native_observation as JsonObject | undefined;
+    const result = native?.response as JsonObject | undefined;
+    const returned = recovered?.delivery_verified === true && recovered.turn_id === row.turn_id
+      && result != null && !["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].some(key => result[key])
+      && Array.isArray(result.proposals) && result.proposals.length === 0;
+    return {pending: !returned && row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
       && (unfinished || turn?.status === "failed"
         && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure",
@@ -543,13 +550,14 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
   if (version < 1) throw new EffectRuntimeRequestError("invalid context representation version");
   const no = {operation: "admit_turn", recovery_candidate: false, recovery_eligible: false};
   if (!current || row.command !== null || row.status !== "accepted" || row.delivery_verified !== true
-      || row.agent_target != null || current.session_id !== row.session_id || current.agent_id !== "codex"
-      || current.upstream_mode !== "chat" || current.status === "closed"
+      || row.agent_target != null || current.session_id !== row.session_id
+      || !["codex", "antigravity"].includes(String(current.agent_id))
+      || current.upstream_mode !== (current.agent_id === "antigravity" ? "default" : "chat") || current.status === "closed"
       || current.session_mode === "attached_host") return no;
   const selected = binding(params.binding);
   if (selected.audience === "group") return no;
   const steward = selected.context_kind === "steward";
-  if (selected.executor_endpoint_id !== "codex" || current.goal_id !== (steward ? "loopx-manager" : null)) return no;
+  if (selected.executor_endpoint_id !== current.agent_id || current.goal_id !== (steward ? "loopx-manager" : null)) return no;
   const saved = steward ? normalizeStewardContext(current.steward_context)
     : normalizeProjectContext(current.project_context);
   const source = requireJsonObject(row.source, "original conversation source");
@@ -560,6 +568,39 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       || saved.project_ref !== selected.project_ref || (!steward && saved.grant !== selected.grant)
       || current.channel_id !== `${steward ? "manager.external.native" : "project.external"}.${selected.binding_id}.${saved.source_ref}`) return no;
   const failed = requireJsonObject(recovery.failed_turn, "original failed Turn");
+  if (current.agent_id === "antigravity") {
+    // Result adoption has no admission/dispatch authority. Keep the failed
+    // Turn and original failure delivery, even when native tools ran.
+    if (!steward || failed.session_id !== current.session_id || failed.turn_id !== row.turn_id
+        || failed.client_turn_id !== `external-${row.request_ref}` || failed.status !== "failed"
+        || failed.response != null || failed.message !== row.message || row.context_recovery != null
+        || !["response_timeout", "antigravity_rpc_error", "antigravity_transport_error",
+          "upstream_transport_failure", "server_restarted"].includes(String(failed.error_code))
+        || recovery.attachments_present !== false || recovery.steering_observed !== false
+        || recovery.stop_observed !== false) return no;
+    const adopted = row.native_result_recovery == null ? null
+      : requireJsonObject(row.native_result_recovery, "adopted native result");
+    if (adopted?.delivery_verified === true) return no;
+    const candidate = {...no, operation: "adopt_native_result", recovery_candidate: true,
+      recover_native_result: true, failed_turn_id: failed.turn_id, session_id: current.session_id};
+    const proof = recovery.native_observation == null ? null
+      : requireJsonObject(recovery.native_observation, "native terminal readback");
+    if (proof === null) return candidate;
+    const created = Date.parse(String(proof.input_created_at));
+    const start = Date.parse(String(failed.started_at)), end = Date.parse(String(failed.completed_at));
+    const response = proof.response == null ? null : requireJsonObject(proof.response, "native result response");
+    return {...candidate, recovery_eligible: recovery.source_verified === true
+      && proof.thread_id === current.upstream_thread_id && proof.request_ref === row.request_ref
+      && typeof proof.execution_id === "string" && proof.execution_id.length > 0
+      && proof.latest_input_verified === true && proof.normal_terminal_verified === true
+      && Number.isFinite(created) && created >= start && created <= end
+      && Number.isSafeInteger(proof.start_step_index) && Number(proof.start_step_index) >= 0
+      && Number.isSafeInteger(proof.total_steps) && Number(proof.total_steps) > Number(proof.start_step_index) + 1
+      && [proof.input_sha256, proof.answer_sha256].every(value => typeof value === "string" && /^[a-f0-9]{64}$/.test(value))
+      && response?.schema_version === "loopx_chat_agent_response_v0"
+      && typeof response.message === "string" && response.message.trim().length > 0
+      && (adopted === null || adopted.turn_id === failed.turn_id && isDeepStrictEqual(adopted.native_observation, proof))};
+  }
   const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery");
   // One completed-work continuation is terminal for this request, including
   // when it later fails. It cannot fall through to routing/quota retries.
diff --git a/loopx/extensions/lark/private_conversations.py b/loopx/extensions/lark/private_conversations.py
index 586ca46cd6..70f25f8e30 100644
--- a/loopx/extensions/lark/private_conversations.py
+++ b/loopx/extensions/lark/private_conversations.py
@@ -1078,6 +1078,9 @@ def edit_attempt(value: Any) -> None:
         elif phase_state.get("started"):
             # A crash between provider write and its receipt is ambiguous. Keep
             # the original source recoverable; never resend blindly.
+            phase_state["blocker"] = "provider_delivery_receipt_unknown"
+            record["deliveries"][phase] = phase_state
+            _atomic_write_json(path, record)
             return False
         else:
             def attempt(value: Any) -> None:
@@ -1493,6 +1496,29 @@ def _received_during_recovery(self, path: Path) -> None:
                 return
             self._feedback(path, record, inbox=lambda: self._inbox(record, observation=selected["observation"]))
 
+    def _reconcile_native_result(self, path: Path, native: dict[str, Any]) -> int:
+        # Use the original reply worker and Inbox intent/readback owner. This
+        # separate phase preserves the already delivered failure notice.
+        with exclusive_file_lock(path, operation="deliver_private_chat_request"):
+            record = _read_json(path)
+            if any(record.get(key) != native.get(key) for key in
+                   ("request_ref", "binding_id", "source", "session_id", "turn_id")):
+                return 0
+            text = self.core.native_result_reply(record["request_ref"])
+            if text is None:
+                return 0
+            recovered = native["native_result_recovery"]
+            def authorized() -> bool:
+                return self.core.native_result_reply(record["request_ref"]) == text
+
+            delivery_id = hashlib.sha256(recovered["message_id"].encode()).hexdigest()
+            if not self._deliver(path, record, f"native_result.{record['turn_id']}", text,
+                    inbox=lambda: self._inbox(record), authorization_check=authorized,
+                    reply_placement="chat_root", delivery_id=delivery_id, finalize_source=False):
+                return 0
+            self.core.record_native_result_delivery(record["request_ref"], message_id=recovered["message_id"])
+            return 1
+
     def reconcile_request(self, path: Path) -> int:
         if path.parent == self.root / "web-projections":
             return self._reconcile_web_projection(path)
@@ -1513,6 +1539,8 @@ def reconcile_request(self, path: Path) -> int:
             # the old Turn and every frozen/ambiguous provider attempt.
             recovered = self.core.recover_failed_request(row["request_ref"],
                 source_observation=lambda selected: self._recovery_source_observation(row, selected))
+            if recovered.get("native_result_recovery"):
+                return self._reconcile_native_result(path, recovered)
             if (recovered.get("context_recovery") and not recovered.get("delivery_verified")
                     and recovered.get("turn_id") != row["turn_id"]):
                 with exclusive_file_lock(path, operation="deliver_private_chat_request"):
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index ab6f3d206d..683f547cbf 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -18,6 +18,61 @@ const current = {schema_version: "loopx_chat_conversation_bindings_v0", revision
 const request = {current, expected_revision: 0, operation: "configure", binding: row, observation,
   available_projects: [project]};
 
+test("native terminal recovery adopts a result without Turn admission or lost failure provenance", () => {
+  const selected = {...row, context_kind: "steward", grant: "portfolio_read", goal_ids: [], executor_endpoint_id: "antigravity"};
+  const configured = planConversationBinding({...request, binding: selected}).state;
+  const source = {source_ref: "e".repeat(24), sender_ref: row.operator_ref, private_human_message: true};
+  const context = resolveBoundConversation({current: configured, binding_id: row.binding_id,
+    observation, available_projects: [project], ...source}).context;
+  const ref = "f".repeat(24);
+  const session = {session_id: "original", goal_id: "loopx-manager", agent_id: "antigravity",
+    upstream_mode: "default", status: "ready", upstream_thread_id: "native", active_turn_id: null,
+    channel_id: `manager.external.native.${row.binding_id}.${source.source_ref}`, steward_context: context};
+  const failed = {session_id: "original", turn_id: "failed", client_turn_id: `external-${ref}`, message: "Original request",
+    status: "failed", delta_count: 1, response: null, error_code: "antigravity_rpc_error",
+    started_at: "2026-01-01T00:00:00Z", completed_at: "2026-01-01T00:00:10Z"};
+  const proof = {thread_id: "native", request_ref: ref, execution_id: "original-execution",
+    input_created_at: "2026-01-01T00:00:05Z", start_step_index: 4, total_steps: 8,
+    input_sha256: "1".repeat(64), answer_sha256: "2".repeat(64),
+    latest_input_verified: true, normal_terminal_verified: true,
+    response: {schema_version: "loopx_chat_agent_response_v0", message: "Existing answer", proposals: [], gate: null}};
+  const packet = {request: {request_ref: ref, command: null, status: "accepted", delivery_verified: true,
+    message: "Original request", origin: "lark", binding_id: row.binding_id, session_id: "original", turn_id: "failed", source},
+    binding: selected, current_session: session, context_recovery: {version: 21, failed_turn: failed,
+      steering_observed: false, stop_observed: false, attachments_present: false, source_verified: true, native_observation: proof}};
+  const plan = planBoundConversationRequest(packet);
+  assert.equal(plan.operation, "adopt_native_result");
+  assert.equal(plan.recovery_eligible, true);
+  assert.equal(plan.client_turn_id, undefined);
+  for (const change of [{source_verified: false}, {stop_observed: true}, {steering_observed: true},
+    {attachments_present: true}, {native_observation: {...proof, thread_id: "other"}},
+    {native_observation: {...proof, request_ref: "a".repeat(24)}},
+    {native_observation: {...proof, normal_terminal_verified: false}},
+    {native_observation: {...proof, latest_input_verified: false}},
+    {native_observation: {...proof, input_created_at: "2026-01-02T00:00:00Z"}},
+    {failed_turn: {...failed, client_turn_id: "another-input"}},
+    {failed_turn: {...failed, error_code: "host_gate"}}, {failed_turn: {...failed, status: "interrupted"}}]) {
+    assert.equal(planBoundConversationRequest({...packet, context_recovery: {...packet.context_recovery, ...change}}).recovery_eligible, false);
+  }
+  for (const change of [{status: "closed"}, {session_mode: "attached_host"}, {agent_id: "codex"}]) {
+    assert.equal(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_candidate, false);
+  }
+  const adopted = {turn_id: "failed", native_observation: proof, delivery_verified: true};
+  const delivered = {...packet.request, native_result_recovery: adopted};
+  assert.equal(planBoundConversationRequest({...packet, request: delivered}).recovery_candidate, false);
+  assert.equal(planBoundConversationRequest({request: delivered, current_session: session,
+    pending_observation: {turn: failed, stop_observed: false}}).pending, false);
+  const incomplete = {...delivered, native_result_recovery: {...adopted,
+    native_observation: {...proof, response: {...proof.response, gate: {kind: "response_metadata_invalid"}}}}};
+  assert.equal(planBoundConversationRequest({request: incomplete, current_session: session,
+    pending_observation: {turn: failed, stop_observed: false}}).pending, true);
+  const unfinished = {...delivered, native_result_recovery: {...adopted,
+    native_observation: {...proof, response: {...proof.response,
+      unfinished_work: {remaining: "Still owed", dependency: "Original worker", effects_settled: true}}}}};
+  assert.equal(planBoundConversationRequest({request: unfinished, current_session: session,
+    pending_observation: {turn: failed, stop_observed: false}}).pending, true);
+});
+
 for (const endpoint of ["codex", "antigravity", "kiro-cli"]) {
   test(`retained steward identity permits bound reads after selecting ${endpoint}`, () => {
     const actor = {schema_version: "loopx_steward_managed_actor_v0",
diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py
index 193a01dbcd..d0e7889ca6 100644
--- a/tests/test_antigravity_steward_adapter.py
+++ b/tests/test_antigravity_steward_adapter.py
@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from datetime import datetime, timezone
+from copy import deepcopy
 import json
 import threading
 import time
@@ -140,6 +142,8 @@ def do_POST(self):
                     self.end_headers()
                     return
                 state["sent_count"] += 1
+                state.setdefault("inputs", []).append({"text": body["items"][0]["text"],
+                    "created_at": datetime.now(timezone.utc).isoformat()})
                 state["active"] = True
                 state["cancelled"].clear()
                 state["sent"].set()
@@ -151,16 +155,29 @@ def do_POST(self):
                     answer = f"Argon answer {index + 1}"
                     if not state.get("plain_text"):
                         answer = '<loopx-review-json>' + json.dumps({"message": answer,
-                            "proposals": [], "context_handoff": [], "gate": None}) + '</loopx-review-json>'
+                            "proposals": [], "context_handoff": ([{"goal_id": "work", "agent_id": "worker", "brief": "invalid"}]
+                                if state.get("invalid_handoff") else []), "gate": None,
+                            **({"unfinished_work": {"remaining": "Original work still owed", "dependency": "Original worker",
+                                "effects_settled": True}} if state.get("unfinished") else {})}) + '</loopx-review-json>'
+                    native = state["inputs"][index]
+                    meta = {"createdAt": native["created_at"], "executionId": f"execution-{index}"}
                     response["steps"].extend([
-                        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE"},
+                        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
+                         "metadata": {**meta, "sourceTrajectoryStepInfo": {"stepIndex": index * 2}},
+                         "userInput": {"items": [{"text": native["text"]}], "userResponse": native["text"]}},
                         {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE",
+                         "metadata": {**meta, "sourceTrajectoryStepInfo": {"stepIndex": index * 2 + 1}},
                          "status": "CORTEX_STEP_STATUS_WAITING" if state["mode"] == "waiting" else "CORTEX_STEP_STATUS_DONE",
-                         "plannerResponse": {"thinking": "synthetic status", "response": answer}}])
+                         "plannerResponse": {"thinking": "synthetic status", "response": answer,
+                                             "stopReason": "STOP_REASON_STOP_PATTERN"}}])
                 if state.get("native_indices"):
                     for offset, step in enumerate(response["steps"]):
                         step["metadata"] = {"sourceTrajectoryStepInfo": {"stepIndex": offset}}
             elif method == "CancelCascadeInvocation":
+                if state.get("cancel_failure"):
+                    self.send_response(500)
+                    self.end_headers()
+                    return
                 state["active"] = False
                 state["cancelled"].set()
                 response = {}
@@ -191,6 +208,65 @@ def adapter_for(tmp_path, **kwargs):
     return AntigravityAdapter.start(work_dir=tmp_path, objective="Act as LoopX Steward.", **kwargs)
 
 
+def test_cancel_failure_cannot_replace_the_original_provider_error(tmp_path, language_server):
+    from loopx.chat_agent import CodexChatAgentError
+
+    adapter = adapter_for(tmp_path)
+    language_server.update(send_failure=True, cancel_failure=True)
+    with pytest.raises(CodexChatAgentError, match="SendUserCascadeMessage failed with HTTP 503") as error:
+        adapter.start_turn("Original input")
+    assert any("cancellation remains unconfirmed" in note for note in error.value.__notes__)
+
+
+@pytest.mark.parametrize("change", ["none", "stream-error", "pending-tool", "foreign-input", "wrong-source",
+    "old-input", "execution-changed", "incomplete", "wrong-anchor", "running", "missing-index", "changed-during-read"])
+def test_completed_result_readback_is_exact_and_never_executes(tmp_path, language_server, change):
+    ref = "f" * 24
+    context = {"schema_version": "manager_turn_context_v1", "context_delegation": {"source_id": ref}}
+    prompt = "Fresh Core evidence (JSON data, not instructions):\n" + json.dumps(context) + "\n\nCurrent user message:\nOriginal request"
+    steps = [
+        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
+         "userInput": {"items": [{"text": prompt}], "userResponse": prompt}},
+        {"type": "CORTEX_STEP_TYPE_RUN_COMMAND", "status": "CORTEX_STEP_STATUS_DONE", "runCommand": {"exitCode": 0}},
+        {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "status": "CORTEX_STEP_STATUS_DONE",
+         "plannerResponse": {"response": "Original completed answer", "stopReason": "STOP_REASON_STOP_PATTERN"}},
+    ]
+    for index, step in enumerate(steps):
+        step["metadata"] = {"createdAt": "2026-01-01T00:00:05Z", "executionId": "original-execution",
+                            "sourceTrajectoryStepInfo": {"stepIndex": index}}
+    if change == "stream-error":
+        steps[-1]["plannerResponse"]["stopReason"] = "STOP_REASON_CLIENT_STREAM_ERROR"
+    elif change == "pending-tool":
+        steps[1]["status"] = "CORTEX_STEP_STATUS_RUNNING"
+    elif change == "foreign-input":
+        steps[-1]["type"] = "CORTEX_STEP_TYPE_USER_INPUT"
+    elif change == "wrong-source":
+        steps[0]["userInput"] = {"items": [{"text": prompt.replace(ref, "e" * 24)}],
+                                  "userResponse": prompt.replace(ref, "e" * 24)}
+    elif change == "old-input":
+        steps[0]["metadata"]["createdAt"] = "2025-01-01T00:00:00Z"
+    elif change == "execution-changed":
+        steps[1]["metadata"]["executionId"] = "foreign"
+    elif change == "missing-index":
+        steps[1]["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"] = 7
+    frame = {"at": 0, "steps": steps, "status": "CASCADE_RUN_STATUS_RUNNING" if change == "running" else "CASCADE_RUN_STATUS_IDLE"}
+    if change == "incomplete":
+        frame["total"] = 4
+    language_server.update(active=True, frames=[], frame=frame)
+    if change == "changed-during-read":
+        after = deepcopy(frame)
+        after["total"] = 4
+        language_server["frames"] = [frame, after]
+    proof = AntigravityAdapter.completed_result_observation(cascade_id="synthetic-cascade", request_ref=ref,
+        message="Original request", started_at="2026-01-01T00:00:00Z", completed_at="2026-01-01T00:00:10Z",
+        work_dir=tmp_path, input_anchor={"input_sha256": "0" * 64} if change == "wrong-anchor" else None)
+    assert (proof is not None) == (change == "none")
+    if proof:
+        assert proof["response"]["message"] == "Original completed answer"
+    assert not any(method in {"StartCascade", "SendUserCascadeMessage", "CancelCascadeInvocation"}
+                   for method, _ in language_server["calls"])
+
+
 def test_antigravity_adapter_executes_repeated_turns_and_streams_response(tmp_path, language_server):
     available, executable, _ = probe_antigravity_endpoint()
     assert available and executable == "antigravity-ls"
diff --git a/tests/test_chat_response_parsing.py b/tests/test_chat_response_parsing.py
index 9ef540e33b..ea3d127b9b 100644
--- a/tests/test_chat_response_parsing.py
+++ b/tests/test_chat_response_parsing.py
@@ -10,6 +10,19 @@
 )
 
 
+@pytest.mark.parametrize("metadata", [
+    {"context_handoff": [{"goal_id": "work", "agent_id": "worker", "brief": "invalid brief"}]},
+    {"context_handoff": {"goal_id": "work", "agent_id": "worker", "brief": []}},
+])
+def test_invalid_action_metadata_preserves_answer_without_admitting_actions(metadata):
+    response = parse_agent_response("The original answer.\n" + CHAT_REVIEW_OPEN_TAG
+        + json.dumps({"message": "", **metadata}) + CHAT_REVIEW_CLOSE_TAG)
+    assert response["message"] == "The original answer."
+    assert response["gate"]["kind"] == "response_metadata_invalid"
+    assert not response.get("context_handoff") and not response.get("goal_draft")
+    assert response["proposals"] == [] and response["protected_action"] is None
+
+
 @pytest.mark.parametrize("change", [None, {"effects_settled": False}, {"effects_settled": "true"},
     {"remaining": ""}, {"remaining": "x" * 1001}, {"dependency": 4}, {"grant": "workspace_write"}])
 def test_unfinished_work_is_explicit_bounded_data_never_inferred_from_prose(change):
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 42a24fc533..122fd172bc 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -10,6 +10,7 @@
 
 import pytest
 from test_chat_ordinary_project import ordinary  # noqa: F401
+from test_antigravity_steward_adapter import language_server  # noqa: F401
 
 from loopx.capabilities.native_chat.conversation_bindings import ChatConversationBindings
 from loopx.capabilities.native_chat.external_conversations import ChatExternalConversations
@@ -220,6 +221,100 @@ def connect(fixture):
                                                              runner=provider, cli_bin="lark-cli")
 
 
+@pytest.mark.parametrize("case", ["normal", "invalid-metadata", "unfinished", "source-changed", "closed", "lost-ack", "message-crash"])
+def test_original_worker_recovers_native_result_without_replay(tmp_path, language_server, monkeypatch, case):  # noqa: F811
+    from test_native_steward_private import steward
+    from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
+    from loopx.chat_agent import CodexChatAgentError
+    from loopx.chat_store import _read_json
+
+    monkeypatch.setenv("LOOPX_MANAGER_MODEL", "gemini-4-argon-high")
+    monkeypatch.setenv("LOOPX_MANAGER_REASONING_EFFORT", "high")
+    fixture = steward.__wrapped__(tmp_path)
+    store, runtime, provider, transport, binding, _, workspace = next(fixture)
+    try:
+        binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
+            executor_endpoint_id="antigravity", context_kind="steward")
+        original = AntigravityAdapter.start_turn
+        language_server["invalid_handoff"] = case == "invalid-metadata"
+        language_server["unfinished"] = case == "unfinished"
+        def lost_core_terminal(adapter, message, event_sink):
+            original(adapter, message, event_sink)
+            raise CodexChatAgentError("Synthetic lost Core terminal commit", gate=None, error_code="antigravity_rpc_error")
+        monkeypatch.setattr(AntigravityAdapter, "start_turn", lost_core_terminal)
+        event = provider.event("steward-app", "native-original", "Complete the original request")
+        transport.admit("steward-app", event)
+        ref = identity_ref(binding["provider_ref"], event["message_id"])
+        row = transport.core.read_request(ref)
+        failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)
+        assert failed["status"] == "failed" and failed["error_code"] == "antigravity_rpc_error"
+        assert transport.reconcile_request(transport.root / f"{ref}.json") == 1
+        path = transport.root / f"{ref}.json"
+        old = _read_json(path)
+        if case not in {"source-changed", "closed"}:
+            assert transport.core.context_recovery_candidate(ref)
+        failed_bytes = store._turn_path(row["session_id"], row["turn_id"]).read_bytes()
+        writes = len(provider.writes)
+        monkeypatch.setattr(runtime, "enqueue_turn", lambda **_: pytest.fail("Result recovery enqueued work"))
+        monkeypatch.setattr(runtime, "resume_session_queue", lambda **_: pytest.fail("Result recovery resumed work"))
+        if case == "source-changed":
+            provider.messages[event["message_id"]]["content"] = "Edited source"
+        elif case == "closed":
+            store.update_session(row["session_id"], status="closed")
+        elif case == "lost-ack":
+            provider.lose_next_reply = True
+        elif case == "message-crash":
+            append = store.append_message
+            def crash(*args, **kwargs):
+                if str(kwargs.get("message_id", "")).startswith("native."):
+                    raise OSError("Synthetic journal/message crash")
+                return append(*args, **kwargs)
+            monkeypatch.setattr(store, "append_message", crash)
+            with pytest.raises(OSError, match="journal/message crash"):
+                transport.reconcile()
+            monkeypatch.setattr(store, "append_message", append)
+        if case == "lost-ack":
+            # The original intent owner retains/reconciles an uncertain write.
+            try:
+                transport.reconcile()
+            except OSError:
+                pass
+        restarted = LarkPrivateConversations(controller=runtime, runtime_root=transport.runtime_root,
+            runner=provider, cli_bin="lark-cli")
+        restarted.reconcile()
+        native = restarted.core.read_request(ref)
+        recovered = native.get("native_result_recovery")
+        if case in {"source-changed", "closed"}:
+            assert recovered is None and len(provider.writes) == writes
+        elif case == "lost-ack":
+            assert recovered["delivery_verified"] is False
+            receipt = _read_json(path)["deliveries"][f"native_result.{row['turn_id']}"]
+            assert receipt["started"] and not receipt.get("attempt")
+            assert receipt["blocker"] == "provider_delivery_receipt_unknown"
+            assert len(provider.writes) == writes + 1
+            restarted.reconcile()
+            assert len(provider.writes) == writes + 1  # Unknown delivery remains owed.
+            assert restarted.health()[row["binding_id"]]["pending_count"] == 1
+        else:
+            assert recovered["delivery_verified"] is True
+            message = next(item for item in store.messages(row["session_id"]) if item["message_id"] == recovered["message_id"])
+            receipt = _read_json(path)["deliveries"][f"native_result.{row['turn_id']}"]
+            body = json.loads(provider.messages[receipt["attempt"]["message_ref"]]["body"]["content"])
+            assert body["zh_cn"]["content"][0][0]["text"] == message["text"]
+            assert "Argon answer 1" in message["text"]
+            assert len(provider.writes) == writes + 1
+            assert restarted.health()[row["binding_id"]]["pending_count"] == (case in {"invalid-metadata", "unfinished"})
+            if case == "unfinished":
+                assert "Original work still owed" in message["text"]
+            assert len([item for item in store.messages(row["session_id"]) if item["message_id"] == recovered["message_id"]]) == 1
+        assert _read_json(path)["deliveries"]["terminal"] == old["deliveries"]["terminal"]
+        assert store._turn_path(row["session_id"], row["turn_id"]).read_bytes() == failed_bytes
+        assert len(list(store._turn_path(row["session_id"], row["turn_id"]).parent.glob("*.json"))) == 1
+        assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+    finally:
+        fixture.close()
+
+
 def connect_group(fixture):
     store, runtime, provider, transport = connect(fixture)
     runtime.codex_home = fixture[-1].parent / "isolated-native-account"

From 4e3e4df75ba8b73f2eac9c3a651b8cf344ee8e29 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Mon, 2 Nov 2026 04:30:34 +0530
Subject: [PATCH 2/2] fix(chat): retain gated replies and freeze recovered
 message intent

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../app-conversation-and-async-inbox-v0.md    |  2 +-
 ...p-conversation-and-async-inbox-v0.zh-CN.md |  2 +-
 .../native_chat/external_conversations.py     |  7 ++++---
 loopx/chat.py                                 |  6 +++---
 .../collaboration/conversation_binding.ts     | 11 +++++++++--
 .../conversation_binding.test.ts              | 16 +++++++++++++++-
 tests/test_chat_response_parsing.py           |  3 ++-
 tests/test_lark_private_conversations.py      | 19 ++++++++++++++++---
 8 files changed, 51 insertions(+), 15 deletions(-)

diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
index e492357a65..3adf3b1f45 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md
@@ -1817,7 +1817,7 @@ and input time inside the failed Turn. New inputs additionally persist the
 pre-send prompt hash and native start index in the existing `turn.started` event.
 Missing, conflicting or stream-error evidence keeps the request pending.
 
-The existing request journal reserves the adopted result before appending one
+The existing request journal freezes the adopted result and rendered message before appending one
 deterministic canonical Chat message. The original failed Turn, events and
 delivered failure notice remain unchanged. Lark uses a distinct phase in the
 same Inbox intent/readback path; a crash resumes the same message and receipt.
diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
index 3d3a7a6ee9..3c24da32cc 100644
--- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
+++ b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md
@@ -1192,7 +1192,7 @@ steering、stop/new 与 attached-host 路由。provider 读回必须完整覆盖
 还在既有 `turn.started` 事件中保存发送前 prompt hash 和原生起始索引。证据缺失、
 冲突或以 stream error 结束时继续保留待办。
 
-既有请求 journal 先预留采用结果，再以确定性 message id 追加一条 canonical Chat
+既有请求 journal 先冻结采用结果和消息正文，再以确定性 message id 追加一条 canonical Chat
 消息；原失败 Turn、事件和已送达的失败通知均不改写。飞书沿同一个 Inbox 意图与
 读回路径使用独立阶段，崩溃后恢复同一消息和回执。写入后丢失确认仍记为欠回，
 不能盲目重发。操作元数据无效时保留正文、明确显示 gate，不提交其中任何动作；
diff --git a/loopx/capabilities/native_chat/external_conversations.py b/loopx/capabilities/native_chat/external_conversations.py
index d015cf0a10..8184ddf341 100644
--- a/loopx/capabilities/native_chat/external_conversations.py
+++ b/loopx/capabilities/native_chat/external_conversations.py
@@ -271,7 +271,7 @@ def _native_result_text(recovered: dict[str, Any]) -> str:
         text = "已恢复本次请求的原答复：\n\n" + response["message"]
         if any(response.get(key) for key in ("context_handoff", "proposals", "protected_action", "goal_draft")):
             text += "\n\n操作状态：本次仅恢复原答复，结构化操作或委派尚未提交；未重新执行原请求。"
-        if (response.get("gate") or {}).get("summary"):
+        if (response.get("gate") or {}).get("summary") and response["gate"]["summary"] not in text:
             text += "\n\n待处理：" + response["gate"]["summary"]
         if response.get("unfinished_work"):
             text += "\n\n未完事项：" + response["unfinished_work"]["remaining"]
@@ -322,11 +322,12 @@ def _recover_native_result(self, path: Path, row: dict[str, Any], session: dict[
             if recovered is None:
                 recovered = {"turn_id": turn["turn_id"], "native_observation": proof,
                     "message_id": f"native.{turn['turn_id']}.recovered", "delivery_verified": False}
+                recovered["message_text"] = self._native_result_text(recovered)
                 row["native_result_recovery"] = recovered
                 _atomic_write_json(path, row)
             # The same explicit message identity closes the journal/message
             # crash gap. Neither the failed Turn nor its events are rewritten.
-            store.append_message(session["session_id"], role="agent", text=self._native_result_text(recovered),
+            store.append_message(session["session_id"], role="agent", text=recovered["message_text"],
                 turn_id=turn["turn_id"], origin=turn["origin"], message_id=recovered["message_id"])
             return row
 
@@ -336,7 +337,7 @@ def native_result_reply(self, request_ref: str) -> str | None:
         if recovered is None:
             return None
         plan = self._context_recovery_plan(row, observation=recovered["native_observation"], source_verified=True)
-        text = self._native_result_text(recovered)
+        text = recovered["message_text"]
         message = next((item for item in self.controller.store.messages(row["session_id"])
                         if item.get("message_id") == recovered["message_id"]), None)
         if (plan.get("recovery_eligible") is True and message is not None
diff --git a/loopx/chat.py b/loopx/chat.py
index 20b333d3db..43dd95fd00 100644
--- a/loopx/chat.py
+++ b/loopx/chat.py
@@ -839,9 +839,9 @@ def parse_agent_response(
                 # Preserve prose, but admit none of the unvalidated actions.
                 if not isinstance(payload.get("message"), str) or not payload["message"].strip():
                     raise
-                return normalize_agent_response({"message": payload["message"],
-                    "gate": {"kind": "response_metadata_invalid", "summary":
-                        "答复正文已保留，但操作或委派元数据未通过校验；相关操作和委派尚未提交。"}},
+                summary = "答复正文已保留，但操作或委派元数据未通过校验；相关操作和委派尚未提交。"
+                return normalize_agent_response({"message": payload["message"] + "\n\n待处理：" + summary,
+                    "gate": {"kind": "response_metadata_invalid", "summary": summary}},
                     protected_paths=protected)
         key = re.search(r'"message"\s*:\s*', body)
         if key:
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index be5a791f7c..8957043df5 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -461,13 +461,17 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
       && turn?.client_turn_id === reservation.client_turn_id
       && gate !== null && typeof gate === "object" && !Array.isArray(gate)
       && typeof (gate as JsonObject).kind === "string" && typeof (gate as JsonObject).summary === "string";
+    const invalidMetadata = gate !== null && typeof gate === "object" && !Array.isArray(gate)
+      && (gate as JsonObject).kind === "response_metadata_invalid" && typeof (gate as JsonObject).summary === "string";
     const unfinished = turn?.status === "completed" && response !== null
-      && (unfinishedProjectWork(response) !== null || quotaGate);
+      && (unfinishedProjectWork(response) !== null || quotaGate || invalidMetadata);
     const recovered = row.native_result_recovery == null ? null
       : requireJsonObject(row.native_result_recovery, "recovered native result");
     const native = recovered?.native_observation as JsonObject | undefined;
     const result = native?.response as JsonObject | undefined;
     const returned = recovered?.delivery_verified === true && recovered.turn_id === row.turn_id
+      && recovered.message_id === `native.${row.turn_id}.recovered`
+      && typeof recovered.message_text === "string" && recovered.message_text.length > 0
       && result != null && !["gate", "context_handoff", "protected_action", "goal_draft", "unfinished_work"].some(key => result[key])
       && Array.isArray(result.proposals) && result.proposals.length === 0;
     return {pending: !returned && row.command === null && row.status === "accepted"
@@ -599,7 +603,10 @@ function planContextRecovery(params: JsonObject, row: JsonObject, current: JsonO
       && [proof.input_sha256, proof.answer_sha256].every(value => typeof value === "string" && /^[a-f0-9]{64}$/.test(value))
       && response?.schema_version === "loopx_chat_agent_response_v0"
       && typeof response.message === "string" && response.message.trim().length > 0
-      && (adopted === null || adopted.turn_id === failed.turn_id && isDeepStrictEqual(adopted.native_observation, proof))};
+      && (adopted === null || adopted.turn_id === failed.turn_id
+        && adopted.message_id === `native.${failed.turn_id}.recovered`
+        && typeof adopted.message_text === "string" && adopted.message_text.length > 0
+        && isDeepStrictEqual(adopted.native_observation, proof))};
   }
   const previous = row.context_recovery == null ? null : requireJsonObject(row.context_recovery, "reserved context recovery");
   // One completed-work continuation is terminal for this request, including
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 683f547cbf..f1a911ea49 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -57,7 +57,18 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
   for (const change of [{status: "closed"}, {session_mode: "attached_host"}, {agent_id: "codex"}]) {
     assert.equal(planBoundConversationRequest({...packet, current_session: {...session, ...change}}).recovery_candidate, false);
   }
-  const adopted = {turn_id: "failed", native_observation: proof, delivery_verified: true};
+  assert.throws(() => planBoundConversationRequest({...packet,
+    binding: {...selected, audience: "group", group_refs: ["a".repeat(24)]}}), /group Chat requires a native project executor/);
+  const newRef = "9".repeat(24);
+  assert.equal(planBoundConversationRequest({...packet, request: {...packet.request, request_ref: newRef},
+    context_recovery: {...packet.context_recovery, failed_turn: {...failed, client_turn_id: `external-${newRef}`},
+      native_observation: {...proof, request_ref: newRef}}}).recovery_eligible, true);
+  const adopted = {turn_id: "failed", message_id: "native.failed.recovered", message_text: "Frozen answer",
+    native_observation: proof, delivery_verified: true};
+  assert.equal(planBoundConversationRequest({...packet,
+    request: {...packet.request, native_result_recovery: {...adopted, delivery_verified: false}}}).recovery_eligible, true);
+  assert.equal(planBoundConversationRequest({...packet,
+    request: {...packet.request, native_result_recovery: {...adopted, delivery_verified: false, message_id: "other"}}}).recovery_eligible, false);
   const delivered = {...packet.request, native_result_recovery: adopted};
   assert.equal(planBoundConversationRequest({...packet, request: delivered}).recovery_candidate, false);
   assert.equal(planBoundConversationRequest({request: delivered, current_session: session,
@@ -71,6 +82,9 @@ test("native terminal recovery adopts a result without Turn admission or lost fa
       unfinished_work: {remaining: "Still owed", dependency: "Original worker", effects_settled: true}}}}};
   assert.equal(planBoundConversationRequest({request: unfinished, current_session: session,
     pending_observation: {turn: failed, stop_observed: false}}).pending, true);
+  assert.equal(planBoundConversationRequest({request: packet.request, current_session: session,
+    pending_observation: {turn: {...failed, status: "completed", response: {...proof.response,
+      gate: {kind: "response_metadata_invalid", summary: "Action metadata was not admitted"}}}, stop_observed: false}}).pending, true);
 });
 
 for (const endpoint of ["codex", "antigravity", "kiro-cli"]) {
diff --git a/tests/test_chat_response_parsing.py b/tests/test_chat_response_parsing.py
index ea3d127b9b..1a18b72f30 100644
--- a/tests/test_chat_response_parsing.py
+++ b/tests/test_chat_response_parsing.py
@@ -17,7 +17,8 @@
 def test_invalid_action_metadata_preserves_answer_without_admitting_actions(metadata):
     response = parse_agent_response("The original answer.\n" + CHAT_REVIEW_OPEN_TAG
         + json.dumps({"message": "", **metadata}) + CHAT_REVIEW_CLOSE_TAG)
-    assert response["message"] == "The original answer."
+    assert response["message"].startswith("The original answer.\n\n待处理：")
+    assert "相关操作和委派尚未提交" in response["message"]
     assert response["gate"]["kind"] == "response_metadata_invalid"
     assert not response.get("context_handoff") and not response.get("goal_draft")
     assert response["proposals"] == [] and response["protected_action"] is None
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 122fd172bc..d008709a1e 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -221,7 +221,7 @@ def connect(fixture):
                                                              runner=provider, cli_bin="lark-cli")
 
 
-@pytest.mark.parametrize("case", ["normal", "invalid-metadata", "unfinished", "source-changed", "closed", "lost-ack", "message-crash"])
+@pytest.mark.parametrize("case", ["normal", "invalid-metadata", "metadata-only", "unfinished", "source-changed", "closed", "lost-ack", "message-crash"])
 def test_original_worker_recovers_native_result_without_replay(tmp_path, language_server, monkeypatch, case):  # noqa: F811
     from test_native_steward_private import steward
     from loopx.capabilities.native_chat.antigravity_adapter import AntigravityAdapter
@@ -236,10 +236,12 @@ def test_original_worker_recovers_native_result_without_replay(tmp_path, languag
         binding = transport.bindings.configure(transport_ref="steward-app", project_ref=binding["project_ref"],
             executor_endpoint_id="antigravity", context_kind="steward")
         original = AntigravityAdapter.start_turn
-        language_server["invalid_handoff"] = case == "invalid-metadata"
+        language_server["invalid_handoff"] = case in {"invalid-metadata", "metadata-only"}
         language_server["unfinished"] = case == "unfinished"
         def lost_core_terminal(adapter, message, event_sink):
-            original(adapter, message, event_sink)
+            result = original(adapter, message, event_sink)
+            if case == "metadata-only":
+                return result
             raise CodexChatAgentError("Synthetic lost Core terminal commit", gate=None, error_code="antigravity_rpc_error")
         monkeypatch.setattr(AntigravityAdapter, "start_turn", lost_core_terminal)
         event = provider.event("steward-app", "native-original", "Complete the original request")
@@ -247,6 +249,15 @@ def lost_core_terminal(adapter, message, event_sink):
         ref = identity_ref(binding["provider_ref"], event["message_id"])
         row = transport.core.read_request(ref)
         failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)
+        if case == "metadata-only":
+            assert failed["status"] == "completed" and failed["response"]["gate"]["kind"] == "response_metadata_invalid"
+            assert transport.reconcile() == 1
+            assert transport.health()[row["binding_id"]]["pending_count"] == 1
+            assert "Argon answer 1" in provider.writes[-1][1]
+            assert "相关操作和委派尚未提交" in provider.writes[-1][1]
+            assert not transport.core.context_recovery_candidate(ref)  # No business replay.
+            assert language_server["sent_count"] == 1
+            return
         assert failed["status"] == "failed" and failed["error_code"] == "antigravity_rpc_error"
         assert transport.reconcile_request(transport.root / f"{ref}.json") == 1
         path = transport.root / f"{ref}.json"
@@ -273,6 +284,8 @@ def crash(*args, **kwargs):
             with pytest.raises(OSError, match="journal/message crash"):
                 transport.reconcile()
             monkeypatch.setattr(store, "append_message", append)
+            monkeypatch.setattr(ChatExternalConversations, "_native_result_text",
+                                staticmethod(lambda _: pytest.fail("Frozen message intent was regenerated")))
         if case == "lost-ack":
             # The original intent owner retains/reconciles an uncertain write.
             try: