{
export interface ApprovalRequest extends KeyedPrompt {
// false when the backend won't honor a permanent allow (tirith warning) → hide "Always allow".
allowPermanent?: boolean
+ choices?: string[]
command: string
description: string
+ smartDenied?: boolean
}
export interface SudoRequest extends KeyedPrompt {
diff --git a/cli.py b/cli.py
index 560b5d9a2e..37c158dabf 100644
--- a/cli.py
+++ b/cli.py
@@ -11683,13 +11683,15 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
return ""
def _approval_callback(self, command: str, description: str,
- *, allow_permanent: bool = True) -> str:
+ *, allow_permanent: bool = True,
+ smart_denied: bool = False) -> str:
"""
Prompt for dangerous command approval through the prompt_toolkit UI.
Called from the agent thread. Shows a selection UI similar to clarify
- with choices: once / session / always / deny. When allow_permanent
- is False (tirith warnings present), the 'always' option is hidden.
+ with choices: once / session / always / deny. Smart DENY owner
+ overrides show only once / deny. When allow_permanent is False for
+ another reason (for example tirith), only 'always' is hidden.
Long commands also get a 'view' option so the full command can be
expanded before deciding.
@@ -11706,7 +11708,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
self._approval_state = {
"command": command,
"description": description,
- "choices": self._approval_choices(command, allow_permanent=allow_permanent),
+ "choices": self._approval_choices(
+ command,
+ allow_permanent=allow_permanent,
+ smart_denied=smart_denied,
+ ),
"selected": 0,
"response_queue": response_queue,
}
@@ -11752,9 +11758,13 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
_cprint(f"\n{_DIM} ⏱ Timeout — denying command{_RST}")
return "deny"
- def _approval_choices(self, command: str, *, allow_permanent: bool = True) -> list[str]:
+ def _approval_choices(self, command: str, *, allow_permanent: bool = True,
+ smart_denied: bool = False) -> list[str]:
"""Return approval choices for a dangerous command prompt."""
- choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
+ if smart_denied:
+ choices = ["once", "deny"]
+ else:
+ choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
if len(command) > 70:
choices.append("view")
return choices
diff --git a/gateway/platforms/api_server.py b/gateway/platforms/api_server.py
index 59deed4e7f..b96adf9e44 100644
--- a/gateway/platforms/api_server.py
+++ b/gateway/platforms/api_server.py
@@ -45,6 +45,12 @@ import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
+def _approval_event_choices(*, smart_denied: bool, allow_permanent: bool) -> list[str]:
+ if smart_denied:
+ return ["once", "deny"]
+ return ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
+
+
try:
from aiohttp import web
AIOHTTP_AVAILABLE = True
@@ -4375,7 +4381,10 @@ class APIServerAdapter(BasePlatformAdapter):
"event": "approval.request",
"run_id": run_id,
"timestamp": time.time(),
- "choices": ["once", "session", "always", "deny"],
+ "choices": _approval_event_choices(
+ smart_denied=bool(event.get("smart_denied")),
+ allow_permanent=event.get("allow_permanent") is not False,
+ ),
})
self._set_run_status(
run_id,
diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py
index 865a38063a..5447d99d99 100644
--- a/gateway/platforms/qqbot/adapter.py
+++ b/gateway/platforms/qqbot/adapter.py
@@ -2648,7 +2648,10 @@ class QQAdapter(BasePlatformAdapter):
return await self.send_with_keyboard(
chat_id,
build_approval_text(req),
- build_approval_keyboard(req.session_key),
+ build_approval_keyboard(
+ req.session_key,
+ allow_permanent=getattr(req, "allow_permanent", True),
+ ),
reply_to=reply_to,
)
@@ -2668,6 +2671,8 @@ class QQAdapter(BasePlatformAdapter):
session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send a button-based exec-approval prompt for a dangerous command.
@@ -2677,6 +2682,8 @@ class QQAdapter(BasePlatformAdapter):
adapter's interaction callback (:meth:`_default_interaction_dispatch`).
"""
del metadata # QQ doesn't have thread_id / DM targeting overrides.
+ if smart_denied:
+ description += " Owner override applies to this one operation only."
# Use the reply-to message for passive-message context when we have one.
# QQ requires a msg_id on outbound messages to a user we've never
@@ -2689,6 +2696,7 @@ class QQAdapter(BasePlatformAdapter):
description=description,
command_preview=command,
timeout_sec=self._APPROVAL_TIMEOUT_SECONDS,
+ allow_permanent=allow_permanent and not smart_denied,
)
return await self.send_approval_request(
chat_id, req, reply_to=msg_id,
diff --git a/gateway/platforms/qqbot/keyboards.py b/gateway/platforms/qqbot/keyboards.py
index 19fd36e370..9c1afbbfa3 100644
--- a/gateway/platforms/qqbot/keyboards.py
+++ b/gateway/platforms/qqbot/keyboards.py
@@ -201,8 +201,8 @@ def _make_callback_button(
)
-def build_approval_keyboard(session_key: str) -> InlineKeyboard:
- """Build the 3-button approval keyboard.
+def build_approval_keyboard(session_key: str, *, allow_permanent: bool = True) -> InlineKeyboard:
+ """Build the approval keyboard, hiding persistent scope when unavailable.
Layout: ``[✅ 允许一次] [⭐ 始终允许] [❌ 拒绝]`` — all three share
``group_id='approval'`` so clicking one greys out the rest.
@@ -210,38 +210,25 @@ def build_approval_keyboard(session_key: str) -> InlineKeyboard:
:param session_key: Embedded into ``button_data`` so the decision
routes back to the right pending approval.
"""
- return InlineKeyboard(
- content=KeyboardContent(
- rows=[
- KeyboardRow(buttons=[
- _make_callback_button(
- btn_id="allow",
- label="✅ 允许一次",
- visited_label="已允许",
- data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-once",
- style=1,
- group_id="approval",
- ),
- _make_callback_button(
- btn_id="always",
- label="⭐ 始终允许",
- visited_label="已始终允许",
- data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-always",
- style=1,
- group_id="approval",
- ),
- _make_callback_button(
- btn_id="deny",
- label="❌ 拒绝",
- visited_label="已拒绝",
- data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:deny",
- style=0,
- group_id="approval",
- ),
- ]),
- ]
+ buttons = [
+ _make_callback_button(
+ btn_id="allow", label="✅ 允许一次", visited_label="已允许",
+ data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-once",
+ style=1, group_id="approval",
)
- )
+ ]
+ if allow_permanent:
+ buttons.append(_make_callback_button(
+ btn_id="always", label="⭐ 始终允许", visited_label="已始终允许",
+ data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-always",
+ style=1, group_id="approval",
+ ))
+ buttons.append(_make_callback_button(
+ btn_id="deny", label="❌ 拒绝", visited_label="已拒绝",
+ data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:deny",
+ style=0, group_id="approval",
+ ))
+ return InlineKeyboard(content=KeyboardContent(rows=[KeyboardRow(buttons=buttons)]))
def build_update_prompt_keyboard() -> InlineKeyboard:
@@ -295,6 +282,7 @@ class ApprovalRequest:
tool_name: str = ""
severity: str = ""
timeout_sec: int = 120
+ allow_permanent: bool = True
def build_approval_text(req: ApprovalRequest) -> str:
diff --git a/gateway/platforms/whatsapp_cloud.py b/gateway/platforms/whatsapp_cloud.py
index bd26252f12..b494fc15e5 100644
--- a/gateway/platforms/whatsapp_cloud.py
+++ b/gateway/platforms/whatsapp_cloud.py
@@ -819,6 +819,8 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Render a dangerous-command approval prompt with native buttons.
@@ -830,6 +832,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
if self._http_client is None:
return SendResult(success=False, error="Not connected")
+ del allow_permanent # This adapter already offers one-shot Approve / Deny only.
# WhatsApp body caps at 1024 chars; reserve room for the
# framing prose around the command.
cmd = command or ""
@@ -838,6 +841,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
f"⚠️ *Command Approval Required*\n\n"
f"```\n{cmd_preview}\n```\n\n"
f"Reason: {description}"
+ + ("\n\nSmart DENY: owner override applies to this one operation only." if smart_denied else "")
)
approval_id = uuid.uuid4().hex[:12]
diff --git a/gateway/run.py b/gateway/run.py
index ef49bf67c4..c038f8c14e 100644
--- a/gateway/run.py
+++ b/gateway/run.py
@@ -353,6 +353,34 @@ def _redact_approval_command(cmd: "str | None") -> str:
return redact_sensitive_text(str(cmd or ""), force=True)
+def _format_exec_approval_fallback(
+ command: str,
+ description: str,
+ command_prefix: str,
+ *,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
+) -> str:
+ """Render the text fallback from approval capabilities, not platform names."""
+ cmd_preview = command[:200] + "..." if len(command) > 200 else command
+ heading = "⚠️ **Dangerous command requires approval:**"
+ if smart_denied:
+ heading = "⚠️ **Smart DENY — owner override for one operation:**"
+
+ choices = [f"Reply `{command_prefix}approve` to execute this one operation"]
+ if not smart_denied:
+ choices.append(
+ f"`{command_prefix}approve session` to approve this pattern for the session"
+ )
+ if allow_permanent:
+ choices.append(f"`{command_prefix}approve always` to approve permanently")
+ choices.append(f"`{command_prefix}deny` to cancel")
+ return (
+ f"{heading}\n```\n{cmd_preview}\n```\nReason: {description}\n\n"
+ + ", ".join(choices[:-1]) + f", or {choices[-1]}."
+ )
+
+
def _gateway_provider_error_reply(text: str) -> str:
"""Map raw provider/API errors to a short user-safe Telegram reply."""
if _GATEWAY_AUTH_ERROR_RE.search(text):
@@ -18641,6 +18669,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
session_key=_approval_session_key,
description=desc,
metadata=_status_thread_metadata,
+ allow_permanent=approval_data.get("allow_permanent", True),
+ smart_denied=approval_data.get("smart_denied", False),
),
_loop_for_step,
logger=logger,
@@ -18665,13 +18695,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
# can actually type (`!approve`) — typed "/" is blocked in
# Slack threads and reserved by Matrix clients.
_p = getattr(_status_adapter, "typed_command_prefix", "/")
- cmd_preview = cmd[:200] + "..." if len(cmd) > 200 else cmd
- msg = (
- f"⚠️ **Dangerous command requires approval:**\n"
- f"```\n{cmd_preview}\n```\n"
- f"Reason: {desc}\n\n"
- f"Reply `{_p}approve` to execute, `{_p}approve session` to approve this pattern "
- f"for the session, `{_p}approve always` to approve permanently, or `{_p}deny` to cancel."
+ msg = _format_exec_approval_fallback(
+ cmd,
+ desc,
+ _p,
+ allow_permanent=approval_data.get("allow_permanent", True),
+ smart_denied=approval_data.get("smart_denied", False),
)
try:
_approval_send_fut = safe_schedule_threadsafe(
diff --git a/locales/af.yaml b/locales/af.yaml
index 6dc9055def..eecc3bdcf8 100644
--- a/locales/af.yaml
+++ b/locales/af.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]eenmalig | [s]sessie | [d]weier"
prompt_long: " Keuse [o/s/a/D]: "
prompt_short: " Keuse [o/s/D]: "
+ choose_smart_deny: " [o]eenmalig | [d]weier"
+ prompt_smart_deny: " Keuse [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Tyd verstreke - opdrag word geweier"
allowed_once: " ✓ Eenmalig toegelaat"
allowed_session: " ✓ Vir hierdie sessie toegelaat"
diff --git a/locales/de.yaml b/locales/de.yaml
index 26acd3eb35..0d85397785 100644
--- a/locales/de.yaml
+++ b/locales/de.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]einmal | [s]sitzung | [d]ablehnen"
prompt_long: " Auswahl [o/s/a/D]: "
prompt_short: " Auswahl [o/s/D]: "
+ choose_smart_deny: " [o]einmal | [d]ablehnen"
+ prompt_smart_deny: " Auswahl [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Zeitüberschreitung – Befehl wird abgelehnt"
allowed_once: " ✓ Einmalig erlaubt"
allowed_session: " ✓ Für diese Sitzung erlaubt"
diff --git a/locales/en.yaml b/locales/en.yaml
index f970b057ec..852ce88b41 100644
--- a/locales/en.yaml
+++ b/locales/en.yaml
@@ -17,6 +17,10 @@ approval:
choose_short: " [o]nce | [s]ession | [d]eny"
prompt_long: " Choice [o/s/a/D]: "
prompt_short: " Choice [o/s/D]: "
+ choose_smart_deny: " [o]nce | [d]eny"
+ prompt_smart_deny: " Choice [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Timeout - denying command"
allowed_once: " ✓ Allowed once"
allowed_session: " ✓ Allowed for this session"
diff --git a/locales/es.yaml b/locales/es.yaml
index 15eedaa869..aaf0f1e14a 100644
--- a/locales/es.yaml
+++ b/locales/es.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]una vez | [s]sesión | [d]denegar"
prompt_long: " Opción [o/s/a/D]: "
prompt_short: " Opción [o/s/D]: "
+ choose_smart_deny: " [o]una vez | [d]denegar"
+ prompt_smart_deny: " Opción [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Tiempo agotado — comando denegado"
allowed_once: " ✓ Permitido una vez"
allowed_session: " ✓ Permitido en esta sesión"
diff --git a/locales/fr.yaml b/locales/fr.yaml
index 78935eed55..6173e684a1 100644
--- a/locales/fr.yaml
+++ b/locales/fr.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]ne fois | [s]ession | [r]efuser"
prompt_long: " Choix [o/s/t/R] : "
prompt_short: " Choix [o/s/R] : "
+ choose_smart_deny: " [o]ne fois | [r]efuser"
+ prompt_smart_deny: " Choix [o/R] : "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "r,refuser"
timeout: " ⏱ Délai dépassé — commande refusée"
allowed_once: " ✓ Autorisé une fois"
allowed_session: " ✓ Autorisé pour cette session"
diff --git a/locales/ga.yaml b/locales/ga.yaml
index bad263ecfc..7317c2eba1 100644
--- a/locales/ga.yaml
+++ b/locales/ga.yaml
@@ -11,6 +11,10 @@ approval:
choose_short: " [o]uair amháin | [s]eisiún | [d]iúltaigh"
prompt_long: " Rogha [o/s/a/D]: "
prompt_short: " Rogha [o/s/D]: "
+ choose_smart_deny: " [o]uair amháin | [d]iúltaigh"
+ prompt_smart_deny: " Rogha [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Am istigh — ag diúltú don ordú"
allowed_once: " ✓ Ceadaithe uair amháin"
allowed_session: " ✓ Ceadaithe don seisiún seo"
diff --git a/locales/hu.yaml b/locales/hu.yaml
index e3bbc6ea60..547aa87676 100644
--- a/locales/hu.yaml
+++ b/locales/hu.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]egyszer | [s]munkamenet | [d]elutasít"
prompt_long: " Választás [o/s/a/D]: "
prompt_short: " Választás [o/s/D]: "
+ choose_smart_deny: " [o]egyszer | [d]elutasít"
+ prompt_smart_deny: " Választás [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Időtúllépés - parancs elutasítva"
allowed_once: " ✓ Egyszer engedélyezve"
allowed_session: " ✓ Engedélyezve ehhez a munkamenethez"
diff --git a/locales/it.yaml b/locales/it.yaml
index f747409053..0cbdcf68fa 100644
--- a/locales/it.yaml
+++ b/locales/it.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]una volta | [s]essione | [d]nega"
prompt_long: " Scelta [o/s/a/D]: "
prompt_short: " Scelta [o/s/D]: "
+ choose_smart_deny: " [o]una volta | [d]nega"
+ prompt_smart_deny: " Scelta [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Tempo scaduto — comando negato"
allowed_once: " ✓ Consentito una volta"
allowed_session: " ✓ Consentito per questa sessione"
diff --git a/locales/ja.yaml b/locales/ja.yaml
index f11725d7ad..38f8822b71 100644
--- a/locales/ja.yaml
+++ b/locales/ja.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]今回のみ | [s]セッション中 | [d]拒否"
prompt_long: " 選択 [o/s/a/D]: "
prompt_short: " 選択 [o/s/D]: "
+ choose_smart_deny: " [o]今回のみ | [d]拒否"
+ prompt_smart_deny: " 選択 [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ タイムアウト — コマンドを拒否しました"
allowed_once: " ✓ 今回のみ許可"
allowed_session: " ✓ このセッション中は許可"
diff --git a/locales/ko.yaml b/locales/ko.yaml
index aae3358bdc..752012b0fd 100644
--- a/locales/ko.yaml
+++ b/locales/ko.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]한 번 | [s]세션 | [d]거부"
prompt_long: " 선택 [o/s/a/D]: "
prompt_short: " 선택 [o/s/D]: "
+ choose_smart_deny: " [o]한 번 | [d]거부"
+ prompt_smart_deny: " 선택 [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ 시간 초과 - 명령을 거부합니다"
allowed_once: " ✓ 한 번 허용됨"
allowed_session: " ✓ 이 세션에서 허용됨"
diff --git a/locales/pt.yaml b/locales/pt.yaml
index 2f8bcd03d4..12d53e995d 100644
--- a/locales/pt.yaml
+++ b/locales/pt.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]uma vez | [s]sessão | [d]negar"
prompt_long: " Escolha [o/s/a/D]: "
prompt_short: " Escolha [o/s/D]: "
+ choose_smart_deny: " [o]uma vez | [d]negar"
+ prompt_smart_deny: " Escolha [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Tempo esgotado — comando negado"
allowed_once: " ✓ Permitido uma vez"
allowed_session: " ✓ Permitido nesta sessão"
diff --git a/locales/ru.yaml b/locales/ru.yaml
index 0450981fc2..c19536afba 100644
--- a/locales/ru.yaml
+++ b/locales/ru.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]один раз | [s]сеанс | [d]отклонить"
prompt_long: " Выбор [o/s/a/D]: "
prompt_short: " Выбор [o/s/D]: "
+ choose_smart_deny: " [o]один раз | [d]отклонить"
+ prompt_smart_deny: " Выбор [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Время ожидания истекло — команда отклонена"
allowed_once: " ✓ Разрешено один раз"
allowed_session: " ✓ Разрешено для этого сеанса"
diff --git a/locales/tr.yaml b/locales/tr.yaml
index 2fd70cd439..9a576624bc 100644
--- a/locales/tr.yaml
+++ b/locales/tr.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [b]ir kez | [o]turum | [r]eddet"
prompt_long: " Seçim [b/o/h/R]: "
prompt_short: " Seçim [b/o/R]: "
+ choose_smart_deny: " [b]ir kez | [r]eddet"
+ prompt_smart_deny: " Seçim [b/R]: "
+ smart_deny_once_inputs: "b,bir kez"
+ smart_deny_deny_inputs: "r,reddet"
timeout: " ⏱ Zaman aşımı — komut reddedildi"
allowed_once: " ✓ Bir kez izin verildi"
allowed_session: " ✓ Bu oturum için izin verildi"
diff --git a/locales/uk.yaml b/locales/uk.yaml
index 5e0391c0db..e78d47653d 100644
--- a/locales/uk.yaml
+++ b/locales/uk.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]один раз | [s]сеанс | [d]відхилити"
prompt_long: " Вибір [o/s/a/D]: "
prompt_short: " Вибір [o/s/D]: "
+ choose_smart_deny: " [o]один раз | [d]відхилити"
+ prompt_smart_deny: " Вибір [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ Час очікування вичерпано — команду відхилено"
allowed_once: " ✓ Дозволено один раз"
allowed_session: " ✓ Дозволено для цього сеансу"
diff --git a/locales/zh-hant.yaml b/locales/zh-hant.yaml
index 8f2d5e5980..6602231812 100644
--- a/locales/zh-hant.yaml
+++ b/locales/zh-hant.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]僅此一次 | [s]本次工作階段 | [d]拒絕"
prompt_long: " 選擇 [o/s/a/D]: "
prompt_short: " 選擇 [o/s/D]: "
+ choose_smart_deny: " [o]僅此一次 | [d]拒絕"
+ prompt_smart_deny: " 選擇 [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ 逾時 — 已拒絕指令"
allowed_once: " ✓ 本次允許"
allowed_session: " ✓ 本次工作階段內允許"
diff --git a/locales/zh.yaml b/locales/zh.yaml
index 7defb22e1e..2e307e4d60 100644
--- a/locales/zh.yaml
+++ b/locales/zh.yaml
@@ -7,6 +7,10 @@ approval:
choose_short: " [o]仅此一次 | [s]本次会话 | [d]拒绝"
prompt_long: " 选择 [o/s/a/D]: "
prompt_short: " 选择 [o/s/D]: "
+ choose_smart_deny: " [o]仅此一次 | [d]拒绝"
+ prompt_smart_deny: " 选择 [o/D]: "
+ smart_deny_once_inputs: "o,once"
+ smart_deny_deny_inputs: "d,deny"
timeout: " ⏱ 超时 — 已拒绝命令"
allowed_once: " ✓ 本次允许"
allowed_session: " ✓ 本次会话内允许"
diff --git a/plugins/platforms/discord/adapter.py b/plugins/platforms/discord/adapter.py
index d162f2afc4..b216ef29d8 100644
--- a/plugins/platforms/discord/adapter.py
+++ b/plugins/platforms/discord/adapter.py
@@ -5571,6 +5571,8 @@ class DiscordAdapter(BasePlatformAdapter):
self, chat_id: str, command: str, session_key: str,
description: str = "dangerous command",
metadata: Optional[dict] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""
Send a button-based exec approval prompt for a dangerous command.
@@ -5606,6 +5608,8 @@ class DiscordAdapter(BasePlatformAdapter):
"Do you want Hermes to run this command?\n\n"
"**Requested command:**\n```bash\n"
)
+ if smart_denied:
+ prompt_prefix += "**Smart DENY:** owner override applies to this one operation only.\n\n"
mention_content = self._approval_mention_content()
if mention_content:
prompt_prefix = f"{mention_content}\n{prompt_prefix}"
@@ -5642,6 +5646,8 @@ class DiscordAdapter(BasePlatformAdapter):
allowed_role_ids=self._allowed_role_ids,
require_admin=require_admin,
admin_user_ids=admin_user_ids,
+ allow_permanent=allow_permanent,
+ smart_denied=smart_denied,
)
send_kwargs: Dict[str, Any] = {"content": content, "embed": embed, "view": view}
@@ -6836,6 +6842,8 @@ def _define_discord_view_classes() -> None:
allowed_role_ids: Optional[set] = None,
require_admin: bool = False,
admin_user_ids: Optional[set] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
):
super().__init__(timeout=_read_discord_prompt_timeout())
self.session_key = session_key
@@ -6849,6 +6857,11 @@ def _define_discord_view_classes() -> None:
str(a).strip() for a in (admin_user_ids or set()) if str(a).strip()
}
self.resolved = False
+ if smart_denied:
+ self.remove_item(self.allow_session)
+ self.remove_item(self.allow_always)
+ elif not allow_permanent:
+ self.remove_item(self.allow_always)
def _check_auth(self, interaction: discord.Interaction) -> bool:
"""Verify the user clicking is authorized.
diff --git a/plugins/platforms/feishu/adapter.py b/plugins/platforms/feishu/adapter.py
index 9fecc91ddf..8f0b61685d 100644
--- a/plugins/platforms/feishu/adapter.py
+++ b/plugins/platforms/feishu/adapter.py
@@ -1980,6 +1980,8 @@ class FeishuAdapter(BasePlatformAdapter):
self, chat_id: str, command: str, session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send an interactive card with approval buttons.
@@ -2002,6 +2004,13 @@ class FeishuAdapter(BasePlatformAdapter):
"value": {"hermes_action": action_name, "approval_id": approval_id},
}
+ actions = [_btn("✅ Allow Once", "approve_once", "primary")]
+ if not smart_denied:
+ actions.append(_btn("✅ Session", "approve_session"))
+ if allow_permanent:
+ actions.append(_btn("✅ Always", "approve_always"))
+ actions.append(_btn("❌ Deny", "deny", "danger"))
+ scope_note = "\n\n**Smart DENY:** owner override applies to this one operation only." if smart_denied else ""
card = {
"config": {"wide_screen_mode": True},
"header": {
@@ -2011,16 +2020,11 @@ class FeishuAdapter(BasePlatformAdapter):
"elements": [
{
"tag": "markdown",
- "content": f"```\n{cmd_preview}\n```\n**Reason:** {description}",
+ "content": f"```\n{cmd_preview}\n```\n**Reason:** {description}{scope_note}",
},
{
"tag": "action",
- "actions": [
- _btn("✅ Allow Once", "approve_once", "primary"),
- _btn("✅ Session", "approve_session"),
- _btn("✅ Always", "approve_always"),
- _btn("❌ Deny", "deny", "danger"),
- ],
+ "actions": actions,
},
],
}
diff --git a/plugins/platforms/matrix/adapter.py b/plugins/platforms/matrix/adapter.py
index d66a1d87b7..39a8fd818f 100644
--- a/plugins/platforms/matrix/adapter.py
+++ b/plugins/platforms/matrix/adapter.py
@@ -2000,6 +2000,8 @@ class MatrixAdapter(BasePlatformAdapter):
session_key: str,
description: str = "dangerous command",
metadata: Optional[dict] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send a reaction-based exec approval prompt for Matrix."""
if not self._client:
@@ -2007,12 +2009,18 @@ class MatrixAdapter(BasePlatformAdapter):
requester_user_id = str((metadata or {}).get("requester_user_id") or "") or None
cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
+ scope_choices = ""
+ if smart_denied:
+ scope_choices = "Smart DENY: owner override applies to this one operation only.\n"
+ else:
+ scope_choices = "Reply `!approve session` to approve this pattern for the session, "
+ if allow_permanent:
+ scope_choices += "`!approve always` to approve permanently, "
text = (
"⚠️ **Dangerous command requires approval**\n"
f"```\n{cmd_preview}\n```\n"
f"Reason: {description}\n\n"
- "Reply `!approve` to execute, `!approve session` to approve this pattern for the session, "
- "`!approve always` to approve permanently, or `!deny` to cancel.\n\n"
+ f"{scope_choices}Reply `!approve` to execute once, or `!deny` to cancel.\n\n"
"You can also click the reaction to approve:\n"
"✅ = approve\n"
"❎ = deny"
@@ -2035,7 +2043,8 @@ class MatrixAdapter(BasePlatformAdapter):
self._approval_prompts_by_event[result.message_id] = prompt
self._approval_prompt_by_session[session_key] = result.message_id
- for emoji in ("✅", "♾️", "❌"):
+ reactions = ("✅", "❌") if smart_denied or not allow_permanent else ("✅", "♾️", "❌")
+ for emoji in reactions:
try:
reaction_result = await self._send_reaction(chat_id, result.message_id, emoji)
# Save the bot's reaction event_id for later cleanup
diff --git a/plugins/platforms/slack/adapter.py b/plugins/platforms/slack/adapter.py
index 05aeaf56f4..c0095d5add 100644
--- a/plugins/platforms/slack/adapter.py
+++ b/plugins/platforms/slack/adapter.py
@@ -3244,6 +3244,8 @@ class SlackAdapter(BasePlatformAdapter):
session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send a Block Kit approval prompt with interactive buttons.
@@ -3264,10 +3266,42 @@ class SlackAdapter(BasePlatformAdapter):
# instead of a flat truncation that overflows once the header +
# reason are added.
header = ":warning: *Command Approval Required*\n"
+ if smart_denied:
+ header += "*Smart DENY:* owner override applies to this one operation only.\n"
reason = f"Reason: {description[:500]}"
budget = 3000 - len(header) - len(reason) - len("``````\n") - len("...")
cmd_preview = command[:budget] + "..." if len(command) > budget else command
+ actions = [
+ {
+ "type": "button",
+ "text": {"type": "plain_text", "text": "Allow Once"},
+ "style": "primary",
+ "action_id": "hermes_approve_once",
+ "value": session_key,
+ },
+ ]
+ if not smart_denied:
+ actions.append({
+ "type": "button",
+ "text": {"type": "plain_text", "text": "Allow Session"},
+ "action_id": "hermes_approve_session",
+ "value": session_key,
+ })
+ if allow_permanent:
+ actions.append({
+ "type": "button",
+ "text": {"type": "plain_text", "text": "Always Allow"},
+ "action_id": "hermes_approve_always",
+ "value": session_key,
+ })
+ actions.append({
+ "type": "button",
+ "text": {"type": "plain_text", "text": "Deny"},
+ "style": "danger",
+ "action_id": "hermes_deny",
+ "value": session_key,
+ })
blocks = [
{
"type": "section",
@@ -3276,37 +3310,7 @@ class SlackAdapter(BasePlatformAdapter):
"text": f"{header}```{cmd_preview}```\n{reason}",
},
},
- {
- "type": "actions",
- "elements": [
- {
- "type": "button",
- "text": {"type": "plain_text", "text": "Allow Once"},
- "style": "primary",
- "action_id": "hermes_approve_once",
- "value": session_key,
- },
- {
- "type": "button",
- "text": {"type": "plain_text", "text": "Allow Session"},
- "action_id": "hermes_approve_session",
- "value": session_key,
- },
- {
- "type": "button",
- "text": {"type": "plain_text", "text": "Always Allow"},
- "action_id": "hermes_approve_always",
- "value": session_key,
- },
- {
- "type": "button",
- "text": {"type": "plain_text", "text": "Deny"},
- "style": "danger",
- "action_id": "hermes_deny",
- "value": session_key,
- },
- ],
- },
+ {"type": "actions", "elements": actions},
]
kwargs: Dict[str, Any] = {
diff --git a/plugins/platforms/teams/adapter.py b/plugins/platforms/teams/adapter.py
index 4323009837..1c02917f6e 100644
--- a/plugins/platforms/teams/adapter.py
+++ b/plugins/platforms/teams/adapter.py
@@ -1098,6 +1098,8 @@ class TeamsAdapter(BasePlatformAdapter):
session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send an Adaptive Card approval prompt with Allow/Deny buttons."""
if not self._app:
@@ -1111,39 +1113,34 @@ class TeamsAdapter(BasePlatformAdapter):
"desc": description,
}
- card = (
- AdaptiveCard()
- .with_version("1.4")
- .with_body([
- TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"),
- TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True),
- TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True),
- ])
- .with_actions([
- ExecuteAction(
- title="Allow Once",
- verb="hermes_approve",
- data={**btn_data_base, "hermes_action": "approve_once"},
- style="positive",
- ),
- ExecuteAction(
- title="Allow Session",
- verb="hermes_approve",
- data={**btn_data_base, "hermes_action": "approve_session"},
- ),
- ExecuteAction(
- title="Always Allow",
- verb="hermes_approve",
+ actions = [ExecuteAction(
+ title="Allow Once", verb="hermes_approve",
+ data={**btn_data_base, "hermes_action": "approve_once"}, style="positive",
+ )]
+ if not smart_denied:
+ actions.append(ExecuteAction(
+ title="Allow Session", verb="hermes_approve",
+ data={**btn_data_base, "hermes_action": "approve_session"},
+ ))
+ if allow_permanent:
+ actions.append(ExecuteAction(
+ title="Always Allow", verb="hermes_approve",
data={**btn_data_base, "hermes_action": "approve_always"},
- ),
- ExecuteAction(
- title="Deny",
- verb="hermes_approve",
- data={**btn_data_base, "hermes_action": "deny"},
- style="destructive",
- ),
- ])
- )
+ ))
+ actions.append(ExecuteAction(
+ title="Deny", verb="hermes_approve",
+ data={**btn_data_base, "hermes_action": "deny"}, style="destructive",
+ ))
+ body = [
+ TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"),
+ TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True),
+ TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True),
+ ]
+ if smart_denied:
+ body.append(TextBlock(
+ text="Smart DENY: owner override applies to this one operation only.", wrap=True
+ ))
+ card = AdaptiveCard().with_version("1.4").with_body(body).with_actions(actions)
try:
result = await self._send_card(chat_id, card)
diff --git a/plugins/platforms/telegram/adapter.py b/plugins/platforms/telegram/adapter.py
index 5ea820761c..de1dad27dd 100644
--- a/plugins/platforms/telegram/adapter.py
+++ b/plugins/platforms/telegram/adapter.py
@@ -4568,6 +4568,8 @@ class TelegramAdapter(BasePlatformAdapter):
self, chat_id: str, command: str, session_key: str,
description: str = "dangerous command",
metadata: Optional[Dict[str, Any]] = None,
+ allow_permanent: bool = True,
+ smart_denied: bool = False,
) -> SendResult:
"""Send an inline-keyboard approval prompt with interactive buttons.
@@ -4584,6 +4586,8 @@ class TelegramAdapter(BasePlatformAdapter):
f"{_html.escape(cmd_preview)}\n\n"
f"Reason: {_html.escape(description)}"
)
+ if smart_denied:
+ text += "\n\nSmart DENY: owner override applies to this one operation only."
# Resolve thread context for thread replies
thread_id = self._metadata_thread_id(metadata)
@@ -4596,16 +4600,19 @@ class TelegramAdapter(BasePlatformAdapter):
self._approval_counter = itertools.count(1)
approval_id = next(self._approval_counter)
- keyboard = InlineKeyboardMarkup([
- [
- InlineKeyboardButton("✅ Allow Once", callback_data=f"ea:once:{approval_id}"),
- InlineKeyboardButton("✅ Session", callback_data=f"ea:session:{approval_id}"),
- ],
- [
- InlineKeyboardButton("✅ Always", callback_data=f"ea:always:{approval_id}"),
- InlineKeyboardButton("❌ Deny", callback_data=f"ea:deny:{approval_id}"),
- ],
- ])
+ buttons = [
+ InlineKeyboardButton("✅ Allow Once", callback_data=f"ea:once:{approval_id}")
+ ]
+ if not smart_denied:
+ buttons.append(
+ InlineKeyboardButton("✅ Session", callback_data=f"ea:session:{approval_id}")
+ )
+ if allow_permanent:
+ buttons.append(
+ InlineKeyboardButton("✅ Always", callback_data=f"ea:always:{approval_id}")
+ )
+ buttons.append(InlineKeyboardButton("❌ Deny", callback_data=f"ea:deny:{approval_id}"))
+ keyboard = InlineKeyboardMarkup([buttons])
kwargs: Dict[str, Any] = {
"chat_id": normalize_telegram_chat_id(chat_id),
diff --git a/tests/acp/test_permissions.py b/tests/acp/test_permissions.py
index a7248aa717..123592d505 100644
--- a/tests/acp/test_permissions.py
+++ b/tests/acp/test_permissions.py
@@ -23,6 +23,7 @@ def _invoke_callback(
outcome,
*,
allow_permanent=True,
+ smart_denied=False,
timeout=60.0,
use_prompt_path=False,
):
@@ -45,6 +46,7 @@ def _invoke_callback(
"rm -rf /",
"dangerous command",
allow_permanent=allow_permanent,
+ smart_denied=smart_denied,
approval_callback=cb,
)
else:
@@ -52,6 +54,7 @@ def _invoke_callback(
"rm -rf /",
"dangerous command",
allow_permanent=allow_permanent,
+ smart_denied=smart_denied,
)
scheduled["coro"].close()
@@ -115,6 +118,30 @@ class TestApprovalBridge:
assert result == "session"
assert option_ids == ["allow_once", "allow_session", "deny", "deny_always"]
+ def test_smart_deny_prompt_only_offers_once_and_deny(self):
+ result, kwargs, _, _, _ = _invoke_callback(
+ AllowedOutcome(option_id="allow_once", outcome="selected"),
+ allow_permanent=False,
+ smart_denied=True,
+ use_prompt_path=True,
+ )
+
+ assert result == "once"
+ assert [option.option_id for option in kwargs["options"]] == [
+ "allow_once", "deny",
+ ]
+
+ def test_smart_deny_rejects_disallowed_session_outcome(self):
+ result, kwargs, _, _, _ = _invoke_callback(
+ AllowedOutcome(option_id="allow_session", outcome="selected"),
+ smart_denied=True,
+ )
+
+ assert result == "deny"
+ assert [option.option_id for option in kwargs["options"]] == [
+ "allow_once", "deny",
+ ]
+
def test_reject_always_outcome_denies_without_changing_policy(self):
result, kwargs, _, _, _ = _invoke_callback(
AllowedOutcome(option_id="deny_always", outcome="selected"),
diff --git a/tests/cli/test_cli_approval_ui.py b/tests/cli/test_cli_approval_ui.py
index 8e1d4e327d..ebca5bd85e 100644
--- a/tests/cli/test_cli_approval_ui.py
+++ b/tests/cli/test_cli_approval_ui.py
@@ -67,6 +67,38 @@ def _make_background_cli_stub():
class TestCliApprovalUi:
+ def test_smart_denied_callback_offers_only_once_and_deny(self):
+ cli = _make_cli_stub()
+ result = {}
+
+ def _run_callback():
+ result["value"] = cli._approval_callback(
+ "rm -rf /tmp/example",
+ "recursive delete",
+ allow_permanent=False,
+ smart_denied=True,
+ )
+
+ thread = threading.Thread(target=_run_callback, daemon=True)
+ thread.start()
+
+ deadline = time.time() + 2
+ while cli._approval_state is None and time.time() < deadline:
+ time.sleep(0.01)
+
+ assert cli._approval_state is not None
+ assert cli._approval_state["choices"] == ["once", "deny"]
+
+ cli._approval_state["response_queue"].put("deny")
+ thread.join(timeout=2)
+ assert result["value"] == "deny"
+
+ def test_non_smart_non_permanent_callback_preserves_session_choice(self):
+ cli = _make_cli_stub()
+ assert cli._approval_choices(
+ "rm -rf /tmp/example", allow_permanent=False, smart_denied=False
+ ) == ["once", "session", "deny"]
+
def test_sudo_prompt_restores_existing_draft_after_response(self):
cli = _make_cli_stub()
cli._app.current_buffer = _FakeBuffer("draft command", cursor_position=5)
diff --git a/tests/gateway/test_api_server_runs.py b/tests/gateway/test_api_server_runs.py
index 9e3180acce..147a75ab4b 100644
--- a/tests/gateway/test_api_server_runs.py
+++ b/tests/gateway/test_api_server_runs.py
@@ -20,6 +20,7 @@ from aiohttp.test_utils import TestClient, TestServer
from gateway.config import PlatformConfig
from gateway.platforms.api_server import (
APIServerAdapter,
+ _approval_event_choices,
cors_middleware,
security_headers_middleware,
)
@@ -31,6 +32,24 @@ from tools import approval as approval_mod
# ---------------------------------------------------------------------------
+@pytest.mark.parametrize(
+ ("smart_denied", "allow_permanent", "expected"),
+ [
+ (False, True, ["once", "session", "always", "deny"]),
+ (False, False, ["once", "session", "deny"]),
+ (True, True, ["once", "deny"]),
+ (True, False, ["once", "deny"]),
+ ],
+)
+def test_approval_event_choices_follow_backend_capabilities(
+ smart_denied, allow_permanent, expected
+):
+ assert _approval_event_choices(
+ smart_denied=smart_denied,
+ allow_permanent=allow_permanent,
+ ) == expected
+
+
def _make_adapter(api_key: str = "") -> APIServerAdapter:
"""Create an adapter with optional API key."""
extra = {}
diff --git a/tests/gateway/test_approval_prompt_redaction.py b/tests/gateway/test_approval_prompt_redaction.py
index fb57a8644a..7aa9c824c8 100644
--- a/tests/gateway/test_approval_prompt_redaction.py
+++ b/tests/gateway/test_approval_prompt_redaction.py
@@ -126,3 +126,63 @@ class TestApprovalCommandWiring:
from gateway.platforms import api_server
self._assert_redacts_then_uses(api_server, "_approval_notify", "put_nowait")
+
+ def test_chat_platform_threads_approval_capabilities_to_adapter(self):
+ """The gateway must not drop the backend's one-operation UI contract."""
+ import ast
+ import inspect
+ import gateway.run as run
+
+ tree = ast.parse(inspect.getsource(run))
+ notify = next(
+ node for node in ast.walk(tree)
+ if isinstance(node, ast.FunctionDef) and node.name == "_approval_notify_sync"
+ )
+ call = next(
+ node for node in ast.walk(notify)
+ if isinstance(node, ast.Call)
+ and isinstance(node.func, ast.Attribute)
+ and node.func.attr == "send_exec_approval"
+ )
+ keywords = {kw.arg: kw.value for kw in call.keywords}
+ for name, default in (("allow_permanent", True), ("smart_denied", False)):
+ value = keywords[name]
+ assert isinstance(value, ast.Call)
+ assert isinstance(value.func, ast.Attribute) and value.func.attr == "get"
+ assert isinstance(value.args[0], ast.Constant) and value.args[0].value == name
+ assert isinstance(value.args[1], ast.Constant) and value.args[1].value is default
+
+
+class TestApprovalTextFallbackContract:
+ def test_smart_deny_only_advertises_one_operation(self):
+ from gateway.run import _format_exec_approval_fallback
+
+ text = _format_exec_approval_fallback(
+ "rm -rf /", "dangerous deletion", "/",
+ allow_permanent=False, smart_denied=True,
+ )
+ assert "owner override" in text.lower()
+ assert "one operation" in text.lower()
+ assert "`/approve`" in text
+ assert "approve session" not in text
+ assert "approve always" not in text
+
+ def test_non_smart_restriction_preserves_session_choice(self):
+ from gateway.run import _format_exec_approval_fallback
+
+ text = _format_exec_approval_fallback(
+ "curl https://example.test", "content warning", "!",
+ allow_permanent=False, smart_denied=False,
+ )
+ assert "`!approve session`" in text
+ assert "approve always" not in text
+
+ def test_manual_prompt_preserves_all_choices(self):
+ from gateway.run import _format_exec_approval_fallback
+
+ text = _format_exec_approval_fallback(
+ "rm -rf /", "dangerous deletion", "/",
+ allow_permanent=True, smart_denied=False,
+ )
+ assert "`/approve session`" in text
+ assert "`/approve always`" in text
diff --git a/tests/gateway/test_slack_approval_buttons.py b/tests/gateway/test_slack_approval_buttons.py
index b85fc37872..2f0f683580 100644
--- a/tests/gateway/test_slack_approval_buttons.py
+++ b/tests/gateway/test_slack_approval_buttons.py
@@ -121,6 +121,24 @@ class TestSlackExecApproval:
for e in elements:
assert e["value"] == "agent:main:slack:group:C1:1111"
+ @pytest.mark.asyncio
+ async def test_smart_deny_owner_override_hides_persistent_buttons(self):
+ adapter = _make_adapter()
+ mock_client = adapter._team_clients["T1"]
+ mock_client.chat_postMessage = AsyncMock(return_value={"ts": "1234.5678"})
+
+ await adapter.send_exec_approval(
+ chat_id="C1", command="rm -rf /", session_key="s",
+ allow_permanent=False, smart_denied=True,
+ )
+
+ kwargs = mock_client.chat_postMessage.call_args.kwargs
+ elements = kwargs["blocks"][1]["elements"]
+ assert [element["action_id"] for element in elements] == [
+ "hermes_approve_once", "hermes_deny",
+ ]
+ assert "one operation" in kwargs["blocks"][0]["text"]["text"].lower()
+
@pytest.mark.asyncio
async def test_sends_in_thread(self):
adapter = _make_adapter()
diff --git a/tests/gateway/test_telegram_approval_buttons.py b/tests/gateway/test_telegram_approval_buttons.py
index 96de984a9c..1698a7010d 100644
--- a/tests/gateway/test_telegram_approval_buttons.py
+++ b/tests/gateway/test_telegram_approval_buttons.py
@@ -105,6 +105,49 @@ class TestTelegramExecApproval:
assert "dangerous deletion" in kwargs["text"]
assert kwargs["reply_markup"] is not None # InlineKeyboardMarkup
+ @pytest.mark.asyncio
+ async def test_smart_deny_owner_override_only_offers_once_and_deny(self, monkeypatch):
+ adapter = _make_adapter()
+ adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=42))
+ buttons = []
+ monkeypatch.setattr(
+ "plugins.platforms.telegram.adapter.InlineKeyboardButton",
+ lambda text, callback_data: buttons.append((text, callback_data)) or (text, callback_data),
+ )
+ monkeypatch.setattr(
+ "plugins.platforms.telegram.adapter.InlineKeyboardMarkup", lambda rows: rows
+ )
+
+ await adapter.send_exec_approval(
+ chat_id="12345", command="rm -rf /", session_key="s",
+ allow_permanent=False, smart_denied=True,
+ )
+
+ labels = [label for label, _ in buttons]
+ assert labels == ["✅ Allow Once", "❌ Deny"]
+ text = adapter._bot.send_message.call_args.kwargs["text"]
+ assert "one operation" in text.lower()
+
+ @pytest.mark.asyncio
+ async def test_non_smart_allow_permanent_false_keeps_session(self, monkeypatch):
+ adapter = _make_adapter()
+ adapter._bot.send_message = AsyncMock(return_value=SimpleNamespace(message_id=42))
+ buttons = []
+ monkeypatch.setattr(
+ "plugins.platforms.telegram.adapter.InlineKeyboardButton",
+ lambda text, callback_data: buttons.append(text) or text,
+ )
+ monkeypatch.setattr(
+ "plugins.platforms.telegram.adapter.InlineKeyboardMarkup", lambda rows: rows
+ )
+
+ await adapter.send_exec_approval(
+ chat_id="12345", command="curl example.test", session_key="s",
+ allow_permanent=False,
+ )
+
+ assert buttons == ["✅ Allow Once", "✅ Session", "❌ Deny"]
+
@pytest.mark.asyncio
async def test_stores_approval_state(self):
adapter = _make_adapter()
diff --git a/tests/gateway/test_tui_approval_redaction.py b/tests/gateway/test_tui_approval_redaction.py
index 04716222e7..bb757bc9b0 100644
--- a/tests/gateway/test_tui_approval_redaction.py
+++ b/tests/gateway/test_tui_approval_redaction.py
@@ -47,6 +47,28 @@ class TestTuiApprovalEmitRedaction:
tui_server._emit_approval_request("s", None)
assert emitted["payload"] == {}
+ @pytest.mark.parametrize(
+ ("data", "expected"),
+ [
+ ({"smart_denied": True, "allow_permanent": True}, ["once", "deny"]),
+ ({"allow_permanent": False}, ["once", "session", "deny"]),
+ ({"allow_permanent": True}, ["once", "session", "always", "deny"]),
+ ],
+ )
+ def test_emit_approval_request_derives_choices(self, monkeypatch, data, expected):
+ from tui_gateway import server as tui_server
+
+ emitted = {}
+ monkeypatch.setattr(
+ tui_server,
+ "_emit",
+ lambda event, sid, payload=None: emitted.update({"payload": payload}),
+ )
+
+ tui_server._emit_approval_request("s", data)
+
+ assert emitted["payload"]["choices"] == expected
+
def test_no_raw_command_emit_in_approval_registrations(self):
"""Every register_gateway_notify approval callback must route through the
redacting `_emit_approval_request` helper — no registration may emit the
diff --git a/tests/tools/test_approval.py b/tests/tools/test_approval.py
index 253c338204..ed67099bbf 100644
--- a/tests/tools/test_approval.py
+++ b/tests/tools/test_approval.py
@@ -9,6 +9,8 @@ from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch as mock_patch
+import pytest
+
import tools.approval as approval_module
from hermes_constants import get_hermes_home
from tools.approval import (
@@ -1140,6 +1142,110 @@ class TestFullCommandAlwaysShown:
assert result == "deny"
+class TestSmartDeniedPrompt:
+ def test_callback_receives_smart_denied_capability(self):
+ captured = {}
+
+ def callback(command, description, **kwargs):
+ captured.update(kwargs)
+ return "deny"
+
+ result = prompt_dangerous_approval(
+ "rm -rf /tmp/example",
+ "recursive delete",
+ allow_permanent=False,
+ smart_denied=True,
+ approval_callback=callback,
+ )
+
+ assert result == "deny"
+ assert captured == {"allow_permanent": False, "smart_denied": True}
+
+ def test_short_prompt_smart_deny_rejects_session_input(self):
+ with mock_patch("builtins.input", return_value="session"):
+ result = prompt_dangerous_approval(
+ "rm -rf /tmp/example",
+ "recursive delete",
+ allow_permanent=False,
+ smart_denied=True,
+ )
+
+ assert result == "deny"
+
+ def test_short_prompt_smart_deny_displays_only_once_and_deny(self, capsys):
+ prompts = []
+
+ def input_once(prompt):
+ prompts.append(prompt)
+ return "deny"
+
+ with mock_patch("builtins.input", side_effect=input_once):
+ prompt_dangerous_approval(
+ "rm -rf /tmp/example",
+ "recursive delete",
+ allow_permanent=False,
+ smart_denied=True,
+ )
+
+ rendered = capsys.readouterr().out
+ assert "[o]nce" in rendered and "[d]eny" in rendered
+ assert "[s]ession" not in rendered and "[a]lways" not in rendered
+ assert prompts == [" Choice [o/D]: "]
+
+ @pytest.mark.parametrize(
+ ("lang", "once_key", "deny_key", "once_label", "deny_label"),
+ [
+ ("tr", "b", "r", "[b]ir kez", "[r]eddet"),
+ ("fr", "o", "r", "[o]ne fois", "[r]efuser"),
+ ("ja", "o", "d", "[o]今回のみ", "[d]拒否"),
+ ],
+ )
+ def test_smart_deny_uses_locale_specific_once_deny_choices(
+ self, monkeypatch, capsys, lang, once_key, deny_key, once_label, deny_label,
+ ):
+ monkeypatch.setenv("HERMES_LANGUAGE", lang)
+ from agent import i18n
+ i18n.reset_language_cache()
+ prompts = []
+
+ def choose_once(prompt):
+ prompts.append(prompt)
+ return once_key
+
+ try:
+ with mock_patch("builtins.input", side_effect=choose_once):
+ result = prompt_dangerous_approval(
+ "rm -rf /tmp/example", "recursive delete",
+ allow_permanent=False, smart_denied=True,
+ )
+ finally:
+ i18n.reset_language_cache()
+
+ rendered = capsys.readouterr().out
+ assert result == "once"
+ assert once_label in rendered
+ assert deny_label in rendered
+ assert i18n.t("approval.choose_short", lang=lang).split("|")[1].strip() not in rendered
+ assert "/".join((once_key, deny_key.upper())) in prompts[0]
+
+ @pytest.mark.parametrize(("lang", "forbidden"), [("tr", "o"), ("fr", "s"), ("ja", "a")])
+ def test_smart_deny_rejects_localized_session_or_always_shortcuts(
+ self, monkeypatch, lang, forbidden,
+ ):
+ monkeypatch.setenv("HERMES_LANGUAGE", lang)
+ from agent import i18n
+ i18n.reset_language_cache()
+ try:
+ with mock_patch("builtins.input", return_value=forbidden):
+ result = prompt_dangerous_approval(
+ "rm -rf /tmp/example", "recursive delete",
+ allow_permanent=False, smart_denied=True,
+ )
+ finally:
+ i18n.reset_language_cache()
+ assert result == "deny"
+
+
class TestForkBombDetection:
"""The fork bomb regex must match the classic :(){ :|:& };: pattern."""
diff --git a/tests/tools/test_execute_code_approval_cluster.py b/tests/tools/test_execute_code_approval_cluster.py
index b51a1c130c..7ea74a53b4 100644
--- a/tests/tools/test_execute_code_approval_cluster.py
+++ b/tests/tools/test_execute_code_approval_cluster.py
@@ -17,6 +17,7 @@ from __future__ import annotations
import concurrent.futures
import contextvars
+import json
import threading
import pytest
@@ -144,6 +145,23 @@ def _register_resolver(session_key: str, result):
A._gateway_notify_cbs[session_key] = cb
+def _register_capturing_resolver(session_key: str, result):
+ """Resolve immediately and retain the exact approval payload shown."""
+ seen = {}
+
+ def cb(approval_data):
+ seen["approval_data"] = approval_data
+ with A._lock:
+ entries = A._gateway_queues.get(session_key, [])
+ if entries:
+ entries[-1].result = result
+ entries[-1].event.set()
+
+ with A._lock:
+ A._gateway_notify_cbs[session_key] = cb
+ return seen
+
+
def test_guard_isolated_backend_approved():
# Container backends already sandbox the child — no-op approve.
assert A.check_execute_code_guard("import os", "docker")["approved"] is True
@@ -160,6 +178,7 @@ def test_guard_headless_local_approved(monkeypatch):
def test_guard_cron_deny_blocks(monkeypatch):
+ monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
monkeypatch.setenv("HERMES_CRON_SESSION", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.setattr(A, "_get_approval_mode", lambda: "manual")
@@ -267,9 +286,11 @@ def test_guard_smart_mode(gw_session, monkeypatch):
res = A.check_execute_code_guard("import os", "local")
assert res["approved"] is True and res.get("smart_approved") is True
+ # Smart DENY on an interactive surface now asks the owner. With no bound
+ # notifier it remains pending rather than being hard-denied.
monkeypatch.setattr(A, "_smart_approve", lambda c, d: "deny")
res = A.check_execute_code_guard("import os", "local")
- assert res["approved"] is False and res.get("smart_denied") is True
+ assert res["approved"] is False and res["status"] == "pending_approval"
# escalate → falls through to manual gateway approval
monkeypatch.setattr(A, "_smart_approve", lambda c, d: "escalate")
@@ -278,6 +299,150 @@ def test_guard_smart_mode(gw_session, monkeypatch):
assert res["approved"] is True
+def test_terminal_smart_deny_owner_override_is_one_operation(gw_session, monkeypatch):
+ """A human may override DENY, but a broad UI choice must not be persisted."""
+ with A._lock:
+ A._permanent_approved.discard("owner-override-test-danger")
+ A._session_approved.get(gw_session, set()).discard("owner-override-test-danger")
+ monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+ monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
+ monkeypatch.setattr(
+ A,
+ "detect_dangerous_command",
+ lambda command: (True, "owner-override-test-danger", f"risk:{command}"),
+ )
+ monkeypatch.setattr(
+ "tools.tirith_security.check_command_security",
+ lambda _command: {"action": "allow", "findings": [], "summary": ""},
+ raising=False,
+ )
+
+ shown = _register_capturing_resolver(gw_session, "always")
+ result = A.check_all_command_guards("dangerous /tmp/first", "local")
+
+ assert result["approved"] is True
+ assert result["user_approved"] is True
+ assert shown["approval_data"]["smart_denied"] is True
+ assert shown["approval_data"]["allow_permanent"] is False
+ assert A.is_approved(gw_session, "owner-override-test-danger") is False
+
+ _register_resolver(gw_session, "deny")
+ changed = A.check_all_command_guards("dangerous /tmp/second", "local")
+ assert changed["approved"] is False
+ assert changed["outcome"] == "denied"
+
+
+def test_execute_code_smart_deny_owner_override_is_one_operation(gw_session, monkeypatch):
+ """Never persist the coarse execute_code key after overriding smart DENY."""
+ with A._lock:
+ A._permanent_approved.discard("execute_code")
+ A._session_approved.get(gw_session, set()).discard("execute_code")
+ monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+ monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
+
+ shown = _register_capturing_resolver(gw_session, "session")
+ result = A.check_execute_code_guard("print('first')", "local")
+
+ assert result["approved"] is True
+ assert result["user_approved"] is True
+ assert shown["approval_data"]["smart_denied"] is True
+ assert shown["approval_data"]["allow_permanent"] is False
+ assert A.is_approved(gw_session, "execute_code") is False
+
+ _register_resolver(gw_session, "deny")
+ changed = A.check_execute_code_guard("print('second')", "local")
+ assert changed["approved"] is False
+ assert changed["outcome"] == "denied"
+
+
+def test_smart_escalate_still_persists_session_choice(gw_session, monkeypatch):
+ """The DENY restriction must not alter Smart ESCALATE's manual choices."""
+ key = "smart-escalate-persistence"
+ with A._lock:
+ A._session_approved.get(gw_session, set()).discard(key)
+ monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+ monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "escalate")
+ monkeypatch.setattr(
+ A, "detect_dangerous_command",
+ lambda command: (True, key, f"risk:{command}"),
+ )
+ monkeypatch.setattr(
+ "tools.tirith_security.check_command_security",
+ lambda _command: {"action": "allow", "findings": [], "summary": ""},
+ raising=False,
+ )
+
+ shown = _register_capturing_resolver(gw_session, "session")
+ result = A.check_all_command_guards("dangerous escalate", "local")
+
+ assert result["approved"] is True
+ assert shown["approval_data"]["allow_permanent"] is True
+ assert "smart_denied" not in shown["approval_data"]
+ assert A.is_approved(gw_session, key) is True
+
+
+def test_terminal_smart_deny_pending_payload_is_one_operation(gw_session, monkeypatch):
+ monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+ monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
+ monkeypatch.setattr(
+ A, "detect_dangerous_command",
+ lambda command: (True, "pending-smart-deny", f"risk:{command}"),
+ )
+ monkeypatch.setattr(
+ "tools.tirith_security.check_command_security",
+ lambda _command: {"action": "allow", "findings": [], "summary": ""},
+ raising=False,
+ )
+
+ result = A.check_all_command_guards("dangerous pending", "local")
+
+ assert result["status"] == "pending_approval"
+ assert result["smart_denied"] is True
+ assert result["allow_permanent"] is False
+ with A._lock:
+ pending = dict(A._pending[gw_session])
+ assert pending["smart_denied"] is True
+ assert pending["allow_permanent"] is False
+
+
+def test_execute_code_smart_deny_pending_payload_is_one_operation(gw_session, monkeypatch):
+ monkeypatch.setattr(A, "_get_approval_mode", lambda: "smart")
+ monkeypatch.setattr(A, "_smart_approve", lambda _command, _description: "deny")
+
+ result = A.check_execute_code_guard("print('pending')", "local")
+
+ assert result["status"] == "pending_approval"
+ assert result["smart_denied"] is True
+ assert result["allow_permanent"] is False
+ with A._lock:
+ pending = dict(A._pending[gw_session])
+ assert pending["smart_denied"] is True
+ assert pending["allow_permanent"] is False
+
+
+def test_terminal_serializes_smart_deny_pending_capabilities(monkeypatch):
+ from tools import terminal_tool as terminal_module
+
+ monkeypatch.setattr(
+ terminal_module,
+ "_check_all_guards",
+ lambda *_args, **_kwargs: {
+ "approved": False,
+ "status": "pending_approval",
+ "command": "rm -rf /tmp/example",
+ "description": "recursive delete",
+ "pattern_key": "rm-rf",
+ "smart_denied": True,
+ "allow_permanent": False,
+ },
+ )
+
+ payload = json.loads(terminal_module.terminal_tool(command="rm -rf /tmp/example"))
+
+ assert payload["smart_denied"] is True
+ assert payload["allow_permanent"] is False
+
+
def test_guard_session_yolo_bypasses(gw_session):
A.enable_session_yolo(gw_session)
try:
@@ -345,6 +510,7 @@ def test_execute_code_entry_blocks_before_spawn_when_guard_denies(monkeypatch, t
from tools import terminal_tool as TT
marker = tmp_path / "child-ran.marker"
+ monkeypatch.setattr(A, "_YOLO_MODE_FROZEN", False)
monkeypatch.setenv("HERMES_CRON_SESSION", "1")
monkeypatch.delenv("HERMES_GATEWAY_SESSION", raising=False)
monkeypatch.delenv("HERMES_INTERACTIVE", raising=False)
diff --git a/tools/approval.py b/tools/approval.py
index faffb9a219..7c58062ed5 100644
--- a/tools/approval.py
+++ b/tools/approval.py
@@ -1735,16 +1735,21 @@ def save_permanent_allowlist(patterns: set):
def prompt_dangerous_approval(command: str, description: str,
timeout_seconds: int | None = None,
allow_permanent: bool = True,
- approval_callback=None) -> str:
+ approval_callback=None,
+ *, smart_denied: bool = False) -> str:
"""Prompt the user to approve a dangerous command (CLI only).
Args:
allow_permanent: When False, hide the [a]lways option (used when
tirith warnings are present, since broad permanent allowlisting
is inappropriate for content-level security findings).
+ smart_denied: When True, this is an owner override of a Smart DENY.
+ Offer only one-operation approval or denial.
approval_callback: Optional callback registered by the CLI for
prompt_toolkit integration. Signature:
- (command, description, *, allow_permanent=True) -> str.
+ (command, description, *, allow_permanent=True,
+ smart_denied=False) -> str. Legacy callback signatures remain
+ supported when ``smart_denied`` is false.
Returns: 'once', 'session', 'always', or 'deny'
"""
@@ -1761,8 +1766,12 @@ def prompt_dangerous_approval(command: str, description: str,
if approval_callback is not None:
try:
- return approval_callback(display_command, display_description,
- allow_permanent=allow_permanent)
+ callback_kwargs = {"allow_permanent": allow_permanent}
+ if smart_denied:
+ callback_kwargs["smart_denied"] = True
+ return approval_callback(
+ display_command, display_description, **callback_kwargs
+ )
except Exception as e:
logger.error("Approval callback failed: %s", e, exc_info=True)
return "deny"
@@ -1804,7 +1813,9 @@ def prompt_dangerous_approval(command: str, description: str,
print(f" {t('approval.dangerous_header', description=display_description)}")
print(f" {display_command}")
print()
- if allow_permanent:
+ if smart_denied:
+ print(t("approval.choose_smart_deny"))
+ elif allow_permanent:
print(t("approval.choose_long"))
else:
print(t("approval.choose_short"))
@@ -1815,7 +1826,10 @@ def prompt_dangerous_approval(command: str, description: str,
def get_input():
try:
- prompt = t("approval.prompt_long") if allow_permanent else t("approval.prompt_short")
+ if smart_denied:
+ prompt = t("approval.prompt_smart_deny")
+ else:
+ prompt = t("approval.prompt_long") if allow_permanent else t("approval.prompt_short")
result["choice"] = input(prompt).strip().lower()
except (EOFError, OSError):
result["choice"] = ""
@@ -1829,6 +1843,21 @@ def prompt_dangerous_approval(command: str, description: str,
return "deny"
choice = result["choice"]
+ if smart_denied:
+ choice_map = {
+ **{
+ value: "once"
+ for value in t("approval.smart_deny_once_inputs").split(",")
+ },
+ **{
+ value: "deny"
+ for value in t("approval.smart_deny_deny_inputs").split(",")
+ },
+ }
+ decision = choice_map.get(choice, "deny")
+ print(t("approval.allowed_once" if decision == "once" else "approval.denied"))
+ return decision
+
if choice in {'o', 'once'}:
print(t("approval.allowed_once"))
return "once"
@@ -2810,6 +2839,7 @@ def check_all_command_guards(command: str, env_type: str,
# When approvals.mode=smart, ask the aux LLM before prompting the user.
# Inspired by OpenAI Codex's Smart Approvals guardian subagent
# (openai/codex#13860).
+ smart_denied_for_owner = False
if approval_mode == "smart":
combined_desc_for_llm = "; ".join(desc for _, desc, _ in warnings)
observer_payload = _prepare_smart_approval_observer(
@@ -2830,15 +2860,17 @@ def check_all_command_guards(command: str, env_type: str,
return {"approved": True, "message": None,
"smart_approved": True,
"description": combined_desc_for_llm}
- elif verdict == "deny":
- combined_desc_for_llm = "; ".join(desc for _, desc, _ in warnings)
+ elif verdict == "deny" and not (is_cli or is_gateway or is_ask):
return {
"approved": False,
"message": f"BLOCKED by smart approval: {combined_desc_for_llm}. "
"The command was assessed as genuinely dangerous. Do NOT retry.",
"smart_denied": True,
}
- # verdict == "escalate" → fall through to manual prompt
+ elif verdict == "deny":
+ smart_denied_for_owner = True
+ # An interactive owner may override DENY for this operation only.
+ # ESCALATE follows the normal, potentially persistent manual behavior.
# --- Phase 3: Approval ---
@@ -2875,10 +2907,12 @@ def check_all_command_guards(command: str, env_type: str,
"pattern_key": primary_key,
"pattern_keys": all_keys,
"description": redact_sensitive_text(combined_desc),
- # Mirror the CLI's allow_permanent gate: a tirith warning downgrades
- # "always" to session scope below, so the UI must not offer it.
- "allow_permanent": not has_tirith,
+ # Smart DENY overrides are one-operation decisions, so the UI
+ # must not offer a permanent scope.
+ "allow_permanent": not has_tirith and not smart_denied_for_owner,
}
+ if smart_denied_for_owner:
+ approval_data["smart_denied"] = True
decision = _await_gateway_decision(
session_key, notify_cb, approval_data, surface="gateway"
)
@@ -2931,16 +2965,17 @@ def check_all_command_guards(command: str, env_type: str,
"deny_reason": deny_reason,
}
- # User approved — persist based on scope (same logic as CLI)
- for key, _, is_tirith in warnings:
- if choice == "session" or (choice == "always" and is_tirith):
- approve_session(session_key, key)
- elif choice == "always":
- approve_session(session_key, key)
- approve_permanent(key)
- save_permanent_allowlist(_permanent_approved)
- # choice == "once": no persistence — command allowed this
- # single time only, matching the CLI's behavior.
+ # A smart-DENY owner override is always one operation, even if an
+ # older client returns "session" or "always". Manual and ESCALATE
+ # choices retain their existing persistence semantics.
+ if not smart_denied_for_owner:
+ for key, _, is_tirith in warnings:
+ if choice == "session" or (choice == "always" and is_tirith):
+ approve_session(session_key, key)
+ elif choice == "always":
+ approve_session(session_key, key)
+ approve_permanent(key)
+ save_permanent_allowlist(_permanent_approved)
return {"approved": True, "message": None,
"user_approved": True, "description": combined_desc}
@@ -2952,13 +2987,16 @@ def check_all_command_guards(command: str, env_type: str,
from agent.redact import redact_sensitive_text
_disp_command = redact_sensitive_text(command)
_disp_combined_desc = redact_sensitive_text(combined_desc)
- submit_pending(session_key, {
+ pending_data = {
"command": _disp_command,
"pattern_key": primary_key,
"pattern_keys": all_keys,
"description": _disp_combined_desc,
- })
- return {
+ }
+ if smart_denied_for_owner:
+ pending_data.update(smart_denied=True, allow_permanent=False)
+ submit_pending(session_key, pending_data)
+ result = {
"approved": False,
"pattern_key": primary_key,
"status": "pending_approval",
@@ -2969,6 +3007,9 @@ def check_all_command_guards(command: str, env_type: str,
f"⚠️ {_disp_combined_desc}. Asking the user for approval.\n\n**Command:**\n```\n{_disp_command}\n```"
),
}
+ if smart_denied_for_owner:
+ result.update(smart_denied=True, allow_permanent=False)
+ return result
# CLI interactive: single combined prompt
# Hide [a]lways when any tirith warning is present
@@ -2981,9 +3022,13 @@ def check_all_command_guards(command: str, env_type: str,
session_key=session_key,
surface="cli",
)
- choice = prompt_dangerous_approval(command, combined_desc,
- allow_permanent=not has_tirith,
- approval_callback=approval_callback)
+ choice = prompt_dangerous_approval(
+ command,
+ combined_desc,
+ allow_permanent=not has_tirith and not smart_denied_for_owner,
+ smart_denied=smart_denied_for_owner,
+ approval_callback=approval_callback,
+ )
_fire_approval_hook(
"post_approval_response",
command=command,
@@ -3012,16 +3057,18 @@ def check_all_command_guards(command: str, env_type: str,
"user_consent": False,
}
- # Persist approval for each warning individually
- for key, _, is_tirith in warnings:
- if choice == "session" or (choice == "always" and is_tirith):
- # tirith: session only (no permanent broad allowlisting)
- approve_session(session_key, key)
- elif choice == "always":
- # dangerous patterns: permanent allowed
- approve_session(session_key, key)
- approve_permanent(key)
- save_permanent_allowlist(_permanent_approved)
+ # Smart-DENY owner overrides are one-operation scoped. Preserve existing
+ # persistence for manual mode and smart ESCALATE.
+ if not smart_denied_for_owner:
+ for key, _, is_tirith in warnings:
+ if choice == "session" or (choice == "always" and is_tirith):
+ # tirith: session only (no permanent broad allowlisting)
+ approve_session(session_key, key)
+ elif choice == "always":
+ # dangerous patterns: permanent allowed
+ approve_session(session_key, key)
+ approve_permanent(key)
+ save_permanent_allowlist(_permanent_approved)
return {"approved": True, "message": None,
"user_approved": True, "description": combined_desc}
@@ -3112,6 +3159,7 @@ def check_execute_code_guard(code: str, env_type: str,
# Smart mode: ask the aux LLM about the whole script. An APPROVE here only
# suppresses the redundant whole-script prompt; the per-call terminal()
# guards (restored by context propagation) still run independently.
+ smart_denied_for_owner = False
if approval_mode == "smart":
observer_payload = _prepare_smart_approval_observer(
command=command,
@@ -3127,7 +3175,7 @@ def check_execute_code_guard(code: str, env_type: str,
session_key)
return {"approved": True, "message": None,
"smart_approved": True, "description": description}
- if verdict == "deny":
+ if verdict == "deny" and not (is_gateway or is_ask):
return {
"approved": False,
"message": ("BLOCKED by smart approval: execute_code script "
@@ -3139,7 +3187,10 @@ def check_execute_code_guard(code: str, env_type: str,
"outcome": "denied",
"user_consent": False,
}
- # verdict == "escalate" → fall through to manual approval
+ if verdict == "deny":
+ smart_denied_for_owner = True
+ # Interactive DENY falls through to one-operation human approval;
+ # ESCALATE retains the normal manual approval behavior.
# Redacted copies for user-visible rendering only. An execute_code script
# can embed credentials (e.g. api_key = "sk-..."), and the gateway renders
@@ -3159,13 +3210,16 @@ def check_execute_code_guard(code: str, env_type: str,
if notify_cb is None:
# No gateway callback registered (e.g. ask-mode without a notifier):
# surface a pending approval for backward compatibility.
- submit_pending(session_key, {
+ pending_data = {
"command": display_command,
"pattern_key": pattern_key,
"pattern_keys": [pattern_key],
"description": display_description,
- })
- return {
+ }
+ if smart_denied_for_owner:
+ pending_data.update(smart_denied=True, allow_permanent=False)
+ submit_pending(session_key, pending_data)
+ result = {
"approved": False,
"pattern_key": pattern_key,
"status": "pending_approval",
@@ -3177,13 +3231,19 @@ def check_execute_code_guard(code: str, env_type: str,
f"**Code:**\n```python\n{display_code}\n```"
),
}
+ if smart_denied_for_owner:
+ result.update(smart_denied=True, allow_permanent=False)
+ return result
approval_data = {
"command": display_command,
"pattern_key": pattern_key,
"pattern_keys": [pattern_key],
"description": display_description,
+ "allow_permanent": not smart_denied_for_owner,
}
+ if smart_denied_for_owner:
+ approval_data["smart_denied"] = True
decision = _await_gateway_decision(
session_key, notify_cb, approval_data, surface="gateway"
)
@@ -3223,13 +3283,16 @@ def check_execute_code_guard(code: str, env_type: str,
"deny_reason": deny_reason,
}
- # Approved — persist based on scope (same logic as check_all_command_guards).
- if choice == "session":
- approve_session(session_key, pattern_key)
- elif choice == "always":
- approve_session(session_key, pattern_key)
- approve_permanent(pattern_key)
- save_permanent_allowlist(_permanent_approved)
+ # Never persist a smart-DENY override under the coarse execute_code key;
+ # doing so would approve unrelated future scripts. Manual and ESCALATE
+ # decisions preserve their existing session/permanent behavior.
+ if not smart_denied_for_owner:
+ if choice == "session":
+ approve_session(session_key, pattern_key)
+ elif choice == "always":
+ approve_session(session_key, pattern_key)
+ approve_permanent(pattern_key)
+ save_permanent_allowlist(_permanent_approved)
# choice == "once": no persistence — approval lasts this single call only.
return {"approved": True, "message": None,
diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py
index 44ef03af78..fc13367eb1 100644
--- a/tools/terminal_tool.py
+++ b/tools/terminal_tool.py
@@ -2295,6 +2295,8 @@ def terminal_tool(
"command": approval.get("command", command),
"description": approval.get("description", "command flagged"),
"pattern_key": approval.get("pattern_key", ""),
+ "smart_denied": approval.get("smart_denied", False),
+ "allow_permanent": approval.get("allow_permanent", True),
}, ensure_ascii=False)
# Command was blocked
desc = approval.get("description", "command flagged")
diff --git a/tui_gateway/server.py b/tui_gateway/server.py
index 0a87f133d4..8f72de9169 100644
--- a/tui_gateway/server.py
+++ b/tui_gateway/server.py
@@ -1152,6 +1152,13 @@ def _emit_approval_request(sid: str, data: dict | None) -> None:
platforms and the SSE/API stream fixed in #50767). Reuse the shared gateway
seam so all approval transports redact consistently."""
payload = dict(data or {})
+ if "choices" not in payload:
+ if payload.get("smart_denied"):
+ payload["choices"] = ["once", "deny"]
+ elif payload.get("allow_permanent") is False:
+ payload["choices"] = ["once", "session", "deny"]
+ elif "allow_permanent" in payload:
+ payload["choices"] = ["once", "session", "always", "deny"]
if "command" in payload:
from gateway.run import _redact_approval_command
diff --git a/ui-tui/src/__tests__/approvalAction.test.ts b/ui-tui/src/__tests__/approvalAction.test.ts
index 662fb71b96..0c817c6441 100644
--- a/ui-tui/src/__tests__/approvalAction.test.ts
+++ b/ui-tui/src/__tests__/approvalAction.test.ts
@@ -1,6 +1,6 @@
import { describe, expect, it } from 'vitest'
-import { approvalAction } from '../components/prompts.js'
+import { approvalAction, approvalOptions } from '../components/prompts.js'
describe('approvalAction — pure key dispatch for ApprovalPrompt', () => {
it('maps Esc to deny — parity with global Ctrl+C cancellation', () => {
@@ -58,4 +58,23 @@ describe('approvalAction — pure key dispatch for ApprovalPrompt', () => {
expect(approvalAction('', { downArrow: true }, 2, opts)).toEqual({ kind: 'noop' })
expect(approvalAction('', { return: true }, 2, opts)).toEqual({ kind: 'choose', choice: 'deny' })
})
+
+ it('offers only once and deny for Smart DENY owner override', () => {
+ const opts = approvalOptions({ allowPermanent: true, command: 'rm -rf /', description: 'blocked', smartDenied: true })
+
+ expect(opts).toEqual(['once', 'deny'])
+ expect(approvalAction('2', {}, 0, opts)).toEqual({ kind: 'choose', choice: 'deny' })
+ expect(approvalAction('3', {}, 0, opts)).toEqual({ kind: 'noop' })
+ })
+
+ it('uses explicit gateway choices as the prompt contract', () => {
+ expect(
+ approvalOptions({
+ allowPermanent: true,
+ choices: ['once', 'deny'],
+ command: 'rm -rf /',
+ description: 'blocked'
+ })
+ ).toEqual(['once', 'deny'])
+ })
})
diff --git a/ui-tui/src/__tests__/createGatewayEventHandler.test.ts b/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
index ba36ef6b49..9103877bac 100644
--- a/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
+++ b/ui-tui/src/__tests__/createGatewayEventHandler.test.ts
@@ -949,6 +949,23 @@ describe('createGatewayEventHandler', () => {
})
})
+ it('preserves Smart DENY and explicit approval choices on the overlay', () => {
+ const onEvent = createGatewayEventHandler(buildCtx([]))
+
+ onEvent({
+ payload: {
+ allow_permanent: true,
+ choices: ['once', 'deny'],
+ command: 'rm -rf /tmp/x',
+ description: 'smart deny override',
+ smart_denied: true
+ },
+ type: 'approval.request'
+ } as any)
+
+ expect(getOverlayState().approval).toMatchObject({ choices: ['once', 'deny'], smartDenied: true })
+ })
+
it('still surfaces terminal turn failures as errors', () => {
const appended: Msg[] = []
const onEvent = createGatewayEventHandler(buildCtx(appended))
diff --git a/ui-tui/src/app/createGatewayEventHandler.ts b/ui-tui/src/app/createGatewayEventHandler.ts
index 72cad059eb..051f093477 100644
--- a/ui-tui/src/app/createGatewayEventHandler.ts
+++ b/ui-tui/src/app/createGatewayEventHandler.ts
@@ -786,7 +786,13 @@ export function createGatewayEventHandler(ctx: GatewayEventHandlerContext): (ev:
const allowPermanent = ev.payload.allow_permanent !== false
patchOverlayState({
- approval: { allowPermanent, command: String(ev.payload.command ?? ''), description }
+ approval: {
+ allowPermanent,
+ choices: ev.payload.choices,
+ command: String(ev.payload.command ?? ''),
+ description,
+ smartDenied: ev.payload.smart_denied === true
+ }
})
setStatus('approval needed')
diff --git a/ui-tui/src/components/prompts.tsx b/ui-tui/src/components/prompts.tsx
index acac12eef1..bfd7bf62d2 100644
--- a/ui-tui/src/components/prompts.tsx
+++ b/ui-tui/src/components/prompts.tsx
@@ -10,11 +10,24 @@ import { TextInput } from './textInput.js'
const APPROVAL_OPTS = ['once', 'session', 'always', 'deny'] as const
// tirith warning present → backend downgrades "always" to session scope, so drop it.
const APPROVAL_OPTS_NO_ALWAYS = APPROVAL_OPTS.filter(o => o !== 'always')
+const APPROVAL_OPTS_SMART_DENY = ['once', 'deny'] as const
const LABELS = { always: 'Always allow', deny: 'Deny', once: 'Allow once', session: 'Allow this session' } as const
const CMD_PREVIEW_LINES = 10
type ApprovalChoice = 'always' | 'deny' | 'once' | 'session'
+export function approvalOptions(req: ApprovalReq): readonly ApprovalChoice[] {
+ if (req.choices) {
+ return req.choices.filter((choice): choice is ApprovalChoice => APPROVAL_OPTS.includes(choice as ApprovalChoice))
+ }
+
+ if (req.smartDenied) {
+ return APPROVAL_OPTS_SMART_DENY
+ }
+
+ return req.allowPermanent === false ? APPROVAL_OPTS_NO_ALWAYS : APPROVAL_OPTS
+}
+
type ApprovalKey = {
downArrow?: boolean
escape?: boolean
@@ -68,7 +81,7 @@ export function approvalAction(
export function ApprovalPrompt({ cols = 80, onChoice, req, t }: ApprovalPromptProps) {
const [sel, setSel] = useState(0)
- const opts = req.allowPermanent === false ? APPROVAL_OPTS_NO_ALWAYS : APPROVAL_OPTS
+ const opts = approvalOptions(req)
useInput((ch, key) => {
const action = approvalAction(ch, key, sel, opts)
diff --git a/ui-tui/src/gatewayTypes.ts b/ui-tui/src/gatewayTypes.ts
index 58f1aec7da..46e3019fde 100644
--- a/ui-tui/src/gatewayTypes.ts
+++ b/ui-tui/src/gatewayTypes.ts
@@ -692,7 +692,7 @@ export type GatewayEvent =
type: 'clarify.request'
}
| {
- payload: { allow_permanent?: boolean; command: string; description: string }
+ payload: { allow_permanent?: boolean; choices?: string[]; command: string; description: string; smart_denied?: boolean }
session_id?: string
type: 'approval.request'
}
diff --git a/ui-tui/src/types.ts b/ui-tui/src/types.ts
index 14ab98ca18..6f6818e37c 100644
--- a/ui-tui/src/types.ts
+++ b/ui-tui/src/types.ts
@@ -92,8 +92,10 @@ export interface DelegationStatus {
export interface ApprovalReq {
// false when the backend won't honor a permanent allow (tirith warning) → hide "Always allow".
allowPermanent?: boolean
+ choices?: string[]
command: string
description: string
+ smartDenied?: boolean
}
export interface ConfirmReq {