feat(approvals): add cross-surface mode command

Co-authored-by: luxiaolu4827 <227715866+luxiaolu4827@users.noreply.github.com>
This commit is contained in:
Teknium 2026-07-12 19:14:01 -07:00
parent 443ce196f4
commit f9cd577915
12 changed files with 452 additions and 2 deletions

View file

@ -929,6 +929,47 @@ describe('usePromptActions slash.exec dispatch payloads', () => {
vi.restoreAllMocks()
})
it('executes /approvals against the focused profile session and persists its mode', async () => {
const focusedProfile = 'work'
const focusedSessionId = 'work-runtime-session'
const persistedModes = new Map<string, string>()
const sessionProfiles = new Map([[focusedSessionId, focusedProfile]])
const requestGateway = vi.fn(async (method: string, params?: Record<string, unknown>) => {
if (method === 'slash.exec') {
const sessionId = String(params?.session_id ?? '')
const profile = sessionProfiles.get(sessionId)
const command = String(params?.command ?? '')
if (profile && command === 'approvals off') persistedModes.set(profile, 'off')
return { output: 'Approval mode: off (persistent profile setting).' } as never
}
return {} as never
})
let handle: HarnessHandle | null = null
render(
<Harness
activeSessionId={focusedSessionId}
activeSessionIdRef={{ current: focusedSessionId }}
onReady={h => (handle = h)}
refreshSessions={async () => undefined}
requestGateway={requestGateway}
storedSessionId={focusedSessionId}
/>
)
await handle!.submitText('/approvals off')
expect(requestGateway).toHaveBeenCalledWith('slash.exec', {
command: 'approvals off',
session_id: focusedSessionId
})
expect(persistedModes.get(focusedProfile)).toBe('off')
expect(persistedModes.has('default')).toBe(false)
})
it('submits /goal send directives returned directly by slash.exec instead of rendering no output', async () => {
const calls: { method: string; params?: Record<string, unknown> }[] = []
const states: Record<string, unknown>[] = []

View file

@ -21,6 +21,9 @@ describe('desktop slash command curation', () => {
expect(isDesktopSlashSuggestion('/version')).toBe(true)
expect(isDesktopSlashSuggestion('/yolo')).toBe(true)
expect(isDesktopSlashCommand('/yolo')).toBe(true)
expect(isDesktopSlashSuggestion('/approvals')).toBe(true)
expect(isDesktopSlashCommand('/approvals')).toBe(true)
expect(resolveDesktopCommand('/approvals')?.surface).toEqual({ kind: 'exec' })
})
it('surfaces skill and quick commands (extensions) in suggestions and lets them run', () => {

View file

@ -190,6 +190,12 @@ const DESKTOP_COMMAND_SPECS: readonly DesktopCommandSpec[] = [
// them, /steer falls back to a next-turn prompt, and /usage is a formatted
// live report. Keep them on slash.exec until their RPC contracts are fully
// equivalent.
{
name: '/approvals',
description: 'Show or set approval mode [manual|smart|off]',
surface: exec(),
args: true
},
{
name: '/agents',
description: 'Show active desktop sessions and running tasks',

2
cli.py
View file

@ -9701,6 +9701,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin):
self._handle_footer_command(cmd_original)
elif canonical == "yolo":
self._toggle_yolo()
elif canonical == "approvals":
self._handle_approvals_command(cmd_original)
elif canonical == "reasoning":
self._handle_reasoning_command(cmd_original)
elif canonical == "fast":

View file

@ -11977,6 +11977,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
if canonical == "yolo":
return await self._handle_yolo_command(event)
if canonical == "approvals":
return await self._handle_approvals_command(event)
if canonical == "model":
return await self._handle_model_command(event)

View file

@ -3649,6 +3649,23 @@ class GatewaySlashCommandsMixin:
return _apply_fast_selection(args, persist=persist_global)
async def _handle_approvals_command(self, event: MessageEvent) -> str:
"""Show or persist the profile-wide dangerous-command approval mode."""
from gateway.slash_access import policy_for_source
from hermes_cli.approval_mode import run_approval_mode_command
requested = event.get_command_args().strip() or None
# This mutates profile-wide security policy. The central slash gate can
# allow selected commands to non-admin users, so enforce admin again at
# this side-effect boundary. Unconfigured policies remain unrestricted.
policy = policy_for_source(self.config, event.source)
if requested and not policy.is_admin(event.source.user_id):
return "Only gateway admins can change the persistent approval mode."
result = run_approval_mode_command(requested)
# Approval checks load config dynamically; do not evict the cached agent
# or alter its system prompt/tool schema (prompt-cache prefix is sacred).
return result.message
async def _handle_yolo_command(self, event: MessageEvent) -> Union[str, EphemeralReply]:
"""Handle /yolo — toggle dangerous command approval bypass for this session only."""
from tools.approval import (

View file

@ -0,0 +1,87 @@
"""Shared persistent approval-mode command logic.
Approval mode is profile-scoped configuration, not conversation state. Changing
it affects subsequent terminal guard checks immediately because approval.py
loads config on each check; it must not rebuild a live agent or mutate its
system prompt/tool schema, preserving the prompt-cache prefix.
"""
from __future__ import annotations
from contextlib import redirect_stderr, redirect_stdout
from dataclasses import dataclass
from io import StringIO
from typing import Optional
VALID_APPROVAL_MODES = ("manual", "smart", "off")
@dataclass(frozen=True)
class ApprovalModeResult:
ok: bool
mode: str
changed: bool
message: str
def _effective_mode() -> str:
"""Return the exact mode enforced by the terminal approval guard."""
from tools.approval import _get_approval_mode
return _get_approval_mode()
def run_approval_mode_command(requested_mode: Optional[str]) -> ApprovalModeResult:
"""Inspect or persist ``approvals.mode`` through canonical config APIs."""
current = _effective_mode()
requested = (requested_mode or "").strip().lower()
if not requested:
return ApprovalModeResult(
True,
current,
False,
f"Approval mode: {current} (persistent profile setting).",
)
if requested not in VALID_APPROVAL_MODES:
return ApprovalModeResult(
False,
current,
False,
"Usage: /approvals [manual|smart|off]",
)
# set_config_value is the canonical managed-scope/write-safety chokepoint.
# It reports managed policy through stderr + SystemExit, so capture that for
# slash-command output instead of terminating the interactive worker.
from hermes_cli.config import set_config_value
output = StringIO()
try:
with redirect_stdout(output), redirect_stderr(output):
set_config_value("approvals.mode", requested)
except SystemExit:
detail = output.getvalue().strip() or "Approval mode is managed and cannot be changed."
return ApprovalModeResult(False, current, False, detail)
except Exception as exc:
return ApprovalModeResult(
False,
current,
False,
f"Failed to save approval mode: {exc}",
)
effective = _effective_mode()
if effective != requested:
return ApprovalModeResult(
False,
effective,
False,
f"Approval mode remains {effective}; the requested value did not become effective.",
)
return ApprovalModeResult(
True,
effective,
effective != current,
f"Approval mode: {effective} (persistent profile setting).",
)

View file

@ -2778,6 +2778,16 @@ class CLICommandsMixin:
except Exception:
pass
def _handle_approvals_command(self, cmd_original: str) -> None:
"""Show or persist the profile-wide dangerous-command approval mode."""
from cli import _cprint
from hermes_cli.approval_mode import run_approval_mode_command
parts = (cmd_original or "").strip().split(None, 1)
requested = parts[1] if len(parts) > 1 else None
result = run_approval_mode_command(requested)
_cprint(f" {result.message}")
def _handle_footer_command(self, cmd_original: str) -> None:
"""Toggle or inspect ``display.runtime_footer.enabled`` from the CLI.

View file

@ -165,6 +165,9 @@ COMMAND_REGISTRY: list[CommandDef] = [
subcommands=("on", "off", "status")),
CommandDef("yolo", "Toggle YOLO mode (skip all dangerous command approvals)",
"Configuration"),
CommandDef("approvals", "Show or set the persistent dangerous-command approval mode",
"Configuration", args_hint="[manual|smart|off]",
subcommands=("manual", "smart", "off")),
CommandDef("reasoning", "Manage reasoning effort and display", "Configuration",
args_hint="[level|show|hide|full|clamp] [--global]",
subcommands=("none", "minimal", "low", "medium", "high", "xhigh", "max", "ultra", "show", "hide", "on", "off", "full", "clamp", "--global")),
@ -1187,10 +1190,15 @@ _SLACK_PRIORITY_ALIASES = ("btw", "bg")
# /init clamps /version off the native list and breaks Telegram parity.
# - version: low-frequency info command; reachable as /hermes version on
# Slack. Demoted when /context claimed a native slot (context is a
# recurring inspection surface; version is a one-off lookup).
# recurring inspection surface; version is a one-off lookup); the demotion
# also absorbs the native slot /approvals now consumes at the 50-cap.
# - diff: git working-tree diff; reached via /hermes diff on Slack so it
# doesn't displace an existing native slash at the 50-command cap.
_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff"})
# - update: low-frequency self-update maintenance command; reached via
# /hermes update on Slack. Demoted to free the native slot /approvals now
# claims — without this entry /approvals tips the registry past the 50-cap
# and silently clamps /update off, breaking Telegram parity.
_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update"})
def _sanitize_slack_name(raw: str) -> str:

View file

@ -0,0 +1,91 @@
"""Gateway contract and live dispatch for /approvals."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import yaml
import gateway.run as gateway_run
from gateway.config import Platform
from gateway.platforms.base import MessageEvent
from gateway.session import SessionSource
def _event(text: str = "/approvals") -> MessageEvent:
return MessageEvent(
text=text,
source=SessionSource(
platform=Platform.TELEGRAM,
user_id="user-1",
chat_id="chat-1",
chat_type="dm",
),
)
def _runner():
runner = object.__new__(gateway_run.GatewayRunner)
runner.config = SimpleNamespace(platforms={})
runner.hooks = MagicMock(loaded_hooks=[])
runner.hooks.emit = AsyncMock(return_value=[])
runner._running_agents = {}
runner._get_or_create_gateway_honcho = lambda _key: (None, None)
runner._is_user_authorized = lambda _source: True
runner.session_store = SimpleNamespace(get_or_create_session=lambda _source: None)
return runner
@pytest.mark.asyncio
async def test_gateway_handler_uses_shared_persistent_logic_without_cache_eviction():
runner = _runner()
result = SimpleNamespace(message="Approval mode: manual (persistent profile setting).")
runner._evict_cached_agent = MagicMock()
with patch("hermes_cli.approval_mode.run_approval_mode_command", return_value=result) as run:
output = await runner._handle_approvals_command(_event("/approvals manual"))
assert output == result.message
run.assert_called_once_with("manual")
runner._evict_cached_agent.assert_not_called()
@pytest.mark.asyncio
async def test_gateway_rejects_non_admin_persistent_approval_change():
runner = _runner()
runner.config = SimpleNamespace(
platforms={
Platform.TELEGRAM: SimpleNamespace(
extra={
"allow_admin_from": ["admin-1"],
"user_allowed_commands": ["approvals"],
}
)
}
)
with patch("hermes_cli.approval_mode.run_approval_mode_command") as run:
output = await runner._handle_approvals_command(_event("/approvals off"))
assert "admin" in output.lower()
run.assert_not_called()
@pytest.mark.asyncio
async def test_gateway_live_dispatch_routes_and_persists_approvals_command(tmp_path, monkeypatch):
runner = _runner()
home = tmp_path / "home"
home.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_MANAGED_DIR", str(tmp_path / "missing-managed"))
from hermes_cli import managed_scope
from hermes_cli.config import _LOAD_CONFIG_CACHE, _RAW_CONFIG_CACHE
_LOAD_CONFIG_CACHE.clear()
_RAW_CONFIG_CACHE.clear()
managed_scope.invalidate_managed_cache()
output = await runner._handle_message(_event("/approvals manual"))
assert output == "Approval mode: manual (persistent profile setting)."
assert yaml.safe_load((home / "config.yaml").read_text())["approvals"]["mode"] == "manual"

View file

@ -0,0 +1,180 @@
"""Cross-surface contract for the persistent /approvals mode command."""
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import yaml
from cli import HermesCLI
from hermes_cli.commands import (
GATEWAY_KNOWN_COMMANDS,
SUBCOMMANDS,
SlashCommandCompleter,
gateway_help_lines,
resolve_command,
telegram_bot_commands,
)
from prompt_toolkit.completion import CompleteEvent
from prompt_toolkit.document import Document
def _completions(text: str) -> set[str]:
return {
item.text
for item in SlashCommandCompleter().get_completions(
Document(text=text), CompleteEvent(completion_requested=True)
)
}
def test_approvals_registry_drives_help_menu_and_autocomplete():
command = resolve_command("approvals")
assert command is not None
assert command.category == "Configuration"
assert command.args_hint == "[manual|smart|off]"
assert SUBCOMMANDS["/approvals"] == ["manual", "smart", "off"]
assert "approvals" in GATEWAY_KNOWN_COMMANDS
assert any("/approvals" in line for line in gateway_help_lines())
assert "approvals" in {name for name, _ in telegram_bot_commands()}
assert _completions("/approvals ") == {"manual", "smart", "off"}
def _isolate_config(monkeypatch, home):
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_MANAGED_DIR", str(home / "missing-managed"))
from hermes_cli import managed_scope
from hermes_cli.config import _LOAD_CONFIG_CACHE, _RAW_CONFIG_CACHE
_LOAD_CONFIG_CACHE.clear()
_RAW_CONFIG_CACHE.clear()
managed_scope.invalidate_managed_cache()
def test_shared_approval_mode_command_reports_effective_default_without_writing(tmp_path, monkeypatch):
from hermes_cli.approval_mode import run_approval_mode_command
from tools.approval import _get_approval_mode
_isolate_config(monkeypatch, tmp_path)
result = run_approval_mode_command(None)
assert result.ok is True
assert result.mode == _get_approval_mode()
assert result.changed is False
assert result.mode in result.message
assert not (tmp_path / "config.yaml").exists()
def test_shared_approval_mode_command_persists_profile_setting(tmp_path, monkeypatch):
from hermes_cli.approval_mode import run_approval_mode_command
_isolate_config(monkeypatch, tmp_path)
path = tmp_path / "config.yaml"
path.write_text("model:\n default: test-model\n", encoding="utf-8")
result = run_approval_mode_command("off")
assert result.ok is True
assert result.mode == "off"
assert result.changed is True
assert yaml.safe_load(path.read_text(encoding="utf-8"))["approvals"]["mode"] in {"off", False}
assert "persistent" in result.message.lower()
def test_shared_approval_mode_command_rejects_unknown_mode_without_writing(tmp_path, monkeypatch):
from hermes_cli.approval_mode import run_approval_mode_command
_isolate_config(monkeypatch, tmp_path)
path = tmp_path / "config.yaml"
path.write_text("approvals:\n mode: smart\n", encoding="utf-8")
before = path.read_bytes()
result = run_approval_mode_command("auto")
assert result.ok is False
assert result.mode == "smart"
assert path.read_bytes() == before
assert "manual|smart|off" in result.message
def test_shared_status_matches_runtime_normalization_for_all_stored_shapes():
from hermes_cli.approval_mode import run_approval_mode_command
from tools.approval import _get_approval_mode
for stored in (None, "manual", "smart", "off", False, True, "", "auto"):
config = {"approvals": {}} if stored is None else {"approvals": {"mode": stored}}
with patch("hermes_cli.config.load_config", return_value=config):
result = run_approval_mode_command(None)
assert result.mode == _get_approval_mode(), stored
def test_shared_command_refuses_managed_mode_override(tmp_path, monkeypatch):
from hermes_cli import managed_scope
from hermes_cli.approval_mode import run_approval_mode_command
home = tmp_path / "home"
managed = tmp_path / "managed"
home.mkdir()
managed.mkdir()
monkeypatch.setenv("HERMES_HOME", str(home))
monkeypatch.setenv("HERMES_MANAGED_DIR", str(managed))
(managed / "config.yaml").write_text("approvals:\n mode: manual\n", encoding="utf-8")
managed_scope.invalidate_managed_cache()
result = run_approval_mode_command("off")
assert result.ok is False
assert result.mode == "manual"
assert result.changed is False
assert "managed" in result.message.lower()
assert not (home / "config.yaml").exists()
def test_cli_dispatch_uses_shared_handler_without_rebuilding_agent():
cli = HermesCLI.__new__(HermesCLI)
cli.config = {}
cli.console = MagicMock()
cli.agent = object()
cli._agent_running = False
cli._pending_input = MagicMock()
with patch.object(cli, "_handle_approvals_command", create=True) as handler:
assert cli.process_command("/approvals manual") is True
handler.assert_called_once_with("/approvals manual")
assert cli.agent is not None
def test_cli_handler_prints_shared_result_and_preserves_agent_cache():
cli = HermesCLI.__new__(HermesCLI)
cached_agent = object()
cli.agent = cached_agent
result = SimpleNamespace(message="Approval mode: smart (persistent profile setting).")
with (
patch("hermes_cli.approval_mode.run_approval_mode_command", return_value=result) as run,
patch("cli._cprint") as output,
):
cli._handle_approvals_command("/approvals smart")
run.assert_called_once_with("smart")
output.assert_called_once_with(" Approval mode: smart (persistent profile setting).")
assert cli.agent is cached_agent
def test_cli_live_process_command_persists_mode(tmp_path, monkeypatch):
cli = HermesCLI.__new__(HermesCLI)
cached_agent = object()
cli.config = {}
cli.console = MagicMock()
cli.agent = cached_agent
cli._agent_running = False
cli._pending_input = MagicMock()
_isolate_config(monkeypatch, tmp_path)
with patch("cli._cprint") as output:
assert cli.process_command("/approvals off") is True
stored = yaml.safe_load((tmp_path / "config.yaml").read_text())["approvals"]["mode"]
assert stored in {"off", False}
assert "persistent profile setting" in output.call_args.args[0]
assert cli.agent is cached_agent

View file

@ -7179,6 +7179,8 @@ def test_commands_catalog_filters_gateway_only_commands_and_keeps_status_visible
assert "/status" in pairs
assert canon["/status"] == "/status"
assert "/approvals" in pairs
assert resp["result"]["sub"]["/approvals"] == ["manual", "smart", "off"]
assert "/topic" not in pairs
assert "/approve" not in pairs