From 6460c48ce69316a7ab4674a3ed7095e19ebbc203 Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Thu, 16 Jul 2026 04:37:50 -0400 Subject: [PATCH] fix(#6008): keep gateway approvals attached to the right run --- api/route_approvals.py | 63 +++- api/routes.py | 85 ++++- tests/test_gateway_approval_runs_api.py | 313 +++++++++++++++++- ...est_issue4771_local_approval_regression.py | 103 ++++++ tests/test_runtime_adapter_seam.py | 2 +- 5 files changed, 546 insertions(+), 20 deletions(-) diff --git a/api/route_approvals.py b/api/route_approvals.py index 6ab739d8..f524e852 100644 --- a/api/route_approvals.py +++ b/api/route_approvals.py @@ -47,6 +47,7 @@ _approval_sse_subscribers: dict[str, list[queue.Queue]] = {} _GATEWAY_MIRROR_FLAG = "_gateway_mirror" _GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token" _GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token" +_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY = "_webui_local_head" def _approval_sse_subscribe(session_id: str) -> queue.Queue: @@ -106,7 +107,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 @@ -115,12 +116,14 @@ def _gateway_mirror_entry_token(entry) -> str: """ data = getattr(entry, "data", None) if isinstance(data, dict): + if data.get(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY): + return None token = data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) if not token: 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: @@ -203,6 +206,13 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No live_mirror_present = True 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: @@ -305,6 +315,9 @@ def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_i continue retained_gateway_queue.append(entry) if not retired and not gateway_queue_changed: + if approval_id: + head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) + _approval_sse_notify_locked(session_key, head, total) return False for match in retired: entries.remove(match) @@ -339,11 +352,30 @@ 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() + approval_id = str(approval.get("approval_id") or "").strip() + live_gateway_queue = _gateway_queues.get(session_key) or [] + live_head_entry = live_gateway_queue[0] if live_gateway_queue else None + live_head_data = getattr(live_head_entry, "data", None) or {} + matching_no_run_live_head = ( + not run_id + and live_head_data + and str(live_head_data.get("command") or "") == str(approval.get("command") or "") + and str(live_head_data.get("description") or "") == str(approval.get("description") or "") + and str(live_head_data.get("pattern_key") or "") == str(approval.get("pattern_key") or "") + and tuple(str(k) for k in (live_head_data.get("pattern_keys") or [])) + == tuple(str(k) for k in (approval.get("pattern_keys") or [])) + ) + if matching_no_run_live_head and not live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY): + live_head_data.pop(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY, None) + live_head_data[_GATEWAY_ENTRY_DATA_TOKEN_KEY] = uuid.uuid4().hex + elif not run_id and live_head_data: + live_head_data[_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY] = True + if not run_id and not approval_id: + approval_id = uuid.uuid4().hex + approval["approval_id"] = approval_id + if matching_no_run_live_head and approval_id and not str(live_head_data.get("approval_id") or "").strip(): + live_head_data["approval_id"] = approval_id if run_id: - live_gateway_queue = _gateway_queues.get(session_key) or [] - 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 = ( @@ -386,6 +418,25 @@ def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: ) if mirror_entry: approval["approval_id"] = str(mirror_entry.get("approval_id") or approval_id).strip() + elif 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) _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 9151abf4..016df3a6 100644 --- a/api/routes.py +++ b/api/routes.py @@ -9715,6 +9715,8 @@ from api.route_approvals import ( # noqa: F401 — re-exports for backward comp _approval_sse_notify_locked, _approval_sse_notify, _GATEWAY_MIRROR_FLAG, + _GATEWAY_MIRROR_TOKEN, + _gateway_mirror_entry_token, gateway_pending_mirror, retire_gateway_pending_mirror, reconcile_gateway_pending_mirror_locked, @@ -23806,15 +23808,24 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st 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: continue if run_id and str(entry.get("run_id") or "").strip() != run_id: continue - pending = queue.pop(i) + 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 - break else: # A stale explicit id must not accidentally approve the # oldest queued command; duplicate/stale responses are @@ -23993,10 +24004,11 @@ def _handle_approval_respond(handler, body): _candidate_run_id = _STREAM_RUN_IDS.get(active_sid) 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 [] + 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 @@ -24014,11 +24026,60 @@ def _handle_approval_respond(handler, body): 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 + 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, + { + "ok": False, + "choice": choice, + "relayed": False, + "code": "gateway_run_unavailable", + "error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE, + }, + 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: diff --git a/tests/test_gateway_approval_runs_api.py b/tests/test_gateway_approval_runs_api.py index 19cd3094..3ba15e9a 100644 --- a/tests/test_gateway_approval_runs_api.py +++ b/tests/test_gateway_approval_runs_api.py @@ -9,6 +9,8 @@ import urllib.error from types import SimpleNamespace from unittest.mock import MagicMock, patch +import pytest + # --------------------------------------------------------------------------- # 1. Capability detection @@ -603,7 +605,10 @@ 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 + ta = pytest.importorskip( + "tools.approval", + reason="tools.approval not available in this environment", + ) sid = "sess-remote-relay-local-head" stream_id = "sid-remote-relay-local-head" @@ -746,6 +751,132 @@ def test_direct_remote_approval_still_mirrors_with_local_gateway_head(): approvals._gateway_queues.pop(sid, None) +def test_legacy_gateway_approval_without_run_gets_browser_visible_id(): + import api.route_approvals as approvals + + sid = "sess-legacy-no-run-id" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [SimpleNamespace(data={"command": "legacy"})] + approval = {"command": "legacy"} + approvals.submit_gateway_pending_mirror(sid, approval) + assert approval.get("approval_id") + queue = approvals._pending.get(sid) or [] + assert queue[0]["approval_id"] == approval["approval_id"] + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_legacy_gateway_approval_without_run_keeps_its_own_id_beside_local_head(): + import api.route_approvals as approvals + + sid = "sess-legacy-no-run-local-head" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [{"approval_id": "local-id", "command": "local"}] + approvals._gateway_queues[sid] = [SimpleNamespace(data={"command": "legacy"})] + approval = {"command": "legacy"} + approvals.submit_gateway_pending_mirror(sid, approval) + assert approval.get("approval_id") + assert approval["approval_id"] != "local-id" + queue = approvals._pending.get(sid) or [] + mirrors = [ + entry for entry in queue + if entry.get(approvals._GATEWAY_MIRROR_FLAG) + and not str(entry.get("run_id") or "").strip() + ] + assert mirrors + assert mirrors[-1]["approval_id"] == approval["approval_id"] + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_legacy_gateway_approval_without_run_keeps_its_own_id_beside_local_gateway_head(): + import api.route_approvals as approvals + ta = pytest.importorskip( + "tools.approval", + reason="tools.approval not available in this environment", + ) + + sid = "sess-legacy-no-run-local-gateway-head" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._gateway_queues[sid] = [ta._ApprovalEntry({ + "approval_id": "local-id", + "command": "local-head", + })] + approval = {"approval_id": "remote-id", "command": "legacy"} + approvals.submit_gateway_pending_mirror(sid, approval) + assert approval["approval_id"] == "remote-id" + queue = approvals._pending.get(sid) or [] + mirrors = [ + entry for entry in queue + if entry.get(approvals._GATEWAY_MIRROR_FLAG) + and not str(entry.get("run_id") or "").strip() + and entry.get("approval_id") == "remote-id" + ] + assert mirrors + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_identical_legacy_gateway_approvals_without_run_keep_distinct_ids(): + import api.route_approvals as approvals + + sid = "sess-legacy-no-run-identical" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + first = {"command": "legacy"} + second = {"command": "legacy"} + approvals.submit_gateway_pending_mirror(sid, first) + approvals.submit_gateway_pending_mirror(sid, second) + assert first["approval_id"] != second["approval_id"] + queue = approvals._pending.get(sid) or [] + mirrored_ids = [ + entry.get("approval_id") + for entry in queue + if entry.get(approvals._GATEWAY_MIRROR_FLAG) + and not str(entry.get("run_id") or "").strip() + ] + assert mirrored_ids == [first["approval_id"], second["approval_id"]] + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_identical_legacy_gateway_approvals_with_explicit_ids_keep_distinct_ids(): + import api.route_approvals as approvals + + sid = "sess-legacy-no-run-identical-explicit" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + first = {"approval_id": "upstream-a", "command": "legacy"} + second = {"approval_id": "upstream-b", "command": "legacy"} + approvals.submit_gateway_pending_mirror(sid, first) + approvals.submit_gateway_pending_mirror(sid, second) + assert first["approval_id"] == "upstream-a" + assert second["approval_id"] == "upstream-b" + queue = approvals._pending.get(sid) or [] + mirrored_ids = [ + entry.get("approval_id") + for entry in queue + if entry.get(approvals._GATEWAY_MIRROR_FLAG) + and not str(entry.get("run_id") or "").strip() + ] + assert mirrored_ids == ["upstream-a", "upstream-b"] + 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 @@ -874,6 +1005,30 @@ def test_exact_one_retirement_keeps_same_run_gateway_queue_state(): approvals._gateway_queues.pop(sid, None) +def test_retire_gateway_pending_mirror_notifies_reconciled_successor_when_target_missing(): + import api.route_approvals as approvals + + sid = "sess-retire-notify-successor" + approvals._pending.pop(sid, None) + try: + approvals._pending[sid] = [{ + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-a", + "approval_id": "appr-b", + "command": "b", + }] + notifications = [] + + def fake_notify(_sid, head, total): + notifications.append((head["approval_id"] if head else None, total)) + + with patch.object(approvals, "_approval_sse_notify_locked", side_effect=fake_notify): + assert approvals.retire_gateway_pending_mirror(sid, approval_id="appr-a", run_id="run-a") is False + assert notifications[-1] == ("appr-b", 1) + finally: + approvals._pending.pop(sid, None) + + def test_local_stop_uses_runs_api_stop_endpoint(): from api.gateway_chat import _STREAM_RUN_IDS, stop_gateway_run @@ -1521,6 +1676,162 @@ def test_stale_same_run_gateway_card_does_not_relay_live_head(): approvals._pending.pop(sid, None) +def test_stale_same_run_gateway_card_without_token_does_not_relay_live_head(): + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + ta = pytest.importorskip( + "tools.approval", + reason="tools.approval not available in this environment", + ) + + sid = "sess-stale-same-run-no-token" + stream_id = "sid-stale-same-run-no-token" + _STREAM_RUN_IDS[stream_id] = "run-shared" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [{ + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-shared", + "approval_id": "stale-a", + "command": "stale", + }] + approvals._gateway_queues[sid] = [ta._ApprovalEntry({ + "run_id": "run-shared", + "approval_id": "live-b", + "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.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-a", + }) + + 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-b", run_id="run-shared") is not None + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_stale_same_run_gateway_card_without_token_after_teardown_does_not_relay_live_head(): + import api.route_approvals as approvals + ta = pytest.importorskip( + "tools.approval", + reason="tools.approval not available in this environment", + ) + + sid = "sess-stale-same-run-no-token-teardown" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [{ + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-shared", + "approval_id": "stale-a", + "command": "stale", + }] + approvals._gateway_queues[sid] = [ta._ApprovalEntry({ + "run_id": "run-shared", + "approval_id": "live-b", + "command": "live", + })] + + 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._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-a", + }) + + 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-b", run_id="run-shared") is not None + finally: + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + + +def test_stale_same_run_gateway_card_without_token_does_not_relay_live_head_when_live_id_is_synthesized(): + from api.gateway_chat import _STREAM_RUN_IDS + import api.route_approvals as approvals + ta = pytest.importorskip( + "tools.approval", + reason="tools.approval not available in this environment", + ) + + sid = "sess-stale-same-run-no-token-synthesized-live-id" + stream_id = "sid-stale-same-run-no-token-synthesized-live-id" + _STREAM_RUN_IDS[stream_id] = "run-shared" + approvals._pending.pop(sid, None) + approvals._gateway_queues.pop(sid, None) + try: + approvals._pending[sid] = [{ + approvals._GATEWAY_MIRROR_FLAG: True, + "run_id": "run-shared", + "approval_id": "stale-a", + "command": "stale", + }] + approvals._gateway_queues[sid] = [ta._ApprovalEntry({ + "run_id": "run-shared", + "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.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-a", + }) + + 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() + live_mirror = approvals.gateway_pending_mirror(sid, run_id="run-shared") + assert live_mirror is not None + assert live_mirror["approval_id"] != "stale-a" + finally: + _STREAM_RUN_IDS.pop(stream_id, None) + approvals._pending.pop(sid, None) + approvals._gateway_queues.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 diff --git a/tests/test_issue4771_local_approval_regression.py b/tests/test_issue4771_local_approval_regression.py index 7b62795a..5a3da490 100644 --- a/tests/test_issue4771_local_approval_regression.py +++ b/tests/test_issue4771_local_approval_regression.py @@ -278,6 +278,109 @@ def test_local_live_head_beats_active_remote_candidate_run(): _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_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_runtime_adapter_seam.py b/tests/test_runtime_adapter_seam.py index ee3a5208..54ca5017 100644 --- a/tests/test_runtime_adapter_seam.py +++ b/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