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

> Discover how ApprovalRuntime manages pending approvals using a UUID-keyed dictionary and streams them as JSON-RPC events via RootWireHub for real-time Kimi CLI display. Learn about asynchronous coordination.

- Repository: [Moonshot AI/kimi-cli](https://github.com/MoonshotAI/kimi-cli)
- Tags: internals
- Published: 2026-07-22

---

**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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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`](https://github.com/MoonshotAI/kimi-cli/blob/main/src/kimi_cli/approval_runtime/models.py)) and stores it in the internal `_requests` dictionary, keyed by the request's UUID.

```python

# 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:

```python

# 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.

```python

# 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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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.

```python

# 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.

```python

# 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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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`](https://github.com/MoonshotAI/kimi-cli/blob/main/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.