How the Kimi CLI Approval Runtime Works: Structure, Lifecycle, and Request Projection

The Kimi CLI approval runtime uses the ApprovalRuntime class in src/kimi_cli/approval_runtime/runtime.py to maintain an in-memory ledger of requests, coordinate asynchronous waiters, and project approval events onto the wire protocol for UI rendering.

The MoonshotAI/kimi-cli repository implements a sophisticated approval system that pauses privileged tool execution until explicitly authorized. Understanding the structure of this approval runtime and how requests are projected to the UI layer is essential for developers extending the CLI or building custom integrations.

Core Components of the Approval Runtime

The approval system centers on four primary data structures defined across the runtime and wire protocol modules:

• ApprovalRuntime (src/kimi_cli/approval_runtime/runtime.py) – The main orchestrator that holds the _requests dictionary, manages asyncio.Future waiters in _waiters, tracks _waiter_counts, and maintains a reference to the RootWireHub for message projection.

• ApprovalRequestRecord (src/kimi_cli/approval_runtime/models.py) – A dataclass storing complete metadata for each request including id, tool_call_id, sender, action, description, display blocks, source, timestamps, status, and the final response.

• ApprovalSource (src/kimi_cli/approval_runtime/models.py) – An enumeration distinguishing request origins as either foreground_turn (direct user interaction) or background_agent (autonomous agent execution), enabling source-aware cancellation.

• ApprovalRuntimeEvent (src/kimi_cli/approval_runtime/models.py) – A simple wrapper carrying request_created or request_resolved events to subscriber callbacks.

• Wire Protocol Types (src/kimi_cli/wire/types.py) – Defines ApprovalRequest and ApprovalResponse message schemas that serialize internal state for transmission over the RootWireHub.

The Approval Request Lifecycle

The runtime manages requests through five distinct phases, each implemented as specific methods on the ApprovalRuntime class.

1. Creation and Event Publication

When a tool requires authorization, it calls ApprovalRuntime.create_request() to instantiate an ApprovalRequestRecord. The runtime immediately stores this in _requests and triggers two simultaneous actions:

• Publishes a request_created event to all subscribers via _publish_event()
• Serializes the request into a wire ApprovalRequest message via _publish_wire_request() for UI consumption

2. Asynchronous Waiting

Consumers awaiting user input call await ApprovalRuntime.wait_for_response(request_id, timeout). The runtime creates a shared asyncio.Future in _waiters for that request ID. Multiple coroutines can await the same future, and the runtime tracks active waiter counts in _waiter_counts to determine when the last observer has timed out.

3. Resolution

Upon user decision, the UI invokes ApprovalRuntime.resolve() with the request ID, response type (approve, approve_for_session, or reject), and optional feedback. The runtime:

1. Updates the record status to resolved
2. Stores the response and feedback
3. Fulfills the future in _waiters, releasing all waiting coroutines
4. Fires a request_resolved event
5. Emits a wire ApprovalResponse via _publish_wire_response()

4. Source-Aware Cancellation

If a background agent terminates or a UI session ends, cancel_by_source() iterates through pending records matching that ApprovalSource. It marks matching requests as cancelled, raises ApprovalCancelledError for active waiters, and publishes resolution events with a reject response.

5. Event Subscription

External components register for lifecycle updates via ApprovalRuntime.subscribe(callback). The runtime invokes these callbacks with ApprovalRuntimeEvent objects whenever requests are created or resolved, enabling logging, telemetry, or custom UI reactions without modifying core logic.

How Approval Requests Are Projected

Projection refers to the transformation of internal Python objects into protocol messages that travel over the wire to UI components. This occurs through two private helper methods in src/kimi_cli/approval_runtime/runtime.py:

_publish_wire_request(request) maps the ApprovalRequestRecord fields into a wire.types.ApprovalRequest message containing the ID, tool call ID, sender, action, description, display blocks, and source identifiers. This message routes through the RootWireHub to reach shell UIs or ACP interfaces.

_publish_wire_response(request_id, response, feedback) constructs a wire.types.ApprovalResponse carrying the final decision and optional feedback string. This signals the UI to clear the prompt and update its state.

