diff --git a/docs/client/oauth-clients.md b/docs/client/oauth-clients.md
index cd7de35626..c3e81c2510 100644
--- a/docs/client/oauth-clients.md
+++ b/docs/client/oauth-clients.md
@@ -52,6 +52,13 @@ The in-memory version above works. It also forgets everything when the process e
     Store `client_info`, not only the tokens. The provider registers dynamically the first time it
     finds no stored `client_info`. Throw it away and you mint a fresh registration on every run.
 
+    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. 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
 
 The authorization code flow needs a human exactly once: someone has to sign in and click "allow".
@@ -95,7 +102,7 @@ The repository ships the live version. `examples/servers/simple-auth/` runs a st
 
 The 2026-07-28 revision of the spec deprecates dynamic client registration in favor of **Client ID Metadata Documents** (CIMD). Instead of POSTing a fresh registration to every authorization server it meets, your client publishes one JSON document about itself at a stable HTTPS URL, and that URL *is* its `client_id`. The authorization server fetches the document; the provider never touches it.
 
-The SDK already speaks it: pass the URL as `client_metadata_url=` when you construct the provider. When the authorization server's metadata advertises `client_id_metadata_document_supported: true`, the provider skips the `/register` request entirely: the URL goes into the flow as the `client_id`, and there is no `client_secret`. When the server doesn't advertise it (most don't yet), or you never pass a URL, the provider falls back to dynamic registration **silently**, and everything above works exactly as described. Stored `client_info` still wins over both.
+The SDK already speaks it: pass the URL as `client_metadata_url=` when you construct the provider. When the authorization server's metadata advertises `client_id_metadata_document_supported: true`, the provider skips the `/register` request entirely: the URL goes into the flow as the `client_id`, and there is no `client_secret`. When the server doesn't advertise it (most don't yet), or you never pass a URL, the provider falls back to dynamic registration **silently**, and everything above works exactly as described. Stored `client_info` still wins over both, as long as its registration is usable — a record whose dynamically issued secret has expired is discarded and the provider registers (or resolves the CIMD URL) afresh.
 
 The URL must be HTTPS with a non-root path; anything else is a `ValueError` at construction, before any network happens. The shipped `examples/clients/simple-auth-client/` takes it as the `MCP_CLIENT_METADATA_URL` environment variable.
 
diff --git a/src/mcp/client/auth/oauth2.py b/src/mcp/client/auth/oauth2.py
index 7dc62b52b9..d5718ac4b9 100644
--- a/src/mcp/client/auth/oauth2.py
+++ b/src/mcp/client/auth/oauth2.py
@@ -10,6 +10,7 @@
 import string
 import time
 from collections.abc import AsyncGenerator, Awaitable, Callable
+from contextlib import aclosing
 from dataclasses import dataclass, field
 from typing import Any, Protocol, get_args
 from urllib.parse import quote, urlencode, urljoin, urlparse
@@ -109,6 +110,22 @@ def check_registration_usable(client_info: OAuthClientInformationFull) -> None:
         )
 
 
+def stored_registration_expired(client_info: OAuthClientInformationFull) -> bool:
+    """Whether a stored registration's minted secret has lapsed and can no longer authenticate.
+
+    RFC 7591 requires `client_secret_expires_at` whenever a secret is issued, with ``0``
+    meaning the secret never expires. Once a non-zero expiry passes, every token-endpoint
+    interaction authenticating with that secret fails with ``invalid_client`` — and with no
+    RFC 7592 rotation endpoint, re-registration is the only standard recovery. The lapse
+    only matters for registrations that authenticate with the minted secret: ``none`` (or
+    an absent method) sends no secret, and `private_key_jwt` signs an assertion instead.
+    """
+    if client_info.token_endpoint_auth_method not in _SECRET_TOKEN_ENDPOINT_AUTH_METHODS:
+        return False
+    expires_at = client_info.client_secret_expires_at
+    return expires_at is not None and expires_at != 0 and expires_at < int(time.time())
+
+
 class PKCEParameters(BaseModel):
     """PKCE (Proof Key for Code Exchange) parameters."""
 
@@ -192,6 +209,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
@@ -548,7 +584,16 @@ async def _handle_refresh_response(self, response: httpx2.Response) -> bool:
             return False
 
     async def _initialize(self) -> None:
-        """Load stored tokens and client info."""
+        """Load stored tokens and client info.
+
+        A stored registration whose minted secret has expired (RFC 7591
+        `client_secret_expires_at`) is loaded as-is rather than discarded here: the auth
+        flow discards it right before re-registering, *after* the SEP-2352 issuer checks,
+        which need the record's issuer stamp — an expired record that is also bound to a
+        different issuer must still get its cross-issuer cleanup (dropping the old
+        issuer's tokens and cached metadata). Until then the dead secret is never
+        presented: the refresh branch and the 403 step-up skip it explicitly.
+        """
         self.context.current_tokens = await self.context.storage.get_tokens()
         self.context.client_info = await self.context.storage.get_client_info()
         self._initialized = True
@@ -577,6 +622,158 @@ async def _validate_resource_match(self, prm: ProtectedResourceMetadata) -> None
         if not check_resource_allowed(requested_resource=default_resource, configured_resource=prm_resource):
             raise OAuthFlowError(f"Protected resource {prm_resource} does not match expected {default_resource}")
 
+    def _registration_issuer(self) -> str | None:
+        """SEP-2352: the issuer to bind newly minted credentials to, when known."""
+        if self.context.oauth_metadata is not None:
+            return self.context.auth_server_url or str(self.context.oauth_metadata.issuer)
+        return None
+
+    async def _prepare_client_registration(self) -> httpx2.Request | None:
+        """Resolve a URL-based client ID (CIMD) or build a Dynamic Client Registration request.
+
+        When the server supports CIMD the client information is created (and persisted)
+        immediately and ``None`` is returned — no network round trip is needed. Otherwise
+        the returned registration request must be sent and its response passed to
+        `_complete_client_registration`.
+        """
+        if should_use_client_metadata_url(self.context.oauth_metadata, self.context.client_metadata_url):
+            # Use URL-based client ID (CIMD). CIMD records are portable across
+            # authorization servers, so the issuer stamp is informational.
+            logger.debug(f"Using URL-based client ID (CIMD): {self.context.client_metadata_url}")
+            client_information = create_client_info_from_metadata_url(
+                self.context.client_metadata_url,  # type: ignore[arg-type]
+                redirect_uris=self.context.client_metadata.redirect_uris,
+            )
+            client_information.issuer = self._registration_issuer()
+            self.context.client_info = client_information
+            await self.context.storage.set_client_info(client_information)
+            return None
+
+        # Fallback to Dynamic Client Registration
+        fallback_base = self.context.get_authorization_base_url(self.context.server_url)
+        return create_client_registration_request(
+            self.context.oauth_metadata, self.context.client_metadata, fallback_base
+        )
+
+    async def _complete_client_registration(self, response: httpx2.Response) -> None:
+        """Handle a Dynamic Client Registration response and persist the minted record."""
+        client_information = await handle_registration_response(response)
+        check_registration_usable(client_information)
+        discovered_issuer = self._registration_issuer()
+        fallback_base = self.context.get_authorization_base_url(self.context.server_url)
+        # Only record the issuer when the registration actually targeted the discovered
+        # AS — either via its published registration_endpoint, or because the
+        # resource-origin /register fallback is on the issuer's own host (legacy
+        # same-origin embedded AS). Otherwise the fallback hit a different server and
+        # recording a binding to the PRM-advertised AS would persist a binding that was
+        # never established.
+        if (
+            self.context.oauth_metadata is not None
+            and discovered_issuer is not None
+            and (
+                self.context.oauth_metadata.registration_endpoint is not None
+                or self.context.get_authorization_base_url(discovered_issuer) == fallback_base
+            )
+        ):
+            client_information.issuer = discovered_issuer
+        self.context.client_info = client_information
+        await self.context.storage.set_client_info(client_information)
+
+    async def _discover_authorization_server_metadata(
+        self, response: httpx2.Response
+    ) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        """Discover protected resource and authorization server metadata.
+
+        Runs the discovery sequence shared by the 401 flow and the 403 step-up:
+        protected resource metadata (SEP-985, seeded from the challenge's
+        `resource_metadata` parameter when present), the SEP-2352 issuer checks on
+        stored credentials, and OAuth authorization server metadata. Yields each
+        discovery request; the caller must send the response back into the generator
+        (the httpx auth-flow protocol).
+        """
+        www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
+
+        # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
+        prm_discovery_urls = build_protected_resource_metadata_discovery_urls(
+            www_auth_resource_metadata_url, self.context.server_url
+        )
+
+        for url in prm_discovery_urls:  # pragma: no branch
+            discovery_request = create_oauth_metadata_request(url)
+
+            discovery_response = yield discovery_request  # sending request
+
+            prm = await handle_protected_resource_response(discovery_response)
+            if prm:
+                # Validate PRM resource matches server URL (RFC 8707)
+                await self._validate_resource_match(prm)
+                self.context.protected_resource_metadata = prm
+
+                # todo: try all authorization_servers to find the OASM
+                assert (
+                    len(prm.authorization_servers) > 0
+                )  # this is always true as authorization_servers has a min length of 1
+
+                self.context.auth_server_url = str(prm.authorization_servers[0])
+                break
+            else:
+                logger.debug(f"Protected resource metadata discovery failed: {url}")
+
+        # SEP-2352: stored credentials are bound to the issuer that registered them.
+        # If the authorization server changed, drop them (and the old tokens) so the
+        # flow re-registers instead of presenting another server's credentials.
+        if (
+            self.context.client_info is not None
+            and self.context.auth_server_url is not None
+            and not credentials_match_issuer(
+                self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url
+            )
+        ):
+            logger.debug("Authorization server changed; discarding bound credentials and re-registering")
+            self.context.client_info = None
+            self.context.clear_tokens()
+            # Any cached AS metadata is for the old server; drop it so a failed
+            # rediscovery cannot leak the old registration/token endpoints into Step 4.
+            self.context.oauth_metadata = None
+
+        asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
+            self.context.auth_server_url, self.context.server_url
+        )
+
+        # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
+        for url in asm_discovery_urls:  # pragma: no branch
+            oauth_metadata_request = create_oauth_metadata_request(url)
+            oauth_metadata_response = yield oauth_metadata_request
+
+            ok, asm = await handle_auth_metadata_response(oauth_metadata_response)
+            if not ok:
+                break
+            if ok and asm:
+                # SEP-2468: metadata issuer must match the discovery issuer
+                if self.context.auth_server_url is not None:
+                    validate_metadata_issuer(asm, self.context.auth_server_url)
+                self.context.oauth_metadata = asm
+                break
+            else:
+                logger.debug(f"OAuth metadata discovery failed: {url}")
+
+        # SEP-2352: on the legacy no-PRM path the issuer is only known after ASM
+        # discovery, so re-evaluate the binding here using the discovered metadata
+        # issuer (mirroring the bound_issuer fallback in Step 4).
+        if (
+            self.context.client_info is not None
+            and self.context.auth_server_url is None
+            and self.context.oauth_metadata is not None
+            and not credentials_match_issuer(
+                self.context.client_info,
+                str(self.context.oauth_metadata.issuer),
+                self.context.client_metadata_url,
+            )
+        ):
+            logger.debug("Authorization server changed; discarding bound credentials and re-registering")
+            self.context.client_info = None
+            self.context.clear_tokens()
+
     async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
         """httpx2 auth flow integration."""
         async with self.context.lock:
