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

refactor(hermes_cli): group C — AST-neutral layout compaction (hug/pack/join, verified ast.dump-equal)

Browse Source
This commit is contained in:
Teknium
2026-09-02 20:56:23 -07:00
parent fe6cefd2d9
commit 0e3203618b
8 changed files with 52 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/approval_transport.py
+4 -16
View File
@@ -54,16 +54,8 @@ class ApprovalRequest:
@classmethod
def create(
cls,
*,
command: str,
description: str,
pattern_key: str,
pattern_keys: tuple[str, ...],
session_key: str,
surface: str,
allow_session: bool,
allow_permanent: bool,
cls, *, command: str, description: str, pattern_key: str, pattern_keys: tuple[str, ...],
session_key: str, surface: str, allow_session: bool, allow_permanent: bool,
timeout_seconds: float = 300,
) -> "ApprovalRequest":
choices: list[ApprovalChoice] = ["once"]
@@ -109,12 +101,8 @@ def _deny(failure: str) -> ApprovalTransportResult:
def invoke_approval_transport(
present: ApprovalPresentFn,
request: ApprovalRequest,
*,
timeout_seconds: float,
poll_interval: float = 1.0,
on_poll: Callable[[], None] | None = None,
present: ApprovalPresentFn, request: ApprovalRequest, *, timeout_seconds: float,
poll_interval: float = 1.0, on_poll: Callable[[], None] | None = None,
is_interrupted: Callable[[], bool] | None = None,
) -> ApprovalTransportResult:
"""Run a sync or async transport on a bounded daemon worker.
hermes_cli/approvals_suggest.py
+3 -7
View File
@@ -248,9 +248,7 @@ def derive_glob(normalized: str) -> Optional[str]:
def build_proposals(
records: Iterable[tuple[str, str]],
existing: Optional[set] = None,
min_count: int = 2,
records: Iterable[tuple[str, str]], existing: Optional[set] = None, min_count: int = 2,
limit: int = 20,
) -> list[Proposal]:
"""Aggregate scan records into a ranked, safety-filtered proposal list.
@@ -352,10 +350,8 @@ def suggest_command(args) -> int:
existing = set(approval_module.load_permanent_allowlist())
proposals = build_proposals(
scan_approval_history(db_path, days=days),
existing=existing,
min_count=getattr(args, "min_count", 2),
limit=getattr(args, "limit", 20),
scan_approval_history(db_path, days=days), existing=existing,
min_count=getattr(args, "min_count", 2), limit=getattr(args, "limit", 20),
)
as_json = getattr(args, "json", False)
hermes_cli/auth_minimax.py
+1 -2
View File
@@ -83,8 +83,7 @@ def _minimax_pkce_pair() -> tuple:
def _minimax_request_user_code(
client: httpx.Client, *, portal_base_url: str, client_id: str,
code_challenge: str, state: str,
client: httpx.Client, *, portal_base_url: str, client_id: str, code_challenge: str, state: str
) -> Dict[str, Any]:
response = _minimax_post_form(
client,
hermes_cli/auth_model_picker.py
+6 -18
View File
@@ -22,11 +22,7 @@ _CURRENT_SUFFIX = " ← currently in use"
def _confirm_selection_guards(
model_id: str,
*,
provider: str = "",
base_url: str = "",
api_key: str = "",
model_id: str, *, provider: str = "", base_url: str = "", api_key: str = "",
include_kinds: Optional[List[str]] = None,
) -> bool:
"""Prompt before saving a model that trips any selection guard (cost, data-policy, ...).
@@ -64,12 +60,8 @@ class _ModelPickerRows:
"""
def __init__(
self,
all_models: List[str],
pricing: Optional[Dict[str, Dict[str, str]]],
*,
current_model: str,
sale_chrome: bool,
self, all_models: List[str], pricing: Optional[Dict[str, Dict[str, str]]], *,
current_model: str, sale_chrome: bool,
) -> None:
from hermes_cli.models import _format_price_per_mtok, compute_sale_discount
@@ -161,14 +153,10 @@ class _ModelPickerRows:
def _prompt_model_selection(
model_ids: List[str],
current_model: str = "",
model_ids: List[str], current_model: str = "",
pricing: Optional[Dict[str, Dict[str, str]]] = None,
unavailable_models: Optional[List[str]] = None,
portal_url: str = "",
unavailable_message: str = "",
confirm_provider: str = "",
confirm_base_url: str = "",
unavailable_models: Optional[List[str]] = None, portal_url: str = "",
unavailable_message: str = "", confirm_provider: str = "", confirm_base_url: str = "",
confirm_api_key: str = "",
) -> Optional[str]:
"""Interactive model picker; current_model listed first. Returns the chosen model ID or None.
hermes_cli/auth_qwen.py
+2 -5
View File
@@ -67,8 +67,7 @@ def _refresh_qwen_cli_tokens(tokens: Dict[str, Any], timeout_seconds: float = 20
try:
response = httpx.post(
QWEN_OAUTH_TOKEN_URL,
headers=_FORM_JSON_HEADERS,
QWEN_OAUTH_TOKEN_URL, headers=_FORM_JSON_HEADERS,
data={"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": QWEN_OAUTH_CLIENT_ID},
timeout=timeout_seconds,
)
@@ -121,9 +120,7 @@ def _mark_qwen_oauth_active(creds: Dict[str, Any]) -> None:
def resolve_qwen_runtime_credentials(
*,
force_refresh: bool = False,
refresh_if_expiring: bool = True,
*, force_refresh: bool = False, refresh_if_expiring: bool = True,
refresh_skew_seconds: int = QWEN_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
) -> Dict[str, Any]:
from hermes_cli.auth import _qwen_cli_auth_path, _refresh_qwen_cli_tokens
hermes_cli/auth_spotify.py
+21 -70
View File
@@ -45,22 +45,15 @@ def _spotify_scope_string(raw_scope: Optional[str] = None) -> str:
def _spotify_setting(
state: Optional[Dict[str, Any]],
state_key: str,
env_vars: Tuple[str, ...],
default: str,
*,
explicit: Optional[str] = None,
strip_slash: bool = False,
state: Optional[Dict[str, Any]], state_key: str, env_vars: Tuple[str, ...], default: str, *,
explicit: Optional[str] = None, strip_slash: bool = False,
) -> str:
"""First non-empty of explicit arg, env vars (``.env`` aware), stored state, then *default*."""
from hermes_cli.config import get_env_value
candidates = (
explicit,
*(get_env_value(var) for var in env_vars),
state.get(state_key) if isinstance(state, dict) else None,
default,
explicit, *(get_env_value(var) for var in env_vars),
state.get(state_key) if isinstance(state, dict) else None, default,
)
for candidate in candidates:
cleaned = _clean(candidate)
@@ -113,12 +106,7 @@ def _spotify_code_challenge(code_verifier: str) -> str:
def _spotify_build_authorize_url(
*,
client_id: str,
redirect_uri: str,
scope: str,
state: str,
code_challenge: str,
*, client_id: str, redirect_uri: str, scope: str, state: str, code_challenge: str,
accounts_base_url: str,
) -> str:
query = urlencode({
@@ -204,14 +192,8 @@ def _spotify_wait_for_callback(redirect_uri: str, *, timeout_seconds: float = 18
def _spotify_token_payload_to_state(
token_payload: Dict[str, Any],
*,
client_id: str,
redirect_uri: str,
requested_scope: str,
accounts_base_url: str,
api_base_url: str,
previous_state: Optional[Dict[str, Any]] = None,
token_payload: Dict[str, Any], *, client_id: str, redirect_uri: str, requested_scope: str,
accounts_base_url: str, api_base_url: str, previous_state: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
from hermes_cli.auth import _coerce_ttl_seconds
now = datetime.now(timezone.utc)
@@ -237,15 +219,8 @@ def _spotify_token_payload_to_state(
def _spotify_token_post(
accounts_base_url: str,
data: Dict[str, str],
*,
timeout_seconds: float,
what: str,
failed_code: str,
invalid_code: str,
invalid_message: str,
failed_suffix: str = "",
accounts_base_url: str, data: Dict[str, str], *, timeout_seconds: float, what: str,
failed_code: str, invalid_code: str, invalid_message: str, failed_suffix: str = "",
relogin_required: bool = False,
) -> Dict[str, Any]:
"""POST to Spotify's ``/api/token`` and return the JSON payload, or raise a shaped AuthError."""
@@ -272,12 +247,7 @@ def _spotify_token_post(
def _spotify_exchange_code_for_tokens(
*,
client_id: str,
code: str,
redirect_uri: str,
code_verifier: str,
accounts_base_url: str,
*, client_id: str, code: str, redirect_uri: str, code_verifier: str, accounts_base_url: str,
timeout_seconds: float = 20.0,
) -> Dict[str, Any]:
return _spotify_token_post(
@@ -310,30 +280,22 @@ def _refresh_spotify_oauth_state(state: Dict[str, Any], *, timeout_seconds: floa
payload = _spotify_token_post(
accounts_base_url,
{"grant_type": "refresh_token", "refresh_token": refresh_token, "client_id": client_id},
timeout_seconds=timeout_seconds,
what="token refresh",
failed_code="spotify_refresh_failed",
timeout_seconds=timeout_seconds, what="token refresh", failed_code="spotify_refresh_failed",
invalid_code="spotify_refresh_invalid",
invalid_message="Spotify refresh response did not include an access_token.",
failed_suffix=" Run `hermes auth spotify` again.",
relogin_required=True,
failed_suffix=" Run `hermes auth spotify` again.", relogin_required=True,
)
return _spotify_token_payload_to_state(
payload,
client_id=client_id,
redirect_uri=_spotify_redirect_uri(state=state),
payload, client_id=client_id, redirect_uri=_spotify_redirect_uri(state=state),
requested_scope=str(state.get("scope") or DEFAULT_SPOTIFY_SCOPE),
accounts_base_url=accounts_base_url,
api_base_url=_spotify_api_base_url(state),
accounts_base_url=accounts_base_url, api_base_url=_spotify_api_base_url(state),
previous_state=state,
)
def resolve_spotify_runtime_credentials(
*,
force_refresh: bool = False,
refresh_if_expiring: bool = True,
*, force_refresh: bool = False, refresh_if_expiring: bool = True,
refresh_skew_seconds: int = SPOTIFY_ACCESS_TOKEN_REFRESH_SKEW_SECONDS,
) -> Dict[str, Any]:
from hermes_cli.auth import _auth_store_lock, _is_expiring, _load_auth_store, _load_provider_state, _quarantine_flat_oauth_state, _refresh_spotify_oauth_state, _save_auth_store, _store_provider_state
@@ -479,12 +441,8 @@ def login_spotify_command(args) -> None:
code_verifier = _spotify_code_verifier()
state_nonce = uuid.uuid4().hex
authorize_url = _spotify_build_authorize_url(
client_id=client_id,
redirect_uri=redirect_uri,
scope=scope,
state=state_nonce,
code_challenge=_spotify_code_challenge(code_verifier),
accounts_base_url=accounts_base_url,
client_id=client_id, redirect_uri=redirect_uri, scope=scope, state=state_nonce,
code_challenge=_spotify_code_challenge(code_verifier), accounts_base_url=accounts_base_url,
)
print(
@@ -512,20 +470,13 @@ def login_spotify_command(args) -> None:
raise SystemExit("Spotify authorization failed: state mismatch.")
token_payload = _spotify_exchange_code_for_tokens(
client_id=client_id,
code=str(callback.get("code") or ""),
redirect_uri=redirect_uri,
code_verifier=code_verifier,
accounts_base_url=accounts_base_url,
client_id=client_id, code=str(callback.get("code") or ""), redirect_uri=redirect_uri,
code_verifier=code_verifier, accounts_base_url=accounts_base_url,
timeout_seconds=float(getattr(args, "timeout", None) or 20.0),
)
spotify_state = _spotify_token_payload_to_state(
token_payload,
client_id=client_id,
redirect_uri=redirect_uri,
requested_scope=scope,
accounts_base_url=accounts_base_url,
api_base_url=api_base_url,
token_payload, client_id=client_id, redirect_uri=redirect_uri, requested_scope=scope,
accounts_base_url=accounts_base_url, api_base_url=api_base_url,
)
with _auth_store_lock():
hermes_cli/auth_xai.py
+14 -39
View File
@@ -134,12 +134,8 @@ def _write_through_xai_oauth_to_global_root(state: Dict[str, Any]) -> None:
def _save_xai_oauth_tokens(
tokens: Dict[str, Any],
*,
discovery: Optional[Dict[str, Any]] = None,
redirect_uri: str = "",
last_refresh: Optional[str] = None,
auth_mode: str = "oauth_device_code",
tokens: Dict[str, Any], *, discovery: Optional[Dict[str, Any]] = None, redirect_uri: str = "",
last_refresh: Optional[str] = None, auth_mode: str = "oauth_device_code",
set_active: bool = True,
) -> None:
"""Persist xAI OAuth tokens into the auth store.
@@ -310,10 +306,7 @@ def _xai_tokens_from_payload(payload: Dict[str, Any], access_token: str, fallbac
def refresh_xai_oauth_pure(
access_token: str,
refresh_token: str,
*,
token_endpoint: str = "",
access_token: str, refresh_token: str, *, token_endpoint: str = "",
timeout_seconds: float = 20.0,
) -> Dict[str, Any]:
from hermes_cli.auth import _nonempty_str, _utc_now_z, _xai_oauth_discovery
@@ -330,8 +323,7 @@ def refresh_xai_oauth_pure(
timeout = httpx.Timeout(max(5.0, float(timeout_seconds)))
with httpx.Client(timeout=timeout, headers={"Accept": "application/json"}) as client:
response = client.post(
endpoint,
headers={"Content-Type": "application/x-www-form-urlencoded"},
endpoint, headers={"Content-Type": "application/x-www-form-urlencoded"},
data={"grant_type": "refresh_token", "client_id": XAI_OAUTH_CLIENT_ID, "refresh_token": refresh_token},
)
if response.status_code != 200:
@@ -356,11 +348,9 @@ def refresh_xai_oauth_pure(
"xAI token refresh failed." + suffix, "xai_refresh_failed", relogin=response.status_code in {400, 401},
)
payload, refreshed_access = _refresh_payload_access_token(
response,
provider="xai-oauth",
response, provider="xai-oauth",
invalid_json=("xAI token refresh returned invalid JSON: {exc}", "xai_refresh_invalid_json"),
invalid_json_relogin=False,
strict_str=False,
invalid_json_relogin=False, strict_str=False,
invalid_response=("xAI token refresh response was not a JSON object.", "xai_refresh_invalid_response"),
missing_access=("xAI token refresh response was missing access_token.", "xai_refresh_missing_access_token"),
)
@@ -368,11 +358,7 @@ def refresh_xai_oauth_pure(
def _refresh_xai_oauth_tokens(
tokens: Dict[str, Any],
*,
token_endpoint: str,
redirect_uri: str = "",
timeout_seconds: float,
tokens: Dict[str, Any], *, token_endpoint: str, redirect_uri: str = "", timeout_seconds: float
) -> Dict[str, Any]:
# Re-persist whatever auth_mode is already stored (legacy pre-device-code logins may still
# carry ``oauth_pkce``): the refresh hot path must not relabel how the grant was obtained.
@@ -436,9 +422,7 @@ def _xai_oauth_inference_base_url() -> str:
def resolve_xai_oauth_runtime_credentials(
*,
force_refresh: bool = False,
refresh_if_expiring: bool = True,
*, force_refresh: bool = False, refresh_if_expiring: bool = True,
refresh_skew_seconds: Optional[int] = None,
) -> Dict[str, Any]:
from hermes_cli.auth import _auth_store_lock, _is_terminal_xai_oauth_refresh_error, _refresh_xai_oauth_tokens, _xai_oauth_discovery
@@ -515,10 +499,8 @@ def _login_xai_oauth(args, pconfig: ProviderConfig, *, force_new_login: bool = F
creds = _xai_oauth_device_code_login(timeout_seconds=timeout_seconds, open_browser=open_browser)
_save_xai_oauth_tokens(
creds["tokens"],
discovery=creds.get("discovery"),
redirect_uri=creds.get("redirect_uri", ""),
last_refresh=creds.get("last_refresh"),
creds["tokens"], discovery=creds.get("discovery"),
redirect_uri=creds.get("redirect_uri", ""), last_refresh=creds.get("last_refresh"),
auth_mode="oauth_device_code",
)
# An explicit interactive re-login means the user wants the xAI credential re-enabled.
@@ -550,11 +532,7 @@ def _xai_oauth_request_device_code(client: httpx.Client, *, scope: str = XAI_OAU
def _xai_oauth_poll_device_token(
client: httpx.Client,
*,
token_endpoint: str,
device_code: str,
expires_in: int,
client: httpx.Client, *, token_endpoint: str, device_code: str, expires_in: int,
poll_interval: int,
) -> Dict[str, Any]:
from hermes_cli.auth import _poll_device_token_generic
@@ -573,8 +551,7 @@ def _xai_oauth_poll_device_token(
return _poll_device_token_generic(
lambda: client.post(
token_endpoint,
headers=_FORM_JSON_HEADERS,
token_endpoint, headers=_FORM_JSON_HEADERS,
data={"grant_type": DEVICE_CODE_GRANT_TYPE, "client_id": XAI_OAUTH_CLIENT_ID, "device_code": device_code},
),
expires_in=int(expires_in),
@@ -603,10 +580,8 @@ def _xai_oauth_device_code_login(*, timeout_seconds: float = 20.0, open_browser:
)
print(f"Waiting for approval (polling every {max(1, interval)}s)...")
payload = _xai_oauth_poll_device_token(
client,
token_endpoint=discovery["token_endpoint"],
device_code=str(device_data["device_code"]),
expires_in=int(device_data["expires_in"]),
client, token_endpoint=discovery["token_endpoint"],
device_code=str(device_data["device_code"]), expires_in=int(device_data["expires_in"]),
poll_interval=interval,
)
hermes_cli/blueprint_cmd.py
+1 -4
View File
@@ -183,10 +183,7 @@ def _manage_hint(surface: str) -> str:
def handle_blueprint_command(
args: str,
*,
origin: Optional[Dict[str, Any]] = None,
surface: str = "cli",
args: str, *, origin: Optional[Dict[str, Any]] = None, surface: str = "cli"
) -> BlueprintCommandResult:
"""Dispatch a ``/blueprint`` invocation.