fix(mcp): expose connection state and reject ineffective credential refresh

crates/kernel/rlm-runtime/README.md

@@ -266,7 +266,8 @@ to host commands.
 There are two layers:
 
 - `rlm.mcp` is a thin facade for the session's Rust MCP manager. Use
-  `await mcp.list_tools(server)` for `{name, description, inputSchema}` entries
+  `await mcp.list_servers()` for redacted observed connection states without
+  starting servers, and `await mcp.list_tools(server)` for `{name, description, inputSchema}` entries
   and `await mcp.call_tool(server, tool, arguments)` for the complete MCP
   `CallToolResult`, including `content`, `structuredContent`, and `isError`.
   Tool-level `isError` results are returned unchanged. Host call rejections
@@ -282,7 +283,12 @@ There are two layers:
   discovered on first use and bound through `__getattr__`; `call_tool(name,
   args)` is the explicit escape hatch. These authored integrations, including
   Linear and Notion, still use the Python SDK and retain their parsed-result
-  contract; they are separate from the generic host facade.
+  contract; they are separate from the generic host facade. An explicit
+  `list_tools()` rechecks the connection and clears stale schemas on failure.
+  Their legacy `auth.json` refresh contract is not implemented by the Rust host:
+  it reports an explicit error, never a successful no-op. Expired grants must be
+  reconnected through their owning login host. See
+  [MCP connection lifecycle](../../../docs/mcp-connection-lifecycle.md).
 
 The MCP exports are lazy. `import rlm` does not import the `mcp` SDK; only an
 integration package that subclasses `McpIntegration` needs it.

crates/kernel/rlm-runtime/src/rlm/mcp.py

@@ -8,7 +8,7 @@
 
 from typing import Any
 
-__all__ = ["call_tool", "list_tools"]
+__all__ = ["call_tool", "list_servers", "list_tools"]
 
 
 async def _request(action: str, payload: dict[str, Any]) -> Any:
@@ -39,6 +39,40 @@ async def _request(action: str, payload: dict[str, Any]) -> Any:
     return reply.get("result")
 
 
+async def list_servers() -> list[dict[str, Any]]:
+    """Return observed native server states, without initiating connections.
+
+    Entries contain ``server``, ``status``, ``authStatus``, and ``toolCount``.
+    Only ``ready`` records a successful handshake/discovery. This snapshot is
+    not a live health check: use ``list_tools(server)`` to verify a connection.
+    Credentials, URLs, commands, headers, and tool schemas are not returned.
+    """
+    result = await _request("mcp.list_servers", {})
+    if not isinstance(result, dict) or not isinstance(result.get("servers"), list):
+        raise RuntimeError("mcp.list_servers host response must contain result.servers")
+    servers = []
+    for server in result["servers"]:
+        if (
+            not isinstance(server, dict)
+            or not isinstance(server.get("server"), str)
+            or not server["server"]
+            or not isinstance(server.get("status"), str)
+            or server["status"] not in {
+                "disabled", "not_started", "starting", "ready", "failed",
+                "auth_required", "degraded", "stopped",
+            }
+            or not isinstance(server.get("authStatus"), str)
+            or server["authStatus"] not in {
+                "not_required", "authenticated", "auth_required", "auth_failed",
+            }
+            or type(server.get("toolCount")) is not int
+            or server["toolCount"] < 0
+        ):
+            raise RuntimeError("mcp.list_servers host response contains an invalid server")
+        servers.append({key: server[key] for key in ("server", "status", "authStatus", "toolCount")})
+    return servers
+
+
 async def list_tools(server: str) -> list[dict[str, Any]]:
     """List one server's tools through the Rust host.
 

crates/kernel/rlm-runtime/src/rlm/mcp_base.py

@@ -8,16 +8,19 @@
     import linear
     issues = await linear.list_issues(team="Engineering")
 
-Credentials live in the host's ``auth.json`` (single store, survives kernel
-rebuilds). This module reads that file directly for the common case; on token
+Authored integration credentials live in their login host's ``auth.json``
+(survives kernel rebuilds). This module reads that file directly for the common case; on token
 expiry it asks the host to refresh via ``rlm.host_request("mcp.refresh", ...)``
-and re-reads. Interactive login runs host-side, never here.
+and re-reads. Interactive login runs host-side, never here. The current Rust
+host does not implement this legacy auth.json refresh contract and reports that
+explicitly; a native configured MCP server is instead used through ``rlm.mcp``.
 """
 
 from __future__ import annotations
 
 import asyncio
 import json
