fix(xai): handle Grok paste flow and tiny screenshots

hermes_cli/auth.py

@@ -2595,12 +2595,23 @@ def _xai_wait_for_callback(
     result: dict[str, Any],
     *,
     timeout_seconds: float = 180.0,
+    manual_paste_redirect_uri: Optional[str] = None,
 ) -> dict[str, Any]:
     deadline = time.monotonic() + max(5.0, timeout_seconds)
+    if manual_paste_redirect_uri and sys.stdin.isatty():
+        print()
+        print("If xAI shows a Grok Build code instead of redirecting,")
+        print("paste that code here and press Enter.")
     try:
         while time.monotonic() < deadline:
             if result["code"] or result["error"]:
                 return result
+            if manual_paste_redirect_uri:
+                raw_paste = _read_ready_stdin_line()
+                if raw_paste and raw_paste.strip():
+                    pasted = _parse_pasted_callback(raw_paste)
+                    pasted["_manual_paste"] = True
+                    return pasted
             time.sleep(0.1)
     finally:
         server.shutdown()
@@ -2624,6 +2635,21 @@ def _xai_wait_for_callback(
     )
 
 
+def _read_ready_stdin_line() -> Optional[str]:
+    """Return one pending stdin line without blocking, if the terminal has one."""
+    try:
+        if not sys.stdin.isatty():
+            return None
+        import select
+
+        ready, _, _ = select.select([sys.stdin], [], [], 0)
+        if not ready:
+            return None
+        return sys.stdin.readline()
+    except Exception:
+        return None
+
+
 def _spotify_token_payload_to_state(
     token_payload: Dict[str, Any],
     *,
@@ -6555,6 +6581,7 @@ def _stdin_supports_manual_paste() -> bool:
     authorization_endpoint = discovery["authorization_endpoint"]
     token_endpoint = discovery["token_endpoint"]
 
+    allow_missing_state = False
     if manual_paste:
         # No HTTP listener — synthesize a redirect_uri matching what
         # the server would have bound to so the authorize URL the user
@@ -6581,6 +6608,7 @@ def _stdin_supports_manual_paste() -> bool:
         print("Open this URL to authorize Hermes with xAI:")
         print(authorize_url)
         callback = _prompt_manual_callback_paste(redirect_uri)
+        allow_missing_state = True
     else:
         server, thread, callback_result, redirect_uri = _xai_start_callback_server()
         try:
@@ -6620,6 +6648,7 @@ def _stdin_supports_manual_paste() -> bool:
                     thread,
                     callback_result,
                     timeout_seconds=max(30.0, timeout_seconds * 9),
+                    manual_paste_redirect_uri=redirect_uri,
                 )
             except AuthError as exc:
                 if (
@@ -6636,6 +6665,7 @@ def _stdin_supports_manual_paste() -> bool:
                 callback = _prompt_manual_callback_paste(redirect_uri)
                 if callback.get("code") is None and callback.get("error") is None:
                     raise exc
+                allow_missing_state = True
         except Exception:
             try:
                 server.shutdown()
@@ -6656,18 +6686,20 @@ def _stdin_supports_manual_paste() -> bool:
             code="xai_authorization_failed",
         )
     callback_state = callback.get("state")
-    # Manual-paste bare-code path: when a user pastes only the opaque
+    # Manual bare-code paths: when a user pastes only the opaque
     # authorization code (no ``code=``/``state=`` query parameters),
     # ``_parse_pasted_callback`` returns ``state=None``.  xAI's consent
     # page renders the code in-page rather than redirecting through the
     # 127.0.0.1 callback, so on many remote setups (Cloud Shell, headless
     # VPS, container consoles) the bare code is the only thing the user
     # can obtain.  PKCE (code_verifier) still binds the exchange to this
     # client, so the local state-equality check is redundant on the
-    # bare-code path — we substitute the locally generated state to keep
+    # bare-code paths — we substitute the locally generated state to keep
     # the rest of the validation chain (and the token exchange) unchanged.
     # See #26923 (AccursedGalaxy comment, 2026-05-20).
