diff --git a/cli.py b/cli.py index 9f86e3e3a4..30b33001c7 100644 --- a/cli.py +++ b/cli.py @@ -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 diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 030421c90c..89397b1cb5 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -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", diff --git a/tests/cli/test_cli_force_redraw.py b/tests/cli/test_cli_force_redraw.py index 24d787c24e..4c7197ad94 100644 --- a/tests/cli/test_cli_force_redraw.py +++ b/tests/cli/test_cli_force_redraw.py @@ -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 diff --git a/tests/cli/test_cprint_bg_thread.py b/tests/cli/test_cprint_bg_thread.py index 3b5db53492..bb0e59d064 100644 --- a/tests/cli/test_cprint_bg_thread.py +++ b/tests/cli/test_cprint_bg_thread.py @@ -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) == [] diff --git a/tests/cli/test_resume_display.py b/tests/cli/test_resume_display.py index bb931bb1fe..ffeb4402cd 100644 --- a/tests/cli/test_resume_display.py +++ b/tests/cli/test_resume_display.py @@ -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."""