fix(chat): follow native Antigravity progress and retain failed requests

loopx/capabilities/native_chat/antigravity_adapter.py

@@ -4,6 +4,7 @@
 
 import ipaddress
 import json
+import math
 import os
 from pathlib import Path
 import re
@@ -26,6 +27,7 @@
 
 ANTIGRAVITY_DEFAULT_MODEL = "gemini-4-argon-high"
 ANTIGRAVITY_DEFAULT_REASONING_EFFORT = "high"
+ANTIGRAVITY_IDLE_TIMEOUT_SEC = 300.0
 
 ANTIGRAVITY_MODEL_PLACEHOLDERS: dict[str, str] = {
     "gemini-4-argon-high": "MODEL_PLACEHOLDER_M417",
@@ -45,6 +47,38 @@
 )
 
 
+def _observed_turn_steps(steps: list[Any], start: int, total: int) -> tuple[list[tuple[int, dict[str, Any]]], int]:
+    """Read absolute native indices; some LS versions ignore range arguments."""
+    bases: set[int] = set()
+    for offset, step in enumerate(steps):
+        if not isinstance(step, dict):
+            raise ValueError("Antigravity returned an invalid trajectory step")
+        metadata = step.get("metadata") or {}
+        source = metadata.get("sourceTrajectoryStepInfo") or {}
+        index = source.get("stepIndex")
+        if index is not None:
+            if type(index) is not int or index < 0:
+                raise ValueError("Antigravity returned an invalid trajectory index")
+            bases.add(index - offset)
+    if len(bases) == 1:
+        base = bases.pop()
+    elif not bases and len(steps) == total - start:
+        base = start  # Older servers honor the requested range without metadata.
+    elif not bases and len(steps) == total:
+        base = 0
+    else:
+        raise ValueError("Antigravity trajectory coverage is unknown")
+    current = [(base + offset, step) for offset, step in enumerate(steps) if start <= base + offset < total]
+    if base < 0 or [index for index, _ in current] != list(range(start, total)):
+        raise ValueError("Antigravity trajectory coverage is incomplete")
+    return current, base + len(steps)
+
+
+def _check_hard_timeout(value: float | None) -> None:
+    if value is not None and (type(value) not in (int, float) or not math.isfinite(value) or value <= 0):
+        raise ValueError("Antigravity hard timeout must be finite and positive")
+
+
 class _LocalRedirectHandler(urllib.request.HTTPRedirectHandler):
     def redirect_request(self, req: Any, fp: Any, code: int, msg: str,
                          headers: Any, newurl: str) -> None:
@@ -261,6 +295,8 @@ class AntigravityAdapter:
     Restricted Chat disables native tools. Trusted-owner uses Antigravity's
     existing approvals; LoopX never accepts those approvals on the user's behalf.
     Governed Tasks and workspace grants retain their existing executor owner.
