From b9c001116e2bc6e2b112d9338ab6ce10040896a0 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Sat, 9 May 2026 11:04:46 -0700 Subject: [PATCH] feat: confirm prompt for destructive slash commands (#4069) (#22687) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit /clear, /new, /reset, and /undo now ask the user to confirm before discarding conversation state — three-option prompt routed through the existing tools.slash_confirm primitive. Native yes/no buttons render on Telegram, Discord, and Slack (their adapters already implement send_slash_confirm); other platforms get a text-fallback prompt and reply with /approve, /always, or /cancel. The classic prompt_toolkit CLI uses the same three-option flow via the established _prompt_text_input pattern (see _confirm_and_reload_mcp). TUI keeps its existing modal overlay (#12312). Gated by new config key approvals.destructive_slash_confirm (default true). Picking 'Always Approve' flips the gate to false so subsequent destructive commands run silently — matches the established mcp_reload_confirm UX. Out of scope: /cron remove (separate domain — scheduled jobs, not session history). Existing TUI overlay env-var (HERMES_TUI_NO_CONFIRM) left unchanged; cosmetic unification can come later. Closes #4069. --- cli.py | 89 ++++++ gateway/run.py | 121 +++++++- hermes_cli/config.py | 9 + tests/cli/test_cli_new_session.py | 5 + tests/cli/test_destructive_slash_confirm.py | 152 ++++++++++ .../gateway/test_destructive_slash_confirm.py | 261 ++++++++++++++++++ tests/gateway/test_telegram_topic_mode.py | 5 + tests/gateway/test_update_streaming.py | 5 + .../test_destructive_slash_confirm_gate.py | 86 ++++++ 9 files changed, 730 insertions(+), 3 deletions(-) create mode 100644 tests/cli/test_destructive_slash_confirm.py create mode 100644 tests/gateway/test_destructive_slash_confirm.py create mode 100644 tests/hermes_cli/test_destructive_slash_confirm_gate.py diff --git a/cli.py b/cli.py index fed96a157bd..585b664f2b2 100644 --- a/cli.py +++ b/cli.py @@ -6751,6 +6751,12 @@ class HermesCLI: self._force_full_redraw() _cprint(f" {_DIM}✓ UI redrawn{_RST}") elif canonical == "clear": + if self._confirm_destructive_slash( + "clear", + "This clears the screen and starts a new session.\n" + "The current conversation history will be discarded.", + ) is None: + return self.new_session(silent=True) _clear_output_history() # Clear terminal screen. Inside the TUI, Rich's console.clear() @@ -6873,6 +6879,12 @@ class HermesCLI: elif canonical == "new": parts = cmd_original.split(maxsplit=1) title = parts[1].strip() if len(parts) > 1 else None + if self._confirm_destructive_slash( + "new", + "This starts a fresh session.\n" + "The current conversation history will be discarded.", + ) is None: + return self.new_session(title=title) elif canonical == "resume": self._handle_resume_command(cmd_original) @@ -6890,6 +6902,11 @@ class HermesCLI: # Re-queue the message so process_loop sends it to the agent self._pending_input.put(retry_msg) elif canonical == "undo": + if self._confirm_destructive_slash( + "undo", + "This removes the last user/assistant exchange from history.", + ) is None: + return self.undo_last() elif canonical == "branch": self._handle_branch_command(cmd_original) @@ -8307,6 +8324,78 @@ class HermesCLI: if _reload_thread.is_alive(): print(" ⚠️ MCP reload timed out (30s). Some servers may not have reconnected.") + def _confirm_destructive_slash(self, command: str, detail: str) -> Optional[str]: + """Prompt the user to confirm a destructive session slash command. + + Used by ``/clear``, ``/new``/``/reset``, and ``/undo`` before they + discard conversation state. Three-option prompt: + + 1. Approve Once — proceed this time only + 2. Always Approve — proceed and persist + ``approvals.destructive_slash_confirm: false`` so future + destructive commands run without confirmation + 3. Cancel — abort + + Gated by ``approvals.destructive_slash_confirm`` (default on). If the + gate is off the function returns ``"once"`` immediately without + prompting. + + Returns ``"once"``, ``"always"``, or ``None`` (cancelled). Callers + proceed with the destructive action when the result is non-None. + """ + # Gate check — respects prior "Always Approve" clicks. + try: + cfg = load_cli_config() + approvals = cfg.get("approvals") if isinstance(cfg, dict) else None + confirm_required = True + if isinstance(approvals, dict): + confirm_required = bool(approvals.get("destructive_slash_confirm", True)) + except Exception: + confirm_required = True + + if not confirm_required: + return "once" + + # Render warning + prompt — single-line composer prompt, mirrors + # ``_confirm_and_reload_mcp``. + print() + print(f"⚠️ /{command} — destroys conversation state") + print() + for line in detail.splitlines(): + print(f" {line}") + print() + print(" [1] Approve Once — proceed this time only") + print(" [2] Always Approve — proceed and silence this prompt permanently") + print(" [3] Cancel — keep current conversation") + print() + raw = self._prompt_text_input("Choice [1/2/3]: ") + if raw is None: + print(f"🟡 /{command} cancelled (no input).") + return None + choice_raw = raw.strip().lower() + if choice_raw in ("1", "once", "approve", "yes", "y", "ok"): + choice = "once" + elif choice_raw in ("2", "always", "remember"): + choice = "always" + elif choice_raw in ("3", "cancel", "nevermind", "no", "n", ""): + choice = "cancel" + else: + print(f"🟡 Unrecognized choice '{raw}'. /{command} cancelled.") + return None + + if choice == "cancel": + print(f"🟡 /{command} cancelled. Conversation unchanged.") + return None + + if choice == "always": + if save_config_value("approvals.destructive_slash_confirm", False): + print("🔒 Future /clear, /new, /reset, and /undo will run without confirmation.") + print(" Re-enable via `approvals.destructive_slash_confirm: true` in config.yaml.") + else: + print("⚠️ Couldn't persist opt-out — proceeding once.") + + return choice + def _confirm_and_reload_mcp(self, cmd_original: str = "") -> None: """Interactive /reload-mcp — confirm with the user, then reload. diff --git a/gateway/run.py b/gateway/run.py index a72a7d411b9..15bfe53c66d 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -5776,7 +5776,18 @@ class GatewayRunner: if canonical == "new": if self._is_telegram_topic_root_lobby(source): return self._telegram_topic_root_new_message() - return await self._handle_reset_command(event) + async def _do_reset(): + return await self._handle_reset_command(event) + return await self._maybe_confirm_destructive_slash( + event=event, + command="new", + title="/new", + detail=( + "This starts a fresh session and discards the current " + "conversation history." + ), + execute=_do_reset, + ) if canonical == "topic": return await self._handle_topic_command(event) @@ -5830,7 +5841,15 @@ class GatewayRunner: return await self._handle_retry_command(event) if canonical == "undo": - return await self._handle_undo_command(event) + async def _do_undo(): + return await self._handle_undo_command(event) + return await self._maybe_confirm_destructive_slash( + event=event, + command="undo", + title="/undo", + detail="This removes the last user/assistant exchange from history.", + execute=_do_undo, + ) if canonical == "sethome": return await self._handle_set_home_command(event) @@ -11304,6 +11323,93 @@ class GatewayRunner: # /cancel; the early intercept in ``_handle_message`` matches # those replies against ``tools.slash_confirm.get_pending()``. + async def _maybe_confirm_destructive_slash( + self, + *, + event: MessageEvent, + command: str, + title: str, + detail: str, + execute, + ) -> Union[str, "EphemeralReply", None]: + """Gate a destructive session slash command (/new, /reset, /undo). + + ``execute`` is an async callable ``execute() -> str | EphemeralReply`` + that performs the destructive action. If the + ``approvals.destructive_slash_confirm`` config gate is off, ``execute`` + runs immediately (returning its result). Otherwise this routes + through ``_request_slash_confirm`` — native yes/no buttons on + Telegram/Discord/Slack, text fallback elsewhere. + + Three-option resolution: + + - ``once`` — run ``execute`` and return its result + - ``always`` — persist ``approvals.destructive_slash_confirm: false``, + then run ``execute`` + - ``cancel`` — return a "cancelled" message; do not run ``execute`` + """ + # Gate check. + confirm_required = True + try: + cfg = self._read_user_config() + approvals = cfg.get("approvals") if isinstance(cfg, dict) else None + if isinstance(approvals, dict): + confirm_required = bool(approvals.get("destructive_slash_confirm", True)) + except Exception: + pass + + if not confirm_required: + return await execute() + + session_key = self._session_key_for_source(event.source) + + async def _on_confirm(choice: str): + if choice == "cancel": + return f"🟡 /{command} cancelled. Conversation unchanged." + if choice == "always": + try: + from cli import save_config_value + save_config_value("approvals.destructive_slash_confirm", False) + logger.info( + "User opted out of destructive slash confirm (session=%s)", + session_key, + ) + except Exception as exc: + logger.warning( + "Failed to persist destructive_slash_confirm=false: %s", exc, + ) + result = await execute() + if choice == "always": + note = ( + "\n\nℹ️ Future /clear, /new, /reset, and /undo will run " + "without confirmation. Re-enable via " + "`approvals.destructive_slash_confirm: true` in config.yaml." + ) + if isinstance(result, str): + return result + note + # EphemeralReply or other — leave untouched; the opt-out note + # would otherwise mangle structured replies. The persist itself + # already happened above; user gets the same UX next time. + return result + return result + + prompt_message = ( + f"⚠️ **Confirm /{command}**\n\n" + f"{detail}\n\n" + "Choose:\n" + "• **Approve Once** — proceed this time only\n" + "• **Always Approve** — proceed and silence this prompt permanently\n" + "• **Cancel** — keep current conversation\n\n" + "_Text fallback: reply `/approve`, `/always`, or `/cancel`._" + ) + return await self._request_slash_confirm( + event=event, + command=command, + title=title, + message=prompt_message, + handler=_on_confirm, + ) + async def _request_slash_confirm( self, *, @@ -11329,7 +11435,16 @@ class GatewayRunner: source = event.source session_key = self._session_key_for_source(source) - confirm_id = f"{next(self._slash_confirm_counter)}" + # Bare-runner test harnesses (object.__new__(GatewayRunner)) skip + # __init__ and don't have the counter attribute — fall back to a + # local counter so tests don't AttributeError. Real runs always + # have the instance attribute. + counter = getattr(self, "_slash_confirm_counter", None) + if counter is None: + import itertools as _itertools + counter = _itertools.count(1) + self._slash_confirm_counter = counter + confirm_id = f"{next(counter)}" # Register the pending confirm FIRST so a super-fast button click # cannot race the send_slash_confirm return. diff --git a/hermes_cli/config.py b/hermes_cli/config.py index 262b8f2285d..117d3e25d04 100644 --- a/hermes_cli/config.py +++ b/hermes_cli/config.py @@ -1204,6 +1204,15 @@ DEFAULT_CONFIG = { # "Always Approve" to silence the prompt permanently; that flips # this key to false. "mcp_reload_confirm": True, + # When true, destructive session slash commands (/clear, /new, /reset, + # /undo) ask the user to confirm before discarding conversation state. + # Three-option prompt (Approve Once / Always Approve / Cancel) routed + # through tools.slash_confirm — native yes/no buttons on Telegram, + # Discord, and Slack; text fallback elsewhere. Users click "Always + # Approve" to silence the prompt permanently; that flips this key to + # false. TUI has its own modal overlay (HERMES_TUI_NO_CONFIRM=1 to + # opt out there). + "destructive_slash_confirm": True, }, # Permanently allowed dangerous command patterns (added via "always" approval) diff --git a/tests/cli/test_cli_new_session.py b/tests/cli/test_cli_new_session.py index 4f453fea32a..05503552cec 100644 --- a/tests/cli/test_cli_new_session.py +++ b/tests/cli/test_cli_new_session.py @@ -130,6 +130,11 @@ def _prepare_cli_with_active_session(tmp_path): old_session_start = cli.session_start - timedelta(seconds=1) cli.session_start = old_session_start cli.agent.session_start = old_session_start + + # Bypass the destructive-slash confirmation gate — these tests focus on + # the new-session mechanics, not the confirm prompt itself (covered in + # tests/cli/test_destructive_slash_confirm.py). + cli._confirm_destructive_slash = lambda *_a, **_kw: "once" return cli diff --git a/tests/cli/test_destructive_slash_confirm.py b/tests/cli/test_destructive_slash_confirm.py new file mode 100644 index 00000000000..290314dc371 --- /dev/null +++ b/tests/cli/test_destructive_slash_confirm.py @@ -0,0 +1,152 @@ +"""Tests for cli.HermesCLI._confirm_destructive_slash. + +Drives the helper directly via __get__ on a SimpleNamespace stand-in so we +don't have to construct a full HermesCLI (which requires extensive setup). +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import patch + + +def _bound(fn, instance): + """Bind an unbound method to a stand-in instance.""" + return fn.__get__(instance, type(instance)) + + +def _make_self(prompt_response): + """Build a minimal stand-in 'self' for _confirm_destructive_slash.""" + return SimpleNamespace( + _app=None, + _prompt_text_input=lambda _prompt: prompt_response, + ) + + +def test_gate_off_returns_once_without_prompting(): + """When approvals.destructive_slash_confirm is False, return 'once' + immediately (caller proceeds without showing a prompt).""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="should not be called") + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": False}}, + ): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result == "once" + + +def test_gate_on_choice_once_returns_once(): + """When the gate is on and the user picks '1', return 'once'.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="1") + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": True}}, + ): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result == "once" + + +def test_gate_on_choice_cancel_returns_none(): + """When the user picks '3' (cancel), return None — caller must abort.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="3") + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": True}}, + ): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result is None + + +def test_gate_on_no_input_returns_none(): + """No input (None / EOF / Ctrl-C) treated as cancel.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response=None) + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": True}}, + ): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result is None + + +def test_gate_on_unknown_choice_returns_none(): + """Garbage input is treated as cancel — fail safe, don't destroy state.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="maybe") + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": True}}, + ): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result is None + + +def test_gate_on_choice_always_persists_and_returns_always(): + """User picks 'always' → returns 'always' AND + save_config_value('approvals.destructive_slash_confirm', False) was called.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="2") + + saves = [] + + def _fake_save(key, value): + saves.append((key, value)) + return True + + with patch( + "cli.load_cli_config", + return_value={"approvals": {"destructive_slash_confirm": True}}, + ), patch("cli.save_config_value", _fake_save): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + assert result == "always" + assert ("approvals.destructive_slash_confirm", False) in saves + + +def test_gate_default_true_when_config_missing(): + """If load_cli_config raises or returns malformed data, treat as + 'gate on' (default safe) — must prompt.""" + from cli import HermesCLI + + self_ = _make_self(prompt_response="3") # cancel + + with patch("cli.load_cli_config", side_effect=Exception("boom")): + result = _bound(HermesCLI._confirm_destructive_slash, self_)( + "clear", "detail", + ) + + # Got prompted (returned None from cancel) — meaning the gate was + # treated as on despite the config error. If the gate had been off + # this would have returned 'once' without consulting the prompt. + assert result is None diff --git a/tests/gateway/test_destructive_slash_confirm.py b/tests/gateway/test_destructive_slash_confirm.py new file mode 100644 index 00000000000..a937852d0ea --- /dev/null +++ b/tests/gateway/test_destructive_slash_confirm.py @@ -0,0 +1,261 @@ +"""Tests for the gateway's destructive-slash-confirm wrapper. + +When ``approvals.destructive_slash_confirm`` is True (default), /new, +/reset, and /undo route through the slash-confirm primitive — native +yes/no buttons on Telegram/Discord/Slack, text fallback elsewhere. +When False (after "Always Approve"), the destructive action runs +immediately. +""" + +from __future__ import annotations + +from datetime import datetime +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from gateway.config import GatewayConfig, Platform, PlatformConfig +from gateway.platforms.base import MessageEvent +from gateway.session import SessionEntry, SessionSource, build_session_key + + +def _make_source() -> SessionSource: + return SessionSource( + platform=Platform.TELEGRAM, + user_id="u1", + chat_id="c1", + user_name="tester", + chat_type="dm", + ) + + +def _make_event(text: str) -> MessageEvent: + return MessageEvent(text=text, source=_make_source(), message_id="m1") + + +def _make_runner(): + """Mirror tests/gateway/test_unknown_command.py::_make_runner.""" + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner.config = GatewayConfig( + platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="***")} + ) + adapter = MagicMock() + adapter.send = AsyncMock() + # No send_slash_confirm override -> button render returns None, + # _request_slash_confirm falls back to text path. + adapter.send_slash_confirm = AsyncMock(return_value=None) + runner.adapters = {Platform.TELEGRAM: adapter} + + session_entry = SessionEntry( + session_key=build_session_key(_make_source()), + session_id="sess-1", + created_at=datetime.now(), + updated_at=datetime.now(), + platform=Platform.TELEGRAM, + chat_type="dm", + ) + runner.session_store = MagicMock() + runner.session_store.get_or_create_session.return_value = session_entry + runner.session_store.load_transcript.return_value = [] + runner.session_store.append_to_transcript = MagicMock() + runner.session_store.rewrite_transcript = MagicMock() + + runner._running_agents = {} + runner._pending_messages = {} + import itertools as _it + runner._slash_confirm_counter = _it.count(1) + runner.hooks = SimpleNamespace( + emit=AsyncMock(), + emit_collect=AsyncMock(return_value=[]), + loaded_hooks=False, + ) + runner._thread_metadata_for_source = lambda *a, **kw: None + runner._reply_anchor_for_event = lambda _e: None + return runner + + +@pytest.mark.asyncio +async def test_gate_off_runs_execute_immediately(monkeypatch): + """When approvals.destructive_slash_confirm is False, the destructive + action runs immediately without prompting.""" + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": False}} + runner._session_key_for_source = lambda src: build_session_key(src) + + sentinel = "✨ Session reset!" + execute = AsyncMock(return_value=sentinel) + + result = await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + execute.assert_awaited_once() + assert result == sentinel + + +@pytest.mark.asyncio +async def test_gate_on_text_fallback_returns_prompt_without_executing(monkeypatch): + """When the gate is on and the adapter has no button UI, the user gets + a text prompt back and the destructive action is NOT yet run.""" + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": True}} + runner._session_key_for_source = lambda src: build_session_key(src) + + execute = AsyncMock(return_value="should not run yet") + + result = await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + execute.assert_not_awaited() + assert isinstance(result, str) + assert "Confirm /new" in result + assert "Approve Once" in result + assert "Cancel" in result + + +@pytest.mark.asyncio +async def test_gate_on_pending_confirm_registered(monkeypatch): + """When the gate is on, a pending slash-confirm entry is registered for + the session — the user's /approve reply will resolve it.""" + from tools import slash_confirm as _slash_confirm_mod + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": True}} + session_key = build_session_key(_make_source()) + runner._session_key_for_source = lambda src: session_key + _slash_confirm_mod.clear(session_key) + + execute = AsyncMock(return_value="reset done") + + await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + pending = _slash_confirm_mod.get_pending(session_key) + assert pending is not None + assert pending["command"] == "new" + _slash_confirm_mod.clear(session_key) + + +@pytest.mark.asyncio +async def test_resolve_once_runs_execute_and_returns_result(): + """Resolving the pending confirm with 'once' runs the destructive + action and returns its output.""" + from tools import slash_confirm as _slash_confirm_mod + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": True}} + session_key = build_session_key(_make_source()) + runner._session_key_for_source = lambda src: session_key + _slash_confirm_mod.clear(session_key) + + execute = AsyncMock(return_value="✨ fresh session") + + await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + pending = _slash_confirm_mod.get_pending(session_key) + assert pending is not None + + resolved = await _slash_confirm_mod.resolve( + session_key, pending["confirm_id"], "once", + ) + + execute.assert_awaited_once() + assert resolved == "✨ fresh session" + # Pending should be cleared after resolve. + assert _slash_confirm_mod.get_pending(session_key) is None + + +@pytest.mark.asyncio +async def test_resolve_cancel_does_not_run_execute(): + """Resolving with 'cancel' must NOT run the destructive action.""" + from tools import slash_confirm as _slash_confirm_mod + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": True}} + session_key = build_session_key(_make_source()) + runner._session_key_for_source = lambda src: session_key + _slash_confirm_mod.clear(session_key) + + execute = AsyncMock(side_effect=AssertionError("execute must NOT run on cancel")) + + await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + pending = _slash_confirm_mod.get_pending(session_key) + assert pending is not None + + resolved = await _slash_confirm_mod.resolve( + session_key, pending["confirm_id"], "cancel", + ) + + execute.assert_not_awaited() + assert resolved is not None + assert "cancelled" in resolved.lower() + + +@pytest.mark.asyncio +async def test_resolve_always_persists_opt_out_and_runs_execute(monkeypatch): + """Resolving with 'always' must (a) flip the config gate to False, + (b) run execute, and (c) include a one-time opt-out note in the reply.""" + from tools import slash_confirm as _slash_confirm_mod + runner = _make_runner() + runner._read_user_config = lambda: {"approvals": {"destructive_slash_confirm": True}} + session_key = build_session_key(_make_source()) + runner._session_key_for_source = lambda src: session_key + _slash_confirm_mod.clear(session_key) + + saved: dict = {} + + def _fake_save(path, value): + saved[path] = value + return True + + import cli as cli_mod + monkeypatch.setattr(cli_mod, "save_config_value", _fake_save) + + execute = AsyncMock(return_value="✨ fresh") + + await runner._maybe_confirm_destructive_slash( + event=_make_event("/new"), + command="new", + title="/new", + detail="Discards history.", + execute=execute, + ) + + pending = _slash_confirm_mod.get_pending(session_key) + assert pending is not None + resolved = await _slash_confirm_mod.resolve( + session_key, pending["confirm_id"], "always", + ) + + execute.assert_awaited_once() + assert saved.get("approvals.destructive_slash_confirm") is False + assert resolved is not None + assert "✨ fresh" in resolved + assert "config.yaml" in resolved diff --git a/tests/gateway/test_telegram_topic_mode.py b/tests/gateway/test_telegram_topic_mode.py index 7c2171c0ae3..eeec2509962 100644 --- a/tests/gateway/test_telegram_topic_mode.py +++ b/tests/gateway/test_telegram_topic_mode.py @@ -144,6 +144,11 @@ def _make_runner(session_db=None): runner._invalidate_session_run_generation = MagicMock() runner._begin_session_run_generation = MagicMock(return_value=1) runner._is_session_run_current = MagicMock(return_value=True) + # Bypass the destructive-slash confirm gate — these tests focus on + # /new topic-mode mechanics, not the confirm prompt itself. + runner._read_user_config = lambda: { + "approvals": {"destructive_slash_confirm": False} + } runner._release_running_agent_state = MagicMock() runner._evict_cached_agent = MagicMock() runner._clear_session_boundary_security_state = MagicMock() diff --git a/tests/gateway/test_update_streaming.py b/tests/gateway/test_update_streaming.py index 36923bc5f05..b1681e1f349 100644 --- a/tests/gateway/test_update_streaming.py +++ b/tests/gateway/test_update_streaming.py @@ -45,6 +45,11 @@ def _make_runner(hermes_home=None): runner._pending_messages = {} runner._pending_approvals = {} runner._failed_platforms = {} + # Bypass the destructive-slash confirm gate — this test exercises + # update-prompt interception, not the confirm prompt. + runner._read_user_config = lambda: { + "approvals": {"destructive_slash_confirm": False} + } return runner diff --git a/tests/hermes_cli/test_destructive_slash_confirm_gate.py b/tests/hermes_cli/test_destructive_slash_confirm_gate.py new file mode 100644 index 00000000000..5f08518e1be --- /dev/null +++ b/tests/hermes_cli/test_destructive_slash_confirm_gate.py @@ -0,0 +1,86 @@ +"""Tests for the approvals.destructive_slash_confirm config gate. + +Destructive session slash commands (/clear, /new, /reset, /undo) discard +conversation state. This config key (default True) gates a three-option +confirmation prompt — "Always Approve" flips the key to False so future +destructive commands run silently. + +See gateway/run.py::_maybe_confirm_destructive_slash and +cli.py::_confirm_destructive_slash for the runtime gate. +""" + +from __future__ import annotations + +from hermes_cli.config import DEFAULT_CONFIG + + +class TestDestructiveSlashConfirmDefault: + def test_default_config_has_the_key(self): + approvals = DEFAULT_CONFIG.get("approvals") + assert isinstance(approvals, dict) + assert "destructive_slash_confirm" in approvals + + def test_default_is_true(self): + # New installs confirm by default — destructive commands must not + # silently wipe history without an explicit user "yes". + assert DEFAULT_CONFIG["approvals"]["destructive_slash_confirm"] is True + + def test_shape_matches_other_approval_keys(self): + approvals = DEFAULT_CONFIG["approvals"] + assert isinstance(approvals.get("destructive_slash_confirm"), bool) + # Sibling key shape sanity — same flat dict level as mcp_reload_confirm. + assert isinstance(approvals.get("mcp_reload_confirm"), bool) + + +class TestUserConfigMerge: + """If a user has a pre-existing config without this key, load_config + should fill it in from DEFAULT_CONFIG (deep merge preserves keys the + user didn't override).""" + + def test_existing_user_config_without_key_gets_default(self, tmp_path, monkeypatch): + import yaml + + home = tmp_path / ".hermes" + home.mkdir() + cfg_path = home / "config.yaml" + legacy = { + "approvals": {"mode": "manual", "timeout": 60, "cron_mode": "deny"}, + } + cfg_path.write_text(yaml.safe_dump(legacy)) + + monkeypatch.setenv("HERMES_HOME", str(home)) + import importlib + import hermes_cli.config as cfg_mod + importlib.reload(cfg_mod) + + cfg = cfg_mod.load_config() + assert cfg["approvals"]["destructive_slash_confirm"] is True + + def test_existing_user_config_with_false_key_survives_merge( + self, tmp_path, monkeypatch, + ): + """A user who clicked "Always Approve" (key=false) must keep that + setting — the default-true value must not win on later loads. + """ + import yaml + + home = tmp_path / ".hermes" + home.mkdir() + cfg_path = home / "config.yaml" + user_cfg = { + "approvals": { + "mode": "manual", + "timeout": 60, + "cron_mode": "deny", + "destructive_slash_confirm": False, + }, + } + cfg_path.write_text(yaml.safe_dump(user_cfg)) + + monkeypatch.setenv("HERMES_HOME", str(home)) + import importlib + import hermes_cli.config as cfg_mod + importlib.reload(cfg_mod) + + cfg = cfg_mod.load_config() + assert cfg["approvals"]["destructive_slash_confirm"] is False