Free tier: a rate limit ends a watched turn at once with the reset time (#134049)

* fix(agent): free-tier rate limits end an attended turn at once

A free-tier (no sign-in) session that hit the service-wide rate limit slept
up to 600 s, retried into the same refusal, slept again, and then sent its
third attempt on a guest JWT that had expired during the wait. The welcome
host answered 403 "invalid jwt", which read as the tier being switched off.
The user watched a spinner for about 18 minutes and then saw the wrong card.

- An attended session (desktop, TUI, the interactive CLI, ACP) on the free
  model ends the attempt cycle at once when the cooldown is longer than
  60 s: fallback first, else the reset time and the sign-in / model doors.
  "Attended" is keyed on the session source, so `hermes chat -q`, cron and
  messaging sessions keep the long wait. The source vocabulary moves from
  run_agent.py into agent/session_source.py, which adds is_attended().
- on_free_model() (welcome host + anonymous credential) is the one free-tier
  check, used by the cutoff, the "free model is busy" wait copy and the
  outage copy.
- A welcome-host 403 "invalid jwt" is its own route kind, session_expired:
  the existing one-shot route heal re-exchanges the credential and retries;
  if that fails the card is the retryable outage card.
- interruptible_backoff_sleep adopts a fresh Nous key after each wait.
  _adopt_nous_key_before_expiry owns its refresh failure, so both call
  sites drop their try/except.
- A retry wait over 60 s is announced in the transcript when it starts, for
  every provider; shorter waits stay buffered. Every wait line names the
  attempt that runs after it.
- max_retries_exhausted_result's max_retries becomes attempts (what it
  carries); a one-attempt rate-limit end gets its own lead.
- One Retry-After reader, retry_utils.provider_retry_after_seconds, shared
  by compute_error_backoff and the recovery ladder.
- _session_default_route replaces _session_default_model everywhere in
  tui_gateway, so a not-yet-built free-tier session reports nous/welcome on
  create, reuse, resume, branch and row persist, not the silent default.

* fix(agent): free-tier cutoff no longer ends subagent or library turns, paid waits look like main (review)

A delegated child of a TUI/Desktop session and a library caller (no platform) are unattended, so they keep waiting.
Rate-limit wait lines keep main's attempt numbers and stay buffered for every provider; only a non-rate-limit Retry-After over 60 s is announced, as on main.
The unattended free-tier wait line keeps the reset time beside "The free model is busy".
The free-tier cutoff names the reset time for any cooldown it fires on (over 60 s), not only from 120 s.
session.info for an env-seeded model off the free tier carries the model alone again, as on main.
Behaviour tests for the cutoff, the subagent and unattended waits, and a paid long wait.

agent/agent_init.py

@@ -690,8 +690,8 @@ def _init_prompt_cache_config(agent):
         elif _ttl == AUTO_CACHE_TTL:
             # Decided once per session from its source (a delegated child is clamped to 5m again
             # in delegate_tool regardless).
-            from run_agent import _session_source_for_agent  # late: run_agent imports this module
-            agent._cache_ttl = auto_cache_ttl_for_source(_session_source_for_agent(getattr(agent, "platform", None)))
+            from agent.session_source import session_source_for
+            agent._cache_ttl = auto_cache_ttl_for_source(session_source_for(getattr(agent, "platform", None)))
         elif cache_ttl_means_disabled(_ttl):
             agent._use_prompt_caching = False
             agent._use_native_cache_layout = False

agent/client_lifecycle.py

@@ -676,7 +676,11 @@ class ClientLifecycleMixin:
         exp, account = claims.get("exp"), claims.get("sub")
         if not account or not isinstance(exp, (int, float)) or exp - time.time() > self._NOUS_KEY_ADOPT_SKEW_S:
             return False
-        return self._try_refresh_nous_client_credentials(force=False, require_account=str(account))
+        try:
+            return self._try_refresh_nous_client_credentials(force=False, require_account=str(account))
+        except Exception:  # noqa: BLE001 — a failed adoption leaves the key in hand; the 401 path still heals
+            logger.debug("Nous key pre-expiry adoption failed", exc_info=True)
+            return False
 
 
     def _resolve_env_credentials(self) -> Optional[tuple]:

agent/conversation_compression.py

@@ -3298,8 +3298,8 @@ def _compression_child_source(agent: Any, parent_session_id: str) -> str:
         parent = agent._session_db.get_session(parent_session_id)
     if parent and parent.get("source"):
         return parent["source"]
