diff --git a/crates/kernel/rlm-runtime/README.md b/crates/kernel/rlm-runtime/README.md
index 220919f8..c535cdef 100644
--- a/crates/kernel/rlm-runtime/README.md
+++ b/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.
diff --git a/crates/kernel/rlm-runtime/src/rlm/mcp.py b/crates/kernel/rlm-runtime/src/rlm/mcp.py
index 355c192d..8dd6b9f5 100644
--- a/crates/kernel/rlm-runtime/src/rlm/mcp.py
+++ b/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.
 
diff --git a/crates/kernel/rlm-runtime/src/rlm/mcp_base.py b/crates/kernel/rlm-runtime/src/rlm/mcp_base.py
index e057935c..54214a58 100644
--- a/crates/kernel/rlm-runtime/src/rlm/mcp_base.py
+++ b/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,15 +165,31 @@ 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
 
@@ -176,24 +197,46 @@ 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.
diff --git a/crates/kernel/rlm-runtime/tests/test_mcp_connection_lifecycle.py b/crates/kernel/rlm-runtime/tests/test_mcp_connection_lifecycle.py
new file mode 100644
index 00000000..22f7ee85
--- /dev/null
+++ b/crates/kernel/rlm-runtime/tests/test_mcp_connection_lifecycle.py
@@ -0,0 +1,176 @@
+"""Authored integration lifecycle through the real typed host reply adapter.
+
+Uses temporary synthetic grants only; never touches a developer's auth store.
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from tempfile import TemporaryDirectory
+import time
+import traceback
+from types import SimpleNamespace
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from rlm import mcp_base, repl
+
+
+class Linear(mcp_base.McpIntegration):
+    server = "linear"
+    url = "https://mcp.linear.app/mcp"
+
+
+class ConnectionLifecycleTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self.temp = TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.home = Path(self.temp.name)
+        self.home_patch = patch.object(mcp_base, "_agent_dir", return_value=self.home)
+        self.home_patch.start()
+        self.addCleanup(self.home_patch.stop)
+        self.integration = Linear()
+
+    def write_grant(self, **overrides):
+        grant = {"type": "oauth", "endpoint": Linear.url, "access": "synthetic-access",
+                 "refresh": "synthetic-refresh", "expires": (time.time() - 60) * 1000}
+        grant.update(overrides)
+        (self.home / "auth.json").write_text(json.dumps({"mcp:linear": grant}))
+
+    async def test_missing_grant_is_not_enabled_and_never_refreshes(self):
+        host = AsyncMock()
+        with patch.object(repl, "host_request", host):
+            with self.assertRaises(mcp_base.NotEnabled):
+                await self.integration._resolve_token()
+        host.assert_not_awaited()
+
+    async def test_expired_grant_refreshes_once_for_concurrent_calls(self):
+        self.write_grant()
+
+        async def rotate(request):
+            self.assertEqual(request, {"type": "mcp.refresh", "server": "linear"})
+            await asyncio.sleep(0)
+            self.write_grant(access="rotated-synthetic-access", expires=(time.time() + 3600) * 1000)
+            return {"status": "ok", "result": {}}
+
+        host = AsyncMock(side_effect=rotate)
+        with patch.object(repl, "host_request", host):
+            results = await asyncio.gather(*(self.integration._resolve_token() for _ in range(5)))
+        self.assertEqual(results, ["rotated-synthetic-access"] * 5)
+        host.assert_awaited_once()
+
+    async def test_noop_refresh_is_not_misreported_as_missing_login(self):
+        self.write_grant()
+        host = AsyncMock(return_value={"status": "ok", "result": {"servers": []}})
+        with patch.object(repl, "host_request", host):
+            with self.assertRaisesRegex(RuntimeError, "without usable credentials") as caught:
+                await self.integration._resolve_token()
+        self.assertNotIsInstance(caught.exception, mcp_base.NotEnabled)
+        host.assert_awaited_once()
+
+    async def test_native_refresh_error_is_actionable_and_redacted(self):
+        self.write_grant()
+        host = AsyncMock(return_value={"status": "error", "error": (
+            "This Rust host cannot refresh authored integration credentials in auth.json. "
+            "synthetic-secret-token"
+        )})
+        with patch.object(repl, "host_request", host):
+            try:
+                await self.integration._resolve_token()
+            except RuntimeError as error:
+                rendered = "".join(traceback.format_exception(error))
+                self.assertIn("Rust host", rendered)
+                self.assertNotIn("synthetic-secret-token", rendered)
+            else:
+                self.fail("host rejection must fail")
+        host.assert_awaited_once()
+
+    async def test_generic_refresh_failure_does_not_misidentify_the_host(self):
+        self.write_grant()
+        host = AsyncMock(return_value={"status": "error", "error": "temporary provider failure"})
+        with patch.object(repl, "host_request", host):
+            with self.assertRaises(RuntimeError) as caught:
+                await self.integration._resolve_token()
+        self.assertIn("Failed to refresh", str(caught.exception))
+        self.assertNotIn("Rust host", str(caught.exception))
+
+    async def test_failed_refresh_rechecks_credentials_written_by_another_host(self):
+        self.write_grant()
+
+        async def concurrent_reconnect(request):
+            self.write_grant(access="reconnected-synthetic-access", expires=(time.time() + 3600) * 1000)
+            return {"status": "error", "error": "old refresh was superseded"}
+
+        host = AsyncMock(side_effect=concurrent_reconnect)
+        with patch.object(repl, "host_request", host):
+            self.assertEqual(await self.integration._resolve_token(), "reconnected-synthetic-access")
+        host.assert_awaited_once()
+
+    async def test_recheck_after_failure_keeps_binding_errors_and_tracebacks_safe(self):
+        self.write_grant()
+
+        async def invalid_reconnect(request):
+            self.write_grant(endpoint="https://another.example/mcp", expires=(time.time() + 3600) * 1000)
+            return {"status": "error", "error": "synthetic-provider-secret"}
+
+        with patch.object(repl, "host_request", AsyncMock(side_effect=invalid_reconnect)):
+            try:
+                await self.integration._resolve_token()
+            except RuntimeError as error:
+                rendered = "".join(traceback.format_exception(error))
+                self.assertIn("another endpoint", rendered)
+                self.assertNotIn("synthetic-provider-secret", rendered)
+            else:
+                self.fail("a retargeted grant must not be returned")
+
+    async def test_endpoint_mismatch_never_refreshes_or_transmits(self):
+        self.write_grant(endpoint="https://another.example/mcp", expires=(time.time() + 3600) * 1000)
+        host = AsyncMock()
+        with patch.object(repl, "host_request", host):
+            with self.assertRaisesRegex(RuntimeError, "another endpoint"):
+                await self.integration._resolve_token()
+        host.assert_not_awaited()
+
+    async def test_bad_expiry_and_access_are_not_treated_as_fresh(self):
+        for expires in (True, float("inf"), float("nan"), "9999999999999", 10**1000, None):
+            with self.subTest(expires=expires):
+                self.write_grant(expires=expires)
+                self.assertIsNone(self.integration._token())
+        self.write_grant(access={"unexpected": "object"}, expires=(time.time() + 3600) * 1000)
+        self.assertIsNone(self.integration._token())
+
+    async def test_cancelled_refresh_does_not_retry_or_poison_lock(self):
+        self.write_grant()
+        host = AsyncMock(side_effect=asyncio.CancelledError())
+        with patch.object(repl, "host_request", host):
+            with self.assertRaises(asyncio.CancelledError):
+                await self.integration._resolve_token()
+        self.assertFalse(self.integration._refresh_lock.locked())
+        host.assert_awaited_once()
+
+    async def test_explicit_discovery_rechecks_connection_and_clears_stale_tools(self):
+        session = SimpleNamespace(list_tools=AsyncMock(return_value=SimpleNamespace(tools=[
+            SimpleNamespace(name="list_issues", description="List", input_schema={})
+        ])))
+        with patch.object(self.integration, "_open_session", AsyncMock(return_value=session)):
+            self.assertEqual(await self.integration.list_tools(), [
+                {"name": "list_issues", "description": "List", "inputSchema": {}}
+            ])
+            session.list_tools.return_value = SimpleNamespace(tools=[])
+            self.assertEqual(await self.integration.list_tools(), [])
+            session.list_tools.side_effect = RuntimeError("disconnected")
+            with self.assertRaisesRegex(RuntimeError, "disconnected"):
+                await self.integration.list_tools()
+        self.assertIsNone(self.integration._tools)
+        self.assertEqual(session.list_tools.await_count, 3)
+
+    async def test_mutating_call_failure_is_never_replayed(self):
+        session = SimpleNamespace(call_tool=AsyncMock(side_effect=RuntimeError("connection lost")))
+        with patch.object(self.integration, "_open_session", AsyncMock(return_value=session)):
+            with self.assertRaisesRegex(RuntimeError, "connection lost"):
+                await self.integration.call_tool("create_issue", {"title": "Test"})
+        session.call_tool.assert_awaited_once_with("create_issue", {"title": "Test"})
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/crates/kernel/rlm-runtime/tests/test_mcp_host_list_tools.py b/crates/kernel/rlm-runtime/tests/test_mcp_host_list_tools.py
index c4a806cd..64c943c9 100644
--- a/crates/kernel/rlm-runtime/tests/test_mcp_host_list_tools.py
+++ b/crates/kernel/rlm-runtime/tests/test_mcp_host_list_tools.py
@@ -7,6 +7,29 @@
 
 
 class McpHostTests(unittest.IsolatedAsyncioTestCase):
+    async def test_list_servers_returns_only_connection_state(self):
+        state = {"server": "docs", "status": "not_started", "authStatus": "not_required", "toolCount": 0}
+        host = AsyncMock(return_value={"status": "ok", "result": {"servers": [
+            {**state, "Authorization": "must-not-escape", "url": "https://private.example"}
+        ]}})
+        with patch.object(repl, "host_request", host):
+            self.assertEqual(await mcp.list_servers(), [state])
+        host.assert_awaited_once_with({"type": "mcp.list_servers"})
+
+    async def test_list_servers_rejects_malformed_state(self):
+        state = {"server": "docs", "status": "ready", "authStatus": "authenticated", "toolCount": 0}
+        for result in ({}, {"servers": None}, {"servers": [None]},
+                       {"servers": [{**state, "status": "connected"}]},
+                       {"servers": [{**state, "status": []}]},
+                       {"servers": [{**state, "authStatus": {}}]},
+                       {"servers": [{**state, "toolCount": True}]},
+                       {"servers": [{**state, "toolCount": -1}]}):
+            with self.subTest(result=result), patch.object(
+                repl, "host_request", AsyncMock(return_value={"status": "ok", "result": result})
+            ):
+                with self.assertRaises(RuntimeError):
+                    await mcp.list_servers()
+
     async def test_native_list_tools_uses_exact_server_and_python_contract(self):
         schema = {"type": "object", "properties": {"q": {"type": "string"}}}
         host = AsyncMock(return_value={"status": "ok", "result": {"tools": [
@@ -65,7 +88,8 @@ async def test_host_rejections_remain_visible_without_retry(self):
                     host.assert_awaited_once()
 
     async def test_unsupported_host_requires_upgrade(self):
-        for action, call in (("mcp.list_tools", lambda: mcp.list_tools("docs")),
+        for action, call in (("mcp.list_servers", mcp.list_servers),
+                             ("mcp.list_tools", lambda: mcp.list_tools("docs")),
                              ("mcp.call", lambda: mcp.call_tool("docs", "search"))):
             for kind in ("unknown", "unsupported"):
                 for label in ("host_request action", "host request action", "action"):
@@ -131,7 +155,7 @@ async def test_repl_shutdown_keeps_local_child_cleanup_with_facade_loaded(self):
         send.assert_called_once_with({"event": "done", "id": "shutdown-1", "status": "ok"})
 
     def test_generic_facade_has_no_local_transport_or_lifecycle(self):
-        self.assertEqual(mcp.__all__, ["call_tool", "list_tools"])
+        self.assertEqual(mcp.__all__, ["call_tool", "list_servers", "list_tools"])
         for name in ("_Registry", "_Generation", "_registry", "_dispatch", "_StderrTail",
                      "reload", "close", "McpStartupError"):
             self.assertFalse(hasattr(mcp, name), name)
diff --git a/crates/kernel/rlm-runtime/tests/test_mcp_integrations.py b/crates/kernel/rlm-runtime/tests/test_mcp_integrations.py
index afb547f2..07b654e3 100644
--- a/crates/kernel/rlm-runtime/tests/test_mcp_integrations.py
+++ b/crates/kernel/rlm-runtime/tests/test_mcp_integrations.py
@@ -45,7 +45,7 @@ async def test_bundled_integrations_keep_discovery_and_parsed_results(self):
                     tools = await integration.list_tools()
                 self.assertEqual(result, {"items": ["found"]})
                 self.assertEqual(tools, [{"name": tool, "description": "Search", "inputSchema": schema}])
-                session.list_tools.assert_awaited_once_with()
+                self.assertEqual(session.list_tools.await_count, 2)
                 session.call_tool.assert_awaited_once_with(tool, {"query": "roadmap"})
                 host.assert_not_awaited()
                 for reserved in ("run", "__wrapped__", "__call__"):
diff --git a/crates/kernel/rlm-runtime/tests/test_mcp_kernel_wire.py b/crates/kernel/rlm-runtime/tests/test_mcp_kernel_wire.py
new file mode 100644
index 00000000..0934c530
--- /dev/null
+++ b/crates/kernel/rlm-runtime/tests/test_mcp_kernel_wire.py
@@ -0,0 +1,110 @@
+"""Real REPL process/wire simulation with synthetic host replies, not live OAuth."""
+
+import asyncio
+import json
+import os
+from pathlib import Path
+import sys
+from tempfile import TemporaryDirectory
+import time
+import unittest
+
+
+class McpKernelWireTests(unittest.IsolatedAsyncioTestCase):
+    async def test_real_kernel_discovery_call_denial_and_auth_rotation(self):
+        runtime = Path(__file__).resolve().parents[1]
+        bundled = runtime.parents[1] / "skills" / "assets" / "bundled"
+        with TemporaryDirectory() as home:
+            auth_file = Path(home) / "auth.json"
+
+            def save_grant(expires):
+                auth_file.write_text(json.dumps({"mcp:linear": {
+                    "type": "oauth", "endpoint": "https://mcp.linear.app/mcp",
+                    "access": "wire-synthetic-token", "refresh": "wire-synthetic-refresh",
+                    "expires": expires,
+                }}))
+
+            save_grant(0)
+            env = {**os.environ, "DEVO_CODING_AGENT_DIR": home,
+                   "PYTHONPATH": os.pathsep.join((str(runtime / "src"), str(bundled / "linear" / "src")))}
+            process = await asyncio.create_subprocess_exec(
+                sys.executable, "-m", "rlm.repl", env=env,
+                stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+
+            async def read():
+                line = await asyncio.wait_for(process.stdout.readline(), 10)
+                self.assertTrue(line, "kernel closed unexpectedly")
+                return json.loads(line)
+
+            async def send(frame):
+                process.stdin.write((json.dumps(frame) + "\n").encode())
+                await process.stdin.drain()
+
+            try:
+                self.assertEqual((await read())["event"], "ready")
+                await send({"type": "execute", "id": "mcp-wire", "code": """
+import linear
+from rlm import mcp
+# The synthetic refresh host updates auth.json; the integration re-reads it.
+assert await linear.linear._resolve_token()
+servers = await mcp.list_servers()
+assert servers == [{"server": "docs", "status": "not_started", "authStatus": "not_required", "toolCount": 0}]
+assert await mcp.list_tools("docs") == []
+try:
+    await mcp.call_tool("docs", "create", {"name": "demo"})
+except PermissionError:
+    print("MCP_WIRE_OK")
+else:
+    raise AssertionError("denial lost")
+"""})
+                calls = []
+                outputs = []
+                while True:
+                    frame = await read()
+                    self.assertNotEqual(frame["event"], "error", frame)
+                    if frame["event"] == "host_request":
+                        self.assertEqual(frame["cell_id"], "mcp-wire")
+                        request = frame["data"]
+                        calls.append(request)
+                        action = request["type"]
+                        if action == "mcp.refresh":
+                            save_grant((time.time() + 3600) * 1000)
+                            result = {"status": "ok", "result": {}}
+                        elif action == "mcp.list_servers":
+                            result = {"status": "ok", "result": {"servers": [{
+                                "server": "docs", "status": "not_started",
+                                "authStatus": "not_required", "toolCount": 0,
+                            }]}}
+                        elif action == "mcp.list_tools":
+                            result = {"status": "ok", "result": {"tools": []}}
+                        elif action == "mcp.call":
+                            result = {"status": "error", "error": "denied by synthetic host"}
+                        else:
+                            self.fail(f"unexpected host request: {action}")
+                        await send({"type": "host_reply", "id": frame["id"], "data": result})
+                    elif frame["event"] == "stdout":
+                        outputs.append(frame["text"])
+                    elif frame["event"] == "done":
+                        self.assertEqual(frame, {"event": "done", "id": "mcp-wire", "status": "ok"})
+                        break
+                self.assertEqual(calls, [
+                    {"type": "mcp.refresh", "server": "linear"},
+                    {"type": "mcp.list_servers"},
+                    {"type": "mcp.list_tools", "server": "docs"},
+                    {"type": "mcp.call", "server": "docs", "tool": "create", "arguments": {"name": "demo"}},
+                ])
+                self.assertEqual("".join(outputs), "MCP_WIRE_OK\n")
+            finally:
+                if process.returncode is None:
+                    await send({"type": "shutdown", "id": "shutdown"})
+                    try:
+                        await asyncio.wait_for(process.wait(), 5)
+                    except TimeoutError:
+                        process.kill()
+                        await process.wait()
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/crates/server/src/runtime/kernel_host.rs b/crates/server/src/runtime/kernel_host.rs
index 593727af..efd479ca 100644
--- a/crates/server/src/runtime/kernel_host.rs
+++ b/crates/server/src/runtime/kernel_host.rs
@@ -233,7 +233,21 @@ pub async fn dispatch_host_request_with_bridge(bridge: &HostBridge, data: Value)
         "mcp.list_tools" => {
             crate::runtime::kernel_host_bridge::handle_mcp_list_tools(bridge, &params).await
         }
-        "mcp.config" | "mcp.refresh" => ok_result(json!({ "servers": [] })),
+        "mcp.list_servers" => {
+            crate::runtime::kernel_host_bridge::handle_mcp_list_servers(bridge).await
+        }
+        // These are the authored Python integration's auth.json contracts, not
+        // the Rust manager's discovery refresh. Never acknowledge a no-op or
+        // return runtime transport config (which can contain credentials).
+        "mcp.refresh" => error_reply(
+            "This Rust host cannot refresh authored integration credentials in auth.json. \
+             Reconnect the integration using the host that owns its login. \
+             Configured native MCP servers use rlm.mcp.list_tools instead.",
+        ),
+        "mcp.config" => error_reply(
+            "This Rust host does not expose authored integration config. \
+             Use rlm.mcp.list_servers for native server status; credentials remain host-owned.",
+        ),
         "model.info" => ok_result(bridge.model_info_payload()),
         "plan.update" => {
             crate::runtime::kernel_host_bridge::handle_plan_update(bridge, &params).await
diff --git a/crates/server/src/runtime/kernel_host_bridge.rs b/crates/server/src/runtime/kernel_host_bridge.rs
index a221f464..71eeaa5b 100644
--- a/crates/server/src/runtime/kernel_host_bridge.rs
+++ b/crates/server/src/runtime/kernel_host_bridge.rs
@@ -22,6 +22,8 @@ use devo_core::McpManager;
 use devo_core::tools::contracts::ToolBudgets;
 mod harness;
 pub(crate) use harness::handle_harness;
+mod mcp;
+pub(crate) use mcp::{handle_mcp_call, handle_mcp_list_servers, handle_mcp_list_tools};
 mod shell;
 pub(crate) use shell::{handle_bash, handle_bash_start};
 
@@ -581,97 +583,6 @@ async fn handle_fs_read_once(bridge: &HostBridge, _params: &Value, path_str: &st
     json!({ "status": "ok", "result": { "content": content, "bytes": n, "once": true } })
 }
 
-pub(crate) async fn handle_mcp_call(bridge: &HostBridge, params: &Value) -> Value {
-    if plan_mode_denies_mutation(bridge) {
-        return deny_plan_mutation("mcp.call");
-    }
-    let Some(manager) = bridge.mcp_manager.as_ref() else {
-        return error_reply("mcp manager not available on this host bridge");
-    };
-    let server = params
-        .get("server")
-        .or_else(|| params.get("serverId"))
-        .or_else(|| params.get("server_id"))
-        .and_then(|v| v.as_str())
-        .unwrap_or("");
-    let tool_name = params
-        .get("tool")
-        .or_else(|| params.get("toolName"))
-        .or_else(|| params.get("name"))
-        .and_then(|v| v.as_str())
-        .unwrap_or("");
-    if server.is_empty() || tool_name.is_empty() {
-        return error_reply("mcp.call requires server and tool");
-    }
-    let input = params
-        .get("arguments")
-        .or_else(|| params.get("input"))
-        .or_else(|| params.get("args"))
-        .cloned()
-        .unwrap_or_else(|| json!({}));
-    let tool_call_id = synthetic_tool_call_id("mcp");
-    let req = permission_request(
-        bridge,
-        tool_call_id,
-        "mcp.call",
-        json!({
-            "server": server,
-            "tool": tool_name,
-            "arguments": input,
-        }),
-        ResourceKind::Custom("mcp.call".into()),
-        format!("Call MCP tool {server}/{tool_name}"),
-        None,
-        Some(format!("{server}/{tool_name}")),
-    );
-    if let Err(reply) = check_or_error(bridge, req).await {
-        return reply;
-    }
-    let server_id = devo_core::McpServerId(server.to_string());
-    match manager.invoke_tool(&server_id, tool_name, input).await {
-        Ok(result) => json!({
-            "status": "ok",
-            "result": result,
-        }),
-        Err(err) => error_reply(err.to_string()),
-    }
-}
-
-pub(crate) async fn handle_mcp_list_tools(bridge: &HostBridge, params: &Value) -> Value {
-    let Some(server) = params
-        .get("server")
-        .and_then(Value::as_str)
-        .filter(|server| !server.is_empty())
-    else {
-        return error_reply("mcp.list_tools requires server");
-    };
-    // Listing is read-only; skip Ask. Refreshing one configured server lists its
-    // tools once and avoids discovering every server for this per-server API.
-    let Some(manager) = bridge.mcp_manager.as_ref() else {
-        return error_reply("mcp manager not available on this host bridge");
-    };
-    let server_id = devo_core::McpServerId(server.to_string());
-    match manager.refresh(&server_id).await {
-        Ok(status) => {
-            let tools: Vec<Value> = status
-                .tools
-                .into_iter()
-                .filter(|tool| tool.server_id == server_id)
-                .map(|tool| {
-                    json!({
-                        "server": tool.server_id.0,
-                        "name": tool.name,
-                        "description": tool.description,
-                        "inputSchema": tool.input_schema,
-                    })
-                })
-                .collect();
-            json!({ "status": "ok", "result": { "tools": tools } })
-        }
-        Err(err) => error_reply(err.to_string()),
-    }
-}
-
 pub(crate) async fn handle_web_search(bridge: &HostBridge, params: &Value) -> Value {
     let query = params
         .get("query")
diff --git a/crates/server/src/runtime/kernel_host_bridge/mcp.rs b/crates/server/src/runtime/kernel_host_bridge/mcp.rs
new file mode 100644
index 00000000..425c77ba
--- /dev/null
+++ b/crates/server/src/runtime/kernel_host_bridge/mcp.rs
@@ -0,0 +1,132 @@
+//! MCP operations use the session manager; credentials never enter host replies.
+
+use devo_core::McpStartupState;
+use serde_json::{Value, json};
+
+use super::{
+    HostBridge, ResourceKind, check_or_error, deny_plan_mutation, error_reply, permission_request,
+    plan_mode_denies_mutation, synthetic_tool_call_id,
+};
+
+/// Return observed connection state without starting servers or exposing config.
+pub(crate) async fn handle_mcp_list_servers(bridge: &HostBridge) -> Value {
+    let Some(manager) = bridge.mcp_manager.as_ref() else {
+        return error_reply("mcp manager not available on this host bridge");
+    };
+    let mut statuses = match manager.statuses().await {
+        Ok(statuses) => statuses,
+        Err(error) => return error_reply(error.to_string()),
+    };
+    statuses.sort_by(|left, right| left.server_id.0.cmp(&right.server_id.0));
+    let servers: Vec<Value> = statuses
+        .into_iter()
+        .map(|status| {
+            json!({
+                "server": status.server_id.0,
+                "status": status.startup_state,
+                "authStatus": status.auth_state,
+                "toolCount": status.tools.len(),
+            })
+        })
+        .collect();
+    json!({ "status": "ok", "result": { "servers": servers } })
+}
+
+pub(crate) async fn handle_mcp_call(bridge: &HostBridge, params: &Value) -> Value {
+    if plan_mode_denies_mutation(bridge) {
+        return deny_plan_mutation("mcp.call");
+    }
+    let Some(manager) = bridge.mcp_manager.as_ref() else {
+        return error_reply("mcp manager not available on this host bridge");
+    };
+    let server = params
+        .get("server")
+        .or_else(|| params.get("serverId"))
+        .or_else(|| params.get("server_id"))
+        .and_then(|v| v.as_str())
+        .unwrap_or("");
+    let tool_name = params
+        .get("tool")
+        .or_else(|| params.get("toolName"))
+        .or_else(|| params.get("name"))
+        .and_then(|v| v.as_str())
+        .unwrap_or("");
+    if server.is_empty() || tool_name.is_empty() {
+        return error_reply("mcp.call requires server and tool");
+    }
+    let input = params
+        .get("arguments")
+        .or_else(|| params.get("input"))
+        .or_else(|| params.get("args"))
+        .cloned()
+        .unwrap_or_else(|| json!({}));
+    let tool_call_id = synthetic_tool_call_id("mcp");
+    let req = permission_request(
+        bridge,
+        tool_call_id,
+        "mcp.call",
+        json!({
+            "server": server,
+            "tool": tool_name,
+            "arguments": input,
+        }),
+        ResourceKind::Custom("mcp.call".into()),
+        format!("Call MCP tool {server}/{tool_name}"),
+        None,
+        Some(format!("{server}/{tool_name}")),
+    );
+    if let Err(reply) = check_or_error(bridge, req).await {
+        return reply;
+    }
+    let server_id = devo_core::McpServerId(server.to_string());
+    match manager.invoke_tool(&server_id, tool_name, input).await {
+        Ok(result) => json!({
+            "status": "ok",
+            "result": result,
+        }),
+        Err(err) => error_reply(err.to_string()),
+    }
+}
+
+pub(crate) async fn handle_mcp_list_tools(bridge: &HostBridge, params: &Value) -> Value {
+    let Some(server) = params
+        .get("server")
+        .and_then(Value::as_str)
+        .filter(|server| !server.is_empty())
+    else {
+        return error_reply("mcp.list_tools requires server");
+    };
+    // Listing is read-only; skip Ask. Refreshing one configured server lists its
+    // tools once and avoids discovering every server for this per-server API.
+    let Some(manager) = bridge.mcp_manager.as_ref() else {
+        return error_reply("mcp manager not available on this host bridge");
+    };
+    let server_id = devo_core::McpServerId(server.to_string());
+    match manager.refresh(&server_id).await {
+        Ok(status) => {
+            // A successful manager call is not necessarily a usable connection.
+            // Disabled/auth-required/failed servers must not look like an empty
+            // but connected server. Ready with zero tools remains valid.
+            if status.server_id != server_id || status.startup_state != McpStartupState::Ready {
+                return error_reply(format!(
+                    "MCP server '{server}' is not ready; inspect mcp.list_servers before retrying"
+                ));
+            }
+            let tools: Vec<Value> = status
+                .tools
+                .into_iter()
+                .filter(|tool| tool.server_id == server_id)
+                .map(|tool| {
+                    json!({
+                        "server": tool.server_id.0,
+                        "name": tool.name,
+                        "description": tool.description,
+                        "inputSchema": tool.input_schema,
+                    })
+                })
+                .collect();
+            json!({ "status": "ok", "result": { "tools": tools } })
+        }
+        Err(err) => error_reply(err.to_string()),
+    }
+}
diff --git a/crates/server/src/runtime/kernel_host_bridge_tests.rs b/crates/server/src/runtime/kernel_host_bridge_tests.rs
index 7e99a58a..45912961 100644
--- a/crates/server/src/runtime/kernel_host_bridge_tests.rs
+++ b/crates/server/src/runtime/kernel_host_bridge_tests.rs
@@ -4,6 +4,7 @@ use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering;
 
 use async_trait::async_trait;
+use pretty_assertions::assert_eq;
 use serde_json::Value;
 use serde_json::json;
 
@@ -25,6 +26,8 @@ struct FakeMcpManager {
     refreshed: Mutex<Vec<McpServerId>>,
     discoveries: AtomicUsize,
     invocations: AtomicUsize,
+    observed_statuses: Vec<McpServerStatus>,
+    refresh_state: McpStartupState,
 }
 
 impl FakeMcpManager {
@@ -33,6 +36,8 @@ impl FakeMcpManager {
             refreshed: Mutex::new(Vec::new()),
             discoveries: AtomicUsize::new(0),
             invocations: AtomicUsize::new(0),
+            observed_statuses: Vec::new(),
+            refresh_state: McpStartupState::Ready,
         }
     }
 }
@@ -40,7 +45,7 @@ impl FakeMcpManager {
 #[async_trait]
 impl McpManager for FakeMcpManager {
     async fn statuses(&self) -> Result<Vec<McpServerStatus>, McpError> {
-        Ok(Vec::new())
+        Ok(self.observed_statuses.clone())
     }
 
     async fn discover_tools(&self) -> Result<Vec<McpToolInfo>, McpError> {
@@ -52,7 +57,7 @@ impl McpManager for FakeMcpManager {
         self.refreshed.lock().unwrap().push(server_id.clone());
         Ok(McpServerStatus {
             server_id: server_id.clone(),
-            startup_state: McpStartupState::Ready,
+            startup_state: self.refresh_state.clone(),
             auth_state: McpAuthState::Authenticated,
             tools: vec![
                 McpToolDescriptor {
@@ -177,6 +182,87 @@ async fn list_tools_requires_server_before_refreshing() {
     assert_eq!(manager.discoveries.load(Ordering::SeqCst), 0);
 }
 
+#[tokio::test]
+async fn mcp_status_dispatch_is_read_only_and_does_not_expose_config_or_schemas() {
+    let mut manager = FakeMcpManager::new();
+    for (server, state) in [
+        ("z-disabled", McpStartupState::Disabled),
+        ("a-ready", McpStartupState::Ready),
+        ("b-auth", McpStartupState::AuthRequired),
+    ] {
+        manager.observed_statuses.push(McpServerStatus {
+            server_id: McpServerId(server.to_string()),
+            startup_state: state,
+            auth_state: McpAuthState::NotRequired,
+            tools: Vec::new(),
+            resources: Vec::new(),
+            resource_templates: Vec::new(),
+            last_refreshed_at: None,
+        });
+    }
+    let manager = Arc::new(manager);
+    let bridge = test_bridge(Arc::clone(&manager));
+    let reply = crate::runtime::kernel_host::dispatch_host_request_with_bridge(
+        &bridge,
+        json!({ "type": "mcp.list_servers" }),
+    )
+    .await;
+    assert_eq!(
+        reply,
+        json!({ "status": "ok", "result": { "servers": [
+            { "server": "a-ready", "status": "ready", "authStatus": "not_required", "toolCount": 0 },
+            { "server": "b-auth", "status": "auth_required", "authStatus": "not_required", "toolCount": 0 },
+            { "server": "z-disabled", "status": "disabled", "authStatus": "not_required", "toolCount": 0 }
+        ] } })
+    );
+    assert!(manager.refreshed.lock().unwrap().is_empty());
+    assert_eq!(manager.discoveries.load(Ordering::SeqCst), 0);
+    assert_eq!(manager.invocations.load(Ordering::SeqCst), 0);
+}
+
+#[tokio::test]
+async fn mcp_discovery_does_not_report_non_ready_servers_as_connected() {
+    for state in [
+        McpStartupState::Disabled,
+        McpStartupState::NotStarted,
+        McpStartupState::Starting,
+        McpStartupState::Failed,
+        McpStartupState::AuthRequired,
+        McpStartupState::Degraded,
+        McpStartupState::Stopped,
+    ] {
+        let mut manager = FakeMcpManager::new();
+        manager.refresh_state = state;
+        let bridge = test_bridge(Arc::new(manager));
+        let reply = handle_mcp_list_tools(&bridge, &json!({ "server": "docs" })).await;
+        assert_eq!(
+            reply,
+            json!({
+                "status": "error",
+                "error": "MCP server 'docs' is not ready; inspect mcp.list_servers before retrying"
+            })
+        );
+    }
+}
+
+#[tokio::test]
+async fn legacy_auth_actions_fail_explicitly_instead_of_acknowledging_noops() {
+    let manager = Arc::new(FakeMcpManager::new());
+    let bridge = test_bridge(Arc::clone(&manager));
+    for action in ["mcp.refresh", "mcp.config"] {
+        let reply = crate::runtime::kernel_host::dispatch_host_request_with_bridge(
+            &bridge,
+            json!({ "type": action, "server": "linear" }),
+        )
+        .await;
+        assert_eq!(reply["status"], "error");
+        assert!(reply.get("result").is_none());
+        assert!(reply["error"].as_str().unwrap().contains("Rust host"));
+    }
+    assert!(manager.refreshed.lock().unwrap().is_empty());
+    assert_eq!(manager.invocations.load(Ordering::SeqCst), 0);
+}
+
 #[cfg(any(unix, windows))]
 #[tokio::test]
 async fn mediated_read_pins_the_file_during_approval_for_both_read_modes() {
@@ -251,3 +337,77 @@ async fn both_mediated_read_modes_reject_oversized_files() {
         );
     }
 }
+
+/// Exercises the real Python REPL transport, Rust dispatcher, and manager/Ask
+/// boundary. The MCP manager and expired grant are synthetic; no live OAuth.
+#[tokio::test]
+#[ignore = "requires Python 3.11+; run explicitly for kernel/host integration changes"]
+async fn mcp_real_kernel_uses_native_dispatch_and_reports_legacy_refresh_failure() {
+    use devo_kernel::{KernelSession, KernelSessionConfig};
+
+    let home = tempfile::tempdir().unwrap();
+    std::fs::write(
+        home.path().join("auth.json"),
+        r#"{"mcp:linear":{"type":"oauth","endpoint":"https://mcp.linear.app/mcp","access":"synthetic-expired","expires":0}}"#,
+    )
+    .unwrap();
+    let crates_dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+        .parent()
+        .unwrap();
+    let mut config = KernelSessionConfig {
+        cwd: home.path().to_path_buf(),
+        python_path_entries: vec![
+            crates_dir.join("kernel/rlm-runtime/src"),
+            crates_dir.join("skills/assets/bundled/linear/src"),
+        ],
+        ..KernelSessionConfig::default()
+    };
+    config.extra_env.push((
+        "DEVO_CODING_AGENT_DIR".into(),
+        home.path().to_string_lossy().into_owned(),
+    ));
+    let kernel = KernelSession::spawn(config).await.unwrap();
+    let manager = Arc::new(FakeMcpManager::new());
+    let mut bridge = test_bridge(Arc::clone(&manager));
+    bridge.permission =
+        PermissionChecker::new(|_| Box::pin(async { Err("denied by synthetic host".to_string()) }));
+    kernel
+        .set_host_handler(Some(super::host_handler_for_bridge(Arc::new(bridge))))
+        .await;
+    let output = kernel
+        .execute(
+            r#"
+from rlm import mcp
+import linear
+assert await mcp.list_servers() == []
+assert await mcp.list_tools("docs") == [{
+    "name": "search:raw/name", "description": "Search docs",
+    "inputSchema": {"type": "object", "properties": {"query": {"type": "string"}}},
+}]
+try:
+    await mcp.call_tool("docs", "create", {"name": "demo"})
+except PermissionError:
+    pass
+else:
+    raise AssertionError("host denial was ignored")
+try:
+    await linear.linear._resolve_token()
+except RuntimeError as error:
+    assert "Rust host" in str(error)
+    from rlm.mcp_base import NotEnabled
+    assert not isinstance(error, NotEnabled)
+else:
+    raise AssertionError("native host falsely refreshed legacy credentials")
+print("MCP_NATIVE_WIRE_OK")
+"#,
+        )
+        .await
+        .unwrap();
+    assert_eq!(output.status, "ok", "{output:?}");
+    assert_eq!(output.stdout, "MCP_NATIVE_WIRE_OK\n");
+    assert_eq!(manager.invocations.load(Ordering::SeqCst), 0);
+    assert_eq!(
+        *manager.refreshed.lock().unwrap(),
+        vec![McpServerId("docs".to_string())]
+    );
+}
diff --git a/crates/skills/assets/bundled/linear/SKILL.md b/crates/skills/assets/bundled/linear/SKILL.md
index 9d81714e..6c00a5b9 100644
--- a/crates/skills/assets/bundled/linear/SKILL.md
+++ b/crates/skills/assets/bundled/linear/SKILL.md
@@ -14,6 +14,13 @@ Connect via `/login` → **Services** tab → **Linear** (OAuth in the browser).
 automatically. If a call raises `NotEnabled`, the user isn't logged in — walk
 them through `/login`; don't ask them to set environment variables.
 
+This authored Python integration uses the login host's `auth.json`, separately
+from native configured servers accessed through `rlm.mcp`. The current Rust
+host cannot refresh this store. A refresh error is not a missing-login error:
+report it accurately and reconnect through the host that owns the login.
+`list_tools()` verifies the connection each time; stored credentials alone do
+not prove it is connected. Never retry a failed write automatically.
+
 ## Usage
 
 The tool set is defined by the server, not by this skill, so **discover before
diff --git a/crates/skills/assets/bundled/notion/SKILL.md b/crates/skills/assets/bundled/notion/SKILL.md
index b99b8e10..909f81ae 100644
--- a/crates/skills/assets/bundled/notion/SKILL.md
+++ b/crates/skills/assets/bundled/notion/SKILL.md
@@ -14,6 +14,13 @@ Connect via `/login` → **Services** tab → **Notion** (OAuth in the browser).
 automatically. If a call raises `NotEnabled`, the user isn't logged in — walk
 them through `/login`; don't ask them to set environment variables.
 
+This authored Python integration uses the login host's `auth.json`, separately
+from native configured servers accessed through `rlm.mcp`. The current Rust
+host cannot refresh this store. A refresh error is not a missing-login error:
+report it accurately and reconnect through the host that owns the login.
+`list_tools()` verifies the connection each time; stored credentials alone do
+not prove it is connected. Never retry a failed write automatically.
+
 ## Usage
 
 The tool set is defined by the server, not by this skill, so **discover before
diff --git a/docs/mcp-connection-lifecycle.md b/docs/mcp-connection-lifecycle.md
new file mode 100644
index 00000000..4297bf1d
--- /dev/null
+++ b/docs/mcp-connection-lifecycle.md
@@ -0,0 +1,102 @@
+# MCP connection lifecycle
+
+## Ownership and current limits
+
+Devo currently has two distinct MCP paths. They are not interchangeable:
+
+| Path | Connection owner | Credentials | Result contract |
+| --- | --- | --- | --- |
+| `rlm.mcp` | Rust session MCP manager | Native runtime credential/config store | Full MCP `CallToolResult` |
+| Bundled `linear` / `notion` | Authored Python integration | Login host's `auth.json` | Parsed structured content or text |
+
+A same-named user server must never repoint an authored integration or receive
+its grant. This change does not migrate credentials between stores, start login,
+or add a second OAuth implementation to Rust.
+
+The Native host previously acknowledged `mcp.refresh` and `mcp.config` with
+`{"servers": []}`. That did not refresh a token or return a valid authored
+configuration. It now rejects these unsupported contracts explicitly. Calling
+Rust `McpManager::refresh` in their place would be incorrect: that method refreshes
+native server discovery, not the Python integration's `auth.json`.
+
+## Native discovery and status
+
+```python
+from rlm import mcp
+
+# Observed state only. Does not launch servers or expose credentials/config.
+print(await mcp.list_servers())
+
+# Verify one configured server's connection and obtain current schemas.
+tools = await mcp.list_tools("docs")
+
+# Uses the existing host approval path. Never retried through Python.
+result = await mcp.call_tool("docs", "search", {"query": "release"})
+```
+
+`list_servers()` returns `server`, `status`, `authStatus`, and `toolCount`.
+Statuses are `disabled`, `not_started`, `starting`, `ready`, `failed`,
+`auth_required`, `degraded`, or `stopped`. `ready` is the manager's observed
+successful initialization/discovery, not proof of current network health.
+Token presence alone is not a successful connection. Use `list_tools()` for a
+fresh check; even a successful refresh result must be in `ready` state. A ready
+server with zero tools is valid.
+
+The status reply intentionally excludes URLs, commands, environment variables,
+headers, tool schemas, and credential values. Tool calls still pass through the
+existing permission and plan-mode checks. Discovery never retries a mutation.
+
+## Authored integrations
+
+`await linear.list_tools()` and `await notion.list_tools()` now verify discovery
+on every explicit call. Dynamic tool methods reuse schemas between explicit
+checks; a failed check clears them rather than presenting stale discovery as a
+current connection. Tool calls keep their parsed-result contract and are never
+automatically replayed after a transport failure.
+
+Token resolution distinguishes:
+
+- Missing grant: `NotEnabled`, with login instructions.
+- Expired/malformed grant: ask its login host to refresh; serialize concurrent
+  refresh attempts and re-read the persisted grant before use.
+- Host failure or an acknowledged no-op: actionable refresh error, not a false
+  “no credentials” diagnosis. Provider error bodies are not copied into the
+  raised traceback.
+- Explicit endpoint mismatch: refuse use and refresh before sending credentials.
+- Cancellation: propagate without retrying or retaining the refresh lock.
+
+Legacy built-in grants without endpoint metadata remain readable for backwards
+compatibility. Their endpoint is fixed in the authored integration. Explicitly
+bound grants must match it exactly. Existing static-token environment support is
+unchanged; this change does not copy credentials into host replies.
+
+## Verification
+
+The regression suite covers both contracts and failure cases:
+
+```sh
+PYTHONPATH=crates/kernel/rlm-runtime/src python -m unittest discover \
+  -s crates/kernel/rlm-runtime/tests -p 'test_mcp*.py' -v
+cargo test -p devo-server --lib kernel_host_bridge
+cargo test -p devo-server --lib mcp_real_kernel -- --ignored
+```
+
+The Python wire test launches the actual REPL and supplies synthetic host
+replies. The Rust integration test launches that same REPL through
+`KernelSession`, uses the real host dispatcher/approval bridge, and substitutes
+only the MCP manager and expired credential fixture. Neither test contacts
+Linear/Notion or verifies a real OAuth browser flow.
+
+## Remaining complete-connection work
+
+This is a correctness and diagnostics slice, not an end-to-end OAuth migration.
+Completing the experience requires a single chosen credential/connection owner
+for login, refresh, logout, settings, and kernel calls; native login/cancellation
+and reconnect UI; endpoint-bound credential migration; and live provider tests
+with explicit user authorization. Reuse the existing Native protocol instead of
+adding parallel provider-specific transports or a copied plugin catalog.
+
+The useful Prime comparison is its separation of stored credentials from a
+verified connection record (`catalog_status_views.rs`, Prime `74591cd`). Devo
+can reuse its existing manager status rather than import that catalog's entire
+architecture.