From b4b9656d4c24aa9f03b1b14a2b10d76b7c25978e Mon Sep 17 00:00:00 2001 From: sqp-adicr <84441485+sqp-adicr@users.noreply.github.com> Date: Sat, 9 May 2026 15:14:26 +0800 Subject: [PATCH 1/2] feat: surface command context in approval prompts Add optional purpose/effect/risk context fields to terminal tool calls and thread them through command approval requests. Gateway approval prompts keep the existing request message intact and send a follow-up context message when explanation data is available. --- tests/tools/test_command_guards.py | 77 +++++++++++++++++++++++++++++- tools/approval.py | 44 ++++++++++++++++- tools/terminal_tool.py | 33 ++++++++++++- 3 files changed, 150 insertions(+), 4 deletions(-) diff --git a/tests/tools/test_command_guards.py b/tests/tools/test_command_guards.py index da385afcc3d0..8d114b608060 100644 --- a/tests/tools/test_command_guards.py +++ b/tests/tools/test_command_guards.py @@ -50,6 +50,8 @@ def _mode_manual(monkeypatch): def _clean_state(): """Clear approval state and relevant env vars between tests.""" approval_module._session_approved.clear() + approval_module._gateway_queues.clear() + approval_module._gateway_notify_cbs.clear() approval_module._pending.clear() approval_module._permanent_approved.clear() saved = {} @@ -58,6 +60,8 @@ def _clean_state(): saved[k] = os.environ.pop(k) yield approval_module._session_approved.clear() + approval_module._gateway_queues.clear() + approval_module._gateway_notify_cbs.clear() approval_module._pending.clear() approval_module._permanent_approved.clear() for k, v in saved.items(): @@ -347,9 +351,80 @@ def test_warn_empty_findings_cli_prompts(self, mock_tirith): # --------------------------------------------------------------------------- -# Programming errors propagate through orchestration +# Approval context # --------------------------------------------------------------------------- +class TestApprovalContext: + def test_clean_approval_context_accepts_tool_schema_aliases(self): + cleaned = approval_module._clean_approval_context({ + "approval_purpose": " explain why ", + "approval_effect": " explain effect ", + "approval_risk": " explain risk ", + "ignored": "value", + "purpose": "overridden by alias order", + }) + assert cleaned == { + "purpose": "explain why", + "effect": "explain effect", + "risk": "explain risk", + } + + def test_clean_approval_context_ignores_empty_and_non_strings(self): + cleaned = approval_module._clean_approval_context({ + "purpose": " ", + "effect": 123, + "risk": "real risk", + }) + assert cleaned == {"risk": "real risk"} + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_gateway_approval_data_includes_context(self, mock_tirith): + os.environ["HERMES_GATEWAY_SESSION"] = "1" + session_key = "test-session" + token = set_current_session_key(session_key) + seen = {} + + def notify_cb(data): + seen.update(data) + queue = approval_module._gateway_queues[session_key] + queue[0].result = "deny" + queue[0].event.set() + + approval_module.register_gateway_notify(session_key, notify_cb) + try: + result = check_all_command_guards( + "rm -rf /tmp/example", + "local", + approval_context={ + "purpose": "clean a temp path", + "effect": "removes temporary files", + "risk": "deleted files cannot be recovered", + }, + ) + finally: + approval_module.unregister_gateway_notify(session_key) + reset_current_session_key(token) + + assert result["approved"] is False + assert seen["explanation"] == { + "purpose": "clean a temp path", + "effect": "removes temporary files", + "risk": "deleted files cannot be recovered", + } + + +# --------------------------------------------------------------------------- +# Terminal schema exposes approval context +# --------------------------------------------------------------------------- + +def test_terminal_schema_exposes_approval_context_fields(): + from tools.terminal_tool import TERMINAL_SCHEMA + + props = TERMINAL_SCHEMA["parameters"]["properties"] + assert "approval_purpose" in props + assert "approval_effect" in props + assert "approval_risk" in props + class TestProgrammingErrorsPropagateFromWrapper: @patch(_TIRITH_PATCH, side_effect=AttributeError("bug in wrapper")) def test_attribute_error_propagates(self, mock_tirith): diff --git a/tools/approval.py b/tools/approval.py index 5db5065f9063..768f82e4e4b6 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -3362,9 +3362,37 @@ def _drop_entry() -> None: return {"resolved": resolved, "choice": choice, "reason": entry.reason} +def _clean_approval_context(approval_context: dict | None) -> dict: + """Normalize optional model-supplied approval context.""" + if not isinstance(approval_context, dict): + return {} + allowed = { + "purpose": "purpose", + "effect": "effect", + "risk": "risk", + "approval_purpose": "purpose", + "approval_effect": "effect", + "approval_risk": "risk", + } + cleaned = {} + for src, dst in allowed.items(): + value = approval_context.get(src) + if isinstance(value, str): + value = value.strip() + if value: + cleaned[dst] = value[:1000] + return cleaned + + +def _approval_context_or_fallback(approval_context: dict | None) -> dict: + """Return normalized model-supplied approval context, if provided.""" + return _clean_approval_context(approval_context) + + def check_all_command_guards(command: str, env_type: str, approval_callback=None, - has_host_access: bool = False) -> dict: + has_host_access: bool = False, + approval_context: dict | None = None) -> dict: """Run all pre-exec security checks and return a single approval decision. Gathers findings from tirith and dangerous-command detection, then @@ -3375,6 +3403,9 @@ def check_all_command_guards(command: str, env_type: str, ``has_host_access`` is True when a Docker sandbox bind-mounts host paths; such a session is no longer isolated, so it goes through the normal flow instead of the container fast-path. + + ``approval_context`` is optional model-supplied context explaining why the + command is being run. It is only surfaced when approval is required. """ # Skip isolated container backends for both checks. Docker stops skipping # once host paths are bind-mounted into the sandbox. @@ -3614,6 +3645,15 @@ def check_all_command_guards(command: str, env_type: str, # Combine descriptions for a single approval prompt combined_desc = "; ".join(desc for _, desc, _ in warnings) + approval_explanation = _approval_context_or_fallback(approval_context) + # Approval output is a secret-egress boundary: model-supplied context is + # displayed verbatim to the user, so redact credential-shaped strings the + # same way the command and description are redacted before display. + if approval_explanation: + from agent.redact import redact_sensitive_text as _redact_explanation + approval_explanation = { + k: _redact_explanation(v) for k, v in approval_explanation.items() + } primary_key = warnings[0][0] all_keys = [key for key, _, _ in warnings] # "Always" is offered when at least one warning is a dangerous-pattern @@ -3653,6 +3693,7 @@ def check_all_command_guards(command: str, env_type: str, "pattern_key": primary_key, "pattern_keys": all_keys, "description": redact_sensitive_text(combined_desc), + "explanation": approval_explanation, # Smart DENY overrides are one-operation decisions, so the UI # must not offer a permanent scope. Otherwise offer Always # whenever any dangerous-pattern warning can actually be @@ -3749,6 +3790,7 @@ def check_all_command_guards(command: str, env_type: str, "pattern_key": primary_key, "pattern_keys": all_keys, "description": _disp_combined_desc, + "explanation": approval_explanation, } if smart_denied_for_owner: pending_data.update(smart_denied=True, allow_permanent=False) diff --git a/tools/terminal_tool.py b/tools/terminal_tool.py index b02c4bd2f910..2b1b13c744b9 100644 --- a/tools/terminal_tool.py +++ b/tools/terminal_tool.py @@ -344,11 +344,13 @@ def _docker_has_host_access(config: Dict[str, Any]) -> bool: def _check_all_guards(command: str, env_type: str, - has_host_access: bool = False) -> dict: + has_host_access: bool = False, + approval_context: dict | None = None) -> dict: """Delegate to consolidated guard (tirith + dangerous cmd) with CLI callback.""" return _check_all_guards_impl(command, env_type, approval_callback=_get_approval_callback(), - has_host_access=has_host_access) + has_host_access=has_host_access, + approval_context=approval_context) # Allowlist: characters that can legitimately appear in directory paths. @@ -2192,6 +2194,9 @@ def terminal_tool( pty: bool = False, notify_on_complete: bool = False, watch_patterns: Optional[List[str]] = None, + approval_purpose: Optional[str] = None, + approval_effect: Optional[str] = None, + approval_risk: Optional[str] = None, ) -> str: """ Execute a command in the configured terminal environment. @@ -2207,6 +2212,9 @@ def terminal_tool( pty: If True, use pseudo-terminal for interactive CLI tools (local backend only) notify_on_complete: If True and background=True, you'll be notified exactly once when the process exits. The right choice for almost every long task. MUTUALLY EXCLUSIVE with watch_patterns. watch_patterns: List of strings to watch for in background output. HARD rate limit: 1 notification per 15s per process. After 3 strike windows in a row, watch_patterns is disabled and the session is auto-promoted to notify_on_complete. Use ONLY for rare, one-shot mid-process signals on long-lived processes (server readiness, migration-done markers). NEVER use in loops/batch jobs — error patterns there will hit the strike limit and get disabled. MUTUALLY EXCLUSIVE with notify_on_complete — set one, not both. + approval_purpose: Optional explanation of why this command is needed, shown to the user if approval is required. + approval_effect: Optional explanation of what this command changes, shown to the user if approval is required. + approval_risk: Optional explanation of risks, shown to the user if approval is required. Returns: str: JSON string with output, exit_code, and error fields @@ -2463,9 +2471,15 @@ def terminal_tool( # the approval-wait (see clear_current_thread_interrupt). _approved_run = bool(force) if not force: + approval_context = { + "purpose": approval_purpose, + "effect": approval_effect, + "risk": approval_risk, + } approval = _check_all_guards( command, env_type, has_host_access=_docker_has_host_access(config), + approval_context=approval_context, ) if not approval["approved"]: # Check if this is an approval_required (gateway ask mode) @@ -3194,6 +3208,18 @@ def check_terminal_requirements() -> bool: "type": "array", "items": {"type": "string"}, "description": "Strings to watch for in background process output. HARD RATE LIMIT: at most 1 notification per 15 seconds per process — matches arriving inside the cooldown are dropped. After 3 consecutive 15-second windows with dropped matches, watch_patterns is automatically disabled for that process and promoted to notify_on_complete behavior (one notification on exit, no more mid-process spam). USE ONLY for truly rare, one-shot mid-process signals on LONG-LIVED processes that will never exit on their own — e.g. ['Application startup complete'] on a server so you know when to hit its endpoint, or ['migration done'] on a daemon. DO NOT use for: (1) end-of-run markers like 'DONE'/'PASS' — use notify_on_complete instead; (2) error patterns like 'ERROR'/'Traceback' in loops or multi-item batch jobs — they fire on every iteration and you'll hit the strike limit fast; (3) anything you'd ever combine with notify_on_complete. When in doubt, choose notify_on_complete. MUTUALLY EXCLUSIVE with notify_on_complete — set one, not both." + }, + "approval_purpose": { + "type": "string", + "description": "If this command triggers approval, explain its purpose to the user. Do not include secrets, tokens, passwords, or credentials." + }, + "approval_effect": { + "type": "string", + "description": "If this command triggers approval, explain what it will change or affect. Do not include secrets, tokens, passwords, or credentials." + }, + "approval_risk": { + "type": "string", + "description": "If this command triggers approval, explain risks the user should consider. Do not include secrets, tokens, passwords, or credentials." } }, "required": ["command"] @@ -3212,6 +3238,9 @@ def _handle_terminal(args, **kw): pty=args.get("pty", False), notify_on_complete=args.get("notify_on_complete", False), watch_patterns=args.get("watch_patterns"), + approval_purpose=args.get("approval_purpose"), + approval_effect=args.get("approval_effect"), + approval_risk=args.get("approval_risk"), ) From 2371649b4b7bb91ecfaac42a419bcf584ad2696b Mon Sep 17 00:00:00 2001 From: troyechung <144773789+troyechung@users.noreply.github.com> Date: Fri, 31 Jul 2026 23:08:58 +0800 Subject: [PATCH 2/2] fix(approvals): include context in approval prompts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Enhance approval prompts with model-supplied Purpose/Effect/Risk context co-located in a single message — no standalone follow-up. - tools/approval.py: _sanitize_explanation, _build_enhanced_description_with_context, enhanced_desc replaces combined_desc on all user-facing surfaces - gateway/run.py: extract _deliver_approval_message (module-level, testable); DeliveryError for fail-closed delivery; button success returns, button explicit failure falls through to text, unknown/None/timeout raises DeliveryError - prompt_dangerous_approval: CLI displays unverified-context annotation - tests: production-delivery, redaction, fail-closed, E2E, AST wiring --- gateway/run.py | 188 +++++--- .../gateway/test_approval_context_message.py | 445 ++++++++++++++++++ .../gateway/test_approval_prompt_redaction.py | 4 +- tests/tools/test_command_guards.py | 171 ++++++- tools/approval.py | 126 ++++- 5 files changed, 844 insertions(+), 90 deletions(-) create mode 100644 tests/gateway/test_approval_context_message.py diff --git a/gateway/run.py b/gateway/run.py index 911021a36ee6..fa94e52bfd4a 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -447,6 +447,12 @@ def _redact_gateway_user_facing_secrets(text: str) -> str: return redacted +class DeliveryError(RuntimeError): + """Raised when approval message delivery fails irrecoverably — + the caller must NOT retry and must NOT proceed as if the user + received an actionable approval prompt.""" + + def _redact_approval_command(cmd: "str | None") -> str: """Redact credentials from a command before it goes into an approval prompt. @@ -492,6 +498,110 @@ def _format_exec_approval_fallback( ) +def _deliver_approval_message( + adapter, + chat_id, + command: str, + description: str, + session_key: str, + metadata, + loop, + logger, + *, + allow_permanent: bool = True, + allow_session: bool = True, + smart_denied: bool = False, +) -> None: + """Deliver the approval request as a single message carrying the full + context (system risk + model-supplied Purpose/Effect/Risk). + + Button-capable platforms receive a button-based prompt whose + *description* already carries the combined context — no separate + follow-up is sent. Text-only platforms receive exactly one + ``send()`` call with the approval, context, and instructions all + in one message. + + Raises ``ValueError`` when *description* is empty (fail-closed: + callers guarantee a non-empty enhanced description upstream). + """ + if not description or not str(description).strip(): + raise ValueError( + "Approval description is empty — refusing to deliver " + "an insufficient approval prompt." + ) + + cmd = _redact_approval_command(command) + + # description is already redacted upstream (_sanitize_explanation), + # but re-redact here as defense-in-depth for the outbound channel. + description = _redact_approval_command(description) + + # --- Button path --- + if getattr(type(adapter), "send_exec_approval", None) is not None: + try: + _fut = safe_schedule_threadsafe( + adapter.send_exec_approval( + chat_id=chat_id, + command=cmd, + session_key=session_key, + description=description, + metadata=metadata, + allow_permanent=allow_permanent, + allow_session=allow_session, + smart_denied=smart_denied, + ), + loop, + logger=logger, + log_message="send_exec_approval scheduling error", + ) + if _fut is None: + raise DeliveryError( + "send_exec_approval: loop unavailable — delivery status unknown" + ) + _result = _fut.result(timeout=15) + if _result.success: + return # button success — single message, no follow-up + # _result.success is False — button explicitly failed, + # fall through to the text path below. + logger.warning( + "Button-based approval failed, falling back to text: %s", + _result.error, + ) + except DeliveryError: + raise + except Exception as _e: + raise DeliveryError( + f"send_exec_approval delivery failed: {_e}" + ) from _e + + # --- Text fallback: single message with full context --- + _p = getattr(adapter, "typed_command_prefix", "/") + msg = _format_exec_approval_fallback( + cmd, description, _p, + allow_permanent=allow_permanent, + allow_session=allow_session, + smart_denied=smart_denied, + ) + try: + _send_fut = safe_schedule_threadsafe( + adapter.send(chat_id, msg, metadata=metadata), + loop, + logger=logger, + log_message="Approval text-send scheduling error", + ) + if _send_fut is None: + raise DeliveryError( + "Approval text-send: loop unavailable" + ) + _send_fut.result(timeout=15) + except DeliveryError: + raise + except Exception as _e: + raise DeliveryError( + f"Failed to send approval request: {_e}" + ) from _e + + 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): @@ -4814,79 +4924,19 @@ def _approval_notify_sync(approval_data: dict) -> None: # Typing resumes in _handle_approve_command/_handle_deny_command. ctx._status_adapter.pause_typing_for_chat(ctx._status_chat_id) - cmd = approval_data.get("command", "") - desc = approval_data.get("description", "dangerous command") - - # Redact credentials from the command before displaying it in - # the approval prompt — Tirith's findings are already redacted, - # but the raw command string still leaks secrets to the chat - # platform (#48456). Applied here so BOTH the button-based - # (send_exec_approval) and plain-text fallback paths below use - # the redacted value. - cmd = _redact_approval_command(cmd) - - # Prefer button-based approval when the adapter supports it. - # Check the *class* for the method, not the instance — avoids - # false positives from MagicMock auto-attribute creation in tests. - if getattr(type(ctx._status_adapter), "send_exec_approval", None) is not None: - try: - _approval_fut = safe_schedule_threadsafe( - ctx._status_adapter.send_exec_approval( - chat_id=ctx._status_chat_id, - command=cmd, - session_key=_approval_session_key, - description=desc, - metadata=ctx._status_thread_metadata, - allow_permanent=approval_data.get("allow_permanent", True), - allow_session=approval_data.get("allow_session", True), - smart_denied=approval_data.get("smart_denied", False), - ), - ctx._loop_for_step, - logger=logger, - log_message="send_exec_approval scheduling error", - ) - if _approval_fut is None: - raise RuntimeError("send_exec_approval: loop unavailable") - _approval_result = _approval_fut.result(timeout=15) - if _approval_result.success: - return - logger.warning( - "Button-based approval failed (send returned error), falling back to text: %s", - _approval_result.error, - ) - except Exception as _e: - logger.warning( - "Button-based approval failed, falling back to text: %s", _e - ) - - # Fallback: plain text approval prompt. Use the adapter's - # typed prefix so Slack/Matrix users are told the form they - # can actually type (`!approve`) — typed "/" is blocked in - # Slack threads and reserved by Matrix clients. - _p = getattr(ctx._status_adapter, "typed_command_prefix", "/") - msg = _format_exec_approval_fallback( - cmd, - desc, - _p, + _deliver_approval_message( + ctx._status_adapter, + ctx._status_chat_id, + approval_data.get("command", ""), + approval_data.get("description", "dangerous command"), + _approval_session_key, + ctx._status_thread_metadata, + ctx._loop_for_step, + logger, allow_permanent=approval_data.get("allow_permanent", True), allow_session=approval_data.get("allow_session", True), smart_denied=approval_data.get("smart_denied", False), ) - try: - _approval_send_fut = safe_schedule_threadsafe( - ctx._status_adapter.send( - ctx._status_chat_id, - msg, - metadata=ctx._status_thread_metadata, - ), - ctx._loop_for_step, - logger=logger, - log_message="Approval text-send scheduling error", - ) - if _approval_send_fut is not None: - _approval_send_fut.result(timeout=15) - except Exception as _e: - logger.error("Failed to send approval request: %s", _e) # Keep real user text separate from API-only recovery guidance. If # an auto-continue note is prepended below, persist the original diff --git a/tests/gateway/test_approval_context_message.py b/tests/gateway/test_approval_context_message.py new file mode 100644 index 000000000000..16ac8a712fba --- /dev/null +++ b/tests/gateway/test_approval_context_message.py @@ -0,0 +1,445 @@ +"""Tests for ``_deliver_approval_message`` — the production delivery helper. + +All tests call the module-level ``_deliver_approval_message`` directly +with a fake adapter so they exercise the real button / text / redaction / +fail-closed paths without touching ``TurnRunner`` internals. +""" + +import pytest +from unittest.mock import MagicMock + + +# --------------------------------------------------------------------------- +# Fake adapter +# --------------------------------------------------------------------------- + +class _FakeSendResult: + def __init__(self, success: bool, error: str = ""): + self.success = success + self.error = error + + +class FakeButtonAdapter: + """Adapter that offers ``send_exec_approval`` (Discord-style).""" + typed_command_prefix = "/" + + def __init__(self, *, send_result: _FakeSendResult | None = None): + self._send_result = send_result or _FakeSendResult(True) + self.sent_messages: list[str] = [] + self.approval_calls: list[dict] = [] + + async def send_exec_approval(self, *, chat_id, command, session_key, + description, metadata, + allow_permanent, allow_session, + smart_denied): + self.approval_calls.append({ + "command": command, + "description": description, + "allow_permanent": allow_permanent, + "allow_session": allow_session, + "smart_denied": smart_denied, + }) + return self._send_result + + async def send(self, chat_id, message, *, metadata=None): + self.sent_messages.append(message) + + +class FakeTextAdapter: + """Text-only adapter (no ``send_exec_approval``).""" + typed_command_prefix = "!" + + def __init__(self): + self.sent_messages: list[str] = [] + + async def send(self, chat_id, message, *, metadata=None): + self.sent_messages.append(message) + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + +_FAKE_OPENAI = "sk-test-" + "X" * 36 +_FAKE_GHP = "ghp_" + "X" * 36 + +# Explanation with Purpose/Effect/Risk but no credentials (clean). +_CLEAN_EXPLANATION = { + "purpose": "clean deployment target", + "effect": "remove all files", + "risk": "irreversible deletion", +} + +# Enhanced description that includes the unverified model context. +_CONTEXT_PREFIX = "—— Model-provided context (unverified) ——" +_CONTEXT_SUFFIX = "—— End unverified context ——" + +_ENHANCED_DESC = ( + "recursive delete in root path\n\n" + + _CONTEXT_PREFIX + "\n" + "Purpose: clean deployment target\n" + "Effect: remove all files\n" + "Risk: irreversible deletion\n" + + _CONTEXT_SUFFIX +) + + +import asyncio + + +class _SyncFuture: + """Minimal Future stand-in so ``_fut.result(timeout=…)`` works.""" + def __init__(self, value): + self._value = value + + def result(self, timeout=None): + return self._value + + +def _sync_schedule(fn, loop, *, logger=None, log_message=""): + """Run the coroutine *fn* to completion synchronously and return a + ``_SyncFuture`` wrapping its result — used to monkeypatch + ``safe_schedule_threadsafe`` in tests so the production delivery + helper can be called without a running event loop.""" + loop = asyncio.get_event_loop() + return _SyncFuture(loop.run_until_complete(fn)) + + +def _make_deliver_kwargs(adapter, monkeypatch): + """Return kwargs + install the synchronous schedule patch.""" + from logging import getLogger + import gateway.run as gw_run + monkeypatch.setattr(gw_run, "safe_schedule_threadsafe", _sync_schedule) + return { + "adapter": adapter, + "chat_id": "test-chat", + "command": "rm -rf /dangerous", + "description": _ENHANCED_DESC, + "session_key": "test-session", + "metadata": None, + "loop": asyncio.get_event_loop(), + "logger": getLogger("test"), + } + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +class TestButtonPath: + """Tests that exercise the ``send_exec_approval`` (button) path.""" + + def test_description_includes_model_context(self, monkeypatch): + """Button-based approval receives the enhanced description so + Purpose/Effect/Risk appear in the same button card.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + assert len(adapter.approval_calls) == 1 + desc = adapter.approval_calls[0]["description"] + assert _CONTEXT_PREFIX in desc + assert "Purpose: clean deployment target" in desc + assert "Effect: remove all files" in desc + assert "Risk: irreversible deletion" in desc + assert _CONTEXT_SUFFIX in desc + + def test_button_success_sends_no_text_fallback(self, monkeypatch): + """When the button path succeeds, ``send()`` is never called.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter(send_result=_FakeSendResult(True)) + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + assert len(adapter.approval_calls) == 1 + assert len(adapter.sent_messages) == 0, ( + "button success must not fall through to text send()" + ) + + def test_button_failure_falls_through_to_text(self, monkeypatch): + """When the button path fails, the text fallback sends exactly one + message.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter(send_result=_FakeSendResult(False, "timeout")) + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + assert len(adapter.approval_calls) == 1 # tried button + assert len(adapter.sent_messages) == 1, ( + "button failure must fall through to a single text send()" + ) + + +class TestTextPath: + """Tests that exercise the plain-text fallback path.""" + + def test_sends_exactly_once(self, monkeypatch): + """Text-only adapters receive exactly one ``send()`` call.""" + from gateway.run import _deliver_approval_message + + adapter = FakeTextAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + assert len(adapter.sent_messages) == 1 + + def test_message_contains_context_and_approve_instruction(self, monkeypatch): + """The single text message carries the full context AND the + /approve instruction in one payload.""" + from gateway.run import _deliver_approval_message + + adapter = FakeTextAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + msg = adapter.sent_messages[0] + assert _CONTEXT_PREFIX in msg + assert "clean deployment target" in msg + assert "!approve" in msg, ( + "text fallback must include the typed approve instruction" + ) + + def test_no_standalone_followup(self, monkeypatch): + """There is never an independent second message — context is + always co-located with the approval prompt.""" + from gateway.run import _deliver_approval_message + + adapter = FakeTextAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + _deliver_approval_message(**kwargs) + + assert len(adapter.sent_messages) <= 1, ( + "must not send a standalone follow-up message" + ) + + +class TestOutboundRedaction: + """Credentials in model-supplied context must be redacted in the + outbound message, not just upstream.""" + + def test_redacts_in_button_description(self, monkeypatch): + """Button path description must not contain raw credentials.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter() + desc_with_creds = ( + "risky command\n\n" + + _CONTEXT_PREFIX + "\n" + "Purpose: deploy with key " + _FAKE_OPENAI + "\n" + "Risk: exposes " + _FAKE_GHP + "\n" + + _CONTEXT_SUFFIX + ) + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + kwargs["description"] = desc_with_creds + _deliver_approval_message(**kwargs) + + out = adapter.approval_calls[0]["description"] + assert _FAKE_OPENAI not in out, "redact_sensitive_text must strip creds" + assert _FAKE_GHP not in out, "redact_sensitive_text must strip creds" + + def test_redacts_in_text_message(self, monkeypatch): + """Text-fallback message must not contain raw credentials.""" + from gateway.run import _deliver_approval_message + + adapter = FakeTextAdapter() + desc_with_creds = ( + "risky command\n\n" + + _CONTEXT_PREFIX + "\n" + "Purpose: deploy with key " + _FAKE_OPENAI + "\n" + "Risk: exposes " + _FAKE_GHP + "\n" + + _CONTEXT_SUFFIX + ) + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + kwargs["description"] = desc_with_creds + _deliver_approval_message(**kwargs) + + out = adapter.sent_messages[0] + assert _FAKE_OPENAI not in out, "redact_sensitive_text must strip creds" + assert _FAKE_GHP not in out, "redact_sensitive_text must strip creds" + + +class TestFailClosed: + """Malformed input must refuse to deliver any approval prompt.""" + + def test_empty_description_raises(self, monkeypatch): + """An empty or blank description must raise ValueError before any + adapter method is called.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + kwargs["description"] = "" + with pytest.raises(ValueError, match="empty"): + _deliver_approval_message(**kwargs) + + assert len(adapter.approval_calls) == 0 + assert len(adapter.sent_messages) == 0 + + def test_whitespace_only_description_raises(self, monkeypatch): + """Whitespace-only description is treated the same as empty.""" + from gateway.run import _deliver_approval_message + + adapter = FakeButtonAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + kwargs["description"] = " \n " + with pytest.raises(ValueError, match="empty"): + _deliver_approval_message(**kwargs) + + assert len(adapter.approval_calls) == 0 + assert len(adapter.sent_messages) == 0 + + def test_fail_closed_on_text_adapter_too(self, monkeypatch): + """Fail-closed applies to text adapters as well.""" + from gateway.run import _deliver_approval_message + + adapter = FakeTextAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + kwargs["description"] = "" + with pytest.raises(ValueError): + _deliver_approval_message(**kwargs) + + assert len(adapter.sent_messages) == 0 + kwargs["description"] = "" + with pytest.raises(ValueError, match="empty"): + _deliver_approval_message(**kwargs) + + assert len(adapter.sent_messages) == 0 + + +class TestDeliveryError: + """Delivery failure must raise ``DeliveryError``, not silently return.""" + + def test_button_future_none_raises(self, monkeypatch): + """When ``safe_schedule_threadsafe`` returns None for the button + call, ``_deliver_approval_message`` must raise DeliveryError + (delivery status is unknown).""" + from gateway.run import _deliver_approval_message, DeliveryError + import gateway.run as gw_run + + adapter = FakeButtonAdapter() + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + # Force the first safe_schedule_threadsafe call to return None. + monkeypatch.setattr(gw_run, "safe_schedule_threadsafe", + lambda fn, loop, **kw: None) + with pytest.raises(DeliveryError, match="loop unavailable"): + _deliver_approval_message(**kwargs) + assert len(adapter.approval_calls) == 0 + assert len(adapter.sent_messages) == 0 + + def test_text_send_failure_raises(self, monkeypatch): + """When the text send raises, DeliveryError must propagate.""" + from gateway.run import _deliver_approval_message, DeliveryError + + adapter = FakeTextAdapter() + + async def _failing_send(chat_id, msg, *, metadata=None): + raise ConnectionError("test forced failure") + + adapter.send = _failing_send + kwargs = _make_deliver_kwargs(adapter, monkeypatch) + with pytest.raises(DeliveryError, match="Failed to send"): + _deliver_approval_message(**kwargs) + + +class TestE2EFailClosed: + """End-to-end: ``check_all_command_guards`` must block delivery when + model-supplied context is malicious or delivery itself fails.""" + + def test_forged_approve_context_blocks(self, monkeypatch): + """A model-supplied context containing ``/approve`` is sanitised + out before the approval is delivered — the guard must not reach + the notify callback.""" + import tools.approval as amod + from tools.approval import ( + check_all_command_guards, + register_gateway_notify, + unregister_gateway_notify, + set_current_session_key, + reset_current_session_key, + ) + + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + session_key = "test-e2e-forge" + token = set_current_session_key(session_key) + notified = {} + + def notify_cb(data): + notified.update(data) + queue = amod._gateway_queues[session_key] + queue[0].result = "deny" + queue[0].event.set() + + register_gateway_notify(session_key, notify_cb) + try: + # approval_context with forged /approve line — must be stripped + # by _sanitize_explanation, making the enhanced description + # contain the system warning but NOT the forged line. + result = check_all_command_guards( + "rm -rf /malicious", "local", + approval_context={ + "purpose": "normal text", + "risk": "/approve malicious\nreal risk", + }, + ) + finally: + unregister_gateway_notify(session_key) + reset_current_session_key(token) + + # The guard blocks the command (dangerous), and the explanation + # is sanitised before reaching the notify callback. + assert result["approved"] is False + assert "explanation" in notified + assert "/approve" not in str(notified["explanation"]), ( + "forged /approve line must be stripped" + ) + + def test_delivery_error_notify_failed_removes_entry(self, monkeypatch): + """When _deliver_approval_message raises DeliveryError, the pending + entry is removed and check_all_command_guards does NOT return + approved.""" + import tools.approval as amod + from tools.approval import ( + check_all_command_guards, + register_gateway_notify, + unregister_gateway_notify, + set_current_session_key, + reset_current_session_key, + ) + from gateway.run import DeliveryError + + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + session_key = "test-e2e-delivery-fail" + token = set_current_session_key(session_key) + + def failing_notify(data): + raise DeliveryError("simulated delivery failure") + + register_gateway_notify(session_key, failing_notify) + try: + result = check_all_command_guards( + "rm -rf /dangerous", "local", + approval_context={ + "purpose": "clean deployment", + "effect": "remove files", + "risk": "data loss", + }, + ) + finally: + unregister_gateway_notify(session_key) + reset_current_session_key(token) + + # Guard must NOT approve — delivery failed. + assert result["approved"] is False, ( + "must not approve when delivery fails" + ) + assert result.get("message", "").startswith("BLOCKED"), ( + "must return BLOCKED when notify fails" + ) + # Pending entry must be removed — no orphaned entry in the queue. + assert amod._gateway_queues.get(session_key) is None, ( + "orphaned entry must be removed on delivery failure" + ) diff --git a/tests/gateway/test_approval_prompt_redaction.py b/tests/gateway/test_approval_prompt_redaction.py index 695448e1407f..d7bd53790d5a 100644 --- a/tests/gateway/test_approval_prompt_redaction.py +++ b/tests/gateway/test_approval_prompt_redaction.py @@ -111,9 +111,11 @@ def _assert_redacts_then_uses(self, module, func_name: str, sink_substr: str): ) def test_chat_platform_path_redacts_before_send(self): + """_deliver_approval_message must assign and use the redacted + command and description before any send_exec_approval or send().""" import gateway.run as run - self._assert_redacts_then_uses(run, "_approval_notify_sync", "send_exec_approval") + self._assert_redacts_then_uses(run, "_deliver_approval_message", "send_exec_approval") def test_sse_api_path_redacts_before_enqueue(self): from gateway.platforms import api_server diff --git a/tests/tools/test_command_guards.py b/tests/tools/test_command_guards.py index 8d114b608060..55f8a47d9078 100644 --- a/tests/tools/test_command_guards.py +++ b/tests/tools/test_command_guards.py @@ -412,10 +412,175 @@ def notify_cb(data): "risk": "deleted files cannot be recovered", } + # ------------------------------------------------------------------- + # Explanation credential redaction + # ------------------------------------------------------------------- + # Synthetic, scanner-safe credential fixtures. Each matches its + # redactor regex (sk-/AKIA/ghp_) but is unmistakably fake — a run of + # X characters, never a real key. Same pattern used by the existing + # gateway test_approval_prompt_redaction.py. + _FAKE_OPENAI = "sk-test-" + "X" * 36 + _FAKE_AWS = "AKIA" + "X" * 16 + _FAKE_GHP = "ghp_" + "X" * 36 -# --------------------------------------------------------------------------- -# Terminal schema exposes approval context -# --------------------------------------------------------------------------- + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_redact_helper_strips_sk_shapes(self, mock_tirith, monkeypatch): + """redact_sensitive_text helper strips OpenAI ``sk-...`` shapes + from model-supplied approval context values.""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + cb = MagicMock(return_value="once") + result = check_all_command_guards( + "echo safe", + "local", + approval_context={ + "purpose": "test with key " + self._FAKE_OPENAI, + }, + approval_callback=cb, + ) + assert result["approved"] is True # safe cmd, no approval prompt + # But if it were blocked, the explanation must not leak the key. + # Validate the redaction path directly via _clean_approval_context + # plus the redact call in check_all_command_guards by running a + # dangerous command and inspecting the returned description. + from agent.redact import redact_sensitive_text + raw_context = {"purpose": "deploy via " + self._FAKE_OPENAI} + cleaned = approval_module._clean_approval_context(raw_context) + assert self._FAKE_OPENAI in cleaned["purpose"], \ + "precondition: raw credential survives _clean_approval_context" + redacted = redact_sensitive_text(cleaned["purpose"]) + assert self._FAKE_OPENAI not in redacted, \ + "redact_sensitive_text must strip sk- shapes" + assert "deploy via" in redacted, \ + "non-credential text must survive redaction" + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_redact_helper_strips_aws_ghp_shapes(self, mock_tirith): + """AWS ``AKIA...`` and GitHub ``ghp_...`` shapes are redacted.""" + from agent.redact import redact_sensitive_text + raw_context = { + "purpose": "use " + self._FAKE_AWS, + "risk": "exposes " + self._FAKE_GHP, + } + cleaned = approval_module._clean_approval_context(raw_context) + assert self._FAKE_AWS in cleaned["purpose"], "precondition" + assert self._FAKE_GHP in cleaned["risk"], "precondition" + # Simulate the redaction step done inside check_all_command_guards. + redacted_purpose = redact_sensitive_text(cleaned["purpose"]) + redacted_risk = redact_sensitive_text(cleaned["risk"]) + assert self._FAKE_AWS not in redacted_purpose + assert self._FAKE_GHP not in redacted_risk + assert "use" in redacted_purpose + assert "exposes" in redacted_risk + + @patch(_TIRITH_PATCH, return_value=_tirith_result("warn", [], + "git reset destructive")) + def test_inbound_notify_payload_redacts_credentials(self, mock_tirith, monkeypatch): + """Inbound notify payload: the callback receives an ``explanation`` + from which credential-shaped strings have been redacted by + check_all_command_guards (first layer, before the defense-in-depth + re-redact in _deliver_approval_message).""" + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + session_key = "test-redact-session" + token = set_current_session_key(session_key) + notified = {} + + def notify_cb(data): + notified.update(data) + queue = approval_module._gateway_queues[session_key] + queue[0].result = "deny" + queue[0].event.set() + + approval_module.register_gateway_notify(session_key, notify_cb) + try: + result = check_all_command_guards( + "git reset --hard origin/main", + "local", + approval_context={ + "purpose": "reset via " + self._FAKE_OPENAI, + "risk": "may expose " + self._FAKE_GHP, + }, + ) + finally: + approval_module.unregister_gateway_notify(session_key) + reset_current_session_key(token) + + assert result["approved"] is False + # Gateway notify callback received the explanation — it must not + # contain the raw credential that was in the model-supplied context. + explanation = notified.get("explanation") or {} + assert "purpose" in explanation + assert self._FAKE_OPENAI not in explanation.get("purpose", "") + assert self._FAKE_GHP not in explanation.get("risk", "") + # Non-credential fragments survive redaction. + assert "reset via" in explanation["purpose"] + assert "may expose" in explanation["risk"] + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_explanation_bound_to_approval_request(self, mock_tirith, monkeypatch): + """The ``explanation`` is NOT a loose follow-up message — it is + bound to the same approval payload as command, description, and + pattern_key. It only appears when approval is required; a safe + command with context must NOT leak explanation into tool output.""" + monkeypatch.setenv("HERMES_GATEWAY_SESSION", "1") + session_key = "test-bound-session" + token = set_current_session_key(session_key) + notified = {} + + def notify_cb(data): + notified.update(data) + queue = approval_module._gateway_queues[session_key] + queue[0].result = "deny" + queue[0].event.set() + + approval_module.register_gateway_notify(session_key, notify_cb) + try: + result = check_all_command_guards( + "rm -rf /important", # dangerous → triggers approval + "local", + approval_context={ + "purpose": "clean deployment target", + "effect": "remove all files", + "risk": "irreversible deletion", + }, + ) + finally: + approval_module.unregister_gateway_notify(session_key) + reset_current_session_key(token) + + assert result["approved"] is False + # All four payload fields must be present together in the same + # approval notification — explanation is NOT a separate message. + assert notified.get("command") + assert notified.get("description") + assert notified.get("pattern_key") + assert notified.get("explanation") + # Verify explanation content is structured, not just a dict stub. + assert notified["explanation"]["purpose"] == "clean deployment target" + assert notified["explanation"]["effect"] == "remove all files" + assert notified["explanation"]["risk"] == "irreversible deletion" + + @patch(_TIRITH_PATCH, return_value=_tirith_result("allow")) + def test_safe_command_with_context_does_not_leak_explanation( + self, mock_tirith, monkeypatch): + """A safe command with ``approval_context`` must NOT surface + explanation in any output — the ``explanation`` field only + exists inside the approval data, not in the tool return value.""" + monkeypatch.setenv("HERMES_INTERACTIVE", "1") + cb = MagicMock(return_value="once") + result = check_all_command_guards( + "echo safe operation", + "local", + approval_context={ + "purpose": "verify shell works", + "effect": "prints text", + "risk": "none", + }, + approval_callback=cb, + ) + # Safe command returns approved without any approval_data + assert result["approved"] is True + assert "explanation" not in result + assert "purpose" not in str(result) def test_terminal_schema_exposes_approval_context_fields(): from tools.terminal_tool import TERMINAL_SCHEMA diff --git a/tools/approval.py b/tools/approval.py index 768f82e4e4b6..843af51618d3 100644 --- a/tools/approval.py +++ b/tools/approval.py @@ -2422,7 +2422,8 @@ def prompt_dangerous_approval(command: str, description: str, timeout_seconds: int | None = None, allow_permanent: bool = True, approval_callback=None, - *, smart_denied: bool = False) -> str: + *, smart_denied: bool = False, + explanation: dict | None = None) -> str: """Prompt the user to approve a dangerous command (CLI only). Args: @@ -2436,6 +2437,9 @@ def prompt_dangerous_approval(command: str, description: str, (command, description, *, allow_permanent=True, smart_denied=False) -> str. Legacy callback signatures remain supported when ``smart_denied`` is false. + explanation: Optional model-supplied dict with 'purpose'/'effect'/ + 'risk' keys explaining why the command is needed. Displayed to + the user before the approve/deny choice. Returns: 'once', 'session', 'always', or 'deny' """ @@ -2498,6 +2502,17 @@ def prompt_dangerous_approval(command: str, description: str, print() print(f" {t('approval.dangerous_header', description=display_description)}") print(f" {display_command}") + if explanation: + print(" —— Model-provided context (unverified) ——") + for _key, _label in ( + ("purpose", "Purpose"), + ("effect", "Effect"), + ("risk", "Risk"), + ): + _val = str(explanation.get(_key, "")).strip() + if _val: + print(f" {_label}: {redact_sensitive_text(_val)}") + print(" —— End unverified context ——") print() if smart_denied: print(t("approval.choose_smart_deny")) @@ -3389,6 +3404,83 @@ def _approval_context_or_fallback(approval_context: dict | None) -> dict: return _clean_approval_context(approval_context) +_FORGE_RE = re.compile(r'^[/!](approve|deny)|^(⚠|⚠️)', re.IGNORECASE) + + +def _sanitize_explanation(explanation: dict | None) -> dict: + """Sanitize model-supplied approval context before it reaches any user surface. + + Applies in order: credential redaction, control-character stripping, + newline normalisation, forged-command-line removal, per-field and total + length caps. Returns an empty dict when the input is empty or invalid. + """ + if not isinstance(explanation, dict) or not explanation: + return {} + from agent.redact import redact_sensitive_text + + cleaned: dict[str, str] = {} + total = 0 + for key in ("purpose", "effect", "risk"): + value = str(explanation.get(key, "")).strip() + if not value: + continue + value = redact_sensitive_text(value, force=True) + value = re.sub(r"[\x00-\x1f\x7f]", "", value) + value = value.replace("\r\n", "\n").replace("\r", "\n") + # Drop lines that try to forge approve/deny commands or approval + # headings so a model cannot hijack the approval-instruction surface. + safe_lines = [ + ln for ln in value.split("\n") if not _FORGE_RE.match(ln) + ] + value = "\n".join(safe_lines) + if len(value) > 1000: + value = value[:1000] + if total + len(value) > 3000: + value = value[:max(0, 3000 - total)] + if value: + cleaned[key] = value + total += len(value) + return cleaned + + +def _build_enhanced_description_with_context( + system_desc: str, + explanation: dict | None, +) -> str: + """Combine the system-level risk description with sanitised model-supplied + purpose/effect/risk context into a single description string suitable for + every approval surface (gateway button, text fallback, CLI prompt). + + Raises ``ValueError`` when the result would be empty — callers MUST + refuse to deliver an insufficient approval prompt (fail-closed). + """ + parts = [system_desc.strip()] + + if isinstance(explanation, dict) and explanation: + ctx = [] + for key, label in ( + ("purpose", "Purpose"), + ("effect", "Effect"), + ("risk", "Risk"), + ): + val = str(explanation.get(key, "")).strip() + if val: + ctx.append(f"{label}: {val}") + if ctx: + parts.append("") + parts.append("—— Model-provided context (unverified) ——") + parts.extend(ctx) + parts.append("—— End unverified context ——") + + result = "\n".join(parts) + if not result or not result.strip(): + raise ValueError( + "Approval description is empty — refusing to deliver " + "an insufficient approval prompt." + ) + return result + + def check_all_command_guards(command: str, env_type: str, approval_callback=None, has_host_access: bool = False, @@ -3646,14 +3738,13 @@ def check_all_command_guards(command: str, env_type: str, # Combine descriptions for a single approval prompt combined_desc = "; ".join(desc for _, desc, _ in warnings) approval_explanation = _approval_context_or_fallback(approval_context) - # Approval output is a secret-egress boundary: model-supplied context is - # displayed verbatim to the user, so redact credential-shaped strings the - # same way the command and description are redacted before display. - if approval_explanation: - from agent.redact import redact_sensitive_text as _redact_explanation - approval_explanation = { - k: _redact_explanation(v) for k, v in approval_explanation.items() - } + # Approval output is a secret-egress boundary: sanitise model-supplied + # context and build a single enhanced description that every surface + # (gateway button, text fallback, CLI prompt) consumes directly. + approval_explanation = _sanitize_explanation(approval_explanation) + enhanced_desc = _build_enhanced_description_with_context( + combined_desc, approval_explanation, + ) primary_key = warnings[0][0] all_keys = [key for key, _, _ in warnings] # "Always" is offered when at least one warning is a dangerous-pattern @@ -3692,7 +3783,7 @@ def check_all_command_guards(command: str, env_type: str, "command": redact_sensitive_text(command), "pattern_key": primary_key, "pattern_keys": all_keys, - "description": redact_sensitive_text(combined_desc), + "description": redact_sensitive_text(enhanced_desc), "explanation": approval_explanation, # Smart DENY overrides are one-operation decisions, so the UI # must not offer a permanent scope. Otherwise offer Always @@ -3715,7 +3806,7 @@ def check_all_command_guards(command: str, env_type: str, "approved": False, "message": "BLOCKED: Failed to send approval request to user. Do NOT retry.", "pattern_key": primary_key, - "description": combined_desc, + "description": enhanced_desc, } resolved = decision["resolved"] choice = decision["choice"] @@ -3754,7 +3845,7 @@ def check_all_command_guards(command: str, env_type: str, f"irreversible action.{timeout_addendum}{breaker_addendum}" ), "pattern_key": primary_key, - "description": combined_desc, + "description": enhanced_desc, "outcome": outcome, "user_consent": False, "deny_reason": deny_reason, @@ -3776,7 +3867,7 @@ def check_all_command_guards(command: str, env_type: str, # smart-DENY owner override) resets the consecutive-denial tally. _reset_denials(session_key) return {"approved": True, "message": None, - "user_approved": True, "description": combined_desc} + "user_approved": True, "description": enhanced_desc} # Fallback: no gateway callback registered (e.g. cron, batch). # Return approval_required for backward compat. Redact secrets in the @@ -3784,7 +3875,7 @@ def check_all_command_guards(command: str, env_type: str, # the allowlist keys off pattern_key, so redaction is display-only. from agent.redact import redact_sensitive_text _disp_command = redact_sensitive_text(command) - _disp_combined_desc = redact_sensitive_text(combined_desc) + _disp_combined_desc = redact_sensitive_text(enhanced_desc) pending_data = { "command": _disp_command, "pattern_key": primary_key, @@ -3815,7 +3906,7 @@ def check_all_command_guards(command: str, env_type: str, _fire_approval_hook( "pre_approval_request", command=command, - description=combined_desc, + description=enhanced_desc, pattern_key=primary_key, pattern_keys=list(all_keys), session_key=session_key, @@ -3823,15 +3914,16 @@ def check_all_command_guards(command: str, env_type: str, ) choice = prompt_dangerous_approval( command, - combined_desc, + enhanced_desc, allow_permanent=has_permanent_capable and not smart_denied_for_owner, smart_denied=smart_denied_for_owner, approval_callback=approval_callback, + explanation=approval_explanation, ) _fire_approval_hook( "post_approval_response", command=command, - description=combined_desc, + description=enhanced_desc, pattern_key=primary_key, pattern_keys=list(all_keys), session_key=session_key,