How ApprovalRuntime Manages Pending Approvals and Streams Them Over the Wire in Kimi CLI

The ApprovalRuntime class stores pending requests in an internal UUID-keyed dictionary and publishes them as JSON-RPC events through the RootWireHub, enabling real-time UI display and asynchronous coordination.

Kimi CLI's approval system centers on the ApprovalRuntime class defined in src/kimi_cli/approval_runtime/runtime.py. This component acts as the central coordinator for user-driven approval flows, bridging tool execution requests with the terminal UI via a wire protocol. Understanding how ApprovalRuntime manages pending approvals and displays them on the wire stream reveals the async architecture that allows multiple callers to await the same user decision.

Storing and Tracking Pending Approvals

At the heart of the approval system lies a robust registry that maintains the state of every pending request.

The Internal Request Registry

When a tool initiates an approval flow, it calls create_request() on the ApprovalRuntime instance. This method constructs an ApprovalRequestRecord dataclass (defined in src/kimi_cli/approval_runtime/models.py) and stores it in the internal _requests dictionary, keyed by the request's UUID.

# Create a new approval request from a tool

request = approval_runtime.create_request(
    sender="my_tool",
    action="delete_file",
    description="Delete /tmp/secret.txt?",
    tool_call_id=tool_call.id,
    display=[...],                     # UI‑ready blocks

    source=ApprovalSource(
        kind="tool",
        id="my_tool",
        agent_id=None,
        subagent_type=None,
    ),
)

The storage structure maps each request ID to its complete record:

# From src/kimi_cli/approval_runtime/runtime.py

_requests: dict[str, ApprovalRequestRecord]

This dictionary tracks the full lifecycle of a request, from creation through resolution or cancellation. Each record contains metadata including the sender, action description, tool call ID, and display blocks ready for UI rendering.

Listing Pending Requests

The runtime exposes list_pending() to return a chronological view of all active requests. This method filters the _requests dictionary for entries where status == "pending" and returns them sorted by time, allowing UI components to render approval queues or dashboards.

Coordinating Concurrent Waiters

Tools often need to block execution until a user makes a decision. The runtime handles this through asyncio primitives that support multiple observers of the same request.

Asyncio Futures for Blocking Operations

The wait_for_response() method creates a shared asyncio.Future for each unique request ID, stored in _waiters: dict[str, asyncio.Future]. When a caller awaits this future, execution pauses until the user approves or rejects the request.

# Somewhere else, block until the user decides

response_kind, feedback = await approval_runtime.wait_for_response(request.id, timeout=30)

if response_kind == "approve":
    # proceed with the operation

    ...
else:
    # abort or handle rejection

    ...

Upon resolution via resolve(), the runtime fulfills the future with the response kind and any feedback text, unblocking all waiting coroutines simultaneously.

Reference Counting with _waiter_counts

To properly manage memory and lifecycle events, ApprovalRuntime tracks how many callers are waiting on each request through _waiter_counts: dict[str, int]. When the last waiter detaches or the request resolves, the runtime cleans up the associated future and count entries, preventing memory leaks in long-running sessions.

Publishing Approval Events to the Wire Stream

The wire stream represents the communication channel between the CLI backend and the terminal frontend. ApprovalRuntime publishes structured events to make pending approvals visible to the user.

RootWireHub Integration

After create_request() stores a new record, the runtime immediately calls _publish_wire_request(). This method pushes an ApprovalRequest message onto the RootWireHub if one is bound to the runtime. Similarly, when resolve() or cancel_by_source() completes, _publish_wire_response() transmits an ApprovalResponse message.

These messages conform to Pydantic models defined in src/kimi_cli/wire/types.py, ensuring type safety across the wire protocol.

Request and Response Message Types

The wire protocol distinguishes between incoming requests and outgoing responses:

• ApprovalRequest: Contains the UUID, sender identity, action description, and display blocks for UI rendering
• ApprovalResponse: Carries the request ID, decision ("approve" or "reject"), and optional feedback text

Both messages traverse the RootWireHub as typed events, decoupling the runtime from specific transport implementations.

