Compare commits

...

5 Commits

Author SHA1 Message Date
Brooklyn Nicholson
82b9c44cbd fix(notify): restrict OSC to native-rendering terminals; hint osascript perms
Research finding: terminfo.dev "support" for OSC 9/777 only means the parser
consumes the sequence — VS Code/Cursor and Apple Terminal silently drop it
without rendering anything (microsoft/vscode#294247, anthropics/claude-code#28338).
Emitting OSC there made notifications no-op AND skipped the OS fallback.

- _detect_terminal_osc now returns a flavor only for terminals that actually
  render: iTerm2, Ghostty, kitty, WezTerm. Everything else (VS Code/Cursor,
  Apple Terminal, unknown) falls through to the osascript path. VS Code/Cursor
  users wanting click-to-focus can install the "Terminal Notification"
  extension, which parses the OSC we already emit — documented, not assumed.
- Add a one-time WARNING when the osascript fallback runs: on macOS Sequoia+,
  osascript notifications are attributed to "Script Editor" and silently
  dropped (exit 0, nothing shown) until the user grants Script Editor
  notification permission once. The hint spells out the fix so users aren't
  stuck staring at a no-op.
2026-06-18 15:33:00 -05:00
Brooklyn Nicholson
b59b1cfb12 feat(notify): terminal-native OSC notifications as primary path
osascript is a dead end for an unsigned CLI on modern macOS (notifications
permanently attributed to "Script Editor"; Apple removed sender override in
Monterey) and terminal-notifier is broken on recent releases. The reliable
approach used across CLI notifier projects is to let the terminal emulator
raise the banner itself via OSC escape sequences — attributed to the terminal
the user already trusts, click focuses it, zero dependencies.

- Emit OSC 9 (iTerm2-style) or OSC 777 (urxvt-style, title+body) to
  /dev/tty — picked per terminal via TERM_PROGRAM/env so terminals that
  support both don't double-fire. Write to /dev/tty (not stdout, which the
  TUI/slash-worker capture); the sequences are non-rendering so they don't
  disturb a live TUI.
- Works in iTerm2, Ghostty, kitty, WezTerm, Warp, VS Code, Cursor. Apple
  Terminal and unknown terminals return False and fall back to the existing
  OS-level path (notify-send / terminal-notifier / osascript / PowerShell).
- tmux passthrough wrapping when $TMUX is set.
- Add `import os` (the module didn't import it before; the new env reads need
  it).
- Tests: terminal detection table, OSC 9/777 payloads, tmux wrap,
  unknown-terminal fallthrough, terminal-preferred-over-OS ordering.
2026-06-18 14:06:08 -05:00
Brooklyn Nicholson
437105c717 fix(notify): prefer terminal-notifier on macOS for reliable banners
Plain `osascript display notification` attributes to the launching process;
for an unsigned CLI that frequently can't register an app entry, so macOS
delivers the notification silently to Notification Center with no banner and
no toggle the user can enable. Prefer `terminal-notifier` when on PATH (it
ships a real app bundle that shows banners and is grantable in System
Settings), falling back to osascript otherwise. `brew install
terminal-notifier` is the documented opt-in for reliable banners.
2026-06-18 14:01:16 -05:00
Brooklyn Nicholson
c4aeb8a931 fix(notify): scope sentinel per-session; dedupe consume; tidy
Builds on PCinkusz's /notify command (previous commit) to fix one design
flaw and tighten the implementation:

