diff --git a/hermes_cli/cli_modal_mixin.py b/hermes_cli/cli_modal_mixin.py index 44b1f466d7..68b993d14e 100644 --- a/hermes_cli/cli_modal_mixin.py +++ b/hermes_cli/cli_modal_mixin.py @@ -547,33 +547,32 @@ class CLIModalMixin: or ``display.bell_on_complete`` (end of turn); works over SSH. The same flag also emits the OSC 9 / Warp OSC 777 desktop notification; ``context`` is the short notification body.""" flag = "bell_on_prompt" if prompt else "bell_on_complete" - if not getattr(self, flag, False): + if not getattr(self, flag, False) or getattr(self, "_terminal_io_broken", False): return - from hermes_cli.terminal_notify import notification_sequence, notify as _terminal_notify + from hermes_cli.cli_terminal_mixin import _run_on_app_loop, _write_terminal_sequence + from hermes_cli.terminal_notify import notification_sequence, write_tty 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 + try: + seq = "\a" + notification_sequence( + body, prompt=prompt, session_id=getattr(self, "session_id", "") or "", detail=detail) + except Exception: return - try: - sys.stdout.write("\a") - sys.stdout.flush() - except Exception: - pass - try: - _terminal_notify(body, prompt=prompt, session_id=session_id, detail=detail) - except Exception: - pass + app = getattr(self, "_app", None) + if app is None or not getattr(app, "_is_running", False): + write_tty(seq) + return + + # 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. + def _emit() -> None: + try: + _write_terminal_sequence(app, seq) + except (OSError, ValueError): + pass # dead tty: same fail-quiet as _pet_flush_kitty_frame + + _run_on_app_loop(app, _emit) def _clarify_teardown(self) -> None: self._clarify_state = None diff --git a/hermes_cli/terminal_notify.py b/hermes_cli/terminal_notify.py index 100ed064e8..db57eae7c8 100644 --- a/hermes_cli/terminal_notify.py +++ b/hermes_cli/terminal_notify.py @@ -4,9 +4,11 @@ OSC 9 (``ESC ] 9 ; BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; BEL``): Warp's structured CLI-agent protocol (tab status + notification mailbox). -Sequences are written to ``/dev/tty`` because prompt_toolkit's stdout wrapper can buffer or strip -raw escapes; when ``/dev/tty`` can't be opened (Windows, no controlling terminal) they fall back to -``sys.stdout``. Never raises. +Inside the running CLI, ``HermesCLI._ring_bell`` sends ``notification_sequence()`` through the +prompt_toolkit output on the app loop (a second writer on the tty would splice into an in-flight kitty +pet frame). ``write_tty`` is the no-app path: ``/dev/tty`` because ``patch_stdout``'s wrapper strips raw +escapes, falling back to ``sys.stdout`` when ``/dev/tty`` can't be opened (Windows, no controlling +terminal). Never raises. """ from __future__ import annotations @@ -23,7 +25,7 @@ _WARP_PROTOCOL_VERSION = 1 _WARP_LAST_BROKEN = {"stable": "v0.2026.03.25.08.24.stable_05", "preview": "v0.2026.03.25.08.24.preview_05"} -def _write_tty(seq: str) -> None: +def write_tty(seq: str) -> None: """Write raw escapes to /dev/tty, falling back to sys.stdout. Never raises.""" try: with open("/dev/tty", "w", encoding="utf-8") as tty: @@ -72,9 +74,3 @@ def notification_sequence(context: str, *, prompt: bool, session_id: str = "", d event = "permission_request" if prompt else "stop" seq += warp_osc777(event, detail or context, session_id) 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 e5a9092dc5..17305a48b0 100644 --- a/tests/hermes_cli/test_terminal_notify.py +++ b/tests/hermes_cli/test_terminal_notify.py @@ -1,8 +1,6 @@ """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 @@ -22,7 +20,7 @@ def _ring(monkeypatch, *, flag_on, env, **kwargs): for key, value in env.items(): monkeypatch.setenv(key, value) written = [] - monkeypatch.setattr(terminal_notify, "_write_tty", written.append) + monkeypatch.setattr(terminal_notify, "write_tty", written.append) cli = HermesCLI.__new__(HermesCLI) cli.bell_on_prompt = flag_on cli.session_id = "sess-1" @@ -32,7 +30,7 @@ def _ring(monkeypatch, *, flag_on, env, **kwargs): def test_osc9_body_emitted_and_sanitized_only_when_flag_on(monkeypatch): out = _ring(monkeypatch, flag_on=True, env={}, context="approval\x1b\x07\x00\x7f!") - assert out == "\x1b]9;Hermes: approval!\x07" + assert out == "\a\x1b]9;Hermes: approval!\x07" assert _ring(monkeypatch, flag_on=False, env={}, context="approval") == "" @@ -55,14 +53,11 @@ def test_warp_osc777_only_under_supported_warp_build(monkeypatch): 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.""" + """With the prompt_toolkit app live, the bell + notification must reach the tty through the + app's output ON THE APP LOOP, never via a second writer from the calling thread.""" 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) + monkeypatch.setattr(terminal_notify, "write_tty", lambda seq: pytest.fail(f"stray tty write: {seq!r}")) class _Output: raw = [] @@ -90,7 +85,7 @@ def test_running_app_gets_bell_and_osc9_on_its_loop_never_a_second_tty_writer(mo 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 _Output.raw == [] assert len(_Loop.queued) == 1 _Loop.queued[0]() assert _Output.raw == ["\a\x1b]9;Hermes: turn complete\x07", ""]