fix(mcp): scope oauth_user server status to the requesting user

Follow-up to f585c47b (review finding #4). _oauth_user_server_status derived
connected + tools/resources/prompts counts from warm[0] — an arbitrary user's
pool entry — and get_all_server_status surfaced that to every read-scoped
caller of /v1/api/_internal/mcp-status, ignoring who was asking. So user B saw
user A's oauth_user server as connected with A's catalog size, over the wire
(connected + the three counts are in _READ_STATUS_PUBLIC_KEYS; user_pools /
auth_type are stripped). Before f585c47b these servers were absent from the
read map entirely.

Thread user_id through get_all_server_status -> get_server_status ->
_oauth_user_server_status; the warm-pool filter now matches uid == user_id, so
connected + counts reflect ONLY the requester's own pool. internal_mcp_status
passes _auth_user_id(request); an empty/absent principal (user_id falsy) sees
oauth_user servers as not-connected. Static-server status is unaffected (the
new param defaults to None and is ignored for them).

Note: the admin console (admin.mcp) reaches this same read endpoint via the
console proxy, which forwards the ADMIN's identity — so an admin now sees an
oauth_user server scoped to their OWN pool (typically not-connected) rather
than the prior any-user aggregate. Server-global health (circuit_open / error /
consecutive_failures) is unchanged, and the consented-users-count is a separate
aggregate. Restoring an aggregate in-use pill for admins (without re-leaking
per-user catalogs) would need a privilege-aware aggregate mode + admin.js
change — deferred.

Tests: updated TestOAuthUserServerStatus to the scoped signature, added the
cross-user isolation regression (user B sees neither A's connected flag nor A's
catalog size) and a no-user-context case.

tests/test_mcp_pool_auth_introspection.py

@@ -1953,34 +1953,71 @@ class TestPoolPrimingAndTokenRotation:
 
 
 class TestOAuthUserServerStatus:
-    """``get_server_status`` for ``auth_type='oauth_user'`` servers reflects
-    per-user pool warmth instead of a permanent global "connecting" — so the
-    console pill flips to connected once a pool is primed (the third leg of the
-    OBO fix set)."""
+    """``get_server_status`` for ``auth_type='oauth_user'`` servers reflects the
+    REQUESTING user's pool warmth (scoped by user_id), never another user's — so
+    the console pill flips to connected once that user's pool is primed, without
+    leaking one user's catalog to another."""
 
-    def test_oauth_user_status_connected_when_pool_warm(self) -> None:
+    @staticmethod
+    def _warm(mgr: MCPClientManager, user_id: str, server: str, n_tools: int = 1) -> None:
         from turnstone.core.mcp_client import PoolEntryState
 
+        entry = PoolEntryState(key=(user_id, server), open_lock=MagicMock())
+        entry.session = MagicMock()
+        entry.tools = [{"function": {"name": f"mcp__{server}__t{i}"}} for i in range(n_tools)]
+        mgr._user_pool_entries[(user_id, server)] = entry
+
+    def test_oauth_user_status_connected_for_own_warm_pool(self) -> None:
         mgr = MCPClientManager({})
         mgr._oauth_user_server_names = {"pool-srv"}
-        entry = PoolEntryState(key=("user-1", "pool-srv"), open_lock=MagicMock())
-        entry.session = MagicMock()
-        entry.tools = [{"function": {"name": "mcp__pool-srv__do"}}]
-        mgr._user_pool_entries[("user-1", "pool-srv")] = entry
+        self._warm(mgr, "user-1", "pool-srv", n_tools=1)
 
-        st = mgr.get_server_status("pool-srv")
+        st = mgr.get_server_status("pool-srv", user_id="user-1")
         assert st["connected"] is True
         assert st["tools"] == 1
         assert st["auth_type"] == "oauth_user"
         assert st["user_pools"] == 1
         # Also surfaced in the all-servers map (oauth_user is absent from
         # _server_configs, so this exercises the explicit union).
-        assert "pool-srv" in mgr.get_all_server_status()
+        assert "pool-srv" in mgr.get_all_server_status(user_id="user-1")
+
+    def test_oauth_user_status_does_not_leak_other_users_pool(self) -> None:
+        """#4 regression: user B must NOT see user A's warm pool — neither the
+        connected flag nor the catalog count. Before scoping, status was derived
+        from warm[0] (an arbitrary user), leaking A's catalog size to B over the
+        read-scoped /mcp-status endpoint."""
+        mgr = MCPClientManager({})
+        mgr._oauth_user_server_names = {"pool-srv"}
+        self._warm(mgr, "user-A", "pool-srv", n_tools=5)
+
+        own = mgr.get_server_status("pool-srv", user_id="user-A")
+        assert own["connected"] is True
+        assert own["tools"] == 5
+
+        other = mgr.get_server_status("pool-srv", user_id="user-B")
+        assert other["connected"] is False, "user B must not see user A's pool as connected"
+        assert other["tools"] == 0, "user B must not see user A's catalog size"
+        assert other["user_pools"] == 0
+
+    def test_oauth_user_status_no_user_context_is_not_connected(self) -> None:
+        """A request with no user context (user_id falsy — e.g. an operator
+        refresh/reconnect) reports not-connected rather than an arbitrary
+        user's pool."""
+        mgr = MCPClientManager({})
+        mgr._oauth_user_server_names = {"pool-srv"}
+        self._warm(mgr, "user-A", "pool-srv", n_tools=3)
+
+        for uid in (None, ""):
+            st = mgr.get_server_status("pool-srv", user_id=uid)
+            assert st["connected"] is False, f"user_id={uid!r} must not see a pool"
+            assert st["tools"] == 0
+            assert st["user_pools"] == 0
+            assert st["auth_type"] == "oauth_user"
 
     def test_oauth_user_status_connecting_when_no_warm_pool(self) -> None:
         mgr = MCPClientManager({})
         mgr._oauth_user_server_names = {"pool-srv"}
-        st = mgr.get_server_status("pool-srv")
+        st = mgr.get_server_status("pool-srv", user_id="user-1")
         assert st["connected"] is False
         assert st["tools"] == 0
         assert st["user_pools"] == 0

turnstone/core/mcp_client.py

@@ -3367,15 +3367,20 @@ class MCPClientManager:
         clean = msg.replace("\n", " ").replace("\r", "")
         self._last_error[name] = clean[: self._MAX_ERROR_LEN]
 
-    def get_server_status(self, name: str) -> dict[str, Any]:
-        """Return live status for a single server, including config details."""
+    def get_server_status(self, name: str, user_id: str | None = None) -> dict[str, Any]:
+        """Return live status for a single server, including config details.
+
+        For ``auth_type='oauth_user'`` servers the result is scoped to *user_id*
+        (see :meth:`_oauth_user_server_status`). ``user_id`` is ignored for
+        static servers, whose session is process-global.
+        """
         # auth_type='oauth_user' servers hold NO process-global session — they
         # are warmed per-user into the pool — so the static-session check below
-        # would always report them "connecting". Derive their status from warm
-        # per-user pool entries instead, so the console pill reflects real
-        # per-user reachability once a pool is primed.
+        # would always report them "connecting". Derive their status from the
+        # REQUESTING user's warm pool entry instead, so the console pill reflects
+        # that user's real reachability once their pool is primed.
         if name in self._oauth_user_server_names:
-            return self._oauth_user_server_status(name)
+            return self._oauth_user_server_status(name, user_id)
         state = self._static_servers.get(name)
         connected = state is not None and state.session is not None
         cfg = self._server_configs.get(name, {})
@@ -3404,26 +3409,33 @@ class MCPClientManager:
             "last_refresh_outcome": last_refresh[1] if last_refresh is not None else None,
         }
 
