Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
102 changes: 101 additions & 1 deletion cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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:
Expand All @@ -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()}")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand All @@ -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 = ""
Expand Down
10 changes: 10 additions & 0 deletions hermes_cli/cli_agent_setup_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down
5 changes: 4 additions & 1 deletion hermes_cli/cli_commands_mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -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``
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 4 additions & 0 deletions hermes_cli/config_defaults.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
105 changes: 105 additions & 0 deletions hermes_cli/terminal_title.py
Original file line number Diff line number Diff line change
@@ -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:

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The classic interactive loop is inside patch_stdout (cli.py:17185), and cli.py:3060 documents that raw ANSI written to its StdoutProxy is swallowed. Please write this sequence through the real terminal/raw prompt_toolkit output path and add a regression test; otherwise the title lifecycle can be invisible in the primary classic-CLI path.

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
Loading