+import math
 import os
 import time
 from contextlib import AsyncExitStack
@@ -44,8 +47,9 @@ def __init__(self, server: str):
         self.server = server
         super().__init__(
             f"The '{server}' integration is not enabled: no credentials found. "
-            f"Tell the user to run `/mcp login {server}` (or `/login` → Services) "
-            f"in Devo to connect it. Do not ask them to set environment variables."
+            f"Connect through its login host (`/mcp login {server}` or `/login` → Services "
+            "where supported). Native Rust-configured MCP servers use rlm.mcp instead. "
+            "Do not ask the user to set environment variables."
         )
 
 
@@ -140,6 +144,7 @@ def __init__(self) -> None:
             raise ValueError(f"{type(self).__name__} must set a non-empty `server`")
         self._tools: dict[str, Any] | None = None
         self._lock = asyncio.Lock()
+        self._refresh_lock = asyncio.Lock()
 
     # -- credentials --------------------------------------------------------
 
@@ -160,40 +165,78 @@ def _token(self) -> str | None:
         cred = _read_auth(self._provider_id)
         if cred is None:
             return None
+        # Never refresh or transmit a credential explicitly bound elsewhere.
+        # Legacy built-in grants may lack a binding; their endpoint is authored
+        # in code and cannot be replaced by same-named user MCP configuration.
+        endpoint = cred.get("endpoint")
+        if endpoint is not None and endpoint != self.url:
+            raise RuntimeError(
+                f"Stored credentials for '{self.server}' belong to another endpoint. "
+                "Reconnect the integration; no credential was sent."
+            )
         if cred.get("type") == "api_key":
-            return _resolve_config_value(str(cred.get("key") or "")) or None
-        # OAuth credential: {access, refresh, expires(ms)}.
-        access = str(cred.get("access") or "")
+            key = cred.get("key")
+            return (_resolve_config_value(key) or None) if isinstance(key, str) else None
+        if cred.get("type") != "oauth":
+            return None
+        # OAuth credential: {access, refresh, expires(ms)}. Malformed expiry
+        # (including booleans and infinity) must not make a token immortal.
+        access = cred.get("access")
         expires = cred.get("expires")
