fix(#5139): keep older gateways on the unsupported warning path

api/config.py

@@ -7463,6 +7463,10 @@ def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
             features = {}
         caps["approval_events"] = bool(features.get("approval_events"))
         caps["run_approval_response"] = bool(features.get("run_approval_response"))
+    except urllib.error.HTTPError as exc:
+        if exc.code == 404:
+            caps["capabilities_reachable"] = True
+        caps["probe_error"] = f"{type(exc).__name__}: {exc}"
     except Exception as exc:
         caps["probe_error"] = f"{type(exc).__name__}: {exc}"
     with _GATEWAY_CAPS_LOCK:

api/gateway_chat.py

@@ -639,8 +639,8 @@ 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:

tests/test_gateway_approval_runs_api.py

@@ -92,6 +92,31 @@ def test_gateway_capability_detection_marks_probe_failures_unreachable():
     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

@@ -22,12 +22,13 @@ def test_gateway_chat_emits_approval_gateway_unsupported_event():
 
 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")

tests/test_issue5139_gateway_approval_offline_notice.py

@@ -2,7 +2,9 @@
 
 from __future__ import annotations
 
+import io
 import json
+import urllib.error
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -101,6 +103,72 @@ def test_gateway_chat_keeps_unsupported_warning_for_reachable_older_gateway():
     )
 
 
+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
+
+    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)
+
+    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