fix(client/auth): drop the orphaned refresh token on expiry discard; centralize the expiry predicate

A refresh token is bound to the client it was issued to (RFC 6749 §6), so
once the expiry discard replaces the registration, the kept refresh token
could only be presented as the wrong client and fail invalid_grant — which
happened whenever a flow failed between re-registration and token exchange.
Both discard sites (401 pre-Step-4 and 403 step-up) now drop the refresh
token with the record and persist the trimmed tokens, so an interrupted flow
or a restart cannot resurrect the orphan; the live access token is a bearer
credential and is kept.

The null-guarded expiry check, previously inlined verbatim at three flow
sites, moves into OAuthContext.registration_secret_expired() beside its
sibling predicates, and the discard mechanics into
OAuthContext.discard_expired_registration(), giving the rule a single home.

Author: claude

docs/client/oauth-clients.md

@@ -55,7 +55,9 @@ The in-memory version above works. It also forgets everything when the process e
     One exception: a stored registration whose dynamically issued secret has expired (a non-zero
     `client_secret_expires_at` in the past) is treated as absent — the expired secret could never
     authenticate again, so the provider discards the record and re-registers on the next flow,
-    overwriting it in your storage with the fresh registration.
+    overwriting it in your storage with the fresh registration. Any refresh token goes with the
+    discarded record (it was issued to that `client_id` and no other client can redeem it), so the
+    stored tokens are rewritten without it; a still-live access token is kept and keeps working.
 
 ### The two handlers
 

src/mcp/client/auth/oauth2.py

@@ -208,6 +208,25 @@ def can_refresh_token(self) -> bool:
         """Check if token can be refreshed."""
         return bool(self.current_tokens and self.current_tokens.refresh_token and self.client_info)
 
+    def registration_secret_expired(self) -> bool:
+        """Whether the loaded registration's minted secret has lapsed (RFC 7591)."""
+        return self.client_info is not None and stored_registration_expired(self.client_info)
+
+    async def discard_expired_registration(self) -> None:
+        """Discard a registration whose minted secret lapsed so the flow re-registers.
+
+        The refresh token goes with it: RFC 6749 §6 binds a refresh token to the client
+        it was issued to, so once the flow re-registers under a fresh `client_id` the
+        orphaned token could only fail `invalid_grant`. The trimmed tokens are persisted
+        so an interrupted flow (or a restart) cannot resurrect the orphan. The live
+        access token is a bearer credential that keeps working without client
+        authentication, so it is kept.
+        """
+        self.client_info = None
+        if self.current_tokens is not None and self.current_tokens.refresh_token is not None:
+            self.current_tokens.refresh_token = None
+            await self.storage.set_tokens(self.current_tokens)
+
     def clear_tokens(self) -> None:
         """Clear current tokens."""
         self.current_tokens = None
@@ -673,9 +692,7 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
             # `invalid_client`. Skip the doomed refresh and fall through to the 401 flow,
             # which re-registers; the record itself is kept for now so the flow's SEP-2352
             # issuer checks can still read its issuer stamp before the expiry discard runs.
-            registration_expired = self.context.client_info is not None and stored_registration_expired(
-                self.context.client_info
-            )
+            registration_expired = self.context.registration_secret_expired()
 
             if not self.context.is_token_valid() and self.context.can_refresh_token() and not registration_expired:
                 # Try to refresh token
@@ -794,13 +811,14 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                     # token endpoint. Discard it only now, after the SEP-2352 issuer
                     # checks above, so an expired record bound to a different issuer
                     # still got its cross-issuer cleanup; Step 4 then re-registers,
-                    # overwriting the dead record in storage. Any stored tokens are kept:
-                    # a live access token keeps working without client authentication.
-                    if self.context.client_info is not None and stored_registration_expired(self.context.client_info):
+                    # overwriting the dead record in storage. The refresh token is
+                    # dropped with the record it was issued to; the live access token
+                    # is kept — it works without client authentication.
+                    if self.context.registration_secret_expired():
                         logger.debug(
                             "Stored client registration secret has expired; discarding so this flow re-registers"
                         )
