# How AionUi's ChannelMessageService Handles Streaming Responses with 500ms Throttling for IM Updates

> Discover how AionUi's ChannelMessageService streams IM updates with 500ms throttling. Learn about its callback system, tool-call tracking, and robust delivery guarantees via pending timer cleanup.

- Repository: [OfficeAI/AionUi](https://github.com/iofficeai/aionui)
- Tags: internals
- Published: 2026-02-19

---

**AionUi's ChannelMessageService manages streaming AI responses by registering per-conversation callbacks that buffer chunks and track tool-call turns, while ActionExecutor enforces a 500ms throttle (`UPDATE_THROTTLE_MS`) on IM platform edits to prevent rate limiting, guaranteeing the final update is always delivered via pending timer cleanup.**

AionUi, an open-source AI interface framework maintained by iOfficeAI, delivers real-time agent responses to enterprise messaging platforms like Lark, Telegram, and DingTalk through a coordinated streaming architecture. The system relies on `ChannelMessageService` to orchestrate the stream lifecycle while delegating platform-specific rate limiting to `ActionExecutor`. This implementation ensures that high-frequency AI token generation never overwhelms IM platform APIs, while strict state management guarantees message integrity across multi-turn tool interactions.

## Core Architecture: ChannelMessageService and ActionExecutor

The streaming pipeline separates concerns between agent-side event management and gateway-side platform throttling. In [`src/channels/agent/ChannelMessageService.ts`](https://github.com/iOfficeAI/AionUi/blob/main/src/channels/agent/ChannelMessageService.ts), the service registers a global listener on `ChannelEventBus` to intercept agent events (`start`, `update`, `finish`) and maintains per-conversation state in an `activeStreams` Map. Each stream entry tracks the callback function, message buffer, and resolution promise.

Conversely, [`src/channels/gateway/ActionExecutor.ts`](https://github.com/iOfficeAI/AionUi/blob/main/src/channels/gateway/ActionExecutor.ts) consumes these streams and handles the platform-specific outgoing messages. This is where the **500ms throttling logic** resides, preventing edit request flooding to IM platforms that enforce strict rate limits on message updates.

## Stream Lifecycle and Turn Tracking

When `ChannelMessageService.sendMessage()` initiates a conversation, it creates a stream state object stored by `conversationId`:

```typescript
this.activeStreams.set(conversationId, {
    msgId,
    callback: onStream,
    buffer: '',
    resolve,
    reject,
    turnCount: 0,
    finishCount: 0,
});

```

The `handleAgentMessage()` method processes three critical event types:

- **`start`** events increment `turnCount`, tracking when the agent invokes external tools
- **`finish`** events increment `finishCount`, signaling tool execution completion
- **Message events** are transformed via `transformMessage()` and composed into the existing message list

The stream only resolves when `finishCount >= turnCount` (or if no turns were initiated), ensuring all tool-call "turns" complete before closing the connection:

```typescript
if (event.type === 'finish') {
    stream.finishCount++;
    if (stream.turnCount === 0 || stream.finishCount >= stream.turnCount) {
        this.activeStreams.delete(conversationId);
        stream.resolve(stream.msgId);
    }
    return;
}

```

## 500ms Throttling Implementation for IM Updates

The throttling mechanism lives entirely within `ActionExecutor` to respect IM platform constraints. The implementation uses a sliding window approach with explicit cleanup guarantees.

### Throttle Constants and State

The executor initializes throttling state variables at the start of each streaming operation:

```typescript
let lastUpdateTime = 0;
const UPDATE_THROTTLE_MS = 500;
let pendingUpdateTimer: ReturnType<typeof setTimeout> | null = null;
let pendingMessage: IUnifiedOutgoingMessage | null = null;

```

### Handling Stream Chunks with Time Windows

When the `ChannelMessageService` invokes the streaming callback with `isInsert` flags, `ActionExecutor` applies different logic for initial versus subsequent chunks. For the **first insert** (updating the "thinking" placeholder) and **all updates**, the system checks the time elapsed since `lastUpdateTime`:

```typescript
const now = Date.now();
pendingMessage = streamOutgoing;

if (now - lastUpdateTime >= UPDATE_THROTTLE_MS) {
    // Immediate edit outside throttle window
    await doEditMessage(streamOutgoing);
} else {
    // Schedule delayed edit within throttle window
    const delay = UPDATE_THROTTLE_MS - (now - lastUpdateTime);
    pendingUpdateTimer = setTimeout(() => {
        if (pendingMessage) {
            void doEditMessage(pendingMessage);
            pendingMessage = null;
        }
        pendingUpdateTimer = null;
    }, delay);
}

```

The `doEditMessage` helper updates `lastUpdateTime` and performs the actual platform edit:

```typescript
const doEditMessage = async (msg: IUnifiedOutgoingMessage) => {
    lastUpdateTime = Date.now();
    const targetMsgId = sentMessageIds.at(-1) ?? thinkingMsgId;
    await context.editMessage(targetMsgId, msg);
};

```

### Guaranteed Final Update Delivery

When the stream promise resolves (indicating all turns completed), `ActionExecutor` performs mandatory cleanup to ensure no content is lost to throttling:

```typescript
if (pendingUpdateTimer) {
    clearTimeout(pendingUpdateTimer);
    pendingUpdateTimer = null;
}
if (pendingMessage) {
    await doEditMessage(pendingMessage); // Forces final content delivery
}

```

After flushing pending updates, the executor appends platform-specific action buttons (reply markup) to the finalized message:

```typescript
const responseMarkup = getResponseActionsMarkup(context.platform as PluginType, lastMessageContent?.text);
await context.editMessage(lastMsgId, {
    ...lastMessageContent,
    replyMarkup: responseMarkup,
});

```

## Practical Code Implementation

To implement this flow, `ActionExecutor` registers a callback with `ChannelMessageService` that handles the throttling logic:

```typescript
await messageService.sendMessage(
    sessionId,
    conversationId,
    text,
    async (message: TMessage, isInsert: boolean) => {
        const now = Date.now();
        const outgoing = convertTMessageToOutgoing(message, context.platform as PluginType, false);
        const streamOutgoing = { ...outgoing, replyMarkup: undefined };
        
        // Throttling logic applied here based on isInsert flag and timing
    }
);

```

This architecture allows `ChannelMessageService` to remain agnostic of platform rate limits while `ActionExecutor` enforces the 500ms constraint required by enterprise IM platforms.

## Summary

- **ChannelMessageService** in [`src/channels/agent/ChannelMessageService.ts`](https://github.com/iOfficeAI/AionUi/blob/main/src/channels/agent/ChannelMessageService.ts) maintains per-conversation state using an `activeStreams` Map with `turnCount` and `finishCount` counters to manage multi-turn tool-call lifecycles
- **ActionExecutor** in [`src/channels/gateway/ActionExecutor.ts`](https://github.com/iOfficeAI/AionUi/blob/main/src/channels/gateway/ActionExecutor.ts) implements the `UPDATE_THROTTLE_MS = 500` constant to restrict IM edit requests to one per 500-millisecond window
- Pending message updates are buffered in `pendingMessage` and scheduled via `setTimeout` when chunks arrive within the throttle window, preventing API rate limit violations
- Stream resolution guarantees final content delivery by clearing `pendingUpdateTimer` and unconditionally invoking `doEditMessage(pendingMessage)` before closing the stream
- The first streamed chunk updates the initial "thinking" placeholder message, while subsequent chunks create new messages to show progressive generation without chat history clutter

## Frequently Asked Questions

### What is the purpose of turnCount and finishCount in ChannelMessageService?

These counters enable robust handling of multi-turn agent interactions involving external tools. When the agent initiates a tool call, the `start` event increments `turnCount`; when the tool returns, the `finish` event increments `finishCount`. The stream only resolves when `finishCount >= turnCount`, ensuring that `ChannelMessageService` waits for all parallel tool executions to complete before signaling completion to the UI layer.

### Why does ActionExecutor use 500ms specifically for throttling IM updates?

The 500-millisecond value (`UPDATE_THROTTLE_MS`) represents a conservative balance between real-time user experience and the rate limits imposed by enterprise messaging platforms like Lark, DingTalk, and Telegram. These platforms typically restrict message edit operations to prevent spam and API abuse; the 500ms window ensures AionUi remains compliant with platform policies while maintaining responsive, visible streaming progress.

### How does AionUi guarantee the final message content is always delivered despite throttling?

When the streaming promise resolves—indicating all agent turns have finished—`ActionExecutor` executes a mandatory cleanup routine. It first clears any active `pendingUpdateTimer` to prevent race conditions, then checks if `pendingMessage` contains buffered content. If so, it immediately calls `doEditMessage(pendingMessage)` to force the final update, ensuring the last accumulated content reaches the IM platform even if it arrived within the 500ms throttle window.

### What happens to the "thinking" placeholder message when streaming begins?

The first token chunk received from the agent (`isInsert === true` when `sentMessageIds.length === 1`) triggers an **edit** operation on the initial "thinking" placeholder rather than creating a new message. This approach provides immediate visual feedback that generation has started. Subsequent chunks create separate messages via `context.sendMessage()`, allowing users to see the response build progressively without generating excessive edit history on the placeholder message.