fix(chat): resume proven unstarted quota failures through the original queue

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -1847,6 +1847,19 @@ card cannot substitute for final-message readback.
 
 ### Recovering a native result after Core failure
 
+A quota-failure notice remains pending work. For Antigravity, the original worker
+may reserve one new quota-continuation Turn only after two independent reads of
+the same idle native trajectory prove its latest invocation contains exactly the
+original input and a structured `QUOTA_EXHAUSTED` error, without model output or
+tool steps. Admission binds the prompt hash, source, execution time, native index,
+current authorization and fresh capacity. Later source input, steering, stop,
+queued input, unknown or cleared steps veto continuation. An unconfirmed cancel
+alone confers no replay eligibility. An enabled flexible endpoint pool may select
+Codex through the existing quota owner; dispatch rechecks current configuration
+and capacity. Keep the failed Turn and its original notice, and adopt the same
+reservation after a crash. The original Bot still delivers and independently
+reads back its normal result in the original audience.
+
 Historical native-result discovery waits while the same Core Session is busy
 or has an active Turn, leaving shared reply workers available for current
 progress. Already verified result metadata keeps its existing correlation owner.

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -1210,6 +1210,15 @@ reaction 清理未完成时，也只恢复原结果 proof，不能再次发送
 
 ### Core 失败后的原生结果恢复
 
+额度耗尽的失败提示仍是待办。对于 Antigravity，只在原 worker 两次独立读回同一
+空闲轨迹，证明最新一次调用恰好是原始输入和结构化 `QUOTA_EXHAUSTED` 错误、
+没有任何模型输出或工具步骤时，现有队列才可预约一次新的 quota 续办 Turn。
+准入绑定原 prompt hash、来源、执行时间、原生索引、当前授权与最新额度；后续
+来源输入、steering、stop、新队列、未知或已清理步骤均阻止续办。取消未确认本身
+不构成重放资格。已启用的灵活端点池可沿既有额度 owner 选择 Codex，派发前再次
+核验当前配置与容量。失败 Turn 和原失败通知保留，崩溃恢复采用同一个预约身份；
+正常结果仍须由原 Bot 沿原受众实际回传并独立读回。
+
 同一 Core Session 忙碌或仍有活跃 Turn 时，历史原生结果暂不进入读回发现，避免
 占用当前进度所需的共享回复 worker。已核验结果的元数据仍由原关联修复路径负责。
 

loopx/canary/module_metric_baseline.json

@@ -36,15 +36,15 @@
     "loopx/chat_runtime.py": {
       "any_count": 37,
       "dict_any_count": 0,
-      "lines": 2395
+      "lines": 2398
     },
     "loopx/chat_server.py": {
       "any_count": 25,
       "dict_any_count": 0,
       "lines": 2039
     },
     "loopx/chat_store.py": {
-      "lines": 2111
+      "lines": 2113
     },
     "loopx/codex_cli_probe.py": {
       "any_count": 37,

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -380,6 +380,70 @@ class AntigravityAdapter:
 
     adapter_kind = "antigravity_ls"
 
+    @staticmethod
+    def quota_failure_observation(*, cascade_id: str, request_ref: str, message: str,
+                                  started_at: str, completed_at: str,
+                                  input_anchor: dict[str, Any]) -> dict[str, Any] | None:
+        """Prove one native input ended at quota before any model/tool output.
+
+        An unconfirmed cancel is not proof of effects or of their absence. Read
+        the exact original input twice from a stable idle cascade instead. Only
+        the latest two-step input/quota-error sequence qualifies; any output,
+        tool, later input, cleared step or unknown native state fails closed.
+        """
+        try:
+            found = discover_antigravity_language_server()
+            start = input_anchor.get("start_step_index")
+            if found is None or input_anchor.get("thread_id") != cascade_id or type(start) is not int or start < 0:
+                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 != start + 2
+                    or (before.get("trajectory") or {}).get("cascadeId") != cascade_id):
+                return None
+            raw = read("GetCascadeTrajectorySteps", startIndex=start, endIndex=total)["steps"]
+            indexed, covered = _observed_turn_steps(raw, start, total)
+            again = read("GetCascadeTrajectorySteps", startIndex=start, endIndex=total)["steps"]
+            after = read("GetCascadeTrajectory")
+            if covered != total or raw != again or before != after or len(indexed) != 2:
+                return None
+            user, terminal = [step for _, step in indexed]
+            if (user.get("type") != "CORTEX_STEP_TYPE_USER_INPUT"
+                    or terminal.get("type") != "CORTEX_STEP_TYPE_ERROR_MESSAGE"
+                    or any(step.get("status") != "CORTEX_STEP_STATUS_DONE" for step in (user, terminal))
+                    or set(user) - {"type", "status", "metadata", "userInput"}
+                    or set(terminal) - {"type", "status", "metadata", "errorMessage"}):
+                return None
+            metadata = user["metadata"]
+            execution = metadata.get("executionId")
+            created = datetime.fromisoformat(metadata["createdAt"])
+            ended = datetime.fromisoformat(terminal["metadata"]["createdAt"])
+            prompt = _native_user_input_text(user)
+            if (not isinstance(execution, str) or not execution
+                    or terminal["metadata"].get("executionId") not in (None, execution)
+                    or not datetime.fromisoformat(started_at) <= created <= ended <= datetime.fromisoformat(completed_at)
+                    or prompt is None or not manager_input_matches(prompt, message=message, request_ref=request_ref)):
+                return None
+            digest = hashlib.sha256(prompt.encode()).hexdigest()
+            if input_anchor.get("input_sha256") != digest:
+                return None
+            error = terminal["errorMessage"]["error"]
+            details = [json.loads(value) for value in error.get("rpcErrorDetails", [])]
+            quota = [value for value in details if value.get("@type") == "type.googleapis.com/google.rpc.ErrorInfo"
+                     and value.get("domain") == "cloudcode-pa.googleapis.com" and value.get("reason") == "QUOTA_EXHAUSTED"]
+            if error.get("errorCode") != 429 or len(quota) != 1 or not quota[0].get("metadata", {}).get("model"):
+                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": digest, "model_id": quota[0]["metadata"]["model"],
+                "pre_execution_quota_failure": True, "only_original_user_item": True,
+                "latest_input_verified": True, "observed_at": time.time()}
+        except (KeyError, TypeError, ValueError, AttributeError, CodexChatAgentError):
+            return None
+
     @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.

