fef0bc56b2
Review fold: build "\a" + sequence once and dispatch it to exactly one sink (app loop when the Application runs, write_tty otherwise) instead of two branches with two writes and two try/excepts. terminal_notify.notify() had a single caller (that fallback), so write_tty is the public no-app entry and notify() is gone. The loop-side write keeps the never-raises contract on a dead tty (EIO/closed file) the same way _pet_flush_kitty_frame does, and _ring_bell skips entirely once _terminal_io_broken is set. pty A/B re-run on this stack: 400 rings vs 90 frames, 0 aborted, 0 painted, 400/400 delivered.
77 lines
3.2 KiB
Python
77 lines
3.2 KiB
Python
"""Terminal-native desktop notifications: OSC 9 and Warp's OSC 777 CLI-agent protocol.
|
|
|
|
OSC 9 (``ESC ] 9 ; <body> BEL``): Ghostty, iTerm2, Kitty and WezTerm raise an OS notification;
|
|
others drop it. OSC 777 (``ESC ] 777 ; notify ; warp://cli-agent ; <json> BEL``): Warp's
|
|
structured CLI-agent protocol (tab status + notification mailbox).
|
|
|
|
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
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
_C0_AND_DEL = re.compile(r"[\x00-\x1f\x7f]")
|
|
_WARP_PROTOCOL_VERSION = 1
|
|
# Last Warp release per channel that set WARP_CLI_AGENT_PROTOCOL_VERSION but could not render
|
|
# structured payloads (Warp's should-use-structured.sh). Bash compares lexicographically; so do we.
|
|
_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:
|
|
"""Write raw escapes to /dev/tty, falling back to sys.stdout. Never raises."""
|
|
try:
|
|
with open("/dev/tty", "w", encoding="utf-8") as tty:
|
|
tty.write(seq)
|
|
return
|
|
except OSError:
|
|
pass
|
|
try:
|
|
sys.stdout.write(seq)
|
|
sys.stdout.flush()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def osc9(body: str) -> str:
|
|
"""OSC 9 sequence with C0 controls and DEL stripped from the body."""
|
|
return f"\x1b]9;{_C0_AND_DEL.sub('', body)}\x07"
|
|
|
|
|
|
def warp_supported(env=None) -> bool:
|
|
"""True when running in a Warp build that can render OSC 777 agent payloads."""
|
|
env = os.environ if env is None else env
|
|
client = env.get("WARP_CLIENT_VERSION", "")
|
|
if env.get("TERM_PROGRAM") != "WarpTerminal" or not env.get("WARP_CLI_AGENT_PROTOCOL_VERSION") or not client:
|
|
return False
|
|
return not any(channel in client and client <= last_broken for channel, last_broken in _WARP_LAST_BROKEN.items())
|
|
|
|
|
|
def warp_osc777(event: str, detail: str, session_id: str = "") -> str:
|
|
"""OSC 777 ``warp://cli-agent`` notification; ``event`` is ``stop`` or ``permission_request``."""
|
|
try:
|
|
advertised = int(os.environ.get("WARP_CLI_AGENT_PROTOCOL_VERSION", "1"))
|
|
except ValueError:
|
|
advertised = 1
|
|
cwd = os.getcwd()
|
|
payload = {"v": min(advertised, _WARP_PROTOCOL_VERSION), "agent": "hermes", "event": event,
|
|
"session_id": session_id, "cwd": cwd, "project": os.path.basename(cwd)}
|
|
payload["summary" if event == "permission_request" else "response"] = detail[:200]
|
|
return f"\x1b]777;notify;warp://cli-agent;{json.dumps(payload, separators=(',', ':'))}\x07"
|
|
|
|
|
|
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)
|
|
return seq
|