APEX/BrowserUse_and_ComputerUse_skills
5
0
Fork 0
Code Issues 6 Pull requests 1 Projects 1 Releases Packages Wiki Activity Actions

fix(cli): throttle UI invalidate to prevent terminal blinking on SSH

Browse source
This commit is contained in:
ygd58 2026-03-02 15:56:53 +01:00
parent 3c13feed4c
commit b603b6e1c9
1 changed files with 20 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

42
cli.py
View file

@ -916,6 +916,15 @@ class HermesCLI:
# History file for persistent input recall across sessions # History file for persistent input recall across sessions
self._history_file = Path.home() / ".hermes_history" self._history_file = Path.home() / ".hermes_history"
self._last_invalidate: float = 0.0 # throttle UI repaints
def _invalidate(self, min_interval: float = 0.25) -> None:
"""Throttled UI repaint — prevents terminal blinking on slow/SSH connections."""
import time as _time
now = _time.monotonic()
if hasattr(self, "_app") and self._app and (now - self._last_invalidate) >= min_interval:
self._last_invalidate = now
self._app.invalidate()
def _ensure_runtime_credentials(self) -> bool: def _ensure_runtime_credentials(self) -> bool:
""" """
@ -1903,8 +1912,7 @@ class HermesCLI:
self._clarify_freetext = is_open_ended self._clarify_freetext = is_open_ended
# Trigger prompt_toolkit repaint from this (non-main) thread # Trigger prompt_toolkit repaint from this (non-main) thread
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
# Poll in 1-second ticks so the countdown refreshes in the UI. # Poll in 1-second ticks so the countdown refreshes in the UI.
# Each tick triggers an invalidate() to repaint the hint line. # Each tick triggers an invalidate() to repaint the hint line.
@ -1918,15 +1926,13 @@ class HermesCLI:
if remaining <= 0: if remaining <= 0:
break break
# Repaint so the countdown updates # Repaint so the countdown updates
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
# Timed out — tear down the UI and let the agent decide # Timed out — tear down the UI and let the agent decide
self._clarify_state = None self._clarify_state = None
self._clarify_freetext = False self._clarify_freetext = False
self._clarify_deadline = 0 self._clarify_deadline = 0
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
_cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}") _cprint(f"\n{_DIM}(clarify timed out after {timeout}s — agent will decide){_RST}")
return ( return (
"The user did not provide a response within the time limit. " "The user did not provide a response within the time limit. "
@ -1951,16 +1957,14 @@ class HermesCLI:
} }
self._sudo_deadline = _time.monotonic() + timeout self._sudo_deadline = _time.monotonic() + timeout
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
while True: while True:
try: try:
result = response_queue.get(timeout=1) result = response_queue.get(timeout=1)
self._sudo_state = None self._sudo_state = None
self._sudo_deadline = 0 self._sudo_deadline = 0
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
if result: if result:
_cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}") _cprint(f"\n{_DIM} ✓ Password received (cached for session){_RST}")
else: else:
@ -1970,13 +1974,11 @@ class HermesCLI:
remaining = self._sudo_deadline - _time.monotonic() remaining = self._sudo_deadline - _time.monotonic()
if remaining <= 0: if remaining <= 0:
break break
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
self._sudo_state = None self._sudo_state = None
self._sudo_deadline = 0 self._sudo_deadline = 0
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
_cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}") _cprint(f"\n{_DIM} ⏱ Timeout — continuing without sudo{_RST}")
return "" return ""
@ -2002,28 +2004,24 @@ class HermesCLI:
} }
self._approval_deadline = _time.monotonic() + timeout self._approval_deadline = _time.monotonic() + timeout
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
while True: while True:
try: try:
result = response_queue.get(timeout=1) result = response_queue.get(timeout=1)
self._approval_state = None self._approval_state = None
self._approval_deadline = 0 self._approval_deadline = 0
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
return result return result
except queue.Empty: except queue.Empty:
remaining = self._approval_deadline - _time.monotonic() remaining = self._approval_deadline - _time.monotonic()
if remaining <= 0: if remaining <= 0:
break break
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
self._approval_state = None self._approval_state = None
self._approval_deadline = 0 self._approval_deadline = 0
if hasattr(self, '_app') and self._app: self._invalidate()
self._app.invalidate()
_cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}") _cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}")
return "deny" return "deny"