loopx/capabilities/native_chat/external_conversations.py

@@ -114,6 +114,10 @@ def _context_recovery_plan(self, row: dict[str, Any], *,
                 "preparation_observation": preparation,
                 "later_equivalent_input_present": later_equivalent,
                 "preparation_scope_available": preparation_scope_available,
+                "later_source_input_present": any(item.get("binding_id") == row["binding_id"]
+                    and item.get("source") == row["source"] and item.get("command") is None
+                    and datetime.fromisoformat(item["created_at"]) > datetime.fromisoformat(row["created_at"])
+                    for item in self.pending()) if turn.get("error_code") == "antigravity_step_error" else None,
                 "native_input": anchors[0] if len(anchors) == 1 else None,
                 "response_literal_open": bool(open_literal),
                 "metadata_capacity": metadata_capacity,
@@ -250,12 +254,20 @@ def source_requests() -> list[dict[str, Any]]:
             # 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,
-                    include_retained_result=turn.get("status") == "completed")
+            if not plan.get("admitted_turn_id") and (metadata_review or turn.get("error_code") == "antigravity_step_error"):
+                from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL
+                events = self.controller.store.events_after(row["session_id"], row["turn_id"], None)
+                anchors = [item["payload"]["native_input"] for item in events
+                    if item.get("kind") == "turn.started" and item.get("payload", {}).get("native_input")]
+                native_session = self._original_native_session(session, anchors)
+                if metadata_review:
+                    proof = self._original_native_result(row, session, turn, context, include_raw_response=True,
+                        include_retained_result=turn.get("status") == "completed")
+                elif len(anchors) == 1:
+                    proof = AntigravityAdapter.quota_failure_observation(cascade_id=native_session["upstream_thread_id"],
+                        request_ref=row["request_ref"], message=row["message"], started_at=turn["started_at"],
+                        completed_at=turn["completed_at"], input_anchor=anchors[0])
                 if proof is not None:
-                    from .antigravity_adapter import AntigravityAdapter, ANTIGRAVITY_DEFAULT_MODEL
-                    native_session = self._original_native_session(session, [proof])
                     capacity = AntigravityAdapter.metadata_review_capacity(
                         cascade_id=native_session["upstream_thread_id"],
                         model_id=native_session.get("agent_model_id") or ANTIGRAVITY_DEFAULT_MODEL)
