mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-06-16 15:11:18 +08:00
Compare commits
3 Commits
claude-cod
...
salvage/pr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1cd5c13f9d | ||
|
|
7ae849db36 | ||
|
|
56167629b0 |
34
cli.py
34
cli.py
@@ -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"
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
206
tests/cli/test_approval_invalidate_bypass.py
Normal file
206
tests/cli/test_approval_invalidate_bypass.py
Normal 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
|
||||
Reference in New Issue
Block a user