diff --git a/src/kimi_cli/app.py b/src/kimi_cli/app.py index a3d4d0a87a..8fa1eecbed 100644 --- a/src/kimi_cli/app.py +++ b/src/kimi_cli/app.py @@ -325,6 +325,11 @@ async def create( hook_engine = HookEngine(config.hooks, cwd=str(session.work_dir)) soul.set_hook_engine(hook_engine) runtime.hook_engine = hook_engine + runtime.approval.set_hook_engine( + hook_engine, + session_id=session.id, + cwd=str(session.work_dir), + ) # --- Initialize telemetry --- from kimi_cli.telemetry import attach_sink, set_context diff --git a/src/kimi_cli/soul/approval.py b/src/kimi_cli/soul/approval.py index 23c1b144d4..30b71e73ae 100644 --- a/src/kimi_cli/soul/approval.py +++ b/src/kimi_cli/soul/approval.py @@ -3,7 +3,7 @@ import time import uuid from collections.abc import Callable -from typing import Any, Literal +from typing import TYPE_CHECKING, Any, Literal from kimi_cli.approval_runtime import ( ApprovalCancelledError, @@ -16,6 +16,9 @@ from kimi_cli.utils.logging import logger from kimi_cli.wire.types import DisplayBlock +if TYPE_CHECKING: + from kimi_cli.hooks.engine import HookEngine + type Response = Literal["approve", "approve_for_session", "reject"] # Maps DisplayBlock.type to the TS approval_surface vocabulary. @@ -137,14 +140,30 @@ def __init__( ): self._state = state or ApprovalState(yolo=yolo) self._runtime = runtime or ApprovalRuntime() + self._hook_engine: HookEngine | None = None + self._hook_session_id = "" + self._hook_cwd = "" def share(self) -> Approval: """Create a new approval queue that shares approval state.""" - return Approval(state=self._state, runtime=self._runtime) + shared = Approval(state=self._state, runtime=self._runtime) + if self._hook_engine is not None: + shared.set_hook_engine( + self._hook_engine, + session_id=self._hook_session_id, + cwd=self._hook_cwd, + ) + return shared def set_runtime(self, runtime: ApprovalRuntime) -> None: self._runtime = runtime + def set_hook_engine(self, engine: HookEngine, *, session_id: str, cwd: str) -> None: + """Configure hooks emitted when a request needs human approval.""" + self._hook_engine = engine + self._hook_session_id = session_id + self._hook_cwd = cwd + @property def runtime(self) -> ApprovalRuntime: return self._runtime @@ -285,6 +304,21 @@ def _elapsed_ms() -> int: kind="foreground_turn", id=tool_call.id, ) + if self._hook_engine is not None: + from kimi_cli.hooks import events + + self._hook_engine.fire_and_forget_trigger( + "Notification", + matcher_value="permission_prompt", + input_data=events.notification( + session_id=self._hook_session_id, + cwd=self._hook_cwd, + sink="shell", + notification_type="permission_prompt", + title=f"{sender} requires approval", + body=description, + ), + ) self._runtime.create_request( request_id=request_id, tool_call_id=tool_call.id, diff --git a/tests/core/test_approval_telemetry.py b/tests/core/test_approval_telemetry.py index 11d373efaf..a366e64b48 100644 --- a/tests/core/test_approval_telemetry.py +++ b/tests/core/test_approval_telemetry.py @@ -1,7 +1,7 @@ """Telemetry parity tests for the permission_approval_result event (TS alignment).""" import asyncio -from unittest.mock import patch +from unittest.mock import MagicMock, patch import pytest from kosong.message import ToolCall @@ -18,6 +18,52 @@ def _permission_events(mock_track) -> list: return [c for c in mock_track.call_args_list if c[0][0] == "permission_approval_result"] +@pytest.mark.asyncio +async def test_manual_approval_triggers_permission_prompt_notification_hook() -> None: + approval = Approval() + hook_engine = MagicMock() + approval.set_hook_engine(hook_engine, session_id="session-1", cwd="/work") + token = current_tool_call.set(_tool_call()) + try: + request_task = asyncio.create_task(approval.request("Bash", "bash:ls", "Run command: ls")) + await asyncio.sleep(0) + pending = approval.runtime.list_pending() + assert len(pending) == 1 + + hook_engine.fire_and_forget_trigger.assert_called_once_with( + "Notification", + matcher_value="permission_prompt", + input_data={ + "hook_event_name": "Notification", + "session_id": "session-1", + "cwd": "/work", + "sink": "shell", + "notification_type": "permission_prompt", + "title": "Bash requires approval", + "body": "Run command: ls", + "severity": "info", + }, + ) + approval.runtime.resolve(pending[0].id, "approve") + assert await request_task + finally: + current_tool_call.reset(token) + + +@pytest.mark.asyncio +async def test_auto_approval_does_not_trigger_permission_prompt_hook() -> None: + approval = Approval(yolo=True) + hook_engine = MagicMock() + approval.set_hook_engine(hook_engine, session_id="session-1", cwd="/work") + token = current_tool_call.set(_tool_call()) + try: + assert await approval.request("Bash", "bash:ls", "Run command: ls") + finally: + current_tool_call.reset(token) + + hook_engine.fire_and_forget_trigger.assert_not_called() + + @pytest.mark.asyncio async def test_yolo_auto_approve_emits_permission_result(): approval = Approval(yolo=True)