Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 14 additions & 4 deletions acp_adapter/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,7 @@ def _build_permission_options(*, allow_permanent: bool) -> list[PermissionOption
return options


def _build_permission_tool_call(command: str, description: str):
def _build_permission_tool_call(command: str, description: str, explanation: dict | None = None):
"""Return the ACP tool-call update attached to a permission request.

``request_permission`` expects a ``ToolCallUpdate`` payload — produced
Expand All @@ -81,14 +81,23 @@ def _build_permission_tool_call(command: str, description: str):

tool_call_id = f"perm-check-{next(_PERMISSION_REQUEST_IDS)}"
title = f"{description}: {command}" if description else command
content_text = f"{description}\n$ {command}" if description else f"$ {command}"
action = str((explanation or {}).get("action") or "")
permission = str((explanation or {}).get("permission") or "")
explanation_lines = []
if action:
explanation_lines.append(f"What Hermes is trying to do: {action}")
if permission:
explanation_lines.append(f"Permission requested: {permission}")
explanation_text = "\n".join(explanation_lines)
base_text = f"{description}\n$ {command}" if description else f"$ {command}"
content_text = f"{explanation_text}\n{base_text}" if explanation_text else base_text
return _acp.update_tool_call(
tool_call_id,
title=title,
kind="execute",
status="pending",
content=[_acp.tool_content(_acp.text_block(content_text))],
raw_input={"command": command, "description": description},
raw_input={"command": command, "description": description, "explanation": explanation or {}},
)


Expand Down Expand Up @@ -129,13 +138,14 @@ def _callback(
description: str,
*,
allow_permanent: bool = True,
approval_explanation: dict | None = None,
**_: object,
) -> str:
from agent.async_utils import safe_schedule_threadsafe

options = _build_permission_options(allow_permanent=allow_permanent)

tool_call = _build_permission_tool_call(command, description)
tool_call = _build_permission_tool_call(command, description, approval_explanation)
coro = request_permission_fn(
session_id=session_id,
tool_call=tool_call,
Expand Down
29 changes: 24 additions & 5 deletions cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -10787,7 +10787,8 @@ def _sudo_password_callback(self) -> str:
return ""

def _approval_callback(self, command: str, description: str,
*, allow_permanent: bool = True) -> str:
*, allow_permanent: bool = True,
approval_explanation: dict | None = None) -> str:
"""
Prompt for dangerous command approval through the prompt_toolkit UI.

Expand All @@ -10810,6 +10811,7 @@ def _approval_callback(self, command: str, description: str,
self._approval_state = {
"command": command,
"description": description,
"explanation": approval_explanation or {},
"choices": self._approval_choices(command, allow_permanent=allow_permanent),
"selected": 0,
"response_queue": response_queue,
Expand Down Expand Up @@ -10951,6 +10953,7 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:

command = state["command"]
description = state["description"]
explanation = state.get("explanation") or {}
choices = state["choices"]
selected = state.get("selected", 0)
show_full = state.get("show_full", False)
Expand All @@ -10966,6 +10969,12 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
}

preview_lines = _wrap_panel_text(description, 60)
action_text = explanation.get("action")
permission_text = explanation.get("permission")
if action_text:
preview_lines.extend(_wrap_panel_text(f"What: {action_text}", 60))
if permission_text:
preview_lines.extend(_wrap_panel_text(f"Permission: {permission_text}", 60))
preview_lines.extend(_wrap_panel_text(cmd_display, 60))
for i, choice in enumerate(choices):
prefix = '❯ ' if i == selected else ' '
Expand All @@ -10978,7 +10987,12 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
box_width = _panel_box_width(title, preview_lines)
inner_text_width = max(8, box_width - 2)

# Pre-wrap the mandatory content — command + choices must always render.
# Pre-wrap the mandatory content — explanation + command + choices must always render.
explanation_wrapped: list[str] = []
if action_text:
explanation_wrapped.extend(_wrap_panel_text(f"What: {action_text}", inner_text_width))
if permission_text:
explanation_wrapped.extend(_wrap_panel_text(f"Permission: {permission_text}", inner_text_width))
cmd_wrapped = _wrap_panel_text(cmd_display, inner_text_width)
if not show_full and "view" in choices and len(cmd_wrapped) > 4:
cmd_wrapped = cmd_wrapped[:3] + _wrap_panel_text(
Expand Down Expand Up @@ -11021,7 +11035,7 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
reserved_below = 6

available = max(0, term_rows - reserved_below)
mandatory_full = chrome_full + len(cmd_wrapped) + len(choice_wrapped)
mandatory_full = chrome_full + len(explanation_wrapped) + len(cmd_wrapped) + len(choice_wrapped)

# If the full-chrome panel doesn't fit, drop the separator blanks.
# This keeps the command and every choice on-screen in compact terminals.
Expand All @@ -11031,7 +11045,7 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
# If the command itself is too long to leave room for choices (e.g. user
# hit "view" on a multi-hundred-character command), truncate it so the
# approve/deny buttons still render. Keep at least 1 row of command.
max_cmd_rows = max(1, available - chrome_rows - len(choice_wrapped))
max_cmd_rows = max(1, available - chrome_rows - len(explanation_wrapped) - len(choice_wrapped))
if len(cmd_wrapped) > max_cmd_rows:
keep = max(1, max_cmd_rows - 1) if max_cmd_rows > 1 else 1
cmd_wrapped = cmd_wrapped[:keep] + _wrap_panel_text(
Expand All @@ -11041,7 +11055,7 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:

# Allocate any remaining rows to description. The extra -1 in full mode
# accounts for the blank separator between choices and description.
mandatory_no_desc = chrome_rows + len(cmd_wrapped) + len(choice_wrapped)
mandatory_no_desc = chrome_rows + len(explanation_wrapped) + len(cmd_wrapped) + len(choice_wrapped)
desc_sep_cost = 0 if use_compact_chrome else 1
available_for_desc = available - mandatory_no_desc - desc_sep_cost
# Even on huge terminals, cap description height so the panel stays compact.
Expand All @@ -11064,6 +11078,11 @@ def _append_blank_panel_line(lines, border_style: str, box_width: int) -> None:
if not use_compact_chrome:
_append_blank_panel_line(lines, 'class:approval-border', box_width)

for wrapped in explanation_wrapped:
_append_panel_line(lines, 'class:approval-border', 'class:approval-desc', wrapped, box_width)
if explanation_wrapped and not use_compact_chrome:
_append_blank_panel_line(lines, 'class:approval-border', box_width)

for wrapped in cmd_wrapped:
_append_panel_line(lines, 'class:approval-border', 'class:approval-cmd', wrapped, box_width)
if not use_compact_chrome:
Expand Down
16 changes: 15 additions & 1 deletion gateway/run.py
Original file line number Diff line number Diff line change
Expand Up @@ -15785,6 +15785,7 @@ def _approval_notify_sync(approval_data: dict) -> None:

cmd = approval_data.get("command", "")
desc = approval_data.get("description", "dangerous command")
explanation = approval_data.get("explanation") or {}

# Redact credentials from the command before displaying it in
# the approval prompt — Tirith's findings are already redacted,
Expand All @@ -15805,7 +15806,10 @@ def _approval_notify_sync(approval_data: dict) -> None:
command=cmd,
session_key=_approval_session_key,
description=desc,
metadata=_status_thread_metadata,
metadata={
**(_status_thread_metadata or {}),
"approval_explanation": explanation,
},
),
_loop_for_step,
logger=logger,
Expand All @@ -15831,8 +15835,18 @@ def _approval_notify_sync(approval_data: dict) -> None:
# Slack threads and reserved by Matrix clients.
_p = getattr(_status_adapter, "typed_command_prefix", "/")
cmd_preview = cmd[:200] + "..." if len(cmd) > 200 else cmd
action_line = (
f"What Hermes is trying to do: {explanation.get('action')}\n"
if explanation.get("action") else ""
)
permission_line = (
f"Permission requested: {explanation.get('permission')}\n"
if explanation.get("permission") else ""
)
msg = (
f"⚠️ **Dangerous command requires approval:**\n"
f"{action_line}"
f"{permission_line}"
f"```\n{cmd_preview}\n```\n"
f"Reason: {desc}\n\n"
f"Reply `{_p}approve` to execute, `{_p}approve session` to approve this pattern "
Expand Down
15 changes: 13 additions & 2 deletions plugins/platforms/discord/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -4714,11 +4714,22 @@ async def send_exec_approval(
channel = await self._client.fetch_channel(int(target_id))

# Discord embed description limit is 4096; show full command up to that
max_desc = 4088
explanation = (metadata or {}).get("approval_explanation") or {}
action_text = str(explanation.get("action") or "")
permission_text = str(explanation.get("permission") or "")
explanation_parts = []
if action_text:
explanation_parts.append(f"**What Hermes is trying to do:** {action_text}")
if permission_text:
explanation_parts.append(f"**Permission requested:** {permission_text}")
explanation_prefix = "\n".join(explanation_parts)
if explanation_prefix:
explanation_prefix += "\n\n"
max_desc = max(200, 4088 - len(explanation_prefix))
cmd_display = command if len(command) <= max_desc else command[: max_desc - 3] + "..."
embed = discord.Embed(
title="⚠️ Command Approval Required",
description=f"```\n{cmd_display}\n```",
description=f"{explanation_prefix}```\n{cmd_display}\n```",
color=discord.Color.orange(),
)
embed.add_field(name="Reason", value=description, inline=False)
Expand Down
10 changes: 9 additions & 1 deletion plugins/platforms/feishu/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -1881,6 +1881,14 @@ async def send_exec_approval(
try:
approval_id = next(self._approval_counter)
cmd_preview = command[:3000] + "..." if len(command) > 3000 else command
explanation = (metadata or {}).get("approval_explanation") or {}
action_text = str(explanation.get("action") or "")
permission_text = str(explanation.get("permission") or "")
explanation_lines = ""
if action_text:
explanation_lines += f"**What Hermes is trying to do:** {action_text}\n"
if permission_text:
explanation_lines += f"**Permission requested:** {permission_text}\n"

def _btn(label: str, action_name: str, btn_type: str = "default") -> dict:
return {
Expand All @@ -1899,7 +1907,7 @@ def _btn(label: str, action_name: str, btn_type: str = "default") -> dict:
"elements": [
{
"tag": "markdown",
"content": f"```\n{cmd_preview}\n```\n**Reason:** {description}",
"content": f"{explanation_lines}```\n{cmd_preview}\n```\n**Reason:** {description}",
},
{
"tag": "action",
Expand Down
9 changes: 9 additions & 0 deletions plugins/platforms/matrix/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -1915,9 +1915,18 @@ async def send_exec_approval(
return SendResult(success=False, error="Not connected")

requester_user_id = str((metadata or {}).get("requester_user_id") or "") or None
explanation = (metadata or {}).get("approval_explanation") or {}
action_text = str(explanation.get("action") or "")
permission_text = str(explanation.get("permission") or "")
explanation_lines = ""
if action_text:
explanation_lines += f"What Hermes is trying to do: {action_text}\n"
if permission_text:
explanation_lines += f"Permission requested: {permission_text}\n"
cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
text = (
"⚠️ **Dangerous command requires approval**\n"
f"{explanation_lines}"
f"```\n{cmd_preview}\n```\n"
f"Reason: {description}\n\n"
"Reply `!approve` to execute, `!approve session` to approve this pattern for the session, "
Expand Down
13 changes: 11 additions & 2 deletions plugins/platforms/slack/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -3011,17 +3011,26 @@ async def send_exec_approval(
# ``command``, so budget the preview against the fixed parts
# instead of a flat truncation that overflows once the header +
# reason are added.
explanation = (metadata or {}).get("approval_explanation") or {}
action_text = str(explanation.get("action") or "")
permission_text = str(explanation.get("permission") or "")
header = ":warning: *Command Approval Required*\n"
explanation_text = ""
if action_text:
explanation_text += f"What Hermes is trying to do: {action_text}\n"
if permission_text:
explanation_text += f"Permission requested: {permission_text}\n"
reason = f"Reason: {description[:500]}"
budget = 3000 - len(header) - len(reason) - len("``````\n") - len("...")
budget = 3000 - len(header) - len(explanation_text) - len(reason) - len("``````\n") - len("...")
budget = max(200, budget)
cmd_preview = command[:budget] + "..." if len(command) > budget else command

blocks = [
{
"type": "section",
"text": {
"type": "mrkdwn",
"text": f"{header}```{cmd_preview}```\n{reason}",
"text": f"{header}{explanation_text}```{cmd_preview}```\n{reason}",
},
},
{
Expand Down
9 changes: 9 additions & 0 deletions plugins/platforms/telegram/adapter.py
Original file line number Diff line number Diff line change
Expand Up @@ -3472,8 +3472,17 @@ async def send_exec_approval(

try:
cmd_preview = command[:3800] + "..." if len(command) > 3800 else command
explanation = (metadata or {}).get("approval_explanation") or {}
action_text = str(explanation.get("action") or "")
permission_text = str(explanation.get("permission") or "")
explanation_lines = ""
if action_text:
explanation_lines += f"What Hermes is trying to do: {_html.escape(action_text)}\n"
if permission_text:
explanation_lines += f"Permission requested: {_html.escape(permission_text)}\n"
text = (
f"⚠️ <b>Command Approval Required</b>\n\n"
f"{explanation_lines}"
f"<pre>{_html.escape(cmd_preview)}</pre>\n\n"
f"Reason: {_html.escape(description)}"
)
Expand Down
34 changes: 34 additions & 0 deletions tests/acp/test_permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ def test_bridge_schedules_request_on_the_given_loop(self):
assert tool_call.raw_input == {
"command": "rm -rf /",
"description": "dangerous command",
"explanation": {},
}
assert option_ids == [
"allow_once",
Expand All @@ -93,6 +94,39 @@ def test_bridge_schedules_request_on_the_given_loop(self):
"deny_always",
]

def test_bridge_includes_permission_context_when_provided(self):
explanation = {
"action": "Run the shown terminal command.",
"permission": "Allow Hermes to execute this command.",
}
loop = MagicMock(spec=asyncio.AbstractEventLoop)
request_permission = AsyncMock(name="request_permission")
future = MagicMock(spec=Future)
future.result.return_value = _make_response(
AllowedOutcome(option_id="allow_once", outcome="selected")
)
scheduled = {}

def _schedule(coro, passed_loop):
scheduled["coro"] = coro
return future

with patch("agent.async_utils.asyncio.run_coroutine_threadsafe", side_effect=_schedule):
cb = make_approval_callback(request_permission, loop, session_id="s1")
result = cb(
"rm -rf /",
"dangerous command",
approval_explanation=explanation,
)

scheduled["coro"].close()
_, kwargs = request_permission.call_args
content_text = kwargs["tool_call"].content[0].content.text
assert result == "once"
assert "What Hermes is trying to do: Run the shown terminal command." in content_text
assert "Permission requested: Allow Hermes to execute this command." in content_text
assert kwargs["tool_call"].raw_input["explanation"] == explanation

def test_tool_call_ids_are_unique(self):
_, first_kwargs, _, _, _ = _invoke_callback(
AllowedOutcome(option_id="allow_once", outcome="selected"),
Expand Down
26 changes: 26 additions & 0 deletions tests/gateway/test_telegram_approval_buttons.py
Original file line number Diff line number Diff line change
Expand Up @@ -105,6 +105,32 @@ async def test_sends_inline_keyboard(self):
assert "dangerous deletion" in kwargs["text"]
assert kwargs["reply_markup"] is not None # InlineKeyboardMarkup

@pytest.mark.asyncio
async def test_includes_permission_context(self):
adapter = _make_adapter()
mock_msg = MagicMock()
mock_msg.message_id = 42
adapter._bot.send_message = AsyncMock(return_value=mock_msg)

await adapter.send_exec_approval(
chat_id="12345",
command="rm -rf /important",
session_key="s",
description="dangerous deletion",
metadata={
"approval_explanation": {
"action": "Run the shown terminal command.",
"permission": "Allow Hermes to execute this command.",
}
},
)

text = adapter._bot.send_message.call_args[1]["text"]
assert "What Hermes is trying to do" in text
assert "Run the shown terminal command." in text
assert "Permission requested" in text
assert "Allow Hermes to execute this command." in text

@pytest.mark.asyncio
async def test_stores_approval_state(self):
adapter = _make_adapter()
Expand Down
Loading