feat: surface command context in approval prompts

gateway/platforms/api_server.py

@@ -4500,6 +4500,12 @@ def _approval_notify(approval_data: Dict[str, Any]) -> None:
                         from gateway.run import _redact_approval_command
 
                         event["command"] = _redact_approval_command(event.get("command"))
+                    if "explanation" in event:
+                        from gateway.run import _redact_approval_explanation
+
+                        event["explanation"] = _redact_approval_explanation(
+                            event.get("explanation")
+                        )
                     event.update({
                         "event": "approval.request",
                         "run_id": run_id,

gateway/run.py

@@ -353,6 +353,21 @@ def _redact_approval_command(cmd: "str | None") -> str:
     return redact_sensitive_text(str(cmd or ""), force=True)
 
 
+def _redact_approval_explanation(explanation: object) -> dict:
+    """Force-redact model-supplied approval context at the display boundary."""
+    if not isinstance(explanation, dict):
+        return {}
+
+    from agent.redact import redact_sensitive_text
+
+    redacted = {}
+    for key in ("purpose", "effect", "risk"):
+        value = explanation.get(key)
+        if isinstance(value, str) and value.strip():
+            redacted[key] = redact_sensitive_text(value.strip(), force=True)
+    return redacted
+
+
 def _format_exec_approval_fallback(
     command: str,
     description: str,
@@ -18878,6 +18893,36 @@ def _approval_notify_sync(approval_data: dict) -> None:
 
                 cmd = approval_data.get("command", "")
                 desc = approval_data.get("description", "dangerous command")
+                explanation = _redact_approval_explanation(
+                    approval_data.get("explanation")
+                )
+                purpose = explanation.get("purpose")
+                effect = explanation.get("effect")
+                risk = explanation.get("risk")
+                followup_msg = ""
+                if purpose or effect or risk:
+                    followup_msg = (
+                        "Command approval context:\n\n"
+                        f"Purpose: {purpose or 'Not provided'}\n\n"
+                        f"Effect: {effect or 'Not provided'}\n\n"
+                        f"Risk: {risk or 'Not provided'}"
+                    )
+
+                def _send_approval_context_followup() -> None:
+                    if not followup_msg:
+                        return
+                    _followup_fut = safe_schedule_threadsafe(
+                        _status_adapter.send(
+                            _status_chat_id,
+                            followup_msg,
+                            metadata=_status_thread_metadata,
+                        ),
+                        _loop_for_step,
+                        logger=logger,
+                        log_message="Approval context-send scheduling error",
+                    )
+                    if _followup_fut is not None:
+                        _followup_fut.result(timeout=15)
 
                 # Redact credentials from the command before displaying it in
                 # the approval prompt — Tirith's findings are already redacted,
@@ -18910,6 +18955,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
                             raise RuntimeError("send_exec_approval: loop unavailable")
                         _approval_result = _approval_fut.result(timeout=15)
                         if _approval_result.success:
+                            _send_approval_context_followup()
                             return
                         logger.warning(
                             "Button-based approval failed (send returned error), falling back to text: %s",
@@ -18945,6 +18991,7 @@ def _approval_notify_sync(approval_data: dict) -> None:
                     )
                     if _approval_send_fut is not None:
                         _approval_send_fut.result(timeout=15)
+                        _send_approval_context_followup()
                 except Exception as _e:
                     logger.error("Failed to send approval request: %s", _e)
 

tests/gateway/test_approval_prompt_redaction.py

@@ -17,7 +17,7 @@
 or real-looking key, so secret scanners do not flag this file.
 """
 
-from gateway.run import _redact_approval_command
+from gateway.run import _redact_approval_command, _redact_approval_explanation
 
 # Synthetic, scanner-safe credential fixtures. Each matches its redactor
 # regex (ghp_/sk-/JWT) but is unmistakably fake -- a run of X's, never a
@@ -67,6 +67,33 @@ def test_handles_none_and_empty(self):
         assert _redact_approval_command(None) == ""
 
 
+class TestRedactApprovalExplanation:
+    """Approval explanations are an equally strict secret-egress boundary."""
+
+    def test_forces_redaction_for_each_context_field(self, monkeypatch):
+        monkeypatch.setattr("agent.redact._REDACT_ENABLED", False, raising=False)
+        explanation = {
+            "purpose": "Call GitHub with " + _FAKE_GHP,
+            "effect": "Export OPENAI_API_KEY=" + _FAKE_OPENAI,
+            "risk": "Bearer " + _FAKE_JWT + " may be logged",
+        }
+
+        out = _redact_approval_explanation(explanation)
+
+        for credential in (_FAKE_GHP, _FAKE_OPENAI, _FAKE_JWT):
+            assert credential not in " ".join(out.values())
+        assert set(out) == {"purpose", "effect", "risk"}
+
+    def test_ignores_unknown_or_invalid_values(self):
+        assert _redact_approval_explanation(None) == {}
+        assert _redact_approval_explanation("raw") == {}
+        assert _redact_approval_explanation({
+            "purpose": " deploy ",
+            "effect": 123,
+            "unknown": "do not forward",
+        }) == {"purpose": "deploy"}
+
+
 class TestApprovalCommandWiring:
     """Guard the production wiring on BOTH approval-notify transports:
     1. the chat-platform path (_approval_notify_sync in gateway/run.py), and
@@ -77,7 +104,10 @@ class TestApprovalCommandWiring:
     benign refactor doesn't cause a false failure, and so a discarded-result
     call (`_redact(cmd); send(cmd)`) does NOT pass."""
 
-    def _assert_redacts_then_uses(self, module, func_name: str, sink_substr: str):
+    def _assert_redacts_then_uses(
+        self, module, func_name: str, sink_substr: str,
+        redactor: str = "_redact_approval_command",
+    ):
         """Parse `module`'s full AST, locate the (possibly nested) function
         `func_name`, and assert it contains an assignment
         `<x> = _redact_approval_command(...)` whose result is then used by a
@@ -100,10 +130,10 @@ def _assert_redacts_then_uses(self, module, func_name: str, sink_substr: str):
         for node in ast.walk(target_fn):
             if isinstance(node, ast.Assign) and isinstance(node.value, ast.Call):
                 fn = node.value.func
-                if isinstance(fn, ast.Name) and fn.id == "_redact_approval_command":
+                if isinstance(fn, ast.Name) and fn.id == redactor:
                     redact_line = node.lineno
         assert redact_line is not None, (
-            f"{func_name} must assign the result of _redact_approval_command(...) "
+            f"{func_name} must assign the result of {redactor}(...) "
             "(a discarded-result call would still leak the raw command)"
         )
 
@@ -127,6 +157,26 @@ def test_sse_api_path_redacts_before_enqueue(self):
 
         self._assert_redacts_then_uses(api_server, "_approval_notify", "put_nowait")
 
+    def test_chat_platform_path_redacts_explanation_before_send(self):
+        import gateway.run as run
+
+        self._assert_redacts_then_uses(
+            run,
+            "_approval_notify_sync",
+            "_status_adapter.send",
+            redactor="_redact_approval_explanation",
+        )
+
+    def test_sse_api_path_redacts_explanation_before_enqueue(self):
+        from gateway.platforms import api_server
+
+        self._assert_redacts_then_uses(
+            api_server,
+            "_approval_notify",
+            "put_nowait",
+            redactor="_redact_approval_explanation",
+        )
+
     def test_chat_platform_threads_approval_capabilities_to_adapter(self):
         """The gateway must not drop the backend's one-operation UI contract."""
         import ast

tests/gateway/test_tui_approval_redaction.py

@@ -47,6 +47,31 @@ def test_emit_approval_request_handles_missing_command(self, monkeypatch):
         tui_server._emit_approval_request("s", None)
         assert emitted["payload"] == {}
 
+    def test_emit_approval_request_force_redacts_explanation(self, monkeypatch):
+        from tui_gateway import server as tui_server
+
+        emitted = {}
+        monkeypatch.setattr(
+            tui_server, "_emit",
+            lambda event, sid, payload=None: emitted.update({"payload": payload}),
+        )
+        monkeypatch.setattr("agent.redact._REDACT_ENABLED", False, raising=False)
+        fake_credential = "sk-proj-" + "X" * 40
+
+        tui_server._emit_approval_request(
+            "s",
+            {
+                "explanation": {
+                    "purpose": "Use " + fake_credential,
+                    "effect": "No credential here",
+                }
+            },
+        )
+
+        explanation = emitted["payload"]["explanation"]
+        assert fake_credential not in explanation["purpose"]
+        assert explanation["effect"] == "No credential here"
+
     @pytest.mark.parametrize(
         ("data", "expected"),
         [

tests/tools/test_command_guards.py

@@ -37,6 +37,8 @@ def _tirith_result(action="allow", findings=None, summary=""):
 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 = {}
@@ -45,6 +47,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():
@@ -345,9 +349,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):

tools/approval.py

@@ -2517,6 +2517,33 @@ def _format_tirith_description(tirith_result: dict) -> str:
     return "Security scan — " + "; ".join(parts)
 
 
+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 _await_gateway_decision(session_key: str, notify_cb, approval_data: dict,
                             *, surface: str = "gateway") -> dict:
     """Enqueue *approval_data*, notify the user, and block the calling agent
@@ -2634,14 +2661,18 @@ def _drop_entry() -> None:
 
 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
     presents them as a single combined approval request. This prevents
     a gateway force=True replay from bypassing one check when only the
     other was shown to the user.
 
+    ``approval_context`` is optional model-supplied context explaining why the
+    command is being run. It is only surfaced when approval is required.
+
     ``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.
@@ -2876,6 +2907,7 @@ 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)
     primary_key = warnings[0][0]
     all_keys = [key for key, _, _ in warnings]
     has_tirith = any(is_t for _, _, is_t in warnings)
@@ -2907,6 +2939,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.
                 "allow_permanent": not has_tirith and not smart_denied_for_owner,
@@ -2992,6 +3025,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)