The UI layer listens for these wire messages, renders the appropriate prompt (e.g., "Delete /tmp/secret.txt?"), and calls back into ApprovalRuntime.resolve() when the user responds, completing the request loop.

Working with the Approval Runtime

The following examples demonstrate common patterns for interacting with the approval system:

from kimi_cli.approval_runtime.runtime import ApprovalRuntime
from kimi_cli.approval_runtime.models import ApprovalSource

# Initialize runtime with reference to the wire hub

approval_runtime = ApprovalRuntime(wire_hub=root_hub)

# Create a request from a privileged tool

request = approval_runtime.create_request(
    sender="file_manager",
    action="delete_file",
    description="Delete /tmp/secret.txt?",
    tool_call_id="tool-42",
    display=[{"type": "text", "value": "Are you sure?"}],
    source=ApprovalSource(kind="foreground_turn", id="turn-1")
)

# Await user decision with timeout handling

from kimi_cli.approval_runtime.errors import ApprovalCancelledError

try:
    response, feedback = await approval_runtime.wait_for_response(
        request.id, 
        timeout=30
    )
    if response == "approve":
        # Execute the privileged operation

        perform_deletion()
except ApprovalCancelledError:
    # Handle cancellation from UI close or agent termination

    logger.info("Request was cancelled")

# Manual resolution for background agents

approval_runtime.resolve(
    request_id=request.id,
    response="reject",
    feedback="User policy violation",
    approved_via_session_cache=False,
)

# Subscribe to lifecycle events for logging

def log_approval_event(event):
    logger.info(f"Approval {event.kind}: {event.request.id}")

token = approval_runtime.subscribe(log_approval_event)

# Later: approval_runtime.unsubscribe(token)

Summary

• Single Source of Truth: All approval state lives in the ApprovalRuntime instance; wire messages are read-only projections.
• Async Coordination: The runtime uses shared asyncio.Future objects in _waiters to allow multiple coroutines to await the same user decision without duplication.
• Source Tracking: ApprovalSource enables bulk cancellation via cancel_by_source() when background agents or UI sessions terminate.
• Event-Driven Architecture: Subscribers receive ApprovalRuntimeEvent notifications for request_created and request_resolved events, supporting extensible side effects like analytics.
• Wire Projection: Private methods _publish_wire_request() and _publish_wire_response() serialize internal records into wire.types messages for UI rendering.

Frequently Asked Questions

What is the difference between ApprovalRequestRecord and the wire ApprovalRequest?

ApprovalRequestRecord is the internal Python dataclass stored in ApprovalRuntime._requests that tracks the full state of a request including timestamps, status, and the final response. The wire ApprovalRequest (defined in src/kimi_cli/wire/types.py) is a serialized message format containing only the fields necessary for UI rendering, such as the description and display blocks. The _publish_wire_request() method transforms the internal record into the wire format for transmission over the RootWireHub.

How does the runtime handle multiple components waiting for the same approval?

The runtime maintains a dictionary of _waiters where each request ID maps to a shared asyncio.Future. When multiple coroutines call wait_for_response() with the same ID, they all receive a reference to the same future object. The _waiter_counts dictionary tracks how many components are awaiting each request, allowing the runtime to clean up resources only when the last waiter has either received a response or timed out.

What happens when a background agent terminates while requests are pending?

The runtime provides cancel_by_source(), which accepts an ApprovalSource parameter identifying the terminated agent. The method iterates through all pending requests in _requests, matches those with the specified source, marks them as cancelled, and raises ApprovalCancelledError for any active futures. It also publishes request_resolved events and wire responses to ensure UIs clear the pending prompts.

Where does the UI send responses back to the runtime?

UI components receive wire ApprovalRequest messages through the RootWireHub (implemented in src/kimi_cli/wire/root_hub.py). When the user makes a decision, the UI calls ApprovalRuntime.resolve() directly with the request ID, response type (approve, approve_for_session, or reject), and optional feedback. This method updates the internal record and fulfills the waiting future, completing the request lifecycle without requiring additional wire protocol messages for the response path.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}