fix(#6008): publish the authoritative approval head

This commit is contained in:
Rod Boev committed 2026-07-22 12:31:12 -04:00
1 parent 6460c48ce6
commit 5826d76da0
10 files changed
+301 -84

No files matched your search

+11 -3
View File
@@ -507,8 +507,8 @@ def _run_gateway_runs_api_streaming(
if not approval_data.get("approval_id"): if not approval_data.get("approval_id"):
approval_data["approval_id"] = f"gwrun:{run_id}:{uuid.uuid4().hex}" approval_data["approval_id"] = f"gwrun:{run_id}:{uuid.uuid4().hex}"
from api.route_approvals import submit_gateway_pending_mirror from api.route_approvals import submit_gateway_pending_mirror
submit_gateway_pending_mirror(session_id, approval_data) head, total = submit_gateway_pending_mirror(session_id, approval_data)
put_gateway_event("approval", approval_data) put_gateway_event("approval", {**(head or approval_data), "pending_count": total})
sse_event = "message" sse_event = "message"
continue continue
if payload_event in {"tool.started", "tool.completed", "reasoning.available"}: if payload_event in {"tool.started", "tool.completed", "reasoning.available"}:
@@ -989,7 +989,8 @@ def _run_gateway_chat_streaming(
_STREAM_RUN_IDS[stream_id] = _approval_run_id _STREAM_RUN_IDS[stream_id] = _approval_run_id
try: try:
from api.route_approvals import submit_gateway_pending_mirror from api.route_approvals import submit_gateway_pending_mirror
submit_gateway_pending_mirror(session_id, approval_data) head, total = submit_gateway_pending_mirror(session_id, approval_data)
approval_data = {**(head or approval_data), "pending_count": total}
except Exception: except Exception:
logger.debug("submit_gateway_pending_mirror failed", exc_info=True) logger.debug("submit_gateway_pending_mirror failed", exc_info=True)
put_gateway_event("approval", approval_data) put_gateway_event("approval", approval_data)
@@ -1259,6 +1260,13 @@ def _run_gateway_chat_streaming(
"hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.", "hint": "Check HERMES_WEBUI_GATEWAY_BASE_URL and Gateway API server health.",
}) })
finally: finally:
mapped_run_id = str(_STREAM_RUN_IDS.get(stream_id) or "").strip()
if mapped_run_id:
try:
from api.route_approvals import retire_gateway_pending_mirror
retire_gateway_pending_mirror(session_id, run_id=mapped_run_id)
except Exception:
logger.debug("Failed to retire gateway pending mirrors during teardown", exc_info=True)
if s is not None: if s is not None:
try: try:
with _get_session_agent_lock(session_id): with _get_session_agent_lock(session_id):
+107 -35
View File
@@ -47,7 +47,6 @@ _approval_sse_subscribers: dict[str, list[queue.Queue]] = {}
_GATEWAY_MIRROR_FLAG = "_gateway_mirror" _GATEWAY_MIRROR_FLAG = "_gateway_mirror"
_GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token" _GATEWAY_MIRROR_TOKEN = "_gateway_mirror_token"
_GATEWAY_ENTRY_DATA_TOKEN_KEY = "_webui_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: def _approval_sse_subscribe(session_id: str) -> queue.Queue:
@@ -116,8 +115,6 @@ def _gateway_mirror_entry_token(entry) -> str | None:
""" """
data = getattr(entry, "data", None) data = getattr(entry, "data", None)
if isinstance(data, dict): if isinstance(data, dict):
if data.get(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY):
return None
token = data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) token = data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY)
if not token: if not token:
token = uuid.uuid4().hex token = uuid.uuid4().hex
@@ -150,8 +147,31 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
live_head_entry = live_gateway_queue[0] if live_gateway_queue else None live_head_entry = live_gateway_queue[0] if live_gateway_queue else None
live_head_data = getattr(live_head_entry, "data", None) or {} live_head_data = getattr(live_head_entry, "data", None) or {}
live_token = _gateway_mirror_entry_token(live_head_entry) if live_head_entry and live_head_data else None has_no_run_mirror = any(
_is_gateway_mirror_entry(entry)
and not str(entry.get("run_id") or "").strip()
for entry in queue_list
)
live_run_id = str(live_head_data.get("run_id") or "").strip() live_run_id = str(live_head_data.get("run_id") or "").strip()
live_has_token = bool(live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY))
live_local_tokens: set[str] = set()
for live_entry in live_gateway_queue:
live_data = getattr(live_entry, "data", None) or {}
if str(live_data.get("run_id") or "").strip():
continue
live_entry_token = str(live_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY) or "").strip()
if live_entry_token or not has_no_run_mirror:
live_entry_token = _gateway_mirror_entry_token(live_entry) or ""
if live_entry_token:
live_local_tokens.add(live_entry_token)
if not str(live_data.get("approval_id") or "").strip():
live_data["approval_id"] = f"gwlocal:{live_entry_token}"
live_token = (
_gateway_mirror_entry_token(live_head_entry)
if live_head_entry and live_head_data
and (live_run_id or live_has_token or not has_no_run_mirror)
else None
)
if live_token and live_run_id and not str(live_head_data.get("approval_id") or "").strip(): 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_head_data["approval_id"] = f"gwrun:{live_run_id}:{live_token}"
live_approval_id = str(live_head_data.get("approval_id") or "").strip() live_approval_id = str(live_head_data.get("approval_id") or "").strip()
@@ -206,6 +226,10 @@ def reconcile_gateway_pending_mirror_locked(session_key: str) -> tuple[dict | No
live_mirror_present = True live_mirror_present = True
continue continue
if entry_token and entry_token in live_local_tokens:
rebuilt.append(entry)
continue
if not live_token: if not live_token:
if entry_token: if entry_token:
changed = True changed = True
@@ -348,34 +372,56 @@ def _gateway_mirrored_pending_run_id(session_key: str, approval_id: str) -> str
return None return None
def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None: def submit_gateway_pending_mirror(session_key: str, approval: dict) -> tuple[dict | None, int]:
"""Mirror the live gateway head into WebUI polling state under a typed tag.""" """Mirror the live gateway head into WebUI polling state under a typed tag."""
with _lock: with _lock:
run_id = str(approval.get("run_id") or "").strip() run_id = str(approval.get("run_id") or "").strip()
approval_id = str(approval.get("approval_id") or "").strip() approval_id = str(approval.get("approval_id") or "").strip()
live_gateway_queue = _gateway_queues.get(session_key) or [] live_gateway_queue = _gateway_queues.get(session_key) or []
live_head_entry = live_gateway_queue[0] if live_gateway_queue else None exact_local_entry = next(
live_head_data = getattr(live_head_entry, "data", None) or {} (
matching_no_run_live_head = ( entry for entry in live_gateway_queue
not run_id if getattr(entry, "data", None) is approval
and live_head_data ),
and str(live_head_data.get("command") or "") == str(approval.get("command") or "") None,
and str(live_head_data.get("description") or "") == str(approval.get("description") or "") ) if not run_id else None
and str(live_head_data.get("pattern_key") or "") == str(approval.get("pattern_key") or "") if exact_local_entry is None and not run_id and approval_id:
and tuple(str(k) for k in (live_head_data.get("pattern_keys") or [])) exact_local_entry = next(
== tuple(str(k) for k in (approval.get("pattern_keys") or [])) (
) entry for entry in live_gateway_queue
if matching_no_run_live_head and not live_head_data.get(_GATEWAY_ENTRY_DATA_TOKEN_KEY): if str(((getattr(entry, "data", None) or {}).get("approval_id") or "")).strip() == approval_id
live_head_data.pop(_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY, None) ),
live_head_data[_GATEWAY_ENTRY_DATA_TOKEN_KEY] = uuid.uuid4().hex None,
elif not run_id and live_head_data: )
live_head_data[_GATEWAY_ENTRY_DATA_LOCAL_HEAD_KEY] = True if exact_local_entry is not None:
if not run_id and not approval_id: mirror_entries = _normalize_pending_queue_locked(session_key)
approval_id = uuid.uuid4().hex entries_to_mirror = [live_gateway_queue[0]] if live_gateway_queue else []
approval["approval_id"] = approval_id if exact_local_entry not in entries_to_mirror:
if matching_no_run_live_head and approval_id and not str(live_head_data.get("approval_id") or "").strip(): entries_to_mirror.append(exact_local_entry)
live_head_data["approval_id"] = approval_id for entry in entries_to_mirror:
local_data = entry.data
token = _gateway_mirror_entry_token(entry)
entry_approval_id = str(local_data.get("approval_id") or "").strip()
if entry is exact_local_entry:
entry_approval_id = approval_id or entry_approval_id or f"gwlocal:{token}"
approval_id = entry_approval_id
approval["approval_id"] = entry_approval_id
elif not entry_approval_id:
entry_approval_id = f"gwlocal:{token}"
local_data["approval_id"] = entry_approval_id
if not any(
_is_gateway_mirror_entry(mirror)
and str(mirror.get(_GATEWAY_MIRROR_TOKEN) or "") == token
for mirror in mirror_entries
):
mirror_entry = dict(local_data)
mirror_entry["approval_id"] = entry_approval_id
mirror_entry[_GATEWAY_MIRROR_FLAG] = True
mirror_entry[_GATEWAY_MIRROR_TOKEN] = token
mirror_entries.append(mirror_entry)
if run_id: if run_id:
live_head_entry = live_gateway_queue[0] if live_gateway_queue else None
live_head_data = getattr(live_head_entry, "data", None) or {}
live_head_run_id = str(live_head_data.get("run_id") or "").strip() live_head_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_head_approval_id = str(live_head_data.get("approval_id") or "").strip()
live_token = ( live_token = (
@@ -410,15 +456,10 @@ def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None:
mirror_entry[_GATEWAY_MIRROR_FLAG] = True mirror_entry[_GATEWAY_MIRROR_FLAG] = True
if not _gateway_pending_mirror_locked(session_key, approval_id=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) _normalize_pending_queue_locked(session_key).append(mirror_entry)
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key) elif not exact_local_entry:
approval_id = str(approval.get("approval_id") or "").strip() if not approval_id:
if run_id and approval_id: approval_id = uuid.uuid4().hex
mirror_entry = _gateway_pending_mirror_locked( approval["approval_id"] = approval_id
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()
elif approval_id:
queue = _pending.get(session_key) queue = _pending.get(session_key)
entries = queue if isinstance(queue, list) else [queue] if queue else [] entries = queue if isinstance(queue, list) else [queue] if queue else []
no_run_mirror = next( no_run_mirror = next(
@@ -437,8 +478,39 @@ def submit_gateway_pending_mirror(session_key: str, approval: dict) -> None:
mirror_entry["approval_id"] = approval_id mirror_entry["approval_id"] = approval_id
mirror_entry[_GATEWAY_MIRROR_FLAG] = True mirror_entry[_GATEWAY_MIRROR_FLAG] = True
_normalize_pending_queue_locked(session_key).append(mirror_entry) _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) _approval_sse_notify_locked(session_key, head, total)
publish_session_list_changed("attention_pending") publish_session_list_changed("attention_pending")
return (dict(head) if head else None), total
def resolve_gateway_pending_local(
session_key: str, approval_id: str, choice: str, reason: str | None = None
) -> tuple[int, dict | None, int]:
"""Resolve the exact parked local entry bound to an approval mirror."""
target = None
with _lock:
approval_id = str(approval_id or "").strip()
gateway_queue = _gateway_queues.get(session_key) or []
for index, entry in enumerate(gateway_queue):
data = getattr(entry, "data", None) or {}
if str(data.get("approval_id") or "").strip() == approval_id:
target = gateway_queue.pop(index)
break
if gateway_queue:
_gateway_queues[session_key] = gateway_queue
else:
_gateway_queues.pop(session_key, None)
head, total, _changed = reconcile_gateway_pending_mirror_locked(session_key)
_approval_sse_notify_locked(session_key, head, total)
if target is None:
return 0, head, total
target.result = choice
if reason:
target.reason = reason
target.event.set()
publish_session_list_changed("attention_resolved")
return 1, head, total
def submit_pending(session_key: str, approval: dict) -> None: def submit_pending(session_key: str, approval: dict) -> None:
+55 -20
View File
@@ -9720,6 +9720,7 @@ from api.route_approvals import ( # noqa: F401 — re-exports for backward comp
gateway_pending_mirror, gateway_pending_mirror,
retire_gateway_pending_mirror, retire_gateway_pending_mirror,
reconcile_gateway_pending_mirror_locked, reconcile_gateway_pending_mirror_locked,
resolve_gateway_pending_local,
submit_gateway_pending_mirror, submit_gateway_pending_mirror,
submit_pending, submit_pending,
) )
@@ -9746,12 +9747,40 @@ except ImportError:
def _session_attention_summary(session_id: str) -> dict | None: def _session_attention_summary(session_id: str) -> dict | None:
"""Return sidebar attention metadata for pending approval/clarify work.""" """Return sidebar attention metadata for pending approval/clarify work."""
active_run_id = ""
try:
from api.gateway_chat import _STREAM_RUN_IDS
session = get_session(session_id)
active_stream_id = getattr(session, "active_stream_id", None) if session is not None else None
if active_stream_id:
active_run_id = str(_STREAM_RUN_IDS.get(active_stream_id) or "").strip()
except Exception:
active_run_id = ""
approval_count = 0 approval_count = 0
with _lock: with _lock:
reconcile_gateway_pending_mirror_locked(session_id) reconcile_gateway_pending_mirror_locked(session_id)
queue_list = _pending.get(session_id) queue_list = _pending.get(session_id)
if isinstance(queue_list, list): if isinstance(queue_list, list):
approval_count = len(queue_list) gateway_queue = _gateway_queues.get(session_id) or []
live_gateway_run_ids = {
str((getattr(entry, "data", None) or {}).get("run_id") or "").strip()
for entry in gateway_queue
if str((getattr(entry, "data", None) or {}).get("run_id") or "").strip()
}
approval_count = sum(
1
for entry in queue_list
if not (
isinstance(entry, dict)
and entry.get(_GATEWAY_MIRROR_FLAG)
and str(entry.get("run_id") or "").strip()
and not str(entry.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
and str(entry.get("run_id") or "").strip() not in live_gateway_run_ids
and str(entry.get("run_id") or "").strip() != active_run_id
)
)
elif queue_list: elif queue_list:
approval_count = 1 approval_count = 1
if approval_count > 0: if approval_count > 0:
@@ -23802,6 +23831,7 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st
pending = None pending = None
found_target = False found_target = False
gateway_keys = [] gateway_keys = []
local_gateway_approval_id = ""
gateway_head_matches_target = False gateway_head_matches_target = False
with _lock: with _lock:
reconcile_gateway_pending_mirror_locked(sid) reconcile_gateway_pending_mirror_locked(sid)
@@ -23874,20 +23904,28 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st
gw_data = getattr(gw_entry, "data", None) or {} gw_data = getattr(gw_entry, "data", None) or {}
gw_approval_id = str(gw_data.get("approval_id") or "").strip() gw_approval_id = str(gw_data.get("approval_id") or "").strip()
gw_run_id = str(gw_data.get("run_id") or "").strip() gw_run_id = str(gw_data.get("run_id") or "").strip()
if run_id: gateway_head_matches_target = bool(
gateway_head_matches_target = gw_approval_id == approval_id and gw_run_id == run_id run_id and gw_approval_id == approval_id and gw_run_id == run_id
else: )
gateway_head_matches_target = gw_approval_id == approval_id if gw_approval_id == approval_id and (not run_id or gw_run_id == run_id):
local_gateway_approval_id = approval_id
elif not run_id and found_target and pending:
pending_token = str(pending.get(_GATEWAY_MIRROR_TOKEN) or "").strip()
gateway_token = str(gw_data.get("_webui_mirror_token") or "").strip()
if pending_token and pending_token == gateway_token:
gw_data["approval_id"] = approval_id
local_gateway_approval_id = approval_id
# Notify SSE subscribers of the new head (or empty state) so the UI # Notify SSE subscribers of the new head (or empty state) so the UI
# surfaces any trailing approvals that were queued behind this one # surfaces any trailing approvals that were queued behind this one
# without waiting for the next submit_pending. Without this, a parallel # without waiting for the next submit_pending. Without this, a parallel
# tool-call scenario (#527) would leave the second approval invisible # tool-call scenario (#527) would leave the second approval invisible
# in the SSE path until the next event ever fired (the agent thread # in the SSE path until the next event ever fired (the agent thread
# would be parked indefinitely from the user's perspective). # would be parked indefinitely from the user's perspective).
if isinstance(_pending.get(sid), list) and _pending[sid]: if not local_gateway_approval_id:
_approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid])) if isinstance(_pending.get(sid), list) and _pending[sid]:
else: _approval_sse_notify_locked(sid, _pending[sid][0], len(_pending[sid]))
_approval_sse_notify_locked(sid, None, 0) else:
_approval_sse_notify_locked(sid, None, 0)
# Collect keys from both _pending and _gateway_queues # Collect keys from both _pending and _gateway_queues
keys_from_pending = pending.get("pattern_keys") or [pending.get("pattern_key", "")] if pending else [] keys_from_pending = pending.get("pattern_keys") or [pending.get("pattern_key", "")] if pending else []
@@ -23909,21 +23947,18 @@ def _resolve_approval_legacy(sid: str, approval_id: str, choice: str, run_id: st
# This is the primary signal when streaming is active — the agent # This is the primary signal when streaming is active — the agent
# thread is parked in entry.event.wait() and needs to be woken up. # thread is parked in entry.event.wait() and needs to be woken up.
gateway_resolved = 0 gateway_resolved = 0
should_resolve_gateway = ( local_gateway_resolved = 0
not approval_id if approval_id and found_target and not run_id and local_gateway_approval_id:
or ( local_gateway_resolved, _head, _total = resolve_gateway_pending_local(
found_target sid, local_gateway_approval_id, choice
and (
not run_id
or gateway_head_matches_target
)
) )
) elif approval_id and found_target and run_id and gateway_head_matches_target:
if should_resolve_gateway: gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0
elif not approval_id:
gateway_resolved = resolve_gateway_approval(sid, choice, resolve_all=False) or 0 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 # Keep the historical no-id response path truthy for old clients/tests while
# making stale explicit ids bounded as not-active for Slice 3b. # making stale explicit ids bounded as not-active for Slice 3b.
resolved = bool(pending) or bool(gateway_resolved) or not bool(approval_id) resolved = bool(pending) or bool(gateway_resolved) or bool(local_gateway_resolved) or not bool(approval_id)
if resolved: if resolved:
publish_session_list_changed("attention_resolved") publish_session_list_changed("attention_resolved")
return resolved return resolved
+5 -15
View File
@@ -7826,7 +7826,8 @@ def _run_agent_streaming(
def _approval_notify_cb(approval_data): def _approval_notify_cb(approval_data):
if _submit_pending_for_polling is not None: if _submit_pending_for_polling is not None:
try: try:
_submit_pending_for_polling(session_id, approval_data) head, total = _submit_pending_for_polling(session_id, approval_data)
approval_data = {**(head or approval_data), "pending_count": total}
except Exception: except Exception:
logger.warning("Failed to mirror approval into WebUI polling state", exc_info=True) logger.warning("Failed to mirror approval into WebUI polling state", exc_info=True)
put('approval', approval_data) put('approval', approval_data)
@@ -8241,20 +8242,9 @@ def _run_agent_streaming(
if _has_blocking_approval(session_id): if _has_blocking_approval(session_id):
p = None p = None
with _approval_lock: with _approval_lock:
_reconcile_gateway_pending_mirror_locked(session_id) p, pending_count, _changed = _reconcile_gateway_pending_mirror_locked(session_id)
queue = _approval_pending.get(session_id) if p:
if isinstance(queue, list): p = {**p, "pending_count": pending_count}
p = dict(queue[0]) if queue else None
elif queue:
p = dict(queue)
if p is None:
gw_queue = _approval_gateway_queues.get(session_id) or []
if gw_queue:
raw = getattr(gw_queue[0], 'data', None) or {}
if raw:
p = dict(raw)
else:
logger.warning("Gateway queue entry for %s has no .data attribute", session_id)
if p: if p:
put('approval', p) put('approval', p)
except ImportError: except ImportError:
+1 -1
View File
@@ -5589,7 +5589,7 @@ function attachLiveStream(activeSid, streamId, uploaded=[], options={}){
source.addEventListener('approval',e=>{ source.addEventListener('approval',e=>{
const d=JSON.parse(e.data); const d=JSON.parse(e.data);
_applyToAnchor('approval',d,e); _applyToAnchor('approval',d,e);
showApprovalForSession(activeSid, d, 1); showApprovalForSession(activeSid, d, d.pending_count || 1);
playAttentionSound(_attentionSoundKey(activeSid,'approval',1)); playAttentionSound(_attentionSoundKey(activeSid,'approval',1));
sendBrowserNotification('Approval required',d.description||'Tool approval needed',{sid:activeSid}); sendBrowserNotification('Approval required',d.description||'Tool approval needed',{sid:activeSid});
}); });
+35
View File
@@ -7,6 +7,8 @@ approval_id so /api/approval/respond can target a specific entry.
import json import json
import pathlib import pathlib
import re import re
import shutil
import subprocess
import sys import sys
REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve() REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve()
@@ -108,6 +110,39 @@ def test_polling_passes_count_to_show():
"Poll loop must pass data.pending_count through showApprovalForSession" "Poll loop must pass data.pending_count through showApprovalForSession"
def test_chat_stream_approval_listener_renders_received_count():
"""A chat-stream approval event with count two renders the head and count."""
node = shutil.which("node")
if node is None:
import pytest
pytest.skip("node is required for the approval listener harness")
start = MESSAGES_JS.index("source.addEventListener('approval',e=>{")
end_marker = "\n });"
end = MESSAGES_JS.index(end_marker, start) + len(end_marker)
listener = MESSAGES_JS[start:end]
script = f"""
const rendered = [];
let activeSid = 'sid-browser';
const source = {{ listeners: {{}}, addEventListener(name, cb) {{ this.listeners[name] = cb; }} }};
function _applyToAnchor() {{}}
function showApprovalForSession(sid, data, count) {{ rendered.push({{sid, data, count}}); }}
function playAttentionSound() {{}}
function _attentionSoundKey() {{ return 'approval'; }}
function sendBrowserNotification() {{}}
{listener}
source.listeners.approval({{ data: JSON.stringify({{command: 'head', approval_id: 'a', pending_count: 2}}) }});
process.stdout.write(JSON.stringify(rendered));
"""
result = subprocess.run([node, "-e", script], capture_output=True, text=True, check=False)
assert result.returncode == 0, result.stderr
rendered = json.loads(result.stdout)
assert rendered == [{
"sid": "sid-browser",
"data": {"command": "head", "approval_id": "a", "pending_count": 2},
"count": 2,
}]
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# HTML: counter element present # HTML: counter element present
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+3 -1
View File
@@ -113,8 +113,10 @@ class TestSSEStaticAnalysis:
cb_start = streaming_src.find("def _approval_notify_cb(approval_data):") cb_start = streaming_src.find("def _approval_notify_cb(approval_data):")
cb_end = streaming_src.find("_reg_notify(session_id, _approval_notify_cb)", cb_start) cb_end = streaming_src.find("_reg_notify(session_id, _approval_notify_cb)", cb_start)
cb_body = streaming_src[cb_start:cb_end] cb_body = streaming_src[cb_start:cb_end]
assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \ assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \
"_approval_notify_cb must mirror approval data into polling state before SSE" "_approval_notify_cb must mirror approval data into polling state before SSE"
assert '"pending_count": total' in cb_body, \
"_approval_notify_cb must emit the reconciled pending count"
assert "put('approval', approval_data)" in cb_body, \ assert "put('approval', approval_data)" in cb_body, \
"_approval_notify_cb must still emit the approval SSE event" "_approval_notify_cb must still emit the approval SSE event"
+47 -1
View File
@@ -232,11 +232,57 @@ class TestApprovalModuleExports:
assert cb_start != -1, "_approval_notify_cb must exist" assert cb_start != -1, "_approval_notify_cb must exist"
cb_end = STREAMING_SRC.find("_reg_notify(session_id, _approval_notify_cb)", cb_start) cb_end = STREAMING_SRC.find("_reg_notify(session_id, _approval_notify_cb)", cb_start)
cb_body = STREAMING_SRC[cb_start:cb_end] cb_body = STREAMING_SRC[cb_start:cb_end]
assert "_submit_pending_for_polling(session_id, approval_data)" in cb_body, \ assert "head, total = _submit_pending_for_polling(session_id, approval_data)" in cb_body, \
"approval notify callback must mirror approval data into polling state" "approval notify callback must mirror approval data into polling state"
assert '"pending_count": total' in cb_body, \
"approval notify callback must publish the reconciled pending count"
assert "put('approval', approval_data)" in cb_body, \ assert "put('approval', approval_data)" in cb_body, \
"approval notify callback must still push the SSE event" "approval notify callback must still push the SSE event"
def test_reversed_local_callbacks_publish_all_parked_entries(self):
"""The local producer must count every exact parked entry, not only its head."""
import api.route_approvals as ra
from api import routes
sid = f"unit-reversed-local-{uuid.uuid4().hex[:8]}"
entry_a = _ApprovalEntry({
"command": "same-command",
"description": "same-description",
"pattern_key": "same-pattern",
"pattern_keys": ["same-pattern"],
})
entry_b = _ApprovalEntry(dict(entry_a.data))
events = []
with _lock:
_gateway_queues[sid] = [entry_a, entry_b]
ra._pending.pop(sid, None)
def local_producer(approval_data):
head, total = ra.submit_gateway_pending_mirror(sid, approval_data)
events.append((head, total))
try:
local_producer(entry_b.data)
local_producer(entry_a.data)
assert [total for _head, total in events] == [2, 2]
assert all(head["command"] == "same-command" for head, _total in events)
head_id = events[0][0]["approval_id"]
assert head_id == events[1][0]["approval_id"]
assert routes._resolve_approval_legacy(sid, head_id, "once") is True
assert entry_a.event.is_set()
assert not entry_b.event.is_set()
with _lock:
successor = ra._pending[sid][0]
assert successor["approval_id"] != head_id
with _lock:
assert ra.reconcile_gateway_pending_mirror_locked(sid)[1] == 1
finally:
with _lock:
ra._pending.pop(sid, None)
_gateway_queues.pop(sid, None)
ra._pending.pop(sid, None)
# ── HTTP regression tests (test server, port 8788) ─────────────────────────── # ── HTTP regression tests (test server, port 8788) ───────────────────────────
+18 -8
View File
@@ -336,6 +336,19 @@ def test_legacy_approval_records_run_id_for_response_relay():
# (connection still open, run parked) — this test's stream already ran to # (connection still open, run parked) — this test's stream already ran to
# completion, which cleared active_stream_id and popped the mapping. # completion, which cleared active_stream_id and popped the mapping.
_STREAM_RUN_IDS[stream_id] = "run-legacy-1" _STREAM_RUN_IDS[stream_id] = "run-legacy-1"
from types import SimpleNamespace
from api import route_approvals as ra
with ra._lock:
ra._gateway_queues["sess-legacy-runid"] = [SimpleNamespace(data={
"command": "rm -rf /tmp/test",
"description": "Delete temporary files",
"pattern_key": "dangerous_command",
"pattern_keys": ["dangerous_command"],
"approval_id": "appr-legacy-runid",
"run_id": "run-legacy-1",
})]
live_entry = ra._gateway_queues["sess-legacy-runid"][0]
ra.submit_gateway_pending_mirror("sess-legacy-runid", live_entry.data)
relay_session = MagicMock() relay_session = MagicMock()
relay_session.active_stream_id = stream_id relay_session.active_stream_id = stream_id
captured = {} captured = {}
@@ -447,8 +460,8 @@ def test_legacy_teardown_clears_stale_gateway_mirror_and_notifies_empty_state():
ra._gateway_queues.pop(session_id, None) ra._gateway_queues.pop(session_id, None)
def test_legacy_teardown_preserves_live_gateway_head_mirror(): def test_legacy_teardown_retires_live_gateway_head_mirror():
"""A live gateway head must still be mirrored after legacy teardown runs.""" """A mapped legacy run retires its live gateway mirror during outer teardown."""
from types import SimpleNamespace from types import SimpleNamespace
from api import route_approvals as ra from api import route_approvals as ra
from api.config import STREAMS, STREAMS_LOCK from api.config import STREAMS, STREAMS_LOCK
@@ -507,14 +520,11 @@ def test_legacy_teardown_preserves_live_gateway_head_mirror():
payloads = _drain_queue(subscriber) payloads = _drain_queue(subscriber)
assert payloads, "Expected mirrored approval notifications" assert payloads, "Expected mirrored approval notifications"
assert payloads[-1]["pending"]["_gateway_mirror"] is True assert payloads[-1]["pending"] is None
assert payloads[-1]["pending_count"] == 1 assert payloads[-1]["pending_count"] == 0
with ra._lock: with ra._lock:
pending = ra._pending.get(session_id) pending = ra._pending.get(session_id)
assert isinstance(pending, list) assert pending is None
assert len(pending) == 1
assert pending[0]["approval_id"] == approval_data["approval_id"]
assert pending[0]["_gateway_mirror"] is True
finally: finally:
ra._approval_sse_unsubscribe(session_id, subscriber) ra._approval_sse_unsubscribe(session_id, subscriber)
with STREAMS_LOCK: with STREAMS_LOCK:
+19
View File
@@ -115,6 +115,25 @@ def test_attention_summary_prefers_pending_approvals_over_clarify_questions():
_clear_attention_state(sid) _clear_attention_state(sid)
def test_attention_summary_ignores_stale_gateway_mirror():
sid = "attention-stale-gateway-session"
_clear_attention_state(sid)
try:
with routes._lock:
routes._pending[sid] = [{
"_gateway_mirror": True,
"approval_id": "appr-stale-gateway",
"run_id": "run-stale-gateway",
"command": "rm -rf /tmp/stale",
"description": "Stale approval",
}]
routes._gateway_queues[sid] = []
assert routes._session_attention_summary(sid) is None
finally:
_clear_attention_state(sid)
def test_attention_summary_reports_clarify_when_no_approval_is_pending(): def test_attention_summary_reports_clarify_when_no_approval_is_pending():
sid = "attention-clarify-session" sid = "attention-clarify-session"
_clear_attention_state(sid) _clear_attention_state(sid)