Compare commits

...

3 Commits

Author SHA1 Message Date
teknium1
1cd5c13f9d chore(release): map islam666 for salvaged PR #41166 2026-06-07 06:42:26 -07:00
teknium1
7ae849db36 fix(cli): also bypass throttle on approval success path + de-bloat test
Hardening on salvaged #41166: the success-path invalidate (after the user
responds) was still throttled while every other approval path bypassed it —
panel dismissal could lag/drop up to 250ms. Convert it to a direct
_app.invalidate() for consistency. Also replace the 6s real-sleep retry test
with a fake monotonic clock (suite 9s -> <3s).
2026-06-07 06:42:26 -07:00
islam666
56167629b0 fix(cli): bypass _invalidate() throttle for approval panel render (#41098)
The approval callback's initial render was going through the 250ms
_invalidate() throttle, which silently dropped the redraw when any
other UI event (spinner, output flush) had triggered an invalidation
within the previous 250ms. This made the approval panel never appear,
causing commands to be silently denied after 60s timeout.

Fix: call app.invalidate() directly in _approval_callback (matching the
precedent set by _force_full_redraw), and add a terminal bell to alert
the user that approval is pending.
2026-06-07 06:40:29 -07:00
3 changed files with 237 additions and 4 deletions

34
cli.py
View File

@@ -11919,7 +11919,18 @@ class HermesCLI:
}
self._approval_deadline = _time.monotonic() + timeout
self._invalidate()
# Bypass the _invalidate() throttle — approval is a user-blocking
# modal that must render immediately. The 250ms throttle exists to
# prevent terminal flicker on slow connections, but dropping an
# approval redraw means the panel never appears and the command is
# silently denied after 60s.
if hasattr(self, "_app") and self._app:
try:
self._app.invalidate()
except Exception:
pass
# Audible signal so the user knows approval is pending.
print("\a", end="", flush=True)
_last_countdown_refresh = _time.monotonic()
while True:
@@ -11927,7 +11938,14 @@ class HermesCLI:
result = response_queue.get(timeout=1)
self._approval_state = None
self._approval_deadline = 0
self._invalidate()
# Bypass the throttle here too — the panel must clear
# immediately once the user responds, consistent with the
# entry/countdown/timeout invalidates above.
if hasattr(self, "_app") and self._app:
try:
self._app.invalidate()
except Exception:
pass
return result
except queue.Empty:
remaining = self._approval_deadline - _time.monotonic()
@@ -11936,11 +11954,19 @@ class HermesCLI:
now = _time.monotonic()
if now - _last_countdown_refresh >= 5.0:
_last_countdown_refresh = now
self._invalidate()
if hasattr(self, "_app") and self._app:
try:
self._app.invalidate()
except Exception:
pass
self._approval_state = None
self._approval_deadline = 0
self._invalidate()
if hasattr(self, "_app") and self._app:
try:
self._app.invalidate()
except Exception:
pass
_cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}")
return "deny"

View File

