Merge pull request #6446 from nesquena/stage/6040

Release: gateway approvals routed by run ID (#6040, fix #6008)

CHANGELOG.md

@@ -23,6 +23,8 @@
 
 ### Fixed
 
+- **Gateway approval cards are now routed by run ID, so approvals and Stop always act on the right run (#6008).** A gateway approval card that arrived without an `approval_id` couldn't be acted on, and with multiple concurrent runs an approval or a Stop could be applied to the wrong run. Approvals are now owned per-`(session, run)`: identity-less and identity-bearing approvals route through the same single-flight owner, concurrent duplicate submits can't double-relay, different runs don't block each other, and a failed relay releases ownership so a retry can proceed. The composer/sidebar Stop now waits boundedly for an acknowledged exact-run gateway stop and surfaces a persistent, actionable "Cancel failed." error toast (with Copy and Dismiss, localized) instead of silently settling the UI when the stop fails. When multiple approvals are queued the card shows a "1 of N pending" counter. Thanks @rodboev. (#6040, #6008)
+
 - **Streaming prose no longer appears one character short before a tool card in Compact Worklog.** While a turn was streaming, the process-prose immediately preceding a tool card could render with its last character missing until the whole turn settled — the streaming-markdown renderer holds the newest character back as lookahead until its parser is finalized, and Anchor-owned prose rows never finalized theirs. The cached parser is now finalized once when the row becomes `completed` (running the same media/link/table cleanup the main answer body already does), so the character appears immediately. Incremental O(n) rendering is preserved and an edit/re-render after finalize self-heals by rebuilding the parser. Thanks @starship-s. (#6408)
 
 - **Security: public share links no longer leak local attachments (#6174).** A public share snapshot could carry `MEDIA:<local path>` references that the public renderer turned into authenticated `/api/media` requests, exposing local workspace/attachment files (and, via crafted paths, files outside them) to anyone with the link. Share snapshots now embed only images that resolve *inside* the session workspace or the attachment root, as inline base64 fixed at build time — enforced by a path allow-list (`is_relative_to`, so `..` traversal and symlink escapes are rejected), a 512 KiB size cap, an image-only MIME allow-list, and magic-byte content validation (an extension-spoofed file is rejected). `file://`, absolute paths outside the allowed roots, SVG, non-image files, and the empty-allowed-roots case all fail closed to a static placeholder — no local bytes ever reach the public page. Public `https://` image URLs still pass through unchanged. Thanks @webtecnica. (#6319, #6174)

api/config.py

@@ -8897,6 +8897,7 @@ def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
     caps = {
         "approval_events": False,
         "run_approval_response": False,
+        "approval_identity_v1": False,
         "capabilities_reachable": False,
         "probe_error": None,
         "fetched_at": 0.0,
@@ -8914,6 +8915,7 @@ def get_gateway_caps(base_url: str, api_key: str = "") -> dict:
             features = {}
         caps["approval_events"] = bool(features.get("approval_events"))
         caps["run_approval_response"] = bool(features.get("run_approval_response"))
+        caps["approval_identity_v1"] = bool(features.get("approval_identity_v1"))
     except urllib.error.HTTPError as exc:
         caps["capabilities_reachable"] = True
         caps["probe_error"] = f"{type(exc).__name__}: {exc}"
@@ -8953,6 +8955,11 @@ def gateway_supports_approval(base_url: str, api_key: str = "") -> bool:
     return bool(caps.get("approval_events") and caps.get("run_approval_response"))
 
 
+def gateway_supports_approval_identity_v1(base_url: str, api_key: str = "") -> bool:
+    """True only when Gateway advertises authoritative approval identities."""
+    return bool(get_gateway_caps(base_url, api_key).get("approval_identity_v1"))
+
+
 def invalidate_gateway_caps(base_url: str | None = None) -> None:
     """Evict capability cache for base_url, or all entries when base_url is None."""
     with _GATEWAY_CAPS_LOCK:

api/gateway_chat.py

@@ -8,6 +8,7 @@ import threading
 import time
 import uuid
 import urllib.error
+import urllib.parse
 import urllib.request
 from typing import Any
 
@@ -39,6 +40,110 @@ logger = logging.getLogger(__name__)
 
 # Maps stream_id -> gateway run_id for approval response relay.
 _STREAM_RUN_IDS: dict[str, str] = {}
+_STREAM_RUN_LIFECYCLE: dict[str, dict[str, Any]] = {}
+_STREAM_RUN_STARTING_CONDITION = threading.Condition()
+GATEWAY_RUN_ID_WAIT_TIMEOUT = 5.0
+
+
+def _mark_gateway_run_starting(stream_id: str) -> None:
+    with _STREAM_RUN_STARTING_CONDITION:
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _STREAM_RUN_LIFECYCLE[stream_id] = {
+            "phase": "pending",
+            "run_id": "",
+            "waiters": 0,
+            "owner_done": False,
+        }
+
+
+def _publish_gateway_run_id(stream_id: str, run_id: str) -> None:
+    with _STREAM_RUN_STARTING_CONDITION:
+        _STREAM_RUN_IDS[stream_id] = run_id
+        state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
+        _STREAM_RUN_LIFECYCLE[stream_id] = {
+            "phase": "ready",
+            "run_id": run_id,
+            "waiters": int(state.get("waiters") or 0),
+            "owner_done": bool(state.get("owner_done")),
+        }
+        _STREAM_RUN_STARTING_CONDITION.notify_all()
+
+
+def _finish_gateway_run_starting(stream_id: str, *, result: str = "failed") -> None:
+    with _STREAM_RUN_STARTING_CONDITION:
+        state = _STREAM_RUN_LIFECYCLE.get(stream_id) or {}
+        if str(state.get("phase") or "").strip().lower() == "ready":
+            return
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _STREAM_RUN_LIFECYCLE[stream_id] = {
+            "phase": "fallback" if result == "fallback" else "failed",
+            "run_id": "",
+            "waiters": int(state.get("waiters") or 0),
+            "owner_done": bool(state.get("owner_done")),
+        }
+        _STREAM_RUN_STARTING_CONDITION.notify_all()
+
+
+def _retire_gateway_run_starting_if_done(stream_id: str) -> bool:
+    state = _STREAM_RUN_LIFECYCLE.get(stream_id)
+    if not state:
+        return False
+    if int(state.get("waiters") or 0) > 0:
+        return False
+    if not bool(state.get("owner_done")):
+        return False
+    _STREAM_RUN_LIFECYCLE.pop(stream_id, None)
+    _STREAM_RUN_IDS.pop(stream_id, None)
+    return True
+
+
+def _clear_gateway_run_starting(stream_id: str) -> None:
+    with _STREAM_RUN_STARTING_CONDITION:
+        state = _STREAM_RUN_LIFECYCLE.get(stream_id)
+        if state:
+            state["owner_done"] = True
+        _retire_gateway_run_starting_if_done(stream_id)
+        _STREAM_RUN_STARTING_CONDITION.notify_all()
+
+
+def gateway_run_id_pending(stream_id: str) -> bool:
+    with _STREAM_RUN_STARTING_CONDITION:
+        return str((_STREAM_RUN_LIFECYCLE.get(stream_id) or {}).get("phase") or "").strip().lower() == "pending"
+
+
+def wait_for_gateway_run_id(stream_id: str, timeout: float) -> tuple[bool, str | None]:
+    deadline = time.monotonic() + max(0.0, float(timeout))
+    with _STREAM_RUN_STARTING_CONDITION:
+        state = _STREAM_RUN_LIFECYCLE.get(stream_id)
+        if state:
+            state["waiters"] = int(state.get("waiters") or 0) + 1
+        try:
+            while True:
+                state = _STREAM_RUN_LIFECYCLE.get(stream_id)
+                phase = str((state or {}).get("phase") or "").strip().lower()
+                if phase == "fallback":
+                    return False, None
+                if phase == "failed":
+                    return True, None
+                run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
+                if phase == "ready":
+                    stored_run_id = str((state or {}).get("run_id") or "").strip()
+                    return True, run_id or stored_run_id or None
+                if run_id:
+                    return True, run_id
+                if not state:
+                    return False, None
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    return True, None
+                _STREAM_RUN_STARTING_CONDITION.wait(timeout=remaining)
+        finally:
+            state = _STREAM_RUN_LIFECYCLE.get(stream_id)
+            if state:
+                waiters = max(0, int(state.get("waiters") or 0) - 1)
+                state["waiters"] = waiters
+                if _retire_gateway_run_starting_if_done(stream_id):
+                    _STREAM_RUN_STARTING_CONDITION.notify_all()
 
 _WEBUI_CHAT_BACKEND_ENV = "HERMES_WEBUI_CHAT_BACKEND"
 _WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL"
@@ -337,7 +442,8 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None:
     pattern_key = str(payload.get("pattern_key") or "").strip()
     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()
+    raw_approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip()
+    approval_id = raw_approval_id
     if not approval_id:
         approval_id = uuid.uuid4().hex
     risk = str(payload.get("risk_level") or "high").strip()
@@ -357,6 +463,7 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None:
         "risk_level": risk,
         "run_id": run_id,
         "approval_id": approval_id,
+        "_gateway_raw_approval_id_present": bool(raw_approval_id),
         "choices": choices,
         "allow_permanent": bool(allow_permanent),
     }
@@ -369,88 +476,92 @@ def _run_gateway_runs_api_streaming(
     attachments=None, cfg=None, session=None,
 ):
     """Submit via POST /v1/runs and relay SSE events including approval."""
-    url_runs = f"{base_url.rstrip('/')}/v1/runs"
-    headers = {
-        "Content-Type": "application/json",
-        "Accept": "application/json",
-        "X-Hermes-Session-Id": session_id,
-    }
-    if api_key:
-        headers["Authorization"] = f"Bearer {api_key}"
-        headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
-    message_content: Any = str(msg_text or "")
-    if attachments:
-        try:
-            from api.streaming import _build_native_multimodal_message
+    try:
+        url_runs = f"{base_url.rstrip('/')}/v1/runs"
+        headers = {
+            "Content-Type": "application/json",
+            "Accept": "application/json",
+            "X-Hermes-Session-Id": session_id,
+        }
+        if api_key:
+            headers["Authorization"] = f"Bearer {api_key}"
+            headers["X-Hermes-Session-Key"] = f"webui:{session_id}"
+        message_content: Any = str(msg_text or "")
+        if attachments:
+            try:
+                from api.streaming import _build_native_multimodal_message
 
-            message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg)
-        except Exception:
-            logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True)
-            message_content = str(msg_text or "")
-    from api.streaming import _strip_oob_blocks
+                message_content = _build_native_multimodal_message("", str(msg_text or ""), attachments, str(workspace), cfg=cfg)
+            except Exception:
+                logger.debug("Failed to build runs-API multimodal attachment payload", exc_info=True)
+                message_content = str(msg_text or "")
+        from api.streaming import _strip_oob_blocks
 
-    instructions_parts = []
-    conversation_history = []
-    for entry in getattr(session, "context_messages", None) or []:
-        if not isinstance(entry, dict):
-            continue
-        role = str(entry.get("role") or "").strip().lower()
-        if role not in {"user", "assistant"}:
-            continue
-        content = entry.get("content")
-        if content is not None:
-            content = _strip_oob_blocks(content)
+        instructions_parts = []
+        conversation_history = []
+        for entry in getattr(session, "context_messages", None) or []:
+            if not isinstance(entry, dict):
+                continue
+            role = str(entry.get("role") or "").strip().lower()
+            if role not in {"user", "assistant"}:
+                continue
+            content = entry.get("content")
+            if content is not None:
+                content = _strip_oob_blocks(content)
+                conversation_history.append({"role": role, "content": content})
+        for entry in prefill_messages or []:
+            if not isinstance(entry, dict):
+                continue
+            role = str(entry.get("role") or "").strip().lower()
+            content = entry.get("content")
+            if role == "system":
+                if isinstance(content, str) and content.strip():
+                    instructions_parts.append(content)
+                elif content is not None:
+                    instructions_parts.append(str(content))
+                continue
+            if role not in {"user", "assistant"}:
+                continue
+            if content is not None:
+                content = _strip_oob_blocks(content)
             conversation_history.append({"role": role, "content": content})
-    for entry in prefill_messages or []:
-        if not isinstance(entry, dict):
-            continue
-        role = str(entry.get("role") or "").strip().lower()
-        content = entry.get("content")
-        if role == "system":
-            if isinstance(content, str) and content.strip():
-                instructions_parts.append(content)
-            elif content is not None:
-                instructions_parts.append(str(content))
-            continue
-        if role not in {"user", "assistant"}:
-            continue
-        if content is not None:
-            content = _strip_oob_blocks(content)
-        conversation_history.append({"role": role, "content": content})
-    run_input = message_content
-    if isinstance(run_input, list):
-        run_input = [{"role": "user", "content": run_input}]
-    run_body = {
-        "model": model or "default",
-        "input": run_input,
-        **body_extras,
-        "session_id": session_id,
-    }
-    if instructions_parts:
-        run_body["instructions"] = "\n\n".join(part for part in instructions_parts if part)
-    if conversation_history:
-        run_body["conversation_history"] = conversation_history
-    req = urllib.request.Request(
-        url_runs,
-        data=json.dumps(run_body).encode("utf-8"),
-        headers=headers,
-        method="POST",
-    )
-    update_active_run(stream_id, phase="gateway-request")
-    with urllib.request.urlopen(req, timeout=30) as resp:
-        run_data = json.loads(resp.read(65536))
-    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}")
+        run_input = message_content
+        if isinstance(run_input, list):
+            run_input = [{"role": "user", "content": run_input}]
+        run_body = {
+            "model": model or "default",
+            "input": run_input,
+            **body_extras,
+            "session_id": session_id,
+        }
+        if instructions_parts:
+            run_body["instructions"] = "\n\n".join(part for part in instructions_parts if part)
+        if conversation_history:
+            run_body["conversation_history"] = conversation_history
+        req = urllib.request.Request(
+            url_runs,
+            data=json.dumps(run_body).encode("utf-8"),
+            headers=headers,
+            method="POST",
+        )
+        update_active_run(stream_id, phase="gateway-request")
+        with urllib.request.urlopen(req, timeout=30) as resp:
+            run_data = json.loads(resp.read(65536))
+        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:
+        _finish_gateway_run_starting(stream_id)
+        raise
 
-    _STREAM_RUN_IDS[stream_id] = run_id
+    usage: dict = {}
+    _publish_gateway_run_id(stream_id, run_id)
 
     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):
