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

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

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:

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:

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:

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:

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:

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:

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:

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

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