diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 261691d4..1f51fbc3 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -347,6 +347,8 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None: run_id = str(payload.get("run_id") or "").strip() raw_approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip() approval_id = raw_approval_id + if not approval_id: + approval_id = uuid.uuid4().hex risk = str(payload.get("risk_level") or "high").strip() choices = payload.get("choices") if isinstance(payload.get("choices"), list) else [] allow_permanent = payload.get("allow_permanent") @@ -703,14 +705,10 @@ def _clear_gateway_pending_state(session: Any, stream_id: str) -> None: 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, + retire_gateway_pending_mirror, ) - with _approval_lock: - head, total, _ = reconcile_gateway_pending_mirror_locked(session_id) - _approval_sse_notify_locked(session_id, head, total) + retire_gateway_pending_mirror(session_id) except Exception: logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True) diff --git a/api/route_approvals.py b/api/route_approvals.py index c3ba6a09..424e099f 100644 --- a/api/route_approvals.py +++ b/api/route_approvals.py @@ -46,6 +46,7 @@ except ImportError: _approval_sse_subscribers: dict[str, list[queue.Queue]] = {} _GATEWAY_MIRROR_FLAG = "_gateway_mirror" _GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token" +_GATEWAY_MIRROR_RETAINED = "_gateway_mirror_retained" _GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token" _GATEWAY_AGENT_IDENTITY_V1 = "_gateway_agent_identity_v1" @@ -219,6 +220,10 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No changed = True continue + if entry.get(_GATEWAY_MIRROR_RETAINED): + rebuilt.append(entry) + continue + if matches_live_head and not live_mirror_present: if entry_token != live_token: entry[_GATEWAY_MIRROR_TOKEN] = live_token @@ -334,7 +339,11 @@ def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_i entry for entry in entries if _is_gateway_mirror_entry(entry) and str(entry.get("run_id") or "").strip() == normalized_run_id - ] if normalized_run_id else [] + ] if normalized_run_id else [ + entry for entry in entries + if _is_gateway_mirror_entry(entry) + and not str(entry.get("run_id") or "").strip() + ] if normalized_run_id: retained_gateway_queue = [] for entry in gateway_queue: @@ -344,10 +353,11 @@ def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_i continue retained_gateway_queue.append(entry) if not retired and not gateway_queue_changed: - if approval_id: - head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) - _approval_sse_notify_locked(session_key, head, total) - return False + head, total, changed = reconcile_gateway_pending_mirror_locked(session_key) + _approval_sse_notify_locked(session_key, head, total) + if changed: + publish_session_list_changed("attention_resolved") + return changed for match in retired: entries.remove(match) if normalized_run_id and not approval_id: @@ -546,6 +556,7 @@ def resolve_gateway_pending_local_no_run_mirror( target = gateway_queue.pop(index) break if target is None: + matched_mirror[_GATEWAY_MIRROR_RETAINED] = True return True, 0, entries[0] if entries else None, len(entries) if gateway_queue: diff --git a/api/streaming.py b/api/streaming.py index bda92823..f4c8a143 100644 --- a/api/streaming.py +++ b/api/streaming.py @@ -7808,14 +7808,10 @@ def _run_agent_streaming( try: from api.route_approvals import ( submit_gateway_pending_mirror as _submit_pending_for_polling, - reconcile_gateway_pending_mirror_locked as _reconcile_gateway_pending_mirror_locked, - _approval_sse_notify_locked as _approval_sse_notify_locked, - _lock as _approval_lock, + retire_gateway_pending_mirror as _retire_gateway_pending_mirror, ) def _cleanup_gateway_pending_mirror(): - with _approval_lock: - head, total, _changed = _reconcile_gateway_pending_mirror_locked(session_id) - _approval_sse_notify_locked(session_id, head, total) + _retire_gateway_pending_mirror(session_id) except ImportError: _submit_pending_for_polling = None _cleanup_gateway_pending_mirror = None diff --git a/tests/test_approval_unblock.py b/tests/test_approval_unblock.py index 05579f13..4ca70897 100644 --- a/tests/test_approval_unblock.py +++ b/tests/test_approval_unblock.py @@ -638,11 +638,11 @@ class TestApprovalHTTPEndpoints: r._gateway_queues.pop(sid, None) _STREAM_RUN_IDS.pop(stream_id, None) - def test_gateway_no_run_mirror_stays_visible_when_exact_local_producer_is_gone(self, monkeypatch): - """A tokenized no-run mirror stays live when its parked producer already vanished.""" + def test_gateway_no_run_mirror_survives_pending_and_attention_reads_until_explicit_teardown(self, monkeypatch): + """A missing-producer 409 stays visible until a real teardown retires it.""" from api import routes as r from api import route_approvals as ra - from api.gateway_chat import _STREAM_RUN_IDS + from api.gateway_chat import _STREAM_RUN_IDS, _cleanup_gateway_pending_mirror sid = f"http-gateway-race-{uuid.uuid4().hex[:8]}" stream_id = f"stream-race-{uuid.uuid4().hex[:8]}" @@ -680,13 +680,76 @@ class TestApprovalHTTPEndpoints: "error": r._GATEWAY_APPROVAL_RELAY_UNAVAILABLE, } assert not entry.event.is_set() - with _lock: - pending_queue = r._pending.get(sid) - assert isinstance(pending_queue, list) - assert len(pending_queue) == 1 - assert pending_queue[0]["approval_id"] == approval_id + pending = r._handle_approval_pending( + object(), urllib.parse.urlparse( + f"/api/approval/pending?session_id={urllib.parse.quote(sid)}" + ) + ) + assert pending["pending"]["approval_id"] == approval_id + assert pending["pending_count"] == 1 + assert r._session_attention_summary(sid) == { + "kind": "approval", "count": 1, "severity": "critical" + } + _cleanup_gateway_pending_mirror(sid) + pending = r._handle_approval_pending( + object(), urllib.parse.urlparse( + f"/api/approval/pending?session_id={urllib.parse.quote(sid)}" + ) + ) + assert pending == {"pending": None, "pending_count": 0} + assert r._session_attention_summary(sid) is None finally: with _lock: r._pending.pop(sid, None) r._gateway_queues.pop(sid, None) _STREAM_RUN_IDS.pop(stream_id, None) + + def test_gateway_no_run_non_head_response_resolves_only_exact_producer(self, monkeypatch): + """An exact non-head response wakes only its matching producer.""" + from api import routes as r + from api import route_approvals as ra + + sid = f"http-gateway-non-head-{uuid.uuid4().hex[:8]}" + approval_a = {"command": "rm -rf /tmp/a", "description": "a"} + approval_b = {"command": "rm -rf /tmp/b", "description": "b"} + entry_a = _ApprovalEntry(approval_a) + entry_b = _ApprovalEntry(approval_b) + captured = {} + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(r, "j", fake_j) + monkeypatch.setattr(r, "get_session", lambda _sid: SimpleNamespace(active_stream_id=None)) + monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway") + with _lock: + r._pending.pop(sid, None) + r._gateway_queues[sid] = [entry_a, entry_b] + try: + ra.submit_gateway_pending_mirror(sid, approval_a) + ra.submit_gateway_pending_mirror(sid, approval_b) + with _lock: + approval_b_id = next( + item["approval_id"] for item in r._pending[sid] + if item["command"] == approval_b["command"] + ) + r._handle_approval_respond( + object(), {"session_id": sid, "choice": "once", "approval_id": approval_b_id} + ) + assert captured["status"] == 200 + assert entry_b.event.is_set() + assert entry_b.result == "once" + assert not entry_a.event.is_set() + pending = r._handle_approval_pending( + object(), urllib.parse.urlparse( + f"/api/approval/pending?session_id={urllib.parse.quote(sid)}" + ) + ) + assert pending["pending"]["command"] == approval_a["command"] + assert pending["pending_count"] == 1 + finally: + with _lock: + r._pending.pop(sid, None) + r._gateway_queues.pop(sid, None) diff --git a/tests/test_gateway_approval_runs_api.py b/tests/test_gateway_approval_runs_api.py index b988a9d4..58f8638a 100644 --- a/tests/test_gateway_approval_runs_api.py +++ b/tests/test_gateway_approval_runs_api.py @@ -432,13 +432,19 @@ def test_gateway_runs_api_streaming_parses_real_run_events(): def test_live_empty_ingress_id_stays_fifo_under_capability_v1(): """A normalized fallback browser ID must not become authoritative Agent identity.""" from api.config import STREAM_PARTIAL_TEXT, STREAM_REASONING_TEXT - from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming + from api.gateway_chat import _STREAM_RUN_IDS, _gateway_runs_approval_event, _run_gateway_runs_api_streaming from api import routes import api.route_approvals as approvals sid = "sid-live-empty-ingress" stream_id = "stream-live-empty-ingress" events = [] + normalized = _gateway_runs_approval_event({ + "command": "rm -rf /tmp/x", "description": "Dangerous", + "run_id": "run-empty-ingress", "approval_id": "", "id": "", + }) + assert normalized["approval_id"] + assert normalized["_gateway_raw_approval_id_present"] is False STREAM_PARTIAL_TEXT[stream_id] = "" STREAM_REASONING_TEXT[stream_id] = "" @@ -472,22 +478,10 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1(): return _JsonResponse() return _SseResponse() - def fake_mapped_approval(payload): - assert payload["approval_id"] == "" - assert payload["id"] == "" - return { - "command": "rm -rf /tmp/x", - "description": "Dangerous", - "run_id": "run-empty-ingress", - "approval_id": "gwrun:normalized-fallback", - "_gateway_raw_approval_id_present": False, - } - handler = MagicMock() handler.wfile = io.BytesIO() try: with patch("urllib.request.urlopen", side_effect=fake_urlopen), \ - patch("api.gateway_chat._gateway_runs_approval_event", side_effect=fake_mapped_approval), \ patch("api.config.gateway_supports_approval_identity_v1", return_value=True): _run_gateway_runs_api_streaming( session_id=sid, msg_text="hi", model="test-model", workspace="/tmp", @@ -496,11 +490,12 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1(): cancel_event=threading.Event(), ) mirror = approvals.gateway_pending_mirror(sid, run_id="run-empty-ingress") + browser_id = mirror["approval_id"] with patch("api.routes.get_session", return_value=SimpleNamespace(active_stream_id=stream_id)), \ patch("api.config.gateway_supports_approval_identity_v1", return_value=True), \ patch("api.runner_client.HttpRunnerClient.respond_approval") as respond: routes._handle_approval_respond(handler, { - "session_id": sid, "choice": "once", "approval_id": "gwrun:normalized-fallback", + "session_id": sid, "choice": "once", "approval_id": browser_id, }) respond.assert_called_once_with("run-empty-ingress", "", "once") finally: @@ -509,9 +504,11 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1(): _STREAM_RUN_IDS.pop(stream_id, None) assert events[0][0] == "approval" - assert events[0][1]["approval_id"] == "gwrun:normalized-fallback" + assert events[0][1]["approval_id"] + assert events[0][1]["_gateway_raw_approval_id_present"] is False assert events[0][1]["_gateway_agent_identity_v1"] is False - assert mirror["approval_id"] == "gwrun:normalized-fallback" + assert mirror["approval_id"] == events[0][1]["approval_id"] + assert mirror["_gateway_raw_approval_id_present"] is False assert mirror["_gateway_agent_identity_v1"] is False approvals._pending.pop(sid, None) approvals._gateway_queues.pop(sid, None)