Merge master into stage-5155 (catch up to v0.51.728)

CHANGELOG.md

@@ -24,6 +24,9 @@
 
 ### Fixed
 
+- **Provider auth failures (e.g. an invalidated 401 token) now surface as a visible error turn in chat instead of a silent empty assistant turn.** Some terminal provider failures returned a result that escaped the existing `apperror` settlement, so the WebUI finalized the turn "done" with an empty body — no `stream error`, no inline error, and a saved transcript that pretended the assistant never responded (so a reload disagreed with what you just saw). The settlement path now evaluates terminal failure against the *merged, save-ready* transcript rather than the raw agent result, and persists the same `_error` turn + `apperror` payload the exception path already uses (even after partially-streamed output). User cancel/interrupt is still never mislabeled as a provider failure, and a legitimate repeated assistant answer still completes normally. Thanks @rodboev. (#5129, fixes #5121)
+- **Background-process wakeups on a slash-qualified custom provider no longer fail with "Invalid model format."** When a process wakeup took the fast model-resolution path, a bare runtime model id (`grok-composer-2.5-fast`) could reach an OpenAI-compatible proxy that requires the slash-qualified form (`x-ai/grok-composer-2.5-fast`), so the upstream rejected the turn with HTTP 400 exactly when a `terminal(notify_on_complete=true)` wakeup needed to consume completion output. The resolver now repairs a suffix-only custom-provider model to the profile's qualified default — gated on an exact suffix match **and** the profile's configured provider matching the session's requested provider, so it can never re-point a different custom provider's bare model. Thanks @claw-io (co-author @b3nw). (#5128, fixes #5127)
+- **Gateway-mode approval prompts now distinguish "gateway offline" from "gateway too old."** When the WebUI ran in gateway mode and the gateway process was down, an approval-requiring turn showed a generic "Approvals not supported" toast — which reads like a feature limitation, not a connectivity failure, so a user had no reason to check whether the gateway was running. The capability probe now tracks whether `/v1/capabilities` was reachable: a genuine connection failure surfaces "Gateway connection failed — check that the connected Hermes gateway is running and reachable," while a reachable-but-older gateway (HTTP error or a slow/timed-out probe) keeps the existing "Approvals require a newer gateway" message. Thanks @rodboev. (#5151, fixes #5139)
 - **Switching sessions feels snappier — non-critical refreshes now happen after the first paint.** The session-switch path deferred three non-essential operations (the workspace-tree/git-badge refresh, the model-dropdown option-list refresh, and a redundant empty-composer-draft save) until after the new session renders, instead of blocking the switch on them. Each deferred op re-checks the active session id at fire time, so a refresh scheduled for a session you've already switched away from is dropped rather than applied to the wrong session; non-empty composer drafts are never skipped. (Internal: added `requestIdleCallback`/`cancelIdleCallback` to the static-JS scope-undef gate's browser-globals allowlist.) Thanks @santastabber. (#5117)
 - **Stale approval cards now clear on stream teardown in legacy gateway mode.** When the WebUI ran against a legacy gateway backend, an approval card could linger after the underlying approval was already resolved/gone, until the next fallback poll reconciled it. Stream teardown now reconciles the gateway approval mirrors against the live queue (under the approval lock), so a stale card clears as soon as teardown proves the approval is gone — still-pending gateway and local approvals are preserved. Thanks @rodboev. (#5147, fixes #5135)
 - **A pinned reader no longer gets bounced off the live tail when a streaming turn rebuilds its rows.** While a response streams, live activity/worklog DOM updates rebuild rows; for a reader pinned to the live tail, restoring the first-visible viewport anchor could remount an older row and jump them away from the bottom. Pinned/following readers now keep a tail-relative scroll position across the rebuild (restored before the semantic anchor pass), while explicitly-unpinned/manual reader positions still use the semantic viewport-anchor restore. Thanks @santastabber. (#5118)

api/config.py

@@ -17,6 +17,7 @@ import logging
 import os
 import queue
 import re
+import socket
 import sys
 import threading
 import time
@@ -7433,6 +7434,15 @@ _GATEWAY_CAPS_LOCK = threading.Lock()
 _GATEWAY_CAPS_TTL_S: float = 60.0
 
 
+def _gateway_caps_probe_timed_out(exc: BaseException) -> bool:
+    """Keep slow capability probes on the legacy reachable-but-unsupported path."""
+    reason = exc.reason if isinstance(exc, urllib.error.URLError) else exc
+    if isinstance(reason, (TimeoutError, socket.timeout)):
+        return True
+    reason_text = str(reason).lower()
+    return "timed out" in reason_text or "timeout" in reason_text
+
+
 def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
     """Return cached gateway capability flags, probing /v1/capabilities if stale."""
     base_url = str(base_url or "").rstrip("/")
@@ -7443,21 +7453,40 @@ def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
         cached = _GATEWAY_CAPS_CACHE.get(cache_key)
         if cached and now - cached.get("fetched_at", 0) < _GATEWAY_CAPS_TTL_S:
             return cached
-    caps = {"approval_events": False, "run_approval_response": False, "fetched_at": 0.0}
+    caps = {
+        "approval_events": False,
+        "run_approval_response": False,
+        "capabilities_reachable": False,
+        "probe_error": None,
+        "fetched_at": 0.0,
+    }
     try:
         headers = {}
         if api_key:
             headers["Authorization"] = f"Bearer {api_key}"
         req = urllib.request.Request(f"{base_url}/v1/capabilities", headers=headers, method="GET")
         with urllib.request.urlopen(req, timeout=3) as resp:
+            caps["capabilities_reachable"] = True
             body = json.loads(resp.read(65536))
         features = body.get("features") if isinstance(body, dict) else {}
         if not isinstance(features, dict):
             features = {}
         caps["approval_events"] = bool(features.get("approval_events"))
         caps["run_approval_response"] = bool(features.get("run_approval_response"))
-    except Exception:
-        pass
+    except urllib.error.HTTPError as exc:
+        caps["capabilities_reachable"] = True
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
+    except urllib.error.URLError as exc:
+        if _gateway_caps_probe_timed_out(exc):
+            caps["capabilities_reachable"] = True
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
+    except (TimeoutError, socket.timeout) as exc:
+        caps["capabilities_reachable"] = True
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
+    except OSError as exc:
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
+    except Exception as exc:
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
     with _GATEWAY_CAPS_LOCK:
         current = _GATEWAY_CAPS_CACHE.get(cache_key)
         if current and current.get("fetched_at", 0) > probe_started_at:
@@ -7467,6 +7496,16 @@ def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
     return caps
 
 
+def gateway_approval_unavailable_reason(base_url: str, api_key: str = "") -> str | None:
+    """Return why approval support is unavailable, if it is unavailable."""
+    caps = get_gateway_caps(base_url, api_key)
+    if bool(caps.get("approval_events") and caps.get("run_approval_response")):
+        return None
+    if not caps.get("capabilities_reachable"):
+        return "unreachable"
+    return "unsupported"
+
+
 def gateway_supports_approval(base_url: str, api_key: str = "") -> bool:
     """True only when the gateway advertises both approval_events and run_approval_response."""
     caps = get_gateway_caps(base_url, api_key)

api/gateway_chat.py

@@ -20,6 +20,7 @@ from api.config import (
     STREAM_REASONING_TEXT,
     _get_session_agent_lock,
     coerce_reasoning_effort_for_model,
+    gateway_approval_unavailable_reason,
     gateway_supports_approval,
     register_active_run,
     unregister_active_run,
@@ -653,13 +654,19 @@ def _run_gateway_chat_streaming(
         else:
             # Legacy gateway path: emit unsupported approval notice once per session,
             # but only when the gateway genuinely lacks approval capability.
-            if not gateway_supports_approval(base_url, api_key):
+            approval_reason = gateway_approval_unavailable_reason(base_url, api_key)
+            if approval_reason is not None:
                 if not hasattr(s, "_approval_notice_emitted"):
                     s._approval_notice_emitted = False
                 if not s._approval_notice_emitted:
+                    approval_message = "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this."
+                    approval_type = "approval_gateway_unsupported"
+                    if approval_reason == "unreachable":
+                        approval_type = "approval_gateway_offline"
+                        approval_message = "Gateway connection failed. Check that the connected Hermes gateway is running and reachable."
                     put_gateway_event("warning", {
-                        "type": "approval_gateway_unsupported",
-                        "message": "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.",
+                        "type": approval_type,
+                        "message": approval_message,
                     })
                     s._approval_notice_emitted = True
 

api/routes.py

@@ -5215,31 +5215,52 @@ def _read_profile_model_config(
 ) -> tuple[str | None, str | None]:
     """Read model.provider and model.default from the session's profile config.
 
-    Returns (profile_provider, profile_default_model). Both are None when
-    the session has no profile, the profile config is unreadable, or an
-    explicit ``requested_provider`` is already set (profile should not
-    override explicit selections).
+    Returns (profile_provider, profile_default_model). Both are None when the
+    session has no profile or the profile config is unreadable.
+
+    When the session already has an explicit ``requested_provider``, the profile
+    ``model.provider`` is not returned (first tuple element is None) so profile
+    does not override the session provider. ``profile_default_model`` is still
+    returned for suffix repair (#5127) only when the profile's configured
+    provider matches ``requested_provider`` after normalization.
     """
-    if _clean_session_model_provider(requested_provider) or not getattr(session, "profile", None):
+    if not getattr(session, "profile", None):
         return None, None
+
     try:
         from api.profiles import get_hermes_home_for_profile
+
         _profile_home = get_hermes_home_for_profile(session.profile)
         _profile_cfg_path = os.path.join(str(_profile_home), "config.yaml")
         if not os.path.isfile(_profile_cfg_path):
             return None, None
         import yaml
+
         with open(_profile_cfg_path, encoding="utf-8") as _f:
             _pcfg = yaml.safe_load(_f) or {}
         if not isinstance(_pcfg, dict):
             return None, None
-        _provider = (_pcfg.get("model", {}).get("provider") or "").strip() or None
-        _default = (_pcfg.get("model", {}).get("default") or "").strip() or None
-        return _provider, _default
+        _model_cfg = _pcfg.get("model") or {}
+        if not isinstance(_model_cfg, dict):
+            return None, None
+        _provider = (_model_cfg.get("provider") or "").strip() or None
+        _default = (_model_cfg.get("default") or "").strip() or None
     except Exception:
-        logger.warning("profile provider read failed for %r", getattr(session, "profile", None), exc_info=True)
+        logger.warning(
+            "profile provider read failed for %r",
+            getattr(session, "profile", None),
+            exc_info=True,
+        )
         return None, None
 
+    _requested = _clean_session_model_provider(requested_provider)
+    if _requested:
+        _profile_prov = _clean_session_model_provider(_provider)
+        if _profile_prov != _requested:
+            return None, None
+        return None, _default
+    return _provider, _default
+
 
 def _resolve_compatible_session_model_state(
     model_id: str | None,
@@ -5303,6 +5324,23 @@ def _resolve_compatible_session_model_state(
             and model_prefix == "openai"
         )
         if not explicit_provider and not stale_codex_openai_slash_id:
+            _profile_default = str(profile_default_model or "").strip()
+            _profile_prov = _clean_session_model_provider(profile_provider)
+            _providers_match_for_repair = (
+                _profile_prov is None or _profile_prov == requested_provider
+            )
+            if (
+                _profile_default
+                and "/" in _profile_default
+                and "/" not in model
+                and _profile_default.rsplit("/", 1)[-1] == model
+                and _providers_match_for_repair
+                and (
+                    requested_provider == "custom"
+                    or str(requested_provider).startswith("custom:")
+                )
+            ):
+                return _profile_default, requested_provider, True
             return model, requested_provider, False
 
     # Default (human chat/start) path calls get_available_models() with NO

api/streaming.py

@@ -1100,13 +1100,61 @@ def _cancelled_turn_hint(agent_name: str | None = None) -> str:
     return f'The run was cancelled by the user before {name} finished. No provider failure occurred.'
 
 
+def _provider_error_probe_text(value) -> tuple[str, int | None]:
+    """Flatten structured provider-error payloads into searchable text."""
+    _texts: list[str] = []
+    _status_code: int | None = None
+    _seen: set[int] = set()
+
+    def _walk(node):
+        nonlocal _status_code
+        if node is None:
+            return
+        if isinstance(node, (dict, list, tuple, set)):
+            _node_id = id(node)
+            if _node_id in _seen:
+                return
+            _seen.add(_node_id)
+        if isinstance(node, dict):
+            for _key in ('type', 'code', 'message', 'detail', 'details', 'name', 'status', 'status_code'):
+                _val = node.get(_key)
+                if _val is None:
+                    continue
+                if _key in ('status', 'status_code') and _status_code is None:
+                    try:
+                        _status_code = int(_val)
+                    except Exception:
+                        pass
+                _texts.append(str(_val))
+            for _key, _val in node.items():
+                if _key in ('type', 'code', 'message', 'detail', 'details', 'name', 'status', 'status_code'):
+                    continue
+                _walk(_val)
+            return
+        if isinstance(node, (list, tuple, set)):
+            for _item in node:
+                _walk(_item)
+            return
+        _texts.append(str(node))
+
+    _walk(value)
+    return ' '.join(t for t in _texts if t).strip(), _status_code
+
+
 def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = False) -> dict:
     """Classify provider/agent failure text for WebUI apperror UX.
 
     Keep this string-based until hermes-agent exposes stable structured
     provider error classes for Codex OAuth plan limits.
     """
-    err_str = str(err_str or '')
+    _probe_text, _probe_status_code = _provider_error_probe_text(err_str)
+    if exc is not None:
+        _exc_probe_text, _exc_status_code = _provider_error_probe_text(exc)
+        if _exc_probe_text:
+            _probe_text = f"{_probe_text} {_exc_probe_text}".strip()
+        if _probe_status_code is None:
+            _probe_status_code = _exc_status_code
+    err_str = str(_probe_text or err_str or '')
     _err_lower = err_str.lower()
     _exc_name = type(exc).__name__ if exc is not None else ''
     _is_cancelled = (
@@ -1147,6 +1195,8 @@ def _classify_provider_error(err_str: str, exc=None, *, silent_failure: bool = F
     _is_quota = _is_quota_error_text(err_str)
     _is_auth = (
         not _is_quota and (
+            _probe_status_code == 401
+            or
             '401' in err_str
             or (exc is not None and 'AuthenticationError' in _exc_name)
             or 'authentication' in _err_lower
@@ -4890,8 +4940,22 @@ def _merge_display_messages_after_agent_result(previous_display, previous_contex
                 previous_user_tail,
                 previous_context=previous_context,
             )
-        candidates = _strip_replayed_prefix(previous_display, candidates)
-        candidates = _strip_replayed_prefix(previous_context, candidates)
+        current_user_key = _message_identity({'role': 'user', 'content': msg_text})
+        current_user_in_candidates = any(
+            _message_identity(m) == current_user_key or _looks_like_current_user_turn(m, msg_text)
+            for m in candidates
+        )
+        assistant_or_tool_only_candidates = bool(candidates) and all(
+            _is_context_compression_marker(m)
+            or (
+                isinstance(m, dict)
+                and m.get('role') in ('assistant', 'tool')
+            )
+            for m in candidates
+        )
+        if not (assistant_or_tool_only_candidates and not current_user_in_candidates):
+            candidates = _strip_replayed_prefix(previous_display, candidates)
+            candidates = _strip_replayed_prefix(previous_context, candidates)
     else:
         current_user_idx = _find_current_user_turn(result_messages, msg_text)
         turn_candidates = result_messages[current_user_idx:] if current_user_idx is not None else []
@@ -5056,6 +5120,64 @@ def _session_lacks_final_assistant_answer(messages) -> bool:
     return True
 
 
+def _merged_transcript_lacks_final_assistant_answer(
+    previous_display,
+    previous_context,
+    result_messages,
+    msg_text,
+    source: str = "webui",
+    drop_replayed_assistant: bool = False,
+) -> bool:
+    """Return True when the current turn still lacks a final assistant answer."""
+    previous_display = list(previous_display or [])
+    merged_messages = _merge_display_messages_after_agent_result(
+        previous_display,
+        previous_context,
+        _restore_reasoning_metadata(previous_display, result_messages),
+        msg_text,
+        source=source,
+    )
+    current_user_idx = _find_current_user_turn(merged_messages, msg_text)
+    if current_user_idx is None or current_user_idx < len(previous_display):
+        # The active turn lives after the durable transcript boundary. If the
+        # merged display only exposes an older user row, materialize the pending
+        # prompt so a replayed assistant row cannot satisfy the wrong turn.
+        pending_user = {
+            'role': 'user',
+            'content': msg_text,
+        }
+        if source and source != 'webui':
+            pending_user['_source'] = source
+        merged_messages.append(pending_user)
+        current_user_idx = len(merged_messages) - 1
+
+    current_user_key = _message_identity(merged_messages[current_user_idx])
+    filtered_messages = merged_messages[:current_user_idx + 1]
+    if drop_replayed_assistant:
+        prior_id_set = {
+            _message_identity(msg)
+            for msg in merged_messages[:current_user_idx]
+            if isinstance(msg, dict)
+        }
+        for msg in merged_messages[current_user_idx + 1:]:
+            if not isinstance(msg, dict):
+                filtered_messages.append(msg)
+                continue
+            if msg.get('role') == 'assistant':
+                key = _message_identity(msg)
+                if key is not None and key in prior_id_set:
+                    continue
+            filtered_messages.append(msg)
+    else:
+        filtered_messages.extend(merged_messages[current_user_idx + 1:])
+    if current_user_key is not None:
+        filtered_messages = [
+            msg for msg in filtered_messages
+            if _message_identity(msg) != current_user_key or msg is merged_messages[current_user_idx]
+        ]
+    return _session_lacks_final_assistant_answer(filtered_messages)
+
+
 def _agent_result_terminal_failure(result) -> bool:
     """Return True for agent results that must not be finalized as done."""
     if not isinstance(result, dict):
@@ -7973,15 +8095,32 @@ def _run_agent_streaming(
                     _previous_context_messages,
                     msg_text,
                 )
+                _last_err = getattr(agent, '_last_error', None) or result.get('error') or ''
+                _classification = _classify_provider_error(
+                    str(_last_err) if _last_err else '',
+                    _last_err,
+                    silent_failure=not bool(_last_err),
+                )
+                _is_quota = _classification['type'] == 'quota_exhausted'
+                _is_auth = _classification['type'] == 'auth_mismatch'
+                _drop_replayed_assistant = (
+                    _agent_result_terminal_failure(result)
+                    or bool(getattr(agent, '_last_error', None))
+                    or ('error' in result and result.get('error') is not None)
+                )
+                _saved_transcript_lacks_final_answer = _merged_transcript_lacks_final_assistant_answer(
+                    _previous_messages,
+                    _previous_context_messages,
+                    _all_result_messages,
+                    msg_text,
+                    source=getattr(s, 'pending_user_source', None) or 'webui',
+                    drop_replayed_assistant=_drop_replayed_assistant,
+                )
                 _terminal_failure = (
                     _agent_result_terminal_failure(result)
                     or (
-                        _tool_limit_reached
-                        and _session_lacks_final_assistant_answer(_all_result_messages)
-                    )
-                    or (
-                        not _token_sent
-                        and _session_lacks_final_assistant_answer(_all_result_messages)
+                        _saved_transcript_lacks_final_answer
+                        and _classification['type'] not in {'cancelled', 'interrupted'}
                     )
                 )
                 if _terminal_failure:
@@ -8007,15 +8146,7 @@ def _run_agent_streaming(
                                 logger.debug("Failed to append cancelled turn journal event", exc_info=True)
                         put('cancel', _cancel_event_payload('Cancelled by user'))
                         return
-                    _last_err = getattr(agent, '_last_error', None) or result.get('error') or ''
                     _err_str = str(_last_err) if _last_err else ''
-                    _classification = _classify_provider_error(
-                        _err_str,
-                        _last_err,
-                        silent_failure=not bool(_err_str),
-                    )
-                    _is_quota = _classification['type'] == 'quota_exhausted'
-                    _is_auth = _classification['type'] == 'auth_mismatch'
                     if _is_quota:
                         _err_label = _classification['label']
                         _err_type = _classification['type']

static/messages.js

@@ -5447,6 +5447,10 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
           if(typeof showToast==='function') showToast(typeof t==='function'?t('approval_gateway_unsupported_label'):'Approvals not supported',4000,'warning');
           return;
         }
+        if(d.type==='approval_gateway_offline'){
+          if(typeof showToast==='function') showToast(d.message||'Gateway offline',4000,'warning');
+          return;
+        }
         // Show as a small inline notice, not a full error
         setComposerStatus(`${d.message||'Warning'}`);
         // If it's a fallback notice, show it briefly then clear

tests/test_auto_compression_terminal_failure.py

@@ -177,9 +177,10 @@ def test_terminal_failure_gates_shape_check_to_no_streamed_text():
     block = src[start:end]
 
     assert "_agent_result_terminal_failure(result)" in block
-    assert "not _token_sent" in block
-    assert "_session_lacks_final_assistant_answer(_all_result_messages)" in block
-    assert "not _assistant_added" not in block
+    assert "_saved_transcript_lacks_final_answer" in block
+    assert "_classification['type'] not in {'cancelled', 'interrupted'}" in block
+    assert "not _token_sent" not in block
+    assert "_session_lacks_final_assistant_answer(_all_result_messages)" not in block
 
 
 def test_completed_tool_tail_without_final_assistant_is_not_successful_done():

tests/test_gateway_approval_runs_api.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 import io
 import json
+import socket
 import threading
 import urllib.error
 from unittest.mock import MagicMock, patch
@@ -15,7 +16,9 @@ from unittest.mock import MagicMock, patch
 def test_gateway_capability_detection():
     """get_gateway_caps / gateway_supports_approval correctly parse /v1/capabilities."""
     from api.config import (
+        gateway_approval_unavailable_reason,
         gateway_supports_approval,
+        get_gateway_caps,
         invalidate_gateway_caps,
     )
 
@@ -38,6 +41,10 @@ def test_gateway_capability_detection():
         return resp
 
     with patch("urllib.request.urlopen", side_effect=_fake_urlopen_capable):
+        caps = get_gateway_caps("http://fake:1234", "secret")
+        assert caps["capabilities_reachable"] is True
+        assert caps["probe_error"] is None
+        assert gateway_approval_unavailable_reason("http://fake:1234", "secret") is None
         assert gateway_supports_approval("http://fake:1234", "secret") is True
 
     invalidate_gateway_caps()
@@ -52,11 +59,90 @@ def test_gateway_capability_detection():
         return resp
 
     with patch("urllib.request.urlopen", side_effect=_fake_urlopen_incapable):
+        caps = get_gateway_caps("http://fake:5678")
+        assert caps["capabilities_reachable"] is True
+        assert caps["probe_error"] is None
+        assert gateway_approval_unavailable_reason("http://fake:5678") == "unsupported"
         assert gateway_supports_approval("http://fake:5678") is False
 
     invalidate_gateway_caps()
 
 
+def test_gateway_capability_detection_marks_probe_failures_unreachable():
+    """Probe failures stay non-fatal but remain distinguishable from unsupported gateways."""
+    from api.config import (
+        gateway_approval_unavailable_reason,
+        gateway_supports_approval,
+        get_gateway_caps,
+        invalidate_gateway_caps,
+    )
+
+    invalidate_gateway_caps()
+
+    def _fake_urlopen_fail(req, *, timeout=None):
+        assert req.full_url == "http://fake:9999/v1/capabilities"
+        raise urllib.error.URLError(ConnectionRefusedError("connection refused"))
+
+    with patch("urllib.request.urlopen", side_effect=_fake_urlopen_fail):
+        caps = get_gateway_caps("http://fake:9999", "secret")
+        assert caps["capabilities_reachable"] is False
+        assert caps["probe_error"]
+        assert gateway_approval_unavailable_reason("http://fake:9999", "secret") == "unreachable"
+        assert gateway_supports_approval("http://fake:9999", "secret") is False
+
+    invalidate_gateway_caps()
+
+
+def test_gateway_capability_detection_treats_timeout_probe_as_reachable_unsupported():
+    """Slow probes should preserve the reachable-but-unsupported warning contract."""
+    from api.config import (
+        gateway_approval_unavailable_reason,
+        gateway_supports_approval,
+        get_gateway_caps,
+        invalidate_gateway_caps,
+    )
+
+    invalidate_gateway_caps()
+
+    def _fake_urlopen_timeout(req, *, timeout=None):
+        assert req.full_url == "http://fake:8888/v1/capabilities"
+        raise socket.timeout("timed out")
+
+    with patch("urllib.request.urlopen", side_effect=_fake_urlopen_timeout):
+        caps = get_gateway_caps("http://fake:8888", "secret")
+        assert caps["capabilities_reachable"] is True
+        assert caps["probe_error"]
+        assert gateway_approval_unavailable_reason("http://fake:8888", "secret") == "unsupported"
+        assert gateway_supports_approval("http://fake:8888", "secret") is False
+
+    invalidate_gateway_caps()
+
+
+def test_gateway_capability_detection_treats_404_probe_as_reachable_unsupported():
+    """Older reachable gateways can 404 /v1/capabilities without becoming "offline"."""
+    from api.config import (
+        gateway_approval_unavailable_reason,
+        gateway_supports_approval,
+        get_gateway_caps,
+        invalidate_gateway_caps,
+    )
+
+    invalidate_gateway_caps()
+
+    def _fake_urlopen_404(req, *, timeout=None):
+        assert req.full_url == "http://fake:7777/v1/capabilities"
+        raise urllib.error.HTTPError(req.full_url, 404, "Not Found", hdrs=None, fp=io.BytesIO(b""))
+
+    with patch("urllib.request.urlopen", side_effect=_fake_urlopen_404):
+        caps = get_gateway_caps("http://fake:7777", "secret")
+        assert caps["capabilities_reachable"] is True
+        assert caps["probe_error"]
+        assert gateway_approval_unavailable_reason("http://fake:7777", "secret") == "unsupported"
+        assert gateway_supports_approval("http://fake:7777", "secret") is False
+
+    invalidate_gateway_caps()
+
+
 def test_gateway_capability_cache_keeps_fresher_success_on_probe_race():
     """A slower failed probe must not overwrite a fresher successful capability result."""
     from api.config import gateway_supports_approval, invalidate_gateway_caps

tests/test_issue4300_gateway_approval_notice.py

@@ -17,17 +17,18 @@ def test_gateway_chat_has_approval_notice_emitted_attribute_check():
 def test_gateway_chat_emits_approval_gateway_unsupported_event():
     """Verify put_gateway_event is called with approval_gateway_unsupported type on non-terminal channel."""
     assert "put_gateway_event(\"warning\"" in GATEWAY_CHAT
-    assert "\"type\": \"approval_gateway_unsupported\"" in GATEWAY_CHAT
+    assert "approval_type = \"approval_gateway_unsupported\"" in GATEWAY_CHAT
 
 
 def test_gateway_chat_once_per_session_guard_pattern():
     """Verify the once-per-session guard: capability check + hasattr + flag check + flag set."""
-    assert "if not gateway_supports_approval(base_url, api_key):" in GATEWAY_CHAT
+    assert "approval_reason = gateway_approval_unavailable_reason(base_url, api_key)" in GATEWAY_CHAT
+    assert "if approval_reason is not None:" in GATEWAY_CHAT
     assert "if not hasattr(s, \"_approval_notice_emitted\"):" in GATEWAY_CHAT
     assert "if not s._approval_notice_emitted:" in GATEWAY_CHAT
     assert "s._approval_notice_emitted = True" in GATEWAY_CHAT
     # Verify order: capability gate before session guard before flag set
-    cap_pos = GATEWAY_CHAT.find("if not gateway_supports_approval(base_url, api_key):")
+    cap_pos = GATEWAY_CHAT.find("if approval_reason is not None:")
     hasattr_pos = GATEWAY_CHAT.find("if not hasattr(s, \"_approval_notice_emitted\"):")
     flag_check_pos = GATEWAY_CHAT.find("if not s._approval_notice_emitted:")
     flag_set_pos = GATEWAY_CHAT.find("s._approval_notice_emitted = True")
@@ -36,8 +37,8 @@ def test_gateway_chat_once_per_session_guard_pattern():
 
 def test_gateway_chat_event_payload_contains_type_and_message():
     """Verify the event payload has type and message fields."""
-    assert "\"type\": \"approval_gateway_unsupported\"" in GATEWAY_CHAT
-    assert "\"message\": \"Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.\"" in GATEWAY_CHAT
+    assert "approval_type = \"approval_gateway_unsupported\"" in GATEWAY_CHAT
+    assert "approval_message = \"Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.\"" in GATEWAY_CHAT
 
 
 def test_messages_js_handles_approval_gateway_unsupported_event():

tests/test_issue5121_provider_auth_terminal_error.py

@@ -0,0 +1,544 @@
+"""Regression tests for issue #5121: provider auth failures must persist as terminal error turns."""
+
+from __future__ import annotations
+
+import queue
+import sys
+import types
+from unittest import mock
+
+import pytest
+
+import api.config as config
+import api.models as models
+import api.streaming as streaming
+from api.models import Session
+
+
+@pytest.fixture(autouse=True)
+def _isolate_session_dir(tmp_path, monkeypatch):
+    session_dir = tmp_path / "sessions"
+    session_dir.mkdir()
+    index_file = session_dir / "_index.json"
+    monkeypatch.setattr(models, "SESSION_DIR", session_dir)
+    monkeypatch.setattr(models, "SESSION_INDEX_FILE", index_file)
+    models.SESSIONS.clear()
+    yield
+    models.SESSIONS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _isolate_stream_state():
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.STREAM_PARTIAL_TEXT.clear()
+    if hasattr(config, "STREAM_REASONING_TEXT"):
+        config.STREAM_REASONING_TEXT.clear()
+    if hasattr(config, "STREAM_LIVE_TOOL_CALLS"):
+        config.STREAM_LIVE_TOOL_CALLS.clear()
+    yield
+    config.STREAMS.clear()
+    config.CANCEL_FLAGS.clear()
+    config.AGENT_INSTANCES.clear()
+    config.STREAM_PARTIAL_TEXT.clear()
+    if hasattr(config, "STREAM_REASONING_TEXT"):
+        config.STREAM_REASONING_TEXT.clear()
+    if hasattr(config, "STREAM_LIVE_TOOL_CALLS"):
+        config.STREAM_LIVE_TOOL_CALLS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _isolate_agent_locks():
+    config.SESSION_AGENT_LOCKS.clear()
+    yield
+    config.SESSION_AGENT_LOCKS.clear()
+
+
+@pytest.fixture(autouse=True)
+def _neutralize_credential_self_heal(monkeypatch):
+    """Make the 401 self-heal path a deterministic no-op by default.
+
+    The terminal-auth-failure tests assert that an unrecoverable 401 surfaces
+    an ``apperror`` / persisted ``_error`` turn. The streaming settlement path
+    first tries ``_attempt_credential_self_heal`` (#1401), which calls
+    ``read_auth_json()``. On a host with a populated ``~/.hermes/auth.json``
+    (e.g. a developer's real Hermes box) self-heal can succeed and silently
+    retry the mock agent, swallowing the error the test expects — so the
+    outcome would depend on host credentials. CI / Windows boxes have no such
+    credentials, which is why the tests pass there but fail on a live agent
+    host. Force self-heal off by default so every host exercises the
+    unrecoverable-failure path; the one test that intentionally verifies a
+    successful retry patches this symbol explicitly inside its own body.
+    """
+    monkeypatch.setattr(streaming, "_attempt_credential_self_heal", lambda *a, **k: None)
+    yield
+
+
+@pytest.fixture(autouse=True)
+def _mock_hermes_modules(monkeypatch):
+    fake_runtime_module = types.ModuleType("hermes_cli.runtime_provider")
+    fake_runtime_module.resolve_runtime_provider = lambda requested=None, **_kw: {
+        "provider": requested or "test-provider",
+        "api_key": "synthetic-key",
+        "base_url": None,
+    }
+    fake_hermes_cli = types.ModuleType("hermes_cli")
+    fake_hermes_cli.runtime_provider = fake_runtime_module
+    fake_hermes_state = types.ModuleType("hermes_state")
+    fake_hermes_state.SessionDB = mock.Mock(return_value=None)
+
+    injected = {
+        "hermes_cli": fake_hermes_cli,
+        "hermes_cli.runtime_provider": fake_runtime_module,
+        "hermes_state": fake_hermes_state,
+    }
+    missing = object()
+    saved = {k: sys.modules.get(k, missing) for k in injected}
+    sys.modules.update(injected)
+    yield
+    for name, prev in saved.items():
+        if prev is missing:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = prev
+
+
+class MockAgent:
+    def __init__(self, **kwargs):
+        self.session_id = kwargs.get("session_id")
+        self.stream_delta_callback = kwargs.get("stream_delta_callback")
+        self.reasoning_callback = kwargs.get("reasoning_callback")
+        self.tool_progress_callback = kwargs.get("tool_progress_callback")
+        self.session_prompt_tokens = 0
+        self.session_completion_tokens = 0
+        self.session_estimated_cost_usd = 0.0
+        self.context_compressor = None
+        self._last_error = None
+        self.ephemeral_system_prompt = None
+
+    def run_conversation(self, **kwargs):
+        raise NotImplementedError
+
+    def interrupt(self, _message):
+        pass
+
+
+def _prepare_session(session_id: str, stream_id: str, *, pending_user_message: str, partial_source: str = "cli"):
+    session = Session(session_id=session_id, title="Test Session")
+    session.messages = []
+    session.context_messages = []
+    session.pending_user_message = pending_user_message
+    session.pending_attachments = ["attachment.txt"]
+    session.pending_started_at = 1234567890.0
+    session.pending_user_source = partial_source
+    session.active_stream_id = stream_id
+    session.save()
+    models.SESSIONS[session_id] = session
+    return session
+
+
+def _seed_prior_turn(session, *, prior_user: str, prior_assistant: str):
+    session.messages = [
+        {"role": "user", "content": prior_user, "timestamp": 1},
+        {"role": "assistant", "content": prior_assistant, "timestamp": 2},
+    ]
+    session.context_messages = [
+        {"role": "user", "content": prior_user},
+        {"role": "assistant", "content": prior_assistant},
+    ]
+    session.save()
+
+
+def _queue_events(fake_queue):
+    return [(item[0], item[1]) for item in list(fake_queue.queue)]
+
+
+def _auth_failure_error_payload():
+    return {
+        "error": {
+            "type": "authentication_error",
+            "status_code": 401,
+            "code": "auth_unavailable",
+            "message": "Your authentication token has been invalidated. Please try signing in again.",
+        }
+    }
+
+
+def _build_auth_failure_agent(*, token_text: str | None, success_text: str = "Recovered auth reply"):
+    class AuthFailureAgent(MockAgent):
+        runs = 0
+
+        def run_conversation(self, **kwargs):
+            type(self).runs += 1
+            history = list(kwargs.get("conversation_history") or [])
+            if type(self).runs == 1:
+                if self.stream_delta_callback is not None and token_text is not None:
+                    self.stream_delta_callback(token_text)
+                return {
+                    "messages": history,
+                    "error": _auth_failure_error_payload(),
+                }
+            return {
+                "status": "ok",
+                "messages": history + [{"role": "assistant", "content": success_text}],
+            }
+
+    return AuthFailureAgent
+
+
+def _run_stream(monkeypatch, session, stream_id, agent_cls, *, workspace):
+    fake_queue = queue.Queue()
+    streaming.STREAMS[stream_id] = fake_queue
+    config.STREAM_PARTIAL_TEXT[stream_id] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=agent_cls), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]):
+        streaming._run_agent_streaming(
+            session_id=session.session_id,
+            msg_text=session.pending_user_message,
+            model="test-model",
+            workspace=workspace,
+            stream_id=stream_id,
+        )
+
+    return fake_queue
+
+
+def test_auth_401_without_delivery_persists_error_turn(tmp_path, monkeypatch):
+    session = _prepare_session("auth_no_delivery", "stream_auth_no_delivery", pending_user_message="Please respond")
+    agent_cls = _build_auth_failure_agent(token_text=None)
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_auth_no_delivery", agent_cls, workspace=str(tmp_path))
+    saved = Session.load("auth_no_delivery")
+    assert saved is not None
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for auth failure"
+    assert apperrors[-1]["type"] == "auth_mismatch"
+    assert not any(event == "done" for event, _ in events)
+
+    assert saved.active_stream_id is None
+    assert saved.pending_user_message is None
+    assert saved.pending_attachments == []
+    assert saved.pending_started_at is None
+    assert saved.pending_user_source is None
+    assert saved.messages[-1]["_error"] is True
+    assert saved.messages[-1]["role"] == "assistant"
+    assert any(msg.get("role") == "user" for msg in saved.messages)
+
+
+def test_auth_401_after_partial_preserves_partial_then_error(tmp_path, monkeypatch):
+    session = _prepare_session("auth_partial", "stream_auth_partial", pending_user_message="Please stream then fail")
+    agent_cls = _build_auth_failure_agent(token_text="Partial auth text")
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_auth_partial", agent_cls, workspace=str(tmp_path))
+    saved = Session.load("auth_partial")
+    assert saved is not None
+
+    partial = next((msg for msg in saved.messages if msg.get("_partial")), None)
+    assert partial is not None
+    assert partial["role"] == "assistant"
+    assert partial["content"] == "Partial auth text"
+
+    error_idx = next(i for i, msg in enumerate(saved.messages) if msg.get("_error"))
+    partial_idx = saved.messages.index(partial)
+    assert partial_idx < error_idx
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors and apperrors[-1]["type"] == "auth_mismatch"
+
+
+def test_auth_401_seeded_multi_turn_partial_persists_error_turn(tmp_path, monkeypatch):
+    session = _prepare_session("auth_seeded_partial", "stream_auth_seeded_partial", pending_user_message="Please stream then fail")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Earlier answer",
+    )
+    agent_cls = _build_auth_failure_agent(token_text="Partial auth text")
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_auth_seeded_partial", agent_cls, workspace=str(tmp_path))
+    saved = Session.load("auth_seeded_partial")
+    assert saved is not None
+
+    assert any(msg.get("role") == "assistant" and msg.get("content") == "Earlier answer" for msg in saved.messages)
+    assert any(msg.get("_partial") and msg.get("content") == "Partial auth text" for msg in saved.messages)
+    assert saved.messages[-1]["_error"] is True
+    assert saved.messages[-1]["role"] == "assistant"
+    assert any(msg.get("role") == "user" and msg.get("content") == "Please stream then fail" for msg in saved.messages)
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors and apperrors[-1]["type"] == "auth_mismatch"
+    assert not any(event == "done" for event, _ in events)
+
+
+def test_auth_401_classification_receives_stringified_probe_text(tmp_path, monkeypatch):
+    session = _prepare_session("auth_probe_text", "stream_auth_probe_text", pending_user_message="Please fail")
+    agent_cls = _build_auth_failure_agent(token_text=None)
+    observed = {}
+    real_classify = streaming._classify_provider_error
+
+    def _spy_classify_provider_error(err_str, exc=None, *, silent_failure=False):
+        observed["err_str"] = err_str
+        observed["exc"] = exc
+        observed["silent_failure"] = silent_failure
+        return real_classify(err_str, exc, silent_failure=silent_failure)
+
+    with mock.patch.object(streaming, "_classify_provider_error", side_effect=_spy_classify_provider_error):
+        _run_stream(monkeypatch, session, "stream_auth_probe_text", agent_cls, workspace=str(tmp_path))
+
+    assert observed["err_str"] == str(_auth_failure_error_payload())
+    assert observed["exc"] == _auth_failure_error_payload()
+    assert observed["silent_failure"] is False
+
+
+def test_auth_401_seeded_replayed_assistant_does_not_satisfy_current_turn(tmp_path, monkeypatch):
+    session = _prepare_session("auth_seeded_replay", "stream_auth_seeded_replay", pending_user_message="Please respond now")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Earlier answer",
+    )
+
+    class ReplayAssistantAuthFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": history + [{"role": "assistant", "content": "Earlier answer"}],
+                "error": _auth_failure_error_payload(),
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_auth_seeded_replay", ReplayAssistantAuthFailureAgent, workspace=str(tmp_path))
+    saved = Session.load("auth_seeded_replay")
+    assert saved is not None
+
+    assert any(msg.get("role") == "user" and msg.get("content") == "Please respond now" for msg in saved.messages)
+    assert saved.messages[-1]["_error"] is True
+    assert saved.messages[-1]["role"] == "assistant"
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors and apperrors[-1]["type"] == "auth_mismatch"
+    assert not any(event == "done" for event, _ in events)
+
+
+def test_auth_retry_success_does_not_append_error_turn(tmp_path, monkeypatch):
+    session = _prepare_session("auth_retry", "stream_auth_retry", pending_user_message="Please retry")
+    agent_cls = _build_auth_failure_agent(token_text="")
+
+    heal_rt = {
+        "provider": "test-provider",
+        "api_key": "fresh-key",
+        "base_url": None,
+    }
+
+    fake_queue = queue.Queue()
+    streaming.STREAMS["stream_auth_retry"] = fake_queue
+    config.STREAM_PARTIAL_TEXT["stream_auth_retry"] = ""
+
+    with mock.patch.object(streaming, "get_session", return_value=session), \
+         mock.patch.object(streaming, "_get_ai_agent", return_value=agent_cls), \
+         mock.patch.object(streaming, "resolve_model_provider", return_value=("test-model", "test-provider", None)), \
+         mock.patch("api.config.get_config", return_value={}), \
+         mock.patch("api.config._resolve_cli_toolsets", return_value=[]), \
+         mock.patch.object(streaming, "_attempt_credential_self_heal", return_value=heal_rt):
+        streaming._run_agent_streaming(
+            session_id=session.session_id,
+            msg_text=session.pending_user_message,
+            model="test-model",
+            workspace=str(tmp_path),
+            stream_id="stream_auth_retry",
+        )
+
+    saved = Session.load("auth_retry")
+    assert saved is not None
+
+    events = _queue_events(fake_queue)
+    assert not any(event == "apperror" for event, _ in events)
+    assert any(event == "done" for event, _ in events)
+    assert saved.messages[-1]["role"] == "assistant"
+    assert saved.messages[-1]["content"] == "Recovered auth reply"
+    assert not any(msg.get("_error") for msg in saved.messages)
+
+
+def test_success_repeated_assistant_text_stays_successful_current_turn(tmp_path, monkeypatch):
+    session = _prepare_session("repeat_success", "stream_repeat_success", pending_user_message="Please say it again")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Same answer",
+    )
+
+    class RepeatedSuccessAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": history + [{"role": "assistant", "content": "Same answer"}],
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_repeat_success", RepeatedSuccessAgent, workspace=str(tmp_path))
+    saved = Session.load("repeat_success")
+    assert saved is not None
+
+    assert any(msg.get("role") == "user" and msg.get("content") == "Please say it again" for msg in saved.messages)
+    assert saved.messages[-1]["role"] == "assistant"
+    assert saved.messages[-1]["content"] == "Same answer"
+    assert not any(msg.get("_error") for msg in saved.messages)
+
+    events = _queue_events(fake_queue)
+    assert any(event == "done" for event, _ in events)
+    assert not any(event == "apperror" for event, _ in events)
+
+
+def test_success_repeated_assistant_text_ignores_empty_error_field(tmp_path, monkeypatch):
+    session = _prepare_session("repeat_success_empty_error", "stream_repeat_success_empty_error", pending_user_message="Please say it again")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Same answer",
+    )
+
+    class RepeatedSuccessWithEmptyErrorAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": history + [{"role": "assistant", "content": "Same answer"}],
+                "error": None,
+            }
+
+    fake_queue = _run_stream(
+        monkeypatch,
+        session,
+        "stream_repeat_success_empty_error",
+        RepeatedSuccessWithEmptyErrorAgent,
+        workspace=str(tmp_path),
+    )
+    saved = Session.load("repeat_success_empty_error")
+    assert saved is not None
+
+    assert any(msg.get("role") == "user" and msg.get("content") == "Please say it again" for msg in saved.messages)
+    assert saved.messages[-1]["role"] == "assistant"
+    assert saved.messages[-1]["content"] == "Same answer"
+    assert not any(msg.get("_error") for msg in saved.messages)
+
+    events = _queue_events(fake_queue)
+    assert any(event == "done" for event, _ in events)
+    assert not any(event == "apperror" for event, _ in events)
+
+
+def test_non_auth_silent_failure_still_uses_no_response(tmp_path, monkeypatch):
+    session = _prepare_session("silent_failure", "stream_silent_failure", pending_user_message="Please handle silence")
+
+    class SilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            return {
+                "messages": list(kwargs.get("conversation_history") or []),
+                "error": "",
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_silent_failure", SilentFailureAgent, workspace=str(tmp_path))
+    saved = Session.load("silent_failure")
+    assert saved is not None
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for silent failure"
+    assert apperrors[-1]["type"] == "no_response"
+    assert apperrors[-1]["type"] != "auth_mismatch"
+    assert saved.messages[-1]["_error"] is True
+
+
+def test_non_auth_partial_delivery_persists_error_turn(tmp_path, monkeypatch):
+    session = _prepare_session("partial_escape", "stream_partial_escape", pending_user_message="Please handle partial silence")
+
+    class PartialSilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback is not None:
+                self.stream_delta_callback("Partial text before failure")
+            return {
+                "messages": list(kwargs.get("conversation_history") or []),
+                "error": "",
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_partial_escape", PartialSilentFailureAgent, workspace=str(tmp_path))
+    saved = Session.load("partial_escape")
+    assert saved is not None
+
+    partial = next((msg for msg in saved.messages if msg.get("_partial")), None)
+    assert partial is not None
+    assert partial["content"] == "Partial text before failure"
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for partial silent failure"
+    assert apperrors[-1]["type"] == "no_response"
+    assert saved.messages[-1]["_error"] is True
+
+
+def test_non_auth_seeded_multi_turn_partial_persists_error_turn(tmp_path, monkeypatch):
+    session = _prepare_session("seeded_partial_escape", "stream_seeded_partial_escape", pending_user_message="Please handle partial silence")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Earlier answer",
+    )
+
+    class PartialSilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            if self.stream_delta_callback is not None:
+                self.stream_delta_callback("Partial text before failure")
+            return {
+                "messages": list(kwargs.get("conversation_history") or []),
+                "error": "",
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_seeded_partial_escape", PartialSilentFailureAgent, workspace=str(tmp_path))
+    saved = Session.load("seeded_partial_escape")
+    assert saved is not None
+
+    assert any(msg.get("role") == "assistant" and msg.get("content") == "Earlier answer" for msg in saved.messages)
+    assert any(msg.get("_partial") and msg.get("content") == "Partial text before failure" for msg in saved.messages)
+    assert saved.messages[-1]["_error"] is True
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for seeded partial silent failure"
+    assert apperrors[-1]["type"] == "no_response"
+    assert not any(event == "done" for event, _ in events)
+
+
+def test_non_auth_seeded_replayed_assistant_does_not_satisfy_current_turn(tmp_path, monkeypatch):
+    session = _prepare_session("seeded_replay_escape", "stream_seeded_replay_escape", pending_user_message="Please handle this now")
+    _seed_prior_turn(
+        session,
+        prior_user="Earlier question",
+        prior_assistant="Earlier answer",
+    )
+
+    class ReplayAssistantSilentFailureAgent(MockAgent):
+        def run_conversation(self, **kwargs):
+            history = list(kwargs.get("conversation_history") or [])
+            return {
+                "messages": history + [{"role": "assistant", "content": "Earlier answer"}],
+                "error": "",
+            }
+
+    fake_queue = _run_stream(monkeypatch, session, "stream_seeded_replay_escape", ReplayAssistantSilentFailureAgent, workspace=str(tmp_path))
+    saved = Session.load("seeded_replay_escape")
+    assert saved is not None
+
+    assert any(msg.get("role") == "user" and msg.get("content") == "Please handle this now" for msg in saved.messages)
+    assert saved.messages[-1]["_error"] is True
+
+    events = _queue_events(fake_queue)
+    apperrors = [data for event, data in events if event == "apperror"]
+    assert apperrors, "expected apperror for seeded replay silent failure"
+    assert apperrors[-1]["type"] == "no_response"
+    assert not any(event == "done" for event, _ in events)

tests/test_issue5127_process_wakeup_bare_model.py

@@ -0,0 +1,152 @@
+"""Tests for issue #5127 — process-wakeup bare custom-provider model repair.
+
+When ``_resolve_compatible_session_model_state`` takes the #1855 fast path
+(both model and model_provider set), a bare runtime model that matches the
+suffix of a slash-qualified ``profile_default_model`` must repair to the full
+qualified ID for named custom providers.
+"""
+from unittest.mock import patch
+
+
+class TestIssue5127CustomProviderBareSuffixRepair:
+    def test_fast_path_repairs_bare_suffix_to_profile_default(self):
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            result = _resolve_compatible_session_model_state(
+                "grok-composer-2.5-fast",
+                "custom:my-proxy",
+                profile_provider="custom:my-proxy",
+                profile_default_model="x-ai/grok-composer-2.5-fast",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count == 0
+        assert result == ("x-ai/grok-composer-2.5-fast", "custom:my-proxy", True)
+
+    def test_fast_path_skips_repair_when_profile_provider_mismatches(self):
+        """Regression: custom:other-proxy must not inherit my-proxy's qualified default."""
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            result = _resolve_compatible_session_model_state(
+                "grok-composer-2.5-fast",
+                "custom:other-proxy",
+                profile_provider="custom:my-proxy",
+                profile_default_model="x-ai/grok-composer-2.5-fast",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count == 0
+        assert result == ("grok-composer-2.5-fast", "custom:other-proxy", False)
+
+    def test_fast_path_repairs_generic_custom_provider(self):
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            result = _resolve_compatible_session_model_state(
+                "some-model",
+                "custom",
+                profile_default_model="vendor/some-model",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count == 0
+        assert result == ("vendor/some-model", "custom", True)
+
+    def test_fast_path_does_not_rewrite_unrelated_bare_model(self):
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            result = _resolve_compatible_session_model_state(
+                "other-model",
+                "custom:my-proxy",
+                profile_default_model="x-ai/grok-composer-2.5-fast",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count == 0
+        assert result == ("other-model", "custom:my-proxy", False)
+
+    def test_fast_path_keeps_qualified_model_unchanged(self):
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            result = _resolve_compatible_session_model_state(
+                "x-ai/grok-composer-2.5-fast",
+                "custom:my-proxy",
+                profile_default_model="x-ai/grok-composer-2.5-fast",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count == 0
+        assert result == ("x-ai/grok-composer-2.5-fast", "custom:my-proxy", False)
+
+    def test_openai_codex_stale_openai_slash_still_uses_slow_path(self):
+        """Regression: openai/... under openai-codex must not take bare fast return."""
+        from api.routes import _resolve_compatible_session_model_state
+
+        with patch("api.routes.get_available_models") as mock_catalog:
+            mock_catalog.return_value = {
+                "active_provider": "openai-codex",
+                "default_model": "gpt-5.5",
+                "groups": [],
+            }
+            result = _resolve_compatible_session_model_state(
+                "openai/gpt-5.4-mini",
+                "openai-codex",
+                profile_default_model="gpt-5.5",
+                prefer_cached_catalog=True,
+            )
+
+        assert mock_catalog.call_count >= 1
+        assert result[0] == "gpt-5.5"
+        assert result[2] is True
+
+
+class TestReadProfileModelConfigWithExplicitProvider:
+    def test_returns_profile_default_when_session_has_model_provider(self, tmp_path, monkeypatch):
+        from api.routes import _read_profile_model_config
+
+        profile_home = tmp_path / "prof"
+        profile_home.mkdir()
+        (profile_home / "config.yaml").write_text(
+            "model:\n  provider: custom:my-proxy\n  default: x-ai/grok-composer-2.5-fast\n",
+            encoding="utf-8",
+        )
+
+        class _Session:
+            profile = "testprof"
+
+        monkeypatch.setattr(
+            "api.profiles.get_hermes_home_for_profile",
+            lambda _p: str(profile_home),
+        )
+
+        provider, default = _read_profile_model_config(_Session(), "custom:my-proxy")
+        assert provider is None
+        assert default == "x-ai/grok-composer-2.5-fast"
+
+    def test_returns_none_default_when_session_provider_differs_from_profile(
+        self, tmp_path, monkeypatch,
+    ):
+        from api.routes import _read_profile_model_config
+
+        profile_home = tmp_path / "prof"
+        profile_home.mkdir()
+        (profile_home / "config.yaml").write_text(
+            "model:\n  provider: custom:my-proxy\n  default: x-ai/grok-composer-2.5-fast\n",
+            encoding="utf-8",
+        )
+
+        class _Session:
+            profile = "testprof"
+
+        monkeypatch.setattr(
+            "api.profiles.get_hermes_home_for_profile",
+            lambda _p: str(profile_home),
+        )
+
+        provider, default = _read_profile_model_config(_Session(), "custom:other-proxy")
+        assert provider is None
+        assert default is None

tests/test_issue5139_gateway_approval_offline_notice.py

@@ -0,0 +1,255 @@
+"""Regression coverage for #5139 gateway approval offline notice."""
+
+from __future__ import annotations
+
+import io
+import json
+import urllib.error
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from api.config import STREAMS, STREAMS_LOCK, invalidate_gateway_caps
+from api.gateway_chat import _run_gateway_chat_streaming
+
+REPO = Path(__file__).resolve().parents[1]
+GATEWAY_CHAT = (REPO / "api" / "gateway_chat.py").read_text(encoding="utf-8")
+MESSAGES_JS = (REPO / "static" / "messages.js").read_text(encoding="utf-8")
+
+
+def _run_gateway_warning_case(unavailable_reason: str) -> list:
+    events = []
+    q = MagicMock()
+    q.put_nowait = lambda item: events.append(item)
+
+    stream_id = f"sid-{unavailable_reason}"
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = q
+
+    mock_session = MagicMock()
+    mock_session.active_stream_id = stream_id
+    mock_session.workspace = "/tmp"
+    mock_session.model = "test"
+    mock_session.model_provider = None
+    mock_session.profile = None
+    mock_session.context_messages = []
+    mock_session.messages = []
+    mock_session.pending_user_message = None
+    mock_session.pending_attachments = None
+    mock_session.pending_started_at = None
+    mock_session._approval_notice_emitted = False
+
+    def fake_urlopen(req, *, timeout=None):
+        assert req.full_url == "http://127.0.0.1:8642/v1/chat/completions"
+        assert req.get_method() == "POST"
+        payload = req.data.decode("utf-8")
+        assert json.loads(payload)["messages"][-1]["content"] == "hi"
+        resp = MagicMock()
+        resp.__iter__ = lambda s: iter(
+            [
+                b'data: {"choices":[{"delta":{"content":"Done"}}]}\n',
+                b"\n",
+                b"data: [DONE]\n",
+            ]
+        )
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = lambda s, *a: None
+        return resp
+
+    try:
+        with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}):
+            with patch("api.gateway_chat.gateway_supports_approval", return_value=False), \
+                 patch("api.gateway_chat.gateway_approval_unavailable_reason", return_value=unavailable_reason), \
+                 patch("urllib.request.urlopen", side_effect=fake_urlopen), \
+                 patch("api.gateway_chat.get_session", return_value=mock_session), \
+                 patch("api.gateway_chat._stream_writeback_is_current", return_value=True), \
+                 patch("api.gateway_chat.merge_session_messages_append_only", return_value=[]):
+                _run_gateway_chat_streaming(
+                    session_id="sess-warning",
+                    msg_text="hi",
+                    model="test-model",
+                    workspace="/tmp",
+                    stream_id=stream_id,
+                )
+    finally:
+        with STREAMS_LOCK:
+            STREAMS.pop(stream_id, None)
+
+    return events
+
+
+def test_gateway_chat_emits_offline_warning_for_unreachable_probe():
+    events = _run_gateway_warning_case("unreachable")
+    warnings = [item for item in events if isinstance(item, tuple) and item[0] == "warning"]
+    assert warnings
+    assert warnings[0][1]["type"] == "approval_gateway_offline"
+    assert warnings[0][1]["message"] == "Gateway connection failed. Check that the connected Hermes gateway is running and reachable."
+    assert any(isinstance(item, tuple) and item[0] == "done" for item in events)
+    assert not any(
+        isinstance(item, tuple) and item[0] == "warning" and item[1].get("type") == "approval_gateway_unsupported"
+        for item in events
+    )
+
+
+def test_gateway_chat_keeps_unsupported_warning_for_reachable_older_gateway():
+    events = _run_gateway_warning_case("unsupported")
+    warnings = [item for item in events if isinstance(item, tuple) and item[0] == "warning"]
+    assert warnings
+    assert warnings[0][1]["type"] == "approval_gateway_unsupported"
+    assert warnings[0][1]["message"] == "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this."
+    assert any(isinstance(item, tuple) and item[0] == "done" for item in events)
+    assert not any(
+        isinstance(item, tuple) and item[0] == "warning" and item[1].get("type") == "approval_gateway_offline"
+        for item in events
+    )
+
+
+def test_gateway_chat_keeps_unsupported_warning_for_404_capabilities_probe():
+    events = []
+    q = MagicMock()
+    q.put_nowait = lambda item: events.append(item)
+
+    stream_id = "sid-http-404"
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = q
+
+    mock_session = MagicMock()
+    mock_session.active_stream_id = stream_id
+    mock_session.workspace = "/tmp"
+    mock_session.model = "test"
+    mock_session.model_provider = None
+    mock_session.profile = None
+    mock_session.context_messages = []
+    mock_session.messages = []
+    mock_session.pending_user_message = None
+    mock_session.pending_attachments = None
+    mock_session.pending_started_at = None
+    mock_session._approval_notice_emitted = False
+
+    def fake_urlopen(req, *, timeout=None):
+        if req.full_url == "http://127.0.0.1:8642/v1/capabilities":
+            raise urllib.error.HTTPError(req.full_url, 404, "Not Found", hdrs=None, fp=io.BytesIO(b""))
+        assert req.full_url == "http://127.0.0.1:8642/v1/chat/completions"
+        resp = MagicMock()
+        resp.__iter__ = lambda s: iter(
+            [
+                b'data: {"choices":[{"delta":{"content":"Done"}}]}\n',
+                b"\n",
+                b"data: [DONE]\n",
+            ]
+        )
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = lambda s, *a: None
+        return resp
+
+    invalidate_gateway_caps()
+
+    try:
+        with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}):
+            with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
+                 patch("api.gateway_chat.get_session", return_value=mock_session), \
+                 patch("api.gateway_chat._stream_writeback_is_current", return_value=True), \
+                 patch("api.gateway_chat.merge_session_messages_append_only", return_value=[]):
+                _run_gateway_chat_streaming(
+                    session_id="sess-404",
+                    msg_text="hi",
+                    model="test-model",
+                    workspace="/tmp",
+                    stream_id=stream_id,
+                )
+    finally:
+        with STREAMS_LOCK:
+            STREAMS.pop(stream_id, None)
+        invalidate_gateway_caps()
+
+    warnings = [item for item in events if isinstance(item, tuple) and item[0] == "warning"]
+    assert warnings
+    assert warnings[0][1]["type"] == "approval_gateway_unsupported"
+    assert warnings[0][1]["message"] == "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this."
+    assert any(isinstance(item, tuple) and item[0] == "done" for item in events)
+    assert not any(
+        isinstance(item, tuple) and item[0] == "warning" and item[1].get("type") == "approval_gateway_offline"
+        for item in events
+    )
+
+
+def test_gateway_chat_keeps_unsupported_warning_for_timeout_capabilities_probe():
+    events = []
+    q = MagicMock()
+    q.put_nowait = lambda item: events.append(item)
+
+    stream_id = "sid-timeout"
+    with STREAMS_LOCK:
+        STREAMS[stream_id] = q
+
+    mock_session = MagicMock()
+    mock_session.active_stream_id = stream_id
+    mock_session.workspace = "/tmp"
+    mock_session.model = "test"
+    mock_session.model_provider = None
+    mock_session.profile = None
+    mock_session.context_messages = []
+    mock_session.messages = []
+    mock_session.pending_user_message = None
+    mock_session.pending_attachments = None
+    mock_session.pending_started_at = None
+    mock_session._approval_notice_emitted = False
+
+    def fake_urlopen(req, *, timeout=None):
+        if req.full_url == "http://127.0.0.1:8642/v1/capabilities":
+            raise TimeoutError("timed out")
+        assert req.full_url == "http://127.0.0.1:8642/v1/chat/completions"
+        resp = MagicMock()
+        resp.__iter__ = lambda s: iter(
+            [
+                b'data: {"choices":[{"delta":{"content":"Done"}}]}\n',
+                b"\n",
+                b"data: [DONE]\n",
+            ]
+        )
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = lambda s, *a: None
+        return resp
+
+    invalidate_gateway_caps()
+
+    try:
+        with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}):
+            with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
+                 patch("api.gateway_chat.get_session", return_value=mock_session), \
+                 patch("api.gateway_chat._stream_writeback_is_current", return_value=True), \
+                 patch("api.gateway_chat.merge_session_messages_append_only", return_value=[]):
+                _run_gateway_chat_streaming(
+                    session_id="sess-timeout",
+                    msg_text="hi",
+                    model="test-model",
+                    workspace="/tmp",
+                    stream_id=stream_id,
+                )
+    finally:
+        with STREAMS_LOCK:
+            STREAMS.pop(stream_id, None)
+        invalidate_gateway_caps()
+
+    warnings = [item for item in events if isinstance(item, tuple) and item[0] == "warning"]
+    assert warnings
+    assert warnings[0][1]["type"] == "approval_gateway_unsupported"
+    assert warnings[0][1]["message"] == "Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this."
+    assert any(isinstance(item, tuple) and item[0] == "done" for item in events)
+    assert not any(
+        isinstance(item, tuple) and item[0] == "warning" and item[1].get("type") == "approval_gateway_offline"
+        for item in events
+    )
+
+
+def test_messages_js_handles_offline_warning_without_touching_unsupported_branch():
+    assert "d.type==='approval_gateway_offline'" in MESSAGES_JS
+    assert "Gateway offline" in MESSAGES_JS
+    assert "d.type==='approval_gateway_unsupported'" in MESSAGES_JS
+    assert "Approvals not supported" in MESSAGES_JS
+    assert "setComposerStatus(`${d.message||'Warning'}`);" in MESSAGES_JS
+
+
+def test_gateway_chat_source_mentions_offline_warning_type():
+    assert "approval_type = \"approval_gateway_offline\"" in GATEWAY_CHAT
+    assert "approval_type = \"approval_gateway_unsupported\"" in GATEWAY_CHAT
+    assert "approval_message = \"Gateway connection failed. Check that the connected Hermes gateway is running and reachable.\"" in GATEWAY_CHAT