@@ -586,7 +783,14 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
             # Capture protocol version from request headers
             self.context.protocol_version = request.headers.get(MCP_PROTOCOL_VERSION_HEADER)
 
-            if not self.context.is_token_valid() and self.context.can_refresh_token():
+            # A refresh request authenticates with the minted secret, so a registration
+            # whose secret has lapsed (RFC 7591 `client_secret_expires_at`) can only fail
+            # `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.registration_secret_expired()
+
+            if not self.context.is_token_valid() and self.context.can_refresh_token() and not registration_expired:
                 # Try to refresh token
                 refresh_request = await self._refresh_token()
                 refresh_response = yield refresh_request
@@ -604,88 +808,22 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 # Perform full OAuth flow
                 try:
                     # OAuth flow must be inline due to generator constraints
-                    www_auth_resource_metadata_url = extract_resource_metadata_from_www_auth(response)
-
-                    # Step 1: Discover protected resource metadata (SEP-985 with fallback support)
-                    prm_discovery_urls = build_protected_resource_metadata_discovery_urls(
-                        www_auth_resource_metadata_url, self.context.server_url
-                    )
-
-                    for url in prm_discovery_urls:  # pragma: no branch
-                        discovery_request = create_oauth_metadata_request(url)
-
-                        discovery_response = yield discovery_request  # sending request
-
-                        prm = await handle_protected_resource_response(discovery_response)
-                        if prm:
-                            # Validate PRM resource matches server URL (RFC 8707)
-                            await self._validate_resource_match(prm)
-                            self.context.protected_resource_metadata = prm
-
-                            # todo: try all authorization_servers to find the OASM
-                            assert (
-                                len(prm.authorization_servers) > 0
-                            )  # this is always true as authorization_servers has a min length of 1
-
-                            self.context.auth_server_url = str(prm.authorization_servers[0])
-                            break
-                        else:
-                            logger.debug(f"Protected resource metadata discovery failed: {url}")
-
-                    # SEP-2352: stored credentials are bound to the issuer that registered them.
-                    # If the authorization server changed, drop them (and the old tokens) so the
-                    # flow re-registers instead of presenting another server's credentials.
-                    if (
-                        self.context.client_info is not None
-                        and self.context.auth_server_url is not None
-                        and not credentials_match_issuer(
-                            self.context.client_info, self.context.auth_server_url, self.context.client_metadata_url
-                        )
-                    ):
-                        logger.debug("Authorization server changed; discarding bound credentials and re-registering")
-                        self.context.client_info = None
-                        self.context.clear_tokens()
-                        # Any cached AS metadata is for the old server; drop it so a failed
-                        # rediscovery cannot leak the old registration/token endpoints into Step 4.
-                        self.context.oauth_metadata = None
-
-                    asm_discovery_urls = build_oauth_authorization_server_metadata_discovery_urls(
-                        self.context.auth_server_url, self.context.server_url
-                    )
 
