fix(hooks): notify on permission prompts

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

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,

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)