-                        self.context.client_info = None
+                        await self.context.discard_expired_registration()
 
                     # Step 4: Register client or use URL-based client ID (CIMD)
                     if not self.context.client_info:
@@ -847,13 +865,11 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         # 401 flow's discard from ever running. Discard it and mint fresh
                         # credentials first (mirroring the 401 flow's Step 4, reusing any
                         # AS metadata already discovered).
-                        if self.context.client_info is not None and stored_registration_expired(
-                            self.context.client_info
-                        ):
+                        if self.context.registration_secret_expired():
                             logger.debug(
                                 "Stored client registration secret has expired; re-registering before the step-up"
                             )
-                            self.context.client_info = None
+                            await self.context.discard_expired_registration()
                         if not self.context.client_info:
                             registration_request = await self._prepare_client_registration()
                             if registration_request is not None:

tests/client/test_auth.py

@@ -3302,6 +3302,24 @@ def test_stored_registration_expired_only_for_lapsed_secret_backed_registrations
     )
 
 
+def test_registration_secret_expired_is_false_until_a_lapsed_registration_is_loaded(
+    oauth_provider: OAuthClientProvider,
+):
+    """`OAuthContext.registration_secret_expired` owns the null check for the flow's call
+    sites (refresh gate, 401 discard, 403 step-up): no registration loaded means "not
+    expired", and a loaded record defers to `stored_registration_expired`.
+    """
+    assert not oauth_provider.context.registration_secret_expired()
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    assert oauth_provider.context.registration_secret_expired()
+
+
 @pytest.mark.anyio
 async def test_expired_stored_registration_is_discarded_and_the_flow_re_registers(
     oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage, valid_tokens: OAuthToken
@@ -3311,8 +3329,9 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     Reusing it makes every token-endpoint interaction fail with ``invalid_client`` — even a
     fresh interactive authorization ends in the same failure, so the client is permanently
     stuck (\"I re-authenticated and nothing changed\"). The 401 flow must discard the lapsed
-    record and re-register instead of presenting the dead secret; stored tokens are kept
-    (a live access token still works without client authentication).
+    record and re-register instead of presenting the dead secret; the access token is kept
+    (it still works without client authentication) while the refresh token — issued to the
+    discarded `client_id` — goes with the record.
     """
     await mock_storage.set_client_info(
         OAuthClientInformationFull(
@@ -3355,6 +3374,13 @@ async def test_expired_stored_registration_is_discarded_and_the_flow_re_register
     assert oauth_provider.context.client_info is None  # discarded in-flow, just before Step 4
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://api.example.com/register"
+    # The discard also dropped the dead client's refresh token, in memory and in storage.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == valid_tokens.access_token
     await auth_flow.aclose()
 
 
@@ -3515,6 +3541,93 @@ async def test_refresh_is_skipped_when_registration_secret_expired_mid_session(
     await auth_flow.aclose()
 
 
+@pytest.mark.anyio
+async def test_interrupted_flow_cannot_pair_the_old_refresh_token_with_the_fresh_registration(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """RFC 6749 §6 binds a refresh token to the client it was issued to, so the expiry
+    discard drops the token alongside the dead record — otherwise a flow failing between
+    re-registration and token exchange leaves fresh `client_info` paired with the old
+    client's refresh token, and the next request presents that token as the new client
+    (a guaranteed `invalid_grant` round trip).
+
+    Steps:
+    1. A registration whose secret lapsed mid-session sits next to an expired access
+       token and a refresh token issued to it.
+    2. The 401 flow discards the record — dropping the refresh token in memory and in
+       storage — and re-registers; the interactive authorization then fails (the user
+       abandons the consent) after the fresh registration was already persisted.
+    3. The next request attempts no refresh: with the orphan gone, `can_refresh_token()`
+       is false and the request goes out unauthenticated.
+    """
+    oauth_provider._initialized = True
+    oauth_provider.context.client_info = OAuthClientInformationFull(
+        client_id="dead-client",
+        client_secret="expired-secret",
+        client_secret_expires_at=int(time.time()) - 3600,  # lapsed after load
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="stale", refresh_token="issued-to-dead-client")
+    oauth_provider.context.token_expiry_time = time.time() - 60  # access token expired
+
+    async def abandoned_consent(url: str) -> None:
+        raise RuntimeError("user closed the browser")
+
+    oauth_provider.context.redirect_handler = abandoned_consent
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → discovery; the lapsed record is discarded and the flow re-registers.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://api.example.com", '
+            b'"authorization_endpoint": "https://api.example.com/authorize", '
+            b'"token_endpoint": "https://api.example.com/token", '
+            b'"registration_endpoint": "https://api.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+    register_req = await auth_flow.asend(asm_response)
+    register_response = httpx2.Response(
+        201,
+        json={
+            "client_id": "fresh-client",
+            "client_secret": "fresh-secret",
+            "redirect_uris": ["http://localhost:3030/callback"],
+            "token_endpoint_auth_method": "client_secret_post",
+        },
+        request=register_req,
+    )
+
+    # The fresh registration is persisted, then the interactive authorization fails.
+    with pytest.raises(RuntimeError, match="user closed the browser"):
+        await auth_flow.asend(register_response)
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "fresh-client"
+
+    # The orphaned refresh token is gone from memory and storage; the access token survives.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == "stale"
+
+    # The next request attempts no refresh: it goes out unauthenticated straight away.
+    retry_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    retry_request = await retry_flow.__anext__()
+    assert str(retry_request.url) == "https://api.example.com/v1/mcp"
+    assert "Authorization" not in retry_request.headers
+    await retry_flow.aclose()
+
+
 @pytest.mark.anyio
 async def test_403_step_up_re_registers_when_registration_secret_expired(
     oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
@@ -3525,7 +3638,8 @@ async def test_403_step_up_re_registers_when_registration_secret_expired(
     without its own re-check every step-up would complete a full interactive authorization
     only to fail ``invalid_client`` at the token exchange, until the access token itself
     expires. The step-up mints a fresh registration first — against the AS metadata already
-    discovered — and completes the exchange with the new credentials.
+    discovered — and completes the exchange with the new credentials; the dead client's
+    refresh token is dropped with its record.
     """
     oauth_provider._initialized = True
     oauth_provider.context.client_info = OAuthClientInformationFull(
@@ -3535,7 +3649,9 @@ async def test_403_step_up_re_registers_when_registration_secret_expired(
         redirect_uris=[AnyUrl("http://localhost:3030/callback")],
         token_endpoint_auth_method="client_secret_post",
     )
-    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token", scope="read")
+    oauth_provider.context.current_tokens = OAuthToken(
+        access_token="live-token", refresh_token="issued-to-dead-client", scope="read"
+    )
     oauth_provider.context.token_expiry_time = time.time() + 1800  # access token still live
     oauth_provider.context.oauth_metadata = OAuthMetadata(
         issuer=AnyHttpUrl("https://auth.example.com"),
@@ -3571,6 +3687,14 @@ async def mock_callback() -> AuthorizationCodeResult:
     register_req = await auth_flow.asend(response_403)
     assert register_req.method == "POST"
     assert str(register_req.url) == "https://auth.example.com/register"
+    # The discard dropped the dead client's refresh token (in memory and in storage) while
+    # keeping the live access token.
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+    stored_tokens = await mock_storage.get_tokens()
+    assert stored_tokens is not None
+    assert stored_tokens.refresh_token is None
+    assert stored_tokens.access_token == "live-token"
     register_response = httpx2.Response(
         201,
         json={