-                    # Step 2: Discover OAuth Authorization Server Metadata (OASM) (with fallback for legacy servers)
-                    for url in asm_discovery_urls:  # pragma: no branch
-                        oauth_metadata_request = create_oauth_metadata_request(url)
-                        oauth_metadata_response = yield oauth_metadata_request
-
-                        ok, asm = await handle_auth_metadata_response(oauth_metadata_response)
-                        if not ok:
-                            break
-                        if ok and asm:
-                            # SEP-2468: metadata issuer must match the discovery issuer
-                            if self.context.auth_server_url is not None:
-                                validate_metadata_issuer(asm, self.context.auth_server_url)
-                            self.context.oauth_metadata = asm
-                            break
-                        else:
-                            logger.debug(f"OAuth metadata discovery failed: {url}")
-
-                    # SEP-2352: on the legacy no-PRM path the issuer is only known after ASM
-                    # discovery, so re-evaluate the binding here using the discovered metadata
-                    # issuer (mirroring the bound_issuer fallback in Step 4).
-                    if (
-                        self.context.client_info is not None
-                        and self.context.auth_server_url is None
-                        and self.context.oauth_metadata is not None
-                        and not credentials_match_issuer(
-                            self.context.client_info,
-                            str(self.context.oauth_metadata.issuer),
-                            self.context.client_metadata_url,
-                        )
-                    ):
-                        logger.debug("Authorization server changed; discarding bound credentials and re-registering")
-                        self.context.client_info = None
-                        self.context.clear_tokens()
+                    # Steps 1-2: Discover protected resource and authorization server
+                    # metadata, applying the SEP-2352 issuer checks along the way. The
+                    # sequence lives in a sub-generator shared with the 403 step-up;
+                    # its requests are relayed by hand (`yield from` cannot cross an
+                    # async generator). `aclosing` finalizes the sub-generator when
+                    # httpx2 closes this flow mid-discovery (transport error or
+                    # cancellation throws `GeneratorExit` at the relay's `yield`).
+                    async with aclosing(self._discover_authorization_server_metadata(response)) as discovery:
+                        try:
+                            discovery_request = await anext(discovery)
+                            while True:
+                                discovery_response = yield discovery_request
+                                discovery_request = await discovery.asend(discovery_response)
+                        except StopAsyncIteration:
+                            pass
 
                     # Step 3: Apply scope selection strategy
                     self.context.client_metadata.scope = get_client_metadata_scopes(
@@ -695,52 +833,28 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         self.context.client_metadata.grant_types,
                     )
 