-    def _oauth_user_server_status(self, name: str) -> dict[str, Any]:
-        """Live status for an ``auth_type='oauth_user'`` server.
+    def _oauth_user_server_status(self, name: str, user_id: str | None) -> dict[str, Any]:
+        """Live status for an ``auth_type='oauth_user'`` server, scoped to *user_id*.
 
         These have no global session (stripped from ``_server_configs`` /
         ``_static_servers``); they connect per-user into ``_user_pool_entries``.
-        Report ``connected=True`` when at least one user has a warm pool entry,
-        with a representative catalog count and the number of warm user pools —
-        so the console stops showing a permanent "connecting"/``---`` for a
-        server that is in fact reachable and in use.
+        ``connected`` and the catalog counts reflect ONLY the requesting user's
+        warm pool entry — never another user's. The per-user pool is per-user
+        data, and ``connected`` / ``tools`` / ``resources`` / ``prompts`` reach
+        read-scoped callers over the wire, so deriving them from an arbitrary
+        other user's pool would leak that user's catalog (and its existence) to
+        anyone with read scope. A request with no user context (``user_id``
+        falsy, e.g. an operator refresh/reconnect) reports ``connected=False``.
         """
         # Snapshot with list(): the mcp-loop thread mutates _user_pool_entries
         # (prime insert / idle eviction) concurrently with status polls from the
         # console/server thread, and iterating a live dict that changes size
         # raises RuntimeError mid-comprehension. The sibling get_all_server_status
         # and the eviction loop snapshot the same way.
-        warm = [
-            entry
-            for (uid, sname), entry in list(self._user_pool_entries.items())
-            if sname == name and entry.session is not None
-        ]
+        warm = (
+            [
+                entry
+                for (uid, sname), entry in list(self._user_pool_entries.items())
+                if sname == name and uid == user_id and entry.session is not None
+            ]
+            if user_id
+            else []
+        )
         rep = warm[0] if warm else None
         cb_deadline = self._circuit_open_until.get(name)
         cb_open = cb_deadline is not None and time.monotonic() < cb_deadline
@@ -3445,19 +3457,20 @@ class MCPClientManager:
             "last_refresh_outcome": last_refresh[1] if last_refresh is not None else None,
         }
 
-    def get_all_server_status(self) -> dict[str, dict[str, Any]]:
+    def get_all_server_status(self, user_id: str | None = None) -> dict[str, dict[str, Any]]:
         """Return live status for all configured servers.
 
         Includes ``oauth_user`` servers (which are absent from
         ``_server_configs``) so the console list reports their real per-user
         pool status instead of falling back to a DB-only "connecting" default.
