fix(gateway): keep retained no-run approvals visible
This commit is contained in:
1 parent
0a3a630856
commit
4a71e407b1
5 files changed
+106
-41
No files matched your search
+4
-6
@@ -347,6 +347,8 @@ def _gateway_runs_approval_event(payload: dict) -> dict | None:
|
||||
run_id = str(payload.get("run_id") or "").strip()
|
||||
raw_approval_id = str(payload.get("approval_id") or payload.get("id") or "").strip()
|
||||
approval_id = raw_approval_id
|
||||
if not approval_id:
|
||||
approval_id = uuid.uuid4().hex
|
||||
risk = str(payload.get("risk_level") or "high").strip()
|
||||
choices = payload.get("choices") if isinstance(payload.get("choices"), list) else []
|
||||
allow_permanent = payload.get("allow_permanent")
|
||||
@@ -703,14 +705,10 @@ def _clear_gateway_pending_state(session: Any, stream_id: str) -> None:
|
||||
def _cleanup_gateway_pending_mirror(session_id: str) -> None:
|
||||
try:
|
||||
from api.route_approvals import (
|
||||
_approval_sse_notify_locked,
|
||||
_lock as _approval_lock,
|
||||
reconcile_gateway_pending_mirror_locked,
|
||||
retire_gateway_pending_mirror,
|
||||
)
|
||||
|
||||
with _approval_lock:
|
||||
head, total, _ = reconcile_gateway_pending_mirror_locked(session_id)
|
||||
_approval_sse_notify_locked(session_id, head, total)
|
||||
retire_gateway_pending_mirror(session_id)
|
||||
except Exception:
|
||||
logger.debug("Failed to reconcile gateway pending mirror during teardown", exc_info=True)
|
||||
|
||||
|
||||
+16
-5
@@ -46,6 +46,7 @@ except ImportError:
|
||||
_approval_sse_subscribers: dict[str, list[queue.Queue]] = {}
|
||||
_GATEWAY_MIRROR_FLAG = "_gateway_mirror"
|
||||
_GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token"
|
||||
_GATEWAY_MIRROR_RETAINED = "_gateway_mirror_retained"
|
||||
_GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_mirror_token"
|
||||
_GATEWAY_AGENT_IDENTITY_V1 = "_gateway_agent_identity_v1"
|
||||
|
||||
@@ -219,6 +220,10 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
|
||||
changed = True
|
||||
continue
|
||||
|
||||
if entry.get(_GATEWAY_MIRROR_RETAINED):
|
||||
rebuilt.append(entry)
|
||||
continue
|
||||
|
||||
if matches_live_head and not live_mirror_present:
|
||||
if entry_token != live_token:
|
||||
entry[_GATEWAY_MIRROR_TOKEN] = live_token
|
||||
@@ -334,7 +339,11 @@ def retire_gateway_pending_mirror(session_key: str, approval_id: str = "", run_i
|
||||
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 else [
|
||||
entry for entry in entries
|
||||
if _is_gateway_mirror_entry(entry)
|
||||
and not str(entry.get("run_id") or "").strip()
|
||||
]
|
||||
if normalized_run_id:
|
||||
retained_gateway_queue = []
|
||||
for entry in gateway_queue:
|
||||
@@ -344,10 +353,11 @@ 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
|
||||
head, total, changed = reconcile_gateway_pending_mirror_locked(session_key)
|
||||
_approval_sse_notify_locked(session_key, head, total)
|
||||
if changed:
|
||||
publish_session_list_changed("attention_resolved")
|
||||
return changed
|
||||
for match in retired:
|
||||
entries.remove(match)
|
||||
if normalized_run_id and not approval_id:
|
||||
@@ -546,6 +556,7 @@ def resolve_gateway_pending_local_no_run_mirror(
|
||||
target = gateway_queue.pop(index)
|
||||
break
|
||||
if target is None:
|
||||
matched_mirror[_GATEWAY_MIRROR_RETAINED] = True
|
||||
return True, 0, entries[0] if entries else None, len(entries)
|
||||
|
||||
if gateway_queue:
|
||||
|
||||
+2
-6
@@ -7808,14 +7808,10 @@ def _run_agent_streaming(
|
||||
try:
|
||||
from api.route_approvals import (
|
||||
submit_gateway_pending_mirror as _submit_pending_for_polling,
|
||||
reconcile_gateway_pending_mirror_locked as _reconcile_gateway_pending_mirror_locked,
|
||||
_approval_sse_notify_locked as _approval_sse_notify_locked,
|
||||
_lock as _approval_lock,
|
||||
retire_gateway_pending_mirror as _retire_gateway_pending_mirror,
|
||||
)
|
||||
def _cleanup_gateway_pending_mirror():
|
||||
with _approval_lock:
|
||||
head, total, _changed = _reconcile_gateway_pending_mirror_locked(session_id)
|
||||
_approval_sse_notify_locked(session_id, head, total)
|
||||
_retire_gateway_pending_mirror(session_id)
|
||||
except ImportError:
|
||||
_submit_pending_for_polling = None
|
||||
_cleanup_gateway_pending_mirror = None
|
||||
|
||||
@@ -638,11 +638,11 @@ class TestApprovalHTTPEndpoints:
|
||||
r._gateway_queues.pop(sid, None)
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
|
||||
def test_gateway_no_run_mirror_stays_visible_when_exact_local_producer_is_gone(self, monkeypatch):
|
||||
"""A tokenized no-run mirror stays live when its parked producer already vanished."""
|
||||
def test_gateway_no_run_mirror_survives_pending_and_attention_reads_until_explicit_teardown(self, monkeypatch):
|
||||
"""A missing-producer 409 stays visible until a real teardown retires it."""
|
||||
from api import routes as r
|
||||
from api import route_approvals as ra
|
||||
from api.gateway_chat import _STREAM_RUN_IDS
|
||||
from api.gateway_chat import _STREAM_RUN_IDS, _cleanup_gateway_pending_mirror
|
||||
|
||||
sid = f"http-gateway-race-{uuid.uuid4().hex[:8]}"
|
||||
stream_id = f"stream-race-{uuid.uuid4().hex[:8]}"
|
||||
@@ -680,13 +680,76 @@ class TestApprovalHTTPEndpoints:
|
||||
"error": r._GATEWAY_APPROVAL_RELAY_UNAVAILABLE,
|
||||
}
|
||||
assert not entry.event.is_set()
|
||||
with _lock:
|
||||
pending_queue = r._pending.get(sid)
|
||||
assert isinstance(pending_queue, list)
|
||||
assert len(pending_queue) == 1
|
||||
assert pending_queue[0]["approval_id"] == approval_id
|
||||
pending = r._handle_approval_pending(
|
||||
object(), urllib.parse.urlparse(
|
||||
f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
|
||||
)
|
||||
)
|
||||
assert pending["pending"]["approval_id"] == approval_id
|
||||
assert pending["pending_count"] == 1
|
||||
assert r._session_attention_summary(sid) == {
|
||||
"kind": "approval", "count": 1, "severity": "critical"
|
||||
}
|
||||
_cleanup_gateway_pending_mirror(sid)
|
||||
pending = r._handle_approval_pending(
|
||||
object(), urllib.parse.urlparse(
|
||||
f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
|
||||
)
|
||||
)
|
||||
assert pending == {"pending": None, "pending_count": 0}
|
||||
assert r._session_attention_summary(sid) is None
|
||||
finally:
|
||||
with _lock:
|
||||
r._pending.pop(sid, None)
|
||||
r._gateway_queues.pop(sid, None)
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
|
||||
def test_gateway_no_run_non_head_response_resolves_only_exact_producer(self, monkeypatch):
|
||||
"""An exact non-head response wakes only its matching producer."""
|
||||
from api import routes as r
|
||||
from api import route_approvals as ra
|
||||
|
||||
sid = f"http-gateway-non-head-{uuid.uuid4().hex[:8]}"
|
||||
approval_a = {"command": "rm -rf /tmp/a", "description": "a"}
|
||||
approval_b = {"command": "rm -rf /tmp/b", "description": "b"}
|
||||
entry_a = _ApprovalEntry(approval_a)
|
||||
entry_b = _ApprovalEntry(approval_b)
|
||||
captured = {}
|
||||
|
||||
def fake_j(handler, data, status=200, extra_headers=None):
|
||||
captured["payload"] = data
|
||||
captured["status"] = status
|
||||
return data
|
||||
|
||||
monkeypatch.setattr(r, "j", fake_j)
|
||||
monkeypatch.setattr(r, "get_session", lambda _sid: SimpleNamespace(active_stream_id=None))
|
||||
monkeypatch.setenv("HERMES_WEBUI_CHAT_BACKEND", "gateway")
|
||||
with _lock:
|
||||
r._pending.pop(sid, None)
|
||||
r._gateway_queues[sid] = [entry_a, entry_b]
|
||||
try:
|
||||
ra.submit_gateway_pending_mirror(sid, approval_a)
|
||||
ra.submit_gateway_pending_mirror(sid, approval_b)
|
||||
with _lock:
|
||||
approval_b_id = next(
|
||||
item["approval_id"] for item in r._pending[sid]
|
||||
if item["command"] == approval_b["command"]
|
||||
)
|
||||
r._handle_approval_respond(
|
||||
object(), {"session_id": sid, "choice": "once", "approval_id": approval_b_id}
|
||||
)
|
||||
assert captured["status"] == 200
|
||||
assert entry_b.event.is_set()
|
||||
assert entry_b.result == "once"
|
||||
assert not entry_a.event.is_set()
|
||||
pending = r._handle_approval_pending(
|
||||
object(), urllib.parse.urlparse(
|
||||
f"/api/approval/pending?session_id={urllib.parse.quote(sid)}"
|
||||
)
|
||||
)
|
||||
assert pending["pending"]["command"] == approval_a["command"]
|
||||
assert pending["pending_count"] == 1
|
||||
finally:
|
||||
with _lock:
|
||||
r._pending.pop(sid, None)
|
||||
r._gateway_queues.pop(sid, None)
|
||||
@@ -432,13 +432,19 @@ def test_gateway_runs_api_streaming_parses_real_run_events():
|
||||
def test_live_empty_ingress_id_stays_fifo_under_capability_v1():
|
||||
"""A normalized fallback browser ID must not become authoritative Agent identity."""
|
||||
from api.config import STREAM_PARTIAL_TEXT, STREAM_REASONING_TEXT
|
||||
from api.gateway_chat import _STREAM_RUN_IDS, _run_gateway_runs_api_streaming
|
||||
from api.gateway_chat import _STREAM_RUN_IDS, _gateway_runs_approval_event, _run_gateway_runs_api_streaming
|
||||
from api import routes
|
||||
import api.route_approvals as approvals
|
||||
|
||||
sid = "sid-live-empty-ingress"
|
||||
stream_id = "stream-live-empty-ingress"
|
||||
events = []
|
||||
normalized = _gateway_runs_approval_event({
|
||||
"command": "rm -rf /tmp/x", "description": "Dangerous",
|
||||
"run_id": "run-empty-ingress", "approval_id": "", "id": "",
|
||||
})
|
||||
assert normalized["approval_id"]
|
||||
assert normalized["_gateway_raw_approval_id_present"] is False
|
||||
STREAM_PARTIAL_TEXT[stream_id] = ""
|
||||
STREAM_REASONING_TEXT[stream_id] = ""
|
||||
|
||||
@@ -472,22 +478,10 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1():
|
||||
return _JsonResponse()
|
||||
return _SseResponse()
|
||||
|
||||
def fake_mapped_approval(payload):
|
||||
assert payload["approval_id"] == ""
|
||||
assert payload["id"] == ""
|
||||
return {
|
||||
"command": "rm -rf /tmp/x",
|
||||
"description": "Dangerous",
|
||||
"run_id": "run-empty-ingress",
|
||||
"approval_id": "gwrun:normalized-fallback",
|
||||
"_gateway_raw_approval_id_present": False,
|
||||
}
|
||||
|
||||
handler = MagicMock()
|
||||
handler.wfile = io.BytesIO()
|
||||
try:
|
||||
with patch("urllib.request.urlopen", side_effect=fake_urlopen), \
|
||||
patch("api.gateway_chat._gateway_runs_approval_event", side_effect=fake_mapped_approval), \
|
||||
patch("api.config.gateway_supports_approval_identity_v1", return_value=True):
|
||||
_run_gateway_runs_api_streaming(
|
||||
session_id=sid, msg_text="hi", model="test-model", workspace="/tmp",
|
||||
@@ -496,11 +490,12 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1():
|
||||
cancel_event=threading.Event(),
|
||||
)
|
||||
mirror = approvals.gateway_pending_mirror(sid, run_id="run-empty-ingress")
|
||||
browser_id = mirror["approval_id"]
|
||||
with patch("api.routes.get_session", return_value=SimpleNamespace(active_stream_id=stream_id)), \
|
||||
patch("api.config.gateway_supports_approval_identity_v1", return_value=True), \
|
||||
patch("api.runner_client.HttpRunnerClient.respond_approval") as respond:
|
||||
routes._handle_approval_respond(handler, {
|
||||
"session_id": sid, "choice": "once", "approval_id": "gwrun:normalized-fallback",
|
||||
"session_id": sid, "choice": "once", "approval_id": browser_id,
|
||||
})
|
||||
respond.assert_called_once_with("run-empty-ingress", "", "once")
|
||||
finally:
|
||||
@@ -509,9 +504,11 @@ def test_live_empty_ingress_id_stays_fifo_under_capability_v1():
|
||||
_STREAM_RUN_IDS.pop(stream_id, None)
|
||||
|
||||
assert events[0][0] == "approval"
|
||||
assert events[0][1]["approval_id"] == "gwrun:normalized-fallback"
|
||||
assert events[0][1]["approval_id"]
|
||||
assert events[0][1]["_gateway_raw_approval_id_present"] is False
|
||||
assert events[0][1]["_gateway_agent_identity_v1"] is False
|
||||
assert mirror["approval_id"] == "gwrun:normalized-fallback"
|
||||
assert mirror["approval_id"] == events[0][1]["approval_id"]
|
||||
assert mirror["_gateway_raw_approval_id_present"] is False
|
||||
assert mirror["_gateway_agent_identity_v1"] is False
|
||||
approvals._pending.pop(sid, None)
|
||||
approvals._gateway_queues.pop(sid, None)
|
||||
|
||||
Reference in new issue
Block a user