+                    # A registration whose minted secret lapsed (RFC 7591
+                    # `client_secret_expires_at`) — whether loaded from storage or expired
+                    # mid-session — can no longer authenticate: reusing it would burn an
+                    # interactive authorization doomed to fail `invalid_client` at the
+                    # 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. 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"
+                        )
+                        await self.context.discard_expired_registration()
+
                     # Step 4: Register client or use URL-based client ID (CIMD)
                     if not self.context.client_info:
-                        # SEP-2352: the issuer to bind these credentials to, when known.
-                        discovered_issuer: str | None = None
-                        if self.context.oauth_metadata is not None:
-                            discovered_issuer = self.context.auth_server_url or str(self.context.oauth_metadata.issuer)
-
-                        if should_use_client_metadata_url(
-                            self.context.oauth_metadata, self.context.client_metadata_url
-                        ):
-                            # Use URL-based client ID (CIMD). CIMD records are portable across
-                            # authorization servers, so the issuer stamp is informational.
-                            logger.debug(f"Using URL-based client ID (CIMD): {self.context.client_metadata_url}")
-                            client_information = create_client_info_from_metadata_url(
-                                self.context.client_metadata_url,  # type: ignore[arg-type]
-                                redirect_uris=self.context.client_metadata.redirect_uris,
-                            )
-                            client_information.issuer = discovered_issuer
-                            self.context.client_info = client_information
-                            await self.context.storage.set_client_info(client_information)
-                        else:
-                            # Fallback to Dynamic Client Registration
-                            fallback_base = self.context.get_authorization_base_url(self.context.server_url)
-                            registration_request = create_client_registration_request(
-                                self.context.oauth_metadata, self.context.client_metadata, fallback_base
-                            )
+                        registration_request = await self._prepare_client_registration()
+                        if registration_request is not None:
                             registration_response = yield registration_request
-                            client_information = await handle_registration_response(registration_response)
-                            check_registration_usable(client_information)
-                            # Only record the issuer when the registration above actually targeted
-                            # the discovered AS — either via its published registration_endpoint,
-                            # or because the resource-origin /register fallback is on the issuer's
-                            # own host (legacy same-origin embedded AS). Otherwise the fallback hit
-                            # a different server and recording a binding to the PRM-advertised AS
-                            # would persist a binding that was never established.
-                            if (
-                                self.context.oauth_metadata is not None
-                                and discovered_issuer is not None
-                                and (
-                                    self.context.oauth_metadata.registration_endpoint is not None
-                                    or self.context.get_authorization_base_url(discovered_issuer) == fallback_base
-                                )
-                            ):
-                                client_information.issuer = discovered_issuer
-                            self.context.client_info = client_information
-                            await self.context.storage.set_client_info(client_information)
+                            await self._complete_client_registration(registration_response)
 
                     # Step 5: Perform authorization and complete token exchange
                     token_response = yield await self._perform_authorization()
@@ -759,6 +873,31 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                 # Step 2: Check if we need to step-up authorization
                 if error == "insufficient_scope":  # pragma: no branch
                     try:
+                        # After a restart the 403 step-up can be the first auth event:
+                        # `_initialize` restores tokens and client info but never AS
+                        # metadata, and the still-live access token keeps the 401 flow
+                        # (and its discovery) from running. When this step-up will need
+                        # to re-register — the stored secret lapsed, or no record is
+                        # stored at all — discover the metadata first instead of
+                        # registering blind: the blind fallback would bypass a
+                        # configured CIMD URL and POST the registration document to the
+                        # resource origin's `/register`, the wrong server in the
+                        # separate-AS topology. Running it before the expiry discard
+                        # lets the SEP-2352 issuer checks still see the record's stamp.
+                        if self.context.oauth_metadata is None and (
+                            self.context.registration_secret_expired() or self.context.client_info is None
+                        ):
+                            # `aclosing` mirrors the 401 relay above: it finalizes the
+                            # sub-generator when httpx2 closes this flow mid-discovery.
+                            async with aclosing(self._discover_authorization_server_metadata(response)) as discovery:
+                                try:
+                                    discovery_request = await anext(discovery)
+                                    while True:
+                                        discovery_response = yield discovery_request
+                                        discovery_request = await discovery.asend(discovery_response)
+                                except StopAsyncIteration:
+                                    pass
+
                         # Step 2a: Union previously requested scopes with the newly challenged
                         # scopes (SEP-2350) so escalating one operation keeps the others' grants.
                         # Fold in the stored token's scope too: on a restart the token is reloaded
@@ -773,10 +912,28 @@ async def async_auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx
                         prior_scope = union_scopes(self.context.client_metadata.scope, granted_scope)
                         self.context.client_metadata.scope = union_scopes(prior_scope, challenged_scope)
 
+                        # A registration whose minted secret lapsed (RFC 7591
+                        # `client_secret_expires_at`) cannot complete the step-up: the
+                        # token exchange would fail `invalid_client` after burning a full
+                        # interactive consent — and the still-live access token keeps the
+                        # 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.registration_secret_expired():
+                            logger.debug(
+                                "Stored client registration secret has expired; re-registering before the step-up"
+                            )
+                            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:
+                                registration_response = yield registration_request
+                                await self._complete_client_registration(registration_response)
+
                         # Step 2b: Perform (re-)authorization and token exchange
                         token_response = yield await self._perform_authorization()
                         await self._handle_token_response(token_response)