- Per-session sentinel. The pending-notify flag was a single global file
  (~/.hermes/.notify_pending). The TUI gateway and dashboard serve many
  sessions from one process sharing one HERMES_HOME, so a /notify set in
  session A would fire on session B's next turn completion. Key the sentinel
  by HERMES_SESSION_KEY (resolved from the per-turn contextvar in the gateway,
  the slash worker's env, or os.environ in the classic CLI). Classic
  single-session CLI keeps the unsuffixed default file — no behavior change.
- Single consume helper. The check->clear->fire block was copy-pasted at four
  sites (2 in cli.py, 2 in tui_gateway/server.py). Extract
  consume_pending_notification(session_key) and call it everywhere; the TUI
  sites pass session["session_key"] explicitly since that process has no
  per-session contextvar bound at the consume point.
- Drop the unused config= param from fire_notification; add the missing
  trailing newline; reuse approval._get_session_platform() in the
  approval-notify guard.
- tests/tools/test_notify_utils.py: per-session isolation, consume
  fire-once/scope, default-key, env-resolution.

Co-authored-by: PCinkusz <pcinkusz123321@gmail.com>
2026-06-18 13:52:15 -05:00
PCinkusz
eb20289f96 feat(cli): add local notify command 2026-06-18 13:48:31 -05:00
8 changed files with 936 additions and 1 deletions

76
cli.py
View File

@@ -7137,6 +7137,53 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
except Exception:
return False
def _should_handle_notify_command_inline(self, text: str, has_images: bool = False) -> bool:
"""Return True when /notify should be dispatched mid-task.
Same pattern as /steer: write the sentinel file without queuing
through _pending_input (which would miss the mid-run window).
"""
if not text or has_images or not _looks_like_slash_command(text):
return False
if not getattr(self, "_agent_running", False):
return False
try:
from hermes_cli.commands import resolve_command
base = text.split(None, 1)[0].lower().lstrip('/')
cmd = resolve_command(base)
return bool(cmd and cmd.name == "notify")
except Exception:
return False
def _handle_notify_command(self, cmd_original: str):
"""Handle /notify [prompt | cancel].
- /notify <prompt> — set flag + submit prompt
- /notify — set flag only (mid-task or pre-turn)
- /notify cancel — clear pending notification
"""
from tools.notify_utils import set_notify_flag, clear_notify_flag
parts = cmd_original.split(None, 1)
sub = parts[1].strip() if len(parts) > 1 else ""
if sub.lower() == "cancel":
if clear_notify_flag():
_cprint(" 🔕 Notification cancelled")
else:
_cprint(" No pending notification")
return
set_notify_flag()
if sub:
# Has a prompt — set flag AND submit
self._pending_input.put(sub)
_cprint(f" 🔔 Will notify when done: "
f"{sub[:80]}{'...' if len(sub) > 80 else ''}")
else:
_cprint(" 🔔 Will notify when this turn finishes")
def _output_console(self):
"""Use prompt_toolkit-safe Rich rendering once the TUI is live."""
if getattr(self, "_app", None):
@@ -7625,6 +7672,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
_cprint(f" Queued for the next turn: {payload[:80]}{'...' if len(payload) > 80 else ''}")
else:
_cprint(f" Queued: {payload[:80]}{'...' if len(payload) > 80 else ''}")
elif canonical == "notify":
self._handle_notify_command(cmd_original)
elif canonical == "steer":
# Inject a message after the next tool call without interrupting.
# If the agent is actively running, push the text into the agent's
@@ -10389,6 +10438,15 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
# Flush any remaining streamed text and close the box
self._flush_stream()
# Notify check — fire if /notify was set for this turn.
# Must run BEFORE goal continuation so the notification
# reflects actual turn completion (mirrors the TUI path).
try:
from tools.notify_utils import consume_pending_notification
consume_pending_notification()
except Exception as e:
logging.debug("notify idle-check failed: %s", e)
# Signal end-of-text to TTS consumer and wait for it to finish
if use_streaming_tts and text_queue is not None:
text_queue.put(None) # sentinel
@@ -11279,6 +11337,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
event.app.invalidate()
return
# Handle /notify while the agent is running — same pattern
# as /steer: write the sentinel file on the UI thread so it
# survives mid-run without queueing through _pending_input.
if self._should_handle_notify_command_inline(text, has_images=has_images):
self.process_command(text)
event.app.current_buffer.reset(append_to_history=True)
return
# Snapshot and clear attached images
images = list(self._attached_images)
self._attached_images.clear()
@@ -13054,6 +13120,16 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
self._pending_tool_info.clear()
self._last_scrollback_tool = ""
# Notify check — fire if /notify was set during this turn.
# Must run BEFORE goal continuation so the notification
# reflects the actual turn completion, not a
# potentially-auto-continued turn.
try:
from tools.notify_utils import consume_pending_notification
consume_pending_notification()
except Exception as e:
logging.debug("notify idle-check failed: %s", e)
app.invalidate() # Refresh status line
# Goal continuation: if a standing goal is active, ask

View File

@@ -103,6 +103,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
aliases=("tasks",)),
CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
aliases=("q",), args_hint="<prompt>"),
CommandDef("notify", "Set a desktop notification when Hermes finishes this turn", "Session",
args_hint="[prompt | cancel]", cli_only=True),
CommandDef("steer", "Inject a message after the next tool call without interrupting", "Session",
args_hint="<prompt>"),
CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session",

View File

