From b907630f8f8cf672b49cff032591f0a4f188a238 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Wed, 15 Jul 2026 18:48:40 -0400 Subject: [PATCH] fix(#6008): keep gateway approvals attached to the right run --- api/gateway_chat.py | 246 ++-- api/route_approvals.py | 139 ++- api/routes.py | 127 +- static/boot.js | 15 +- static/commands.js | 11 +- static/messages.js | 4 +- static/sessions.js | 4 +- tests/test_cancel_stream_owner_guard.py | 47 +- tests/test_gateway_approval_runs_api.py | 1052 ++++++++++++++++- ...est_issue4771_local_approval_regression.py | 96 ++ .../test_issue6008_sidebar_stop_failure_ui.py | 144 +++ 11 files changed, 1698 insertions(+), 187 deletions(-) create mode 100644 tests/test_issue6008_sidebar_stop_failure_ui.py diff --git a/api/gateway_chat.py b/api/gateway_chat.py index b19eb2e1..db157b68 100644 --- a/api/gateway_chat.py +++ b/api/gateway_chat.py @@ -8,7 +8,9 @@ import threading import time import uuid import urllib.error +import urllib.parse import urllib.request +import uuid from typing import Any from api.config import ( @@ -39,6 +41,13 @@ logger = logging.getLogger(__name__) # Maps stream_id -> gateway run_id for approval response relay. _STREAM_RUN_IDS: dict[str, str] = {} +_STREAM_RUN_STARTING: set[str] = set() +_STREAM_RUN_STARTING_LOCK = threading.Lock() + + +def gateway_run_id_pending(stream_id: str) -> bool: + with _STREAM_RUN_STARTING_LOCK: + return stream_id in _STREAM_RUN_STARTING _WEBUI_CHAT_BACKEND_ENV = "HERMES_WEBUI_CHAT_BACKEND" _WEBUI_GATEWAY_BASE_URL_ENV = "HERMES_WEBUI_GATEWAY_BASE_URL" @@ -338,8 +347,6 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None: args = payload.get("args") if isinstance(payload.get("args"), (list, dict)) else [] run_id = str(payload.get("run_id") or "").strip() approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip() - if not approval_id and run_id: - approval_id = f"gwrun:{run_id}" risk = str(payload.get("risk_level") or "high").strip() choices = payload.get("choices") if isinstance(payload.get("choices"), list) else [] allow_permanent = payload.get("allow_permanent") @@ -369,88 +376,107 @@ 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 + with _STREAM_RUN_STARTING_LOCK: + _STREAM_RUN_STARTING.add(stream_id) + try: + url_runs = f"{base_url.rstrip('/')}/v1/runs" + headers = { + "Content-Type": "application/json", + "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: + with _STREAM_RUN_STARTING_LOCK: + _STREAM_RUN_STARTING.discard(stream_id) + raise + usage: dict = {} _STREAM_RUN_IDS[stream_id] = run_id + with _STREAM_RUN_STARTING_LOCK: + _STREAM_RUN_STARTING.discard(stream_id) + if cancel_event.is_set(): + stop_confirmed = False + try: + stop_confirmed = stop_gateway_run(stream_id) + except Exception: + logger.debug("Failed to stop gateway run after pre-stream cancellation", exc_info=True) + if stop_confirmed: + put_gateway_event("cancel", {"message": "Cancelled by user"}) + return None, usage + cancel_event.clear() url_events = f"{base_url.rstrip('/')}/v1/runs/{run_id}/events" headers_sse = dict(headers) headers_sse["Accept"] = "text/event-stream" req_events = urllib.request.Request(url_events, headers=headers_sse, method="GET") final_text = "" - usage: dict = {} sse_event = "message" with urllib.request.urlopen(req_events, timeout=_gateway_read_timeout_secs()) as resp: for raw_line in _iter_sse_lines_cancellable(resp, cancel_event): @@ -479,7 +505,7 @@ def _run_gateway_runs_api_streaming( if approval_data: approval_data["run_id"] = run_id if not approval_data.get("approval_id"): - approval_data["approval_id"] = f"gwrun:{run_id}" + approval_data["approval_id"] = f"gwrun:{run_id}:{uuid.uuid4().hex}" from api.route_approvals import submit_gateway_pending_mirror submit_gateway_pending_mirror(session_id, approval_data) put_gateway_event("approval", approval_data) @@ -562,6 +588,40 @@ def _run_gateway_runs_api_streaming( return final_text, usage +def stop_gateway_run(stream_id: str) -> bool: + """Request gateway interruption and report whether it was acknowledged.""" + run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip() + if not run_id: + return False + from api.config import get_config + + cfg = get_config() + base_url = _gateway_base_url(cfg) + api_key = _gateway_api_key() + headers = {"Accept": "application/json", "Content-Type": "application/json"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + req = urllib.request.Request( + f"{base_url.rstrip('/')}/v1/runs/{urllib.parse.quote(run_id, safe='')}/stop", + data=b"{}", + headers=headers, + method="POST", + ) + class _NoRedirect(urllib.request.HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + try: + opener = urllib.request.build_opener(_NoRedirect) + with opener.open(req, timeout=10) as response: + final_url = str(getattr(response, "geturl", lambda: req.full_url)() or "") + status = int(getattr(response, "status", getattr(response, "code", 0)) or 0) + return 200 <= status < 300 and final_url == req.full_url + except (urllib.error.HTTPError, urllib.error.URLError, OSError, ValueError): + logger.debug("Gateway stop failed for run %s", run_id, exc_info=True) + return False + + def _settle_gateway_terminal_error(session_id, stream_id, workspace, model, model_provider, terminal_error): from api.streaming import ( _classify_provider_error, @@ -701,6 +761,7 @@ def _run_gateway_chat_streaming( STREAM_LIVE_TOOL_CALLS[stream_id] = [] success_writeback_committed = False + runs_api_pending_marked = False def put_gateway_event(event, data): if cancel_event.is_set() and not success_writeback_committed and event not in ("cancel", "error", "apperror"): @@ -742,6 +803,27 @@ def _run_gateway_chat_streaming( model=model, model_provider=model_provider, ) + base_url = _gateway_base_url(cfg) + api_key = _gateway_api_key() + try: + from api.config import _main_model_request_overrides + _gw_overrides = _main_model_request_overrides( + cfg, + effective_model=model, + effective_provider=model_provider, + ) + except Exception: + _gw_overrides = {} + _runs_api_enabled = _gateway_use_runs_api_enabled(cfg) + if _runs_api_enabled: + with _STREAM_RUN_STARTING_LOCK: + _STREAM_RUN_STARTING.add(stream_id) + runs_api_pending_marked = True + _use_runs_api = _runs_api_enabled and gateway_supports_approval(base_url, api_key) + if not _use_runs_api and runs_api_pending_marked: + with _STREAM_RUN_STARTING_LOCK: + _STREAM_RUN_STARTING.discard(stream_id) + runs_api_pending_marked = False try: from api.streaming import ( _load_webui_prefill_context, @@ -780,19 +862,6 @@ def _run_gateway_chat_streaming( except Exception: logger.debug("Failed to load WebUI gateway prefill context", exc_info=True) prefill_messages = [] - base_url = _gateway_base_url(cfg) - api_key = _gateway_api_key() - try: - from api.config import _main_model_request_overrides - _gw_overrides = _main_model_request_overrides( - cfg, - effective_model=model, - effective_provider=model_provider, - ) - except Exception: - _gw_overrides = {} - # Capability gate: use runs API when gateway advertises approval support. - _use_runs_api = _gateway_use_runs_api_enabled(cfg) and gateway_supports_approval(base_url, api_key) if _use_runs_api: body_extras = {} if model_provider: @@ -1207,5 +1276,8 @@ def _run_gateway_chat_streaming( STREAM_LAST_EVENT_ID.pop(stream_id, None) STREAMS.pop(stream_id, None) _STREAM_RUN_IDS.pop(stream_id, None) + if runs_api_pending_marked and gateway_run_id_pending(stream_id): + _finish_gateway_run_starting(stream_id) + _clear_gateway_run_starting(stream_id) unregister_stream_owner(stream_id) unregister_active_run(stream_id) diff --git a/api/route_approvals.py b/api/route_approvals.py index 97bdba37..6ab739d8 100644 --- a/api/route_approvals.py +++ b/api/route_approvals.py @@ -149,9 +149,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No live_head_data = getattr(live_head_entry, "data", None) or {} live_token = _gateway_mirror_entry_token(live_head_entry) if live_head_entry and live_head_data else None live_run_id = str(live_head_data.get("run_id") or "").strip() + if live_token and live_run_id and not str(live_head_data.get("approval_id") or "").strip(): + live_head_data["approval_id"] = f"gwrun:{live_run_id}:{live_token}" live_approval_id = str(live_head_data.get("approval_id") or "").strip() rebuilt: list[dict] = [] + deferred_run_entries: list[dict] = [] live_mirror_present = False for entry in queue_list: if not _is_gateway_mirror_entry(entry): @@ -164,9 +167,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No if live_token: if entry_token and entry_token == live_token: matches_live_head = True - elif live_run_id and entry_run_id == live_run_id: - matches_live_head = True - elif live_approval_id and entry_approval_id == live_approval_id: + elif ( + live_approval_id + and live_run_id + and entry_approval_id == live_approval_id + and entry_run_id == live_run_id + ): matches_live_head = True if entry_run_id: @@ -177,6 +183,12 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No rebuilt.append(entry) live_mirror_present = True continue + if live_token: + if entry_token: + changed = True + continue + deferred_run_entries.append(entry) + continue if not entry_token: rebuilt.append(entry) continue @@ -196,13 +208,19 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No if live_token and not live_mirror_present: mirror_entry = dict(live_head_data) mirror_run_id = str(mirror_entry.get("run_id") or "").strip() - mirror_entry.setdefault("approval_id", f"gwrun:{mirror_run_id}" if mirror_run_id else uuid.uuid4().hex) + mirror_entry.setdefault( + "approval_id", + f"gwrun:{mirror_run_id}:{live_token}" if mirror_run_id else uuid.uuid4().hex, + ) mirror_entry[_GATEWAY_MIRROR_FLAG] = True mirror_entry[_GATEWAY_MIRROR_TOKEN] = live_token rebuilt.append(mirror_entry) live_mirror_present = True changed = True + if deferred_run_entries: + rebuilt.extend(deferred_run_entries) + if rebuilt: if rebuilt != queue_list: _pending[session_key] = rebuilt @@ -224,14 +242,25 @@ def _gateway_pending_mirror_locked(session_key: str, approval_id: str = "", run_ queue = _pending.get(session_key) entries = queue if isinstance(queue, list) else [queue] if queue else [] if approval_id: + matched_entry: dict | None = None for entry in entries: - if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip(): + if not _is_gateway_mirror_entry(entry): continue - if entry.get("approval_id") == approval_id: - if run_id and entry.get("run_id") != run_id: + if entry.get("approval_id") != approval_id: + continue + entry_run_id = str(entry.get("run_id") or "").strip() + if not entry_run_id: + if not run_id: return None + continue + if run_id and entry_run_id != run_id: + continue + if run_id: return entry - return None + if matched_entry is not None: + return None + matched_entry = entry + return matched_entry for entry in entries: if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip(): continue @@ -249,21 +278,47 @@ def gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str = "") -> bool: - """Retire only the exact run-backed mirror and notify its new queue head.""" + """Retire one approval, or every mirror for a terminal run.""" with _lock: reconcile_gateway_pending_mirror_locked(session_key) queue = _pending.get(session_key) entries = queue if isinstance(queue, list) else [queue] if queue else [] - match = _gateway_pending_mirror_locked(session_key, approval_id, run_id) - if not match: + 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) + retired = [match] if match else [] + else: + retired = [ + entry for entry in entries + if _is_gateway_mirror_entry(entry) + and str(entry.get("run_id") or "").strip() == normalized_run_id + ] if normalized_run_id else [] + if normalized_run_id: + retained_gateway_queue = [] + for entry in gateway_queue: + data = getattr(entry, "data", None) or {} + if str(data.get("run_id") or "").strip() == normalized_run_id: + gateway_queue_changed = True + continue + retained_gateway_queue.append(entry) + if not retired and not gateway_queue_changed: return False - entries.remove(match) + for match in retired: + entries.remove(match) + if normalized_run_id and not approval_id: + if retained_gateway_queue: + _gateway_queues[session_key] = retained_gateway_queue + else: + _gateway_queues.pop(session_key, None) if entries: _pending[session_key] = entries else: _pending.pop(session_key, None) - head = entries[0] if entries else None - _approval_sse_notify_locked(session_key, head, len(entries)) + head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) + _approval_sse_notify_locked(session_key, head, total) publish_session_list_changed("attention_resolved") return True @@ -280,33 +335,57 @@ def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str return None -def _gateway_mirrored_pending_approval_id_by_run_id(session_key: str, run_id: str) -> str | None: - """Return the mirrored approval_id for a matching gateway run.""" - run_id = str(run_id or "").strip() - if not run_id: - return None - with _lock: - entry = _gateway_pending_mirror_locked(session_key, run_id=run_id) - if entry: - approval_id = str(entry.get("approval_id") or "").strip() - return approval_id or None - return None - - def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: """Mirror the live gateway head into WebUI polling state under a typed tag.""" with _lock: run_id = str(approval.get("run_id") or "").strip() if run_id: live_gateway_queue = _gateway_queues.get(session_key) or [] - if not live_gateway_queue: + approval_id = str(approval.get("approval_id") or "").strip() + live_head_entry = live_gateway_queue[0] if live_gateway_queue else None + live_head_data = getattr(live_head_entry, "data", None) or {} + live_head_run_id = str(live_head_data.get("run_id") or "").strip() + live_head_approval_id = str(live_head_data.get("approval_id") or "").strip() + live_token = ( + _gateway_mirror_entry_token(live_head_entry) + if live_head_entry and live_head_data + else None + ) + if ( + live_token + and live_head_run_id == run_id + and ( + not approval_id + or not live_head_approval_id + or live_head_approval_id == approval_id + ) + ): + if approval_id: + live_head_data["approval_id"] = approval_id + else: + approval_id = live_head_approval_id + if not approval_id: + approval_id = f"gwrun:{run_id}:{live_token}" + live_head_data["approval_id"] = approval_id + approval["approval_id"] = approval_id + else: + if not approval_id: + approval_id = f"gwrun:{run_id}:{uuid.uuid4().hex}" + approval["approval_id"] = approval_id mirror_entry = dict(approval) mirror_entry["run_id"] = run_id - mirror_entry["approval_id"] = str(mirror_entry.get("approval_id") or f"gwrun:{run_id}").strip() + mirror_entry["approval_id"] = approval_id mirror_entry[_GATEWAY_MIRROR_FLAG] = True - if not _gateway_pending_mirror_locked(session_key, approval_id=mirror_entry["approval_id"], run_id=run_id): + if not _gateway_pending_mirror_locked(session_key, approval_id=approval_id, run_id=run_id): _normalize_pending_queue_locked(session_key).append(mirror_entry) head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) + approval_id = str(approval.get("approval_id") or "").strip() + if run_id and approval_id: + mirror_entry = _gateway_pending_mirror_locked( + session_key, approval_id=approval_id, run_id=run_id + ) + if mirror_entry: + approval["approval_id"] = str(mirror_entry.get("approval_id") or approval_id).strip() _approval_sse_notify_locked(session_key, head, total) publish_session_list_changed("attention_pending") diff --git a/api/routes.py b/api/routes.py index 5d72e0f8..9151abf4 100644 --- a/api/routes.py +++ b/api/routes.py @@ -13368,6 +13368,46 @@ def handle_get(handler, parsed) -> bool: return bad(handler, "stream_id required") if not _stream_id_visible_to_request_profile(handler, stream_id): return True + gateway_stop_blocked = False + try: + from api.gateway_chat import _STREAM_RUN_IDS, gateway_run_id_pending, stop_gateway_run + + run_id = _STREAM_RUN_IDS.get(stream_id) + pending_run_id = False + if not run_id: + import time as _time + + pending_run_id = gateway_run_id_pending(stream_id) + attempts_remaining = 10 + while not run_id and pending_run_id and attempts_remaining > 0: + _time.sleep(0.05) + run_id = _STREAM_RUN_IDS.get(stream_id) + pending_run_id = gateway_run_id_pending(stream_id) + attempts_remaining -= 1 + if run_id: + if stop_gateway_run(stream_id): + owner_sid = stream_owner_session_id(stream_id) + if owner_sid: + retire_gateway_pending_mirror(owner_sid, run_id=run_id) + else: + gateway_stop_blocked = True + elif pending_run_id: + gateway_stop_blocked = True + except Exception: + logger.debug("Failed to stop gateway run during chat cancellation", exc_info=True) + gateway_stop_blocked = True + if gateway_stop_blocked: + return j( + handler, + { + "ok": False, + "cancelled": False, + "stream_id": stream_id, + "error": "Gateway stop failed", + }, + status=502, + ) + from api.runtime_adapter import LegacyJournalRuntimeAdapter, runtime_adapter_enabled if runtime_adapter_enabled(): @@ -23749,7 +23789,7 @@ def _handle_workspace_reorder(handler, body): return j(handler, {"ok": True, "workspaces": reordered}) -def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: +def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: str = "") -> bool: """Resolve an approval through the existing callback path. Slice 3b keeps the RuntimeAdapter as a protocol translator: it delegates to @@ -23760,6 +23800,7 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: pending = None found_target = False gateway_keys = [] + gateway_head_matches_target = False with _lock: reconcile_gateway_pending_mirror_locked(sid) queue = _pending.get(sid) @@ -23767,10 +23808,13 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: if approval_id: # Find and remove the specific entry by approval_id. for i, entry in enumerate(queue): - if entry.get("approval_id") == approval_id: - pending = queue.pop(i) - found_target = True - break + if entry.get("approval_id") != approval_id: + continue + if run_id and str(entry.get("run_id") or "").strip() != run_id: + continue + pending = queue.pop(i) + found_target = True + break else: # A stale explicit id must not accidentally approve the # oldest queued command; duplicate/stale responses are @@ -23783,7 +23827,13 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: _pending.pop(sid, None) elif queue: # Legacy single-dict value. - if not approval_id or queue.get("approval_id") == approval_id: + if ( + not approval_id + or ( + queue.get("approval_id") == approval_id + and (not run_id or str(queue.get("run_id") or "").strip() == run_id) + ) + ): pending = _pending.pop(sid, None) found_target = pending is not None # When no _pending entry found AND no explicit approval_id was @@ -23806,6 +23856,17 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: # idempotent over the session key set so the outcome is # the same regardless of which entry wins the race. found_target = True + elif approval_id: + gw_queue = _gateway_queues.get(sid) + if gw_queue and len(gw_queue) > 0: + gw_entry = gw_queue[0] + gw_data = getattr(gw_entry, "data", None) or {} + gw_approval_id = str(gw_data.get("approval_id") or "").strip() + gw_run_id = str(gw_data.get("run_id") or "").strip() + if run_id: + gateway_head_matches_target = gw_approval_id == approval_id and gw_run_id == run_id + else: + gateway_head_matches_target = gw_approval_id == approval_id # Notify SSE subscribers of the new head (or empty state) so the UI # surfaces any trailing approvals that were queued behind this one # without waiting for the next submit_pending. Without this, a parallel @@ -23837,7 +23898,17 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str) -> bool: # This is the primary signal when streaming is active — the agent # thread is parked in entry.event.wait() and needs to be woken up. gateway_resolved = 0 - if found_target or not approval_id: + should_resolve_gateway = ( + not approval_id + or ( + found_target + and ( + not run_id + or gateway_head_matches_target + ) + ) + ) + if should_resolve_gateway: gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0 # Keep the historical no-id response path truthy for old clients/tests while # making stale explicit ids bounded as not-active for Slice 3b. @@ -23920,20 +23991,50 @@ def _handle_approval_respond(handler, body): active_sid = getattr(s, "active_stream_id", None) if active_sid: _candidate_run_id = _STREAM_RUN_IDS.get(active_sid) - matched_mirror = gateway_pending_mirror(sid, approval_id=approval_id, run_id=_candidate_run_id) - _run_id = matched_mirror["run_id"] if matched_mirror else None - if not matched_mirror and approval_id: + local_match = False + run_backed_gateway_matches = 0 + if approval_id: with _lock: queue = _pending.get(sid) entries = queue if isinstance(queue, list) else [queue] if queue else [] local_match = any( isinstance(entry, dict) and entry.get("approval_id") == approval_id - and not entry.get(_GATEWAY_MIRROR_FLAG) + and ( + not entry.get(_GATEWAY_MIRROR_FLAG) + or not str(entry.get("run_id") or "").strip() + ) for entry in entries ) + run_backed_gateway_matches = sum( + 1 + for entry in entries + if isinstance(entry, dict) + and entry.get("approval_id") == approval_id + and entry.get(_GATEWAY_MIRROR_FLAG) + and str(entry.get("run_id") or "").strip() + ) if local_match: _candidate_run_id = None + matched_mirror = gateway_pending_mirror( + sid, approval_id=approval_id, run_id=_candidate_run_id + ) if approval_id else None + _run_id = matched_mirror["run_id"] if matched_mirror else None + if not matched_mirror and approval_id: + if local_match: + _candidate_run_id = None + elif run_backed_gateway_matches > 1 and not _candidate_run_id: + return j( + handler, + { + "ok": False, + "choice": choice, + "relayed": False, + "code": "gateway_run_unavailable", + "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE, + }, + status=409, + ) if _run_id: from api.runner_client import HttpRunnerClient, RunnerClientError _cfg = _get_config() @@ -23949,10 +24050,10 @@ def _handle_approval_respond(handler, body): # still needs the same cleanup path so the parked entry, mirrored # card, and agent signal all settle here too. cleanup_approval_id = matched_mirror["approval_id"] - _resolve_approval_legacy(sid, cleanup_approval_id, choice) + _resolve_approval_legacy(sid, cleanup_approval_id, choice, run_id=_run_id) retire_gateway_pending_mirror(sid, approval_id=cleanup_approval_id, run_id=_run_id) return j(handler, {"ok": True, "choice": choice, "relayed": True}) - if _candidate_run_id or approval_id.startswith("gwrun:"): + if _candidate_run_id: return j(handler, {"ok": False, "choice": choice, "relayed": False, "code": "gateway_run_unavailable", "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE}, status=409) diff --git a/static/boot.js b/static/boot.js index 1e79ac4e..a45a33a4 100644 --- a/static/boot.js +++ b/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){ diff --git a/static/commands.js b/static/commands.js index dc6a0d43..a2377655 100644 --- a/static/commands.js +++ b/static/commands.js @@ -1213,7 +1213,10 @@ async function cmdPersonality(args){ async function cmdStop(){ if(!S.session){showToast(t('no_active_session'));return;} if(!S.activeStreamId){showToast(t('no_active_task'));return;} - if(typeof cancelStream==='function'){await cancelStream('slash-stop');showToast(t('stream_stopped'));} + if(typeof cancelStream==='function'){ + if(await cancelStream('slash-stop')) showToast(t('stream_stopped')); + else showToast(t('cancel_failed')); + } else showToast(t('cancel_unavailable')); } @@ -1319,8 +1322,10 @@ async function cmdInterrupt(args){ updateQueueBadge(S.session.session_id); S.pendingFiles=[];renderTray(); // Cancel the active stream; setBusy(false) will drain the queue - if(typeof cancelStream==='function'){await cancelStream('slash-interrupt');} - showToast(t('cmd_interrupt_confirm'),2000); + if(typeof cancelStream==='function'){ + if(await cancelStream('slash-interrupt')) showToast(t('cmd_interrupt_confirm'),2000); + else showToast(t('cancel_failed')); + } } /** diff --git a/static/messages.js b/static/messages.js index b67a95e8..ea19f451 100644 --- a/static/messages.js +++ b/static/messages.js @@ -1395,8 +1395,8 @@ async function send(){ _clearComposerAfterQueuedSelectionSend(S.session&&S.session.session_id); S.pendingFiles=[];renderTray(); if(S.activeStreamId&&typeof cancelStream==='function'){ - showToast(t('busy_interrupt_confirm'),2000); - await cancelStream('busy-interrupt'); + if(await cancelStream('busy-interrupt')) showToast(t('busy_interrupt_confirm'),2000); + else showToast(t('cancel_failed')); } else { showToast(`Queued: "${text.slice(0,40)}${text.length>40?'…':''}"`,2000); } diff --git a/static/sessions.js b/static/sessions.js index ed41c6d8..1100794a 100644 --- a/static/sessions.js +++ b/static/sessions.js @@ -4905,8 +4905,8 @@ function _openSessionActionMenu(session, anchorEl){ ICONS.stop, async()=>{ closeSessionActionMenu(); - await cancelSessionStream(session); - showToast(t('stream_stopped')); + if(await cancelSessionStream(session)) showToast(t('stream_stopped')); + else showToast(t('cancel_failed')); } )); } diff --git a/tests/test_cancel_stream_owner_guard.py b/tests/test_cancel_stream_owner_guard.py index e149dd42..bc5e3333 100644 --- a/tests/test_cancel_stream_owner_guard.py +++ b/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'), " diff --git a/tests/test_gateway_approval_runs_api.py b/tests/test_gateway_approval_runs_api.py index 3b6cb1eb..19cd3094 100644 --- a/tests/test_gateway_approval_runs_api.py +++ b/tests/test_gateway_approval_runs_api.py @@ -303,7 +303,7 @@ def test_gateway_approval_event_translation(): assert result["run_id"] == "run-999" assert result["approval_id"] == "appr-1" empty_id = _gateway_runs_approval_event({**payload, "approval_id": "", "id": ""}) - assert empty_id["approval_id"] == "gwrun:run-999" + assert empty_id["approval_id"] == "" downgraded = _gateway_runs_approval_event({ "command": "rm -rf /tmp/x", @@ -467,7 +467,7 @@ def test_empty_id_runs_approval_reaches_real_response_lifecycle(): cancel_event=threading.Event(), ) approval_id = events[0][1]["approval_id"] - assert approval_id == f"gwrun:run-real-{choice}" + assert approval_id.startswith(f"gwrun:run-real-{choice}:") assert approvals.gateway_pending_mirror(sid, approval_id=approval_id)["run_id"] == f"run-real-{choice}" routes._handle_approval_respond(handler, { "session_id": sid, "choice": choice, "approval_id": approval_id, @@ -488,10 +488,10 @@ def test_gateway_mirror_match_is_session_scoped_and_retires_one_run(): try: approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-a", "command": "a"}) approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-b", "command": "b"}) - first = approvals.gateway_pending_mirror(sid, approval_id="gwrun:run-a") + first = approvals.gateway_pending_mirror(sid, run_id="run-a") second = approvals.gateway_pending_mirror(sid, run_id="run-b") - assert first["approval_id"] == "gwrun:run-a" - assert second["approval_id"] == "gwrun:run-b" + assert first["approval_id"].startswith("gwrun:run-a:") + assert second["approval_id"].startswith("gwrun:run-b:") assert approvals.gateway_pending_mirror(other_sid, run_id="run-a") is None assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") is True assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None @@ -501,6 +501,904 @@ def test_gateway_mirror_match_is_session_scoped_and_retires_one_run(): approvals._pending.pop(other_sid, None) +def test_gateway_mirror_exact_pair_lookup_skips_same_id_other_run(): + import api.route_approvals as approvals + + sid = "sess-same-id-other-run" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-old", "approval_id": "same-id", "command": "old" + }) + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-live", "approval_id": "same-id", "command": "live" + }) + assert approvals.gateway_pending_mirror(sid, approval_id="same-id") is None + live = approvals.gateway_pending_mirror(sid, approval_id="same-id", run_id="run-live") + assert live is not None + assert live["run_id"] == "run-live" + assert approvals.retire_gateway_pending_mirror(sid, approval_id="same-id", run_id="run-live") + assert approvals.gateway_pending_mirror(sid, approval_id="same-id", run_id="run-live") is None + assert approvals.gateway_pending_mirror(sid, approval_id="same-id", run_id="run-old") is not None + finally: + approvals._pending.pop(sid, None) + + +def test_gateway_approval_response_relay_cleans_only_selected_shared_id_run(): + from api.gateway_chat import _STREAM_RUN_IDS + from api import routes + import api.route_approvals as approvals + + sid = "sess-shared-id-relay" + stream_id = "sid-shared-id-relay" + _STREAM_RUN_IDS[stream_id] = "run-live" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-old", "approval_id": "same-id", "command": "old" + }) + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-live", "approval_id": "same-id", "command": "live" + }) + + mock_session = MagicMock() + mock_session.active_stream_id = stream_id + handler = MagicMock() + handler.wfile = io.BytesIO() + + with patch("api.routes.get_session", return_value=mock_session), \ + patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval, \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=""): + routes._handle_approval_respond(handler, { + "session_id": sid, "choice": "once", "approval_id": "same-id", + }) + + handler.send_response.assert_called_with(200) + respond_approval.assert_called_once_with("run-live", "same-id", "once") + assert approvals.gateway_pending_mirror(sid, approval_id="same-id", run_id="run-live") is None + assert approvals.gateway_pending_mirror(sid, approval_id="same-id", run_id="run-old") is not None + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + + +def test_gateway_approval_response_without_active_run_rejects_shared_upstream_id(): + from api import routes + import api.route_approvals as approvals + + sid = "sess-shared-id-no-active-run" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-a", "approval_id": "shared-upstream-id", "command": "a" + }) + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-b", "approval_id": "shared-upstream-id", "command": "b" + }) + + mock_session = MagicMock() + mock_session.active_stream_id = None + handler = MagicMock() + handler.wfile = io.BytesIO() + + with patch("api.routes.get_session", return_value=mock_session), \ + patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=True), \ + patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval: + routes._handle_approval_respond(handler, { + "session_id": sid, "choice": "once", "approval_id": "shared-upstream-id", + }) + + payload = json.loads(handler.wfile.getvalue().decode("utf-8")) + assert handler.send_response.call_args.args[0] == 409 + assert payload["code"] == "gateway_run_unavailable" + respond_approval.assert_not_called() + assert approvals.gateway_pending_mirror(sid, approval_id="shared-upstream-id", run_id="run-a") is not None + assert approvals.gateway_pending_mirror(sid, approval_id="shared-upstream-id", run_id="run-b") is not None + finally: + approvals._pending.pop(sid, None) + + +def test_gateway_approval_relay_does_not_unblock_unrelated_local_head(): + from api.gateway_chat import _STREAM_RUN_IDS + from api import routes + import api.route_approvals as approvals + import tools.approval as ta + + sid = "sess-remote-relay-local-head" + stream_id = "sid-remote-relay-local-head" + _STREAM_RUN_IDS[stream_id] = "run-remote" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, { + "run_id": "run-remote", "approval_id": "remote-a", "command": "remote" + }) + local_entry = ta._ApprovalEntry({ + "command": "local", + "description": "local approval", + "approval_id": "local-b", + }) + with approvals._lock: + approvals._gateway_queues[sid] = [local_entry] + + mock_session = MagicMock() + mock_session.active_stream_id = stream_id + handler = MagicMock() + handler.wfile = io.BytesIO() + + with patch("api.routes.get_session", return_value=mock_session), \ + patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval, \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=""): + routes._handle_approval_respond(handler, { + "session_id": sid, "choice": "once", "approval_id": "remote-a", + }) + + handler.send_response.assert_called_with(200) + respond_approval.assert_called_once_with("run-remote", "remote-a", "once") + assert local_entry.event.is_set() is False + assert local_entry.result is None + assert approvals.gateway_pending_mirror(sid, approval_id="remote-a", run_id="run-remote") is None + with approvals._lock: + assert sid in approvals._gateway_queues + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_empty_id_approvals_same_run_keep_distinct_fifo_cards(): + import api.route_approvals as approvals + + sid = "sess-same-run" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-a", "command": "first"}) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-a", "command": "second"}) + queue = approvals._pending[sid] + assert [entry["command"] for entry in queue] == ["first", "second"] + assert queue[0]["approval_id"] != queue[1]["approval_id"] + assert approvals.retire_gateway_pending_mirror(sid, approval_id=queue[0]["approval_id"], run_id="run-a") + assert approvals._pending[sid][0]["command"] == "second" + finally: + approvals._pending.pop(sid, None) + + +def test_live_gateway_head_gets_token_scoped_approval_id(): + import api.route_approvals as approvals + + sid = "sess-live-head" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [SimpleNamespace(data={"run_id": "run-live", "command": "head"})] + with approvals._lock: + head, total, changed = approvals.reconcile_gateway_pending_mirror_locked(sid) + assert changed is True + assert total == 1 + assert head["approval_id"].startswith("gwrun:run-live:") + assert approvals._gateway_queues[sid][0].data["approval_id"] == head["approval_id"] + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_live_gateway_head_emits_the_stored_token_scoped_identity(): + import api.route_approvals as approvals + + sid = "sess-live-head-emitted-id" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [SimpleNamespace(data={"run_id": "run-live", "command": "head"})] + approval_data = {"run_id": "run-live", "command": "head"} + approvals.submit_gateway_pending_mirror(sid, approval_data) + mirror = approvals.gateway_pending_mirror(sid, run_id="run-live") + assert mirror is not None + assert approval_data["approval_id"] == mirror["approval_id"] + assert approvals._gateway_queues[sid][0].data["approval_id"] == mirror["approval_id"] + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_distinct_same_run_approval_does_not_overwrite_live_head_identity(): + import api.route_approvals as approvals + + sid = "sess-same-run-distinct-approval" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [ + SimpleNamespace(data={"run_id": "run-live", "approval_id": "appr-a", "command": "head"}) + ] + approvals.submit_gateway_pending_mirror( + sid, + {"run_id": "run-live", "approval_id": "appr-b", "command": "second"}, + ) + assert approvals._gateway_queues[sid][0].data["approval_id"] == "appr-a" + assert approvals.gateway_pending_mirror(sid, approval_id="appr-a", run_id="run-live") is not None + assert approvals.gateway_pending_mirror(sid, approval_id="appr-b", run_id="run-live") is not None + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_direct_remote_approval_still_mirrors_with_local_gateway_head(): + import api.route_approvals as approvals + + sid = "sess-local-head-remote-mirror" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [SimpleNamespace(data={"command": "local-head"})] + approvals.submit_gateway_pending_mirror( + sid, + {"run_id": "run-remote", "approval_id": "remote-a", "command": "remote"}, + ) + mirror = approvals.gateway_pending_mirror(sid, approval_id="remote-a", run_id="run-remote") + assert mirror is not None + assert mirror["run_id"] == "run-remote" + assert mirror["approval_id"] == "remote-a" + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_reconcile_does_not_bind_live_token_by_approval_id_alone(): + import api.route_approvals as approvals + + sid = "sess-cross-run-shared-id" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [{ + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-stale", + "approval_id": "appr-shared", + "command": "stale", + }] + approvals._gateway_queues[sid] = [SimpleNamespace(data={ + "run_id": "run-live", + "approval_id": "appr-shared", + "command": "live", + })] + with approvals._lock: + head, total, _changed = approvals.reconcile_gateway_pending_mirror_locked(sid) + queue = list(approvals._pending[sid]) + assert total == 2 + assert head["run_id"] == "run-live" + assert queue[0]["run_id"] == "run-live" + assert queue[1]["run_id"] == "run-stale" + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_terminal_run_retirement_removes_all_same_run_mirrors(): + import api.route_approvals as approvals + + sid = "sess-terminal-run" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-a", "command": "first"}) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-a", "command": "second"}) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-b", "command": "other"}) + assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") + assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None + assert approvals.gateway_pending_mirror(sid, run_id="run-b") is not None + finally: + approvals._pending.pop(sid, None) + + +def test_terminal_run_retirement_clears_same_run_gateway_queue_state(): + import api.route_approvals as approvals + + sid = "sess-terminal-run-queue" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [ + SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}), + SimpleNamespace(data={"command": "local-head"}), + ] + approvals.submit_gateway_pending_mirror( + sid, + {"run_id": "run-a", "approval_id": "appr-a", "command": "a"}, + ) + assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") + assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None + assert not any( + str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a" + for entry in approvals._gateway_queues.get(sid, []) + ) + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_terminal_run_retirement_clears_non_head_same_run_gateway_queue_state(): + import api.route_approvals as approvals + + sid = "sess-terminal-run-non-head-queue" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [ + SimpleNamespace(data={"command": "local-head"}), + SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-a", "command": "a"}), + ] + assert approvals.retire_gateway_pending_mirror(sid, run_id="run-a") + assert approvals.gateway_pending_mirror(sid, run_id="run-a") is None + assert not any( + str((getattr(entry, "data", None) or {}).get("run_id") or "").strip() == "run-a" + for entry in approvals._gateway_queues.get(sid, []) + ) + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_exact_one_retirement_keeps_same_run_gateway_queue_state(): + import api.route_approvals as approvals + + sid = "sess-exact-one-queue" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [ + { + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-a", + "approval_id": "appr-a", + "command": "a", + }, + { + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-a", + "approval_id": "appr-b", + "command": "b", + }, + ] + approvals._gateway_queues[sid] = [ + SimpleNamespace(data={"run_id": "run-a", "approval_id": "appr-b", "command": "b"}) + ] + assert approvals.retire_gateway_pending_mirror(sid, approval_id="appr-a", run_id="run-a") + assert approvals.gateway_pending_mirror(sid, approval_id="appr-b", run_id="run-a") is not None + queue = approvals._gateway_queues.get(sid) or [] + assert len(queue) == 1 + assert queue[0].data["approval_id"] == "appr-b" + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_local_stop_uses_runs_api_stop_endpoint(): + from api.gateway_chat import _STREAM_RUN_IDS, stop_gateway_run + + stream_id = "stream-stop" + _STREAM_RUN_IDS[stream_id] = "run-stop" + + class _Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def geturl(self): + return "http://gw:8642/v1/runs/run-stop/stop" + + requests = [] + + class _Opener: + def open(self, req, *, timeout=None): + requests.append((req, timeout)) + return _Response() + + def fake_build_opener(*handlers): + assert handlers, "stop must install a no-redirect handler" + return _Opener() + + try: + with patch("urllib.request.build_opener", side_effect=fake_build_opener), \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value="secret"): + assert stop_gateway_run(stream_id) is True + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + request, timeout = requests[0] + assert request.full_url == "http://gw:8642/v1/runs/run-stop/stop" + assert request.get_method() == "POST" + assert request.get_header("Authorization") == "Bearer secret" + assert timeout == 10 + + +def test_runs_api_cancel_after_run_creation_stops_remote_run_before_events(): + from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming + + stream_id = "sid-run-create-cancel" + cancel_event = threading.Event() + requests = [] + + class _JsonResponse: + def __init__(self, payload): + self._payload = json.dumps(payload).encode("utf-8") + + def read(self, _limit=None): + return self._payload + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def fake_urlopen(req, *, timeout=None): + requests.append(req.full_url) + if req.full_url.endswith("/v1/runs"): + cancel_event.set() + return _JsonResponse({"run_id": "run-pre-events-cancel"}) + raise AssertionError(f"unexpected urlopen after pre-stream cancel: {req.full_url}") + + try: + with patch("urllib.request.urlopen", side_effect=fake_urlopen), \ + patch("api.gateway_chat.stop_gateway_run", return_value=True) as stop_run: + final_text, usage = _run_gateway_runs_api_streaming( + session_id="sess-run-create-cancel", + msg_text="hi", + model="test-model", + workspace="/tmp", + stream_id=stream_id, + base_url="http://gw:8642", + api_key="secret", + prefill_messages=[], + body_extras={}, + put_gateway_event=lambda *_args, **_kwargs: None, + cancel_event=cancel_event, + cfg={}, + ) + assert final_text is None + assert usage == {} + stop_run.assert_called_once_with(stream_id) + assert requests == ["http://gw:8642/v1/runs"] + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + +def test_runs_api_cancel_after_run_creation_keeps_stream_open_when_stop_fails(): + from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming + + stream_id = "sid-run-create-cancel-stop-failed" + cancel_event = threading.Event() + requests = [] + events = [] + + class _JsonResponse: + def __init__(self, payload): + self._payload = json.dumps(payload).encode("utf-8") + + def read(self, _limit=None): + return self._payload + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + class _SseResponse: + def __init__(self, lines): + self._lines = lines + + def __iter__(self): + return iter(self._lines) + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def fake_urlopen(req, *, timeout=None): + requests.append(req.full_url) + if req.full_url.endswith("/v1/runs"): + cancel_event.set() + return _JsonResponse({"run_id": "run-pre-events-stop-failed"}) + if req.full_url.endswith("/events"): + return _SseResponse([ + b'data: {"event":"message.delta","delta":"Hello"}\n', + b'\n', + b'data: {"event":"run.completed","output":"Hello"}\n', + b'\n', + ]) + raise AssertionError(f"unexpected urlopen: {req.full_url}") + + try: + with patch("urllib.request.urlopen", side_effect=fake_urlopen), \ + patch("api.gateway_chat.stop_gateway_run", return_value=False) as stop_run: + final_text, usage = _run_gateway_runs_api_streaming( + session_id="sess-run-create-cancel-stop-failed", + msg_text="hi", + model="test-model", + workspace="/tmp", + stream_id=stream_id, + base_url="http://gw:8642", + api_key="secret", + prefill_messages=[], + body_extras={}, + put_gateway_event=lambda event, data: events.append((event, data)), + cancel_event=cancel_event, + cfg={}, + ) + assert final_text == "Hello" + assert usage == {} + stop_run.assert_called_once_with(stream_id) + assert requests == [ + "http://gw:8642/v1/runs", + "http://gw:8642/v1/runs/run-pre-events-stop-failed/events", + ] + assert [event for event, _data in events] == ["token"] + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + +def test_runs_api_marks_run_pending_before_request_preparation(): + from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming, gateway_run_id_pending + + stream_id = "sid-run-starting-prep" + cancel_event = threading.Event() + observed = {"pending_during_prepare": False, "pending_during_post": False} + + class _JsonResponse: + def __init__(self, payload): + self._payload = json.dumps(payload).encode("utf-8") + + def read(self, _limit=None): + return self._payload + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + class _SseResponse: + def __iter__(self): + return iter([ + b'data: {"event":"run.completed","output":"Hello"}\n', + b'\n', + ]) + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def fake_strip(text): + observed["pending_during_prepare"] = gateway_run_id_pending(stream_id) + return text + + def fake_urlopen(req, *, timeout=None): + if req.full_url.endswith("/v1/runs"): + observed["pending_during_post"] = gateway_run_id_pending(stream_id) + return _JsonResponse({"run_id": "run-starting"}) + if req.full_url.endswith("/events"): + return _SseResponse() + raise AssertionError(f"unexpected urlopen: {req.full_url}") + + try: + with patch("urllib.request.urlopen", side_effect=fake_urlopen), \ + patch("api.streaming._strip_oob_blocks", side_effect=fake_strip): + final_text, usage = _run_gateway_runs_api_streaming( + session_id="sess-run-starting-prep", + msg_text="hi", + model="test-model", + workspace="/tmp", + stream_id=stream_id, + base_url="http://gw:8642", + api_key="secret", + prefill_messages=[], + body_extras={}, + put_gateway_event=lambda *_args, **_kwargs: None, + cancel_event=cancel_event, + cfg={}, + session=SimpleNamespace(context_messages=[{"role": "user", "content": "ctx"}]), + ) + assert final_text == "Hello" + assert usage == {} + assert observed["pending_during_prepare"] is True + assert observed["pending_during_post"] is True + assert gateway_run_id_pending(stream_id) is False + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + +def test_gateway_worker_marks_run_pending_before_runs_api_prelude(): + from api.gateway_chat import STREAMS, _run_gateway_chat_streaming, gateway_run_id_pending + + stream_id = "sid-worker-run-starting" + observed = {"pending_during_support_check": False, "pending_during_runs_call": False} + STREAMS[stream_id] = SimpleNamespace(put_nowait=lambda *_args, **_kwargs: None) + session = SimpleNamespace( + profile=None, + workspace="/tmp", + context_messages=[], + messages=[], + active_stream_id=stream_id, + pending_user_source="webui", + process_wakeup_pause={}, + save=lambda: None, + ) + + def fake_supports(_base_url, _api_key): + observed["pending_during_support_check"] = gateway_run_id_pending(stream_id) + return True + + def fake_runs(*_args, **_kwargs): + observed["pending_during_runs_call"] = gateway_run_id_pending(stream_id) + return None, {} + + with patch("api.gateway_chat.RunJournalWriter", return_value=SimpleNamespace(append_sse_event=lambda *_a, **_k: None)), \ + patch("api.gateway_chat.get_session", return_value=session), \ + patch("api.config.get_config", return_value={}), \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=""), \ + patch("api.gateway_chat._gateway_use_runs_api_enabled", return_value=True), \ + patch("api.gateway_chat.gateway_supports_approval", side_effect=fake_supports), \ + patch("api.gateway_chat._gateway_reasoning_effort_for_request", return_value=None), \ + patch("api.gateway_chat._run_gateway_runs_api_streaming", side_effect=fake_runs), \ + patch("api.streaming._load_webui_prefill_context", return_value={}), \ + patch("api.streaming._prefill_messages_with_webui_context", return_value=[]), \ + patch("api.streaming._normalize_prefill_messages_before_user_turn", side_effect=lambda messages: messages), \ + patch("api.streaming._public_prefill_context_status", return_value={}), \ + patch("api.streaming._webui_ephemeral_system_prompt", return_value="sys"): + _run_gateway_chat_streaming( + session_id="sess-worker-run-starting", + msg_text="hi", + model="test-model", + workspace="/tmp", + stream_id=stream_id, + ) + + assert observed["pending_during_support_check"] is True + assert observed["pending_during_runs_call"] is True + assert gateway_run_id_pending(stream_id) is False + + +def test_local_stop_rejects_redirected_success(): + from api.gateway_chat import _STREAM_RUN_IDS, stop_gateway_run + + stream_id = "stream-stop-redirect" + _STREAM_RUN_IDS[stream_id] = "run-stop" + + class _Response: + status = 200 + + def __enter__(self): + return self + + def __exit__(self, *args): + return None + + def geturl(self): + return "http://gw:8642/login" + + class _Opener: + def open(self, req, *, timeout=None): + return _Response() + + try: + with patch("urllib.request.build_opener", return_value=_Opener()), \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=None): + assert stop_gateway_run(stream_id) is False + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + +def test_chat_cancel_surfaces_redirected_gateway_stop(monkeypatch): + import urllib.parse + + from api import routes + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + + sid = "sess-cancel-stop-redirect" + stream_id = "stream-cancel-stop-redirect" + _STREAM_RUN_IDS[stream_id] = "run-stop" + approvals._pending.pop(sid, None) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop", "command": "first"}) + captured = {} + called = {"cancel": False} + + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_args: True) + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "stream_owner_session_id", lambda _stream_id: sid) + monkeypatch.setattr(routes, "cancel_stream", lambda _stream_id: called.__setitem__("cancel", True)) + monkeypatch.setattr("api.gateway_chat.stop_gateway_run", lambda _stream_id: False) + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(routes, "j", fake_j) + parsed = urllib.parse.urlparse(f"/api/chat/cancel?stream_id={stream_id}") + try: + routes.handle_get(object(), parsed) + assert captured["status"] == 502 + assert captured["payload"] == { + "ok": False, + "cancelled": False, + "stream_id": stream_id, + "error": "Gateway stop failed", + } + assert called["cancel"] is False + assert sid in approvals._pending + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + + +def test_chat_cancel_retires_same_run_gateway_mirrors_after_stop(monkeypatch): + import urllib.parse + + from api import routes + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + + sid = "sess-cancel-stop" + stream_id = "stream-cancel-stop" + _STREAM_RUN_IDS[stream_id] = "run-stop" + approvals._pending.pop(sid, None) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop", "command": "first"}) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop", "command": "second"}) + captured = {} + + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_args: True) + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "stream_owner_session_id", lambda _stream_id: sid) + monkeypatch.setattr(routes, "cancel_stream", lambda _stream_id: True) + monkeypatch.setattr("api.gateway_chat.stop_gateway_run", lambda _stream_id: True) + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(routes, "j", fake_j) + parsed = urllib.parse.urlparse(f"/api/chat/cancel?stream_id={stream_id}") + try: + routes.handle_get(object(), parsed) + assert captured["status"] == 200 + assert captured["payload"] == {"ok": True, "cancelled": True, "stream_id": stream_id} + assert sid not in approvals._pending + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + + +def test_chat_cancel_surfaces_gateway_stop_failure(monkeypatch): + import urllib.parse + + from api import routes + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + + sid = "sess-cancel-stop-fail" + stream_id = "stream-cancel-stop-fail" + _STREAM_RUN_IDS[stream_id] = "run-stop" + approvals._pending.pop(sid, None) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-stop", "command": "first"}) + captured = {} + called = {"cancel": False} + + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_args: True) + monkeypatch.setattr(routes, "webui_gateway_chat_enabled", lambda *_args, **_kwargs: True) + monkeypatch.setattr(routes, "stream_owner_session_id", lambda _stream_id: sid) + monkeypatch.setattr(routes, "cancel_stream", lambda _stream_id: called.__setitem__("cancel", True)) + monkeypatch.setattr("api.gateway_chat.stop_gateway_run", lambda _stream_id: False) + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(routes, "j", fake_j) + parsed = urllib.parse.urlparse(f"/api/chat/cancel?stream_id={stream_id}") + try: + routes.handle_get(object(), parsed) + assert captured["status"] == 502 + assert captured["payload"] == { + "ok": False, + "cancelled": False, + "stream_id": stream_id, + "error": "Gateway stop failed", + } + assert called["cancel"] is False + assert sid in approvals._pending + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + + +def test_chat_cancel_blocks_while_gateway_run_id_is_pending(monkeypatch): + import urllib.parse + + from api import routes + + stream_id = "stream-cancel-pending-run-id" + captured = {} + called = {"cancel": False, "stop": False} + + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_args: True) + monkeypatch.setattr(routes, "cancel_stream", lambda _stream_id: called.__setitem__("cancel", True)) + monkeypatch.setattr("api.gateway_chat.gateway_run_id_pending", lambda _stream_id: True) + monkeypatch.setattr("api.gateway_chat.stop_gateway_run", lambda _stream_id: called.__setitem__("stop", True)) + monkeypatch.setattr("time.sleep", lambda _seconds: None) + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(routes, "j", fake_j) + parsed = urllib.parse.urlparse(f"/api/chat/cancel?stream_id={stream_id}") + routes.handle_get(object(), parsed) + assert captured["status"] == 502 + assert captured["payload"] == { + "ok": False, + "cancelled": False, + "stream_id": stream_id, + "error": "Gateway stop failed", + } + assert called["cancel"] is False + assert called["stop"] is False + + +def test_chat_cancel_waits_for_pending_gateway_run_id_then_stops_remote_run(monkeypatch): + import urllib.parse + + from api import routes + from api.gateway_chat import _STREAM_RUN_IDS + + stream_id = "stream-cancel-pending-wait" + captured = {} + called = {"cancel": False, "stop": False, "pending": True} + + def fake_sleep(_seconds): + _STREAM_RUN_IDS[stream_id] = "run-stop" + called["pending"] = False + + monkeypatch.setattr(routes, "_stream_id_visible_to_request_profile", lambda *_args: True) + monkeypatch.setattr(routes, "cancel_stream", lambda _stream_id: called.__setitem__("cancel", True) or True) + monkeypatch.setattr("api.gateway_chat.gateway_run_id_pending", lambda _stream_id: called["pending"]) + monkeypatch.setattr("api.gateway_chat.stop_gateway_run", lambda _stream_id: called.__setitem__("stop", True) or True) + monkeypatch.setattr("time.sleep", fake_sleep) + + def fake_j(handler, data, status=200, extra_headers=None): + captured["payload"] = data + captured["status"] = status + return data + + monkeypatch.setattr(routes, "j", fake_j) + parsed = urllib.parse.urlparse(f"/api/chat/cancel?stream_id={stream_id}") + try: + routes.handle_get(object(), parsed) + assert captured["status"] == 200 + assert captured["payload"] == {"ok": True, "cancelled": True, "stream_id": stream_id} + assert called["stop"] is True + assert called["cancel"] is True + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + + def test_stale_gateway_card_does_not_relay_live_run(): from api.gateway_chat import _STREAM_RUN_IDS import api.route_approvals as approvals @@ -581,6 +1479,101 @@ def test_retained_gateway_card_does_not_relay_different_active_run(): approvals._pending.pop(sid, None) +def test_stale_same_run_gateway_card_does_not_relay_live_head(): + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + + sid = "sess-stale-same-run" + stream_id = "sid-stale-same-run" + _STREAM_RUN_IDS[stream_id] = "run-shared" + approvals._pending.pop(sid, None) + try: + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-shared", "command": "first"}) + approvals.submit_gateway_pending_mirror(sid, {"run_id": "run-shared", "command": "second"}) + queue = approvals._pending[sid] + stale_id = queue[0]["approval_id"] + live_id = queue[1]["approval_id"] + assert approvals.retire_gateway_pending_mirror(sid, approval_id=stale_id, run_id="run-shared") + + mock_session = MagicMock() + mock_session.active_stream_id = stream_id + handler = MagicMock() + handler.wfile = io.BytesIO() + + with patch("api.routes.get_session", return_value=mock_session), \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=""), \ + patch("api.runner_client.HttpRunnerClient._request_json") as request_json: + from api.routes import _handle_approval_respond + _handle_approval_respond(handler, { + "session_id": sid, + "choice": "once", + "approval_id": stale_id, + }) + + payload = json.loads(handler.wfile.getvalue().decode("utf-8")) + assert handler.send_response.call_args.args[0] == 409 + assert payload["code"] == "gateway_run_unavailable" + request_json.assert_not_called() + assert approvals.gateway_pending_mirror(sid, approval_id=live_id) is not None + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + + +def test_token_stale_same_run_gateway_card_does_not_relay_advanced_live_head(): + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + + sid = "sess-token-stale-same-run" + stream_id = "sid-token-stale-same-run" + _STREAM_RUN_IDS[stream_id] = "run-shared" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [SimpleNamespace(data={ + "run_id": "run-shared", + "approval_id": "gwrun:run-shared:a", + "command": "first", + })] + approvals.submit_gateway_pending_mirror(sid, dict(approvals._gateway_queues[sid][0].data)) + stale_id = approvals._pending[sid][0]["approval_id"] + + approvals._gateway_queues[sid] = [SimpleNamespace(data={ + "run_id": "run-shared", + "approval_id": "gwrun:run-shared:b", + "command": "second", + })] + approvals.submit_gateway_pending_mirror(sid, dict(approvals._gateway_queues[sid][0].data)) + + mock_session = MagicMock() + mock_session.active_stream_id = stream_id + handler = MagicMock() + handler.wfile = io.BytesIO() + + with patch("api.routes.get_session", return_value=mock_session), \ + patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ + patch("api.gateway_chat._gateway_api_key", return_value=""), \ + patch("api.runner_client.HttpRunnerClient._request_json") as request_json: + from api.routes import _handle_approval_respond + _handle_approval_respond(handler, { + "session_id": sid, + "choice": "once", + "approval_id": stale_id, + }) + + payload = json.loads(handler.wfile.getvalue().decode("utf-8")) + assert handler.send_response.call_args.args[0] == 409 + assert payload["code"] == "gateway_run_unavailable" + request_json.assert_not_called() + assert approvals.gateway_pending_mirror(sid, approval_id="gwrun:run-shared:a") is None + assert approvals.gateway_pending_mirror(sid, approval_id="gwrun:run-shared:b") is not None + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + def test_gateway_runs_api_streaming_preserves_multimodal_input(): """Attachment-backed runs requests must keep multimodal content lists.""" from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming @@ -752,8 +1745,8 @@ def test_gateway_approval_response_relay(): approvals._pending.pop("sess-relay", None) -def test_gateway_approval_response_without_approval_id_routes_by_run_id(): - """Runs API approvals resolve by the server-owned run ID.""" +def test_gateway_approval_response_without_approval_id_409s(): + """Gateway relay requires the emitted per-approval id, not bare run state.""" from api.gateway_chat import _STREAM_RUN_IDS import api.route_approvals as approvals @@ -764,20 +1757,13 @@ def test_gateway_approval_response_without_approval_id_routes_by_run_id(): mock_session = MagicMock() mock_session.active_stream_id = "sid-relay-empty-id" - captured = {} - - def fake_request_json(self, req): - captured["url"] = req.full_url - captured["body"] = json.loads(req.data) - return {"ok": True} - handler = MagicMock() handler.wfile = io.BytesIO() with patch("api.routes.get_session", return_value=mock_session), \ - patch("api.runner_client.HttpRunnerClient._request_json", new=fake_request_json), \ patch("api.gateway_chat._gateway_base_url", return_value="http://gw:8642"), \ patch("api.gateway_chat._gateway_api_key", return_value=""), \ + patch("api.runner_client.HttpRunnerClient._request_json") as request_json, \ patch("api.routes._resolve_approval_legacy", return_value=True) as cleanup: from api.routes import _handle_approval_respond _handle_approval_respond( @@ -787,41 +1773,17 @@ def test_gateway_approval_response_without_approval_id_routes_by_run_id(): payload = json.loads(handler.wfile.getvalue().decode("utf-8")) status = handler.send_response.call_args.args[0] - assert status == 200, f"status={status} payload={payload} outbound={captured}" - assert captured["url"] == "http://gw:8642/v1/runs/run%20abc%2F1/approval" - assert captured["body"] == {"choice": "once", "approval_id": "gwrun:run abc/1"} - assert payload["ok"] is True - assert payload["relayed"] is True - cleanup.assert_called_once_with("sess-relay", "gwrun:run abc/1", "once") + assert status == 409, f"status={status} payload={payload}" + assert payload["ok"] is False + assert payload["relayed"] is False + assert payload["code"] == "gateway_run_unavailable" + request_json.assert_not_called() + cleanup.assert_not_called() _STREAM_RUN_IDS.pop("sid-relay-empty-id", None) approvals._pending.pop("sess-relay", None) -def test_gateway_mirrored_pending_approval_id_by_run_id_skips_older_local_entries(): - """Run-scoped cleanup must target the mirrored gateway card, not queue head.""" - import api.route_approvals as approvals - - sid = "sess-relay-mirror" - approvals._pending.pop(sid, None) - approvals._gateway_queues.pop(sid, None) - try: - approvals._pending[sid] = [{"approval_id": "local-1", "command": "echo local"}] - approvals._gateway_queues[sid] = [SimpleNamespace(data={ - "approval_id": "gateway-real", - "command": "echo gateway", - "run_id": "run-xyz", - })] - approvals.submit_gateway_pending_mirror(sid, {}) - queue = approvals._pending[sid] - assert queue[0]["approval_id"] == "local-1" - assert queue[1]["approval_id"] == "gateway-real" - assert approvals._gateway_mirrored_pending_approval_id_by_run_id(sid, "run-xyz") == "gateway-real" - finally: - approvals._pending.pop(sid, None) - approvals._gateway_queues.pop(sid, None) - - def test_gateway_approval_response_relay_failure_returns_502(): """Gateway relay failures must surface as HTTP errors to the frontend.""" from api.gateway_chat import _STREAM_RUN_IDS diff --git a/tests/test_issue4771_local_approval_regression.py b/tests/test_issue4771_local_approval_regression.py index 2718c18e..7b62795a 100644 --- a/tests/test_issue4771_local_approval_regression.py +++ b/tests/test_issue4771_local_approval_regression.py @@ -40,6 +40,7 @@ import pytest # Import order matters: tools.* will not resolve until api.config has run. from api import routes from api import models +from api.gateway_chat import _STREAM_RUN_IDS try: import tools.approval as ta @@ -182,6 +183,101 @@ def test_local_mirrored_approval_deny_resolves(): _cleanup(sid) +def test_local_mirrored_approval_with_gwrun_prefix_still_resolves(): + """A local approval id that happens to start with gwrun: stays local.""" + sid = f"local-approval-prefixed-{uuid.uuid4().hex[:8]}" + _register_session(sid) + try: + entry, _approval_id = _seed_local_pending_approval(sid) + with ta._lock: + head = ta._pending[sid][0] + head["approval_id"] = "gwrun:local-test" + handler = _FakeHandler() + with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False): + routes._handle_approval_respond( + handler, + {"session_id": sid, "choice": "once", "approval_id": "gwrun:local-test"}, + ) + resp = handler.json() + assert handler.status == 200, f"expected 200, got {handler.status}: {resp}" + assert resp.get("ok") is True + assert resp.get("code") != "gateway_run_unavailable" + assert entry.event.is_set() + assert entry.result == "once" + finally: + _cleanup(sid) + + +def test_local_live_head_does_not_inherit_stale_remote_run(): + """A stale remote mirror cannot seize a live local approval with the same id.""" + sid = f"local-approval-collision-{uuid.uuid4().hex[:8]}" + _register_session(sid) + try: + entry, approval_id = _seed_local_pending_approval(sid) + with ta._lock: + local_queue = ta._pending.get(sid) + local_entries = list(local_queue) if isinstance(local_queue, list) else [local_queue] + ta._pending[sid] = [{ + ra._GATEWAY_MIRROR_FLAG: True, + "run_id": "remote-run", + "approval_id": approval_id, + "command": "stale-remote", + }, *local_entries] + handler = _FakeHandler() + with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \ + patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval: + routes._handle_approval_respond( + handler, + {"session_id": sid, "choice": "once", "approval_id": approval_id}, + ) + resp = handler.json() + assert handler.status == 200, f"expected 200, got {handler.status}: {resp}" + assert resp.get("ok") is True + assert resp.get("relayed") is not True + respond_approval.assert_not_called() + assert entry.event.is_set() + assert entry.result == "once" + finally: + _cleanup(sid) + + +def test_local_live_head_beats_active_remote_candidate_run(): + """A local approval still wins when an active remote run shares its id.""" + sid = f"local-approval-active-remote-{uuid.uuid4().hex[:8]}" + session = _register_session(sid) + stream_id = f"remote-stream-{uuid.uuid4().hex[:8]}" + session.active_stream_id = stream_id + _STREAM_RUN_IDS[stream_id] = "remote-run" + try: + entry, approval_id = _seed_local_pending_approval(sid) + with ta._lock: + local_queue = ta._pending.get(sid) + local_entries = list(local_queue) if isinstance(local_queue, list) else [local_queue] + ta._pending[sid] = [{ + ra._GATEWAY_MIRROR_FLAG: True, + "run_id": "remote-run", + "approval_id": approval_id, + "command": "remote-collision", + }, *local_entries] + handler = _FakeHandler() + with patch("api.gateway_chat.webui_gateway_chat_enabled", return_value=False), \ + patch("api.runner_client.HttpRunnerClient.respond_approval") as respond_approval: + routes._handle_approval_respond( + handler, + {"session_id": sid, "choice": "once", "approval_id": approval_id}, + ) + resp = handler.json() + assert handler.status == 200, f"expected 200, got {handler.status}: {resp}" + assert resp.get("ok") is True + assert resp.get("relayed") is not True + respond_approval.assert_not_called() + assert entry.event.is_set() + assert entry.result == "once" + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + _cleanup(sid) + + def test_gateway_mirrored_approval_without_run_still_409s(): """#4771 behaviour preserved: on the gateway backend, a mirrored approval whose run is gone must still surface the actionable 409 so the card stays diff --git a/tests/test_issue6008_sidebar_stop_failure_ui.py b/tests/test_issue6008_sidebar_stop_failure_ui.py new file mode 100644 index 00000000..6965ab3b --- /dev/null +++ b/tests/test_issue6008_sidebar_stop_failure_ui.py @@ -0,0 +1,144 @@ +"""Sidebar stop must not settle the UI when /api/chat/cancel fails.""" +from __future__ import annotations + +import json +import pathlib +import re +import subprocess + + +REPO = pathlib.Path(__file__).parent.parent +BOOT_JS = (REPO / "static" / "boot.js").read_text(encoding="utf-8") +COMMANDS_JS = (REPO / "static" / "commands.js").read_text(encoding="utf-8") +MESSAGES_JS = (REPO / "static" / "messages.js").read_text(encoding="utf-8") +SESSIONS_JS = (REPO / "static" / "sessions.js").read_text(encoding="utf-8") + + +def _extract_function(src: str, name: str) -> str: + m = re.search(rf"async function {name}\s*\(", src) + assert m, f"{name} not found in static/boot.js" + brace_pos = src.index("{", m.end()) + depth = 1 + pos = brace_pos + 1 + while pos < len(src) and depth > 0: + ch = src[pos] + if ch == "{": + depth += 1 + elif ch == "}": + depth -= 1 + pos += 1 + return src[m.start():pos] + + +CANCEL_SESSION_STREAM_SRC = _extract_function(BOOT_JS, "cancelSessionStream") + + +def test_source_gates_sidebar_settle_on_http_success(): + assert "return false" in CANCEL_SESSION_STREAM_SRC + assert "return true" in CANCEL_SESSION_STREAM_SRC + assert "r.ok" in CANCEL_SESSION_STREAM_SRC, ( + "cancelSessionStream() must check the /api/chat/cancel HTTP status before " + "closing local UI state" + ) + assert "if(!respOk)returnfalse;" in "".join(CANCEL_SESSION_STREAM_SRC.split()), ( + "cancelSessionStream() must bail out on failed stop responses" + ) + + +def test_stop_callers_gate_success_toasts_on_cancel_result(): + compact_commands = "".join(COMMANDS_JS.split()) + compact_messages = "".join(MESSAGES_JS.split()) + compact_sessions = "".join(SESSIONS_JS.split()) + assert "if(awaitcancelStream('slash-stop'))showToast(t('stream_stopped'))" in compact_commands + assert "elseshowToast(t('cancel_failed'))" in compact_commands + assert "if(awaitcancelStream('slash-interrupt'))showToast(t('cmd_interrupt_confirm'),2000)" in compact_commands + assert "if(awaitcancelStream('busy-interrupt'))showToast(t('busy_interrupt_confirm'),2000)" in compact_messages + assert "if(awaitcancelSessionStream(session))showToast(t('stream_stopped'))" in compact_sessions + + +_NODE_SCRIPT = r''' +const M = { + closeCalls: [], + busyCalls: [], + composerCalls: [], + statusCalls: [], + renderCalls: 0, + clearCalls: [], + approvalStops: 0, + approvalHides: 0, + clarifyStops: 0, + clarifyHides: 0, + fetchCalls: [], +}; + +globalThis.INFLIGHT = { 'sid-1': { streamId: 'stream-1' } }; +globalThis.S = { activeStreamId: 'stream-1', session: { session_id: 'sid-1', active_stream_id: 'stream-1' } }; +globalThis.closeLiveStream = (...a) => M.closeCalls.push(a); +globalThis.clearInflightState = (sid) => M.clearCalls.push(['clearInflightState', sid]); +globalThis.clearInflight = () => M.clearCalls.push(['clearInflight']); +globalThis.setBusy = (v) => M.busyCalls.push(v); +globalThis.setComposerStatus = (v) => M.composerCalls.push(v); +globalThis.setStatus = (v) => M.statusCalls.push(v); +globalThis.stopApprovalPolling = () => M.approvalStops += 1; +globalThis.hideApprovalCard = () => M.approvalHides += 1; +globalThis.stopClarifyPolling = () => M.clarifyStops += 1; +globalThis.hideClarifyCard = () => M.clarifyHides += 1; +globalThis.renderSessionList = () => M.renderCalls += 1; +globalThis._approvalSessionId = 'sid-1'; +globalThis._clarifySessionId = 'sid-1'; +globalThis.document = { baseURI: 'http://localhost:8787/' }; +globalThis.location = { href: 'http://localhost:8787/' }; +globalThis.fetch = (url, opts) => { + M.fetchCalls.push({ url: String(url), opts }); + return Promise.resolve({ + ok: false, + json: () => Promise.resolve({ ok: false, cancelled: false, stream_id: 'stream-1' }), + }); +}; + +__CANCEL_SESSION_STREAM_SRC__ + +const session = { session_id: 'sid-1', active_stream_id: 'stream-1' }; +await cancelSessionStream(session); +console.log(JSON.stringify({ + sessionActiveStreamId: session.active_stream_id, + activeStreamId: globalThis.S.activeStreamId, + closeCalls: M.closeCalls, + busyCalls: M.busyCalls, + composerCalls: M.composerCalls, + renderCalls: M.renderCalls, + clearCalls: M.clearCalls, + approvalStops: M.approvalStops, + approvalHides: M.approvalHides, + clarifyStops: M.clarifyStops, + clarifyHides: M.clarifyHides, + fetchCalls: M.fetchCalls.length, +})); +''' + + +def test_failed_sidebar_stop_keeps_local_state(): + script = _NODE_SCRIPT.replace("__CANCEL_SESSION_STREAM_SRC__", CANCEL_SESSION_STREAM_SRC) + completed = subprocess.run( + ["node", "--input-type=module", "-e", script], + cwd=str(REPO), + text=True, + capture_output=True, + timeout=30, + ) + assert completed.returncode == 0, ( + f"node subprocess failed:\n--- stdout ---\n{completed.stdout}\n--- stderr ---\n{completed.stderr}" + ) + result = json.loads(completed.stdout.splitlines()[-1]) + assert result["fetchCalls"] == 1 + assert result["sessionActiveStreamId"] == "stream-1" + assert result["activeStreamId"] == "stream-1" + assert result["closeCalls"] == [] + assert result["busyCalls"] == [] + assert result["composerCalls"] == [] + assert result["renderCalls"] == 0 + assert result["clearCalls"] == [] + assert result["approvalStops"] == 0 + assert result["approvalHides"] == 0 + assert result["clarifyStops"] == 0 + assert result["clarifyHides"] == 0