diff --git a/loopx/capabilities/native_chat/antigravity_adapter.py b/loopx/capabilities/native_chat/antigravity_adapter.py index 6bf7428478..133c8b3ab5 100644 --- a/loopx/capabilities/native_chat/antigravity_adapter.py +++ b/loopx/capabilities/native_chat/antigravity_adapter.py @@ -149,8 +149,13 @@ def _rpc_post( def discover_antigravity_language_server( env: Mapping[str, str] | None = None, + *, deadline: float | None = None, ) -> dict[str, Any] | None: - """Discover a running local Antigravity language_server endpoint and CSRF token.""" + """Discover a running local endpoint without reusing an availability verdict. + + Readiness can bound the whole observation, including process listing. An + execution caller without that deadline retains the existing phase limits. + """ source_env = os.environ if env is None else env explicit_url = str(source_env.get("ANTIGRAVITY_LS_URL") or "").strip() explicit_token = str(source_env.get("ANTIGRAVITY_CSRF_TOKEN") or "").strip() @@ -159,12 +164,15 @@ def discover_antigravity_language_server( return None try: explicit_url = _local_url(explicit_url) + timeout = 2.5 if deadline is None else min(2.5, deadline - time.monotonic()) + if timeout <= 0: + return None _rpc_post( explicit_url, explicit_token, "GetCascadeModelConfigData", {}, - timeout=2.5, + timeout=timeout, ) return { "url": explicit_url.rstrip("/"), @@ -175,6 +183,9 @@ def discover_antigravity_language_server( except (CodexChatAgentError, ValueError): return None + process_timeout = 3.0 if deadline is None else min(3.0, deadline - time.monotonic()) + if process_timeout <= 0: + return None try: ps_out = subprocess.check_output( ["ps", "-u", str(os.getuid()), "-o", "pid=,command="], @@ -182,7 +193,7 @@ def discover_antigravity_language_server( encoding="utf-8", errors="replace", stderr=subprocess.DEVNULL, - timeout=3.0, + timeout=process_timeout, ) except Exception: return None @@ -201,7 +212,8 @@ def discover_antigravity_language_server( if token_match: candidates.append((pid, token_match.group(1))) - deadline = time.monotonic() + 4.0 + discovery_deadline = time.monotonic() + 4.0 + deadline = min(discovery_deadline, deadline) if deadline is not None else discovery_deadline for pid, token in candidates: if time.monotonic() >= deadline: break @@ -243,7 +255,9 @@ def probe_antigravity_endpoint( env: Mapping[str, str] | None = None, ) -> tuple[bool, str, str]: """Return `(available, command_display, detail)` for `/api/chat/capabilities`.""" - discovered = discover_antigravity_language_server(env=env) + # The four-second discovery allowance covers this whole advisory read, + # rather than beginning again after a slow process-listing phase. + discovered = discover_antigravity_language_server(env=env, deadline=time.monotonic() + 4.0) if discovered is not None: return ( True, diff --git a/loopx/chat_manager.py b/loopx/chat_manager.py index 548a899377..fe42e6071c 100644 --- a/loopx/chat_manager.py +++ b/loopx/chat_manager.py @@ -309,7 +309,10 @@ def operator_credential_pair(controller: Any) -> dict[str, str]: return operator_provider_host_credential(runtime_root) -def manager_capabilities_projection(controller: Any, store: Any) -> dict[str, Any]: +def manager_capabilities_projection( + controller: Any, store: Any, *, + endpoint_capabilities: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: """Compose the steward section of the shared chat capabilities payload. Every steward entry point resolves the same three things -- this machine's @@ -328,6 +331,7 @@ def manager_capabilities_projection(controller: Any, store: Any) -> dict[str, An controller, machine_defaults=machine_defaults, environ=credential["environ"], + endpoint_capabilities=endpoint_capabilities, ) return manager_runtime_capability_projection( controller, @@ -397,14 +401,18 @@ def _manager_eligible_endpoints( return [str(item) for item in raw if str(item)] -def _controller_endpoint_availability(controller: Any) -> dict[str, bool]: - resolver = getattr(controller, "capabilities", None) - if not callable(resolver): - return {} - try: - rows = resolver() - except Exception: # noqa: BLE001 - availability readback must remain advisory. - return {} +def _controller_endpoint_availability( + controller: Any, *, endpoint_capabilities: list[dict[str, Any]] | None = None, +) -> dict[str, bool]: + rows = endpoint_capabilities + if rows is None: + resolver = getattr(controller, "capabilities", None) + if not callable(resolver): + return {} + try: + rows = resolver() + except Exception: # noqa: BLE001 - availability readback must remain advisory. + return {} if not isinstance(rows, list): return {} return { @@ -432,6 +440,7 @@ def manager_executor_allocation( *, machine_defaults: Mapping[str, Any] | None = None, environ: dict[str, str] | None = None, + endpoint_capabilities: list[dict[str, Any]] | None = None, ) -> dict[str, Any]: """Resolve one bounded steward executor choice and explain why it won. @@ -452,7 +461,9 @@ def manager_executor_allocation( ) policy = _manager_selection_policy(defaults) eligible = _manager_eligible_endpoints(defaults) - availability = _controller_endpoint_availability(controller) + availability = _controller_endpoint_availability( + controller, endpoint_capabilities=endpoint_capabilities, + ) requested = str(requested_endpoint or "").strip() reason = "" diff --git a/loopx/chat_server.py b/loopx/chat_server.py index d6ec5d89c1..c8bb089721 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -1660,14 +1660,14 @@ def do_GET(self) -> None: "state": "ready", "runtime_identity": self.server.runtime_identity}) return if path == CHAT_CAPABILITIES_PATH: - # The steward section is composed by its own owner so the model - # arguments, the availability verdict and the readback cannot - # disagree about which executor and credential they describe. + # The channel and adapter rows share one fresh endpoint observation; + # separate requests and execution retain their own discovery. + adapters = self.server.runtime_controller.capabilities() capabilities = { "ok": True, "schema_version": "loopx_chat_capabilities_v1", "manager": manager_capabilities_projection( - self.server.runtime_controller, self.server.chat_store + self.server.runtime_controller, self.server.chat_store, endpoint_capabilities=adapters, ), "runtime_identity": self.server.runtime_identity, "agent_backend": "multi_adapter", @@ -1681,7 +1681,7 @@ def do_GET(self) -> None: "typed_actions": True, "attached_session_broker": True, "action_kinds": sorted(ACTION_KINDS), - "adapters": self.server.runtime_controller.capabilities(), + "adapters": adapters, "lark_cli": self.server.lark_cli_resolution.public_snapshot(), } add_goal_subagent_capability(capabilities, server=self.server) diff --git a/tests/test_antigravity_steward_adapter.py b/tests/test_antigravity_steward_adapter.py index 4438c5a13b..c1bc08b1a0 100644 --- a/tests/test_antigravity_steward_adapter.py +++ b/tests/test_antigravity_steward_adapter.py @@ -263,6 +263,39 @@ def listing(command, **kwargs): assert language_server["sent_count"] == 0 +def test_readiness_budget_includes_process_listing_but_execution_discovery_stays_fresh(monkeypatch): + from loopx.capabilities.native_chat import antigravity_adapter as provider + + clock = [0.0] + calls = [] + monkeypatch.setattr(provider.time, "monotonic", lambda: clock[0]) + monkeypatch.setattr(provider.os, "getuid", lambda: 501, raising=False) + + def listing(command, **kwargs): + calls.append(command[0]) + duration = 2.8 if command[0] == "ps" else 1.4 + clock[0] += min(duration, kwargs["timeout"]) + if duration > kwargs["timeout"]: + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + return ("123 /Applications/Antigravity.app/language_server --csrf_token synthetic-token\n" + if command[0] == "ps" else "server 123 user TCP 127.0.0.1:12345 (LISTEN)\n") + + def rpc(*args, **kwargs): + calls.append("rpc") + return {} + + monkeypatch.setattr(provider.subprocess, "check_output", listing) + monkeypatch.setattr(provider, "_rpc_post", rpc) + assert provider.probe_antigravity_endpoint(env={})[0] is False + assert clock[0] == pytest.approx(4.0) + assert calls == ["ps", "lsof"] + # A readiness timeout is not a cached denial for the next real operation. + discovered = provider.discover_antigravity_language_server(env={}) + assert discovered is not None + assert discovered["url"] == "http://127.0.0.1:12345" + assert calls == ["ps", "lsof", "ps", "lsof", "rpc"] + + def test_cancel_failure_cannot_replace_the_original_provider_error(tmp_path, language_server): from loopx.chat_agent import CodexChatAgentError diff --git a/tests/test_chat_manager_model_adoption.py b/tests/test_chat_manager_model_adoption.py index 0091fa1633..1384f6fa1d 100644 --- a/tests/test_chat_manager_model_adoption.py +++ b/tests/test_chat_manager_model_adoption.py @@ -107,6 +107,28 @@ def test_idle_conversation_adopts_model_preserving_context_and_authority(convers assert len(starts) == 2 +def test_readback_observation_does_not_cache_later_allocation(conversation, monkeypatch): + _, store, runtime, session, _ = conversation + before = store.load_session(session["session_id"]) + observed = [] + + def capabilities(): + available = bool(observed) + observed.append(available) + return [{"agent_id": "codex", "available": available}] + + monkeypatch.setattr(runtime, "capabilities", capabilities) + assert manager_executor_allocation(runtime, environ={}, + endpoint_capabilities=[{"agent_id": "codex", "available": True}])["available"] is True + # An explicitly empty observation remains unknown, without a hidden retry. + assert manager_executor_allocation(runtime, environ={}, endpoint_capabilities=[])["available"] is None + assert observed == [] + assert manager_executor_allocation(runtime, environ={})["available"] is False + assert manager_executor_allocation(runtime, environ={})["available"] is True + assert observed == [False, True] + assert store.load_session(session["session_id"]) == before + + def test_restart_migrates_legacy_history_once_preserving_source_and_native_provenance(conversation): root, store, runtime, session, starts = conversation sid = session["session_id"] diff --git a/tests/test_chat_transport_composition.py b/tests/test_chat_transport_composition.py index 2c1f2512d1..f011ddb119 100644 --- a/tests/test_chat_transport_composition.py +++ b/tests/test_chat_transport_composition.py @@ -171,6 +171,7 @@ def test_return_rechecks_actual_typed_binding_after_owner_disconnect(ordinary): @pytest.mark.parametrize("installed", [False, True]) def test_real_chat_entrypoint_composes_one_store_controller_and_return_service(tmp_path, monkeypatch, installed): import loopx.chat_server as chat + import loopx.chat_endpoint_catalog as endpoints from loopx.cli import build_parser from loopx.cli_commands.support_control_chat import chat_server_options from loopx.extensions.lark.cli_resolution import LarkCliResolution @@ -185,6 +186,17 @@ def test_real_chat_entrypoint_composes_one_store_controller_and_return_service(t monkeypatch.chdir(tmp_path) monkeypatch.setattr(chat, "build_lark_goal_topic_runtime_snapshot", lambda **kwargs: {}) monkeypatch.setattr(chat, "resolve_lark_cli_for_runtime", lambda **kwargs: LarkCliResolution(None, False, "missing", None, "lark_cli_not_installed")) + observations = [] + + def observe_endpoint(): + available = len(observations) % 2 == 0 + observations.append(available) + return {"available": available, "executable": "synthetic-endpoint"} + + monkeypatch.setattr(endpoints, "_antigravity_readiness", observe_endpoint) + monkeypatch.setattr(endpoints, "_executable_readiness", lambda executable: { + "available": executable == "unavailable-fixture", "executable": executable, + }) entered = threading.Event() captured = [] external = Transport("external-owner") @@ -217,6 +229,32 @@ def __init__(self, *args, **kwargs): server = captured[0] with urlopen(f"http://127.0.0.1:{server.server_port}/healthz", timeout=8) as response: assert json.load(response)["ok"] is True + from loopx.capabilities.machine_configuration.builtins import build_builtin_machine_configuration_registry + from loopx.capabilities.machine_configuration.store import configure_machine_configuration + + configuration = {"schema_version": "loopx_machine_configuration_v0", "namespaces": { + "steward_executor": {"schema_version": "steward_executor_machine_defaults_v1", + "selection_policy": "flexible", "eligible_endpoints": ["antigravity", "codex"], + "executor_endpoint": "antigravity", "executor_model": "gemini-4-argon-high", + "executor_reasoning_effort": "high"}}} + configuration_args = {"runtime_root": server.runtime_root, "configuration": configuration, + "registry": build_builtin_machine_configuration_registry()} + preview = configure_machine_configuration(**configuration_args, execute=False) + configure_machine_configuration(**configuration_args, execute=True, + expected_plan_revision=preview["plan_revision"]) + observations.clear() + for expected in (True, False, True): + with urlopen(f"http://127.0.0.1:{server.server_port}/api/chat/capabilities", timeout=8) as response: + capabilities = json.load(response) + available = {item["agent_id"]: item["available"] for item in capabilities["adapters"]} + assert available["antigravity"] is expected + # The channel must choose from the same observation shown in Settings, + # including when the preferred endpoint disappears and later returns. + selected = capabilities["manager"]["channel_binding"]["executor_endpoint"] + assert selected == ("antigravity" if expected else "codex") + assert available[selected] is True + assert observations == [True, False, True] + assert server.chat_store.list_sessions() == [] assert server.runtime_controller.startup_timeout_sec == 30.0 assert server.runtime_controller.acp_startup_timeout_sec is None if server.codex_account_service.provider is not None: