@@ -38,6 +38,7 @@
 from .exchanger.base_credential_exchanger import BaseCredentialExchanger
 from .exchanger.credential_exchanger_registry import CredentialExchangerRegistry
 from .oauth2_discovery import OAuth2DiscoveryManager
+from .refresher.base_credential_refresher import _CredentialRevokedError
 from .refresher.credential_refresher_registry import CredentialRefresherRegistry
 
 logger = logging.getLogger("google_adk." + __name__)
@@ -331,7 +332,53 @@ async def get_auth_credential(
       # Step 7: Refresh credential if expired
       was_refreshed = False
       if not was_exchanged:
-        credential, was_refreshed = await self._refresh_credential(credential)
+        try:
+          credential, was_refreshed = await self._refresh_credential(credential)
+        except _CredentialRevokedError:
+          logger.warning(
+              "The refresh token for auth scheme %s was rejected.",
+              type(self._auth_config.auth_scheme).__name__,
+          )
+          if was_from_auth_response:
+            return None
+          revoked_credential = credential
+          # Evict the dead stored credential so later turns load the user's
+          # new authorization instead of refreshing it again.
+          try:
+            await self._save_credential(context, None)
+          except Exception:  # pylint: disable=broad-exception-caught
+            # A failed eviction only costs one more rejected refresh next turn.
+            logger.warning(
+                "Failed to remove the revoked credential for auth scheme %s.",
+                type(self._auth_config.auth_scheme).__name__,
+                exc_info=True,
+            )
+          # A stored dead credential must not shadow a fresh auth response.
+          credential = await self._load_from_auth_response(context)
+          revoked_oauth2 = revoked_credential.oauth2
+          new_oauth2 = credential.oauth2 if credential else None
+          if (
+              not credential
+              or credential == revoked_credential
+              or (
+                  revoked_oauth2
+                  and new_oauth2
+                  and revoked_oauth2.refresh_token
+                  and new_oauth2.refresh_token == revoked_oauth2.refresh_token
+              )
+          ):
+            return None
+          was_from_auth_response = True
+          credential, was_exchanged = await self._exchange_credential(
+              credential
+          )
+          if not was_exchanged:
+            try:
+              credential, was_refreshed = await self._refresh_credential(
+                  credential
+              )
+            except _CredentialRevokedError:
+              return None
 
       if _has_unredeemed_auth_code(credential):
         logger.warning(
@@ -432,9 +479,24 @@ async def _refresh_credential(
         if was_exchanged:
           return new_credential, True
         return credential, False
-      refreshed_credential = await refresher.refresh(
-          credential, self._auth_config.auth_scheme
-      )
+      try:
+        refreshed_credential = await refresher.refresh(
+            credential, self._auth_config.auth_scheme
+        )
+      except _CredentialRevokedError:
+        if (
+            raw_auth_credential is None
+            or not self._is_client_credentials_flow()
+        ):
+          raise
+        # A client credentials flow can get a new token without the user. Copy
+        # the raw credential so the exchange does not mutate the shared config.
+        new_credential, was_exchanged = await self._exchange_credential(
+            raw_auth_credential.model_copy(deep=True)
+        )
+        if not was_exchanged:
+          raise
+        return new_credential, True
       return refreshed_credential, True
 
     return credential, False
@@ -488,9 +550,9 @@ async def _validate_credential(self) -> None:
     # Additional validation can be added here
 
   async def _save_credential(
-      self, context: CallbackContext, credential: AuthCredential
+      self, context: CallbackContext, credential: AuthCredential | None
   ) -> None:
-    """Save credential to credential service if available."""
+    """Save credential to credential service if available; None removes it."""
     credential_service = context._invocation_context.credential_service
     if credential_service:
       auth_config_to_save = self._auth_config.model_copy(deep=True)
 
@@ -63,6 +63,9 @@ async def save_credential(
     Saves the exchanged_auth_credential in auth config to the backend credential
     store.
 
+    An exchanged_auth_credential of None means the stored credential must be
+    removed, so that load_credential returns None for that key afterwards.
+
     Args:
         auth_config: The auth config which contains the auth scheme and auth
         credential information. auth_config.get_credential_key will be used to
 
@@ -27,6 +27,14 @@ class CredentialRefresherError(Exception):
   """Base exception for credential refresh errors."""
 
 
+class _CredentialRevokedError(CredentialRefresherError):
+  """Raised when the refresh token was permanently rejected.
+
+  The credential can never be refreshed again, so only a new user
+  authorization can replace it.
+  """
+
+
 class BaseCredentialRefresher(abc.ABC):
   """Base interface for credential refreshers.
 
 
@@ -27,6 +27,7 @@
 import requests
 from typing_extensions import override
 
+from .base_credential_refresher import _CredentialRevokedError
 from .base_credential_refresher import BaseCredentialRefresher
 
 try:
@@ -80,7 +81,8 @@ async def refresh(
       auth_scheme: Optional[AuthScheme] = None,
   ) -> AuthCredential:
     """Refresh the OAuth2 credential.
-    If refresh failed, return the original credential.
+
+    If refresh failed transiently, return the original credential.
 
     Args:
         auth_credential: The OAuth2 credential to refresh.
@@ -89,6 +91,9 @@ async def refresh(
     Returns:
         The refreshed credential.
 
+    Raises:
+        CredentialRefresherError: If the authorization server rejected the
+            refresh token with `invalid_grant` (revoked or expired).
     """
 
     # Handle regular OAuth2 credentials
@@ -128,6 +133,12 @@ async def refresh(
           update_credential_with_tokens(auth_credential, tokens)
           logger.debug("Successfully refreshed OAuth2 tokens")
         except (AuthlibBaseError, requests.RequestException) as e:
+          if getattr(e, "error", None) == "invalid_grant":
+            # Permanent: the refresh token is dead, so a stale access token
+            # would only fail later; the caller must ask for re-authorization.
+            raise _CredentialRevokedError(
+                f"OAuth2 refresh token was rejected: {e}"
+            ) from e
           # Non-fatal: keep the stale token so its eventual 401
           # re-triggers auth.
           logger.error("Failed to refresh OAuth2 tokens: %s", e)
 
@@ -16,12 +16,15 @@
 from unittest.mock import Mock
 from unittest.mock import patch
 
+from authlib.integrations.base_client.errors import OAuthError
 from authlib.oauth2.rfc6749 import OAuth2Token
+from authlib.oauth2.rfc6749.errors import InvalidGrantError
 from authlib.oauth2.rfc6749.errors import OAuth2Error
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.auth.auth_credential import AuthCredentialTypes
 from google.adk.auth.auth_credential import OAuth2Auth
 from google.adk.auth.auth_schemes import OpenIdConnectWithConfig
+from google.adk.auth.refresher import base_credential_refresher
 from google.adk.auth.refresher.oauth2_credential_refresher import OAuth2CredentialRefresher
 import pytest
 import requests
@@ -195,7 +198,7 @@ async def test_refresh_oauth2_error_returns_original_and_logs(
     mock_client = Mock()
     mock_oauth2_session.return_value = mock_client
     mock_client.refresh_token.side_effect = OAuth2Error(
-        description="invalid_grant"
+        description="temporarily_unavailable"
     )
 
     scheme = OpenIdConnectWithConfig(
@@ -309,6 +312,94 @@ async def test_refresh_unexpected_error_propagates(
     with pytest.raises(ValueError, match="unexpected bug"):
       await refresher.refresh(credential, scheme)
 
+  @patch("google.adk.auth.oauth2_credential_util.OAuth2Session")
+  @patch("google.adk.auth.oauth2_credential_util.OAuth2Token")
+  @pytest.mark.parametrize(
+      "error",
+      [
+          InvalidGrantError(description="Token has been revoked."),
+          OAuth2Error(error="invalid_grant"),
+          OAuthError(error="invalid_grant", description="Bad Request"),
+      ],
+  )
+  @pytest.mark.asyncio
+  async def test_refresh_invalid_grant_raises_revoked(
+      self, mock_oauth2_token, mock_oauth2_session, error
+  ):
+    """invalid_grant means the refresh token is dead and must not be kept."""
+    mock_token_instance = Mock()
+    mock_token_instance.is_expired.return_value = True
+    mock_oauth2_token.return_value = mock_token_instance
+
+    mock_client = Mock()
+    mock_oauth2_session.return_value = mock_client
+    mock_client.refresh_token.side_effect = error
+
+    scheme = OpenIdConnectWithConfig(
+        type_="openIdConnect",
+        openId_connect_url=(
+            "https://example.com/.well-known/openid_configuration"
+        ),
+        authorization_endpoint="https://example.com/auth",
+        token_endpoint="https://example.com/token",
+        scopes=["openid"],
+    )
+    credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OPEN_ID_CONNECT,
+        oauth2=OAuth2Auth(
+            client_id="test_client_id",
+            client_secret="test_client_secret",
+            access_token="old_token",
+            refresh_token="old_refresh_token",
+            expires_at=int(time.time()) - 3600,  # Expired
+        ),
+    )
+
+    refresher = OAuth2CredentialRefresher()
+    with pytest.raises(base_credential_refresher._CredentialRevokedError):
+      await refresher.refresh(credential, scheme)
+
+  @patch("google.adk.auth.oauth2_credential_util.OAuth2Session")
+  @patch("google.adk.auth.oauth2_credential_util.OAuth2Token")
+  @pytest.mark.asyncio
+  async def test_refresh_other_oauth2_error_returns_original(
+      self, mock_oauth2_token, mock_oauth2_session
+  ):
+    """An OAuth2 error other than invalid_grant is transient: keep the token."""
+    mock_token_instance = Mock()
+    mock_token_instance.is_expired.return_value = True
+    mock_oauth2_token.return_value = mock_token_instance
+
+    mock_client = Mock()
+    mock_oauth2_session.return_value = mock_client
+    mock_client.refresh_token.side_effect = OAuth2Error(error="server_error")
+
+    scheme = OpenIdConnectWithConfig(
+        type_="openIdConnect",
+        openId_connect_url=(
+            "https://example.com/.well-known/openid_configuration"
+        ),
+        authorization_endpoint="https://example.com/auth",
+        token_endpoint="https://example.com/token",
+        scopes=["openid"],
+    )
+    credential = AuthCredential(
+        auth_type=AuthCredentialTypes.OPEN_ID_CONNECT,
+        oauth2=OAuth2Auth(
+            client_id="test_client_id",
+            client_secret="test_client_secret",
+            access_token="old_token",
+            refresh_token="old_refresh_token",
+            expires_at=int(time.time()) - 3600,  # Expired
+        ),
+    )
+
+    refresher = OAuth2CredentialRefresher()
+    result = await refresher.refresh(credential, scheme)
+
+    assert result is credential
+    assert result.oauth2.access_token == "old_token"
+
   @patch("google.adk.auth.refresher.oauth2_credential_refresher.logger")
   @pytest.mark.asyncio
   async def test_refresh_without_refresh_token_returns_original_and_logs(