feat: confirm prompt for destructive slash commands (#4069) (#22687)

/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.

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
 
 

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

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

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()

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
 
 

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