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
1 change: 1 addition & 0 deletions src/kimi_cli/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -255,6 +255,7 @@ 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)

return KimiCLI(soul, runtime, env_overrides)

Expand Down
1 change: 1 addition & 0 deletions src/kimi_cli/hooks/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
"PreCompact",
"PostCompact",
"Notification",
"PermissionRequest",
]

HOOK_EVENT_TYPES: list[str] = list(HookEventType.__args__) # type: ignore[attr-defined]
Expand Down
20 changes: 20 additions & 0 deletions src/kimi_cli/hooks/events.py
Original file line number Diff line number Diff line change
Expand Up @@ -188,3 +188,23 @@ def notification(
"body": body,
"severity": severity,
}


def permission_request(
*,
session_id: str,
cwd: str,
tool_name: str,
tool_input: dict[str, Any],
tool_call_id: str = "",
action: str = "",
description: str = "",
) -> dict[str, Any]:
return {
**_base("PermissionRequest", session_id, cwd),
"tool_name": tool_name,
"tool_input": tool_input,
"tool_call_id": tool_call_id,
"action": action,
"description": description,
}
24 changes: 21 additions & 3 deletions src/kimi_cli/hooks/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
class HookResult:
"""Result of a single hook execution."""

action: Literal["allow", "block"] = "allow"
action: Literal["allow", "block", "ask"] = "allow"
reason: str = ""
stdout: str = ""
stderr: str = ""
Expand Down Expand Up @@ -75,10 +75,28 @@ async def run_hook(
if isinstance(raw, dict):
parsed = cast(dict[str, Any], raw)
hook_output = cast(dict[str, Any], parsed.get("hookSpecificOutput", {}))
if hook_output.get("permissionDecision") == "deny":
decision = hook_output.get("permissionDecision") or hook_output.get("decision")
if decision == "deny":
return HookResult(
action="block",
reason=str(hook_output.get("permissionDecisionReason", "")),
reason=str(hook_output.get("permissionDecisionReason", hook_output.get("reason", ""))),
stdout=stdout,
stderr=stderr,
exit_code=0,
)
elif decision == "ask":
# Explicitly ask user in terminal
return HookResult(
action="ask",
reason=str(hook_output.get("reason", "")),
stdout=stdout,
stderr=stderr,
exit_code=0,
)
elif decision == "allow":
return HookResult(
action="allow",
reason=str(hook_output.get("reason", "")),
stdout=stdout,
stderr=stderr,
exit_code=0,
Expand Down
76 changes: 74 additions & 2 deletions src/kimi_cli/soul/approval.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import uuid
from collections.abc import Callable
from typing import Literal
from typing import TYPE_CHECKING, Literal

from kimi_cli.approval_runtime import (
ApprovalCancelledError,
Expand All @@ -15,6 +15,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"]


Expand Down Expand Up @@ -76,17 +79,22 @@ def __init__(
*,
state: ApprovalState | None = None,
runtime: ApprovalRuntime | None = None,
hook_engine: HookEngine | None = None,
):
self._state = state or ApprovalState(yolo=yolo)
self._runtime = runtime or ApprovalRuntime()
self._hook_engine = hook_engine

def share(self) -> Approval:
"""Create a new approval queue that shares state (yolo + auto-approve)."""
return Approval(state=self._state, runtime=self._runtime)
return Approval(state=self._state, runtime=self._runtime, hook_engine=self._hook_engine)

def set_runtime(self, runtime: ApprovalRuntime) -> None:
self._runtime = runtime

def set_hook_engine(self, hook_engine: HookEngine) -> None:
self._hook_engine = hook_engine

@property
def runtime(self) -> ApprovalRuntime:
return self._runtime
Expand Down Expand Up @@ -138,6 +146,70 @@ async def request(
if action in self._state.auto_approve_actions:
return ApprovalResult(approved=True)

# --- PermissionRequest hook ---
if self._hook_engine is not None:
from pathlib import Path

from kimi_cli.hooks import events
from kimi_cli.soul.toolset import _get_session_id

# Build tool_input from action and description
tool_input: dict = {"action": action, "description": description}

hook_results = await self._hook_engine.trigger(
"PermissionRequest",
matcher_value=tool_call.function.name,
input_data=events.permission_request(
session_id=_get_session_id(), # Use actual CLI session id
cwd=str(Path.cwd()),
tool_name=tool_call.function.name,
tool_input=tool_input,
tool_call_id=tool_call.id,
Comment on lines +159 to +167
Copy link

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

permission_request() is being built with session_id=tool_call.id, but other hook events use the CLI session id (see soul/toolset.py:_get_session_id()). Using tool_call.id here makes PermissionRequest events inconsistent and harder to correlate across a session; consider using the runtime session id / ContextVar session id and keep tool_call_id for the tool call identifier.

Copilot uses AI. Check for mistakes.
action=action,
description=description,
),
)

# Aggregate results: block takes priority, then allow, then ask
# Timeouts/errors result in "ask" (fall back to terminal approval)
has_explicit_allow = False
allow_reason = ""
has_block = False
block_reason = ""

for result in hook_results:
if result.action == "block":
# Block takes highest priority - check action, not reason
has_block = True
block_reason = result.reason
break
elif result.action == "allow" and not result.timed_out and result.exit_code == 0:
# Allow only if:
# - no block found
# - not timed out (timed-out hooks return action="allow")
# - exit_code is 0 (errors/crashes return action="allow" with non-zero exit)
has_explicit_allow = True
allow_reason = result.reason
# result.action == "ask" or timeout/error/crash: continue to check other results

if has_block:
logger.debug(
"PermissionRequest hook denied {tool_name}: {reason}",
tool_name=tool_call.function.name,
reason=block_reason,
)
return ApprovalResult(
approved=False,
feedback=block_reason or "Denied by PermissionRequest hook",
)
elif has_explicit_allow:
logger.debug(
"PermissionRequest hook allowed {tool_name}",
tool_name=tool_call.function.name,
)
return ApprovalResult(approved=True)
# If no explicit decision or only "ask"/timeouts, continue to terminal approval

Comment on lines +180 to +212
Copy link

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When multiple PermissionRequest hooks match, this loop returns on the first "allow" result, which can bypass a later "block" result depending on execution order. Deny/block should take precedence over allow, so consider aggregating all hook results first (e.g., if any block -> reject; elif any allow -> approve; else -> ask/terminal).

Suggested change
for result in hook_results:
if result.action == "allow":
# Hook explicitly allows - skip terminal approval
logger.debug(
"PermissionRequest hook allowed {tool_name}",
tool_name=tool_call.function.name,
)
return ApprovalResult(approved=True)
elif result.action == "block":
# Hook explicitly denies
logger.debug(
"PermissionRequest hook denied {tool_name}: {reason}",
tool_name=tool_call.function.name,
reason=result.reason,
)
return ApprovalResult(
approved=False,
feedback=result.reason or "Denied by PermissionRequest hook",
)
# result.action == "ask" means continue to terminal approval
has_allow = False
block_reason: str | None = None
for result in hook_results:
if result.action == "block":
# Hook explicitly denies; deny takes precedence over allow
if block_reason is None:
block_reason = result.reason or "Denied by PermissionRequest hook"
elif result.action == "allow":
# Hook explicitly allows - approve only if no hook blocks
has_allow = True
# result.action == "ask" means continue evaluating other hooks
if block_reason is not None:
logger.debug(
"PermissionRequest hook denied {tool_name}: {reason}",
tool_name=tool_call.function.name,
reason=block_reason,
)
return ApprovalResult(
approved=False,
feedback=block_reason,
)
if has_allow:
logger.debug(
"PermissionRequest hook allowed {tool_name}",
tool_name=tool_call.function.name,
)
return ApprovalResult(approved=True)

Copilot uses AI. Check for mistakes.
request_id = str(uuid.uuid4())
display_blocks = display or []
source = get_current_approval_source_or_none() or ApprovalSource(
Expand Down