diff --git a/plugins/memory/honcho/__init__.py b/plugins/memory/honcho/__init__.py index 7be1b0b3d8355..030f0a4cd9d57 100644 --- a/plugins/memory/honcho/__init__.py +++ b/plugins/memory/honcho/__init__.py @@ -883,13 +883,7 @@ class HonchoMemoryProvider(MemoryProvider): def _pop_auth_notice(self) -> str: """One-time model-facing notice that Honcho auth expired and memory is paused.""" - msg = None - pop = getattr(self._manager, "pop_auth_notice", None) - if callable(pop): - try: - msg = pop() - except Exception: - msg = None + msg = self._manager.pop_auth_notice() if self._manager else None if not isinstance(msg, str) or not msg: # Init failures discard the manager; the provider kept the detail. if self._init_auth_failure is None or self._init_auth_notice_emitted: diff --git a/plugins/memory/honcho/oauth.py b/plugins/memory/honcho/oauth.py index 02a1be0e55dc2..a51f03cf3e61d 100644 --- a/plugins/memory/honcho/oauth.py +++ b/plugins/memory/honcho/oauth.py @@ -45,16 +45,35 @@ _REFRESH_TIMEOUT_SECONDS = 15.0 # Retry pause, kept short: the server honors a replayed refresh token only briefly after rotating it. _REFRESH_RETRY_DELAY_SECONDS = 2.0 +# Total wall-clock budget for one exchange cycle (first attempt + pause + retry). +# The exchange runs while holding the global refresh locks on the path to a +# memory call, so a stalled token endpoint must not hold them for two full +# HTTP timeouts back to back. +_REFRESH_TOTAL_BUDGET_SECONDS = 20.0 + +# After a transient exchange failure, fail open without re-exchanging for this +# long. Prevents N waiting threads (or turns) from serializing N full exchange +# cycles against an endpoint that just failed. +_REFRESH_FAILURE_COOLDOWN_SECONDS = 30.0 + # OAuth error codes that a retry can never fix — the grant itself is dead. _PERMANENT_OAUTH_ERRORS = frozenset({"invalid_grant", "invalid_client", "unauthorized_client"}) # Token values are secret even though their prefixes are not; redact before logging. -_TOKEN_VALUE_RE = re.compile(r"hch-(at|rt)-[A-Za-z0-9._~+/=-]+") +# Derived from the canonical prefixes above so a prefix change can't silently +# break redaction. +_TOKEN_VALUE_RE = re.compile( + rf"({re.escape(ACCESS_TOKEN_PREFIX)}|{re.escape(REFRESH_TOKEN_PREFIX)})[A-Za-z0-9._~+/=-]+" +) -def _redact_tokens(text: str) -> str: +def redact_tokens(text: str) -> str: """Replace any embedded token values with their prefix plus a placeholder.""" - return _TOKEN_VALUE_RE.sub(lambda m: f"hch-{m.group(1)}-[redacted]", text) + return _TOKEN_VALUE_RE.sub(lambda m: f"{m.group(1)}[redacted]", text) + + +# Backward-compat alias for oauth-internal call sites and older importers. +_redact_tokens = redact_tokens class OAuthRefreshError(Exception): @@ -132,6 +151,26 @@ _expiry_cache: dict[tuple[str, str], tuple[float, str]] = {} # Permanently rejected grants: (config path, host) → sha256 of the dead refresh token; a re-login rotates the token, so the digest check self-clears. _dead_grants: dict[tuple[str, str], str] = {} +# Last transient exchange failure per grant: key → monotonic timestamp. While +# inside the cooldown window callers fail open to the stale token without +# re-exchanging, so waiting threads don't serialize repeated full exchange +# cycles against an endpoint that just failed. +_refresh_failure_at: dict[tuple[str, str], float] = {} + + +def _in_failure_cooldown(key: tuple[str, str]) -> bool: + failed_at = _refresh_failure_at.get(key) + return ( + failed_at is not None + and (time.monotonic() - failed_at) < _REFRESH_FAILURE_COOLDOWN_SECONDS + ) + + +# Memoized reauth_required verdict per grant: key → (config mtime_ns, result). +# The verdict only changes when the config file is rewritten (re-login), so an +# unchanged mtime short-circuits the read+parse on the dead-grant hot path. +_reauth_check_cache: dict[tuple[str, str], tuple[int, bool]] = {} + def _refresh_token_digest(cred: OAuthCredential) -> str: return hashlib.sha256(cred.refresh_token.encode("utf-8")).hexdigest() @@ -143,6 +182,8 @@ def _grant_is_dead(key: tuple[str, str], cred: OAuthCredential) -> bool: def _mark_grant_dead(key: tuple[str, str], cred: OAuthCredential) -> None: _dead_grants[key] = _refresh_token_digest(cred) + # The verdict changed without a config rewrite; drop any memoized answer. + _reauth_check_cache.pop(key, None) def reauth_required(path: Path, host: str) -> bool: @@ -150,9 +191,29 @@ def reauth_required(path: Path, host: str) -> bool: key = (str(path), host) if key not in _dead_grants: return False + # A re-login rewrites the config file, so gate the read+parse on mtime: + # while the file is unchanged the answer cannot change. + try: + mtime = path.stat().st_mtime_ns + except OSError: + mtime = -1 + cached = _reauth_check_cache.get(key) + if cached is not None and cached[0] == mtime: + return cached[1] block = (_read_config(path).get("hosts") or {}).get(host) or {} cred = OAuthCredential.from_host_block(block) - return cred is not None and _grant_is_dead(key, cred) + result = cred is not None and _grant_is_dead(key, cred) + _reauth_check_cache[key] = (mtime, result) + return result + + +def any_dead_grants() -> bool: + """Cheap predicate: has any grant in this process been marked dead? + + Lets hot-path callers skip config-path resolution entirely in the + overwhelmingly common healthy state. + """ + return bool(_dead_grants) def is_oauth_access_token(value: str | None) -> bool: @@ -260,7 +321,9 @@ def _http_get_json(url: str, timeout: float) -> dict[str, Any]: return body if isinstance(body, dict) else {} -def _exchange_refresh_token(cred: OAuthCredential, *, now: float) -> OAuthCredential: +def _exchange_refresh_token( + cred: OAuthCredential, *, now: float, timeout: float = _REFRESH_TIMEOUT_SECONDS +) -> OAuthCredential: """Run the refresh_token grant and return the rotated credential. Raises ``OAuthRefreshError`` (with the endpoint's error body) on an error @@ -273,7 +336,7 @@ def _exchange_refresh_token(cred: OAuthCredential, *, now: float) -> OAuthCreden "client_id": cred.client_id, "refresh_token": cred.refresh_token, }, - _REFRESH_TIMEOUT_SECONDS, + timeout, ) if status >= 400: error = str(body.get("error") or "") @@ -306,8 +369,13 @@ def _exchange_refresh_token(cred: OAuthCredential, *, now: float) -> OAuthCreden def _exchange_with_retry(cred: OAuthCredential, *, now: float) -> OAuthCredential: """Exchange the refresh token, retrying once on transient failure. - The server accepts a replayed token only briefly after rotating it, so the retry cannot wait. + The server accepts a replayed token only briefly after rotating it, so the + retry cannot wait — and the whole cycle is capped by + ``_REFRESH_TOTAL_BUDGET_SECONDS`` because it runs under the global refresh + locks: a fast first failure gets a full-timeout retry, a slow (timed-out) + first attempt gets only the remaining budget. """ + deadline = time.monotonic() + _REFRESH_TOTAL_BUDGET_SECONDS try: return _exchange_refresh_token(cred, now=now) except OAuthRefreshError as exc: @@ -316,12 +384,55 @@ def _exchange_with_retry(cred: OAuthCredential, *, now: float) -> OAuthCredentia first: Exception = exc except Exception as exc: first = exc + remaining = deadline - time.monotonic() - _REFRESH_RETRY_DELAY_SECONDS + if remaining <= 0: + raise first logger.warning( "Honcho OAuth token exchange failed, retrying once: %s", _redact_tokens(str(first)), ) time.sleep(_REFRESH_RETRY_DELAY_SECONDS) - return _exchange_refresh_token(cred, now=now) + return _exchange_refresh_token( + cred, now=now, timeout=min(remaining, _REFRESH_TIMEOUT_SECONDS) + ) + + +def _rotate_and_persist( + path: Path, + host: str, + key: tuple[str, str], + cred: OAuthCredential, + *, + now: float, + op_label: str = "refresh", +) -> OAuthCredential | None: + """Exchange ``cred`` and persist the rotation; ``None`` on failure (logged). + + A permanent OAuth error marks the grant dead so later calls skip the + endpoint until a new login rotates the refresh token. + """ + try: + rotated = _exchange_with_retry(cred, now=now) + except OAuthRefreshError as exc: + if exc.permanent: + _mark_grant_dead(key, cred) + logger.error( + "Honcho OAuth grant for host %s is no longer valid (%s); " + "run 'hermes honcho setup' to re-authenticate", host, exc, + ) + else: + _refresh_failure_at[key] = time.monotonic() + logger.warning("Honcho OAuth %s failed for host %s: %s", op_label, host, exc) + return None + except Exception as exc: + _refresh_failure_at[key] = time.monotonic() + logger.warning( + "Honcho OAuth %s failed for host %s: %s", + op_label, host, _redact_tokens(str(exc)), + ) + return None + _persist_credential(path, host, rotated) + return rotated def _read_config(path: Path) -> dict[str, Any]: @@ -366,6 +477,7 @@ def _persist_credential(path: Path, host: str, cred: OAuthCredential) -> None: _atomic_write_config(path, raw) _expiry_cache[(str(path), host)] = (cred.expires_at, cred.access_token) _dead_grants.pop((str(path), host), None) + _refresh_failure_at.pop((str(path), host), None) def ensure_fresh_token( @@ -404,6 +516,9 @@ def ensure_fresh_token( _expiry_cache[key] = (cred.expires_at, cred.access_token) if not cred.is_expired(now=now): return cred.access_token, False + if _in_failure_cooldown(key): + # An exchange just failed transiently; don't pile on the endpoint. + return cred.access_token, False with _refresh_lock, _config_refresh_lock(path): # Re-read under both locks: another thread or process may have just @@ -414,25 +529,12 @@ def ensure_fresh_token( return current.access_token, current.access_token != cred.access_token if _grant_is_dead(key, current): return current.access_token, False - try: - rotated = _exchange_with_retry(current, now=now) - except OAuthRefreshError as exc: - if exc.permanent: - _mark_grant_dead(key, current) - logger.error( - "Honcho OAuth grant for host %s is no longer valid (%s); " - "run 'hermes honcho setup' to re-authenticate", host, exc, - ) - else: - logger.warning("Honcho OAuth refresh failed for host %s: %s", host, exc) + if _in_failure_cooldown(key): + # The lock holder we waited on just failed; fail open too. return current.access_token, False - except Exception as exc: - logger.warning( - "Honcho OAuth refresh failed for host %s: %s", - host, _redact_tokens(str(exc)), - ) + rotated = _rotate_and_persist(path, host, key, current, now=now) + if rotated is None: return current.access_token, False - _persist_credential(path, host, rotated) logger.info("Honcho OAuth token refreshed for host %s", host) return rotated.access_token, True @@ -452,30 +554,18 @@ def force_refresh_token(path: Path, host: str) -> str | None: return None if _grant_is_dead(key, cred): return None + if _in_failure_cooldown(key): + # An exchange just failed transiently; don't force another full + # cycle — callers fail open and retry after the cooldown. + return None cached = _expiry_cache.get(key) # Another thread or process already rotated: adopt the newer on-disk token. if cached is not None and cred.access_token != cached[1] and not cred.is_expired(now=now): _expiry_cache[key] = (cred.expires_at, cred.access_token) return cred.access_token - try: - rotated = _exchange_with_retry(cred, now=now) - except OAuthRefreshError as exc: - if exc.permanent: - _mark_grant_dead(key, cred) - logger.error( - "Honcho OAuth grant for host %s is no longer valid (%s); " - "run 'hermes honcho setup' to re-authenticate", host, exc, - ) - else: - logger.warning("Honcho OAuth forced refresh failed for host %s: %s", host, exc) + rotated = _rotate_and_persist(path, host, key, cred, now=now, op_label="forced refresh") + if rotated is None: return None - except Exception as exc: - logger.warning( - "Honcho OAuth forced refresh failed for host %s: %s", - host, _redact_tokens(str(exc)), - ) - return None - _persist_credential(path, host, rotated) logger.info("Honcho OAuth token force-refreshed for host %s after an auth failure", host) return rotated.access_token @@ -526,6 +616,7 @@ def install_grant( _deep_merge(raw, granted_config) _expiry_cache[(str(path), host)] = (cred.expires_at, cred.access_token) _dead_grants.pop((str(path), host), None) + _refresh_failure_at.pop((str(path), host), None) hosts = raw.setdefault("hosts", {}) block = hosts.setdefault(host, {}) block["apiKey"] = cred.access_token diff --git a/plugins/memory/honcho/session.py b/plugins/memory/honcho/session.py index acd312915f68f..b8d2db4ef3c90 100644 --- a/plugins/memory/honcho/session.py +++ b/plugins/memory/honcho/session.py @@ -12,7 +12,7 @@ from datetime import datetime from typing import Any, Callable, TYPE_CHECKING from plugins.memory.honcho.client import get_honcho_client -from plugins.memory.honcho.oauth import _redact_tokens +from plugins.memory.honcho.oauth import redact_tokens as _redact_tokens if TYPE_CHECKING: from honcho import Honcho @@ -150,6 +150,10 @@ class HonchoSessionManager: self._cache_lock = threading.RLock() self._peers_cache: dict[str, Any] = {} self._sessions_cache: dict[str, Any] = {} + # Bumped (under _cache_lock) whenever _force_reauth rebuilds the client. + # In-flight resolvers compare it around their SDK fetch so an object + # bound to the discarded client is never stored into the fresh cache. + self._client_generation = 0 # Set when a call still fails auth after a forced token refresh; cleared on the next success. self._auth_failure: str | None = None @@ -241,6 +245,12 @@ class HonchoSessionManager: """ try: from plugins.memory.honcho import oauth + + # Fast path: no grant in this process is dead — skip config-path + # resolution entirely (this runs before every SDK call). + if not oauth.any_dead_grants(): + return False + from plugins.memory.honcho.client import resolve_config_path host = getattr(self._config, "host", "") or "" @@ -272,6 +282,7 @@ class HonchoSessionManager: # SDK shape changed: rebuild the client and drop objects holding the old transport. reset_honcho_client() with self._cache_lock: + self._client_generation += 1 self._peers_cache.clear() self._sessions_cache.clear() return True @@ -315,25 +326,32 @@ class HonchoSessionManager: def _sdk_session(self, session_id: str) -> Any: """Get or create the SDK session; a client rebuild clears the cache, so re-fetch.""" - with self._cache_lock: - cached = self._sessions_cache.get(session_id) - if cached is not None: - return cached - sdk_session = self.honcho.session(session_id) - with self._cache_lock: - self._sessions_cache[session_id] = sdk_session - return sdk_session + while True: + with self._cache_lock: + cached = self._sessions_cache.get(session_id) + generation = self._client_generation + if cached is not None: + return cached + sdk_session = self.honcho.session(session_id) + with self._cache_lock: + if self._client_generation == generation: + return self._sessions_cache.setdefault(session_id, sdk_session) + # The client was rebuilt while we resolved; this object holds the + # discarded transport. Don't cache it — resolve afresh. def _get_or_create_peer(self, peer_id: str) -> Any: """Get or create a Honcho peer (one get-or-create API call, then cached).""" - with self._cache_lock: - if peer_id in self._peers_cache: - return self._peers_cache[peer_id] + while True: + with self._cache_lock: + if peer_id in self._peers_cache: + return self._peers_cache[peer_id] + generation = self._client_generation - peer = self._authed_call("peer setup", lambda: self.honcho.peer(peer_id)) - with self._cache_lock: - self._peers_cache[peer_id] = peer - return peer + peer = self._authed_call("peer setup", lambda: self.honcho.peer(peer_id)) + with self._cache_lock: + if self._client_generation == generation: + return self._peers_cache.setdefault(peer_id, peer) + # Client rebuilt mid-resolve — drop the stale object and retry. def _get_or_create_honcho_session( self, session_id: str, user_peer: Any, assistant_peer: Any diff --git a/tests/honcho_plugin/test_auth_recovery.py b/tests/honcho_plugin/test_auth_recovery.py index 9f0903a916201..bf76a9913e1e5 100644 --- a/tests/honcho_plugin/test_auth_recovery.py +++ b/tests/honcho_plugin/test_auth_recovery.py @@ -985,3 +985,164 @@ class TestInitAuthFailureNotice: out = provider.handle_tool_call("honcho_profile", {}) assert "could not be initialized" in out + + +# --------------------------------------------------------------------------- +# hardening: exchange budget, failure cooldown, client-generation cache guard +# --------------------------------------------------------------------------- + + +class TestExchangeBudget: + def test_timed_out_first_attempt_skips_retry_when_budget_spent(self, tmp_path, monkeypatch): + """A first attempt that consumed the whole budget must not start a + second full-timeout exchange while holding the global refresh locks.""" + path = tmp_path / "honcho.json" + _write(path, {"hosts": {"hermes": _host_block()}}) + monkeypatch.setattr(oauth, "_REFRESH_RETRY_DELAY_SECONDS", 0) + + calls = [] + clock = {"now": 1000.0} + monkeypatch.setattr(oauth.time, "monotonic", lambda: clock["now"]) + + def slow_timeout(url, data, timeout): + calls.append(timeout) + clock["now"] += oauth._REFRESH_TOTAL_BUDGET_SECONDS + 1 + raise TimeoutError("token exchange timed out") + + monkeypatch.setattr(oauth, "_http_post_form_status", slow_timeout) + token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000) + + assert token == "hch-at-old" and refreshed is False + assert len(calls) == 1 # no second exchange after the budget is gone + + def test_fast_failure_retry_gets_remaining_budget(self, tmp_path, monkeypatch): + path = tmp_path / "honcho.json" + _write(path, {"hosts": {"hermes": _host_block()}}) + monkeypatch.setattr(oauth, "_REFRESH_RETRY_DELAY_SECONDS", 0) + + timeouts = [] + + def flaky(url, data, timeout): + timeouts.append(timeout) + if len(timeouts) == 1: + raise ConnectionError("reset") + return 200, _rotated_body() + + monkeypatch.setattr(oauth, "_http_post_form_status", flaky) + token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000) + + assert refreshed is True and token == "hch-at-new1" + assert len(timeouts) == 2 + # Retry timeout is bounded by both the per-attempt cap and the budget. + assert 0 < timeouts[1] <= oauth._REFRESH_TIMEOUT_SECONDS + + +class TestFailureCooldown: + def test_repeated_calls_within_cooldown_do_not_reexchange(self, tmp_path, monkeypatch): + """After a transient failure, waiting callers fail open instead of + serializing their own full exchange cycles (dogpile guard).""" + path = tmp_path / "honcho.json" + _write(path, {"hosts": {"hermes": _host_block()}}) + monkeypatch.setattr(oauth, "_REFRESH_RETRY_DELAY_SECONDS", 0) + + calls = [] + + def boom(*a, **k): + calls.append(1) + raise ConnectionError("network down") + + monkeypatch.setattr(oauth, "_http_post_form_status", boom) + oauth.ensure_fresh_token(path, "hermes", now=1000) + assert len(calls) == 2 # first attempt + its one retry + + # Subsequent callers inside the cooldown window skip the endpoint. + for _ in range(3): + token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000) + assert token == "hch-at-old" and refreshed is False + assert oauth.force_refresh_token(path, "hermes") is None + assert len(calls) == 2 + + # After the cooldown expires the exchange is attempted again. + key = (str(path), "hermes") + oauth._refresh_failure_at[key] -= oauth._REFRESH_FAILURE_COOLDOWN_SECONDS + 1 + oauth.ensure_fresh_token(path, "hermes", now=1000) + assert len(calls) == 4 + + def test_relogin_clears_the_cooldown(self, tmp_path, monkeypatch): + path = tmp_path / "honcho.json" + _write(path, {"hosts": {"hermes": _host_block()}}) + monkeypatch.setattr(oauth, "_REFRESH_RETRY_DELAY_SECONDS", 0) + + def boom(*a, **k): + raise ConnectionError("network down") + + monkeypatch.setattr(oauth, "_http_post_form_status", boom) + oauth.ensure_fresh_token(path, "hermes", now=1000) + assert oauth._in_failure_cooldown((str(path), "hermes")) is True + + _relogin(path) + assert oauth._in_failure_cooldown((str(path), "hermes")) is False + + def test_successful_rotation_clears_the_cooldown(self, tmp_path, monkeypatch): + path = tmp_path / "honcho.json" + _write(path, {"hosts": {"hermes": _host_block()}}) + monkeypatch.setattr(oauth, "_REFRESH_RETRY_DELAY_SECONDS", 0) + key = (str(path), "hermes") + oauth._refresh_failure_at[key] = ( + oauth.time.monotonic() - oauth._REFRESH_FAILURE_COOLDOWN_SECONDS - 1 + ) + monkeypatch.setattr( + oauth, "_http_post_form_status", lambda *a, **k: (200, _rotated_body()) + ) + token, refreshed = oauth.ensure_fresh_token(path, "hermes", now=1000) + assert refreshed is True + assert key not in oauth._refresh_failure_at + + +class TestClientGenerationGuard: + def test_stale_object_resolved_across_rebuild_is_not_cached(self, monkeypatch): + """A resolver that fetched from the OLD client must not store its + object into the cache after _force_reauth rebuilt the client.""" + from plugins.memory.honcho import session as session_mod + + cfg = HonchoClientConfig(host="hermes", api_key="hch-at-x", enabled=True) + mgr = HonchoSessionManager(config=cfg) + + stale_session = object() + fresh_session = object() + resolutions = [] + + class _Client: + def session(self, sid): + # First resolve returns the stale object and simulates a + # concurrent rebuild landing mid-flight; the retry gets fresh. + if not resolutions: + resolutions.append("stale") + with mgr._cache_lock: + mgr._client_generation += 1 + mgr._sessions_cache.clear() + return stale_session + resolutions.append("fresh") + return fresh_session + + client = _Client() + monkeypatch.setattr(session_mod, "get_honcho_client", lambda *a, **k: client) + got = mgr._sdk_session("s") + assert got is fresh_session + assert mgr._sessions_cache["s"] is fresh_session + assert resolutions == ["stale", "fresh"] + + def test_dead_grant_check_fast_path_skips_path_resolution(self, monkeypatch): + """With no dead grants, _reauth_required must not resolve the config + path at all (it runs before every SDK call).""" + from plugins.memory.honcho import client as client_mod + + oauth._dead_grants.clear() + + def _fail(): + raise AssertionError("resolve_config_path must not run on the fast path") + + monkeypatch.setattr(client_mod, "resolve_config_path", _fail) + cfg = HonchoClientConfig(host="hermes", api_key="hch-at-x", enabled=True) + mgr = HonchoSessionManager(config=cfg) + assert mgr._reauth_required() is False