fix(#6008): route gateway approvals by run ID
This commit is contained in:
1 parent
e9e9ed09a5
commit
39fc169834
4 files changed
+375
-31
No files matched your search
+13
-3
@@ -338,8 +338,8 @@ 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:
|
||||
approval_id = uuid.uuid4().hex
|
||||
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")
|
||||
@@ -478,6 +478,10 @@ def _run_gateway_runs_api_streaming(
|
||||
approval_data = _gateway_runs_approval_event(payload)
|
||||
if approval_data:
|
||||
approval_data["run_id"] = run_id
|
||||
if not approval_data.get("approval_id"):
|
||||
approval_data["approval_id"] = f"gwrun:{run_id}"
|
||||
from api.route_approvals import submit_gateway_pending_mirror
|
||||
submit_gateway_pending_mirror(session_id, approval_data)
|
||||
put_gateway_event("approval", approval_data)
|
||||
sse_event = "message"
|
||||
continue
|
||||
@@ -522,6 +526,8 @@ def _run_gateway_runs_api_streaming(
|
||||
sse_event = "message"
|
||||
continue
|
||||
if payload_event == "run.completed":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
if payload.get("error"):
|
||||
raise RuntimeError(str(payload["error"]))
|
||||
output = str(payload.get("output") or "")
|
||||
@@ -533,8 +539,12 @@ def _run_gateway_runs_api_streaming(
|
||||
sse_event = "message"
|
||||
continue
|
||||
if payload_event == "run.failed":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
raise RuntimeError(str(payload.get("error") or "Gateway run failed"))
|
||||
if payload_event == "run.cancelled":
|
||||
from api.route_approvals import retire_gateway_pending_mirror
|
||||
retire_gateway_pending_mirror(session_id, run_id=run_id)
|
||||
put_gateway_event("cancel", {"message": "Cancelled by gateway"})
|
||||
return None, usage
|
||||
reasoning_delta = _gateway_sse_reasoning_delta(payload)
|
||||
@@ -908,12 +918,12 @@ def _run_gateway_chat_streaming(
|
||||
_approval_run_id = str(approval_data.get("run_id") or "").strip()
|
||||
if _approval_run_id:
|
||||
_STREAM_RUN_IDS[stream_id] = _approval_run_id
|
||||
put_gateway_event("approval", approval_data)
|
||||
try:
|
||||
from api.route_approvals import submit_gateway_pending_mirror
|
||||
submit_gateway_pending_mirror(session_id, approval_data)
|
||||
except Exception:
|
||||
logger.debug("submit_gateway_pending_mirror failed", exc_info=True)
|
||||
put_gateway_event("approval", approval_data)
|
||||
else:
|
||||
logger.debug("Ignoring malformed gateway approval payload")
|
||||
sse_event = "message"
|
||||
|
||||
+83
-19
@@ -155,6 +155,9 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
|
||||
if not _is_gateway_mirror_entry(entry):
|
||||
rebuilt.append(entry)
|
||||
continue
|
||||
if str(entry.get("run_id") or "").strip():
|
||||
rebuilt.append(entry)
|
||||
continue
|
||||
if live_token and entry.get(_GATEWAY_MIRROR_TOKEN) == live_token and not live_mirror_present:
|
||||
rebuilt.append(entry)
|
||||
live_mirror_present = True
|
||||
@@ -163,7 +166,8 @@ 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_entry.setdefault("approval_id", uuid.uuid4().hex)
|
||||
mirror_run_id = str(mirror_entry.get("run_id") or "").strip()
|
||||
mirror_entry.setdefault("approval_id", f"gwrun:{mirror_run_id}" 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)
|
||||
@@ -184,35 +188,95 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
|
||||
return head, total, changed
|
||||
|
||||
|
||||
def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str | None:
|
||||
"""Return the mirrored gateway approval run_id for a matching pending card.
|
||||
def _gateway_pending_mirror_locked(session_key: str, approval_id: str = "", run_id: str = "") -> dict | None:
|
||||
"""Return the exact live run-backed mirror under `_lock`."""
|
||||
approval_id = str(approval_id or "").strip()
|
||||
run_id = str(run_id or "").strip()
|
||||
queue = _pending.get(session_key)
|
||||
entries = queue if isinstance(queue, list) else [queue] if queue else []
|
||||
if approval_id:
|
||||
for entry in entries:
|
||||
if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip():
|
||||
continue
|
||||
if entry.get("approval_id") == approval_id:
|
||||
if run_id and entry.get("run_id") != run_id:
|
||||
return None
|
||||
return entry
|
||||
return None
|
||||
for entry in entries:
|
||||
if not _is_gateway_mirror_entry(entry) or not str(entry.get("run_id") or "").strip():
|
||||
continue
|
||||
if run_id and entry.get("run_id") == run_id:
|
||||
return entry
|
||||
return None
|
||||
|
||||
Reconciles the mirror first so a live gateway head still survives a lost
|
||||
`active_stream_id` pointer.
|
||||
"""
|
||||
|
||||
def gateway_pending_mirror(session_key: str, approval_id: str = "", run_id: str = "") -> dict | None:
|
||||
"""Return an exact live run-backed mirror for this session."""
|
||||
with _lock:
|
||||
reconcile_gateway_pending_mirror_locked(session_key)
|
||||
entry = _gateway_pending_mirror_locked(session_key, approval_id, run_id)
|
||||
return dict(entry) if entry else None
|
||||
|
||||
|
||||
def 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."""
|
||||
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:
|
||||
return False
|
||||
entries.remove(match)
|
||||
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))
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return True
|
||||
|
||||
|
||||
def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str | None:
|
||||
"""Compatibility wrapper for exact run-backed lookup."""
|
||||
approval_id = str(approval_id or "").strip()
|
||||
if not approval_id:
|
||||
return None
|
||||
with _lock:
|
||||
reconcile_gateway_pending_mirror_locked(session_key)
|
||||
queue = _pending.get(session_key)
|
||||
if isinstance(queue, list):
|
||||
entries = queue
|
||||
elif queue:
|
||||
entries = [queue]
|
||||
else:
|
||||
return None
|
||||
for entry in entries:
|
||||
if isinstance(entry, dict) and entry.get("approval_id") == approval_id and entry.get(_GATEWAY_MIRROR_FLAG):
|
||||
run_id = str(entry.get("run_id") or "").strip()
|
||||
return run_id or None
|
||||
entry = _gateway_pending_mirror_locked(session_key, approval_id=approval_id)
|
||||
if entry:
|
||||
return str(entry.get("run_id") or "").strip() or None
|
||||
return None
|
||||
|
||||
|
||||
def _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."""
|
||||
del approval # mirror from the live gateway head under `_lock`, not from callback input
|
||||
with _lock:
|
||||
run_id = str(approval.get("run_id") or "").strip()
|
||||
if run_id:
|
||||
live_gateway_queue = _gateway_queues.get(session_key) or []
|
||||
if not live_gateway_queue:
|
||||
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[_GATEWAY_MIRROR_FLAG] = True
|
||||
if not _gateway_pending_mirror_locked(session_key, approval_id=mirror_entry["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_sse_notify_locked(session_key, head, total)
|
||||
publish_session_list_changed("attention_pending")
|
||||
|
||||
+28
-9
@@ -9715,7 +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_mirrored_pending_run_id,
|
||||
gateway_pending_mirror,
|
||||
retire_gateway_pending_mirror,
|
||||
reconcile_gateway_pending_mirror_locked,
|
||||
submit_gateway_pending_mirror,
|
||||
submit_pending,
|
||||
@@ -23914,29 +23915,47 @@ def _handle_approval_respond(handler, body):
|
||||
)
|
||||
from api.config import get_config as _get_config
|
||||
s = get_session(sid)
|
||||
_run_id = None
|
||||
_candidate_run_id = None
|
||||
if s is not None:
|
||||
active_sid = getattr(s, "active_stream_id", None)
|
||||
if active_sid:
|
||||
_run_id = _STREAM_RUN_IDS.get(active_sid)
|
||||
if not _run_id and approval_id:
|
||||
_run_id = _gateway_mirrored_pending_run_id(sid, approval_id)
|
||||
_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:
|
||||
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)
|
||||
for entry in entries
|
||||
)
|
||||
if local_match:
|
||||
_candidate_run_id = None
|
||||
if _run_id:
|
||||
if not approval_id:
|
||||
return bad(handler, "approval_id is required for gateway approvals")
|
||||
from api.runner_client import HttpRunnerClient, RunnerClientError
|
||||
_cfg = _get_config()
|
||||
_base = _gateway_base_url(_cfg)
|
||||
_key = _gateway_api_key()
|
||||
try:
|
||||
HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(_run_id, approval_id, choice)
|
||||
HttpRunnerClient(base_url=_base, api_key=_key).respond_approval(
|
||||
_run_id, matched_mirror["approval_id"], choice
|
||||
)
|
||||
except (RunnerClientError, ValueError) as exc:
|
||||
return j(handler, {"ok": False, "choice": choice, "relayed": True, "error": str(exc)}, status=502)
|
||||
# The outbound relay only resumes the remote run; the local mirror
|
||||
# still needs the same cleanup path so the parked entry, mirrored
|
||||
# card, and agent signal all settle here too.
|
||||
_resolve_approval_legacy(sid, approval_id, choice)
|
||||
cleanup_approval_id = matched_mirror["approval_id"]
|
||||
_resolve_approval_legacy(sid, cleanup_approval_id, choice)
|
||||
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:"):
|
||||
return j(handler, {"ok": False, "choice": choice, "relayed": False,
|
||||
"code": "gateway_run_unavailable",
|
||||
"error": _GATEWAY_APPROVAL_RELAY_UNAVAILABLE}, status=409)
|
||||
# Only a still-mirrored gateway approval with a missing run should 409;
|
||||
# stale or empty gateway clicks fall through to local resolution.
|
||||
if webui_gateway_chat_enabled(_get_config()) and _gateway_pending_approval_without_run_id(
|
||||
|
||||
@@ -6,6 +6,7 @@ import json
|
||||
import socket
|
||||
import threading
|
||||
import urllib.error
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
|
||||
@@ -301,6 +302,8 @@ def test_gateway_approval_event_translation():
|
||||
assert result["risk_level"] == "high"
|
||||
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"
|
||||
|
||||
downgraded = _gateway_runs_approval_event({
|
||||
"command": "rm -rf /tmp/x",
|
||||
@@ -412,6 +415,169 @@ def test_gateway_runs_api_streaming_parses_real_run_events():
|
||||
assert events[0][1]["approval_id"] == "appr-1"
|
||||
assert events[1] == ("reasoning", {"text": "thinking..."})
|
||||
assert events[2] == ("token", {"text": "Hello"})
|
||||
import api.route_approvals as approvals
|
||||
assert approvals.gateway_pending_mirror("sess1", run_id="run-abc") is None
|
||||
|
||||
|
||||
def test_empty_id_runs_approval_reaches_real_response_lifecycle():
|
||||
"""Runs API empty IDs are carried and resolved through the live mirror."""
|
||||
from api.gateway_chat import _run_gateway_runs_api_streaming
|
||||
import api.route_approvals as approvals
|
||||
import api.routes as routes
|
||||
|
||||
for choice in ("once", "deny"):
|
||||
sid = f"sess-real-{choice}"
|
||||
stream_id = f"stream-real-{choice}"
|
||||
events = []
|
||||
|
||||
class _JsonResponse:
|
||||
def read(self, _limit=None):
|
||||
return b'{"run_id":"run-real-' + choice.encode() + b'"}'
|
||||
def __enter__(self):
|
||||
return self
|
||||
def __exit__(self, *args):
|
||||
return None
|
||||
|
||||
class _SseResponse:
|
||||
def __iter__(self):
|
||||
return iter([
|
||||
b'data: {"event":"approval.request","command":"echo x",'
|
||||
b'"description":"approval","approval_id":""}\n',
|
||||
b'\n', b'data: [DONE]\n', b'\n'
|
||||
])
|
||||
def __enter__(self):
|
||||
return self
|
||||
def __exit__(self, *args):
|
||||
return None
|
||||
|
||||
def fake_urlopen(req, *, timeout=None):
|
||||
return _JsonResponse() if req.full_url.endswith("/v1/runs") else _SseResponse()
|
||||
|
||||
handler = MagicMock()
|
||||
handler.wfile = io.BytesIO()
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
|
||||
patch("api.routes.get_session", return_value=SimpleNamespace(active_stream_id=None)), \
|
||||
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", return_value={"ok": True}):
|
||||
_run_gateway_runs_api_streaming(
|
||||
sid, "hi", "test", "/tmp", stream_id, "http://gw:8642", "", [], {},
|
||||
put_gateway_event=lambda event, data: events.append((event, data)),
|
||||
cancel_event=threading.Event(),
|
||||
)
|
||||
approval_id = events[0][1]["approval_id"]
|
||||
assert approval_id == 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,
|
||||
})
|
||||
assert handler.send_response.call_args.args[0] == 200
|
||||
assert json.loads(handler.wfile.getvalue().decode())["ok"] is True
|
||||
assert approvals.gateway_pending_mirror(sid, approval_id=approval_id) is None
|
||||
approvals._pending.pop(sid, None)
|
||||
|
||||
|
||||
def test_gateway_mirror_match_is_session_scoped_and_retires_one_run():
|
||||
import api.route_approvals as approvals
|
||||
|
||||
sid = "sess-multi-run"
|
||||
other_sid = "sess-other-run"
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._pending.pop(other_sid, None)
|
||||
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")
|
||||
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 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
|
||||
assert approvals.gateway_pending_mirror(sid, run_id="run-b") is not None
|
||||
finally:
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._pending.pop(other_sid, 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
|
||||
|
||||
sid = "sess-stale-gateway"
|
||||
stream_id = "sid-stale-live-run"
|
||||
_STREAM_RUN_IDS[stream_id] = "run-live"
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals.submit_gateway_pending_mirror(sid, {
|
||||
"run_id": "run-live", "approval_id": "gwrun:run-live", "command": "echo 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": "gwrun:run-stale",
|
||||
})
|
||||
|
||||
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, run_id="run-live") is not None
|
||||
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
approvals._pending.pop(sid, None)
|
||||
|
||||
|
||||
def test_retained_gateway_card_does_not_relay_different_active_run():
|
||||
from api.gateway_chat import _STREAM_RUN_IDS
|
||||
import api.route_approvals as approvals
|
||||
|
||||
sid = "sess-retained-gateway"
|
||||
stream_id = "sid-retained-live-run"
|
||||
_STREAM_RUN_IDS[stream_id] = "run-live"
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals.submit_gateway_pending_mirror(sid, {
|
||||
"run_id": "run-retained", "approval_id": "gwrun:run-retained", "command": "echo retained"
|
||||
})
|
||||
approvals.submit_gateway_pending_mirror(sid, {
|
||||
"run_id": "run-live", "approval_id": "gwrun:run-live", "command": "echo 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": "gwrun:run-retained",
|
||||
})
|
||||
|
||||
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-retained") is not None
|
||||
assert approvals.gateway_pending_mirror(sid, approval_id="gwrun:run-live") is not None
|
||||
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
approvals._pending.pop(sid, None)
|
||||
|
||||
|
||||
def test_gateway_runs_api_streaming_preserves_multimodal_input():
|
||||
@@ -546,9 +712,13 @@ def test_gateway_runs_api_cancel_does_not_emit_empty_response():
|
||||
def test_gateway_approval_response_relay():
|
||||
"""_handle_approval_respond relays the real gateway approval body."""
|
||||
from api.gateway_chat import _STREAM_RUN_IDS
|
||||
import api.route_approvals as approvals
|
||||
|
||||
# Seed the mapping.
|
||||
_STREAM_RUN_IDS["sid-relay"] = "run abc/1"
|
||||
approvals.submit_gateway_pending_mirror("sess-relay", {
|
||||
"run_id": "run abc/1", "approval_id": "appr x/y", "command": "echo x"
|
||||
})
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.active_stream_id = "sid-relay"
|
||||
@@ -578,14 +748,89 @@ def test_gateway_approval_response_relay():
|
||||
|
||||
# Cleanup.
|
||||
_STREAM_RUN_IDS.pop("sid-relay", None)
|
||||
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."""
|
||||
from api.gateway_chat import _STREAM_RUN_IDS
|
||||
import api.route_approvals as approvals
|
||||
|
||||
_STREAM_RUN_IDS["sid-relay-empty-id"] = "run abc/1"
|
||||
approvals.submit_gateway_pending_mirror("sess-relay", {
|
||||
"run_id": "run abc/1", "approval_id": "", "command": "echo x"
|
||||
})
|
||||
|
||||
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.routes._resolve_approval_legacy", return_value=True) as cleanup:
|
||||
from api.routes import _handle_approval_respond
|
||||
_handle_approval_respond(
|
||||
handler,
|
||||
{"session_id": "sess-relay", "choice": "once", "approval_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")
|
||||
|
||||
_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
|
||||
from api.runner_client import RunnerClientError
|
||||
import api.route_approvals as approvals
|
||||
|
||||
_STREAM_RUN_IDS["sid-relay-fail"] = "run-abc"
|
||||
approvals.submit_gateway_pending_mirror("sess-relay", {
|
||||
"run_id": "run-abc", "approval_id": "appr-x", "command": "echo x"
|
||||
})
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.active_stream_id = "sid-relay-fail"
|
||||
@@ -609,13 +854,18 @@ def test_gateway_approval_response_relay_failure_returns_502():
|
||||
assert "relay failed" in payload["error"]
|
||||
|
||||
_STREAM_RUN_IDS.pop("sid-relay-fail", None)
|
||||
approvals._pending.pop("sess-relay", None)
|
||||
|
||||
|
||||
def test_gateway_approval_response_invalid_gateway_base_returns_502():
|
||||
"""Misconfigured gateway bases must not fall through to the local approval path."""
|
||||
from api.gateway_chat import _STREAM_RUN_IDS
|
||||
import api.route_approvals as approvals
|
||||
|
||||
_STREAM_RUN_IDS["sid-relay-invalid-base"] = "run-abc"
|
||||
approvals.submit_gateway_pending_mirror("sess-relay", {
|
||||
"run_id": "run-abc", "approval_id": "appr-x", "command": "echo x"
|
||||
})
|
||||
|
||||
mock_session = MagicMock()
|
||||
mock_session.active_stream_id = "sid-relay-invalid-base"
|
||||
@@ -638,6 +888,7 @@ def test_gateway_approval_response_invalid_gateway_base_returns_502():
|
||||
assert "runner base_url must be http(s)" in payload["error"]
|
||||
|
||||
_STREAM_RUN_IDS.pop("sid-relay-invalid-base", None)
|
||||
approvals._pending.pop("sess-relay", None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in new issue
Block a user