Merge pull request #7704 from nesquena/stage/7570-approval-retire

Release: retire producer on gateway approval deny (#7570)

CHANGELOG.md

@@ -5,6 +5,14 @@
 
 ### Fixed
 
+- **Denying a gateway approval now retires the request instead of leaving the run waiting.**
+  When an approval routed through the gateway was denied, the deny reached the agent but the
+  local producer was never retired, so the run could sit waiting on a decision that had already
+  been made. Denial now settles the producer atomically, scoped to the exact
+  `(session_id, run_id, approval_id)` that was answered — sibling approvals from the same run
+  and approvals from other runs are left untouched — and repeated or late responses are bounded
+  no-ops rather than resurrecting a settled request. Thanks @snoyberg. (#7570)
+
 - **The approval card's "Skip all this session" button no longer shows two lightning bolts.**
   The button rendered its ⚡ twice — once from the icon span in the markup and again from the
   translated label, which carried its own leading glyph in 14 of 15 locales. The icon now comes

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):

api/route_approvals.py

@@ -2,6 +2,24 @@
 
 State-extraction prelude to the routes.py split tracked in #1907.
 Extracts approval state, not handlers, by design.
+
+Gateway approvals carry two pieces of state with one lifecycle:
+
+* the **producer** entry in ``tools.approval._gateway_queues`` — the queued
+  approval itself, which owns the lifecycle; and
+* the **mirror** entry in ``_pending`` — this module's projection of that
+  producer for the UI.
+
+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
@@ -505,21 +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 = _gateway_queues.get(session_key) or []
-        retained_gateway_queue = gateway_queue
-        gateway_queue_changed = False
         if approval_id:
             match = _gateway_pending_mirror_locked(
                 session_key,
@@ -542,15 +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:
-                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:
-                        gateway_queue_changed = True
-                        continue
-                    retained_gateway_queue.append(entry)
-        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:
@@ -558,11 +573,6 @@ def retire_gateway_pending_mirror(
             return changed
         for match in retired:
             entries.remove(match)
-        if normalized_run_id and not approval_id:
-            if retained_gateway_queue:
-                _gateway_queues[session_key] = retained_gateway_queue
-            else:
-                _gateway_queues.pop(session_key, None)
         if entries:
             _pending[session_key] = entries
         else:
@@ -573,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()
@@ -804,14 +868,50 @@ 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
+    publish_session_list_changed("attention_resolved")
+    return 1, head, total
+
+
+def resolve_gateway_pending_run(
+    session_key: str,
+    approval_id: str,
+    run_id: str,
+    choice: str,
+    reason: str | None = None,
+) -> tuple[int, dict | None, int]:
+    """Resolve one exact run-backed producer and wake its parked waiter."""
+    target = None
+    approval_id = str(approval_id or "").strip()
+    run_id = str(run_id or "").strip()
+    if not approval_id or not run_id:
+        return 0, None, 0
+    with _lock:
+        gateway_queue = _gateway_queues.get(session_key) or []
+        retained_gateway_queue = []
+        for entry in gateway_queue:
+            data = getattr(entry, "data", None) or {}
+            entry_approval_id = str(data.get("approval_id") or "").strip()
+            entry_run_id = str(data.get("run_id") or "").strip()
+            if target is None and entry_approval_id == approval_id and entry_run_id == run_id:
+                target = 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)
+        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
 
@@ -856,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
 
@@ -904,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

api/routes.py

@@ -10771,8 +10771,10 @@ 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,
     resolve_gateway_pending_local_all,
     resolve_gateway_pending_local_no_run_mirror,
     set_session_yolo_enabled,
@@ -14484,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:
@@ -26219,7 +26225,6 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st
     found_target = False
     gateway_keys = []
     local_gateway_approval_id = ""
-    gateway_head_matches_target = False
     with _lock:
         reconcile_gateway_pending_mirror_locked(sid)
         queue = _pending.get(sid)
@@ -26291,9 +26296,6 @@ 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()
-                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:
@@ -26354,8 +26356,10 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st
         local_gateway_resolved, _head, _total = resolve_gateway_pending_local(
             sid, local_gateway_approval_id, choice
         )
-    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 approval_id and found_target and run_id:
+        gateway_resolved, _head, _total = resolve_gateway_pending_run(
+            sid, approval_id, run_id, choice
+        )
     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

tests/test_approval_unblock.py

@@ -601,6 +601,85 @@ class TestApprovalHTTPEndpoints:
                 r._gateway_queues.pop(sid, None)
                 _STREAM_RUN_IDS.pop(stream_id, None)
 
+    def test_identity_v1_non_head_run_approval_settles_exact_producer(self, monkeypatch):
+        """A relayed non-head run approval must wake only its own parked waiter."""
+        from api import route_approvals as ra
+        from api import routes as r
+
+        sid = f"identity-v1-non-head-{uuid.uuid4().hex[:8]}"
+        run_id = f"run-identity-v1-{uuid.uuid4().hex[:8]}"
+        head = {
+            "approval_id": "head-approval",
+            "run_id": run_id,
+            "command": "head command",
+            "_gateway_agent_identity_v1": True,
+        }
+        target = {
+            "approval_id": "target-approval",
+            "run_id": run_id,
+            "command": "target command",
+            "_gateway_agent_identity_v1": True,
+        }
+        head_entry = _ApprovalEntry(dict(head))
+        target_entry = _ApprovalEntry(dict(target))
+        captured = {}
+        relays = []
+
+        def fake_j(_handler, data, status=200, extra_headers=None):
+            captured.update(payload=data, status=status)
+            return data
+
+        def fake_respond(_self, got_run_id, got_approval_id, choice):
+            relays.append((got_run_id, got_approval_id, choice))
+            return {"resolved": 1}
+
+        monkeypatch.setattr(r, "j", fake_j)
+        monkeypatch.setattr("api.runner_client.HttpRunnerClient.respond_approval", fake_respond)
+        monkeypatch.setattr(
+            "api.config.gateway_supports_approval_identity_v1", lambda *_args: True
+        )
+        with _lock:
+            r._pending.pop(sid, None)
+            r._gateway_queues[sid] = [head_entry, target_entry]
+        try:
+            ra.submit_gateway_pending_mirror(sid, dict(target_entry.data))
+            mirror = ra.gateway_pending_mirror(
+                sid, approval_id=target["approval_id"], run_id=run_id
+            )
+            assert mirror is not None
+
+            r._handle_approval_respond(
+                object(),
+                {
+                    "session_id": sid,
+                    "choice": "deny",
+                    "approval_id": target["approval_id"],
+                    "run_id": run_id,
+                    "mirror_token": mirror[ra._GATEWAY_MIRROR_TOKEN],
+                },
+            )
+
+            assert captured == {
+                "payload": {"ok": True, "choice": "deny", "relayed": True},
+                "status": 200,
+            }
+            assert relays == [(run_id, target["approval_id"], "deny")]
+            assert target_entry.event.is_set()
+            assert target_entry.result == "deny"
+            assert not head_entry.event.is_set()
+            with _lock:
+                assert r._gateway_queues[sid] == [head_entry]
+            for _ in range(2):
+                with _lock:
+                    ra.reconcile_gateway_pending_mirror_locked(sid)
+                assert ra.gateway_pending_mirror(
+                    sid, approval_id=target["approval_id"], run_id=run_id
+                ) is None
+        finally:
+            with _lock:
+                r._pending.pop(sid, None)
+                r._gateway_queues.pop(sid, None)
+
     def test_gateway_mirror_without_run_id_with_one_producer_resolves_exactly(self, monkeypatch):
         """A no-run mirror retires only after its exact local producer resolves."""
         from api import routes as r

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
@@ -873,3 +875,346 @@ def test_legacy_approval_without_run_id_retires_locally():
             r._pending.pop(session_id, None)
             r._gateway_queues.pop(session_id, None)
         _STREAM_RUN_IDS.pop(stream_id, None)
+
+
+def test_route_deny_settles_exact_non_head_run_producer():
+    """The response route must settle the exact producer whose card was denied.
+
+    The route is the only caller that supplies a target's ``(run_id,
+    approval_id)`` to the legacy resolver, so a regression in that wiring is
+    invisible to helper-level coverage: the producer would be dropped from the
+    queue without its waiter ever being woken, leaving the agent thread parked
+    until the approval timeout.
+
+    Unlike the route-level test in ``test_approval_unblock.py`` (which needs the
+    installed agent's ``tools.approval`` and is skipped without it), this one
+    runs wherever the suite runs.
+    """
+    from types import SimpleNamespace
+    from api import route_approvals as ra
+    from api import routes as r
+
+    sid = "sess-route-non-head-deny"
+    run_id = "run-route-non-head-deny"
+    head_id = "approval-route-head"
+    target_id = "approval-route-target"
+    sibling_id = "approval-route-sibling"
+    other_run_id = "run-route-unrelated"
+    head = {
+        "approval_id": head_id,
+        "run_id": run_id,
+        "command": "echo head",
+        "description": "Head approval on the run",
+        "_gateway_agent_identity_v1": True,
+    }
+    sibling = {
+        "approval_id": sibling_id,
+        "run_id": run_id,
+        "command": "echo sibling",
+        "description": "Non-head sibling approval on the same run",
+        "_gateway_agent_identity_v1": True,
+    }
+    target = {
+        "approval_id": target_id,
+        "run_id": run_id,
+        "command": "echo target",
+        "description": "Non-head approval on the same run",
+        "_gateway_agent_identity_v1": True,
+    }
+    other_run = {
+        "approval_id": "approval-route-other",
+        "run_id": other_run_id,
+        "command": "echo else",
+        "description": "Approval on an unrelated run",
+    }
+
+    def producer(payload):
+        """The agent's real producer entry when installed, an equivalent one otherwise.
+
+        The suite runs without hermes-agent in CI, where
+        ``tools.approval._ApprovalEntry`` does not exist; the stand-in carries the
+        same ``data`` / ``event`` / ``result`` / ``reason`` contract the resolution
+        path touches, so the assertions below mean the same thing in both shapes.
+        """
+        try:
+            from tools.approval import _ApprovalEntry
+        except Exception:
+            _ApprovalEntry = None
+        if _ApprovalEntry is not None:
+            return _ApprovalEntry(dict(payload))
+        return SimpleNamespace(
+            data=dict(payload), event=threading.Event(), result=None, reason=None
+        )
+
+    head_entry = producer(head)
+    sibling_entry = producer(sibling)
+    target_entry = producer(target)
+    other_run_entry = producer(other_run)
+    relayed = []
+    captured = {}
+
+    def fake_j(_handler, data, status=200, extra_headers=None):
+        captured.update(payload=data, status=status)
+        return data
+
+    def fake_respond(_self, got_run_id, got_approval_id, choice):
+        relayed.append((got_run_id, got_approval_id, choice))
+        return {"resolved": 1}
+
+    try:
+        with ra._lock:
+            ra._pending.pop(sid, None)
+            ra._gateway_queues[sid] = [head_entry, sibling_entry, target_entry, other_run_entry]
+        ra.submit_gateway_pending_mirror(sid, dict(target_entry.data))
+        mirror = ra.gateway_pending_mirror(sid, approval_id=target_id, run_id=run_id)
+        assert mirror is not None, "the non-head approval must be mirrored before it is answered"
+
+        with patch("api.routes.j", new=fake_j), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval", new=fake_respond), \
+             patch("api.config.gateway_supports_approval_identity_v1", return_value=True):
+            r._handle_approval_respond(
+                object(),
+                {
+                    "session_id": sid,
+                    "choice": "deny",
+                    "approval_id": target_id,
+                    "run_id": run_id,
+                    "mirror_token": mirror[ra._GATEWAY_MIRROR_TOKEN],
+                },
+            )
+
+        assert captured == {
+            "payload": {"ok": True, "choice": "deny", "relayed": True},
+            "status": 200,
+        }
+        assert relayed == [(run_id, target_id, "deny")]
+
+        assert target_entry.event.is_set(), "the denied approval's waiter must be woken"
+        assert target_entry.result == "deny"
+        with ra._lock:
+            remaining = list(ra._gateway_queues.get(sid) or [])
+        assert target_entry not in remaining, "the denied approval's producer must be consumed"
+        assert head_entry in remaining, (
+            "the run's own head approval must not be consumed by answering another approval"
+        )
+        assert not head_entry.event.is_set()
+        assert sibling_entry in remaining, (
+            "a sibling approval on the same run must not be consumed"
+        )
+        assert not sibling_entry.event.is_set()
+        assert other_run_entry in remaining, (
+            "an unrelated run's producer must not be consumed"
+        )
+        assert not other_run_entry.event.is_set()
+
+        # These are the observations made by subsequent HTTP polling.
+        for _ in range(2):
+            with ra._lock:
+                ra.reconcile_gateway_pending_mirror_locked(sid)
+            assert ra.gateway_pending_mirror(
+                sid, approval_id=target_id, run_id=run_id
+            ) is None, "reconciliation must not resurrect the denied approval's mirror"
+    finally:
+        with ra._lock:
+            ra._pending.pop(sid, None)
+            ra._gateway_queues.pop(sid, None)
+
+
+def test_deny_settles_run_producer_so_reconciliation_cannot_resurrect_mirror():
+    """Deny must settle the producer as well as its WebUI mirror.
+
+    Selection is the behaviour under test, so the queue carries more than the
+    one entry being resolved: a sibling approval for the same run, and a producer
+    belonging to a different run. Resolving one approval must consume exactly
+    that producer, wake its waiter, and leave both others queued.
+    """
+    from types import SimpleNamespace
+    from api import route_approvals as ra
+
+    sid = "sess-deny-no-resurrection"
+    approval_id = "approval-deny-no-resurrection"
+    run_id = "run-deny-no-resurrection"
+    sibling_approval_id = "approval-deny-sibling"
+    other_run_id = "run-deny-unrelated"
+    approval = {
+        "approval_id": approval_id,
+        "run_id": run_id,
+        "command": "echo pricing",
+        "description": "Run pricing probe",
+    }
+    sibling_approval = {
+        "approval_id": sibling_approval_id,
+        "run_id": run_id,
+        "command": "echo sibling",
+        "description": "Sibling approval on the same run",
+    }
+    other_run_approval = {
+        "approval_id": "approval-deny-other-run",
+        "run_id": other_run_id,
+        "command": "echo else",
+        "description": "Approval on an unrelated run",
+    }
+
+    def producer(payload):
+        return SimpleNamespace(data=dict(payload), event=threading.Event(), result=None)
+
+    target = producer(approval)
+    same_run_sibling = producer(sibling_approval)
+    other_run = producer(other_run_approval)
+    try:
+        with ra._lock:
+            ra._pending.pop(sid, None)
+            ra._gateway_queues[sid] = [target, same_run_sibling, other_run]
+            queued_before = list(ra._gateway_queues[sid])
+        ra.submit_gateway_pending_mirror(sid, dict(approval))
+        mirror = ra.gateway_pending_mirror(sid, approval_id=approval_id, run_id=run_id)
+        assert mirror is not None
+
+        resolved, _head, _total = ra.resolve_gateway_pending_run(
+            sid,
+            approval_id=approval_id,
+            run_id=run_id,
+            choice="deny",
+        )
+        assert resolved == 1
+
+        assert queued_before == [target, same_run_sibling, other_run], (
+            "the queue should have held all three producers before retirement"
+        )
+        with ra._lock:
+            remaining = list(ra._gateway_queues.get(sid) or [])
+        assert target not in remaining, "the denied approval's producer must be consumed"
+        assert target.event.is_set(), "the denied approval's waiter must be woken"
+        assert target.result == "deny"
+        assert same_run_sibling in remaining, (
+            "a sibling approval on the same run must not be consumed by retiring another approval"
+        )
+        assert other_run in remaining, (
+            "an unrelated run's producer must not be consumed"
+        )
+
+        # These are the same observations made by subsequent HTTP polling: the
+        # denied approval must not come back, and the survivors must stay live.
+        # `gateway_pending_mirror` acquires `_lock` itself, so it must be called
+        # outside the locked section.
+        for _ in range(2):
+            with ra._lock:
+                ra.reconcile_gateway_pending_mirror_locked(sid)
+            assert ra.gateway_pending_mirror(
+                sid, approval_id=approval_id, run_id=run_id
+            ) is None, "reconciliation must not resurrect the denied approval's mirror"
+            with ra._lock:
+                refreshed = list(ra._gateway_queues.get(sid) or [])
+            assert target not in refreshed
+            assert same_run_sibling in refreshed
+            assert other_run in refreshed
+    finally:
+        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)

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)