diff --git a/api/gateway_chat.py b/api/gateway_chat.py index 0807af48..c66175d5 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -741,8 +741,12 @@ def _run_gateway_runs_api_streaming( sse_event = "message" continue if payload_event == "run.completed": - from api.route_approvals import retire_gateway_pending_mirror - retire_gateway_pending_mirror(session_id, run_id=run_id) + from api.route_approvals import settle_gateway_pending_run + settle_gateway_pending_run( + session_id, + run_id, + reason="Gateway run completed before approval resolution", + ) if payload.get("error"): raise RuntimeError(str(payload["error"])) output = str(payload.get("output") or "") @@ -754,12 +758,20 @@ def _run_gateway_runs_api_streaming( sse_event = "message" continue if payload_event == "run.failed": - from api.route_approvals import retire_gateway_pending_mirror - retire_gateway_pending_mirror(session_id, run_id=run_id) + from api.route_approvals import settle_gateway_pending_run + settle_gateway_pending_run( + session_id, + run_id, + reason="Gateway run failed before approval resolution", + ) raise RuntimeError(str(payload.get("error") or "Gateway run failed")) if payload_event == "run.cancelled": - from api.route_approvals import retire_gateway_pending_mirror - retire_gateway_pending_mirror(session_id, run_id=run_id) + from api.route_approvals import settle_gateway_pending_run + settle_gateway_pending_run( + session_id, + run_id, + reason="Gateway run was cancelled before approval resolution", + ) put_gateway_event("cancel", {"message": "Cancelled by gateway"}) return None, usage reasoning_delta = _gateway_sse_reasoning_delta(payload) @@ -1467,10 +1479,14 @@ def _run_gateway_chat_streaming( 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) + from api.route_approvals import settle_gateway_pending_run + settle_gateway_pending_run( + session_id, + mapped_run_id, + reason="Gateway run ended during teardown before approval resolution", + ) except Exception: - logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True) + logger.debug("Failed to settle gateway pending approvals 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 d956ab27..da3a27d9 100644 --- a/api/route_approvals.py +++ b/api/route_approvals.py @@ -10,13 +10,16 @@ Gateway approvals carry two pieces of state with one lifecycle: * the **mirror** entry in ``_pending`` — this module's projection of that producer for the UI. -A producer is consumed only by an explicit resolution operation, which assigns -its result and signals its waiter before reconciliation can observe it again. +A producer is consumed only by an explicit settlement operation, which assigns +its result and signals its waiter before fallible projection work begins. ``retire_gateway_pending_mirror`` cleans up the UI projection; it must not silently delete a producer, because that would strand the agent thread parked on the producer's event. For a provider-qualified response, ``resolve_gateway_pending_run`` matches the exact ``(run_id, approval_id)`` pair, settles that producer, and preserves same-run siblings and other runs. +When a run becomes terminal, ``settle_gateway_pending_run`` fail-closes every +producer owned by that run with ``deny`` before retiring its mirrors; producers +owned by another run remain live. """ import queue import threading @@ -520,19 +523,26 @@ def release_gateway_approval_relay_owner(session_key: str, run_id: str, approval _gateway_relay_owners.pop(key, None) +def _settle_gateway_entry(entry, choice: str, reason: str | None = None) -> None: + """Complete a removed producer before any fallible projection work.""" + entry.result = choice + if reason: + entry.reason = reason + entry.event.set() + + def retire_gateway_pending_mirror( session_key: str, approval_id: str = "", run_id: str = "", mirror_token: str = "", ) -> bool: - """Retire one approval, or every mirror for a terminal run.""" + """Retire one approval mirror, or every mirror for a terminal run.""" with _lock: reconcile_gateway_pending_mirror_locked(session_key) queue = _pending.get(session_key) entries = queue if isinstance(queue, list) else [queue] if queue else [] normalized_run_id = str(run_id or "").strip() - gateway_queue_changed = False if approval_id: match = _gateway_pending_mirror_locked( session_key, @@ -555,20 +565,7 @@ def retire_gateway_pending_mirror( if _is_gateway_mirror_entry(entry) and not str(entry.get("run_id") or "").strip() ] - if normalized_run_id: - gateway_queue = _gateway_queues.get(session_key) or [] - retained_gateway_queue = [] - for entry in gateway_queue: - data = getattr(entry, "data", None) or {} - if str(data.get("run_id") or "").strip() == normalized_run_id: - continue - retained_gateway_queue.append(entry) - gateway_queue_changed = len(retained_gateway_queue) != len(gateway_queue) - if retained_gateway_queue: - _gateway_queues[session_key] = retained_gateway_queue - else: - _gateway_queues.pop(session_key, None) - if not retired and not gateway_queue_changed: + if not retired: head, total, changed = reconcile_gateway_pending_mirror_locked(session_key) _approval_sse_notify_locked(session_key, head, total) if changed: @@ -586,6 +583,60 @@ def retire_gateway_pending_mirror( return True +def settle_gateway_pending_run( + session_key: str, + run_id: str, + *, + reason: str, +) -> tuple[int, dict | None, int]: + """Fail-close every producer and mirror owned by a terminal gateway run.""" + run_id = str(run_id or "").strip() + if not run_id: + return 0, None, 0 + + targets = [] + with _lock: + gateway_queue = _gateway_queues.get(session_key) or [] + retained_gateway_queue = [] + for entry in gateway_queue: + data = getattr(entry, "data", None) or {} + if str(data.get("run_id") or "").strip() == run_id: + targets.append(entry) + else: + retained_gateway_queue.append(entry) + if retained_gateway_queue: + _gateway_queues[session_key] = retained_gateway_queue + else: + _gateway_queues.pop(session_key, None) + + # Removal and waiter settlement are one semantic operation. In + # particular, reconciliation and SSE notification below may raise. + for entry in targets: + _settle_gateway_entry(entry, "deny", reason) + + queue = _pending.get(session_key) + entries = queue if isinstance(queue, list) else [queue] if queue else [] + retained_pending = [ + entry + for entry in entries + if not ( + _is_gateway_mirror_entry(entry) + and str(entry.get("run_id") or "").strip() == run_id + ) + ] + mirrors_retired = len(retained_pending) != len(entries) + if retained_pending: + _pending[session_key] = retained_pending + else: + _pending.pop(session_key, None) + + head, total, changed = reconcile_gateway_pending_mirror_locked(session_key) + _approval_sse_notify_locked(session_key, head, total) + if targets or mirrors_retired or changed: + publish_session_list_changed("attention_resolved") + return len(targets), head, total + + def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str | None: """Compatibility wrapper for exact run-backed lookup.""" approval_id = str(approval_id or "").strip() @@ -817,14 +868,12 @@ def resolve_gateway_pending_local( _gateway_queues[session_key] = gateway_queue else: _gateway_queues.pop(session_key, None) + if target is not None: + _settle_gateway_entry(target, choice, reason) 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 @@ -857,14 +906,12 @@ def resolve_gateway_pending_run( _gateway_queues[session_key] = retained_gateway_queue else: _gateway_queues.pop(session_key, None) + if target is not None: + _settle_gateway_entry(target, choice, reason) 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 @@ -909,12 +956,9 @@ def resolve_gateway_pending_local_no_run_mirror( _pending[session_key] = entries else: _pending.pop(session_key, None) + _settle_gateway_entry(target, choice, reason) head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) _approval_sse_notify_locked(session_key, head, total) - target.result = choice - if reason: - target.reason = reason - target.event.set() publish_session_list_changed("attention_resolved") return True, 1, head, total @@ -957,14 +1001,11 @@ def resolve_gateway_pending_local_all( else: _pending.pop(session_key, None) + for entry in targets: + _settle_gateway_entry(entry, choice, reason) head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) _approval_sse_notify_locked(session_key, head, total) - for entry in targets: - entry.result = choice - if reason: - entry.reason = reason - entry.event.set() if targets or removed_pending: publish_session_list_changed("attention_resolved") return len(targets), head, total diff --git a/api/routes.py b/api/routes.py index 883dbbd7..6afd81ad 100644 --- a/api/routes.py +++ b/api/routes.py @@ -10771,6 +10771,7 @@ from api.route_approvals import ( # noqa: F401 — re-exports for backward comp gateway_pending_mirrors, release_gateway_approval_relay_owner, retire_gateway_pending_mirror, + settle_gateway_pending_run, reconcile_gateway_pending_mirror_locked, resolve_gateway_pending_local, resolve_gateway_pending_run, @@ -14485,7 +14486,11 @@ def handle_get(handler, parsed) -> bool: if stop_gateway_run(run_id): owner_sid = stream_owner_session_id(stream_id) if owner_sid: - retire_gateway_pending_mirror(owner_sid, run_id=run_id) + settle_gateway_pending_run( + owner_sid, + run_id, + reason="Gateway run was cancelled before approval resolution", + ) else: gateway_stop_blocked = True except Exception: diff --git a/tests/test_gateway_approval_legacy_path.py b/tests/test_gateway_approval_legacy_path.py index a2b0d74c..20169d19 100644 --- a/tests/test_gateway_approval_legacy_path.py +++ b/tests/test_gateway_approval_legacy_path.py @@ -10,6 +10,8 @@ import threading from pathlib import Path from unittest.mock import MagicMock, patch +import pytest + from api.gateway_chat import _gateway_runs_approval_event REPO_ROOT = Path(__file__).parent.parent @@ -1110,3 +1112,109 @@ def test_deny_settles_run_producer_so_reconciliation_cannot_resurrect_mirror(): with ra._lock: ra._pending.pop(sid, None) ra._gateway_queues.pop(sid, None) + + +@pytest.mark.parametrize( + "failure_point", + ["reconcile_gateway_pending_mirror_locked", "_approval_sse_notify_locked"], +) +def test_exact_run_resolution_settles_before_projection_failure(failure_point): + """Projection failures must not strand an already-consumed producer.""" + from types import SimpleNamespace + from api import route_approvals as ra + + sid = f"sess-exact-projection-failure-{failure_point}" + run_id = "run-exact-projection-failure" + target = SimpleNamespace( + data={"run_id": run_id, "approval_id": "approval-target"}, + event=threading.Event(), + result=None, + reason=None, + ) + sibling = SimpleNamespace( + data={"run_id": run_id, "approval_id": "approval-sibling"}, + event=threading.Event(), + result=None, + reason=None, + ) + try: + with ra._lock: + ra._pending.pop(sid, None) + ra._gateway_queues[sid] = [target, sibling] + + with patch.object(ra, failure_point, side_effect=RuntimeError("projection failed")): + with pytest.raises(RuntimeError, match="projection failed"): + ra.resolve_gateway_pending_run( + sid, + approval_id="approval-target", + run_id=run_id, + choice="deny", + reason="user denied", + ) + + assert target.result == "deny" + assert target.reason == "user denied" + assert target.event.is_set(), "the consumed producer must be woken before projection work" + assert sibling.result is None + assert sibling.reason is None + assert not sibling.event.is_set() + with ra._lock: + assert ra._gateway_queues[sid] == [sibling] + finally: + with ra._lock: + ra._pending.pop(sid, None) + ra._gateway_queues.pop(sid, None) + + +@pytest.mark.parametrize( + "terminal_reason", + [ + "Gateway run completed before approval resolution", + "Gateway run failed before approval resolution", + "Gateway run was cancelled before approval resolution", + "Gateway run ended during teardown before approval resolution", + ], +) +def test_terminal_run_settlement_denies_all_run_producers_and_preserves_other_runs( + terminal_reason, +): + """Every terminal exit fails closed without consuming another run's producer.""" + from types import SimpleNamespace + from api import route_approvals as ra + + sid = f"sess-terminal-run-{terminal_reason.split()[2]}" + run_id = "run-terminal" + + def producer(entry_run_id, approval_id): + return SimpleNamespace( + data={"run_id": entry_run_id, "approval_id": approval_id}, + event=threading.Event(), + result=None, + reason=None, + ) + + targets = [producer(run_id, "approval-a"), producer(run_id, "approval-b")] + survivor = producer("run-other", "approval-other") + try: + with ra._lock: + ra._pending.pop(sid, None) + ra._gateway_queues[sid] = [targets[0], survivor, targets[1]] + + settled, _head, _total = ra.settle_gateway_pending_run( + sid, run_id, reason=terminal_reason + ) + + assert settled == 2 + for target in targets: + assert target.result == "deny" + assert target.reason == terminal_reason + assert target.event.is_set() + assert survivor.result is None + assert survivor.reason is None + assert not survivor.event.is_set() + with ra._lock: + assert ra._gateway_queues[sid] == [survivor] + finally: + with ra._lock: + ra._pending.pop(sid, None) + ra._gateway_queues.pop(sid, None) diff --git a/tests/test_gateway_approval_runs_api.py b/tests/test_gateway_approval_runs_api.py index 7635aefb..51022542 100644 --- a/tests/test_gateway_approval_runs_api.py +++ b/tests/test_gateway_approval_runs_api.py @@ -333,6 +333,178 @@ def test_gateway_runs_api_submission(): assert captured["body_extras"]["reasoning_effort"] == "high" +@pytest.mark.parametrize( + ("terminal_event", "terminal_reason"), + [ + ("run.completed", "Gateway run completed before approval resolution"), + ("run.failed", "Gateway run failed before approval resolution"), + ("run.cancelled", "Gateway run was cancelled before approval resolution"), + ], +) +def test_gateway_terminal_event_fail_closes_parked_run_producer( + terminal_event, + terminal_reason, +): + """Each terminal gateway event denies and wakes the run's parked producers.""" + from api import gateway_chat + from api import route_approvals as approvals + + sid = f"sess-terminal-event-{terminal_event}" + stream_id = f"stream-terminal-event-{terminal_event}" + run_id = f"run-terminal-event-{terminal_event}" + target = SimpleNamespace( + data={"run_id": run_id, "approval_id": "approval-target"}, + event=threading.Event(), + result=None, + reason=None, + ) + survivor = SimpleNamespace( + data={"run_id": "run-other", "approval_id": "approval-other"}, + event=threading.Event(), + result=None, + reason=None, + ) + + class JsonResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def read(self, _limit=None): + return json.dumps({"run_id": run_id}).encode() + + class SseResponse: + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def __iter__(self): + payload = {"event": terminal_event} + if terminal_event == "run.failed": + payload["error"] = "terminal failure" + return iter([f"data: {json.dumps(payload)}".encode(), b""]) + + def fake_urlopen(req, *, timeout=None): + del timeout + return JsonResponse() if req.full_url.endswith("/v1/runs") else SseResponse() + + try: + with approvals._lock: + approvals._pending.pop(sid, None) + approvals._gateway_queues[sid] = [target, survivor] + with patch("urllib.request.urlopen", side_effect=fake_urlopen): + if terminal_event == "run.failed": + with pytest.raises(RuntimeError, match="terminal failure"): + gateway_chat._run_gateway_runs_api_streaming( + sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {}, + put_gateway_event=lambda *_args: None, + cancel_event=threading.Event(), + session=SimpleNamespace(context_messages=[]), + ) + else: + gateway_chat._run_gateway_runs_api_streaming( + sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {}, + put_gateway_event=lambda *_args: None, + cancel_event=threading.Event(), + session=SimpleNamespace(context_messages=[]), + ) + + assert target.result == "deny" + assert target.reason == terminal_reason + assert target.event.is_set() + assert survivor.result is None + assert not survivor.event.is_set() + with approvals._lock: + assert approvals._gateway_queues[sid] == [survivor] + finally: + gateway_chat._STREAM_RUN_IDS.pop(stream_id, None) + with approvals._lock: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_gateway_stream_teardown_fail_closes_parked_run_producer(): + """A disconnect before a terminal event still settles the mapped run.""" + from api import gateway_chat + from api import route_approvals as approvals + from api.config import STREAMS, STREAMS_LOCK + + sid = "sess-terminal-teardown" + stream_id = "stream-terminal-teardown" + run_id = "run-terminal-teardown" + target = SimpleNamespace( + data={"run_id": run_id, "approval_id": "approval-target"}, + event=threading.Event(), + result=None, + reason=None, + ) + survivor = SimpleNamespace( + data={"run_id": "run-other", "approval_id": "approval-other"}, + event=threading.Event(), + result=None, + reason=None, + ) + events = [] + queue = MagicMock() + queue.put_nowait = lambda item: events.append(item) + with STREAMS_LOCK: + STREAMS[stream_id] = queue + + session = MagicMock() + session.active_stream_id = stream_id + session.workspace = "/tmp" + session.model = "test" + session.model_provider = None + session.profile = None + session.context_messages = [] + session.messages = [] + session.pending_user_message = None + session.pending_attachments = None + session.pending_started_at = None + + def disconnect_after_run_id(*_args, **_kwargs): + gateway_chat._publish_gateway_run_id(stream_id, run_id) + raise RuntimeError("gateway disconnected") + + try: + with approvals._lock: + approvals._pending.pop(sid, None) + approvals._gateway_queues[sid] = [target, survivor] + with patch.dict( + "os.environ", + {"HERMES_WEBUI_CHAT_BACKEND": "gateway", "HERMES_WEBUI_GATEWAY_USE_RUNS_API": "1"}, + ), patch("api.gateway_chat.gateway_supports_approval", return_value=True), patch( + "api.gateway_chat._run_gateway_runs_api_streaming", + side_effect=disconnect_after_run_id, + ), patch("api.gateway_chat.get_session", return_value=session): + gateway_chat._run_gateway_chat_streaming( + session_id=sid, + msg_text="hi", + model="test-model", + workspace="/tmp", + stream_id=stream_id, + ) + + assert target.result == "deny" + assert target.reason == "Gateway run ended during teardown before approval resolution" + assert target.event.is_set() + assert survivor.result is None + assert not survivor.event.is_set() + with approvals._lock: + assert approvals._gateway_queues[sid] == [survivor] + finally: + gateway_chat._STREAM_RUN_IDS.pop(stream_id, None) + with STREAMS_LOCK: + STREAMS.pop(stream_id, None) + with approvals._lock: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + # --------------------------------------------------------------------------- # 3. Approval event translation # --------------------------------------------------------------------------- @@ -1268,22 +1440,32 @@ def test_terminal_run_retirement_removes_all_same_run_mirrors(): approvals._pending.pop(sid, None) -def test_terminal_run_retirement_clears_same_run_gateway_queue_state(): +def test_terminal_run_settlement_clears_same_run_gateway_queue_state(): import api.route_approvals as approvals sid = "sess-terminal-run-queue" approvals._pending.pop(sid, None) approvals._gateway_queues.pop(sid, None) try: - approvals._gateway_queues[sid] = [ - SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}), - SimpleNamespace(data={"command": "local-head"}), - ] + target = SimpleNamespace( + data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}, + event=threading.Event(), + result=None, + reason=None, + ) + local_head = SimpleNamespace(data={"command": "local-head"}) + approvals._gateway_queues[sid] = [target, local_head] approvals.submit_gateway_pending_mirror( sid, {"run_id": "run-a", "approval_id": "appr-a", "command": "a"}, ) - assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") + settled, _head, _total = approvals.settle_gateway_pending_run( + sid, "run-a", reason="terminal" + ) + assert settled == 1 + assert target.result == "deny" + assert target.reason == "terminal" + assert target.event.is_set() assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None assert not any( str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a" @@ -1294,18 +1476,30 @@ def test_terminal_run_retirement_clears_same_run_gateway_queue_state(): approvals._gateway_queues.pop(sid, None) -def test_terminal_run_retirement_clears_non_head_same_run_gateway_queue_state(): +def test_terminal_run_settlement_clears_non_head_same_run_gateway_queue_state(): import api.route_approvals as approvals sid = "sess-terminal-run-non-head-queue" approvals._pending.pop(sid, None) approvals._gateway_queues.pop(sid, None) try: + target = SimpleNamespace( + data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}, + event=threading.Event(), + result=None, + reason=None, + ) approvals._gateway_queues[sid] = [ SimpleNamespace(data={"command": "local-head"}), - SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}), + target, ] - assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") + settled, _head, _total = approvals.settle_gateway_pending_run( + sid, "run-a", reason="terminal" + ) + assert settled == 1 + assert target.result == "deny" + assert target.reason == "terminal" + assert target.event.is_set() assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None assert not any( str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a" @@ -1865,7 +2059,15 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement( STREAMS[stream_id] = SimpleNamespace(put_nowait=lambda *_args, **_kwargs: None) register_stream_owner(stream_id, sid) approvals._pending.pop(sid, None) - approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop/1", "command": "first"}) + approval_entry = SimpleNamespace( + data={"run_id": "run-stop/1", "approval_id": "approval-stop"}, + event=threading.Event(), + result=None, + reason=None, + ) + with approvals._lock: + approvals._gateway_queues[sid] = [approval_entry] + approvals.submit_gateway_pending_mirror(sid, approval_entry.data) session = SimpleNamespace( profile=None, workspace="/tmp", @@ -1945,8 +2147,15 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement( assert called["stop"] == "run-stop/1" assert called["cancel"] is expect_cancel if stop_result: + assert approval_entry.result == "deny" + assert approval_entry.reason == ( + "Gateway run was cancelled before approval resolution" + ) + assert approval_entry.event.is_set() assert approvals.gateway_pending_mirror(sid, run_id="run-stop/1") is None else: + assert approval_entry.result is None + assert not approval_entry.event.is_set() assert stream_id in STREAMS assert stream_owner_session_id(stream_id) == sid assert session.active_stream_id == stream_id @@ -1957,7 +2166,9 @@ def test_chat_cancel_waits_for_worker_published_run_id_before_settlement( release_worker.set() request_thread.join(timeout=5) worker_thread.join(timeout=5) - approvals._pending.pop(sid, None) + with approvals._lock: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) STREAMS.pop(stream_id, None) ACTIVE_RUNS.pop(stream_id, None) unregister_stream_owner(stream_id)