mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-20 15:33:54 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

fix(approval): scope smart deny owner overrides to one operation

Browse source 
Co-authored-by: Sergei Ivanov <kavi@local.hermes>
This commit is contained in:
Teknium 2026-07-12 20:06:15 -07:00
parent f96b2e6ef7
commit d48bf743f2
52 changed files with 1011 additions and 203 deletions
Download patch file Download diff file Expand all files Collapse all files

25
acp_adapter/permissions.py
View file

@ -38,19 +38,22 @@ def _permission_option_supports_kind(kind: str) -> bool:
return True
def _build_permission_options(*, allow_permanent: bool) -> list[PermissionOption]:
def _build_permission_options(
*, allow_permanent: bool, smart_denied: bool = False,
) -> list[PermissionOption]:
"""Return ACP options that match Hermes approval semantics."""
options = [
PermissionOption(option_id="allow_once", kind="allow_once", name="Allow once"),
PermissionOption(
options = [PermissionOption(
option_id="allow_once", kind="allow_once", name="Allow once",
)]
if not smart_denied:
options.append(PermissionOption(
option_id="allow_session",
# ACP has no session-scoped kind, so use the closest persistent
# hint while keeping Hermes semantics in the option id.
kind="allow_always",
name="Allow for session",
),
]
if allow_permanent:
))
if allow_permanent and not smart_denied:
options.append(
PermissionOption(
option_id="allow_always",
@ -59,7 +62,7 @@ def _build_permission_options(*, allow_permanent: bool) -> list[PermissionOption
),
)
options.append(PermissionOption(option_id="deny", kind="reject_once", name="Deny"))
if _permission_option_supports_kind("reject_always"):
if not smart_denied and _permission_option_supports_kind("reject_always"):
options.append(
PermissionOption(
option_id="deny_always",
@ -129,11 +132,15 @@ def make_approval_callback(
description: str,
*,
allow_permanent: bool = True,
smart_denied: bool = False,
**_: object,
) -> str:
from agent.async_utils import safe_schedule_threadsafe
options = _build_permission_options(allow_permanent=allow_permanent)
options = _build_permission_options(
allow_permanent=allow_permanent,
smart_denied=smart_denied,
)
tool_call = _build_permission_tool_call(command, description)
coro = request_permission_fn(

4
apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts
View file

@ -489,9 +489,11 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
setApprovalRequest({
// false only when a tirith warning forbids it; backend omits the field otherwise.
allowPermanent: payload?.allow_permanent !== false,
choices: Array.isArray(payload?.choices) ? payload.choices.filter(choice => typeof choice === 'string') : undefined,
command,
description,
sessionId: sessionId ?? null
sessionId: sessionId ?? null,
smartDenied: payload?.smart_denied === true
})
if (sessionId) {

28
apps/desktop/src/components/assistant-ui/tool/approval.test.tsx
View file

@ -30,9 +30,13 @@ function part(toolName: string): ToolPart {
return { toolName, type: `tool-${toolName}` } as unknown as ToolPart
}
function setRequest(command = 'rm -rf /tmp/x', allowPermanent?: boolean) {
function setRequest(
command = 'rm -rf /tmp/x',
allowPermanent?: boolean,
extra: { choices?: string[]; smartDenied?: boolean } = {}
) {
$activeSessionId.set('sess-1')
setApprovalRequest({ allowPermanent, command, description: 'dangerous command', sessionId: 'sess-1' })
setApprovalRequest({ allowPermanent, command, description: 'dangerous command', sessionId: 'sess-1', ...extra })
}
function mockGateway() {
@ -131,6 +135,26 @@ describe('PendingToolApproval', () => {
expect(screen.queryByRole('menuitem', { name: /Always allow/ })).toBeNull()
})
it('renders only Once and Deny for a Smart DENY owner override', () => {
setRequest('rm -rf /tmp/x', true, { smartDenied: true })
render(<PendingToolApproval part={part('terminal')} />)
expect(screen.getByRole('button', { name: /Run/ })).toBeTruthy()
expect(screen.getByRole('button', { name: /Reject/ })).toBeTruthy()
expect(screen.queryByRole('button', { name: /More approval options/ })).toBeNull()
expect(screen.queryByText(/Allow this session/)).toBeNull()
expect(screen.queryByText(/Always allow/)).toBeNull()
})
it('renders only choices explicitly supplied by the gateway event', () => {
setRequest('rm -rf /tmp/x', true, { choices: ['once', 'deny'] })
render(<PendingToolApproval part={part('terminal')} />)
expect(screen.getByRole('button', { name: /Run/ })).toBeTruthy()
expect(screen.getByRole('button', { name: /Reject/ })).toBeTruthy()
expect(screen.queryByRole('button', { name: /More approval options/ })).toBeNull()
})
it('renders a floating fallback when no pending tool row is mounted', () => {
setRequest('rm /tmp/hermes_approval_test.txt')
const { container } = render(<PendingApprovalFallback />)

14
apps/desktop/src/components/assistant-ui/tool/approval.tsx
View file

@ -110,6 +110,10 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
const busy = submitting !== null
// false when the backend won't honor a permanent allow (tirith warning) → hide "Always allow".
const allowPermanent = request.allowPermanent !== false
const choices = request.choices ?? (request.smartDenied ? ['once', 'deny'] : undefined)
const allowSession = choices ? choices.includes('session') : true
const allowAlways = choices ? choices.includes('always') : allowPermanent
const hasMoreOptions = allowSession || allowAlways
const hasCommand = request.command.trim().length > 0
const respond = useCallback(
@ -183,8 +187,8 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
{submitting === 'once' ? <Loader2 className="size-3 animate-spin" /> : copy.run}
{submitting !== 'once' && <span className="text-[0.625rem] text-primary/60">{isMac ? '⌘⏎' : 'Ctrl⏎'}</span>}
</Button>
<span aria-hidden className="w-px self-stretch bg-primary/20" />
<DropdownMenu>
{hasMoreOptions && <span aria-hidden className="w-px self-stretch bg-primary/20" />}
{hasMoreOptions && <DropdownMenu>
<DropdownMenuTrigger asChild>
<Button
aria-label={copy.moreOptions}
@ -197,8 +201,8 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
</Button>
</DropdownMenuTrigger>
<DropdownMenuContent align="start" className="min-w-44">
<DropdownMenuItem onSelect={() => void respond('session')}>{copy.allowSession}</DropdownMenuItem>
{allowPermanent && (
{allowSession && <DropdownMenuItem onSelect={() => void respond('session')}>{copy.allowSession}</DropdownMenuItem>}
{allowAlways && (
<DropdownMenuItem
onSelect={() => {
// Defer one tick so the menu fully unmounts before the dialog
@ -214,7 +218,7 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
{copy.reject}
</DropdownMenuItem>
</DropdownMenuContent>
</DropdownMenu>
</DropdownMenu>}
</div>
<Button

1
apps/desktop/src/lib/chat-messages.ts
View file

@ -67,6 +67,7 @@ export type GatewayEventPayload = {
description?: string
// False when a tirith content-security warning forbids a permanent allow.
allow_permanent?: boolean
smart_denied?: boolean
// secret.request (skill credential capture)
env_var?: string
prompt?: string

2
apps/desktop/src/store/prompts.ts
View file

@ -71,8 +71,10 @@ function keyedPromptStore<T extends KeyedPrompt>(): PromptStore<T> {
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 {

22
cli.py
View file

@ -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

11
gateway/platforms/api_server.py
View file

@ -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,

10
gateway/platforms/qqbot/adapter.py
View file

@ -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,

54
gateway/platforms/qqbot/keyboards.py
View file

@ -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:

4
gateway/platforms/whatsapp_cloud.py
View file

@ -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]

43
gateway/run.py
View file

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

4
locales/af.yaml
View file

@ -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"

4
locales/de.yaml
View file

@ -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"

4
locales/en.yaml
View file

@ -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"

4
locales/es.yaml
View file

@ -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"

4
locales/fr.yaml
View file

@ -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"

4
locales/ga.yaml
View file

@ -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"

4
locales/hu.yaml
View file

@ -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"

4
locales/it.yaml
View file

@ -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"

4
locales/ja.yaml
View file

@ -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: " ✓ このセッション中は許可"

4
locales/ko.yaml
View file

@ -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: " ✓ 이 세션에서 허용됨"

4
locales/pt.yaml
View file

@ -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"

4
locales/ru.yaml
View file

@ -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: " ✓ Разрешено для этого сеанса"

4
locales/tr.yaml
View file

@ -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"

4
locales/uk.yaml
View file

@ -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: " ✓ Дозволено для цього сеансу"

4
locales/zh-hant.yaml
View file

@ -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: " ✓ 本次工作階段內允許"

4
locales/zh.yaml
View file

@ -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: " ✓ 本次会话内允许"

13
plugins/platforms/discord/adapter.py
View file

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

18
plugins/platforms/feishu/adapter.py
View file

@ -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,
},
],
}

15
plugins/platforms/matrix/adapter.py
View file

@ -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

66
plugins/platforms/slack/adapter.py
View file

@ -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] = {

61
plugins/platforms/teams/adapter.py
View file

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

27
plugins/platforms/telegram/adapter.py
View file

@ -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"<pre>{_html.escape(cmd_preview)}</pre>\n\n"
f"Reason: {_html.escape(description)}"
)
if smart_denied:
text += "\n\n<b>Smart DENY:</b> 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),

27
tests/acp/test_permissions.py
View file

@ -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"),

32
tests/cli/test_cli_approval_ui.py
View file

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

19
tests/gateway/test_api_server_runs.py
View file

@ -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 = {}

60
tests/gateway/test_approval_prompt_redaction.py
View file

@ -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

18
tests/gateway/test_slack_approval_buttons.py
View file

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

43
tests/gateway/test_telegram_approval_buttons.py
View file

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

22
tests/gateway/test_tui_approval_redaction.py
View file

@ -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

106
tests/tools/test_approval.py
View file

@ -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."""

168
tests/tools/test_execute_code_approval_cluster.py
View file

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

163
tools/approval.py
View file

@ -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,

2
tools/terminal_tool.py
View file

@ -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")

7
tui_gateway/server.py
View file

@ -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

21
ui-tui/src/__tests__/approvalAction.test.ts
View file

@ -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'])
})
})

17
ui-tui/src/__tests__/createGatewayEventHandler.test.ts
View file

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

8
ui-tui/src/app/createGatewayEventHandler.ts
View file

@ -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')

15
ui-tui/src/components/prompts.tsx
View file

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

2
ui-tui/src/gatewayTypes.ts
View file

@ -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'
}

2
ui-tui/src/types.ts
View file

@ -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 {