# Message Bus

> Inter-agent message bus with channels, queues, routing, reliability, and multiple communication patterns

- Skill: `jrennie99-glitch/message-bus` (Agent Skill)
- Install (CLI): `npx skillmds add jrennie99-glitch/message-bus`
- Raw SKILL.md: https://api.skillmd.com/api/skills/jrennie99-glitch/message-bus/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: jrennie99-glitch (https://skillmd.com/u/jrennie99-glitch)
- Updated: 2026-08-19
- Page: https://skillmd.com/skills/jrennie99-glitch/message-bus

---


# Message Bus — Inter-Agent Communication System

## Purpose

The Message Bus provides an advanced messaging and communication layer for swarm coordination. It supports multiple communication patterns (direct, broadcast, multicast, topic-based, queue-based), message routing, reliability guarantees (at-most-once, at-least-once, exactly-once), message filtering, priority queuing, dead letter handling, and comprehensive metrics.

## Configuration

```typescript
import { MessageBus } from './communication/message-bus';

const bus = new MessageBus({
  strategy: 'event-driven',       // Communication strategy
  enablePersistence: true,         // Persist messages to disk
  enableReliability: true,         // Enable delivery guarantees
  enableOrdering: false,           // Ordered message delivery
  enableFiltering: true,           // Message filtering
  maxMessageSize: 1048576,         // 1MB max per message
  maxQueueSize: 10000,             // Max queue depth
  messageRetention: 86400000,      // 24 hours
  acknowledgmentTimeout: 30000,    // 30 seconds
  retryAttempts: 3,
  backoffMultiplier: 2,
  compressionEnabled: false,
  encryptionEnabled: false,
  metricsEnabled: true,
  debugMode: false,
}, logger, eventBus);

await bus.initialize();
```

## Message Structure

```typescript
interface Message {
  id: string;                      // Unique message ID (msg-*)
  type: string;                    // Message type/topic
  sender: AgentId;                 // Sending agent
  receivers: AgentId[];            // Target agent(s)
  content: any;                    // Message payload
  metadata: MessageMetadata;       // Routing, compression, encryption info
  timestamp: Date;
  expiresAt?: Date;                // TTL-based expiration
  priority: 'low' | 'normal' | 'high' | 'critical';
  reliability: 'best-effort' | 'at-least-once' | 'exactly-once';
}
```

### Message Metadata

```typescript
interface MessageMetadata {
  correlationId?: string;          // For request-reply correlation
  causationId?: string;            // Causal chain tracking
  replyTo?: string;                // Reply channel
  ttl?: number;                    // Time-to-live in ms
  compressed: boolean;
  encrypted: boolean;
  size: number;
  contentType: string;
  encoding: string;
  checksum?: string;
  route?: string[];                // Routing path
  deadLetterReason?: string;       // Why message was dead-lettered
}
```

## Sending Messages

```typescript
const messageId = await bus.sendMessage(
  'task:assigned',                 // type
  { taskId: 't-1', description: 'Build feature' },  // content
  senderAgentId,                   // sender
  [receiverAgentId],               // receivers (single or array)
  {
    priority: 'high',
    reliability: 'at-least-once',
    ttl: 60000,                    // 1 minute TTL
    correlationId: 'req-123',
    replyTo: 'response-channel',
    channel: 'task-channel',
  }
);
```

## Channels

Channels are communication pathways between agents:

```typescript
interface MessageChannel {
  id: string;
  name: string;
  type: 'direct' | 'broadcast' | 'multicast' | 'topic' | 'queue';
  participants: AgentId[];
  config: ChannelConfig;
  statistics: ChannelStatistics;
  filters: MessageFilter[];
  middleware: ChannelMiddleware[];
}
```

### Channel Types

| Type       | Description                                    |
|-----------|------------------------------------------------|
| `direct`   | Point-to-point between two agents              |
| `broadcast`| One-to-all agents in the channel               |
| `multicast`| One-to-selected subset of agents               |
| `topic`    | Pub/sub based on topic matching                |
| `queue`    | Competing consumers with load distribution     |

