fix(chat): recover completed steward handoffs after quota restoration

loopx/canary/module_metric_baseline.json

@@ -44,7 +44,7 @@
       "lines": 2035
     },
     "loopx/chat_store.py": {
-      "lines": 2066
+      "lines": 2074
     },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -105,6 +105,27 @@ def _check_hard_timeout(value: float | None) -> None:
         raise ValueError("Antigravity hard timeout must be finite and positive")
 
 
+def _native_completion_facts(cascade_id: str, current: list[tuple[int, dict[str, Any]]]) -> dict[str, Any]:
+    """One private completion projection, without tool outputs or reasoning."""
+    facts = []
+    for index, step in current:
+        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[-1][0]:
+            planner = step.get("plannerResponse") or {}
+            fact["plannerResponse"] = {key: planner[key] for key in
+                ("response", "modifiedResponse", "stopReason", "toolCalls") if key in planner}
+        facts.append(fact)
+    return {"thread_id": cascade_id, "start_step_index": current[0][0],
+        "total_steps": current[-1][0] + 1, "steps": facts}
+
+
 class _LocalRedirectHandler(urllib.request.HTTPRedirectHandler):
     def redirect_request(self, req: Any, fp: Any, code: int, msg: str,
                          headers: Any, newurl: str) -> None:
@@ -359,12 +380,80 @@ class AntigravityAdapter:
 
     adapter_kind = "antigravity_ls"
 
