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

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

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

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

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:

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

Channel Types

Channel Configuration

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

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

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:

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

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:

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

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:

Lifecycle

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