diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py
index 3a5714a026..bf27fff59a 100644
--- a/loopx/capabilities/native_chat/antigravity_adapter.py
+++ b/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()
diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py
index 76dc0d81c7..d5a6a58821 100644
--- a/loopx/chat_runtime.py
+++ b/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
diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts
index 35a7a8fe27..9a2ca48744 100644
--- a/loopx/control_plane/collaboration/conversation_binding.ts
+++ b/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};
}
diff --git a/tests/control_plane_ts/conversation_binding.test.ts b/tests/control_plane_ts/conversation_binding.test.ts
index 1352baeb04..ab6f3d206d 100644
--- a/tests/control_plane_ts/conversation_binding.test.ts
+++ b/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",
diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py
index 995840b2e9..193a01dbcd 100644
--- a/tests/test_antigravity_steward_adapter.py
+++ b/tests/test_antigravity_steward_adapter.py
@@ -5,6 +5,8 @@
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import threading
+import time
+from types import SimpleNamespace
import pytest
from typing import Any
@@ -115,10 +117,21 @@ def do_POST(self):
elif method == "StartCascade":
response = {"cascadeId": "synthetic-cascade"}
elif method == "GetCascadeTrajectory":
- response = {"trajectory": {"cascadeId": "synthetic-cascade"},
- "numTotalSteps": state["sent_count"] * 2,
- "status": ("CASCADE_RUN_STATUS_RUNNING" if state["mode"] in {"running", "waiting"}
- and state["active"] else "CASCADE_RUN_STATUS_IDLE")}
+ if state.get("frames") is not None:
+ if state["active"]:
+ if state["frames"]:
+ state["frame"] = state["frames"].pop(0)
+ frame = state["frame"]
+ state["now"] = frame["at"]
+ else:
+ frame = {"steps": [], "status": "CASCADE_RUN_STATUS_IDLE"}
+ response = {"trajectory": {"cascadeId": "synthetic-cascade"},
+ "numTotalSteps": frame.get("total", len(frame["steps"])), "status": frame["status"]}
+ else:
+ response = {"trajectory": {"cascadeId": "synthetic-cascade"},
+ "numTotalSteps": state["sent_count"] * 2,
+ "status": ("CASCADE_RUN_STATUS_RUNNING" if state["mode"] in {"running", "waiting"}
+ and state["active"] else "CASCADE_RUN_STATUS_IDLE")}
if state.get("missing_resume"):
response.pop("trajectory")
elif method == "SendUserCascadeMessage":
@@ -132,15 +145,21 @@ def do_POST(self):
state["sent"].set()
response = {}
elif method == "GetCascadeTrajectorySteps":
- response = {"steps": []}
- for index in range(body["startIndex"] // 2, state["sent_count"]):
+ response = {"steps": state["frame"]["steps"] if state.get("frames") is not None else []}
+ start = 0 if state.get("ignore_step_range") else body["startIndex"] // 2
+ for index in (() if state.get("frames") is not None else range(start, state["sent_count"])):
+ answer = f"Argon answer {index + 1}"
+ if not state.get("plain_text"):
+ answer = '' + json.dumps({"message": answer,
+ "proposals": [], "context_handoff": [], "gate": None}) + ''
response["steps"].extend([
{"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE"},
{"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE",
"status": "CORTEX_STEP_STATUS_WAITING" if state["mode"] == "waiting" else "CORTEX_STEP_STATUS_DONE",
- "plannerResponse": {"thinking": "synthetic status",
- "response": '' + json.dumps({"message": f"Argon answer {index + 1}",
- "proposals": [], "context_handoff": [], "gate": None}) + ''}}])
+ "plannerResponse": {"thinking": "synthetic status", "response": answer}}])
+ if state.get("native_indices"):
+ for offset, step in enumerate(response["steps"]):
+ step["metadata"] = {"sourceTrajectoryStepInfo": {"stepIndex": offset}}
elif method == "CancelCascadeInvocation":
state["active"] = False
state["cancelled"].set()
@@ -237,6 +256,7 @@ def test_timeout_with_partial_response_never_reports_success(tmp_path, language_
assert error.value.error_code == "response_timeout"
assert language_server["cancelled"].is_set()
assert not any(kind == "answer.final" for kind, _ in events)
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
def test_pending_native_approval_is_actionable_and_not_auto_accepted(tmp_path, language_server):
@@ -308,7 +328,7 @@ def test_deadline_expiry_after_partial_text_fails_closed(tmp_path, language_serv
from loopx.capabilities.native_chat import antigravity_adapter as provider
from loopx.chat_agent import CodexChatAgentError
language_server["mode"] = "running"
- adapter = adapter_for(tmp_path)
+ adapter = adapter_for(tmp_path, hard_timeout_sec=300.0)
ticks = iter([0.0, 0.0, 301.0])
monkeypatch.setattr(provider.time, "monotonic", lambda: next(ticks, 301.0))
with pytest.raises(CodexChatAgentError) as error:
@@ -326,3 +346,127 @@ def test_busy_native_conversation_is_not_sent_to_or_cancelled(tmp_path, language
assert error.value.error_code == "resume_failed"
assert not any(method in {"SendUserCascadeMessage", "CancelCascadeInvocation"}
for method, _ in language_server["calls"])
+
+
+@pytest.mark.parametrize("native_indices", [False, True])
+def test_native_full_history_is_not_streamed_as_the_current_answer(tmp_path, language_server, native_indices):
+ language_server["plain_text"] = True
+ adapter = adapter_for(tmp_path)
+ assert adapter.start_turn("first")["message"] == "Argon answer 1"
+ language_server["ignore_step_range"] = True
+ language_server["native_indices"] = native_indices
+ events = []
+ result = adapter.start_turn("second", lambda kind, data: events.append((kind, data)))
+ assert result["message"] == "Argon answer 2"
+ assert "Argon answer 1" not in "".join(data.get("text", "") for kind, data in events if kind == "answer.delta")
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 2
+
+
+def _native_frames(*, stopped: str = "STOP_REASON_STOP_PATTERN", stagnant: bool = False):
+ frames = []
+ for index in range(5):
+ final = index == 4 and not stagnant
+ steps = [
+ {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
+ "metadata": {"sourceTrajectoryStepInfo": {}}},
+ {"type": "CORTEX_STEP_TYPE_PLANNER_RESPONSE",
+ "status": "CORTEX_STEP_STATUS_DONE" if final else "CORTEX_STEP_STATUS_GENERATING",
+ "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 1}},
+ "plannerResponse": {"response": "native final" if final else ("same partial" if stagnant else f"progress {index}"),
+ "thinkingDuration": index, **({"stopReason": stopped} if final else {})}},
+ ]
+ frames.append({"at": 80.0 * (index + 1), "steps": steps,
+ "status": "CASCADE_RUN_STATUS_IDLE" if final else "CASCADE_RUN_STATUS_RUNNING"})
+ return frames
+
+
+def _native_clock(monkeypatch, language_server, frames):
+ from loopx.capabilities.native_chat import antigravity_adapter as provider
+ language_server.update(frames=frames, now=0.0)
+ # Keep HTTP/socket/thread clocks real. Only the provider's invocation clock
+ # advances, in response to completed synthetic native HTTP observations.
+ monkeypatch.setattr(provider, "time", SimpleNamespace(monotonic=lambda: language_server["now"], sleep=time.sleep))
+
+
+def test_default_core_keeps_an_original_progressing_invocation_past_five_minutes(tmp_path, language_server, monkeypatch):
+ from loopx.chat_runtime import ChatRuntimeController
+ from loopx.chat_store import ChatSessionStore
+ _native_clock(monkeypatch, language_server, _native_frames())
+ store = ChatSessionStore(tmp_path / "runtime" / "chat")
+ controller = ChatRuntimeController(store=store, codex_bin="unused-codex")
+ session, _ = controller.open_session(goal_id="synthetic-goal", agent_id="antigravity", work_dir=tmp_path,
+ objective="Synthetic conversation", mode="new", channel_id="synthetic-argon-chat")
+ sid = session["session_id"]
+ try:
+ turn, _ = controller.submit_turn(session_id=sid, client_turn_id="one-original-input", message="long task",
+ work_dir=tmp_path, objective="Synthetic conversation")
+ result = controller.wait_for_turn(session_id=sid, turn_id=turn["turn_id"], timeout_sec=8)
+ assert result["status"] == "completed"
+ assert result["response"]["message"] == "native final"
+ assert language_server["now"] == 400.0
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+ assert not language_server["cancelled"].is_set()
+ assert [m["text"] for m in store.messages(sid) if m["role"] == "agent"] == ["native final"]
+ finally:
+ controller.close_session(sid)
+
+
+@pytest.mark.parametrize("hard_limit,stagnant", [(300.0, False), (None, True)])
+def test_explicit_deadline_and_default_no_progress_timeout_still_stop(tmp_path, language_server, monkeypatch, hard_limit, stagnant):
+ from loopx.chat_agent import CodexChatAgentError
+ _native_clock(monkeypatch, language_server, _native_frames(stagnant=stagnant))
+ events = []
+ with pytest.raises(CodexChatAgentError) as error:
+ options = {"hard_timeout_sec": hard_limit} if hard_limit is not None else {}
+ adapter_for(tmp_path, **options).start_turn("long task", lambda kind, data: events.append((kind, data)))
+ assert error.value.error_code == "response_timeout"
+ assert language_server["cancelled"].is_set()
+ assert not any(kind == "answer.final" for kind, _ in events)
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+
+
+@pytest.mark.parametrize("value", [0, -1, float("nan"), float("inf"), True])
+def test_invalid_hard_timeout_creates_no_native_invocation(tmp_path, language_server, value):
+ with pytest.raises(ValueError, match="finite and positive"):
+ adapter_for(tmp_path, hard_timeout_sec=value)
+ assert language_server["calls"] == []
+
+
+@pytest.mark.parametrize("reason", ["STOP_REASON_CLIENT_STREAM_ERROR", "unrecognized-native-stop"])
+def test_idle_done_with_an_unverified_native_stop_never_becomes_success(tmp_path, language_server, monkeypatch, reason):
+ from loopx.chat_agent import CodexChatAgentError
+ frames = _native_frames(stopped=reason)
+ frames[-1]["at"] = 1.0
+ _native_clock(monkeypatch, language_server, [frames[-1]])
+ events = []
+ with pytest.raises(CodexChatAgentError) as error:
+ adapter_for(tmp_path).start_turn("task", lambda kind, data: events.append((kind, data)))
+ assert error.value.error_code == "upstream_transport_failure"
+ assert not any(kind == "answer.final" for kind, _ in events)
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
+
+
+def test_partial_native_inventory_is_not_a_complete_response(tmp_path, language_server, monkeypatch):
+ from loopx.chat_agent import CodexChatAgentError
+ frame = _native_frames()[-1]
+ frame.update(at=1.0, total=4)
+ _native_clock(monkeypatch, language_server, [frame])
+ with pytest.raises(CodexChatAgentError) as error:
+ adapter_for(tmp_path).start_turn("task")
+ assert error.value.error_code == "antigravity_rpc_error"
+ assert language_server["cancelled"].is_set()
+
+
+def test_another_native_input_is_neither_adopted_nor_cancelled(tmp_path, language_server, monkeypatch):
+ from loopx.chat_agent import CodexChatAgentError
+ frame = _native_frames()[-1]
+ frame["at"] = 1.0
+ frame["steps"].insert(1, {"type": "CORTEX_STEP_TYPE_USER_INPUT", "status": "CORTEX_STEP_STATUS_DONE",
+ "metadata": {"sourceTrajectoryStepInfo": {"stepIndex": 1}}})
+ frame["steps"][-1]["metadata"]["sourceTrajectoryStepInfo"]["stepIndex"] = 2
+ _native_clock(monkeypatch, language_server, [frame])
+ with pytest.raises(CodexChatAgentError) as error:
+ adapter_for(tmp_path).start_turn("original task")
+ assert error.value.error_code == "upstream_transport_failure"
+ assert not language_server["cancelled"].is_set()
+ assert len([1 for method, _ in language_server["calls"] if method == "SendUserCascadeMessage"]) == 1
diff --git a/tests/test_lark_private_conversations.py b/tests/test_lark_private_conversations.py
index 576e0615a4..42a24fc533 100644
--- a/tests/test_lark_private_conversations.py
+++ b/tests/test_lark_private_conversations.py
@@ -276,7 +276,8 @@ def test_delivered_quota_failure_stays_pending_across_restart_without_replaying(
@pytest.mark.parametrize("command", ["stop", "new"])
-def test_delivered_timeout_stays_pending_without_replaying_and_honors_stop(ordinary, monkeypatch, command): # noqa: F811
+@pytest.mark.parametrize("error_code", ["response_timeout", "upstream_transport_failure"])
+def test_delivered_timeout_stays_pending_without_replaying_and_honors_stop(ordinary, monkeypatch, command, error_code): # noqa: F811
from loopx.chat_agent import CodexChatAgentError
from loopx.capabilities.native_chat.codex_adapter import CodexAppServerAdapter
@@ -285,7 +286,7 @@ def test_delivered_timeout_stays_pending_without_replaying_and_honors_stop(ordin
def timeout_after_start(_adapter, message, event_sink):
starts.append(message)
event_sink("turn.started", {"upstream_turn_id": "synthetic-timeout"})
- raise CodexChatAgentError("Synthetic upstream timeout", gate=None, error_code="response_timeout")
+ raise CodexChatAgentError("Synthetic upstream failure", gate=None, error_code=error_code)
monkeypatch.setattr(CodexAppServerAdapter, "start_turn", timeout_after_start)
store, runtime, provider, transport = connect(ordinary)
@@ -293,7 +294,7 @@ def timeout_after_start(_adapter, message, event_sink):
transport.admit("notes-app", provider.event("notes-app", "timeout-pending", "Complete this request."))
row = transport.core.pending()[0]
failed = runtime.wait_for_turn(session_id=row["session_id"], turn_id=row["turn_id"], timeout_sec=10)
- assert failed["status"] == "failed" and failed["error_code"] == "response_timeout"
+ assert failed["status"] == "failed" and failed["error_code"] == error_code
transport.reconcile()
saved = json.loads((transport.root / f"{row['request_ref']}.json").read_text())
assert saved["status"] == "delivered" and saved["deliveries"]["terminal"]["verified"]