@@ -0,0 +1,140 @@
"""Tests for /notify approval-request notifications."""
def test_fire_approval_request_notification_uses_input_needed_message(monkeypatch):
from tools import notify_utils
calls = []
monkeypatch.setattr(
notify_utils,
"fire_notification",
lambda *, title="Hermes Agent", message="Task complete", config=None: calls.append(
{"title": title, "message": message, "config": config}
),
)
notify_utils.fire_approval_request_notification()
assert calls == [
{"title": "Hermes Agent", "message": "Input needed: approval required", "config": None}
]
def test_fire_approval_request_notification_does_not_clear_pending_notify(monkeypatch, tmp_path):
from tools import notify_utils
monkeypatch.setattr(notify_utils, "_hermes_home", lambda: tmp_path)
monkeypatch.setattr(notify_utils, "fire_notification", lambda **kwargs: None)
notify_utils.set_notify_flag()
notify_utils.fire_approval_request_notification()
assert notify_utils.is_notify_pending() is True
def test_approval_request_notification_skips_messaging_gateway_platform(monkeypatch):
from gateway import session_context
from tools import approval
from tools import notify_utils
calls = []
monkeypatch.setattr(
session_context,
"get_session_env",
lambda name, default="": "telegram" if name == "HERMES_SESSION_PLATFORM" else default,
)
monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
approval._notify_approval_request_if_pending()
assert calls == []
def test_check_all_command_guards_notifies_when_cli_approval_requested(monkeypatch):
from tools import approval
from tools import notify_utils
calls = []
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
monkeypatch.setattr(
approval,
"detect_dangerous_command",
lambda command: (True, "dangerous:test", "test approval"),
)
monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *args, **kwargs: "deny")
monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
assert result["approved"] is False
assert calls == ["approval"]
def test_gateway_approval_prompt_is_emitted_before_desktop_notification(monkeypatch):
from tools import approval
from tools import notify_utils
calls = []
session_key = "notify-order-session"
def notify_cb(_approval_data):
calls.append("approval-prompt")
approval.resolve_gateway_approval(session_key, "deny")
monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1")
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
monkeypatch.setattr(approval, "get_current_session_key", lambda default="default": session_key)
monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval, "_get_approval_config", lambda: {"gateway_timeout": 1})
monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
monkeypatch.setattr(
approval,
"detect_dangerous_command",
lambda command: (True, "dangerous:test", "test approval"),
)
monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: True)
monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("desktop-notify"))
approval.register_gateway_notify(session_key, notify_cb)
result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
assert result["approved"] is False
assert calls == ["approval-prompt", "desktop-notify"]
def test_check_all_command_guards_skips_approval_notification_without_notify_pending(monkeypatch):
from tools import approval
from tools import notify_utils
calls = []
monkeypatch.setenv("HERMES_INTERACTIVE", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_EXEC_ASK", raising=False)
monkeypatch.delenv("HERMES_YOLO_MODE", raising=False)
monkeypatch.setattr(approval, "_get_approval_mode", lambda: "manual")
monkeypatch.setattr(approval, "detect_hardline_command", lambda command: (False, None))
monkeypatch.setattr(
approval,
"detect_dangerous_command",
lambda command: (True, "dangerous:test", "test approval"),
)
monkeypatch.setattr(approval, "is_approved", lambda session_key, pattern_key: False)
monkeypatch.setattr(approval, "prompt_dangerous_approval", lambda *args, **kwargs: "deny")
monkeypatch.setattr(notify_utils, "is_notify_pending", lambda: False)
monkeypatch.setattr(notify_utils, "fire_approval_request_notification", lambda: calls.append("approval"))
result = approval.check_all_command_guards("rm -rf /tmp/demo", "local")
assert result["approved"] is False
assert calls == []

View File

@@ -0,0 +1,201 @@
"""Tests for the per-session /notify sentinel + consume helper.
Covers the behavior the supersede added on top of the original PR:
the sentinel is scoped per session so a /notify in one TUI/dashboard
session never fires on another session's turn completion.
"""
import pytest
from tools import notify_utils
@pytest.fixture
def home(tmp_path, monkeypatch):
monkeypatch.setattr(notify_utils, "_hermes_home", lambda: tmp_path)
# Never hit the OS during tests.
monkeypatch.setattr(notify_utils, "_show_desktop_notification", lambda *a, **k: None)
return tmp_path
def test_set_is_pending_clear_roundtrip(home):
assert notify_utils.is_notify_pending("sess-a") is False
assert notify_utils.set_notify_flag("sess-a") is True
assert notify_utils.is_notify_pending("sess-a") is True
assert notify_utils.clear_notify_flag("sess-a") is True
assert notify_utils.is_notify_pending("sess-a") is False
# Clearing again is a no-op (nothing to remove).
assert notify_utils.clear_notify_flag("sess-a") is False
def test_sessions_are_independent(home):
"""A /notify set in session A must not register as pending for B."""
notify_utils.set_notify_flag("sess-a")
assert notify_utils.is_notify_pending("sess-a") is True
assert notify_utils.is_notify_pending("sess-b") is False
def test_distinct_keys_get_distinct_sentinels(home):
pa = notify_utils.get_notify_sentinel_path("sess-a")
pb = notify_utils.get_notify_sentinel_path("sess-b")
assert pa != pb
def test_empty_key_uses_default_sentinel(home):
# Classic CLI (no session key) gets the unsuffixed default file.
assert notify_utils.get_notify_sentinel_path("").name == ".notify_pending"
assert notify_utils.get_notify_sentinel_path(None).name == ".notify_pending"
def test_consume_fires_once_and_clears(home, monkeypatch):
fired = []
monkeypatch.setattr(
notify_utils, "fire_notification",
lambda **kw: fired.append(kw),
)
notify_utils.set_notify_flag("sess-a")
assert notify_utils.consume_pending_notification("sess-a") is True
assert len(fired) == 1
# Sentinel consumed — a second consume is a no-op.
assert notify_utils.consume_pending_notification("sess-a") is False
assert len(fired) == 1
def test_consume_is_scoped_to_its_session(home, monkeypatch):
fired = []
monkeypatch.setattr(
notify_utils, "fire_notification",
lambda **kw: fired.append(kw),
)
notify_utils.set_notify_flag("sess-a")
# Another session completing must NOT consume A's pending notify.
assert notify_utils.consume_pending_notification("sess-b") is False
assert fired == []
assert notify_utils.is_notify_pending("sess-a") is True
@pytest.mark.parametrize(
"env,expected",
[
({"TERM_PROGRAM": "iTerm.app"}, "osc9"),
({"TERM_PROGRAM": "WarpTerminal"}, "osc9"),
({"KITTY_WINDOW_ID": "1"}, "osc9"),
({"WEZTERM_PANE": "0"}, "osc777"),
# VS Code / Cursor and Apple Terminal parse but DON'T render OSC
# notifications — must fall through to the OS-level path.
({"TERM_PROGRAM": "vscode"}, None),
({"TERM_PROGRAM": "Apple_Terminal"}, None),
({}, None),
],
)
def test_detect_terminal_osc(monkeypatch, env, expected):
for k in ("TERM_PROGRAM", "KITTY_WINDOW_ID", "WEZTERM_PANE",
"GHOSTTY_RESOURCES_DIR"):
monkeypatch.delenv(k, raising=False)
for k, v in env.items():
monkeypatch.setenv(k, v)
assert notify_utils._detect_terminal_osc() == expected
def test_emit_terminal_notification_writes_osc9(monkeypatch):
written = []
monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc9")
monkeypatch.delenv("TMUX", raising=False)
monkeypatch.setattr(notify_utils, "_write_tty",
lambda payload: written.append(payload) or True)
assert notify_utils._emit_terminal_notification("Hermes", "done") is True
assert written == ["\033]9;Hermes: done\007"]
def test_emit_terminal_notification_writes_osc777(monkeypatch):
written = []
monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc777")
monkeypatch.delenv("TMUX", raising=False)
monkeypatch.setattr(notify_utils, "_write_tty",
lambda payload: written.append(payload) or True)
assert notify_utils._emit_terminal_notification("Hermes", "done") is True
assert written == ["\033]777;notify;Hermes;done\007"]
def test_emit_terminal_notification_unknown_terminal_returns_false(monkeypatch):
monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: None)
called = []
monkeypatch.setattr(notify_utils, "_write_tty",
lambda payload: called.append(payload) or True)
assert notify_utils._emit_terminal_notification("Hermes", "done") is False
assert called == [] # no tty write attempted
def test_emit_terminal_notification_wraps_for_tmux(monkeypatch):
written = []
monkeypatch.setattr(notify_utils, "_detect_terminal_osc", lambda: "osc9")
monkeypatch.setenv("TMUX", "/tmp/tmux-1000/default,123,0")
monkeypatch.setattr(notify_utils, "_write_tty",
lambda payload: written.append(payload) or True)
notify_utils._emit_terminal_notification("Hermes", "done")
assert written and written[0].startswith("\033Ptmux;")
assert written[0].endswith("\033\\")
def test_desktop_notification_prefers_terminal_over_os(monkeypatch):
"""Terminal OSC short-circuits the OS-level fallbacks."""
monkeypatch.setattr(notify_utils, "_emit_terminal_notification",
lambda t, m: True)
macos_called = []
monkeypatch.setattr(notify_utils, "_show_notification_macos",
lambda t, m: macos_called.append((t, m)))
notify_utils._show_desktop_notification("T", "M")
assert macos_called == []
def test_osascript_permission_hint_fires_once(monkeypatch, caplog):
import logging
monkeypatch.setattr(notify_utils, "_OSASCRIPT_HINT_SHOWN", False)
with caplog.at_level(logging.WARNING, logger=notify_utils.logger.name):
notify_utils._osascript_permission_hint_once()
notify_utils._osascript_permission_hint_once()
hits = [r for r in caplog.records if "Script Editor" in r.getMessage()]
assert len(hits) == 1
def test_macos_prefers_terminal_notifier_when_present(monkeypatch):
runs = []
monkeypatch.setattr(notify_utils.shutil, "which",
lambda name: "/opt/homebrew/bin/terminal-notifier")
monkeypatch.setattr(notify_utils.subprocess, "run",
lambda argv, **kw: runs.append(argv))
notify_utils._show_notification_macos("T", "M")
assert len(runs) == 1
assert runs[0][0].endswith("terminal-notifier")
assert "-message" in runs[0] and "M" in runs[0]
def test_macos_falls_back_to_osascript_without_terminal_notifier(monkeypatch):
runs = []
monkeypatch.setattr(notify_utils.shutil, "which", lambda name: None)
monkeypatch.setattr(notify_utils.subprocess, "run",
lambda argv, **kw: runs.append(argv))
notify_utils._show_notification_macos("T", "M")
assert len(runs) == 1
assert runs[0][0] == "osascript"
def test_key_resolves_from_session_env(home, monkeypatch):
"""When no explicit key is passed, the current session context is used."""
monkeypatch.setattr(
notify_utils, "_resolve_session_key",
lambda sk: "ctx-key" if sk is None else sk,
)
notify_utils.set_notify_flag() # resolves to "ctx-key"
assert notify_utils.is_notify_pending("ctx-key") is True
assert notify_utils.is_notify_pending() is True

