From 165454dd7b90e57cbccacb7fc831ea6a1b9bed46 Mon Sep 17 00:00:00 2001 From: nesquena-hermes Date: Fri, 5 Jun 2026 13:10:45 -0700 Subject: [PATCH] =?UTF-8?q?Release=20v0.51.275=20=E2=80=94=20Release=20IQ?= =?UTF-8?q?=20(stage-p3d=20=E2=80=94=20approval=20SSE=20state=20extraction?= =?UTF-8?q?=20#3575)=20(#3681)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor(routes): extract approval SSE state into api/route_approvals.py (#3575) Co-authored-by: Rod Boev * docs(changelog): v0.51.275 — Release IQ (stage-p3d) --------- Co-authored-by: nesquena-hermes <[email protected]> Co-authored-by: Rod Boev --- CHANGELOG.md | 5 + api/route_approvals.py | 136 +++++++++++++++++++++ api/routes.py | 148 ++++------------------- tests/test_approval_queue.py | 6 +- tests/test_approval_sse.py | 26 ++-- tests/test_attention_session_events.py | 9 +- tests/test_route_approvals_extraction.py | 83 +++++++++++++ 7 files changed, 272 insertions(+), 141 deletions(-) create mode 100644 api/route_approvals.py create mode 100644 tests/test_route_approvals_extraction.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 20442a34..2cf31525 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,11 @@ ## [Unreleased] +## [v0.51.275] — 2026-06-05 — Release IQ (stage-p3d — approval SSE state extraction) + +### Internal +- **Approval SSE/queue state is extracted from `api/routes.py` into a dedicated `api/route_approvals.py` module**, reducing `routes.py` by ~140 lines with no behavior change. Pure refactor — the approval queue, SSE broadcast, and attention-session-event paths are unchanged (verified by the existing approval/SSE/attention test suites plus a new extraction-contract test). (#3575, @rodboev) + ## [v0.51.274] — 2026-06-05 — Release IP (stage-p3c — symlink-swap TOCTOU hardening for file APIs) ### Security diff --git a/api/route_approvals.py b/api/route_approvals.py new file mode 100644 index 00000000..698725a8 --- /dev/null +++ b/api/route_approvals.py @@ -0,0 +1,136 @@ +"""Approval SSE state and helpers. + +State-extraction prelude to the routes.py split tracked in #1907. +Extracts approval state, not handlers, by design. +""" +import queue +import threading +import uuid + +from api.session_events import publish_session_list_changed + +# Approval system (optional -- graceful fallback if agent not available) +try: + from tools.approval import ( + submit_pending as _submit_pending_raw, + approve_session, + approve_permanent, + save_permanent_allowlist, + is_approved, + _pending, + _lock, + _permanent_approved, + _gateway_queues, + resolve_gateway_approval, + enable_session_yolo, + disable_session_yolo, + is_session_yolo_enabled, + ) +except ImportError: + _submit_pending_raw = lambda *a, **k: None + approve_session = lambda *a, **k: None + approve_permanent = lambda *a, **k: None + save_permanent_allowlist = lambda *a, **k: None + is_approved = lambda *a, **k: True + resolve_gateway_approval = lambda *a, **k: 0 + enable_session_yolo = lambda *a, **k: None + disable_session_yolo = lambda *a, **k: None + is_session_yolo_enabled = lambda *a, **k: False + _pending = {} + _lock = threading.Lock() + _permanent_approved = set() + _gateway_queues = {} + + +# ── Approval SSE subscribers (long-connection push) ────────────────────────── +_approval_sse_subscribers: dict[str, list[queue.Queue]] = {} + + +def _approval_sse_subscribe(session_id: str) -> queue.Queue: + """Register an SSE subscriber for approval events on a given session.""" + q = queue.Queue(maxsize=16) + with _lock: + _approval_sse_subscribers.setdefault(session_id, []).append(q) + return q + + +def _approval_sse_unsubscribe(session_id: str, q: queue.Queue) -> None: + """Remove an SSE subscriber.""" + with _lock: + subs = _approval_sse_subscribers.get(session_id) + if subs and q in subs: + subs.remove(q) + if not subs: + _approval_sse_subscribers.pop(session_id, None) + + +def _approval_sse_notify_locked(session_id: str, head: dict | None, total: int) -> None: + """Push an approval event to all SSE subscribers for a session. + + CALLER MUST HOLD `_lock`. Snapshots the subscriber list under the held + lock and then calls `q.put_nowait()` on each (which is itself thread-safe). + + `head` is the approval entry currently at the head of the queue (the one + the UI should display) — NOT the just-appended entry. With multiple + parallel approvals (#527), the just-appended entry is at the TAIL, but + `/api/approval/pending` always returns the HEAD, so SSE must match. + + `total` is the total number of pending approvals. + + Pass `head=None` and `total=0` when the queue has just been emptied (e.g. + `_handle_approval_respond` popped the last entry) so the client knows to + hide its approval card. + """ + payload = {"pending": dict(head) if head else None, "pending_count": total} + subs = _approval_sse_subscribers.get(session_id, ()) + for q in subs: + try: + q.put_nowait(payload) + except queue.Full: + pass # drop if subscriber is slow (bounded queue prevents memory leak) + + +def _approval_sse_notify(session_id: str, head: dict | None, total: int) -> None: + """Convenience wrapper that takes `_lock` itself. + + Use only from contexts that don't already hold `_lock`. Production call + sites (submit_pending, _handle_approval_respond) MUST hold the lock and + call `_approval_sse_notify_locked` directly to avoid a notify-ordering + race where a later append's notify can fire before an earlier append's + notify (resulting in stale `pending_count`). + """ + with _lock: + _approval_sse_notify_locked(session_id, head, total) + + +def submit_pending(session_key: str, approval: dict) -> None: + """Append a pending approval to the per-session queue. + + Wraps the agent's submit_pending to: + - Add a stable approval_id (uuid4 hex) so the respond endpoint can target + a specific entry even when multiple approvals are queued simultaneously. + - Change the storage from a single overwriting dict value to a list, so + parallel tool calls each get their own approval slot (fixes #527). + - Notify any connected SSE subscribers immediately. + """ + entry = dict(approval) + entry.setdefault("approval_id", uuid.uuid4().hex) + with _lock: + queue_list = _pending.setdefault(session_key, []) + # Replace a legacy non-list value if the agent version uses the old pattern. + if not isinstance(queue_list, list): + _pending[session_key] = [queue_list] + queue_list = _pending[session_key] + queue_list.append(entry) + total = len(queue_list) + head = queue_list[0] # /api/approval/pending always returns head + # Push to SSE subscribers from inside _lock so two parallel + # submit_pending calls can't deliver out-of-order (T2's later + # notify arriving before T1's earlier notify with a stale count). + _approval_sse_notify_locked(session_key, head, total) + publish_session_list_changed("attention_pending") + # NOTE: We do NOT call _submit_pending_raw here — that function overwrites + # _pending[session_key] with a single dict, which would undo the list we just + # built. The gateway blocking path uses _gateway_queues (a separate mechanism + # managed by check_all_command_guards / register_gateway_notify), which is + # unaffected by _pending. The _pending dict is only used for UI polling. diff --git a/api/routes.py b/api/routes.py index ebe41042..14efa50e 100644 --- a/api/routes.py +++ b/api/routes.py @@ -2958,131 +2958,29 @@ from api.oauth import ( start_onboarding_oauth_flow, ) -# Approval system (optional -- graceful fallback if agent not available) -try: - from tools.approval import ( - submit_pending as _submit_pending_raw, - approve_session, - approve_permanent, - save_permanent_allowlist, - is_approved, - _pending, - _lock, - _permanent_approved, - _gateway_queues, - resolve_gateway_approval, - enable_session_yolo, - disable_session_yolo, - is_session_yolo_enabled, - ) -except ImportError: - _submit_pending_raw = lambda *a, **k: None - approve_session = lambda *a, **k: None - approve_permanent = lambda *a, **k: None - save_permanent_allowlist = lambda *a, **k: None - is_approved = lambda *a, **k: True - resolve_gateway_approval = lambda *a, **k: 0 - enable_session_yolo = lambda *a, **k: None - disable_session_yolo = lambda *a, **k: None - is_session_yolo_enabled = lambda *a, **k: False - _pending = {} - _lock = threading.Lock() - _permanent_approved = set() - _gateway_queues = {} - - -# ── Approval SSE subscribers (long-connection push) ────────────────────────── -_approval_sse_subscribers: dict[str, list[queue.Queue]] = {} - - -def _approval_sse_subscribe(session_id: str) -> queue.Queue: - """Register an SSE subscriber for approval events on a given session.""" - q = queue.Queue(maxsize=16) - with _lock: - _approval_sse_subscribers.setdefault(session_id, []).append(q) - return q - - -def _approval_sse_unsubscribe(session_id: str, q: queue.Queue) -> None: - """Remove an SSE subscriber.""" - with _lock: - subs = _approval_sse_subscribers.get(session_id) - if subs and q in subs: - subs.remove(q) - if not subs: - _approval_sse_subscribers.pop(session_id, None) - - -def _approval_sse_notify_locked(session_id: str, head: dict | None, total: int) -> None: - """Push an approval event to all SSE subscribers for a session. - - CALLER MUST HOLD `_lock`. Snapshots the subscriber list under the held - lock and then calls `q.put_nowait()` on each (which is itself thread-safe). - - `head` is the approval entry currently at the head of the queue (the one - the UI should display) — NOT the just-appended entry. With multiple - parallel approvals (#527), the just-appended entry is at the TAIL, but - `/api/approval/pending` always returns the HEAD, so SSE must match. - - `total` is the total number of pending approvals. - - Pass `head=None` and `total=0` when the queue has just been emptied (e.g. - `_handle_approval_respond` popped the last entry) so the client knows to - hide its approval card. - """ - payload = {"pending": dict(head) if head else None, "pending_count": total} - subs = _approval_sse_subscribers.get(session_id, ()) - for q in subs: - try: - q.put_nowait(payload) - except queue.Full: - pass # drop if subscriber is slow (bounded queue prevents memory leak) - - -def _approval_sse_notify(session_id: str, head: dict | None, total: int) -> None: - """Convenience wrapper that takes `_lock` itself. - - Use only from contexts that don't already hold `_lock`. Production call - sites (submit_pending, _handle_approval_respond) MUST hold the lock and - call `_approval_sse_notify_locked` directly to avoid a notify-ordering - race where a later append's notify can fire before an earlier append's - notify (resulting in stale `pending_count`). - """ - with _lock: - _approval_sse_notify_locked(session_id, head, total) - - -def submit_pending(session_key: str, approval: dict) -> None: - """Append a pending approval to the per-session queue. - - Wraps the agent's submit_pending to: - - Add a stable approval_id (uuid4 hex) so the respond endpoint can target - a specific entry even when multiple approvals are queued simultaneously. - - Change the storage from a single overwriting dict value to a list, so - parallel tool calls each get their own approval slot (fixes #527). - - Notify any connected SSE subscribers immediately. - """ - entry = dict(approval) - entry.setdefault("approval_id", uuid.uuid4().hex) - with _lock: - queue_list = _pending.setdefault(session_key, []) - # Replace a legacy non-list value if the agent version uses the old pattern. - if not isinstance(queue_list, list): - _pending[session_key] = [queue_list] - queue_list = _pending[session_key] - queue_list.append(entry) - total = len(queue_list) - head = queue_list[0] # /api/approval/pending always returns head - # Push to SSE subscribers from inside _lock so two parallel - # submit_pending calls can't deliver out-of-order (T2's later - # notify arriving before T1's earlier notify with a stale count). - _approval_sse_notify_locked(session_key, head, total) - publish_session_list_changed("attention_pending") - # NOTE: We do NOT call _submit_pending_raw here — that function overwrites - # _pending[session_key] with a single dict, which would undo the list we just - # built. The gateway blocking path uses _gateway_queues (a separate mechanism - # managed by check_all_command_guards / register_gateway_notify), which is - # unaffected by _pending. The _pending dict is only used for UI polling. +# Approval system -- state and helpers live in api.route_approvals; imported +# here for backward compatibility so existing call sites continue to resolve. +from api.route_approvals import ( # noqa: F401 — re-exports for backward compat + _submit_pending_raw, + approve_session, + approve_permanent, + save_permanent_allowlist, + is_approved, + _pending, + _lock, + _permanent_approved, + _gateway_queues, + resolve_gateway_approval, + enable_session_yolo, + disable_session_yolo, + is_session_yolo_enabled, + _approval_sse_subscribers, + _approval_sse_subscribe, + _approval_sse_unsubscribe, + _approval_sse_notify_locked, + _approval_sse_notify, + submit_pending, +) # Clarify prompts (optional -- graceful fallback if agent not available) try: diff --git a/tests/test_approval_queue.py b/tests/test_approval_queue.py index d52a3421..2ca10f58 100644 --- a/tests/test_approval_queue.py +++ b/tests/test_approval_queue.py @@ -13,6 +13,10 @@ REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve() sys.path.insert(0, str(REPO_ROOT)) ROUTES_SRC = (REPO_ROOT / "api" / "routes.py").read_text(encoding="utf-8") +# Approval helpers moved to api.route_approvals after the #1907 extraction; +# combine both files so static-analysis assertions still pass. +_ROUTE_APPROVALS = REPO_ROOT / "api" / "route_approvals.py" +ROUTES_SRC_FULL = ROUTES_SRC + (_ROUTE_APPROVALS.read_text(encoding="utf-8") if _ROUTE_APPROVALS.exists() else "") MESSAGES_JS = (REPO_ROOT / "static" / "messages.js").read_text(encoding="utf-8") INDEX_HTML = (REPO_ROOT / "static" / "index.html").read_text(encoding="utf-8") @@ -24,7 +28,7 @@ INDEX_HTML = (REPO_ROOT / "static" / "index.html").read_text(encoding="utf-8") def test_submit_pending_appends_to_list(): """submit_pending() must append to a list, not overwrite.""" # The new wrapper must contain a queue append (list mutation pattern) - assert "queue_list.append(entry)" in ROUTES_SRC or "queue.append(entry)" in ROUTES_SRC, \ + assert "queue_list.append(entry)" in ROUTES_SRC_FULL or "queue.append(entry)" in ROUTES_SRC_FULL, \ "submit_pending() must append entry to a list queue, not overwrite _pending[sid]" diff --git a/tests/test_approval_sse.py b/tests/test_approval_sse.py index 0ff950d8..fae8cf0f 100644 --- a/tests/test_approval_sse.py +++ b/tests/test_approval_sse.py @@ -24,6 +24,11 @@ REPO_ROOT = pathlib.Path(__file__).parent.parent.resolve() sys.path.insert(0, str(REPO_ROOT)) ROUTES_SRC = (REPO_ROOT / "api" / "routes.py").read_text(encoding="utf-8") +# Approval SSE state and helpers live in route_approvals after the #1907 +# extraction; combine both files so structural assertions below still pass. +_ROUTE_APPROVALS = REPO_ROOT / "api" / "route_approvals.py" +APPROVAL_SRC = _ROUTE_APPROVALS.read_text(encoding="utf-8") if _ROUTE_APPROVALS.exists() else "" +ROUTES_SRC_FULL = ROUTES_SRC + APPROVAL_SRC MESSAGES_JS = (REPO_ROOT / "static" / "messages.js").read_text(encoding="utf-8") @@ -46,17 +51,17 @@ class TestSSEStaticAnalysis: def test_subscribe_function_exists(self): """_approval_sse_subscribe must exist and use a Queue.""" - assert "def _approval_sse_subscribe(" in ROUTES_SRC, \ + assert "def _approval_sse_subscribe(" in ROUTES_SRC_FULL, \ "_approval_sse_subscribe must be defined" def test_unsubscribe_function_exists(self): """_approval_sse_unsubscribe must exist and clean up empty lists.""" - assert "def _approval_sse_unsubscribe(" in ROUTES_SRC, \ + assert "def _approval_sse_unsubscribe(" in ROUTES_SRC_FULL, \ "_approval_sse_unsubscribe must be defined" def test_notify_function_exists(self): """_approval_sse_notify must exist and push to subscriber queues.""" - assert "def _approval_sse_notify(" in ROUTES_SRC, \ + assert "def _approval_sse_notify(" in ROUTES_SRC_FULL, \ "_approval_sse_notify must be defined" def test_sse_subscribers_dict_exists(self): @@ -95,7 +100,7 @@ class TestSSEStaticAnalysis: # block as the queue mutation so two parallel submit_pending calls can't # deliver out-of-order with stale pending_count. Tracks the v0.50.248 # MUST-FIX A fix. - assert "_approval_sse_notify_locked(session_key, head, total)" in ROUTES_SRC, \ + assert "_approval_sse_notify_locked(session_key, head, total)" in ROUTES_SRC_FULL, \ ("submit_pending() must call _approval_sse_notify_locked(session_key, head, total) " "from inside the `with _lock:` block — not the unlocked _approval_sse_notify wrapper, " "and head must be queue_list[0] (the head, not the just-appended entry).") @@ -119,25 +124,26 @@ class TestSSEStaticAnalysis: def test_notify_drops_on_full(self): """_approval_sse_notify must silently drop events when subscriber is slow.""" # The queue.Full exception handler - assert "queue.Full" in ROUTES_SRC, \ + assert "queue.Full" in ROUTES_SRC_FULL, \ "_approval_sse_notify must handle queue.Full to drop events for slow subscribers" def test_subscribe_uses_shared_lock(self): """subscribe/unsubscribe/notify must all use the same _lock.""" - # All three functions must use _lock + # All three functions must use _lock; search the combined corpus since the + # helpers live in api.route_approvals after the #1907 extraction. for func in ["_approval_sse_subscribe", "_approval_sse_unsubscribe", "_approval_sse_notify"]: # Find the function and verify it uses "with _lock" - func_start = ROUTES_SRC.find(f"def {func}(") + func_start = ROUTES_SRC_FULL.find(f"def {func}(") assert func_start != -1, f"{func} must exist" # Find the next function definition after this one - next_func = ROUTES_SRC.find("\ndef ", func_start + 1) - func_body = ROUTES_SRC[func_start:next_func] if next_func != -1 else ROUTES_SRC[func_start:] + next_func = ROUTES_SRC_FULL.find("\ndef ", func_start + 1) + func_body = ROUTES_SRC_FULL[func_start:next_func] if next_func != -1 else ROUTES_SRC_FULL[func_start:] assert "with _lock:" in func_body, \ f"{func} must use 'with _lock:' for thread safety" def test_unsubscribe_cleans_empty_session(self): """Unsubscribe must remove empty session keys from the dict.""" - assert "_approval_sse_subscribers.pop(session_id, None)" in ROUTES_SRC, \ + assert "_approval_sse_subscribers.pop(session_id, None)" in ROUTES_SRC_FULL, \ "_approval_sse_unsubscribe must pop session_id when subscriber list is empty" diff --git a/tests/test_attention_session_events.py b/tests/test_attention_session_events.py index 66dfed4f..8de56f78 100644 --- a/tests/test_attention_session_events.py +++ b/tests/test_attention_session_events.py @@ -42,13 +42,12 @@ def test_clarify_pending_and_resolved_publish_session_list_changed(monkeypatch): def test_approval_pending_and_resolved_publish_session_list_changed(monkeypatch): import api.routes as routes + import api.route_approvals as route_approvals events = [] - monkeypatch.setattr( - routes, - "publish_session_list_changed", - lambda reason: events.append(reason), - ) + capture = lambda reason: events.append(reason) + monkeypatch.setattr(route_approvals, "publish_session_list_changed", capture) + monkeypatch.setattr(routes, "publish_session_list_changed", capture) sid = "test-attention-approval" with routes._lock: diff --git a/tests/test_route_approvals_extraction.py b/tests/test_route_approvals_extraction.py new file mode 100644 index 00000000..b1bb5150 --- /dev/null +++ b/tests/test_route_approvals_extraction.py @@ -0,0 +1,83 @@ +"""Import-resolution and identity tests for the route_approvals extraction. + +Verifies that api.route_approvals is self-contained and that api.routes +re-exports the same live objects (not copies) so existing callers are +unaffected by the move. +""" + +def test_route_approvals_imports(): + from api.route_approvals import submit_pending, _approval_sse_subscribers + assert callable(submit_pending) + assert isinstance(_approval_sse_subscribers, dict) + + +def test_backward_compat_imports(): + from api.routes import submit_pending + assert callable(submit_pending) + + +def test_identity(): + from api.route_approvals import _approval_sse_subscribers as a + from api.routes import _approval_sse_subscribers as b + assert a is b, "routes.py must re-export the same object, not a copy" + + +def test_pending_identity(): + """_pending dict must be the same object in both modules.""" + from api.route_approvals import _pending as a + from api.routes import _pending as b + assert a is b, "routes.py must re-export the same _pending object" + + +def test_lock_identity(): + """_lock must be the same object in both modules.""" + from api.route_approvals import _lock as a + from api.routes import _lock as b + assert a is b, "routes.py must re-export the same _lock object" + + +def test_sse_helpers_importable_from_route_approvals(): + """All SSE helpers must be importable directly from route_approvals.""" + from api.route_approvals import ( + _approval_sse_subscribe, + _approval_sse_unsubscribe, + _approval_sse_notify_locked, + _approval_sse_notify, + ) + assert callable(_approval_sse_subscribe) + assert callable(_approval_sse_unsubscribe) + assert callable(_approval_sse_notify_locked) + assert callable(_approval_sse_notify) + + +def test_sse_helpers_backward_compat(): + """SSE helpers must still be importable from routes for backward compat.""" + from api.routes import ( + _approval_sse_subscribe, + _approval_sse_unsubscribe, + _approval_sse_notify_locked, + _approval_sse_notify, + ) + assert callable(_approval_sse_subscribe) + assert callable(_approval_sse_unsubscribe) + assert callable(_approval_sse_notify_locked) + assert callable(_approval_sse_notify) + + +def test_sse_helper_identity(): + """SSE helpers imported from routes must be the same callables from route_approvals.""" + import api.route_approvals as ra + import api.routes as r + assert ra._approval_sse_subscribe is r._approval_sse_subscribe + assert ra._approval_sse_unsubscribe is r._approval_sse_unsubscribe + assert ra._approval_sse_notify_locked is r._approval_sse_notify_locked + assert ra._approval_sse_notify is r._approval_sse_notify + assert ra.submit_pending is r.submit_pending + + +def test_no_circular_import(): + """route_approvals must not import from api.routes (no circular dep).""" + import pathlib + src = (pathlib.Path(__file__).parent.parent / "api" / "route_approvals.py").read_text() + assert "from api.routes" not in src, "route_approvals.py must not import from api.routes" + assert "import api.routes" not in src, "route_approvals.py must not import api.routes"