diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index c2c940da0d..44b1f466d7 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -549,18 +549,29 @@ class CLIModalMixin: flag = "bell_on_prompt" if prompt else "bell_on_complete" if not getattr(self, flag, False): return + from hermes_cli.terminal_notify import notification_sequence, notify as _terminal_notify + body = context or ("input needed" if prompt else "turn complete") + session_id = getattr(self, "session_id", "") or "" + app = getattr(self, "_app", None) + if app is not None and getattr(app, "_is_running", False): + # Agent thread. The loop thread may be mid-write of a 12 KB kitty pet frame that the tty + # drains ~1 KB at a time; a second writer on the same tty (/dev/tty, sys.stdout) splices + # in, the foreign ESC aborts the APC, and the terminal paints the rest of the payload as + # base64 at the input cursor. Serialize behind the renderer instead. + from hermes_cli.cli_terminal_mixin import _run_on_app_loop, _write_terminal_sequence + try: + seq = "\a" + notification_sequence(body, prompt=prompt, session_id=session_id, detail=detail) + _run_on_app_loop(app, lambda: _write_terminal_sequence(app, seq)) + except Exception: + pass + return try: sys.stdout.write("\a") sys.stdout.flush() except Exception: pass try: - from hermes_cli.terminal_notify import notify as _terminal_notify - _terminal_notify( - context or ("input needed" if prompt else "turn complete"), - prompt=prompt, - session_id=getattr(self, "session_id", "") or "", - detail=detail) + _terminal_notify(body, prompt=prompt, session_id=session_id, detail=detail) except Exception: pass diff --git a/hermes_cli/terminal_notify.py b/hermes_cli/terminal_notify.py index 4b370ef8d3..100ed064e8 100644 --- a/hermes_cli/terminal_notify.py +++ b/hermes_cli/terminal_notify.py @@ -65,10 +65,16 @@ def warp_osc777(event: str, detail: str, session_id: str = "") -> str: return f"\x1b]777;notify;warp://cli-agent;{json.dumps(payload, separators=(',', ':'))}\x07" -def notify(context: str, *, prompt: bool, session_id: str = "", detail: str = "") -> None: - """Emit OSC 9 (plus Warp OSC 777 when supported) for a blocking prompt or turn end.""" +def notification_sequence(context: str, *, prompt: bool, session_id: str = "", detail: str = "") -> str: + """OSC 9 (plus Warp OSC 777 when supported) for a blocking prompt or turn end.""" seq = osc9(f"Hermes: {context}") if warp_supported(): event = "permission_request" if prompt else "stop" seq += warp_osc777(event, detail or context, session_id) - _write_tty(seq) + return seq + + +def notify(context: str, *, prompt: bool, session_id: str = "", detail: str = "") -> None: + """Emit the notification straight to the tty. Only for callers that do not own a running + prompt_toolkit app; inside the CLI, ``_ring_bell`` routes it through the app output instead.""" + _write_tty(notification_sequence(context, prompt=prompt, session_id=session_id, detail=detail)) diff --git a/tests/hermes_cli/test_terminal_notify.py b/tests/hermes_cli/test_terminal_notify.py index fce8b90d5e..e5a9092dc5 100644 --- a/tests/hermes_cli/test_terminal_notify.py +++ b/tests/hermes_cli/test_terminal_notify.py @@ -1,6 +1,10 @@ """display.bell_on_prompt / bell_on_complete also drive OSC 9 + Warp OSC 777 via _ring_bell.""" +import io import json +import sys + +import pytest from cli import HermesCLI from hermes_cli import terminal_notify @@ -48,3 +52,45 @@ def test_warp_osc777_only_under_supported_warp_build(monkeypatch): # Not Warp at all → OSC 9 only. not_warp = dict(_WARP_OK, TERM_PROGRAM="ghostty") assert prefix not in _ring(monkeypatch, flag_on=True, env=not_warp, context="approval") + + +def test_running_app_gets_bell_and_osc9_on_its_loop_never_a_second_tty_writer(monkeypatch): + """With the prompt_toolkit app live, the notification must reach the tty through the app's + output ON THE APP LOOP. A parallel /dev/tty or sys.stdout write from the agent thread splices + into an in-flight kitty pet frame and the terminal paints the frame's base64 as text.""" + for key in _WARP_OK: + monkeypatch.delenv(key, raising=False) + monkeypatch.setattr(terminal_notify, "_write_tty", lambda seq: pytest.fail(f"stray tty write: {seq!r}")) + fake_stdout = io.StringIO() + monkeypatch.setattr(sys, "stdout", fake_stdout) + + class _Output: + raw = [] + + def write_raw(self, data): + self.raw.append(data) + + def flush(self): + self.raw.append("") + + class _Loop: + queued = [] + + def call_soon_threadsafe(self, fn): + self.queued.append(fn) + + class _App: + _is_running = True + loop = _Loop() + output = _Output() + + cli = HermesCLI.__new__(HermesCLI) + cli.bell_on_complete = True + cli.session_id = "sess-1" + cli._app = _App() + cli._ring_bell(context="turn complete") + # Nothing touched the tty from the calling thread; the write is queued for the loop. + assert _Output.raw == [] and fake_stdout.getvalue() == "" + assert len(_Loop.queued) == 1 + _Loop.queued[0]() + assert _Output.raw == ["\a\x1b]9;Hermes: turn complete\x07", ""]