diff --git a/packages/google-auth/google/auth/aio/transport/sessions.py b/packages/google-auth/google/auth/aio/transport/sessions.py
index d88162667bda..91bb6c3073b3 100644
--- a/packages/google-auth/google/auth/aio/transport/sessions.py
+++ b/packages/google-auth/google/auth/aio/transport/sessions.py
@@ -13,10 +13,14 @@
 # limitations under the License.
 
 import asyncio
+import collections.abc
 from contextlib import asynccontextmanager
 import functools
+import http.client as http_client
+import logging
 import time
 from typing import Mapping, Optional, TYPE_CHECKING, Union
+import urllib.parse
 import warnings
 
 from google.auth import _exponential_backoff, exceptions
@@ -37,6 +41,9 @@
     except (ImportError, AttributeError):
         ClientTimeout = None
 
+_LOGGER = logging.getLogger(__name__)
+MTLS_URL_PREFIXES = ["mtls.googleapis.com", "mtls.sandbox.googleapis.com"]
+
 
 # Tracks the internal aiohttp installation and usage
 try:
@@ -66,6 +73,8 @@ async def timeout_guard(timeout):
     total_timeout = timeout
 
     def _remaining_time():
+        if total_timeout is None:
+            return None
         elapsed = time.monotonic() - start
         remaining = total_timeout - elapsed
         if remaining <= 0:
