fix(chat): restore verified native answers after Core failures

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.

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 错误和清理未确认状态。

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()

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:

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: