Skip to content
Closed
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
7 changes: 7 additions & 0 deletions acp_adapter/__main__.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,12 @@
"""Allow running the ACP adapter as ``python -m acp_adapter``."""

import os

# ACP mode owns stdout for JSON-RPC and stderr for logs; the controlling editor
# draws its own chrome. Suppress OSC-based tab-title emission from any code path
# that would otherwise write escape sequences to the TTY.
os.environ.setdefault("HERMES_DISABLE_TAB_TITLE", "1")

from .entry import main

main()
13 changes: 13 additions & 0 deletions agent/context_engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,6 +124,19 @@ def on_session_reset(self) -> None:
self.last_total_tokens = 0
self.compression_count = 0

# -- Optional: per-turn context recall ---------------------------------

def prefetch(self, query: str, **kwargs) -> str:
"""Return ephemeral context to inject into the current user turn.

This is called once per user turn before the tool-calling loop.
The returned text is injected at API-call time only and is NOT
persisted to the session transcript. Engines can use this to surface
domain-specific operating context without mutating the stable system
prompt or waiting for compression to fire.
"""
return ""

# -- Optional: tools ---------------------------------------------------

def get_tool_schemas(self) -> List[Dict[str, Any]]:
Expand Down
10 changes: 10 additions & 0 deletions agent/title_generator.py
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,16 @@ def auto_title_session(
logger.debug("Auto-generated session title: %s", title)
except Exception as e:
logger.debug("Failed to set auto-generated title: %s", e)
return

# Refresh the terminal tab title so the new auto-generated title shows up
# immediately. No-op in non-CLI contexts (gateway, ACP, etc.) per the
# module's TTY / env guards.
try:
from hermes_cli.terminal_title import update_for_session
update_for_session(session_id)
except Exception:
pass


def maybe_auto_title(
Expand Down
21 changes: 20 additions & 1 deletion cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -2005,7 +2005,16 @@ def __init__(
timestamp_str = self.session_start.strftime("%Y%m%d_%H%M%S")
short_uuid = uuid.uuid4().hex[:6]
self.session_id = f"{timestamp_str}_{short_uuid}"


# Terminal tab title: "<Persona>: <session title>" + OSC 7 cwd.
# Safe no-op on non-TTY / TERM=dumb / HERMES_DISABLE_TAB_TITLE=1.
try:
from hermes_cli.terminal_title import set_cwd, update_for_session
set_cwd()
update_for_session(self.session_id)
except Exception:
pass

# History file for persistent input recall across sessions
self._history_file = _hermes_home / ".hermes_history"
self._last_invalidate: float = 0.0 # throttle UI repaints
Expand Down Expand Up @@ -3311,6 +3320,11 @@ def _init_agent(self, *, model_override: str = None, runtime_override: dict = No
except (ValueError, Exception) as e:
_cprint(f" Could not apply pending title: {e}")
self._pending_title = None
try:
from hermes_cli.terminal_title import update_for_session
update_for_session(self.session_id)
except Exception:
pass
return True
except Exception as e:
ChatConsole().print(f"[bold red]Failed to initialize agent: {e}[/]")
Expand Down Expand Up @@ -5942,6 +5956,11 @@ def process_command(self, command: str) -> bool:
try:
if self._session_db.set_session_title(self.session_id, new_title):
_cprint(f" Session title set: {new_title}")
try:
from hermes_cli.terminal_title import update_for_session
update_for_session(self.session_id)
except Exception:
pass
else:
_cprint(" Session not found in database.")
except ValueError as e:
Expand Down
207 changes: 207 additions & 0 deletions hermes_cli/terminal_title.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,207 @@
"""Emit OSC escape sequences so the terminal tab shows: "<persona>: <session title>".

Drop this file in as ``hermes_cli/terminal_title.py`` in the Hermes Agent repo and
wire it into the chat loop (see WIRING.md in this directory).

Format emitted (OSC 0 — sets both window title and icon title):

ESC ] 0 ; <persona>: <session title> BEL

Plus OSC 7 for working directory, which Warp / iTerm2 / Kitty / WezTerm use to
display the cwd subtitle:

ESC ] 7 ; file://<host><abs-cwd> BEL

No-ops gracefully when stderr/stdout aren't TTYs (pipes, captured output, ACP
JSON-RPC mode), when ``TERM=dumb``, or when ``HERMES_DISABLE_TAB_TITLE=1``.
"""
from __future__ import annotations

import atexit
import os
import re
import socket
import sys
from pathlib import Path
from typing import Optional
from urllib.parse import quote

# OSC framing. BEL terminator works everywhere; ST (\x1b\\) is the formal one.
_OSC = "\x1b]"
_BEL = "\x07"

_MAX_TITLE_LEN = 120
_DEFAULT_PERSONA = "Hermes"

# Strip everything that could break the OSC frame or look like garbage in a tab.
_UNSAFE = re.compile(r"[\x00-\x1f\x7f]")

_prior_title_set = False # whether we've stored anything to restore on exit


# ---------------------------------------------------------------------------
# Output channel
# ---------------------------------------------------------------------------

def _tty_stream():
"""Return a writable TTY stream, or None if we shouldn't emit.

Prefers ``/dev/tty`` so the escape lands on the real terminal even when
stdout/stderr are redirected. Falls back to stderr if it's a tty.
"""
if os.environ.get("HERMES_DISABLE_TAB_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.

This opens the controlling terminal even when stdout and stderr are pipes, so the implementation does not satisfy the stated no-op behavior for captured/piped output. Gate this on the intended interactive condition, or explicitly document and test the controlling-TTY behavior.

return None
term = os.environ.get("TERM", "")
if term in ("", "dumb"):
return None
# /dev/tty is the controlling terminal regardless of stdio redirection.
try:
return open("/dev/tty", "w", buffering=1, encoding="utf-8")
except OSError:
pass
if sys.stderr.isatty():
return sys.stderr
return None


def _write(seq: str) -> None:
stream = _tty_stream()
if stream is None:
return
try:
stream.write(seq)
stream.flush()
except Exception:
pass
finally:
# If we opened /dev/tty, close it; don't close stderr.
if stream is not sys.stderr:
try:
stream.close()
except Exception:
pass


# ---------------------------------------------------------------------------
# Sanitisation
# ---------------------------------------------------------------------------

def _clean(text: str, max_len: int = _MAX_TITLE_LEN) -> str:
if not text:
return ""
text = _UNSAFE.sub("", text).strip()
if len(text) > max_len:
text = text[: max_len - 1].rstrip() + "\u2026" # …
return text


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def set_tab_title(persona: Optional[str], session_title: Optional[str]) -> None:
"""Set the terminal tab/window title to ``"<persona>: <session_title>"``.

Empty / missing values are tolerated. Called frequently; cheap and safe.
"""
persona_clean = _clean(persona or _DEFAULT_PERSONA, max_len=40)
title_clean = _clean(session_title or "", max_len=_MAX_TITLE_LEN - len(persona_clean) - 2)

if title_clean:
full = f"{persona_clean}: {title_clean}"
else:
full = persona_clean

global _prior_title_set
_prior_title_set = True
# OSC 0 sets both window and icon title at once.
_write(f"{_OSC}0;{full}{_BEL}")


def set_cwd(path: Optional[str | os.PathLike] = None) -> None:
"""Emit OSC 7 so the terminal knows the current working directory."""
p = Path(path or os.getcwd()).resolve()
try:
host = socket.gethostname()
except Exception:
host = ""
# Path components must be percent-encoded (per RFC 3986); leave the slashes.
encoded = quote(str(p), safe="/")
_write(f"{_OSC}7;file://{host}{encoded}{_BEL}")


def reset_tab_title() -> None:
"""Clear our title; most terminals will fall back to the shell's default."""
if _prior_title_set:
_write(f"{_OSC}0;{_BEL}")


# Restore on normal interpreter exit.
atexit.register(reset_tab_title)


# ---------------------------------------------------------------------------
# Persona / title resolution helpers (callers can use or ignore)
# ---------------------------------------------------------------------------

_PERSONA_NAME_RX = re.compile(r"^\s*-?\s*Name\s*:\s*(.+?)\s*$", re.MULTILINE | re.IGNORECASE)
_PERSONA_HEADING_RX = re.compile(r"^\s*#\s+(.+?)\s*$", re.MULTILINE)


def resolve_persona_name(soul_path: Optional[Path] = None,
config_value: Optional[str] = None) -> str:
"""Best-effort persona name.

Resolution order:
1. Explicit ``config_value`` (caller-provided, e.g. ``agent.persona_name``).
2. ``Name: <X>`` line in SOUL.md.
3. First markdown ``# heading`` in SOUL.md.
4. ``"Hermes"`` fallback.
"""
if config_value:
cleaned = _clean(config_value, max_len=40)
if cleaned:
return cleaned

if soul_path is None:
home = Path(os.environ.get("HERMES_HOME", Path.home() / ".hermes"))
soul_path = home / "SOUL.md"

try:
text = soul_path.read_text(encoding="utf-8", errors="ignore")
except OSError:
return _DEFAULT_PERSONA

m = _PERSONA_NAME_RX.search(text)
if m:
return _clean(m.group(1), max_len=40) or _DEFAULT_PERSONA

m = _PERSONA_HEADING_RX.search(text)
if m:
return _clean(m.group(1), max_len=40) or _DEFAULT_PERSONA

return _DEFAULT_PERSONA


def resolve_session_title(session_id: str) -> Optional[str]:
"""Look up the session title from the Hermes SQLite store, or return None."""
try:
from hermes_state import SessionDB # type: ignore[import-not-found]
from hermes_constants import get_hermes_home # type: ignore[import-not-found]
except Exception:
return None
try:
db = SessionDB(db_path=get_hermes_home() / "state.db")
return db.get_session_title(session_id)
except Exception:
return None


# Convenience one-liner for callers that already know both.
def update_for_session(session_id: str,
persona_override: Optional[str] = None,
title_override: Optional[str] = None) -> None:
"""Resolve persona + title (with optional overrides) and emit."""

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.

update_for_session() never loads or passes agent.persona_name, so this always skips the first advertised resolution source. Please retrieve the configured value here (or at a single shared call site) and add an integration test that proves it reaches the OSC payload.

persona = persona_override or resolve_persona_name()
title = title_override or resolve_session_title(session_id) or ""
set_tab_title(persona, title)
Loading