@@ -58,6 +58,7 @@ AUTHOR_MAP = {
"129007007+HeLLGURD@users.noreply.github.com": "HeLLGURD",
"290859878+synapsesx@users.noreply.github.com": "synapsesx",
"dirtyren@users.noreply.github.com": "dirtyren",
"islam666@users.noreply.github.com": "islam666",
"zhaolei.vc@bytedance.com": "zhaoleibd",
"jeffrobodie@gmail.com": "jeffrobodie-glitch",
"kyssta-exe@users.noreply.github.com": "kyssta-exe",

View File

@@ -0,0 +1,206 @@
"""Tests for the approval panel invalidate throttle bypass (issue #41098).
The approval callback must bypass the _invalidate() 250ms throttle and call
app.invalidate() directly, otherwise the approval panel never renders when
another UI event triggered an invalidation within the previous 250ms.
"""
import queue
import threading
import time
from types import SimpleNamespace
from unittest.mock import MagicMock, patch, call
import cli as cli_module
from cli import HermesCLI
def _make_cli_stub():
cli = HermesCLI.__new__(HermesCLI)
cli._approval_state = None
cli._approval_deadline = 0
cli._approval_lock = threading.Lock()
cli._sudo_state = None
cli._sudo_deadline = 0
cli._modal_input_snapshot = None
cli._invalidate = MagicMock()
cli._app = SimpleNamespace(invalidate=MagicMock(), current_buffer=MagicMock())
cli._cprint = MagicMock()
return cli
def test_approval_callback_bypasses_throttle_on_entry():
"""_approval_callback must call app.invalidate() directly, not _invalidate()."""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
# Simulate a response after 0.1s
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("once")
t = threading.Thread(target=respond_later)
t.start()
result = cli._approval_callback("rm -rf /", "Delete everything")
t.join()
assert result == "once"
# app.invalidate() should have been called (direct bypass)
assert cli._app.invalidate.call_count >= 1
# _invalidate() should NOT have been called for the initial render
# (it may be called for cleanup, but the critical path is app.invalidate)
def test_approval_callback_calls_app_invalidate_not_throttled():
"""The initial render must use app.invalidate(), bypassing the 250ms throttle."""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("deny")
t = threading.Thread(target=respond_later)
t.start()
cli._approval_callback("dangerous_cmd", "Run dangerous command")
t.join()
# app.invalidate must be called at least once for the initial render
cli._app.invalidate.assert_called()
def test_approval_callback_retry_bypasses_throttle():
"""The 5-second retry invalidation must also bypass the throttle.
Uses a fake monotonic clock so the >=5s countdown-refresh branch fires
without sleeping 6 real seconds.
"""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
# Fake clock: each call to monotonic() jumps forward 3s, so by the second
# queue.Empty timeout the loop has "elapsed" >=5s and fires the retry
# invalidate. The responder answers on the 3rd poll.
clock = {"t": 0.0}
def fake_monotonic():
clock["t"] += 3.0
return clock["t"]
polls = {"n": 0}
def fake_queue_get(self, timeout=None):
polls["n"] += 1
if polls["n"] >= 3:
return "once"
raise queue.Empty
with patch.object(cli_module, "CLI_CONFIG", {"approvals": {"timeout": 60}}), \
patch("time.monotonic", side_effect=fake_monotonic):
# Swap the response queue's get for a deterministic stub once state exists.
orig_callback = cli._approval_callback
def run():
return orig_callback("cmd", "desc")
# Patch queue.Queue.get globally for this call.
with patch.object(queue.Queue, "get", fake_queue_get):
result = run()
assert result == "once"
# initial render + at least one 5s-retry invalidate
assert cli._app.invalidate.call_count >= 2
def test_approval_callback_timeout_bypasses_throttle():
"""Timeout cleanup must also bypass the throttle."""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
# Short timeout so we don't wait 60s
with patch.object(cli_module, "CLI_CONFIG", {"approvals": {"timeout": 2}}):
result = cli._approval_callback("cmd", "desc")
assert result == "deny"
# app.invalidate should be called for the initial render
assert cli._app.invalidate.call_count >= 1
def test_approval_callback_no_app_graceful():
"""If _app is not set, the callback must not crash."""
cli = _make_cli_stub()
cli._app = None
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("once")
t = threading.Thread(target=respond_later)
t.start()
result = cli._approval_callback("cmd", "desc")
t.join()
assert result == "once"
def test_approval_callback_app_invalidate_exception_graceful():
"""If app.invalidate() raises, the callback must not crash."""
cli = _make_cli_stub()
cli._app.invalidate.side_effect = RuntimeError("renderer dead")
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("once")
t = threading.Thread(target=respond_later)
t.start()
result = cli._approval_callback("cmd", "desc")
t.join()
assert result == "once"
def test_approval_callback_emits_bell():
"""The approval callback should emit a terminal bell on first render."""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("once")
t = threading.Thread(target=respond_later)
t.start()
with patch("cli.print") as mock_print:
cli._approval_callback("cmd", "desc")
t.join()
# Check that print was called with the bell character
bell_calls = [c for c in mock_print.call_args_list if c == call("\a", end="", flush=True)]
assert len(bell_calls) >= 1, "Terminal bell should be emitted on approval"
def test_invalidate_method_still_exists():
"""The _invalidate() method should still exist and be callable."""
cli = _make_cli_stub()
# _invalidate is a real method on the class — just verify it's callable
assert callable(getattr(cli, "_invalidate", None))
def test_approval_state_cleared_after_response():
"""_approval_state should be None after the callback returns."""
cli = _make_cli_stub()
cli._approval_choices = MagicMock(return_value=["once", "session", "always", "deny"])
def respond_later():
time.sleep(0.1)
cli._approval_state["response_queue"].put("once")
t = threading.Thread(target=respond_later)
t.start()
cli._approval_callback("cmd", "desc")
t.join()
assert cli._approval_state is None
assert cli._approval_deadline == 0