### Channel Configuration

```typescript
interface ChannelConfig {
  persistent: boolean;
  ordered: boolean;
  reliable: boolean;
  maxParticipants: number;
  maxMessageSize: number;
  maxQueueDepth: number;
  retentionPeriod: number;
  accessControl: {
    readPermission: 'public' | 'participants' | 'restricted';
    writePermission: 'public' | 'participants' | 'restricted';
    adminPermission: 'creator' | 'administrators' | 'system';
    allowedSenders: AgentId[];
    allowedReceivers: AgentId[];
    bannedAgents: AgentId[];
  };
}
```

## Message Queues

```typescript
interface MessageQueue {
  id: string;
  name: string;
  type: 'fifo' | 'lifo' | 'priority' | 'delay' | 'round-robin';
  config: {
    maxSize: number;
    persistent: boolean;
    ordered: boolean;
    durability: 'memory' | 'disk' | 'distributed';
    deliveryMode: 'at-most-once' | 'at-least-once' | 'exactly-once';
    deadLetterQueue?: string;
    retryPolicy: RetryPolicy;
  };
  subscribers: QueueSubscriber[];
}
```

### Queue Subscribers

```typescript
interface QueueSubscriber {
  id: string;
  agent: AgentId;
  filter?: MessageFilter;
  ackMode: 'auto' | 'manual';     // Auto or manual acknowledgment
  prefetchCount: number;           // Max unacked messages
  lastActivity: Date;
}
```

## Message Filtering

Filters control which messages pass through channels:

```typescript
interface MessageFilter {
  id: string;
  name: string;
  enabled: boolean;
  conditions: FilterCondition[];
  action: 'allow' | 'deny' | 'modify' | 'route';
  priority: number;
}

interface FilterCondition {
  field: string;                   // Message field path
  operator: 'eq' | 'ne' | 'gt' | 'lt' | 'contains' | 'matches' | 'in';
  value: any;
  caseSensitive?: boolean;
}
```

## Routing Rules

```typescript
interface RoutingRule {
  id: string;
  name: string;
  enabled: boolean;
  priority: number;
  conditions: FilterCondition[];
  actions: RoutingAction[];
}

interface RoutingAction {
  type: 'forward' | 'duplicate' | 'transform' | 'aggregate' | 'delay';
  target?: string;
  config: Record<string, any>;
}
```

## Channel Middleware

Process messages as they flow through channels:

```typescript
interface ChannelMiddleware {
  id: string;
  name: string;
  enabled: boolean;
  order: number;
  process: (message: Message, context: MiddlewareContext) => Promise<Message | null>;
  // Return null to drop the message
}
```

## Retry Policy

```typescript
interface RetryPolicy {
  maxAttempts: number;
  initialDelay: number;            // ms
  maxDelay: number;                // ms
  backoffMultiplier: number;
  jitter: boolean;                 // Add randomized jitter
}
```

## Quality of Service (QoS)

MQTT-style QoS levels:

| Level | Name           | Guarantee                    |
|-------|----------------|------------------------------|
| 0     | At most once   | Fire and forget              |
| 1     | At least once  | Guaranteed delivery (may dup)|
| 2     | Exactly once   | No duplicates, no loss       |

## Lifecycle

```typescript
await bus.initialize();
// Initializes router, delivery manager, retry manager
// Creates default channels
// Starts metrics collection

await bus.shutdown();
// Stops metrics, shuts down components
// Persists remaining messages if persistence enabled
```

## Events

- `messagebus:initialized` — Bus ready
- `messagebus:shutdown` — Bus stopped
- `delivery:success` — Message delivered
- `delivery:failure` — Delivery failed
- `retry:exhausted` — All retry attempts failed

## Internal Components

- **MessageRouter** — Routes messages to channels/queues based on rules
- **DeliveryManager** — Handles actual message delivery with receipt tracking
- **RetryManager** — Manages retry logic with exponential backoff
- **MessageBusMetrics** — Tracks throughput, latency, error rates