fix(oauth): let Google refresh path run instead of forcing hourly reconsent

backend/src/agents/main_agent/base_agent.py

@@ -340,26 +340,13 @@ async def disconnected_lookup(provider_id: str) -> bool:
                 self.user_id, provider_id
             )
 
-        async def mark_disconnected(provider_id: str) -> None:
-            # Persist a disconnect from the AfterToolCallEvent 401-retry
-            # path so subsequent requests (potentially on other replicas)
-            # also force a fresh consent.
-            from apis.shared.oauth.disconnect_repository import (
-                get_disconnect_repository,
-            )
-
-            await get_disconnect_repository().mark_disconnected(
-                self.user_id, provider_id
-            )
-
         return OAuthConsentHook(
             user_id=self.user_id,
             provider_lookup=provider_lookup,
             scopes_lookup=scopes_lookup,
             provider_type_lookup=provider_type_lookup,
             custom_parameters_lookup=custom_parameters_lookup,
             disconnected_lookup=disconnected_lookup,
-            mark_disconnected=mark_disconnected,
         )
 
     def _build_filtered_tools(self) -> List:

backend/src/agents/main_agent/integrations/oauth_token_cache.py

@@ -9,33 +9,66 @@
   * `OAuthConsentHook` warms the cache after a successful vault lookup, or
   * the resume path re-fetches a token after the user completes consent.
 