+    @staticmethod
+    def metadata_review_capacity(*, cascade_id: str, model_id: str) -> dict[str, Any] | None:
+        """Observe the existing model's quota; never probe it with generation.
+
+        A positive fraction alone is insufficient: an invocation can have
+        reached its limit with a small advertised remainder. Preserve that
+        native failure alongside the exact model's quota epoch.
+        """
+        try:
+            found = discover_antigravity_language_server()
+            if found is None:
+                return None
+            def read(method: str, payload: dict[str, Any]) -> dict[str, Any]:
+                return _rpc_post(found["url"], found["csrf_token"], method, payload, timeout=5.0)
+            before = read("GetCascadeTrajectory", {"cascadeId": cascade_id})
+            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
+            start = max(0, total - 128)
+            steps, covered = _observed_turn_steps(read("GetCascadeTrajectorySteps",
+                {"cascadeId": cascade_id, "startIndex": start, "endIndex": total})["steps"], start, total)
+            inputs = [i for i, step in steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+            if not inputs and start:
+                steps, covered = _observed_turn_steps(read("GetCascadeTrajectorySteps",
+                    {"cascadeId": cascade_id, "startIndex": 0, "endIndex": total})["steps"], 0, total)
+                inputs = [i for i, step in steps if step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT"]
+            if not inputs or covered != total:
+                return None
+            latest = [step for i, step in steps if i >= inputs[-1]]
+            exhausted = any(str(((step.get("errorMessage") or {}).get("error") or {}).get(
+                "userErrorMessage") or "").startswith("Individual quota reached.")
+                for step in latest)
+            planner = latest[-1].get("plannerResponse") or {}
+            normal = (latest[-1].get("type") == "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
+                and planner.get("stopReason") == "STOP_REASON_STOP_PATTERN" and not planner.get("toolCalls")
+                and all(step.get("status") == "CORTEX_STEP_STATUS_DONE" for step in latest)
+                and isinstance((latest[0].get("metadata") or {}).get("executionId"), str)
+                and bool((latest[0].get("metadata") or {}).get("executionId"))
+                and len({(step.get("metadata") or {}).get("executionId") for step in latest}) == 1)
+            if not normal and not exhausted:
+                return None  # An unknown native failure is not capacity evidence.
+            status = read("GetUserStatus", {})
+            user = status["userStatus"]
+            account = user.get("email")
+            if not isinstance(account, str) or not account.strip():
+                return None
+            rows = user["cascadeModelConfigData"]["clientModelConfigs"]
+            matching = [row for row in rows if row.get("modelId") == model_id]
+            if len(matching) != 1:
+                return None
+            quota = matching[0]["quotaInfo"]
+            fraction, reset = quota.get("remainingFraction", 0), quota.get("resetTime")
+            if (type(fraction) not in (int, float) or not math.isfinite(fraction)
+                    or not 0 <= fraction <= 1 or not isinstance(reset, str)
+                    or datetime.fromisoformat(reset).tzinfo is None):
+                return None
+            after = read("GetCascadeTrajectory", {"cascadeId": cascade_id})
+            if (after.get("status") != before["status"] or after.get("numTotalSteps") != total
+                    or (after.get("trajectory") or {}).get("cascadeId") != cascade_id):
+                return None
+            return {"model_id": model_id, "account_ref": hashlib.sha256(account.strip().casefold().encode()).hexdigest(),
+                "remaining_fraction": fraction, "reset_time": reset,
+                "observed_at": time.time(), "quota_exhausted": exhausted, "normal_completion": normal}
+        except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
+            return None
+
     @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,
                                      include_raw_response: bool = False,
-                                     retained_result: dict[str, Any] | None = None) -> dict[str, Any] | None:
+                                     retained_result: dict[str, Any] | None = None,
+                                     include_retained_result: bool = False) -> dict[str, Any] | None:
         """Read an existing result; never start, send to or cancel a cascade.
 
         Legacy owner requests require the exact Core-generated source context,
@@ -461,6 +550,7 @@ def read(method: str, **payload: Any) -> dict[str, Any]:
                 **({"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,
+                **({"retained_result": _native_completion_facts(cascade_id, current)} if include_retained_result else {}),
                 **({"raw_response": answer} if include_raw_response else {})}
         except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
             return None
@@ -984,23 +1074,9 @@ def _run_turn(
             # 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})
+            event_sink("turn.native_result", {**_native_completion_facts(self.cascade_id, current_steps),
+                "upstream_turn_id": turn_id})
             tail = stream_filter.finish()
             if tail:
                 emitted_visible_delta = True

loopx/capabilities/native_chat/external_conversations.py

@@ -51,7 +51,8 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                                source_verified: bool = False,
                                selected_binding: dict[str, Any] | None = None,
                                discovery_only: bool = False,
-                               preparation_scope_available: bool = False) -> dict[str, Any]:
+                               preparation_scope_available: bool = False,
+                               metadata_capacity: dict[str, Any] | None = None) -> dict[str, Any]:
         from ...chat_manager import MANAGER_CONTEXT_VERSION
         from ...control_plane.effect_runtime import effect_runtime_result
 
@@ -113,6 +114,7 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "preparation_scope_available": preparation_scope_available,
                 "native_input": anchors[0] if len(anchors) == 1 else None,
                 "response_literal_open": bool(open_literal),
+                "metadata_capacity": metadata_capacity,
                 "native_observation": observation, "source_verified": source_verified},
         })
 
@@ -184,7 +186,8 @@ def context_recovery_candidate(self, request_ref: str) -> bool:
             if turn.get("status") == "completed" and not (turn.get("response") or {}).get("unfinished_work"):
                 from ...chat import _review_openings
                 _, open_literal, _ = _review_openings(str((turn.get("response") or {}).get("message") or ""), 0)
-                if not open_literal:
+                invalid_metadata = ((turn.get("response") or {}).get("gate") or {}).get("kind") == "response_metadata_invalid"
+                if not open_literal and not invalid_metadata:
                     return False  # A lexical probe hint, never permission to adopt or send.
             return self._context_recovery_plan(row, discovery_only=True).get("recovery_candidate") is True
         except (KeyError, ValueError, RuntimeError, OSError):
@@ -230,15 +233,23 @@ def source_requests() -> list[dict[str, Any]]:
             # matches after reacquiring the route fence; fresh authority/source
             # checks and the typed eligibility decision remain below that fence.
             proof = None
+            capacity = None
             if not plan.get("admitted_turn_id") and metadata_review:
-                proof = self._original_native_result(row, session, turn, context, include_raw_response=True)
+                proof = self._original_native_result(row, session, turn, context, include_raw_response=True,
+                    include_retained_result=turn.get("status") == "completed")
+                if proof is not None:
+                    from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL
+                    capacity = AntigravityAdapter.metadata_review_capacity(
+                        cascade_id=session["upstream_thread_id"],
+                        model_id=session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL)
             elif not plan.get("admitted_turn_id") and not preparation_retry:
                 proof = CodexAppServerAdapter.failed_request_observation(
                     codex_bin=self.controller.codex_bin, codex_home=Path(session["codex_home"]),
                     model_auth_home=model_auth_home,
                     **({"read_only_review": True} if plan.get("read_only_review") else {}),
                     **({"completed_work": True, "work_dir": Path(context["project"])} if plan.get("work_attempt") else {}),
                     thread_ids=plan["native_thread_ids"], turn_id=turn["upstream_turn_id"])
+            facts = proof.pop("retained_result", None) if proof is not None and metadata_review else None
             selected = self.bindings.resolve(binding_id=row["binding_id"], **row["source"], include_observation=True)
             source = source_observation(selected)
             scope_available = (self._preparation_scope_available(session)
@@ -265,8 +276,23 @@ def source_requests() -> list[dict[str, Any]]:
                         or source_requests() != requests):
                     return row
                 plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
-                    selected_binding=selected["binding"], preparation_scope_available=scope_available)
+                    selected_binding=selected["binding"], preparation_scope_available=scope_available,
+                    metadata_capacity=capacity)
+                if plan.get("retain_metadata_result") and facts and not self.controller.store.native_result(row["session_id"], turn["turn_id"]):
+                    try:
+                        self.controller.store.retain_native_result(row["session_id"], turn["turn_id"],
+                            {**facts, "upstream_turn_id": turn["upstream_turn_id"]}, completed_turn=turn)
+                    except (OSError, ValueError) as exc:
+                        import logging
+                        logging.getLogger(__name__).warning("Metadata proof retention held (%s)", type(exc).__name__)
+                        return row  # Preserve the delivered answer and unspent review.
+                    proof = self._original_native_result(row, session, turn, context, include_raw_response=True)
+                    plan = self._context_recovery_plan(row, observation=proof, source_verified=source is not None,
+                        selected_binding=selected["binding"], metadata_capacity=capacity)
                 if not plan.get("recovery_eligible"):
+                    if plan.get("metadata_quota_wait") is not None and row.get("metadata_quota_wait") != plan["metadata_quota_wait"]:
+                        row["metadata_quota_wait"] = plan["metadata_quota_wait"]
+                        _atomic_write_json(path, row)
                     return row
                 recovery = row.get("context_recovery")
                 if recovery is None or recovery.get("client_turn_id") != plan["client_turn_id"]:
@@ -343,7 +369,8 @@ def _native_result_text(recovered: dict[str, Any]) -> str:
 
     def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
                                 turn: dict[str, Any], context: dict[str, Any], *,
-                                include_raw_response: bool = False) -> dict[str, Any] | None:
+                                include_raw_response: bool = False,
+                                include_retained_result: bool = False) -> dict[str, Any] | None:
         from .antigravity_adapter import AntigravityAdapter
 
         events = self.controller.store.events_after(session["session_id"], turn["turn_id"], None)
@@ -360,11 +387,13 @@ def _original_native_result(self, row: dict[str, Any], session: dict[str, Any],
             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)
+                input_anchor=anchors[0], include_raw_response=include_raw_response, retained_result=retained,
+                **({"include_retained_result": True} if include_retained_result else {}))
         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] if anchors else None,
+            **({"include_retained_result": True} if include_retained_result else {}),
             **({"include_raw_response": True} if include_raw_response else {}))
 
     def _recover_native_result(self, path: Path, row: dict[str, Any], session: dict[str, Any],

loopx/chat_store.py

@@ -1395,11 +1395,19 @@ def native_result(self, session_id: str, turn_id: str) -> dict[str, Any]:
         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."""
+    def retain_native_result(self, session_id: str, turn_id: str, result: dict[str, Any], *,
+                             completed_turn: dict[str, Any] | None = None) -> None:
+        """Freeze once; a completed backfill requires the exact unchanged Turn.
+
+        The automatic recovery owner first verifies the original native proof
+        and current source. This fence changes neither its Turn nor its events.
+        """
         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":
+            if completed_turn is not None:
+                if turn != completed_turn or turn is None or turn.get("status") != "completed":
+                    raise ValueError("the original completed Turn changed")
+            elif 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")