fix(cli): recover classic CLI output after resize

This commit is contained in:
helix4u 2026-05-05 16:10:26 -06:00 committed by Teknium
parent 17687911b7
commit 76074d9ee6
5 changed files with 389 additions and 21 deletions

200
cli.py
View file

@ -27,6 +27,7 @@ import tempfile
import time
import uuid
import textwrap
from collections import deque
from urllib.parse import unquote, urlparse
from contextlib import contextmanager
from pathlib import Path
@ -335,6 +336,8 @@ def load_cli_config() -> Dict[str, Any]:
"show_reasoning": False,
"streaming": True,
"busy_input_mode": "interrupt",
"persistent_output": True,
"persistent_output_max_lines": 200,
"skin": "default",
},
@ -1276,6 +1279,87 @@ def _render_final_assistant_content(text: str, mode: str = "render"):
return Markdown(plain)
_OUTPUT_HISTORY_ENABLED = True
_OUTPUT_HISTORY_REPLAYING = False
_OUTPUT_HISTORY_SUPPRESSED = False
_OUTPUT_HISTORY_MAX_LINES = 200
_OUTPUT_HISTORY = deque(maxlen=_OUTPUT_HISTORY_MAX_LINES)
_ANSI_CONTROL_RE = re.compile(
r"\x1b(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))"
)
def _coerce_output_history_limit(value) -> int:
try:
return max(10, int(value))
except (TypeError, ValueError):
return 200
def _configure_output_history(enabled: bool, max_lines=200) -> None:
"""Configure recent CLI output replayed after terminal redraws."""
global _OUTPUT_HISTORY_ENABLED, _OUTPUT_HISTORY_MAX_LINES, _OUTPUT_HISTORY
_OUTPUT_HISTORY_ENABLED = bool(enabled)
_OUTPUT_HISTORY_MAX_LINES = _coerce_output_history_limit(max_lines)
_OUTPUT_HISTORY = deque(maxlen=_OUTPUT_HISTORY_MAX_LINES)
def _clear_output_history() -> None:
_OUTPUT_HISTORY.clear()
@contextmanager
def _suspend_output_history():
global _OUTPUT_HISTORY_SUPPRESSED
old_value = _OUTPUT_HISTORY_SUPPRESSED
_OUTPUT_HISTORY_SUPPRESSED = True
try:
yield
finally:
_OUTPUT_HISTORY_SUPPRESSED = old_value
def _record_output_history_entry(entry) -> None:
if not _OUTPUT_HISTORY_ENABLED or _OUTPUT_HISTORY_REPLAYING or _OUTPUT_HISTORY_SUPPRESSED:
return
_OUTPUT_HISTORY.append(entry)
def _record_output_history(text: str) -> None:
if not _OUTPUT_HISTORY_ENABLED or _OUTPUT_HISTORY_REPLAYING or _OUTPUT_HISTORY_SUPPRESSED:
return
clean = _ANSI_CONTROL_RE.sub("", str(text)).replace("\r", "").rstrip("\n")
if not clean:
return
for line in clean.splitlines():
_record_output_history_entry(line)
def _replay_output_history() -> None:
"""Repaint recent output above the prompt after a full screen clear."""
global _OUTPUT_HISTORY_REPLAYING
if not _OUTPUT_HISTORY_ENABLED or not _OUTPUT_HISTORY:
return
_OUTPUT_HISTORY_REPLAYING = True
try:
for entry in tuple(_OUTPUT_HISTORY):
if callable(entry):
try:
lines = entry()
except Exception:
continue
if isinstance(lines, str):
lines = lines.splitlines()
else:
lines = [entry]
for line in lines:
_pt_print(_PT_ANSI(str(line)))
except Exception:
pass
finally:
_OUTPUT_HISTORY_REPLAYING = False
def _cprint(text: str):
"""Print ANSI-colored text through prompt_toolkit's native renderer.
@ -1292,6 +1376,8 @@ def _cprint(text: str):
``loop.call_soon_threadsafe``, which pauses the input area, prints
the line above it, and redraws the prompt cleanly.
"""
_record_output_history(text)
try:
from prompt_toolkit.application import get_app_or_none, run_in_terminal
except Exception:
@ -2048,6 +2134,10 @@ class HermesCLI:
self.bell_on_complete = CLI_CONFIG["display"].get("bell_on_complete", False)
# show_reasoning: display model thinking/reasoning before the response
self.show_reasoning = CLI_CONFIG["display"].get("show_reasoning", False)
_configure_output_history(
enabled=CLI_CONFIG["display"].get("persistent_output", True),
max_lines=CLI_CONFIG["display"].get("persistent_output_max_lines", 200),
)
# busy_input_mode: "interrupt" (Enter interrupts current run),
# "queue" (Enter queues for next turn), or "steer" (Enter injects
# mid-run via /steer, arriving after the next tool call).
@ -2325,6 +2415,9 @@ class HermesCLI:
# Status bar visibility (toggled via /statusbar)
self._status_bar_visible = True
self._resize_recovery_lock = threading.Lock()
self._resize_recovery_timer = None
self._resize_recovery_pending = False
# Background task tracking: {task_id: threading.Thread}
self._background_tasks: Dict[str, threading.Thread] = {}
@ -2332,6 +2425,8 @@ class HermesCLI:
def _invalidate(self, min_interval: float = 0.25) -> None:
"""Throttled UI repaint — prevents terminal blinking on slow/SSH connections."""
if getattr(self, "_resize_recovery_pending", False):
return
now = time.monotonic()
if hasattr(self, "_app") and self._app and (now - self._last_invalidate) >= min_interval:
self._last_invalidate = now
@ -2355,11 +2450,25 @@ class HermesCLI:
app = getattr(self, "_app", None)
if not app:
return
self._clear_prompt_toolkit_screen(app)
_replay_output_history()
try:
app.invalidate()
except Exception:
pass
def _clear_prompt_toolkit_screen(self, app, *, rebuild_scrollback: bool = False) -> None:
"""Clear the terminal and reset prompt_toolkit renderer state."""
try:
renderer = app.renderer
out = renderer.output
out.reset_attributes()
out.erase_screen()
if rebuild_scrollback:
try:
out.write_raw("\x1b[3J")
except Exception:
pass
out.cursor_goto(0, 0)
out.flush()
# Drop prompt_toolkit's cached screen + cursor state so the
@ -2368,10 +2477,57 @@ class HermesCLI:
renderer.reset(leave_alternate_screen=False)
except Exception:
pass
def _recover_after_resize(self, app, original_on_resize) -> None:
"""Recover a resized classic CLI without desynchronizing cursor state."""
self._clear_prompt_toolkit_screen(app, rebuild_scrollback=True)
_replay_output_history()
original_on_resize()
def _schedule_resize_recovery(self, app, original_on_resize, delay: float = 0.12) -> None:
"""Debounce resize redraws so footer chrome is not stamped into scrollback."""
try:
app.invalidate()
old_timer = getattr(self, "_resize_recovery_timer", None)
lock = getattr(self, "_resize_recovery_lock", None)
if lock is None:
lock = threading.Lock()
self._resize_recovery_lock = lock
def _timer_fired(timer_ref):
def _run_recovery():
with lock:
if getattr(self, "_resize_recovery_timer", None) is not timer_ref:
return
self._resize_recovery_timer = None
self._resize_recovery_pending = False
self._recover_after_resize(app, original_on_resize)
try:
loop = app.loop # type: ignore[attr-defined]
except Exception:
loop = None
if loop is not None:
try:
loop.call_soon_threadsafe(_run_recovery)
return
except Exception:
pass
_run_recovery()
with lock:
if old_timer is not None:
try:
old_timer.cancel()
except Exception:
pass
self._resize_recovery_pending = True
timer = threading.Timer(delay, lambda: _timer_fired(timer))
timer.daemon = True
self._resize_recovery_timer = timer
timer.start()
except Exception:
pass
self._resize_recovery_pending = False
self._recover_after_resize(app, original_on_resize)
def _status_bar_context_style(self, percent_used: Optional[int]) -> str:
if percent_used is None:
@ -4046,7 +4202,26 @@ class HermesCLI:
padding=(0, 1),
style=_history_text_c,
)
self._console_print(panel)
_record_output_history_entry(lambda: self._render_resume_history_panel_lines(panel))
with _suspend_output_history():
self._console_print(panel)
def _render_resume_history_panel_lines(self, panel) -> list[str]:
"""Render the resume panel at the current terminal width for resize replay."""
from io import StringIO
buf = StringIO()
width = shutil.get_terminal_size((80, 24)).columns
console = Console(
file=buf,
force_terminal=True,
color_system="truecolor",
highlight=False,
width=width,
)
with _suspend_output_history():
console.print(panel)
return buf.getvalue().rstrip("\n").splitlines()
def _try_attach_clipboard_image(self) -> bool:
"""Check clipboard for an image and attach it if found.
@ -6405,6 +6580,7 @@ class HermesCLI:
_cprint(f" {_DIM}✓ UI redrawn{_RST}")
elif canonical == "clear":
self.new_session(silent=True)
_clear_output_history()
# Clear terminal screen. Inside the TUI, Rich's console.clear()
# goes through patch_stdout's StdoutProxy which swallows the
# screen-clear escape sequences. Use prompt_toolkit's output
@ -11672,23 +11848,7 @@ class HermesCLI:
_original_on_resize = app._on_resize
def _resize_clear_ghosts():
renderer = app.renderer
try:
out = renderer.output
# Reset attributes, erase the entire screen, and home the
# cursor. This overwrites any reflowed status-bar rows or
# stale content the terminal kept from the prior layout.
out.reset_attributes()
out.erase_screen()
out.cursor_goto(0, 0)
out.flush()
# Tell the renderer its tracked position is fresh so its
# own erase() inside _on_resize doesn't cursor_up() past
# the top of the screen.
renderer.reset(leave_alternate_screen=False)
except Exception:
pass # never break resize handling
_original_on_resize()
self._schedule_resize_recovery(app, _original_on_resize)
app._on_resize = _resize_clear_ghosts

View file

@ -785,6 +785,11 @@ DEFAULT_CONFIG = {
"show_reasoning": False,
"streaming": False,
"final_response_markdown": "strip", # render | strip | raw
# Preserve recent classic CLI output across Ctrl+L, /redraw, and
# terminal resize full-screen clears. Disable if a terminal emulator
# behaves badly with replayed scrollback.
"persistent_output": True,
"persistent_output_max_lines": 200,
"inline_diffs": True, # Show inline diff previews for write actions (write_file, patch, skill_manage)
"show_cost": False, # Show $ cost in the status bar (off by default)
"skin": "default",

View file

@ -13,6 +13,7 @@ from unittest.mock import MagicMock
import pytest
import cli as cli_mod
from cli import HermesCLI
@ -33,10 +34,18 @@ class TestForceFullRedraw:
# Simulate HermesCLI before the TUI has ever been constructed.
bare_cli._force_full_redraw() # must not raise
def test_sends_full_clear_and_invalidates(self, bare_cli):
def test_sends_full_clear_replays_then_invalidates(self, bare_cli, monkeypatch):
app = MagicMock()
out = app.renderer.output
bare_cli._app = app
events = []
out.reset_attributes.side_effect = lambda: events.append("reset_attrs")
out.erase_screen.side_effect = lambda: events.append("erase")
out.cursor_goto.side_effect = lambda *_: events.append("home")
out.flush.side_effect = lambda: events.append("flush")
app.renderer.reset.side_effect = lambda **_: events.append("renderer_reset")
monkeypatch.setattr(cli_mod, "_replay_output_history", lambda: events.append("replay"))
app.invalidate.side_effect = lambda: events.append("invalidate")
bare_cli._force_full_redraw()
@ -52,6 +61,109 @@ class TestForceFullRedraw:
# Must schedule a repaint.
app.invalidate.assert_called_once()
assert events == [
"reset_attrs",
"erase",
"home",
"flush",
"renderer_reset",
"replay",
"invalidate",
]
def test_resize_rebuilds_scrollback_before_prompt_toolkit_redraw(self, bare_cli, monkeypatch):
app = MagicMock()
out = app.renderer.output
events = []
out.reset_attributes.side_effect = lambda: events.append("reset_attrs")
out.erase_screen.side_effect = lambda: events.append("erase")
out.write_raw.side_effect = lambda text: events.append(("raw", text))
out.cursor_goto.side_effect = lambda *_: events.append("home")
out.flush.side_effect = lambda: events.append("flush")
app.renderer.reset.side_effect = lambda **_: events.append("renderer_reset")
monkeypatch.setattr(cli_mod, "_replay_output_history", lambda: events.append("replay"))
original_on_resize = lambda: events.append("original_resize")
bare_cli._recover_after_resize(app, original_on_resize)
assert events == [
"reset_attrs",
"erase",
("raw", "\x1b[3J"),
"home",
"flush",
"renderer_reset",
"replay",
"original_resize",
]
app.invalidate.assert_not_called()
def test_force_redraw_uses_full_screen_clear_without_scrollback_clear(self, bare_cli):
app = MagicMock()
bare_cli._app = app
bare_cli._force_full_redraw()
app.renderer.output.erase_screen.assert_called_once()
app.renderer.output.cursor_goto.assert_called_once_with(0, 0)
app.renderer.output.write_raw.assert_not_called()
def test_resize_recovery_is_debounced(self, bare_cli, monkeypatch):
timers = []
calls = []
class FakeTimer:
def __init__(self, delay, callback):
self.delay = delay
self.callback = callback
self.cancelled = False
self.daemon = False
timers.append(self)
def start(self):
calls.append(("start", self.delay))
def cancel(self):
self.cancelled = True
calls.append(("cancel", self.delay))
def fire(self):
self.callback()
app = MagicMock()
app.loop.call_soon_threadsafe.side_effect = lambda cb: cb()
monkeypatch.setattr(cli_mod.threading, "Timer", FakeTimer)
monkeypatch.setattr(
bare_cli,
"_recover_after_resize",
lambda _app, _orig: calls.append(("recover", _orig())),
)
original_one = lambda: "first"
original_two = lambda: "second"
bare_cli._schedule_resize_recovery(app, original_one, delay=0.25)
assert bare_cli._resize_recovery_pending is True
bare_cli._schedule_resize_recovery(app, original_two, delay=0.25)
assert len(timers) == 2
assert timers[0].cancelled is True
timers[0].fire()
assert ("recover", "first") not in calls
timers[1].fire()
assert ("recover", "second") in calls
assert bare_cli._resize_recovery_pending is False
def test_invalidate_is_suppressed_while_resize_recovery_is_pending(self, bare_cli):
app = MagicMock()
bare_cli._app = app
bare_cli._last_invalidate = 0.0
bare_cli._resize_recovery_pending = True
bare_cli._invalidate(min_interval=0)
app.invalidate.assert_not_called()
def test_swallows_renderer_exceptions(self, bare_cli):
# If the renderer blows up for any reason, the helper must not

View file

@ -16,9 +16,18 @@ import sys
import types
from types import SimpleNamespace
import pytest
import cli
@pytest.fixture(autouse=True)
def reset_output_history():
cli._configure_output_history(False, 200)
yield
cli._configure_output_history(True, 200)
def test_cprint_no_app_direct_print(monkeypatch):
"""No active app → direct _pt_print, no run_in_terminal involvement."""
calls = []
@ -204,3 +213,69 @@ def test_cprint_swallows_prompt_toolkit_import_error(monkeypatch):
sys.meta_path.remove(blocker)
assert direct_prints == ["fallback2"]
def test_output_history_strips_ansi_and_keeps_recent_lines():
cli._configure_output_history(True, 10)
for idx in range(12):
cli._record_output_history(f"\x1b[31mline-{idx}\x1b[0m")
assert list(cli._OUTPUT_HISTORY) == [f"line-{idx}" for idx in range(2, 12)]
def test_replay_output_history_does_not_record_replayed_lines(monkeypatch):
cli._configure_output_history(True, 10)
cli._record_output_history("visible output")
printed = []
def _fake_print(value):
printed.append(value)
cli._record_output_history("duplicated replay")
monkeypatch.setattr(cli, "_pt_print", _fake_print)
monkeypatch.setattr(cli, "_PT_ANSI", lambda text: text)
cli._replay_output_history()
assert printed == ["visible output"]
assert list(cli._OUTPUT_HISTORY) == ["visible output"]
def test_replay_output_history_rerenders_callable_entries(monkeypatch):
cli._configure_output_history(True, 10)
widths_seen = []
printed = []
def _render_current_width():
widths_seen.append("called")
return ["top border", "body"]
cli._record_output_history_entry(_render_current_width)
monkeypatch.setattr(cli, "_pt_print", lambda value: printed.append(value))
monkeypatch.setattr(cli, "_PT_ANSI", lambda text: text)
cli._replay_output_history()
assert widths_seen == ["called"]
assert printed == ["top border", "body"]
assert list(cli._OUTPUT_HISTORY) == [_render_current_width]
def test_suspend_output_history_blocks_recording():
cli._configure_output_history(True, 10)
with cli._suspend_output_history():
cli._record_output_history("hidden")
cli._record_output_history_entry("also hidden")
assert list(cli._OUTPUT_HISTORY) == []
def test_clear_output_history_removes_replayable_lines():
cli._configure_output_history(True, 10)
cli._record_output_history("before clear")
cli._clear_output_history()
assert list(cli._OUTPUT_HISTORY) == []

View file

@ -11,6 +11,7 @@ from io import StringIO
from unittest.mock import MagicMock, patch
import pytest
import cli as cli_mod
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
@ -286,6 +287,21 @@ class TestDisplayResumedHistory:
assert "Previous Conversation" in output
def test_panel_is_stored_as_resize_aware_history_entry(self):
cli = _make_cli()
cli.conversation_history = _simple_history()
cli_mod._configure_output_history(True, 10)
cli_mod._clear_output_history()
try:
output = self._capture_display(cli)
assert "Previous Conversation" in output
assert len(cli_mod._OUTPUT_HISTORY) == 1
assert callable(cli_mod._OUTPUT_HISTORY[0])
finally:
cli_mod._configure_output_history(True, 200)
def test_assistant_with_no_content_no_tools_skipped(self):
"""Assistant messages with no visible output (e.g. pure reasoning)
are skipped in the recap."""