@@ -143,11 +152,15 @@ def __init__(
         self._is_mtls = False
         self._mtls_init_task = None
         self._cached_cert = None
+        self._client_cert_callback = None
+        self._old_auth_requests: list[transport.Request] = []
         if _auth_request is None:
             raise exceptions.TransportError(
                 "`auth_request` must either be configured or the external package `aiohttp` must be installed to use the default value."
             )
         self._auth_request = _auth_request
+        self._mtls_rotation_lock = None  # type: Optional[asyncio.Lock]
+        self._mtls_check_counter = 0
 
     async def configure_mtls_channel(self, client_cert_callback=None):
         """Configure the client certificate and key for SSL connection.
@@ -175,6 +188,7 @@ async def configure_mtls_channel(self, client_cert_callback=None):
                 creation failed for any reason.
         """
         if self._mtls_init_task is None:
+            self._client_cert_callback = client_cert_callback
 
             async def _do_configure():
                 # Run the blocking check in an executor
@@ -204,12 +218,8 @@ async def _do_configure():
 
                             old_auth_request = self._auth_request
                             self._auth_request = AiohttpRequest(session=new_session)
+                            self._old_auth_requests.append(old_auth_request)
 
-                            try:
-                                await old_auth_request.close()
-                            except Exception:
-                                # Suppress so it doesn't abort the mTLS configuration
-                                pass
                         else:
                             is_mtls = False
                             warnings.warn(
@@ -277,7 +287,10 @@ async def request(
                 google.auth.exceptions.TimeoutError: If the method does not complete within
                 the configured `max_allowed_time` or the request exceeds the configured
                 `timeout`.
+                google.auth.exceptions.MutualTLSChannelError: If mutual TLS
+                channel reconfiguration fails for any reason during certificate rotation.
         """
+        _auth_retry_count = kwargs.pop("_auth_retry_count", 0)
         if self._mtls_init_task:
             try:
                 await self._mtls_init_task
@@ -290,6 +303,7 @@ async def request(
         )
         if headers is None:
             headers = {}
+        start_time = time.monotonic()
         async with timeout_guard(max_allowed_time) as with_timeout:
             await with_timeout(
                 # Note: before_request will attempt to refresh credentials if expired.
@@ -310,8 +324,166 @@ async def request(
                         url, method, data, headers, actual_timeout, **kwargs
                     )
                 )
+
                 if response.status_code not in transport.DEFAULT_RETRYABLE_STATUS_CODES:
                     break
+
+        if response.status_code == http_client.UNAUTHORIZED:
+            if _auth_retry_count < 2:
+                if max_allowed_time is not None:
+                    elapsed = time.monotonic() - start_time
+                    remaining_time = max(0.0, max_allowed_time - elapsed)
+                    if remaining_time == 0.0:
+                        raise google.auth.exceptions.TimeoutError(
+                            "Timeout exceeded before credential refresh could begin"
+                        )
+                else:
+                    remaining_time = None
+                is_streaming = data is not None and (
+                    isinstance(
+                        data, (collections.abc.Iterator, collections.abc.AsyncIterable)
+                    )
+                    or hasattr(data, "read")
+                )
+
+                async def _recover_auth_state():
+                    is_mtls_endpoint = False
+                    if getattr(self, "is_mtls", False):
+                        hostname = urllib.parse.urlsplit(url).hostname
+                        if hostname:
+                            is_mtls_endpoint = any(
+                                hostname == prefix or hostname.endswith("." + prefix)
+                                for prefix in MTLS_URL_PREFIXES
+                            )
+                        # Snapshot the stale certificate state BEFORE acquiring the lock.
+                        # This represents the cert that caused the 401 rejection.
+                        if is_mtls_endpoint:
+                            if self._mtls_rotation_lock is None:
+                                self._mtls_rotation_lock = asyncio.Lock()
+                            # Snapshot the counter state BEFORE acquiring the lock.
+                            check_counter_at_error = self._mtls_check_counter
+
+                            async with self._mtls_rotation_lock:
+                                # Check if another coroutine already reconfigured mTLS or
+                                # ran the validation check.
+                                if self._mtls_check_counter > check_counter_at_error:
+                                    pass
+                                else:
+                                    try:
+                                        (
+                                            call_cert_bytes,
+                                            call_key_bytes,
+                                            cached_fingerprint,
+                                            current_cert_fingerprint,
+                                        ) = await mtls._run_in_executor(
+                                            google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response,
+                                            self._cached_cert,
+                                            self._client_cert_callback,
+                                        )
+                                    except (
+                                        exceptions.ClientCertError,
+                                        exceptions.MutualTLSChannelError,
+                                        OSError,
+                                        ValueError,
+                                        ImportError,
+                                    ) as e:
+                                        _LOGGER.warning(
+                                            "Failed to check client certificate parameters: %s. Proceeding with original response.",
+                                            e,
+                                        )
+                                        return response
+                                    else:
+                                        if (
+                                            cached_fingerprint
+                                            != current_cert_fingerprint
+                                        ):
+                                            try:
+                                                _LOGGER.info(
+                                                    "Client certificate has changed, reconfiguring mTLS "
+                                                    "channel."
+                                                )
+                                                if (
+                                                    self._mtls_init_task
+                                                    and self._mtls_init_task.done()
+                                                ):
+                                                    self._mtls_init_task = None
+                                                await self.configure_mtls_channel(
+                                                    self._client_cert_callback
+                                                )
+                                            except Exception as e:
+                                                _LOGGER.error(
+                                                    "Failed to reconfigure mTLS channel: %s",
+                                                    e,
+                                                )
+                                                if hasattr(response, "close"):
+                                                    if asyncio.iscoroutinefunction(
+                                                        response.close
+                                                    ):
+                                                        await response.close()
+                                                    else:
+                                                        response.close()
+                                                raise exceptions.MutualTLSChannelError(
+                                                    "Failed to reconfigure mTLS channel"
+                                                ) from e
+                                        else:
+                                            _LOGGER.info(
+                                                "Skipping reconfiguration of mTLS channel because the client"
+                                                " certificate has not changed."
+                                            )
+                                    finally:
+                                        # Always increment so waiting tasks skip the check block
+                                        self._mtls_check_counter += 1
+                    if is_streaming:
+                        return response
+                    try:
+                        await self._credentials.refresh(self._auth_request)
+                    except (
+                        exceptions.RefreshError,
+                        getattr(exceptions, "InvalidOperation", Exception),
+                    ) as e:
+                        _LOGGER.debug(
+                            "Credential refresh failed, returning 401 response. Error: %s",
+                            e,
+                        )
+                        return response
+
+                    # Return None to explicitly signal successful recovery
+                    return None
+
+                async with timeout_guard(remaining_time) as auth_with_timeout:
+                    early_return_response = await auth_with_timeout(
+                        _recover_auth_state()
+                    )
+                # If it returned a response (meaning streaming or error), bail out
+                if early_return_response is not None:
+                    return early_return_response
+
+                if hasattr(response, "close"):
+                    if asyncio.iscoroutinefunction(response.close):
+                        await response.close()
+                    else:
+                        response.close()
+
+                if max_allowed_time is not None:
+                    remaining_time = max(
+                        0.0, max_allowed_time - (time.monotonic() - start_time)
+                    )
+                    if remaining_time == 0.0:
+                        raise google.auth.exceptions.TimeoutError(
+                            "Timeout exceeded before retrying the request"
+                        )
+
+                kwargs["_auth_retry_count"] = _auth_retry_count + 1
+                return await self.request(
+                    method,
+                    url,
+                    data=data,
+                    headers=headers,
+                    max_allowed_time=remaining_time,
+                    timeout=timeout,
+                    total_attempts=total_attempts,
+                    **kwargs,
+                )
         return response
 
     @functools.wraps(request)
@@ -594,4 +766,12 @@ async def close(self) -> None:
                 await self._mtls_init_task
             except asyncio.CancelledError:
                 pass
-        await self._auth_request.close()
+        try:
+            await self._auth_request.close()
+        finally:
+            for old_request in self._old_auth_requests:
+                try:
+                    await old_request.close()
+                except Exception:
+                    pass
+            self._old_auth_requests.clear()
diff --git a/packages/google-auth/google/auth/transport/_mtls_helper.py b/packages/google-auth/google/auth/transport/_mtls_helper.py
index 7779c484c713..b36742f15455 100644
--- a/packages/google-auth/google/auth/transport/_mtls_helper.py
+++ b/packages/google-auth/google/auth/transport/_mtls_helper.py
@@ -808,11 +808,13 @@ def check_use_client_cert():
     return False
 
 
-def check_parameters_for_unauthorized_response(cached_cert):
+def check_parameters_for_unauthorized_response(cached_cert, client_cert_callback=None):
     """Returns the cached and current cert fingerprint for reconfiguring mTLS.
 
     Args:
         cached_cert(bytes): The cached client certificate.
+        client_cert_callback(Optional[Callable[[], (bytes, bytes)]]):
+            The optional callback that returns the client certificate and private key bytes.
 
     Returns:
         bytes: The client callback cert bytes.
@@ -820,7 +822,10 @@ def check_parameters_for_unauthorized_response(cached_cert):
         str: The base64-encoded SHA256 cached fingerprint.
         str: The base64-encoded SHA256 current cert fingerprint.
     """
-    call_cert_bytes, call_key_bytes = call_client_cert_callback()
+    if client_cert_callback:
+        call_cert_bytes, call_key_bytes = client_cert_callback()
+    else:
+        call_cert_bytes, call_key_bytes = call_client_cert_callback()
     cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
     current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
         cert_obj
diff --git a/packages/google-auth/tests/transport/aio/test_sessions.py b/packages/google-auth/tests/transport/aio/test_sessions.py
index de283b7b2e7f..ecdb60e4af5a 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions.py
@@ -105,7 +105,9 @@ async def test_timeout_with_simple_async_task_within_bounds(
         self, simple_async_task
     ):
         task = False
-        with patch("time.monotonic", side_effect=[0, 0.25, 0.75]):
+        with patch(
+            "time.monotonic", side_effect=lambda it=iter([0, 0.25]): next(it, 0.75)
+        ):
             with patch("asyncio.wait_for", lambda coro, _: coro):
                 async with self.make_timeout_guard(
                     timeout=self.default_timeout
@@ -255,7 +257,7 @@ async def test_request_raises_transport_error(self):
     async def test_request_max_allowed_time_exceeded_error(self):
         auth_request = MockRequest(side_effect=TransportError)
         authed_session = sessions.AsyncAuthorizedSession(self.credentials, auth_request)
-        with patch("time.monotonic", side_effect=[0, 1, 1]):
+        with patch("time.monotonic", side_effect=[0, 0] + [2] * 10):
             with pytest.raises(TimeoutError):
                 await authed_session.request("GET", self.TEST_URL, max_allowed_time=1)
 
diff --git a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
index b68766ca5b5d..b3e458d338b5 100644
--- a/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
+++ b/packages/google-auth/tests/transport/aio/test_sessions_mtls.py
@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
+import http.client as http_client
 import json
 import os
 import ssl
@@ -344,3 +346,453 @@ async def test_configure_mtls_channel_close_exception_does_not_abort(self):
             assert session._is_mtls is True
             assert session._cached_cert == b"fake_cert_data"
             await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_failure_raises_error(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        new_cert = b"new_cert"
+        new_key = b"new_key"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp")
+            mock_conf.side_effect = Exception("Failed to reconfigure")
+
+            with pytest.raises(exceptions.MutualTLSChannelError):
+                await session.request("GET", "https://pubsub.mtls.googleapis.com/test")
+
+            mock_check.assert_called_once()
+            mock_conf.assert_called_once()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_check_params_fails(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock()
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.side_effect = exceptions.MutualTLSChannelError(
+                "Failed to check params"
+            )
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+
+            assert resp == mock_resp
+            mock_check.assert_called_once()
+            mock_creds.refresh.assert_not_called()
+            mock_conf.assert_not_called()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_no_cert_rotation_when_cert_matches_and_mtls_enabled(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        new_cert = b"new_cert"
+        new_key = b"new_key"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            # Matching fingerprints mean no layout rotation is needed
+            mock_check.return_value = (new_cert, new_key, b"old_fp", b"old_fp")
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+
+            assert resp == mock_resp
+            assert mock_check.call_count >= 1
+            mock_conf.assert_not_called()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_success_and_retry(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        # Initial request fails natively with 401. Retry succeeds with 200.
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+
+        # Use side_effect to dynamically yield responses
+        mock_auth_req = mock.AsyncMock(side_effect=[mock_resp_401, mock_resp_200])
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        new_cert = b"new_cert"
+        new_key = b"new_key"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.return_value = (new_cert, new_key, b"old_fp", b"new_fp")
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+
+            # 1. Assert the retried 200 response is successfully returned to the user
+            assert resp == mock_resp_200
+
+            # 2. Assert rotation logic correctly executed
+            mock_check.assert_called_once()
+            mock_conf.assert_called_once_with(mock.ANY)
+
+            # 3. Assert credentials were explicitly refreshed
+            mock_creds.refresh.assert_called_once()
+
+            # 4. Assert headers were explicitly rebound on the recursive retry (2 invocations)
+            assert mock_creds.before_request.call_count == 2
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_lock_contention(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp_401)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        new_cert = b"new_cert"
+        new_key = b"new_key"
+
+        async def mock_configure_mtls_channel(*args, **kwargs):
+            # Introduce a small delay to guarantee lock contention from asyncio.gather tasks
+            await asyncio.sleep(0.01)
+            # Simulate the channel update to ensure following tasks take the skip branch
+            session._cached_cert = new_cert
+
+        def mock_check_side_effect(cached_cert, callback):
+            # Return new fingerprints on first check, but matching fingerprints on retries
+            if cached_cert == b"old_cert":
+                return (new_cert, new_key, b"old_fp", b"new_fp")
+            return (new_cert, new_key, b"new_fp", b"new_fp")
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.side_effect = mock_check_side_effect
+            mock_conf.side_effect = mock_configure_mtls_channel
+
+            # Launch multiple concurrent requests triggering 401s
+            tasks = [
+                session.request("GET", "https://pubsub.mtls.googleapis.com/test")
+                for _ in range(3)
+            ]
+            responses = await asyncio.gather(*tasks)
+
+            for resp in responses:
+                assert resp == mock_resp_401
+
+            # Confirm the channel is only reconfigured once across all concurrent requests
+            mock_conf.assert_called_once()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_lock_contention_no_cert_change(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_resp_200 = mock.Mock()
+        mock_resp_200.status_code = http_client.OK
+
+        # Return 401 for the first 3 requests, then 200 for the retries.
+        mock_auth_req = mock.AsyncMock(
+            side_effect=[mock_resp_401] * 3 + [mock_resp_200] * 3
+        )
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        def mock_check_side_effect(cached_cert, callback):
+            import time
+
+            # Introduce a small delay to guarantee lock contention from asyncio.gather tasks
+            time.sleep(0.01)
+            # Return matching fingerprints to skip reconfiguration
+            return (b"old_cert", b"old_key", b"old_fp", b"old_fp")
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.side_effect = mock_check_side_effect
+
+            # Launch multiple concurrent requests triggering 401s
+            tasks = [
+                session.request("GET", "https://pubsub.mtls.googleapis.com/test")
+                for _ in range(3)
+            ]
+            responses = await asyncio.gather(*tasks)
+
+            for resp in responses:
+                assert resp == mock_resp_200
+
+            # Confirm that the cert parameters were only checked once across the concurrent requests
+            mock_check.assert_called_once()
+            # Because fingerprints match, reconfiguration should not happen
+            mock_conf.assert_not_called()
+            # Credentials should still be refreshed for each task as they fall back to normal refresh logic
+            assert mock_creds.refresh.call_count == 3
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_non_mtls_url_bypasses_rotation(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp_401 = mock.Mock()
+        mock_resp_401.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp_401)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+
+        # Even if mTLS is enabled globally...
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            # Will attempt to refresh credentials and retry the request up to 2 times (3 requests total).
+            resp = await session.request("GET", "https://pubsub.googleapis.com/test")
+
+            assert resp == mock_resp_401
+            mock_check.assert_not_called()
+            mock_conf.assert_not_called()
+
+            # Verify the standard retry behavior executed
+            assert mock_creds.refresh.call_count == 2
+            assert mock_auth_req.call_count == 3
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_skips_retry_for_streaming(self):
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock()
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        # Simulate a streaming object (hasattr(data, 'read'))
+        class MockStream:
+            def read(self):
+                pass
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ) as mock_conf:
+            mock_check.return_value = (b"new", b"new", b"old_fp", b"new_fp")
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test", data=MockStream()
+            )
+
+            assert resp == mock_resp
+            mock_conf.assert_called_once()
+            # Because it is streaming, it skips credentials refresh and retry
+            mock_creds.refresh.assert_not_called()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_credential_refresh_fails(self):
+        """Covers the except block for RefreshError when credentials fail to refresh."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        # Simulate a credentials refresh failure
+        mock_creds.refresh = mock.AsyncMock(
+            side_effect=exceptions.RefreshError("Refresh failed")
+        )
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ):
+            mock_check.return_value = (b"new", b"new", b"old_fp", b"new_fp")
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+
+            # It should catch the RefreshError and return the 401 response directly
+            assert resp == mock_resp
+            mock_creds.refresh.assert_called_once()
+            # Ensure it didn't retry the request by checking auth_request was only called once
+            mock_auth_req.assert_called_once()
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_cert_rotation_max_retries_exceeded(self):
+        """Covers the `if _auth_retry_count < 2:` max retry limit."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        mock_creds.before_request = mock.AsyncMock(return_value=None)
+        mock_creds.refresh = mock.AsyncMock(return_value=None)
+
+        mock_resp = mock.Mock()
+        mock_resp.status_code = http_client.UNAUTHORIZED
+        mock_resp.close = mock.AsyncMock()
+        # Always return 401
+        mock_auth_req = mock.AsyncMock(return_value=mock_resp)
+
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock_auth_req
+        )
+        session._is_mtls = True
+        session._cached_cert = b"old_cert"
+
+        with mock.patch(
+            "google.auth.transport._mtls_helper.check_parameters_for_unauthorized_response"
+        ) as mock_check, mock.patch.object(
+            session, "configure_mtls_channel", new_callable=mock.AsyncMock
+        ):
+            # Yield new fingerprints to trigger reconfiguration branches
+            mock_check.return_value = (b"new", b"new", b"old_fp", b"new_fp")
+
+            resp = await session.request(
+                "GET", "https://pubsub.mtls.googleapis.com/test"
+            )
+
+            assert resp == mock_resp
+            # Expecting exactly 3 requests (Initial try, Retry 1, Retry 2).
+            # When _auth_retry_count reaches 2, it drops out of the retry condition.
+            assert mock_auth_req.call_count == 3
+            # It should have checked the cert twice before hitting the retry cap
+            assert mock_check.call_count == 2
+            # Verify the stale response is closed before the retry
+            assert mock_resp.close.call_count == 2
+
+        await session.close()
+
+    @pytest.mark.asyncio
+    async def test_session_close_cleans_old_auth_requests(self):
+        """Covers the loop in the `close()` method that drains `_old_auth_requests`."""
+        mock_creds = mock.AsyncMock(spec=credentials.Credentials)
+        session = sessions.AsyncAuthorizedSession(
+            mock_creds, auth_request=mock.AsyncMock()
+        )
+
+        # Manually inject mocked old requests simulating past channel reconfigurations
+        mock_old_req_1 = mock.AsyncMock()
+        mock_old_req_2 = mock.AsyncMock()
+        mock_old_req_3_fails = mock.AsyncMock()
+        mock_old_req_3_fails.close.side_effect = Exception("Close error")
+
+        session._old_auth_requests.extend(
+            [mock_old_req_1, mock_old_req_2, mock_old_req_3_fails]
+        )
+
+        # Triggers `await old_request.close()` for each item
+        await session.close()
+
+        # Ensure all were called
+        mock_old_req_1.close.assert_called_once()
+        mock_old_req_2.close.assert_called_once()
+        mock_old_req_3_fails.close.assert_called_once()
+        # Ensure the list was cleared
+        assert len(session._old_auth_requests) == 0