fix(cli): honor pooled credentials in model wizard

Signed-off-by: Bao <nnqbao@gmail.com>

hermes_cli/main.py

@@ -4269,7 +4269,12 @@ def _prompt_reasoning_effort_selection(efforts, current_effort=""):
 
 
 
-def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple:
+def _prompt_api_key(
+    pconfig,
+    existing_key: str,
+    provider_id: str = "",
+    existing_source: str = "",
+) -> tuple:
     """Shared API-key entry point for ``hermes setup`` / ``hermes model``.
 
     Handles both first-time entry and the already-configured case.  When a key
@@ -4323,8 +4328,14 @@ def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple:
         # Nothing we can rewrite; just acknowledge and move on.
         print()
         return existing_key, False
+    pool_backed = existing_source.startswith("credential_pool:")
+    menu = (
+        "  [K]eep / [R]eplace (default K): "
+        if pool_backed
+        else "  [K]eep / [R]eplace / [C]lear (default K): "
+    )
     try:
-        choice = input("  [K]eep / [R]eplace / [C]lear (default K): ").strip().lower()
+        choice = input(menu).strip().lower()
     except (KeyboardInterrupt, EOFError):
         print()
         choice = "k"
@@ -4340,7 +4351,7 @@ def _prompt_api_key(pconfig, existing_key: str, provider_id: str = "") -> tuple:
         print()
         return new_key, False
 
-    if choice.startswith("c"):
+    if choice.startswith("c") and not pool_backed:
         save_env_value(key_env, "")
         print(
             f"  API key cleared.  Re-run `hermes setup` to configure {pconfig.name} again."

hermes_cli/model_setup_flows.py

@@ -76,6 +76,13 @@ def bedrock_model_routable_from_region(model_id: str, region_name: str) -> bool:
     return matched_geo == geo
 
 
+def _existing_api_key_for_model_flow(provider_id: str, pconfig) -> tuple[str, str]:
+    """Resolve an existing wizard credential without changing its storage."""
+    from hermes_cli.auth import _resolve_api_key_provider_secret
+
+    return _resolve_api_key_provider_secret(provider_id, pconfig)
+
+
 def _prune_replaced_custom_model_config_credentials(
     base_url: str,
     *,
@@ -177,8 +184,6 @@ def _model_flow_openrouter(config, current_model=""):
         _save_model_choice,
         deactivate_provider,
     )
-    from hermes_cli.config import get_env_value
-
     # Route through _prompt_api_key so users can replace a stale/broken key
     # in-flow (K/R/C) instead of having to edit ~/.hermes/.env by hand. The
     # previous bypass-when-key-exists branch left no way to recover from a
@@ -190,11 +195,16 @@ def _model_flow_openrouter(config, current_model=""):
         auth_type="api_key",
         api_key_env_vars=("OPENROUTER_API_KEY",),
     )
-    existing_key = get_env_value("OPENROUTER_API_KEY") or ""
+    existing_key, existing_source = _existing_api_key_for_model_flow("openrouter", pconfig)
     if not existing_key:
         print("Get one at: https://openrouter.ai/keys")
         print()
-    _resolved, abort = _prompt_api_key(pconfig, existing_key, provider_id="openrouter")
+    _resolved, abort = _prompt_api_key(
+        pconfig,
+        existing_key,
+        provider_id="openrouter",
+        existing_source=existing_source,
+    )
     if abort:
         return
 
@@ -1975,14 +1985,13 @@ def _model_flow_kimi(config, current_model=""):
     base_url_env = pconfig.base_url_env_var or ""
 
     # Step 1: Check / prompt for API key
-    existing_key = ""
-    for ev in pconfig.api_key_env_vars:
-        existing_key = get_env_value(ev) or os.getenv(ev, "")
-        if existing_key:
-            break
+    existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig)
 
     existing_key, abort = _prompt_api_key(
-        pconfig, existing_key, provider_id=provider_id
+        pconfig,
+        existing_key,
+        provider_id=provider_id,
+        existing_source=existing_source,
     )
     if abort:
         return
@@ -2060,14 +2069,13 @@ def _model_flow_stepfun(config, current_model=""):
     key_env = pconfig.api_key_env_vars[0] if pconfig.api_key_env_vars else ""
     base_url_env = pconfig.base_url_env_var or ""
 
-    existing_key = ""
-    for ev in pconfig.api_key_env_vars:
-        existing_key = get_env_value(ev) or os.getenv(ev, "")
-        if existing_key:
-            break
+    existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig)
 
     existing_key, abort = _prompt_api_key(
-        pconfig, existing_key, provider_id=provider_id
+        pconfig,
+        existing_key,
+        provider_id=provider_id,
+        existing_source=existing_source,
     )
     if abort:
         return
@@ -2642,14 +2650,13 @@ def _model_flow_api_key_provider(config, provider_id, current_model=""):
     base_url_env = pconfig.base_url_env_var or ""
 
     # Check / prompt for API key
-    existing_key = ""
-    for ev in pconfig.api_key_env_vars:
-        existing_key = get_env_value(ev) or os.getenv(ev, "")
-        if existing_key:
-            break
+    existing_key, existing_source = _existing_api_key_for_model_flow(provider_id, pconfig)
 
     existing_key, abort = _prompt_api_key(
-        pconfig, existing_key, provider_id=provider_id
+        pconfig,
+        existing_key,
+        provider_id=provider_id,
+        existing_source=existing_source,
     )
     if abort:
         return

tests/hermes_cli/test_model_flow_pooled_credentials.py

@@ -0,0 +1,129 @@
+"""Pool-only credentials must be visible to interactive model setup flows."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from hermes_cli.auth import PROVIDER_REGISTRY
+from hermes_cli.model_setup_flows import _existing_api_key_for_model_flow
+
+
+class _PoolEntry:
+    access_token = "pool-secret"
+    runtime_api_key = ""
+
+
+class _AvailablePool:
+    def has_credentials(self) -> bool:
+        return True
+
+    def peek(self):
+        return _PoolEntry()
+
+
+class _ExhaustedPool:
+    def has_credentials(self) -> bool:
+        return True
+
+    def peek(self):
+        return None
+
+
+def test_existing_key_precedence_is_dotenv_then_process_then_pool(tmp_path, monkeypatch):
+    pconfig = PROVIDER_REGISTRY["deepseek"]
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    monkeypatch.setenv("DEEPSEEK_API_KEY", "process-secret")
+    (hermes_home / ".env").write_text("DEEPSEEK_API_KEY=dotenv-secret\n", encoding="utf-8")
+
+    with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()):
+        assert _existing_api_key_for_model_flow("deepseek", pconfig) == (
+            "dotenv-secret",
+            "DEEPSEEK_API_KEY",
+        )
+
+    (hermes_home / ".env").write_text("", encoding="utf-8")
+    with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()):
+        assert _existing_api_key_for_model_flow("deepseek", pconfig) == (
+            "process-secret",
+            "DEEPSEEK_API_KEY",
+        )
+
+    monkeypatch.delenv("DEEPSEEK_API_KEY")
+    with patch("agent.credential_pool.load_pool", return_value=_AvailablePool()):
+        assert _existing_api_key_for_model_flow("deepseek", pconfig) == (
+            "pool-secret",
+            "credential_pool:deepseek",
+        )
+
+
+def test_exhausted_pool_is_not_an_existing_key(monkeypatch):
+    pconfig = PROVIDER_REGISTRY["deepseek"]
+    monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
+    with (
+        patch("hermes_cli.config.get_env_value", return_value=""),
+        patch("agent.credential_pool.load_pool", return_value=_ExhaustedPool()),
+    ):
+        assert _existing_api_key_for_model_flow("deepseek", pconfig) == ("", "")
+
+
+def test_generic_api_key_flow_passes_pool_key_to_existing_key_prompt(monkeypatch):
+    from hermes_cli.model_setup_flows import _model_flow_api_key_provider
+
+    monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
+    captured: dict[str, str] = {}
+
+    def capture_prompt(_pconfig, existing_key, **_kwargs):
+        captured["existing_key"] = existing_key
+        return existing_key, True
+
+    with (
+        patch("hermes_cli.config.get_env_value", return_value=""),
+        patch("agent.credential_pool.load_pool", return_value=_AvailablePool()),
+        patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt),
+    ):
+        _model_flow_api_key_provider({}, "deepseek")
+
+    assert captured["existing_key"] == "pool-secret"
+
+
+def test_kimi_flow_passes_pool_key_to_existing_key_prompt(monkeypatch):
+    from hermes_cli.model_setup_flows import _model_flow_kimi
+
+    monkeypatch.delenv("KIMI_API_KEY", raising=False)
+    monkeypatch.delenv("MOONSHOT_API_KEY", raising=False)
+    captured: dict[str, str] = {}
+
+    def capture_prompt(_pconfig, existing_key, **_kwargs):
+        captured["existing_key"] = existing_key
+        return existing_key, True
+
+    with (
+        patch("hermes_cli.config.get_env_value", return_value=""),
+        patch("agent.credential_pool.load_pool", return_value=_AvailablePool()),
+        patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt),
+    ):
+        _model_flow_kimi({})
+
+    assert captured["existing_key"] == "pool-secret"
+
+
+def test_exhausted_pool_still_uses_first_time_prompt(monkeypatch):
+    from hermes_cli.model_setup_flows import _model_flow_api_key_provider
+
+    monkeypatch.delenv("DEEPSEEK_API_KEY", raising=False)
+    captured: dict[str, str] = {}
+
+    def capture_prompt(_pconfig, existing_key, **_kwargs):
+        captured["existing_key"] = existing_key
+        return "", True
+
+    with (
+        patch("hermes_cli.config.get_env_value", return_value=""),
+        patch("agent.credential_pool.load_pool", return_value=_ExhaustedPool()),
+        patch("hermes_cli.main._prompt_api_key", side_effect=capture_prompt),
+    ):
+        _model_flow_api_key_provider({}, "deepseek")
+
+    assert captured["existing_key"] == ""

tests/hermes_cli/test_prompt_api_key.py

@@ -37,6 +37,33 @@ def _run_prompt(existing_key, choice, new_key="", provider_id="", pconfig_name="
         return m._prompt_api_key(pconfig, existing_key, provider_id=provider_id)
 
 
+def test_pool_only_key_does_not_offer_or_execute_clear(profile_env, monkeypatch, capsys):
+    from hermes_cli import main as m
+
+    pconfig = _pconfig("deepseek")
+    prompts = []
+
+    def choose_clear(prompt):
+        prompts.append(prompt)
+        return "c"
+
+    monkeypatch.setattr("builtins.input", choose_clear)
+    with patch("hermes_cli.config.save_env_value") as save_env:
+        key, abort = m._prompt_api_key(
+            pconfig,
+            "pool-secret",
+            provider_id="deepseek",
+            existing_source="credential_pool:deepseek",
+        )
+
+    assert key == "pool-secret"
+    assert abort is False
+    assert prompts and "[C]lear" not in prompts[0]
+    assert "[K]eep / [R]eplace" in prompts[0]
+    save_env.assert_not_called()
+    assert "API key cleared" not in capsys.readouterr().out
+
+
 # First-time entry ────────────────────────────────────────────────────────────
 
 def test_first_time_save_new_key(profile_env):