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

refactor(hermes_cli): group C — pack short dict/kwarg spans (AST-verified)

Browse Source
This commit is contained in:
Teknium
2026-09-02 21:26:16 -07:00
parent f8e71375a7
commit 7010f2d996
8 changed files with 34 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/approvals_suggest.py
+2 -6
View File
@@ -367,12 +367,8 @@ def suggest_command(args) -> int:
"days": days,
"proposals": [
{
"n": i,
"pattern": p.pattern,
"kind": p.kind,
"count": p.count,
"classes": sorted(p.classes),
"examples": p.examples,
"n": i, "pattern": p.pattern, "kind": p.kind, "count": p.count,
"classes": sorted(p.classes), "examples": p.examples,
}
for i, p in enumerate(proposals, 1)
],
hermes_cli/approvals_test.py
+3 -9
View File
@@ -20,9 +20,7 @@ EXIT_ASK = 2
EXIT_DENY = 3
_VERDICT_EXIT = {
"allow": EXIT_ALLOW,
"ask-approval": EXIT_ASK,
"hardline-deny": EXIT_DENY,
"allow": EXIT_ALLOW, "ask-approval": EXIT_ASK, "hardline-deny": EXIT_DENY,
"user-deny": EXIT_DENY,
}
@@ -46,12 +44,8 @@ def evaluate_command(command: str, env_type: str = "local") -> dict:
def result(verdict: str, rule=None, detail: str = "") -> dict:
return {
"command": command,
"env_type": env_type,
"verdict": verdict,
"exit_code": _VERDICT_EXIT[verdict],
"rule": rule,
"detail": detail,
"command": command, "env_type": env_type, "verdict": verdict,
"exit_code": _VERDICT_EXIT[verdict], "rule": rule, "detail": detail,
"normalized_variants": variants,
}
hermes_cli/auth_minimax.py
+4 -10
View File
@@ -76,12 +76,8 @@ def _minimax_request_user_code(
client,
f"{portal_base_url}/oauth/code",
data={
"response_type": "code",
"client_id": client_id,
"scope": MINIMAX_OAUTH_SCOPE,
"code_challenge": code_challenge,
"code_challenge_method": "S256",
"state": state,
"response_type": "code", "client_id": client_id, "scope": MINIMAX_OAUTH_SCOPE,
"code_challenge": code_challenge, "code_challenge_method": "S256", "state": state,
},
headers={**_FORM_JSON_HEADERS, "x-request-id": str(uuid.uuid4())},
)
@@ -135,10 +131,8 @@ def _minimax_poll_token(
client,
f"{portal_base_url}/oauth/token",
data={
"grant_type": MINIMAX_OAUTH_GRANT_TYPE,
"client_id": client_id,
"user_code": user_code,
"code_verifier": code_verifier,
"grant_type": MINIMAX_OAUTH_GRANT_TYPE, "client_id": client_id,
"user_code": user_code, "code_verifier": code_verifier,
},
headers=_FORM_JSON_HEADERS,
)
hermes_cli/auth_qwen.py
+3 -8
View File
@@ -137,9 +137,7 @@ def resolve_qwen_runtime_credentials(
return {
"provider": "qwen-oauth",
"base_url": os.getenv("HERMES_QWEN_BASE_URL", "").strip().rstrip("/") or DEFAULT_QWEN_BASE_URL,
"api_key": access_token,
"source": "qwen-cli",
"expires_at_ms": tokens.get("expiry_date"),
"api_key": access_token, "source": "qwen-cli", "expires_at_ms": tokens.get("expiry_date"),
"auth_file": str(_qwen_cli_auth_path()),
}
@@ -153,11 +151,8 @@ def get_qwen_auth_status() -> Dict[str, Any]:
# broken Qwen setup flow.
creds = resolve_qwen_runtime_credentials(refresh_if_expiring=True)
return {
"logged_in": True,
"auth_file": str(auth_path),
"source": creds.get("source"),
"api_key": creds.get("api_key"),
"expires_at_ms": creds.get("expires_at_ms"),
"logged_in": True, "auth_file": str(auth_path), "source": creds.get("source"),
"api_key": creds.get("api_key"), "expires_at_ms": creds.get("expires_at_ms"),
}
except AuthError as exc:
return {"logged_in": False, "auth_file": str(auth_path), "error": str(exc)}
hermes_cli/auth_spotify.py
+13 -29
View File
@@ -106,12 +106,8 @@ def _spotify_build_authorize_url(
accounts_base_url: str,
) -> str:
query = urlencode({
"client_id": client_id,
"response_type": "code",
"redirect_uri": redirect_uri,
"scope": scope,
"state": state,
"code_challenge_method": "S256",
"client_id": client_id, "response_type": "code", "redirect_uri": redirect_uri,
"scope": scope, "state": state, "code_challenge_method": "S256",
"code_challenge": code_challenge,
})
return f"{accounts_base_url}/authorize?{query}"
@@ -197,19 +193,15 @@ def _spotify_token_payload_to_state(
expires_at = datetime.fromtimestamp(now.timestamp() + expires_in, tz=timezone.utc)
state = dict(previous_state or {})
state.update({
"client_id": client_id,
"redirect_uri": redirect_uri,
"accounts_base_url": accounts_base_url,
"api_base_url": api_base_url,
"client_id": client_id, "redirect_uri": redirect_uri,
"accounts_base_url": accounts_base_url, "api_base_url": api_base_url,
"scope": requested_scope,
"granted_scope": str(token_payload.get("scope") or requested_scope).strip(),
"token_type": _clean(token_payload.get("token_type", "Bearer") or "Bearer") or "Bearer",
"access_token": _clean(token_payload.get("access_token")),
"refresh_token": _clean(token_payload.get("refresh_token") or state.get("refresh_token")),
"obtained_at": now.isoformat(),
"expires_at": expires_at.isoformat(),
"expires_in": expires_in,
"auth_type": "oauth_pkce",
"obtained_at": now.isoformat(), "expires_at": expires_at.isoformat(),
"expires_in": expires_in, "auth_type": "oauth_pkce",
})
return state
@@ -308,15 +300,12 @@ def resolve_spotify_runtime_credentials(
)
return {
"provider": "spotify",
"access_token": access_token,
"api_key": access_token,
"provider": "spotify", "access_token": access_token, "api_key": access_token,
"token_type": str(state.get("token_type", "Bearer") or "Bearer"),
"base_url": _spotify_api_base_url(state),
"scope": _clean(state.get("granted_scope") or state.get("scope")),
"client_id": _spotify_client_id(state=state),
"redirect_uri": _spotify_redirect_uri(state=state),
"expires_at": state.get("expires_at"),
"redirect_uri": _spotify_redirect_uri(state=state), "expires_at": state.get("expires_at"),
"refresh_token": _clean(state.get("refresh_token")),
}
@@ -331,13 +320,10 @@ def get_spotify_auth_status() -> Dict[str, Any]:
refresh_token = _clean(state.get("refresh_token"))
return {
"logged_in": bool(refresh_token or not _is_expiring(expires_at, 0)),
"auth_type": state.get("auth_type", "oauth_pkce"),
"client_id": state.get("client_id"),
"auth_type": state.get("auth_type", "oauth_pkce"), "client_id": state.get("client_id"),
"redirect_uri": state.get("redirect_uri"),
"scope": state.get("granted_scope") or state.get("scope"),
"expires_at": expires_at,
"api_base_url": state.get("api_base_url"),
"has_refresh_token": bool(refresh_token),
"scope": state.get("granted_scope") or state.get("scope"), "expires_at": expires_at,
"api_base_url": state.get("api_base_url"), "has_refresh_token": bool(refresh_token),
}
@@ -445,10 +431,8 @@ def login_spotify_command(args) -> None:
token_payload = _spotify_token_post(
accounts_base_url,
{
"client_id": client_id,
"grant_type": "authorization_code",
"code": str(callback.get("code") or ""),
"redirect_uri": redirect_uri,
"client_id": client_id, "grant_type": "authorization_code",
"code": str(callback.get("code") or ""), "redirect_uri": redirect_uri,
"code_verifier": code_verifier,
},
timeout_seconds=float(getattr(args, "timeout", None) or 20.0),
hermes_cli/auth_xai.py
+6 -13
View File
@@ -58,8 +58,7 @@ def _xai_oauth_state_from_store(auth_store: Dict[str, Any]) -> Optional[Dict[str
continue
merged = dict(state or {})
merged["tokens"] = {
"access_token": access_token,
"refresh_token": refresh_token,
"access_token": access_token, "refresh_token": refresh_token,
"token_type": str(entry.get("token_type") or "Bearer"),
}
if entry.get("last_refresh"):
@@ -96,10 +95,8 @@ def _read_xai_oauth_tokens(*, _lock: bool = True) -> Dict[str, Any]:
f"xAI OAuth state is missing {field}. {_RELOGIN}", f"xai_auth_missing_{field}", relogin=True,
)
return {
"tokens": tokens,
"last_refresh": state.get("last_refresh"),
"discovery": state.get("discovery") or {},
"redirect_uri": state.get("redirect_uri"),
"tokens": tokens, "last_refresh": state.get("last_refresh"),
"discovery": state.get("discovery") or {}, "redirect_uri": state.get("redirect_uri"),
}
@@ -308,8 +305,7 @@ def _xai_tokens_from_payload(payload: Dict[str, Any], access_token: str, fallbac
return {
"access_token": access_token,
"refresh_token": str(payload.get("refresh_token") or fallback_refresh).strip(),
"id_token": _clean(payload.get("id_token")),
"expires_in": payload.get("expires_in"),
"id_token": _clean(payload.get("id_token")), "expires_in": payload.get("expires_in"),
"token_type": _clean(payload.get("token_type") or "Bearer") or "Bearer",
}
@@ -597,9 +593,6 @@ def _xai_oauth_device_code_login(*, timeout_seconds: float = 20.0, open_browser:
raise _xai_err("xAI device-code token response was missing required tokens.", "xai_device_token_invalid")
return {
"tokens": _xai_tokens_from_payload(payload, access_token, refresh_token),
"discovery": discovery,
"redirect_uri": "",
"base_url": _xai_oauth_inference_base_url(),
"last_refresh": _utc_now_z(),
"source": "oauth-device-code",
"discovery": discovery, "redirect_uri": "", "base_url": _xai_oauth_inference_base_url(),
"last_refresh": _utc_now_z(), "source": "oauth-device-code",
}
hermes_cli/auth_zai_kimi.py
+2 -4
View File
@@ -130,10 +130,8 @@ def _resolve_zai_base_url(api_key: str, default_url: str, env_override: str) ->
return default_url
detected_endpoint = {
"base_url": detected["base_url"],
"endpoint_id": detected.get("id", ""),
"model": detected.get("model", ""),
"label": detected.get("label", ""),
"base_url": detected["base_url"], "endpoint_id": detected.get("id", ""),
"model": detected.get("model", ""), "label": detected.get("label", ""),
"key_hash": key_hash,
}
# Persist failure (disk full, permissions, lock timeout) must not break resolution — detection
hermes_cli/blueprint_cmd.py
+1 -2
View File
@@ -33,8 +33,7 @@ def _resolve_origin(explicit: Optional[Dict[str, Any]]) -> Optional[Dict[str, An
chat_id = get_session_env("HERMES_SESSION_CHAT_ID")
if platform and chat_id:
return {
"platform": platform,
"chat_id": chat_id,
"platform": platform, "chat_id": chat_id,
"chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None,
"thread_id": get_session_env("HERMES_SESSION_THREAD_ID") or None,
}