-Tokens are evicted explicitly via `clear_user_provider` when consent is
-revoked or expires; we don't track expiry locally because AgentCore
-Identity owns refresh.
+Each entry has a TTL — by default a little under the upstream token's
+lifetime (Google's access tokens are 3600s; we expire locally at 3000s).
+On expiry `get` returns None so `OAuthConsentHook._gate` re-asks AgentCore
+Identity, which transparently refreshes the access token using the
+refresh_token in the vault. Without a TTL the cache would hand out an
+expired token until the upstream MCP server returned 401, and the 401
+retry path would force a full re-consent — i.e. the user reconnects
+hourly even though refresh would have worked.
 
-Disconnect intent ("user pressed Disconnect" / "tool returned 401") is *not*
-held here — it lives in the DDB-backed `OAuthDisconnectRepository` so it's
-visible across replicas. The cache only holds tokens.
+Tokens are evicted explicitly via `clear_user_provider` when consent is
+revoked. Disconnect intent ("user pressed Disconnect") is *not* held here
+— it lives in the DDB-backed `OAuthDisconnectRepository` so it's visible
+across replicas. The cache only holds tokens.
 """
 
 from __future__ import annotations
 
 import threading
+import time
 from typing import Optional
 
+# Default TTL for cached access tokens. Google access tokens last 3600s;
+# expiring locally at 3000s gives us a 10-minute safety margin so we
+# refresh before the upstream rejects the token. Most other providers
+# (Microsoft, GitHub) use the same or longer lifetimes, so this is a safe
+# floor across the board.
+DEFAULT_TTL_SECONDS = 3000
 
 _lock = threading.Lock()
-_cache: dict[tuple[str, str], str] = {}
+# (user_id, provider_id) -> (token, expires_at_monotonic)
+_cache: dict[tuple[str, str], tuple[str, float]] = {}
 
 
 def get(user_id: str, provider_id: str) -> Optional[str]:
-    with _lock:
-        return _cache.get((user_id, provider_id))
+    """Return the cached token if present and not expired, else None.
 
-
-def set(user_id: str, provider_id: str, token: str) -> None:
+    Expired entries are evicted on access; we don't run a sweeper because
+    the cache is small (one entry per (user, provider) pair) and a stale
+    entry that's never read costs nothing.
+    """
+    with _lock:
+        entry = _cache.get((user_id, provider_id))
+        if entry is None:
+            return None
+        token, expires_at = entry
+        if time.monotonic() >= expires_at:
+            del _cache[(user_id, provider_id)]
+            return None
+        return token
+
+
+def set(
+    user_id: str,
+    provider_id: str,
+    token: str,
+    *,
+    ttl_seconds: float = DEFAULT_TTL_SECONDS,
+) -> None:
     with _lock:
-        _cache[(user_id, provider_id)] = token
+        _cache[(user_id, provider_id)] = (token, time.monotonic() + ttl_seconds)
 
 
 def clear_user_provider(user_id: str, provider_id: str) -> None:

backend/src/agents/main_agent/session/hooks/oauth_consent.py

@@ -130,12 +130,6 @@ def _looks_like_auth_failure(tool_result: Any) -> bool:
 # fresh consent URL with `force_authentication=True`.
 DisconnectedLookup = Callable[[str], Union[bool, Awaitable[bool]]]
 
-# Records a disconnect for the caller — invoked from the AfterToolCallEvent
-# path when a tool returns a 401 against the cached vault token. Called
-# instead of mutating per-process state so the intent is durable across
-# replicas.
-MarkDisconnected = Callable[[str], Union[None, Awaitable[None]]]
-
 
 class OAuthConsentHook(HookProvider):
     """Pause the agent if a tool needs OAuth and we don't have a token yet."""
@@ -148,7 +142,6 @@ def __init__(
         provider_type_lookup: Optional[ProviderTypeLookup] = None,
         custom_parameters_lookup: Optional[CustomParametersLookup] = None,
         disconnected_lookup: Optional[DisconnectedLookup] = None,
-        mark_disconnected: Optional[MarkDisconnected] = None,
     ):
         """Initialize.
 
@@ -169,19 +162,13 @@ def __init__(
                 effectively assumes the user has not disconnected. Wire
                 this to the durable disconnect repository in production so
                 a /disconnect on one replica is visible from any other.
-            mark_disconnected: See `MarkDisconnected`. Optional. Invoked
-                from the 401-retry path; without it, a 401 still flips
-                `event.retry = True` but leaves no durable record, so the
-                next BeforeToolCallEvent on a different replica won't know
-                to force a fresh consent.
         """
         self._user_id = user_id
         self._provider_lookup = provider_lookup
         self._scopes_lookup = scopes_lookup
         self._provider_type_lookup = provider_type_lookup
         self._custom_parameters_lookup = custom_parameters_lookup
         self._disconnected_lookup = disconnected_lookup
-        self._mark_disconnected = mark_disconnected
         # Cache scopes per provider for the lifetime of this hook (one agent
         # invocation). Avoids repeated DB hits if the same provider is used
         # across multiple tool calls in a single turn.
@@ -196,10 +183,9 @@ def __init__(
         # Providers that already burned their one 401-retry in the current
         # turn. The agent instance is cached across turns by `get_agent`, so
         # this set must be reset on `BeforeInvocationEvent`. Without the cap,
-        # a misconfigured provider (wrong scope, perma-401) would surface a
-        # consent prompt on every tool call in the turn — `_record_disconnect`
-        # forces fresh consent on the next BeforeToolCallEvent, the user
-        # consents, the tool 401s again, and the loop repeats per tool use.
+        # a misconfigured provider (wrong scope, perma-401) would loop:
+        # the cache is cleared, AgentCore returns the same expired/invalid
+        # token from the vault, the tool 401s again, and so on.
         self._reauth_attempted_providers: set[str] = set()
 
     def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
@@ -298,7 +284,11 @@ async def _fetch_token_or_url(
                 scopes=scopes,
                 user_id=self._user_id,
                 force_authentication=force_authentication,
-                custom_parameters=custom_parameters_for(provider_type, admin_extras),
+                custom_parameters=custom_parameters_for(
+                    provider_type,
+                    admin_extras,
+                    force_authentication=force_authentication,
+                ),
             )
         except WorkloadTokenUnavailableError:
             logger.error(
@@ -328,19 +318,28 @@ async def _fetch_token_or_url(
         }
 
     async def _handle_auth_failure(self, event: AfterToolCallEvent) -> None:
-        """Detect a 401 from an OAuth-gated MCP tool and retry with fresh consent.
-
-        AgentCore Identity has no revoke API, so when the user revokes our
-        app at the provider (or the refresh token expires), AgentCore's
-        vault keeps serving the now-stale token. The MCP server rejects it
-        with a 401 — and that's where the staleness first becomes visible.
+        """Detect a 401 from an OAuth-gated MCP tool and retry.
 
-        We detect the 401 in the tool result, mark the (user, provider)
-        for forced re-consent in the cache, and set `event.retry = True`.
+        We clear the local hot-path cache and set `event.retry = True`.
         Strands' tool executor then re-fires `BeforeToolCallEvent`, our
-        `_gate` callback sees the force-reauth flag, asks AgentCore for a
-        fresh consent URL with `force_authentication=True`, and raises an
-        interrupt — same path as a first-time consent.
+        `_gate` callback misses the cache, and re-asks AgentCore Identity.
+        AgentCore handles refresh transparently:
+          * If the access_token was just expired, it uses the vault's
+            refresh_token to mint a new one and returns it — the user
+            never sees a prompt.
+          * If the refresh_token itself is dead (user revoked our app at
+            the provider, or it lapsed), AgentCore returns an
+            authorization URL instead, which `_gate` surfaces as the
+            standard `oauth_required` interrupt.
+
+        We deliberately do NOT write the durable disconnect flag here. A
+        401 is most commonly an expired access token, and writing the
+        flag would force `_gate` to call AgentCore with
+        `force_authentication=True` — which bypasses the vault entirely,
+        ignores the still-valid refresh_token, and prompts the user to
+        re-consent unnecessarily. The disconnect flag is reserved for
+        explicit user intent (the "Disconnect" button in the settings
+        page).
         """
         provider_id = self._provider_lookup(event.selected_tool)
         if not provider_id:
@@ -370,11 +369,9 @@ async def _handle_auth_failure(self, event: AfterToolCallEvent) -> None:
             provider_id,
         )
         # Drop the local hot-path token so the BeforeToolCallEvent retry
-        # doesn't short-circuit to it, and record the intent durably so
-        # other replicas (and subsequent requests on this one) also force a
-        # fresh consent.
+        # doesn't short-circuit to it. The retry will re-fetch from
+        # AgentCore Identity, which handles refresh internally.
         oauth_token_cache.clear_user_provider(self._user_id, provider_id)
-        await self._record_disconnect(provider_id)
         event.retry = True
 
     async def _resolve_scopes(self, provider_id: str) -> list[str]:
@@ -434,10 +431,3 @@ async def _is_disconnected(self, provider_id: str) -> bool:
             return bool(await result)
         return bool(result)
 
-    async def _record_disconnect(self, provider_id: str) -> None:
-        """Persist a disconnect from the AfterToolCallEvent retry path."""
-        if self._mark_disconnected is None:
-            return
-        result = self._mark_disconnected(provider_id)
-        if inspect.isawaitable(result):
-            await result

backend/src/apis/app_api/connectors/routes.py

@@ -229,7 +229,9 @@ async def initiate_consent(
             user_id=current_user.user_id,
             force_authentication=force_auth,
             custom_parameters=custom_parameters_for(
-                provider.provider_type.value, provider.custom_parameters
+                provider.provider_type.value,
+                provider.custom_parameters,
+                force_authentication=force_auth,
             ),
             # No custom_state: AgentCore appears to treat its presence as a
             # signal to start a fresh flow, never short-circuiting to the

backend/src/apis/shared/oauth/agentcore_identity.py

@@ -88,7 +88,9 @@ async def poll_for_token(self) -> str:
 _CALLBACK_URL_ENV = "AGENTCORE_LOCAL_OAUTH_CALLBACK_URL"
 
 
-def _vendor_baseline_params(provider_type: Optional[str]) -> Dict[str, str]:
+def _vendor_baseline_params(
+    provider_type: Optional[str], *, force_authentication: bool = False
+) -> Dict[str, str]:
     """Hardcoded params AgentCore Identity *requires* for a given vendor.
 
     Per the AgentCore Identity authentication docs
@@ -97,30 +99,51 @@ def _vendor_baseline_params(provider_type: Optional[str]) -> Dict[str, str]:
     without it the vault entry expires after ~1 hour with no refresh
     path. This is non-negotiable: it always wins over admin-supplied
     extras to prevent an admin from accidentally turning it off.
+
+    When `force_authentication=True` we additionally send `prompt=consent`
+    for Google. Google only re-issues a refresh token on subsequent
+    authorizations if the user is shown the consent screen again — so a
+    user who clicks "Disconnect" then "Reconnect" would otherwise get a
+    vault entry with an access token but no refresh token, putting them
+    right back in the hourly-reconsent loop on the next access-token
+    expiry. We only set this on the explicit re-consent path; first-time
+    consent and silent refreshes don't trigger it.
     """
     if not provider_type:
         return {}
     if provider_type.lower() == "google":
-        return {"access_type": "offline"}
+        params = {"access_type": "offline"}
+        if force_authentication:
+            params["prompt"] = "consent"
+        return params
     return {}
 
 
 def custom_parameters_for(
     provider_type: Optional[str],
     admin_extras: Optional[Dict[str, str]] = None,
+    *,
+    force_authentication: bool = False,
 ) -> Optional[Dict[str, str]]:
     """Build the `customParameters` payload AgentCore Identity wants forwarded.
 
     Merges admin-supplied extras (e.g. Google `hd=mycorp.com` for domain
-    restriction, `prompt=consent` for stricter UX) with the hardcoded
-    vendor baseline. Baseline keys win on conflict — admins cannot turn
-    off a documented requirement.
+    restriction) with the hardcoded vendor baseline. Baseline keys win on
+    conflict — admins cannot turn off a documented requirement.
+
+    `force_authentication=True` mirrors the same flag on
+    `get_token_for_user`: when the caller is forcing AgentCore to bypass
+    the vault and walk the user through consent again, this enables any
+    vendor-specific extras needed for that path (e.g. Google's
+    `prompt=consent` so a refresh token is re-issued).
 
     Returns None when the merged result would be empty, so callers can
     pass the value through to the SDK unconditionally without sending
     an empty `customParameters` map.
     """
-    baseline = _vendor_baseline_params(provider_type)
+    baseline = _vendor_baseline_params(
+        provider_type, force_authentication=force_authentication
+    )
     merged = {**(admin_extras or {}), **baseline}
     return merged or None