mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-05-29 06:31:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 14

refactor(auth): mostly cleanups and style changes

Browse source
This commit is contained in:
Robin Fernandes 2026-05-17 21:18:53 +10:00 • committed by Teknium
parent 0bac7dd05b
commit 20bffa5b37
10 changed files with 145 additions and 125 deletions
Download patch file Download diff file Expand all files Collapse all files

17
agent/auxiliary_client.py
View file

@ -1253,18 +1253,18 @@ def _resolve_nous_runtime_api(*, force_refresh: bool = False) -> Optional[tuple[
"""
try:
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_AUTO,
NOUS_INFERENCE_AUTH_LEGACY,
NOUS_INFERENCE_AUTH_MODE_AUTO,
NOUS_INFERENCE_AUTH_MODE_LEGACY,
resolve_nous_runtime_credentials,
)
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=max(60, int(os.getenv("HERMES_NOUS_MIN_KEY_TTL_SECONDS", "1800"))),
timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
auth_mode=(
NOUS_INFERENCE_AUTH_LEGACY
inference_auth_mode=(
NOUS_INFERENCE_AUTH_MODE_LEGACY
if force_refresh
else NOUS_INFERENCE_AUTH_AUTO
else NOUS_INFERENCE_AUTH_MODE_AUTO
),
)
except Exception as exc:
@ -2509,12 +2509,15 @@ def _refresh_provider_credentials(provider: str) -> bool:
_evict_cached_clients(normalized)
return True
if normalized == "nous":
from hermes_cli.auth import NOUS_INFERENCE_AUTH_LEGACY, resolve_nous_runtime_credentials
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_MODE_LEGACY,
resolve_nous_runtime_credentials,
)
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=max(60, int(os.getenv("HERMES_NOUS_MIN_KEY_TTL_SECONDS", "1800"))),
timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
auth_mode=NOUS_INFERENCE_AUTH_LEGACY,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
if not str(creds.get("api_key", "") or "").strip():
return False

6
agent/credential_pool.py
View file

@ -831,10 +831,10 @@ class CredentialPool:
nous_state,
min_key_ttl_seconds=DEFAULT_AGENT_KEY_MIN_TTL_SECONDS,
force_refresh=force,
auth_mode=(
auth_mod.NOUS_INFERENCE_AUTH_LEGACY
inference_auth_mode=(
auth_mod.NOUS_INFERENCE_AUTH_MODE_LEGACY
if force
else auth_mod.NOUS_INFERENCE_AUTH_AUTO
else auth_mod.NOUS_INFERENCE_AUTH_MODE_AUTO
),
)
# Apply returned fields: dataclass fields via replace, extras via dict update

146
hermes_cli/auth.py
View file

@ -78,13 +78,21 @@ NOUS_INFERENCE_INVOKE_SCOPE = "inference:invoke"
DEFAULT_NOUS_SCOPE = f"{NOUS_INFERENCE_INVOKE_SCOPE} {NOUS_LEGACY_AGENT_KEY_SCOPE}"
NOUS_LEGACY_SESSION_KEYS_ENV = "HERMES_AGENT_USE_LEGACY_SESSION_KEYS"
NOUS_DEVICE_CODE_SOURCE = "device_code"
NOUS_INFERENCE_AUTH_AUTO = "auto"
NOUS_INFERENCE_AUTH_FRESH = "fresh"
NOUS_INFERENCE_AUTH_LEGACY = "legacy"
NOUS_INFERENCE_AUTH_MODE_AUTO = "auto"
NOUS_INFERENCE_AUTH_MODE_FRESH = "fresh"
NOUS_INFERENCE_AUTH_MODE_LEGACY = "legacy"
NOUS_INFERENCE_AUTH_MODES = frozenset({
NOUS_INFERENCE_AUTH_AUTO,
NOUS_INFERENCE_AUTH_FRESH,
NOUS_INFERENCE_AUTH_LEGACY,
NOUS_INFERENCE_AUTH_MODE_AUTO,
NOUS_INFERENCE_AUTH_MODE_FRESH,
NOUS_INFERENCE_AUTH_MODE_LEGACY,
})
NOUS_AUTH_PATH_INVOKE_JWT = "invoke_jwt"
NOUS_AUTH_PATH_LEGACY_SESSION_KEY_CACHE = "legacy_session_key_cache"
NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT = "legacy_session_key_mint"
NOUS_AUTH_PATHS = frozenset({
NOUS_AUTH_PATH_INVOKE_JWT,
NOUS_AUTH_PATH_LEGACY_SESSION_KEY_CACHE,
NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT,
})
DEFAULT_AGENT_KEY_MIN_TTL_SECONDS = 30 * 60 # 30 minutes
ACCESS_TOKEN_REFRESH_SKEW_SECONDS = 120 # refresh 2 min before expiry
@ -1592,12 +1600,13 @@ def _nous_scope_has_invoke(raw_scope: Any) -> bool:
return NOUS_INFERENCE_INVOKE_SCOPE in _scope_values(raw_scope)
def _normalize_nous_auth_mode(auth_mode: Optional[str]) -> str:
mode = str(auth_mode or NOUS_INFERENCE_AUTH_AUTO).strip().lower()
def _normalize_nous_inference_auth_mode(inference_auth_mode: Optional[str]) -> str:
mode = str(inference_auth_mode or NOUS_INFERENCE_AUTH_MODE_AUTO).strip().lower()
if mode not in NOUS_INFERENCE_AUTH_MODES:
allowed = ", ".join(sorted(NOUS_INFERENCE_AUTH_MODES))
raise ValueError(
f"Invalid Nous inference auth mode {auth_mode!r}; expected one of: {allowed}"
"Invalid Nous inference auth mode "
f"{inference_auth_mode!r}; expected one of: {allowed}"
)
return mode
@ -1649,89 +1658,57 @@ def _nous_invoke_jwt_is_usable(
)
def _nous_invoke_jwt_unavailable_reason(
token: Any,
*,
scope: Any = None,
expires_at: Any = None,
min_ttl_seconds: int = NOUS_INVOKE_JWT_MIN_TTL_SECONDS,
) -> str:
return (
_nous_invoke_jwt_status(
token,
scope=scope,
expires_at=expires_at,
min_ttl_seconds=min_ttl_seconds,
)
or "invoke_jwt_unavailable"
)
def _nous_can_select_invoke_jwt(auth_mode: str = NOUS_INFERENCE_AUTH_AUTO) -> bool:
return (
not _nous_legacy_session_keys_forced()
and _normalize_nous_auth_mode(auth_mode) != NOUS_INFERENCE_AUTH_LEGACY
)
def _nous_legacy_session_key_reason(
token: Any,
*,
scope: Any = None,
expires_at: Any = None,
auth_mode: str = NOUS_INFERENCE_AUTH_AUTO,
inference_auth_mode: str = NOUS_INFERENCE_AUTH_MODE_AUTO,
) -> str:
if _normalize_nous_auth_mode(auth_mode) == NOUS_INFERENCE_AUTH_LEGACY:
if inference_auth_mode == NOUS_INFERENCE_AUTH_MODE_LEGACY:
return "forced_legacy_session_key"
if _nous_legacy_session_keys_forced():
return "forced_legacy_session_keys"
return _nous_invoke_jwt_unavailable_reason(
token,
scope=scope,
expires_at=expires_at,
return (
_nous_invoke_jwt_status(token, scope=scope, expires_at=expires_at)
or "invoke_jwt_unavailable"
)
def _nous_cached_agent_key_is_usable(
state: Dict[str, Any],
min_ttl_seconds: int,
) -> bool:
return _agent_key_is_usable(state, min_ttl_seconds)
def _choose_nous_inference_auth_path(
state: Dict[str, Any],
*,
access_token: Any = None,
min_key_ttl_seconds: int = DEFAULT_AGENT_KEY_MIN_TTL_SECONDS,
auth_mode: str = NOUS_INFERENCE_AUTH_AUTO,
inference_auth_mode: str = NOUS_INFERENCE_AUTH_MODE_AUTO,
) -> Tuple[str, Optional[str]]:
auth_mode = _normalize_nous_auth_mode(auth_mode)
inference_auth_mode = _normalize_nous_inference_auth_mode(inference_auth_mode)
token = state.get("access_token") if access_token is None else access_token
if (
_nous_can_select_invoke_jwt(auth_mode)
not _nous_legacy_session_keys_forced()
and inference_auth_mode != NOUS_INFERENCE_AUTH_MODE_LEGACY
and _nous_invoke_jwt_is_usable(
token,
scope=state.get("scope"),
expires_at=state.get("expires_at"),
)
):
return "invoke_jwt", None
return NOUS_AUTH_PATH_INVOKE_JWT, None
if (
auth_mode == NOUS_INFERENCE_AUTH_AUTO
and _nous_cached_agent_key_is_usable(
inference_auth_mode == NOUS_INFERENCE_AUTH_MODE_AUTO
and _agent_key_is_usable(
state,
max(60, int(min_key_ttl_seconds)),
)
):
return "legacy_session_key_cache", None
return NOUS_AUTH_PATH_LEGACY_SESSION_KEY_CACHE, None
return (
"legacy_session_key_mint",
NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT,
_nous_legacy_session_key_reason(
token,
scope=state.get("scope"),
expires_at=state.get("expires_at"),
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
),
)
@ -3660,7 +3637,7 @@ def _is_nous_invoke_scope_refusal(exc: Exception) -> bool:
)
def _nous_device_scope(
def _nous_device_scope_with_env_override(
requested_scope: Optional[str],
*,
default_scope: str = DEFAULT_NOUS_SCOPE,
@ -4131,7 +4108,7 @@ def _try_import_shared_nous_state(
min_key_ttl_seconds=min_key_ttl_seconds,
timeout_seconds=timeout_seconds,
force_refresh=True,
auth_mode=NOUS_INFERENCE_AUTH_FRESH,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
)
_write_shared_nous_state(refreshed)
except AuthError as exc:
@ -4440,10 +4417,10 @@ def refresh_nous_oauth_pure(
insecure: Optional[bool] = None,
ca_bundle: Optional[str] = None,
force_refresh: bool = False,
auth_mode: str = NOUS_INFERENCE_AUTH_AUTO,
inference_auth_mode: str = NOUS_INFERENCE_AUTH_MODE_AUTO,
) -> Dict[str, Any]:
"""Refresh Nous OAuth state without mutating auth.json."""
auth_mode = _normalize_nous_auth_mode(auth_mode)
inference_auth_mode = _normalize_nous_inference_auth_mode(inference_auth_mode)
state: Dict[str, Any] = {
"access_token": access_token,
"refresh_token": refresh_token,
@ -4506,11 +4483,11 @@ def refresh_nous_oauth_pure(
selected_auth_path, fallback_reason = _choose_nous_inference_auth_path(
state,
min_key_ttl_seconds=min_agent_key_ttl,
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
)
if selected_auth_path == "invoke_jwt":
if selected_auth_path == NOUS_AUTH_PATH_INVOKE_JWT:
_select_nous_invoke_jwt(state)
elif selected_auth_path == "legacy_session_key_mint":
elif selected_auth_path == NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT:
_log_nous_legacy_session_key_selected(
fallback_reason or "legacy_session_key_required",
access_token=state.get("access_token"),
@ -4541,7 +4518,7 @@ def refresh_nous_oauth_from_state(
min_key_ttl_seconds: int = DEFAULT_AGENT_KEY_MIN_TTL_SECONDS,
timeout_seconds: float = 15.0,
force_refresh: bool = False,
auth_mode: str = NOUS_INFERENCE_AUTH_AUTO,
inference_auth_mode: str = NOUS_INFERENCE_AUTH_MODE_AUTO,
) -> Dict[str, Any]:
"""Refresh Nous OAuth from a state dict. Thin wrapper around refresh_nous_oauth_pure."""
tls = state.get("tls") or {}
@ -4562,7 +4539,7 @@ def refresh_nous_oauth_from_state(
insecure=tls.get("insecure"),
ca_bundle=tls.get("ca_bundle"),
force_refresh=force_refresh,
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
)
@ -4640,7 +4617,7 @@ def resolve_nous_runtime_credentials(
timeout_seconds: float = 15.0,
insecure: Optional[bool] = None,
ca_bundle: Optional[str] = None,
auth_mode: str = NOUS_INFERENCE_AUTH_AUTO,
inference_auth_mode: str = NOUS_INFERENCE_AUTH_MODE_AUTO,
) -> Dict[str, Any]:
"""
Resolve Nous inference credentials for runtime use.
@ -4652,7 +4629,7 @@ def resolve_nous_runtime_credentials(
Returns dict with: provider, base_url, api_key, key_id, expires_at,
expires_in, source ("invoke_jwt", "cache", or "portal"), and auth_path.
"""
auth_mode = _normalize_nous_auth_mode(auth_mode)
inference_auth_mode = _normalize_nous_inference_auth_mode(inference_auth_mode)
min_key_ttl_seconds = max(60, int(min_key_ttl_seconds))
sequence_id = uuid.uuid4().hex[:12]
@ -4682,6 +4659,8 @@ def resolve_nous_runtime_credentials(
def _persist_state(reason: str) -> None:
nonlocal persisted_state, state_persisted
# Skip writes where only derived TTL countdowns changed; this keeps
# the mtime-keyed Nous auth-status cache warm during read paths.
if (
_nous_effective_provider_state(state)
== _nous_effective_provider_state(persisted_state)
@ -4723,7 +4702,7 @@ def resolve_nous_runtime_credentials(
_oauth_trace(
"nous_runtime_credentials_start",
sequence_id=sequence_id,
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
min_key_ttl_seconds=min_key_ttl_seconds,
refresh_token_fp=_token_fingerprint(state.get("refresh_token")),
)
@ -4830,16 +4809,16 @@ def resolve_nous_runtime_credentials(
state,
access_token=access_token,
min_key_ttl_seconds=min_key_ttl_seconds,
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
)
if selected_auth_path == "invoke_jwt":
if selected_auth_path == NOUS_AUTH_PATH_INVOKE_JWT:
_select_nous_invoke_jwt(
state,
access_token=access_token,
sequence_id=sequence_id,
)
elif selected_auth_path == "legacy_session_key_cache":
elif selected_auth_path == NOUS_AUTH_PATH_LEGACY_SESSION_KEY_CACHE:
used_cached_key = True
logger.info("Nous inference auth: using cached agent_key")
_oauth_trace("agent_key_reuse", sequence_id=sequence_id)
@ -4929,20 +4908,20 @@ def resolve_nous_runtime_credentials(
# Persist retry refresh immediately for crash safety and cross-process visibility.
_persist_state("post_refresh_mint_retry")
retry_auth_mode = (
NOUS_INFERENCE_AUTH_LEGACY
if auth_mode == NOUS_INFERENCE_AUTH_LEGACY
else NOUS_INFERENCE_AUTH_FRESH
retry_inference_auth_mode = (
NOUS_INFERENCE_AUTH_MODE_LEGACY
if inference_auth_mode == NOUS_INFERENCE_AUTH_MODE_LEGACY
else NOUS_INFERENCE_AUTH_MODE_FRESH
)
retry_auth_path, _ = _choose_nous_inference_auth_path(
state,
access_token=access_token,
min_key_ttl_seconds=min_key_ttl_seconds,
auth_mode=retry_auth_mode,
inference_auth_mode=retry_inference_auth_mode,
)
if retry_auth_path == "invoke_jwt":
if retry_auth_path == NOUS_AUTH_PATH_INVOKE_JWT:
mint_payload = None
selected_auth_path = "invoke_jwt"
selected_auth_path = NOUS_AUTH_PATH_INVOKE_JWT
_select_nous_invoke_jwt(
state,
access_token=access_token,
@ -5008,8 +4987,8 @@ def resolve_nous_runtime_credentials(
"expires_at": expires_at,
"expires_in": expires_in,
"source": (
"invoke_jwt"
if selected_auth_path == "invoke_jwt"
NOUS_AUTH_PATH_INVOKE_JWT
if selected_auth_path == NOUS_AUTH_PATH_INVOKE_JWT
else ("cache" if used_cached_key else "portal")
),
"auth_path": selected_auth_path,
@ -6691,7 +6670,10 @@ def _nous_device_code_login(
or pconfig.inference_base_url
).rstrip("/")
client_id = client_id or pconfig.client_id
scope, explicit_scope = _nous_device_scope(scope, default_scope=pconfig.scope)
scope, explicit_scope = _nous_device_scope_with_env_override(
scope,
default_scope=pconfig.scope,
)
timeout = httpx.Timeout(timeout_seconds)
verify: bool | str = False if insecure else (ca_bundle if ca_bundle else True)
@ -6781,7 +6763,7 @@ def _nous_device_code_login(
min_key_ttl_seconds=min_key_ttl_seconds,
timeout_seconds=timeout_seconds,
force_refresh=False,
auth_mode=NOUS_INFERENCE_AUTH_FRESH,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
)
except AuthError as exc:
if exc.code == "subscription_required":

2
hermes_cli/proxy/adapters/base.py
View file

@ -93,7 +93,7 @@ class UpstreamAdapter(ABC):
fallback paths, such as switching from a preferred token type to a
legacy bearer after the upstream rejects the first request.
"""
del failed_credential, status_code
_ = failed_credential, status_code
return None
def describe(self) -> str:

21
hermes_cli/proxy/adapters/nous_portal.py
View file

@ -19,8 +19,8 @@ from typing import Any, Dict, FrozenSet, Optional
from hermes_cli.auth import (
AuthError,
DEFAULT_NOUS_INFERENCE_URL,
NOUS_INFERENCE_AUTH_AUTO,
NOUS_INFERENCE_AUTH_LEGACY,
NOUS_INFERENCE_AUTH_MODE_AUTO,
NOUS_INFERENCE_AUTH_MODE_LEGACY,
_load_auth_store,
_is_terminal_nous_refresh_error,
_quarantine_nous_oauth_state,
@ -28,7 +28,7 @@ from hermes_cli.auth import (
_save_auth_store,
_write_shared_nous_state,
refresh_nous_oauth_from_state,
)
)
from hermes_cli.proxy.adapters.base import UpstreamAdapter, UpstreamCredential
logger = logging.getLogger(__name__)
@ -79,7 +79,9 @@ class NousPortalAdapter(UpstreamAdapter):
)
def get_credential(self) -> UpstreamCredential:
return self._get_credential(auth_mode=NOUS_INFERENCE_AUTH_AUTO)
return self._get_credential(
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_AUTO,
)
def get_retry_credential(
self,
@ -87,13 +89,16 @@ class NousPortalAdapter(UpstreamAdapter):
failed_credential: UpstreamCredential,
status_code: int,
) -> Optional[UpstreamCredential]:
del failed_credential
if status_code != 401:
return None
if failed_credential.bearer.count(".") != 2:
return None
logger.info("proxy: Nous upstream rejected bearer; retrying with legacy session key")
return self._get_credential(auth_mode=NOUS_INFERENCE_AUTH_LEGACY)
return self._get_credential(
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
def _get_credential(self, *, auth_mode: str) -> UpstreamCredential:
def _get_credential(self, *, inference_auth_mode: str) -> UpstreamCredential:
with self._lock:
state = self._read_state()
if state is None:
@ -104,7 +109,7 @@ class NousPortalAdapter(UpstreamAdapter):
try:
refreshed = refresh_nous_oauth_from_state(
state,
auth_mode=auth_mode,
inference_auth_mode=inference_auth_mode,
)
except AuthError as exc:
if _is_terminal_nous_refresh_error(exc):

11
hermes_cli/web_server.py
View file

@ -1816,7 +1816,7 @@ async def _start_device_code_flow(provider_id: str) -> Dict[str, Any]:
"""
if provider_id == "nous":
from hermes_cli.auth import (
_nous_device_scope,
_nous_device_scope_with_env_override,
_request_nous_device_code_with_scope_fallback,
PROVIDER_REGISTRY,
)
@ -1828,7 +1828,10 @@ async def _start_device_code_flow(provider_id: str) -> Dict[str, Any]:
or pconfig.portal_base_url
).rstrip("/")
client_id = pconfig.client_id
scope, explicit_scope = _nous_device_scope(None, default_scope=pconfig.scope)
scope, explicit_scope = _nous_device_scope_with_env_override(
None,
default_scope=pconfig.scope,
)
def _do_nous_device_request():
with httpx.Client(
@ -1982,7 +1985,7 @@ async def _start_device_code_flow(provider_id: str) -> Dict[str, Any]:
def _nous_poller(session_id: str) -> None:
"""Background poller that drives a Nous device-code flow to completion."""
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_FRESH,
NOUS_INFERENCE_AUTH_MODE_FRESH,
_poll_for_token,
refresh_nous_oauth_from_state,
)
@ -2031,7 +2034,7 @@ def _nous_poller(session_id: str) -> None:
min_key_ttl_seconds=300,
timeout_seconds=15.0,
force_refresh=False,
auth_mode=NOUS_INFERENCE_AUTH_FRESH,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
)
from hermes_cli.auth import persist_nous_credentials
persist_nous_credentials(full_state)

10
run_agent.py
View file

@ -2629,18 +2629,18 @@ class AIAgent:
try:
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_AUTO,
NOUS_INFERENCE_AUTH_LEGACY,
NOUS_INFERENCE_AUTH_MODE_AUTO,
NOUS_INFERENCE_AUTH_MODE_LEGACY,
resolve_nous_runtime_credentials,
)
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=max(60, int(os.getenv("HERMES_NOUS_MIN_KEY_TTL_SECONDS", "1800"))),
timeout_seconds=float(os.getenv("HERMES_NOUS_TIMEOUT_SECONDS", "15")),
auth_mode=(
NOUS_INFERENCE_AUTH_LEGACY
inference_auth_mode=(
NOUS_INFERENCE_AUTH_MODE_LEGACY
if force
else NOUS_INFERENCE_AUTH_AUTO
else NOUS_INFERENCE_AUTH_MODE_AUTO
),
)
except Exception as exc:

25
tests/hermes_cli/test_auth_nous_provider.py
View file

@ -217,8 +217,8 @@ def test_resolve_nous_runtime_credentials_prefers_invoke_jwt_and_mirrors(
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == token
assert creds["source"] == "invoke_jwt"
assert creds["auth_path"] == "invoke_jwt"
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
assert creds["auth_path"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
payload = json.loads((hermes_home / "auth.json").read_text())
singleton = payload["providers"]["nous"]
@ -297,7 +297,7 @@ def test_resolve_nous_runtime_credentials_invoke_jwt_is_idempotent(
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == token
assert creds["source"] == "invoke_jwt"
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
assert auth_path.read_text() == before_content
assert auth_path.stat().st_mtime_ns == before_mtime
assert sync_calls == []
@ -339,7 +339,7 @@ def test_resolve_nous_runtime_credentials_trusts_invoke_jwt_exp_over_stale_metad
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=300)
assert creds["api_key"] == token
assert creds["source"] == "invoke_jwt"
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
payload = json.loads((hermes_home / "auth.json").read_text())
singleton = payload["providers"]["nous"]
assert singleton["agent_key"] == token
@ -372,7 +372,7 @@ def test_resolve_nous_runtime_credentials_does_not_apply_legacy_ttl_to_invoke_jw
creds = auth_mod.resolve_nous_runtime_credentials(min_key_ttl_seconds=1800)
assert creds["api_key"] == token
assert creds["source"] == "invoke_jwt"
assert creds["source"] == auth_mod.NOUS_AUTH_PATH_INVOKE_JWT
payload = json.loads((hermes_home / "auth.json").read_text())
assert payload["providers"]["nous"]["agent_key"] == token
assert payload["credential_pool"]["nous"][0]["agent_key"] == token
@ -403,12 +403,12 @@ def test_legacy_auth_mode_bypasses_usable_invoke_jwt(tmp_path, monkeypatch):
creds = auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
auth_mode=auth_mod.NOUS_INFERENCE_AUTH_LEGACY,
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_LEGACY,
)
assert mint_calls == [token]
assert creds["api_key"] == "legacy-after-jwt-401"
assert creds["auth_path"] == "legacy_session_key_mint"
assert creds["auth_path"] == auth_mod.NOUS_AUTH_PATH_LEGACY_SESSION_KEY_MINT
payload = json.loads((hermes_home / "auth.json").read_text())
assert payload["providers"]["nous"]["agent_key"] == "legacy-after-jwt-401"
@ -1199,7 +1199,7 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
providers.nous was empty.
"""
from hermes_cli.auth import (
NOUS_INFERENCE_AUTH_FRESH,
NOUS_INFERENCE_AUTH_MODE_FRESH,
persist_nous_credentials,
resolve_nous_runtime_credentials,
)
@ -1232,7 +1232,7 @@ def test_persist_nous_credentials_allows_recovery_from_401(tmp_path, monkeypatch
creds = resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
auth_mode=NOUS_INFERENCE_AUTH_FRESH,
inference_auth_mode=NOUS_INFERENCE_AUTH_MODE_FRESH,
)
assert creds["api_key"] == "new-agent-key"
@ -1698,7 +1698,10 @@ def test_try_import_shared_rehydrates_on_success(shared_store_env, monkeypatch):
def _fake_refresh(state, **kwargs):
# Simulate portal returning fresh tokens + a new agent_key
assert kwargs.get("force_refresh") is True
assert kwargs.get("auth_mode") == auth_mod.NOUS_INFERENCE_AUTH_FRESH
assert (
kwargs.get("inference_auth_mode")
== auth_mod.NOUS_INFERENCE_AUTH_MODE_FRESH
)
return {
**state,
"access_token": "fresh-access-tok",
@ -1826,7 +1829,7 @@ def test_runtime_refresh_uses_newer_shared_token_before_local_stale_token(
creds = auth_mod.resolve_nous_runtime_credentials(
min_key_ttl_seconds=300,
auth_mode=auth_mod.NOUS_INFERENCE_AUTH_FRESH,
inference_auth_mode=auth_mod.NOUS_INFERENCE_AUTH_MODE_FRESH,
)
assert creds["api_key"] == "agent-key-from-shared-token"

30
tests/hermes_cli/test_proxy.py
View file

@ -169,7 +169,7 @@ def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch)
adapter = NousPortalAdapter()
cred = adapter.get_retry_credential(
failed_credential=UpstreamCredential(
bearer="jwt-access",
bearer="header.jwt.signature",
base_url="https://inference-api.nousresearch.com/v1",
),
status_code=401,
@ -177,7 +177,31 @@ def test_nous_adapter_retry_credential_forces_legacy_mint(tmp_path, monkeypatch)
assert cred is not None
assert cred.bearer == "legacy-bearer"
assert mock_refresh.call_args.kwargs["auth_mode"] == "legacy"
assert mock_refresh.call_args.kwargs["inference_auth_mode"] == "legacy"
def test_nous_adapter_retry_credential_skips_opaque_bearer(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
_write_auth_store(tmp_path, {
"access_token": "jwt-access",
"refresh_token": "refresh-tok",
"agent_key": "opaque-bearer",
})
with patch(
"hermes_cli.proxy.adapters.nous_portal.refresh_nous_oauth_from_state",
) as mock_refresh:
adapter = NousPortalAdapter()
cred = adapter.get_retry_credential(
failed_credential=UpstreamCredential(
bearer="opaque-bearer",
base_url="https://inference-api.nousresearch.com/v1",
),
status_code=401,
)
assert cred is None
mock_refresh.assert_not_called()
def test_nous_adapter_get_credential_raises_when_not_logged_in(tmp_path, monkeypatch):
@ -364,7 +388,7 @@ class FakeAdapter(UpstreamAdapter):
)
def get_retry_credential(self, *, failed_credential, status_code):
del failed_credential
_ = failed_credential
self.retry_calls += 1
if status_code != 401 or not self._retry_bearer:
return None

2
tests/run_agent/test_run_agent.py
View file

@ -3667,7 +3667,7 @@ class TestNousCredentialRefresh:
assert ok is True
assert closed["value"] is True
assert captured["auth_mode"] == "legacy"
assert captured["inference_auth_mode"] == "legacy"
assert rebuilt["kwargs"]["api_key"] == "new-nous-key"
assert (
rebuilt["kwargs"]["base_url"] == "https://inference-api.nousresearch.com/v1"