fix(#6008): keep gateway approvals attached to the right run

api/gateway_chat.py

@@ -8,7 +8,9 @@ import threading
 import time
 import uuid
 import urllib.error
+import urllib.parse
 import urllib.request
+import uuid
 from typing import Any
 
 from api.config import (
@@ -39,6 +41,13 @@ logger = logging.getLogger(__name__)
 
 # Maps stream_id -> gateway run_id for approval response relay.
 _STREAM_RUN_IDS: dict[str, str] = {}
+_STREAM_RUN_STARTING: set[str] = set()
+_STREAM_RUN_STARTING_LOCK = threading.Lock()
+
+
+def gateway_run_id_pending(stream_id: str) -> bool:
+    with _STREAM_RUN_STARTING_LOCK:
+        return stream_id in _STREAM_RUN_STARTING
 
 _WEBUI_CHAT_BACKEND_ENV = "HERMES_WEBUI_CHAT_BACKEND"
 _WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL"
@@ -338,8 +347,6 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None:
     args = payload.get("args") if isinstance(payload.get("args"), (list, dict)) else []
     run_id = str(payload.get("run_id") or "").strip()
     approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip()
-    if not approval_id and run_id:
-        approval_id = f"gwrun:{run_id}"
     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")
@@ -369,6 +376,9 @@ def _run_gateway_runs_api_streaming(
     attachments=None, cfg=None, session=None,
 ):
     """Submit via POST /v1/runs and relay SSE events including approval."""
+    with _STREAM_RUN_STARTING_LOCK:
+        _STREAM_RUN_STARTING.add(stream_id)
+    try:
         url_runs = f"{base_url.rstrip('/')}/v1/runs"
         headers = {
             "Content-Type": "application/json",
@@ -442,15 +452,31 @@ def _run_gateway_runs_api_streaming(
         run_id = str(run_data.get("run_id") or run_data.get("id") or "").strip()
         if not run_id:
             raise ValueError(f"Gateway runs API returned no run_id: {run_data!r}")
+    except Exception:
+        with _STREAM_RUN_STARTING_LOCK:
+            _STREAM_RUN_STARTING.discard(stream_id)
+        raise
 
+    usage: dict = {}
     _STREAM_RUN_IDS[stream_id] = run_id
+    with _STREAM_RUN_STARTING_LOCK:
+        _STREAM_RUN_STARTING.discard(stream_id)
+    if cancel_event.is_set():
+        stop_confirmed = False
+        try:
+            stop_confirmed = stop_gateway_run(stream_id)
+        except Exception:
+            logger.debug("Failed to stop gateway run after pre-stream cancellation", exc_info=True)
+        if stop_confirmed:
+            put_gateway_event("cancel", {"message": "Cancelled by user"})
+            return None, usage
+        cancel_event.clear()
 
     url_events = f"{base_url.rstrip('/')}/v1/runs/{run_id}/events"
     headers_sse = dict(headers)
     headers_sse["Accept"] = "text/event-stream"
     req_events = urllib.request.Request(url_events, headers=headers_sse, method="GET")
     final_text = ""
-    usage: dict = {}
     sse_event = "message"
     with urllib.request.urlopen(req_events, timeout=_gateway_read_timeout_secs()) as resp:
         for raw_line in _iter_sse_lines_cancellable(resp, cancel_event):
@@ -479,7 +505,7 @@ def _run_gateway_runs_api_streaming(
                 if approval_data:
                     approval_data["run_id"] = run_id
                     if not approval_data.get("approval_id"):
-                        approval_data["approval_id"] = f"gwrun:{run_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)
@@ -562,6 +588,40 @@ def _run_gateway_runs_api_streaming(
     return final_text, usage
 
 
+def stop_gateway_run(stream_id: str) -> bool:
+    """Request gateway interruption and report whether it was acknowledged."""
+    run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
+    if not run_id:
+        return False
+    from api.config import get_config
+
+    cfg = get_config()
+    base_url = _gateway_base_url(cfg)
+    api_key = _gateway_api_key()
+    headers = {"Accept": "application/json", "Content-Type": "application/json"}
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    req = urllib.request.Request(
+        f"{base_url.rstrip('/')}/v1/runs/{urllib.parse.quote(run_id, safe='')}/stop",
+        data=b"{}",
+        headers=headers,
+        method="POST",
+    )
+    class _NoRedirect(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, req, fp, code, msg, headers, newurl):
+            return None
+
+    try:
+        opener = urllib.request.build_opener(_NoRedirect)
+        with opener.open(req, timeout=10) as response:
+            final_url = str(getattr(response, "geturl", lambda: req.full_url)() or "")
+            status = int(getattr(response, "status", getattr(response, "code", 0)) or 0)
+            return 200 <= status < 300 and final_url == req.full_url
+    except (urllib.error.HTTPError, urllib.error.URLError, OSError, ValueError):
+        logger.debug("Gateway stop failed for run %s", run_id, exc_info=True)
+        return False
+
+
 def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, model_provider, terminal_error):
     from api.streaming import (
         _classify_provider_error,
@@ -701,6 +761,7 @@ def _run_gateway_chat_streaming(
         STREAM_LIVE_TOOL_CALLS[stream_id] = []
 
     success_writeback_committed = False
+    runs_api_pending_marked = False
 
     def put_gateway_event(event, data):
         if cancel_event.is_set() and not success_writeback_committed and event not in ("cancel", "error", "apperror"):
@@ -742,6 +803,27 @@ def _run_gateway_chat_streaming(
             model=model,
             model_provider=model_provider,
         )
+        base_url = _gateway_base_url(cfg)
+        api_key = _gateway_api_key()
+        try:
+            from api.config import _main_model_request_overrides
+            _gw_overrides = _main_model_request_overrides(
+                cfg,
+                effective_model=model,
+                effective_provider=model_provider,
+            )
+        except Exception:
+            _gw_overrides = {}
+        _runs_api_enabled = _gateway_use_runs_api_enabled(cfg)
+        if _runs_api_enabled:
+            with _STREAM_RUN_STARTING_LOCK:
+                _STREAM_RUN_STARTING.add(stream_id)
+            runs_api_pending_marked = True
+        _use_runs_api = _runs_api_enabled and gateway_supports_approval(base_url, api_key)
+        if not _use_runs_api and runs_api_pending_marked:
+            with _STREAM_RUN_STARTING_LOCK:
+                _STREAM_RUN_STARTING.discard(stream_id)
+            runs_api_pending_marked = False
         try:
             from api.streaming import (
                 _load_webui_prefill_context,
@@ -780,19 +862,6 @@ def _run_gateway_chat_streaming(
         except Exception:
             logger.debug("Failed to load WebUI gateway prefill context", exc_info=True)
             prefill_messages = []
-        base_url = _gateway_base_url(cfg)
-        api_key = _gateway_api_key()
-        try:
-            from api.config import _main_model_request_overrides
-            _gw_overrides = _main_model_request_overrides(
-                cfg,
-                effective_model=model,
-                effective_provider=model_provider,
-            )
-        except Exception:
-            _gw_overrides = {}
-        # Capability gate: use runs API when gateway advertises approval support.
-        _use_runs_api = _gateway_use_runs_api_enabled(cfg) and gateway_supports_approval(base_url, api_key)
         if _use_runs_api:
             body_extras = {}
             if model_provider:
@@ -1207,5 +1276,8 @@ def _run_gateway_chat_streaming(
             STREAM_LAST_EVENT_ID.pop(stream_id, None)
             STREAMS.pop(stream_id, None)
         _STREAM_RUN_IDS.pop(stream_id, None)
+        if runs_api_pending_marked and gateway_run_id_pending(stream_id):
+            _finish_gateway_run_starting(stream_id)
+        _clear_gateway_run_starting(stream_id)
         unregister_stream_owner(stream_id)
         unregister_active_run(stream_id)

api/route_approvals.py

@@ -149,9 +149,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
     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
     live_run_id = str(live_head_data.get("run_id") or "").strip()
+    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()
 
     rebuilt: list[dict] = []
+    deferred_run_entries: list[dict] = []
     live_mirror_present = False
     for entry in queue_list:
         if not _is_gateway_mirror_entry(entry):
@@ -164,9 +167,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
         if live_token:
             if entry_token and entry_token == live_token:
                 matches_live_head = True
-            elif live_run_id and entry_run_id == live_run_id:
-                matches_live_head = True
-            elif live_approval_id and entry_approval_id == live_approval_id:
+            elif (
+                live_approval_id
+                and live_run_id
+                and entry_approval_id == live_approval_id
+                and entry_run_id == live_run_id
+            ):
                 matches_live_head = True
 
         if entry_run_id:
@@ -177,6 +183,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
                 rebuilt.append(entry)
                 live_mirror_present = True
                 continue
+            if live_token:
+                if entry_token:
+                    changed = True
+                    continue
+                deferred_run_entries.append(entry)
+                continue
             if not entry_token:
                 rebuilt.append(entry)
                 continue
@@ -196,13 +208,19 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
     if live_token and not live_mirror_present:
         mirror_entry = dict(live_head_data)
         mirror_run_id = str(mirror_entry.get("run_id") or "").strip()
-        mirror_entry.setdefault("approval_id", f"gwrun:{mirror_run_id}" if mirror_run_id else uuid.uuid4().hex)
+        mirror_entry.setdefault(
+            "approval_id",
+            f"gwrun:{mirror_run_id}:{live_token}" if mirror_run_id else uuid.uuid4().hex,
+        )
         mirror_entry[_GATEWAY_MIRROR_FLAG] = True
         mirror_entry[_GATEWAY_MIRROR_TOKEN] = live_token
         rebuilt.append(mirror_entry)
         live_mirror_present = True
         changed = True
 
+    if deferred_run_entries:
+        rebuilt.extend(deferred_run_entries)
+
     if rebuilt:
         if rebuilt != queue_list:
             _pending[session_key] = rebuilt
@@ -224,14 +242,25 @@ def _gateway_pending_mirror_locked(session_key: str, approval_id: str = "", run_
     queue = _pending.get(session_key)
     entries = queue if isinstance(queue, list) else [queue] if queue else []
     if approval_id:
+        matched_entry: dict | None = None
         for entry in entries:
-            if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip():
+            if not _is_gateway_mirror_entry(entry):
                 continue
-            if entry.get("approval_id") == approval_id:
-                if run_id and entry.get("run_id") != run_id:
+            if entry.get("approval_id") != approval_id:
+                continue
+            entry_run_id = str(entry.get("run_id") or "").strip()
+            if not entry_run_id:
+                if not run_id:
                     return None
+                continue
+            if run_id and entry_run_id != run_id:
+                continue
+            if run_id:
                 return entry
+            if matched_entry is not None:
                 return None
+            matched_entry = entry
+        return matched_entry
     for entry in entries:
         if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip():
             continue
@@ -249,21 +278,47 @@ def gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str
 
 
 def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str = "") -> bool:
-    """Retire only the exact run-backed mirror and notify its new queue head."""
+    """Retire one approval, 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, approval_id, run_id)
-        if not match:
+            retired = [match] if match else []
+        else:
+            retired = [
+                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:
+                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:
             return False
+        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:
             _pending.pop(session_key, None)
-        head = entries[0] if entries else None
-        _approval_sse_notify_locked(session_key, head, len(entries))
+        head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
+        _approval_sse_notify_locked(session_key, head, total)
     publish_session_list_changed("attention_resolved")
     return True
 
@@ -280,33 +335,57 @@ def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str
     return None
 
 
-def _gateway_mirrored_pending_approval_id_by_run_id(session_key: str, run_id: str) -> str | None:
-    """Return the mirrored approval_id for a matching gateway run."""
-    run_id = str(run_id or "").strip()
-    if not run_id:
-        return None
-    with _lock:
-        entry = _gateway_pending_mirror_locked(session_key, run_id=run_id)
-        if entry:
-            approval_id = str(entry.get("approval_id") or "").strip()
-            return approval_id or None
-    return None
-
-
 def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None:
     """Mirror the live gateway head into WebUI polling state under a typed tag."""
     with _lock:
         run_id = str(approval.get("run_id") or "").strip()
         if run_id:
             live_gateway_queue = _gateway_queues.get(session_key) or []
-            if not live_gateway_queue:
+            approval_id = str(approval.get("approval_id") or "").strip()
+            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 = (
+                _gateway_mirror_entry_token(live_head_entry)
+                if live_head_entry and live_head_data
+                else None
+            )
+            if (
+                live_token
+                and live_head_run_id == run_id
+                and (
+                    not approval_id
+                    or not live_head_approval_id
+                    or live_head_approval_id == approval_id
+                )
+            ):
+                if approval_id:
+                    live_head_data["approval_id"] = approval_id
+                else:
+                    approval_id = live_head_approval_id
+                    if not approval_id:
+                        approval_id = f"gwrun:{run_id}:{live_token}"
+                        live_head_data["approval_id"] = approval_id
+                    approval["approval_id"] = approval_id
+            else:
+                if not approval_id:
+                    approval_id = f"gwrun:{run_id}:{uuid.uuid4().hex}"
+                    approval["approval_id"] = approval_id
                 mirror_entry = dict(approval)
                 mirror_entry["run_id"] = run_id
-                mirror_entry["approval_id"] = str(mirror_entry.get("approval_id") or f"gwrun:{run_id}").strip()
+                mirror_entry["approval_id"] = approval_id
                 mirror_entry[_GATEWAY_MIRROR_FLAG] = True
-                if not _gateway_pending_mirror_locked(session_key, approval_id=mirror_entry["approval_id"], run_id=run_id):
+                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()
         _approval_sse_notify_locked(session_key, head, total)
     publish_session_list_changed("attention_pending")
 

api/routes.py

@@ -13368,6 +13368,46 @@ def handle_get(handler, parsed) -> bool:
             return bad(handler, "stream_id required")
         if not _stream_id_visible_to_request_profile(handler, stream_id):
             return True
+        gateway_stop_blocked = False
+        try:
+            from api.gateway_chat import _STREAM_RUN_IDS, gateway_run_id_pending, stop_gateway_run
+
+            run_id = _STREAM_RUN_IDS.get(stream_id)
+            pending_run_id = False
+            if not run_id:
+                import time as _time
+
+                pending_run_id = gateway_run_id_pending(stream_id)
+                attempts_remaining = 10
+                while not run_id and pending_run_id and attempts_remaining > 0:
+                    _time.sleep(0.05)
+                    run_id = _STREAM_RUN_IDS.get(stream_id)
+                    pending_run_id = gateway_run_id_pending(stream_id)
+                    attempts_remaining -= 1
+            if run_id:
+                if stop_gateway_run(stream_id):
+                    owner_sid = stream_owner_session_id(stream_id)
+                    if owner_sid:
+                        retire_gateway_pending_mirror(owner_sid, run_id=run_id)
+                else:
+                    gateway_stop_blocked = True
+            elif pending_run_id:
+                gateway_stop_blocked = True
+        except Exception:
+            logger.debug("Failed to stop gateway run during chat cancellation", exc_info=True)
+            gateway_stop_blocked = True
+        if gateway_stop_blocked:
+            return j(
+                handler,
+                {
+                    "ok": False,
+                    "cancelled": False,
+                    "stream_id": stream_id,
+                    "error": "Gateway stop failed",
+                },
+                status=502,
+            )
+
         from api.runtime_adapter import LegacyJournalRuntimeAdapter, runtime_adapter_enabled
 
         if runtime_adapter_enabled():
@@ -23749,7 +23789,7 @@ def _handle_workspace_reorder(handler, body):
     return j(handler, {"ok": True, "workspaces": reordered})
 
 
-def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
+def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: str = "") -> bool:
     """Resolve an approval through the existing callback path.
 
     Slice 3b keeps the RuntimeAdapter as a protocol translator: it delegates to
@@ -23760,6 +23800,7 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
     pending = None
     found_target = False
     gateway_keys = []
+    gateway_head_matches_target = False
     with _lock:
         reconcile_gateway_pending_mirror_locked(sid)
         queue = _pending.get(sid)
@@ -23767,7 +23808,10 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
             if approval_id:
                 # Find and remove the specific entry by approval_id.
                 for i, entry in enumerate(queue):
-                    if entry.get("approval_id") == approval_id:
+                    if entry.get("approval_id") != approval_id:
+                        continue
+                    if run_id and str(entry.get("run_id") or "").strip() != run_id:
+                        continue
                     pending = queue.pop(i)
                     found_target = True
                     break
@@ -23783,7 +23827,13 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
                 _pending.pop(sid, None)
         elif queue:
             # Legacy single-dict value.
-            if not approval_id or queue.get("approval_id") == approval_id:
+            if (
+                not approval_id
+                or (
+                    queue.get("approval_id") == approval_id
+                    and (not run_id or str(queue.get("run_id") or "").strip() == run_id)
+                )
+            ):
                 pending = _pending.pop(sid, None)
                 found_target = pending is not None
         # When no _pending entry found AND no explicit approval_id was
@@ -23806,6 +23856,17 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
                 # idempotent over the session key set so the outcome is
                 # the same regardless of which entry wins the race.
                 found_target = True
+        elif approval_id:
+            gw_queue = _gateway_queues.get(sid)
+            if gw_queue and len(gw_queue) > 0:
+                gw_entry = gw_queue[0]
+                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
         # 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
@@ -23837,7 +23898,17 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
     # 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
-    if found_target or not approval_id:
+    should_resolve_gateway = (
+        not approval_id
+        or (
+            found_target
+            and (
+                not run_id
+                or gateway_head_matches_target
+            )
+        )
+    )
+    if should_resolve_gateway:
         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.
@@ -23920,20 +23991,50 @@ def _handle_approval_respond(handler, body):
             active_sid = getattr(s, "active_stream_id", None)
             if active_sid:
                 _candidate_run_id = _STREAM_RUN_IDS.get(active_sid)
-        matched_mirror = gateway_pending_mirror(sid, approval_id=approval_id, run_id=_candidate_run_id)
-        _run_id = matched_mirror["run_id"] if matched_mirror else None
-        if not matched_mirror and approval_id:
+        local_match = False
+        run_backed_gateway_matches = 0
+        if approval_id:
             with _lock:
                 queue = _pending.get(sid)
                 entries = queue if isinstance(queue, list) else [queue] if queue else []
                 local_match = any(
                     isinstance(entry, dict)
                     and entry.get("approval_id") == approval_id
-                    and not entry.get(_GATEWAY_MIRROR_FLAG)
+                    and (
+                        not entry.get(_GATEWAY_MIRROR_FLAG)
+                        or not str(entry.get("run_id") or "").strip()
+                    )
                     for entry in entries
                 )
+                run_backed_gateway_matches = sum(
+                    1
+                    for entry in entries
+                    if isinstance(entry, dict)
+                    and entry.get("approval_id") == approval_id
+                    and entry.get(_GATEWAY_MIRROR_FLAG)
+                    and str(entry.get("run_id") or "").strip()
+                )
             if local_match:
                 _candidate_run_id = None
+        matched_mirror = gateway_pending_mirror(
+            sid, approval_id=approval_id, run_id=_candidate_run_id
+        ) if approval_id else None
+        _run_id = matched_mirror["run_id"] if matched_mirror else None
+        if not matched_mirror and approval_id:
+            if local_match:
+                _candidate_run_id = None
+            elif run_backed_gateway_matches > 1 and not _candidate_run_id:
+                return j(
+                    handler,
+                    {
+                        "ok": False,
+                        "choice": choice,
+                        "relayed": False,
+                        "code": "gateway_run_unavailable",
+                        "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE,
+                    },
+                    status=409,
+                )
         if _run_id:
             from api.runner_client import HttpRunnerClient, RunnerClientError
             _cfg = _get_config()
@@ -23949,10 +24050,10 @@ def _handle_approval_respond(handler, body):
             # still needs the same cleanup path so the parked entry, mirrored
             # card, and agent signal all settle here too.
             cleanup_approval_id = matched_mirror["approval_id"]
-            _resolve_approval_legacy(sid, cleanup_approval_id, choice)
+            _resolve_approval_legacy(sid, cleanup_approval_id, choice, run_id=_run_id)
             retire_gateway_pending_mirror(sid, approval_id=cleanup_approval_id, run_id=_run_id)
             return j(handler, {"ok": True, "choice": choice, "relayed": True})
-        if _candidate_run_id or approval_id.startswith("gwrun:"):
+        if _candidate_run_id:
             return j(handler, {"ok": False, "choice": choice, "relayed": False,
                                "code": "gateway_run_unavailable",
                                "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE}, status=409)

static/boot.js

@@ -17,7 +17,7 @@
 async function cancelStream(reason){
   const sid = S.session && S.session.session_id;
   const streamId = S.activeStreamId;
-  if(!streamId) return;
+  if(!streamId) return false;
   // Interrupt provenance: log WHY the active run is being cancelled so operators
   // can tell an explicit Stop / interrupt from any other trigger when they see a
   // SIGINT/exit-code-130 in the backend logs. Only explicit user paths reach
@@ -30,8 +30,10 @@ async function cancelStream(reason){
     console.info('[stream] cancel requested', {reason:_reason, streamId, sessionId:sid});
   }
   let respBody=null;
+  let respOk=false;
   try{
     const r=await fetch(new URL(`api/chat/cancel?stream_id=${encodeURIComponent(streamId)}`,document.baseURI||location.href).href,{credentials:'include'});
+    respOk=!!(r&&r.ok);
     try{respBody=await r.json();}catch(_){}
   }catch(e){
     if(typeof console !== 'undefined' && console.warn){
@@ -50,7 +52,7 @@ async function cancelStream(reason){
   // `cancel` event can clear INFLIGHT, render "Task cancelled", and refresh
   // the sidebar. Only clear locally when the backend says there is no active
   // stream left to settle.
-  if(respBody && respBody.cancelled===false && S.activeStreamId===streamId){
+  if(respOk && respBody && respBody.cancelled===false && S.activeStreamId===streamId){
     S.activeStreamId=null;
     setBusy(false);
     if(typeof setComposerStatus==='function') setComposerStatus('');
@@ -59,20 +61,24 @@ async function cancelStream(reason){
     // distinguish reasons — keep the toast generic and short.
     if(typeof showToast==='function') showToast('Stream is no longer active',2000);
   }
+  return respOk;
 }
 
 async function cancelSessionStream(session){
   const streamId = session&&session.active_stream_id;
   const sid = session&&session.session_id;
-  if(!streamId||!sid) return;
+  if(!streamId||!sid) return false;
   // Explicit sidebar "Stop response" — log provenance for the same reason as
   // cancelStream(). (#5345)
   if(typeof console !== 'undefined' && console.info){
     console.info('[stream] cancel requested', {reason:'sidebar-stop', streamId, sessionId:sid});
   }
+  let respOk=false;
   try{
-    await fetch(new URL(`api/chat/cancel?stream_id=${encodeURIComponent(streamId)}`,document.baseURI||location.href).href,{credentials:'include'});
+    const r=await fetch(new URL(`api/chat/cancel?stream_id=${encodeURIComponent(streamId)}`,document.baseURI||location.href).href,{credentials:'include'});
+    respOk=!!(r&&r.ok);
   }catch(e){/* close local stream; keep UI state honest below */}
+  if(!respOk) return false;
   if(typeof closeLiveStream==='function') closeLiveStream(sid, streamId);
   session.active_stream_id=null;
   delete INFLIGHT[sid];
@@ -94,6 +100,7 @@ async function cancelSessionStream(session){
     hideClarifyCard(true, 'cancelled');
   }
   if(typeof renderSessionList==='function') renderSessionList();
+  return true;
 }
 
 async function _savedSessionShouldStaySidebarOnly(sid){

static/commands.js

@@ -1213,7 +1213,10 @@ async function cmdPersonality(args){
 async function cmdStop(){
   if(!S.session){showToast(t('no_active_session'));return;}
   if(!S.activeStreamId){showToast(t('no_active_task'));return;}
-  if(typeof cancelStream==='function'){await cancelStream('slash-stop');showToast(t('stream_stopped'));}
+  if(typeof cancelStream==='function'){
+    if(await cancelStream('slash-stop')) showToast(t('stream_stopped'));
+    else showToast(t('cancel_failed'));
+  }
   else showToast(t('cancel_unavailable'));
 }
 
@@ -1319,8 +1322,10 @@ async function cmdInterrupt(args){
   updateQueueBadge(S.session.session_id);
   S.pendingFiles=[];renderTray();
   // Cancel the active stream; setBusy(false) will drain the queue
-  if(typeof cancelStream==='function'){await cancelStream('slash-interrupt');}
-  showToast(t('cmd_interrupt_confirm'),2000);
+  if(typeof cancelStream==='function'){
+    if(await cancelStream('slash-interrupt')) showToast(t('cmd_interrupt_confirm'),2000);
+    else showToast(t('cancel_failed'));
+  }
 }
 
 /**

static/messages.js

@@ -1395,8 +1395,8 @@ async function send(){
         _clearComposerAfterQueuedSelectionSend(S.session&&S.session.session_id);
         S.pendingFiles=[];renderTray();
         if(S.activeStreamId&&typeof cancelStream==='function'){
-          showToast(t('busy_interrupt_confirm'),2000);
-          await cancelStream('busy-interrupt');
+          if(await cancelStream('busy-interrupt')) showToast(t('busy_interrupt_confirm'),2000);
+          else showToast(t('cancel_failed'));
         } else {
           showToast(`Queued: "${text.slice(0,40)}${text.length>40?'…':''}"`,2000);
         }

static/sessions.js

@@ -4905,8 +4905,8 @@ function _openSessionActionMenu(session, anchorEl){
       ICONS.stop,
       async()=>{
         closeSessionActionMenu();
-        await cancelSessionStream(session);
-        showToast(t('stream_stopped'));
+        if(await cancelSessionStream(session)) showToast(t('stream_stopped'));
+        else showToast(t('cancel_failed'));
       }
     ));
   }

tests/test_cancel_stream_owner_guard.py

@@ -135,6 +135,10 @@ class TestCancelStreamOwnerGuardStructural:
         """When the backend reports ``cancelled:false``, the fix should
         show a lightweight toast (the backend may have already finalized
         the turn, or a newer turn may hold the session lock)."""
+        assert "r.ok" in CANCEL_STREAM_SRC, (
+            "cancelStream() must gate cancelled:false cleanup on an HTTP-success "
+            "response so failed gateway stops do not settle the UI locally"
+        )
         assert "cancelled" in CANCEL_STREAM_SRC and "false" in CANCEL_STREAM_SRC, (
             "cancelStream() must check respBody.cancelled === false and surface "
             "a toast (e.g. 'Stream is no longer active')"
@@ -254,6 +258,25 @@ async function runAll() {
     toastCalls: [...M.toastCalls],
   };
 
+  // T3b — failed HTTP response with cancelled:false body: keep the active
+  // owner path intact because the cancel did not succeed.
+  reset();
+  globalThis.S = { activeStreamId: 'stream-1', session: { session_id: 'sid-1' } };
+  _fetchResponse = {
+    ok: false,
+    json: () => Promise.resolve({ ok: false, cancelled: false, stream_id: 'stream-1' }),
+  };
+  _fetchThrows = false;
+  await cancelStream();
+  out.t3b_failed_http = {
+    finalActiveStreamId: globalThis.S.activeStreamId,
+    fetchCalls: M.fetchCalls.length,
+    closeCalls: [...M.closeCalls],
+    busyCalls: [...M.busyCalls],
+    composerCalls: [...M.composerCalls],
+    toastCalls: [...M.toastCalls],
+  };
+
   // T4 — network error: no throw, active owner kept because we do not know
   // whether the cancel landed.
   reset();
@@ -466,6 +489,26 @@ class TestCancelStreamOwnerGuardRuntime:
             f"got {r['toastCalls']}"
         )
 
+    def test_t3b_failed_http_keeps_owner_state(self, runtime_results):
+        r = runtime_results["t3b_failed_http"]
+        assert r["fetchCalls"] == 1, (
+            f"cancelStream() must still issue the cancel request on failed HTTP, "
+            f"got {r['fetchCalls']}"
+        )
+        assert r["finalActiveStreamId"] == "stream-1", (
+            f"cancelStream() must keep S.activeStreamId when /api/chat/cancel "
+            f"fails, got {r['finalActiveStreamId']!r}"
+        )
+        assert r["closeCalls"] == [], (
+            f"cancelStream() must keep owned SSE open on failed HTTP, got {r['closeCalls']}"
+        )
+        assert r["busyCalls"] == [], (
+            f"cancelStream() must not call setBusy(false) on failed HTTP, got {r['busyCalls']}"
+        )
+        assert r["toastCalls"] == [], (
+            f"cancelStream() should not show a toast on failed HTTP, got {r['toastCalls']}"
+        )
+
     def test_t5_owner_guard_preserves_new_turn(self, runtime_results):
         r = runtime_results["t5_owner_guard"]
         # The fetch still happened for the OLD stream.
@@ -512,12 +555,14 @@ class TestCancelStreamOwnerGuardRuntime:
             )
         assert runtime_results["t3_cancelled_false"]["finalActiveStreamId"] is None
         assert runtime_results["t3_cancelled_false"]["busyCalls"] == [False]
+        assert runtime_results["t3b_failed_http"]["finalActiveStreamId"] == "stream-1"
+        assert runtime_results["t3b_failed_http"]["busyCalls"] == []
         # T5 is the owner-rotation case and must preserve the new turn.
         assert runtime_results["t5_owner_guard"]["finalActiveStreamId"] == "stream-2"
         assert runtime_results["t5_owner_guard"]["busyCalls"] == []
         # Only the stale-owner path closes the old SSE; active owned paths
         # keep it open so the backend terminal event can settle/render.
-        for key in ("t2_happy_path", "t3_cancelled_false", "t4_network_error", "t5_owner_guard"):
+        for key in ("t2_happy_path", "t3_cancelled_false", "t3b_failed_http", "t4_network_error", "t5_owner_guard"):
             if key == "t5_owner_guard":
                 assert runtime_results[key]["closeCalls"] == [["sid-1", "stream-1"]], (
                     f"{key} must close stale old SSE for ('sid-1', 'stream-1'), "

tests/test_issue4771_local_approval_regression.py

@@ -40,6 +40,7 @@ import pytest
 # Import order matters: tools.* will not resolve until api.config has run.
 from api import routes
 from api import models
+from api.gateway_chat import _STREAM_RUN_IDS
 
 try:
     import tools.approval as ta
@@ -182,6 +183,101 @@ def test_local_mirrored_approval_deny_resolves():
         _cleanup(sid)
 
 
+def test_local_mirrored_approval_with_gwrun_prefix_still_resolves():
+    """A local approval id that happens to start with gwrun: stays local."""
+    sid = f"local-approval-prefixed-{uuid.uuid4().hex[:8]}"
+    _register_session(sid)
+    try:
+        entry, _approval_id = _seed_local_pending_approval(sid)
+        with ta._lock:
+            head = ta._pending[sid][0]
+            head["approval_id"] = "gwrun:local-test"
+        handler = _FakeHandler()
+        with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False):
+            routes._handle_approval_respond(
+                handler,
+                {"session_id": sid, "choice": "once", "approval_id": "gwrun:local-test"},
+            )
+        resp = handler.json()
+        assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
+        assert resp.get("ok") is True
+        assert resp.get("code") != "gateway_run_unavailable"
+        assert entry.event.is_set()
+        assert entry.result == "once"
+    finally:
+        _cleanup(sid)
+
+
+def test_local_live_head_does_not_inherit_stale_remote_run():
+    """A stale remote mirror cannot seize a live local approval with the same id."""
+    sid = f"local-approval-collision-{uuid.uuid4().hex[:8]}"
+    _register_session(sid)
+    try:
+        entry, approval_id = _seed_local_pending_approval(sid)
+        with ta._lock:
+            local_queue = ta._pending.get(sid)
+            local_entries = list(local_queue) if isinstance(local_queue, list) else [local_queue]
+            ta._pending[sid] = [{
+                ra._GATEWAY_MIRROR_FLAG: True,
+                "run_id": "remote-run",
+                "approval_id": approval_id,
+                "command": "stale-remote",
+            }, *local_entries]
+        handler = _FakeHandler()
+        with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval:
+            routes._handle_approval_respond(
+                handler,
+                {"session_id": sid, "choice": "once", "approval_id": approval_id},
+            )
+        resp = handler.json()
+        assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
+        assert resp.get("ok") is True
+        assert resp.get("relayed") is not True
+        respond_approval.assert_not_called()
+        assert entry.event.is_set()
+        assert entry.result == "once"
+    finally:
+        _cleanup(sid)
+
+
+def test_local_live_head_beats_active_remote_candidate_run():
+    """A local approval still wins when an active remote run shares its id."""
+    sid = f"local-approval-active-remote-{uuid.uuid4().hex[:8]}"
+    session = _register_session(sid)
+    stream_id = f"remote-stream-{uuid.uuid4().hex[:8]}"
+    session.active_stream_id = stream_id
+    _STREAM_RUN_IDS[stream_id] = "remote-run"
+    try:
+        entry, approval_id = _seed_local_pending_approval(sid)
+        with ta._lock:
+            local_queue = ta._pending.get(sid)
+            local_entries = list(local_queue) if isinstance(local_queue, list) else [local_queue]
+            ta._pending[sid] = [{
+                ra._GATEWAY_MIRROR_FLAG: True,
+                "run_id": "remote-run",
+                "approval_id": approval_id,
+                "command": "remote-collision",
+            }, *local_entries]
+        handler = _FakeHandler()
+        with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval:
+            routes._handle_approval_respond(
+                handler,
+                {"session_id": sid, "choice": "once", "approval_id": approval_id},
+            )
+        resp = handler.json()
+        assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
+        assert resp.get("ok") is True
+        assert resp.get("relayed") is not True
+        respond_approval.assert_not_called()
+        assert entry.event.is_set()
+        assert entry.result == "once"
+    finally:
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _cleanup(sid)
+
+
 def test_gateway_mirrored_approval_without_run_still_409s():
     """#4771 behaviour preserved: on the gateway backend, a mirrored approval
     whose run is gone must still surface the actionable 409 so the card stays

tests/test_issue6008_sidebar_stop_failure_ui.py

@@ -0,0 +1,144 @@
+"""Sidebar stop must not settle the UI when /api/chat/cancel fails."""
+from __future__ import annotations
+
+import json
+import pathlib
+import re
+import subprocess
+
+
+REPO = pathlib.Path(__file__).parent.parent
+BOOT_JS = (REPO / "static" / "boot.js").read_text(encoding="utf-8")
+COMMANDS_JS = (REPO / "static" / "commands.js").read_text(encoding="utf-8")
+MESSAGES_JS = (REPO / "static" / "messages.js").read_text(encoding="utf-8")
+SESSIONS_JS = (REPO / "static" / "sessions.js").read_text(encoding="utf-8")
+
+
+def _extract_function(src: str, name: str) -> str:
+    m = re.search(rf"async function {name}\s*\(", src)
+    assert m, f"{name} not found in static/boot.js"
+    brace_pos = src.index("{", m.end())
+    depth = 1
+    pos = brace_pos + 1
+    while pos < len(src) and depth > 0:
+        ch = src[pos]
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+        pos += 1
+    return src[m.start():pos]
+
+
+CANCEL_SESSION_STREAM_SRC = _extract_function(BOOT_JS, "cancelSessionStream")
+
+
+def test_source_gates_sidebar_settle_on_http_success():
+    assert "return false" in CANCEL_SESSION_STREAM_SRC
+    assert "return true" in CANCEL_SESSION_STREAM_SRC
+    assert "r.ok" in CANCEL_SESSION_STREAM_SRC, (
+        "cancelSessionStream() must check the /api/chat/cancel HTTP status before "
+        "closing local UI state"
+    )
+    assert "if(!respOk)returnfalse;" in "".join(CANCEL_SESSION_STREAM_SRC.split()), (
+        "cancelSessionStream() must bail out on failed stop responses"
+    )
+
+
+def test_stop_callers_gate_success_toasts_on_cancel_result():
+    compact_commands = "".join(COMMANDS_JS.split())
+    compact_messages = "".join(MESSAGES_JS.split())
+    compact_sessions = "".join(SESSIONS_JS.split())
+    assert "if(awaitcancelStream('slash-stop'))showToast(t('stream_stopped'))" in compact_commands
+    assert "elseshowToast(t('cancel_failed'))" in compact_commands
+    assert "if(awaitcancelStream('slash-interrupt'))showToast(t('cmd_interrupt_confirm'),2000)" in compact_commands
+    assert "if(awaitcancelStream('busy-interrupt'))showToast(t('busy_interrupt_confirm'),2000)" in compact_messages
+    assert "if(awaitcancelSessionStream(session))showToast(t('stream_stopped'))" in compact_sessions
+
+
+_NODE_SCRIPT = r'''
+const M = {
+  closeCalls: [],
+  busyCalls: [],
+  composerCalls: [],
+  statusCalls: [],
+  renderCalls: 0,
+  clearCalls: [],
+  approvalStops: 0,
+  approvalHides: 0,
+  clarifyStops: 0,
+  clarifyHides: 0,
+  fetchCalls: [],
+};
+
+globalThis.INFLIGHT = { 'sid-1': { streamId: 'stream-1' } };
+globalThis.S = { activeStreamId: 'stream-1', session: { session_id: 'sid-1', active_stream_id: 'stream-1' } };
+globalThis.closeLiveStream = (...a) => M.closeCalls.push(a);
+globalThis.clearInflightState = (sid) => M.clearCalls.push(['clearInflightState', sid]);
+globalThis.clearInflight = () => M.clearCalls.push(['clearInflight']);
+globalThis.setBusy = (v) => M.busyCalls.push(v);
+globalThis.setComposerStatus = (v) => M.composerCalls.push(v);
+globalThis.setStatus = (v) => M.statusCalls.push(v);
+globalThis.stopApprovalPolling = () => M.approvalStops += 1;
+globalThis.hideApprovalCard = () => M.approvalHides += 1;
+globalThis.stopClarifyPolling = () => M.clarifyStops += 1;
+globalThis.hideClarifyCard = () => M.clarifyHides += 1;
+globalThis.renderSessionList = () => M.renderCalls += 1;
+globalThis._approvalSessionId = 'sid-1';
+globalThis._clarifySessionId = 'sid-1';
+globalThis.document = { baseURI: 'http://localhost:8787/' };
+globalThis.location = { href: 'http://localhost:8787/' };
+globalThis.fetch = (url, opts) => {
+  M.fetchCalls.push({ url: String(url), opts });
+  return Promise.resolve({
+    ok: false,
+    json: () => Promise.resolve({ ok: false, cancelled: false, stream_id: 'stream-1' }),
+  });
+};
+
+__CANCEL_SESSION_STREAM_SRC__
+
+const session = { session_id: 'sid-1', active_stream_id: 'stream-1' };
+await cancelSessionStream(session);
+console.log(JSON.stringify({
+  sessionActiveStreamId: session.active_stream_id,
+  activeStreamId: globalThis.S.activeStreamId,
+  closeCalls: M.closeCalls,
+  busyCalls: M.busyCalls,
+  composerCalls: M.composerCalls,
+  renderCalls: M.renderCalls,
+  clearCalls: M.clearCalls,
+  approvalStops: M.approvalStops,
+  approvalHides: M.approvalHides,
+  clarifyStops: M.clarifyStops,
+  clarifyHides: M.clarifyHides,
+  fetchCalls: M.fetchCalls.length,
+}));
+'''
+
+
+def test_failed_sidebar_stop_keeps_local_state():
+    script = _NODE_SCRIPT.replace("__CANCEL_SESSION_STREAM_SRC__", CANCEL_SESSION_STREAM_SRC)
+    completed = subprocess.run(
+        ["node", "--input-type=module", "-e", script],
+        cwd=str(REPO),
+        text=True,
+        capture_output=True,
+        timeout=30,
+    )
+    assert completed.returncode == 0, (
+        f"node subprocess failed:\n--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}"
+    )
+    result = json.loads(completed.stdout.splitlines()[-1])
+    assert result["fetchCalls"] == 1
+    assert result["sessionActiveStreamId"] == "stream-1"
+    assert result["activeStreamId"] == "stream-1"
+    assert result["closeCalls"] == []
+    assert result["busyCalls"] == []
+    assert result["composerCalls"] == []
+    assert result["renderCalls"] == 0
+    assert result["clearCalls"] == []
+    assert result["approvalStops"] == 0
+    assert result["approvalHides"] == 0
+    assert result["clarifyStops"] == 0
+    assert result["clarifyHides"] == 0