-    from run_agent import _session_source_for_agent  # late: run_agent imports this module
-    return _session_source_for_agent(getattr(agent, "platform", None))
+    from agent.session_source import session_source_for
+    return session_source_for(getattr(agent, "platform", None))
 
 
 def _publish_rotated_compaction(

agent/prompt_cache_scope.py

@@ -57,16 +57,8 @@ def _agent_source(
     if row_source:
         return row_source
     platform = getattr(agent, "platform", None)
-    try:
-        # Lazy: run_agent imports this module.
-        from run_agent import _session_source_for_agent
-
-        source = str(_session_source_for_agent(platform) or "").strip()
-        if source:
-            return source
-    except Exception:
-        logger.debug("declared-scope source authority unavailable", exc_info=True)
-    return str(platform or "").strip()
+    from agent.session_source import session_source_for
+    return str(session_source_for(platform)).strip()
 
 
 def _conversation_generation(session_key: str, source: str, session_db: Any) -> str:

agent/retry_utils.py

@@ -172,3 +172,24 @@ def zai_coding_overload_retry_ceiling(short_attempts: int = _ZAI_CODING_OVERLOAD
     because the loop gives up when ``retry_count >= ceiling`` BEFORE computing the attempt's
     backoff (the default ``api_max_retries`` of 3 equals ``short_attempts``)."""
     return short_attempts + len(_ZAI_CODING_OVERLOAD_LONG_BACKOFF) + 1
+
+
+# A wait longer than this is one a person feels: a non-rate-limit Retry-After this long is announced
+# when it starts, and on the Nous free tier an attended session ends the turn instead of sitting through it.
+LIVE_RETRY_WAIT_CAP_S = 60.0
+# Anthropic Tier 1 input-token buckets reset in ~171s, so a 120s cap re-tripped the limit; 600s
+# covers realistic provider windows while still rejecting pathological values (#26293).
+RETRY_AFTER_CAP_S = 600.0
+
+
+def provider_retry_after_seconds(error: Any) -> Optional[float]:
+    """Provider-declared cooldown: the ``Retry-After`` header, else a ``retry_after`` body field
+    (top level or nested under ``error``). None when absent, unparseable or zero: a zero or expired
+    cooldown carries no usable wait, and treating it as one would hot-loop the provider."""
+    value = parse_retry_after_seconds(getattr(getattr(error, "response", None), "headers", None))
+    if value is None:
+        body = getattr(error, "body", None)
+        if isinstance(body, dict):
+            nested = body.get("error")
+            value = parse_retry_after_seconds((nested if isinstance(nested, dict) else body).get("retry_after"))
+    return value if value is not None and value > 0 else None

agent/session_source.py

@@ -0,0 +1,44 @@
+"""Session source vocabulary: the ``source`` label a session row carries, and which sources have a
+person reading the turn as it runs. A leaf (its one import is the stdlib-only session context), so
+the ``run_agent`` facade and its siblings import it directly."""
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+from gateway.session_context import get_session_env
+
+# Sources that label the human conversation an interactive UI transport hosts. A finite ``hermes chat -q`` /
+# one-shot child spawned from such a session inherits HERMES_SESSION_SOURCE (the terminal tool bridges the
+# session env into child processes) but is NOT that conversation: labelling it ``tui``/``desktop`` lists it
+# in the TUI/WebUI pickers as a resumable chat and lets ``hermes -c`` in the TUI continue it (#112550).
+# Automation sources (kanban, tool, cron, a2a, ...) are inherited on purpose.
+UI_TRANSPORT_SOURCES = frozenset({"tui", "desktop"})
+
+# Finite non-interactive CLI runs (``hermes chat -q``/``--oneshot``, ``hermes -z``) get their own source so human
+# pickers hide them without title/cwd heuristics; ``hermes -c`` still treats them as CLI history.
+ONESHOT_SOURCE = "oneshot"
+CLI_FAMILY_SOURCES = frozenset({"cli", ONESHOT_SOURCE})
+
+# A person reads the turn as it runs: not cron, batch, a messaging chat, or a one-shot run.
+ATTENDED_SOURCES = UI_TRANSPORT_SOURCES | {"cli", "acp"}
+
+
+def session_source_for(platform: Optional[str]) -> str:
+    source = str(get_session_env("HERMES_SESSION_SOURCE", "") or "").strip()
+    single_query = get_session_env("HERMES_SINGLE_QUERY_SESSION", "") == "1"
+    explicit = get_session_env("HERMES_SESSION_SOURCE_EXPLICIT", "") == "1"
+    if single_query and not explicit and source in UI_TRANSPORT_SOURCES:
+        source = ""
+    if single_query and not source and (platform or "cli") == "cli":
+        return ONESHOT_SOURCE
+    return source or platform or "cli"
+
+
+def is_attended(agent: Any) -> bool:
+    """A delegated child answers its parent, not a person, though its copied context still carries
+    the parent's source; a library caller (no platform) has no one reading either."""
+    platform = getattr(agent, "platform", None)
+    if not platform or platform == "subagent":
+        return False
+    return session_source_for(platform) in ATTENDED_SOURCES

agent/turn_api_error.py

@@ -19,8 +19,8 @@ from agent.api_error_summary import is_provider_stream_parse_error
 from agent.error_classifier import RETRYABLE_CLIENT_REASONS, FailoverReason, classify_api_error
 from agent.turn_overflow import recover_from_overflow
 from agent.turn_recovery import (
-    _NONRETRYABLE_LABELS, abort_turn_on_interrupt, compute_error_backoff, interruptible_backoff_sleep,
-    log_api_error_attempt,
+    _NONRETRYABLE_LABELS, abort_turn_on_interrupt, compute_error_backoff, free_tier_cooldown_ends_turn,
+    interruptible_backoff_sleep, log_api_error_attempt,
     max_retries_exhausted_result, nonretryable_client_error_result, recover_after_classification,
     recover_before_classification, route_classified_error, settle_delivered_partial,
 )
@@ -353,6 +353,13 @@ def settle_unrecovered_error(
             base_url=_base, model=_model, delivered=_delivered,
         ))
 
+    # An attended session on the free model does not sit through a long cooldown: end the attempt
+    # cycle now (fallback, else the reset time and the ways forward); the copy counts attempts made.
+    attempts_made = min(retry_count, max_retries)
+    if is_rate_limited and retry_count < max_retries and free_tier_cooldown_ends_turn(agent, api_error, _base):
+        logger.info("%sFree-tier cooldown outlasts the attended wait — ending retries after attempt %s",
+                    agent.log_prefix, retry_count)
+        retry_count = max_retries
     if retry_count >= max_retries:
         # Before fallback, rebuild the primary client once per API call block for
         # transient transport errors (stale pool, TCP reset).
@@ -389,7 +396,7 @@ def settle_unrecovered_error(
             return _verdict(_ladder["action"], _ladder.get("result"))
         _delivered = settle_delivered_partial(agent, messages, current_turn_user_idx)
         return _verdict("return", max_retries_exhausted_result(
-            agent, api_error, classified, max_retries=max_retries, is_rate_limited=is_rate_limited,
+            agent, api_error, classified, attempts=attempts_made, is_rate_limited=is_rate_limited,
             error_msg=error_msg, api_kwargs=api_kwargs, api_messages=api_messages,
             messages=messages, conversation_history=conversation_history,
             api_call_count=api_call_count, approx_tokens=approx_tokens, provider=_provider,

agent/turn_failure_copy.py

@@ -177,6 +177,11 @@ _EXHAUSTED_LEADS: Dict[str, str] = {
     FailoverReason.timeout.value: "{label} didn't respond in time on any of {attempts} attempts",
 }
 _EXHAUSTED_DEFAULT_LEAD = "{label} didn't answer after {attempts} attempts"
+# One attempt: an attended free-tier session ended the cycle on a cooldown longer than it waits.
+_EXHAUSTED_FIRST_ATTEMPT_LEADS: Dict[str, str] = {
+    FailoverReason.rate_limit.value: "{label} is rate-limiting requests right now",
+    FailoverReason.upstream_rate_limit.value: "{label} is rate-limiting requests right now",
+}
 
 # Terminal copy for a non-retryable provider rejection, keyed by classifier reason.
 _NONRETRYABLE_COPY: Dict[str, str] = {
@@ -358,8 +363,11 @@ def site_copy(code: str, **fields: Any) -> str:
 def exhausted_copy(reason: str, *, label: str, attempts: int, summary: str, reset_seconds: Optional[float] = None) -> str:
     """Chat copy once retries + fallback are exhausted (``max_retries_exhausted_result``). A rate
     limit whose reset window is known names it: an 8.6h plan quota is not "wait a minute" (#89401)."""
-    lead = _EXHAUSTED_LEADS.get(reason, _EXHAUSTED_DEFAULT_LEAD).format(label=label, attempts=attempts)
-    if reset_seconds is not None and reset_seconds >= 120:
+    first_attempt = attempts == 1 and reason in _EXHAUSTED_FIRST_ATTEMPT_LEADS
+    lead = (_EXHAUSTED_FIRST_ATTEMPT_LEADS if first_attempt else _EXHAUSTED_LEADS).get(
+        reason, _EXHAUSTED_DEFAULT_LEAD).format(label=label, attempts=attempts)
+    # The free-tier cutoff (one attempt) fires on any cooldown over a minute, so it always names the reset.
+    if reset_seconds is not None and (reset_seconds >= 120 or first_attempt and reset_seconds > 0):
         from agent.retry_utils import format_reset_window
         situation = (f"its usage limit resets in {format_reset_window(reset_seconds)}. "
                      "Send /retry after that, or switch models with /model.")

agent/turn_iteration_prep.py

@@ -141,10 +141,7 @@ def prepare_iteration(
     # key before the one in hand expires (local JWT exp read; no network unless inside the skew)
     # instead of letting this iteration's request 401. With many agents sharing the hour that
     # 401 was a storm, and the pool benched the sole credential for all of them.
-    try:
-        agent._adopt_nous_key_before_expiry()
-    except Exception:
-        logger.debug("Nous key pre-expiry adoption failed", exc_info=True)
+    agent._adopt_nous_key_before_expiry()
 
     # Drain a /steer sent during the last API call so it lands THIS iteration. Delivered as a
     # standalone user row after the newest tool result (never smeared onto the tool row: that

agent/turn_recovery.py

@@ -593,6 +593,7 @@ def _recover_format_errors(
 _WELCOME_ROUTE_HEAL_COPY = {
     "anon_on_paid_host": "Reconnected to the free model's own route.",
     "named_on_welcome_host": "Reconnected to your Nous account's own route.",
+    "session_expired": "Renewed the connection to the free model.",
 }
 
 
@@ -604,7 +605,8 @@ def _recover_welcome_tier(agent: Any, classified: Any, _retry: TurnRetryState) -
     turn. ``anon_on_paid_host`` / ``named_on_welcome_host``: this process is pointed at the other
     identity's host (a stale route); re-read the credentials, which heals the URL, and retry. The
     refresh reports False when the store yields the same route, so a user-set
-    ``NOUS_INFERENCE_BASE_URL`` falls straight through to the terminal copy.
+    ``NOUS_INFERENCE_BASE_URL`` falls straight through to the terminal copy. ``session_expired``
+    (403 ``invalid jwt``): the same forced refresh re-exchanges the free credential.
 
     Reads the CLASSIFIER's context (``classified.error_context``): that is where
     ``_nous_welcome_tier`` parks ``welcome_refusal`` / ``welcome_route``. The turn's other context
@@ -949,6 +951,9 @@ def _welcome_surface_kind(classified: Any) -> str:
     route = ctx.get("welcome_route") if isinstance(ctx, dict) else None
     if route == "tier_disabled":
         return "disabled"
+    # The renewal failed but the tier is up: a plain Retry mints a new credential.
+    if route == "session_expired":
+        return "outage"
     # A named account on the welcome host has already signed in: no sign-in card, copy only.
     if route == "named_on_welcome_host":
         return ""
@@ -962,21 +967,16 @@ def _stamp_free_tier(result: Dict[str, Any], kind: str, message: str) -> Dict[st
     return result
 
 
-def _welcome_outage_copy(base_url: Any, classified: Any, *, anonymous: bool = False) -> str:
+def _welcome_outage_copy(agent: Any, base_url: Any, classified: Any) -> str:
     """On the Nous free tier, a transport / server failure that outlived every retry reads as one
     plain sentence (the free model is having trouble) rather than the technical summary. Empty
     for every other route and for rate limits / billing, which have their own copy."""
-    try:
-        from hermes_cli.anon_auth import FREE_TIER_OUTAGE_COPY, route_is_welcome_host
-        # Both: an anonymous JWT sent to a user-overridden paid host never reached the free model.
-        if not anonymous or not route_is_welcome_host(base_url):
-            return ""
-        # Not ``unknown``: that is the classifier's catch-all for status-less local failures, which
-        # are not the free model's trouble.
-        if classified.reason in (FailoverReason.timeout, FailoverReason.overloaded, FailoverReason.server_error):
-            return FREE_TIER_OUTAGE_COPY
-    except Exception:
-        pass
+    from hermes_cli.anon_auth import FREE_TIER_OUTAGE_COPY, on_free_model
+    # Not ``unknown``: that is the classifier's catch-all for status-less local failures, which
+    # are not the free model's trouble.
+    if on_free_model(agent, base_url) and classified.reason in (
+            FailoverReason.timeout, FailoverReason.overloaded, FailoverReason.server_error):
+        return FREE_TIER_OUTAGE_COPY
     return ""
 
 
@@ -1138,7 +1138,7 @@ _STREAM_DROP_MARKERS = (
 
 
 def max_retries_exhausted_result(
-    agent: Any, api_error: Exception, classified: Any, *, max_retries: int, is_rate_limited: bool,
+    agent: Any, api_error: Exception, classified: Any, *, attempts: int, is_rate_limited: bool,
     error_msg: str, api_kwargs: Any, api_messages: Any, messages: List[Dict[str, Any]],
     conversation_history: Any, api_call_count: int, approx_tokens: int, provider: Any,
     base_url: Any, model: Any, delivered: str = "",
@@ -1148,7 +1148,6 @@ def max_retries_exhausted_result(
     guidance (the latter wins), persist, build the result with ``failure_reason`` /
     ``failure_retryable`` / ``billing_block``."""
     # Result/guidance helpers stay in the loop module (tests import + patch them there).
-    from hermes_cli.anon_auth import is_anonymous_agent
     from agent.conversation_loop import (
         _billing_block_dict, _billing_or_entitlement_message, _billing_terminal_label,
         _print_billing_or_entitlement_guidance,
@@ -1176,11 +1175,11 @@ def max_retries_exhausted_result(
     elif is_rate_limited:
         _reset = reset_hint(api_error)
         agent._emit_diagnostic_status(
-            f"❌ Rate limited after {max_retries} retries — {_final_summary}"
+            f"❌ Rate limited after {attempts} {'attempt' if attempts == 1 else 'retries'} — {_final_summary}"
             f"{f' (resets in {_reset})' if _reset else ''}"
         )
     else:
-        agent._emit_diagnostic_status(f"❌ API failed after {max_retries} retries — {_final_summary}")
+        agent._emit_diagnostic_status(f"❌ API failed after {attempts} retries — {_final_summary}")
     _vlines(agent, f"   💀 Final error: {_final_summary}")
     _welcome_hint = _welcome_tier_guidance(classified, model=model, in_chat=False)
     if _welcome_hint:
@@ -1209,7 +1208,7 @@ def max_retries_exhausted_result(
 
     logger.error(
         "%sAPI call failed after %s retries. %s | provider=%s model=%s msgs=%s tokens=~%s",
-        agent.log_prefix, max_retries, _final_summary,
+        agent.log_prefix, attempts, _final_summary,
         provider, model, len(api_messages), f"{approx_tokens:,}",
     )
     if api_kwargs is not None:
@@ -1232,13 +1231,13 @@ def max_retries_exhausted_result(
         # text carries the plain what-happened + next step itself.
         _reset_at = classified.error_context.get("reset_at")
         _final_response = exhausted_copy(
-            classified.reason.value, label=provider_label_for(provider), attempts=max_retries,
+            classified.reason.value, label=provider_label_for(provider), attempts=attempts,
             summary=_final_summary, reset_seconds=_reset_at - time.time() if _reset_at else None,
         )
         if _welcome_hint:
             _final_response = _welcome_tier_guidance(classified, model=model, in_chat=True)
             _free_tier_kind = _welcome_surface_kind(classified)
-        elif _outage := _welcome_outage_copy(base_url, classified, anonymous=is_anonymous_agent(agent)):
+        elif _outage := _welcome_outage_copy(agent, base_url, classified):
             _final_response, _free_tier_kind = _outage, "outage"
     if _is_thinking_timeout:
         # Thinking-timeout guidance overrides stream-drop guidance, which would wrongly
@@ -1390,6 +1389,9 @@ def interruptible_backoff_sleep(
         _touch_counter += 1
         if _touch_counter % 150 == 0:  # 150 × 0.2s = 30s
             agent._touch_activity(f"{activity_label}, {int(sleep_end - time.time())}s remaining")
+    # A long wait can outlive a short-lived credential (the free tier's JWT lives 15 minutes), and
+    # the per-iteration pre-expiry adoption does not run between retries.
+    agent._adopt_nous_key_before_expiry()
     return None
 
 
@@ -1419,39 +1421,25 @@ def compute_error_backoff(
     is_zai_coding_overload: bool, base_url: Any, model: Any,
 ) -> float:
     """Pick the wait before the next API retry and announce it. Retry-After wins for
-    rate limits and any other retryable error (capped at 600s: Anthropic Tier 1 buckets
-    reset in ~171s, so a 120s cap re-tripped the limit); otherwise jittered backoff,
-    replaced by the adaptive policy for 429s / Z.AI overloads. Normal retries are
-    buffered; long Z.AI Coding waits surface immediately."""
+    rate limits and any other retryable error (capped at ``RETRY_AFTER_CAP_S``); otherwise
+    jittered backoff, replaced by the adaptive policy for 429s / Z.AI overloads. Retries are
+    buffered (replayed only if every retry fails); a long Z.AI Coding wait or a non-rate-limit
+    ``Retry-After`` over ``LIVE_RETRY_WAIT_CAP_S`` is announced when it starts."""
     # Imported lazily so tests that patch ``agent.retry_utils.jittered_backoff`` /
     # ``adaptive_rate_limit_backoff`` (incl. the run_agent conftest fast-backoff fixture) intercept.
-    from agent.retry_utils import adaptive_rate_limit_backoff, jittered_backoff, parse_retry_after_seconds
+    from agent.retry_utils import (
+        LIVE_RETRY_WAIT_CAP_S, RETRY_AFTER_CAP_S, adaptive_rate_limit_backoff, jittered_backoff,
+        provider_retry_after_seconds,
+    )
+    from hermes_cli.anon_auth import on_free_model
 
     # Respect Retry-After on every retryable provider error, not just 429s. Retryable
     # 5xx responses (e.g. Cloudflare 520/524) also carry the header or a structured
     # ``retry_after`` problem-detail body field; ignoring either turns an origin
     # outage into a retry storm.
-    _retry_after = parse_retry_after_seconds(
-        getattr(getattr(api_error, "response", None), "headers", None)
-    )
-    if _retry_after is None:
-        _error_body = getattr(api_error, "body", None)
-        if isinstance(_error_body, dict):
-            # Some providers nest it as error.retry_after (the same unwrap
-            # extract_api_error_context uses), others put it at the top level.
-            _nested = _error_body.get("error")
-            _payload = _nested if isinstance(_nested, dict) else _error_body
-            _retry_after = parse_retry_after_seconds(_payload.get("retry_after"))
+    _retry_after = provider_retry_after_seconds(api_error)
     if _retry_after is not None:
-        # Cap at 10 minutes. Anthropic Tier 1 input-token buckets reset in ~171s, so a 120s cap
-        # caused us to retry before the actual reset window and re-trip the limit. 600s covers all
-        # realistic provider reset windows while still rejecting pathological values. (#26293)
-        _retry_after = min(_retry_after, 600)
-        if _retry_after <= 0:
-            # A zero/expired cooldown (retry-after: 0, or an HTTP-date in the
-            # past, which the parser clamps to 0.0) carries no usable wait —
-            # treat it as absent so we never hot-loop the provider.
-            _retry_after = None
+        _retry_after = min(_retry_after, RETRY_AFTER_CAP_S)
     wait_time = _retry_after if _retry_after is not None else jittered_backoff(retry_count, base_delay=2.0, max_delay=60.0)
     _backoff_policy = None
     _adaptive = is_rate_limited or is_zai_coding_overload
@@ -1460,36 +1448,30 @@ def compute_error_backoff(
             retry_count, base_url=str(base_url), model=model, error=api_error, default_wait=wait_time,
         )
     _reset = reset_hint(api_error) if _adaptive else ""
-    _wait_reason = "Provider overloaded" if is_zai_coding_overload and not is_rate_limited else "Rate limited"
+    _free_busy = is_rate_limited and on_free_model(agent, base_url)
+    _wait_reason = ("The free model is busy" if _free_busy
+                    else "Provider overloaded" if is_zai_coding_overload and not is_rate_limited else "Rate limited")
     if _adaptive:
         _policy_note = _ZAI_POLICY_NOTES.get(_backoff_policy or "", "")
-        _rate_limit_status = (
-            f"⏱️ {_wait_reason}.{f' Resets in {_reset}.' if _reset else ''} Waiting {wait_time:.1f}s "
-            f"(attempt {retry_count + 1}/{max_retries}){_policy_note}..."
-        )
-        if _backoff_policy == "zai_coding_overload_long":
-            agent._emit_diagnostic_status(_rate_limit_status)
-        else:
-            agent._buffer_diagnostic_status(_rate_limit_status)
+        _status = (f"⏱️ {_wait_reason}.{f' Resets in {_reset}.' if _reset else ''} Waiting {wait_time:.1f}s "
+                   f"(attempt {retry_count + 1}/{max_retries}){_policy_note}...")
+        _announce_now = _backoff_policy == "zai_coding_overload_long"
     else:
-        _retry_status = (
-            f"⏳ Retrying in {wait_time:.1f}s (attempt {retry_count}/{max_retries})..."
-        )
-        if _retry_after is not None and _retry_after > 60:
-            # A 5xx Retry-After can now reach the 600s cap; buffering that wait
-            # would leave the user silent for minutes, so surface long provider
-            # cooldowns immediately (mirrors the zai_coding_overload_long path).
-            agent._emit_diagnostic_status(_retry_status)
-        else:
-            agent._buffer_diagnostic_status(_retry_status)
-    # The buffered line only replays if every retry fails; the live status
-    # line is the one thing the user sees meanwhile. Name the wait there so a
-    # 60s backoff after a 5xx is not an anonymous spinner — this is transient
-    # (rewritten by the next frame, cleared on recovery), so it does not add
-    # the transcript chatter the buffer exists to avoid. The reset window
-    # belongs here too: during the wait this line is the only place the user
-    # can learn whether to sit it out or switch models.
-    _live_reason = f"{_wait_reason.lower()} — resets in {_reset}," if _reset else "waiting on provider —"
+        _status = f"⏳ Retrying in {wait_time:.1f}s (attempt {retry_count}/{max_retries})..."
+        # A 5xx Retry-After can reach the cap; buffering it would leave the user silent for minutes.
+        _announce_now = _retry_after is not None and _retry_after > LIVE_RETRY_WAIT_CAP_S
+    if _announce_now:
+        agent._emit_diagnostic_status(_status)
+    else:
+        agent._buffer_diagnostic_status(_status)
+    # The buffered line replays only if every retry fails; the live status line is transient
+    # (rewritten by the next frame, cleared on recovery), so it names the wait on every retry
+    # without adding transcript chatter. The reset window belongs here too: during the wait this
+    # line is where the user learns whether to sit it out or switch models.
+    if _reset:
+        _live_reason = f"{_wait_reason.lower()} — resets in {_reset},"
+    else:
+        _live_reason = "the free model is busy —" if _free_busy else "waiting on provider —"
     agent._emit_diagnostic_wait(
         f"⏳ {_live_reason} retrying in {wait_time:.0f}s (attempt {retry_count}/{max_retries})"
     )
@@ -1501,6 +1483,22 @@ def compute_error_backoff(
     return wait_time
 
 
+def free_tier_cooldown_ends_turn(agent: Any, api_error: Exception, base_url: Any) -> bool:
+    """True when an attended session on the free model meets a cooldown longer than a person
+    sits through. The free tier's pause is service-wide: a 600 s retry met a second 429 while its
+    1069 s window ran, so the caller ends the attempt cycle now (fallback, or the reset time and
+    the sign-in / model doors). Every other provider, and every unattended run (cron, messaging,
+    ``hermes chat -q``), waits out its ``Retry-After``: a paid bucket reopens on time, and
+    ``fallback.min_switch_reset_seconds`` relies on that wait."""
+    from agent.retry_utils import LIVE_RETRY_WAIT_CAP_S, provider_retry_after_seconds
+    from agent.session_source import is_attended
+    from hermes_cli.anon_auth import on_free_model
+    if not (on_free_model(agent, base_url) and is_attended(agent)):
+        return False
+    cooldown = provider_retry_after_seconds(api_error)
+    return cooldown is not None and cooldown > LIVE_RETRY_WAIT_CAP_S
+
+
 def _codex_soft_failure_error(response: Any) -> Dict[str, Any]:
     """``response.error`` of a Codex ``failed``/``cancelled`` Response as ``{"code", "message"}``
     (the SDK types it as ``ResponseError``; the raw-SSE assembler keeps the dict); ``{}`` when absent."""

agent/turn_recovery_autorecover.py

@@ -46,23 +46,11 @@ def auto_recovery_cycles(agent: Any) -> int:
     return max(int(getattr(agent, "_auto_recovery_cycles", 0) or 0), 0)
 
 
-def _retry_after_seconds(api_error: Any) -> Optional[float]:
-    """Provider-declared cooldown from the ``Retry-After`` header or a ``retry_after`` body field."""
-    from agent.retry_utils import parse_retry_after_seconds
-    value = parse_retry_after_seconds(getattr(getattr(api_error, "response", None), "headers", None))
-    if value is None:
-        body = getattr(api_error, "body", None)
-        if isinstance(body, dict):
-            nested = body.get("error")
-            value = parse_retry_after_seconds((nested if isinstance(nested, dict) else body).get("retry_after"))
-    return value if value is not None and value > 0 else None
-
-
 def ladder_wait_seconds(cycle: int, api_error: Any) -> float:
     """Wait before recovery ``cycle`` (1-based): jittered 15/30/60/60/60 s, or the provider's
     ``Retry-After`` when present (honoured past the 60 s cap, up to 120 s)."""
-    from agent.retry_utils import jittered_backoff
-    retry_after = _retry_after_seconds(api_error)
+    from agent.retry_utils import jittered_backoff, provider_retry_after_seconds
+    retry_after = provider_retry_after_seconds(api_error)
     if retry_after is not None:
         return min(retry_after, _RETRY_AFTER_CAP_S)
     return jittered_backoff(cycle, base_delay=_LADDER_BASE_DELAY_S, max_delay=_LADDER_CAP_S, jitter_ratio=0.2)

agent/turn_usage.py

@@ -25,8 +25,8 @@ logger = logging.getLogger("agent.conversation_loop")
 def _agent_session_source(agent: Any) -> str:
     """The surface the agent's own row create would stamp (``_ensure_db_session``), so an
     accounting guard that wins the row-creation race never mints an anonymous session."""
-    from run_agent import _session_source_for_agent  # late: run_agent imports this module
-    return _session_source_for_agent(getattr(agent, "platform", None))
+    from agent.session_source import session_source_for
+    return session_source_for(getattr(agent, "platform", None))
 
 
 @dataclass

hermes_cli/anon_auth.py

@@ -262,6 +262,11 @@ def route_is_welcome_host(base_url: Any) -> bool:
     return host in welcome_hosts()
 
 
+def on_free_model(agent: Any, base_url: Any) -> bool:
+    """The request went to the free tier's host on a free-tier credential."""
+    return route_is_welcome_host(base_url) and is_anonymous_agent(agent)
+
+
 def anon_secret() -> str:
     return (os.environ.get(ANON_SECRET_ENV) or "").strip()
 
@@ -677,6 +682,9 @@ _WELCOME_ROUTE_REFUSALS = (
     ("anonymous accounts must use", "anon_on_paid_host"),
     ("serves anonymous hermes agent accounts only", "named_on_welcome_host"),
     ("anonymous accounts are not accepted", "tier_disabled"),
+    # The gateway's answer to an expired or unreadable bearer: the credential, not the tier. A
+    # retry that waited out a long rate limit outlives the 15-minute free-tier JWT and lands here.
+    ("invalid jwt", "session_expired"),
 )
 _WELCOME_ROUTE_COPY = {
     # Only reachable when the route heal (``turn_recovery._recover_welcome_tier``) could not move
@@ -686,6 +694,9 @@ _WELCOME_ROUTE_COPY = {
     "named_on_welcome_host": "This Nous account needs to reconnect. {model_hint}",
     "tier_disabled": "Using Hermes without signing in is switched off right now. "
                      "Sign in to keep chatting, it's free. {signin}",
+    # Only reachable when re-exchanging the free credential failed (``turn_recovery._recover_welcome_tier``).
+    "session_expired": "Hermes couldn't renew its connection to the free model. "
+                       "Send your message again, or sign in to keep chatting, it's free. {signin}",
 }
 # The sign-in door, phrased for a chat surface (slash command) and for a terminal.
 _SIGNIN_CHAT = "To sign in: /login."
@@ -756,6 +767,7 @@ def welcome_route_refusal(status: Any, message: Any, base_url: Any = None) -> Op
     ``"anon_on_paid_host"``: a free-tier JWT reached the paid host. ``"named_on_welcome_host"``: an
     account or API key reached the free tier's host. ``"tier_disabled"``: the tier is dark
     (``WELCOME_MODE=off``). Each is deterministic for the request: retrying cannot help.
+    ``"session_expired"``: the 403 names the bearer (``invalid jwt``); a fresh credential heals it.
 
     The dark-tier 403 is keyed on the ROUTE, not the message: the gateway's permission error
     carries only its generic sentence (the detail stays in its logs), so any 403 answered by the

hermes_cli/main.py

@@ -1338,7 +1338,7 @@ def _resolve_last_session(source: str = "cli") -> Optional[str]:
     """
     # A finite `hermes -z`/`chat -q` run is CLI history too: `hermes -z … --resume latest` chains on it.
     if source == "cli":
-        from run_agent import CLI_FAMILY_SOURCES
+        from agent.session_source import CLI_FAMILY_SOURCES
         source = sorted(CLI_FAMILY_SOURCES)
     with _session_db() as db:
         ws_key = _resolve_workspace_key()
@@ -1769,7 +1769,7 @@ def cmd_chat(args):
     if getattr(args, "source", None):
         os.environ["HERMES_SESSION_SOURCE"] = args.source
         # Explicit flag, not a label inherited from a parent TUI/Desktop session — one-shot
-        # runs must keep it (see run_agent._session_source_for_agent).
+        # runs must keep it (see agent.session_source.session_source_for).
         os.environ["HERMES_SESSION_SOURCE_EXPLICIT"] = "1"
 
     _pin_kanban_board_env()

run_agent.py

@@ -36,6 +36,7 @@ from typing import List, Dict, Any, Optional, Callable
 from datetime import datetime
 from pathlib import Path
 
+from agent.session_source import CLI_FAMILY_SOURCES, session_source_for
 from hermes_constants import get_hermes_home
 
 
@@ -53,34 +54,6 @@ def _launch_cwd_for_session(source: str) -> Optional[str]:
         return None
 
 
-# Sources that label the human conversation an interactive UI transport hosts. A finite ``hermes chat -q`` /
-# one-shot child spawned from such a session inherits HERMES_SESSION_SOURCE (the terminal tool bridges the
-# session env into child processes) but is NOT that conversation: labelling it ``tui``/``desktop`` lists it
-# in the TUI/WebUI pickers as a resumable chat and lets ``hermes -c`` in the TUI continue it (#112550).
-# Automation sources (kanban, tool, cron, a2a, ...) are inherited on purpose.
-_UI_TRANSPORT_SOURCES = frozenset({"tui", "desktop"})
-
-# Finite non-interactive CLI runs (``hermes chat -q``/``--oneshot``, ``hermes -z``) get their own source so human
-# pickers hide them without title/cwd heuristics; ``hermes -c`` still treats them as CLI history.
-ONESHOT_SOURCE = "oneshot"
-CLI_FAMILY_SOURCES = frozenset({"cli", ONESHOT_SOURCE})
-
-
-def _session_source_for_agent(platform: Optional[str]) -> str:
-    try:
-        from gateway.session_context import get_session_env
-    except Exception:
-        get_session_env = os.environ.get
-    source = str(get_session_env("HERMES_SESSION_SOURCE", "") or "").strip()
-    single_query = get_session_env("HERMES_SINGLE_QUERY_SESSION", "") == "1"
-    explicit = get_session_env("HERMES_SESSION_SOURCE_EXPLICIT", "") == "1"
-    if single_query and not explicit and source in _UI_TRANSPORT_SOURCES:
-        source = ""
-    if single_query and not source and (platform or "cli") == "cli":
-        return ONESHOT_SOURCE
-    return source or platform or "cli"
-
-
 def _gateway_origin_json(agent: "AIAgent") -> Optional[str]:
     """Gateway routing ``origin_json`` for a session row; None when the agent carries no gateway identity.
 
@@ -350,7 +323,7 @@ class AIAgent(
         """Create the session DB row on first use; a transient failure leaves it to retry next turn."""
         if getattr(self, "_persist_disabled", False) or self._session_db_created or not self._session_db:
             return
-        source = _session_source_for_agent(self.platform)
+        source = session_source_for(self.platform)
         try:
             # Persist the profile name explicitly, including "default": profile-keyed consumers treat NULL
             # as unowned.
@@ -401,7 +374,7 @@ class AIAgent(
         if should_start and target_session_id and hasattr(engine, "on_session_start"):
             start_context = {
                 "old_session_id": old_session_id, "carry_over_context": carry_over_context,
-                "platform": _session_source_for_agent(getattr(self, "platform", None)),
+                "platform": session_source_for(getattr(self, "platform", None)),
                 "model": getattr(self, "model", ""), "context_length": getattr(engine, "context_length", None),
                 "conversation_id": getattr(self, "_gateway_session_key", None), **extra_context,
             }

tests/agent/test_declared_conversation_scope.py

@@ -774,7 +774,7 @@ class TestSourceOverrideDomain:
     """The scope is memoized immediately, so the source must be right first.
 
     ``_agent_source`` used ``agent.platform`` before the row landed while
-    persistence uses ``_session_source_for_agent``, which honors
+    persistence uses ``session_source_for``, which honors
     ``HERMES_SESSION_SOURCE``. Under an override both sides of a ``/new``
     queried the platform domain, missed the boundary stored under the
     override, and hashed the same scope (@andrexibiza on #98811).

tests/agent/test_failed_turn_chat_copy.py

@@ -105,7 +105,7 @@ def test_max_retries_exhausted_chat_text_has_next_step_and_no_mechanism_lead():
     error = _Http(503, "HTTP 503: upstream unavailable")
     classified = classify_api_error(error, provider="openrouter", model="m")
     result = max_retries_exhausted_result(
-        _Agent(), error, classified, max_retries=3, is_rate_limited=False, error_msg=str(error).lower(),
+        _Agent(), error, classified, attempts=3, is_rate_limited=False, error_msg=str(error).lower(),
         api_kwargs=None, api_messages=[], messages=[], conversation_history=None, api_call_count=1,
         approx_tokens=10, provider="openrouter", base_url="https://openrouter.ai/api/v1", model="m",
     )
@@ -130,7 +130,7 @@ def test_exhausted_plan_quota_429_names_the_reset_window_not_wait_a_minute():
     agent = _Agent()
     agent._summarize_api_error = ApiErrorSummaryMixin._summarize_api_error
     result = max_retries_exhausted_result(
-        agent, error, classified, max_retries=3, is_rate_limited=True, error_msg=str(error).lower(),
+        agent, error, classified, attempts=3, is_rate_limited=True, error_msg=str(error).lower(),
         api_kwargs=None, api_messages=[], messages=[], conversation_history=None, api_call_count=3,
         approx_tokens=10, provider="openai-codex", base_url="https://chatgpt.com/backend-api/codex", model="gpt-5.3-codex",
     )
@@ -141,7 +141,7 @@ def test_exhausted_plan_quota_429_names_the_reset_window_not_wait_a_minute():
     # A throttle with no reset window keeps the short-wait copy.
     short = _Http(429, "HTTP 429: Rate limit exceeded")
     plain = max_retries_exhausted_result(
-        _Agent(), short, classify_api_error(short, provider="openrouter", model="m"), max_retries=3,
+        _Agent(), short, classify_api_error(short, provider="openrouter", model="m"), attempts=3,
         is_rate_limited=True, error_msg=str(short).lower(), api_kwargs=None, api_messages=[], messages=[],
         conversation_history=None, api_call_count=3, approx_tokens=10, provider="openrouter",
         base_url="https://openrouter.ai/api/v1", model="m",

tests/agent/test_session_source.py

@@ -1,7 +1,7 @@
 import pytest
 
 from gateway.session_context import _UNSET, _VAR_MAP, clear_session_vars, set_session_vars
-from run_agent import _session_source_for_agent
+from agent.session_source import session_source_for
 
 
 @pytest.fixture(autouse=True)
@@ -18,7 +18,7 @@ def test_session_source_context_overrides_platform(monkeypatch):
 
     tokens = set_session_vars(source="tool")
     try:
-        assert _session_source_for_agent("tui") == "tool"
+        assert session_source_for("tui") == "tool"
     finally:
         clear_session_vars(tokens)
 
@@ -26,7 +26,7 @@ def test_session_source_context_overrides_platform(monkeypatch):
 def test_session_source_falls_back_to_platform(monkeypatch):
     monkeypatch.delenv("HERMES_SESSION_SOURCE", raising=False)
 
-    assert _session_source_for_agent("tui") == "tui"
+    assert session_source_for("tui") == "tui"
 
 
 
@@ -38,7 +38,7 @@ def test_oneshot_run_gets_distinct_source(monkeypatch, inherited):
     monkeypatch.setenv("HERMES_SESSION_SOURCE", inherited)
     monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
 
-    assert _session_source_for_agent("cli") == "oneshot"
+    assert session_source_for("cli") == "oneshot"
 
 
 def test_oneshot_marker_does_not_relabel_subagents(monkeypatch):
@@ -46,7 +46,7 @@ def test_oneshot_marker_does_not_relabel_subagents(monkeypatch):
     monkeypatch.delenv("HERMES_SESSION_SOURCE", raising=False)
     monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
 
-    assert _session_source_for_agent("subagent") == "subagent"
+    assert session_source_for("subagent") == "subagent"
 
 
 @pytest.mark.parametrize("inherited", ["kanban", "tool", "a2a"])
@@ -54,7 +54,7 @@ def test_oneshot_child_keeps_inherited_automation_source(monkeypatch, inherited)
     monkeypatch.setenv("HERMES_SESSION_SOURCE", inherited)
     monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
 
-    assert _session_source_for_agent("cli") == inherited
+    assert session_source_for("cli") == inherited
 
 
 @pytest.mark.parametrize("explicit", ["tui", "desktop"])
@@ -65,4 +65,4 @@ def test_oneshot_keeps_explicit_source_flag(monkeypatch, explicit):
     monkeypatch.setenv("HERMES_SESSION_SOURCE_EXPLICIT", "1")
     monkeypatch.setenv("HERMES_SINGLE_QUERY_SESSION", "1")
 
-    assert _session_source_for_agent("cli") == explicit
+    assert session_source_for("cli") == explicit

tests/agent/test_welcome_error_identity.py

@@ -45,7 +45,7 @@ def test_named_errors_do_not_offer_anonymous_recovery(api_key, base_url, case):
     if status == 403:
         result = nonretryable_client_error_result(agent, error, classified, status_code=status, **common)
     else:
-        result = max_retries_exhausted_result(agent, error, classified, max_retries=3,
+        result = max_retries_exhausted_result(agent, error, classified, attempts=3,
                                              is_rate_limited=status == 429, error_msg=message, **common)
     assert "welcome_refusal" not in classified.error_context
     assert "welcome_route" not in classified.error_context

tests/agent/test_welcome_tier_recovery.py

@@ -177,22 +177,26 @@ class TestLongWaitRule:
         assert recorded == []
 
 
+_ANON = SimpleNamespace(provider="nous", api_key=make_jwt())
+
+
 class TestOutageCopy:
     @pytest.mark.parametrize("reason", [FailoverReason.timeout, FailoverReason.overloaded,
                                         FailoverReason.server_error])
     def test_a_spent_transport_failure_on_the_welcome_host_reads_as_one_sentence(self, reason):
         from agent.turn_recovery import _welcome_outage_copy
         from hermes_cli.anon_auth import FREE_TIER_OUTAGE_COPY
-        assert _welcome_outage_copy(WELCOME, SimpleNamespace(reason=reason), anonymous=True) == FREE_TIER_OUTAGE_COPY
+        assert _welcome_outage_copy(_ANON, WELCOME, SimpleNamespace(reason=reason)) == FREE_TIER_OUTAGE_COPY
 
     def test_other_routes_and_other_reasons_keep_the_technical_summary(self):
         from agent.turn_recovery import _welcome_outage_copy
-        assert _welcome_outage_copy(PAID, SimpleNamespace(reason=FailoverReason.timeout), anonymous=True) == ""
-        assert _welcome_outage_copy(WELCOME, SimpleNamespace(reason=FailoverReason.rate_limit), anonymous=True) == ""
+        assert _welcome_outage_copy(_ANON, PAID, SimpleNamespace(reason=FailoverReason.timeout)) == ""
+        assert _welcome_outage_copy(_ANON, WELCOME, SimpleNamespace(reason=FailoverReason.rate_limit)) == ""
         # ``unknown`` is the catch-all for status-less local failures, not the free model's trouble.
-        assert _welcome_outage_copy(WELCOME, SimpleNamespace(reason=FailoverReason.unknown), anonymous=True) == ""
+        assert _welcome_outage_copy(_ANON, WELCOME, SimpleNamespace(reason=FailoverReason.unknown)) == ""
         # A named account's outage is its provider's trouble, not the free model's.
-        assert _welcome_outage_copy(WELCOME, SimpleNamespace(reason=FailoverReason.timeout)) == ""
+        named = SimpleNamespace(provider="nous", api_key=make_jwt(account_tier="free"))
+        assert _welcome_outage_copy(named, WELCOME, SimpleNamespace(reason=FailoverReason.timeout)) == ""
 
 
 class TestTerminalResultsCarryTheFreeTierBlock:
@@ -229,7 +233,7 @@ class TestTerminalResultsCarryTheFreeTierBlock:
         err = _refusal("at_capacity", retry_after=30)
         classified = _classify(err)
         result = max_retries_exhausted_result(
-            self._terminal_agent(), err, classified, max_retries=3, is_rate_limited=True, error_msg="429",
+            self._terminal_agent(), err, classified, attempts=3, is_rate_limited=True, error_msg="429",
             api_kwargs=None, api_messages=[], messages=[], conversation_history=[], api_call_count=3,
             approx_tokens=10, provider="nous", base_url=WELCOME, model="nous/welcome")
         assert result["free_tier"]["kind"] == "at_capacity"
@@ -241,7 +245,7 @@ class TestTerminalResultsCarryTheFreeTierBlock:
         err = _gateway_error(503, {"status": 503, "message": "The requested model is currently unavailable."})
         classified = _classify(err)
         result = max_retries_exhausted_result(
-            self._terminal_agent(), err, classified, max_retries=3, is_rate_limited=False, error_msg="503",
+            self._terminal_agent(), err, classified, attempts=3, is_rate_limited=False, error_msg="503",
             api_kwargs=None, api_messages=[], messages=[], conversation_history=[], api_call_count=3,
             approx_tokens=10, provider="nous", base_url=WELCOME, model="nous/welcome")
         assert result["free_tier"]["kind"] == "outage"
@@ -250,7 +254,74 @@ class TestTerminalResultsCarryTheFreeTierBlock:
         from agent.turn_recovery import max_retries_exhausted_result
         err = _gateway_error(503, {"status": 503, "message": "The requested model is currently unavailable."})
         result = max_retries_exhausted_result(
-            self._terminal_agent(), err, _classify(err, base_url=PAID), max_retries=3, is_rate_limited=False,
+            self._terminal_agent(), err, _classify(err, base_url=PAID), attempts=3, is_rate_limited=False,
             error_msg="503", api_kwargs=None, api_messages=[], messages=[], conversation_history=[],
             api_call_count=3, approx_tokens=10, provider="nous", base_url=PAID, model="hermes-4")
         assert "free_tier" not in result
+
+
+def _plain_429(retry_after: int) -> MockAPIError:
+    """A 429 with no structured welcome refusal: only the provider's ``retry_after``."""
+    return _gateway_error(429, {"error": {"message": "Rate limit exceeded", "retry_after": retry_after}})
+
+
+def _backoff_agent(**overrides):
+    agent = _agent(_client_log_context=lambda: "", **overrides)
+    agent.emitted, agent.buffered, agent.waits = [], [], []
+    agent._emit_diagnostic_status = agent.emitted.append
+    agent._buffer_diagnostic_status = agent.buffered.append
+    agent._emit_diagnostic_wait = agent.waits.append
+    return agent
+
+
+class TestFreeTierCooldownCutoff:
+    """Q9: an attended session on the free model ends the turn at once with the reset time; a
+    delegated child, a library caller and an unattended run keep waiting."""
+
+    @pytest.fixture(autouse=True)
+    def _no_inherited_source(self, monkeypatch):
+        for name in ("HERMES_SESSION_SOURCE", "HERMES_SINGLE_QUERY_SESSION", "HERMES_SESSION_SOURCE_EXPLICIT"):
+            monkeypatch.delenv(name, raising=False)
+
+    def test_an_attended_desktop_session_ends_the_turn_and_names_the_reset(self):
+        from agent.turn_recovery import free_tier_cooldown_ends_turn, max_retries_exhausted_result
+        err = _plain_429(90)
+        assert free_tier_cooldown_ends_turn(_agent(platform="desktop"), err, WELCOME) is True
+        result = max_retries_exhausted_result(
+            TestTerminalResultsCarryTheFreeTierBlock._terminal_agent(), err, _classify(err), attempts=1,
+            is_rate_limited=True, error_msg="429", api_kwargs=None, api_messages=[], messages=[],
+            conversation_history=[], api_call_count=1, approx_tokens=10, provider="nous", base_url=WELCOME,
+            model="nous/welcome")
+        assert "resets in ~2 min" in result["final_response"]
+
+    def test_a_subagent_of_a_desktop_session_keeps_waiting(self):
+        """The child runs in a copy of the parent's context, so it reads the parent's source."""
+        from agent.turn_recovery import free_tier_cooldown_ends_turn
+        from gateway.session_context import clear_session_vars, set_session_vars
+        tokens = set_session_vars(source="desktop")
+        try:
+            assert free_tier_cooldown_ends_turn(_agent(platform="subagent"), _plain_429(90), WELCOME) is False
+        finally:
+            clear_session_vars(tokens)
+
+    @pytest.mark.parametrize("platform", ["cron", None])
+    def test_an_unattended_run_keeps_waiting_and_is_told_the_reset(self, platform):
+        from agent.turn_recovery import compute_error_backoff, free_tier_cooldown_ends_turn
+        err = _plain_429(90)
+        agent = _backoff_agent(platform=platform)
+        assert free_tier_cooldown_ends_turn(agent, err, WELCOME) is False
+        compute_error_backoff(agent, err, retry_count=1, max_retries=3, is_rate_limited=True,
+                              is_zai_coding_overload=False, base_url=WELCOME, model="nous/welcome")
+        assert "Resets in ~" in agent.buffered[0]
+
+    def test_a_paid_long_wait_stays_buffered_with_mains_attempt_numbers(self):
+        """A paid 300 s Retry-After: the status line is buffered (replayed only if every retry
+        fails) and the live wait line names the attempt that just failed, as on main."""
+        from agent.turn_recovery import compute_error_backoff
+        agent = _backoff_agent(provider="openrouter", api_key="sk-or-test", platform="desktop")
+        err = _plain_429(300)
+        wait = compute_error_backoff(agent, err, retry_count=1, max_retries=3, is_rate_limited=True,
+                                     is_zai_coding_overload=False, base_url="https://openrouter.ai/api/v1", model="m")
+        assert wait == 300
+        assert agent.emitted == [] and len(agent.buffered) == 1
+        assert "(attempt 1/3)" in agent.waits[0]

tests/gateway/test_compress_command.py

@@ -229,7 +229,7 @@ async def test_compress_command_in_place_skips_destructive_rewrite():
 async def test_compress_command_preserves_platform_and_gateway_session_key():
     """The temporary compression agent must carry the originating source's
     platform and stable gateway session key, matching a normal gateway turn.
-    Without them ``_session_source_for_agent`` falls back to a default "cli"
+    Without them ``session_source_for`` falls back to a default "cli"
     host source, so an external context engine misattributes the retained
     transcript tail and later duplicates it on resume (#50422)."""
     history = _make_history()

tests/tui_gateway/test_session_create_model_provider_guard.py

@@ -63,7 +63,7 @@ def test_session_create_logs_when_a_client_override_beats_the_profile_default(_c
     """A composer pick silently decided every new chat's model; agent.log must name it (#107410)."""
     from tui_gateway import server
 
-    monkeypatch.setattr(server, "_session_default_model", lambda session: "deepseek-v4-flash")
+    monkeypatch.setattr(server, "_session_default_route", lambda session: ("deepseek-v4-flash", ""))
     with caplog.at_level(logging.INFO):
         response, _ = _create({"model": "gpt-5.5", "provider": "openrouter"})
         pinned = [r.getMessage() for r in caplog.records if "client override" in r.getMessage()]

tui_gateway/methods_session.py

@@ -294,7 +294,7 @@ def _seed_branch_row(record: dict, key: str, parent_session_id: str, history: li
             _persist_branch(db, key, parent_session_id, _branch_title(db, parent_session_id), history,
                             source=source, cwd=None if _is_remote_launch_cwd(record) else record["cwd"],
                             profile_name=profile_name_for_home(profile_home) or _current_profile_name(),
-                            model=_session_default_model(record), compensate=True, title_source="derived", user_id=_session_auth_user_id(record))
+                            model=_session_default_route(record)[0], compensate=True, title_source="derived", user_id=_session_auth_user_id(record))
             record["pending_title"] = None
             # The first submit's _persist_branch_seed is the fallback for a failed seed, not a second copy.
             record["_branch_seed_persisted"] = True
@@ -370,9 +370,7 @@ def _create_session(rid, params: dict, *, copy_parent_history: bool = False) ->
                         "message_count": len(history),
                         **({"messages_omitted": True} if copy_parent_history
                            else {"messages": _history_to_messages(history, profile_home=session.get("profile_home"))}),
-                        "info": {"model": override.get("model") if override else _session_default_model(session),
-                                 **({"provider": override["provider"]} if override.get("provider") else {}),
-                                 "tools": {}, "skills": {}, "cwd": session["cwd"], "branch": git_probe.branch(session["cwd"]),
+                        "info": {**_lazy_info_route(session, override), "tools": {}, "skills": {}, "cwd": session["cwd"], "branch": git_probe.branch(session["cwd"]),
                                  "project": _project_info_for_cwd(session["cwd"]), "lazy": True,
                                  "desktop_contract": DESKTOP_BACKEND_CONTRACT,
                                  "profile_name": _response_profile_name(profile)}})
@@ -450,7 +448,7 @@ def _create_session(rid, params: dict, *, copy_parent_history: bool = False) ->
         # name both so agent.log alone explains which model a new chat runs, and why (#107410).
         logger.info("session.create %s: model=%s provider=%s source=client override (profile default: %s)",
                     key, session_model_override["model"], session_model_override.get("provider") or "-",
-                    _session_default_model(_sessions[sid]))
+                    _session_default_route(_sessions[sid])[0])
     # No DB row here (drafts left "Untitled" litter): created on the first prompt — except seeded sessions.
     # NOTE: we intentionally do NOT persist a DB row here. Every TUI/desktop launch (and every "New agent" /
     # draft) opens a session here just to paint the composer, so eagerly creating a row left an "Untitled"
@@ -485,10 +483,7 @@ def _create_session(rid, params: dict, *, copy_parent_history: bool = False) ->
     return _ok(rid, {
         "session_id": sid, "stored_session_id": key, "message_count": len(messages),
         **({"messages_omitted": True} if copy_parent_history else {"messages": messages}),
-        # Reflect the override now so the client doesn't clobber its sticky pick.
-        "info": {"model": override.get("model") if override else _session_default_model(_sessions[sid]),
-                 **({"provider": override["provider"]} if override.get("provider") else {}),
-                 "tools": {}, "skills": {}, "cwd": cwd, "branch": git_probe.branch(cwd),
+        "info": {**_lazy_info_route(_sessions[sid], override), "tools": {}, "skills": {}, "cwd": cwd, "branch": git_probe.branch(cwd),
                  "project": _project_info_for_cwd(cwd), "lazy": True, "desktop_contract": DESKTOP_BACKEND_CONTRACT,
                  "profile_name": _response_profile_name(profile)}})
 
@@ -2306,7 +2301,7 @@ def _branch_live(rid, params: dict, session: dict, *, omit_messages: bool = Fals
             _persist_branch(db, new_key, old_key, title, history, source=source,
                             cwd=None if _is_remote_launch_cwd(session) else _session_cwd(session),
                             profile_name=profile_name_for_home(home) or _current_profile_name(),
-                            model=_session_default_model(session), copy_fields=_BRANCH_COPY_FIELDS,
+                            model=_session_default_route(session)[0], copy_fields=_BRANCH_COPY_FIELDS,
                             title_source="user" if params.get("name") else "derived",
                             user_id=_session_auth_user_id(session))
         except Exception as e:

tui_gateway/model_switch.py

@@ -141,13 +141,26 @@ def _session_profile_runtime_scope(session: dict, *, hydrate_secrets: bool = Tru
         _release_profile_runtime_scope_tokens(scopes)
 
 
-def _session_default_model(session: dict) -> str:
-    """The configured default model of the session's OWN profile. Bare ``_resolve_model()`` reads the
-    LAUNCH profile's config, so a secondary session's reply or first state.db row carried the launch
-    profile's model id."""
+def _session_default_route(session: dict) -> tuple[str, str]:
+    """``(model, provider)`` a not-yet-built session of its OWN profile will run on. Bare
+    ``_resolve_startup_runtime()`` reads the LAUNCH profile's config, so a secondary session's reply or
+    first state.db row carried the launch profile's model id. On the Nous free tier the agent build pins
+    ``nous/welcome`` (``pin_model_for_route``), so the configured default (often the silent default,
+    with no provider) is not what the session runs; report the pinned route instead."""
+    from hermes_cli.anon_auth import GUEST_MODEL, free_tier_route
     with _session_profile_runtime_scope({"profile_home": session.get("profile_home") or None},
                                         hydrate_secrets=False):
-        return _resolve_model()
+        if not _resolve_startup_runtime()[1] and free_tier_route():
+            return GUEST_MODEL, "nous"
+        # Off the free tier, the model id alone: the provider is resolved when the agent is built.
+        return _resolve_model(), ""
+
+
+def _lazy_info_route(session: dict, override: dict) -> dict:
+    """``session.info``'s model and provider for a not-yet-built session: the client's sticky pick when it
+    sent one (so the client does not clobber it), else the profile's default route."""
+    model, provider = (override.get("model"), override.get("provider")) if override else _session_default_route(session)
+    return {"model": model, **({"provider": provider} if provider else {})}
 
 
 def _restart_completed_failed_agent_build(sid: str, session: dict, failed_ready: threading.Event | None) -> bool:

tui_gateway/server.py

@@ -2319,10 +2319,11 @@ def _live_session_identity(session: dict) -> tuple[str, str]:
     agent = session.get("agent")
     override = session.get("model_override") or {}
     model = (str(pending.get("display_model") or "").strip() or mirror.get("model")
-             or getattr(agent, "model", "") or override.get("model") or _session_default_model(session))
+             or getattr(agent, "model", "") or override.get("model"))
     provider = (str(pending.get("display_provider") or "").strip() or mirror.get("provider")
-                or getattr(agent, "provider", "") or override.get("provider") or "")
-    return str(model), str(provider or "")
+                or getattr(agent, "provider", "") or override.get("provider"))
+    default = ("", "") if model else _session_default_route(session)
+    return str(model or default[0]), str(provider or default[1])
 
 
 def _fast_tier_applies(agent, model: str, provider: str, *, route_known: bool, tier: str | None = None) -> bool:
@@ -2829,10 +2830,9 @@ def _lazy_resume_info(cwd: str, *, model: str = "", provider: str = "", profile:
     """session.info for a not-yet-built session (session.create's shape); tools/skills land with the deferred build."""
     return {
         "cwd": cwd, "branch": git_probe.branch(cwd), "project": _project_info_for_cwd(cwd),
-        "model": model or _session_default_model({"profile_home": _profile_home(profile)}),
+        **_lazy_info_route({"profile_home": _profile_home(profile)}, {"model": model, "provider": provider} if model else {}),
         "tools": {}, "skills": {}, "lazy": True,
         "desktop_contract": DESKTOP_BACKEND_CONTRACT, "profile_name": _response_profile_name(profile),
-        **({"provider": provider} if provider else {}),
     }
 
 
@@ -3099,7 +3099,7 @@ def _fallback_session_info(session: dict) -> dict:
     cwd = _session_cwd(session)
     return {
         "cwd": cwd, "branch": git_probe.branch(cwd), "project": _project_info_for_cwd(cwd), "lazy": True,
-        "model": _session_default_model(session), "skills": {}, "tools": {}, "desktop_contract": DESKTOP_BACKEND_CONTRACT,
+        **_lazy_info_route(session, {}), "skills": {}, "tools": {}, "desktop_contract": DESKTOP_BACKEND_CONTRACT,
     }
 
 

tui_gateway/session_workdir.py

@@ -416,7 +416,7 @@ def _workdir_row_model_config(session: dict) -> tuple[str, dict]:
     global default here wins the INSERT-OR-IGNORE race (a reconnect silently reverts to the profile default).
     model_config carries provider/reasoning/service_tier so resume restores effort + fast too."""
     override = raw if isinstance(raw := session.get("model_override"), dict) else {}
-    row_model = str(override.get("model") or "").strip() or _session_default_model(session)
+    row_model = str(override.get("model") or "").strip() or _session_default_route(session)[0]
     model_config: dict = {k: str(v) for k in ("model", "provider", "base_url", "api_mode") if (v := override.get(k))}
     # A RESOLVED provider "custom" (named ``providers:``/``custom_providers:`` entry) persisted bare here is the origin
     # of "No LLM provider configured" rows (resume routes to OpenRouter with no key). Recover the durable

website/docs/user-guide/configuration.md

@@ -1237,6 +1237,8 @@ agent:
 
 `agent.api_max_retries` controls how many times Hermes retries a provider API call on transient errors (rate limits, connection drops, 5xx) **before** fallback-provider switching engages. The default is `3` — four attempts total. If you have [fallback providers](./features/fallback-providers.md) configured and want to fail over faster, drop this to `0` so the first transient error on your primary immediately hands off to the fallback instead of churning retries against the flaky endpoint.
 
+A rate limit that names its reset (`Retry-After` or a `retry_after` body field) is waited out, capped at 600 s per retry. The live status line shows the wait and, when the provider names it, when the limit resets. One exception: on the free Nous tier (no sign-in), a session someone is watching (the Desktop app, the TUI, the `hermes` terminal chat, an ACP editor) does not sit through a cooldown longer than 60 s, because the free tier's pause covers the whole service and a retry before the reset only meets the same refusal. The turn ends at once and says when the limit lifts; a configured fallback runs first. Unattended runs (`hermes chat -q`, cron, messaging platforms, delegated subagents) keep waiting.
+
 `agent.auto_recovery_cycles` is the safety net *after* both the retries and the fallback chain are spent. When the failure is a transient outage (HTTP 5xx, an `overloaded`/529 response, a connect or read timeout) and no answer text has reached you yet, Hermes does not end the turn with "API failed after N retries" — it waits and tries again, up to this many cycles (default `5`), with a jittered 15/30/60/60/60 s schedule. A provider `Retry-After` header wins over the schedule (honoured up to 120 s). Every surface shows the same line while it waits — `⏳ Provider temporarily unavailable — retrying automatically in 30s (cycle 2/5); press Esc to stop` on the CLI/TUI/Desktop, a status bubble on messaging platforms (`send /stop to cancel`), a `hermes.status` SSE event on the API server, and a log line for cron jobs. Pressing Esc (or `/stop`) cancels the wait immediately. Fallback still comes first: with a fallback chain configured, exhaustion moves to the next provider as before, and the ladder only engages once the chain has nothing left. Authentication, billing, request-format, entitlement, content-policy and account-policy errors never enter the ladder. Set `0` to disable it.
 
 ## Wall-Clock Run Budget