-        fresh = isinstance(expires, (int, float)) and (
-            time.time() * 1000 < expires - _EXPIRY_SKEW_SECONDS * 1000
+        fresh = (
+            type(expires) in (int, float)
+            and 0 <= expires <= 2**53 - 1
+            and math.isfinite(expires)
+            and time.time() * 1000 < expires - _EXPIRY_SKEW_SECONDS * 1000
         )
-        if access and fresh:
+        if isinstance(access, str) and access.strip() and fresh:
             return access
         return None  # signal: needs refresh
 
     async def _resolve_token(self) -> str:
         token = self._token()
         if token:
             return token
-        # Expired or missing-access: ask the host to refresh, then re-validate via
-        # _token() (which re-checks expiry) rather than trusting any access value.
-        if _read_auth(self._provider_id) is not None:
-            refresh_error: Exception | None = None
+        # Concurrent calls share one refresh and re-read the store afterwards.
+        # This matters for providers that rotate single-use refresh tokens.
+        async with self._refresh_lock:
+            token = self._token()
+            if token:
+                return token
+            if _read_auth(self._provider_id) is None:
+                raise NotEnabled(self.server)
+            refresh_failure: str | None = None
             try:
                 await host_request("mcp.refresh", {"server": self.server})
-            except RuntimeError as exc:
-                refresh_error = exc
+            except RuntimeError as error:
+                # Provider errors may contain response bodies or credentials.
+                # Keep the recovery actionable without leaking them in a
+                # chained traceback. Cancellation is deliberately not caught.
+                reason = (
+                    "The Rust host does not refresh authored auth.json credentials; "
+                    "native configured servers use rlm.mcp instead. "
+                    if str(error).startswith(
+                        "This Rust host cannot refresh authored integration credentials in auth.json."
+                    ) else ""
+                )
+                refresh_failure = (
+                    f"Failed to refresh credentials for '{self.server}'. {reason}"
+                    "Reconnect using the host that owns this integration's login."
+                )
+            # A second host may have reconnected while this request failed.
+            # Re-read outside the exception handler: the new grant must pass
+            # the same binding/expiry checks, without chaining a provider body.
             token = self._token()
             if token:
                 return token
-            # A refresh that failed (vs. genuinely-absent creds) is a recoverable
-            # error; don't mislabel it as "not enabled / re-login".
-            if refresh_error is not None:
-                raise RuntimeError(
-                    f"Failed to refresh credentials for '{self.server}': {refresh_error}"
-                ) from refresh_error
-        raise NotEnabled(self.server)
+            if refresh_failure is not None:
+                raise RuntimeError(refresh_failure) from None
+            # Acknowledgement is not proof of rotation. Do not turn a stale or
+            # malformed grant into the misleading 'not enabled' diagnosis.
+            raise RuntimeError(
+                f"Credential refresh for '{self.server}' returned without usable credentials. "
+                "Reconnect the integration using the host that owns its login."
+            )
 
     # -- connection ---------------------------------------------------------
 
@@ -262,31 +305,41 @@ async def _open_session(self, stack: AsyncExitStack):
     # -- tools --------------------------------------------------------------
 
     async def list_tools(self) -> list[dict[str, Any]]:
-        """Return the server's tools as ``[{name, description, inputSchema}]``."""
-        await self._ensure_tools()
-        return [dict(t) for t in (self._tools or {}).values()]
+        """Verify the connection and return its current tools and schemas.
+
+        An explicit listing reconnects even when tools are cached. A failed
+        verification clears stale discovery, and never retries a tool call.
+        """
+        async with self._lock:
+            await self._discover_tools()
+            return [dict(t) for t in (self._tools or {}).values()]
 
     async def _ensure_tools(self) -> None:
         if self._tools is not None:
             return
         async with self._lock:
-            if self._tools is not None:
-                return
-            async with AsyncExitStack() as stack:
-                session = await self._open_session(stack)
-                resp = await session.list_tools()
-                tools: dict[str, Any] = {}
-                for t in resp.tools:
-                    # mcp>=2 exposes the pydantic field input_schema; inputSchema is the wire alias.
-                    schema = getattr(t, "input_schema", None)
-                    if schema is None:
-                        schema = getattr(t, "inputSchema", None)
-                    tools[t.name] = {
-                        "name": t.name,
-                        "description": getattr(t, "description", "") or "",
-                        "inputSchema": schema if isinstance(schema, dict) else {},
-                    }
-                self._tools = tools
+            if self._tools is None:
+                await self._discover_tools()
+
+    async def _discover_tools(self) -> None:
+        # Caller holds _lock. Clear first so a failed reconnect never leaves a
+        # stale schema presented as a current, usable connection.
+        self._tools = None
+        async with AsyncExitStack() as stack:
+            session = await self._open_session(stack)
+            resp = await session.list_tools()
+            tools: dict[str, Any] = {}
+            for t in resp.tools:
+                # mcp>=2 exposes input_schema; inputSchema is the wire alias.
+                schema = getattr(t, "input_schema", None)
+                if schema is None:
+                    schema = getattr(t, "inputSchema", None)
+                tools[t.name] = {
+                    "name": t.name,
+                    "description": getattr(t, "description", "") or "",
+                    "inputSchema": schema if isinstance(schema, dict) else {},
+                }
+            self._tools = tools
 
     async def call_tool(self, tool: str, arguments: dict[str, Any] | None = None) -> Any:
         """Call ``tool`` on the server and return its parsed result.