WireServer: Bridging Runtime and UI

While ApprovalRuntime produces events, the WireServer in src/kimi_cli/wire/server.py handles the actual network transmission to the client interface.

Subscription to RootWireHub

WireServer subscribes to the RootWireHub during initialization. Its internal _root_hub_loop continuously monitors for new messages, creating the bridge between the runtime's internal state and the external JSON-RPC wire protocol.

JSON-RPC Event Forwarding

When _root_hub_loop receives an ApprovalRequest, the server wraps it in a JSONRPCEventMessage and forwards it to the client UI as an event notification. The UI renders the request—typically displaying the sender, description, and approval controls.

# WireServer side – forwarding a request to the client UI

if isinstance(msg, ApprovalRequest):
    # The request appears as a JSON‑RPC event to the UI

    await self._send_msg(JSONRPCEventMessage(method="event", params=msg))

When the user selects approve or reject, the client sends an ApprovalResponse back through the wire. The WireServer receives this response and routes it to the ApprovalRuntime, which calls resolve() to update the internal state and fulfill waiting futures.

# UI (shell) receives the event and renders it

def handle_event(event: ApprovalRequest) -> None:
    console.print(f"[bold]{event.sender}[/] asks: {event.description}")
    # User selects approve/reject → send ApprovalResponse back

    send_wire_message(ApprovalResponse(
        request_id=event.id,
        response="approve",
        feedback="Looks safe",
    ))

Request Lifecycle and Cancellation

The runtime supports bulk cancellation through cancel_by_source(), which walks the _requests dictionary to find pending requests matching specific source criteria (such as a particular tool or agent ID). For each match, it marks the request as cancelled, publishes a cancellation response on the wire, and raises CancelledError in any waiting futures.

This ensures that when a tool or agent terminates unexpectedly, its pending approvals don't remain orphaned in the queue, and waiting coroutines receive immediate notification of the cancellation.

Summary

• Storage: ApprovalRuntime maintains pending approvals in _requests, a UUID-keyed dictionary of ApprovalRequestRecord objects defined in src/kimi_cli/approval_runtime/models.py.
• Coordination: Async callers await decisions through shared asyncio.Future objects tracked in _waiters, with reference counting via _waiter_counts to manage cleanup.
• Wire Publishing: The runtime pushes ApprovalRequest and ApprovalResponse messages to the RootWireHub through _publish_wire_request() and _publish_wire_response().
• Server Bridge: WireServer subscribes to the hub in src/kimi_cli/wire/server.py and forwards these messages as JSON-RPC events over the wire stream to the terminal UI.
• Lifecycle: cancel_by_source() enables bulk cancellation of pending requests, propagating cancellation signals to both the wire and waiting callers.

Frequently Asked Questions

How does ApprovalRuntime handle multiple tools waiting for the same approval?

When multiple callers invoke wait_for_response() for the same request ID, ApprovalRuntime creates a single shared asyncio.Future stored in _waiters. The _waiter_counts dictionary tracks how many callers reference this future. When the user resolves the request, ApprovalRuntime fulfills the future once, and all awaiting callers receive the same response simultaneously.

What happens to pending approvals if the user closes the terminal?

If the terminal closes, the WireServer connection drops, but the ApprovalRuntime persists in memory if the process continues running. However, tools awaiting responses via wait_for_response() would hang until the specified timeout expires. The runtime's cancel_by_source() method can be invoked during shutdown sequences to mark all pending requests as cancelled and unblock waiting coroutines with CancelledError.

Where are the wire protocol message types defined?

The Pydantic models for wire communication are defined in src/kimi_cli/wire/types.py. This module exports ApprovalRequest and ApprovalResponse classes that specify the exact schema for data transmitted between the ApprovalRuntime and the UI through the RootWireHub and WireServer.

Can I query pending approvals without blocking execution?

Yes. Instead of calling wait_for_response(), which blocks until resolution, use the list_pending() method on the ApprovalRuntime instance. This returns a time-sorted list of all ApprovalRequestRecord objects where status == "pending", allowing non-blocking inspection of the current approval queue for dashboard or monitoring purposes.

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