fix(auth): memoize the valid-token fast path too, add memo tests

Follow-ups on the startup-burst memo:

- Populate the memo on the valid-token fast path as well. The startup
  burst usually finds a VALID token, and each check_fn call still paid
  two cross-process file locks + state reads to reach that return; the
  original memo only engaged after a refresh. The token has at least
  refresh_skew_seconds (>=120s) of life at that return, so a 5s memo can
  never serve an expired token.
- Clear the module-level memo in test_nous_portal_staging_allowlist's
  refresh-capture helper: with the fast-path populate, a token memoized
  by an earlier test would otherwise short-circuit the refresh these
  tests assert on (3 tests failed without this).
- Add dedicated memo behavior tests (TTL hit, TTL expiry, insecure
  bypass) — the original PR shipped none. Mutation-checked: all 3 fail
  against main's un-memoized function, pass on this branch.
This commit is contained in:
kshitijk4poor 2026-08-02 22:08:31 +05:30 committed by kshitij
parent 9929743b72
commit 06b4f64c31
3 changed files with 110 additions and 0 deletions

View File

@ -5825,6 +5825,15 @@ def resolve_nous_access_token(
if not _is_expiring(state.get("expires_at"), refresh_skew_seconds):
if merged_shared:
_save_provider_state_to_source(auth_store, "nous", state, state_source_path)
# Populate the memo on the valid-token fast path too: the
# startup burst usually finds a *valid* token, but each
# check_fn call still pays two cross-process file locks and
# state reads to reach this return. The token has at least
# refresh_skew_seconds (>= 120s) of life here, so a 5s memo
# can never serve an expired token.
if not insecure and ca_bundle is None:
with _RESOLVE_TOKEN_CACHE_LOCK:
_RESOLVE_TOKEN_CACHE = (time.monotonic(), access_token)
return access_token
if not isinstance(refresh_token, str) or not refresh_token:

View File

@ -87,6 +87,11 @@ class TestResolveAccessTokenEnvOverrideWins:
def _run_and_capture(self, monkeypatch, auth):
seen_portal_urls = []
# The resolve memo is module-level state; clear it so each test's
# resolution actually exercises the refresh path instead of serving
# a token cached by a previous test.
monkeypatch.setattr(auth, "_RESOLVE_TOKEN_CACHE", None)
def _fake_refresh(*, client, portal_base_url, client_id, refresh_token):
seen_portal_urls.append(portal_base_url)
return {

View File

@ -0,0 +1,96 @@
"""Tests for the resolve_nous_access_token startup-burst memo (PR #66016).
The memo collapses the startup burst of managed-tool check_fn calls into a
single expensive resolution: within the short TTL, repeat calls return the
cached token without re-entering _provider_state_transaction (two
cross-process file locks + state reads) or triggering a network refresh.
"""
import json
import time
import pytest
import hermes_cli.auth as auth
@pytest.fixture(autouse=True)
def _fresh_memo(monkeypatch, tmp_path):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("HERMES_PORTAL_BASE_URL", raising=False)
monkeypatch.delenv("NOUS_PORTAL_BASE_URL", raising=False)
monkeypatch.setattr(auth, "_RESOLVE_TOKEN_CACHE", None)
yield
def _write_valid_auth_file(tmp_path, token="memo-token"):
(tmp_path / "auth.json").write_text(
json.dumps(
{
"version": 1,
"active_provider": "nous",
"providers": {
"nous": {
"access_token": token,
"refresh_token": "r",
"client_id": "hermes-cli-vps",
"expires_at": time.strftime(
"%Y-%m-%dT%H:%M:%S+00:00", time.gmtime(time.time() + 3600)
),
}
},
}
)
)
def _count_transactions(monkeypatch):
calls = {"n": 0}
real = auth._provider_state_transaction
def _counting(provider):
calls["n"] += 1
return real(provider)
monkeypatch.setattr(auth, "_provider_state_transaction", _counting)
return calls
def test_repeat_calls_within_ttl_hit_memo(monkeypatch, tmp_path):
_write_valid_auth_file(tmp_path)
calls = _count_transactions(monkeypatch)
first = auth.resolve_nous_access_token()
second = auth.resolve_nous_access_token()
third = auth.resolve_nous_access_token()
assert first == second == third == "memo-token"
assert calls["n"] == 1, (
"repeat calls within the TTL must not re-enter the state transaction"
)
def test_memo_expires_after_ttl(monkeypatch, tmp_path):
_write_valid_auth_file(tmp_path)
calls = _count_transactions(monkeypatch)
auth.resolve_nous_access_token()
cached_at, tok = auth._RESOLVE_TOKEN_CACHE
monkeypatch.setattr(
auth,
"_RESOLVE_TOKEN_CACHE",
(cached_at - auth._RESOLVE_TOKEN_CACHE_TTL_S - 1.0, tok),
)
auth.resolve_nous_access_token()
assert calls["n"] == 2, "an expired memo must re-resolve"
def test_insecure_callers_bypass_memo(monkeypatch, tmp_path):
_write_valid_auth_file(tmp_path)
calls = _count_transactions(monkeypatch)
auth.resolve_nous_access_token()
auth.resolve_nous_access_token(insecure=True)
assert calls["n"] == 2, "insecure callers must bypass the memo entirely"