Merge pull request #5150 from nesquena/stage/5147-approval

fix(gateway): clear stale approval mirrors on stream teardown (#5147)

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)

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)

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