<!-- source: https://deepwiki.com/qhkm/zeptoclaw/3.3-message-bus-and-routing -->

# Message Bus and Routing

- Cargo.lock
- Cargo.toml
- src/bus/message.rs
- src/bus/mod.rs
- src/channels/factory.rs
- src/channels/mod.rs
- src/cli/channel.rs
- src/session/mod.rs
- src/session/types.rs

## Purpose and Scope

The `MessageBus` is the central nervous system of ZeptoClaw, implementing a pub/sub pattern that decouples communication channels from agent processing. This page documents the `MessageBus` architecture, the core message types (`InboundMessage` and `OutboundMessage`), routing mechanics, and the channel registration flow.

The bus facilitates asynchronous communication between the **Channel Layer** (responsible for external I/O) and the **Agent Layer** (responsible for LLM orchestration and tool execution).

## Architecture Overview

The `MessageBus` enables loose coupling between channels (Telegram, Discord, Slack, etc.) and the agent loop. Channels publish incoming user messages as `InboundMessage` and consume agent responses as `OutboundMessage`. The agent loop does the inverse: it consumes `InboundMessage` for processing and publishes `OutboundMessage` for delivery.

### MessageBus in the System

```

```

**Sources:** [src/bus/mod.rs63-72](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L63-L72) [src/channels/mod.rs9-27](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/mod.rs#L9-L27) [src/channels/factory.rs23-163](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/factory.rs#L23-L163)

## Message Types

### InboundMessage

Standardized format for messages flowing **into** the agent from external sources. Each `InboundMessage` represents a user prompt or system-triggered task.

**Structure:**

- channel: String — Source channel name (e.g., "telegram" , "discord" , "cron" ) src/bus/message.rs 13
- sender_id: String — Unique identifier of the sender src/bus/message.rs 15
- chat_id: String — Unique identifier of the chat/conversation src/bus/message.rs 17
- content: String — The text content of the message src/bus/message.rs 19
- media: Vec<MediaAttachment> — Optional images, audio, or documents src/bus/message.rs 21
- session_key: String — Routing key formatted as channel:chat_id src/bus/message.rs 23
- metadata: HashMap<String, String> — Additional key-value pairs src/bus/message.rs 25

**Construction:**
The `InboundMessage::new` function automatically generates the `session_key` used by the `SessionManager` [src/bus/message.rs90-100](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L90-L100)

**Sources:** [src/bus/message.rs11-26](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L11-L26) [src/bus/message.rs90-100](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L90-L100)

### OutboundMessage

Standardized format for messages flowing **out** from the agent to channels for delivery.

**Structure:**

- channel: String — The target channel to send through src/bus/message.rs 32
- chat_id: String — The target chat/conversation src/bus/message.rs 34
- content: String — The text content to send src/bus/message.rs 36
- reply_to: Option<String> — Optional message ID to reply to src/bus/message.rs 38
- metadata: HashMap<String, String> — Channel-specific delivery hints src/bus/message.rs 41

**Usage:**
Agents typically use `OutboundMessage::reply_to(&inbound, content)` to ensure the response is routed back to the correct source [src/bus/message.rs196-199](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L196-L199)

**Sources:** [src/bus/message.rs30-42](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L30-L42) [src/bus/message.rs189-200](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L189-L200)

## Message Flow Patterns

### Inbound Path: Channel → Bus → Agent

The inbound path uses a Multi-Producer, Single-Consumer (MPSC) queue. Multiple channels can concurrently push messages to a single `AgentLoop`.

```

```

**Key Points:**

- Channels call MessageBus::publish_inbound() which returns a Result src/bus/mod.rs 134-139
- The MessageBus holds the receiver inbound_rx behind a Mutex to allow shared access across async tasks src/bus/mod.rs 70
- DEFAULT_BUFFER_SIZE is set to 100 messages to provide backpressure if the agent is overwhelmed src/bus/mod.rs 53

**Sources:** [src/bus/mod.rs53](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L53-L53) [src/bus/mod.rs134-139](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L134-L139) [src/bus/mod.rs167-169](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L167-L169)

### Outbound Path: Agent → Bus → Channels

The outbound path also utilizes an MPSC channel. While the agent publishes messages, the `ChannelManager` or individual channel tasks consume them to perform delivery.

```

```

**Key Points:**

- Agents publish responses via publish_outbound src/bus/mod.rs 193-198
- Channels implement the Channel trait, which requires a send(&self, msg: OutboundMessage) method src/channels/mod.rs 148
- The MessageBus handles the routing logic by ensuring messages are available for consumption by the registered channel handlers src/bus/mod.rs 208-210

**Sources:** [src/bus/mod.rs193-198](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L193-L198) [src/bus/mod.rs208-210](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L208-L210) [src/channels/mod.rs148](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/mod.rs#L148-L148)

## Implementation Details

### MessageBus Structure

The `MessageBus` encapsulates the Tokio MPSC channels and provides a thread-safe interface using `Arc` and `Mutex`.

```

```

- MPSC (Multi-Producer, Single-Consumer): Used for both directions to manage flow control and buffering src/bus/mod.rs 63-72
- Shared Ownership: The Arc<Mutex<...>> wrapping of receivers allows multiple parts of the system to wait for messages without race conditions src/bus/mod.rs 106-108

**Sources:** [src/bus/mod.rs63-72](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L63-L72) [src/bus/mod.rs100-110](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L100-L110)

## Channel Registration

### ChannelManager Integration

The `ChannelManager` is responsible for the lifecycle of all communication channels. It requires a reference to the `MessageBus` during initialization to pass it to the channels it manages.

```

```

**Sources:** [src/channels/manager.rs118](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/manager.rs#L118-L118) [src/channels/factory.rs23-27](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/factory.rs#L23-L27) [src/channels/factory.rs34-46](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/factory.rs#L34-L46)

### Factory Pattern

The `register_configured_channels` function (found in `src/channels/factory.rs`) acts as a factory that:

1. Iterates through the Config src/channels/factory.rs 26
2. Checks if a channel (e.g., Telegram) is enabled src/channels/factory.rs 30
3. Instantiates the channel with the shared MessageBus src/channels/factory.rs 35-37
4. Registers it with the ChannelManager via manager.register(Box::new(...)) src/channels/factory.rs 34

**Sources:** [src/channels/factory.rs23-50](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/factory.rs#L23-L50) [src/channels/factory.rs53-67](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/factory.rs#L53-L67)

## Channel Trait and Interface

Every channel must implement the `Channel` trait, which defines the standard interface for the `MessageBus` and `ChannelManager` to interact with.

**Sources:** [src/channels/mod.rs148](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/mod.rs#L148-L148)

## Summary of Data Flow

**Sources:** [src/bus/mod.rs134-210](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/mod.rs#L134-L210) [src/bus/message.rs11-42](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/bus/message.rs#L11-L42) [src/channels/mod.rs148](https://github.com/qhkm/zeptoclaw/blob/2c743d31/src/channels/mod.rs#L148-L148)
