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..3adf3b1f45 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 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.
+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..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
@@ -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..8184ddf341 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,96 @@ 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") 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"]
+            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}
+                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=recovered["message_text"],
+                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 = 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
+                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..43dd95fd00 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
+                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:
             try:
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 9a2ca48744..8957043df5 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/loopx/control_plane/collaboration/conversation_binding.ts
@@ -461,9 +461,20 @@ 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);
-    return {pending: row.command === null && row.status === "accepted"
+      && (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"
       && 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 +554,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 +572,42 @@ 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
+        && 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
   // 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..f1a911ea49 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/tests/control_plane_ts/conversation_binding.test.ts
@@ -18,6 +18,75 @@ 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);
+  }
+  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,
+    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);
+  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"]) {
   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..1a18b72f30 100644
--- a/tests/test_chat_response_parsing.py
+++ b/tests/test_chat_response_parsing.py
@@ -10,6 +10,20 @@
 )
 
 
+@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"].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
+
+
 @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..d008709a1e 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,113 @@ def connect(fixture):
                                                              runner=provider, cli_bin="lark-cli")
 
 
+@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
+    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 in {"invalid-metadata", "metadata-only"}
+        language_server["unfinished"] = case == "unfinished"
+        def lost_core_terminal(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")
+        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)
+        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"
+        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)
+            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:
+                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"