fix(chat): retain native completion before response parsing

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -342,35 +342,49 @@ class AntigravityAdapter:
     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,
-                                     include_raw_response: bool = False) -> dict[str, Any] | None:
+                                     include_raw_response: bool = False,
+                                     retained_result: 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
+            if retained_result is not None:
+                # Core retained this exact invocation before parsing/delivery.
+                # It is not a fresh claim about the provider's latest input.
+                if (input_anchor is None or retained_result.get("thread_id") != cascade_id
+                        or retained_result.get("start_step_index") != input_anchor.get("start_step_index")):
+                    return None
+                total = retained_result.get("total_steps")
+                if type(total) is not int or total < 2:
+                    return None
+                indexed, covered = _observed_turn_steps(retained_result["steps"], input_anchor["start_step_index"], total)
+                if covered != total:
+                    return None
+            else:
+                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
@@ -425,6 +439,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                 "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": start == inputs[-1][0],
+                **({"observation_source": "retained_turn_result"} if retained_result is not None else {}),
                 **({"end_step_index": end} if start != inputs[-1][0] else {}),
                 "normal_terminal_verified": True, "response": response,
                 **({"raw_response": answer} if include_raw_response else {})}
@@ -921,6 +936,26 @@ def _run_turn(
                 break
 
         if event_sink is not None:
+            # Keep only completion facts and the original input/final text,
+            # never native tool output or raw reasoning. Core consumes this
+            # privately and durably before parsing can fail or lose text.
+            facts = []
+            for index, step in current_steps:
+                metadata = step.get("metadata") or {}
+                fact = {"type": step.get("type"), "status": step.get("status"), "metadata": {
+                    "createdAt": metadata.get("createdAt"), "executionId": metadata.get("executionId"),
+                    "sourceTrajectoryStepInfo": {"stepIndex": index}}}
+                if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT":
+                    fact["userInput"] = step.get("userInput")
+                if step.get("type") == "CORTEX_STEP_TYPE_ERROR_MESSAGE":
+                    fact["errorMessage"] = {"shouldShowUser": (step.get("errorMessage") or {}).get("shouldShowUser")}
+                if index == current_steps[-1][0]:
+                    fact["plannerResponse"] = {key: planner[key] for key in
+                        ("response", "modifiedResponse", "stopReason", "toolCalls") if key in planner}
+                facts.append(fact)
+            self._check_interrupted()
+            event_sink("turn.native_result", {"thread_id": self.cascade_id, "upstream_turn_id": turn_id,
+                "start_step_index": start_step_index, "total_steps": total_steps, "steps": facts})
             tail = stream_filter.finish()
             if tail:
                 emitted_visible_delta = True

loopx/capabilities/native_chat/external_conversations.py

@@ -351,6 +351,16 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
                    if event.get("kind") == "turn.started" and event.get("payload", {}).get("native_input")]
         if len(anchors) > 1:
             return None
+        # Retained facts can repair only a completed Turn's text. Failed-Turn
+        # recovery keeps its live/latest-input requirements; old capture does
+        # not authorize another invocation or change unknown-effect handling.
+        retained = self.controller.store.native_result(session["session_id"], turn["turn_id"])
+        if (turn.get("status") == "completed" and len(anchors) == 1 and retained
+                and retained.get("upstream_turn_id") == turn.get("upstream_turn_id")):
+            return 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], include_raw_response=include_raw_response, retained_result=retained)
         return 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"]),

loopx/chat_runtime.py

@@ -1722,6 +1722,16 @@ def event_sink(kind: str, payload: dict[str, Any]) -> None:
             with self.lock:
                 if (session_id, turn_id) in self.cancelled_turns:
                     return
+            if kind == "turn.native_result":
+                if getattr(adapter, "adapter_kind", None) != "antigravity_ls":
+                    raise ValueError("native result capture is unavailable for this adapter")
+                try:
+                    self.store.retain_native_result(session_id, turn_id, payload)
+                except OSError as exc:
+                    # Supplemental retention cannot discard a valid primary
+                    # answer. Missing facts still cannot authorize recovery.
+                    logging.getLogger(__name__).warning("Native completion capture unavailable: %s", type(exc).__name__)
+                return
             event_buffer.emit(kind, payload)
             if kind == "native_goal.status":
                 self.store.update_session(session_id, native_goal=payload)

loopx/chat_store.py

@@ -1390,6 +1390,28 @@ def load_turn(self, session_id: str, turn_id: str) -> dict[str, Any] | None:
         payload = _read_json(self._turn_path(session_id, turn_id))
         return payload if payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION else None
 
+    def native_result(self, session_id: str, turn_id: str) -> dict[str, Any]:
+        """Private provider facts, outside public Turn/event projections."""
+        token = _opaque_id(turn_id, field="turn_id")
+        return _read_json(self._session_dir(session_id) / "native-results" / f"{token}.json")
+
+    def retain_native_result(self, session_id: str, turn_id: str, result: dict[str, Any]) -> None:
+        """Freeze once under the original Turn fence, before response handling."""
+        with exclusive_file_lock(self._turn_path(session_id, turn_id), agent_id="loopx-chat", operation="retain_native_result"):
+            turn = self.load_turn(session_id, turn_id)
+            if turn is None or turn.get("status") != "running":
+                return
+            if not result.get("upstream_turn_id") or result["upstream_turn_id"] != turn.get("upstream_turn_id"):
+                raise ValueError("native result belongs to another invocation")
+            token = _opaque_id(turn_id, field="turn_id")
+            path = self._session_dir(session_id) / "native-results" / f"{token}.json"
+            if path.exists():
+                if _read_json(path) != result:
+                    raise ValueError("the original native result is immutable")
+                return
+            path.parent.mkdir(mode=0o700, exist_ok=True)
+            _atomic_write_json(path, result)
+
     def update_turn(
         self,
         session_id: str,