@@ -342,6 +354,8 @@ def source_requests() -> list[dict[str, Any]]:
                             recovery.update({f"{kind}_attempt": plan[f"{kind}_attempt"],
                                 "presentation_key": f"{MANAGER_CONTEXT_VERSION}.{kind}.{plan[f'{kind}_attempt']}",
                                 "not_before": plan["not_before"]})
+                    if plan.get("quota_executor"):
+                        recovery["quota_executor"] = plan["quota_executor"]
                     if preparation_retry:
                         recovery["preparation_observation"] = plan["preparation_observation"]
                     row["context_recovery"] = recovery

loopx/chat_runtime.py

@@ -832,14 +832,15 @@ def _ensure_adapter_locked(
             recovery = ChatExternalConversations(self).queued_recovery_context(session, active, accepted_turn_id)
         quota_route = None
         metadata_fallback = recovery.get("metadata_review") is True and recovery.get("metadata_executor") == "codex"
+        quota_fallback = bool(recovery.get("quota_attempt") and recovery.get("quota_executor") == "codex")
         if (manager_runtime is not None and active and active_id == accepted_turn_id
                 and active.get("status") == "queued" and not active.get("upstream_turn_id")
-                and active.get("origin") != "native_loopx" and (not recovery or metadata_fallback)
-                and ("-recovery-" not in str(active.get("client_turn_id")) or metadata_fallback)
+                and active.get("origin") != "native_loopx" and (not recovery or metadata_fallback or quota_fallback)
+                and ("-recovery-" not in str(active.get("client_turn_id")) or metadata_fallback or quota_fallback)
                 and session.get("upstream_mode") in {"default", "chat"}):
             quota_route = manager_session_quota_route(self, session)
-            if metadata_fallback and (not quota_route or quota_route["allocation"]["executor_endpoint"] != "codex"):
-                raise CodexChatAgentError("The reserved formatter no longer has verified Codex capacity.", error_code="manager_authorization_unavailable", gate=None)
+            if (metadata_fallback or quota_fallback) and (not quota_route or quota_route["allocation"]["executor_endpoint"] != "codex"):
+                raise CodexChatAgentError("The reserved recovery no longer has verified Codex capacity.", error_code="manager_authorization_unavailable", gate=None)
             if quota_route and quota_route["allocation"]["executor_endpoint"] == session.get("agent_id"):
                 quota_route = None
             if quota_route is not None:
@@ -971,6 +972,7 @@ def check_route_configuration():
                     session.get("manager_context_version") != MANAGER_CONTEXT_VERSION
                     or manager_profile_changed
                     or model_allocation is not None
+                    or quota_fallback  # Do not resume a prior tools-disabled formatter as business execution.
                 )
             )
             legacy_codex_goal_thread = (
@@ -1096,6 +1098,7 @@ def check_route_configuration():
                             agent_model_id=adapter.current_model_id, **observed_settings,
                             manager_quota_block=quota_route["blocked_primary"],
                             metadata_review=metadata_fallback,
+                            quota_recovery=quota_fallback,
                             expected_upstream_thread_id=session["upstream_thread_id"])
                         if selected_agent == "codex":
                             changes["codex_home"] = str(selected_codex_home)

loopx/chat_store.py

@@ -440,6 +440,7 @@ def reserve_loopx_outcome_request(
 
     def update_session(self, session_id: str, *, expected_statuses: set[str] | None = None,
                        expected_upstream_thread_id: str | None = None, metadata_review: bool = False,
+                       quota_recovery: bool = False,
                        **changes: Any) -> dict[str, Any]:
         path = self._session_path(session_id)
         with self._session_lock(session_id):
@@ -502,9 +503,10 @@ def update_session(self, session_id: str, *, expected_statuses: set[str] | None
                             or payload.get("upstream_mode") not in {"default", "chat"}
                             or not active or active.get("status") != "queued" or active.get("upstream_turn_id")
                             or active.get("origin") == "native_loopx"
-                            or ("-recovery-" in str(active.get("client_turn_id")) and not (metadata_review
+                            or ("-recovery-" in str(active.get("client_turn_id")) and not ((metadata_review or quota_recovery)
                                 and payload.get("steward_context") is not None and endpoint == "codex"
-                                and str(active.get("client_turn_id")).endswith("-metadata-1")))
+                                and str(active.get("client_turn_id")).endswith(
+                                    "-metadata-1" if metadata_review else "-quota-1")))
                             or allocation["selection_policy"] != "flexible"
                             or allocation["configured_endpoint"] != "antigravity"
                             or set(allocation["eligible_endpoints"]) != {"antigravity", "codex"}