Fallback provider handling in agent and gateway sessions

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #61119 fix(gateway): reload fallback_providers for live messaging sessions (#60987 salvage)

  merged to main

  files: gateway/run.py, tests/gateway/test_fallback_chain_reload.py

  Show the diff

  • gateway/run.py

    @@ -5008,6 +5008,74 @@ def _load_fallback_model() -> list | None:
                 pass
             return None
     
    +    def _refresh_fallback_model(self) -> list | None:
    +        """Re-read fallback_providers from disk for the next agent create/reuse.
    +
    +        Cron already does this per job via ``get_fallback_chain``; the gateway
    +        previously froze ``self._fallback_model`` at process start, so a chain
    +        configured (or changed) after ``hermes gateway`` was running never
    +        reached messaging sessions even though the same process's cron jobs
    +        fell back correctly. Fixes #60955.
    +
    +        A TRANSIENT read/parse failure (user mid-edit of config.yaml with a
    +        non-atomic write) keeps the last known-good chain instead of wiping a
    +        cached agent's working fallback for that turn.  Only a successful read
    +        that genuinely lacks the key clears the chain.
    +        """
    +        try:
    +            import yaml as _y
    +            cfg_path = _hermes_home / "config.yaml"
    +            if not cfg_path.exists():
    +                self._fallback_model = None
    +                return self._fallback_model
    +            with open(cfg_path, encoding="utf-8") as _f:
    +                cfg = _y.safe_load(_f) or {}
    +        except Exception:
    +            # Transient failure — keep last known-good chain.
    +            logger.debug(
    +                "fallback_providers refresh: config.yaml read failed; "
    +                "keeping last known-good chain", exc_info=True,
    +            )
    +            return self._fallback_model
    +        self._fallback_model = get_fallback_chain(cfg) or None
    +        return self._fallback_model
    +
    +    @staticmethod
    +    def _apply_fallback_chain_to_agent(agent: Any, chain: list | None) -> None:
    +        """Keep a cached agent's fallback chain aligned with current config.
    +
    +        Skips rewrite while a cooldown is holding the agent on an already-
    +        activated fallback provider — ``restore_primary_runtime`` owns that
    +        turn-scoped lifecycle. When primary is active (or cooldown expired),
    +        replace the chain so mid-uptime ``fallback_providers`` edits take
    +        effect without requiring a gateway restart (#60955).
    +        """
    +        if agent is None:
    +            return
    +        new_chain = list(chain or [])
    +        rate_limited_until = getattr(agent, "_rate_limited_until", 0) or 0
    +        if (
    +            getattr(agent, "_fallback_activated", False)
    +            and rate_limited_until > time.monotonic()
    +        ):
    +            return
    +        old_chain = list(getattr(agent, "_fallback_chain", []) or [])
    +        agent._fallback_chain = new_chain
    +        agent._fallback_model = new_chain[0] if new_chain else None
    +        if not getattr(agent, "_fallback_activated", False):
    +            agent._fallback_index = 0
    +        # A config edit signals the user changed something — drop the
    +        # session-scoped unavailability memo so re-configured entries
    +        # (e.g. credentials added mid-uptime for a previously-failing
    +        # provider) get retried instead of staying suppressed for the
    +        # cached agent's lifetime.  Only on actual content change, so
    +        # the per-message no-op refresh keeps the memo's rate-limiting
    +        # benefit (#60955).
    +        if new_chain != old_chain:
    +            unavailable = getattr(agent, "_unavailable_fallback_keys", None)
    +            if unavailable:
    +                unavailable.clear()
    +
         def _snapshot_running_agents(self) -> Dict[str, Any]:
             return {
                 session_key: agent
    @@ -13244,7 +13312,8 @@ def run_sync():
                         chat_type=source.chat_type,
                         thread_id=source.thread_id,
                         session_db=getattr(self._session_db, "_db", self._session_db),
    -                    fallback_model=self._fallback_model,
    +                    # Reload from disk — do not reuse the startup snapshot (#60955).
    +                    fallback_model=self._refresh_fallback_model(),
                     )
                     try:
                         return agent.run_conversation(
    @@ -18065,6 +18134,18 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                                 logger.debug("Reusing cached agent for session %s", session_key)
                                 reused_cached_agent = True
     
    +            # Lock released — refresh the fallback chain from disk for the
    +            # reused agent OUTSIDE the cache lock (config.yaml read is disk
    +            # I/O; the idle-sweep watcher contends on this lock and stalls
    +            # Discord heartbeats — same reasoning as #52197).  A chain
    +            # configured after this agent was cached (or after gateway start)
    +            # must reach the next turn (#60955).  Per-session turn
    +            # serialization (_running_agents) keeps this safe post-lock.
    +            if reused_cached_agent and agent is not None:
    +                self._apply_fallback_chain_to_agent(
    +                    agent, self._refresh_fallback_model(),
    +                )
    +
                 # Lock released — now schedule cleanup of any cross-process-evicted
                 # agent on a daemon thread so memory-provider shutdown / socket
                 # teardown never blocks the gateway event loop or the cache lock
    @@ -18117,7 +18198,8 @@ def _interim_assistant_cb(text: str, *, already_streamed: bool = False) -> None:
                         thread_id=source.thread_id,
                         gateway_session_key=session_key,
                         session_db=getattr(self._session_db, "_db", self._session_db),
    -                    fallback_model=self._fallback_model,
    +                    # Reload from disk — do not reuse the startup snapshot (#60955).
    +                    fallback_model=self._refresh_fallback_model(),
                     )
                     if _cache_lock and _cache is not None:
                         with _cache_lock:

  • tests/gateway/test_fallback_chain_reload.py

    @@ -0,0 +1,239 @@
    +"""Regression tests for #60955: gateway must not freeze fallback_providers.
    +
    +Cron reloads ``fallback_providers`` from disk on every job. The gateway used to
    +freeze ``self._fallback_model`` at process start, so a chain configured (or
    +edited) after ``hermes gateway`` was already running never reached messaging
    +sessions — even though cron in the same process fell back correctly.
    +
    +These tests pin the reload + cached-agent apply helpers without driving the
    +full Feishu session path.
    +"""
    +
    +from __future__ import annotations
    +
    +import time
    +from types import SimpleNamespace
    +
    +
    +def test_refresh_fallback_model_rereads_config(tmp_path, monkeypatch):
    +    from gateway.run import GatewayRunner
    +
    +    monkeypatch.setattr("gateway.run._hermes_home", tmp_path)
    +    cfg = tmp_path / "config.yaml"
    +    cfg.write_text(
    +        "fallback_providers:\n"
    +        "  - provider: deepseek\n"
    +        "    model: deepseek-v4-flash\n"
    +    )
    +
    +    runner = SimpleNamespace(
    +        _fallback_model=None,
    +    )
    +    runner._load_fallback_model = GatewayRunner._load_fallback_model
    +    bound = GatewayRunner._refresh_fallback_model.__get__(runner)
    +    chain = bound()
    +
    +    assert chain == [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +    assert runner._fallback_model == chain
    +
    +    cfg.write_text(
    +        "fallback_providers:\n"
    +        "  - provider: openrouter\n"
    +        "    model: anthropic/claude-sonnet-4.6\n"
    +    )
    +    updated = bound()
    +    assert updated == [
    +        {"provider": "openrouter", "model": "anthropic/claude-sonnet-4.6"}
    +    ]
    +    assert runner._fallback_model == updated
    +
    +
    +def test_refresh_fallback_model_clears_when_config_removed(tmp_path, monkeypatch):
    +    from gateway.run import GatewayRunner
    +
    +    monkeypatch.setattr("gateway.run._hermes_home", tmp_path)
    +    cfg = tmp_path / "config.yaml"
    +    cfg.write_text(
    +        "fallback_providers:\n"
    +        "  - provider: deepseek\n"
    +        "    model: deepseek-v4-flash\n"
    +    )
    +
    +    runner = SimpleNamespace(
    +        _fallback_model=[{"provider": "stale", "model": "x"}],
    +    )
    +    runner._load_fallback_model = GatewayRunner._load_fallback_model
    +    bound = GatewayRunner._refresh_fallback_model.__get__(runner)
    +    assert bound() is not None
    +
    +    cfg.write_text("model:\n  provider: nvidia\n")
    +    assert bound() is None
    +    assert runner._fallback_model is None
    +
    +
    +def test_refresh_fallback_model_keeps_last_known_good_on_read_failure(
    +    tmp_path, monkeypatch,
    +):
    +    """A transient config.yaml read/parse failure (user mid-edit, non-atomic
    +    write) must NOT wipe the last known-good chain — only a successful read
    +    that genuinely lacks the key clears it."""
    +    from gateway.run import GatewayRunner
    +
    +    monkeypatch.setattr("gateway.run._hermes_home", tmp_path)
    +    cfg = tmp_path / "config.yaml"
    +    cfg.write_text(
    +        "fallback_providers:\n"
    +        "  - provider: deepseek\n"
    +        "    model: deepseek-v4-flash\n"
    +    )
    +
    +    runner = SimpleNamespace(_fallback_model=None)
    +    runner._load_fallback_model = GatewayRunner._load_fallback_model
    +    bound = GatewayRunner._refresh_fallback_model.__get__(runner)
    +    good = bound()
    +    assert good == [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +
    +    # Simulate a mid-edit torn write: invalid YAML.
    +    cfg.write_text("fallback_providers:\n  - provider: [unclosed\n")
    +    assert bound() == good
    +    assert runner._fallback_model == good
    +
    +
    +def test_apply_fallback_chain_updates_primary_agent():
    +    from gateway.run import GatewayRunner
    +
    +    agent = SimpleNamespace(
    +        _fallback_chain=[],
    +        _fallback_model=None,
    +        _fallback_index=0,
    +        _fallback_activated=False,
    +        _rate_limited_until=0,
    +    )
    +    chain = [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +    GatewayRunner._apply_fallback_chain_to_agent(agent, chain)
    +
    +    assert agent._fallback_chain == chain
    +    assert agent._fallback_model == chain[0]
    +    assert agent._fallback_index == 0
    +
    +
    +def test_apply_fallback_chain_skips_while_cooldown_holds_fallback():
    +    """Do not clobber a live fallback activation during its cooldown window."""
    +    from gateway.run import GatewayRunner
    +
    +    live = [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +    agent = SimpleNamespace(
    +        _fallback_chain=live,
    +        _fallback_model=live[0],
    +        _fallback_index=1,
    +        _fallback_activated=True,
    +        _rate_limited_until=time.monotonic() + 30,
    +    )
    +    GatewayRunner._apply_fallback_chain_to_agent(
    +        agent,
    +        [{"provider": "openrouter", "model": "anthropic/claude-sonnet-4.6"}],
    +    )
    +
    +    assert agent._fallback_chain == live
    +    assert agent._fallback_index == 1
    +    assert agent._fallback_activated is True
    +
    +
    +def test_apply_fallback_chain_updates_after_cooldown_expires():
    +    from gateway.run import GatewayRunner
    +
    +    agent = SimpleNamespace(
    +        _fallback_chain=[{"provider": "deepseek", "model": "old"}],
    +        _fallback_model={"provider": "deepseek", "model": "old"},
    +        _fallback_index=1,
    +        _fallback_activated=True,
    +        _rate_limited_until=time.monotonic() - 1,
    +    )
    +    new_chain = [{"provider": "openrouter", "model": "anthropic/claude-sonnet-4.6"}]
    +    GatewayRunner._apply_fallback_chain_to_agent(agent, new_chain)
    +
    +    assert agent._fallback_chain == new_chain
    +    assert agent._fallback_model == new_chain[0]
    +    # Activated agents keep their index; restore_primary_runtime owns reset.
    +    assert agent._fallback_index == 1
    +
    +
    +def test_apply_fallback_chain_clears_unavailable_memo_on_content_change():
    +    """A config edit must drop the session-scoped unavailability memo so a
    +    re-configured entry (credentials added mid-uptime) is retried instead of
    +    staying suppressed for the cached agent's lifetime."""
    +    from gateway.run import GatewayRunner
    +
    +    agent = SimpleNamespace(
    +        _fallback_chain=[{"provider": "deepseek", "model": "old"}],
    +        _fallback_model={"provider": "deepseek", "model": "old"},
    +        _fallback_index=0,
    +        _fallback_activated=False,
    +        _rate_limited_until=0,
    +        _unavailable_fallback_keys={("deepseek", "old", "")},
    +    )
    +    new_chain = [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +    GatewayRunner._apply_fallback_chain_to_agent(agent, new_chain)
    +
    +    assert agent._fallback_chain == new_chain
    +    assert agent._unavailable_fallback_keys == set()
    +
    +
    +def test_apply_fallback_chain_keeps_unavailable_memo_when_unchanged():
    +    """The per-message no-op refresh must NOT clear the memo — it exists to
    +    rate-limit repeated activation attempts against dead entries."""
    +    from gateway.run import GatewayRunner
    +
    +    chain = [{"provider": "deepseek", "model": "deepseek-v4-flash"}]
    +    memo = {("deepseek", "deepseek-v4-flash", "")}
    +    agent = SimpleNamespace(
    +        _fallback_chain=list(chain),
    +        _fallback_model=chain[0],
    +        _fallback_index=0,
    +        _fallback_activated=False,
    +        _rate_limited_until=0,
    +        _unavailable_fallback_keys=set(memo),
    +    )
    +    GatewayRunner._apply_fallback_chain_to_agent(agent, list(chain))
    +
    +    assert agent._unavailable_fallback_keys == memo
    +
    +
    +def test_background_and_main_agent_paths_call_refresh():
    +    """Both AIAgent construction sites must pass a refreshed chain, not the
    +    startup snapshot, and the cached-agent reuse path must apply the refreshed
    +    chain. Source-level invariant for call sites that resist unit testing.
    +    """
    +    from pathlib import Path
    +
    +    source = (
    +        Path(__file__).resolve().parent.parent.parent / "gateway" / "run.py"
    +    ).read_text(encoding="utf-8")
    +    assert "fallback_model=self._refresh_fallback_model()" in source
    +    assert source.count("fallback_model=self._refresh_fallback_model()") >= 2
    +    # The cached-agent reuse path (the load-bearing fix for a long-lived
    +    # session in a running gateway) must apply the refreshed chain.
    +    assert "self._apply_fallback_chain_to_agent(" in source
    +    # The stale startup-snapshot form must not remain at create sites.
    +    assert "fallback_model=self._fallback_model," not in source
    +
    +
    +def test_load_fallback_model_static_unchanged_contract(tmp_path, monkeypatch):
    +    """_load_fallback_model remains a pure static reader used by refresh."""
    +    from gateway.run import GatewayRunner
    +
    +    monkeypatch.setattr("gateway.run._hermes_home", tmp_path)
    +    (tmp_path / "config.yaml").write_text(
    +        "fallback_providers:\n"
    +        "  - provider: deepseek\n"
    +        "    model: deepseek-v4-flash\n"
    +        "fallback_model:\n"
    +        "  provider: nous\n"
    +        "  model: Hermes-4\n"
    +    )
    +
    +    chain = GatewayRunner._load_fallback_model()
    +    assert chain == [
    +        {"provider": "deepseek", "model": "deepseek-v4-flash"},
    +        {"provider": "nous", "model": "Hermes-4"},
    +    ]