-                    except Exception:  # pragma: no cover
+                    except Exception:
                         logger.exception("OAuth flow error")
                         raise
 
diff --git a/src/mcp/client/auth/utils.py b/src/mcp/client/auth/utils.py
index 31e2e5cade..5210e821a3 100644
--- a/src/mcp/client/auth/utils.py
+++ b/src/mcp/client/auth/utils.py
@@ -51,10 +51,13 @@ def extract_scope_from_www_auth(response: Response) -> str | None:
 def extract_resource_metadata_from_www_auth(response: Response) -> str | None:
     """Extract protected resource metadata URL from WWW-Authenticate header as per RFC 9728.
 
+    RFC 9728 attaches the `resource_metadata` parameter to any WWW-Authenticate
+    challenge, so both 401 and 403 (scope step-up) responses are honored.
+
     Returns:
         Resource metadata URL if found in WWW-Authenticate header, None otherwise
     """
-    if not response or response.status_code != 401:
+    if not response or response.status_code not in (401, 403):
         return None  # pragma: no cover
 
     return extract_field_from_www_auth(response, "resource_metadata")
diff --git a/tests/client/test_auth.py b/tests/client/test_auth.py
index be96cc8eec..a209020902 100644
--- a/tests/client/test_auth.py
+++ b/tests/client/test_auth.py
@@ -3,6 +3,7 @@
 import base64
 import json
 import time
+from collections.abc import AsyncGenerator
 from unittest import mock
 from urllib.parse import parse_qs, quote, unquote, urlparse
 
@@ -13,6 +14,7 @@
 
 from mcp.client.auth import OAuthClientProvider, PKCEParameters
 from mcp.client.auth.exceptions import OAuthFlowError, OAuthRegistrationError, OAuthTokenError
