ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(hermes_cli/web_routers): oauth status cards via one _status_card builder + table (json-parity verified)

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:21:56 -07:00
parent 7d60d28f41
commit e2ba8a4e53
1 changed files with 47 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/web_routers/oauth.py
+47 -70
View File
@@ -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}