From 62be5b02eaf77da7378d566c723ae1af1ba10f15 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Sun, 28 Jun 2026 17:32:55 -0400 Subject: [PATCH 1/2] fix(#5135): clear stale gateway approval mirrors on stream teardown --- api/gateway_chat.py | 16 ++ tests/test_gateway_approval_legacy_path.py | 275 +++++++++++++++++++++ 2 files changed, 291 insertions(+) diff --git a/api/gateway_chat.py b/api/gateway_chat.py index ebd202a7b..3eb2ee545 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -477,6 +477,21 @@ def _clear_gateway_pending_state(session: Any, stream_id: str) -> None: session.save() +def _cleanup_gateway_pending_mirror(session_id: str) -> None: + try: + from api.route_approvals import ( + _approval_sse_notify_locked, + _lock as _approval_lock, + reconcile_gateway_pending_mirror_locked, + ) + + with _approval_lock: + head, total, _ = reconcile_gateway_pending_mirror_locked(session_id) + _approval_sse_notify_locked(session_id, head, total) + except Exception: + logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True) + + def _run_gateway_chat_streaming( session_id, msg_text, @@ -888,6 +903,7 @@ def _run_gateway_chat_streaming( _clear_gateway_pending_state(get_session(session_id), stream_id) except Exception: logger.debug("Failed to clear gateway stream state", exc_info=True) + _cleanup_gateway_pending_mirror(session_id) with STREAMS_LOCK: CANCEL_FLAGS.pop(stream_id, None) STREAM_PARTIAL_TEXT.pop(stream_id, None) diff --git a/tests/test_gateway_approval_legacy_path.py b/tests/test_gateway_approval_legacy_path.py index 1ad5eed4b..a170bb728 100644 --- a/tests/test_gateway_approval_legacy_path.py +++ b/tests/test_gateway_approval_legacy_path.py @@ -28,6 +28,41 @@ def _extract_legacy_sse_loop(): return GATEWAY_CHAT_SRC[start:end] +def _drain_queue(q): + import queue + + items = [] + while True: + try: + items.append(q.get_nowait()) + except queue.Empty: + return items + + +def _make_legacy_gateway_urlopen(approval_payload: str, after_approval=None): + def fake_urlopen(req, *, timeout=None): + del req, timeout + + def _iter(): + yield b"event: approval.request" + yield f"data: {approval_payload}".encode("utf-8") + yield b"" + if after_approval is not None: + after_approval() + yield b'data: {"choices":[{"delta":{"content":"Done"}}]}' + yield b"" + yield b"data: [DONE]" + yield b"" + + resp = MagicMock() + resp.__iter__ = lambda s: _iter() + resp.__enter__ = lambda s: s + resp.__exit__ = lambda s, *a: None + return resp + + return fake_urlopen + + def test_legacy_loop_checks_approval_request_event(): """Legacy SSE loop must handle `approval.request` events.""" loop = _extract_legacy_sse_loop() @@ -333,6 +368,246 @@ def test_legacy_approval_records_run_id_for_response_relay(): _STREAM_RUN_IDS.pop(stream_id, None) +def test_legacy_teardown_clears_stale_gateway_mirror_and_notifies_empty_state(): + """Legacy teardown must remove a stale gateway mirror and publish empty SSE state.""" + from types import SimpleNamespace + from api import route_approvals as ra + from api.config import STREAMS, STREAMS_LOCK + from api.gateway_chat import _run_gateway_chat_streaming + + session_id = "sess-legacy-teardown-stale" + stream_id = "sid-legacy-teardown-stale" + approval_data = { + "command": "rm -rf /tmp/test", + "description": "Delete temporary files", + "pattern_key": "dangerous_command", + "pattern_keys": ["dangerous_command"], + "approval_id": "appr-legacy-teardown-stale", + "choices": ["once", "session", "always", "deny"], + "run_id": "run-legacy-teardown-stale", + } + approval_payload = json.dumps(approval_data) + + subscriber = ra._approval_sse_subscribe(session_id) + q = MagicMock() + q.put_nowait = lambda item: None + + 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.pending_user_source = None + + try: + with ra._lock: + ra._pending.pop(session_id, None) + ra._gateway_queues[session_id] = [SimpleNamespace(data=dict(approval_data))] + with STREAMS_LOCK: + STREAMS[stream_id] = q + + def clear_gateway_queue(): + with ra._lock: + ra._gateway_queues.pop(session_id, None) + + with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}): + with patch("api.gateway_chat.gateway_supports_approval", return_value=False), \ + patch("urllib.request.urlopen", side_effect=_make_legacy_gateway_urlopen(approval_payload, clear_gateway_queue)), \ + 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=session_id, + msg_text="do something risky", + model="test", + workspace="/tmp", + stream_id=stream_id, + ) + + payloads = _drain_queue(subscriber) + assert payloads, "Expected approval SSE notifications from mirror and teardown" + assert payloads[0]["pending"]["_gateway_mirror"] is True + assert payloads[0]["pending_count"] == 1 + assert payloads[-1]["pending"] is None + assert payloads[-1]["pending_count"] == 0 + with ra._lock: + assert session_id not in ra._pending + finally: + ra._approval_sse_unsubscribe(session_id, subscriber) + with STREAMS_LOCK: + STREAMS.pop(stream_id, None) + with ra._lock: + ra._pending.pop(session_id, None) + ra._gateway_queues.pop(session_id, None) + + +def test_legacy_teardown_preserves_live_gateway_head_mirror(): + """A live gateway head must still be mirrored after legacy teardown runs.""" + from types import SimpleNamespace + from api import route_approvals as ra + from api.config import STREAMS, STREAMS_LOCK + from api.gateway_chat import _run_gateway_chat_streaming + + session_id = "sess-legacy-teardown-live" + stream_id = "sid-legacy-teardown-live" + approval_data = { + "command": "rm -rf /tmp/test", + "description": "Delete temporary files", + "pattern_key": "dangerous_command", + "pattern_keys": ["dangerous_command"], + "approval_id": "appr-legacy-teardown-live", + "choices": ["once", "session", "always", "deny"], + "run_id": "run-legacy-teardown-live", + } + approval_payload = json.dumps(approval_data) + + subscriber = ra._approval_sse_subscribe(session_id) + q = MagicMock() + q.put_nowait = lambda item: None + + 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.pending_user_source = None + + try: + with ra._lock: + ra._pending.pop(session_id, None) + ra._gateway_queues[session_id] = [SimpleNamespace(data=dict(approval_data))] + with STREAMS_LOCK: + STREAMS[stream_id] = q + + with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}): + with patch("api.gateway_chat.gateway_supports_approval", return_value=False), \ + patch("urllib.request.urlopen", side_effect=_make_legacy_gateway_urlopen(approval_payload)), \ + 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=session_id, + msg_text="do something risky", + model="test", + workspace="/tmp", + stream_id=stream_id, + ) + + payloads = _drain_queue(subscriber) + assert payloads, "Expected mirrored approval notifications" + assert payloads[-1]["pending"]["_gateway_mirror"] is True + assert payloads[-1]["pending_count"] == 1 + with ra._lock: + pending = ra._pending.get(session_id) + assert isinstance(pending, list) + assert len(pending) == 1 + assert pending[0]["approval_id"] == approval_data["approval_id"] + assert pending[0]["_gateway_mirror"] is True + finally: + ra._approval_sse_unsubscribe(session_id, subscriber) + with STREAMS_LOCK: + STREAMS.pop(stream_id, None) + with ra._lock: + ra._pending.pop(session_id, None) + ra._gateway_queues.pop(session_id, None) + + +def test_legacy_teardown_preserves_local_pending_entry(): + """Legacy gateway teardown must not remove non-gateway pending approvals.""" + from api import route_approvals as ra + from api.config import STREAMS, STREAMS_LOCK + from api.gateway_chat import _run_gateway_chat_streaming + + session_id = "sess-legacy-teardown-local" + stream_id = "sid-legacy-teardown-local" + local_pending = { + "command": "echo local", + "description": "Local approval", + "pattern_key": "local_command", + "pattern_keys": ["local_command"], + "approval_id": "appr-legacy-local", + } + approval_data = { + "command": "rm -rf /tmp/test", + "description": "Delete temporary files", + "pattern_key": "dangerous_command", + "pattern_keys": ["dangerous_command"], + "approval_id": "appr-legacy-teardown-local", + "choices": ["once", "session", "always", "deny"], + "run_id": "run-legacy-teardown-local", + } + approval_payload = json.dumps(approval_data) + + subscriber = ra._approval_sse_subscribe(session_id) + q = MagicMock() + q.put_nowait = lambda item: None + + 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.pending_user_source = None + + try: + with ra._lock: + ra._gateway_queues.pop(session_id, None) + ra._pending[session_id] = [dict(local_pending)] + with STREAMS_LOCK: + STREAMS[stream_id] = q + + with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}): + with patch("api.gateway_chat.gateway_supports_approval", return_value=False), \ + patch("urllib.request.urlopen", side_effect=_make_legacy_gateway_urlopen(approval_payload)), \ + 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=session_id, + msg_text="do something risky", + model="test", + workspace="/tmp", + stream_id=stream_id, + ) + + payloads = _drain_queue(subscriber) + assert payloads, "Expected local pending approval notifications" + assert any( + payload["pending"] and payload["pending"]["approval_id"] == local_pending["approval_id"] + for payload in payloads + ) + with ra._lock: + pending = ra._pending.get(session_id) + assert isinstance(pending, list) + assert pending[0]["approval_id"] == local_pending["approval_id"] + assert pending[0].get(ra._GATEWAY_MIRROR_FLAG) is not True + finally: + ra._approval_sse_unsubscribe(session_id, subscriber) + with STREAMS_LOCK: + STREAMS.pop(stream_id, None) + with ra._lock: + ra._pending.pop(session_id, None) + ra._gateway_queues.pop(session_id, None) + + def test_mirrored_run_id_survives_active_stream_loss(): """A mirrored gateway approval must still relay after active_stream_id is lost.""" import io From beb883a159bd5185b4d13b7e7b9e986f91c662ae Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Sun, 28 Jun 2026 21:57:12 +0000 Subject: [PATCH 2/2] docs(changelog): clear stale gateway approval mirrors on teardown (#5147) --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index a4cdf0f3f..ff467b8f0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,6 +22,7 @@ ### Fixed +- **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) - **Scheduled jobs created under a selected profile no longer trip the provider/model drift guard.** The cron profile selector lets you choose which profile a job runs under, but `_handle_cron_create()` snapshotted unpinned provider/model state *before* that selected runtime profile was applied — so the agent later compared runtime resolution against a stale create-time snapshot and could refuse to run a job whose intended runtime profile never changed. Snapshot recompute now runs inside the selected profile's context (validated at parity with cron update; only provider/model name strings are stored, no cross-profile state), while the default unpinned-create path is unchanged. Thanks @rodboev. (#5131, fixes #5130) - **A live assistant turn no longer briefly blanks out and reappears on a transient stream hiccup.** When the chat `EventSource` fired a transient `error`, `attachLiveStream` made a single 1.5s reconnect probe; if `/api/chat/stream/status` didn't report `active`/`replay_available` at that one instant it fell straight through to the terminal error path — clearing inflight state, nulling the active stream id, pushing a "Connection interrupted" message and re-rendering, which wiped the live message DOM even while the backend was still producing tokens (or the run-journal replay file was a beat from ready). The reconnect probes are now staged (retried on a short schedule) before the stream is declared dead, so a momentary blip recovers in place instead of blanking the turn. Thanks @allenliang2022. (#5122)