@@ -478,7 +589,11 @@ def _run_gateway_runs_api_streaming(
                 approval_data = _gateway_runs_approval_event(payload)
                 if approval_data:
                     approval_data["run_id"] = run_id
-                    put_gateway_event("approval", approval_data)
+                    from api.config import gateway_supports_approval_identity_v1
+                    approval_data["_gateway_agent_identity_v1"] = bool(approval_data.get("_gateway_raw_approval_id_present")) and gateway_supports_approval_identity_v1(base_url, api_key)
+                    from api.route_approvals import submit_gateway_pending_mirror
+                    head, total = submit_gateway_pending_mirror(session_id, approval_data)
+                    put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
                 sse_event = "message"
                 continue
             if payload_event in {"tool.started", "tool.completed", "reasoning.available"}:
@@ -522,6 +637,8 @@ 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)
                 if payload.get("error"):
                     raise RuntimeError(str(payload["error"]))
                 output = str(payload.get("output") or "")
@@ -533,8 +650,12 @@ 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)
                 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)
                 put_gateway_event("cancel", {"message": "Cancelled by gateway"})
                 return None, usage
             reasoning_delta = _gateway_sse_reasoning_delta(payload)
@@ -552,6 +673,40 @@ def _run_gateway_runs_api_streaming(
     return final_text, usage
 
 
+def stop_gateway_run(run_id: str) -> bool:
+    """Request gateway interruption and report whether it was acknowledged."""
+    run_id = str(run_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,
@@ -631,14 +786,10 @@ def _clear_gateway_pending_state(session: Any, stream_id: str) -> None:
 def _cleanup_gateway_pending_mirror(session_id: str) -> None:
     try:
         from api.route_approvals import (
-            _approval_sse_notify_locked,
-            _lock as _approval_lock,
-            reconcile_gateway_pending_mirror_locked,
+            retire_gateway_pending_mirror,
         )
 
-        with _approval_lock:
-            head, total, _ = reconcile_gateway_pending_mirror_locked(session_id)
-            _approval_sse_notify_locked(session_id, head, total)
+        retire_gateway_pending_mirror(session_id)
     except Exception:
         logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True)
 
@@ -664,6 +815,8 @@ def _run_gateway_chat_streaming(
     """
     q = STREAMS.get(stream_id)
     if q is None:
+        _finish_gateway_run_starting(stream_id, result="fallback")
+        _clear_gateway_run_starting(stream_id)
         # Cancelled before the worker started; release the owner entry the route
         # layer registered so STREAM_SESSION_OWNERS does not leak (no teardown finally runs).
         unregister_stream_owner(stream_id)
@@ -691,6 +844,7 @@ def _run_gateway_chat_streaming(
         STREAM_LIVE_TOOL_CALLS[stream_id] = []
 
     success_writeback_committed = False
+    runs_api_pending_marked = True
 
     def put_gateway_event(event, data):
         if cancel_event.is_set() and not success_writeback_committed and event not in ("cancel", "error", "apperror"):
@@ -732,6 +886,22 @@ 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)
+        _use_runs_api = _runs_api_enabled and gateway_supports_approval(base_url, api_key)
+        if not _use_runs_api and runs_api_pending_marked:
+            _finish_gateway_run_starting(stream_id, result="fallback")
+            runs_api_pending_marked = False
         try:
             from api.streaming import (
                 _load_webui_prefill_context,
@@ -770,19 +940,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:
@@ -908,12 +1065,13 @@ def _run_gateway_chat_streaming(
                             _approval_run_id = str(approval_data.get("run_id") or "").strip()
                             if _approval_run_id:
                                 _STREAM_RUN_IDS[stream_id] = _approval_run_id
-                            put_gateway_event("approval", approval_data)
                             try:
                                 from api.route_approvals import submit_gateway_pending_mirror
-                                submit_gateway_pending_mirror(session_id, approval_data)
+                                head, total = submit_gateway_pending_mirror(session_id, approval_data)
+                                approval_data = {**(head or approval_data), "pending_count": total}
                             except Exception:
                                 logger.debug("submit_gateway_pending_mirror failed", exc_info=True)
+                            put_gateway_event("approval", approval_data)
                         else:
                             logger.debug("Ignoring malformed gateway approval payload")
                         sse_event = "message"
@@ -1180,6 +1338,13 @@ def _run_gateway_chat_streaming(
             "hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.",
         })
     finally:
+        mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
+        if mapped_run_id:
+            try:
+                from api.route_approvals import retire_gateway_pending_mirror
+                retire_gateway_pending_mirror(session_id, run_id=mapped_run_id)
+            except Exception:
+                logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True)
         if s is not None:
             try:
                 with _get_session_agent_lock(session_id):
@@ -1196,6 +1361,8 @@ def _run_gateway_chat_streaming(
             STREAM_LIVE_TOOL_CALLS.pop(stream_id, None)
             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

@@ -46,7 +46,10 @@ except ImportError:
 _approval_sse_subscribers: dict[str, list[queue.Queue]] = {}
 _GATEWAY_MIRROR_FLAG = "_gateway_mirror"
 _GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token"
+_GATEWAY_MIRROR_RETAINED = "_gateway_mirror_retained"
 _GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token"
+_GATEWAY_AGENT_IDENTITY_V1 = "_gateway_agent_identity_v1"
+_gateway_relay_owners: dict[tuple[str, str], str] = {}
 
 
 def _approval_sse_subscribe(session_id: str) -> queue.Queue:
@@ -106,7 +109,7 @@ def _approval_sse_notify(session_id: str, head: dict | None, total: int) -> None
         _approval_sse_notify_locked(session_id, head, total)
 
 
-def _gateway_mirror_entry_token(entry) -> str:
+def _gateway_mirror_entry_token(entry) -> str | None:
     """Return a stable token for the current process lifetime of a gateway head.
 
     Stamps a token key into the entry's `.data` dict so
@@ -120,7 +123,7 @@ def _gateway_mirror_entry_token(entry) -> str:
             token = uuid.uuid4().hex
             data[_GATEWAY_ENTRY_DATA_TOKEN_KEY] = token
         return token
-    return uuid.uuid4().hex
+    return None
 
 
 def _is_gateway_mirror_entry(entry: dict | None) -> bool:
@@ -147,29 +150,118 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
 
     live_head_entry = live_gateway_queue[0] if live_gateway_queue else None
     live_head_data = getattr(live_head_entry, "data", None) or {}
-    live_token = _gateway_mirror_entry_token(live_head_entry) if live_head_entry and live_head_data else None
+    has_no_run_mirror = any(
+        _is_gateway_mirror_entry(entry)
+        and not str(entry.get("run_id") or "").strip()
+        for entry in queue_list
+    )
+    live_run_id = str(live_head_data.get("run_id") or "").strip()
+    live_has_token = bool(live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY))
+    live_local_tokens: set[str] = set()
+    for live_entry in live_gateway_queue:
+        live_data = getattr(live_entry, "data", None) or {}
+        if str(live_data.get("run_id") or "").strip():
+            continue
+        live_entry_token = str(live_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) or "").strip()
+        if live_entry_token or not has_no_run_mirror:
+            live_entry_token = _gateway_mirror_entry_token(live_entry) or ""
+            if live_entry_token:
+                live_local_tokens.add(live_entry_token)
+                if not str(live_data.get("approval_id") or "").strip():
+                    live_data["approval_id"] = f"gwlocal:{live_entry_token}"
+    live_token = (
+        _gateway_mirror_entry_token(live_head_entry)
+        if live_head_entry and live_head_data
+        and (live_run_id or live_has_token or not has_no_run_mirror)
+        else None
+    )
+    if live_token and live_run_id and not str(live_head_data.get("approval_id") or "").strip():
+        live_head_data["approval_id"] = f"gwrun:{live_run_id}:{live_token}"
+    live_approval_id = str(live_head_data.get("approval_id") or "").strip()
 
     rebuilt: list[dict] = []
+    deferred_run_entries: list[dict] = []
     live_mirror_present = False
     for entry in queue_list:
         if not _is_gateway_mirror_entry(entry):
             rebuilt.append(entry)
             continue
