fix(chat): keep metadata recovery after handled native tool errors

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -591,7 +591,7 @@ def read(method: str, payload: dict[str, Any]) -> dict[str, Any]:
             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 all(_step_allows_completed_turn(step, final=latest[-1]) 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)

tests/test_lark_private_conversations.py

@@ -1375,7 +1375,7 @@ def crash(**kwargs):
 @pytest.mark.parametrize("fault", [None, "codex-fallback", "quota-restore", "quota-unchanged", "quota-unknown",
     "stop", "new-input", "no-anchor", "changed-proof", "unknown-outer", "wrong-target", "changed-brief", "tool-step",
     "cleared-before-proof", "retention-io", "link-inputs", "link-inputs-new-input", "link-inputs-bad-reference",
-    "link-inputs-ungranted", "link-inputs-codex-fallback", "link-inputs-oversized-description"])
+    "link-inputs-ungranted", "link-inputs-codex-fallback", "link-inputs-oversized-description", "settled-tool-error"])
 def test_completed_invalid_metadata_uses_one_original_format_review(tmp_path, language_server, monkeypatch, fault):  # noqa: F811 - imported pytest fixture
     """An already delivered, historical answer must not be sent/executed twice."""
     from copy import deepcopy
@@ -1444,6 +1444,25 @@ def legacy_completion(adapter, message, event_sink, **kwargs):
         # Another native input is historical context, not authority to replay it.
         language_server["inputs"].append({"text": "An unrelated later native input", "created_at": datetime.now(timezone.utc).isoformat()})
         language_server["sent_count"] += 1
+        if fault == "settled-tool-error":
+            from loopx.capabilities.native_chat import antigravity_adapter as native
+            rpc = native._rpc_post
+            language_server["ignore_step_range"] = True
+            def later_handled_tool_error(url, token, method, body, **kwargs):
+                result = rpc(url, token, method, body, **kwargs)
+                if method == "GetCascadeTrajectory":
+                    result["numTotalSteps"] += 1
+                elif method == "GetCascadeTrajectorySteps":
+                    result = deepcopy(result)
+                    for step in result["steps"]:
+                        index = step["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"]
+                        if index >= 3:
+                            step["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"] += 1
+                    later = next(s for s in result["steps"] if s["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"] == 2)
+                    result["steps"].insert(3, {"type": "CORTEX_STEP_TYPE_VIEW_FILE", "status": "CORTEX_STEP_STATUS_ERROR",
+                        "metadata": {**later["metadata"], "sourceTrajectoryStepInfo": {"stepIndex": 3}}})
+                return result
+            monkeypatch.setattr(native, "_rpc_post", later_handled_tool_error)
         if fault in {"quota-restore", "quota-unchanged"} or codex_fallback:
             language_server.update(quota_error_indices=[1], quota_fraction=0.006)
         elif fault == "quota-unknown":
@@ -1537,7 +1556,27 @@ def start(**kwargs):
                 return Formatter()
             monkeypatch.setattr(runtime, "_start_adapter", start)
         path = transport.root / f"{ref}.json"
-        transport.reconcile_request(path)
+        if fault == "settled-tool-error":
+            from loopx.extensions.lark.goal_topic_runtime_service import LarkGoalTopicRuntimeService
+            service = LarkGoalTopicRuntimeService(snapshot_provider=lambda: {}, runtime_root=transport.runtime_root,
+                runtime_controller=runtime, private_conversations=transport)
+            def drive(step, **_):
+                deadline = time.monotonic() + 15
+                try:
+                    while time.monotonic() < deadline:
+                        step()
+                        current = transport.core.read_request(ref)
+                        if (current["turn_id"] != original["turn_id"]
+                                and current.get("context_recovery", {}).get("metadata_review") and current.get("delivery_verified")):
+                            return
+                        time.sleep(.01)
+                    pytest.fail("The existing discovery/reply worker did not complete the original metadata review")
+                finally:
+                    service._closed.set()
+            monkeypatch.setattr(service._delivery_worker, "repeat", drive)
+            service._reconcile_private()
+        else:
+            transport.reconcile_request(path)
         if fault in {"quota-restore", "quota-unchanged"}:
             held = transport.core.read_request(ref)
             assert held["metadata_quota_wait"]["remaining_fraction"] == 0.006