+    By default, five minutes without native progress stops an invocation; an
+    explicit hard_timeout_sec also bounds wall time even while it progresses.
     """
 
     adapter_kind = "antigravity_ls"
@@ -280,9 +316,10 @@ def __init__(
         team_plan_context: dict[str, Any] | None = None,
         initial_history: list[dict[str, str]] | None = None,
         resumed: bool = False,
-        hard_timeout_sec: float = 300.0,
+        hard_timeout_sec: float | None = None,
     ) -> None:
         self.work_dir = work_dir.resolve()
+        _check_hard_timeout(hard_timeout_sec)
         self.objective = objective
         self.cascade_id = cascade_id
         self.upstream_thread_id = cascade_id
@@ -327,10 +364,11 @@ def start(
         team_plan_context: dict[str, Any] | None = None,
         executor_model: Mapping[str, Any] | None = None,
         env: Mapping[str, str] | None = None,
-        hard_timeout_sec: float = 300.0,
+        hard_timeout_sec: float | None = None,
     ) -> "AntigravityAdapter":
         if runtime_profile not in {"restricted", "trusted_owner"}:
             raise ValueError("unsupported Antigravity Chat runtime profile")
+        _check_hard_timeout(hard_timeout_sec)
         # Validate user intent before creating an upstream cascade.
         requested = dict(executor_model or {})
         resolve_antigravity_model_placeholder(requested.get("model"), requested.get("reasoning_effort"))
@@ -526,10 +564,16 @@ def _run_turn(
         emitted_text_by_step: dict[int, str] = {}
         announced_phases: set[tuple[int, str]] = set()
         latest_response_text = ""
-        deadline = time.monotonic() + self.hard_timeout_sec
+        last_progress_at = time.monotonic()
+        deadline = last_progress_at + self.hard_timeout_sec if self.hard_timeout_sec is not None else None
+        previous_progress: Any = None
         saw_running_or_new_step = False
 
-        while time.monotonic() < deadline:
+        while True:
+            now = time.monotonic()
+            if (deadline is not None and now >= deadline) or now - last_progress_at >= ANTIGRAVITY_IDLE_TIMEOUT_SEC:
+                raise CodexChatAgentError("Antigravity did not finish before the turn deadline.",
+                                          gate=None, error_code="response_timeout")
             self._check_interrupted()
             self._interrupted.wait(0.25)
             self._check_interrupted()
@@ -545,6 +589,8 @@ def _run_turn(
             if status == "CASCADE_RUN_STATUS_RUNNING" or total_steps > start_step_index:
                 saw_running_or_new_step = True
 
+            current_steps: list[tuple[int, dict[str, Any]]] = []
+            observed_total = total_steps
             if total_steps > start_step_index:
                 self._check_interrupted()
                 steps_resp = _rpc_post(
@@ -558,12 +604,21 @@ def _run_turn(
                     },
                     timeout=2.0,
                 )
-                steps = steps_resp.get("steps") or []
-                if isinstance(steps, list):
-                    for offset, step in enumerate(steps):
-                        if not isinstance(step, dict):
-                            continue
-                        step_idx = start_step_index + offset
+                steps = steps_resp.get("steps")
+                try:
+                    if not isinstance(steps, list):
+                        raise ValueError("Antigravity returned no trajectory coverage")
+                    current_steps, observed_total = _observed_turn_steps(steps, start_step_index, total_steps)
+                except (AttributeError, TypeError, ValueError) as exc:
+                    raise CodexChatAgentError(str(exc), gate=None, error_code="antigravity_rpc_error") from exc
+                if current_steps:
+                    if sum(step.get("type") == "CORTEX_STEP_TYPE_USER_INPUT" for _, step in current_steps) > 1:
+                        # Another native input is not this invocation's result or
+                        # cancellation target. Preserve it for its current owner.
+                        self._active_turn_id = None
+                        raise CodexChatAgentError("The native Antigravity input changed; no result was adopted or invocation cancelled.",
+                                                  gate=None, error_code="upstream_transport_failure")
+                    for step_idx, step in current_steps:
                         step_type = str(step.get("type") or "")
                         step_status = str(step.get("status") or "")
                         if step_status == "CORTEX_STEP_STATUS_WAITING":
@@ -631,12 +686,33 @@ def _run_turn(
                                 error_code="antigravity_step_error",
                             )
 
+            # A running status alone is not progress. Observe only this input's
+            # steps, including streamed text and command output; old history or
+            # a stuck RUNNING flag cannot keep an invocation alive indefinitely.
+            progress = [(index, step.get("status"),
+                         (step.get("plannerResponse") or {}).get("modifiedResponse")
+                         or (step.get("plannerResponse") or {}).get("response"),
+                         (step.get("plannerResponse") or {}).get("thinking"),
+                         (step.get("runCommand") or {}).get("combinedOutput"),
+                         (step.get("runCommand") or {}).get("exitCode")) for index, step in current_steps]
+            now = time.monotonic()
+            if progress and progress != previous_progress:
+                previous_progress = progress
+                last_progress_at = now
             self._check_interrupted()
-            if saw_running_or_new_step and status == "CASCADE_RUN_STATUS_IDLE":
+            if deadline is not None and now >= deadline:
+                raise CodexChatAgentError("Antigravity did not finish before the turn deadline.",
+                                          gate=None, error_code="response_timeout")
+            if saw_running_or_new_step and status == "CASCADE_RUN_STATUS_IDLE" and observed_total == total_steps:
+                final = current_steps[-1][1] if current_steps else {}
+                planner = final.get("plannerResponse") or {}
+                if (final.get("type") != "CORTEX_STEP_TYPE_PLANNER_RESPONSE"
+                        or final.get("status") != "CORTEX_STEP_STATUS_DONE" or planner.get("toolCalls")
+                        or not str(planner.get("modifiedResponse") or planner.get("response") or "").strip()
+                        or planner.get("stopReason") not in {None, "STOP_REASON_STOP_PATTERN"}):
+                    raise CodexChatAgentError("Antigravity ended without a verified final response; execution may have occurred.",
+                                              gate=None, error_code="upstream_transport_failure")
                 break
-        else:
-            raise CodexChatAgentError("Antigravity did not finish before the turn deadline.",
-                                      gate=None, error_code="response_timeout")
 
         if event_sink is not None:
             tail = stream_filter.finish()

loopx/chat_runtime.py

@@ -398,7 +398,7 @@ def _start_adapter(
                 resume_thread_id=resume_thread_id,
                 history=history,
                 executor_model=model_config,
-                hard_timeout_sec=self.hard_timeout_sec or 300.0,
+                hard_timeout_sec=self.hard_timeout_sec,
             )
         if agent_id == "codex":
             from .capabilities.manager_context.inspection import READ_TOOL, CONTEXT_READ_TOOL

loopx/control_plane/collaboration/conversation_binding.ts

@@ -444,7 +444,7 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
     const turn = observation.turn == null ? null : requireJsonObject(observation.turn, "observed Turn");
     // Pending work is not current retry eligibility: backoff, disabled bindings,
     // exhausted attempts and unavailable providers do not retire the request.
-    // A delivered timeout notice is not a business result. Its effects can
+    // A delivered transport failure is not a business result. Its effects can
     // remain unknown, so visibility must not grant another execution.
     // A delivered partial answer is not completion either. Keep explicit
     // unfinished work and a correlated quota continuation's unresolved gate
@@ -466,7 +466,8 @@ export function planBoundConversationRequest(params: JsonObject): JsonObject {
     return {pending: row.command === null && row.status === "accepted"
       && turn?.session_id === row.session_id && turn?.turn_id === row.turn_id
       && (unfinished || turn?.status === "failed"
-        && (turn.error_code === "usage_limit_exceeded" || turn.error_code === "response_timeout"))
+        && ["usage_limit_exceeded", "response_timeout", "upstream_transport_failure",
+          "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error"].includes(String(turn.error_code)))
       && !(current?.session_id === row.session_id && current?.status === "closed")
       && observation.stop_observed !== true};
   }

tests/control_plane_ts/conversation_binding.test.ts

@@ -764,10 +764,12 @@ test("quota backlog projection retires explicit lifecycle stops, not unavailable
   }
 });
 
-test("delivered timeout notices retain the exact failed request, including other executors", () => {
+for (const errorCode of ["response_timeout", "upstream_transport_failure",
+  "antigravity_rpc_error", "antigravity_transport_error", "antigravity_step_error"]) {
+test(`delivered ${errorCode} notices retain the exact failed request, including other executors`, () => {
   const request = {request_ref: "f".repeat(24), command: null, status: "accepted",
     session_id: "original", turn_id: "timed-out", delivery_verified: true};
-  const turn = {session_id: "original", turn_id: "timed-out", status: "failed", error_code: "response_timeout"};
+  const turn = {session_id: "original", turn_id: "timed-out", status: "failed", error_code: errorCode};
   const packet = {request, current_session: {session_id: "original", status: "ready", agent_id: "antigravity"},
     pending_observation: {turn, stop_observed: false}};
   assert.deepEqual(planBoundConversationRequest(packet), {pending: true});
@@ -783,6 +785,7 @@ test("delivered timeout notices retain the exact failed request, including other
   }
   assert.equal(planBoundConversationRequest({...packet, request: {...request, command: "stop"}}).pending, false);
 });
+}
 
 test("delivered partial results remain pending without authorizing another execution", () => {
   const original = {request_ref: "f".repeat(24), command: null, status: "accepted",