View File

@@ -151,6 +151,33 @@ def _is_gateway_approval_context() -> bool:
return True
return bool(_get_session_platform())
def _notify_approval_request_if_pending() -> None:
"""Fire a /notify input-needed notification for approval prompts.
Best-effort only: approval safety flow must not depend on desktop
notification delivery. Do not clear the sentinel here; the final
turn-complete notification should still fire after the user responds.
"""
try:
from tools.notify_utils import (
fire_approval_request_notification,
is_notify_pending,
)
# /notify is local CLI/TUI-only. TUI gateway sessions set only a
# session key; messaging gateway sessions also set a platform. Do not
# let a local sentinel trigger desktop notifications from Telegram,
# Discord, Slack, etc. approval flows.
if _get_session_platform():
return
if is_notify_pending():
fire_approval_request_notification()
except Exception as exc:
logger.debug("Approval-request notification failed: %s", exc)
# Sensitive write targets that should trigger approval even when referenced
# via shell expansions like $HOME or $HERMES_HOME, or by the resolved absolute
# active profile home path such as /home/hermes/.hermes/config.yaml. The
@@ -1207,6 +1234,7 @@ def check_dangerous_command(command: str, env_type: str,
"pattern_key": pattern_key,
"description": description,
})
_notify_approval_request_if_pending()
return {
"approved": False,
"pattern_key": pattern_key,
@@ -1219,6 +1247,7 @@ def check_dangerous_command(command: str, env_type: str,
),
}
_notify_approval_request_if_pending()
choice = prompt_dangerous_approval(command, description,
approval_callback=approval_callback)
@@ -1547,6 +1576,10 @@ def check_all_command_guards(command: str, env_type: str,
"pattern_key": primary_key,
"description": combined_desc,
}
# The approval prompt reached the user — surface a local /notify
# input-needed desktop notification if one is pending (no-op on
# messaging-gateway sessions, which carry a platform).
_notify_approval_request_if_pending()
resolved = decision["resolved"]
choice = decision["choice"]
@@ -1596,7 +1629,9 @@ def check_all_command_guards(command: str, env_type: str,
"user_approved": True, "description": combined_desc}
# Fallback: no gateway callback registered (e.g. cron, batch).
# Return approval_required for backward compat.
# Return approval_required for backward compat. Do not fire the local
# desktop /notify hook here because there is no local UI prompt to pair
# it with.
submit_pending(session_key, {
"command": command,
"pattern_key": primary_key,
@@ -1626,6 +1661,7 @@ def check_all_command_guards(command: str, env_type: str,
session_key=session_key,
surface="cli",
)
_notify_approval_request_if_pending()
choice = prompt_dangerous_approval(command, combined_desc,
allow_permanent=not has_tirith,
approval_callback=approval_callback)

441
tools/notify_utils.py Normal file
View File

@@ -0,0 +1,441 @@
"""Desktop notification delivery for the /notify slash command.
All functions are fail-safe — notification errors are logged but never
propagate to the agent loop.
Cross-platform: Linux (notify-send), macOS (osascript),
Windows (PowerShell), and WSL (bridges to Windows via powershell.exe,
preferring notify-send via WSLg when available).
"""
import hashlib
import logging
import os
import platform
import shutil
import subprocess
from pathlib import Path
from typing import Optional
logger = logging.getLogger(__name__)
_SYSTEM = platform.system()
def _hermes_home() -> Path:
from hermes_constants import get_hermes_home
return get_hermes_home()
# ---------------------------------------------------------------------------
# WSL detection
# ---------------------------------------------------------------------------
_WSL_CACHE: Optional[bool] = None
def _is_wsl() -> bool:
"""Return True when running under Windows Subsystem for Linux."""
global _WSL_CACHE
if _WSL_CACHE is not None:
return _WSL_CACHE
try:
with open("/proc/version", "r") as f:
_WSL_CACHE = "microsoft" in f.read().lower()
except Exception:
_WSL_CACHE = False
return _WSL_CACHE
# ---------------------------------------------------------------------------
# Sentinel file (per-session)
# ---------------------------------------------------------------------------
#
# The pending-notify flag is scoped to a *session*, not the whole process.
# The TUI gateway and dashboard serve many concurrent sessions from one
# process sharing one HERMES_HOME; a single global sentinel would let a
# ``/notify`` set in session A fire on session B's turn completion. Keying
# the sentinel by HERMES_SESSION_KEY keeps each session's pending flag
# independent. Classic single-process CLI has no session key and falls back
# to the unsuffixed default file — same behavior as before.
def _resolve_session_key(session_key: Optional[str]) -> str:
"""Resolve the session key for the current context.
Explicit *session_key* wins (used by the TUI gateway, which serves many
sessions from one process and must name them explicitly). Otherwise read
``HERMES_SESSION_KEY`` from the session context — a contextvar bound
per-turn in the gateway, or ``os.environ`` in the classic CLI and the
slash worker. Falls back to ``""`` (the default sentinel).
"""
if session_key is not None:
return session_key
try:
from gateway.session_context import get_session_env
return get_session_env("HERMES_SESSION_KEY", "") or ""
except Exception:
return ""
def _sentinel_name(session_key: str) -> str:
key = (session_key or "").strip()
if not key:
return ".notify_pending"
digest = hashlib.sha1(key.encode("utf-8", "replace")).hexdigest()[:16]
return f".notify_pending-{digest}"
def get_notify_sentinel_path(session_key: Optional[str] = None) -> Path:
return _hermes_home() / _sentinel_name(_resolve_session_key(session_key))
def set_notify_flag(session_key: Optional[str] = None) -> bool:
"""Write the sentinel file to signal a pending notification."""
try:
p = get_notify_sentinel_path(session_key)
p.parent.mkdir(parents=True, exist_ok=True)
p.touch()
return True
except Exception as e:
logger.warning("Failed to write notify sentinel: %s", e)
return False
def clear_notify_flag(session_key: Optional[str] = None) -> bool:
"""Remove the sentinel file (cancel or consume notification)."""
try:
p = get_notify_sentinel_path(session_key)
if not p.exists():
return False
p.unlink()
return True
except Exception as e:
logger.warning("Failed to clear notify sentinel: %s", e)
return False
def is_notify_pending(session_key: Optional[str] = None) -> bool:
"""Check if a notification is pending for this session."""
return get_notify_sentinel_path(session_key).exists()
# ---------------------------------------------------------------------------
# Desktop notification
# ---------------------------------------------------------------------------
def _notify_send_available() -> bool:
"""Return True if notify-send is available and D-Bus is reachable."""
if not shutil.which("notify-send"):
return False
# Quick smoke-test: verify D-Bus notification service exists
try:
result = subprocess.run(
["notify-send", "--version"],
timeout=3, capture_output=True,
)
return result.returncode == 0
except Exception:
return False
def _show_notification_linux(title: str, message: str) -> None:
"""Desktop notification on native Linux via notify-send."""
try:
subprocess.run(
["notify-send", title, message],
timeout=5, capture_output=True,
)
logger.debug("notify: Linux notification sent via notify-send")
except FileNotFoundError:
logger.debug("notify: notify-send not found on Linux")
except subprocess.TimeoutExpired:
logger.debug("notify: notify-send timed out on Linux")
def _ps_single_quote(value: str) -> str:
"""Quote a string for a single-quoted PowerShell literal."""
return "'" + value.replace("'", "''") + "'"
def _show_notification_wsl(title: str, message: str) -> None:
"""Desktop notification in WSL via Windows balloon tip (PowerShell)."""
logger.debug("notify: attempting WSL notification via PowerShell")
try:
ps_code = (
"Add-Type -AssemblyName System.Windows.Forms; "
"$n = New-Object System.Windows.Forms.NotifyIcon; "
"$n.Icon = [System.Drawing.SystemIcons]::Information; "
f"$n.BalloonTipTitle = {_ps_single_quote(title)}; "
f"$n.BalloonTipText = {_ps_single_quote(message)}; "
"$n.Visible = $true; "
"$n.ShowBalloonTip(3000); "
"[System.Windows.Forms.Application]::DoEvents(); "
"Start-Sleep -Seconds 4; "
"$n.Dispose()"
)
result = subprocess.run(
["powershell.exe", "-c", ps_code],
timeout=8, capture_output=True,
)
if result.returncode != 0:
logger.debug("notify: PowerShell balloon failed (rc=%d, stderr=%s)",
result.returncode, result.stderr.decode(errors="replace")[:200])
else:
logger.debug("notify: WSL notification sent via PowerShell")
except subprocess.TimeoutExpired:
logger.debug("notify: PowerShell balloon timed out")
except FileNotFoundError:
logger.debug("notify: powershell.exe not found — is WSL properly configured?")
except Exception as e:
logger.warning("WSL notification failed: %s", e)
def _show_notification_macos(title: str, message: str) -> None:
"""Desktop notification on macOS.
Prefer ``terminal-notifier`` when it's on PATH: it ships a real app
bundle, so notifications attribute to it, show as banners, and are
grantable in System Settings → Notifications. Plain ``osascript display
notification`` attributes to the *launching* process — for an unsigned
CLI that often can't register an app entry, so macOS delivers it silently
to Notification Center with no banner (and no toggle the user can flip).
Install with ``brew install terminal-notifier`` for reliable banners;
otherwise fall back to osascript.
"""
tn = shutil.which("terminal-notifier")
if tn:
try:
subprocess.run(
[tn, "-title", title, "-message", message],
timeout=5, capture_output=True,
)
logger.debug("notify: macOS notification sent via terminal-notifier")
return
except Exception as e:
logger.debug(
"notify: terminal-notifier failed (%s), falling back to osascript", e
)
try:
escaped_title = title.replace('\\', '\\\\').replace('"', '\\"')
escaped_message = message.replace('\\', '\\\\').replace('"', '\\"')
subprocess.run(
["osascript", "-e",
f"display notification \"{escaped_message}\" with title \"{escaped_title}\""],
timeout=5, capture_output=True,
)
logger.debug("notify: macOS notification sent via osascript")
_osascript_permission_hint_once()
except Exception as e:
logger.debug("notify: osascript notification failed: %s", e)
_OSASCRIPT_HINT_SHOWN = False
def _osascript_permission_hint_once() -> None:
"""Warn once that osascript notifications need Script Editor permission.
On macOS Sequoia+, ``osascript display notification`` is attributed to
``com.apple.ScriptEditor2`` and is silently dropped until the user grants
Script Editor notification permission — the command still exits 0, so a
user sees "nothing happened" with no error. Surface the one-time fix so
they aren't stuck. (Native terminals using the OSC path never reach here.)
"""
global _OSASCRIPT_HINT_SHOWN
if _OSASCRIPT_HINT_SHOWN:
return
_OSASCRIPT_HINT_SHOWN = True
logger.warning(
"Desktop notifications use osascript on this terminal, which macOS "
"delivers as 'Script Editor' — banners are suppressed until you grant "
"permission once: run `open -a 'Script Editor'`, execute "
"`display notification \"test\" with title \"test\"` inside it, click "
"Allow, then check System Settings > Notifications > Script Editor. "
"For native banners, run Hermes in iTerm2/Ghostty/kitty/WezTerm, or "
"install the VS Code 'Terminal Notification' extension for Cursor."
)
# ---------------------------------------------------------------------------
# Terminal-native notifications (OSC escape sequences)
# ---------------------------------------------------------------------------
#
# The most reliable way to notify from an unsigned CLI — especially on modern
# macOS, where ``osascript`` notifications are permanently stuck under the
# "Script Editor" identity (Apple removed sender override in Monterey) and
# ``terminal-notifier`` is broken on recent releases. Here the *terminal
# emulator itself* raises the banner: it's attributed to the terminal (which
# the user already trusts), clicking it focuses the terminal, and there's no
# extra dependency.
#
# We emit the escape sequence directly to the controlling terminal
# (``/dev/tty``), NOT stdout — the TUI/slash-worker capture stdout, and the
# sequences are non-rendering so they don't disturb a live TUI screen.
#
# Two flavors cover the field (see terminfo.dev): OSC 9 (iTerm2-style, single
# string) and OSC 777 (urxvt-style, title+body). We pick ONE per terminal via
# TERM_PROGRAM/env so a terminal that supports both doesn't double-fire.
# Apple Terminal ignores both → returns False so the caller falls back.
def _detect_terminal_osc() -> Optional[str]:
"""Return the OSC notification flavor for the current terminal, or None.
Only terminals that actually *render* an OS notification from the escape
sequence are listed. terminfo.dev marks many terminals as "supporting"
OSC 9/777, but that only means their parser consumes the sequence — VS
Code/Cursor and Apple Terminal silently drop it without showing anything
(confirmed: microsoft/vscode#294247, anthropics/claude-code#28338). Those
return None so the caller falls back to an OS-level notifier (osascript).
VS Code / Cursor users who want click-to-focus can install the "Terminal
Notification" extension (it parses OSC 9/777 from the terminal stream);
that's a user-side opt-in, not something we can assume here.
"""
if os.environ.get("KITTY_WINDOW_ID"):
return "osc9" # kitty also speaks the legacy OSC 9
if os.environ.get("WEZTERM_PANE") or os.environ.get("GHOSTTY_RESOURCES_DIR"):
return "osc777"
tp = os.environ.get("TERM_PROGRAM", "")
return {
"iTerm.app": "osc9",
"WarpTerminal": "osc9",
"ghostty": "osc777",
"WezTerm": "osc777",
}.get(tp)
def _tmux_wrap(seq: str) -> str:
"""Wrap an escape sequence for tmux passthrough so it reaches the outer
terminal. Requires ``set -g allow-passthrough on`` in tmux >= 3.3."""
return "\033Ptmux;" + seq.replace("\033", "\033\033") + "\033\\"
def _emit_terminal_notification(title: str, message: str) -> bool:
"""Emit an OSC desktop-notification sequence to the controlling terminal.
Returns True when written to a terminal known to support it. Fully
fail-safe.
"""
kind = _detect_terminal_osc()
if not kind:
return False
if kind == "osc777":
seq = f"\033]777;notify;{title};{message}\007"
else: # osc9 — single string
seq = f"\033]9;{title}: {message}\007" if title else f"\033]9;{message}\007"
if os.environ.get("TMUX"):
seq = _tmux_wrap(seq)
if _write_tty(seq):
logger.debug("notify: terminal notification sent via %s", kind)
return True
return False
def _write_tty(payload: str) -> bool:
"""Write *payload* to the controlling terminal (/dev/tty). Fail-safe."""
try:
with open("/dev/tty", "w") as tty:
tty.write(payload)
tty.flush()
return True
except Exception as e:
logger.debug("notify: /dev/tty write failed: %s", e)
return False
def _show_desktop_notification(title: str, message: str) -> None:
"""Show a desktop notification bubble.
Order of preference:
1. Terminal-native OSC sequence (works in iTerm2/Ghostty/kitty/WezTerm/
Warp/VS Code/Cursor; reliable for an unsigned CLI on modern macOS).
2. OS-level fallback per platform (notify-send / terminal-notifier /
osascript / PowerShell). Used for Apple Terminal and unknown terminals.
"""
try:
if _emit_terminal_notification(title, message):
return
if _is_wsl():
# WSLg path: notify-send bridges to native Windows notifications
if _notify_send_available():
logger.debug("notify: WSLg notify-send available, using D-Bus path")
_show_notification_linux(title, message)
return
logger.debug("notify: notify-send not available in WSL, falling back to PowerShell")
_show_notification_wsl(title, message)
elif _SYSTEM == "Linux":
_show_notification_linux(title, message)
elif _SYSTEM == "Darwin":
_show_notification_macos(title, message)
elif _SYSTEM == "Windows":
ps_code = (
"Add-Type -AssemblyName System.Windows.Forms; "
"$n = New-Object System.Windows.Forms.NotifyIcon; "
"$n.Icon = [System.Drawing.SystemIcons]::Information; "
f"$n.BalloonTipTitle = {_ps_single_quote(title)}; "
f"$n.BalloonTipText = {_ps_single_quote(message)}; "
"$n.Visible = $true; "
"$n.ShowBalloonTip(3000); "
"Start-Sleep -Seconds 4"
)
subprocess.run(
["powershell", "-c", ps_code],
timeout=8, capture_output=True,
)
logger.debug("notify: Windows notification sent via PowerShell")
except Exception as e:
logger.debug("Desktop notification failed: %s", e)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def fire_notification(
*,
title: str = "Hermes Agent",
message: str = "Task complete",
) -> None:
"""Fire a desktop notification.
All errors are caught silently — notification failure must never
crash the idle loop.
Args:
title: Desktop notification title.
message: Desktop notification body.
"""
_show_desktop_notification(title, message)
def fire_approval_request_notification() -> None:
"""Notify that Hermes is blocked waiting for command approval.
This intentionally does not clear the /notify sentinel; the final
turn-complete notification should still fire after the user responds.
"""
fire_notification(message="Input needed: approval required")
def consume_pending_notification(
session_key: Optional[str] = None,
*,
title: str = "Hermes Agent",
message: str = "Task complete",
) -> bool:
"""Fire-and-clear the pending notification for *session_key*, if any.
Single entry point for the turn-complete sites (CLI idle loop, CLI
process loop, TUI gateway success/error paths) so the
check→clear→fire sequence lives in one place. Returns True when a
notification was fired. Fully fail-safe.
"""
try:
if is_notify_pending(session_key):
clear_notify_flag(session_key)
fire_notification(title=title, message=message)
return True
except Exception as e:
logger.debug("notify consume failed: %s", e)
return False

View File

@@ -6330,6 +6330,16 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
file=sys.stderr,
)
# Notify check — fire if /notify was set for THIS session's turn.
# The sentinel is per-session (keyed by session_key), so a
# /notify in one TUI session never fires on another session's
# completion. Set by the SlashWorker that handled /notify.
try:
from tools.notify_utils import consume_pending_notification
consume_pending_notification(session.get("session_key"))
except Exception as e:
logging.debug("tui notify idle-check failed: %s", e)
# Apply pending_title now that the DB row exists.
_pending = session.get("pending_title")
if _pending and status == "complete":
@@ -6410,6 +6420,13 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True
)
_emit("error", sid, {"message": str(e)})
# If /notify was set for this failed turn, consume it here too so
# the (per-session) sentinel doesn't survive into a later turn.
try:
from tools.notify_utils import consume_pending_notification
consume_pending_notification(session.get("session_key"))
except Exception as notify_exc:
logging.debug("tui notify error-path check failed: %s", notify_exc)
finally:
try:
if approval_token is not None:

View File

@@ -62,6 +62,28 @@ interface SkillsReloadResponse {
}
export const opsCommands: SlashCommand[] = [
{
help: 'notify when this turn finishes; optionally submit a prompt',
name: 'notify',
run: (arg, ctx, cmd) => {
const trimmed = arg.trim()
ctx.gateway
.rpc<SlashExecResponse>('slash.exec', { command: cmd.slice(1), session_id: ctx.sid })
.then(
ctx.guarded<SlashExecResponse>(r => {
const body = r?.output || '/notify: no output'
ctx.transcript.sys(body)
if (trimmed && trimmed.toLowerCase() !== 'cancel') {
ctx.transcript.send(trimmed)
}
})
)
.catch(ctx.guardedErr)
}
},
{
help: 'stop background processes',
name: 'stop',