diff --git a/api/gateway_chat.py b/api/gateway_chat.py index db157b68..f4e5158a 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -507,8 +507,8 @@ def _run_gateway_runs_api_streaming( if not approval_data.get("approval_id"): approval_data["approval_id"] = f"gwrun:{run_id}:{uuid.uuid4().hex}" from api.route_approvals import submit_gateway_pending_mirror - submit_gateway_pending_mirror(session_id, approval_data) - put_gateway_event("approval", approval_data) + head, total = submit_gateway_pending_mirror(session_id, approval_data) + put_gateway_event("approval", {**(head or approval_data), "pending_count": total}) sse_event = "message" continue if payload_event in {"tool.started", "tool.completed", "reasoning.available"}: @@ -989,7 +989,8 @@ def _run_gateway_chat_streaming( _STREAM_RUN_IDS[stream_id] = _approval_run_id try: from api.route_approvals import submit_gateway_pending_mirror - submit_gateway_pending_mirror(session_id, approval_data) + head, total = submit_gateway_pending_mirror(session_id, approval_data) + approval_data = {**(head or approval_data), "pending_count": total} except Exception: logger.debug("submit_gateway_pending_mirror failed", exc_info=True) put_gateway_event("approval", approval_data) @@ -1259,6 +1260,13 @@ def _run_gateway_chat_streaming( "hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.", }) finally: + mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip() + if mapped_run_id: + try: + from api.route_approvals import retire_gateway_pending_mirror + retire_gateway_pending_mirror(session_id, run_id=mapped_run_id) + except Exception: + logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True) if s is not None: try: with _get_session_agent_lock(session_id): diff --git a/api/route_approvals.py b/api/route_approvals.py index f524e852..41bb6e5c 100644 --- a/api/route_approvals.py +++ b/api/route_approvals.py @@ -47,7 +47,6 @@ _approval_sse_subscribers: dict[str, list[queue.Queue]] = {} _GATEWAY_MIRROR_FLAG = "_gateway_mirror" _GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token" _GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token" -_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY = "_webui_local_head" def _approval_sse_subscribe(session_id: str) -> queue.Queue: @@ -116,8 +115,6 @@ def _gateway_mirror_entry_token(entry) -> str | None: """ data = getattr(entry, "data", None) if isinstance(data, dict): - if data.get(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY): - return None token = data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) if not token: token = uuid.uuid4().hex @@ -150,8 +147,31 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No live_head_entry = live_gateway_queue[0] if live_gateway_queue else None live_head_data = getattr(live_head_entry, "data", None) or {} - live_token = _gateway_mirror_entry_token(live_head_entry) if live_head_entry and live_head_data else None + has_no_run_mirror = any( + _is_gateway_mirror_entry(entry) + and not str(entry.get("run_id") or "").strip() + for entry in queue_list + ) live_run_id = str(live_head_data.get("run_id") or "").strip() + live_has_token = bool(live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY)) + live_local_tokens: set[str] = set() + for live_entry in live_gateway_queue: + live_data = getattr(live_entry, "data", None) or {} + if str(live_data.get("run_id") or "").strip(): + continue + live_entry_token = str(live_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) or "").strip() + if live_entry_token or not has_no_run_mirror: + live_entry_token = _gateway_mirror_entry_token(live_entry) or "" + if live_entry_token: + live_local_tokens.add(live_entry_token) + if not str(live_data.get("approval_id") or "").strip(): + live_data["approval_id"] = f"gwlocal:{live_entry_token}" + live_token = ( + _gateway_mirror_entry_token(live_head_entry) + if live_head_entry and live_head_data + and (live_run_id or live_has_token or not has_no_run_mirror) + else None + ) if live_token and live_run_id and not str(live_head_data.get("approval_id") or "").strip(): live_head_data["approval_id"] = f"gwrun:{live_run_id}:{live_token}" live_approval_id = str(live_head_data.get("approval_id") or "").strip() @@ -206,6 +226,10 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No live_mirror_present = True continue + if entry_token and entry_token in live_local_tokens: + rebuilt.append(entry) + continue + if not live_token: if entry_token: changed = True @@ -348,34 +372,56 @@ def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str return None -def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: +def submit_gateway_pending_mirror(session_key: str, approval: dict) -> tuple[dict | None, int]: """Mirror the live gateway head into WebUI polling state under a typed tag.""" with _lock: run_id = str(approval.get("run_id") or "").strip() approval_id = str(approval.get("approval_id") or "").strip() live_gateway_queue = _gateway_queues.get(session_key) or [] - live_head_entry = live_gateway_queue[0] if live_gateway_queue else None - live_head_data = getattr(live_head_entry, "data", None) or {} - matching_no_run_live_head = ( - not run_id - and live_head_data - and str(live_head_data.get("command") or "") == str(approval.get("command") or "") - and str(live_head_data.get("description") or "") == str(approval.get("description") or "") - and str(live_head_data.get("pattern_key") or "") == str(approval.get("pattern_key") or "") - and tuple(str(k) for k in (live_head_data.get("pattern_keys") or [])) - == tuple(str(k) for k in (approval.get("pattern_keys") or [])) - ) - if matching_no_run_live_head and not live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY): - live_head_data.pop(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY, None) - live_head_data[_GATEWAY_ENTRY_DATA_TOKEN_KEY] = uuid.uuid4().hex - elif not run_id and live_head_data: - live_head_data[_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY] = True - if not run_id and not approval_id: - approval_id = uuid.uuid4().hex - approval["approval_id"] = approval_id - if matching_no_run_live_head and approval_id and not str(live_head_data.get("approval_id") or "").strip(): - live_head_data["approval_id"] = approval_id + exact_local_entry = next( + ( + entry for entry in live_gateway_queue + if getattr(entry, "data", None) is approval + ), + None, + ) if not run_id else None + if exact_local_entry is None and not run_id and approval_id: + exact_local_entry = next( + ( + entry for entry in live_gateway_queue + if str(((getattr(entry, "data", None) or {}).get("approval_id") or "")).strip() == approval_id + ), + None, + ) + if exact_local_entry is not None: + mirror_entries = _normalize_pending_queue_locked(session_key) + entries_to_mirror = [live_gateway_queue[0]] if live_gateway_queue else [] + if exact_local_entry not in entries_to_mirror: + entries_to_mirror.append(exact_local_entry) + for entry in entries_to_mirror: + local_data = entry.data + token = _gateway_mirror_entry_token(entry) + entry_approval_id = str(local_data.get("approval_id") or "").strip() + if entry is exact_local_entry: + entry_approval_id = approval_id or entry_approval_id or f"gwlocal:{token}" + approval_id = entry_approval_id + approval["approval_id"] = entry_approval_id + elif not entry_approval_id: + entry_approval_id = f"gwlocal:{token}" + local_data["approval_id"] = entry_approval_id + if not any( + _is_gateway_mirror_entry(mirror) + and str(mirror.get(_GATEWAY_MIRROR_TOKEN) or "") == token + for mirror in mirror_entries + ): + mirror_entry = dict(local_data) + mirror_entry["approval_id"] = entry_approval_id + mirror_entry[_GATEWAY_MIRROR_FLAG] = True + mirror_entry[_GATEWAY_MIRROR_TOKEN] = token + mirror_entries.append(mirror_entry) if run_id: + live_head_entry = live_gateway_queue[0] if live_gateway_queue else None + live_head_data = getattr(live_head_entry, "data", None) or {} live_head_run_id = str(live_head_data.get("run_id") or "").strip() live_head_approval_id = str(live_head_data.get("approval_id") or "").strip() live_token = ( @@ -410,15 +456,10 @@ def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: mirror_entry[_GATEWAY_MIRROR_FLAG] = True if not _gateway_pending_mirror_locked(session_key, approval_id=approval_id, run_id=run_id): _normalize_pending_queue_locked(session_key).append(mirror_entry) - head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) - approval_id = str(approval.get("approval_id") or "").strip() - if run_id and approval_id: - mirror_entry = _gateway_pending_mirror_locked( - session_key, approval_id=approval_id, run_id=run_id - ) - if mirror_entry: - approval["approval_id"] = str(mirror_entry.get("approval_id") or approval_id).strip() - elif approval_id: + elif not exact_local_entry: + if not approval_id: + approval_id = uuid.uuid4().hex + approval["approval_id"] = approval_id queue = _pending.get(session_key) entries = queue if isinstance(queue, list) else [queue] if queue else [] no_run_mirror = next( @@ -437,8 +478,39 @@ def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: mirror_entry["approval_id"] = approval_id mirror_entry[_GATEWAY_MIRROR_FLAG] = True _normalize_pending_queue_locked(session_key).append(mirror_entry) + head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) _approval_sse_notify_locked(session_key, head, total) publish_session_list_changed("attention_pending") + return (dict(head) if head else None), total + + +def resolve_gateway_pending_local( + session_key: str, approval_id: str, choice: str, reason: str | None = None +) -> tuple[int, dict | None, int]: + """Resolve the exact parked local entry bound to an approval mirror.""" + target = None + with _lock: + approval_id = str(approval_id or "").strip() + gateway_queue = _gateway_queues.get(session_key) or [] + for index, entry in enumerate(gateway_queue): + data = getattr(entry, "data", None) or {} + if str(data.get("approval_id") or "").strip() == approval_id: + target = gateway_queue.pop(index) + break + if gateway_queue: + _gateway_queues[session_key] = gateway_queue + else: + _gateway_queues.pop(session_key, None) + head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) + _approval_sse_notify_locked(session_key, head, total) + if target is None: + return 0, head, total + target.result = choice + if reason: + target.reason = reason + target.event.set() + publish_session_list_changed("attention_resolved") + return 1, head, total def submit_pending(session_key: str, approval: dict) -> None: diff --git a/api/routes.py b/api/routes.py index 016df3a6..76dfa066 100644 --- a/api/routes.py +++ b/api/routes.py @@ -9720,6 +9720,7 @@ from api.route_approvals import ( # noqa: F401 — re-exports for backward comp gateway_pending_mirror, retire_gateway_pending_mirror, reconcile_gateway_pending_mirror_locked, + resolve_gateway_pending_local, submit_gateway_pending_mirror, submit_pending, ) @@ -9746,12 +9747,40 @@ except ImportError: def _session_attention_summary(session_id: str) -> dict | None: """Return sidebar attention metadata for pending approval/clarify work.""" + active_run_id = "" + try: + from api.gateway_chat import _STREAM_RUN_IDS + + session = get_session(session_id) + active_stream_id = getattr(session, "active_stream_id", None) if session is not None else None + if active_stream_id: + active_run_id = str(_STREAM_RUN_IDS.get(active_stream_id) or "").strip() + except Exception: + active_run_id = "" + approval_count = 0 with _lock: reconcile_gateway_pending_mirror_locked(session_id) queue_list = _pending.get(session_id) if isinstance(queue_list, list): - approval_count = len(queue_list) + gateway_queue = _gateway_queues.get(session_id) or [] + live_gateway_run_ids = { + str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() + for entry in gateway_queue + if str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() + } + approval_count = sum( + 1 + for entry in queue_list + if not ( + isinstance(entry, dict) + and entry.get(_GATEWAY_MIRROR_FLAG) + and str(entry.get("run_id") or "").strip() + and not str(entry.get(_GATEWAY_MIRROR_TOKEN) or "").strip() + and str(entry.get("run_id") or "").strip() not in live_gateway_run_ids + and str(entry.get("run_id") or "").strip() != active_run_id + ) + ) elif queue_list: approval_count = 1 if approval_count > 0: @@ -23802,6 +23831,7 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st pending = None found_target = False gateway_keys = [] + local_gateway_approval_id = "" gateway_head_matches_target = False with _lock: reconcile_gateway_pending_mirror_locked(sid) @@ -23874,20 +23904,28 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st gw_data = getattr(gw_entry, "data", None) or {} gw_approval_id = str(gw_data.get("approval_id") or "").strip() gw_run_id = str(gw_data.get("run_id") or "").strip() - if run_id: - gateway_head_matches_target = gw_approval_id == approval_id and gw_run_id == run_id - else: - gateway_head_matches_target = gw_approval_id == approval_id + gateway_head_matches_target = bool( + run_id and gw_approval_id == approval_id and gw_run_id == run_id + ) + if gw_approval_id == approval_id and (not run_id or gw_run_id == run_id): + local_gateway_approval_id = approval_id + elif not run_id and found_target and pending: + pending_token = str(pending.get(_GATEWAY_MIRROR_TOKEN) or "").strip() + gateway_token = str(gw_data.get("_webui_mirror_token") or "").strip() + if pending_token and pending_token == gateway_token: + gw_data["approval_id"] = approval_id + local_gateway_approval_id = approval_id # Notify SSE subscribers of the new head (or empty state) so the UI # surfaces any trailing approvals that were queued behind this one # without waiting for the next submit_pending. Without this, a parallel # tool-call scenario (#527) would leave the second approval invisible # in the SSE path until the next event ever fired (the agent thread # would be parked indefinitely from the user's perspective). - if isinstance(_pending.get(sid), list) and _pending[sid]: - _approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid])) - else: - _approval_sse_notify_locked(sid, None, 0) + if not local_gateway_approval_id: + if isinstance(_pending.get(sid), list) and _pending[sid]: + _approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid])) + else: + _approval_sse_notify_locked(sid, None, 0) # Collect keys from both _pending and _gateway_queues keys_from_pending = pending.get("pattern_keys") or [pending.get("pattern_key", "")] if pending else [] @@ -23909,21 +23947,18 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st # This is the primary signal when streaming is active — the agent # thread is parked in entry.event.wait() and needs to be woken up. gateway_resolved = 0 - should_resolve_gateway = ( - not approval_id - or ( - found_target - and ( - not run_id - or gateway_head_matches_target - ) + local_gateway_resolved = 0 + if approval_id and found_target and not run_id and local_gateway_approval_id: + local_gateway_resolved, _head, _total = resolve_gateway_pending_local( + sid, local_gateway_approval_id, choice ) - ) - if should_resolve_gateway: + elif approval_id and found_target and run_id and gateway_head_matches_target: + gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0 + elif not approval_id: gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0 # Keep the historical no-id response path truthy for old clients/tests while # making stale explicit ids bounded as not-active for Slice 3b. - resolved = bool(pending) or bool(gateway_resolved) or not bool(approval_id) + resolved = bool(pending) or bool(gateway_resolved) or bool(local_gateway_resolved) or not bool(approval_id) if resolved: publish_session_list_changed("attention_resolved") return resolved diff --git a/api/streaming.py b/api/streaming.py index 2528baf4..bda92823 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -7826,7 +7826,8 @@ def _run_agent_streaming( def _approval_notify_cb(approval_data): if _submit_pending_for_polling is not None: try: - _submit_pending_for_polling(session_id, approval_data) + head, total = _submit_pending_for_polling(session_id, approval_data) + approval_data = {**(head or approval_data), "pending_count": total} except Exception: logger.warning("Failed to mirror approval into WebUI polling state", exc_info=True) put('approval', approval_data) @@ -8241,20 +8242,9 @@ def _run_agent_streaming( if _has_blocking_approval(session_id): p = None with _approval_lock: - _reconcile_gateway_pending_mirror_locked(session_id) - queue = _approval_pending.get(session_id) - if isinstance(queue, list): - p = dict(queue[0]) if queue else None - elif queue: - p = dict(queue) - if p is None: - gw_queue = _approval_gateway_queues.get(session_id) or [] - if gw_queue: - raw = getattr(gw_queue[0], 'data', None) or {} - if raw: - p = dict(raw) - else: - logger.warning("Gateway queue entry for %s has no .data attribute", session_id) + p, pending_count, _changed = _reconcile_gateway_pending_mirror_locked(session_id) + if p: + p = {**p, "pending_count": pending_count} if p: put('approval', p) except ImportError: diff --git a/static/messages.js b/static/messages.js index ea19f451..c45ff129 100644 --- a/static/messages.js +++ b/static/messages.js @@ -5589,7 +5589,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){ source.addEventListener('approval',e=>{ const d=JSON.parse(e.data); _applyToAnchor('approval',d,e); - showApprovalForSession(activeSid, d, 1); + showApprovalForSession(activeSid, d, d.pending_count || 1); playAttentionSound(_attentionSoundKey(activeSid,'approval',1)); sendBrowserNotification('Approval required',d.description||'Tool approval needed',{sid:activeSid}); }); diff --git a/tests/test_approval_queue.py b/tests/test_approval_queue.py index 2ca10f58..4de98b72 100644 --- a/tests/test_approval_queue.py +++ b/tests/test_approval_queue.py @@ -7,6 +7,8 @@ approval_id so /api/approval/respond can target a specific entry. import json import pathlib import re +import shutil +import subprocess import sys REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve() @@ -108,6 +110,39 @@ def test_polling_passes_count_to_show(): "Poll loop must pass data.pending_count through showApprovalForSession" +def test_chat_stream_approval_listener_renders_received_count(): + """A chat-stream approval event with count two renders the head and count.""" + node = shutil.which("node") + if node is None: + import pytest + pytest.skip("node is required for the approval listener harness") + start = MESSAGES_JS.index("source.addEventListener('approval',e=>{") + end_marker = "\n });" + end = MESSAGES_JS.index(end_marker, start) + len(end_marker) + listener = MESSAGES_JS[start:end] + script = f""" + const rendered = []; + let activeSid = 'sid-browser'; + const source = {{ listeners: {{}}, addEventListener(name, cb) {{ this.listeners[name] = cb; }} }}; + function _applyToAnchor() {{}} + function showApprovalForSession(sid, data, count) {{ rendered.push({{sid, data, count}}); }} + function playAttentionSound() {{}} + function _attentionSoundKey() {{ return 'approval'; }} + function sendBrowserNotification() {{}} + {listener} + source.listeners.approval({{ data: JSON.stringify({{command: 'head', approval_id: 'a', pending_count: 2}}) }}); + process.stdout.write(JSON.stringify(rendered)); + """ + result = subprocess.run([node, "-e", script], capture_output=True, text=True, check=False) + assert result.returncode == 0, result.stderr + rendered = json.loads(result.stdout) + assert rendered == [{ + "sid": "sid-browser", + "data": {"command": "head", "approval_id": "a", "pending_count": 2}, + "count": 2, + }] + + # --------------------------------------------------------------------------- # HTML: counter element present # --------------------------------------------------------------------------- diff --git a/tests/test_approval_sse.py b/tests/test_approval_sse.py index f10ea608..dfc4dfc6 100644 --- a/tests/test_approval_sse.py +++ b/tests/test_approval_sse.py @@ -113,8 +113,10 @@ class TestSSEStaticAnalysis: cb_start = streaming_src.find("def _approval_notify_cb(approval_data):") cb_end = streaming_src.find("_reg_notify(session_id, _approval_notify_cb)", cb_start) cb_body = streaming_src[cb_start:cb_end] - assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \ + assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \ "_approval_notify_cb must mirror approval data into polling state before SSE" + assert '"pending_count": total' in cb_body, \ + "_approval_notify_cb must emit the reconciled pending count" assert "put('approval', approval_data)" in cb_body, \ "_approval_notify_cb must still emit the approval SSE event" diff --git a/tests/test_approval_unblock.py b/tests/test_approval_unblock.py index d9040d95..757ab5e6 100644 --- a/tests/test_approval_unblock.py +++ b/tests/test_approval_unblock.py @@ -232,11 +232,57 @@ class TestApprovalModuleExports: assert cb_start != -1, "_approval_notify_cb must exist" cb_end = STREAMING_SRC.find("_reg_notify(session_id, _approval_notify_cb)", cb_start) cb_body = STREAMING_SRC[cb_start:cb_end] - assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \ + assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \ "approval notify callback must mirror approval data into polling state" + assert '"pending_count": total' in cb_body, \ + "approval notify callback must publish the reconciled pending count" assert "put('approval', approval_data)" in cb_body, \ "approval notify callback must still push the SSE event" + def test_reversed_local_callbacks_publish_all_parked_entries(self): + """The local producer must count every exact parked entry, not only its head.""" + import api.route_approvals as ra + from api import routes + + sid = f"unit-reversed-local-{uuid.uuid4().hex[:8]}" + entry_a = _ApprovalEntry({ + "command": "same-command", + "description": "same-description", + "pattern_key": "same-pattern", + "pattern_keys": ["same-pattern"], + }) + entry_b = _ApprovalEntry(dict(entry_a.data)) + events = [] + with _lock: + _gateway_queues[sid] = [entry_a, entry_b] + ra._pending.pop(sid, None) + + def local_producer(approval_data): + head, total = ra.submit_gateway_pending_mirror(sid, approval_data) + events.append((head, total)) + + try: + local_producer(entry_b.data) + local_producer(entry_a.data) + + assert [total for _head, total in events] == [2, 2] + assert all(head["command"] == "same-command" for head, _total in events) + head_id = events[0][0]["approval_id"] + assert head_id == events[1][0]["approval_id"] + assert routes._resolve_approval_legacy(sid, head_id, "once") is True + assert entry_a.event.is_set() + assert not entry_b.event.is_set() + with _lock: + successor = ra._pending[sid][0] + assert successor["approval_id"] != head_id + with _lock: + assert ra.reconcile_gateway_pending_mirror_locked(sid)[1] == 1 + finally: + with _lock: + ra._pending.pop(sid, None) + _gateway_queues.pop(sid, None) + ra._pending.pop(sid, None) + # ── HTTP regression tests (test server, port 8788) ─────────────────────────── diff --git a/tests/test_gateway_approval_legacy_path.py b/tests/test_gateway_approval_legacy_path.py index a170bb72..d9a1c7d1 100644 --- a/tests/test_gateway_approval_legacy_path.py +++ b/tests/test_gateway_approval_legacy_path.py @@ -336,6 +336,19 @@ def test_legacy_approval_records_run_id_for_response_relay(): # (connection still open, run parked) — this test's stream already ran to # completion, which cleared active_stream_id and popped the mapping. _STREAM_RUN_IDS[stream_id] = "run-legacy-1" + from types import SimpleNamespace + from api import route_approvals as ra + with ra._lock: + ra._gateway_queues["sess-legacy-runid"] = [SimpleNamespace(data={ + "command": "rm -rf /tmp/test", + "description": "Delete temporary files", + "pattern_key": "dangerous_command", + "pattern_keys": ["dangerous_command"], + "approval_id": "appr-legacy-runid", + "run_id": "run-legacy-1", + })] + live_entry = ra._gateway_queues["sess-legacy-runid"][0] + ra.submit_gateway_pending_mirror("sess-legacy-runid", live_entry.data) relay_session = MagicMock() relay_session.active_stream_id = stream_id captured = {} @@ -447,8 +460,8 @@ def test_legacy_teardown_clears_stale_gateway_mirror_and_notifies_empty_state(): 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.""" +def test_legacy_teardown_retires_live_gateway_head_mirror(): + """A mapped legacy run retires its live gateway mirror during outer teardown.""" from types import SimpleNamespace from api import route_approvals as ra from api.config import STREAMS, STREAMS_LOCK @@ -507,14 +520,11 @@ def test_legacy_teardown_preserves_live_gateway_head_mirror(): payloads = _drain_queue(subscriber) assert payloads, "Expected mirrored approval notifications" - assert payloads[-1]["pending"]["_gateway_mirror"] is True - assert payloads[-1]["pending_count"] == 1 + assert payloads[-1]["pending"] is None + assert payloads[-1]["pending_count"] == 0 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 + assert pending is None finally: ra._approval_sse_unsubscribe(session_id, subscriber) with STREAMS_LOCK: diff --git a/tests/test_session_attention_badges.py b/tests/test_session_attention_badges.py index 8955d000..91bf7c13 100644 --- a/tests/test_session_attention_badges.py +++ b/tests/test_session_attention_badges.py @@ -115,6 +115,25 @@ def test_attention_summary_prefers_pending_approvals_over_clarify_questions(): _clear_attention_state(sid) +def test_attention_summary_ignores_stale_gateway_mirror(): + sid = "attention-stale-gateway-session" + _clear_attention_state(sid) + try: + with routes._lock: + routes._pending[sid] = [{ + "_gateway_mirror": True, + "approval_id": "appr-stale-gateway", + "run_id": "run-stale-gateway", + "command": "rm -rf /tmp/stale", + "description": "Stale approval", + }] + routes._gateway_queues[sid] = [] + + assert routes._session_attention_summary(sid) is None + finally: + _clear_attention_state(sid) + + def test_attention_summary_reports_clarify_when_no_approval_is_pending(): sid = "attention-clarify-session" _clear_attention_state(sid)