diff --git a/hermes_cli/approval_transport.py b/hermes_cli/approval_transport.py index 64b7a75520..bf141d0b9f 100644 --- a/hermes_cli/approval_transport.py +++ b/hermes_cli/approval_transport.py @@ -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. diff --git a/hermes_cli/approvals_suggest.py b/hermes_cli/approvals_suggest.py index e0a432d9ed..231338cf14 100644 --- a/hermes_cli/approvals_suggest.py +++ b/hermes_cli/approvals_suggest.py @@ -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) diff --git a/hermes_cli/auth_minimax.py b/hermes_cli/auth_minimax.py index d7b9b349d4..3dc798ea14 100644 --- a/hermes_cli/auth_minimax.py +++ b/hermes_cli/auth_minimax.py @@ -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, diff --git a/hermes_cli/auth_model_picker.py b/hermes_cli/auth_model_picker.py index 2a9497b046..36616ffea9 100644 --- a/hermes_cli/auth_model_picker.py +++ b/hermes_cli/auth_model_picker.py @@ -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. diff --git a/hermes_cli/auth_qwen.py b/hermes_cli/auth_qwen.py index 1f82e3daab..ad15b86300 100644 --- a/hermes_cli/auth_qwen.py +++ b/hermes_cli/auth_qwen.py @@ -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 diff --git a/hermes_cli/auth_spotify.py b/hermes_cli/auth_spotify.py index f8fd068e93..bbf56e4672 100644 --- a/hermes_cli/auth_spotify.py +++ b/hermes_cli/auth_spotify.py @@ -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(): diff --git a/hermes_cli/auth_xai.py b/hermes_cli/auth_xai.py index 565aeb1e8e..92b546cea2 100644 --- a/hermes_cli/auth_xai.py +++ b/hermes_cli/auth_xai.py @@ -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, ) diff --git a/hermes_cli/blueprint_cmd.py b/hermes_cli/blueprint_cmd.py index d51cd42fb7..68a2413045 100644 --- a/hermes_cli/blueprint_cmd.py +++ b/hermes_cli/blueprint_cmd.py @@ -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.