diff --git a/hermes_cli/web_routers/oauth.py b/hermes_cli/web_routers/oauth.py index 6e33afc809..ac8905ca3c 100644 --- a/hermes_cli/web_routers/oauth.py +++ b/hermes_cli/web_routers/oauth.py @@ -240,63 +240,46 @@ def _codex_full_login_worker(session_id: str) -> None: s["error_message"] = str(e) -# Hand-written status card shapes per provider id: (hauth getter name, shaper). +_OMIT: Any = object() + + +def _status_card( + raw: dict, source, source_label, token_preview, expires_at, has_refresh_token, last_refresh=_OMIT +) -> Dict[str, Any]: + card = { + "logged_in": bool(raw.get("logged_in")), "source": source, "source_label": source_label, + "token_preview": token_preview, "expires_at": expires_at, "has_refresh_token": has_refresh_token, + } + if last_refresh is not _OMIT: + card["last_refresh"] = last_refresh + return card + + +# Hand-written status cards per provider id: (hauth getter name, raw -> card). # Providers absent here fall through to the slug-driven ``get_auth_status``. -def _nous_status(raw): - # Refresh-free snapshot so listing providers never performs an OAuth refresh. - return { - "logged_in": bool(raw.get("logged_in")), "source": "nous_portal", - "source_label": raw.get("portal_base_url") or "Nous Portal", - "token_preview": _truncate_token(raw.get("access_token")), - "expires_at": raw.get("access_expires_at"), - "has_refresh_token": bool(raw.get("has_refresh_token")), - } - - -def _codex_status(raw): - return { - "logged_in": bool(raw.get("logged_in")), "source": raw.get("source") or "openai_codex", - "source_label": raw.get("auth_mode") or "OpenAI Codex", - "token_preview": _truncate_token(raw.get("api_key")), "expires_at": None, - "has_refresh_token": False, "last_refresh": raw.get("last_refresh"), - } - - -def _qwen_status(raw): - return { - "logged_in": bool(raw.get("logged_in")), "source": "qwen_cli", - "source_label": raw.get("auth_store_path") or "Qwen CLI", - "token_preview": _truncate_token(raw.get("access_token")), - "expires_at": raw.get("expires_at"), - "has_refresh_token": bool(raw.get("has_refresh_token")), - } - - -def _minimax_status(raw): - return { - "logged_in": bool(raw.get("logged_in")), "source": "minimax_oauth", - "source_label": f"MiniMax ({raw.get('region', 'global')})", "token_preview": None, - "expires_at": raw.get("expires_at"), "has_refresh_token": True, - } - - -def _xai_status(raw): - # source_label is a human-readable origin (auth-store path / credential - # source), not the internal auth_mode string ("oauth_pkce"). - return { - "logged_in": bool(raw.get("logged_in")), "source": raw.get("source") or "xai_oauth", - "source_label": raw.get("auth_store") or raw.get("source") or "xAI Grok OAuth", - "token_preview": _truncate_token(raw.get("api_key")), "expires_at": None, - "has_refresh_token": True, "last_refresh": raw.get("last_refresh"), - } - - +# nous: refresh-free local snapshot so listing providers never performs an OAuth +# refresh. xai: source_label is a human-readable origin (auth-store path / +# credential source), not the internal auth_mode string ("oauth_pkce"). _PROVIDER_STATUS: Dict[str, tuple[str, Callable[[dict], dict]]] = { - "nous": ("get_nous_auth_status_local", _nous_status), - "openai-codex": ("get_codex_auth_status", _codex_status), - "qwen-oauth": ("get_qwen_auth_status", _qwen_status), - "minimax-oauth": ("get_minimax_oauth_auth_status", _minimax_status), - "xai-oauth": ("get_xai_oauth_auth_status", _xai_status), + "nous": ("get_nous_auth_status_local", lambda r: _status_card( + r, "nous_portal", r.get("portal_base_url") or "Nous Portal", + _truncate_token(r.get("access_token")), r.get("access_expires_at"), bool(r.get("has_refresh_token")), + )), + "openai-codex": ("get_codex_auth_status", lambda r: _status_card( + r, r.get("source") or "openai_codex", r.get("auth_mode") or "OpenAI Codex", + _truncate_token(r.get("api_key")), None, False, r.get("last_refresh"), + )), + "qwen-oauth": ("get_qwen_auth_status", lambda r: _status_card( + r, "qwen_cli", r.get("auth_store_path") or "Qwen CLI", + _truncate_token(r.get("access_token")), r.get("expires_at"), bool(r.get("has_refresh_token")), + )), + "minimax-oauth": ("get_minimax_oauth_auth_status", lambda r: _status_card( + r, "minimax_oauth", f"MiniMax ({r.get('region', 'global')})", None, r.get("expires_at"), True, + )), + "xai-oauth": ("get_xai_oauth_auth_status", lambda r: _status_card( + r, r.get("source") or "xai_oauth", r.get("auth_store") or r.get("source") or "xAI Grok OAuth", + _truncate_token(r.get("api_key")), None, True, r.get("last_refresh"), + )), } @@ -318,21 +301,15 @@ def _resolve_provider_status(provider_id: str, status_fn) -> Dict[str, Any]: # a new OAuth/account provider plugin never renders permanently logged-out. raw = hauth.get_auth_status(provider_id) if isinstance(raw, dict) and "logged_in" in raw: - return { - "logged_in": bool(raw.get("logged_in")), - "source": raw.get("source") or raw.get("provider") or provider_id, - "source_label": ( - raw.get("source_label") - or raw.get("auth_store") - or raw.get("auth_store_path") - or raw.get("base_url") - or raw.get("name") - or "" - ), - "token_preview": _truncate_token(raw.get("access_token") or raw.get("api_key")), - "expires_at": raw.get("expires_at") or raw.get("access_expires_at"), - "has_refresh_token": bool(raw.get("has_refresh_token")), - } + return _status_card( + raw, + raw.get("source") or raw.get("provider") or provider_id, + raw.get("source_label") or raw.get("auth_store") or raw.get("auth_store_path") + or raw.get("base_url") or raw.get("name") or "", + _truncate_token(raw.get("access_token") or raw.get("api_key")), + raw.get("expires_at") or raw.get("access_expires_at"), + bool(raw.get("has_refresh_token")), + ) except Exception as e: return {"logged_in": False, "error": str(e)} return {"logged_in": False}