Compare commits
5 Commits
main
...
hermes/not
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
82b9c44cbd | ||
|
|
b59b1cfb12 | ||
|
|
437105c717 | ||
|
|
c4aeb8a931 | ||
|
|
eb20289f96 |
76
cli.py
76
cli.py
@ -7137,6 +7137,53 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
|||||||
except Exception:
|
except Exception:
|
||||||
return False
|
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):
|
def _output_console(self):
|
||||||
"""Use prompt_toolkit-safe Rich rendering once the TUI is live."""
|
"""Use prompt_toolkit-safe Rich rendering once the TUI is live."""
|
||||||
if getattr(self, "_app", None):
|
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 ''}")
|
_cprint(f" Queued for the next turn: {payload[:80]}{'...' if len(payload) > 80 else ''}")
|
||||||
else:
|
else:
|
||||||
_cprint(f" Queued: {payload[:80]}{'...' if len(payload) > 80 else ''}")
|
_cprint(f" Queued: {payload[:80]}{'...' if len(payload) > 80 else ''}")
|
||||||
|
elif canonical == "notify":
|
||||||
|
self._handle_notify_command(cmd_original)
|
||||||
elif canonical == "steer":
|
elif canonical == "steer":
|
||||||
# Inject a message after the next tool call without interrupting.
|
# Inject a message after the next tool call without interrupting.
|
||||||
# If the agent is actively running, push the text into the agent's
|
# 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
|
# Flush any remaining streamed text and close the box
|
||||||
self._flush_stream()
|
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
|
# Signal end-of-text to TTS consumer and wait for it to finish
|
||||||
if use_streaming_tts and text_queue is not None:
|
if use_streaming_tts and text_queue is not None:
|
||||||
text_queue.put(None) # sentinel
|
text_queue.put(None) # sentinel
|
||||||
@ -11279,6 +11337,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
|||||||
event.app.invalidate()
|
event.app.invalidate()
|
||||||
return
|
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
|
# Snapshot and clear attached images
|
||||||
images = list(self._attached_images)
|
images = list(self._attached_images)
|
||||||
self._attached_images.clear()
|
self._attached_images.clear()
|
||||||
@ -13054,6 +13120,16 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
|
|||||||
self._pending_tool_info.clear()
|
self._pending_tool_info.clear()
|
||||||
self._last_scrollback_tool = ""
|
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
|
app.invalidate() # Refresh status line
|
||||||
|
|
||||||
# Goal continuation: if a standing goal is active, ask
|
# Goal continuation: if a standing goal is active, ask
|
||||||
|
|||||||
@ -103,6 +103,8 @@ COMMAND_REGISTRY: list[CommandDef] = [
|
|||||||
aliases=("tasks",)),
|
aliases=("tasks",)),
|
||||||
CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
|
CommandDef("queue", "Queue a prompt for the next turn (doesn't interrupt)", "Session",
|
||||||
aliases=("q",), args_hint="<prompt>"),
|
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",
|
CommandDef("steer", "Inject a message after the next tool call without interrupting", "Session",
|
||||||
args_hint="<prompt>"),
|
args_hint="<prompt>"),
|
||||||
CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session",
|
CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session",
|
||||||
|
|||||||
140
tests/test_notify_approval.py
Normal file
140
tests/test_notify_approval.py
Normal 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 == []
|
||||||
201
tests/tools/test_notify_utils.py
Normal file
201
tests/tools/test_notify_utils.py
Normal 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
|
||||||
@ -151,6 +151,33 @@ def _is_gateway_approval_context() -> bool:
|
|||||||
return True
|
return True
|
||||||
return bool(_get_session_platform())
|
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
|
# Sensitive write targets that should trigger approval even when referenced
|
||||||
# via shell expansions like $HOME or $HERMES_HOME, or by the resolved absolute
|
# 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
|
# 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,
|
"pattern_key": pattern_key,
|
||||||
"description": description,
|
"description": description,
|
||||||
})
|
})
|
||||||
|
_notify_approval_request_if_pending()
|
||||||
return {
|
return {
|
||||||
"approved": False,
|
"approved": False,
|
||||||
"pattern_key": pattern_key,
|
"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,
|
choice = prompt_dangerous_approval(command, description,
|
||||||
approval_callback=approval_callback)
|
approval_callback=approval_callback)
|
||||||
|
|
||||||
@ -1547,6 +1576,10 @@ def check_all_command_guards(command: str, env_type: str,
|
|||||||
"pattern_key": primary_key,
|
"pattern_key": primary_key,
|
||||||
"description": combined_desc,
|
"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"]
|
resolved = decision["resolved"]
|
||||||
choice = decision["choice"]
|
choice = decision["choice"]
|
||||||
|
|
||||||
@ -1596,7 +1629,9 @@ def check_all_command_guards(command: str, env_type: str,
|
|||||||
"user_approved": True, "description": combined_desc}
|
"user_approved": True, "description": combined_desc}
|
||||||
|
|
||||||
# Fallback: no gateway callback registered (e.g. cron, batch).
|
# 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, {
|
submit_pending(session_key, {
|
||||||
"command": command,
|
"command": command,
|
||||||
"pattern_key": primary_key,
|
"pattern_key": primary_key,
|
||||||
@ -1626,6 +1661,7 @@ def check_all_command_guards(command: str, env_type: str,
|
|||||||
session_key=session_key,
|
session_key=session_key,
|
||||||
surface="cli",
|
surface="cli",
|
||||||
)
|
)
|
||||||
|
_notify_approval_request_if_pending()
|
||||||
choice = prompt_dangerous_approval(command, combined_desc,
|
choice = prompt_dangerous_approval(command, combined_desc,
|
||||||
allow_permanent=not has_tirith,
|
allow_permanent=not has_tirith,
|
||||||
approval_callback=approval_callback)
|
approval_callback=approval_callback)
|
||||||
|
|||||||
441
tools/notify_utils.py
Normal file
441
tools/notify_utils.py
Normal 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
|
||||||
@ -6330,6 +6330,16 @@ def _run_prompt_submit(rid, sid: str, session: dict, text: Any) -> None:
|
|||||||
file=sys.stderr,
|
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.
|
# Apply pending_title now that the DB row exists.
|
||||||
_pending = session.get("pending_title")
|
_pending = session.get("pending_title")
|
||||||
if _pending and status == "complete":
|
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
|
f"[gateway-turn] {type(e).__name__}: {e}", file=sys.stderr, flush=True
|
||||||
)
|
)
|
||||||
_emit("error", sid, {"message": str(e)})
|
_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:
|
finally:
|
||||||
try:
|
try:
|
||||||
if approval_token is not None:
|
if approval_token is not None:
|
||||||
|
|||||||
@ -62,6 +62,28 @@ interface SkillsReloadResponse {
|
|||||||
}
|
}
|
||||||
|
|
||||||
export const opsCommands: SlashCommand[] = [
|
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',
|
help: 'stop background processes',
|
||||||
name: 'stop',
|
name: 'stop',
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user