+from mcp.client.auth.oauth2 import stored_registration_expired
 from mcp.client.auth.utils import (
     build_oauth_authorization_server_metadata_discovery_urls,
     build_protected_resource_metadata_discovery_urls,
@@ -3253,3 +3255,896 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+def test_stored_registration_expired_only_for_lapsed_secret_backed_registrations():
+    """RFC 7591: only a non-zero, past `client_secret_expires_at` on a secret-authenticating
+    registration marks the stored record as expired; `0` means the secret never expires, and
+    methods that send no secret (`none`) are unaffected by the lapse.
+    """
+    base: dict[str, object] = {
+        "client_id": "c",
+        "client_secret": "s",
+        "redirect_uris": [AnyUrl("http://localhost:3030/callback")],
+    }
+    lapsed = int(time.time()) - 3600
+    live = int(time.time()) + 3600
+
+    expired = OAuthClientInformationFull.model_validate(
+        {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": lapsed}
+    )
+    assert stored_registration_expired(expired)
+    assert stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_basic", "client_secret_expires_at": lapsed}
+        )
+    )
+
+    # 0 means "never expires" (RFC 7591); absent means no expiry was declared.
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": 0}
+        )
+    )
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate({**base, "token_endpoint_auth_method": "client_secret_post"})
+    )
+
+    # Still-live secret, and methods that never present the secret.
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "client_secret_post", "client_secret_expires_at": live}
+        )
+    )
+    assert not stored_registration_expired(
+        OAuthClientInformationFull.model_validate(
+            {**base, "token_endpoint_auth_method": "none", "client_secret_expires_at": lapsed}
+        )
+    )
+
+
+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
+):
+    """Regression for #3256: a stored DCR registration whose secret has lapsed is not reused.
+
+    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; 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(
+            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",
+        )
+    )
+    await mock_storage.set_tokens(valid_tokens)
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+
+    # The record is loaded as-is (the flow discards it later, after the SEP-2352 issuer
+    # checks); the stored access token is kept and used.
+    request = await auth_flow.__anext__()
+    assert oauth_provider.context.client_info is not None
+    assert oauth_provider.context.current_tokens is not None
+    assert request.headers["Authorization"] == f"Bearer {valid_tokens.access_token}"
+
+    # `MockTokenStorage` stores by reference and `_initialize` loaded that same object into
+    # `context.current_tokens`, so the discard's in-place trim would reach the stored object
+    # even if it never persisted. Re-seed storage with a distinct copy so the storage-side
+    # assertions below pass only if the discard actually calls `set_tokens`.
+    await mock_storage.set_tokens(valid_tokens.model_copy())
+
+    # Server rejects the stale token: the 401 flow re-registers instead of reusing the record.
+    response_401 = httpx2.Response(401, request=request)
+    prm_req = await auth_flow.asend(response_401)
+    prm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=prm_req))
+    assert str(asm_req.url) == "https://api.example.com/.well-known/oauth-authorization-server"
+    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)
+    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()
+
+
+@pytest.mark.anyio
+async def test_registration_that_expires_mid_session_is_discarded_by_the_401_flow(
+    oauth_provider: OAuthClientProvider,
+):
+    """A secret that lapses after load (long-lived process) is also discarded, in-flow.
+
+    ``_initialize`` runs once per provider instance, so a registration that expires while
+    the process is running would otherwise be reused by the 401 flow: Step 4 would skip
+    re-registration and the interactive authorization would burn a user consent only to
+    fail ``invalid_client`` at the token exchange. The 401 flow re-checks the expiry
+    (after the SEP-2352 issuer checks) and discards the dead record so Step 4
+    re-registers instead.
+    """
+    oauth_provider._initialized = True  # already initialized while the secret was live
+    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",
+    )
+
+    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 registration is discarded, so 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)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_expired_and_issuer_mismatched_registration_still_gets_cross_issuer_cleanup(
+    oauth_provider: OAuthClientProvider,
+):
+    """SEP-2352: the expiry discard must not shadow the issuer-binding cleanup.
+
+    A record that is both expired and bound to a different issuer (the server migrated
+    authorization servers while the secret lapsed) still needs the cross-issuer cleanup:
+    the old issuer's tokens are dropped so they can never be presented to the new AS, and
+    the old AS's cached metadata is dropped so a failed rediscovery cannot leak its
+    registration/token endpoints into Step 4. The expiry discard therefore runs only
+    after the issuer checks, which read the issuer stamp off the record.
+    """
+    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,
+        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+        token_endpoint_auth_method="client_secret_post",
+        issuer="https://old-as.example.com",
+    )
+    oauth_provider.context.current_tokens = OAuthToken(access_token="old-as-access", refresh_token="old-as-refresh")
+    oauth_provider.context.token_expiry_time = time.time() + 1800
+    oauth_provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://old-as.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://old-as.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://old-as.example.com/token"),
+        registration_endpoint=AnyHttpUrl("https://old-as.example.com/register"),
+    )
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → PRM discovery now advertises a different authorization server.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+
+    # ASM rediscovery for the new AS fails at both well-known URLs.
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_req = await auth_flow.asend(httpx2.Response(404, request=asm_req))
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/openid-configuration"
+    register_req = await auth_flow.asend(httpx2.Response(404, request=asm_req))
+
+    # The issuer-binding cleanup fired despite the lapsed secret: the old issuer's tokens
+    # and cached metadata are gone, so registration goes to the resource-origin fallback,
+    # not the old AS's registration endpoint.
+    assert oauth_provider.context.current_tokens is None
+    assert oauth_provider.context.oauth_metadata is None
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    await auth_flow.aclose()
+
+
+@pytest.mark.anyio
+async def test_refresh_is_skipped_when_registration_secret_expired_mid_session(
+    oauth_provider: OAuthClientProvider,
+):
+    """The refresh path never presents a lapsed secret.
+
+    ``can_refresh_token()`` ignores expiry, so without the extra check the refresh would
+    authenticate with the dead secret and fail ``invalid_client`` before recovering via
+    the subsequent 401. Instead the doomed refresh is skipped outright: the request goes
+    out unauthenticated and the 401 flow re-registers.
+    """
+    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="still-there")
+    oauth_provider.context.token_expiry_time = time.time() - 60  # access token expired → refresh territory
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # No refresh request was yielded: the first request out is the original one, unauthenticated.
+    assert request.method == "GET"
+    assert str(request.url) == "https://api.example.com/v1/mcp"
+    assert "Authorization" not in request.headers
+
+    # The 401 flow then recovers by re-registering.
+    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)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://api.example.com/register"
+    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
+):
+    """A 403 scope step-up with a lapsed secret re-registers instead of burning a consent.
+
+    On this path the access token is still live, so the 401 flow's discard can never run;
+    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; the dead client's
+    refresh token is dropped with its record.
+    """
+    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="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"),
+        authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        registration_endpoint=AnyHttpUrl("https://auth.example.com/register"),
+    )
+
+    captured_state: str | None = None
+    authorize_client_id: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state, authorize_client_id
+        params = parse_qs(urlparse(url).query)
+        authorize_client_id = params["client_id"][0]
+        captured_state = params.get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    oauth_provider.context.redirect_handler = capture_redirect
+    oauth_provider.context.callback_handler = mock_callback
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # The step-up re-registers before authorizing, using the cached registration endpoint.
+    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={
+            "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 interactive authorization and token exchange use the fresh credentials only.
+    token_exchange_request = await auth_flow.asend(register_response)
+    assert authorize_client_id == "fresh-client"
+    content = token_exchange_request.content.decode()
+    assert "client_id=fresh-client" in content
+    assert "client_secret=fresh-secret" in content
+    assert "expired-secret" not in content
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "fresh-client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_403_step_up_discovers_the_as_before_re_registering_after_a_restart(
+    oauth_provider: OAuthClientProvider, mock_storage: MockTokenStorage
+):
+    """A 403 step-up that must re-register with no AS metadata cached discovers it first.
+
+    After a restart `_initialize` restores tokens and client info but no AS metadata, and
+    the still-live access token keeps the 401 flow's discovery from ever running. Without
+    its own discovery the step-up would POST the registration document to the resource
+    origin's `/register` fallback — the wrong server in the separate-AS topology — so it
+    runs the 401 flow's discovery sequence first, honoring the 403's `resource_metadata`
+    pointer, and registers at the discovered endpoint.
+    """
+    await mock_storage.set_client_info(
+        OAuthClientInformationFull(
+            client_id="dead-client",
+            client_secret="expired-secret",
+            client_secret_expires_at=int(time.time()) - 3600,  # lapsed while stored
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
+            issuer="https://auth.example.com",
+        )
+    )
+    await mock_storage.set_tokens(
+        OAuthToken(access_token="live-token", refresh_token="issued-to-dead-client", scope="read")
+    )
+
+    captured_state: str | None = None
+    authorize_client_id: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state, authorize_client_id
+        params = parse_qs(urlparse(url).query)
+        authorize_client_id = params["client_id"][0]
+        captured_state = params.get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    oauth_provider.context.redirect_handler = capture_redirect
+    oauth_provider.context.callback_handler = mock_callback
+
+    # The restarted provider initializes from storage: the loaded token records no expiry,
+    # so it is presented as-is and the 401 flow never runs.
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    assert request.headers["Authorization"] == "Bearer live-token"
+
+    response_403 = httpx2.Response(
+        403,
+        headers={
+            "WWW-Authenticate": (
+                'Bearer error="insufficient_scope", scope="write", '
+                'resource_metadata="https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"'
+            )
+        },
+        request=request,
+    )
+
+    # The step-up discovers the PRM first, seeded from the 403's resource_metadata pointer.
+    prm_req = await auth_flow.asend(response_403)
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+
+    # Then the AS metadata.
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://auth.example.com", '
+            b'"authorization_endpoint": "https://auth.example.com/authorize", '
+            b'"token_endpoint": "https://auth.example.com/token", '
+            b'"registration_endpoint": "https://auth.example.com/register"}'
+        ),
+        request=asm_req,
+    )
+
+    # The registration goes to the discovered endpoint, not the resource origin's /register.
+    register_req = await auth_flow.asend(asm_response)
+    assert register_req.method == "POST"
+    assert str(register_req.url) == "https://auth.example.com/register"
+    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 step-up completes with the fresh credentials; the dead client's refresh token
+    # was dropped with its record.
+    token_exchange_request = await auth_flow.asend(register_response)
+    assert authorize_client_id == "fresh-client"
+    content = token_exchange_request.content.decode()
+    assert "client_id=fresh-client" in content
+    assert "expired-secret" not in content
+    assert oauth_provider.context.current_tokens is not None
+    assert oauth_provider.context.current_tokens.refresh_token is None
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_403_step_up_after_a_restart_resolves_cimd_once_metadata_is_discovered(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+):
+    """A 403 step-up with a configured CIMD URL and no cached AS metadata still uses CIMD.
+
+    Registering blind would silently bypass the CIMD capability the (undiscovered) AS
+    metadata advertises; the step-up's discovery restores it, so the dead record is
+    replaced locally by the URL-based client ID — no `/register` round trip at all.
+    """
+    captured_state: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state
+        captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+        redirect_handler=capture_redirect,
+        callback_handler=mock_callback,
+        client_metadata_url="https://example.com/client",
+    )
+    await mock_storage.set_client_info(
+        OAuthClientInformationFull(
+            client_id="dead-client",
+            client_secret="expired-secret",
+            client_secret_expires_at=int(time.time()) - 3600,  # lapsed while stored
+            redirect_uris=[AnyUrl("http://localhost:3030/callback")],
+            token_endpoint_auth_method="client_secret_post",
+            issuer="https://auth.example.com",
+        )
+    )
+    await mock_storage.set_tokens(OAuthToken(access_token="live-token", scope="read"))
+
+    auth_flow = provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    assert request.headers["Authorization"] == "Bearer live-token"
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # No resource_metadata pointer this time: discovery falls back to the well-known PRM URL.
+    prm_req = await auth_flow.asend(response_403)
+    assert str(prm_req.url) == "https://api.example.com/.well-known/oauth-protected-resource/v1/mcp"
+    prm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://api.example.com/v1/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+    asm_req = await auth_flow.asend(prm_response)
+    assert str(asm_req.url) == "https://auth.example.com/.well-known/oauth-authorization-server"
+    asm_response = httpx2.Response(
+        200,
+        content=(
+            b'{"issuer": "https://auth.example.com", '
+            b'"authorization_endpoint": "https://auth.example.com/authorize", '
+            b'"token_endpoint": "https://auth.example.com/token", '
+            b'"client_id_metadata_document_supported": true}'
+        ),
+        request=asm_req,
+    )
+
+    # No /register round trip: the next request is already the token exchange, and the
+    # dead record was replaced by the URL-based client ID.
+    token_exchange_request = await auth_flow.asend(asm_response)
+    assert str(token_exchange_request.url) == "https://auth.example.com/token"
+    content = token_exchange_request.content.decode()
+    assert "client_id=https%3A%2F%2Fexample.com%2Fclient" in content
+    assert "expired-secret" not in content
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "https://example.com/client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_403_step_up_resolves_cimd_when_registration_secret_expired(
+    client_metadata: OAuthClientMetadata, mock_storage: MockTokenStorage
+):
+    """A 403 step-up with a lapsed secret resolves the CIMD URL when the AS supports it.
+
+    Mirrors Step 4 of the 401 flow: with `client_id_metadata_document_supported` in the
+    cached AS metadata and a configured `client_metadata_url`, the dead registration is
+    replaced locally — no ``/register`` round trip — and the step-up proceeds straight to
+    authorization with the URL-based client ID (which has no secret to expire).
+    """
+    captured_state: str | None = None
+
+    async def capture_redirect(url: str) -> None:
+        nonlocal captured_state
+        captured_state = parse_qs(urlparse(url).query).get("state", [None])[0]
+
+    async def mock_callback() -> AuthorizationCodeResult:
+        return AuthorizationCodeResult(code="auth_code", state=captured_state)
+
+    provider = OAuthClientProvider(
+        server_url="https://api.example.com/v1/mcp",
+        client_metadata=client_metadata,
+        storage=mock_storage,
+        redirect_handler=capture_redirect,
+        callback_handler=mock_callback,
+        client_metadata_url="https://example.com/client",
+    )
+    provider._initialized = True
+    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",
+    )
+    provider.context.current_tokens = OAuthToken(access_token="live-token", scope="read")
+    provider.context.token_expiry_time = time.time() + 1800  # access token still live
+    provider.context.oauth_metadata = OAuthMetadata(
+        issuer=AnyHttpUrl("https://auth.example.com"),
+        authorization_endpoint=AnyHttpUrl("https://auth.example.com/authorize"),
+        token_endpoint=AnyHttpUrl("https://auth.example.com/token"),
+        client_id_metadata_document_supported=True,
+    )
+
+    auth_flow = provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # No /register round trip: the next request is already the token exchange, and the
+    # dead record was replaced by the URL-based client ID.
+    token_exchange_request = await auth_flow.asend(response_403)
+    assert str(token_exchange_request.url) == "https://auth.example.com/token"
+    content = token_exchange_request.content.decode()
+    assert "client_id=https%3A%2F%2Fexample.com%2Fclient" in content
+    assert "expired-secret" not in content
+    assert provider.context.client_info is not None
+    assert provider.context.client_info.token_endpoint_auth_method == "none"
+    stored = await mock_storage.get_client_info()
+    assert stored is not None
+    assert stored.client_id == "https://example.com/client"  # dead record overwritten in storage
+
+    token_response = httpx2.Response(
+        200,
+        json={"access_token": "stepped-up", "token_type": "Bearer", "expires_in": 3600, "scope": "read write"},
+        request=token_exchange_request,
+    )
+    final_request = await auth_flow.asend(token_response)
+    assert final_request.headers["Authorization"] == "Bearer stepped-up"
+    try:
+        await auth_flow.asend(httpx2.Response(200, request=final_request))
+    except StopAsyncIteration:
+        pass
+
+
+@pytest.mark.anyio
+async def test_closing_the_flow_mid_discovery_finalizes_the_401_discovery_sub_generator(
+    oauth_provider: OAuthClientProvider,
+):
+    """httpx2's `_send_handling_auth` acloses the auth flow when a discovery request
+    fails at the transport level (or the request is cancelled), throwing `GeneratorExit`
+    at the 401 relay's `yield`; the relay must finalize the discovery sub-generator with
+    the flow rather than abandon it suspended to GC (a `ResourceWarning` under trio).
+    The sub-generator is captured via a wrapper because its finalization is not
+    observable through the auth-flow protocol itself.
+    """
+    captured: list[AsyncGenerator[httpx2.Request, httpx2.Response]] = []
+    original = oauth_provider._discover_authorization_server_metadata
+
+    def capturing(response: httpx2.Response) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        discovery = original(response)
+        captured.append(discovery)
+        return discovery
+
+    oauth_provider._discover_authorization_server_metadata = capturing
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    # 401 → the flow yields the first discovery request, suspending both generators mid-relay.
+    prm_req = await auth_flow.asend(httpx2.Response(401, request=request))
+    assert "oauth-protected-resource" in str(prm_req.url)
+    assert len(captured) == 1
+
+    # Closing the flow here mirrors httpx2 aborting it on a transport error.
+    await auth_flow.aclose()
+
+    # The sub-generator was closed with the flow: probing it reports exhaustion instead
+    # of resuming a suspended frame.
+    with pytest.raises(StopAsyncIteration):
+        await captured[0].__anext__()
+
+
+@pytest.mark.anyio
+async def test_closing_the_flow_mid_discovery_finalizes_the_403_step_up_discovery_sub_generator(
+    oauth_provider: OAuthClientProvider,
+):
+    """The 403 step-up's discovery relay must finalize its sub-generator when httpx2
+    acloses the flow mid-discovery, exactly like the 401 relay (same abandonment
+    hazard, same fix); the sub-generator is captured via a wrapper because its
+    finalization is not observable through the auth-flow protocol itself.
+    """
+    captured: list[AsyncGenerator[httpx2.Request, httpx2.Response]] = []
+    original = oauth_provider._discover_authorization_server_metadata
+
+    def capturing(response: httpx2.Response) -> AsyncGenerator[httpx2.Request, httpx2.Response]:
+        discovery = original(response)
+        captured.append(discovery)
+        return discovery
+
+    oauth_provider._discover_authorization_server_metadata = capturing
+    # Restart shape: a live token and no stored registration, so the 403 step-up must
+    # discover before registering (`oauth_metadata` is never restored by `_initialize`).
+    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token")
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+
+    # 403 → the step-up yields the first discovery request, suspending both generators mid-relay.
+    prm_req = await auth_flow.asend(response_403)
+    assert "oauth-protected-resource" in str(prm_req.url)
+    assert len(captured) == 1
+
+    await auth_flow.aclose()
+
+    with pytest.raises(StopAsyncIteration):
+        await captured[0].__anext__()
+
+
+@pytest.mark.anyio
+async def test_403_step_up_surfaces_oauth_flow_errors_to_the_caller(
+    oauth_provider: OAuthClientProvider,
+):
+    """An `OAuthFlowError` raised inside the 403 step-up — here the SEP-985/RFC 8707
+    resource-identity check failing during the step-up's discovery — propagates to the
+    caller instead of being swallowed, matching the 401 branch's error contract.
+    """
+    # Restart shape: a live token and no stored registration, so the step-up discovers.
+    oauth_provider.context.current_tokens = OAuthToken(access_token="live-token")
+    oauth_provider._initialized = True
+
+    auth_flow = oauth_provider.async_auth_flow(httpx2.Request("GET", "https://api.example.com/v1/mcp"))
+    request = await auth_flow.__anext__()
+
+    response_403 = httpx2.Response(
+        403,
+        headers={"WWW-Authenticate": 'Bearer error="insufficient_scope", scope="write"'},
+        request=request,
+    )
+    prm_req = await auth_flow.asend(response_403)
+
+    # The PRM advertises a different resource, so the RFC 8707 identity check must fail.
+    mismatched_prm = httpx2.Response(
+        200,
+        content=(
+            b'{"resource": "https://other.example.com/mcp", "authorization_servers": ["https://auth.example.com"]}'
+        ),
+        request=prm_req,
+    )
+    with pytest.raises(OAuthFlowError):
+        await auth_flow.asend(mismatched_prm)