-    if callback_state is None and manual_paste:
+    if callback.get("_manual_paste"):
+        allow_missing_state = True
+    if callback_state is None and (manual_paste or allow_missing_state):
         callback_state = state
     if callback_state != state:
         raise AuthError(

hermes_cli/main.py

@@ -2525,7 +2525,12 @@ def _active_custom_key_from_base_url() -> str:
         member_labels = [
             provider_labels.get(m, m) for m in selected_members
         ]
-        member_idx = _prompt_provider_choice(member_labels, default=member_default)
+        group_label = ordered[provider_idx][1].split(" ▸", 1)[0]
+        member_idx = _prompt_provider_choice(
+            member_labels,
+            default=member_default,
+            title=f"Select {group_label} provider:",
+        )
         if member_idx is None:
             print("No change.")
             return
@@ -3026,7 +3031,7 @@ def _aux_flow_custom_endpoint(task: str, task_cfg: dict) -> None:
     print(f"{display_name}: custom ({short_url})" + (f" · {model}" if model else ""))
 
 
-def _prompt_provider_choice(choices, *, default=0):
+def _prompt_provider_choice(choices, *, default=0, title="Select provider:"):
     """Show provider selection menu with curses arrow-key navigation.
 
     Falls back to a numbered list when curses is unavailable (e.g. piped
@@ -3036,15 +3041,15 @@ def _prompt_provider_choice(choices, *, default=0):
     try:
         from hermes_cli.setup import _curses_prompt_choice
 
-        idx = _curses_prompt_choice("Select provider:", choices, default)
+        idx = _curses_prompt_choice(title, choices, default)
         if idx >= 0:
             print()
             return idx
     except Exception:
         pass
 
     # Fallback: numbered list
-    print("Select provider:")
+    print(title)
     for i, c in enumerate(choices, 1):
         marker = "→" if i - 1 == default else " "
         print(f"  {marker} {i}. {c}")
@@ -3065,6 +3070,40 @@ def _prompt_provider_choice(choices, *, default=0):
             return None
 
 
+def _prompt_auth_credentials_choice(title: str) -> str:
+    """Prompt for reuse / reauthenticate / cancel with the standard radio UI."""
+    choices = [
+        "Use existing credentials",
+        "Reauthenticate (new OAuth login)",
+        "Cancel",
+    ]
+    try:
+        from hermes_cli.setup import _curses_prompt_choice
+
+        idx = _curses_prompt_choice(title, choices, 0)
+        if idx >= 0:
+            print()
+            return ("use", "reauth", "cancel")[idx]
+    except Exception:
+        pass
+
+    print(title)
+    for i, label in enumerate(choices, 1):
+        marker = "→" if i == 1 else " "
+        print(f"  {marker} {i}. {label}")
+    print()
+    try:
+        choice = input("  Choice [1/2/3]: ").strip()
+    except (KeyboardInterrupt, EOFError):
+        choice = "1"
+
+    if choice == "2":
+        return "reauth"
+    if choice == "3":
+        return "cancel"
+    return "use"
+
+
 def _model_flow_openrouter(config, current_model=""):
     """OpenRouter provider: ensure API key, then pick model."""
     from hermes_constants import OPENROUTER_BASE_URL
@@ -3453,16 +3492,11 @@ def _model_flow_xai_oauth(_config, current_model="", *, args=None):
     if status.get("logged_in"):
         print("  xAI Grok OAuth (SuperGrok / Premium+) credentials: ✓")
         print()
-        print("    1. Use existing credentials")
-        print("    2. Reauthenticate (new OAuth login)")
-        print("    3. Cancel")
-        print()
-        try:
-            choice = input("  Choice [1/2/3]: ").strip()
-        except (KeyboardInterrupt, EOFError):
-            choice = "1"
+        choice = _prompt_auth_credentials_choice(
+            "xAI Grok OAuth (SuperGrok / Premium+) credentials:"
+        )
 
-        if choice == "2":
+        if choice == "reauth":
             print("Starting a fresh xAI OAuth login...")
             print()
             try:
@@ -3486,7 +3520,7 @@ def _model_flow_xai_oauth(_config, current_model="", *, args=None):
             except Exception as exc:
                 print(f"Login failed: {exc}")
                 return
-        elif choice == "3":
+        elif choice == "cancel":
             return
     else:
         print("Not logged into xAI Grok OAuth (SuperGrok / Premium+). Starting login...")

tests/hermes_cli/test_auth_manual_paste.py

@@ -465,7 +465,7 @@ def _capture(**kw):
 
 
 def test_xai_loopback_login_timeout_falls_back_to_manual_paste(monkeypatch):
-    """Loopback timeout should offer the existing manual-paste path."""
+    """Loopback timeout should accept a bare Grok Build code paste."""
     monkeypatch.setattr(
         auth_mod, "_xai_oauth_discovery",
         lambda *_a, **_k: {
@@ -523,7 +523,7 @@ def _fake_prompt(_redirect_uri):
         captured["prompt_calls"] += 1
         return {
             "code": "manual-auth-code",
-            "state": captured["state"],
+            "state": None,
             "error": None,
             "error_description": None,
         }
@@ -558,6 +558,48 @@ def _fake_prompt(_redirect_uri):
     assert creds["tokens"]["refresh_token"] == "rt-timeout"
 
 
+def test_xai_wait_for_callback_accepts_ready_stdin_code(monkeypatch):
+    """Users can paste the Grok Build code while Hermes is still waiting."""
+    class _StubServer:
+        shutdown_called = False
+        close_called = False
+
+        def shutdown(self):
+            self.shutdown_called = True
+
+        def server_close(self):
+            self.close_called = True
+
+    class _StubThread:
+        joined = False
+
+        def join(self, timeout=None):
+            self.joined = True
+
+    server = _StubServer()
+    thread = _StubThread()
+    monkeypatch.setattr(
+        auth_mod,
+        "_read_ready_stdin_line",
+        lambda: "ready-grok-build-code\n",
+    )
+
+    out = auth_mod._xai_wait_for_callback(
+        server,
+        thread,
+        {"code": None, "error": None},
+        timeout_seconds=5,
+        manual_paste_redirect_uri="http://127.0.0.1:56121/callback",
+    )
+
+    assert out["code"] == "ready-grok-build-code"
+    assert out["state"] is None
+    assert out["_manual_paste"] is True
+    assert server.shutdown_called is True
+    assert server.close_called is True
+    assert thread.joined is True
+
+
 def test_xai_loopback_login_timeout_noninteractive_reraises(monkeypatch):
     """Non-interactive stdin must keep the original timeout error."""
     monkeypatch.setattr(

tests/hermes_cli/test_xai_model_flow.py

@@ -0,0 +1,78 @@
+import argparse
+
+
+def test_xai_model_flow_reauth_uses_standard_radio_prompt(monkeypatch):
+    from hermes_cli import main as main_mod
+
+    captured = {"login_calls": 0}
+
+    monkeypatch.setattr(
+        "hermes_cli.auth.get_xai_oauth_auth_status",
+        lambda: {"logged_in": True},
+    )
+    monkeypatch.setattr(
+        "hermes_cli.setup._curses_prompt_choice",
+        lambda title, choices, default, description=None: 1,
+    )
+
+    def _fake_login(args, provider, force_new_login=False):
+        captured["login_calls"] += 1
+        captured["force_new_login"] = force_new_login
+        captured["args"] = args
+
+    monkeypatch.setattr("hermes_cli.auth._login_xai_oauth", _fake_login)
+    monkeypatch.setattr(
+        "hermes_cli.auth.resolve_xai_oauth_runtime_credentials",
+        lambda *args, **kwargs: {"base_url": "https://api.x.ai/v1"},
+    )
+    monkeypatch.setattr(
+        "hermes_cli.auth._prompt_model_selection",
+        lambda model_ids, current_model="": None,
+    )
+
+    main_mod._model_flow_xai_oauth(
+        {},
+        current_model="grok-build-0.1",
+        args=argparse.Namespace(manual_paste=True, no_browser=True, timeout=3),
+    )
+
+    assert captured["login_calls"] == 1
+    assert captured["force_new_login"] is True
+    assert captured["args"].manual_paste is True
+    assert captured["args"].no_browser is True
+    assert captured["args"].timeout == 3
+
+
+def test_xai_model_flow_cancel_skips_reauth(monkeypatch):
+    from hermes_cli import main as main_mod
+
+    monkeypatch.setattr(
+        "hermes_cli.auth.get_xai_oauth_auth_status",
+        lambda: {"logged_in": True},
+    )
+    monkeypatch.setattr(
+        "hermes_cli.setup._curses_prompt_choice",
+        lambda title, choices, default, description=None: 2,
+    )
+    monkeypatch.setattr(
+        "hermes_cli.auth._login_xai_oauth",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not reauthenticate")),
+    )
+    monkeypatch.setattr(
+        "hermes_cli.auth._prompt_model_selection",
+        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("should not pick a model")),
+    )
+
+    main_mod._model_flow_xai_oauth({}, current_model="grok-build-0.1")
+
+
+def test_auth_credentials_choice_falls_back_to_numbered_prompt(monkeypatch):
+    from hermes_cli import main as main_mod
+
+    monkeypatch.setattr(
+        "hermes_cli.setup._curses_prompt_choice",
+        lambda title, choices, default, description=None: -1,
+    )
+    monkeypatch.setattr("builtins.input", lambda prompt="": "2")
+
+    assert main_mod._prompt_auth_credentials_choice("Credentials:") == "reauth"