@@ -1571,7 +1610,7 @@ def start(**kwargs):
         reviewed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=15)
         transport.reconcile_request(path)
         entries = pending(runtime.coordination_runtime_root, "work", "worker")["items"]
-        if codex_fallback or fault in {None, "quota-restore", "link-inputs", "link-inputs-bad-reference",
+        if codex_fallback or fault in {None, "quota-restore", "link-inputs", "link-inputs-bad-reference", "settled-tool-error",
                                      "link-inputs-oversized-description"}:
             assert reviewed["status"] == "completed", reviewed
             assert len(entries) == 1 and entries[0]["source_id"] == ref

tests/test_steward_interrupted_answer.py

@@ -16,6 +16,45 @@
 from loopx.extensions.lark.conversation_identity import identity_ref
 
 
+@pytest.mark.parametrize("fault, expected", [
+    ("done", True), ("error", True), ("canceled", True), ("cleared", True),
+    ("running", False), ("generating", False), ("waiting", False),
+    ("missing-status", False), ("visible-error", False),
+    ("mixed-execution", False), ("final-error", False), ("missing-stop", False),
+])
+def test_capacity_readback_uses_the_completed_native_turn_boundary(language_server, fault, expected):
+    """Handled tool failures do not hide quota; unknown/active work stays held."""
+    meta = {"createdAt": "2026-01-01T00:00:05Z", "executionId": "completed-execution"}
+    steps = [
+        {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta},
+        {"type": "CORTEX_STEP_TYPE_VIEW_FILE", "status": "CORTEX_STEP_STATUS_" + fault.upper(), "metadata": meta},
+        {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE", "status": "CORTEX_STEP_STATUS_DONE", "metadata": meta,
+         "plannerResponse": {"response": "Completed after the tool attempt.", "stopReason": "STOP_REASON_STOP_PATTERN"}},
+    ]
+    if fault == "missing-status":
+        steps[1].pop("status")
+    elif fault == "visible-error":
+        steps[1].update(type="CORTEX_STEP_TYPE_ERROR_MESSAGE", status="CORTEX_STEP_STATUS_DONE",
+                        errorMessage={"shouldShowUser": True})
+    elif fault == "mixed-execution":
+        steps[1].update(status="CORTEX_STEP_STATUS_DONE", metadata={**meta, "executionId": "another-execution"})
+    elif fault == "final-error":
+        steps[1]["status"] = steps[2]["status"] = "CORTEX_STEP_STATUS_ERROR"
+    elif fault == "missing-stop":
+        steps[1]["status"] = "CORTEX_STEP_STATUS_DONE"
+        steps[2]["plannerResponse"].pop("stopReason")
+    language_server.update(active=True, frames=[], frame={"at": 1, "steps": steps, "status": "CASCADE_RUN_STATUS_IDLE"})
+    capacity = native.AntigravityAdapter.metadata_review_capacity(cascade_id="synthetic-cascade", model_id="gemini-4-argon-high")
+    if expected:
+        assert capacity and capacity["normal_completion"] is True and capacity["remaining_fraction"] == 1
+        assert capacity["quota_exhausted"] is False
+    else:
+        assert capacity is None
+    assert {method for method, _ in language_server["calls"]} <= {
+        "GetCascadeModelConfigData", "GetCascadeTrajectory", "GetCascadeTrajectorySteps", "GetUserStatus"}
+    assert language_server["sent_count"] == 0 and not language_server["cancelled"].is_set()
+
+
 @pytest.mark.parametrize("prior_status, error_status, cancellation_needed", [
     ("CORTEX_STEP_STATUS_DONE", "CORTEX_STEP_STATUS_DONE", False),
     ("CORTEX_STEP_STATUS_ERROR", "CORTEX_STEP_STATUS_DONE", False),