-        if live_token and entry.get(_GATEWAY_MIRROR_TOKEN) == live_token and not live_mirror_present:
+        entry_run_id = str(entry.get("run_id") or "").strip()
+        entry_approval_id = str(entry.get("approval_id") or "").strip()
+        entry_token = str(entry.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
+        matches_live_head = False
+        if live_token:
+            if entry_token and entry_token == live_token:
+                matches_live_head = True
+            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:
+            if matches_live_head and not live_mirror_present:
+                if entry_token != live_token:
+                    entry[_GATEWAY_MIRROR_TOKEN] = live_token
+                    changed = True
+                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
+            changed = True
+            continue
+
+        if entry.get(_GATEWAY_MIRROR_RETAINED):
+            rebuilt.append(entry)
+            continue
+
+        if matches_live_head and not live_mirror_present:
+            if entry_token != live_token:
+                entry[_GATEWAY_MIRROR_TOKEN] = live_token
+                changed = True
             rebuilt.append(entry)
             live_mirror_present = True
             continue
+
+        if entry_token and entry_token in live_local_tokens:
+            rebuilt.append(entry)
+            continue
+
+        if not live_token:
+            if entry_token:
+                changed = True
+                continue
+            rebuilt.append(entry)
+            continue
+
         changed = True
 
     if live_token and not live_mirror_present:
         mirror_entry = dict(live_head_data)
-        mirror_entry.setdefault("approval_id", uuid.uuid4().hex)
+        mirror_run_id = str(mirror_entry.get("run_id") or "").strip()
+        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
@@ -184,38 +276,337 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
     return head, total, changed
 
 
-def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str | None:
-    """Return the mirrored gateway approval run_id for a matching pending card.
+def _gateway_pending_mirror_locked(session_key: str, approval_id: str = "", run_id: str = "") -> dict | None:
+    """Return the exact live run-backed mirror under `_lock`."""
+    approval_id = str(approval_id or "").strip()
+    run_id = str(run_id or "").strip()
+    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):
+                continue
+            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
+        if run_id and entry.get("run_id") == run_id:
+            return entry
+    return None
 
-    Reconciles the mirror first so a live gateway head still survives a lost
-    `active_stream_id` pointer.
-    """
+
+def gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str = "") -> dict | None:
+    """Return an exact live run-backed mirror for this session."""
+    with _lock:
+        reconcile_gateway_pending_mirror_locked(session_key)
+        entry = _gateway_pending_mirror_locked(session_key, approval_id, run_id)
+        return dict(entry) if entry else None
+
+
+def claim_gateway_approval_relay_owner(session_key: str, run_id: str, approval_id: str) -> bool:
+    """Claim the single-flight relay owner for one `(session, run)` pair."""
+    session_key = str(session_key or "").strip()
+    run_id = str(run_id or "").strip()
+    approval_id = str(approval_id or "").strip()
+    if not session_key or not run_id:
+        return False
+    with _lock:
+        key = (session_key, run_id)
+        if key in _gateway_relay_owners:
+            return False
+        _gateway_relay_owners[key] = approval_id
+        return True
+
+
+def release_gateway_approval_relay_owner(session_key: str, run_id: str, approval_id: str = "") -> None:
+    """Release the single-flight relay owner for one `(session, run)` pair."""
+    session_key = str(session_key or "").strip()
+    run_id = str(run_id or "").strip()
+    approval_id = str(approval_id or "").strip()
+    if not session_key or not run_id:
+        return
+    with _lock:
+        key = (session_key, run_id)
+        current = str(_gateway_relay_owners.get(key) or "").strip()
+        if approval_id and current and current != approval_id:
+            return
+        _gateway_relay_owners.pop(key, None)
+
+
+def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str = "") -> bool:
+    """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 match is None and not normalized_run_id:
+                match = next((entry for entry in entries if _is_gateway_mirror_entry(entry)
+                              and not str(entry.get("run_id") or "").strip()
+                              and str(entry.get("approval_id") or "").strip() == approval_id), None)
+            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 [
+                entry for entry in entries
+                if _is_gateway_mirror_entry(entry)
+                and not str(entry.get("run_id") or "").strip()
+            ]
+            if normalized_run_id:
+                retained_gateway_queue = []
+                for entry in gateway_queue:
+                    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:
+            head, total, changed = reconcile_gateway_pending_mirror_locked(session_key)
+            _approval_sse_notify_locked(session_key, head, total)
+            if changed:
+                publish_session_list_changed("attention_resolved")
+            return changed
+        for match in retired:
+            entries.remove(match)
+        if normalized_run_id and not approval_id:
+            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, 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
+
+
+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()
     if not approval_id:
         return None
     with _lock:
-        reconcile_gateway_pending_mirror_locked(session_key)
-        queue = _pending.get(session_key)
-        if isinstance(queue, list):
-            entries = queue
-        elif queue:
-            entries = [queue]
-        else:
-            return None
-        for entry in entries:
-            if isinstance(entry, dict) and entry.get("approval_id") == approval_id and entry.get(_GATEWAY_MIRROR_FLAG):
-                run_id = str(entry.get("run_id") or "").strip()
-                return run_id or None
+        entry = _gateway_pending_mirror_locked(session_key, approval_id=approval_id)
+        if entry:
+            return str(entry.get("run_id") or "").strip() or None
     return None
 
 
-def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None:
+def submit_gateway_pending_mirror(session_key: str, approval: dict) -> tuple[dict | None, int]:
     """Mirror the live gateway head into WebUI polling state under a typed tag."""
-    del approval  # mirror from the live gateway head under `_lock`, not from callback input
     with _lock:
+        run_id = str(approval.get("run_id") or "").strip()
+        approval_id = str(approval.get("approval_id") or "").strip()
+        live_gateway_queue = _gateway_queues.get(session_key) or []
+        exact_local_entry = next(
+            (
+                entry for entry in live_gateway_queue
+                if getattr(entry, "data", None) is approval
+            ),
+            None,
+        ) if not run_id else None
+        if exact_local_entry is None and not run_id and approval_id:
+            exact_local_entry = next(
+                (
+                    entry for entry in live_gateway_queue
+                    if str(((getattr(entry, "data", None) or {}).get("approval_id") or "")).strip() == approval_id
+                ),
+                None,
+            )
+        if exact_local_entry is not None:
+            mirror_entries = _normalize_pending_queue_locked(session_key)
+            entries_to_mirror = [live_gateway_queue[0]] if live_gateway_queue else []
+            if exact_local_entry not in entries_to_mirror:
+                entries_to_mirror.append(exact_local_entry)
+            for entry in entries_to_mirror:
+                local_data = entry.data
+                token = _gateway_mirror_entry_token(entry)
+                entry_approval_id = str(local_data.get("approval_id") or "").strip()
+                if entry is exact_local_entry:
+                    entry_approval_id = approval_id or entry_approval_id or f"gwlocal:{token}"
+                    approval_id = entry_approval_id
+                    approval["approval_id"] = entry_approval_id
+                elif not entry_approval_id:
+                    entry_approval_id = f"gwlocal:{token}"
+                local_data["approval_id"] = entry_approval_id
+                if not any(
+                    _is_gateway_mirror_entry(mirror)
+                    and str(mirror.get(_GATEWAY_MIRROR_TOKEN) or "") == token
+                    for mirror in mirror_entries
+                ):
+                    mirror_entry = dict(local_data)
+                    mirror_entry["approval_id"] = entry_approval_id
+                    mirror_entry[_GATEWAY_MIRROR_FLAG] = True
+                    mirror_entry[_GATEWAY_MIRROR_TOKEN] = token
+                    mirror_entries.append(mirror_entry)
+        if run_id:
+            live_head_entry = live_gateway_queue[0] if live_gateway_queue else None
+            live_head_data = getattr(live_head_entry, "data", None) or {}
+            live_head_run_id = str(live_head_data.get("run_id") or "").strip()
+            live_head_approval_id = str(live_head_data.get("approval_id") or "").strip()
+            live_token = (
+                _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"] = approval_id
+                mirror_entry[_GATEWAY_MIRROR_FLAG] = True
+                if not _gateway_pending_mirror_locked(session_key, approval_id=approval_id, run_id=run_id):
+                    _normalize_pending_queue_locked(session_key).append(mirror_entry)
+        elif not exact_local_entry:
+            if not approval_id:
+                approval_id = uuid.uuid4().hex
+                approval["approval_id"] = approval_id
+            queue = _pending.get(session_key)
+            entries = queue if isinstance(queue, list) else [queue] if queue else []
+            no_run_mirror = next(
+                (
+                    entry for entry in reversed(entries)
+                    if _is_gateway_mirror_entry(entry)
+                    and not str(entry.get("run_id") or "").strip()
+                    and str(entry.get("approval_id") or "").strip() == approval_id
+                ),
+                None,
+            )
+            if no_run_mirror:
+                approval["approval_id"] = str(no_run_mirror.get("approval_id") or approval_id).strip()
+            elif not _gateway_pending_mirror_locked(session_key, approval_id=approval_id):
+                mirror_entry = dict(approval)
+                mirror_entry["approval_id"] = approval_id
+                mirror_entry[_GATEWAY_MIRROR_FLAG] = True
+                _normalize_pending_queue_locked(session_key).append(mirror_entry)
         head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
         _approval_sse_notify_locked(session_key, head, total)
     publish_session_list_changed("attention_pending")
+    return (dict(head) if head else None), total
+
+
+def resolve_gateway_pending_local(
+    session_key: str, approval_id: str, choice: str, reason: str | None = None
+) -> tuple[int, dict | None, int]:
+    """Resolve the exact parked local entry bound to an approval mirror."""
+    target = None
+    with _lock:
+        approval_id = str(approval_id or "").strip()
+        gateway_queue = _gateway_queues.get(session_key) or []
+        for index, entry in enumerate(gateway_queue):
+            data = getattr(entry, "data", None) or {}
+            if str(data.get("approval_id") or "").strip() == approval_id:
+                target = gateway_queue.pop(index)
+                break
+        if gateway_queue:
+            _gateway_queues[session_key] = gateway_queue
+        else:
+            _gateway_queues.pop(session_key, None)
+        head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
+        _approval_sse_notify_locked(session_key, head, total)
+    if target is None:
+        return 0, head, total
+    target.result = choice
+    if reason:
+        target.reason = reason
+    target.event.set()
+    publish_session_list_changed("attention_resolved")
+    return 1, head, total
+
+
+def resolve_gateway_pending_local_no_run_mirror(
+    session_key: str, approval_id: str, choice: str, reason: str | None = None
+) -> tuple[bool, int, dict | None, int]:
+    """Resolve an exact no-run mirror only while its parked producer still exists."""
+    target = None
+    with _lock:
+        approval_id = str(approval_id or "").strip()
+        queue = _pending.get(session_key)
+        entries = queue if isinstance(queue, list) else [queue] if queue else []
+        matched_mirror = next(
+            (
+                entry for entry in entries
+                if _is_gateway_mirror_entry(entry)
+                and not str(entry.get("run_id") or "").strip()
+                and str(entry.get("approval_id") or "").strip() == approval_id
+            ),
+            None,
+        )
+        if matched_mirror is None:
+            return False, 0, entries[0] if entries else None, len(entries)
+
+        gateway_queue = _gateway_queues.get(session_key) or []
+        for index, entry in enumerate(gateway_queue):
+            data = getattr(entry, "data", None) or {}
+            if str(data.get("approval_id") or "").strip() == approval_id:
+                target = gateway_queue.pop(index)
+                break
+        if target is None:
+            matched_mirror[_GATEWAY_MIRROR_RETAINED] = True
+            return True, 0, entries[0] if entries else None, len(entries)
+
+        if gateway_queue:
+            _gateway_queues[session_key] = gateway_queue
+        else:
+            _gateway_queues.pop(session_key, None)
+        entries.remove(matched_mirror)
+        if entries:
+            _pending[session_key] = entries
+        else:
+            _pending.pop(session_key, None)
+        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
 
 
 def submit_pending(session_key: str, approval: dict) -> None:

api/routes.py

@@ -9715,8 +9715,15 @@ from api.route_approvals import (  # noqa: F401 — re-exports for backward comp
     _approval_sse_notify_locked,
     _approval_sse_notify,
     _GATEWAY_MIRROR_FLAG,
-    _gateway_mirrored_pending_run_id,
+    _GATEWAY_MIRROR_TOKEN,
+    _gateway_mirror_entry_token,
+    claim_gateway_approval_relay_owner,
+    gateway_pending_mirror,
+    release_gateway_approval_relay_owner,
+    retire_gateway_pending_mirror,
     reconcile_gateway_pending_mirror_locked,
+    resolve_gateway_pending_local,
+    resolve_gateway_pending_local_no_run_mirror,
     submit_gateway_pending_mirror,
     submit_pending,
 )
@@ -13367,6 +13374,39 @@ 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 (
+                GATEWAY_RUN_ID_WAIT_TIMEOUT,
+                stop_gateway_run,
+                wait_for_gateway_run_id,
+            )
+
+            structured_gateway, run_id = wait_for_gateway_run_id(stream_id, GATEWAY_RUN_ID_WAIT_TIMEOUT)
+            if not run_id and structured_gateway:
+                gateway_stop_blocked = True
+            if run_id:
+                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)
+                else:
+                    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():
@@ -21206,13 +21246,27 @@ def _start_chat_stream_for_session(
     worker_kwargs = {"model_provider": model_provider, "goal_related": goal_related}
     if moa_config and not backend_is_gateway:
         worker_kwargs["moa_config"] = moa_config
+    if backend_is_gateway:
+        from api.gateway_chat import _mark_gateway_run_starting
+        _mark_gateway_run_starting(stream_id)
     thr = threading.Thread(
         target=worker_target,
         args=(s.session_id, msg, model, workspace, stream_id, attachments),
         kwargs=worker_kwargs,
         daemon=True,
     )
-    thr.start()
+    try:
+        thr.start()
+    except Exception:
+        if backend_is_gateway:
+            try:
+                from api.gateway_chat import _finish_gateway_run_starting
+                _finish_gateway_run_starting(stream_id)
+                from api.gateway_chat import _clear_gateway_run_starting
+                _clear_gateway_run_starting(stream_id)
+            except Exception:
+                logger.debug("Failed to record gateway run-start failure for stream %s", stream_id, exc_info=True)
+        raise
     response = {
         "stream_id": stream_id,
         "session_id": s.session_id,
@@ -23748,7 +23802,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
@@ -23759,17 +23813,31 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool:
     pending = None
     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)
         if isinstance(queue, list):
             if approval_id:
-                # Find and remove the specific entry by approval_id.
+                # Prefer a local exact-id match over a mirrored one, so a
+                # preceding remote mirror cannot consume the user's local choice.
+                preferred_index = None
+                fallback_index = None
                 for i, entry in enumerate(queue):
-                    if entry.get("approval_id") == approval_id:
-                        pending = queue.pop(i)
-                        found_target = True
+                    if entry.get("approval_id") != approval_id:
+                        continue
+                    if run_id and str(entry.get("run_id") or "").strip() != run_id:
+                        continue
+                    if not entry.get(_GATEWAY_MIRROR_FLAG) or not str(entry.get("run_id") or "").strip():
+                        preferred_index = i
                         break
+                    if fallback_index is None:
+                        fallback_index = i
+                match_index = preferred_index if preferred_index is not None else fallback_index
+                if match_index is not None:
+                    pending = queue.pop(match_index)
+                    found_target = True
                 else:
                     # A stale explicit id must not accidentally approve the
                     # oldest queued command; duplicate/stale responses are
@@ -23782,7 +23850,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
@@ -23805,16 +23879,35 @@ 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()
+                gateway_head_matches_target = bool(
+                    run_id and gw_approval_id == approval_id and gw_run_id == run_id
+                )
+                if gw_approval_id == approval_id and (not run_id or gw_run_id == run_id):
+                    local_gateway_approval_id = approval_id
+                elif not run_id and found_target and pending:
+                    pending_token = str(pending.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
+                    gateway_token = str(gw_data.get("_webui_mirror_token") or "").strip()
+                    if pending_token and pending_token == gateway_token:
+                        gw_data["approval_id"] = approval_id
+                        local_gateway_approval_id = approval_id
         # Notify SSE subscribers of the new head (or empty state) so the UI
         # surfaces any trailing approvals that were queued behind this one
         # without waiting for the next submit_pending. Without this, a parallel
         # tool-call scenario (#527) would leave the second approval invisible
         # in the SSE path until the next event ever fired (the agent thread
         # would be parked indefinitely from the user's perspective).
-        if isinstance(_pending.get(sid), list) and _pending[sid]:
-            _approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid]))
-        else:
-            _approval_sse_notify_locked(sid, None, 0)
+        if not local_gateway_approval_id:
+            if isinstance(_pending.get(sid), list) and _pending[sid]:
+                _approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid]))
+            else:
+                _approval_sse_notify_locked(sid, None, 0)
 
     # Collect keys from both _pending and _gateway_queues
     keys_from_pending = pending.get("pattern_keys") or [pending.get("pattern_key", "")] if pending else []
@@ -23836,11 +23929,18 @@ 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:
+    local_gateway_resolved = 0
+    if approval_id and found_target and not run_id and local_gateway_approval_id:
+        local_gateway_resolved, _head, _total = resolve_gateway_pending_local(
+            sid, local_gateway_approval_id, choice
+        )
+    elif approval_id and found_target and run_id and gateway_head_matches_target:
+        gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0
+    elif not approval_id:
         gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0
     # Keep the historical no-id response path truthy for old clients/tests while
     # making stale explicit ids bounded as not-active for Slice 3b.
-    resolved = bool(pending) or bool(gateway_resolved) or not bool(approval_id)
+    resolved = bool(pending) or bool(gateway_resolved) or bool(local_gateway_resolved) or not bool(approval_id)
     if resolved:
         publish_session_list_changed("attention_resolved")
     return resolved
@@ -23850,6 +23950,10 @@ _GATEWAY_APPROVAL_RELAY_UNAVAILABLE = (
     "Gateway approval could not be relayed because the active run is unavailable. "
     "Reopen the session or retry after it reconnects."
 )
+_GATEWAY_APPROVAL_RELAY_IN_PROGRESS = (
+    "Another approval response for this Gateway run is already in progress. "
+    "Wait for it to finish, then retry if the card is still visible."
+)
 
 
 def _gateway_pending_approval_without_run_id(sid: str, approval_id: str) -> bool:
@@ -23865,11 +23969,11 @@ def _gateway_pending_approval_without_run_id(sid: str, approval_id: str) -> bool
         if approval_id:
             for entry in entries:
                 if isinstance(entry, dict) and entry.get("approval_id") == approval_id:
-                    return bool(entry.get(_GATEWAY_MIRROR_FLAG))
+                    return bool(entry.get(_GATEWAY_MIRROR_FLAG)) and not str(entry.get("run_id") or "").strip()
             return False
         if not entries or not isinstance(entries[0], dict):
             return False
-        return bool(entries[0].get(_GATEWAY_MIRROR_FLAG))
+        return bool(entries[0].get(_GATEWAY_MIRROR_FLAG)) and not str(entries[0].get("run_id") or "").strip()
 
 
 def _session_has_pending_approval(sid: str) -> bool:
@@ -23912,36 +24016,76 @@ def _handle_approval_respond(handler, body):
             _gateway_api_key,
             webui_gateway_chat_enabled,
         )
-        from api.config import get_config as _get_config
+        from api.config import get_config as _get_config, gateway_supports_approval_identity_v1
+        from api.route_approvals import _GATEWAY_AGENT_IDENTITY_V1
         s = get_session(sid)
-        _run_id = None
+        _candidate_run_id = None
         if s is not None:
             active_sid = getattr(s, "active_stream_id", None)
             if active_sid:
-                _run_id = _STREAM_RUN_IDS.get(active_sid)
-            if not _run_id and approval_id:
-                _run_id = _gateway_mirrored_pending_run_id(sid, approval_id)
-        if _run_id:
-            if not approval_id:
-                return bad(handler, "approval_id is required for gateway approvals")
-            from api.runner_client import HttpRunnerClient, RunnerClientError
-            _cfg = _get_config()
-            _base = _gateway_base_url(_cfg)
-            _key = _gateway_api_key()
-            try:
-                HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(_run_id, approval_id, choice)
-            except (RunnerClientError, ValueError) as exc:
-                return j(handler, {"ok": False, "choice": choice, "relayed": True, "error": str(exc)}, status=502)
-            # The outbound relay only resumes the remote run; the local mirror
-            # still needs the same cleanup path so the parked entry, mirrored
-            # card, and agent signal all settle here too.
-            _resolve_approval_legacy(sid, approval_id, choice)
-            return j(handler, {"ok": True, "choice": choice, "relayed": True})
-        # Only a still-mirrored gateway approval with a missing run should 409;
-        # stale or empty gateway clicks fall through to local resolution.
-        if webui_gateway_chat_enabled(_get_config()) and _gateway_pending_approval_without_run_id(
-            sid, approval_id
-        ):
+                _candidate_run_id = _STREAM_RUN_IDS.get(active_sid)
+        local_match = False
+        run_backed_gateway_matches = 0
+        same_run_stale_without_token = False
+        with _lock:
+            queue = _pending.get(sid)
+            entries = queue if isinstance(queue, list) else [queue] if queue else []
+            if approval_id:
+                local_match = any(
+                    isinstance(entry, dict)
+                    and entry.get("approval_id") == approval_id
+                    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()
+                )
+                gateway_queue = _gateway_queues.get(sid) or []
+                live_head_data = getattr(gateway_queue[0], "data", None) or {} if gateway_queue else {}
+                live_head_run_id = str(live_head_data.get("run_id") or "").strip()
+                live_head_token = (
+                    _gateway_mirror_entry_token(gateway_queue[0])
+                    if gateway_queue and live_head_data
+                    else None
+                )
+                live_head_approval_id = str(live_head_data.get("approval_id") or "").strip()
+                if not live_head_approval_id and live_head_token and live_head_run_id:
+                    live_head_approval_id = f"gwrun:{live_head_run_id}:{live_head_token}"
+                stale_same_run_id = _candidate_run_id or live_head_run_id
+                if (
+                    stale_same_run_id
+                    and live_head_run_id == stale_same_run_id
+                    and live_head_approval_id
+                    and live_head_approval_id != approval_id
+                ):
+                    same_run_stale_without_token = any(
+                        isinstance(entry, dict)
+                        and entry.get("approval_id") == approval_id
+                        and entry.get(_GATEWAY_MIRROR_FLAG)
+                        and str(entry.get("run_id") or "").strip() == stale_same_run_id
+                        and not str(entry.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
+                        for entry in entries
+                    )
+            else:
+                local_match = any(
+                    isinstance(entry, dict)
+                    and (
+                        not entry.get(_GATEWAY_MIRROR_FLAG)
+                        or not str(entry.get("run_id") or "").strip()
+                    )
+                    for entry in entries
+                )
+        if local_match:
+            _candidate_run_id = None
+        if approval_id and not local_match and same_run_stale_without_token:
             return j(
                 handler,
                 {
@@ -23953,6 +24097,111 @@ def _handle_approval_respond(handler, body):
                 },
                 status=409,
             )
+        matched_mirror = (
+            gateway_pending_mirror(sid, approval_id=approval_id, run_id=_candidate_run_id)
+            if approval_id and not local_match
+            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()
+            _base = _gateway_base_url(_cfg)
+            _key = _gateway_api_key()
+            claimed_approval_id = str(matched_mirror.get("approval_id") or "").strip()
+            if not claim_gateway_approval_relay_owner(sid, _run_id, claimed_approval_id):
+                return j(
+                    handler,
+                    {
+                        "ok": False,
+                        "choice": choice,
+                        "relayed": False,
+                        "code": "gateway_approval_in_progress",
+                        "error": _GATEWAY_APPROVAL_RELAY_IN_PROGRESS,
+                    },
+                    status=409,
+                )
+            try:
+                current_mirror = gateway_pending_mirror(
+                    sid,
+                    approval_id=claimed_approval_id,
+                    run_id=_run_id,
+                )
+                if not current_mirror:
+                    return j(
+                        handler,
+                        {
+                            "ok": False,
+                            "choice": choice,
+                            "relayed": False,
+                            "code": "gateway_run_unavailable",
+                            "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE,
+                        },
+                        status=409,
+                    )
+                identity_v1 = bool(current_mirror.get(_GATEWAY_AGENT_IDENTITY_V1)) and gateway_supports_approval_identity_v1(_base, _key)
+                if not identity_v1:
+                    run_head = gateway_pending_mirror(sid, run_id=_run_id)
+                    if not run_head or str(run_head.get("approval_id") or "").strip() != claimed_approval_id:
+                        return j(
+                            handler,
+                            {
+                                "ok": False,
+                                "choice": choice,
+                                "relayed": False,
+                                "code": "gateway_run_unavailable",
+                                "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE,
+                            },
+                            status=409,
+                        )
+                matched_mirror = current_mirror
+                try:
+                    HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(
+                        _run_id, matched_mirror["approval_id"] if identity_v1 else "", choice
+                    )
+                except (RunnerClientError, ValueError) as exc:
+                    return j(handler, {"ok": False, "choice": choice, "relayed": True, "error": str(exc)}, status=502)
+                # The outbound relay only resumes the remote run; the local mirror
+                # 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, 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})
+            finally:
+                release_gateway_approval_relay_owner(sid, _run_id, claimed_approval_id)
+        if _candidate_run_id:
+            return j(handler, {"ok": False, "choice": choice, "relayed": False,
+                               "code": "gateway_run_unavailable",
+                               "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE}, status=409)
+        # A no-run mirror is local visibility state only. Resolve it only while
+        # the exact parked producer still exists; otherwise keep the card live
+        # and fail closed instead of claiming success.
+        if webui_gateway_chat_enabled(_get_config()):
+            handled_no_run_mirror, resolved_count, _, _ = resolve_gateway_pending_local_no_run_mirror(
+                sid, approval_id, choice
+            )
+            if handled_no_run_mirror and resolved_count == 1:
+                return j(handler, {"ok": True, "choice": choice, "local_retired": True})
+            if handled_no_run_mirror:
+                return j(handler, {"ok": False, "choice": choice, "relayed": False,
+                                   "code": "gateway_run_unavailable",
+                                   "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE}, status=409)
     except Exception:
         pass  # fall through to local approval path
 

api/streaming.py

@@ -7808,14 +7808,10 @@ def _run_agent_streaming(
             try:
                 from api.route_approvals import (
                     submit_gateway_pending_mirror as _submit_pending_for_polling,
-                    reconcile_gateway_pending_mirror_locked as _reconcile_gateway_pending_mirror_locked,
-                    _approval_sse_notify_locked as _approval_sse_notify_locked,
-                    _lock as _approval_lock,
+                    retire_gateway_pending_mirror as _retire_gateway_pending_mirror,
                 )
                 def _cleanup_gateway_pending_mirror():
-                    with _approval_lock:
-                        head, total, _changed = _reconcile_gateway_pending_mirror_locked(session_id)
-                        _approval_sse_notify_locked(session_id, head, total)
+                    _retire_gateway_pending_mirror(session_id)
             except ImportError:
                 _submit_pending_for_polling = None
                 _cleanup_gateway_pending_mirror = None
@@ -7826,7 +7822,8 @@ def _run_agent_streaming(
             def _approval_notify_cb(approval_data):
                 if _submit_pending_for_polling is not None:
                     try:
-                        _submit_pending_for_polling(session_id, approval_data)
+                        head, total = _submit_pending_for_polling(session_id, approval_data)
+                        approval_data = {**(head or approval_data), "pending_count": total}
                     except Exception:
                         logger.warning("Failed to mirror approval into WebUI polling state", exc_info=True)
                 put('approval', approval_data)
@@ -8241,20 +8238,9 @@ def _run_agent_streaming(
                         if _has_blocking_approval(session_id):
                             p = None
                             with _approval_lock:
-                                _reconcile_gateway_pending_mirror_locked(session_id)
-                                queue = _approval_pending.get(session_id)
-                                if isinstance(queue, list):
-                                    p = dict(queue[0]) if queue else None
-                                elif queue:
-                                    p = dict(queue)
-                                if p is None:
-                                    gw_queue = _approval_gateway_queues.get(session_id) or []
-                                    if gw_queue:
-                                        raw = getattr(gw_queue[0], 'data', None) or {}
-                                        if raw:
-                                            p = dict(raw)
-                                        else:
-                                            logger.warning("Gateway queue entry for %s has no .data attribute", session_id)
+                                p, pending_count, _changed = _reconcile_gateway_pending_mirror_locked(session_id)
+                                if p:
+                                    p = {**p, "pending_count": pending_count}
                             if p:
                                 put('approval', p)
                     except ImportError:

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'),null,'error');
+  }
   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'),null,'error');
+  }
 }
 
 /**

static/i18n.js

@@ -17,7 +17,7 @@ const LOCALES = {
     _speech: 'en-US',
     // boot.js
     cancelling: 'Cancelling\u2026',
-    cancel_failed: 'Cancel failed: ',
+    cancel_failed: 'Cancel failed.',
     mic_denied: 'Microphone access denied. Check browser permissions.',
     mic_insecure_origin: 'Voice input needs a secure connection. Open Hermes over HTTPS or from localhost to use the microphone.',
     mic_no_speech: 'No speech detected. Try again.',
@@ -165,6 +165,7 @@ const LOCALES = {
     approval_responding: 'Responding\u2026',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Approvals not supported',
+    approval_pending_count: (n) => `1 of ${n} pending`,
     clarify_heading: 'Clarification needed',
     clarify_hint: 'Pick a choice, or type your own answer below.',
     clarify_other: 'Other',
@@ -1777,7 +1778,7 @@ const LOCALES = {
     _speech: 'it-IT',
     // boot.js
     cancelling: 'Annullamento\u2026',
-    cancel_failed: 'Annullamento fallito: ',
+    cancel_failed: 'Annullamento fallito.',
     mic_denied: 'Accesso al microfono negato. Controlla i permessi del browser.',
     mic_insecure_origin: 'L\'input vocale richiede una connessione sicura. Apri Hermes tramite HTTPS o da localhost per usare il microfono.',
     mic_no_speech: 'Nessuna voce rilevata. Riprova.',
@@ -1925,6 +1926,7 @@ const LOCALES = {
     approval_responding: 'Rispondendo\u2026',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Approvazioni non supportate',
+    approval_pending_count: (n) => `1 di ${n} in sospeso`,
     clarify_heading: 'Chiarimento necessario',
     clarify_hint: "Scegli un'opzione o scrivi la tua risposta qui sotto.",
     clarify_other: 'Altro',
@@ -3519,7 +3521,7 @@ const LOCALES = {
     _speech: 'ja-JP',
     // boot.js
     cancelling: 'キャンセル中…',
-    cancel_failed: 'キャンセル失敗: ',
+    cancel_failed: 'キャンセルに失敗しました。',
     mic_denied: 'マイクへのアクセスが拒否されました。ブラウザの権限を確認してください。',
     mic_insecure_origin: '音声入力にはセキュアな接続が必要です。マイクを使うには HTTPS または localhost 経由で Hermes を開いてください。',
     mic_no_speech: '音声が検出されませんでした。もう一度お試しください。',
@@ -3667,6 +3669,7 @@ const LOCALES = {
     approval_responding: '応答中…',
     approval_gateway_unsupported: '承認を実行するには新しいゲートウェイが必要です。有効にするには接続されている Hermes ゲートウェイをアップグレードしてください。',
     approval_gateway_unsupported_label: '承認はサポートされていません',
+    approval_pending_count: (n) => `${n}件中1件保留中`,
     clarify_heading: '確認が必要',
     clarify_hint: '選択肢を選ぶか、下に独自の回答を入力してください。',
     clarify_other: 'その他',
@@ -5265,7 +5268,7 @@ const LOCALES = {
     _label: 'Русский',
     _speech: 'ru-RU',
     cancelling: 'Отменяю…',
-    cancel_failed: 'Не удалось отменить: ',
+    cancel_failed: 'Не удалось отменить.',
     mic_denied: 'Доступ к микрофону запрещён. Проверьте разрешения браузера.',
     mic_insecure_origin: 'Для голосового ввода нужно защищённое соединение. Откройте Hermes по HTTPS или с localhost, чтобы использовать микрофон.',
     mic_no_speech: 'Речь не распознана. Попробуйте ещё раз.',
@@ -5397,6 +5400,7 @@ const LOCALES = {
     approval_responding: 'Отвечаю…',
     approval_gateway_unsupported: 'Для подтверждений нужен более новый Gateway. Обновите подключённый Hermes Gateway, чтобы включить их.',
     approval_gateway_unsupported_label: 'Одобрение не поддерживается',
+    approval_pending_count: (n) => `1 из ${n} в ожидании`,
     untitled: 'Без названия',
     n_messages: (n) => `${n} сообщений`,
     load_older_messages: '↑ Прокрутите вверх или нажмите, чтобы загрузить ранние сообщения',
@@ -6987,7 +6991,7 @@ const LOCALES = {
     _speech: 'es-ES',
     // boot.js
     cancelling: 'Cancelando…',
-    cancel_failed: 'Error al cancelar: ',
+    cancel_failed: 'Error al cancelar.',
     mic_denied: 'Acceso al micrófono denegado. Revisa los permisos del navegador.',
     mic_insecure_origin: 'La entrada de voz necesita una conexión segura. Abre Hermes con HTTPS o desde localhost para usar el micrófono.',
     mic_no_speech: 'No se detectó voz. Inténtalo de nuevo.',
@@ -7107,6 +7111,7 @@ const LOCALES = {
     approval_responding: 'Respondiendo…',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Las aprobaciones no son compatibles',
+    approval_pending_count: (n) => `1 de ${n} pendientes`,
     clarify_heading: 'Se necesita aclaración',
     clarify_hint: 'Elige una opción o escribe tu propia respuesta abajo.',
     clarify_other: 'Otra',
@@ -8675,7 +8680,7 @@ const LOCALES = {
     _speech: 'de-DE',
     // boot.js
     cancelling: 'Wird abgebrochen\u2026',
-    cancel_failed: 'Abbrechen fehlgeschlagen: ',
+    cancel_failed: 'Abbrechen fehlgeschlagen.',
     mic_denied: 'Mikrofonzugriff verweigert. Überprüfen Sie die Browserberechtigungen.',
     mic_insecure_origin: 'Für die Spracheingabe ist eine sichere Verbindung erforderlich. Öffne Hermes über HTTPS oder von localhost, um das Mikrofon zu verwenden.',
     mic_no_speech: 'Keine Sprache erkannt. Versuchen Sie es erneut.',
@@ -8795,6 +8800,7 @@ const LOCALES = {
     approval_responding: 'Antwortet\u2026',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Genehmigungen nicht unterstützt',
+    approval_pending_count: (n) => `1 von ${n} ausstehend`,
     clarify_heading: 'Klärung erforderlich',
     clarify_hint: 'Wähle eine Option oder schreibe deine eigene Antwort unten.',
     clarify_other: 'Andere',
@@ -10357,7 +10363,7 @@ const LOCALES = {
     _speech: 'zh-CN',
     // boot.js
     cancelling: '正在取消...',
-    cancel_failed: '取消失败：',
+    cancel_failed: '取消失败。',
     mic_denied: '麦克风访问被拒绝，请检查浏览器权限。',
     mic_insecure_origin: '语音输入需要安全连接。请通过 HTTPS 协议或从本地主机打开Hermes，以便使用麦克风。', 
     mic_no_speech: '没有检测到语音，请再试一次。',
@@ -10477,6 +10483,7 @@ const LOCALES = {
     approval_responding: '处理中…',
     approval_gateway_unsupported: '批准操作需要更新版本的网关。请升级已连接的Hermes网关以实现此功能。',
     approval_gateway_unsupported_label: '不支持批准',
+    approval_pending_count: (n) => `${n} 个待处理中的第 1 个`,
     clarify_heading: '需要澄清',
     clarify_hint: '请选择一个选项，或在下方输入你自己的回答。',
     clarify_other: '其他',
@@ -12034,7 +12041,7 @@ const LOCALES = {
     _speech: 'zh-TW',
     // boot.js
     cancelling: '正在取消……',
-    cancel_failed: '取消失敗：',
+    cancel_failed: '取消失敗。',
     mic_denied: '麥克風存取遭拒。請檢查瀏覽器權限。',
     mic_insecure_origin: '語音輸入需要安全連線。請透過 HTTPS 或 localhost 開啟 Hermes 才能使用麥克風。',
     mic_no_speech: '未偵測到語音。請再試一次。',
@@ -12182,6 +12189,7 @@ const LOCALES = {
     approval_responding: '正在回應……',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: '不支援核准',
+    approval_pending_count: (n) => `${n} 個待處理中的第 1 個`,
     clarify_heading: '需要釐清',
     clarify_hint: '請選擇一個選項，或在下方輸入回答。',
     clarify_other: '其他',
@@ -13778,7 +13786,7 @@ const LOCALES = {
     _speech: 'pt-BR',
     // boot.js
     cancelling: 'Cancelando…',
-    cancel_failed: 'Falha ao cancelar: ',
+    cancel_failed: 'Falha ao cancelar.',
     mic_denied: 'Acesso ao microfone negado. Verifique as permissões do navegador.',
     mic_insecure_origin: 'A entrada de voz precisa de uma conexão segura. Abra o Hermes por HTTPS ou a partir do localhost para usar o microfone.',
     mic_no_speech: 'Nenhuma fala detectada. Tente novamente.',
@@ -13832,6 +13840,7 @@ const LOCALES = {
     approval_responding: 'Respondendo…',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Aprovações não suportadas',
+    approval_pending_count: (n) => `1 de ${n} pendentes`,
     clarify_heading: 'Esclarecimento necessário',
     clarify_hint: 'Escolha uma opção ou digite sua resposta abaixo.',
     clarify_other: 'Outro',
@@ -15340,7 +15349,7 @@ const LOCALES = {
     _speech: 'ko-KR',
     // boot.js
     cancelling: '취소 중\u2026',
-    cancel_failed: '취소 실패: ',
+    cancel_failed: '취소에 실패했습니다.',
     mic_denied: '마이크 접근이 거부되었습니다. 브라우저 권한을 확인하세요.',
     mic_insecure_origin: '음성 입력에는 보안 연결이 필요합니다. 마이크를 사용하려면 HTTPS 또는 localhost를 통해 Hermes를 여세요.',
     mic_no_speech: '음성이 감지되지 않았습니다. 다시 시도하세요.',
@@ -15460,6 +15469,7 @@ const LOCALES = {
     approval_responding: '응답 중\u2026',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: '승인이 지원되지 않음',
+    approval_pending_count: (n) => `${n}개 중 1개 대기 중`,
     clarify_heading: '확인 필요',
     clarify_hint: '선택지를 고르거나 아래에 직접 입력하세요.',
     clarify_other: '기타',
@@ -17071,7 +17081,7 @@ const LOCALES = {
     _label: 'Français',
     _speech: 'fr-FR',
     cancelling: 'Annulation\u2026',
-    cancel_failed: 'Échec de l\'annulation : ',
+    cancel_failed: 'Échec de l\'annulation.',
     mic_denied: 'Accès au microphone refusé. Vérifiez les autorisations du navigateur.',
     mic_insecure_origin: 'La saisie vocale nécessite une connexion sécurisée. Ouvrez Hermes en HTTPS ou depuis localhost pour utiliser le microphone.',
     mic_no_speech: 'Aucune parole détectée. Réessayez.',
@@ -17212,6 +17222,7 @@ const LOCALES = {
     approval_responding: 'Réponse en cours\u2026',
     approval_gateway_unsupported: 'Les approbations nécessitent une passerelle plus récente. Mettez à niveau la passerelle Hermes connectée pour activer cette fonctionnalité.',
     approval_gateway_unsupported_label: 'Approbations non prises en charge',
+    approval_pending_count: (n) => `1 sur ${n} en attente`,
     clarify_heading: 'Clarification nécessaire',
     clarify_hint: 'Choisissez une option ou saisissez votre propre réponse ci-dessous.',
     clarify_other: 'Autre',
@@ -19980,6 +19991,7 @@ const LOCALES = {
     approval_desc_prefix: 'Byl zjištěn nebezpečný příkaz',
     approval_gateway_unsupported: 'Schválení vyžadují novější gateway. Upgradujte připojený Hermes gateway.',
     approval_gateway_unsupported_label: 'Schválení nejsou podporována',
+    approval_pending_count: (n) => `1 z ${n} čeká`,
     approval_heading: 'Požadováno schválení',
     approval_responding: 'Reakce\u2026',
     auto_compress_label: 'Automatická komprese',
@@ -20000,7 +20012,7 @@ const LOCALES = {
     btw_no_answer: 'Nebyla přijata žádná odpověď.',
     busy_interrupt_confirm: 'Přerušeno — odeslání nové zprávy',
     busy_steer_fallback: 'Řízení není k dispozici — návrh byl obnoven',
-    cancel_failed: 'Zrušení selhalo: ',
+    cancel_failed: 'Zrušení selhalo.',
     cancel_unavailable: 'Zrušení není k dispozici.',
     cancelling: 'Rušení\u2026',
     clarify_heading: 'Zde je zapotřebí vyjasnění.',
@@ -20492,7 +20504,7 @@ const LOCALES = {
     _speech: 'tr-TR',
     // boot.js
     cancelling: 'İptal ediliyor\u2026',
-    cancel_failed: 'İptal başarısız oldu:',
+    cancel_failed: 'İptal başarısız oldu.',
     mic_denied: 'Mikrofon erişimi reddedildi. Tarayıcı izinlerini kontrol edin.',
     mic_insecure_origin: 'Sesli giriş güvenli bir bağlantı gerektirir. Mikrofonu kullanmak için Hermes\'i HTTPS üzerinden veya localhost\'tan açın.',
     mic_no_speech: 'Konuşma algılanmadı. Tekrar deneyin.',
@@ -20629,6 +20641,7 @@ const LOCALES = {
     approval_responding: 'Yanıt veriliyor\u2026',
     approval_gateway_unsupported: 'Approvals require a newer gateway. Upgrade the connected Hermes gateway to enable this.',
     approval_gateway_unsupported_label: 'Onaylar desteklenmiyor',
+    approval_pending_count: (n) => `${n} bekleyenden 1'i`,
     clarify_heading: 'Açıklama gerekli',
     clarify_hint: 'Bir seçim yapın veya kendi cevabınızı aşağıya yazın.',
     clarify_other: 'Diğer',
@@ -22225,7 +22238,7 @@ const LOCALES = {
     _speech: 'pl-PL',
     // boot.js
     cancelling: 'Anulowanie…',
-    cancel_failed: 'Anulowanie nie powiodło się: ',
+    cancel_failed: 'Anulowanie nie powiodło się.',
     mic_denied: 'Odmowa dostępu do mikrofonu. Sprawdź uprawnienia przeglądarki.',
     mic_insecure_origin: 'Wprowadzanie głosowe wymaga bezpiecznego połączenia. Otwórz Hermes przez HTTPS lub z localhost, aby użyć mikrofonu.',
     mic_no_speech: 'Nie wykryto mowy. Spróbuj ponownie.',
@@ -22373,6 +22386,7 @@ const LOCALES = {
     approval_responding: 'Odpowiadanie…',
     approval_gateway_unsupported: 'Zatwierdzenia wymagają nowszej bramy. Zaktualizuj połączoną bramę Hermes, aby włączyć tę funkcję.',
     approval_gateway_unsupported_label: 'Zatwierdzenia nie są obsługiwane',
+    approval_pending_count: (n) => `1 z ${n} oczekujących`,
     clarify_heading: 'Wymagane wyjaśnienie',
     clarify_hint: 'Wybierz opcję lub wpisz własną odpowiedź poniżej.',
     clarify_other: 'Inne',
@@ -23962,7 +23976,7 @@ const LOCALES = {
     _speech: 'vi-VN',
     // boot.js
     cancelling: 'Đang hủy…',
-    cancel_failed: 'Hủy thất bại: ',
+    cancel_failed: 'Hủy thất bại.',
     mic_denied: 'Quyền truy cập microphone bị từ chối. Kiểm tra quyền trình duyệt.',
     mic_no_speech: 'Không phát hiện giọng nói. Hãy thử lại.',
     mic_network: 'Nhận diện giọng nói không khả dụng.',
@@ -25467,6 +25481,7 @@ const LOCALES = {
 
     approval_gateway_unsupported: 'Phê duyệt cần gateway mới hơn. Hãy nâng cấp Hermes gateway đang kết nối để bật tính năng này.',
     approval_gateway_unsupported_label: 'Không hỗ trợ phê duyệt',
+    approval_pending_count: (n) => `1 trong ${n} đang chờ`,
     approval_skip_all: '⚡ Bỏ qua tất cả trong phiên này',
     approval_skip_all_title: 'Bỏ qua mọi yêu cầu phê duyệt cho phiên này',
     checkpoint_title: 'Checkpoint',

static/index.html

@@ -496,7 +496,7 @@
           </div>
           <div class="approval-desc" id="approvalDesc"></div>
           <div class="approval-cmd" id="approvalCmd"></div>
-          <div class="approval-counter" id="approvalCounter" style="display:none;font-size:0.75em;opacity:0.6;margin-top:4px;"></div>
+          <div class="approval-counter" id="approvalCounter" style="display:none;"></div>
           <div class="approval-btns" id="approvalBtns">
             <button class="approval-btn once" id="approvalBtnOnce" onclick="respondApproval('once')" title="Allow this one command (Enter)" data-i18n-title="approval_btn_once_title">
               <span class="approval-btn-icon"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2.5" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><polyline points="20 6 9 17 4 12"/></svg></span>

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'),null,'error');
         } else {
           showToast(`Queued: "${text.slice(0,40)}${text.length>40?'…':''}"`,2000);
         }
@@ -5619,7 +5619,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
     source.addEventListener('approval',e=>{
       const d=JSON.parse(e.data);
       _applyToAnchor('approval',d,e);
-      showApprovalForSession(activeSid, d, 1);
+      showApprovalForSession(activeSid, d, d.pending_count || 1);
       playAttentionSound(_attentionSoundKey(activeSid,'approval',1));
       sendBrowserNotification('Approval required',d.description||'Tool approval needed',{sid:activeSid});
     });
@@ -7106,7 +7106,9 @@ function showApprovalCard(pending, pendingCount) {
   const counter = $("approvalCounter");
   if (counter) {
     if (pendingCount && pendingCount > 1) {
-      counter.textContent = "1 of " + pendingCount + " pending";
+      counter.textContent = (typeof t === "function")
+        ? t("approval_pending_count", pendingCount)
+        : ("1 of " + pendingCount + " pending");
       counter.style.display = "";
     } else {
       counter.style.display = "none";

static/sessions.js

@@ -2315,6 +2315,7 @@ async function loadSession(sid){
       setComposerStatus('');
       updateQueueBadge(sid);
       syncTopbar();renderMessages(sameSessionForceReload?{preserveScroll:true}:undefined);
+      startApprovalPolling(sid);
       if(typeof resumeManualCompressionForSession==='function') resumeManualCompressionForSession(sid);
       // Workspace refresh is guarded by session id inside loadDir(); keep it
       // after the transcript's first paint so chat switching is not competing
@@ -4905,8 +4906,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'),null,'error');
       }
     ));
   }

static/style.css

@@ -1992,6 +1992,7 @@
   .approval-card.collapsed .approval-desc,.approval-card.collapsed .approval-cmd,.approval-card.collapsed .approval-counter,.approval-card.collapsed .approval-btns{display:none;}
   .approval-desc{font-size:12px;color:var(--muted);margin-bottom:8px;line-height:1.5;}
   .approval-cmd{background:var(--code-bg);border:1px solid var(--border);border-radius:8px;padding:8px 12px;font-family:"SF Mono",ui-monospace,monospace;font-size:12px;color:var(--pre-text);white-space:pre-wrap;word-break:break-all;margin-bottom:14px;max-height:120px;overflow-y:auto;}
+  .approval-counter{font-size:11px;color:var(--muted);margin:0 0 12px;font-variant-numeric:tabular-nums;letter-spacing:0.01em;}
   .approval-btns{display:flex;gap:8px;flex-wrap:wrap;align-items:center;}
   .approval-btn{display:inline-flex;align-items:center;gap:6px;padding:8px 16px;border-radius:8px;font-size:12px;font-weight:600;border:1px solid var(--border2);background:var(--hover-bg);color:var(--text);cursor:pointer;transition:all .15s;white-space:nowrap;}
   .approval-btn:hover{background:rgba(255,255,255,0.12);transform:translateY(-1px);box-shadow:0 2px 8px rgba(0,0,0,0.2);}

static/ui.js

@@ -7912,7 +7912,7 @@ async function handleComposerPrimaryAction(){
   const action=typeof getComposerPrimaryAction==='function'?getComposerPrimaryAction():'send';
   if(action==='disabled') return;
   if(action==='stop'){
-    if(typeof cancelStream==='function') await cancelStream('composer-stop');
+    if(typeof cancelStream==='function' && !await cancelStream('composer-stop')) showToast(t('cancel_failed'),null,'error');
     return;
   }
   await send();

tests/test_approval_queue.py

@@ -7,6 +7,8 @@ approval_id so /api/approval/respond can target a specific entry.
 import json
 import pathlib
 import re
+import shutil
+import subprocess
 import sys
 
 REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve()
@@ -108,6 +110,39 @@ def test_polling_passes_count_to_show():
         "Poll loop must pass data.pending_count through showApprovalForSession"
 
 
+def test_chat_stream_approval_listener_renders_received_count():
+    """A chat-stream approval event with count two renders the head and count."""
+    node = shutil.which("node")
+    if node is None:
+        import pytest
+        pytest.skip("node is required for the approval listener harness")
+    start = MESSAGES_JS.index("source.addEventListener('approval',e=>{")
+    end_marker = "\n    });"
+    end = MESSAGES_JS.index(end_marker, start) + len(end_marker)
+    listener = MESSAGES_JS[start:end]
+    script = f"""
+    const rendered = [];
+    let activeSid = 'sid-browser';
+    const source = {{ listeners: {{}}, addEventListener(name, cb) {{ this.listeners[name] = cb; }} }};
+    function _applyToAnchor() {{}}
+    function showApprovalForSession(sid, data, count) {{ rendered.push({{sid, data, count}}); }}
+    function playAttentionSound() {{}}
+    function _attentionSoundKey() {{ return 'approval'; }}
+    function sendBrowserNotification() {{}}
+    {listener}
+    source.listeners.approval({{ data: JSON.stringify({{command: 'head', approval_id: 'a', pending_count: 2}}) }});
+    process.stdout.write(JSON.stringify(rendered));
+    """
+    result = subprocess.run([node, "-e", script], capture_output=True, text=True, check=False)
+    assert result.returncode == 0, result.stderr
+    rendered = json.loads(result.stdout)
+    assert rendered == [{
+        "sid": "sid-browser",
+        "data": {"command": "head", "approval_id": "a", "pending_count": 2},
+        "count": 2,
+    }]
+
+
 # ---------------------------------------------------------------------------
 # HTML: counter element present
 # ---------------------------------------------------------------------------

tests/test_approval_sse.py

@@ -113,8 +113,10 @@ class TestSSEStaticAnalysis:
         cb_start = streaming_src.find("def _approval_notify_cb(approval_data):")
         cb_end = streaming_src.find("_reg_notify(session_id, _approval_notify_cb)", cb_start)
         cb_body = streaming_src[cb_start:cb_end]
-        assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \
+        assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \
             "_approval_notify_cb must mirror approval data into polling state before SSE"
+        assert '"pending_count": total' in cb_body, \
+            "_approval_notify_cb must emit the reconciled pending count"
         assert "put('approval', approval_data)" in cb_body, \
             "_approval_notify_cb must still emit the approval SSE event"
 

tests/test_approval_unblock.py

@@ -232,11 +232,57 @@ class TestApprovalModuleExports:
         assert cb_start != -1, "_approval_notify_cb must exist"
         cb_end = STREAMING_SRC.find("_reg_notify(session_id, _approval_notify_cb)", cb_start)
         cb_body = STREAMING_SRC[cb_start:cb_end]
-        assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \
+        assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \
             "approval notify callback must mirror approval data into polling state"
+        assert '"pending_count": total' in cb_body, \
+            "approval notify callback must publish the reconciled pending count"
         assert "put('approval', approval_data)" in cb_body, \
             "approval notify callback must still push the SSE event"
 
+    def test_reversed_local_callbacks_publish_all_parked_entries(self):
+        """The local producer must count every exact parked entry, not only its head."""
+        import api.route_approvals as ra
+        from api import routes
+
+        sid = f"unit-reversed-local-{uuid.uuid4().hex[:8]}"
+        entry_a = _ApprovalEntry({
+            "command": "same-command",
+            "description": "same-description",
+            "pattern_key": "same-pattern",
+            "pattern_keys": ["same-pattern"],
+        })
+        entry_b = _ApprovalEntry(dict(entry_a.data))
+        events = []
+        with _lock:
+            _gateway_queues[sid] = [entry_a, entry_b]
+            ra._pending.pop(sid, None)
+
+        def local_producer(approval_data):
+            head, total = ra.submit_gateway_pending_mirror(sid, approval_data)
+            events.append((head, total))
+
+        try:
+            local_producer(entry_b.data)
+            local_producer(entry_a.data)
+
+            assert [total for _head, total in events] == [2, 2]
+            assert all(head["command"] == "same-command" for head, _total in events)
+            head_id = events[0][0]["approval_id"]
+            assert head_id == events[1][0]["approval_id"]
+            assert routes._resolve_approval_legacy(sid, head_id, "once") is True
+            assert entry_a.event.is_set()
+            assert not entry_b.event.is_set()
+            with _lock:
+                successor = ra._pending[sid][0]
+            assert successor["approval_id"] != head_id
+            with _lock:
+                assert ra.reconcile_gateway_pending_mirror_locked(sid)[1] == 1
+        finally:
+            with _lock:
+                ra._pending.pop(sid, None)
+                _gateway_queues.pop(sid, None)
+                ra._pending.pop(sid, None)
+
 
 # ── HTTP regression tests (test server, port 8788) ───────────────────────────
 
@@ -481,8 +527,8 @@ class TestApprovalHTTPEndpoints:
                 r._gateway_queues.pop(sid, None)
                 pass  # no external token state to clean
 
-    def test_gateway_mirror_without_run_id_returns_explicit_conflict(self, monkeypatch):
-        """Gateway-mirrored approvals without relayable run state must stay actionable."""
+    def test_gateway_mirror_without_run_id_and_no_producer_returns_explicit_conflict(self, monkeypatch):
+        """A no-run mirror stays actionable when no local producer is parked."""
         from api import routes as r
         from api import route_approvals as ra
         from api.gateway_chat import _STREAM_RUN_IDS
@@ -515,7 +561,6 @@ class TestApprovalHTTPEndpoints:
             r._pending.pop(sid, None)
             r._gateway_queues.pop(sid, None)
             _STREAM_RUN_IDS.pop(stream_id, None)
-            r._gateway_queues[sid] = [_ApprovalEntry(approval)]
         try:
             ra.submit_gateway_pending_mirror(sid, approval)
             with _lock:
@@ -539,10 +584,172 @@ class TestApprovalHTTPEndpoints:
                 assert isinstance(pending_queue, list)
                 assert len(pending_queue) == 1
                 assert pending_queue[0]["approval_id"] == approval_id
-                assert sid in r._gateway_queues
-                assert not r._gateway_queues[sid][0].event.is_set()
+                assert sid not in r._gateway_queues
         finally:
             with _lock:
                 r._pending.pop(sid, None)
                 r._gateway_queues.pop(sid, None)
                 _STREAM_RUN_IDS.pop(stream_id, 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
+        from api import route_approvals as ra
+        from api.gateway_chat import _STREAM_RUN_IDS
+
+        sid = f"http-gateway-local-{uuid.uuid4().hex[:8]}"
+        stream_id = f"stream-local-{uuid.uuid4().hex[:8]}"
+        approval = {"command": "rm -rf /tmp/local", "description": "local"}
+        sibling = {"command": "rm -rf /tmp/sibling", "description": "sibling"}
+        captured = {}
+
+        def fake_j(handler, data, status=200, extra_headers=None):
+            captured["payload"] = data
+            captured["status"] = status
+            return data
+
+        monkeypatch.setattr(r, "j", fake_j)
+        monkeypatch.setattr(r, "get_session", lambda _sid: SimpleNamespace(active_stream_id=stream_id))
+        monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway")
+        entry = _ApprovalEntry(approval)
+        sibling_entry = _ApprovalEntry(sibling)
+        with _lock:
+            r._pending.pop(sid, None)
+            r._gateway_queues[sid] = [entry, sibling_entry]
+        try:
+            ra.submit_gateway_pending_mirror(sid, approval)
+            with _lock:
+                approval_id = r._pending[sid][0]["approval_id"]
+
+            r._handle_approval_respond(
+                object(), {"session_id": sid, "choice": "once", "approval_id": approval_id}
+            )
+
+            assert captured["status"] == 200
+            assert captured["payload"] == {"ok": True, "choice": "once", "local_retired": True}
+            assert entry.event.is_set()
+            assert entry.result == "once"
+            assert not sibling_entry.event.is_set()
+            with _lock:
+                assert r._pending[sid][0]["command"] == sibling["command"]
+        finally:
+            with _lock:
+                r._pending.pop(sid, None)
+                r._gateway_queues.pop(sid, None)
+                _STREAM_RUN_IDS.pop(stream_id, None)
+
+    def test_gateway_no_run_mirror_survives_pending_and_attention_reads_until_explicit_teardown(self, monkeypatch):
+        """A missing-producer 409 stays visible until a real teardown retires it."""
+        from api import routes as r
+        from api import route_approvals as ra
+        from api.gateway_chat import _STREAM_RUN_IDS, _cleanup_gateway_pending_mirror
+
+        sid = f"http-gateway-race-{uuid.uuid4().hex[:8]}"
+        stream_id = f"stream-race-{uuid.uuid4().hex[:8]}"
+        approval = {"command": "rm -rf /tmp/race", "description": "race"}
+        captured = {}
+
+        def fake_j(handler, data, status=200, extra_headers=None):
+            captured["payload"] = data
+            captured["status"] = status
+            return data
+
+        monkeypatch.setattr(r, "j", fake_j)
+        monkeypatch.setattr(r, "get_session", lambda _sid: SimpleNamespace(active_stream_id=stream_id))
+        monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway")
+        entry = _ApprovalEntry(approval)
+        with _lock:
+            r._pending.pop(sid, None)
+            r._gateway_queues[sid] = [entry]
+        try:
+            ra.submit_gateway_pending_mirror(sid, approval)
+            with _lock:
+                approval_id = r._pending[sid][0]["approval_id"]
+                r._gateway_queues.pop(sid, None)
+
+            r._handle_approval_respond(
+                object(), {"session_id": sid, "choice": "once", "approval_id": approval_id}
+            )
+
+            assert captured["status"] == 409
+            assert captured["payload"] == {
+                "ok": False,
+                "choice": "once",
+                "relayed": False,
+                "code": "gateway_run_unavailable",
+                "error": r._GATEWAY_APPROVAL_RELAY_UNAVAILABLE,
+            }
+            assert not entry.event.is_set()
+            pending = r._handle_approval_pending(
+                object(), urllib.parse.urlparse(
+                    f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
+                )
+            )
+            assert pending["pending"]["approval_id"] == approval_id
+            assert pending["pending_count"] == 1
+            assert r._session_attention_summary(sid) == {
+                "kind": "approval", "count": 1, "severity": "critical"
+            }
+            _cleanup_gateway_pending_mirror(sid)
+            pending = r._handle_approval_pending(
+                object(), urllib.parse.urlparse(
+                    f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
+                )
+            )
+            assert pending == {"pending": None, "pending_count": 0}
+            assert r._session_attention_summary(sid) is None
+        finally:
+            with _lock:
+                r._pending.pop(sid, None)
+                r._gateway_queues.pop(sid, None)
+                _STREAM_RUN_IDS.pop(stream_id, None)
+
+    def test_gateway_no_run_non_head_response_resolves_only_exact_producer(self, monkeypatch):
+        """An exact non-head response wakes only its matching producer."""
+        from api import routes as r
+        from api import route_approvals as ra
+
+        sid = f"http-gateway-non-head-{uuid.uuid4().hex[:8]}"
+        approval_a = {"command": "rm -rf /tmp/a", "description": "a"}
+        approval_b = {"command": "rm -rf /tmp/b", "description": "b"}
+        entry_a = _ApprovalEntry(approval_a)
+        entry_b = _ApprovalEntry(approval_b)
+        captured = {}
+
+        def fake_j(handler, data, status=200, extra_headers=None):
+            captured["payload"] = data
+            captured["status"] = status
+            return data
+
+        monkeypatch.setattr(r, "j", fake_j)
+        monkeypatch.setattr(r, "get_session", lambda _sid: SimpleNamespace(active_stream_id=None))
+        monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway")
+        with _lock:
+            r._pending.pop(sid, None)
+            r._gateway_queues[sid] = [entry_a, entry_b]
+        try:
+            ra.submit_gateway_pending_mirror(sid, approval_a)
+            ra.submit_gateway_pending_mirror(sid, approval_b)
+            with _lock:
+                approval_b_id = next(
+                    item["approval_id"] for item in r._pending[sid]
+                    if item["command"] == approval_b["command"]
+                )
+            r._handle_approval_respond(
+                object(), {"session_id": sid, "choice": "once", "approval_id": approval_b_id}
+            )
+            assert captured["status"] == 200
+            assert entry_b.event.is_set()
+            assert entry_b.result == "once"
+            assert not entry_a.event.is_set()
+            pending = r._handle_approval_pending(
+                object(), urllib.parse.urlparse(
+                    f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
+                )
+            )
+            assert pending["pending"]["command"] == approval_a["command"]
+            assert pending["pending_count"] == 1
+        finally:
+            with _lock:
+                r._pending.pop(sid, None)
+                r._gateway_queues.pop(sid, None)

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_gateway_approval_legacy_path.py

@@ -6,6 +6,7 @@ PR #4495 fixed the runs API path but left the legacy path without approval handl
 from __future__ import annotations
 
 import json
+import threading
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
@@ -336,6 +337,19 @@ def test_legacy_approval_records_run_id_for_response_relay():
         # (connection still open, run parked) — this test's stream already ran to
         # completion, which cleared active_stream_id and popped the mapping.
         _STREAM_RUN_IDS[stream_id] = "run-legacy-1"
+        from types import SimpleNamespace
+        from api import route_approvals as ra
+        with ra._lock:
+            ra._gateway_queues["sess-legacy-runid"] = [SimpleNamespace(data={
+                "command": "rm -rf /tmp/test",
+                "description": "Delete temporary files",
+                "pattern_key": "dangerous_command",
+                "pattern_keys": ["dangerous_command"],
+                "approval_id": "appr-legacy-runid",
+                "run_id": "run-legacy-1",
+            })]
+            live_entry = ra._gateway_queues["sess-legacy-runid"][0]
+        ra.submit_gateway_pending_mirror("sess-legacy-runid", live_entry.data)
         relay_session = MagicMock()
         relay_session.active_stream_id = stream_id
         captured = {}
@@ -360,7 +374,7 @@ def test_legacy_approval_records_run_id_for_response_relay():
         assert captured.get("url", "") == "http://gw:8642/v1/runs/run-legacy-1/approval", (
             f"approval respond must relay to the gateway run; got {captured.get('url')!r}"
         )
-        assert captured["body"] == {"choice": "once", "approval_id": "appr-legacy-runid"}
+        assert captured["body"] == {"choice": "once", "approval_id": ""}
         handler.send_response.assert_called_with(200)
     finally:
         with STREAMS_LOCK:
@@ -447,8 +461,8 @@ def test_legacy_teardown_clears_stale_gateway_mirror_and_notifies_empty_state():
             ra._gateway_queues.pop(session_id, None)
 
 
-def test_legacy_teardown_preserves_live_gateway_head_mirror():
-    """A live gateway head must still be mirrored after legacy teardown runs."""
+def test_legacy_teardown_retires_live_gateway_head_mirror():
+    """A mapped legacy run retires its live gateway mirror during outer teardown."""
     from types import SimpleNamespace
     from api import route_approvals as ra
     from api.config import STREAMS, STREAMS_LOCK
@@ -507,14 +521,11 @@ def test_legacy_teardown_preserves_live_gateway_head_mirror():
 
         payloads = _drain_queue(subscriber)
         assert payloads, "Expected mirrored approval notifications"
-        assert payloads[-1]["pending"]["_gateway_mirror"] is True
-        assert payloads[-1]["pending_count"] == 1
+        assert payloads[-1]["pending"] is None
+        assert payloads[-1]["pending_count"] == 0
         with ra._lock:
             pending = ra._pending.get(session_id)
-            assert isinstance(pending, list)
-            assert len(pending) == 1
-            assert pending[0]["approval_id"] == approval_data["approval_id"]
-            assert pending[0]["_gateway_mirror"] is True
+            assert pending is None
     finally:
         ra._approval_sse_unsubscribe(session_id, subscriber)
         with STREAMS_LOCK:
@@ -686,7 +697,7 @@ def test_mirrored_run_id_survives_active_stream_loss():
         assert captured.get("url", "") == f"http://gw:8642/v1/runs/{run_id}/approval", (
             f"approval respond must relay to the mirrored gateway run; got {captured.get('url')!r}"
         )
-        assert captured["body"] == {"choice": "once", "approval_id": approval_id}
+        assert captured["body"] == {"choice": "once", "approval_id": ""}
         handler.send_response.assert_called_with(200)
         assert entry.event.is_set(), "mirrored gateway approval was not resolved"
         assert entry.result == "once"
@@ -752,8 +763,8 @@ def test_gateway_mode_no_pending_click_stays_non_409():
     assert captured["payload"].get("code") != "gateway_run_unavailable"
 
 
-def test_legacy_approval_without_run_id_stays_actionable():
-    """Legacy approvals without a run_id must fail explicitly and keep the mirror live."""
+def test_legacy_approval_without_run_id_retires_locally():
+    """A no-run legacy mirror retires locally instead of becoming a ghost card."""
     from types import SimpleNamespace
     from api import routes as r
     from api import route_approvals as ra
@@ -832,22 +843,15 @@ def test_legacy_approval_without_run_id_stays_actionable():
         ]
         assert approval_events
         approval_data = approval_events[0][1]
+        local_entry = SimpleNamespace(data=dict(approval_data), event=threading.Event(), result=None)
         with ra._lock:
-            r._gateway_queues[session_id] = [SimpleNamespace(data=dict(approval_data))]
+            r._gateway_queues[session_id] = [local_entry]
         ra.submit_gateway_pending_mirror(session_id, approval_data)
         with ra._lock:
             pending_queue = r._pending.get(session_id)
             assert isinstance(pending_queue, list)
             approval_id = pending_queue[0]["approval_id"]
 
-        # The relay-unavailable 409 is only meaningful when the WebUI is
-        # actually running the gateway chat backend. On a gateway deployment
-        # the backend env is process-wide (not just during the stream), so
-        # assert the 409 with HERMES_WEBUI_CHAT_BACKEND=gateway active at
-        # respond time. Without this scope the handler now (correctly) treats
-        # a mirrored approval on the default LOCAL backend as locally
-        # resolvable and falls through instead of 409ing — see
-        # test_issue4771_local_approval_regression.py (#4771 follow-up).
         with patch.dict("os.environ", {"HERMES_WEBUI_CHAT_BACKEND": "gateway"}), \
              patch("api.routes.get_session", return_value=mock_session), \
              patch("api.routes.j", new=fake_j):
@@ -856,13 +860,12 @@ def test_legacy_approval_without_run_id_stays_actionable():
                 {"session_id": session_id, "choice": "once", "approval_id": approval_id},
             )
 
-        assert captured["status"] == 409
-        assert captured["payload"]["code"] == "gateway_run_unavailable"
-        assert captured["payload"]["error"] == r._GATEWAY_APPROVAL_RELAY_UNAVAILABLE
+        assert captured["status"] == 200
+        assert captured["payload"] == {"ok": True, "choice": "once", "local_retired": True}
+        assert local_entry.event.is_set()
+        assert local_entry.result == "once"
         with ra._lock:
-            pending_queue = r._pending.get(session_id)
-            assert isinstance(pending_queue, list)
-            assert pending_queue[0]["approval_id"] == approval_id
+            assert session_id not in r._pending
     finally:
         with STREAMS_LOCK:
             STREAMS.pop(stream_id, None)

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,10 +183,206 @@ def test_local_mirrored_approval_deny_resolves():
         _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
-    visible instead of silently failing."""
+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_local_plain_pending_exact_id_beats_remote_mirror():
+    sid = f"local-plain-pending-id-{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:
+        with ta._lock:
+            ta._pending[sid] = [
+                {"approval_id": "shared-id", "command": "local"},
+                {
+                    ra._GATEWAY_MIRROR_FLAG: True,
+                    "run_id": "remote-run",
+                    "approval_id": "shared-id",
+                    "command": "remote",
+                },
+            ]
+        handler = _FakeHandler()
+        with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval, \
+             patch.object(routes, "_resolve_approval_legacy", return_value=True) as resolve_legacy:
+            routes._handle_approval_respond(
+                handler,
+                {"session_id": sid, "choice": "once", "approval_id": "shared-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
+        resolve_legacy.assert_called_once_with(sid, "shared-id", "once")
+        respond_approval.assert_not_called()
+    finally:
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _cleanup(sid)
+
+
+def test_local_plain_pending_exact_id_beats_remote_first_queue_order():
+    sid = f"local-plain-pending-remote-first-{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:
+        with ta._lock:
+            ta._pending[sid] = [
+                {
+                    ra._GATEWAY_MIRROR_FLAG: True,
+                    "run_id": "remote-run",
+                    "approval_id": "shared-id",
+                    "command": "remote",
+                },
+                {"approval_id": "shared-id", "command": "local"},
+            ]
+        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": "shared-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()
+        with ta._lock:
+            queue = ta._pending.get(sid) or []
+            assert len(queue) == 1
+            assert queue[0].get("command") == "remote"
+            assert queue[0].get(ra._GATEWAY_MIRROR_FLAG) is True
+    finally:
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _cleanup(sid)
+
+
+def test_local_plain_pending_without_id_beats_active_remote_candidate_run():
+    sid = f"local-plain-pending-no-id-{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:
+        with ta._lock:
+            ta._pending[sid] = [{"command": "local"}]
+        handler = _FakeHandler()
+        with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval, \
+             patch.object(routes, "_resolve_approval_legacy", return_value=True) as resolve_legacy:
+            routes._handle_approval_respond(
+                handler,
+                {"session_id": sid, "choice": "once"},
+            )
+        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
+        resolve_legacy.assert_called_once_with(sid, "", "once")
+        respond_approval.assert_not_called()
+    finally:
+        _STREAM_RUN_IDS.pop(stream_id, None)
+        _cleanup(sid)
+
+
+def test_gateway_mirrored_approval_without_run_retires_locally():
+    """A gateway mirror without a run retires instead of leaving a ghost card."""
     sid = f"gw-approval-{uuid.uuid4().hex[:8]}"
     _register_session(sid)  # no active_stream_id -> no _STREAM_RUN_IDS mapping
     try:
@@ -197,10 +394,13 @@ def test_gateway_mirrored_approval_without_run_still_409s():
                 {"session_id": sid, "choice": "once", "approval_id": approval_id},
             )
         resp = handler.json()
-        assert handler.status == 409, f"expected 409, got {handler.status}: {resp}"
-        assert resp.get("ok") is False
-        assert resp.get("relayed") is False
-        assert resp.get("code") == "gateway_run_unavailable"
+        assert handler.status == 200, f"expected 200, got {handler.status}: {resp}"
+        assert resp == {"ok": True, "choice": "once", "local_retired": True}
+        assert entry.event.is_set()
+        assert entry.result == "once"
+        with ta._lock:
+            assert sid not in ta._pending
+            assert sid not in ta._gateway_queues
     finally:
         _cleanup(sid)
 

tests/test_issue6008_sidebar_stop_failure_ui.py

@@ -0,0 +1,238 @@
+"""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")
+UI_JS = (REPO / "static" / "ui.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")
+I18N_JS = (REPO / "static" / "i18n.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")
+COMPOSER_PRIMARY_ACTION_SRC = _extract_function(UI_JS, "handleComposerPrimaryAction")
+_TOAST_DEFAULT_NAME = "TOAST_" + "DEFAULT_MS"
+_TOAST_ERROR_DEFAULT_NAME = "TOAST_ERROR_" + "DEFAULT_MS"
+SHOW_TOAST_SRC = _extract_function(UI_JS, "show" + "Toast").replace(
+    _TOAST_DEFAULT_NAME, "toast_default_ms"
+).replace(_TOAST_ERROR_DEFAULT_NAME, "toast_error_default_ms")
+
+
+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'));"
+        "elseshowToast(t('cancel_failed'),null,'error');"
+    ) in compact_commands
+    assert (
+        "if(awaitcancelStream('slash-interrupt'))showToast(t('cmd_interrupt_confirm'),2000);"
+        "elseshowToast(t('cancel_failed'),null,'error');"
+    ) in compact_commands
+    assert (
+        "if(awaitcancelStream('busy-interrupt'))showToast(t('busy_interrupt_confirm'),2000);"
+        "elseshowToast(t('cancel_failed'),null,'error');"
+    ) in compact_messages
+    assert (
+        "if(awaitcancelSessionStream(session))showToast(t('stream_stopped'));"
+        "elseshowToast(t('cancel_failed'),null,'error');"
+    ) in compact_sessions
+    assert (
+        "if(typeofcancelStream==='function'&&!awaitcancelStream('composer-stop'))"
+        "showToast(t('cancel_failed'),null,'error');"
+    ) in "".join(UI_JS.split())
+
+
+_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),
+        capture_output=True,
+        encoding="utf-8",
+        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
+
+
+def test_primary_composer_stop_renders_localized_error_and_preserves_success():
+    cancel_key = "cancel_" + "failed"
+    english = re.search(rf"{cancel_key}:\s*'((?:\\'|[^'])*)'", I18N_JS)
+    japanese = re.search(
+        rf"{cancel_key}:\s*'((?:\\'|[^'])*)'", I18N_JS[I18N_JS.index("ja:"):]
+    )
+    assert english and japanese
+    english_message = english.group(1).replace("\\'", "'")
+    japanese_message = japanese.group(1).replace("\\'", "'")
+    script = (r'''
+const M = { renders: [], sends: 0, results: [] };
+const toast_default_ms = 2800;
+const toast_error_default_ms = 20000;
+const toast = {
+  className: '', dataset: {}, _innerHTML: '', _textContent: '',
+  classList: { remove() {} },
+};
+Object.defineProperty(toast, 'innerHTML', {
+  get() { return this._innerHTML; },
+  set(value) { this._innerHTML = String(value); this._textContent = ''; },
+});
+Object.defineProperty(toast, 'textContent', {
+  get() { return this._textContent; },
+  set(value) { this._textContent = String(value); this._innerHTML = ''; },
+});
+globalThis.$ = (id) => id === 'toast' ? toast : null;
+globalThis.esc = (value) => String(value);
+globalThis.clearToastDismissTimer = () => {};
+globalThis.setToastDismissTimer = (el, duration) => {
+  M.renders.push({ message: el.dataset.toastMessage, className: el.className,
+    duration, copy: el.innerHTML.includes('data-toast-' + 'copy="1"') });
+};
+globalThis.setTimeout = () => 0;
+globalThis.window = {};
+globalThis.S = {};
+globalThis.getComposerPrimaryAction = () => 'stop';
+__SHOW_TOAST_SRC__
+globalThis.send = () => { M.sends += 1; };
+globalThis.cancelStream = async () => M.cancelResult;
+__COMPOSER_PRIMARY_ACTION_SRC__
+for (const [message, result] of [[__ENGLISH__, false], [__JAPANESE__, false], ['unused', true]]) {
+  M.cancelResult = result;
+  globalThis.t = () => message;
+  const before = M.renders.length;
+  await handleComposerPrimaryAction();
+  M.results.push({ result, rendered: M.renders.length - before, sends: M.sends });
+}
+console.log(JSON.stringify(M));
+'''.replace("__SHOW_TOAST_SRC__", SHOW_TOAST_SRC)
+    .replace("__COMPOSER_PRIMARY_ACTION_SRC__", COMPOSER_PRIMARY_ACTION_SRC)
+    .replace("__ENGLISH__", json.dumps(english_message))
+    .replace("__JAPANESE__", json.dumps(japanese_message)))
+    completed = subprocess.run(
+        ["node", "--input-type=module", "-e", script],
+        cwd=str(REPO), capture_output=True, encoding="utf-8", 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["renders"] == [
+        {"message": english_message, "className": "toast show error", "duration": 20000, "copy": True},
+        {"message": japanese_message, "className": "toast show error", "duration": 20000, "copy": True},
+    ]
+    assert result["results"] == [
+        {"result": False, "rendered": 1, "sends": 0},
+        {"result": False, "rendered": 1, "sends": 0},
+        {"result": True, "rendered": 0, "sends": 0},
+    ]

tests/test_runtime_adapter_seam.py

@@ -325,7 +325,7 @@ def test_approval_respond_does_not_fallback_to_oldest_when_explicit_id_is_stale(
     helper_body = src[helper_idx:src.index("def _handle_approval_respond", helper_idx)]
 
     assert "A stale explicit id must not accidentally approve" in helper_body
-    assert "if found_target or not approval_id:" in helper_body
+    assert "if not pending and not approval_id:" in helper_body
     stale_branch = helper_body[helper_body.index("else:", helper_body.index("for i, entry")):helper_body.index("else:\n                pending = queue.pop(0)")]
     assert "pending = None" in stale_branch
     assert "queue.pop(0)" not in stale_branch

tests/test_session_attention_badges.py

@@ -115,6 +115,68 @@ def test_attention_summary_prefers_pending_approvals_over_clarify_questions():
         _clear_attention_state(sid)
 
 
+def test_attention_summary_keeps_direct_runs_mirror_after_stream_pointer_clears():
+    from types import SimpleNamespace
+    from unittest.mock import MagicMock, patch
+
+    import api.route_approvals as approvals
+
+    sid = "attention-retained-gateway-session"
+    _clear_attention_state(sid)
+    try:
+        approvals.submit_gateway_pending_mirror(sid, {
+            "_gateway_mirror": True,
+            "approval_id": "approval-retained",
+            "run_id": "run-retained",
+            "command": "rm -rf /tmp/retained",
+            "description": "Retained approval",
+        })
+        handler = MagicMock()
+        handler.wfile = io.BytesIO()
+
+        with patch("api.routes.get_session", return_value=SimpleNamespace(active_stream_id=None)), \
+             patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval:
+            assert routes._session_attention_summary(sid) == {
+                "kind": "approval",
+                "count": 1,
+                "severity": "critical",
+            }
+            routes._handle_approval_respond(handler, {
+                "session_id": sid,
+                "choice": "deny",
+                "approval_id": "approval-retained",
+            })
+
+        handler.send_response.assert_called_with(200)
+        respond_approval.assert_called_once_with("run-retained", "", "deny")
+        assert json.loads(handler.wfile.getvalue().decode("utf-8"))["ok"] is True
+    finally:
+        approvals._pending.pop(sid, None)
+        _clear_attention_state(sid)
+
+
+def test_attention_summary_hides_retired_direct_runs_mirror():
+    import api.route_approvals as approvals
+
+    sid = "attention-retired-gateway-session"
+    _clear_attention_state(sid)
+    try:
+        approvals.submit_gateway_pending_mirror(sid, {
+            "_gateway_mirror": True,
+            "approval_id": "approval-retired",
+            "run_id": "run-retired",
+            "command": "rm -rf /tmp/retired",
+            "description": "Retired approval",
+        })
+
+        assert approvals.retire_gateway_pending_mirror(sid, run_id="run-retired") is True
+        assert routes._session_attention_summary(sid) is None
+        assert approvals.gateway_pending_mirror(sid, run_id="run-retired") is None
+    finally:
+        approvals._pending.pop(sid, None)
+        _clear_attention_state(sid)
+
+
 def test_attention_summary_reports_clarify_when_no_approval_is_pending():
     sid = "attention-clarify-session"
     _clear_attention_state(sid)