diff --git a/cli.py b/cli.py index ab2f2d3d7f96..ecaada574153 100644 --- a/cli.py +++ b/cli.py @@ -515,6 +515,7 @@ def load_cli_config() -> Dict[str, Any]: # enable when a terminal/tmux stack stamps stale prompt chrome into # scrollback during fullscreen/restore resizes. "cli_rebuild_scrollback_on_redraw": False, + "terminal_title": True, # Print a one-line summary of resolved modal prompts (approval / # clarify) into scrollback so the decision survives the repaint. "persist_prompts": True, @@ -5264,6 +5265,12 @@ def __init__( self.bell_on_prompt = CLI_CONFIG["display"].get("bell_on_prompt", False) # show_reasoning: display model thinking/reasoning before the response self.show_reasoning = CLI_CONFIG["display"].get("show_reasoning", True) + # The classic prompt_toolkit CLI owns its OSC title lifecycle. The Ink + # TUI manages its title in TypeScript, so this setting intentionally + # applies only to this surface. + self._terminal_title_enabled = bool( + CLI_CONFIG["display"].get("terminal_title", True) + ) # reasoning_full: when reasoning display is on, print the post-response # recap box uncollapsed instead of clamping to the first 10 lines. self.reasoning_full = CLI_CONFIG["display"].get("reasoning_full", False) @@ -5855,6 +5862,82 @@ def __init__( self._cache_hit_baseline_model: Optional[str] = None self._cache_hit_baseline_compressions = 0 + self._update_terminal_title() + + def _current_session_title(self) -> str: + """Return the current persisted or pending session title.""" + pending = getattr(self, "_pending_title", None) + if pending: + return str(pending) + session_db = getattr(self, "_session_db", None) + if session_db is None: + return "" + try: + return str(session_db.get_session_title(self.session_id) or "") + except Exception: + return "" + + def _response_panel_label(self, label: str) -> str: + """Append the active session name to a skin response label when present.""" + session_title = self._current_session_title() + base = (label or "⚕ Hermes").rstrip() + return f"{base} — {session_title}" if session_title else base + + def _update_terminal_title( + self, + *, + session_title: str | None = None, + expected_session_id: str | None = None, + ) -> None: + """Schedule a best-effort title update on prompt_toolkit's output loop.""" + if not getattr(self, "_terminal_title_enabled", False): + return + + def _write() -> None: + try: + # Auto-title callbacks run on a background thread. Check the + # captured session immediately before the write so a later + # session switch cannot let an old callback overwrite the tab. + if ( + expected_session_id is not None + and self.session_id != expected_session_id + ): + return + from hermes_cli.skin_engine import get_active_skin + from hermes_cli.terminal_title import ( + compose_terminal_title, + write_terminal_title, + ) + + label = get_active_skin().get_branding("response_label", "⚕ Hermes") + title = ( + self._current_session_title() + if session_title is None + else session_title + ) + write_terminal_title( + compose_terminal_title( + label, + title, + busy=bool(getattr(self, "_agent_running", False)), + ), + getattr(getattr(self, "_app", None), "output", None), + ) + except Exception: + pass + + app = getattr(self, "_app", None) + if app is None: + _write() + return + try: + app.loop.call_soon_threadsafe(_write) + except Exception: + # A live prompt_toolkit Output must only be touched by its owner + # loop. If that loop is unavailable during teardown, skip a + # cosmetic update rather than race the renderer. + pass + def _claim_active_session(self, surface: str = "cli", *, stderr: bool = False) -> bool: """Claim a global active-session slot for this CLI process.""" if self._active_session_lease is not None: @@ -8060,6 +8143,7 @@ def _on_thinking(self, text: str) -> None: if not text: self._flush_reasoning_preview(force=True) self._spinner_text = text or "" + self._update_terminal_title() self._tool_start_time = 0.0 # clear tool timer when switching to thinking self._invalidate() @@ -8497,6 +8581,7 @@ def _emit_stream_text(self, text: str) -> None: except Exception: label = "⚕ Hermes" _text_hex = "#FFF8DC" + label = self._response_panel_label(label) # Build a true-color ANSI escape for the response text color # so streamed content matches the Rich Panel appearance. try: @@ -10617,6 +10702,9 @@ def new_session(self, silent=False, title=None): print(f"(^_^)v New session started: {title}") else: print("(^_^)v New session started!") + update_terminal_title = getattr(self, "_update_terminal_title", None) + if callable(update_terminal_title): + update_terminal_title(session_title=title or "") def _consume_pending_resume_selection(self, text: str) -> bool: @@ -12730,6 +12818,7 @@ def process_command(self, command: str) -> bool: if self._session_db.set_session_title(self.session_id, new_title): self._status_bar_title_checked_at = 0.0 _cprint(f" Session title set: {new_title}") + self._update_terminal_title(session_title=new_title) else: _cprint(" Session not found in database.") except ValueError as e: @@ -12743,6 +12832,7 @@ def process_command(self, command: str) -> bool: else: self._pending_title = new_title _cprint(f" Session title queued: {new_title} (will be saved on first message)") + self._update_terminal_title(session_title=new_title) else: from hermes_state import format_session_db_unavailable _cprint(f" {format_session_db_unavailable()}") @@ -17185,6 +17275,12 @@ def display_callback(sentence: str): _streaming_box_opened = True w = self._scrollback_box_width(getattr(self.console, "width", 80)) label = " ⚕ Hermes " + try: + from hermes_cli.skin_engine import get_active_skin + label = get_active_skin().get_branding("response_label", "⚕ Hermes") + except Exception: + pass + label = self._response_panel_label(label) if self.show_timestamps: label = f"{label}{datetime.now().strftime(getattr(self, 'timestamp_format', '%H:%M'))} " fill = w - 2 - HermesCLI._status_bar_display_width(label) @@ -17623,6 +17719,7 @@ def run_agent(): label = "⚕ Hermes" _resp_color = _maybe_remap_for_light_mode("#CD7F32") _resp_text = _maybe_remap_for_light_mode("#FFF8DC") + label = self._response_panel_label(label) is_error_response = result and (result.get("failed") or result.get("partial")) already_streamed = self._stream_started and self._stream_box_opened and not is_error_response @@ -17642,7 +17739,7 @@ def run_agent(): _chat_console = ChatConsole() _chat_console.print(Panel( _render_final_assistant_content(response, mode=self.final_response_markdown), - title=f"[{_resp_color} bold]{label}[/]", + title=f"[{_resp_color} bold]{_escape(label)}[/]", title_align="left", border_style=_resp_color, style=_resp_text, @@ -17791,6 +17888,7 @@ def run_agent(): stop_event.set() if tts_thread is not None and tts_thread.is_alive(): tts_thread.join(timeout=5) + self._update_terminal_title() def _clear_terminal_on_exit(self): """Clear screen + scrollback so nothing is stranded above the exit summary. @@ -21193,6 +21291,7 @@ def process_loop(): # Regular chat - run agent self._agent_running = True + self._update_terminal_title() self._interactive_turn = True self._pet_turn_error = False self._pet_reasoning = False @@ -21204,6 +21303,7 @@ def process_loop(): finally: self._agent_running = False self._spinner_text = "" + self._update_terminal_title() self._tool_start_time = 0.0 self._pending_tool_info.clear() self._last_scrollback_tool = "" diff --git a/hermes_cli/cli_agent_setup_mixin.py b/hermes_cli/cli_agent_setup_mixin.py index 46eae97b0d04..a48c96188a71 100644 --- a/hermes_cli/cli_agent_setup_mixin.py +++ b/hermes_cli/cli_agent_setup_mixin.py @@ -465,6 +465,7 @@ def _init_agent(self, *, model_override: str = None, runtime_override: dict = No f"[bold red]Cannot resume session:[/] {_escape(resume_limit_error)}" ) return False + self._update_terminal_title(session_title=session_meta.get("title") or "") restored = self._session_db.get_messages_as_conversation( self.session_id, repair_alternation=True ) @@ -598,6 +599,14 @@ def _init_agent(self, *, model_override: str = None, runtime_override: dict = No # Route agent status output through prompt_toolkit so ANSI escape # sequences aren't garbled by patch_stdout's StdoutProxy (#2262). self.agent._print_fn = _cprint + # The shared turn prologue now creates session titles before the + # model runs. Keep the classic CLI's tab current by passing its + # UI-thread-safe writer to that common title callback. + _title_session_id = self.session_id + self.agent._on_session_title = lambda title, _source: self._update_terminal_title( + session_title=title, + expected_session_id=_title_session_id, + ) # Hydrate credits notices at session OPEN (parity with the TUI), so a # depletion / usage-band warning shows before the first message. The # notice_callback is bound above → _on_notice renders the line. Idempotent @@ -723,6 +732,7 @@ def _preload_resumed_session(self) -> bool: ) return False + self._update_terminal_title(session_title=session_meta.get("title") or "") model_history, display_history = self._session_db.get_resume_conversations(self.session_id) restored = model_history if restored: diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index b78ff6157c16..fcb6035c9b8e 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -1210,6 +1210,7 @@ def _handle_resume_command(self, cmd_original: str) -> None: self._resumed = True self._pending_title = None _sync_process_session_id(target_id) + self._update_terminal_title(session_title=session_meta.get("title") or "") # Load conversation history (strip transcript-only metadata entries). # repair_alternation: this /resume feeds LIVE REPLAY — ``restored`` @@ -1593,6 +1594,7 @@ def _handle_branch_command(self, cmd_original: str) -> None: self._pending_title = None self._resumed = True # Prevents auto-title generation _sync_process_session_id(new_session_id) + self._update_terminal_title(session_title=branch_title) # Sync the agent if self.agent: @@ -2439,11 +2441,12 @@ def _bg_thinking(text: str) -> None: label = "⚕ Hermes" _resp_color = "#CD7F32" _resp_text = "#FFF8DC" + label = self._response_panel_label(label) _chat_console = ChatConsole() _chat_console.print(Panel( _render_final_assistant_content(response, mode=self.final_response_markdown), - title=f"[{_resp_color} bold]{label} (background #{task_num})[/]", + title=f"[{_resp_color} bold]{_escape(label)} (background #{task_num})[/]", title_align="left", border_style=_resp_color, style=_resp_text, diff --git a/hermes_cli/config_defaults.py b/hermes_cli/config_defaults.py index 6cfc67692f94..bfac511a6d5c 100644 --- a/hermes_cli/config_defaults.py +++ b/hermes_cli/config_defaults.py @@ -1481,6 +1481,10 @@ "bell_on_complete": False, # Bell when a blocking prompt opens (clarify/approval/sudo/secret). "bell_on_prompt": False, + # Keep a classic-CLI terminal tab/window title in sync with the active + # session. Set false for terminals or multiplexers where OSC titles are + # undesirable. The Ink TUI manages its own title independently. + "terminal_title": True, # Stream the model's reasoning/thinking live before the response. # Default ON: on thinking models the reasoning phase can run tens of # seconds, and with this off the user stares at a spinner the whole diff --git a/hermes_cli/terminal_title.py b/hermes_cli/terminal_title.py new file mode 100644 index 000000000000..fda3d72cc9e0 --- /dev/null +++ b/hermes_cli/terminal_title.py @@ -0,0 +1,105 @@ +"""Best-effort terminal tab and window title updates for the classic CLI.""" + +from __future__ import annotations + +import os +import re +import sys +import threading + + +_CONTROL_CHARS = re.compile(r"[\x00-\x1f\x7f-\x9f]") +_MAX_TITLE_LENGTH = 200 +_WRITE_LOCK = threading.Lock() + + +def sanitize_terminal_title(value: object) -> str: + """Return a printable, bounded title that cannot inject terminal escapes.""" + text = _CONTROL_CHARS.sub("", str(value or "")) + return " ".join(text.split())[:_MAX_TITLE_LENGTH] + + +def terminal_title_symbol(response_label: object, fallback: str = "⚕") -> str: + """Extract the skin's leading symbol from its response-panel label.""" + label = sanitize_terminal_title(response_label) + return label.split(maxsplit=1)[0] if label else fallback + + +def compose_terminal_title( + response_label: object, + session_title: object = "", + *, + busy: bool = False, +) -> str: + """Compose the short tab title for an idle or active classic CLI session.""" + parts = [terminal_title_symbol(response_label)] + title = sanitize_terminal_title(session_title) + if title: + parts.append(title) + if busy: + parts.append("⏳") + return " ".join(parts) + + +def _set_windows_console_title(title: str) -> bool: + """Set the native Windows console title without relying on OSC support.""" + try: + import ctypes + + return bool(ctypes.windll.kernel32.SetConsoleTitleW(title)) + except Exception: + return False + + +def _is_interactive_output(output: object) -> bool: + """Check the underlying stream for prompt_toolkit Output instances.""" + stream = getattr(output, "stdout", output) + try: + return bool(stream.isatty()) + except Exception: + return False + + +def _write_osc_terminal_title(output: object, title: str) -> bool: + """Write OSC title sequences, returning whether the terminal accepted them.""" + try: + with _WRITE_LOCK: + sequence = f"\033]1;{title}\a\033]2;{title}\a" + write_raw = getattr(output, "write_raw", None) + if callable(write_raw): + write_raw(sequence) + else: + output.write(sequence) + output.flush() + return True + except Exception: + return False + + +def write_terminal_title(title: object, output: object | None = None) -> bool: + """Set an interactive terminal's tab and window title. + + On Windows, this uses ``SetConsoleTitleW`` for classic conhost and also + writes OSC 1/2 for terminal emulators such as mintty and VS Code. Elsewhere, + OSC 1 updates a terminal icon/tab label and OSC 2 updates the window title. + Prompt_toolkit ``Output`` objects are supported so callers can bypass + ``patch_stdout`` safely. The writer deliberately avoids logging failures + because it may be called from an agent callback. + """ + if os.environ.get("TERM", "").lower() == "dumb": + return False + + try: + output = output if output is not None else sys.stdout + if output is None or not _is_interactive_output(output): + return False + clean_title = sanitize_terminal_title(title) + if not clean_title: + return False + if sys.platform == "win32": + native_updated = _set_windows_console_title(clean_title) + osc_updated = _write_osc_terminal_title(output, clean_title) + return native_updated or osc_updated + return _write_osc_terminal_title(output, clean_title) + except Exception: + return False diff --git a/tests/hermes_cli/test_terminal_title.py b/tests/hermes_cli/test_terminal_title.py new file mode 100644 index 000000000000..79194c1fe04a --- /dev/null +++ b/tests/hermes_cli/test_terminal_title.py @@ -0,0 +1,184 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from cli import HermesCLI +from hermes_cli.terminal_title import ( + compose_terminal_title, + sanitize_terminal_title, + write_terminal_title, +) +import hermes_cli.terminal_title as terminal_title + + +class _Terminal: + def __init__(self, *, isatty: bool = True) -> None: + self._isatty = isatty + self.writes: list[str] = [] + self.flushed = False + + def isatty(self) -> bool: + return self._isatty + + def write(self, value: str) -> None: + self.writes.append(value) + + def flush(self) -> None: + self.flushed = True + + +class _PromptToolkitOutput: + """Minimal prompt_toolkit Output stand-in backed by a real terminal.""" + + def __init__(self, stdout: _Terminal) -> None: + self.stdout = stdout + self.raw_writes: list[str] = [] + self.flushed = False + + def write_raw(self, value: str) -> None: + self.raw_writes.append(value) + + def flush(self) -> None: + self.flushed = True + + +class _ScheduledLoop: + """Capture callbacks so tests can control their event-loop ordering.""" + + def __init__(self) -> None: + self.callbacks = [] + + def call_soon_threadsafe(self, callback) -> None: + self.callbacks.append(callback) + + +def test_compose_terminal_title_uses_skin_symbol_session_and_busy_marker(): + assert compose_terminal_title(" ⚔ Ares ", "Release prep") == "⚔ Release prep" + assert compose_terminal_title(" ⚔ Ares ", "Release prep", busy=True) == "⚔ Release prep ⏳" + assert compose_terminal_title(" ⚔ Ares ") == "⚔" + + +def test_sanitize_terminal_title_removes_terminal_control_sequences_and_bounds_length(): + title = sanitize_terminal_title("Build\x1b]2;injected\a\nready") + + assert title == "Build]2;injectedready" + assert len(sanitize_terminal_title("x" * 201)) == 200 + + +def test_write_terminal_title_emits_icon_and_window_sequences(monkeypatch): + monkeypatch.setenv("TERM", "xterm-256color") + terminal = _Terminal() + + assert write_terminal_title("⚕ Planning", terminal) + assert terminal.writes == ["\033]1;⚕ Planning\a\033]2;⚕ Planning\a"] + assert terminal.flushed + + +def test_write_terminal_title_uses_prompt_toolkit_raw_output(monkeypatch): + monkeypatch.setenv("TERM", "xterm-256color") + terminal = _Terminal() + output = _PromptToolkitOutput(terminal) + + assert write_terminal_title("⚕ Planning", output) + assert output.raw_writes == ["\033]1;⚕ Planning\a\033]2;⚕ Planning\a"] + assert output.flushed + assert terminal.writes == [] + + +def test_write_terminal_title_updates_windows_console_and_osc_terminals(monkeypatch): + monkeypatch.setenv("TERM", "xterm-256color") + monkeypatch.setattr(terminal_title.sys, "platform", "win32") + titles: list[str] = [] + monkeypatch.setattr( + terminal_title, + "_set_windows_console_title", + lambda title: titles.append(title) or True, + ) + terminal = _Terminal() + + assert write_terminal_title("⚕ Planning", terminal) + assert titles == ["⚕ Planning"] + assert terminal.writes == ["\033]1;⚕ Planning\a\033]2;⚕ Planning\a"] + assert terminal.flushed + + +def test_write_terminal_title_skips_dumb_and_noninteractive_output(monkeypatch): + terminal = _Terminal() + monkeypatch.setenv("TERM", "dumb") + assert not write_terminal_title("Hermes", terminal) + + monkeypatch.setenv("TERM", "xterm-256color") + assert not write_terminal_title("Hermes", _Terminal(isatty=False)) + + +def test_cli_terminal_title_writes_on_the_prompt_toolkit_event_loop(monkeypatch): + loop = _ScheduledLoop() + cli = HermesCLI.__new__(HermesCLI) + cli._terminal_title_enabled = True + cli._agent_running = False + cli._app = SimpleNamespace(loop=loop, output=object()) + cli._current_session_title = lambda: "Current session" + cli.session_id = "session-a" + writes = [] + monkeypatch.setattr( + terminal_title, + "write_terminal_title", + lambda title, output: writes.append((title, output)), + ) + + cli._update_terminal_title() + + assert writes == [] + assert len(loop.callbacks) == 1 + loop.callbacks.pop()() + assert writes == [("⚕ Current session", cli._app.output)] + + +def test_cli_auto_title_skips_a_session_that_changed_before_the_scheduled_write(monkeypatch): + loop = _ScheduledLoop() + cli = HermesCLI.__new__(HermesCLI) + cli._terminal_title_enabled = True + cli._agent_running = False + cli._app = SimpleNamespace(loop=loop, output=object()) + cli._current_session_title = lambda: "Current session" + cli.session_id = "session-a" + writes = [] + monkeypatch.setattr( + terminal_title, + "write_terminal_title", + lambda title, output: writes.append((title, output)), + ) + + cli._update_terminal_title( + session_title="Generated A", + expected_session_id="session-a", + ) + cli.session_id = "session-b" + loop.callbacks.pop()() + + assert writes == [] + + +def test_cli_auto_title_uses_busy_state_when_the_scheduled_write_runs(monkeypatch): + loop = _ScheduledLoop() + cli = HermesCLI.__new__(HermesCLI) + cli._terminal_title_enabled = True + cli._agent_running = False + cli._app = SimpleNamespace(loop=loop, output=object()) + cli._current_session_title = lambda: "Current session" + cli.session_id = "session-a" + writes = [] + monkeypatch.setattr( + terminal_title, + "write_terminal_title", + lambda title, output: writes.append((title, output)), + ) + + cli._update_terminal_title( + session_title="Generated A", + expected_session_id="session-a", + ) + cli._agent_running = True + loop.callbacks.pop()() + + assert writes == [("⚕ Generated A ⏳", cli._app.output)]