fix(chat): share endpoint observations within capability responses

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,14 +183,17 @@ 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="],
             text=True,
             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,

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 = ""
 

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)

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
 

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"]

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: