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

fix(fallback): restore primary route after picker errors

Browse Source
This commit is contained in:
Xipong
2026-08-02 05:20:19 +03:00
committed by Teknium
parent bc31e04d83
commit 33cefdd8e5
2 changed files with 196 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hermes_cli/fallback_cmd.py
+51 -34
View File
@@ -13,6 +13,9 @@ from hermes_cli.fallback_config import get_fallback_chain
_read_chain = get_fallback_chain
_MISSING_ACTIVE_PROVIDER = object()
def _identity(entry: Dict[str, Any]):
"""BackendIdentity for a ``{provider, model, base_url?}`` entry."""
from agent.backend_identity import BackendIdentity
@@ -45,23 +48,25 @@ def _extract_fallback_from_model_cfg(model_cfg: Any) -> Optional[Dict[str, Any]]
def _snapshot_auth_active_provider() -> Any:
"""Current ``active_provider`` in auth.json, or None if unavailable."""
try:
from hermes_cli.auth import _load_auth_store
return _load_auth_store().get("active_provider")
except Exception:
return None
"""Return the current ``active_provider`` in auth.json."""
from hermes_cli.auth import _auth_store_lock, _load_auth_store
with _auth_store_lock():
store = _load_auth_store()
return store.get("active_provider", _MISSING_ACTIVE_PROVIDER)
def _restore_auth_active_provider(value: Any) -> None:
"""Write back a snapshotted ``active_provider``; best-effort (user re-runs `hermes model`), never fails the add."""
try:
from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
with _auth_store_lock():
_save_auth_store({**_load_auth_store(), "active_provider": value})
except Exception:
pass
"""Write back a previously snapshotted ``active_provider`` value."""
from hermes_cli.auth import _auth_store_lock, _load_auth_store, _save_auth_store
with _auth_store_lock():
store = _load_auth_store()
if value is _MISSING_ACTIVE_PROVIDER:
store.pop("active_provider", None)
else:
store["active_provider"] = value
_save_auth_store(store)
def _restore_model_cfg(model_before: Any) -> None:
"""Restore ``config["model"]`` to a previously-captured snapshot."""
@@ -73,6 +78,21 @@ def _restore_model_cfg(model_before: Any) -> None:
save_config(cfg)
def _restore_primary_route(model_before: Any, active_provider_before: Any) -> None:
"""Attempt both halves of temporary picker-route restoration."""
errors: list[BaseException] = []
try:
_restore_model_cfg(model_before)
except BaseException as exc:
errors.append(exc)
try:
_restore_auth_active_provider(active_provider_before)
except BaseException as exc:
errors.append(exc)
if errors:
details = "; ".join(str(exc) for exc in errors)
raise RuntimeError(f"Could not fully restore the primary route: {details}") from errors[0]
def _entries(n: int) -> str:
return f"{n} {'entry' if n == 1 else 'entries'}"
@@ -122,45 +142,43 @@ def cmd_fallback_add(args) -> None:
from hermes_cli.config import load_config, save_config
_require_tty("fallback add")
# Snapshot BEFORE the picker runs: "picked" vs "cancelled" is decided by comparing before/after,
# and the primary must be restored either way.
# Snapshot BEFORE the picker runs; both route stores must be restored on every exit path.
model_before = copy.deepcopy(load_config().get("model"))
active_provider_before = _snapshot_auth_active_provider()
print("\n Adding a fallback provider. The picker below is the same one used by\n"
" `hermes model` — select the provider + model you want as a fallback.\n")
def _restore() -> None:
_restore_model_cfg(model_before)
_restore_auth_active_provider(active_provider_before)
try:
select_provider_and_model(args=args)
except SystemExit: # some provider flows exit on auth failure — restore state and re-raise
_restore()
after_cfg = load_config()
model_after = after_cfg.get("model")
new_entry = _extract_fallback_from_model_cfg(model_after)
except BaseException as picker_error:
try:
_restore_primary_route(model_before, active_provider_before)
except Exception as restore_error:
picker_error.add_note(
"Could not fully restore the primary route after fallback "
f"selection failed: {restore_error}"
)
raise
new_entry = _extract_fallback_from_model_cfg(load_config().get("model"))
if not new_entry: # picker didn't complete (user cancelled or flow bailed)
_restore()
# From here onward no identity/import/append failure can strand the temporary picker route.
_restore_primary_route(model_before, active_provider_before)
if not new_entry:
print("\n No fallback added.")
return
# Same deployment as the primary → nothing to add. Identity semantics are owned by
# agent.backend_identity: same provider+model on a DIFFERENT explicit base_url is a different
# backend (multi-endpoint pool) and a legitimate fallback.
# Picker picked the same thing that's already the primary → nothing changed, and there's nothing useful
# to add as a fallback to itself. See #54250, #57584, #62984.
from agent.backend_identity import same_deployment
new_ident = _identity(new_entry)
primary_entry = _extract_fallback_from_model_cfg(model_before)
if primary_entry and same_deployment(_identity(primary_entry), new_ident):
_restore()
print(f"\n Selected model matches the current primary ({_format_entry(new_entry)}).")
print(" A provider cannot be a fallback for itself — no change.")
return
# Restore the primary, then re-load (rather than mutating the post-picker config) because the
# picker may have touched other top-level keys (custom_providers, credentials) we want to keep.
_restore()
# Reload after primary restoration; picker-created providers/credentials remain.
final_cfg = load_config()
chain = _read_chain(final_cfg)
if any(same_deployment(_identity(existing), new_ident) for existing in chain):
@@ -173,7 +191,6 @@ def cmd_fallback_add(args) -> None:
print(f" Chain is now {_entries(len(chain))} long.\n")
print(" Run `hermes fallback list` to view, or `hermes fallback remove` to delete.")
def cmd_fallback_remove(args) -> None: # noqa: ARG001
"""Pick an entry from the chain and remove it."""
from hermes_cli.config import save_config
tests/hermes_cli/test_fallback_cmd.py
+145 -1
View File
@@ -15,7 +15,6 @@ import yaml
@pytest.fixture()
def isolated_home(tmp_path, monkeypatch):
monkeypatch.setattr(Path, "home", lambda: tmp_path)
home = tmp_path / ".hermes"
home.mkdir(exist_ok=True)
monkeypatch.setenv("HERMES_HOME", str(home))
@@ -118,6 +117,20 @@ class TestListCommand:
# ---------------------------------------------------------------------------
class TestAddCommand:
def test_auth_snapshot_failure_aborts_before_picker(self):
from hermes_cli import fallback_cmd
picker = object()
with patch(
"hermes_cli.auth._load_auth_store",
side_effect=OSError("auth read failed"),
), patch(
"hermes_cli.main.select_provider_and_model",
picker,
), patch("hermes_cli.main._require_tty"):
with pytest.raises(OSError, match="auth read failed"):
fallback_cmd.cmd_fallback_add(types.SimpleNamespace())
def test_add_appends_new_entry(self, isolated_home, capsys):
_write_config(isolated_home, {
"model": {"provider": "anthropic", "default": "claude-sonnet-4-6"},
@@ -199,6 +212,13 @@ class TestAddCommand:
"base_url": "https://openrouter.ai/api/v1",
"api_mode": "chat_completions",
}
cfg["custom_providers"] = [
{
"name": "Picker-created endpoint",
"base_url": "https://picker.example/v1",
"model": "picker-model",
}
]
save_config(cfg)
with patch("hermes_cli.main.select_provider_and_model", side_effect=fake_picker), \
@@ -215,6 +235,130 @@ class TestAddCommand:
# Fallback added
assert len(cfg["fallback_providers"]) == 1
assert cfg["fallback_providers"][0]["provider"] == "openrouter"
assert cfg["custom_providers"] == [
{
"name": "Picker-created endpoint",
"base_url": "https://picker.example/v1",
"model": "picker-model",
}
]
def test_post_picker_config_read_failure_restores_route(self):
from hermes_cli import fallback_cmd
primary_model = {
"provider": "anthropic",
"default": "claude-sonnet-4-6",
}
post_read_error = OSError("post-picker config read failed")
reads = iter([{"model": primary_model}, post_read_error])
def load_config():
value = next(reads)
if isinstance(value, BaseException):
raise value
return value
restore_calls = []
with patch("hermes_cli.config.load_config", side_effect=load_config), patch(
"hermes_cli.main._require_tty"
), patch("hermes_cli.main.select_provider_and_model"), patch.object(
fallback_cmd, "_snapshot_auth_active_provider", return_value="old-provider"
), patch.object(
fallback_cmd,
"_restore_primary_route",
side_effect=lambda model, provider: restore_calls.append((model, provider)),
):
with pytest.raises(OSError) as exc_info:
fallback_cmd.cmd_fallback_add(types.SimpleNamespace())
assert exc_info.value is post_read_error
assert restore_calls == [(primary_model, "old-provider")]
@pytest.mark.parametrize(
("failure_type", "message"),
[
(OSError, "config write failed"),
(KeyboardInterrupt, "config restore interrupted"),
],
)
def test_restore_attempts_auth_after_model_restore_failure(
self, failure_type, message
):
from hermes_cli import fallback_cmd
auth_calls = []
with patch.object(
fallback_cmd,
"_restore_model_cfg",
side_effect=failure_type(message),
), patch.object(
fallback_cmd,
"_restore_auth_active_provider",
side_effect=lambda value: auth_calls.append(value),
):
with pytest.raises(RuntimeError, match=message):
fallback_cmd._restore_primary_route("old-model", "old-provider")
assert auth_calls == ["old-provider"]
def test_restore_preserves_absent_active_provider(self):
from contextlib import nullcontext
from hermes_cli import auth, fallback_cmd
store = {"version": 1, "providers": {}}
def save_auth(value):
store.clear()
store.update(value)
with patch.object(auth, "_load_auth_store", lambda: dict(store)), patch.object(
auth, "_save_auth_store", save_auth
), patch.object(auth, "_auth_store_lock", nullcontext):
before = fallback_cmd._snapshot_auth_active_provider()
fallback_cmd._restore_auth_active_provider(before)
assert "active_provider" not in store
def test_picker_failure_restores_persisted_primary_without_masking_error(
self, isolated_home
):
from hermes_cli import fallback_cmd
primary_model = {
"provider": "anthropic",
"default": "claude-sonnet-4-6",
"base_url": "https://api.anthropic.com",
"api_mode": "anthropic_messages",
}
_write_config(isolated_home, {"model": primary_model, "theme": "midnight"})
picker_error = LookupError("picker failed")
def failing_picker(args=None):
from hermes_cli.config import load_config, save_config
cfg = load_config()
cfg["model"] = {
"provider": "openrouter",
"default": "anthropic/claude-sonnet-4.6",
"base_url": "https://openrouter.ai/api/v1",
"api_mode": "chat_completions",
}
save_config(cfg)
raise picker_error
with patch(
"hermes_cli.main.select_provider_and_model",
side_effect=failing_picker,
), patch("hermes_cli.main._require_tty"):
with pytest.raises(LookupError, match="picker failed") as exc_info:
fallback_cmd.cmd_fallback_add(types.SimpleNamespace())
assert exc_info.value is picker_error
persisted = _read_config(isolated_home)
assert persisted["model"] == primary_model
assert persisted["theme"] == "midnight"
# ---------------------------------------------------------------------------