+        Their status is scoped to *user_id* (see :meth:`get_server_status`).
         """
         result: dict[str, dict[str, Any]] = {}
         for name in list(self._server_configs):
-            result[name] = self.get_server_status(name)
+            result[name] = self.get_server_status(name, user_id)
         for name in list(self._oauth_user_server_names):
             if name not in result:
-                result[name] = self.get_server_status(name)
+                result[name] = self.get_server_status(name, user_id)
         return result
 
     def reconcile_sync(self, storage: Any, timeout: int = 30) -> dict[str, Any]:

turnstone/server.py

@@ -3188,11 +3188,15 @@ def internal_mcp_status(request: Request) -> JSONResponse:
     if mcp_mgr is None:
         return JSONResponse({"servers": {}})
 
+    # Scope oauth_user server status to the requesting user — their per-user pool
+    # catalog (and its existence) must not leak to other read-scoped callers.
+    # _auth_user_id returns "" when unauthenticated, which the manager treats as
+    # "no user context" (oauth_user servers then report not-connected).
+    all_status = mcp_mgr.get_all_server_status(_auth_user_id(request))
     return JSONResponse(
         {
             "servers": {
-                name: _strip_server_status_for_read(status)
-                for name, status in mcp_mgr.get_all_server_status().items()
+                name: _strip_server_status_for_read(status) for name, status in all_status.items()
             }
         }
     )